# 03 — Domain Events, Outbox & Jobs

Every mutation in SoloPM emits a domain event. One mechanism feeds five consumers: **audit log, notifications, SSE, Meilisearch sync, outgoing webhooks**. The design is a **hand-rolled transactional outbox in Postgres with a single in-process dispatcher** — no external queue, no River. Rationale in [17-decisions.md](17-decisions.md); the explicit assumption is **one app instance** (`FOR UPDATE SKIP LOCKED` makes even accidental multi-instance mostly safe, but the SSE hub is per-process).

## Emit contract

- [ ] Implement `events.Emit(ctx, tx, Event{Type, ActorID, ProjectID, ResourceType, ResourceID, Payload, ViaWebhook})` which, **inside the caller's transaction**:
  - [ ] Inserts the `audit_log` row (exactly-once, atomic with the mutation — an audit entry can never be lost or orphaned)
  - [ ] Inserts the `events` outbox row
- [ ] `Payload` includes: changed-field map `{field: {prev, new}}`, denormalized display data (issue title, project title, `#N`), and server-parsed `@mention` usernames where applicable ([08-collaboration.md](08-collaboration.md))
- [ ] After commit, the service fires `pg_notify('solopm_events', '')` — best-effort wake-up; polling is the safety net
- [ ] Every service mutation calls `Emit`; a mutation without an event is a spec violation

## Event taxonomy

Format `resource.action`, past tense. `action_type` in audit_log uses the same strings.

- [ ] `project.created` / `project.updated` / `project.status_changed` / `project.deleted`
- [ ] `project.member_added` / `project.member_removed` / `project.role_changed` / `project.ownership_transferred`
- [ ] `project.dependency_added` / `project.dependency_removed`
- [ ] `project.exported` (audit-only; no notification/SSE/search/webhook fan-out)
- [ ] `label.created` / `label.updated` / `label.deleted`
- [ ] `epic.created` / `epic.updated` / `epic.status_changed` / `epic.deleted`
- [ ] `issue.created` / `issue.updated` / `issue.status_changed` / `issue.assigned` / `issue.unassigned` / `issue.moved` (rank/epic/parent changes)
- [ ] `issue.blocker_added` / `issue.blocker_removed`
- [ ] `issue.label_added` / `issue.label_removed`
- [ ] `comment.created` / `comment.updated` / `comment.resolved` / `comment.unresolved` / `comment.deleted`
- [ ] `attachment.uploaded` / `attachment.deleted`
- [ ] `link.added` / `link.removed`
- [ ] `wiki.created` / `wiki.updated` / `wiki.deleted`
- [ ] `user.created` / `user.updated` / `user.deleted`
- [ ] `auth.login_succeeded` / `auth.login_failed` / `auth.token_created` / `auth.token_revoked`
- [ ] `webhook.received` / `webhook.processed` / `webhook.failed` (incoming-webhook lifecycle)

## Dispatcher

- [ ] One goroutine started in `main`, stopped on shutdown: `LISTEN solopm_events` + 2-second poll fallback
- [ ] Claims batches: `SELECT … FROM events WHERE processed_at IS NULL ORDER BY id LIMIT 50 FOR UPDATE SKIP LOCKED`
- [ ] For each event, run the fan-out router (below), then `UPDATE events SET processed_at = now()`
- [ ] Crash mid-fan-out → event re-processed on restart; **all consumers must be idempotent** (see guarantees table)
- [ ] `attempts` incremented per processing try; events failing 8 times are marked processed with an error log (never wedge the queue)

## Consumers & guarantees

| Consumer | Mechanism | Guarantee |
|---|---|---|
| Audit log | synchronous inside `Emit` (same tx) | exactly-once |
| Notifications | inline in dispatcher → `notifications` inserts + `send_email` jobs | at-least-once, deduped |
| SSE | inline publish to in-memory hub | best-effort (clients refetch on reconnect) |
| Meilisearch | `search_sync` job | at-least-once, idempotent upsert/delete by doc id |
| Outgoing webhooks | delivery row + `webhook_delivery` job | at-least-once; UNIQUE(webhook_id, event_id) dedupes |

- [ ] **Notifications consumer**: compute recipients per the rules in [09-notifications-realtime.md](09-notifications-realtime.md); skip the actor themself; insert `notifications` rows (dedupe: skip if an identical `(user_id, type, resource, event)` row was already created for this event id); for each recipient with `email_mode = instant`, enqueue a `send_email` job; `daily_digest` users are handled by the nightly digest job
- [ ] **SSE consumer**: build the envelope ([09-notifications-realtime.md](09-notifications-realtime.md)) and publish to the hub; hub delivers `data` frames only to connected users who are members of `event.project_id`
- [ ] **Search consumer**: map event → `{index, doc_id, op: upsert|delete}` per [10-search.md](10-search.md); enqueue `search_sync` job; skip event types with no search impact
- [ ] **Webhook consumer**: for each active `outgoing_webhooks` row of the project whose `event_types` is empty or contains the event type: insert `outgoing_webhook_deliveries` (pending) + `webhook_delivery` job

## Job queue

- [ ] Worker pool: 3 goroutines polling `jobs` every 1s: `SELECT … WHERE status = 'pending' AND run_at <= now() ORDER BY run_at LIMIT 10 FOR UPDATE SKIP LOCKED`, set `running`, execute, set `done`
- [ ] On failure: `attempts++`, `last_error` recorded, `run_at = now() + 30s × 2^attempts` (capped at 1h), back to `pending`
- [ ] After `max_attempts` (8): status `dead`; dead jobs visible in admin (and webhook deliveries surface as `failed` in the delivery log)
- [ ] Job kinds: `send_email`, `webhook_delivery`, `search_sync`, `daily_digest` (scheduled nightly), `cull` (scheduled daily; see [12-retention.md](12-retention.md))
- [ ] Scheduled jobs: on startup and every hour, ensure the next `daily_digest` and `cull` jobs exist (idempotent upsert by kind + date)
- [ ] Email and webhook delivery never block an HTTP request; search sync typically lands < 2s after the mutation