diff --git a/docs/PILOT_READINESS.md b/docs/PILOT_READINESS.md index f86b090..66320d0 100644 --- a/docs/PILOT_READINESS.md +++ b/docs/PILOT_READINESS.md @@ -62,7 +62,7 @@ the decisions taken along the way, and what is still open. | Decision | Why | |---|---| | Offline sales are always accepted, even if they take stock below zero. | The sale already happened. Refusing it would lose real cash. Negative stock is shown on the Stock page and fixed with a physical count. | -| Transfers are refused when the sending branch lacks the stock. | Transfers are made online by a manager, so they can be checked. | +| Transfers are refused when the sending branch lacks the stock. | Transfers are made online by a manager, so they can be checked. The check and the move run under a short per-branch, per-product lock (`withStockLock` in `src/lib/server/stock.ts`), so simultaneous transfers cannot together overdraw a branch; physical counts take the same lock. | | The server price is authoritative; a different price shown on the device is kept (`clientUnitPrice`) and flagged (`priceMismatch`). | Trusting the device's price would let a cashier under-charge. Overwriting it silently would hide a cash discrepancy. | | A queued sale stays with the branch where it was recorded, even if the cashier has since moved. | That is where the stock left and the cash was taken. | | A product deactivated while a phone was offline is still recorded at its price. | Same reason as the first row: the sale happened. | @@ -76,7 +76,7 @@ the decisions taken along the way, and what is still open. sales from the old version are sent on the next sync. ## Testing -- `npm test` runs 56 API tests against an in-memory MongoDB, covering role +- `npm test` runs 62 API tests against an in-memory MongoDB, covering role enforcement, tenant isolation for sales, stock and transfers, suspension, the sync protocol (including a 120-sale backlog replayed twice), and the business day. @@ -137,8 +137,6 @@ prices read-only. - **Isolation by convention.** Tenant filters are still written per route. The tests cover the sales, stock, transfer and access paths, not every route. -- **Concurrent transfers.** Two transfers submitted at the same moment - could together send more than the branch holds. - **Cost of goods is an average.** It uses each product's average buying price across the business, not first-in-first-out per branch. Units sold with no purchase on record are counted and flagged, but have no cost. diff --git a/src/app/api/stock/adjustments/route.ts b/src/app/api/stock/adjustments/route.ts index 06f5fd8..c375d0c 100644 --- a/src/app/api/stock/adjustments/route.ts +++ b/src/app/api/stock/adjustments/route.ts @@ -6,7 +6,7 @@ import { Product } from "@/lib/server/models/Product"; import { StockMovement } from "@/lib/server/models/StockMovement"; import { authorize, MANAGERS } from "@/lib/server/auth"; import { objectId } from "@/lib/server/validation"; -import { clientRefId, recordMovement, stockOf } from "@/lib/server/stock"; +import { clientRefId, recordMovement, stockOf, StockBusyError, withStockLock } from "@/lib/server/stock"; export const runtime = "nodejs"; @@ -57,27 +57,36 @@ export async function POST(req: NextRequest) { if (!branch) return NextResponse.json({ error: "Branch not found in your business" }, { status: 404 }); if (!product) return NextResponse.json({ error: "Product not found in your business" }, { status: 404 }); - // A retried submit with the same clientUuid is a no-op. + // A retried submit with the same clientUuid is a no-op. A count reads the + // balance to work out its correction, so it runs under the stock lock: + // a transfer cannot slip in between the read and the write. const refId = clientRefId(auth.businessId, data.clientUuid); - const already = await StockMovement.exists({ type: "ADJUSTMENT", refId }); - if (!already) { - const before = await stockOf(auth.businessId, branchId, data.productId); - const delta = data.mode === "COUNT" ? data.quantity - before : data.quantity; - if (delta !== 0) { - const label = data.mode === "COUNT" ? `Count: ${before} -> ${data.quantity}` : "Adjustment"; - await recordMovement({ - businessId: auth.businessId, - branchId, - productId: data.productId, - type: "ADJUSTMENT", - quantity: delta, - refId, - note: data.note ? `${label}. ${data.note}` : label, - recordedBy: auth.sub, - }); - } + const lockedBranch = branchId; + try { + const quantity = await withStockLock(auth.businessId, lockedBranch, data.productId, async () => { + const already = await StockMovement.exists({ type: "ADJUSTMENT", refId }); + if (!already) { + const before = await stockOf(auth.businessId, lockedBranch, data.productId); + const delta = data.mode === "COUNT" ? data.quantity - before : data.quantity; + if (delta !== 0) { + const label = data.mode === "COUNT" ? `Count: ${before} -> ${data.quantity}` : "Adjustment"; + await recordMovement({ + businessId: auth.businessId, + branchId: lockedBranch, + productId: data.productId, + type: "ADJUSTMENT", + quantity: delta, + refId, + note: data.note ? `${label}. ${data.note}` : label, + recordedBy: auth.sub, + }); + } + } + return stockOf(auth.businessId, lockedBranch, data.productId); + }); + return NextResponse.json({ branchId, productId: data.productId, quantity }, { status: 201 }); + } catch (err) { + if (err instanceof StockBusyError) return NextResponse.json({ error: err.message }, { status: 409 }); + throw err; } - - const quantity = await stockOf(auth.businessId, branchId, data.productId); - return NextResponse.json({ branchId, productId: data.productId, quantity }, { status: 201 }); } diff --git a/src/app/api/stock/transfers/route.ts b/src/app/api/stock/transfers/route.ts index b8465f0..1a0f806 100644 --- a/src/app/api/stock/transfers/route.ts +++ b/src/app/api/stock/transfers/route.ts @@ -6,7 +6,7 @@ import { Product } from "@/lib/server/models/Product"; import { StockTransfer, IStockTransfer } from "@/lib/server/models/StockTransfer"; import { authorize, branchParam, MANAGERS } from "@/lib/server/auth"; import { objectId } from "@/lib/server/validation"; -import { recordMovement, stockOf } from "@/lib/server/stock"; +import { recordMovement, stockOf, StockBusyError, withStockLock } from "@/lib/server/stock"; export const runtime = "nodejs"; export const dynamic = "force-dynamic"; @@ -104,33 +104,49 @@ export async function POST(req: NextRequest) { if (!product) return NextResponse.json({ error: "Product not found in your business" }, { status: 404 }); // Transfers are made online by a manager, so unlike offline sales they - // can be refused when the sending branch does not have the stock. - const available = await stockOf(auth.businessId, fromBranchId, data.productId); - if (available < data.quantity) { - return NextResponse.json( - { error: `Only ${Math.max(available, 0)} in stock at the sending branch`, available }, - { status: 409 } - ); - } - - let transfer: IStockTransfer; + // can be refused when the sending branch does not have the stock. The + // balance check and the movements run under the sending branch's stock + // lock, so two simultaneous transfers cannot both spend the same units. + const sendingBranch = fromBranchId; try { - transfer = await StockTransfer.create({ - businessId: auth.businessId, - fromBranchId, - toBranchId: data.toBranchId, - productId: data.productId, - quantity: data.quantity, - clientUuid: data.clientUuid, - note: data.note ?? null, - recordedBy: auth.sub, - }); - } catch (err: unknown) { - // Two identical submits raced and the other one created it. - if ((err as { code?: number })?.code !== 11000) throw err; - transfer = (await StockTransfer.findOne({ businessId: auth.businessId, clientUuid: data.clientUuid }))!; - } - await applyTransfer(transfer); + return await withStockLock(auth.businessId, sendingBranch, data.productId, async () => { + // An identical submit may have completed while we waited for the lock. + const done = await StockTransfer.findOne({ businessId: auth.businessId, clientUuid: data.clientUuid }); + if (done) { + await applyTransfer(done); + return NextResponse.json({ transfer: done }, { status: 200 }); + } - return NextResponse.json({ transfer }, { status: 201 }); + const available = await stockOf(auth.businessId, sendingBranch, data.productId); + if (available < data.quantity) { + return NextResponse.json( + { error: `Only ${Math.max(available, 0)} in stock at the sending branch`, available }, + { status: 409 } + ); + } + + let transfer: IStockTransfer; + try { + transfer = await StockTransfer.create({ + businessId: auth.businessId, + fromBranchId: sendingBranch, + toBranchId: data.toBranchId, + productId: data.productId, + quantity: data.quantity, + clientUuid: data.clientUuid, + note: data.note ?? null, + recordedBy: auth.sub, + }); + } catch (err: unknown) { + // Defensive: the unique clientUuid index still guards double submits. + if ((err as { code?: number })?.code !== 11000) throw err; + transfer = (await StockTransfer.findOne({ businessId: auth.businessId, clientUuid: data.clientUuid }))!; + } + await applyTransfer(transfer); + return NextResponse.json({ transfer }, { status: 201 }); + }); + } catch (err) { + if (err instanceof StockBusyError) return NextResponse.json({ error: err.message }, { status: 409 }); + throw err; + } } diff --git a/src/lib/server/models/StockLock.ts b/src/lib/server/models/StockLock.ts new file mode 100644 index 0000000..117c042 --- /dev/null +++ b/src/lib/server/models/StockLock.ts @@ -0,0 +1,25 @@ +import { Schema, model, models, Model } from "mongoose"; + +export interface IStockLock { + _id: string; // `${businessId}:${branchId}:${productId}` + owner: string; // random id of the request holding the lock + lockedUntil: Date; // lease expiry: a crashed request cannot hold the lock forever +} + +/** + * A short lease on one product's stock at one branch, held while a request + * reads the current balance and writes a movement that depends on it (see + * withStockLock in src/lib/server/stock.ts). Rows are deleted on release; + * an expired row is simply taken over by the next request. + */ +const stockLockSchema = new Schema( + { + _id: { type: String, required: true }, + owner: { type: String, required: true }, + lockedUntil: { type: Date, required: true }, + }, + { versionKey: false } +); + +export const StockLock = + (models.StockLock as Model) || model("StockLock", stockLockSchema); diff --git a/src/lib/server/stock.ts b/src/lib/server/stock.ts index 3e4f8ec..454c108 100644 --- a/src/lib/server/stock.ts +++ b/src/lib/server/stock.ts @@ -3,6 +3,7 @@ import { StockMovement, type MovementType } from "@/lib/server/models/StockMovem 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; @@ -173,3 +174,58 @@ export async function backfillPurchaseMovements(businessId?: Id): Promise(businessId: Id, branchId: Id, productId: Id, fn: () => Promise): Promise { + 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 }); + } +} diff --git a/tests/concurrency.test.ts b/tests/concurrency.test.ts new file mode 100644 index 0000000..b14525e --- /dev/null +++ b/tests/concurrency.test.ts @@ -0,0 +1,119 @@ +import { describe, expect, it } from "vitest"; +import { stockOf } from "@/lib/server/stock"; +import { StockTransfer } from "@/lib/server/models/StockTransfer"; +import { StockLock } from "@/lib/server/models/StockLock"; +import * as adjustments from "@/app/api/stock/adjustments/route"; +import * as transfers from "@/app/api/stock/transfers/route"; +import { makeTenant, req, uuid, type Tenant } from "./helpers"; + +async function openingStock(t: Tenant, quantity: number) { + await adjustments.POST( + req("POST", "/x", { + as: t.owner, + body: { clientUuid: uuid(), branchId: t.branchA, productId: t.water20, mode: "DELTA", quantity }, + }) + ); +} + +const transferOut = (t: Tenant, quantity: number) => + transfers.POST( + req("POST", "/x", { + as: t.owner, + body: { clientUuid: uuid(), fromBranchId: t.branchA, toBranchId: t.branchB, productId: t.water20, quantity }, + }) + ); + +describe("concurrent stock changes", () => { + it("never lets simultaneous transfers send more than the branch holds", async () => { + const t = await makeTenant(); + await openingStock(t, 10); + + // Five managers press "Record transfer" for 4 units at the same moment. + const statuses = (await Promise.all(Array.from({ length: 5 }, () => transferOut(t, 4)))).map((r) => r.status); + + expect(statuses.filter((s) => s === 201)).toHaveLength(2); + expect(statuses.filter((s) => s === 409)).toHaveLength(3); + expect(await stockOf(t.businessId, t.branchA, t.water20)).toBe(2); + expect(await stockOf(t.businessId, t.branchB, t.water20)).toBe(8); + expect(await StockTransfer.countDocuments({})).toBe(2); + }); + + it("holds across repeated races, not just once", async () => { + for (let round = 0; round < 5; round++) { + const t = await makeTenant("Round" + round); + await openingStock(t, 9); + const statuses = (await Promise.all(Array.from({ length: 4 }, () => transferOut(t, 3)))).map((r) => r.status); + expect(statuses.filter((s) => s === 201)).toHaveLength(3); + expect(await stockOf(t.businessId, t.branchA, t.water20)).toBe(0); + } + }); + + it("applies a double-tapped transfer once even when both taps race", async () => { + const t = await makeTenant(); + await openingStock(t, 10); + const body = { clientUuid: uuid(), fromBranchId: t.branchA, toBranchId: t.branchB, productId: t.water20, quantity: 6 }; + const statuses = ( + await Promise.all([ + transfers.POST(req("POST", "/x", { as: t.owner, body })), + transfers.POST(req("POST", "/x", { as: t.owner, body })), + ]) + ).map((r) => r.status); + expect(statuses.sort()).toEqual([200, 201]); + expect(await stockOf(t.businessId, t.branchA, t.water20)).toBe(4); + }); + + it("takes over a lock left behind by a crashed request once it expires", async () => { + const t = await makeTenant(); + await openingStock(t, 10); + await StockLock.create({ + _id: `${t.businessId}:${t.branchA}:${t.water20}`, + owner: "crashed-request", + lockedUntil: new Date(Date.now() - 1000), + }); + expect((await transferOut(t, 3)).status).toBe(201); + expect(await StockLock.countDocuments({})).toBe(0); + }); + + it("serialises a physical count with a transfer, so the count is never applied to a stale balance", async () => { + const t = await makeTenant(); + await openingStock(t, 10); + + // A count of 10 and a transfer of 6 race. Whichever runs second must + // see the first: either the count resets 4 back to 10 after the + // transfer, or the transfer moves 6 out of the counted 10. + const [count, transfer] = await Promise.all([ + adjustments.POST( + req("POST", "/x", { + as: t.owner, + body: { clientUuid: uuid(), branchId: t.branchA, productId: t.water20, mode: "COUNT", quantity: 10 }, + }) + ), + transferOut(t, 6), + ]); + expect([count.status, transfer.status]).toEqual([201, 201]); + const a = await stockOf(t.businessId, t.branchA, t.water20); + expect([4, 10]).toContain(a); + expect(await stockOf(t.businessId, t.branchB, t.water20)).toBe(6); + }); + + it("does not block transfers of different products or branches on each other", async () => { + const t = await makeTenant(); + await openingStock(t, 10); + await adjustments.POST( + req("POST", "/x", { + as: t.owner, + body: { clientUuid: uuid(), branchId: t.branchA, productId: t.water5, mode: "DELTA", quantity: 10 }, + }) + ); + const results = await Promise.all([ + transferOut(t, 5), + transfers.POST( + req("POST", "/x", { + as: t.owner, + body: { clientUuid: uuid(), fromBranchId: t.branchA, toBranchId: t.branchB, productId: t.water5, quantity: 5 }, + }) + ), + ]); + expect(results.map((r) => r.status)).toEqual([201, 201]); + }); +});