diff --git a/hosts/Volcanic.AgenticCore.Telegram/Groups/GroupMessage.cs b/hosts/Volcanic.AgenticCore.Telegram/Groups/GroupMessage.cs new file mode 100644 index 0000000..864c25c --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Groups/GroupMessage.cs @@ -0,0 +1,71 @@ +using System.Globalization; +using Telegram.Bot.Types; +using Telegram.Bot.Types.Enums; + +namespace Volcanic.AgenticCore.Telegram.Groups; + +/// The message a group message replies to. +public sealed record GroupReply(string Sender, string? Text); + +/// A group message as triggers see it: who, where, when, the text (or caption) and what it replies to. +public sealed record GroupMessage( + long GroupId, + string GroupTitle, + string? GroupUsername, + bool IsSupergroup, + int MessageId, + DateTimeOffset Time, + string Sender, + string? Text, + GroupReply? ReplyTo = null) +{ + /// + /// A link to the message: t.me/<username>/<id> for public groups, t.me/c/<id without -100>/<id> + /// for private supergroups (members only), none for basic groups. + /// + public string? Link + { + get + { + var mid = MessageId.ToString(CultureInfo.InvariantCulture); + if (!string.IsNullOrEmpty(GroupUsername)) return $"https://t.me/{GroupUsername}/{mid}"; + if (!IsSupergroup) return null; + var id = (-GroupId).ToString(CultureInfo.InvariantCulture); + return id.StartsWith("100", StringComparison.Ordinal) && id.Length > 3 ? $"https://t.me/c/{id[3..]}/{mid}" : null; + } + } + + public static GroupMessage From(Message message) + { + ArgumentNullException.ThrowIfNull(message); + var reply = message.ReplyToMessage is { } replied + ? new GroupReply(SenderOf(replied), replied.Text ?? replied.Caption) + : null; + return new GroupMessage( + message.Chat.Id, + TitleOf(message.Chat), + message.Chat.Username, + message.Chat.Type == ChatType.Supergroup, + message.Id, + new DateTimeOffset(DateTime.SpecifyKind(message.Date, DateTimeKind.Utc)), + SenderOf(message), + message.Text ?? message.Caption, + reply); + } + + public static string TitleOf(Chat chat) + { + ArgumentNullException.ThrowIfNull(chat); + return string.IsNullOrWhiteSpace(chat.Title) ? chat.Id.ToString(CultureInfo.InvariantCulture) : chat.Title; + } + + /// "First Last (@username)", or the chat an anonymous admin / channel posted as. + public static string SenderOf(Message message) + { + ArgumentNullException.ThrowIfNull(message); + if (message.SenderChat is { } chat) return TitleOf(chat); + if (message.From is not { } user) return "unknown"; + var name = string.IsNullOrWhiteSpace(user.LastName) ? user.FirstName : $"{user.FirstName} {user.LastName}"; + return string.IsNullOrWhiteSpace(user.Username) ? name : $"{name} (@{user.Username})"; + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Groups/GroupRegistry.cs b/hosts/Volcanic.AgenticCore.Telegram/Groups/GroupRegistry.cs new file mode 100644 index 0000000..0080fb5 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Groups/GroupRegistry.cs @@ -0,0 +1,147 @@ +using System.Globalization; +using System.Text.Json; +using Microsoft.Extensions.Logging; +using Telegram.Bot.Types; +using Telegram.Bot.Types.Enums; +using Volcanic.AgenticCore.Abstractions.Json; + +namespace Volcanic.AgenticCore.Telegram.Groups; + +/// A group the bot watches. Inactive once the bot left or was removed; its triggers stay but never fire. +public sealed record KnownGroup(long Id, string Title, string? Username, bool IsSupergroup, long AddedBy, DateTimeOffset Since, bool Active = true); + +/// +/// The groups an allowed user vouched for (added the bot, or posted while it was there), in +/// <AgentDir>/telegram/groups.json: { "version": 1, "groups": [ … ] }. Loaded once, written atomically +/// through one semaphore; a corrupt file is set aside and loading starts empty. +/// +public sealed class GroupRegistry +{ + public const int Version = 1; + private readonly string _path; + private readonly ILogger _logger; + private readonly TimeProvider _time; + private readonly SemaphoreSlim _writeGate = new(1, 1); + private readonly Lock _lock = new(); + private readonly Dictionary _groups = []; + + public GroupRegistry(string path, ILogger logger, TimeProvider? time = null) + { + _path = path; + _logger = logger; + _time = time ?? TimeProvider.System; + Load(); + } + + public static string DefaultPath(string agentDir) => Path.Combine(agentDir, "telegram", "groups.json"); + + public KnownGroup? Find(long id) + { + lock (_lock) return _groups.GetValueOrDefault(id); + } + + public IReadOnlyList All() + { + lock (_lock) return [.. _groups.Values.OrderBy(g => g.Title, StringComparer.CurrentCultureIgnoreCase)]; + } + + /// Adds the group (or reactivates it) on behalf of allowed user . True when it changed. + public Task VouchAsync(Chat chat, long by, CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(chat); + return ChangeAsync(groups => + { + if (groups.TryGetValue(chat.Id, out var known)) + { + var updated = known with { Title = GroupMessage.TitleOf(chat), Username = chat.Username, IsSupergroup = chat.Type == ChatType.Supergroup, Active = true }; + if (updated == known) return false; + groups[chat.Id] = updated; + return true; + } + groups[chat.Id] = new KnownGroup(chat.Id, GroupMessage.TitleOf(chat), chat.Username, chat.Type == ChatType.Supergroup, by, _time.GetUtcNow()); + _logger.LogInformation("Group {GroupId} \"{Title}\" is now watched (vouched for by {UserId})", chat.Id, GroupMessage.TitleOf(chat), by); + return true; + }, cancellationToken); + } + + /// Keeps a known group's title and username current. No-op for unknown groups or when nothing changed. + public Task RefreshAsync(Chat chat, CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(chat); + lock (_lock) + { + if (!_groups.TryGetValue(chat.Id, out var known) || (known.Title == GroupMessage.TitleOf(chat) && known.Username == chat.Username)) + return Task.FromResult(false); + } + return ChangeAsync(groups => + { + if (!groups.TryGetValue(chat.Id, out var known)) return false; + groups[chat.Id] = known with { Title = GroupMessage.TitleOf(chat), Username = chat.Username }; + return true; + }, cancellationToken); + } + + public Task SetInactiveAsync(long id, CancellationToken cancellationToken = default) => ChangeAsync(groups => + { + if (!groups.TryGetValue(id, out var known) || !known.Active) return false; + groups[id] = known with { Active = false }; + _logger.LogInformation("Group {GroupId} \"{Title}\": the bot left; its triggers are paused", id, known.Title); + return true; + }, cancellationToken); + + /// A group upgraded to a supergroup gets a new id. + public Task MigrateAsync(long from, long to, CancellationToken cancellationToken = default) => ChangeAsync(groups => + { + if (!groups.Remove(from, out var known)) return false; + groups[to] = known with { Id = to, IsSupergroup = true }; + _logger.LogInformation("Group {From} became supergroup {To}", from, to); + return true; + }, cancellationToken); + + private async Task ChangeAsync(Func, bool> change, CancellationToken cancellationToken) + { + await _writeGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + GroupsFile snapshot; + lock (_lock) + { + if (!change(_groups)) return false; + snapshot = new GroupsFile(Version, [.. _groups.Values.OrderBy(g => g.Id)]); + } + Directory.CreateDirectory(Path.GetDirectoryName(_path)!); + var temp = $"{_path}.{Environment.ProcessId.ToString(CultureInfo.InvariantCulture)}.tmp"; + await File.WriteAllTextAsync(temp, JsonSerializer.Serialize(snapshot, VolcanicJson.Indented), CancellationToken.None).ConfigureAwait(false); + File.Move(temp, _path, overwrite: true); + return true; + } + finally + { + _writeGate.Release(); + } + } + + private void Load() + { + if (!File.Exists(_path)) return; + GroupsFile? file; + try + { + file = VolcanicJson.Deserialize(File.ReadAllText(_path)); + } + catch (JsonException) + { + file = null; + } + if (file is not { Version: Version, Groups: not null }) + { + var aside = $"{_path}.corrupt-{_time.GetUtcNow().ToString("yyyyMMddHHmmss", CultureInfo.InvariantCulture)}"; + File.Move(_path, aside, overwrite: true); + _logger.LogError("Groups file {Path} could not be read; moved to {Aside}, no group is watched until an allowed user vouches again", _path, aside); + return; + } + foreach (var group in file.Groups) _groups[group.Id] = group; + } + + private sealed record GroupsFile(int Version, List? Groups); +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Triggers/SnoozeQueue.cs b/hosts/Volcanic.AgenticCore.Telegram/Triggers/SnoozeQueue.cs new file mode 100644 index 0000000..f756640 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Triggers/SnoozeQueue.cs @@ -0,0 +1,104 @@ +using System.Text; +using System.Text.Json; +using Volcanic.AgenticCore.Abstractions.Json; +using Volcanic.AgenticCore.Telegram.Groups; + +namespace Volcanic.AgenticCore.Telegram.Triggers; + +/// +/// The messages a snoozed trigger collects: <ownerWorkspace>/.volcanic/triggers/snoozed/<triggerId>.jsonl, +/// one per line, so a restart during a snooze loses nothing. Past +/// maxPerTrigger the oldest lines are dropped. One gate for every file: appends are rare and small. +/// +public sealed class SnoozeQueue(int maxPerTrigger) +{ + private readonly SemaphoreSlim _gate = new(1, 1); + + public static string PathFor(string workspace, string triggerId) => + Path.Combine(TriggerStore.DirectoryIn(workspace), "snoozed", triggerId + ".jsonl"); + + /// Appends the message; returns the queue length and whether the oldest message was dropped to make room. + public async Task<(int Count, bool Dropped)> AppendAsync(string workspace, string triggerId, GroupMessage message, CancellationToken cancellationToken) + { + var path = PathFor(workspace, triggerId); + var line = JsonSerializer.Serialize(message, VolcanicJson.Options); + await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + Directory.CreateDirectory(Path.GetDirectoryName(path)!); + var lines = File.Exists(path) ? File.ReadAllLines(path).Where(l => l.Length > 0).ToList() : []; + if (lines.Count < maxPerTrigger) + { + await File.AppendAllTextAsync(path, line + "\n", CancellationToken.None).ConfigureAwait(false); + return (lines.Count + 1, false); + } + lines.RemoveRange(0, lines.Count - maxPerTrigger + 1); + lines.Add(line); + await File.WriteAllTextAsync(path, string.Join('\n', lines) + "\n", CancellationToken.None).ConfigureAwait(false); + return (lines.Count, true); + } + finally + { + _gate.Release(); + } + } + + /// Reads and deletes the queue in one step, so nothing appended in between is lost. + public async Task> TakeAllAsync(string workspace, string triggerId, CancellationToken cancellationToken) + { + var path = PathFor(workspace, triggerId); + await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + if (!File.Exists(path)) return []; + var lines = await File.ReadAllLinesAsync(path, Encoding.UTF8, CancellationToken.None).ConfigureAwait(false); + File.Delete(path); + var messages = new List(lines.Length); + foreach (var l in lines) + { + if (l.Length == 0) continue; + try + { + if (VolcanicJson.Deserialize(l) is { } m) messages.Add(m); + } + catch (JsonException) + { + // a torn line from a crash mid-append: skip it + } + } + return messages; + } + finally + { + _gate.Release(); + } + } + + public int Count(string workspace, string triggerId) + { + var path = PathFor(workspace, triggerId); + _gate.Wait(); + try + { + return File.Exists(path) ? File.ReadLines(path).Count(l => l.Length > 0) : 0; + } + finally + { + _gate.Release(); + } + } + + public async Task DeleteAsync(string workspace, string triggerId, CancellationToken cancellationToken) + { + await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + var path = PathFor(workspace, triggerId); + if (File.Exists(path)) File.Delete(path); + } + finally + { + _gate.Release(); + } + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerStore.cs b/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerStore.cs new file mode 100644 index 0000000..9599228 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerStore.cs @@ -0,0 +1,214 @@ +using System.Globalization; +using System.Text.Json; +using Microsoft.Extensions.Logging; +using Volcanic.AgenticCore.Abstractions.Json; + +namespace Volcanic.AgenticCore.Telegram.Triggers; + +/// +/// A topic trigger: a yes/no question Jev asks about every message of one group, owned by one private chat. It matches +/// when the yes-probability reaches . +/// +public sealed record Trigger( + string Id, + string Name, + long GroupId, + string GroupTitle, + string Instructions, + string? CriteriaTrue, + string? CriteriaFalse, + double Threshold, + bool Enabled, + DateTimeOffset? SnoozedUntil, + DateTimeOffset CreatedAt) +{ + public bool IsSnoozed(DateTimeOffset now) => SnoozedUntil is { } until && until > now; +} + +/// A trigger and the private chat it belongs to. +public sealed record WatchingTrigger(long Owner, Trigger Trigger); + +/// What needs; the store assigns the id and creation time. +public sealed record NewTrigger(string Name, long GroupId, string GroupTitle, string Instructions, string? CriteriaTrue, string? CriteriaFalse, double Threshold); + +public sealed class TriggerLimitException(string message) : Exception(message); + +/// +/// Every owner's triggers, in <WorkspaceRoot>/<chatId>/.volcanic/triggers/triggers.json +/// ({ "version": 1, "triggers": [ … ] }). All files are loaded at construction into one index by group; every +/// change rewrites the owner's file atomically under one semaphore. A corrupt file is set aside. +/// +public sealed class TriggerStore +{ + public const int Version = 1; + private readonly string _workspaceRoot; + private readonly int _maxPerChat; + private readonly ILogger _logger; + private readonly TimeProvider _time; + private readonly SemaphoreSlim _writeGate = new(1, 1); + private readonly Lock _lock = new(); + private readonly Dictionary> _byOwner = []; + + public TriggerStore(string workspaceRoot, int maxPerChat, ILogger logger, TimeProvider? time = null) + { + _workspaceRoot = workspaceRoot; + _maxPerChat = maxPerChat; + _logger = logger; + _time = time ?? TimeProvider.System; + Load(); + } + + /// The owner's working directory (its session's cwd). + public string WorkspaceOf(long owner) => Path.Combine(_workspaceRoot, owner.ToString(CultureInfo.InvariantCulture)); + + public static string DirectoryIn(string workspace) => Path.Combine(workspace, ".volcanic", "triggers"); + + public static string PathIn(string workspace) => Path.Combine(DirectoryIn(workspace), "triggers.json"); + + public IReadOnlyList List(long owner) + { + lock (_lock) return _byOwner.TryGetValue(owner, out var list) ? [.. list] : []; + } + + public Trigger? Find(long owner, string id) + { + lock (_lock) return _byOwner.TryGetValue(owner, out var list) ? list.Find(t => string.Equals(t.Id, id, StringComparison.OrdinalIgnoreCase)) : null; + } + + /// Every trigger (enabled or not) watching , across owners. + public IReadOnlyList Watching(long groupId) + { + lock (_lock) + return [.. _byOwner.SelectMany(kv => kv.Value.Where(t => t.GroupId == groupId).Select(t => new WatchingTrigger(kv.Key, t)))]; + } + + /// The soonest snooze end of any trigger, or null. + public DateTimeOffset? NextSnoozeEnd() + { + lock (_lock) return _byOwner.Values.SelectMany(l => l).Select(t => t.SnoozedUntil).Where(u => u is not null).Min(); + } + + /// Triggers whose snooze has ended by (still marked snoozed). + public IReadOnlyList SnoozeDue(DateTimeOffset now) + { + lock (_lock) + return [.. _byOwner.SelectMany(kv => kv.Value.Where(t => t.SnoozedUntil is { } u && u <= now).Select(t => new WatchingTrigger(kv.Key, t)))]; + } + + public async Task AddAsync(long owner, NewTrigger trigger, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(trigger); + Trigger? created = null; + await ChangeAsync(owner, list => + { + if (list.Count >= _maxPerChat) + throw new TriggerLimitException($"This chat already has {list.Count} triggers (the limit is {_maxPerChat}); delete one first."); + var next = list.Select(t => int.TryParse(t.Id.AsSpan(1), NumberStyles.None, CultureInfo.InvariantCulture, out var n) ? n : 0).DefaultIfEmpty(0).Max() + 1; + created = new Trigger("t" + next.ToString(CultureInfo.InvariantCulture), trigger.Name, trigger.GroupId, trigger.GroupTitle, trigger.Instructions, + trigger.CriteriaTrue, trigger.CriteriaFalse, trigger.Threshold, Enabled: true, SnoozedUntil: null, _time.GetUtcNow()); + list.Add(created); + return true; + }, cancellationToken).ConfigureAwait(false); + return created!; + } + + /// Applies to the trigger; null when the owner has no such trigger. + public async Task UpdateAsync(long owner, string id, Func change, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(change); + Trigger? updated = null; + await ChangeAsync(owner, list => + { + var index = list.FindIndex(t => string.Equals(t.Id, id, StringComparison.OrdinalIgnoreCase)); + if (index < 0) return false; + updated = change(list[index]) with { Id = list[index].Id, CreatedAt = list[index].CreatedAt }; + list[index] = updated; + return true; + }, cancellationToken).ConfigureAwait(false); + return updated; + } + + public async Task RemoveAsync(long owner, string id, CancellationToken cancellationToken) + { + Trigger? removed = null; + await ChangeAsync(owner, list => + { + var index = list.FindIndex(t => string.Equals(t.Id, id, StringComparison.OrdinalIgnoreCase)); + if (index < 0) return false; + removed = list[index]; + list.RemoveAt(index); + return true; + }, cancellationToken).ConfigureAwait(false); + return removed; + } + + /// A group upgraded to a supergroup: every trigger on moves to . + public async Task MoveGroupAsync(long from, long to, CancellationToken cancellationToken) + { + long[] owners; + lock (_lock) owners = [.. _byOwner.Where(kv => kv.Value.Any(t => t.GroupId == from)).Select(kv => kv.Key)]; + foreach (var owner in owners) + await ChangeAsync(owner, list => + { + for (var i = 0; i < list.Count; i++) + if (list[i].GroupId == from) list[i] = list[i] with { GroupId = to }; + return true; + }, cancellationToken).ConfigureAwait(false); + } + + private async Task ChangeAsync(long owner, Func, bool> change, CancellationToken cancellationToken) + { + await _writeGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + TriggersFile snapshot; + lock (_lock) + { + var list = _byOwner.TryGetValue(owner, out var existing) ? existing : []; + var working = new List(list); + if (!change(working)) return; + _byOwner[owner] = working; + snapshot = new TriggersFile(Version, working); + } + var path = PathIn(WorkspaceOf(owner)); + Directory.CreateDirectory(Path.GetDirectoryName(path)!); + var temp = $"{path}.{Environment.ProcessId.ToString(CultureInfo.InvariantCulture)}.tmp"; + await File.WriteAllTextAsync(temp, JsonSerializer.Serialize(snapshot, VolcanicJson.Indented), CancellationToken.None).ConfigureAwait(false); + File.Move(temp, path, overwrite: true); + } + finally + { + _writeGate.Release(); + } + } + + private void Load() + { + if (!Directory.Exists(_workspaceRoot)) return; + foreach (var dir in Directory.EnumerateDirectories(_workspaceRoot)) + { + if (!long.TryParse(Path.GetFileName(dir), NumberStyles.AllowLeadingSign, CultureInfo.InvariantCulture, out var owner)) continue; + var path = PathIn(dir); + if (!File.Exists(path)) continue; + TriggersFile? file; + try + { + file = VolcanicJson.Deserialize(File.ReadAllText(path)); + } + catch (JsonException) + { + file = null; + } + if (file is not { Version: Version, Triggers: not null }) + { + var aside = $"{path}.corrupt-{_time.GetUtcNow().ToString("yyyyMMddHHmmss", CultureInfo.InvariantCulture)}"; + File.Move(path, aside, overwrite: true); + _logger.LogError("Triggers of chat {ChatId} ({Path}) could not be read; moved to {Aside}", owner, path, aside); + continue; + } + _byOwner[owner] = file.Triggers; + } + } + + private sealed record TriggersFile(int Version, List? Triggers); +} diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/TriggerStoreTests.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/TriggerStoreTests.cs new file mode 100644 index 0000000..e80c766 --- /dev/null +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/TriggerStoreTests.cs @@ -0,0 +1,156 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Telegram.Bot.Types; +using Telegram.Bot.Types.Enums; +using Volcanic.AgenticCore.Telegram.Groups; +using Volcanic.AgenticCore.Telegram.Triggers; + +namespace Volcanic.AgenticCore.Telegram.Tests; + +/// The group registry, the trigger store and the snooze queue: persistence and bookkeeping. +public sealed class TriggerStoreTests : IDisposable +{ + private const long Owner = 1001; + private const long OtherOwner = 2002; + private const long Group = -1001234567890; + private readonly string _root = Path.Combine(Path.GetTempPath(), "volcanic-triggers", Guid.NewGuid().ToString("N")); + + public void Dispose() + { + try { Directory.Delete(_root, true); } catch (IOException) { } + } + + private TriggerStore Store(int max = 20) => new(Path.Combine(_root, "workspaces"), max, NullLogger.Instance); + + private static NewTrigger Deploys(long group = Group) => new("deploys", group, "Dev Team", "Is this about deploys?", "Mentions deploys", "Small talk", 0.7); + + internal static GroupMessage Message(string text, int id = 1, string sender = "Alice (@alice)", long group = Group) => + new(group, "Dev Team", null, true, id, new DateTimeOffset(2026, 10, 5, 12, 2, 0, TimeSpan.Zero), sender, text); + + [Fact] + public async Task Triggers_get_sequential_ids_per_owner_and_survive_a_reload() + { + var store = Store(); + Assert.Equal("t1", (await store.AddAsync(Owner, Deploys(), CancellationToken.None)).Id); + Assert.Equal("t2", (await store.AddAsync(Owner, Deploys(), CancellationToken.None)).Id); + Assert.Equal("t1", (await store.AddAsync(OtherOwner, Deploys(), CancellationToken.None)).Id); + await store.RemoveAsync(Owner, "t1", CancellationToken.None); + Assert.Equal("t3", (await store.AddAsync(Owner, Deploys(), CancellationToken.None)).Id); + + var reloaded = Store(); + Assert.Equal(["t2", "t3"], reloaded.List(Owner).Select(t => t.Id)); + Assert.Equal(3, reloaded.Watching(Group).Count); + Assert.True(File.Exists(TriggerStore.PathIn(reloaded.WorkspaceOf(Owner)))); + } + + [Fact] + public async Task The_per_chat_limit_is_enforced() + { + var store = Store(max: 1); + await store.AddAsync(Owner, Deploys(), CancellationToken.None); + await Assert.ThrowsAsync(() => store.AddAsync(Owner, Deploys(), CancellationToken.None)); + await store.AddAsync(OtherOwner, Deploys(), CancellationToken.None); // per chat, not global + } + + [Fact] + public async Task Updates_keep_id_and_creation_time_and_snoozes_are_tracked() + { + var store = Store(); + var t = await store.AddAsync(Owner, Deploys(), CancellationToken.None); + var until = DateTimeOffset.UtcNow.AddHours(1); + + var updated = await store.UpdateAsync(Owner, "T1", x => x with { Id = "hijack", Threshold = 0.9, SnoozedUntil = until }, CancellationToken.None); + + Assert.Equal(("t1", t.CreatedAt, 0.9), (updated!.Id, updated.CreatedAt, updated.Threshold)); + Assert.Equal(until, store.NextSnoozeEnd()); + Assert.Empty(store.SnoozeDue(until.AddSeconds(-1))); + Assert.Single(store.SnoozeDue(until)); + Assert.Null(await store.UpdateAsync(Owner, "t9", x => x, CancellationToken.None)); + } + + [Fact] + public async Task Moving_a_group_moves_every_owners_triggers() + { + var store = Store(); + await store.AddAsync(Owner, Deploys(-5), CancellationToken.None); + await store.AddAsync(OtherOwner, Deploys(-5), CancellationToken.None); + + await store.MoveGroupAsync(-5, Group, CancellationToken.None); + + Assert.Empty(store.Watching(-5)); + Assert.Equal(2, Store().Watching(Group).Count); + } + + [Fact] + public async Task A_corrupt_file_is_set_aside() + { + var store = Store(); + var path = TriggerStore.PathIn(store.WorkspaceOf(Owner)); + Directory.CreateDirectory(Path.GetDirectoryName(path)!); + await File.WriteAllTextAsync(path, "{ not json"); + + Assert.Empty(Store().List(Owner)); + Assert.Single(Directory.GetFiles(Path.GetDirectoryName(path)!, "triggers.json.corrupt-*")); + } + + [Fact] + public async Task The_snooze_queue_caps_drops_the_oldest_and_takes_all_at_once() + { + var queue = new SnoozeQueue(maxPerTrigger: 2); + var ws = Path.Combine(_root, "ws"); + Assert.Equal((1, false), await queue.AppendAsync(ws, "t1", Message("one", 1), CancellationToken.None)); + Assert.Equal((2, false), await queue.AppendAsync(ws, "t1", Message("two", 2), CancellationToken.None)); + Assert.Equal((2, true), await queue.AppendAsync(ws, "t1", Message("three", 3), CancellationToken.None)); + Assert.Equal(2, queue.Count(ws, "t1")); + + var taken = await queue.TakeAllAsync(ws, "t1", CancellationToken.None); + + Assert.Equal(["two", "three"], taken.Select(m => m.Text)); + Assert.Equal(Message("two", 2), taken[0]); + Assert.Equal(0, queue.Count(ws, "t1")); + Assert.Empty(await queue.TakeAllAsync(ws, "t1", CancellationToken.None)); + await queue.DeleteAsync(Path.Combine(_root, "missing"), "t1", CancellationToken.None); + } + + [Fact] + public async Task Groups_are_vouched_refreshed_paused_migrated_and_persisted() + { + var path = GroupRegistry.DefaultPath(Path.Combine(_root, "agent")); + var registry = new GroupRegistry(path, NullLogger.Instance); + var chat = new Chat { Id = -5, Type = ChatType.Group, Title = "Dev" }; + + Assert.True(await registry.VouchAsync(chat, Owner)); + Assert.False(await registry.VouchAsync(chat, Owner)); + Assert.True(await registry.RefreshAsync(new Chat { Id = -5, Type = ChatType.Group, Title = "Dev Team" })); + Assert.False(await registry.RefreshAsync(new Chat { Id = -9, Type = ChatType.Group, Title = "Unknown" })); + Assert.True(await registry.SetInactiveAsync(-5)); + Assert.False(registry.Find(-5)!.Active); + Assert.True(await registry.VouchAsync(new Chat { Id = -5, Type = ChatType.Group, Title = "Dev Team" }, Owner)); + Assert.True(registry.Find(-5)!.Active); + Assert.True(await registry.MigrateAsync(-5, Group)); + + var reloaded = new GroupRegistry(path, NullLogger.Instance); + var group = Assert.Single(reloaded.All()); + Assert.Equal((Group, "Dev Team", true, Owner, true), (group.Id, group.Title, group.IsSupergroup, group.AddedBy, group.Active)); + } + + [Fact] + public void Message_links_and_senders() + { + Assert.Equal("https://t.me/c/1234567890/7", (Message("x") with { MessageId = 7 }).Link); + Assert.Equal("https://t.me/devteam/7", (Message("x") with { MessageId = 7, GroupUsername = "devteam" }).Link); + Assert.Null((Message("x") with { IsSupergroup = false, GroupId = -5 }).Link); + + var message = new Message + { + Id = 3, + Chat = new Chat { Id = Group, Type = ChatType.Supergroup, Title = "Dev Team" }, + From = new User { Id = 7, FirstName = "Alice", LastName = "Smith", Username = "alice" }, + Caption = "deploy log attached", + Date = new DateTime(2026, 10, 5, 12, 0, 0, DateTimeKind.Utc), + ReplyToMessage = new Message { Id = 2, Chat = new Chat { Id = Group }, From = new User { Id = 8, FirstName = "Bob" }, Text = "status?" }, + }; + var built = GroupMessage.From(message); + Assert.Equal(("Alice Smith (@alice)", "deploy log attached", "Bob", "status?"), (built.Sender, built.Text, built.ReplyTo!.Sender, built.ReplyTo.Text)); + Assert.Equal("Dev Team", GroupMessage.SenderOf(new Message { Chat = new Chat { Id = Group }, SenderChat = new Chat { Id = Group, Title = "Dev Team" } })); + } +}