From 0e074a259955bf128dd01d5c2cf9e8a1c6be9542 Mon Sep 17 00:00:00 2001
From: Acho Arnold Ewin <arnold@ndolestudio.com>
Date: Sat, 7 Nov 2026 14:24:19 +0200
Subject: [PATCH 1/7] fix: deduplicate heartbeat checks by queue task

Match delivered task IDs against each monitor's queue_id before processing.
Preserve emulator and local fallback delivery.
Restart stopped checks when heartbeat history resumes.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 742e0212-0866-4232-89d8-7d194134448a
---
 api/pkg/entities/heartbeat_monitor.go         |   2 +-
 api/pkg/events/listeners.go                   |   3 +
 api/pkg/handlers/events_handler.go            |   7 +
 api/pkg/handlers/events_handler_test.go       |  72 +++++++++
 api/pkg/listeners/heartbeat_listener.go       |   2 +
 api/pkg/listeners/heartbeat_listener_test.go  | 142 ++++++++++++++++++
 api/pkg/services/emulator_push_queue.go       |   4 +-
 api/pkg/services/event_dispatcher_service.go  |   1 +
 .../services/event_dispatcher_service_test.go |  72 +++++++++
 api/pkg/services/heartbeat_service.go         |  12 +-
 api/pkg/services/heartbeat_service_test.go    |  42 +++++-
 11 files changed, 349 insertions(+), 10 deletions(-)
 create mode 100644 api/pkg/handlers/events_handler_test.go
 create mode 100644 api/pkg/listeners/heartbeat_listener_test.go
 create mode 100644 api/pkg/services/event_dispatcher_service_test.go

diff --git a/api/pkg/entities/heartbeat_monitor.go b/api/pkg/entities/heartbeat_monitor.go
index 7151f1952..dc05b3b30 100644
--- a/api/pkg/entities/heartbeat_monitor.go
+++ b/api/pkg/entities/heartbeat_monitor.go
@@ -20,7 +20,7 @@ type HeartbeatMonitor struct {
 
 // RequiresCheck returns true if the heartbeat monitor requires a check
 func (h *HeartbeatMonitor) RequiresCheck() bool {
-	return h.UpdatedAt.Add(2 * time.Hour).Before(time.Now())
+	return h.QueueID == "" || h.UpdatedAt.Add(2*time.Hour).Before(time.Now())
 }
 
 // PhoneIsOffline returns true if the phone is offline
diff --git a/api/pkg/events/listeners.go b/api/pkg/events/listeners.go
index d5285f75d..7f6e3c817 100644
--- a/api/pkg/events/listeners.go
+++ b/api/pkg/events/listeners.go
@@ -6,5 +6,8 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 )
 
+// ExtensionQueueID identifies the queue task delivering an event.
+const ExtensionQueueID = "queueid"
+
 // 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..af674778e 100644
--- a/api/pkg/handlers/events_handler.go
+++ b/api/pkg/handlers/events_handler.go
@@ -5,6 +5,7 @@ import (
 
 	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 +64,12 @@ func (h *EventsHandler) Dispatch(c fiber.Ctx) error {
 		return h.responseForbidden(c)
 	}
 
+	queueID := ""
+	if taskName := c.Get("X-CloudTasks-TaskName"); taskName != "" {
+		queueID = fmt.Sprintf("%s/tasks/%s", h.queueConfig.Name, taskName)
+	}
+	request.SetExtension(events.ExtensionQueueID, queueID)
+
 	ctxLogger.Info(fmt.Sprintf("handling [%s] event with ID [%s]", request.Type(), request.ID()))
 	err := h.service.DispatchSync(ctx, request)
 	if err != nil {
diff --git a/api/pkg/handlers/events_handler_test.go b/api/pkg/handlers/events_handler_test.go
new file mode 100644
index 000000000..a06385ace
--- /dev/null
+++ b/api/pkg/handlers/events_handler_test.go
@@ -0,0 +1,72 @@
+package handlers
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"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 delivered cloudevents.Event
+			dispatcher.Subscribe(events.EventTypePhoneHeartbeatCheck, func(_ context.Context, event cloudevents.Event) error {
+				delivered = event
+				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, delivered.Extensions()["queueid"])
+		})
+	}
+}
diff --git a/api/pkg/listeners/heartbeat_listener.go b/api/pkg/listeners/heartbeat_listener.go
index 4e0cdc0d7..444d4eabe 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, _ := event.Extensions()[events.ExtensionQueueID].(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..6f9f12558
--- /dev/null
+++ b/api/pkg/listeners/heartbeat_listener_test.go
@@ -0,0 +1,142 @@
+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_DoesNotRescheduleWhenMonitorCannotBeLoaded(t *testing.T) {
+	logger := &noopListenerLogger{}
+	tracer := telemetry.NewOtelLogger("test", logger)
+	repository := &heartbeatListenerRepository{}
+	queue := &heartbeatListenerQueue{}
+	service := services.NewHeartbeatService(logger, tracer, repository,
+		&heartbeatListenerMonitorRepository{
+			monitor: &entities.HeartbeatMonitor{}, 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)
+	event.SetExtension("queueid", "old-check")
+	require.NoError(t, event.SetData(cloudevents.ApplicationJSON, events.PhoneHeartbeatCheckPayload{
+		UserID: "user-1", Owner: "+18005550199",
+	}))
+
+	err := routes[event.Type()](context.Background(), event)
+
+	require.ErrorContains(t, err, "database unavailable")
+	assert.Zero(t, repository.lastCalls)
+	assert.Empty(t, queue.events)
+}
+
+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
+	}{
+		{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: "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},
+				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)
+			if test.eventQueue != "" {
+				event.SetExtension("queueid", 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()](context.Background(), 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()](context.Background(), 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..7080dee9f 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
+		event.SetExtension(events.ExtensionQueueID, 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..3c48912f3
--- /dev/null
+++ b/api/pkg/services/event_dispatcher_service_test.go
@@ -0,0 +1,72 @@
+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 cloudevents.Event, 1)
+	dispatcher.Subscribe(events.EventTypePhoneHeartbeatCheck, func(_ context.Context, event cloudevents.Event) error {
+		delivered <- event
+		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.Extensions()["queueid"])
+	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..33ec5a4ae 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
@@ -296,8 +297,12 @@ func (service *HeartbeatService) Monitor(ctx context.Context, params *HeartbeatM
 		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))
-		return service.scheduleHeartbeatCheck(ctx, time.Now().UTC(), params)
+		return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot check if monitor exists with userID [%s] and owner [%s]", params.UserID, params.Owner))
+	}
+
+	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
@@ -306,6 +311,9 @@ func (service *HeartbeatService) Monitor(ctx context.Context, params *HeartbeatM
 
 	heartbeat, err := service.repository.Last(ctx, params.UserID, params.Owner)
 	if stacktrace.GetCode(err) == repositories.ErrCodeNotFound {
+		if err = service.monitorRepository.UpdateQueueID(ctx, monitor.ID, ""); err != nil {
+			return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot clear queue ID for stopped heartbeat monitor [%s]", monitor.ID))
+		}
 		ctxLogger.Info(fmt.Sprintf("heartbeat history does not exist for userID [%s] and owner [%s], stopping check", params.UserID, params.Owner))
 		return nil
 	}
diff --git a/api/pkg/services/heartbeat_service_test.go b/api/pkg/services/heartbeat_service_test.go
index ff607240e..6d2c4a09b 100644
--- a/api/pkg/services/heartbeat_service_test.go
+++ b/api/pkg/services/heartbeat_service_test.go
@@ -33,8 +33,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 +43,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
@@ -81,7 +85,7 @@ func TestHeartbeatService_RestartsChecksAfterHistoryExpires(t *testing.T) {
 	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)
@@ -105,6 +109,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 +119,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: "stopped monitor", updatedAt: time.Now().UTC(), wantCheck: true},
 	} {
 		t.Run(test.name, func(t *testing.T) {
 			logger := &phoneNotificationLogger{}
@@ -125,7 +132,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},
@@ -145,6 +152,29 @@ func TestHeartbeatServiceStore_OnlyRestartsInactiveChecks(t *testing.T) {
 	}
 }
 
+func TestHeartbeatServiceMonitor_ReportsFailureToClearStoppedCheck(t *testing.T) {
+	logger := &phoneNotificationLogger{}
+	tracer := telemetry.NewOtelLogger("test", logger)
+	queue := &heartbeatServiceQueue{}
+	monitor := &entities.HeartbeatMonitor{
+		ID: uuid.New(), UserID: "user-1", Owner: "+18005550199", QueueID: "expired-check",
+	}
+	monitorRepository := &heartbeatServiceMonitorRepository{
+		monitor: monitor, updateQueueErr: stacktrace.NewError("queue update failed"),
+	}
+	service := NewHeartbeatService(logger, tracer, &heartbeatServiceRepository{
+		err: stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "heartbeat not found"),
+	}, monitorRepository, NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{}))
+
+	err := service.Monitor(context.Background(), &HeartbeatMonitorParams{
+		UserID: monitor.UserID, Owner: monitor.Owner, QueueID: monitor.QueueID,
+	})
+
+	require.ErrorContains(t, err, "queue update failed")
+	assert.Equal(t, "expired-check", monitor.QueueID)
+	assert.Empty(t, queue.events)
+}
+
 func TestHeartbeatServiceMonitor_StopsCheckingDeletedMonitor(t *testing.T) {
 	logger := &phoneNotificationLogger{}
 	tracer := telemetry.NewOtelLogger("test", logger)

From 291aacaaa663e1fb1aa46ca7a22ab59b47fa62f1 Mon Sep 17 00:00:00 2001
From: Acho Arnold Ewin <arnold@ndolestudio.com>
Date: Sat, 7 Nov 2026 14:36:15 +0200
Subject: [PATCH 2/7] fix: preserve heartbeat checks across database failures

Return heartbeat listener errors to queue consumers for retry.
Require next-task ID persistence and conditionally clear observed task IDs.
Cover HTTP retries, failed writes, and overlapping heartbeat restarts.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 742e0212-0866-4232-89d8-7d194134448a
---
 api/pkg/handlers/events_handler_test.go       |  50 ++++++++
 .../gorm_heartbeat_monitor_clear_test.go      |  37 ++++++
 .../gorm_heartbeat_monitor_repository.go      |  22 ++++
 .../heartbeat_monitor_repository.go           |   3 +
 .../mongo_heartbeat_monitor_repository.go     |  22 ++++
 api/pkg/services/event_dispatcher_service.go  |  22 +++-
 api/pkg/services/heartbeat_service.go         |   7 +-
 api/pkg/services/heartbeat_service_test.go    | 111 +++++++++++++++++-
 8 files changed, 264 insertions(+), 10 deletions(-)
 create mode 100644 api/pkg/repositories/gorm_heartbeat_monitor_clear_test.go

diff --git a/api/pkg/handlers/events_handler_test.go b/api/pkg/handlers/events_handler_test.go
index a06385ace..3f6608c5f 100644
--- a/api/pkg/handlers/events_handler_test.go
+++ b/api/pkg/handlers/events_handler_test.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -37,6 +38,7 @@ func TestEventsHandler_UsesDeliveredTaskQueueID(t *testing.T) {
 			config := services.PushQueueConfig{
 				Name: "projects/test/locations/us/queues/heartbeats", UserID: "queue-user",
 			}
+
 			dispatcher := services.NewEventDispatcher(logger, tracer, meter, nil, config)
 			var delivered cloudevents.Event
 			dispatcher.Subscribe(events.EventTypePhoneHeartbeatCheck, func(_ context.Context, event cloudevents.Event) error {
@@ -70,3 +72,51 @@ func TestEventsHandler_UsesDeliveredTaskQueueID(t *testing.T) {
 		})
 	}
 }
+
+func TestEventsHandler_RetriesFailedHeartbeatChecks(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}
+			if eventType == events.EventTypePhoneHeartbeatCheck {
+				statuses[0] = http.StatusInternalServerError
+			}
+			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/repositories/gorm_heartbeat_monitor_clear_test.go b/api/pkg/repositories/gorm_heartbeat_monitor_clear_test.go
new file mode 100644
index 000000000..0c1470643
--- /dev/null
+++ b/api/pkg/repositories/gorm_heartbeat_monitor_clear_test.go
@@ -0,0 +1,37 @@
+package repositories
+
+import (
+	"context"
+	"testing"
+
+	"github.com/NdoleStudio/httpsms/pkg/telemetry"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+)
+
+func TestHeartbeatMonitorClearQueueID_MatchesObservedTask(t *testing.T) {
+	db, err := gorm.Open(postgres.New(postgres.Config{Conn: &messageThreadTestConnPool{}}), &gorm.Config{
+		DisableAutomaticPing: true, SkipDefaultTransaction: true, DryRun: true,
+	})
+	require.NoError(t, err)
+	var query string
+	var args []any
+	require.NoError(t, db.Callback().Update().After("gorm:update").Register("capture", func(db *gorm.DB) {
+		query = db.Statement.SQL.String()
+		args = db.Statement.Vars
+	}))
+	logger := &messageThreadTestLogger{}
+	repository := NewGormHeartbeatMonitorRepository(logger, telemetry.NewOtelLogger("test", logger), db)
+	monitorID := uuid.New()
+
+	require.NoError(t, repository.ClearQueueID(context.Background(), monitorID, "old-check"))
+
+	assert.Contains(t, query, "WHERE id = $3 AND queue_id = $4")
+	require.Len(t, args, 4)
+	assert.Equal(t, "", args[0])
+	assert.Equal(t, monitorID, args[2])
+	assert.Equal(t, "old-check", args[3])
+}
diff --git a/api/pkg/repositories/gorm_heartbeat_monitor_repository.go b/api/pkg/repositories/gorm_heartbeat_monitor_repository.go
index 85d1b2098..5c67dbe59 100644
--- a/api/pkg/repositories/gorm_heartbeat_monitor_repository.go
+++ b/api/pkg/repositories/gorm_heartbeat_monitor_repository.go
@@ -88,6 +88,28 @@ func (repository *gormHeartbeatMonitorRepository) UpdateQueueID(ctx context.Cont
 	return nil
 }
 
+// ClearQueueID clears the observed queue ID without erasing a replacement task.
+func (repository *gormHeartbeatMonitorRepository) ClearQueueID(ctx context.Context, monitorID uuid.UUID, expectedQueueID string) error {
+	ctx, span := repository.tracer.Start(ctx)
+	defer span.End()
+
+	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
+	defer cancel()
+
+	err := repository.db.WithContext(ctx).
+		Model(&entities.HeartbeatMonitor{}).
+		Where("id = ?", monitorID).
+		Where("queue_id = ?", expectedQueueID).
+		Updates(map[string]any{
+			"queue_id":   "",
+			"updated_at": time.Now().UTC(),
+		}).Error
+	if err != nil {
+		return repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot clear queue ID for heartbeat monitor [%s]", monitorID))
+	}
+	return nil
+}
+
 func (repository *gormHeartbeatMonitorRepository) Delete(ctx context.Context, userID entities.UserID, owner string) error {
 	ctx, span := repository.tracer.Start(ctx)
 	defer span.End()
diff --git a/api/pkg/repositories/heartbeat_monitor_repository.go b/api/pkg/repositories/heartbeat_monitor_repository.go
index ddcde94d8..d039302f8 100644
--- a/api/pkg/repositories/heartbeat_monitor_repository.go
+++ b/api/pkg/repositories/heartbeat_monitor_repository.go
@@ -22,6 +22,9 @@ type HeartbeatMonitorRepository interface {
 	// UpdateQueueID updates the queueID of a monitor
 	UpdateQueueID(ctx context.Context, monitorID uuid.UUID, queueID string) error
 
+	// ClearQueueID clears a queue ID only if it still matches the observed task.
+	ClearQueueID(ctx context.Context, monitorID uuid.UUID, expectedQueueID string) error
+
 	// Delete an entities.HeartbeatMonitor
 	Delete(ctx context.Context, userID entities.UserID, phoneNumber string) error
 
diff --git a/api/pkg/repositories/mongo_heartbeat_monitor_repository.go b/api/pkg/repositories/mongo_heartbeat_monitor_repository.go
index fea05e993..387e73e61 100644
--- a/api/pkg/repositories/mongo_heartbeat_monitor_repository.go
+++ b/api/pkg/repositories/mongo_heartbeat_monitor_repository.go
@@ -114,6 +114,28 @@ func (repository *mongoHeartbeatMonitorRepository) UpdateQueueID(ctx context.Con
 	return nil
 }
 
+func (repository *mongoHeartbeatMonitorRepository) ClearQueueID(ctx context.Context, monitorID uuid.UUID, expectedQueueID string) error {
+	ctx, span := repository.tracer.Start(ctx)
+	defer span.End()
+
+	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
+	defer cancel()
+
+	filter := bson.D{
+		{Key: "_id", Value: monitorID.String()},
+		{Key: "queue_id", Value: expectedQueueID},
+	}
+	update := bson.D{{Key: "$set", Value: bson.D{
+		{Key: "queue_id", Value: ""},
+		{Key: "updated_at", Value: time.Now().UTC()},
+	}}}
+	_, err := repository.collection.UpdateOne(ctx, filter, update)
+	if err != nil {
+		return repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot clear queue ID for heartbeat monitor [%s]", monitorID))
+	}
+	return nil
+}
+
 func (repository *mongoHeartbeatMonitorRepository) Delete(ctx context.Context, userID entities.UserID, phoneNumber string) error {
 	ctx, span := repository.tracer.Start(ctx)
 	defer span.End()
diff --git a/api/pkg/services/event_dispatcher_service.go b/api/pkg/services/event_dispatcher_service.go
index 7080dee9f..e2258b66e 100644
--- a/api/pkg/services/event_dispatcher_service.go
+++ b/api/pkg/services/event_dispatcher_service.go
@@ -56,7 +56,9 @@ func (dispatcher *EventDispatcher) DispatchSync(ctx context.Context, event cloud
 		return dispatcher.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot dispatch event with ID [%s] and type [%s] because it is invalid", event.ID(), event.Type()))
 	}
 
-	dispatcher.Publish(ctx, event)
+	if err := dispatcher.Publish(ctx, event); err != nil && event.Type() == events.EventTypePhoneHeartbeatCheck {
+		return dispatcher.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot dispatch heartbeat check with ID [%s]", event.ID()))
+	}
 	return nil
 }
 
@@ -115,7 +117,7 @@ func (dispatcher *EventDispatcher) Subscribe(eventType string, listener events.E
 }
 
 // Publish an event to subscribers
-func (dispatcher *EventDispatcher) Publish(ctx context.Context, event cloudevents.Event) {
+func (dispatcher *EventDispatcher) Publish(ctx context.Context, event cloudevents.Event) error {
 	ctx, span, ctxLogger := dispatcher.tracer.StartWithLogger(ctx, dispatcher.logger)
 	defer span.End()
 
@@ -126,21 +128,29 @@ func (dispatcher *EventDispatcher) Publish(ctx context.Context, event cloudevent
 	subscribers, ok := dispatcher.listeners[event.Type()]
 	if !ok {
 		ctxLogger.Info(fmt.Sprintf("no listener is configured for event type [%s] with id [%s]", event.Type(), event.ID()))
-		return
+		return nil
 	}
 
+	listenerErrors := make(chan error, len(subscribers))
 	var wg sync.WaitGroup
 	for _, sub := range subscribers {
 		wg.Add(1)
 		go func(ctx context.Context, sub events.EventListener) {
 			if err := sub(ctx, event); err != nil {
-				ctxLogger.Error(stacktrace.Propagatef(err, "subscriber [%T] cannot handle event [%s]", sub, event.Type()))
+				wrapped := stacktrace.Propagatef(err, "subscriber [%T] cannot handle event [%s]", sub, event.Type())
+				ctxLogger.Error(wrapped)
+				listenerErrors <- wrapped
 			}
 			wg.Done()
 		}(ctx, sub)
 	}
 
 	wg.Wait()
+	close(listenerErrors)
+	var failures []error
+	for err := range listenerErrors {
+		failures = append(failures, err)
+	}
 
 	dispatcher.meter.Record(
 		ctx,
@@ -150,6 +160,10 @@ func (dispatcher *EventDispatcher) Publish(ctx context.Context, event cloudevent
 			semconv.CloudeventsEventSpecVersion(event.SpecVersion()),
 		),
 	)
+	if len(failures) != 0 {
+		return dispatcher.tracer.WrapErrorSpan(span, stacktrace.Propagatef(errors.Join(failures...), "cannot publish event [%s] with ID [%s]", event.Type(), event.ID()))
+	}
+	return nil
 }
 
 func (dispatcher *EventDispatcher) addCloudEventAttributes(span trace.Span, event cloudevents.Event) {
diff --git a/api/pkg/services/heartbeat_service.go b/api/pkg/services/heartbeat_service.go
index 33ec5a4ae..b05073f3a 100644
--- a/api/pkg/services/heartbeat_service.go
+++ b/api/pkg/services/heartbeat_service.go
@@ -311,15 +311,14 @@ func (service *HeartbeatService) Monitor(ctx context.Context, params *HeartbeatM
 
 	heartbeat, err := service.repository.Last(ctx, params.UserID, params.Owner)
 	if stacktrace.GetCode(err) == repositories.ErrCodeNotFound {
-		if err = service.monitorRepository.UpdateQueueID(ctx, monitor.ID, ""); err != nil {
+		if err = service.monitorRepository.ClearQueueID(ctx, monitor.ID, params.QueueID); err != nil {
 			return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot clear queue ID for stopped heartbeat monitor [%s]", monitor.ID))
 		}
 		ctxLogger.Info(fmt.Sprintf("heartbeat history does not exist for userID [%s] and owner [%s], stopping check", params.UserID, params.Owner))
 		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))
-		return nil
+		return 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))
 	}
 
 	// send urgent FCM message if the last heartbeat is late
@@ -408,7 +407,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))
+		return 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 6d2c4a09b..56a210355 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 {
@@ -51,6 +57,13 @@ func (repository *heartbeatServiceMonitorRepository) UpdateQueueID(_ context.Con
 	return nil
 }
 
+func (repository *heartbeatServiceMonitorRepository) ClearQueueID(ctx context.Context, monitorID uuid.UUID, expectedQueueID string) error {
+	if repository.monitor.QueueID != expectedQueueID {
+		return nil
+	}
+	return repository.UpdateQueueID(ctx, monitorID, "")
+}
+
 func (repository *heartbeatServiceMonitorRepository) UpdatePhoneOnline(_ context.Context, _ entities.UserID, _ uuid.UUID, online bool) error {
 	repository.monitor.PhoneOnline = online
 	return nil
@@ -64,7 +77,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) {
@@ -190,3 +203,97 @@ func TestHeartbeatServiceMonitor_StopsCheckingDeletedMonitor(t *testing.T) {
 	}))
 	assert.Empty(t, queue.events)
 }
+
+func TestHeartbeatServiceMonitor_ReportsHeartbeatReadFailure(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.ErrorContains(t, err, "heartbeat read failed")
+	assert.Equal(t, "current-check", monitor.QueueID)
+	assert.Empty(t, queue.events)
+}
+
+func TestHeartbeatServiceMonitor_RetriesFailedQueueIDSaves(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,
+	}
+	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.ErrorContains(t, service.Monitor(context.Background(), params), "queue update failed")
+	assert.Equal(t, "current-check", monitor.QueueID)
+	require.Len(t, queue.events, 1)
+
+	monitorRepository.updateQueueErr = nil
+	require.NoError(t, service.Monitor(context.Background(), params))
+	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(-3 * 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)
+}

From 6e6da5b053eb959628b787a872dc894f579fba18 Mon Sep 17 00:00:00 2001
From: Acho Arnold Ewin <arnold@ndolestudio.com>
Date: Sat, 7 Nov 2026 14:42:00 +0200
Subject: [PATCH 3/7] fix: use queue.task.id for task context metadata

Carry queue task IDs in a typed context key named queue.task.id.
CloudEvents rejects dotted extension names, so keep this metadata internal.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 742e0212-0866-4232-89d8-7d194134448a
---
 api/pkg/events/listeners.go                       |  6 ++++--
 api/pkg/handlers/events_handler.go                |  3 ++-
 api/pkg/handlers/events_handler_test.go           |  8 ++++----
 api/pkg/listeners/heartbeat_listener.go           |  2 +-
 api/pkg/listeners/heartbeat_listener_test.go      | 11 ++++++-----
 api/pkg/services/event_dispatcher_service.go      |  2 +-
 api/pkg/services/event_dispatcher_service_test.go |  9 +++++----
 7 files changed, 23 insertions(+), 18 deletions(-)

diff --git a/api/pkg/events/listeners.go b/api/pkg/events/listeners.go
index 7f6e3c817..1ebc7f087 100644
--- a/api/pkg/events/listeners.go
+++ b/api/pkg/events/listeners.go
@@ -6,8 +6,10 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 )
 
-// ExtensionQueueID identifies the queue task delivering an event.
-const ExtensionQueueID = "queueid"
+type contextKey string
+
+// ExtensionQueueID identifies the queue task delivering an event in its context.
+const ExtensionQueueID 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 af674778e..f3b8329f5 100644
--- a/api/pkg/handlers/events_handler.go
+++ b/api/pkg/handlers/events_handler.go
@@ -1,6 +1,7 @@
 package handlers
 
 import (
+	"context"
 	"fmt"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
@@ -68,7 +69,7 @@ func (h *EventsHandler) Dispatch(c fiber.Ctx) error {
 	if taskName := c.Get("X-CloudTasks-TaskName"); taskName != "" {
 		queueID = fmt.Sprintf("%s/tasks/%s", h.queueConfig.Name, taskName)
 	}
-	request.SetExtension(events.ExtensionQueueID, queueID)
+	ctx = context.WithValue(ctx, events.ExtensionQueueID, queueID)
 
 	ctxLogger.Info(fmt.Sprintf("handling [%s] event with ID [%s]", request.Type(), request.ID()))
 	err := h.service.DispatchSync(ctx, request)
diff --git a/api/pkg/handlers/events_handler_test.go b/api/pkg/handlers/events_handler_test.go
index 3f6608c5f..5e1035969 100644
--- a/api/pkg/handlers/events_handler_test.go
+++ b/api/pkg/handlers/events_handler_test.go
@@ -40,9 +40,9 @@ func TestEventsHandler_UsesDeliveredTaskQueueID(t *testing.T) {
 			}
 
 			dispatcher := services.NewEventDispatcher(logger, tracer, meter, nil, config)
-			var delivered cloudevents.Event
-			dispatcher.Subscribe(events.EventTypePhoneHeartbeatCheck, func(_ context.Context, event cloudevents.Event) error {
-				delivered = event
+			var deliveredQueueID string
+			dispatcher.Subscribe(events.EventTypePhoneHeartbeatCheck, func(ctx context.Context, _ cloudevents.Event) error {
+				deliveredQueueID, _ = ctx.Value(events.ExtensionQueueID).(string)
 				return nil
 			})
 			handler := NewEventsHandler(logger, tracer, config, dispatcher)
@@ -68,7 +68,7 @@ func TestEventsHandler_UsesDeliveredTaskQueueID(t *testing.T) {
 			require.NoError(t, err)
 			defer response.Body.Close()
 			require.Equal(t, http.StatusNoContent, response.StatusCode)
-			assert.Equal(t, test.queueID, delivered.Extensions()["queueid"])
+			assert.Equal(t, test.queueID, deliveredQueueID)
 		})
 	}
 }
diff --git a/api/pkg/listeners/heartbeat_listener.go b/api/pkg/listeners/heartbeat_listener.go
index 444d4eabe..46a6ab1f9 100644
--- a/api/pkg/listeners/heartbeat_listener.go
+++ b/api/pkg/listeners/heartbeat_listener.go
@@ -92,7 +92,7 @@ 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, _ := event.Extensions()[events.ExtensionQueueID].(string)
+	queueID, _ := ctx.Value(events.ExtensionQueueID).(string)
 	monitorParams := &services.HeartbeatMonitorParams{
 		Owner:     payload.Owner,
 		PhoneID:   payload.PhoneID,
diff --git a/api/pkg/listeners/heartbeat_listener_test.go b/api/pkg/listeners/heartbeat_listener_test.go
index 6f9f12558..2aea19100 100644
--- a/api/pkg/listeners/heartbeat_listener_test.go
+++ b/api/pkg/listeners/heartbeat_listener_test.go
@@ -53,12 +53,12 @@ func TestHeartbeatListener_DoesNotRescheduleWhenMonitorCannotBeLoaded(t *testing
 	event.SetID("event-123")
 	event.SetSource("https://example.com")
 	event.SetType(events.EventTypePhoneHeartbeatCheck)
-	event.SetExtension("queueid", "old-check")
+	ctx := context.WithValue(context.Background(), events.ExtensionQueueID, "old-check")
 	require.NoError(t, event.SetData(cloudevents.ApplicationJSON, events.PhoneHeartbeatCheckPayload{
 		UserID: "user-1", Owner: "+18005550199",
 	}))
 
-	err := routes[event.Type()](context.Background(), event)
+	err := routes[event.Type()](ctx, event)
 
 	require.ErrorContains(t, err, "database unavailable")
 	assert.Zero(t, repository.lastCalls)
@@ -113,14 +113,15 @@ func TestHeartbeatListener_ChecksOnlyCurrentQueueTask(t *testing.T) {
 			event.SetID(uuid.NewString())
 			event.SetSource("https://example.com")
 			event.SetType(events.EventTypePhoneHeartbeatCheck)
+			ctx := context.Background()
 			if test.eventQueue != "" {
-				event.SetExtension("queueid", test.eventQueue)
+				ctx = context.WithValue(ctx, events.ExtensionQueueID, 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()](context.Background(), event))
+			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 {
@@ -134,7 +135,7 @@ func TestHeartbeatListener_ChecksOnlyCurrentQueueTask(t *testing.T) {
 			assert.Equal(t, monitor.PhoneID, payload.PhoneID)
 
 			// Redelivery of the previous task must not start another stream.
-			require.NoError(t, routes[event.Type()](context.Background(), event))
+			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/event_dispatcher_service.go b/api/pkg/services/event_dispatcher_service.go
index e2258b66e..59e27d7ea 100644
--- a/api/pkg/services/event_dispatcher_service.go
+++ b/api/pkg/services/event_dispatcher_service.go
@@ -91,7 +91,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
-		event.SetExtension(events.ExtensionQueueID, queueID)
+		ctx = context.WithValue(ctx, events.ExtensionQueueID, 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
index 3c48912f3..7074ce30c 100644
--- a/api/pkg/services/event_dispatcher_service_test.go
+++ b/api/pkg/services/event_dispatcher_service_test.go
@@ -27,9 +27,10 @@ func TestEventDispatcher_LocalFallbackIncludesQueueID(t *testing.T) {
 	meter, err := noop.NewMeterProvider().Meter("test").Float64Histogram("events")
 	require.NoError(t, err)
 	dispatcher := NewEventDispatcher(logger, tracer, meter, &deadlineEventQueue{}, PushQueueConfig{})
-	delivered := make(chan cloudevents.Event, 1)
-	dispatcher.Subscribe(events.EventTypePhoneHeartbeatCheck, func(_ context.Context, event cloudevents.Event) error {
-		delivered <- event
+	delivered := make(chan string, 1)
+	dispatcher.Subscribe(events.EventTypePhoneHeartbeatCheck, func(ctx context.Context, _ cloudevents.Event) error {
+		queueID, _ := ctx.Value(events.ExtensionQueueID).(string)
+		delivered <- queueID
 		return nil
 	})
 	event := cloudevents.NewEvent()
@@ -41,7 +42,7 @@ func TestEventDispatcher_LocalFallbackIncludesQueueID(t *testing.T) {
 	require.NoError(t, err)
 	select {
 	case received := <-delivered:
-		assert.Equal(t, queueID, received.Extensions()["queueid"])
+		assert.Equal(t, queueID, received)
 	case <-time.After(time.Second):
 		t.Fatal("local fallback did not deliver the event")
 	}

From 83e57837443897b1ab3933fdabf21243667cb549 Mon Sep 17 00:00:00 2001
From: Acho Arnold Ewin <arnold@ndolestudio.com>
Date: Sat, 7 Nov 2026 14:44:27 +0200
Subject: [PATCH 4/7] fix: preserve queue IDs when heartbeat history expires

Stop checks without clearing task IDs when heartbeat history is absent.
Restore timestamp-based RequiresCheck behavior and remove clearing helpers.
Model long-inactive monitors in retention restart tests.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 742e0212-0866-4232-89d8-7d194134448a
---
 api/pkg/entities/heartbeat_monitor.go         |  2 +-
 .../gorm_heartbeat_monitor_clear_test.go      | 37 ------------------
 .../gorm_heartbeat_monitor_repository.go      | 22 -----------
 .../heartbeat_monitor_repository.go           |  3 --
 .../mongo_heartbeat_monitor_repository.go     | 22 -----------
 api/pkg/services/heartbeat_service.go         |  3 --
 api/pkg/services/heartbeat_service_test.go    | 38 ++-----------------
 7 files changed, 5 insertions(+), 122 deletions(-)
 delete mode 100644 api/pkg/repositories/gorm_heartbeat_monitor_clear_test.go

diff --git a/api/pkg/entities/heartbeat_monitor.go b/api/pkg/entities/heartbeat_monitor.go
index dc05b3b30..7151f1952 100644
--- a/api/pkg/entities/heartbeat_monitor.go
+++ b/api/pkg/entities/heartbeat_monitor.go
@@ -20,7 +20,7 @@ type HeartbeatMonitor struct {
 
 // RequiresCheck returns true if the heartbeat monitor requires a check
 func (h *HeartbeatMonitor) RequiresCheck() bool {
-	return h.QueueID == "" || h.UpdatedAt.Add(2*time.Hour).Before(time.Now())
+	return h.UpdatedAt.Add(2 * time.Hour).Before(time.Now())
 }
 
 // PhoneIsOffline returns true if the phone is offline
diff --git a/api/pkg/repositories/gorm_heartbeat_monitor_clear_test.go b/api/pkg/repositories/gorm_heartbeat_monitor_clear_test.go
deleted file mode 100644
index 0c1470643..000000000
--- a/api/pkg/repositories/gorm_heartbeat_monitor_clear_test.go
+++ /dev/null
@@ -1,37 +0,0 @@
-package repositories
-
-import (
-	"context"
-	"testing"
-
-	"github.com/NdoleStudio/httpsms/pkg/telemetry"
-	"github.com/google/uuid"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"gorm.io/driver/postgres"
-	"gorm.io/gorm"
-)
-
-func TestHeartbeatMonitorClearQueueID_MatchesObservedTask(t *testing.T) {
-	db, err := gorm.Open(postgres.New(postgres.Config{Conn: &messageThreadTestConnPool{}}), &gorm.Config{
-		DisableAutomaticPing: true, SkipDefaultTransaction: true, DryRun: true,
-	})
-	require.NoError(t, err)
-	var query string
-	var args []any
-	require.NoError(t, db.Callback().Update().After("gorm:update").Register("capture", func(db *gorm.DB) {
-		query = db.Statement.SQL.String()
-		args = db.Statement.Vars
-	}))
-	logger := &messageThreadTestLogger{}
-	repository := NewGormHeartbeatMonitorRepository(logger, telemetry.NewOtelLogger("test", logger), db)
-	monitorID := uuid.New()
-
-	require.NoError(t, repository.ClearQueueID(context.Background(), monitorID, "old-check"))
-
-	assert.Contains(t, query, "WHERE id = $3 AND queue_id = $4")
-	require.Len(t, args, 4)
-	assert.Equal(t, "", args[0])
-	assert.Equal(t, monitorID, args[2])
-	assert.Equal(t, "old-check", args[3])
-}
diff --git a/api/pkg/repositories/gorm_heartbeat_monitor_repository.go b/api/pkg/repositories/gorm_heartbeat_monitor_repository.go
index 5c67dbe59..85d1b2098 100644
--- a/api/pkg/repositories/gorm_heartbeat_monitor_repository.go
+++ b/api/pkg/repositories/gorm_heartbeat_monitor_repository.go
@@ -88,28 +88,6 @@ func (repository *gormHeartbeatMonitorRepository) UpdateQueueID(ctx context.Cont
 	return nil
 }
 
-// ClearQueueID clears the observed queue ID without erasing a replacement task.
-func (repository *gormHeartbeatMonitorRepository) ClearQueueID(ctx context.Context, monitorID uuid.UUID, expectedQueueID string) error {
-	ctx, span := repository.tracer.Start(ctx)
-	defer span.End()
-
-	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
-	defer cancel()
-
-	err := repository.db.WithContext(ctx).
-		Model(&entities.HeartbeatMonitor{}).
-		Where("id = ?", monitorID).
-		Where("queue_id = ?", expectedQueueID).
-		Updates(map[string]any{
-			"queue_id":   "",
-			"updated_at": time.Now().UTC(),
-		}).Error
-	if err != nil {
-		return repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot clear queue ID for heartbeat monitor [%s]", monitorID))
-	}
-	return nil
-}
-
 func (repository *gormHeartbeatMonitorRepository) Delete(ctx context.Context, userID entities.UserID, owner string) error {
 	ctx, span := repository.tracer.Start(ctx)
 	defer span.End()
diff --git a/api/pkg/repositories/heartbeat_monitor_repository.go b/api/pkg/repositories/heartbeat_monitor_repository.go
index d039302f8..ddcde94d8 100644
--- a/api/pkg/repositories/heartbeat_monitor_repository.go
+++ b/api/pkg/repositories/heartbeat_monitor_repository.go
@@ -22,9 +22,6 @@ type HeartbeatMonitorRepository interface {
 	// UpdateQueueID updates the queueID of a monitor
 	UpdateQueueID(ctx context.Context, monitorID uuid.UUID, queueID string) error
 
-	// ClearQueueID clears a queue ID only if it still matches the observed task.
-	ClearQueueID(ctx context.Context, monitorID uuid.UUID, expectedQueueID string) error
-
 	// Delete an entities.HeartbeatMonitor
 	Delete(ctx context.Context, userID entities.UserID, phoneNumber string) error
 
diff --git a/api/pkg/repositories/mongo_heartbeat_monitor_repository.go b/api/pkg/repositories/mongo_heartbeat_monitor_repository.go
index 387e73e61..fea05e993 100644
--- a/api/pkg/repositories/mongo_heartbeat_monitor_repository.go
+++ b/api/pkg/repositories/mongo_heartbeat_monitor_repository.go
@@ -114,28 +114,6 @@ func (repository *mongoHeartbeatMonitorRepository) UpdateQueueID(ctx context.Con
 	return nil
 }
 
-func (repository *mongoHeartbeatMonitorRepository) ClearQueueID(ctx context.Context, monitorID uuid.UUID, expectedQueueID string) error {
-	ctx, span := repository.tracer.Start(ctx)
-	defer span.End()
-
-	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
-	defer cancel()
-
-	filter := bson.D{
-		{Key: "_id", Value: monitorID.String()},
-		{Key: "queue_id", Value: expectedQueueID},
-	}
-	update := bson.D{{Key: "$set", Value: bson.D{
-		{Key: "queue_id", Value: ""},
-		{Key: "updated_at", Value: time.Now().UTC()},
-	}}}
-	_, err := repository.collection.UpdateOne(ctx, filter, update)
-	if err != nil {
-		return repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot clear queue ID for heartbeat monitor [%s]", monitorID))
-	}
-	return nil
-}
-
 func (repository *mongoHeartbeatMonitorRepository) Delete(ctx context.Context, userID entities.UserID, phoneNumber string) error {
 	ctx, span := repository.tracer.Start(ctx)
 	defer span.End()
diff --git a/api/pkg/services/heartbeat_service.go b/api/pkg/services/heartbeat_service.go
index b05073f3a..8f5b73edd 100644
--- a/api/pkg/services/heartbeat_service.go
+++ b/api/pkg/services/heartbeat_service.go
@@ -311,9 +311,6 @@ func (service *HeartbeatService) Monitor(ctx context.Context, params *HeartbeatM
 
 	heartbeat, err := service.repository.Last(ctx, params.UserID, params.Owner)
 	if stacktrace.GetCode(err) == repositories.ErrCodeNotFound {
-		if err = service.monitorRepository.ClearQueueID(ctx, monitor.ID, params.QueueID); err != nil {
-			return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot clear queue ID for stopped heartbeat monitor [%s]", monitor.ID))
-		}
 		ctxLogger.Info(fmt.Sprintf("heartbeat history does not exist for userID [%s] and owner [%s], stopping check", params.UserID, params.Owner))
 		return nil
 	}
diff --git a/api/pkg/services/heartbeat_service_test.go b/api/pkg/services/heartbeat_service_test.go
index 56a210355..7fa84d278 100644
--- a/api/pkg/services/heartbeat_service_test.go
+++ b/api/pkg/services/heartbeat_service_test.go
@@ -57,13 +57,6 @@ func (repository *heartbeatServiceMonitorRepository) UpdateQueueID(_ context.Con
 	return nil
 }
 
-func (repository *heartbeatServiceMonitorRepository) ClearQueueID(ctx context.Context, monitorID uuid.UUID, expectedQueueID string) error {
-	if repository.monitor.QueueID != expectedQueueID {
-		return nil
-	}
-	return repository.UpdateQueueID(ctx, monitorID, "")
-}
-
 func (repository *heartbeatServiceMonitorRepository) UpdatePhoneOnline(_ context.Context, _ entities.UserID, _ uuid.UUID, online bool) error {
 	repository.monitor.PhoneOnline = online
 	return nil
@@ -93,7 +86,7 @@ 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,
@@ -102,7 +95,7 @@ func TestHeartbeatService_RestartsChecksAfterHistoryExpires(t *testing.T) {
 
 	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(),
@@ -137,7 +130,7 @@ func TestHeartbeatServiceStore_OnlyRestartsInactiveChecks(t *testing.T) {
 	}{
 		{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: "stopped monitor", updatedAt: time.Now().UTC(), wantCheck: true},
+		{name: "recent monitor without task ID", updatedAt: time.Now().UTC()},
 	} {
 		t.Run(test.name, func(t *testing.T) {
 			logger := &phoneNotificationLogger{}
@@ -165,29 +158,6 @@ func TestHeartbeatServiceStore_OnlyRestartsInactiveChecks(t *testing.T) {
 	}
 }
 
-func TestHeartbeatServiceMonitor_ReportsFailureToClearStoppedCheck(t *testing.T) {
-	logger := &phoneNotificationLogger{}
-	tracer := telemetry.NewOtelLogger("test", logger)
-	queue := &heartbeatServiceQueue{}
-	monitor := &entities.HeartbeatMonitor{
-		ID: uuid.New(), UserID: "user-1", Owner: "+18005550199", QueueID: "expired-check",
-	}
-	monitorRepository := &heartbeatServiceMonitorRepository{
-		monitor: monitor, updateQueueErr: stacktrace.NewError("queue update failed"),
-	}
-	service := NewHeartbeatService(logger, tracer, &heartbeatServiceRepository{
-		err: stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "heartbeat not found"),
-	}, monitorRepository, NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{}))
-
-	err := service.Monitor(context.Background(), &HeartbeatMonitorParams{
-		UserID: monitor.UserID, Owner: monitor.Owner, QueueID: monitor.QueueID,
-	})
-
-	require.ErrorContains(t, err, "queue update failed")
-	assert.Equal(t, "expired-check", monitor.QueueID)
-	assert.Empty(t, queue.events)
-}
-
 func TestHeartbeatServiceMonitor_StopsCheckingDeletedMonitor(t *testing.T) {
 	logger := &phoneNotificationLogger{}
 	tracer := telemetry.NewOtelLogger("test", logger)
@@ -269,7 +239,7 @@ func TestHeartbeatServiceMonitor_PreservesConcurrentReplacement(t *testing.T) {
 	queue := &heartbeatServiceQueue{}
 	monitor := &entities.HeartbeatMonitor{
 		ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", Owner: "+18005550199",
-		QueueID: "old-check", UpdatedAt: time.Now().UTC().Add(-3 * time.Hour), PhoneOnline: true,
+		QueueID: "old-check", UpdatedAt: time.Now().UTC().Add(-61 * 24 * time.Hour), PhoneOnline: true,
 	}
 	repository := &heartbeatServiceRepository{
 		err: stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "heartbeat not found"),

From 724b2093378529e0fa88ff6034e33d24fde7baff Mon Sep 17 00:00:00 2001
From: Acho Arnold Ewin <arnold@ndolestudio.com>
Date: Sat, 7 Nov 2026 14:47:37 +0200
Subject: [PATCH 5/7] refactor: simplify heartbeat monitor decision flow

Separate monitor lookup and deduplication from heartbeat timing decisions.
Calculate heartbeat age once and reuse the missed-heartbeat window.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 742e0212-0866-4232-89d8-7d194134448a
---
 api/pkg/services/heartbeat_service.go | 18 ++++++++++++------
 1 file changed, 12 insertions(+), 6 deletions(-)

diff --git a/api/pkg/services/heartbeat_service.go b/api/pkg/services/heartbeat_service.go
index 8f5b73edd..2d3e8a49c 100644
--- a/api/pkg/services/heartbeat_service.go
+++ b/api/pkg/services/heartbeat_service.go
@@ -318,15 +318,21 @@ func (service *HeartbeatService) Monitor(ctx context.Context, params *HeartbeatM
 		return 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))
 	}
 
-	// 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)

From 832853bf8f74d0e8798cafd34bfded1e65b8ff16 Mon Sep 17 00:00:00 2001
From: Acho Arnold Ewin <arnold@ndolestudio.com>
Date: Sat, 7 Nov 2026 15:09:29 +0200
Subject: [PATCH 6/7] fix: acknowledge failed heartbeat checks without
 replaying events

Extract task context handling and rename the typed queue task key.
Log heartbeat lookup and queue-ID save errors with best-effort recovery.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 742e0212-0866-4232-89d8-7d194134448a
---
 api/pkg/events/listeners.go                   |  4 ++--
 api/pkg/handlers/events_handler.go            | 14 +++++++----
 api/pkg/handlers/events_handler_test.go       |  7 ++----
 api/pkg/listeners/heartbeat_listener.go       |  2 +-
 api/pkg/listeners/heartbeat_listener_test.go  | 18 ++++++++------
 api/pkg/services/event_dispatcher_service.go  | 24 ++++---------------
 .../services/event_dispatcher_service_test.go |  2 +-
 api/pkg/services/heartbeat_service.go         |  8 ++++---
 api/pkg/services/heartbeat_service_test.go    | 15 +++++++-----
 9 files changed, 45 insertions(+), 49 deletions(-)

diff --git a/api/pkg/events/listeners.go b/api/pkg/events/listeners.go
index 1ebc7f087..e73f1d957 100644
--- a/api/pkg/events/listeners.go
+++ b/api/pkg/events/listeners.go
@@ -8,8 +8,8 @@ import (
 
 type contextKey string
 
-// ExtensionQueueID identifies the queue task delivering an event in its context.
-const ExtensionQueueID contextKey = "queue.task.id"
+// 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 f3b8329f5..14fbfa71c 100644
--- a/api/pkg/handlers/events_handler.go
+++ b/api/pkg/handlers/events_handler.go
@@ -65,11 +65,7 @@ func (h *EventsHandler) Dispatch(c fiber.Ctx) error {
 		return h.responseForbidden(c)
 	}
 
-	queueID := ""
-	if taskName := c.Get("X-CloudTasks-TaskName"); taskName != "" {
-		queueID = fmt.Sprintf("%s/tasks/%s", h.queueConfig.Name, taskName)
-	}
-	ctx = context.WithValue(ctx, events.ExtensionQueueID, queueID)
+	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)
@@ -80,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
index 5e1035969..36d93cfa9 100644
--- a/api/pkg/handlers/events_handler_test.go
+++ b/api/pkg/handlers/events_handler_test.go
@@ -42,7 +42,7 @@ func TestEventsHandler_UsesDeliveredTaskQueueID(t *testing.T) {
 			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.ExtensionQueueID).(string)
+				deliveredQueueID, _ = ctx.Value(events.QueueIDKey).(string)
 				return nil
 			})
 			handler := NewEventsHandler(logger, tracer, config, dispatcher)
@@ -73,7 +73,7 @@ func TestEventsHandler_UsesDeliveredTaskQueueID(t *testing.T) {
 	}
 }
 
-func TestEventsHandler_RetriesFailedHeartbeatChecks(t *testing.T) {
+func TestEventsHandler_AcknowledgesFailedListeners(t *testing.T) {
 	for _, eventType := range []string{events.EventTypePhoneHeartbeatCheck, events.EventTypePhoneUpdated} {
 		t.Run(eventType, func(t *testing.T) {
 			logger := &messageThreadHandlerNoopLogger{}
@@ -104,9 +104,6 @@ func TestEventsHandler_RetriesFailedHeartbeatChecks(t *testing.T) {
 			require.NoError(t, err)
 
 			statuses := []int{http.StatusNoContent, http.StatusNoContent}
-			if eventType == events.EventTypePhoneHeartbeatCheck {
-				statuses[0] = http.StatusInternalServerError
-			}
 			for _, status := range statuses {
 				request := httptest.NewRequest(http.MethodPost, "/v1/events", bytes.NewReader(body))
 				request.Header.Set("Content-Type", "application/json")
diff --git a/api/pkg/listeners/heartbeat_listener.go b/api/pkg/listeners/heartbeat_listener.go
index 46a6ab1f9..29235bac0 100644
--- a/api/pkg/listeners/heartbeat_listener.go
+++ b/api/pkg/listeners/heartbeat_listener.go
@@ -92,7 +92,7 @@ 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.ExtensionQueueID).(string)
+	queueID, _ := ctx.Value(events.QueueIDKey).(string)
 	monitorParams := &services.HeartbeatMonitorParams{
 		Owner:     payload.Owner,
 		PhoneID:   payload.PhoneID,
diff --git a/api/pkg/listeners/heartbeat_listener_test.go b/api/pkg/listeners/heartbeat_listener_test.go
index 2aea19100..574362282 100644
--- a/api/pkg/listeners/heartbeat_listener_test.go
+++ b/api/pkg/listeners/heartbeat_listener_test.go
@@ -38,14 +38,17 @@ func (repository *heartbeatListenerMonitorRepository) Load(context.Context, enti
 	return repository.monitor, repository.err
 }
 
-func TestHeartbeatListener_DoesNotRescheduleWhenMonitorCannotBeLoaded(t *testing.T) {
+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: &entities.HeartbeatMonitor{}, err: errors.New("database unavailable"),
+			monitor: monitor, err: errors.New("database unavailable"),
 		},
 		services.NewEventDispatcher(logger, tracer, nil, queue, services.PushQueueConfig{}))
 	_, routes := NewHeartbeatListener(logger, tracer, service)
@@ -53,16 +56,17 @@ func TestHeartbeatListener_DoesNotRescheduleWhenMonitorCannotBeLoaded(t *testing
 	event.SetID("event-123")
 	event.SetSource("https://example.com")
 	event.SetType(events.EventTypePhoneHeartbeatCheck)
-	ctx := context.WithValue(context.Background(), events.ExtensionQueueID, "old-check")
+	ctx := context.WithValue(context.Background(), events.QueueIDKey, "old-check")
 	require.NoError(t, event.SetData(cloudevents.ApplicationJSON, events.PhoneHeartbeatCheckPayload{
-		UserID: "user-1", Owner: "+18005550199",
+		UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, PhoneID: monitor.PhoneID,
 	}))
 
 	err := routes[event.Type()](ctx, event)
 
-	require.ErrorContains(t, err, "database unavailable")
+	require.NoError(t, err)
 	assert.Zero(t, repository.lastCalls)
-	assert.Empty(t, queue.events)
+	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 {
@@ -115,7 +119,7 @@ func TestHeartbeatListener_ChecksOnlyCurrentQueueTask(t *testing.T) {
 			event.SetType(events.EventTypePhoneHeartbeatCheck)
 			ctx := context.Background()
 			if test.eventQueue != "" {
-				ctx = context.WithValue(ctx, events.ExtensionQueueID, 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(),
diff --git a/api/pkg/services/event_dispatcher_service.go b/api/pkg/services/event_dispatcher_service.go
index 59e27d7ea..1b2e3478b 100644
--- a/api/pkg/services/event_dispatcher_service.go
+++ b/api/pkg/services/event_dispatcher_service.go
@@ -56,9 +56,7 @@ func (dispatcher *EventDispatcher) DispatchSync(ctx context.Context, event cloud
 		return dispatcher.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot dispatch event with ID [%s] and type [%s] because it is invalid", event.ID(), event.Type()))
 	}
 
-	if err := dispatcher.Publish(ctx, event); err != nil && event.Type() == events.EventTypePhoneHeartbeatCheck {
-		return dispatcher.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot dispatch heartbeat check with ID [%s]", event.ID()))
-	}
+	dispatcher.Publish(ctx, event)
 	return nil
 }
 
@@ -91,7 +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.ExtensionQueueID, queueID)
+		ctx = context.WithValue(ctx, events.QueueIDKey, queueID)
 		time.AfterFunc(timeout, func() {
 			dispatcher.Publish(ctx, event)
 		})
@@ -117,7 +115,7 @@ func (dispatcher *EventDispatcher) Subscribe(eventType string, listener events.E
 }
 
 // Publish an event to subscribers
-func (dispatcher *EventDispatcher) Publish(ctx context.Context, event cloudevents.Event) error {
+func (dispatcher *EventDispatcher) Publish(ctx context.Context, event cloudevents.Event) {
 	ctx, span, ctxLogger := dispatcher.tracer.StartWithLogger(ctx, dispatcher.logger)
 	defer span.End()
 
@@ -128,29 +126,21 @@ func (dispatcher *EventDispatcher) Publish(ctx context.Context, event cloudevent
 	subscribers, ok := dispatcher.listeners[event.Type()]
 	if !ok {
 		ctxLogger.Info(fmt.Sprintf("no listener is configured for event type [%s] with id [%s]", event.Type(), event.ID()))
-		return nil
+		return
 	}
 
-	listenerErrors := make(chan error, len(subscribers))
 	var wg sync.WaitGroup
 	for _, sub := range subscribers {
 		wg.Add(1)
 		go func(ctx context.Context, sub events.EventListener) {
 			if err := sub(ctx, event); err != nil {
-				wrapped := stacktrace.Propagatef(err, "subscriber [%T] cannot handle event [%s]", sub, event.Type())
-				ctxLogger.Error(wrapped)
-				listenerErrors <- wrapped
+				ctxLogger.Error(stacktrace.Propagatef(err, "subscriber [%T] cannot handle event [%s]", sub, event.Type()))
 			}
 			wg.Done()
 		}(ctx, sub)
 	}
 
 	wg.Wait()
-	close(listenerErrors)
-	var failures []error
-	for err := range listenerErrors {
-		failures = append(failures, err)
-	}
 
 	dispatcher.meter.Record(
 		ctx,
@@ -160,10 +150,6 @@ func (dispatcher *EventDispatcher) Publish(ctx context.Context, event cloudevent
 			semconv.CloudeventsEventSpecVersion(event.SpecVersion()),
 		),
 	)
-	if len(failures) != 0 {
-		return dispatcher.tracer.WrapErrorSpan(span, stacktrace.Propagatef(errors.Join(failures...), "cannot publish event [%s] with ID [%s]", event.Type(), event.ID()))
-	}
-	return nil
 }
 
 func (dispatcher *EventDispatcher) addCloudEventAttributes(span trace.Span, event cloudevents.Event) {
diff --git a/api/pkg/services/event_dispatcher_service_test.go b/api/pkg/services/event_dispatcher_service_test.go
index 7074ce30c..8702ad7d4 100644
--- a/api/pkg/services/event_dispatcher_service_test.go
+++ b/api/pkg/services/event_dispatcher_service_test.go
@@ -29,7 +29,7 @@ func TestEventDispatcher_LocalFallbackIncludesQueueID(t *testing.T) {
 	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.ExtensionQueueID).(string)
+		queueID, _ := ctx.Value(events.QueueIDKey).(string)
 		delivered <- queueID
 		return nil
 	})
diff --git a/api/pkg/services/heartbeat_service.go b/api/pkg/services/heartbeat_service.go
index 2d3e8a49c..7af3c9efe 100644
--- a/api/pkg/services/heartbeat_service.go
+++ b/api/pkg/services/heartbeat_service.go
@@ -297,7 +297,8 @@ func (service *HeartbeatService) Monitor(ctx context.Context, params *HeartbeatM
 		return nil
 	}
 	if err != nil {
-		return service.tracer.WrapErrorSpan(span, 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 {
@@ -315,7 +316,8 @@ func (service *HeartbeatService) Monitor(ctx context.Context, params *HeartbeatM
 		return nil
 	}
 	if err != nil {
-		return 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))
+		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
 	}
 
 	return service.checkLastHeartbeat(ctx, heartbeat, monitor, params)
@@ -410,7 +412,7 @@ func (service *HeartbeatService) scheduleHeartbeatCheck(ctx context.Context, las
 	}
 
 	if err = service.monitorRepository.UpdateQueueID(ctx, params.MonitorID, queueID); err != nil {
-		return service.tracer.WrapErrorSpan(span, 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 7fa84d278..7e7ae31c7 100644
--- a/api/pkg/services/heartbeat_service_test.go
+++ b/api/pkg/services/heartbeat_service_test.go
@@ -174,7 +174,7 @@ func TestHeartbeatServiceMonitor_StopsCheckingDeletedMonitor(t *testing.T) {
 	assert.Empty(t, queue.events)
 }
 
-func TestHeartbeatServiceMonitor_ReportsHeartbeatReadFailure(t *testing.T) {
+func TestHeartbeatServiceMonitor_StopsOnHeartbeatReadFailure(t *testing.T) {
 	logger := &phoneNotificationLogger{}
 	tracer := telemetry.NewOtelLogger("test", logger)
 	queue := &heartbeatServiceQueue{}
@@ -190,18 +190,18 @@ func TestHeartbeatServiceMonitor_ReportsHeartbeatReadFailure(t *testing.T) {
 		UserID: monitor.UserID, Owner: monitor.Owner, QueueID: monitor.QueueID,
 	})
 
-	require.ErrorContains(t, err, "heartbeat read failed")
+	require.NoError(t, err)
 	assert.Equal(t, "current-check", monitor.QueueID)
 	assert.Empty(t, queue.events)
 }
 
-func TestHeartbeatServiceMonitor_RetriesFailedQueueIDSaves(t *testing.T) {
+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,
+		QueueID: "current-check", PhoneOnline: true, UpdatedAt: time.Now().UTC().Add(-3 * time.Hour),
 	}
 	monitorRepository := &heartbeatServiceMonitorRepository{
 		monitor: monitor, updateQueueErr: stacktrace.NewError("queue update failed"),
@@ -213,12 +213,15 @@ func TestHeartbeatServiceMonitor_RetriesFailedQueueIDSaves(t *testing.T) {
 		UserID: monitor.UserID, Owner: monitor.Owner, QueueID: monitor.QueueID, Source: "https://example.com",
 	}
 
-	require.ErrorContains(t, service.Monitor(context.Background(), params), "queue update failed")
+	require.NoError(t, service.Monitor(context.Background(), params))
 	assert.Equal(t, "current-check", monitor.QueueID)
 	require.Len(t, queue.events, 1)
 
 	monitorRepository.updateQueueErr = nil
-	require.NoError(t, service.Monitor(context.Background(), params))
+	_, 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)
 

From 13c10dc8ba031aad4b363bb39c55b80725174b16 Mon Sep 17 00:00:00 2001
From: Acho Arnold Ewin <arnold@ndolestudio.com>
Date: Sat, 7 Nov 2026 15:19:26 +0200
Subject: [PATCH 7/7] fix: reject heartbeat checks without a delivery task ID

Stop anonymous checks before monitor lookup or recovery scheduling.
Cover empty stored IDs and unavailable monitors in listener regressions.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 742e0212-0866-4232-89d8-7d194134448a
---
 api/pkg/listeners/heartbeat_listener_test.go | 5 ++++-
 api/pkg/services/heartbeat_service.go        | 5 +++++
 api/pkg/services/heartbeat_service_test.go   | 2 +-
 3 files changed, 10 insertions(+), 2 deletions(-)

diff --git a/api/pkg/listeners/heartbeat_listener_test.go b/api/pkg/listeners/heartbeat_listener_test.go
index 574362282..c68f7cc03 100644
--- a/api/pkg/listeners/heartbeat_listener_test.go
+++ b/api/pkg/listeners/heartbeat_listener_test.go
@@ -93,10 +93,13 @@ func TestHeartbeatListener_ChecksOnlyCurrentQueueTask(t *testing.T) {
 		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) {
@@ -109,7 +112,7 @@ func TestHeartbeatListener_ChecksOnlyCurrentQueueTask(t *testing.T) {
 			repository := &heartbeatListenerRepository{}
 			queue := &heartbeatListenerQueue{}
 			service := services.NewHeartbeatService(logger, tracer, repository,
-				&heartbeatListenerMonitorRepository{monitor: monitor},
+				&heartbeatListenerMonitorRepository{monitor: monitor, err: test.loadErr},
 				services.NewEventDispatcher(logger, tracer, nil, queue, services.PushQueueConfig{}))
 			_, routes := NewHeartbeatListener(logger, tracer, service)
 
diff --git a/api/pkg/services/heartbeat_service.go b/api/pkg/services/heartbeat_service.go
index 7af3c9efe..059897061 100644
--- a/api/pkg/services/heartbeat_service.go
+++ b/api/pkg/services/heartbeat_service.go
@@ -291,6 +291,11 @@ 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))
diff --git a/api/pkg/services/heartbeat_service_test.go b/api/pkg/services/heartbeat_service_test.go
index 7e7ae31c7..dcd150b4a 100644
--- a/api/pkg/services/heartbeat_service_test.go
+++ b/api/pkg/services/heartbeat_service_test.go
@@ -169,7 +169,7 @@ 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)
 }