Compare commits

...
3 Commits
Author SHA1 Message Date
natercio 4fc5cd22ea make use of the store optional 2026-07-11 16:43:27 +01:00
natercio b70ac40e37 implement a store abstraction 2026-07-11 16:38:40 +01:00
natercio 0a657682f6 use options in the report 2026-07-11 16:28:35 +01:00
4 changed files with 68 additions and 11 deletions
+1 -1
View File
@@ -71,7 +71,7 @@ func run(ctx context.Context) error {
}
eg.Go(func() error {
return internal.BuildReport(ctx, reader, writer, selector)
return internal.BuildReport(ctx, reader, writer, internal.WithSelector(selector))
})
err = eg.Wait()
+12
View File
@@ -0,0 +1,12 @@
package internal
// EphemeralStore loads an empty state and discards everything on save
type EphemeralStore struct{}
func (EphemeralStore) Load() (map[string]*FillerQueue, error) {
return make(map[string]*FillerQueue), nil
}
func (EphemeralStore) Save(map[string]*FillerQueue) error {
return nil
}
+54 -9
View File
@@ -51,13 +51,52 @@ 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() (map[string]*FillerQueue, error)
Save(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)
buys, err := optionals.store.Load()
if err != nil {
return fmt.Errorf("loading state: %w", err)
}
var buysCount, sellsCount int64
var lastTimestamp time.Time
@@ -79,6 +118,11 @@ func BuildReport(ctx context.Context, reader RecordReader, writer ReportWriter,
rec, err := reader.ReadRecord(ctx)
if err != nil {
if errors.Is(err, io.EOF) {
err = optionals.store.Save(buys)
if err != nil {
return fmt.Errorf("saving state: %w", err)
}
return nil
}
return err
@@ -98,21 +142,18 @@ 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)
}
}
}
return nil
}
// 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 +163,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,7 +186,8 @@ 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.
// Since we don't apply selectors while processing buys we need to apply them here, befre we
// actually use them.
if !sel(buy) {
continue
}
+1 -1
View File
@@ -43,7 +43,7 @@ func TestBuildReport(t *testing.T) {
Taxes: decimal.Decimal{},
})).Times(1)
gotErr := internal.BuildReport(t.Context(), reader, writer, internal.Any())
gotErr := internal.BuildReport(t.Context(), reader, writer)
if gotErr != nil {
t.Fatalf("got unexpected err: %v", gotErr)
}