multica/server/internal/handler/cloud_billing.go

package handler

import (
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"strings"

	"github.com/go-chi/chi/v5"
	"github.com/multica-ai/multica/server/internal/cloudruntime"
	"github.com/multica-ai/multica/server/internal/featureflags"
	"github.com/multica-ai/multica/server/internal/util"
)

// Cloud billing endpoints proxy to the same multica-cloud HTTP service
// that backs cloud-runtime (Fleet and Billing share `:8080` per the
// upstream README). All paths here forward verbatim to /api/v1/billing/*
// on the cloud side, mirroring the cloud-runtime handler shape:
//
//   - Owner-credit endpoints sit under /api/cloud-billing/* in our router.
//     Workspace subscription endpoints sit under /api/cloud-subscriptions/*
//     and add membership / role checks plus an authoritative workspace ID.
//     Both require regular Auth and inject the resolved user_id as `X-User-ID`.
//
//   - The Stripe webhook is the one outlier: it lives at
//     /api/webhooks/stripe (outside the Auth group), takes the raw
//     request body byte-for-byte, and forwards `Stripe-Signature` for
//     the cloud side to verify against its STRIPE_WEBHOOK_SECRET. The
//     upstream contract is explicit about this:
//     "webhook 使用原始请求体进行签名校验，不要在反向代理里改写 body."
//
// Owner-credit paths share `proxyCloudRuntime`. Workspace subscriptions use a
// small stricter wrapper to inject workspace_id and allowlist Idempotency-Key;
// the webhook remains the only raw-body forwarder.

// maxStripeWebhookBodySize bounds the raw body we'll forward upstream.
// Stripe's documented event payload upper bound is well under this;
// the cap exists to keep a malicious sender from making us read
// arbitrary memory before the upstream gets to reject the signature.
const maxStripeWebhookBodySize = 1 << 20 // 1 MiB

// stripeSignatureHeader is the canonical name of the header Stripe
// uses to ship its HMAC over the raw body. We forward whatever the
// client sent verbatim; the cloud side is the one that knows the
// shared secret and rejects on mismatch.
const stripeSignatureHeader = "Stripe-Signature"

const idempotencyKeyHeader = "Idempotency-Key"
const maxCloudSubscriptionIdempotencyKeyLength = 255

type cloudSubscriptionCheckoutRequest struct {
	Interval       string `json:"interval"`
	IdempotencyKey string `json:"idempotency_key,omitempty"`
	CustomerEmail  string `json:"customer_email,omitempty"`
}

// This is an intentional allowlist: additive cloud fields are not forwarded
// until the main repository explicitly reviews and adds them here.
type cloudSubscriptionCheckoutUpstreamRequest struct {
	WorkspaceID    string `json:"workspace_id"`
	Interval       string `json:"interval"`
	IdempotencyKey string `json:"idempotency_key,omitempty"`
	CustomerEmail  string `json:"customer_email,omitempty"`
}

// requireCloudSubscriptionWorkspace is the handler-level backstop behind the
// router's RequireWorkspaceMember / RequireWorkspaceRole middleware. Keeping
// the check here makes a future route refactor fail closed instead of exposing
// a workspace billing write without its tenant/role guard.
func (h *Handler) requireCloudSubscriptionWorkspace(w http.ResponseWriter, r *http.Request, roles ...string) (string, string, bool) {
	if isMachineCredentialActor(r) {
		writeError(w, http.StatusForbidden, "this endpoint is only available to human actors")
		return "", "", false
	}
	if !featureflags.BillingWorkspaceSubscriptionsEnabled(r.Context(), h.FeatureFlags) {
		writeError(w, http.StatusServiceUnavailable, "workspace subscriptions are not enabled")
		return "", "", false
	}

	workspaceID := ctxWorkspaceID(r.Context())
	workspaceUUID, err := util.ParseUUID(workspaceID)
	if err != nil {
		writeError(w, http.StatusBadRequest, "workspace_id or workspace_slug is required")
		return "", "", false
	}
	member, ok := ctxMember(r.Context())
	if !ok {
		writeError(w, http.StatusForbidden, "workspace membership required")
		return "", "", false
	}
	if len(roles) > 0 && !roleAllowed(member.Role, roles...) {
		writeError(w, http.StatusForbidden, "insufficient permissions")
		return "", "", false
	}
	userID, ok := requireUserID(w, r)
	if !ok {
		return "", "", false
	}
	return util.UUIDToString(workspaceUUID), userID, true
}

func (h *Handler) proxyCloudSubscription(w http.ResponseWriter, r *http.Request, method, path, userID string, body []byte, headers http.Header) {
	if h.CloudRuntime == nil || !h.CloudRuntime.Enabled() {
		writeError(w, http.StatusServiceUnavailable, "cloud runtime is not configured")
		return
	}
	resp, err := h.CloudRuntime.Do(r.Context(), cloudruntime.Request{
		Method:    method,
		Path:      path,
		Body:      body,
		UserID:    userID,
		RequestID: cloudRuntimeRequestID(r),
		Op:        "billing",
		Headers:   headers,
	})
	if err != nil {
		writeCloudRuntimeError(w, r, err)
		return
	}
	writeCloudRuntimeResponse(w, resp)
}

func cloudSubscriptionIdempotencyHeaders(r *http.Request) http.Header {
	key := strings.TrimSpace(r.Header.Get(idempotencyKeyHeader))
	if key == "" {
		return nil
	}
	return http.Header{idempotencyKeyHeader: []string{key}}
}

func requireCloudSubscriptionIdempotencyKey(w http.ResponseWriter, r *http.Request, bodyKey string) (string, bool) {
	key := strings.TrimSpace(bodyKey)
	if key == "" {
		key = strings.TrimSpace(r.Header.Get(idempotencyKeyHeader))
	}
	if key == "" {
		writeError(w, http.StatusBadRequest, "Idempotency-Key or idempotency_key is required")
		return "", false
	}
	if len(key) > maxCloudSubscriptionIdempotencyKeyLength {
		writeError(w, http.StatusBadRequest, "idempotency key must be at most 255 bytes")
		return "", false
	}
	return key, true
}

// GetCloudWorkspaceEntitlements forwards the active workspace to cloud's
// resolved entitlement endpoint. Any workspace member may read this snapshot;
// cloud independently verifies the stamped X-User-ID against its read-only
// product database.
func (h *Handler) GetCloudWorkspaceEntitlements(w http.ResponseWriter, r *http.Request) {
	workspaceID, userID, ok := h.requireCloudSubscriptionWorkspace(w, r)
	if !ok {
		return
	}
	h.proxyCloudSubscription(w, r, http.MethodGet, "/api/v1/entitlements/"+workspaceID, userID, nil, nil)
}

// GetCloudWorkspaceSubscriptionSummary forwards cloud's Billing-page read:
// the resolved entitlement plus local subscription facts (seats, interval,
// cancellation and grace state, Portal availability). Member-readable like
// entitlements; cloud re-verifies membership against its product database.
//
// Cloud serves this from its own database without calling Stripe, so this stays
// available while Stripe is degraded. A cloud that predates the endpoint
// answers 404 and the client treats the page as unavailable — never as Free.
func (h *Handler) GetCloudWorkspaceSubscriptionSummary(w http.ResponseWriter, r *http.Request) {
	workspaceID, userID, ok := h.requireCloudSubscriptionWorkspace(w, r)
	if !ok {
		return
	}
	h.proxyCloudSubscription(w, r, http.MethodGet, "/api/v1/subscriptions/"+workspaceID+"/summary", userID, nil, nil)
}

// GetCloudWorkspaceSubscriptionPrices forwards cloud's validated per-seat price
// read so the client never hardcodes an amount that Stripe can change.
//
// Member-readable: seeing what Pro costs is not a privileged action, and cloud
// scopes the read to a workspace this caller belongs to. Cloud returns only
// currency, minor-unit amount and interval — Price and Product IDs stay server
// side — and answers 503 when a configured Price cannot be validated, which the
// client renders as "price confirmed at Checkout" rather than a guessed number.
func (h *Handler) GetCloudWorkspaceSubscriptionPrices(w http.ResponseWriter, r *http.Request) {
	workspaceID, userID, ok := h.requireCloudSubscriptionWorkspace(w, r)
	if !ok {
		return
	}
	h.proxyCloudSubscription(w, r, http.MethodGet, "/api/v1/subscriptions/"+workspaceID+"/prices", userID, nil, nil)
}

// CreateCloudWorkspaceSubscriptionCheckout injects the middleware-resolved
// workspace into the upstream body. A caller cannot use a valid role in one
// workspace to smuggle a different workspace_id through JSON.
func (h *Handler) CreateCloudWorkspaceSubscriptionCheckout(w http.ResponseWriter, r *http.Request) {
	workspaceID, userID, ok := h.requireCloudSubscriptionWorkspace(w, r, "owner", "admin")
	if !ok {
		return
	}
	body, ok := readCloudRuntimeJSONBody(w, r)
	if !ok {
		return
	}
	var in cloudSubscriptionCheckoutRequest
	if err := json.Unmarshal(body, &in); err != nil {
		writeError(w, http.StatusBadRequest, "invalid request body")
		return
	}
	if in.Interval != "month" && in.Interval != "year" {
		writeError(w, http.StatusBadRequest, "interval must be month or year")
		return
	}
	idempotencyKey, ok := requireCloudSubscriptionIdempotencyKey(w, r, in.IdempotencyKey)
	if !ok {
		return
	}
	upstreamBody, err := json.Marshal(cloudSubscriptionCheckoutUpstreamRequest{
		WorkspaceID:    workspaceID,
		Interval:       in.Interval,
		IdempotencyKey: idempotencyKey,
		CustomerEmail:  in.CustomerEmail,
	})
	if err != nil {
		writeError(w, http.StatusInternalServerError, "failed to build subscription request")
		return
	}
	h.proxyCloudSubscription(w, r, http.MethodPost, "/api/v1/subscriptions/checkout-sessions", userID, upstreamBody, cloudSubscriptionIdempotencyHeaders(r))
}

// ReconcileCloudWorkspaceSubscriptionSeats is a hint only: neither this
// handler nor its caller supplies a seat quantity. Cloud re-counts human
// members from its least-privilege product-database connection.
func (h *Handler) ReconcileCloudWorkspaceSubscriptionSeats(w http.ResponseWriter, r *http.Request) {
	workspaceID, userID, ok := h.requireCloudSubscriptionWorkspace(w, r, "owner", "admin")
	if !ok {
		return
	}
	h.proxyCloudSubscription(w, r, http.MethodPost, "/api/v1/subscriptions/"+workspaceID+"/seats/reconcile", userID, nil, nil)
}

func (h *Handler) CreateCloudWorkspaceSubscriptionPortal(w http.ResponseWriter, r *http.Request) {
	workspaceID, userID, ok := h.requireCloudSubscriptionWorkspace(w, r, "owner", "admin")
	if !ok {
		return
	}
	if _, ok := requireCloudSubscriptionIdempotencyKey(w, r, ""); !ok {
		return
	}
	h.proxyCloudSubscription(w, r, http.MethodPost, "/api/v1/subscriptions/"+workspaceID+"/portal-sessions", userID, nil, cloudSubscriptionIdempotencyHeaders(r))
}

// GetCloudBillingBalance forwards GET /api/v1/billing/balance.
//
// Returns the caller's wallet balance. Cloud reads `X-User-ID`; we
// stamp it from the authenticated context.
func (h *Handler) GetCloudBillingBalance(w http.ResponseWriter, r *http.Request) {
	h.proxyCloudRuntime(w, r, http.MethodGet, "/api/v1/billing/balance", cloudRuntimeProxyOptions{
		withUserID: true,
	})
}

// ListCloudBillingTransactions forwards GET /api/v1/billing/transactions.
//
// The upstream supports `page` / `page_size`; we forward the query
// string unchanged.
func (h *Handler) ListCloudBillingTransactions(w http.ResponseWriter, r *http.Request) {
	h.proxyCloudRuntime(w, r, http.MethodGet, "/api/v1/billing/transactions", cloudRuntimeProxyOptions{
		withUserID: true,
		withQuery:  true,
	})
}

// ListCloudBillingBatches forwards GET /api/v1/billing/batches.
//
// Returns paginated topup / bonus batches for the owner; same query
// shape as transactions.
func (h *Handler) ListCloudBillingBatches(w http.ResponseWriter, r *http.Request) {
	h.proxyCloudRuntime(w, r, http.MethodGet, "/api/v1/billing/batches", cloudRuntimeProxyOptions{
		withUserID: true,
		withQuery:  true,
	})
}

// ListCloudBillingTopups forwards GET /api/v1/billing/topups.
func (h *Handler) ListCloudBillingTopups(w http.ResponseWriter, r *http.Request) {
	h.proxyCloudRuntime(w, r, http.MethodGet, "/api/v1/billing/topups", cloudRuntimeProxyOptions{
		withUserID: true,
		withQuery:  true,
	})
}

// ListCloudBillingPriceTiers forwards GET /api/v1/billing/price-tiers.
//
// Per the upstream doc, this endpoint requires `X-User-ID` (it sits
// under the same auth fence as the rest of /api/v1/billing/*), even
// though the response is the same for every owner today. We stamp the
// header so cloud can audit who's listing tiers — and so the contract
// stays uniform if pricing later differentiates per-customer.
func (h *Handler) ListCloudBillingPriceTiers(w http.ResponseWriter, r *http.Request) {
	h.proxyCloudRuntime(w, r, http.MethodGet, "/api/v1/billing/price-tiers", cloudRuntimeProxyOptions{
		withUserID: true,
	})
}

// CreateCloudBillingCheckoutSession forwards POST /api/v1/billing/checkout-sessions.
//
// Body shape (per upstream): `{tier_id, customer_email?}`. We don't
// care about its contents — proxyCloudRuntime validates only that
// it's syntactically JSON and forwards the bytes.
func (h *Handler) CreateCloudBillingCheckoutSession(w http.ResponseWriter, r *http.Request) {
	h.proxyCloudRuntime(w, r, http.MethodPost, "/api/v1/billing/checkout-sessions", cloudRuntimeProxyOptions{
		withUserID: true,
		withBody:   true,
	})
}

// GetCloudBillingCheckoutSession forwards GET /api/v1/billing/checkout-sessions/{session_id}.
//
// The path param goes into the upstream URL; if missing we return 400
// before doing any network work.
func (h *Handler) GetCloudBillingCheckoutSession(w http.ResponseWriter, r *http.Request) {
	sessionID := chi.URLParam(r, "sessionId")
	if sessionID == "" {
		writeError(w, http.StatusBadRequest, "session_id is required")
		return
	}
	// Stripe Checkout session IDs are `cs_<base62>` by construction —
	// alphanumeric plus underscore, nothing else. We splice the value
	// straight into the upstream URL path, so anything outside that
	// alphabet could in principle introduce path / query / fragment
	// segments that re-target the request. Allow-list rather than
	// deny-list: a future Stripe-format bump cannot quietly slip a
	// new "harmful" character past us, it just rejects until we
	// widen the rule consciously.
	if !isValidStripeSessionID(sessionID) {
		writeError(w, http.StatusBadRequest, "invalid session_id")
		return
	}
	h.proxyCloudRuntime(w, r, http.MethodGet, "/api/v1/billing/checkout-sessions/"+sessionID, cloudRuntimeProxyOptions{
		withUserID: true,
	})
}

// isValidStripeSessionID reports whether s is a syntactically
// plausible Stripe Checkout session id: a non-empty string of
// `[A-Za-z0-9_]`. Stripe's own format is `cs_<base62>`, but we
// accept the broader alphanumeric+underscore set because (a) it
// covers every Stripe ID variant that might ever route here and
// (b) it stays in lockstep with Stripe's own ID grammar without
// hard-coding the `cs_` prefix.
func isValidStripeSessionID(s string) bool {
	if s == "" {
		return false
	}
	for _, c := range s {
		switch {
		case c >= 'a' && c <= 'z':
		case c >= 'A' && c <= 'Z':
		case c >= '0' && c <= '9':
		case c == '_':
		default:
			return false
		}
	}
	return true
}

// CreateCloudBillingPortalSession forwards POST /api/v1/billing/portal-sessions.
//
// Body shape is upstream-defined; can be empty. We treat it as
// optional JSON: cloud_runtime helper rejects empty bodies on
// withBody=true, so for portal-sessions we explicitly do NOT mark
// withBody and also send no body upstream. If the upstream contract
// later requires a body, switch this to withBody and let cloud
// validate.
func (h *Handler) CreateCloudBillingPortalSession(w http.ResponseWriter, r *http.Request) {
	h.proxyCloudRuntime(w, r, http.MethodPost, "/api/v1/billing/portal-sessions", cloudRuntimeProxyOptions{
		withUserID: true,
	})
}

// HandleCloudBillingStripeWebhook is the public ingress for Stripe
// webhook deliveries. Three things are critical here and are *not*
// shared with the standard proxyCloudRuntime path:
//
//  1. NO authentication. Stripe POSTs from its own infrastructure;
//     we don't have a user context and don't try to invent one.
//     Application-layer auth is replaced by Stripe's HMAC signature,
//     which the upstream cloud service verifies. This route therefore
//     sits OUTSIDE the Auth group in router.go.
//
//  2. The body is forwarded byte-for-byte. Stripe's signature is
//     computed over the exact bytes it sent. We must NOT json.Unmarshal
//     /re-marshal, trim whitespace, or otherwise touch the payload —
//     even a single byte difference fails verification.
//
//  3. The `Stripe-Signature` header is forwarded verbatim. That's
//     the entire authentication channel from Stripe to the upstream.
//
// We DO bound the read with MaxBytesReader so a malicious or
// misconfigured sender can't make us buffer arbitrary memory before
// the upstream rejects on signature.
//
// Two pre-checks happen BEFORE we read the body:
//
//   - Per-IP rate limit (mirrors HandleAutopilotWebhook). The
//     endpoint is public; without rate limiting an attacker spraying
//     bogus payloads forces an upstream round-trip per request and
//     burns the cloud's webhook handling budget. A spray of bad
//     signatures still counts toward this limit, so the fast-path
//     429 stops the bleed.
//
//   - Mandatory `Stripe-Signature` header. Real Stripe deliveries
//     always include it; absence ≡ not from Stripe. Returning 401
//     locally saves an upstream RTT on the most common kind of
//     unauthorized poke (curl from a script kiddie). This is a strict
//     superset of what the upstream would do — Cloud also rejects
//     missing-signature with 401 — so it does not change Stripe's
//     own delivery dashboard view.
func (h *Handler) HandleCloudBillingStripeWebhook(w http.ResponseWriter, r *http.Request) {
	if h.CloudRuntime == nil || !h.CloudRuntime.Enabled() {
		writeError(w, http.StatusServiceUnavailable, "cloud runtime is not configured")
		return
	}

	// Per-IP rate limit BEFORE reading the body or hitting upstream.
	// We deliberately reuse the same limiter as the autopilot
	// webhook: both are public unauthenticated ingress with the same
	// abuse profile, and budgeting them together gives a single knob
	// to tune.
	if h.WebhookIPRateLimiter != nil {
		if ip := h.clientIPForRateLimit(r); ip != "" {
			if !h.WebhookIPRateLimiter.Allow(r.Context(), ip) {
				writeError(w, http.StatusTooManyRequests, "rate limit exceeded")
				return
			}
		}
	}

	// Real Stripe deliveries always include Stripe-Signature; the
	// absence is a confident "not from Stripe" signal. Reject locally
	// to save the upstream RTT. NOTE: we use Header.Values to detect
	// presence rather than Get, so a header explicitly set to "" still
	// counts as missing (matches the upstream's interpretation).
	if len(r.Header.Values(stripeSignatureHeader)) == 0 {
		writeError(w, http.StatusUnauthorized, "missing Stripe-Signature header")
		return
	}

	// Body cap matches the JSON proxy. Stripe's documented payload
	// ceiling is much smaller; the limit is defense, not contract.
	r.Body = http.MaxBytesReader(w, r.Body, maxStripeWebhookBodySize)
	// io.ReadAll is appropriate here because the webhook body is
	// fully consumed before forwarding (Stripe signs the bytes; we
	// can't stream). Unlike the JSON proxy we deliberately do NOT
	// trim whitespace or json-validate — the upstream signature
	// check is computed over exactly what we received, so any
	// transformation here would silently break verification.
	body, err := io.ReadAll(r.Body)
	if err != nil {
		var maxErr *http.MaxBytesError
		if errors.As(err, &maxErr) {
			writeError(w, http.StatusRequestEntityTooLarge, "request body is too large")
			return
		}
		writeError(w, http.StatusBadRequest, "invalid request body")
		return
	}

	// Forward Stripe-Signature verbatim, plus Stripe's original
	// Content-Type (so the upstream sees the exact same header set
	// it would if Stripe were calling it directly). cloudruntime's
	// default Content-Type would override `application/json;
	// charset=utf-8` to plain `application/json`; signature
	// verification doesn't care (HMAC is over the body), but
	// preserving the exact header is cheap and removes a debug-time
	// "why does this header look different?" surprise. Caller-
	// supplied Headers in cloudruntime.Request override defaults, so
	// putting Content-Type here is enough.
	headers := http.Header{}
	if sigs := r.Header.Values(stripeSignatureHeader); len(sigs) > 0 {
		headers[stripeSignatureHeader] = sigs
	}
	if cts := r.Header.Values("Content-Type"); len(cts) > 0 {
		headers["Content-Type"] = cts
	}

	resp, err := h.CloudRuntime.Do(r.Context(), cloudruntime.Request{
		Method:    http.MethodPost,
		Path:      "/api/v1/webhooks/stripe",
		Body:      body,
		Headers:   headers,
		RequestID: cloudRuntimeRequestID(r),
		Op:        "billing",
		// Intentionally no UserID — webhook is unauthenticated by
		// design; injecting an empty header would still be harmless,
		// but staying explicit makes the contract obvious to readers.
	})
	if err != nil {
		writeCloudRuntimeError(w, r, err)
		return
	}
	writeCloudRuntimeResponse(w, resp)
}