From 87df874f75e6f56f995d79a0522f4e2517b9c598 Mon Sep 17 00:00:00 2001 From: Dominic Ferrando Date: Sat, 4 Jul 2026 00:13:25 -0400 Subject: [PATCH] Implement session tracking product syncs --- apps/api/src/server.ts | 60 ++++++++------ apps/api/src/services/sync.ts | 81 ++++++++++++++----- .../src/routes/(app)/commerce/+page.svelte | 21 +++-- 3 files changed, 112 insertions(+), 50 deletions(-) diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index a10ce9a..bc1486e 100644 --- a/apps/api/src/server.ts +++ b/apps/api/src/server.ts @@ -24,6 +24,7 @@ import serverTiming from "@elysia/server-timing"; import jwt from "@elysia/jwt"; import { SyncService } from "./services/sync"; import cluster from "node:cluster"; +import { randomUUIDv7 } from "bun"; // SERVICES // ----------- @@ -104,7 +105,7 @@ export const app = new Elysia() if (body.password !== env.ADMIN_PASSWORD) throw status(401, "Invalid credentials"); auth.set({ - value: await jwt.sign({ id: "admin", exp: "7d" }), + value: await jwt.sign({ sessionId: randomUUIDv7(), exp: "7d" }), path: "/", maxAge: 60 * 60 * 24 * 7, sameSite: "lax", @@ -115,17 +116,13 @@ export const app = new Elysia() undefined }); }, { - body: t.Object({ - password: t.String() - }), + body: t.Object({ password: t.String() }), cookie: t.Cookie({ auth: t.Optional(t.String()) }) }) // CALCULATOR .post("/calculate", async ({ body }) => { - return { - levels: levelCalculator.calculate(body.player, body.activityPerformances), - }; + return { levels: levelCalculator.calculate(body.player, body.activityPerformances) }; }, { body: t.Object({ player: PlayerSchema, @@ -134,29 +131,34 @@ export const app = new Elysia() }) // COMMERCE - .group("/products", (app) => - app - .guard({ cookie: t.Cookie({ auth: t.Optional(t.String()) }) }) - .onBeforeHandle(async ({ jwt, cookie: { auth } }) => { + .group("/products", + { cookie: t.Cookie({ auth: t.Optional(t.String()) }) }, + (app) => app + .resolve(async ({ jwt, cookie: { auth } }) => { const token = auth.value && await jwt.verify(auth.value); - if (!token) throw status(401, "Unauthorized"); + if (!token || !token.sessionId) throw status(401, "Unauthorized"); + return { sessionId: token.sessionId.toString() }; }) .group("/sync", (app) => app // Sync status - .get("/", async ({ }) => { - const activeSync = await syncService.getActiveSync(); - if (!activeSync) throw new NotFoundError("No sync in progress"); + .get("/", async ({ sessionId }) => { + const latestSync = await syncService.getLatestSync(sessionId); + if (!latestSync) throw new NotFoundError("No sync found"); return { - startDate: activeSync.started_at, - syncingPrintfulProductIds: activeSync.syncing_printful_product_ids + startDate: latestSync.started_at, + status: syncService.deriveSyncStatus(latestSync), + syncingPrintfulProductIds: latestSync.syncing_printful_product_ids } }) // Run sync - .post("/:printfulProductId?", async ({ params: { printfulProductId } }) => { - const printfulProductIds = printfulProductId ? [printfulProductId] : null; - log.debug({ printfulProductIds }); - await syncService.sync({ filter: { printfulProductIds } }); + .post("/:printfulProductId?", async ({ params: { printfulProductId }, sessionId }) => { + await syncService.sync({ + session: { id: sessionId, name: "Portal" }, + filter: { + printfulProductIds: printfulProductId ? [printfulProductId] : null + } + }); }, { params: t.Object({ printfulProductId: t.Optional(t.Numeric()) }) }), ) .get("/:printfulProductId", async ({ params }) => { @@ -184,7 +186,12 @@ export const app = new Elysia() const printfulProduct = payload.data.sync_product; log.info({ productId: printfulProduct.id }, "printful webhook: product updated"); - await syncService.sync({ filter: { printfulProductIds: [printfulProduct.id] } }); + await syncService.sync({ + session: { id: randomUUIDv7(), name: "Printful" }, + filter: { + printfulProductIds: [printfulProduct.id] + } + }); break; } case Printful.Webhook.Event.ProductDeleted: { @@ -257,7 +264,14 @@ export const app = new Elysia() app.listen(3000, async () => { if (cluster.worker?.id === 1) { log.info({ port: 3000 }, "server started") - await syncService.syncNext().catch((err) => log.error({ err }, "failed to resume product sync queue")); + + try { + const frontQueuedSync = await syncService.queue.front(); + if (frontQueuedSync) await syncService.sync(frontQueuedSync); + } + catch (err) { + log.error({ err }, "failed to resume product sync queue") + } } }); diff --git a/apps/api/src/services/sync.ts b/apps/api/src/services/sync.ts index 62e0c7c..ad8d5c5 100644 --- a/apps/api/src/services/sync.ts +++ b/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 = { + 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): 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 = { + 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 { + 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 + } + } } } diff --git a/apps/portal/src/routes/(app)/commerce/+page.svelte b/apps/portal/src/routes/(app)/commerce/+page.svelte index 8e6c83a..3b9d431 100644 --- a/apps/portal/src/routes/(app)/commerce/+page.svelte +++ b/apps/portal/src/routes/(app)/commerce/+page.svelte @@ -3,18 +3,23 @@ import { onMount } from "svelte"; import { api } from "$lib/api.js"; + type SyncStatus = "queued" | "active" | "failed" | "succeeded" | "none"; + const { data } = $props(); const synchronizer = $state({ - isRunning: false, + status: "none" as SyncStatus, syncingPrintfulProductIds: [] as number[], + get isInProgress() { + return this.status === "active" || this.status === "queued"; + }, poll: async function () { const res = await api.products.sync.get(); if (res.error) { - synchronizer.isRunning = false; + synchronizer.status = "none"; synchronizer.syncingPrintfulProductIds = []; } else { - synchronizer.isRunning = true; + synchronizer.status = res.data.status; synchronizer.syncingPrintfulProductIds = res.data.syncingPrintfulProductIds; } @@ -22,7 +27,7 @@ startPolling: () => { const intervalId = setInterval(async () => { await synchronizer.poll(); - if (!synchronizer.isRunning) clearInterval(intervalId); + if (!synchronizer.isInProgress) clearInterval(intervalId); }, 100); }, syncAll: async () => { @@ -37,7 +42,7 @@ onMount(async () => { await synchronizer.poll(); - if (synchronizer.isRunning) { + if (synchronizer.isInProgress) { synchronizer.startPolling(); } }); @@ -56,11 +61,11 @@

Loading...

{:then printfulProducts}