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/entities/heartbeat_monitor.go

@@ -1,16 +1,21 @@
 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"`
+	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"`
 	Owner       string    `json:"owner" bson:"owner" example:"+18005550199"`
 	PhoneOnline bool      `json:"phone_online" bson:"phone_online" example:"true" default:"true"`
@@ -20,7 +25,8 @@ 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()) ||
+		(strings.HasPrefix(h.QueueID, "pending::") && h.UpdatedAt.Add(HeartbeatCheckClaimLease).Before(time.Now()))
 }
 
 // PhoneIsOffline returns true if the phone is offline

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

@@ -96,6 +96,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,7 +108,6 @@ 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 {
 	ctx, span := listener.tracer.Start(ctx)
 	defer span.End()
@@ -118,7 +118,7 @@ func (listener *HeartbeatListener) onPhoneHeartbeatOffline(ctx context.Context,
 	}
 
 	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()))
+		return listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot mark heartbeat monitor [%s] offline", payload.MonitorID))
 	}
 
 	return nil

api/pkg/listeners/heartbeat_listener_test.go

@@ -0,0 +1,64 @@
+package listeners
+
+import (
+	"context"
+	"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"
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/metric/noop"
+)
+
+type offlineMonitorRepository struct {
+	repositories.HeartbeatMonitorRepository
+	monitor *entities.HeartbeatMonitor
+}
+
+func (repository *offlineMonitorRepository) UpdatePhoneOnline(_ context.Context, userID entities.UserID, monitorID uuid.UUID, online bool) error {
+	if repository.monitor.UserID == userID && repository.monitor.ID == monitorID {
+		repository.monitor.PhoneOnline = online
+	}
+	return nil
+}
+
+func TestHeartbeatOfflineDelivery_AllowsRepeatedAndOutOfOrderEvents(t *testing.T) {
+	logger := &noopListenerLogger{}
+	tracer := telemetry.NewOtelLogger("test", logger)
+	meter, err := noop.NewMeterProvider().Meter("test").Float64Histogram("dispatch")
+	require.NoError(t, err)
+	dispatcher := services.NewEventDispatcher(logger, tracer, meter, nil, services.PushQueueConfig{})
+	monitor := &entities.HeartbeatMonitor{ID: uuid.New(), UserID: "user-1", PhoneOnline: true}
+	repository := &offlineMonitorRepository{monitor: monitor}
+	service := services.NewHeartbeatService(logger, tracer, nil, repository, dispatcher)
+	_, routes := NewHeartbeatListener(logger, tracer, service)
+	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
+	})
+	event := cloudevents.NewEvent()
+	event.SetID(uuid.NewString())
+	event.SetSource("https://example.com")
+	event.SetType(events.EventTypePhoneHeartbeatOffline)
+	require.NoError(t, event.SetData(cloudevents.ApplicationJSON, events.PhoneHeartbeatOfflinePayload{
+		MonitorID: monitor.ID, UserID: monitor.UserID, LastHeartbeatTimestamp: time.Now().UTC().Add(-24 * time.Hour),
+	}))
+
+	for delivery := 1; delivery <= 2; delivery++ {
+		monitor.PhoneOnline = true
+		require.NoError(t, dispatcher.DispatchSync(context.Background(), event))
+		require.False(t, monitor.PhoneOnline)
+		require.EqualValues(t, delivery, notifications.Load())
+	}
+}

api/pkg/repositories/gorm_heartbeat_monitor_repository.go

@@ -106,6 +106,27 @@ func (repository *gormHeartbeatMonitorRepository) Delete(ctx context.Context, us
 	return nil
 }
 
+func (repository *gormHeartbeatMonitorRepository) CompareAndSwapQueueID(ctx context.Context, monitorID uuid.UUID, expected, queueID string) (bool, error) {
+	ctx, span := repository.tracer.Start(ctx)
+	defer span.End()
+
+	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
+	defer cancel()
+
+	result := repository.db.WithContext(ctx).
+		Model(&entities.HeartbeatMonitor{}).
+		Where("id = ?", monitorID).
+		Where("queue_id = ?", expected).
+		Updates(map[string]any{
+			"queue_id":   queueID,
+			"updated_at": time.Now().UTC(),
+		})
+	if result.Error != nil {
+		return false, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(result.Error, "cannot claim heartbeat monitor ID [%s]", monitorID))
+	}
+	return result.RowsAffected == 1, nil
+}
+
 // Index entities.Message between 2 parties
 func (repository *gormHeartbeatMonitorRepository) Index(ctx context.Context, userID entities.UserID, owner string, params IndexParams) (*[]entities.Heartbeat, error) {
 	ctx, span := repository.tracer.Start(ctx)

api/pkg/repositories/gorm_heartbeat_monitor_repository_test.go

@@ -19,7 +19,12 @@ func TestHeartbeatMonitorUpdatesCarryContext(t *testing.T) {
 		"queue ID": func(repository HeartbeatMonitorRepository, ctx context.Context) error {
 			return repository.UpdateQueueID(ctx, uuid.New(), "queue-id")
 		},
+		"conditional queue ID": func(repository HeartbeatMonitorRepository, ctx context.Context) error {
+			_, err := repository.CompareAndSwapQueueID(ctx, uuid.New(), "old-check", "new-check")
+			return err
+		},
 	}
+
 	for name, update := range updates {
 		t.Run(name, func(t *testing.T) {
 			db, err := gorm.Open(postgres.New(postgres.Config{Conn: &messageThreadTestConnPool{}}), &gorm.Config{
@@ -45,3 +50,28 @@ func TestHeartbeatMonitorUpdatesCarryContext(t *testing.T) {
 		})
 	}
 }
+
+func TestHeartbeatMonitorCompareAndSwapQueueID_RequiresMatchingOwnership(t *testing.T) {
+	for _, matched := range []bool{false, true} {
+		t.Run(map[bool]string{false: "lost claim", true: "claimed"}[matched], func(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)
+			monitorID := uuid.New()
+			require.NoError(t, db.Callback().Update().After("gorm:update").Register("check_ownership", func(db *gorm.DB) {
+				require.Contains(t, db.Statement.SQL.String(), `WHERE id = $3 AND queue_id = $4`)
+				require.Equal(t, monitorID, db.Statement.Vars[2])
+				require.Equal(t, "old-check", db.Statement.Vars[3])
+				if matched {
+					db.RowsAffected = 1
+				}
+			}))
+			logger := &messageThreadTestLogger{}
+			repository := NewGormHeartbeatMonitorRepository(logger, telemetry.NewOtelLogger("test", logger), db)
+			claimed, err := repository.CompareAndSwapQueueID(context.Background(), monitorID, "old-check", "new-check")
+			require.NoError(t, err)
+			require.Equal(t, matched, claimed)
+		})
+	}
+}

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
 
+	// CompareAndSwapQueueID changes check ownership only if the expected marker still matches.
+	CompareAndSwapQueueID(ctx context.Context, monitorID uuid.UUID, expected, queueID string) (bool, error)
+
 	// Delete an entities.HeartbeatMonitor
 	Delete(ctx context.Context, userID entities.UserID, phoneNumber string) error
 

api/pkg/repositories/mongo_heartbeat_monitor_repository.go

@@ -134,6 +134,28 @@ func (repository *mongoHeartbeatMonitorRepository) Delete(ctx context.Context, u
 	return nil
 }
 
+func (repository *mongoHeartbeatMonitorRepository) CompareAndSwapQueueID(ctx context.Context, monitorID uuid.UUID, expected, queueID string) (bool, 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: expected},
+	}
+	update := bson.D{{Key: "$set", Value: bson.D{
+		{Key: "queue_id", Value: queueID},
+		{Key: "updated_at", Value: time.Now().UTC()},
+	}}}
+	result, err := repository.collection.UpdateOne(ctx, filter, update)
+	if err != nil {
+		return false, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot claim heartbeat monitor ID [%s]", monitorID))
+	}
+	return result.MatchedCount == 1, nil
+}
+
 func (repository *mongoHeartbeatMonitorRepository) UpdatePhoneOnline(ctx context.Context, userID entities.UserID, monitorID uuid.UUID, online bool) error {
 	ctx, span := repository.tracer.Start(ctx)
 	defer span.End()

api/pkg/repositories/mongo_heartbeat_repository.go

@@ -3,6 +3,7 @@ package repositories
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"go.mongodb.org/mongo-driver/v2/bson"
 	"go.mongodb.org/mongo-driver/v2/mongo"
@@ -99,17 +100,40 @@ func (repository *mongoHeartbeatRepository) Last(ctx context.Context, userID ent
 		{Key: "owner", Value: owner},
 	}
 
-	opts := options.FindOne().SetSort(bson.D{{Key: "timestamp", Value: -1}})
+	opts := options.Find().SetSort(bson.D{{Key: "timestamp", Value: -1}}).SetLimit(2)
+	cursor, err := repository.collection.Find(ctx, filter, opts)
+	if err != nil {
+		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot load heartbeat with userID [%s] and owner [%s]", userID, owner))
+	}
+
+	var candidates []entities.Heartbeat
+	if err := cursor.All(ctx, &candidates); err != nil {
+		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot decode latest heartbeats with userID [%s] and owner [%s]", userID, owner))
+	}
+	if len(candidates) == 0 {
+		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.PropagateWithCodef(mongo.ErrNoDocuments, ErrCodeNotFound, "heartbeat with userID [%s] and owner [%s] does not exist", userID, owner))
+	}
+	if len(candidates) == 1 || !candidates[0].Timestamp.Equal(candidates[1].Timestamp) {
+		return &candidates[0], nil
+	}
+	return repository.lastHeartbeatAtTimestamp(ctx, userID, owner, candidates[0].Timestamp)
+}
 
+func (repository *mongoHeartbeatRepository) lastHeartbeatAtTimestamp(ctx context.Context, userID entities.UserID, owner string, timestamp time.Time) (*entities.Heartbeat, error) {
+	ctx, span := repository.tracer.Start(ctx)
+	defer span.End()
+
+	filter := bson.D{{Key: "user_id", Value: string(userID)}, {Key: "owner", Value: owner}, {Key: "timestamp", Value: timestamp}}
+	// Restrict sorting to the tied timestamp; the existing history index bounds both queries.
+	opts := options.FindOne().SetSort(bson.D{{Key: "_id", Value: 1}})
 	var heartbeat entities.Heartbeat
 	err := repository.collection.FindOne(ctx, filter, opts).Decode(&heartbeat)
 	if err == mongo.ErrNoDocuments {
 		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.PropagateWithCodef(err, ErrCodeNotFound, "heartbeat with userID [%s] and owner [%s] does not exist", userID, owner))
 	}
 	if err != nil {
-		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot load heartbeat with userID [%s] and owner [%s]", userID, owner))
+		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot resolve tied heartbeats with userID [%s] and owner [%s]", userID, owner))
 	}
-
 	return &heartbeat, nil
 }