// Package webhook provides webhook dispatch functionality. package webhook import ( "bytes" "context" "crypto/hmac" "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "io" "log/slog" "net/http" "sync" "time" "github.com/google/uuid" "github.com/orchard9/rdev/internal/domain" "github.com/orchard9/rdev/internal/port" ) // DispatcherConfig holds configuration for the webhook dispatcher. type DispatcherConfig struct { // WorkerCount is the number of concurrent delivery workers. WorkerCount int // MaxRetries is the maximum number of retry attempts for failed deliveries. MaxRetries int // Timeout is the HTTP request timeout for webhook deliveries. Timeout time.Duration // RetryBackoff defines the base backoff duration for retries (exponential). RetryBackoff time.Duration // MaxResponseBodySize is the maximum size of response body to store. MaxResponseBodySize int // Logger is the logger to use. Logger *slog.Logger } // DefaultDispatcherConfig returns sensible defaults. func DefaultDispatcherConfig() *DispatcherConfig { return &DispatcherConfig{ WorkerCount: 10, MaxRetries: 3, Timeout: 30 * time.Second, RetryBackoff: 5 * time.Second, MaxResponseBodySize: 1024, // 1KB Logger: slog.Default(), } } // deliveryJob represents a webhook delivery job. type deliveryJob struct { webhook *domain.Webhook event *domain.WebhookEvent deliveryID string retryCount int } // Dispatcher handles webhook delivery with worker pool and retry logic. type Dispatcher struct { repo port.WebhookRepository config *DispatcherConfig client *http.Client // Job queue jobs chan deliveryJob // Shutdown management ctx context.Context cancel context.CancelFunc wg sync.WaitGroup } // NewDispatcher creates a new webhook dispatcher. func NewDispatcher(repo port.WebhookRepository, cfg *DispatcherConfig) *Dispatcher { if cfg == nil { cfg = DefaultDispatcherConfig() } ctx, cancel := context.WithCancel(context.Background()) return &Dispatcher{ repo: repo, config: cfg, client: &http.Client{ Timeout: cfg.Timeout, }, jobs: make(chan deliveryJob, 1000), // Buffered channel for job queue ctx: ctx, cancel: cancel, } } // Ensure Dispatcher implements port.WebhookDispatcher at compile time. var _ port.WebhookDispatcher = (*Dispatcher)(nil) // Start starts the background dispatcher workers. func (d *Dispatcher) Start() error { d.config.Logger.Info("webhook dispatcher starting", "workers", d.config.WorkerCount) // Start worker goroutines for i := 0; i < d.config.WorkerCount; i++ { d.wg.Add(1) go d.worker(i) } return nil } // Stop gracefully shuts down the dispatcher. func (d *Dispatcher) Stop() { d.config.Logger.Info("webhook dispatcher stopping") d.cancel() close(d.jobs) d.wg.Wait() d.config.Logger.Info("webhook dispatcher stopped") } // Health returns true if the dispatcher is running and healthy. func (d *Dispatcher) Health() bool { select { case <-d.ctx.Done(): return false default: return true } } // QueueSize returns the current number of pending jobs in the queue. func (d *Dispatcher) QueueSize() int { return len(d.jobs) } // Dispatch sends an event to all subscribed webhooks for a project. // This is a non-blocking operation - deliveries happen in the background. func (d *Dispatcher) Dispatch(ctx context.Context, projectID string, event *domain.WebhookEvent) error { // Find all enabled webhooks that subscribe to this event type webhooks, err := d.repo.ListEnabledByProjectAndEvent(ctx, projectID, event.Type) if err != nil { return fmt.Errorf("list webhooks: %w", err) } if len(webhooks) == 0 { return nil // No webhooks to dispatch to } d.config.Logger.Debug("dispatching webhook event", "project_id", projectID, "event_type", event.Type, "webhook_count", len(webhooks), ) // Queue delivery jobs for each webhook for _, webhook := range webhooks { deliveryID := uuid.New().String() select { case d.jobs <- deliveryJob{ webhook: webhook, event: event, deliveryID: deliveryID, retryCount: 0, }: // Job queued successfully default: // Job queue is full, log warning d.config.Logger.Warn("webhook job queue full, dropping event", "webhook_id", webhook.ID, "event_type", event.Type, ) } } return nil } // worker processes delivery jobs from the queue. func (d *Dispatcher) worker(id int) { defer d.wg.Done() d.config.Logger.Debug("webhook worker started", "worker_id", id) for { select { case <-d.ctx.Done(): d.config.Logger.Debug("webhook worker stopping", "worker_id", id) return case job, ok := <-d.jobs: if !ok { d.config.Logger.Debug("webhook worker job channel closed", "worker_id", id) return } d.processJob(job) } } } // processJob delivers a webhook and handles retries. func (d *Dispatcher) processJob(job deliveryJob) { delivery := d.deliver(job) // Record the delivery attempt recordCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() if err := d.repo.RecordDelivery(recordCtx, delivery); err != nil { d.config.Logger.Error("failed to record webhook delivery", "webhook_id", job.webhook.ID, "delivery_id", delivery.ID, "error", err, ) } // Handle retry if delivery failed if !delivery.Success && job.retryCount < d.config.MaxRetries { // Calculate exponential backoff backoff := d.config.RetryBackoff * time.Duration(1<= 200 && resp.StatusCode < 300 { delivery.Success = true d.config.Logger.Debug("webhook delivered successfully", "webhook_id", job.webhook.ID, "delivery_id", delivery.ID, "status", resp.StatusCode, ) } else { delivery.Success = false delivery.ErrorMessage = fmt.Sprintf("received non-2xx status: %d", resp.StatusCode) d.config.Logger.Debug("webhook delivery failed", "webhook_id", job.webhook.ID, "delivery_id", delivery.ID, "status", resp.StatusCode, ) } return delivery } // signPayload creates an HMAC-SHA256 signature of the payload. func (d *Dispatcher) signPayload(payload []byte, secret string) string { mac := hmac.New(sha256.New, []byte(secret)) mac.Write(payload) return "sha256=" + hex.EncodeToString(mac.Sum(nil)) }