from __future__ import annotations

import os
import sqlite3
from pathlib import Path
from uuid import uuid4

from sqlalchemy import create_engine, text


def _source_database(path: Path) -> None:
    connection = sqlite3.connect(path)
    connection.executescript(
        """
        CREATE TABLE sessions (
            id TEXT PRIMARY KEY, stage TEXT NOT NULL, revision INTEGER NOT NULL,
            profile_json TEXT NOT NULL, draft_id TEXT, resume_id TEXT,
            created_at TEXT NOT NULL, updated_at TEXT NOT NULL
        );
        CREATE TABLE turns (
            id TEXT PRIMARY KEY, session_id TEXT NOT NULL, sequence INTEGER NOT NULL,
            role TEXT NOT NULL, content TEXT, composer_mode TEXT NOT NULL, created_at TEXT NOT NULL
        );
        CREATE TABLE blocks (
            id TEXT PRIMARY KEY, session_id TEXT NOT NULL, turn_id TEXT NOT NULL,
            block_index INTEGER NOT NULL, type TEXT NOT NULL, lifecycle TEXT NOT NULL,
            data_json TEXT NOT NULL, version INTEGER NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL
        );
        CREATE TABLE resumes (
            id TEXT PRIMARY KEY, session_id TEXT NOT NULL, idempotency_key TEXT,
            revision INTEGER NOT NULL, content_json TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL
        );
        CREATE TABLE resume_imports (
            id TEXT PRIMARY KEY, session_id TEXT NOT NULL, file_name TEXT NOT NULL,
            mime_type TEXT NOT NULL, size_bytes INTEGER NOT NULL, sha256 TEXT NOT NULL,
            object_key TEXT NOT NULL, status TEXT NOT NULL, document_json TEXT,
            field_reviews_json TEXT NOT NULL, error_code TEXT,
            created_at TEXT NOT NULL, updated_at TEXT NOT NULL
        );
        CREATE TABLE optimization_runs (
            id TEXT PRIMARY KEY, session_id TEXT NOT NULL, entry_id TEXT NOT NULL,
            mode TEXT NOT NULL, status TEXT NOT NULL, source_revision INTEGER NOT NULL,
            state_json TEXT NOT NULL, proposal_json TEXT,
            created_at TEXT NOT NULL, updated_at TEXT NOT NULL
        );
        """
    )
    connection.execute(
        "INSERT INTO sessions VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
        (
            "session-1", "CONTENT_READY", 7,
            '{"job_type":"other","name":"张三","metadata":{"city":"上海"}}',
            "draft-1", "resume-1", "2026-01-01T00:00:00+00:00", "2026-01-02T00:00:00+00:00",
        ),
    )
    connection.execute(
        "INSERT INTO turns VALUES (?, ?, ?, ?, ?, ?, ?)",
        ("turn-1", "session-1", 1, "assistant", "欢迎", "ui_only", "2026-01-01T00:00:00+00:00"),
    )
    connection.execute(
        "INSERT INTO blocks VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
        ("block-1", "session-1", "turn-1", 0, "component", "active", '{"label":"基本信息"}', 2, "2026-01-01T00:00:00+00:00", "2026-01-01T00:00:00+00:00"),
    )
    connection.execute(
        "INSERT INTO resumes VALUES (?, ?, ?, ?, ?, ?, ?)",
        ("resume-1", "session-1", "create-1", 3, '{"schema_version":3,"basics":{"name":"张三"}}', "2026-01-01T00:00:00+00:00", "2026-01-02T00:00:00+00:00"),
    )
    connection.execute(
        "INSERT INTO resume_imports VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
        ("import-1", "session-1", "resume.pdf", "application/pdf", 42, "a" * 64,
         "aa/import-1.pdf", "awaiting_review", '{"schema_version":3}',
         '[{"field_path":"basics.name"}]', None,
         "2026-01-01T00:00:00+00:00", "2026-01-02T00:00:00+00:00"),
    )
    connection.execute(
        "INSERT INTO optimization_runs VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
        ("run-1", "session-1", "entry-1", "deep", "proposal_pending", 3,
         '{"question_index":1}', '{"summary":"improved"}',
         "2026-01-01T00:00:00+00:00", "2026-01-02T00:00:00+00:00"),
    )

    connection.commit()
    connection.close()
def test_sqlite_migration_preserves_ids_order_and_normalizes_legacy_job_type(tmp_path: Path) -> None:
    from app.db.sqlite_migration import migrate_sqlite_to_postgres

    source = tmp_path / "legacy.db"
    _source_database(source)
    schema = f"test_migration_{uuid4().hex}"
    target_url = os.environ["RESUME_AGENT_TEST_DATABASE_URL"]

    report = migrate_sqlite_to_postgres(source, target_url, schema=schema)

    assert report.source_counts == {
        "sessions": 1,
        "turns": 1,
        "blocks": 1,
        "resumes": 1,
        "resume_imports": 1,
        "optimization_runs": 1,
    }
    assert report.target_counts == report.source_counts
    assert report.source_checksum == report.target_checksum
    engine = create_engine(target_url)
    try:
        with engine.connect() as connection:
            profile = connection.execute(
                text(f'SELECT profile FROM "{schema}".sessions WHERE id = :id'), {"id": "session-1"}
            ).scalar_one()
            sequence = connection.execute(
                text(f'SELECT sequence FROM "{schema}".turns WHERE id = :id'), {"id": "turn-1"}
            ).scalar_one()
        assert profile["job_type"] == "internship"
        assert profile["name"] == "张三"
        assert sequence == 1
    finally:
        with engine.begin() as connection:
            connection.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
        engine.dispose()


def test_sqlite_migration_dry_run_does_not_create_target_schema(tmp_path: Path) -> None:
    from app.db.sqlite_migration import migrate_sqlite_to_postgres

    source = tmp_path / "legacy.db"
    _source_database(source)
    schema = f"test_migration_{uuid4().hex}"

    report = migrate_sqlite_to_postgres(
        source, os.environ["RESUME_AGENT_TEST_DATABASE_URL"], schema=schema, dry_run=True
    )

    assert report.target_counts == {}
    engine = create_engine(os.environ["RESUME_AGENT_TEST_DATABASE_URL"])
    try:
        with engine.connect() as connection:
            exists = connection.execute(
                text("SELECT EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = :schema)"),
                {"schema": schema},
            ).scalar_one()
        assert exists is False
    finally:
        engine.dispose()