From fc3cfb523b54b62753d0b5c7d857db04b16780d2 Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Sat, 2 May 2026 04:23:37 -0400 Subject: [PATCH] feat(monitor/redisring): sorted-set ring buffer with score-based retention --- .../backend/internal/redisring/ring.go | 79 +++++++++++++++++ .../backend/internal/redisring/ring_test.go | 87 +++++++++++++++++++ 2 files changed, 166 insertions(+) create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/redisring/ring.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/redisring/ring_test.go diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/redisring/ring.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/redisring/ring.go new file mode 100644 index 00000000..cf6be854 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/redisring/ring.go @@ -0,0 +1,79 @@ +// ©AngelaMos | 2026 +// ring.go + +package redisring + +import ( + "context" + "fmt" + "strconv" + "time" + + "github.com/redis/go-redis/v9" +) + +const ( + defaultRetention = 24 * time.Hour +) + +type Config struct { + Retention time.Duration +} + +type Client struct { + rc *redis.Client + retention time.Duration +} + +func New(rc *redis.Client, cfg Config) *Client { + if cfg.Retention <= 0 { + cfg.Retention = defaultRetention + } + return &Client{rc: rc, retention: cfg.Retention} +} + +func (c *Client) Push(ctx context.Context, key string, score int64, payload []byte) error { + cutoff := time.UnixMilli(score).Add(-c.retention).UnixMilli() + pipe := c.rc.Pipeline() + pipe.ZAdd(ctx, key, redis.Z{Score: float64(score), Member: payload}) + pipe.ZRemRangeByScore(ctx, key, "-inf", "("+strconv.FormatInt(cutoff, 10)) + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("ring push %s @ %d: %w", key, score, err) + } + return nil +} + +func (c *Client) Recent(ctx context.Context, key string, n int) ([][]byte, error) { + if n <= 0 { + return nil, nil + } + res, err := c.rc.ZRevRangeByScore(ctx, key, &redis.ZRangeBy{ + Min: "-inf", + Max: "+inf", + Offset: 0, + Count: int64(n), + }).Result() + if err != nil { + return nil, fmt.Errorf("ring recent %s: %w", key, err) + } + out := make([][]byte, 0, len(res)) + for _, s := range res { + out = append(out, []byte(s)) + } + return out, nil +} + +func (c *Client) Range(ctx context.Context, key string, fromScore, toScore int64) ([][]byte, error) { + res, err := c.rc.ZRangeByScore(ctx, key, &redis.ZRangeBy{ + Min: strconv.FormatInt(fromScore, 10), + Max: strconv.FormatInt(toScore, 10), + }).Result() + if err != nil { + return nil, fmt.Errorf("ring range %s: %w", key, err) + } + out := make([][]byte, 0, len(res)) + for _, s := range res { + out = append(out, []byte(s)) + } + return out, nil +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/redisring/ring_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/redisring/ring_test.go new file mode 100644 index 00000000..6ec06567 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/redisring/ring_test.go @@ -0,0 +1,87 @@ +// ©AngelaMos | 2026 +// ring_test.go + +package redisring_test + +import ( + "context" + "fmt" + "testing" + "time" + + goredis "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/require" + tcredis "github.com/testcontainers/testcontainers-go/modules/redis" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/redisring" +) + +func setupRedis(t *testing.T) *goredis.Client { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + + rc, err := tcredis.Run(ctx, "redis:7-alpine") + require.NoError(t, err) + t.Cleanup(func() { _ = rc.Terminate(context.Background()) }) + + uri, err := rc.ConnectionString(ctx) + require.NoError(t, err) + opt, err := goredis.ParseURL(uri) + require.NoError(t, err) + c := goredis.NewClient(opt) + t.Cleanup(func() { _ = c.Close() }) + return c +} + +func TestRing_PushAndRecentRoundtrips(t *testing.T) { + c := setupRedis(t) + r := redisring.New(c, redisring.Config{Retention: 24 * time.Hour}) + ctx := context.Background() + + now := time.Now().UTC() + for i := 0; i < 5; i++ { + ts := now.Add(-time.Duration(i) * time.Minute) + require.NoError(t, r.Push(ctx, "test:key", ts.UnixMilli(), []byte(fmt.Sprintf(`{"i":%d}`, i)))) + } + + got, err := r.Recent(ctx, "test:key", 3) + require.NoError(t, err) + require.Len(t, got, 3) +} + +func TestRing_RetentionPrunesOldEntries(t *testing.T) { + c := setupRedis(t) + r := redisring.New(c, redisring.Config{Retention: 1 * time.Hour}) + ctx := context.Background() + + now := time.Now().UTC() + for i := 0; i < 5; i++ { + old := now.Add(-2 * time.Hour).Add(-time.Duration(i) * time.Minute) + require.NoError(t, r.Push(ctx, "test:key", old.UnixMilli(), []byte(fmt.Sprintf(`old-%d`, i)))) + } + require.NoError(t, r.Push(ctx, "test:key", now.UnixMilli(), []byte("now"))) + + got, err := r.Recent(ctx, "test:key", 100) + require.NoError(t, err) + require.Len(t, got, 1) + require.Equal(t, "now", string(got[0])) +} + +func TestRing_RangeReturnsItemsWithinScoreWindow(t *testing.T) { + c := setupRedis(t) + r := redisring.New(c, redisring.Config{Retention: 24 * time.Hour}) + ctx := context.Background() + + now := time.Now().UTC() + for i := 0; i < 10; i++ { + ts := now.Add(-time.Duration(i) * time.Minute) + require.NoError(t, r.Push(ctx, "test:key", ts.UnixMilli(), []byte(fmt.Sprintf("i=%d", i)))) + } + + from := now.Add(-3 * time.Minute).UnixMilli() + to := now.UnixMilli() + got, err := r.Range(ctx, "test:key", from, to) + require.NoError(t, err) + require.Len(t, got, 4, "expected items in [now-3m, now] inclusive") +}