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) <noreply@anthropic.com>

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": {

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");

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<PublishJobData> =>
+  new Queue<PublishJobData>(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,
+};

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<PublishJobData>;
+  let active: { worker: Worker<PublishJobData>; 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<PublishJobData> => {
+    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();
+  });
+});

apps/scheduler/src/worker.ts

@@ -1,2 +1,110 @@
-// BullMQ worker entry point. Full implementation in Stage 4.
+import { Worker, type Job } from "bullmq";
-export const SCHEDULER_NAME = "@stargue/scheduler";
+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<DispatchResult>;
+
+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<PublishJobData> => {
+  const maxRetries = deps.maxRetries ?? 3;
+
+  return new Worker<PublishJobData>(
+    PUBLISH_QUEUE,
+    async (job: Job<PublishJobData>) => {
+      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 },
+  );
+};

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": {