import json from hashlib import sha256 from django.db import transaction from django.db.models import Q from django.utils import timezone from competition_calendar_governance.constants import CalendarEditionSource, CalendarEditionStatus from competition_calendar_governance.models import CalendarFixtureAssignment, CompetitionCalendarEdition from competition_calendar_governance.services import ( CalendarGovernancePolicyService, CompetitionCalendarGovernanceService, VenueAssignmentService, _current_draw_ratification_hash, _event as calendar_event, ) from competition_calendar_governance.constants import CalendarGovernanceEventType from core_runtime.services import OutboxService from federations.constants import FederationAdminRole from federations.services import FederationService from match_result_governance.constants import ( CertifiedResultOutcome, ReplayOrderStatus, ResultCertificationStatus, ) from match_result_governance.models import MatchReplayOrder, MatchResultCertification from match_result_governance.services import ReplayOrderService from matches.constants import MatchStatus from matches.models import Fixture, Match from matches.services import MatchSchedulingService from .constants import ( REPLAY_EXECUTION_SCHEMA_VERSION, ReplayConsolidationOutcome, ReplayEventType, ReplayExecutionStatus, ) from .exceptions import ReplayExecutionIntegrityError, ReplayExecutionStateError from .models import ReplayExecutionEvent, ReplayFixtureExecution, ReplayResultConsolidation def stable_hash(payload) -> str: raw = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str).encode("utf-8") return sha256(raw).hexdigest() def _iso(value): return value.isoformat() if value else None def _require_authority(federation, actor): FederationService.require_role( federation=federation, user=actor, roles=[FederationAdminRole.GOVERNOR, FederationAdminRole.COMPETITION, FederationAdminRole.ADJUDICATION], ) def _event(*, execution, event_type, actor=None, reference=None, payload=None): execution = ReplayFixtureExecution.objects.select_for_update().get(pk=execution.pk) latest = execution.events.order_by("-sequence").first() sequence = (latest.sequence if latest else 0) + 1 previous_hash = latest.event_hash if latest else "" ref_type = reference.__class__.__name__ if reference is not None else "" ref_id = str(getattr(reference, "pk", "") or "") body = { "schema_version": REPLAY_EXECUTION_SCHEMA_VERSION, "execution_id": str(execution.pk), "sequence": sequence, "event_type": str(event_type), "reference_type": ref_type, "reference_id": ref_id, "payload": payload or {}, "previous_hash": previous_hash, } row = ReplayExecutionEvent.objects.create( execution=execution, sequence=sequence, event_type=event_type, reference_type=ref_type, reference_id=ref_id, payload=payload or {}, previous_hash=previous_hash, event_hash=stable_hash(body), actor=actor, ) OutboxService.enqueue( topic="match.replay-execution", event_type=f"match.replay-execution.{str(event_type).lower()}", aggregate_type="ReplayFixtureExecution", aggregate_id=str(execution.pk), deduplication_key=f"replay-execution:{row.pk}", payload={"execution_id": str(execution.pk), "event_id": str(row.pk)}, ) return row class ReplayFixtureExecutionService: @staticmethod def _parent_execution(original_match): return ReplayFixtureExecution.objects.filter(replacement_match=original_match).select_related("root_match").first() @classmethod @transaction.atomic def create(cls, *, replay_order, scheduled_at, actor): order = MatchReplayOrder.objects.select_for_update(of=("self", "original_match", "decision")).select_related( "original_match__fixture__federation_competition__federation", "original_match__fixture__federation_competition__season", "original_match__fixture__competition__season", "original_match__fixture__competition__league__federation_authority__federation", "decision__case__certification", ).get(pk=replay_order.pk) if order.status != ReplayOrderStatus.ORDERED or order.fulfilled_at is not None: if hasattr(order, "execution"): return order.execution raise ReplayExecutionStateError("Replay order is not open for execution creation.") original = order.original_match fixture = original.fixture federation_competition = fixture.federation_competition league_competition = fixture.competition if federation_competition is not None: federation = federation_competition.federation else: authority = getattr(league_competition.league, "federation_authority", None) if league_competition else None if authority is None: raise ReplayExecutionStateError("League replay requires a federation authority for the source league.") federation = authority.federation _require_authority(federation, actor) current_cert = original.result_certifications.exclude(status=ResultCertificationStatus.SUPERSEDED).order_by("-sequence").first() if current_cert is None or current_cert.outcome != CertifiedResultOutcome.REPLAY_REQUIRED or current_cert.finality_at is None: raise ReplayExecutionStateError("Original match does not have a final REPLAY_REQUIRED certification.") parent = cls._parent_execution(original) root = parent.root_match if parent else original next_seq = ( Fixture.objects.filter(replay_of_fixture=fixture).order_by("-replay_sequence") .values_list("replay_sequence", flat=True).first() or 0 ) + 1 replacement_fixture = Fixture.objects.create( competition=league_competition, federation_competition=federation_competition, draw_generation=fixture.draw_generation, replay_of_fixture=fixture, replay_sequence=next_seq, round_number=fixture.round_number, leg=fixture.leg, home_club=fixture.home_club, away_club=fixture.away_club, competition_type=fixture.competition_type, home_youth_team=fixture.home_youth_team, away_youth_team=fixture.away_youth_team, ) replacement_match = Match.objects.create(fixture=replacement_fixture, status=MatchStatus.PENDING) MatchSchedulingService.schedule( match_id=replacement_match.pk, scheduled_at=scheduled_at, record_revision=True, allow_reschedule=False, revision_reason_text=f"Replay fixture ordered by MatchReplayOrder {order.pk}.", ) replacement_match.refresh_from_db() draw_hash = _current_draw_ratification_hash(federation_competition) if federation_competition is not None else "" payload = { "schema_version": REPLAY_EXECUTION_SCHEMA_VERSION, "replay_order_id": str(order.pk), "replay_order_hash": order.order_hash, "root_match_id": str(root.pk), "original_match_id": str(original.pk), "original_fixture_id": str(fixture.pk), "replacement_fixture_id": str(replacement_fixture.pk), "replacement_match_id": str(replacement_match.pk), "parent_execution_id": str(parent.pk) if parent else None, "replay_sequence": int(next_seq), "competition_source": "FEDERATION" if federation_competition is not None else "LEAGUE", "competition_id": str(federation_competition.pk if federation_competition is not None else league_competition.pk), "draw_generation": int(fixture.draw_generation), "draw_ratification_hash": draw_hash, "scheduled_at": _iso(replacement_match.scheduled_at), "created_by_id": str(actor.pk), } execution = ReplayFixtureExecution.objects.create( replay_order=order, root_match=root, original_match=original, original_fixture=fixture, replacement_fixture=replacement_fixture, replacement_match=replacement_match, parent_execution=parent, replay_sequence=next_seq, status=ReplayExecutionStatus.CREATED, draw_ratification_hash=draw_hash, execution_payload=payload, execution_hash=stable_hash(payload), created_by=actor, ) _event( execution=execution, event_type=ReplayEventType.EXECUTION_CREATED, actor=actor, reference=replacement_match, payload={"execution_hash": execution.execution_hash, "scheduled_at": _iso(replacement_match.scheduled_at)}, ) ReplayOrderService.fulfill(replay_order=order, replacement_match=replacement_match, actor=actor) _event( execution=execution, event_type=ReplayEventType.ORDER_FULFILLED, actor=actor, reference=order, payload={"fulfillment_hash": order.fulfillment_hash, "replacement_match_id": str(replacement_match.pk)}, ) calendar = ReplayCalendarBridge.create_amendment(execution=execution, actor=actor) if federation_competition is not None else None execution.calendar_edition = calendar execution.status = ReplayExecutionStatus.CALENDAR_PENDING if calendar and calendar.status == CalendarEditionStatus.DRAFT else ReplayExecutionStatus.READY execution.save(update_fields=["calendar_edition", "status", "updated_at"]) from regulatory_evidence.services import RegulatoryEvidenceBridgeService RegulatoryEvidenceBridgeService.capture_replay_fixture_execution(execution=execution, actor=actor) return execution @classmethod def refresh_status(cls, execution): execution = ReplayFixtureExecution.objects.select_related("replacement_match", "calendar_edition").get(pk=execution.pk) if hasattr(execution, "consolidation"): wanted = ReplayExecutionStatus.CONSOLIDATED elif execution.replacement_match.status == MatchStatus.IN_PROGRESS: wanted = ReplayExecutionStatus.IN_PROGRESS elif execution.replacement_match.status == MatchStatus.COMPLETED: wanted = ReplayExecutionStatus.AWAITING_RESULT elif execution.calendar_edition_id and execution.calendar_edition.status != CalendarEditionStatus.RATIFIED: wanted = ReplayExecutionStatus.CALENDAR_PENDING else: wanted = ReplayExecutionStatus.READY if wanted != execution.status: execution.status = wanted execution.save(update_fields=["status", "updated_at"]) return execution class ReplayCalendarBridge: @classmethod @transaction.atomic def create_amendment(cls, *, execution, actor): competition = execution.replacement_fixture.federation_competition policy, policy_payload, governance = CalendarGovernancePolicyService.resolve(competition=competition) if policy is None and not governance: return None previous = CompetitionCalendarGovernanceService.current_edition(competition) if previous is None or previous.status != CalendarEditionStatus.RATIFIED: raise ReplayExecutionStateError("Replay fixture requires an active RATIFIED calendar edition before amendment.") if previous.draw_generation != execution.replacement_fixture.draw_generation: raise ReplayExecutionStateError("Active calendar edition belongs to a different draw generation.") latest = CompetitionCalendarGovernanceService.latest_edition(competition) if latest and latest.status == CalendarEditionStatus.DRAFT: raise ReplayExecutionStateError("An unresolved DRAFT calendar edition already exists.") sequence = (latest.sequence if latest else 0) + 1 assignments = list(previous.fixture_assignments.select_related( "match", "home_stadium_profile", "stadium_asset", "schedule_revision" ).order_by("match_id")) match = execution.replacement_match venue = VenueAssignmentService.default_for_match(match=match, policy_payload=policy_payload) revision = match.schedule_revisions.order_by("-revision_number").first() replay_payload = CompetitionCalendarGovernanceService._assignment_payload( match=match, scheduled_at=match.scheduled_at, venue=venue, schedule_revision=revision ) manifest_rows = [(str(row.match_id), row.assignment_hash) for row in assignments] manifest_rows.append((str(match.pk), stable_hash(replay_payload))) manifest_rows.sort(key=lambda item: item[0]) hashes = [item[1] for item in manifest_rows] calendar_payload = { "schema_version": 1, "competition_id": str(competition.pk), "sequence": sequence, "draw_generation": int(previous.draw_generation), "draw_ratification_hash": _current_draw_ratification_hash(competition), "source": CalendarEditionSource.AMENDMENT, "amendment_reference_type": "ReplayFixtureExecution", "amendment_reference_id": str(execution.pk), "assignment_hashes": hashes, "assignment_count": len(hashes), } ratification_payload = { "schema_version": 1, "competition_id": str(competition.pk), "sequence": sequence, "draw_generation": int(previous.draw_generation), "draw_ratification_hash": calendar_payload["draw_ratification_hash"], "calendar_hash": stable_hash(calendar_payload), "policy_hash": policy.policy_hash if policy else "", "rulebook_governance_hash": stable_hash(governance), "previous_ratification_hash": previous.ratification_hash, } edition = CompetitionCalendarEdition.objects.create( competition=competition, previous_edition=previous, policy=policy, sequence=sequence, draw_generation=previous.draw_generation, source=CalendarEditionSource.AMENDMENT, amendment_reference_type="ReplayFixtureExecution", amendment_reference_id=str(execution.pk), calendar_payload=calendar_payload, calendar_hash=stable_hash(calendar_payload), policy_snapshot={ "policy_id": str(policy.pk) if policy else None, "policy_hash": policy.policy_hash if policy else "", "policy_payload": policy_payload, }, rulebook_governance=governance, ratification_payload=ratification_payload, ratification_hash=stable_hash(ratification_payload), status=CalendarEditionStatus.DRAFT, created_by=actor, ) for old in assignments: CalendarFixtureAssignment.objects.create( edition=edition, match=old.match, scheduled_at=old.scheduled_at, timezone_name=old.timezone_name, venue_source=old.venue_source, home_stadium_profile=old.home_stadium_profile, stadium_asset=old.stadium_asset, venue_name=old.venue_name, venue_snapshot=old.venue_snapshot, schedule_revision=old.schedule_revision, assignment_payload=old.assignment_payload, assignment_hash=old.assignment_hash, ) CalendarFixtureAssignment.objects.create( edition=edition, match=match, scheduled_at=match.scheduled_at, timezone_name=venue["timezone_name"], venue_source=venue["venue_source"], home_stadium_profile=venue["home_stadium_profile"], stadium_asset=venue["stadium_asset"], venue_name=venue["venue_name"], venue_snapshot=venue["venue_snapshot"], schedule_revision=revision, assignment_payload=replay_payload, assignment_hash=stable_hash(replay_payload), ) calendar_event( competition=competition, event_type=CalendarGovernanceEventType.EDITION_CREATED, actor=actor, edition=edition, reference=edition, payload={"calendar_hash": edition.calendar_hash, "source": CalendarEditionSource.AMENDMENT, "replay_execution_id": str(execution.pk)}, ) _event( execution=execution, event_type=ReplayEventType.CALENDAR_DRAFTED, actor=actor, reference=edition, payload={"calendar_hash": edition.calendar_hash, "ratification_hash": edition.ratification_hash}, ) return edition class ReplayResultConsolidationService: COUNTABLE = { CertifiedResultOutcome.CONFIRMED, CertifiedResultOutcome.AWARDED_HOME, CertifiedResultOutcome.AWARDED_AWAY, CertifiedResultOutcome.VOID, } @classmethod def current_for_original(cls, match): execution = ReplayFixtureExecution.objects.filter(original_match=match).select_related( "replacement_match", "root_match" ).order_by("-created_at").first() if execution is None: return None try: return execution.consolidation except ReplayResultConsolidation.DoesNotExist: return None @classmethod @transaction.atomic def consolidate_for_replacement(cls, *, replacement_match_id, actor=None): execution = ReplayFixtureExecution.objects.select_for_update(of=("self", "replacement_match", "original_match", "root_match", "replay_order", "replacement_fixture")).select_related( "replacement_match", "original_match", "root_match", "parent_execution", "replay_order", "replacement_fixture" ).filter(replacement_match_id=replacement_match_id).first() if execution is None: return None try: return execution.consolidation except ReplayResultConsolidation.DoesNotExist: pass cert = execution.replacement_match.result_certifications.exclude( status=ResultCertificationStatus.SUPERSEDED ).order_by("-sequence").first() if cert is None or cert.status != ResultCertificationStatus.CERTIFIED or cert.finality_at is None: return None if cert.outcome == CertifiedResultOutcome.REPLAY_REQUIRED: execution.status = ReplayExecutionStatus.AWAITING_RESULT execution.save(update_fields=["status", "updated_at"]) return None if cert.outcome not in cls.COUNTABLE: return None consolidation = cls._create_for_execution( execution=execution, terminal_match=execution.replacement_match, terminal_cert=cert, lineage=[{"execution_id": str(execution.pk), "execution_hash": execution.execution_hash}], actor=actor, ) cls._propagate_to_parents(execution=execution, terminal_match=execution.replacement_match, terminal_cert=cert, actor=actor) transaction.on_commit(lambda: cls._progress_root(match_id=execution.root_match_id)) return consolidation @classmethod def _create_for_execution(cls, *, execution, terminal_match, terminal_cert, lineage, actor=None): existing = ReplayResultConsolidation.objects.filter(execution=execution).first() if existing: return existing payload = { "schema_version": REPLAY_EXECUTION_SCHEMA_VERSION, "execution_id": str(execution.pk), "execution_hash": execution.execution_hash, "replay_order_id": str(execution.replay_order_id), "replay_order_hash": execution.replay_order.order_hash, "root_match_id": str(execution.root_match_id), "original_match_id": str(execution.original_match_id), "direct_replacement_match_id": str(execution.replacement_match_id), "terminal_match_id": str(terminal_match.pk), "terminal_certification_id": str(terminal_cert.pk), "terminal_certification_hash": terminal_cert.certification_hash, "outcome": terminal_cert.outcome, "effective_home_score": terminal_cert.effective_home_score, "effective_away_score": terminal_cert.effective_away_score, "lineage": lineage, } evidence_actor = actor or execution.created_by if evidence_actor is None: raise ReplayExecutionStateError("Replay consolidation requires an accountable federation actor.") row = ReplayResultConsolidation.objects.create( execution=execution, root_match=execution.root_match, direct_replacement_match=execution.replacement_match, terminal_match=terminal_match, terminal_certification=terminal_cert, outcome=terminal_cert.outcome, effective_home_score=terminal_cert.effective_home_score, effective_away_score=terminal_cert.effective_away_score, lineage_depth=len(lineage), lineage_payload=payload, consolidation_hash=stable_hash(payload), consolidated_by=evidence_actor, ) execution.status = ReplayExecutionStatus.CONSOLIDATED execution.save(update_fields=["status", "updated_at"]) from regulatory_evidence.services import RegulatoryEvidenceBridgeService RegulatoryEvidenceBridgeService.capture_replay_result_consolidation(consolidation=row, actor=evidence_actor) _event( execution=execution, event_type=ReplayEventType.CONSOLIDATED, actor=actor, reference=row, payload={"consolidation_hash": row.consolidation_hash, "terminal_certification_hash": terminal_cert.certification_hash}, ) return row @classmethod def _propagate_to_parents(cls, *, execution, terminal_match, terminal_cert, actor=None): current = execution.parent_execution child = execution lineage = [{"execution_id": str(execution.pk), "execution_hash": execution.execution_hash}] while current is not None: lineage = [{"execution_id": str(current.pk), "execution_hash": current.execution_hash}] + lineage row = cls._create_for_execution( execution=current, terminal_match=terminal_match, terminal_cert=terminal_cert, lineage=lineage, actor=actor, ) _event( execution=current, event_type=ReplayEventType.ANCESTOR_CONSOLIDATED, actor=actor, reference=row, payload={"child_execution_id": str(child.pk), "terminal_match_id": str(terminal_match.pk)}, ) child = current current = current.parent_execution @staticmethod def _progress_root(*, match_id): from match_result_governance.services import ResultProjectionService try: return ResultProjectionService.try_progress(match_id=match_id) except Exception: return None @classmethod def projection_for_original(cls, *, match, require_final=False): row = cls.current_for_original(match) if row is None: return None cert = row.terminal_certification if cert.status != ResultCertificationStatus.CERTIFIED: return None if require_final and cert.finality_at is None: return None if stable_hash(row.lineage_payload) != row.consolidation_hash: raise ReplayExecutionIntegrityError("Replay consolidation hash verification failed.") if row.lineage_payload.get("terminal_certification_hash") != cert.certification_hash: raise ReplayExecutionIntegrityError("Replay consolidation references a stale terminal certification.") if row.outcome == CertifiedResultOutcome.VOID: return None return { "governed": True, "certification_id": str(cert.pk), "certification_hash": cert.certification_hash, "final": cert.finality_at is not None, "outcome": cert.outcome, "home_score": row.effective_home_score, "away_score": row.effective_away_score, "result_type": "PLAYED" if cert.outcome == CertifiedResultOutcome.CONFIRMED else "ADMINISTRATIVE", "replay_consolidation_id": str(row.pk), "replay_consolidation_hash": row.consolidation_hash, "replay_terminal_match_id": str(row.terminal_match_id), } class ReplayExecutionIntegrityService: @classmethod def verify_execution(cls, execution): errors = [] if stable_hash(execution.execution_payload) != execution.execution_hash: errors.append("execution-hash") if execution.replacement_fixture.replay_of_fixture_id != execution.original_fixture_id: errors.append("fixture-lineage") if execution.replacement_fixture.replay_sequence != execution.replay_sequence: errors.append("fixture-sequence") order = execution.replay_order if order.status != ReplayOrderStatus.FULFILLED or order.replacement_match_id != execution.replacement_match_id: errors.append("order-fulfillment") try: con = execution.consolidation except ReplayResultConsolidation.DoesNotExist: con = None if con: if stable_hash(con.lineage_payload) != con.consolidation_hash: errors.append("consolidation-hash") if con.lineage_payload.get("terminal_certification_hash") != con.terminal_certification.certification_hash: errors.append("terminal-certification") previous = "" expected = 1 for event in execution.events.order_by("sequence"): body = { "schema_version": REPLAY_EXECUTION_SCHEMA_VERSION, "execution_id": str(execution.pk), "sequence": event.sequence, "event_type": event.event_type, "reference_type": event.reference_type, "reference_id": event.reference_id, "payload": event.payload or {}, "previous_hash": previous, } if event.sequence != expected or event.previous_hash != previous or stable_hash(body) != event.event_hash: errors.append(f"event:{event.pk}") previous = event.event_hash expected += 1 return {"execution_id": str(execution.pk), "valid": not errors, "errors": errors} @classmethod def verify_all(cls): executions = ReplayFixtureExecution.objects.select_related( "replacement_fixture", "original_fixture", "replay_order" ).order_by("created_at") results = [cls.verify_execution(row) for row in executions] return { "executions": len(results), "invalid": sum(1 for row in results if not row["valid"]), "results": results, } def consume_domain_event(event_id): from core_runtime.models import OutboxEvent event = OutboxEvent.objects.filter(pk=event_id).first() if event is None: return {"consumer": "replay_execution", "handled": False, "reason": "event-not-found"} if event.event_type != "match.result-governance.certification_final": return {"consumer": "replay_execution", "handled": False, "reason": "not-replay-result-event"} match_id = (event.payload or {}).get("match_id") if not match_id: return {"consumer": "replay_execution", "handled": False, "reason": "missing-match"} row = ReplayResultConsolidationService.consolidate_for_replacement(replacement_match_id=match_id) return { "consumer": "replay_execution", "handled": bool(row), "consolidation_id": str(row.pk) if row else None, }