admin/Playa
1
0
Fork 0
forked from admin/VolcanicAgenticCore
Code Pull Requests Activity

feat(telegram): group registry, trigger store and snooze queue

Browse Source
This commit is contained in:
EugeneTes
2026-10-05 16:17:17 +00:00
parent af714cc50b
commit 50d0e81aac
5 changed files with 692 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
hosts/Volcanic.AgenticCore.Telegram/Groups/GroupMessage.cs Normal file
View File

@@ -0,0 +1,71 @@
using System.Globalization;
using Telegram.Bot.Types;
using Telegram.Bot.Types.Enums;
namespace Volcanic.AgenticCore.Telegram.Groups;
/// <summary>The message a group message replies to.</summary>
public sealed record GroupReply(string Sender, string? Text);
/// <summary>A group message as triggers see it: who, where, when, the text (or caption) and what it replies to.</summary>
public sealed record GroupMessage(
long GroupId,
string GroupTitle,
string? GroupUsername,
bool IsSupergroup,
int MessageId,
DateTimeOffset Time,
string Sender,
string? Text,
GroupReply? ReplyTo = null)
{
/// <summary>
/// A link to the message: <c>t.me/&lt;username&gt;/&lt;id&gt;</c> for public groups, <c>t.me/c/&lt;id without -100&gt;/&lt;id&gt;</c>
/// for private supergroups (members only), none for basic groups.
/// </summary>
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;
}
/// <summary>"First Last (@username)", or the chat an anonymous admin / channel posted as.</summary>
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})";
}
}

147
hosts/Volcanic.AgenticCore.Telegram/Groups/GroupRegistry.cs Normal file
View File

@@ -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;
/// <summary>A group the bot watches. Inactive once the bot left or was removed; its triggers stay but never fire.</summary>
public sealed record KnownGroup(long Id, string Title, string? Username, bool IsSupergroup, long AddedBy, DateTimeOffset Since, bool Active = true);
/// <summary>
/// The groups an allowed user vouched for (added the bot, or posted while it was there), in
/// <c>&lt;AgentDir&gt;/telegram/groups.json</c>: <c>{ "version": 1, "groups": [ … ] }</c>. Loaded once, written atomically
/// through one semaphore; a corrupt file is set aside and loading starts empty.
/// </summary>
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<long, KnownGroup> _groups = [];
public GroupRegistry(string path, ILogger<GroupRegistry> 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<KnownGroup> All()
{
lock (_lock) return [.. _groups.Values.OrderBy(g => g.Title, StringComparer.CurrentCultureIgnoreCase)];
}
/// <summary>Adds the group (or reactivates it) on behalf of allowed user <paramref name="by"/>. True when it changed.</summary>
public Task<bool> 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);
}
/// <summary>Keeps a known group's title and username current. No-op for unknown groups or when nothing changed.</summary>
public Task<bool> 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<bool> 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);
/// <summary>A group upgraded to a supergroup gets a new id.</summary>
public Task<bool> 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<bool> ChangeAsync(Func<Dictionary<long, KnownGroup>, 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<GroupsFile>(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<KnownGroup>? Groups);
}

104
hosts/Volcanic.AgenticCore.Telegram/Triggers/SnoozeQueue.cs Normal file
View File

@@ -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;
/// <summary>
/// The messages a snoozed trigger collects: <c>&lt;ownerWorkspace&gt;/.volcanic/triggers/snoozed/&lt;triggerId&gt;.jsonl</c>,
/// one <see cref="GroupMessage"/> per line, so a restart during a snooze loses nothing. Past
/// <c>maxPerTrigger</c> the oldest lines are dropped. One gate for every file: appends are rare and small.
/// </summary>
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");
/// <summary>Appends the message; returns the queue length and whether the oldest message was dropped to make room.</summary>
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();
}
}
/// <summary>Reads and deletes the queue in one step, so nothing appended in between is lost.</summary>
public async Task<IReadOnlyList<GroupMessage>> 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<GroupMessage>(lines.Length);
foreach (var l in lines)
{
if (l.Length == 0) continue;
try
{
if (VolcanicJson.Deserialize<GroupMessage>(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();
}
}
}

214
hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerStore.cs Normal file
View File

@@ -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;
/// <summary>
/// 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 <see cref="Threshold"/>.
/// </summary>
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;
}
/// <summary>A trigger and the private chat it belongs to.</summary>
public sealed record WatchingTrigger(long Owner, Trigger Trigger);
/// <summary>What <see cref="TriggerStore.AddAsync"/> needs; the store assigns the id and creation time.</summary>
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);
/// <summary>
/// Every owner's triggers, in <c>&lt;WorkspaceRoot&gt;/&lt;chatId&gt;/.volcanic/triggers/triggers.json</c>
/// (<c>{ "version": 1, "triggers": [ … ] }</c>). 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.
/// </summary>
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<long, List<Trigger>> _byOwner = [];
public TriggerStore(string workspaceRoot, int maxPerChat, ILogger<TriggerStore> logger, TimeProvider? time = null)
{
_workspaceRoot = workspaceRoot;
_maxPerChat = maxPerChat;
_logger = logger;
_time = time ?? TimeProvider.System;
Load();
}
/// <summary>The owner's working directory (its session's cwd).</summary>
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<Trigger> 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;
}
/// <summary>Every trigger (enabled or not) watching <paramref name="groupId"/>, across owners.</summary>
public IReadOnlyList<WatchingTrigger> Watching(long groupId)
{
lock (_lock)
return [.. _byOwner.SelectMany(kv => kv.Value.Where(t => t.GroupId == groupId).Select(t => new WatchingTrigger(kv.Key, t)))];
}
/// <summary>The soonest snooze end of any trigger, or null.</summary>
public DateTimeOffset? NextSnoozeEnd()
{
lock (_lock) return _byOwner.Values.SelectMany(l => l).Select(t => t.SnoozedUntil).Where(u => u is not null).Min();
}
/// <summary>Triggers whose snooze has ended by <paramref name="now"/> (still marked snoozed).</summary>
public IReadOnlyList<WatchingTrigger> 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<Trigger> 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!;
}
/// <summary>Applies <paramref name="change"/> to the trigger; null when the owner has no such trigger.</summary>
public async Task<Trigger?> UpdateAsync(long owner, string id, Func<Trigger, Trigger> 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<Trigger?> 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;
}
/// <summary>A group upgraded to a supergroup: every trigger on <paramref name="from"/> moves to <paramref name="to"/>.</summary>
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<List<Trigger>, 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<Trigger>(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<TriggersFile>(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<Trigger>? Triggers);
}