fix(infra): don't advance account cursor unless all transactions proceed successfully

src/FiscalOS.Infra/Transactions/Plaid/PlaidAddedTransactionHandler.cs

@@ -5,28 +5,48 @@ namespace FiscalOS.Infra.Transactions.Plaid;
 
 internal interface IPlaidAddedTransactionHandler
 {
-  void Handle(Account account, IEnumerable<Going.Plaid.Entity.Transaction> added);
+  Task<int> HandleAsync(
+    Account account,
+    IEnumerable<Going.Plaid.Entity.Transaction> added,
+    CancellationToken ct
+  );
 }
 
 internal sealed class PlaidAddedTransactionHandler : IPlaidAddedTransactionHandler
 {
   private readonly ILogger<PlaidAddedTransactionHandler> _logger;
+  private readonly AppDbContext _appDbContext;
 
-  private PlaidAddedTransactionHandler(ILogger<PlaidAddedTransactionHandler> logger)
+  private PlaidAddedTransactionHandler(
+    ILogger<PlaidAddedTransactionHandler> logger,
+    AppDbContext appDbContext
+  )
   {
     _logger = logger;
+    _appDbContext = appDbContext;
   }
 
   public static PlaidAddedTransactionHandler From(IServiceProvider serviceProvider)
   {
     return new(
-      serviceProvider.GetRequiredService<ILogger<PlaidAddedTransactionHandler>>()
+      serviceProvider.GetRequiredService<ILogger<PlaidAddedTransactionHandler>>(),
+      serviceProvider.GetRequiredService<AppDbContext>()
     );
   }
 
-  public void Handle(Account account, IEnumerable<Going.Plaid.Entity.Transaction> added)
+  public async Task<int> HandleAsync(Account account, IEnumerable<Going.Plaid.Entity.Transaction> added, CancellationToken ct)
   {
+    var existingTransactionIds = added.Select(t => t.TransactionId);
+    var existingTransactions = await _appDbContext.Transactions
+      .Where(t => t.Metadata is PlaidTransactionMetadata && existingTransactionIds.Contains(((PlaidTransactionMetadata)t.Metadata).PlaidId))
+      .ToListAsync(ct)
+      .ConfigureAwait(false);
+
+    var addedCount = 0;
+
     foreach (var addedTransaction in added)
+    {
+      try
       {
         if (addedTransaction.Pending.GetValueOrDefault())
         {
@@ -35,6 +55,23 @@ internal sealed class PlaidAddedTransactionHandler : IPlaidAddedTransactionHandl
             addedTransaction.TransactionId,
             account.Id
           );
+          addedCount++;
+          continue;
+        }
+
+        var existingTransaction = existingTransactions.FirstOrDefault(
+          t => t.Metadata is PlaidTransactionMetadata metadata && metadata.PlaidId == addedTransaction.TransactionId
+        );
+
+        if (existingTransaction is not null)
+        {
+          _logger.LogInformation(
+            "Skipping added transaction {PlaidTransactionId} for account {AccountId} as it has already been added as transaction {TransactionId}",
+            addedTransaction.TransactionId,
+            account.Id,
+            existingTransaction.Id
+          );
+          addedCount++;
           continue;
         }
 
@@ -42,13 +79,28 @@ internal sealed class PlaidAddedTransactionHandler : IPlaidAddedTransactionHandl
         var transaction = Transaction.From(
           account.UserId,
           account.Id,
-        addedTransaction.MerchantName,
+          addedTransaction.Merchant,
           addedTransaction.Description,
           addedTransaction.Amount,
           addedTransaction.PostedDate,
           transactionMetadata
         );
         account.AddTransaction(transaction);
+        addedCount++;
       }
+      catch (Exception ex)
+      {
+        _logger.LogError(
+          ex,
+          "Failed to added transaction {PlaidTransactionId} to account {AccountId}",
+          account.Id,
+          addedTransaction.TransactionId
+        );
+      }
+    }
+
+    _logger.LogInformation("Added {AddedCount} transactions to account {AccountId}", addedCount, account.Id);
+
+    return addedCount;
   }
 }

src/FiscalOS.Infra/Transactions/Plaid/PlaidExtensions.cs

@@ -5,6 +5,7 @@ public static class PlaidExtensions
 #pragma warning disable CA1034
   extension(Transaction transaction)
   {
+    public string Merchant => transaction.MerchantName ?? "Unknown merchant";
     public string Description => transaction.OriginalDescription ?? "";
     public DateTimeOffset PostedDate => transaction.Datetime ?? (
         transaction.Date.HasValue

src/FiscalOS.Infra/Transactions/Plaid/PlaidModifiedTransactionHandler.cs

@@ -5,7 +5,11 @@ namespace FiscalOS.Infra.Transactions.Plaid;
 
 internal interface IPlaidModifiedTransactionHandler
 {
-  void Handle(Account account, IEnumerable<Going.Plaid.Entity.Transaction> modified);
+  Task<int> HandleAsync(
+    Account account,
+    IEnumerable<Going.Plaid.Entity.Transaction> modified,
+    CancellationToken ct
+  );
 }
 
 internal sealed class PlaidModifiedTransactionHandler : IPlaidModifiedTransactionHandler
@@ -30,14 +34,19 @@ internal sealed class PlaidModifiedTransactionHandler : IPlaidModifiedTransactio
     );
   }
 
-  public void Handle(Account account, IEnumerable<Going.Plaid.Entity.Transaction> modified)
+  public async Task<int> HandleAsync(Account account, IEnumerable<Going.Plaid.Entity.Transaction> modified, CancellationToken ct)
   {
     var modifiedTransactionIds = modified.Select(t => t.TransactionId);
-    var existingModifiedTransactions = _appDbContext.Transactions
+    var existingModifiedTransactions = await _appDbContext.Transactions
       .Where(t => t.Metadata is PlaidTransactionMetadata && modifiedTransactionIds.Contains(((PlaidTransactionMetadata)t.Metadata).PlaidId))
-      .ToList();
+      .ToListAsync(ct)
+      .ConfigureAwait(false);
+
+    var modifiedCount = 0;
 
     foreach (var existing in existingModifiedTransactions)
+    {
+      try
       {
         if (existing.Metadata is not PlaidTransactionMetadata plaidMetadata)
         {
@@ -45,6 +54,7 @@ internal sealed class PlaidModifiedTransactionHandler : IPlaidModifiedTransactio
             "Existing transaction {TransactionId} has non-Plaid metadata. Skipping update for this transaction.",
             existing.Id
           );
+          modifiedCount++;
           continue;
         }
 
@@ -56,13 +66,14 @@ internal sealed class PlaidModifiedTransactionHandler : IPlaidModifiedTransactio
             "No corresponding modified transaction found in Plaid response for existing transaction {TransactionId}. Skipping update for this transaction.",
             existing.Id
           );
+          modifiedCount++;
           continue;
         }
 
         var newTransactionData = Transaction.From(
           existing.UserId,
           existing.AccountId,
-        plaidModifiedTransaction.MerchantName,
+          plaidModifiedTransaction.Merchant,
           plaidModifiedTransaction.Description,
           plaidModifiedTransaction.Amount,
           plaidModifiedTransaction.PostedDate,
@@ -70,6 +81,21 @@ internal sealed class PlaidModifiedTransactionHandler : IPlaidModifiedTransactio
         );
 
         _appDbContext.Entry(existing).CurrentValues.SetValues(newTransactionData);
+        modifiedCount++;
       }
+      catch (Exception ex)
+      {
+        _logger.LogError(
+          ex,
+          "Failed to update transaction {TransactionId} for account {AccountId}",
+          account.Id,
+          existing.Id
+        );
+      }
+    }
+
+    _logger.LogInformation("Modified {ModifiedCount} transactions for account {AccountId}", modifiedCount, account.Id);
+
+    return modifiedCount;
   }
 }

src/FiscalOS.Infra/Transactions/Plaid/PlaidRemovedTransactionHandler.cs

@@ -2,32 +2,54 @@ namespace FiscalOS.Infra.Transactions.Plaid;
 
 internal interface IPlaidRemovedTransactionHandler
 {
-  Task HandleAsync(IEnumerable<RemovedTransaction> removed, CancellationToken cancellationToken);
+  Task<int> HandleAsync(IEnumerable<RemovedTransaction> removed, CancellationToken cancellationToken);
 }
 
 internal sealed class PlaidRemovedTransactionHandler : IPlaidRemovedTransactionHandler
 {
   private readonly AppDbContext _appDbContext;
+  private readonly ILogger<PlaidRemovedTransactionHandler> _logger;
 
-  private PlaidRemovedTransactionHandler(AppDbContext appDbContext)
+  private PlaidRemovedTransactionHandler(
+    AppDbContext appDbContext,
+    ILogger<PlaidRemovedTransactionHandler> logger
+  )
   {
     _appDbContext = appDbContext;
+    _logger = logger;
   }
 
   public static PlaidRemovedTransactionHandler From(IServiceProvider serviceProvider)
   {
     return new(
-      serviceProvider.GetRequiredService<AppDbContext>()
+      serviceProvider.GetRequiredService<AppDbContext>(),
+      serviceProvider.GetRequiredService<ILogger<PlaidRemovedTransactionHandler>>()
     );
   }
 
-  public async Task HandleAsync(IEnumerable<RemovedTransaction> removed, CancellationToken cancellationToken)
+  public async Task<int> HandleAsync(IEnumerable<RemovedTransaction> removed, CancellationToken cancellationToken)
   {
+    var removedCount = 0;
     var removedIdsList = removed.Select(t => t.TransactionId);
 
-    await _appDbContext.Transactions
+    try
+    {
+      removedCount = await _appDbContext.Transactions
         .Where(t => t.Metadata is PlaidTransactionMetadata && removedIdsList.Contains(((PlaidTransactionMetadata)t.Metadata).PlaidId))
         .ExecuteDeleteAsync(cancellationToken)
         .ConfigureAwait(false);
     }
+    catch (Exception ex)
+    {
+      _logger.LogError(
+        ex,
+        "Failed to removed transactions {TransactionIds}",
+        removedIdsList
+      );
+    }
+
+    _logger.LogInformation("Removed {RemovedCount} transactions", removedCount);
+
+    return removedCount;
+  }
 }

src/FiscalOS.Infra/Transactions/Plaid/PlaidTransactionProcessor.cs

@@ -4,7 +4,7 @@ namespace FiscalOS.Infra.Transactions.Plaid;
 
 internal interface IPlaidTransactionProcessor
 {
-  Task ProcessAsync(
+  Task<bool> ProcessAsync(
     Account account,
     TransactionsSyncResponse syncResponse,
     CancellationToken cancellationToken
@@ -37,10 +37,31 @@ internal sealed class PlaidTransactionProcessor : IPlaidTransactionProcessor
     );
   }
 
-  public async Task ProcessAsync(Account account, TransactionsSyncResponse syncResponse, CancellationToken cancellationToken)
+  public async Task<bool> ProcessAsync(
+    Account account,
+    TransactionsSyncResponse syncResponse,
+    CancellationToken cancellationToken
+  )
   {
-    _addedHandler.Handle(account, syncResponse.Added);
+    var numAdded = await _addedHandler.HandleAsync(
-    _modifiedHandler.Handle(account, syncResponse.Modified);
+      account,
-    await _removedHandler.HandleAsync(syncResponse.Removed, cancellationToken).ConfigureAwait(false);
+      syncResponse.Added,
+      cancellationToken
+    ).ConfigureAwait(false);
+
+    var numModified = await _modifiedHandler.HandleAsync(
+      account,
+      syncResponse.Modified,
+      cancellationToken
+    ).ConfigureAwait(false);
+
+    var numRemoved = await _removedHandler.HandleAsync(
+      syncResponse.Removed,
+      cancellationToken
+    ).ConfigureAwait(false);
+
+    return numAdded == syncResponse.Added.Count &&
+      numModified == syncResponse.Modified.Count &&
+      numRemoved == syncResponse.Removed.Count;
   }
 }

src/FiscalOS.Infra/Transactions/Plaid/PlaidTransactionSyncer.cs

@@ -83,11 +83,22 @@ internal sealed class PlaidTransactionSyncer : IPlaidTransactionSyncer
         break;
       }
 
-      plaidAccountMetadata.SetCursor(response.NextCursor);
       var balance = Balance.From(plaidAccount.Current, plaidAccount.Available, plaidAccount.CurrencyCode);
       account.AddBalance(balance);
 
-      await _transactionProcessor.ProcessAsync(account, response, cancellationToken).ConfigureAwait(false);
+      var isProcessedSuccessfully = await _transactionProcessor.ProcessAsync(account, response, cancellationToken).ConfigureAwait(false);
+
+      if (isProcessedSuccessfully is false)
+      {
+        _logger.LogWarning(
+          "Unable to process transactions for request {RequestId} for account {AccountId} successfully",
+          response.RequestId,
+          account.Id
+        );
+        break;
+      }
+
+      plaidAccountMetadata.SetCursor(response.NextCursor);
     }
   }
 }

src/FiscalOS.Infra/Transactions/Plaid/SyncUpdatesProcessor.cs

@@ -74,17 +74,24 @@ internal sealed class SyncUpdatesProcessor : IAsyncQueueProcessor<SyncUpdatesQue
     }
 
     var decryptedAccessToken = await _encryptor.DecryptAsyncFor(
-        institution.User!,
+        institution.User,
         plaidMetadata.EncryptedAccessToken,
         cancellationToken
       )
       .ConfigureAwait(false);
 
     foreach (var account in institution.Accounts)
+    {
+      try
       {
         await _plaidTransactionSyncer.SyncTransactionsForAccountAsync(account, decryptedAccessToken, cancellationToken)
           .ConfigureAwait(false);
       }
+      catch (Exception ex)
+      {
+        _logger.LogError(ex, "Failed to sync transactions for account {AccountId}", account.Id);
+      }
+    }
 
     await _appDbContext.SaveChangesAsync(cancellationToken).ConfigureAwait(false);