using System.Data; using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.Storage; using Microsoft.Extensions.Options; using Nexus.Api.Data; namespace Nexus.Api.Services; /// /// Durable, metadata-only idempotency claims for OpenClaw mutations. /// The legacy JSONL path is exposed for archive discovery only; this store /// never reads from or appends to that file. /// public sealed class PostgresOpenClawOperationAuditStore( IServiceScopeFactory scopeFactory, IOptions options) : IOpenClawOperationAuditStore { private const string OperationNamespace = "openclaw.mutation"; private static readonly TimeSpan TerminalRetention = TimeSpan.FromDays(7); private readonly SemaphoreSlim processGate = new(1, 1); public string AuditPath { get; } = OpenClawOperationAuditStore.ResolveAuditPath( options.Value.OperationAuditPath, options.Value.DeviceStatePath); public async Task ClaimAsync( OpenClawInvocationContext context, OpenClawOperationDescriptor operation, CancellationToken cancellationToken = default) { Validate(operation); var keyHash = OpenClawInvocationContextFactory.Hash( context.IdempotencyKey); var requestHash = RequestHash(operation); await processGate.WaitAsync(cancellationToken); try { await using var scope = scopeFactory.CreateAsyncScope(); var db = scope.ServiceProvider.GetRequiredService(); await using var transaction = await BeginTransactionAsync( db, cancellationToken); await AcquireDatabaseLockAsync( db, keyHash, cancellationToken); var now = DateTimeOffset.UtcNow; var existing = await db.OperationClaims.SingleOrDefaultAsync( item => item.Operation == OperationNamespace && item.IdempotencyKeyHash == keyHash, cancellationToken); if (existing is not null) { var terminal = existing.CompletedAt is not null; if (terminal && existing.ExpiresAt <= now) { db.OperationClaims.Remove(existing); await db.SaveChangesAsync(cancellationToken); existing = null; } else { var result = ExistingClaim(existing, requestHash); await CommitAsync(transaction, cancellationToken); return result; } } db.OperationClaims.Add(new OperationClaim { Operation = OperationNamespace, IdempotencyKeyHash = keyHash, RequestHash = requestHash, State = "started", CreatedAt = now, ExpiresAt = now.Add(TerminalRetention) }); await db.SaveChangesAsync(cancellationToken); await CommitAsync(transaction, cancellationToken); return new OpenClawOperationClaim( OpenClawOperationClaimDisposition.Started); } finally { processGate.Release(); } } public async Task CompleteAsync( OpenClawInvocationContext context, OpenClawOperationDescriptor operation, bool ok, string state, string message, string? errorCode = null, CancellationToken cancellationToken = default) { Validate(operation); var keyHash = OpenClawInvocationContextFactory.Hash( context.IdempotencyKey); var requestHash = RequestHash(operation); await processGate.WaitAsync(cancellationToken); try { await using var scope = scopeFactory.CreateAsyncScope(); var db = scope.ServiceProvider.GetRequiredService(); await using var transaction = await BeginTransactionAsync( db, cancellationToken); await AcquireDatabaseLockAsync( db, keyHash, cancellationToken); var claim = await db.OperationClaims.SingleOrDefaultAsync( item => item.Operation == OperationNamespace && item.IdempotencyKeyHash == keyHash, cancellationToken); if (claim is null || !string.Equals( claim.RequestHash, requestHash, StringComparison.Ordinal)) { throw new InvalidOperationException( "OpenClaw operation must be claimed before it is completed."); } var resultState = NormalizeResultState(state); if (claim.CompletedAt is not null) { var sameOutcome = string.Equals( claim.State, ok ? "completed" : "failed", StringComparison.Ordinal) && string.Equals( claim.ResultCode, resultState, StringComparison.Ordinal); if (!sameOutcome) { throw new InvalidOperationException( "OpenClaw operation already has a different terminal result."); } await CommitAsync(transaction, cancellationToken); return; } var now = DateTimeOffset.UtcNow; claim.State = ok ? "completed" : "failed"; claim.ResultCode = resultState; claim.CompletedAt = now; claim.ExpiresAt = now.Add(TerminalRetention); await db.SaveChangesAsync(cancellationToken); await CommitAsync(transaction, cancellationToken); } finally { processGate.Release(); } } private static OpenClawOperationClaim ExistingClaim( OperationClaim existing, string requestHash) { if (!string.Equals( existing.RequestHash, requestHash, StringComparison.Ordinal)) { return new OpenClawOperationClaim( OpenClawOperationClaimDisposition.Conflict, PreviousMessage: "Der Idempotency-Key wurde bereits für eine andere Aktion verwendet."); } if (existing.CompletedAt is null) { return new OpenClawOperationClaim( OpenClawOperationClaimDisposition.InDoubt, PreviousMessage: "Die frühere Aktion besitzt kein bestätigtes terminales Ergebnis."); } var ok = string.Equals( existing.State, "completed", StringComparison.Ordinal); return new OpenClawOperationClaim( OpenClawOperationClaimDisposition.Replayed, ok, existing.ResultCode, ok ? "Das bereits bestätigte Operationsergebnis wurde wiederverwendet." : "Das bereits protokollierte Fehlerergebnis wurde wiederverwendet."); } private static async Task BeginTransactionAsync( NexusDbContext db, CancellationToken cancellationToken) { if (!db.Database.IsRelational()) return null; return await db.Database.BeginTransactionAsync( IsolationLevel.ReadCommitted, cancellationToken); } private static async Task AcquireDatabaseLockAsync( NexusDbContext db, string keyHash, CancellationToken cancellationToken) { if (!string.Equals( db.Database.ProviderName, "Npgsql.EntityFrameworkCore.PostgreSQL", StringComparison.Ordinal)) { return; } await db.Database.ExecuteSqlInterpolatedAsync( $"SELECT pg_advisory_xact_lock(hashtextextended({keyHash}, 0))", cancellationToken); } private static Task CommitAsync( IDbContextTransaction? transaction, CancellationToken cancellationToken) => transaction is null ? Task.CompletedTask : transaction.CommitAsync(cancellationToken); private static string RequestHash( OpenClawOperationDescriptor operation) => OpenClawInvocationContextFactory.Hash(string.Join( '\u001f', operation.Method.Trim(), operation.TargetType.Trim(), operation.TargetId.Trim(), operation.IntentFingerprint.Trim())); private static string NormalizeResultState(string value) { if (string.IsNullOrWhiteSpace(value)) return "unknown"; var normalized = new string(value .Trim() .ToLowerInvariant() .Where(character => char.IsAsciiLetterOrDigit(character) || character is '_' or '-' or '.') .Take(120) .ToArray()); return string.IsNullOrWhiteSpace(normalized) ? "unknown" : normalized; } private static void Validate(OpenClawOperationDescriptor operation) { if (string.IsNullOrWhiteSpace(operation.Method) || string.IsNullOrWhiteSpace(operation.TargetType) || string.IsNullOrWhiteSpace(operation.TargetId) || string.IsNullOrWhiteSpace(operation.IntentFingerprint)) { throw new ArgumentException( "OpenClaw operation audit metadata is incomplete.", nameof(operation)); } } }