webhooker/internal/delivery/engine.go

// Package delivery manages asynchronous event delivery
// to configured targets.
package delivery

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"sync"
	"time"

	"go.uber.org/fx"
	"gorm.io/gorm"
	"sneak.berlin/go/webhooker/internal/database"
	"sneak.berlin/go/webhooker/internal/lifecycle"
	"sneak.berlin/go/webhooker/internal/logger"
	"sneak.berlin/go/webhooker/internal/metrics"
)

const (
	// deliveryChannelSize is the buffer size for the delivery
	// channel. New Tasks from the webhook handler are sent
	// here. Workers drain this channel. Sized large enough
	// that the webhook handler should never block under
	// normal load.
	deliveryChannelSize = 10000

	// retryChannelSize is the buffer size for the retry
	// channel. Timer-fired retries are sent here for
	// processing by workers.
	retryChannelSize = 10000

	// defaultWorkers is the number of worker goroutines in
	// the delivery engine pool. At most this many deliveries
	// are in-flight at any time, preventing goroutine
	// explosions regardless of queue depth.
	defaultWorkers = 10

	// retrySweepInterval is how often the periodic retry
	// sweep runs.
	retrySweepInterval = 60 * time.Second

	// pendingSweepMinAge is how long a delivery must have sat
	// untouched at pending before the sweep will look at it.
	//
	// It is not what keeps the sweep off live work — inflightSet is,
	// and it is exact. This bound sets the re-dispatch cadence for a
	// delivery that really is stranded: without it, a delivery the
	// database will not let the engine settle would be re-sent on
	// every 60-second tick.
	//
	// It is nonetheless set clear of the longest legitimate attempt,
	// so that the two guards do not both have to be right. That
	// length is MaxTargetTimeoutSeconds (300s), the per-target
	// timeout the target form accepts — not httpClientTimeout, which
	// is merely the default. Fifteen minutes leaves a margin of
	// three times the ceiling rather than the zero margin the two
	// equal values would have given.
	pendingSweepMinAge = 15 * time.Minute

	// pendingSweepBatch bounds how many stranded pending deliveries
	// one sweep of one webhook re-dispatches. The sweep runs every
	// retrySweepInterval, so a larger backlog drains across
	// successive sweeps instead of arriving as one burst against a
	// database that was already struggling to accept writes.
	pendingSweepBatch = 500

	// MaxInlineBodySize is the maximum event body size that
	// will be carried inline in a Task through the channel.
	// Bodies at or above this size are left nil and fetched
	// from the per-webhook database on demand.
	MaxInlineBodySize = 16 * 1024

	// httpClientTimeout is the timeout for outbound HTTP
	// requests.
	httpClientTimeout = 30 * time.Second

	// maxBodyLog is the maximum response body length to
	// store in DeliveryResult.
	maxBodyLog = 4096

	// maxBackoffShift caps the exponential backoff shift to
	// avoid integer overflow in the 1<<shift expression.
	maxBackoffShift = 30

	// httpSuccessMin is the lower bound (inclusive) of the
	// HTTP success status code range.
	httpSuccessMin = 200

	// httpSuccessMax is the upper bound (exclusive) of the
	// HTTP success status code range.
	httpSuccessMax = 300
)

// Task contains everything needed to deliver an event to a
// single target.
type Task struct {
	DeliveryID   string
	EventID      string
	WebhookID    string
	EntrypointID string

	TargetID     string
	TargetName   string
	TargetType   database.TargetType
	TargetConfig string
	MaxRetries   int

	Method      string
	Headers     string
	ContentType string
	Body        *string

	AttemptNum int
}

// Notifier is the interface for notifying the delivery
// engine about new deliveries.
type Notifier interface {
	Notify(tasks []Task)
}

// WebhookEvictor releases the delivery engine's per-webhook
// state for a webhook that no longer needs it — currently the
// cached archive writer of the database target, whose open
// file handle would otherwise outlive the webhook.
//
// It is deliberately separate from Notifier and deliberately
// one method wide: archiving lifecycle is not notification, and
// a single-method interface keeps the handlers package free of
// any dependency on the engine's internals while staying
// trivially fakeable in tests.
//
// EvictWebhook never deletes an archive file. It is idempotent
// and is a no-op for a webhook with no engine state.
type WebhookEvictor interface {
	EvictWebhook(webhookID string)
}

// EngineParams are the fx dependencies for the delivery
// engine.
type EngineParams struct {
	fx.In

	DB        *database.Database
	DBManager *database.WebhookDBManager
	Logger    *logger.Logger
	SSRFGuard *Guard
}

// Engine processes queued deliveries in the background
// using a bounded worker pool architecture. It owns only
// the cross-target machinery: the worker pool, the queue
// and retry channels, restart recovery, and the persistence
// helpers and Scheduler that individual targets rely on.
// Each target type owns its own delivery, including retries,
// backoff, and circuit breaking.
type Engine struct {
	database   *database.Database
	dbManager  *database.WebhookDBManager
	log        *slog.Logger
	cancel     context.CancelFunc
	wg         sync.WaitGroup
	deliveryCh chan Task
	retryCh    chan Task
	workers    int

	// mtr is the delivery metric set. Production wires the
	// process-wide one; a test can substitute a set registered on
	// a private registry so its assertions are not disturbed by
	// deliveries other tests are making at the same time.
	mtr *metrics.Set

	// targets maps each target type to its implementation.
	targets map[database.TargetType]Target

	// httpTarget is retained so tests can reach the HTTP
	// target's shared client and circuit breakers.
	httpTarget *httpTarget

	// dbTarget is retained so the engine can reach the archive
	// writer registry for webhook eviction and the idle sweep.
	dbTarget *databaseTarget

	// inflight is the set of deliveries this engine currently owns.
	// Recovery and the sweeps re-dispatch only what it does not
	// hold. Held by value: its zero value works, so no constructor
	// can leave it out. See inflight.go.
	inflight inflightSet
}

// New creates and registers the delivery engine with the
// fx lifecycle.
func New(
	lc fx.Lifecycle,
	params EngineParams,
) *Engine {
	e := &Engine{
		database:   params.DB,
		dbManager:  params.DBManager,
		log:        params.Logger.Get(),
		deliveryCh: make(chan Task, deliveryChannelSize),
		retryCh:    make(chan Task, retryChannelSize),
		workers:    defaultWorkers,
		mtr:        metrics.Default(),
	}

	e.initTargets(&http.Client{
		Timeout:   httpClientTimeout,
		Transport: params.SSRFGuard.NewSSRFSafeTransport(),
	})

	e.registerHooks(lc)

	return e
}

// Notify signals the delivery engine that new deliveries
// are ready.
func (e *Engine) Notify(tasks []Task) {
	for i := range tasks {
		// Owned before it is queued, and until the worker that runs
		// it returns. A task can sit in a 10000-deep channel for a
		// long time on a healthy system, and nothing may re-send it
		// while it waits. See inflight.go.
		if !e.inflight.retainIdle(tasks[i].DeliveryID) {
			e.log.Warn(
				"delivery already in flight, not queued again",
				"delivery_id", tasks[i].DeliveryID,
				"event_id", tasks[i].EventID,
			)

			continue
		}

		select {
		case e.deliveryCh <- tasks[i]:
		default:
			e.inflight.release(tasks[i].DeliveryID)
			e.log.Warn(
				"delivery channel full, "+
					"task will be recovered by the sweep",
				"delivery_id", tasks[i].DeliveryID,
				"event_id", tasks[i].EventID,
			)
		}
	}
}

// EvictWebhook implements WebhookEvictor. It releases the
// engine's per-webhook archiving state: the database target's
// cached archive writer is dropped from the registry and its
// file handle closed. The archive file itself is left on disk
// — it is long-term storage the operator owns.
func (e *Engine) EvictWebhook(webhookID string) {
	if e.dbTarget == nil {
		return
	}

	e.dbTarget.evict(webhookID)
}

// ScheduleRetry schedules a task to be re-enqueued onto the
// retry channel after delay. It implements the Scheduler
// interface the targets use to own their durable retries.
func (e *Engine) ScheduleRetry(
	task Task, delay time.Duration,
) {
	e.log.Debug(
		"scheduling delivery retry",
		"webhook_id", task.WebhookID,
		"delivery_id", task.DeliveryID,
		"delay", delay,
		"next_attempt", task.AttemptNum,
	)

	// The reference is taken here rather than when the timer fires,
	// so the delivery stays owned across the whole backoff window.
	// Its caller is a target inside Deliver, so the engine already
	// owns it; this second reference is what keeps that ownership
	// alive after the worker returns and the row sits at retrying
	// with nothing running. Without it the sweep finds the row
	// orphaned and sends it again.
	e.inflight.retain(task.DeliveryID)

	time.AfterFunc(delay, func() {
		select {
		case e.retryCh <- task:
		default:
			e.inflight.release(task.DeliveryID)
			e.log.Warn(
				"retry channel full, delivery "+
					"will be recovered by periodic sweep",
				"delivery_id", task.DeliveryID,
				"webhook_id", task.WebhookID,
			)
		}
	})
}

// registerHooks wires the engine's start and stop into the fx
// lifecycle. The start hook's context is deliberately ignored
// (see start for why the worker pool must not inherit it); the
// stop hook's context is honoured (see stop).
func (e *Engine) registerHooks(lc fx.Lifecycle) {
	lc.Append(fx.Hook{
		//nolint:contextcheck // Not inheriting the hook context
		// is the point: see start.
		OnStart: func(_ context.Context) error {
			e.start()

			return nil
		},
		OnStop: func(ctx context.Context) error {
			return e.stop(ctx)
		},
	})
}

// start launches the worker pool, restart recovery, and the
// periodic retry sweep.
//
// Their context is derived from context.Background(), NOT from
// the fx OnStart hook context. The hook context carries fx's
// start timeout (15s by default) and is cancelled once the start
// phase completes, so goroutines derived from it stop a few
// seconds into the process: every worker would return and the
// engine would silently stop delivering webhooks entirely. A
// long-lived goroutine must outlive the startup phase, so its
// lifetime is bounded by OnStop instead: stop cancels this
// context and waits on the WaitGroup.
func (e *Engine) start() {
	ctx, cancel := context.WithCancel(context.Background())
	e.cancel = cancel

	for range e.workers {
		e.wg.Add(1)

		go e.worker(ctx)
	}

	e.wg.Add(1)

	go e.recoverPending(ctx)

	e.wg.Add(1)

	go e.retrySweep(ctx)

	e.wg.Add(1)

	go e.queueDepthSampler(ctx)

	e.log.Info(
		"delivery engine started",
		"workers", e.workers,
	)
}

// stop cancels the worker pool's context and waits for the pool
// to drain, bounded by the stop hook's context: a wedged worker
// must not hang the process past fx's stop timeout.
func (e *Engine) stop(ctx context.Context) error {
	e.log.Info("delivery engine stopping")

	if e.cancel != nil {
		e.cancel()
	}

	err := lifecycle.WaitForShutdown(
		ctx, e.log, "delivery engine", &e.wg,
	)
	if err != nil {
		return err
	}

	e.log.Info("delivery engine stopped")

	return nil
}

func (e *Engine) worker(ctx context.Context) {
	defer e.wg.Done()

	for {
		select {
		case <-ctx.Done():
			return
		case task := <-e.deliveryCh:
			e.runTask(ctx, &task, e.processNewTask)
		case task := <-e.retryCh:
			e.runTask(ctx, &task, e.processRetryTask)
		}
	}
}

// runTask runs one task and then drops the reference the queueing
// side took on its delivery.
//
// The release is deferred rather than written after the call because
// every early return inside the processing paths must drop it too: a
// delivery whose database could not be opened is one the engine has
// stopped working on, and leaving it owned would hide it from the
// sweep forever.
//
// Ownership does not necessarily end here. A target that scheduled a
// retry took its own reference before this one is dropped, so the
// delivery stays owned through the backoff window.
func (e *Engine) runTask(
	ctx context.Context,
	task *Task,
	run func(context.Context, *Task),
) {
	defer e.inflight.release(task.DeliveryID)

	run(ctx, task)
}

func (e *Engine) recoverPending(ctx context.Context) {
	defer e.wg.Done()

	e.recoverInFlight(ctx)
}

func (e *Engine) processNewTask(
	ctx context.Context, task *Task,
) {
	webhookDB, err := e.dbManager.GetDB(task.WebhookID)
	if err != nil {
		e.log.Error(
			"failed to get webhook database",
			"webhook_id", task.WebhookID,
			"error", err,
		)

		return
	}

	event := buildEventFromTask(task)

	event, err = e.hydrateEvent(
		webhookDB, event, task,
	)
	if err != nil {
		e.log.Error(
			"failed to fetch event body from database",
			"event_id", task.EventID,
			"error", err,
		)

		return
	}

	target := buildTargetFromTask(task)

	d := &database.Delivery{
		EventID:  task.EventID,
		TargetID: task.TargetID,
		Status:   database.DeliveryStatusPending,
		Event:    event,
		Target:   target,
	}
	d.ID = task.DeliveryID

	e.processDelivery(ctx, webhookDB, d, task)
}

func (e *Engine) processRetryTask(
	ctx context.Context, task *Task,
) {
	webhookDB, err := e.dbManager.GetDB(task.WebhookID)
	if err != nil {
		e.log.Error(
			"failed to get webhook database for retry",
			"webhook_id", task.WebhookID,
			"delivery_id", task.DeliveryID,
			"error", err,
		)

		return
	}

	d, err := e.loadRetryDelivery(
		webhookDB, task.DeliveryID,
	)
	if err != nil {
		e.log.Error(
			"failed to load delivery for retry",
			"delivery_id", task.DeliveryID,
			"error", err,
		)

		return
	}

	if d.Status != database.DeliveryStatusRetrying {
		e.log.Debug(
			"skipping retry for delivery "+
				"no longer in retrying status",
			"delivery_id", d.ID,
			"status", d.Status,
		)

		return
	}

	if e.abandonRetryForMissingTarget(webhookDB, d, task) {
		return
	}

	event := buildEventFromTask(task)

	event, err = e.hydrateEvent(
		webhookDB, event, task,
	)
	if err != nil {
		e.log.Error(
			"failed to fetch event body for retry",
			"event_id", task.EventID,
			"error", err,
		)

		return
	}

	target := buildTargetFromTask(task)
	d.EventID = task.EventID
	d.TargetID = task.TargetID
	d.Event = event
	d.Target = target

	e.processDelivery(ctx, webhookDB, d, task)
}

// abandonRetryForMissingTarget stops a retry chain whose target has
// been deleted, and reports whether it did.
//
// A scheduled retry lives in memory as a time.AfterFunc holding the
// target's configuration as it was when the chain began, and nothing
// else on this path reads the target row. Without this check a
// deletion stops nothing: the timer keeps firing and keeps sending to
// the destination the operator removed, for the whole remaining
// backoff chain. Terminalising in the recovery and sweep paths alone
// is not enough, because those only see the delivery once nothing
// holds it in memory — which is to say after a restart.
//
// The worker already owns this delivery, so the terminal write happens
// here directly, exactly as a target's own Deliver fails one. Claiming
// it again through the recovery gate would only fail against the
// reference the worker itself is holding.
//
// A lookup that fails for any other reason is not a deletion — it is
// the main database being unreadable — and the delivery goes ahead as
// it did before. A guard that terminally failed deliveries on a
// transient fault would be worse than the bug it fixes.
func (e *Engine) abandonRetryForMissingTarget(
	webhookDB *gorm.DB,
	d *database.Delivery,
	task *Task,
) bool {
	_, err := e.loadTarget(task.TargetID)
	if err == nil {
		return false
	}

	if !errors.Is(err, gorm.ErrRecordNotFound) {
		e.log.Warn(
			"could not confirm the target of a retrying "+
				"delivery still exists; attempting anyway",
			"delivery_id", task.DeliveryID,
			"target_id", task.TargetID,
			"error", err,
		)

		return false
	}

	targetType, reason := e.missingTargetReason(task.TargetID)

	e.log.Warn(
		"abandoning scheduled retry: target is gone",
		"webhook_id", task.WebhookID,
		"delivery_id", task.DeliveryID,
		"target_id", task.TargetID,
		"target_type", targetType,
	)

	e.failDelivery(webhookDB, d, targetType, reason)

	return true
}

func (e *Engine) recoverInFlight(ctx context.Context) {
	var webhookIDs []string

	err := e.database.DB().
		Model(&database.Webhook{}).
		Pluck("id", &webhookIDs).Error
	if err != nil {
		e.log.Error(
			"failed to query webhook IDs for recovery",
			"error", err,
		)

		return
	}

	for _, webhookID := range webhookIDs {
		select {
		case <-ctx.Done():
			return
		default:
		}

		if !e.dbManager.DBExists(webhookID) {
			continue
		}

		e.recoverWebhookDeliveries(ctx, webhookID)
	}
}

func (e *Engine) recoverWebhookDeliveries(
	ctx context.Context, webhookID string,
) {
	webhookDB, err := e.dbManager.GetDB(webhookID)
	if err != nil {
		e.log.Error(
			"failed to get webhook database for recovery",
			"webhook_id", webhookID,
			"error", err,
		)

		return
	}

	e.recoverPendingDeliveries(
		ctx, webhookDB, webhookID,
	)

	e.recoverRetryingDeliveries(
		webhookDB, webhookID,
	)
}

func (e *Engine) recoverRetryingDeliveries(
	webhookDB *gorm.DB, webhookID string,
) {
	var retrying []database.Delivery

	err := webhookDB.
		Where(
			"status = ?",
			database.DeliveryStatusRetrying,
		).
		Find(&retrying).Error
	if err != nil {
		e.log.Error(
			"failed to query retrying deliveries "+
				"for recovery",
			"webhook_id", webhookID,
			"error", err,
		)

		return
	}

	settled := e.reconcileDelivered(
		webhookDB, webhookID, retrying,
		e.loadTargetMap(retrying),
	)

	for i := range retrying {
		if _, ok := settled[retrying[i].ID]; ok {
			continue
		}

		e.recoverSingleRetry(
			webhookDB, webhookID, &retrying[i],
		)
	}
}

// recoverSingleRetry hands an orphaned retrying delivery back
// to its target to recompute the remaining backoff, then
// reschedules it. Targets that do not own durable retries
// (fire-and-forget) never produce retrying deliveries, so a
// delivery found in that state has had its target's type
// changed underneath it and is terminally failed.
func (e *Engine) recoverSingleRetry(
	webhookDB *gorm.DB,
	webhookID string,
	d *database.Delivery,
) {
	target, err := e.loadTarget(d.TargetID)
	if err != nil {
		// A target that is merely gone is an operator action with a
		// terminal answer. Any other failure is the main database
		// refusing to read, which is transient and must leave the
		// delivery alone: failing every retrying delivery of every
		// webhook on one bad read would be a far larger fault than
		// the strand it is meant to clear.
		if errors.Is(err, gorm.ErrRecordNotFound) {
			e.failMissingTargetRetry(
				webhookDB, webhookID, d,
			)

			return
		}

		e.log.Error(
			"failed to load target for retrying "+
				"delivery recovery",
			"delivery_id", d.ID,
			"target_id", d.TargetID,
			"error", err,
		)

		return
	}

	rs, ok := e.targets[target.Type].(rescheduler)
	if !ok {
		e.failUnretryableRetry(
			webhookDB, webhookID, d, &target,
		)

		return
	}

	attemptNum := e.countAttempts(webhookDB, d.ID)
	remaining := rs.remainingBackoff(
		webhookDB, d.ID, attemptNum,
	)

	event, err := e.loadEvent(webhookDB, d.EventID)
	if err != nil {
		e.log.Error(
			"failed to load event for retrying "+
				"delivery recovery",
			"delivery_id", d.ID,
			"event_id", d.EventID,
			"error", err,
		)

		return
	}

	task := buildRecoveryTask(
		d, webhookID, &event, &target, attemptNum+1,
	)

	if !e.rescheduleRecovered(webhookDB, task, remaining) {
		return
	}

	e.log.Info(
		"recovering retrying delivery",
		"webhook_id", webhookID,
		"delivery_id", d.ID,
		"attempt", attemptNum,
		"remaining_backoff", remaining,
	)
}

func (e *Engine) recoverPendingDeliveries(
	ctx context.Context,
	webhookDB *gorm.DB,
	webhookID string,
) {
	var deliveries []database.Delivery

	// No Preload: event bodies are read one at a time in
	// sendRecoveredDeliveries, and only for the deliveries actually
	// being sent.
	result := webhookDB.
		Where(
			"status = ?",
			database.DeliveryStatusPending,
		).
		Find(&deliveries)

	if result.Error != nil {
		e.log.Error(
			"failed to query pending deliveries",
			"webhook_id", webhookID,
			"error", result.Error,
		)

		return
	}

	if len(deliveries) == 0 {
		return
	}

	e.log.Info(
		"recovering pending deliveries",
		"webhook_id", webhookID,
		"count", len(deliveries),
	)

	e.recoverPendingBatch(
		ctx, webhookDB, webhookID, deliveries,
	)
}

// recoverPendingBatch settles every delivery in the batch that was
// already delivered, and re-dispatches only the rest. Both the
// restart-time recovery and the periodic sweep go through it, so a
// pending delivery is treated the same however it was found.
func (e *Engine) recoverPendingBatch(
	ctx context.Context,
	webhookDB *gorm.DB,
	webhookID string,
	deliveries []database.Delivery,
) {
	targetMap := e.loadTargetMap(deliveries)

	settled := e.reconcileDelivered(
		webhookDB, webhookID, deliveries, targetMap,
	)

	e.sendRecoveredDeliveries(
		ctx, webhookDB, deliveries, webhookID,
		targetMap, settled,
	)
}

// reconcileDelivered finds the deliveries in a recovered batch that
// already have a successful DeliveryResult, marks them delivered, and
// returns their ids so the caller does not send them a second time.
//
// This is the state the engine previously had no way to represent. A
// delivery is left in a non-terminal state by a failed bookkeeping
// write, and that covers two different histories: nothing was ever
// sent, or the send reached the receiver and only the status write
// failed. Re-sending was the sole option, so every stranded row
// produced a duplicate at the receiver and an event log that recorded
// one attempt for two POSTs. A successful result row distinguishes
// them: it is written before the status, so its presence means the
// wire I/O happened and was recorded, and all that is missing is the
// status.
//
// Every recovery path runs this, not only the pending one. A delivery
// abandoned at retrying can hold a successful result just as a pending
// one can — a second attempt that reached the receiver and whose status
// write then failed sits at retrying with success recorded — and
// re-sending it is the same duplicate.
//
// Deliveries whose result row itself never landed are not in the
// returned set and are re-sent, recorded as the further attempt they
// are. That is honest at-least-once delivery rather than a silent
// duplicate.
func (e *Engine) reconcileDelivered(
	webhookDB *gorm.DB,
	webhookID string,
	deliveries []database.Delivery,
	targetMap map[string]database.Target,
) map[string]struct{} {
	settled := make(map[string]struct{})

	if len(deliveries) == 0 {
		return settled
	}

	ids := make([]string, 0, len(deliveries))
	for i := range deliveries {
		ids = append(ids, deliveries[i].ID)
	}

	var deliveredIDs []string

	err := webhookDB.
		Model(&database.DeliveryResult{}).
		Where(
			"delivery_id IN ? AND success = ?", ids, true,
		).
		Distinct().
		Pluck("delivery_id", &deliveredIDs).Error
	if err != nil {
		// Every delivery stays out of the settled set, so the batch
		// is re-sent exactly as it was before this check existed.
		// That is the safe direction: a duplicate delivery beats
		// declaring a delivery successful on a query that failed.
		e.log.Error(
			"failed to query successful delivery results; "+
				"pending deliveries will be re-sent",
			"webhook_id", webhookID,
			"error", err,
		)

		return settled
	}

	for _, id := range deliveredIDs {
		settled[id] = struct{}{}
	}

	if len(settled) == 0 {
		return settled
	}

	e.log.Info(
		"settling recovered deliveries that already succeeded",
		"webhook_id", webhookID,
		"count", len(settled),
	)

	for i := range deliveries {
		if _, ok := settled[deliveries[i].ID]; !ok {
			continue
		}

		// A delivery the engine is working on right now settles
		// itself; writing over it from here would race that worker.
		if !e.inflight.retainIdle(deliveries[i].ID) {
			delete(settled, deliveries[i].ID)

			continue
		}

		e.settleStatus(
			webhookDB,
			&deliveries[i],
			targetMap[deliveries[i].TargetID].Type,
			database.DeliveryStatusDelivered,
		)

		e.inflight.release(deliveries[i].ID)
	}

	return settled
}

func (e *Engine) retrySweep(ctx context.Context) {
	defer e.wg.Done()

	ticker := time.NewTicker(retrySweepInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			e.sweepOrphanedRetries(ctx)
		}
	}
}

func (e *Engine) sweepOrphanedRetries(
	ctx context.Context,
) {
	var webhookIDs []string

	err := e.database.DB().
		Model(&database.Webhook{}).
		Pluck("id", &webhookIDs).Error
	if err != nil {
		e.log.Error(
			"retry sweep: failed to query webhook IDs",
			"error", err,
		)

		return
	}

	for _, webhookID := range webhookIDs {
		select {
		case <-ctx.Done():
			return
		default:
		}

		if !e.dbManager.DBExists(webhookID) {
			continue
		}

		e.sweepWebhookRetries(ctx, webhookID)
	}
}

func (e *Engine) sweepWebhookRetries(
	ctx context.Context, webhookID string,
) {
	webhookDB, err := e.dbManager.GetDB(webhookID)
	if err != nil {
		e.log.Error(
			"retry sweep: failed to get webhook database",
			"webhook_id", webhookID,
			"error", err,
		)

		return
	}

	var retrying []database.Delivery

	err = webhookDB.
		Where(
			"status = ?",
			database.DeliveryStatusRetrying,
		).
		Find(&retrying).Error
	if err != nil {
		e.log.Error(
			"retry sweep: "+
				"failed to query retrying deliveries",
			"webhook_id", webhookID,
			"error", err,
		)

		return
	}

	settled := e.reconcileDelivered(
		webhookDB, webhookID, retrying,
		e.loadTargetMap(retrying),
	)

	for i := range retrying {
		select {
		case <-ctx.Done():
			return
		default:
		}

		if _, ok := settled[retrying[i].ID]; ok {
			continue
		}

		e.sweepSingleRetry(
			webhookDB, webhookID, &retrying[i],
		)
	}

	e.sweepWebhookPending(ctx, webhookDB, webhookID)
}

// sweepWebhookPending recovers deliveries stranded at pending.
//
// A delivery is created pending and leaves that state only when its
// outcome is written, so a pending row the engine does not own is one
// whose bookkeeping write failed — the state that used to sit there
// until a restart, and then produce a duplicate at the receiver. The
// sweep gives it the same reconcile-then-dispatch treatment restart
// recovery gets, so it costs a minute rather than an operator
// noticing.
//
// What keeps the sweep off live work is ownership, checked per
// delivery in takeForRedispatch, not the age bound in this query.
// A delivery waiting in deliveryCh is pending and arbitrarily old —
// the channel holds 10000 tasks and 10 workers drain it — so
// reasoning from the row's age alone re-sends it. See inflight.go.
func (e *Engine) sweepWebhookPending(
	ctx context.Context,
	webhookDB *gorm.DB,
	webhookID string,
) {
	var pending []database.Delivery

	err := webhookDB.
		Where(
			"status = ? AND updated_at < ?",
			database.DeliveryStatusPending,
			time.Now().Add(-pendingSweepMinAge),
		).
		Limit(pendingSweepBatch).
		Find(&pending).Error
	if err != nil {
		e.log.Error(
			"retry sweep: "+
				"failed to query pending deliveries",
			"webhook_id", webhookID,
			"error", err,
		)

		return
	}

	if len(pending) == 0 {
		return
	}

	e.log.Info(
		"retry sweep: recovering stranded pending deliveries",
		"webhook_id", webhookID,
		"count", len(pending),
	)

	e.recoverPendingBatch(ctx, webhookDB, webhookID, pending)
}

// sweepSingleRetry re-enqueues an orphaned retrying delivery
// whose backoff window has elapsed, delegating the backoff
// decision to the delivery's target. A delivery whose target
// no longer owns durable retries is terminally failed.
func (e *Engine) sweepSingleRetry(
	webhookDB *gorm.DB,
	webhookID string,
	d *database.Delivery,
) {
	target, err := e.loadTarget(d.TargetID)
	if err != nil {
		// Deleted is terminal, unreadable is not; see
		// recoverSingleRetry.
		if errors.Is(err, gorm.ErrRecordNotFound) {
			e.failMissingTargetRetry(
				webhookDB, webhookID, d,
			)

			return
		}

		e.log.Error(
			"retry sweep: failed to load target",
			"delivery_id", d.ID,
			"target_id", d.TargetID,
			"error", err,
		)

		return
	}

	rs, ok := e.targets[target.Type].(rescheduler)
	if !ok {
		e.failUnretryableRetry(
			webhookDB, webhookID, d, &target,
		)

		return
	}

	attemptNum := e.countAttempts(webhookDB, d.ID)

	if !rs.backoffElapsed(
		webhookDB, d.ID, attemptNum,
	) {
		return
	}

	event, err := e.loadEvent(webhookDB, d.EventID)
	if err != nil {
		e.log.Error(
			"retry sweep: failed to load event",
			"delivery_id", d.ID,
			"event_id", d.EventID,
			"error", err,
		)

		return
	}

	task := buildRecoveryTask(
		d, webhookID, &event, &target, attemptNum+1,
	)

	if !e.redispatch(
		e.retryCh, webhookDB, task,
		database.DeliveryStatusRetrying,
	) {
		return
	}

	e.log.Info(
		"retry sweep: "+
			"recovered orphaned retrying delivery",
		"delivery_id", d.ID,
		"webhook_id", webhookID,
		"attempt", attemptNum+1,
	)
}

// failUnretryableRetry terminally fails an orphaned retrying
// delivery whose target type no longer supports retries. Both
// restart recovery and the periodic sweep call it, so the
// terminal transition exists once.
//
// This is only reachable when a target's type has been changed
// out from under an in-flight retrying delivery (or the type is
// unknown to the registry): fire-and-forget targets never set
// status retrying themselves. Re-dispatching under the new type
// would be a delivery the operator never asked for, and leaving
// the row retrying strands it forever, so the delivery is
// failed with a recorded reason. The event stays stored, but
// nothing redelivers it today. Logged at warn, not error: this
// is operator-caused state, not a system fault.
func (e *Engine) failUnretryableRetry(
	webhookDB *gorm.DB,
	webhookID string,
	d *database.Delivery,
	target *database.Target,
) {
	// Terminal, and reached from the recovery paths, so it takes
	// ownership like every other write they make: a delivery the
	// engine is still attempting must not be failed underneath the
	// worker running it.
	if !e.inflight.retainIdle(d.ID) {
		return
	}

	defer e.inflight.release(d.ID)

	e.log.Warn(
		"failing orphaned retrying delivery: target "+
			"type no longer supports retries",
		"webhook_id", webhookID,
		"delivery_id", d.ID,
		"target_id", target.ID,
		"target_name", target.Name,
		"target_type", target.Type,
	)

	reason := fmt.Sprintf(
		"target type %q does not support retries; "+
			"delivery was left retrying by a previous "+
			"target type and has been failed terminally",
		target.Type,
	)

	e.failDelivery(webhookDB, d, target.Type, reason)
}

// failMissingTargetRetry terminally fails an orphaned retrying
// delivery whose target row is gone. Both restart recovery and the
// periodic sweep call it, so the transition exists once.
//
// Until it existed both paths logged the failed lookup and returned,
// which left the delivery retrying for the life of the database and
// the sweep repeating the same error every minute forever. Failing it
// with a recorded reason is the treatment the other orphaned-retry
// cases already get, so all of them read alike in the event log.
//
// Logged at warn rather than error: a deleted target is an operator
// action, not a system fault.
func (e *Engine) failMissingTargetRetry(
	webhookDB *gorm.DB,
	webhookID string,
	d *database.Delivery,
) {
	// Terminal, and reached from the recovery paths, so it takes
	// ownership like every other write they make.
	if !e.inflight.retainIdle(d.ID) {
		return
	}

	defer e.inflight.release(d.ID)

	targetType, reason := e.missingTargetReason(d.TargetID)

	e.log.Warn(
		"failing orphaned retrying delivery: "+
			"its target no longer exists",
		"webhook_id", webhookID,
		"delivery_id", d.ID,
		"target_id", d.TargetID,
		"target_type", targetType,
	)

	e.failDelivery(webhookDB, d, targetType, reason)
}

// missingTargetReason describes a target id that no longer resolves,
// and returns the type of the deleted row where there still is one.
//
// The lookup is Unscoped because deletes are soft: the row survives
// with deleted_at set, invisible to loadTarget's default scope.
// Reading it is what separates "you deleted this target" from "this id
// never named a row" — different things to whoever reads the event
// log, and only the first is something an operator did. The widened
// scope is deliberately confined to this terminal path: the engine's
// normal target loading must go on refusing a deleted target, or
// deleting one would stop nothing.
//
// The type comes back so the caller can label the delivery's status
// transition with it. Where the row is gone entirely there is no type
// to give, and updateDeliveryStatus leaves the counter alone rather
// than opening a series named by the empty string.
func (e *Engine) missingTargetReason(
	targetID string,
) (database.TargetType, string) {
	var target database.Target

	err := e.database.DB().Unscoped().
		First(&target, "id = ?", targetID).Error
	if err != nil {
		return "", fmt.Sprintf(
			"target %s no longer exists; the delivery "+
				"cannot be retried and has been failed "+
				"terminally",
			targetID,
		)
	}

	return target.Type, fmt.Sprintf(
		"target %q (type %s) was deleted; the delivery "+
			"cannot be retried and has been failed terminally",
		target.Name, target.Type,
	)
}

// failDelivery records why a delivery is over and then marks it
// failed. The caller must already own the delivery: every call site is
// either a worker holding the reference runTask took, or a recovery
// path that took one through retainIdle.
//
// The result row is written first and a failure to write it stops the
// transition, which is what keeps a delivery from ending failed with
// an empty event log — the state that leaves an operator with nothing
// but a server log line to work out what happened. A delivery whose
// reason could not be recorded stays in the non-terminal state it
// already holds, where the sweep will find it again; see
// bookkeepingFailed.
//
// The target type is a parameter rather than read off d because the
// orphaned-retry callers deliberately hold a delivery loaded without
// its Target relation: populating d.Target would make GORM's
// SaveBeforeAssociations upsert the whole target row — plaintext
// config, which for a slack target is the credential — into the
// per-webhook event database. See
// https://git.eeqj.de/sneak/webhooker/issues/206.
func (e *Engine) failDelivery(
	webhookDB *gorm.DB,
	d *database.Delivery,
	targetType database.TargetType,
	reason string,
) {
	err := e.recordResult(
		webhookDB,
		d,
		e.countAttempts(webhookDB, d.ID)+1,
		false,
		0,
		"",
		reason,
		0,
	)
	if err != nil {
		e.bookkeepingFailed(d, err)

		return
	}

	e.settleStatus(
		webhookDB, d, targetType,
		database.DeliveryStatusFailed,
	)
}

// processDelivery dispatches a delivery to the target that
// owns its type. Unknown target types fail the delivery.
func (e *Engine) processDelivery(
	ctx context.Context,
	webhookDB *gorm.DB,
	d *database.Delivery,
	task *Task,
) {
	target, ok := e.targets[d.Target.Type]
	if !ok {
		e.log.Error(
			"unknown target type",
			"target_id", d.TargetID,
			"type", d.Target.Type,
		)

		// The reason is recorded, not just logged. This branch used
		// to fail the delivery with no DeliveryResult at all, which
		// showed in the event log as "failed, no attempts recorded
		// yet" and left one server log line as the only account of
		// why anywhere.
		e.failDelivery(
			webhookDB, d, d.Target.Type,
			fmt.Sprintf(
				"unknown target type %q: this build has no "+
					"delivery implementation for it, so no "+
					"attempt was made",
				d.Target.Type,
			),
		)

		return
	}

	target.Deliver(ctx, webhookDB, d, task, e)
}

// observeAttempt counts one delivery attempt that was actually
// dispatched to a target, and records how long it took.
//
// It is called from the dispatch paths rather than from around
// Target.Deliver, because Deliver is also entered for deliveries
// that never reach the wire: a delivery an open circuit breaker
// refuses sends nothing, records no DeliveryResult, and is
// rescheduled. Counting those would climb the attempts counter with
// no traffic behind it and fill the duration histogram with
// microsecond samples, which would make the delivery-duration
// quantiles improve during exactly the outage they exist to reveal.
func (e *Engine) observeAttempt(
	t database.TargetType, dur time.Duration,
) {
	e.mtr.DeliveryAttempted(t)
	e.mtr.ObserveDeliveryDuration(t, dur)
}

// recordResult persists a DeliveryResult row describing a
// single attempt. It is a cross-target helper the targets
// call.
//
// It returns its error rather than swallowing it. A DeliveryResult
// row is the only record that an attempt happened at all, so a
// caller that ignored a failed write would go on to mark the
// delivery delivered — leaving the event log claiming one attempt
// for a receiver that got two. Every caller must instead stop
// advancing the delivery's status and let it stay in the
// non-terminal state it already holds; see bookkeepingFailed.
func (e *Engine) recordResult(
	webhookDB *gorm.DB,
	d *database.Delivery,
	attemptNum int,
	success bool,
	statusCode int,
	respBody, errMsg string,
	durationMs int64,
) error {
	result := &database.DeliveryResult{
		DeliveryID:   d.ID,
		AttemptNum:   attemptNum,
		Success:      success,
		StatusCode:   statusCode,
		ResponseBody: truncate(respBody, maxBodyLog),
		Error:        errMsg,
		Duration:     durationMs,
	}

	err := webhookDB.Create(result).Error
	if err != nil {
		return fmt.Errorf(
			"recording delivery result for %s: %w", d.ID, err,
		)
	}

	return nil
}

// bookkeepingFailed reports that a delivery's own record of what
// happened could not be written, and deliberately writes nothing in
// response.
//
// Leaving the row alone is the whole point. A delivery is created
// pending and only ever leaves that state through
// updateDeliveryStatus, so a delivery whose bookkeeping write failed
// is still pending or retrying — the two non-terminal states, per
// DeliveryStatus.Terminal — and both are swept and recovered. Writing
// anything here would need the very database that just refused a
// write, and would be one more thing to fail; not writing cannot.
//
// The cost is honest at-least-once behaviour: a send that reached the
// receiver but whose result row did not land is attempted again, and
// recorded as the further attempt it is. What no longer happens is the
// silent duplicate — a second POST the event log denies ever
// occurred. Where the result row *did* land and only the status write
// failed, reconcileDelivered settles the row without re-sending.
func (e *Engine) bookkeepingFailed(
	d *database.Delivery, err error,
) {
	e.log.Error(
		"delivery bookkeeping write failed; leaving delivery "+
			"in a recoverable state",
		"delivery_id", d.ID,
		"event_id", d.EventID,
		"target_id", d.TargetID,
		"status", d.Status,
		"error", err,
	)
}

// updateDeliveryStatus persists a new status for a delivery.
// It is a cross-target helper the targets call, and therefore the
// single point where a delivery's outcome — delivered, terminally
// failed, or put back into retry — is counted.
//
// The target type is a parameter rather than read off d.Target
// because one caller — failUnretryableRetry — deliberately holds a
// delivery loaded without its target relation, and must keep it that
// way: a populated d.Target makes GORM upsert the target row, config
// and all, into the per-webhook database.
//
// The counter moves only after the row is written, so a transition
// the database rejected is not claimed as an outcome that happened.
// For the same reason the error is returned rather than logged and
// dropped: a delivery whose status write failed has not reached that
// status, and its caller must not act as though it had.
func (e *Engine) updateDeliveryStatus(
	webhookDB *gorm.DB,
	d *database.Delivery,
	targetType database.TargetType,
	status database.DeliveryStatus,
) error {
	err := webhookDB.Model(d).
		Update("status", status).Error
	if err != nil {
		return fmt.Errorf(
			"updating delivery %s to status %s: %w",
			d.ID, status, err,
		)
	}

	// An empty type means the target row is gone — a delivery being
	// settled long after its target was deleted. The row still has to
	// be settled, but the counter is left alone rather than given a
	// series labelled with the empty string.
	if targetType != "" {
		e.mtr.DeliveryStatusChanged(targetType, status)
	}

	return nil
}

// settleStatus moves a delivery to its outcome status and reports a
// failed write through bookkeepingFailed, which leaves the row
// recoverable. It exists so the target call sites read as one
// statement rather than four lines of identical error handling.
func (e *Engine) settleStatus(
	webhookDB *gorm.DB,
	d *database.Delivery,
	targetType database.TargetType,
	status database.DeliveryStatus,
) {
	err := e.updateDeliveryStatus(
		webhookDB, d, targetType, status,
	)
	if err != nil {
		e.bookkeepingFailed(d, err)
	}
}

func truncate(s string, maxLen int) string {
	if len(s) <= maxLen {
		return s
	}

	return s[:maxLen]
}

// --- Helper functions ---

// buildEventFromTask reconstructs the event a Task describes, as far
// as the Task itself goes. The fields it cannot fill — the body when
// it was too large to inline, and the receipt time, which no Task
// carries — come from the stored row in hydrateEvent, which every
// caller of this function runs next.
func buildEventFromTask(task *Task) database.Event {
	event := database.Event{
		EntrypointID: task.EntrypointID,
		Method:       task.Method,
		Headers:      task.Headers,
		ContentType:  task.ContentType,
	}

	event.ID = task.EventID
	event.WebhookID = task.WebhookID

	return event
}

func buildTargetFromTask(task *Task) database.Target {
	target := database.Target{
		Name:       task.TargetName,
		Type:       task.TargetType,
		Config:     task.TargetConfig,
		MaxRetries: task.MaxRetries,
	}

	target.ID = task.TargetID

	return target
}

// hydrateEvent fills in the event fields a Task does not carry, by
// reading the stored event row.
//
// CreatedAt is the event's receipt time and lives only in that row.
// The Slack target renders it into every message it sends, so an
// unhydrated event puts the zero time in front of a human on every
// notification the product delivers. See
// https://git.eeqj.de/sneak/webhooker/issues/257.
//
// The body comes from the same row when the Task did not inline it,
// which is the case for a body at or above MaxInlineBodySize.
//
// A read failure is fatal to the delivery only when the body depended
// on it. When the Task inlined the body, the delivery has everything
// it needs to be sent and goes ahead with the timestamp unset: the row
// can be gone under a retention reap while a queued delivery still
// holds its body, and dropping a deliverable event to protect one
// metadata field would be a worse failure than the one it prevents.
func (e *Engine) hydrateEvent(
	webhookDB *gorm.DB,
	event database.Event,
	task *Task,
) (database.Event, error) {
	columns := []string{"created_at"}

	if task.Body == nil {
		columns = append(columns, "body")
	}

	var dbEvent database.Event

	err := webhookDB.Select(columns).
		First(&dbEvent, "id = ?", task.EventID).Error
	if err != nil {
		if task.Body == nil {
			return event, fmt.Errorf(
				"fetching event body: %w", err,
			)
		}

		e.log.Warn(
			"could not read the stored event; delivering "+
				"the inlined body without its receipt time",
			"event_id", task.EventID,
			"delivery_id", task.DeliveryID,
			"error", err,
		)

		event.Body = *task.Body

		return event, nil
	}

	event.CreatedAt = dbEvent.CreatedAt

	if task.Body != nil {
		event.Body = *task.Body
	} else {
		event.Body = dbEvent.Body
	}

	return event, nil
}

func (e *Engine) loadRetryDelivery(
	webhookDB *gorm.DB, deliveryID string,
) (*database.Delivery, error) {
	var d database.Delivery

	err := webhookDB.Select("id", "status").
		First(&d, "id = ?", deliveryID).Error
	if err != nil {
		return nil, fmt.Errorf(
			"loading delivery: %w", err,
		)
	}

	return &d, nil
}

func (e *Engine) countAttempts(
	webhookDB *gorm.DB, deliveryID string,
) int {
	var resultCount int64

	webhookDB.Model(&database.DeliveryResult{}).
		Where("delivery_id = ?", deliveryID).
		Count(&resultCount)

	return int(resultCount)
}

// takeForRedispatch decides whether a recovered delivery may be sent
// again, and takes it if so. It is the single gate every re-dispatch
// path goes through, and it asks two separate questions in order.
//
// First, does the engine already own this delivery? Ownership is
// exact and mutually exclusive, so a delivery queued, being attempted,
// or waiting out a retry backoff is refused here, and two dispatchers
// racing for the same delivery cannot both win. See inflight.go.
//
// Second, is the row still in the status that made it eligible? The
// batch was read some time ago and a worker may have settled a row
// since. The check is a conditional update rather than a read so the
// answer cannot go stale between asking and acting.
//
// Stamping updated_at is the same statement, and it is a cadence
// control rather than a claim: the pending sweep selects on that
// column, so a delivery handed out now is not selected again on the
// next tick a minute later but after pendingSweepMinAge. A delivery
// the database refuses to settle is therefore retried on that
// interval instead of every tick.
//
// A failed write is a refusal. It means the database is not accepting
// writes, which is the condition that stranded the delivery in the
// first place; an attempt that cannot be recorded is exactly the
// unlogged duplicate this is all here to prevent.
//
// The caller must release ownership if it then fails to queue the
// task.
func (e *Engine) takeForRedispatch(
	webhookDB *gorm.DB,
	deliveryID string,
	eligible database.DeliveryStatus,
) bool {
	if !e.inflight.retainIdle(deliveryID) {
		return false
	}

	res := webhookDB.
		Model(&database.Delivery{}).
		Where(
			"id = ? AND status = ?", deliveryID, eligible,
		).
		UpdateColumn("updated_at", time.Now())

	if res.Error != nil {
		e.log.Error(
			"failed to mark delivery for re-dispatch; "+
				"leaving it for a later sweep",
			"delivery_id", deliveryID,
			"error", res.Error,
		)
		e.inflight.release(deliveryID)

		return false
	}

	if res.RowsAffected != 1 {
		// Settled underneath us between the query and here.
		e.inflight.release(deliveryID)

		return false
	}

	return true
}

// queueRecovered puts an owned delivery's task on a worker channel,
// dropping the ownership the gate took if it does not fit.
func (e *Engine) queueRecovered(
	ch chan<- Task, task Task,
) bool {
	select {
	case ch <- task:
		return true
	default:
		e.inflight.release(task.DeliveryID)
		e.log.Warn(
			"worker channel full during recovery; "+
				"delivery will be recovered by a later sweep",
			"delivery_id", task.DeliveryID,
			"webhook_id", task.WebhookID,
		)

		return false
	}
}

// redispatch hands a recovered delivery to a worker channel through
// takeForRedispatch, and reports whether the task was queued.
func (e *Engine) redispatch(
	ch chan<- Task,
	webhookDB *gorm.DB,
	task Task,
	eligible database.DeliveryStatus,
) bool {
	if !e.takeForRedispatch(
		webhookDB, task.DeliveryID, eligible,
	) {
		return false
	}

	return e.queueRecovered(ch, task)
}

// rescheduleRecovered hands an orphaned retrying delivery back to the
// retry timer, through the same gate. It reports whether the delivery
// was rescheduled.
//
// The reference taken by the gate is dropped as soon as ScheduleRetry
// has taken its own, which it does before returning: what keeps the
// delivery owned through the backoff window is ScheduleRetry's
// reference, not this one.
func (e *Engine) rescheduleRecovered(
	webhookDB *gorm.DB, task Task, delay time.Duration,
) bool {
	if !e.takeForRedispatch(
		webhookDB, task.DeliveryID,
		database.DeliveryStatusRetrying,
	) {
		return false
	}

	defer e.inflight.release(task.DeliveryID)

	e.ScheduleRetry(task, delay)

	return true
}

// countAttemptsBatch counts the recorded attempts of every delivery
// in a batch with one grouped query, keyed by delivery id. Deliveries
// with no attempts are simply absent from the result, which reads back
// as the zero this caller wants.
//
// One query rather than one per delivery: this runs on the recovery
// path, which is a burst of writes against a database that has just
// been under enough contention to strand these rows in the first
// place. See https://git.eeqj.de/sneak/webhooker/issues/256.
func (e *Engine) countAttemptsBatch(
	webhookDB *gorm.DB, deliveries []database.Delivery,
) map[string]int {
	counts := make(map[string]int, len(deliveries))

	if len(deliveries) == 0 {
		return counts
	}

	ids := make([]string, 0, len(deliveries))
	for i := range deliveries {
		ids = append(ids, deliveries[i].ID)
	}

	// One delivery id per recorded attempt, tallied here rather than
	// grouped in SQL: internal/gormlog forbids (*gorm.DB).Scan, which
	// a GROUP BY into a struct would need, and an attempt row per
	// delivery is bounded by the target's MaxRetries.
	var attemptIDs []string

	err := webhookDB.
		Model(&database.DeliveryResult{}).
		Where("delivery_id IN ?", ids).
		Pluck("delivery_id", &attemptIDs).Error
	if err != nil {
		e.log.Error(
			"failed to count delivery attempts for recovery",
			"error", err,
		)

		return counts
	}

	for _, id := range attemptIDs {
		counts[id]++
	}

	return counts
}

func (e *Engine) loadEvent(
	webhookDB *gorm.DB, eventID string,
) (database.Event, error) {
	var event database.Event

	err := webhookDB.
		First(&event, "id = ?", eventID).Error
	if err != nil {
		return event, fmt.Errorf(
			"loading event: %w", err,
		)
	}

	return event, nil
}

func (e *Engine) loadTarget(
	targetID string,
) (database.Target, error) {
	var target database.Target

	err := e.database.DB().
		First(&target, "id = ?", targetID).Error
	if err != nil {
		return target, fmt.Errorf(
			"loading target: %w", err,
		)
	}

	return target, nil
}

func buildRecoveryTask(
	d *database.Delivery,
	webhookID string,
	event *database.Event,
	target *database.Target,
	attemptNum int,
) Task {
	var bodyPtr *string

	if len(event.Body) < MaxInlineBodySize {
		bodyStr := event.Body
		bodyPtr = &bodyStr
	}

	return Task{
		DeliveryID:   d.ID,
		EventID:      d.EventID,
		WebhookID:    webhookID,
		EntrypointID: event.EntrypointID,
		TargetID:     target.ID,
		TargetName:   target.Name,
		TargetType:   target.Type,
		TargetConfig: target.Config,
		MaxRetries:   target.MaxRetries,
		Method:       event.Method,
		Headers:      event.Headers,
		ContentType:  event.ContentType,
		Body:         bodyPtr,
		AttemptNum:   attemptNum,
	}
}

func (e *Engine) loadTargetMap(
	deliveries []database.Delivery,
) map[string]database.Target {
	if len(deliveries) == 0 {
		return nil
	}

	seen := make(map[string]bool)

	targetIDs := make([]string, 0, len(deliveries))

	for _, d := range deliveries {
		if !seen[d.TargetID] {
			targetIDs = append(targetIDs, d.TargetID)
			seen[d.TargetID] = true
		}
	}

	var targets []database.Target

	err := e.database.DB().
		Where("id IN ?", targetIDs).
		Find(&targets).Error
	if err != nil {
		e.log.Error(
			"failed to load targets from main DB",
			"error", err,
		)

		return nil
	}

	targetMap := make(
		map[string]database.Target, len(targets),
	)

	for _, t := range targets {
		targetMap[t.ID] = t
	}

	return targetMap
}

// sendRecoveredDeliveries re-dispatches pending deliveries, skipping
// the ids in settled — those already reached their receiver and have
// been marked delivered by reconcileDelivered.
//
// The skip and takeForRedispatch's status check answer different
// questions and neither replaces the other. This one is "did this
// delivery already succeed", which is what settles the row to
// delivered instead of sending it, and which is the only thing that
// keeps the retrying paths from terminally failing a delivery that
// reconcile just settled. The status check is "is the row still what
// the batch query said it was", which catches a worker settling it to
// anything at all in between.
func (e *Engine) sendRecoveredDeliveries(
	ctx context.Context,
	webhookDB *gorm.DB,
	deliveries []database.Delivery,
	webhookID string,
	targetMap map[string]database.Target,
	settled map[string]struct{},
) {
	// The attempt number continues each delivery's own history
	// rather than restarting at 1. A recovered delivery may already
	// have recorded attempts, and numbering the next one 1 again
	// both collides in the event log and hands the retry path a
	// backoff computed from the wrong attempt.
	attempts := e.countAttemptsBatch(webhookDB, deliveries)

	for i := range deliveries {
		select {
		case <-ctx.Done():
			return
		default:
		}

		if _, ok := settled[deliveries[i].ID]; ok {
			continue
		}

		target, ok := targetMap[deliveries[i].TargetID]
		if !ok {
			e.log.Error(
				"target not found for delivery",
				"delivery_id", deliveries[i].ID,
				"target_id", deliveries[i].TargetID,
			)

			continue
		}

		if !e.takeForRedispatch(
			webhookDB, deliveries[i].ID,
			database.DeliveryStatusPending,
		) {
			continue
		}

		// The body is read here, one delivery at a time and only for
		// deliveries that are actually being sent, rather than
		// preloaded across the whole batch. A batch is up to
		// pendingSweepBatch rows at up to the 1 MB ingest cap, and
		// most of a sweep's batch is refused by the gate above — so
		// preloading would hold hundreds of megabytes per webhook per
		// tick to build tasks it then discards.
		event, err := e.loadEvent(
			webhookDB, deliveries[i].EventID,
		)
		if err != nil {
			e.log.Error(
				"failed to load event for recovered delivery",
				"delivery_id", deliveries[i].ID,
				"event_id", deliveries[i].EventID,
				"error", err,
			)
			e.inflight.release(deliveries[i].ID)

			continue
		}

		task := buildRecoveryTask(
			&deliveries[i], webhookID, &event, &target,
			attempts[deliveries[i].ID]+1,
		)

		e.queueRecovered(e.deliveryCh, task)
	}
}