Method Tick

internal/webhookpub/worker.go:185–238

Tick processes one batch of pending events. Exposed (not just the private processBatch) so integration tests can drive the worker synchronously instead of waiting on the timer.

(ctx context.Context)

// private processBatch) so integration tests can drive the worker
// synchronously instead of waiting on the timer.
func (w *OutboxWorker) Tick(ctx context.Context) {
	tickStart := time.Now()
	notifyWake := w.notifySawLastTick
	w.notifySawLastTick = false

	// Publisher-lag gauge: age of the oldest pending row.
	var oldestAge float64
	if err := w.pool.QueryRow(ctx,
		`SELECT EXTRACT(EPOCH FROM (now() - min(created_at)))
		 FROM webhook_events WHERE status = 'pending'`,
	).Scan(&oldestAge); err == nil {
		w.metrics.SetPublisherLag(oldestAge)
	}

	events, err := w.leasePending(ctx)
	if err != nil {
		log.Printf("[outbox-worker] leasePending err: %v", err)
		w.metrics.OutboxFailures("lease")
		return
	}
	if len(events) == 0 {
		return
	}
	// If we picked up events without a NOTIFY wakeup, the fallback
	// poll saved us. Non-zero rate signals LISTEN churn.
	if !notifyWake {
		w.metrics.NotifyMissed()
	}
	// Slice 10 telemetry hook: log batch size + age of oldest row so
	// publisher lag can be derived from access logs. A future
	// follow-up wires real Prometheus/OTLP counters.
	var oldest time.Time
	for _, ev := range events {
		_ = ev
	}
	log.Printf("[outbox-worker-metrics] tick batch=%d oldest_age_estimate=lease-bound elapsed_ms_so_far=%d",
		len(events), time.Since(tickStart).Milliseconds())
	_ = oldest
	_ = oldestAge // already emitted via SetPublisherLag

	sem := make(chan struct{}, w.concurrency)
	var wg sync.WaitGroup
	for _, ev := range events {
		ev := ev
		wg.Add(1)
		sem <- struct{}{}
		go func() {
			defer wg.Done()
			defer func() { <-sem }()
			w.fanOutOne(ctx, ev)
		}()
	}
	wg.Wait()
}

// leasedEvent is the worker's row-shape for an in-progress event.
type leasedEvent struct {