XFEWorkflow

using System.Net.Http.Headers;
using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;
using XFEWorkflow.Abstractions;

namespace XFEWorkflow.Plugins.Standard;

public abstract class StandardPlugin(string id) : IWorkflowPlugin
{
    public PluginManifest Manifest
    {
        get
        {
            using var stream = typeof(StandardPlugin).Assembly.GetManifestResourceStream("XFEWorkflow.Plugins.Standard.Manifests." + id + ".json")!;
            return JsonSerializer.Deserialize<PluginManifest>(stream, WorkflowJson.Options)!;
        }
    }
    public virtual void Register(IPluginRegistry registry)
    {
        foreach (var capability in Manifest.Capabilities)
        {
            if (capability.IsTrigger) registry.RegisterTrigger(capability, () => new TimerTrigger());
            else registry.RegisterAction(capability, () => new StandardAction(capability.Id));
        }
    }
}
public sealed class HttpPlugin() : StandardPlugin("xfe.http");
public sealed class FilePlugin() : StandardPlugin("xfe.file");
public sealed class AiPlugin() : StandardPlugin("xfe.ai");
public sealed class WeComPlugin() : StandardPlugin("xfe.wecom");
public sealed class BrowserPlugin() : StandardPlugin("xfe.browser");
public sealed class TimerPlugin() : StandardPlugin("xfe.timer");
public sealed class CommandPlugin() : StandardPlugin("xfe.command");

public sealed class TimerTrigger : IWorkflowTrigger
{
    public async Task RunAsync(JsonObject parameters, IWorkflowContext context, Func<JsonObject, ValueTask> emit, CancellationToken cancellationToken)
    {
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(Math.Max(1, parameters.Integer("seconds", 60))));
        while (await timer.WaitForNextTickAsync(cancellationToken)) await emit(new() { ["timestamp"] = DateTimeOffset.UtcNow.ToString("O") });
    }
}

public sealed class StandardAction(string capabilityId) : IWorkflowAction
{
    private static readonly HttpClient Http = new(new SocketsHttpHandler { PooledConnectionLifetime = TimeSpan.FromMinutes(5) })
    { Timeout = Timeout.InfiniteTimeSpan };

    public async ValueTask<JsonObject> ExecuteAsync(JsonObject input, IWorkflowContext context, CancellationToken ct)
    {
        if (capabilityId.StartsWith("xfe.browser.", StringComparison.Ordinal))
            return await context.InvokeHostAsync("browser." + capabilityId["xfe.browser.".Length..], input, ct);
        if (capabilityId is "xfe.command.powershell" or "xfe.command.cmd")
        {
            var command = WorkflowJson.Copy(input); command["shell"] = capabilityId["xfe.command.".Length..];
            var result = await context.InvokeHostAsync("command.run", command, ct); result["result"] = result.DeepClone(); return result;
        }
        switch (capabilityId)
        {
            case "xfe.http.request":
                using (var request = new HttpRequestMessage(new HttpMethod(input.String("method", "GET")), RequireHttp(input.String("url"))))
                {
                    if (input["body"] is { } body) request.Content = new StringContent(body.Text(), Encoding.UTF8, input.String("contentType", "application/json"));
                    if (input["headers"] is JsonObject headers)
                        foreach (var pair in headers)
                            if (!request.Headers.TryAddWithoutValidation(pair.Key, pair.Value.Text()))
                                request.Content?.Headers.TryAddWithoutValidation(pair.Key, pair.Value.Text());
                    using var response = await Http.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, ct);
                    var text = await ReadBoundedAsync(response, ct);
                    JsonNode? json = null;
                    try { json = JsonNode.Parse(text); } catch (JsonException) { }
                    if (input.Boolean("ensureSuccess", true)) response.EnsureSuccessStatusCode();
                    return new() { ["status"] = (int)response.StatusCode, ["text"] = text, ["json"] = json,
                        ["headers"] = JsonSerializer.SerializeToNode(response.Headers.ToDictionary(x => x.Key, x => string.Join(", ", x.Value))) };
                }
            case "xfe.file.read":
                var path = Path.GetFullPath(input.String("path"));
                if (new FileInfo(path).Length > 32 * 1024 * 1024) throw new IOException("文本文件超过 32 MiB 限制");
                var content = await File.ReadAllTextAsync(path, Encoding.UTF8, ct);
                JsonNode? parsed = null;
                try { parsed = JsonNode.Parse(content); } catch (JsonException) { }
                return new() { ["text"] = content, ["json"] = parsed };
            case "xfe.file.write":
                var destination = Path.GetFullPath(input.String("path"));
                Directory.CreateDirectory(Path.GetDirectoryName(destination)!);
                if (input.Boolean("append")) await File.AppendAllTextAsync(destination, input["text"].Text(), Encoding.UTF8, ct);
                else await File.WriteAllTextAsync(destination, input["text"].Text(), Encoding.UTF8, ct);
                return new() { ["path"] = destination };
            case "xfe.file.list":
                return new() { ["files"] = new JsonArray(Directory.EnumerateFiles(input.String("path"), input.String("pattern", "*"),
                    input.Boolean("recursive") ? SearchOption.AllDirectories : SearchOption.TopDirectoryOnly)
                    .Take(100_000).Select(x => (JsonNode?)JsonValue.Create(x)).ToArray()) };
            case "xfe.ai.chat":
                var endpoint = RequireHttp(input.String("endpoint", "https://api.openai.com/v1/")).ToString().TrimEnd('/');
                if (!endpoint.EndsWith("/chat/completions", StringComparison.Ordinal)) endpoint += "/chat/completions";
                var messages = input["messages"] as JsonArray ?? new JsonArray(
                    new JsonObject { ["role"] = "system", ["content"] = input.String("system", "你是一个自动化助手。") },
                    new JsonObject { ["role"] = "user", ["content"] = input["prompt"].Text() });
                var payload = new JsonObject { ["model"] = input.String("model"), ["messages"] = messages.DeepClone(), ["stream"] = false };
                if (input["temperature"] is { } temperature) payload["temperature"] = temperature.DeepClone();
                using (var request = JsonRequest(endpoint, payload))
                {
                    var apiKey = input.String("apiKey");
                    if (input.String("credential") is { Length: > 0 } reference) apiKey = await context.GetSecretAsync(reference, ct);
                    if (apiKey.Length > 0) request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", apiKey);
                    using var response = await Http.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, ct);
                    response.EnsureSuccessStatusCode();
                    var result = JsonNode.Parse(await ReadBoundedAsync(response, ct))!.AsObject();
                    return new() { ["text"] = result["choices"]?[0]?["message"]?["content"]?.DeepClone(), ["response"] = result };
                }
            case "xfe.wecom.send":
                var webhook = input.String("webhook");
                if (input.String("credential") is { Length: > 0 } secret) webhook = await context.GetSecretAsync(secret, ct);
                RequireHttp(webhook);
                var format = input.String("format", "text");
                if (format is not ("text" or "markdown")) throw new ArgumentException("机器人只支持 text 和 markdown");
                var data = new JsonObject { ["content"] = input["content"].Text() };
                if (format == "text" && input["mentionedList"] is JsonArray mentioned) data["mentioned_list"] = mentioned.DeepClone();
                using (var request = JsonRequest(webhook, new() { ["msgtype"] = format, [format] = data }))
                {
                    using var response = await Http.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, ct);
                    response.EnsureSuccessStatusCode();
                    var result = JsonNode.Parse(await ReadBoundedAsync(response, ct))!.AsObject();
                    if (result.Integer("errcode") != 0) throw new InvalidOperationException("企业微信拒绝通知：" + result.String("errmsg"));
                    return new() { ["response"] = result };
                }
            default: throw new NotSupportedException("未知能力：" + capabilityId);
        }
    }
    private static Uri RequireHttp(string value)
    {
        if (!Uri.TryCreate(value, UriKind.Absolute, out var uri) || uri.Scheme is not ("https" or "http")) throw new ArgumentException("必须使用 HTTP 或 HTTPS 地址");
        return uri;
    }
    private static HttpRequestMessage JsonRequest(string url, JsonObject body) => new(HttpMethod.Post, url)
    { Content = new StringContent(body.ToJsonString(), Encoding.UTF8, "application/json") };
    private static async Task<string> ReadBoundedAsync(HttpResponseMessage response, CancellationToken ct)
    {
        const int limit = 32 * 1024 * 1024;
        if (response.Content.Headers.ContentLength > limit) throw new IOException("HTTP 响应超过 32 MiB 限制");
        await using var stream = await response.Content.ReadAsStreamAsync(ct);
        using var buffer = new MemoryStream();
        var block = new byte[81920];
        while (true)
        {
            var read = await stream.ReadAsync(block, ct);
            if (read == 0) break;
            if (buffer.Length + read > limit) throw new IOException("HTTP 响应超过 32 MiB 限制");
            await buffer.WriteAsync(block.AsMemory(0, read), ct);
        }
        return Encoding.UTF8.GetString(buffer.GetBuffer(), 0, (int)buffer.Length);
    }
}