diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index 5f298c6..66ef883 100644 --- a/apps/api/src/server.ts +++ b/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: { diff --git a/apps/api/src/services/commerce.ts b/apps/api/src/services/commerce.ts index 1de787e..f75f9a6 100644 --- a/apps/api/src/services/commerce.ts +++ b/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; - const ProductSyncUpdateStateSchema = t.Union([ t.Object({ pProductIds: t.Optional(t.Array(t.Number())) @@ -28,9 +28,15 @@ const ProductSyncUpdateStateSchema = t.Union([ ]); type ProductSyncUpdateState = Static; +// product_sync_delete +const ProductSyncDeletePayloadSchema = t.Object({ wProductId: t.String() }); +type ProductSyncDeletePayload = Static; +const ProductSyncDeleteStateSchema = t.Union([t.Object({}), t.Null()]); +type ProductSyncDeleteState = Static; + export type ProductSyncEvents = { - product_sync_update: EventSpec - // TODO: product_sync_delete, once its schema/processor are written + product_sync_update: EventSpec; + product_sync_delete: EventSpec; }; 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`payload->'session'->>'id'`, "=", sessionId) .orderBy(sql`(started_at is not null and ended_at is null)`, "desc") .orderBy("created_at", "desc") diff --git a/apps/api/src/services/event-queue.ts b/apps/api/src/services/event-queue.ts index 382cc1b..c432682 100644 --- a/apps/api/src/services/event-queue.ts +++ b/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

= { +export type EventSpec

= { payload: P, state: S } export type EventSource = "portal" | "printful" | "webflow"; -type EventProcessor = +type EventProcessor> = (id: string, payload: S["payload"], setState: (state: S["state"]) => Promise) => Promise; type EventStatus = "pending" | "processing" | "failed" | "fulfilled"; -type EventTypeOptions = { +type EventTypeOptions> = { schemas: { payload: TSchema & { static: S["payload"] }, state: TSchema & { static: S["state"] }, @@ -24,7 +24,7 @@ type EventTypeOptions = { processor: EventProcessor, }; -type EventQueueOptions> = { +type EventQueueOptions>> = { group: G, cfg: { retry: { @@ -36,12 +36,12 @@ type EventQueueOptions> = timeoutMs: number, }, }, - types: { + events: { [T in keyof M]: EventTypeOptions }, } -type EnqueueOptions, T extends keyof M & string> = { +type EnqueueOptions>, T extends keyof M & string> = { source: EventSource, type: T, payload: M[T]["payload"] @@ -50,7 +50,7 @@ type EnqueueOptions, T extends keyof M & str class AlreadyClaimedEventError extends Error { } class ConcurrencyLimitReachedError extends Error { } -export class EventQueue> { +export class EventQueue>> { private _lastCleanDate: Date = new Date(0); readonly cfg: EventQueueOptions["cfg"]; readonly group: G; @@ -201,8 +201,8 @@ export class EventQueue> { 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}`); - const cfg = this.opt.types[type as keyof M]; + if (!(type in this.opt.events)) throw new Error(`Unknown event type: ${type}`); + const cfg = this.opt.events[type as keyof M]; try { Value.Assert(cfg.schemas.payload, payload); await cfg.processor(id, payload, async (state) => { diff --git a/packages/commerce/src/sync.ts b/packages/commerce/src/sync.ts index 91660ab..21cb976 100644 --- a/packages/commerce/src/sync.ts +++ b/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);