async-jobs-and-events

name	async-jobs-and-events
description	Queues and workers, domain event publishers, async notifications or projections, or not doing that work inside HTTP handlers.

Background jobs, domain events, and side effects

When to use: Queues and workers, domain event publishers, async notifications or projections, or not doing that work inside HTTP handlers.

Side effects and eventing

• Domain code emits domain events through domain-level publisher abstractions (ports), not ad-hoc calls from use-cases to email/HTTP/Slack.
• Workers handle notifications, integrations, projections, and other I/O asynchronously.
• Do not orchestrate side effects (fan-out integrations, “fire and forget” HTTP, etc.) inside HTTP handlers — enqueue / publish and return.

Domain event naming and publisher–consumer decoupling

Domain events represent facts that happened — state transitions on an aggregate — not instructions for what should happen next. The publisher must never know or care which handlers are subscribed.

Rules

1. Name events after what the aggregate did, not what consumers need to hear. Good: ScoreCreated, ScoreStatusChanged. Bad: ScoreDraftSaved (named to route around a handler), ScoreReadyForDiscovery (named after a consumer concern).
2. Smell test: if you deleted every event handler, would you still emit this event because it describes a meaningfully different thing that happened? If the answer is no, the event is coupling in disguise.
3. One canonical event per state transition. Do not split a single write operation into multiple event types to route to different handlers. If a score is written, emit ScoreCreated — regardless of whether the score is a draft or published.
4. Consumers own their filtering logic. If a handler only cares about published scores, the handler checks the payload or re-fetches state and skips drafts. The publisher does not pre-filter by emitting different event names.
5. Dedupe keys must not collide across lifecycle stages. If the same entity emits the same event at different lifecycle points (e.g., draft save then final publish), include the relevant discriminator in the dedupe key — not in the event name. Example: issues:discovery:${scoreId}:${status} instead of splitting into separate event types.
6. Payload carries facts, not routing hints. Include the aggregate's current state (or the fields consumers might filter on) in the payload. Let consumers decide relevance from payload data.

When a new event type IS warranted

A new event type is justified when it represents a genuinely distinct state transition that would exist even with zero handlers — for example, ScoreDeleted is a different fact from ScoreCreated. The test: does the aggregate's lifecycle model include this transition independently of downstream concerns?

Anti-pattern: conditional event publishing

// BAD — publisher decides routing based on consumer needs
const eventName = score.draftedAt === null ? “ScorePublished” : “ScoreDraftSaved”
yield* outboxEventWriter.write({ eventName, ... })

// GOOD — one canonical event, consumers filter
yield* outboxEventWriter.write({
  eventName: “ScoreCreated”,
  payload: { scoreId: score.id, organizationId, projectId, status: score.draftedAt === null ? “published” : “draft” },
})

// Consumer side — handler owns its filtering
ScoreCreated: (event) =>
  Effect.all([
    // discovery uses status-aware dedupe key, skips drafts internally
    pub.publish(“issues”, “discovery”, event.payload, {
      dedupeKey: `issues:discovery:${event.payload.scoreId}:${event.payload.status}`,
    }),
    pub.publish(“annotation-scores”, “publishHumanAnnotation”, event.payload, {
      debounceMs: SCORE_PUBLICATION_DEBOUNCE,
    }),
  ])

Async and background tasks

• Put IDs or opaque storage keys in job payloads — not full mutable entities.
• Re-fetch authoritative state inside the worker before acting.
• Make stale or deleted entities an explicit outcome (skip, dead-letter, or record failure) instead of assuming rows still exist.
• Worker handler effects must include withTracing from @repo/observability in their pipe chain so that Effect spans flow into the OTel pipeline. See effect-and-errors for tracing rules.
• For transactional domain events, write through the OutboxEventWriter service (or a plain OutboxEventWriterShape from createOutboxWriter in @platform/db-postgres) instead of inserting outbox rows directly.
• For high-volume or otherwise non-transactional producers whose upstream write is already durable, publish directly through createEventsPublisher(queuePublisher) into domain-events instead of persisting an outbox row only to forward it.
• Domain-event consumers should act as dispatchers: publish downstream topic tasks or start workflows, but do not run synchronous business logic inline inside the event handler.
• Extend the repo’s existing async rails instead of inventing new ones: BullMQ-backed queue workers live in apps/workers, and durable multi-step workflows live in the Temporal-backed apps/workflows app.
• Changing a Temporal workflow that may have in-flight runs is a versioning event, not a refactor. Reordering, inserting, or removing activities (or sleeps, timers, child workflows) breaks replay for any history written under the old code and the workflow gets stuck on a non-determinism error — its workflow task keeps failing forever, and from the UI's perspective it stays "running" (e.g. an evaluation issue sits on "Generating" because getIssueAlignmentState sees status === "running"). Guard the change with patched("<descriptive-id>") from @temporalio/workflow, or pin the new code to a new Worker Deployment Version, before merging. See temporal-developer and its references/typescript/versioning.md for the three-step patched → deprecatePatch → remove flow and Worker Versioning setup. Draining in-flight workflows before deploy is rarely viable here: optimizeEvaluationWorkflow activities are sized up to 75 min (GEPA budget) and the full pipeline can run substantially longer, so a deploy window almost never finds all evaluations:* workflow IDs idle.
• Queue topics may own several related lower-kebab-case task names; one worker module owns the topic and dispatches by task name.
• Queue publication should expose logical dedupe/debounce or throttle keyed by the relevant entity identity when the transport supports it.
• Use queue topics for single-step tasks and the workflow abstraction for long-running or multi-step orchestration.
• For reliability async contracts, include both organizationId and projectId in domain-event payloads, topic/task payloads, and workflow inputs by default. Exceptions: MagicLinkEmailRequested, InvitationEmailRequested, UserDeletionRequested, the domain-events topic payload, the magic-link-email topic payload, the invitation-email topic payload, and the user-deletion topic payload.
• When BullMQ delay is the chosen coalescing mechanism, key the delayed job by the logical entity identity so newer writes can replace, extend, or be dropped against the pending job — see the debounce-vs-throttle guidance below for which semantic to pick.
• When a delayed queue topic semantically marks a lifecycle edge, let the delayed task publish a domain event through the appropriate rail after the delay elapses: use OutboxEventWriter / OutboxEventWriterShape for transactional boundaries and direct EventsPublisher publication for non-transactional or high-volume worker flows. Downstream side effects should run from the domain-event consumers rather than inline in the delayed task.

Choosing between debounceMs and throttleMs

PublishOptions exposes two mutually exclusive delay fields. Both accept a window in ms and coalesce repeated publishes against dedupeKey, but they answer different questions.

debounceMs — fires after N ms of quiet on the dedupe key. Each publish within the window pushes the fire time forward and replaces the pending payload (BullMQ extend: true, replace: true). Use when the task should wait for a stream of events to settle.

Example: trace-end:run after TracesIngested. The batch event fans out one publish per deduped trace id; every new publish for the same trace resets the clock, so end-of-trace work fires once that trace is actually idle. If spans keep arriving every few seconds, that means the trace is still active — not firing is correct.

throttleMs — fires at most once per N ms per dedupe key. The first publish schedules the fire time; subsequent publishes within the window are dropped (BullMQ extend: false, replace: false). Requires dedupeKey. Use when you need a hard upper bound on fire latency and a cap on frequency, and where starvation under a continuous publish stream would be a product bug.

Example: annotation-driven alignment refresh (evaluations:automaticRefreshAlignment, 1h) and its escalation (evaluations:automaticOptimization, 8h). We want at most one refresh per evaluation per hour, firing at most 1h after the first new annotation, even if annotations keep arriving every 30 min.

Decision heuristic

Ask: if a publisher fires every 30 min forever on the same dedupeKey, what should happen?

• "Fire after they stop" → debounceMs. Classic debounce. Fire time keeps sliding forward; fires only during quiet periods.
• "Fire every N min regardless" → throttleMs. Bounded latency, bounded frequency; never starves.

Anti-pattern

Reaching for debounceMs when the intent is "run at most once per hour". With a continuous publish stream every publish extends the TTL and the task never fires — silent starvation. If the wording in the spec or PR is "at most once per X" or "every X at most", that is throttle semantics; use throttleMs.

// Debounce — wait for events to settle
pub.publish("trace-end", "run", payload, {
  dedupeKey: `trace-end:run:${traceId}`,
  debounceMs: TRACE_END_DEBOUNCE_MS,
})

// Throttle — at most once per window, bounded latency
pub.publish("issues", "refresh", payload, {
  dedupeKey: `issues:refresh:${issueId}`,
  throttleMs: ISSUE_REFRESH_THROTTLE_MS,
})

Name the constant to match the semantic: *_DEBOUNCE_MS vs. *_THROTTLE_MS. A constant named for one semantic that is passed as the other is a lie readers will trip over.

New infrastructure dependencies

When adding a new external system the product talks to:

1. Add a concrete provider package in packages/platform/*-<provider>.
2. Wire it in the app composition root from environment-driven config.
3. Change domain only if business rules change — not for every new adapter.

For env var naming when wiring config, see env-configuration. For layer rules, see architecture-boundaries.