webhooker/internal/delivery/export_test.go

package delivery

import (
	"context"
	"log/slog"
	"net"
	"net/http"
	"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
)

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

// 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.
func ExportApplyRequestHeaders(
	req *http.Request,
	event *database.Event,
	cfg *HTTPTargetConfig,
) {
	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)
}

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

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

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

// ExportClientForConfig exposes the http target's
// clientForConfig.
func (e *Engine) ExportClientForConfig(
	cfg *HTTPTargetConfig,
) *http.Client {
	return e.httpTarget.clientForConfig(cfg)
}

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

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