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/heartbeat_service.go b/api/pkg/services/heartbeat_service.go index d6c442347..2e8ae2df9 100644 --- a/api/pkg/services/heartbeat_service.go +++ b/api/pkg/services/heartbeat_service.go @@ -124,6 +124,20 @@ func (service *HeartbeatService) Store(ctx context.Context, params HeartbeatStor return heartbeat, nil } + if monitor.RequiresCheck() { + ctxLogger.Info(fmt.Sprintf("heartbeat [%s] with heartbeat monitor [%s] for owner [%s] and user [%s] requires a check because it was updated at [%s]", heartbeat.ID, monitor.ID, monitor.Owner, monitor.UserID, monitor.UpdatedAt)) + if err = service.scheduleHeartbeatCheck(ctx, heartbeat.Timestamp, &HeartbeatMonitorParams{ + Owner: monitor.Owner, + PhoneID: monitor.PhoneID, + UserID: monitor.UserID, + MonitorID: monitor.ID, + Source: params.Source, + }); err != nil { + ctxLogger.Error(stacktrace.Propagatef(err, "cannot schedule heartbeat check for monitor [%s] due to heartbeat [%s]", monitor.ID, heartbeat.ID)) + return heartbeat, nil + } + } + if monitor.PhoneIsOffline() { 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) @@ -273,17 +287,16 @@ type HeartbeatMonitorParams struct { // Monitor the heartbeats of an owner and phone number func (service *HeartbeatService) Monitor(ctx context.Context, params *HeartbeatMonitorParams) error { - ctx, span := service.tracer.Start(ctx) + ctx, span, ctxLogger := service.tracer.StartWithLogger(ctx, service.logger) defer span.End() - ctxLogger := service.tracer.CtxLogger(service.logger, span) + ctxLogger.Info(fmt.Sprintf("monitoring heartbeat for owner [%s] and user [%s] with monitor ID [%s] and phone ID [%s]", params.Owner, params.UserID, params.MonitorID, params.PhoneID)) monitor, err := service.monitorRepository.Load(ctx, params.UserID, params.Owner) if err != nil && stacktrace.GetCode(err) == repositories.ErrCodeNotFound { ctxLogger.Info(fmt.Sprintf("heartbeat monitor does not exist for owner [%s] and user [%s]", params.Owner, params.UserID)) 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) @@ -294,6 +307,10 @@ func (service *HeartbeatService) Monitor(ctx context.Context, params *HeartbeatM params.MonitorID = monitor.ID heartbeat, err := service.repository.Last(ctx, params.UserID, params.Owner) + if stacktrace.GetCode(err) == repositories.ErrCodeNotFound { + ctxLogger.Info(fmt.Sprintf("heartbeat history does not exist for userID [%s] and owner [%s], stopping check", params.UserID, params.Owner)) + return nil + } if err != nil { ctxLogger.Error(stacktrace.Propagatef(err, "cannot fetch last heartbeat for userID [%s] and owner [%s] and ID [%s] removing check", params.UserID, params.Owner, params.MonitorID)) return nil diff --git a/api/pkg/services/heartbeat_service_test.go b/api/pkg/services/heartbeat_service_test.go new file mode 100644 index 000000000..ff607240e --- /dev/null +++ b/api/pkg/services/heartbeat_service_test.go @@ -0,0 +1,162 @@ +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 +} + +func (repository *heartbeatServiceRepository) Last(context.Context, entities.UserID, string) (*entities.Heartbeat, error) { + return repository.heartbeat, repository.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 +} + +func (repository *heartbeatServiceMonitorRepository) Load(context.Context, entities.UserID, string) (*entities.HeartbeatMonitor, error) { + return repository.monitor, repository.err +} + +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 + return nil +} + +type heartbeatServiceQueue struct { + phoneNotificationEventQueue +} + +func (queue *heartbeatServiceQueue) Enqueue(ctx context.Context, task *PushQueueTask, timeout time.Duration) (string, error) { + if _, err := queue.phoneNotificationEventQueue.Enqueue(ctx, task, timeout); err != nil { + return "", err + } + 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.Equal(t, "queue-1", monitor.QueueID) + assert.True(t, monitor.PhoneOnline) + var payload events.PhoneHeartbeatCheckPayload + require.NoError(t, queue.events[0].DataAs(&payload)) + assert.Equal(t, monitor.ID, payload.MonitorID) + assert.Equal(t, monitor.PhoneID, payload.PhoneID) + 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) + require.NoError(t, service.Monitor(context.Background(), params)) + require.Len(t, queue.events, 5) + assert.Equal(t, events.EventTypePhoneHeartbeatOffline, queue.events[4].Type()) +} + +func TestHeartbeatServiceStore_OnlyRestartsInactiveChecks(t *testing.T) { + for _, test := range []struct { + name string + updatedAt time.Time + wantCheck bool + }{ + {name: "active monitor", updatedAt: time.Now().UTC()}, + {name: "stale monitor", updatedAt: time.Now().UTC().Add(-3 * time.Hour), 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, + } + 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) +}