Skip to content

Commit 1324e6d

Browse files
mekilismekilis
authored
feat(dataplane): record delivery attempt request/response timestamps (#2694)
add per-attempt requested_at (request sent) and responded_at (response received) to delivery attempts. responded_at is null when the endpoint returns no HTTP response (network error or timeout). include acknowledged_at, requested_at and responded_at in eventdelivery.* meta event payloads so the push payload matches the delivery attempt polling api. surface delivery_mode in the dashboard: a selector on the subscription form and a delivery mode indicator on the event delivery details page. migration adds nullable requested_at/responded_at columns to delivery_attempts; partition and unpartition rebuilds preserve them.
1 parent 9276942 commit 1324e6d

21 files changed

Lines changed: 472 additions & 18 deletions

‎database/listener/event_delivery_listener.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ type MetaEventDelivery struct {
4040
Metadata *datastore.Metadata `json:"metadata"`
4141
CLIMetadata *datastore.CLIMetadata `json:"cli_metadata"`
4242
Description string `json:"description,omitempty"`
43+
AcknowledgedAt null.Time `json:"acknowledged_at,omitempty"`
4344
CreatedAt time.Time `json:"created_at,omitempty"`
4445
UpdatedAt time.Time `json:"updated_at,omitempty"`
4546
DeletedAt null.Time `json:"deleted_at,omitempty"`
@@ -101,6 +102,7 @@ func getMetaEventDelivery(eventDelivery *datastore.EventDelivery) *MetaEventDeli
101102
Metadata: eventDelivery.Metadata,
102103
CLIMetadata: eventDelivery.CLIMetadata,
103104
Description: eventDelivery.Description,
105+
AcknowledgedAt: eventDelivery.AcknowledgedAt,
104106
CreatedAt: eventDelivery.CreatedAt,
105107
UpdatedAt: eventDelivery.UpdatedAt,
106108
}

‎database/listener/event_delivery_listener_test.go‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
package listener
2+
3+
import (
4+
"encoding/json"
5+
"testing"
6+
"time"
7+
8+
"github.com/stretchr/testify/require"
9+
"gopkg.in/guregu/null.v4"
10+
11+
"github.com/frain-dev/convoy/datastore"
12+
)
13+
14+
func TestGetMetaEventDelivery_CopiesAcknowledgedAt(t *testing.T) {
15+
ackedAt := time.Now().UTC()
16+
eventDelivery := &datastore.EventDelivery{
17+
UID: "ed-1",
18+
ProjectID: "proj-1",
19+
Status: datastore.SuccessEventStatus,
20+
AcknowledgedAt: null.TimeFrom(ackedAt),
21+
}
22+
23+
meta := getMetaEventDelivery(eventDelivery)
24+
25+
require.True(t, meta.AcknowledgedAt.Valid)
26+
require.True(t, meta.AcknowledgedAt.Time.Equal(ackedAt))
27+
}
28+
29+
func TestMetaEventDelivery_PayloadIncludesTimestamps(t *testing.T) {
30+
requestedAt := time.Now().UTC()
31+
respondedAt := requestedAt.Add(40 * time.Millisecond)
32+
33+
meta := getMetaEventDelivery(&datastore.EventDelivery{
34+
UID: "ed-2",
35+
ProjectID: "proj-1",
36+
Status: datastore.SuccessEventStatus,
37+
AcknowledgedAt: null.TimeFrom(requestedAt),
38+
})
39+
meta.DeliveryAttempt = datastore.DeliveryAttempt{
40+
UID: "att-1",
41+
RequestedAt: null.TimeFrom(requestedAt),
42+
RespondedAt: null.TimeFrom(respondedAt),
43+
}
44+
45+
raw, err := json.Marshal(meta)
46+
require.NoError(t, err)
47+
48+
var payload struct {
49+
AcknowledgedAt *time.Time `json:"acknowledged_at"`
50+
Attempt struct {
51+
RequestedAt *time.Time `json:"requested_at"`
52+
RespondedAt *time.Time `json:"responded_at"`
53+
} `json:"attempt"`
54+
}
55+
require.NoError(t, json.Unmarshal(raw, &payload))
56+
57+
require.NotNil(t, payload.AcknowledgedAt)
58+
require.NotNil(t, payload.Attempt.RequestedAt)
59+
require.NotNil(t, payload.Attempt.RespondedAt)
60+
}

‎datastore/models.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1021,6 +1021,9 @@ type DeliveryAttempt struct {
10211021
Error string `json:"error,omitempty" db:"error"`
10221022
Status bool `json:"status,omitempty" db:"status"`
10231023

1024+
RequestedAt null.Time `json:"requested_at,omitempty" db:"requested_at" swaggertype:"string"`
1025+
RespondedAt null.Time `json:"responded_at,omitempty" db:"responded_at" swaggertype:"string"`
1026+
10241027
CreatedAt time.Time `json:"created_at,omitempty" db:"created_at" swaggertype:"string"`
10251028
UpdatedAt time.Time `json:"updated_at,omitempty" db:"updated_at" swaggertype:"string"`
10261029
DeletedAt null.Time `json:"deleted_at,omitempty" db:"deleted_at" swaggertype:"string"`

‎docs/docs.go‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// Package docs Code generated by swaggo/swag at 2026-06-25 06:45:12.598443 +0100 WAT m=+2.279863418. DO NOT EDIT
1+
// Package docs Code generated by swaggo/swag at 2026-06-26 09:39:28.105879 +0100 WAT m=+2.295455626. DO NOT EDIT
22
package docs
33

44
import "github.com/swaggo/swag"
@@ -7699,6 +7699,12 @@ const docTemplate = `{
76997699
"request_http_header": {
77007700
"$ref": "#/definitions/datastore.HttpHeader"
77017701
},
7702+
"requested_at": {
7703+
"type": "string"
7704+
},
7705+
"responded_at": {
7706+
"type": "string"
7707+
},
77027708
"response_data": {
77037709
"type": "string"
77047710
},

‎docs/swagger.json‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7696,6 +7696,12 @@
76967696
"request_http_header": {
76977697
"$ref": "#/definitions/datastore.HttpHeader"
76987698
},
7699+
"requested_at": {
7700+
"type": "string"
7701+
},
7702+
"responded_at": {
7703+
"type": "string"
7704+
},
76997705
"response_data": {
77007706
"type": "string"
77017707
},

‎docs/swagger.yaml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,10 @@ definitions:
108108
type: string
109109
request_http_header:
110110
$ref: '#/definitions/datastore.HttpHeader'
111+
requested_at:
112+
type: string
113+
responded_at:
114+
type: string
111115
response_data:
112116
type: string
113117
response_http_header:

‎docs/v3/openapi3.json‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8730,6 +8730,12 @@
87308730
"request_http_header": {
87318731
"$ref": "#/components/schemas/datastore.HttpHeader"
87328732
},
8733+
"requested_at": {
8734+
"type": "string"
8735+
},
8736+
"responded_at": {
8737+
"type": "string"
8738+
},
87338739
"response_data": {
87348740
"type": "string"
87358741
},

‎docs/v3/openapi3.yaml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4825,6 +4825,10 @@ components:
48254825
type: string
48264826
request_http_header:
48274827
$ref: '#/components/schemas/datastore.HttpHeader'
4828+
requested_at:
4829+
type: string
4830+
responded_at:
4831+
type: string
48284832
response_data:
48294833
type: string
48304834
response_http_header:

‎internal/delivery_attempts/create_delivery_attempt_test.go‎

Lines changed: 128 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99

1010
"github.com/oklog/ulid/v2"
1111
"github.com/stretchr/testify/require"
12+
"gopkg.in/guregu/null.v4"
1213

1314
"github.com/frain-dev/convoy/config"
1415
"github.com/frain-dev/convoy/database"
@@ -85,6 +86,133 @@ func TestCreateDeliveryAttempt_ValidRequest(t *testing.T) {
8586
require.Equal(t, attempt.Status, fetched.Status)
8687
}
8788

89+
func TestCreateDeliveryAttempt_PersistsRequestedAndRespondedAt(t *testing.T) {
90+
db, ctx := setupTestDB(t)
91+
defer db.Close()
92+
93+
project := seedTestData(t, db, ctx)
94+
endpoint := seedEndpoint(t, db, ctx, project)
95+
eventDelivery := seedEventDelivery(t, db, ctx, project, endpoint)
96+
97+
service := New(log.New("convoy", log.LevelError), db)
98+
99+
requestedAt := time.Now().UTC().Truncate(time.Millisecond)
100+
respondedAt := requestedAt.Add(75 * time.Millisecond)
101+
102+
attempt := &datastore.DeliveryAttempt{
103+
UID: ulid.Make().String(),
104+
URL: "https://example.com/webhook",
105+
Method: "POST",
106+
APIVersion: "2023.12.25",
107+
EndpointID: endpoint.UID,
108+
EventDeliveryId: eventDelivery.UID,
109+
ProjectId: project.UID,
110+
HttpResponseCode: "200",
111+
Status: true,
112+
RequestedAt: null.TimeFrom(requestedAt),
113+
RespondedAt: null.TimeFrom(respondedAt),
114+
}
115+
116+
err := service.CreateDeliveryAttempt(ctx, attempt)
117+
require.NoError(t, err)
118+
119+
fetched, err := service.FindDeliveryAttemptById(ctx, eventDelivery.UID, attempt.UID)
120+
require.NoError(t, err)
121+
require.True(t, fetched.RequestedAt.Valid)
122+
require.True(t, fetched.RespondedAt.Valid)
123+
require.True(t, fetched.RequestedAt.Time.Equal(requestedAt))
124+
require.True(t, fetched.RespondedAt.Time.Equal(respondedAt))
125+
126+
// And via the list path used by the meta-event listener.
127+
attempts, err := service.FindDeliveryAttempts(ctx, eventDelivery.UID)
128+
require.NoError(t, err)
129+
require.Len(t, attempts, 1)
130+
require.True(t, attempts[0].RequestedAt.Valid)
131+
require.True(t, attempts[0].RespondedAt.Valid)
132+
}
133+
134+
func TestCreateDeliveryAttempt_NoResponseLeavesRespondedAtNull(t *testing.T) {
135+
db, ctx := setupTestDB(t)
136+
defer db.Close()
137+
138+
project := seedTestData(t, db, ctx)
139+
endpoint := seedEndpoint(t, db, ctx, project)
140+
eventDelivery := seedEventDelivery(t, db, ctx, project, endpoint)
141+
142+
service := New(log.New("convoy", log.LevelError), db)
143+
144+
attempt := &datastore.DeliveryAttempt{
145+
UID: ulid.Make().String(),
146+
URL: "https://example.com/webhook",
147+
Method: "POST",
148+
APIVersion: "2023.12.25",
149+
EndpointID: endpoint.UID,
150+
EventDeliveryId: eventDelivery.UID,
151+
ProjectId: project.UID,
152+
Error: "connection refused",
153+
Status: false,
154+
RequestedAt: null.TimeFrom(time.Now().UTC()),
155+
// RespondedAt intentionally left zero (no HTTP response).
156+
}
157+
158+
err := service.CreateDeliveryAttempt(ctx, attempt)
159+
require.NoError(t, err)
160+
161+
fetched, err := service.FindDeliveryAttemptById(ctx, eventDelivery.UID, attempt.UID)
162+
require.NoError(t, err)
163+
require.True(t, fetched.RequestedAt.Valid)
164+
require.False(t, fetched.RespondedAt.Valid)
165+
}
166+
167+
func TestPartitionRoundTrip_PreservesRequestedAndRespondedAt(t *testing.T) {
168+
db, ctx := setupTestDB(t)
169+
defer db.Close()
170+
171+
project := seedTestData(t, db, ctx)
172+
endpoint := seedEndpoint(t, db, ctx, project)
173+
eventDelivery := seedEventDelivery(t, db, ctx, project, endpoint)
174+
175+
service := New(log.New("convoy", log.LevelError), db)
176+
177+
requestedAt := time.Now().UTC().Truncate(time.Millisecond)
178+
respondedAt := requestedAt.Add(60 * time.Millisecond)
179+
180+
attempt := &datastore.DeliveryAttempt{
181+
UID: ulid.Make().String(),
182+
URL: "https://example.com/webhook",
183+
Method: "POST",
184+
APIVersion: "2023.12.25",
185+
EndpointID: endpoint.UID,
186+
EventDeliveryId: eventDelivery.UID,
187+
ProjectId: project.UID,
188+
HttpResponseCode: "200",
189+
Status: true,
190+
RequestedAt: null.TimeFrom(requestedAt),
191+
RespondedAt: null.TimeFrom(respondedAt),
192+
}
193+
require.NoError(t, service.CreateDeliveryAttempt(ctx, attempt))
194+
195+
// Partition rebuild must carry the new columns over.
196+
require.NoError(t, service.PartitionDeliveryAttemptsTable(ctx))
197+
198+
fetched, err := service.FindDeliveryAttemptById(ctx, eventDelivery.UID, attempt.UID)
199+
require.NoError(t, err)
200+
require.True(t, fetched.RequestedAt.Valid)
201+
require.True(t, fetched.RespondedAt.Valid)
202+
require.True(t, fetched.RequestedAt.Time.Equal(requestedAt))
203+
require.True(t, fetched.RespondedAt.Time.Equal(respondedAt))
204+
205+
// Un-partition rebuild must also carry them over.
206+
require.NoError(t, service.UnPartitionDeliveryAttemptsTable(ctx))
207+
208+
fetched, err = service.FindDeliveryAttemptById(ctx, eventDelivery.UID, attempt.UID)
209+
require.NoError(t, err)
210+
require.True(t, fetched.RequestedAt.Valid)
211+
require.True(t, fetched.RespondedAt.Valid)
212+
require.True(t, fetched.RequestedAt.Time.Equal(requestedAt))
213+
require.True(t, fetched.RespondedAt.Time.Equal(respondedAt))
214+
}
215+
88216
func TestCreateDeliveryAttempt_WithHeaders(t *testing.T) {
89217
db, ctx := setupTestDB(t)
90218
defer db.Close()

‎internal/delivery_attempts/impl.go‎

Lines changed: 17 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,8 @@ func (s *Service) CreateDeliveryAttempt(ctx context.Context, attempt *datastore.
5757
HttpStatus: pgtype.Text{String: attempt.HttpResponseCode, Valid: attempt.HttpResponseCode != ""},
5858
Error: pgtype.Text{String: attempt.Error, Valid: attempt.Error != ""},
5959
Status: pgtype.Bool{Bool: attempt.Status, Valid: true},
60+
RequestedAt: common.NullTimeToPgTimestamptz(attempt.RequestedAt),
61+
RespondedAt: common.NullTimeToPgTimestamptz(attempt.RespondedAt),
6062
}
6163

6264
// Marshal headers to JSON bytes
@@ -337,6 +339,8 @@ BEGIN
337339
response_data BYTEA,
338340
error TEXT,
339341
status BOOLEAN,
342+
requested_at TIMESTAMP WITH TIME ZONE,
343+
responded_at TIMESTAMP WITH TIME ZONE,
340344
created_at TIMESTAMP WITH TIME ZONE default now() not null,
341345
updated_at TIMESTAMP WITH TIME ZONE default now() not null,
342346
deleted_at TIMESTAMP WITH TIME ZONE,
@@ -367,12 +371,12 @@ BEGIN
367371
INSERT INTO convoy.delivery_attempts_new (
368372
id, url, method, api_version, project_id, endpoint_id,
369373
event_delivery_id, ip_address, request_http_header, response_http_header,
370-
http_status, response_data, error, status, created_at,
374+
http_status, response_data, error, status, requested_at, responded_at, created_at,
371375
updated_at, deleted_at
372376
)
373377
SELECT id, url, method, api_version, project_id, endpoint_id,
374378
event_delivery_id, ip_address, request_http_header, response_http_header,
375-
http_status, response_data, error, status, created_at,
379+
http_status, response_data, error, status, requested_at, responded_at, created_at,
376380
updated_at, deleted_at
377381
FROM convoy.delivery_attempts;
378382
@@ -435,6 +439,8 @@ begin
435439
response_data BYTEA,
436440
error TEXT,
437441
status BOOLEAN,
442+
requested_at TIMESTAMP WITH TIME ZONE,
443+
responded_at TIMESTAMP WITH TIME ZONE,
438444
created_at TIMESTAMP WITH TIME ZONE default now() not null,
439445
updated_at TIMESTAMP WITH TIME ZONE default now() not null,
440446
deleted_at TIMESTAMP WITH TIME ZONE
@@ -444,12 +450,12 @@ begin
444450
INSERT INTO convoy.delivery_attempts_new (
445451
id, url, method, api_version, project_id, endpoint_id,
446452
event_delivery_id, ip_address, request_http_header, response_http_header,
447-
http_status, response_data, error, status, created_at,
453+
http_status, response_data, error, status, requested_at, responded_at, created_at,
448454
updated_at, deleted_at
449455
)
450456
SELECT id, url, method, api_version, project_id, endpoint_id,
451457
event_delivery_id, ip_address, request_http_header, response_http_header,
452-
http_status, response_data::bytea, error, status, created_at,
458+
http_status, response_data::bytea, error, status, requested_at, responded_at, created_at,
453459
updated_at, deleted_at
454460
FROM convoy.delivery_attempts;
455461
@@ -490,6 +496,7 @@ func rowToDeliveryAttempt(row interface{}) (*datastore.DeliveryAttempt, error) {
490496
ipAddress, httpStatus, errorMsg pgtype.Text
491497
requestHeader, responseHeader, responseData []byte
492498
status pgtype.Bool
499+
requestedAt, respondedAt pgtype.Timestamptz
493500
createdAt, updatedAt pgtype.Timestamptz
494501
deletedAt pgtype.Timestamptz
495502
)
@@ -511,6 +518,8 @@ func rowToDeliveryAttempt(row interface{}) (*datastore.DeliveryAttempt, error) {
511518
responseData = r.ResponseData
512519
errorMsg = r.Error
513520
status = r.Status
521+
requestedAt = r.RequestedAt
522+
respondedAt = r.RespondedAt
514523
createdAt = r.CreatedAt
515524
updatedAt = r.UpdatedAt
516525
deletedAt = r.DeletedAt
@@ -529,6 +538,8 @@ func rowToDeliveryAttempt(row interface{}) (*datastore.DeliveryAttempt, error) {
529538
responseData = r.ResponseData
530539
errorMsg = r.Error
531540
status = r.Status
541+
requestedAt = r.RequestedAt
542+
respondedAt = r.RespondedAt
532543
createdAt = r.CreatedAt
533544
updatedAt = r.UpdatedAt
534545
deletedAt = r.DeletedAt
@@ -565,6 +576,8 @@ func rowToDeliveryAttempt(row interface{}) (*datastore.DeliveryAttempt, error) {
565576
ResponseDataString: string(responseData),
566577
Error: errorMsg.String,
567578
Status: status.Bool,
579+
RequestedAt: common.PgTimestamptzToNullTime(requestedAt),
580+
RespondedAt: common.PgTimestamptzToNullTime(respondedAt),
568581
CreatedAt: createdAt.Time,
569582
UpdatedAt: updatedAt.Time,
570583
}

0 commit comments

Comments
 (0)