package service

import (
	"bytes"
	"context"
	"crypto/hmac"
	"crypto/rand"
	"crypto/sha1"
	"crypto/sha256"
	"crypto/tls"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/url"
	"strings"
	"time"

	"gorm.io/gorm"

	"github.com/pingcap/agent-git-service/internal/db"
)

const (
	webhookStatusPending                = "pending"
	webhookStatusOK                     = "ok"
	webhookStatusFailed                 = "failed"
	webhookStatusError                  = "error"
	webhookQueueTimeout                 = 5 * time.Second
	webhookRequestTimeout               = 10 * time.Second
	webhookResponseBodyLimit            = 64 * 1024
	webhookWorkerCount                  = 4
	webhookWorkerQueueSize              = 256
	durableWebhookPayloadMax            = 8 << 20
	webhookRequestBodyMax               = 8 << 20
	webhookRequestURLMax                = 2048
	webhookRequestHeadersMax            = (64 << 10) - 1
	webhookTerminalErrorMax             = 1024
	webhookDeliveryPageSize             = 16
	webhookDeliveryLease                = 30 * time.Second
	webhookDeliveryPoll                 = 5 * time.Second
	webhookDeliveryRetryMax             = time.Minute
	durableWebhookSnapshotVersion uint8 = 1
)

type webhookConfig struct {
	URL         string `json:"url"`
	ContentType string `json:"content_type"`
	Secret      string `json:"secret"`
	InsecureSSL string `json:"insecure_ssl"`
}

type webhookJob struct {
	db         *gorm.DB
	deliveryID uint
	url        string
	headers    http.Header
	body       []byte
	insecure   bool
	leaseToken string
	attempts   uint
}

// RepositoryEvent is the canonical, product-neutral event emitted to an
// embedding host. AGS does not interpret it as workflow or product state.
type RepositoryEvent struct {
	RepositoryID uint
	Event        string
	Action       string
	// OperationID and EventKey are stable for durable outbox replay. Sinks
	// provide at-least-once handling and use this pair for deduplication.
	OperationID string
	EventKey    string
	PayloadJSON []byte
}

type RepositoryEventSink interface {
	HandleRepositoryEvent(context.Context, RepositoryEvent) error
}

// CreateWebhook creates a new webhook for a repository.
func (s *Service) CreateWebhook(ctx context.Context, w *db.Webhook) error {
	return s.DBForCtx(ctx).Create(w).Error
}

// GetWebhook returns a webhook by ID for a repository.
func (s *Service) GetWebhook(ctx context.Context, repoID, hookID uint) (*db.Webhook, error) {
	var w db.Webhook
	err := s.DBForCtx(ctx).
		Where("repository_id = ? AND id = ?", repoID, hookID).
		First(&w).Error
	if err != nil {
		return nil, wrapErr(err)
	}
	return &w, nil
}

// UpdateWebhook updates an existing webhook.
func (s *Service) UpdateWebhook(ctx context.Context, w *db.Webhook) error {
	var existing db.Webhook
	if err := s.DBForCtx(ctx).
		Where("repository_id = ? AND id = ?", w.RepositoryID, w.ID).
		First(&existing).Error; err != nil {
		return wrapErr(err)
	}
	return s.DBForCtx(ctx).Save(w).Error
}

// DeleteWebhook deletes a webhook by ID for a repository.
func (s *Service) DeleteWebhook(ctx context.Context, repoID, hookID uint) error {
	return s.DBForCtx(ctx).
		Where("repository_id = ? AND id = ?", repoID, hookID).
		Delete(&db.Webhook{}).Error
}

// ListWebhooks returns all webhooks for a repository.
func (s *Service) ListWebhooks(ctx context.Context, repoID uint) ([]db.Webhook, error) {
	var hooks []db.Webhook
	err := s.DBForCtx(ctx).
		Where("repository_id = ?", repoID).
		Find(&hooks).Error
	return hooks, err
}

// DispatchWebhookEvent fans out an event to all matching active repository hooks.
func (s *Service) DispatchWebhookEvent(ctx context.Context, repoID uint, event, action string, payload any) error {
	return s.dispatchWebhookEvent(ctx, repoID, event, action, payload, "", "")
}

// DispatchWebhookEventIdempotent materializes durable push-recovery events
// once per operation/event/hook. Replays re-enqueue the existing pending row
// rather than creating a duplicate delivery record.
func (s *Service) DispatchWebhookEventIdempotent(ctx context.Context, repoID uint, event, action string, payload any, operationID, eventKey string) error {
	if operationID == "" || eventKey == "" || len(operationID) > 64 || len(eventKey) > 400 {
		return fmt.Errorf("webhook idempotency identity is invalid")
	}
	body, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("webhook payload marshal: %w", err)
	}
	return s.DispatchWebhookOperationIdempotent(ctx, operationID, []RepositoryEvent{{
		RepositoryID: repoID, Event: event, Action: action, EventKey: eventKey, PayloadJSON: body,
	}})
}

// DispatchWebhookOperationIdempotent freezes and materializes the complete
// durable event operation before any delivery can be claimed. The operation
// bound is evaluated against final encoded request snapshots, not raw event
// payloads, so refs x subscribers cannot amplify into partial fanout.
func (s *Service) DispatchWebhookOperationIdempotent(ctx context.Context, operationID string, events []RepositoryEvent) error {
	frozen, err := s.freezeDurableWebhookOperation(ctx, operationID, events)
	if err != nil {
		return err
	}
	var sinkErrs []error
	if s.RepositoryEvents != nil {
		for _, event := range frozen {
			if err := s.RepositoryEvents.HandleRepositoryEvent(ctx, event); err != nil {
				slog.WarnContext(ctx, "repository event sink failed", "repo_id", event.RepositoryID, "event", event.Event, "action", event.Action, "error", err)
				sinkErrs = append(sinkErrs, fmt.Errorf("repository event sink: %w", err))
			}
		}
	}
	materializationErr := s.materializeDurableWebhookOperation(ctx, operationID)
	if materializationErr == nil {
		s.kickWebhookDeliveryWorker()
	}
	return errors.Join(append(sinkErrs, materializationErr)...)
}

func (s *Service) dispatchWebhookEvent(ctx context.Context, repoID uint, event, action string, payload any, operationID, eventKey string) error {
	body, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("webhook payload marshal: %w", err)
	}
	durable := operationID != ""
	if durable {
		return s.DispatchWebhookOperationIdempotent(ctx, operationID, []RepositoryEvent{{
			RepositoryID: repoID, Event: event, Action: action, EventKey: eventKey, PayloadJSON: body,
		}})
	}
	if s.RepositoryEvents != nil {
		if err := s.RepositoryEvents.HandleRepositoryEvent(ctx, RepositoryEvent{
			RepositoryID: repoID, Event: event, Action: action, OperationID: operationID, EventKey: eventKey,
			PayloadJSON: append([]byte(nil), body...),
		}); err != nil {
			slog.WarnContext(ctx, "repository event sink failed", "repo_id", repoID, "event", event, "action", action, "error", err)
		}
	}

	var hooks []db.Webhook
	if err := s.DBForCtx(ctx).
		Where("repository_id = ? AND active = ?", repoID, true).
		Find(&hooks).Error; err != nil {
		return err
	}
	for _, hook := range hooks {
		if !webhookSubscribed(hook.EventsJSON, event) {
			continue
		}
		if err := s.dispatchWebhookAttempt(ctx, hook, repoID, event, action, body, nil, "", false); err != nil {
			slog.WarnContext(ctx, "webhook dispatch enqueue failed", "hook_id", hook.ID, "event", event, "error", err)
		}
	}
	return nil
}

// DispatchWebhookPing sends the initial ping delivery for a newly created hook.
func (s *Service) DispatchWebhookPing(ctx context.Context, hook db.Webhook, payload any) error {
	body, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("webhook ping payload marshal: %w", err)
	}
	return s.dispatchWebhookAttempt(ctx, hook, hook.RepositoryID, "ping", "", body, nil, "", false)
}

// ListHookDeliveries returns all deliveries for a repository webhook.
func (s *Service) ListHookDeliveries(ctx context.Context, repoID, hookID uint) ([]db.HookDelivery, error) {
	var deliveries []db.HookDelivery
	err := s.DBForCtx(ctx).
		Where("repository_id = ? AND webhook_id = ?", repoID, hookID).
		Order("id desc").
		Find(&deliveries).Error
	return deliveries, err
}

// GetHookDelivery returns a single persisted delivery attempt.
func (s *Service) GetHookDelivery(ctx context.Context, repoID, hookID, deliveryID uint) (*db.HookDelivery, error) {
	var delivery db.HookDelivery
	err := s.DBForCtx(ctx).
		Where("repository_id = ? AND webhook_id = ? AND id = ?", repoID, hookID, deliveryID).
		First(&delivery).Error
	if err != nil {
		return nil, wrapErr(err)
	}
	return &delivery, nil
}

// RedeliverHookDelivery creates a new delivery attempt using the original payload.
func (s *Service) RedeliverHookDelivery(ctx context.Context, repoID, hookID, deliveryID uint) (*db.HookDelivery, error) {
	hook, err := s.GetWebhook(ctx, repoID, hookID)
	if err != nil {
		return nil, err
	}
	original, err := s.GetHookDelivery(ctx, repoID, hookID, deliveryID)
	if err != nil {
		return nil, err
	}
	if len(original.RequestPayload) > webhookRequestBodyMax {
		return nil, fmt.Errorf("webhook request body exceeds limit")
	}

	headers, err := unmarshalWebhookHeaders(original.RequestHeaders)
	if err != nil {
		return nil, fmt.Errorf("webhook redelivery request headers: %w", err)
	}

	cfg, err := parseWebhookConfig(hook.ConfigJSON)
	if err != nil {
		delivery := db.HookDelivery{
			WebhookID:        hook.ID,
			RepositoryID:     repoID,
			ParentDeliveryID: &original.ID,
			GUID:             original.GUID,
			Event:            original.Event,
			Action:           original.Action,
			Status:           webhookStatusError,
			Redelivery:       true,
			RequestHeaders:   original.RequestHeaders,
			RequestPayload:   original.RequestPayload,
			LastError:        err.Error(),
		}
		now := time.Now()
		delivery.DeliveredAt = &now
		if createErr := s.DBForCtx(ctx).Create(&delivery).Error; createErr != nil {
			return nil, createErr
		}
		return &delivery, nil
	}

	delivery := db.HookDelivery{
		WebhookID:        hook.ID,
		RepositoryID:     repoID,
		ParentDeliveryID: &original.ID,
		GUID:             original.GUID,
		Event:            original.Event,
		Action:           original.Action,
		Status:           webhookStatusPending,
		Redelivery:       true,
		RequestURL:       cfg.URL,
		RequestInsecure:  webhookConfigAllowsInsecureSSL(cfg.InsecureSSL),
		RequestHeaders:   original.RequestHeaders,
		RequestPayload:   original.RequestPayload,
		NextAttemptAt:    time.Now().UTC(),
	}
	directLeaseToken, err := newWebhookGUID()
	if err != nil {
		return nil, err
	}
	directLeaseUntil := time.Now().UTC().Add(webhookDeliveryLease)
	delivery.LeaseToken = directLeaseToken
	delivery.LeaseUntil = &directLeaseUntil
	deliveryDB := s.DBForCtx(ctx)
	if err := deliveryDB.Create(&delivery).Error; err != nil {
		return nil, err
	}

	if err := s.enqueueWebhookJob(ctx, webhookJob{
		db: deliveryDB, deliveryID: delivery.ID, url: cfg.URL, headers: headers,
		body: []byte(original.RequestPayload), insecure: webhookConfigAllowsInsecureSSL(cfg.InsecureSSL), leaseToken: directLeaseToken,
	}); err != nil {
		return nil, err
	}
	return &delivery, nil
}

func (s *Service) dispatchWebhookAttempt(
	ctx context.Context,
	hook db.Webhook,
	repoID uint,
	event, action string,
	payload []byte,
	parent *db.HookDelivery,
	guid string,
	redelivery bool,
) error {
	deliveryDB := s.DBForCtx(ctx)
	if guid == "" {
		var err error
		guid, err = newWebhookGUID()
		if err != nil {
			return err
		}
	}

	cfg, cfgErr := parseWebhookConfig(hook.ConfigJSON)
	headers, requestBody := buildWebhookRequest(hook.ID, event, guid, payload, cfg.Secret, cfg.ContentType)
	if len(requestBody) > webhookRequestBodyMax {
		return fmt.Errorf("webhook request body exceeds limit")
	}
	now := time.Now().UTC()

	delivery := db.HookDelivery{
		WebhookID:       hook.ID,
		RepositoryID:    repoID,
		GUID:            guid,
		Event:           event,
		Action:          action,
		Status:          webhookStatusPending,
		Redelivery:      redelivery,
		RequestURL:      cfg.URL,
		RequestInsecure: webhookConfigAllowsInsecureSSL(cfg.InsecureSSL),
		RequestHeaders:  marshalWebhookHeaders(headers),
		RequestPayload:  db.LargeText(requestBody),
		NextAttemptAt:   now,
	}
	if parent != nil {
		delivery.ParentDeliveryID = &parent.ID
	}
	if cfgErr != nil {
		delivery.Status = webhookStatusError
		delivery.LastError = cfgErr.Error()
		now := time.Now()
		delivery.DeliveredAt = &now
	}
	directLeaseToken := ""
	if cfgErr == nil {
		var err error
		directLeaseToken, err = newWebhookGUID()
		if err != nil {
			return err
		}
		directLeaseUntil := now.Add(webhookDeliveryLease)
		delivery.LeaseToken = directLeaseToken
		delivery.LeaseUntil = &directLeaseUntil
	}

	if err := deliveryDB.Create(&delivery).Error; err != nil {
		return err
	}
	if cfgErr != nil {
		return nil
	}
	return s.enqueueWebhookJob(ctx, webhookJob{
		db: deliveryDB, deliveryID: delivery.ID, url: cfg.URL, headers: headers,
		body: requestBody, insecure: webhookConfigAllowsInsecureSSL(cfg.InsecureSSL), leaseToken: directLeaseToken,
	})
}

// enqueueWebhookJob preserves the ordinary request-scoped live-delivery
// contract. A scoped DB is not a cross-process durable queue: after restart
// only the embedder that owns that DB can decide how to reconcile its audit
// rows. Canonical durable push replay never uses this path; it is claimed from
// the private construction-bound database by kickWebhookDeliveryWorker.
func (s *Service) enqueueWebhookJob(ctx context.Context, job webhookJob) error {
	if job.db == nil || job.deliveryID == 0 {
		return fmt.Errorf("webhook delivery job is invalid")
	}
	if s.Ctx == nil {
		s.processWebhookJob(context.Background(), job)
		return nil
	}
	s.ensureWebhookWorkers()
	s.Wg.Add(1)
	go func() {
		defer s.Wg.Done()
		select {
		case s.webhookJobs <- job:
		case <-s.ServerCtx().Done():
			s.completeWebhookResult(job.db, job.deliveryID, webhookStatusError, 0, time.Now(), nil,
				"server shutting down", "server shutting down", job.leaseToken)
		case <-time.After(webhookQueueTimeout):
			s.completeWebhookResult(job.db, job.deliveryID, webhookStatusError, 0, time.Now(), nil,
				"webhook queue timeout", "webhook queue timeout", job.leaseToken)
		}
	}()
	return nil
}

func (s *Service) kickWebhookDeliveryWorker() {
	if s == nil || s.durableWebhookDB == nil {
		return
	}
	if s.Ctx == nil {
		s.webhookFallbackMu.Lock()
		if s.webhookFallbackRunning {
			s.webhookFallbackPending = true
			s.webhookFallbackMu.Unlock()
			return
		}
		if s.webhookFallbackTimer != nil {
			s.webhookFallbackTimer.Stop()
			s.webhookFallbackTimer = nil
			s.webhookFallbackDue = time.Time{}
		}
		s.webhookFallbackRunning = true
		s.webhookFallbackMu.Unlock()
		s.Wg.Add(1)
		go s.runWebhookDeliveryFallback()
		return
	}
	s.ensureWebhookWorkers()
	select {
	case s.webhookDeliveryKick <- struct{}{}:
	default:
	}
}

func (s *Service) runWebhookDeliveryFallback() {
	defer s.Wg.Done()
	for {
		for {
			attempted, err := s.processPendingWebhookDeliveryPage(context.Background())
			if err != nil {
				slog.Error("recover pending webhook deliveries", "error", err)
			}
			if attempted < webhookDeliveryPageSize {
				break
			}
		}
		s.webhookFallbackMu.Lock()
		if s.webhookFallbackPending {
			s.webhookFallbackPending = false
			s.webhookFallbackMu.Unlock()
			continue
		}
		s.webhookFallbackRunning = false
		s.webhookFallbackMu.Unlock()
		s.scheduleWebhookDeliveryFallback()
		return
	}
}

func (s *Service) scheduleWebhookDeliveryFallback() {
	if s == nil || s.durableWebhookDB == nil || s.Ctx != nil {
		return
	}
	var next struct {
		NextAttemptAt time.Time
		LeaseUntil    *time.Time
	}
	result := s.durableWebhookDB.WithContext(context.Background()).Model(&db.HookDelivery{}).
		Select("next_attempt_at", "lease_until").
		Scopes(durablePendingWebhookScope).
		Order("next_attempt_at ASC, id ASC").Limit(1).Scan(&next)
	if result.Error != nil {
		slog.Error("inspect pending webhook delivery wake", "error", result.Error)
	}
	if result.Error == nil && result.RowsAffected == 0 {
		s.webhookFallbackMu.Lock()
		if s.webhookFallbackTimer != nil {
			s.webhookFallbackTimer.Stop()
			s.webhookFallbackTimer = nil
			s.webhookFallbackDue = time.Time{}
		}
		s.webhookFallbackMu.Unlock()
		return
	}
	due := time.Now().Add(webhookDeliveryPoll)
	if result.Error == nil {
		due = next.NextAttemptAt
		if next.LeaseUntil != nil && next.LeaseUntil.After(due) {
			due = *next.LeaseUntil
		}
	}
	delay := time.Until(due)
	if delay < time.Second {
		delay = time.Second
	}
	if delay > webhookDeliveryPoll {
		delay = webhookDeliveryPoll
	}

	s.webhookFallbackMu.Lock()
	defer s.webhookFallbackMu.Unlock()
	if s.webhookFallbackTimer != nil && !s.webhookFallbackDue.After(due) {
		return
	}
	if s.webhookFallbackTimer != nil {
		s.webhookFallbackTimer.Stop()
	}
	var timer *time.Timer
	timer = time.AfterFunc(delay, func() {
		s.webhookFallbackMu.Lock()
		if s.webhookFallbackTimer != timer {
			s.webhookFallbackMu.Unlock()
			return
		}
		s.webhookFallbackTimer = nil
		s.webhookFallbackDue = time.Time{}
		s.webhookFallbackMu.Unlock()
		s.kickWebhookDeliveryWorker()
	})
	s.webhookFallbackTimer = timer
	s.webhookFallbackDue = due
}

func (s *Service) processPendingWebhookDeliveryPage(ctx context.Context) (int, error) {
	jobs, attempted, claimErr := s.claimPendingWebhookDeliveries(ctx, webhookDeliveryPageSize)
	var enqueueErrs []error
	for _, job := range jobs {
		if s.Ctx == nil {
			s.processWebhookJob(context.Background(), job)
			continue
		}
		select {
		case s.webhookJobs <- job:
		case <-s.ServerCtx().Done():
			err := s.ServerCtx().Err()
			enqueueErrs = append(enqueueErrs, err)
			if retryErr := s.retryPendingWebhookDelivery(context.Background(), job, err); retryErr != nil {
				enqueueErrs = append(enqueueErrs, retryErr)
			}
		case <-time.After(webhookQueueTimeout):
			err := fmt.Errorf("webhook queue timeout")
			enqueueErrs = append(enqueueErrs, err)
			if retryErr := s.retryPendingWebhookDelivery(context.Background(), job, err); retryErr != nil {
				enqueueErrs = append(enqueueErrs, retryErr)
			}
		}
	}
	return attempted, errors.Join(claimErr, errors.Join(enqueueErrs...))
}

func (s *Service) claimPendingWebhookDeliveries(ctx context.Context, limit int) ([]webhookJob, int, error) {
	if limit <= 0 || limit > webhookDeliveryPageSize {
		limit = webhookDeliveryPageSize
	}
	leaseToken, err := newWebhookGUID()
	if err != nil {
		return nil, 0, err
	}
	now := time.Now().UTC()
	leaseUntil := now.Add(webhookDeliveryLease)
	var rows []db.HookDelivery
	err = s.durableWebhookDB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		var ids []uint
		if err := tx.Model(&db.HookDelivery{}).Select("id").Scopes(durablePendingWebhookScope).
			Where("next_attempt_at <= ? AND (lease_until IS NULL OR lease_until < ?)", now, now).
			Order("next_attempt_at ASC, id ASC").Limit(limit).Pluck("id", &ids).Error; err != nil || len(ids) == 0 {
			return err
		}
		if err := tx.Model(&db.HookDelivery{}).Scopes(durablePendingWebhookScope).
			Where("id IN ? AND (lease_until IS NULL OR lease_until < ?)", ids, now).
			Updates(map[string]any{"lease_until": leaseUntil, "lease_token": leaseToken}).Error; err != nil {
			return err
		}
		return tx.Where("lease_token = ?", leaseToken).Order("id ASC").Limit(limit).Find(&rows).Error
	})
	if err != nil {
		return nil, 0, err
	}
	jobs := make([]webhookJob, 0, len(rows))
	var rowErrs []error
	for _, row := range rows {
		headers, headerErr := unmarshalWebhookHeaders(row.RequestHeaders)
		requestURL, urlErr := url.ParseRequestURI(row.RequestURL)
		var rowErr error
		switch {
		case row.GUID == "" || row.RequestURL == "" || requestURL == nil || urlErr != nil:
			rowErr = fmt.Errorf("persisted webhook target is invalid")
		case headerErr != nil:
			rowErr = fmt.Errorf("persisted webhook headers: %w", headerErr)
		case len(row.RequestPayload) > webhookRequestBodyMax:
			rowErr = fmt.Errorf("persisted webhook request body exceeds limit")
		}
		job := webhookJob{db: s.durableWebhookDB, deliveryID: row.ID, url: row.RequestURL, headers: headers,
			body: []byte(row.RequestPayload), insecure: row.RequestInsecure, leaseToken: leaseToken, attempts: row.Attempts}
		if rowErr != nil {
			if quarantineErr := s.quarantinePendingWebhookDelivery(ctx, job, rowErr); quarantineErr != nil {
				rowErrs = append(rowErrs, errors.Join(rowErr, quarantineErr))
			} else {
				rowErrs = append(rowErrs, rowErr)
			}
			continue
		}
		jobs = append(jobs, job)
	}
	return jobs, len(rows), errors.Join(rowErrs...)
}

func durablePendingWebhookScope(query *gorm.DB) *gorm.DB {
	return query.Where("snapshot_version = ? AND status = ?", durableWebhookSnapshotVersion, webhookStatusPending).
		Where("operation_id IS NOT NULL AND operation_id <> '' AND operation_event_key IS NOT NULL AND operation_event_key <> ''").
		Where("request_url <> '' AND request_headers <> '' AND next_attempt_at IS NOT NULL")
}

func (s *Service) quarantinePendingWebhookDelivery(ctx context.Context, job webhookJob, cause error) error {
	if job.deliveryID == 0 || job.leaseToken == "" {
		return fmt.Errorf("webhook delivery lease is invalid")
	}
	message := "persisted webhook request is not recoverable"
	if cause != nil {
		message = cause.Error()
	}
	if len(message) > 1024 {
		message = message[:1024]
	}
	now := time.Now().UTC()
	result := s.durableWebhookDB.WithContext(ctx).Model(&db.HookDelivery{}).
		Where("id = ? AND status = ? AND lease_token = ?", job.deliveryID, webhookStatusPending, job.leaseToken).
		Updates(map[string]any{"status": webhookStatusError, "delivered_at": &now, "lease_until": nil, "lease_token": "", "last_error": message})
	if result.Error != nil {
		return result.Error
	}
	if result.RowsAffected != 1 {
		return fmt.Errorf("webhook delivery lease changed")
	}
	return nil
}

func (s *Service) retryPendingWebhookDelivery(ctx context.Context, job webhookJob, cause error) error {
	if job.deliveryID == 0 || job.leaseToken == "" {
		return fmt.Errorf("webhook delivery lease is invalid")
	}
	delay := time.Second << min(job.attempts, 6)
	if delay > webhookDeliveryRetryMax {
		delay = webhookDeliveryRetryMax
	}
	message := "webhook delivery retry"
	if cause != nil {
		message = cause.Error()
	}
	if len(message) > 1024 {
		message = message[:1024]
	}
	result := s.durableWebhookDB.WithContext(ctx).Model(&db.HookDelivery{}).
		Where("id = ? AND status = ? AND lease_token = ?", job.deliveryID, webhookStatusPending, job.leaseToken).
		Updates(map[string]any{"attempts": gorm.Expr("attempts + 1"), "next_attempt_at": time.Now().UTC().Add(delay), "lease_until": nil, "lease_token": "", "last_error": message})
	if result.Error != nil {
		return result.Error
	}
	if result.RowsAffected != 1 {
		return fmt.Errorf("webhook delivery lease changed")
	}
	return nil
}

func (s *Service) ensureWebhookWorkers() {
	s.webhookWorkersOnce.Do(func() {
		s.webhookJobs = make(chan webhookJob, webhookWorkerQueueSize)
		s.webhookDeliveryKick = make(chan struct{}, 1)
		for i := 0; i < webhookWorkerCount; i++ {
			s.Wg.Add(1)
			go func() {
				defer s.Wg.Done()
				for {
					select {
					case <-s.ServerCtx().Done():
						return
					case job := <-s.webhookJobs:
						s.processWebhookJob(s.ServerCtx(), job)
					}
				}
			}()
		}
		if s.Ctx != nil {
			s.Wg.Add(1)
			go func() {
				defer s.Wg.Done()
				ticker := time.NewTicker(webhookDeliveryPoll)
				defer ticker.Stop()
				for {
					select {
					case <-s.ServerCtx().Done():
						return
					case <-s.webhookDeliveryKick:
					case <-ticker.C:
					}
					for {
						attempted, err := s.processPendingWebhookDeliveryPage(s.ServerCtx())
						if err != nil {
							slog.ErrorContext(s.ServerCtx(), "recover pending webhook deliveries", "error", err)
						}
						if attempted < webhookDeliveryPageSize {
							break
						}
					}
				}
			}()
			s.webhookDeliveryKick <- struct{}{}
		}
	})
}

func (s *Service) processWebhookJob(ctx context.Context, job webhookJob) {
	started := time.Now()
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, job.url, bytes.NewReader(job.body))
	if err != nil {
		s.completeWebhookResult(job.db, job.deliveryID, webhookStatusError, 0, started, nil, err.Error(), err.Error(), job.leaseToken)
		return
	}
	for key, values := range job.headers {
		for _, value := range values {
			req.Header.Add(key, value)
		}
	}

	client := webhookHTTPClient(job.insecure)
	resp, err := client.Do(req)
	if err != nil {
		s.completeWebhookResult(job.db, job.deliveryID, webhookStatusError, 0, started, nil, err.Error(), err.Error(), job.leaseToken)
		return
	}
	defer resp.Body.Close()

	responseBody := readWebhookBody(resp.Body, webhookResponseBodyLimit)
	status := webhookStatusOK
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		status = webhookStatusFailed
	}
	s.completeWebhookResult(
		job.db,
		job.deliveryID,
		status,
		resp.StatusCode,
		started,
		resp.Header,
		string(responseBody),
		"",
		job.leaseToken,
	)
}

func webhookHTTPClient(insecure bool) *http.Client {
	client := &http.Client{Timeout: webhookRequestTimeout}
	if !insecure {
		return client
	}
	baseTransport, ok := http.DefaultTransport.(*http.Transport)
	if !ok {
		return client
	}
	transport := baseTransport.Clone()
	if transport.TLSClientConfig == nil {
		transport.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12}
	}
	transport.TLSClientConfig.InsecureSkipVerify = true
	client.Transport = transport
	return client
}

func (s *Service) completeWebhookResult(
	database *gorm.DB,
	deliveryID uint,
	status string,
	statusCode int,
	started time.Time,
	headers http.Header,
	payload string,
	lastError string,
	leaseToken string,
) {
	now := time.Now()
	updates := map[string]any{
		"status":           status,
		"status_code":      statusCode,
		"duration_millis":  now.Sub(started).Milliseconds(),
		"delivered_at":     &now,
		"response_headers": marshalWebhookHeaders(headers),
		"response_payload": db.LargeText(payload),
		"last_error":       lastError,
		"lease_until":      nil,
		"lease_token":      "",
	}
	query := database.WithContext(context.Background()).Model(&db.HookDelivery{}).Where("id = ?", deliveryID)
	if leaseToken != "" {
		query = query.Where("status = ? AND lease_token = ?", webhookStatusPending, leaseToken)
	}
	if err := query.Updates(updates).Error; err != nil {
		slog.Error("webhook delivery update failed", "delivery_id", deliveryID, "error", err)
	}
}

func parseWebhookConfig(raw string) (webhookConfig, error) {
	var cfg webhookConfig
	if raw == "" {
		return cfg, fmt.Errorf("webhook config missing")
	}
	if err := json.Unmarshal([]byte(raw), &cfg); err != nil {
		return cfg, fmt.Errorf("webhook config decode: %w", err)
	}
	cfg.URL = strings.TrimSpace(cfg.URL)
	if cfg.URL == "" {
		return cfg, fmt.Errorf("webhook config missing url")
	}
	if _, err := url.ParseRequestURI(cfg.URL); err != nil {
		return cfg, fmt.Errorf("webhook config invalid url: %w", err)
	}
	return cfg, nil
}

func webhookConfigAllowsInsecureSSL(raw string) bool {
	switch strings.TrimSpace(strings.ToLower(raw)) {
	case "1", "true":
		return true
	default:
		return false
	}
}

func webhookSubscribed(eventsJSON, event string) bool {
	if event == "ping" {
		return true
	}
	var events []string
	if err := json.Unmarshal([]byte(eventsJSON), &events); err != nil {
		return false
	}
	for _, candidate := range events {
		candidate = strings.TrimSpace(strings.ToLower(candidate))
		if candidate == "*" || candidate == strings.ToLower(event) {
			return true
		}
	}
	return false
}

func buildWebhookRequest(hookID uint, event, guid string, payload []byte, secret, contentType string) (http.Header, []byte) {
	requestBody := payload
	headers := make(http.Header)
	headers.Set("User-Agent", fmt.Sprintf("GitHub-Hookshot/%d", hookID))
	headers.Set("X-GitHub-Event", event)
	headers.Set("X-GitHub-Delivery", guid)

	if strings.EqualFold(strings.TrimSpace(contentType), "form") {
		form := url.Values{}
		form.Set("payload", string(payload))
		requestBody = []byte(form.Encode())
		headers.Set("Content-Type", "application/x-www-form-urlencoded")
	} else {
		headers.Set("Content-Type", "application/json")
	}

	if secret != "" {
		mac256 := hmac.New(sha256.New, []byte(secret))
		_, _ = mac256.Write(requestBody)
		headers.Set("X-Hub-Signature-256", "sha256="+hex.EncodeToString(mac256.Sum(nil)))

		mac1 := hmac.New(sha1.New, []byte(secret))
		_, _ = mac1.Write(requestBody)
		headers.Set("X-Hub-Signature", "sha1="+hex.EncodeToString(mac1.Sum(nil)))
	}

	return headers, requestBody
}

func marshalWebhookHeaders(headers http.Header) string {
	if headers == nil {
		return "{}"
	}
	data, err := json.Marshal(headers)
	if err != nil {
		return "{}"
	}
	return string(data)
}

func unmarshalWebhookHeaders(encoded string) (http.Header, error) {
	headers := make(http.Header)
	if encoded == "" {
		return headers, nil
	}
	if err := json.Unmarshal([]byte(encoded), &headers); err != nil {
		return nil, err
	}
	return headers, nil
}

func readWebhookBody(r io.Reader, limit int64) []byte {
	if r == nil {
		return nil
	}
	body, err := io.ReadAll(io.LimitReader(r, limit+1))
	if err != nil {
		return []byte(err.Error())
	}
	if int64(len(body)) > limit {
		return body[:limit]
	}
	return body
}

func newWebhookGUID() (string, error) {
	var raw [16]byte
	if _, err := rand.Read(raw[:]); err != nil {
		return "", fmt.Errorf("webhook guid: %w", err)
	}
	raw[6] = (raw[6] & 0x0f) | 0x40
	raw[8] = (raw[8] & 0x3f) | 0x80
	return fmt.Sprintf(
		"%08x-%04x-%04x-%04x-%012x",
		raw[0:4],
		raw[4:6],
		raw[6:8],
		raw[8:10],
		raw[10:16],
	), nil
}