Offline-first POS, part 4: idempotent sync and the last-item problem

INSERT INTO sales (id, device_id, cashier_id, occurred_at, total_cents, ...)
VALUES ($1, $2, $3, $4, $5, ...)
ON CONFLICT (id) DO NOTHING
RETURNING id;

import { Router } from "express";
import { dataSource } from "../db";
import { Sale } from "../entities/Sale";
import { z } from "zod";

const SaleInput = z.object({
  id: z.string().uuid(),
  deviceId: z.string().min(1),
  cashierId: z.string().uuid(),
  occurredAt: z.string().datetime(),
  totalCents: z.number().int().nonnegative(),
  paymentMethod: z.enum(["cash", "khqr", "card"]),
  paymentState: z.enum(["paid", "pending_verification"]),
  lines: z
    .array(
      z.object({
        productId: z.string().uuid(),
        quantity: z.number().int().positive(),
        unitPriceCents: z.number().int().nonnegative(),
      }),
    )
    .min(1),
});

const BatchInput = z.object({ sales: z.array(SaleInput).max(200) });

export const syncRouter = Router();

syncRouter.post("/sync/sales", async (req, res) => {
  const parsed = BatchInput.safeParse(req.body);
  if (!parsed.success) {
    return res
      .status(400)
      .json({ ok: false, error: "invalid", details: parsed.error.flatten() });
  }

  const accepted: string[] = [];
  const rejected: { id: string; reason: string }[] = [];

  // Each sale gets its own transaction. One malformed sale in a batch of fifty
  // must not roll back the other forty-nine — the till would resend all of them
  // forever and never make progress.
  for (const sale of parsed.data.sales) {
    try {
      await persistSale(sale);
      accepted.push(sale.id);
    } catch (err) {
      rejected.push({ id: sale.id, reason: describe(err) });
    }
  }

  if (accepted.length > 0) void refreshStockView();

  return res.json({ ok: true, accepted, rejected });
});

{
  "ok": true,
  "accepted": ["0191f3a2-...", "0191f3a3-..."],
  "rejected": [{ "id": "0191f3a4-...", "reason": "unknown_product" }]
}

/**
 * Inserts a sale, its lines and its stock movements in one transaction.
 *
 * A duplicate is a no-op, not an error: the till is retrying because it never
 * saw our acknowledgement, and it deserves a clean 'accepted' so it can stop.
 */
export async function persistSale(input: SaleInput): Promise<void> {
  await dataSource.transaction(async (manager) => {
    const result = await manager
      .createQueryBuilder()
      .insert()
      .into(Sale)
      .values({
        id: input.id,
        deviceId: input.deviceId,
        cashierId: input.cashierId,
        occurredAt: new Date(input.occurredAt),
        totalCents: input.totalCents,
        paymentMethod: input.paymentMethod,
        paymentState: input.paymentState,
      })
      .orIgnore() // ON CONFLICT (id) DO NOTHING
      .returning("id")
      .execute();

    // No row returned → we already had this sale. Its lines and movements were
    // written by the original request, in this same transaction shape. Writing
    // them again would double-count the stock.
    const isNew = result.raw.length > 0;
    if (!isNew) return;

    await manager.insert(SaleLine, input.lines.map((l) => ({ ... })));

    // One negative movement per line. Same transaction: a sale whose stock
    // never moved is worse than no sale at all.
    await manager.insert(
      StockMovement,
      input.lines.map((l) => ({
        id: newId(),
        productId: l.productId,
        quantity: -l.quantity,
        reason: "sale",
        saleId: input.id,
        occurredAt: new Date(input.occurredAt),
      })),
    );
  });
}

/**
 * Negative stock is not a bug to be prevented — it is the accurate record of
 * something that physically happened while two tills could not see each other.
 *
 * The system's job is to notice and tell a human, not to hide it by clamping
 * to zero. Clamping destroys the only evidence that the shop has an
 * unfulfillable order.
 */
export async function detectOversells(productIds: string[]) {
  const rows = await dataSource.query(
    `SELECT product_id, SUM(quantity) AS on_hand
       FROM stock_movements
      WHERE product_id = ANY($1)
      GROUP BY product_id
     HAVING SUM(quantity) < 0`,
    [productIds],
  );

  for (const row of rows) {
    await raiseAlert({
      type: "oversell",
      productId: row.product_id,
      shortBy: Math.abs(Number(row.on_hand)),
      // Whoever is standing in the shop is the only one who can resolve this.
      message: `Sold ${Math.abs(row.on_hand)} more than were in stock. Refund, reorder, or correct the count.`,
    });
  }

  return rows;
}

/**
 * CONCURRENTLY so tills can keep reading during the refresh; it requires the
 * unique index created in part 2.
 *
 * Debounced because a batch of fifty sales should trigger one refresh, not
 * fifty — and a shop coming back online after a day sends many batches.
 */
let pending: NodeJS.Timeout | null = null;

export function refreshStockView(): void {
  if (pending) clearTimeout(pending);
  pending = setTimeout(async () => {
    pending = null;
    await dataSource.query(
      "REFRESH MATERIALIZED VIEW CONCURRENTLY product_stock",
    );
  }, 2_000);
}

const MAX_DRIFT_MS = 6 * 60 * 60 * 1000; // 6 hours

/**
 * A till's clock is the only source for when a sale happened, so it cannot
 * simply be replaced with server time — that would attribute an offline day's
 * trade to the moment the wifi returned.
 *
 * But an obviously wrong timestamp poisons every report. Keep the claim, flag
 * it, and let a human decide.
 */
export function assessTimestamp(occurredAt: Date, receivedAt: Date) {
  const drift = receivedAt.getTime() - occurredAt.getTime();

  if (drift < -MAX_DRIFT_MS) {
    return { suspect: true, reason: "sale timestamped in the future" };
  }
  if (drift > 30 * 24 * 60 * 60 * 1000) {
    return { suspect: true, reason: "sale over 30 days old" };
  }
  return { suspect: false };
}