"""Workflow engine: nodes, branching, triggers, approvals, scheduling, events."""

from __future__ import annotations

import json
from datetime import UTC, datetime

import httpx
from legate.policy.roles import Role
from legate.workflow.service import emit_event, is_cron_due, run_due_workflows
from legate.workflow.webhooks import sign


def _nodes_edges(nodes, edges):
    return {"nodes": nodes, "edges": edges}


def _steps(client, helpers, token, run_id):
    return client.get(f"/runs/{run_id}/steps", headers=helpers.headers(token)).json()


def test_manual_run_executes_graph_and_records_steps(client, helpers):
    owner = helpers.bootstrap_owner(client)
    token = owner["access_token"]
    definition = _nodes_edges(
        [
            {"id": "t1", "type": "trigger", "config": {}},
            {"id": "x1", "type": "transform", "config": {"template": {"q": "{{ trigger.query }}"}}},
            {
                "id": "c1",
                "type": "condition",
                "config": {"expr": "{{ trigger.priority }} == 'high'"},
            },
            {
                "id": "n1",
                "type": "tool",
                "config": {"tool": "kb.search", "args": {"query": "{{ trigger.query }}"}},
            },
        ],
        [
            {"from": "t1", "to": "x1"},
            {"from": "x1", "to": "c1"},
            {"from": "c1", "to": "n1", "when": "true"},
        ],
    )
    wf = helpers.create_workflow(client, token, definition=definition)["workflow"]

    created = helpers.run_workflow(
        client, token, wf["id"], input={"query": "refunds", "priority": "high"}
    )
    run_id = created["run"]["id"]

    run = client.get(f"/runs/{run_id}", headers=helpers.headers(token)).json()
    assert run["status"] == "succeeded"
    types = [s["type"] for s in _steps(client, helpers, token, run_id)]
    assert types == ["trigger", "transform", "condition", "tool_call"]


def test_condition_false_skips_branch(client, helpers):
    owner = helpers.bootstrap_owner(client)
    token = owner["access_token"]
    definition = _nodes_edges(
        [
            {"id": "t1", "type": "trigger", "config": {}},
            {
                "id": "c1",
                "type": "condition",
                "config": {"expr": "{{ trigger.priority }} == 'high'"},
            },
            {"id": "n1", "type": "tool", "config": {"tool": "kb.search", "args": {"query": "x"}}},
        ],
        [{"from": "t1", "to": "c1"}, {"from": "c1", "to": "n1", "when": "true"}],
    )
    wf = helpers.create_workflow(client, token, definition=definition)["workflow"]
    created = helpers.run_workflow(client, token, wf["id"], input={"priority": "low"})
    run_id = created["run"]["id"]

    assert (
        client.get(f"/runs/{run_id}", headers=helpers.headers(token)).json()["status"]
        == "succeeded"
    )
    types = [s["type"] for s in _steps(client, helpers, token, run_id)]
    assert types == ["trigger", "condition"]  # tool branch not taken


def test_webhook_trigger_verifies_signature(client, helpers):
    owner = helpers.bootstrap_owner(client)
    token = owner["access_token"]
    definition = _nodes_edges(
        [
            {"id": "t1", "type": "trigger", "config": {}},
            {
                "id": "n1",
                "type": "tool",
                "config": {"tool": "kb.search", "args": {"query": "{{ trigger.body.q }}"}},
            },
        ],
        [{"from": "t1", "to": "n1"}],
    )
    created = helpers.create_workflow(client, token, definition=definition, trigger_type="webhook")
    wf_token = created["workflow"]["webhook_token"]
    secret = created["webhook_secret"]
    assert wf_token and secret

    body = json.dumps({"q": "refunds"}).encode()
    good = client.post(
        f"/hooks/{wf_token}", content=body, headers={"X-Legate-Signature": sign(secret, body)}
    )
    assert good.status_code == 202, good.text
    run_id = good.json()["run_id"]
    assert (
        client.get(f"/runs/{run_id}", headers=helpers.headers(token)).json()["status"]
        == "succeeded"
    )

    bad = client.post(
        f"/hooks/{wf_token}", content=body, headers={"X-Legate-Signature": "deadbeef"}
    )
    assert bad.status_code == 401
    missing = client.post(f"/hooks/{wf_token}", content=body)
    assert missing.status_code == 401


def test_agent_node_runs_agent_and_feeds_condition(client, helpers, set_agent_responses):
    owner = helpers.bootstrap_owner(client)
    token = owner["access_token"]
    agent = helpers.create_agent(client, token, name="Classifier")
    set_agent_responses(
        [
            json.dumps(
                {"thought": "classify", "action": "final", "args": {"output": {"priority": "high"}}}
            )
        ]
    )
    definition = _nodes_edges(
        [
            {"id": "t1", "type": "trigger", "config": {}},
            {
                "id": "a1",
                "type": "agent",
                "config": {"agent_id": agent["id"], "input": "{{ trigger.text }}"},
            },
            {
                "id": "c1",
                "type": "condition",
                "config": {"expr": "{{ a1.output.priority }} == 'high'"},
            },
            {"id": "d1", "type": "transform", "config": {"template": {"handled": True}}},
        ],
        [
            {"from": "t1", "to": "a1"},
            {"from": "a1", "to": "c1"},
            {"from": "c1", "to": "d1", "when": "true"},
        ],
    )
    wf = helpers.create_workflow(client, token, definition=definition)["workflow"]
    created = helpers.run_workflow(client, token, wf["id"], input={"text": "urgent billing issue"})
    run_id = created["run"]["id"]

    run = client.get(f"/runs/{run_id}", headers=helpers.headers(token)).json()
    assert run["status"] == "succeeded"
    types = [s["type"] for s in _steps(client, helpers, token, run_id)]
    assert types == ["trigger", "agent_reason", "condition", "transform"]


def test_approval_node_pauses_then_resumes(client, helpers):
    owner = helpers.bootstrap_owner(client)
    token = owner["access_token"]
    definition = _nodes_edges(
        [
            {"id": "t1", "type": "trigger", "config": {}},
            {"id": "ap1", "type": "approval", "config": {"summary": "please review"}},
            {"id": "n1", "type": "tool", "config": {"tool": "kb.search", "args": {"query": "x"}}},
        ],
        [{"from": "t1", "to": "ap1"}, {"from": "ap1", "to": "n1", "when": "true"}],
    )
    wf = helpers.create_workflow(client, token, definition=definition)["workflow"]
    created = helpers.run_workflow(client, token, wf["id"], input={})
    run_id = created["run"]["id"]
    assert created["run"]["status"] == "waiting_approval"

    approval_id = client.get("/approvals", headers=helpers.headers(token)).json()[0]["id"]
    res = client.post(f"/approvals/{approval_id}/approve", headers=helpers.headers(token))
    assert res.status_code == 200
    assert res.json()["run_status"] == "succeeded"

    types = [s["type"] for s in _steps(client, helpers, token, run_id)]
    assert "approval" in types and "tool_call" in types


def test_approval_node_reject_skips_downstream(client, helpers):
    owner = helpers.bootstrap_owner(client)
    token = owner["access_token"]
    definition = _nodes_edges(
        [
            {"id": "t1", "type": "trigger", "config": {}},
            {"id": "ap1", "type": "approval", "config": {}},
            {"id": "n1", "type": "tool", "config": {"tool": "kb.search", "args": {"query": "x"}}},
        ],
        [{"from": "t1", "to": "ap1"}, {"from": "ap1", "to": "n1", "when": "true"}],
    )
    wf = helpers.create_workflow(client, token, definition=definition)["workflow"]
    run_id = helpers.run_workflow(client, token, wf["id"])["run"]["id"]
    approval_id = client.get("/approvals", headers=helpers.headers(token)).json()[0]["id"]

    client.post(f"/approvals/{approval_id}/reject", headers=helpers.headers(token))
    run = client.get(f"/runs/{run_id}", headers=helpers.headers(token)).json()
    assert run["status"] == "succeeded"
    assert not any(s["name"] == "n1" for s in _steps(client, helpers, token, run_id))


def test_slack_tool_node_posts_via_connector(client, helpers, set_connector_http):
    owner = helpers.bootstrap_owner(client)
    token = owner["access_token"]
    helpers.install_connector(client, token, type="slack", secrets={"bot_token": "xoxb-x"})

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"ok": True, "ts": "1.1", "channel": "C1"})

    set_connector_http(handler)
    definition = _nodes_edges(
        [
            {"id": "t1", "type": "trigger", "config": {}},
            {
                "id": "n1",
                "type": "tool",
                "config": {
                    "tool": "slack.post_message",
                    "args": {"channel": "C1", "text": "hi {{ trigger.name }}"},
                },
            },
        ],
        [{"from": "t1", "to": "n1"}],
    )
    wf = helpers.create_workflow(client, token, definition=definition)["workflow"]
    run_id = helpers.run_workflow(client, token, wf["id"], input={"name": "team"})["run"]["id"]

    assert (
        client.get(f"/runs/{run_id}", headers=helpers.headers(token)).json()["status"]
        == "succeeded"
    )
    tool_steps = [s for s in _steps(client, helpers, token, run_id) if s["type"] == "tool_call"]
    assert tool_steps and tool_steps[0]["output"]["data"]["ts"] == "1.1"


def test_failing_node_retries_then_fails(client, helpers):
    owner = helpers.bootstrap_owner(client)
    token = owner["access_token"]
    definition = _nodes_edges(
        [
            {"id": "t1", "type": "trigger", "config": {}},
            {
                "id": "bad",
                "type": "tool",
                "config": {"tool": "does.not.exist", "args": {}, "retry": {"max_attempts": 2}},
            },
        ],
        [{"from": "t1", "to": "bad"}],
    )
    wf = helpers.create_workflow(client, token, definition=definition)["workflow"]
    run_id = helpers.run_workflow(client, token, wf["id"])["run"]["id"]

    run = client.get(f"/runs/{run_id}", headers=helpers.headers(token)).json()
    assert run["status"] == "failed"
    error_steps = [s for s in _steps(client, helpers, token, run_id) if s["status"] == "error"]
    assert error_steps and error_steps[0]["input"]["attempts"] == 2


def test_is_cron_due():
    now = datetime(2026, 7, 1, 12, 0, 30, tzinfo=UTC)
    assert is_cron_due("* * * * *", None, now) is True
    assert is_cron_due("* * * * *", datetime(2026, 7, 1, 12, 0, 0, tzinfo=UTC), now) is False
    assert is_cron_due("* * * * *", datetime(2026, 7, 1, 11, 59, 0, tzinfo=UTC), now) is True


async def test_scheduled_workflow_runs_when_due(client, helpers, session_factory):
    owner = helpers.bootstrap_owner(client)
    token = owner["access_token"]
    definition = _nodes_edges(
        [
            {"id": "t1", "type": "trigger", "config": {}},
            {"id": "n1", "type": "tool", "config": {"tool": "kb.search", "args": {"query": "x"}}},
        ],
        [{"from": "t1", "to": "n1"}],
    )
    helpers.create_workflow(
        client,
        token,
        definition=definition,
        trigger_type="schedule",
        trigger_config={"cron": "* * * * *"},
    )

    started = await run_due_workflows(session_factory, datetime.now(UTC))
    assert len(started) == 1
    # Running again immediately does not re-fire (last_triggered_at just set).
    assert await run_due_workflows(session_factory, datetime.now(UTC)) == []


async def test_event_trigger_starts_subscribed_workflow(client, helpers, session_factory):
    owner = helpers.bootstrap_owner(client)
    token = owner["access_token"]
    import uuid

    workspace_id = uuid.UUID(owner["workspace"]["id"])
    definition = _nodes_edges(
        [
            {"id": "t1", "type": "trigger", "config": {}},
            {"id": "n1", "type": "tool", "config": {"tool": "kb.search", "args": {"query": "x"}}},
        ],
        [{"from": "t1", "to": "n1"}],
    )
    helpers.create_workflow(
        client,
        token,
        definition=definition,
        trigger_type="event",
        trigger_config={"event": "doc.ingested"},
    )

    started = await emit_event(
        session_factory, workspace_id=workspace_id, event="doc.ingested", payload={"id": "1"}
    )
    assert len(started) == 1
    none = await emit_event(
        session_factory, workspace_id=workspace_id, event="other.event", payload={}
    )
    assert none == []


def test_workflow_rbac_and_tenant_isolation(client, helpers):
    owner = helpers.bootstrap_owner(client)
    builder = helpers.add_member(client, owner["access_token"], email="b@x.com", role=Role.BUILDER)
    viewer = helpers.add_member(client, owner["access_token"], email="v@x.com", role=Role.VIEWER)

    wf = helpers.create_workflow(client, builder["access_token"], name="Owned")["workflow"]
    forbidden = client.post(
        "/workflows", json={"name": "No"}, headers=helpers.headers(viewer["access_token"])
    )
    assert forbidden.status_code == 403

    other = helpers.bootstrap_owner(client, email="z@x.com", name="Z")
    assert (
        client.get(
            f"/workflows/{wf['id']}", headers=helpers.headers(other["access_token"])
        ).status_code
        == 404
    )
