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"` } 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) (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") } 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") } s.Data.Transactions[i].Enrichment = c.After needed[c.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) }