"""Draft -> posted receiving and draft -> transit -> received transfers.""" import uuid from contextlib import contextmanager from django.core.exceptions import ValidationError from django.db.models import Q from django.utils import timezone from apps.parts.locking import persisted_pk from apps.parts.models import SparePart from .locking import inventory_context, lock_positions from .models import ( InventoryLocation, InventoryDocumentSequence, GoodsReceipt, GoodsReceiptLine, GoodsReceiptIdentifier, SerializedStockUnit, StockTransfer, StockTransferLine, StockTransferUnit, ) from .services import check_revision, text_value, quantity_value, normalized_identifier, _register_unit, _post from .queries import USABLE_TYPES def _allocate(company, kind): counter, _ = InventoryDocumentSequence.objects.select_for_update().get_or_create(company=company, kind=kind) if counter.next_value > 999999999999999999: raise ValidationError("Document number capacity exhausted.") number = f"{'GRN' if kind == 'RECEIPT' else 'TRF'}-{counter.next_value:08d}" counter.next_value += 1 counter.save(update_fields=["next_value"]) return number def _physical(location): if not location.is_active or location.location_type in ("TRANSIT", "CUSTODY"): raise ValidationError("Select an active physical inventory location, not managed transit.") def create_goods_receipt(*, actor, destination, external_reference="", note=""): with inventory_context(actor=actor, company=destination.company, locations=[destination], permission="inventory.receive_stock") as (actor, company, locations, _): destination = locations[destination.pk] _physical(destination) row = GoodsReceipt(company=company, destination=destination, number=_allocate(company, "RECEIPT"), created_by=actor, external_reference=text_value(external_reference, maximum=128), note=text_value(note)) row._persist() return row def create_stock_transfer(*, actor, source, destination, note=""): with inventory_context(actor=actor, company=source.company, locations=[source, destination], permission="inventory.transfer_stock") as (actor, company, locations, _): source, destination = locations[source.pk], locations[destination.pk] _physical(source) _physical(destination) if source.pk == destination.pk or (source.location_type in ("QUARANTINE", "DEFECTIVE") and destination.location_type in USABLE_TYPES): raise ValidationError("Transfer endpoints cannot coincide or silently promote non-serviceable stock.") row = StockTransfer(company=company, source=source, destination=destination, number=_allocate(company, "TRANSFER"), created_by=actor, note=text_value(note)) row._persist() return row @contextmanager def _document(document, actor, expected_revision, *, parts=(), include_current_parts=False): model = GoodsReceipt if isinstance(document, GoodsReceipt) else StockTransfer pk = persisted_pk(document, model) snapshot = model.objects.select_related("company", "destination", *(["source", "transit_location"] if model is StockTransfer else [])).get(pk=pk) locations = [snapshot.destination] + ([snapshot.source] if model is StockTransfer else []) permission = "inventory.receive_stock" if model is GoodsReceipt else "inventory.transfer_stock" if include_current_parts: parts = list(SparePart.objects.filter(pk__in=snapshot.lines.filter(is_active=True).values("spare_part_id"))) part_ids = {part.pk for part in parts} with inventory_context(actor=actor, company=snapshot.company, locations=locations, parts=parts, permission=permission) as (actor, company, locations, parts): # A transfer's managed transit location is a system dependency, authorized # through BOTH real endpoints. Lock it before the document/positions. if model is StockTransfer and snapshot.transit_location_id: from apps.access.locks import share_rows share_rows(InventoryLocation, [snapshot.transit_location_id]) transit = InventoryLocation.objects.get(pk=snapshot.transit_location_id) if transit.company_id != company.pk or transit.location_type != "TRANSIT": raise ValidationError("Transfer transit identity is inconsistent.") locations[transit.pk] = transit current = model.objects.select_for_update().get(pk=pk) check_revision(current, expected_revision) if current.updated_at != snapshot.updated_at: raise ValidationError("Document changed while dependencies were locked; reload.") if include_current_parts and set(current.lines.filter(is_active=True).values_list("spare_part_id", flat=True)) != part_ids: raise ValidationError("Document lines changed; reload.") yield current, actor, company, locations, parts def _draft(document): if document.status != "DRAFT": raise ValidationError("Only draft documents can be edited or cancelled.") def update_goods_receipt(*, actor, receipt, expected_revision, external_reference="", note=""): with _document(receipt, actor, expected_revision) as (current, _, _, _, _): _draft(current) current.external_reference = text_value(external_reference, maximum=128) current.note = text_value(note) current._persist() return current def update_stock_transfer(*, actor, transfer, expected_revision, note=""): with _document(transfer, actor, expected_revision) as (current, _, _, _, _): _draft(current) current.note = text_value(note) current._persist() return current def _line_inputs(lines, *, receipt): if not isinstance(lines, (list, tuple)): raise ValidationError("Supply an explicit document line list.") result = {} for entry in lines: allowed = {"spare_part", "quantity", "identifiers" if receipt else "units"} if not isinstance(entry, dict) or set(entry) - allowed: raise ValidationError("Invalid line data.") part = entry.get("spare_part") pk = persisted_pk(part, SparePart) if pk in result: raise ValidationError("A part may appear only once per document.") quantity = quantity_value(entry.get("quantity")) values = entry.get("identifiers" if receipt else "units", []) if not isinstance(values, (list, tuple)): raise ValidationError("Supply an explicit identifier/unit list.") normalized = [normalized_identifier(value) if receipt else persisted_pk(value, SerializedStockUnit) for value in values] if len(set(normalized)) != len(normalized) or len(normalized) > quantity: raise ValidationError("Duplicate units or more identifiers than quantity.") result[pk] = (part, quantity, normalized) return result def set_goods_receipt_lines(*, actor, receipt, expected_revision, lines): inputs = _line_inputs(lines, receipt=True) with _document(receipt, actor, expected_revision, parts=[entry[0] for entry in inputs.values()]) as (current, _, _, _, _): _draft(current) existing = {line.spare_part_id: line for line in current.lines.select_for_update().order_by("pk")} for pk, line in existing.items(): if pk not in inputs and line.is_active: line.is_active = False line._persist() for pk, (part, quantity, identifiers) in sorted(inputs.items(), key=lambda item: str(item[0])): line = existing.get(pk) or GoodsReceiptLine(receipt=current, spare_part=part) line.quantity, line.is_active = quantity, True line._persist() old = {row.identifier: row for row in line.identifiers.all()} for value, row in old.items(): active = value in identifiers if row.is_active != active: row.is_active = active row._persist() for identifier in identifiers: if identifier not in old: GoodsReceiptIdentifier(line=line, identifier=identifier)._persist() current._persist() return current def set_stock_transfer_lines(*, actor, transfer, expected_revision, lines): inputs = _line_inputs(lines, receipt=False) with _document(transfer, actor, expected_revision, parts=[entry[0] for entry in inputs.values()]) as (current, _, company, _, _): _draft(current) existing = {line.spare_part_id: line for line in current.lines.select_for_update().order_by("pk")} for pk, line in existing.items(): if pk not in inputs and line.is_active: line.is_active = False line._persist() for pk, (part, quantity, unit_ids) in sorted(inputs.items(), key=lambda item: str(item[0])): if SerializedStockUnit.objects.filter(pk__in=unit_ids, company=company, spare_part=part).count() != len(unit_ids): raise ValidationError("Selected unit does not belong to this Company and part.") line = existing.get(pk) or StockTransferLine(transfer=current, spare_part=part) line.quantity, line.is_active = quantity, True line._persist() old = {row.unit_id: row for row in line.units.all()} for unit_id, row in old.items(): active = unit_id in unit_ids if row.is_active != active: row.is_active = active row._persist() for unit_id in unit_ids: if unit_id not in old: StockTransferUnit(line=line, unit_id=unit_id)._persist() current._persist() return current def post_goods_receipt(*, actor, receipt, expected_revision): with _document(receipt, actor, expected_revision, include_current_parts=True) as (current, actor, company, locations, parts): _draft(current) destination = locations[current.destination_id] _physical(destination) lines = list(current.lines.filter(is_active=True).prefetch_related("identifiers").order_by("spare_part_id")) if not lines: raise ValidationError("A receipt requires at least one active line.") lock_positions([destination], parts.values()) identifiers = {(line.spare_part_id, item.identifier) for line in lines for item in line.identifiers.all() if item.is_active} # Lock all existing identities in UUID order, before creating missing ones # in common (part UUID, identifier) order. Draft text is not authoritative. matching = Q(pk__in=[]) for part_id, identifier in identifiers: matching |= Q(spare_part_id=part_id, identifier=identifier) units = {(unit.spare_part_id, unit.identifier): unit for unit in SerializedStockUnit.objects.filter(matching).select_for_update().order_by("pk")} for part_id, identifier in sorted(identifiers, key=lambda pair: (str(pair[0]), pair[1])): if (part_id, identifier) not in units: units[(part_id, identifier)] = _register_unit(actor=actor, company=company, part=parts[part_id], identifier=identifier) for line in lines: entries = [entry for entry in line.identifiers.all() if entry.is_active] selected = [units[(line.spare_part_id, entry.identifier)] for entry in entries] line.movement = _post(actor=actor, company=company, spare_part=parts[line.spare_part_id], destination=destination, quantity=line.quantity, reference=current.number, idempotency_key=line.pk, units=selected, note=current.note) line._persist() for entry, unit in zip(entries, selected): entry.unit = unit entry._persist() current.status, current.received_by, current.received_at = "POSTED", actor, timezone.now() current._persist() return current def dispatch_stock_transfer(*, actor, transfer, expected_revision): with _document(transfer, actor, expected_revision, include_current_parts=True) as (current, actor, company, locations, parts): _draft(current) source, destination = locations[current.source_id], locations[current.destination_id] _physical(source) _physical(destination) lines = list(current.lines.filter(is_active=True).prefetch_related("units__unit").order_by("spare_part_id")) if not lines: raise ValidationError("A transfer requires at least one active line.") transit = InventoryLocation(company=company, code=f"TRANSIT-{current.pk.hex.upper()}", name=f"In transit: {current.number}", location_type="TRANSIT") transit._persist() # New private transit rows cannot be visible to another transaction yet. lock_positions([source, transit], parts.values()) all_units = [entry.unit_id for line in lines for entry in line.units.all() if entry.is_active] list(SerializedStockUnit.objects.filter(pk__in=all_units).select_for_update().order_by("pk")) for line in lines: line.dispatch_movement = _post(actor=actor, company=company, spare_part=parts[line.spare_part_id], source=source, destination=transit, quantity=line.quantity, reference=current.number, idempotency_key=line.pk, units=[entry.unit for entry in line.units.all() if entry.is_active]) line._persist() current.transit_location = transit current.status, current.dispatched_by, current.dispatched_at = "DISPATCHED", actor, timezone.now() current._persist() return current def receive_stock_transfer(*, actor, transfer, expected_revision): with _document(transfer, actor, expected_revision, include_current_parts=True) as (current, actor, company, locations, parts): if current.status != "DISPATCHED": raise ValidationError("Only a dispatched transfer can be received.") destination, transit = locations[current.destination_id], locations[current.transit_location_id] _physical(destination) lines = list(current.lines.filter(is_active=True).prefetch_related("units__unit").order_by("spare_part_id")) lock_positions([transit, destination], parts.values()) all_units = [entry.unit_id for line in lines for entry in line.units.all() if entry.is_active] list(SerializedStockUnit.objects.filter(pk__in=all_units).select_for_update().order_by("pk")) for line in lines: line.receive_movement = _post(actor=actor, company=company, spare_part=parts[line.spare_part_id], source=transit, destination=destination, quantity=line.quantity, reference=current.number, idempotency_key=uuid.uuid5(line.pk, "receive"), units=[entry.unit for entry in line.units.all() if entry.is_active]) line._persist() current.status, current.received_by, current.received_at = "RECEIVED", actor, timezone.now() current._persist() return current def _cancel(document, actor, expected_revision, reason): with _document(document, actor, expected_revision) as (current, actor, _, _, _): _draft(current) current.status, current.cancelled_by, current.cancelled_at = "CANCELLED", actor, timezone.now() current.cancellation_reason = text_value(reason, required=True, maximum=500) current._persist() return current def cancel_goods_receipt(*, actor, receipt, expected_revision, reason): return _cancel(receipt, actor, expected_revision, reason) def cancel_stock_transfer(*, actor, transfer, expected_revision, reason): return _cancel(transfer, actor, expected_revision, reason)