Add a stock ledger, inter-branch transfers and a sync protocol that cannot jam

Stock (PRD 5.2): a branch's stock of a product is now the sum of its
StockMovement rows. Every movement names the document that caused it and
(type, refId) is unique, so replaying a sale, retrying a transfer or a
second M-Pesa callback never moves the same stock twice.

- Cash sales and confirmed M-Pesa payments deduct stock; purchases of a
  catalogue product add it.
- POST /api/stock/transfers moves stock between two branches (owner: any
  two; manager: out of their own) and refuses to send more than the
  sending branch holds. Double submits are applied once.
- POST /api/stock/adjustments records opening stock, breakage or a
  physical count (the ledger is corrected to the counted number).
- GET /api/stock returns every branch x product with LOW/OUT flags
  (per-product lowStockThreshold, default 10); GET /api/stock/movements
  is the history.

Sync (PRD 5.3/5.5): POST /api/transactions/cash now returns a result per
sale - saved, duplicate or rejected - instead of silently skipping
replays. The old client never learned a replayed sale had synced, so
rows piled up until the batch passed the 100-sale limit and every flush
failed with 400. Also:
- A sale can hold several products (one Transaction per line, grouped by
  saleId); the old single-product payload is still accepted.
- One bad sale is rejected on its own instead of failing the batch.
- A queued sale is only accepted from the cashier who recorded it, and
  stays with the branch where it happened.
- A product deactivated while offline is still recorded at its price; an
  invalid phone is dropped rather than losing the sale; the phone is now
  optional for cash sales.
- The price the device showed is kept and a difference is flagged
  (priceMismatch) instead of being silently overwritten.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/lib/server/models/StockMovement.ts

@@ -0,0 +1,51 @@
+import { Schema, model, models, Model, Types, Document } from "mongoose";
+
+export const MOVEMENT_TYPES = ["SALE", "PURCHASE", "TRANSFER_OUT", "TRANSFER_IN", "ADJUSTMENT"] as const;
+export type MovementType = (typeof MOVEMENT_TYPES)[number];
+
+export interface IStockMovement extends Document {
+  _id: Types.ObjectId;
+  businessId: Types.ObjectId;
+  branchId: Types.ObjectId;
+  productId: Types.ObjectId;
+  type: MovementType;
+  quantity: number; // signed: + adds stock to the branch, - removes it
+  refId: string; // the document that caused it (transaction, purchase, transfer, adjustment)
+  note: string | null;
+  recordedBy: Types.ObjectId | null;
+  createdAt: Date; // authoritative — server clock
+}
+
+/**
+ * The stock ledger. A branch's stock of a product is the SUM of its
+ * movements — there is no separately stored balance that could drift out
+ * of step with history.
+ *
+ * Every movement names the document that caused it, and (type, refId) is
+ * unique. Writing a movement is therefore idempotent: replaying an offline
+ * sale, retrying a transfer, or a second M-Pesa callback can never deduct
+ * or add the same stock twice. A transfer writes two movements with the
+ * same refId but different types (TRANSFER_OUT, TRANSFER_IN).
+ */
+const stockMovementSchema = new Schema<IStockMovement>(
+  {
+    businessId: { type: Schema.Types.ObjectId, ref: "Business", required: true },
+    branchId: { type: Schema.Types.ObjectId, ref: "Branch", required: true },
+    productId: { type: Schema.Types.ObjectId, ref: "Product", required: true },
+    type: { type: String, enum: MOVEMENT_TYPES, required: true },
+    quantity: { type: Number, required: true },
+    refId: { type: String, required: true },
+    note: { type: String, default: null, trim: true },
+    recordedBy: { type: Schema.Types.ObjectId, ref: "User", default: null },
+  },
+  { timestamps: { createdAt: true, updatedAt: false } }
+);
+
+stockMovementSchema.index({ type: 1, refId: 1 }, { unique: true });
+// Stock-level aggregation and the per-branch movement history.
+stockMovementSchema.index({ businessId: 1, branchId: 1, productId: 1 });
+stockMovementSchema.index({ businessId: 1, branchId: 1, createdAt: -1 });
+
+export const StockMovement =
+  (models.StockMovement as Model<IStockMovement>) ||
+  model<IStockMovement>("StockMovement", stockMovementSchema);

src/lib/server/models/StockTransfer.ts

@@ -0,0 +1,40 @@
+import { Schema, model, models, Model, Types, Document } from "mongoose";
+
+export interface IStockTransfer extends Document {
+  _id: Types.ObjectId;
+  businessId: Types.ObjectId;
+  fromBranchId: Types.ObjectId;
+  toBranchId: Types.ObjectId;
+  productId: Types.ObjectId;
+  quantity: number;
+  clientUuid: string; // idempotency key from the form, so a double-submit moves stock once
+  note: string | null;
+  recordedBy: Types.ObjectId;
+  createdAt: Date;
+}
+
+/**
+ * An inter-branch stock transfer. The stock itself moves through two
+ * StockMovement rows (TRANSFER_OUT at the source, TRANSFER_IN at the
+ * destination) whose refId is this transfer's id.
+ */
+const stockTransferSchema = new Schema<IStockTransfer>(
+  {
+    businessId: { type: Schema.Types.ObjectId, ref: "Business", required: true },
+    fromBranchId: { type: Schema.Types.ObjectId, ref: "Branch", required: true },
+    toBranchId: { type: Schema.Types.ObjectId, ref: "Branch", required: true },
+    productId: { type: Schema.Types.ObjectId, ref: "Product", required: true },
+    quantity: { type: Number, required: true, min: 1 },
+    clientUuid: { type: String, required: true },
+    note: { type: String, default: null, trim: true },
+    recordedBy: { type: Schema.Types.ObjectId, ref: "User", required: true },
+  },
+  { timestamps: { createdAt: true, updatedAt: false } }
+);
+
+stockTransferSchema.index({ businessId: 1, clientUuid: 1 }, { unique: true });
+stockTransferSchema.index({ businessId: 1, createdAt: -1 });
+
+export const StockTransfer =
+  (models.StockTransfer as Model<IStockTransfer>) ||
+  model<IStockTransfer>("StockTransfer", stockTransferSchema);

src/lib/server/models/Transaction.ts

@@ -9,7 +9,7 @@ export interface ITransaction extends Document {
   branchId: Types.ObjectId;
   staffId: Types.ObjectId;
   customerName: string | null;
-  customerPhone: string;
+  customerPhone: string | null; // optional for cash sales; required for M-Pesa
   productId: Types.ObjectId;
   sizeLiters: number; // snapshotted at sale time
   quantity: number;
@@ -26,8 +26,15 @@ export interface ITransaction extends Document {
     resultDesc: string | null;
   };
 
-  clientUuid: string; // idempotency key generated on-device
+  clientUuid: string; // idempotency key for this line: `${saleId}:${lineIndex}` for cash sales
+  saleId: string | null; // groups the lines of one multi-item sale (the device's sale UUID)
+  lineIndex: number;
   syncedFromOffline: boolean;
+  // What the device showed and collected per unit. The server price stays
+  // authoritative; a difference (e.g. the price changed while the device
+  // was offline) is kept and flagged rather than silently overwritten.
+  clientUnitPrice: number | null;
+  priceMismatch: boolean;
 
   // TAMPER PREVENTION: this is what the device *claimed* the time was when
   // the sale happened — informational only, e.g. for offline-queue
@@ -50,7 +57,7 @@ const transactionSchema = new Schema<ITransaction>(
     branchId: { type: Schema.Types.ObjectId, ref: "Branch", required: true },
     staffId: { type: Schema.Types.ObjectId, ref: "User", required: true },
     customerName: { type: String, default: null, trim: true },
-    customerPhone: { type: String, required: true },
+    customerPhone: { type: String, default: null },
     productId: { type: Schema.Types.ObjectId, ref: "Product", required: true },
     sizeLiters: { type: Number, required: true },
     quantity: { type: Number, required: true, min: 1 },
@@ -73,7 +80,11 @@ const transactionSchema = new Schema<ITransaction>(
     },
 
     clientUuid: { type: String, required: true },
+    saleId: { type: String, default: null },
+    lineIndex: { type: Number, default: 0 },
     syncedFromOffline: { type: Boolean, default: false },
+    clientUnitPrice: { type: Number, default: null },
+    priceMismatch: { type: Boolean, default: false },
     clientSubmittedAt: { type: Date, default: null },
   },
   { timestamps: true } // createdAt/updatedAt set from the server's Date.now(), not client input
@@ -92,6 +103,7 @@ transactionSchema.index({ clientUuid: 1 }, { unique: true });
 transactionSchema.index({ branchId: 1, createdAt: -1 });
 transactionSchema.index({ businessId: 1, createdAt: -1 });
 transactionSchema.index({ paymentStatus: 1 });
+transactionSchema.index({ businessId: 1, saleId: 1 });
 
 export const Transaction =
   (models.Transaction as Model<ITransaction>) ||

src/lib/server/stock.ts

@@ -0,0 +1,140 @@
+import mongoose, { Types } from "mongoose";
+import { StockMovement, type MovementType } from "@/lib/server/models/StockMovement";
+import { Branch } from "@/lib/server/models/Branch";
+import { Product } from "@/lib/server/models/Product";
+import type { ITransaction } from "@/lib/server/models/Transaction";
+
+type Id = Types.ObjectId | string;
+
+export interface MovementInput {
+  businessId: Id;
+  branchId: Id;
+  productId: Id;
+  type: MovementType;
+  quantity: number; // signed
+  refId: string;
+  note?: string | null;
+  recordedBy?: Id | null;
+}
+
+/**
+ * Write one ledger movement, idempotently: if a movement with the same
+ * (type, refId) already exists nothing changes. Safe to call again on any
+ * retry or replay.
+ */
+export async function recordMovement(m: MovementInput): Promise<void> {
+  await StockMovement.updateOne(
+    { type: m.type, refId: m.refId },
+    {
+      $setOnInsert: {
+        businessId: m.businessId,
+        branchId: m.branchId,
+        productId: m.productId,
+        quantity: m.quantity,
+        note: m.note ?? null,
+        recordedBy: m.recordedBy ?? null,
+      },
+    },
+    { upsert: true }
+  );
+}
+
+/**
+ * Deduct a completed sale from its branch's stock. Only SUCCESS sales move
+ * stock — a PENDING or FAILED M-Pesa prompt never does.
+ */
+export async function recordSaleMovement(
+  txn: Pick<ITransaction, "_id" | "businessId" | "branchId" | "productId" | "quantity" | "paymentStatus" | "staffId">
+): Promise<void> {
+  if (txn.paymentStatus !== "SUCCESS") return;
+  await recordMovement({
+    businessId: txn.businessId,
+    branchId: txn.branchId,
+    productId: txn.productId,
+    type: "SALE",
+    quantity: -Math.abs(txn.quantity),
+    refId: txn._id.toString(),
+    recordedBy: txn.staffId,
+  });
+}
+
+/** Ledger refIds derived from client-generated keys are namespaced by tenant. */
+export function clientRefId(businessId: Id, clientUuid: string): string {
+  return `${businessId.toString()}:${clientUuid}`;
+}
+
+/** Current stock of one product at one branch (the sum of its movements). */
+export async function stockOf(businessId: Id, branchId: Id, productId: Id): Promise<number> {
+  const [row] = await StockMovement.aggregate<{ total: number }>([
+    {
+      $match: {
+        businessId: new mongoose.Types.ObjectId(businessId.toString()),
+        branchId: new mongoose.Types.ObjectId(branchId.toString()),
+        productId: new mongoose.Types.ObjectId(productId.toString()),
+      },
+    },
+    { $group: { _id: null, total: { $sum: "$quantity" } } },
+  ]);
+  return row?.total ?? 0;
+}
+
+export type StockStatus = "OK" | "LOW" | "OUT";
+
+export interface StockLevel {
+  branchId: string;
+  branchName: string;
+  productId: string;
+  productLabel: string;
+  unit: string;
+  quantity: number;
+  lowStockThreshold: number;
+  status: StockStatus;
+}
+
+export function stockStatus(quantity: number, threshold: number): StockStatus {
+  if (quantity <= 0) return "OUT";
+  if (quantity <= threshold) return "LOW";
+  return "OK";
+}
+
+/**
+ * Stock of every active product at every branch in scope — including
+ * zero rows, so an owner sees "0" rather than a missing product. A
+ * negative quantity is reported as-is: it means sales were recorded
+ * (typically offline) for stock the ledger never received, and the branch
+ * needs a count/adjustment.
+ */
+export async function getStockLevels(businessId: Id, branchId?: Id | null): Promise<StockLevel[]> {
+  const bizObj = new mongoose.Types.ObjectId(businessId.toString());
+  const branchFilter: Record<string, unknown> = { businessId: bizObj };
+  if (branchId) branchFilter._id = new mongoose.Types.ObjectId(branchId.toString());
+
+  const [branches, products, totals] = await Promise.all([
+    Branch.find(branchFilter).sort({ branchName: 1 }).lean(),
+    Product.find({ businessId: bizObj, isActive: true }).sort({ sortOrder: 1 }).lean(),
+    StockMovement.aggregate<{ _id: { branchId: Types.ObjectId; productId: Types.ObjectId }; total: number }>([
+      { $match: branchId ? { businessId: bizObj, branchId: branchFilter._id } : { businessId: bizObj } },
+      { $group: { _id: { branchId: "$branchId", productId: "$productId" }, total: { $sum: "$quantity" } } },
+    ]),
+  ]);
+
+  const totalByKey = new Map(totals.map((t) => [`${t._id.branchId}:${t._id.productId}`, t.total]));
+  const levels: StockLevel[] = [];
+  for (const b of branches) {
+    for (const p of products) {
+      const quantity = totalByKey.get(`${b._id}:${p._id}`) ?? 0;
+      const lowStockThreshold = p.lowStockThreshold ?? 10;
+      levels.push({
+        branchId: b._id.toString(),
+        branchName: b.branchName,
+        productId: p._id.toString(),
+        productLabel: p.label,
+        unit: p.unit ?? "L",
+        quantity,
+        lowStockThreshold,
+        status: stockStatus(quantity, lowStockThreshold),
+      });
+    }
+  }
+  return levels;
+}

src/lib/syncProtocol.ts

@@ -0,0 +1,11 @@
+// Shared between the terminal's offline queue and POST /api/transactions/cash.
+
+/** Largest batch the offline queue sends in one request. */
+export const MAX_SALES_PER_REQUEST = 50;
+
+/** Largest number of product lines in one sale. */
+export const MAX_LINES_PER_SALE = 20;
+
+export type SaleResult =
+  | { clientUuid: string; status: "saved" | "duplicate"; amountTotal: number; priceMismatch: boolean }
+  | { clientUuid: string; status: "rejected"; error: string };