"""Run ledger (db.RunLedger) — the observability columns + real durability (P1a)."""

from __future__ import annotations

from lawn_estimator.db import (
    ApiKeyStore,
    AuditLog,
    ConfigStore,
    JobStore,
    LeadStore,
    MeasurementEditStore,
    RunLedger,
    TenantStore,
    UserStore,
    connect,
)
from lawn_estimator.metering import SINGLE


def test_record_stores_the_observability_columns():
    conn = connect(":memory:")
    RunLedger(conn).record(
        "acme", "5534 Mayberry St", SINGLE,
        zip="68106", method="lidar+rgb", confidence="high", duration_s=21.4, result_id="abc123",
    )
    row = conn.execute("SELECT * FROM runs").fetchone()
    assert row["tenant_id"] == "acme"
    assert row["surface"] == SINGLE
    assert row["zip"] == "68106"
    assert row["method"] == "lidar+rgb"
    assert row["confidence"] == "high"
    assert row["duration_s"] == 21.4
    assert row["result_id"] == "abc123"
    assert row["billable"] == 1
    assert row["outcome"] == "ok"   # set explicitly (not via the column default)
    assert row["created_at"]  # server-stamped


def test_record_sets_outcome_ok_without_a_column_default():
    # Reproduce the migrated-Neon condition: `outcome` exists but has NO default (added via
    # ADD COLUMN). record() must write 'ok' explicitly, else `ok` stays 0 on prod forever.
    db = connect(":memory:")
    db.execute("DROP TABLE runs")
    db.execute(
        "CREATE TABLE runs (id INTEGER PRIMARY KEY AUTOINCREMENT, tenant_id TEXT, surface TEXT, "
        "address TEXT, address_norm TEXT, zip TEXT, billable INTEGER, method TEXT, confidence TEXT, "
        "lawn_sqft REAL, duration_s REAL, result_id TEXT, outcome TEXT, error_class TEXT, created_at TEXT)"
    )
    db.commit()
    RunLedger(db).record("acme", "1 A St", "single", result_id="r1")
    row = db.execute("SELECT outcome FROM runs WHERE result_id = 'r1'").fetchone()
    assert row["outcome"] == "ok"   # explicit — there is no default to fall back on


def test_ensure_columns_backfills_legacy_null_outcome_as_ok():
    # A pre-observability row (outcome NULL) counts in total but is invisible to ok — the
    # `total=2, ok=0` gap. Re-running the migration backfills it to 'ok'.
    db = connect(":memory:")
    led = RunLedger(db)
    # Recreate runs with a NULLABLE outcome (the migrated Neon column has no NOT NULL / default),
    # then insert a pre-observability completed quote as outcome=NULL.
    db.execute("DROP TABLE runs")
    db.execute(
        "CREATE TABLE runs (id INTEGER PRIMARY KEY AUTOINCREMENT, tenant_id TEXT, surface TEXT, "
        "address TEXT, address_norm TEXT, zip TEXT, billable INTEGER, method TEXT, confidence TEXT, "
        "lawn_sqft REAL, duration_s REAL, result_id TEXT, outcome TEXT, error_class TEXT, "
        "created_at TEXT DEFAULT '2020-01-01T00:00:00Z')"
    )
    db.execute("INSERT INTO runs (tenant_id, surface, address, address_norm, billable, outcome) "
               "VALUES ('acme', 'batch', '1 A St', '1 a st', 1, NULL)")
    db.commit()
    assert led.stats("acme")["total"] == 1 and led.stats("acme")["ok"] == 0   # the gap
    db._ensure_columns()                                     # re-running the migration backfills
    s = led.stats("acme")
    assert s["total"] == 1 and s["ok"] == 1                  # now counted as ok


def test_usage_survives_a_new_connection_to_the_same_file(tmp_path):
    db = str(tmp_path / "app.db")
    RunLedger(connect(db)).record("acme", "123 A St", SINGLE)
    # A brand-new ledger over the same file (i.e. a "redeploy") still sees the count.
    assert RunLedger(connect(db)).usage("acme") == 1


def test_run_ledger_records_failures_and_stats():
    led = RunLedger(connect(":memory:"))
    led.record("acme", "1 A St", "single", duration_s=20.0)   # ok
    led.record("acme", "2 B St", "batch", duration_s=30.0)    # ok
    led.record_failure("acme", "bad addr", "single", error_class="ValueError", outcome="rejected")
    led.record_failure("acme", "3 C St", "batch", error_class="RuntimeError", outcome="error")
    s = led.stats("acme")
    assert s["total"] == 4 and s["ok"] == 2 and s["rejected"] == 1 and s["errors"] == 1
    assert s["failure_rate"] == 0.25          # errors / total
    assert s["avg_duration_s"] == 25.0        # avg over the two ok runs (failures excluded)
    assert RunLedger(connect(":memory:")).stats("nobody")["total"] == 0


def test_stats_by_tenant_is_cross_tenant():
    led = RunLedger(connect(":memory:"))
    led.record("acme", "1", "batch", duration_s=20.0)
    led.record("acme", "2", "single", duration_s=30.0)
    led.record_failure("acme", "x", "single", error_class="ValueError", outcome="rejected")
    led.record("beta", "9", "batch", duration_s=10.0)
    rows = led.stats_by_tenant()
    by = {r["tenant"]: r for r in rows}
    assert by["acme"]["total"] == 3 and by["acme"]["ok"] == 2 and by["acme"]["rejected"] == 1
    assert by["acme"]["billable"] == 2         # 2 billable ok runs; the rejected one isn't
    assert by["beta"]["total"] == 1
    assert rows[0]["tenant"] == "acme"         # most active first


def test_run_ledger_stores_lawn_sqft_and_fetches_by_result_id():
    led = RunLedger(connect(":memory:"))
    led.record("acme", "1 A St", "batch", lawn_sqft=4200.0, result_id="r1", zip="68106")
    row = led.get_by_result_id("acme", "r1")
    assert row["lawn_sqft"] == 4200.0 and row["surface"] == "batch" and row["zip"] == "68106"
    assert led.get_by_result_id("acme", "nope") is None
    assert led.get_by_result_id("beta", "r1") is None      # tenant-scoped


def test_measurement_edit_store():
    st = MeasurementEditStore(connect(":memory:"))
    st.record("acme", "r1", 4500.0, original_sqft=4000.0, address="1 A St", user_id=1)
    got = st.recent("acme")
    assert got[0]["edited_sqft"] == 4500.0 and got[0]["original_sqft"] == 4000.0
    assert st.recent("beta") == []


def test_lead_store_records_and_lists_newest_first():
    store = LeadStore(connect(":memory:"))
    store.record("acme", "widget", "1 A St", name="Sam", email="s@x.com", phone="402-555-0100")
    store.record("acme", "quote", "2 B St", email="b@x.com", zip="68106",
                 lawn_sqft=4200.0, price_total=90.0, confidence="high", result_id="r1")
    store.record("beta", "widget", "9 Z St", email="z@x.com")   # other tenant, filtered out

    leads = store.recent("acme")
    assert [lead["address"] for lead in leads] == ["2 B St", "1 A St"]   # newest first
    quote_lead = leads[0]
    assert quote_lead["source"] == "quote"
    assert quote_lead["lawn_sqft"] == 4200.0
    assert quote_lead["price_total"] == 90.0
    assert quote_lead["result_id"] == "r1"
    widget_lead = leads[1]
    assert widget_lead["phone"] == "402-555-0100"
    assert widget_lead["lawn_sqft"] is None   # captured before measurement
    assert store.recent("gamma") == []        # unknown tenant


def test_tenant_store_roundtrips_and_upserts():
    store = TenantStore(connect(":memory:"))
    assert store.get("acme") is None                       # never stored
    store.put("acme", {"company": "Acme", "currency": "USD"})
    assert store.get("acme") == {"company": "Acme", "currency": "USD"}
    store.put("acme", {"company": "Acme Renamed"})          # upsert replaces
    assert store.get("acme") == {"company": "Acme Renamed"}
    assert store.all_ids() == ["acme"]


def test_user_store_maps_clerk_user_to_tenant_and_role():
    users = UserStore(connect(":memory:"))
    uid = users.create("acme", "clerk_abc", "owner@acme.com", "owner", status="active")
    got = users.by_clerk_id("clerk_abc")
    assert got["tenant_id"] == "acme" and got["role"] == "owner" and got["email"] == "owner@acme.com"
    assert users.by_clerk_id("clerk_nope") is None          # unprovisioned Clerk user → denied
    users.set_role(uid, "staff")
    assert users.by_clerk_id("clerk_abc")["role"] == "staff"
    users.create("acme", "clerk_def", "staff@acme.com", "staff")
    assert {u["email"] for u in users.by_tenant("acme")} == {"owner@acme.com", "staff@acme.com"}


def test_audit_log_is_append_only_scoped_and_newest_first():
    audit = AuditLog(connect(":memory:"))
    audit.record("config.publish", tenant_id="acme", user_id=1,
                 before={"rate": 5}, after={"rate": 6}, ip="1.2.3.4")
    audit.record("user.invite", tenant_id="acme", user_id=1, target_type="user", target_id="7")
    audit.record("config.publish", tenant_id="beta", user_id=9)   # other tenant

    rows = audit.recent("acme")
    assert [r["action"] for r in rows] == ["user.invite", "config.publish"]   # newest first
    assert '"rate": 6' in rows[1]["after"] and rows[1]["ip"] == "1.2.3.4"
    assert audit.recent("beta")[0]["action"] == "config.publish"              # scoped


def test_config_store_drafts_and_versions():
    cs = ConfigStore(connect(":memory:"))
    assert cs.get_draft("acme") is None
    cs.save_draft("acme", {"company": "Acme", "rate": 5}, user_id=1)
    assert cs.get_draft("acme") == {"company": "Acme", "rate": 5}
    cs.save_draft("acme", {"company": "Acme", "rate": 6})               # upsert
    assert cs.get_draft("acme")["rate"] == 6
    cs.add_version("acme", {"v": 1})
    cs.add_version("acme", {"v": 2})
    versions = cs.versions("acme")
    assert [v["config"]["v"] for v in versions] == [2, 1]               # newest first
    cs.clear_draft("acme")
    assert cs.get_draft("acme") is None


class _Dropped(Exception):
    """Stand-in for psycopg.OperationalError/InterfaceError (psycopg isn't installed locally)."""


def _fake_pg_retry(db):
    """Retrofit a real SQLite Database into the postgres retry path with fake connections."""
    db.is_postgres = True
    db._conn_errors = (_Dropped,)


def test_execute_reconnects_once_on_a_dropped_connection():
    # Neon drops idle connections; a dropped conn must reconnect + retry, not surface a 500.
    db = connect(":memory:")
    _fake_pg_retry(db)
    calls = {"n": 0}

    class DeadConn:
        def execute(self, sql, params):
            calls["n"] += 1
            raise _Dropped("server closed the connection unexpectedly")

    class LiveConn:
        def execute(self, sql, params):
            calls["n"] += 1
            return "ok-cursor"

    db._conn = DeadConn()
    db._connect_pg = lambda: setattr(db, "_conn", LiveConn())   # "reconnect" swaps in a live conn

    assert db.execute("SELECT 1") == "ok-cursor"   # retried on the reconnected connection
    assert calls["n"] == 2                          # exactly one failed attempt + one retry


def test_execute_retries_only_once_then_propagates():
    # A real outage (still dead after reconnect) must raise, not loop forever.
    import pytest

    db = connect(":memory:")
    _fake_pg_retry(db)
    calls = {"n": 0}

    class DeadConn:
        def execute(self, sql, params):
            calls["n"] += 1
            raise _Dropped("still down")

    db._conn = DeadConn()
    db._connect_pg = lambda: setattr(db, "_conn", DeadConn())   # reconnect, but still dead

    with pytest.raises(_Dropped):
        db.execute("SELECT 1")
    assert calls["n"] == 2   # original + exactly one retry, then propagate


def test_user_store_invite_activate_and_caps():
    users = UserStore(connect(":memory:"))
    users.create("acme", "clerk_owner", "owner@acme.com", "owner", status="active")
    # A pending invite has no Clerk id and is found by email until it's accepted.
    iid = users.invite("acme", "Staff@Acme.com", "staff", invited_by=1)
    assert users.by_clerk_id("clerk_new") is None
    pending = users.pending_by_email("staff@acme.com")   # case-insensitive
    assert pending["id"] == iid and pending["role"] == "staff" and pending["clerk_user_id"] is None
    # First sign-in binds the Clerk id + activates it.
    users.activate(iid, "clerk_new")
    assert users.pending_by_email("staff@acme.com") is None            # no longer pending
    assert users.by_clerk_id("clerk_new")["status"] == "active"
    # Seats count active + pending (not disabled); owners are counted for the last-owner guard.
    assert users.seat_count("acme") == 2 and users.active_owner_count("acme") == 1
    users.set_status(iid, "disabled")
    assert users.seat_count("acme") == 1


def test_job_store_persist_resume_scope_and_evict():
    jobs = JobStore(connect(":memory:"))
    jobs.create("j1", client="acme", imagery="google", addresses=["1 A St", "2 B St", "3 C St"])

    # Freshly created -> queued, worker sees it with completed=0 and decoded addresses.
    assert jobs.pending_ids() == ["j1"]
    w = jobs.for_worker("j1")
    assert w["client"] == "acme" and w["imagery"] == "google" and w["completed"] == 0
    assert w["addresses"] == ["1 A St", "2 B St", "3 C St"]

    # Worker runs the first address, then the app "restarts" mid-batch (nothing lost).
    jobs.mark_running("j1")
    jobs.append_result("j1", 1, {"address": "1 A St", "status": "ok", "result_id": "r1"})
    assert jobs.pending_ids() == ["j1"]                    # still resumable after a restart
    assert jobs.for_worker("j1")["completed"] == 1        # resumes from index 1, not 0

    # Finish the batch; the poll view is scoped to the owning client.
    jobs.append_result("j1", 2, {"address": "2 B St", "status": "ok", "result_id": "r2"})
    jobs.append_result("j1", 3, {"address": "3 C St", "status": "error"})
    jobs.mark_done("j1")
    assert jobs.pending_ids() == []
    view = jobs.get("j1", "acme")
    assert view["status"] == "done" and view["total"] == 3 and view["completed"] == 3
    assert [r.get("result_id") for r in view["results"]] == ["r1", "r2", None]
    assert jobs.get("j1", "someone-else") is None         # a job never leaks across clients

    # Idempotent replay of an index overwrites rather than duplicating.
    jobs.append_result("j1", 2, {"address": "2 B St", "status": "ok", "result_id": "r2b"})
    assert jobs.get("j1", "acme")["results"][1]["result_id"] == "r2b"

    # Eviction returns the result_ids to forget and only removes DONE jobs beyond `keep`.
    jobs.create("j2", client="acme", imagery="google", addresses=["x"])   # queued, must survive
    forget = jobs.evict_old(keep=0)
    assert set(forget) == {"r1", "r2b"}                   # None result_ids are skipped
    assert jobs.get("j1", "acme") is None                 # the done job is gone
    assert jobs.pending_ids() == ["j2"]                   # the queued job is untouched


def test_run_ledger_overview_aggregates():
    led = RunLedger(connect(":memory:"))
    led.record("acme", "1 A St", "single", zip="68106", duration_s=20.0, lawn_sqft=4000.0)
    led.record("acme", "2 B St", "batch", zip="68106", duration_s=40.0, lawn_sqft=6000.0)
    led.record_failure("acme", "x", "single", error_class="ValueError", outcome="rejected", zip="68164")
    led.record("beta", "9 Z", "batch", zip="68007")   # other tenant — excluded
    ov = led.overview("acme", days=30)
    assert ov["quotes"] == 3 and ov["ok"] == 2 and ov["rejected"] == 1 and ov["errors"] == 0
    assert ov["by_surface"] == {"single": 2, "batch": 1}
    assert ov["avg_duration_s"] == 30.0 and ov["avg_lawn_sqft"] == 5000   # ok runs only
    assert ov["top_zips"][0] == {"zip": "68106", "count": 2}
    assert len(ov["volume_series"]) == 30
    assert ov["volume_series"][-1]["count"] == 3 and sum(p["count"] for p in ov["volume_series"]) == 3
    assert RunLedger(connect(":memory:")).overview("nobody")["quotes"] == 0


def test_api_key_store_issue_and_lookup():
    store = ApiKeyStore(connect(":memory:"))
    store.issue("acme", "pub-abc", kind="publishable", label="acme")
    store.issue("acme", "sec-xyz", kind="secret")
    rows = store.all()
    assert {"pub-abc", "sec-xyz"} <= {r["api_key"] for r in rows}
    assert store.key_for_tenant("acme", "publishable") == "pub-abc"
    assert store.key_for_tenant("acme", "secret") == "sec-xyz"
    assert store.key_for_tenant("nobody") is None


def test_lead_store_summary_conversion():
    store = LeadStore(connect(":memory:"))
    store.record("acme", "widget", "1 A", email="a@x")
    store.record("acme", "booking", "1 A", price_total=90.0, result_id="r1")
    store.record("acme", "booking", "2 B", price_total=110.0, result_id="r2")
    store.record("acme", "batch", "3 C", price_total=200.0)   # operator quote — NOT a customer lead
    store.record("beta", "widget", "9 Z")   # other tenant
    s = store.summary("acme", days=30)
    assert s["leads"] == 3 and s["bookings"] == 2 and s["conversion"] == round(2 / 3, 3)  # batch excluded
    assert s["by_source"] == {"widget": 1, "booking": 2} and "batch" not in s["by_source"]
    assert s["avg_price"] == 100.0    # avg of BOOKED prices (90, 110); the batch 200 is ignored


def test_lead_store_summary_top_packages():
    import json
    store = LeadStore(connect(":memory:"))
    spring = json.dumps({"bundle": {"id": "spring", "name": "Spring Package", "cadence": "seasonal"}})
    store.record("acme", "booking", "1 A", price_total=90.0, result_id="r1", detail=spring)
    store.record("acme", "booking", "2 B", price_total=90.0, result_id="r2", detail=spring)
    store.record("acme", "booking", "3 C", price_total=50.0, result_id="r3")   # à-la-carte, no bundle
    top = store.summary("acme", days=30)["top_packages"]
    assert top and top[0] == {"name": "Spring Package", "count": 2}