AzuTear
72 lines
2.8 KiB
C#
72 lines
2.8 KiB
C#
namespace Nexus.Api.Services;
|
|
|
|
/// <summary>
|
|
/// Continuously folds bounded Gateway chat/run events into the durable Nexus
|
|
/// run projection. Re-processing is safe because per-run sequence cursors and
|
|
/// persisted terminal/gap event ids reject duplicates.
|
|
/// </summary>
|
|
public sealed class OpenClawRunEventReconciler(
|
|
IGatewayConnector connector,
|
|
IServiceScopeFactory scopeFactory,
|
|
ILogger<OpenClawRunEventReconciler> logger) : BackgroundService
|
|
{
|
|
private static readonly TimeSpan PollInterval = TimeSpan.FromMilliseconds(750);
|
|
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
string? cursor = null;
|
|
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
try
|
|
{
|
|
var snapshot = connector.GetRecentEvents(500)
|
|
.OrderBy(item => item.ReceivedAt)
|
|
.ThenBy(item => item.Sequence)
|
|
.ThenBy(OpenClawEventIdentity.Create, StringComparer.Ordinal)
|
|
.ToList();
|
|
var startIndex = cursor is null
|
|
? 0
|
|
: snapshot.FindIndex(item =>
|
|
string.Equals(
|
|
OpenClawEventIdentity.Create(item),
|
|
cursor,
|
|
StringComparison.Ordinal)) + 1;
|
|
if (startIndex <= 0 && cursor is not null)
|
|
{
|
|
// The bounded buffer rolled over. Re-process what remains;
|
|
// run-level sequence cursors and persisted event ids dedupe it.
|
|
startIndex = 0;
|
|
}
|
|
|
|
if (startIndex < snapshot.Count)
|
|
{
|
|
await using var scope = scopeFactory.CreateAsyncScope();
|
|
var runs = scope.ServiceProvider.GetRequiredService<IOpenClawRunService>();
|
|
foreach (var gatewayEvent in snapshot.Skip(startIndex))
|
|
{
|
|
await runs.ReconcileAsync(gatewayEvent, stoppingToken);
|
|
cursor = OpenClawEventIdentity.Create(gatewayEvent);
|
|
}
|
|
}
|
|
else if (snapshot.Count > 0)
|
|
{
|
|
cursor = OpenClawEventIdentity.Create(snapshot[^1]);
|
|
}
|
|
}
|
|
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
|
{
|
|
break;
|
|
}
|
|
catch (Exception exception)
|
|
{
|
|
logger.LogWarning(
|
|
exception,
|
|
"Could not reconcile the latest OpenClaw run events; retrying without dropping the durable projection.");
|
|
}
|
|
|
|
await Task.Delay(PollInterval, stoppingToken);
|
|
}
|
|
}
|
|
}
|