kenn-io--agentsview/internal/server/huma_routes.go

package server

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

	"github.com/danielgtaylor/huma/v2"
	"github.com/danielgtaylor/huma/v2/adapters/humago"

	"go.kenn.io/agentsview/internal/db"
	"go.kenn.io/agentsview/internal/timeutil"
)

type emptyInput struct{}

type jsonOutput[T any] struct {
	Body T
}

type createdOutput[T any] struct {
	Status int `status:"201"`
	Body   T
}

type noContentOutput struct {
	Status int `status:"204"`
}

type bytesOutput struct {
	ContentType        string `header:"Content-Type"`
	ContentDisposition string `header:"Content-Disposition"`
	NoSniff            string `header:"X-Content-Type-Options"`
	CacheControl       string `header:"Cache-Control"`
	Body               []byte
}

type apiErrorResponse struct {
	Status  int    `json:"-"`
	Code    string `json:"code,omitempty"`
	Message string `json:"error"`
}

func (e *apiErrorResponse) Error() string {
	return e.Message
}

func (e *apiErrorResponse) GetStatus() int {
	return e.Status
}

func apiError(status int, message string) error {
	return &apiErrorResponse{Status: status, Message: message}
}

func apiErrorWithCode(status int, code, message string) error {
	return &apiErrorResponse{Status: status, Code: code, Message: message}
}

var configureHumaErrorsOnce stdsync.Once

func configureHumaErrors() {
	configureHumaErrorsOnce.Do(func() {
		huma.NewError = func(status int, message string, errs ...error) huma.StatusError {
			if status == http.StatusUnprocessableEntity {
				status = http.StatusBadRequest
			}
			if len(errs) > 0 {
				var details []string
				for _, err := range errs {
					if err == nil {
						continue
					}
					details = append(details, err.Error())
				}
				if len(details) > 0 {
					message = strings.Join(details, "; ")
				}
			}
			if strings.Contains(message, "(query.type:") {
				message = "invalid type: " + message
			}
			return &apiErrorResponse{
				Status:  status,
				Message: message,
			}
		}
		huma.NewErrorWithContext = func(
			_ huma.Context,
			status int,
			message string,
			errs ...error,
		) huma.StatusError {
			return huma.NewError(status, message, errs...)
		}
	})
}

type requestInfo struct {
	RemoteAddr string
	Forwarded  bool
}

type optionalIntParam struct {
	Value int
	IsSet bool
}

func (p optionalIntParam) Schema(r huma.Registry) *huma.Schema {
	return huma.SchemaFromType(r, reflect.TypeFor[int]())
}

func (p *optionalIntParam) Receiver() reflect.Value {
	return reflect.ValueOf(p).Elem().FieldByName("Value")
}

func (p *optionalIntParam) OnParamSet(isSet bool, _ any) {
	p.IsSet = isSet
}

func optionalIntValue(p optionalIntParam) *int {
	if !p.IsSet {
		return nil
	}
	return &p.Value
}

// optionalBoolParam distinguishes an omitted query param from an explicit
// false, so a sort key's canonical direction is used unless ?descending is set.
type optionalBoolParam struct {
	Value bool
	IsSet bool
}

func (p optionalBoolParam) Schema(r huma.Registry) *huma.Schema {
	return huma.SchemaFromType(r, reflect.TypeFor[bool]())
}

func (p *optionalBoolParam) Receiver() reflect.Value {
	return reflect.ValueOf(p).Elem().FieldByName("Value")
}

func (p *optionalBoolParam) OnParamSet(isSet bool, _ any) {
	p.IsSet = isSet
}

func optionalBoolValue(p optionalBoolParam) *bool {
	if !p.IsSet {
		return nil
	}
	return &p.Value
}

const ctxKeyHumaRequestInfo contextKey = 100

func humaRequestInfoMiddleware(ctx huma.Context, next func(huma.Context)) {
	info := requestInfo{
		RemoteAddr: ctx.RemoteAddr(),
		Forwarded: ctx.Header("X-Forwarded-For") != "" ||
			ctx.Header("X-Real-IP") != "" ||
			ctx.Header("Forwarded") != "",
	}
	next(huma.WithValue(ctx, ctxKeyHumaRequestInfo, info))
}

func isLocalhostContext(ctx context.Context) bool {
	info, _ := ctx.Value(ctxKeyHumaRequestInfo).(requestInfo)
	if info.Forwarded {
		return false
	}
	host, _, err := net.SplitHostPort(info.RemoteAddr)
	if err != nil {
		host = info.RemoteAddr
	}
	ip := net.ParseIP(host)
	return ip != nil && ip.IsLoopback()
}

func agentsViewSchemaNamer(t reflect.Type, hint string) string {
	name := huma.DefaultSchemaNamer(t, hint)
	base := schemaNamedType(t)
	pkgPath := base.PkgPath()
	const internalPrefix = "go.kenn.io/agentsview/internal/"
	if pkgPath == "" ||
		!strings.HasPrefix(pkgPath, internalPrefix) ||
		strings.HasSuffix(pkgPath, "/server") {
		return name
	}
	pkg := strings.TrimPrefix(pkgPath, internalPrefix)
	pkg = strings.NewReplacer("/", "", "-", "", "_", "").Replace(pkg)
	if pkg == "" {
		return name
	}
	return pascalASCII(pkg) + name
}

func schemaNamedType(t reflect.Type) reflect.Type {
	for t.Kind() == reflect.Pointer ||
		t.Kind() == reflect.Slice ||
		t.Kind() == reflect.Array {
		t = t.Elem()
	}
	return t
}

func pascalASCII(s string) string {
	if s == "" || s[0] < 'a' || s[0] > 'z' {
		return s
	}
	return string(s[0]-('a'-'A')) + s[1:]
}

func get[I, O any](
	s *Server, group routeGroup, path, summary string,
	handler func(context.Context, *I) (*O, error),
) {
	registerRoute(group, http.MethodGet, path, summary, handler, s.humaTimeout())
}

func getLong[I, O any](
	_ *Server, group routeGroup, path, summary string,
	handler func(context.Context, *I) (*O, error),
) {
	registerRoute(group, http.MethodGet, path, summary, handler)
}

func post[I, O any](
	s *Server, group routeGroup, path, summary string,
	handler func(context.Context, *I) (*O, error),
) {
	registerRoute(group, http.MethodPost, path, summary, handler, s.humaTimeout())
}

func postLong[I, O any](
	_ *Server, group routeGroup, path, summary string,
	handler func(context.Context, *I) (*O, error),
) {
	registerRoute(group, http.MethodPost, path, summary, handler)
}

func put[I, O any](
	s *Server, group routeGroup, path, summary string,
	handler func(context.Context, *I) (*O, error),
) {
	registerRoute(group, http.MethodPut, path, summary, handler, s.humaTimeout())
}

func patch[I, O any](
	s *Server, group routeGroup, path, summary string,
	handler func(context.Context, *I) (*O, error),
) {
	registerRoute(group, http.MethodPatch, path, summary, handler, s.humaTimeout())
}

func deleteRoute[I, O any](
	s *Server, group routeGroup, path, summary string,
	handler func(context.Context, *I) (*O, error),
) {
	registerRoute(group, http.MethodDelete, path, summary, handler, s.humaTimeout())
}

func stream[I any](
	_ *Server, group routeGroup, method, path, summary string,
	handler func(context.Context, *I) (*huma.StreamResponse, error),
	options ...func(*huma.Operation),
) {
	routeOptions := append([]func(*huma.Operation){streamResponse()}, options...)
	registerRoute(group, method, path, summary, handler, routeOptions...)
}

func raw[I any](
	_ *Server, group routeGroup, method, path, summary string,
	handler func(context.Context, *I) (*bytesOutput, error),
) {
	registerRoute(group, method, path, summary, handler)
}

func operationID(method, path string) string {
	var b strings.Builder
	b.WriteString(strings.ToLower(method))
	lastDash := false
	for _, r := range path {
		switch {
		case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
			b.WriteRune(r)
			lastDash = false
		case r >= 'A' && r <= 'Z':
			b.WriteRune(r + ('a' - 'A'))
			lastDash = false
		default:
			if b.Len() > 0 && !lastDash {
				b.WriteByte('-')
				lastDash = true
			}
		}
	}
	return strings.Trim(b.String(), "-")
}

func registerRoute[I, O any](
	group routeGroup, method, path, summary string,
	handler func(context.Context, *I) (*O, error),
	options ...func(*huma.Operation),
) {
	op := huma.Operation{
		OperationID: operationID(method, group.fullPath(path)),
		Method:      method,
		Path:        path,
		Summary:     summary,
		Errors: []int{
			http.StatusBadRequest,
			http.StatusUnauthorized,
			http.StatusForbidden,
			http.StatusNotFound,
			http.StatusConflict,
			http.StatusInternalServerError,
			http.StatusNotImplemented,
			http.StatusBadGateway,
			http.StatusServiceUnavailable,
			http.StatusGatewayTimeout,
		},
	}
	for _, option := range options {
		option(&op)
	}
	huma.Register(group.api, op, handler)
}

func streamResponse() func(*huma.Operation) {
	return func(op *huma.Operation) {
		op.Responses = map[string]*huma.Response{
			"200": {
				Description: "OK",
				Content: map[string]*huma.MediaType{
					"text/event-stream": {Schema: &huma.Schema{Type: huma.TypeString}},
				},
			},
		}
	}
}

func streamJSONResponse() func(*huma.Operation) {
	return func(op *huma.Operation) {
		if op.Responses == nil {
			op.Responses = map[string]*huma.Response{}
		}
		resp := op.Responses["200"]
		if resp == nil {
			resp = &huma.Response{Description: "OK"}
			op.Responses["200"] = resp
		}
		if resp.Content == nil {
			resp.Content = map[string]*huma.MediaType{}
		}
		resp.Content["application/json"] = &huma.MediaType{
			Schema: &huma.Schema{Type: huma.TypeObject},
		}
	}
}

func (s *Server) humaTimeout() func(*huma.Operation) {
	return func(op *huma.Operation) {
		op.Middlewares = append(op.Middlewares, func(ctx huma.Context, next func(huma.Context)) {
			if errors.Is(ctx.Context().Err(), context.Canceled) {
				return
			}
			if errors.Is(ctx.Context().Err(), context.DeadlineExceeded) {
				next(ctx)
				return
			}
			if s.cfg.WriteTimeout <= 0 {
				if s.handlerDelay > 0 {
					time.Sleep(s.handlerDelay)
				}
				next(ctx)
				return
			}

			req, writer := humago.Unwrap(ctx)
			timeoutHandler := http.TimeoutHandler(
				http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
					if s.handlerDelay > 0 {
						time.Sleep(s.handlerDelay)
					}
					next(huma.WithContext(humago.NewContext(ctx.Operation(), r, w), r.Context()))
				}),
				s.cfg.WriteTimeout,
				`{"error":"request timed out"}`,
			)
			tw := &contentTypeWrapper{
				ResponseWriter: writer,
				contentType:    "application/json",
				triggerStatus:  http.StatusServiceUnavailable,
			}
			timeoutHandler.ServeHTTP(tw, req.WithContext(ctx.Context()))
		})
	}
}

func handleHumaContextError(err error) error {
	if errors.Is(err, context.Canceled) {
		return nil
	}
	if errors.Is(err, context.DeadlineExceeded) {
		return apiError(http.StatusGatewayTimeout, "gateway timeout")
	}
	return nil
}

func handleHumaReadOnly(err error) error {
	if errors.Is(err, db.ErrReadOnly) {
		return apiError(http.StatusNotImplemented, "not available in remote mode")
	}
	return nil
}

func serverError(err error) error {
	if errors.Is(err, context.Canceled) {
		return nil
	}
	if handled := handleHumaContextError(err); handled != nil {
		return handled
	}
	return apiError(http.StatusInternalServerError, err.Error())
}

func internalError(logPrefix string, err error) error {
	if errors.Is(err, context.Canceled) {
		return nil
	}
	if handled := handleHumaContextError(err); handled != nil {
		return handled
	}
	if err != nil {
		log.Printf("%s: %v", logPrefix, err)
	}
	return apiError(http.StatusInternalServerError, "internal error")
}

type idPathInput struct {
	ID string `path:"id" required:"true" doc:"Session ID"`
}

type intIDPathInput struct {
	ID int64 `path:"id" required:"true" doc:"Numeric ID"`
}

type messagePathInput struct {
	ID        string `path:"id" required:"true" doc:"Session ID"`
	MessageID int64  `path:"messageId" required:"true" doc:"Message ordinal"`
}

type BoolIncludeInput struct {
	IncludeOneShot   bool `query:"include_one_shot" doc:"Include one-shot sessions"`
	IncludeAutomated bool `query:"include_automated" doc:"Include automated sessions"`
}

func validateDateFilterValues(date, dateFrom, dateTo, activeSince string) error {
	for _, d := range []string{date, dateFrom, dateTo} {
		if d != "" && !timeutil.IsValidDate(d) {
			return apiError(http.StatusBadRequest, "invalid date format: use YYYY-MM-DD")
		}
	}
	if dateFrom != "" && dateTo != "" && dateFrom > dateTo {
		return apiError(http.StatusBadRequest, "date_from must not be after date_to")
	}
	if activeSince != "" && !timeutil.IsValidTimestamp(activeSince) {
		return apiError(http.StatusBadRequest, "invalid active_since: use RFC3339 timestamp")
	}
	return nil
}

func newHumaSSEStream(ctx huma.Context) (*SSEStream, bool) {
	w, ok := ctx.BodyWriter().(http.ResponseWriter)
	if !ok {
		return nil, false
	}
	f, ok := w.(http.Flusher)
	if !ok {
		return nil, false
	}
	ctx.SetHeader("Content-Type", "text/event-stream")
	ctx.SetHeader("Cache-Control", "no-cache")
	ctx.SetHeader("Connection", "keep-alive")
	f.Flush()
	return &SSEStream{w: w, f: f}, true
}

func writeHumaJSON(ctx huma.Context, status int, value any) {
	ctx.SetHeader("Content-Type", "application/json")
	ctx.SetStatus(status)
	_ = sjson(ctx.BodyWriter(), value)
}

func sjson(w io.Writer, value any) error {
	return json.NewEncoder(w).Encode(value)
}