webhooker/internal/delivery/engine_test.go

package delivery_test

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
	_ "modernc.org/sqlite"
	"sneak.berlin/go/webhooker/internal/database"
	"sneak.berlin/go/webhooker/internal/delivery"
	"sneak.berlin/go/webhooker/internal/metrics"
)

// testContentType is the event content type used in tests.
const testContentType = "application/json"

func testWebhookDB(t *testing.T) *gorm.DB {
	t.Helper()

	dbPath := filepath.Join(
		t.TempDir(), "events-test.db",
	)

	// Opened the way the service opens a per-webhook database, so
	// these tests cannot pass against journal and locking settings
	// production does not use.
	sqlDB, err := database.OpenSQLite(
		dbPath, database.SQLiteModeCreate,
	)
	require.NoError(t, err)

	t.Cleanup(func() { _ = sqlDB.Close() })

	db, err := gorm.Open(
		sqlite.Dialector{Conn: sqlDB}, &gorm.Config{},
	)
	require.NoError(t, err)

	require.NoError(t, db.AutoMigrate(
		&database.Event{},
		&database.Delivery{},
		&database.DeliveryResult{},
	))

	return db
}

func testEngine(
	t *testing.T, workers int,
) *delivery.Engine {
	t.Helper()

	return delivery.NewTestEngine(
		slog.New(slog.NewTextHandler(
			os.Stderr,
			&slog.HandlerOptions{Level: slog.LevelDebug},
		)),
		&http.Client{Timeout: 5 * time.Second},
		workers,
	)
}

func newHTTPTargetConfig(url string) string {
	cfg := delivery.HTTPTargetConfig{URL: url}

	data, err := json.Marshal(cfg)
	if err != nil {
		panic(
			"failed to marshal HTTPTargetConfig: " +
				err.Error(),
		)
	}

	return string(data)
}

func seedEvent(
	t *testing.T, db *gorm.DB, body string,
) database.Event {
	t.Helper()

	event := database.Event{
		WebhookID:    uuid.New().String(),
		EntrypointID: uuid.New().String(),
		Method:       http.MethodPost,
		Headers:      `{"Content-Type":["application/json"]}`,
		Body:         body,
		ContentType:  testContentType,
	}

	require.NoError(t, db.Create(&event).Error)

	return event
}

func seedDelivery(
	t *testing.T,
	db *gorm.DB,
	eventID, targetID string,
	status database.DeliveryStatus,
) database.Delivery {
	t.Helper()

	d := database.Delivery{
		EventID:  eventID,
		TargetID: targetID,
		Status:   status,
	}

	require.NoError(t, db.Create(&d).Error)

	return d
}

// httpDeliveryFixture holds the task and delivery needed
// for HTTP delivery tests.
type httpDeliveryFixture struct {
	Task     *delivery.Task
	Delivery *database.Delivery
}

// buildHTTPFixture creates a task and delivery pair for
// HTTP delivery tests.
func buildHTTPFixture(
	dlv database.Delivery,
	event database.Event,
	targetID, name, config string,
	maxRetries, attemptNum int,
) httpDeliveryFixture {
	task := &delivery.Task{
		DeliveryID:   dlv.ID,
		EventID:      event.ID,
		WebhookID:    event.WebhookID,
		TargetID:     targetID,
		TargetName:   name,
		TargetType:   database.TargetTypeHTTP,
		TargetConfig: config,
		MaxRetries:   maxRetries,
		AttemptNum:   attemptNum,
	}

	d := &database.Delivery{
		EventID:  event.ID,
		TargetID: targetID,
		Status:   dlv.Status,
		Event:    event,
		Target: database.Target{
			Name:       name,
			Type:       database.TargetTypeHTTP,
			Config:     config,
			MaxRetries: maxRetries,
		},
	}
	d.ID = dlv.ID

	return httpDeliveryFixture{
		Task: task, Delivery: d,
	}
}

// assertDeliveryStatus checks that the delivery has the
// expected status in the database.
func assertDeliveryStatus(
	t *testing.T,
	db *gorm.DB,
	deliveryID string,
	expected database.DeliveryStatus,
) {
	t.Helper()

	var updated database.Delivery

	require.NoError(t, db.First(
		&updated, "id = ?", deliveryID,
	).Error)

	assert.Equal(t, expected, updated.Status)
}

// assertDeliveryResult checks that a delivery result
// exists with the expected success and status code.
func assertDeliveryResult(
	t *testing.T,
	db *gorm.DB,
	deliveryID string,
	success bool,
	statusCode int,
) {
	t.Helper()

	var result database.DeliveryResult

	require.NoError(t, db.Where(
		"delivery_id = ?", deliveryID,
	).First(&result).Error)

	assert.Equal(t, success, result.Success)

	assert.Equal(t, statusCode, result.StatusCode)
}

// --- Tests ---

func TestNotify_NonBlocking(t *testing.T) {
	t.Parallel()

	e := testEngine(t, 1)

	for range delivery.ExportDeliveryChannelSize {
		e.ExportDeliveryCh() <- delivery.Task{
			DeliveryID: "fill",
		}
	}

	done := make(chan struct{})

	go func() {
		e.Notify([]delivery.Task{
			{DeliveryID: "overflow-1"},
			{DeliveryID: "overflow-2"},
		})
		close(done)
	}()

	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal(
			"Notify blocked when delivery channel was full",
		)
	}
}

func TestDeliverHTTP_Success(t *testing.T) {
	t.Parallel()

	db := testWebhookDB(t)

	var received atomic.Bool

	ts := httptest.NewServer(
		http.HandlerFunc(
			func(w http.ResponseWriter, _ *http.Request) {
				received.Store(true)
				w.WriteHeader(http.StatusOK)
				_, _ = fmt.Fprint(w, `{"ok":true}`)
			},
		),
	)
	defer ts.Close()

	e := testEngine(t, 1)
	targetID := uuid.New().String()
	event := seedEvent(t, db, `{"hello":"world"}`)

	dlv := seedDelivery(
		t, db, event.ID, targetID,
		database.DeliveryStatusPending,
	)

	cfg := newHTTPTargetConfig(ts.URL)
	fix := buildHTTPFixture(
		dlv, event, targetID, "test-http", cfg, 0, 1,
	)

	e.ExportDeliverHTTP(
		context.TODO(), db, fix.Delivery, fix.Task,
	)

	assert.True(t, received.Load(),
		"HTTP target should have received request",
	)

	assertDeliveryStatus(t, db, dlv.ID,
		database.DeliveryStatusDelivered,
	)

	assertDeliveryResult(
		t, db, dlv.ID, true, http.StatusOK,
	)
}

func TestDeliverHTTP_Failure(t *testing.T) {
	t.Parallel()

	db := testWebhookDB(t)

	ts := httptest.NewServer(
		http.HandlerFunc(
			func(w http.ResponseWriter, _ *http.Request) {
				w.WriteHeader(http.StatusInternalServerError)
				_, _ = fmt.Fprint(w, "internal error")
			},
		),
	)
	defer ts.Close()

	e := testEngine(t, 1)
	targetID := uuid.New().String()
	event := seedEvent(t, db, `{"test":true}`)

	dlv := seedDelivery(
		t, db, event.ID, targetID,
		database.DeliveryStatusPending,
	)

	cfg := newHTTPTargetConfig(ts.URL)
	fix := buildHTTPFixture(
		dlv, event, targetID,
		"test-http-fail", cfg, 0, 1,
	)

	e.ExportDeliverHTTP(
		context.TODO(), db, fix.Delivery, fix.Task,
	)

	assertDeliveryStatus(t, db, dlv.ID,
		database.DeliveryStatusFailed,
	)

	assertDeliveryResult(
		t, db, dlv.ID, false,
		http.StatusInternalServerError,
	)
}

func TestDeliverDatabase_ImmediateSuccess(
	t *testing.T,
) {
	t.Parallel()

	db := testWebhookDB(t)

	// The database target archives for real now, so the engine
	// needs a webhook DB manager to locate the data directory.
	e := delivery.NewTestEngineWithDB(
		nil,
		database.NewTestWebhookDBManager(t.TempDir()),
		slog.New(slog.NewTextHandler(
			os.Stderr,
			&slog.HandlerOptions{Level: slog.LevelDebug},
		)),
		&http.Client{Timeout: 5 * time.Second},
		1,
	)

	event := seedEvent(t, db, `{"db":"target"}`)
	d := seedDatabaseTargetDelivery(t, db, event, "")

	e.ExportDeliverDatabase(db, d)

	var updated database.Delivery

	require.NoError(t, db.First(
		&updated, "id = ?", d.ID,
	).Error)

	assert.Equal(t,
		database.DeliveryStatusDelivered, updated.Status,
		"database target should immediately succeed",
	)

	var result database.DeliveryResult

	require.NoError(t, db.Where(
		"delivery_id = ?", d.ID,
	).First(&result).Error)

	assert.True(t, result.Success)

	assert.Equal(t, 0, result.StatusCode,
		"database target should not have an HTTP status",
	)
}

func TestDeliverLog_ImmediateSuccess(t *testing.T) {
	t.Parallel()

	db := testWebhookDB(t)
	e := testEngine(t, 1)

	event := seedEvent(t, db, `{"log":"target"}`)

	dlv := seedDelivery(
		t, db, event.ID, uuid.New().String(),
		database.DeliveryStatusPending,
	)

	d := &database.Delivery{
		EventID:  event.ID,
		TargetID: dlv.TargetID,
		Status:   database.DeliveryStatusPending,
		Event:    event,
		Target: database.Target{
			Name: "test-log",
			Type: database.TargetTypeLog,
		},
	}
	d.ID = dlv.ID

	e.ExportDeliverLog(db, d)

	var updated database.Delivery

	require.NoError(t, db.First(
		&updated, "id = ?", dlv.ID,
	).Error)

	assert.Equal(t,
		database.DeliveryStatusDelivered, updated.Status,
		"log target should immediately succeed",
	)

	var result database.DeliveryResult

	require.NoError(t, db.Where(
		"delivery_id = ?", dlv.ID,
	).First(&result).Error)

	assert.True(t, result.Success)
}

func TestDeliverHTTP_WithRetries_Success(t *testing.T) {
	t.Parallel()

	db := testWebhookDB(t)

	ts := httptest.NewServer(
		http.HandlerFunc(
			func(w http.ResponseWriter, _ *http.Request) {
				w.WriteHeader(http.StatusOK)
			},
		),
	)
	defer ts.Close()

	e := testEngine(t, 1)
	targetID := uuid.New().String()
	event := seedEvent(t, db, `{"retry":"ok"}`)

	dlv := seedDelivery(
		t, db, event.ID, targetID,
		database.DeliveryStatusPending,
	)

	cfg := newHTTPTargetConfig(ts.URL)
	fix := buildHTTPFixture(
		dlv, event, targetID,
		"test-http-retry", cfg, 5, 1,
	)

	e.ExportDeliverHTTP(
		context.TODO(), db, fix.Delivery, fix.Task,
	)

	assertDeliveryStatus(t, db, dlv.ID,
		database.DeliveryStatusDelivered,
	)

	cb := e.ExportGetCircuitBreaker(targetID)

	assert.Equal(t,
		delivery.CircuitClosed, cb.State(),
	)
}

func TestDeliverHTTP_MaxRetriesExhausted(t *testing.T) {
	t.Parallel()

	db := testWebhookDB(t)

	ts := httptest.NewServer(
		http.HandlerFunc(
			func(w http.ResponseWriter, _ *http.Request) {
				w.WriteHeader(http.StatusBadGateway)
			},
		),
	)
	defer ts.Close()

	e := testEngine(t, 1)
	targetID := uuid.New().String()
	event := seedEvent(t, db, `{"retry":"exhaust"}`)

	dlv := seedDelivery(
		t, db, event.ID, targetID,
		database.DeliveryStatusRetrying,
	)

	cfg := newHTTPTargetConfig(ts.URL)
	fix := buildHTTPFixture(
		dlv, event, targetID,
		"test-http-exhaust", cfg, 3, 3,
	)

	e.ExportDeliverHTTP(
		context.TODO(), db, fix.Delivery, fix.Task,
	)

	assertDeliveryStatus(t, db, dlv.ID,
		database.DeliveryStatusFailed,
	)
}

func TestDeliverHTTP_SchedulesRetryOnFailure(
	t *testing.T,
) {
	t.Parallel()

	db := testWebhookDB(t)

	ts := httptest.NewServer(
		http.HandlerFunc(
			func(w http.ResponseWriter, _ *http.Request) {
				w.WriteHeader(
					http.StatusServiceUnavailable,
				)
			},
		),
	)
	defer ts.Close()

	e := testEngine(t, 1)
	targetID := uuid.New().String()
	event := seedEvent(t, db, `{"retry":"schedule"}`)

	dlv := seedDelivery(
		t, db, event.ID, targetID,
		database.DeliveryStatusPending,
	)

	cfg := newHTTPTargetConfig(ts.URL)
	fix := buildHTTPFixture(
		dlv, event, targetID,
		"test-http-schedule", cfg, 5, 1,
	)

	e.ExportDeliverHTTP(
		context.TODO(), db, fix.Delivery, fix.Task,
	)

	assertDeliveryStatus(t, db, dlv.ID,
		database.DeliveryStatusRetrying,
	)

	assertDeliveryResult(
		t, db, dlv.ID, false,
		http.StatusServiceUnavailable,
	)
}

func TestExponentialBackoff_Durations(t *testing.T) {
	t.Parallel()

	expected := []time.Duration{
		1 * time.Second,
		2 * time.Second,
		4 * time.Second,
		8 * time.Second,
		16 * time.Second,
	}

	for attemptNum := 1; attemptNum <= 5; attemptNum++ {
		shift := attemptNum - 1

		shift = min(shift, 30)

		backoff := time.Duration(
			1<<uint(shift),
		) * time.Second

		assert.Equal(t,
			expected[attemptNum-1], backoff,
			"backoff for attempt %d should be %v",
			attemptNum,
			expected[attemptNum-1],
		)
	}
}

func TestExponentialBackoff_CappedAt30(t *testing.T) {
	t.Parallel()

	attemptNum := 50
	shift := attemptNum - 1

	shift = min(shift, 30)

	backoff := time.Duration(
		1<<uint(shift),
	) * time.Second

	assert.Equal(t,
		time.Duration(1<<30)*time.Second, backoff,
		"backoff shift should be capped at 30",
	)
}

func TestBodyPointer_SmallBodyInline(t *testing.T) {
	t.Parallel()

	smallBody := `{"small": true}`

	assert.Less(t,
		len(smallBody), delivery.MaxInlineBodySize,
	)

	var bodyPtr *string
	if len(smallBody) < delivery.MaxInlineBodySize {
		bodyPtr = &smallBody
	}

	require.NotNil(t, bodyPtr,
		"small body should be inline (non-nil)",
	)

	assert.Equal(t, smallBody, *bodyPtr)
}

func TestBodyPointer_LargeBodyNil(t *testing.T) {
	t.Parallel()

	largeBody := strings.Repeat(
		"x", delivery.MaxInlineBodySize,
	)

	assert.GreaterOrEqual(t,
		len(largeBody), delivery.MaxInlineBodySize,
	)

	var bodyPtr *string
	if len(largeBody) < delivery.MaxInlineBodySize {
		bodyPtr = &largeBody
	}

	assert.Nil(t, bodyPtr,
		"large body should be nil",
	)
}

func TestBodyPointer_ExactBoundary(t *testing.T) {
	t.Parallel()

	exactBody := strings.Repeat(
		"y", delivery.MaxInlineBodySize,
	)

	assert.Len(t, exactBody, delivery.MaxInlineBodySize)

	var bodyPtr *string
	if len(exactBody) < delivery.MaxInlineBodySize {
		bodyPtr = &exactBody
	}

	assert.Nil(t, bodyPtr,
		"body at exactly MaxInlineBodySize should be nil",
	)
}

// concurrencyTracker tracks concurrent request count.
type concurrencyTracker struct {
	mu      sync.Mutex
	current int
	maxSeen int
}

func (ct *concurrencyTracker) enter() {
	ct.mu.Lock()
	ct.current++

	if ct.current > ct.maxSeen {
		ct.maxSeen = ct.current
	}

	ct.mu.Unlock()
}

func (ct *concurrencyTracker) leave() {
	ct.mu.Lock()
	ct.current--
	ct.mu.Unlock()
}

func (ct *concurrencyTracker) max() int {
	ct.mu.Lock()
	defer ct.mu.Unlock()

	return ct.maxSeen
}

// seedConcurrencyTasks creates tasks and deliveries
// for the concurrency test.
func seedConcurrencyTasks(
	t *testing.T,
	db *gorm.DB,
	n int,
	targetCfg string,
) ([]database.Delivery, []delivery.Task) {
	t.Helper()

	tasks := make([]database.Delivery, n)
	dtasks := make([]delivery.Task, n)

	for i := range n {
		event := seedEvent(
			t, db, fmt.Sprintf(`{"task":%d}`, i),
		)

		dlv := seedDelivery(
			t, db, event.ID,
			uuid.New().String(),
			database.DeliveryStatusPending,
		)

		tasks[i] = database.Delivery{
			EventID:  event.ID,
			TargetID: dlv.TargetID,
			Status:   database.DeliveryStatusPending,
			Event:    event,
			Target: database.Target{
				Name:   fmt.Sprintf("task-%d", i),
				Type:   database.TargetTypeHTTP,
				Config: targetCfg,
			},
		}
		tasks[i].ID = dlv.ID

		dtasks[i] = delivery.Task{
			DeliveryID:   tasks[i].ID,
			EventID:      tasks[i].EventID,
			TargetID:     tasks[i].TargetID,
			TargetName:   tasks[i].Target.Name,
			TargetType:   tasks[i].Target.Type,
			TargetConfig: tasks[i].Target.Config,
			MaxRetries:   0,
			AttemptNum:   1,
		}
	}

	return tasks, dtasks
}

func TestWorkerPool_BoundedConcurrency(t *testing.T) {
	if testing.Short() {
		t.Skip("skipping concurrency test in short mode")
	}

	t.Parallel()

	const (
		numWorkers = 3
		numTasks   = 10
	)

	db := testWebhookDB(t)
	ct := &concurrencyTracker{}

	ts := httptest.NewServer(
		http.HandlerFunc(
			func(w http.ResponseWriter, _ *http.Request) {
				ct.enter()
				time.Sleep(100 * time.Millisecond)
				ct.leave()

				w.WriteHeader(http.StatusOK)
			},
		),
	)
	defer ts.Close()

	e := testEngine(t, numWorkers)
	cfg := newHTTPTargetConfig(ts.URL)
	tasks, dtasks := seedConcurrencyTasks(
		t, db, numTasks, cfg,
	)

	runConcurrencyWorkers(
		e, db, tasks, dtasks, numWorkers,
	)

	assert.LessOrEqual(t, ct.max(), numWorkers)

	assertAllDelivered(t, db, tasks)
}

func runConcurrencyWorkers(
	e *delivery.Engine,
	db *gorm.DB,
	tasks []database.Delivery,
	dtasks []delivery.Task,
	workers int,
) {
	var wg sync.WaitGroup

	taskCh := make(chan int, len(tasks))

	for i := range tasks {
		taskCh <- i
	}

	close(taskCh)

	for range workers {
		wg.Go(func() {
			for idx := range taskCh {
				e.ExportDeliverHTTP(
					context.TODO(), db,
					&tasks[idx],
					&dtasks[idx],
				)
			}
		})
	}

	wg.Wait()
}

func assertAllDelivered(
	t *testing.T,
	db *gorm.DB,
	tasks []database.Delivery,
) {
	t.Helper()

	for i := range tasks {
		assertDeliveryStatus(
			t, db, tasks[i].ID,
			database.DeliveryStatusDelivered,
		)
	}
}

func TestDeliverHTTP_CircuitBreakerBlocks(t *testing.T) {
	t.Parallel()

	db := testWebhookDB(t)
	e := testEngine(t, 1)
	targetID := uuid.New().String()

	cb := e.ExportGetCircuitBreaker(targetID)

	for range delivery.ExportDefaultFailureThreshold {
		cb.RecordFailure()
	}

	require.Equal(t, delivery.CircuitOpen, cb.State())

	event := seedEvent(t, db, `{"cb":"blocked"}`)

	dlv := seedDelivery(
		t, db, event.ID, targetID,
		database.DeliveryStatusPending,
	)

	cfg := newHTTPTargetConfig(
		"http://will-not-be-called.invalid",
	)
	fix := buildHTTPFixture(
		dlv, event, targetID,
		"test-cb-block", cfg, 5, 1,
	)

	e.ExportDeliverHTTP(
		context.TODO(), db, fix.Delivery, fix.Task,
	)

	assertDeliveryStatus(t, db, dlv.ID,
		database.DeliveryStatusRetrying,
	)

	var resultCount int64

	db.Model(&database.DeliveryResult{}).
		Where("delivery_id = ?", dlv.ID).
		Count(&resultCount)

	assert.Equal(t, int64(0), resultCount,
		"no delivery result when circuit is open",
	)
}

// recordingScheduler keeps the delay of every retry it is asked to
// schedule, and schedules nothing.
type recordingScheduler struct {
	delays []time.Duration
}

func (s *recordingScheduler) ScheduleRetry(
	_ delivery.Task, delay time.Duration,
) {
	s.delays = append(s.delays, delay)
}

// TestDeliverHTTP_HalfOpenBreakerDelaysQueuedTasks proves that while a
// half-open breaker's one probe delivery is in flight, every other task
// for the target is put back with a whole cooldown as its delay rather
// than none, and that its status is written the first time the breaker
// turns it away and not on each pass after that.
func TestDeliverHTTP_HalfOpenBreakerDelaysQueuedTasks(t *testing.T) {
	t.Parallel()

	db := testWebhookDB(t)
	e := testEngine(t, 1)

	// Every write of retrying moves the retry counter, so on a registry
	// this test owns the counter is the number of those writes.
	reg := prometheus.NewRegistry()
	e.ExportSetMetrics(metrics.New(reg))

	targetID := uuid.New().String()
	cb := newShortCooldownCB(t)
	e.ExportSetCircuitBreaker(targetID, cb)

	for range delivery.ExportDefaultFailureThreshold {
		cb.RecordFailure()
	}

	time.Sleep(60 * time.Millisecond)

	require.True(t, cb.Allow(), "the probe delivery should go through")
	require.Equal(t, delivery.CircuitHalfOpen, cb.State())

	cfg := newHTTPTargetConfig(
		"http://will-not-be-called.invalid",
	)
	sched := &recordingScheduler{}

	const queued, passes = 3, 4

	for range queued {
		event := seedEvent(t, db, `{"cb":"half-open"}`)
		dlv := seedDelivery(
			t, db, event.ID, targetID,
			database.DeliveryStatusPending,
		)

		for range passes {
			// Each pass starts from the stored row, as a retry does.
			var row database.Delivery

			require.NoError(t, db.First(
				&row, "id = ?", dlv.ID,
			).Error)

			fix := buildHTTPFixture(
				row, event, targetID,
				"test-cb-half-open", cfg, 5, 1,
			)

			e.ExportDeliverHTTPWithScheduler(
				context.TODO(), db, fix.Delivery, fix.Task, sched,
			)
		}

		assertDeliveryStatus(t, db, dlv.ID,
			database.DeliveryStatusRetrying,
		)
	}

	require.Len(t, sched.delays, queued*passes)

	for _, delay := range sched.delays {
		// The cooldown newShortCooldownCB gives the breaker.
		assert.Equal(t, 50*time.Millisecond, delay,
			"a task turned away while half-open should wait "+
				"a whole cooldown",
		)
	}

	assert.InDelta(t, float64(queued),
		mCounter(t, reg, mRetries, mTypeHTTP), 0,
		"status should be written once per task, not once per pass",
	)
}

func TestGetCircuitBreaker_CreatesOnDemand(t *testing.T) {
	t.Parallel()

	e := testEngine(t, 1)

	targetID := uuid.New().String()
	cb1 := e.ExportGetCircuitBreaker(targetID)

	require.NotNil(t, cb1)

	assert.Equal(t,
		delivery.CircuitClosed, cb1.State(),
	)

	cb2 := e.ExportGetCircuitBreaker(targetID)

	assert.Same(t, cb1, cb2,
		"same target ID should return the "+
			"same circuit breaker",
	)

	otherID := uuid.New().String()
	cb3 := e.ExportGetCircuitBreaker(otherID)

	assert.NotSame(t, cb1, cb3,
		"different target ID should return a "+
			"different circuit breaker",
	)
}

func TestParseHTTPConfig_Valid(t *testing.T) {
	t.Parallel()

	e := testEngine(t, 1)

	cfg, err := e.ExportParseHTTPConfig(
		`{"url":"https://example.com/hook",` +
			`"headers":{"X-Token":"secret"}}`,
	)

	require.NoError(t, err)
	assert.Equal(t, "https://example.com/hook", cfg.URL)
	assert.Equal(t, "secret", cfg.Headers["X-Token"])
}

func TestParseHTTPConfig_Empty(t *testing.T) {
	t.Parallel()

	e := testEngine(t, 1)

	_, err := e.ExportParseHTTPConfig("")

	assert.Error(t, err,
		"empty config should return error",
	)
}

func TestParseHTTPConfig_MissingURL(t *testing.T) {
	t.Parallel()

	e := testEngine(t, 1)

	_, err := e.ExportParseHTTPConfig(
		`{"headers":{"X-Token":"secret"}}`,
	)

	assert.Error(t, err,
		"config without URL should return error",
	)
}

func TestScheduleRetry_SendsToRetryChannel(
	t *testing.T,
) {
	t.Parallel()

	e := testEngine(t, 1)

	task := delivery.Task{
		DeliveryID: uuid.New().String(),
		EventID:    uuid.New().String(),
		WebhookID:  uuid.New().String(),
		TargetID:   uuid.New().String(),
		AttemptNum: 2,
	}

	e.ExportScheduleRetry(task, 10*time.Millisecond)

	select {
	case received := <-e.ExportRetryCh():
		assert.Equal(t,
			task.DeliveryID, received.DeliveryID,
		)

		assert.Equal(t,
			task.AttemptNum, received.AttemptNum,
		)
	case <-time.After(2 * time.Second):
		t.Fatal(
			"retry task was not sent to retry " +
				"channel within timeout",
		)
	}
}

func TestScheduleRetry_DropsWhenChannelFull(
	t *testing.T,
) {
	t.Parallel()

	e := delivery.NewTestEngineSmallRetry(
		slog.New(slog.NewTextHandler(
			os.Stderr,
			&slog.HandlerOptions{Level: slog.LevelDebug},
		)),
	)

	e.ExportRetryCh() <- delivery.Task{DeliveryID: "fill"}

	task := delivery.Task{
		DeliveryID: "overflow",
		AttemptNum: 2,
	}

	e.ExportScheduleRetry(task, 0)

	time.Sleep(50 * time.Millisecond)

	received := <-e.ExportRetryCh()

	assert.Equal(t, "fill", received.DeliveryID,
		"only the original task should be in the "+
			"channel (overflow was dropped)",
	)
}

func TestIsForwardableHeader(t *testing.T) {
	t.Parallel()

	assert.True(t,
		delivery.ExportIsForwardableHeader("X-Custom-Header"),
	)

	assert.True(t,
		delivery.ExportIsForwardableHeader("Authorization"),
	)

	assert.True(t,
		delivery.ExportIsForwardableHeader("Accept"),
	)

	assert.True(t,
		delivery.ExportIsForwardableHeader("X-GitHub-Event"),
	)

	assert.False(t,
		delivery.ExportIsForwardableHeader("Host"),
	)

	assert.False(t,
		delivery.ExportIsForwardableHeader("Connection"),
	)

	assert.False(t,
		delivery.ExportIsForwardableHeader("Keep-Alive"),
	)

	assert.False(t,
		delivery.ExportIsForwardableHeader(
			"Transfer-Encoding",
		),
	)

	assert.False(t,
		delivery.ExportIsForwardableHeader("Content-Length"),
	)
}

func TestTruncate(t *testing.T) {
	t.Parallel()

	assert.Equal(t,
		"hello", delivery.ExportTruncate("hello", 10),
	)

	assert.Equal(t,
		"hello", delivery.ExportTruncate("hello", 5),
	)

	assert.Equal(t,
		"hel", delivery.ExportTruncate("hello", 3),
	)

	assert.Empty(t, delivery.ExportTruncate("", 5))
}

func TestDoHTTPRequest_ForwardsHeaders(t *testing.T) {
	t.Parallel()

	var receivedHeaders http.Header

	ts := httptest.NewServer(
		http.HandlerFunc(
			func(w http.ResponseWriter, r *http.Request) {
				receivedHeaders = r.Header.Clone()

				w.WriteHeader(http.StatusOK)
			},
		),
	)
	defer ts.Close()

	e := testEngine(t, 1)

	cfg := &delivery.HTTPTargetConfig{
		URL: ts.URL,
		Headers: map[string]string{
			"X-Target-Auth": "bearer xyz",
		},
	}

	event := &database.Event{
		Method:      http.MethodPost,
		Headers:     `{"X-Custom":["value1"],"Content-Type":["application/json"]}`,
		Body:        `{"test":true}`,
		ContentType: testContentType,
	}

	statusCode, _, _, err := e.ExportDoHTTPRequest(
		context.TODO(), cfg, event,
	)

	require.NoError(t, err)

	assert.Equal(t, http.StatusOK, statusCode)

	assert.Equal(t,
		"value1",
		receivedHeaders.Get("X-Custom"),
	)

	assert.Equal(t,
		"bearer xyz",
		receivedHeaders.Get("X-Target-Auth"),
	)

	assert.Equal(t,
		testContentType,
		receivedHeaders.Get("Content-Type"),
	)

	assert.Equal(t,
		"webhooker/1.0",
		receivedHeaders.Get("User-Agent"),
	)
}

func TestProcessDelivery_RoutesToCorrectHandler(
	t *testing.T,
) {
	t.Parallel()

	db := testWebhookDB(t)

	// The database target archives for real now, so the engine
	// needs a webhook DB manager to locate the data directory.
	e := delivery.NewTestEngineWithDB(
		nil,
		database.NewTestWebhookDBManager(t.TempDir()),
		slog.New(slog.NewTextHandler(
			os.Stderr,
			&slog.HandlerOptions{Level: slog.LevelDebug},
		)),
		&http.Client{Timeout: 5 * time.Second},
		1,
	)

	tests := []struct {
		name       string
		targetType database.TargetType
		wantStatus database.DeliveryStatus
	}{
		{
			"database target",
			database.TargetTypeDatabase,
			database.DeliveryStatusDelivered,
		},
		{
			"log target",
			database.TargetTypeLog,
			database.DeliveryStatusDelivered,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()

			runRoutingSubtest(
				t, db, e, tt.targetType,
				tt.wantStatus,
			)
		})
	}
}

func runRoutingSubtest(
	t *testing.T,
	db *gorm.DB,
	e *delivery.Engine,
	targetType database.TargetType,
	wantStatus database.DeliveryStatus,
) {
	t.Helper()

	event := seedEvent(t, db, `{"routing":"test"}`)

	dlv := seedDelivery(
		t, db, event.ID,
		uuid.New().String(),
		database.DeliveryStatusPending,
	)

	d := &database.Delivery{
		EventID:  event.ID,
		TargetID: dlv.TargetID,
		Status:   database.DeliveryStatusPending,
		Event:    event,
		Target: database.Target{
			Name: "test-" + string(targetType),
			Type: targetType,
		},
	}
	d.ID = dlv.ID

	task := &delivery.Task{
		DeliveryID: dlv.ID,
		TargetType: targetType,
	}

	e.ExportProcessDelivery(
		context.TODO(), db, d, task,
	)

	assertDeliveryStatus(
		t, db, dlv.ID, wantStatus,
	)
}

func TestMaxInlineBodySize_Constant(t *testing.T) {
	t.Parallel()

	assert.Equal(t, 16*1024, delivery.MaxInlineBodySize,
		"MaxInlineBodySize should be 16KB",
	)
}

func TestParseSlackConfig_Valid(t *testing.T) {
	t.Parallel()

	e := testEngine(t, 1)

	cfg, err := e.ExportParseSlackConfig(
		`{"webhookUrl":"https://hooks.slack.com/services/T00/B00/xxx"}`,
	)

	require.NoError(t, err)

	assert.Equal(t,
		"https://hooks.slack.com/services/T00/B00/xxx",
		cfg.WebhookURL,
	)
}

func TestParseSlackConfig_Empty(t *testing.T) {
	t.Parallel()

	e := testEngine(t, 1)

	_, err := e.ExportParseSlackConfig("")

	assert.Error(t, err,
		"empty config should return error",
	)
}

func TestParseSlackConfig_MissingWebhookURL(
	t *testing.T,
) {
	t.Parallel()

	e := testEngine(t, 1)

	_, err := e.ExportParseSlackConfig(
		`{"other":"field"}`,
	)

	assert.Error(t, err,
		"config without webhook_url should return error",
	)
}

func TestFormatSlackMessage_JSONBody(t *testing.T) {
	t.Parallel()

	event := &database.Event{
		Method:      http.MethodPost,
		ContentType: testContentType,
		Body: `{"action":"push",` +
			`"repo":"test/repo",` +
			`"ref":"refs/heads/main"}`,
	}
	event.CreatedAt = time.Date(
		2025, 1, 15, 10, 30, 0, 0, time.UTC,
	)

	msg := delivery.FormatSlackMessage(event)

	assert.Contains(t, msg, "*Webhook Event Received*")
	assert.Contains(t, msg, "`POST`")
	assert.Contains(t, msg, "`application/json`")
	assert.Contains(t, msg, "```")
	assert.NotContains(t, msg, "```json")

	assert.Contains(t, msg, `  "action": "push"`)
	assert.Contains(t, msg, `  "repo": "test/repo"`)
}

func TestFormatSlackMessage_NonJSONBody(t *testing.T) {
	t.Parallel()

	event := &database.Event{
		Method:      http.MethodPost,
		ContentType: "text/plain",
		Body:        "hello world plain text",
	}
	event.CreatedAt = time.Date(
		2025, 1, 15, 10, 30, 0, 0, time.UTC,
	)

	msg := delivery.FormatSlackMessage(event)

	assert.Contains(t, msg, "*Webhook Event Received*")

	assert.Contains(t, msg,
		"```\nhello world plain text\n```",
	)

	assert.NotContains(t, msg, "```json")
}

func TestFormatSlackMessage_EmptyBody(t *testing.T) {
	t.Parallel()

	event := &database.Event{
		Method:      http.MethodPost,
		ContentType: testContentType,
		Body:        "",
	}
	event.CreatedAt = time.Date(
		2025, 1, 15, 10, 30, 0, 0, time.UTC,
	)

	msg := delivery.FormatSlackMessage(event)

	assert.Contains(t, msg, "_(empty body)_")
	assert.NotContains(t, msg, "```")
}

func TestFormatSlackMessage_LargeJSONTruncated(
	t *testing.T,
) {
	t.Parallel()

	largeObj := make(map[string]string)

	for i := range 200 {
		largeObj[fmt.Sprintf("key_%03d", i)] = strings.Repeat("v", 20)
	}

	largeJSON, err := json.Marshal(largeObj)
	require.NoError(t, err)

	event := &database.Event{
		Method:      http.MethodPost,
		ContentType: testContentType,
		Body:        string(largeJSON),
	}
	event.CreatedAt = time.Date(
		2025, 1, 15, 10, 30, 0, 0, time.UTC,
	)

	msg := delivery.FormatSlackMessage(event)

	assert.Contains(t, msg, "... (truncated)")
}

// buildSlackDelivery creates a database.Delivery
// with a Slack target config for testing.
func buildSlackDelivery(
	dlv database.Delivery,
	event database.Event,
	targetID, name, slackCfg string,
) *database.Delivery {
	d := &database.Delivery{
		EventID:  event.ID,
		TargetID: targetID,
		Status:   dlv.Status,
		Event:    event,
		Target: database.Target{
			Name:   name,
			Type:   database.TargetTypeSlack,
			Config: slackCfg,
		},
	}
	d.ID = dlv.ID

	return d
}

func TestDeliverSlack_Success(t *testing.T) {
	t.Parallel()

	db := testWebhookDB(t)

	var receivedBody string

	ts := httptest.NewServer(
		http.HandlerFunc(
			func(w http.ResponseWriter, r *http.Request) {
				bodyBytes, readErr := readAll(r.Body)
				if readErr != nil {
					http.Error(
						w, "read error",
						http.StatusInternalServerError,
					)

					return
				}

				receivedBody = string(bodyBytes)

				w.WriteHeader(http.StatusOK)
				_, _ = fmt.Fprint(w, "ok")
			},
		),
	)
	defer ts.Close()

	e, targetID, slackCfg, event, dlv :=
		setupSlackTest(t, db, ts.URL)

	d := buildSlackDelivery(
		dlv, event, targetID, "test-slack", slackCfg,
	)

	e.ExportDeliverSlack(context.TODO(), db, d)

	assertDeliveryStatus(t, db, dlv.ID,
		database.DeliveryStatusDelivered,
	)

	assertDeliveryResult(
		t, db, dlv.ID, true, http.StatusOK,
	)

	assertSlackPayload(t, receivedBody)
}

func setupSlackTest(
	t *testing.T,
	db *gorm.DB,
	serverURL string,
) (
	*delivery.Engine, string, string,
	database.Event, database.Delivery,
) {
	t.Helper()

	e := testEngine(t, 1)
	targetID := uuid.New().String()

	cfgBytes, err := json.Marshal(
		delivery.SlackTargetConfig{
			WebhookURL: serverURL,
		},
	)
	require.NoError(t, err)

	event := seedEvent(
		t, db, `{"action":"test","data":"value"}`,
	)

	dlv := seedDelivery(
		t, db, event.ID, targetID,
		database.DeliveryStatusPending,
	)

	return e, targetID, string(cfgBytes), event, dlv
}

func assertSlackPayload(
	t *testing.T, receivedBody string,
) {
	t.Helper()

	var slackPayload map[string]string

	require.NoError(t, json.Unmarshal(
		[]byte(receivedBody), &slackPayload,
	))

	assert.Contains(t,
		slackPayload["text"],
		"*Webhook Event Received*",
	)

	assert.NotContains(t,
		slackPayload["text"],
		"**Webhook Event Received**",
	)

	assert.Contains(t, slackPayload["text"], "```")

	assert.NotContains(t,
		slackPayload["text"], "```json",
	)
}

func TestDeliverSlack_Failure(t *testing.T) {
	t.Parallel()

	db := testWebhookDB(t)

	ts := httptest.NewServer(
		http.HandlerFunc(
			func(w http.ResponseWriter, _ *http.Request) {
				w.WriteHeader(http.StatusForbidden)
				_, _ = fmt.Fprint(w, "invalid_token")
			},
		),
	)
	defer ts.Close()

	e := testEngine(t, 1)
	targetID := uuid.New().String()

	slackCfg, err := json.Marshal(
		delivery.SlackTargetConfig{
			WebhookURL: ts.URL,
		},
	)
	require.NoError(t, err)

	event := seedEvent(t, db, `{"test":true}`)

	dlv := seedDelivery(
		t, db, event.ID, targetID,
		database.DeliveryStatusPending,
	)

	d := buildSlackDelivery(
		dlv, event, targetID,
		"test-slack-fail", string(slackCfg),
	)

	e.ExportDeliverSlack(context.TODO(), db, d)

	assertDeliveryStatus(t, db, dlv.ID,
		database.DeliveryStatusFailed,
	)

	assertDeliveryResult(
		t, db, dlv.ID, false, http.StatusForbidden,
	)
}

func TestDeliverSlack_InvalidConfig(t *testing.T) {
	t.Parallel()

	db := testWebhookDB(t)
	e := testEngine(t, 1)

	event := seedEvent(t, db, `{"test":true}`)

	dlv := seedDelivery(
		t, db, event.ID, uuid.New().String(),
		database.DeliveryStatusPending,
	)

	d := &database.Delivery{
		EventID:  event.ID,
		TargetID: dlv.TargetID,
		Status:   database.DeliveryStatusPending,
		Event:    event,
		Target: database.Target{
			Name:   "test-slack-bad",
			Type:   database.TargetTypeSlack,
			Config: `{"not_webhook_url":"missing"}`,
		},
	}
	d.ID = dlv.ID

	e.ExportDeliverSlack(context.TODO(), db, d)

	var updated database.Delivery

	require.NoError(t, db.First(
		&updated, "id = ?", dlv.ID,
	).Error)

	assert.Equal(t,
		database.DeliveryStatusFailed, updated.Status,
	)
}

func TestProcessDelivery_RoutesToSlack(t *testing.T) {
	t.Parallel()

	db := testWebhookDB(t)

	var received atomic.Bool

	ts := httptest.NewServer(
		http.HandlerFunc(
			func(w http.ResponseWriter, _ *http.Request) {
				received.Store(true)
				w.WriteHeader(http.StatusOK)
			},
		),
	)
	defer ts.Close()

	e := testEngine(t, 1)

	slackCfg, err := json.Marshal(
		delivery.SlackTargetConfig{
			WebhookURL: ts.URL,
		},
	)
	require.NoError(t, err)

	event := seedEvent(t, db, `{"route":"slack"}`)

	dlv := seedDelivery(
		t, db, event.ID, uuid.New().String(),
		database.DeliveryStatusPending,
	)

	d := buildSlackDelivery(
		dlv, event, dlv.TargetID,
		"test-slack-route", string(slackCfg),
	)

	task := &delivery.Task{
		DeliveryID: dlv.ID,
		TargetType: database.TargetTypeSlack,
	}

	e.ExportProcessDelivery(
		context.TODO(), db, d, task,
	)

	assert.True(t, received.Load())

	assertDeliveryStatus(t, db, dlv.ID,
		database.DeliveryStatusDelivered,
	)
}

// newLogCaptureEngine builds a test engine whose logger
// writes to the returned buffer, for inspecting log output.
func newLogCaptureEngine(
	t *testing.T,
) (*delivery.Engine, *bytes.Buffer) {
	t.Helper()

	var buf bytes.Buffer

	log := slog.New(slog.NewTextHandler(
		&buf,
		&slog.HandlerOptions{Level: slog.LevelDebug},
	))

	e := delivery.NewTestEngine(
		log, &http.Client{Timeout: 5 * time.Second}, 1,
	)

	return e, &buf
}

// assertLogLineComplete asserts the captured log output
// carries the full inbound webhook content and ids.
func assertLogLineComplete(
	t *testing.T, out string, event database.Event,
) {
	t.Helper()

	assert.Contains(t, out, "log-body-marker",
		"log line must contain the full request body",
	)

	assert.Contains(t, out, "Content-Type",
		"log line must contain the full request headers",
	)

	assert.Contains(t, out, event.EntrypointID,
		"log line must contain the entrypoint id",
	)

	assert.Contains(t, out, event.WebhookID,
		"log line must contain the webhook id",
	)

	assert.Contains(t, out, testContentType,
		"log line must contain the content type",
	)
}

func TestDeliverLog_LogsFullContent(t *testing.T) {
	t.Parallel()

	db := testWebhookDB(t)
	e, buf := newLogCaptureEngine(t)

	event := seedEvent(
		t, db, `{"log-body-marker":"abc123"}`,
	)

	dlv := seedDelivery(
		t, db, event.ID, uuid.New().String(),
		database.DeliveryStatusPending,
	)

	d := &database.Delivery{
		EventID:  event.ID,
		TargetID: dlv.TargetID,
		Status:   database.DeliveryStatusPending,
		Event:    event,
		Target: database.Target{
			Name: "test-log-full",
			Type: database.TargetTypeLog,
		},
	}
	d.ID = dlv.ID

	e.ExportDeliverLog(db, d)

	assertLogLineComplete(t, buf.String(), event)

	assertDeliveryStatus(t, db, dlv.ID,
		database.DeliveryStatusDelivered,
	)
}

// buildSlackRetryDelivery builds a Slack delivery whose
// target is configured with retries enabled.
func buildSlackRetryDelivery(
	dlv database.Delivery,
	event database.Event,
	targetID, cfg string,
) *database.Delivery {
	d := &database.Delivery{
		EventID:  event.ID,
		TargetID: targetID,
		Status:   database.DeliveryStatusPending,
		Event:    event,
		Target: database.Target{
			Name:       "test-slack-retry",
			Type:       database.TargetTypeSlack,
			Config:     cfg,
			MaxRetries: 5,
		},
	}
	d.ID = dlv.ID

	return d
}

func TestDeliverSlack_WithRetries_SchedulesRetry(
	t *testing.T,
) {
	t.Parallel()

	db := testWebhookDB(t)
	ts := newStatusServer(t, http.StatusServiceUnavailable)
	e := testEngine(t, 1)
	targetID := uuid.New().String()

	slackCfg, err := json.Marshal(
		delivery.SlackTargetConfig{WebhookURL: ts.URL},
	)
	require.NoError(t, err)

	event := seedEvent(t, db, `{"slack":"retry"}`)

	dlv := seedDelivery(
		t, db, event.ID, targetID,
		database.DeliveryStatusPending,
	)

	d := buildSlackRetryDelivery(
		dlv, event, targetID, string(slackCfg),
	)

	task := &delivery.Task{
		DeliveryID: dlv.ID,
		TargetID:   targetID,
		TargetType: database.TargetTypeSlack,
		MaxRetries: 5,
		AttemptNum: 1,
	}

	e.ExportProcessDelivery(context.TODO(), db, d, task)

	assertDeliveryStatus(t, db, dlv.ID,
		database.DeliveryStatusRetrying,
	)

	assertDeliveryResult(
		t, db, dlv.ID, false,
		http.StatusServiceUnavailable,
	)
}

// newStatusServer starts a test server that always responds
// with the given status code.
func newStatusServer(
	t *testing.T, code int,
) *httptest.Server {
	t.Helper()

	ts := httptest.NewServer(http.HandlerFunc(
		func(w http.ResponseWriter, _ *http.Request) {
			w.WriteHeader(code)
		},
	))

	t.Cleanup(ts.Close)

	return ts
}

// readAll is a small helper to avoid importing io in
// a test handler inline.
func readAll(r interface {
	Read(p []byte) (n int, err error)
}) ([]byte, error) {
	var buf []byte

	tmp := make([]byte, 1024)

	for {
		n, err := r.Read(tmp)
		buf = append(buf, tmp[:n]...)

		if err != nil {
			if err.Error() == "EOF" {
				return buf, nil
			}

			return buf, err
		}
	}
}