Backend orchestrator workflows

Integration refresh: Social integrations that expose refreshCron need a supervisor that waits until the stored access token is close to expiry, then calls the provider refresh path and updates the database. By default the API can run this supervisor in the same Node process using the flowcraft package. You can instead enqueue the same blueprint on BullMQ so a separate worker process executes it (distributed execution, BullMQ adapter).

Notification email and scheduled social posting use the same model: orchestratorFlows defaults plus optional ORCHESTRATOR_* transport overrides, with dedicated Flowcraft blueprints and workers when you choose transport: “bullmq”.

Where the code lives

Integration refresh (OAuth token supervisor)

• Blueprint and tick logic: orchestrator/flows/refreshTokenWorkflow.ts
• Activity-style steps (timeout + retries on failure): orchestrator/activities/integrationRefreshActivities.ts
• OAuth completion starts the supervisor from RefreshIntegrationService.startRefreshWorkflow (fire-and-forget).

Notification email

• Blueprint: orchestrator/blueprints/notificationEmailBlueprint.ts (plain send + digest flush paths).

Scheduled social posts (calendar)

• Enqueue and in-process / BullMQ entry: orchestrator/flows/scheduledSocialPostWorkflow.ts (blueprint scheduled-social-post, one publish pass per org/post group)
• Node implementation: orchestrator/nodes/scheduledSocialPostNodes.ts; publish path: orchestrator/activities/scheduledSocialPostActivities.ts
• If the worker is down, optional rescan re-lists QUEUE posts that should have published and re-enqueues: orchestrator/flows/missingScheduledPostReconciliation.ts (interval from orchestratorFlows.scheduledSocialPost.missingPostRescanIntervalMs)

Shared

• Barrel exports: orchestrator/index.ts
• Per-flow transport, queue names, and feature flags: backend/config/orchestratorFlows.ts (defaults in TypeScript). Optional runtime overrides (without editing that file): ORCHESTRATOR_INTEGRATION_REFRESH_TRANSPORT, ORCHESTRATOR_NOTIFICATION_EMAIL_TRANSPORT, and ORCHESTRATOR_SCHEDULED_SOCIAL_POST_TRANSPORT — each in_process or bullmq — merged in backend/config/GlobalConfig.ts.

Activity-style resilience

Loads and token refresh run through helpers that mirror common durable-activity settings: 10 minutes max per attempt, 3 attempts, 2 minutes initial delay between attempts, backoff coefficient 1 (fixed delay). Retries apply when the step throws or hits the timeout; a normal false from refresh (provider failure handled inside the service) is not retried.

Behavior (one iteration)

Each loop iteration:

1. Loads the integration by organization and id; stops if it is missing, soft-deleted, mid–OAuth step, or marked refresh_needed.
2. Computes milliseconds until token_expiration; stops if there is no expiry or the token is already expired.
3. Sleeps for that duration (chunked so individual timers stay within the 32-bit setTimeout limit).
4. Reloads the row and re-applies the same guards.
5. Runs refresh (provider token exchange + upsert). Stops if refresh fails.

The graph uses Flowcraft’s loop construct: a begin node enters the loop controller; the body is a single tick node that performs the steps above. Continuation is driven by the workflow key loopShouldContinue (the loop controller evaluates conditions against flat serialized state, not a nested context.* path).

Limits vs an external workflow engine

The default in-process transport does not survive API restarts: if the API redeploys during a sleep, the supervisor for that integration is gone until the next OAuth connect or manual trigger.

With transport: bullmq in backend/config/orchestratorFlows.ts, run state and the job queue live in Redis, and you run a worker process for each flow you enable:

• Integration refresh — pnpm orchestrator:dev:worker:integration-refresh-bullmq locally (or pnpm worker:integration-refresh-bullmq under backend/); production pnpm railway:orchestrator:start:integration-refresh after pnpm railway:orchestrator:build, on an always-on host such as Railway.
• Notification email (when that transport is bullmq) — pnpm railway:orchestrator:start:notification-email in production; local pnpm orchestrator:dev:worker:notification-email-bullmq.
• Scheduled social posts (when that transport is bullmq) — pnpm railway:orchestrator:start:scheduled-social-post in production (Railway); local pnpm orchestrator:dev:worker:scheduled-social-post-bullmq.

For integration refresh, that improves durability across API deploys, but each tick still performs a long in-node sleep while a worker job is active (see Flowcraft pause/sleep guidance). Tune BullMQ concurrency and monitor queue depth accordingly.

Managed Redis (for example Redis Cloud) works the same as for cache: set REDIS_* and optionally REDIS_BULLMQ_DB; see Redis cache. Workers must use the same Redis as the API. Env and deployment: Configuration - Worker.

Configuration

• Transport, queue names, and feature flags: backend/config/orchestratorFlows.ts (for example integrationRefresh.enabled and scheduledSocialPost.missingPostRescanIntervalMs). backend/config/GlobalConfig.ts copies integrationRefresh.enabled into config.bullmq.integrationRefresh.enabled except under Jest, where it is always false; scheduledSocialPost.enabled is also forced false under Jest. Add future flows as sibling entries instead of per-flow keys in backend/.env.development.example.
• Jest: JEST_WORKER_ID forces the integration-refresh supervisor off regardless of integrationRefresh.enabled. To test the supervisor, mock config or reload modules with different orchestratorFlows settings.
• Elsewhere: toggle integrationRefresh.enabled and other flow flags in code for the deployment profile you want (or split config by environment in that file).

Worker health and Sentry

Each *BullMqWorker bootstraps via orchestrator/worker/bootstrapOrchestratorWorker.ts:

• HTTP — GET /health and GET /health/status return JSON with status, worker, redis, uptimeSeconds, and optional BullMQ queue counts. Use these URLs for uptime monitors (Railway health checks, Better Stack, etc.). Port resolution: host PORT when set, else ORCHESTRATOR_WORKER_HEALTH_PORT (default 3091); 0 disables the listener.
• Sentry — Workers import the same backend/connections/sentry module as the API. Set SENTRY_DSN on worker services; uncaught errors are tagged openquok.worker and openquok.worker.queue.

BullMQ reconciliation (Flowcraft adapter)

The BullMQ adapter guide describes createBullMQReconciler: it scans Redis keys that hold workflow state, treats runs idle longer than a threshold as stalled, and re-enqueues the appropriate next jobs so a worker can continue. That is aimed at production reliability when jobs or workers disappear between steps.

Clearing stale Flowcraft runs (Redis)

If you deploy a workflow change and see runs failing repeatedly with errors like:

• Implementation for 'fn_…' not found
• Blueprint with ID '…' not found

those are usually stale workflow runs persisted in Redis from an older build. The BullMQ reconciler will keep resuming them until you delete their run state keys.

OpenQuok provides a helper script that deletes Flowcraft run state for a specific blueprintId:

pnpm --filter openquok-orchestrator script:clear-flowcraft-runs scheduled-social-post

It loads dotenv the same way workers do (via backend/config/loadBackendDotenv.cjs), so it works with:

• backend/.env.development.local when NODE_ENV=development
• backend/.env.production.local when NODE_ENV=production
• injected environment variables in production hosts (Railway, Fly, etc.)

BullMQ webhook endpoints (Flowcraft adapter)

The same guide describes wiring HTTP routes so external systems can POST payloads that resume workflows using Flowcraft webhook nodes (Flow.createWebhook() / wait-for-callback style graphs).

Further reading

• Configuration - Worker — env vars, production scripts, dotenv resolution
• Railway Deployment — persistent services, CLI, railway.toml
• Fluent API
• Pausing and sleep nodes (this workflow uses an imperative delay inside tick because the wait length comes from the database at runtime)
• Runtime adapter: BullMQ (reconciliation, webhooks, worker/client setup)