gitFlow/internal/webhook/webhook.go

// Package webhook sends scan results to external services (Slack, Discord,
// generic HTTP) when repositories change during a watch session.
package webhook

import (
	"context"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"

	"gitea.oblak.solutions/dimitar/gitFlow/internal/httpclient"
	"gitea.oblak.solutions/dimitar/gitFlow/pkg/status"
)

// Payload is the structured data sent to every webhook.
type Payload struct {
	Timestamp time.Time         `json:"timestamp"`
	ParentDir string            `json:"parent_dir"`
	Summary   status.Summary    `json:"summary"`
	Changed   []status.RepoInfo `json:"changed"`
	All       []status.RepoInfo `json:"all,omitempty"`
}

// Sender is the interface every webhook provider implements.
type Sender interface {
	Name() string
	Send(ctx context.Context, payload Payload) error
}

// Config holds per-webhook settings from the configuration file.
type Config struct {
	Name         string            `yaml:"name" mapstructure:"name"`
	Type         string            `yaml:"type" mapstructure:"type"`
	URL          string            `yaml:"url" mapstructure:"url"`
	SendAll      bool              `yaml:"send_all" mapstructure:"send_all"`
	OnChangeOnly bool              `yaml:"on_change_only" mapstructure:"on_change_only"`
	MinPriority  int               `yaml:"min_priority" mapstructure:"min_priority"`
	RateLimit    time.Duration     `yaml:"rate_limit" mapstructure:"-"`
	Timeout      time.Duration     `yaml:"timeout" mapstructure:"-"`
	Headers      map[string]string `yaml:"headers,omitempty" mapstructure:"headers"`
	Retry        RetryConfig       `yaml:"retry" mapstructure:"retry"`
}

// RetryConfig controls retry behaviour per webhook.
type RetryConfig struct {
	MaxAttempts int           `yaml:"max_attempts" mapstructure:"max_attempts"`
	Backoff     time.Duration `yaml:"backoff" mapstructure:"-"`
}

// NewSender builds a Sender from a configuration.
func NewSender(cfg Config) (Sender, error) {
	switch cfg.Type {
	case "slack":
		return newSlack(cfg), nil
	case "discord":
		return newDiscord(cfg), nil
	case "generic":
		return newGeneric(cfg), nil
	default:
		return nil, fmt.Errorf("webhook: unknown type %q (want slack, discord, or generic)", cfg.Type)
	}
}

// Dispatcher fans a payload out to every sender, respecting per-sender
// rate limits and retry policies. It is safe for concurrent use.
type Dispatcher struct {
	senders []senderHandle
}

type senderHandle struct {
	sender   Sender
	cfg      Config
	http     *httpclient.Client
	lastSent time.Time
	mu       sync.Mutex
}

// NewDispatcher builds a Dispatcher from a list of webhook configs. Senders
// whose creation fails are logged and skipped (a single bad webhook should
// not prevent others from firing).
func NewDispatcher(cfgs []Config) *Dispatcher {
	d := &Dispatcher{}
	for _, cfg := range cfgs {
		s, err := NewSender(cfg)
		if err != nil {
			log.Printf("webhook: %s: %v", cfg.Name, err)
			continue
		}
		d.senders = append(d.senders, senderHandle{
			sender: s,
			cfg:    cfg,
			http:   httpclient.New(httpclient.WithTimeout(cfg.TimeoutOrDefault())),
		})
	}
	return d
}

// Dispatch sends the payload to every sender whose filters match. Errors
// from individual senders are collected and returned; a failing sender
// never prevents others from firing.
func (d *Dispatcher) Dispatch(ctx context.Context, payload Payload, changedCount int) []error {
	var errs []error
	for i := range d.senders {
		h := &d.senders[i]
		if h.cfg.OnChangeOnly && changedCount == 0 {
			continue
		}
		if !h.cfg.allowNow() {
			continue
		}
		h.mu.Lock()
		h.lastSent = time.Now()
		h.mu.Unlock()

		if err := h.sendWithRetry(ctx, payload); err != nil {
			errs = append(errs, fmt.Errorf("webhook %s: %w", h.cfg.Name, err))
		}
	}
	return errs
}

// allowNow checks whether the rate limit has elapsed since lastSend.
func (c Config) allowNow() bool {
	return c.RateLimit <= 0
}

// allowNow checks the sender-level rate limit.
func (h *senderHandle) allowNow() bool {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.cfg.RateLimit <= 0 || time.Since(h.lastSent) >= h.cfg.RateLimit
}

// sendWithRetry attempts to send with exponential backoff for transient
// failures. Non-retryable errors (4xx client errors) fail immediately.
func (h *senderHandle) sendWithRetry(ctx context.Context, payload Payload) error {
	maxAttempts := h.cfg.Retry.MaxAttempts
	if maxAttempts < 1 {
		maxAttempts = 1
	}
	backoff := h.cfg.Retry.Backoff
	if backoff <= 0 {
		backoff = 1 * time.Second
	}

	var lastErr error
	for attempt := 1; attempt <= maxAttempts; attempt++ {
		lastErr = h.sender.Send(ctx, payload)
		if lastErr == nil {
			return nil
		}
		if !isRetryable(lastErr) {
			return lastErr
		}
		if attempt < maxAttempts {
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(backoff * time.Duration(1<<(attempt-1))):
			}
		}
	}
	return fmt.Errorf("%d attempts: %w", maxAttempts, lastErr)
}

func isRetryable(err error) bool {
	// 4xx = client error (non-retryable); 5xx/timeout/DNS = retryable.
	if strings.Contains(err.Error(), "400") || strings.Contains(err.Error(), "401") ||
		strings.Contains(err.Error(), "403") || strings.Contains(err.Error(), "404") {
		return false
	}
	return true
}

// TimeoutOrDefault returns the configured timeout or 10s when unset.
func (c Config) TimeoutOrDefault() time.Duration {
	if c.Timeout > 0 {
		return c.Timeout
	}
	return 10 * time.Second
}