"""services/audit — every state-changing action writes an audit row; bulk mutations carry COMPLETE before-images and undo is implemented as replay-from-audit-row (04-architecture "Audit"; judgment #11 — ONE mechanism, not an audit log plus a separate undo log). Identity convention (matches the AuditLog columns): `actor_user_id` is the EFFECTIVE user (view-as is true impersonation — ADR #033 lineage); `impersonated_by` is the REAL super admin when view-as is active, NULL otherwise. Both identities are therefore always recoverable. Undo semantics (entries.md rebuild delta / T-ENT-44): - only actions in UNDOABLE_ACTIONS replay; - an audit row undoes at most ONCE (the undo itself is audited with entity_type="audit_log" pointing back — that row is the marker); - rows that changed lock state since the original mutation are SKIPPED (never silently rewrite an invoice-attached row), as are locked rows when the undoing actor lacks can_transfer; - deleted rows are re-inserted verbatim (ids preserved); - exact counts are reported: restored / reinserted / skipped_locked / missing. """ from __future__ import annotations import uuid from datetime import date, datetime from decimal import Decimal from sqlalchemy import select from sqlalchemy.orm import Session from app.models import AuditLog, TimeEntry from app.services.errors import ServiceError #: Actions the undo endpoint will replay (each writes full before-images). UNDOABLE_ACTIONS = frozenset( { "entries.update", "entries.delete", "entries.reresolve", "entries.bulk_edit", "entries.bulk_recalculate", "entries.bulk_delete", } ) #: Columns never overwritten by an undo replay. id/tenant/created_at are #: identity; invoice_line_id is THE lock — M5 mutations never touch it, #: and undo must not either (lock changes are M6's invoice actions). _PROTECTED_COLUMNS = frozenset({"id", "tenant_id", "created_at", "invoice_line_id"}) # --------------------------------------------------------------- snapshots def snapshot_entry(entry: TimeEntry) -> dict: """COMPLETE JSON-serializable image of one time_entries row.""" out: dict = {} for column in TimeEntry.__table__.columns: value = getattr(entry, column.name) if isinstance(value, uuid.UUID): value = str(value) elif isinstance(value, datetime | date): value = value.isoformat() elif isinstance(value, Decimal): value = str(value) out[column.name] = value return out def _parse_value(column, value): if value is None: return None python_type = column.type.python_type if python_type is uuid.UUID: return uuid.UUID(str(value)) if python_type is datetime: return datetime.fromisoformat(value) if python_type is date: return date.fromisoformat(value) if python_type is Decimal: return Decimal(str(value)) if python_type is bool: return bool(value) if python_type is int: return int(value) return value def apply_snapshot(entry: TimeEntry, snapshot: dict) -> bool: """Restore every non-protected column from the snapshot. Returns True when anything actually changed (honest counts — T-ADJ-015).""" changed = False for column in TimeEntry.__table__.columns: if column.name in _PROTECTED_COLUMNS or column.name not in snapshot: continue new_value = _parse_value(column, snapshot[column.name]) if getattr(entry, column.name) != new_value: setattr(entry, column.name, new_value) changed = True return changed def entry_from_snapshot(snapshot: dict) -> TimeEntry: """Rebuild a deleted row verbatim (id preserved — the audit trail stays coherent across delete+undo).""" kwargs = {} for column in TimeEntry.__table__.columns: if column.name in snapshot: kwargs[column.name] = _parse_value(column, snapshot[column.name]) return TimeEntry(**kwargs) # --------------------------------------------------------------- writing def write_audit( session: Session, actor, *, action: str, entity_type: str | None, entity_id: uuid.UUID | None, before: dict | None = None, after: dict | None = None, ) -> AuditLog: """Append one audit row (NOT committed — rides the caller's transaction so a failed mutation leaves no orphan audit).""" row = AuditLog( tenant_id=actor.tenant_id, actor_user_id=actor.user_id, impersonated_by=actor.real_user_id if actor.is_viewing_as else None, action=action, entity_type=entity_type, entity_id=entity_id, before=before, after=after, ) session.add(row) return row # --------------------------------------------------------------- undo def _already_undone(session: Session, audit_id: uuid.UUID) -> bool: return ( session.scalar( select(AuditLog.id).where( AuditLog.action == "entries.undo", AuditLog.entity_type == "audit_log", AuditLog.entity_id == audit_id, ) ) is not None ) def undo_entries_action(session: Session, actor, audit_id: uuid.UUID) -> dict: """Replay the before-images of one undoable audit row. One transaction; commits on success.""" if not actor.allowed("can_manage_imports"): raise ServiceError(403, "NOT_ALLOWED", "You do not have permission to undo entry edits.") audit = session.get(AuditLog, audit_id) if audit is None or audit.tenant_id != actor.tenant_id: raise ServiceError(404, "UNDO_NOT_FOUND", "Nothing to undo — audit record not found.") if audit.action not in UNDOABLE_ACTIONS: raise ServiceError( 409, "NOT_UNDOABLE", f"The action {audit.action!r} can't be undone.", "Only entry edits, deletes, and bulk operations are undoable.", ) if audit.actor_user_id != actor.user_id and not actor.allowed("can_transfer"): raise ServiceError( 403, "NOT_YOUR_ACTION", "Only the person who made this change (or an owner) can undo it.", ) if _already_undone(session, audit_id): raise ServiceError( 409, "ALREADY_UNDONE", "This action was already undone.", "An undo replays at most once — re-apply changes by hand if needed.", ) snapshots: list[dict] = (audit.before or {}).get("rows", []) if not snapshots: raise ServiceError( 409, "NO_BEFORE_IMAGE", "This audit record carries no before-image to replay." ) can_touch_locked = actor.allowed("can_transfer") restored = reinserted = skipped_locked = unchanged = 0 current_images: list[dict] = [] for snap in snapshots: entry_id = uuid.UUID(str(snap["id"])) entry = session.get(TimeEntry, entry_id) if entry is None: # deleted since (or BY the undone action) -> re-insert verbatim if snap.get("invoice_line_id") and not can_touch_locked: skipped_locked += 1 continue session.add(entry_from_snapshot(snap)) reinserted += 1 continue current_images.append(snapshot_entry(entry)) snap_line = snap.get("invoice_line_id") cur_line = str(entry.invoice_line_id) if entry.invoice_line_id else None if cur_line != snap_line: # lock state changed since the mutation — never rewrite it skipped_locked += 1 continue if cur_line is not None and not can_touch_locked: skipped_locked += 1 continue if apply_snapshot(entry, snap): restored += 1 else: unchanged += 1 write_audit( session, actor, action="entries.undo", entity_type="audit_log", entity_id=audit_id, before={"rows": current_images}, after={ "undone_action": audit.action, "restored": restored, "reinserted": reinserted, "skipped_locked": skipped_locked, "unchanged": unchanged, }, ) session.commit() return { "undone_action": audit.action, "restored": restored, "reinserted": reinserted, "skipped_locked": skipped_locked, "unchanged": unchanged, }