diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index a0013a0..11fddc1 100644 --- a/apps/api/src/server.ts +++ b/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({ - type: "product_sync_update", + await s.Commerce.Apparel.Syncs.Queue.enqueue({ + 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({ - type: "product_sync_update", + await s.Commerce.Apparel.Syncs.Queue.enqueue({ + 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({ - type: "product_sync_delete", + await s.Commerce.Apparel.Syncs.Queue.enqueue({ + 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({ - 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, - })), + await s.Commerce.Apparel.Orders.Queue.enqueue({ + type: "apparel_order_create", + source: "webflow", + payload: { wOrder } }); break; @@ -326,7 +310,7 @@ app.listen(3000, async () => { log.info({ port: 3000 }, "server started") // MANAGE QUEUES - const queues: EventQueue>>[] = [s.Commerce.ProductSync.Queue]; + const queues = [s.Commerce.Apparel.Syncs.Queue]; for (const queue of queues) { (async () => { while (true) { diff --git a/apps/api/src/services/commerce.ts b/apps/api/src/services/commerce.ts index 87b6be1..3725d2a 100644 --- a/apps/api/src/services/commerce.ts +++ b/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 { - static readonly EVENT_GROUP = "product_sync"; +class ApparelService { + 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("type", "=", "product_sync_update") + .where("group", "=", this.Queue.group) + .where("type", "=", "apparel_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") @@ -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, diff --git a/packages/commerce/src/sync.ts b/packages/commerce/src/sync.ts index 21cb976..91660ab 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 sync(opt: ProductSyncerOptions = {}) { + async syncApparel(opt: ProductSyncerOptions = {}) { const startDate = new Date(); await opt.beforeStart?.(startDate);