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/archive"
	"sneak.berlin/go/webhooker/internal/database"
	"sneak.berlin/go/webhooker/internal/globals"
	"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 recovery routine runs
	// once it has run at start. Each of those runs is a tick of
	// the sweep.
	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 tick of the sweep re-dispatches for one webhook. The sweep
	// ticks every retrySweepInterval, so a larger backlog drains
	// across successive ticks 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
	MaxAttempts  int

	Method      string
	RawQuery    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)
}

// CircuitBreakers is how the handlers read a target's circuit
// breaker, so the webhook page and the event log can say that
// deliveries to the target are paused and until when. It keeps the
// handlers free of the engine's internals and is trivially faked in
// tests.
type CircuitBreakers interface {
	StateAndCooldown(targetID string) (CircuitState, time.Duration)
}

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

	DB        *database.Database
	DBManager *database.WebhookDBManager
	Globals   *globals.Globals
	Logger    *logger.Logger
	SSRFGuard *Guard
	Metrics   *metrics.Set
	Archives  *archive.Writers
}

// 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, the recovery routine, 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

	// version is the running build's version, the one the web UI
	// footer shows. userAgent puts it on every outbound request.
	version string

	// mtr is the delivery metric set. Production wires the one
	// registered on the registry /metrics serves; a test can
	// substitute a set registered on a registry it holds, so it can
	// gather what its own deliveries recorded.
	mtr *metrics.Set

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

	// httpTarget and slackTarget are retained so StateAndCooldown
	// can read their circuit breakers, and so tests can reach the
	// HTTP target's shared client.
	httpTarget  *httpTarget
	slackTarget *slackTarget

	// archives is the database targets' archive writers. The
	// database target writes each event through them, and stop
	// closes them.
	archives *archive.Writers

	// inflight is the set of deliveries this engine currently owns.
	// The recovery routine re-dispatches 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,
		version:    params.Globals.Version,
		mtr:        params.Metrics,
		archives:   params.Archives,
	}

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

// StateAndCooldown implements CircuitBreakers. It returns the state of
// the target's circuit breaker and, while the breaker is open, what is
// left of its cooldown; the cooldown is zero once that has passed and
// in any other state. A target with no breaker reads as closed with no
// cooldown, and reading never creates one.
func (e *Engine) StateAndCooldown(
	targetID string,
) (CircuitState, time.Duration) {
	for _, core := range []*httpCore{
		e.httpTarget.httpCore, e.slackTarget.httpCore,
	} {
		val, ok := core.circuitBreakers.Load(targetID)
		if ok {
			cb, _ := val.(*CircuitBreaker)

			return cb.StateAndCooldown()
		}
	}

	return CircuitClosed, 0
}

// 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,
			)
		}
	})
}

// userAgent is the User-Agent header of every http and slack
// delivery request: the program name and the running build's
// version.
func (e *Engine) userAgent() string {
	return "webhooker/" + e.version
}

// 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 and the recovery routine, which
// runs now and then on every tick of the sweep (see runRecovery).
//
// 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.runRecovery(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.
//
// Once the pool has drained it closes the archive writers, so a
// clean stop leaves no archive -wal behind. Nothing else holds a
// writer for long by then: the archive sweeper stops before the
// engine, and deleting or renaming a webhook or target only closes
// or moves one. If the pool did
// not drain in time, the writers are left open, as a kill would
// leave them. Closing them would wait for any write in progress,
// and a worker still running would then open new writers that
// nothing closes, so it gains nothing over a kill.
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.archives.EvictAll()

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

	// The recovery routine's run at start can send and release this
	// delivery before the receiver's Notify queues it. Ownership cannot
	// refuse a delivery nobody holds, so the row decides whether it
	// still needs sending.
	row, err := e.loadDelivery(webhookDB, task.DeliveryID)
	if err != nil {
		e.log.Error(
			"failed to load delivery",
			"delivery_id", task.DeliveryID,
			"error", err,
		)

		return
	}

	if row.Status != database.DeliveryStatusPending {
		e.log.Info(
			"delivery already handled, not sent again",
			"delivery_id", task.DeliveryID,
			"event_id", task.EventID,
			"status", row.Status,
		)

		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.loadDelivery(
		webhookDB, task.DeliveryID,
	)
	if err != nil {
		e.log.Error(
			"failed to load delivery for retry",
			"delivery_id", task.DeliveryID,
			"error", err,
		)

		return
	}

	// Set before anything below can fail the delivery: the failure is
	// added to this target's totals.
	d.EventID = task.EventID
	d.TargetID = task.TargetID

	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.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 routine alone is not
// enough, because it only sees 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
}

// runRecovery runs the recovery routine, recoverDeliveries, once when
// the engine starts and then on every tick of the sweep, every
// retrySweepInterval, until the engine stops.
func (e *Engine) runRecovery(ctx context.Context) {
	defer e.wg.Done()

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

	e.recoverDeliveries(ctx, true)

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

// recoverDeliveries is the engine's one recovery routine: atStart is
// true for its run when the engine starts and false on a tick of the
// sweep.
//
// For every webhook it finds the pending and retrying deliveries the
// engine is not working on, such as those a stop or a crash cut off, a
// full channel dropped, or a failed bookkeeping write left behind. One
// that already has a successful attempt is marked delivered. A retrying
// one is scheduled after what is left of its backoff, or now when
// nothing is left. A pending one goes to a worker now.
//
// Ownership keeps it off a delivery the engine is working on, has
// queued, or holds a retry timer for, the same at start as on a tick:
// every delivery it acts on is taken through the inflight set first.
// See inflight.go.
//
// A run at start differs from a tick in which webhooks it opens (see
// recoverWebhookDeliveries) and which pending deliveries it takes (see
// recoverPendingDeliveries).
func (e *Engine) recoverDeliveries(ctx context.Context, atStart bool) {
	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:
		}

		e.recoverWebhookDeliveries(ctx, webhookID, atStart)
	}
}

// recoverWebhookDeliveries runs the recovery routine over one webhook.
//
// At start the webhook's database is opened even when its file is
// missing, so that a lost database is reported at start, not when the
// webhook next receives an event, which for a quiet webhook may be
// never. A tick leaves a missing file alone rather than create one: a
// webhook being created has its row saved a moment before its file,
// and a tick in between would report that file as lost.
func (e *Engine) recoverWebhookDeliveries(
	ctx context.Context, webhookID string, atStart bool,
) {
	if !atStart && !e.dbManager.DBExists(webhookID) {
		return
	}

	// The web interface is serving, so the webhook may have been
	// deleted since the list was read. Opening its database then
	// would create the file again after the delete removed it.
	stillExists := func() (bool, error) {
		var count int64

		err := e.database.DB().
			Model(&database.Webhook{}).
			Where("id = ?", webhookID).
			Count(&count).Error
		if err != nil {
			return false, fmt.Errorf("confirming webhook exists: %w", err)
		}

		return count > 0, nil
	}

	webhookDB, err := e.dbManager.GetDBIf(webhookID, stillExists)
	if err != nil {
		e.log.Error(
			"failed to get webhook database for recovery",
			"webhook_id", webhookID,
			"error", err,
		)

		return
	}

	if webhookDB == nil {
		return
	}

	e.recoverPendingDeliveries(
		ctx, webhookDB, webhookID, atStart,
	)

	e.recoverRetryingDeliveries(
		ctx, webhookDB, webhookID,
	)
}

// recoverRetryingDeliveries recovers every retrying delivery of one
// webhook, at start and on a tick alike.
func (e *Engine) recoverRetryingDeliveries(
	ctx context.Context, 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 {
		select {
		case <-ctx.Done():
			return
		default:
		}

		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 work out what is left of its backoff, then
// schedules its next attempt after that, or now when nothing is
// left. 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.failMissingTarget(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,
	)

	// Taken before the event is read, so that a delivery the engine
	// already holds, most often one waiting out its backoff on a
	// timer, costs a tick no read of its body.
	if !e.takeForRedispatch(
		webhookDB, d.ID, database.DeliveryStatusRetrying,
	) {
		return
	}

	// ScheduleRetry takes a reference of its own before it returns,
	// and that one keeps the delivery owned through the backoff.
	defer e.inflight.release(d.ID)

	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
	}

	e.ScheduleRetry(buildRecoveryTask(
		d, webhookID, &event, &target, attemptNum+1,
	), remaining)

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

// recoverPendingDeliveries recovers the pending deliveries of one
// webhook: it marks delivered those that already have a successful
// attempt and sends the rest.
//
// 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
// it stopped working on before then: for example one whose bookkeeping
// write failed, whose attempt a stop cut off, that was still queued
// when the engine last stopped, whose task was dropped because the
// delivery channel was full (by Notify or by queueRecovered), or whose
// worker could not open the webhook's database.
//
// At start every pending delivery is taken, however recent, so one a
// stop cut off moments ago is sent again now. A tick takes only those
// untouched for pendingSweepMinAge, at most pendingSweepBatch of them.
// The age bound is not what keeps a tick off live work: a delivery
// waiting in deliveryCh is pending and arbitrarily old. Ownership is,
// checked per delivery in takeForRedispatch. See inflight.go.
func (e *Engine) recoverPendingDeliveries(
	ctx context.Context,
	webhookDB *gorm.DB,
	webhookID string,
	atStart bool,
) {
	query := webhookDB.Where(
		"status = ?", database.DeliveryStatusPending,
	)

	if !atStart {
		query = query.
			Where(
				"updated_at < ?",
				time.Now().Add(-pendingSweepMinAge),
			).
			Limit(pendingSweepBatch)
	}

	var deliveries []database.Delivery

	// No Preload: event bodies are read one at a time in
	// sendRecoveredDeliveries, and only for the deliveries actually
	// being sent.
	err := query.Find(&deliveries).Error
	if err != nil {
		e.log.Error(
			"failed to query pending deliveries for recovery",
			"webhook_id", webhookID,
			"error", err,
		)

		return
	}

	if len(deliveries) == 0 {
		return
	}

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

	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.
//
// The recovery routine runs this for retrying deliveries as well as
// pending ones. 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
}

// failUnretryableRetry terminally fails an orphaned retrying
// delivery whose target type no longer supports retries.
//
// 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)
}

// failMissingTarget terminally fails a recovered delivery, pending or
// retrying, whose target row is gone. The recovery routine calls it for
// both statuses.
//
// Until it existed those paths logged the failed lookup and moved on,
// which left the delivery where it was 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) failMissingTarget(
	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)

	// The batch was read before ownership was taken, and a worker may
	// have settled the delivery and let it go in between. Only a row
	// still in the status the batch read is failed.
	row, err := e.loadDelivery(webhookDB, d.ID)
	if err != nil {
		e.log.Error(
			"failed to load delivery",
			"delivery_id", d.ID,
			"error", err,
		)

		return
	}

	if row.Status != d.Status {
		e.log.Debug(
			"delivery already handled, not failed",
			"delivery_id", d.ID,
			"status", row.Status,
		)

		return
	}

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

	e.log.Warn(
		"failing recovered delivery: its target no longer exists",
		"webhook_id", webhookID,
		"delivery_id", d.ID,
		"status", d.Status,
		"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 "+
				"has been failed terminally",
			targetID,
		)
	}

	return target.Type, fmt.Sprintf(
		"target %q (type %s) was deleted; the delivery "+
			"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.Transaction(func(tx *gorm.DB) error {
		return writeDeliveryStatus(tx, d, status)
	})
	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
}

// writeDeliveryStatus writes a delivery's new status. A delivery that
// becomes delivered or failed also gets the time it finished, and is
// added to its target's delivered or failed total. That write changes
// only a delivery not yet delivered or failed, and the total moves
// only when it changed a row: retention may have deleted the delivery
// while the engine was working on it, and a recovery path may settle
// a delivery that a worker has already settled.
func writeDeliveryStatus(
	tx *gorm.DB,
	d *database.Delivery,
	status database.DeliveryStatus,
) error {
	if !status.Terminal() {
		return tx.Model(d).Update("status", status).Error
	}

	res := tx.Model(d).
		Where("status NOT IN ?", []database.DeliveryStatus{
			database.DeliveryStatusDelivered,
			database.DeliveryStatusFailed,
		}).
		Updates(map[string]any{
			"status": status,
			// GORM's clock, the one that stamps created_at. A bare
			// time.Now() carries a monotonic clock reading, which the
			// driver would store as an " m=+…" suffix.
			"finished_at": tx.NowFunc(),
		})
	if res.Error != nil || res.RowsAffected == 0 {
		return res.Error
	}

	add := database.TargetTotals{TargetID: d.TargetID, Delivered: 1}
	if status == database.DeliveryStatusFailed {
		add = database.TargetTotals{TargetID: d.TargetID, Failed: 1}
	}

	return database.AddTargetTotals(tx, add)
}

// 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,
		RawQuery:     task.RawQuery,
		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,
		MaxAttempts: task.MaxAttempts,
	}

	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) loadDelivery(
	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: a tick of the sweep selects pending
// deliveries on that column, so one 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
	}

	// GORM's clock, as writeDeliveryStatus explains.
	res := webhookDB.
		Model(&database.Delivery{}).
		Where(
			"id = ? AND status = ?", deliveryID, eligible,
		).
		UpdateColumn("updated_at", webhookDB.NowFunc())

	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 pending delivery's task on the delivery
// channel, dropping the ownership the gate took if it does not fit.
func (e *Engine) queueRecovered(task Task) {
	select {
	case e.deliveryCh <- task:
	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,
		)
	}
}

// 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 MaxAttempts.
	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,
		MaxAttempts:  target.MaxAttempts,
		Method:       event.Method,
		RawQuery:     event.RawQuery,
		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 {
			// A missing entry does not mean the target is gone: the
			// map is also empty when its query failed. Only a lookup
			// that finds no row ends the delivery; any other error
			// leaves it pending for the next sweep. See
			// recoverSingleRetry.
			var err error

			target, err = e.loadTarget(deliveries[i].TargetID)
			if errors.Is(err, gorm.ErrRecordNotFound) {
				e.failMissingTarget(webhookDB, webhookID, &deliveries[i])

				continue
			}

			if err != nil {
				e.log.Error(
					"failed to load target for recovered delivery",
					"delivery_id", deliveries[i].ID,
					"target_id", deliveries[i].TargetID,
					"error", err,
				)

				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(task)
	}
}