From fd94f22d21cd670c73772d415c7e6d27deee9de7 Mon Sep 17 00:00:00 2001 From: Dominic Ferrando Date: Mon, 13 Jul 2026 16:30:21 -0400 Subject: [PATCH] Implement custom retry handling of rate limit errors --- .../api/src/database/migrations/2026-06-03.ts | 1 + apps/api/src/services/commerce.ts | 7 +++--- apps/api/src/services/event-queue.ts | 24 ++++++++++++++----- packages/commerce/package.json | 4 +++- packages/commerce/src/printful.ts | 15 ++++++++++++ packages/commerce/src/webflow.ts | 20 ++++++++++++++++ packages/domain/src/utils.ts | 19 +++++++++++++++ 7 files changed, 80 insertions(+), 10 deletions(-) diff --git a/apps/api/src/database/migrations/2026-06-03.ts b/apps/api/src/database/migrations/2026-06-03.ts index bcfe1d7..9b7db97 100644 --- a/apps/api/src/database/migrations/2026-06-03.ts +++ b/apps/api/src/database/migrations/2026-06-03.ts @@ -14,6 +14,7 @@ export async function up(db: Kysely): Promise { .addColumn("ended_at", "timestamptz") .addColumn("available_at", "timestamptz", (cb) => cb.notNull().defaultTo(sql`now()`)) .addColumn("tries", "integer", (cb) => cb.notNull().defaultTo(0)) + .addColumn("rate_limits", "integer", (cb) => cb.notNull().defaultTo(0)) .addColumn("has_failed", "boolean", (cb) => cb.notNull().defaultTo(false)) .execute() diff --git a/apps/api/src/services/commerce.ts b/apps/api/src/services/commerce.ts index b5d5d3b..5aa59a5 100644 --- a/apps/api/src/services/commerce.ts +++ b/apps/api/src/services/commerce.ts @@ -7,9 +7,6 @@ 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({ token: env.PRINTFUL_AUTH, @@ -49,6 +46,7 @@ class ApparelOrdersService { processor: async (payload) => { const wOrder = payload.wOrder as Webflow.Orders.Order; + // creates a DRAFT order await this.Printful.Orders.create({ external_id: wOrder.orderId, // TODO: derive from webflow @@ -69,6 +67,9 @@ class ApparelOrdersService { quantity: wOrderSku.count, })), }); + + // TODO: confirm the above created order + // right now we manually review and confirm it in the printful portal } }), "apparel_order_fulfill": defineEventType({ diff --git a/apps/api/src/services/event-queue.ts b/apps/api/src/services/event-queue.ts index d02bf45..83c0d7e 100644 --- a/apps/api/src/services/event-queue.ts +++ b/apps/api/src/services/event-queue.ts @@ -5,6 +5,7 @@ import { sleep } from "bun"; import { db } from "../database/db"; import { AssertError, Value } from "@sinclair/typebox/value"; import type { TSchema, Static } from "@sinclair/typebox"; +import { RateLimitError } from "@blade-and-brawn/domain"; export type EventSource = "portal" | "printful" | "webflow"; export type EventStatus = "waiting" | "available" | "processing" | "failed" | "fulfilled"; @@ -38,6 +39,7 @@ export function defineEventType

(opt: Event export class EventQueue>> { private static readonly DEQUEUE_RETRY_DELAY_MS = 10000; // 10 seconds private static readonly DEQUEUE_RETRY_LIMIT = 3; + private static readonly RATE_LIMIT_RETRY_LIMIT = 15; private _lastCleanDate: Date = new Date(0); readonly cfg: EventQueueCfg; @@ -179,7 +181,7 @@ export class EventQueue) { const result = await (trx ?? db).selectFrom("events") - .select(["id", "type", "payload", "available_at", "tries"]) + .select(["id", "type", "payload", "available_at", "tries", "rate_limits"]) .where("group", "=", this.group) .where("started_at", "is", null) .where("available_at", "<=", new Date()) @@ -220,7 +222,7 @@ export class EventQueue, "id" | "type" | "tries" | "payload">) { + private async processEvent(event: Pick, "id" | "type" | "tries" | "payload" | "rate_limits">) { log.info({ id: event.id, type: event.type, payload: event.payload }, "processing event"); if (!(event.type in this.events)) throw new Error(`event type not registered in this queue (${this.group}): ${event.type}`); const cfg = this.events[event.type as keyof T]!; @@ -231,15 +233,24 @@ export class EventQueue ({ tries: eb("tries", "+", 1), + rate_limits: opt.isRateLimit ? eb("rate_limits", "+", 1) : eb.ref("rate_limits"), started_at: null, ended_at: null, has_failed: false, - available_at: new Date(Date.now() + this.cfg.retry.delayMs) + available_at: new Date(Date.now() + delayMs) })) .where("id", "=", id) .execute(); diff --git a/packages/commerce/package.json b/packages/commerce/package.json index 9c0dcdc..309f8c3 100644 --- a/packages/commerce/package.json +++ b/packages/commerce/package.json @@ -6,5 +6,7 @@ "exports": { ".": "./src/index.ts" }, - "dependencies": {} + "dependencies": { + "@blade-and-brawn/domain": "workspace:*" + } } diff --git a/packages/commerce/src/printful.ts b/packages/commerce/src/printful.ts index 1e3f8d6..4a656da 100644 --- a/packages/commerce/src/printful.ts +++ b/packages/commerce/src/printful.ts @@ -1,5 +1,6 @@ import type { Printful } from "./util/types"; import { type DeepPartial, type PagingOptions } from "./util/misc"; +import { RateLimitError, parseRetryAfterMs } from "@blade-and-brawn/domain"; const parsePayload = (res: Response): Promise => res.json().catch(() => res.text().catch(() => undefined)); @@ -26,6 +27,12 @@ export class PrintfulError extends Error { } } +class PrintfulRateLimitError extends RateLimitError { + constructor(retryAfterHeader: string | null, payload: unknown) { + super("Printful", parseRetryAfterMs(retryAfterHeader, 60_000), payload); + } +} + class VariantsClient { constructor(private creds: Credentials) { } @@ -34,6 +41,7 @@ class VariantsClient { method: "GET", headers: this.creds.authHeaders, }); + if (res.status === 429) throw new PrintfulRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (res.status === 404 || res.status === 400) return; if (!res.ok) throw new PrintfulError("Failed to get Printful variant", res.status, await parsePayload(res)); const payload = (await res.json()) as Printful.MetaDataSingle; @@ -56,6 +64,7 @@ class ProductsClient { method: "GET", headers: this.creds.authHeaders, }); + if (res.status === 429) throw new PrintfulRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new PrintfulError("Failed to list Printful products", res.status, await parsePayload(res)); const payload = (await res.json()) as Printful.MetaDataMulti; allSyncProducts.push(...payload.result); @@ -70,6 +79,7 @@ class ProductsClient { method: "GET", headers: this.creds.authHeaders, }); + if (res.status === 429) throw new PrintfulRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (res.status === 404 || res.status === 400) return; if (!res.ok) throw new PrintfulError("Failed to get Printful product", res.status, await parsePayload(res)); const payload = (await res.json()) as Printful.MetaDataSingle; @@ -85,6 +95,7 @@ class ProductsClient { }, body: JSON.stringify(pProduct), }); + if (res.status === 429) throw new PrintfulRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new PrintfulError("Failed to update Printful product", res.status, await parsePayload(res)); } } @@ -101,6 +112,7 @@ class OrdersClient { }, body: JSON.stringify(pOrder), }); + if (res.status === 429) throw new PrintfulRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new PrintfulError("Failed to create Printful order", res.status, await parsePayload(res)); } @@ -112,6 +124,7 @@ class OrdersClient { method: "GET", headers: this.creds.authHeaders, }); + if (res.status === 429) throw new PrintfulRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new PrintfulError("Failed to list Printful orders", res.status, await parsePayload(res)); const payload = (await res.json()) as Printful.MetaDataMulti; allPOrders.push(...payload.result); @@ -133,6 +146,7 @@ class WebhooksClient { ...this.creds.authHeaders, } }); + if (res.status === 429) throw new PrintfulRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (res.status === 404 || res.status === 400) return; if (!res.ok) throw new PrintfulError("Failed to get webhook configuration", res.status, await parsePayload(res)); const payload = (await res.json()) as Printful.Webhook.Config; @@ -155,6 +169,7 @@ class WebhooksClient { "types": events }), }); + if (res.status === 429) throw new PrintfulRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new PrintfulError("Failed to register webhook URL", res.status, await parsePayload(res)); } } diff --git a/packages/commerce/src/webflow.ts b/packages/commerce/src/webflow.ts index 1d9e134..d245ff3 100644 --- a/packages/commerce/src/webflow.ts +++ b/packages/commerce/src/webflow.ts @@ -1,6 +1,7 @@ import type { Webflow } from "./util/types"; import { type DeepPartial, type PagingOptions } from "./util/misc"; import crypto from "node:crypto"; +import { RateLimitError, parseRetryAfterMs } from "@blade-and-brawn/domain"; const parsePayload = (res: Response): Promise => res.json().catch(() => res.text().catch(() => undefined)); @@ -31,6 +32,12 @@ export class WebflowError extends Error { } } +class WebflowRateLimitError extends RateLimitError { + constructor(retryAfterHeader: string | null, payload: unknown) { + super("Webflow", parseRetryAfterMs(retryAfterHeader, 60_000), payload); + } +} + class SkusClient { constructor(private creds: Credentials) { } @@ -49,6 +56,7 @@ class SkusClient { body: JSON.stringify({ skus }), }, ); + if (res.status === 429) throw new WebflowRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new WebflowError("Failed to create Webflow product SKU", res.status, await parsePayload(res)); const payload = (await res.json()) as { skus: { id: string }[] }; return payload.skus.map((sku) => sku.id); @@ -70,6 +78,7 @@ class SkusClient { body: JSON.stringify({ sku: wSku }), }, ); + if (res.status === 429) throw new WebflowRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new WebflowError("Failed to update Webflow product SKU", res.status, await parsePayload(res)); } } @@ -92,6 +101,7 @@ class ProductsClient { }, body: JSON.stringify(wProductAndSku), }); + if (res.status === 429) throw new WebflowRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new WebflowError("Failed to create Webflow product", res.status, await parsePayload(res)); const payload = (await res.json()) as { product: { id: string } }; return payload.product.id; @@ -105,6 +115,7 @@ class ProductsClient { method: "GET", headers: this.creds.authHeader, }); + if (res.status === 429) throw new WebflowRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new WebflowError("Failed to list Webflow products", res.status, await parsePayload(res)); const payload = (await res.json()) as Webflow.MetaDataMulti<"items", Webflow.Products.ProductAndSkus>; allWProducts.push(...payload.items); @@ -119,6 +130,7 @@ class ProductsClient { method: "GET", headers: this.creds.authHeader, }); + if (res.status === 429) throw new WebflowRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (res.status === 404 || res.status === 400) return; if (!res.ok) throw new WebflowError("Failed to get Webflow product", res.status, await parsePayload(res)); return (await res.json()) as Webflow.Products.ProductAndSkus; @@ -136,6 +148,7 @@ class ProductsClient { }, body: JSON.stringify(wProduct), }); + if (res.status === 429) throw new WebflowRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new WebflowError("Failed to update Webflow product", res.status, await parsePayload(res)); } @@ -151,6 +164,7 @@ class ProductsClient { body: JSON.stringify({}), }, ); + if (res.status === 429) throw new WebflowRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new WebflowError("Failed to remove Webflow product", res.status, await parsePayload(res)); } } @@ -170,6 +184,7 @@ class OrdersClient { method: "GET", headers: this.creds.authHeader, }); + if (res.status === 429) throw new WebflowRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new WebflowError("Failed to list Webflow orders", res.status, await parsePayload(res)); const payload = (await res.json()) as Webflow.MetaDataMulti<"orders", Webflow.Orders.Order>; allOrders.push(...payload.orders); @@ -196,6 +211,7 @@ class OrdersClient { }, body: JSON.stringify(wOrderUpdate), }); + if (res.status === 429) throw new WebflowRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new WebflowError("Failed to update Webflow order", res.status, await parsePayload(res)); } @@ -211,6 +227,7 @@ class OrdersClient { }, body: JSON.stringify(opt), }); + if (res.status === 429) throw new WebflowRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new WebflowError("Failed to fulfill Webflow order", res.status, await parsePayload(res)); } } @@ -226,6 +243,7 @@ class WebhooksClient { method: "GET", headers: this.creds.authHeader, }); + if (res.status === 429) throw new WebflowRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new WebflowError("Failed to list Webflow webhooks", res.status, await parsePayload(res)); const payload = (await res.json()) as Webflow.MetaDataMulti<"webhooks", Webflow.Webhook.Config>; allWebhooks.push(...payload.webhooks); @@ -250,6 +268,7 @@ class WebhooksClient { "triggerType": event }), }); + if (res.status === 429) throw new WebflowRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new WebflowError("Failed to create Webflow webhook", res.status, await parsePayload(res)); } @@ -265,6 +284,7 @@ class WebhooksClient { body: JSON.stringify({}), }, ); + if (res.status === 429) throw new WebflowRateLimitError(res.headers.get("Retry-After"), await parsePayload(res)); if (!res.ok) throw new WebflowError("Failed to remove Webflow webhook", res.status, await parsePayload(res)); } } diff --git a/packages/domain/src/utils.ts b/packages/domain/src/utils.ts index 6c6d256..dd6de05 100644 --- a/packages/domain/src/utils.ts +++ b/packages/domain/src/utils.ts @@ -26,3 +26,22 @@ export const range = (length: number) => export const clamp = (x: number, lo: number, hi: number) => Math.max(lo, Math.min(hi, x)); + +export const parseRetryAfterMs = (header: string | null, defaultMs: number): number => { + if (header === null) return defaultMs; + + if (/^\d+$/.test(header.trim())) return secToMs(Number(header)); + + const date = new Date(header); + if (!Number.isNaN(date.getTime())) return clamp(date.getTime() - Date.now(), 0, Infinity); + + return defaultMs; +}; + +export class RateLimitError extends Error { + status: number = 429; + + constructor(public source: string, public retryDelayMs: number, public payload: unknown) { + super(`${source}: Too many requests`); + } +}