engine

package
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Mar 23, 2026 License: MIT Imports: 20 Imported by: 0

Documentation

Overview

Package engine provides the core reconciliation types and logic.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ParseCSVEach 

func ParseCSVEach(
	ctx context.Context,
	sourceName string,
	filePath string,
	cfg config.CSVParserCfg,
	fn func(tx Transaction, rowNum int) error,
) error

ParseCSVEach streams a CSV file, calling fn for each parsed transaction.

Ownership: each call receives a distinct Transaction by value. The callback owns the value and may retain it without copying. No struct is reused between calls.

Error format: all errors include filePath, row number (1-based), the original field value, and the source name. Example: 

bank.csv: row 452133: source "bank": invalid date "2023-99-01" with layout "2006-01-02"

Context: checked between rows. Returns ctx.Err() on cancellation.

func ReconcileStreaming 

func ReconcileStreaming(
	ctx context.Context,
	pairName string,
	leftSource string,
	rightSource string,
	leftPath string,
	rightPath string,
	leftCfg config.CSVParserCfg,
	rightCfg config.CSVParserCfg,
	pair config.Pair,
	idx RightIndex,
	w ResultWriter,
	maxTokenBuffer int,
) error

ReconcileStreaming is the canonical streaming reconciliation function.

It accepts a RightIndex (created by the caller) and a ResultWriter, and emits events incrementally as it processes both sides. ReconcileStreaming never assumes a specific RightIndex implementation — passing newMemoryIndex() is the default; a disk-backed index can be substituted without changing this function.

Token mode ordering: reference matching always runs first. Token/name matching applies only to transactions that remain unmatched after the reference pass. Token matches cannot override reference matches.

Memory complexity:

  • Right-side index: O(n_right) via the provided RightIndex
  • Right dup tracking: O(unique_refs_right) for counting + O(dup_right_txns) for groups
  • Left ref tracking: O(unique_refs_left) for counting + O(dup_left_txns) for groups
  • Token mode buffer: O(n_unmatched) worst case — guarded by maxTokenBuffer

Empty references ("") are treated as unmatched and are never grouped as duplicates.

func ReconcileStreamingWithProgress 

func ReconcileStreamingWithProgress(
	ctx context.Context,
	pairName string,
	leftSource string,
	rightSource string,
	leftPath string,
	rightPath string,
	leftCfg config.CSVParserCfg,
	rightCfg config.CSVParserCfg,
	pair config.Pair,
	idx RightIndex,
	w ResultWriter,
	maxTokenBuffer int,
	progress ProgressFunc,
	progressEvery int,
) error

ReconcileStreamingWithProgress is identical to ReconcileStreaming, but emits progress updates every progressEvery rows when progress != nil. If progressEvery <= 0, it defaults to 1,000,000 rows.

Types

type AmountDiffPair 

type AmountDiffPair struct {
	Left      Transaction `json:"left"`
	Right     Transaction `json:"right"`
	DiffMinor int64       `json:"diff_minor"`
}

AmountDiffPair is a pair where reference matched but the amount differs beyond tolerance.

type DuplicateGroup 

type DuplicateGroup struct {
	Source       string        `json:"source"`
	Reference    string        `json:"reference"`
	Transactions []Transaction `json:"transactions"`
}

DuplicateGroup is a set of transactions in the same source sharing the same reference.

type FileInfo 

type FileInfo struct {
	Path   string `json:"path"`
	SHA256 string `json:"sha256"` // lowercase hex, 64 characters
}

FileInfo holds the resolved path and SHA-256 digest of an input file.

type MatchedPair 

type MatchedPair struct {
	Left  Transaction `json:"left"`
	Right Transaction `json:"right"`
}

MatchedPair is a left+right pair that reconciled cleanly.

type PairConfigSnap 

type PairConfigSnap struct {
	DateWindow           string `json:"date_window"`
	AmountToleranceMinor int64  `json:"amount_tolerance_minor"`
	NameMode             string `json:"name_mode"`
}

PairConfigSnap is a read-only snapshot of the pair's matching rules as they were applied during this run. Embedding it in the output means an auditor reading the file later knows exactly what tolerance and window were in effect.

type ProgressEvent 

type ProgressEvent struct {
	Phase   string
	Rows    int
	Elapsed time.Duration
	Done    bool
}

ProgressEvent reports incremental progress for long-running reconciliations. Phase is one of: "right_index", "left_match".

type ProgressFunc 

type ProgressFunc func(ProgressEvent)

ProgressFunc is invoked by ReconcileStreamingWithProgress at row intervals. It runs on the calling goroutine; avoid heavy work.

type Result 

type Result struct {
	RunInfo        *RunInfo         `json:"run_info,omitempty"` // nil unless --audit mode
	PairName       string           `json:"pair"`
	LeftSource     string           `json:"left_source"`
	RightSource    string           `json:"right_source"`
	Summary        Summary          `json:"summary"`
	Matched        []MatchedPair    `json:"matched"`
	UnmatchedLeft  []Transaction    `json:"unmatched_left"`
	UnmatchedRight []Transaction    `json:"unmatched_right"`
	AmountDiff     []AmountDiffPair `json:"amount_diff"`
	TimingDiff     []TimingDiffPair `json:"timing_diff"`
	Duplicates     []DuplicateGroup `json:"duplicates"`
}

Result is the full output of a reconciliation run.

func Reconcile 

func Reconcile(pairName, leftSource, rightSource string, left, right []Transaction, pair config.Pair) (*Result, error)

Reconcile runs the reconciliation algorithm between two slices of transactions.

type ResultWriter 

type ResultWriter interface {
	WriteMatch(pair MatchedPair) error
	WriteAmountDiff(pair AmountDiffPair) error
	WriteTimingDiff(pair TimingDiffPair) error
	// WriteUnmatched writes an unmatched transaction. side must be "left" or "right".
	WriteUnmatched(tx Transaction, side string) error
	WriteDuplicate(group DuplicateGroup) error
	WriteSummary(s Summary) error
	// Flush finalizes the output (closes JSON arrays/objects, flushes CSV buffers,
	// renders table). Must be called exactly once after all events.
	Flush() error
}

ResultWriter receives reconciliation events and writes them to an output format.

Not thread-safe. Must be called from a single goroutine. Flush() must be called exactly once after all events have been written.

func NewResultWriter 

func NewResultWriter(format string, w io.Writer) (ResultWriter, error)

NewResultWriter returns a ResultWriter for the given format name. Valid formats: "json", "json-stream", "ndjson", "csv", "table".

type RightIndex 

type RightIndex interface {
	// Add inserts a transaction into the index.
	// Implementations strip Raw and convert Date to a pointer-free representation
	// before storage. Callers should not assume Raw or Date are preserved as-is.
	Add(tx Transaction) error

	// Get returns all buckets matching the given reference key.
	// Implementations may return already-used buckets; callers should check b.used.
	Get(ref string) ([]*bucket, error)

	// MarkUsed marks a bucket as consumed so future lookups cannot re-match it.
	MarkUsed(b *bucket) error

	// IterateUnused calls fn for every transaction not marked as used.
	// Iteration order is unspecified.
	IterateUnused(fn func(tx Transaction) error) error

	// Close releases any resources held by the index (file handles, temp files, etc.).
	Close() error
}

RightIndex is the right-side lookup store for ReconcileStreaming.

Not thread-safe. Single-threaded use only. Close() must be called when the index is no longer needed.

The default in-memory implementation (NewMemoryIndex) requires the right side to fit in RAM. For datasets where the right side exceeds available memory, a disk-backed implementation of this interface can be substituted. ReconcileStreaming accepts any RightIndex without modification — it is the only coupling point between the reconciler and the storage backend.

func NewDiskIndex 

func NewDiskIndex(spillDir string) (RightIndex, error)

NewDiskIndex creates a disk-backed RightIndex.

If spillDir is empty, os.TempDir() is used. A private temporary directory is created and removed on Close().

func NewMemoryIndex 

func NewMemoryIndex() RightIndex

NewMemoryIndex returns a new empty in-memory RightIndex.

type RunInfo 

type RunInfo struct {
	RunID       string         `json:"run_id"`       // 16 hex chars derived from file hashes + timestamp
	Timestamp   time.Time      `json:"timestamp"`    // UTC wall-clock time captured before parsing began
	ToolVersion string         `json:"tool_version"` // set from build -ldflags Version variable
	LeftFile    FileInfo       `json:"left_file"`
	RightFile   FileInfo       `json:"right_file"`
	PairConfig  PairConfigSnap `json:"pair_config"`
}

RunInfo carries provenance metadata for a reconciliation run. It is embedded in structured output formats (json, json-stream, ndjson) when the --audit flag is set. It is never populated in the default path.

func BuildRunInfo 

func BuildRunInfo(
	toolVersion string,
	leftPath string,
	rightPath string,
	pair config.Pair,
	ts time.Time,
) (RunInfo, error)

BuildRunInfo hashes both input files and constructs a RunInfo value. It performs two sequential file reads (one per file) before parsing begins, so the OS page cache is warm for the subsequent ParseCSVEach passes. On M1 with hardware SHA-256 acceleration, hashing ~2 GB takes roughly 1-2 seconds.

type RunInfoSetter 

type RunInfoSetter interface {
	SetRunInfo(info RunInfo) error
}

RunInfoSetter is an optional interface implemented by writers that support embedding run provenance metadata in their output. ReconcileStreaming calls this via type assertion before any events are written, so for streaming writers (ndjson) the run_info line is guaranteed to be the first line of output.

Writers that do not implement this interface silently skip run_info (csv, table).

type Summary 

type Summary struct {
	// Row counts
	TotalLeft       int     `json:"total_left"`
	TotalRight      int     `json:"total_right"`
	MatchedCount    int     `json:"matched"`
	UnmatchedLeft   int     `json:"unmatched_left"`
	UnmatchedRight  int     `json:"unmatched_right"`
	AmountDiffCount int     `json:"amount_diff_count"`
	TimingDiffCount int     `json:"timing_diff_count"`
	DuplicateCount  int     `json:"duplicate_count"`
	MatchRatePct    float64 `json:"match_rate_pct"`

	// Monetary totals (all values in minor units, e.g. cents).
	// These are always populated regardless of --audit mode.
	MatchedAmountLeft    int64 `json:"matched_amount_left"`    // sum of left.Amount for all matched pairs
	MatchedAmountRight   int64 `json:"matched_amount_right"`   // sum of right.Amount for all matched pairs
	UnmatchedAmountLeft  int64 `json:"unmatched_amount_left"`  // sum of Amount for unmatched left transactions
	UnmatchedAmountRight int64 `json:"unmatched_amount_right"` // sum of Amount for unmatched right transactions
	AmountDiffTotal      int64 `json:"amount_diff_total"`      // sum of abs(DiffMinor) across all amount_diff pairs
	TotalDiscrepancy     int64 `json:"total_discrepancy"`      // UnmatchedAmountLeft + UnmatchedAmountRight + AmountDiffTotal
}

Summary holds aggregate counts, match rate, and monetary totals for a reconciliation run.

type TimingDiffPair 

type TimingDiffPair struct {
	Left     Transaction `json:"left"`
	Right    Transaction `json:"right"`
	DaysDiff int         `json:"days_diff"`
}

TimingDiffPair is a pair where reference+amount matched but the date is outside the window.

type Transaction 

type Transaction struct {
	ID        string            `json:"id"`
	Date      time.Time         `json:"date"`
	Amount    int64             `json:"amount"` // always in minor units (e.g. kobo, cents)
	Currency  string            `json:"currency"`
	Reference string            `json:"reference"`
	Name      string            `json:"name"`
	Source    string            `json:"source"` // source name from config
	Raw       map[string]string `json:"raw,omitempty"`
}

Transaction is a normalized financial record from any source.

func ParseCSV 

func ParseCSV(sourceName string, filePath string, cfg config.CSVParserCfg) ([]Transaction, error)

ParseCSV reads a CSV file and returns normalized transactions according to the source config. It is a convenience wrapper around ParseCSVEach for callers that need a complete slice.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.