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:

// 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:

// 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:

// 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:

// 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:

// 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#

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
• BullMQ Documentation
• Event-Driven Architecture Patterns
• Zod Documentation

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#