solopm-server/plans/03-domain-events.md

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; 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)
• 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; 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) 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; 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)
• 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