feat(memory): AW-backed Store with over-fetch kind filtering

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
+}

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))
+	}
+}