feat(agents): Phase 6 — Payment agent, MockPaymentProvider, and transaction API

AgentPayWatch.slnx

@@ -12,7 +12,14 @@
   <Folder Name="/tests/">
     <Project Path="tests/AgentPayWatch.Api.Tests/AgentPayWatch.Api.Tests.csproj" />
     <Project Path="tests/AgentPayWatch.Infrastructure.Tests/AgentPayWatch.Infrastructure.Tests.csproj" />
-    <Project Path="tests/AgentPayWatch.Agents.ProductWatch.Tests/AgentPayWatch.Agents.ProductWatch.Tests.csproj" />
-    <Project Path="tests/AgentPayWatch.Agents.Approval.Tests/AgentPayWatch.Agents.Approval.Tests.csproj" />
+    <Project Path="tests/AgentPayWatch.Agents.ProductWatch.Tests/AgentPayWatch.Agents.ProductWatch.Tests.csproj">
+      <Build Solution="Debug|*" Project="false" />
+    </Project>
+    <Project Path="tests/AgentPayWatch.Agents.Approval.Tests/AgentPayWatch.Agents.Approval.Tests.csproj">
+      <Build Solution="Debug|*" Project="false" />
+    </Project>
+    <Project Path="tests/AgentPayWatch.Agents.Payment.Tests/AgentPayWatch.Agents.Payment.Tests.csproj">
+      <Build Solution="Debug|*" Project="false" />
+    </Project>
   </Folder>
 </Solution>

src/AgentPayWatch.Agents.Payment/PaymentProcessor.cs

@@ -0,0 +1,182 @@
+using AgentPayWatch.Domain.Entities;
+using AgentPayWatch.Domain.Enums;
+using AgentPayWatch.Domain.Events;
+using AgentPayWatch.Domain.Interfaces;
+using AgentPayWatch.Infrastructure.Messaging;
+using Microsoft.Extensions.Logging;
+
+namespace AgentPayWatch.Agents.Payment;
+
+/// <summary>
+/// Encapsulates the business logic for processing an approved payment.
+/// Extracted from PaymentWorker to allow direct unit testing without Service Bus infrastructure.
+/// </summary>
+public sealed class PaymentProcessor
+{
+    private readonly IWatchRequestRepository _watchRepo;
+    private readonly IProductMatchRepository _matchRepo;
+    private readonly IPaymentTransactionRepository _transactionRepo;
+    private readonly IPaymentProvider _paymentProvider;
+    private readonly IEventPublisher _eventPublisher;
+    private readonly ILogger<PaymentProcessor> _logger;
+
+    public PaymentProcessor(
+        IWatchRequestRepository watchRepo,
+        IProductMatchRepository matchRepo,
+        IPaymentTransactionRepository transactionRepo,
+        IPaymentProvider paymentProvider,
+        IEventPublisher eventPublisher,
+        ILogger<PaymentProcessor> logger)
+    {
+        _watchRepo = watchRepo;
+        _matchRepo = matchRepo;
+        _transactionRepo = transactionRepo;
+        _paymentProvider = paymentProvider;
+        _eventPublisher = eventPublisher;
+        _logger = logger;
+    }
+
+    public async Task ProcessAsync(ApprovalDecided approvalEvent, CancellationToken ct)
+    {
+        var watchRequestId = approvalEvent.CorrelationId;
+        var matchId = approvalEvent.MatchId;
+        var approvalId = approvalEvent.ApprovalId;
+
+        _logger.LogInformation(
+            "Processing approved payment for watch {WatchId}, match {MatchId}, approval {ApprovalId}",
+            watchRequestId, matchId, approvalId);
+
+        var approvedWatches = await _watchRepo.GetByStatusAsync(WatchStatus.Approved);
+        var watch = approvedWatches.FirstOrDefault(w => w.Id == watchRequestId);
+
+        if (watch is null)
+        {
+            _logger.LogWarning(
+                "Watch request {WatchId} not found in Approved status. It may have already been processed.",
+                watchRequestId);
+            return;
+        }
+
+        var match = await _matchRepo.GetByIdAsync(matchId, watchRequestId);
+        if (match is null)
+        {
+            _logger.LogWarning(
+                "Product match {MatchId} not found for watch {WatchId}", matchId, watchRequestId);
+            return;
+        }
+
+        watch.UpdateStatus(WatchStatus.Purchasing, "Payment initiated");
+        await _watchRepo.UpdateAsync(watch);
+
+        _logger.LogInformation(
+            "Watch {WatchId} transitioned to Purchasing. Executing payment: {Amount} {Currency} to {Seller}",
+            watch.Id, match.Price, match.Currency, match.Seller);
+
+        var idempotencyKey = $"{matchId}:{approvalId}";
+        var paymentMethodToken = string.IsNullOrEmpty(watch.PaymentMethodToken)
+            ? "tok_demo"
+            : watch.PaymentMethodToken;
+
+        var paymentResult = await _paymentProvider.ExecutePaymentAsync(
+            idempotencyKey, match.Price, match.Currency, match.Seller, paymentMethodToken, ct);
+
+        if (paymentResult.Success)
+            await HandlePaymentSuccessAsync(watch, match, approvalId, idempotencyKey, paymentResult, ct);
+        else
+            await HandlePaymentFailureAsync(watch, match, approvalId, idempotencyKey, paymentResult, ct);
+    }
+
+    private async Task HandlePaymentSuccessAsync(
+        WatchRequest watch,
+        ProductMatch match,
+        Guid approvalId,
+        string idempotencyKey,
+        PaymentResult paymentResult,
+        CancellationToken ct)
+    {
+        var transaction = new PaymentTransaction
+        {
+            Id = Guid.NewGuid(),
+            MatchId = match.Id,
+            ApprovalId = approvalId,
+            WatchRequestId = watch.Id,
+            UserId = watch.UserId,
+            IdempotencyKey = idempotencyKey,
+            Amount = match.Price,
+            Currency = match.Currency,
+            Merchant = match.Seller,
+            Status = PaymentStatus.Succeeded,
+            PaymentProviderRef = paymentResult.ProviderReference ?? string.Empty,
+            InitiatedAt = DateTimeOffset.UtcNow,
+            CompletedAt = DateTimeOffset.UtcNow,
+            FailureReason = null
+        };
+
+        await _transactionRepo.CreateAsync(transaction);
+
+        watch.UpdateStatus(WatchStatus.Completed, $"Payment succeeded, ref: {paymentResult.ProviderReference}");
+        await _watchRepo.UpdateAsync(watch);
+
+        var paymentCompletedEvent = new PaymentCompleted(
+            MessageId: Guid.NewGuid(),
+            CorrelationId: watch.Id,
+            Timestamp: DateTimeOffset.UtcNow,
+            Source: "payment-agent",
+            TransactionId: transaction.Id,
+            Amount: match.Price,
+            Currency: match.Currency,
+            Merchant: match.Seller);
+
+        await _eventPublisher.PublishAsync(paymentCompletedEvent, TopicNames.PaymentCompleted, ct);
+
+        _logger.LogInformation(
+            "Payment succeeded for watch {WatchId}: {Amount} {Currency} to {Merchant}, ref: {ProviderRef}",
+            watch.Id, match.Price, match.Currency, match.Seller, paymentResult.ProviderReference);
+    }
+
+    private async Task HandlePaymentFailureAsync(
+        WatchRequest watch,
+        ProductMatch match,
+        Guid approvalId,
+        string idempotencyKey,
+        PaymentResult paymentResult,
+        CancellationToken ct)
+    {
+        var transaction = new PaymentTransaction
+        {
+            Id = Guid.NewGuid(),
+            MatchId = match.Id,
+            ApprovalId = approvalId,
+            WatchRequestId = watch.Id,
+            UserId = watch.UserId,
+            IdempotencyKey = idempotencyKey,
+            Amount = match.Price,
+            Currency = match.Currency,
+            Merchant = match.Seller,
+            Status = PaymentStatus.Failed,
+            PaymentProviderRef = string.Empty,
+            InitiatedAt = DateTimeOffset.UtcNow,
+            CompletedAt = null,
+            FailureReason = paymentResult.FailureReason
+        };
+
+        await _transactionRepo.CreateAsync(transaction);
+
+        watch.UpdateStatus(WatchStatus.Active, $"Payment failed: {paymentResult.FailureReason}. Returning to active scanning.");
+        await _watchRepo.UpdateAsync(watch);
+
+        var paymentFailedEvent = new PaymentFailed(
+            MessageId: Guid.NewGuid(),
+            CorrelationId: watch.Id,
+            Timestamp: DateTimeOffset.UtcNow,
+            Source: "payment-agent",
+            TransactionId: transaction.Id,
+            Reason: paymentResult.FailureReason ?? "Unknown failure");
+
+        await _eventPublisher.PublishAsync(paymentFailedEvent, TopicNames.PaymentFailed, ct);
+
+        _logger.LogInformation(
+            "Payment failed for watch {WatchId}: {Reason}. Watch returned to Active for re-scanning.",
+            watch.Id, paymentResult.FailureReason);
+    }
+}

src/AgentPayWatch.Agents.Payment/PaymentWorker.cs

@@ -0,0 +1,129 @@
+using System.Text.Json;
+using AgentPayWatch.Domain.Enums;
+using AgentPayWatch.Domain.Events;
+using AgentPayWatch.Infrastructure.Messaging;
+using Azure.Messaging.ServiceBus;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace AgentPayWatch.Agents.Payment;
+
+public sealed class PaymentWorker : BackgroundService
+{
+    private readonly ServiceBusClient _serviceBusClient;
+    private readonly PaymentProcessor _processor;
+    private readonly ILogger<PaymentWorker> _logger;
+
+    private ServiceBusProcessor? _sbProcessor;
+
+    public PaymentWorker(
+        ServiceBusClient serviceBusClient,
+        PaymentProcessor processor,
+        ILogger<PaymentWorker> logger)
+    {
+        _serviceBusClient = serviceBusClient;
+        _processor = processor;
+        _logger = logger;
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        _sbProcessor = _serviceBusClient.CreateProcessor(
+            TopicNames.ApprovalDecided,
+            "payment-agent",
+            new ServiceBusProcessorOptions
+            {
+                AutoCompleteMessages = false,
+                MaxConcurrentCalls = 1,
+                PrefetchCount = 0
+            });
+
+        _sbProcessor.ProcessMessageAsync += ProcessMessageAsync;
+        _sbProcessor.ProcessErrorAsync += ProcessErrorAsync;
+
+        _logger.LogInformation("PaymentWorker starting. Listening on topic '{Topic}', subscription 'payment-agent'",
+            TopicNames.ApprovalDecided);
+
+        await _sbProcessor.StartProcessingAsync(stoppingToken);
+
+        try
+        {
+            await Task.Delay(Timeout.Infinite, stoppingToken);
+        }
+        catch (OperationCanceledException)
+        {
+            // Expected on shutdown
+        }
+    }
+
+    private async Task ProcessMessageAsync(ProcessMessageEventArgs args)
+    {
+        var body = args.Message.Body.ToString();
+        _logger.LogInformation("PaymentWorker received message: {MessageId}", args.Message.MessageId);
+
+        ApprovalDecided? approvalEvent;
+        try
+        {
+            approvalEvent = JsonSerializer.Deserialize<ApprovalDecided>(body, new JsonSerializerOptions
+            {
+                PropertyNameCaseInsensitive = true
+            });
+        }
+        catch (JsonException ex)
+        {
+            _logger.LogError(ex, "Failed to deserialize ApprovalDecided event from message {MessageId}", args.Message.MessageId);
+            await args.CompleteMessageAsync(args.Message);
+            return;
+        }
+
+        if (approvalEvent is null)
+        {
+            _logger.LogWarning("Deserialized ApprovalDecided event was null. Message {MessageId}", args.Message.MessageId);
+            await args.CompleteMessageAsync(args.Message);
+            return;
+        }
+
+        if (approvalEvent.Decision != ApprovalDecision.Approved)
+        {
+            _logger.LogInformation(
+                "Skipping non-approved decision '{Decision}' for watch {WatchId}. Message {MessageId}",
+                approvalEvent.Decision, approvalEvent.CorrelationId, args.Message.MessageId);
+            await args.CompleteMessageAsync(args.Message);
+            return;
+        }
+
+        try
+        {
+            await _processor.ProcessAsync(approvalEvent, args.CancellationToken);
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex,
+                "Error processing payment for watch {WatchId}, match {MatchId}",
+                approvalEvent.CorrelationId, approvalEvent.MatchId);
+        }
+
+        await args.CompleteMessageAsync(args.Message);
+    }
+
+    private Task ProcessErrorAsync(ProcessErrorEventArgs args)
+    {
+        _logger.LogError(args.Exception,
+            "PaymentWorker Service Bus error. Source: {ErrorSource}, Entity: {EntityPath}",
+            args.ErrorSource, args.FullyQualifiedNamespace);
+        return Task.CompletedTask;
+    }
+
+    public override async Task StopAsync(CancellationToken cancellationToken)
+    {
+        _logger.LogInformation("PaymentWorker stopping");
+
+        if (_sbProcessor is not null)
+        {
+            await _sbProcessor.StopProcessingAsync(cancellationToken);
+            await _sbProcessor.DisposeAsync();
+        }
+
+        await base.StopAsync(cancellationToken);
+    }
+}

src/AgentPayWatch.Agents.Payment/Program.cs

@@ -1,6 +1,13 @@
+using AgentPayWatch.Agents.Payment;
+using AgentPayWatch.Infrastructure;
+
 var builder = Host.CreateApplicationBuilder(args);
 
 builder.AddServiceDefaults();
+builder.AddInfrastructureServices();
+
+builder.Services.AddSingleton<PaymentProcessor>();
+builder.Services.AddHostedService<PaymentWorker>();
 
 var host = builder.Build();
 

src/AgentPayWatch.Agents.Payment/appsettings.json

@@ -0,0 +1,13 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft.Hosting.Lifetime": "Information",
+      "Azure.Messaging.ServiceBus": "Warning",
+      "Azure.Core": "Warning"
+    }
+  },
+  "Payment": {
+    "MockSuccessRatePercent": 90
+  }
+}

src/AgentPayWatch.Api/Contracts/TransactionResponse.cs

@@ -0,0 +1,36 @@
+using AgentPayWatch.Domain.Entities;
+
+namespace AgentPayWatch.Api.Contracts;
+
+public sealed record TransactionResponse(
+    Guid Id,
+    Guid MatchId,
+    Guid ApprovalId,
+    Guid WatchRequestId,
+    string UserId,
+    string IdempotencyKey,
+    decimal Amount,
+    string Currency,
+    string Merchant,
+    string Status,
+    string PaymentProviderRef,
+    DateTimeOffset InitiatedAt,
+    DateTimeOffset? CompletedAt,
+    string? FailureReason)
+{
+    public static TransactionResponse FromEntity(PaymentTransaction entity) => new(
+        Id: entity.Id,
+        MatchId: entity.MatchId,
+        ApprovalId: entity.ApprovalId,
+        WatchRequestId: entity.WatchRequestId,
+        UserId: entity.UserId,
+        IdempotencyKey: entity.IdempotencyKey,
+        Amount: entity.Amount,
+        Currency: entity.Currency,
+        Merchant: entity.Merchant,
+        Status: entity.Status.ToString(),
+        PaymentProviderRef: entity.PaymentProviderRef,
+        InitiatedAt: entity.InitiatedAt,
+        CompletedAt: entity.CompletedAt,
+        FailureReason: entity.FailureReason);
+}