Two of three banks were only pacing us, yet the dashboard demanded attention, printed four nested wrappers and a nanosecond UTC deadline, and the scheduler retried hourly into a refusal whose end time the bank had already given. A rate limit now carries its retry time as data: Status.SyncRetryAt is set when every failure is self-clearing, the connection reports rate_limited with that deadline, the dashboard says synchronization resumes by itself and renders the time in the browser's zone, and the scheduler sleeps until the deadline instead of spending hourly session checks. Sync now still tries immediately. The third bank's "transaction retrieval failed" hid its cause. Provider failures Finance Duck determines itself are typed as banking.ProviderError, so an unreachable provider, a timeout or an unusable response, such as a booked transaction without a booking date, is reported instead of the opaque fallback. Provider response text still never reaches the message.
867 lines
28 KiB
Go
867 lines
28 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"slices"
|
|
"strings"
|
|
"time"
|
|
|
|
"finance-duck/internal/banking"
|
|
"finance-duck/internal/classification"
|
|
"finance-duck/internal/domain"
|
|
"finance-duck/internal/ratelimit"
|
|
)
|
|
|
|
type ImportResult struct {
|
|
Imported int `json:"imported"`
|
|
// RequestedFrom and EarliestFetched describe manual history retrieval:
|
|
// the requested window start, and the oldest booking date the bank
|
|
// actually returned (empty when it returned nothing).
|
|
RequestedFrom string `json:"requested_from,omitempty"`
|
|
EarliestFetched string `json:"earliest_fetched,omitempty"`
|
|
State State `json:"state"`
|
|
}
|
|
|
|
func addProposal(d *domain.Dataset, p classification.Proposal) error {
|
|
if p.NewMerchant != nil {
|
|
m := *p.NewMerchant
|
|
if slices.ContainsFunc(d.Merchants, func(v domain.Merchant) bool { return v.ID == m.ID }) {
|
|
return errors.New("proposed merchant ID already exists")
|
|
}
|
|
d.Merchants = append(d.Merchants, m)
|
|
}
|
|
return nil
|
|
}
|
|
func (a *App) importFacts(ctx context.Context, s State, facts []domain.Facts) (ImportResult, error) {
|
|
added, err := banking.NormalizeAndDedupe(s.Data, facts)
|
|
if err != nil {
|
|
return ImportResult{}, err
|
|
}
|
|
if len(added) == 0 {
|
|
return ImportResult{State: s}, nil
|
|
}
|
|
s.Data.Transactions = append(s.Data.Transactions, added...)
|
|
banking.MatchTransfers(&s.Data)
|
|
// Commit imported facts before calling any model: remote failures cannot lose money records.
|
|
s, err = a.commit(ctx, s.Revision, s.Data)
|
|
if err != nil {
|
|
return ImportResult{}, err
|
|
}
|
|
ids := make(map[string]bool, len(added))
|
|
for _, t := range added {
|
|
ids[t.Facts.ID] = true
|
|
}
|
|
for i, t := range s.Data.Transactions {
|
|
if !ids[t.Facts.ID] || t.Enrichment.Kind == "transfer" {
|
|
continue
|
|
}
|
|
// With AI classification off for imports, no provider is contacted at
|
|
// all: deterministic merchant rules still apply.
|
|
p, e := classification.Rules(t.Facts, s.Data)
|
|
if a.settings.ClassifyOnImport {
|
|
p, e = a.classifier.Classify(ctx, t.Facts, s.Data, false)
|
|
}
|
|
if e == nil {
|
|
e = addProposal(&s.Data, p)
|
|
}
|
|
if e == nil {
|
|
e = domain.ValidateEnrichment(s.Data, t.Facts, p.Enrichment)
|
|
}
|
|
if e != nil {
|
|
s.Data.Transactions[i].Enrichment.Classification = domain.Provenance{Source: "unclassified", Timestamp: time.Now().UTC().Format(time.RFC3339), Error: e.Error()}
|
|
continue
|
|
}
|
|
s.Data.Transactions[i].Enrichment = p.Enrichment
|
|
}
|
|
state, err := a.commit(ctx, s.Revision, s.Data)
|
|
if err != nil {
|
|
return ImportResult{}, fmt.Errorf("facts imported; enrichment commit failed: %w", err)
|
|
}
|
|
return ImportResult{Imported: len(added), State: state}, nil
|
|
}
|
|
|
|
// CSVColumn is one reviewable source-column assignment.
|
|
type CSVColumn struct {
|
|
Field string `json:"field"`
|
|
Column string `json:"column"`
|
|
}
|
|
|
|
// CSVImport is a parsed statement awaiting confirmation. Nothing is written to
|
|
// the journal until ConfirmCSVImport applies the exact facts previewed here.
|
|
type CSVImport struct {
|
|
ID string `json:"id"`
|
|
Revision string `json:"revision"`
|
|
AccountID string `json:"account_id"`
|
|
Source string `json:"source"`
|
|
SourceLabel string `json:"source_label"`
|
|
MappedBy string `json:"mapped_by"`
|
|
Model string `json:"model,omitempty"`
|
|
Mapping banking.CSVMapping `json:"mapping"`
|
|
Columns []CSVColumn `json:"columns"`
|
|
Records int `json:"records"`
|
|
New int `json:"new"`
|
|
Duplicates int `json:"duplicates"`
|
|
Samples []domain.Facts `json:"samples"`
|
|
|
|
facts []domain.Facts
|
|
created time.Time
|
|
}
|
|
|
|
const csvImportLifetime = time.Hour
|
|
const maxPreparedCSVImports = 5
|
|
const maxCSVSamples = 10
|
|
|
|
// PrepareCSVImport maps and parses an uploaded statement without importing it.
|
|
// Known N26, ING and Kontist exports are recognized locally; any other layout
|
|
// needs a configured model to propose a column mapping from the statement's
|
|
// redacted shape. The result must be reviewed and confirmed.
|
|
func (a *App) PrepareCSVImport(ctx context.Context, rev, accountID string, r io.Reader) (CSVImport, error) {
|
|
a.mu.Lock()
|
|
s, err := a.snapshot(ctx)
|
|
model := strings.TrimSpace(a.settings.Model)
|
|
client := a.classifier.WithModel(model)
|
|
configured := strings.TrimSpace(a.classifier.APIKey) != "" && model != ""
|
|
a.mu.Unlock()
|
|
if err != nil {
|
|
return CSVImport{}, err
|
|
}
|
|
if rev != s.Revision {
|
|
return CSVImport{}, errors.New("revision conflict: reload before importing")
|
|
}
|
|
index := slices.IndexFunc(s.Data.Accounts, func(account domain.Account) bool { return account.ID == accountID })
|
|
if index < 0 {
|
|
return CSVImport{}, errors.New("unknown account")
|
|
}
|
|
account := s.Data.Accounts[index]
|
|
file, err := banking.ReadCSV(r)
|
|
if err != nil {
|
|
return CSVImport{}, err
|
|
}
|
|
prepared := CSVImport{ID: domain.NewID("csvimport"), Revision: s.Revision, AccountID: account.ID, MappedBy: "preset", created: time.Now()}
|
|
mapping, source, label, recognized := banking.DetectCSVMapping(file)
|
|
if !recognized {
|
|
sample, e := file.Sample()
|
|
if e != nil {
|
|
return CSVImport{}, e
|
|
}
|
|
if !configured {
|
|
return CSVImport{}, errors.New("unrecognized CSV layout: import an N26, ING or Kontist export, or configure an OpenRouter key and model in Settings to map these columns")
|
|
}
|
|
proposal, e := client.ProposeCSVMapping(ctx, classification.CSVMappingRequest{
|
|
Delimiter: sample.Delimiter, Headers: sample.Headers, ShapedRows: sample.ShapedRows,
|
|
DateFormats: banking.CSVDateFormats(), DecimalFormats: banking.CSVDecimalFormats(),
|
|
})
|
|
if e != nil {
|
|
return CSVImport{}, e
|
|
}
|
|
mapping = banking.CSVMapping{
|
|
HeaderRow: sample.HeaderRow,
|
|
BookingDateColumn: proposal.BookingDateColumn,
|
|
ValueDateColumn: proposal.ValueDateColumn,
|
|
AmountColumn: proposal.AmountColumn,
|
|
DebitColumn: proposal.DebitColumn,
|
|
CreditColumn: proposal.CreditColumn,
|
|
CurrencyColumn: proposal.CurrencyColumn,
|
|
DescriptionColumn: proposal.DescriptionColumn,
|
|
CounterpartyColumn: proposal.CounterpartyColumn,
|
|
CounterpartyIBANColumn: proposal.CounterpartyIBANColumn,
|
|
DateFormat: proposal.DateFormat,
|
|
DecimalFormat: proposal.DecimalFormat,
|
|
}
|
|
source, label = "csv", "AI-mapped CSV"
|
|
prepared.MappedBy, prepared.Model = "openrouter", proposal.Model
|
|
}
|
|
facts, err := banking.ParseMappedCSV(file, account, mapping, source)
|
|
if err != nil {
|
|
return CSVImport{}, err
|
|
}
|
|
// Dedupe now so the preview reports what confirming would actually add, and
|
|
// so cross-source conflicts are reported before anything is written.
|
|
added, err := banking.NormalizeAndDedupe(s.Data, facts)
|
|
if err != nil {
|
|
return CSVImport{}, err
|
|
}
|
|
prepared.Source, prepared.SourceLabel, prepared.Mapping = source, label, mapping
|
|
prepared.Columns = csvColumns(mapping, account)
|
|
prepared.Records, prepared.New, prepared.Duplicates = len(facts), len(added), len(facts)-len(added)
|
|
prepared.Samples, prepared.facts = csvSamples(facts), facts
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
for id, old := range a.csvImports {
|
|
if time.Since(old.created) > csvImportLifetime {
|
|
delete(a.csvImports, id)
|
|
}
|
|
}
|
|
if len(a.csvImports) >= maxPreparedCSVImports {
|
|
return CSVImport{}, errors.New("too many statements awaiting confirmation; confirm or cancel one first")
|
|
}
|
|
a.csvImports[prepared.ID] = prepared
|
|
return prepared, nil
|
|
}
|
|
|
|
// ConfirmCSVImport imports exactly the facts that were previewed, provided the
|
|
// journal has not changed since.
|
|
func (a *App) ConfirmCSVImport(ctx context.Context, id, rev string) (ImportResult, error) {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
prepared, ok := a.csvImports[id]
|
|
if !ok || time.Since(prepared.created) > csvImportLifetime {
|
|
return ImportResult{}, errors.New("prepared import expired or unknown; upload the statement again")
|
|
}
|
|
if rev != prepared.Revision {
|
|
return ImportResult{}, errors.New("revision conflict: reload before importing")
|
|
}
|
|
s, err := a.snapshot(ctx)
|
|
if err != nil {
|
|
return ImportResult{}, err
|
|
}
|
|
if s.Revision != prepared.Revision {
|
|
return ImportResult{}, errors.New("revision conflict: data changed after the preview; upload the statement again")
|
|
}
|
|
result, err := a.importFacts(ctx, s, prepared.facts)
|
|
if err != nil {
|
|
return ImportResult{}, err
|
|
}
|
|
delete(a.csvImports, id)
|
|
return result, nil
|
|
}
|
|
|
|
// CancelCSVImport discards a prepared statement without importing anything.
|
|
func (a *App) CancelCSVImport(id string) {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
delete(a.csvImports, id)
|
|
}
|
|
|
|
// csvColumns lists the mapping as reviewable field/value pairs, including where
|
|
// the currency comes from and how dates and decimals are read: an inferred
|
|
// convention is the easiest part of a mapping to get wrong.
|
|
func csvColumns(mapping banking.CSVMapping, account domain.Account) []CSVColumn {
|
|
columns := make([]CSVColumn, 0, 13)
|
|
for _, field := range []CSVColumn{
|
|
{"Booking date", mapping.BookingDateColumn}, {"Value date", mapping.ValueDateColumn},
|
|
{"Amount", mapping.AmountColumn}, {"Debit", mapping.DebitColumn}, {"Credit", mapping.CreditColumn},
|
|
{"Currency", mapping.CurrencyColumn}, {"Description", mapping.DescriptionColumn},
|
|
{"Secondary description", mapping.FallbackDescriptionColumn}, {"Counterparty", mapping.CounterpartyColumn},
|
|
{"Counterparty IBAN", mapping.CounterpartyIBANColumn}, {"Transaction reference", mapping.ExternalIDColumn},
|
|
} {
|
|
if field.Column != "" {
|
|
columns = append(columns, field)
|
|
}
|
|
}
|
|
if mapping.CurrencyColumn == "" {
|
|
currency, origin := mapping.FixedCurrency, "from the amount column header"
|
|
if currency == "" {
|
|
currency, origin = account.Currency, "from the selected account"
|
|
}
|
|
columns = append(columns, CSVColumn{"Currency", currency + " (" + origin + ")"})
|
|
}
|
|
return append(columns,
|
|
CSVColumn{"Dates read as", csvDateFormatLabel(mapping.DateFormat)},
|
|
CSVColumn{"Decimal separator", csvDecimalFormatLabel(mapping.DecimalFormat)},
|
|
)
|
|
}
|
|
|
|
func csvDateFormatLabel(format string) string {
|
|
switch format {
|
|
case "yyyy-mm-dd":
|
|
return "2026-09-01"
|
|
case "dd.mm.yyyy":
|
|
return "01.09.2026 (day first)"
|
|
case "mm/dd/yyyy":
|
|
return "09/01/2026 (month first)"
|
|
case "dd/mm/yyyy":
|
|
return "01/09/2026 (day first)"
|
|
case "iso-date-time":
|
|
return "2026-09-01T14:30:00 (date and time)"
|
|
case "iso-or-german":
|
|
return "2026-09-01 or 01.09.2026"
|
|
default:
|
|
return format
|
|
}
|
|
}
|
|
|
|
func csvDecimalFormatLabel(format string) string {
|
|
switch format {
|
|
case "dot":
|
|
return "point (1234.56)"
|
|
case "comma":
|
|
return "comma (1.234,56)"
|
|
case "dot-or-comma":
|
|
return "point or comma"
|
|
default:
|
|
return format
|
|
}
|
|
}
|
|
|
|
// csvSamples keeps a bounded, ordered excerpt that always shows the extremes and
|
|
// both directions of money when the statement contains them: an inverted sign or
|
|
// a misread date convention has to be visible before confirming.
|
|
func csvSamples(facts []domain.Facts) []domain.Facts {
|
|
if len(facts) == 0 {
|
|
return []domain.Facts{}
|
|
}
|
|
chosen := map[int]bool{0: true, len(facts) - 1: true}
|
|
if len(facts) > 1 {
|
|
chosen[1] = true
|
|
}
|
|
if len(facts) > 2 {
|
|
chosen[len(facts)-2] = true
|
|
}
|
|
credit, debit, largest := -1, -1, 0
|
|
for i, f := range facts {
|
|
minor, err := f.Amount.Minor()
|
|
if err != nil {
|
|
continue
|
|
}
|
|
if minor >= 0 && credit < 0 {
|
|
credit = i
|
|
}
|
|
if minor < 0 && debit < 0 {
|
|
debit = i
|
|
}
|
|
if previous, e := facts[largest].Amount.Minor(); e != nil || abs64(minor) > abs64(previous) {
|
|
largest = i
|
|
}
|
|
}
|
|
for _, index := range []int{credit, debit, largest} {
|
|
if index >= 0 && len(chosen) < maxCSVSamples {
|
|
chosen[index] = true
|
|
}
|
|
}
|
|
indexes := make([]int, 0, len(chosen))
|
|
for index := range chosen {
|
|
indexes = append(indexes, index)
|
|
}
|
|
slices.Sort(indexes)
|
|
samples := make([]domain.Facts, 0, len(indexes))
|
|
for _, index := range indexes {
|
|
samples = append(samples, facts[index])
|
|
}
|
|
return samples
|
|
}
|
|
|
|
func abs64(v int64) int64 {
|
|
if v < 0 {
|
|
return -v
|
|
}
|
|
return v
|
|
}
|
|
|
|
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 {
|
|
return ImportResult{}, bankFailure(err, "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.Contains(current.AccountIDs, account.ExternalAccountID) {
|
|
return ImportResult{}, banking.ErrReconnect
|
|
}
|
|
now := time.Now().UTC()
|
|
from := now.AddDate(0, -historyMonths, 0).Format("2006-01-02")
|
|
// The longest fetching strategy imports whatever period the bank still
|
|
// permits: many banks cap history on an established consent instead of
|
|
// serving the full requested range.
|
|
facts, err := a.bank.Transactions(ctx, account, from, now.Format("2006-01-02"), true)
|
|
if err != nil {
|
|
return ImportResult{}, bankFailure(err, "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.
|
|
result, err := a.importFacts(ctx, s, facts)
|
|
if err != nil {
|
|
return ImportResult{}, err
|
|
}
|
|
result.RequestedFrom = from
|
|
for _, f := range facts {
|
|
if result.EarliestFetched == "" || f.BookingDate < result.EarliestFetched {
|
|
result.EarliestFetched = f.BookingDate
|
|
}
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
// Authorize starts a consent for one account-holder kind. An empty psuType
|
|
// keeps the previous personal default for existing API callers.
|
|
func (a *App) Authorize(ctx context.Context, institution, country, psuType string, historyMonths int) (string, error) {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
if historyMonths < 1 || historyMonths > 120 {
|
|
return "", errors.New("history_months must be an integer between 1 and 120")
|
|
}
|
|
if a.bank == nil {
|
|
return "", errors.New("Enable Banking is not configured")
|
|
}
|
|
institution = strings.TrimSpace(institution)
|
|
country = strings.ToUpper(strings.TrimSpace(country))
|
|
if institution == "" {
|
|
return "", errors.New("institution is required")
|
|
}
|
|
if len(country) != 2 {
|
|
return "", errors.New("country must be a two-letter code")
|
|
}
|
|
if psuType == "" {
|
|
psuType = banking.PSUPersonal
|
|
}
|
|
if !banking.ValidPSUType(psuType) {
|
|
return "", errors.New("account type must be personal or business")
|
|
}
|
|
for state, auth := range a.authStates {
|
|
if time.Now().After(auth.Expires) {
|
|
delete(a.authStates, state)
|
|
}
|
|
}
|
|
state := domain.NewID("auth")
|
|
url, err := a.bank.Authorize(ctx, institution, country, psuType, state)
|
|
if err != nil {
|
|
return "", bankFailure(err, "bank authorization unavailable; retry connecting")
|
|
}
|
|
a.authStates[state] = authorization{Expires: time.Now().Add(15 * time.Minute), Institution: institution, Country: country, PSUType: psuType, HistoryMonths: historyMonths}
|
|
return url, nil
|
|
}
|
|
|
|
// Institutions lists connectable banks for the country so the UI can offer
|
|
// a picker instead of free-text entry. Provider failures stay sanitized.
|
|
func (a *App) Institutions(ctx context.Context, country string) ([]banking.Institution, error) {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
if a.bank == nil {
|
|
return nil, errors.New("Enable Banking is not configured")
|
|
}
|
|
list, err := a.bank.Institutions(ctx, country)
|
|
if err != nil {
|
|
return nil, bankFailure(err, "institution list unavailable; retry")
|
|
}
|
|
return list, nil
|
|
}
|
|
func normalizedIBAN(s string) string { return strings.ToUpper(strings.Join(strings.Fields(s), "")) }
|
|
func connectAccounts(d *domain.Dataset, session *banking.Session, reconnect bool) {
|
|
for i, account := range session.Accounts {
|
|
found := -1
|
|
for j, local := range d.Accounts {
|
|
if local.ID == account.ID || (account.ExternalAccountID != "" && local.ExternalAccountID == account.ExternalAccountID) || (account.IBAN != "" && normalizedIBAN(account.IBAN) == normalizedIBAN(local.IBAN)) {
|
|
found = j
|
|
break
|
|
}
|
|
}
|
|
if found >= 0 {
|
|
local := d.Accounts[found]
|
|
local.ExternalAccountID = account.ExternalAccountID
|
|
if account.IBAN != "" {
|
|
local.IBAN = account.IBAN
|
|
}
|
|
if reconnect {
|
|
local.Active = true
|
|
}
|
|
session.Accounts[i] = local
|
|
d.Accounts[found] = local
|
|
} else {
|
|
if account.ID == "" {
|
|
account.ID = domain.NewID("acct")
|
|
}
|
|
account.Active = true
|
|
session.Accounts[i] = account
|
|
d.Accounts = append(d.Accounts, account)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Callback completes a bank authorization and returns how many accounts the
|
|
// bank shared that could not be linked to a journal account.
|
|
func (a *App) Callback(ctx context.Context, code, state string) (int, error) {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
auth, ok := a.authStates[state]
|
|
delete(a.authStates, state)
|
|
if !ok || time.Now().After(auth.Expires) {
|
|
return 0, errors.New("authorization state expired or invalid; reconnect again")
|
|
}
|
|
if a.bank == nil || code == "" {
|
|
return 0, errors.New("authorization did not provide a code")
|
|
}
|
|
session, err := a.bank.Exchange(ctx, code)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
// A consent without linkable accounts can never sync and its stored
|
|
// session would be reaped silently. Fail visibly instead.
|
|
if len(session.Accounts) == 0 {
|
|
if session.Unlinkable > 0 {
|
|
return 0, fmt.Errorf("the bank shared %d account(s), but none could be linked: they lack an IBAN or stable identification, or use an unsupported currency", session.Unlinkable)
|
|
}
|
|
return 0, errors.New("the bank authorized the connection but shared no accounts, so nothing was linked; accounts of another type (for example business) may need a separate consent")
|
|
}
|
|
a.ops.Sessions = append(a.ops.Sessions, session)
|
|
a.ops.Consents[session.ID] = Consent{Institution: auth.Institution, Country: auth.Country, PSUType: auth.PSUType, HistoryMonths: auth.HistoryMonths}
|
|
if err = a.saveOps(); err != nil {
|
|
return 0, err
|
|
}
|
|
s, err := a.snapshot(ctx)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
connectAccounts(&s.Data, &session, true)
|
|
// Remove superseded account bindings, not unrelated bank consents.
|
|
replacements := map[string]bool{}
|
|
for _, account := range session.Accounts {
|
|
replacements[account.ID] = true
|
|
}
|
|
sessions := make([]banking.Session, 0, len(a.ops.Sessions)+1)
|
|
for _, old := range a.ops.Sessions {
|
|
if old.ID == session.ID {
|
|
continue
|
|
}
|
|
old.Accounts = slices.DeleteFunc(slices.Clone(old.Accounts), func(account domain.Account) bool { return replacements[account.ID] })
|
|
if len(old.Accounts) > 0 {
|
|
sessions = append(sessions, old)
|
|
} else {
|
|
delete(a.ops.Consents, old.ID)
|
|
}
|
|
}
|
|
a.ops.Sessions = append(sessions, session)
|
|
// Save once-only provider details before the canonical commit. Sync can recover
|
|
// the account bindings if a crash or external edit interrupts that commit.
|
|
if err = a.saveOps(); err != nil {
|
|
return 0, err
|
|
}
|
|
if _, err = a.commit(ctx, s.Revision, s.Data); err != nil {
|
|
return 0, err
|
|
}
|
|
select {
|
|
case a.syncRequested <- struct{}{}:
|
|
default:
|
|
}
|
|
return session.Unlinkable, nil
|
|
}
|
|
func (a *App) Balances(ctx context.Context, id string) ([]banking.Balance, error) {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
if a.bank == nil {
|
|
return nil, errors.New("Enable Banking is not configured")
|
|
}
|
|
s, err := a.snapshot(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, account := range s.Data.Accounts {
|
|
if account.ID == id && account.ExternalAccountID != "" {
|
|
for _, session := range a.ops.Sessions {
|
|
for _, linked := range session.Accounts {
|
|
if linked.ID == account.ID && linked.ExternalAccountID == account.ExternalAccountID {
|
|
return a.bank.Balances(ctx, linked.ExternalAccountID)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return nil, errors.New("account is not connected")
|
|
}
|
|
|
|
// Only typed, locally generated errors are safe to expose; provider errors may
|
|
// wrap private response data even when their underlying cause is recognizable.
|
|
// The fallback is a last resort: an unnamed cause leaves an operator with
|
|
// nothing to act on.
|
|
func bankFailure(err error, fallback string) error {
|
|
var background *banking.BackgroundQuotaError
|
|
if errors.As(err, &background) {
|
|
return background
|
|
}
|
|
var limited *ratelimit.RateLimitError
|
|
if errors.As(err, &limited) {
|
|
return fmt.Errorf("Enable Banking: %w", limited)
|
|
}
|
|
if errors.Is(err, banking.ErrReconnect) {
|
|
return banking.ErrReconnect
|
|
}
|
|
var api *banking.APIError
|
|
if errors.As(err, &api) {
|
|
return api
|
|
}
|
|
var consent *banking.ConsentError
|
|
if errors.As(err, &consent) {
|
|
return consent
|
|
}
|
|
var provider *banking.ProviderError
|
|
if errors.As(err, &provider) {
|
|
return provider
|
|
}
|
|
return errors.New(fallback)
|
|
}
|
|
|
|
// syncRetryAt reports the bank's own retry time for a failure that clears
|
|
// itself. A rate limit without a usable deadline is not treated as waiting:
|
|
// nothing would ever announce that it had expired.
|
|
func syncRetryAt(err error) (time.Time, bool) {
|
|
var limited *ratelimit.RateLimitError
|
|
if !errors.As(err, &limited) {
|
|
return time.Time{}, false
|
|
}
|
|
at := limited.RetryAt()
|
|
return at, !at.IsZero()
|
|
}
|
|
|
|
func (a *App) Sync(ctx context.Context) (State, error) {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
if a.bank == nil {
|
|
return State{}, errors.New("Enable Banking is not configured")
|
|
}
|
|
s, err := a.snapshot(ctx)
|
|
if err != nil {
|
|
return State{}, err
|
|
}
|
|
var failures []string
|
|
// waitUntil is the earliest time the banks themselves allow a retry, used
|
|
// only while every failure is such a self-clearing rate limit.
|
|
var waitUntil time.Time
|
|
waiting := true
|
|
for i := range a.ops.Sessions {
|
|
connectAccounts(&s.Data, &a.ops.Sessions[i], false)
|
|
}
|
|
// Recovery may have both the old consent and its once-only replacement.
|
|
// Keep the newest binding for each local account before checking bank status.
|
|
claimed := map[string]bool{}
|
|
retained := make([]banking.Session, 0, len(a.ops.Sessions))
|
|
for i := len(a.ops.Sessions) - 1; i >= 0; i-- {
|
|
session := a.ops.Sessions[i]
|
|
session.Accounts = slices.DeleteFunc(slices.Clone(session.Accounts), func(account domain.Account) bool {
|
|
if claimed[account.ID] {
|
|
return true
|
|
}
|
|
claimed[account.ID] = true
|
|
return false
|
|
})
|
|
if len(session.Accounts) == 0 {
|
|
delete(a.ops.Consents, session.ID)
|
|
} else {
|
|
retained = append(retained, session)
|
|
}
|
|
}
|
|
slices.Reverse(retained)
|
|
a.ops.Sessions = retained
|
|
s, err = a.commit(ctx, s.Revision, s.Data)
|
|
if err != nil {
|
|
return State{}, err
|
|
}
|
|
validAccounts := map[string]bool{}
|
|
accountSession := map[string]string{}
|
|
failedSessions := map[string]bool{}
|
|
for i, session := range a.ops.Sessions {
|
|
for _, account := range session.Accounts {
|
|
accountSession[account.ID] = session.ID
|
|
}
|
|
meta := a.ops.Consents[session.ID]
|
|
current, e := a.bank.Status(ctx, session.ID)
|
|
if e == nil {
|
|
expiry, parseErr := time.Parse(time.RFC3339, current.ValidUntil)
|
|
if parseErr != nil || !expiry.After(time.Now()) {
|
|
e = banking.ErrReconnect
|
|
}
|
|
}
|
|
if e != nil {
|
|
meta.Error = bankFailure(e, "bank connection unavailable; retry synchronization").Error()
|
|
meta.RetryAt = ""
|
|
if at, ok := syncRetryAt(e); ok {
|
|
meta.RetryAt = at.UTC().Format(time.RFC3339)
|
|
if waitUntil.IsZero() || at.Before(waitUntil) {
|
|
waitUntil = at
|
|
}
|
|
} else {
|
|
waiting = false
|
|
}
|
|
meta.NeedsReconnect = errors.Is(e, banking.ErrReconnect)
|
|
a.ops.Consents[session.ID] = meta
|
|
failures = append(failures, meta.Institution+": "+meta.Error)
|
|
failedSessions[session.ID] = true
|
|
continue
|
|
}
|
|
meta.Error = ""
|
|
meta.RetryAt = ""
|
|
meta.NeedsReconnect = false
|
|
a.ops.Consents[session.ID] = meta
|
|
a.ops.Sessions[i].ValidUntil = current.ValidUntil
|
|
for _, account := range session.Accounts {
|
|
if account.ExternalAccountID != "" && slices.Contains(current.AccountIDs, account.ExternalAccountID) {
|
|
validAccounts[account.ID] = true
|
|
}
|
|
}
|
|
}
|
|
now := time.Now().UTC()
|
|
to := now.Format("2006-01-02")
|
|
for _, account := range s.Data.Accounts {
|
|
if !account.Active || account.ExternalAccountID == "" {
|
|
continue
|
|
}
|
|
sessionID := accountSession[account.ID]
|
|
if failedSessions[sessionID] {
|
|
continue
|
|
}
|
|
if !validAccounts[account.ID] {
|
|
failures = append(failures, account.DisplayName+": bank connection unavailable")
|
|
waiting = false
|
|
if sessionID != "" {
|
|
meta := a.ops.Consents[sessionID]
|
|
meta.Error = banking.ErrReconnect.Error()
|
|
meta.RetryAt = ""
|
|
meta.NeedsReconnect = true
|
|
a.ops.Consents[sessionID] = meta
|
|
}
|
|
continue
|
|
}
|
|
var from string
|
|
if last, e := time.Parse(time.RFC3339, a.ops.AccountSync[account.ID]); e == nil {
|
|
from = last.AddDate(0, 0, -14).Format("2006-01-02")
|
|
} else {
|
|
months := a.ops.Consents[accountSession[account.ID]].historyMonths()
|
|
from = now.AddDate(0, -months, 0).Format("2006-01-02")
|
|
}
|
|
facts, e := a.bank.Transactions(ctx, account, from, to, false)
|
|
if e != nil {
|
|
meta := a.ops.Consents[sessionID]
|
|
meta.Error = bankFailure(e, "transaction retrieval failed; retry synchronization").Error()
|
|
meta.RetryAt = ""
|
|
if at, ok := syncRetryAt(e); ok {
|
|
meta.RetryAt = at.UTC().Format(time.RFC3339)
|
|
if waitUntil.IsZero() || at.Before(waitUntil) {
|
|
waitUntil = at
|
|
}
|
|
} else {
|
|
waiting = false
|
|
}
|
|
meta.NeedsReconnect = meta.NeedsReconnect || errors.Is(e, banking.ErrReconnect)
|
|
a.ops.Consents[sessionID] = meta
|
|
failures = append(failures, account.DisplayName+": "+meta.Error)
|
|
continue
|
|
}
|
|
result, e := a.importFacts(ctx, s, facts)
|
|
if e != nil {
|
|
failures = append(failures, account.DisplayName+": "+e.Error())
|
|
waiting = false
|
|
s, err = a.snapshot(ctx)
|
|
if err != nil {
|
|
return State{}, err
|
|
}
|
|
continue
|
|
}
|
|
s = result.State
|
|
a.ops.AccountSync[account.ID] = now.Format(time.RFC3339)
|
|
}
|
|
a.ops.SyncError = strings.Join(failures, "; ")
|
|
a.ops.SyncRetryAt = ""
|
|
if len(failures) == 0 {
|
|
a.ops.LastSync = now.Format(time.RFC3339)
|
|
} else if waiting && !waitUntil.IsZero() {
|
|
// Every bank named its own retry time: this is a wait, not a fault.
|
|
a.ops.SyncRetryAt = waitUntil.UTC().Format(time.RFC3339)
|
|
}
|
|
if err = a.saveOps(); err != nil {
|
|
return State{}, err
|
|
}
|
|
return a.snapshot(ctx)
|
|
}
|
|
|
|
// syncSchedule decides whether an automatic sync may run now, and how long to
|
|
// wait otherwise. While a bank has named its own retry time, waiting is the
|
|
// only useful action: retrying earlier spends session-status calls on a refusal
|
|
// that is already known. A manual request always proceeds.
|
|
func syncSchedule(now time.Time, ops operational, force bool) (time.Duration, bool) {
|
|
if retry, err := time.Parse(time.RFC3339, ops.SyncRetryAt); err == nil && !force && now.Before(retry) {
|
|
return min(retry.Sub(now)+time.Minute, time.Hour), false
|
|
}
|
|
last, err := time.Parse(time.RFC3339, ops.LastSync)
|
|
if force || err != nil || ops.SyncError != "" || now.Sub(last) >= 24*time.Hour {
|
|
return 0, true
|
|
}
|
|
return time.Minute, false
|
|
}
|
|
|
|
// syncBackoff spaces the next attempt after a sync. A failure with a known bank
|
|
// retry time waits for it; other failures retry hourly so transient provider
|
|
// problems clear without waiting a day, while bounding unattended traffic.
|
|
func syncBackoff(now time.Time, ops operational) time.Duration {
|
|
if ops.SyncError == "" {
|
|
return 24 * time.Hour
|
|
}
|
|
if retry, err := time.Parse(time.RFC3339, ops.SyncRetryAt); err == nil && now.Before(retry) {
|
|
return min(retry.Sub(now)+time.Minute, 24*time.Hour)
|
|
}
|
|
return time.Hour
|
|
}
|
|
|
|
func (a *App) RunScheduler(ctx context.Context) {
|
|
timer := time.NewTimer(time.Minute)
|
|
defer timer.Stop()
|
|
for {
|
|
force := false
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-a.syncRequested:
|
|
force = true
|
|
case <-timer.C:
|
|
}
|
|
a.mu.Lock()
|
|
configured := a.bank != nil
|
|
wait, due := syncSchedule(time.Now(), a.ops, force)
|
|
a.mu.Unlock()
|
|
if !configured || !due {
|
|
if wait <= 0 {
|
|
wait = time.Minute
|
|
}
|
|
timer.Reset(wait)
|
|
continue
|
|
}
|
|
a.Sync(ctx)
|
|
a.mu.Lock()
|
|
wait = syncBackoff(time.Now(), a.ops)
|
|
a.mu.Unlock()
|
|
timer.Reset(wait)
|
|
}
|
|
}
|