Implement session tracking product syncs

apps/api/src/services/sync.ts

@@ -2,9 +2,17 @@ import { PrintfulClient, ProductSyncer, WebflowClient, type ProductSyncerOptions
 import { log } from "../util";
 import { db } from "../database/db";
 import { DatabaseError } from "pg";
+import type { Insertable, Selectable } from "kysely";
+import type { ProductSyncs } from "../database/out/db";
+
+type SyncStatus = "queued" | "active" | "failed" | "succeeded";
 
 type QueuedSync = {
-    id?: string | null,
+    id?: string,
+    session: {
+        name: string,
+        id: string
+    },
     filter: ProductSyncerOptions["filter"]
 }
 
@@ -23,11 +31,13 @@ export class SyncService {
         let id = queuedSync.id ?? null;
         let freedSlot = false;
         try {
-            log.info("starting sync run");
+            log.info({ session: queuedSync.session }, "starting sync run");
             await this.productSyncer.sync({
                 filter: queuedSync.filter,
                 onStart: async (startDate) => {
-                    const values = {
+                    const values: Insertable<ProductSyncs> = {
+                        session_id: queuedSync.session.id,
+                        session_name: queuedSync.session.name,
                         started_at: startDate,
                         printful_product_id_filter: queuedSync.filter?.printfulProductIds ?? null,
                     };
@@ -103,34 +113,55 @@ export class SyncService {
             }
         }
         finally {
-            if (freedSlot)
-                this.syncNext().catch((err) => log.error({ err }, "failed to process next queued sync"));
+            if (freedSlot) {
+                try {
+                    const frontQueuedSync = await this.queue.front();
+                    if (frontQueuedSync) await this.sync(frontQueuedSync);
+                }
+                catch (err) {
+                    log.error({ err }, "failed to process next queued sync")
+                }
+            }
         }
     }
 
-    async syncNext() {
-        const nextSync = await this.queue.front();
-        if (nextSync) {
-            await this.sync({
-                id: nextSync.id,
-                filter: { printfulProductIds: nextSync.printful_product_id_filter }
-            });
-        }
-    }
-
-    async getActiveSync() {
+    async getActiveSync(sessionId?: string) {
         return await db.selectFrom("product_syncs")
             .selectAll()
+            .$if(sessionId !== undefined, (qb) =>
+                qb.where("session_id", "=", sessionId!)
+            )
             .where("started_at", "is not", null)
             .where("ended_at", "is", null)
             .limit(1)
             .executeTakeFirst();
     }
+
+    async getLatestSync(sessionId?: string) {
+        return await db.selectFrom("product_syncs")
+            .selectAll()
+            .$if(sessionId !== undefined, (qb) =>
+                qb.where("session_id", "=", sessionId!)
+            )
+            .orderBy("created_at", "desc")
+            .limit(1)
+            .executeTakeFirst();
+    }
+
+    deriveSyncStatus(sync: Selectable<ProductSyncs>): SyncStatus {
+        if (!sync.started_at) return "queued";
+        if (!sync.ended_at) return "active";
+        return sync.has_failed ? "failed" : "succeeded";
+    }
 }
 
 class SyncQueue {
     async enqueue(queuedSync: QueuedSync) {
-        const values = { printful_product_id_filter: queuedSync.filter?.printfulProductIds };
+        const values: Insertable<ProductSyncs> = {
+            session_id: queuedSync.session.id,
+            session_name: queuedSync.session.name,
+            printful_product_id_filter: queuedSync.filter?.printfulProductIds ?? null
+        };
         if (queuedSync.id) {
             await db.updateTable("product_syncs")
                 .set(values)
@@ -144,12 +175,24 @@ class SyncQueue {
         }
     }
 
-    async front() {
-        return await db.selectFrom("product_syncs")
+    async front(): Promise<QueuedSync | undefined> {
+        const result = await db.selectFrom("product_syncs")
             .selectAll()
             .where("started_at", "is", null)
             .orderBy("created_at", "asc")
             .limit(1)
             .executeTakeFirst();
+        if (!result) return;
+
+        return {
+            id: result.id,
+            session: {
+                id: result.session_id,
+                name: result.session_name
+            },
+            filter: {
+                printfulProductIds: result.printful_product_id_filter
+            }
+        }
     }
 }