finance-duck/internal/analytics/store.go

// Package analytics maintains a disposable DuckDB projection of the plaintext dataset.
package analytics

import (
	"context"
	"database/sql"
	"fmt"
	"strings"
	"time"

	"finance-duck/internal/domain"
	_ "github.com/duckdb/duckdb-go/v2"
)

type Store struct{ db *sql.DB }

type Filter struct {
	From       string `json:"from"`
	To         string `json:"to"`
	Currency   string `json:"currency"`
	AccountID  string `json:"account_id"`
	CategoryID string `json:"category_id"`
	TagID      string `json:"tag_id"`
	MerchantID string `json:"merchant_id"`
}

type Total struct {
	Currency string `json:"currency"`
	Expenses string `json:"expenses"`
	Income   string `json:"income"`
	Net      string `json:"net"`
}

// Amount is signed net movement, not an absolute expense. Category groups overlap
// because each ancestor includes its descendants; totals never sum these groups.
type Group struct {
	ID       string `json:"id"`
	Name     string `json:"name"`
	Currency string `json:"currency"`
	Period   string `json:"period"`
	Amount   string `json:"amount"`
	Count    int64  `json:"count"`
}

type Dashboard struct {
	Totals     []Total `json:"totals"`
	Previous   []Total `json:"previous"`
	Monthly    []Group `json:"monthly"`
	Categories []Group `json:"categories"`
	Tags       []Group `json:"tags"`
	Merchants  []Group `json:"merchants"`
	Accounts   []Group `json:"accounts"`
	Recurring  []Group `json:"recurring"`
}

func Open(path string) (*Store, error) {
	db, err := sql.Open("duckdb", path)
	if err != nil {
		return nil, fmt.Errorf("open analytics: %w", err)
	}
	// A single connection bounds native resources and serializes entire dashboard
	// snapshots with rebuilds, rather than interleaving individual group queries.
	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)
	fail := func(err error) (*Store, error) { db.Close(); return nil, fmt.Errorf("initialize analytics: %w", err) }
	for _, statement := range []string{
		"SET threads = 2",
		"SET memory_limit = '256MB'",
		"SET max_temp_directory_size = '1GB'",
		"SET autoinstall_known_extensions = false",
		"SET autoload_known_extensions = false",
		"SET enable_external_access = false",
	} {
		if _, err := db.Exec(statement); err != nil {
			return fail(err)
		}
	}
	tx, err := db.Begin()
	if err != nil {
		return fail(err)
	}
	defer tx.Rollback()
	for _, statement := range schema {
		if _, err := tx.Exec(statement); err != nil {
			tx.Rollback()
			return fail(err)
		}
	}
	if err := tx.Commit(); err != nil {
		return fail(err)
	}
	return &Store{db: db}, nil
}

func (s *Store) Close() error { return s.db.Close() }

var schema = []string{
	`CREATE TABLE IF NOT EXISTS accounts (id VARCHAR PRIMARY KEY, display_name VARCHAR NOT NULL, institution VARCHAR NOT NULL, currency VARCHAR NOT NULL, external_account_id VARCHAR NOT NULL, iban VARCHAR NOT NULL, active BOOLEAN NOT NULL)`,
	`CREATE TABLE IF NOT EXISTS categories (id VARCHAR PRIMARY KEY, name VARCHAR NOT NULL, parent_id VARCHAR NOT NULL, kind VARCHAR NOT NULL)`,
	`CREATE TABLE IF NOT EXISTS tags (id VARCHAR PRIMARY KEY, name VARCHAR NOT NULL)`,
	`CREATE TABLE IF NOT EXISTS merchants (id VARCHAR PRIMARY KEY, name VARCHAR NOT NULL)`,
	`CREATE TABLE IF NOT EXISTS transactions (id VARCHAR PRIMARY KEY, source VARCHAR NOT NULL, account_id VARCHAR NOT NULL, booking_date DATE NOT NULL, value_date DATE, amount DECIMAL(24,4) NOT NULL, currency VARCHAR NOT NULL, raw_description VARCHAR NOT NULL, external_id VARCHAR NOT NULL, fingerprint VARCHAR NOT NULL, counterparty VARCHAR NOT NULL, counterparty_iban VARCHAR NOT NULL, kind VARCHAR NOT NULL, merchant_id VARCHAR NOT NULL, category_id VARCHAR NOT NULL, transfer_peer_id VARCHAR NOT NULL, classification_source VARCHAR NOT NULL, classification_model VARCHAR NOT NULL, classification_timestamp VARCHAR NOT NULL, classification_error VARCHAR NOT NULL)`,
	`CREATE TABLE IF NOT EXISTS transaction_tags (transaction_id VARCHAR NOT NULL, tag_id VARCHAR NOT NULL, PRIMARY KEY (transaction_id, tag_id))`,
	`CREATE TABLE IF NOT EXISTS category_ancestors (category_id VARCHAR NOT NULL, ancestor_id VARCHAR NOT NULL, depth INTEGER NOT NULL, PRIMARY KEY (category_id, ancestor_id))`,
	`CREATE TABLE IF NOT EXISTS postings (transaction_id VARCHAR NOT NULL, line INTEGER NOT NULL, ledger_account VARCHAR NOT NULL, account_id VARCHAR NOT NULL, category_id VARCHAR NOT NULL, currency VARCHAR NOT NULL, amount DECIMAL(24,4) NOT NULL, PRIMARY KEY (transaction_id, line))`,
}

// Rebuild replaces both schema and contents in one transaction. The database is
// only a cache: no previous index contents participate in the derived dataset.
func (s *Store) Rebuild(ctx context.Context, data domain.Dataset) error {
	if err := domain.Validate(data); err != nil {
		return fmt.Errorf("validate analytics dataset: %w", err)
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	for _, name := range []string{"postings", "category_ancestors", "transaction_tags", "transactions", "merchants", "tags", "categories", "accounts"} {
		if _, err := tx.ExecContext(ctx, "DROP TABLE IF EXISTS "+name); err != nil {
			return fmt.Errorf("reset analytics: %w", err)
		}
	}
	for _, statement := range schema {
		if _, err := tx.ExecContext(ctx, statement); err != nil {
			return fmt.Errorf("create analytics schema: %w", err)
		}
	}
	// Prepared statements prevent repeated SQL parsing and keep every dataset
	// field, including registry IDs, out of SQL source text.
	inserts := []string{
		"INSERT INTO accounts VALUES (?, ?, ?, ?, ?, ?, ?)",
		"INSERT INTO categories VALUES (?, ?, ?, ?)",
		"INSERT INTO tags VALUES (?, ?)",
		"INSERT INTO merchants VALUES (?, ?)",
		"INSERT INTO transactions VALUES (?, ?, ?, CAST(? AS DATE), CAST(NULLIF(?, '') AS DATE), CAST(? AS DECIMAL(24,4)), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
		"INSERT INTO transaction_tags VALUES (?, ?)",
		"INSERT INTO category_ancestors VALUES (?, ?, ?)",
	}
	statements := make([]*sql.Stmt, 0, len(inserts))
	defer func() {
		for _, stmt := range statements {
			stmt.Close()
		}
	}()
	for _, query := range inserts {
		stmt, err := tx.PrepareContext(ctx, query)
		if err != nil {
			return err
		}
		statements = append(statements, stmt)
	}
	exec := func(index int, args ...any) error {
		_, err := statements[index].ExecContext(ctx, args...)
		if err != nil {
			return fmt.Errorf("populate analytics: %w", err)
		}
		return nil
	}
	for _, a := range data.Accounts {
		if err := exec(0, a.ID, a.DisplayName, a.Institution, a.Currency, a.ExternalAccountID, a.IBAN, a.Active); err != nil {
			return err
		}
	}
	parents := make(map[string]string, len(data.Categories))
	for _, c := range data.Categories {
		if err := exec(1, c.ID, c.Name, c.ParentID, c.Kind); err != nil {
			return err
		}
		parents[c.ID] = c.ParentID
	}
	for _, t := range data.Tags {
		if err := exec(2, t.ID, t.Name); err != nil {
			return err
		}
	}
	for _, m := range data.Merchants {
		if err := exec(3, m.ID, m.Name); err != nil {
			return err
		}
	}
	for _, c := range data.Categories {
		ancestor := c.ID
		for depth := 0; ancestor != ""; depth++ {
			if depth >= len(data.Categories) {
				return fmt.Errorf("category ancestry cycle at %q", c.ID)
			}
			if err := exec(6, c.ID, ancestor, depth); err != nil {
				return err
			}
			ancestor = parents[ancestor]
		}
	}
	for _, t := range data.Transactions {
		f, e := t.Facts, t.Enrichment
		if err := exec(4, f.ID, f.Source, f.AccountID, f.BookingDate, f.ValueDate, string(f.Amount), f.Currency, f.RawDescription, f.ExternalID, f.Fingerprint, f.Counterparty, f.CounterpartyIBAN, e.Kind, e.MerchantID, e.CategoryID, e.TransferPeerID, e.Classification.Source, e.Classification.Model, e.Classification.Timestamp, e.Classification.Error); err != nil {
			return err
		}
		for _, tag := range e.TagIDs {
			if err := exec(5, f.ID, tag); err != nil {
				return err
			}
		}
	}
	// Each bank fact produces a balanced asset/counterpart pair. Own-account
	// transfers cancel through one clearing ledger; broker facts cancel through
	// another, where the residue left behind is exactly the cash an investment
	// account has returned: distributions and interest received, less fees.
	if _, err := tx.ExecContext(ctx, `INSERT INTO postings
  SELECT id, 1, 'asset:' || account_id, account_id, '', currency, amount FROM transactions
  UNION ALL
  SELECT id, 2, CASE kind WHEN 'transfer' THEN 'clearing:transfers' WHEN 'investment' THEN 'clearing:investments' ELSE 'category:' || category_id END,
   '', CASE WHEN kind IN ('transfer', 'investment') THEN '' ELSE category_id END, currency, -amount FROM transactions`); err != nil {
		return fmt.Errorf("derive postings: %w", err)
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit analytics rebuild: %w", err)
	}
	return nil
}

func (f Filter) validate() error {
	for _, item := range []struct{ name, value string }{{"from", f.From}, {"to", f.To}} {
		if item.value != "" {
			if _, err := time.Parse(time.DateOnly, item.value); err != nil {
				return fmt.Errorf("%s must be YYYY-MM-DD", item.name)
			}
		}
	}
	if f.From != "" && f.To != "" && f.From > f.To {
		return fmt.Errorf("from must not be after to")
	}
	return nil
}

// where uses EXISTS for many-to-many filters so a transaction carrying several
// selected tags, or several matching ancestors, can never multiply totals.
// Transfers and broker facts are excluded: moving your own money between your
// own cash and your own positions is neither spending nor income, and a broker
// history is large enough to swamp everything else if it leaked in.
func (f Filter) where() (string, []any) {
	clauses := []string{"t.kind NOT IN ('transfer', 'investment')"}
	args := []any{}
	add := func(clause string, value string) {
		if value != "" {
			clauses = append(clauses, clause)
			args = append(args, value)
		}
	}
	add("t.booking_date >= CAST(? AS DATE)", f.From)
	add("t.booking_date <= CAST(? AS DATE)", f.To)
	add("t.currency = ?", f.Currency)
	add("t.account_id = ?", f.AccountID)
	add("t.merchant_id = ?", f.MerchantID)
	add("EXISTS (SELECT 1 FROM category_ancestors ca WHERE ca.category_id = t.category_id AND ca.ancestor_id = ?)", f.CategoryID)
	if f.TagID != "" {
		ids := strings.Split(f.TagID, ",")
		placeholders := make([]string, 0, len(ids))
		for _, id := range ids {
			id = strings.TrimSpace(id)
			if id != "" {
				placeholders = append(placeholders, "?")
				args = append(args, id)
			}
		}
		if len(placeholders) == 0 {
			clauses = append(clauses, "FALSE")
		} else {
			clauses = append(clauses, "EXISTS (SELECT 1 FROM transaction_tags tt WHERE tt.transaction_id = t.id AND tt.tag_id IN ("+strings.Join(placeholders, ",")+"))")
		}
	}
	return strings.Join(clauses, " AND "), args
}