package delivery import ( "context" "log/slog" "net" "net/http" "time" "go.uber.org/fx" "gorm.io/gorm" "sneak.berlin/go/webhooker/internal/database" ) // 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) } // 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) } // ExportStart exposes start for testing. func (e *Engine) ExportStart(ctx context.Context) { e.start(ctx) } // ExportStop exposes stop for testing. func (e *Engine) ExportStop() { e.stop() } // 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, } 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), } 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, } e.initTargets(client) return e } // 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() { s.stop() } // 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) }