Enqueue product sync delete event

apps/api/src/server.ts

@@ -253,7 +253,12 @@ export const app = new Elysia()
                 log.info({ externalId: payload.data.sync_product.external_id, wProductId }, "printful webhook: product deleted");
                 if (!wProductId) throw new NotFoundError("Missing webflow product ID");
 
-                await s.Commerce.Webflow.Products.remove(wProductId);
+                await s.Commerce.ProductSync.Queue.enqueue({
+                    type: "product_sync_delete",
+                    source: "printful",
+                    payload: { wProductId }
+                });
+
                 break;
             }
             case Printful.Webhook.Event.PackageShipped: {

apps/api/src/services/commerce.ts

@@ -7,6 +7,7 @@ import type { Static } from "@sinclair/typebox";
 import { Value } from "@sinclair/typebox/value";
 import { sql } from "kysely";
 
+// product_sync_update
 const ProductSyncUpdatePayloadSchema = t.Object({
     session: t.Object({
         name: t.String(),
@@ -19,7 +20,6 @@ const ProductSyncUpdatePayloadSchema = t.Object({
     )
 });
 type ProductSyncUpdatePayload = Static<typeof ProductSyncUpdatePayloadSchema>;
-
 const ProductSyncUpdateStateSchema = t.Union([
     t.Object({
         pProductIds: t.Optional(t.Array(t.Number()))
@@ -28,9 +28,15 @@ const ProductSyncUpdateStateSchema = t.Union([
 ]);
 type ProductSyncUpdateState = Static<typeof ProductSyncUpdateStateSchema>;
 
+// product_sync_delete
+const ProductSyncDeletePayloadSchema = t.Object({ wProductId: t.String() });
+type ProductSyncDeletePayload = Static<typeof ProductSyncDeletePayloadSchema>;
+const ProductSyncDeleteStateSchema = t.Union([t.Object({}), t.Null()]);
+type ProductSyncDeleteState = Static<typeof ProductSyncDeleteStateSchema>;
+
 export type ProductSyncEvents = {
-    product_sync_update: EventSpec<ProductSyncUpdatePayload, ProductSyncUpdateState>
+    product_sync_update: EventSpec<ProductSyncUpdatePayload, ProductSyncUpdateState>;
-    // TODO: product_sync_delete, once its schema/processor are written
+    product_sync_delete: EventSpec<ProductSyncDeletePayload, ProductSyncDeleteState>;
 };
 
 export class CommerceService {
@@ -62,17 +68,23 @@ class ProductSyncService {
                 retry: { limit: 3, delayMs: 10000 },
                 concurrency: { limit: 1, timeoutMs: 600000 }
             },
-            types: {
+            events: {
                 "product_sync_update": {
                     schemas: { payload: ProductSyncUpdatePayloadSchema, state: ProductSyncUpdateStateSchema },
                     processor: async (id, payload, setState) => {
-                        await this.ProductSyncer.syncApparel({
+                        await this.ProductSyncer.sync({
                             filter: payload.filter,
                             beforeStep: async (pProductIds: number[]) => {
                                 await setState({ pProductIds });
                             }
                         });
                     }
+                },
+                "product_sync_delete": {
+                    schemas: { payload: ProductSyncDeletePayloadSchema, state: ProductSyncDeleteStateSchema },
+                    processor: async (id, payload, setState) => {
+                        await this.Webflow.Products.remove(payload.wProductId);
+                    }
                 }
             }
         });
@@ -82,6 +94,7 @@ class ProductSyncService {
         const latestSync = await db.selectFrom("events")
             .select(["started_at", "ended_at", "has_failed", "state"])
             .where("group", "=", ProductSyncService.EVENT_GROUP)
+            .where("type", "=", "product_sync_update")
             .where(sql<string>`payload->'session'->>'id'`, "=", sessionId)
             .orderBy(sql`(started_at is not null and ended_at is null)`, "desc")
             .orderBy("created_at", "desc")

apps/api/src/services/event-queue.ts

@@ -6,17 +6,17 @@ import { db } from "../database/db";
 import { Value } from "@sinclair/typebox/value";
 import type { TSchema } from "@sinclair/typebox";
 
-export type EventSpec<P extends JsonValue = JsonValue, S extends JsonValue = JsonValue> = {
+export type EventSpec<P extends JsonValue, S extends JsonValue> = {
     payload: P,
     state: S
 }
 export type EventSource = "portal" | "printful" | "webflow";
 
-type EventProcessor<S extends EventSpec> =
+type EventProcessor<S extends EventSpec<JsonValue, JsonValue>> =
     (id: string, payload: S["payload"], setState: (state: S["state"]) => Promise<void>) => Promise<void>;
 type EventStatus = "pending" | "processing" | "failed" | "fulfilled";
 
-type EventTypeOptions<S extends EventSpec> = {
+type EventTypeOptions<S extends EventSpec<JsonValue, JsonValue>> = {
     schemas: {
         payload: TSchema & { static: S["payload"] },
         state: TSchema & { static: S["state"] },
@@ -24,7 +24,7 @@ type EventTypeOptions<S extends EventSpec> = {
     processor: EventProcessor<S>,
 };
 
-type EventQueueOptions<G extends string, M extends Record<string, EventSpec>> = {
+type EventQueueOptions<G extends string, M extends Record<string, EventSpec<JsonValue, JsonValue>>> = {
     group: G,
     cfg: {
         retry: {
@@ -36,12 +36,12 @@ type EventQueueOptions<G extends string, M extends Record<string, EventSpec>> =
             timeoutMs: number,
         },
     },
-    types: {
+    events: {
         [T in keyof M]: EventTypeOptions<M[T]>
     },
 }
 
-type EnqueueOptions<M extends Record<string, EventSpec>, T extends keyof M & string> = {
+type EnqueueOptions<M extends Record<string, EventSpec<JsonValue, JsonValue>>, T extends keyof M & string> = {
     source: EventSource,
     type: T,
     payload: M[T]["payload"]
@@ -50,7 +50,7 @@ type EnqueueOptions<M extends Record<string, EventSpec>, T extends keyof M & str
 class AlreadyClaimedEventError extends Error { }
 class ConcurrencyLimitReachedError extends Error { }
 
-export class EventQueue<G extends string, M extends Record<string, EventSpec>> {
+export class EventQueue<G extends string, M extends Record<string, EventSpec<JsonValue, JsonValue>>> {
     private _lastCleanDate: Date = new Date(0);
     readonly cfg: EventQueueOptions<G, M>["cfg"];
     readonly group: G;
@@ -201,8 +201,8 @@ export class EventQueue<G extends string, M extends Record<string, EventSpec>> {
 
     private async handleEvent(id: string, type: string, payload: JsonValue) {
         log.info({ id, payload }, "handling event");
-        if (!(type in this.opt.types)) throw new Error(`Unknown event type: ${type}`);
+        if (!(type in this.opt.events)) throw new Error(`Unknown event type: ${type}`);
-        const cfg = this.opt.types[type as keyof M];
+        const cfg = this.opt.events[type as keyof M];
         try {
             Value.Assert(cfg.schemas.payload, payload);
             await cfg.processor(id, payload, async (state) => {

packages/commerce/src/sync.ts

@@ -37,7 +37,7 @@ export class ProductSyncer {
         this.log = (log ?? pino({ level: "silent" })).child({ component: "ProductSyncer" });
     }
 
-    async syncApparel(opt: ProductSyncerOptions = {}) {
+    async sync(opt: ProductSyncerOptions = {}) {
         const startDate = new Date();
         await opt.beforeStart?.(startDate);