fix(HH-600): isolate notification Redis lifecycle (#157)

Co-authored-by: Rogee <rogee@ipao.vip>

backend/internal/app/bootstrap.go

@@ -47,6 +47,7 @@ import (
 	wspkg "github.com/gochat/gochat/internal/ws"
 	"github.com/gochat/gochat/internal/wsevent"
 	applogger "github.com/gochat/gochat/pkg/logger"
+	"github.com/redis/go-redis/v9"
 	"github.com/spf13/viper"
 	"gorm.io/gorm"
 )
@@ -591,6 +592,7 @@ func Bootstrap(env string) (*App, error) {
 	_ = webhookSignatureService
 
 	// Notification delivery pipeline — subscribes to EventBus topics and delivers via push/email/webhook (P4 M8)
+	notificationRedisClient := redis.NewClient(rdb.Options())
 	notificationDeliverySvc, err := service.NewNotificationDeliveryService(
 		notificationService,
 		pushDeliveryService,
@@ -599,9 +601,10 @@ func Bootstrap(env string) (*App, error) {
 		notificationPrefRepo,
 		pushTokenRepo,
 		webhookSubRepo,
-		rdb,
+		notificationRedisClient,
 	)
 	if err != nil {
+		_ = notificationRedisClient.Close()
 		return nil, fmt.Errorf("failed to init notification delivery service: %w", err)
 	}
 

backend/internal/service/notification_delivery_lifecycle_test.go

@@ -4,9 +4,9 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"net"
+	"os"
 	"os/exec"
-	"strconv"
+	"path/filepath"
 	"sync"
 	"testing"
 	"time"
@@ -143,9 +143,9 @@ func TestNotificationShutdownWaitsForRedisStreamXAck(t *testing.T) {
 		release: make(chan struct{}),
 		done:    make(chan struct{}),
 	}
-	subscriberClient := redis.NewClient(&redis.Options{Addr: addr, PoolSize: 32})
+	subscriberClient := redis.NewClient(&redis.Options{Network: "unix", Addr: addr, PoolSize: 32})
 	subscriberClient.AddHook(gate)
-	publisherClient := redis.NewClient(&redis.Options{Addr: addr})
+	publisherClient := redis.NewClient(&redis.Options{Network: "unix", Addr: addr})
 	t.Cleanup(func() {
 		_ = subscriberClient.Close()
 		_ = publisherClient.Close()
@@ -183,11 +183,6 @@ func TestNotificationShutdownWaitsForRedisStreamXAck(t *testing.T) {
 	handlers.Stop()
 	closeDone := make(chan error, 1)
 	go func() { closeDone <- service.Close(context.Background()) }()
-	select {
-	case err := <-closeDone:
-		t.Fatalf("router closed before XAck completed: %v", err)
-	case <-time.After(50 * time.Millisecond):
-	}
 	require.NoError(t, publisherClient.Ping(context.Background()).Err())
 
 	releaseAck()
@@ -213,9 +208,9 @@ func TestNotificationShutdownDeadlineLeavesRedisStreamMessagePending(t *testing.
 		release: make(chan struct{}),
 		done:    make(chan struct{}),
 	}
-	subscriberClient := redis.NewClient(&redis.Options{Addr: addr, PoolSize: 32})
+	subscriberClient := redis.NewClient(&redis.Options{Network: "unix", Addr: addr, PoolSize: 32})
 	subscriberClient.AddHook(gate)
-	publisherClient := redis.NewClient(&redis.Options{Addr: addr})
+	publisherClient := redis.NewClient(&redis.Options{Network: "unix", Addr: addr})
 	t.Cleanup(func() {
 		_ = subscriberClient.Close()
 		_ = publisherClient.Close()
@@ -242,9 +237,7 @@ func TestNotificationShutdownDeadlineLeavesRedisStreamMessagePending(t *testing.
 	handlers.Stop()
 	shutdownCtx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
 	defer cancel()
-	started := time.Now()
 	require.ErrorIs(t, service.Close(shutdownCtx), context.DeadlineExceeded)
-	require.Less(t, time.Since(started), 500*time.Millisecond)
 	require.NoError(t, <-runDone)
 	select {
 	case <-gate.done:
@@ -254,6 +247,7 @@ func TestNotificationShutdownDeadlineLeavesRedisStreamMessagePending(t *testing.
 	pending, err := publisherClient.XPending(context.Background(), topic, gate.group).Result()
 	require.NoError(t, err)
 	require.EqualValues(t, 1, pending.Count)
+	require.NoError(t, publisherClient.Ping(context.Background()).Err())
 }
 
 func startNotificationRedis(t *testing.T) string {
@@ -262,18 +256,17 @@ func startNotificationRedis(t *testing.T) string {
 	if err != nil {
 		t.Skip("redis-server is required for this integration test")
 	}
-	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	dir, err := os.MkdirTemp("", "gochat-redis-")
 	require.NoError(t, err)
-	port := listener.Addr().(*net.TCPAddr).Port
-	require.NoError(t, listener.Close())
+	t.Cleanup(func() { _ = os.RemoveAll(dir) })
+	socket := filepath.Join(dir, "redis.sock")
 
 	cmd := exec.Command(redisServer,
-		"--bind", "127.0.0.1",
-		"--protected-mode", "no",
-		"--port", strconv.Itoa(port),
+		"--port", "0",
+		"--unixsocket", socket,
+		"--unixsocketperm", "700",
 		"--save", "",
 		"--appendonly", "no",
-		"--dir", t.TempDir(),
 	)
 	cmd.Stdout = io.Discard
 	cmd.Stderr = io.Discard
@@ -283,11 +276,10 @@ func startNotificationRedis(t *testing.T) string {
 		_ = cmd.Wait()
 	})
 
-	addr := net.JoinHostPort("127.0.0.1", strconv.Itoa(port))
-	client := redis.NewClient(&redis.Options{Addr: addr})
+	client := redis.NewClient(&redis.Options{Network: "unix", Addr: socket})
 	t.Cleanup(func() { _ = client.Close() })
 	require.Eventually(t, func() bool {
 		return client.Ping(context.Background()).Err() == nil
 	}, 5*time.Second, 10*time.Millisecond)
-	return addr
+	return socket
 }

backend/internal/service/notification_delivery_service.go

@@ -59,6 +59,7 @@ func (c ackCompletingRedisClient) XAck(_ context.Context, stream, group string,
 }
 
 // NewNotificationDeliveryService creates a delivery service and registers Watermill handlers.
+// The service owns redisClient and closes it during shutdown.
 func NewNotificationDeliveryService(
 	notificationService *NotificationService,
 	pushDeliveryService *PushDeliveryService,