finance-duck/internal/app/reclassify.go

package app

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"slices"
	"strings"
	"time"

	"finance-duck/internal/classification"
	"finance-duck/internal/domain"
)

type Fields struct {
	Merchant bool `json:"merchant"`
	Category bool `json:"category"`
	Tags     bool `json:"tags"`
}
type PreviewRequest struct {
	Revision string `json:"revision"`
	From     string `json:"from"`
	To       string `json:"to"`
	Model    string `json:"model"`
	Fields   Fields `json:"fields"`
}
type Change struct {
	ID           string            `json:"id"`
	Description  string            `json:"description"`
	Counterparty string            `json:"counterparty"`
	Amount       domain.Money      `json:"amount"`
	Currency     string            `json:"currency"`
	Before       domain.Enrichment `json:"before"`
	After        domain.Enrichment `json:"after"`
}

// EnrichmentEdit is a reviewer's correction to one proposal: it replaces the
// proposed category and tags before the change is applied. A corrected
// transaction is classified by the human, not the model, so its provenance
// becomes manual and later runs treat it accordingly.
type EnrichmentEdit struct {
	ID         string   `json:"id"`
	CategoryID string   `json:"category_id"`
	TagIDs     []string `json:"tag_ids"`
}
type ClassificationError struct {
	ID    string `json:"id"`
	Error string `json:"error"`
}
type Preview struct {
	ID           string                `json:"id"`
	Revision     string                `json:"revision"`
	Changes      []Change              `json:"changes"`
	Analysed     int                   `json:"analysed"`
	Unchanged    int                   `json:"unchanged"`
	Errors       []ClassificationError `json:"errors"`
	NewMerchants []domain.Merchant     `json:"new_merchants"`
	created      time.Time
}

// PreviewProgress is the live state of one preview run. Errors accumulate as
// they happen so a failing provider is visible after seconds, not after the
// whole paced range. Preview is set only when Done with an empty Error.
type PreviewProgress struct {
	ID        string                `json:"id"`
	Total     int                   `json:"total"`
	Analysed  int                   `json:"analysed"`
	Changes   int                   `json:"changes"`
	Unchanged int                   `json:"unchanged"`
	Errors    []ClassificationError `json:"errors"`
	Done      bool                  `json:"done"`
	Error     string                `json:"error,omitempty"`
	Preview   *Preview              `json:"preview,omitempty"`
}

func (p PreviewProgress) clone() PreviewProgress {
	p.Errors = append([]ClassificationError{}, p.Errors...)
	return p
}

// previewJob is the single in-flight (or most recently finished) preview run.
// status is guarded by App.mu; cancel stops the goroutine cooperatively.
type previewJob struct {
	cancel context.CancelFunc
	status PreviewProgress
}

func validRange(from, to string) error {
	f, e := time.Parse("2006-01-02", from)
	if e != nil {
		return errors.New("from must be YYYY-MM-DD")
	}
	t, e := time.Parse("2006-01-02", to)
	if e != nil {
		return errors.New("to must be YYYY-MM-DD")
	}
	if f.After(t) {
		return errors.New("from must not exceed to")
	}
	return nil
}
func validatePreviewRequest(r PreviewRequest) error {
	if err := validRange(r.From, r.To); err != nil {
		return err
	}
	if !r.Fields.Merchant && !r.Fields.Category && !r.Fields.Tags {
		return errors.New("select at least one enrichment field")
	}
	if strings.TrimSpace(r.Model) == "" {
		return errors.New("model is required")
	}
	return nil
}
func previewEligible(t domain.Transaction, r PreviewRequest) bool {
	return t.Facts.BookingDate >= r.From && t.Facts.BookingDate <= r.To &&
		t.Enrichment.Kind != "transfer" && t.Enrichment.Kind != domain.KindInvestment
}

// StartPreview validates the request against the current journal and starts a
// background classification run. The provider is paced to one request every
// few seconds, so any real range takes minutes: the caller polls
// PreviewProgress instead of holding an HTTP request open for the duration.
// Only one run exists at a time; the run owns its own snapshot and never
// touches canonical data.
func (a *App) StartPreview(ctx context.Context, r PreviewRequest) (PreviewProgress, error) {
	if err := validatePreviewRequest(r); err != nil {
		return PreviewProgress{}, err
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.previewRun != nil && !a.previewRun.status.Done {
		return PreviewProgress{}, errors.New("a preview is already being generated; stop it first")
	}
	s, err := a.snapshot(ctx)
	if err != nil {
		return PreviewProgress{}, err
	}
	if r.Revision != s.Revision {
		return PreviewProgress{}, errors.New("revision conflict: reload before analysing")
	}
	client := a.classifier.WithModel(r.Model)
	total := 0
	for _, t := range s.Data.Transactions {
		if previewEligible(t, r) {
			total++
		}
	}
	runCtx, cancel := context.WithCancel(context.Background())
	job := &previewJob{cancel: cancel, status: PreviewProgress{ID: domain.NewID("preview"), Total: total, Errors: []ClassificationError{}}}
	a.previewRun = job
	go a.runPreview(runCtx, cancel, client, s, r, job)
	return job.status.clone(), nil
}

func (a *App) runPreview(ctx context.Context, cancel context.CancelFunc, client *classification.Client, s State, r PreviewRequest, job *previewJob) {
	defer cancel()
	p, err := classifyRange(ctx, client, s, r, job.status.ID, func(u PreviewProgress) {
		a.mu.Lock()
		if a.previewRun == job {
			job.status = u
		}
		a.mu.Unlock()
	})
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.previewRun != job {
		return // stopped by CancelPreview; discard the result
	}
	job.status.Done = true
	if err != nil {
		job.status.Error = err.Error()
		return
	}
	for id, old := range a.previews {
		if time.Since(old.created) > time.Hour {
			delete(a.previews, id)
		}
	}
	if len(a.previews) >= 20 {
		job.status.Error = "too many active previews; cancel one first"
		return
	}
	a.previews[p.ID] = p
	job.status.Analysed = p.Analysed
	job.status.Changes = len(p.Changes)
	job.status.Unchanged = p.Unchanged
	job.status.Errors = append([]ClassificationError{}, p.Errors...)
	job.status.Preview = &p
}

// PreviewProgress reports the current (or most recently finished) preview run.
// An empty id re-attaches to whatever run exists, so navigating away from the
// page does not orphan a run that is still spending provider requests.
func (a *App) PreviewProgress(id string) (PreviewProgress, error) {
	a.mu.Lock()
	defer a.mu.Unlock()
	job := a.previewRun
	if job == nil || (id != "" && job.status.ID != id) {
		return PreviewProgress{}, errors.New("no matching preview run; analyse again")
	}
	return job.status.clone(), nil
}

// classifyRange proposes enrichment for every eligible transaction in the
// snapshot, reporting progress after each one. It stops early when the run has
// produced no successful proposal yet and the same error message repeats three
// times in a row: an identical repeated failure is a configuration or provider
// problem, and grinding through the rest of the paced range would only repeat
// it a few seconds apart.
func classifyRange(ctx context.Context, client *classification.Client, s State, r PreviewRequest, id string, report func(PreviewProgress)) (Preview, error) {
	p := Preview{ID: id, Revision: s.Revision, Changes: []Change{}, Errors: []ClassificationError{}, created: time.Now()}
	baseMerchants := len(s.Data.Merchants)
	eligible := []domain.Transaction{}
	for _, t := range s.Data.Transactions {
		if previewEligible(t, r) {
			eligible = append(eligible, t)
		}
	}
	total := len(eligible)
	progress := func() {
		if report != nil {
			report(PreviewProgress{ID: id, Total: total, Analysed: p.Analysed, Changes: len(p.Changes), Unchanged: p.Unchanged, Errors: append([]ClassificationError{}, p.Errors...)})
		}
	}
	succeeded := false
	repeated := 0
	// One provider request classifies a whole chunk. Rows are partitioned by
	// transaction kind because expense and income use different category
	// enums; within a kind they keep journal order. New merchants proposed by
	// one chunk are registered before the next chunk runs, so later
	// duplicates link instead of minting again.
	chunks := [][]domain.Transaction{}
	for _, kind := range []string{"expense", "income"} {
		group := []domain.Transaction{}
		for _, t := range eligible {
			if domain.Fallback(t.Facts).Kind == kind {
				group = append(group, t)
			}
		}
		for start := 0; start < len(group); start += classification.MaxBatch {
			chunks = append(chunks, group[start:min(start+classification.MaxBatch, len(group))])
		}
	}
	for _, chunk := range chunks {
		if err := ctx.Err(); err != nil {
			return Preview{}, err
		}
		facts := make([]domain.Facts, len(chunk))
		for i, t := range chunk {
			facts[i] = t.Facts
		}
		results := client.ClassifyBatch(ctx, facts, s.Data)
		if err := ctx.Err(); err != nil {
			return Preview{}, err
		}
		// A chunk can mix one slow request's failures with later successes;
		// count the successes first so a working run is never aborted by the
		// repeated-identical-failure heuristic.
		for _, result := range results {
			if result.Err == nil {
				succeeded = true
			}
		}
		for i, t := range chunk {
			p.Analysed++
			proposal, e := results[i].Proposal, results[i].Err
			if e != nil {
				if n := len(p.Errors); n > 0 && p.Errors[n-1].Error == e.Error() {
					repeated++
				} else {
					repeated = 1
				}
				p.Errors = append(p.Errors, ClassificationError{t.Facts.ID, e.Error()})
				if !succeeded && repeated >= 3 {
					return Preview{}, fmt.Errorf("stopped after %d identical failures — %s — with %d of %d transactions not analysed", repeated, e.Error(), total-p.Analysed, total)
				}
				progress()
				continue
			}
			succeeded = true
			after := t.Enrichment
			if r.Fields.Merchant {
				after.MerchantID = proposal.Enrichment.MerchantID
				if e = addProposal(&s.Data, proposal, t.Facts); e != nil {
					p.Errors = append(p.Errors, ClassificationError{t.Facts.ID, e.Error()})
					progress()
					continue
				}
			}
			if r.Fields.Category {
				after.CategoryID = proposal.Enrichment.CategoryID
			}
			if r.Fields.Tags {
				after.TagIDs = slices.Clone(proposal.Enrichment.TagIDs)
			}
			if e = domain.ValidateEnrichment(s.Data, t.Facts, after); e != nil {
				p.Errors = append(p.Errors, ClassificationError{t.Facts.ID, e.Error()})
				progress()
				continue
			}
			beforeComparable, afterComparable := t.Enrichment, after
			beforeComparable.Classification = domain.Provenance{}
			afterComparable.Classification = domain.Provenance{}
			beforeComparable.TagIDs = slices.Clone(beforeComparable.TagIDs)
			afterComparable.TagIDs = slices.Clone(afterComparable.TagIDs)
			slices.Sort(beforeComparable.TagIDs)
			slices.Sort(afterComparable.TagIDs)
			if reflect.DeepEqual(beforeComparable, afterComparable) {
				p.Unchanged++
				progress()
				continue
			}
			after.Classification = proposal.Enrichment.Classification
			p.Changes = append(p.Changes, Change{
				ID: t.Facts.ID, Description: t.Facts.RawDescription, Counterparty: t.Facts.Counterparty,
				Amount: t.Facts.Amount, Currency: t.Facts.Currency,
				Before: t.Enrichment, After: after,
			})
			progress()
		}
	}
	p.NewMerchants = append([]domain.Merchant{}, s.Data.Merchants[baseMerchants:]...)
	return p, nil
}

// enrichmentEqual compares enrichment semantically: tag order is not a change.
func enrichmentEqual(a, b domain.Enrichment) bool {
	a.TagIDs = slices.Clone(a.TagIDs)
	b.TagIDs = slices.Clone(b.TagIDs)
	slices.Sort(a.TagIDs)
	slices.Sort(b.TagIDs)
	return reflect.DeepEqual(a, b)
}

// ApplyPreview rebases the selected proposals onto the current journal. A
// preview run is minutes long by design, so unrelated commits (a scheduled
// sync, an import, an earlier partial apply of this same preview) must not
// invalidate the review; only a selected transaction whose own enrichment
// changed since the preview snapshot conflicts. Applied changes are pruned so
// the remaining proposals stay appliable without another paced provider run.
func (a *App) ApplyPreview(ctx context.Context, id, rev string, ids []string, edits []EnrichmentEdit) (State, error) {
	a.mu.Lock()
	defer a.mu.Unlock()
	p, ok := a.previews[id]
	if !ok || time.Since(p.created) > time.Hour {
		return State{}, errors.New("preview expired or unknown; analyse again")
	}
	if rev != p.Revision {
		return State{}, errors.New("revision conflict: preview was generated from different records")
	}
	s, err := a.snapshot(ctx)
	if err != nil {
		return State{}, err
	}
	changes := map[string]Change{}
	for _, c := range p.Changes {
		changes[c.ID] = c
	}
	selected := map[string]bool{}
	for _, id := range ids {
		if _, ok := changes[id]; !ok {
			return State{}, errors.New("selected transaction is not in preview")
		}
		selected[id] = true
	}
	if len(selected) == 0 {
		return State{}, errors.New("select at least one change")
	}
	edited := map[string]EnrichmentEdit{}
	for _, e := range edits {
		if !selected[e.ID] {
			return State{}, errors.New("edited transaction is not selected")
		}
		edited[e.ID] = e
	}
	// Edits are validated against the dataset the change will land in, which
	// includes merchants this preview mints only when the change is applied.
	validation := s.Data
	validation.Merchants = append(append([]domain.Merchant{}, s.Data.Merchants...), p.NewMerchants...)
	applied := 0
	needed := map[string]bool{}
	for i, t := range s.Data.Transactions {
		if !selected[t.Facts.ID] {
			continue
		}
		c := changes[t.Facts.ID]
		if !enrichmentEqual(t.Enrichment, c.Before) {
			return State{}, errors.New("revision conflict: a selected transaction changed after the preview; analyse it again")
		}
		after := c.After
		if e, ok := edited[t.Facts.ID]; ok {
			after.CategoryID = e.CategoryID
			after.TagIDs = append([]string{}, e.TagIDs...)
			after.Classification = domain.Provenance{Source: "manual", Timestamp: time.Now().UTC().Format(time.RFC3339)}
			if err := domain.ValidateEnrichment(validation, t.Facts, after); err != nil {
				return State{}, fmt.Errorf("edited classification for %s is invalid: %w", t.Facts.ID, err)
			}
		}
		s.Data.Transactions[i].Enrichment = after
		needed[after.MerchantID] = true
		applied++
	}
	if applied != len(selected) {
		return State{}, errors.New("revision conflict: a selected transaction no longer exists; analyse again")
	}
	existing := map[string]bool{}
	for _, m := range s.Data.Merchants {
		existing[m.ID] = true
	}
	for _, m := range p.NewMerchants {
		if needed[m.ID] && !existing[m.ID] {
			s.Data.Merchants = append(s.Data.Merchants, m)
		}
	}
	for _, t := range s.Data.Transactions {
		if selected[t.Facts.ID] && t.Enrichment.MerchantID != "" {
			// New merchants already carry their first alias; existing merchants
			// learn only when the real matcher stays unambiguous.
			LearnAlias(&s.Data, t.Facts, t.Enrichment.MerchantID)
		}
	}
	state, err := a.commit(ctx, s.Revision, s.Data)
	if err != nil {
		return State{}, err
	}
	kept := make([]Change, 0, len(p.Changes)-applied)
	for _, c := range p.Changes {
		if !selected[c.ID] {
			kept = append(kept, c)
		}
	}
	if len(kept) == 0 {
		delete(a.previews, id)
		if job := a.previewRun; job != nil && job.status.ID == id {
			a.previewRun = nil
		}
		return state, nil
	}
	p.Changes = kept
	a.previews[id] = p
	return state, nil
}

// CancelPreview stops a running preview job and discards a finished preview.
// A run and its stored preview share one id, so a single cancel covers both.
func (a *App) CancelPreview(id string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	if job := a.previewRun; job != nil && job.status.ID == id {
		job.cancel()
		a.previewRun = nil
	}
	delete(a.previews, id)
}