fix(api): expire MongoDB heartbeats after 60 days

.github/copilot-instructions.md

@@ -123,6 +123,13 @@ The API uses a **DI container** (`pkg/di/container.go`) that lazily initializes
 
 ## Key Conventions
 
+### Clean Code & SOLID
+
+- Write the simplest, clearest code that fully satisfies the requirement. Follow SOLID principles without introducing unnecessary abstractions or overengineering.
+- Keep methods short and focused on a single responsibility. Avoid long methods that combine unrelated functionality.
+- Extract distinct responsibilities into clearly named private methods (unexported methods in Go), keeping the main method a readable orchestration of those steps.
+- Keep responsibilities cohesive, depend on appropriate interfaces, and prefer small, focused interfaces over broad ones. Preserve existing behavior when refactoring.
+
 ### API (Go)
 
 - **Error handling**: Use `github.com/palantir/stacktrace` — wrap errors with `stacktrace.Propagate(err, "context")` or `stacktrace.PropagateWithCode()`. Never return bare errors.

api/pkg/di/container.go

@@ -64,6 +64,7 @@ import (
 	"github.com/gofiber/fiber/v3/middleware/cors"
 
 	"github.com/NdoleStudio/httpsms/pkg/entities"
+	"github.com/NdoleStudio/httpsms/pkg/events"
 	"github.com/NdoleStudio/httpsms/pkg/listeners"
 	"github.com/NdoleStudio/httpsms/pkg/repositories"
 	"github.com/NdoleStudio/httpsms/pkg/services"
@@ -1501,11 +1502,12 @@ func (container *Container) RegisterNotificationListeners() {
 // RegisterHeartbeatListeners registers event listeners for listeners.HeartbeatListener
 func (container *Container) RegisterHeartbeatListeners() {
 	container.logger.Debug(fmt.Sprintf("registering listners for %T", listeners.HeartbeatListener{}))
-	_, routes := listeners.NewHeartbeatListener(
+	heartbeatListener, routes := listeners.NewHeartbeatListener(
 		container.Logger(),
 		container.Tracer(),
 		container.HeartbeatService(),
 	)
+	container.EventDispatcher().SubscribeFilter(events.EventTypePhoneHeartbeatOffline, heartbeatListener.FilterOfflineEvent)
 
 	for event, handler := range routes {
 		container.EventDispatcher().Subscribe(event, handler)

api/pkg/entities/heartbeat_monitor.go

@@ -1,26 +1,34 @@
 package entities
 
 import (
+	"strings"
 	"time"
 
 	"github.com/google/uuid"
 )
 
+// HeartbeatCheckClaimLease bounds recovery of interrupted heartbeat check claims.
+const HeartbeatCheckClaimLease = time.Minute
+
 // HeartbeatMonitor is used to monitor heartbeats of a phone
 type HeartbeatMonitor struct {
-	ID          uuid.UUID `json:"id" gorm:"primaryKey;type:uuid;" bson:"_id" example:"32343a19-da5e-4b1b-a767-3298a73703cb"`
-	PhoneID     uuid.UUID `json:"phone_id" bson:"phone_id" example:"32343a19-da5e-4b1b-a767-3298a73703cb"`
-	UserID      UserID    `json:"user_id" bson:"user_id" example:"WB7DRDWrJZRGbYrv2CKGkqbzvqdC"`
-	QueueID     string    `json:"queue_id" bson:"queue_id" example:"0360259236613675274"`
-	Owner       string    `json:"owner" bson:"owner" example:"+18005550199"`
-	PhoneOnline bool      `json:"phone_online" bson:"phone_online" example:"true" default:"true"`
-	CreatedAt   time.Time `json:"created_at" bson:"created_at" example:"2022-06-05T14:26:02.302718+03:00"`
-	UpdatedAt   time.Time `json:"updated_at" bson:"updated_at" example:"2022-06-05T14:26:10.303278+03:00"`
+	ID      uuid.UUID `json:"id" gorm:"primaryKey;type:uuid;" bson:"_id" example:"32343a19-da5e-4b1b-a767-3298a73703cb"`
+	PhoneID uuid.UUID `json:"phone_id" bson:"phone_id" example:"32343a19-da5e-4b1b-a767-3298a73703cb"`
+	UserID  UserID    `json:"user_id" bson:"user_id" example:"WB7DRDWrJZRGbYrv2CKGkqbzvqdC"`
+	// QueueID is the current check ownership marker, not the external task queue's ID.
+	QueueID string `json:"queue_id" bson:"queue_id" example:"0360259236613675274"`
+	// LastHeartbeatID fences delayed offline transitions independently of check leases.
+	LastHeartbeatID uuid.UUID `json:"-" bson:"last_heartbeat_id,omitempty" gorm:"type:uuid"`
+	Owner           string    `json:"owner" bson:"owner" example:"+18005550199"`
+	PhoneOnline     bool      `json:"phone_online" bson:"phone_online" example:"true" default:"true"`
+	CreatedAt       time.Time `json:"created_at" bson:"created_at" example:"2022-06-05T14:26:02.302718+03:00"`
+	UpdatedAt       time.Time `json:"updated_at" bson:"updated_at" example:"2022-06-05T14:26:10.303278+03:00"`
 }
 
 // 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()) ||
+		(strings.HasPrefix(h.QueueID, "pending::") && h.UpdatedAt.Add(HeartbeatCheckClaimLease).Before(time.Now()))
 }
 
 // PhoneIsOffline returns true if the phone is offline

api/pkg/events/listeners.go

@@ -8,3 +8,6 @@ import (
 
 // EventListener is the type for processing events
 type EventListener func(ctx context.Context, event cloudevents.Event) error
+
+// EventFilter decides whether an event may reach any subscribers.
+type EventFilter func(ctx context.Context, event cloudevents.Event) (bool, error)

api/pkg/events/phone_heartbeat_check_event.go

@@ -17,4 +17,5 @@ type PhoneHeartbeatCheckPayload struct {
 	ScheduledAt time.Time       `json:"scheduled_at"`
 	Owner       string          `json:"owner"`
 	MonitorID   uuid.UUID       `json:"monitor_id"`
+	CheckID     string          `json:"check_id,omitempty"`
 }

api/pkg/events/phone_heartbeat_offline_event.go

@@ -13,6 +13,7 @@ const EventTypePhoneHeartbeatOffline = "phone.heartbeat.offline"
 // PhoneHeartbeatOfflinePayload is the payload of the EventTypePhoneHeartbeatOffline event
 type PhoneHeartbeatOfflinePayload struct {
 	PhoneID                uuid.UUID       `json:"phone_id"`
+	HeartbeatID            uuid.UUID       `json:"heartbeat_id,omitempty"`
 	UserID                 entities.UserID `json:"user_id"`
 	LastHeartbeatTimestamp time.Time       `json:"last_heartbeat_timestamp"`
 	Timestamp              time.Time       `json:"timestamp"`

api/pkg/listeners/heartbeat_listener.go

@@ -33,11 +33,10 @@ func NewHeartbeatListener(
 	}
 
 	return l, map[string]events.EventListener{
-		events.EventTypePhoneUpdated:          l.onPhoneUpdated,
-		events.EventTypePhoneDeleted:          l.onPhoneDeleted,
-		events.EventTypePhoneHeartbeatCheck:   l.onPhoneHeartbeatCheck,
-		events.EventTypePhoneHeartbeatOffline: l.onPhoneHeartbeatOffline,
-		events.UserAccountDeleted:             l.onUserAccountDeleted,
+		events.EventTypePhoneUpdated:        l.onPhoneUpdated,
+		events.EventTypePhoneDeleted:        l.onPhoneDeleted,
+		events.EventTypePhoneHeartbeatCheck: l.onPhoneHeartbeatCheck,
+		events.UserAccountDeleted:           l.onUserAccountDeleted,
 	}
 }
 
@@ -96,6 +95,7 @@ func (listener *HeartbeatListener) onPhoneHeartbeatCheck(ctx context.Context, ev
 		Owner:     payload.Owner,
 		PhoneID:   payload.PhoneID,
 		MonitorID: payload.MonitorID,
+		CheckID:   payload.CheckID,
 		UserID:    payload.UserID,
 		Source:    event.Source(),
 	}
@@ -107,21 +107,22 @@ func (listener *HeartbeatListener) onPhoneHeartbeatCheck(ctx context.Context, ev
 	return nil
 }
 
-// onPhoneDeleted handles the events.EventTypePhoneDeleted event
-func (listener *HeartbeatListener) onPhoneHeartbeatOffline(ctx context.Context, event cloudevents.Event) error {
+// FilterOfflineEvent validates and applies the transition before notification subscribers run.
+func (listener *HeartbeatListener) FilterOfflineEvent(ctx context.Context, event cloudevents.Event) (bool, error) {
 	ctx, span := listener.tracer.Start(ctx)
 	defer span.End()
 
 	var payload events.PhoneHeartbeatOfflinePayload
 	if err := event.DataAs(&payload); err != nil {
-		return listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot decode [%s] into [%T]", event.Data(), payload))
+		return false, listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot decode [%s] into [%T]", event.Data(), payload))
 	}
 
-	if err := listener.service.UpdatePhoneOnline(ctx, payload.UserID, payload.MonitorID, false); err != nil {
-		return listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot delete heartbeat monitor with userID [%s] and owner [%s] for event with ID [%s]", payload.UserID, payload.Owner, event.ID()))
+	allowed, err := listener.service.ApplyOfflineHeartbeat(ctx, &payload)
+	if err != nil {
+		return false, listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot validate offline heartbeat for monitor [%s]", payload.MonitorID))
 	}
 
-	return nil
+	return allowed, nil
 }
 
 func (listener *HeartbeatListener) onUserAccountDeleted(ctx context.Context, event cloudevents.Event) error {

api/pkg/listeners/heartbeat_listener_test.go

@@ -0,0 +1,180 @@
+package listeners
+
+import (
+	"context"
+	"encoding/json"
+	"sync/atomic"
+	"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"
+	"github.com/NdoleStudio/stacktrace"
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/metric/noop"
+)
+
+type offlineHeartbeatRepository struct {
+	repositories.HeartbeatRepository
+	heartbeat *entities.Heartbeat
+	err       error
+}
+
+func (repository *offlineHeartbeatRepository) Last(context.Context, entities.UserID, string) (*entities.Heartbeat, error) {
+	return repository.heartbeat, repository.err
+}
+
+func (repository *offlineHeartbeatRepository) Store(_ context.Context, heartbeat *entities.Heartbeat) error {
+	repository.heartbeat = heartbeat
+	return nil
+}
+
+type offlineMonitorRepository struct {
+	repositories.HeartbeatMonitorRepository
+	monitor         *entities.HeartbeatMonitor
+	lastHeartbeatID uuid.UUID
+	beforeOffline   func()
+}
+
+func (repository *offlineMonitorRepository) Load(context.Context, entities.UserID, string) (*entities.HeartbeatMonitor, error) {
+	monitor := *repository.monitor
+	monitor.LastHeartbeatID = repository.lastHeartbeatID
+	return &monitor, nil
+}
+
+func (repository *offlineMonitorRepository) CompareAndSwapQueueID(_ context.Context, _ uuid.UUID, expected, next string) (bool, error) {
+	if repository.monitor.QueueID != expected {
+		return false, nil
+	}
+	repository.monitor.QueueID = next
+	repository.monitor.UpdatedAt = time.Now().UTC()
+	return true, nil
+}
+
+func (repository *offlineMonitorRepository) RecordHeartbeat(_ context.Context, _ entities.UserID, _ uuid.UUID, expected, heartbeatID uuid.UUID) (*entities.HeartbeatMonitor, error) {
+	if repository.lastHeartbeatID != expected {
+		return nil, nil
+	}
+	previous := *repository.monitor
+	previous.LastHeartbeatID = repository.lastHeartbeatID
+	repository.lastHeartbeatID = heartbeatID
+	repository.monitor.PhoneOnline = true
+	return &previous, nil
+}
+
+func (repository *offlineMonitorRepository) UpdatePhoneOfflineForHeartbeat(_ context.Context, _ entities.UserID, _ uuid.UUID, expected, heartbeatID uuid.UUID) (bool, error) {
+	repository.runOfflineHook()
+	if repository.lastHeartbeatID != expected {
+		return false, nil
+	}
+	repository.monitor.PhoneOnline = false
+	if heartbeatID != uuid.Nil {
+		repository.lastHeartbeatID = heartbeatID
+	}
+	return true, nil
+}
+
+func (repository *offlineMonitorRepository) UpdatePhoneOnline(_ context.Context, _ entities.UserID, _ uuid.UUID, online bool) error {
+	if !online {
+		repository.runOfflineHook()
+	}
+	repository.monitor.PhoneOnline = online
+	return nil
+}
+
+func (repository *offlineMonitorRepository) runOfflineHook() {
+	if repository.beforeOffline != nil {
+		hook := repository.beforeOffline
+		repository.beforeOffline = nil
+		hook()
+	}
+}
+
+type offlineEventQueue struct{ events []cloudevents.Event }
+
+func (queue *offlineEventQueue) 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 event.ID(), nil
+}
+
+func TestHeartbeatOfflineDelivery_RejectsNewerHeartbeatBeforeSubscribers(t *testing.T) {
+	for _, arrival := range []string{"before delivery", "between history and status update", "history read fails", "history expired", "no newer heartbeat", "legacy event"} {
+		t.Run(arrival, func(t *testing.T) {
+			logger := &noopListenerLogger{}
+			tracer := telemetry.NewOtelLogger("test", logger)
+			meter, err := noop.NewMeterProvider().Meter("test").Float64Histogram("dispatch")
+			require.NoError(t, err)
+			queue := &offlineEventQueue{}
+			dispatcher := services.NewEventDispatcher(logger, tracer, meter, queue, services.PushQueueConfig{})
+			history := &offlineHeartbeatRepository{heartbeat: &entities.Heartbeat{
+				ID: uuid.New(), Timestamp: time.Now().UTC().Add(-65 * time.Minute),
+			}}
+			monitor := &offlineMonitorRepository{monitor: &entities.HeartbeatMonitor{
+				ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", Owner: "+18005550199",
+				PhoneOnline: true, QueueID: "check:current-check", UpdatedAt: time.Now().UTC(),
+			}, lastHeartbeatID: history.heartbeat.ID}
+			service := services.NewHeartbeatService(logger, tracer, history, monitor, dispatcher)
+			heartbeatListener, routes := NewHeartbeatListener(logger, tracer, service)
+			dispatcher.SubscribeFilter(events.EventTypePhoneHeartbeatOffline, heartbeatListener.FilterOfflineEvent)
+			for eventType, listener := range routes {
+				dispatcher.Subscribe(eventType, listener)
+			}
+			var notifications atomic.Int32
+			dispatcher.Subscribe(events.EventTypePhoneHeartbeatOffline, func(context.Context, cloudevents.Event) error {
+				notifications.Add(1)
+				return nil
+			})
+			require.NoError(t, service.Monitor(context.Background(), &services.HeartbeatMonitorParams{
+				UserID: monitor.monitor.UserID, Owner: monitor.monitor.Owner, MonitorID: monitor.monitor.ID,
+				Source: "https://example.com", CheckID: "current-check",
+			}))
+			require.Len(t, queue.events, 3)
+			event := queue.events[1]
+			storeHeartbeat := func() {
+				_, err := service.Store(context.Background(), services.HeartbeatStoreParams{
+					UserID: monitor.monitor.UserID, Owner: monitor.monitor.Owner,
+					Source: "https://example.com", Timestamp: time.Now().UTC(),
+				})
+				require.NoError(t, err)
+			}
+			switch arrival {
+			case "before delivery":
+				storeHeartbeat()
+			case "between history and status update":
+				monitor.beforeOffline = storeHeartbeat
+			case "history read fails":
+				history.err = stacktrace.NewError("history unavailable")
+			case "history expired":
+				history.err = stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "history expired")
+			case "legacy event":
+				var data map[string]any
+				require.NoError(t, event.DataAs(&data))
+				delete(data, "heartbeat_id")
+				require.NoError(t, event.SetData(cloudevents.ApplicationJSON, data))
+				monitor.lastHeartbeatID = uuid.Nil
+			}
+			err = dispatcher.DispatchSync(context.Background(), event)
+			if arrival == "history read fails" {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+			if arrival == "no newer heartbeat" || arrival == "legacy event" {
+				require.EqualValues(t, 1, notifications.Load())
+				require.False(t, monitor.monitor.PhoneOnline)
+			} else {
+				require.Zero(t, notifications.Load(), "stale or unverifiable events must not reach email/webhook subscribers")
+				require.True(t, monitor.monitor.PhoneOnline)
+			}
+		})
+	}
+}