"""Rebuild the transaction participation index from the revision catalogue.
The participation index is a derived, read-side cache co-written atomically with
revision persistence; the calculation-revision catalogue is the authoritative
source of truth. This module regenerates the index from scratch by iterating the
full finalized-revision catalogue, so a stale or corrupt index can be replaced
without data loss, in line with the
``aeat-ledger-contract`` project rule.
The rebuild reads the :class:`CalculationRevisionCatalogue` (the finalized
revisions and their ``source_transaction_ids``), the work-unit catalogue (for the
``modelo`` / ``filing_year`` / ``period`` axis per revision), and the
:class:`ModeloRecordCatalogue` / :class:`ModeloRecord` rows (to attach the
``filing_record_id`` and any justificante reference to filed participations).
Borrador and discarded revisions are excluded — the index records only the
finalized audit surface.
See Also:
:class:`TransactionRevisionParticipationIndex`:
Per-transaction secure object rebuilt by this module.
:func:`cadrumo.application.modelo.verification_actions._build_participation_writes`:
Write-time co-emission path that keeps the index current during
verification persistence.
"""
from __future__ import annotations
from pydantic import BaseModel, NonNegativeInt
from ...core.identity.hex_ids import CalculationRevisionId
from ...core.models import STRICT_FROZEN_CONFIG as _STRICT_FROZEN
from ...domain.modelos.calculation_revision import (
SEALED_REVISION_STATES,
CalculationRevision,
CalculationRevisionState,
)
from ...domain.modelos.filing_record import ExternalEvidence, ModeloRecord, ModeloRecordCatalogue
from ...domain.modelos.participation_index import (
TransactionRevisionParticipation,
TransactionRevisionParticipationIndex,
upsert_transaction_participation,
)
from ...domain.modelos.work_unit import WorkUnitCatalogue
from .participation_index_rebuild_ports import ParticipationIndexRebuildPorts
[docs]
class ParticipationRebuildStats(BaseModel):
"""Outcome of one participation-index rebuild pass.
Attributes:
transaction_count: Number of distinct ledger transactions that gained at
least one participation entry.
participation_count: Total number of (transaction, finalized-revision)
participation entries written across all transactions.
revision_count: Number of finalized revisions folded into the index.
stale_removed_count: Number of previously persisted participation objects
removed because the regenerated catalogue no longer records them —
the transactions whose finalized revisions were discarded, removed,
or reverted to borrador since the last rebuild.
"""
model_config = _STRICT_FROZEN
transaction_count: NonNegativeInt
participation_count: NonNegativeInt
revision_count: NonNegativeInt
stale_removed_count: NonNegativeInt
def _filing_record_for_revision(
calculation_revision_id: CalculationRevisionId,
filings: ModeloRecordCatalogue,
) -> ModeloRecord | None:
"""Return the filing record bound to ``calculation_revision_id``, if any."""
for record in filings.records.values():
if record.calculation_revision_id == calculation_revision_id:
return record
return None
def _justificante_reference(record: ModeloRecord | None) -> str | None:
"""Return the imported justificante reference from a filing record, if present."""
evidence: ExternalEvidence | None = record.external_evidence if record is not None else None
return evidence.reference_id if evidence is not None else None
def _participation_for_sealed_revision(
revision: CalculationRevision,
work_units: WorkUnitCatalogue,
filings: ModeloRecordCatalogue,
) -> tuple[TransactionRevisionParticipation, tuple[str, ...]] | None:
"""Build one participation row and its source ids when ``revision`` is indexable."""
if revision.state not in SEALED_REVISION_STATES:
return None
work_unit = work_units.get(revision.work_unit_id)
if work_unit is None:
return None
is_filed = revision.state in {
CalculationRevisionState.PRESENTADO,
CalculationRevisionState.PRESENTADO_SUPERSEDIDO,
}
filing_record = _filing_record_for_revision(revision.calculation_revision_id, filings) if is_filed else None
participation = TransactionRevisionParticipation(
calculation_revision_id=revision.calculation_revision_id,
work_unit_id=work_unit.work_unit_id,
modelo=work_unit.modelo,
filing_year=work_unit.filing_year,
period=work_unit.period,
revision_state=revision.state.value,
filing_record_id=filing_record.filing_record_id if filing_record is not None else None,
justificante_reference=_justificante_reference(filing_record),
)
return participation, revision.source_transaction_ids
def _fold_participation(
rebuilt: dict[str, TransactionRevisionParticipationIndex],
participation: TransactionRevisionParticipation,
transaction_ids: tuple[str, ...],
) -> int:
"""Merge one revision's participation into each contributing transaction."""
participation_count = 0
for transaction_id in transaction_ids:
current = rebuilt.get(transaction_id) or TransactionRevisionParticipationIndex(
transaction_id=transaction_id,
)
rebuilt[transaction_id] = upsert_transaction_participation(current, participation)
participation_count += 1
return participation_count
[docs]
def rebuild_participation_index(
*,
ports: ParticipationIndexRebuildPorts,
) -> ParticipationRebuildStats:
"""Regenerate the participation index from the finalized-revision catalogue.
Iterates every finalized :class:`CalculationRevision`, resolves its
``modelo`` / ``filing_year`` / ``period`` from the work-unit catalogue and its
``filing_record_id`` from the filing-record catalogue, and writes one
:class:`TransactionRevisionParticipationIndex` per contributing transaction.
Borrador and discarded revisions are excluded.
The regenerated set REPLACES the persisted index rather than being upserted
over it: a transaction whose finalized revisions have since been discarded or
removed contributes no entry here, and its stale secure object is pruned in
the same atomic batch. Without that prune a removed revision would stay
visible through the derived cache while the authoritative catalogue no longer
records it. Returns a :class:`ParticipationRebuildStats` summary.
"""
revisions = ports.calculation_repository.load()
work_units = ports.work_unit_repository.load()
filings = ports.filing_repository.load()
rebuilt: dict[str, TransactionRevisionParticipationIndex] = {}
participation_count = 0
revision_count = 0
for revision in revisions.values():
resolved = _participation_for_sealed_revision(revision, work_units, filings)
if resolved is None:
continue
participation, transaction_ids = resolved
revision_count += 1
participation_count += _fold_participation(rebuilt, participation, transaction_ids)
stale_removed_count = ports.participation_index_repository.replace_all(rebuilt.values())
return ParticipationRebuildStats(
transaction_count=len(rebuilt),
participation_count=participation_count,
revision_count=revision_count,
stale_removed_count=stale_removed_count,
)
__all__ = [
"ParticipationRebuildStats",
"rebuild_participation_index",
]