"""FlowDeck — v7.0.0 Automations v2 (steps) + Workers lite.

Covers migration 26, steps CRUD + validation + auth, trigger modes any/all,
chained actions with interpolation, condition/delay steps, new actions
(slack/email/forge_issue/agent_trigger, secret encryption), native DB button,
legacy no-double-run compat, workers CRUD/run/sandbox/budget/fork/usage/cron.
"""
from __future__ import annotations

import secrets

import pytest
from conftest import anon, anon_csrf

from app.db import get_conn
from app.services import automations as auto_svc

# ── helpers ────────────────────────────────────────────────────────────────

def _login(client):
    from app.auth.session import SessionManager
    login = f"v70_{secrets.token_hex(4)}"
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO users (login, full_name, email, is_admin) VALUES (?, 'V70', ?, 0)",
            (login, f"{login}@test.com"),
        )
        uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
        conn.commit()
    session = SessionManager.create_session({"id": uid, "login": login})
    return session, uid


def _cookies(session):
    return {"flowdeck_session": session}


def _mkcollection(client, name="Tasks"):
    r = client.post("/db/api", json={"name": name,
                                     "schema": [{"name": "Status", "type": "text"}]})
    assert r.status_code == 200, r.text
    return r.json()["id"]


def _mkrow(client, cid, title="Row"):
    r = client.post(f"/db/{cid}/pages/api", json={"title": title})
    assert r.status_code == 200, r.text
    return r.json()["id"]


def _mkauto(client, session, **kw):
    body = {"name": "Auto", "trigger_type": "event", "event": "page.created",
            "actions": []}
    body.update(kw)
    r = client.post("/workspace/automations", json=body, cookies=_cookies(session))
    assert r.status_code == 200, r.text
    return r.json()["id"]


def _add_step(client, session, aid, kind, config, position=None):
    body = {"kind": kind, "config": config}
    if position is not None:
        body["position"] = position
    r = client.post(f"/workspace/automations/{aid}/steps", json=body,
                    cookies=_cookies(session))
    assert r.status_code == 200, r.text
    return r.json()["id"]


def _runs(aid):
    with get_conn() as conn:
        return conn.execute(
            "SELECT * FROM automation_runs WHERE automation_id=? ORDER BY id", (aid,)
        ).fetchall()


# ── migration ──────────────────────────────────────────────────────────────

def test_migration_26_tables(client):
    with get_conn() as conn:
        tables = {r[0] for r in conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
    for t in ("automation_steps", "workers", "worker_runs"):
        assert t in tables
    with get_conn() as conn:
        auto_cols = {r[1] for r in conn.execute("PRAGMA table_info(automations)").fetchall()}
        prop_cols = {r[1] for r in conn.execute(
            "PRAGMA table_info(collection_properties)").fetchall()}
    assert "trigger_mode" in auto_cols
    assert "button_automation_id" in prop_cols
    with get_conn() as conn:
        v = conn.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]
    assert v >= 26


# ── steps CRUD ─────────────────────────────────────────────────────────────

def test_steps_crud_and_order(client):
    session, _ = _login(client)
    aid = _mkauto(client, session)
    s1 = _add_step(client, session, aid, "trigger", {"event": "page.created"}, position=0)
    s2 = _add_step(client, session, aid, "action",
                   {"type": "notify", "message": "hi"}, position=1)
    r = client.get(f"/workspace/automations/{aid}/steps", cookies=_cookies(session))
    assert [s["id"] for s in r.json()["steps"]] == [s1, s2]
    r = client.put(f"/workspace/automations/steps/{s2}",
                   json={"kind": "action", "config": {"type": "notify", "message": "yo"}},
                   cookies=_cookies(session))
    assert r.status_code == 200
    r = client.delete(f"/workspace/automations/steps/{s2}", cookies=_cookies(session))
    assert r.status_code == 200
    r = client.get(f"/workspace/automations/{aid}/steps", cookies=_cookies(session))
    assert len(r.json()["steps"]) == 1


def test_steps_validation_and_auth(client):
    anon_csrf(client)
    session, _ = _login(client)
    aid = _mkauto(client, session)
    r = client.post(f"/workspace/automations/{aid}/steps",
                    json={"kind": "nope", "config": {}}, cookies=_cookies(session))
    assert r.status_code == 400
    r = client.post(f"/workspace/automations/{aid}/steps",
                    json={"kind": "action", "config": {"type": "nope"}},
                    cookies=_cookies(session))
    assert r.status_code == 400
    r = client.post(f"/workspace/automations/{aid}/steps",
                    json={"kind": "trigger", "config": {}})
    assert r.status_code == 401
    r = client.post("/workspace/automations/999999/steps",
                    json={"kind": "trigger", "config": {"event": "x"}},
                    cookies=_cookies(session))
    assert r.status_code == 404


def test_trigger_mode_endpoint(client):
    session, _ = _login(client)
    aid = _mkauto(client, session)
    r = client.put(f"/workspace/automations/{aid}/mode", json={"mode": "all"},
                   cookies=_cookies(session))
    assert r.json()["trigger_mode"] == "all"
    r = client.put(f"/workspace/automations/{aid}/mode", json={"mode": "sometimes"},
                   cookies=_cookies(session))
    assert r.status_code == 400


# ── multi-trigger any/all ──────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_mode_any_two_triggers(client):
    auto_svc.reset_all_pending()
    session, _ = _login(client)
    aid = _mkauto(client, session)
    _add_step(client, session, aid, "trigger", {"event": "page.created"})
    _add_step(client, session, aid, "trigger", {"event": "form.submitted"})
    _add_step(client, session, aid, "action", {"type": "notify", "message": "fired"})
    await auto_svc.fire_event("page.created", {"collection_id": 0})
    await auto_svc.fire_event("form.submitted", {"collection_id": 0})
    assert len([r for r in _runs(aid) if r["status"] == "fired"]) == 2


@pytest.mark.asyncio
async def test_mode_all_needs_every_trigger(client):
    auto_svc.reset_all_pending()
    session, _ = _login(client)
    aid = _mkauto(client, session)
    client.put(f"/workspace/automations/{aid}/mode", json={"mode": "all"},
               cookies=_cookies(session))
    _add_step(client, session, aid, "trigger", {"event": "page.created"})
    _add_step(client, session, aid, "trigger", {"event": "form.submitted"})
    _add_step(client, session, aid, "action", {"type": "notify", "message": "both"})
    await auto_svc.fire_event("page.created", {"collection_id": 0})
    assert _runs(aid) == []
    await auto_svc.fire_event("form.submitted", {"collection_id": 0})
    assert len([r for r in _runs(aid) if r["status"] == "fired"]) == 1


@pytest.mark.asyncio
async def test_form_submitted_trigger_end_to_end(client):
    auto_svc.reset_all_pending()
    session, _ = _login(client)
    aid = _mkauto(client, session)
    _add_step(client, session, aid, "trigger", {"event": "form.submitted"})
    _add_step(client, session, aid, "action", {"type": "notify", "message": "form in"})
    await auto_svc.fire_event("form.submitted", {"collection_id": 0, "row_id": 5})
    runs = _runs(aid)
    assert len(runs) == 1 and runs[0]["status"] == "fired"


# ── chains: order, interpolation, conditions, delay ────────────────────────

@pytest.mark.asyncio
async def test_chained_actions_interpolation(client):
    session, uid = _login(client)
    cid = _mkcollection(client)
    aid = _mkauto(client, session)
    _add_step(client, session, aid, "trigger", {"event": "page.created"})
    _add_step(client, session, aid, "action",
              {"type": "create_page", "collection_id": cid, "title": "Copy of {{title}}"})
    _add_step(client, session, aid, "action",
              {"type": "notify", "message": "made [[Copy of {{title}}]]"})
    res = await auto_svc.run_automation(aid, "manual",
                                        {"collection_id": cid, "title": "Alpha"})
    assert res["status"] == "fired"
    with get_conn() as conn:
        row = conn.execute("SELECT title FROM collection_pages WHERE collection_id=?",
                           (cid,)).fetchone()
    assert row and row["title"] == "Copy of Alpha"


@pytest.mark.asyncio
async def test_condition_step_blocks(client):
    session, _ = _login(client)
    cid = _mkcollection(client)
    aid = _mkauto(client, session)
    _add_step(client, session, aid, "trigger", {"event": "page.created"})
    _add_step(client, session, aid, "condition",
              {"property": "Status", "op": "eq", "value": "Done"})
    _add_step(client, session, aid, "action", {"type": "notify", "message": "x"})
    res = await auto_svc.run_automation(
        aid, "event", {"collection_id": cid, "properties": {"Status": "Todo"}})
    assert res["status"] == "skipped"
    assert _runs(aid)[0]["status"] == "skipped"
    res = await auto_svc.run_automation(
        aid, "event", {"collection_id": cid, "properties": {"Status": "Done"}})
    assert res["status"] == "fired"


@pytest.mark.asyncio
async def test_delay_step(client):
    import time as _t
    session, _ = _login(client)
    aid = _mkauto(client, session)
    _add_step(client, session, aid, "action", {"type": "notify", "message": "a"})
    _add_step(client, session, aid, "delay", {"seconds": 1})
    _add_step(client, session, aid, "action", {"type": "notify", "message": "b"})
    start = _t.time()
    res = await auto_svc.run_automation(aid, "manual", {})
    assert res["status"] == "fired"
    assert _t.time() - start >= 1.0
    assert "delay 1s" in res["detail"]


# ── new actions ────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_slack_action(client, monkeypatch):
    session, _ = _login(client)
    aid = _mkauto(client, session)
    seen = {}
    async def fake(url, text):
        seen["url"] = url
        seen["text"] = text
        return "slack → (200)"
    monkeypatch.setattr(auto_svc, "_post_slack", fake)
    _add_step(client, session, aid, "action",
              {"type": "slack", "webhook_url": "https://hooks.test/x", "text": "Hi {{title}}"})
    res = await auto_svc.run_automation(aid, "manual", {"title": "Bob"})
    assert res["status"] == "fired"
    assert seen == {"url": "https://hooks.test/x", "text": "Hi Bob"}


@pytest.mark.asyncio
async def test_slack_secret_encrypted_at_rest(client, monkeypatch):
    session, _ = _login(client)
    aid = _mkauto(client, session)
    seen = {}
    async def fake(url, text):
        seen["url"] = url
        return "ok"
    monkeypatch.setattr(auto_svc, "_post_slack", fake)
    _add_step(client, session, aid, "action",
              {"type": "slack", "webhook_url": "https://hooks.test/secret"})
    with get_conn() as conn:
        stored = conn.execute(
            "SELECT config_json FROM automation_steps WHERE automation_id=?", (aid,)).fetchone()[0]
    assert "hooks.test/secret" not in stored  # encrypted at rest
    res = await auto_svc.run_automation(aid, "manual", {})
    assert res["status"] == "fired"
    assert seen["url"] == "https://hooks.test/secret"  # decrypted on execute


@pytest.mark.asyncio
async def test_email_action_no_smtp_skips(client):
    session, uid = _login(client)
    aid = _mkauto(client, session)
    _add_step(client, session, aid, "action",
              {"type": "email", "to": f"user:{uid}", "subject": "S", "body": "B"})
    res = await auto_svc.run_automation(aid, "manual", {"created_by": uid})
    # user has email but SMTP unconfigured in tests → skipped, not error
    assert res["status"] == "fired"
    assert "email" in res["detail"]


@pytest.mark.asyncio
async def test_forge_issue_action(client, monkeypatch):
    session, _ = _login(client)
    aid = _mkauto(client, session)
    async def fake(provider, owner, repo, title, body, labels=None, user_id=None):
        return f"{provider} issue #7 in {owner}/{repo}"
    monkeypatch.setattr(auto_svc, "_create_forge_issue", fake)
    _add_step(client, session, aid, "action",
              {"type": "forge_issue", "provider": "gitea", "owner": "o", "repo": "r",
               "title": "Bug {{title}}"})
    res = await auto_svc.run_automation(aid, "manual", {"title": "X"})
    assert res["status"] == "fired"
    assert "gitea issue #7 in o/r" in res["detail"]


@pytest.mark.asyncio
async def test_forge_issue_needs_token(client):
    session, uid = _login(client)
    aid = _mkauto(client, session)
    _add_step(client, session, aid, "action",
              {"type": "forge_issue", "provider": "github", "owner": "o", "repo": "r",
               "title": "T"})
    res = await auto_svc.run_automation(aid, "manual", {"created_by": uid})
    assert res["status"] == "error"
    assert "token" in res["detail"].lower()


@pytest.mark.asyncio
async def test_agent_trigger_action(client, monkeypatch):
    session, _ = _login(client)
    aid = _mkauto(client, session)
    async def fake(agent_id, user_id, workspace_id, message, context):
        return f"agent {agent_id} ran: {message}"
    monkeypatch.setattr(auto_svc, "_run_linked_agent", fake)
    with get_conn() as conn:
        conn.execute("INSERT INTO agents (name, system_instructions) VALUES ('A', 'Do X')")
        agid = conn.execute("SELECT id FROM agents WHERE name='A'").fetchone()["id"]
        conn.commit()
    _add_step(client, session, aid, "action",
              {"type": "agent_trigger", "agent_id": agid, "message": "go"})
    res = await auto_svc.run_automation(aid, "manual", {})
    assert res["status"] == "fired"
    assert f"agent {agid} ran: go" in res["detail"]


@pytest.mark.asyncio
async def test_agent_trigger_missing_agent_errors(client):
    session, _ = _login(client)
    aid = _mkauto(client, session)
    _add_step(client, session, aid, "action", {"type": "agent_trigger", "agent_id": 999999})
    res = await auto_svc.run_automation(aid, "manual", {})
    assert res["status"] == "error"


# ── native button ──────────────────────────────────────────────────────────

def _make_button(client, cid, aid, name="Ship it"):
    r = client.post(f"/db/{cid}/properties/api",
                    json={"name": name, "prop_type": "button"})
    assert r.status_code == 200, r.text
    pid = r.json()["id"]
    with get_conn() as conn:
        conn.execute("UPDATE collection_properties SET button_automation_id=? WHERE id=?",
                     (aid, pid))
        conn.commit()
    return pid


def test_press_button_runs_automation(client):
    session, _ = _login(client)
    cid = _mkcollection(client)
    rid = _mkrow(client, cid)
    aid = _mkauto(client, session)
    _add_step(client, session, aid, "action", {"type": "notify", "message": "shipped"})
    prop_id = _make_button(client, cid, aid)
    r = client.post("/api/automations/press-button",
                    json={"collection_id": cid, "row_id": rid, "property_id": prop_id})
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "fired"
    assert len([x for x in _runs(aid) if x["status"] == "fired"]) == 1


def test_press_button_validation(client):
    session, _ = _login(client)
    cid = _mkcollection(client)
    rid = _mkrow(client, cid)
    r = client.post(f"/db/{cid}/properties/api",
                    json={"name": "Plain", "prop_type": "text"})
    text_pid = r.json()["id"]
    r = client.post("/api/automations/press-button",
                    json={"collection_id": cid, "row_id": rid, "property_id": text_pid})
    assert r.status_code == 400  # not a button
    r = client.post("/api/automations/press-button",
                    json={"collection_id": cid, "row_id": rid, "property_id": 999999})
    assert r.status_code == 400  # unknown
    aid = _mkauto(client, session)
    unlinked = _make_button(client, cid, aid, name="Unlinked")
    with get_conn() as conn:
        conn.execute("UPDATE collection_properties SET button_automation_id=NULL WHERE id=?",
                     (unlinked,))
        conn.commit()
    r = client.post("/api/automations/press-button",
                    json={"collection_id": cid, "row_id": rid, "property_id": unlinked})
    assert r.status_code == 400  # no linked automation


# ── legacy compat: no double run ───────────────────────────────────────────

@pytest.mark.asyncio
async def test_legacy_automation_single_run(client):
    session, _ = _login(client)
    cid = _mkcollection(client)
    r = client.post("/workspace/automations",
                    json={"name": "Legacy", "trigger_type": "event", "event": "page.created",
                          "collection_id": cid,
                          "actions": [{"type": "notify", "message": "legacy"}]},
                    cookies=_cookies(session))
    aid = r.json()["id"]
    await auto_svc.fire_event("page.created", {"collection_id": cid})
    assert len([x for x in _runs(aid) if x["status"] == "fired"]) == 1


# ── workers ────────────────────────────────────────────────────────────────

def _mkworker(client, session, **kw):
    body = {"name": "W", "code_py": "result['x'] = 1"}
    body.update(kw)
    r = client.post("/api/v2/workers", json=body, cookies=_cookies(session))
    assert r.status_code == 201, r.text
    return r.json()


def test_workers_crud_and_auth(client):
    anon(client)
    session, _ = _login(client)
    w = _mkworker(client, session, name="Hello")
    assert w["slug"].startswith("hello") or w["slug"]
    wid = w["id"]
    r = client.get(f"/api/v2/workers/{wid}", cookies=_cookies(session))
    assert r.status_code == 200
    assert r.json()["code_py"] == "result['x'] = 1"  # owner sees code
    r = client.patch(f"/api/v2/workers/{wid}", json={"shared": True},
                     cookies=_cookies(session))
    assert r.json()["shared"] == 1
    r = client.get("/api/v2/workers", cookies=_cookies(session))
    assert "X-Total-Count" in r.headers
    r = client.post("/api/v2/workers", json={"name": "X", "code_py": "x = 1"})
    assert r.status_code == 401
    r = client.delete(f"/api/v2/workers/{wid}", cookies=_cookies(session))
    assert r.status_code == 200


def test_worker_rejects_bad_code(client):
    session, _ = _login(client)
    r = client.post("/api/v2/workers",
                    json={"name": "Bad", "code_py": "import os\nresult['x']=1"},
                    cookies=_cookies(session))
    assert r.status_code == 400
    r = client.post("/api/v2/workers",
                    json={"name": "Bad2", "code_py": "open('/etc/passwd').read()"},
                    cookies=_cookies(session))
    assert r.status_code == 400
    r = client.post("/api/v2/workers",
                    json={"name": "Bad3", "code_py": "def broken(:\n  pass"},
                    cookies=_cookies(session))
    assert r.status_code == 400


def test_worker_run_ok(client):
    session, _ = _login(client)
    w = _mkworker(client, session, code_py="log('hello'); result['total'] = sum([1, 2, 3])")
    r = client.post(f"/api/v2/workers/{w['id']}/run", json={"ctx": {}},
                    cookies=_cookies(session))
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["status"] == "ok"
    assert body["result"] == {"total": 6}
    r = client.get(f"/api/v2/workers/{w['id']}/runs", cookies=_cookies(session))
    assert r.json()["runs"][0]["status"] == "ok"


def test_worker_run_error(client):
    session, _ = _login(client)
    w = _mkworker(client, session, code_py="1 / 0")
    r = client.post(f"/api/v2/workers/{w['id']}/run", json={}, cookies=_cookies(session))
    assert r.json()["status"] == "error"
    assert "ZeroDivision" in r.json()["error"]


def test_worker_run_timeout(client, monkeypatch):
    from app.services import workers as wsvc
    monkeypatch.setattr(wsvc, "RUN_TIMEOUT_S", 1)
    session, _ = _login(client)
    w = _mkworker(client, session, code_py="while True:\n  pass")
    r = client.post(f"/api/v2/workers/{w['id']}/run", json={}, cookies=_cookies(session))
    assert r.json()["status"] == "timeout"


def test_worker_budget_enforced(client):
    session, _ = _login(client)
    w = _mkworker(client, session, code_py="result['x'] = 1")
    with get_conn() as conn:
        conn.execute("UPDATE workers SET daily_budget_s=1 WHERE id=?", (w["id"],))
        conn.execute("INSERT INTO worker_runs (worker_id, status, duration_ms)"
                     " VALUES (?, 'ok', 60000)", (w["id"],))
        conn.commit()
    r = client.post(f"/api/v2/workers/{w['id']}/run", json={}, cookies=_cookies(session))
    assert r.status_code == 429


def test_worker_fork_and_privacy(client):
    s1, _ = _login(client)
    s2, _ = _login(client)
    w = _mkworker(client, s1, name="Private")
    r = client.post(f"/api/v2/workers/{w['id']}/fork", cookies=_cookies(s2))
    assert r.status_code == 403  # private
    client.patch(f"/api/v2/workers/{w['id']}", json={"shared": True},
                 cookies=_cookies(s1))
    r = client.post(f"/api/v2/workers/{w['id']}/fork", cookies=_cookies(s2))
    assert r.status_code == 201
    assert r.json()["from"] == w["id"]


def test_worker_private_run_forbidden(client):
    s1, _ = _login(client)
    s2, _ = _login(client)
    w = _mkworker(client, s1)
    r = client.post(f"/api/v2/workers/{w['id']}/run", json={}, cookies=_cookies(s2))
    assert r.status_code == 403


def test_workers_usage(client):
    session, _ = _login(client)
    w = _mkworker(client, session)
    client.post(f"/api/v2/workers/{w['id']}/run", json={}, cookies=_cookies(session))
    r = client.get("/api/v2/workers-usage", cookies=_cookies(session))
    assert r.json()["used_seconds_today"] >= 0


@pytest.mark.asyncio
async def test_run_due_workers(client):
    from app.services import workers as wsvc
    session, _ = _login(client)
    w = _mkworker(client, session, code_py="result['cron'] = True",
                  schedule_cron="@hourly")
    fired = await wsvc.run_due_workers()
    assert fired >= 1
    with get_conn() as conn:
        row = conn.execute("SELECT status FROM worker_runs WHERE worker_id=? ORDER BY id DESC",
                           (w["id"],)).fetchone()
    assert row and row["status"] == "ok"


def test_worker_code_hidden_from_strangers(client):
    s1, _ = _login(client)
    s2, _ = _login(client)
    w = _mkworker(client, s1)
    client.patch(f"/api/v2/workers/{w['id']}", json={"shared": True},
                 cookies=_cookies(s1))
    r = client.get(f"/api/v2/workers/{w['id']}", cookies=_cookies(s2))
    assert r.status_code == 200
    assert "code_py" not in r.json()  # no include_code for non-owner