Hook up syncing with postgres

apps/api/src/database/seed.ts

@@ -6,12 +6,26 @@ import { log } from "../util";
 // SYNC RUNS
 // =====================
 {
-    log.info({ name: "sync_runs" }, "Creating table");
-    await db.schema.dropTable("sync_runs").ifExists().execute()
-    await db.schema.createTable("sync_runs")
+    log.info({ name: "product_syncs" }, "Creating table");
+    await db.schema.dropTable("product_syncs").ifExists().execute()
+    await db.schema.createTable("product_syncs")
         .$call(addDefaultColumns)
-        .addColumn("completed_at", "timestamptz")
+        .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("ended_at", "timestamptz")
+        .addColumn("has_failed", "boolean", (cb) => cb.notNull().defaultTo(false))
         .execute()
+
+    await db.schema.dropIndex("idx_one_active_product_sync").ifExists().execute();
+    await db.schema.createIndex("idx_one_active_product_sync")
+        .on("product_syncs")
+        .unique()
+        .column("ended_at")
+        .where(sql.ref("started_at"), "is not", null)
+        .where("ended_at", "is", null)
+        .nullsNotDistinct()
+        .execute();
 }
 
 function addDefaultColumns(ctb: CreateTableBuilder<any, any>) {
@@ -23,6 +37,5 @@ function addDefaultColumns(ctb: CreateTableBuilder<any, any>) {
         .addColumn("created_at", "timestamptz", (cb) => cb
             .notNull()
             .defaultTo(sql`now()`)
-            .unique()
         )
 };

apps/api/src/server.ts

@@ -13,7 +13,6 @@ import { Elysia, NotFoundError, status, t } from "elysia";
 import {
     PrintfulClient,
     PrintfulError,
-    ProductSyncer,
     WebflowClient,
     WebflowError,
     Printful,
@@ -23,6 +22,7 @@ import zipcodesUs from "zipcodes-us";
 import { env, log } from "./util";
 import serverTiming from "@elysia/server-timing";
 import jwt from "@elysia/jwt";
+import { SyncService } from "./services/sync";
 
 // SERVICES
 // -----------
@@ -42,7 +42,7 @@ const webflow = new WebflowClient({
     webhookSecret: env.WEBFLOW_WEBHOOK_SECRET,
 });
 
-const productSyncer = new ProductSyncer(printful, webflow, log);
+const syncService = new SyncService(printful, webflow);
 
 // ELYSIA
 // -----------
@@ -143,15 +143,22 @@ export const app = new Elysia()
             .group("/sync", (app) =>
                 app
                     // Sync status
-                    .get("/", async ({ }) => ({
-                        state: await productSyncer.getState(),
-                        isRunning: await productSyncer.isRunning()
-                    }))
+                    .get("/", async ({ }) => {
+                        const activeRun = await syncService.getActiveRun();
+                        if (!activeRun) throw new NotFoundError("No sync in progress");
+                        return {
+                            startDate: activeRun.started_at,
+                            syncingPrintfulProductIds: activeRun.syncing_printful_product_ids
+                        }
+                    })
                     // Run sync
                     .post("/:printfulProductId?", async ({ params: { printfulProductId } }) => {
                         log.debug({ printfulProductId });
-                        const canSync = await productSyncer.sync({ filter: { printfulProductIds: printfulProductId } });
-                        if (!canSync) throw status(409, "Sync already in progress");
+                        await syncService.start({
+                            printfulProductIds: printfulProductId ?
+                                [printfulProductId] :
+                                undefined
+                        });
                     }, { params: t.Object({ printfulProductId: t.Optional(t.Numeric()) }) }),
             )
             .get("/:printfulProductId", async ({ params }) => {
@@ -178,7 +185,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 productSyncer.sync({ filter: { printfulProductIds: printfulProduct.id } });
+                await syncService.start({ printfulProductIds: [printfulProduct.id] });
                 break;
             }
             case Printful.Webhook.Event.ProductDeleted: {

apps/api/src/services/sync.ts

@@ -0,0 +1,58 @@
+import { PrintfulClient, ProductSyncer, WebflowClient, type SyncOptions } from "@blade-and-brawn/commerce";
+import { log } from "../util";
+import { db } from "../database/db";
+
+export class SyncService {
+    readonly productSyncer: ProductSyncer
+
+    constructor(printful: PrintfulClient, webflow: WebflowClient) {
+        this.productSyncer = new ProductSyncer(printful, webflow, log);
+    }
+
+    async start(filter: SyncOptions["filter"]) {
+        let id: string | null = null;
+        try {
+            await this.productSyncer.sync({
+                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;
+                },
+                onStep: async (printfulProductIds: number[]) => {
+                    if (!id) throw new Error("Expected sync run ID");
+                    await db.updateTable("product_syncs")
+                        .set({ syncing_printful_product_ids: printfulProductIds })
+                        .where("id", "=", id)
+                        .execute();
+                },
+                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, })
+                        .where("id", "=", id)
+                        .execute();
+                }
+            });
+        }
+        catch (err) {
+            if (id) {
+                await db.updateTable("product_syncs")
+                    .set({ syncing_printful_product_ids: [], ended_at: new Date(), has_failed: true })
+                    .where("id", "=", id)
+                    .execute();
+            }
+            throw err;
+        }
+    }
+
+    async getActiveRun() {
+        return await db.selectFrom("product_syncs").selectAll().where("ended_at", "is", null).executeTakeFirst();
+    }
+}