broker/pkg/api/server.go

package api

import (
	"context"
	"encoding/json"
	"net"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/go-chi/chi/v5"
	chimw "github.com/go-chi/chi/v5/middleware"
	"github.com/go-chi/cors"

	"github.com/luxfi/broker/pkg/accounts"
	"github.com/luxfi/broker/pkg/audit"
	"github.com/luxfi/broker/pkg/auth"
	"github.com/luxfi/broker/pkg/funding"
	"github.com/luxfi/broker/pkg/marketdata"
	"github.com/luxfi/broker/pkg/provider"
	"github.com/luxfi/broker/pkg/risk"
	"github.com/luxfi/broker/pkg/router"
	"github.com/luxfi/broker/pkg/types"
	"github.com/luxfi/broker/pkg/webhook"
	"github.com/luxfi/broker/pkg/ws"
)

type Server struct {
	registry                *provider.Registry
	resolver                *accounts.Resolver
	sor                     *router.Router
	riskEng                 *risk.Engine
	auditLog                *audit.Log
	feed                    *marketdata.Feed
	stream                  *ws.Server
	funding                 *funding.Service
	twap                    *router.TWAPScheduler
	arbDetector             *marketdata.ArbitrageDetector
	arbDetectorThresholdBps float64
	webhookStore            webhook.Store
	router                  chi.Router
	server                  *http.Server
}

func NewServer(registry *provider.Registry, listenAddr string) *Server {
	s := &Server{
		registry: registry,
		resolver: accounts.NewResolver(),
		sor:      router.New(registry),
		riskEng:  risk.NewEngine(risk.DefaultLimits()),
		auditLog: audit.NewLog(),
		feed:     marketdata.NewFeed(),
	}
	s.stream = ws.NewServer(s.feed)

	iamEndpoint := os.Getenv("IAM_ENDPOINT")
	if iamEndpoint == "" {
		iamEndpoint = "http://localhost:8000"
	}

	r := chi.NewRouter()
	r.Use(chimw.RequestID)
	r.Use(chimw.RealIP)
	r.Use(chimw.Recoverer)
	r.Use(chimw.Timeout(60 * time.Second))
	r.Use(cors.Handler(cors.Options{
		AllowedOrigins:   corsOriginsFromEnv(),
		AllowedMethods:   []string{"GET", "POST", "PATCH", "DELETE", "OPTIONS"},
		AllowedHeaders:   []string{"Accept", "Authorization", "Content-Type", "X-User-Id", "X-Org-Id", "X-User-Email", "X-IAM-Org"},
		AllowCredentials: true,
		MaxAge:           300,
	}))

	// Public routes — no auth. Webhook handlers validate processor signatures.
	r.Get("/healthz", func(w http.ResponseWriter, r *http.Request) {
		writeJSON(w, http.StatusOK, map[string]interface{}{
			"status":    "ok",
			"providers": registry.List(),
		})
	})
	r.Post("/v1/fund/webhook/{processor}", s.handleFundingWebhook)

	// All remaining routes require IAM JWT auth.
	r.Group(func(r chi.Router) {
		r.Use(auth.Middleware(iamEndpoint))

		r.Route("/v1", func(r chi.Router) {
			r.Get("/providers", s.handleListProviders)
			r.Get("/providers/capabilities", s.handleGetCapabilities)

			r.Get("/accounts", s.handleListAccounts)
			r.Post("/accounts", s.handleCreateAccount)

			// Account-scoped routes require ownership verification.
			r.Route("/accounts/{provider}/{accountId}", func(r chi.Router) {
				r.Use(s.requireAccountOwnership)

				r.Get("/", s.handleGetAccount)
				r.Get("/portfolio", s.handleGetPortfolio)

				r.Get("/orders", s.handleListOrders)
				r.Post("/orders", s.handleCreateOrder)
				r.Delete("/orders", s.handleCancelAllOrders)
				r.Get("/orders/{orderId}", s.handleGetOrder)
				r.Patch("/orders/{orderId}", s.handleReplaceOrder)
				r.Delete("/orders/{orderId}", s.handleCancelOrder)

				r.Get("/positions/{symbol}", s.handleGetPosition)
				r.Delete("/positions/{symbol}", s.handleClosePosition)
				r.Delete("/positions", s.handleCloseAllPositions)

				r.Get("/transfers", s.handleListTransfers)
				r.Post("/transfers", s.handleCreateTransfer)

				r.Get("/bank-relationships", s.handleListBankRelationships)
				r.Post("/bank-relationships", s.handleCreateBankRelationship)

				// Extended Account Management
				r.Patch("/", s.handleUpdateAccount)
				r.Delete("/", s.handleCloseAccount)
				r.Get("/activities", s.handleGetAccountActivities)

				// Documents
				r.Post("/documents", s.handleUploadDocument)
				r.Get("/documents", s.handleListDocuments)
				r.Get("/documents/{documentId}", s.handleGetDocument)
				r.Get("/documents/{documentId}/download", s.handleDownloadDocument)

				// Transfer Extended (cancel, ACH delete, wire banks)
				r.Delete("/transfers/{transferId}", s.handleCancelTransfer)
				r.Delete("/ach-relationships/{achId}", s.handleDeleteACHRelationship)
				r.Post("/recipient-banks", s.handleCreateRecipientBank)
				r.Get("/recipient-banks", s.handleListRecipientBanks)
				r.Delete("/recipient-banks/{bankId}", s.handleDeleteRecipientBank)

				// ACATS Transfers
				r.Post("/acats", s.handleCreateACATSTransfer)
				r.Get("/acats", s.handleListACATSTransfers)
				r.Get("/acats/{transferId}", s.handleGetACATSTransfer)
				r.Delete("/acats/{transferId}", s.handleCancelACATSTransfer)

				// Fixed Income
				r.Post("/fixed-income/orders", s.handleCreateFixedIncomeOrder)
				r.Get("/fixed-income/orders/{orderId}", s.handleGetFixedIncomeOrder)
				r.Delete("/fixed-income/orders/{orderId}", s.handleCancelFixedIncomeOrder)
				r.Get("/fixed-income/positions", s.handleGetFixedIncomePositions)
				r.Delete("/fixed-income/positions/{symbol}", s.handleCloseFixedIncomePosition)

				// Portfolio History
				r.Get("/portfolio/history", s.handleGetPortfolioHistory)

				// Watchlists
				r.Post("/watchlists", s.handleCreateWatchlist)
				r.Get("/watchlists", s.handleListWatchlists)
				r.Get("/watchlists/{watchlistId}", s.handleGetWatchlist)
				r.Put("/watchlists/{watchlistId}", s.handleUpdateWatchlist)
				r.Delete("/watchlists/{watchlistId}", s.handleDeleteWatchlist)
				r.Post("/watchlists/{watchlistId}/assets", s.handleAddWatchlistAsset)
				r.Delete("/watchlists/{watchlistId}/{symbol}", s.handleRemoveWatchlistAsset)

				// Options Trading (account-scoped)
				r.Post("/options/orders", s.handleCreateOptionOrder)
				r.Post("/options/multi-leg", s.handleCreateMultiLegOrder)
				r.Post("/options/exercise", s.handleExerciseOption)
				r.Post("/options/do-not-exercise", s.handleDoNotExercise)
				r.Get("/options/positions", s.handleGetOptionPositions)
			}) // end /accounts/{provider}/{accountId}

			// ACATS disclosure is provider-scoped, not account-scoped.
			r.Get("/accounts/{provider}/acats/disclosure", s.handleGetACATSDisclosure)

			r.Get("/assets/{provider}", s.handleListAssets)
			r.Get("/assets/{provider}/{symbolOrId}", s.handleGetAsset)

			// Smart Order Routing
			r.Get("/route/{symbol}", s.handleGetRoutes)
			r.Get("/route/{symbol}/{quote}", s.handleGetRoutesPair)
			r.Get("/route/{symbol}/split", s.handleGetSplitPlan)
			r.Get("/route/{symbol}/{quote}/split", s.handleGetSplitPlanPair)
			r.Get("/assets", s.handleAggregatedAssets)
			r.Post("/smart-order", s.handleSmartOrder)
			r.Post("/smart-order/split", s.handleExecuteSplit)

			// TWAP Execution
			r.Post("/twap", s.handleStartTWAP)
			r.Get("/twap", s.handleGetTWAP)
			r.Delete("/twap", s.handleCancelTWAP)

			// Arbitrage Detection
			r.Get("/arbitrage", s.handleScanArbitrage)

			// Market Data
			r.Get("/market/{provider}/snapshot/{symbol}", s.handleGetSnapshot)
			r.Get("/market/{provider}/snapshots", s.handleGetSnapshots)
			r.Get("/market/{provider}/bars/{symbol}", s.handleGetBars)
			r.Get("/market/{provider}/trades/latest", s.handleGetLatestTrades)
			r.Get("/market/{provider}/quotes/latest", s.handleGetLatestQuotes)
			r.Get("/market/{provider}/clock", s.handleGetClock)
			r.Get("/market/{provider}/calendar", s.handleGetCalendar)

			// Consolidated Market Data
			r.Get("/bbo/{symbol}", s.handleGetBBO)
			r.Get("/bbo/{symbol}/{quote}", s.handleGetBBOPair)
			r.Get("/stream", s.stream.HandleSSE)

			// Risk & Audit
			r.Get("/risk/check", s.handleRiskCheck)
			r.Get("/audit", s.handleAuditQuery)
			r.Get("/audit/stats", s.handleAuditStats)
			r.Get("/audit/export", s.handleAuditExport)

			// Funding (deposit/withdraw via payment processors)
			// Note: /v1/fund/webhook/{processor} is registered before auth middleware.
			r.Route("/fund", func(r chi.Router) {
				r.Post("/deposit", s.handleDeposit)
				r.Post("/withdraw", s.handleWithdraw)
				r.Get("/processors", s.handleListProcessors)
			})

			// Journals (inter-account transfers)
			r.Post("/journals/{provider}", s.handleCreateJournal)
			r.Get("/journals/{provider}", s.handleListJournals)
			r.Get("/journals/{provider}/{journalId}", s.handleGetJournal)
			r.Delete("/journals/{provider}/{journalId}", s.handleDeleteJournal)
			r.Post("/journals/{provider}/batch", s.handleCreateBatchJournal)
			r.Post("/journals/{provider}/reverse_batch", s.handleReverseBatchJournal)

			// Fixed Income (non-account-scoped)
			r.Get("/assets/fixed-income/corporates", s.handleListCorporateBonds)
			r.Get("/assets/fixed-income/treasuries", s.handleListTreasuryBonds)

			// Crypto Market Data
			r.Get("/market/{provider}/crypto/bars", s.handleGetCryptoBars)
			r.Get("/market/{provider}/crypto/quotes", s.handleGetCryptoQuotes)
			r.Get("/market/{provider}/crypto/trades", s.handleGetCryptoTrades)
			r.Get("/market/{provider}/crypto/snapshots", s.handleGetCryptoSnapshots)

			// Event Streaming (SSE)
			r.Get("/events/{provider}/trades", s.handleStreamTradeEvents)
			r.Get("/events/{provider}/accounts", s.handleStreamAccountEvents)
			r.Get("/events/{provider}/transfers", s.handleStreamTransferEvents)
			r.Get("/events/{provider}/journals", s.handleStreamJournalEvents)

			// Options Trading (non-account-scoped)
			r.Get("/options/{provider}/expirations/{symbol}", s.handleGetOptionExpirations)
			r.Get("/options/{provider}/chain/{symbol}", s.handleGetOptionChain)
			r.Get("/options/{provider}/chain/{symbol}/{expiration}", s.handleGetOptionChain)
			r.Get("/options/{provider}/quote/{contractSymbol}", s.handleGetOptionQuote)
			r.Get("/options/{provider}/greeks/{contractSymbol}", s.handleGetOptionGreeks)
			r.Get("/options/route", s.handleRouteOptionOrder)

			// Exchange frontend API (provider-agnostic, user-resolved)
			r.Route("/exchange", func(r chi.Router) {
				r.Get("/assets", s.handleFrontendAssets)
				r.Get("/crypto-prices", s.handleCryptoPrices)
				r.Get("/charts/{symbol}", s.handleChartData)
				r.Get("/orders", s.handleFrontendOrders)
				r.Post("/orders", s.handleFrontendCreateOrder)
				r.Get("/positions", s.handleFrontendPositions)
				r.Get("/portfolio", s.handleFrontendPortfolio)

				// Options (provider-agnostic, user-resolved)
				r.Route("/options", func(r chi.Router) {
					r.Post("/orders", s.handleExchangeCreateOptionOrder)
					r.Post("/multi-leg", s.handleExchangeCreateMultiLegOrder)
					r.Get("/chain/{symbol}", s.handleExchangeOptionChain)
					r.Get("/expirations/{symbol}", s.handleExchangeOptionExpirations)
					r.Get("/positions", s.handleExchangeOptionPositions)
					r.Post("/orders/{id}/cancel", s.handleExchangeCancelOptionOrder)
					r.Post("/exercise/{id}", s.handleExchangeExerciseOption)
				})
			})
		})
	}) // end auth group

	s.router = r
	s.server = &http.Server{
		Addr:         listenAddr,
		Handler:      r,
		ReadTimeout:  15 * time.Second,
		WriteTimeout: 60 * time.Second,
		IdleTimeout:  120 * time.Second,
	}
	return s
}

func (s *Server) Start() error {
	return s.server.ListenAndServe()
}

func (s *Server) Shutdown(ctx context.Context) error {
	return s.server.Shutdown(ctx)
}

// SetFunding attaches a funding service for deposit/withdraw operations.
func (s *Server) SetFunding(f *funding.Service) {
	s.funding = f
}

// SetTWAP attaches a TWAP scheduler for time-weighted execution.
func (s *Server) SetTWAP(t *router.TWAPScheduler) {
	s.twap = t
}

// SetArbitrageDetector attaches an arbitrage detector for cross-venue scanning.
func (s *Server) SetArbitrageDetector(d *marketdata.ArbitrageDetector, thresholdBps float64) {
	s.arbDetector = d
	s.arbDetectorThresholdBps = thresholdBps
}

// SetWebhookStore attaches a webhook store for event delivery.
// Also mounts webhook management routes at /v1/bd/webhooks.
func (s *Server) SetWebhookStore(store webhook.Store) {
	s.webhookStore = store
	s.router.Route("/v1/bd/webhooks", func(r chi.Router) {
		r.Mount("/", webhook.NewRouter(store))
	})
}

// Handler returns the HTTP handler for testing.
func (s *Server) Handler() http.Handler {
	return s.router
}

// Mount attaches a sub-router at the given pattern on the main router.
func (s *Server) Mount(pattern string, handler http.Handler) {
	s.router.Mount(pattern, handler)
}

// Resolver returns the account resolver for user-to-provider account mapping.
func (s *Server) Resolver() *accounts.Resolver {
	return s.resolver
}

func (s *Server) getProvider(r *http.Request) (provider.Provider, error) {
	return s.registry.Get(chi.URLParam(r, "provider"))
}

// --- Handlers ---

func (s *Server) handleListProviders(w http.ResponseWriter, r *http.Request) {
	writeJSON(w, http.StatusOK, map[string]interface{}{"providers": s.registry.List()})
}

func (s *Server) handleGetCapabilities(w http.ResponseWriter, r *http.Request) {
	writeJSON(w, http.StatusOK, s.sor.GetCapabilities())
}

func (s *Server) handleListAccounts(w http.ResponseWriter, r *http.Request) {
	providerName := r.URL.Query().Get("provider")
	if providerName == "" {
		all := make([]*types.Account, 0)
		for _, name := range s.registry.List() {
			p, _ := s.registry.Get(name)
			accts, err := p.ListAccounts(r.Context())
			if err == nil {
				all = append(all, accts...)
			}
		}
		writeJSON(w, http.StatusOK, all)
		return
	}
	p, err := s.registry.Get(providerName)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	accts, err := p.ListAccounts(r.Context())
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, accts)
}

func (s *Server) handleCreateAccount(w http.ResponseWriter, r *http.Request) {
	var req struct {
		Provider      string   `json:"provider"`
		GivenName     string   `json:"given_name"`
		FamilyName    string   `json:"family_name"`
		DateOfBirth   string   `json:"date_of_birth"`
		TaxID         string   `json:"tax_id"`
		TaxIDType     string   `json:"tax_id_type"`
		CountryOfTax  string   `json:"country_of_tax_residence"`
		Email         string   `json:"email"`
		Phone         string   `json:"phone"`
		Street        []string `json:"street"`
		City          string   `json:"city"`
		State         string   `json:"state"`
		PostalCode    string   `json:"postal_code"`
		Country       string   `json:"country"`
		EnabledAssets []string `json:"enabled_assets"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "invalid request body")
		return
	}
	if req.Provider == "" {
		req.Provider = "alpaca"
	}

	p, err := s.registry.Get(req.Provider)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}

	acct, err := p.CreateAccount(r.Context(), &types.CreateAccountRequest{
		Provider: req.Provider,
		Identity: &types.Identity{
			GivenName: req.GivenName, FamilyName: req.FamilyName,
			DateOfBirth: req.DateOfBirth, TaxID: req.TaxID,
			TaxIDType: req.TaxIDType, CountryOfTax: req.CountryOfTax,
		},
		Contact: &types.Contact{
			Email: req.Email, Phone: req.Phone, Street: req.Street,
			City: req.City, State: req.State, PostalCode: req.PostalCode, Country: req.Country,
		},
		EnabledAssets: req.EnabledAssets,
		IPAddress:     extractClientIP(r),
	})
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}

	s.auditLog.Record(audit.Entry{
		Action:    audit.ActionAccountCreate,
		Provider:  req.Provider,
		AccountID: acct.ID,
		Status:    "success",
	})

	orgID := r.Header.Get("X-Org-Id")
	webhook.Deliver(s.webhookStore, orgID, "account.created", map[string]interface{}{
		"account_id": acct.ID,
		"provider":   req.Provider,
		"status":     acct.Status,
	})

	writeJSON(w, http.StatusCreated, acct)
}

func (s *Server) handleGetAccount(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	acct, err := p.GetAccount(r.Context(), chi.URLParam(r, "accountId"))
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, acct)
}

func (s *Server) handleGetPortfolio(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	portfolio, err := p.GetPortfolio(r.Context(), chi.URLParam(r, "accountId"))
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, portfolio)
}

func (s *Server) handleListOrders(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	accountID := chi.URLParam(r, "accountId")
	q := r.URL.Query()

	// If provider supports filtered listing and query params are provided, use it
	if ext, ok := p.(provider.TradingExtended); ok {
		params := &types.ListOrdersParams{
			Status:    q.Get("status"),
			After:     q.Get("after"),
			Until:     q.Get("until"),
			Direction: q.Get("direction"),
			Nested:    q.Get("nested") == "true",
		}
		if l := q.Get("limit"); l != "" {
			params.Limit, _ = strconv.Atoi(l)
		}
		orders, err := ext.ListOrdersFiltered(r.Context(), accountID, params)
		if err != nil {
			writeError(w, http.StatusBadGateway, err.Error())
			return
		}
		writeJSON(w, http.StatusOK, orders)
		return
	}

	orders, err := p.ListOrders(r.Context(), accountID)
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, orders)
}

func (s *Server) handleCreateOrder(w http.ResponseWriter, r *http.Request) {
	providerName := chi.URLParam(r, "provider")
	accountID := chi.URLParam(r, "accountId")

	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}

	var req types.CreateOrderRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "invalid request body")
		return
	}

	// Pre-trade risk check
	riskResult := s.riskEng.PreTradeCheck(risk.CheckRequest{
		Provider:  providerName,
		AccountID: accountID,
		Symbol:    req.Symbol,
		Side:      req.Side,
		Qty:       req.Qty,
		OrderType: req.Type,
	})
	if !riskResult.Allowed {
		s.auditLog.Record(audit.Entry{
			Action:    audit.ActionOrderReject,
			Provider:  providerName,
			AccountID: accountID,
			Symbol:    req.Symbol,
			Side:      req.Side,
			Qty:       req.Qty,
			Status:    "rejected",
			Metadata:  map[string]interface{}{"risk_errors": riskResult.Errors},
		})
		writeJSON(w, http.StatusForbidden, map[string]interface{}{
			"error":    "risk check failed",
			"details":  riskResult.Errors,
			"warnings": riskResult.Warnings,
		})
		return
	}

	start := time.Now()
	order, err := p.CreateOrder(r.Context(), accountID, &req)
	latency := time.Since(start)

	status := "success"
	if err != nil {
		status = "failure"
	}

	s.auditLog.RecordOrder(r.Context(), audit.ActionOrderCreate,
		providerName, accountID, req.Symbol, req.Side, req.Qty,
		"", "", status, latency, err)

	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}

	// Estimate order value for risk tracking: qty * limit_price, or notional, or qty alone
	var estimatedValue float64
	if qty, err := strconv.ParseFloat(req.Qty, 64); err == nil && qty > 0 {
		if price, err := strconv.ParseFloat(req.LimitPrice, 64); err == nil && price > 0 {
			estimatedValue = qty * price
		} else if price, err := strconv.ParseFloat(req.StopPrice, 64); err == nil && price > 0 {
			estimatedValue = qty * price
		}
	}
	if estimatedValue == 0 {
		if notional, err := strconv.ParseFloat(req.Notional, 64); err == nil && notional > 0 {
			estimatedValue = notional
		}
	}
	s.riskEng.RecordOrder(providerName, accountID, estimatedValue)

	webhook.Deliver(s.webhookStore, r.Header.Get("X-Org-Id"), "order.placed", map[string]interface{}{
		"order_id":   order.ID,
		"account_id": accountID,
		"provider":   providerName,
		"symbol":     req.Symbol,
		"side":       req.Side,
		"qty":        req.Qty,
		"type":       req.Type,
	})

	writeJSON(w, http.StatusCreated, order)
}

func (s *Server) handleGetOrder(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	order, err := p.GetOrder(r.Context(), chi.URLParam(r, "accountId"), chi.URLParam(r, "orderId"))
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, order)
}

func (s *Server) handleCancelOrder(w http.ResponseWriter, r *http.Request) {
	providerName := chi.URLParam(r, "provider")
	accountID := chi.URLParam(r, "accountId")
	orderID := chi.URLParam(r, "orderId")

	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	if err := p.CancelOrder(r.Context(), accountID, orderID); err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}

	s.auditLog.Record(audit.Entry{
		Action:    audit.ActionOrderCancel,
		Provider:  providerName,
		AccountID: accountID,
		OrderID:   orderID,
		Status:    "success",
	})

	s.riskEng.RecordFill(providerName, accountID)

	webhook.Deliver(s.webhookStore, r.Header.Get("X-Org-Id"), "order.canceled", map[string]interface{}{
		"order_id":   orderID,
		"account_id": accountID,
		"provider":   providerName,
	})

	writeJSON(w, http.StatusOK, map[string]string{"status": "cancelled"})
}

func (s *Server) handleReplaceOrder(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	ext, ok := p.(provider.TradingExtended)
	if !ok {
		writeError(w, http.StatusNotImplemented, "provider does not support order replacement")
		return
	}
	accountID := chi.URLParam(r, "accountId")
	orderID := chi.URLParam(r, "orderId")
	var req types.ReplaceOrderRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "invalid request body")
		return
	}
	order, err := ext.ReplaceOrder(r.Context(), accountID, orderID, &req)
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, order)
}

func (s *Server) handleCancelAllOrders(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	ext, ok := p.(provider.TradingExtended)
	if !ok {
		writeError(w, http.StatusNotImplemented, "provider does not support cancel all orders")
		return
	}
	accountID := chi.URLParam(r, "accountId")
	if err := ext.CancelAllOrders(r.Context(), accountID); err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, map[string]string{"status": "cancelled"})
}

func (s *Server) handleGetPosition(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	ext, ok := p.(provider.TradingExtended)
	if !ok {
		writeError(w, http.StatusNotImplemented, "provider does not support get position")
		return
	}
	position, err := ext.GetPosition(r.Context(), chi.URLParam(r, "accountId"), chi.URLParam(r, "symbol"))
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, position)
}

func (s *Server) handleClosePosition(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	ext, ok := p.(provider.TradingExtended)
	if !ok {
		writeError(w, http.StatusNotImplemented, "provider does not support close position")
		return
	}
	accountID := chi.URLParam(r, "accountId")
	symbol := chi.URLParam(r, "symbol")
	var qty *float64
	if qStr := r.URL.Query().Get("qty"); qStr != "" {
		if q, err := strconv.ParseFloat(qStr, 64); err == nil {
			qty = &q
		}
	}
	order, err := ext.ClosePosition(r.Context(), accountID, symbol, qty)
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, order)
}

func (s *Server) handleCloseAllPositions(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	ext, ok := p.(provider.TradingExtended)
	if !ok {
		writeError(w, http.StatusNotImplemented, "provider does not support close all positions")
		return
	}
	orders, err := ext.CloseAllPositions(r.Context(), chi.URLParam(r, "accountId"))
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, orders)
}

func (s *Server) handleListTransfers(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	transfers, err := p.ListTransfers(r.Context(), chi.URLParam(r, "accountId"))
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, transfers)
}

func (s *Server) handleCreateTransfer(w http.ResponseWriter, r *http.Request) {
	providerName := chi.URLParam(r, "provider")
	accountID := chi.URLParam(r, "accountId")

	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	var req types.CreateTransferRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "invalid request body")
		return
	}
	transfer, err := p.CreateTransfer(r.Context(), accountID, &req)
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}

	s.auditLog.Record(audit.Entry{
		Action:    audit.ActionTransfer,
		Provider:  providerName,
		AccountID: accountID,
		Status:    "success",
		Metadata:  map[string]interface{}{"type": req.Type, "direction": req.Direction, "amount": req.Amount},
	})

	webhook.Deliver(s.webhookStore, r.Header.Get("X-Org-Id"), "transfer.initiated", map[string]interface{}{
		"transfer_id": transfer.ID,
		"account_id":  accountID,
		"provider":    providerName,
		"type":        req.Type,
		"direction":   req.Direction,
		"amount":      req.Amount,
	})

	writeJSON(w, http.StatusCreated, transfer)
}

func (s *Server) handleListBankRelationships(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	rels, err := p.ListBankRelationships(r.Context(), chi.URLParam(r, "accountId"))
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, rels)
}

func (s *Server) handleCreateBankRelationship(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	var req struct {
		AccountOwnerName  string `json:"account_owner_name"`
		BankAccountType   string `json:"bank_account_type"`
		BankAccountNumber string `json:"bank_account_number"`
		BankRoutingNumber string `json:"bank_routing_number"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "invalid request body")
		return
	}
	rel, err := p.CreateBankRelationship(r.Context(), chi.URLParam(r, "accountId"),
		req.AccountOwnerName, req.BankAccountType, req.BankAccountNumber, req.BankRoutingNumber)
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusCreated, rel)
}

func (s *Server) handleListAssets(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	class := r.URL.Query().Get("class")
	assets, err := p.ListAssets(r.Context(), class)
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	if r.URL.Query().Get("all") == "" {
		tradable := make([]*types.Asset, 0, len(assets))
		for _, a := range assets {
			if a.Tradable {
				tradable = append(tradable, a)
			}
		}
		writeJSON(w, http.StatusOK, tradable)
		return
	}
	writeJSON(w, http.StatusOK, assets)
}

func (s *Server) handleGetAsset(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	asset, err := p.GetAsset(r.Context(), chi.URLParam(r, "symbolOrId"))
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, asset)
}

// --- Smart Order Routing Handlers ---

func (s *Server) handleGetRoutes(w http.ResponseWriter, r *http.Request) {
	symbol := chi.URLParam(r, "symbol")
	side := r.URL.Query().Get("side")
	if side == "" {
		side = "buy"
	}
	routes, err := s.sor.GetAllRoutes(r.Context(), symbol, side)
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, routes)
}

func (s *Server) handleGetRoutesPair(w http.ResponseWriter, r *http.Request) {
	symbol := chi.URLParam(r, "symbol") + "/" + chi.URLParam(r, "quote")
	side := r.URL.Query().Get("side")
	if side == "" {
		side = "buy"
	}
	routes, err := s.sor.GetAllRoutes(r.Context(), symbol, side)
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, routes)
}

func (s *Server) handleGetSplitPlan(w http.ResponseWriter, r *http.Request) {
	symbol := chi.URLParam(r, "symbol")
	q := r.URL.Query()
	side := q.Get("side")
	if side == "" {
		side = "buy"
	}
	qty := q.Get("qty")
	if qty == "" {
		writeError(w, http.StatusBadRequest, "qty query param required")
		return
	}

	plan, err := s.sor.BuildSplitPlan(r.Context(), symbol, side, qty)
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}

	s.auditLog.Record(audit.Entry{
		Action:    audit.ActionSplitPlan,
		Symbol:    symbol,
		Side:      side,
		Qty:       qty,
		Algorithm: plan.Algorithm,
		Status:    "success",
	})

	writeJSON(w, http.StatusOK, plan)
}

func (s *Server) handleGetSplitPlanPair(w http.ResponseWriter, r *http.Request) {
	symbol := chi.URLParam(r, "symbol") + "/" + chi.URLParam(r, "quote")
	q := r.URL.Query()
	side := q.Get("side")
	if side == "" {
		side = "buy"
	}
	qty := q.Get("qty")
	if qty == "" {
		writeError(w, http.StatusBadRequest, "qty query param required")
		return
	}

	plan, err := s.sor.BuildSplitPlan(r.Context(), symbol, side, qty)
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, plan)
}

func (s *Server) handleSmartOrder(w http.ResponseWriter, r *http.Request) {
	var req struct {
		Symbol      string `json:"symbol"`
		Qty         string `json:"qty"`
		Side        string `json:"side"`
		Type        string `json:"type"`
		TimeInForce string `json:"time_in_force"`
		LimitPrice  string `json:"limit_price,omitempty"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "invalid request body")
		return
	}

	if req.Type == "" {
		req.Type = "market"
	}
	if req.TimeInForce == "" {
		req.TimeInForce = "day"
	}

	// Build accounts map server-side from authenticated user's resolver mappings.
	// Never trust client-supplied account IDs.
	userID := r.Header.Get("X-User-Id")
	accounts := s.resolver.UserAccounts(userID)
	if len(accounts) == 0 {
		writeError(w, http.StatusBadRequest, "no trading accounts configured")
		return
	}

	start := time.Now()
	order, err := s.sor.SmartOrder(r.Context(), accounts, &types.CreateOrderRequest{
		Symbol:      req.Symbol,
		Qty:         req.Qty,
		Side:        req.Side,
		Type:        req.Type,
		TimeInForce: req.TimeInForce,
		LimitPrice:  req.LimitPrice,
	})
	latency := time.Since(start)

	status := "success"
	var providerUsed string
	if err != nil {
		status = "failure"
	} else {
		providerUsed = order.Provider
	}

	s.auditLog.RecordOrder(r.Context(), audit.ActionRouteDecision,
		providerUsed, "", req.Symbol, req.Side, req.Qty,
		"", "smart_route", status, latency, err)

	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusCreated, order)
}

func (s *Server) handleExecuteSplit(w http.ResponseWriter, r *http.Request) {
	var req struct {
		Symbol string `json:"symbol"`
		Qty    string `json:"qty"`
		Side   string `json:"side"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "invalid request body")
		return
	}

	// Build accounts map server-side from authenticated user's resolver mappings.
	userID := r.Header.Get("X-User-Id")
	accounts := s.resolver.UserAccounts(userID)
	if len(accounts) == 0 {
		writeError(w, http.StatusBadRequest, "no trading accounts configured")
		return
	}

	plan, err := s.sor.BuildSplitPlan(r.Context(), req.Symbol, req.Side, req.Qty)
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}

	result, err := s.sor.ExecuteSplitPlan(r.Context(), plan, accounts)
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}

	s.auditLog.Record(audit.Entry{
		Action:    audit.ActionSplitExecute,
		Symbol:    req.Symbol,
		Side:      req.Side,
		Qty:       req.Qty,
		Algorithm: "split",
		Status:    result.Status,
		Metadata: map[string]interface{}{
			"plan_id":    result.PlanID,
			"vwap":       result.VWAP,
			"total_fees": result.TotalFees,
			"legs":       len(result.Legs),
		},
	})

	writeJSON(w, http.StatusOK, result)
}

func (s *Server) handleAggregatedAssets(w http.ResponseWriter, r *http.Request) {
	assets, err := s.sor.AggregatedAssets(r.Context())
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, assets)
}

// --- Market Data Handlers ---

func (s *Server) handleGetSnapshot(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	snap, err := p.GetSnapshot(r.Context(), chi.URLParam(r, "symbol"))
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, snap)
}

func (s *Server) handleGetSnapshots(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	syms := r.URL.Query().Get("symbols")
	if syms == "" {
		writeError(w, http.StatusBadRequest, "symbols query param required")
		return
	}
	snaps, err := p.GetSnapshots(r.Context(), strings.Split(syms, ","))
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, snaps)
}

func (s *Server) handleGetBars(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	q := r.URL.Query()
	timeframe := q.Get("timeframe")
	if timeframe == "" {
		timeframe = "1Day"
	}
	limit := 0
	if l := q.Get("limit"); l != "" {
		limit, _ = strconv.Atoi(l)
	}
	bars, err := p.GetBars(r.Context(), chi.URLParam(r, "symbol"), timeframe, q.Get("start"), q.Get("end"), limit)
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, bars)
}

func (s *Server) handleGetLatestTrades(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	syms := r.URL.Query().Get("symbols")
	if syms == "" {
		writeError(w, http.StatusBadRequest, "symbols query param required")
		return
	}
	trades, err := p.GetLatestTrades(r.Context(), strings.Split(syms, ","))
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, trades)
}

func (s *Server) handleGetLatestQuotes(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	syms := r.URL.Query().Get("symbols")
	if syms == "" {
		writeError(w, http.StatusBadRequest, "symbols query param required")
		return
	}
	quotes, err := p.GetLatestQuotes(r.Context(), strings.Split(syms, ","))
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, quotes)
}

func (s *Server) handleGetClock(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	clock, err := p.GetClock(r.Context())
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, clock)
}

func (s *Server) handleGetCalendar(w http.ResponseWriter, r *http.Request) {
	p, err := s.getProvider(r)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	q := r.URL.Query()
	days, err := p.GetCalendar(r.Context(), q.Get("start"), q.Get("end"))
	if err != nil {
		writeError(w, http.StatusBadGateway, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, days)
}

// --- Consolidated BBO ---

func (s *Server) handleGetBBO(w http.ResponseWriter, r *http.Request) {
	symbol := chi.URLParam(r, "symbol")
	t, err := s.feed.GetTicker(symbol)
	if err != nil {
		writeError(w, http.StatusNotFound, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, t)
}

func (s *Server) handleGetBBOPair(w http.ResponseWriter, r *http.Request) {
	symbol := chi.URLParam(r, "symbol") + "/" + chi.URLParam(r, "quote")
	t, err := s.feed.GetTicker(symbol)
	if err != nil {
		writeError(w, http.StatusNotFound, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, t)
}

// --- Risk & Audit ---

func (s *Server) handleRiskCheck(w http.ResponseWriter, r *http.Request) {
	q := r.URL.Query()
	result := s.riskEng.PreTradeCheck(risk.CheckRequest{
		Provider:  q.Get("provider"),
		AccountID: q.Get("account_id"),
		Symbol:    q.Get("symbol"),
		Side:      q.Get("side"),
		Qty:       q.Get("qty"),
		Price:     q.Get("price"),
		OrderType: q.Get("type"),
	})
	writeJSON(w, http.StatusOK, result)
}

func (s *Server) handleAuditQuery(w http.ResponseWriter, r *http.Request) {
	q := r.URL.Query()
	filter := audit.Filter{
		Action:    audit.Action(q.Get("action")),
		Provider:  q.Get("provider"),
		AccountID: q.Get("account_id"),
		Symbol:    q.Get("symbol"),
	}
	if since := q.Get("since"); since != "" {
		filter.Since, _ = time.Parse(time.RFC3339, since)
	}
	if until := q.Get("until"); until != "" {
		filter.Until, _ = time.Parse(time.RFC3339, until)
	}
	entries := s.auditLog.Query(filter)
	writeJSON(w, http.StatusOK, entries)
}

func (s *Server) handleAuditStats(w http.ResponseWriter, r *http.Request) {
	writeJSON(w, http.StatusOK, s.auditLog.Stats())
}

func (s *Server) handleAuditExport(w http.ResponseWriter, r *http.Request) {
	data, err := s.auditLog.Export()
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("Content-Disposition", "attachment; filename=audit_export.json")
	w.WriteHeader(http.StatusOK)
	w.Write(data)
}

// requireAccountOwnership is a chi middleware that verifies the authenticated
// user owns the account identified by {provider}/{accountId} in the URL path.
// Returns 403 if the user does not own the account.
func (s *Server) requireAccountOwnership(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		userID := r.Header.Get("X-User-Id")
		providerName := chi.URLParam(r, "provider")
		accountID := chi.URLParam(r, "accountId")

		if userID == "" || providerName == "" || accountID == "" {
			writeError(w, http.StatusForbidden, "account access denied")
			return
		}

		if !s.resolver.OwnsAccount(userID, providerName, accountID) {
			writeError(w, http.StatusForbidden, "account access denied")
			return
		}

		next.ServeHTTP(w, r)
	})
}

// --- Helpers ---

// extractClientIP returns the client IP from the HTTP request.
// In production behind hanzoai/ingress, RemoteAddr is set to the real client IP.
func extractClientIP(r *http.Request) string {
	host, _, err := net.SplitHostPort(r.RemoteAddr)
	if err != nil {
		return r.RemoteAddr
	}
	return host
}

func writeJSON(w http.ResponseWriter, status int, v interface{}) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	json.NewEncoder(w).Encode(v)
}

func writeError(w http.ResponseWriter, status int, msg string) {
	writeJSON(w, status, map[string]string{"error": msg})
}

// corsOriginsFromEnv returns CORS origins from CORS_ALLOWED_ORIGINS env var.
// Defaults to lux.exchange if not set. WL consumers set their own origins.
func corsOriginsFromEnv() []string {
	defaults := []string{"https://lux.exchange", "https://admin.lux.exchange", "https://mpc.lux.network"}
	if env := os.Getenv("BROKER_ENV"); env != "production" {
		defaults = append(defaults, "http://localhost:3000", "http://localhost:3001")
	}
	if extra := os.Getenv("CORS_ALLOWED_ORIGINS"); extra != "" {
		for _, o := range strings.Split(extra, ",") {
			if trimmed := strings.TrimSpace(o); trimmed != "" {
				defaults = append(defaults, trimmed)
			}
		}
	}
	return defaults
}