feat(monitor/redisring): sorted-set ring buffer with score-based retention

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

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