datewise/app/jobs/worker.ts

/**
 * 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);
});