"""Finance domain service — period close/reopen + rollup orchestration. Thin layer over the cross-domain :mod:`app.core.coordinators.finance_spend_coordinator`. Owns the period-close lifecycle (the accounting freeze) and the FX-freeze step that close depends on; the actual cross-domain aggregation lives in the coordinator. PR2 adds the read surface (summary / pivot / timeseries / detail) on top of the materialised ledger; this module stays focused on materialisation and close-state mutation. """ from __future__ import annotations import re from datetime import UTC, datetime from decimal import Decimal from sqlalchemy import select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.core.coordinators import finance_spend_coordinator as coord from app.core.exceptions import AppException from app.core.safe_logging import get_safe_logger from app.domains.finance.models import ( CLOSE_CLOSED, CLOSE_OPEN, FinancePeriodClose, ) logger = get_safe_logger(__name__) _PERIOD_RE = re.compile(r"^\d{4}-(0[1-9]|1[0-2])$") def validate_period(period: str) -> str: """Validate a ``YYYY-MM`` period string, returning it normalised.""" p = (period or "").strip() if not _PERIOD_RE.match(p): raise AppException( code="BAD_PERIOD", message="period must be a valid calendar month 'YYYY-MM'.", status_code=400, ) return p async def get_close(db: AsyncSession, period: str) -> FinancePeriodClose | None: return ( await db.execute( select(FinancePeriodClose).where(FinancePeriodClose.period == period) ) ).scalar_one_or_none() async def refresh_period(db: AsyncSession, period: str) -> dict: """Materialise the ledger for ``period`` (respects the closed freeze). A closed period is not replaced — drift is recorded as a restatement. Use :func:`reopen_period` first to recompute and republish a closed month. """ period = validate_period(period) return await coord.materialize_period(db, period) async def close_period( db: AsyncSession, period: str, *, actor_id: str | None, note: str | None = None, ) -> dict: """Freeze ``period``: finalise its facts and snapshot the total. Steps, in order: 1. Freeze the month's FX rates (best-effort backfill) so the USD figures are final. 2. Mark the close row ``closed`` so the subsequent materialise stamps facts non-provisional. 3. Materialise (allow_closed) → replaces facts as final and writes ``closed_total_usd``. """ period = validate_period(period) existing = await get_close(db, period) if existing is not None and existing.status == CLOSE_CLOSED: raise AppException( code="PERIOD_ALREADY_CLOSED", message=f"Period {period} is already closed.", status_code=409, ) await _freeze_period_fx(db, period) if existing is None: # First-ever close of this period: no ``finance_period_close`` row yet. # Two near-simultaneous first-closes both read ``None`` above and both # try to INSERT — the second trips ``uq_finance_period_close_period``. # Wrap the INSERT in a SAVEPOINT so the 1062 rolls back only the # nested write (not the whole session), then re-select the winner with # a LOCKING current-read — a plain snapshot read under REPEATABLE READ # was fixed at the ``get_close`` above and would still see no row. # Same idiom as ``close_service.ensure_run``. existing = FinancePeriodClose(period=period) try: async with db.begin_nested(): db.add(existing) await db.flush() except IntegrityError: existing = ( await db.execute( select(FinancePeriodClose) .where(FinancePeriodClose.period == period) .with_for_update() .execution_options(populate_existing=True) ) ).scalar_one_or_none() if existing is None: # pragma: no cover — non-period IntegrityError raise if existing.status == CLOSE_CLOSED: raise AppException( code="PERIOD_ALREADY_CLOSED", message=f"Period {period} is already closed.", status_code=409, ) existing.status = CLOSE_CLOSED existing.closed_at = datetime.now(UTC) existing.closed_by_id = actor_id existing.note = note await db.flush() result = await coord.materialize_period(db, period, allow_closed=True) logger.info("finance.close: period=%s by=%s total=%s", period, actor_id, result.get("consumption_total_usd")) return {"period": period, "status": CLOSE_CLOSED, **result} async def reopen_period( db: AsyncSession, period: str, *, actor_id: str | None, note: str | None = None, ) -> dict: """Reopen a closed ``period`` for correction. Sets the close row back to ``open`` and re-materialises (facts become provisional again). The prior ``closed_total_usd`` is retained as the restatement baseline so the next close can diff against it. """ period = validate_period(period) existing = await get_close(db, period) if existing is None or existing.status != CLOSE_CLOSED: raise AppException( code="PERIOD_NOT_CLOSED", message=f"Period {period} is not closed; nothing to reopen.", status_code=409, ) existing.status = CLOSE_OPEN existing.reopened_at = datetime.now(UTC) existing.reopened_by_id = actor_id if note: existing.note = note await db.flush() result = await coord.materialize_period(db, period) return {"period": period, "status": CLOSE_OPEN, **result} async def reconcile_internal(db: AsyncSession, period: str) -> dict: period = validate_period(period) return await coord.reconcile_internal(db, period) async def _freeze_period_fx(db: AsyncSession, period: str) -> None: """Backfill + freeze FX rates for every native currency in the period, then REFUSE to close if any currency can't resolve to a real rate. Best-effort freeze: a currency with a fetchable rate is frozen so the published USD figure is stable against later live-rate moves. But the close also stamps ``closed_total_usd`` as the authoritative restatement baseline — so if a non-USD currency would fall through to the unconverted ``1.0`` sentinel (frankfurter unreachable AND no frozen/platform rate), materialising would book e.g. ¥1,500,000 as $1,500,000 (~156× over) into the final closed total, and a later correct recompute would log a giant bogus "restatement". Rather than freeze a wrong baseline we block the close; the admin retries once the FX source is reachable or pins the rate. """ from app.core.coordinators import finance_spend_coordinator as coord from app.domains.finance import fx as finance_fx currencies = await coord.discover_native_currencies(db) for currency in currencies: row = await finance_fx.get_monthly_row(db, currency, period) if row is not None and row.frozen: continue rate = await finance_fx.fetch_rate_from_frankfurter(currency, period) if rate is None: # Keep an existing live_fallback row (unfrozen) if present; # otherwise the rollup falls back to the live platform rate. continue await finance_fx.upsert_monthly_rate( db, currency, period, rate, source="frankfurter", frozen=True, note="frozen at period close", ) # Verification gate: resolve every native currency exactly as the rollup # will (this also persists any live-fallback rows). A non-USD currency that # resolves to the unconverted 1.0 sentinel (rate == 1, basis 'live') would # pollute the closed baseline — refuse the close listing the offenders. unresolved: list[str] = [] for currency in currencies: cur = currency.upper() if cur == "USD": continue res = await finance_fx.resolve_usd_per_unit(db, cur, period) if res.usd_per_unit == Decimal("1") and res.fx_basis == "live": unresolved.append(cur) if unresolved: raise AppException( code="PERIOD_FX_UNRESOLVED", message=( f"Cannot close {period}: no usable FX rate for " f"{', '.join(sorted(unresolved))}. Retry once the FX source " f"is reachable, or pin the rate manually before closing." ), status_code=409, )