"""ReportStore — persists/reads versioned AnalysisReports (KM-644). Mirrors `PostgresReportInputStore`: each call opens its own `AsyncSessionLocal`. Version assignment is serialized per `analysis_id` with a Postgres transaction-level advisory lock so concurrent button presses can't compute the same version number; the `(analysis_id, version)` unique constraint is the backstop. Per decision 4A every generation is a new version, so two near-simultaneous presses legitimately produce V and V — the lock only prevents a duplicate-number race, not double generation. """ from __future__ import annotations import hashlib from sqlalchemy import func, select, text from src.db.postgres.connection import AsyncSessionLocal from src.db.postgres.models import AnalysisReportRow from src.middlewares.logging import get_logger from .schemas import AnalysisReport logger = get_logger("report_store") def _lock_key(analysis_id: str) -> int: """Stable signed 64-bit key for `pg_advisory_xact_lock`. Python's builtin `hash(str)` is randomized per process, so derive a deterministic key from a digest instead. """ digest = hashlib.sha256(analysis_id.encode()).digest() return int.from_bytes(digest[:8], "big", signed=True) def _report_title(report: AnalysisReport) -> str: """Title for the dedorch `reports.title` column — the goal, else a generic label.""" objective = (report.problem_statement.objective or "").strip() return objective[:200] if objective else "Analysis Report" def _row_to_report(row) -> AnalysisReport: """Rebuild a minimal AnalysisReport from the flat dedorch row. dedorch stores markdown only, so structured fields (findings/caveats/…) come back empty; `rendered_markdown` carries the content the FE renders/downloads. """ return AnalysisReport( report_id=row.id, analysis_id=row.analysis_id, version=row.version, generated_at=row.generated_at, rendered_markdown=row.content, ) class ReportStore: """Read/write versioned reports keyed by `analysis_id`.""" async def save(self, report: AnalysisReport) -> AnalysisReport: """Assign the next version under an advisory lock and persist. Mutates and returns `report` with its final `version`. """ async with AsyncSessionLocal() as session: async with session.begin(): await session.execute( text("SELECT pg_advisory_xact_lock(:k)"), {"k": _lock_key(report.analysis_id)}, ) result = await session.execute( select(func.max(AnalysisReportRow.version)).where( AnalysisReportRow.analysis_id == report.analysis_id ) ) report.version = (result.scalar_one_or_none() or 0) + 1 session.add( AnalysisReportRow( id=report.report_id, analysis_id=report.analysis_id, title=_report_title(report), content=report.rendered_markdown or "", generated_at=report.generated_at, version=report.version, ) ) # leaving session.begin() commits, which releases the advisory lock logger.info( "report persisted", analysis_id=report.analysis_id, version=report.version, report_id=report.report_id, ) return report async def list_for_analysis(self, analysis_id: str) -> list[AnalysisReport]: async with AsyncSessionLocal() as session: result = await session.execute( select(AnalysisReportRow) .where(AnalysisReportRow.analysis_id == analysis_id) .order_by(AnalysisReportRow.version.asc()) ) rows = result.scalars().all() return [_row_to_report(row) for row in rows] async def get(self, analysis_id: str, version: int) -> AnalysisReport | None: async with AsyncSessionLocal() as session: result = await session.execute( select(AnalysisReportRow).where( AnalysisReportRow.analysis_id == analysis_id, AnalysisReportRow.version == version, ) ) row = result.scalar_one_or_none() if row is None: return None return _row_to_report(row)