"""Run slug normalization, migration, and collision-safe allocation."""

from __future__ import annotations

import hashlib
import sqlite3
import threading
from pathlib import Path

import pytest

from adw_modules.run_slug import (
    derive_slug_base,
    normalize_slug_segment,
    pick_unique_slug,
    slug_base_from_hint,
    workflow_slug_fallback,
)
from adw_modules.tracer import (
    RUN_SLUG_INDEX_NAME,
    RUN_SLUG_UNIQUE_INDEX,
    Tracer,
    connect_db,
    verify_run_slug_index,
)
from adw_modules.utils import classify_slug_hint


def test_normalize_slug_segment_strips_punctuation_and_case() -> None:
    assert normalize_slug_segment("Hello, World!!") == "hello-world"
    assert normalize_slug_segment("---a--b---") == "a-b"


def test_slug_base_from_existing_plan_file(tmp_path: Path) -> None:
    hint = "docs/specs/2026-08-09-incident-dispatch-selector-options-design.md"
    path = tmp_path / hint
    path.parent.mkdir(parents=True)
    path.write_text("# spec\n")
    classified = classify_slug_hint(str(path))
    assert classified == (str(path), True)
    assert slug_base_from_hint(str(path), is_path=True) == (
        "incident-dispatch-selector-options-design"
    )


def test_slug_base_from_dotted_inline_text() -> None:
    assert slug_base_from_hint("Fix src/auth.ts and add tests", is_path=False) == (
        "fix-src-auth-ts-and-add-tests"
    )


def test_slug_base_from_inline_text() -> None:
    assert slug_base_from_hint("Add Health Endpoint ASAP!!!", is_path=False) == (
        "add-health-endpoint-asap"
    )


def test_slug_base_from_nonexistent_path_like_text() -> None:
    classified = classify_slug_hint("docs/specs/missing-file.md")
    assert classified == ("docs/specs/missing-file.md", False)
    assert slug_base_from_hint("docs/specs/missing-file.md", is_path=False) == (
        "docs-specs-missing-file-md"
    )


def test_slug_base_from_empty_hint_is_none() -> None:
    assert slug_base_from_hint("") is None
    assert slug_base_from_hint("   ") is None
    assert slug_base_from_hint("!!!", is_path=False) is None
    assert classify_slug_hint("") is None
    assert classify_slug_hint("   ") is None


def test_workflow_fallback_includes_adw_id() -> None:
    slug = workflow_slug_fallback("adw_plan", "abc12345")
    assert slug == "plan-abc12345"
    assert derive_slug_base(None, "adw_plan", "deadbeef") == "plan-deadbeef"


def test_workflow_fallback_uses_digest_for_punctuation_only_id() -> None:
    adw_id = "!!!"
    expected = f"plan-{hashlib.sha256(adw_id.encode()).hexdigest()[:8]}"
    assert workflow_slug_fallback("adw_plan", adw_id) == expected
    assert derive_slug_base(None, "adw_plan", adw_id) == expected


def test_workflow_fallback_uses_digest_for_unicode_only_id() -> None:
    adw_id = "日本語"
    expected = f"plan-{hashlib.sha256(adw_id.encode()).hexdigest()[:8]}"
    assert workflow_slug_fallback("adw_plan", adw_id) == expected


def test_fresh_db_migrates_run_slug_column_and_unique_index(tmp_path: Path) -> None:
    tracer = Tracer(tmp_path / "fresh.db", tmp_path / "events.jsonl")
    try:
        columns = {row[1] for row in tracer.conn.execute("PRAGMA table_info(sessions)")}
        assert "run_slug" in columns
        verify_run_slug_index(tracer.conn)
    finally:
        tracer.conn.close()


def test_run_slug_index_is_unique_and_partial(tmp_path: Path) -> None:
    tracer = Tracer(tmp_path / "fresh.db", tmp_path / "events.jsonl")
    try:
        row = tracer.conn.execute(
            "SELECT name, [unique], partial FROM pragma_index_list('sessions')"
            f" WHERE name='{RUN_SLUG_INDEX_NAME}'",
        ).fetchone()
        assert row is not None
        assert row[1] == 1
        assert row[2] == 1
        sql = tracer.conn.execute(
            "SELECT sql FROM sqlite_master WHERE type='index' AND name=?",
            (RUN_SLUG_INDEX_NAME,),
        ).fetchone()[0].lower()
        assert "run_slug is not null" in sql
    finally:
        tracer.conn.close()


def test_incompatible_existing_run_slug_index_fails_closed(tmp_path: Path) -> None:
    db = tmp_path / "bad-index.db"
    conn = sqlite3.connect(db, isolation_level=None)
    try:
        conn.executescript(
            "CREATE TABLE sessions (adw_id TEXT PRIMARY KEY, run_slug TEXT);"
            "CREATE INDEX idx_sessions_run_slug ON sessions(run_slug);"
        )
    finally:
        conn.close()

    with pytest.raises(RuntimeError, match="not unique"):
        Tracer(db, tmp_path / "events.jsonl")


def test_sequential_collision_allocation(tmp_path: Path) -> None:
    db = tmp_path / "sssf.db"
    tracer = Tracer(db, tmp_path / "events.jsonl")
    try:
        base = "incident-dispatch-selector-options-design"
        tracer.session_start("run1", "alice", slug_hint=base, adw_script_stem="adw_plan")
        tracer.session_start("run2", "bob", slug_hint=base, adw_script_stem="adw_plan")
        tracer.session_start("run3", "carol", slug_hint=base, adw_script_stem="adw_plan")
        rows = tracer.conn.execute(
            "SELECT adw_id, run_slug FROM sessions ORDER BY adw_id",
        ).fetchall()
    finally:
        tracer.conn.close()

    assert rows == [
        ("run1", base),
        ("run2", f"{base}-2"),
        ("run3", f"{base}-3"),
    ]


def test_concurrent_collision_allocation(tmp_path: Path) -> None:
    db = tmp_path / "sssf.db"
    base = "shared-base"
    barrier = threading.Barrier(3)
    errors: list[BaseException] = []
    allocated: list[str] = []
    lock = threading.Lock()

    def start_session(adw_id: str) -> None:
        try:
            barrier.wait(timeout=10)
            tracer = Tracer(db, tmp_path / f"{adw_id}.jsonl")
            tracer.session_start(adw_id, "tester", slug_hint=base, adw_script_stem="adw_plan")
            row = tracer.conn.execute(
                "SELECT run_slug FROM sessions WHERE adw_id=?", (adw_id,),
            ).fetchone()
            tracer.conn.close()
            with lock:
                allocated.append(row[0])
        except BaseException as exc:
            errors.append(exc)

    threads = [threading.Thread(target=start_session, args=(f"run{i}",)) for i in range(3)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join(timeout=30)

    assert not errors
    assert sorted(allocated) == sorted([base, f"{base}-2", f"{base}-3"])


def test_join_preserves_existing_run_slug(tmp_path: Path) -> None:
    db = tmp_path / "sssf.db"
    tracer = Tracer(db, tmp_path / "events.jsonl")
    try:
        tracer.session_start(
            "pinned", "alice", adw_name="adw_plan",
            slug_hint="first-topic", adw_script_stem="adw_plan",
        )
        tracer.session_start(
            "pinned", "alice", adw_name="adw_build", slug_hint="other-topic",
            adw_script_stem="adw_build",
        )
        row = tracer.conn.execute(
            "SELECT run_slug, adw_name FROM sessions WHERE adw_id='pinned'",
        ).fetchone()
    finally:
        tracer.conn.close()

    assert row == ("first-topic", "adw_plan + adw_build")


def test_legacy_session_without_slug_stays_null(tmp_path: Path) -> None:
    db = tmp_path / "legacy.db"
    conn = sqlite3.connect(db, isolation_level=None)
    try:
        conn.executescript(
            "CREATE TABLE sessions ("
            "adw_id TEXT PRIMARY KEY, status TEXT, engineer TEXT, started_at TEXT)"
        )
        conn.execute(
            "INSERT INTO sessions VALUES (?, ?, ?, ?)",
            ("legacy", "success", "alice", "2026-01-01T00:00:00Z"),
        )
    finally:
        conn.close()

    tracer = Tracer(db, tmp_path / "events.jsonl")
    try:
        verify_run_slug_index(tracer.conn)
        tracer.session_start("legacy", "alice", slug_hint="new-hint", adw_script_stem="adw_plan")
        row = tracer.conn.execute(
            "SELECT run_slug FROM sessions WHERE adw_id='legacy'",
        ).fetchone()
    finally:
        tracer.conn.close()

    assert row == (None,)


def test_pick_unique_slug_respects_existing_rows(tmp_path: Path) -> None:
    conn = connect_db(tmp_path / "slug.db")
    try:
        conn.executescript(
            "CREATE TABLE sessions (adw_id TEXT PRIMARY KEY, run_slug TEXT);"
            + RUN_SLUG_UNIQUE_INDEX
        )
        conn.execute(
            "INSERT INTO sessions (adw_id, run_slug) VALUES (?, ?)",
            ("existing", "collision-base"),
        )
        conn.execute("BEGIN IMMEDIATE")
        slug = pick_unique_slug(conn, "collision-base")
        conn.execute("COMMIT")
    finally:
        conn.close()

    assert slug == "collision-base-2"