HH-564: harden durable realtime publish boundaries (#139)

* fix(HH-564): harden durable realtime enqueue

* fix(HH-564): wire production SSE stream

---------

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

backend/internal/app/app.go

@@ -39,6 +39,7 @@ type App struct {
 	engine                  *gin.Engine
 	wsHub                   *ws.Hub
 	wsRelay                 *wspkg.BroadcastRelay
+	eventPublisher          *wspkg.EventPublisher
 	notificationDeliverySvc *service.NotificationDeliveryService
 	workerPool              *worker.WorkerPool
 	ready                   *atomic.Bool

backend/internal/app/bootstrap.go

@@ -779,13 +779,14 @@ func Bootstrap(env string) (*App, error) {
 	// Must be created before handlers so the hubTypingAdapter can reference it.
 	wsHub := ws.NewHubSimple()
 	wsRelay := wspkg.NewBroadcastRelay(rdb, wsHub)
+	sseRegistry := wspkg.NewSSERegistry()
 	agentService.WithDeactivation(refreshStore, func(userID uint) {
 		wsHub.DisconnectUser(userID)
 		if err := wsRelay.PublishUserDisconnect(context.Background(), userID); err != nil {
 			applogger.L().Errorf("failed to broadcast user disconnect for user_id=%d: %v", userID, err)
 		}
 	})
-	eventPublisher := wspkg.NewEventPublisher(wsHub, nil, wsRelay)
+	eventPublisher := wspkg.NewEventPublisher(wsHub, sseRegistry, wsRelay)
 	eventPublisher.SetWorkerPool(workerPool)
 	presenceTracker := wspkg.NewPresenceTracker(rdb, wsRelay)
 
@@ -849,6 +850,7 @@ func Bootstrap(env string) (*App, error) {
 		Message:           v1.NewMessageHandler(messageService),
 		Profile:           v1.NewProfileHandler(profileService, uploadService),
 		Notification:      v1.NewNotificationHandler(notificationService).WithEventPublisher(eventPublisher),
+		SSEEvent:          v1.NewSSEEventHandler(sseRegistry),
 		PlatformApp:       v1.NewPlatformAppHandler(platformAppService),
 		Team:              v1.NewTeamHandler(teamService),
 		CaptainAssistant:  v1.NewCaptainAssistantHandler(captainAssistantService),
@@ -1039,6 +1041,7 @@ func Bootstrap(env string) (*App, error) {
 		engine:                  engine,
 		wsHub:                   wsHub,
 		wsRelay:                 wsRelay,
+		eventPublisher:          eventPublisher,
 		notificationDeliverySvc: notificationDeliverySvc,
 		workerPool:              workerPool,
 		ready:                   ready,

backend/internal/app/bootstrap_realtime_test.go

@@ -0,0 +1,226 @@
+package app
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"os"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/alicebob/miniredis/v2"
+	miniredisserver "github.com/alicebob/miniredis/v2/server"
+	"github.com/stretchr/testify/require"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
+
+	handlerws "github.com/gochat/gochat/internal/handler/ws"
+	"github.com/gochat/gochat/internal/model"
+	wspkg "github.com/gochat/gochat/internal/ws"
+)
+
+func TestBootstrapRoutesDurableRealtimeToHubAndSSEWithoutDuplicates(t *testing.T) {
+	if os.Getenv("GOCHAT_TEST_DB") == "sqlite" {
+		t.Skip("bootstrap production wiring requires PostgreSQL")
+	}
+	dsn := os.Getenv("GOCHAT_TEST_DB_URL")
+	if dsn == "" {
+		t.Skip("GOCHAT_TEST_DB_URL is not set")
+	}
+
+	db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+	require.NoError(t, err)
+	sqlDB, err := db.DB()
+	require.NoError(t, err)
+	require.NoError(t, sqlDB.Ping())
+	t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
+	schema := fmt.Sprintf("hh564_bootstrap_%d", time.Now().UnixNano())
+	require.NoError(t, db.Exec(fmt.Sprintf(`CREATE SCHEMA %q`, schema)).Error)
+	t.Cleanup(func() { require.NoError(t, db.Exec(fmt.Sprintf(`DROP SCHEMA %q CASCADE`, schema)).Error) })
+	databaseURL, err := url.Parse(dsn)
+	require.NoError(t, err)
+	query := databaseURL.Query()
+	query.Set("search_path", schema+",public")
+	databaseURL.RawQuery = query.Encode()
+
+	mini := miniredis.RunT(t)
+	t.Setenv("GOCHAT_DATABASE_DSN", databaseURL.String())
+	t.Setenv("GOCHAT_DATABASE_RUN_MIGRATIONS", "true")
+	t.Setenv("GOCHAT_REDIS_DSN", "redis://"+mini.Addr()+"/0")
+	t.Setenv("GOCHAT_JWT_ALLOW_INSECURE_HEADER_AUTH", "true")
+	t.Setenv("GOCHAT_SEARCH_ENGINE", "db")
+	t.Setenv("GOCHAT_RATE_LIMIT_ENABLED", "false")
+	t.Setenv("GOCHAT_LOG_LEVEL", "error")
+	t.Chdir("../..")
+
+	application, err := Bootstrap("default")
+	require.NoError(t, err)
+	relayCtx, stopRelay := context.WithCancel(context.Background())
+	require.NoError(t, application.wsRelay.Start(relayCtx))
+	t.Cleanup(func() {
+		stopRelay()
+		if err := application.Shutdown(2 * time.Second); err != nil && !strings.Contains(err.Error(), "redis: client is closed") {
+			t.Errorf("shutdown bootstrap app: %v", err)
+		}
+	})
+
+	require.NoError(t, application.db.Exec("DELETE FROM background_jobs").Error)
+	stamp := time.Now().UnixNano()
+	account := model.Account{Name: fmt.Sprintf("bootstrap-sse-%d", stamp), Active: true}
+	require.NoError(t, application.db.Create(&account).Error)
+	user := model.User{AccountID: account.ID, Name: "Bootstrap SSE", Email: fmt.Sprintf("bootstrap-sse-%d@example.test", stamp), Password: "unused", Provider: "email", Active: true}
+	require.NoError(t, application.db.Create(&user).Error)
+	require.NoError(t, application.db.Create(&model.AccountUser{UserID: user.ID, AccountID: account.ID, Role: "administrator"}).Error)
+
+	dashboard := handlerws.NewClient(user.ID, account.ID, nil, application.wsHub)
+	dashboard.Identifier = fmt.Sprintf(`{"channel":"AccountChannel","account_id":%d}`, account.ID)
+	application.wsHub.Register(dashboard)
+	visitorToken := fmt.Sprintf("bootstrap-visitor-%d", stamp)
+	visitor := handlerws.NewClient(0, account.ID, nil, application.wsHub)
+	visitor.IsContact = true
+	visitor.PubsubToken = visitorToken
+	visitor.Identifier = fmt.Sprintf(`{"channel":"RoomChannel","pubsub_token":%q}`, visitorToken)
+	application.wsHub.Register(visitor)
+	t.Cleanup(func() {
+		application.wsHub.Unregister(dashboard)
+		application.wsHub.Unregister(visitor)
+	})
+
+	server := httptest.NewServer(application.Handler())
+	t.Cleanup(server.Close)
+	requestCtx, cancelRequest := context.WithCancel(context.Background())
+	t.Cleanup(cancelRequest)
+	req, err := http.NewRequestWithContext(requestCtx, http.MethodGet, fmt.Sprintf("%s/api/v1/accounts/%d/events", server.URL, account.ID), nil)
+	require.NoError(t, err)
+	req.Header.Set("X-User-ID", fmt.Sprint(user.ID))
+	req.Header.Set("X-Account-ID", fmt.Sprint(account.ID))
+	resp, err := server.Client().Do(req)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, resp.Body.Close()) })
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+	require.Equal(t, "text/event-stream", resp.Header.Get("Content-Type"))
+	sse := bufio.NewReader(resp.Body)
+	eventType, _ := readBootstrapSSE(t, sse)
+	require.Equal(t, "connected", eventType)
+
+	tokenChannel := wspkg.RedisPrefixRoom + "pubsub_token_" + visitorToken
+	var tokenFailed atomic.Bool
+	mini.Server().SetPreHook(func(peer *miniredisserver.Peer, command string, args ...string) bool {
+		if strings.EqualFold(command, "publish") && len(args) > 0 && args[0] == tokenChannel && tokenFailed.CompareAndSwap(false, true) {
+			peer.WriteError("token room unavailable")
+			return true
+		}
+		return false
+	})
+
+	require.NoError(t, application.eventPublisher.PublishWidgetEvent(account.ID, visitorToken, wspkg.EventMessageCreated, map[string]any{"id": 7}))
+	processed, err := application.workerPool.ProcessOne(context.Background())
+	require.True(t, processed)
+	require.NoError(t, err)
+	requireHubEvent(t, dashboard.Send)
+	eventType, payload := readBootstrapSSE(t, sse)
+	require.Equal(t, wspkg.EventMessageCreated, eventType)
+	require.JSONEq(t, `{"id":7}`, payload)
+
+	processed, err = application.workerPool.ProcessOne(context.Background())
+	require.True(t, processed)
+	require.ErrorContains(t, err, "token room unavailable")
+	select {
+	case <-visitor.Send:
+		t.Fatal("token subscriber received failed delivery")
+	default:
+	}
+	require.NoError(t, application.db.Model(&model.BackgroundJob{}).
+		Where("status = ?", model.BackgroundJobStatusRetrying).
+		Update("scheduled_at", time.Now().Add(-time.Second)).Error)
+
+	processed, err = application.workerPool.ProcessOne(context.Background())
+	require.True(t, processed)
+	require.NoError(t, err)
+	requireHubEvent(t, visitor.Send)
+	require.True(t, tokenFailed.Load())
+	assertNoHubEvent(t, dashboard.Send)
+	assertNoHubEvent(t, visitor.Send)
+	assertNoBootstrapSSE(t, sse)
+}
+
+func readBootstrapSSE(t *testing.T, reader *bufio.Reader) (string, string) {
+	t.Helper()
+	resultCh := make(chan bootstrapSSEResult, 1)
+	go func() {
+		resultCh <- scanBootstrapSSE(reader)
+	}()
+	select {
+	case got := <-resultCh:
+		require.NoError(t, got.err)
+		return got.event, got.data
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for SSE event")
+		return "", ""
+	}
+}
+
+func requireHubEvent(t *testing.T, events <-chan []byte) {
+	t.Helper()
+	select {
+	case <-events:
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for Hub event")
+	}
+}
+
+func assertNoHubEvent(t *testing.T, events <-chan []byte) {
+	t.Helper()
+	time.Sleep(20 * time.Millisecond)
+	select {
+	case <-events:
+		t.Fatal("subscriber received duplicate event")
+	default:
+	}
+}
+
+func assertNoBootstrapSSE(t *testing.T, reader *bufio.Reader) {
+	t.Helper()
+	eventCh := make(chan bootstrapSSEResult, 1)
+	go func() {
+		eventCh <- scanBootstrapSSE(reader)
+	}()
+	select {
+	case event := <-eventCh:
+		require.NoError(t, event.err)
+		t.Fatalf("SSE subscriber received duplicate %s", event.event)
+	case <-time.After(50 * time.Millisecond):
+	}
+}
+
+type bootstrapSSEResult struct {
+	event string
+	data  string
+	err   error
+}
+
+func scanBootstrapSSE(reader *bufio.Reader) bootstrapSSEResult {
+	var got bootstrapSSEResult
+	for {
+		line, err := reader.ReadString('\n')
+		if err != nil {
+			got.err = err
+			return got
+		}
+		line = strings.TrimSpace(line)
+		switch {
+		case strings.HasPrefix(line, "event: "):
+			got.event = strings.TrimPrefix(line, "event: ")
+		case strings.HasPrefix(line, "data: "):
+			got.data = strings.TrimPrefix(line, "data: ")
+		case line == "":
+			return got
+		}
+	}
+}

backend/internal/handler/api/v1/company_handler.go

@@ -113,7 +113,7 @@ func (h *CompanyHandler) publishCompanyEvent(accountID uint, eventType string, c
 	if h.eventPublisher == nil || company == nil {
 		return
 	}
-	h.eventPublisher.PublishEvent(accountID, eventType, serializeCompany(context.Background(), h.svc.DB(), company))
+	publishRealtimeEvent(h.eventPublisher, accountID, eventType, serializeCompany(context.Background(), h.svc.DB(), company))
 }
 
 // Create creates a new company.

backend/internal/handler/api/v1/contact_handler.go

@@ -289,7 +289,7 @@ func (h *ContactHandler) publishContactEvent(accountID uint, eventType string, c
 	if h.eventPublisher == nil || contact == nil {
 		return
 	}
-	h.eventPublisher.PublishEvent(accountID, eventType, serializeCRMContact(context.Background(), h.svc.DB(), contact, true))
+	publishRealtimeEvent(h.eventPublisher, accountID, eventType, serializeCRMContact(context.Background(), h.svc.DB(), contact, true))
 }
 
 // @Summary Delete a contact

backend/internal/handler/api/v1/helpers.go

@@ -11,9 +11,20 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/gochat/gochat/internal/middleware"
 	"github.com/gochat/gochat/internal/model"
+	"github.com/gochat/gochat/internal/ws"
+	applogger "github.com/gochat/gochat/pkg/logger"
 	"gorm.io/gorm"
 )
 
+func publishRealtimeEvent(publisher *ws.EventPublisher, accountID uint, eventType string, payload any) {
+	if publisher == nil {
+		return
+	}
+	if err := publisher.PublishEvent(accountID, eventType, payload); err != nil {
+		applogger.L().Errorf("publish realtime event %s for account %d: %v", eventType, accountID, err)
+	}
+}
+
 func bindJSONWrappedOrRaw(c *gin.Context, wrapperKey string, target any) error {
 	if c.Request.Body == nil {
 		return fmt.Errorf("empty request body")

backend/internal/handler/api/v1/notification_handler.go

@@ -320,7 +320,7 @@ func (h *NotificationHandler) publishNotificationEvent(ctx context.Context, acco
 		applogger.L().Warnf("notification event %s total count: %v", eventType, err)
 		return
 	}
-	h.eventPublisher.PublishEvent(accountID, eventType, gin.H{
+	publishRealtimeEvent(h.eventPublisher, accountID, eventType, gin.H{
 		"notification": h.serializeNotification(ctx, notification),
 		"unread_count": unreadCount,
 		"count":        total,

backend/internal/handler/api/v1/realtime_publish_test.go

@@ -0,0 +1,30 @@
+package v1
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/gochat/gochat/internal/worker"
+	"github.com/gochat/gochat/internal/ws"
+	applogger "github.com/gochat/gochat/pkg/logger"
+)
+
+func TestPublishRealtimeEventLogsEnqueueFailure(t *testing.T) {
+	logPath := filepath.Join(t.TempDir(), "realtime.log")
+	require.NoError(t, applogger.Init(applogger.Config{Level: "error", Format: "console", Output: logPath, ErrorOutput: logPath}))
+	t.Cleanup(func() {
+		require.NoError(t, applogger.Init(applogger.Config{Level: "info", Format: "console", Output: "stdout", ErrorOutput: "stderr"}))
+	})
+
+	publisher := ws.NewEventPublisherLocal(nil, nil)
+	publisher.SetWorkerPool(worker.NewWorkerPool(nil))
+	publishRealtimeEvent(publisher, 42, ws.EventContactUpdated, map[string]any{"id": 7})
+	applogger.Sync()
+
+	logs, err := os.ReadFile(logPath)
+	require.NoError(t, err)
+	require.Contains(t, string(logs), "publish realtime event contact.updated for account 42: worker database is required")
+}

backend/internal/handler/ws/ws_test.go

@@ -339,6 +339,8 @@ func TestDurableWidgetRetryDoesNotDuplicateRealHubSubscribers(t *testing.T) {
 	visitor.PubsubToken = "visitor"
 	visitor.Identifier = `{"channel":"RoomChannel","pubsub_token":"visitor"}`
 	hub.Register(visitor)
+	sse := wspkg.NewSSERegistry()
+	dashboardSSE := sse.Subscribe("dashboard", 1, 1)
 
 	relay := wspkg.NewBroadcastRelay(rdb, hub)
 	ctx, cancel := context.WithCancel(context.Background())
@@ -348,7 +350,7 @@ func TestDurableWidgetRetryDoesNotDuplicateRealHubSubscribers(t *testing.T) {
 		require.NoError(t, relay.Stop())
 	})
 	pool := worker.NewWorkerPoolWithOptions(db, worker.WithBackoff(func(int) time.Duration { return 0 }))
-	publisher := wspkg.NewEventPublisher(hub, nil, relay)
+	publisher := wspkg.NewEventPublisher(hub, sse, relay)
 	publisher.SetWorkerPool(pool)
 	require.NoError(t, publisher.PublishWidgetEvent(1, "visitor", wspkg.EventMessageCreated, map[string]any{"id": 7}))
 
@@ -360,6 +362,12 @@ func TestDurableWidgetRetryDoesNotDuplicateRealHubSubscribers(t *testing.T) {
 	case <-time.After(time.Second):
 		t.Fatal("account subscriber did not receive message.created")
 	}
+	select {
+	case event := <-dashboardSSE.Events:
+		require.Equal(t, wspkg.EventMessageCreated, event.Type)
+	case <-time.After(time.Second):
+		t.Fatal("SSE subscriber did not receive message.created")
+	}
 
 	processed, err = pool.ProcessOne(context.Background())
 	require.True(t, processed)
@@ -385,6 +393,11 @@ func TestDurableWidgetRetryDoesNotDuplicateRealHubSubscribers(t *testing.T) {
 	default:
 	}
 	select {
+	case <-dashboardSSE.Events:
+		t.Fatal("SSE subscriber received duplicate message.created")
+	default:
+	}
+	select {
 	case <-visitor.Send:
 		t.Fatal("token subscriber received duplicate message.created")
 	default:

backend/internal/router/router.go

@@ -46,6 +46,7 @@ type Handlers struct {
 	Message                  *v1.MessageHandler
 	Profile                  *v1.ProfileHandler
 	Notification             *v1.NotificationHandler
+	SSEEvent                 *v1.SSEEventHandler
 	PlatformApp              *v1.PlatformAppHandler
 	Team                     *v1.TeamHandler
 	CaptainAssistant         *v1.CaptainAssistantHandler
@@ -711,6 +712,7 @@ func registerV1Routes(g *gin.RouterGroup, h *Handlers) {
 
 		// Account settings (ref: Chatwoot accounts#update settings subset)
 		accounts.PUT("/:account_id/settings", middleware.SuperAdminOrAdministrator(), h.Account.UpdateSettings)
+		accounts.GET("/:account_id/events", h.SSEEvent.StreamEvents)
 
 		// Account file upload route (ref: Chatwoot api/v1/accounts/:account_id/upload)
 		accounts.POST("/:account_id/upload", h.Upload.Upload)

backend/internal/router/router_test.go

@@ -101,6 +101,7 @@ func TestRegisterRoutesBootsWithChatwootParityConflictGroups(t *testing.T) {
 		"GET /api/v2/accounts/:account_id/year_in_review",
 		"GET /api/v2/accounts/:account_id/live_reports/grouped_conversation_metrics",
 		"POST /api/v1/accounts",
+		"GET /api/v1/accounts/:account_id/events",
 		"GET /webhooks/twitter",
 		"POST /webhooks/twitter",
 		"POST /webhooks/telegram/:bot_token",

backend/internal/worker/worker.go

@@ -317,14 +317,20 @@ func (wp *WorkerPool) persistJob(ctx context.Context, db *gorm.DB, jobType strin
 		}
 	}
 
-	if err := db.WithContext(ctx).Create(job).Error; err != nil {
-		if job.IdempotencyKey != "" {
-			var existing model.BackgroundJob
-			if findErr := db.WithContext(ctx).Where("idempotency_key = ?", job.IdempotencyKey).First(&existing).Error; findErr == nil {
-				return &existing, false, nil
-			}
+	query := db.WithContext(ctx)
+	if job.IdempotencyKey != "" {
+		query = query.Clauses(clause.OnConflict{DoNothing: true})
+	}
+	result := query.Create(job)
+	if result.Error != nil {
+		return nil, false, result.Error
+	}
+	if job.IdempotencyKey != "" && result.RowsAffected == 0 {
+		var existing model.BackgroundJob
+		if err := db.WithContext(ctx).Where("idempotency_key = ?", job.IdempotencyKey).First(&existing).Error; err != nil {
+			return nil, false, err
 		}
-		return nil, false, err
+		return &existing, false, nil
 	}
 	return job, true, nil
 }

backend/internal/ws/event_publisher_pg_test.go

@@ -0,0 +1,97 @@
+package ws
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
+
+	"github.com/gochat/gochat/internal/model"
+	"github.com/gochat/gochat/internal/worker"
+)
+
+func TestEventPublisherConcurrentTransactionalEnqueueIsIdempotentPostgres(t *testing.T) {
+	if os.Getenv("GOCHAT_TEST_DB") == "sqlite" {
+		t.Skip("PostgreSQL-only concurrency regression")
+	}
+
+	dsn := os.Getenv("GOCHAT_TEST_DB_URL")
+	if dsn == "" {
+		dsn = "host=localhost port=5432 user=postgres password=postgres dbname=gochat_test sslmode=disable"
+	}
+	db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+	require.NoError(t, err)
+	sqlDB, err := db.DB()
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
+
+	schema := fmt.Sprintf("realtime_enqueue_%d", time.Now().UnixNano())
+	require.NoError(t, db.Exec(`CREATE SCHEMA "`+schema+`"`).Error)
+	t.Cleanup(func() { require.NoError(t, db.Exec(`DROP SCHEMA "`+schema+`" CASCADE`).Error) })
+	inSchema := func(tx *gorm.DB) error {
+		return tx.Exec(`SET LOCAL search_path TO "` + schema + `", public`).Error
+	}
+	require.NoError(t, db.Transaction(func(tx *gorm.DB) error {
+		if err := inSchema(tx); err != nil {
+			return err
+		}
+		if err := tx.AutoMigrate(&model.BackgroundJob{}); err != nil {
+			return err
+		}
+		return tx.Exec(`CREATE UNIQUE INDEX idx_background_jobs_idempotency_key_unique ON background_jobs(idempotency_key) WHERE idempotency_key IS NOT NULL AND idempotency_key <> ''`).Error
+	}))
+
+	pool := worker.NewWorkerPool(db)
+	publisher := NewEventPublisherLocal(nil, nil)
+	publisher.SetWorkerPool(pool)
+
+	const writers = 8
+	barrier := make(chan struct{})
+	var arrived atomic.Int32
+	require.NoError(t, db.Callback().Create().Before("gorm:create").Register("test:concurrent_realtime_enqueue", func(tx *gorm.DB) {
+		if tx.Statement.Schema != nil && tx.Statement.Schema.Table == "background_jobs" && arrived.Add(1) == writers {
+			close(barrier)
+		}
+		<-barrier
+	}))
+	start := make(chan struct{})
+	errs := make(chan error, writers)
+	var wg sync.WaitGroup
+	for range writers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			errs <- db.Transaction(func(tx *gorm.DB) error {
+				if err := inSchema(tx); err != nil {
+					return err
+				}
+				_, err := publisher.EnqueueInTransaction(context.Background(), tx, 1, "visitor", EventMessageCreated, map[string]any{"id": 7})
+				return err
+			})
+		}()
+	}
+	close(start)
+	wg.Wait()
+	close(errs)
+	for err := range errs {
+		require.NoError(t, err)
+	}
+
+	var count int64
+	require.NoError(t, db.Transaction(func(tx *gorm.DB) error {
+		if err := inSchema(tx); err != nil {
+			return err
+		}
+		return tx.Model(&model.BackgroundJob{}).Where("job_type = ?", taskTypeRealtimeEventPublish).Count(&count).Error
+	}))
+	require.Equal(t, int64(2), count, "one account and one token job must survive concurrent duplicate enqueue")
+}

backend/internal/ws/event_publisher_test.go

@@ -809,3 +809,24 @@ func TestEventPublisher_DurableWidgetPublishRetriesPartialFailure(t *testing.T)
 	require.Equal(t, 2, hook.attempts[tokenChannel])
 	require.Equal(t, 1, hook.succeeded[tokenChannel])
 }
+
+func TestEventPublisher_DurableMessageIDsHaveDistinctIdempotencyKeys(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
+	require.NoError(t, err)
+	require.NoError(t, db.AutoMigrate(&model.BackgroundJob{}))
+
+	pool := worker.NewWorkerPool(db)
+	publisher := NewEventPublisherLocal(nil, nil)
+	publisher.SetWorkerPool(pool)
+	require.NoError(t, publisher.PublishWidgetEvent(1, "visitor", EventMessageCreated, map[string]any{"id": 7}))
+	require.NoError(t, publisher.PublishWidgetEvent(1, "visitor", EventMessageCreated, map[string]any{"id": 8}))
+
+	var jobs []model.BackgroundJob
+	require.NoError(t, db.Where("job_type = ?", taskTypeRealtimeEventPublish).Order("id").Find(&jobs).Error)
+	require.Len(t, jobs, 4, "each message needs independent account and token jobs")
+	keys := make(map[string]struct{}, len(jobs))
+	for _, job := range jobs {
+		keys[job.IdempotencyKey] = struct{}{}
+	}
+	require.Len(t, keys, 4)
+}