362 lines
10 KiB
Go
362 lines
10 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"slices"
|
|
"strings"
|
|
"time"
|
|
|
|
"finance-duck/internal/banking"
|
|
"finance-duck/internal/classification"
|
|
"finance-duck/internal/domain"
|
|
)
|
|
|
|
type ImportResult struct {
|
|
Imported int `json:"imported"`
|
|
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
|
|
}
|
|
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
|
|
}
|
|
func (a *App) ImportCSV(ctx context.Context, rev, accountID string, r io.Reader) (ImportResult, error) {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
s, err := a.snapshot(ctx)
|
|
if err != nil {
|
|
return ImportResult{}, err
|
|
}
|
|
if rev != s.Revision {
|
|
return ImportResult{}, errors.New("revision conflict: reload before importing")
|
|
}
|
|
for _, account := range s.Data.Accounts {
|
|
if account.ID == accountID {
|
|
facts, e := banking.ParseCSV(r, account)
|
|
if e != nil {
|
|
return ImportResult{}, e
|
|
}
|
|
return a.importFacts(ctx, s, facts)
|
|
}
|
|
}
|
|
return ImportResult{}, errors.New("unknown account")
|
|
}
|
|
func (a *App) Authorize(ctx context.Context, institution, country string) (string, error) {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
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")
|
|
}
|
|
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, state)
|
|
if err == nil {
|
|
a.authStates[state] = authorization{time.Now().Add(15 * time.Minute), institution, country}
|
|
}
|
|
return url, err
|
|
}
|
|
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)
|
|
}
|
|
}
|
|
}
|
|
func (a *App) Callback(ctx context.Context, code, state string) 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 errors.New("authorization state expired or invalid; reconnect again")
|
|
}
|
|
if a.bank == nil || code == "" {
|
|
return errors.New("authorization did not provide a code")
|
|
}
|
|
session, err := a.bank.Exchange(ctx, code)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
a.ops.Sessions = append(a.ops.Sessions, session)
|
|
a.ops.Consents[session.ID] = Consent{Institution: auth.Institution, Country: auth.Country}
|
|
if err = a.saveOps(); err != nil {
|
|
return err
|
|
}
|
|
s, err := a.snapshot(ctx)
|
|
if err != nil {
|
|
return 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 err
|
|
}
|
|
_, err = a.commit(ctx, s.Revision, s.Data)
|
|
if err == nil {
|
|
select {
|
|
case a.syncRequested <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
return err
|
|
}
|
|
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")
|
|
}
|
|
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
|
|
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{}
|
|
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 {
|
|
meta.Error = e.Error()
|
|
meta.NeedsReconnect = errors.Is(e, banking.ErrReconnect)
|
|
a.ops.Consents[session.ID] = meta
|
|
failures = append(failures, meta.Institution+": "+meta.Error)
|
|
continue
|
|
}
|
|
meta.Error = ""
|
|
meta.NeedsReconnect = false
|
|
a.ops.Consents[session.ID] = meta
|
|
a.ops.Sessions[i].ValidUntil = current.ValidUntil
|
|
for _, account := range current.Accounts {
|
|
validAccounts[account.ExternalAccountID] = true
|
|
}
|
|
}
|
|
now := time.Now().UTC()
|
|
to := now.Format("2006-01-02")
|
|
for _, account := range s.Data.Accounts {
|
|
if !account.Active || account.ExternalAccountID == "" {
|
|
continue
|
|
}
|
|
if !validAccounts[account.ExternalAccountID] {
|
|
failures = append(failures, account.DisplayName+": bank connection unavailable")
|
|
continue
|
|
}
|
|
from := now.AddDate(0, 0, -90).Format("2006-01-02")
|
|
if last, e := time.Parse(time.RFC3339, a.ops.AccountSync[account.ID]); e == nil {
|
|
from = last.AddDate(0, 0, -14).Format("2006-01-02")
|
|
}
|
|
facts, e := a.bank.Transactions(ctx, account, from, to)
|
|
if e != nil {
|
|
meta := a.ops.Consents[accountSession[account.ID]]
|
|
meta.Error = "Transaction retrieval failed; retry synchronization"
|
|
a.ops.Consents[accountSession[account.ID]] = meta
|
|
failures = append(failures, account.DisplayName+": transaction retrieval failed")
|
|
continue
|
|
}
|
|
result, e := a.importFacts(ctx, s, facts)
|
|
if e != nil {
|
|
failures = append(failures, account.DisplayName+": "+e.Error())
|
|
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, "; ")
|
|
if len(failures) == 0 {
|
|
a.ops.LastSync = now.Format(time.RFC3339)
|
|
}
|
|
if err = a.saveOps(); err != nil {
|
|
return State{}, err
|
|
}
|
|
return a.snapshot(ctx)
|
|
}
|
|
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
|
|
last, err := time.Parse(time.RFC3339, a.ops.LastSync)
|
|
due := force || err != nil || time.Since(last) >= 24*time.Hour
|
|
a.mu.Unlock()
|
|
if configured && due {
|
|
a.Sync(ctx)
|
|
timer.Reset(24 * time.Hour)
|
|
} else {
|
|
timer.Reset(time.Minute)
|
|
}
|
|
}
|
|
}
|