at-container-registry/pkg/appview/webhooks/dispatch.go

package webhooks

import (
	"context"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"math/rand/v2"
	"net/http"
	"slices"
	"strings"
	"time"

	"atcr.io/pkg/appview/db"
	"atcr.io/pkg/appview/storage"
	"atcr.io/pkg/atproto"
)

// WebhookLimiter reports a user's current webhook allowance: the max number of
// webhooks (-1 = unlimited) and whether paid trigger types are allowed. It is
// consulted at dispatch time so losing entitlement (switching to a self-hosted
// hold, or a plan downgrade) immediately caps delivery — the creation-time gate
// alone can't catch a transition after the webhook already exists.
type WebhookLimiter func(userDID string) (maxWebhooks int, allTriggers bool)

// Dispatcher handles webhook delivery for push and scan notifications.
// It reads webhooks from the appview DB and delivers payloads
// with Discord/Slack formatting and HMAC signing.
type Dispatcher struct {
	db   db.DBTX
	meta atproto.AppviewMetadata
	// httpClient talks to internal services (hold quota stats). Holds may
	// legitimately live on a private address, so it is deliberately NOT the
	// SSRF-guarded client.
	httpClient *http.Client
	// deliveryClient is the only client that ever dials a user-supplied webhook
	// URL. Its dialer refuses non-public addresses at connect time, which is
	// what stops DNS rebinding and redirect chains from reaching loopback,
	// RFC1918 or the cloud metadata endpoint.
	deliveryClient *http.Client
	// validateURL is the destination policy applied before every attempt. Nil
	// means ValidateWebhookURL. Only tests replace it, so that they can deliver
	// to their own loopback httptest servers.
	validateURL func(string) error
	limits      WebhookLimiter
}

// NewDispatcher creates a new webhook dispatcher. limits may be nil (treated as
// unlimited), but production wires it to the billing manager's GetWebhookLimits.
func NewDispatcher(database db.DBTX, meta atproto.AppviewMetadata, limits WebhookLimiter) *Dispatcher {
	return &Dispatcher{
		db:             database,
		meta:           meta,
		httpClient:     http.DefaultClient,
		deliveryClient: NewSafeClient(deliveryTimeout),
		limits:         limits,
	}
}

// entitledWebhooks caps a user's webhook list to their current allowance and
// reports whether paid triggers are allowed right now. The list must already be
// ordered oldest-first (GetWebhooksForUser) so the cap keeps the oldest N — the
// same ones the creation-time count gate would have permitted. A nil limiter
// (e.g. tests, billing disabled) means unlimited.
func (d *Dispatcher) entitledWebhooks(userDID string, hooks []db.Webhook) ([]db.Webhook, bool) {
	if d.limits == nil {
		return hooks, true
	}
	max, allTriggers := d.limits(userDID)
	if max >= 0 && len(hooks) > max {
		// Log it: the webhooks still show in settings, so from the user's side
		// delivery just stops. This is also the only signal if the limiter itself
		// degraded (a billing lookup failure falls back to free-tier limits),
		// which would silently demote a paying user mid-dispatch.
		slog.Info("Capping webhook delivery to current entitlement",
			"userDID", userDID,
			"configured", len(hooks),
			"allowed", max)
		hooks = hooks[:max]
	}
	return hooks, allTriggers
}

// DispatchForScan fires matching webhooks after a scan record arrives via Jetstream.
// previousScan is nil for first-time scans. userHandle is used for payload enrichment.
func (d *Dispatcher) DispatchForScan(ctx context.Context, scan, previousScan *db.Scan, userHandle, tag, holdEndpoint string) {
	webhooks, err := db.GetWebhooksForUser(d.db, scan.UserDID)
	if err != nil || len(webhooks) == 0 {
		return
	}
	webhooks, allTriggers := d.entitledWebhooks(scan.UserDID, webhooks)

	isFirst := previousScan == nil
	isChanged := previousScan != nil && vulnCountsChanged(scan, previousScan)

	scanInfo := WebhookScanInfo{
		ScannedAt:      scan.ScannedAt.Format(time.RFC3339),
		ScannerVersion: scan.ScannerVersion,
		Vulnerabilities: WebhookVulnCounts{
			Critical: scan.Critical,
			High:     scan.High,
			Medium:   scan.Medium,
			Low:      scan.Low,
			Total:    scan.Total,
		},
	}

	manifestInfo := WebhookManifestInfo{
		Digest:     scan.ManifestDigest,
		Repository: scan.Repository,
		Tag:        tag,
		UserDID:    scan.UserDID,
		UserHandle: userHandle,
	}

	for _, wh := range webhooks {
		// Check each trigger condition against bitmask. Read flag bits via
		// TriggerFlags so packed threshold bits (quota webhooks) don't leak in.
		// Drop paid trigger bits when the user isn't currently entitled.
		flags := TriggerFlags(wh.Triggers)
		if !allTriggers {
			flags &= FreeTriggerMask
		}
		var triggers []string
		if flags&TriggerFirst != 0 && isFirst {
			triggers = append(triggers, "scan:first")
		}
		if flags&TriggerAll != 0 {
			triggers = append(triggers, "scan:all")
		}
		if flags&TriggerChanged != 0 && isChanged {
			triggers = append(triggers, "scan:changed")
		}

		for _, trigger := range triggers {
			payload := WebhookPayload{
				Trigger:      trigger,
				HoldDID:      scan.HoldDID,
				HoldEndpoint: holdEndpoint,
				Manifest:     manifestInfo,
				Scan:         scanInfo,
			}

			// Include previous counts for scan:changed
			if trigger == "scan:changed" && previousScan != nil {
				payload.Previous = &WebhookVulnCounts{
					Critical: previousScan.Critical,
					High:     previousScan.High,
					Medium:   previousScan.Medium,
					Low:      previousScan.Low,
					Total:    previousScan.Total,
				}
			}

			payloadBytes, err := json.Marshal(payload)
			if err != nil {
				slog.Error("Failed to marshal webhook payload", "error", err)
				continue
			}

			go d.deliverWithRetry(wh.URL, wh.Secret, payloadBytes)
		}
	}
}

// DispatchForQuota evaluates each of the user's quota-trigger webhooks and
// fires those whose configured threshold has just been crossed upward. Re-arms
// webhooks whose usage has fallen back below their threshold.
//
// Edge-triggered semantics, keyed by webhook.last_fired_at:
//   - nil  = armed; fire if cur_pct >= threshold, then stamp last_fired_at
//   - set  = already fired and still above; skip. If cur_pct drops below the
//     threshold, clear last_fired_at to re-arm.
//
// Called from the post-push hook in manifest_store. The hold's getQuota
// endpoint is public, so failure is non-fatal — push has already succeeded.
func (d *Dispatcher) DispatchForQuota(ctx context.Context, event storage.QuotaWebhookEvent) {
	hooks, err := db.GetWebhooksForUser(d.db, event.UserDID)
	if err != nil || len(hooks) == 0 {
		return
	}
	// Most pushes have no quota webhook configured at all, and the entitlement
	// lookup can reach the billing backend, so check cheaply whether there is any
	// quota work before paying for it. This preserves the original "do no work on
	// the common path" property that fetching the allowance up front would lose.
	if !slices.ContainsFunc(hooks, func(wh db.Webhook) bool {
		return TriggerFlags(wh.Triggers)&TriggerQuota != 0
	}) {
		return
	}

	// Cap to the user's current allowance before thresholding. The cap is applied
	// over the full list, not the quota subset, because the count limit spans all
	// webhook types, matching the creation-time gate.
	hooks, _ = d.entitledWebhooks(event.UserDID, hooks)

	// Filter to quota-trigger webhooks before doing any work; most pushes
	// will have no quota webhooks configured.
	var quotaHooks []db.Webhook
	for _, wh := range hooks {
		if TriggerFlags(wh.Triggers)&TriggerQuota != 0 {
			quotaHooks = append(quotaHooks, wh)
		}
	}
	if len(quotaHooks) == 0 {
		return
	}

	stats, err := atproto.FetchQuotaStats(ctx, d.httpClient, event.HoldDID, event.UserDID)
	if err != nil {
		slog.Warn("quota webhook: fetch stats failed", "user_did", event.UserDID, "hold_did", event.HoldDID, "error", err)
		return
	}
	if stats.Limit == nil || *stats.Limit <= 0 {
		// Unlimited (captain or no quota tier) — nothing to threshold.
		return
	}

	curPct := int((stats.TotalSize * 100) / *stats.Limit)
	now := time.Now().UTC()

	for _, wh := range quotaHooks {
		threshold := ThresholdPct(wh.Triggers)
		if threshold <= 0 {
			continue
		}

		if curPct < threshold {
			if wh.LastFiredAt != nil {
				if err := db.ClearWebhookLastFiredAt(d.db, wh.ID); err != nil {
					slog.Warn("quota webhook: clear last_fired_at failed", "id", wh.ID, "error", err)
				}
			}
			continue
		}

		if wh.LastFiredAt != nil {
			// Still above threshold and already fired — suppress.
			continue
		}

		payload := QuotaWebhookPayload{
			Trigger: "quota",
			QuotaData: QuotaData{
				Timestamp:        now.Format(time.RFC3339),
				UsageBytes:       stats.TotalSize,
				LimitBytes:       *stats.Limit,
				UsagePercent:     curPct,
				ThresholdPercent: threshold,
			},
			Hold: PushHold{DID: event.HoldDID, Endpoint: event.HoldEndpoint},
			User: QuotaUserInfo{DID: event.UserDID, Handle: event.UserHandle},
		}

		payloadBytes, err := json.Marshal(payload)
		if err != nil {
			slog.Error("quota webhook: marshal payload failed", "error", err)
			continue
		}

		if err := db.UpdateWebhookLastFiredAt(d.db, wh.ID, now); err != nil {
			slog.Warn("quota webhook: stamp last_fired_at failed", "id", wh.ID, "error", err)
			// Stamp first so a transient retry doesn't double-fire; if the
			// write fails, deliver anyway (better one extra alert than none).
		}
		go d.deliverWithRetry(wh.URL, wh.Secret, payloadBytes)
	}
}

// DispatchForPush fires matching webhooks after a manifest is pushed.
func (d *Dispatcher) DispatchForPush(ctx context.Context, event storage.PushWebhookEvent) {
	webhooks, err := db.GetWebhooksForUser(d.db, event.OwnerDID)
	if err != nil || len(webhooks) == 0 {
		return
	}
	webhooks, _ = d.entitledWebhooks(event.OwnerDID, webhooks)

	// Fetch star/pull counts for payload enrichment
	var starCount, pullCount int
	stats, err := db.GetRepositoryStats(d.db, event.OwnerDID, event.Repository)
	if err == nil && stats != nil {
		starCount = stats.StarCount
		pullCount = stats.PullCount
	}

	// Build repo URL using the primary registry domain (pull domain) if available
	baseURL := d.meta.BaseURL
	if len(d.meta.RegistryDomains) > 0 {
		baseURL = "https://" + d.meta.RegistryDomains[0]
	}
	repoURL := fmt.Sprintf("%s/%s/%s", baseURL, event.OwnerHandle, event.Repository)

	payload := PushWebhookPayload{
		Trigger: "push",
		PushData: PushData{
			PushedAt:  time.Now().Format(time.RFC3339),
			Pusher:    event.PusherHandle,
			PusherDID: event.PusherDID,
			Tag:       event.Tag,
			Digest:    event.Digest,
		},
		Repository: PushRepository{
			Name:      event.Repository,
			Namespace: event.OwnerHandle,
			RepoName:  event.OwnerHandle + "/" + event.Repository,
			RepoURL:   repoURL,
			MediaType: event.MediaType,
			StarCount: starCount,
			PullCount: pullCount,
		},
		Hold: PushHold{
			DID:      event.HoldDID,
			Endpoint: event.HoldEndpoint,
		},
	}

	payloadBytes, err := json.Marshal(payload)
	if err != nil {
		slog.Error("Failed to marshal push webhook payload", "error", err)
		return
	}

	for _, wh := range webhooks {
		if TriggerFlags(wh.Triggers)&TriggerPush == 0 {
			continue
		}
		go d.deliverWithRetry(wh.URL, wh.Secret, payloadBytes)
	}
}

// DeliverTest sends a test payload to a specific webhook (synchronous, single attempt)
func (d *Dispatcher) DeliverTest(ctx context.Context, webhookID, userDID, userHandle string) (bool, error) {
	wh, err := db.GetWebhookByID(d.db, webhookID)
	if err != nil {
		return false, err
	}
	if wh.UserDID != userDID {
		return false, fmt.Errorf("unauthorized")
	}

	// Randomize vulnerability counts so each test shows a different severity color
	critical := rand.IntN(3)
	high := rand.IntN(5)
	medium := rand.IntN(8)
	low := rand.IntN(10)
	total := critical + high + medium + low

	payload := WebhookPayload{
		Trigger: "test",
		Manifest: WebhookManifestInfo{
			Digest:     "sha256:0000000000000000000000000000000000000000000000000000000000000000",
			Repository: "test-repo",
			Tag:        "latest",
			UserDID:    userDID,
			UserHandle: userHandle,
		},
		Scan: WebhookScanInfo{
			ScannedAt:      time.Now().Format(time.RFC3339),
			ScannerVersion: "atcr-scanner-v1.0.0",
			Vulnerabilities: WebhookVulnCounts{
				Critical: critical, High: high, Medium: medium, Low: low, Total: total,
			},
		},
	}

	payloadBytes, _ := json.Marshal(payload)
	success := d.attemptDelivery(wh.URL, wh.Secret, payloadBytes)
	return success, nil
}

// deliverWithRetry attempts to deliver a webhook with exponential backoff
func (d *Dispatcher) deliverWithRetry(webhookURL, secret string, payload []byte) {
	// A destination that fails validation will never become valid, so don't
	// hold a goroutine open through the backoff schedule for it.
	if err := d.validateDestination(webhookURL); err != nil {
		slog.Warn("Refusing webhook delivery to disallowed URL", "url", maskURL(webhookURL), "error", err)
		return
	}

	delays := []time.Duration{0, 30 * time.Second, 2 * time.Minute, 8 * time.Minute}
	for attempt, delay := range delays {
		if attempt > 0 {
			time.Sleep(delay)
		}
		if d.attemptDelivery(webhookURL, secret, payload) {
			return
		}
	}
	slog.Warn("Webhook delivery failed after retries", "url", maskURL(webhookURL))
}

// validateDestination applies the destination policy. Tests override
// validateURL so they can deliver to their own loopback httptest servers;
// production always gets ValidateWebhookURL.
func (d *Dispatcher) validateDestination(webhookURL string) error {
	if d.validateURL != nil {
		return d.validateURL(webhookURL)
	}
	return ValidateWebhookURL(webhookURL)
}

// deliveryTimeout bounds a single webhook POST, including redirects.
const deliveryTimeout = 10 * time.Second

// attemptDelivery sends a single webhook HTTP POST
func (d *Dispatcher) attemptDelivery(webhookURL, secret string, payload []byte) bool {
	// Re-validate on every attempt, not just at creation. Rows predating the
	// guard (or written by any future path that skips the handler) are refused
	// here rather than dialed.
	if err := d.validateDestination(webhookURL); err != nil {
		slog.Warn("Refusing webhook delivery to disallowed URL", "url", maskURL(webhookURL), "error", err)
		return false
	}

	ctx, cancel := context.WithTimeout(context.Background(), deliveryTimeout)
	defer cancel()

	// Reformat payload for platform-specific webhook APIs
	sendPayload := payload
	if isDiscordWebhook(webhookURL) || isSlackWebhook(webhookURL) {
		formatted, fmtErr := formatPlatformPayload(payload, webhookURL, d.meta)
		if fmtErr == nil {
			sendPayload = formatted
		}
	}

	req, err := http.NewRequestWithContext(ctx, "POST", webhookURL, strings.NewReader(string(sendPayload)))
	if err != nil {
		slog.Warn("Failed to create webhook request", "error", err)
		return false
	}

	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("User-Agent", d.meta.ClientShortName+"-Webhook/1.0")

	// HMAC signing if secret is set (signs the actual payload sent)
	if secret != "" {
		mac := hmac.New(sha256.New, []byte(secret))
		mac.Write(sendPayload)
		sig := hex.EncodeToString(mac.Sum(nil))
		req.Header.Set("X-Webhook-Signature-256", "sha256="+sig)
	}

	client := d.deliveryClient
	if client == nil {
		// Zero-value Dispatchers (tests) must not fall back to an unguarded
		// client.
		client = NewSafeClient(deliveryTimeout)
	}
	resp, err := client.Do(req)
	if err != nil {
		slog.Warn("Webhook delivery attempt failed", "url", maskURL(webhookURL), "error", err)
		return false
	}
	defer resp.Body.Close()

	if resp.StatusCode >= 200 && resp.StatusCode < 300 {
		slog.Info("Webhook delivered successfully", "url", maskURL(webhookURL), "status", resp.StatusCode)
		return true
	}

	// Read response body for debugging
	body, _ := io.ReadAll(io.LimitReader(resp.Body, 256))
	slog.Warn("Webhook delivery got non-2xx response",
		"url", maskURL(webhookURL),
		"status", resp.StatusCode,
		"body", string(body))
	return false
}

// vulnCountsChanged checks if vulnerability counts differ between scans
func vulnCountsChanged(current, previous *db.Scan) bool {
	return current.Critical != previous.Critical ||
		current.High != previous.High ||
		current.Medium != previous.Medium ||
		current.Low != previous.Low
}