README.mdapps/worker/README.md← worker5,017 B← Back to the app

@factory/worker

Headless background worker for the AI Software Factory. No HTTP port — it talks to Redis (BullMQ + domain-event streams), Postgres, OpenSearch, S3-compatible storage, SMTP, ClamAV and the AI endpoint.

Queues

Queue	Purpose
`files`	Storage pipeline: S3 head/get + sha256, magic-byte MIME check, ClamAV scan, metadata (image dims, pdf pages), `File.status` lifecycle, audit
`search-index`	Explicit index ops: `reindex` (workspace-scoped, 500-doc bulk batches), `reconcile` (PG↔OS drift), `delete`
`webhooks`	Outbound webhook dispatch (row-based via `Webhook` + direct URL); retry-safe via BullMQ attempts
`automations`	Automation run jobs (condition eval → executors) + schedule repeatables
`email`	Outbound mail via the built-in dependency-free SMTP client (EHLO/STARTTLS/AUTH LOGIN/dot-stuffing). No `SMTP_HOST` → dev-safe no-op
`ai`	`embed` (OpenAI-compatible `/embeddings` → vector field + `AiEmbedding`) and `generate` (`/chat/completions` → `AiMessage`). No `AI_API_KEY` → dev-safe no-op
`maintenance`	Repeatables: retention (daily 03:00, dry-run default), search reconcile (weekly Sun 02:30), webhook DLQ report (hourly), session/token/invite/key cleanup (hourly), workspace resync (ad-hoc)

Event → queue mapping

Domain events are published by the API on the Redis stream factory:events as {event, payload (JSON), ts}. Each queue has its own consumer group so queues lag independently.

Event	search-index consumer	webhooks consumer	automations consumer
`page.created` / `page.updated` / `page.shared`	✅ index `factory_pages`	✅ fan-out	`page.created` → PAGE_CREATED
`page.deleted`	✅ delete (permanent) / mark trashed	✅	—
`database.created` / `database.updated`	✅ `factory_databases`	✅	—
`record.created` / `record.updated`	✅ `factory_records`	✅	✅ RECORD_CREATED / RECORD_UPDATED (database + property filters)
`comment.created`	✅ `factory_comments`	✅	✅
`form.submitted`	—	✅	✅ FORM_SUBMITTED
`file.uploaded`	✅ `factory_files`	✅	—
`automation.executed`	—	✅	— (depth-guarded)
`api_key.created`, `user.login`, `user.register`, `permission.changed`, `workspace.created`	`permission.changed` only	✅ fan-out	—

permission.changed is handled simplified: targeted reindex when the payload names a single resource, otherwise a workspace resync is enqueued on maintenance.

Inbound webhooks are handled by apps/api — this worker is outbound only.

Running

# from the monorepo root
pnpm install
cp apps/worker/.env.example apps/worker/.env   # defaults are dev-safe
pnpm --dir apps/worker dev          # tsx watch src/index.ts
pnpm --dir apps/worker test         # vitest
pnpm --dir apps/worker typecheck
pnpm --dir apps/worker build && pnpm --dir apps/worker start

Docker (headless, workspace note in the Dockerfile header):

docker build -f apps/worker/Dockerfile -t factory-worker .
docker run --env-file apps/worker/.env factory-worker

Set WORKER_ID uniquely per replica. Liveness: the worker writes worker:heartbeat:<WORKER_ID> (TTL 90s, every 30s) so api /health/deep can check without a port.

Retry / delivery semantics

Every BullMQ job: 8 attempts, exponential backoff base 5s ×2 (delays 5s→320s ≈ 5 min cap), removeOnComplete 1000, removeOnFail false, UUID job ids.
Stream consumers are at-least-once: XACK on success, failure leaves the entry pending and the XAUTOCLAIM loop (60s, idle > 60s, 20/pass) redelivers it. All handlers are written to be idempotent.
Webhooks: non-2xx/timeout throws → BullMQ retry; final failure → WebhookDelivery.status=FAILED (attempts=8) + admin notification. Test sends: {webhookId, test:true} → event webhook.test.
Automations: run lifecycle PENDING→RUNNING→SUCCEEDED/FAILED/DEAD_LETTER, one AutomationLog per step, automation.executed emitted on success. Self-trigger guard: emitted events carry {automationId, depth}; the consumer drops events from the same automation or deeper than 5.

Known gaps (documented)

Thumbnails: images get metadata.needsThumbnail=true; actual rasterization needs sharp — plugs into the sharp-based thumbnail task (Phase 9 task 3).
Media duration: video duration is null (mp4 moov atom parsing not implemented); pdf page count is a regex approximation that misses compressed object streams.
Retention file sweep deletes File rows but S3 object deletion is delegated to bucket lifecycle policy (no object deletion in the sweep).