Skip to content
SiteSwarm Platform Documentation

Event Processing Platform

Event Processing Platform (EVENT_PROCESSING_PLATFORM.md)

1. Executive Summary & Core Architectural Intent

The Event Processing Platform delivers an edge-native, decoupled asynchronous event processing backbone across the SiteSwarm platform. It establishes a resilient store-and-forward architecture to ingest, validate, record, and dispatch asynchronous workloads without blocking client-facing web requests or risking dropped critical events.

While SiteSwarm operates entirely on Cloudflare Workers and serverless primitives rather than heavy self-hosted clusters (such as Apache Kafka or RabbitMQ), this platform emulates traditional message broker semantics using Cloudflare Queues as topic conduits.

flowchart LR
    Producers["Event Producers\n(External Webhooks & Internal Apps)"]
    Ingestion["Edge Ingestion Worker\n(Signature Verification & Handshake)"]
    Idemp[("Idempotency Ledger\n(Cloudflare D1)")]
    Topics[["Dedicated Topics\n(Cloudflare Queues)"]]
    Consumers["Domain Consumer Services\n(Decoupled Worker Services)"]


    Producers ==> Ingestion
    Ingestion -.->|"Atomic Lock & Dedupe"| Idemp
    Ingestion ==>|"Route & Enqueue"| Topics
    Topics ==>|"Competing Consumers"| Consumers

Core Invariants:

  1. Store-and-Forward Handshake: Ingestion immediately verifies requests, locks idempotency, enqueues the payload, and returns 200 OK (sub-50ms). Zero synchronous downstream operations.
  2. Zero-Tolerance for Critical Events: High-integrity workloads (e.g. payment/billing webhooks) enforce strict ACID deduplication via Cloudflare D1 and automatic Dead-Letter Queue (DLQ) retention.
  3. Broker Topic Emulation over Dedicated Queues: Cloudflare Queues natively enforce the Competing Consumers pattern (one consumer worker per queue). To achieve pub/sub topic isolation, each domain topic is modeled as an independent Cloudflare Queue.

[!NOTE] Specific domain events referenced throughout this spec (such as billing webhooks, customer inquiries, and cache purges) represent illustrative reference implementations. The underlying platform is completely domain-agnostic and extensible to any future asynchronous event workload.

2. Broker Emulation & The Competing Consumers Invariant

In traditional enterprise message brokers (RabbitMQ, Kafka, AWS SNS/SQS):

  • Topics / Exchanges: Broadcast an event to multiple independent subscriptions.
  • Consumer Groups / Competing Consumers: Multiple instances of the same worker pull from a queue to share the load.

Cloudflare Queues implements the Competing Consumers pattern on a single queue: multiple worker instances scale horizontally to process batches in parallel, but a queue can only be attached to a single consumer Worker. Once a message is acknowledged, it is removed from that queue.

flowchart LR
    Event["Domain Event\n(e.g., 'domain.resource.action')"] --> Router["Topic Router\n(Ingestion Worker)"]


    Router -->|"Publish copy"| Q_A["Topic: domain-a.events\n(Cloudflare Queue)"]
    Router -->|"Publish copy"| Q_B["Topic: domain-b.events\n(Cloudflare Queue)"]


    Q_A --> W_A["Domain A Consumer Worker\n(Dedicated Consumer Service)"]
    Q_B --> W_B["Domain B Consumer Worker\n(Dedicated Consumer Service)"]

Architectural Rules:

  • Dedicated Topic per Consumer Subsystem: Separate downstream workloads MUST bind to separate, dedicated Cloudflare Queues.
  • Blast Radius Isolation: If a downstream service or third-party dependency suffers degradation, its dedicated queue buffers and retries independently. Other topics remain completely unaffected.
  • Customized Batch & Concurrency Tuning: Each topic queue defines independently calibrated batch sizes, delay windows, and retry limits tailored to its workload profile (e.g., immediate sub-second single processing vs. batched bulk processing).

3. Ingestion Protocols & Decoupled Adapter Gateways

To keep the platform core unbloated and strictly honor the Open/Closed Principle, the Core Ingestion Worker (services/event-platform) is strictly an internal, private RPC service accessible exclusively via Cloudflare Service Bindings. It has no public HTTP routes, no external internet exposure, and zero built-in knowledge of third-party signature formats (like Stripe HMAC secrets or GitHub webhook tokens).

External HTTP webhooks are terminated by dedicated ingress adapter workers that verify provider-specific signatures, canonicalize payloads, and forward them over internal Service Binding RPC to the core ingestion worker:

flowchart LR
    subgraph ExternalIngress ["1. External Public Ingress"]
        StripeWebhook["Stripe Public Webhook\n(POST /webhook/stripe)"]
        StripeAdapter["Dedicated Stripe Webhook Adapter\n• Public HTTP Endpoint\n• Verifies Stripe-Signature\n• Normalizes Event Payload"]
        StripeWebhook ==>|"Public Internet HTTP + HMAC"| StripeAdapter
    end


    subgraph InternalIngress ["1. Internal Tenant Ingress"]
        TenantApps["Internal Tenant Apps & CMS\n(apps/*)"]
    end


    subgraph CorePlatform ["2. Private Core Ingestion Worker (RPC-Only)"]
        direction TB
        RpcInterface["Service Binding RPC Interface\n(WorkerEntrypoint: publish())"]
        PlatformAuth["Platform Authentication Gate\n(Mutual Service Binding Trust)"]
        D1Idemp[("D1 Idempotency Ledger\n(INSERT INTO event_ledger)")]
        TopicRouter["Topic Router & Enqueue\n(env.TOPIC_QUEUE.send())"]
        FastAck["Immediate Resolved Promise\n(<50ms Handshake)"]


        RpcInterface --> PlatformAuth
        PlatformAuth --> D1Idemp
        D1Idemp -->|"First Seen"| TopicRouter
        D1Idemp -->|"Duplicate"| FastAck
        TopicRouter --> FastAck
    end


    %% Ingress to Private RPC Core
    StripeAdapter ==>|"Internal RPC\nenv.EVENT_PLATFORM.publish()"| RpcInterface
    TenantApps ==>|"Internal RPC\nenv.EVENT_PLATFORM.publish()"| RpcInterface


    %% Outbound to Topics
    TopicRouter ==>|"Enqueue Payload"| TopicQueues[["Dedicated Topic Queues\n(Cloudflare Queues)"]]

3.1 The Private Ingestion Core (RPC-Only Invariant)

The core platform worker (services/event-platform):

  • No Public HTTP Routes: Deployed with workers_dev: false and no public route patterns, fully shielding the event bus and D1 database from direct internet exposure.
  • Single Ingress Protocol: Exposes only WorkerEntrypoint RPC (EventPlatformRpc).
  • Platform Invariant: The core only manages generic platform invariants: validating internal platform identity, checking the D1 strongly consistent idempotency ledger, enqueuing to topic queues, and enforcing DLQ retention. Adding new third-party providers never requires touching the core ingestion code.

3.2 Pluggable External Webhook Adapters

Each external third-party webhook provider is deployed as its own lightweight edge adapter (e.g. services/webhook-stripe):

  1. Public Webhook Termination: Listens on public HTTP (POST /webhook/stripe).
  2. Provider Cryptography: Implements provider-specific verification (e.g. stripe.webhooks.constructEvent() with STRIPE_WEBHOOK_SECRET).
  3. Payload Canonicalization: Maps the third-party payload into the standard SiteSwarm event envelope (id, tenantId, type, payload, timestamp).
  4. Internal RPC Forwarding: Calls await env.EVENT_PLATFORM.publish(platformEvent) over the local V8 isolate binding.

3.3 Internal Tenant Storefront & CMS Ingress

Tenant edge applications (apps/*) and platform services (like CMS publish hooks) bind directly to env.EVENT_PLATFORM in wrangler.jsonc. Because both caller and callee live within the Cloudflare edge runtime, calls execute with zero network hops, zero TLS negotiation, and sub-millisecond local latency.

3.4 Handshake Sequence & Decoupled Acknowledgment Flow

With external webhook termination decoupled into adapter workers, the end-to-end handshake sequence cleanly separates third-party HTTP termination from the private core’s RPC execution:

sequenceDiagram
    autonumber
    actor ExtProducer as External Producer (e.g. Stripe)
    participant Adapter as Dedicated Webhook Adapter
    actor IntWorker as Internal Tenant App
    participant CoreWorker as Private Core Ingestion Worker
    participant D1 as D1 Idempotency Ledger
    participant Queue as Dedicated Topic (Cloudflare Queue)


    alt Path A: External Public Webhook Ingress
        ExtProducer->>Adapter: POST /webhook/stripe (HTTP + Stripe-Signature)
        critical Vendor Signature Verification
            Adapter->>Adapter: Verify Stripe HMAC Secret (Reject 401 if invalid)
            Adapter->>Adapter: Canonicalize into PlatformEvent
        end
        Adapter->>CoreWorker: env.EVENT_PLATFORM.publish(event) [Service Binding RPC]
    else Path B: Internal Worker Ingress
        IntWorker->>CoreWorker: env.EVENT_PLATFORM.publish(event) [Service Binding RPC]
    end


    critical Core Ingestion Handshake (RPC Core)
        CoreWorker->>CoreWorker: Platform Auth & Contract Check
        CoreWorker->>D1: INSERT INTO event_ledger (event_id, status='received')
        alt Duplicate Key (Already in Ledger)
            D1-->>CoreWorker: Unique Constraint Violation (SQLITE_CONSTRAINT)
            CoreWorker-->>Adapter: Resolve RPC: { status: 'duplicate_acknowledged' }
            CoreWorker-->>IntWorker: Resolve RPC: { status: 'duplicate_acknowledged' }
        else First Time Seen
            D1-->>CoreWorker: Insert Success
            CoreWorker->>Queue: env.TOPIC_QUEUE.send(eventPayload)
            CoreWorker-->>Adapter: Resolve RPC: { status: 'enqueued' }
            CoreWorker-->>IntWorker: Resolve RPC: { status: 'enqueued' }
        end
    end


    opt Return External HTTP Response
        Adapter-->>ExtProducer: HTTP 200 OK (Webhook Safely Persisted)
    end

Key Handshake Guarantees:

  1. Third-Party Signature Isolation: Vendor cryptography (e.g. Stripe HMAC verification) happens strictly inside the adapter isolate before reaching the platform core.
  2. Deterministic Internal RPC Interface: Whether invoked by an adapter or a tenant storefront, the core ingestion worker receives a pre-validated PlatformEvent over an internal Service Binding.
  3. Store-and-Forward Confirmation: External providers receive their HTTP 200 OK only after the core worker has durably committed the event into both D1 and the topic queue.

4. Reliability & Strongly Consistent Idempotency Engine

Eventual consistency (such as Cloudflare KV) is insufficient for zero-tolerance events like billing. Rapid duplicate retries from Stripe hitting different global Cloudflare PoPs can execute simultaneously if an idempotency check relies on eventually-consistent KV.

The platform uses Cloudflare D1 SQLite for strongly consistent ACID transactions:

flowchart TD
    subgraph D1Ledger ["D1 Idempotency & Audit Table: 'event_ledger'"]
        Cols["• event_id (TEXT PRIMARY KEY)\n• tenant_id (TEXT NOT NULL)\n• topic (TEXT NOT NULL)\n• status ('received' | 'processing' | 'completed' | 'failed')\n• payload_hash (TEXT NOT NULL)\n• attempts (INTEGER DEFAULT 0)\n• created_at (TIMESTAMP)\n• updated_at (TIMESTAMP)"]
    end


    RetryRequest["Incoming Event\n(ID: evt_stripe_1029)"] --> Check{"INSERT INTO event_ledger\nON CONFLICT(event_id) DO NOTHING"}
    Check -->|"Rows Written = 1"| Valid["Proceed to Queue"]
    Check -->|"Rows Written = 0"| Dup["Duplicate: Return 200 OK / Skip"]

Schema Definition:

CREATE TABLE IF NOT EXISTS event_ledger (
  event_id TEXT PRIMARY KEY,
  tenant_id TEXT NOT NULL,
  topic TEXT NOT NULL,
  status TEXT CHECK(status IN ('received', 'processing', 'completed', 'failed', 'dlq')) DEFAULT 'received',
  payload TEXT NOT NULL,
  attempts INTEGER DEFAULT 0,
  last_error TEXT,
  created_at TEXT NOT NULL DEFAULT (datetime('now')),
  updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);


CREATE INDEX IF NOT EXISTS idx_event_ledger_tenant ON event_ledger(tenant_id, created_at);
CREATE INDEX IF NOT EXISTS idx_event_ledger_status ON event_ledger(status, updated_at);

4.2 The Transactional Outbox Pattern & Fan-Out Failure Remediation

The Classic Dual-Write Problem at the Edge

A fundamental dilemma in distributed systems is the Dual-Write Problem: when an application must both persist state in a database (D1) and emit a message to an external broker or queue (Cloudflare Queues), achieving atomicity across these two heterogeneous systems is notoriously difficult. Two-phase commit (2PC) protocols are brittle, introduce massive latency, and are unsupported in serverless edge runtimes.

If an ingestion worker writes to the database first and then attempts to enqueue:

  • A sudden isolate crash, edge CPU limit, or Cloudflare Queue blip causes the enqueue operation to fail. The database records the event, but downstream consumers never hear about it.
  • Conversely, if the worker enqueues first and the database write fails, downstream consumers begin processing a phantom event that has no corresponding ledger record.

When fan-out is introduced (where a single event must be delivered to multiple topic queues), the problem compounds into Partial Dispatch Failure:

  • Sending to Topic A succeeds, but sending to Topic B fails (due to queue backlog limits, transient 500s, or rate limits).
  • If the worker throws an error to prompt the upstream caller (e.g., Stripe) to retry, a naive retry re-sends to Topic A, causing duplicate work or duplicate billing notices.
  • If the worker swallows the error and returns 200 OK, Topic B’s message is silently lost forever.

How the Transactional Outbox Solves It

The Transactional Outbox Pattern eliminates the dual-write dilemma by converting the asynchronous messaging step into a local, single-system ACID transaction:

  1. Atomic Pre-Commit: When an event arrives, the ingestion engine determines all target topics. Rather than immediately calling out to Cloudflare Queues, it writes both the master event (event_ledger) and an outbox entry for each target destination (event_dispatches) in a single atomic D1 SQLite transaction. Because SQLite enforces strict ACID guarantees, it is physically impossible for an event to be accepted without its destination outbox entries also being locked in storage.
  2. Optimistic Immediate Dispatch: Immediately following the database commit, the worker makes a best-effort asynchronous attempt to push the payload to each destination queue using Promise.allSettled(). If all sends succeed, the outbox rows transition to enqueued, and the worker returns a sub-50ms success response.
  3. Decoupled Recovery (The Outbox Sweeper): If an immediate queue push fails or the worker isolate is abruptly evicted by Cloudflare before marking rows as enqueued, the event is not lost. The dispatch row remains marked pending or failed in D1. An asynchronous background process (an Outbox Sweeper running via Cloudflare Cron Triggers) continuously polls for stranded dispatches and pushes them to their target queues.
  4. Selective Re-Entrancy: Because dispatch status is tracked per topic, any subsequent retry—whether initiated by an upstream provider like Stripe or by the platform’s internal outbox sweeper—only targets destinations that have not yet been marked enqueued. Topics that already received the message are cleanly bypassed.
flowchart TD
    Event["Incoming Event\n(Targets: [Topic A, Topic B])"] --> TxStep["1. Atomic D1 Transaction\n• INSERT INTO event_ledger\n• INSERT INTO event_dispatches (Topic A, pending)\n• INSERT INTO event_dispatches (Topic B, pending)"]


    TxStep --> DispatchStep["2. Dispatch via Promise.allSettled()"]


    DispatchStep --> SendA["send(Topic A) -> Success"]
    DispatchStep --> SendB["send(Topic B) -> FAILS (Transient 500)"]


    SendA --> UpdateA["UPDATE event_dispatches\nSET status='enqueued' WHERE topic='A'"]
    SendB --> UpdateB["UPDATE event_dispatches\nSET status='failed', error=err WHERE topic='B'"]


    UpdateB --> Eval{"Caller Retry or Outbox Sweeper?"}


    Eval -->|"Path 1: Caller Retries (e.g. Stripe)"| RetryFlow["D1 Checks Dispatches:\n• Topic A already enqueued (SKIP)\n• Topic B failed/pending (RETRY ONLY TOPIC B)"]
    Eval -->|"Path 2: Caller Does Not Retry"| CronSweeper["Platform Cron Sweeper (every 2m):\n• Queries failed/pending dispatches\n• Re-attempts send() to Topic B\n• If retries exhausted -> routes to dlq_incidents"]

4.3 Dispatch Ledger Schema (event_dispatches)

CREATE TABLE IF NOT EXISTS event_dispatches (
  dispatch_id TEXT PRIMARY KEY,
  event_id TEXT NOT NULL,
  topic TEXT NOT NULL,
  status TEXT CHECK(status IN ('pending', 'enqueued', 'failed', 'dlq_abandoned')) DEFAULT 'pending',
  attempts INTEGER NOT NULL DEFAULT 0,
  max_attempts INTEGER NOT NULL DEFAULT 5,
  next_retry_at TEXT NOT NULL DEFAULT (datetime('now')),
  last_error TEXT,
  created_at TEXT NOT NULL DEFAULT (datetime('now')),
  updated_at TEXT NOT NULL DEFAULT (datetime('now')),
  FOREIGN KEY (event_id) REFERENCES event_ledger(event_id),
  UNIQUE(event_id, topic)
);


CREATE INDEX IF NOT EXISTS idx_dispatches_sweeper ON event_dispatches(status, next_retry_at);

4.4 Guaranteed Invariants for Partial Dispatch & Sweeper Retry Tracking:

  1. Atomic Pre-Registration: Destinations are registered in event_dispatches within the same D1 transaction as event_ledger before any queue send() calls execute, initialized with attempts = 0 and status = 'pending'.
  2. Immediate Attempt Accounting: If the initial optimistic send() fails during the ingress request, the worker immediately updates the row: 
    UPDATE event_dispatches
    SET attempts = attempts + 1,
        status = 'failed',
        last_error = ?,
        next_retry_at = datetime('now', '+2 minutes'),
        updated_at = datetime('now')
    WHERE dispatch_id = ?;
  3. Selective Retry Re-Entrancy: When an external webhook provider (like Stripe) retries an event following a partial failure, the ingestion worker inspects event_dispatches. Any topic already marked enqueued is skipped. Only failed or pending topics are attempted, preventing duplicate messages to already-notified topics.
  4. Exponential Backoff Sweeper Execution: The background Cloudflare Cron Trigger (running every 2 minutes) queries only records due for retry: 
    SELECT d.*, l.payload
    FROM event_dispatches d
    JOIN event_ledger l ON d.event_id = l.event_id
    WHERE d.status IN ('pending', 'failed')
      AND d.attempts < d.max_attempts
      AND d.next_retry_at <= datetime('now')
    ORDER BY d.next_retry_at ASC
    LIMIT 50;
    For each processed row:
    • On retry success: marks status = 'enqueued' and updated_at = datetime('now').
    • On retry failure: increments attempts = attempts + 1, logs last_error, and calculates exponential backoff for the next sweep (next_retry_at = datetime('now', '+' || (2 * (1 << attempts)) || ' minutes')).
  5. Exhaustion Escalation to DLQ (dlq_abandoned): When attempts >= max_attempts (default 5 attempts over ~1 hour with exponential backoff), the outbox sweeper transitions the dispatch row to status = 'dlq_abandoned' and immediately creates a record in dlq_incidents with error_message = 'Outbox sweeper retries exhausted'. This triggers operator alerting and exposes the stranded dispatch for manual replay without losing any data.

5. Consumer Execution, Retries & Dead-Letter Queue (DLQ) Architecture

When a dedicated topic consumer worker processes a batch of messages, it must handle transient errors gracefully and isolate permanent failures without losing data.

flowchart TD
    subgraph ProcessingFlow ["Consumer Execution & Error Handling"]
        TopicQueue[["Topic Queue (Cloudflare Queue)"]]
        ConsumerWorker["Consumer Worker Isolate\n(Batch: 1..N messages)"]
        ExternalAPI["Downstream Target / API"]
        CF_DLQ[["Cloudflare DLQ Queue\n(Ephemeral Transport Buffer)"]]


        TopicQueue -->|"Pull Message Batch"| ConsumerWorker
        ConsumerWorker -->|"Execute Workload"| ExternalAPI


        ExternalAPI -->|"Success (200)"| Ack["message.ack()"]
        ExternalAPI -.->|"Transient Error (500 / Rate Limit)"| RetryCheck{"Retries < Max?"}


        RetryCheck -->|"Yes"| Retry["message.retry({ delaySeconds: 2^attempt * 5 })"]
        RetryCheck -->|"No (Max Retries Exceeded)"| DLQRoute["Route to DLQ"]
        DLQRoute --> CF_DLQ
    end

5.1 The 4-Day Retention Problem & The Drain-to-Storage Pattern

A critical limitation of Cloudflare Queues is that messages have a maximum retention window (default 4 days) before being automatically purged by the edge network.

For mission-critical workloads (such as billing events, tenant subscriptions, and customer inquiries), relying on a human operator to notice and triage an ephemeral queue within 4 days is an unacceptable failure mode.

To solve this, the Cloudflare DLQ Queue is treated strictly as an ephemeral transport buffer, NOT a permanent resting place.

A dedicated DLQ Archival Worker consumes the dead-letter queue immediately upon arrival, draining messages out of the 4-day ephemeral window and into permanent, infinite-retention storage:

flowchart LR
    subgraph EdgeBuffer ["Ephemeral Transport (Max 4 Days)"]
        CF_DLQ[["Cloudflare DLQ Queue\n(siteswarm-*-dlq)"]]
    end


    subgraph ArchivalWorker ["Immediate DLQ Consumer"]
        DLQConsumer["DLQ Archival Worker\n(Batch: 1..10)"]
    end


    subgraph PermanentStorage ["Permanent Storage & Operations"]
        D1Incidents[("D1 SQLite: 'dlq_incidents'\n(Infinite Retention)")]
        R2Archive[("R2 Object Storage\n(Large Raw Payloads)")]
        OpsAlert["Operator Alerting\n(Command Center / Webhook)"]
    end


    CF_DLQ ==>|"Immediately Drains"| DLQConsumer
    DLQConsumer -->|"1. Record Incident & Stack Trace"| D1Incidents
    DLQConsumer -->|"2. Archive Large Body (>64KB)"| R2Archive
    DLQConsumer -->|"3. Dispatch Alert"| OpsAlert
    DLQConsumer ==>|"4. message.ack() (Purge from 4-day queue)"| CF_DLQ

5.2 The Incident Ledger Schema (dlq_incidents)

When the DLQ Archival Worker drains a failed message, it persists an audit record in Cloudflare D1:

CREATE TABLE IF NOT EXISTS dlq_incidents (
  incident_id TEXT PRIMARY KEY,
  event_id TEXT NOT NULL,
  tenant_id TEXT NOT NULL,
  topic TEXT NOT NULL,
  source_queue TEXT NOT NULL,
  error_message TEXT NOT NULL,
  error_stack TEXT,
  retry_count INTEGER NOT NULL,
  payload TEXT NOT NULL,
  status TEXT CHECK(status IN ('open', 'triaged', 'retrying', 'resolved', 'ignored')) DEFAULT 'open',
  replayed_at TEXT,
  resolved_at TEXT,
  resolved_by TEXT,
  created_at TEXT NOT NULL DEFAULT (datetime('now')),
  updated_at TEXT NOT NULL DEFAULT (datetime('now')),
  FOREIGN KEY (event_id) REFERENCES event_ledger(event_id)
);


CREATE INDEX IF NOT EXISTS idx_dlq_status ON dlq_incidents(status, created_at);
CREATE INDEX IF NOT EXISTS idx_dlq_tenant ON dlq_incidents(tenant_id, created_at);

5.3 Post-Failure Lifecycle: Triage, Operator Remediation & Replay Engine

What happens after an event lands in dlq_incidents? The platform defines an explicit mechanical triage and replay workflow:

sequenceDiagram
    autonumber
    participant Worker as DLQ Archival Worker
    participant DB as D1 dlq_incidents
    actor Operator as Platform Operator / Developer
    participant ReplayAPI as Platform Replay RPC / API
    participant TopicQueue as Primary Topic Queue (Cloudflare Queue)
    participant Consumer as Primary Consumer Worker


    Worker->>DB: Persist Incident (status='open')
    Note over Operator: 1. Alert fires. Operator investigates failure.
    Note over Operator: 2. Root cause resolved (e.g. bug patched, API key rotated).


    Operator->>ReplayAPI: Trigger Replay (incident_id)
    ReplayAPI->>DB: SELECT * FROM dlq_incidents WHERE incident_id = ?
    ReplayAPI->>DB: UPDATE dlq_incidents SET status='retrying', replayed_at=now()
    ReplayAPI->>TopicQueue: env.TOPIC_QUEUE.send(payload, { metadata: { replayOf: incident_id } })


    TopicQueue->>Consumer: Consume re-injected message
    alt Reprocessing Succeeds
        Consumer->>DB: UPDATE dlq_incidents SET status='resolved', resolved_at=now()
    else Reprocessing Fails Again
        Consumer-->>TopicQueue: Retries exhausted -> routes back to DLQ
        Worker->>DB: UPDATE dlq_incidents SET status='open', retry_count = retry_count + 1
    end

The 4 Operational Stages of DLQ Recovery:

  1. Automated Quarantining & Immediate Alert:
    • The DLQ worker consumes the failed message within seconds, acknowledges it off the 4-day Cloudflare Queue, and inserts it into dlq_incidents as status='open'.
    • An alert is dispatched with the incident_id, event_id, and error_message.
  2. Root Cause Remediation:
    • Developers inspect the incident in the SiteSwarm Command Center or via CLI:
      Terminal window
      pnpm run swarm dlq list --status open
      pnpm run swarm dlq view <incident_id>
    • Developers resolve the root cause (e.g., deploying a fix for an edge case, adjusting a database migration, or renewing an external service credential).
  3. Controlled Single or Batch Replay:
    • Operators trigger mechanical replay without re-posting from the original client:
      Terminal window
      pnpm run swarm dlq replay <incident_id>
      # Or replay all failures for a specific tenant:
      pnpm run swarm dlq replay --tenant <tenant_id> --topic billing.events
    • The Replay API fetches the original canonical payload and re-enqueues it directly into the primary topic queue with a replayOf tracing header.
  4. Resolution Attestation:
    • When the primary consumer successfully processes a replayed message, it acknowledges the message and attests resolution in D1 (status='resolved', resolved_at=now()).
    • If the downstream service still fails, the event routes back to the DLQ, where the incident counter is incremented without creating duplicate ledger records.

6. Migration Case Study: Cache Purge Service

Currently, services/cache-purge operates as an imperative RPC service (CACHE_PURGE_SERVICE.purgeTags(...)) called directly by client applications like clk-construction-cms.

Migrating Cache Purge to the Event Processing Platform transitions it from a coupled imperative dependency to an asynchronous event subscriber:

flowchart LR
    subgraph Before ["Current: Imperative RPC Coupling"]
        CMS1["Client CMS Worker\n(clk-construction-cms)"]
        RPC["CACHE_PURGE_SERVICE.purgeTags()"]
        PurgeWorker1["Cache Purge Worker\n(services/cache-purge)"]
        CMS1 ==>|"Synchronous RPC Binding"| RPC ==> PurgeWorker1
    end


    subgraph After ["Target: Decoupled Event Publication"]
        CMS2["Client CMS Worker\n(clk-construction-cms)"]
        EventPub["emit('content.published', tags)"]
        TopicPurge[["Topic Queue: 'cache-purge-events'"]]
        PurgeWorker2["Cache Purge Consumer Worker"]
        SitemapWorker["SEO / Sitemap Worker (Future)"]


        CMS2 ==>|"Event Ingress"| EventPub
        EventPub ==> TopicPurge
        TopicPurge ==> PurgeWorker2
        EventPub -.->|"Future Fan-Out"| SitemapWorker
    end

Migration Advantages:

  • Zero Client Coupling: Client CMS workers only know how to emit domain events (content.published, content.deleted). They require no knowledge of Cloudflare zone IDs or cache purge endpoints.
  • Coalescing & Batching: The existing queue consumer inside services/cache-purge already supports batch coalescing (max_batch_size: 10). Migrating to a topic queue retains this exact optimization.
  • Multi-Subscriber Extensibility: Publishing content.published enables future subscribers (e.g. sitemap regeneration, RSS feed updates) without modifying the client CMS.

7. Putting It All Together: Complete System Topology

The composite diagram below illustrates the generic platform architecture connecting producers, ingestion, the D1 reliability engine, emulated broker topics, and decoupled consumer services:

flowchart TB
    subgraph ExternalProducers ["External Ingress"]
        ExtProducer["External Public Producers\n(e.g., Stripe Webhook)"]
        WebhookAdapter["Dedicated Ingress Adapter Worker\n(Verifies Signature, Normalizes Payload)"]
        ExtProducer ==>|"Public Internet HTTP + HMAC"| WebhookAdapter
    end


    subgraph InternalProducers ["Internal Platform Ingress"]
        IntProducer["Tenant Apps & Internal Services\n(apps/*)"]
    end


    subgraph PrivateCore ["Private Ingestion Worker (RPC-Only)"]
        RpcIngress["Service Binding RPC Ingress\n(WorkerEntrypoint: EventPlatformRpc)"]
        PlatformAuth["Platform Authentication & Trust Check"]
        D1Ledger[("D1 Idempotency Ledger\n(event_ledger)")]
        TopicRouter["Topic Router & Dispatcher"]


        RpcIngress --> PlatformAuth
        PlatformAuth -->|"1. ACID Dedupe & Store"| D1Ledger
        PlatformAuth -->|"2. Return Instant Fast Handshake"| FastACK["Fast Handshake (<50ms)"]
        D1Ledger -->|"First Seen"| TopicRouter
    end


    subgraph BrokerTopics ["Emulated Broker Topics (Cloudflare Queues)"]
        TopicA[["Topic: 'domain-a.events'\n(Cloudflare Queue A)"]]
        TopicB[["Topic: 'domain-b.events'\n(Cloudflare Queue B)"]]
        TopicC[["Topic: 'domain-c.events'\n(Cloudflare Queue C)"]]
    end


    subgraph Consumers ["Domain Consumer Services"]
        ConsumerA["Consumer Service A\n• Dedicated Domain Workload"]
        ConsumerB["Consumer Service B\n• Multi-Channel / Fan-Out Workload"]
        ConsumerC["Consumer Service C\n• Batched / Background Workload"]
    end


    subgraph SafetyNet ["Dead-Letter Storage & Operator Tools"]
        DLQStore[["Dead-Letter Queues (DLQ)"]]
        AdminCockpit["Platform Operations Cockpit & Replay API"]
    end


    %% Producer connections
    WebhookAdapter ==>|"Service Binding RPC"| RpcIngress
    IntProducer ==>|"Service Binding RPC"| RpcIngress


    %% Ingestion fan-out to topics
    TopicRouter ==>|"Route: domain-a.*"| TopicA
    TopicRouter ==>|"Route: domain-b.*"| TopicB
    TopicRouter ==>|"Route: domain-c.*"| TopicC


    %% Queues to consumers
    TopicA ==> ConsumerA
    TopicB ==> ConsumerB
    TopicC ==> ConsumerC


    %% DLQ routes
    ConsumerA -.->|"On Fatal Failure"| DLQStore
    ConsumerB -.->|"On Fatal Failure"| DLQStore
    ConsumerC -.->|"On Fatal Failure"| DLQStore
    DLQStore ==> AdminCockpit
    AdminCockpit -.->|"Manual Replay"| BrokerTopics

Reference Domain Mappings (Illustrative Examples)

To maintain strict domain neutrality, the platform does not hardcode business workflows into the ingestion core. The table below illustrates how candidate use cases map onto these abstract architectural tiers:

Workload Domain Producer Example Ingress Protocol Target Topic Queue Consumer Service Role
Mission-Critical Ingestion Stripe Webhook (invoice.paid) Protocol A (HTTP + HMAC) billing.events Subscription provisioning, customer ledger sync
Tenant Engagement Storefront Inquiry (inquiry.created) Protocol B (Worker RPC) notifications.events, crm.sync Email/SMS dispatch (fan-out), CRM contact push
Platform Maintenance CMS Publish (content.published) Protocol B (Worker RPC) cache.purge Cloudflare edge cache tag eviction
Auditing & Metrics Traffic / Event anomalies Protocol B (Worker RPC) audit.events Bulk background metrics logging to D1/R2

8. Decisions Intentionally Deferred

In alignment with SiteSwarm’s Minimal Viable Abstraction (MVA) invariant, the following capabilities are explicitly deferred:

  1. Dynamic Runtime Topic Registration: Topics are declared statically via Cloudflare Queues configuration in code and wrangler.jsonc. Dynamic topic creation via database-driven pub/sub registries is deferred until tenant volume demands it.
  2. Cross-Cloud Message Replication: Multi-region failover to non-Cloudflare queue backbones (e.g. AWS SQS or Kafka Cloud) is deferred. Cloudflare Queues’ globally distributed edge ingress provides sufficient resilience.
  3. Complex Distributed Saga Orchestration: Multi-step workflows requiring compensating rollbacks (e.g. multi-step booking payment holding) will utilize Cloudflare Workflows when required, rather than custom saga coordinators over queues.
  4. Full Schema Registry & Runtime Protobuf Serialization: Event payloads will be standardized via TypeScript interfaces and JSON Schema validation within packages/governance rather than a binary schema registry service.