using System.Text.Json;
using System.Text.Json.Nodes;
using Microsoft.EntityFrameworkCore;
using w4c_workflows.Data;
using w4c_workflows.Models;
using w4c_workflows.Models.Credentials;
using w4c_workflows.Models.Nodes;
using w4c_workflows.Services.Audit;
using w4c_workflows.Services.Credentials;
using w4c_workflows.Services.Execution;

namespace w4c_workflows.Services.Nodes;

/// <summary>Outcome of an in-process node-workflow run.</summary>
public sealed record NodeWorkflowRunOutcome(bool Succeeded, string? Error, string? Output);

/// <summary>
/// Runs a persisted node-mode workflow through <see cref="NodeGraphRunner"/>,
/// replacing the linear <c>NextId</c> subprocess chain for node definitions
/// (script workflows keep the legacy subprocess path in
/// <see cref="Runs.RunLifecycleEngine"/>). Reconstructs the executable graph from
/// the stored tasks + <see cref="WorkflowTaskEdge"/> rows, writes a
/// <see cref="TaskRun"/> row per executed node for history, and records the run's
/// terminal output/status.
///
/// Invoked by the worker (<see cref="Execution.GraphJobExecutor"/>) when it
/// claims a <c>graph.run</c> job, and recursively in-process by
/// <see cref="SubWorkflowInvoker"/> for <c>core.executeWorkflow</c> nodes.
/// </summary>
public sealed class NodeWorkflowRunner
{
    private static readonly JsonSerializerOptions Json = new(JsonSerializerDefaults.Web);

    private readonly NodeBlueprintCatalog _catalog;
    private readonly NodeGraphRunner _runner;
    private readonly CredentialVault _vault;
    private readonly ILogger<NodeWorkflowRunner>? _logger;
    private readonly int _maxSubWorkflowDepth;
    private readonly IActionAuditSink? _audit;

    public NodeWorkflowRunner(
        NodeBlueprintCatalog catalog,
        NodeGraphRunner runner,
        CredentialVault vault,
        ILogger<NodeWorkflowRunner>? logger = null,
        IActionAuditSink? audit = null)
        : this(catalog, runner, vault, logger, SubWorkflowInvoker.DefaultMaxDepth, audit)
    {
    }

    public NodeWorkflowRunner(
        NodeBlueprintCatalog catalog,
        NodeGraphRunner runner,
        CredentialVault vault,
        ILogger<NodeWorkflowRunner>? logger,
        int maxSubWorkflowDepth,
        IActionAuditSink? audit = null)
    {
        _catalog = catalog;
        _runner = runner;
        _vault = vault;
        _logger = logger;
        _maxSubWorkflowDepth = maxSubWorkflowDepth;
        _audit = audit;
    }

    public Task<NodeWorkflowRunOutcome> RunAsync(
        WorkflowsDbContext db,
        WorkflowRun run,
        Workflow workflow,
        string? workingDirectory,
        CancellationToken ct)
        // A top-level run's chain starts at its own workflow.
        => ExecuteAsync(db, run, workflow, workingDirectory, new[] { workflow.Id }, recordTaskRuns: true, ct);

    /// <summary>
    /// Runs a workflow's node graph and owns the run's terminal status. Child runs
    /// started by a <c>core.executeWorkflow</c> node call this with their own
    /// ancestry chain (for the recursion guard) and may opt out of per-node task
    /// history.
    /// </summary>
    internal async Task<NodeWorkflowRunOutcome> ExecuteAsync(
        WorkflowsDbContext db,
        WorkflowRun run,
        Workflow workflow,
        string? workingDirectory,
        IReadOnlyList<Guid> ancestry,
        bool recordTaskRuns,
        CancellationToken ct)
    {
        var build = BuildGraph(workflow);
        if (build.Graph == null)
            return await FailAsync(db, run, build.Error!, ct);

        // Resolve every credential the graph references once, so executors only
        // see decrypted data. Aliases are unique across a workflow (compile-time
        // rule), so a single alias → data map is unambiguous.
        IReadOnlyDictionary<string, CredentialData> credentials;
        try
        {
            credentials = await _vault.ResolveAsync(db, run.TenantId, CollectCredentialRefs(build.Graph), ct);
        }
        catch (CredentialResolutionException ex)
        {
            return await FailAsync(db, run, ex.Message, ct);
        }

        // The literal secret values resolved for this run. They are scrubbed from
        // every persisted/logged error so a token embedded in a URL path or query
        // (connector secrets are substituted into the URL) never reaches history.
        var secretValues = CollectSecretValues(credentials);

        var environment = new NodeRunEnvironment
        {
            TenantId = run.TenantId,
            RunId = run.Id.ToString(),
            WorkingDirectory = workingDirectory,
            Credentials = credentials,
            Depth = run.Depth,
            Ancestry = ancestry,
            // One invoker per run: it carries this run's chain so a nested
            // core.executeWorkflow can recurse with a correct depth/cycle view.
            SubWorkflows = new SubWorkflowInvoker(
                db, this, run, build.TaskIds, ancestry, _maxSubWorkflowDepth, _logger),
        };

        // At-least-once: number this graph execution and retire any in-flight
        // history left by a crashed/abandoned attempt (see BeginGraphAttemptAsync).
        var attempt = recordTaskRuns ? await BeginGraphAttemptAsync(db, run.Id, ct) : 1;

        var listener = BuildListener(db, run, build.TaskIds, recordTaskRuns, attempt, secretValues);

        NodeGraphRunResult result;
        try
        {
            result = await _runner.RunAsync(
                build.Graph, FlowItemJson.Parse(run.InputJson), environment, listener, ct);
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            _logger?.LogError("Node workflow run {RunId} threw: {Error}", run.Id, SecretRedactor.Redact(ex.Message, secretValues));
            return await FailAsync(db, run, ex.Message, ct, secretValues);
        }

        if (!result.Succeeded)
            return await FailAsync(db, run, result.Failure!.Message, ct, secretValues);

        var output = TerminalOutput(result);
        // Guarded write: if the control-plane timeout sweep already failed this
        // run while the worker was executing it, do not resurrect it as succeeded.
        if (!await RunTerminalWriter.TrySetTerminalAsync(
                db, run, RunStatus.Succeeded, error: null, output, ct))
        {
            _logger?.LogWarning(
                "Run {RunId} was terminalized before the graph finished; not overwriting", run.Id);
            return new NodeWorkflowRunOutcome(false, "the run was already terminal", null);
        }

        _logger?.LogDebug(
            "Node workflow run {RunId} succeeded: {Nodes} nodes executed", run.Id, result.ExecutionOrder.Count);

        return new NodeWorkflowRunOutcome(true, null, output);
    }

    /// <summary>
    /// At-least-once delivery: number this graph execution after any prior history
    /// for the run, and retire the previous attempt's in-flight <see cref="TaskRun"/>
    /// rows. A worker crash mid-graph leaves the job unacked; on redelivery the
    /// whole graph re-runs, so without this the run detail would accumulate
    /// indistinguishable duplicate rows and dangling <c>running</c> tasks.
    /// </summary>
    private static async Task<int> BeginGraphAttemptAsync(
        WorkflowsDbContext db, Guid runId, CancellationToken ct)
    {
        var previousMax = await db.TaskRuns
            .Where(t => t.RunId == runId)
            .Select(t => (int?)t.Attempt)
            .MaxAsync(ct) ?? 0;

        var attempt = previousMax + 1;

        await db.TaskRuns
            .Where(t => t.RunId == runId
                        && (t.Status == TaskRunStatus.Pending || t.Status == TaskRunStatus.Running))
            .ExecuteUpdateAsync(setters => setters
                .SetProperty(t => t.Status, TaskRunStatus.Dead)
                .SetProperty(t => t.Error, $"superseded by graph attempt {attempt}")
                .SetProperty(t => t.FinishedAt, DateTime.UtcNow), ct);

        return attempt;
    }

    // ------------------------------------------------------------------ graph build

    private sealed record GraphBuild(
        NodeGraph? Graph,
        IReadOnlyDictionary<string, Guid> TaskIds,
        string? Error);

    /// <summary>
    /// Reconstructs the executable graph from the compiled entities. The compile
    /// step already validated blueprint/port shape; here we only re-resolve the
    /// blueprints (the installed catalog may have changed since compile) and fail
    /// the run with a clear message when a type is missing.
    /// </summary>
    private GraphBuild BuildGraph(Workflow workflow)
    {
        var tasks = workflow.Tasks.ToList();
        var byId = tasks.ToDictionary(t => t.Id);
        var taskIds = tasks.ToDictionary(t => t.Key, t => t.Id, StringComparer.Ordinal);

        var nodes = new List<NodeGraphNode>(tasks.Count);
        foreach (var task in tasks)
        {
            if (string.IsNullOrWhiteSpace(task.NodeType))
                return new GraphBuild(null, taskIds, $"task '{task.Key}' is not a node step");

            var blueprint = _catalog.Resolve(task.NodeType!, task.NodeVersion);
            if (blueprint == null)
            {
                return new GraphBuild(null, taskIds,
                    $"node type '{task.NodeType}' is not installed at version {task.NodeVersion?.ToString() ?? "default"}");
            }

            nodes.Add(new NodeGraphNode
            {
                Id = task.Key,
                Blueprint = blueprint,
                Parameters = ParseObject(task.ParametersJson),
                CredentialRefs = ParseRefs(task.CredentialsJson),
                RunMode = task.RunMode,
                ContinueOnFail = task.ContinueOnFail,
                Retry = ParseRetry(task.RetryJson),
            });
        }

        var edges = new List<NodeGraphEdge>(workflow.TaskEdges.Count);
        foreach (var edge in workflow.TaskEdges)
        {
            if (!byId.TryGetValue(edge.FromTaskId, out var from) || !byId.TryGetValue(edge.ToTaskId, out var to))
                return new GraphBuild(null, taskIds, "a task edge references a task that is not in the workflow");

            edges.Add(new NodeGraphEdge
            {
                FromNodeId = from.Key,
                FromOutput = edge.FromOutput,
                ToNodeId = to.Key,
                ToInput = edge.ToInput,
                IsLoopBack = edge.IsLoopBack,
            });
        }

        var incoming = nodes.ToDictionary(n => n.Id, _ => 0, StringComparer.Ordinal);
        foreach (var edge in edges.Where(e => !e.IsLoopBack))
        {
            if (incoming.ContainsKey(edge.ToNodeId))
                incoming[edge.ToNodeId]++;
        }

        var roots = nodes.Where(n => incoming[n.Id] == 0).Select(n => n.Id).ToList();
        if (roots.Count == 0)
            return new GraphBuild(null, taskIds, "node graph has no entry node (it is a cycle)");
        if (roots.Count > 1)
            return new GraphBuild(null, taskIds,
                $"node graph has multiple entry nodes ({string.Join(", ", roots)})");

        return new GraphBuild(
            new NodeGraph { Nodes = nodes, Edges = edges, EntryNodeId = roots[0] },
            taskIds,
            null);
    }

    private static string? TerminalOutput(NodeGraphRunResult result)
    {
        var last = result.ExecutionOrder.LastOrDefault();
        return last == null ? null : FlowItemJson.Serialize(result.OutputOf(last));
    }

    /// <summary>Union of every alias → reference the graph's nodes declare.</summary>
    private static IReadOnlyDictionary<string, string> CollectCredentialRefs(NodeGraph graph)
    {
        var references = new Dictionary<string, string>(StringComparer.Ordinal);
        foreach (var node in graph.Nodes)
        {
            foreach (var (alias, reference) in node.CredentialRefs)
                references[alias] = reference;
        }
        return references;
    }

    private async Task<NodeWorkflowRunOutcome> FailAsync(
        WorkflowsDbContext db,
        WorkflowRun run,
        string error,
        CancellationToken ct,
        IReadOnlyCollection<string>? secretValues = null)
    {
        error = SecretRedactor.Redact(error, secretValues) ?? error;
        _logger?.LogWarning("Node workflow run {RunId} failed: {Error}", run.Id, error);

        // Guarded: a run the timeout sweep already failed keeps its original error.
        await RunTerminalWriter.TrySetTerminalAsync(
            db, run, RunStatus.Failed, error, run.OutputJson, ct);

        return new NodeWorkflowRunOutcome(false, error, null);
    }

    /// <summary>
    /// Flattens every string value in the resolved credentials. Used to scrub the
    /// actual secret literals from error messages, since the regex redactor cannot
    /// know an arbitrary token embedded in a URL path or query.
    /// </summary>
    private static List<string> CollectSecretValues(IReadOnlyDictionary<string, CredentialData> credentials)
    {
        var values = new List<string>();
        foreach (var credential in credentials.Values)
        {
            foreach (var (_, node) in credential.Data)
                CollectJsonStrings(node, values);
        }
        return values;
    }

    private static void CollectJsonStrings(JsonNode? node, List<string> values)
    {
        switch (node)
        {
            case JsonValue value when value.TryGetValue<string>(out var text)
                                      && !string.IsNullOrWhiteSpace(text)
                                      && text.Length >= 6:
                values.Add(text);
                break;
            case JsonObject obj:
                foreach (var (_, child) in obj)
                    CollectJsonStrings(child, values);
                break;
            case JsonArray array:
                foreach (var child in array)
                    CollectJsonStrings(child, values);
                break;
        }
    }

    // ------------------------------------------------------------------ history

    /// <summary>
    /// Composes the per-run listeners: task history when requested plus the
    /// audit listener when a sink is configured. Returns null when neither is
    /// needed so the kernel skips the callbacks entirely.
    /// </summary>
    private INodeRunListener? BuildListener(
        WorkflowsDbContext db,
        WorkflowRun run,
        IReadOnlyDictionary<string, Guid> taskIds,
        bool recordTaskRuns,
        int attempt,
        IReadOnlyCollection<string>? secretValues = null)
    {
        var listeners = new List<INodeRunListener>(2);
        if (recordTaskRuns)
            listeners.Add(new TaskRunHistory(db, run, taskIds, attempt, secretValues));
        if (_audit != null)
            listeners.Add(new AuditingNodeRunListener(_audit, run, secretValues));

        return listeners.Count switch
        {
            0 => null,
            1 => listeners[0],
            _ => new CompositeNodeRunListener(listeners),
        };
    }

    /// <summary>
    /// Writes one <see cref="TaskRun"/> per executed node: created <c>running</c>
    /// when the node starts, completed with its output and (if it errored) the
    /// failure when it finishes. Nodes skipped for lack of input leave no row.
    /// </summary>
    private sealed class TaskRunHistory : INodeRunListener
    {
        private readonly WorkflowsDbContext _db;
        private readonly WorkflowRun _run;
        private readonly IReadOnlyDictionary<string, Guid> _taskIds;
        private readonly int _attempt;
        private readonly IReadOnlyCollection<string>? _secretValues;
        private readonly Dictionary<string, TaskRun> _rows = new(StringComparer.Ordinal);

        public TaskRunHistory(
            WorkflowsDbContext db,
            WorkflowRun run,
            IReadOnlyDictionary<string, Guid> taskIds,
            int attempt,
            IReadOnlyCollection<string>? secretValues = null)
        {
            _db = db;
            _run = run;
            _taskIds = taskIds;
            _attempt = attempt;
            _secretValues = secretValues;
        }

        public async Task NodeStartedAsync(
            NodeGraphNode node, IReadOnlyList<IReadOnlyList<FlowItem>> inputs, CancellationToken ct)
        {
            if (!_taskIds.TryGetValue(node.Id, out var taskId))
                return;

            var row = new TaskRun
            {
                Id = Guid.NewGuid(),
                RunId = _run.Id,
                TaskId = taskId,
                Attempt = _attempt,
                Status = TaskRunStatus.Running,
                InputJson = MainOutput(inputs),
                StartedAt = DateTime.UtcNow,
            };

            _db.TaskRuns.Add(row);
            _rows[node.Id] = row;
            await _db.SaveChangesAsync(ct);
        }

        public async Task NodeFinishedAsync(
            NodeGraphNode node,
            IReadOnlyList<IReadOnlyList<FlowItem>> outputs,
            NodeFailure? failure,
            CancellationToken ct)
        {
            if (!_rows.TryGetValue(node.Id, out var row))
                return;

            row.Status = failure == null ? TaskRunStatus.Succeeded : TaskRunStatus.Failed;
            row.OutputJson = MainOutput(outputs);
            row.Error = SecretRedactor.Redact(failure?.Message, _secretValues);
            row.FinishedAt = DateTime.UtcNow;

            await _db.SaveChangesAsync(ct);
        }

        private static string? MainOutput(IReadOnlyList<IReadOnlyList<FlowItem>> ports)
            => ports.Count > 0 ? FlowItemJson.Serialize(ports[0]) : null;
    }

    // ------------------------------------------------------------------ parsing

    private static JsonObject ParseObject(string? json)
    {
        if (string.IsNullOrWhiteSpace(json))
            return new JsonObject();
        try
        {
            return JsonNode.Parse(json) as JsonObject ?? new JsonObject();
        }
        catch (JsonException)
        {
            return new JsonObject();
        }
    }

    private static IReadOnlyDictionary<string, string> ParseRefs(string? json)
    {
        if (string.IsNullOrWhiteSpace(json))
            return new Dictionary<string, string>();
        try
        {
            return JsonSerializer.Deserialize<Dictionary<string, string>>(json, Json)
                ?? new Dictionary<string, string>();
        }
        catch (JsonException)
        {
            return new Dictionary<string, string>();
        }
    }

    private static TaskRetryDefinition? ParseRetry(string? json)
    {
        if (string.IsNullOrWhiteSpace(json))
            return null;
        try
        {
            return JsonSerializer.Deserialize<TaskRetryDefinition>(json, Json);
        }
        catch (JsonException)
        {
            return null;
        }
    }
}