diff --git a/backend/internal/app/app.go b/backend/internal/app/app.go index b6d431ee..7ce19c41 100644 --- a/backend/internal/app/app.go +++ b/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 diff --git a/backend/internal/app/bootstrap.go b/backend/internal/app/bootstrap.go index cb525862..5874a7ed 100644 --- a/backend/internal/app/bootstrap.go +++ b/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, diff --git a/backend/internal/app/bootstrap_realtime_test.go b/backend/internal/app/bootstrap_realtime_test.go new file mode 100644 index 00000000..da651eae --- /dev/null +++ b/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 + } + } +} diff --git a/backend/internal/handler/api/v1/company_handler.go b/backend/internal/handler/api/v1/company_handler.go index 98f53d9f..be12d5ac 100644 --- a/backend/internal/handler/api/v1/company_handler.go +++ b/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. diff --git a/backend/internal/handler/api/v1/contact_handler.go b/backend/internal/handler/api/v1/contact_handler.go index c84bb7e3..09e30d41 100644 --- a/backend/internal/handler/api/v1/contact_handler.go +++ b/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 diff --git a/backend/internal/handler/api/v1/helpers.go b/backend/internal/handler/api/v1/helpers.go index 510d9018..d6c04fd6 100644 --- a/backend/internal/handler/api/v1/helpers.go +++ b/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") diff --git a/backend/internal/handler/api/v1/notification_handler.go b/backend/internal/handler/api/v1/notification_handler.go index 635e2aee..49c6ecc3 100644 --- a/backend/internal/handler/api/v1/notification_handler.go +++ b/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, diff --git a/backend/internal/handler/api/v1/realtime_publish_test.go b/backend/internal/handler/api/v1/realtime_publish_test.go new file mode 100644 index 00000000..6062e33d --- /dev/null +++ b/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") +} diff --git a/backend/internal/handler/ws/ws_test.go b/backend/internal/handler/ws/ws_test.go index 56f234c6..c89ce51f 100644 --- a/backend/internal/handler/ws/ws_test.go +++ b/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: diff --git a/backend/internal/router/router.go b/backend/internal/router/router.go index a8127399..8524e547 100644 --- a/backend/internal/router/router.go +++ b/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) diff --git a/backend/internal/router/router_test.go b/backend/internal/router/router_test.go index 957ac582..139b30a9 100644 --- a/backend/internal/router/router_test.go +++ b/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", diff --git a/backend/internal/worker/worker.go b/backend/internal/worker/worker.go index 4cf915af..f192dea6 100644 --- a/backend/internal/worker/worker.go +++ b/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 } diff --git a/backend/internal/ws/event_publisher_pg_test.go b/backend/internal/ws/event_publisher_pg_test.go new file mode 100644 index 00000000..6e850a95 --- /dev/null +++ b/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") +} diff --git a/backend/internal/ws/event_publisher_test.go b/backend/internal/ws/event_publisher_test.go index 5bbb7f54..02db3066 100644 --- a/backend/internal/ws/event_publisher_test.go +++ b/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) +}