diff --git a/apps/api/src/services/commerce.ts b/apps/api/src/services/commerce.ts index 91ec889..eef20c8 100644 --- a/apps/api/src/services/commerce.ts +++ b/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) { - let id = queuedProductSync.id ?? null; + async start(payload: ProductSyncPayload) { + 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 = { - 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; - } + await this.Queue.enqueue(payload, startDate); }, 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") - .set({ syncing_p_product_ids: [], ended_at: completionDate }) - .where("id", "=", id) - .execute(); + await this.Queue.fulfill(id, completionDate) 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") - .set({ syncing_p_product_ids: [], ended_at: new Date(), has_failed: true }) - .where("id", "=", activeSync.id) - .execute(); - await this.start(queuedProductSync); + await this.Queue.fail(activeSync.id, new Date()); + await this.start(payload); } 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") - .set({ syncing_p_product_ids: [], ended_at: new Date(), has_failed: true }) - .where("id", "=", id) - .execute(); + await this.Queue.fail(id, new Date()); freedSlot = true; } throw err; @@ -164,31 +132,39 @@ class ProductSyncService { deriveStatus(sync: Selectable): 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 { const values: Insertable = { - session_id: queuedProductSync.session.id, - session_name: queuedProductSync.session.name, - p_product_id_filter: queuedProductSync.filter?.pProductIds ?? null + session_id: payload.session.id, + session_name: payload.session.name, + p_product_id_filter: payload.filter?.pProductIds ?? null }; - if (queuedProductSync.id) { - await db.updateTable("product_syncs") + if (startDate) values.started_at = startDate; + if (payload.id) { + const result = await db.updateTable("product_syncs") .set(values) - .where("id", "=", queuedProductSync.id) - .execute(); + .where("id", "=", payload.id) + .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 { + async front(): Promise { 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(); + } } diff --git a/apps/portal/src/routes/(app)/commerce/products/+page.svelte b/apps/portal/src/routes/(app)/commerce/products/+page.svelte index e914ca6..9d72923 100644 --- a/apps/portal/src/routes/(app)/commerce/products/+page.svelte +++ b/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();