# ADR-0004: Eventing Strategy - Event Broker and Contract Patterns

**Status**: Accepted **Date**: 2026-01-10 **Authors**: Development Team
**Reviewers**: Architecture Team, Platform Team **Supersedes**: N/A **Superseded
by**: N/A

## Context and Problem Statement

The Oshun monorepo requires a unified eventing strategy to enable asynchronous
communication between services, job processing, and real-time features across
all domains (lilith, yemaya, isis, sophia, hathor, bellona). Currently, both
primary codebases implement different but compatible eventing patterns:

**Lilith Eventing (Current State)**:

- Custom in-memory queue designed for Redis backing (`@lilith/queue`)
- Fastify WebSocket with Redis Pub/Sub for real-time sync
- Redis for conversation state, read receipts, presence
- WebRTC SFU for voice/video
- Pattern: Monolithic real-time service with shared Redis

**Yemaya Eventing (Current State)**:

- BullMQ for GPU/compute job queues (12 job types)
- Native `ws` library with Redis adapter for horizontal scaling
- Yjs CRDTs with y-websocket for collaborative editing
- Node heartbeat system for cluster coordination
- Pattern: Modular packages with production-grade scaling

We need to standardize on:

1. **Event Broker Choice**: Which message broker(s) for async communication
2. **Event Contract Patterns**: How to define, version, and validate event
   schemas

This decision affects service decoupling, system scalability, debugging
capabilities, and cross-domain integration.

## Decision Drivers

- **Scalability**: Support 100k+ concurrent users with high-throughput events
- **Reliability**: At-least-once delivery, persistence, and recovery
- **Developer Experience**: Easy event publishing/consuming, good debugging
- **Performance**: Low latency for real-time features (<50ms)
- **Cost Efficiency**: Minimize infrastructure complexity and costs
- **Existing Investment**: Leverage proven patterns from both codebases
- **Operational Simplicity**: Manageable infrastructure footprint
- **Type Safety**: Strongly typed event contracts for compile-time safety
- **Evolvability**: Schema versioning for backward compatibility

## Considered Options

### Event Broker Options

#### Option A: Redis (Pub/Sub + Streams) + BullMQ (Recommended)

**Description**: Use Redis as the primary message broker with Pub/Sub for
real-time fan-out events, Redis Streams for persistent event logs, and BullMQ
for job queue management.

**Pros**:

- ✅ **Already In Use**: Both codebases use Redis extensively
- ✅ **Single Infrastructure**: One Redis cluster handles caching, sessions,
  pub/sub, and queues
- ✅ **Low Latency**: Sub-millisecond publish, ideal for real-time
- ✅ **BullMQ Maturity**: Production-proven job queue with UI, metrics, and
  advanced features
- ✅ **Redis Streams**: Persistent message log with consumer groups for
  at-least-once delivery
- ✅ **Operational Simplicity**: Teams already know Redis
- ✅ **Cost Effective**: Redis Cluster scales well without additional services

**Cons**:

- ❌ **Memory Bound**: Large event volumes require careful memory management
- ❌ **Pub/Sub Limitations**: No persistence (use Streams for guaranteed
  delivery)
- ❌ **Single Point**: Requires Redis Cluster for HA (complexity)

**Use Cases**:

- Real-time notifications (Pub/Sub)
- Job queues (BullMQ)
- Event sourcing (Redis Streams)
- Cross-service communication

#### Option B: Apache Kafka

**Description**: Kafka as a distributed event streaming platform for all async
communication.

**Pros**:

- ✅ **Massive Scale**: Handles millions of events/second
- ✅ **Durability**: Persistent log with configurable retention
- ✅ **Consumer Groups**: Parallel processing with exactly-once semantics
- ✅ **Replay**: Re-process events from any point in history

**Cons**:

- ❌ **Complexity**: ZooKeeper/KRaft coordination, partition management
- ❌ **Latency**: Higher latency than Redis for small messages
- ❌ **Operational Overhead**: Requires dedicated expertise
- ❌ **Overkill**: Over-engineered for current scale requirements
- ❌ **Cost**: Additional infrastructure, managed Kafka is expensive
- ❌ **No Current Usage**: Neither codebase uses Kafka

#### Option C: RabbitMQ

**Description**: RabbitMQ as a traditional message broker with exchanges and
queues.

**Pros**:

- ✅ **Flexible Routing**: Exchanges, bindings, topic patterns
- ✅ **Guaranteed Delivery**: Acknowledgments, dead letter queues
- ✅ **Management UI**: Good visibility into queues and messages
- ✅ **Protocol Support**: AMQP, MQTT, STOMP

**Cons**:

- ❌ **Additional Infrastructure**: Another service to manage
- ❌ **Not In Use**: Neither codebase uses RabbitMQ
- ❌ **Performance**: Slower than Redis for simple pub/sub
- ❌ **Complexity**: Exchange types, bindings add cognitive load

#### Option D: Cloud-Native (AWS SQS/SNS, GCP Pub/Sub)

**Description**: Use cloud provider's managed messaging services.

**Pros**:

- ✅ **Fully Managed**: No infrastructure to maintain
- ✅ **Scalable**: Automatic scaling
- ✅ **Integrated**: Works well with other cloud services

**Cons**:

- ❌ **Vendor Lock-In**: Tied to specific cloud provider
- ❌ **Latency**: Higher latency than self-hosted Redis
- ❌ **Cost at Scale**: Per-message pricing adds up
- ❌ **Local Dev**: Complex local development setup

### Event Contract Options

#### Contract Option 1: Zod Schemas with TypeScript (Recommended)

**Description**: Define event contracts using Zod schemas that generate
TypeScript types and provide runtime validation.

**Pros**:

- ✅ **Type Safety**: Full TypeScript inference from schemas
- ✅ **Runtime Validation**: Validate events at publish and consume
- ✅ **Already In Use**: Both codebases use Zod extensively
- ✅ **Composable**: Extend/merge schemas easily
- ✅ **Self-Documenting**: Schema is the documentation

#### Contract Option 2: Protocol Buffers

**Description**: Use protobuf for event schema definition with code generation.

**Pros**:

- ✅ **Compact**: Binary serialization, smaller payloads
- ✅ **Language Agnostic**: Generate clients for any language
- ✅ **Versioning**: Built-in backward compatibility rules

**Cons**:

- ❌ **Build Step**: Requires code generation
- ❌ **Debugging**: Binary format harder to inspect
- ❌ **TypeScript Integration**: Extra tooling needed

#### Contract Option 3: JSON Schema

**Description**: Use JSON Schema for event validation.

**Pros**:

- ✅ **Standard**: Widely supported
- ✅ **Language Agnostic**: Works everywhere

**Cons**:

- ❌ **Weak TypeScript Integration**: Types need separate definitions
- ❌ **Verbose**: More boilerplate than Zod

## Decision Outcome

**Chosen Event Broker**: Option A - Redis (Pub/Sub + Streams) + BullMQ **Chosen
Contract Pattern**: Contract Option 1 - Zod Schemas with TypeScript

**Justification**:

### Event Broker Decision

Redis + BullMQ is the optimal choice because:

1. **Existing Infrastructure**: Both Lilith and Yemaya already use Redis for
   caching, sessions, and pub/sub. Adding Streams and BullMQ is an incremental
   enhancement, not a new system.

2. **Right-Sized Solution**: For the current scale (target: 100k users), Redis
   provides sufficient throughput (100k+ ops/second). Kafka's complexity is not
   justified.

3. **Unified Stack**: Redis serves multiple purposes:
   - Caching (application cache)
   - Sessions (auth sessions from ADR-0003)
   - Pub/Sub (real-time notifications)
   - Streams (event sourcing, guaranteed delivery)
   - BullMQ (job queues)

4. **Proven at Scale**: Yemaya's BullMQ implementation handles GPU jobs with
   sophisticated features (rate limiting, priority, deduplication). This can be
   generalized.

5. **Operational Simplicity**: One distributed system to manage (Redis Cluster)
   instead of multiple (Redis + Kafka + RabbitMQ).

### Event Contract Decision

Zod schemas are optimal because:

1. **Type Safety**: Full compile-time and runtime validation with zero runtime
   type erasure.

2. **Existing Investment**: Both codebases use Zod for API validation. Event
   schemas are a natural extension.

3. **Developer Experience**: Define once, get types and validation. No code
   generation step.

4. **Evolvability**: Zod's `.extend()`, `.merge()`, and `.transform()` support
   schema evolution.

**Implementation Plan**:

1. **Phase 1: Event Infrastructure** (Week 1)
   - Create `@oshun/events` package in `libs/shared/events/`
   - Define event envelope schema
   - Create Redis Streams publisher/consumer utilities
   - Port BullMQ patterns from Yemaya

2. **Phase 2: Event Contracts** (Week 1-2)
   - Define domain event schemas:
     - `isis.*` (generation events)
     - `sophia.*` (research events)
     - `hathor.*` (narrative events)
     - `bellona.*` (build events)
     - `yemaya.*` (project events)
     - `lilith.*` (experience events)
   - Create event registry with versioning

3. **Phase 3: Integration** (Week 2-3)
   - Migrate Lilith services to use @oshun/events
   - Migrate Yemaya services to use @oshun/events
   - Implement cross-domain event flows

4. **Phase 4: Observability** (Week 3)
   - Event tracing integration
   - Dead letter queue handling
   - Event metrics and dashboards

**Success Metrics**:

- Event publish latency < 10ms at p99
- Event delivery success rate > 99.9%
- Zero event loss during service restarts
- Cross-domain events flow correctly
- All events validated against schemas

**Review Schedule**: 45 days post-implementation

## Implementation Details

### Technical Specifications

**Event Envelope Schema**:

```typescript
// libs/shared/events/src/envelope.ts
import { z } from 'zod';

export const EventEnvelopeSchema = z.object({
  // Identity
  id: z.string().uuid(),
  type: z.string().regex(/^[a-z]+\.[a-z]+(\.[a-z]+)?$/), // domain.action or domain.entity.action
  version: z.string().regex(/^v\d+$/), // v1, v2, etc.

  // Timing
  timestamp: z.string().datetime(),
  scheduledFor: z.string().datetime().optional(),

  // Source
  source: z.object({
    service: z.string(),
    instance: z.string().optional(),
    traceId: z.string().optional(),
    spanId: z.string().optional(),
  }),

  // Context
  context: z.object({
    userId: z.string().optional(),
    organizationId: z.string().optional(),
    projectId: z.string().optional(),
    correlationId: z.string().optional(),
  }),

  // Payload
  payload: z.unknown(),

  // Metadata
  metadata: z.record(z.unknown()).optional(),
});

export type EventEnvelope = z.infer<typeof EventEnvelopeSchema>;
```

**Domain Event Schemas**:

```typescript
// libs/contracts/events/src/isis.ts
import { z } from 'zod';

// Isis domain events
export const IsisEvents = {
  'isis.asset.generated': z.object({
    assetId: z.string().uuid(),
    jobId: z.string().uuid(),
    assetType: z.enum(['image', 'video', '3d', 'audio']),
    workflowId: z.string(),
    outputUrl: z.string().url(),
    metadata: z.object({
      duration: z.number().optional(),
      dimensions: z
        .object({
          width: z.number(),
          height: z.number(),
        })
        .optional(),
      fileSize: z.number(),
      format: z.string(),
    }),
    provenance: z.object({
      model: z.string(),
      seed: z.number().optional(),
      prompt: z.string().optional(),
    }),
  }),

  'isis.job.failed': z.object({
    jobId: z.string().uuid(),
    jobType: z.string(),
    errorCode: z.string(),
    errorMessage: z.string(),
    attempts: z.number(),
    willRetry: z.boolean(),
  }),

  'isis.workflow.registered': z.object({
    workflowId: z.string(),
    name: z.string(),
    version: z.string(),
    inputSchema: z.unknown(),
    outputSchema: z.unknown(),
  }),
};

// libs/contracts/events/src/yemaya.ts
export const YemayaEvents = {
  'yemaya.project.created': z.object({
    projectId: z.string().uuid(),
    name: z.string(),
    type: z.enum(['game', 'film', 'animation']),
    ownerId: z.string().uuid(),
    organizationId: z.string().uuid().optional(),
  }),

  'yemaya.asset.approved': z.object({
    assetId: z.string().uuid(),
    projectId: z.string().uuid(),
    approvedBy: z.string().uuid(),
    approvalNote: z.string().optional(),
  }),

  'yemaya.collaboration.joined': z.object({
    sessionId: z.string().uuid(),
    userId: z.string().uuid(),
    documentId: z.string().uuid(),
    role: z.enum(['viewer', 'editor', 'owner']),
  }),
};

// libs/contracts/events/src/lilith.ts
export const LilithEvents = {
  'lilith.meditation.started': z.object({
    sessionId: z.string().uuid(),
    userId: z.string().uuid(),
    meditationType: z.string(),
    guidedBy: z.string().optional(),
    isGroup: z.boolean(),
  }),

  'lilith.meditation.completed': z.object({
    sessionId: z.string().uuid(),
    userId: z.string().uuid(),
    duration: z.number(),
    metrics: z
      .object({
        heartRateVariability: z.number().optional(),
        breathingRate: z.number().optional(),
      })
      .optional(),
  }),

  'lilith.content.purchased': z.object({
    contentId: z.string().uuid(),
    userId: z.string().uuid(),
    contentType: z.enum(['course', 'meditation', 'nft']),
    price: z.object({
      amount: z.number(),
      currency: z.string(),
    }),
    transactionId: z.string(),
  }),
};
```

**Event Publisher**:

```typescript
// libs/shared/events/src/publisher.ts
import { Redis } from 'ioredis';
import { EventEnvelope, EventEnvelopeSchema } from './envelope';
import { randomUUID } from 'crypto';

export interface PublishOptions {
  stream?: string;
  maxLen?: number;
  correlationId?: string;
}

export class EventPublisher {
  constructor(
    private redis: Redis,
    private serviceName: string
  ) {}

  async publish<T>(
    type: string,
    payload: T,
    options: PublishOptions = {}
  ): Promise<string> {
    const eventId = randomUUID();
    const stream = options.stream || 'oshun:events';

    const envelope: EventEnvelope = {
      id: eventId,
      type,
      version: 'v1',
      timestamp: new Date().toISOString(),
      source: {
        service: this.serviceName,
        instance: process.env.INSTANCE_ID,
      },
      context: {
        correlationId: options.correlationId,
      },
      payload,
    };

    // Validate envelope
    EventEnvelopeSchema.parse(envelope);

    // Publish to Redis Stream
    await this.redis.xadd(
      stream,
      'MAXLEN',
      '~',
      options.maxLen || 100000,
      '*',
      'event',
      JSON.stringify(envelope)
    );

    // Also publish to Pub/Sub for real-time subscribers
    await this.redis.publish(`events:${type}`, JSON.stringify(envelope));

    return eventId;
  }
}
```

**Event Consumer**:

```typescript
// libs/shared/events/src/consumer.ts
import { Redis } from 'ioredis';
import { EventEnvelope, EventEnvelopeSchema } from './envelope';
import { z } from 'zod';

export interface ConsumerOptions {
  stream?: string;
  group: string;
  consumer: string;
  batchSize?: number;
  blockMs?: number;
}

export type EventHandler<T> = (
  event: EventEnvelope & { payload: T }
) => Promise<void>;

export class EventConsumer {
  private handlers = new Map<string, EventHandler<unknown>>();

  constructor(
    private redis: Redis,
    private options: ConsumerOptions
  ) {}

  register<T>(
    eventType: string,
    schema: z.ZodSchema<T>,
    handler: EventHandler<T>
  ): void {
    this.handlers.set(eventType, async (event) => {
      const validated = schema.parse(event.payload);
      await handler({ ...event, payload: validated });
    });
  }

  async start(): Promise<void> {
    const stream = this.options.stream || 'oshun:events';
    const { group, consumer, batchSize = 10, blockMs = 5000 } = this.options;

    // Create consumer group if not exists
    try {
      await this.redis.xgroup('CREATE', stream, group, '0', 'MKSTREAM');
    } catch (e) {
      // Group already exists
    }

    // Consume loop
    while (true) {
      const results = await this.redis.xreadgroup(
        'GROUP',
        group,
        consumer,
        'COUNT',
        batchSize,
        'BLOCK',
        blockMs,
        'STREAMS',
        stream,
        '>'
      );

      if (!results) continue;

      for (const [, messages] of results) {
        for (const [messageId, fields] of messages) {
          try {
            const eventData = fields[1]; // fields is [key, value]
            const envelope = EventEnvelopeSchema.parse(JSON.parse(eventData));

            const handler = this.handlers.get(envelope.type);
            if (handler) {
              await handler(envelope as EventEnvelope & { payload: unknown });
            }

            // Acknowledge message
            await this.redis.xack(stream, group, messageId);
          } catch (error) {
            // Move to dead letter queue
            await this.handleFailedMessage(messageId, error);
          }
        }
      }
    }
  }

  private async handleFailedMessage(
    messageId: string,
    error: unknown
  ): Promise<void> {
    // Implementation: move to DLQ, alert, etc.
  }
}
```

**BullMQ Job Queue**:

```typescript
// libs/shared/events/src/job-queue.ts
import { Queue, Worker, Job } from 'bullmq';
import { Redis } from 'ioredis';
import { z } from 'zod';

export interface JobQueueOptions {
  name: string;
  redis: Redis;
  defaultJobOptions?: {
    attempts?: number;
    backoff?: { type: 'exponential' | 'fixed'; delay: number };
    removeOnComplete?: number;
    removeOnFail?: number;
  };
}

export class TypedJobQueue<T> {
  private queue: Queue;
  private worker?: Worker;

  constructor(
    private schema: z.ZodSchema<T>,
    private options: JobQueueOptions
  ) {
    this.queue = new Queue(options.name, {
      connection: options.redis,
      defaultJobOptions: {
        attempts: options.defaultJobOptions?.attempts || 3,
        backoff: options.defaultJobOptions?.backoff || {
          type: 'exponential',
          delay: 1000,
        },
        removeOnComplete: options.defaultJobOptions?.removeOnComplete || 1000,
        removeOnFail: options.defaultJobOptions?.removeOnFail || 5000,
      },
    });
  }

  async add(
    name: string,
    data: T,
    opts?: { priority?: number; delay?: number; jobId?: string }
  ): Promise<Job<T>> {
    // Validate data against schema
    const validated = this.schema.parse(data);
    return this.queue.add(name, validated, opts);
  }

  process(handler: (job: Job<T>) => Promise<void>): void {
    this.worker = new Worker(
      this.options.name,
      async (job) => {
        const validated = this.schema.parse(job.data);
        job.data = validated;
        await handler(job);
      },
      { connection: this.options.redis }
    );
  }

  async close(): Promise<void> {
    await this.queue.close();
    await this.worker?.close();
  }
}
```

### Event Flow Architecture

```
┌─────────────────────────────────────────────────────────────────────────┐
│                         Redis Cluster                                    │
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────────────┐  │
│  │ Redis Streams   │  │ Redis Pub/Sub   │  │ BullMQ Queues           │  │
│  │ (Event Log)     │  │ (Real-time)     │  │ (Job Processing)        │  │
│  │                 │  │                 │  │                         │  │
│  │ oshun:events    │  │ events:*        │  │ isis:gpu-jobs           │  │
│  │ oshun:dlq       │  │ presence:*      │  │ bellona:builds          │  │
│  │                 │  │ notifications:* │  │ sophia:indexing         │  │
│  └─────────────────┘  └─────────────────┘  └─────────────────────────┘  │
└─────────────────────────────────────────────────────────────────────────┘
           │                    │                        │
           │                    │                        │
    ┌──────▼──────┐      ┌──────▼──────┐         ┌──────▼──────┐
    │ Event       │      │ Real-time   │         │ Job Workers │
    │ Consumers   │      │ Subscribers │         │             │
    │             │      │             │         │ GPU Worker  │
    │ Isis API    │      │ WebSocket   │         │ Build Worker│
    │ Sophia Svc  │      │ Servers     │         │ Index Worker│
    │ Hathor Svc  │      │             │         │             │
    └─────────────┘      └─────────────┘         └─────────────┘
```

### Event Naming Convention

```
<domain>.<entity>.<action>

Examples:
- isis.asset.generated
- isis.job.failed
- isis.job.progress
- sophia.document.ingested
- sophia.index.updated
- hathor.world.published
- hathor.simulation.completed
- bellona.build.started
- bellona.build.completed
- bellona.export.ready
- yemaya.project.created
- yemaya.asset.approved
- yemaya.collaboration.joined
- lilith.meditation.started
- lilith.content.purchased
```

### Migration Strategy

**From**: Separate Lilith and Yemaya eventing systems **To**: Unified
@oshun/events package

**Steps**:

1. Create @oshun/events package with unified interfaces
2. Migrate Yemaya GPU queue to use @oshun/events (already close to target)
3. Migrate Lilith queue package to use @oshun/events
4. Create domain event schemas in @oshun/contracts-events
5. Update services to publish/consume via unified package
6. Deprecate old queue implementations

**Rollback Plan**: Services can fall back to direct Redis/BullMQ usage if
@oshun/events has issues.

## Consequences

### Positive Consequences

- ✅ **Unified Eventing**: Single pattern for all async communication
- ✅ **Type Safety**: Compile-time validation of event payloads
- ✅ **Debugging**: Structured events with correlation IDs for tracing
- ✅ **Scalability**: Redis Cluster handles growth without architecture change
- ✅ **Operational Simplicity**: One infrastructure (Redis) for multiple
  purposes
- ✅ **Cross-Domain Integration**: Events flow naturally between domains
- ✅ **Event Sourcing Ready**: Redis Streams provide persistent event log

### Negative Consequences

- ❌ **Redis Dependency**: Heavy reliance on Redis availability
- ❌ **Schema Discipline**: All events must have defined schemas
- ❌ **Migration Effort**: Existing code needs updating
- ❌ **Learning Curve**: Teams need to learn unified patterns

### Risks and Mitigation

| Risk                          | Probability | Impact   | Mitigation Strategy                                 |
| ----------------------------- | ----------- | -------- | --------------------------------------------------- |
| Redis cluster failure         | Low         | Critical | Multi-AZ deployment, automatic failover, monitoring |
| Event schema breaking changes | Medium      | High     | Schema versioning, backward compatibility rules     |
| Message loss                  | Low         | High     | Redis persistence (AOF), consumer acknowledgments   |
| High memory usage             | Medium      | Medium   | Stream trimming, TTL policies, monitoring alerts    |
| Consumer lag                  | Medium      | Medium   | Autoscaling consumers, backpressure handling        |

## Compliance and Security

### Security Implications

- **Event Encryption**: Sensitive payloads encrypted at rest
- **Access Control**: Redis ACLs for service-specific access
- **Audit Trail**: All events logged with source and context
- **PII Handling**: Events should not contain raw PII; use IDs and references

### Compliance Requirements

- **Data Retention**: Event streams trimmed per retention policy
- **Audit Logging**: Event publish/consume logged for compliance
- **GDPR**: User-related events support data subject access requests

## Monitoring and Observability

### Metrics to Track

- **Publishing**: Events published/second, publish latency, publish errors
- **Consuming**: Consumer lag, processing time, ack rate, DLQ size
- **Queues**: Queue depth, worker utilization, job completion rate
- **Infrastructure**: Redis memory, CPU, connections

### Alerting Strategy

- **Critical**: Consumer lag > 10,000, DLQ growth, Redis master failover
- **Warning**: Consumer lag > 1,000, high memory usage, publish latency > 100ms
- **Info**: New event types, consumer group changes, stream trimming

## Related Decisions

### Upstream Dependencies

- **ADR-0001**: Git Consolidation (single repo for shared event packages)
- **ADR-0002**: pnpm (workspace dependencies)
- **ADR-0003**: Unified Auth (event context includes auth info)

### Downstream Impacts

- **All Services**: Must adopt @oshun/events for async communication
- **Observability**: Tracing integration for event flows
- **API Gateway**: May forward events to clients via SSE/WebSocket

## References

### External Resources

- [Redis Streams Documentation](https://redis.io/docs/data-types/streams/)
- [BullMQ Documentation](https://docs.bullmq.io/)
- [Event-Driven Architecture Patterns](https://microservices.io/patterns/data/event-driven-architecture.html)
- [Zod Documentation](https://zod.dev/)

### Internal Resources

- Lilith queue package — `lilith/packages/queue/` (pre-consolidation source
  repo)
- Yemaya GPU jobs package — `yemaya/packages/gpu-jobs/` (pre-consolidation
  source repo)
- Yemaya WebSocket package — `yemaya/packages/websocket/` (pre-consolidation
  source repo)

---

## Revision History

| Version | Date       | Author           | Changes         |
| ------- | ---------- | ---------------- | --------------- |
| 1.0     | 2026-01-10 | Development Team | Initial version |