Extract product sync queue db calls into Queue Service methods

apps/api/src/services/commerce.ts

@@ -5,9 +5,9 @@ import { DatabaseError } from "pg";
 import type { Insertable, Selectable } from "kysely";
 import type { ProductSyncs } from "../database/out/db";
 
-type SyncStatus = "queued" | "active" | "failed" | "succeeded";
+type SyncStatus = "queued" | "active" | "failed" | "fulfilled";
 
-type QueuedProductSync = {
+type ProductSyncPayload = {
     id?: string,
     session: {
         name: string,
@@ -40,38 +40,15 @@ class ProductSyncService {
         this.ProductSyncer = new ProductSyncer(this.Printful, this.Webflow, log);
     }
 
-    async start(queuedProductSync: QueuedProductSync) {
+    async start(payload: ProductSyncPayload) {
-        let id = queuedProductSync.id ?? null;
+        let id = payload.id ?? null;
         let freedSlot = false;
         try {
-            log.info({ session: queuedProductSync.session }, "starting sync run");
+            log.info({ session: payload.session }, "starting sync run");
             await this.ProductSyncer.run({
-                filter: queuedProductSync.filter,
+                filter: payload.filter,
                 onStart: async (startDate) => {
-                    const values: Insertable<ProductSyncs> = {
+                    await this.Queue.enqueue(payload, startDate);
-                        session_id: queuedProductSync.session.id,
-                        session_name: queuedProductSync.session.name,
-                        started_at: startDate,
-                        p_product_id_filter: queuedProductSync.filter?.pProductIds ?? null,
-                    };
-                    if (id) {
-                        const result = await db.updateTable("product_syncs")
-                            .set(values)
-                            .where("id", "=", id)
-                            .where("started_at", "is", null)
-                            .executeTakeFirst();
-
-                        // Guard against another worker/machine having already claimed
-                        // this queued row between `queue.front()` and now.
-                        if (result.numUpdatedRows === 0n) throw new AlreadyClaimedError();
-                    }
-                    else {
-                        const result = await db.insertInto("product_syncs")
-                            .values(values)
-                            .returning("id")
-                            .executeTakeFirstOrThrow();
-                        id = result.id;
-                    }
                 },
                 onStep: async (pProductIds: number[]) => {
                     if (!id) throw new Error("Expected sync run ID");
@@ -82,10 +59,7 @@ class ProductSyncService {
                 },
                 onCompletion: async (completionDate, duration) => {
                     if (!id) throw new Error("Expected sync run ID");
-                    await db.updateTable("product_syncs")
+                    await this.Queue.fulfill(id, completionDate)
-                        .set({ syncing_p_product_ids: [], ended_at: completionDate })
-                        .where("id", "=", id)
-                        .execute();
                     freedSlot = true;
                 }
             });
@@ -103,23 +77,17 @@ class ProductSyncService {
                 // If active sync has been syncing for 10 min (600000ms), clear it and retry this sync
                 if (Date.now() - (activeSync.started_at?.getTime() ?? 0) > 600000) {
                     log.info("already active sync exceeded timeout");
-                    await db.updateTable("product_syncs")
+                    await this.Queue.fail(activeSync.id, new Date());
-                        .set({ syncing_p_product_ids: [], ended_at: new Date(), has_failed: true })
+                    await this.start(payload);
-                        .where("id", "=", activeSync.id)
-                        .execute();
-                    await this.start(queuedProductSync);
                 }
                 else {
                     log.info("sync already active, enqueuing next sync");
-                    await this.Queue.enqueue(queuedProductSync);
+                    await this.Queue.enqueue(payload);
                 }
             }
             else {
                 if (id) {
-                    await db.updateTable("product_syncs")
+                    await this.Queue.fail(id, new Date());
-                        .set({ syncing_p_product_ids: [], ended_at: new Date(), has_failed: true })
-                        .where("id", "=", id)
-                        .execute();
                     freedSlot = true;
                 }
                 throw err;
@@ -164,31 +132,39 @@ class ProductSyncService {
     deriveStatus(sync: Selectable<ProductSyncs>): SyncStatus {
         if (!sync.started_at) return "queued";
         if (!sync.ended_at) return "active";
-        return sync.has_failed ? "failed" : "succeeded";
+        return sync.has_failed ? "failed" : "fulfilled";
     }
 }
 
 class ProductSyncQueueService {
-    async enqueue(queuedProductSync: QueuedProductSync) {
+    async enqueue(payload: ProductSyncPayload, startDate?: Date): Promise<string> {
         const values: Insertable<ProductSyncs> = {
-            session_id: queuedProductSync.session.id,
+            session_id: payload.session.id,
-            session_name: queuedProductSync.session.name,
+            session_name: payload.session.name,
-            p_product_id_filter: queuedProductSync.filter?.pProductIds ?? null
+            p_product_id_filter: payload.filter?.pProductIds ?? null
         };
-        if (queuedProductSync.id) {
+        if (startDate) values.started_at = startDate;
-            await db.updateTable("product_syncs")
+        if (payload.id) {
+            const result = await db.updateTable("product_syncs")
                 .set(values)
-                .where("id", "=", queuedProductSync.id)
+                .where("id", "=", payload.id)
-                .execute();
+                .where("started_at", "is", null)
+                .executeTakeFirstOrThrow();
+            // Guard against another worker/machine having already claimed
+            // this queued row between `queue.front()` and now.
+            if (result.numUpdatedRows === 0n) throw new AlreadyClaimedError();
+            return payload.id;
         }
         else {
-            await db.insertInto("product_syncs")
+            const result = await db.insertInto("product_syncs")
                 .values(values)
-                .execute();
+                .returning("id")
+                .executeTakeFirstOrThrow();
+            return result.id;
         }
     }
 
-    async front(): Promise<QueuedProductSync | undefined> {
+    async front(): Promise<ProductSyncPayload | undefined> {
         const result = await db.selectFrom("product_syncs")
             .selectAll()
             .where("started_at", "is", null)
@@ -208,4 +184,18 @@ class ProductSyncQueueService {
             }
         }
     }
+
+    async fail(id: string, endDate: Date) {
+        await db.updateTable("product_syncs")
+            .set({ syncing_p_product_ids: [], ended_at: endDate, has_failed: true })
+            .where("id", "=", id)
+            .execute();
+    }
+
+    async fulfill(id: string, endDate: Date) {
+        await db.updateTable("product_syncs")
+            .set({ syncing_p_product_ids: [], ended_at: endDate })
+            .where("id", "=", id)
+            .execute();
+    }
 }

apps/portal/src/routes/(app)/commerce/products/+page.svelte

@@ -3,7 +3,7 @@
     import { onMount } from "svelte";
     import { api } from "$lib/api.js";
 
-    type SyncStatus = "queued" | "active" | "failed" | "succeeded" | "none";
+    type SyncStatus = "queued" | "active" | "failed" | "fulfilled" | "none";
 
     const { data } = $props();