fix(api): fence offline alerts with heartbeat versions

Revalidate running ownership and history before offline detection and restart.

Atomically record online arrivals and guard offline fanout with heartbeat-version CAS.

Keep version writes independent of check leases and preserve fail-open heartbeat storage.

Cover delayed delivery, lease takeover, concurrent arrivals, and version-name collisions.

Narrowly suppress the Semgrep transport warning for the bufconn-only test server.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot-Session: f27d8f52-9620-4ee2-a564-ec4870a5f840

Author: AchoArnold

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

@@ -16,11 +16,13 @@ type HeartbeatMonitor struct {
 	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"`
-	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"`
+	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

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_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,
 	}
 }
 
@@ -108,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)
+			}
+		})
+	}
+}

api/pkg/repositories/gorm_heartbeat_monitor_repository.go

@@ -13,6 +13,7 @@ import (
 	"github.com/NdoleStudio/httpsms/pkg/telemetry"
 	"github.com/NdoleStudio/stacktrace"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )
 
 // gormHeartbeatRepository is responsible for persisting entities.Heartbeat
@@ -127,6 +128,63 @@ func (repository *gormHeartbeatMonitorRepository) CompareAndSwapQueueID(ctx cont
 	return result.RowsAffected == 1, nil
 }
 
+func (repository *gormHeartbeatMonitorRepository) RecordHeartbeat(ctx context.Context, userID entities.UserID, monitorID, expected, heartbeatID uuid.UUID) (*entities.HeartbeatMonitor, error) {
+	ctx, span := repository.tracer.Start(ctx)
+	defer span.End()
+	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
+	defer cancel()
+
+	var previous *entities.HeartbeatMonitor
+	err := repository.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		monitor := new(entities.HeartbeatMonitor)
+		query := heartbeatMonitorWithVersion(tx.Where("id = ?", monitorID).Where("user_id = ?", userID), expected)
+		if err := query.Clauses(clause.Locking{Strength: "UPDATE"}).First(monitor).Error; err != nil {
+			if errors.Is(err, gorm.ErrRecordNotFound) {
+				return nil
+			}
+			return stacktrace.Propagatef(err, "cannot lock heartbeat version for monitor [%s]", monitorID)
+		}
+		before := *monitor
+		if err := tx.Model(monitor).UpdateColumns(map[string]any{
+			"last_heartbeat_id": heartbeatID,
+			"phone_online":      true,
+		}).Error; err != nil {
+			return stacktrace.Propagatef(err, "cannot record heartbeat version for monitor [%s]", monitorID)
+		}
+		previous = &before
+		return nil
+	})
+	if err != nil {
+		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot mark heartbeat monitor [%s] online", monitorID))
+	}
+	return previous, nil
+}
+
+func (repository *gormHeartbeatMonitorRepository) UpdatePhoneOfflineForHeartbeat(ctx context.Context, userID entities.UserID, monitorID, expected, heartbeatID uuid.UUID) (bool, error) {
+	ctx, span := repository.tracer.Start(ctx)
+	defer span.End()
+	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
+	defer cancel()
+
+	query := heartbeatMonitorWithVersion(repository.db.WithContext(ctx).
+		Model(&entities.HeartbeatMonitor{}).Where("id = ?", monitorID).Where("user_id = ?", userID), expected)
+	result := query.UpdateColumns(map[string]any{
+		"last_heartbeat_id": heartbeatID,
+		"phone_online":      false,
+	})
+	if result.Error != nil {
+		return false, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(result.Error, "cannot apply offline heartbeat version for monitor [%s]", monitorID))
+	}
+	return result.RowsAffected == 1, nil
+}
+
+func heartbeatMonitorWithVersion(query *gorm.DB, heartbeatID uuid.UUID) *gorm.DB {
+	if heartbeatID == uuid.Nil {
+		return query.Where("last_heartbeat_id IS NULL OR last_heartbeat_id = ?", uuid.Nil)
+	}
+	return query.Where("last_heartbeat_id = ?", heartbeatID)
+}
+
 // 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)