webhooker/internal/delivery/target_http.go

package delivery

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"sort"
	"sync"
	"time"

	"gorm.io/gorm"
	"sneak.berlin/go/webhooker/internal/database"
)

// Sentinel errors returned by the config parsers.
var (
	errEmptyTargetConfig = errors.New(
		"empty target config",
	)
	errMissingTargetURL = errors.New(
		"target URL is required",
	)
)

// HTTPTargetConfig holds configuration for http target
// types.
type HTTPTargetConfig struct {
	URL     string            `json:"url"`
	Headers map[string]string `json:"headers,omitempty"`
	Timeout int               `json:"timeout,omitempty"`

	// ForwardQuery passes each event's query string on to the target,
	// appended to URL. Off, the target URL is sent exactly as
	// configured.
	ForwardQuery bool `json:"forwardQuery,omitempty"`
}

// httpCore holds the retry, backoff, and circuit-breaker
// machinery shared by the HTTP and Slack targets. Each of
// those targets owns its own httpCore instance (and thus its
// own circuit breakers); the per-attempt request differs
// between them and is supplied as a closure.
type httpCore struct {
	eng *Engine

	// circuitBreakers stores a *CircuitBreaker per target ID.
	circuitBreakers sync.Map
}

// deliver runs one delivery attempt through the retry core.
// A maxAttempts of 1 is a single attempt, recorded without
// consulting a circuit breaker. A higher maxAttempts gates the
// attempt on the circuit breaker and schedules a backed-off retry
// on failure, until maxAttempts attempts have been made.
//
// ctx is the context attempt sends its request with. Stopping the
// engine cancels it; see interruptedByStop.
func (c *httpCore) deliver(
	ctx context.Context,
	webhookDB *gorm.DB,
	d *database.Delivery,
	task *Task,
	sched Scheduler,
	maxAttempts int,
	attempt func() attemptResult,
) {
	if maxAttempts == 1 {
		c.fireAndForget(ctx, webhookDB, d, attempt())

		return
	}

	c.withRetry(
		ctx, webhookDB, d, task, sched, maxAttempts, attempt,
	)
}

// interruptedByStop reports whether an attempt failed because the
// engine is stopping, which cancels the context its request is made
// with. Such an attempt is not recorded: no DeliveryResult, no circuit
// breaker outcome, no status change, no count in the attempt metrics.
// The delivery stays pending or retrying, and startup recovery sends
// it again on the next start.
//
// A target's own timeout leaves that context alone, so its attempt is
// still recorded as the failure it is.
func (c *httpCore) interruptedByStop(
	ctx context.Context,
	d *database.Delivery,
	res attemptResult,
) bool {
	if res.success || ctx.Err() == nil {
		return false
	}

	c.eng.log.Warn(
		"delivery attempt interrupted by shutdown; not recorded, "+
			"it is sent again on the next start",
		"delivery_id", d.ID,
		"event_id", d.EventID,
		"target_id", d.TargetID,
	)

	return true
}

func (c *httpCore) fireAndForget(
	ctx context.Context,
	webhookDB *gorm.DB,
	d *database.Delivery,
	res attemptResult,
) {
	if c.interruptedByStop(ctx, d, res) {
		return
	}

	c.eng.observeAttempt(d.Target.Type, res.elapsed())

	err := c.eng.recordResult(
		webhookDB, d, 1, res.success,
		res.statusCode, res.respBody, res.errMsg,
		res.duration,
	)
	if err != nil {
		c.eng.bookkeepingFailed(d, err)

		return
	}

	if res.success {
		c.eng.settleStatus(
			webhookDB, d, d.Target.Type,
			database.DeliveryStatusDelivered,
		)

		return
	}

	c.eng.settleStatus(
		webhookDB, d, d.Target.Type,
		database.DeliveryStatusFailed,
	)
}

func (c *httpCore) withRetry(
	ctx context.Context,
	webhookDB *gorm.DB,
	d *database.Delivery,
	task *Task,
	sched Scheduler,
	maxAttempts int,
	attempt func() attemptResult,
) {
	cb := c.getCircuitBreaker(task.TargetID)
	if c.circuitBreakerBlock(webhookDB, d, task, sched, cb) {
		return
	}

	// Allow may have moved the breaker to half-open, and the
	// attempt below may open or close it, so the gauge is
	// republished on every exit from here.
	defer c.publishCircuitState(d.Target.Type)

	attemptNum := task.AttemptNum

	res := attempt()

	if c.interruptedByStop(ctx, d, res) {
		return
	}

	c.eng.observeAttempt(d.Target.Type, res.elapsed())

	err := c.eng.recordResult(
		webhookDB, d, attemptNum, res.success,
		res.statusCode, res.respBody, res.errMsg,
		res.duration,
	)
	if err != nil {
		// The breaker still learns the outcome: it describes the
		// target's health, which is unaffected by this database's.
		c.recordCircuitOutcome(cb, res.success)

		c.eng.bookkeepingFailed(d, err)

		return
	}

	if res.success {
		cb.RecordSuccess()

		c.eng.settleStatus(
			webhookDB, d, d.Target.Type,
			database.DeliveryStatusDelivered,
		)

		return
	}

	cb.RecordFailure()

	c.handleRetry(
		webhookDB, d, task, sched, maxAttempts, attemptNum,
	)
}

// recordCircuitOutcome feeds one attempt's outcome to the target's
// circuit breaker.
func (c *httpCore) recordCircuitOutcome(
	cb *CircuitBreaker, success bool,
) {
	if success {
		cb.RecordSuccess()

		return
	}

	cb.RecordFailure()
}

func (c *httpCore) circuitBreakerBlock(
	webhookDB *gorm.DB,
	d *database.Delivery,
	task *Task,
	sched Scheduler,
	cb *CircuitBreaker,
) bool {
	if cb.Allow() {
		return false
	}

	defer c.publishCircuitState(d.Target.Type)

	remaining := cb.CooldownRemaining()

	c.eng.log.Info(
		"circuit breaker open, skipping delivery",
		"target_id", task.TargetID,
		"target_name", task.TargetName,
		"delivery_id", d.ID,
		"cooldown_remaining", remaining,
	)

	// A delivery already at retrying is left as it is, so a task
	// the breaker keeps turning away writes nothing each time.
	if d.Status != database.DeliveryStatusRetrying {
		c.eng.settleStatus(
			webhookDB, d, d.Target.Type,
			database.DeliveryStatusRetrying,
		)
	}

	retryTask := *task
	sched.ScheduleRetry(retryTask, remaining)

	return true
}

func (c *httpCore) handleRetry(
	webhookDB *gorm.DB,
	d *database.Delivery,
	task *Task,
	sched Scheduler,
	maxAttempts int,
	attemptNum int,
) {
	if attemptNum >= maxAttempts {
		c.eng.settleStatus(
			webhookDB, d, d.Target.Type,
			database.DeliveryStatusFailed,
		)

		return
	}

	c.eng.settleStatus(
		webhookDB, d, d.Target.Type,
		database.DeliveryStatusRetrying,
	)

	backoff := Backoff(attemptNum)

	retryTask := *task
	retryTask.AttemptNum = attemptNum + 1
	sched.ScheduleRetry(retryTask, backoff)
}

func (c *httpCore) getCircuitBreaker(
	targetID string,
) *CircuitBreaker {
	if val, ok := c.circuitBreakers.Load(targetID); ok {
		cb, _ := val.(*CircuitBreaker)

		return cb
	}

	fresh := NewCircuitBreaker()

	actual, _ := c.circuitBreakers.LoadOrStore(
		targetID, fresh,
	)

	cb, _ := actual.(*CircuitBreaker)

	return cb
}

// publishCircuitState recounts this core's open breakers and
// publishes the gauge. Each core holds the breakers of exactly one
// target type, so the recount is over that type's targets alone.
// Counting rather than adjusting a delta keeps the gauge honest
// however a breaker changed state.
func (c *httpCore) publishCircuitState(
	targetType database.TargetType,
) {
	open := 0

	c.circuitBreakers.Range(func(_, val any) bool {
		cb, ok := val.(*CircuitBreaker)
		if ok && cb.State() == CircuitOpen {
			open++
		}

		return true
	})

	c.eng.mtr.SetCircuitBreakersOpen(targetType, open)
}

// remainingBackoff returns how long remains of the backoff
// window for the last attempt of a recovered retrying
// delivery. It implements rescheduler.
func (c *httpCore) remainingBackoff(
	webhookDB *gorm.DB,
	deliveryID string,
	attemptNum int,
) time.Duration {
	var lastResult database.DeliveryResult

	err := webhookDB.
		Where("delivery_id = ?", deliveryID).
		Order("created_at DESC").
		First(&lastResult).Error
	if err != nil {
		return 0
	}

	backoff := Backoff(attemptNum)
	elapsed := time.Since(lastResult.CreatedAt)
	remaining := backoff - elapsed

	return max(remaining, 0)
}

// backoffElapsed reports whether the backoff window for the
// last attempt of a retrying delivery has passed. It
// implements rescheduler.
func (c *httpCore) backoffElapsed(
	webhookDB *gorm.DB,
	deliveryID string,
	attemptNum int,
) bool {
	var lastResult database.DeliveryResult

	err := webhookDB.
		Where("delivery_id = ?", deliveryID).
		Order("created_at DESC").
		First(&lastResult).Error
	if err != nil {
		return true
	}

	backoff := Backoff(attemptNum)

	return time.Since(lastResult.CreatedAt) >= backoff
}

// Backoff is how long an http or slack target with retries waits after
// a delivery's failed attempt attemptNum before trying it again.
func Backoff(attemptNum int) time.Duration {
	shift := max(attemptNum-1, 0)
	shift = min(shift, maxBackoffShift)

	return time.Duration(1<<uint(shift)) * time.Second
}

// httpTarget delivers events to http targets. It forwards the
// event body and (filtered) request headers to the configured
// URL and owns retry, backoff, and circuit breaking through
// the shared httpCore.
type httpTarget struct {
	*httpCore

	client *http.Client
}

// Deliver implements Target.
func (t *httpTarget) Deliver(
	ctx context.Context,
	webhookDB *gorm.DB,
	d *database.Delivery,
	task *Task,
	sched Scheduler,
) {
	cfg, err := parseHTTPConfig(d.Target.Config)
	if err != nil {
		t.eng.log.Error(
			"invalid HTTP target config",
			"target_id", d.TargetID,
			"error", err,
		)

		recErr := t.eng.recordResult(
			webhookDB, d, task.AttemptNum,
			false, 0, "", err.Error(), 0,
		)
		if recErr != nil {
			t.eng.bookkeepingFailed(d, recErr)

			return
		}

		t.eng.settleStatus(
			webhookDB, d, d.Target.Type,
			database.DeliveryStatusFailed,
		)

		return
	}

	attempt := func() attemptResult {
		return t.attempt(ctx, cfg, &d.Event)
	}

	t.deliver(
		ctx, webhookDB, d, task, sched,
		d.Target.MaxAttempts, attempt,
	)
}

// attempt performs a single HTTP delivery attempt and derives
// the success flag and error message the same way the engine
// did: a non-2xx response is a failure but carries no error
// string; only a transport-level error does.
func (t *httpTarget) attempt(
	ctx context.Context,
	cfg *HTTPTargetConfig,
	event *database.Event,
) attemptResult {
	statusCode, respBody, duration, reqErr :=
		t.doHTTPRequest(ctx, cfg, event)

	success := reqErr == nil &&
		statusCode >= httpSuccessMin &&
		statusCode < httpSuccessMax

	errMsg := ""
	if reqErr != nil {
		errMsg = reqErr.Error()
	}

	return attemptResult{
		statusCode: statusCode,
		respBody:   respBody,
		duration:   duration,
		success:    success,
		errMsg:     errMsg,
	}
}

func (t *httpTarget) doHTTPRequest(
	ctx context.Context,
	cfg *HTTPTargetConfig,
	event *database.Event,
) (int, string, int64, error) {
	start := time.Now()

	req, reqErr := http.NewRequestWithContext(
		ctx,
		http.MethodPost,
		cfg.URL,
		bytes.NewReader([]byte(event.Body)),
	)
	if reqErr != nil {
		return 0, "", 0, fmt.Errorf(
			"creating request: %w",
			maskURLError(reqErr),
		)
	}

	if cfg.ForwardQuery {
		appendQuery(req.URL, event.RawQuery)
	}

	originScoped := applyRequestHeaders(
		req, event, cfg, t.eng.userAgent(),
	)

	client := t.clientForRequest(cfg, originScoped)

	resp, doErr := executeHTTPRequest(client, req)

	dur := time.Since(start).Milliseconds()
	if doErr != nil {
		return 0, "", dur, fmt.Errorf(
			"sending request: %w", doErr,
		)
	}

	defer func() { _ = resp.Body.Close() }()

	body, readErr := io.ReadAll(
		io.LimitReader(resp.Body, maxBodyLog),
	)
	if readErr != nil {
		return resp.StatusCode, "", dur,
			fmt.Errorf(
				"reading response body: %w", readErr,
			)
	}

	return resp.StatusCode, string(body), dur, nil
}

// appendQuery adds an event's query string to a delivery's URL, joined
// with "&" to any query string the target URL already has.
func appendQuery(u *url.URL, rawQuery string) {
	switch {
	case rawQuery == "":
		return
	case u.RawQuery == "":
		u.RawQuery = rawQuery
	default:
		u.RawQuery += "&" + rawQuery
	}
}

// clientForRequest returns the client for one delivery attempt.
// originScoped is the header set applyRequestHeaders built for that
// attempt; a request with neither a per-target timeout nor an
// origin-scoped header gets the shared client, because there is
// then nothing for the redirect policy to strip and net/http's
// default policy already withholds Authorization and Cookie across
// hosts.
func (t *httpTarget) clientForRequest(
	cfg *HTTPTargetConfig,
	originScoped []string,
) *http.Client {
	if cfg.Timeout <= 0 && len(originScoped) == 0 {
		return t.client
	}

	// Reuse the shared client's SSRF-safe transport so neither a
	// per-target timeout nor the redirect policy drops the
	// request-time private-IP guard — which, being a dial hook,
	// also covers every redirect hop.
	client := &http.Client{
		Timeout:   t.client.Timeout,
		Transport: t.client.Transport,
	}

	if cfg.Timeout > 0 {
		client.Timeout = time.Duration(
			cfg.Timeout,
		) * time.Second
	}

	if len(originScoped) > 0 {
		client.CheckRedirect = offOriginHeaderPolicy(originScoped)
	}

	return client
}

func parseHTTPConfig(
	configJSON string,
) (*HTTPTargetConfig, error) {
	if configJSON == "" {
		return nil, errEmptyTargetConfig
	}

	var cfg HTTPTargetConfig

	err := json.Unmarshal(
		[]byte(configJSON), &cfg,
	)
	if err != nil {
		return nil, fmt.Errorf(
			"parsing config JSON: %w", err,
		)
	}

	if cfg.URL == "" {
		return nil, errMissingTargetURL
	}

	return &cfg, nil
}

// isForwardableHeader returns true if the header should
// be forwarded to targets.
func isForwardableHeader(name string) bool {
	switch http.CanonicalHeaderKey(name) {
	case "Host", "Connection", "Keep-Alive",
		"Transfer-Encoding", "Te", "Trailer",
		"Upgrade", "Proxy-Authorization",
		"Proxy-Connection", "Content-Length":
		return false
	case "Content-Type":
		// applyRequestHeaders sets Content-Type itself. The receiver
		// already stored this inbound value as the event's
		// ContentType, so forwarding it too would send it twice.
		return false
	default:
		return true
	}
}

// applyRequestHeaders builds one outbound delivery's header set and
// returns the canonical names of every header in it that is scoped
// to the configured origin: the inbound event headers this delivery
// forwarded, plus the operator's configured headers. The redirect
// policy strips exactly that set on a hop that leaves the origin,
// so the forward set is decided here and only here — a header added
// to it is covered off-origin without a second edit elsewhere.
//
// Content-Type goes out once: a Content-Type configured on the target
// wins, otherwise the event's ContentType, otherwise none. The inbound
// Content-Type in the event's headers is never forwarded.
//
// userAgent is set last, over any configured or inbound User-Agent.
func applyRequestHeaders(
	req *http.Request,
	event *database.Event,
	cfg *HTTPTargetConfig,
	userAgent string,
) []string {
	if event.ContentType != "" {
		req.Header.Set(
			"Content-Type", event.ContentType,
		)
	}

	originScoped := forwardEventHeaders(req, event)

	for k, v := range cfg.Headers {
		req.Header.Set(k, v)
		originScoped[http.CanonicalHeaderKey(k)] = struct{}{}
	}

	req.Header.Set("User-Agent", userAgent)

	// A Content-Type configured on the target describes the body
	// being sent rather than the sender. A 307/308 preserves the
	// body across hosts, so stripping it would send that body
	// untyped.
	delete(originScoped, "Content-Type")

	// User-Agent is overwritten just above, so an inbound one never
	// reaches the wire and the value that does identifies this
	// delivery path rather than the sender. Reporting it would strip
	// it off-origin and leave net/http's own default in its place.
	delete(originScoped, "User-Agent")

	names := make([]string, 0, len(originScoped))
	for name := range originScoped {
		names = append(names, name)
	}

	sort.Strings(names)

	return names
}

// forwardEventHeaders copies the inbound event's forwardable
// headers onto the outbound request and returns the canonical names
// it forwarded. Headers the event never carried are absent from the
// result, so the redirect policy strips what was actually sent.
func forwardEventHeaders(
	req *http.Request,
	event *database.Event,
) map[string]struct{} {
	forwarded := make(map[string]struct{})

	if event.Headers == "" {
		return forwarded
	}

	var inbound map[string][]string

	if json.Unmarshal([]byte(event.Headers), &inbound) != nil {
		return forwarded
	}

	for k, vals := range inbound {
		if !isForwardableHeader(k) || len(vals) == 0 {
			continue
		}

		for _, v := range vals {
			req.Header.Add(k, v)
		}

		forwarded[http.CanonicalHeaderKey(k)] = struct{}{}
	}

	return forwarded
}

// executeHTTPRequest sends an HTTP request using the provided
// client. URLs are validated by the config parsers and the
// SSRF-safe transport before reaching here.
//
// Transport failures are masked here, at the single point
// where every target's request errors are born, because the
// caller stores them in DeliveryResult.Error: an unmasked
// *url.Error would write the target URL — the credential for
// a Slack incoming webhook — into the per-webhook database.
func executeHTTPRequest(
	client *http.Client, req *http.Request,
) (*http.Response, error) {
	resp, err := client.Do(req) //#nosec G704 -- validated URL, SSRF-safe transport
	if err != nil {
		return nil, maskURLError(err)
	}

	return resp, nil
}