Queue Package

@activation-sys/queue — BullMQ workers and queues for async processing, backed by Redis.

Installation

Already included in the monorepo workspace.

import { addActivationJob, startAllWorkers, initializeQueues } from '@activation-sys/queue';

Exports

// Redis connection
export { createConnection, redis, disconnect } from './config/redis.js';

// Queues
export { activationQueue, addActivationJob, getActivationJobCounts };
export { paymentQueue, addPaymentJob, getPaymentJobCounts };
export { paymentReconciliationQueue, addPaymentReconciliationJob, closePaymentReconciliationQueue };
export type { PaymentReconciliationJob };
export { storeQueue, addStoreJob, addStoreReconciliationJob, getStoreJobCounts,
  scheduleStoreRecoveryScan, removeStoreRecoveryScheduler };
export { notificationEmailQueue, notificationSmsQueue, addNotificationChannelJob,
  addNotificationChannelJobs, getNotificationChannelJobCounts };
export { notificationOrchestrationQueue, addBatchRunJob, removeBatchRunJob,
  scheduleNotificationHandoffSweep, removeNotificationHandoffSweep };

// Notification options
export { NOTIFICATION_CHANNEL_CONCURRENCY, NOTIFICATION_CHANNEL_LIMITERS,
  NOTIFICATION_JOB_PRIORITY, NOTIFICATION_HANDOFF_SWEEP_DEFAULTS,
  NOTIFICATION_SENDING_STALE_MS }

// Workers
export { startActivationWorker, stopActivationWorker };
export { getCompletionJobState, retryCompletion };
export { startPaymentWorker, stopPaymentWorker };
export { startPaymentReconciliationWorker, stopPaymentReconciliationWorker };
export type { PaymentReconciliationHandler };
export { startStoreWorker, stopStoreWorker, startAllWorkers, stopAllWorkers };
export { startNotificationChannelWorkers, stopNotificationChannelWorkers };
export { startNotificationOrchestrationWorker, stopNotificationOrchestrationWorker };

// Lifecycle
export { initializeQueues, shutdownQueues, closeNotificationQueues };

Notification payload types come from @activation-sys/shared, not the queue package root. They do not extend BaseJob or belong to the QueueJob union. Worker dependency interfaces are defined in the worker source files but are not re-exported at the root. Callers can infer them from Parameters<typeof startNotificationChannelWorkers>[0] and Parameters<typeof startNotificationOrchestrationWorker>[0].

Notification Export	Contract
notificationEmailQueue	Producer for notification-email. Accepts email delivery IDs.
notificationSmsQueue	Producer for notification-sms. Accepts SMS delivery IDs.
notificationOrchestrationQueue	Producer for notification-orchestration. Accepts batch, handoff sweep and callback jobs.
addNotificationChannelJob(job, urgency)	Routes one job by channel. Returns the job ID or undefined. Recovers retained failed jobs in place.
addNotificationChannelJobs(jobs, urgency)	Groups fresh jobs by channel and adds each group in bulk. Returns no value. Does not recover retained failures.
getNotificationChannelJobCounts()	Returns separate email and sms counts for waiting, active, completed, failed, delayed and prioritized.
addBatchRunJob(batchId, runAt)	Takes a Date or null. Returns the job ID or undefined. A future date delays execution. A null or past date is immediate.
removeBatchRunJob(batchId)	Returns whether the delayed job was removed. Returns false on removal errors. Batch status, not job removal, decides cancellation.
scheduleNotificationHandoffSweep(options)	Adds an immediate sweep and registers a recurring sweep. Requires intervalMs, staleAfterMs and batchSize.
removeNotificationHandoffSweep()	Removes the recurring schedule. Returns false on removal errors. Does not remove an already queued sweep.
startNotificationChannelWorkers(deps)	Async startup for both channels. Returns { email, sms }. Cleans up channel workers if startup fails.
stopNotificationChannelWorkers()	Stops both channel workers and closes their owned clients.
startNotificationOrchestrationWorker(deps)	Starts the orchestration worker and returns it. Reuses the running worker on repeated calls.
stopNotificationOrchestrationWorker()	Stops the orchestration worker and closes its owned clients.
closeNotificationQueues()	Closes all three notification producers and their clients after consumers stop.
NOTIFICATION_CHANNEL_CONCURRENCY	{ email: 5, sms: 2 }. Applied per worker.
NOTIFICATION_CHANNEL_LIMITERS	Email uses { max: 30, duration: 60000 }. SMS uses { max: 10, duration: 60000 }. Each queue has its own limiter.
NOTIFICATION_JOB_PRIORITY	{ urgent: 1, normal: 5 }.
NOTIFICATION_HANDOFF_SWEEP_DEFAULTS	{ intervalMs: 60000, staleAfterMs: 900000, batchSize: 200 }.
NOTIFICATION_SENDING_STALE_MS	60000 ms before an unresolved send claim can be reclaimed.

Job Types

ActivationJob

{
  type: 'activation',
  activationId: string,
  userId: string,
  operatorSlug: 'stc' | 'mobily' | 'zain',
  step: 'initiate' | 'verify_identity' | 'complete'
}

PaymentJob

{
  type: 'payment',
  action: 'settle_payment' | 'reconcile_failure',
  paymentId: string,
  orderId: string,
  userId: string,
  gateway: string // Runtime providers: mock, moyasar or paymob; not a payment method
}

PaymentReconciliationJob

Queue moyasar-reconciliation (the original Redis name, kept so queued Moyasar jobs drain across upgrades; the code calls it the payment reconciliation queue), job name reconcile-payment. Webhook routes for Moyasar and Paymob enqueue it; the API's PaymentReconciliationHandler verifies the payment with the named gateway.

{
  provider?: 'moyasar' | 'paymob', // omitted means moyasar
  paymentId: string,
  environment: 'sandbox' | 'production',
  eventId: string,
  eventType: string,
  transactionId?: string // Paymob: the transaction the callback named
}

The job ID is a SHA-256 of [provider (default moyasar), environment, eventId], so existing Moyasar job IDs are unchanged.

NotificationChannelJob

{
  deliveryId: string,
  notificationId: string,
  channel: 'sms' | 'email'
}

NotificationChannelJob is the union of NotificationEmailJob and NotificationSmsJob from the shared package. Redis carries IDs, not recipient addresses or rendered content. Channel workers load the stored rows, claim the send conditionally, check recipient gates and call injected render and send functions.

addNotificationChannelJob(job, urgency) uses deliveryId as the deterministic job ID. Job names are deliver-email and deliver-sms. Priority is always explicit, with 1 for urgent or 5 for normal. A retained failed job is retried in place with both attemptsMade and attemptsStarted reset. A retained completed or live job is not replayed. Bulk adds are for fresh deliveries only. Durable delivery outcomes still prevent repeat sends after queue retention removes an old job.

NotificationOrchestrationJob

type NotificationOrchestrationJob =
  | { kind: 'run_batch', batchId: string }
  | { kind: 'reconcile_handoffs', staleAfterMs: number, batchSize: number }
  | { kind: 'normalise_callback', eventId: string }

Kind	Enqueue Behavior	Injected Hook
run_batch	addBatchRunJob(batchId, runAt) creates run-notification-batch with ID notification-batch-${batchId}. A retained failed job is retried with attemptsMade reset. Live and completed jobs are not replayed.	runBatch(batchId)
reconcile_handoffs	scheduleNotificationHandoffSweep(options) adds reconcile-notification-handoffs immediately with ID startup-notification-reconciliation. It registers recurring ID repeat-notification-reconciliation. Defaults are a 60-second interval, 15-minute handoff cooldown and 200 rows.	reconcileHandoffs({ staleAfterMs, batchSize })
normalise_callback	The API adds normalise-notification-callback with ID notification-callback-${eventId}. Only the stored event ID enters the queue. Callback jobs override default retry and retention options as shown below.	normaliseCallback(eventId)

Orchestration has no provider limiter, so delayed batches and recovery are independent of channel throttling.

Activation Completion Controls

The activation worker exports three completion controls used by the officer console (the fallout queue and the completion tool):

Function	Contract
claimCompletion(activationId, simType, iccid)	Transactionally checks in_progress, completed identity and paid order, then sets completionAttemptedAt. For physical SIMs it locks the serial and applies first-claim-wins. Returns claimed, claimed_elsewhere, serial_in_use with a transition, or skipped with a reason.
getCompletionJobState(activationId)	Reads the fixed ${activationId}-complete job and returns none, waiting, active, delayed, failed, completed, or unknown. Officer actions treat waiting, active, delayed, and unknown as potentially live; a none (no retained job) is not live.
retryCompletion(activationId, { now?, repeatSafe? })	Replays payment settlement only when the activation is in_progress, identity is completed, payment/order are paid, and no live completion job remains. By default the row must also have no completion claim. When repeatSafe is true (the row's operator declares repeated activation requests safe), a set claim no longer refuses the retry: once the job reads as not live, the claim is released with a conditional update before the settlement job is enqueued. A retained failed/completed job is removed before enqueueing a unique settlement job. Returns replayed, in_flight, or not_eligible.

retryCompletion is advisory against a concurrent recovery scan; claimCompletion is the final atomic guard. A completion claim that may have reached the operator is deliberately not cleared by retry logic unless the caller asserts repeatSafe.

initiateActivation receives reference: activationId from the activation worker (processComplete in workers/activation-worker.ts), so an operator plugin that declares capabilities.requestLookup can later look up what became of that same request by this id.

Queue Configuration

Queue	Concurrency	Attempts	Backoff Strategy	Completed Retention	Failed Retention
Activation	10	3	Custom: 1s → 5s → 30s	100 jobs	7 days / 1000
Payment	5	5	Exponential (1s base)	500 jobs	7 days / 5000
Store Fulfillment	5	3	Exponential (1s base)	500 jobs	7 days / 5000
Notification Email	5	5	Exponential (5s base)	500 jobs	7 days / 5000
Notification SMS	2	3	Exponential (5s base)	500 jobs	7 days / 5000
Notification Orchestration	2	5	Exponential (1s base)	500 jobs	7 days / 5000

The table shows queue defaults. Email and SMS concurrency caps apply per worker and are independent. Their queue rate limiters are 30 and 10 jobs per minute respectively. Orchestration has no provider limiter. Attempts include the initial execution.

Lane or Job	attempts	backoff	priority	removeOnComplete	removeOnFail
notification-email default	5	{ type: 'exponential', delay: 5000 }	5	500	{ age: 604800, count: 5000 }
notification-sms default	3	{ type: 'exponential', delay: 5000 }	5	500	{ age: 604800, count: 5000 }
notification-orchestration default	5	{ type: 'exponential', delay: 1000 }	5	500	{ age: 604800, count: 5000 }
API callback normalization job	9	{ type: 'exponential', delay: 10000 }	5, inherited	true	true

Completed retention is a count of 500 jobs, not a time window. Failed retention has both a seven-day age limit and a 5000-job count limit. Callback jobs are removed on completion or failure. Their eight backoff delays span 42.5 minutes. An unmatched callback retries while its receipt is younger than 30 minutes. After that cutoff, normalization records it as processed without changing a delivery.

Workers write retry bookkeeping for failed attempts. They write failed_terminal on exhaustion or unrecoverable failure, including exhausted stalled recovery. Send-time suppression completes without a provider call. Recorded sent, delivered, failed_terminal and suppressed deliveries are not resent. SMTP email records sent because there is no SMTP callback. A crash after SMTP acceptance but before the outcome is recorded can still cause a duplicate on recovery.

await startNotificationChannelWorkers(deps) requires render, send and resolveLocale. Workers access stored rows through the queue package connection. Both channels set maxStartedAttempts to their attempt budget. This bounds processor starts, including deferrals that do not increase attemptsMade. The worker lock is 30 seconds. The injected send promise must settle within 30 seconds. SMTP sends have a 20-second overall response deadline. Deadline expiry returns retryable SMTP_SEND_TIMEOUT. The underlying SMTP operation can still deliver, creating a duplicate window on retry.

A sending claim at least 60 seconds old is reclaimable. A claim with failure bookkeeping at or after its claim timestamp is also reclaimable. Workers read account activity, channel opt-outs and the recipient address after claiming. Transactional categories ignore opt-outs. Workers revalidate the exact claim after rendering and before calling the provider. Claim timestamps fence stale writes. An unresolved live claim delays the job by 60 seconds rather than discarding recovery work. A lost claim whose delivery has completed sends nothing.

startNotificationOrchestrationWorker(deps) requires runBatch, reconcileHandoffs and normaliseCallback. The API injects the batch service, handoff repair service and callback normalization function. The registered callback route verifies and stores a parsed receipt before enqueueing its ID. Normalization correlates by channel, provider and provider message ID. It only advances allowed outcomes and marks the receipt processed. Repeated and backward status reports do not change a delivery.

Each notification worker has a five-second shutdown deadline covering graceful close, pending metrics or bookkeeping, and client quit. Channel shutdown runs both drains in parallel. On timeout, cleanup stops lock renewal and disconnects owned clients without waiting for a hung processor. It does not cancel arbitrary rendering or send work. closeNotificationQueues() closes the three producers in parallel after workers stop. Each queue and client pair has its own five-second deadline.

The handoff sweep selects email and SMS deliveries in queued, plus sending rows whose claim is older than 60 seconds. By default, it locks up to 200 rows without waiting for already locked rows. It updates only queuedAt before enqueueing, leaving the outcome and send claim timestamp unchanged. A null queuedAt is eligible immediately. Otherwise, the default 15-minute handoff cooldown must pass. Failed enqueue attempts remain eligible after that cooldown. Recovery uses normal priority and the single-job helper to recover retained failed jobs.

The sweep then selects unfinished scheduled or running batches whose creation or start is older than the same 15-minute stale window, up to the same 200-row cap. A scheduled batch also needs its scheduled time to be past that window, so a batch waiting for a future time is never re-selected. Each is re-enqueued through addBatchRunJob(batchId, scheduledAt) using the original scheduled time, so a null scheduled time runs immediately. The deterministic notification-batch-{batchId} job ID keeps concurrent sweeps from doubling a run. A missing or throwing re-enqueue is logged by batch ID only and stays eligible for the next run.

Mock provider message capture is restricted to test mode. Runtime notification sends retain no recipient or content history in memory.

Usage Examples

The host supplies channelDeps and orchestrationDeps using the contracts above. This example selects notification workers, but is not an isolated notification-only host: importing the package entry point also opens activation, payment, and store producer clients. startWorkers: false prevents worker startup, not producer construction. shutdownQueues() stops workers, closes notification and payment reconciliation producers, and disconnects the default singleton. It does not close the other lanes' dedicated producer/worker clients. The host must account for those connections separately.

initializeQueues also accepts notificationChannels and notificationOrchestration dependency options. With worker startup enabled, each notification worker group starts only when its matching dependencies are supplied. Activation and payment startup is independent of those options. Sweep registration remains a separate call.

import {
  addNotificationChannelJob,
  getNotificationChannelJobCounts,
  initializeQueues,
  NOTIFICATION_HANDOFF_SWEEP_DEFAULTS,
  scheduleNotificationHandoffSweep,
  shutdownQueues,
  startNotificationChannelWorkers,
  startNotificationOrchestrationWorker,
} from '@activation-sys/queue'

const status = await initializeQueues({ startWorkers: false });
if (!status.connected) throw new Error('Redis unavailable');

await startNotificationChannelWorkers(channelDeps);
startNotificationOrchestrationWorker(orchestrationDeps);
await scheduleNotificationHandoffSweep(NOTIFICATION_HANDOFF_SWEEP_DEFAULTS);

// Delivery rows must exist before their IDs are enqueued.
await addNotificationChannelJob(
  { deliveryId: '...', notificationId: '...', channel: 'email' },
  'normal',
);

// Check job counts
const counts = await getNotificationChannelJobCounts();
// { email: { waiting, active, completed, failed, delayed, prioritized }, sms: { ... } }

// Stops workers, closes notification and payment reconciliation producers, and disconnects the default singleton, not every package client.
await shutdownQueues();

Redis Configuration

• Production: rediss:// with TLS 1.3
• Each queue gets a dedicated connection; BullMQ's prefix produces namespaces such as activation:bull:activation:*, payment:bull:payment:*, notification-email:bull:notification-email:*, notification-sms:bull:notification-sms:*, and notification-orchestration:bull:notification-orchestration:*. Do not set ioredis keyPrefix.
• maxRetriesPerRequest: null required for BullMQ
• Explicitly supplied Redis clients need explicit closure after workers, queues, and queue-event listeners stop. Closing the default singleton alone does not release their connections.