knowledge-governance/internal/objectgateway/http.go

package objectgateway

import (
	"bytes"
	"context"
	"crypto/subtle"
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"
)

type ExecuteRequest struct {
	Authorization *Grant  `json:"authorization,omitempty"`
	Request       Request `json:"request"`
	Body          []byte  `json:"body,omitempty"`
}
type AdvanceRequest struct {
	Expected int64 `json:"expected"`
	Grant    Grant `json:"grant"`
}
type Capabilities struct {
	AutomaticGrants bool   `json:"automatic_grants"`
	Protocol        string `json:"protocol"`
	Scope           string `json:"scope"`
	Ready           bool   `json:"ready"`
}
type HTTPServer struct {
	Gateway                 *Gateway
	DataToken, ControlToken string
}

func (s HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	token := s.DataToken
	if r.URL.Path == "/v1/advance" || r.URL.Path == "/v1/plans" {
		token = s.ControlToken
	}
	if len(token) < 32 || s.DataToken == s.ControlToken || subtle.ConstantTimeCompare([]byte(r.Header.Get("Authorization")), []byte("Bearer "+token)) != 1 {
		http.Error(w, "gateway_unauthorized", 401)
		return
	}
	if r.Header.Get("X-KG-Gateway-Protocol") != Protocol || r.Header.Get("X-KG-Gateway-Scope") != s.Gateway.Scope {
		http.Error(w, "gateway_protocol_or_target_mismatch", 409)
		return
	}
	fail := func(e error) {
		code := 503
		if errors.Is(e, ErrEpoch) || errors.Is(e, ErrConflict) || errors.Is(e, ErrHeld) {
			code = 409
		}
		http.Error(w, e.Error(), code)
	}
	decode := func(v any) bool {
		r.Body = http.MaxBytesReader(w, r.Body, s.Gateway.MaxBytes*2+16384)
		d := json.NewDecoder(r.Body)
		d.DisallowUnknownFields()
		if d.Decode(v) != nil || d.Decode(&struct{}{}) != io.EOF {
			http.Error(w, "gateway_bad_request", 400)
			return false
		}
		return true
	}
	if r.Method == "GET" && r.URL.Path == "/v1/capabilities" {
		if e := s.Gateway.Ready(r.Context()); e != nil {
			fail(e)
			return
		}
		if authority, ok := s.Gateway.Authority.(interface{ Ready(context.Context) error }); ok {
			if e := authority.Ready(r.Context()); e != nil {
				fail(e)
				return
			}
		}
		_ = json.NewEncoder(w).Encode(Capabilities{Protocol: Protocol, Scope: s.Gateway.Scope, Ready: true, AutomaticGrants: s.Gateway.Authority != nil})
		return
	}
	if r.Method != "POST" {
		http.Error(w, "gateway_route_not_found", 404)
		return
	}
	switch r.URL.Path {
	case "/v1/acquire":
		var in Claim
		if !decode(&in) {
			return
		}
		out, e := s.Gateway.Acquire(r.Context(), in)
		if e != nil {
			fail(e)
			return
		}
		_ = json.NewEncoder(w).Encode(out)
	case "/v1/plans":
		if s.Gateway.Authority != nil {
			fail(ErrConflict)
			return
		}
		var in Plan
		if !decode(&in) {
			return
		}
		if e := s.Gateway.InstallPlan(r.Context(), in); e != nil {
			fail(e)
			return
		}
		_ = json.NewEncoder(w).Encode(in)
	case "/v1/advance":
		if s.Gateway.Authority != nil {
			fail(ErrConflict)
			return
		}
		var in AdvanceRequest
		if !decode(&in) {
			return
		}
		if e := s.Gateway.Advance(r.Context(), in.Expected, in.Grant); e != nil {
			fail(e)
			return
		}
		_ = json.NewEncoder(w).Encode(in)
	case "/v1/execute", "/v1/reconcile":
		var in ExecuteRequest
		if !decode(&in) {
			return
		}
		var out Receipt
		var e error
		if r.URL.Path == "/v1/execute" {
			if in.Authorization != nil {
				out, e = s.Gateway.ExecuteAuthorized(r.Context(), *in.Authorization, in.Request, in.Body)
			} else {
				out, e = s.Gateway.Execute(r.Context(), in.Request, in.Body)
			}
		} else {
			out, e = s.Gateway.Reconcile(r.Context(), in.Request)
		}
		if e != nil {
			fail(e)
			return
		}
		_ = json.NewEncoder(w).Encode(out)
	case "/v1/resolve":
		var in Request
		if !decode(&in) {
			return
		}
		out, e := s.Gateway.Resolve(r.Context(), in)
		if e != nil {
			fail(e)
			return
		}
		_ = json.NewEncoder(w).Encode(out)
	case "/v1/observe":
		var in struct {
			Key     string `json:"key"`
			Version string `json:"version"`
		}
		if !decode(&in) {
			return
		}
		if in.Key == "" {
			http.Error(w, "gateway_bad_key", 400)
			return
		}
		if e := s.Gateway.Ready(r.Context()); e != nil {
			fail(e)
			return
		}
		v, exists, e := s.Gateway.Backend.Observe(r.Context(), in.Key, in.Version)
		if e != nil {
			http.Error(w, "gateway_backend_read_failed", 503)
			return
		}
		_ = json.NewEncoder(w).Encode(struct {
			Version Version `json:"version"`
			Exists  bool    `json:"exists"`
		}{v, exists})
	default:
		http.Error(w, "gateway_route_not_found", 404)
	}
}

type Client struct {
	Endpoint, Token, Scope string
	HTTP                   *http.Client
}

func NewClient(endpoint, token, scope string) (*Client, error) {
	u, e := url.Parse(endpoint)
	if e != nil || u.Host == "" || u.User != nil || u.RawQuery != "" || u.Fragment != "" || u.Path != "" || len(token) < 32 || scope == "" {
		return nil, errors.New("gateway_configuration_invalid")
	}
	if u.Scheme != "https" && (u.Scheme != "http" || (u.Hostname() != "127.0.0.1" && u.Hostname() != "::1")) {
		return nil, errors.New("gateway_transport_requires_tls")
	}
	return &Client{endpoint, token, scope, &http.Client{Timeout: 2 * time.Minute, CheckRedirect: func(*http.Request, []*http.Request) error { return errors.New("gateway_redirect_refused") }}}, nil
}
func (c *Client) call(ctx context.Context, path string, in, out any) error {
	method := "GET"
	var body io.Reader
	if in != nil {
		raw, e := json.Marshal(in)
		if e != nil {
			return ErrConflict
		}
		method = "POST"
		body = struct{ io.Reader }{bytes.NewReader(raw)}
	}
	req, e := http.NewRequestWithContext(ctx, method, c.Endpoint+path, body)
	if e != nil {
		return errors.New("gateway_request_invalid")
	}
	req.Header.Set("Authorization", "Bearer "+c.Token)
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("X-KG-Gateway-Protocol", Protocol)
	req.Header.Set("X-KG-Gateway-Scope", c.Scope)
	resp, e := c.HTTP.Do(req)
	if e != nil {
		return errors.New("gateway_unreachable_or_outcome_unknown")
	}
	defer resp.Body.Close()
	if resp.StatusCode != 200 {
		raw, _ := io.ReadAll(io.LimitReader(resp.Body, 128))
		reason := strings.TrimSpace(string(raw))
		switch reason {
		case ErrAuthority.Error():
			return ErrAuthority
		case ErrAuthorityPrivileges.Error():
			return ErrAuthorityPrivileges
		case ErrEpoch.Error():
			return ErrEpoch
		case ErrConflict.Error():
			return ErrConflict
		case ErrHeld.Error():
			return ErrHeld
		case ErrPersistence.Error():
			return ErrPersistence
		case ErrIsolation.Error():
			return ErrIsolation
		}
		return errors.New("gateway_request_refused")
	}
	if json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(out) != nil {
		return errors.New("gateway_response_invalid")
	}
	return nil
}
func (c *Client) Ready(ctx context.Context) error               { return c.checkReady(ctx, false) }
func (c *Client) ReadyForAcquisition(ctx context.Context) error { return c.checkReady(ctx, true) }
func (c *Client) checkReady(ctx context.Context, automatic bool) error {
	var caps Capabilities
	if e := c.call(ctx, "/v1/capabilities", nil, &caps); e != nil {
		return e
	}
	if automatic && !caps.AutomaticGrants {
		return ErrAuthority
	}
	if caps.Protocol != Protocol || caps.Scope != c.Scope || !caps.Ready {
		return errors.New("gateway_protocol_or_target_mismatch")
	}
	return nil
}
func (c *Client) InstallPlan(ctx context.Context, p Plan) error {
	var out Plan
	if e := c.call(ctx, "/v1/plans", p, &out); e != nil {
		return e
	}
	want, _ := json.Marshal(p)
	got, _ := json.Marshal(out)
	if string(want) != string(got) {
		return ErrConflict
	}
	return nil
}
func (c *Client) Advance(ctx context.Context, expected int64, g Grant) error {
	var out AdvanceRequest
	in := AdvanceRequest{expected, g}
	if e := c.call(ctx, "/v1/advance", in, &out); e != nil {
		return e
	}
	if in != out {
		return ErrConflict
	}
	return nil
}
func (c *Client) Execute(ctx context.Context, r Request, body []byte) (Receipt, error) {
	var out Receipt
	e := c.call(ctx, "/v1/execute", ExecuteRequest{Request: r, Body: body}, &out)
	if e == nil && (out.Request != r || !out.valid()) {
		e = ErrConflict
	}
	return out, e
}
func (c *Client) Reconcile(ctx context.Context, r Request) (Receipt, error) {
	var out Receipt
	e := c.call(ctx, "/v1/reconcile", ExecuteRequest{Request: r}, &out)
	if e == nil && (out.Request != r || !out.valid()) {
		e = ErrConflict
	}
	return out, e
}
func (c *Client) Observe(ctx context.Context, key, id string) (Version, bool, error) {
	var out struct {
		Version Version `json:"version"`
		Exists  bool    `json:"exists"`
	}
	e := c.call(ctx, "/v1/observe", struct {
		Key     string `json:"key"`
		Version string `json:"version"`
	}{key, id}, &out)
	return out.Version, out.Exists, e
}