fix: deduplicate heartbeat checks by queue task

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

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)
+}

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)
+		})
+	}
+}

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 {

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)
+		})
+	}
+}

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))

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)
 		})