createDispatcher

Create the pure ingest core: schema resolution, strict validation, identity resolution, durable persistence, and isolated fan-out.

Creates the dispatcher — the pure, framework-free ingest core. Mount it anywhere: an HTTP route (via toHandler), a queue worker, a cron job, tests, or in-process server-side tracking via the direct transport.

import { createDispatcher } from 'dotyc/ingest'

Signature

function createDispatcher<C extends Catalog>(options: {
  catalog: C
  wiring: Wiring<C>
  identity?: IdentityProvider
  store?: DotycStore
  strict?: boolean
  onError?: (err: unknown, ctx: ErrorContext) => void
}): Dispatcher

Options

Option	Description
`catalog`	The catalog from `defineCatalog`. Drives validation and computed fields.
`wiring`	The consumer registry from `createWiring`.
`identity`	Optional identity provider (ingest facet). Resolves the session or JWT from the request into a verified identity, which overrides the emitter's claim.
`store`	The state brick. Required if the catalog contains any `delivery: 'durable'` event — it backs the outbox that makes at-least-once delivery real. The same store also backs the identify dedupe cache and replay dedupe (on `eventId`, over a TTL window). Use `memoryStore()` in development; official drivers target Redis, Postgres and KV stores. In serverless, always use a shared store — never per-instance memory.
`strict`	Default `false`: an event that matches no consumer produces a warning in development and is silent in production. `true`: an unmatched event is an error.
`onError`	Global error hook: receives adapter errors (isolated — a throwing adapter affects nothing else), invalid events (rejected, never dispatched), and durable events that exhaust their retries (dead-letter). Wire it to your logger or Sentry.

Draft — subject to change: the exact DotycStore interface, the store driver packages, and the default retry policy (attempts, backoff) are not finalized.

Return value

A Dispatcher with:

await dispatcher.ingest(envelope, { request? })

envelope is the wire format — one POST, N events:

interface Envelope {
  events: RawEvent[]
  identity: IdentityClaim // emitter claim, without `verified`
}

request is optional and only needed when an identity provider must resolve a session or JWT from it.

The result of ingest carries per-event acks: each event resolves to accepted, rejected or legacy. This is what toHandler returns to the emitter, and what lets the emitter's durable lane retry exactly the events that need it.

The ingest pipeline

Parse the envelope.
Resolve the schema. Each event carries the schemaHash of the definition it was emitted against. Current hash → strict validation. A hash found in the committed schema registry (.dotyc/schema-history.json) → the event is validated against its own schema version and flagged legacy — never blocked. Unknown hash → never blocked either: flagged unknown, properties passed through raw. A stale client is not a data-loss window.
Validate — strict on the current schema. An invalid current-schema event is rejected, reported via onError, and never dispatched.
Resolve identity. The provider's verified identity overrides the claim. When lightweight ids (anonymousId/sessionId) meet a verified userId, the link is signaled to adapters via their identify hook — deduplicated through the store.
Apply computed fields.
Persist durables. Every delivery: 'durable' event is written to the outbox (via the store) before anything else happens to it. Legacy durable events are persisted like the others — a business fact is not lost because a client is one deploy behind.
Ack the transport. From here on, durable events survive a crash.
Match against the wiring criteria.
Fan out — parallel and isolated per consumer.
Retry (durables). Each consumer marks its own completion; a failed consumer is retried with backoff from the store, independently of the others. Exhausted retries move the event to a dead-letter status and report through onError. Best-effort events keep the fire-and-forget contract: no retry.

Under pressure, ingest prioritizes durable events and sheds best-effort ones — dropping best-effort traffic is its contract, degrading everyone is not.

Example

import { createDispatcher } from 'dotyc/ingest'
import { clerkIdentity } from '@dotyc/clerk/server'

export const dispatcher = createDispatcher({
  catalog,
  wiring,
  identity: clerkIdentity(),
  store: redisStore(process.env.REDIS_URL!), // memoryStore() in dev
  strict: false,
  onError: (err, ctx) => logger.error('dotyc', { err, ctx }),
})

// In a queue worker, a cron, or a test — no HTTP required:
await dispatcher.ingest(envelope)

Notes

Durability is per event, not per pipeline. Only delivery: 'durable' events pay for the outbox; a catalog with none needs no store for durability and nothing changes in the hot path.
source: 'trusted' events are rejected here if they arrived over HTTP. They may only enter through direct(dispatcher) — see defineEvent.
The core is the contract. toHandler is a thin convenience layer, never a required dependency.