XFEWorkflow

using System.Net.Http.Json;
using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;
using XFEWorkflow.Abstractions;
using XFEWorkflow.Plugins.WeCom.Native.Core;

namespace XFEWorkflow.Plugins.WeCom;

public sealed class WeComPlugin : IWorkflowPlugin
{
    public PluginManifest Manifest
    {
        get
        {
            using var source = typeof(WeComPlugin).Assembly.GetManifestResourceStream("XFEWorkflow.Plugins.WeCom.plugin.json")!;
            return JsonSerializer.Deserialize<PluginManifest>(source, WorkflowJson.Options)!;
        }
    }
    public void Register(IPluginRegistry registry)
    {
        foreach (var capability in Manifest.Capabilities)
            if (capability.IsTrigger) registry.RegisterTrigger(capability, () => new WeComMessageTrigger());
            else registry.RegisterAction(capability, () => new WeComAction(capability.Id));
    }
}

public sealed class WeComMessageTrigger : IWorkflowTrigger
{
    public async Task RunAsync(JsonObject parameters, IWorkflowContext context, Func<JsonObject, ValueTask> emit, CancellationToken ct)
    {
        var receive = new JsonObject { ["_deliveryMode"] = "all", ["_workflowId"] = parameters["_workflowId"]?.DeepClone(), ["_priority"] = parameters["_priority"]?.DeepClone() };
        var subscription = (await context.InvokeHostAsync("wecom.subscribe", receive, ct)).String("subscription");
        try
        {
            while (true)
            {
                var message = await context.InvokeHostAsync("wecom.next", new() { ["subscription"] = subscription }, ct);
                await emit(message);
                // The desktop processes trigger events before reading this request. This
                // acknowledgement therefore happens after the triggered workflow finishes.
                await context.InvokeHostAsync("wecom.ack", new() { ["subscription"] = subscription, ["dispatchId"] = message.String("dispatchId") }, ct);
            }
        }
        finally
        {
            try { await context.InvokeHostAsync("wecom.unsubscribe", new() { ["subscription"] = subscription }, CancellationToken.None).AsTask().WaitAsync(TimeSpan.FromSeconds(2)); }
            catch (Exception) { /* The desktop also releases subscriptions when this Runner exits. */ }
        }
    }
}

public sealed class WeComAction(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)
    {
        ct.ThrowIfCancellationRequested();
        if (capabilityId == "xfe.wecom.send") return await SendWebhookAsync(input, context, ct);
        if (capabilityId.StartsWith("xfe.wecom.filter.", StringComparison.Ordinal)) return Filter(capabilityId, input);
        var operation = capabilityId["xfe.wecom.".Length..];
        var parameters = WorkflowJson.Copy(input);
        if (operation is "quoteReply" or "dryRun")
        { parameters[operation == "quoteReply" ? "quoteOriginal" : "dryRun"] = true; operation = "reply"; }
        var output = await context.InvokeHostAsync("wecom." + operation, parameters, ct);
        if (operation is "reply" or "selfTest") await context.LogAsync("企业微信 · " + output.Text(), ct);
        // A result object is available on every native action, alongside its named returns.
        output["result"] = output.DeepClone(); return output;
    }
    public static JsonObject Filter(string id, JsonObject input)
    {
        var message = input["message"] as JsonObject ?? throw new ArgumentException("请连接企业微信触发节点的 message 消息引用输出");
        bool matched;
        switch (id)
        {
            case "xfe.wecom.filter.group":
                var groupId = input.String("groupId"); var groupName = input.String("groupName");
                if (groupId.Length == 0 && groupName.Length == 0) throw new ArgumentException("请填写群 ID 或群名称");
                matched = (groupId.Length == 0 || message.String("conversationId") == groupId) && (groupName.Length == 0 || TextMatching.SameDisplayName(message.String("groupName"), groupName)); break;
            case "xfe.wecom.filter.member":
                var members = MemberIdFilter.Resolve(input.String("memberIds"), "");
                if (members.Any(x => !MemberIdFilter.IsNativeId(x))) throw new ArgumentException("请填写完整数字成员 ID");
                matched = members.Contains(message.String("senderId"), StringComparer.Ordinal); break;
            case "xfe.wecom.filter.kind":
                var kind = input.String("messageKind", "MergedForward");
                if (kind is not ("Any" or "Text" or "MergedForward")) throw new ArgumentException("未知消息类型");
                matched = kind == "Any" || message.String("kind") == kind; break;
            case "xfe.wecom.filter.keyword":
                if (!message.Boolean("hasContent")) throw new NotSupportedException("当前原生接口只提供消息元数据，没有正文，不能执行关键词筛选");
                var keywords = input.String("keywords").Split(['\r', '\n', ',', '，', ';', '；'], StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries).Select(TextMatching.Normalize).Where(x => x.Length > 0).ToArray();
                if (keywords.Length == 0) throw new ArgumentException("请填写至少一个关键词");
                var content = TextMatching.Normalize(message.String("content"));
                matched = input.Boolean("matchAll") ? keywords.All(x => content.Contains(x, StringComparison.OrdinalIgnoreCase)) : keywords.Any(x => content.Contains(x, StringComparison.OrdinalIgnoreCase)); break;
            default: throw new NotSupportedException(id);
        }
        return new() { ["matched"] = matched, ["branch"] = matched ? "match" : "noMatch", ["message"] = message.DeepClone() };
    }
    private static async Task<JsonObject> SendWebhookAsync(JsonObject input, IWorkflowContext context, CancellationToken ct)
    {
        var webhook = input.String("webhook"); if (input.String("credential") is { Length: > 0 } reference) webhook = await context.GetSecretAsync(reference, ct);
        if (!Uri.TryCreate(webhook, UriKind.Absolute, out var uri) || uri.Scheme is not ("http" or "https")) throw new ArgumentException("请输入 HTTP/HTTPS 群机器人地址或凭据引用");
        var format = input.String("format", "text"); if (format is not ("text" or "markdown")) throw new ArgumentException("机器人只支持 text 和 markdown");
        var payload = new JsonObject { ["content"] = input["content"].Text() };
        if (format == "text" && input["mentionedList"] is JsonArray mentions) payload["mentioned_list"] = mentions.DeepClone();
        using var request = new HttpRequestMessage(HttpMethod.Post, uri) { Content = new StringContent(new JsonObject { ["msgtype"] = format, [format] = payload }.ToJsonString(), Encoding.UTF8, "application/json") };
        using var response = await http.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, ct); response.EnsureSuccessStatusCode();
        if (response.Content.Headers.ContentLength > 1024 * 1024) throw new IOException("机器人响应超过 1 MiB");
        await using var stream = await response.Content.ReadAsStreamAsync(ct); using var buffer = new MemoryStream(); var bytes = new byte[8192];
        int read; while ((read = await stream.ReadAsync(bytes, ct)) > 0) { if (buffer.Length + read > 1024 * 1024) throw new IOException("机器人响应超过 1 MiB"); await buffer.WriteAsync(bytes.AsMemory(0, read), ct); }
        var result = JsonNode.Parse(buffer.ToArray())?.AsObject() ?? throw new IOException("机器人返回了空响应");
        if (result.Integer("errcode") != 0) throw new InvalidOperationException("企业微信拒绝通知：" + result.String("errmsg"));
        return new() { ["response"] = result };
    }
}