POSTGRESQL REFERENCE · WORKING SOURCE
src/custody.ts
This file ships in the runnable source archive. Synthetic fixtures and demonstration policies are identified in the README.
Download the complete project ↓import type {
Actor,
PalletResponse,
ReceiveRequest,
ReceiveResponse,
ReleaseRequest,
ReleaseResponse,
Row,
} from "./contracts.ts";
import { audit, type DB, pool, warehouseTx } from "./db.ts";
import { operator, palletAccess } from "./access.ts";
import { assert, fail, instant, integer, text, uid } from "./errors.ts";
import { compatible, freePositions, loadCapacity } from "./capacity.ts";
import { completeIfFinished, reservationSummary, withReservation } from "./booking.ts";
import { handlingCharge, rate, storageCharges } from "./billing.ts";
export async function palletDTO(db: DB, id: string): Promise<PalletResponse> {
const p = (await db.query(
"SELECT p.*,c.position_id,c.received_at,c.released_at,o.name AS customer_name,pos.label AS position_label FROM pallets p JOIN custodies c ON c.pallet_id=p.id JOIN organizations o ON o.id=p.owner_org_id LEFT JOIN positions pos ON pos.id=c.position_id WHERE p.id=$1",
[id],
)).rows[0];
return {
id: p.id,
label: p.label,
externalId: p.external_id,
externalScheme: p.external_scheme,
reservationId: p.reservation_id,
ownerOrgId: p.owner_org_id,
positionId: p.position_id,
positionLabel: p.position_label,
receivedAt: p.received_at,
releasedAt: p.released_at,
customerName: p.customer_name,
declaration: p.declaration,
};
}
export async function movement(
db: DB,
actor: Actor,
r: Row,
kind: string,
clock: string,
reason: string,
p: { id?: string; custodyId?: string; from?: string; to?: string; adjustmentOf?: string; data?: unknown } =
{},
) {
const id = uid("movement");
await db.query(
"INSERT INTO movements(id,kind,reservation_id,pallet_id,custody_id,owner_org_id,actor_id,warehouse_id,from_position_id,to_position_id,occurred_at,reason,adjustment_of,data) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)",
[
id,
kind,
r.id,
p.id ?? null,
p.custodyId ?? null,
r.customer_org_id,
actor.id,
r.warehouse_id,
p.from ?? null,
p.to ?? null,
clock,
reason,
p.adjustmentOf ?? null,
JSON.stringify(p.data ?? {}),
],
);
return id;
}
export async function receive(
actor: Actor,
id: string,
payload: ReceiveRequest,
key: string,
): Promise<ReceiveResponse> {
operator(actor);
assert(
Array.isArray(payload.pallets) && payload.pallets.length > 0 && payload.pallets.length <= 100,
"VALIDATION",
"Enter between 1 and 100 arriving pallets.",
);
assert(
payload.pallets.every((p) => p !== null && typeof p === "object" && !Array.isArray(p)),
"VALIDATION",
"Every pallet entry must be an object.",
);
assert(
payload.approveVariance === undefined || typeof payload.approveVariance === "boolean",
"VALIDATION",
"approveVariance must be a boolean.",
);
const entries = payload.pallets.map((p) => ({
label: text(p.label, "pallet label", 100),
externalId: p.externalId !== undefined && p.externalId !== ""
? text(p.externalId, "external identifier", 100)
: null,
actual: Object.fromEntries(
["lengthMm", "widthMm", "heightMm", "weightKg", "condition", "goods"].filter((k) => k in p).map(
(k) => [k, (p as Row)[k]],
),
),
}));
assert(
new Set(entries.map((p) => p.label)).size === entries.length,
"DUPLICATE_PALLET",
"Pallet labels in a receipt must be distinct.",
409,
);
return withReservation(actor, id, async (db, r, clock) => {
assert(
r.status === "confirmed" && r.receiving_status === "open",
"INVALID_TRANSITION",
"Receiving is not open for this reservation.",
409,
);
const reason = payload.reason
? text(payload.reason, "reason", 500)
: "Accepted against confirmed reservation";
let end = r.end_at;
const early = clock < r.start_at, late = clock >= r.end_at;
if (early) {
assert(
payload.approveVariance === true && payload.reason,
"EARLY_RECEIPT",
"Early arrival requires explicit approval, a reason, and protected capacity validation.",
409,
);
}
if (late) {
assert(
payload.approveVariance === true && payload.reason && payload.expectedEndAt,
"LATE_RECEIPT",
"This interval ended. Approve a variance and specify a future expectedEndAt.",
409,
);
end = instant(payload.expectedEndAt, "expectedEndAt", true);
assert(
end > clock && Date.parse(end) - Date.parse(clock) <= 90 * 86400000,
"INVALID_INTERVAL",
"The revised expected departure must be in the next 90 days.",
);
}
const pending = (await db.query(
"SELECT * FROM allocations WHERE reservation_id=$1 AND state='committed' ORDER BY position_id,id",
[id],
)).rows;
const excess = Math.max(0, entries.length - pending.length);
if (excess) {
assert(
payload.approveVariance === true && payload.reason,
"EXCESS_DELIVERY",
"Delivery exceeds the unreceived commitment. Explicitly approve a capacity-checked variance and give a reason.",
409,
);
}
for (const p of entries) {
const declaration = { ...r.declaration, ...p.actual };
await compatible(db, r.warehouse_id, r.zone_id, declaration);
if (
(await db.query("SELECT 1 FROM pallets WHERE reservation_id=$1 AND label=$2", [id, p.label])).rowCount
) fail(409, "DUPLICATE_PALLET", "This pallet label has already been received for this reservation.");
if (
p.externalId &&
(await db.query("SELECT 1 FROM pallets WHERE owner_org_id=$1 AND external_id=$2", [
r.customer_org_id,
p.externalId,
])).rowCount
) {
fail(
409,
"DUPLICATE_IDENTIFIER",
"This external identifier is already linked to a pallet owned by this customer.",
);
}
}
const state = await loadCapacity(db, r.warehouse_id, clock);
// Exclude only the unfulfilled slots being converted. Existing custody remains
// demand, so a receipt cannot reuse a position with goods still on it.
const excluded = pending.map((a) => a.id);
const available = freePositions(state, r.zone_id, clock, end, { allocationIds: excluded });
const currentOwned = state.pallets.filter((p) => p.reservation_id === id && !p.released_at);
if (late) {
const extensionFree = freePositions(state, r.zone_id, clock, end, { reservationId: id });
if (currentOwned.some((p) => !extensionFree.some((f) => f.id === p.position_id))) {
fail(
409,
"CAPACITY_CONFLICT",
"The late receipt would extend existing custody into a future commitment or block. Record and resolve the overstay first.",
);
}
}
// Every untouched pending commitment must also remain feasible if extending.
const needed = Math.max(entries.length, pending.length);
if (available.length < needed) {
fail(
409,
"CAPACITY_CONFLICT",
"The arrival cannot fit through the expected departure without breaking another commitment.",
{ available: available.length, requested: needed },
);
}
const preferred = [
...available.filter((p) => pending.some((a) => a.position_id === p.id)),
...available.filter((p) => !pending.some((a) => a.position_id === p.id)),
];
const allocations = [...pending];
for (let i = pending.length; i < entries.length; i++) {
const a = { id: uid("alloc"), position_id: preferred[i].id };
await db.query("INSERT INTO allocations VALUES($1,$2,$3,$4,$5,$6,'committed')", [
a.id,
r.hold_id,
id,
a.position_id,
clock,
end,
]);
allocations.push(a);
}
for (let i = 0; i < pending.length; i++) {
await db.query(
"UPDATE allocations SET position_id=$1,start_at=LEAST(start_at,$2),end_at=$3 WHERE id=$4",
[preferred[i].id, clock, end, pending[i].id],
);
}
if (early || late || excess) {
await db.query(
"UPDATE reservations SET start_at=LEAST(start_at,$1),end_at=$2,quantity=quantity+$3 WHERE id=$4",
[clock, end, excess, id],
);
if (late) {
await db.query("UPDATE allocations SET end_at=$1 WHERE reservation_id=$2 AND state='occupied'", [
end,
id,
]);
}
await audit(db, actor, clock, "receiving.variance_approved", {
warehouseId: r.warehouse_id,
customerOrgId: r.customer_org_id,
reservationId: id,
reason,
data: { early, late, excess, expectedEndAt: end },
});
}
const tariff = await rate(db, r.rate_card_id), pallets = [], movementIds = [];
for (let i = 0; i < entries.length; i++) {
const entry = entries[i],
a = allocations[i],
palletId = uid("DEMO-PAL"),
custodyId = uid("custody"),
positionId = preferred[i].id;
const declaration = { ...r.declaration, ...entry.actual };
await db.query(
"INSERT INTO pallets(id,label,external_id,external_scheme,owner_org_id,reservation_id,allocation_id,declaration) VALUES($1,$2,$3,$4,$5,$6,$7,$8)",
[
palletId,
entry.label,
entry.externalId,
entry.externalId ? "external_unvalidated" : null,
r.customer_org_id,
id,
a.id,
JSON.stringify(declaration),
],
);
await db.query("INSERT INTO custodies(id,pallet_id,position_id,received_at) VALUES($1,$2,$3,$4)", [
custodyId,
palletId,
positionId,
clock,
]);
await db.query("UPDATE allocations SET state='occupied',position_id=$1 WHERE id=$2", [
positionId,
a.id,
]);
const m = await movement(db, actor, r, "received", clock, reason, {
id: palletId,
custodyId,
to: positionId,
data: { declaration },
});
movementIds.push(m);
await handlingCharge(db, r, palletId, custodyId, "receiving", m, tariff);
pallets.push(await palletDTO(db, palletId));
}
await audit(db, actor, clock, "custody.received", {
warehouseId: r.warehouse_id,
customerOrgId: r.customer_org_id,
reservationId: id,
reason,
data: { palletIds: pallets.map((p) => p.id), movementIds },
});
return {
reservation: await reservationSummary(db, id),
pallets,
movementIds,
variance: { early, late, excess },
};
}, { operation: `receive:${id}`, key, payload });
}
function ids(value: unknown): string[] {
assert(
Array.isArray(value) && value.length > 0 && value.length <= 100 &&
value.every((v) => typeof v === "string"),
"VALIDATION",
"Select between 1 and 100 pallets.",
);
assert(new Set(value).size === value.length, "VALIDATION", "Each pallet may appear only once.");
return value;
}
export async function requestRelease(actor: Actor, id: string, payload: Row) {
const palletIds = ids(payload.palletIds), reason = text(payload.reason, "reason", 500);
return withReservation(actor, id, async (db, r, clock) => {
for (const pId of palletIds) {
const p = await palletAccess(db, actor, pId);
assert(
p.reservation_id === id && !p.released_at,
"INVALID_RELEASE",
"Choose pallets still in this reservation's custody.",
409,
);
}
const requestId = uid("release-request");
await db.query(
"INSERT INTO release_requests(id,reservation_id,customer_org_id,pallet_ids,reason,requested_by) VALUES($1,$2,$3,$4,$5,$6)",
[requestId, id, r.customer_org_id, palletIds, reason, actor.id],
);
await audit(db, actor, clock, "release.requested", {
warehouseId: r.warehouse_id,
customerOrgId: r.customer_org_id,
reservationId: id,
reason,
data: { releaseRequestId: requestId, palletIds },
});
return { id: requestId, reservationId: id, palletIds, reason, status: "requested" };
});
}
export async function release(
actor: Actor,
id: string,
payload: ReleaseRequest,
key: string,
): Promise<ReleaseResponse> {
operator(actor);
const palletIds = ids(payload.palletIds), reason = text(payload.reason, "reason", 500);
return withReservation(actor, id, async (db, r, clock) => {
const pallets = [];
for (const pId of palletIds) {
const p = await palletAccess(db, actor, pId);
assert(
p.reservation_id === id && !p.released_at,
"INVALID_RELEASE",
"A selected pallet is not in this reservation's active custody.",
409,
);
assert(p.received_at <= clock, "INVALID_RELEASE", "A release cannot precede its receipt.", 409);
pallets.push(p);
}
if (payload.releaseRequestId) {
const request: Row | undefined =
(await db.query("SELECT * FROM release_requests WHERE id=$1 AND reservation_id=$2", [
payload.releaseRequestId,
id,
])).rows[0];
assert(
request && request.status === "requested",
"INVALID_RELEASE_REQUEST",
"The release request is unavailable.",
409,
);
assert(
palletIds.every((p) => request.pallet_ids.includes(p)),
"INVALID_RELEASE_REQUEST",
"The release request does not match these pallets.",
409,
);
}
const tariff = await rate(db, r.rate_card_id), movementIds = [];
for (const p of pallets) {
const m = await movement(db, actor, r, "released", clock, reason, {
id: p.id,
custodyId: p.custody_id,
from: p.position_id,
});
movementIds.push(m);
await db.query("UPDATE custodies SET position_id=NULL,released_at=$1 WHERE id=$2", [
clock,
p.custody_id,
]);
await db.query("UPDATE allocations SET state='released' WHERE id=$1", [p.allocation_id]);
await handlingCharge(db, r, p.id, p.custody_id, "release", m, tariff);
await storageCharges(db, r, { ...p, released_at: clock }, m, tariff);
}
// Complete requests only when every requested pallet has physically departed.
await db.query(
`UPDATE release_requests rr SET status='completed' WHERE rr.reservation_id=$1 AND rr.status='requested' AND NOT EXISTS(SELECT 1 FROM custodies c WHERE c.pallet_id=ANY(rr.pallet_ids) AND c.released_at IS NULL)`,
[id],
);
await completeIfFinished(db, id);
await audit(db, actor, clock, "custody.released", {
warehouseId: r.warehouse_id,
customerOrgId: r.customer_org_id,
reservationId: id,
reason,
data: { palletIds, movementIds },
});
return {
reservation: await reservationSummary(db, id),
released: await Promise.all(palletIds.map((p) => palletDTO(db, p))),
movementIds,
};
}, { operation: `release:${id}`, key, payload });
}
export async function relocate(actor: Actor, palletId: string, payload: Row) {
operator(actor);
const positionId = text(payload.positionId, "positionId"), reason = text(payload.reason, "reason", 500);
const lookup = await pool.connect();
let p;
try {
p = await palletAccess(lookup, actor, palletId);
} finally {
lookup.release();
}
return withReservation(actor, p.reservation_id, async (db, r, clock) => {
p = await palletAccess(db, actor, palletId);
assert(!p.released_at, "INVALID_TRANSITION", "Released pallets cannot be relocated.", 409);
assert(p.position_id !== positionId, "INVALID_TRANSITION", "Select a different position.", 409);
const state = await loadCapacity(db, r.warehouse_id, clock),
dest = state.positions.find((x) => x.id === positionId);
assert(
dest && dest.zone_id === r.zone_id,
"INCOMPATIBLE_POSITION",
"Relocation in this release is limited to the reservation's compatible rack zone.",
409,
);
await compatible(db, r.warehouse_id, dest.zone_id, p.declaration);
const end = r.end_at > clock ? r.end_at : "9999-12-31T23:59:59.999Z";
assert(
freePositions(state, dest.zone_id, clock, end, { palletIds: [palletId] }).some((x) =>
x.id === positionId
),
"CAPACITY_CONFLICT",
"The destination is occupied, blocked, or committed before the expected departure.",
409,
);
await db.query("UPDATE custodies SET position_id=$1 WHERE id=$2", [positionId, p.custody_id]);
await db.query("UPDATE allocations SET position_id=$1 WHERE id=$2", [positionId, p.allocation_id]);
const m = await movement(db, actor, r, "relocated", clock, reason, {
id: palletId,
custodyId: p.custody_id,
from: p.position_id,
to: positionId,
});
await audit(db, actor, clock, "custody.relocated", {
warehouseId: r.warehouse_id,
customerOrgId: r.customer_org_id,
reservationId: r.id,
reason,
data: { movementId: m },
});
return palletDTO(db, palletId);
});
}
export async function adjustMovement(actor: Actor, movementId: string, payload: Row, key: string) {
operator(actor);
const reason = text(payload.reason, "reason", 500),
note = text(payload.note, "note", 2000),
amount = payload.amountCents === undefined
? 0
: integer(payload.amountCents, "amountCents", -1000000, 1000000);
const lookup = await pool.connect();
let m;
try {
m = (await lookup.query(
"SELECT m.* FROM movements m JOIN warehouses w ON w.id=m.warehouse_id WHERE m.id=$1 AND w.operator_org_id=$2",
[movementId, actor.orgId],
)).rows[0];
assert(m, "NOT_FOUND", "Movement not found.", 404);
} finally {
lookup.release();
}
return withReservation(actor, m.reservation_id, async (db, r, clock) => {
const id = await movement(db, actor, r, "adjustment", clock, reason, {
id: m.pallet_id,
custodyId: m.custody_id,
adjustmentOf: movementId,
data: { note, amountCents: amount },
});
if (amount !== 0) {
await db.query(
"INSERT INTO charge_lines(id,reservation_id,pallet_id,custody_id,kind,quantity,unit_cents,total_cents,rate_card_id,movement_ids) VALUES($1,$2,$3,$4,'adjustment',1,$5,$5,$6,$7)",
[uid("charge"), r.id, m.pallet_id, m.custody_id, amount, r.rate_card_id, [movementId, id]],
);
}
await audit(db, actor, clock, "movement.adjusted", {
warehouseId: r.warehouse_id,
customerOrgId: r.customer_org_id,
reservationId: r.id,
reason,
data: { movementId: id, adjustmentOf: movementId, note, amountCents: amount },
});
return { id, adjustmentOf: movementId, amountCents: amount, note };
}, { operation: `adjust:${movementId}`, key, payload });
}