Implement sync service queue

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()

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

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();
     }
 }

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<void>
     onCompletion?: (completionDate: Date, duration: number) => Promise<void>
@@ -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))