Skip to content

background-processing.md

Latest commit

 

History

History
191 lines (142 loc) · 10.5 KB

background-processing.md

File metadata and controls

191 lines (142 loc) · 10.5 KB
Raw

Background Processing

Overview

Unitae uses a Redis-based background job processing system built on BullMQ. A single multi-queue worker process handles six job types: territory data sync, email notifications, PDF thumbnail generation, data transfer (export/import), data retention (daily auto-anonymisation), and the campaign lifecycle (daily date-driven activation/end of publishing campaigns). Jobs carry congregationId to maintain tenant isolation.

Architecture

Web Pod                                    Worker Pod (workers/worker.server.ts)
┌──────────────────────────────┐           ┌─────────────────────────────────────────┐
│  Route / Cron Action         │           │  syncWorker           (concurrency 1)   │
│                              │           │    → handleSyncWork()                   │
│  syncQueue.add(...)          │── Redis ─▶│                                         │
│  emailQueue.add(...)         │── Redis ─▶│  emailWorker          (concurrency 5)   │
│  thumbnailQueue.add(...)     │── Redis ─▶│    → handleEmailWork()                  │
│  dataTransferQueue.add(...)  │── Redis ─▶│                                         │
│                              │           │  thumbnailWorker      (concurrency 2)   │
└──────────────────────────────┘           │    → handleThumbnailWork()              │
                                           │                                         │
                                           │  dataTransferWorker   (concurrency 1)   │
                                           │    → handleDataTransferWork()           │
                                           │                                         │
                                           │  retentionWorker      (concurrency 1)   │
                                           │    → handleRetentionWork()              │
                                           │    ▲ daily cron @ 03:00 UTC             │
                                           │      (upsertJobScheduler)               │
                                           │                                         │
                                           │  Health: :9090                          │
                                           └─────────────────────────────────────────┘

Queue Registry

Queue names are centralized in app/shared/infra/queues.server.ts:

export const QUEUE_NAMES = {
  sync: 'syncQueue',
  email: 'emailQueue',
  thumbnail: 'thumbnailQueue',
  dataTransfer: 'dataTransferQueue',
  retention: 'retentionQueue',
  campaign: 'campaignQueue',
} as const

Queues

Sync Queue

Imports and processes open data (BANO addresses) for territory management.

  • Producer: app/features/territories/server/sync-queue.server.ts
  • Handler: app/features/territories/jobs/handle-sync-work.server.ts
  • Concurrency: 1 (CPU/IO-intensive import)
  • Retries: 3 attempts, exponential backoff (10s base)
  • Tenant isolation: Uses withScope(congregationId, ...) for RLS-scoped DB access
  • Job data: { userId, congregationId }

Email Queue

Sends notification emails asynchronously with automatic retries.

  • Producer: app/shared/infra/email-queue.server.ts
  • Handler: app/features/notifications/jobs/handle-email-work.server.tsx
  • Concurrency: 5 (IO-bound Resend API calls)
  • Retries: 3 attempts, exponential backoff (5s base)
  • Tenant isolation: Uses unscopedDb with explicit congregationId filtering
  • Locale: Wraps email rendering in runInWorkerContext(congregation.locale, congregation.timezone, ...) for i18n

Job types (discriminated union on type):

  • notification-digest: Batched notification email after the debounce window settles
  • notification-instant: Immediate notification email (no debounce)

Thumbnail Queue

Generates PDF thumbnails asynchronously after document upload.

  • Producer: app/features/display-board/server/thumbnail-queue.server.ts
  • Handler: app/features/display-board/jobs/handle-thumbnail-work.server.ts
  • Concurrency: 2 (CPU-bound pdftoppm subprocess)
  • Retries: 3 attempts, exponential backoff (5s base)
  • Job data: { congregationId, documentId, pdfStorageKey }
  • Flow: Fetch PDF from storage → run pdftoppm → upload thumbnail → update document record

Documents are created with thumbnailUri: null and updated asynchronously when the worker completes.

Data Transfer Queue

Handles congregation data export and import as background jobs.

  • Producer: app/features/settings/server/data-transfer-queue.server.ts
  • Handler: app/features/settings/jobs/handle-data-transfer-work.server.ts
  • Concurrency: 1 (IO-intensive archive creation/extraction)
  • Retries: None (1 attempt only)
  • Tenant isolation: Uses withScope(congregationId, ...) for RLS-scoped DB access

Job types (discriminated union on type):

  • export: Creates a .unitae archive (ZIP) with congregation data and optional uploaded files
  • import: Extracts a .unitae archive and imports entities with ID remapping

Retention Queue

Runs the daily data-retention sweep that auto-anonymises members who left the congregation longer ago than the retention window.

  • Producer: app/features/settings/server/retention-queue.server.ts
  • Handler: app/features/settings/jobs/handle-retention-work.server.ts
  • Concurrency: 1
  • Retries: None (1 attempt only)
  • Schedule: A single repeating job registered at worker startup via retentionQueue.upsertJobScheduler('retention-daily', { pattern: '0 <RETENTION_CRON_HOUR_UTC> * * *', tz: 'UTC' }) — runs daily at 03:00 UTC by default. upsertJobScheduler is idempotent, so restarts don't pile up duplicate schedules.
  • Behaviour: Iterates every active congregation and calls autoAnonymizeRetentionCandidates(db, congregationId, DEFAULT_RETENTION_MONTHS, ...), which anonymises members whose leftAt is older than the window. Group responsibles are skipped with a warning (they must be reassigned by an admin first).

This is distinct from the /cron/retention HTTP endpoint (expired-token / withdrawn-consent cleanup), which is triggered by an external scheduler — see Cron Jobs.

Campaign Lifecycle Queue

Runs the daily date-driven pass over publishing campaigns: activates campaigns whose start day has arrived and ends campaigns whose inclusive end date is fully past, applying each campaign's configured start/end transitions (pause/close regular attributions, auto-reassign, auto-close campaign attributions, resume).

  • Producer: app/features/territories/server/campaign-queue.server.ts
  • Handler: app/features/territories/jobs/handle-campaign-lifecycle-work.server.ts
  • Concurrency: 1
  • Retries: None (1 attempt only — transitions are idempotent and the next tick converges)
  • Schedule: A repeating job registered at worker startup via campaignQueue.upsertJobScheduler('campaign-lifecycle-daily', { pattern: '0 <CAMPAIGN_CRON_HOUR_UTC> * * *', tz: 'UTC' }) — daily at 02:00 UTC by default, deliberately an hour before the retention sweep. The worker also enqueues one catch-up sweep at boot so a worker that was down over a start/end date converges immediately.
  • Behaviour: Iterates every active congregation with per-tenant error isolation and runs runCampaignLifecycleSweep, which delegates to the campaign lifecycle workflow (campaign-lifecycle.workflow.ts) as a synthetic system actor. Saving a campaign create/edit form also runs the sweep inline for that congregation, so a campaign that is already due activates without waiting for the cron.

Worker Locale Support

Background emails must render in the congregation's language. The worker uses AsyncLocalStorage via app/shared/utils/worker-locale.server.ts:

import { runInWorkerContext } from '~/shared/utils/worker-locale.server'

// In email handler:
await runInWorkerContext(congregation.locale, congregation.timezone, async () => {
  // All Paraglide m.*() calls resolve to the correct locale
  await mailer.emails.send({ subject: m.email_subject(), ... })
})

This module is imported at the top of workers/worker.server.ts before any handler imports, ensuring overwriteGetLocale() is called once at startup.

Worker Health & Lifecycle

The unified worker (workers/worker.server.ts) manages all six queue workers:

  • Health endpoint: HTTP server on port UNITAE_WORKER_HEALTH_PORT (default 9090)
  • Ready check: Returns 200 only when ALL workers have fired ready and none are closing
  • Graceful shutdown: On SIGINT/SIGTERM, closes the health server then calls worker.close() on all workers via Promise.allSettled

Development

# Start Redis
docker compose -f docker/docker-compose.dev.yml up -d

# Start worker (separate terminal)
pnpm start:worker

# Start web app
pnpm start:dev

The worker is only needed if you're working on territory sync, document uploads, or board notifications. Most development can be done without it.

Deployment

  • Worker runs as a separate process from the web server
  • Same Docker image, different command: pnpm start:worker
  • Health endpoint on port 9090 for K8s liveness/readiness probes
  • Scales independently from web processes

Adding New Job Types

  1. Add queue name to QUEUE_NAMES in app/shared/infra/queues.server.ts
  2. Create queue producer: app/features/{feature}/server/{name}-queue.server.ts
  3. Create handler: app/features/{feature}/jobs/handle-{name}-work.server.ts (handlers live in a per-feature jobs/ directory, separate from server/)
    • For scoped DB access: use withScope(congregationId, ...)
    • For cross-tenant queries: use unscopedDb with explicit where: { congregationId }
    • For email rendering: wrap in runInWorkerContext(congregation.locale, congregation.timezone, ...)
  4. Register worker in workers/worker.server.ts with appropriate concurrency
  5. No new K8s deployment needed — the unified worker handles all queues

Related