webhooker/internal/delivery/export_test.go

package delivery

import (
	"context"
	"log/slog"
	"net"
	"net/http"
	"net/netip"
	"net/url"
	"time"

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

// ErrExportArchiveWriterEvicted exposes the sentinel returned by
// an evicted archive writer. It carries the Err prefix rather
// than this file's usual Export one because it is a sentinel
// error.
var ErrExportArchiveWriterEvicted = errArchiveWriterEvicted

// Exported constants for test access.
const (
	ExportDeliveryChannelSize     = deliveryChannelSize
	ExportRetryChannelSize        = retryChannelSize
	ExportDefaultFailureThreshold = defaultFailureThreshold
	ExportDefaultCooldown         = defaultCooldown

	// ExportMaxBodyLog is the cap the engine applies to a
	// recorded response body. The event log's handling of a cut
	// response is written against this number, so a test has to
	// be able to name it.
	ExportMaxBodyLog = maxBodyLog

	// ExportPendingSweepMinAge is how long a delivery must sit at
	// pending before the sweep treats it as stranded. A test has to
	// name it to age a row past the bound.
	ExportPendingSweepMinAge = pendingSweepMinAge
)

// ExportIsBlockedIP exposes isBlockedIP for testing.
func ExportIsBlockedIP(ip net.IP) bool {
	return isBlockedIP(ip)
}

// NewTestGuard builds an SSRF Guard from an explicit egress
// allowlist, without going through config. Passing no prefixes
// yields the default guard, which blocks every private/reserved
// range.
func NewTestGuard(allowed ...netip.Prefix) *Guard {
	return &Guard{allowed: allowed}
}

// ExportCheckIP exposes the guard's single decision point, so a
// test can assert the policy both the validator and the dialer
// inherit without needing a live destination.
func (g *Guard) ExportCheckIP(ip net.IP) error {
	return g.checkIP(ip)
}

// ExportAlwaysBlockedNetworks exposes alwaysBlockedNetworks.
func ExportAlwaysBlockedNetworks() []*net.IPNet {
	return alwaysBlockedNetworks
}

// ExportBlockedNetworks exposes blockedNetworks.
func ExportBlockedNetworks() []*net.IPNet {
	return blockedNetworks
}

// ExportIsForwardableHeader exposes isForwardableHeader.
func ExportIsForwardableHeader(name string) bool {
	return isForwardableHeader(name)
}

// ExportApplyRequestHeaders exposes applyRequestHeaders, so a test
// can inspect the header set an outbound delivery actually carries
// and the origin-scoped names it reports for the redirect policy.
func ExportApplyRequestHeaders(
	req *http.Request,
	event *database.Event,
	cfg *HTTPTargetConfig,
) []string {
	return applyRequestHeaders(req, event, cfg)
}

// ExportTruncate exposes truncate for testing.
func ExportTruncate(s string, maxLen int) string {
	return truncate(s, maxLen)
}

// ExportDeliverHTTP delivers via the http target for testing.
func (e *Engine) ExportDeliverHTTP(
	ctx context.Context,
	webhookDB *gorm.DB,
	d *database.Delivery,
	task *Task,
) {
	e.httpTarget.Deliver(ctx, webhookDB, d, task, e)
}

// ExportDeliverHTTPWithScheduler delivers via the http target, handing
// any retry to sched instead of the engine, so a test can see the
// delay each retry is given.
func (e *Engine) ExportDeliverHTTPWithScheduler(
	ctx context.Context,
	webhookDB *gorm.DB,
	d *database.Delivery,
	task *Task,
	sched Scheduler,
) {
	e.httpTarget.Deliver(ctx, webhookDB, d, task, sched)
}

// ExportDeliverDatabase delivers via the database target.
func (e *Engine) ExportDeliverDatabase(
	webhookDB *gorm.DB, d *database.Delivery,
) {
	e.targets[database.TargetTypeDatabase].Deliver(
		context.Background(), webhookDB, d, &Task{}, e,
	)
}

// ExportDeliverLog delivers via the log target for testing.
func (e *Engine) ExportDeliverLog(
	webhookDB *gorm.DB, d *database.Delivery,
) {
	e.targets[database.TargetTypeLog].Deliver(
		context.Background(), webhookDB, d, &Task{}, e,
	)
}

// ExportDeliverSlack delivers via the slack target for
// testing.
func (e *Engine) ExportDeliverSlack(
	ctx context.Context,
	webhookDB *gorm.DB,
	d *database.Delivery,
) {
	task := &Task{
		DeliveryID: d.ID,
		TargetID:   d.TargetID,
		AttemptNum: 1,
	}

	e.targets[database.TargetTypeSlack].Deliver(
		ctx, webhookDB, d, task, e,
	)
}

// ExportProcessNewTask exposes processNewTask.
func (e *Engine) ExportProcessNewTask(
	ctx context.Context, task *Task,
) {
	e.processNewTask(ctx, task)
}

// ExportProcessRetryTask exposes processRetryTask.
func (e *Engine) ExportProcessRetryTask(
	ctx context.Context, task *Task,
) {
	e.processRetryTask(ctx, task)
}

// ExportEventForTask exposes the event reconstruction the delivery
// paths run: buildEventFromTask followed by hydrateEvent.
func (e *Engine) ExportEventForTask(
	webhookDB *gorm.DB, task *Task,
) (database.Event, error) {
	return e.hydrateEvent(
		webhookDB, buildEventFromTask(task), task,
	)
}

// ExportProcessDelivery exposes processDelivery.
func (e *Engine) ExportProcessDelivery(
	ctx context.Context,
	webhookDB *gorm.DB,
	d *database.Delivery,
	task *Task,
) {
	e.processDelivery(ctx, webhookDB, d, task)
}

// ExportGetCircuitBreaker exposes the http target's
// getCircuitBreaker.
func (e *Engine) ExportGetCircuitBreaker(
	targetID string,
) *CircuitBreaker {
	return e.httpTarget.getCircuitBreaker(targetID)
}

// ExportSetCircuitBreaker makes cb the http target's circuit breaker
// for targetID, so a test can use one with a short cooldown.
func (e *Engine) ExportSetCircuitBreaker(
	targetID string, cb *CircuitBreaker,
) {
	e.httpTarget.circuitBreakers.Store(targetID, cb)
}

// ExportParseHTTPConfig exposes parseHTTPConfig.
func (e *Engine) ExportParseHTTPConfig(
	configJSON string,
) (*HTTPTargetConfig, error) {
	return parseHTTPConfig(configJSON)
}

// ExportParseSlackConfig exposes parseSlackConfig.
func (e *Engine) ExportParseSlackConfig(
	configJSON string,
) (*SlackTargetConfig, error) {
	return parseSlackConfig(configJSON)
}

// ExportDoHTTPRequest exposes the http target's
// doHTTPRequest.
func (e *Engine) ExportDoHTTPRequest(
	ctx context.Context,
	cfg *HTTPTargetConfig,
	event *database.Event,
) (int, string, int64, error) {
	return e.httpTarget.doHTTPRequest(ctx, cfg, event)
}

// ExportClientForRequest exposes the http target's
// clientForRequest.
func (e *Engine) ExportClientForRequest(
	cfg *HTTPTargetConfig,
	originScoped []string,
) *http.Client {
	return e.httpTarget.clientForRequest(cfg, originScoped)
}

// ErrExportTooManyRedirects exposes the sentinel the redirect
// policy returns once a chain exceeds the hop cap. It carries the
// Err prefix rather than this file's usual Export one because it
// is a sentinel error.
var ErrExportTooManyRedirects = errTooManyRedirects

// ExportMaxDeliveryRedirects exposes the redirect hop cap.
const ExportMaxDeliveryRedirects = maxDeliveryRedirects

// ExportSameDeliveryOrigin exposes sameDeliveryOrigin.
func ExportSameDeliveryOrigin(origin, dest *url.URL) bool {
	return sameDeliveryOrigin(origin, dest)
}

// ExportClient returns the http target's shared HTTP client.
func (e *Engine) ExportClient() *http.Client {
	return e.httpTarget.client
}

// ExportScheduleRetry exposes ScheduleRetry.
func (e *Engine) ExportScheduleRetry(
	task Task, delay time.Duration,
) {
	e.ScheduleRetry(task, delay)
}

// ExportRecoverPendingDeliveries exposes
// recoverPendingDeliveries.
func (e *Engine) ExportRecoverPendingDeliveries(
	ctx context.Context,
	webhookDB *gorm.DB,
	webhookID string,
) {
	e.recoverPendingDeliveries(
		ctx, webhookDB, webhookID,
	)
}

// ExportRecoverWebhookDeliveries exposes
// recoverWebhookDeliveries.
func (e *Engine) ExportRecoverWebhookDeliveries(
	ctx context.Context, webhookID string,
) {
	e.recoverWebhookDeliveries(ctx, webhookID)
}

// ExportRecoverInFlight exposes recoverInFlight.
func (e *Engine) ExportRecoverInFlight(
	ctx context.Context,
) {
	e.recoverInFlight(ctx)
}

// ExportSweepWebhookRetries exposes sweepWebhookRetries.
func (e *Engine) ExportSweepWebhookRetries(
	ctx context.Context, webhookID string,
) {
	e.sweepWebhookRetries(ctx, webhookID)
}

// ExportStart exposes start for testing.
func (e *Engine) ExportStart() {
	e.start()
}

// ExportRegisterHooks registers the engine's real fx lifecycle
// hooks on a lifecycle supplied by a test, so a test can drive
// the exact OnStart/OnStop functions the application runs and
// hand OnStart the kind of context fx actually supplies.
func (e *Engine) ExportRegisterHooks(lc fx.Lifecycle) {
	e.registerHooks(lc)
}

// ExportStop exposes stop for testing.
func (e *Engine) ExportStop(ctx context.Context) error {
	return e.stop(ctx)
}

// ExportWedgeWorker adds a goroutine to the engine's WaitGroup
// that never observes cancellation and returns only when release
// is closed. It stands in for a worker stuck inside a delivery
// target that never returns, which is the only way stop can be
// made to outlast its context.
func (e *Engine) ExportWedgeWorker(release <-chan struct{}) {
	e.wg.Go(func() {
		<-release
	})
}

// ExportInflightHeld reports how many deliveries the engine currently
// owns, so a test can prove ownership is released rather than leaked.
func (e *Engine) ExportInflightHeld() int {
	return e.inflight.held()
}

// ExportRetainDelivery takes the first reference on a delivery, as the
// queueing side does. It lets a test put a delivery into the state a
// worker or a full channel would, without running the pool.
func (e *Engine) ExportRetainDelivery(deliveryID string) bool {
	return e.inflight.retainIdle(deliveryID)
}

// ExportRecoverRetryingDeliveries exposes recoverRetryingDeliveries.
func (e *Engine) ExportRecoverRetryingDeliveries(
	webhookDB *gorm.DB, webhookID string,
) {
	e.recoverRetryingDeliveries(webhookDB, webhookID)
}

// ExportDeliveryCh returns the delivery channel.
func (e *Engine) ExportDeliveryCh() chan Task {
	return e.deliveryCh
}

// ExportRetryCh returns the retry channel.
func (e *Engine) ExportRetryCh() chan Task {
	return e.retryCh
}

// NewTestEngine creates an Engine for unit tests without
// database dependencies.
func NewTestEngine(
	log *slog.Logger,
	client *http.Client,
	workers int,
) *Engine {
	e := &Engine{
		log:        log,
		deliveryCh: make(chan Task, deliveryChannelSize),
		retryCh:    make(chan Task, retryChannelSize),
		workers:    workers,
		mtr:        metrics.Default(),
	}
	e.initTargets(client)

	return e
}

// NewTestEngineSmallRetry creates an Engine with a tiny
// retry channel buffer for overflow testing.
func NewTestEngineSmallRetry(
	log *slog.Logger,
) *Engine {
	e := &Engine{
		log:     log,
		retryCh: make(chan Task, 1),
		mtr:     metrics.Default(),
	}
	e.initTargets(nil)

	return e
}

// NewTestEngineWithDB creates an Engine with a real
// database and dbManager for integration tests.
func NewTestEngineWithDB(
	db *database.Database,
	dbMgr *database.WebhookDBManager,
	log *slog.Logger,
	client *http.Client,
	workers int,
) *Engine {
	e := &Engine{
		database:   db,
		dbManager:  dbMgr,
		log:        log,
		deliveryCh: make(chan Task, deliveryChannelSize),
		retryCh:    make(chan Task, retryChannelSize),
		workers:    workers,
		mtr:        metrics.Default(),
	}
	e.initTargets(client)

	return e
}

// ExportSetMetrics substitutes the engine's metric set, so a test can
// assert on collectors registered on a private registry instead of
// the process-wide ones every other test is also moving.
func (e *Engine) ExportSetMetrics(mtr *metrics.Set) {
	e.mtr = mtr
}

// ExportSampleQueueDepths runs one queue depth sample synchronously.
func (e *Engine) ExportSampleQueueDepths(ctx context.Context) {
	e.sampleQueueDepths(ctx)
}

// NewTestCircuitBreaker creates a CircuitBreaker with
// custom settings for testing.
func NewTestCircuitBreaker(
	threshold int, cooldown time.Duration,
) *CircuitBreaker {
	return &CircuitBreaker{
		state:     CircuitClosed,
		threshold: threshold,
		cooldown:  cooldown,
	}
}

// ExportArchivedEvent aliases the archive row type so black-box
// tests can construct and read archive rows.
type ExportArchivedEvent = archivedEvent

// ExportArchiveWriter wraps an archiveWriter so black-box tests
// can exercise the per-webhook archive file mechanics.
type ExportArchiveWriter struct {
	w *archiveWriter
}

// NewExportArchiveWriter builds an archive writer for tests,
// optionally overriding the reopen debounce (a non-positive
// debounce keeps the production default).
func NewExportArchiveWriter(
	path string, log *slog.Logger, debounce time.Duration,
) *ExportArchiveWriter {
	w := newArchiveWriter(path, log)
	if debounce > 0 {
		w.debounce = debounce
	}

	return &ExportArchiveWriter{w: w}
}

// Write archives a row through the writer.
func (e *ExportArchiveWriter) Write(
	row ExportArchivedEvent, expiry time.Duration,
) error {
	return e.w.write(row, expiry)
}

// Open opens the archive file, pruning when expiry is positive.
func (e *ExportArchiveWriter) Open(expiry time.Duration) error {
	return e.w.open(expiry)
}

// Reopen closes and reopens the archive file.
func (e *ExportArchiveWriter) Reopen(
	expiry time.Duration,
) error {
	return e.w.reopen(expiry)
}

// Reopens reports how many times the file has been opened.
func (e *ExportArchiveWriter) Reopens() int {
	return e.w.reopens
}

// DB returns the writer's current open handle for row
// inspection in tests.
func (e *ExportArchiveWriter) DB() *gorm.DB {
	return e.w.db
}

// Path returns the archive file the writer owns.
func (e *ExportArchiveWriter) Path() string {
	return e.w.path
}

// OpenExisting opens the archive without permitting creation,
// the way the idle sweep does.
func (e *ExportArchiveWriter) OpenExisting(
	expiry time.Duration,
) error {
	return e.w.openMode(archiveModeExisting, expiry)
}

// SweepExpired runs an idle sweep of the archive.
func (e *ExportArchiveWriter) SweepExpired(
	expiry time.Duration,
) error {
	return e.w.sweepExpired(expiry)
}

// Evict marks the writer evicted and closes its handle, exactly
// as leaving the registry does.
func (e *ExportArchiveWriter) Evict() {
	e.w.evict()
}

// HandleOpen reports whether the writer currently holds an open
// archive handle.
func (e *ExportArchiveWriter) HandleOpen() bool {
	e.w.mu.Lock()
	defer e.w.mu.Unlock()

	return e.w.db != nil
}

// Same reports whether both wrappers refer to the very same
// underlying archive writer, so a test can prove a registry entry
// is the writer it was handed rather than a replacement.
func (e *ExportArchiveWriter) Same(
	other *ExportArchiveWriter,
) bool {
	return other != nil && e.w == other.w
}

// ExportArchiveWriterFor returns the archive writer the registry
// currently caches for a webhook, or nil when none is cached. It
// never creates one, so a test can hold a reference to the very
// writer an eviction is about to detach.
func (e *Engine) ExportArchiveWriterFor(
	webhookID string,
) *ExportArchiveWriter {
	e.dbTarget.mu.Lock()
	defer e.dbTarget.mu.Unlock()

	w, ok := e.dbTarget.writers[webhookID]
	if !ok {
		return nil
	}

	return &ExportArchiveWriter{w: w}
}

// ExportHasArchiveWriter reports whether the database target
// currently caches an archive writer for a webhook.
func (e *Engine) ExportHasArchiveWriter(
	webhookID string,
) bool {
	e.dbTarget.mu.Lock()
	defer e.dbTarget.mu.Unlock()

	_, ok := e.dbTarget.writers[webhookID]

	return ok
}

// ExportArchiveHandleOpen reports whether the cached archive
// writer for a webhook holds an open database handle. It
// returns false when no writer is cached.
func (e *Engine) ExportArchiveHandleOpen(
	webhookID string,
) bool {
	e.dbTarget.mu.Lock()
	w, ok := e.dbTarget.writers[webhookID]
	e.dbTarget.mu.Unlock()

	if !ok {
		return false
	}

	w.mu.Lock()
	defer w.mu.Unlock()

	return w.db != nil
}

// ExportEnsureArchiveWriter creates (if needed) and returns the
// archive file path of the cached writer for a webhook, so a
// test can prime the registry the way a delivery would.
func (e *Engine) ExportEnsureArchiveWriter(
	webhookID string,
) (string, error) {
	w, err := e.dbTarget.writerFor(webhookID)
	if err != nil {
		return "", err
	}

	return w.path, nil
}

// ExportSweepWriterFor takes a webhook's registry writer exactly
// as the idle sweep does, reporting whether the sweep had to
// create the entry. It lets a test drive the registry through the
// sweep's own entry point instead of choreographing goroutines.
func (e *Engine) ExportSweepWriterFor(
	webhookID string,
) (*ExportArchiveWriter, bool, error) {
	w, created, err := e.dbTarget.sweepWriterFor(webhookID)
	if err != nil {
		return nil, false, err
	}

	return &ExportArchiveWriter{w: w}, created, nil
}

// ExportReleaseSweepWriter releases a sweep-created registry entry
// exactly as a finished sweep does.
func (e *Engine) ExportReleaseSweepWriter(
	webhookID string, w *ExportArchiveWriter,
) {
	e.dbTarget.releaseSweepWriter(webhookID, w.w)
}

// NewTestArchiveSweeper builds an ArchiveSweeper backed by the
// given main database and engine, without the fx lifecycle.
// Intended for tests.
func NewTestArchiveSweeper(
	db *database.Database,
	eng *Engine,
	log *slog.Logger,
) *ArchiveSweeper {
	return &ArchiveSweeper{
		db:       db,
		eng:      eng,
		log:      log,
		interval: time.Hour,
	}
}

// ExportSweep runs a single archive sweep synchronously for
// tests.
func (s *ArchiveSweeper) ExportSweep(ctx context.Context) {
	s.sweep(ctx)
}

// ExportStart starts the sweeper's background loop for tests.
func (s *ArchiveSweeper) ExportStart() {
	s.start()
}

// ExportRegisterHooks registers the sweeper's real fx lifecycle
// hooks on a lifecycle supplied by a test, so a test can drive
// the exact OnStart/OnStop functions the application runs and
// hand OnStart the kind of context fx actually supplies.
func (s *ArchiveSweeper) ExportRegisterHooks(lc fx.Lifecycle) {
	s.registerHooks(lc)
}

// ExportStop stops the sweeper's background loop for tests.
func (s *ArchiveSweeper) ExportStop(ctx context.Context) error {
	return s.stop(ctx)
}

// ExportWedgeLoop adds a goroutine to the sweeper's WaitGroup
// that never observes cancellation and returns only when release
// is closed. It stands in for a prune stuck on a locked archive.
func (s *ArchiveSweeper) ExportWedgeLoop(
	release <-chan struct{},
) {
	s.wg.Go(func() {
		<-release
	})
}

// ExportSetInterval overrides the sweep interval for tests.
func (s *ArchiveSweeper) ExportSetInterval(d time.Duration) {
	s.interval = d
}

// ExportParseArchiveExpiry exposes parseArchiveExpiry.
func ExportParseArchiveExpiry(
	configJSON string,
) (time.Duration, error) {
	return parseArchiveExpiry(configJSON)
}