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, facts ...domain.Facts) error { if p.NewMerchant != nil { m := *p.NewMerchant if len(facts) > 0 { if alias := strings.Join(strings.Fields(facts[0].Counterparty), " "); alias != "" && !slices.Contains(m.Aliases, alias) { m.Aliases = append(m.Aliases, alias) } } if slices.ContainsFunc(d.Merchants, func(v domain.Merchant) bool { return v.ID == m.ID }) { // A batch resolves several rows against one snapshot: an earlier // row already registered this same proposal. return nil } d.Merchants = append(d.Merchants, m) } return nil } func (a *App) importFacts(ctx context.Context, s State, facts []domain.Facts, instruments []domain.Instrument) (ImportResult, error) { // Instruments first: a broker fact references one, and the canonical // dataset is validated as a whole, so a trade cannot be committed before // the security it trades exists. known := make(map[string]bool, len(s.Data.Instruments)) for _, v := range s.Data.Instruments { known[v.ID] = true } registered := false for _, v := range instruments { if !known[v.ID] { known[v.ID], registered = true, true s.Data.Instruments = append(s.Data.Instruments, v) } } added, err := banking.NormalizeAndDedupe(s.Data, facts) if err != nil { return ImportResult{}, err } if len(added) == 0 && !registered { 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" || t.Enrichment.Kind == domain.KindInvestment { 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) // A low-confidence category is never auto-applied on import: the // merchant link and provenance stay, and Analyse shows the model's // suggestion for review instead. if e == nil && p.Enrichment.Classification.Confidence == "low" { p.Enrichment.CategoryID = domain.Fallback(t.Facts).CategoryID } } if e == nil { e = addProposal(&s.Data, p, t.Facts) if e == nil && p.Enrichment.MerchantID != "" { classification.LearnAlias(&s.Data, t.Facts, p.Enrichment.MerchantID) } } 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"` // Broker is present when the statement is a broker export. Its rows carry // positions as well as cash, so they are read by a dedicated parser rather // than by a column mapping, and the review needs to show what that parser // decided: which securities it would register, which rows it skipped, and // which figures it deliberately did not apply. Broker *banking.BrokerImport `json:"broker,omitempty"` facts []domain.Facts instruments []domain.Instrument 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()} // A broker export is recognized before any column mapping is attempted. Its // rows are not interchangeable statement lines: the same amount column is // cash on one row, a gross to be netted on another, and a position // valuation that must not touch cash on a third, so a column mapping cannot // describe it. if source, label, header, broker := banking.DetectBrokerCSV(file); broker { read, e := banking.ParseBrokerCSV(file, account, s.Data.Instruments) if e != nil { return CSVImport{}, e } added, e := banking.NormalizeAndDedupe(s.Data, read.Facts) if e != nil { return CSVImport{}, e } prepared.Source, prepared.SourceLabel = source, label prepared.Mapping = banking.CSVMapping{HeaderRow: header} prepared.Columns = brokerColumns(source) prepared.Records, prepared.New, prepared.Duplicates = len(read.Facts), len(added), len(read.Facts)-len(added) prepared.Samples, prepared.facts, prepared.instruments = csvSamples(read.Facts), read.Facts, read.Instruments prepared.Broker = &read return a.retain(prepared) } 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 return a.retain(prepared) } // retain holds a reviewed statement until it is confirmed or expires. Nothing // is written to the journal here. func (a *App) retain(prepared CSVImport) (CSVImport, error) { 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 } // brokerColumns describes what the broker parser decided, in the same // reviewable shape as a column mapping. The dispatch is the part that can be // wrong in a way that moves money, so it is the part shown. func brokerColumns(source string) []CSVColumn { if source == banking.SourceTradeRepublic { return []CSVColumn{ {Field: "Booking date", Column: "date, exactly as printed; the datetime column is UTC and disagrees with it late in the evening"}, {Field: "Cash movement", Column: "amount − fee − tax, where the export writes fee and tax as the signed adjustments it made and the amount is the gross"}, {Field: "Position change", Column: "shares, already signed; a dividend's shares are the holding it was paid on and move nothing"}, {Field: "Instrument", Column: "symbol when it is an ISIN, else the one ISIN the description names; crypto carries a ticker in the column"}, {Field: "Counterparty", Column: "counterparty_iban, else the IBAN the description names in parentheses, else this account's settlement IBAN"}, {Field: "Reference", Column: "transaction_id"}, {Field: "Decimals", Column: "plain decimal point; trailing zeros are padding, not precision"}, } } return []CSVColumn{ {Field: "Booking date", Column: "date, exactly as printed; the time column is local and crosses midnight, so it is ignored"}, {Field: "Imported rows", Column: `status "Executed" only; cancelled retries are all zeros and would import as phantom trades`}, {Field: "Cash movement", Column: "cash rows: amount, already net of tax; trades: amount − fee − tax; corporate actions and depot transfers: none"}, {Field: "Position change", Column: "shares, signed by type for buys and sells and exactly as printed for corporate actions and depot transfers"}, {Field: "Instrument", Column: "isin; the description only names it"}, {Field: "Reference", Column: "reference, which the broker reuses across every leg of one event"}, {Field: "Decimals", Column: "German: comma decimal, and a dot only groups thousands in exact three-digit runs"}, } } // 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, prepared.instruments) 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, nil) 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, nil) 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) } // syncInterval is how often connected accounts synchronize on their own. Twice // a day halves how long a booking can sit unseen while staying inside Enable // Banking's documented background allowance of roughly four fetches per day per // account, which a failing sync's hourly retries also draw from. const syncInterval = 12 * time.Hour // 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) >= syncInterval { 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 for the next scheduled run, while bounding // unattended traffic. func syncBackoff(now time.Time, ops operational) time.Duration { if ops.SyncError == "" { return syncInterval } if retry, err := time.Parse(time.RFC3339, ops.SyncRetryAt); err == nil && now.Before(retry) { return min(retry.Sub(now)+time.Minute, syncInterval) } return time.Hour } func (a *App) RunScheduler(ctx context.Context) { timer := time.NewTimer(time.Minute) defer timer.Stop() // Prices keep their own clock: they come from a different provider, they are // wanted even when no bank is connected, and a sync backoff must not delay // them. The first run is shortly after start, so a fresh install or a // restart does not leave a day's holdings unvalued waiting for the tick; // after that it is daily, which is as often as a close changes. prices := time.NewTimer(quoteStartup) defer prices.Stop() for { force := false select { case <-ctx.Done(): return case <-a.syncRequested: force = true case <-prices.C: a.RefreshQuotes(ctx) prices.Reset(quoteInterval) continue 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) } }