From c0dad39d55ef0c039a82b67d1997382fba170d44 Mon Sep 17 00:00:00 2001 From: "Angelo B. J. Luidens" Date: Wed, 3 Jun 2026 15:12:44 -0400 Subject: [PATCH] Stage 4.1: BullMQ publish worker + Redis/Postgres integration - apps/scheduler/src/queue.ts: PUBLISH_QUEUE, BullMQ-compatible Redis connection factory, queue factory, default retry/backoff job opts. - apps/scheduler/src/worker.ts: makePublishWorker factory drives the publish-loop state machine (queued -> dispatching -> published | failed/dlq) and persists each transition to publications. Dispatcher is injected (fake in tests; the LinkedIn client swaps in once Gate 0.9 / CMA lands). - apps/scheduler/src/main.ts: runnable entry; boots the worker; LinkedIn dispatch is fail-closed until CMA is approved. - apps/scheduler/src/worker.integration.test.ts: gated behind INTEGRATION=1; verifies enqueue -> consume -> row published within 5s, and failure -> dlq with error recorded. Run against real Redis + Postgres. Closes the Stage 4.1 DoD (enqueue -> worker-consume -> DB row updated within 5s). Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/scheduler/package.json | 4 +- apps/scheduler/src/main.ts | 48 +++++++ apps/scheduler/src/queue.ts | 27 ++++ apps/scheduler/src/worker.integration.test.ts | 130 ++++++++++++++++++ apps/scheduler/src/worker.ts | 112 ++++++++++++++- bun.lock | 1 + 6 files changed, 319 insertions(+), 3 deletions(-) create mode 100644 apps/scheduler/src/main.ts create mode 100644 apps/scheduler/src/queue.ts create mode 100644 apps/scheduler/src/worker.integration.test.ts diff --git a/apps/scheduler/package.json b/apps/scheduler/package.json index 083edba..f059928 100644 --- a/apps/scheduler/package.json +++ b/apps/scheduler/package.json @@ -5,7 +5,8 @@ "type": "module", "scripts": { "build": "tsc --noEmit", - "dev": "bun --watch src/worker.ts", + "dev": "bun --watch src/main.ts", + "start": "bun src/main.ts", "typecheck": "tsc --noEmit", "test": "vitest run", "lint": "echo 'lint pending'" @@ -15,6 +16,7 @@ "@stargue/schema": "workspace:*", "@stargue/observability": "workspace:*", "bullmq": "^5.30.0", + "drizzle-orm": "^0.36.0", "ioredis": "^5.4.0" }, "devDependencies": { diff --git a/apps/scheduler/src/main.ts b/apps/scheduler/src/main.ts new file mode 100644 index 0000000..d114aa4 --- /dev/null +++ b/apps/scheduler/src/main.ts @@ -0,0 +1,48 @@ +#!/usr/bin/env bun +/** + * Scheduler service entry point — runs the BullMQ publish worker. + * + * Env: DATABASE_URL, REDIS_URL (both required). + */ +import { createDb } from "@stargue/schema/client"; +import { makeRedisConnection } from "./queue"; +import { makePublishWorker, type Dispatcher } from "./worker"; + +const DATABASE_URL = process.env.DATABASE_URL; +const REDIS_URL = process.env.REDIS_URL; +if (!DATABASE_URL) throw new Error("DATABASE_URL is required"); +if (!REDIS_URL) throw new Error("REDIS_URL is required"); + +const { db, sql } = createDb(DATABASE_URL); +const connection = makeRedisConnection(REDIS_URL); + +/** + * Gate 0.9: the real LinkedIn dispatcher is wired once Community Management API + * access is granted. Until then the worker boots and consumes jobs but refuses + * to dispatch — surfacing a clear, audited failure rather than silently + * dropping or fake-publishing. + */ +const dispatch: Dispatcher = async (target) => { + throw new Error( + `LinkedIn dispatch not yet enabled (Gate 0.9 / CMA pending) — outlet=${target.outlet} publication=${target.id}`, + ); +}; + +const worker = makePublishWorker({ db, connection, dispatch }); +worker.on("ready", () => console.log("[scheduler] publish worker ready")); +worker.on("completed", (job) => console.log(`[scheduler] job ${job.id} completed`)); +worker.on("failed", (job, err) => + console.error(`[scheduler] job ${job?.id} failed: ${err.message}`), +); + +const shutdown = async () => { + console.log("[scheduler] shutting down…"); + await worker.close(); + connection.disconnect(); + await sql.end(); + process.exit(0); +}; +process.on("SIGTERM", shutdown); +process.on("SIGINT", shutdown); + +console.log("[scheduler] started"); diff --git a/apps/scheduler/src/queue.ts b/apps/scheduler/src/queue.ts new file mode 100644 index 0000000..3131271 --- /dev/null +++ b/apps/scheduler/src/queue.ts @@ -0,0 +1,27 @@ +import { Queue } from "bullmq"; +import IORedis from "ioredis"; + +export const PUBLISH_QUEUE = "publish"; + +export interface PublishJobData { + publicationId: number; +} + +/** + * A BullMQ-compatible Redis connection. `maxRetriesPerRequest: null` is required + * by BullMQ for blocking commands. Create a separate connection per Queue and + * per Worker — they should not share one blocking client. + */ +export const makeRedisConnection = (url: string): IORedis => + new IORedis(url, { maxRetriesPerRequest: null }); + +export const makePublishQueue = (connection: IORedis): Queue => + new Queue(PUBLISH_QUEUE, { connection }); + +/** Default per-job retry policy (Stage 4.3 chaos test exercises the retry/DLQ path). */ +export const DEFAULT_JOB_OPTS = { + attempts: 3, + backoff: { type: "exponential" as const, delay: 2000 }, + removeOnComplete: 1000, + removeOnFail: 5000, +}; diff --git a/apps/scheduler/src/worker.integration.test.ts b/apps/scheduler/src/worker.integration.test.ts new file mode 100644 index 0000000..662b6d9 --- /dev/null +++ b/apps/scheduler/src/worker.integration.test.ts @@ -0,0 +1,130 @@ +import { describe, it, expect, beforeAll, afterAll, beforeEach, afterEach } from "vitest"; +import { eq } from "drizzle-orm"; +import type { Queue, Worker } from "bullmq"; +import type IORedis from "ioredis"; +import { createDb, type DbHandle } from "@stargue/schema/client"; +import { content, publications } from "@stargue/schema/db"; +import { makeRedisConnection, makePublishQueue, type PublishJobData } from "./queue"; +import { makePublishWorker, type Dispatcher } from "./worker"; + +/** + * Stage 4.1 DoD: enqueue → worker consumes → DB row updated within 5s. + * Gated behind INTEGRATION=1; needs the local infra: + * docker compose -f infra/docker-compose.yml up -d --wait postgres redis + * INTEGRATION=1 bun --filter @stargue/scheduler test + */ +const RUN = !!process.env.INTEGRATION; +const PG = + process.env.DATABASE_URL ?? + "postgres://stargue:stargue_dev@localhost:5433/stargue_publishing_engine"; +const REDIS = process.env.REDIS_URL ?? "redis://localhost:6380"; + +describe.skipIf(!RUN)("BullMQ publish worker — real Redis + Postgres (Stage 4.1)", () => { + let h: DbHandle; + let connQ: IORedis; + let queue: Queue; + let active: { worker: Worker; conn: IORedis } | null = null; + + beforeAll(() => { + h = createDb(PG); + connQ = makeRedisConnection(REDIS); + queue = makePublishQueue(connQ); + }); + + afterAll(async () => { + await queue.close(); + connQ.disconnect(); + await h.sql.end(); + }); + + beforeEach(async () => { + await h.sql`TRUNCATE content, publications RESTART IDENTITY CASCADE`; + await queue.obliterate({ force: true }).catch(() => {}); + }); + + afterEach(async () => { + if (active) { + await active.worker.close(); + active.conn.disconnect(); + active = null; + } + }); + + const startWorker = (dispatch: Dispatcher, maxRetries?: number): Worker => { + const conn = makeRedisConnection(REDIS); + const worker = makePublishWorker({ + db: h.db, + connection: conn, + dispatch, + ...(maxRetries !== undefined ? { maxRetries } : {}), + }); + active = { worker, conn }; + return worker; + }; + + const seedQueuedPublication = async () => { + const [c] = await h.db + .insert(content) + .values({ + vault_path: "Stargue/Projects/Publishing Engine/post.md", + slug: "post", + title: "Post", + body_sanitized: "body", + frontmatter_jsonb: { status: "ready" }, + content_hash: "b".repeat(64), + }) + .returning(); + const [pub] = await h.db + .insert(publications) + .values({ + content_id: c!.id, + outlet: "linkedin", + status: "queued", + idempotency_key: `${c!.id}|linkedin|immediate`, + }) + .returning(); + return pub!; + }; + + const waitForStatus = async (id: number, status: string, timeoutMs = 5000) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const [row] = await h.db.select().from(publications).where(eq(publications.id, id)); + if (row && row.status === status) return row; + await new Promise((r) => setTimeout(r, 100)); + } + const [row] = await h.db.select().from(publications).where(eq(publications.id, id)); + return row; + }; + + it("enqueue → worker consumes → row published within 5s", async () => { + const pub = await seedQueuedPublication(); + startWorker(async () => ({ + externalId: "urn:li:share:7100000000000000000", + externalUrl: + "https://www.linkedin.com/feed/update/urn:li:share:7100000000000000000", + })); + + await queue.add("publish", { publicationId: pub.id }); + + const row = await waitForStatus(pub.id, "published"); + expect(row?.status).toBe("published"); + expect(row?.external_id).toBe("urn:li:share:7100000000000000000"); + expect(row?.external_url).toContain("linkedin.com"); + expect(row?.published_at).toBeInstanceOf(Date); + }); + + it("dispatch failure with retries exhausted marks the row dlq + records error", async () => { + const pub = await seedQueuedPublication(); + startWorker(async () => { + throw new Error("LinkedIn 503 Service Unavailable"); + }, 0); + + await queue.add("publish", { publicationId: pub.id }, { attempts: 1 }); + + const row = await waitForStatus(pub.id, "dlq"); + expect(row?.status).toBe("dlq"); + expect(row?.error).toMatch(/LinkedIn 503/); + expect(row?.external_id).toBeNull(); + }); +}); diff --git a/apps/scheduler/src/worker.ts b/apps/scheduler/src/worker.ts index c2b1800..4de84cc 100644 --- a/apps/scheduler/src/worker.ts +++ b/apps/scheduler/src/worker.ts @@ -1,2 +1,110 @@ -// BullMQ worker entry point. Full implementation in Stage 4. -export const SCHEDULER_NAME = "@stargue/scheduler"; +import { Worker, type Job } from "bullmq"; +import type IORedis from "ioredis"; +import { eq } from "drizzle-orm"; +import { publications } from "@stargue/schema/db"; +import type { Db } from "@stargue/schema/client"; +import { PUBLISH_QUEUE, type PublishJobData } from "./queue"; +import { transition, type PublishContext } from "./publish-loop"; + +export interface DispatchResult { + externalId: string; + externalUrl: string; +} + +export interface DispatchTarget { + id: number; + outlet: string; + content_id: number; +} + +/** + * Outlet dispatcher — publishes a single publication to its outlet and returns + * the external id/url. Injected so the worker is testable with a fake and so + * the real LinkedIn dispatch (Gate 0.9) can be swapped in without touching the + * worker's state-machine wiring. + */ +export type Dispatcher = (target: DispatchTarget) => Promise; + +export interface PublishWorkerDeps { + db: Db; + connection: IORedis; + dispatch: Dispatcher; + maxRetries?: number; +} + +/** + * Stage 4 BullMQ worker. Consumes publish jobs, drives the publish-loop state + * machine (queued → dispatching → published | failed), and persists each + * transition to the `publications` row. Throwing lets BullMQ apply the job's + * retry/backoff policy; the row is marked `dlq` once retries are exhausted. + */ +export const makePublishWorker = (deps: PublishWorkerDeps): Worker => { + const maxRetries = deps.maxRetries ?? 3; + + return new Worker( + PUBLISH_QUEUE, + async (job: Job) => { + const { publicationId } = job.data; + + const [pub] = await deps.db + .select() + .from(publications) + .where(eq(publications.id, publicationId)); + if (!pub) throw new Error(`publication ${publicationId} not found`); + + // The DB row is the source of truth; reconstruct the machine at "queued". + let ctx: PublishContext = { + state: "queued", + externalId: pub.external_id, + externalUrl: pub.external_url, + lastError: pub.error, + retriesSoFar: job.attemptsMade, + }; + + ctx = transition(ctx, { type: "dispatch_start" }); + await deps.db + .update(publications) + .set({ status: ctx.state }) + .where(eq(publications.id, publicationId)); + + try { + const res = await deps.dispatch({ + id: pub.id, + outlet: pub.outlet, + content_id: pub.content_id, + }); + ctx = transition(ctx, { + type: "dispatch_success", + externalId: res.externalId, + externalUrl: res.externalUrl, + }); + await deps.db + .update(publications) + .set({ + status: ctx.state, + external_id: res.externalId, + external_url: res.externalUrl, + published_at: new Date(), + error: null, + }) + .where(eq(publications.id, publicationId)); + return { status: ctx.state, externalId: res.externalId }; + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + ctx = transition(ctx, { + type: "dispatch_failure", + error: message, + retriesSoFar: job.attemptsMade, + maxRetries, + }); + await deps.db + .update(publications) + .set({ status: ctx.state, error: message }) + .where(eq(publications.id, publicationId)); + // Re-throw so BullMQ records the attempt and applies backoff/retry. + throw err; + } + }, + { connection: deps.connection }, + ); +}; diff --git a/bun.lock b/bun.lock index f774696..393fdb4 100644 --- a/bun.lock +++ b/bun.lock @@ -56,6 +56,7 @@ "@stargue/observability": "workspace:*", "@stargue/schema": "workspace:*", "bullmq": "^5.30.0", + "drizzle-orm": "^0.36.0", "ioredis": "^5.4.0", }, "devDependencies": {