diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index d1a887310..05ad45ec4 100644 --- a/.github/copilot-instructions.md +++ b/.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. diff --git a/api/pkg/di/container.go b/api/pkg/di/container.go index e7ca20d28..1569de48f 100644 --- a/api/pkg/di/container.go +++ b/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) diff --git a/api/pkg/entities/heartbeat_monitor.go b/api/pkg/entities/heartbeat_monitor.go index 7151f1952..428324db6 100644 --- a/api/pkg/entities/heartbeat_monitor.go +++ b/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 diff --git a/api/pkg/events/listeners.go b/api/pkg/events/listeners.go index d5285f75d..524023e34 100644 --- a/api/pkg/events/listeners.go +++ b/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) diff --git a/api/pkg/events/phone_heartbeat_check_event.go b/api/pkg/events/phone_heartbeat_check_event.go index 305ff59c0..1c674c106 100644 --- a/api/pkg/events/phone_heartbeat_check_event.go +++ b/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"` } diff --git a/api/pkg/events/phone_heartbeat_offline_event.go b/api/pkg/events/phone_heartbeat_offline_event.go index d5eed76e4..4a601a15f 100644 --- a/api/pkg/events/phone_heartbeat_offline_event.go +++ b/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"` diff --git a/api/pkg/listeners/heartbeat_listener.go b/api/pkg/listeners/heartbeat_listener.go index 4e0cdc0d7..6d0b402c6 100644 --- a/api/pkg/listeners/heartbeat_listener.go +++ b/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 { diff --git a/api/pkg/listeners/heartbeat_listener_test.go b/api/pkg/listeners/heartbeat_listener_test.go new file mode 100644 index 000000000..e51a1266c --- /dev/null +++ b/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) + } + }) + } +} diff --git a/api/pkg/repositories/gorm_heartbeat_monitor_repository.go b/api/pkg/repositories/gorm_heartbeat_monitor_repository.go index 85d1b2098..d819cb4bb 100644 --- a/api/pkg/repositories/gorm_heartbeat_monitor_repository.go +++ b/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 @@ -106,6 +107,84 @@ 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 +} + +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) diff --git a/api/pkg/repositories/gorm_heartbeat_monitor_repository_test.go b/api/pkg/repositories/gorm_heartbeat_monitor_repository_test.go index 51e792e85..fb87224b8 100644 --- a/api/pkg/repositories/gorm_heartbeat_monitor_repository_test.go +++ b/api/pkg/repositories/gorm_heartbeat_monitor_repository_test.go @@ -2,8 +2,10 @@ package repositories import ( "context" + "database/sql" "testing" + "github.com/NdoleStudio/httpsms/pkg/entities" "github.com/NdoleStudio/httpsms/pkg/telemetry" "github.com/google/uuid" "github.com/stretchr/testify/require" @@ -19,7 +21,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 +52,94 @@ 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) + }) + } +} + +type heartbeatVersionTestPool struct{ messageThreadTestConnPool } + +func (pool *heartbeatVersionTestPool) BeginTx(context.Context, *sql.TxOptions) (gorm.ConnPool, error) { + return pool, nil +} + +func (*heartbeatVersionTestPool) Commit() error { return nil } +func (*heartbeatVersionTestPool) Rollback() error { return nil } + +func TestHeartbeatMonitorRecordHeartbeat_PreservesPreviousStatusAndLease(t *testing.T) { + db, err := gorm.Open(postgres.New(postgres.Config{Conn: &heartbeatVersionTestPool{}}), &gorm.Config{ + DisableAutomaticPing: true, DryRun: true, + }) + require.NoError(t, err) + monitorID, previousID, nextID := uuid.New(), uuid.New(), uuid.New() + require.NoError(t, db.Callback().Query().After("gorm:query").Register("existing_monitor", func(db *gorm.DB) { + require.Contains(t, db.Statement.SQL.String(), "FOR UPDATE") + require.Contains(t, db.Statement.SQL.String(), "last_heartbeat_id =") + require.Contains(t, db.Statement.Vars, previousID) + require.Contains(t, db.Statement.Vars, entities.UserID("user-id")) + _, deadline := db.Statement.Context.Deadline() + require.True(t, deadline) + monitor := db.Statement.Dest.(*entities.HeartbeatMonitor) + *monitor = entities.HeartbeatMonitor{ID: monitorID, LastHeartbeatID: previousID, PhoneOnline: false} + db.RowsAffected = 1 + })) + require.NoError(t, db.Callback().Update().After("gorm:update").Register("record_online", func(db *gorm.DB) { + require.NotContains(t, db.Statement.SQL.String(), "updated_at") + require.Contains(t, db.Statement.Vars, nextID) + require.Contains(t, db.Statement.Vars, true) + db.RowsAffected = 1 + })) + logger := &messageThreadTestLogger{} + repository := NewGormHeartbeatMonitorRepository(logger, telemetry.NewOtelLogger("test", logger), db) + previous, err := repository.RecordHeartbeat(context.Background(), "user-id", monitorID, previousID, nextID) + require.NoError(t, err) + require.NotNil(t, previous) + require.False(t, previous.PhoneOnline, "returning-phone notifications need the status before the atomic update") + require.Equal(t, previousID, previous.LastHeartbeatID) +} + +func TestHeartbeatMonitorOfflineVersion_RequiresMatchingHeartbeatWithoutRefreshingLease(t *testing.T) { + for _, matched := range []bool{false, true} { + t.Run(map[bool]string{false: "new heartbeat wins", true: "offline transition wins"}[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, previousID, nextID := uuid.New(), uuid.New(), uuid.New() + require.NoError(t, db.Callback().Update().After("gorm:update").Register("offline_version", func(db *gorm.DB) { + require.Contains(t, db.Statement.SQL.String(), "last_heartbeat_id =") + require.NotContains(t, db.Statement.SQL.String(), "updated_at") + require.Equal(t, []any{nextID, false, monitorID, entities.UserID("user-id"), previousID}, db.Statement.Vars) + if matched { + db.RowsAffected = 1 + } + })) + logger := &messageThreadTestLogger{} + repository := NewGormHeartbeatMonitorRepository(logger, telemetry.NewOtelLogger("test", logger), db) + applied, err := repository.UpdatePhoneOfflineForHeartbeat(context.Background(), "user-id", monitorID, previousID, nextID) + require.NoError(t, err) + require.Equal(t, matched, applied) + }) + } +} diff --git a/api/pkg/repositories/heartbeat_monitor_repository.go b/api/pkg/repositories/heartbeat_monitor_repository.go index ddcde94d8..7bfa0e9da 100644 --- a/api/pkg/repositories/heartbeat_monitor_repository.go +++ b/api/pkg/repositories/heartbeat_monitor_repository.go @@ -22,6 +22,16 @@ 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) + + // RecordHeartbeat atomically records the heartbeat and marks the phone online. + // It returns the previous monitor, or nil when the expected heartbeat version changed. + RecordHeartbeat(ctx context.Context, userID entities.UserID, monitorID, expected, heartbeatID uuid.UUID) (*entities.HeartbeatMonitor, error) + + // UpdatePhoneOfflineForHeartbeat applies an offline transition only to the observed heartbeat version. + UpdatePhoneOfflineForHeartbeat(ctx context.Context, userID entities.UserID, monitorID, expected, heartbeatID uuid.UUID) (bool, 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..4f0b59035 100644 --- a/api/pkg/repositories/mongo_heartbeat_monitor_repository.go +++ b/api/pkg/repositories/mongo_heartbeat_monitor_repository.go @@ -8,6 +8,7 @@ import ( "github.com/google/uuid" "go.mongodb.org/mongo-driver/v2/bson" "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" "github.com/NdoleStudio/httpsms/pkg/entities" "github.com/NdoleStudio/httpsms/pkg/telemetry" @@ -134,6 +135,78 @@ 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) 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() + + update := bson.D{{Key: "$set", Value: bson.D{ + {Key: "last_heartbeat_id", Value: heartbeatID.String()}, + {Key: "phone_online", Value: true}, + }}} + var previous entities.HeartbeatMonitor + err := repository.collection.FindOneAndUpdate(ctx, heartbeatMonitorVersionFilter(userID, monitorID, expected), update, + options.FindOneAndUpdate().SetReturnDocument(options.Before)).Decode(&previous) + if err == mongo.ErrNoDocuments { + return nil, nil + } + if err != nil { + return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot record heartbeat version for monitor [%s]", monitorID)) + } + return &previous, nil +} + +func (repository *mongoHeartbeatMonitorRepository) 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() + + update := bson.D{{Key: "$set", Value: bson.D{ + {Key: "last_heartbeat_id", Value: heartbeatID.String()}, + {Key: "phone_online", Value: false}, + }}} + result, err := repository.collection.UpdateOne(ctx, heartbeatMonitorVersionFilter(userID, monitorID, expected), update) + if err != nil { + return false, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot apply offline heartbeat version for monitor [%s]", monitorID)) + } + return result.MatchedCount == 1, nil +} + +func heartbeatMonitorVersionFilter(userID entities.UserID, monitorID, heartbeatID uuid.UUID) bson.D { + filter := bson.D{{Key: "_id", Value: monitorID.String()}, {Key: "user_id", Value: string(userID)}} + if heartbeatID == uuid.Nil { + return append(filter, bson.E{Key: "$or", Value: bson.A{ + bson.D{{Key: "last_heartbeat_id", Value: uuid.Nil.String()}}, + bson.D{{Key: "last_heartbeat_id", Value: nil}}, + }}) + } + return append(filter, bson.E{Key: "last_heartbeat_id", Value: heartbeatID.String()}) +} + 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() diff --git a/api/pkg/repositories/mongo_heartbeat_monitor_repository_test.go b/api/pkg/repositories/mongo_heartbeat_monitor_repository_test.go new file mode 100644 index 000000000..8cd6f733a --- /dev/null +++ b/api/pkg/repositories/mongo_heartbeat_monitor_repository_test.go @@ -0,0 +1,29 @@ +package repositories + +import ( + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + "go.mongodb.org/mongo-driver/v2/bson" +) + +func TestHeartbeatMonitorVersionFilter_FencesUserAndHeartbeatIncludingLegacyRows(t *testing.T) { + monitorID, heartbeatID := uuid.New(), uuid.New() + for _, expected := range []uuid.UUID{uuid.Nil, heartbeatID} { + data, err := bson.Marshal(heartbeatMonitorVersionFilter("user-id", monitorID, expected)) + require.NoError(t, err) + var filter bson.M + require.NoError(t, bson.Unmarshal(data, &filter)) + require.Equal(t, monitorID.String(), filter["_id"]) + require.Equal(t, "user-id", filter["user_id"]) + if expected != uuid.Nil { + require.Equal(t, heartbeatID.String(), filter["last_heartbeat_id"]) + } else { + require.ElementsMatch(t, bson.A{ + bson.D{{Key: "last_heartbeat_id", Value: uuid.Nil.String()}}, + bson.D{{Key: "last_heartbeat_id", Value: nil}}, + }, filter["$or"], "legacy rows may encode zero UUID or have a missing/null version") + } + } +} diff --git a/api/pkg/repositories/mongo_heartbeat_repository.go b/api/pkg/repositories/mongo_heartbeat_repository.go index 44ed280db..0d3f77398 100644 --- a/api/pkg/repositories/mongo_heartbeat_repository.go +++ b/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 } diff --git a/api/pkg/repositories/mongo_heartbeat_repository_test.go b/api/pkg/repositories/mongo_heartbeat_repository_test.go new file mode 100644 index 000000000..5a42b7b3e --- /dev/null +++ b/api/pkg/repositories/mongo_heartbeat_repository_test.go @@ -0,0 +1,180 @@ +package repositories + +import ( + "bytes" + "context" + "encoding/binary" + "fmt" + "io" + "net" + "sort" + "strings" + "testing" + "time" + + "github.com/NdoleStudio/httpsms/pkg/telemetry" + "github.com/NdoleStudio/stacktrace" + "github.com/google/uuid" + "github.com/stretchr/testify/require" + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" +) + +func TestMongoHeartbeatLast_StableIdentityForEqualTimestamps(t *testing.T) { + for _, reversed := range []bool{false, true} { + t.Run(fmt.Sprint(reversed), func(t *testing.T) { + timestamp := time.Date(2026, 1, 1, 12, 0, 0, 123000000, time.UTC) + low := bson.M{"_id": "00000000-0000-4000-8000-000000000001", "timestamp": timestamp, "user_id": "user-id", "owner": "+18005550199"} + high := bson.M{"_id": "00000000-0000-4000-8000-000000000002", "timestamp": timestamp, "user_id": "user-id", "owner": "+18005550199"} + higher := bson.M{"_id": "00000000-0000-4000-8000-000000000003", "timestamp": timestamp, "user_id": "user-id", "owner": "+18005550199"} + older := bson.M{"_id": "00000000-0000-4000-8000-000000000000", "timestamp": timestamp.Add(-time.Second), "user_id": "user-id", "owner": "+18005550199"} + documents := []bson.M{older, high, higher, low} + if reversed { + documents = []bson.M{low, higher, high, older} + } + repository := heartbeatMongoRepositoryForTest(t, documents) + for range 2 { + heartbeat, err := repository.Last(context.Background(), "user-id", "+18005550199") + require.NoError(t, err) + require.Equal(t, uuid.MustParse("00000000-0000-4000-8000-000000000001"), heartbeat.ID) + } + }) + } +} + +func TestMongoHeartbeatLast_MissingAndSingleHistory(t *testing.T) { + for _, missing := range []bool{false, true} { + t.Run(fmt.Sprint(missing), func(t *testing.T) { + id := uuid.New() + documents := []bson.M{{"_id": id.String(), "timestamp": time.Now().UTC(), "user_id": "user-id", "owner": "+18005550199"}} + if missing { + documents = []bson.M{} + } + heartbeat, err := heartbeatMongoRepositoryForTest(t, documents).Last(context.Background(), "user-id", "+18005550199") + if missing { + require.Equal(t, ErrCodeNotFound, stacktrace.GetCode(err)) + require.Nil(t, heartbeat) + return + } + require.NoError(t, err) + require.Equal(t, id, heartbeat.ID) + }) + } +} + +func heartbeatMongoRepositoryForTest(t *testing.T, documents []bson.M) HeartbeatRepository { + t.Helper() + client, err := mongo.Connect(options.Client().SetHosts([]string{"memory:27017"}). + SetDirect(true).SetRegistry(newMongoRegistry()).SetDialer(&heartbeatMongoTestDialer{documents: documents}). + SetServerSelectionTimeout(time.Second)) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Disconnect(context.Background())) }) + logger := &messageThreadTestLogger{} + return NewMongoHeartbeatRepository(logger, telemetry.NewOtelLogger("test", logger), client.Database("test")) +} + +// The driver exchanges real MongoDB commands over net.Pipe; no database or network socket is needed. +type heartbeatMongoTestDialer struct{ documents []bson.M } + +func (dialer *heartbeatMongoTestDialer) DialContext(context.Context, string, string) (net.Conn, error) { + client, server := net.Pipe() + go func() { + defer func() { _ = server.Close() }() + for { + command, requestID, opcode, err := readHeartbeatMongoCommand(server) + if err != nil { + return + } + response := dialer.response(command) + if err := writeHeartbeatMongoResponse(server, response, requestID, opcode); err != nil { + return + } + } + }() + return client, nil +} + +func (dialer *heartbeatMongoTestDialer) response(command bson.Raw) bson.D { + if _, find := command.Lookup("find").StringValueOK(); !find { + return bson.D{ + {Key: "ok", Value: 1}, + {Key: "isWritablePrimary", Value: true}, + {Key: "maxWireVersion", Value: 25}, + {Key: "maxBsonObjectSize", Value: 16777216}, + {Key: "maxMessageSizeBytes", Value: 48000000}, + {Key: "maxWriteBatchSize", Value: 100000}, + } + } + var order bson.D + _ = bson.Unmarshal(command.Lookup("sort").Document(), &order) + documents := []bson.M{} + timestamp, bounded := command.Lookup("filter").Document().Lookup("timestamp").DateTimeOK() + for _, document := range dialer.documents { + if !bounded || document["timestamp"].(time.Time).UnixMilli() == timestamp { + documents = append(documents, document) + } + } + sort.SliceStable(documents, func(i, j int) bool { + for _, key := range order { + comparison := 0 + switch key.Key { + case "timestamp": + comparison = documents[i]["timestamp"].(time.Time).Compare(documents[j]["timestamp"].(time.Time)) + case "_id": + comparison = strings.Compare(documents[i]["_id"].(string), documents[j]["_id"].(string)) + } + if comparison != 0 { + return comparison*int(key.Value.(int32)) < 0 + } + } + return false + }) + limit := int(command.Lookup("limit").AsInt64()) + if limit > 0 && limit < len(documents) { + documents = documents[:limit] + } + return bson.D{{Key: "ok", Value: 1}, {Key: "cursor", Value: bson.D{ + {Key: "id", Value: int64(0)}, + {Key: "ns", Value: "test.heartbeats"}, + {Key: "firstBatch", Value: documents}, + }}} +} + +func readHeartbeatMongoCommand(connection net.Conn) (bson.Raw, int32, int32, error) { + header := make([]byte, 16) + if _, err := io.ReadFull(connection, header); err != nil { + return nil, 0, 0, err + } + body := make([]byte, int(binary.LittleEndian.Uint32(header))-16) + if _, err := io.ReadFull(connection, body); err != nil { + return nil, 0, 0, err + } + opcode := int32(binary.LittleEndian.Uint32(header[12:])) + offset := 5 + if opcode == 2004 { + offset = 4 + bytes.IndexByte(body[4:], 0) + 1 + 8 + } + return bson.Raw(body[offset:]), int32(binary.LittleEndian.Uint32(header[4:])), opcode, nil +} + +func writeHeartbeatMongoResponse(connection net.Conn, response bson.D, requestID, opcode int32) error { + document, err := bson.Marshal(response) + if err != nil { + return err + } + header := make([]byte, 16) + body := make([]byte, 5) + responseOpcode := uint32(2013) + if opcode == 2004 { + responseOpcode = 1 + body = make([]byte, 20) + binary.LittleEndian.PutUint32(body[16:], 1) + } + body = append(body, document...) + binary.LittleEndian.PutUint32(header, uint32(len(header)+len(body))) + binary.LittleEndian.PutUint32(header[8:], uint32(requestID)) + binary.LittleEndian.PutUint32(header[12:], responseOpcode) + _, err = connection.Write(append(header, body...)) + return err +} diff --git a/api/pkg/repositories/mongodb.go b/api/pkg/repositories/mongodb.go index 44f35bd4b..c0b7f49ee 100644 --- a/api/pkg/repositories/mongodb.go +++ b/api/pkg/repositories/mongodb.go @@ -109,6 +109,12 @@ func createMongoIndexes(ctx context.Context, db *mongo.Database) error { _, err := heartbeatsCol.Indexes().CreateMany(ctx, []mongo.IndexModel{ {Keys: bson.D{{Key: "user_id", Value: 1}, {Key: "owner", Value: 1}, {Key: "timestamp", Value: -1}}}, + { + Keys: bson.D{{Key: "timestamp", Value: 1}}, + Options: options.Index(). + SetName("heartbeats_retention"). + SetExpireAfterSeconds(int32((60 * 24 * time.Hour) / time.Second)), + }, }) if err != nil { return stacktrace.Propagatef(err, "cannot create indexes on heartbeats collection") diff --git a/api/pkg/services/emulator_push_queue.go b/api/pkg/services/emulator_push_queue.go index 634689f1e..2abe26a30 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" + "sync" "time" "github.com/NdoleStudio/stacktrace" @@ -14,10 +15,11 @@ import ( ) type emulatorPushQueue struct { - config PushQueueConfig - client *http.Client - logger telemetry.Logger - tracer telemetry.Tracer + config PushQueueConfig + client *http.Client + logger telemetry.Logger + tracer telemetry.Tracer + namedTasks sync.Map } // EmulatorPushQueue creates a new googlePushQueue @@ -41,6 +43,14 @@ func (queue *emulatorPushQueue) Enqueue(ctx context.Context, task *PushQueueTask defer span.End() queueID = uuid.New().String() + if task.ID != "" { + queueID = task.ID + if _, exists := queue.namedTasks.LoadOrStore(queueID, struct{}{}); exists { + return queueID, nil + } + // Mirror Cloud Tasks' named-task retention in this non-durable local emulator. + time.AfterFunc(24*time.Hour, func() { queue.namedTasks.Delete(queueID) }) + } time.AfterFunc(timeout, queue.push(*task, queueID)) @@ -56,28 +66,30 @@ func (queue *emulatorPushQueue) Enqueue(ctx context.Context, task *PushQueueTask func (queue *emulatorPushQueue) push(task PushQueueTask, queueID string) func() { return func() { - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) - defer cancel() - - request := requests. - URL(task.URL). - Client(queue.client). - Method(task.Method). - BodyBytes(task.Body) - - // add headers - for key, value := range task.Headers { - request.Header(key, value) + backoff := time.Second + for { + if err := queue.send(task); err == nil { + queue.logger.Info(fmt.Sprintf("queue task [%s] sent to URL [%s]", queueID, task.URL)) + return + } else { + queue.logger.Error(stacktrace.Propagatef(err, "cannot send http request to [%s] for queue task [%s]", task.URL, queueID)) + } + time.Sleep(backoff) + backoff = min(2*backoff, time.Minute) } + } +} - // add content type - request.Header("Content-Type", "application/json") - - 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)) - return - } +func (queue *emulatorPushQueue) send(task PushQueueTask) error { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() - queue.logger.Info(fmt.Sprintf("queue task [%s] sent to URL [%s]", queueID, task.URL)) + request := requests.URL(task.URL).Client(queue.client).Method(task.Method).BodyBytes(task.Body) + for key, value := range task.Headers { + request.Header(key, value) + } + if err := request.Header("Content-Type", "application/json").Fetch(ctx); err != nil { + return stacktrace.Propagatef(err, "cannot deliver queue HTTP request to [%s]", task.URL) } + return nil } diff --git a/api/pkg/services/emulator_push_queue_test.go b/api/pkg/services/emulator_push_queue_test.go new file mode 100644 index 000000000..fc1fa4c94 --- /dev/null +++ b/api/pkg/services/emulator_push_queue_test.go @@ -0,0 +1,87 @@ +package services + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/NdoleStudio/httpsms/pkg/events" + "github.com/NdoleStudio/httpsms/pkg/telemetry" + cloudevents "github.com/cloudevents/sdk-go/v2" + "github.com/google/uuid" + "github.com/stretchr/testify/require" +) + +func TestEmulatorPushQueue_DeduplicatesOfflineEvents(t *testing.T) { + deliveries := make(chan struct{}, 2) + consumer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) { + deliveries <- struct{}{} + writer.WriteHeader(http.StatusNoContent) + })) + t.Cleanup(consumer.Close) + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + dispatcher := NewEventDispatcher(logger, tracer, nil, + EmulatorPushQueue(logger, tracer, consumer.Client(), PushQueueConfig{}), + PushQueueConfig{ConsumerEndpoint: consumer.URL}) + event := cloudevents.NewEvent() + event.SetID(uuid.NewString()) + event.SetType(events.EventTypePhoneHeartbeatOffline) + event.SetSource("https://example.com") + require.NoError(t, dispatcher.Dispatch(context.Background(), event)) + require.NoError(t, dispatcher.Dispatch(context.Background(), event)) + select { + case <-deliveries: + case <-time.After(time.Second): + t.Fatal("offline alert was not delivered") + } + select { + case <-deliveries: + t.Fatal("offline alert was delivered twice") + case <-time.After(100 * time.Millisecond): + } +} + +func TestEmulatorPushQueue_RetriesFailedHeartbeatDelivery(t *testing.T) { + var attempts atomic.Int32 + delivered := make(chan string, 1) + firstFailure := make(chan struct{}) + consumer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + body, _ := io.ReadAll(request.Body) + if attempts.Add(1) < 3 { + writer.WriteHeader(http.StatusServiceUnavailable) + if attempts.Load() == 1 { + close(firstFailure) + } + return + } + writer.WriteHeader(http.StatusNoContent) + delivered <- string(body) + })) + t.Cleanup(consumer.Close) + logger := &phoneNotificationLogger{} + queue := EmulatorPushQueue(logger, telemetry.NewOtelLogger("test", logger), consumer.Client(), PushQueueConfig{}) + task := &PushQueueTask{ID: "retryable-heartbeat", URL: consumer.URL, Method: http.MethodPost, Body: []byte(`{"id":"same-heartbeat-check"}`)} + ctx, cancel := context.WithCancel(context.Background()) + _, err := queue.Enqueue(ctx, task, 0) + require.NoError(t, err) + cancel() + select { + case <-firstFailure: + case <-time.After(time.Second): + t.Fatal("initial heartbeat delivery did not run") + } + _, err = queue.Enqueue(context.Background(), task, 0) + require.NoError(t, err) + select { + case body := <-delivered: + require.Equal(t, `{"id":"same-heartbeat-check"}`, body) + require.Equal(t, int32(3), attempts.Load()) + case <-time.After(5 * time.Second): + t.Fatal("transient HTTP failures dropped the heartbeat check instead of retrying it") + } +} diff --git a/api/pkg/services/event_dispatcher_service.go b/api/pkg/services/event_dispatcher_service.go index 245ab0b2d..967180ee4 100644 --- a/api/pkg/services/event_dispatcher_service.go +++ b/api/pkg/services/event_dispatcher_service.go @@ -24,6 +24,7 @@ type EventDispatcher struct { logger telemetry.Logger tracer telemetry.Tracer listeners map[string][]events.EventListener + filters map[string]events.EventFilter meter metric.Float64Histogram queue PushQueue queueConfig PushQueueConfig @@ -42,6 +43,7 @@ func NewEventDispatcher( tracer: tracer, meter: meter, listeners: make(map[string][]events.EventListener), + filters: make(map[string]events.EventFilter), queue: queue, queueConfig: queueConfig, } @@ -56,7 +58,14 @@ 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) + allowed, err := dispatcher.shouldPublish(ctx, event) + if err != nil || !allowed { + return err + } + if event.Type() == events.EventTypePhoneHeartbeatCheck { + return dispatcher.publishSubscribers(ctx, event) + } + _ = dispatcher.publishSubscribers(ctx, event) return nil } @@ -86,7 +95,7 @@ func (dispatcher *EventDispatcher) enqueue(ctx context.Context, event cloudevent defer span.End() queueID, err := dispatcher.queue.Enqueue(ctx, task, timeout) - if errors.Is(err, context.DeadlineExceeded) { + if errors.Is(err, context.DeadlineExceeded) && task.ID == "" && event.Type() != events.EventTypePhoneHeartbeatCheck { 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 time.AfterFunc(timeout, func() { @@ -115,6 +124,40 @@ func (dispatcher *EventDispatcher) Subscribe(eventType string, listener events.E // Publish an event to subscribers func (dispatcher *EventDispatcher) Publish(ctx context.Context, event cloudevents.Event) { + _ = dispatcher.publish(ctx, event) +} + +// SubscribeFilter registers a guard that runs before any subscriber sees an event. +func (dispatcher *EventDispatcher) SubscribeFilter(eventType string, filter events.EventFilter) { + dispatcher.filters[eventType] = filter +} + +func (dispatcher *EventDispatcher) publish(ctx context.Context, event cloudevents.Event) error { + allowed, err := dispatcher.shouldPublish(ctx, event) + if err != nil || !allowed { + return err + } + return dispatcher.publishSubscribers(ctx, event) +} + +func (dispatcher *EventDispatcher) shouldPublish(ctx context.Context, event cloudevents.Event) (bool, error) { + ctx, span, ctxLogger := dispatcher.tracer.StartWithLogger(ctx, dispatcher.logger) + defer span.End() + + filter, exists := dispatcher.filters[event.Type()] + if !exists { + return true, nil + } + allowed, err := filter(ctx, event) + if err != nil { + err = dispatcher.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot validate event [%s] with ID [%s]", event.Type(), event.ID())) + ctxLogger.Error(err) + return false, err + } + return allowed, nil +} + +func (dispatcher *EventDispatcher) publishSubscribers(ctx context.Context, event cloudevents.Event) error { ctx, span, ctxLogger := dispatcher.tracer.StartWithLogger(ctx, dispatcher.logger) defer span.End() @@ -125,21 +168,28 @@ 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 } var wg sync.WaitGroup + failures := make(chan error, len(subscribers)) 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())) + failures <- err } wg.Done() }(ctx, sub) } wg.Wait() + close(failures) + var listenerErrors []error + for err := range failures { + listenerErrors = append(listenerErrors, err) + } dispatcher.meter.Record( ctx, @@ -149,6 +199,10 @@ func (dispatcher *EventDispatcher) Publish(ctx context.Context, event cloudevent semconv.CloudeventsEventSpecVersion(event.SpecVersion()), ), ) + if err := errors.Join(listenerErrors...); err != nil { + return dispatcher.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot publish event [%s] with ID [%s]", event.Type(), event.ID())) + } + return nil } func (dispatcher *EventDispatcher) addCloudEventAttributes(span trace.Span, event cloudevents.Event) { @@ -166,12 +220,16 @@ func (dispatcher *EventDispatcher) createCloudTask(event cloudevents.Event) (*Pu return nil, stacktrace.Propagatef(err, "cannot marshall [%T] with ID [%s]", event, event.ID()) } - return &PushQueueTask{ + task := &PushQueueTask{ Method: http.MethodPost, URL: dispatcher.queueConfig.ConsumerEndpoint, Body: eventContent, Headers: map[string]string{ "x-api-key": dispatcher.queueConfig.UserAPIKey, }, - }, nil + } + if event.Type() == events.EventTypePhoneHeartbeatOffline { + task.ID = event.ID() + } + return task, nil } 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..fb05a3e6a --- /dev/null +++ b/api/pkg/services/event_dispatcher_service_test.go @@ -0,0 +1,61 @@ +package services + +import ( + "context" + "testing" + "time" + + "github.com/NdoleStudio/httpsms/pkg/events" + "github.com/NdoleStudio/httpsms/pkg/telemetry" + "github.com/NdoleStudio/stacktrace" + cloudevents "github.com/cloudevents/sdk-go/v2" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/metric/noop" +) + +func TestEventDispatcherDispatchSync_RetriesOnlyHeartbeatCheckFailures(t *testing.T) { + for _, eventType := range []string{events.EventTypePhoneHeartbeatCheck, events.EventTypePhoneHeartbeatOnline} { + t.Run(eventType, func(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + meter, err := noop.NewMeterProvider().Meter("test").Float64Histogram("dispatch") + require.NoError(t, err) + dispatcher := NewEventDispatcher(logger, tracer, meter, nil, PushQueueConfig{}) + failure := stacktrace.NewError("database unavailable") + dispatcher.Subscribe(eventType, func(context.Context, cloudevents.Event) error { return failure }) + event := cloudevents.NewEvent() + event.SetID("check-1") + event.SetSource("https://example.com") + event.SetType(eventType) + err = dispatcher.DispatchSync(context.Background(), event) + if eventType == events.EventTypePhoneHeartbeatCheck { + require.ErrorIs(t, err, failure) + } else { + require.NoError(t, err) + } + }) + } +} + +func TestEventDispatcherDispatchWithTimeout_DoesNotUseLocalFallbackForHeartbeatChecks(t *testing.T) { + for _, eventType := range []string{events.EventTypePhoneHeartbeatCheck, events.EventTypePhoneHeartbeatOffline, events.EventTypePhoneHeartbeatOnline} { + t.Run(eventType, func(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue := &heartbeatServiceQueue{err: context.DeadlineExceeded} + dispatcher := NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{}) + event := cloudevents.NewEvent() + event.SetID("check-1") + event.SetSource("https://example.com") + event.SetType(eventType) + queueID, err := dispatcher.DispatchWithTimeout(context.Background(), event, time.Hour) + if eventType != events.EventTypePhoneHeartbeatOnline { + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Empty(t, queueID) + } else { + require.NoError(t, err) + require.Equal(t, "local-check-1", queueID) + } + }) + } +} diff --git a/api/pkg/services/google_cloud_push_queue_service.go b/api/pkg/services/google_cloud_push_queue_service.go index 2208cf218..7ecca0250 100644 --- a/api/pkg/services/google_cloud_push_queue_service.go +++ b/api/pkg/services/google_cloud_push_queue_service.go @@ -12,6 +12,8 @@ import ( "cloud.google.com/go/cloudtasks/apiv2/cloudtaskspb" "github.com/NdoleStudio/httpsms/pkg/telemetry" "github.com/NdoleStudio/stacktrace" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -75,11 +77,18 @@ func (queue *googlePushQueue) enqueueImpl(ctx context.Context, task *PushQueueTa // Add a payload message if one is present. req.Task.GetHttpRequest().Body = task.Body + if task.ID != "" { + req.Task.Name = queue.queueConfig.Name + "/tasks/" + task.ID + } requestCtx, cancel := context.WithTimeout(ctx, time.Second) defer cancel() queueTask, err := queue.client.CreateTask(requestCtx, req) + if status.Code(err) == codes.AlreadyExists && req.Task.Name != "" { + ctxLogger.Info(fmt.Sprintf("task [%s] was already queued", req.Task.Name)) + return req.Task.Name, nil + } if err != nil { ctxLogger.Error(stacktrace.Propagatef(err, "cannot schedule task [%s] to URL [%s]", string(task.Body), task.URL)) return queueID, queue.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot schedule task [%s] to URL [%s]", string(task.Body), task.URL)) diff --git a/api/pkg/services/google_cloud_push_queue_service_test.go b/api/pkg/services/google_cloud_push_queue_service_test.go new file mode 100644 index 000000000..2f20be410 --- /dev/null +++ b/api/pkg/services/google_cloud_push_queue_service_test.go @@ -0,0 +1,122 @@ +package services + +import ( + "context" + "encoding/json" + "net" + "sync" + "testing" + + cloudtasks "cloud.google.com/go/cloudtasks/apiv2" + "cloud.google.com/go/cloudtasks/apiv2/cloudtaskspb" + "github.com/NdoleStudio/httpsms/pkg/events" + "github.com/NdoleStudio/httpsms/pkg/telemetry" + cloudevents "github.com/cloudevents/sdk-go/v2" + "github.com/google/uuid" + "github.com/stretchr/testify/require" + "google.golang.org/api/option" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" + "google.golang.org/grpc/test/bufconn" + "google.golang.org/protobuf/proto" +) + +type heartbeatCloudTaskServer struct { + cloudtaskspb.UnimplementedCloudTasksServer + mu sync.Mutex + tasks []*cloudtaskspb.Task + rejectChecks bool +} + +func (server *heartbeatCloudTaskServer) CreateTask(_ context.Context, request *cloudtaskspb.CreateTaskRequest) (*cloudtaskspb.Task, error) { + server.mu.Lock() + defer server.mu.Unlock() + + var event cloudevents.Event + if err := json.Unmarshal(request.Task.GetHttpRequest().Body, &event); err != nil { + return nil, status.Error(codes.InvalidArgument, err.Error()) + } + if server.rejectChecks && event.Type() == events.EventTypePhoneHeartbeatCheck { + return nil, status.Error(codes.Unavailable, "heartbeat queue unavailable") + } + for _, existing := range server.tasks { + if existing.Name == request.Task.Name { + return nil, status.Error(codes.AlreadyExists, "task already exists") + } + } + task := proto.Clone(request.Task).(*cloudtaskspb.Task) + if task.Name == "" { + task.Name = request.Parent + "/tasks/" + uuid.NewString() + } + server.tasks = append(server.tasks, task) + return task, nil +} + +func (server *heartbeatCloudTaskServer) queuedEvents(t *testing.T, eventType string) []cloudevents.Event { + t.Helper() + server.mu.Lock() + defer server.mu.Unlock() + + var result []cloudevents.Event + for _, task := range server.tasks { + var event cloudevents.Event + require.NoError(t, json.Unmarshal(task.GetHttpRequest().Body, &event)) + if event.Type() == eventType { + result = append(result, event) + } + } + return result +} + +func newHeartbeatCloudTaskQueue(t *testing.T, logger telemetry.Logger, tracer telemetry.Tracer) (PushQueue, *heartbeatCloudTaskServer) { + t.Helper() + listener := bufconn.Listen(1024 * 1024) + server := &heartbeatCloudTaskServer{} + // Both the server and custom dialer use bufconn only; no network socket is opened. + rpcServer := grpc.NewServer() // nosemgrep: go.grpc.security.grpc-server-insecure-connection.grpc-server-insecure-connection + cloudtaskspb.RegisterCloudTasksServer(rpcServer, server) + go func() { _ = rpcServer.Serve(listener) }() + connection, err := grpc.NewClient("passthrough:///cloudtasks", + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { + return listener.DialContext(ctx) + }), + grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + client, err := cloudtasks.NewClient(context.Background(), option.WithGRPCConn(connection)) + require.NoError(t, err) + t.Cleanup(func() { + _ = client.Close() + _ = connection.Close() + rpcServer.Stop() + _ = listener.Close() + }) + return NewGooglePushQueue(logger, tracer, client, PushQueueConfig{ + Name: "projects/test/locations/test/queues/heartbeats", + }), server +} + +func TestGooglePushQueue_DeduplicatesOfflineEventsWithoutDeduplicatingOtherEvents(t *testing.T) { + for _, eventType := range []string{events.EventTypePhoneHeartbeatOffline, events.EventTypePhoneHeartbeatOnline} { + t.Run(eventType, func(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue, server := newHeartbeatCloudTaskQueue(t, logger, tracer) + dispatcher := NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{ + ConsumerEndpoint: "https://example.com/events", + }) + event := cloudevents.NewEvent() + event.SetID(uuid.NewString()) + event.SetType(eventType) + event.SetSource("https://example.com") + require.NoError(t, dispatcher.Dispatch(context.Background(), event)) + require.NoError(t, dispatcher.Dispatch(context.Background(), event)) + want := 2 + if eventType == events.EventTypePhoneHeartbeatOffline { + want = 1 + } + require.Len(t, server.queuedEvents(t, eventType), want) + }) + } +} diff --git a/api/pkg/services/heartbeat_service.go b/api/pkg/services/heartbeat_service.go index d6c442347..14cd19791 100644 --- a/api/pkg/services/heartbeat_service.go +++ b/api/pkg/services/heartbeat_service.go @@ -3,6 +3,7 @@ package services import ( "context" "fmt" + "strings" "time" "github.com/NdoleStudio/httpsms/pkg/events" @@ -20,6 +21,8 @@ import ( const ( // select id, a.timestamp, a.owner, a.timestamp - (SELECT timestamp from heartbeats b where b.timestamp < a.timestamp and a.owner = b.owner and a.user_id = b.user_id order by b.timestamp desc limit 1) as diff from heartbeats a; heartbeatCheckInterval = 16 * time.Minute + heartbeatClaimLease = entities.HeartbeatCheckClaimLease + legacyHeartbeatCheckID = "legacy" ) // HeartbeatService is handles heartbeat requests @@ -124,7 +127,10 @@ func (service *HeartbeatService) Store(ctx context.Context, params HeartbeatStor return heartbeat, nil } - if monitor.PhoneIsOffline() { + wasOffline := service.recordHeartbeatArrival(ctx, heartbeat, monitor) + service.restartMonitorOnHeartbeat(ctx, heartbeat, monitor, params.Source) + + if wasOffline { ctxLogger.Info(fmt.Sprintf("phone with monitor ID [%s] was offline for user [%s]", monitor.ID, monitor.UserID)) service.handleHeartbeatWhenPhoneWasOffline(ctx, params.Source, heartbeat, monitor) } @@ -132,6 +138,38 @@ func (service *HeartbeatService) Store(ctx context.Context, params HeartbeatStor return heartbeat, nil } +func (service *HeartbeatService) recordHeartbeatArrival(ctx context.Context, heartbeat *entities.Heartbeat, monitor *entities.HeartbeatMonitor) bool { + ctx, span, ctxLogger := service.tracer.StartWithLogger(ctx, service.logger) + defer span.End() + + current := monitor + for attempt := 0; attempt < 3; attempt++ { + latest, err := service.repository.Last(ctx, heartbeat.UserID, heartbeat.Owner) + if err != nil { + ctxLogger.Error(service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot load saved heartbeat version for monitor [%s]", monitor.ID))) + return monitor.PhoneIsOffline() + } + previous, err := service.monitorRepository.RecordHeartbeat(ctx, monitor.UserID, monitor.ID, current.LastHeartbeatID, latest.ID) + if err != nil { + ctxLogger.Error(service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot record saved heartbeat version for monitor [%s]", monitor.ID))) + return monitor.PhoneIsOffline() + } + if previous != nil { + return previous.PhoneIsOffline() + } + current, err = service.monitorRepository.Load(ctx, monitor.UserID, monitor.Owner) + if stacktrace.GetCode(err) == repositories.ErrCodeNotFound || (err == nil && current.ID != monitor.ID) { + return false + } + if err != nil { + ctxLogger.Error(service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot reload heartbeat version for monitor [%s]", monitor.ID))) + return monitor.PhoneIsOffline() + } + } + ctxLogger.Error(service.tracer.WrapErrorSpan(span, stacktrace.NewErrorf("heartbeat version repeatedly changed for monitor [%s]; saved heartbeat retained", monitor.ID))) + return monitor.PhoneIsOffline() +} + // HeartbeatMonitorStoreParams are parameters for creating a new entities.Heartbeat type HeartbeatMonitorStoreParams struct { Owner string @@ -194,7 +232,7 @@ func (service *HeartbeatService) StoreMonitor(ctx context.Context, params *Heart MonitorID: monitor.ID, Source: params.Source, } - if err = service.scheduleHeartbeatCheck(ctx, time.Now().UTC(), monitorParams); err != nil { + if err = service.scheduleHeartbeatCheck(ctx, time.Now().UTC(), monitorParams, monitor, true); err != nil { return nil, service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot schedule healthcheck for monitor [%s] with owner [%s] and userID [%s]", monitor.ID, params.Owner, params.UserID)) } @@ -269,6 +307,7 @@ type HeartbeatMonitorParams struct { PhoneID uuid.UUID UserID entities.UserID Source string + CheckID string } // Monitor the heartbeats of an owner and phone number @@ -277,27 +316,165 @@ func (service *HeartbeatService) Monitor(ctx context.Context, params *HeartbeatM defer span.End() ctxLogger := service.tracer.CtxLogger(service.logger, span) - monitor, err := service.monitorRepository.Load(ctx, params.UserID, params.Owner) - if err != nil && stacktrace.GetCode(err) == repositories.ErrCodeNotFound { + if stacktrace.GetCode(err) == repositories.ErrCodeNotFound { ctxLogger.Info(fmt.Sprintf("heartbeat monitor does not exist for owner [%s] and user [%s]", params.Owner, params.UserID)) 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 load heartbeat monitor for check")) } - // Update params in case of ID duplicate + retryID, canRun, err := service.heartbeatCheckOwnership(ctx, monitor, params) + if err != nil || !canRun { + return err + } params.PhoneID = monitor.PhoneID params.MonitorID = monitor.ID + // An empty marker can follow an interrupted stop; history still gates any restart. + heartbeat, err := service.repository.Last(ctx, params.UserID, params.Owner) + if stacktrace.GetCode(err) == repositories.ErrCodeNotFound { + return service.stopHeartbeatCheck(ctx, params, monitor) + } + if err != nil { + return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot fetch heartbeat history for check")) + } + return service.runHeartbeatCheck(ctx, params, monitor, heartbeat, retryID) +} + +func (service *HeartbeatService) restartMonitorOnHeartbeat(ctx context.Context, heartbeat *entities.Heartbeat, monitor *entities.HeartbeatMonitor, source string) { + ctx, span, ctxLogger := service.tracer.StartWithLogger(ctx, service.logger) + defer span.End() + + if !monitor.RequiresCheck() { + return + } + latest, err := service.repository.Last(ctx, heartbeat.UserID, heartbeat.Owner) + if stacktrace.GetCode(err) == repositories.ErrCodeNotFound { + return + } + if err != nil { + ctxLogger.Error(service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot validate heartbeat history before restarting monitor [%s]", monitor.ID))) + return + } + params := &HeartbeatMonitorParams{ + Owner: monitor.Owner, + PhoneID: monitor.PhoneID, + UserID: monitor.UserID, + MonitorID: monitor.ID, + Source: source, + } + if err := service.scheduleHeartbeatCheck(ctx, latest.Timestamp, params, monitor, true); err != nil { + ctxLogger.Error(service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot restart heartbeat check for monitor [%s]", monitor.ID))) + } +} + +func (service *HeartbeatService) heartbeatCheckOwnership(ctx context.Context, monitor *entities.HeartbeatMonitor, params *HeartbeatMonitorParams) (string, bool, error) { + _, span := service.tracer.Start(ctx) + defer span.End() + + if params.CheckID != "" && params.MonitorID != monitor.ID { + return "", false, nil + } + incomingID := params.CheckID + if incomingID == "" { + incomingID = legacyHeartbeatCheckID + } + // Claims retain the incoming check ID so redelivery can recover an interrupted worker. + claim := strings.SplitN(monitor.QueueID, ":", 3) + if len(claim) == 3 && (claim[0] == "running" || claim[0] == "pending") { + deliveredSuccessor := claim[0] == "pending" && params.CheckID != "" && claim[2] == params.CheckID + restartOnly := claim[0] == "pending" && claim[1] == "" + if !deliveredSuccessor && !restartOnly && claim[1] != incomingID { + return "", false, nil + } + if !deliveredSuccessor && monitor.UpdatedAt.Add(heartbeatClaimLease).After(time.Now()) { + return "", false, service.tracer.WrapErrorSpan(span, stacktrace.NewErrorf("heartbeat check for monitor [%s] is still claimed; retry after its lease", monitor.ID)) + } + retryID := "" + if params.CheckID != "" { + retryID = "check:" + params.CheckID + } + return retryID, true, nil + } + if params.CheckID == "" && strings.HasPrefix(monitor.QueueID, "check:") { + return "", false, nil + } + if params.CheckID != "" && monitor.QueueID != "" && params.CheckID != strings.TrimPrefix(monitor.QueueID, "check:") { + return "", false, nil + } + return monitor.QueueID, true, nil +} + +func (service *HeartbeatService) stopHeartbeatCheck(ctx context.Context, params *HeartbeatMonitorParams, monitor *entities.HeartbeatMonitor) error { + ctx, span, ctxLogger := service.tracer.StartWithLogger(ctx, service.logger) + defer span.End() + + ctxLogger.Info(fmt.Sprintf("heartbeat history does not exist for userID [%s] and owner [%s], stopping check", params.UserID, params.Owner)) + stopped, err := service.monitorRepository.CompareAndSwapQueueID(ctx, monitor.ID, monitor.QueueID, "") + if err != nil { + return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot clear heartbeat check queue for monitor [%s]", monitor.ID)) + } + if !stopped { + return service.tracer.WrapErrorSpan(span, stacktrace.NewErrorf("heartbeat stop ownership changed for monitor [%s]; retry to verify its restart", monitor.ID)) + } + return service.restartAfterHeartbeatStop(ctx, params) +} + +func (service *HeartbeatService) restartAfterHeartbeatStop(ctx context.Context, params *HeartbeatMonitorParams) error { + ctx, span := service.tracer.Start(ctx) + defer span.End() + + // A heartbeat may have been saved while its Store call still saw the old active marker. heartbeat, err := service.repository.Last(ctx, params.UserID, params.Owner) + if stacktrace.GetCode(err) == repositories.ErrCodeNotFound { + 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 service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot recheck heartbeat history after stopping monitor [%s]", params.MonitorID)) + } + monitor, err := service.monitorRepository.Load(ctx, params.UserID, params.Owner) + if stacktrace.GetCode(err) == repositories.ErrCodeNotFound { return nil } + if err != nil { + return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot reload stopped heartbeat monitor [%s]", params.MonitorID)) + } + if !monitor.RequiresCheck() { + return service.requireHeartbeatCheckCommit(ctx, monitor, params.CheckID) + } + params.PhoneID = monitor.PhoneID + params.MonitorID = monitor.ID + return service.scheduleHeartbeatCheck(ctx, heartbeat.Timestamp, params, monitor, true) +} + +func (service *HeartbeatService) runHeartbeatCheck(ctx context.Context, params *HeartbeatMonitorParams, monitor *entities.HeartbeatMonitor, heartbeat *entities.Heartbeat, retryID string) error { + ctx, span := service.tracer.Start(ctx) + defer span.End() + + incomingID := params.CheckID + if incomingID == "" { + incomingID = legacyHeartbeatCheckID + } + processingID := "running:" + incomingID + ":" + uuid.NewString() + claimed, err := service.monitorRepository.CompareAndSwapQueueID(ctx, monitor.ID, monitor.QueueID, processingID) + if err != nil { + return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot claim running heartbeat check for monitor [%s]", monitor.ID)) + } + if !claimed { + return service.tracer.WrapErrorSpan(span, stacktrace.NewErrorf("heartbeat check ownership changed for monitor [%s]; retry to verify its successor", monitor.ID)) + } + monitor.QueueID = processingID + // Restore the incoming check on failure so the task queue can retry it, without stealing a successor. + defer service.releaseHeartbeatCheck(ctx, monitor.ID, processingID, retryID) + return service.checkHeartbeatAvailability(ctx, params, monitor, heartbeat) +} + +func (service *HeartbeatService) checkHeartbeatAvailability(ctx context.Context, params *HeartbeatMonitorParams, monitor *entities.HeartbeatMonitor, heartbeat *entities.Heartbeat) error { + ctx, span, ctxLogger := service.tracer.StartWithLogger(ctx, service.logger) + defer span.End() // 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) { @@ -305,12 +482,48 @@ func (service *HeartbeatService) Monitor(ctx context.Context, params *HeartbeatM 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) + heartbeat, owned, err := service.refreshHeartbeatCheck(ctx, params, monitor) + if err != nil || !owned { + return err } + age := time.Since(heartbeat.Timestamp) + failed := age > heartbeatCheckInterval*4 && age < heartbeatCheckInterval*5 && monitor.PhoneOnline + nextTimestamp := heartbeat.Timestamp + if failed { + nextTimestamp = time.Now().UTC() + if err := service.handleFailedMonitor(ctx, heartbeat, params); err != nil { + return err + } + } + return service.scheduleHeartbeatCheck(ctx, nextTimestamp, params, monitor, false) +} - return service.scheduleHeartbeatCheck(ctx, heartbeat.Timestamp, params) +func (service *HeartbeatService) refreshHeartbeatCheck(ctx context.Context, params *HeartbeatMonitorParams, monitor *entities.HeartbeatMonitor) (*entities.Heartbeat, bool, error) { + ctx, span := service.tracer.Start(ctx) + defer span.End() + + current, err := service.monitorRepository.Load(ctx, params.UserID, params.Owner) + if stacktrace.GetCode(err) == repositories.ErrCodeNotFound { + return nil, false, nil + } + if err != nil { + return nil, false, service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot revalidate running heartbeat monitor [%s]", monitor.ID)) + } + if current.ID != monitor.ID { + return nil, false, nil + } + if current.QueueID != monitor.QueueID { + return nil, false, service.requireHeartbeatCheckCommit(ctx, current, params.CheckID) + } + heartbeat, err := service.repository.Last(ctx, params.UserID, params.Owner) + if stacktrace.GetCode(err) == repositories.ErrCodeNotFound { + return nil, false, service.stopHeartbeatCheck(ctx, params, monitor) + } + if err != nil { + return nil, false, service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot revalidate heartbeat history for monitor [%s]", monitor.ID)) + } + monitor.PhoneOnline = current.PhoneOnline + return heartbeat, true, nil } func (service *HeartbeatService) handleMissedMonitor(ctx context.Context, lastTimestamp time.Time, params *HeartbeatMonitorParams) { @@ -335,20 +548,16 @@ func (service *HeartbeatService) handleMissedMonitor(ctx context.Context, lastTi } } -func (service *HeartbeatService) handleFailedMonitor(ctx context.Context, lastTimestamp time.Time, params *HeartbeatMonitorParams) error { +func (service *HeartbeatService) handleFailedMonitor(ctx context.Context, heartbeat *entities.Heartbeat, params *HeartbeatMonitorParams) error { ctx, span, ctxLogger := service.tracer.StartWithLogger(ctx, service.logger) defer span.End() - err := service.scheduleHeartbeatCheck(ctx, time.Now().UTC(), params) - if err != nil { - return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot schedule healthcheck for monitor with owner [%s] and userID [%s]", params.Owner, params.UserID)) - } - - event, err := service.createPhoneHeartbeatOfflineEvent(params.Source, &events.PhoneHeartbeatOfflinePayload{ + event, err := service.createPhoneHeartbeatOfflineEvent(params.Source, params.CheckID, &events.PhoneHeartbeatOfflinePayload{ PhoneID: params.PhoneID, + HeartbeatID: heartbeat.ID, UserID: params.UserID, MonitorID: params.MonitorID, - LastHeartbeatTimestamp: lastTimestamp, + LastHeartbeatTimestamp: heartbeat.Timestamp, Timestamp: time.Now().UTC(), Owner: params.Owner, }) @@ -364,32 +573,132 @@ func (service *HeartbeatService) handleFailedMonitor(ctx context.Context, lastTi return nil } -func (service *HeartbeatService) scheduleHeartbeatCheck(ctx context.Context, lastTimestamp time.Time, params *HeartbeatMonitorParams) error { - ctx, span, ctxLogger := service.tracer.StartWithLogger(ctx, service.logger) +// ApplyOfflineHeartbeat rejects stale events before any notification subscriber runs. +func (service *HeartbeatService) ApplyOfflineHeartbeat(ctx context.Context, payload *events.PhoneHeartbeatOfflinePayload) (bool, error) { + ctx, span := service.tracer.Start(ctx) defer span.End() - event, err := service.createPhoneHeartbeatCheckEvent(params.Source, &events.PhoneHeartbeatCheckPayload{ - PhoneID: params.PhoneID, - UserID: params.UserID, - MonitorID: params.MonitorID, - ScheduledAt: lastTimestamp.Add(heartbeatCheckInterval), - Owner: params.Owner, - }) + monitor, err := service.monitorRepository.Load(ctx, payload.UserID, payload.Owner) + if stacktrace.GetCode(err) == repositories.ErrCodeNotFound { + return false, nil + } + if err != nil { + return false, service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot load monitor for offline heartbeat [%s]", payload.MonitorID)) + } + if monitor.ID != payload.MonitorID || (payload.HeartbeatID == uuid.Nil && monitor.LastHeartbeatID != uuid.Nil) { + return false, nil + } + heartbeat, err := service.repository.Last(ctx, payload.UserID, payload.Owner) + if stacktrace.GetCode(err) == repositories.ErrCodeNotFound { + return false, nil + } + if err != nil { + return false, service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot validate offline heartbeat for monitor [%s]", monitor.ID)) + } + if !heartbeat.Timestamp.Equal(payload.LastHeartbeatTimestamp) || (payload.HeartbeatID != uuid.Nil && heartbeat.ID != payload.HeartbeatID) { + return false, nil + } + applied, err := service.monitorRepository.UpdatePhoneOfflineForHeartbeat(ctx, payload.UserID, monitor.ID, monitor.LastHeartbeatID, payload.HeartbeatID) if err != nil { + return false, service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot apply offline heartbeat version for monitor [%s]", monitor.ID)) + } + return applied, nil +} + +func (service *HeartbeatService) scheduleHeartbeatCheck(ctx context.Context, lastTimestamp time.Time, params *HeartbeatMonitorParams, monitor *entities.HeartbeatMonitor, restart bool) error { + ctx, span := service.tracer.Start(ctx) + defer span.End() + + checkID := uuid.NewString() + event, err := service.createPhoneHeartbeatCheckEvent(params, lastTimestamp, checkID) + if err != nil { + if restart { + service.releaseHeartbeatCheck(ctx, monitor.ID, monitor.QueueID, "") + } return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot create event when phone monitor failed")) } + incomingID := params.CheckID + if !restart && incomingID == "" { + incomingID = legacyHeartbeatCheckID + } + pendingID := "pending:" + incomingID + ":" + checkID + claimed, err := service.monitorRepository.CompareAndSwapQueueID(ctx, monitor.ID, monitor.QueueID, pendingID) + if err != nil { + return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot claim heartbeat check for monitor [%s]", monitor.ID)) + } + if !claimed { + return service.verifyHeartbeatCheckCommit(ctx, params) + } + return service.enqueueHeartbeatCheck(ctx, event, monitor, pendingID, checkID, restart) +} + +func (service *HeartbeatService) enqueueHeartbeatCheck(ctx context.Context, event cloudevents.Event, monitor *entities.HeartbeatMonitor, pendingID, checkID string, restart bool) error { + ctx, span, ctxLogger := service.tracer.StartWithLogger(ctx, service.logger) + defer span.End() + queueID, err := service.dispatcher.DispatchWithTimeout(ctx, event, heartbeatCheckInterval) if err != nil { - return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot dispatch event [%s] for heartbeat monitor with phone id [%s]", event.Type(), params.PhoneID)) + releaseID := monitor.QueueID + if restart { + releaseID = "" + } + service.releaseHeartbeatCheck(ctx, monitor.ID, pendingID, releaseID) + return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot dispatch event [%s] for heartbeat monitor with phone id [%s]", event.Type(), monitor.PhoneID)) + } + + if err := service.finalizeHeartbeatCheck(ctx, monitor.ID, pendingID, checkID); err != nil { + return err + } + ctxLogger.Info(fmt.Sprintf("heartbeat check scheduled for monitor with id [%s] and phone id [%s] and queue id [%s] for user [%s]", monitor.ID, monitor.PhoneID, queueID, monitor.UserID)) + + return nil +} + +func (service *HeartbeatService) finalizeHeartbeatCheck(ctx context.Context, monitorID uuid.UUID, pendingID, checkID string) error { + ctx, span := service.tracer.Start(ctx) + defer span.End() + + committed, err := service.monitorRepository.CompareAndSwapQueueID(ctx, monitorID, pendingID, "check:"+checkID) + if err != nil { + return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot finalize queued heartbeat check for monitor [%s]", monitorID)) + } + if !committed { + return service.tracer.WrapErrorSpan(span, stacktrace.NewErrorf("heartbeat finalization ownership changed for monitor [%s]; retry delivery", monitorID)) + } + return nil +} + +func (service *HeartbeatService) releaseHeartbeatCheck(ctx context.Context, monitorID uuid.UUID, ownedID, retryID string) { + ctx, span, ctxLogger := service.tracer.StartWithLogger(ctx, service.logger) + defer span.End() + + if _, err := service.monitorRepository.CompareAndSwapQueueID(ctx, monitorID, ownedID, retryID); err != nil { + ctxLogger.Error(service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot release heartbeat check claim for monitor [%s]", monitorID))) } +} + +func (service *HeartbeatService) verifyHeartbeatCheckCommit(ctx context.Context, params *HeartbeatMonitorParams) error { + ctx, span := service.tracer.Start(ctx) + defer span.End() - 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)) + monitor, err := service.monitorRepository.Load(ctx, params.UserID, params.Owner) + if stacktrace.GetCode(err) == repositories.ErrCodeNotFound { + return nil + } + if err != nil { + return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot verify successor for heartbeat monitor [%s]", params.MonitorID)) } + return service.requireHeartbeatCheckCommit(ctx, monitor, params.CheckID) +} - 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)) +func (service *HeartbeatService) requireHeartbeatCheckCommit(ctx context.Context, monitor *entities.HeartbeatMonitor, incomingCheckID string) error { + _, span := service.tracer.Start(ctx) + defer span.End() + if !strings.HasPrefix(monitor.QueueID, "check:") || strings.TrimPrefix(monitor.QueueID, "check:") == incomingCheckID { + return service.tracer.WrapErrorSpan(span, stacktrace.NewErrorf("heartbeat successor for monitor [%s] is not committed; retry delivery", monitor.ID)) + } return nil } @@ -397,10 +706,23 @@ func (service *HeartbeatService) createPhoneHeartbeatMissedEvent(source string, return service.createEvent(events.PhoneHeartbeatMissed, source, payload) } -func (service *HeartbeatService) createPhoneHeartbeatOfflineEvent(source string, payload *events.PhoneHeartbeatOfflinePayload) (cloudevents.Event, error) { - return service.createEvent(events.EventTypePhoneHeartbeatOffline, source, payload) +func (service *HeartbeatService) createPhoneHeartbeatOfflineEvent(source, checkID string, payload *events.PhoneHeartbeatOfflinePayload) (cloudevents.Event, error) { + event, err := service.createEvent(events.EventTypePhoneHeartbeatOffline, source, payload) + if err != nil { + return event, err + } + // Queue-level deduplication survives retries before the offline listener updates phone status. + event.SetID(uuid.NewSHA1(payload.MonitorID, []byte(checkID+":"+payload.HeartbeatID.String()+":"+payload.LastHeartbeatTimestamp.UTC().Format(time.RFC3339Nano))).String()) + return event, nil } -func (service *HeartbeatService) createPhoneHeartbeatCheckEvent(source string, payload *events.PhoneHeartbeatCheckPayload) (cloudevents.Event, error) { - return service.createEvent(events.EventTypePhoneHeartbeatCheck, source, payload) +func (service *HeartbeatService) createPhoneHeartbeatCheckEvent(params *HeartbeatMonitorParams, lastTimestamp time.Time, checkID string) (cloudevents.Event, error) { + return service.createEvent(events.EventTypePhoneHeartbeatCheck, params.Source, &events.PhoneHeartbeatCheckPayload{ + PhoneID: params.PhoneID, + UserID: params.UserID, + MonitorID: params.MonitorID, + ScheduledAt: lastTimestamp.Add(heartbeatCheckInterval), + Owner: params.Owner, + CheckID: checkID, + }) } diff --git a/api/pkg/services/heartbeat_service_test.go b/api/pkg/services/heartbeat_service_test.go new file mode 100644 index 000000000..22a659e7f --- /dev/null +++ b/api/pkg/services/heartbeat_service_test.go @@ -0,0 +1,923 @@ +package services + +import ( + "context" + "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/telemetry" + "github.com/NdoleStudio/stacktrace" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type heartbeatServiceRepository struct { + repositories.HeartbeatRepository + heartbeat *entities.Heartbeat + err error + afterLast func() +} + +func (repository *heartbeatServiceRepository) Last(context.Context, entities.UserID, string) (*entities.Heartbeat, error) { + heartbeat, err := repository.heartbeat, repository.err + if repository.afterLast != nil { + hook := repository.afterLast + repository.afterLast = nil + hook() + } + return heartbeat, err +} + +func (repository *heartbeatServiceRepository) Store(_ context.Context, heartbeat *entities.Heartbeat) error { + repository.heartbeat = heartbeat + repository.err = nil + return nil +} + +type heartbeatServiceMonitorRepository struct { + repositories.HeartbeatMonitorRepository + monitor *entities.HeartbeatMonitor + err error + afterLoad func() + beforeSwap func() + swapErr error + beforeRecord func() + recordErr error +} + +func (repository *heartbeatServiceMonitorRepository) Load(context.Context, entities.UserID, string) (*entities.HeartbeatMonitor, error) { + if repository.monitor == nil { + return nil, repository.err + } + monitor := *repository.monitor + if repository.afterLoad != nil { + hook := repository.afterLoad + repository.afterLoad = nil + hook() + } + return &monitor, repository.err +} + +func (repository *heartbeatServiceMonitorRepository) CompareAndSwapQueueID(_ context.Context, monitorID uuid.UUID, expected, queueID string) (bool, error) { + if repository.beforeSwap != nil { + hook := repository.beforeSwap + repository.beforeSwap = nil + hook() + } + if repository.swapErr != nil { + err := repository.swapErr + repository.swapErr = nil + return false, err + } + if repository.monitor.ID != monitorID || repository.monitor.QueueID != expected { + return false, nil + } + repository.monitor.QueueID = queueID + repository.monitor.UpdatedAt = time.Now().UTC() + return true, nil +} + +func (repository *heartbeatServiceMonitorRepository) UpdateQueueID(_ context.Context, _ uuid.UUID, queueID string) error { + repository.monitor.QueueID = queueID + repository.monitor.UpdatedAt = time.Now().UTC() + return nil +} + +func (repository *heartbeatServiceMonitorRepository) UpdatePhoneOnline(_ context.Context, _ entities.UserID, _ uuid.UUID, online bool) error { + repository.monitor.PhoneOnline = online + repository.monitor.UpdatedAt = time.Now().UTC() + return nil +} + +func (repository *heartbeatServiceMonitorRepository) RecordHeartbeat(_ context.Context, _ entities.UserID, _ uuid.UUID, expected, heartbeatID uuid.UUID) (*entities.HeartbeatMonitor, error) { + if repository.beforeRecord != nil { + hook := repository.beforeRecord + repository.beforeRecord = nil + hook() + } + if repository.recordErr != nil { + return nil, repository.recordErr + } + if repository.monitor.LastHeartbeatID != expected { + return nil, nil + } + previous := *repository.monitor + repository.monitor.LastHeartbeatID = heartbeatID + repository.monitor.PhoneOnline = true + return &previous, nil +} + +type heartbeatServiceQueue struct { + phoneNotificationEventQueue + err error + afterEnqueue func() +} + +func (queue *heartbeatServiceQueue) Enqueue(ctx context.Context, task *PushQueueTask, timeout time.Duration) (string, error) { + if queue.err != nil { + return "", queue.err + } + if _, err := queue.phoneNotificationEventQueue.Enqueue(ctx, task, timeout); err != nil { + return "", err + } + if queue.afterEnqueue != nil { + hook := queue.afterEnqueue + queue.afterEnqueue = nil + hook() + } + return "queue-1", nil +} + +func TestHeartbeatService_RestartsChecksAfterHistoryExpires(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue := &heartbeatServiceQueue{} + repository := &heartbeatServiceRepository{ + err: stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "heartbeat not found"), + } + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), + PhoneID: uuid.New(), + UserID: entities.UserID("user-1"), + Owner: "+18005550199", + QueueID: "expired-check", + UpdatedAt: time.Now().UTC(), + } + 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"} + + require.NoError(t, service.Monitor(context.Background(), params)) + assert.Empty(t, queue.events) + assert.Empty(t, monitor.QueueID) + + heartbeat, err := service.Store(context.Background(), HeartbeatStoreParams{ + UserID: monitor.UserID, Owner: monitor.Owner, Source: params.Source, Timestamp: time.Now().UTC(), + }) + require.NoError(t, err) + require.Len(t, queue.events, 2) + assert.Equal(t, events.EventTypePhoneHeartbeatCheck, queue.events[0].Type()) + assert.Equal(t, events.EventTypePhoneHeartbeatOnline, queue.events[1].Type()) + assert.True(t, monitor.PhoneOnline) + var payload events.PhoneHeartbeatCheckPayload + require.NoError(t, queue.events[0].DataAs(&payload)) + assert.Equal(t, "check:"+payload.CheckID, monitor.QueueID) + assert.NotEmpty(t, payload.CheckID) + assert.Equal(t, monitor.ID, payload.MonitorID) + assert.Equal(t, monitor.PhoneID, payload.PhoneID) + assert.Equal(t, monitor.UserID, payload.UserID) + assert.Equal(t, monitor.Owner, payload.Owner) + assert.WithinDuration(t, heartbeat.Timestamp.Add(heartbeatCheckInterval), payload.ScheduledAt, time.Millisecond) + + repository.heartbeat.Timestamp = time.Now().UTC().Add(-4*heartbeatCheckInterval - time.Minute) + params.CheckID = payload.CheckID + require.NoError(t, service.Monitor(context.Background(), params)) + require.Len(t, queue.events, 5) + assert.Equal(t, events.EventTypePhoneHeartbeatOffline, queue.events[3].Type()) +} + +func TestHeartbeatServiceStore_OnlyRestartsInactiveChecks(t *testing.T) { + for _, test := range []struct { + name string + updatedAt time.Time + wantCheck bool + queueID string + }{ + {name: "active monitor", updatedAt: time.Now().UTC()}, + {name: "stale monitor", updatedAt: time.Now().UTC().Add(-3 * time.Hour), wantCheck: true}, + {name: "fresh restart claim", updatedAt: time.Now().UTC(), queueID: "pending::new-check"}, + {name: "expired restart claim", updatedAt: time.Now().UTC().Add(-2 * heartbeatClaimLease), queueID: "pending::new-check", wantCheck: true}, + } { + t.Run(test.name, func(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + 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, + } + if test.queueID != "" { + monitor.QueueID = test.queueID + } + service := NewHeartbeatService(logger, tracer, &heartbeatServiceRepository{}, + &heartbeatServiceMonitorRepository{monitor: monitor}, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + + _, err := service.Store(context.Background(), HeartbeatStoreParams{ + UserID: monitor.UserID, Owner: monitor.Owner, Source: "https://example.com", Timestamp: time.Now().UTC(), + }) + require.NoError(t, err) + if test.wantCheck { + require.Len(t, queue.events, 1) + assert.Equal(t, events.EventTypePhoneHeartbeatCheck, queue.events[0].Type()) + } else { + assert.Empty(t, queue.events) + } + }) + } +} + +func TestHeartbeatServiceMonitor_StopsCheckingDeletedMonitor(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue := &heartbeatServiceQueue{} + monitorRepository := &heartbeatServiceMonitorRepository{ + err: stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "monitor not found"), + } + service := NewHeartbeatService(logger, tracer, &heartbeatServiceRepository{}, monitorRepository, + 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", + })) + assert.Empty(t, queue.events) +} + +func TestHeartbeatServiceStore_ReturnsHeartbeatWhenRestartFails(t *testing.T) { + for _, test := range []struct { + name string + online bool + stale bool + }{ + {name: "online phone", online: true}, + {name: "offline phone"}, + {name: "stale offline phone", stale: true}, + } { + t.Run(test.name, func(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue := &heartbeatServiceQueue{err: stacktrace.NewError("queue unavailable")} + repository := &heartbeatServiceRepository{} + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), PhoneID: uuid.New(), UserID: entities.UserID("user-1"), + Owner: "+18005550199", UpdatedAt: time.Now().UTC(), PhoneOnline: test.online, + } + if test.stale { + monitor.QueueID = "stale-check" + monitor.UpdatedAt = time.Now().UTC().Add(-3 * time.Hour) + } + service := NewHeartbeatService(logger, tracer, repository, + &heartbeatServiceMonitorRepository{monitor: monitor}, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + + heartbeat, err := service.Store(context.Background(), HeartbeatStoreParams{ + UserID: monitor.UserID, Owner: monitor.Owner, Source: "https://example.com", Timestamp: time.Now().UTC(), + }) + + require.NoError(t, err) + require.NotNil(t, heartbeat) + assert.Same(t, repository.heartbeat, heartbeat) + assert.Empty(t, monitor.QueueID) + assert.True(t, monitor.PhoneOnline) + queue.err = 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) + require.Len(t, queue.events, 1) + assert.Equal(t, events.EventTypePhoneHeartbeatCheck, queue.events[0].Type()) + }) + } +} + +func TestHeartbeatServiceStore_ClaimsOverlappingRestarts(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", PhoneOnline: true, UpdatedAt: time.Now().UTC(), + } + monitorRepository := &heartbeatServiceMonitorRepository{monitor: monitor} + service := NewHeartbeatService(logger, tracer, &heartbeatServiceRepository{}, monitorRepository, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + params := HeartbeatStoreParams{ + UserID: monitor.UserID, Owner: monitor.Owner, Source: "https://example.com", Timestamp: time.Now().UTC(), + } + monitorRepository.afterLoad = func() { + _, err := service.Store(context.Background(), params) + require.NoError(t, err) + } + _, err := service.Store(context.Background(), params) + require.NoError(t, err) + require.Len(t, queue.events, 1) +} + +func TestHeartbeatServiceMonitor_RestartsHeartbeatArrivingDuringStop(t *testing.T) { + for _, arrival := range []string{"before clear", "after clear", "restart wins"} { + t.Run(arrival, func(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue := &heartbeatServiceQueue{} + repository := &heartbeatServiceRepository{ + err: stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "heartbeat not found"), + } + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", + Owner: "+18005550199", PhoneOnline: true, QueueID: "current-check", UpdatedAt: time.Now().UTC(), + } + monitorRepository := &heartbeatServiceMonitorRepository{monitor: monitor} + if arrival == "restart wins" { + monitor.UpdatedAt = time.Now().UTC().Add(-3 * time.Hour) + } + service := NewHeartbeatService(logger, tracer, repository, monitorRepository, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + store := func() { + _, err := service.Store(context.Background(), HeartbeatStoreParams{ + UserID: monitor.UserID, Owner: monitor.Owner, Source: "https://example.com", Timestamp: time.Now().UTC(), + }) + require.NoError(t, err) + } + if arrival != "after clear" { + repository.afterLast = store + } else { + repository.afterLast = func() { repository.afterLast = store } + } + err := service.Monitor(context.Background(), &HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, + Source: "https://example.com", CheckID: monitor.QueueID, + }) + if arrival == "restart wins" { + require.Error(t, err) + } else { + require.NoError(t, err) + } + require.Len(t, queue.events, 1) + var payload events.PhoneHeartbeatCheckPayload + require.NoError(t, queue.events[0].DataAs(&payload)) + assert.Equal(t, "check:"+payload.CheckID, monitor.QueueID) + }) + } +} + +func TestHeartbeatServiceMonitor_RejectsSupersededAndOverlappingChecks(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", PhoneOnline: true, QueueID: "current-check", UpdatedAt: time.Now().UTC(), + } + repository := &heartbeatServiceRepository{heartbeat: &entities.Heartbeat{ + Timestamp: time.Now().UTC().Add(-4*heartbeatCheckInterval - time.Minute), + }} + monitorRepository := &heartbeatServiceMonitorRepository{monitor: monitor} + service := NewHeartbeatService(logger, tracer, repository, monitorRepository, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + params := HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, + Source: "https://example.com", CheckID: monitor.QueueID, + } + monitorRepository.beforeSwap = func() { + copy := params + require.NoError(t, service.Monitor(context.Background(), ©)) + } + require.Error(t, service.Monitor(context.Background(), ¶ms)) + require.Len(t, queue.events, 3) + assert.Equal(t, events.PhoneHeartbeatMissed, queue.events[0].Type()) + assert.Equal(t, events.EventTypePhoneHeartbeatOffline, queue.events[1].Type()) + assert.Equal(t, events.EventTypePhoneHeartbeatCheck, queue.events[2].Type()) + require.NoError(t, service.Monitor(context.Background(), ¶ms)) + assert.Len(t, queue.events, 3) +} + +func TestHeartbeatServiceMonitor_RetriesFailedSuccessor(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue := &heartbeatServiceQueue{err: stacktrace.NewError("queue unavailable")} + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", + Owner: "+18005550199", PhoneOnline: true, QueueID: "current-check", UpdatedAt: time.Now().UTC(), + } + service := NewHeartbeatService(logger, tracer, &heartbeatServiceRepository{heartbeat: &entities.Heartbeat{ + Timestamp: time.Now().UTC(), + }}, &heartbeatServiceMonitorRepository{monitor: monitor}, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + params := &HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, + Source: "https://example.com", CheckID: monitor.QueueID, + } + require.Error(t, service.Monitor(context.Background(), params)) + assert.Equal(t, params.CheckID, monitor.QueueID) + queue.err = nil + require.NoError(t, service.Monitor(context.Background(), params)) + require.Len(t, queue.events, 1) +} + +func TestHeartbeatServiceMonitor_RecoversInterruptedClaims(t *testing.T) { + for _, test := range []struct { + name string + marker string + expired bool + wantRetry bool + wantCheck bool + missingHistory bool + legacy bool + }{ + {name: "running lease active", marker: "running:current-check:claim", wantRetry: true}, + {name: "running lease expired", marker: "running:current-check:claim", expired: true, wantCheck: true}, + {name: "pending lease active", marker: "pending:current-check:next-check", wantRetry: true}, + {name: "pending lease expired", marker: "pending:current-check:next-check", expired: true, wantCheck: true}, + {name: "restart-only lease active", marker: "pending::next-check", wantRetry: true}, + {name: "restart-only lease expired", marker: "pending::next-check", expired: true, wantCheck: true}, + {name: "queued pending check", marker: "pending:previous-check:current-check", wantCheck: true}, + {name: "interrupted stop", wantCheck: true}, + {name: "interrupted stop without history", missingHistory: true}, + {name: "legacy running recovery", marker: "running:legacy:claim", expired: true, wantCheck: true, legacy: true}, + {name: "legacy pending recovery", marker: "pending:legacy:next-check", expired: true, wantCheck: true, legacy: true}, + {name: "legacy waits for restart-only claim", marker: "pending::next-check", wantRetry: true, legacy: true}, + {name: "legacy recovers abandoned restart-only claim", marker: "pending::next-check", expired: true, wantCheck: true, legacy: true}, + {name: "unrelated expired running check", marker: "running:other-check:claim", expired: true}, + } { + t.Run(test.name, func(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", PhoneOnline: true, QueueID: test.marker, UpdatedAt: time.Now().UTC(), + } + if test.expired { + monitor.UpdatedAt = time.Now().UTC().Add(-2 * heartbeatClaimLease) + } + repository := &heartbeatServiceRepository{heartbeat: &entities.Heartbeat{ + Timestamp: time.Now().UTC(), + }} + if test.missingHistory { + repository.err = stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "heartbeat not found") + } + service := NewHeartbeatService(logger, tracer, repository, &heartbeatServiceMonitorRepository{monitor: monitor}, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + params := &HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, + Source: "https://example.com", CheckID: "current-check", + } + if test.legacy { + params.CheckID = "" + } + err := service.Monitor(context.Background(), params) + if test.wantRetry { + require.Error(t, err) + } else { + require.NoError(t, err) + } + if test.wantCheck { + require.Len(t, queue.events, 1) + } else { + assert.Empty(t, queue.events) + assert.Equal(t, test.marker, monitor.QueueID) + } + }) + } +} + +func TestHeartbeatServiceMonitor_RetriesHistoryRecheckFailure(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue := &heartbeatServiceQueue{} + repository := &heartbeatServiceRepository{ + err: stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "heartbeat not found"), + } + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", + Owner: "+18005550199", PhoneOnline: true, QueueID: "current-check", UpdatedAt: time.Now().UTC(), + } + repository.afterLast = func() { repository.err = stacktrace.NewError("database unavailable") } + service := NewHeartbeatService(logger, tracer, repository, &heartbeatServiceMonitorRepository{monitor: monitor}, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + params := &HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, + Source: "https://example.com", CheckID: monitor.QueueID, + } + require.Error(t, service.Monitor(context.Background(), params)) + assert.Empty(t, monitor.QueueID) + assert.Empty(t, queue.events) + repository.heartbeat = &entities.Heartbeat{Timestamp: time.Now().UTC()} + repository.err = nil + require.NoError(t, service.Monitor(context.Background(), params)) + require.Len(t, queue.events, 1) +} + +func TestHeartbeatServiceMonitor_RetriesPendingRestartAfterStop(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", PhoneOnline: true, QueueID: "current-check", UpdatedAt: time.Now().UTC(), + } + repository := &heartbeatServiceRepository{ + err: stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "heartbeat not found"), + } + repository.afterLast = func() { + repository.heartbeat = &entities.Heartbeat{Timestamp: time.Now().UTC()} + repository.err = nil + repository.afterLast = func() { + monitor.QueueID = "pending::concurrent-restart" + monitor.UpdatedAt = time.Now().UTC() + } + } + service := NewHeartbeatService(logger, tracer, repository, &heartbeatServiceMonitorRepository{monitor: monitor}, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + params := &HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, + Source: "https://example.com", CheckID: monitor.QueueID, + } + require.Error(t, service.Monitor(context.Background(), params)) + assert.Equal(t, "pending::concurrent-restart", monitor.QueueID) + assert.Empty(t, queue.events) + monitor.UpdatedAt = time.Now().UTC().Add(-2 * heartbeatClaimLease) + require.NoError(t, service.Monitor(context.Background(), params)) + assert.Len(t, queue.events, 1) +} + +func TestHeartbeatServiceMonitor_RetriesUncommittedSuccessorOnClaimLoss(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", PhoneOnline: true, QueueID: "current-check", UpdatedAt: time.Now().UTC(), + } + monitorRepository := &heartbeatServiceMonitorRepository{monitor: monitor} + monitorRepository.beforeSwap = func() { + monitorRepository.beforeSwap = func() { + monitor.QueueID = "pending::concurrent-restart" + monitor.UpdatedAt = time.Now().UTC() + } + } + 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, MonitorID: monitor.ID, + Source: "https://example.com", CheckID: monitor.QueueID, + } + require.Error(t, service.Monitor(context.Background(), params)) + assert.Equal(t, "pending::concurrent-restart", monitor.QueueID) + assert.Empty(t, queue.events) + monitor.UpdatedAt = time.Now().UTC().Add(-2 * heartbeatClaimLease) + require.NoError(t, service.Monitor(context.Background(), params)) + assert.Len(t, queue.events, 1) +} + +func TestHeartbeatServiceMonitor_LegacyMigrationIsOneWay(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", PhoneOnline: true, QueueID: "legacy-task", UpdatedAt: time.Now().UTC(), + } + service := NewHeartbeatService(logger, tracer, &heartbeatServiceRepository{heartbeat: &entities.Heartbeat{ + Timestamp: time.Now().UTC(), + }}, &heartbeatServiceMonitorRepository{monitor: monitor}, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + params := &HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, Source: "https://example.com", + } + require.NoError(t, service.Monitor(context.Background(), params)) + require.Len(t, queue.events, 1) + var payload events.PhoneHeartbeatCheckPayload + require.NoError(t, queue.events[0].DataAs(&payload)) + assert.Equal(t, "check:"+payload.CheckID, monitor.QueueID) + require.NoError(t, service.Monitor(context.Background(), params)) + assert.Len(t, queue.events, 1) + params.CheckID = payload.CheckID + require.NoError(t, service.Monitor(context.Background(), params)) + assert.Len(t, queue.events, 2) +} + +func TestHeartbeatServiceMonitor_RetriesLostStopOwnership(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", PhoneOnline: true, QueueID: "current-check", UpdatedAt: time.Now().UTC(), + } + repository := &heartbeatServiceRepository{ + err: stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "heartbeat not found"), + } + monitorRepository := &heartbeatServiceMonitorRepository{monitor: monitor} + monitorRepository.beforeSwap = func() { monitor.QueueID = "" } + service := NewHeartbeatService(logger, tracer, repository, monitorRepository, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + params := &HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, + Source: "https://example.com", CheckID: monitor.QueueID, + } + require.Error(t, service.Monitor(context.Background(), params)) + assert.Empty(t, queue.events) + monitor.QueueID = params.CheckID + repository.err = nil + repository.heartbeat = &entities.Heartbeat{Timestamp: time.Now().UTC()} + require.NoError(t, service.Monitor(context.Background(), params)) + assert.Len(t, queue.events, 1) +} + +func TestHeartbeatServiceMonitor_DoesNotResurrectAcknowledgedStop(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue := &heartbeatServiceQueue{} + notFound := stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "heartbeat not found") + repository := &heartbeatServiceRepository{err: notFound} + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", + Owner: "+18005550199", PhoneOnline: true, QueueID: "check:current-check", UpdatedAt: time.Now().UTC(), + } + service := NewHeartbeatService(logger, tracer, repository, &heartbeatServiceMonitorRepository{monitor: monitor}, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + params := HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, + Source: "https://example.com", CheckID: "current-check", + } + repository.afterLast = func() { + repository.err = stacktrace.NewError("history recheck unavailable") + repository.afterLast = func() { + repository.err = notFound + duplicate := params + require.NoError(t, service.Monitor(context.Background(), &duplicate)) + } + } + require.Error(t, service.Monitor(context.Background(), ¶ms)) + assert.Empty(t, queue.events) + assert.True(t, monitor.RequiresCheck(), "an acknowledged stop must remain restartable") + _, err := service.Store(context.Background(), HeartbeatStoreParams{ + UserID: monitor.UserID, Owner: monitor.Owner, Source: params.Source, Timestamp: time.Now().UTC(), + }) + require.NoError(t, err) + require.Len(t, queue.events, 1) + assert.Equal(t, events.EventTypePhoneHeartbeatCheck, queue.events[0].Type()) +} + +func TestHeartbeatServiceMonitor_RetriesFailedFinalization(t *testing.T) { + for _, failure := range []string{"database error", "lease recovered by interrupted worker"} { + t.Run(failure, func(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", PhoneOnline: true, QueueID: "check:current-check", UpdatedAt: time.Now().UTC(), + } + monitorRepository := &heartbeatServiceMonitorRepository{monitor: monitor} + 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, MonitorID: monitor.ID, + Source: "https://example.com", CheckID: "current-check", + } + queue.afterEnqueue = func() { + if failure == "database error" { + monitorRepository.swapErr = stacktrace.NewError("finalization unavailable") + } else { + monitor.QueueID = "running:current-check:recovered-worker" + monitor.UpdatedAt = time.Now().UTC() + } + } + require.Error(t, service.Monitor(context.Background(), ¶ms), "incoming task must remain retryable") + require.Len(t, queue.events, 1) + monitor.UpdatedAt = time.Now().UTC().Add(-2 * heartbeatClaimLease) + require.NoError(t, service.Monitor(context.Background(), ¶ms)) + require.Len(t, queue.events, 2) + var successor events.PhoneHeartbeatCheckPayload + require.NoError(t, queue.events[1].DataAs(&successor)) + assert.Equal(t, "check:"+successor.CheckID, monitor.QueueID) + var abandoned events.PhoneHeartbeatCheckPayload + require.NoError(t, queue.events[0].DataAs(&abandoned)) + params.CheckID = abandoned.CheckID + require.NoError(t, service.Monitor(context.Background(), ¶ms)) + assert.Len(t, queue.events, 2, "superseded queued task must not create another chain") + }) + } +} + +func TestHeartbeatServiceMonitor_RejectsRestoredIncomingAsSuccessor(t *testing.T) { + for _, path := range []string{"scheduling claim loss", "post-stop recheck"} { + t.Run(path, func(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", PhoneOnline: true, QueueID: "check:current-check", UpdatedAt: time.Now().UTC(), + } + repository := &heartbeatServiceRepository{heartbeat: &entities.Heartbeat{Timestamp: time.Now().UTC()}} + monitorRepository := &heartbeatServiceMonitorRepository{monitor: monitor} + restoreIncoming := func() { + monitor.QueueID = "check:current-check" + monitor.UpdatedAt = time.Now().UTC() + } + if path == "scheduling claim loss" { + monitorRepository.beforeSwap = func() { monitorRepository.beforeSwap = restoreIncoming } + } else { + repository.err = stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "heartbeat not found") + repository.afterLast = func() { + repository.err = nil + repository.afterLast = restoreIncoming + } + } + + service := NewHeartbeatService(logger, tracer, repository, monitorRepository, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + params := &HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, + Source: "https://example.com", CheckID: "current-check", + } + require.Error(t, service.Monitor(context.Background(), params), "incoming marker is not a queued successor") + assert.Empty(t, queue.events) + require.NoError(t, service.Monitor(context.Background(), params)) + require.Len(t, queue.events, 1) + }) + } +} + +func TestHeartbeatServiceMonitor_DeduplicatesOfflineAlertOnSuccessorRetry(t *testing.T) { + for _, checkID := range []string{"current-check", ""} { + t.Run("incoming:"+checkID, func(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue, server := newHeartbeatCloudTaskQueue(t, logger, tracer) + server.rejectChecks = true + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", + Owner: "+18005550199", PhoneOnline: true, QueueID: "current-check", UpdatedAt: time.Now().UTC(), + } + service := NewHeartbeatService(logger, tracer, &heartbeatServiceRepository{ + heartbeat: &entities.Heartbeat{Timestamp: time.Now().UTC().Add(-4*heartbeatCheckInterval - time.Minute)}, + }, &heartbeatServiceMonitorRepository{monitor: monitor}, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{ConsumerEndpoint: "https://example.com/events"})) + params := &HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, + Source: "https://example.com", CheckID: checkID, + } + require.Error(t, service.Monitor(context.Background(), params)) + require.Len(t, server.queuedEvents(t, events.EventTypePhoneHeartbeatOffline), 1) + assert.True(t, monitor.PhoneOnline, "offline listener has not run yet") + server.mu.Lock() + server.rejectChecks = false + server.mu.Unlock() + require.NoError(t, service.Monitor(context.Background(), params)) + require.Len(t, server.queuedEvents(t, events.EventTypePhoneHeartbeatOffline), 1, "retry must not queue another email/webhook alert") + checks := server.queuedEvents(t, events.EventTypePhoneHeartbeatCheck) + require.Len(t, checks, 1) + var payload events.PhoneHeartbeatCheckPayload + require.NoError(t, checks[0].DataAs(&payload)) + assert.Equal(t, "check:"+payload.CheckID, monitor.QueueID) + }) + } +} + +func TestHeartbeatServiceMonitor_RevalidatesAfterMissedNotification(t *testing.T) { + for _, takeover := range []bool{false, true} { + t.Run(map[bool]string{false: "returning heartbeat", true: "lease takeover"}[takeover], func(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", + PhoneOnline: true, QueueID: "check:current-check", UpdatedAt: time.Now().UTC(), + } + repository := &heartbeatServiceRepository{heartbeat: &entities.Heartbeat{ + ID: uuid.New(), Timestamp: time.Now().UTC().Add(-4*heartbeatCheckInterval - time.Minute), + }} + service := NewHeartbeatService(logger, tracer, repository, &heartbeatServiceMonitorRepository{monitor: monitor}, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + params := HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, + Source: "https://example.com", CheckID: "current-check", + } + queue.afterEnqueue = func() { + monitor.UpdatedAt = time.Now().UTC().Add(-2 * heartbeatClaimLease) + _, err := service.Store(context.Background(), HeartbeatStoreParams{ + UserID: monitor.UserID, Owner: monitor.Owner, Source: params.Source, Timestamp: time.Now().UTC(), + }) + require.NoError(t, err) + if takeover { + competing := params + require.NoError(t, service.Monitor(context.Background(), &competing)) + } + } + require.NoError(t, service.Monitor(context.Background(), ¶ms)) + require.Len(t, queue.events, 2, "only the missed signal and one owned successor should be queued") + assert.Equal(t, events.EventTypePhoneHeartbeatCheck, queue.events[1].Type()) + var successor events.PhoneHeartbeatCheckPayload + require.NoError(t, queue.events[1].DataAs(&successor)) + assert.Equal(t, "check:"+successor.CheckID, monitor.QueueID) + assert.WithinDuration(t, repository.heartbeat.Timestamp.Add(heartbeatCheckInterval), successor.ScheduledAt, time.Millisecond) + }) + } +} + +func TestHeartbeatServiceStore_OverlappingArrivalsKeepLatestVersion(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue := &heartbeatServiceQueue{} + history := &heartbeatServiceRepository{} + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", Owner: "+18005550199", + PhoneOnline: true, QueueID: "check:active-check", UpdatedAt: time.Now().UTC(), + } + repository := &heartbeatServiceMonitorRepository{monitor: monitor} + service := NewHeartbeatService(logger, tracer, history, repository, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + params := HeartbeatStoreParams{UserID: monitor.UserID, Owner: monitor.Owner, Source: "https://example.com", Timestamp: time.Now().UTC()} + var newer *entities.Heartbeat + repository.beforeRecord = func() { + params.Timestamp = params.Timestamp.Add(time.Second) + var err error + newer, err = service.Store(context.Background(), params) + require.NoError(t, err) + } + older, err := service.Store(context.Background(), params) + require.NoError(t, err) + require.NotEqual(t, older.ID, newer.ID) + require.Equal(t, newer.ID, monitor.LastHeartbeatID, "a paused earlier arrival must not restore an older alert version") + assert.Empty(t, queue.events, "active monitoring must not be restarted by overlapping arrivals") +} + +func TestHeartbeatServiceStore_ReturnsSavedHeartbeatWhenVersionRecordFails(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue := &heartbeatServiceQueue{} + history := &heartbeatServiceRepository{} + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", Owner: "+18005550199", UpdatedAt: time.Now().UTC(), + } + repository := &heartbeatServiceMonitorRepository{monitor: monitor, recordErr: stacktrace.NewError("version update unavailable")} + service := NewHeartbeatService(logger, tracer, history, repository, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + heartbeat, err := service.Store(context.Background(), HeartbeatStoreParams{ + UserID: monitor.UserID, Owner: monitor.Owner, Source: "https://example.com", Timestamp: time.Now().UTC(), + }) + require.NoError(t, err) + require.Same(t, history.heartbeat, heartbeat) + require.True(t, monitor.PhoneOnline) + require.Len(t, queue.events, 2) + assert.Equal(t, events.EventTypePhoneHeartbeatCheck, queue.events[0].Type()) + assert.Equal(t, events.EventTypePhoneHeartbeatOnline, queue.events[1].Type()) +} + +func TestHeartbeatServiceMonitor_OfflineRetryNamesFollowHeartbeatIdentity(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue, server := newHeartbeatCloudTaskQueue(t, logger, tracer) + server.rejectChecks = true + first := &entities.Heartbeat{ID: uuid.New(), Timestamp: time.Now().UTC().Add(-65 * time.Minute)} + history := &heartbeatServiceRepository{heartbeat: first} + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", Owner: "+18005550199", + PhoneOnline: true, QueueID: "check:current-check", UpdatedAt: time.Now().UTC(), + } + service := NewHeartbeatService(logger, tracer, history, &heartbeatServiceMonitorRepository{monitor: monitor}, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{ConsumerEndpoint: "https://example.com/events"})) + params := &HeartbeatMonitorParams{ + UserID: monitor.UserID, Owner: monitor.Owner, MonitorID: monitor.ID, Source: "https://example.com", CheckID: "current-check", + } + require.Error(t, service.Monitor(context.Background(), params)) + second, err := service.Store(context.Background(), HeartbeatStoreParams{ + UserID: monitor.UserID, Owner: monitor.Owner, Source: params.Source, Timestamp: first.Timestamp, + }) + require.NoError(t, err) + server.mu.Lock() + server.rejectChecks = false + server.mu.Unlock() + require.NoError(t, service.Monitor(context.Background(), params)) + alerts := server.queuedEvents(t, events.EventTypePhoneHeartbeatOffline) + require.Len(t, alerts, 2, "a stale queued alert must not suppress a distinct heartbeat version with the same timestamp") + var firstPayload, secondPayload events.PhoneHeartbeatOfflinePayload + require.NoError(t, alerts[0].DataAs(&firstPayload)) + require.NoError(t, alerts[1].DataAs(&secondPayload)) + require.Equal(t, first.ID, firstPayload.HeartbeatID) + require.Equal(t, second.ID, secondPayload.HeartbeatID) +} + +func TestHeartbeatServiceStore_DoesNotRestartHistoryRemovedAfterSaving(t *testing.T) { + logger := &phoneNotificationLogger{} + tracer := telemetry.NewOtelLogger("test", logger) + queue := &heartbeatServiceQueue{} + history := &heartbeatServiceRepository{} + monitor := &entities.HeartbeatMonitor{ + ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", Owner: "+18005550199", PhoneOnline: true, + } + history.afterLast = func() { + history.heartbeat = nil + history.err = stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "history expired after saving") + } + service := NewHeartbeatService(logger, tracer, history, &heartbeatServiceMonitorRepository{monitor: monitor}, + NewEventDispatcher(logger, tracer, nil, queue, PushQueueConfig{})) + heartbeat, err := service.Store(context.Background(), HeartbeatStoreParams{ + UserID: monitor.UserID, Owner: monitor.Owner, Source: "https://example.com", Timestamp: time.Now().UTC(), + }) + require.NoError(t, err) + require.NotNil(t, heartbeat) + assert.Empty(t, queue.events, "known missing history must not schedule checks") + assert.True(t, monitor.RequiresCheck(), "a later heartbeat must still be able to restart") +} diff --git a/api/pkg/services/push_queue_service.go b/api/pkg/services/push_queue_service.go index b67fb23ce..4ab7bafd1 100644 --- a/api/pkg/services/push_queue_service.go +++ b/api/pkg/services/push_queue_service.go @@ -9,6 +9,8 @@ import ( // PushQueueTask represents a push queue task type PushQueueTask struct { + // ID deduplicates named task creation; empty IDs retain anonymous queue behavior. + ID string Method string URL string Body []byte