From 5118eb2e290c2da59b214937aa7d84d250cb6933 Mon Sep 17 00:00:00 2001
From: Chris Guidry <chris@theguidrys.us>
Date: Thu, 5 Mar 2026 08:29:51 -0500
Subject: [PATCH] Clean up empty subscriber map entries and add TTL to Redis
 buffer
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The broker's subscriber map accumulated empty slices for paths that no
longer had any subscribers. Now it deletes the key when the last one
unsubscribes. The Redis buffer key also had no expiry, so it would
stick around forever after all instances shut down — now it gets a 24h
TTL refreshed on each publish.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 backend_redis.go      |  2 ++
 backend_redis_test.go | 15 +++++++++++++++
 broker.go             |  3 +++
 broker_test.go        | 23 +++++++++++++++++++++++
 4 files changed, 43 insertions(+)

diff --git a/backend_redis.go b/backend_redis.go
index c5e2d8b..864ed3e 100644
--- a/backend_redis.go
+++ b/backend_redis.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"log"
 	"sync"
+	"time"
 
 	"github.com/redis/go-redis/v9"
 )
@@ -53,6 +54,7 @@ func (r *RedisBackend) Publish(event *Event) error {
 	pipe := r.client.Pipeline()
 	pipe.LPush(ctx, redisBufferKey, data)
 	pipe.LTrim(ctx, redisBufferKey, 0, int64(r.bufferSize-1))
+	pipe.Expire(ctx, redisBufferKey, 24*time.Hour)
 	pipe.Publish(ctx, redisChannel, data)
 	_, err = pipe.Exec(ctx)
 	return err
diff --git a/backend_redis_test.go b/backend_redis_test.go
index 5852a09..e73a644 100644
--- a/backend_redis_test.go
+++ b/backend_redis_test.go
@@ -169,6 +169,21 @@ func TestRedisBackend_bufferTrims(t *testing.T) {
 	}
 }
 
+func TestRedisBackend_bufferHasTTL(t *testing.T) {
+	backend, mr := newTestRedisBackend(t, 100)
+	defer backend.Close()
+
+	backend.Publish(&Event{ID: "e1", Path: "test", Payload: map[string]any{}})
+
+	ttl := mr.TTL(redisBufferKey)
+	if ttl <= 0 {
+		t.Fatalf("expected positive TTL on buffer key, got %v", ttl)
+	}
+	if ttl > 24*time.Hour {
+		t.Fatalf("expected TTL <= 24h, got %v", ttl)
+	}
+}
+
 func TestNewRedisBackend_success(t *testing.T) {
 	mr := miniredis.RunT(t)
 	backend, err := NewRedisBackend("redis://"+mr.Addr(), 100)
diff --git a/broker.go b/broker.go
index 60b6bad..91e9cf9 100644
--- a/broker.go
+++ b/broker.go
@@ -85,6 +85,9 @@ func (b *Broker) Subscribe(path string, lastEventID string) (<-chan *Event, func
 				break
 			}
 		}
+		if len(b.subscribers[path]) == 0 {
+			delete(b.subscribers, path)
+		}
 		func() {
 			defer func() { recover() }()
 			close(ch)
diff --git a/broker_test.go b/broker_test.go
index 871f5f8..9af2da9 100644
--- a/broker_test.go
+++ b/broker_test.go
@@ -132,6 +132,29 @@ func TestBroker_unsubscribeStopsDelivery(t *testing.T) {
 	}
 }
 
+func TestBroker_unsubscribeCleansUpEmptyPath(t *testing.T) {
+	b, cancel := newTestBroker(100)
+	defer cancel()
+
+	_, unsub := b.Subscribe("ephemeral/path", "")
+
+	b.mu.RLock()
+	_, exists := b.subscribers["ephemeral/path"]
+	b.mu.RUnlock()
+	if !exists {
+		t.Fatal("expected subscriber map entry to exist")
+	}
+
+	unsub()
+
+	b.mu.RLock()
+	_, exists = b.subscribers["ephemeral/path"]
+	b.mu.RUnlock()
+	if exists {
+		t.Fatal("expected subscriber map entry to be deleted after last unsub")
+	}
+}
+
 func TestBroker_ringBufferWraps(t *testing.T) {
 	b, cancel := newTestBroker(5)
 	defer cancel()
-- 
2.51.2