from datetime import UTC, datetime
from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest

from core.app.apps.common import workflow_response_converter
from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
from core.app.apps.workflow.app_runner import WorkflowAppRunner
from core.app.entities.app_invoke_entities import InvokeFrom
from core.app.entities.queue_entities import QueueWorkflowPausedEvent
from core.app.entities.task_entities import HumanInputRequiredResponse, WorkflowPauseStreamResponse
from core.workflow.nodes.human_input.entities import (
    ParagraphInputConfig,
    SelectInputConfig,
    StringListSource,
    UserActionConfig,
)
from core.workflow.nodes.human_input.enums import ValueSourceType
from core.workflow.nodes.human_input.pause_reason import HumanInputRequired
from core.workflow.system_variables import build_system_variables
from graphon.entities import WorkflowStartReason
from graphon.entities.pause_reason import HitlRequired
from graphon.graph_events import GraphRunPausedEvent
from graphon.runtime import GraphRuntimeState, VariablePool
from models.account import Account
from models.human_input import RecipientType


class _FakeSession:
    """Stub session: `execute` feeds the form-expiration query, `scalars` the recipients."""

    def __init__(self, *, execute_rows=(), scalars_rows=()):
        self._execute_rows = execute_rows
        self._scalars_rows = scalars_rows

    def execute(self, _stmt):
        return list(self._execute_rows)

    def scalars(self, _stmt):
        return list(self._scalars_rows)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, tb):
        return False


class _RecordingWorkflowAppRunner(WorkflowAppRunner):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.published_events = []

    def _publish_event(self, event):
        self.published_events.append(event)


class _FakeRuntimeState:
    variable_pool = object()

    def get_paused_nodes(self):
        return ["node-pause-1"]


def _build_runner():
    app_entity = SimpleNamespace(
        app_config=SimpleNamespace(app_id="app-id"),
        inputs={},
        files=[],
        invoke_from=InvokeFrom.SERVICE_API,
        single_iteration_run=None,
        single_loop_run=None,
        workflow_execution_id="run-id",
        user_id="user-id",
    )
    workflow = SimpleNamespace(
        graph_dict={},
        tenant_id="tenant-id",
        environment_variables={},
        id="workflow-id",
    )
    queue_manager = SimpleNamespace(publish=lambda event, pub_from: None)
    return _RecordingWorkflowAppRunner(
        application_generate_entity=app_entity,
        queue_manager=queue_manager,
        variable_loader=MagicMock(),
        workflow=workflow,
        system_user_id="sys-user",
        root_node_id=None,
        workflow_execution_repository=MagicMock(),
        workflow_node_execution_repository=MagicMock(),
        graph_engine_layers=(),
        graph_runtime_state=None,
    )


def test_graph_run_paused_event_emits_queue_pause_event(monkeypatch: pytest.MonkeyPatch):
    runner = _build_runner()
    graph_reason = HitlRequired(
        session_id="form-1",
        node_id="node-human",
        node_title="Human Step",
    )
    event = GraphRunPausedEvent(reasons=[graph_reason], outputs={"foo": "bar"})
    workflow_entry = SimpleNamespace(
        graph_engine=SimpleNamespace(graph_runtime_state=_FakeRuntimeState()),
    )

    enriched_reason = HumanInputRequired(
        form_id="form-1",
        form_content="content",
        inputs=[],
        actions=[],
        node_id="node-human",
        node_title="Human Step",
    )
    monkeypatch.setattr(
        "core.app.apps.workflow_app_runner.enrich_graph_pause_reasons",
        lambda **_: [enriched_reason],
    )

    runner._handle_event(workflow_entry, event)

    assert len(runner.published_events) == 1
    queue_event = runner.published_events[0]
    assert isinstance(queue_event, QueueWorkflowPausedEvent)
    assert queue_event.reasons == [enriched_reason]
    assert queue_event.outputs == {"foo": "bar"}
    assert queue_event.paused_nodes == ["node-pause-1"]


def _build_converter(*, invoke_from: InvokeFrom = InvokeFrom.SERVICE_API):
    application_generate_entity = SimpleNamespace(
        inputs={},
        files=[],
        invoke_from=invoke_from,
        app_config=SimpleNamespace(app_id="app-id", tenant_id="tenant-id"),
    )
    system_variables = build_system_variables(
        user_id="user",
        app_id="app-id",
        workflow_id="workflow-id",
        workflow_execution_id="run-id",
    )
    user = MagicMock(spec=Account)
    user.id = "account-id"
    user.name = "Tester"
    user.email = "tester@example.com"
    return WorkflowResponseConverter(
        application_generate_entity=application_generate_entity,
        user=user,
        system_variables=system_variables,
    )


def test_queue_workflow_paused_event_to_stream_responses(monkeypatch: pytest.MonkeyPatch):
    converter = _build_converter()
    converter.workflow_start_to_stream_response(
        task_id="task",
        workflow_run_id="run-id",
        workflow_id="workflow-id",
        reason=WorkflowStartReason.INITIAL,
    )

    expiration_time = datetime(2024, 1, 1, tzinfo=UTC)
    session = _FakeSession(
        execute_rows=[("form-1", expiration_time, '{"display_in_ui": true}')],
        scalars_rows=[
            SimpleNamespace(form_id="form-1", recipient_type=RecipientType.CONSOLE, access_token="console-token"),
            SimpleNamespace(form_id="form-1", recipient_type=RecipientType.BACKSTAGE, access_token="backstage-token"),
        ],
    )

    monkeypatch.setattr(workflow_response_converter, "Session", lambda **_: session)
    monkeypatch.setattr(workflow_response_converter, "db", SimpleNamespace(engine=object()))

    reason = HumanInputRequired(
        form_id="form-1",
        form_content="Rendered",
        inputs=[ParagraphInputConfig(output_variable_name="field")],
        actions=[UserActionConfig(id="approve", title="Approve")],
        node_id="node-id",
        node_title="Human Step",
    )
    queue_event = QueueWorkflowPausedEvent(
        reasons=[reason],
        outputs={"answer": "value"},
        paused_nodes=["node-id"],
    )

    runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=0.0)
    responses = converter.workflow_pause_to_stream_response(
        event=queue_event,
        task_id="task",
        graph_runtime_state=runtime_state,
    )

    assert isinstance(responses[-1], WorkflowPauseStreamResponse)
    pause_resp = responses[-1]
    assert pause_resp.workflow_run_id == "run-id"
    assert pause_resp.data.paused_nodes == ["node-id"]
    assert pause_resp.data.outputs == {}
    assert pause_resp.data.reasons[0]["form_id"] == "form-1"

    assert isinstance(responses[0], HumanInputRequiredResponse)
    hi_resp = responses[0]
    assert hi_resp.data.form_id == "form-1"
    assert hi_resp.data.node_id == "node-id"
    assert hi_resp.data.node_title == "Human Step"
    assert hi_resp.data.inputs[0].output_variable_name == "field"
    assert hi_resp.data.actions[0].id == "approve"
    assert hi_resp.data.display_in_ui is True
    assert hi_resp.data.form_token is None
    assert hi_resp.data.approval_channels == ["console"]
    assert hi_resp.data.expiration_time == int(expiration_time.timestamp())


def _build_paused_human_input_response(monkeypatch, recipients):
    """Drive the live OPENAPI pause path with the given recipients via a fake session."""
    converter = _build_converter(invoke_from=InvokeFrom.OPENAPI)
    converter.workflow_start_to_stream_response(
        task_id="task",
        workflow_run_id="run-id",
        workflow_id="workflow-id",
        reason=WorkflowStartReason.INITIAL,
    )

    expiration_time = datetime(2024, 1, 1, tzinfo=UTC)
    session = _FakeSession(
        execute_rows=[("form-1", expiration_time, '{"display_in_ui": true}')],
        scalars_rows=list(recipients),
    )

    monkeypatch.setattr(workflow_response_converter, "Session", lambda **_: session)
    monkeypatch.setattr(workflow_response_converter, "db", SimpleNamespace(engine=object()))

    reason = HumanInputRequired(
        form_id="form-1",
        form_content="Rendered",
        inputs=[ParagraphInputConfig(output_variable_name="field")],
        actions=[UserActionConfig(id="approve", title="Approve")],
        node_id="node-id",
        node_title="Human Step",
    )
    queue_event = QueueWorkflowPausedEvent(
        reasons=[reason],
        outputs={},
        paused_nodes=["node-id"],
    )

    runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=0.0)
    responses = converter.workflow_pause_to_stream_response(
        event=queue_event,
        task_id="task",
        graph_runtime_state=runtime_state,
    )
    assert isinstance(responses[0], HumanInputRequiredResponse)
    return responses


def test_openapi_pause_without_web_app_recipient_emits_approval_channels(monkeypatch: pytest.MonkeyPatch):
    responses = _build_paused_human_input_response(
        monkeypatch,
        recipients=[
            SimpleNamespace(form_id="form-1", recipient_type=RecipientType.EMAIL_MEMBER, access_token="email-token"),
            SimpleNamespace(form_id="form-1", recipient_type=RecipientType.BACKSTAGE, access_token="backstage-token"),
        ],
    )

    hi_resp = responses[0]
    assert hi_resp.data.form_token is None
    assert hi_resp.data.approval_channels == ["console", "email"]

    pause_resp = responses[-1]
    assert pause_resp.data.reasons[0]["approval_channels"] == ["console", "email"]


def test_openapi_pause_with_web_app_recipient_sets_token_and_channels(monkeypatch: pytest.MonkeyPatch):
    responses = _build_paused_human_input_response(
        monkeypatch,
        recipients=[
            SimpleNamespace(
                form_id="form-1",
                recipient_type=RecipientType.STANDALONE_WEB_APP,
                access_token="web-app-token",
            ),
            SimpleNamespace(form_id="form-1", recipient_type=RecipientType.BACKSTAGE, access_token="backstage-token"),
        ],
    )

    hi_resp = responses[0]
    assert hi_resp.data.form_token == "web-app-token"
    assert hi_resp.data.approval_channels == ["console"]

    pause_resp = responses[-1]
    assert pause_resp.data.reasons[0]["approval_channels"] == ["console"]


def test_queue_workflow_paused_event_resolves_variable_select_options(monkeypatch: pytest.MonkeyPatch):
    converter = _build_converter()
    converter.workflow_start_to_stream_response(
        task_id="task",
        workflow_run_id="run-id",
        workflow_id="workflow-id",
        reason=WorkflowStartReason.INITIAL,
    )

    expiration_time = datetime(2024, 1, 1, tzinfo=UTC)
    session = _FakeSession(execute_rows=[("form-1", expiration_time, '{"display_in_ui": true}')])

    monkeypatch.setattr(workflow_response_converter, "Session", lambda **_: session)
    monkeypatch.setattr(workflow_response_converter, "db", SimpleNamespace(engine=object()))

    reason = HumanInputRequired(
        form_id="form-1",
        form_content="Rendered",
        inputs=[
            SelectInputConfig(
                output_variable_name="decision",
                option_source=StringListSource(
                    type=ValueSourceType.VARIABLE,
                    selector=["start", "options"],
                    value=[],
                ),
            )
        ],
        actions=[UserActionConfig(id="approve", title="Approve")],
        node_id="node-id",
        node_title="Human Step",
    )
    queue_event = QueueWorkflowPausedEvent(
        reasons=[reason],
        outputs={},
        paused_nodes=["node-id"],
    )

    runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=0.0)
    runtime_state.variable_pool.add(("start", "options"), ["approve", "reject"])
    responses = converter.workflow_pause_to_stream_response(
        event=queue_event,
        task_id="task",
        graph_runtime_state=runtime_state,
    )

    assert isinstance(responses[0], HumanInputRequiredResponse)
    hi_resp = responses[0]
    assert hi_resp.data.inputs[0].option_source.value == ["approve", "reject"]

    assert isinstance(responses[-1], WorkflowPauseStreamResponse)
    pause_resp = responses[-1]
    assert pause_resp.data.reasons[0]["inputs"][0]["option_source"]["value"] == ["approve", "reject"]
