diff --git a/api/pkg/events/listeners.go b/api/pkg/events/listeners.go index d5285f75d..e73f1d957 100644 --- a/api/pkg/events/listeners.go +++ b/api/pkg/events/listeners.go @@ -6,5 +6,10 @@ import ( cloudevents "github.com/cloudevents/sdk-go/v2" ) +type contextKey string + +// QueueIDKey identifies the queue task delivering an event in its context. +const QueueIDKey contextKey = "queue.task.id" + // EventListener is the type for processing events type EventListener func(ctx context.Context, event cloudevents.Event) error diff --git a/api/pkg/handlers/events_handler.go b/api/pkg/handlers/events_handler.go index 5e856cc70..14fbfa71c 100644 --- a/api/pkg/handlers/events_handler.go +++ b/api/pkg/handlers/events_handler.go @@ -1,10 +1,12 @@ package handlers import ( + "context" "fmt" cloudevents "github.com/cloudevents/sdk-go/v2" + "github.com/NdoleStudio/httpsms/pkg/events" "github.com/NdoleStudio/httpsms/pkg/services" "github.com/NdoleStudio/httpsms/pkg/telemetry" "github.com/NdoleStudio/stacktrace" @@ -63,6 +65,8 @@ func (h *EventsHandler) Dispatch(c fiber.Ctx) error { return h.responseForbidden(c) } + ctx = h.withQueueTaskID(ctx, c.Get("X-CloudTasks-TaskName")) + ctxLogger.Info(fmt.Sprintf("handling [%s] event with ID [%s]", request.Type(), request.ID())) err := h.service.DispatchSync(ctx, request) if err != nil { @@ -72,3 +76,11 @@ func (h *EventsHandler) Dispatch(c fiber.Ctx) error { return h.responseNoContent(c, "event dispatched successfully") } + +func (h *EventsHandler) withQueueTaskID(ctx context.Context, taskName string) context.Context { + queueID := "" + if taskName != "" { + queueID = fmt.Sprintf("%s/tasks/%s", h.queueConfig.Name, taskName) + } + return context.WithValue(ctx, events.QueueIDKey, queueID) +} diff --git a/api/pkg/handlers/events_handler_test.go b/api/pkg/handlers/events_handler_test.go new file mode 100644 index 000000000..36d93cfa9 --- /dev/null +++ b/api/pkg/handlers/events_handler_test.go @@ -0,0 +1,119 @@ +package handlers + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "testing" + + "github.com/NdoleStudio/httpsms/pkg/entities" + "github.com/NdoleStudio/httpsms/pkg/events" + "github.com/NdoleStudio/httpsms/pkg/middlewares" + "github.com/NdoleStudio/httpsms/pkg/services" + "github.com/NdoleStudio/httpsms/pkg/telemetry" + cloudevents "github.com/cloudevents/sdk-go/v2" + "github.com/gofiber/fiber/v3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/metric/noop" +) + +func TestEventsHandler_UsesDeliveredTaskQueueID(t *testing.T) { + for _, test := range []struct { + name string + taskID string + queueID string + }{ + {name: "cloud task", taskID: "task-123", queueID: "projects/test/locations/us/queues/heartbeats/tasks/task-123"}, + {name: "missing task header"}, + } { + t.Run(test.name, func(t *testing.T) { + logger := &messageThreadHandlerNoopLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + meter, err := noop.NewMeterProvider().Meter("test").Float64Histogram("events") + require.NoError(t, err) + config := services.PushQueueConfig{ + Name: "projects/test/locations/us/queues/heartbeats", UserID: "queue-user", + } + + dispatcher := services.NewEventDispatcher(logger, tracer, meter, nil, config) + var deliveredQueueID string + dispatcher.Subscribe(events.EventTypePhoneHeartbeatCheck, func(ctx context.Context, _ cloudevents.Event) error { + deliveredQueueID, _ = ctx.Value(events.QueueIDKey).(string) + return nil + }) + handler := NewEventsHandler(logger, tracer, config, dispatcher) + app := fiber.New() + app.Use(func(c fiber.Ctx) error { + c.Locals(middlewares.ContextKeyAuthUserID, entities.AuthContext{ID: config.UserID, Email: "queue@example.com"}) + return c.Next() + }) + handler.RegisterRoutes(app) + event := cloudevents.NewEvent() + event.SetID("event-123") + event.SetSource("https://example.com") + event.SetType(events.EventTypePhoneHeartbeatCheck) + event.SetExtension("queueid", "untrusted-body-queue-id") + require.NoError(t, event.SetData(cloudevents.ApplicationJSON, events.PhoneHeartbeatCheckPayload{})) + body, err := json.Marshal(event) + require.NoError(t, err) + request := httptest.NewRequest(http.MethodPost, "/v1/events", bytes.NewReader(body)) + request.Header.Set("Content-Type", "application/json") + request.Header.Set("X-CloudTasks-TaskName", test.taskID) + + response, err := app.Test(request) + require.NoError(t, err) + defer response.Body.Close() + require.Equal(t, http.StatusNoContent, response.StatusCode) + assert.Equal(t, test.queueID, deliveredQueueID) + }) + } +} + +func TestEventsHandler_AcknowledgesFailedListeners(t *testing.T) { + for _, eventType := range []string{events.EventTypePhoneHeartbeatCheck, events.EventTypePhoneUpdated} { + t.Run(eventType, func(t *testing.T) { + logger := &messageThreadHandlerNoopLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + meter, err := noop.NewMeterProvider().Meter("test").Float64Histogram("events") + require.NoError(t, err) + config := services.PushQueueConfig{Name: "projects/test/locations/us/queues/heartbeats", UserID: "queue-user"} + dispatcher := services.NewEventDispatcher(logger, tracer, meter, nil, config) + attempts := 0 + dispatcher.Subscribe(eventType, func(context.Context, cloudevents.Event) error { + attempts++ + if attempts == 1 { + return errors.New("database unavailable") + } + return nil + }) + app := fiber.New() + app.Use(func(c fiber.Ctx) error { + c.Locals(middlewares.ContextKeyAuthUserID, entities.AuthContext{ID: config.UserID, Email: "queue@example.com"}) + return c.Next() + }) + NewEventsHandler(logger, tracer, config, dispatcher).RegisterRoutes(app) + event := cloudevents.NewEvent() + event.SetID("event-123") + event.SetSource("https://example.com") + event.SetType(eventType) + body, err := json.Marshal(event) + require.NoError(t, err) + + statuses := []int{http.StatusNoContent, http.StatusNoContent} + for _, status := range statuses { + request := httptest.NewRequest(http.MethodPost, "/v1/events", bytes.NewReader(body)) + request.Header.Set("Content-Type", "application/json") + request.Header.Set("X-CloudTasks-TaskName", "task-123") + response, err := app.Test(request) + require.NoError(t, err) + require.NoError(t, response.Body.Close()) + assert.Equal(t, status, response.StatusCode) + } + assert.Equal(t, 2, attempts) + }) + } +} diff --git a/api/pkg/listeners/heartbeat_listener.go b/api/pkg/listeners/heartbeat_listener.go index 4e0cdc0d7..29235bac0 100644 --- a/api/pkg/listeners/heartbeat_listener.go +++ b/api/pkg/listeners/heartbeat_listener.go @@ -92,12 +92,14 @@ func (listener *HeartbeatListener) onPhoneHeartbeatCheck(ctx context.Context, ev return listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot decode [%s] into [%T]", event.Data(), payload)) } + queueID, _ := ctx.Value(events.QueueIDKey).(string) monitorParams := &services.HeartbeatMonitorParams{ Owner: payload.Owner, PhoneID: payload.PhoneID, MonitorID: payload.MonitorID, UserID: payload.UserID, Source: event.Source(), + QueueID: queueID, } if err := listener.service.Monitor(ctx, monitorParams); err != nil { diff --git a/api/pkg/listeners/heartbeat_listener_test.go b/api/pkg/listeners/heartbeat_listener_test.go new file mode 100644 index 000000000..c68f7cc03 --- /dev/null +++ b/api/pkg/listeners/heartbeat_listener_test.go @@ -0,0 +1,150 @@ +package listeners + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + "github.com/NdoleStudio/httpsms/pkg/entities" + "github.com/NdoleStudio/httpsms/pkg/events" + "github.com/NdoleStudio/httpsms/pkg/repositories" + "github.com/NdoleStudio/httpsms/pkg/services" + "github.com/NdoleStudio/httpsms/pkg/telemetry" + cloudevents "github.com/cloudevents/sdk-go/v2" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type heartbeatListenerRepository struct { + repositories.HeartbeatRepository + lastCalls int +} + +func (repository *heartbeatListenerRepository) Last(context.Context, entities.UserID, string) (*entities.Heartbeat, error) { + repository.lastCalls++ + return &entities.Heartbeat{Timestamp: time.Now().UTC()}, nil +} + +type heartbeatListenerMonitorRepository struct { + repositories.HeartbeatMonitorRepository + monitor *entities.HeartbeatMonitor + err error +} + +func (repository *heartbeatListenerMonitorRepository) Load(context.Context, entities.UserID, string) (*entities.HeartbeatMonitor, error) { + return repository.monitor, repository.err +} + +func TestHeartbeatListener_ReschedulesWhenMonitorCannotBeLoaded(t *testing.T) { + logger := &noopListenerLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + repository := &heartbeatListenerRepository{} + queue := &heartbeatListenerQueue{} + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", Owner: "+18005550199", QueueID: "old-check", + } + service := services.NewHeartbeatService(logger, tracer, repository, + &heartbeatListenerMonitorRepository{ + monitor: monitor, err: errors.New("database unavailable"), + }, + services.NewEventDispatcher(logger, tracer, nil, queue, services.PushQueueConfig{})) + _, routes := NewHeartbeatListener(logger, tracer, service) + event := cloudevents.NewEvent() + event.SetID("event-123") + event.SetSource("https://example.com") + event.SetType(events.EventTypePhoneHeartbeatCheck) + ctx := context.WithValue(context.Background(), events.QueueIDKey, "old-check") + require.NoError(t, event.SetData(cloudevents.ApplicationJSON, events.PhoneHeartbeatCheckPayload{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, PhoneID: monitor.PhoneID, + })) + + err := routes[event.Type()](ctx, event) + + require.NoError(t, err) + assert.Zero(t, repository.lastCalls) + require.Len(t, queue.events, 1) + assert.Equal(t, events.EventTypePhoneHeartbeatCheck, queue.events[0].Type()) +} + +func (repository *heartbeatListenerMonitorRepository) UpdateQueueID(_ context.Context, _ uuid.UUID, queueID string) error { + repository.monitor.QueueID = queueID + return nil +} + +type heartbeatListenerQueue struct { + events []cloudevents.Event +} + +func (queue *heartbeatListenerQueue) Enqueue(_ context.Context, task *services.PushQueueTask, _ time.Duration) (string, error) { + var event cloudevents.Event + if err := json.Unmarshal(task.Body, &event); err != nil { + return "", err + } + queue.events = append(queue.events, event) + return "next-check", nil +} + +func TestHeartbeatListener_ChecksOnlyCurrentQueueTask(t *testing.T) { + for _, test := range []struct { + name string + storedQueue string + eventQueue string + wantChecks int + loadErr error + }{ + {name: "current task", storedQueue: "current-check", eventQueue: "current-check", wantChecks: 1}, + {name: "superseded task", storedQueue: "current-check", eventQueue: "old-check"}, + {name: "missing task ID", storedQueue: "current-check"}, + {name: "missing task ID and empty stored ID"}, + {name: "missing task ID and unavailable monitor", storedQueue: "current-check", loadErr: errors.New("database unavailable")}, + {name: "inactive monitor", eventQueue: "old-check"}, + } { + t.Run(test.name, func(t *testing.T) { + logger := &noopListenerLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", Owner: "+18005550199", + QueueID: test.storedQueue, PhoneOnline: true, + } + repository := &heartbeatListenerRepository{} + queue := &heartbeatListenerQueue{} + service := services.NewHeartbeatService(logger, tracer, repository, + &heartbeatListenerMonitorRepository{monitor: monitor, err: test.loadErr}, + services.NewEventDispatcher(logger, tracer, nil, queue, services.PushQueueConfig{})) + _, routes := NewHeartbeatListener(logger, tracer, service) + + event := cloudevents.NewEvent() + event.SetID(uuid.NewString()) + event.SetSource("https://example.com") + event.SetType(events.EventTypePhoneHeartbeatCheck) + ctx := context.Background() + if test.eventQueue != "" { + ctx = context.WithValue(ctx, events.QueueIDKey, test.eventQueue) + } + require.NoError(t, event.SetData(cloudevents.ApplicationJSON, events.PhoneHeartbeatCheckPayload{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: uuid.New(), PhoneID: uuid.New(), + })) + + require.NoError(t, routes[event.Type()](ctx, event)) + assert.Equal(t, test.wantChecks, repository.lastCalls) + require.Len(t, queue.events, test.wantChecks) + if test.wantChecks == 0 { + assert.Equal(t, test.storedQueue, monitor.QueueID) + return + } + assert.Equal(t, "next-check", monitor.QueueID) + var payload events.PhoneHeartbeatCheckPayload + require.NoError(t, queue.events[0].DataAs(&payload)) + assert.Equal(t, monitor.ID, payload.MonitorID) + assert.Equal(t, monitor.PhoneID, payload.PhoneID) + + // Redelivery of the previous task must not start another stream. + require.NoError(t, routes[event.Type()](ctx, event)) + assert.Equal(t, 1, repository.lastCalls) + assert.Len(t, queue.events, 1) + }) + } +} diff --git a/api/pkg/services/emulator_push_queue.go b/api/pkg/services/emulator_push_queue.go index 634689f1e..59a38f812 100644 --- a/api/pkg/services/emulator_push_queue.go +++ b/api/pkg/services/emulator_push_queue.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "net/http" + "path" "time" "github.com/NdoleStudio/stacktrace" @@ -40,7 +41,7 @@ func (queue *emulatorPushQueue) Enqueue(ctx context.Context, task *PushQueueTask ctx, span, ctxLogger := queue.tracer.StartWithLogger(ctx, queue.logger) defer span.End() - queueID = uuid.New().String() + queueID = fmt.Sprintf("%s/tasks/%s", queue.config.Name, uuid.New().String()) time.AfterFunc(timeout, queue.push(*task, queueID)) @@ -72,6 +73,7 @@ func (queue *emulatorPushQueue) push(task PushQueueTask, queueID string) func() // add content type request.Header("Content-Type", "application/json") + request.Header("X-CloudTasks-TaskName", path.Base(queueID)) if err := request.Fetch(ctx); err != nil { queue.logger.Error(stacktrace.Propagatef(err, "cannot send http request to [%s] for queue task [%s]", task.URL, queueID)) diff --git a/api/pkg/services/event_dispatcher_service.go b/api/pkg/services/event_dispatcher_service.go index 245ab0b2d..1b2e3478b 100644 --- a/api/pkg/services/event_dispatcher_service.go +++ b/api/pkg/services/event_dispatcher_service.go @@ -89,6 +89,7 @@ func (dispatcher *EventDispatcher) enqueue(ctx context.Context, event cloudevent if errors.Is(err, context.DeadlineExceeded) { ctxLogger.Warn(stacktrace.Propagatef(err, "cannot enqueue event with ID [%s] and type [%s] to [%T]", event.ID(), event.Type(), dispatcher.queue)) queueID, err = fmt.Sprintf("local-%s", event.ID()), nil + ctx = context.WithValue(ctx, events.QueueIDKey, queueID) time.AfterFunc(timeout, func() { dispatcher.Publish(ctx, event) }) diff --git a/api/pkg/services/event_dispatcher_service_test.go b/api/pkg/services/event_dispatcher_service_test.go new file mode 100644 index 000000000..8702ad7d4 --- /dev/null +++ b/api/pkg/services/event_dispatcher_service_test.go @@ -0,0 +1,73 @@ +package services + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/NdoleStudio/httpsms/pkg/events" + "github.com/NdoleStudio/httpsms/pkg/telemetry" + cloudevents "github.com/cloudevents/sdk-go/v2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/metric/noop" +) + +type deadlineEventQueue struct{} + +func (queue *deadlineEventQueue) Enqueue(context.Context, *PushQueueTask, time.Duration) (string, error) { + return "", context.DeadlineExceeded +} + +func TestEventDispatcher_LocalFallbackIncludesQueueID(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + meter, err := noop.NewMeterProvider().Meter("test").Float64Histogram("events") + require.NoError(t, err) + dispatcher := NewEventDispatcher(logger, tracer, meter, &deadlineEventQueue{}, PushQueueConfig{}) + delivered := make(chan string, 1) + dispatcher.Subscribe(events.EventTypePhoneHeartbeatCheck, func(ctx context.Context, _ cloudevents.Event) error { + queueID, _ := ctx.Value(events.QueueIDKey).(string) + delivered <- queueID + return nil + }) + event := cloudevents.NewEvent() + event.SetID("event-123") + event.SetSource("https://example.com") + event.SetType(events.EventTypePhoneHeartbeatCheck) + + queueID, err := dispatcher.DispatchWithTimeout(context.Background(), event, time.Millisecond) + require.NoError(t, err) + select { + case received := <-delivered: + assert.Equal(t, queueID, received) + case <-time.After(time.Second): + t.Fatal("local fallback did not deliver the event") + } +} + +func TestEmulatorPushQueue_DeliversTaskNameMatchingQueueID(t *testing.T) { + delivered := make(chan string, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + delivered <- request.Header.Get("X-CloudTasks-TaskName") + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + logger := &phoneNotificationLogger{} + queue := EmulatorPushQueue(logger, telemetry.NewOtelLogger("test", logger), server.Client(), + PushQueueConfig{Name: "projects/test/locations/us/queues/heartbeats"}) + + queueID, err := queue.Enqueue(context.Background(), &PushQueueTask{ + Method: http.MethodPost, URL: server.URL, Body: []byte(`{}`), + }, time.Millisecond) + require.NoError(t, err) + select { + case taskName := <-delivered: + require.NotEmpty(t, taskName) + assert.Equal(t, "projects/test/locations/us/queues/heartbeats/tasks/"+taskName, queueID) + case <-time.After(time.Second): + t.Fatal("emulator did not deliver the task") + } +} diff --git a/api/pkg/services/heartbeat_service.go b/api/pkg/services/heartbeat_service.go index da578549b..059897061 100644 --- a/api/pkg/services/heartbeat_service.go +++ b/api/pkg/services/heartbeat_service.go @@ -283,6 +283,7 @@ type HeartbeatMonitorParams struct { PhoneID uuid.UUID UserID entities.UserID Source string + QueueID string } // Monitor the heartbeats of an owner and phone number @@ -290,16 +291,26 @@ func (service *HeartbeatService) Monitor(ctx context.Context, params *HeartbeatM ctx, span, ctxLogger := service.tracer.StartWithLogger(ctx, service.logger) defer span.End() + if params.QueueID == "" { + ctxLogger.Info(fmt.Sprintf("ignoring heartbeat check without a queue ID for owner [%s] and user [%s]", params.Owner, params.UserID)) + return nil + } + monitor, err := service.monitorRepository.Load(ctx, params.UserID, params.Owner) if err != nil && stacktrace.GetCode(err) == repositories.ErrCodeNotFound { ctxLogger.Info(fmt.Sprintf("heartbeat monitor does not exist for owner [%s] and user [%s]", params.Owner, params.UserID)) return nil } if err != nil { - ctxLogger.Error(stacktrace.Propagatef(err, "cannot check if monitor exists with userID [%s] and owner [%s]", params.UserID, params.Owner)) + ctxLogger.Error(service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot check if monitor exists with userID [%s] and owner [%s]", params.UserID, params.Owner))) return service.scheduleHeartbeatCheck(ctx, time.Now().UTC(), params) } + if params.QueueID != monitor.QueueID { + ctxLogger.Info(fmt.Sprintf("ignoring heartbeat check with queue ID [%s] for monitor [%s] with current queue ID [%s]", params.QueueID, monitor.ID, monitor.QueueID)) + return nil + } + // Update params in case of ID duplicate params.PhoneID = monitor.PhoneID params.MonitorID = monitor.ID @@ -310,19 +321,25 @@ func (service *HeartbeatService) Monitor(ctx context.Context, params *HeartbeatM return nil } if err != nil { - ctxLogger.Error(stacktrace.Propagatef(err, "cannot fetch last heartbeat for userID [%s] and owner [%s] and ID [%s] removing check", params.UserID, params.Owner, params.MonitorID)) + ctxLogger.Error(service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot fetch last heartbeat for userID [%s] and owner [%s] and ID [%s]", params.UserID, params.Owner, params.MonitorID))) return nil } - // send urgent FCM message if the last heartbeat is late - if time.Now().UTC().Sub(heartbeat.Timestamp) > heartbeatCheckInterval && time.Now().UTC().Sub(heartbeat.Timestamp) < (heartbeatCheckInterval*5) { + return service.checkLastHeartbeat(ctx, heartbeat, monitor, params) +} + +func (service *HeartbeatService) checkLastHeartbeat(ctx context.Context, heartbeat *entities.Heartbeat, monitor *entities.HeartbeatMonitor, params *HeartbeatMonitorParams) error { + ctx, span, ctxLogger := service.tracer.StartWithLogger(ctx, service.logger) + defer span.End() + + heartbeatAge := time.Now().UTC().Sub(heartbeat.Timestamp) + if heartbeatAge > heartbeatCheckInterval && heartbeatAge < heartbeatCheckInterval*5 { ctxLogger.Info(fmt.Sprintf("sending missed heartbeat notification for userID [%s] and owner [%s] and monitor ID [%s]", params.UserID, params.Owner, params.MonitorID)) service.handleMissedMonitor(ctx, heartbeat.Timestamp, params) - } - if time.Now().UTC().Sub(heartbeat.Timestamp) > (heartbeatCheckInterval*4) && - time.Now().UTC().Sub(heartbeat.Timestamp) < (heartbeatCheckInterval*5) && monitor.PhoneOnline { - return service.handleFailedMonitor(ctx, heartbeat.Timestamp, params) + if heartbeatAge > heartbeatCheckInterval*4 && monitor.PhoneOnline { + return service.handleFailedMonitor(ctx, heartbeat.Timestamp, params) + } } return service.scheduleHeartbeatCheck(ctx, heartbeat.Timestamp, params) @@ -400,7 +417,7 @@ func (service *HeartbeatService) scheduleHeartbeatCheck(ctx context.Context, las } if err = service.monitorRepository.UpdateQueueID(ctx, params.MonitorID, queueID); err != nil { - service.logger.Error(stacktrace.Propagatef(err, "cannot update monitor with id [%s] with queue with ID [%s]", params.MonitorID, queueID)) + ctxLogger.Error(service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot update monitor with id [%s] with queue with ID [%s]", params.MonitorID, queueID))) } ctxLogger.Info(fmt.Sprintf("heartbeat check scheduled for monitor with id [%s] and phone id [%s] and queue id [%s] for user [%s]", params.MonitorID, params.PhoneID, queueID, params.UserID)) diff --git a/api/pkg/services/heartbeat_service_test.go b/api/pkg/services/heartbeat_service_test.go index ff607240e..dcd150b4a 100644 --- a/api/pkg/services/heartbeat_service_test.go +++ b/api/pkg/services/heartbeat_service_test.go @@ -2,6 +2,7 @@ package services import ( "context" + "fmt" "testing" "time" @@ -19,10 +20,15 @@ type heartbeatServiceRepository struct { repositories.HeartbeatRepository heartbeat *entities.Heartbeat err error + onLast func() } func (repository *heartbeatServiceRepository) Last(context.Context, entities.UserID, string) (*entities.Heartbeat, error) { - return repository.heartbeat, repository.err + heartbeat, err := repository.heartbeat, repository.err + if repository.onLast != nil { + repository.onLast() + } + return heartbeat, err } func (repository *heartbeatServiceRepository) Store(_ context.Context, heartbeat *entities.Heartbeat) error { @@ -33,8 +39,9 @@ func (repository *heartbeatServiceRepository) Store(_ context.Context, heartbeat type heartbeatServiceMonitorRepository struct { repositories.HeartbeatMonitorRepository - monitor *entities.HeartbeatMonitor - err error + monitor *entities.HeartbeatMonitor + err error + updateQueueErr error } func (repository *heartbeatServiceMonitorRepository) Load(context.Context, entities.UserID, string) (*entities.HeartbeatMonitor, error) { @@ -42,6 +49,9 @@ func (repository *heartbeatServiceMonitorRepository) Load(context.Context, entit } func (repository *heartbeatServiceMonitorRepository) UpdateQueueID(_ context.Context, _ uuid.UUID, queueID string) error { + if repository.updateQueueErr != nil { + return repository.updateQueueErr + } repository.monitor.QueueID = queueID repository.monitor.UpdatedAt = time.Now().UTC() return nil @@ -60,7 +70,7 @@ func (queue *heartbeatServiceQueue) Enqueue(ctx context.Context, task *PushQueue if _, err := queue.phoneNotificationEventQueue.Enqueue(ctx, task, timeout); err != nil { return "", err } - return "queue-1", nil + return fmt.Sprintf("queue-%d", len(queue.events)), nil } func TestHeartbeatService_RestartsChecksAfterHistoryExpires(t *testing.T) { @@ -76,16 +86,16 @@ func TestHeartbeatService_RestartsChecksAfterHistoryExpires(t *testing.T) { UserID: entities.UserID("user-1"), Owner: "+18005550199", QueueID: "expired-check", - UpdatedAt: time.Now().UTC(), + UpdatedAt: time.Now().UTC().Add(-61 * 24 * time.Hour), } monitorRepository := &heartbeatServiceMonitorRepository{monitor: monitor} service := NewHeartbeatService(logger, tracer, repository, monitorRepository, NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) - params := &HeartbeatMonitorParams{UserID: monitor.UserID, Owner: monitor.Owner, Source: "https://example.com"} + params := &HeartbeatMonitorParams{UserID: monitor.UserID, Owner: monitor.Owner, Source: "https://example.com", QueueID: monitor.QueueID} require.NoError(t, service.Monitor(context.Background(), params)) assert.Empty(t, queue.events) - assert.Empty(t, monitor.QueueID) + assert.Equal(t, "expired-check", monitor.QueueID) heartbeat, err := service.Store(context.Background(), HeartbeatStoreParams{ UserID: monitor.UserID, Owner: monitor.Owner, Source: params.Source, Timestamp: time.Now().UTC(), @@ -105,6 +115,7 @@ func TestHeartbeatService_RestartsChecksAfterHistoryExpires(t *testing.T) { assert.WithinDuration(t, heartbeat.Timestamp.Add(heartbeatCheckInterval), payload.ScheduledAt, time.Millisecond) repository.heartbeat.Timestamp = time.Now().UTC().Add(-4*heartbeatCheckInterval - time.Minute) + params.QueueID = monitor.QueueID require.NoError(t, service.Monitor(context.Background(), params)) require.Len(t, queue.events, 5) assert.Equal(t, events.EventTypePhoneHeartbeatOffline, queue.events[4].Type()) @@ -114,10 +125,12 @@ func TestHeartbeatServiceStore_OnlyRestartsInactiveChecks(t *testing.T) { for _, test := range []struct { name string updatedAt time.Time + queueID string wantCheck bool }{ - {name: "active monitor", updatedAt: time.Now().UTC()}, - {name: "stale monitor", updatedAt: time.Now().UTC().Add(-3 * time.Hour), wantCheck: true}, + {name: "active monitor", updatedAt: time.Now().UTC(), queueID: "previous-check"}, + {name: "stale monitor", updatedAt: time.Now().UTC().Add(-3 * time.Hour), queueID: "previous-check", wantCheck: true}, + {name: "recent monitor without task ID", updatedAt: time.Now().UTC()}, } { t.Run(test.name, func(t *testing.T) { logger := &phoneNotificationLogger{} @@ -125,7 +138,7 @@ func TestHeartbeatServiceStore_OnlyRestartsInactiveChecks(t *testing.T) { queue := &heartbeatServiceQueue{} monitor := &entities.HeartbeatMonitor{ ID: uuid.New(), PhoneID: uuid.New(), UserID: entities.UserID("user-1"), - Owner: "+18005550199", QueueID: "previous-check", UpdatedAt: test.updatedAt, PhoneOnline: true, + Owner: "+18005550199", QueueID: test.queueID, UpdatedAt: test.updatedAt, PhoneOnline: true, } service := NewHeartbeatService(logger, tracer, &heartbeatServiceRepository{}, &heartbeatServiceMonitorRepository{monitor: monitor}, @@ -156,7 +169,104 @@ func TestHeartbeatServiceMonitor_StopsCheckingDeletedMonitor(t *testing.T) { NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) require.NoError(t, service.Monitor(context.Background(), &HeartbeatMonitorParams{ - UserID: entities.UserID("user-1"), Owner: "+18005550199", Source: "https://example.com", + UserID: entities.UserID("user-1"), Owner: "+18005550199", Source: "https://example.com", QueueID: "current-check", })) assert.Empty(t, queue.events) } + +func TestHeartbeatServiceMonitor_StopsOnHeartbeatReadFailure(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue := &heartbeatServiceQueue{} + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), UserID: "user-1", Owner: "+18005550199", QueueID: "current-check", + } + service := NewHeartbeatService(logger, tracer, &heartbeatServiceRepository{ + err: stacktrace.NewError("heartbeat read failed"), + }, &heartbeatServiceMonitorRepository{monitor: monitor}, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + + err := service.Monitor(context.Background(), &HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, QueueID: monitor.QueueID, + }) + + require.NoError(t, err) + assert.Equal(t, "current-check", monitor.QueueID) + assert.Empty(t, queue.events) +} + +func TestHeartbeatServiceMonitor_RecoversFailedQueueIDSavesOnHeartbeat(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue := &heartbeatServiceQueue{} + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", Owner: "+18005550199", + QueueID: "current-check", PhoneOnline: true, UpdatedAt: time.Now().UTC().Add(-3 * time.Hour), + } + monitorRepository := &heartbeatServiceMonitorRepository{ + monitor: monitor, updateQueueErr: stacktrace.NewError("queue update failed"), + } + service := NewHeartbeatService(logger, tracer, &heartbeatServiceRepository{ + heartbeat: &entities.Heartbeat{Timestamp: time.Now().UTC()}, + }, monitorRepository, NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + params := &HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, QueueID: monitor.QueueID, Source: "https://example.com", + } + + require.NoError(t, service.Monitor(context.Background(), params)) + assert.Equal(t, "current-check", monitor.QueueID) + require.Len(t, queue.events, 1) + + monitorRepository.updateQueueErr = nil + _, err := service.Store(context.Background(), HeartbeatStoreParams{ + UserID: monitor.UserID, Owner: monitor.Owner, Source: params.Source, Timestamp: time.Now().UTC(), + }) + require.NoError(t, err) + assert.Equal(t, "queue-2", monitor.QueueID) + require.Len(t, queue.events, 2) + + params.QueueID = "queue-1" + require.NoError(t, service.Monitor(context.Background(), params)) + assert.Equal(t, "queue-2", monitor.QueueID) + assert.Len(t, queue.events, 2) + + params.QueueID = "queue-2" + require.NoError(t, service.Monitor(context.Background(), params)) + assert.Equal(t, "queue-3", monitor.QueueID) + assert.Len(t, queue.events, 3) +} + +func TestHeartbeatServiceMonitor_PreservesConcurrentReplacement(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue := &heartbeatServiceQueue{} + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", Owner: "+18005550199", + QueueID: "old-check", UpdatedAt: time.Now().UTC().Add(-61 * 24 * time.Hour), PhoneOnline: true, + } + repository := &heartbeatServiceRepository{ + err: stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "heartbeat not found"), + } + service := NewHeartbeatService(logger, tracer, repository, + &heartbeatServiceMonitorRepository{monitor: monitor}, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + repository.onLast = func() { + repository.onLast = nil + _, err := service.Store(context.Background(), HeartbeatStoreParams{ + UserID: monitor.UserID, Owner: monitor.Owner, Source: "https://example.com", Timestamp: time.Now().UTC(), + }) + require.NoError(t, err) + } + params := &HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, QueueID: "old-check", Source: "https://example.com", + } + + require.NoError(t, service.Monitor(context.Background(), params)) + assert.Equal(t, "queue-1", monitor.QueueID) + require.Len(t, queue.events, 1) + + params.QueueID = "queue-1" + require.NoError(t, service.Monitor(context.Background(), params)) + assert.Equal(t, "queue-2", monitor.QueueID) + assert.Len(t, queue.events, 2) +}