feat: add destination expiry

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))

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,

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"}))))

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
 }

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

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) {

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))
+}

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 }

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)