From faa30fd27bb90f05a90cbe53c93ec756c38e6578 Mon Sep 17 00:00:00 2001 From: Jim Lancaster <40281233+zoltan57@users.noreply.github.com> Date: Tue, 25 Aug 2026 11:44:59 -0500 Subject: [PATCH] V6 Phase 2 complete --- .gitignore | 18 +-- docker-compose.production.yml | 2 + docs/data_migration.md | 48 +++++- docs/production-runbook.md | 1 + docs/v6_0_hosting_migration_plan.md | 5 + src/transcription/db/migration.py | 168 ++++++++++++++++++++ tests/tools/test_export_import_migration.py | 141 ++++++++++++++++ tools/export_import_migration.py | 8 + 8 files changed, 376 insertions(+), 15 deletions(-) diff --git a/.gitignore b/.gitignore index c9dc1f0..875ec08 100644 --- a/.gitignore +++ b/.gitignore @@ -11,19 +11,19 @@ wheels/ # Environment secrets .env +.env.production # SQLite database *.db -# Document images -uploads/* +# All data including db, backups, document images, photos, and logs: data/* +data/backups/* +data/documents/* +data/logs/* +data/photos/* -# Local destructive-test backups -.test-backups/ - -# Temporary migration files -.migration-bundle-v51 -data.pre-v50-20260823/* -data.pre-v51-20260823-120434/* +# Migration tests +data-migration-test/* +.migration-bundle/* diff --git a/docker-compose.production.yml b/docker-compose.production.yml index ef79a31..a8136a7 100644 --- a/docker-compose.production.yml +++ b/docker-compose.production.yml @@ -54,6 +54,8 @@ services: timeout: 5s retries: 10 start_period: 10s + ports: + - "5432:5432" cloudflared: image: cloudflare/cloudflared:2026.8.0 diff --git a/docs/data_migration.md b/docs/data_migration.md index 3cdcb4d..06f44f4 100644 --- a/docs/data_migration.md +++ b/docs/data_migration.md @@ -19,13 +19,36 @@ Optional source overrides: - `--source-db ` - `--source-upload-dir ` -### 2) Import bundle into a fresh DB + uploads root +### 2) Import bundle into a fresh target (SQLite or PostgreSQL) ```bash uv run python tools/export_import_migration.py import --bundle-dir .migration-bundle --target-db .\data\transcription-new.db --target-upload-dir .\data-new ``` -### 3) One-shot export+import +PostgreSQL target example: + +```bash +uv run python tools/export_import_migration.py import --bundle-dir .migration-bundle --target-db postgresql://transcription:change-me@localhost:5432/transcription --target-upload-dir .\data-new +``` + +### 3) Verify migration parity and integrity + +```bash +uv run python tools/export_import_migration.py verify --source-db .\data\transcription.db --target-db postgresql://transcription:change-me@localhost:5432/transcription +``` + +The verify command checks: + +- row-count parity across migration tables +- orphan-reference checks for `source`, `job`, `job_source`, and `execution_attempt` +- duplicate `(job_id, source_id, attempt_number)` in `execution_attempt` + +Exit code: + +- `0` when counts and integrity checks pass +- `1` when mismatches or integrity violations are detected + +### 4) One-shot export+import ```bash uv run python tools/export_import_migration.py migrate --bundle-dir .migration-bundle --target-db .\data\transcription-new.db --target-upload-dir .\data-new @@ -50,9 +73,22 @@ Legacy V4.x portrait/homepage backfill in the export step: - Legacy homepage markdown is relocated from `UPLOAD_DIR/homepage/homepage.md` to `UPLOAD_DIR/homepage.md`. - Legacy `person.full_name` values are split into `given_names` + `last_name` for V5.1 schema compatibility. -## Cutover +## Cutover (SQLite -> PostgreSQL) After importing to a fresh target: -1. Stop the app. -2. Point `DATABASE__*` and `UPLOAD_DIR` to the new targets. -3. Start the app and run smoke checks (`/healthz`, create/upload/process one job). +1. Stop app and worker services to freeze writes. +2. Export a migration bundle from the last SQLite state. +3. Import bundle to PostgreSQL target. +4. Run `verify` against source and target before switching runtime. +5. Switch runtime config to PostgreSQL (`DATABASE__DRIVER=postgres` and related `DATABASE__*` values). +6. Start app and worker services. +7. Run smoke checks (`/healthz`, create/upload/process one job). + +## Rollback + +If verify or smoke checks fail: + +1. Stop app and worker services. +2. Revert runtime config to SQLite. +3. Start app and worker against pre-cutover SQLite database. +4. Preserve failed migration bundle and logs for analysis. diff --git a/docs/production-runbook.md b/docs/production-runbook.md index 19b05e9..86d3028 100644 --- a/docs/production-runbook.md +++ b/docs/production-runbook.md @@ -20,6 +20,7 @@ This runbook is the operational checklist for releasing and monitoring the trans 1. Deploy artifact/config to target environment. - V6.0 Phase 1 production stack: `docker compose -f docker-compose.production.yml up -d --build` + - For SQLite -> PostgreSQL cutover, run `uv run python tools/export_import_migration.py verify --source-db --target-db ` before switching runtime. 2. Validate service startup: - `/healthz` responds `200` - if `RUN_EMBEDDED_WORKER=true`, `worker.state` is `running` diff --git a/docs/v6_0_hosting_migration_plan.md b/docs/v6_0_hosting_migration_plan.md index bfef8bc..3c695b0 100644 --- a/docs/v6_0_hosting_migration_plan.md +++ b/docs/v6_0_hosting_migration_plan.md @@ -38,6 +38,11 @@ Persistence: 2. Run migration in staging-like environment and validate entity counts and key relationships. 3. Execute cutover with rollback guardrails and preserved evidence/provenance history. +Implemented workflow references: + +- `tools/export_import_migration.py` (`export`, `import`, `migrate`, `verify`) +- `docs/data_migration.md` for cutover and rollback procedure + ## 3.3 Cloudflare remote access 1. Configure tunnel routing for service hostnames. diff --git a/src/transcription/db/migration.py b/src/transcription/db/migration.py index 8760a81..3d57dac 100644 --- a/src/transcription/db/migration.py +++ b/src/transcription/db/migration.py @@ -16,9 +16,12 @@ from uuid import uuid4 from sqlalchemy import URL from sqlalchemy import MetaData from sqlalchemy import Table +from sqlalchemy import bindparam from sqlalchemy import create_engine +from sqlalchemy import func from sqlalchemy import inspect as sqlalchemy_inspect from sqlalchemy import select +from sqlalchemy import text from sqlalchemy.engine import RowMapping from sqlalchemy.engine import make_url from sqlmodel import SQLModel @@ -47,6 +50,7 @@ EXPORT_TABLE_ORDER = ( ) BYTES_FIELDS = {"transport_body"} +VERIFICATION_TABLES = EXPORT_TABLE_ORDER @dataclass(frozen=True) @@ -142,21 +146,74 @@ def import_bundle(*, target_db_url: str, target_upload_dir: Path, bundle_dir: Pa engine = create_engine(target_db_url) try: SQLModel.metadata.create_all(engine) + execution_attempt_ids = _collect_execution_attempt_ids(payload) with engine.begin() as connection: for table_name in reversed(EXPORT_TABLE_ORDER): table = SQLModel.metadata.tables[table_name] connection.execute(table.delete()) + deferred_source_preferred_attempt_updates: list[dict[str, Any]] = [] for table_name in EXPORT_TABLE_ORDER: rows = payload.get("tables", {}).get(table_name, []) if not rows: continue table = SQLModel.metadata.tables[table_name] + if table_name == "source": + prepared_source_rows, updates = _prepare_source_rows_for_import( + rows=rows, + source_table=table, + execution_attempt_ids=execution_attempt_ids, + ) + deferred_source_preferred_attempt_updates.extend(updates) + connection.execute(table.insert(), prepared_source_rows) + continue connection.execute(table.insert(), [_deserialize_row(row, table) for row in rows]) + + if deferred_source_preferred_attempt_updates: + source_table = SQLModel.metadata.tables["source"] + connection.execute( + source_table.update() + .where(source_table.c.id == bindparam("source_id")) + .values(preferred_execution_attempt_id=bindparam("preferred_execution_attempt_id")), + deferred_source_preferred_attempt_updates, + ) finally: engine.dispose() +def _collect_execution_attempt_ids(payload: dict[str, Any]) -> set[str]: + execution_attempt_rows = payload.get("tables", {}).get("execution_attempt", []) + return {_normalize_uuid_like(row.get("id")) for row in execution_attempt_rows if row.get("id") is not None} + + +def _prepare_source_rows_for_import( + *, + rows: list[dict[str, Any]], + source_table: Table, + execution_attempt_ids: set[str], +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + prepared_source_rows: list[dict[str, Any]] = [] + updates: list[dict[str, Any]] = [] + for row in rows: + source_row = _deserialize_row(row, source_table) + source_id = source_row.get("id") + preferred_attempt_id = source_row.get("preferred_execution_attempt_id") + if ( + source_id is not None + and preferred_attempt_id is not None + and _normalize_uuid_like(preferred_attempt_id) in execution_attempt_ids + ): + updates.append( + { + "source_id": source_id, + "preferred_execution_attempt_id": preferred_attempt_id, + } + ) + source_row["preferred_execution_attempt_id"] = None + prepared_source_rows.append(source_row) + return prepared_source_rows, updates + + def _ensure_sqlite_target_parent_exists(target_db_url: str) -> None: parsed = make_url(target_db_url) if not parsed.drivername.startswith("sqlite"): @@ -192,6 +249,24 @@ def migrate_via_bundle(paths: MigrationPaths) -> None: ) +@dataclass(frozen=True) +class MigrationVerificationReport: + source_counts: dict[str, int] + target_counts: dict[str, int] + mismatched_tables: dict[str, dict[str, int]] + integrity_violations: dict[str, int] + success: bool + + def to_dict(self) -> dict[str, Any]: + return { + "success": self.success, + "source_counts": self.source_counts, + "target_counts": self.target_counts, + "mismatched_tables": self.mismatched_tables, + "integrity_violations": self.integrity_violations, + } + + def sqlite_url_from_path(path: Path) -> str: return URL.create(drivername="sqlite", database=str(path)).render_as_string(hide_password=False) @@ -201,6 +276,87 @@ def default_sync_db_url(settings: Settings | None = None) -> str: return get_database_url(runtime_settings).replace("+aiosqlite", "").replace("+asyncpg", "").replace("+psycopg", "") +def verify_migration(*, source_db_url: str, target_db_url: str) -> MigrationVerificationReport: + source_counts = _table_counts(source_db_url) + target_counts = _table_counts(target_db_url) + mismatched_tables = { + table_name: {"source": source_counts[table_name], "target": target_counts[table_name]} + for table_name in VERIFICATION_TABLES + if source_counts[table_name] != target_counts[table_name] + } + integrity_violations = _integrity_violations(target_db_url) + success = not mismatched_tables and all(count == 0 for count in integrity_violations.values()) + return MigrationVerificationReport( + source_counts=source_counts, + target_counts=target_counts, + mismatched_tables=mismatched_tables, + integrity_violations=integrity_violations, + success=success, + ) + + +def _table_counts(db_url: str) -> dict[str, int]: + engine = create_engine(db_url) + try: + metadata = MetaData() + metadata.reflect(bind=engine) + counts: dict[str, int] = {} + with engine.connect() as connection: + for table_name in VERIFICATION_TABLES: + table = metadata.tables.get(table_name) + if table is None: + counts[table_name] = 0 + continue + counts[table_name] = int(connection.execute(select(func.count()).select_from(table)).scalar_one()) + return counts + finally: + engine.dispose() + + +def _integrity_violations(db_url: str) -> dict[str, int]: + checks = { + "orphan_source_document": ( + "select count(*) from source s left join document d on d.id = s.document_id where d.id is null" + ), + "orphan_job_document": ( + "select count(*) from job j left join document d on d.id = j.document_id where d.id is null" + ), + "orphan_job_source_job": ( + "select count(*) from job_source js left join job j on j.id = js.job_id where j.id is null" + ), + "orphan_job_source_source": ( + "select count(*) from job_source js left join source s on s.id = js.source_id where s.id is null" + ), + "orphan_attempt_job_source": ( + "select count(*) from execution_attempt ea " + "left join job_source js on js.id = ea.job_source_id " + "where js.id is null" + ), + "orphan_attempt_job": ( + "select count(*) from execution_attempt ea left join job j on j.id = ea.job_id where j.id is null" + ), + "orphan_attempt_source": ( + "select count(*) from execution_attempt ea left join source s on s.id = ea.source_id where s.id is null" + ), + "duplicate_attempt_numbers": ( + "select count(*) from (" + " select job_id, source_id, attempt_number, count(*) as c" + " from execution_attempt" + " group by job_id, source_id, attempt_number" + " having count(*) > 1" + ") x" + ), + } + engine = create_engine(db_url) + try: + with engine.connect() as connection: + return { + check_name: int(connection.execute(text(query)).scalar_one()) for check_name, query in checks.items() + } + finally: + engine.dispose() + + def _serialize_row(row: RowMapping, *, table_name: str, source_upload_dir: Path) -> dict[str, Any]: serialized: dict[str, Any] = {} for raw_key, value in row.items(): @@ -283,6 +439,18 @@ def _deserialize_value(python_type: type[Any], value: Any) -> Any: return value +def _normalize_uuid_like(value: Any) -> str: + if isinstance(value, UUID): + return str(value) + if isinstance(value, str): + text_value = value.strip() + try: + return str(UUID(text_value)) + except ValueError: + return text_value + return str(value) + + def _canonical_media_relative_path(value: str, *, source_upload_dir: Path, preferred_prefix: str) -> str: normalized = value.strip().replace("\\", "/") lowered = normalized.casefold() diff --git a/tests/tools/test_export_import_migration.py b/tests/tools/test_export_import_migration.py index 6455dbe..14b674d 100644 --- a/tests/tools/test_export_import_migration.py +++ b/tests/tools/test_export_import_migration.py @@ -16,6 +16,7 @@ from transcription.db.migration import export_bundle from transcription.db.migration import import_bundle from transcription.db.migration import migrate_via_bundle from transcription.db.migration import sqlite_url_from_path +from transcription.db.migration import verify_migration def test_export_import_migration_round_trips_db_and_uploads(tmp_path): @@ -339,3 +340,143 @@ def test_import_bundle_does_not_leave_upload_copy_db_as_final_database(tmp_path) assert count == 1 finally: target_engine.dispose() + + +def test_verify_migration_succeeds_for_matching_databases(tmp_path): + source_db_url = sqlite_url_from_path(tmp_path / "source.db") + target_db_url = sqlite_url_from_path(tmp_path / "target.db") + + source_engine = create_engine(source_db_url) + target_engine = create_engine(target_db_url) + try: + SQLModel.metadata.create_all(source_engine) + SQLModel.metadata.create_all(target_engine) + finally: + source_engine.dispose() + target_engine.dispose() + + report = verify_migration(source_db_url=source_db_url, target_db_url=target_db_url) + + assert report.success is True + assert report.mismatched_tables == {} + assert all(count == 0 for count in report.integrity_violations.values()) + + +def test_verify_migration_detects_count_mismatches(tmp_path): + source_db_url = sqlite_url_from_path(tmp_path / "source.db") + target_db_url = sqlite_url_from_path(tmp_path / "target.db") + document_id = uuid4() + + source_engine = create_engine(source_db_url) + target_engine = create_engine(target_db_url) + try: + SQLModel.metadata.create_all(source_engine) + SQLModel.metadata.create_all(target_engine) + with source_engine.begin() as connection: + connection.execute( + SQLModel.metadata.tables["document"].insert(), + [{"id": document_id, "name": "Source-only row"}], + ) + finally: + source_engine.dispose() + target_engine.dispose() + + report = verify_migration(source_db_url=source_db_url, target_db_url=target_db_url) + + assert report.success is False + assert report.mismatched_tables["document"] == {"source": 1, "target": 0} + + +def test_export_import_preserves_source_preferred_attempt_reference(tmp_path): + source_db_path = tmp_path / "source-preferred.db" + target_db_path = tmp_path / "target-preferred.db" + source_upload_dir = tmp_path / "source_uploads" + target_upload_dir = tmp_path / "target_uploads" + bundle_dir = tmp_path / "bundle-preferred" + + source_db_url = sqlite_url_from_path(source_db_path) + target_db_url = sqlite_url_from_path(target_db_path) + + document_id = uuid4() + job_id = uuid4() + source_id = uuid4() + job_source_id = uuid4() + attempt_id = uuid4() + filename = f"{source_id}.jpg" + media_path = source_upload_dir / "documents" / str(document_id) / filename + media_path.parent.mkdir(parents=True, exist_ok=True) + media_path.write_bytes(b"sample-image") + + source_engine = create_engine(source_db_url) + try: + SQLModel.metadata.create_all(source_engine) + with source_engine.begin() as connection: + connection.execute( + SQLModel.metadata.tables["document"].insert(), + [{"id": document_id, "name": "Preferred attempt doc"}], + ) + connection.execute( + SQLModel.metadata.tables["job"].insert(), + [{"id": job_id, "document_id": document_id, "status": "queued"}], + ) + connection.execute( + SQLModel.metadata.tables["source"].insert(), + [ + { + "id": source_id, + "document_id": document_id, + "page_number": 1, + "upload_name": "upload.jpg", + "filename": filename, + "file_path": str(media_path), + "file_hash": "b" * 64, + "file_size_bytes": len(b"sample-image"), + } + ], + ) + connection.execute( + SQLModel.metadata.tables["job_source"].insert(), + [{"id": job_source_id, "job_id": job_id, "source_id": source_id, "status": "pending"}], + ) + connection.execute( + SQLModel.metadata.tables["execution_attempt"].insert(), + [ + { + "id": attempt_id, + "job_source_id": job_source_id, + "job_id": job_id, + "source_id": source_id, + "attempt_number": 1, + "status": "transcribed", + "provider": "fixture", + "model": "fixture-model", + "raw_transcription": "hello", + "started_at": datetime.now(UTC), + "finished_at": datetime.now(UTC), + "duration_ms": 10, + } + ], + ) + connection.execute( + SQLModel.metadata.tables["source"] + .update() + .where(SQLModel.metadata.tables["source"].c.id == source_id) + .values(preferred_execution_attempt_id=attempt_id), + ) + finally: + source_engine.dispose() + + export_bundle(source_db_url=source_db_url, source_upload_dir=source_upload_dir, bundle_dir=bundle_dir) + import_bundle(target_db_url=target_db_url, target_upload_dir=target_upload_dir, bundle_dir=bundle_dir) + + target_engine = create_engine(target_db_url) + try: + with target_engine.connect() as connection: + preferred_attempt = connection.execute( + select(SQLModel.metadata.tables["source"].c.preferred_execution_attempt_id).where( + SQLModel.metadata.tables["source"].c.id == source_id + ) + ).scalar_one() + assert preferred_attempt == attempt_id + finally: + target_engine.dispose() diff --git a/tools/export_import_migration.py b/tools/export_import_migration.py index c308b9c..7a04ea2 100644 --- a/tools/export_import_migration.py +++ b/tools/export_import_migration.py @@ -17,6 +17,7 @@ Usage examples: from __future__ import annotations import argparse +import json from pathlib import Path from typing import Final @@ -27,6 +28,7 @@ from transcription.db.migration import export_bundle from transcription.db.migration import import_bundle from transcription.db.migration import migrate_via_bundle from transcription.db.migration import sqlite_url_from_path +from transcription.db.migration import verify_migration DEFAULT_BUNDLE_DIR: Final[Path] = Path(".migration-bundle") @@ -77,6 +79,7 @@ def main() -> int: subparsers.add_parser("export", parents=[common], help="Export DB rows and uploads into bundle") subparsers.add_parser("import", parents=[common], help="Import bundle into target DB and uploads") subparsers.add_parser("migrate", parents=[common], help="Run export then import in one command") + subparsers.add_parser("verify", parents=[common], help="Compare source/target row counts and integrity checks") args = parser.parse_args() bundle_dir = Path(args.bundle_dir) @@ -104,6 +107,11 @@ def main() -> int: print(f"Import complete: db={target_db_url} uploads={target_upload_dir}") return 0 + if args.command == "verify": + report = verify_migration(source_db_url=source_db_url, target_db_url=target_db_url) + print(json.dumps(report.to_dict(), indent=2)) + return 0 if report.success else 1 + migrate_via_bundle( MigrationPaths( source_db_url=source_db_url,