fluid-core/internal/runtime/gateway.go

package runtime

import (
	"bytes"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"io"
	"net/http"
	"strings"
	"time"

	"github.com/tegwick/fluid-core/internal/contract"
)

// CohortResolver assigns an inbound request to a consumer cohort.
//
// Cohorts should be coarse (FluidAPIStandards.md section 14): granular enough
// to compare populations, never more specific than the analysis requires.
type CohortResolver interface {
	Cohort(*http.Request) (contract.CohortID, string)
}

// StaticCohort assigns everything to one cohort. Useful before cohort analysis
// exists, and for interfaces with a single kind of consumer.
type StaticCohort contract.CohortID

// Cohort implements CohortResolver.
func (s StaticCohort) Cohort(r *http.Request) (contract.CohortID, string) {
	return contract.CohortID(s), consumerRef(r)
}

// HeaderCohort reads the cohort from a request header, falling back to a
// default when absent or unrecognized.
type HeaderCohort struct {
	Header  string
	Known   map[string]contract.CohortID
	Default contract.CohortID
}

// Cohort implements CohortResolver.
func (h HeaderCohort) Cohort(r *http.Request) (contract.CohortID, string) {
	if v := r.Header.Get(h.Header); v != "" {
		if c, ok := h.Known[v]; ok {
			return c, consumerRef(r)
		}
	}
	return h.Default, consumerRef(r)
}

// consumerRef extracts a stable, pseudonymous consumer identity.
//
// It must never be a raw end-user identifier: the telemetry envelope carries
// this value into the evidence store, and Blueprint section 6.2 requires
// pseudonymization there.
func consumerRef(r *http.Request) string {
	if v := r.Header.Get("X-FLUID-Consumer"); v != "" {
		return v
	}
	return ""
}

// GatewayOptions configures the data plane.
type GatewayOptions struct {
	Interface contract.InterfaceID
	Registry  *Registry
	Resolver  *Resolver
	Connector *Connector
	Emitter   *Emitter
	Cohorts   CohortResolver
	Response  ResponsePolicy

	// MaxBodyBytes bounds request size. Zero applies a 1 MiB default.
	MaxBodyBytes int64
	// Validator, when set, checks requests against the revision contract.
	Validator ContractValidator
}

// ContractValidator checks a request against a revision's declared contract.
//
// It is an interface rather than a concrete OpenAPI implementation because
// FluidAPIStandards.md section 4 admits several contract forms, and the gateway
// should not know which one an interface chose.
type ContractValidator interface {
	Validate(rev contract.Revision, r *http.Request, body []byte) error
}

// ValidationError reports a contract violation.
type ValidationError struct {
	Field   string
	Message string
}

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

// Gateway is the deterministic entry point for interface traffic.
//
// It terminates transport, assigns correlation, resolves and routes a revision,
// calls the adapter, and emits telemetry. It does not interpret semantics:
// Blueprint section 5.1 forbids the gateway from inventing them, and section
// 48.1 names an LLM in the request path as an anti-pattern.
type Gateway struct {
	opts GatewayOptions
	now  func() time.Time
}

// NewGateway returns a gateway. Registry, Resolver and Connector are required.
func NewGateway(opts GatewayOptions) (*Gateway, error) {
	if opts.Registry == nil || opts.Resolver == nil || opts.Connector == nil {
		return nil, errors.New("gateway requires a registry, resolver and connector")
	}
	if opts.Cohorts == nil {
		opts.Cohorts = StaticCohort("unclassified")
	}
	if opts.MaxBodyBytes <= 0 {
		opts.MaxBodyBytes = 1 << 20
	}
	return &Gateway{opts: opts, now: time.Now}, nil
}

func (g *Gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	started := g.now()
	correlation := correlationID(r)
	w.Header().Set("X-FLUID-Correlation", correlation)

	cohort, consumer := g.opts.Cohorts.Cohort(r)

	body, err := io.ReadAll(io.LimitReader(r.Body, g.opts.MaxBodyBytes+1))
	if err != nil {
		g.fail(w, r, correlation, cohort, "", ErrorValidation, "could not read request body", "", started)
		return
	}
	if int64(len(body)) > g.opts.MaxBodyBytes {
		g.fail(w, r, correlation, cohort, "", ErrorValidation, "request body exceeds the configured limit", "", started)
		return
	}

	req := Request{
		ExplicitRevision: contract.RevisionID(r.Header.Get("X-FLUID-Revision")),
		BoundRevision:    contract.RevisionID(r.Header.Get("X-FLUID-Bound-Revision")),
		Cohort:           cohort,
		Tenant:           r.Header.Get("X-FLUID-Tenant"),
		ConsumerRef:      consumer,
		CorrelationID:    correlation,
	}

	resolution, err := g.opts.Resolver.Resolve(req)
	if err != nil {
		kind := ErrorUnavailable
		message := "no revision is currently able to serve this request"
		if errors.Is(err, ErrRevisionNotRoutable) || errors.Is(err, ErrUnknownRevision) {
			kind = ErrorValidation
			message = "the requested revision is not available"
		}
		g.fail(w, r, correlation, cohort, "", kind, message, "", started)
		return
	}

	rev, err := g.opts.Registry.Revision(resolution.Revision)
	if err != nil {
		g.fail(w, r, correlation, cohort, resolution.Revision, ErrorUnavailable,
			"the resolved revision is not published", "", started)
		return
	}

	if g.opts.Validator != nil {
		if verr := g.opts.Validator.Validate(rev, r, body); verr != nil {
			field := ""
			var ve *ValidationError
			if errors.As(verr, &ve) {
				field = ve.Field
			}
			g.failWith(w, r, correlation, cohort, rev.ID, ErrorValidation, verr.Error(), field, started, &resolution)
			return
		}
	}

	w.Header().Set("X-FLUID-Revision", string(rev.ID))

	resp, err := g.opts.Connector.Call(r.Context(), rev, r, bytes.NewReader(body))
	if err != nil {
		kind := ErrorUnavailable
		var ue *UpstreamError
		if errors.As(err, &ue) {
			kind = ue.Kind
		}
		g.failWith(w, r, correlation, cohort, rev.ID, kind,
			"the interface could not complete this request", "", started, &resolution)
		return
	}
	defer resp.Body.Close()

	for k, vs := range resp.Header {
		if hopByHop[strings.ToLower(k)] {
			continue
		}
		for _, v := range vs {
			w.Header().Add(k, v)
		}
	}
	w.WriteHeader(resp.StatusCode)
	written, _ := io.Copy(w, resp.Body)

	g.emitRequest(r, correlation, cohort, rev.ID, &resolution, resp.StatusCode, written, int64(len(body)), started, nil)
}

func (g *Gateway) fail(w http.ResponseWriter, r *http.Request, correlation string, cohort contract.CohortID, rev contract.RevisionID, kind ErrorKind, msg, field string, started time.Time) {
	g.failWith(w, r, correlation, cohort, rev, kind, msg, field, started, nil)
}

func (g *Gateway) failWith(w http.ResponseWriter, r *http.Request, correlation string, cohort contract.CohortID, rev contract.RevisionID, kind ErrorKind, msg, field string, started time.Time, res *Resolution) {
	g.opts.Response.WriteError(w, kind, correlation, msg, rev, field)
	status := statusFor[kind]
	if status == 0 {
		status = http.StatusInternalServerError
	}
	g.emitRequest(r, correlation, cohort, rev, res, status, 0, 0, started, &kind)
}

// emitRequest records what happened. Errors are evidence, not noise:
// FluidAPIStandards.md principle 3 treats them as product signals.
func (g *Gateway) emitRequest(r *http.Request, correlation string, cohort contract.CohortID, rev contract.RevisionID, res *Resolution, status int, respBytes, reqBytes int64, started time.Time, errKind *ErrorKind) {
	if g.opts.Emitter == nil {
		return
	}

	latency := float64(g.now().Sub(started).Microseconds()) / 1000.0
	statusCode := int64(status)
	ev := contract.FluidTelemetry{
		SchemaVersion: "0.1",
		ID:            newID("tl-"),
		OccurredAt:    g.now().UTC(),
		InterfaceID:   g.opts.Interface,
		Kind:          contract.FluidTelemetryKindRequest,
		CorrelationID: correlation,
		ConsumerRef:   r.Header.Get("X-FLUID-Consumer"),
		Cohort:        &cohort,
		Revision:      &rev,
		Request: &contract.FluidTelemetryRequest{
			Route:         r.URL.Path,
			Method:        r.Method,
			Status:        &statusCode,
			LatencyMS:     &latency,
			RequestBytes:  &reqBytes,
			ResponseBytes: &respBytes,
		},
		Redaction: &contract.FluidTelemetryRedaction{Applied: false},
	}

	if res != nil {
		ev.Resolution = &contract.FluidTelemetryResolution{
			Reason:           res.Reason,
			PolicyGeneration: &res.PolicyGeneration,
		}
		ev.Experiment = res.Experiment
	}

	if errKind != nil {
		ev.Kind = contract.FluidTelemetryKindError
		class := contract.FluidTelemetryErrorClass(*errKind)
		if class.Valid() {
			ev.Error = &contract.FluidTelemetryError{Class: class}
		}
	}

	g.opts.Emitter.Emit(ev)
}

// correlationID reuses an inbound correlation reference when the consumer
// supplied one, so a call chain stays linked across services.
func correlationID(r *http.Request) string {
	if v := r.Header.Get("X-FLUID-Correlation"); v != "" {
		return v
	}
	return newID("c-")
}

func newID(prefix string) string {
	var b [12]byte
	if _, err := rand.Read(b[:]); err != nil {
		return prefix + "0"
	}
	return prefix + hex.EncodeToString(b[:])
}