sift/internal/api/api.go

package api

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"unicode/utf8"

	"git.wntrmute.dev/kyle/sift/internal/auth"
	"git.wntrmute.dev/kyle/sift/internal/clean"
	"git.wntrmute.dev/kyle/sift/internal/dcache"
	"git.wntrmute.dev/kyle/sift/internal/digest"
	"git.wntrmute.dev/kyle/sift/internal/fetch"
	"git.wntrmute.dev/kyle/sift/internal/fleetctx"
	"git.wntrmute.dev/kyle/sift/internal/guard"
	"git.wntrmute.dev/kyle/sift/internal/metrics"
	"git.wntrmute.dev/kyle/sift/internal/pdfdoc"
	"git.wntrmute.dev/kyle/sift/internal/pdfpool"
	"git.wntrmute.dev/kyle/sift/internal/pdftext"
	"git.wntrmute.dev/kyle/sift/internal/recover"
	"git.wntrmute.dev/kyle/sift/internal/search"
	"git.wntrmute.dev/kyle/sift/internal/selfip"
)

const (
	maxReqBody      = 64 << 10
	writeTimeout    = 240 * time.Second // P5a: a PDF may take 120 s to download + 20 s queued + 60 s to extract
	requestDeadline = 210 * time.Second // P5a: the PDF path's worst case (see writeTimeout); other paths keep their own shorter timeouts
)

type Server struct {
	Version string
	Log     *slog.Logger
	Guard   *guard.Guard
	Fetch   *fetch.Fetcher
	Auth    *auth.Store
	Mux     *http.ServeMux
	OutCap  int
	SelfIP  *selfip.Discoverer
	Digest  *digest.Engine

	PDFRunner            *pdftext.Runner
	PDFPool              *pdfpool.Pool
	PDFTTL               time.Duration
	PDFMaxQuestionChunks int
	Search               *search.Engine
	Ctx                  *fleetctx.Client
	// PermitRestricted is context.permit_restricted: those callers never
	// have a fetch host written to the log (host=-). Independent of Honcho.
	PermitRestricted []string
	Prom             *metrics.Set

	global chan struct{}
	perMu  sync.Mutex
	per    map[string]chan struct{}

	reqN, blockN, scrapeN atomic.Uint64

	pdfPoolOnce sync.Once
}

func New(version string, g *guard.Guard, f *fetch.Fetcher, keys *auth.Store, log *slog.Logger) *Server {
	if log == nil {
		log = slog.Default()
	}
	s := &Server{
		Version: version,
		Log:     log,
		Guard:   g,
		Fetch:   f,
		Auth:    keys,
		Mux:     http.NewServeMux(),
		OutCap:  200_000,
		global:  make(chan struct{}, 8),
		per:     map[string]chan struct{}{},
	}
	s.Prom = metrics.New(s)
	s.Mux.HandleFunc("GET /healthz", s.healthz)
	s.Mux.HandleFunc("GET /healthz/", s.healthz)
	s.Mux.HandleFunc("GET /metrics", s.metrics)
	s.Mux.HandleFunc("GET /metrics.json", s.metrics)
	s.Mux.HandleFunc("POST /v2/scrape", s.scrape)
	s.Mux.HandleFunc("POST /v1/fetch", s.v1fetch)
	s.Mux.HandleFunc("POST /v1/search", s.v1search)
	s.Mux.HandleFunc("POST /v2/search", s.v2search)
	return s
}

func (s *Server) HTTPServer(addr string) *http.Server {
	return &http.Server{
		Addr:              addr,
		Handler:           s,
		ReadHeaderTimeout: 5 * time.Second,
		ReadTimeout:       15 * time.Second,
		WriteTimeout:      writeTimeout,
		IdleTimeout:       30 * time.Second,
		MaxHeaderBytes:    16 << 10,
	}
}

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	r.Body = http.MaxBytesReader(w, r.Body, maxReqBody)
	switch r.URL.Path {
	case "/v1/fetch", "/v2/scrape", "/v1/search", "/v2/search":
		ctx, cancel := context.WithTimeout(r.Context(), requestDeadline)
		defer cancel()
		r = r.WithContext(ctx)
		cw := &codeWriter{ResponseWriter: w}
		start := time.Now()
		s.Mux.ServeHTTP(cw, r)
		code := cw.code
		if code == 0 {
			code = http.StatusOK
		}
		s.Prom.Observe(r.URL.Path, time.Since(start), code)
		return
	}
	s.Mux.ServeHTTP(w, r)
}

type codeWriter struct {
	http.ResponseWriter
	code int
}

func (c *codeWriter) WriteHeader(status int) {
	if c.code == 0 {
		c.code = status
	}
	c.ResponseWriter.WriteHeader(status)
}

func (c *codeWriter) Write(b []byte) (int, error) {
	if c.code == 0 {
		c.code = http.StatusOK
	}
	return c.ResponseWriter.Write(b)
}

func (s *Server) healthz(w http.ResponseWriter, _ *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	out := map[string]any{"status": "ok", "version": s.Version}
	if s.SelfIP != nil {
		out["self_ip"] = s.SelfIP.Status()
	}
	if s.Search != nil {
		if d := s.Search.Health(); len(d) > 0 {
			out["status"] = "degraded"
			out["search"] = d
		}
	}
	_ = json.NewEncoder(w).Encode(out)
}

func (s *Server) metrics(w http.ResponseWriter, r *http.Request) {
	if !s.allowMetrics(w, r) {
		return
	}
	if r.URL.Path == "/metrics.json" || r.URL.Query().Get("format") == "json" {
		s.writeMetricsJSON(w)
		return
	}
	s.Prom.Handler().ServeHTTP(w, r)
}

func (s *Server) allowMetrics(w http.ResponseWriter, r *http.Request) bool {
	if metrics.LoopbackOpen(r) {
		return true
	}
	h := r.Header.Get("Authorization")
	tok := strings.TrimSpace(strings.TrimPrefix(h, "Bearer"))
	tok = strings.TrimSpace(tok)
	if tok == "" || s.Auth == nil {
		s.Log.Warn("unauthorized", "caller", "-", "host", "-", "reason", "missing metrics key")
		writeErr(w, 401, "unauthorized", "")
		return false
	}
	c := s.Auth.LookupBearer(tok)
	if c == nil {
		s.Log.Warn("unauthorized", "caller", "-", "host", "-", "reason", "unknown metrics key")
		writeErr(w, 401, "unauthorized", "")
		return false
	}
	if !c.MetricsOnly() {
		s.Log.Warn("unauthorized", "caller", c.Name, "host", "-", "reason", "not metrics caller")
		writeErr(w, 403, "unauthorized", "")
		return false
	}
	return true
}

func (s *Server) writeMetricsJSON(w http.ResponseWriter) {
	w.Header().Set("Content-Type", "application/json")
	fails := uint64(0)
	if s.SelfIP != nil {
		fails = s.SelfIP.Failures.Load()
	}
	m := map[string]any{
		"requests":         s.reqN.Load(),
		"blocks":           s.blockN.Load(),
		"scrapes":          s.scrapeN.Load(),
		"self_ip_failures": fails,
	}
	if s.Search != nil {
		for k, v := range s.Search.Metrics() {
			m[k] = v
		}
	}
	calls, fb := uint64(0), map[string]uint64{}
	var inflight, waiters int
	if s.Digest != nil {
		calls, fb = s.Digest.Stats()
		inflight, waiters = s.Digest.Gate.Snapshot()
	}
	m["digest_calls"] = calls
	m["digest_fallbacks"] = fb
	m["dixie_inflight"] = inflight
	m["dixie_queue_waiters"] = waiters
	var dH, dM, eH, eM uint64
	if cache := s.cache(); cache != nil {
		dH, dM, eH, eM, _, _, _ = cache.Stats()
	}
	m["dcache_digest_hit"] = dH
	m["dcache_digest_miss"] = dM
	m["dcache_embed_hit"] = eH
	m["dcache_embed_miss"] = eM
	_ = json.NewEncoder(w).Encode(m)
}

func (s *Server) cache() *dcache.DB {
	if s.Digest != nil && s.Digest.Cache != nil {
		return s.Digest.Cache
	}
	if s.Search != nil {
		return s.Search.Cache
	}
	return nil
}

func (s *Server) Snapshot() metrics.Snapshot {
	out := metrics.Snapshot{
		Blocks:     s.blockN.Load(),
		Scrapes:    s.scrapeN.Load(),
		Restricted: map[string]struct{}{},
	}
	for _, n := range s.PermitRestricted {
		out.Restricted[n] = struct{}{}
	}
	if s.SelfIP != nil {
		out.SelfIPFailures = s.SelfIP.Failures.Load()
	}
	if s.Digest != nil {
		out.DigestCalls, out.DigestFallbacks = s.Digest.Stats()
		out.DixieInflight, out.DixieWaiters = s.Digest.Gate.Snapshot()
	}
	if cache := s.cache(); cache != nil {
		out.DigestHit, out.DigestMiss, out.EmbedHit, out.EmbedMiss, _, _, _ = cache.Stats()
	}
	if s.Search != nil {
		m := s.Search.Metrics()
		if v, ok := m["search_queries"].(map[string]map[string]uint64); ok {
			out.SearchQueries = v
		}
		if v, ok := m["search_credits"].(map[string]map[string]uint64); ok {
			out.SearchCredits = v
		}
		if v, ok := m["search_daily_credits"].(map[string]uint64); ok {
			out.SearchDaily = v
		}
		if v, ok := m["search_capped"].(map[string]uint64); ok {
			out.SearchCapped = v
		}
		if v, ok := m["search_cache_hit"].(uint64); ok {
			out.SearchHit = v
		}
		if v, ok := m["search_cache_miss"].(uint64); ok {
			out.SearchMiss = v
		}
	}
	return out
}

type scrapeReq struct {
	URL                 string   `json:"url"`
	Formats             []string `json:"formats"`
	OnlyMainContent     *bool    `json:"onlyMainContent"`
	SkipTlsVerification *bool    `json:"skipTlsVerification"`
	RemoveBase64Images  *bool    `json:"removeBase64Images"`
	FastMode            *bool    `json:"fastMode"`
	BlockAds            *bool    `json:"blockAds"`
	StoreInCache        *bool    `json:"storeInCache"`
	MaxAge              *int     `json:"maxAge"`
	Mobile              *bool    `json:"mobile"`
	Origin              string   `json:"origin"`
}

type v1Req struct {
	URL      string `json:"url"`
	Question string `json:"question"`
	Mode     string `json:"mode"`
	Context  *bool  `json:"context"`
	JQ       string `json:"jq"`
	Recover  bool   `json:"recover"`

	// Pages narrows a PDF digest to "2-3"; Grep matches lines in it. Optional; task-07.
	Pages string `json:"pages"`
	Grep  string `json:"grep"`
}

func (s *Server) caller(w http.ResponseWriter, r *http.Request) *auth.Caller {
	h := r.Header.Get("Authorization")
	tok := strings.TrimSpace(strings.TrimPrefix(h, "Bearer"))
	tok = strings.TrimSpace(tok)
	if tok == "" || s.Auth == nil {
		s.Log.Warn("unauthorized", "caller", "-", "host", "-", "reason", "missing key")
		writeErr(w, 401, "unauthorized", "")
		return nil
	}
	c := s.Auth.LookupBearer(tok)
	if c == nil {
		s.Log.Warn("unauthorized", "caller", "-", "host", "-", "reason", "unknown key")
		writeErr(w, 401, "unauthorized", "")
		return nil
	}
	if c.MetricsOnly() {
		s.Log.Warn("unauthorized", "caller", c.Name, "host", "-", "reason", "metrics caller")
		writeErr(w, 403, "unauthorized", "")
		return nil
	}
	return c
}

func (s *Server) scrape(w http.ResponseWriter, r *http.Request) {
	s.reqN.Add(1)
	c := s.caller(w, r)
	if c == nil {
		return
	}
	var req scrapeReq
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.URL == "" {
		writeErr(w, 400, "invalid url", "")
		return
	}
	_ = req.SkipTlsVerification // MUST ignore; never disable TLS verify
	onlyMain := true
	if req.OnlyMainContent != nil {
		onlyMain = *req.OnlyMainContent
	}
	wantMD, wantHTML := true, false
	if len(req.Formats) > 0 {
		wantMD, wantHTML = false, false
		for _, f := range req.Formats {
			switch strings.ToLower(f) {
			case "markdown":
				wantMD = true
			case "html":
				wantHTML = true
			}
		}
		if !wantMD && !wantHTML {
			wantMD = true
		}
	}
	s.scrapeN.Add(1)
	res, cl, doc, ferr := s.doFetch(r, c, req.URL, "", onlyMain)
	if ferr != nil {
		s.writeFetchErr(w, c, req.URL, ferr)
		return
	}
	md, htmlOut := cl.Markdown, ""
	pages := 0
	if doc != nil {
		rp, err := s.renderPDF(doc, pdfdoc.Options{OutCap: s.OutCap})
		if err != nil {
			s.writeFetchErr(w, c, req.URL, err)
			return
		}
		md = rp.Content
		htmlOut = ""
		pages = doc.Pages
	}
	if wantHTML {
		htmlOut = cl.HTML
	}
	if !wantMD {
		md = ""
	}
	rawTok := len(res.Body) / 4
	retTok := len(md) / 4
	s.Log.Info("scrape", "caller", c.Name, "host", s.logHost(c, res.FinalURL), "status", res.Status, "bytes", len(res.Body), "pdf_pages", pages)
	writeJSON(w, 200, map[string]any{
		"success": true,
		"data": map[string]any{
			"markdown": md,
			"html":     htmlOut,
			"metadata": map[string]any{
				"title":       cl.Title,
				"sourceURL":   res.FinalURL,
				"statusCode":  res.Status,
				"contentType": res.ContentType,
				"numPages":    pages,
				"sift": map[string]any{
					"hash":      res.SHA256(),
					"fetchedAt": res.FetchedAt.Format(time.RFC3339),
					"cached":    false,
					"truncated": cl.Trunc,
					"tokens":    map[string]int{"raw": rawTok, "returned": retTok},
				},
			},
		},
	})
}

func (s *Server) v1fetch(w http.ResponseWriter, r *http.Request) {
	s.reqN.Add(1)
	c := s.caller(w, r)
	if c == nil {
		return
	}
	var req v1Req
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.URL == "" {
		writeErr(w, 400, "invalid url", "")
		return
	}
	q := req.Question
	if len([]rune(q)) > 500 {
		rns := []rune(q)
		q = string(rns[:500])
	}
	mode := req.Mode
	if mode == "" {
		if q != "" {
			mode = "digest"
		} else {
			mode = "clean"
		}
	}
	if mode != "clean" && mode != "raw" && mode != "digest" {
		writeErr(w, 400, "invalid url", "")
		return
	}
	wantCtx := q != ""
	if req.Context != nil {
		wantCtx = *req.Context
	}

	t0 := time.Now()
	onlyMain := true
	res, cl, doc, ferr := s.doFetch(r, c, req.URL, req.JQ, onlyMain)
	var snap *recover.Result
	if ferr != nil && req.Recover && recoverWorthy(ferr) {
		sr, rerr := recover.Try(r.Context(), s.Fetch, req.URL)
		if rerr == nil && sr != nil && sr.Res != nil {
			res, cl, doc, ferr = s.cleanFetched(sr.Res, req.JQ, onlyMain)
			snap = sr
		}
	}
	if ferr != nil {
		s.writeFetchErr(w, c, req.URL, ferr)
		return
	}
	fetchMS := time.Since(t0).Milliseconds()

	if doc == nil && (req.Pages != "" || req.Grep != "") {
		s.writeFetchErr(w, c, req.URL, &fetch.Error{Status: 400, Msg: "pages and grep apply to PDFs only"})
		return
	}

	var docLen, pdfPages int
	var content string
	var pdfMap map[string]any
	if doc != nil {
		pdfPages = doc.Pages
		if mode == "raw" {
			writeErr(w, 400, "raw mode does not apply to PDFs", "")
			return
		}
		rp, err := s.renderPDF(doc, pdfdoc.Options{Pages: req.Pages, Grep: req.Grep, OutCap: s.OutCap})
		if err != nil {
			if errors.Is(err, pdfdoc.ErrBadPages) || errors.Is(err, pdfdoc.ErrBadGrep) {
				s.writeFetchErr(w, c, req.URL, &fetch.Error{Status: 400, Msg: err.Error()})
				return
			}
			s.writeFetchErr(w, c, req.URL, err)
			return
		}
		full := pdfdoc.Full(doc)
		docLen = utf8.RuneCountInString(full)
		if mode == "digest" {
			content = full
		} else {
			content = rp.Content
		}
		pdfMap = map[string]any{
			"pages":     doc.Pages,
			"title":     doc.Title,
			"outline":   doc.Outline,
			"extracted": len(doc.Text),
			"first":     rp.First,
			"last":      rp.Last,
			"map":       rp.Map,
			"grep_hits": rp.GrepHits,
			"note":      rp.Note,
		}
	} else {
		content = cl.Markdown
		if cl.Kind == "json" {
			content = cl.HTML
		}
		if mode == "raw" {
			content = string(res.Body)
			if len(content) > s.OutCap {
				content = clean.TruncateUTF8(content, s.OutCap)
				cl.Trunc = true
				cl.Warning = "truncated to output cap"
			}
		}
	}
	warn := []string{}
	if cl.Warning != "" {
		warn = append(warn, cl.Warning)
	}

	var digMeta map[string]any
	var digestMS int64
	if mode == "digest" {
		if doc != nil {
			n := len(digest.Split(pdfdoc.Full(doc), digest.ChunkTokens))
			limit := s.PDFMaxQuestionChunks
			if limit == 0 {
				limit = DefaultPDFMaxQuestionChunks
			}
			if n > limit {
				s.writeFetchErr(w, c, req.URL, &fetch.Error{Status: 422, Msg: fmt.Sprintf("document too long for question (%d chunks); use grep or pages", n)})
				return
			}
		}
		if s.Digest == nil {
			warn = append(warn, "digest fell back to clean: unset")
			mode = "clean"
			content, warn = capDigestFallback(content, warn)
			digMeta = map[string]any{"fallback": "unset"}
		} else {
			dr := s.Digest.Run(r.Context(), content, q, res.FinalURL, s.callerClass(c))
			digestMS = dr.DigestMS
			if !dr.Cached {
				s.Prom.ObserveDigest(time.Duration(dr.DigestMS) * time.Millisecond)
			}
			if dr.Fallback != "" {
				warn = append(warn, "digest fell back to clean: "+dr.Fallback)
				mode = "clean"
				content, warn = capDigestFallback(content, warn)
				digMeta = map[string]any{"fallback": dr.Fallback, "model": dr.Model}
			} else {
				content = dr.Text
				digMeta = map[string]any{
					"model":        dr.Model,
					"chunks_used":  dr.ChunksUsed,
					"chunks_total": dr.ChunksTotal,
					"fallback":     false,
					"cached":       dr.Cached,
				}
			}
		}
	}

	var ctxItems []fleetctx.Item
	var ctxMS int64
	if wantCtx && s.Ctx != nil {
		ct0 := time.Now()
		query := fleetctx.Query(q, cl.Title, headings(cl.Markdown))
		items, cw := s.Ctx.Search(r.Context(), c.Workspace, c.Name, query)
		ctxMS = time.Since(ct0).Milliseconds()
		if cw != "" {
			warn = append(warn, cw)
		}
		ctxItems = items
	}

	qLen := len(q)
	if qLen > 80 {
		qLen = 80
	}
	fb, _ := "", false
	if digMeta != nil {
		if s, ok := digMeta["fallback"].(string); ok {
			fb = s
		}
	}
	s.Log.Info("fetch", "caller", c.Name, "host", s.logHost(c, res.FinalURL), "mode", mode,
		"q_len", qLen, "raw_tokens", len(res.Body)/4, "returned_tokens", len(content)/4,
		"pdf_pages", pdfPages, "digest_ms", digestMS, "fallback", fb)

	src := map[string]any{
		"url":          res.URL,
		"final_url":    res.FinalURL,
		"status":       res.Status,
		"content_type": res.ContentType,
		"fetched_at":   res.FetchedAt.Format(time.RFC3339),
		"hash":         res.SHA256(),
		"bytes":        len(res.Body),
		"cached":       len(res.Body) == 0,
		"kind":         "live",
	}
	if snap != nil {
		src["kind"] = "snapshot"
		src["via"] = string(snap.Via)
		src["original_url"] = snap.OriginalURL
		if snap.SnapshotAt != "" {
			src["snapshot_at"] = snap.SnapshotAt
		} else {
			src["snapshot_at"] = nil
			warn = append(warn, "snapshot date unknown")
		}
	}
	out := map[string]any{
		"content":   content,
		"mode":      mode,
		"untrusted": true,
		"source":    src,
		"warnings":  warn,
		"sizes": map[string]int{
			"raw_tokens":      docLen / 4,
			"clean_tokens":    digest.EstTokens(cl.Markdown),
			"returned_tokens": digest.EstTokens(content),
		},
		"timings": map[string]int64{"fetch_ms": fetchMS, "digest_ms": digestMS, "context_ms": ctxMS},
	}
	if pdfMap != nil {
		out["pdf"] = pdfMap
	}
	if digMeta != nil {
		out["digest"] = digMeta
	}
	if wantCtx {
		out["context"] = ctxItems
	}
	writeJSON(w, 200, out)
}

func recoverWorthy(err error) bool {
	var fe *fetch.Error
	if !errors.As(err, &fe) {
		return false
	}
	if strings.Contains(fe.Msg, "blocked:") {
		return false
	}
	return fe.Hint == "browser" || fe.Status == 422 || fe.Status == 429 || fe.Status == 504
}

func headings(md string) []string {
	var h []string
	for _, line := range strings.Split(md, "\n") {
		if strings.HasPrefix(line, "#") {
			h = append(h, strings.TrimLeft(line, "# "))
		}
	}
	return h
}

func (s *Server) cleanFetched(res *fetch.Resource, jq string, onlyMain bool) (*fetch.Resource, *clean.Result, *pdftext.Doc, error) {
	kind := clean.Kind(res.ContentType, res.Body)
	if kind == "other" || kind == "pdf" {
		// The recover path (archive snapshots) never serves a PDF.
		return nil, nil, nil, &fetch.Error{Status: 415, Hint: "browser", Msg: "unsupported content type"}
	}
	var cl *clean.Result
	var cerr error
	switch kind {
	case "html":
		cl, cerr = clean.HTML(res, onlyMain, s.OutCap)
	case "json":
		cl, cerr = clean.JSON(res, jq, s.OutCap)
	default:
		cl, cerr = clean.Text(res, s.OutCap)
	}
	if cerr != nil {
		return nil, nil, nil, &fetch.Error{Status: 422, Msg: cerr.Error()}
	}
	return res, cl, nil, nil
}

func (s *Server) doFetch(r *http.Request, c *auth.Caller, rawURL, jq string, onlyMain bool) (*fetch.Resource, *clean.Result, *pdftext.Doc, error) {
	res, hitDoc, ok, err := s.pdfFromCache(r.Context(), c, rawURL)
	if err != nil {
		return nil, nil, nil, err
	}
	if ok {
		return res, &clean.Result{Kind: "pdf", Title: hitDoc.Title}, hitDoc, nil
	}
	if res == nil {
		if err := s.acquire(c.Name, rawURL); err != nil {
			return nil, nil, nil, err
		}
		defer s.release(c.Name, rawURL)

		res, err = s.Fetch.Fetch(r.Context(), rawURL)
		if err != nil {
			return nil, nil, nil, err
		}
	}
	kind := clean.Kind(res.ContentType, res.Body)
	if kind == "other" {
		ct := res.ContentType
		if ct == "" {
			ct = "unknown"
		}
		return nil, nil, nil, &fetch.Error{Status: 415, Hint: "browser", Msg: "unsupported content type " + ct + "; hint: browser"}
	}
	var cl *clean.Result
	var cerr error
	switch kind {
	case "html":
		cl, cerr = clean.HTML(res, onlyMain, s.OutCap)
	case "json":
		cl, cerr = clean.JSON(res, jq, s.OutCap)
	case "text":
		cl, cerr = clean.Text(res, s.OutCap)
	}
	if cerr != nil {
		var tl *clean.TooLargeError
		if errors.As(cerr, &tl) {
			return nil, nil, nil, &fetch.Error{Status: 400, Msg: tl.Error()}
		}
		return nil, nil, nil, &fetch.Error{Status: 422, Msg: cerr.Error()}
	}
	var doc *pdftext.Doc
	if kind == "pdf" {
		doc, err = s.extractPDF(r.Context(), c, res)
		if err != nil {
			return nil, nil, nil, err
		}
		s.storePDF(dcache.PDFKey(rawURL, s.callerClass(c)), res, doc)
		cl = &clean.Result{Kind: "pdf", Title: doc.Title}
	}
	return res, cl, doc, nil
}

func (s *Server) writeFetchErr(w http.ResponseWriter, c *auth.Caller, rawURL string, err error) {
	name := "-"
	if c != nil {
		name = c.Name
	}
	host := s.logHost(c, rawURL)
	var fe *fetch.Error
	if errors.As(err, &fe) {
		if fe.Status == 403 {
			s.blockN.Add(1)
		}
		if fe.Status == 502 {
			fe.Status = 500
			fe.Msg = "internal error"
		}
		if fe.Status == 401 || fe.Status == 403 {
			s.Log.Warn("refused", "caller", name, "host", host, "reason", s.logReason(c, fe))
		} else {
			s.Log.Warn("fetch failed", "caller", name, "host", host, "reason", s.logReason(c, fe), "status", fe.Status)
		}
		if fe.RetryAfter > 0 {
			w.Header().Set("Retry-After", strconv.Itoa(fe.RetryAfter))
		}
		writeErr(w, fe.Status, fe.Msg, fe.Hint)
		return
	}
	s.Log.Error("internal", "caller", name, "host", host, "err", s.logErr(c, err))
	writeErr(w, 500, "internal error", "")
}

func (s *Server) acquire(caller, rawURL string) error {
	host := hostOf(rawURL)
	key := caller + "|" + host
	select {
	case s.global <- struct{}{}:
	case <-time.After(5 * time.Second):
		return &fetch.Error{Status: 429, Msg: "busy"}
	}
	s.perMu.Lock()
	ch, ok := s.per[key]
	if !ok {
		ch = make(chan struct{}, 2)
		s.per[key] = ch
	}
	s.perMu.Unlock()
	select {
	case ch <- struct{}{}:
		return nil
	case <-time.After(5 * time.Second):
		<-s.global
		return &fetch.Error{Status: 429, Msg: "busy"}
	}
}

func (s *Server) release(caller, rawURL string) {
	key := caller + "|" + hostOf(rawURL)
	s.perMu.Lock()
	ch := s.per[key]
	if ch != nil {
		select {
		case <-ch:
		default:
		}
		if len(ch) == 0 {
			delete(s.per, key)
		}
	}
	s.perMu.Unlock()
	select {
	case <-s.global:
	default:
	}
}

func capDigestFallback(content string, warn []string) (string, []string) {
	if digest.EstTokens(content) <= digest.SmallDoc {
		return content, warn
	}
	return digest.CapBudget(content), append(warn, "clean fallback truncated to digest budget")
}

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

func permittedCaller(name string, permit []string) bool {
	if name == "" {
		return false
	}
	for _, p := range permit {
		if p == name {
			return true
		}
	}
	return false
}

// logHost is the host field for request logs. permit_restricted callers get "-".
func (s *Server) logHost(c *auth.Caller, raw string) string {
	if c != nil && permittedCaller(c.Name, s.PermitRestricted) {
		return "-"
	}
	return hostOf(raw)
}

func (s *Server) restrictedLog(c *auth.Caller) bool {
	return c != nil && permittedCaller(c.Name, s.PermitRestricted)
}

func (s *Server) callerClass(c *auth.Caller) string {
	return dcache.CallerClass(s.restrictedLog(c))
}

// logReason is the reason field for fetch error logs. Restricted callers get a
// coarse class so the hostname in guard messages does not land in the journal.
func (s *Server) logReason(c *auth.Caller, fe *fetch.Error) string {
	if fe == nil {
		return "-"
	}
	if s.restrictedLog(c) {
		return coarseReason(fe)
	}
	return fe.Msg
}

func (s *Server) logErr(c *auth.Caller, err error) any {
	if s.restrictedLog(c) {
		return "-"
	}
	return err
}

func coarseReason(fe *fetch.Error) string {
	switch fe.Status {
	case 401:
		return "unauthorized"
	case 403:
		return "blocked"
	case 400:
		return "bad request"
	case 413:
		return "too large"
	case 415:
		return "unsupported"
	case 429:
		return "busy"
	case 504:
		return "timeout"
	default:
		if fe.Status >= 400 && fe.Status < 500 {
			return "upstream 4xx"
		}
		if fe.Status >= 500 {
			return "upstream 5xx"
		}
		return "-"
	}
}

func writeErr(w http.ResponseWriter, status int, msg, hint string) {
	if status == 502 {
		status = 500
		msg = "internal error"
	}
	body := map[string]any{"success": false, "error": msg}
	if hint != "" {
		body["hint"] = hint
	}
	writeJSON(w, status, body)
}

func writeJSON(w http.ResponseWriter, status int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(v)
}

func HashBody(b []byte) string {
	sum := sha256.Sum256(b)
	return hex.EncodeToString(sum[:])
}

func Drain(r io.Reader) {
	_, _ = io.Copy(io.Discard, r)
}