Queues

Overview

We use BullMQ backed by Redis for async processing. Nine queues handle the core async workflows.

Queue Configuration

Queue	Name	Location	Purpose
Activation	activation	packages/queue	Drives operator activation steps (initiate, verify_identity, complete)
Payment	payment	packages/queue	Settles completed payments by enqueuing activation handoff
Notification Email	notification-email	packages/queue	Renders and sends persisted email deliveries
Notification SMS	notification-sms	packages/queue	Renders and sends persisted SMS deliveries through the mock provider
Notification Orchestration	notification-orchestration	packages/queue	Dispatches batch, handoff reconciliation, and callback-normalization hooks
Store Fulfillment	store-fulfillment	packages/queue	Issues eSIM profiles for paid store orders
Package Sync	package-sync	apps/api	Periodic sync of telecom operator packages
eSIM Product Sync	esim-product-sync	apps/api	Periodic sync of eSIM provider catalogs
FX Refresh	fx-refresh	apps/api	Scheduled fetch of USD exchange rates from the configured provider

The packages/queue queues use shared constants from @activation-sys/shared. The apps/api queues configure attempts, backoff, removeOnComplete, and removeOnFail directly at queue creation time, and share the API's Redis singleton without a BullMQ prefix.

For the six packages/queue queues, connections do not use ioredis keyPrefix, because BullMQ requires unprefixed Redis connections. Their namespaces are set with BullMQ's prefix option through getQueuePrefix(queueName), such as activation:bull:activation:* and notification-email:bull:notification-email:*. Supabase Realtime carries activation status changes independently of these queues.

Usage

import { addActivationJob, addPaymentJob, addNotificationChannelJob, addBatchRunJob } from '@activation-sys/queue';

// Add an activation job
await addActivationJob({
  type: 'activation',
  jobId: 'activation-id-initiate',
  createdAt: new Date().toISOString(),
  activationId: '...',
  userId: '...',
  operatorSlug: 'stc',
  step: 'initiate',
});

// Add a payment job
await addPaymentJob({
  type: 'payment',
  action: 'settle_payment',
  jobId: 'payment-id-settle',
  createdAt: new Date().toISOString(),
  paymentId: '...',
  orderId: '...',
  userId: '...',
  gateway: 'stc_pay',
});

// Enqueue only after notification and delivery rows have been persisted.
await addNotificationChannelJob(
  { deliveryId: '...', notificationId: '...', channel: 'email' },
  'normal',
);

// Schedule an existing batch. Null runs immediately.
await addBatchRunJob('...', null);

Notification Queues

The former shared stub and API-local status lane are removed. Use the notification orchestration service for domain events rather than constructing channel jobs in a route. The low-level example above assumes matching notification and delivery rows already exist. In-app is persisted directly as delivered, with no queue. Push is disabled and produces no delivery row.

Email and SMS workers claim a delivery before reading current account activity, non-transactional opt-out, address, and locale. A blocked gate writes suppressed and completes without sending. Rendering uses injected functions and preserves attachments. Before dispatch, the worker checks that the claim still belongs to this attempt. Recorded sent, delivered, failed_terminal, and suppressed outcomes are skipped.

Queue / Job	Concurrency Per Worker	Queue Rate Limit	Attempts	Exponential Base	Retention
notification-email	5	30 jobs / 60 seconds	5	5 seconds	500 completed, failed up to 7 days or 5,000 jobs
notification-sms	2	10 jobs / 60 seconds	3	5 seconds	500 completed, failed up to 7 days or 5,000 jobs
notification-orchestration defaults	2	None	5	1 second	500 completed, failed up to 7 days or 5,000 jobs
normalise_callback override	Shares orchestration worker	None	9	10 seconds	Removed on completion or failure

These values come from the shared retry constants, notification queue definitions, and enqueueNotificationCallback() in the API. Channel concurrency is independent, not a combined ceiling. Additional API processes start additional workers. The HTTP request counter is separate and in memory per API process, not a delivery limiter.

Every notification job has an explicit priority. Channel producers map urgent to 1 and normal to 5. Batch, sweep, and callback jobs use 5. Urgency changes priority only, with no quiet-hours policy.

Deduplication and Retries

The delivery ID is the channel job ID. addNotificationChannelJob() retries a retained failed job in place, resetting both failed-attempt and started-attempt counters, but leaves completed and live jobs alone. addNotificationChannelJobs() groups fresh jobs by channel for bulk enqueue and must not be used to recover retained failures. Notification rows have a separate durable recipient/occurrence dedupe key, which survives Redis job removal.

Retryable send failures keep the outcome at sending and record failure bookkeeping before the next retry is scheduled. A recorded failure releases that attempt's claim for retry. Otherwise a claim becomes stale after 60 seconds. An attempt encountering a live claim is delayed rather than completed. Writes are fenced by the claim timestamp so an old failed attempt cannot overwrite a newer result.

Unrecoverable failures, exhausted attempts, and exhausted stalled recovery record failed_terminal in Postgres. Channel workers set maxStartedAttempts to 5 for email and 3 for SMS, bounding repeated starts even when a control-flow deferral does not increment the failed-attempt count. A retained failed Redis job is not itself the durable terminal record, and retrying it does not reset a terminal delivery row.

Handoff Reconciliation

Startup schedules an immediate reconcile_handoffs job and a recurring run every minute. NOTIFICATION_HANDOFF_SWEEP_DEFAULTS sets a 15-minute handoff age and a maximum of 200 rows per run. The sweep selects email/SMS delivery rows, oldest first, when both conditions hold:

• The outcome is queued, or it is sending with a claim older than 60 seconds.
• queuedAt is null, or the last handoff stamp is older than 15 minutes. An unstamped queued row need not be 15 minutes old.

The claim uses row locks with SKIP LOCKED and stamps queuedAt before enqueueing. It leaves the outcome and worker claim timestamp unchanged. This repairs a crash between persistence and enqueue, a lost Redis job, or an abandoned send claim. Overlapping scans cannot claim the same locked row, and deterministic job IDs deduplicate overlapping handoffs. A failed sweep enqueue remains eligible again after the handoff age passes.

Recovery uses the single-add helper with normal urgency. A missing job is recreated at priority 5, while a retained job keeps its existing priority. The sweep does not resend successful, terminally failed, or suppressed deliveries, and does not create notifications for a domain event that never reached the trigger. A retained completed job is left alone even if its delivery row is still pending, so the sweep is not a universal repair for inconsistent records.

After the delivery claim, the sweep also selects unfinished batches whose scheduled creation or running start is older than the same 15-minute stale window, oldest first, up to the same 200-row cap. A scheduled batch also needs its scheduled time to be past that window, so a healthy batch waiting for a future time is never re-selected and cannot crowd out an overdue one. Each is re-enqueued with addBatchRunJob(batchId, scheduledAt): a null scheduled time runs immediately, and the deterministic batch job ID keeps concurrent sweeps from doubling a run. Missing or throwing re-enqueues are logged by batch ID only and stay eligible next run. This repairs a batch whose worker died mid-run or whose delayed run_batch job was lost.

Last, the sweep selects unprocessed callback receipts, oldest received first, up to the same 200-row cap, when either the enqueue stamp is missing or it is older than the same 15-minute stale window. The first case is an ingestion request that ended between persisting the receipt and queueing it, the second a callback job that was lost or that exhausted its nine attempts. Nothing else re-drives those receipts, because a provider redelivering the same event is answered as a duplicate once the stamp is set. Each receipt is re-added under its deterministic notification-callback-{eventId} job ID, which is a no-op while a live job holds it and free again once a failed job is removed, and the stamp is rewritten so the next sweep skips the receipt for a full window. A failed re-add is logged by receipt ID only and stays eligible next run. Normalization marks an unmatched receipt processed once it is past its correlation cutoff, so a receipt is not re-driven indefinitely.

Batches and Callbacks

run_batch loads a persisted batch and calls the trigger in recipient chunks. Future batches use a delayed job with ID notification-batch-{batchId}. A past schedule runs immediately, and a delayed job may start later than requested. A batch becomes finished when notification persistence completes, even if fullyEnqueued is false. Delivery success is a separate aggregate, not the batch status.

Cancellation conditionally changes a scheduled batch to cancelled. A running or finished batch cannot be cancelled. Removing its Redis job is best-effort cleanup, not the cancellation decision. The runner checks persisted status before doing work.

normalise_callback carries only the persisted receipt ID. Callback ingestion verifies the signature, persists normalized fields, and stamps the enqueue only after success. A 503 asks the sender to retry the same event, which can repair an unstamped receipt. Normalization correlates by channel, provider, and provider message ID and never moves an outcome backwards. An unmatched receipt is retried until it is at least 30 minutes old, then marked processed without changing a delivery. These callback jobs override the orchestration defaults as shown above.

Provider Limits

Every notification channel sends through its service's active provider (see Service providers): email through SMTP, OneSignal or the mock; SMS through Taqnyat, OneSignal, Twilio or the mock; push through OneSignal or the mock. A selected provider without sendNotification produces suppressed with reason provider_unsupported, a push with no active provider does the same, and a push to an account with no signed-in device produces suppressed with reason no_address. SMTP email ends at sent, not delivered.

The SMTP notification adapter imposes a 20-second overall send deadline and closes its transport, below the channel worker's 30-second lock duration. A relay accepting mail before a crash or lost acknowledgement can still cause a duplicate on retry. A closed connection cannot retract accepted mail and SMTP supplies no idempotency operation. Do not promise exactly-once email delivery.

The renderer supports eSIM QR attachments and the automatic activation and store eSIM email hooks trigger through the orchestration queue with a dedupe key per issued profile. See notification architecture for the current producer boundaries and notification data lifecycle for retention and export behavior.

FX Refresh Queue

The fx-refresh queue fetches USD exchange rates from the configured provider. It uses 3 attempts with exponential backoff starting at 5000ms and concurrency of 1. A repeat scheduler runs every FX_REFRESH_INTERVAL_MS plus a startup job triggers an immediate fetch. The queue is gated by FEATURE_ESIM_STORE. A store-disabled boot also clears any repeatable scheduler left in Redis by an earlier store-enabled boot (both the FX and eSIM sync lanes), and logs a warning if that cleanup fails so a leftover scheduler is visible. Transient provider fetch failures throw so the queue retries. Validation and jump-guard rejects keep stored rates and complete the job.

Catalog Sync Engine

One shared engine (catalog-sync-engine.ts) drives both the eSIM product sync and the local package sync. Each lane provides a thin adapter around the engine. Every provider or operator gets its own DB transaction. One owner failing does not stop the rest. When all owners fail the engine throws so the queue retries.

Store Fulfillment State Machine

The store worker drives a provider-agnostic issuance lifecycle on esim_store_orders.issuance_state. It stamps in_flight before every provider call and persists the provider's own order reference as soon as it is known. A retry that finds in_flight, reconciliation_required, or pending_verification goes through the plugin's reconcile capability instead of calling issue again — the double-billing guard for providers without idempotency keys.

Classified provider failures map to queue behavior. Transient errors reset to pending and rethrow so the 3-attempt backoff engages — the failure reason is recorded only on the final attempt. Out-of-stock and terminal errors mark the order failed and stop retries with an unrecoverable error. Ambiguous results park the order as reconciliation_required. An unclassified throw keeps in_flight so the next attempt reconciles. A pending_verification outcome records the customer verification URL and completes the job without completing the order.

Every write the worker makes to issuance_state, error_code or failed_reason goes through transitionStoreOrder from the database package, keeping the condition each write has always used. When a write changes the state or the error code, the helper adds a row to store_order_status_events in the same transaction, with source system and a short note naming what happened and which run did it: fresh issuance, or reconcile followed by the trigger when it is known (a provider webhook, the webhook fallback check, the verification deadline check, the stalled order scan, or a retry or requeue). The in_flight stamp still commits before the provider call. The issued write shares one transaction with the order's move to completed, locking the store order row before the order row.

The eSIM catalog sync worker also refreshes each registered provider's health reading after every run — the data behind the admin providers list and the checkout circuit breaker.

Payment Queue Deduplication

Re-adding a job retries a retained failure in place. Settlement jobs also support replay after completion so a failed downstream handoff can recover; other completed payment actions remain deduplicated.

Worker Lifecycle

apps/api/src/index.ts starts workers through startup-workers.ts inside the API process after Redis connects, except in NODE_ENV=test. Payment, activation, and store workers run alongside notification email, SMS, and orchestration workers. Startup injects operator/provider managers, notification rendering and dispatch, and orchestration hooks, then schedules handoff reconciliation. A business worker or store-recovery startup failure stops workers and closes the three notification producers. A notification startup failure stops only the notification workers and leaves the producers open, leaving business workers and store recovery running.

Shutdown stops workers before closing notification producers so active orchestration can finish enqueueing. Notification workers and producer clients have five-second cleanup bounds with forced disconnection when needed. Activation, payment, and store drains are waited for without a shutdown deadline. stopAllWorkers() does not close producers by itself. closeNotificationQueues() closes all three notification queues and their supplied clients. Notification metrics use worker events, not separate QueueEvents listeners. The older non-notification queue clients are not covered by this close helper.

The API lifecycle wrapper provides the fully wired application startup:

import { startInProcessWorkers, stopInProcessWorkers } from './startup-workers.js';

await startInProcessWorkers();

// Graceful shutdown
await stopInProcessWorkers();

For a notification-only host, see the queue package example. It supplies channel and orchestration dependencies explicitly rather than starting a channel worker without rendering and provider dispatch.

Redis Requirements

• Redis 7+ with TLS 1.3 in production (rediss://)
• Each packages/queue queue gets a dedicated connection with a BullMQ prefix via getQueuePrefix(queueName). The apps/api queues share the API's Redis singleton, unprefixed
• maxRetriesPerRequest: null required for BullMQ compatibility