Add per-account historical bank imports without resetting sync cursors
This commit is contained in:
Lars Nolden
@@ -0,0 +1,249 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"finance-duck/internal/banking"
|
||||
"finance-duck/internal/domain"
|
||||
)
|
||||
|
||||
type backfillBank struct {
|
||||
historyBank
|
||||
statusIDs []string
|
||||
accounts []domain.Account
|
||||
toDates []string
|
||||
statusErr error
|
||||
fetchErr error
|
||||
}
|
||||
|
||||
func (b *backfillBank) Status(ctx context.Context, id string) (banking.Session, error) {
|
||||
b.statusIDs = append(b.statusIDs, id)
|
||||
if b.statusErr != nil {
|
||||
return banking.Session{}, b.statusErr
|
||||
}
|
||||
return b.historyBank.Status(ctx, id)
|
||||
}
|
||||
|
||||
func (b *backfillBank) Transactions(ctx context.Context, account domain.Account, from, to string) ([]domain.Facts, error) {
|
||||
b.accounts = append(b.accounts, account)
|
||||
b.toDates = append(b.toDates, to)
|
||||
rows, err := b.historyBank.Transactions(ctx, account, from, to)
|
||||
if b.fetchErr != nil {
|
||||
// A provider can fail after accumulating a page: none of it is importable.
|
||||
return rows, b.fetchErr
|
||||
}
|
||||
return rows, err
|
||||
}
|
||||
|
||||
func backfillApp(t *testing.T) (*App, State, *backfillBank) {
|
||||
t.Helper()
|
||||
a, s := testApp(t)
|
||||
s, err := a.Mutate(context.Background(), s.Revision, func(d *domain.Dataset) error {
|
||||
d.Accounts[0].ExternalAccountID = "selected_uid"
|
||||
d.Accounts = append(d.Accounts, domain.Account{ID: "other", DisplayName: "Other", Currency: "EUR", Active: true, ExternalAccountID: "other_uid"})
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
session := banking.Session{ID: "current", ValidUntil: time.Now().Add(24 * time.Hour).Format(time.RFC3339), Accounts: s.Data.Accounts}
|
||||
b := &backfillBank{historyBank: historyBank{bankScenario: bankScenario{session: session}}}
|
||||
a.bank = b
|
||||
a.ops.Sessions = []banking.Session{session}
|
||||
a.ops.Consents[session.ID] = Consent{Institution: "Bank", Country: "DE", HistoryMonths: 3}
|
||||
a.ops.LastSync = time.Now().UTC().Add(-time.Hour).Format(time.RFC3339)
|
||||
for _, account := range s.Data.Accounts {
|
||||
a.ops.AccountSync[account.ID] = a.ops.LastSync
|
||||
}
|
||||
if err := a.saveOps(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return a, s, b
|
||||
}
|
||||
|
||||
func TestBackfillImportsOlderFactsOnceWithoutChangingSyncState(t *testing.T) {
|
||||
a, _, b := backfillApp(t)
|
||||
ctx := context.Background()
|
||||
s, err := a.Sync(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(s.Data.Transactions) != 2 {
|
||||
t.Fatal("recent sync did not seed both accounts")
|
||||
}
|
||||
s, err = a.Mutate(ctx, s.Revision, func(d *domain.Dataset) error {
|
||||
d.Accounts[0].Active = false // Inactive disables scheduling, not explicit backfill.
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
old := a.ops.Sessions[0]
|
||||
old.ID = "superseded"
|
||||
a.ops.Sessions = append([]banking.Session{old}, a.ops.Sessions...)
|
||||
a.ops.Consents["superseded"] = Consent{HistoryMonths: 120}
|
||||
meta := a.ops.Consents["current"]
|
||||
meta.Error = "Previous temporary failure"
|
||||
a.ops.Consents["current"] = meta
|
||||
a.ops.SyncError = "Previous temporary failure"
|
||||
if err := a.saveOps(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
beforeOps, err := json.Marshal(a.ops)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.statusIDs, b.accounts, b.fromDates, b.toDates = nil, nil, nil, nil
|
||||
// Provider-local IDs need not match our canonical account ID.
|
||||
providerAccount := s.Data.Accounts[0]
|
||||
providerAccount.ID = "provider_generated_id"
|
||||
b.session.Accounts = []domain.Account{providerAccount}
|
||||
fromBefore := time.Now().UTC().AddDate(0, -24, 0).Format("2006-01-02")
|
||||
toBefore := time.Now().UTC().Format("2006-01-02")
|
||||
result, err := a.Backfill(ctx, s.Revision, s.Data.Accounts[0].ID, 24)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fromAfter := time.Now().UTC().AddDate(0, -24, 0).Format("2006-01-02")
|
||||
toAfter := time.Now().UTC().Format("2006-01-02")
|
||||
if result.Imported != 2 || len(result.State.Data.Transactions) != 4 {
|
||||
t.Fatalf("older history import count is wrong: %+v", result)
|
||||
}
|
||||
if !reflect.DeepEqual(b.statusIDs, []string{"current"}) || !reflect.DeepEqual(b.accounts, []domain.Account{s.Data.Accounts[0]}) {
|
||||
t.Fatal("backfill did not use only the newest matching consent and canonical account")
|
||||
}
|
||||
if len(b.fromDates) != 1 || (b.fromDates[0] != fromBefore && b.fromDates[0] != fromAfter) || (b.toDates[0] != toBefore && b.toDates[0] != toAfter) {
|
||||
t.Fatalf("backfill did not request the selected calendar-month range: %v to %v", b.fromDates, b.toDates)
|
||||
}
|
||||
for _, existing := range s.Data.Transactions {
|
||||
found := false
|
||||
for _, tx := range result.State.Data.Transactions {
|
||||
if tx.Facts.ID == existing.Facts.ID {
|
||||
found = reflect.DeepEqual(tx, existing)
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatal("backfill changed an existing transaction")
|
||||
}
|
||||
}
|
||||
counts := map[string]int{}
|
||||
for _, tx := range result.State.Data.Transactions {
|
||||
counts[tx.Facts.AccountID]++
|
||||
}
|
||||
if counts[s.Data.Accounts[0].ID] != 3 || counts["other"] != 1 || !reflect.DeepEqual(result.State.Data.Accounts, s.Data.Accounts) {
|
||||
t.Fatal("backfill changed unrelated accounts or imported their history")
|
||||
}
|
||||
again, err := a.Backfill(ctx, result.State.Revision, s.Data.Accounts[0].ID, 24)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if again.Imported != 0 || !reflect.DeepEqual(again.State.Data, result.State.Data) {
|
||||
t.Fatal("repeated historical range was not idempotent")
|
||||
}
|
||||
afterOps, err := json.Marshal(a.ops)
|
||||
if err != nil || string(afterOps) != string(beforeOps) {
|
||||
t.Fatal("backfill changed operational state")
|
||||
}
|
||||
a = reopenBankingApp(t, a)
|
||||
afterOps, err = json.Marshal(a.ops)
|
||||
if err != nil || string(afterOps) != string(beforeOps) {
|
||||
t.Fatal("backfill changed persisted cursors, consent history or bindings")
|
||||
}
|
||||
persisted, err := a.Snapshot(ctx)
|
||||
if err != nil || !reflect.DeepEqual(persisted.Data, result.State.Data) {
|
||||
t.Fatal("historical facts did not survive restart")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackfillRejectsUnsafePrerequisitesBeforeProviderContact(t *testing.T) {
|
||||
for _, scenario := range []string{"stale revision", "zero months", "too many months", "missing account", "local account", "unbound account", "mismatched provider ID", "mismatched canonical ID", "expired consent", "reconnect required"} {
|
||||
t.Run(scenario, func(t *testing.T) {
|
||||
a, s, b := backfillApp(t)
|
||||
rev, id, months := s.Revision, s.Data.Accounts[0].ID, 12
|
||||
switch scenario {
|
||||
case "stale revision":
|
||||
rev = "stale"
|
||||
case "zero months":
|
||||
months = 0
|
||||
case "too many months":
|
||||
months = 121
|
||||
case "missing account":
|
||||
id = "missing"
|
||||
case "local account":
|
||||
var err error
|
||||
s, err = a.Mutate(context.Background(), rev, func(d *domain.Dataset) error { d.Accounts[0].ExternalAccountID = ""; return nil })
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rev = s.Revision
|
||||
case "unbound account":
|
||||
a.ops.Sessions = nil
|
||||
case "mismatched provider ID":
|
||||
a.ops.Sessions[0].Accounts[0].ExternalAccountID = "wrong_uid"
|
||||
case "mismatched canonical ID":
|
||||
a.ops.Sessions[0].Accounts[0].ID = "wrong_id"
|
||||
case "expired consent":
|
||||
a.ops.Sessions[0].ValidUntil = time.Now().Add(-time.Hour).Format(time.RFC3339)
|
||||
case "reconnect required":
|
||||
a.ops.Consents["current"] = Consent{NeedsReconnect: true}
|
||||
}
|
||||
if _, err := a.Backfill(context.Background(), rev, id, months); err == nil {
|
||||
t.Fatal("unsafe backfill was accepted")
|
||||
}
|
||||
if len(b.statusIDs) != 0 || len(b.accounts) != 0 {
|
||||
t.Fatal("unsafe prerequisites contacted the provider")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackfillProviderFailuresDoNotImportPartialData(t *testing.T) {
|
||||
for _, scenario := range []string{"not configured", "unavailable", "revoked", "provider expired", "account absent", "partial retrieval"} {
|
||||
t.Run(scenario, func(t *testing.T) {
|
||||
a, s, b := backfillApp(t)
|
||||
s = seed(t, a, s)
|
||||
switch scenario {
|
||||
case "not configured":
|
||||
a.bank = nil
|
||||
case "unavailable":
|
||||
b.statusErr = errors.New("private provider response")
|
||||
case "revoked":
|
||||
b.statusErr = banking.ErrReconnect
|
||||
case "provider expired":
|
||||
b.session.ValidUntil = time.Now().Add(-time.Hour).Format(time.RFC3339)
|
||||
case "account absent":
|
||||
b.session.Accounts = []domain.Account{{ID: s.Data.Accounts[0].ID, ExternalAccountID: "wrong_uid"}}
|
||||
case "partial retrieval":
|
||||
b.fetchErr = errors.New("private provider response")
|
||||
}
|
||||
beforeOps, err := json.Marshal(a.ops)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
result, err := a.Backfill(context.Background(), s.Revision, s.Data.Accounts[0].ID, 24)
|
||||
if err == nil || result.Imported != 0 {
|
||||
t.Fatal("failed provider request reported a successful import")
|
||||
}
|
||||
if strings.Contains(err.Error(), "private provider response") {
|
||||
t.Fatal("provider error exposed private response data")
|
||||
}
|
||||
if scenario != "partial retrieval" && len(b.accounts) != 0 {
|
||||
t.Fatal("unavailable consent reached transaction retrieval")
|
||||
}
|
||||
current, err := a.Snapshot(context.Background())
|
||||
if err != nil || current.Revision != s.Revision || !reflect.DeepEqual(current.Data, s.Data) {
|
||||
t.Fatal("provider failure changed canonical data")
|
||||
}
|
||||
afterOps, err := json.Marshal(a.ops)
|
||||
if err != nil || string(afterOps) != string(beforeOps) {
|
||||
t.Fatal("provider failure changed operational state")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -92,6 +92,76 @@ func (a *App) ImportCSV(ctx context.Context, rev, accountID string, r io.Reader)
|
||||
}
|
||||
return ImportResult{}, errors.New("unknown account")
|
||||
}
|
||||
|
||||
func (a *App) Backfill(ctx context.Context, rev, accountID string, historyMonths int) (ImportResult, error) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
if historyMonths < 1 || historyMonths > 120 {
|
||||
return ImportResult{}, errors.New("history_months must be an integer between 1 and 120")
|
||||
}
|
||||
s, err := a.snapshot(ctx)
|
||||
if err != nil {
|
||||
return ImportResult{}, err
|
||||
}
|
||||
if rev != s.Revision {
|
||||
return ImportResult{}, errors.New("revision conflict: reload before importing")
|
||||
}
|
||||
if a.bank == nil {
|
||||
return ImportResult{}, errors.New("Enable Banking is not configured")
|
||||
}
|
||||
index := slices.IndexFunc(s.Data.Accounts, func(account domain.Account) bool { return account.ID == accountID })
|
||||
if index < 0 {
|
||||
return ImportResult{}, errors.New("unknown account")
|
||||
}
|
||||
account := s.Data.Accounts[index]
|
||||
if account.ExternalAccountID == "" {
|
||||
return ImportResult{}, errors.New("account is not connected")
|
||||
}
|
||||
var session *banking.Session
|
||||
for i := len(a.ops.Sessions) - 1; i >= 0; i-- {
|
||||
saved := &a.ops.Sessions[i]
|
||||
if slices.ContainsFunc(saved.Accounts, func(linked domain.Account) bool {
|
||||
return linked.ID == account.ID && linked.ExternalAccountID == account.ExternalAccountID
|
||||
}) {
|
||||
session = saved
|
||||
break
|
||||
}
|
||||
}
|
||||
if session == nil || session.ID == "" {
|
||||
return ImportResult{}, errors.New("account is not connected")
|
||||
}
|
||||
expiry, err := time.Parse(time.RFC3339, session.ValidUntil)
|
||||
if a.ops.Consents[session.ID].NeedsReconnect || err != nil || !expiry.After(time.Now()) {
|
||||
return ImportResult{}, banking.ErrReconnect
|
||||
}
|
||||
current, err := a.bank.Status(ctx, session.ID)
|
||||
if err != nil {
|
||||
if errors.Is(err, banking.ErrReconnect) {
|
||||
return ImportResult{}, banking.ErrReconnect
|
||||
}
|
||||
return ImportResult{}, errors.New("bank connection unavailable; retry importing history")
|
||||
}
|
||||
expiry, err = time.Parse(time.RFC3339, current.ValidUntil)
|
||||
if err != nil || !expiry.After(time.Now()) {
|
||||
return ImportResult{}, banking.ErrReconnect
|
||||
}
|
||||
if !slices.ContainsFunc(current.Accounts, func(linked domain.Account) bool {
|
||||
return linked.ExternalAccountID == account.ExternalAccountID
|
||||
}) {
|
||||
return ImportResult{}, banking.ErrReconnect
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
facts, err := a.bank.Transactions(ctx, account, now.AddDate(0, -historyMonths, 0).Format("2006-01-02"), now.Format("2006-01-02"))
|
||||
if err != nil {
|
||||
if errors.Is(err, banking.ErrReconnect) {
|
||||
return ImportResult{}, banking.ErrReconnect
|
||||
}
|
||||
return ImportResult{}, errors.New("transaction retrieval failed; retry importing history")
|
||||
}
|
||||
// Use normal import processing without changing sync cursors or saved consent
|
||||
// settings, including when the requested range adds no transactions.
|
||||
return a.importFacts(ctx, s, facts)
|
||||
}
|
||||
func (a *App) Authorize(ctx context.Context, institution, country string, historyMonths int) (string, error) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
|
||||
@@ -41,6 +41,7 @@ func New(a *app.App, assets fs.FS, publicURL string) (http.Handler, error) {
|
||||
s.mux.HandleFunc("POST /api/transactions/{id}", s.transaction)
|
||||
s.mux.HandleFunc("POST /api/manage", s.manage)
|
||||
s.mux.HandleFunc("POST /api/import", s.importCSV)
|
||||
s.mux.HandleFunc("POST /api/backfill", s.backfill)
|
||||
s.mux.HandleFunc("POST /api/rebuild", func(w http.ResponseWriter, r *http.Request) { v, e := a.Rebuild(r.Context()); respond(w, v, e) })
|
||||
s.mux.HandleFunc("POST /api/sync", func(w http.ResponseWriter, r *http.Request) { v, e := a.Sync(r.Context()); respond(w, v, e) })
|
||||
s.mux.HandleFunc("POST /api/settings", s.settings)
|
||||
@@ -281,6 +282,18 @@ func (s *Server) importCSV(w http.ResponseWriter, r *http.Request) {
|
||||
v, e := s.app.ImportCSV(r.Context(), r.FormValue("revision"), r.FormValue("account_id"), f)
|
||||
respond(w, v, e)
|
||||
}
|
||||
func (s *Server) backfill(w http.ResponseWriter, r *http.Request) {
|
||||
var b struct {
|
||||
Revision string `json:"revision"`
|
||||
AccountID string `json:"account_id"`
|
||||
HistoryMonths int `json:"history_months"`
|
||||
}
|
||||
if !decode(w, r, &b) {
|
||||
return
|
||||
}
|
||||
v, e := s.app.Backfill(r.Context(), b.Revision, b.AccountID, b.HistoryMonths)
|
||||
respond(w, v, e)
|
||||
}
|
||||
func (s *Server) settings(w http.ResponseWriter, r *http.Request) {
|
||||
var b app.Settings
|
||||
if !decode(w, r, &b) {
|
||||
|
||||
Reference in New Issue
Block a user