Apparel order fulfill

apps/api/src/server.ts

@@ -264,17 +264,15 @@ export const app = new Elysia()
             }
             case Printful.Webhook.Event.PackageShipped: {
                 const wOrderId = payload.data.order.external_id;
-                const shipInfo = payload.data.shipment;
-                log.info({ wOrderId, carrier: shipInfo.carrier, tracking: shipInfo.tracking_number }, "printful webhook: package shipped");
+                const shipment = payload.data.shipment;
+                log.info({ wOrderId, carrier: shipment.carrier, tracking: shipment.tracking_number }, "printful webhook: package shipped");
 
-                await s.Commerce.Webflow.Orders.update(wOrderId, {
-                    shippingTrackingURL: shipInfo.tracking_url,
-                    shippingTracking: shipInfo.tracking_number,
-                    shippingProvider: shipInfo.carrier,
-                });
-                await s.Commerce.Webflow.Orders.fulfill(wOrderId, {
-                    sendOrderFulfilledEmail: true,
+                await s.Commerce.Apparel.Orders.Queue.enqueue({
+                    type: "apparel_order_fulfill",
+                    source: "printful",
+                    payload: { wOrderId, shipment }
                 });
+
                 break;
             }
             default:
@@ -310,7 +308,7 @@ app.listen(3000, async () => {
         log.info({ port: 3000 }, "server started")
 
         // MANAGE QUEUES
-        const queues = [s.Commerce.Apparel.Syncs.Queue];
+        const queues = [s.Commerce.Apparel.Syncs.Queue, s.Commerce.Apparel.Orders.Queue];
         for (const queue of queues) {
             (async () => {
                 while (true) {

apps/api/src/services/commerce.ts

@@ -1,4 +1,4 @@
-import { PrintfulClient, ProductSyncer, Webflow, WebflowClient } from "@blade-and-brawn/commerce";
+import { Printful, 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";
@@ -46,7 +46,7 @@ class ApparelOrdersService {
             events: {
                 "apparel_order_create": defineEventType({
                     schemas: { payload: t.Object({ wOrder: t.Object({}) }), state: t.Null() },
-                    processor: async (id, payload, setState) => {
+                    processor: async (payload) => {
                         const wOrder = payload.wOrder as Webflow.Orders.Order;
 
                         await this.Printful.Orders.create({
@@ -70,6 +70,23 @@ class ApparelOrdersService {
                             })),
                         });
                     }
+                }),
+                "apparel_order_fulfill": defineEventType({
+                    schemas: { payload: t.Object({ wOrderId: t.String(), shipment: t.Object({}) }), state: t.Null() },
+                    processor: async (payload) => {
+                        const wOrderId = payload.wOrderId;
+                        const shipment = payload.shipment as Printful.Webhook.PackageShipped["data"]["shipment"];
+                        log.info({ wOrderId, carrier: shipment.carrier, tracking: shipment.tracking_number }, "printful webhook: package shipped");
+
+                        await this.Webflow.Orders.update(wOrderId, {
+                            shippingTrackingURL: shipment.tracking_url,
+                            shippingTracking: shipment.tracking_number,
+                            shippingProvider: shipment.carrier,
+                        });
+                        await this.Webflow.Orders.fulfill(wOrderId, {
+                            sendOrderFulfilledEmail: true,
+                        });
+                    }
                 })
             }
         });
@@ -111,7 +128,7 @@ class ApparelSyncService {
                             t.Null()
                         ])
                     },
-                    processor: async (id, payload, setState) => {
+                    processor: async (payload, setState) => {
                         await this.ProductSyncer.syncApparel({
                             filter: payload.filter,
                             beforeStep: async (pProductIds: number[]) => {
@@ -125,7 +142,7 @@ class ApparelSyncService {
                         payload: t.Object({ wProductId: t.String() }),
                         state: t.Union([t.Object({}), t.Null()])
                     },
-                    processor: async (id, payload, setState) => {
+                    processor: async (payload) => {
                         await this.Webflow.Products.remove(payload.wProductId);
                     }
                 })

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

@@ -10,7 +10,7 @@ export type EventSource = "portal" | "printful" | "webflow";
 export type EventStatus = "pending" | "processing" | "failed" | "fulfilled";
 
 type EventProcessor<P extends TSchema, S extends TSchema> =
-    (id: string, payload: Static<P>, setState: (state: Static<S>) => Promise<void>) => Promise<void>;
+    (payload: Static<P>, setState: (state: Static<S>) => Promise<void>, id: string) => Promise<void>;
 
 export type EventTypeOptions<P extends TSchema, S extends TSchema> = {
     schemas: { payload: P, state: S },
@@ -196,12 +196,12 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
         const cfg = this.events[type as keyof T]!;
         try {
             Value.Assert(cfg.schemas.payload, payload);
-            await cfg.processor(id, payload, async (state) => {
+            await cfg.processor(payload, async (state) => {
                 await db.updateTable("events")
                     .set({ state })
                     .where("id", "=", id)
                     .execute();
-            });
+            }, id);
         }
         catch (err) {
             await this.fail(id);