internal/webhooks/webhook_service_test.go
Rama main · réplica de lectura del repositorio original.
package webhooks
import (
"bytes"
"context"
"database/sql"
"errors"
"strings"
"testing"
"time"
"github.com/google/uuid"
"github.com/lib/pq"
"github.com/riverqueue/river"
"github.com/riverqueue/river/rivertype"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
cryptosvc "github.com/sarathsp06/sparrow/internal/crypto"
svcerrors "github.com/sarathsp06/sparrow/internal/errors"
"github.com/sarathsp06/sparrow/internal/storage"
"github.com/sarathsp06/sparrow/internal/webhooks/queue"
"github.com/sarathsp06/sparrow/internal/webhooks/store"
)
type mockJobInserter struct {
mock.Mock
}
func (m *mockJobInserter) Insert(ctx context.Context, args river.JobArgs) (*rivertype.JobInsertResult, error) {
callArgs := m.Called(ctx, args)
res := callArgs.Get(0)
if res == nil {
return nil, callArgs.Error(1)
}
return res.(*rivertype.JobInsertResult), callArgs.Error(1)
}
func (m *mockJobInserter) BatchInsert(ctx context.Context, args []river.JobArgs) ([]*rivertype.JobInsertResult, error) {
callArgs := m.Called(ctx, args)
res := callArgs.Get(0)
if res == nil {
return nil, callArgs.Error(1)
}
return res.([]*rivertype.JobInsertResult), callArgs.Error(1)
}
func (m *mockJobInserter) BatchInsertTx(ctx context.Context, tx *sql.Tx, args []river.JobArgs) error {
return m.Called(ctx, tx, args).Error(0)
}
// CancelWaitingDeliveryJobsTx claims nothing; the store mock decides what a
// retry resets.
func (m *mockJobInserter) CancelWaitingDeliveryJobsTx(context.Context, *sql.Tx, []uuid.UUID) ([]uuid.UUID, error) {
return nil, nil
}
type mockRepo struct {
store.RepositoryInterface
mock.Mock
// gotClaim records whether the last ResetDeliveriesForRetry call was
// given a claimer (needed to retry a retrying delivery).
gotClaim bool
}
func (m *mockRepo) ListWebhooksPaginated(ctx context.Context, tenantID uuid.UUID, consumer, event string, activeOnly bool, health store.WebhookHealth, limit, offset int) ([]*store.WebhookRegistration, int, error) {
args := m.Called(ctx, tenantID, consumer, event, activeOnly, health, limit, offset)
return args.Get(0).([]*store.WebhookRegistration), args.Int(1), args.Error(2)
}
func (m *mockRepo) ListEventsPaginated(ctx context.Context, tenantID uuid.UUID, filter store.EventTypeFilter, limit, offset int) ([]*store.EventRegistration, int, error) {
args := m.Called(ctx, tenantID, filter, limit, offset)
return args.Get(0).([]*store.EventRegistration), args.Int(1), args.Error(2)
}
func (m *mockRepo) ListSubscriptions(ctx context.Context, tenantID uuid.UUID, webhookID uuid.UUID) ([]*store.EventSubscription, error) {
args := m.Called(ctx, tenantID, webhookID)
return args.Get(0).([]*store.EventSubscription), args.Error(1)
}
func (m *mockRepo) GetWebhookByID(ctx context.Context, tenantID uuid.UUID, webhookID uuid.UUID, consumer string) (*store.WebhookRegistration, error) {
args := m.Called(ctx, tenantID, webhookID, consumer)
res := args.Get(0)
if res == nil {
return nil, args.Error(1)
}
return res.(*store.WebhookRegistration), args.Error(1)
}
func (m *mockRepo) LockWebhook(ctx context.Context, tenantID uuid.UUID, webhookID uuid.UUID, consumer string, exclusive bool) (*store.WebhookRegistration, error) {
args := m.Called(ctx, tenantID, webhookID, consumer, exclusive)
res := args.Get(0)
if res == nil {
return nil, args.Error(1)
}
return res.(*store.WebhookRegistration), args.Error(1)
}
func (m *mockRepo) UpdateWebhook(ctx context.Context, tenantID uuid.UUID, webhook *store.WebhookRegistration) error {
args := m.Called(ctx, tenantID, webhook)
return args.Error(0)
}
func (m *mockRepo) RunInTransaction(fn func(store.RepositoryInterface) error) error {
return fn(m)
}
func (m *mockRepo) UpsertRateLimitState(ctx context.Context, _ uuid.UUID, webhookID uuid.UUID) error {
args := m.Called(ctx, webhookID)
return args.Error(0)
}
func (m *mockRepo) DeleteRateLimitState(ctx context.Context, _ uuid.UUID, webhookID uuid.UUID) error {
args := m.Called(ctx, webhookID)
return args.Error(0)
}
func (m *mockRepo) GetDeliveryByID(ctx context.Context, tenantID uuid.UUID, deliveryID uuid.UUID, consumer string) (*store.WebhookDelivery, error) {
args := m.Called(ctx, tenantID, deliveryID, consumer)
res := args.Get(0)
if res == nil {
return nil, args.Error(1)
}
return res.(*store.WebhookDelivery), args.Error(1)
}
func (m *mockRepo) GetRetriableDeliveries(ctx context.Context, tenantID uuid.UUID, webhookID uuid.UUID, consumer string, force bool) ([]*store.WebhookDelivery, error) {
args := m.Called(ctx, tenantID, webhookID, consumer, force)
return args.Get(0).([]*store.WebhookDelivery), args.Error(1)
}
// ResetDeliveriesForRetry resets the IDs the test returns and runs enqueue
// for them, as the store does inside its transaction.
func (m *mockRepo) ResetDeliveriesForRetry(ctx context.Context, _ uuid.UUID, deliveryIDs []uuid.UUID, claim store.RetryClaimer, enqueue store.RetryEnqueuer) ([]uuid.UUID, error) {
m.gotClaim = claim != nil
args := m.Called(ctx, deliveryIDs)
reset, _ := args.Get(0).([]uuid.UUID)
if err := args.Error(1); err != nil || len(reset) == 0 {
return nil, err
}
if err := enqueue(ctx, nil, reset); err != nil {
return nil, err
}
return reset, nil
}
func (m *mockRepo) ListDeliveriesFiltered(ctx context.Context, tenantID uuid.UUID, filter store.DeliveryFilter) ([]*store.WebhookDelivery, bool, error) {
args := m.Called(ctx, tenantID, filter)
return args.Get(0).([]*store.WebhookDelivery), args.Bool(1), args.Error(2)
}
func (m *mockRepo) CountDeliveries(ctx context.Context, tenantID uuid.UUID, filter store.DeliveryFilter) (int, error) {
args := m.Called(ctx, tenantID, filter)
return args.Int(0), args.Error(1)
}
func (m *mockRepo) ListEventReportsFiltered(ctx context.Context, tenantID uuid.UUID, filter store.EventReportFilter) ([]*store.EventReportWithStats, bool, error) {
args := m.Called(ctx, tenantID, filter)
return args.Get(0).([]*store.EventReportWithStats), args.Bool(1), args.Error(2)
}
func (m *mockRepo) CreateSubscription(ctx context.Context, tenantID uuid.UUID, sub *store.EventSubscription) error {
args := m.Called(ctx, tenantID, sub)
sub.ID = uuid.New()
sub.CreatedAt = time.Now()
return args.Error(0)
}
func (m *mockRepo) GetEventByName(ctx context.Context, tenantID uuid.UUID, eventName string) (*store.EventRegistration, error) {
args := m.Called(ctx, tenantID, eventName)
res := args.Get(0)
if res == nil {
return nil, args.Error(1)
}
return res.(*store.EventRegistration), args.Error(1)
}
func (m *mockRepo) RegisterEvent(ctx context.Context, tenantID uuid.UUID, event *store.EventRegistration) error {
args := m.Called(ctx, tenantID, event)
event.TenantID = tenantID
event.CreatedAt = time.Now()
event.UpdatedAt = time.Now()
return args.Error(0)
}
// StoreEventAndEnqueue stores the event as the test says and runs enqueue,
// as the store does inside its transaction.
func (m *mockRepo) StoreEventAndEnqueue(ctx context.Context, tenantID uuid.UUID, event *store.EventRecord, enqueue store.EventEnqueuer) error {
if err := m.Called(ctx, tenantID, event).Error(0); err != nil {
return err
}
return enqueue(ctx, nil)
}
func (m *mockRepo) StoreEvent(ctx context.Context, tenantID uuid.UUID, event *store.EventRecord) error {
args := m.Called(ctx, tenantID, event)
return args.Error(0)
}
func (m *mockRepo) GetEventByIdempotencyKey(ctx context.Context, tenantID uuid.UUID, consumer, idempotencyKey string) (*store.EventRecord, error) {
args := m.Called(ctx, tenantID, consumer, idempotencyKey)
res := args.Get(0)
if res == nil {
return nil, args.Error(1)
}
return res.(*store.EventRecord), args.Error(1)
}
// testContext returns a context for testing.
func testContext() context.Context {
return context.Background()
}
func TestWebhookService_ListWebhooks_Pagination(t *testing.T) {
repo := new(mockRepo)
inserter := new(mockJobInserter)
service := NewWebhookService(inserter, repo, nil)
ctx := testContext()
consumer := "default"
limit := int32(10)
offset := int32(0)
expectedWebhooks := []*store.WebhookRegistration{
{ID: uuid.New(), Consumer: consumer, URL: "http://example.com"},
}
repo.On("ListWebhooksPaginated", mock.Anything, mock.Anything, consumer, "", false, store.WebhookHealth(""), int(limit), int(offset)).
Return(expectedWebhooks, 1, nil)
webhooks, totalCount, err := service.ListWebhooks(ctx, consumer, "", "", false, "", limit, offset)
assert.NoError(t, err)
assert.Equal(t, int32(1), totalCount)
assert.Equal(t, len(expectedWebhooks), len(webhooks))
repo.AssertExpectations(t)
}
func TestWebhookService_RetryDelivery(t *testing.T) {
repo := new(mockRepo)
inserter := new(mockJobInserter)
service := NewWebhookService(inserter, repo, nil)
ctx := testContext()
consumer := "default"
deliveryID := uuid.New()
webhookID := uuid.New()
delivery := &store.WebhookDelivery{
ID: deliveryID,
WebhookID: webhookID,
EventID: uuid.New(),
Status: store.StatusFailed,
ExpiresAt: time.Now().Add(time.Hour),
}
webhook := &store.WebhookRegistration{
ID: webhookID,
Consumer: consumer,
}
repo.On("GetDeliveryByID", mock.Anything, mock.Anything, deliveryID, consumer).Return(delivery, nil)
repo.On("ResetDeliveriesForRetry", mock.Anything, []uuid.UUID{deliveryID}).Return([]uuid.UUID{deliveryID}, nil)
repo.On("GetWebhookByID", mock.Anything, mock.Anything, webhookID, consumer).Return(webhook, nil).Once()
inserter.On("BatchInsertTx", mock.Anything, mock.Anything, mock.MatchedBy(func(jobs []river.JobArgs) bool { return len(jobs) == 1 })).
Return(nil)
ids, count, err := service.RetryDelivery(ctx, consumer, deliveryID.String(), "", false)
assert.NoError(t, err)
assert.Equal(t, int32(1), count)
assert.Contains(t, ids, deliveryID.String())
repo.AssertExpectations(t)
inserter.AssertExpectations(t)
}
// A delivery the store does not reset (pending, or its job already picked up
// by a worker) is left to that job, even by a forced retry, and the caller
// gets FailedPrecondition instead of a second job.
func TestWebhookService_RetryDeliveryInFlight(t *testing.T) {
repo := new(mockRepo)
inserter := new(mockJobInserter)
service := NewWebhookService(inserter, repo, nil)
ctx := testContext()
deliveryID, webhookID := uuid.New(), uuid.New()
repo.On("GetDeliveryByID", mock.Anything, mock.Anything, deliveryID, "default").
Return(&store.WebhookDelivery{ID: deliveryID, WebhookID: webhookID, EventID: uuid.New(), Status: store.StatusRetrying}, nil)
repo.On("GetWebhookByID", mock.Anything, mock.Anything, webhookID, "default").Return(&store.WebhookRegistration{ID: webhookID, Consumer: "default"}, nil)
repo.On("ResetDeliveriesForRetry", mock.Anything, []uuid.UUID{deliveryID}).Return([]uuid.UUID(nil), nil)
_, _, err := service.RetryDelivery(ctx, "default", deliveryID.String(), "", true)
var svcErr *svcerrors.ServiceError
if !errors.As(err, &svcErr) || svcErr.Status != svcerrors.FailedPrecondition {
t.Fatalf("err = %v, want FailedPrecondition", err)
}
inserter.AssertNotCalled(t, "BatchInsertTx", mock.Anything, mock.Anything, mock.Anything)
assert.Contains(t, svcErr.Msg, "being sent right now", "the error says why and what to do")
}
// A retrying delivery whose waiting job the store's claim cancels is reset
// and queued like a failed one: the service always hands the store its
// claimer.
func TestWebhookService_RetryDeliveryRetryingPassesClaimer(t *testing.T) {
repo := new(mockRepo)
inserter := new(mockJobInserter)
service := NewWebhookService(inserter, repo, nil)
ctx := testContext()
deliveryID, webhookID := uuid.New(), uuid.New()
repo.On("GetDeliveryByID", mock.Anything, mock.Anything, deliveryID, "default").
Return(&store.WebhookDelivery{ID: deliveryID, WebhookID: webhookID, EventID: uuid.New(), Status: store.StatusRetrying, MaxAttempts: 4}, nil)
repo.On("GetWebhookByID", mock.Anything, mock.Anything, webhookID, "default").Return(&store.WebhookRegistration{ID: webhookID, Consumer: "default"}, nil)
repo.On("ResetDeliveriesForRetry", mock.Anything, []uuid.UUID{deliveryID}).Return([]uuid.UUID{deliveryID}, nil)
inserter.On("BatchInsertTx", mock.Anything, mock.Anything, mock.MatchedBy(func(jobs []river.JobArgs) bool { return len(jobs) == 1 })).Return(nil)
ids, count, err := service.RetryDelivery(ctx, "default", deliveryID.String(), "", false)
require.NoError(t, err)
assert.Equal(t, int32(1), count)
assert.Equal(t, []string{deliveryID.String()}, ids)
assert.True(t, repo.gotClaim, "the service must pass a claimer, or a retrying delivery is never retried")
inserter.AssertExpectations(t)
}
func TestWebhookService_ListDeliveries_Pagination(t *testing.T) {
repo := new(mockRepo)
inserter := new(mockJobInserter)
service := NewWebhookService(inserter, repo, nil)
ctx := testContext()
consumer := "default"
expectedDeliveries := []*store.WebhookDelivery{
{ID: uuid.New(), WebhookID: uuid.New(), EventID: uuid.New()},
}
// The service normalises the limit, so match the filter as built by the service.
repo.On("ListDeliveriesFiltered", mock.Anything, mock.Anything, mock.MatchedBy(func(f store.DeliveryFilter) bool {
return f.Consumer == consumer && f.Limit == 20
})).Return(expectedDeliveries, true, nil)
filter := store.DeliveryFilter{
Consumer: consumer,
Limit: 20,
}
page, err := service.ListDeliveries(ctx, filter)
assert.NoError(t, err)
assert.True(t, page.HasMore)
assert.Equal(t, len(expectedDeliveries), len(page.Items))
repo.AssertExpectations(t)
}
func TestWebhookService_CreateSubscription(t *testing.T) {
repo := new(mockRepo)
inserter := new(mockJobInserter)
service := NewWebhookService(inserter, repo, nil)
ctx := testContext()
webhookID := uuid.New().String()
consumer := "default"
eventName := "user.created"
repo.On("LockWebhook", mock.Anything, mock.Anything, mock.Anything, mock.Anything, false).Return(&store.WebhookRegistration{}, nil)
repo.On("CreateSubscription", mock.Anything, mock.Anything, mock.Anything).Return(nil)
id, createdAt, err := service.CreateSubscription(ctx, webhookID, eventName, consumer, nil, "POST", 30, false, "", nil, SubscriptionTemplateSettings{})
assert.NoError(t, err)
assert.NotEmpty(t, id)
assert.False(t, createdAt.IsZero())
repo.AssertExpectations(t)
}
func TestWebhookService_GetEvent(t *testing.T) {
repo := new(mockRepo)
service := NewWebhookService(nil, repo, nil)
ctx := testContext()
eventName := "test.event"
event := &store.EventRegistration{
Name: eventName,
}
repo.On("GetEventByName", mock.Anything, mock.Anything, eventName).Return(event, nil)
res, err := service.GetEvent(ctx, eventName)
assert.NoError(t, err)
assert.Equal(t, event, res)
repo.AssertExpectations(t)
}
func TestWebhookService_TestSubscriptionTemplate(t *testing.T) {
repo := new(mockRepo)
service := NewWebhookService(nil, repo, nil)
ctx := testContext()
eventName := "test.event"
event := &store.EventRegistration{
Name: eventName,
SamplePayload: map[string]any{
"id": "123",
},
}
repo.On("GetEventByName", mock.Anything, mock.Anything, eventName).Return(event, nil)
template := `{"new_id": "{{ .payload.id }}"}`
res, err := service.TestSubscriptionTemplate(ctx, eventName, template, "default", true, nil)
assert.NoError(t, err)
assert.Equal(t, `{"new_id": "123"}`, res, "nil payload renders the sample")
res, err = service.TestSubscriptionTemplate(ctx, eventName, template, "default", true, map[string]any{"id": "ord_7"})
assert.NoError(t, err)
assert.Equal(t, `{"new_id": "ord_7"}`, res, "a given payload replaces the sample")
_, err = service.TestSubscriptionTemplate(ctx, eventName, template, "default", true, map[string]any{"other": 1})
assert.Error(t, err, "strict render of a payload without the key fails")
res, err = service.TestSubscriptionTemplate(ctx, eventName, `{"n": {{len .payload}}}`, "default", true, map[string]any{})
assert.NoError(t, err)
assert.Equal(t, `{"n": 0}`, res, "an empty payload is used as given, not replaced by the sample")
repo.AssertExpectations(t)
}
func TestWebhookService_PushEvent_AutoRegister(t *testing.T) {
repo := new(mockRepo)
inserter := new(mockJobInserter)
service := NewWebhookService(inserter, repo, nil, WithAutoRegisterEvents(true))
ctx := testContext()
consumer := "default"
eventName := "user.signup"
payload := map[string]any{"user_id": "123"}
// Event does not exist — should be auto-registered
repo.On("GetEventByName", mock.Anything, mock.Anything, eventName).Return(nil, nil)
repo.On("RegisterEvent", mock.Anything, mock.Anything, mock.MatchedBy(func(e *store.EventRegistration) bool {
return e.Name == eventName && e.Active == true
})).Return(nil)
repo.On("StoreEventAndEnqueue", mock.Anything, mock.Anything, mock.Anything).Return(nil)
inserter.On("BatchInsertTx", mock.Anything, mock.Anything, mock.Anything).Return(nil)
eventID, _, _, _, err := service.PushEvent(ctx, consumer, eventName, payload, 0, nil, nil, nil, nil)
assert.NoError(t, err)
assert.NotEmpty(t, eventID)
repo.AssertExpectations(t)
inserter.AssertExpectations(t)
}
func TestWebhookService_PushEvent_ExistingEvent(t *testing.T) {
repo := new(mockRepo)
inserter := new(mockJobInserter)
service := NewWebhookService(inserter, repo, nil)
ctx := testContext()
consumer := "default"
eventName := "user.created"
payload := map[string]any{"user_id": "456"}
// Event already registered and active
repo.On("GetEventByName", mock.Anything, mock.Anything, eventName).Return(&store.EventRegistration{
Name: eventName,
Active: true,
}, nil)
repo.On("StoreEventAndEnqueue", mock.Anything, mock.Anything, mock.Anything).Return(nil)
inserter.On("BatchInsertTx", mock.Anything, mock.Anything, mock.Anything).Return(nil)
eventID, _, _, _, err := service.PushEvent(ctx, consumer, eventName, payload, 0, nil, nil, nil, nil)
assert.NoError(t, err)
assert.NotEmpty(t, eventID)
// RegisterEvent should NOT be called since event already exists
repo.AssertNotCalled(t, "RegisterEvent", mock.Anything, mock.Anything, mock.Anything)
repo.AssertExpectations(t)
inserter.AssertExpectations(t)
}
func TestWebhookService_PushEvent_Scheduled(t *testing.T) {
repo := new(mockRepo)
inserter := new(mockJobInserter)
service := NewWebhookService(inserter, repo, nil)
ctx := testContext()
repo.On("GetEventByName", mock.Anything, mock.Anything, "user.created").Return(&store.EventRegistration{
Name: "user.created",
Active: true,
}, nil)
var rec *store.EventRecord
repo.On("StoreEventAndEnqueue", mock.Anything, mock.Anything, mock.MatchedBy(func(e *store.EventRecord) bool {
rec = e
return true
})).Return(nil)
var jobArgs []river.JobArgs
inserter.On("BatchInsertTx", mock.Anything, mock.Anything, mock.MatchedBy(func(a []river.JobArgs) bool {
jobArgs = a
return true
})).Return(nil)
// Future scheduled_at lands on the record, the job args and the
// River insert opts; TTL counts from the scheduled time.
when := time.Now().Add(2 * time.Hour).Truncate(time.Second)
_, _, _, _, err := service.PushEvent(ctx, "default", "user.created", nil, 600, nil, nil, nil, &when)
require.NoError(t, err)
require.NotNil(t, rec.ScheduledAt)
assert.True(t, rec.ScheduledAt.Equal(when))
assert.True(t, rec.ExpiresAt.Equal(when.Add(600*time.Second)), "expiry anchors at the scheduled time, not the push")
ea := jobArgs[0].(queue.EventArgs)
assert.True(t, ea.ScheduledAt.Equal(when))
assert.True(t, ea.InsertOpts().ScheduledAt.Equal(when))
// A past scheduled_at means immediate: nothing scheduled anywhere.
past := time.Now().Add(-time.Minute)
_, _, _, _, err = service.PushEvent(ctx, "default", "user.created", nil, 0, nil, nil, nil, &past)
require.NoError(t, err)
assert.Nil(t, rec.ScheduledAt)
assert.True(t, jobArgs[0].(queue.EventArgs).ScheduledAt.IsZero())
// Beyond the one-year cap is rejected before anything is stored.
far := time.Now().Add(maxScheduleAhead + time.Hour)
_, _, _, _, err = service.PushEvent(ctx, "default", "user.created", nil, 0, nil, nil, nil, &far)
require.Error(t, err)
var svcErr *svcerrors.ServiceError
require.ErrorAs(t, err, &svcErr)
assert.Equal(t, svcerrors.InvalidArgument, svcErr.Status)
}
func TestWebhookService_PushEvent_InactiveEvent(t *testing.T) {
repo := new(mockRepo)
inserter := new(mockJobInserter)
service := NewWebhookService(inserter, repo, nil)
ctx := testContext()
// Event exists but inactive
repo.On("GetEventByName", mock.Anything, mock.Anything, "user.deleted").Return(&store.EventRegistration{
Name: "user.deleted",
Active: false,
}, nil)
_, _, _, _, err := service.PushEvent(ctx, "default", "user.deleted", nil, 0, nil, nil, nil, nil)
assert.Error(t, err)
assert.Contains(t, err.Error(), "inactive")
repo.AssertExpectations(t)
}
func TestWebhookService_CreateSubscription_CatchAll(t *testing.T) {
repo := new(mockRepo)
inserter := new(mockJobInserter)
service := NewWebhookService(inserter, repo, nil)
ctx := testContext()
webhookID := uuid.New().String()
consumer := "default"
repo.On("LockWebhook", mock.Anything, mock.Anything, mock.Anything, mock.Anything, false).Return(&store.WebhookRegistration{}, nil)
repo.On("CreateSubscription", mock.Anything, mock.Anything, mock.MatchedBy(func(sub *store.EventSubscription) bool {
return sub.EventName == store.CatchAllEventName
})).Return(nil)
id, createdAt, err := service.CreateSubscription(ctx, webhookID, store.CatchAllEventName, consumer, nil, "POST", 30, false, "", nil, SubscriptionTemplateSettings{})
assert.NoError(t, err)
assert.NotEmpty(t, id)
assert.False(t, createdAt.IsZero())
repo.AssertExpectations(t)
}
func TestWebhookService_UpdateWebhookConfig_MergesSecretHeaderChanges(t *testing.T) {
repo := new(mockRepo)
key := bytes.Repeat([]byte{1}, 32)
cryptoSvc, err := cryptosvc.NewService(key)
require.NoError(t, err)
service := NewWebhookService(nil, repo, cryptoSvc)
ctx := testContext()
consumer := "default"
webhookID := uuid.New()
existingHeaders := map[string]string{
"Authorization": "Bearer old",
"X-Trace": "keep-me",
"X-Remove": "drop-me",
}
encryptedHeaders, err := service.EncryptSecretHeaders(existingHeaders)
require.NoError(t, err)
repo.On("GetWebhookByID", mock.Anything, mock.Anything, webhookID, consumer).Return(&store.WebhookRegistration{
ID: webhookID,
Consumer: consumer,
URL: "https://example.com/webhook",
Active: true,
SecretHeaders: encryptedHeaders,
}, nil)
repo.On("UpdateWebhook", mock.Anything, mock.Anything, mock.MatchedBy(func(webhook *store.WebhookRegistration) bool {
decrypted, err := service.DecryptSecretHeaders(webhook.SecretHeaders)
require.NoError(t, err)
return assert.Equal(t, map[string]string{
"Authorization": "Bearer new",
"X-Trace": "keep-me",
}, decrypted)
})).Return(nil)
repo.On("DeleteRateLimitState", mock.Anything, webhookID).Return(nil)
err = service.UpdateWebhookConfig(ctx, webhookID.String(), consumer, nil, "", nil, true, "", nil, map[string]string{
"Authorization": "Bearer new",
"X-Remove": "",
}, "", false, []string{"secret_headers"})
require.NoError(t, err)
repo.AssertExpectations(t)
}
func TestWebhookService_UpdateWebhookConfig_RejectsOutOfBoundsValues(t *testing.T) {
repo := new(mockRepo)
service := NewWebhookService(nil, repo, nil)
ctx := testContext()
consumer := "default"
webhookID := uuid.New()
repo.On("GetWebhookByID", mock.Anything, mock.Anything, webhookID, consumer).Return(&store.WebhookRegistration{
ID: webhookID,
Consumer: consumer,
URL: "https://example.com/webhook",
Active: true,
MaxRetries: 3,
RetryBackoffSeconds: 60,
RequestTimeoutSeconds: 30,
ExpectedStatusCodes: pq.Int64Array{200},
ContentType: "application/json",
}, nil)
// request_timeout_seconds above the 300s create-path bound must be
// rejected on update too; UpdateWebhook must never be called.
err := service.UpdateWebhookConfig(ctx, webhookID.String(), consumer, nil, "", nil, true, "", &WebhookHTTPConfig{
RequestTimeoutSeconds: 86400,
}, nil, "", false, []string{"http_config"})
require.Error(t, err)
var svcErr *svcerrors.ServiceError
require.ErrorAs(t, err, &svcErr)
assert.Equal(t, svcerrors.InvalidArgument, svcErr.Status)
repo.AssertNotCalled(t, "UpdateWebhook", mock.Anything, mock.Anything, mock.Anything)
}
func TestWebhookService_RequiresTransform(t *testing.T) {
ctx := testContext()
const consumer = "default"
webhookID := uuid.New()
requiring := &store.WebhookRegistration{ID: webhookID, Consumer: consumer, RequiresTransform: true}
t.Run("create webhook with events needs a template", func(t *testing.T) {
service := NewWebhookService(nil, new(mockRepo), nil)
_, _, err := service.CreateWebhook(ctx, WebhookRegistrationRequest{
Consumer: consumer, URL: "https://example.com/hook", Events: []string{"order.created"}, RequiresTransform: true,
})
var svcErr *svcerrors.ServiceError
require.ErrorAs(t, err, &svcErr)
assert.Equal(t, svcerrors.InvalidArgument, svcErr.Status)
})
t.Run("subscription without a transform is refused", func(t *testing.T) {
repo := new(mockRepo)
service := NewWebhookService(nil, repo, nil)
repo.On("LockWebhook", mock.Anything, mock.Anything, webhookID, consumer, false).Return(requiring, nil)
_, _, err := service.CreateSubscription(ctx, webhookID.String(), "order.created", consumer, nil, "POST", 30, false, "", nil, SubscriptionTemplateSettings{})
var svcErr *svcerrors.ServiceError
require.ErrorAs(t, err, &svcErr)
assert.Equal(t, svcerrors.InvalidArgument, svcErr.Status)
_, _, err = service.CreateSubscription(ctx, webhookID.String(), "order.created", consumer, nil, "POST", 30, true, " ", nil, SubscriptionTemplateSettings{})
require.ErrorAs(t, err, &svcErr)
repo.AssertNotCalled(t, "CreateSubscription", mock.Anything, mock.Anything, mock.Anything)
})
t.Run("turning it on needs every subscription transformed", func(t *testing.T) {
repo := new(mockRepo)
service := NewWebhookService(nil, repo, nil)
repo.On("GetWebhookByID", mock.Anything, mock.Anything, webhookID, consumer).Return(&store.WebhookRegistration{ID: webhookID, Consumer: consumer}, nil)
repo.On("LockWebhook", mock.Anything, mock.Anything, webhookID, consumer, true).Return(&store.WebhookRegistration{ID: webhookID, Consumer: consumer}, nil)
repo.On("ListSubscriptions", mock.Anything, mock.Anything, webhookID).Return([]*store.EventSubscription{
{EventName: "order.created", TransformEnabled: true, TransformTemplate: "{}"},
{EventName: "order.refunded"},
}, nil)
err := service.UpdateWebhookConfig(ctx, webhookID.String(), consumer, nil, "", nil, false, "", nil, nil, "", true, []string{"requires_transform"})
var svcErr *svcerrors.ServiceError
require.ErrorAs(t, err, &svcErr)
assert.Equal(t, svcerrors.FailedPrecondition, svcErr.Status)
assert.Contains(t, svcErr.Error(), "order.refunded")
repo.AssertNotCalled(t, "UpdateWebhook", mock.Anything, mock.Anything, mock.Anything)
})
t.Run("events cannot be replaced in bulk", func(t *testing.T) {
repo := new(mockRepo)
service := NewWebhookService(nil, repo, nil)
repo.On("GetWebhookByID", mock.Anything, mock.Anything, webhookID, consumer).Return(requiring, nil)
err := service.UpdateWebhookConfig(ctx, webhookID.String(), consumer, []string{"order.created"}, "", nil, false, "", nil, nil, "", false, []string{"events"})
var svcErr *svcerrors.ServiceError
require.ErrorAs(t, err, &svcErr)
assert.Equal(t, svcerrors.FailedPrecondition, svcErr.Status)
})
}
func TestWebhookService_RotateWebhookSecret(t *testing.T) {
repo := new(mockRepo)
cryptoSvc, err := cryptosvc.NewService(bytes.Repeat([]byte{1}, 32))
require.NoError(t, err)
service := NewWebhookService(nil, repo, cryptoSvc)
ctx := testContext()
consumer := "default"
webhookID := uuid.New()
oldSecret, err := service.EncryptWebhookSecret("whsec_old")
require.NoError(t, err)
repo.On("GetWebhookByID", mock.Anything, mock.Anything, webhookID, consumer).Return(&store.WebhookRegistration{
ID: webhookID,
Consumer: consumer,
URL: "https://example.com/webhook",
Active: true,
WebhookSecret: oldSecret,
}, nil)
var stored []byte
repo.On("UpdateWebhook", mock.Anything, mock.Anything, mock.MatchedBy(func(webhook *store.WebhookRegistration) bool {
stored = webhook.WebhookSecret
return webhook.ID == webhookID && webhook.URL == "https://example.com/webhook"
})).Return(nil)
secret, err := service.RotateWebhookSecret(ctx, webhookID.String(), consumer)
require.NoError(t, err)
assert.True(t, strings.HasPrefix(secret, "whsec_"), "secret %q is not in Standard Webhooks format", secret)
assert.NotEqual(t, "whsec_old", secret)
decrypted, err := service.DecryptWebhookSecret(stored)
require.NoError(t, err)
assert.Equal(t, secret, decrypted, "the returned secret must be the one stored")
repo.AssertExpectations(t)
}
func (m *mockRepo) GetWebhookHealthState(ctx context.Context, _ uuid.UUID, webhookID uuid.UUID) (*store.WebhookHealthMetrics, error) {
args := m.Called(ctx, webhookID)
s, _ := args.Get(0).(*store.WebhookHealthMetrics)
return s, args.Error(1)
}
func (m *mockRepo) GetWebhookHealthSummary(ctx context.Context, _ uuid.UUID, webhookID uuid.UUID, hours int) (*store.WebhookHealthSummary, error) {
args := m.Called(ctx, webhookID, hours)
s, _ := args.Get(0).(*store.WebhookHealthSummary)
return s, args.Error(1)
}
// Before the health evaluator's first pass there is no health state row, but
// the attempts are already in the attempt log: their counts are reported
// (the API promises zeros only before the first attempt).
func TestWebhookService_GetWebhookHealthBeforeFirstEvaluation(t *testing.T) {
repo := new(mockRepo)
service := NewWebhookService(new(mockJobInserter), repo, nil)
ctx := testContext()
webhookID := uuid.New()
repo.On("GetWebhookByID", mock.Anything, mock.Anything, webhookID, "default").
Return(&store.WebhookRegistration{ID: webhookID, Consumer: "default", Health: store.HealthUnknown}, nil)
repo.On("GetWebhookHealthState", mock.Anything, webhookID).Return(nil, storage.ErrNotFound)
repo.On("GetWebhookHealthSummary", mock.Anything, webhookID, store.DefaultHealthRules.WindowHours).
Return(&store.WebhookHealthSummary{TotalDeliveries: 3, SuccessfulDeliveries: 1, FailedDeliveries: 2, SuccessRate: 1.0 / 3, ServerErrors: 2}, nil)
h, err := service.GetWebhookHealth(ctx, webhookID.String(), "default")
require.NoError(t, err)
assert.Equal(t, 3, h.TotalDeliveries)
assert.Equal(t, 1, h.SuccessfulDeliveries)
assert.Equal(t, 2, h.FailedDeliveries)
assert.Equal(t, 2, h.ServerErrors)
assert.Zero(t, h.ConsecutiveFailures, "state fields stay empty until the evaluator runs")
assert.Nil(t, h.LastFailureAt)
}