webhooker/internal/delivery/target_http.go

package delivery

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"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"`
}

// 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 maxRetries of 0 is fire-and-forget: a single attempt is
// recorded and no circuit breaker is consulted. A positive
// maxRetries gates the attempt on the circuit breaker and
// schedules a backed-off retry on failure.
func (c *httpCore) deliver(
	webhookDB *gorm.DB,
	d *database.Delivery,
	task *Task,
	sched Scheduler,
	maxRetries int,
	attempt func() attemptResult,
) {
	if maxRetries == 0 {
		c.fireAndForget(webhookDB, d, attempt())

		return
	}

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

func (c *httpCore) fireAndForget(
	webhookDB *gorm.DB,
	d *database.Delivery,
	res attemptResult,
) {
	c.eng.recordResult(
		webhookDB, d, 1, res.success,
		res.statusCode, res.respBody, res.errMsg,
		res.duration,
	)

	if res.success {
		c.eng.updateDeliveryStatus(
			webhookDB, d,
			database.DeliveryStatusDelivered,
		)

		return
	}

	c.eng.updateDeliveryStatus(
		webhookDB, d, database.DeliveryStatusFailed,
	)
}

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

	attemptNum := task.AttemptNum

	res := attempt()

	c.eng.recordResult(
		webhookDB, d, attemptNum, res.success,
		res.statusCode, res.respBody, res.errMsg,
		res.duration,
	)

	if res.success {
		cb.RecordSuccess()

		c.eng.updateDeliveryStatus(
			webhookDB, d,
			database.DeliveryStatusDelivered,
		)

		return
	}

	cb.RecordFailure()

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

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

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

	c.eng.updateDeliveryStatus(
		webhookDB, d,
		database.DeliveryStatusRetrying,
	)

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

	return true
}

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

		return
	}

	c.eng.updateDeliveryStatus(
		webhookDB, d, database.DeliveryStatusRetrying,
	)

	backoff := calcBackoff(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
}

// 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 := calcBackoff(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 := calcBackoff(attemptNum)

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

func calcBackoff(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,
		)

		t.eng.recordResult(
			webhookDB, d, task.AttemptNum,
			false, 0, "", err.Error(), 0,
		)

		t.eng.updateDeliveryStatus(
			webhookDB, d, database.DeliveryStatusFailed,
		)

		return
	}

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

	t.deliver(
		webhookDB, d, task, sched,
		d.Target.MaxRetries, 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", reqErr,
		)
	}

	applyRequestHeaders(req, event, cfg)

	client := t.clientForConfig(cfg)

	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
}

func (t *httpTarget) clientForConfig(
	cfg *HTTPTargetConfig,
) *http.Client {
	if cfg.Timeout > 0 {
		// Reuse the shared client's SSRF-safe transport so
		// a per-target timeout does not drop the
		// request-time private-IP guard. Only the timeout
		// is overridden.
		return &http.Client{
			Timeout: time.Duration(
				cfg.Timeout,
			) * time.Second,
			Transport: t.client.Transport,
		}
	}

	return t.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
	default:
		return true
	}
}

func applyRequestHeaders(
	req *http.Request,
	event *database.Event,
	cfg *HTTPTargetConfig,
) {
	if event.ContentType != "" {
		req.Header.Set(
			"Content-Type", event.ContentType,
		)
	}

	var originalHeaders map[string][]string

	if event.Headers != "" {
		jsonErr := json.Unmarshal(
			[]byte(event.Headers),
			&originalHeaders,
		)
		if jsonErr == nil {
			for k, vals := range originalHeaders {
				if isForwardableHeader(k) {
					for _, v := range vals {
						req.Header.Add(k, v)
					}
				}
			}
		}
	}

	for k, v := range cfg.Headers {
		req.Header.Set(k, v)
	}

	req.Header.Set("User-Agent", "webhooker/1.0")
}

// executeHTTPRequest sends an HTTP request using the provided
// client. URLs are validated by the config parsers and the
// SSRF-safe transport before reaching here.
func executeHTTPRequest(
	client *http.Client, req *http.Request,
) (*http.Response, error) {
	return client.Do(req) //#nosec G704 -- URL validated by parseHTTPConfig/parseSlackConfig and SSRF-safe transport
}