diff --git a/apps/api/src/database/seed.ts b/apps/api/src/database/seed.ts index 3aab615..4ce8825 100644 --- a/apps/api/src/database/seed.ts +++ b/apps/api/src/database/seed.ts @@ -3,7 +3,7 @@ import { db } from "./db"; import { log } from "../util"; // ===================== -// SYNC RUNS +// PRODUCT SYNCS // ===================== { log.info({ name: "product_syncs" }, "Creating table"); @@ -11,8 +11,11 @@ import { log } from "../util"; await db.schema.createTable("product_syncs") .$call(addDefaultColumns) .addColumn("printful_product_id_filter", sql`integer[]`) - .addColumn("syncing_printful_product_ids", sql`integer[]`, (cb) => cb.notNull()) - .addColumn("started_at", "timestamptz", (cb) => cb.unique()) + .addColumn("syncing_printful_product_ids", sql`integer[]`, (cb) => cb + .notNull() + .defaultTo(sql`'{}'::integer[]`) + ) + .addColumn("started_at", "timestamptz") .addColumn("ended_at", "timestamptz") .addColumn("has_failed", "boolean", (cb) => cb.notNull().defaultTo(false)) .execute() diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index b909c01..0acf5c2 100644 --- a/apps/api/src/server.ts +++ b/apps/api/src/server.ts @@ -144,21 +144,18 @@ export const app = new Elysia() app // Sync status .get("/", async ({ }) => { - const activeRun = await syncService.getActiveRun(); - if (!activeRun) throw new NotFoundError("No sync in progress"); + const activeSync = await syncService.getActiveSync(); + if (!activeSync) throw new NotFoundError("No sync in progress"); return { - startDate: activeRun.started_at, - syncingPrintfulProductIds: activeRun.syncing_printful_product_ids + startDate: activeSync.started_at, + syncingPrintfulProductIds: activeSync.syncing_printful_product_ids } }) // Run sync .post("/:printfulProductId?", async ({ params: { printfulProductId } }) => { - log.debug({ printfulProductId }); - await syncService.start({ - printfulProductIds: printfulProductId ? - [printfulProductId] : - undefined - }); + const printfulProductIds = printfulProductId ? [printfulProductId] : null; + log.debug({ printfulProductIds }); + await syncService.sync({ filter: { printfulProductIds } }); }, { params: t.Object({ printfulProductId: t.Optional(t.Numeric()) }) }), ) .get("/:printfulProductId", async ({ params }) => { @@ -185,7 +182,7 @@ export const app = new Elysia() case Printful.Webhook.Event.ProductUpdated: { const printfulProduct = payload.data.sync_product; log.info({ productId: printfulProduct.id }, "printful webhook: product updated"); - await syncService.start({ printfulProductIds: [printfulProduct.id] }); + await syncService.sync({ filter: { printfulProductIds: [printfulProduct.id] } }); break; } case Printful.Webhook.Event.ProductDeleted: { diff --git a/apps/api/src/services/sync.ts b/apps/api/src/services/sync.ts index 0a1237a..fda6333 100644 --- a/apps/api/src/services/sync.ts +++ b/apps/api/src/services/sync.ts @@ -1,29 +1,72 @@ -import { PrintfulClient, ProductSyncer, WebflowClient, type SyncOptions } from "@blade-and-brawn/commerce"; +import { PrintfulClient, ProductSyncer, WebflowClient, type ProductSyncerOptions } from "@blade-and-brawn/commerce"; 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 = { + id?: string | null, + filter: ProductSyncerOptions["filter"] +} export class SyncService { - readonly productSyncer: ProductSyncer + private readonly productSyncer: ProductSyncer; + private queue: SyncQueue; constructor(printful: PrintfulClient, webflow: WebflowClient) { this.productSyncer = new ProductSyncer(printful, webflow, log); + this.queue = new SyncQueue(); } - async start(filter: SyncOptions["filter"]) { - let id: string | null = null; + async sync(opt: SyncServiceOptions) { + let id = opt.id ?? null; try { + log.info("starting sync run"); await this.productSyncer.sync({ - filter, + filter: opt.filter, onStart: async (startDate) => { - const result = await db.insertInto("product_syncs") - .values({ - started_at: startDate, - printful_product_id_filter: filter?.printfulProductIds, - syncing_printful_product_ids: [] - }) - .returning("id") - .executeTakeFirstOrThrow(); - id = result.id; + const values = { + started_at: startDate, + printful_product_id_filter: opt.filter?.printfulProductIds ?? null, + }; + if (id) { + await db.updateTable("product_syncs") + .set(values) + .where("id", "=", id) + .execute(); + } + else { + const result = await db.insertInto("product_syncs") + .values(values) + .returning("id") + .executeTakeFirstOrThrow(); + id = result.id; + } }, onStep: async (printfulProductIds: number[]) => { if (!id) throw new Error("Expected sync run ID"); @@ -35,14 +78,29 @@ export class SyncService { onCompletion: async (completionDate, duration) => { if (!id) throw new Error("Expected sync run ID"); await db.updateTable("product_syncs") - .set({ syncing_printful_product_ids: [], ended_at: completionDate, }) + .set({ syncing_printful_product_ids: [], ended_at: completionDate }) .where("id", "=", id) .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 } + }); + } } }); } catch (err) { - if (id) { + const isActiveSyncConflict = err instanceof DatabaseError && err.code === "23505" && (await this.getActiveSync()); + if (isActiveSyncConflict) { + log.info("sync already active, enqueuing next sync"); + await this.queue.enqueue(opt); + return; + } + else if (id) { await db.updateTable("product_syncs") .set({ syncing_printful_product_ids: [], ended_at: new Date(), has_failed: true }) .where("id", "=", id) @@ -52,7 +110,12 @@ export class SyncService { } } - async getActiveRun() { - return await db.selectFrom("product_syncs").selectAll().where("ended_at", "is", null).executeTakeFirst(); + async getActiveSync() { + return await db.selectFrom("product_syncs") + .selectAll() + .where("started_at", "is not", null) + .where("ended_at", "is", null) + .limit(1) + .executeTakeFirst(); } } diff --git a/packages/commerce/src/sync.ts b/packages/commerce/src/sync.ts index dbba7b6..2d0d695 100644 --- a/packages/commerce/src/sync.ts +++ b/packages/commerce/src/sync.ts @@ -17,9 +17,9 @@ type MetaPrintfulProduct = { entries: ProductEntry[]; }; -export type SyncOptions = { +export type ProductSyncerOptions = { filter?: { - printfulProductIds?: number[]; + printfulProductIds: number[] | null; }, onStart?: (startDate: Date) => Promise onCompletion?: (completionDate: Date, duration: number) => Promise @@ -37,8 +37,7 @@ export class ProductSyncer { this.log = (log ?? pino({ level: "silent" })).child({ component: "ProductSyncer" }); } - async sync(opt: SyncOptions = {}) { - this.log.info("starting sync run"); + async sync(opt: ProductSyncerOptions = {}) { const startDate = new Date(); await opt.onStart?.(startDate); @@ -256,7 +255,7 @@ export class ProductSyncer { async generateMetaPrintfulProducts( allWebflowProducts: Webflow.Products.ProductAndSkus[], allPrintfulProducts: Printful.Products.SyncProduct[], - opt: SyncOptions, + opt: ProductSyncerOptions, ) { const metaNameFilter = opt.filter?.printfulProductIds ?.map((id) => allPrintfulProducts.find((p) => p.id === id))