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 PrepareForSend(ReportingConfig config) { lock (_gate) { var ready = new List(_state.Items.Count); var kept = new List(_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(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 Sequences { get; set; } = new(StringComparer.Ordinal); [JsonPropertyName("items")] public List Items { get; set; } = []; } }