Skip to content

Commit a85c540

Browse files
AchoArnoldCopilot
AchoArnold
and
Copilot
committed
refactor(api): remove offline heartbeat delivery guard
Process offline events with an ordinary listener, including stale and repeated deliveries. Remove dispatcher filters and heartbeat-version fencing used only by the guard. Preserve heartbeat retention, check ownership, restart behavior, and fail-open storage. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: f27d8f52-9620-4ee2-a564-ec4870a5f840
1 parent 22adc51 commit a85c540

13 files changed

Lines changed: 59 additions & 565 deletions

‎api/pkg/di/container.go‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,6 @@ import (
6464
"github.com/gofiber/fiber/v3/middleware/cors"
6565

6666
"github.com/NdoleStudio/httpsms/pkg/entities"
67-
"github.com/NdoleStudio/httpsms/pkg/events"
6867
"github.com/NdoleStudio/httpsms/pkg/listeners"
6968
"github.com/NdoleStudio/httpsms/pkg/repositories"
7069
"github.com/NdoleStudio/httpsms/pkg/services"
@@ -1502,12 +1501,11 @@ func (container *Container) RegisterNotificationListeners() {
15021501
// RegisterHeartbeatListeners registers event listeners for listeners.HeartbeatListener
15031502
func (container *Container) RegisterHeartbeatListeners() {
15041503
container.logger.Debug(fmt.Sprintf("registering listners for %T", listeners.HeartbeatListener{}))
1505-
heartbeatListener, routes := listeners.NewHeartbeatListener(
1504+
_, routes := listeners.NewHeartbeatListener(
15061505
container.Logger(),
15071506
container.Tracer(),
15081507
container.HeartbeatService(),
15091508
)
1510-
container.EventDispatcher().SubscribeFilter(events.EventTypePhoneHeartbeatOffline, heartbeatListener.FilterOfflineEvent)
15111509

15121510
for event, handler := range routes {
15131511
container.EventDispatcher().Subscribe(event, handler)

‎api/pkg/entities/heartbeat_monitor.go‎

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,11 @@ type HeartbeatMonitor struct {
1616
PhoneID uuid.UUID `json:"phone_id" bson:"phone_id" example:"32343a19-da5e-4b1b-a767-3298a73703cb"`
1717
UserID UserID `json:"user_id" bson:"user_id" example:"WB7DRDWrJZRGbYrv2CKGkqbzvqdC"`
1818
// QueueID is the current check ownership marker, not the external task queue's ID.
19-
QueueID string `json:"queue_id" bson:"queue_id" example:"0360259236613675274"`
20-
// LastHeartbeatID fences delayed offline transitions independently of check leases.
21-
LastHeartbeatID uuid.UUID `json:"-" bson:"last_heartbeat_id,omitempty" gorm:"type:uuid"`
22-
Owner string `json:"owner" bson:"owner" example:"+18005550199"`
23-
PhoneOnline bool `json:"phone_online" bson:"phone_online" example:"true" default:"true"`
24-
CreatedAt time.Time `json:"created_at" bson:"created_at" example:"2022-06-05T14:26:02.302718+03:00"`
25-
UpdatedAt time.Time `json:"updated_at" bson:"updated_at" example:"2022-06-05T14:26:10.303278+03:00"`
19+
QueueID string `json:"queue_id" bson:"queue_id" example:"0360259236613675274"`
20+
Owner string `json:"owner" bson:"owner" example:"+18005550199"`
21+
PhoneOnline bool `json:"phone_online" bson:"phone_online" example:"true" default:"true"`
22+
CreatedAt time.Time `json:"created_at" bson:"created_at" example:"2022-06-05T14:26:02.302718+03:00"`
23+
UpdatedAt time.Time `json:"updated_at" bson:"updated_at" example:"2022-06-05T14:26:10.303278+03:00"`
2624
}
2725

2826
// RequiresCheck returns true if the heartbeat monitor requires a check

‎api/pkg/events/listeners.go‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,3 @@ import (
88

99
// EventListener is the type for processing events
1010
type EventListener func(ctx context.Context, event cloudevents.Event) error
11-
12-
// EventFilter decides whether an event may reach any subscribers.
13-
type EventFilter func(ctx context.Context, event cloudevents.Event) (bool, error)

‎api/pkg/listeners/heartbeat_listener.go‎

Lines changed: 10 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -33,10 +33,11 @@ func NewHeartbeatListener(
3333
}
3434

3535
return l, map[string]events.EventListener{
36-
events.EventTypePhoneUpdated: l.onPhoneUpdated,
37-
events.EventTypePhoneDeleted: l.onPhoneDeleted,
38-
events.EventTypePhoneHeartbeatCheck: l.onPhoneHeartbeatCheck,
39-
events.UserAccountDeleted: l.onUserAccountDeleted,
36+
events.EventTypePhoneUpdated: l.onPhoneUpdated,
37+
events.EventTypePhoneDeleted: l.onPhoneDeleted,
38+
events.EventTypePhoneHeartbeatCheck: l.onPhoneHeartbeatCheck,
39+
events.EventTypePhoneHeartbeatOffline: l.onPhoneHeartbeatOffline,
40+
events.UserAccountDeleted: l.onUserAccountDeleted,
4041
}
4142
}
4243

@@ -107,22 +108,20 @@ func (listener *HeartbeatListener) onPhoneHeartbeatCheck(ctx context.Context, ev
107108
return nil
108109
}
109110

110-
// FilterOfflineEvent validates and applies the transition before notification subscribers run.
111-
func (listener *HeartbeatListener) FilterOfflineEvent(ctx context.Context, event cloudevents.Event) (bool, error) {
111+
func (listener *HeartbeatListener) onPhoneHeartbeatOffline(ctx context.Context, event cloudevents.Event) error {
112112
ctx, span := listener.tracer.Start(ctx)
113113
defer span.End()
114114

115115
var payload events.PhoneHeartbeatOfflinePayload
116116
if err := event.DataAs(&payload); err != nil {
117-
return false, listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot decode [%s] into [%T]", event.Data(), payload))
117+
return listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot decode [%s] into [%T]", event.Data(), payload))
118118
}
119119

120-
allowed, err := listener.service.ApplyOfflineHeartbeat(ctx, &payload)
121-
if err != nil {
122-
return false, listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot validate offline heartbeat for monitor [%s]", payload.MonitorID))
120+
if err := listener.service.UpdatePhoneOnline(ctx, payload.UserID, payload.MonitorID, false); err != nil {
121+
return listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot mark heartbeat monitor [%s] offline", payload.MonitorID))
123122
}
124123

125-
return allowed, nil
124+
return nil
126125
}
127126

128127
func (listener *HeartbeatListener) onUserAccountDeleted(ctx context.Context, event cloudevents.Event) error {

‎api/pkg/listeners/heartbeat_listener_test.go‎

Lines changed: 33 additions & 149 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@ package listeners
22

33
import (
44
"context"
5-
"encoding/json"
65
"sync/atomic"
76
"testing"
87
"time"
@@ -12,169 +11,54 @@ import (
1211
"github.com/NdoleStudio/httpsms/pkg/repositories"
1312
"github.com/NdoleStudio/httpsms/pkg/services"
1413
"github.com/NdoleStudio/httpsms/pkg/telemetry"
15-
"github.com/NdoleStudio/stacktrace"
1614
cloudevents "github.com/cloudevents/sdk-go/v2"
1715
"github.com/google/uuid"
1816
"github.com/stretchr/testify/require"
1917
"go.opentelemetry.io/otel/metric/noop"
2018
)
2119

22-
type offlineHeartbeatRepository struct {
23-
repositories.HeartbeatRepository
24-
heartbeat *entities.Heartbeat
25-
err error
26-
}
27-
28-
func (repository *offlineHeartbeatRepository) Last(context.Context, entities.UserID, string) (*entities.Heartbeat, error) {
29-
return repository.heartbeat, repository.err
30-
}
31-
32-
func (repository *offlineHeartbeatRepository) Store(_ context.Context, heartbeat *entities.Heartbeat) error {
33-
repository.heartbeat = heartbeat
34-
return nil
35-
}
36-
3720
type offlineMonitorRepository struct {
3821
repositories.HeartbeatMonitorRepository
39-
monitor *entities.HeartbeatMonitor
40-
lastHeartbeatID uuid.UUID
41-
beforeOffline func()
42-
}
43-
44-
func (repository *offlineMonitorRepository) Load(context.Context, entities.UserID, string) (*entities.HeartbeatMonitor, error) {
45-
monitor := *repository.monitor
46-
monitor.LastHeartbeatID = repository.lastHeartbeatID
47-
return &monitor, nil
48-
}
49-
50-
func (repository *offlineMonitorRepository) CompareAndSwapQueueID(_ context.Context, _ uuid.UUID, expected, next string) (bool, error) {
51-
if repository.monitor.QueueID != expected {
52-
return false, nil
53-
}
54-
repository.monitor.QueueID = next
55-
repository.monitor.UpdatedAt = time.Now().UTC()
56-
return true, nil
22+
monitor *entities.HeartbeatMonitor
5723
}
5824

59-
func (repository *offlineMonitorRepository) RecordHeartbeat(_ context.Context, _ entities.UserID, _ uuid.UUID, expected, heartbeatID uuid.UUID) (*entities.HeartbeatMonitor, error) {
60-
if repository.lastHeartbeatID != expected {
61-
return nil, nil
25+
func (repository *offlineMonitorRepository) UpdatePhoneOnline(_ context.Context, userID entities.UserID, monitorID uuid.UUID, online bool) error {
26+
if repository.monitor.UserID == userID && repository.monitor.ID == monitorID {
27+
repository.monitor.PhoneOnline = online
6228
}
63-
previous := *repository.monitor
64-
previous.LastHeartbeatID = repository.lastHeartbeatID
65-
repository.lastHeartbeatID = heartbeatID
66-
repository.monitor.PhoneOnline = true
67-
return &previous, nil
68-
}
69-
70-
func (repository *offlineMonitorRepository) UpdatePhoneOfflineForHeartbeat(_ context.Context, _ entities.UserID, _ uuid.UUID, expected, heartbeatID uuid.UUID) (bool, error) {
71-
repository.runOfflineHook()
72-
if repository.lastHeartbeatID != expected {
73-
return false, nil
74-
}
75-
repository.monitor.PhoneOnline = false
76-
if heartbeatID != uuid.Nil {
77-
repository.lastHeartbeatID = heartbeatID
78-
}
79-
return true, nil
80-
}
81-
82-
func (repository *offlineMonitorRepository) UpdatePhoneOnline(_ context.Context, _ entities.UserID, _ uuid.UUID, online bool) error {
83-
if !online {
84-
repository.runOfflineHook()
85-
}
86-
repository.monitor.PhoneOnline = online
8729
return nil
8830
}
8931

90-
func (repository *offlineMonitorRepository) runOfflineHook() {
91-
if repository.beforeOffline != nil {
92-
hook := repository.beforeOffline
93-
repository.beforeOffline = nil
94-
hook()
95-
}
96-
}
97-
98-
type offlineEventQueue struct{ events []cloudevents.Event }
99-
100-
func (queue *offlineEventQueue) Enqueue(_ context.Context, task *services.PushQueueTask, _ time.Duration) (string, error) {
101-
var event cloudevents.Event
102-
if err := json.Unmarshal(task.Body, &event); err != nil {
103-
return "", err
32+
func TestHeartbeatOfflineDelivery_AllowsRepeatedAndOutOfOrderEvents(t *testing.T) {
33+
logger := &noopListenerLogger{}
34+
tracer := telemetry.NewOtelLogger("test", logger)
35+
meter, err := noop.NewMeterProvider().Meter("test").Float64Histogram("dispatch")
36+
require.NoError(t, err)
37+
dispatcher := services.NewEventDispatcher(logger, tracer, meter, nil, services.PushQueueConfig{})
38+
monitor := &entities.HeartbeatMonitor{ID: uuid.New(), UserID: "user-1", PhoneOnline: true}
39+
repository := &offlineMonitorRepository{monitor: monitor}
40+
service := services.NewHeartbeatService(logger, tracer, nil, repository, dispatcher)
41+
_, routes := NewHeartbeatListener(logger, tracer, service)
42+
for eventType, listener := range routes {
43+
dispatcher.Subscribe(eventType, listener)
10444
}
105-
queue.events = append(queue.events, event)
106-
return event.ID(), nil
107-
}
45+
var notifications atomic.Int32
46+
dispatcher.Subscribe(events.EventTypePhoneHeartbeatOffline, func(context.Context, cloudevents.Event) error {
47+
notifications.Add(1)
48+
return nil
49+
})
50+
event := cloudevents.NewEvent()
51+
event.SetID(uuid.NewString())
52+
event.SetSource("https://example.com")
53+
event.SetType(events.EventTypePhoneHeartbeatOffline)
54+
require.NoError(t, event.SetData(cloudevents.ApplicationJSON, events.PhoneHeartbeatOfflinePayload{
55+
MonitorID: monitor.ID, UserID: monitor.UserID, LastHeartbeatTimestamp: time.Now().UTC().Add(-24 * time.Hour),
56+
}))
10857

109-
func TestHeartbeatOfflineDelivery_RejectsNewerHeartbeatBeforeSubscribers(t *testing.T) {
110-
for _, arrival := range []string{"before delivery", "between history and status update", "history read fails", "history expired", "no newer heartbeat", "legacy event"} {
111-
t.Run(arrival, func(t *testing.T) {
112-
logger := &noopListenerLogger{}
113-
tracer := telemetry.NewOtelLogger("test", logger)
114-
meter, err := noop.NewMeterProvider().Meter("test").Float64Histogram("dispatch")
115-
require.NoError(t, err)
116-
queue := &offlineEventQueue{}
117-
dispatcher := services.NewEventDispatcher(logger, tracer, meter, queue, services.PushQueueConfig{})
118-
history := &offlineHeartbeatRepository{heartbeat: &entities.Heartbeat{
119-
ID: uuid.New(), Timestamp: time.Now().UTC().Add(-65 * time.Minute),
120-
}}
121-
monitor := &offlineMonitorRepository{monitor: &entities.HeartbeatMonitor{
122-
ID: uuid.New(), PhoneID: uuid.New(), UserID: "user-1", Owner: "+18005550199",
123-
PhoneOnline: true, QueueID: "check:current-check", UpdatedAt: time.Now().UTC(),
124-
}, lastHeartbeatID: history.heartbeat.ID}
125-
service := services.NewHeartbeatService(logger, tracer, history, monitor, dispatcher)
126-
heartbeatListener, routes := NewHeartbeatListener(logger, tracer, service)
127-
dispatcher.SubscribeFilter(events.EventTypePhoneHeartbeatOffline, heartbeatListener.FilterOfflineEvent)
128-
for eventType, listener := range routes {
129-
dispatcher.Subscribe(eventType, listener)
130-
}
131-
var notifications atomic.Int32
132-
dispatcher.Subscribe(events.EventTypePhoneHeartbeatOffline, func(context.Context, cloudevents.Event) error {
133-
notifications.Add(1)
134-
return nil
135-
})
136-
require.NoError(t, service.Monitor(context.Background(), &services.HeartbeatMonitorParams{
137-
UserID: monitor.monitor.UserID, Owner: monitor.monitor.Owner, MonitorID: monitor.monitor.ID,
138-
Source: "https://example.com", CheckID: "current-check",
139-
}))
140-
require.Len(t, queue.events, 3)
141-
event := queue.events[1]
142-
storeHeartbeat := func() {
143-
_, err := service.Store(context.Background(), services.HeartbeatStoreParams{
144-
UserID: monitor.monitor.UserID, Owner: monitor.monitor.Owner,
145-
Source: "https://example.com", Timestamp: time.Now().UTC(),
146-
})
147-
require.NoError(t, err)
148-
}
149-
switch arrival {
150-
case "before delivery":
151-
storeHeartbeat()
152-
case "between history and status update":
153-
monitor.beforeOffline = storeHeartbeat
154-
case "history read fails":
155-
history.err = stacktrace.NewError("history unavailable")
156-
case "history expired":
157-
history.err = stacktrace.NewErrorWithCode(repositories.ErrCodeNotFound, "history expired")
158-
case "legacy event":
159-
var data map[string]any
160-
require.NoError(t, event.DataAs(&data))
161-
delete(data, "heartbeat_id")
162-
require.NoError(t, event.SetData(cloudevents.ApplicationJSON, data))
163-
monitor.lastHeartbeatID = uuid.Nil
164-
}
165-
err = dispatcher.DispatchSync(context.Background(), event)
166-
if arrival == "history read fails" {
167-
require.Error(t, err)
168-
} else {
169-
require.NoError(t, err)
170-
}
171-
if arrival == "no newer heartbeat" || arrival == "legacy event" {
172-
require.EqualValues(t, 1, notifications.Load())
173-
require.False(t, monitor.monitor.PhoneOnline)
174-
} else {
175-
require.Zero(t, notifications.Load(), "stale or unverifiable events must not reach email/webhook subscribers")
176-
require.True(t, monitor.monitor.PhoneOnline)
177-
}
178-
})
58+
for delivery := 1; delivery <= 2; delivery++ {
59+
monitor.PhoneOnline = true
60+
require.NoError(t, dispatcher.DispatchSync(context.Background(), event))
61+
require.False(t, monitor.PhoneOnline)
62+
require.EqualValues(t, delivery, notifications.Load())
17963
}
18064
}

‎api/pkg/repositories/gorm_heartbeat_monitor_repository.go‎

Lines changed: 0 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,6 @@ import (
1313
"github.com/NdoleStudio/httpsms/pkg/telemetry"
1414
"github.com/NdoleStudio/stacktrace"
1515
"gorm.io/gorm"
16-
"gorm.io/gorm/clause"
1716
)
1817

1918
// gormHeartbeatRepository is responsible for persisting entities.Heartbeat
@@ -128,63 +127,6 @@ func (repository *gormHeartbeatMonitorRepository) CompareAndSwapQueueID(ctx cont
128127
return result.RowsAffected == 1, nil
129128
}
130129

131-
func (repository *gormHeartbeatMonitorRepository) RecordHeartbeat(ctx context.Context, userID entities.UserID, monitorID, expected, heartbeatID uuid.UUID) (*entities.HeartbeatMonitor, error) {
132-
ctx, span := repository.tracer.Start(ctx)
133-
defer span.End()
134-
ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
135-
defer cancel()
136-
137-
var previous *entities.HeartbeatMonitor
138-
err := repository.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
139-
monitor := new(entities.HeartbeatMonitor)
140-
query := heartbeatMonitorWithVersion(tx.Where("id = ?", monitorID).Where("user_id = ?", userID), expected)
141-
if err := query.Clauses(clause.Locking{Strength: "UPDATE"}).First(monitor).Error; err != nil {
142-
if errors.Is(err, gorm.ErrRecordNotFound) {
143-
return nil
144-
}
145-
return stacktrace.Propagatef(err, "cannot lock heartbeat version for monitor [%s]", monitorID)
146-
}
147-
before := *monitor
148-
if err := tx.Model(monitor).UpdateColumns(map[string]any{
149-
"last_heartbeat_id": heartbeatID,
150-
"phone_online": true,
151-
}).Error; err != nil {
152-
return stacktrace.Propagatef(err, "cannot record heartbeat version for monitor [%s]", monitorID)
153-
}
154-
previous = &before
155-
return nil
156-
})
157-
if err != nil {
158-
return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot mark heartbeat monitor [%s] online", monitorID))
159-
}
160-
return previous, nil
161-
}
162-
163-
func (repository *gormHeartbeatMonitorRepository) UpdatePhoneOfflineForHeartbeat(ctx context.Context, userID entities.UserID, monitorID, expected, heartbeatID uuid.UUID) (bool, error) {
164-
ctx, span := repository.tracer.Start(ctx)
165-
defer span.End()
166-
ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
167-
defer cancel()
168-
169-
query := heartbeatMonitorWithVersion(repository.db.WithContext(ctx).
170-
Model(&entities.HeartbeatMonitor{}).Where("id = ?", monitorID).Where("user_id = ?", userID), expected)
171-
result := query.UpdateColumns(map[string]any{
172-
"last_heartbeat_id": heartbeatID,
173-
"phone_online": false,
174-
})
175-
if result.Error != nil {
176-
return false, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(result.Error, "cannot apply offline heartbeat version for monitor [%s]", monitorID))
177-
}
178-
return result.RowsAffected == 1, nil
179-
}
180-
181-
func heartbeatMonitorWithVersion(query *gorm.DB, heartbeatID uuid.UUID) *gorm.DB {
182-
if heartbeatID == uuid.Nil {
183-
return query.Where("last_heartbeat_id IS NULL OR last_heartbeat_id = ?", uuid.Nil)
184-
}
185-
return query.Where("last_heartbeat_id = ?", heartbeatID)
186-
}
187-
188130
// Index entities.Message between 2 parties
189131
func (repository *gormHeartbeatMonitorRepository) Index(ctx context.Context, userID entities.UserID, owner string, params IndexParams) (*[]entities.Heartbeat, error) {
190132
ctx, span := repository.tracer.Start(ctx)

0 commit comments

Comments
 (0)