wx-win-agent/node-agent/WxAgent.Core/RemoteEventQueue.cs

using System.Security.Cryptography;
using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;

namespace WxAgent.Core;

public sealed record RemoteEventEnqueueResult(bool Accepted, string Reason, RemoteMessageEvent? Event);

public sealed class RemoteEventQueue
{
    private const int DefaultMaxItems = 1000;
    private readonly string _path;
    private readonly int _maxItems;
    private readonly object _gate = new();
    private QueueState _state;

    public RemoteEventQueue(string path, int maxItems = DefaultMaxItems)
    {
        if (maxItems is < 1 or > 100_000)
            throw new ArgumentOutOfRangeException(nameof(maxItems));
        _path = Path.GetFullPath(path);
        _maxItems = maxItems;
        _state = Load(_path);
    }

    public RemoteEventEnqueueResult Enqueue(
        ReportingConfig config,
        string nodeId,
        string accountId,
        string chatId,
        ReportingChatType chatType,
        string eventType,
        DateTimeOffset occurredAt,
        string? content)
    {
        var decision = ReportingAuthorization.Check(config, accountId, chatId, chatType, ReportingDataType.Message);
        if (!decision.Allowed)
            return new RemoteEventEnqueueResult(false, decision.Reason, null);
        RemoteAgentOptions.ValidateIdentifier(nodeId, "nodeId", 200);
        RemoteAgentOptions.ValidateIdentifier(eventType, "eventType", 80);
        RemoteAgentOptions.ValidateIdentifier(accountId, "accountId", 200);
        RemoteAgentOptions.ValidateIdentifier(chatId, "chatId", 512);
        if (content is { Length: > RemoteProtocol.MaxEventContentLength })
            throw new WxAgentException(WxAgentErrorCode.InvalidArgument, "Event content is too large.");

        lock (_gate)
        {
            if (_state.Items.Count >= _maxItems)
                return new RemoteEventEnqueueResult(false, "ReportingQueueFull", null);
            var key = Key(accountId, chatId);
            _state.Sequences.TryGetValue(key, out var previous);
            var messageEvent = new RemoteMessageEvent(nodeId, accountId, chatId, chatType, checked(previous + 1), eventType,
                occurredAt, content, config.ConfigVersion, decision.AuthorizationVersion,
                Guid.NewGuid().ToString("N"), true);
            _state.Sequences[key] = messageEvent.EventSeq;
            _state.Items.Add(messageEvent);
            SaveLocked();
            return new RemoteEventEnqueueResult(true, "Queued", messageEvent);
        }
    }

    public IReadOnlyList<RemoteMessageEvent> PrepareForSend(ReportingConfig config)
    {
        lock (_gate)
        {
            var ready = new List<RemoteMessageEvent>(_state.Items.Count);
            var kept = new List<RemoteMessageEvent>(_state.Items.Count);
            foreach (var item in _state.Items)
            {
                var decision = ReportingAuthorization.Check(config, item.AccountId, item.ChatId, item.ChatType, ReportingDataType.Message);
                if (!decision.Allowed)
                    continue;
                var refreshed = item with { AuthorizationVersion = decision.AuthorizationVersion, Authorized = true };
                ready.Add(refreshed);
                kept.Add(refreshed);
            }
            if (kept.Count != _state.Items.Count || !kept.SequenceEqual(_state.Items))
            {
                _state.Items = kept;
                SaveLocked();
            }
            return ready;
        }
    }

    public bool MarkSent(RemoteMessageEvent messageEvent)
    {
        lock (_gate)
        {
            var index = _state.Items.FindIndex(item => item.AccountId == messageEvent.AccountId
                && item.ChatId == messageEvent.ChatId
                && item.ChatType == messageEvent.ChatType
                && item.EventSeq == messageEvent.EventSeq
                && Fingerprint(item) == Fingerprint(messageEvent));
            if (index < 0) return false;
            _state.Items.RemoveAt(index);
            SaveLocked();
            return true;
        }
    }

    public int PendingCount
    {
        get { lock (_gate) return _state.Items.Count; }
    }

    private void SaveLocked()
    {
        var directory = Path.GetDirectoryName(_path) ?? AppContext.BaseDirectory;
        Directory.CreateDirectory(directory);
        var temporary = _path + ".tmp-" + Guid.NewGuid().ToString("N");
        try
        {
            File.WriteAllText(temporary, JsonSerializer.Serialize(_state, RemoteJson.Options), Encoding.UTF8);
            if (!OperatingSystem.IsWindows())
            {
                try { File.SetUnixFileMode(temporary, UnixFileMode.UserRead | UnixFileMode.UserWrite); }
                catch (PlatformNotSupportedException) { }
            }
            if (OperatingSystem.IsWindows() && File.Exists(_path)) File.Replace(temporary, _path, null);
            else File.Move(temporary, _path, true);
        }
        finally
        {
            if (File.Exists(temporary)) File.Delete(temporary);
        }
    }

    private static QueueState Load(string path)
    {
        if (!File.Exists(path)) return new QueueState();
        try
        {
            var state = JsonSerializer.Deserialize<QueueState>(File.ReadAllText(path), RemoteJson.Options) ?? new QueueState();
            state.Sequences ??= new(StringComparer.Ordinal);
            state.Items ??= [];
            return state;
        }
        catch (Exception exception) when (exception is IOException or JsonException or NotSupportedException or ArgumentException)
        {
            throw new WxAgentException(WxAgentErrorCode.InvalidOperationState, "Remote event queue could not be loaded; reporting is blocked.", exception);
        }
    }

    private static string Key(string accountId, string chatId) => $"{accountId}\u001f{chatId}";

    private static string Fingerprint(RemoteMessageEvent value)
    {
        var bytes = Encoding.UTF8.GetBytes($"{value.NodeId}\n{value.AccountId}\n{value.ChatId}\n{value.ChatType}\n{value.EventSeq}\n{value.EventType}\n{value.OccurredAt:O}\n{value.Content}\n{value.ConfigVersion}");
        return Convert.ToHexString(SHA256.HashData(bytes));
    }

    private sealed class QueueState
    {
        [JsonPropertyName("sequences")]
        public Dictionary<string, long> Sequences { get; set; } = new(StringComparer.Ordinal);

        [JsonPropertyName("items")]
        public List<RemoteMessageEvent> Items { get; set; } = [];
    }
}