diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index bb898d6..cb1376f 100644 --- a/apps/api/src/index.ts +++ b/apps/api/src/index.ts @@ -4,9 +4,10 @@ import process from 'node:process' import { log } from "./util"; if (cluster.isPrimary) { - for (let i = 0; i < os.availableParallelism(); i++) + const n = os.availableParallelism(); + log.info(`Starting ${n} worker processes`) + for (let i = 0; i < n; i++) cluster.fork() } else { await import('./server') - log.info(`Worker ${process.pid} started`) } diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index 0acf5c2..9d03d4a 100644 --- a/apps/api/src/server.ts +++ b/apps/api/src/server.ts @@ -23,6 +23,7 @@ import { env, log } from "./util"; import serverTiming from "@elysia/server-timing"; import jwt from "@elysia/jwt"; import { SyncService } from "./services/sync"; +import cluster from "node:cluster"; // SERVICES // ----------- @@ -248,6 +249,11 @@ export const app = new Elysia() } }); -app.listen(3000, () => log.info({ port: 3000 }, "server started")); +app.listen(3000, async () => { + log.info({ port: 3000 }, "server started") + if (cluster.worker?.id === 1) { + await syncService.nextSync().catch((err) => log.error({ err }, "failed to resume product sync queue")); + } +}); export type API = typeof app diff --git a/apps/api/src/services/sync.ts b/apps/api/src/services/sync.ts index 11b61b0..4da95a1 100644 --- a/apps/api/src/services/sync.ts +++ b/apps/api/src/services/sync.ts @@ -3,62 +3,42 @@ import { log } from "../util"; import { db } from "../database/db"; import { DatabaseError } from "pg"; -class SyncQueue { - async enqueue(item: SyncServiceOptions) { - const values = { printful_product_id_filter: item.filter?.printfulProductIds }; - if (item.id) { - await db.updateTable("product_syncs") - .set(values) - .where("id", "=", item.id) - .execute(); - } - else { - await db.insertInto("product_syncs") - .values(values) - .execute(); - } - } - - async front() { - return await db.selectFrom("product_syncs") - .selectAll() - .where("started_at", "is", null) - .orderBy("created_at", "asc") - .limit(1) - .executeTakeFirst(); - } -} - -type SyncServiceOptions = { +type QueuedSync = { id?: string | null, filter: ProductSyncerOptions["filter"] } +class AlreadyClaimedError extends Error { } + export class SyncService { private readonly productSyncer: ProductSyncer; - private queue: SyncQueue; + readonly queue: SyncQueue; constructor(printful: PrintfulClient, webflow: WebflowClient) { this.productSyncer = new ProductSyncer(printful, webflow, log); this.queue = new SyncQueue(); } - async sync(opt: SyncServiceOptions) { - let id = opt.id ?? null; + async sync(queuedSync: QueuedSync) { + let id = queuedSync.id ?? null; try { log.info("starting sync run"); await this.productSyncer.sync({ - filter: opt.filter, + filter: queuedSync.filter, onStart: async (startDate) => { const values = { started_at: startDate, - printful_product_id_filter: opt.filter?.printfulProductIds ?? null, + printful_product_id_filter: queuedSync.filter?.printfulProductIds ?? null, }; if (id) { - await db.updateTable("product_syncs") + // Guard against another worker/machine having already claimed + // this queued row between `queue.front()` and now. + const result = await db.updateTable("product_syncs") .set(values) .where("id", "=", id) - .execute(); + .where("started_at", "is", null) + .executeTakeFirst(); + if (result.numUpdatedRows === 0n) throw new AlreadyClaimedError(); } else { const result = await db.insertInto("product_syncs") @@ -83,17 +63,16 @@ export class SyncService { .execute(); // Sync next item in queue - const nextSync = await this.queue.front(); - if (nextSync) { - await this.sync({ - id: nextSync.id, - filter: { printfulProductIds: nextSync.printful_product_id_filter } - }); - } + await this.nextSync(); } }); } catch (err) { + if (err instanceof AlreadyClaimedError) { + log.info("sync already claimed by another worker, skipping"); + return; + } + const activeSync = await this.getActiveSync(); const activeSyncConflict = activeSync && err instanceof DatabaseError && err.code === "23505"; @@ -106,11 +85,11 @@ export class SyncService { .set({ syncing_printful_product_ids: [], ended_at: new Date(), has_failed: true }) .where("id", "=", activeSync.id) .execute(); - await this.sync(opt); + await this.sync(queuedSync); } else { log.info("sync already active, enqueuing next sync"); - await this.queue.enqueue(opt); + await this.queue.enqueue(queuedSync); } } else { @@ -126,6 +105,16 @@ export class SyncService { } } + async nextSync() { + const nextSync = await this.queue.front(); + if (nextSync) { + await this.sync({ + id: nextSync.id, + filter: { printfulProductIds: nextSync.printful_product_id_filter } + }); + } + } + async getActiveSync() { return await db.selectFrom("product_syncs") .selectAll() @@ -135,3 +124,29 @@ export class SyncService { .executeTakeFirst(); } } + +class SyncQueue { + async enqueue(queuedSync: QueuedSync) { + const values = { printful_product_id_filter: queuedSync.filter?.printfulProductIds }; + if (queuedSync.id) { + await db.updateTable("product_syncs") + .set(values) + .where("id", "=", queuedSync.id) + .execute(); + } + else { + await db.insertInto("product_syncs") + .values(values) + .execute(); + } + } + + async front() { + return await db.selectFrom("product_syncs") + .selectAll() + .where("started_at", "is", null) + .orderBy("created_at", "asc") + .limit(1) + .executeTakeFirst(); + } +} diff --git a/apps/api/src/util.ts b/apps/api/src/util.ts index b3cc7df..9c4fde1 100644 --- a/apps/api/src/util.ts +++ b/apps/api/src/util.ts @@ -1,3 +1,4 @@ +import cluster from 'node:cluster'; import { pino } from 'pino'; export const log = pino({ @@ -24,7 +25,8 @@ export const env = { function requireEnv(key: string): string { const val = Bun.env[key]; if (!val) { - log.error({ name: key }, "Missing required environment variable"); + if (cluster.worker?.id === 1) + log.error({ name: key }, "Missing required environment variable"); throw new Error(`Missing required environment variable: ${key}`) }; return val; @@ -33,7 +35,8 @@ function requireEnv(key: string): string { function optionEnv(key: string, fallback: string = ""): string { const val = Bun.env[key]; if (!val) { - log.warn({ name: key }, `Missing optional environmental variable, falling back to "${fallback}"`) + if (cluster.worker?.id === 1) + log.warn({ name: key }, `Missing optional environmental variable, falling back to "${fallback}"`) return fallback }; return val;