// Package webhook delivers repository event notifications to user-configured
// HTTP endpoints. It is the spindle-side successor to the appview webhook
// notifier: spindle owns webhook configuration and delivery, firing from the
// repo events it already ingests (push, pull_request:*, repository:renamed).
package webhook

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

	"github.com/avast/retry-go/v4"
	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/google/uuid"
	"tangled.org/core/hostutil"
	"tangled.org/core/log"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/models"
)

const (
	deliveryTimeout   = 10 * time.Second
	deliveryRetention = 50
)

// a dead host can only hog perHost workers and its own queue
type deliveryLimits struct {
	workers      int
	perHost      int
	perHostQueue int
	queue        int
}

var defaultLimits = deliveryLimits{workers: 8, perHost: 2, perHostQueue: 64, queue: 1024}

var errQueueSaturated = errors.New("delivery queue saturated")

type hostQueue struct {
	host    string
	pending []deliveryJob
	running int
}

// Redeliver returns this for a paused owner
var ErrPaused = errors.New("the repo's owner is paused")

type Service struct {
	db     *db.DB
	logger *slog.Logger
	client *http.Client
	limits deliveryLimits
	send   func(deliveryJob)
	paused func(syntax.DID) (bool, error)
	wg     sync.WaitGroup

	mu      sync.Mutex
	hosts   map[string]*hostQueue
	pending int
	running int
	closed  bool
}

type Option func(*Service)

func WithGate(paused func(syntax.DID) (bool, error)) Option {
	return func(s *Service) { s.paused = paused }
}

func New(database *db.DB, dev bool, opts ...Option) *Service {
	s := &Service{
		db:     database,
		logger: log.New("webhook"),
		// user-supplied webhook URLs are untrusted: block internal address
		// ranges and don't follow redirects to guard against SSRF.
		client: hostutil.SafeClient(dev, deliveryTimeout),
		limits: defaultLimits,
	}
	s.send = s.sendWebhook
	for _, opt := range opts {
		opt(s)
	}
	return s
}

// FirePush delivers a push (git ref-update) event to subscribed webhooks.
func (s *Service) FirePush(ctx context.Context, repo *db.Repo, committerDid, ref, oldSha, newSha string) {
	webhooks := s.activeWebhooksForEvent(repo.RepoDid, models.WebhookEventPush)
	if len(webhooks) == 0 {
		return
	}

	pusher := committerDid
	if pusher == "" {
		pusher = repo.Owner.String()
	}
	payload := &models.WebhookPayload{
		Ref:        ref,
		Before:     oldSha,
		After:      newSha,
		Repository: buildRepository(repo),
		Pusher:     models.WebhookUser{Did: pusher},
	}
	payloadBytes, err := json.Marshal(payload)
	if err != nil {
		s.logger.Error("failed to marshal push payload", "repo_did", repo.RepoDid, "err", err)
		return
	}

	userAgent := "Tangled-Hook/push"
	if len(newSha) >= 7 {
		userAgent = "Tangled-Hook/" + newSha[:7]
	}
	for _, webhook := range webhooks {
		s.dispatch(deliveryJob{ctx: ctx, webhook: webhook, event: string(models.WebhookEventPush), repoFullName: payload.Repository.FullName, userAgent: userAgent, payload: payloadBytes})
	}
}

// FireRename delivers a repository:renamed event to subscribed webhooks.
func (s *Service) FireRename(ctx context.Context, actor syntax.DID, repo *db.Repo, oldName, newName string) {
	webhooks := s.activeWebhooksForEvent(repo.RepoDid, models.WebhookEventRepoRenamed)
	if len(webhooks) == 0 {
		return
	}

	payload := &models.WebhookRenamePayload{
		OldName:    oldName,
		NewName:    newName,
		Repository: buildRepository(repo),
		Sender:     models.WebhookUser{Did: actor.String()},
	}
	payloadBytes, err := json.Marshal(payload)
	if err != nil {
		s.logger.Error("failed to marshal rename payload", "repo_did", repo.RepoDid, "err", err)
		return
	}

	for _, webhook := range webhooks {
		s.dispatch(deliveryJob{ctx: ctx, webhook: webhook, event: string(models.WebhookEventRepoRenamed), repoFullName: payload.Repository.FullName, userAgent: "Tangled-Hook/rename", payload: payloadBytes})
	}
}

// FirePullRequest delivers a pull_request:* event to subscribed webhooks. repo is
// the target repository, sender is who caused this transition.
func (s *Service) FirePullRequest(ctx context.Context, event models.WebhookEvent, repo *db.Repo, pr models.WebhookPullRequest, sender syntax.DID) {
	webhooks := s.activeWebhooksForEvent(repo.RepoDid, event)
	if len(webhooks) == 0 {
		return
	}

	payload := &models.WebhookPullRequestPayload{
		Action:      strings.TrimPrefix(string(event), "pull_request:"),
		PullRequest: pr,
		Repository:  buildRepository(repo),
		Sender:      models.WebhookUser{Did: sender.String()},
	}
	payloadBytes, err := json.Marshal(payload)
	if err != nil {
		s.logger.Error("failed to marshal pull request payload", "repo_did", repo.RepoDid, "err", err)
		return
	}

	for _, webhook := range webhooks {
		s.dispatch(deliveryJob{ctx: ctx, webhook: webhook, event: string(event), repoFullName: payload.Repository.FullName, userAgent: "Tangled-Hook/pull_request", payload: payloadBytes})
	}
}

// Redeliver re-sends a stored delivery via the live send-and-record path,
// signing with the webhook's current secret.
func (s *Service) Redeliver(ctx context.Context, webhook models.Webhook, prev models.WebhookDelivery) error {
	if paused, err := s.isPaused(webhook.RepoDid); err != nil {
		return err
	} else if paused {
		return ErrPaused
	}
	// recover the repo full name (X-Tangled-Repo header) from the stored payload
	var meta struct {
		Repository struct {
			FullName string `json:"full_name"`
		} `json:"repository"`
	}
	_ = json.Unmarshal([]byte(prev.RequestBody), &meta)

	s.dispatch(deliveryJob{ctx: ctx, webhook: webhook, event: prev.Event, repoFullName: meta.Repository.FullName, userAgent: "Tangled-Hook/retry", payload: []byte(prev.RequestBody)})
	return nil
}

func (s *Service) isPaused(repoDid syntax.DID) (bool, error) {
	if s.paused == nil {
		return false, nil
	}
	return s.paused(repoDid)
}

func (s *Service) activeWebhooksForEvent(repoDid syntax.DID, event models.WebhookEvent) []models.Webhook {
	webhooks, err := s.db.GetActiveWebhooksForRepo(repoDid)
	if err != nil {
		s.logger.Error("failed to get webhooks for repo", "repo_did", repoDid, "err", err)
		return nil
	}
	var matching []models.Webhook
	for _, webhook := range webhooks {
		if webhook.HasEvent(event) {
			matching = append(matching, webhook)
		}
	}
	if len(matching) == 0 {
		return nil
	}
	if paused, err := s.isPaused(repoDid); err != nil {
		s.logger.Error("failed to check whether the repo is paused", "repo_did", repoDid, "err", err)
		return nil
	} else if paused {
		return nil
	}
	return matching
}

func buildRepository(repo *db.Repo) models.WebhookRepository {
	owner := repo.Owner.String()
	rkey := repo.Rkey.String()
	return models.WebhookRepository{
		Name:      repo.Name,
		FullName:  fmt.Sprintf("%s/%s", owner, rkey),
		HtmlUrl:   fmt.Sprintf("https://%s/%s/%s", repo.Knot, owner, rkey),
		CloneUrl:  fmt.Sprintf("https://%s/%s/%s", repo.Knot, owner, rkey),
		SshUrl:    fmt.Sprintf("ssh://git@%s/%s/%s", repo.Knot, owner, rkey),
		CreatedAt: repo.CreatedAt,
		UpdatedAt: repo.CreatedAt,
		Owner:     models.WebhookUser{Did: owner},
	}
}

type deliveryJob struct {
	ctx          context.Context
	webhook      models.Webhook
	event        string
	repoFullName string
	userAgent    string
	payload      []byte
}

func hostOf(raw string) string {
	u, err := url.Parse(raw)
	if err != nil {
		return ""
	}
	return strings.TrimSuffix(strings.ToLower(u.Hostname()), ".")
}

func (s *Service) dispatch(job deliveryJob) {
	// the caller's ctx dies on return, but a queued delivery still has to go out
	job.ctx = context.WithoutCancel(job.ctx)
	host := hostOf(job.webhook.Url)
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		s.logger.Warn("webhook delivery dropped, the service is closed", "webhook_id", job.webhook.Id)
		return
	}
	q := s.hosts[host]
	if s.pending >= s.limits.queue || (q != nil && len(q.pending) >= s.limits.perHostQueue) {
		s.mu.Unlock()
		s.logger.Warn("webhook delivery queue saturated", "webhook_id", job.webhook.Id, "url", job.webhook.Url)
		s.record(&models.WebhookDelivery{
			WebhookId:    job.webhook.Id,
			Event:        job.event,
			DeliveryId:   uuid.New().String(),
			Url:          job.webhook.Url,
			RequestBody:  string(job.payload),
			ResponseBody: errQueueSaturated.Error(),
		})
		return
	}
	if q == nil {
		if s.hosts == nil {
			s.hosts = map[string]*hostQueue{}
		}
		q = &hostQueue{host: host}
		s.hosts[host] = q
	}
	q.pending = append(q.pending, job)
	s.pending++
	s.startReady()
	s.mu.Unlock()
}

// caller holds mu
func (s *Service) startReady() {
	for s.running < s.limits.workers && s.pending > 0 {
		var ready *hostQueue
		for _, q := range s.hosts {
			if len(q.pending) > 0 && q.running < s.limits.perHost {
				ready = q
				break
			}
		}
		if ready == nil {
			return
		}
		job := ready.pending[0]
		ready.pending = ready.pending[1:]
		if len(ready.pending) == 0 {
			ready.pending = nil
		}
		ready.running++
		s.pending--
		s.running++
		s.wg.Add(1)
		go s.deliver(ready, job)
	}
}

func (s *Service) deliver(q *hostQueue, job deliveryJob) {
	defer s.wg.Done()
	s.send(job)

	s.mu.Lock()
	defer s.mu.Unlock()
	q.running--
	s.running--
	if q.running == 0 && len(q.pending) == 0 {
		delete(s.hosts, q.host)
	}
	s.startReady()
}

// close waits out the queue, pending deliveries still go out
func (s *Service) Close(ctx context.Context) {
	s.mu.Lock()
	s.closed = true
	s.mu.Unlock()
	done := make(chan struct{})
	go func() {
		s.wg.Wait()
		close(done)
	}()
	select {
	case <-done:
	case <-ctx.Done():
		s.logger.Warn("webhook deliveries still in flight at shutdown")
	}
}

func (s *Service) sendWebhook(job deliveryJob) {
	ctx := job.ctx

	var signature string
	if job.webhook.Secret != "" {
		signature = "sha256=" + s.computeSignature(job.payload, job.webhook.Secret)
	}

	deliveryId := uuid.New().String()

	retryOpts := []retry.Option{
		retry.Attempts(3),
		retry.Delay(1 * time.Second),
		retry.MaxDelay(10 * time.Second),
		retry.DelayType(retry.BackOffDelay),
		retry.LastErrorOnly(true),
		retry.OnRetry(func(n uint, err error) {
			s.logger.Info("retrying webhook delivery", "webhook_id", job.webhook.Id, "attempt", n+1, "err", err)
		}),
		retry.Context(ctx),
	}

	var (
		resp         *http.Response
		lastStatus   int
		lastResponse string
	)
	err := retry.Do(func() error {
		// build a fresh request each attempt so the body reader is not
		// exhausted after the first try
		req, err := http.NewRequestWithContext(ctx, "POST", job.webhook.Url, bytes.NewReader(job.payload))
		if err != nil {
			return retry.Unrecoverable(err)
		}
		req.Header.Set("Content-Type", "application/json")
		req.Header.Set("User-Agent", job.userAgent)
		req.Header.Set("X-Tangled-Event", job.event)
		req.Header.Set("X-Tangled-Hook-ID", fmt.Sprintf("%d", job.webhook.Id))
		req.Header.Set("X-Tangled-Delivery", deliveryId)
		req.Header.Set("X-Tangled-Repo", job.repoFullName)
		if signature != "" {
			req.Header.Set("X-Tangled-Signature-256", signature)
		}

		r, err := s.client.Do(req)
		if err != nil {
			return err
		}
		lastStatus = r.StatusCode
		body, readErr := io.ReadAll(io.LimitReader(r.Body, 10*1024))
		r.Body.Close()
		if readErr != nil {
			s.logger.Warn("failed to read webhook response body", "webhook_id", job.webhook.Id, "err", readErr)
		}
		lastResponse = string(body)
		if r.StatusCode >= 500 {
			return fmt.Errorf("server error: %d", r.StatusCode)
		}
		resp = r
		return nil
	}, retryOpts...)

	delivery := &models.WebhookDelivery{
		WebhookId:    job.webhook.Id,
		Event:        job.event,
		DeliveryId:   deliveryId,
		Url:          job.webhook.Url,
		RequestBody:  string(job.payload),
		ResponseCode: lastStatus,
		ResponseBody: lastResponse,
	}

	switch {
	case err != nil:
		s.logger.Error("webhook request failed after retries", "webhook_id", job.webhook.Id, "err", err)
		delivery.ResponseBody = err.Error()
	case resp != nil && resp.StatusCode >= 200 && resp.StatusCode < 300:
		delivery.Success = true
		s.logger.Info("webhook delivered successfully", "webhook_id", job.webhook.Id, "url", job.webhook.Url, "delivery_id", deliveryId)
	default:
		s.logger.Warn("webhook delivery failed", "webhook_id", job.webhook.Id, "status", lastStatus, "url", job.webhook.Url)
	}

	s.record(delivery)
}

func (s *Service) record(delivery *models.WebhookDelivery) {
	if err := s.db.AddWebhookDelivery(delivery); err != nil {
		s.logger.Error("failed to record webhook delivery", "webhook_id", delivery.WebhookId, "err", err)
		return
	}
	if err := s.db.PruneWebhookDeliveries(delivery.WebhookId, deliveryRetention); err != nil {
		s.logger.Error("failed to prune webhook deliveries", "webhook_id", delivery.WebhookId, "err", err)
	}
}

func (s *Service) computeSignature(payload []byte, secret string) string {
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write(payload)
	return hex.EncodeToString(mac.Sum(nil))
}