ee5bb5beb8
A transfer read the sending branch's stock and then wrote its movements as two separate steps. Requests arriving together all saw the same balance and all passed the check: five simultaneous transfers of 4 from a stock of 10 all succeeded, sending 20. The check and the write now run under withStockLock: a short lease on one product's stock at one branch, taken with a single atomic upsert (it only succeeds when no unexpired lease exists) and released when the request finishes. A second request waits briefly, then sees the updated balance; after 5 seconds it gets 409 "try again". A lease expires after 10 seconds, so a crashed request cannot block the stock for good. Physical counts take the same lock, so a count is never computed from a balance a transfer is changing. A double-tapped transfer that races itself still moves stock once. Sales do not take the lock: they are recorded even when stock runs out, by design. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
232 lines
8.0 KiB
TypeScript
232 lines
8.0 KiB
TypeScript
import mongoose, { Types } from "mongoose";
|
|
import { StockMovement, type MovementType } from "@/lib/server/models/StockMovement";
|
|
import { ACTIVE_BRANCH, Branch } from "@/lib/server/models/Branch";
|
|
import { Product } from "@/lib/server/models/Product";
|
|
import { Purchase } from "@/lib/server/models/Purchase";
|
|
import { StockLock } from "@/lib/server/models/StockLock";
|
|
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;
|
|
productActive: boolean;
|
|
}
|
|
|
|
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 active branch in scope —
|
|
* including zero rows, so an owner sees "0" rather than a missing product.
|
|
* A deactivated product is listed too wherever it still has stock, so
|
|
* units bought before deactivation never silently disappear. 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, ...ACTIVE_BRANCH };
|
|
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 }).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;
|
|
if (!p.isActive && quantity === 0) continue;
|
|
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),
|
|
productActive: p.isActive,
|
|
});
|
|
}
|
|
}
|
|
return levels;
|
|
}
|
|
|
|
/**
|
|
* Write the PURCHASE movement for every catalogue purchase that has none.
|
|
* Purchases recorded before the stock ledger existed (or while a write
|
|
* failed) otherwise never show up as stock. Idempotent: run it as often
|
|
* as you like. Returns how many movements were added.
|
|
*/
|
|
export async function backfillPurchaseMovements(businessId?: Id): Promise<number> {
|
|
const filter: Record<string, unknown> = { productId: { $ne: null } };
|
|
if (businessId) filter.businessId = new mongoose.Types.ObjectId(businessId.toString());
|
|
|
|
let added = 0;
|
|
for await (const p of Purchase.find(filter).lean().cursor()) {
|
|
const refId = p._id.toString();
|
|
if (await StockMovement.exists({ type: "PURCHASE", refId })) continue;
|
|
await recordMovement({
|
|
businessId: p.businessId,
|
|
branchId: p.branchId,
|
|
productId: p.productId!,
|
|
type: "PURCHASE",
|
|
quantity: p.quantity,
|
|
refId,
|
|
note: `From ${p.supplierName} (backfilled)`,
|
|
recordedBy: p.recordedBy,
|
|
});
|
|
added += 1;
|
|
}
|
|
return added;
|
|
}
|
|
|
|
const LOCK_LEASE_MS = 10_000;
|
|
const LOCK_WAIT_MS = 5_000;
|
|
|
|
/** Another request kept the stock busy for longer than we are willing to wait. */
|
|
export class StockBusyError extends Error {
|
|
constructor() {
|
|
super("This stock is being updated by someone else. Please try again.");
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Run `fn` while holding the lock on one product's stock at one branch.
|
|
*
|
|
* Anything that reads the balance and then writes a movement based on it
|
|
* (a transfer refusing to overdraw, a physical count computing its
|
|
* correction) must run inside this. Without it, two transfers that each
|
|
* see 10 in stock can both send 6, because the read and the write are
|
|
* separate operations. With it, the second waits for the first and then
|
|
* sees 4.
|
|
*
|
|
* The lock is a lease row: acquired with one atomic upsert that only
|
|
* succeeds if no unexpired lease exists (otherwise the insert hits the
|
|
* unique _id and fails), released by deleting it. Sales do not take the
|
|
* lock - they are recorded even when stock runs out (see PILOT_READINESS).
|
|
*/
|
|
export async function withStockLock<T>(businessId: Id, branchId: Id, productId: Id, fn: () => Promise<T>): Promise<T> {
|
|
const key = `${businessId}:${branchId}:${productId}`;
|
|
const owner = crypto.randomUUID();
|
|
const deadline = Date.now() + LOCK_WAIT_MS;
|
|
let delay = 10;
|
|
|
|
for (;;) {
|
|
const now = Date.now();
|
|
try {
|
|
await StockLock.updateOne(
|
|
{ _id: key, lockedUntil: { $lt: new Date(now) } },
|
|
{ $set: { owner, lockedUntil: new Date(now + LOCK_LEASE_MS) } },
|
|
{ upsert: true }
|
|
);
|
|
break;
|
|
} catch (err: unknown) {
|
|
if ((err as { code?: number })?.code !== 11000) throw err;
|
|
}
|
|
if (Date.now() > deadline) throw new StockBusyError();
|
|
await new Promise((r) => setTimeout(r, delay + Math.random() * delay));
|
|
delay = Math.min(delay * 2, 200);
|
|
}
|
|
|
|
try {
|
|
return await fn();
|
|
} finally {
|
|
await StockLock.deleteOne({ _id: key, owner });
|
|
}
|
|
}
|