XFEWorkflow

using System.Text.Json;
using System.Text.Json.Serialization;
using XFEExtension.NetCore.XFEConsole;
using XFEWorkflow.Abstractions;
using XFEWorkflow.Core.Runtime;
using XFEWorkflow.Infrastructure.Execution;

namespace XFEWorkflow.Infrastructure.Storage;

public sealed class WorkflowLogEntry : XFELogEntry
{
    public string Id { get; set; } = Guid.NewGuid().ToString("N");
    public string Category { get; set; } = "应用";
    public string WorkflowId { get; set; } = "";
    public string WorkflowName { get; set; } = "";
    public string RunId { get; set; } = "";
    public string? NodeId { get; set; }
    public string NodeName { get; set; } = "";
    public string Trigger { get; set; } = "";
    public WorkflowDiagnostic? Diagnostic { get; set; }
    [JsonIgnore] public string TimeText => Time.ToString("HH:mm:ss.fff");
    [JsonIgnore] public string LevelLabel => Level.ToString().ToUpperInvariant();
    [JsonIgnore] public string SourceLabel => WorkflowName.Length == 0 ? Category : WorkflowName + (NodeName.Length == 0 ? "" : " / " + NodeName);
    [JsonIgnore] public string SearchText => $"{LogText} {SourceLabel} {RunId} {NodeId} {Trigger}";
}

/// <summary>Runner events retain their original run record; diagnostics use a separate journal.
/// Both are XFE log entries and share the same filtering, rendering and export pipeline.</summary>
public sealed class WorkflowLogService(WorkspacePaths paths, RunCoordinator runs, SecretRedactor redactor)
{
    private readonly object gate = new();
    private readonly Dictionary<string, (int Count, bool Finished)> observed = [];
    private readonly Dictionary<string, List<WorkflowLogEntry>> live = [];
    private readonly Dictionary<string, string> diagnosticVersions = [];
    private readonly SemaphoreSlim journalGate = new(1);
    public event Action? Changed;

    public static IReadOnlyList<WorkflowLogEntry> RunEntries(RunRecord run, int maxEntries = int.MaxValue)
    {
        ArgumentOutOfRangeException.ThrowIfNegative(maxEntries);
        if (maxEntries == 0) return [];
        var terminal = TerminalEntry(run);
        (RunEvent Event, int Index)[] events;
        lock (run.Events) events = run.Events.Select((e, i) => (Event: e, Index: i)).Where(x => IsLog(x.Event))
            .TakeLast(maxEntries - (terminal is null ? 0 : 1)).ToArray();
        var result = events.Select(x => Entry(run, x.Event.Message, x.Event.Timestamp.LocalDateTime,
            x.Event.Kind == RunEventKind.NodeFailed ? LogLevel.Error : LogLevel.Info, x.Event.NodeId, run.Id + "/event/" + x.Index)).ToList();
        if (terminal is not null) result.Add(terminal);
        return result;
    }
    private static bool IsLog(RunEvent ev) => ev.Kind is RunEventKind.Log or RunEventKind.NodeFailed;
    private static WorkflowLogEntry? TerminalEntry(RunRecord run)
    {
        if (run.FinishedAt is not { } finished || run.Status is not (RunStatus.Failed or RunStatus.Cancelled or RunStatus.Interrupted or RunStatus.Skipped)) return null;
        lock (run.Events) if (run.Status == RunStatus.Failed && run.Events.Any(e => e.Kind == RunEventKind.NodeFailed && e.NodeId == run.ErrorNodeId)) return null;
        return Entry(run, run.Error ?? "运行状态：" + run.Status, finished.LocalDateTime,
            run.Status == RunStatus.Failed ? LogLevel.Error : LogLevel.Warning, run.ErrorNodeId, run.Id + "/finished");
    }
    private static WorkflowLogEntry Entry(RunRecord run, string message, DateTime time, LogLevel level, string? node, string id) => new()
    {
        Id = id, Category = "工作流运行", Time = time, Level = level, LogText = message, WorkflowId = run.WorkflowId,
        WorkflowName = run.WorkflowName, RunId = run.Id, NodeId = node, NodeName = node is null ? "" : run.NodeNames.GetValueOrDefault(node, node), Trigger = run.Trigger
    };
    public void Observe(RunRecord run)
    {
        lock (gate)
        {
            var previous = observed.GetValueOrDefault(run.Id);
            int eventCount; RunEvent[] events;
            lock (run.Events) { eventCount = run.Events.Count; events = run.Events.Skip(previous.Count).ToArray(); }
            if (!live.TryGetValue(run.Id, out var entries)) live[run.Id] = entries = [];
            foreach (var (ev, index) in events.Select((ev, index) => (ev, index)))
            {
                if (!IsLog(ev)) continue;
                var entry = Entry(run, ev.Message, ev.Timestamp.LocalDateTime, ev.Kind == RunEventKind.NodeFailed ? LogLevel.Error : LogLevel.Info, ev.NodeId, run.Id + "/event/" + (previous.Count + index));
                entries.Add(entry); Publish(entry);
            }
            if (run.FinishedAt is not null && !previous.Finished)
            {
                if (TerminalEntry(run) is { } terminal) Publish(terminal);
                // The coordinator persists the finished record before this notification.
                live.Remove(run.Id);
            }
            observed[run.Id] = (eventCount, run.FinishedAt is not null);
            if (observed.Count > 2048) foreach (var id in observed.Where(x => x.Value.Finished).Take(512).Select(x => x.Key).ToArray()) { observed.Remove(id); live.Remove(id); }
        }
        Changed?.Invoke();
    }
    private void Publish(WorkflowLogEntry entry)
    { entry.LogText = redactor.Redact(entry.LogText); XFEConsole.Log.AddLog(entry); }

    public async Task<IReadOnlyList<WorkflowLogEntry>> DiagnosticsAsync(WorkflowDefinition workflow, IReadOnlyList<WorkflowDiagnostic> diagnostics, CancellationToken ct = default)
    {
        var entries = diagnostics.Select(d => new WorkflowLogEntry
        {
            Category = "诊断", WorkflowId = workflow.Id, WorkflowName = workflow.Name, NodeId = d.NodeId,
            NodeName = d.NodeId ?? d.File ?? "编译与校验", Level = d.IsError ? LogLevel.Error : LogLevel.Warning,
            LogText = redactor.Redact(d.Message), Diagnostic = d with { Message = redactor.Redact(d.Message) }
        }).ToArray();
        await journalGate.WaitAsync(ct);
        try
        {
            var signature = JsonSerializer.Serialize(diagnostics, WorkflowJson.Options);
            if (diagnosticVersions.GetValueOrDefault(workflow.Id) == signature) return entries;
            diagnosticVersions[workflow.Id] = signature;
            if (entries.Length == 0) return entries;
            var path = WorkspacePaths.Inside(paths.Logs, workflow.Id + ".diagnostics.json");
            var previous = await AtomicJson.ReadAsync<List<WorkflowLogEntry>>(path, ct) ?? [];
            await AtomicJson.WriteAsync(path, previous.Concat(entries).TakeLast(5000).ToList(), ct);
            lock (gate) foreach (var entry in entries) Publish(entry);
        }
        finally { journalGate.Release(); }
        Changed?.Invoke(); return entries;
    }
    public async Task<List<WorkflowLogEntry>> HistoryAsync(CancellationToken ct = default)
    {
        var entries = (await runs.HistoryAsync(ct)).SelectMany(run => RunEntries(run)).ToList();
        lock (gate) entries.AddRange(live.Values.SelectMany(x => x));
        foreach (var file in Directory.EnumerateFiles(paths.Logs, "*.diagnostics.json"))
        {
            try { entries.AddRange(await AtomicJson.ReadAsync<List<WorkflowLogEntry>>(file, ct) ?? []); }
            catch (JsonException) { }
        }
        // Workflow entries in the XFE console are already represented by their
        // durable run record or diagnostic journal. Do not duplicate them.
        XFELogEntry[] application;
        try { lock (gate) application = XFEConsole.Log.Logs.ToArray(); }
        catch (FileNotFoundException) { application = []; } // File logs are created on the first write.
        entries.AddRange(application.Select((x, i) => (Entry: x, Index: i)).Where(x => x.Entry is not WorkflowLogEntry)
            .Select(x => new WorkflowLogEntry { Id = "application/" + x.Entry.Time.Ticks + "/" + x.Index, Time = x.Entry.Time, Level = x.Entry.Level, LogText = redactor.Redact(x.Entry.LogText) }));
        foreach (var entry in entries) entry.LogText = redactor.Redact(entry.LogText);
        return entries.DistinctBy(x => x.Id).OrderByDescending(x => x.Time).ToList();
    }
}