Initial commit: PodcastYes — AI podcast platform

lib/queue/jobs.ts

@@ -0,0 +1,38 @@
+/** Queue/job names and their typed payloads. Shared by the web (producer) and worker (consumer). */
+
+export const QUEUES = {
+  generateEpisode: "episode.generate",
+  generateSeries: "series.generate",
+  repurpose: "episode.repurpose",
+  reconcileBilling: "billing.reconcile",
+  echo: "system.echo",
+} as const;
+
+export type QueueName = (typeof QUEUES)[keyof typeof QUEUES];
+
+export type GenerationType = "full" | "script" | "audio" | "art" | "section" | "repurpose";
+
+export interface GenerateEpisodePayload {
+  episodeId: string;
+  /** "full" runs the whole pipeline; the others re-run a single stage. */
+  type?: GenerationType;
+  /** For type="section", the script section to regenerate. */
+  sectionId?: string;
+}
+
+export interface RepurposePayload {
+  episodeId: string;
+  format: "blog" | "social_thread" | "newsletter";
+}
+
+export interface GenerateSeriesPayload {
+  seriesId: string;
+}
+
+export interface EchoPayload {
+  message: string;
+  episodeId?: string;
+}
+
+/** All queues that must exist before send/work. */
+export const ALL_QUEUES: QueueName[] = Object.values(QUEUES);

lib/queue/pgboss.ts

@@ -0,0 +1,60 @@
+import PgBoss from "pg-boss";
+import { ALL_QUEUES, QUEUES, type GenerateEpisodePayload } from "./jobs";
+
+// One pg-boss instance per process, lazily started. The worker process supervises
+// (maintenance/scheduling); the web process only sends, so it skips supervision.
+let instance: PgBoss | null = null;
+let startup: Promise<PgBoss> | null = null;
+let queuesReady = false;
+
+interface BossOptions {
+  /** True only in the worker process. */
+  supervise?: boolean;
+}
+
+export async function getBoss(opts: BossOptions = {}): Promise<PgBoss> {
+  if (instance) return instance;
+  if (!startup) {
+    const boss = new PgBoss({
+      connectionString: process.env.DATABASE_URL,
+      supervise: opts.supervise ?? false,
+      schedule: opts.supervise ?? false,
+      // Keep the pool small — the web process shares the DB with Prisma.
+      max: opts.supervise ? 10 : 3,
+    });
+    boss.on("error", (err) => console.error("[pg-boss] error", err));
+    startup = boss.start().then(async () => {
+      instance = boss;
+      await ensureQueues(boss);
+      return boss;
+    });
+  }
+  return startup;
+}
+
+/** Create every known queue (idempotent) so send()/work() never race on existence. */
+async function ensureQueues(boss: PgBoss): Promise<void> {
+  if (queuesReady) return;
+  for (const name of ALL_QUEUES) {
+    await boss.createQueue(name);
+  }
+  queuesReady = true;
+}
+
+/** Enqueue a full or partial episode generation. */
+export async function enqueueEpisodeGeneration(
+  payload: GenerateEpisodePayload,
+  options?: { priority?: number; singletonKey?: string }
+): Promise<string | null> {
+  const boss = await getBoss();
+  return boss.send(QUEUES.generateEpisode, payload, {
+    retryLimit: 2,
+    retryDelay: 30,
+    retryBackoff: true,
+    expireInMinutes: 30,
+    ...(options?.priority ? { priority: options.priority } : {}),
+    ...(options?.singletonKey ? { singletonKey: options.singletonKey } : {}),
+  });
+}
+
+export { QUEUES };