package console

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"io"
	"log"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/objstore"
	"github.com/carlosframework/platform/internal/release"
)

const (
	scanDeliveryWorkers = 4
	scanDeliveryQueue   = 64
	scanDeliveryTimeout = 10 * time.Second
)

var scanDeliveryRetryDelays = []time.Duration{time.Second, 4 * time.Second, 16 * time.Second}

type scanDeliveryJob struct {
	ref  objstore.AppRef
	done objstore.Drained
	path string
	gen  int
}

type scanCallbackBody struct {
	Account   string           `json:"account"`
	App       string           `json:"app"`
	Key       string           `json:"key"`
	ETag      string           `json:"etag"`
	Verdict   objstore.Verdict `json:"verdict"`
	Recorded  bool             `json:"recorded"`
	Engine    string           `json:"engine"`
	Signature string           `json:"signature"`
}

// scanDeliveryPool keeps tenant HTTP outside the drain goroutine. Its queue is
// intentionally bounded: the status door is durable, while allowing callbacks
// to grow memory without limit would turn a best-effort convenience into a
// control-plane denial of service.
type scanDeliveryPool struct {
	s      *Server
	ctx    context.Context
	jobs   chan scanDeliveryJob
	work   chan scanDeliveryJob
	done   chan objstore.AppRef
	wg     sync.WaitGroup
	client *http.Client
}

func newScanDeliveryPool(ctx context.Context, s *Server) *scanDeliveryPool {
	transport := s.cfg.CallbackTransport
	if transport == nil {
		// http.DefaultTransport verifies both the public CA chain and the target
		// hostname. Do not clone it merely to set TLSClientConfig: that is where
		// an InsecureSkipVerify regression can hide.
		transport = http.DefaultTransport
	}
	p := &scanDeliveryPool{
		s: s, ctx: ctx, jobs: make(chan scanDeliveryJob, scanDeliveryQueue),
		work: make(chan scanDeliveryJob), done: make(chan objstore.AppRef),
		client: &http.Client{
			Transport: transport,
			Timeout:   scanDeliveryTimeout,
			CheckRedirect: func(*http.Request, []*http.Request) error {
				return http.ErrUseLastResponse
			},
		},
	}
	p.wg.Add(1)
	go p.dispatch()
	for range scanDeliveryWorkers {
		p.wg.Add(1)
		go p.worker()
	}
	return p
}

func (p *scanDeliveryPool) stop() { p.wg.Wait() }

func (p *scanDeliveryPool) enqueue(job scanDeliveryJob) {
	if p == nil || job.path == "" {
		return
	}
	select {
	case p.jobs <- job:
	case <-p.ctx.Done():
	default:
		log.Printf("console: scan callback: %s/%s: dropped — delivery queue full", job.ref.Account, job.ref.App)
	}
}

func (p *scanDeliveryPool) dispatch() {
	defer p.wg.Done()
	busy := make(map[objstore.AppRef]bool)
	queued := make(map[objstore.AppRef]int)
	pending := make([]scanDeliveryJob, 0, scanDeliveryQueue)
	for {
		selected := -1
		for i, job := range pending {
			if !busy[job.ref] {
				selected = i
				break
			}
		}
		if selected >= 0 {
			job := pending[selected]
			select {
			case p.work <- job:
				busy[job.ref] = true
				queued[job.ref]--
				pending = append(pending[:selected], pending[selected+1:]...)
			case ref := <-p.done:
				delete(busy, ref)
			case <-p.ctx.Done():
				return
			}
			continue
		}
		select {
		case job := <-p.jobs:
			if len(pending) >= scanDeliveryQueue || queued[job.ref] >= 8 {
				log.Printf("console: scan callback: %s/%s: dropped — delivery pending cap reached", job.ref.Account, job.ref.App)
			} else {
				pending = append(pending, job)
				queued[job.ref]++
			}
		case ref := <-p.done:
			delete(busy, ref)
		case <-p.ctx.Done():
			return
		}
	}
}

func (p *scanDeliveryPool) worker() {
	defer p.wg.Done()
	for {
		select {
		case <-p.ctx.Done():
			return
		case job := <-p.work:
			p.deliver(job)
			select {
			case p.done <- job.ref:
			case <-p.ctx.Done():
				return
			}
		}
	}
}

func (p *scanDeliveryPool) deliver(job scanDeliveryJob) {
	if p.s.cfg.CallbackSigner == nil {
		return // an unsigned POST is worse than a dropped best-effort callback
	}
	host, err := p.s.scanCallbackHost(job.ref)
	if err != nil {
		log.Printf("console: scan callback: %s/%s: %v", job.ref.Account, job.ref.App, err)
		return
	}
	if host == "" {
		return // no ready route means there is no platform host yet; the app polls
	}
	target, err := callbackTarget(host, job.path)
	if err != nil {
		log.Printf("console: scan callback: %s/%s: %v", job.ref.Account, job.ref.App, err)
		return
	}
	verdict := job.done.Verdict
	if job.done.Gone {
		verdict = objstore.Verdict("gone")
	}
	body, err := json.Marshal(scanCallbackBody{
		Account: job.ref.Account, App: job.ref.App, Key: job.done.Key,
		ETag: strings.Trim(job.done.ETag, `"`), Verdict: verdict,
		Recorded: job.done.Recorded, Engine: job.done.Engine, Signature: job.done.Signature,
	})
	if err != nil {
		return
	}
	for attempt := 0; ; attempt++ {
		retry, err := p.attempt(job, target.String(), body)
		if errors.Is(err, errScanCallbackNotPermitted) {
			return
		}
		if err != nil {
			log.Printf("console: scan callback: %s/%s: %v", job.ref.Account, job.ref.App, err)
		}
		if !retry || attempt == len(scanDeliveryRetryDelays) {
			return
		}
		t := time.NewTimer(scanDeliveryRetryDelays[attempt])
		select {
		case <-p.ctx.Done():
			t.Stop()
			return
		case <-t.C:
		}
	}
}

var errScanCallbackNotPermitted = errors.New("console: scan callback no longer permitted")

// attempt is one POST as one app mutation, so a foreign exclusive operation
// on the app refuses the delivery. Consent and the callback generation are
// re-read inside it, before anything is sent.
func (p *scanDeliveryPool) attempt(job scanDeliveryJob, target string, body []byte) (bool, error) {
	acct, app := job.ref.Account, job.ref.App
	var retry bool
	err := p.s.claimAppMutation(p.ctx, &Account{Sqid: acct}, app, "scan-callback-delivery", func(held context.Context) error {
		if p.s.cfg.HostClaim {
			appID, err := p.s.mutationAppID(held, acct, app)
			if err == nil {
				err = hostops.RequireAppMutation(held, p.s.cfg.Releases, acct, app, appID)
			}
			if err != nil {
				return err
			}
		}
		// BOUNDED. p.ctx is the pool's lifetime context and carries no deadline of
		// its own, so a release store that stops answering would hold this worker
		// for the life of the process — a quarter of the pool per hung read.
		consentCtx, consentCancel := context.WithTimeout(held, storeTimeout)
		defer consentCancel()
		rec, refusal, err := objstore.QueuePermitted(consentCtx, p.s.cfg.Releases, acct, app)
		if err != nil || refusal != objstore.RefusalNone || rec == nil || rec.Scanning == nil || rec.Scanning.WebhookGen != job.gen {
			return errScanCallbackNotPermitted
		}
		var perr error
		retry, perr = p.post(held, job, target, body, job.gen)
		return perr
	})
	return retry || errors.Is(err, hostops.ErrPending), err
}

func (p *scanDeliveryPool) post(ctx context.Context, job scanDeliveryJob, target string, body []byte, gen int) (bool, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, target, bytes.NewReader(body))
	if err != nil {
		return false, err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("User-Agent", "carlos-scan-callback/1")
	req.Header.Set("Carlos-Signature", p.s.cfg.CallbackSigner.Sign(job.ref.Account, job.ref.App, gen, body, time.Now()))
	res, err := p.client.Do(req)
	if err != nil {
		return !errors.Is(err, context.Canceled), err
	}
	defer res.Body.Close()
	_, _ = io.Copy(io.Discard, io.LimitReader(res.Body, 4<<10))
	if res.StatusCode >= 200 && res.StatusCode < 300 {
		return false, nil
	}
	if res.StatusCode >= 300 && res.StatusCode < 400 {
		// SAID OUT LOUD, because silence here is indistinguishable from
		// success to the only person who can fix it. Redirects are refused by
		// design (a tenant answering 302 to an internal address is the attack),
		// but the ordinary cause is dull: a framework that redirects /hook to
		// /hook/ turns every callback into a refusal, and nothing anywhere
		// would otherwise say so.
		log.Printf("console: scan callback %s/%s: %d redirect refused — a callback path must answer directly",
			job.ref.Account, job.ref.App, res.StatusCode)
	}
	retry := res.StatusCode == http.StatusTooManyRequests || res.StatusCode >= 500
	return retry, nil
}

// scanCallbackHost returns only a platform-assigned name that exists in the
// registry: the canonical app.account.apps-domain when it is a ready platform
// route, otherwise the lexicographically first ready platform route of the app.
func (s *Server) scanCallbackHost(ref objstore.AppRef) (string, error) {
	if s.cfg.AppsDomain == "" || s.cfg.Registry == "" {
		return "", nil
	}
	reg, err := openRegistry(s.cfg.Registry)
	if err != nil {
		return "", err
	}
	routes, err := reg.List()
	if err != nil {
		return "", err
	}
	canonical := ref.App + "." + ref.Account + "." + strings.TrimPrefix(s.cfg.AppsDomain, ".")
	best := ""
	for _, rt := range routes {
		if rt.App != ref.App || release.RouteAccount(rt) != ref.Account || rt.Status != "ready" ||
			!rt.Platform || rt.Canary || rt.IsSidecar() {
			continue
		}
		if rt.Host == canonical {
			return canonical, nil
		}
		if best == "" || rt.Host < best {
			best = rt.Host
		}
	}
	return best, nil
}