Continue product sync queue on startup

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