import uuid from dataclasses import dataclass from django.conf import settings from django.db import transaction from django.db.models import Max from api.models import ( IntegrationOutbox, IntegrationTransportCursor, TtLocation, TtStaff, ) from api.services.integration.outbox import ( append_resource_event, canonical_hash, capture_resource_state, finalize_change_set, integration_source_scope, ) class ResourceBootstrapSafetyError(RuntimeError): pass RESOURCE_BOOTSTRAP_ORIGIN = "resource_bootstrap" RESOURCE_PROJECTION_REVISION_ORIGIN = "resource_projection_revision" @dataclass class ResourceBootstrapSummary: run_id: str staff_scanned: int = 0 staff_emitted: int = 0 staff_skipped: int = 0 location_scanned: int = 0 location_emitted: int = 0 location_skipped: int = 0 change_sets: int = 0 last_sequence: int = 0 @property def resources_scanned(self) -> int: return self.staff_scanned + self.location_scanned @property def resources_emitted(self) -> int: return self.staff_emitted + self.location_emitted @property def resources_skipped(self) -> int: return self.staff_skipped + self.location_skipped def as_dict(self) -> dict[str, int | str]: return { "run_id": self.run_id, "staff_scanned": self.staff_scanned, "staff_emitted": self.staff_emitted, "staff_skipped": self.staff_skipped, "location_scanned": self.location_scanned, "location_emitted": self.location_emitted, "location_skipped": self.location_skipped, "resources_scanned": self.resources_scanned, "resources_emitted": self.resources_emitted, "resources_skipped": self.resources_skipped, "change_sets": self.change_sets, "last_sequence": self.last_sequence, } def assert_resource_bootstrap_safe() -> None: config = settings.RESOURCE_BOOKING_INTEGRATION if not config["CAPTURE_ENABLED"]: raise ResourceBootstrapSafetyError( "resource bootstrap requires outbox capture enabled" ) if config["PUBLISH_ENABLED"] or config["ACTIVATION_APPROVED"]: raise ResourceBootstrapSafetyError( "resource bootstrap requires publication and activation approval disabled" ) if config["REVERSE_DELIVERY_ENABLED"]: raise ResourceBootstrapSafetyError( "resource bootstrap requires reverse delivery disabled" ) if config["PHASE"] != "phase1" or config["SCHEMA_VERSION"] != "2": raise ResourceBootstrapSafetyError( "resource bootstrap requires Phase 1 schema 2" ) def _latest_events(aggregate_type: str, resources) -> dict[str, IntegrationOutbox]: aggregate_ids = [ f"{settings.RESOURCE_BOOKING_INTEGRATION['DEPLOYMENT_ID']}:{resource.pk}" for resource in resources ] rows = IntegrationOutbox.objects.filter( source_scope=integration_source_scope(), aggregate_type=aggregate_type, aggregate_id__in=aggregate_ids, transaction_finalized=True, ).order_by("aggregate_id", "-event_version", "-id") latest = {} for event in rows: latest.setdefault(event.aggregate_id, event) return latest def _latest_current_state(event: IntegrationOutbox | None): if event is None: return None committed_state = event.payload.get("committed_state") or {} replacement = committed_state.get("replacement") or {} if committed_state.get("tombstone"): return None return replacement.get("current") def _validate_identifier(value: str, label: str) -> str: if ( not value or len(value) > 190 or any(character in value for character in "\r\n\x00") ): raise ValueError(f"{label} must be a non-empty bounded single-line identifier") return value def _capture_resource_snapshots( *, batch_size: int, actor_id: int | None, run_id: str, projection_operation_key: str | None, ) -> ResourceBootstrapSummary: summary = ResourceBootstrapSummary(run_id=run_id) origin = ( RESOURCE_PROJECTION_REVISION_ORIGIN if projection_operation_key else RESOURCE_BOOTSTRAP_ORIGIN ) boundaries = { "staff": TtStaff.objects.aggregate(value=Max("id"))["value"] or 0, "location": TtLocation.objects.aggregate(value=Max("id"))["value"] or 0, } for aggregate_type, model in (("staff", TtStaff), ("location", TtLocation)): after_id = 0 boundary = boundaries[aggregate_type] while after_id < boundary: candidate_ids = list( model.objects.filter(id__gt=after_id, id__lte=boundary) .order_by("id") .values_list("id", flat=True)[:batch_size] ) if not candidate_ids: break after_id = candidate_ids[-1] batch_emitted = 0 batch_skipped = 0 batch_sequence = 0 with transaction.atomic(): resources = list( model.objects.select_for_update() .filter(id__in=candidate_ids) .order_by("id") ) latest = _latest_events(aggregate_type, resources) completed_aggregate_ids = set() if projection_operation_key: completed_aggregate_ids = set( IntegrationOutbox.objects.filter( source_scope=integration_source_scope(), aggregate_type=aggregate_type, aggregate_id__in=latest.keys(), origin=RESOURCE_PROJECTION_REVISION_ORIGIN, request_id=projection_operation_key, transaction_finalized=True, ).values_list("aggregate_id", flat=True) ) change_set_id = uuid.uuid4() for resource in resources: current = capture_resource_state(resource) aggregate_id = f"{settings.RESOURCE_BOOKING_INTEGRATION['DEPLOYMENT_ID']}:{resource.pk}" prior_event = latest.get(aggregate_id) prior_current = _latest_current_state(prior_event) if projection_operation_key: should_skip = aggregate_id in completed_aggregate_ids else: should_skip = prior_current is not None and canonical_hash( prior_current ) == canonical_hash(current) if should_skip: batch_skipped += 1 continue event = append_resource_event( resource, action="snapshot", previous=prior_current, actor_id=actor_id, origin=origin, correlation_id=run_id, request_id=run_id, change_set_id=change_set_id, defer_change_set_finalization=True, ) if event is None: raise ResourceBootstrapSafetyError( "capture was disabled while resource recovery was running" ) batch_emitted += 1 batch_sequence = event.transport_sequence if batch_emitted: finalized = finalize_change_set(change_set_id) if finalized != batch_emitted: raise ResourceBootstrapSafetyError( "resource recovery change-set finalization count is inconsistent" ) scanned = batch_emitted + batch_skipped setattr( summary, f"{aggregate_type}_scanned", getattr(summary, f"{aggregate_type}_scanned") + scanned, ) setattr( summary, f"{aggregate_type}_emitted", getattr(summary, f"{aggregate_type}_emitted") + batch_emitted, ) setattr( summary, f"{aggregate_type}_skipped", getattr(summary, f"{aggregate_type}_skipped") + batch_skipped, ) if batch_emitted: summary.change_sets += 1 summary.last_sequence = max(summary.last_sequence, batch_sequence) current_cursor = IntegrationTransportCursor.objects.filter( source_scope=integration_source_scope() ).first() if current_cursor is not None: summary.last_sequence = max(summary.last_sequence, current_cursor.last_sequence) return summary def bootstrap_resource_snapshots( *, batch_size: int = 100, actor_id: int | None = None, run_id: str | None = None, ) -> ResourceBootstrapSummary: assert_resource_bootstrap_safe() maximum = settings.RESOURCE_BOOKING_INTEGRATION.get( "MAX_EVENTS_PER_CHANGE_SET", 5000 ) if batch_size < 1 or batch_size > maximum: raise ValueError(f"batch_size must be between 1 and {maximum}") resolved_run_id = run_id or f"resource-bootstrap-{uuid.uuid4()}" _validate_identifier(resolved_run_id, "run_id") return _capture_resource_snapshots( batch_size=batch_size, actor_id=actor_id, run_id=resolved_run_id, projection_operation_key=None, ) def revise_resource_projection_snapshots( *, operation_key: str, batch_size: int = 100, actor_id: int | None = None, ) -> ResourceBootstrapSummary: """Re-emit current resources once for an explicit projection-contract revision.""" assert_resource_bootstrap_safe() maximum = settings.RESOURCE_BOOKING_INTEGRATION.get( "MAX_EVENTS_PER_CHANGE_SET", 5000 ) if batch_size < 1 or batch_size > maximum: raise ValueError(f"batch_size must be between 1 and {maximum}") resolved_operation_key = _validate_identifier(operation_key, "operation_key") return _capture_resource_snapshots( batch_size=batch_size, actor_id=actor_id, run_id=resolved_operation_key, projection_operation_key=resolved_operation_key, )