Stevan/fiscalos
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Stevan Freeborn
2026-03-03 05:09:03 -06:00
parent 19a4520a7b
commit 562a708c74
7 changed files with 212 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/FiscalOS.Infra/Transactions/Plaid/PlaidAddedTransactionHandler.cs
+75 -23
View File
@@ -5,50 +5,102 @@ 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)
{
if (addedTransaction.Pending.GetValueOrDefault())
try
{
_logger.LogInformation(
"Skipping pending transaction {TransactionId} for account {AccountId} as it has not been posted yet",
addedTransaction.TransactionId,
account.Id
);
continue;
}
if (addedTransaction.Pending.GetValueOrDefault())
{
_logger.LogInformation(
"Skipping pending transaction {TransactionId} for account {AccountId} as it has not been posted yet",
addedTransaction.TransactionId,
account.Id
);
addedCount++;
continue;
}
var transactionMetadata = PlaidTransactionMetadata.From(addedTransaction.TransactionId);
var transaction = Transaction.From(
account.UserId,
account.Id,
addedTransaction.MerchantName,
addedTransaction.Description,
addedTransaction.Amount,
addedTransaction.PostedDate,
transactionMetadata
);
account.AddTransaction(transaction);
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;
}
var transactionMetadata = PlaidTransactionMetadata.From(addedTransaction.TransactionId);
var transaction = Transaction.From(
account.UserId,
account.Id,
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
+1
View File
@@ -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
+57 -31
View File
@@ -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,46 +34,68 @@ 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)
{
if (existing.Metadata is not PlaidTransactionMetadata plaidMetadata)
try
{
_logger.LogWarning(
"Existing transaction {TransactionId} has non-Plaid metadata. Skipping update for this transaction.",
if (existing.Metadata is not PlaidTransactionMetadata plaidMetadata)
{
_logger.LogWarning(
"Existing transaction {TransactionId} has non-Plaid metadata. Skipping update for this transaction.",
existing.Id
);
modifiedCount++;
continue;
}
var plaidModifiedTransaction = modified.FirstOrDefault(t => t.TransactionId == plaidMetadata.PlaidId);
if (plaidModifiedTransaction is null)
{
_logger.LogWarning(
"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.Merchant,
plaidModifiedTransaction.Description,
plaidModifiedTransaction.Amount,
plaidModifiedTransaction.PostedDate,
plaidMetadata
);
_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
);
continue;
}
var plaidModifiedTransaction = modified.FirstOrDefault(t => t.TransactionId == plaidMetadata.PlaidId);
if (plaidModifiedTransaction is null)
{
_logger.LogWarning(
"No corresponding modified transaction found in Plaid response for existing transaction {TransactionId}. Skipping update for this transaction.",
existing.Id
);
continue;
}
var newTransactionData = Transaction.From(
existing.UserId,
existing.AccountId,
plaidModifiedTransaction.MerchantName,
plaidModifiedTransaction.Description,
plaidModifiedTransaction.Amount,
plaidModifiedTransaction.PostedDate,
plaidMetadata
);
_appDbContext.Entry(existing).CurrentValues.SetValues(newTransactionData);
}
_logger.LogInformation("Modified {ModifiedCount} transactions for account {AccountId}", modifiedCount, account.Id);
return modifiedCount;
}
}
src/FiscalOS.Infra/Transactions/Plaid/PlaidRemovedTransactionHandler.cs
+30 -8
View File
@@ -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
.Where(t => t.Metadata is PlaidTransactionMetadata && removedIdsList.Contains(((PlaidTransactionMetadata)t.Metadata).PlaidId))
.ExecuteDeleteAsync(cancellationToken)
.ConfigureAwait(false);
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
+26 -5
View File
@@ -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);
_modifiedHandler.Handle(account, syncResponse.Modified);
await _removedHandler.HandleAsync(syncResponse.Removed, cancellationToken).ConfigureAwait(false);
var numAdded = await _addedHandler.HandleAsync(
account,
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
+13 -2
View File
@@ -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
+10 -3
View File
@@ -74,7 +74,7 @@ internal sealed class SyncUpdatesProcessor : IAsyncQueueProcessor<SyncUpdatesQue
}
var decryptedAccessToken = await _encryptor.DecryptAsyncFor(
institution.User!,
institution.User,
plaidMetadata.EncryptedAccessToken,
cancellationToken
)
@@ -82,8 +82,15 @@ internal sealed class SyncUpdatesProcessor : IAsyncQueueProcessor<SyncUpdatesQue
foreach (var account in institution.Accounts)
{
await _plaidTransactionSyncer.SyncTransactionsForAccountAsync(account, decryptedAccessToken, cancellationToken)
.ConfigureAwait(false);
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);