some organization and implement apparel_order_created event

apps/api/src/server.ts

@@ -10,7 +10,6 @@ import {
     Printful,
     Webflow,
 } from "@blade-and-brawn/commerce";
-import zipcodesUs from "zipcodes-us";
 import { DEFAULT_NAME, env, log } from "./util";
 import serverTiming from "@elysia/server-timing";
 import jwt from "@elysia/jwt";
@@ -20,7 +19,6 @@ import { randomUUIDv7, sleep } from "bun";
 import { CalculatorService, CalculatorUnavailableError } from "./services/calculator";
 import { StandardsParamsSchema } from "@blade-and-brawn/calculator";
 import { StandardsService } from "./services/standards";
-import type { EventQueue, EventTypeOptions } from "./services/event-queue";
 
 // CONSTANTS
 // -----------------------
@@ -196,14 +194,14 @@ export const app = new Elysia()
             .group("/sync", (app) => app
                 // Sync status
                 .get("/", async ({ sessionId }) => {
-                    const latestSyncState = await s.Commerce.ProductSync.getLatestSyncState(sessionId);
+                    const latestSyncState = await s.Commerce.Apparel.Syncs.getLatestSyncState(sessionId);
                     if (!latestSyncState) throw new NotFoundError("No product sync found for the provided session");
                     return latestSyncState;
                 })
                 // Run sync
                 .post("/:pProductId?", async ({ params: { pProductId }, sessionId }) => {
-                    await s.Commerce.ProductSync.Queue.enqueue({
+                    await s.Commerce.Apparel.Syncs.Queue.enqueue({
-                        type: "product_sync_update",
+                        type: "apparel_sync_update",
                         source: "portal",
                         payload: {
                             session: { id: sessionId, name: "Portal" },
@@ -230,6 +228,7 @@ export const app = new Elysia()
         ))
 
     // WEBHOOKS
+    // TODO: account for automatic webhook retries from the webhook sender end (especially timeouts)
     .post("/webhooks/printful", async ({ body }) => {
         // https://webflow.com/integrations/printful
         const payload = body as Printful.Webhook.EventPayload;
@@ -239,8 +238,8 @@ export const app = new Elysia()
                 const pProduct = payload.data.sync_product;
                 log.info({ productId: pProduct.id }, "printful webhook: product updated");
 
-                await s.Commerce.ProductSync.Queue.enqueue({
+                await s.Commerce.Apparel.Syncs.Queue.enqueue({
-                    type: "product_sync_update",
+                    type: "apparel_sync_update",
                     source: "printful",
                     payload: {
                         session: { id: randomUUIDv7(), name: "Printful" },
@@ -255,8 +254,8 @@ 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.ProductSync.Queue.enqueue({
+                await s.Commerce.Apparel.Syncs.Queue.enqueue({
-                    type: "product_sync_delete",
+                    type: "apparel_sync_delete",
                     source: "printful",
                     payload: { wProductId }
                 });
@@ -293,25 +292,10 @@ export const app = new Elysia()
                 const wOrder = payload.payload;
                 log.info({ orderId: wOrder.orderId }, "webflow webhook: order created");
 
-                await s.Commerce.Printful.Orders.create({
+                await s.Commerce.Apparel.Orders.Queue.enqueue({
-                    external_id: wOrder.orderId,
+                    type: "apparel_order_create",
-                    // TODO: derive from webflow
+                    source: "webflow",
-                    shipping: "STANDARD",
+                    payload: { wOrder }
-                    recipient: {
-                        name: wOrder.shippingAddress.addressee,
-                        address1: wOrder.shippingAddress.line1,
-                        address2: wOrder.shippingAddress.line2,
-                        city: wOrder.shippingAddress.city,
-                        state_code: zipcodesUs.find(
-                            wOrder.shippingAddress.postalCode.split("-")[0] ?? "",
-                        ).stateCode,
-                        country_code: wOrder.shippingAddress.country,
-                        zip: wOrder.shippingAddress.postalCode,
-                    },
-                    items: wOrder.purchasedItems.map((wOrderSku) => ({
-                        external_variant_id: wOrderSku.variantId,
-                        quantity: wOrderSku.count,
-                    })),
                 });
 
                 break;
@@ -326,7 +310,7 @@ app.listen(3000, async () => {
         log.info({ port: 3000 }, "server started")
 
         // MANAGE QUEUES
-        const queues: EventQueue<string, Record<string, EventTypeOptions<any, any>>>[] = [s.Commerce.ProductSync.Queue];
+        const queues = [s.Commerce.Apparel.Syncs.Queue];
         for (const queue of queues) {
             (async () => {
                 while (true) {

apps/api/src/services/commerce.ts

@@ -1,10 +1,14 @@
-import { PrintfulClient, ProductSyncer, WebflowClient } from "@blade-and-brawn/commerce";
+import { PrintfulClient, ProductSyncer, Webflow, WebflowClient } from "@blade-and-brawn/commerce";
 import { env, log } from "../util";
 import { db } from "../database/db";
 import { defineEventType, EventQueue } from "./event-queue";
 import { Type as t } from "@sinclair/typebox";
 import { Value } from "@sinclair/typebox/value";
 import { sql } from "kysely";
+import zipcodesUs from "zipcodes-us";
+
+// TODO: read Retry-After header for how long to delay retry on 429 Too many Requests
+// https://developers.webflow.com/data/reference/rate-limits ...(do printful too)
 
 export class CommerceService {
     readonly Printful = new PrintfulClient({
@@ -17,11 +21,62 @@ export class CommerceService {
         token: env.WEBFLOW_AUTH,
         webhookSecret: env.WEBFLOW_WEBHOOK_SECRET,
     });
-    readonly ProductSync = new ProductSyncService(this.Printful, this.Webflow);
+    readonly Apparel = new ApparelService(this.Printful, this.Webflow);
 }
 
-class ProductSyncService {
+class ApparelService {
-    static readonly EVENT_GROUP = "product_sync";
+    Syncs: ApparelSyncService;
+    Orders: ApparelOrdersService;
+    constructor(private readonly Printful: PrintfulClient, private readonly Webflow: WebflowClient) {
+        this.Syncs = new ApparelSyncService(this.Printful, this.Webflow);
+        this.Orders = new ApparelOrdersService(this.Printful, this.Webflow);
+    }
+}
+
+class ApparelOrdersService {
+    readonly Queue;
+
+    constructor(private readonly Printful: PrintfulClient, private readonly Webflow: WebflowClient) {
+        this.Queue = new EventQueue({
+            group: "apparel_order",
+            cfg: {
+                retry: { limit: 3, delayMs: 10000 },
+                concurrency: { limit: 1, timeoutMs: 600000 }
+            },
+            events: {
+                "apparel_order_create": defineEventType({
+                    schemas: { payload: t.Object({ wOrder: t.Object({}) }), state: t.Null() },
+                    processor: async (id, payload, setState) => {
+                        const wOrder = payload.wOrder as Webflow.Orders.Order;
+
+                        await this.Printful.Orders.create({
+                            external_id: wOrder.orderId,
+                            // TODO: derive from webflow
+                            shipping: "STANDARD",
+                            recipient: {
+                                name: wOrder.shippingAddress.addressee,
+                                address1: wOrder.shippingAddress.line1,
+                                address2: wOrder.shippingAddress.line2,
+                                city: wOrder.shippingAddress.city,
+                                state_code: zipcodesUs.find(
+                                    wOrder.shippingAddress.postalCode.split("-")[0] ?? "",
+                                ).stateCode,
+                                country_code: wOrder.shippingAddress.country,
+                                zip: wOrder.shippingAddress.postalCode,
+                            },
+                            items: wOrder.purchasedItems.map((wOrderSku) => ({
+                                external_variant_id: wOrderSku.variantId,
+                                quantity: wOrderSku.count,
+                            })),
+                        });
+                    }
+                })
+            }
+        });
+    }
+}
+
+class ApparelSyncService {
     readonly Queue;
     private readonly ProductSyncer: ProductSyncer;
 
@@ -30,13 +85,13 @@ class ProductSyncService {
     constructor(private readonly Printful: PrintfulClient, private readonly Webflow: WebflowClient) {
         this.ProductSyncer = new ProductSyncer(this.Printful, this.Webflow, log);
         this.Queue = new EventQueue({
-            group: ProductSyncService.EVENT_GROUP,
+            group: "apparel_sync",
             cfg: {
                 retry: { limit: 3, delayMs: 10000 },
                 concurrency: { limit: 1, timeoutMs: 600000 }
             },
             events: {
-                "product_sync_update": defineEventType({
+                "apparel_sync_update": defineEventType({
                     schemas: {
                         payload: t.Object({
                             session: t.Object({
@@ -57,7 +112,7 @@ class ProductSyncService {
                         ])
                     },
                     processor: async (id, payload, setState) => {
-                        await this.ProductSyncer.sync({
+                        await this.ProductSyncer.syncApparel({
                             filter: payload.filter,
                             beforeStep: async (pProductIds: number[]) => {
                                 await setState({ pProductIds });
@@ -65,7 +120,7 @@ class ProductSyncService {
                         });
                     }
                 }),
-                "product_sync_delete": defineEventType({
+                "apparel_sync_delete": defineEventType({
                     schemas: {
                         payload: t.Object({ wProductId: t.String() }),
                         state: t.Union([t.Object({}), t.Null()])
@@ -81,8 +136,8 @@ class ProductSyncService {
     async getLatestSyncState(sessionId: string) {
         const latestSync = await db.selectFrom("events")
             .select(["started_at", "ended_at", "has_failed", "state"])
-            .where("group", "=", ProductSyncService.EVENT_GROUP)
+            .where("group", "=", this.Queue.group)
-            .where("type", "=", "product_sync_update")
+            .where("type", "=", "apparel_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")
@@ -90,7 +145,7 @@ class ProductSyncService {
             .executeTakeFirst();
         if (!latestSync) return;
 
-        Value.Assert(this.Queue.events["product_sync_update"].schemas.state, latestSync.state);
+        Value.Assert(this.Queue.events["apparel_sync_update"].schemas.state, latestSync.state);
 
         return {
             startDate: latestSync.started_at,

packages/commerce/src/sync.ts

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