Dual Stream Architecture

Dual-Stream Architecture

Publish events to Kafka (durability) and Redis Pub/Sub (real-time) simultaneously for systems needing both guaranteed delivery and instant updates.

OpenClaw / Moltbot / Clawbot

npx clawhub@latest install dual-stream-architecture

When to Use

Event-driven systems needing both durability AND real-time WebSocket/SSE backends that push live updates Dashboards showing events as they happen Kafka consumers have lag but users expect instant updates

Core Pattern

type DualPublisher struct { kafka *kafka.Writer redis *redis.Client logger *slog.Logger } func (p *DualPublisher) Publish(ctx context.Context, event Event) error { // 1. Kafka: Critical path - must succeed payload, _ := json.Marshal(event) err := p.kafka.WriteMessages(ctx, kafka.Message{ Key: []byte(event.SourceID), Value: payload, }) if err != nil { return fmt.Errorf("kafka publish failed: %w", err) } // 2. Redis: Best-effort - don't fail the operation p.publishToRedis(ctx, event) return nil } func (p *DualPublisher) publishToRedis(ctx context.Context, event Event) { // Lightweight payload (full event in Kafka) notification := map[string]interface{}{ "id": event.ID, "type": event.Type, "source_id": event.SourceID, } payload, _ := json.Marshal(notification) channel := fmt.Sprintf("events:%s:%s", event.SourceType, event.SourceID) // Fire and forget - log errors but don't propagate if err := p.redis.Publish(ctx, channel, payload).Err(); err != nil { p.logger.Warn("redis publish failed", "error", err) } }

Architecture

┌──────────────┐ ┌─────────────────┐ ┌──────────────┐ │ Ingester │────▶│ DualPublisher │────▶│ Kafka │──▶ Event Processor │ │ │ │ │ (durable) │ └──────────────┘ │ │ └──────────────┘ │ │ ┌──────────────┐ │ │────▶│ Redis PubSub │──▶ WebSocket Gateway │ │ │ (real-time) │ └─────────────────┘ └──────────────┘

Channel Naming Convention

• events:{source_type}:{source_id}
• Examples:
• events:user:octocat - Events for user octocat
• events:repo:owner/repo - Events for a repository
• events:org:microsoft - Events for an organization

Batch Publishing

For high throughput: func (p *DualPublisher) PublishBatch(ctx context.Context, events []Event) error { // 1. Batch to Kafka messages := make([]kafka.Message, len(events)) for i, event := range events { payload, _ := json.Marshal(event) messages[i] = kafka.Message{ Key: []byte(event.SourceID), Value: payload, } } if err := p.kafka.WriteMessages(ctx, messages...); err != nil { return fmt.Errorf("kafka batch failed: %w", err) } // 2. Redis: Pipeline for efficiency pipe := p.redis.Pipeline() for _, event := range events { channel := fmt.Sprintf("events:%s:%s", event.SourceType, event.SourceID) notification, _ := json.Marshal(map[string]interface{}{ "id": event.ID, "type": event.Type, }) pipe.Publish(ctx, channel, notification) } if _, err := pipe.Exec(ctx); err != nil { p.logger.Warn("redis batch failed", "error", err) } return nil }

Decision Tree

RequirementStreamWhyMust not lose eventKafka onlyAck required, replicatedUser sees immediatelyRedis onlySub-ms deliveryBoth durability + real-timeDual streamThis patternHigh volume (>10k/sec)Kafka, batch RedisRedis can bottleneckMany subscribers per channelRedis + local fan-outDon't hammer Redis

Related Skills

Meta-skill: ai/skills/meta/realtime-dashboard/ — Complete realtime dashboard guide websocket-hub-patterns — WebSocket gateway backend/service-layer-architecture — Service integration

NEVER Do

NEVER fail on Redis errors — Redis is best-effort. Log and continue. NEVER send full payload to Redis — Send IDs only, clients fetch from API. NEVER create one Redis channel per event — Use source-level channels. NEVER skip Kafka for "unimportant" events — All events go to Kafka for replay. NEVER use Redis Pub/Sub for persistence — Messages are fire-and-forget.

Edge Cases

CaseSolutionRedis downLog warning, continue with Kafka onlyClient connects mid-streamQuery API for recent events, then subscribeHigh channel cardinalityUse wildcard patterns or aggregate channelsKafka backpressureBuffer in memory with timeout, fail if fullNeed event replayConsume from Kafka from offset, not Redis