/** * Background worker — run as a separate process (`npm run worker`), one or more instances. * * Every minute: * - release expired holds (capacity returns even if no shopper touches that date) * - send due reminders (claimed row-by-row, safe with several workers) * Every 15 minutes: * - reconcile: recent orders that carry a selection but have no booking (missed/failed * webhooks) are replayed through the same handler, so nothing is silently lost. */ import prisma from "../db.server"; import { handleOrderCreate, type OrderPayload } from "../services/orders.server"; import { runDueReminders } from "../services/reminders.server"; import { ReservationService } from "../services/reservations.server"; import { SELECTION_ATTR } from "../services/settings"; import shopify from "../shopify.server"; import { pruneAccessLogs } from "../lib/audit.server"; const reservations = new ReservationService(prisma); let stopping = false; async function tick() { const now = new Date(); const released = await reservations.sweepExpired(now).catch((e) => (console.error("sweep", e), 0)); const sent = await runDueReminders(now).catch((e) => (console.error("reminders", e), 0)); if (released || sent) console.info(`[worker] released ${released} holds, sent ${sent} reminders`); } async function reconcile() { const shops = await prisma.shop.findMany({ where: { uninstalledAt: null }, select: { id: true, domain: true } }); for (const s of shops) { try { const { admin } = await shopify.unauthenticated.admin(s.domain); const since = new Date(Date.now() - 6 * 3600_000).toISOString(); const res = await admin.graphql( `#graphql query($q: String!) { orders(first: 100, query: $q, sortKey: CREATED_AT, reverse: true) { nodes { id legacyResourceId name email phone customAttributes { key value } customer { firstName lastName defaultEmailAddress { emailAddress } defaultPhoneNumber { phoneNumber } } shippingAddress { address1 address2 city zip latitude longitude phone firstName lastName } lineItems(first: 100) { nodes { title quantity originalUnitPriceSet { shopMoney { amount } } product { legacyResourceId } variant { legacyResourceId } } } } } }`, { variables: { q: `created_at:>='${since}'` } }, ); const orders = ((await res.json()) as { data: { orders: { nodes: GqlOrder[] } } }).data.orders.nodes; const ids = orders.map((o) => o.id); const have = new Set((await prisma.booking.findMany({ where: { shopId: s.id, orderId: { in: ids } }, select: { orderId: true } })).map((b) => b.orderId)); for (const o of orders) { if (have.has(o.id) || !o.customAttributes.some((a) => a.key === SELECTION_ATTR)) continue; console.warn(`[reconcile] ${s.domain} ${o.name} had no booking; replaying`); await handleOrderCreate(s.domain, toPayload(o), admin); } } catch (e) { console.error(`[reconcile] ${s.domain}`, e); } } } interface GqlOrder { id: string; legacyResourceId: string; name: string; email: string | null; phone: string | null; customAttributes: { key: string; value: string }[]; customer: { firstName: string | null; lastName: string | null; defaultEmailAddress: { emailAddress: string } | null; defaultPhoneNumber: { phoneNumber: string } | null } | null; shippingAddress: { address1: string; address2: string; city: string; zip: string; latitude: number | null; longitude: number | null; phone: string | null; firstName: string; lastName: string } | null; lineItems: { nodes: { title: string; quantity: number; originalUnitPriceSet: { shopMoney: { amount: string } }; product: { legacyResourceId: string } | null; variant: { legacyResourceId: string } | null }[] }; } function toPayload(o: GqlOrder): OrderPayload { const a = o.shippingAddress; return { id: Number(o.legacyResourceId), admin_graphql_api_id: o.id, name: o.name, email: o.email, phone: o.phone, note_attributes: o.customAttributes.map((x) => ({ name: x.key, value: x.value })), customer: o.customer ? { first_name: o.customer.firstName, last_name: o.customer.lastName, email: o.customer.defaultEmailAddress?.emailAddress ?? null, phone: o.customer.defaultPhoneNumber?.phoneNumber ?? null } : null, shipping_address: a ? { address1: a.address1, address2: a.address2, city: a.city, zip: a.zip, latitude: a.latitude, longitude: a.longitude, phone: a.phone, first_name: a.firstName, last_name: a.lastName } : null, line_items: o.lineItems.nodes.map((l) => ({ title: l.title, quantity: l.quantity, price: l.originalUnitPriceSet.shopMoney.amount, product_id: l.product ? Number(l.product.legacyResourceId) : null, variant_id: l.variant ? Number(l.variant.legacyResourceId) : null, })), }; } async function main() { console.info("[worker] started"); let n = 0; while (!stopping) { const started = Date.now(); await tick(); if (n % 1440 === 0) await pruneAccessLogs().catch((e) => console.error("prune access logs", e)); // daily if (n++ % 15 === 0) await reconcile(); await new Promise((r) => setTimeout(r, Math.max(1000, 60_000 - (Date.now() - started)))); } } process.on("SIGTERM", () => (stopping = true)); process.on("SIGINT", () => (stopping = true)); main().catch((e) => { console.error(e); process.exit(1); });