sift/internal/api/api.go

package api

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

	"git.wntrmute.dev/kyle/sift/internal/auth"
	"git.wntrmute.dev/kyle/sift/internal/clean"
	"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/recover"
	"git.wntrmute.dev/kyle/sift/internal/selfip"
)

const (
	maxReqBody      = 64 << 10
	writeTimeout    = 120 * time.Second
	requestDeadline = 90 * time.Second
)

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
	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

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

	reqN, blockN, scrapeN atomic.Uint64
}

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.Mux.HandleFunc("GET /healthz", s.healthz)
	s.Mux.HandleFunc("GET /healthz/", s.healthz)
	s.Mux.HandleFunc("GET /metrics", s.metrics)
	s.Mux.HandleFunc("POST /v2/scrape", s.scrape)
	s.Mux.HandleFunc("POST /v1/fetch", s.v1fetch)
	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":
		ctx, cancel := context.WithTimeout(r.Context(), requestDeadline)
		defer cancel()
		r = r.WithContext(ctx)
	}
	s.Mux.ServeHTTP(w, r)
}

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()
	}
	_ = json.NewEncoder(w).Encode(out)
}

func (s *Server) metrics(w http.ResponseWriter, _ *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	fails := uint64(0)
	if s.SelfIP != nil {
		fails = s.SelfIP.Failures.Load()
	}
	_ = json.NewEncoder(w).Encode(map[string]any{
		"requests":         s.reqN.Load(),
		"blocks":           s.blockN.Load(),
		"scrapes":          s.scrapeN.Load(),
		"self_ip_failures": fails,
	})
}

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"`
}

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
	}
	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, ferr := s.doFetch(r, c, req.URL, "", onlyMain)
	if ferr != nil {
		s.writeFetchErr(w, c, req.URL, ferr)
		return
	}
	md, htmlOut := cl.Markdown, ""
	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))
	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,
				"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, 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, 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()

	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 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)
			digestMS = dr.DigestMS
			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,
		"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":       false,
		"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":      len(res.Body) / 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 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, error) {
	kind := clean.MediaKind(res.ContentType)
	if kind == "sniff" {
		kind = clean.Sniff(res.Body)
	}
	if kind == "other" {
		return 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, &fetch.Error{Status: 422, Msg: cerr.Error()}
	}
	return res, cl, nil
}

func (s *Server) doFetch(r *http.Request, c *auth.Caller, rawURL, jq string, onlyMain bool) (*fetch.Resource, *clean.Result, error) {
	if err := s.acquire(c.Name, rawURL); err != nil {
		return nil, nil, err
	}
	defer s.release(c.Name, rawURL)

	res, err := s.Fetch.Fetch(r.Context(), rawURL)
	if err != nil {
		return nil, nil, err
	}
	kind := clean.MediaKind(res.ContentType)
	if kind == "sniff" {
		kind = clean.Sniff(res.Body)
	}
	if kind == "other" {
		ct := res.ContentType
		if ct == "" {
			ct = "unknown"
		}
		return 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)
	default:
		cl, cerr = clean.Text(res, s.OutCap)
	}
	if cerr != nil {
		var tl *clean.TooLargeError
		if errors.As(cerr, &tl) {
			return nil, nil, &fetch.Error{Status: 400, Msg: tl.Error()}
		}
		return nil, nil, &fetch.Error{Status: 422, Msg: cerr.Error()}
	}
	return res, cl, 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)
		}
		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)
}

// 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)
}