nexus/backend/Services/PostgresOpenClawOperationAuditStore.cs

using System.Data;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Storage;
using Microsoft.Extensions.Options;
using Nexus.Api.Data;

namespace Nexus.Api.Services;

/// <summary>
/// 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.
/// </summary>
public sealed class PostgresOpenClawOperationAuditStore(
    IServiceScopeFactory scopeFactory,
    IOptions<GatewayConnectorOptions> 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<OpenClawOperationClaim> 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<NexusDbContext>();
            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<NexusDbContext>();
            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<IDbContextTransaction?> 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));
        }
    }
}