"""Legacy -> R1 transforms, 05-migration-plan.md step 2 (2.1 - 2.9). Reads ``legacy.*`` (the landed snapshot), writes ``public.*``. Idempotent: each run clears the target tables (dependency-safe list; ``users`` is column-scoped-upserted instead — login-provisioned rows must survive) and rebuilds everything from legacy inside ONE transaction. A failed run rolls back completely; the report row is written outside the transaction either way. App-owned tenants survive the rebuild (ADR #021): the reset is tenant-scoped — rows belonging to a tenant whose ``settings.books_owner == "app"`` are LEFT ALONE (the app is their system of record), and parties are UPSERTED BY SLUG so a party keeps the SAME uuid across runs (an app-owned tenant's party FKs must stay valid across the nightly). With no app-owned tenants (every harness/parity test DB) the reset degenerates to the historical TRUNCATE — behavior unchanged, harness byte-identical. Ordering (05 "order matters"): 2.1 parties (operators, clients w/ name-collision merge, person parties from team_members ONLY deduped by lower(email)) -> the tenant -> users (profiles keyed by id-auth usernames) 2.2 client->operator edges (direct_client engagements; tenant-scoped, so this MUST follow tenant creation) 2.3 engagements + terms (one engagement per team_member row; TWO terms per the kind model — 02 §2; Gary cost 0 + billout 25) + the seeded official_partnership / BW<->PlusROI partnership 2.4 projects (operator chain; income + adjustments) + rate_overrides (05 lists them in step 3; they carry near-verbatim but reference projects, so they land right after — FK ordering) 2.5 import batches (imported_by/imported_at load-bearing) 2.6 time entries (id-carry, raw_*, verbatim stamps, all-CAD currency default, baseline-entry protection) 2.7 invoices + time-kind lines (pinned from/to algorithm; judgment #22 transferred-straggler rule: check -> report -> attach to the EXISTING per-org "Legacy transfers" invoice; never abort) 2.8 CC expenses (near-verbatim; invoice_line_id starts NULL) 2.9 comments (re-keyed authors) Every step logs (source count, written, skipped w/ reasons); the reconciliation accounts for every legacy table AND column (dispositions.py) and the run aborts (rolls back) on unexplained deltas. """ from __future__ import annotations import json import logging import re import unicodedata import uuid from dataclasses import dataclass, field from datetime import UTC, date, datetime from decimal import Decimal from pathlib import Path from typing import Any from sqlalchemy import Engine, create_engine, text from sqlalchemy.engine import Connection from app.services.migrate.dispositions import verify_columns from app.services.migrate.report import RunReport log = logging.getLogger("with.migrate") #: Committed all-CAD default (07 #21) — the `_usd` suffix was a misnomer. CURRENCY = "CAD" #: profiles.email (lowercased) -> id-auth username (05 step 2.11; accounts #: created at Day-0 so usernames are fixed before the first rehearsal). IDENTITY_MAP = { "rian@rian.ca": "rian", "info@adipramono.com": "adi", } #: The tenant's org party is matched to the same-named operator party #: ("one party wearing two hats"); the partnership seed looks for this #: exact counterparty name (05 step 2.3). PARTNER_PARTY_NAME = "PlusROI" #: Rebuilt-from-legacy target tables, cleared together each run (one #: TRUNCATE when no app-owned tenant exists; tenant-scoped DELETEs in #: this same child-before-parent order otherwise — see _reset_targets). #: users is NOT here (upsert — login-provisioned rows survive); #: audit_log IS here (judgment #23: rehearsal-native state is discarded #: on rebuild — but an app-owned tenant's audit trail is REAL history #: and is preserved by the scoped path); migration_runs is the log and #: is never touched; parties are UPSERTED by slug, never bulk-deleted #: (ADR #021 — stable uuids keep app-owned tenants' party FKs valid). TRUNCATE_TABLES = ( "cc_auto_rules", "cc_highlight_keywords", "cc_expense_lines", "cc_expense_batches", "comments", "time_entries", # project_milestones (R3, alembic 0008) references projects AND invoices, # both truncated below — so it MUST be truncated too, or TRUNCATE of its # parents fails ("referenced in a foreign key constraint"). Same reason # tenant_vendors (references tenants) is here. "project_milestones", "invoice_lines", "invoices", # PlusROI books (R2): the sheet step rebuilds these each run too, so # they join the TRUNCATE-and-rebuild set. project_billing/expenses # reference projects, client_profiles references tenants/parties — all # truncated in the same statement (FK-safe). "expenses", "client_profiles", "project_billing", "import_batches", "rate_overrides", "compensation_terms", "projects", "engagements", "audit_log", "tenant_vendors", "tenants", ) #: Tenant-scoped clear (ADR #021): how each rebuilt table's rows map to #: a tenant when app-owned tenants must survive. Same order as #: TRUNCATE_TABLES (children before parents — FK-safe DELETEs). Tables #: without their own tenant_id scope through their parent. _SCOPED_DELETES = ( ("cc_auto_rules", "tenant_id"), ("cc_highlight_keywords", "tenant_id"), ( "cc_expense_lines", "batch_id IN (SELECT id FROM cc_expense_batches b WHERE {gone:b.tenant_id})", ), ("cc_expense_batches", "tenant_id"), ("comments", "tenant_id"), ("time_entries", "tenant_id"), # project_milestones (R3, alembic 0008) has NO tenant_id — it scopes # THROUGH its project, EXACTLY like project_billing. Cleared before # projects/invoices (children-before-parents): its project FK CASCADEs, # and its generated_invoice_id -> invoices is SET NULL, so either order # is FK-safe, but the explicit scoped DELETE (app-owned rows survive) # must see the projects rows still present to read their tenant. ( "project_milestones", "project_id IN (SELECT id FROM projects p WHERE {gone:p.tenant_id})", ), ( "invoice_lines", "invoice_id IN (SELECT id FROM invoices i WHERE {gone:i.tenant_id})", ), ("invoices", "tenant_id"), ("expenses", "tenant_id"), ("client_profiles", "tenant_id"), ( "project_billing", "project_id IN (SELECT id FROM projects p WHERE {gone:p.tenant_id})", ), ("import_batches", "tenant_id"), ("rate_overrides", "tenant_id"), ( "compensation_terms", "engagement_id IN (SELECT id FROM engagements e WHERE {gone:e.tenant_id})", ), ("projects", "tenant_id"), ("engagements", "tenant_id"), ("audit_log", "tenant_id"), # tenant-less audit rows are still discarded # tenant_vendors (R3, alembic 0008) has its OWN tenant_id — cleared # before tenants (its FK parent). The vendor party FK is to the never- # truncated parties table, so only the tenant scope matters here. ("tenant_vendors", "tenant_id"), ("tenants", "id"), ) def _reset_targets(conn: Connection) -> list[uuid.UUID]: """Clear the rebuilt tables, PRESERVING app-owned tenants' rows. No app-owned tenant (every harness/parity scratch DB, and production before the PlusROI flip) → the exact historical TRUNCATE, so the fast path — and the harness — are byte-for-byte unchanged. With app-owned tenants, tenant-scoped DELETEs in child-before-parent order: a row is deleted iff its tenant is NOT app-owned (rows with a NULL tenant, e.g. tenant-less audit entries, are discarded exactly as the TRUNCATE did). Returns the preserved tenant ids. """ app_owned = [ row[0] for row in conn.execute( text("SELECT id FROM tenants WHERE settings->>'books_owner' = 'app'") ).fetchall() ] if not app_owned: conn.execute(text("TRUNCATE TABLE " + ", ".join(TRUNCATE_TABLES))) return [] conn.execute( text( "CREATE TEMP TABLE tmp_keep_tenants (tenant_id uuid PRIMARY KEY)" " ON COMMIT DROP" ) ) for tenant_id in app_owned: conn.execute( text("INSERT INTO tmp_keep_tenants VALUES (:t)"), {"t": tenant_id} ) def gone(column: str) -> str: # "this row's tenant is not one of the preserved (app-owned) # ones". `column` MUST be table-qualified: an unqualified # tenant_id inside the subquery would resolve to k.tenant_id # (inner scope wins) and turn the predicate into a no-op. return ( f"({column} IS NULL OR NOT EXISTS (SELECT 1 FROM tmp_keep_tenants k" f" WHERE k.tenant_id = {column}))" ) for table, scope_ in _SCOPED_DELETES: if "{gone:" in scope_: head, rest = scope_.split("{gone:", 1) column, tail = rest.split("}", 1) predicate = head + gone(column) + tail else: predicate = gone(f"{table}.{scope_}") conn.execute(text(f"DELETE FROM {table} WHERE {predicate}")) # noqa: S608 return app_owned LEGACY_TABLES = ( "organizations", "profiles", "organization_members", "teams", "team_members", "clockify_imports", "time_entries", "comments", "operators", "clients", "projects", "pending_imports", "project_rate_overrides", "invoices", "cc_expense_batches", "cc_expense_lines", "cc_highlight_keywords", "cc_auto_rules", "users", ) BASELINE_PREDICATE = ( "start_at::date = DATE '2025-12-31'" " AND description LIKE 'Transfer baseline from Toggl%'" ) class MigrationError(Exception): """Hard transform failure — aborts (rolls back) the run.""" @dataclass class Ctx: report: RunReport identity_map: dict[str, str] migration_date: date # PlusROI sheet-books source dir (R2). None = the sheet step is not # invoked at all (the default for every harness/parity/reporting test # that runs run_all(engine=...)); "auto" = resolve the app-relative # books-import/source; a path = use it. See sheet_books.step_sheet_books. books_source: str | Path | None = None tenant_by_org: dict[uuid.UUID, uuid.UUID] = field(default_factory=dict) tenant_party_by_org: dict[uuid.UUID, uuid.UUID] = field(default_factory=dict) party_by_operator: dict[uuid.UUID, uuid.UUID] = field(default_factory=dict) operator_party_by_name: dict[str, uuid.UUID] = field(default_factory=dict) party_by_client: dict[uuid.UUID, uuid.UUID] = field(default_factory=dict) client_merged: dict[uuid.UUID, bool] = field(default_factory=dict) person_party_by_email: dict[str, uuid.UUID] = field(default_factory=dict) member_party: dict[uuid.UUID, uuid.UUID] = field(default_factory=dict) member_engagement: dict[uuid.UUID, uuid.UUID] = field(default_factory=dict) member_email: dict[uuid.UUID, str] = field(default_factory=dict) user_by_legacy_uid: dict[uuid.UUID, uuid.UUID] = field(default_factory=dict) slugs: set[str] = field(default_factory=set) # reconciliation accumulators: table -> disposition -> count dispositions: dict[str, dict[str, int]] = field(default_factory=dict) def account(self, table: str, disposition: str, count: int = 1) -> None: table_map = self.dispositions.setdefault(table, {}) table_map[disposition] = table_map.get(disposition, 0) + count # --- small helpers -------------------------------------------------------- def slugify(name: str) -> str: ascii_name = unicodedata.normalize("NFKD", name).encode("ascii", "ignore").decode() slug = re.sub(r"[^a-z0-9]+", "-", ascii_name.lower()).strip("-") return slug or "party" def _allocate_slug(ctx: Ctx, name: str) -> str: base = slugify(name) slug, n = base, 1 while slug in ctx.slugs: n += 1 slug = f"{base}-{n}" ctx.slugs.add(slug) return slug def _insert_party( conn: Connection, ctx: Ctx, *, kind: str, name: str, email: str | None = None, notes: str | None = None, is_active: bool = True, ) -> uuid.UUID: party_id = uuid.uuid4() conn.execute( text( "INSERT INTO parties (id, kind, name, slug, email, notes, is_active)" " VALUES (:id, :kind, :name, :slug, :email, :notes, :is_active)" ), { "id": party_id, "kind": kind, "name": name, "slug": _allocate_slug(ctx, name), "email": email, "notes": notes, "is_active": is_active, }, ) _ensure_subtype_row(conn, kind=kind, party_id=party_id) return party_id def _ensure_subtype_row(conn: Connection, *, kind: str, party_id: uuid.UUID) -> None: """Keep the 1:1 subtype detail row in step (ADR #012/#013) — the raw-SQL mirror of the ORM choke point in services.entities.parties. Idempotent (ON CONFLICT DO NOTHING) so the upsert path re-asserts the invariant.""" table = "organizations" if kind == "org" else "people" conn.execute( text( f"INSERT INTO {table} (party_id) VALUES (:id)" # noqa: S608 " ON CONFLICT (party_id) DO NOTHING" ), {"id": party_id}, ) def _upsert_party( conn: Connection, ctx: Ctx, db_slugs: dict[str, tuple[uuid.UUID, str]], *, kind: str, name: str, email: str | None = None, notes: str | None = None, is_active: bool = True, ) -> uuid.UUID: """STABLE party identity (ADR #021): reuse-by-slug instead of insert. The slug is DETERMINISTIC: ``slugify(name)`` plus a ``-2``/``-3``… suffix allocated in the caller's iteration order — and every caller iterates its legacy rows in a stable ``ORDER BY created_at, id`` — so the same legacy row lands on the same slug run after run. If a party with that slug (and kind) already exists it keeps its uuid (name/email/notes/is_active refreshed from legacy, the same values a fresh build would write); otherwise a new uuid is inserted. A slug held by a party of the OTHER kind is treated like an in-run collision (suffix bumps) — identities never change kind. ``db_slugs`` is the slug -> (id, kind) snapshot taken at step start; ``ctx.slugs`` tracks slugs claimed by THIS run so two same-named legacy parties keep distinct identities (in iteration order). """ base = slugify(name) slug, n = base, 1 while True: if slug not in ctx.slugs: hit = db_slugs.get(slug) if hit is None or hit[1] == kind: break n += 1 slug = f"{base}-{n}" ctx.slugs.add(slug) if hit is None: party_id = uuid.uuid4() conn.execute( text( "INSERT INTO parties (id, kind, name, slug, email, notes," " is_active) VALUES (:id, :kind, :name, :slug, :email, :notes," " :is_active)" ), { "id": party_id, "kind": kind, "name": name, "slug": slug, "email": email, "notes": notes, "is_active": is_active, }, ) else: party_id = hit[0] conn.execute( text( "UPDATE parties SET name = :name, email = :email, notes = :notes," " is_active = :is_active WHERE id = :id" ), { "id": party_id, "name": name, "email": email, "notes": notes, "is_active": is_active, }, ) _ensure_subtype_row(conn, kind=kind, party_id=party_id) return party_id def _insert_engagement( conn: Connection, ctx: Ctx, *, tenant_id: uuid.UUID, type_: str, party_a: uuid.UUID, party_b: uuid.UUID, role_a: str, role_b: str, ended_on: date | None = None, can_manage_imports: bool = False, consolidate_as: str | None = None, notes: str | None = None, ) -> uuid.UUID: engagement_id = uuid.uuid4() conn.execute( text( "INSERT INTO engagements (id, tenant_id, type, party_a_id, party_b_id," " role_a, role_b, ended_on, can_manage_imports, consolidate_as, notes)" " VALUES (:id, :tenant_id, :type, :a, :b, :ra, :rb, :ended_on, :cmi," " :consolidate_as, :notes)" ), { "id": engagement_id, "tenant_id": tenant_id, "type": type_, "a": party_a, "b": party_b, "ra": role_a, "rb": role_b, "ended_on": ended_on, "cmi": can_manage_imports, "consolidate_as": consolidate_as, "notes": notes, }, ) return engagement_id def _count(conn: Connection, sql: str, params: dict | None = None) -> int: return conn.execute(text(sql), params or {}).scalar_one() # --- preflight ------------------------------------------------------------ def preflight(conn: Connection, ctx: Ctx) -> None: step = ctx.report.step("preflight") if not conn.execute( text( "SELECT EXISTS (SELECT 1 FROM information_schema.tables" " WHERE table_schema = 'legacy' AND table_name = 'time_entries')" ) ).scalar_one(): raise MigrationError( "no legacy schema in this database — run the snapshot pipeline" " (scripts/nightly_snapshot.py) or scripts/sync_legacy_host.py first" ) # Column-disposition enumeration (05 step 2.10): every legacy column # must be mentioned; unknown tables/columns fail the run. rows = conn.execute( text( "SELECT table_name, column_name FROM information_schema.columns" " WHERE table_schema = 'legacy' ORDER BY table_name, ordinal_position" ) ).fetchall() legacy_columns: dict[str, list[str]] = {} for table, column in rows: legacy_columns.setdefault(table, []).append(column) unmapped, stale, unknown_tables = verify_columns(legacy_columns) ctx.report.column_check = { "tables": len(legacy_columns), "unmapped": unmapped, "stale": stale, "unknown_tables": unknown_tables, } if unknown_tables: ctx.report.errors.append( f"legacy tables with no disposition (not mentioned anywhere): {unknown_tables}" ) if unmapped: ctx.report.errors.append( f"legacy columns with no disposition (not mentioned anywhere): {unmapped}" ) for table, columns in stale.items(): step.warn(f"stale column dispositions for legacy.{table}: {columns}") # T-MIG-013 guard: stamps must be copyable VERBATIM into Numeric(12,2). # v7 re-stamped everything to 2dp; if that ever stops being true the # copy would silently round — fail instead. overscale = _count( conn, "SELECT count(*) FROM legacy.time_entries" " WHERE (billout_cost_usd IS NOT NULL" " AND billout_cost_usd <> round(billout_cost_usd, 2))" " OR (billout_amount_usd IS NOT NULL" " AND billout_amount_usd <> round(billout_amount_usd, 2))", ) if overscale: raise MigrationError( f"{overscale} legacy money stamp(s) have >2 decimal places —" " verbatim copy into Numeric(12,2) would round (T-MIG-013)" ) # Rebuild delta #4: text-cache vs FK-chain agreement. Drift is a data # bug to fix in the OLD app before cutover; rehearsals report it. drift = _count( conn, "SELECT count(*) FROM legacy.time_entries te" " JOIN legacy.projects p ON p.id = te.project_id" " JOIN legacy.clients c ON c.id = p.client_id" " JOIN legacy.operators o ON o.id = c.operator_id" " WHERE lower(coalesce(te.project, '')) IS DISTINCT FROM lower(coalesce(p.name, ''))" " OR lower(coalesce(te.client, '')) IS DISTINCT FROM lower(coalesce(c.name, ''))" " OR lower(coalesce(te.operator, '')) IS DISTINCT FROM lower(coalesce(o.name, ''))", ) step.details["text_cache_drift_rows"] = drift if drift: message = ( f"{drift} resolved entr(y/ies) where the text triple disagrees with the" " FK entity chain (rebuild delta #4) — harmless here (new side reads" " FKs + raw_*), but a data bug to fix old-side before cutover" ) step.warn(message) ctx.report.warnings.append(message) # --- 2.1 parties, tenant, users -------------------------------------------- def step_parties_tenant_users(conn: Connection, ctx: Ctx) -> None: step = ctx.report.step("2.1 parties -> tenant -> users") # Reset target state — tenant-scoped (ADR #021): app-owned tenants' # rows survive; everything else clears exactly as before. users # survives (upsert below) and KEEPS person_party_id: parties are no # longer bulk-deleted, so there is no FK to sidestep — and an # app-owned tenant's partner logins (rob/danielle) must keep their # person-party links even though the sheet step no longer runs for it. preserved_tenants = _reset_targets(conn) if preserved_tenants: step.details["app_owned_tenants_preserved"] = [ str(t) for t in preserved_tenants ] # Stable-identity snapshot: slug -> (party id, kind) as of the last # run. Parties matched here KEEP their uuid; legacy-gone parties are # left in place untouched (their history may still be referenced by # an app-owned tenant) — acceptable lingering, reported nowhere # because parties are inventory, not a reconciled legacy table. db_slugs: dict[str, tuple[uuid.UUID, str]] = { slug: (pid, kind) for pid, kind, slug in conn.execute( text("SELECT id, kind, slug FROM parties") ).fetchall() } # Operators -> org parties. operators = conn.execute( text( "SELECT id, org_id, name, notes FROM legacy.operators" " ORDER BY created_at, id" ) ).fetchall() for op_id, _org_id, name, notes in operators: key = name.strip().lower() if key in ctx.operator_party_by_name: raise MigrationError( f"duplicate operator name {name!r} across orgs — parties are" " keyed by name; disambiguate before migration" ) party_id = _upsert_party( conn, ctx, db_slugs, kind="org", name=name.strip(), notes=notes ) ctx.operator_party_by_name[key] = party_id ctx.party_by_operator[op_id] = party_id ctx.account("operators", "party_created") # Clients -> org parties, with the pinned merge rule: a client whose # lower(name) collides with an OPERATOR name merges into that party # (one party wearing two hats); same-named clients under different # operators stay separate parties (05 step 2.1). clients = conn.execute( text( "SELECT id, org_id, operator_id, name, notes FROM legacy.clients" " ORDER BY created_at, id" ) ).fetchall() for client_id, _org_id, _operator_id, name, notes in clients: key = name.strip().lower() merged = ctx.operator_party_by_name.get(key) if merged is not None: ctx.party_by_client[client_id] = merged ctx.client_merged[client_id] = True ctx.account("clients", "party_merged_into_operator") else: ctx.party_by_client[client_id] = _upsert_party( conn, ctx, db_slugs, kind="org", name=name.strip(), notes=notes ) ctx.client_merged[client_id] = False ctx.account("clients", "party_created") # Person parties from team_members ONLY, deduped by lower(email) # (unmatched source_user_emails create NOTHING — pinned rule). members = conn.execute( text( "SELECT id, email, display_name FROM legacy.team_members" " ORDER BY created_at, id" ) ).fetchall() person_names: dict[str, str] = {} for member_id, email, display_name in members: key = email.strip().lower() ctx.member_email[member_id] = key if key not in ctx.person_party_by_email: ctx.person_party_by_email[key] = _upsert_party( conn, ctx, db_slugs, kind="person", name=display_name, email=key ) person_names[key] = display_name ctx.account("team_members", "person_party_created") else: ctx.account("team_members", "person_party_deduped") if person_names[key] != display_name: step.warn( f"deduped person {key} has divergent display names" f" ({person_names[key]!r} kept, {display_name!r} dropped)" ) ctx.member_party[member_id] = ctx.person_party_by_email[key] # THEN the tenant(s): the org party merges with a same-named operator # party (Bowden Works is both the tenant and an operator). orgs = conn.execute( text("SELECT id, name, slug FROM legacy.organizations ORDER BY created_at, id") ).fetchall() for org_id, name, _slug in orgs: key = name.strip().lower() party_id = ctx.operator_party_by_name.get(key) if party_id is None: party_id = _upsert_party(conn, ctx, db_slugs, kind="org", name=name.strip()) ctx.operator_party_by_name[key] = party_id tenant_id = uuid.uuid4() conn.execute( text( "INSERT INTO tenants (id, party_id, settings)" " VALUES (:id, :party_id, cast(:settings AS jsonb))" ), { "id": tenant_id, "party_id": party_id, "settings": json.dumps({"default_currency": CURRENCY}), }, ) ctx.tenant_by_org[org_id] = tenant_id ctx.tenant_party_by_org[org_id] = party_id ctx.account("organizations", "tenant_created") # Users: column-scoped upsert keyed on idauth_username — profiles map # via the identity map; login-provisioned rows keep their preferences. profiles = conn.execute( text("SELECT id, email, is_super_admin FROM legacy.profiles ORDER BY email") ).fetchall() for profile_id, email, is_super_admin in profiles: key = (email or "").strip().lower() username = ctx.identity_map.get(key) if username is None: ctx.account("profiles", "skipped_no_idauth_mapping") step.warn( f"profile {key or profile_id} has no id-auth username mapping —" " no users row; its *_by references will migrate as NULL" ) continue person_party = ctx.person_party_by_email.get(key) if person_party is None: step.warn( f"user {username} ({key}) has no person party (not a team member)" " — person_party_id stays NULL" ) user_id = conn.execute( text( "INSERT INTO users (id, idauth_username, email, person_party_id," " is_super_admin, preferences)" " VALUES (:id, :username, :email, :party, :admin, cast('{}' AS jsonb))" " ON CONFLICT (idauth_username) DO UPDATE" " SET is_super_admin = excluded.is_super_admin," " person_party_id = excluded.person_party_id" " RETURNING id" ), { "id": uuid.uuid4(), "username": username, "email": key, "party": person_party, "admin": is_super_admin, }, ).scalar_one() ctx.user_by_legacy_uid[profile_id] = user_id ctx.account("profiles", "user_upserted") # auth.users: identity mapping only. auth_users = _count(conn, "SELECT count(*) FROM legacy.users") ctx.account("users", "identity_map_only", auth_users) # Temp maps for the SQL-side bulk steps. conn.execute( text( "CREATE TEMP TABLE tmp_org_tenant" " (org_id uuid PRIMARY KEY, tenant_id uuid NOT NULL) ON COMMIT DROP" ) ) for org_id, tenant_id in ctx.tenant_by_org.items(): conn.execute( text("INSERT INTO tmp_org_tenant VALUES (:org_id, :tenant_id)"), {"org_id": org_id, "tenant_id": tenant_id}, ) conn.execute( text( "CREATE TEMP TABLE tmp_user_map" " (legacy_uid uuid PRIMARY KEY, user_id uuid NOT NULL) ON COMMIT DROP" ) ) for legacy_uid, user_id in ctx.user_by_legacy_uid.items(): conn.execute( text("INSERT INTO tmp_user_map VALUES (:legacy_uid, :user_id)"), {"legacy_uid": legacy_uid, "user_id": user_id}, ) step.source_count = len(operators) + len(clients) + len(members) + len(orgs) + len(profiles) step.written = _count(conn, "SELECT count(*) FROM parties") + len(orgs) step.details = { "operator_parties": len(operators), "client_parties_created": sum(1 for m in ctx.client_merged.values() if not m), "client_parties_merged_into_operator": sum( 1 for m in ctx.client_merged.values() if m ), "person_parties": len(ctx.person_party_by_email), "tenants": len(orgs), "users_upserted": len(ctx.user_by_legacy_uid), } # --- 2.2 client -> operator edges ------------------------------------------- def step_client_edges(conn: Connection, ctx: Ctx) -> None: step = ctx.report.step("2.2 direct_client engagements") clients = conn.execute( text( "SELECT id, org_id, operator_id, name FROM legacy.clients" " ORDER BY created_at, id" ) ).fetchall() step.source_count = len(clients) for client_id, org_id, operator_id, _name in clients: _insert_engagement( conn, ctx, tenant_id=ctx.tenant_by_org[org_id], type_="direct_client", party_a=ctx.party_by_operator[operator_id], party_b=ctx.party_by_client[client_id], role_a="operator", role_b="client", ) disposition = ( "direct_client_edge_merged_party" if ctx.client_merged[client_id] else "direct_client_edge" ) ctx.account("clients", disposition) step.written += 1 # --- 2.3 engagements + terms ------------------------------------------------- def step_engagements_terms(conn: Connection, ctx: Ctx) -> None: step = ctx.report.step("2.3 worker engagements + compensation terms") members = conn.execute( text( "SELECT id, org_id, email, cost_rate_usd, billout_rate_usd," " consolidate_as, is_active, notes" " FROM legacy.team_members ORDER BY created_at, id" ) ).fetchall() step.source_count = len(members) # canManageImports parity: owner/manager org members grant the person's # worker engagements the capability (02 §4 parity mapping). manager_emails = { row[0].strip().lower() for row in conn.execute( text( "SELECT p.email FROM legacy.organization_members om" " JOIN legacy.profiles p ON p.id = om.user_id" " WHERE om.role IN ('owner', 'manager')" ) ).fetchall() if row[0] } first_entry_dates = dict( conn.execute( text( "SELECT team_member_id, min(start_at)::date FROM legacy.time_entries" " WHERE team_member_id IS NOT NULL GROUP BY 1" ) ).fetchall() ) terms_written = 0 for row in members: (member_id, org_id, email, cost_rate, billout_rate, consolidate_as, is_active, notes) = row key = email.strip().lower() engagement_id = _insert_engagement( conn, ctx, tenant_id=ctx.tenant_by_org[org_id], type_="subcontract", party_a=ctx.tenant_party_by_org[org_id], party_b=ctx.member_party[member_id], role_a="prime", role_b="subcontractor", ended_on=None if is_active else ctx.migration_date, can_manage_imports=key in manager_emails, consolidate_as=consolidate_as, notes=notes, ) ctx.member_engagement[member_id] = engagement_id ctx.account("team_members", "engagement_created") # Only current rate values exist -> effective_from = min(entry date) # (fallback: migration date). Historical stamps are copied verbatim # in 2.6 — the backdated effective_from is NEVER license to # recompute history (recalc guard, 05 step 2.3). effective_from = first_entry_dates.get(member_id) or ctx.migration_date for kind, rate in (("cost", cost_rate), ("billout", billout_rate)): if rate is None: # NULL = never priced -> no term (judgment #19); 0 is a real # zero-rate term (Gary: cost 0 + billout 25). step.skip(f"{kind}_rate_null_no_term") continue conn.execute( text( "INSERT INTO compensation_terms (id, engagement_id, kind, model," " rate_amount, currency, effective_from)" " VALUES (:id, :engagement_id, :kind, 'hourly', :rate, :currency," " :effective_from)" ), { "id": uuid.uuid4(), "engagement_id": engagement_id, "kind": kind, "rate": rate, "currency": CURRENCY, "effective_from": effective_from, }, ) terms_written += 1 step.written += 1 # Seeded grants from organization_members (02 §4 parity mapping). org_members = conn.execute( text( "SELECT om.organization_id, om.role, p.email" " FROM legacy.organization_members om" " JOIN legacy.profiles p ON p.id = om.user_id ORDER BY p.email" ) ).fetchall() for org_id, role, email in org_members: key = (email or "").strip().lower() person_party = ctx.person_party_by_email.get(key) if role == "owner" and person_party is not None: _insert_engagement( conn, ctx, tenant_id=ctx.tenant_by_org[org_id], type_="official_partnership", party_a=ctx.tenant_party_by_org[org_id], party_b=person_party, role_a="owner", role_b="owner", ) ctx.account("organization_members", "official_partnership_seeded") elif role in ("owner", "manager"): # capability already applied to the worker engagements above ctx.account("organization_members", "import_capability_carried") else: ctx.account("organization_members", "worker_grant_via_engagement") if role == "owner" and person_party is None: step.warn(f"org owner {key} has no person party — partnership not seeded") # The BW<->PlusROI partnership engagement (no terms yet — R4's job). partner = ctx.operator_party_by_name.get(PARTNER_PARTY_NAME.lower()) for org_id, tenant_party in ctx.tenant_party_by_org.items(): if partner is not None and partner != tenant_party: _insert_engagement( conn, ctx, tenant_id=ctx.tenant_by_org[org_id], type_="partnership", party_a=tenant_party, party_b=partner, role_a="partner", role_b="partner", ) step.details["partnership_seeded"] = f"tenant <-> {PARTNER_PARTY_NAME}" else: step.warn( f"no {PARTNER_PARTY_NAME!r} party distinct from the tenant —" " partnership engagement not seeded" ) step.details["terms_written"] = terms_written # --- 2.4 projects + rate overrides ------------------------------------------- def step_projects(conn: Connection, ctx: Ctx) -> None: step = ctx.report.step("2.4 projects (+ rate overrides)") projects = conn.execute( text( "SELECT p.id, p.org_id, p.client_id, c.operator_id, p.name, p.income_usd," " p.notes, p.billout_adjustment_pct, p.billout_adjustment_amount," " p.created_at" " FROM legacy.projects p JOIN legacy.clients c ON c.id = p.client_id" " ORDER BY p.created_at, p.id" ) ).fetchall() step.source_count = len(projects) seen: dict[tuple[uuid.UUID, uuid.UUID, str], str] = {} for ( project_id, org_id, client_id, operator_id, name, income, notes, adj_pct, adj_amount, created_at, ) in projects: uniq = (ctx.tenant_by_org[org_id], ctx.party_by_client[client_id], name.lower()) if uniq in seen: raise MigrationError( f"project name collision after party merge: {name!r} under the same" f" (tenant, client party) as {seen[uniq]!r} — owner must rename" " one legacy project before migration (judgment #33 uniqueness)" ) seen[uniq] = name conn.execute( text( "INSERT INTO projects (id, tenant_id, operator_party_id," " client_party_id, name, status, income, income_currency," " billout_adjustment_pct, billout_adjustment_amount, notes, created_at)" " VALUES (:id, :tenant_id, :operator_party, :client_party, :name," " 'active', :income, :income_currency, :adj_pct, :adj_amount, :notes," " :created_at)" ), { "id": project_id, # project ids carry unchanged (harness key) "tenant_id": ctx.tenant_by_org[org_id], "operator_party": ctx.party_by_operator[operator_id], "client_party": ctx.party_by_client[client_id], "name": name, "income": income, "income_currency": CURRENCY if income is not None else None, "adj_pct": adj_pct, "adj_amount": adj_amount, "notes": notes, "created_at": created_at, }, ) ctx.account("projects", "migrated") step.written += 1 # rate_overrides — near-verbatim (05 step 2.3 / judgment #25); placed # after projects because of the FK. XOR + precedence semantics carry. overrides = conn.execute( text( "SELECT id, org_id, project_id, team_member_id, override_rate_usd," " override_pct, notes, created_at FROM legacy.project_rate_overrides" " ORDER BY created_at, id" ) ).fetchall() seen_override: dict[tuple[uuid.UUID, uuid.UUID | None], uuid.UUID] = {} for ( override_id, org_id, project_id, member_id, rate, pct, notes, created_at, ) in overrides: worker_party = ctx.member_party.get(member_id) if member_id else None if member_id and worker_party is None: raise MigrationError( f"rate override {override_id} references unknown team member {member_id}" ) uniq_key = (project_id, worker_party) if uniq_key in seen_override: raise MigrationError( f"rate overrides {seen_override[uniq_key]} and {override_id} collapse" " to the same (project, worker party) after the person-party dedupe" " — owner must resolve which rule wins" ) seen_override[uniq_key] = override_id conn.execute( text( "INSERT INTO rate_overrides (id, tenant_id, project_id," " worker_party_id, override_rate, currency, override_pct, notes," " created_at)" " VALUES (:id, :tenant_id, :project_id, :worker_party, :rate," " :currency, :pct, :notes, :created_at)" ), { "id": override_id, "tenant_id": ctx.tenant_by_org[org_id], "project_id": project_id, "worker_party": worker_party, "rate": rate, "currency": CURRENCY if rate is not None else None, "pct": pct, "notes": notes, "created_at": created_at, }, ) ctx.account("project_rate_overrides", "migrated") step.details["rate_overrides"] = len(overrides) # --- 2.5 import batches -------------------------------------------------------- def step_import_batches(conn: Connection, ctx: Ctx) -> None: step = ctx.report.step("2.5 import batches") step.source_count = _count(conn, "SELECT count(*) FROM legacy.clockify_imports") result = conn.execute( text( "INSERT INTO import_batches (id, tenant_id, source, filename, name," " notes, row_count, imported_by, imported_at)" " SELECT ci.id, t.tenant_id, ci.source, ci.filename, ci.name, ci.notes," " ci.row_count, um.user_id, ci.imported_at" " FROM legacy.clockify_imports ci" " JOIN tmp_org_tenant t ON t.org_id = ci.org_id" " LEFT JOIN tmp_user_map um ON um.legacy_uid = ci.imported_by" ) ) step.written = result.rowcount ctx.account("clockify_imports", "import_batch_created", result.rowcount) unmapped = _count( conn, "SELECT count(*) FROM legacy.clockify_imports ci" " LEFT JOIN tmp_user_map um ON um.legacy_uid = ci.imported_by" " WHERE ci.imported_by IS NOT NULL AND um.user_id IS NULL", ) if unmapped: step.warn( f"{unmapped} batch(es) have an imported_by with no users mapping —" " carried as NULL (importer-scoped visibility loses those batches)" ) pending = _count(conn, "SELECT count(*) FROM legacy.pending_imports") ctx.account("pending_imports", "dropped_staging", pending) if pending: step.warn( f"{pending} pending (staged) import row(s) dropped — anything" " unresolved at cutover gets re-uploaded (05 step 2.5)" ) # --- 2.6 time entries ----------------------------------------------------------- def step_time_entries(conn: Connection, ctx: Ctx) -> None: step = ctx.report.step("2.6 time entries") step.source_count = _count(conn, "SELECT count(*) FROM legacy.time_entries") conn.execute( text( "CREATE TEMP TABLE tmp_member_map (member_id uuid PRIMARY KEY," " party_id uuid NOT NULL, engagement_id uuid NOT NULL) ON COMMIT DROP" ) ) for member_id, party_id in ctx.member_party.items(): conn.execute( text("INSERT INTO tmp_member_map VALUES (:m, :p, :e)"), {"m": member_id, "p": party_id, "e": ctx.member_engagement[member_id]}, ) result = conn.execute( text( "INSERT INTO time_entries (id, tenant_id, project_id, worker_party_id," " via_engagement_id, source, source_timezone, import_id," " source_user_name, source_user_email, raw_operator, raw_client," " raw_project, description, billable, start_at, end_at," " duration_seconds, source_rate, source_amount, cost_amount," " cost_currency, billout_amount, billout_currency, created_at)" " SELECT te.id, t.tenant_id, te.project_id, mm.party_id," " mm.engagement_id," # source: from the batch (toggl/backfill/clockify carried # VERBATIM); no batch = manual entry (02 §3) " CASE WHEN te.import_id IS NULL THEN 'manual'" " ELSE coalesce(nullif(trim(ci.source), ''), 'clockify') END," # source_timezone: unrecoverable for historical rows (07 #17) " NULL," " te.import_id, te.source_user_name, te.source_user_email," " te.operator, te.client, te.project," # the raw_* payload, verbatim " te.description, te.billable, te.start_at, te.end_at," " te.duration_seconds, te.source_rate_usd, te.source_amount_usd," # stamps copied VERBATIM, never recomputed (T-MIG-013); the # currency is part of the stamp — all-CAD default (07 #21) " te.billout_cost_usd," " CASE WHEN te.billout_cost_usd IS NULL THEN NULL ELSE :currency END," " te.billout_amount_usd," " CASE WHEN te.billout_amount_usd IS NULL THEN NULL ELSE :currency END," " te.created_at" " FROM legacy.time_entries te" " JOIN tmp_org_tenant t ON t.org_id = te.org_id" " LEFT JOIN tmp_member_map mm ON mm.member_id = te.team_member_id" " LEFT JOIN legacy.clockify_imports ci ON ci.id = te.import_id" ), {"currency": CURRENCY}, ) step.written = result.rowcount ctx.account("time_entries", "migrated", result.rowcount) if step.written != step.source_count: raise MigrationError( f"time_entries row-count mismatch: {step.source_count} legacy," f" {step.written} written" ) defaulted = _count( conn, "SELECT count(*) FROM legacy.time_entries te" " JOIN legacy.clockify_imports ci ON ci.id = te.import_id" " WHERE nullif(trim(ci.source), '') IS NULL", ) if defaulted: step.warn( f"{defaulted} entr(y/ies) belong to a batch with no source value —" " source defaulted to 'clockify'" ) # Baseline-entry protection (05 step 2.6): the 2025-12-31 sentinel rows # carry the prepaid-block opening balances — they are load-bearing data. legacy_baseline = _count( conn, f"SELECT count(*) FROM legacy.time_entries WHERE {BASELINE_PREDICATE}" ) new_baseline = _count( conn, f"SELECT count(*) FROM time_entries WHERE {BASELINE_PREDICATE}" ) step.details["baseline_entries"] = new_baseline if new_baseline != legacy_baseline: raise MigrationError( f"baseline-entry protection failed: {legacy_baseline} legacy sentinel" f" rows, {new_baseline} migrated (02 §2 prepaid_block)" ) blocked = _count(conn, "SELECT count(*) FROM time_entries WHERE project_id IS NULL") unmatched_workers = _count( conn, "SELECT count(*) FROM time_entries" " WHERE worker_party_id IS NULL AND source_user_email IS NOT NULL", ) step.details["unassigned_blocked"] = blocked step.details["worker_unmatched"] = unmatched_workers # --- 2.7 invoices + lines + the straggler rule ----------------------------------- def _pick_unique(values: set) -> Any | None: return next(iter(values)) if len(values) == 1 else None def step_invoices(conn: Connection, ctx: Ctx) -> None: step = ctx.report.step("2.7 invoices + time-kind lines") invoices = conn.execute( text( "SELECT id, org_id, name, status, operator_id, client_id," " manual_total_usd, notes, created_by, created_at, invoice_date," " sent_at, paid_at FROM legacy.invoices ORDER BY created_at, id" ) ).fetchall() step.source_count = len(invoices) # Operator/client scope derived from attached entries via the FK chain. derived_ops: dict[uuid.UUID, set[uuid.UUID]] = {} derived_clients: dict[uuid.UUID, set[uuid.UUID]] = {} for invoice_id, operator_id, client_id in conn.execute( text( "SELECT te.invoice_id, c.operator_id, p.client_id" " FROM legacy.time_entries te" " JOIN legacy.projects p ON p.id = te.project_id" " JOIN legacy.clients c ON c.id = p.client_id" " WHERE te.invoice_id IS NOT NULL GROUP BY 1, 2, 3" ) ).fetchall(): derived_ops.setdefault(invoice_id, set()).add(operator_id) derived_clients.setdefault(invoice_id, set()).add(client_id) review: list[str] = [] for ( invoice_id, org_id, name, status, operator_chip, client_chip, manual_total, notes, created_by, created_at, invoice_date, sent_at, paid_at, ) in invoices: tenant_party = ctx.tenant_party_by_org[org_id] # PINNED algorithm (05 step 2.7, critique-corrected): from_party is # ALWAYS the tenant party; to_party from the operator scope. op_scope = operator_chip or _pick_unique(derived_ops.get(invoice_id, set())) op_party = ctx.party_by_operator.get(op_scope) if op_scope else None if op_party is not None and op_party != tenant_party: to_party = op_party # a PlusROI-scoped invoice is BW billing PlusROI else: client_scope = client_chip or _pick_unique( derived_clients.get(invoice_id, set()) ) to_party = ctx.party_by_client.get(client_scope) if client_scope else None if to_party is None: review.append(f"{name} ({invoice_id}): to_party unresolved (owner review)") if name.strip().lower() == "legacy transfers": review.append(f"{name} ({invoice_id}): 'Legacy transfers' sentinel") conn.execute( text( "INSERT INTO invoices (id, tenant_id, name, status, currency," " invoice_date, sent_at, paid_at, from_party_id, to_party_id," " manual_total, notes, created_by, created_at)" " VALUES (:id, :tenant_id, :name, :status, :currency, :invoice_date," " :sent_at, :paid_at, :from_party, :to_party, :manual_total, :notes," " :created_by, :created_at)" ), { "id": invoice_id, # invoice ids carry unchanged "tenant_id": ctx.tenant_by_org[org_id], "name": name, "status": status, "currency": CURRENCY, "invoice_date": invoice_date, "sent_at": sent_at, "paid_at": paid_at, "from_party": tenant_party, "to_party": to_party, "manual_total": manual_total, "notes": notes, "created_by": ctx.user_by_legacy_uid.get(created_by), "created_at": created_at, }, ) ctx.account("invoices", "migrated") step.written += 1 step.details["owner_review"] = review # Attached entries become time-kind lines; the entry references its # line (the lock chain). Snapshot names as of migration — best # available approximation of attach-time state (judgment #9). conn.execute( text( "CREATE TEMP TABLE tmp_lines ON COMMIT DROP AS" " SELECT te.id AS entry_id, te.invoice_id, gen_random_uuid() AS line_id," " te.description, te.duration_seconds, te.billout_amount_usd," " te.invoice_applied_at, te.invoice_applied_by," " c.name AS client_name, p.name AS project_name" " FROM legacy.time_entries te" " LEFT JOIN legacy.projects p ON p.id = te.project_id" " LEFT JOIN legacy.clients c ON c.id = p.client_id" " WHERE te.invoice_id IS NOT NULL" ) ) lines = conn.execute( text( "INSERT INTO invoice_lines (id, invoice_id, kind, description, quantity," " amount, snapshot_client_name, snapshot_project_name, attached_at," " attached_by)" " SELECT l.line_id, l.invoice_id, 'time', l.description," " round(l.duration_seconds / 3600.0, 2), l.billout_amount_usd," " l.client_name, l.project_name, l.invoice_applied_at, um.user_id" " FROM tmp_lines l" " LEFT JOIN tmp_user_map um ON um.legacy_uid = l.invoice_applied_by" ) ) conn.execute( text( "UPDATE time_entries SET invoice_line_id = l.line_id" " FROM tmp_lines l WHERE time_entries.id = l.entry_id" ) ) step.details["time_lines"] = lines.rowcount # Judgment #22 — transferred-but-not-invoiced stragglers: expected ZERO # (the 2026-05-25 backfill put every transferred row on a per-org # "Legacy transfers" invoice). Check -> report -> attach to the # EXISTING sentinel; create-if-absent with a louder warning; NEVER abort. stragglers = conn.execute( text( "SELECT te.id, te.org_id, te.transferred_at, te.transferred_by," " te.description, te.duration_seconds, te.billout_amount_usd," " c.name, p.name" " FROM legacy.time_entries te" " LEFT JOIN legacy.projects p ON p.id = te.project_id" " LEFT JOIN legacy.clients c ON c.id = p.client_id" " WHERE te.transferred_at IS NOT NULL AND te.invoice_id IS NULL" " ORDER BY te.start_at, te.id" ) ).fetchall() ctx.report.straggler_count = len(stragglers) ctx.report.straggler_entry_ids = [str(row[0]) for row in stragglers] if stragglers: message = ( f"JUDGMENT #22: {len(stragglers)} transferred-but-not-invoiced" " straggler(s) found (expected 0) — attached to the per-org" " 'Legacy transfers' invoice for owner review;" f" entry ids: {ctx.report.straggler_entry_ids}" ) step.warn(message) ctx.report.warnings.append(message) sentinel_by_org: dict[uuid.UUID, uuid.UUID] = dict( conn.execute( text( "SELECT org_id, id FROM legacy.invoices" " WHERE lower(name) = 'legacy transfers'" ) ).fetchall() ) for ( entry_id, org_id, transferred_at, transferred_by, description, duration_seconds, billout_amount, client_name, project_name, ) in stragglers: sentinel = sentinel_by_org.get(org_id) if sentinel is None: sentinel = uuid.uuid4() conn.execute( text( "INSERT INTO invoices (id, tenant_id, name, status," " currency, from_party_id, notes)" " VALUES (:id, :tenant_id, 'Legacy transfers', 'paid'," " :currency, :from_party," " 'Created by migration for judgment-#22 stragglers —" " the org somehow lacked the 2026-05-25 backfill sentinel')" ), { "id": sentinel, "tenant_id": ctx.tenant_by_org[org_id], "currency": CURRENCY, "from_party": ctx.tenant_party_by_org[org_id], }, ) sentinel_by_org[org_id] = sentinel louder = ( "JUDGMENT #22 ESCALATION: org lacked a 'Legacy transfers'" " invoice entirely — created one (owner MUST review)" ) step.warn(louder) ctx.report.warnings.append(louder) ctx.account("invoices", "straggler_sentinel_created") line_id = uuid.uuid4() conn.execute( text( "INSERT INTO invoice_lines (id, invoice_id, kind, description," " quantity, amount, snapshot_client_name, snapshot_project_name," " attached_at, attached_by)" " VALUES (:id, :invoice_id, 'time', :description, :quantity," " :amount, :client_name, :project_name, :attached_at," " :attached_by)" ), { "id": line_id, "invoice_id": sentinel, "description": description, "quantity": ( round(Decimal(duration_seconds) / 3600, 2) if duration_seconds is not None else None ), "amount": billout_amount, "client_name": client_name, "project_name": project_name, "attached_at": transferred_at, "attached_by": ctx.user_by_legacy_uid.get(transferred_by), }, ) conn.execute( text( "UPDATE time_entries SET invoice_line_id = :line_id" " WHERE id = :entry_id" ), {"line_id": line_id, "entry_id": entry_id}, ) # Single-currency document check (02 §3): every attached entry's # billout currency must match its invoice's currency. conflicts = _count( conn, "SELECT count(*) FROM time_entries te" " JOIN invoice_lines il ON il.id = te.invoice_line_id" " JOIN invoices i ON i.id = il.invoice_id" " WHERE te.billout_currency IS NOT NULL AND te.billout_currency <> i.currency", ) if conflicts: raise MigrationError( f"{conflicts} attached entr(y/ies) carry a billout currency that" " differs from their invoice's currency (single-currency documents)" ) # --- 2.8 CC expenses --------------------------------------------------------------- def step_cc(conn: Connection, ctx: Ctx) -> None: step = ctx.report.step("2.8 CC expense batches/lines/rules/keywords") step.source_count = _count(conn, "SELECT count(*) FROM legacy.cc_expense_batches") batches = conn.execute( text( "INSERT INTO cc_expense_batches (id, tenant_id, name, currency," " created_by, created_at)" " SELECT b.id, t.tenant_id, b.name, :currency, um.user_id, b.created_at" " FROM legacy.cc_expense_batches b" " JOIN tmp_org_tenant t ON t.org_id = b.org_id" " LEFT JOIN tmp_user_map um ON um.legacy_uid = b.created_by" ), {"currency": CURRENCY}, ) ctx.account("cc_expense_batches", "migrated", batches.rowcount) lines = conn.execute( text( "INSERT INTO cc_expense_lines (id, batch_id, line_index, raw_line," " expense_date, description, amount, balance, project_id," " assignment_description, category, auto_assigned, invoice_line_id," " created_at)" " SELECT id, batch_id, line_index, raw_line, expense_date, description," " amount_usd, balance_usd, project_id, assignment_description, category," " auto_assigned, NULL, created_at" # invoice_line_id starts NULL (05 §2.8) " FROM legacy.cc_expense_lines" ) ) ctx.account("cc_expense_lines", "migrated", lines.rowcount) keywords = conn.execute( text( "INSERT INTO cc_highlight_keywords (id, tenant_id, keyword, created_at)" " SELECT k.id, t.tenant_id, k.keyword, k.created_at" " FROM legacy.cc_highlight_keywords k" " JOIN tmp_org_tenant t ON t.org_id = k.org_id" ) ) ctx.account("cc_highlight_keywords", "migrated", keywords.rowcount) rules = conn.execute( text( "INSERT INTO cc_auto_rules (id, tenant_id, keyword, project_id," " assignment_description, category, created_at)" " SELECT r.id, t.tenant_id, r.keyword, r.project_id," " r.assignment_description, r.category, r.created_at" " FROM legacy.cc_auto_rules r" " JOIN tmp_org_tenant t ON t.org_id = r.org_id" ) ) ctx.account("cc_auto_rules", "migrated", rules.rowcount) step.written = batches.rowcount + lines.rowcount + keywords.rowcount + rules.rowcount step.details = { "batches": batches.rowcount, "lines": lines.rowcount, "highlight_keywords": keywords.rowcount, "auto_rules": rules.rowcount, } # --- 2.9 comments ------------------------------------------------------------------- def step_comments(conn: Connection, ctx: Ctx) -> None: step = ctx.report.step("2.9 comments") step.source_count = _count(conn, "SELECT count(*) FROM legacy.comments") result = conn.execute( text( "INSERT INTO comments (id, tenant_id, author_id, body, resolved_at," " resolved_by, created_at)" " SELECT c.id, t.tenant_id, ua.user_id, c.body, c.resolved_at," " ur.user_id, c.created_at" " FROM legacy.comments c" " JOIN tmp_org_tenant t ON t.org_id = c.org_id" " LEFT JOIN tmp_user_map ua ON ua.legacy_uid = c.author_id" " LEFT JOIN tmp_user_map ur ON ur.legacy_uid = c.resolved_by" ) ) step.written = result.rowcount ctx.account("comments", "migrated", result.rowcount) # --- reconciliation + currency counts -------------------------------------------------- def finalize(conn: Connection, ctx: Ctx) -> None: # teams dissolve — accounted as a whole table. ctx.account( "teams", "dissolved_dropped", _count(conn, "SELECT count(*) FROM legacy.teams") ) for table in LEGACY_TABLES: source = _count(conn, f"SELECT count(*) FROM legacy.{table}") dispositions = dict(ctx.dispositions.get(table, {})) # Composite-disposition tables: rows are accounted once per pass — # normalize to a single per-row accounting before the delta check. if table == "clients": # each client row = a party decision (2.1) AND an edge (2.2); # the edge pass is authoritative for row accounting. dispositions = { k: v for k, v in dispositions.items() if k.startswith("direct_client") } if table == "team_members": # each row = one engagement (2.3); party creation/dedupe (2.1) # is informational. dispositions = { k: v for k, v in dispositions.items() if k == "engagement_created" } if table == "invoices": dispositions.pop("straggler_sentinel_created", None) ctx.report.reconcile(table, source, dispositions) for column, sql in { "time_entries.cost_currency": ( "SELECT cost_currency, count(*) FROM time_entries" " WHERE cost_currency IS NOT NULL GROUP BY 1" ), "time_entries.billout_currency": ( "SELECT billout_currency, count(*) FROM time_entries" " WHERE billout_currency IS NOT NULL GROUP BY 1" ), "invoices.currency": "SELECT currency, count(*) FROM invoices GROUP BY 1", "cc_expense_batches.currency": ( "SELECT currency, count(*) FROM cc_expense_batches GROUP BY 1" ), "compensation_terms.currency": ( "SELECT currency, count(*) FROM compensation_terms" " WHERE currency IS NOT NULL GROUP BY 1" ), "rate_overrides.currency": ( "SELECT currency, count(*) FROM rate_overrides" " WHERE currency IS NOT NULL GROUP BY 1" ), "projects.income_currency": ( "SELECT income_currency, count(*) FROM projects" " WHERE income_currency IS NOT NULL GROUP BY 1" ), }.items(): ctx.report.currency_counts[column] = { currency: count for currency, count in conn.execute(text(sql)).fetchall() } # --- orchestration ----------------------------------------------------------------------- class _ReportedFailure(Exception): """Internal: reconciliation failed — roll back, report already filled.""" def _default_url() -> str: """In-container: WITH_DB_URL is in the process env (compose env_file). Host-side: only the .env FILE has it (pointing at the compose-internal host 'db'), so prefer WITH_DB_URL_HOST — same rule as alembic/env.py.""" import os from app.config import settings url = ( os.environ.get("WITH_DB_URL") or settings.with_db_url_host or settings.with_db_url ) if not url: raise RuntimeError("set WITH_DB_URL / WITH_DB_URL_HOST (see .env)") return url def run_all( dsn: str | None = None, *, engine: Engine | None = None, identity_map: dict[str, str] | None = None, books_source: str | Path | None = None, ) -> RunReport: """Run every transform in ONE transaction; write the report row after. Returns the RunReport; ``report.ok`` False means the transaction was rolled back (public.* untouched) — except judgment #22 stragglers, which are warnings by design, never aborts. ``books_source`` (R2, ADR #016) controls the PlusROI sheet-books step: None (default) skips it entirely — the legacy-only behavior every harness/parity/reporting test relies on; ``"auto"`` resolves the app-relative ``books-import/source`` (present → full run, missing → LOUD warn + skip); a path uses that directory (used by the sheet tests). The step runs INSIDE this transaction, so a sheet reconciliation failure rolls back the whole rebuild (by design — sheet ownership means the sheet books are part of the nightly rebuild). When the PlusROI tenant is APP-owned (settings.books_owner == "app", ADR #021) the step skips it and the tenant-scoped reset preserves every one of its rows. """ own_engine = engine is None eng = engine or create_engine(dsn or _default_url(), pool_pre_ping=True) report = RunReport() ctx = Ctx( report=report, identity_map=dict(identity_map or IDENTITY_MAP), migration_date=datetime.now(UTC).date(), books_source=books_source, ) try: with eng.begin() as conn: preflight(conn, ctx) if not report.ok: # column-disposition failures raise _ReportedFailure() step_parties_tenant_users(conn, ctx) step_client_edges(conn, ctx) step_engagements_terms(conn, ctx) step_projects(conn, ctx) step_import_batches(conn, ctx) step_time_entries(conn, ctx) step_invoices(conn, ctx) step_cc(conn, ctx) step_comments(conn, ctx) if ctx.books_source is not None: # local import breaks the transforms<->sheet_books cycle from app.services.migrate.sheet_books import step_sheet_books step_sheet_books(conn, ctx) finalize(conn, ctx) if not report.ok: raise _ReportedFailure() except _ReportedFailure: log.error("migration run failed — transaction rolled back") except Exception as exc: # hard failure: also rolled back report.errors.append(f"{type(exc).__name__}: {exc}") log.exception("migration run crashed — transaction rolled back") report.finish() _write_report_row(eng, report) if own_engine: eng.dispose() return report def _write_report_row(engine: Engine, report: RunReport) -> None: """Persist the run report OUTSIDE the transform transaction so a rolled-back run still leaves its report behind.""" try: with engine.begin() as conn: conn.execute( text( "INSERT INTO migration_runs (id, started_at, finished_at, ok," " report) VALUES (:id, cast(:started_at AS timestamptz)," " cast(:finished_at AS timestamptz), :ok, cast(:report AS jsonb))" ), { "id": uuid.uuid4(), "started_at": report.started_at, "finished_at": report.finished_at, "ok": report.ok, "report": json.dumps(report.as_dict(), default=str), }, ) except Exception: log.exception("could not persist migration_runs report row")