finance-duck/internal/app/sync_test.go

package app

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"reflect"
	"strings"
	"testing"
	"time"

	"finance-duck/internal/banking"
	"finance-duck/internal/domain"
	"finance-duck/internal/ratelimit"
)

type bankScenario struct {
	session banking.Session
	fail    bool
}

func (b *bankScenario) Authorize(context.Context, string, string, string, string) (string, error) {
	return "https://bank.example/authorize", nil
}
func (b *bankScenario) Institutions(context.Context, string) ([]banking.Institution, error) {
	return []banking.Institution{{Name: "N26", Country: "DE"}}, nil
}
func (b *bankScenario) Exchange(context.Context, string) (banking.Session, error) {
	return b.session, nil
}
func (b *bankScenario) Status(context.Context, string) (banking.SessionStatus, error) {
	if b.fail {
		return banking.SessionStatus{}, errors.New("expired")
	}
	status := banking.SessionStatus{ValidUntil: b.session.ValidUntil}
	for _, account := range b.session.Accounts {
		status.AccountIDs = append(status.AccountIDs, account.ExternalAccountID)
	}
	return status, nil
}
func (b *bankScenario) Balances(context.Context, string) ([]banking.Balance, error) {
	return []banking.Balance{{Amount: "100.00", Currency: "EUR", Type: "CLBD"}}, nil
}
func (b *bankScenario) Transactions(_ context.Context, a domain.Account, from, to string, _ bool) ([]domain.Facts, error) {
	if b.fail {
		return nil, errors.New("offline")
	}
	return []domain.Facts{{Source: "enablebanking", AccountID: a.ID, BookingDate: time.Now().UTC().AddDate(0, 0, -1).Format("2006-01-02"), Amount: "-42.80", Currency: "EUR", RawDescription: "REWE", ExternalID: "entry_stable"}}, nil
}
func TestSyncRestoresSavedConsentBindingsAndDoesNotDuplicateFacts(t *testing.T) {
	a, s := testApp(t)
	account := s.Data.Accounts[0]
	account.ExternalAccountID = "provider_new"
	account.IBAN = "DE89370400440532013000"
	provider := &bankScenario{session: banking.Session{ID: "new_session", ValidUntil: time.Now().Add(24 * time.Hour).Format(time.RFC3339), Accounts: []domain.Account{account}}}
	a.bank = provider
	a.ops.Sessions = []banking.Session{provider.session}
	if err := a.saveOps(); err != nil {
		t.Fatal(err)
	}
	first, err := a.Sync(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if len(first.Data.Accounts) != 1 || first.Data.Accounts[0].ExternalAccountID != "provider_new" || len(first.Data.Transactions) != 1 {
		t.Fatalf("saved session did not recover/import: %+v", first.Data)
	}
	again, err := a.Sync(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if !reflect.DeepEqual(first.Data, again.Data) {
		t.Fatal("repeated bank synchronization changed canonical financial data")
	}
	provider.fail = true
	failed, err := a.Sync(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if failed.Status.SyncError == "" || !reflect.DeepEqual(again.Data, failed.Data) {
		t.Fatal("provider failure was not isolated from canonical data")
	}
}
func TestReconnectReplacesOldConsentWithoutDuplicatingLocalAccount(t *testing.T) {
	a, s := testApp(t)
	account := s.Data.Accounts[0]
	account.ExternalAccountID = "old_uid"
	account.IBAN = "DE89370400440532013000"
	s, err := a.Mutate(context.Background(), s.Revision, func(d *domain.Dataset) error { return SaveAccount(d, account) })
	if err != nil {
		t.Fatal(err)
	}
	a.ops.Sessions = []banking.Session{{ID: "old_session", Accounts: []domain.Account{account}}}
	a.ops.Consents["old_session"] = Consent{Institution: "N26", Country: "DE", HistoryMonths: 24}
	cursor := time.Now().UTC().Add(-24 * time.Hour).Format(time.RFC3339)
	a.ops.AccountSync[account.ID] = cursor
	renewed := account
	renewed.ID = "provider_local_id"
	renewed.ExternalAccountID = "new_uid"
	renewed.DisplayName = "Bank-generated name"
	a.bank = &bankScenario{session: banking.Session{ID: "new_session", ValidUntil: time.Now().Add(24 * time.Hour).Format(time.RFC3339), Accounts: []domain.Account{renewed}}}
	a.authStates["one_time_state"] = authorization{Expires: time.Now().Add(time.Minute), Institution: "N26", Country: "DE", HistoryMonths: 24}
	if _, err = a.Callback(context.Background(), "bank_code", "one_time_state"); err != nil {
		t.Fatal(err)
	}
	after, err := a.Snapshot(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if len(after.Data.Accounts) != 1 || after.Data.Accounts[0].ID != account.ID || after.Data.Accounts[0].DisplayName != account.DisplayName || after.Data.Accounts[0].ExternalAccountID != "new_uid" {
		t.Fatal("reconnect duplicated account or lost local display name")
	}
	if len(after.Sessions) != 1 || after.Sessions[0].ID != "new_session" {
		t.Fatal("expired session remains active after reconnect")
	}
	if a.ops.AccountSync[account.ID] != cursor || after.Connections[0].HistoryMonths != 24 {
		t.Fatal("reconnect reset the account cursor or lost the history choice")
	}
	if _, err = a.Callback(context.Background(), "bank_code", "one_time_state"); err == nil {
		t.Fatal("authorization state replay was accepted")
	}
}

// Exercise the controller's typed error without contacting a provider or waiting.
func bankRateError(t *testing.T) error {
	t.Helper()
	var controller ratelimit.Controller
	ctx := context.Background()
	if err := controller.Acquire(ctx); err != nil {
		t.Fatal(err)
	}
	defer controller.Release()
	_, err := controller.Do(ctx, func(context.Context) (*http.Response, error) {
		return &http.Response{
			StatusCode: http.StatusTooManyRequests,
			Header:     http.Header{"Retry-After": []string{"300"}},
			Body:       io.NopCloser(strings.NewReader("private provider response")),
		}, nil
	}, false)
	if err == nil {
		t.Fatal("rate limit response was accepted")
	}
	return fmt.Errorf("private provider response: %w", err)
}

type sessionBank struct {
	bankScenario
	statuses map[string]banking.SessionStatus
	failures map[string]error
	fetched  []string
}

func (b *sessionBank) Status(_ context.Context, id string) (banking.SessionStatus, error) {
	return b.statuses[id], b.failures[id]
}

func (b *sessionBank) Transactions(ctx context.Context, account domain.Account, from, to string, longest bool) ([]domain.Facts, error) {
	b.fetched = append(b.fetched, account.ID)
	return b.bankScenario.Transactions(ctx, account, from, to, longest)
}

func TestSyncSessionRateLimitPreservesBindingsAndRecovers(t *testing.T) {
	a, s := testApp(t)
	ctx := context.Background()
	s, err := a.Mutate(ctx, s.Revision, func(d *domain.Dataset) error {
		d.Accounts[0].ExternalAccountID = "main_uid"
		for _, id := range []string{"mwst", "tax", "independent"} {
			d.Accounts = append(d.Accounts, domain.Account{ID: id, DisplayName: id, Currency: "EUR", Active: true, ExternalAccountID: id + "_uid"})
		}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	expiry := time.Now().Add(24 * time.Hour).Format(time.RFC3339)
	limited := banking.Session{ID: "limited", ValidUntil: expiry}
	healthy := banking.Session{ID: "healthy", ValidUntil: expiry}
	for _, account := range s.Data.Accounts {
		if account.ID == "independent" {
			healthy.Accounts = append(healthy.Accounts, account)
		} else {
			limited.Accounts = append(limited.Accounts, account)
		}
	}
	a.ops.Sessions = []banking.Session{limited, healthy}
	a.ops.Consents["limited"] = Consent{Institution: "N26", Country: "DE", HistoryMonths: 24}
	a.ops.Consents["healthy"] = Consent{Institution: "Other", Country: "DE", HistoryMonths: 12}
	b := &sessionBank{
		statuses: map[string]banking.SessionStatus{
			"limited": {ValidUntil: expiry, AccountIDs: []string{"main_uid", "mwst_uid", "tax_uid"}},
			// Another consent must not authorize an account whose own status failed.
			"healthy": {ValidUntil: expiry, AccountIDs: []string{"independent_uid", "main_uid"}},
		},
		failures: map[string]error{},
	}
	a.bank = b
	before, err := a.Sync(ctx)
	if err != nil || len(before.Data.Transactions) != 4 {
		t.Fatalf("initial sync: transactions=%d, error=%v, sync error=%s", len(before.Data.Transactions), err, before.Status.SyncError)
	}
	old := time.Now().Add(-48 * time.Hour).UTC().Format(time.RFC3339)
	a.ops.LastSync = old
	for _, account := range before.Data.Accounts {
		a.ops.AccountSync[account.ID] = old
	}
	b.fetched = nil
	b.failures["limited"] = bankRateError(t)
	failed, err := a.Sync(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if strings.Count(failed.Status.SyncError, "429") != 1 || strings.Contains(failed.Status.SyncError, "bank connection unavailable") || strings.Contains(failed.Status.SyncError, "private provider response") || !strings.Contains(failed.Status.SyncError, "retry") {
		t.Fatal("session rate limit was duplicated, obscured, or exposed private data")
	}
	if !reflect.DeepEqual(b.fetched, []string{"independent"}) {
		t.Fatal("failed consent authorized retrieval or independent consent stopped syncing")
	}
	if !reflect.DeepEqual(before.Data, failed.Data) || !reflect.DeepEqual(before.Sessions, failed.Sessions) || a.ops.LastSync != old {
		t.Fatal("rate limit changed existing facts, bindings, metadata, or last successful sync")
	}
	for _, account := range before.Data.Accounts {
		if account.ID != "independent" && a.ops.AccountSync[account.ID] != old {
			t.Fatal("failed account advanced its cursor")
		}
	}
	if a.ops.AccountSync["independent"] == old || a.ops.Consents["limited"].NeedsReconnect || a.ops.Consents["limited"].HistoryMonths != 24 {
		t.Fatal("rate limit lost consent settings, required reconnect, or stopped the healthy cursor")
	}
	a = reopenBankingApp(t, a)
	a.bank = b
	if a.ops.Consents["limited"].NeedsReconnect || a.ops.LastSync != old || a.ops.AccountSync["mwst"] != old {
		t.Fatal("rate failure state did not survive restart safely")
	}
	delete(b.failures, "limited")
	b.fetched = nil
	recovered, err := a.Sync(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if recovered.Status.SyncError != "" || a.ops.Consents["limited"].Error != "" || a.ops.Consents["limited"].NeedsReconnect || a.ops.LastSync == old {
		t.Fatal("successful retry did not clear the transient failure")
	}
	if !reflect.DeepEqual(before.Data, recovered.Data) || len(b.fetched) != 4 {
		t.Fatal("recovery duplicated facts or skipped an account")
	}
	for _, account := range recovered.Data.Accounts {
		if a.ops.AccountSync[account.ID] == old {
			t.Fatal("recovered account cursor did not advance")
		}
	}
}

func TestSyncMissingMembershipStillRejectsAccount(t *testing.T) {
	a, s, b := backfillApp(t)
	b.session.Accounts = []domain.Account{s.Data.Accounts[1]}
	before := domain.Clone(s.Data)
	last := a.ops.LastSync
	after, err := a.Sync(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if !strings.Contains(after.Status.SyncError, s.Data.Accounts[0].DisplayName+": bank connection unavailable") || !a.ops.Consents["current"].NeedsReconnect {
		t.Fatal("missing account membership was treated as authorized")
	}
	if len(b.accounts) != 1 || b.accounts[0].ID != "other" || a.ops.AccountSync[s.Data.Accounts[0].ID] != last || a.ops.LastSync != last {
		t.Fatal("missing member was fetched or advanced its cursor, or valid member was skipped")
	}
	if !reflect.DeepEqual(before.Accounts, after.Data.Accounts) || len(after.Data.Transactions) != 1 || after.Data.Transactions[0].Facts.AccountID != "other" {
		t.Fatal("missing membership changed bindings or imported unauthorized facts")
	}
}

func TestSyncTransactionFailuresPreserveProgressAndSafeErrors(t *testing.T) {
	for _, scenario := range []string{"rate limit", "background rate limit", "reconnect", "private response", "unusable response"} {
		t.Run(scenario, func(t *testing.T) {
			a, s, b := backfillApp(t)
			s = seed(t, a, s)
			old := a.ops.LastSync
			cursors := map[string]string{}
			for id, cursor := range a.ops.AccountSync {
				cursors[id] = cursor
			}
			switch scenario {
			case "rate limit":
				b.fetchErr = bankRateError(t)
			case "background rate limit":
				quota := &banking.BackgroundQuotaError{RateLimitError: ratelimit.NewError(time.Now().Add(6 * time.Hour))}
				b.fetchErr = fmt.Errorf("private provider response: %w", quota)
			case "reconnect":
				b.fetchErr = fmt.Errorf("private provider response: %w", banking.ErrReconnect)
			case "private response":
				b.fetchErr = errors.New("private provider response")
			case "unusable response":
				unusable := &banking.ProviderError{Detail: "a booked transaction has no valid booking date"}
				b.fetchErr = fmt.Errorf("private provider response: %w", unusable)
			}
			failed, err := a.Sync(context.Background())
			if err != nil {
				t.Fatal(err)
			}
			meta := a.ops.Consents["current"]
			if failed.Status.SyncError == "" || meta.Error == "" || strings.Contains(failed.Status.SyncError+meta.Error, "private provider response") {
				t.Fatal("transaction failure was lost or exposed provider data")
			}
			if strings.Contains(scenario, "rate limit") && (!strings.Contains(failed.Status.SyncError, "429") || !strings.Contains(meta.Error, "429") || !strings.Contains(meta.Error, "retry")) {
				t.Fatal("transaction rate error was obscured")
			}
			if scenario == "background rate limit" && !strings.Contains(meta.Error, "background") {
				t.Fatal("daily bank quota was confused with a short request throttle")
			}
			if meta.NeedsReconnect != (scenario == "reconnect") {
				t.Fatal("transaction failure classified consent incorrectly")
			}
			if !reflect.DeepEqual(s.Data, failed.Data) || !reflect.DeepEqual(cursors, a.ops.AccountSync) || a.ops.LastSync != old {
				t.Fatal("failed retrieval imported partial data or advanced synchronization")
			}
			rateLimited := strings.Contains(scenario, "rate limit")
			// A bank that named its own retry time is a wait, not a fault: the
			// UI and the scheduler both rely on this distinction.
			if (failed.Status.SyncRetryAt != "") != rateLimited {
				t.Fatalf("waiting state is wrong for %s: retry at %q", scenario, failed.Status.SyncRetryAt)
			}
			for _, connection := range failed.Connections {
				if connection.Status == "local" {
					continue
				}
				want := "error"
				switch {
				case rateLimited:
					want = "rate_limited"
				case scenario == "reconnect":
					want = "reconnect_required"
				}
				if connection.Status != want || (connection.RetryAt != "") != rateLimited {
					t.Fatalf("connection reported %q with retry %q, want %q", connection.Status, connection.RetryAt, want)
				}
			}
			if rateLimited {
				at, e := time.Parse(time.RFC3339, failed.Status.SyncRetryAt)
				if e != nil || !at.After(time.Now()) {
					t.Fatalf("unusable retry deadline %q: %v", failed.Status.SyncRetryAt, e)
				}
				// Whole seconds, once: operators read this message.
				if !strings.Contains(meta.Error, at.UTC().Format(time.RFC3339)) || strings.Count(meta.Error, "429") != 1 {
					t.Fatalf("rate limit message is not legible: %q", meta.Error)
				}
			}
			if scenario == "unusable response" && (!strings.Contains(meta.Error, "cannot use") || !strings.Contains(meta.Error, "booking date")) {
				t.Fatalf("unusable provider response was reduced to an opaque failure: %q", meta.Error)
			}
			if scenario == "private response" && !strings.Contains(meta.Error, "transaction retrieval failed") {
				t.Fatalf("unrecognized failure lost its fallback: %q", meta.Error)
			}
			b.fetchErr = nil
			recovered, err := a.Sync(context.Background())
			if err != nil || recovered.Status.SyncError != "" || a.ops.Consents["current"].Error != "" || a.ops.Consents["current"].NeedsReconnect {
				t.Fatal("successful retrieval did not clear the failure")
			}
		})
	}
}

// The scheduler must not spend session-status calls on a refusal the bank has
// already scheduled, and must not sit on a deadline that has passed.
func TestSyncSchedulingRespectsTheBanksOwnRetryTime(t *testing.T) {
	// Persisted deadlines carry whole seconds; compare against the same grid.
	now := time.Now().UTC().Truncate(time.Second)
	stale := now.Add(-48 * time.Hour).Format(time.RFC3339)
	waiting := operational{LastSync: stale, SyncError: "N26: rate limited", SyncRetryAt: now.Add(3 * time.Hour).Format(time.RFC3339)}
	broken := operational{LastSync: stale, SyncError: "Trade Republic: retrieval failed"}
	healthy := operational{LastSync: now.Format(time.RFC3339)}
	// Twice daily: still fresh at six hours, due again after thirteen.
	fresh := operational{LastSync: now.Add(-6 * time.Hour).Format(time.RFC3339)}
	overdue := operational{LastSync: now.Add(-13 * time.Hour).Format(time.RFC3339)}
	elapsed := operational{LastSync: stale, SyncError: waiting.SyncError, SyncRetryAt: now.Add(-time.Minute).Format(time.RFC3339)}
	cases := []struct {
		name  string
		ops   operational
		force bool
		wait  time.Duration
		due   bool
	}{
		{"waiting for the bank", waiting, false, time.Hour, false},
		{"manual sync during a wait", waiting, true, 0, true},
		{"deadline elapsed", elapsed, false, 0, true},
		{"failure without a deadline", broken, false, 0, true},
		{"recent success", healthy, false, time.Minute, false},
		{"halfway through the interval", fresh, false, time.Minute, false},
		{"interval elapsed", overdue, false, 0, true},
	}
	for _, tt := range cases {
		t.Run(tt.name, func(t *testing.T) {
			wait, due := syncSchedule(now, tt.ops, tt.force)
			if wait != tt.wait || due != tt.due {
				t.Fatalf("schedule = (%s, %t), want (%s, %t)", wait, due, tt.wait, tt.due)
			}
		})
	}
	if backoff := syncBackoff(now, waiting); backoff != 3*time.Hour+time.Minute {
		t.Fatalf("rate-limited backoff = %s, want the bank's own deadline", backoff)
	}
	if backoff := syncBackoff(now, broken); backoff != time.Hour {
		t.Fatalf("failure backoff = %s, want hourly retries", backoff)
	}
	if backoff := syncBackoff(now, healthy); backoff != 12*time.Hour {
		t.Fatalf("successful backoff = %s, want twice-daily synchronization", backoff)
	}
}