"""SQLite database — schema, connections, helpers."""

from __future__ import annotations

import hashlib
import sqlite3
from contextlib import contextmanager
from pathlib import Path

import structlog

from hive_compute.config import get_settings
from hive_compute.coordinator.worker_credentials import hash_worker_token, is_hashed_worker_token

logger = structlog.get_logger()

SCHEMA = """
CREATE TABLE IF NOT EXISTS workers (
    worker_id         TEXT PRIMARY KEY,
    worker_token      TEXT NOT NULL UNIQUE,
    worker_token_last_four TEXT,
    worker_token_issued_at TEXT NOT NULL DEFAULT (datetime('now')),
    worker_token_revoked_at TEXT,
    status            TEXT NOT NULL DEFAULT 'offline',
    chip              TEXT,
    memory_gb         INTEGER,
    gpu_cores         INTEGER,
    available_handlers TEXT NOT NULL DEFAULT '["transcribe"]',
    available_models  TEXT NOT NULL DEFAULT '[]',
    model_capabilities TEXT NOT NULL DEFAULT '[]',
    capabilities_advertised_at TEXT,
    capabilities_confirmed_at TEXT,
    last_poll_at      TEXT,
    last_heartbeat    TEXT,
    last_claimed_at   TEXT,
    last_claimed_job_id TEXT,
    last_auth_failure_at TEXT,
    last_auth_failure_detail TEXT,
    last_job_failure_at TEXT,
    last_job_failure_detail TEXT,
    last_failed_job_id TEXT,
    jobs_completed    INTEGER NOT NULL DEFAULT 0,
    registered_at     TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE TABLE IF NOT EXISTS jobs (
    job_id            TEXT PRIMARY KEY,
    job_type          TEXT NOT NULL DEFAULT 'transcribe',
    status            TEXT NOT NULL DEFAULT 'pending',
    priority          INTEGER NOT NULL DEFAULT 3,
    audio_url         TEXT NOT NULL,
    input_config      TEXT NOT NULL DEFAULT '{}',
    claimed_by        TEXT REFERENCES workers(worker_id),
    claimed_at        TEXT,
    last_heartbeat    TEXT,
    result_data       TEXT,
    error_message     TEXT,
    max_attempts      INTEGER NOT NULL DEFAULT 3,
    attempts          INTEGER NOT NULL DEFAULT 0,
    timeout_seconds   INTEGER NOT NULL DEFAULT 3600,
    callback_url      TEXT,
    audio_duration    REAL,
    shard_count       INTEGER,
    partner_id        TEXT,
    estimated_cost_cents INTEGER,
    idempotency_key   TEXT,
    external_job_id   TEXT,
    content_hash      TEXT,
    measured_duration  REAL,
    channel_count     INTEGER,
    original_format   TEXT,
    normalized_path   TEXT,
    audio_token       TEXT,
    created_at        TEXT NOT NULL DEFAULT (datetime('now')),
    completed_at      TEXT
);

CREATE UNIQUE INDEX IF NOT EXISTS idx_jobs_idempotency
    ON jobs(partner_id, idempotency_key)
    WHERE idempotency_key IS NOT NULL;

CREATE INDEX IF NOT EXISTS idx_jobs_pending
    ON jobs(status, priority, created_at)
    WHERE status = 'pending';

CREATE INDEX IF NOT EXISTS idx_jobs_stale
    ON jobs(status, last_heartbeat)
    WHERE status IN ('claimed', 'processing');

CREATE TABLE IF NOT EXISTS api_keys (
    key_hash          TEXT PRIMARY KEY,
    name              TEXT NOT NULL,
    created_at        TEXT NOT NULL DEFAULT (datetime('now')),
    revoked_at        TEXT
);

CREATE TABLE IF NOT EXISTS shards (
    shard_id          TEXT PRIMARY KEY,
    job_id            TEXT NOT NULL REFERENCES jobs(job_id),
    shard_index       INTEGER NOT NULL,
    start_seconds     REAL NOT NULL,
    end_seconds       REAL NOT NULL,
    overlap_seconds   REAL NOT NULL DEFAULT 1.0,
    status            TEXT NOT NULL DEFAULT 'pending',
    claimed_by        TEXT REFERENCES workers(worker_id),
    claimed_at        TEXT,
    last_heartbeat    TEXT,
    result_data       TEXT,
    error_message     TEXT,
    attempts          INTEGER NOT NULL DEFAULT 0,
    max_attempts      INTEGER NOT NULL DEFAULT 3,
    canary_id         TEXT REFERENCES canary_corpus(canary_id),
    created_at        TEXT NOT NULL DEFAULT (datetime('now')),
    completed_at      TEXT
);

CREATE INDEX IF NOT EXISTS idx_shards_pending
    ON shards(status, job_id, shard_index)
    WHERE status = 'pending';

CREATE INDEX IF NOT EXISTS idx_shards_job
    ON shards(job_id, shard_index);

CREATE TABLE IF NOT EXISTS worker_reputation (
    worker_id         TEXT PRIMARY KEY REFERENCES workers(worker_id),
    trust_tier        TEXT NOT NULL DEFAULT 'new',
    quality_score     REAL NOT NULL DEFAULT 0.5,
    total_shards      INTEGER NOT NULL DEFAULT 0,
    successful_shards INTEGER NOT NULL DEFAULT 0,
    failed_shards     INTEGER NOT NULL DEFAULT 0,
    canary_passes     INTEGER NOT NULL DEFAULT 0,
    canary_fails      INTEGER NOT NULL DEFAULT 0,
    updated_at        TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE TABLE IF NOT EXISTS canary_corpus (
    canary_id         TEXT PRIMARY KEY,
    audio_url         TEXT NOT NULL,
    expected_transcript TEXT NOT NULL,
    language          TEXT NOT NULL DEFAULT 'en',
    duration_seconds  REAL NOT NULL,
    difficulty        TEXT NOT NULL DEFAULT 'normal',
    created_at        TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE TABLE IF NOT EXISTS validation_results (
    validation_id     INTEGER PRIMARY KEY AUTOINCREMENT,
    shard_id          TEXT NOT NULL,
    job_id            TEXT NOT NULL,
    worker_id         TEXT NOT NULL,
    validation_type   TEXT NOT NULL,
    expected_text     TEXT,
    actual_text       TEXT,
    similarity_score  REAL,
    passed            INTEGER NOT NULL DEFAULT 0,
    detail            TEXT,
    created_at        TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_validation_worker
    ON validation_results(worker_id, created_at);

CREATE INDEX IF NOT EXISTS idx_validation_shard
    ON validation_results(shard_id);

CREATE TABLE IF NOT EXISTS partners (
    partner_id        TEXT PRIMARY KEY,
    name              TEXT NOT NULL,
    api_key_hash      TEXT NOT NULL UNIQUE,
    callback_url      TEXT,
    callback_secret   TEXT,
    rate_per_minute   REAL NOT NULL DEFAULT 0.006,
    credits_usd       REAL,
    monthly_budget_cents INTEGER,
    status            TEXT NOT NULL DEFAULT 'active',
    created_at        TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE TABLE IF NOT EXISTS billing_items (
    billing_id        INTEGER PRIMARY KEY AUTOINCREMENT,
    partner_id        TEXT NOT NULL REFERENCES partners(partner_id),
    job_id            TEXT NOT NULL UNIQUE REFERENCES jobs(job_id),
    duration_seconds  REAL NOT NULL,
    rate_per_minute   REAL NOT NULL,
    amount_cents      INTEGER NOT NULL,
    status            TEXT NOT NULL DEFAULT 'pending',
    created_at        TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_billing_partner
    ON billing_items(partner_id, created_at);

CREATE TABLE IF NOT EXISTS payout_items (
    payout_id         INTEGER PRIMARY KEY AUTOINCREMENT,
    worker_id         TEXT NOT NULL REFERENCES workers(worker_id),
    job_id            TEXT NOT NULL REFERENCES jobs(job_id),
    shard_id          TEXT,
    duration_seconds  REAL NOT NULL,
    rate_per_minute   REAL NOT NULL DEFAULT 0.001,
    amount_cents      INTEGER NOT NULL,
    status            TEXT NOT NULL DEFAULT 'held',
    hold_until        TEXT,
    released_at       TEXT,
    reversed_at       TEXT,
    created_at        TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE UNIQUE INDEX IF NOT EXISTS idx_payout_shard_worker
    ON payout_items(shard_id, worker_id)
    WHERE shard_id IS NOT NULL;

CREATE UNIQUE INDEX IF NOT EXISTS idx_payout_job_worker
    ON payout_items(job_id, worker_id)
    WHERE shard_id IS NULL;

CREATE INDEX IF NOT EXISTS idx_payout_worker
    ON payout_items(worker_id, status);

CREATE TABLE IF NOT EXISTS worker_payout_events (
    payout_event_id   INTEGER PRIMARY KEY AUTOINCREMENT,
    worker_id         TEXT NOT NULL REFERENCES workers(worker_id),
    amount_cents      INTEGER NOT NULL,
    status            TEXT NOT NULL DEFAULT 'sent',
    note              TEXT,
    created_at        TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_worker_payout_events_worker
    ON worker_payout_events(worker_id, created_at);

CREATE TABLE IF NOT EXISTS worker_benchmarks (
    benchmark_id      INTEGER PRIMARY KEY AUTOINCREMENT,
    worker_id         TEXT NOT NULL REFERENCES workers(worker_id),
    benchmark_created_at TEXT NOT NULL,
    audio_path        TEXT NOT NULL,
    machine_json      TEXT NOT NULL DEFAULT '{}',
    environment_json  TEXT NOT NULL DEFAULT '{}',
    model             TEXT NOT NULL,
    backend           TEXT NOT NULL,
    language          TEXT NOT NULL DEFAULT 'en',
    audio_seconds     REAL NOT NULL,
    elapsed_seconds   REAL,
    realtime_factor   REAL,
    memory_rss_mb     REAL,
    time_to_first_token_seconds REAL,
    word_count        INTEGER,
    transcript_preview TEXT,
    error             TEXT,
    created_at        TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_worker_benchmarks_worker
    ON worker_benchmarks(worker_id, model, created_at DESC);

CREATE TABLE IF NOT EXISTS webhook_deliveries (
    delivery_id       INTEGER PRIMARY KEY AUTOINCREMENT,
    job_id            TEXT NOT NULL,
    target_url        TEXT NOT NULL,
    payload           TEXT NOT NULL,
    signature         TEXT,
    status_code       INTEGER,
    status            TEXT NOT NULL DEFAULT 'pending',
    attempts          INTEGER NOT NULL DEFAULT 0,
    max_attempts      INTEGER NOT NULL DEFAULT 5,
    next_retry_at     TEXT,
    last_error        TEXT,
    created_at        TEXT NOT NULL DEFAULT (datetime('now')),
    delivered_at      TEXT
);

CREATE INDEX IF NOT EXISTS idx_webhook_pending
    ON webhook_deliveries(status, next_retry_at)
    WHERE status IN ('pending', 'retrying');

CREATE TABLE IF NOT EXISTS job_events (
    event_id          INTEGER PRIMARY KEY AUTOINCREMENT,
    job_id            TEXT NOT NULL REFERENCES jobs(job_id),
    event_type        TEXT NOT NULL,
    from_status       TEXT,
    to_status         TEXT NOT NULL,
    actor             TEXT,
    detail            TEXT,
    created_at        TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_job_events_job_id
    ON job_events(job_id, created_at);

CREATE TABLE IF NOT EXISTS admin_users (
    admin_user_id     TEXT PRIMARY KEY,
    username          TEXT NOT NULL UNIQUE,
    display_name      TEXT NOT NULL,
    role              TEXT NOT NULL DEFAULT 'developer',
    status            TEXT NOT NULL DEFAULT 'active',
    created_at        TEXT NOT NULL DEFAULT (datetime('now')),
    last_login_at     TEXT
);

CREATE TABLE IF NOT EXISTS admin_webauthn_credentials (
    credential_id     TEXT PRIMARY KEY,
    admin_user_id     TEXT NOT NULL REFERENCES admin_users(admin_user_id),
    public_key        BLOB NOT NULL,
    sign_count        INTEGER NOT NULL DEFAULT 0,
    transports        TEXT,
    aaguid            TEXT,
    credential_device_type TEXT,
    credential_backed_up INTEGER NOT NULL DEFAULT 0,
    created_at        TEXT NOT NULL DEFAULT (datetime('now')),
    last_used_at      TEXT
);

CREATE INDEX IF NOT EXISTS idx_admin_credentials_user
    ON admin_webauthn_credentials(admin_user_id, created_at DESC);

CREATE TABLE IF NOT EXISTS admin_challenges (
    challenge_id      TEXT PRIMARY KEY,
    admin_user_id     TEXT REFERENCES admin_users(admin_user_id),
    challenge_type    TEXT NOT NULL,
    challenge         TEXT NOT NULL,
    recovery_code_id  TEXT,
    expires_at        TEXT NOT NULL,
    created_at        TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_admin_challenges_expiry
    ON admin_challenges(expires_at);

CREATE TABLE IF NOT EXISTS admin_sessions (
    session_id_hash   TEXT PRIMARY KEY,
    admin_user_id     TEXT NOT NULL REFERENCES admin_users(admin_user_id),
    created_at        TEXT NOT NULL DEFAULT (datetime('now')),
    expires_at        TEXT NOT NULL,
    last_seen_at      TEXT,
    user_agent        TEXT,
    revoked_at        TEXT
);

CREATE INDEX IF NOT EXISTS idx_admin_sessions_user
    ON admin_sessions(admin_user_id, expires_at DESC);

CREATE TABLE IF NOT EXISTS admin_audit_log (
    audit_id          INTEGER PRIMARY KEY AUTOINCREMENT,
    admin_user_id     TEXT REFERENCES admin_users(admin_user_id),
    action            TEXT NOT NULL,
    detail            TEXT,
    created_at        TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_admin_audit_created
    ON admin_audit_log(created_at DESC);

CREATE TABLE IF NOT EXISTS abuse_events (
    abuse_event_id     INTEGER PRIMARY KEY AUTOINCREMENT,
    subject_type       TEXT NOT NULL,
    subject_id         TEXT NOT NULL,
    event_type         TEXT NOT NULL,
    severity           TEXT NOT NULL DEFAULT 'medium',
    status             TEXT NOT NULL DEFAULT 'open',
    detail             TEXT,
    job_id             TEXT REFERENCES jobs(job_id),
    shard_id           TEXT REFERENCES shards(shard_id),
    reviewed_by_admin_user_id TEXT REFERENCES admin_users(admin_user_id),
    review_action      TEXT,
    resolution         TEXT,
    created_at         TEXT NOT NULL DEFAULT (datetime('now')),
    reviewed_at        TEXT
);

CREATE INDEX IF NOT EXISTS idx_abuse_events_created
    ON abuse_events(created_at DESC);

CREATE INDEX IF NOT EXISTS idx_abuse_events_subject
    ON abuse_events(subject_type, subject_id, status, created_at DESC);

CREATE TABLE IF NOT EXISTS worker_enrollment_tokens (
    token_id               TEXT PRIMARY KEY,
    token_hash             TEXT NOT NULL UNIQUE,
    label                  TEXT,
    created_by_admin_user_id TEXT REFERENCES admin_users(admin_user_id),
    created_at             TEXT NOT NULL DEFAULT (datetime('now')),
    expires_at             TEXT NOT NULL,
    max_uses               INTEGER NOT NULL DEFAULT 1,
    used_count             INTEGER NOT NULL DEFAULT 0,
    last_used_at           TEXT,
    last_used_by_worker_id TEXT,
    revoked_at             TEXT
);

CREATE INDEX IF NOT EXISTS idx_worker_enrollment_expiry
    ON worker_enrollment_tokens(expires_at DESC);

CREATE TABLE IF NOT EXISTS admin_recovery_codes (
    recovery_code_id  TEXT PRIMARY KEY,
    admin_user_id     TEXT NOT NULL REFERENCES admin_users(admin_user_id),
    code_hash         TEXT NOT NULL UNIQUE,
    created_at        TEXT NOT NULL DEFAULT (datetime('now')),
    used_at           TEXT,
    revoked_at        TEXT
);

CREATE INDEX IF NOT EXISTS idx_admin_recovery_codes_user
    ON admin_recovery_codes(admin_user_id, created_at DESC);
"""


def get_db_path() -> str:
    return get_settings().database_path


def _record_database_error(operation: str) -> None:
    try:
        from hive_compute.coordinator.metrics import increment_database_error

        increment_database_error(operation)
    except Exception:
        logger.debug("Failed to record database error metric", operation=operation)


def get_db() -> sqlite3.Connection:
    """Get a SQLite connection with WAL mode and row factory."""
    try:
        db = sqlite3.connect(get_db_path())
        db.row_factory = sqlite3.Row
        db.execute("PRAGMA journal_mode=WAL")
        db.execute("PRAGMA foreign_keys=ON")
        return db
    except sqlite3.Error:
        _record_database_error("connect")
        raise


@contextmanager
def get_db_ctx():
    """Context manager for database connections."""
    db = get_db()
    try:
        yield db
        db.commit()
    except sqlite3.Error:
        db.rollback()
        _record_database_error("transaction")
        raise
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()


def init_db():
    """Create tables and seed initial API key."""
    settings = get_settings()
    settings.validate_secrets()
    Path(settings.database_path).parent.mkdir(parents=True, exist_ok=True)

    # Run migrations BEFORE schema so indexes referencing new columns don't fail
    # on existing databases.  For a fresh DB the migration is a no-op (tables
    # don't exist yet so PRAGMA table_info returns nothing).
    _run_migrations()

    with get_db_ctx() as db:
        db.executescript(SCHEMA)

    # Seed the client API key if set, revoking older default keys so rotation is
    # authoritative instead of additive.
    if settings.client_api_key:
        key_hash = hashlib.sha256(settings.client_api_key.encode()).hexdigest()
        with get_db_ctx() as db:
            db.execute(
                """
                UPDATE api_keys
                SET revoked_at = datetime('now')
                WHERE name = 'default' AND key_hash != ? AND revoked_at IS NULL
                """,
                (key_hash,),
            )
            db.execute(
                """
                INSERT INTO api_keys (key_hash, name, revoked_at)
                VALUES (?, 'default', NULL)
                ON CONFLICT(key_hash) DO UPDATE SET
                    name = excluded.name,
                    revoked_at = NULL
                """,
                (key_hash,),
            )

    logger.info("Database initialized", path=settings.database_path)


def _run_migrations():
    """Add columns/tables that may be missing from older databases."""
    migrations = [
        ("jobs", "audio_duration", "ALTER TABLE jobs ADD COLUMN audio_duration REAL"),
        ("jobs", "shard_count", "ALTER TABLE jobs ADD COLUMN shard_count INTEGER"),
        ("jobs", "callback_url", "ALTER TABLE jobs ADD COLUMN callback_url TEXT"),
        ("shards", "canary_id", "ALTER TABLE shards ADD COLUMN canary_id TEXT REFERENCES canary_corpus(canary_id)"),
        ("jobs", "partner_id", "ALTER TABLE jobs ADD COLUMN partner_id TEXT"),
        ("jobs", "estimated_cost_cents", "ALTER TABLE jobs ADD COLUMN estimated_cost_cents INTEGER"),
        ("jobs", "idempotency_key", "ALTER TABLE jobs ADD COLUMN idempotency_key TEXT"),
        ("workers", "available_models", "ALTER TABLE workers ADD COLUMN available_models TEXT NOT NULL DEFAULT '[]'"),
        (
            "workers",
            "model_capabilities",
            "ALTER TABLE workers ADD COLUMN model_capabilities TEXT NOT NULL DEFAULT '[]'",
        ),
        ("jobs", "external_job_id", "ALTER TABLE jobs ADD COLUMN external_job_id TEXT"),
        ("admin_challenges", "recovery_code_id", "ALTER TABLE admin_challenges ADD COLUMN recovery_code_id TEXT"),
        ("jobs", "content_hash", "ALTER TABLE jobs ADD COLUMN content_hash TEXT"),
        ("jobs", "measured_duration", "ALTER TABLE jobs ADD COLUMN measured_duration REAL"),
        ("jobs", "channel_count", "ALTER TABLE jobs ADD COLUMN channel_count INTEGER"),
        ("jobs", "original_format", "ALTER TABLE jobs ADD COLUMN original_format TEXT"),
        ("jobs", "normalized_path", "ALTER TABLE jobs ADD COLUMN normalized_path TEXT"),
        ("jobs", "audio_token", "ALTER TABLE jobs ADD COLUMN audio_token TEXT"),
        ("partners", "monthly_budget_cents", "ALTER TABLE partners ADD COLUMN monthly_budget_cents INTEGER"),
        ("partners", "credits_usd", "ALTER TABLE partners ADD COLUMN credits_usd REAL"),
        (
            "worker_benchmarks",
            "environment_json",
            "ALTER TABLE worker_benchmarks ADD COLUMN environment_json TEXT NOT NULL DEFAULT '{}'",
        ),
    ]

    with get_db_ctx() as db:
        for table, column, sql in migrations:
            cols = [row[1] for row in db.execute(f"PRAGMA table_info({table})").fetchall()]
            if not cols:
                # Table doesn't exist yet (fresh DB) — SCHEMA will create it
                continue
            if column not in cols:
                db.execute(sql)
                logger.info("Migration applied", table=table, column=column)

        _migrate_worker_credentials(db)


def _migrate_worker_credentials(db: sqlite3.Connection) -> None:
    cols = [row[1] for row in db.execute("PRAGMA table_info(workers)").fetchall()]
    if not cols:
        return

    worker_migrations = [
        ("worker_token_last_four", "ALTER TABLE workers ADD COLUMN worker_token_last_four TEXT"),
        ("worker_token_issued_at", "ALTER TABLE workers ADD COLUMN worker_token_issued_at TEXT"),
        ("worker_token_revoked_at", "ALTER TABLE workers ADD COLUMN worker_token_revoked_at TEXT"),
        ("last_poll_at", "ALTER TABLE workers ADD COLUMN last_poll_at TEXT"),
        ("last_claimed_at", "ALTER TABLE workers ADD COLUMN last_claimed_at TEXT"),
        ("last_claimed_job_id", "ALTER TABLE workers ADD COLUMN last_claimed_job_id TEXT"),
        ("last_auth_failure_at", "ALTER TABLE workers ADD COLUMN last_auth_failure_at TEXT"),
        ("last_auth_failure_detail", "ALTER TABLE workers ADD COLUMN last_auth_failure_detail TEXT"),
        ("last_job_failure_at", "ALTER TABLE workers ADD COLUMN last_job_failure_at TEXT"),
        ("last_job_failure_detail", "ALTER TABLE workers ADD COLUMN last_job_failure_detail TEXT"),
        ("last_failed_job_id", "ALTER TABLE workers ADD COLUMN last_failed_job_id TEXT"),
        ("available_models", "ALTER TABLE workers ADD COLUMN available_models TEXT NOT NULL DEFAULT '[]'"),
        ("model_capabilities", "ALTER TABLE workers ADD COLUMN model_capabilities TEXT NOT NULL DEFAULT '[]'"),
        ("capabilities_advertised_at", "ALTER TABLE workers ADD COLUMN capabilities_advertised_at TEXT"),
        ("capabilities_confirmed_at", "ALTER TABLE workers ADD COLUMN capabilities_confirmed_at TEXT"),
    ]

    for column, sql in worker_migrations:
        if column not in cols:
            db.execute(sql)
            logger.info("Migration applied", table="workers", column=column)

    rows = db.execute(
        """
        SELECT worker_id, worker_token, worker_token_last_four, worker_token_issued_at, registered_at
        FROM workers
        """
    ).fetchall()
    migrated = 0

    for row in rows:
        stored_token = row["worker_token"]
        if not stored_token:
            continue

        next_token = stored_token
        if not is_hashed_worker_token(stored_token):
            next_token = hash_worker_token(stored_token)
            migrated += 1

        last_four = row["worker_token_last_four"] or stored_token[-4:]
        issued_at = row["worker_token_issued_at"] or row["registered_at"]

        db.execute(
            """
            UPDATE workers
            SET worker_token = ?, worker_token_last_four = ?, worker_token_issued_at = ?
            WHERE worker_id = ?
            """,
            (next_token, last_four, issued_at, row["worker_id"]),
        )

    if migrated:
        logger.info("Worker credential hashes migrated", workers=migrated)


def record_event(
    db: sqlite3.Connection,
    job_id: str,
    event_type: str,
    to_status: str,
    from_status: str | None = None,
    actor: str | None = None,
    detail: str | None = None,
) -> None:
    """Append a row to the job_events audit trail."""
    db.execute(
        """
        INSERT INTO job_events (job_id, event_type, from_status, to_status, actor, detail)
        VALUES (?, ?, ?, ?, ?, ?)
        """,
        (job_id, event_type, from_status, to_status, actor, detail),
    )
