forked from admin/VolcanicAgenticCore
106 lines
4.0 KiB
C#
106 lines
4.0 KiB
C#
using Microsoft.Extensions.Logging;
|
|
using Volcanic.AgenticCore.Telegram.Groups;
|
|
|
|
namespace Volcanic.AgenticCore.Telegram.Triggers;
|
|
|
|
/// <summary>Where trigger prompts go: the owner's private session.</summary>
|
|
public interface ITriggerDelivery
|
|
{
|
|
/// <summary>Runs <paramref name="prompt"/> in the owner's session (as a follow-up while it is busy).</summary>
|
|
Task DeliverAsync(long owner, string prompt, CancellationToken cancellationToken);
|
|
|
|
TimeZoneInfo TimeZoneFor(long owner);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Collects live matches per (owner, trigger): the first match opens a <see cref="TriggerOptions.Debounce"/> window,
|
|
/// later ones join it without extending it (a busy group can't hold a match back), and the window becomes one prompt.
|
|
/// </summary>
|
|
public sealed class TriggerDispatcher(ITriggerDelivery delivery, TriggerOptions options, TimeProvider time, ILogger<TriggerDispatcher> logger) : IDisposable
|
|
{
|
|
private sealed class Window(Trigger trigger)
|
|
{
|
|
public Trigger Trigger { get; set; } = trigger;
|
|
public List<(GroupMessage Message, double Score)> Hits { get; } = [];
|
|
public ITimer? Timer { get; set; }
|
|
}
|
|
|
|
private readonly Lock _lock = new();
|
|
private readonly Dictionary<(long Owner, string Trigger), Window> _windows = [];
|
|
private readonly List<Task> _deliveries = [];
|
|
|
|
public void Add(long owner, Trigger trigger, GroupMessage message, double score)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(trigger);
|
|
ArgumentNullException.ThrowIfNull(message);
|
|
var key = (owner, trigger.Id);
|
|
Window window;
|
|
lock (_lock)
|
|
{
|
|
if (_windows.TryGetValue(key, out var open))
|
|
{
|
|
open.Trigger = trigger;
|
|
open.Hits.Add((message, score));
|
|
return;
|
|
}
|
|
window = new Window(trigger);
|
|
window.Hits.Add((message, score));
|
|
_windows[key] = window;
|
|
}
|
|
// Outside the lock: a zero window may fire inside CreateTimer.
|
|
var timer = time.CreateTimer(_ => Fire(key, window), null, options.Debounce, Timeout.InfiniteTimeSpan);
|
|
lock (_lock)
|
|
{
|
|
if (_windows.TryGetValue(key, out var current) && ReferenceEquals(current, window)) window.Timer = timer;
|
|
else timer.Dispose(); // already fired
|
|
}
|
|
}
|
|
|
|
/// <summary>Completes when every prompt handed to the delivery so far has been delivered (for tests and shutdown).</summary>
|
|
public Task IdleAsync()
|
|
{
|
|
lock (_lock) return Task.WhenAll([.. _deliveries]);
|
|
}
|
|
|
|
private void Fire((long Owner, string Trigger) key, Window window)
|
|
{
|
|
List<(GroupMessage, double)> hits;
|
|
lock (_lock)
|
|
{
|
|
if (!_windows.Remove(key, out var current) || !ReferenceEquals(current, window)) return;
|
|
window.Timer?.Dispose();
|
|
hits = [.. window.Hits];
|
|
}
|
|
var prompt = TriggerPrompts.Live(window.Trigger, hits, delivery.TimeZoneFor(key.Owner));
|
|
var task = DeliverAsync(key.Owner, window.Trigger, prompt, hits.Count);
|
|
lock (_lock)
|
|
{
|
|
_deliveries.RemoveAll(t => t.IsCompleted);
|
|
_deliveries.Add(task);
|
|
}
|
|
}
|
|
|
|
private async Task DeliverAsync(long owner, Trigger trigger, string prompt, int count)
|
|
{
|
|
try
|
|
{
|
|
await Task.Yield(); // never deliver on the timer thread
|
|
logger.LogInformation("Trigger {TriggerId} \"{Name}\": {Count} match(es) delivered to chat {Owner}", trigger.Id, trigger.Name, count, owner);
|
|
await delivery.DeliverAsync(owner, prompt, CancellationToken.None).ConfigureAwait(false);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
logger.LogError(ex, "Delivering trigger {TriggerId} to chat {Owner} failed", trigger.Id, owner);
|
|
}
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
lock (_lock)
|
|
{
|
|
foreach (var window in _windows.Values) window.Timer?.Dispose();
|
|
_windows.Clear();
|
|
}
|
|
}
|
|
}
|