diff --git a/internal/memory/memory.go b/internal/memory/memory.go index 3b82f63..aa8e147 100644 --- a/internal/memory/memory.go +++ b/internal/memory/memory.go @@ -8,6 +8,8 @@ import ( "context" "sync" "time" + + "keel/internal/aw" ) // Event is one derived event Keel chooses to remember. Kind is the event type @@ -68,3 +70,63 @@ func (f *Fake) Events() []Event { defer f.mu.Unlock() return append([]Event(nil), f.events...) } + +// overfetch bounds how many recent bucket events awStore pulls before filtering +// by kind, because keel.events interleaves all kinds. Events older than this +// window are not seen — acceptable for "recent". +const overfetch = 200 + +// kindKey is the data field carrying the Event.Kind inside an AW event. +const kindKey = "_kind" + +// AWClient is the subset of *aw.Client that awStore needs. Declaring it here +// lets tests inject a fake without HTTP; *aw.Client satisfies it structurally. +type AWClient interface { + Insert(ctx context.Context, bucketID string, e aw.Event) error + Recent(ctx context.Context, bucketID string, limit int) ([]aw.Event, error) +} + +type awStore struct { + c AWClient + bucketID string +} + +// NewAWStore returns a Store that persists events to the given AW bucket. +func NewAWStore(c AWClient, bucketID string) Store { + return &awStore{c: c, bucketID: bucketID} +} + +func (s *awStore) Record(ctx context.Context, e Event) error { + data := make(map[string]any, len(e.Data)+1) + for k, v := range e.Data { + data[k] = v + } + data[kindKey] = e.Kind + return s.c.Insert(ctx, s.bucketID, aw.Event{Timestamp: e.At, Duration: 0, Data: data}) +} + +func (s *awStore) Recent(ctx context.Context, kind string, n int) ([]Event, error) { + if n <= 0 { + return nil, nil + } + raw, err := s.c.Recent(ctx, s.bucketID, overfetch) + if err != nil { + return nil, err + } + var out []Event + for _, ev := range raw { + if len(out) >= n { + break + } + if k, _ := ev.Data[kindKey].(string); k == kind { + data := make(map[string]any, len(ev.Data)) + for dk, dv := range ev.Data { + if dk != kindKey { + data[dk] = dv + } + } + out = append(out, Event{Kind: kind, At: ev.Timestamp, Data: data}) + } + } + return out, nil +} diff --git a/internal/memory/memory_test.go b/internal/memory/memory_test.go index 2bc14c7..c7d72b1 100644 --- a/internal/memory/memory_test.go +++ b/internal/memory/memory_test.go @@ -4,6 +4,8 @@ import ( "context" "testing" "time" + + "keel/internal/aw" ) func TestFakeRecordThenRecentNewestFirstByKind(t *testing.T) { @@ -44,3 +46,70 @@ func TestNopStoreIsInert(t *testing.T) { t.Fatalf("nop Recent = %+v, %v", got, err) } } + +// fakeAWClient is the AWClient subset, recording inserts and replaying a fixed +// recent list so we can test mapping and over-fetch filtering without HTTP. +type fakeAWClient struct { + inserted []aw.Event + recent []aw.Event // newest-first, mixed kinds + lastLimit int +} + +func (f *fakeAWClient) Insert(_ context.Context, _ string, e aw.Event) error { + f.inserted = append(f.inserted, e) + return nil +} +func (f *fakeAWClient) Recent(_ context.Context, _ string, limit int) ([]aw.Event, error) { + f.lastLimit = limit + return f.recent, nil +} + +func TestAWStoreRecordMapsKindIntoData(t *testing.T) { + fc := &fakeAWClient{} + s := NewAWStore(fc, "keel.events") + at := time.Unix(1000, 0).UTC() + if err := s.Record(context.Background(), Event{Kind: "proposal_made", At: at, Data: map[string]any{"proposal_id": "x"}}); err != nil { + t.Fatalf("Record: %v", err) + } + if len(fc.inserted) != 1 { + t.Fatalf("inserted %d", len(fc.inserted)) + } + got := fc.inserted[0] + if !got.Timestamp.Equal(at) || got.Data["_kind"] != "proposal_made" || got.Data["proposal_id"] != "x" { + t.Fatalf("mapped event = %+v", got) + } +} + +func TestAWStoreRecentOverfetchesAndFiltersByKind(t *testing.T) { + fc := &fakeAWClient{recent: []aw.Event{ + {Timestamp: time.Unix(3, 0), Data: map[string]any{"_kind": "proposal_made", "n": "b"}}, + {Timestamp: time.Unix(2, 0), Data: map[string]any{"_kind": "action_taken"}}, + {Timestamp: time.Unix(1, 0), Data: map[string]any{"_kind": "proposal_made", "n": "a"}}, + }} + s := NewAWStore(fc, "keel.events") + got, err := s.Recent(context.Background(), "proposal_made", 5) + if err != nil { + t.Fatalf("Recent: %v", err) + } + if fc.lastLimit != overfetch { + t.Fatalf("over-fetch limit = %d, want %d", fc.lastLimit, overfetch) + } + if len(got) != 2 || got[0].Kind != "proposal_made" || got[0].Data["n"] != "b" { + t.Fatalf("filtered = %+v", got) + } + if _, leaked := got[0].Data["_kind"]; leaked { + t.Fatalf("_kind leaked into returned Data: %+v", got[0].Data) + } +} + +func TestAWStoreRecentCapsAtN(t *testing.T) { + var recent []aw.Event + for i := 0; i < 5; i++ { + recent = append(recent, aw.Event{Timestamp: time.Unix(int64(i), 0), Data: map[string]any{"_kind": "proposal_made"}}) + } + s := NewAWStore(&fakeAWClient{recent: recent}, "keel.events") + got, _ := s.Recent(context.Background(), "proposal_made", 2) + if len(got) != 2 { + t.Fatalf("cap not applied: %d", len(got)) + } +}