Internal state persistence (#27)

Allow us to generate year over year reports without having to rerun everything from the beginning.

Co-authored-by: Natercio Moniz <[email protected]>

internal/report.go

@@ -51,18 +51,58 @@ type ReportWriter interface {
 	Write(context.Context, ReportItem) error
 }
 
+type optionals struct {
+	selector Selector
+	store    Store
+}
+
+func applyOptions(defaults optionals, opts []Option) optionals {
+	for _, opt := range opts {
+		opt(&defaults)
+	}
+	return defaults
+}
+
+type Option func(*optionals)
+
 // Selector returns true if a record should be selected for processing, false otherwise.
 type Selector func(Record) bool
 
+func WithSelector(s Selector) Option {
+	return func(o *optionals) {
+		o.selector = s
+	}
+}
+
+type Store interface {
+	Load(context.Context) (map[string]*FillerQueue, error)
+	Save(context.Context, map[string]*FillerQueue) error
+}
+
+func WithStore(s Store) Option {
+	return func(o *optionals) {
+		o.store = s
+	}
+}
+
 // BuildReport reads records from a RecordReader and, if the record passes the Selector, it is
 // processed into the ReportWriter.
-func BuildReport(ctx context.Context, reader RecordReader, writer ReportWriter, sel Selector) error {
-	buys := make(map[string]*FillerQueue)
+func BuildReport(ctx context.Context, reader RecordReader, writer ReportWriter, options ...Option) error {
+	optionals := applyOptions(optionals{
+		selector: Any(),
+		store:    EphemeralStore{},
+	}, options)
 
-	var buysCount, sellsCount int64
-	var lastTimestamp time.Time
-	progTicker := time.NewTicker(10 * time.Second)
+	buys, err := optionals.store.Load(ctx)
+	if err != nil {
+		return fmt.Errorf("loading state: %w", err)
+	}
 
+	var (
+		recordsCount  int64
+		lastTimestamp time.Time
+		progTicker    = time.NewTicker(10 * time.Second)
+	)
 	for {
 		select {
 		case <-ctx.Done():
@@ -70,27 +110,29 @@ func BuildReport(ctx context.Context, reader RecordReader, writer ReportWriter,
 		case <-progTicker.C:
 			slog.InfoContext(
 				ctx, "Progress update",
-				slog.Int64("total_records", buysCount+sellsCount),
-				slog.Int64("sell_records", sellsCount),
-				slog.Int64("buy_records", buysCount),
+				slog.Int64("records_count", recordsCount),
 				slog.Time("last_record_timestamp", lastTimestamp),
 			)
 		default:
 			rec, err := reader.ReadRecord(ctx)
 			if err != nil {
 				if errors.Is(err, io.EOF) {
+					err = optionals.store.Save(ctx, buys)
+					if err != nil {
+						return fmt.Errorf("saving state: %w", err)
+					}
+
 					return nil
 				}
 				return err
 			}
 
-			if rec.Kind().Is(KindBuy) {
-				buysCount++
-			} else if rec.Kind().Is(KindSell) {
-				sellsCount++
+			if !rec.Kind().Valid() {
+				return fmt.Errorf("found invalid Kind(%d)", rec.Kind())
 			}
 
 			lastTimestamp = rec.Timestamp()
+			recordsCount++
 
 			buyQueue, ok := buys[rec.Symbol()]
 			if !ok {
@@ -98,21 +140,16 @@ func BuildReport(ctx context.Context, reader RecordReader, writer ReportWriter,
 				buys[rec.Symbol()] = buyQueue
 			}
 
-			err = processRecord(ctx, buyQueue, rec, sel, writer)
+			err = processRecord(ctx, buyQueue, rec, optionals.selector, writer)
 			if err != nil {
 				return fmt.Errorf("processing record: %w", err)
 			}
-
 		}
 	}
 }
 
 // processRecord either adds buys to the queue or consumes buys from the queue when processing a
 // sell record.
-//
-// NOTE: Selectors are only applied when processing sell records for performance reasons. It's much
-// cheaper to just accumulate buys and only actually inspect any records once a sell happens. This
-// avoids potential network requests to for every single record.
 func processRecord(ctx context.Context, q *FillerQueue, rec Record, sel Selector, writer ReportWriter) error {
 	slog.Debug(
 		"Report: processing record",
@@ -122,6 +159,9 @@ func processRecord(ctx context.Context, q *FillerQueue, rec Record, sel Selector
 
 	switch rec.Kind() {
 	case KindBuy:
+		// Selectors are only applied when processing sell records for performance reasons. It's much
+		// cheaper to just accumulate buys and only actually inspect any records once a sell happens. This
+		// avoids potential network requests to for every single record.
 		q.Push(NewFiller(rec))
 
 	case KindSell:
@@ -142,11 +182,6 @@ func processRecord(ctx context.Context, q *FillerQueue, rec Record, sel Selector
 				return ErrInsufficientBoughtVolume
 			}
 
-			// Since we don't apply selectors while processing buys we need to apply them here.
-			if !sel(buy) {
-				continue
-			}
-
 			matchedQty, filled := buy.Fill(unmatchedQty)
 
 			if filled {