From c20ac1a560ae4c6433a2769554c67e66b05d70cf Mon Sep 17 00:00:00 2001 From: Alex Luong Date: Tue, 27 Oct 2026 12:09:59 +0700 Subject: [PATCH 1/2] feat: add destination expiry --- .../apirouter/destination_handlers_test.go | 21 ++++++ internal/apirouter/retry_handlers.go | 14 +++- internal/apirouter/retry_handlers_test.go | 14 ++++ internal/models/entities.go | 11 +++- internal/models/entities_test.go | 13 ++++ internal/tenantstore/drivertest/crud.go | 50 ++++++++++++++ internal/tenantstore/drivertest/match.go | 57 ++++++++++++++++ internal/tenantstore/drivertest/misc.go | 64 ++++++++++++++++++ .../memtenantstore/memtenantstore.go | 22 +++++-- .../redistenantstore/redistenantstore.go | 65 +++++++++++++++++-- .../redistenantstore/serialization.go | 29 +++++++-- internal/util/testutil/destination.go | 6 ++ 12 files changed, 348 insertions(+), 18 deletions(-) diff --git a/internal/apirouter/destination_handlers_test.go b/internal/apirouter/destination_handlers_test.go index 15ac183bf..9d12dcfd1 100644 --- a/internal/apirouter/destination_handlers_test.go +++ b/internal/apirouter/destination_handlers_test.go @@ -382,6 +382,27 @@ func TestAPI_Destinations(t *testing.T) { assert.Equal(t, "d1", dest.ID) }) + t.Run("expires_at is not in the response", func(t *testing.T) { + h := newAPITest(t) + h.tenantStore.UpsertTenant(t.Context(), tf.Any(tf.WithID("t1"))) + require.NoError(t, h.tenantStore.CreateDestination(t.Context(), df.Any(df.WithID("d1"), df.WithTenantID("t1"), df.WithExpiresAt(time.Now().Add(time.Hour))))) + + get := func(path string) []byte { + resp := h.do(h.withAPIKey(httptest.NewRequest(http.MethodGet, path, nil))) + require.Equal(t, http.StatusOK, resp.Code) + return resp.Body.Bytes() + } + + var one map[string]any + require.NoError(t, json.Unmarshal(get("/api/v1/tenants/t1/destinations/d1"), &one)) + assert.NotContains(t, one, "expires_at") + + var list []map[string]any + require.NoError(t, json.Unmarshal(get("/api/v1/tenants/t1/destinations"), &list)) + require.Len(t, list, 1) + assert.NotContains(t, list[0], "expires_at") + }) + t.Run("wildcard patterns are hidden when disabled without changing storage", func(t *testing.T) { store := tenantstore.NewMemTenantStore() h := newAPITest(t, withTenantStore(store)) diff --git a/internal/apirouter/retry_handlers.go b/internal/apirouter/retry_handlers.go index 9b9da7b84..b90d9fcc7 100644 --- a/internal/apirouter/retry_handlers.go +++ b/internal/apirouter/retry_handlers.go @@ -4,6 +4,7 @@ import ( "context" "errors" "net/http" + "time" "github.com/gin-gonic/gin" "github.com/hookdeck/outpost/internal/logging" @@ -106,7 +107,7 @@ func (h *RetryHandlers) Retry(c *gin.Context) { } } - // 2. Check destination exists and is enabled + // 2. Check destination exists, is enabled and not expired destination, err := h.tenantStore.RetrieveDestination(c.Request.Context(), event.TenantID, req.DestinationID) if err != nil { if errors.Is(err, tenantstore.ErrDestinationDeleted) { @@ -131,6 +132,17 @@ func (h *RetryHandlers) Retry(c *gin.Context) { return } + if destination.IsExpired(time.Now()) { + AbortWithError(c, http.StatusBadRequest, ErrorResponse{ + Code: http.StatusBadRequest, + Message: "Destination is expired", + Data: map[string]string{ + "error": "destination_expired", + }, + }) + return + } + if !destination.MatchEvent(*event, h.topicsAllowWildcards) { AbortWithError(c, http.StatusBadRequest, ErrorResponse{ Code: http.StatusBadRequest, diff --git a/internal/apirouter/retry_handlers_test.go b/internal/apirouter/retry_handlers_test.go index 2c3526d75..d90375731 100644 --- a/internal/apirouter/retry_handlers_test.go +++ b/internal/apirouter/retry_handlers_test.go @@ -402,6 +402,20 @@ func TestAPI_Retry(t *testing.T) { assert.Equal(t, "destination_disabled", body.Data["error"]) }) + t.Run("expired destination returns 400", func(t *testing.T) { + h := setup(t) + require.NoError(t, h.tenantStore.UpsertDestination(t.Context(), df.Any(df.WithID("d2"), df.WithTenantID("t1"), df.WithTopics([]string{"*"}), df.WithExpiresAt(time.Now().Add(-time.Minute))))) + + resp := h.do(h.withAPIKey(retry(h, "e1", "d2"))) + + testutil.RequireErrorResponse(t, resp, http.StatusBadRequest, "Destination is expired") + var body struct { + Data map[string]string `json:"data"` + } + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &body)) + assert.Equal(t, "destination_expired", body.Data["error"]) + }) + t.Run("topic mismatch returns 400", func(t *testing.T) { h := setup(t) require.NoError(t, h.tenantStore.UpsertDestination(t.Context(), df.Any(df.WithID("d2"), df.WithTenantID("t1"), df.WithTopics([]string{"user.deleted"})))) diff --git a/internal/models/entities.go b/internal/models/entities.go index ec5ca0fd8..6625bd699 100644 --- a/internal/models/entities.go +++ b/internal/models/entities.go @@ -37,6 +37,12 @@ type Destination struct { CreatedAt time.Time `json:"created_at" redis:"created_at"` UpdatedAt time.Time `json:"updated_at" redis:"updated_at"` DisabledAt *time.Time `json:"disabled_at" redis:"disabled_at"` + // ExpiresAt is not exposed in the API. Once past, the destination receives no events. + ExpiresAt *time.Time `json:"-" redis:"expires_at"` +} + +func (d *Destination) IsExpired(now time.Time) bool { + return d.ExpiresAt != nil && !now.Before(*d.ExpiresAt) } func (d *Destination) Validate(topics []string, allowWildcards bool) error { @@ -47,10 +53,11 @@ func (d *Destination) Validate(topics []string, allowWildcards bool) error { } // MatchEvent checks if the destination matches the given event. -// Returns true if the destination is enabled, topic matches, and filter matches. +// Returns true if the destination is enabled and not expired, topic matches, +// and filter matches. // Embedded wildcard topic patterns are ignored when allowWildcards is false. func (d *Destination) MatchEvent(event Event, allowWildcards bool) bool { - if d.DisabledAt != nil { + if d.DisabledAt != nil || d.IsExpired(time.Now()) { return false } if !d.Topics.MatchTopic(event.Topic, allowWildcards) { diff --git a/internal/models/entities_test.go b/internal/models/entities_test.go index 5effe44f1..92abbe001 100644 --- a/internal/models/entities_test.go +++ b/internal/models/entities_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "testing" + "time" "github.com/hookdeck/outpost/internal/models" "github.com/hookdeck/outpost/internal/util/testutil" @@ -990,3 +991,15 @@ func TestEvent_ParsedData_NilData_ReturnsNil(t *testing.T) { assert.NoError(t, err) assert.Nil(t, parsed) } + +func TestDestination_IsExpired(t *testing.T) { + t.Parallel() + + now := time.Now() + at := func(d time.Duration) *time.Time { t := now.Add(d); return &t } + + assert.False(t, (&models.Destination{}).IsExpired(now)) + assert.False(t, (&models.Destination{ExpiresAt: at(time.Second)}).IsExpired(now)) + assert.True(t, (&models.Destination{ExpiresAt: at(0)}).IsExpired(now)) + assert.True(t, (&models.Destination{ExpiresAt: at(-time.Second)}).IsExpired(now)) +} diff --git a/internal/tenantstore/drivertest/crud.go b/internal/tenantstore/drivertest/crud.go index c8f04d781..5a688253c 100644 --- a/internal/tenantstore/drivertest/crud.go +++ b/internal/tenantstore/drivertest/crud.go @@ -431,6 +431,53 @@ func testCRUD(t *testing.T, newHarness HarnessMaker) { }) }) + t.Run("DestinationExpiry", func(t *testing.T) { + ctx := context.Background() + h, err := newHarness(ctx, t) + require.NoError(t, err) + t.Cleanup(h.Close) + + store, err := h.MakeDriver(ctx) + require.NoError(t, err) + + tenantID := idgen.String() + expiresAt := time.Now().Add(time.Hour).Truncate(time.Millisecond) + destination := testutil.DestinationFactory.Any( + testutil.DestinationFactory.WithTenantID(tenantID), + testutil.DestinationFactory.WithExpiresAt(expiresAt), + ) + + t.Run("persists expires_at", func(t *testing.T) { + require.NoError(t, store.CreateDestination(ctx, destination)) + actual, err := store.RetrieveDestination(ctx, tenantID, destination.ID) + require.NoError(t, err) + assertEqualDestination(t, destination, *actual) + assert.False(t, actual.IsExpired(time.Now())) + + listed, err := store.ListDestination(ctx, driver.ListDestinationRequest{TenantID: tenantID}) + require.NoError(t, err) + require.Len(t, listed, 1) + assertEqualTimePtr(t, &expiresAt, listed[0].ExpiresAt, "ExpiresAt") + }) + + t.Run("an expired destination is still returned", func(t *testing.T) { + destination.ExpiresAt = ptr(time.Now().Add(-time.Minute).Truncate(time.Millisecond)) + require.NoError(t, store.UpsertDestination(ctx, destination)) + actual, err := store.RetrieveDestination(ctx, tenantID, destination.ID) + require.NoError(t, err) + assertEqualTimePtr(t, destination.ExpiresAt, actual.ExpiresAt, "ExpiresAt") + assert.True(t, actual.IsExpired(time.Now())) + }) + + t.Run("upsert without expires_at clears it", func(t *testing.T) { + destination.ExpiresAt = nil + require.NoError(t, store.UpsertDestination(ctx, destination)) + actual, err := store.RetrieveDestination(ctx, tenantID, destination.ID) + require.NoError(t, err) + assert.Nil(t, actual.ExpiresAt) + }) + }) + t.Run("FilterPersistence", func(t *testing.T) { ctx := context.Background() h, err := newHarness(ctx, t) @@ -545,4 +592,7 @@ func assertEqualDestination(t *testing.T, expected, actual models.Destination) { assertEqualTime(t, expected.CreatedAt, actual.CreatedAt, "CreatedAt") assertEqualTime(t, expected.UpdatedAt, actual.UpdatedAt, "UpdatedAt") assertEqualTimePtr(t, expected.DisabledAt, actual.DisabledAt, "DisabledAt") + assertEqualTimePtr(t, expected.ExpiresAt, actual.ExpiresAt, "ExpiresAt") } + +func ptr[T any](v T) *T { return &v } diff --git a/internal/tenantstore/drivertest/match.go b/internal/tenantstore/drivertest/match.go index f30911f1f..599265187 100644 --- a/internal/tenantstore/drivertest/match.go +++ b/internal/tenantstore/drivertest/match.go @@ -3,6 +3,7 @@ package drivertest import ( "context" "encoding/json" + "slices" "testing" "time" @@ -375,6 +376,62 @@ func testMatch(t *testing.T, newHarness HarnessMaker) { }) }) + t.Run("ExpireAndMatch", func(t *testing.T) { + ctx := context.Background() + h, err := newHarness(ctx, t) + require.NoError(t, err) + t.Cleanup(h.Close) + + store, err := h.MakeDriver(ctx) + require.NoError(t, err) + + tenant := models.Tenant{ID: idgen.String(), CreatedAt: time.Now()} + require.NoError(t, store.UpsertTenant(ctx, tenant)) + create := func(opts ...func(*models.Destination)) models.Destination { + opts = append([]func(*models.Destination){ + testutil.DestinationFactory.WithTenantID(tenant.ID), + testutil.DestinationFactory.WithTopics([]string{"user.created"}), + }, opts...) + d := testutil.DestinationFactory.Any(opts...) + require.NoError(t, store.CreateDestination(ctx, d)) + return d + } + plain := create() + future := create(testutil.DestinationFactory.WithExpiresAt(time.Now().Add(time.Hour))) + past := create(testutil.DestinationFactory.WithExpiresAt(time.Now().Add(-time.Hour))) + + event := testutil.EventFactory.Any( + testutil.EventFactory.WithTenantID(tenant.ID), + testutil.EventFactory.WithTopic("user.created"), + ) + matched, err := store.MatchEvent(ctx, event, true) + require.NoError(t, err) + assert.ElementsMatch(t, []string{plain.ID, future.ID}, matched) + assert.NotContains(t, matched, past.ID) + + t.Run("matches again once expires_at is cleared", func(t *testing.T) { + past.ExpiresAt = nil + require.NoError(t, store.UpsertDestination(ctx, past)) + matched, err := store.MatchEvent(ctx, event, true) + require.NoError(t, err) + assert.ElementsMatch(t, []string{plain.ID, future.ID, past.ID}, matched) + }) + + t.Run("expires without any write", func(t *testing.T) { + expiresAt := time.Now().Add(500 * time.Millisecond).Truncate(time.Millisecond) + soon := create(testutil.DestinationFactory.WithExpiresAt(expiresAt)) + matched, err := store.MatchEvent(ctx, event, true) + require.NoError(t, err) + if time.Now().Before(expiresAt) { + assert.Contains(t, matched, soon.ID) + } + require.Eventually(t, func() bool { + matched, err := store.MatchEvent(ctx, event, true) + return err == nil && !slices.Contains(matched, soon.ID) + }, 5*time.Second, 50*time.Millisecond) + }) + }) + t.Run("DeleteAndMatch", func(t *testing.T) { ctx := context.Background() h, err := newHarness(ctx, t) diff --git a/internal/tenantstore/drivertest/misc.go b/internal/tenantstore/drivertest/misc.go index 1fcfc8274..4f38b8f49 100644 --- a/internal/tenantstore/drivertest/misc.go +++ b/internal/tenantstore/drivertest/misc.go @@ -61,6 +61,70 @@ func testMisc(t *testing.T, newHarness HarnessMaker) { require.NoError(t, err, "Should be able to create destination after deleting one") }) + t.Run("MaxDestinationsPerTenantPurgesExpired", func(t *testing.T) { + ctx := context.Background() + h, err := newHarness(ctx, t) + require.NoError(t, err) + t.Cleanup(h.Close) + + store, err := h.MakeDriverWithMaxDest(ctx, 2) + require.NoError(t, err) + + tenant := models.Tenant{ID: idgen.String(), CreatedAt: time.Now()} + require.NoError(t, store.UpsertTenant(ctx, tenant)) + live := testutil.DestinationFactory.Any( + testutil.DestinationFactory.WithTenantID(tenant.ID), + testutil.DestinationFactory.WithExpiresAt(time.Now().Add(time.Hour)), + ) + expired := testutil.DestinationFactory.Any( + testutil.DestinationFactory.WithTenantID(tenant.ID), + testutil.DestinationFactory.WithExpiresAt(time.Now().Add(-time.Minute)), + ) + require.NoError(t, store.CreateDestination(ctx, live)) + require.NoError(t, store.CreateDestination(ctx, expired)) + + fresh := testutil.DestinationFactory.Any(testutil.DestinationFactory.WithTenantID(tenant.ID)) + require.NoError(t, store.CreateDestination(ctx, fresh)) + + _, err = store.RetrieveDestination(ctx, tenant.ID, expired.ID) + require.ErrorIs(t, err, driver.ErrDestinationDeleted) + listed, err := store.ListDestination(ctx, driver.ListDestinationRequest{TenantID: tenant.ID}) + require.NoError(t, err) + ids := make([]string, len(listed)) + for i, d := range listed { + ids[i] = d.ID + } + assert.ElementsMatch(t, []string{live.ID, fresh.ID}, ids) + + err = store.CreateDestination(ctx, testutil.DestinationFactory.Any(testutil.DestinationFactory.WithTenantID(tenant.ID))) + require.ErrorIs(t, err, driver.ErrMaxDestinationsPerTenantReached) + }) + + t.Run("MaxDestinationsPerTenantKeepsExtendedExpiry", func(t *testing.T) { + ctx := context.Background() + h, err := newHarness(ctx, t) + require.NoError(t, err) + t.Cleanup(h.Close) + + store, err := h.MakeDriverWithMaxDest(ctx, 1) + require.NoError(t, err) + + tenant := models.Tenant{ID: idgen.String(), CreatedAt: time.Now()} + require.NoError(t, store.UpsertTenant(ctx, tenant)) + d := testutil.DestinationFactory.Any( + testutil.DestinationFactory.WithTenantID(tenant.ID), + testutil.DestinationFactory.WithExpiresAt(time.Now().Add(-time.Minute)), + ) + require.NoError(t, store.CreateDestination(ctx, d)) + d.ExpiresAt = ptr(time.Now().Add(time.Hour)) + require.NoError(t, store.UpsertDestination(ctx, d)) + + err = store.CreateDestination(ctx, testutil.DestinationFactory.Any(testutil.DestinationFactory.WithTenantID(tenant.ID))) + require.ErrorIs(t, err, driver.ErrMaxDestinationsPerTenantReached) + _, err = store.RetrieveDestination(ctx, tenant.ID, d.ID) + require.NoError(t, err) + }) + t.Run("DeploymentIsolation", func(t *testing.T) { ctx := context.Background() h, err := newHarness(ctx, t) diff --git a/internal/tenantstore/memtenantstore/memtenantstore.go b/internal/tenantstore/memtenantstore/memtenantstore.go index 6fff3cdfb..cd391c593 100644 --- a/internal/tenantstore/memtenantstore/memtenantstore.go +++ b/internal/tenantstore/memtenantstore/memtenantstore.go @@ -353,15 +353,29 @@ func (s *store) CreateDestination(_ context.Context, destination models.Destinat return driver.ErrDuplicateDestination } - // Check max destinations - destIDs := s.destsByTenant[destination.TenantID] - if len(destIDs) >= s.maxDestinationsPerTenant { - return driver.ErrMaxDestinationsPerTenantReached + // Check max destinations; expired destinations are deleted to free a slot. + if len(s.destsByTenant[destination.TenantID]) >= s.maxDestinationsPerTenant { + s.deleteExpiredLocked(destination.TenantID) + if len(s.destsByTenant[destination.TenantID]) >= s.maxDestinationsPerTenant { + return driver.ErrMaxDestinationsPerTenantReached + } } return s.upsertDestinationLocked(destination) } +func (s *store) deleteExpiredLocked(tenantID string) { + now := time.Now() + for destID := range s.destsByTenant[tenantID] { + drec, ok := s.destinations[destKey(tenantID, destID)] + if !ok || !drec.destination.IsExpired(now) { + continue + } + drec.deletedAt = &now + delete(s.destsByTenant[tenantID], destID) + } +} + func (s *store) UpsertDestination(_ context.Context, destination models.Destination) error { s.mu.Lock() defer s.mu.Unlock() diff --git a/internal/tenantstore/redistenantstore/redistenantstore.go b/internal/tenantstore/redistenantstore/redistenantstore.go index 92be2161c..fb9dae0f8 100644 --- a/internal/tenantstore/redistenantstore/redistenantstore.go +++ b/internal/tenantstore/redistenantstore/redistenantstore.go @@ -15,10 +15,13 @@ import ( "github.com/hookdeck/outpost/internal/pagination" "github.com/hookdeck/outpost/internal/redis" "github.com/hookdeck/outpost/internal/tenantstore/driver" + goredis "github.com/redis/go-redis/v9" ) const defaultMaxDestinationsPerTenant = 20 +const deletedRecordTTL = 7 * 24 * time.Hour + const ( defaultListTenantLimit = 20 maxListTenantLimit = 100 @@ -256,12 +259,12 @@ func (s *store) DeleteTenant(ctx context.Context, tenantID string) error { for _, destinationID := range destinationIDs { destKey := s.redisDestinationID(destinationID, tenantID) pipe.HSet(ctx, destKey, "deleted_at", nowUnixMilli) - pipe.Expire(ctx, destKey, 7*24*time.Hour) + pipe.Expire(ctx, destKey, deletedRecordTTL) } pipe.Del(ctx, s.redisTenantDestinationSummaryKey(tenantID)) pipe.HSet(ctx, s.redisTenantID(tenantID), "deleted_at", nowUnixMilli) - pipe.Expire(ctx, s.redisTenantID(tenantID), 7*24*time.Hour) + pipe.Expire(ctx, s.redisTenantID(tenantID), deletedRecordTTL) return nil }) @@ -527,17 +530,62 @@ func (s *store) CreateDestination(ctx context.Context, destination models.Destin } } + // Check max destinations; expired destinations are deleted to free a slot. count, err := s.redisClient.HLen(ctx, s.redisTenantDestinationSummaryKey(destination.TenantID)).Result() if err != nil { return err } if count >= int64(s.maxDestinationsPerTenant) { - return driver.ErrMaxDestinationsPerTenantReached + if err := s.deleteExpiredDestinations(ctx, destination.TenantID); err != nil { + return err + } + count, err = s.redisClient.HLen(ctx, s.redisTenantDestinationSummaryKey(destination.TenantID)).Result() + if err != nil { + return err + } + if count >= int64(s.maxDestinationsPerTenant) { + return driver.ErrMaxDestinationsPerTenantReached + } } return s.UpsertDestination(ctx, destination) } +// Re-reads expires_at so a destination extended since the summary was read is kept. +var deleteIfExpiredScript = goredis.NewScript(` +local expires = redis.call('HGET', KEYS[1], 'expires_at') +if not expires or redis.call('HEXISTS', KEYS[1], 'deleted_at') == 1 then + return 0 +end +if tonumber(expires) > tonumber(ARGV[1]) then + return 0 +end +redis.call('HDEL', KEYS[2], ARGV[2]) +redis.call('HSET', KEYS[1], 'deleted_at', ARGV[1]) +redis.call('EXPIRE', KEYS[1], ARGV[3]) +return 1 +`) + +func (s *store) deleteExpiredDestinations(ctx context.Context, tenantID string) error { + summaries, err := s.listDestinationSummaryByTenant(ctx, tenantID, nil) + if err != nil { + return err + } + now := time.Now() + for _, ds := range summaries { + if !ds.expired(now) { + continue + } + err := deleteIfExpiredScript.Run(ctx, s.redisClient, + []string{s.redisDestinationID(ds.ID, tenantID), s.redisTenantDestinationSummaryKey(tenantID)}, + now.UnixMilli(), ds.ID, int64(deletedRecordTTL.Seconds())).Err() + if err != nil { + return err + } + } + return nil +} + func (s *store) UpsertDestination(ctx context.Context, destination models.Destination) error { key := s.redisDestinationID(destination.ID, destination.TenantID) @@ -591,6 +639,12 @@ func (s *store) UpsertDestination(ctx context.Context, destination models.Destin pipe.HDel(ctx, key, "disabled_at") } + if destination.ExpiresAt != nil { + pipe.HSet(ctx, key, "expires_at", destination.ExpiresAt.UnixMilli()) + } else { + pipe.HDel(ctx, key, "expires_at") + } + if destination.DeliveryMetadata != nil { pipe.HSet(ctx, key, "delivery_metadata", encryptedDeliveryMetadata) } else { @@ -631,7 +685,7 @@ func (s *store) DeleteDestination(ctx context.Context, tenantID, destinationID s pipe.HDel(ctx, summaryKey, destinationID) pipe.HSet(ctx, key, "deleted_at", nowUnixMilli) - pipe.Expire(ctx, key, 7*24*time.Hour) + pipe.Expire(ctx, key, deletedRecordTTL) return nil }) @@ -646,9 +700,10 @@ func (s *store) MatchEvent(ctx context.Context, event models.Event, allowWildcar } var matched []string + now := time.Now() for _, ds := range destinationSummaryList { - if ds.Disabled { + if ds.Disabled || ds.expired(now) { continue } if event.Topic != "" && !ds.Topics.MatchTopic(event.Topic, allowWildcards) { diff --git a/internal/tenantstore/redistenantstore/serialization.go b/internal/tenantstore/redistenantstore/serialization.go index 034213012..151b69f5f 100644 --- a/internal/tenantstore/redistenantstore/serialization.go +++ b/internal/tenantstore/redistenantstore/serialization.go @@ -14,21 +14,31 @@ import ( // destinationSummary is a package-private summary used for Redis storage. type destinationSummary struct { - ID string `json:"id"` - Type string `json:"type"` - Topics models.Topics `json:"topics"` - Filter models.Filter `json:"filter,omitempty"` - Disabled bool `json:"disabled"` + ID string `json:"id"` + Type string `json:"type"` + Topics models.Topics `json:"topics"` + Filter models.Filter `json:"filter,omitempty"` + Disabled bool `json:"disabled"` + ExpiresAt *int64 `json:"expires_at,omitempty"` +} + +func (ds *destinationSummary) expired(now time.Time) bool { + return ds.ExpiresAt != nil && now.UnixMilli() >= *ds.ExpiresAt } func newDestinationSummary(d models.Destination) *destinationSummary { - return &destinationSummary{ + ds := &destinationSummary{ ID: d.ID, Type: d.Type, Topics: d.Topics, Filter: d.Filter, Disabled: d.DisabledAt != nil, } + if d.ExpiresAt != nil { + ms := d.ExpiresAt.UnixMilli() + ds.ExpiresAt = &ms + } + return ds } func (ds *destinationSummary) MarshalBinary() ([]byte, error) { @@ -114,6 +124,13 @@ func parseDestinationHash(cmd *redis.MapStringStringCmd, tenantID string, cipher } } + if hash["expires_at"] != "" { + expiresAt, err := parseTimestamp(hash["expires_at"]) + if err == nil { + d.ExpiresAt = &expiresAt + } + } + if err := d.Topics.UnmarshalBinary([]byte(hash["topics"])); err != nil { return nil, fmt.Errorf("invalid topics: %w", err) } diff --git a/internal/util/testutil/destination.go b/internal/util/testutil/destination.go index 71306d60a..878c8a173 100644 --- a/internal/util/testutil/destination.go +++ b/internal/util/testutil/destination.go @@ -89,6 +89,12 @@ func (f *mockDestinationFactory) WithDisabledAt(disabledAt time.Time) func(*mode } } +func (f *mockDestinationFactory) WithExpiresAt(expiresAt time.Time) func(*models.Destination) { + return func(destination *models.Destination) { + destination.ExpiresAt = &expiresAt + } +} + func (f *mockDestinationFactory) WithDeliveryMetadata(deliveryMetadata map[string]string) func(*models.Destination) { return func(destination *models.Destination) { destination.DeliveryMetadata = deliveryMetadata From e02f93b4026bccdd0cb12a0660197e1fd75c4cf3 Mon Sep 17 00:00:00 2001 From: Alex Luong Date: Tue, 27 Oct 2026 12:09:59 +0700 Subject: [PATCH 2/2] feat: skip deliveries to expired destinations --- internal/deliverymq/messagehandler.go | 14 +++++- internal/deliverymq/messagehandler_test.go | 58 ++++++++++++++++++++++ 2 files changed, 70 insertions(+), 2 deletions(-) diff --git a/internal/deliverymq/messagehandler.go b/internal/deliverymq/messagehandler.go index b8a219478..538819afd 100644 --- a/internal/deliverymq/messagehandler.go +++ b/internal/deliverymq/messagehandler.go @@ -25,6 +25,7 @@ func idempotencyKeyFromDeliveryTask(task models.DeliveryTask) string { var ( errDestinationDisabled = errors.New("destination disabled") + errDestinationExpired = errors.New("destination expired") ) // Error types to distinguish between different stages of delivery @@ -168,7 +169,7 @@ func (h *messageHandler) handleError(msg *mqs.Message, err error) error { // Don't return error for expected cases var preErr *PreDeliveryError if errors.As(err, &preErr) { - if errors.Is(preErr.err, tenantstore.ErrDestinationDeleted) || errors.Is(preErr.err, errDestinationDisabled) { + if errors.Is(preErr.err, tenantstore.ErrDestinationDeleted) || errors.Is(preErr.err, errDestinationDisabled) || errors.Is(preErr.err, errDestinationExpired) { return nil } } @@ -376,7 +377,7 @@ func (h *messageHandler) shouldNackError(err error) bool { var preErr *PreDeliveryError if errors.As(err, &preErr) { // Don't nack if it's a permanent error - if errors.Is(preErr.err, tenantstore.ErrDestinationDeleted) || errors.Is(preErr.err, errDestinationDisabled) { + if errors.Is(preErr.err, tenantstore.ErrDestinationDeleted) || errors.Is(preErr.err, errDestinationDisabled) || errors.Is(preErr.err, errDestinationExpired) { return false } return true // Nack other pre-delivery errors @@ -464,5 +465,14 @@ func (h *messageHandler) ensurePublishableDestination(ctx context.Context, task zap.Time("disabled_at", *destination.DisabledAt)) return nil, errDestinationDisabled } + if destination.IsExpired(time.Now()) { + h.logger.Ctx(ctx).Debug("skipping expired destination", + zap.String("event_id", task.Event.ID), + zap.String("tenant_id", task.Event.TenantID), + zap.String("destination_id", destination.ID), + zap.String("destination_type", destination.Type), + zap.Time("expires_at", *destination.ExpiresAt)) + return nil, errDestinationExpired + } return destination, nil } diff --git a/internal/deliverymq/messagehandler_test.go b/internal/deliverymq/messagehandler_test.go index 31378897f..b3562da02 100644 --- a/internal/deliverymq/messagehandler_test.go +++ b/internal/deliverymq/messagehandler_test.go @@ -557,6 +557,64 @@ func TestMessageHandler_DestinationDisabled(t *testing.T) { assert.Empty(t, logPublisher.entries, "should not log delivery for pre-delivery error") } +func TestMessageHandler_DestinationExpired(t *testing.T) { + // Test scenario: + // - Destination is expired + // - Should be treated as a destination error (not system error) + // - Should ack without retry or publish attempt + + // Setup test data + tenant := models.Tenant{ID: idgen.String()} + destination := testutil.DestinationFactory.Any( + testutil.DestinationFactory.WithType("webhook"), + testutil.DestinationFactory.WithTenantID(tenant.ID), + testutil.DestinationFactory.WithExpiresAt(time.Now().Add(-time.Minute)), + ) + event := testutil.EventFactory.Any( + testutil.EventFactory.WithTenantID(tenant.ID), + testutil.EventFactory.WithDestinationID(destination.ID), + testutil.EventFactory.WithEligibleForRetry(false), + ) + + // Setup mocks + destGetter := &mockDestinationGetter{dest: &destination} + retryScheduler := newMockRetryScheduler() + publisher := newMockPublisher([]error{nil}) // won't be called + logPublisher := newMockLogPublisher(nil) + + // Setup message handler + handler := deliverymq.NewMessageHandler( + testutil.CreateTestLogger(t), + logPublisher, + destGetter, + publisher, + testutil.NewMockEventTracer(nil), + retryScheduler, + &backoff.ConstantBackoff{Interval: 1 * time.Second}, + 10, + idempotence.New(testutil.CreateTestRedisClient(t), idempotence.WithSuccessfulTTL(24*time.Hour)), + ) + + // Create and handle message + task := models.DeliveryTask{ + Event: event, + DestinationID: destination.ID, + } + mockMsg, msg := newDeliveryMockMessage(task) + + // Handle message + err := handler.Handle(context.Background(), msg) + require.NoError(t, err) + + // Assert behavior + assert.True(t, mockMsg.acked, "message should be acked for expired destination") + assert.False(t, mockMsg.nacked, "message should not be nacked for expired destination") + assert.Equal(t, 0, publisher.current, "should not attempt to publish to expired destination") + assert.Empty(t, retryScheduler.schedules, "should not schedule retry") + assert.Empty(t, retryScheduler.canceled, "should not attempt to cancel retries") + assert.Empty(t, logPublisher.entries, "should not log delivery for pre-delivery error") +} + func TestMessageHandler_LogPublisherError(t *testing.T) { // Test scenario: // - Publish succeeds but log publisher fails