homelab-frontend/internal/serviceadapter/router.go

package serviceadapter

import (
	"context"
	"fmt"
	"net"
	"net/http"
	"net/http/httputil"
	"net/url"
	"strings"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"

	"forgejo.riotpiao.com/rock/homelab-frontend/internal/auth"
	"forgejo.riotpiao.com/rock/homelab-frontend/internal/problem"
)

// Dispatcher routes X-Service requests to upstreams.
// Auth per service:
//   SQS: Gateway validates JWT (kmsvc code unverified)
//   MinIO, Temporal: Native JWT support (dumb pipe pass-through)
//   Memory, IAM: Services validate JWTs themselves
type Dispatcher struct {
	registry   *Registry
	sqsJWTAuth *auth.Validator
}

// NewDispatcher creates a new service adapter dispatcher.
func NewDispatcher(registry *Registry) *Dispatcher {
	// Create JWT validator for SQS
	// Issuer and JWKS URL should match Authentik application config
	sqsValidator := auth.NewValidator(
		"https://authentik.riotpiao.com/application/o/sqs/",
		"sqs",
		"https://authentik.riotpiao.com/application/o/sqs/jwks/",
	)

	return &Dispatcher{
		registry:   registry,
		sqsJWTAuth: sqsValidator,
	}
}

// Matches returns true if the request should be dispatched based on X-Service header.
func (d *Dispatcher) Matches(r *http.Request) bool {
	return r.Header.Get("X-Service") != ""
}

// Dispatch routes a request to the appropriate adapter.
// Returns a problem document if the adapter or resource is not found.
func (d *Dispatcher) Dispatch(w http.ResponseWriter, r *http.Request) {
	serviceName := r.Header.Get("X-Service")
	if serviceName == "" {
		d.writeError(w, problem.BadRequest("X-Service header required"))
		return
	}

	// Look up service adapter
	adapter := d.registry.Get(serviceName)
	if adapter == nil {
		p := problem.NotFound(fmt.Sprintf("service '%s' not found", serviceName))
		_ = p.Write(w)
		return
	}

	// Get resource and method from request
	resourceName := r.Header.Get("X-Resource")
	if resourceName == "" {
		d.writeError(w, problem.BadRequest("X-Resource header required"))
		return
	}

	// Find resource
	var resource *Resource
	for i := range adapter.Spec.Resources {
		if adapter.Spec.Resources[i].Name == resourceName {
			resource = &adapter.Spec.Resources[i]
			break
		}
	}

	if resource == nil {
		p := problem.NotFound(fmt.Sprintf("resource '%s' not found in service '%s'", resourceName, serviceName))
		_ = p.Write(w)
		return
	}

	// Find method matching HTTP verb
	var method *Method
	for i := range resource.Methods {
		if resource.Methods[i].Verb == r.Method {
			method = &resource.Methods[i]
			break
		}
	}

	if method == nil {
		p := problem.NotFound(fmt.Sprintf("method %s not defined for resource '%s'", r.Method, resourceName))
		_ = p.Write(w)
		return
	}

	// Gateway-level JWT validation for SQS (code unverified in kmsvc)
	// MinIO, Temporal, Memory, IAM have native JWT support - pass through
	if adapter.Spec.Auth.Required && serviceName == "sqs" {
		authHeader := r.Header.Get("Authorization")
		if authHeader == "" {
			p := problem.NewProblem(http.StatusForbidden, "about:blank#forbidden",
				"Forbidden", "SQS requires Authorization header")
			_ = p.Write(w)
			return
		}

		// Validate JWT signature against Authentik JWKS
		claims, err := d.sqsJWTAuth.ValidateBearerToken(authHeader)
		if err != nil {
			p := problem.NewProblem(http.StatusForbidden, "about:blank#forbidden",
				"Forbidden", fmt.Sprintf("JWT validation failed: %v", err))
			_ = p.Write(w)
			return
		}

		// Check required permissions (sqs:read or sqs:write or *)
		hasPermission := d.sqsJWTAuth.CheckPermissions(claims, "sqs:read", "sqs:write", "*")
		if !hasPermission {
			p := problem.NewProblem(http.StatusForbidden, "about:blank#forbidden",
				"Forbidden", "Insufficient permissions for SQS")
			_ = p.Write(w)
			return
		}
	}

	// Detect protocol from upstream URL scheme
	upstreamURL := adapter.Spec.Upstream.URL
	if strings.HasPrefix(upstreamURL, "grpc://") {
		// gRPC upstream (Temporal, etc.)
		d.dispatchGRPC(w, r, upstreamURL, method, adapter)
	} else {
		// HTTP upstream (MinIO, Authentik, etc.)
		d.dispatchHTTP(w, r, upstreamURL, method, adapter)
	}
}

// dispatchHTTP forwards HTTP requests to upstream, passing Authorization header through.
func (d *Dispatcher) dispatchHTTP(w http.ResponseWriter, r *http.Request, upstreamURL string, method *Method, adapter *ServiceAdapter) {
	parsedURL, err := url.Parse(upstreamURL)
	if err != nil {
		d.writeError(w, problem.NewProblem(http.StatusInternalServerError, "about:blank#server-error",
			"Internal Server Error", fmt.Sprintf("invalid upstream URL: %v", err)))
		return
	}

	// Create reverse proxy
	proxy := httputil.NewSingleHostReverseProxy(parsedURL)
	proxy.Director = func(req *http.Request) {
		req.URL.Scheme = parsedURL.Scheme
		req.URL.Host = parsedURL.Host
		req.URL.Path = method.UpstreamPath
		req.RequestURI = ""
		req.Host = parsedURL.Host
		// Authorization header passes through unchanged
	}

	// Set timeout
	timeout := adapter.Spec.Upstream.TimeoutSeconds
	if timeout <= 0 {
		timeout = 30
	}
	proxy.Transport = &http.Transport{
		DialContext:         (&net.Dialer{Timeout: time.Duration(timeout) * time.Second}).DialContext,
		TLSHandshakeTimeout: time.Duration(timeout) * time.Second,
	}

	// Forward the request
	proxy.ServeHTTP(w, r)
}

// dispatchGRPC forwards gRPC requests to upstream.
// gRPC URL format: grpc://host:port
func (d *Dispatcher) dispatchGRPC(w http.ResponseWriter, r *http.Request, upstreamURL string, method *Method, adapter *ServiceAdapter) {
	// Extract host:port from grpc://host:port
	host := strings.TrimPrefix(upstreamURL, "grpc://")
	if host == upstreamURL {
		d.writeError(w, problem.NewProblem(http.StatusInternalServerError, "about:blank#server-error",
			"Internal Server Error", "invalid gRPC URL format"))
		return
	}

	// Validate that this is a gRPC request
	if !strings.HasPrefix(r.Header.Get("Content-Type"), "application/grpc") {
		d.writeError(w, problem.NewProblem(http.StatusBadRequest, "about:blank#bad-request",
			"Bad Request", "gRPC service requires application/grpc content-type"))
		return
	}

	// Set timeout
	timeout := adapter.Spec.Upstream.TimeoutSeconds
	if timeout <= 0 {
		timeout = 30
	}

	ctx, cancel := context.WithTimeout(r.Context(), time.Duration(timeout)*time.Second)
	defer cancel()

	// Dial gRPC upstream
	conn, err := grpc.DialContext(ctx, host,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithDefaultCallOptions(
			grpc.MaxCallRecvMsgSize(100 * 1024 * 1024), // 100MB
		),
	)
	if err != nil {
		d.writeError(w, problem.NewProblem(http.StatusBadGateway, "about:blank#bad-gateway",
			"Bad Gateway", fmt.Sprintf("failed to dial gRPC upstream: %v", err)))
		return
	}
	defer conn.Close()

	// Forward gRPC request
	// Note: Full gRPC forwarding requires grpcproxy or custom middleware.
	// For now, return unimplemented (Temporal support coming in Phase 9)
	d.writeError(w, problem.NewProblem(http.StatusNotImplemented, "about:blank#not-implemented",
		"Not Implemented", "gRPC forwarding not yet implemented - use in-cluster gRPC clients directly"))
}

func (d *Dispatcher) writeError(w http.ResponseWriter, p *problem.Problem) {
	_ = p.Write(w)
}