neural-hunt/internal/customer/server.go

package customer

import (
	"context"
	"crypto/subtle"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"math"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/go-chi/chi/v5"
)

type CreditPackage struct {
	ID            string `json:"id"`
	AmountCents   int64  `json:"amount_cents"`
	Currency      string `json:"currency"`
	CreditsMicros int64  `json:"credits_micros"`
}

type Config struct {
	PublicAddr, AdminAddr, InternalAddr                      string
	PublicBaseURL                                            string
	GamePublicURL, GameAdminURL, SharedSecret                string
	DockerHost, WorkerImage, WorkerEntrypoint, WorkerNetwork string
	WorkerRegisterURL                                        string
	WorkerOrchestrationMode                                  string
	ControllerSharedSecret                                   string
	ControllerOfflineAfter                                   time.Duration
	WorkerRegistryUsername                                   string
	WorkerRegistryPassword                                   string
	WorkerRegistryServer                                     string
	WorkerAutoPull                                           bool
	WorkerRateMicrosPerMinute                                int64
	MaxWorkersPerCustomer                                    int
	MaxWorkersGlobal                                         int
	MaxRunningPerCustomer                                    int
	MaxRunningGlobal                                         int
	SessionTTL                                               time.Duration
	CookieSecure                                             bool
	AdminCookieSecureMode                                    string
	AdminUser, AdminPassword                                 string
	AllowManualCredits                                       bool
	LoginEnabled                                             bool
	RegistrationEnabled                                      bool
	RegistrationPOWBits                                      int
	RegistrationInviteRequired                               bool
	NewCustomerCreditsMicros                                 int64
	PositiveTipCreditsMicros                                 int64
	PayPalEnabled                                            bool
	PayPalEnvironment                                        string
	PayPalWebhookID                                          string
	PayPalLiveApprovalAck                                    string
	Packages                                                 []CreditPackage
}

type Service struct {
	store         *Store
	docker        WorkerRuntime
	paypal        *PayPalClient
	cfg           Config
	adminSessions sync.Map
	workerImageMu sync.Mutex
	hc            *http.Client
}

func NewService(store *Store, docker WorkerRuntime, paypal *PayPalClient, cfg Config) *Service {
	if cfg.SessionTTL <= 0 {
		cfg.SessionTTL = 24 * time.Hour
	}
	if cfg.WorkerRateMicrosPerMinute <= 0 {
		cfg.WorkerRateMicrosPerMinute = 1_000_000
	}
	if cfg.MaxWorkersPerCustomer <= 0 {
		cfg.MaxWorkersPerCustomer = 20
	}
	if cfg.MaxWorkersGlobal <= 0 {
		cfg.MaxWorkersGlobal = 1000
	}
	if cfg.MaxRunningPerCustomer <= 0 {
		cfg.MaxRunningPerCustomer = 10
	}
	if cfg.MaxRunningGlobal <= 0 {
		cfg.MaxRunningGlobal = 100
	}
	if cfg.ControllerOfflineAfter < 15*time.Second {
		cfg.ControllerOfflineAfter = 45 * time.Second
	}
	mode := strings.ToLower(strings.TrimSpace(cfg.WorkerOrchestrationMode))
	if mode == "" {
		mode = "direct"
	}
	if mode != "direct" && mode != "controller" && mode != "hybrid" {
		mode = "direct"
	}
	cfg.WorkerOrchestrationMode = mode
	return &Service{store: store, docker: docker, paypal: paypal, cfg: cfg, hc: &http.Client{Timeout: 10 * time.Second}}
}

func requestIsHTTPS(r *http.Request) bool {
	if r.TLS != nil {
		return true
	}
	proto := strings.TrimSpace(strings.Split(r.Header.Get("X-Forwarded-Proto"), ",")[0])
	return strings.EqualFold(proto, "https")
}

func (s *Service) security(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("X-Content-Type-Options", "nosniff")
		w.Header().Set("X-Frame-Options", "DENY")
		w.Header().Set("Referrer-Policy", "no-referrer")
		w.Header().Set("Permissions-Policy", "camera=(), microphone=(), geolocation=()")
		w.Header().Set("Content-Security-Policy", "default-src 'self'; base-uri 'none'; object-src 'none'; frame-ancestors 'none'; script-src 'self'; style-src 'self'; img-src 'self' data:; connect-src 'self'; form-action 'self' https://www.paypal.com https://www.sandbox.paypal.com")
		// HSTS is meaningful only for HTTPS responses. In particular, do not emit it
		// on the VPN-only plain-HTTP admin listener.
		if requestIsHTTPS(r) {
			w.Header().Set("Strict-Transport-Security", "max-age=31536000")
		}
		next.ServeHTTP(w, r)
	})
}
func jsonOut(w http.ResponseWriter, status int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(v)
}
func decode(r *http.Request, v any) error {
	d := json.NewDecoder(io.LimitReader(r.Body, 1<<20))
	d.DisallowUnknownFields()
	return d.Decode(v)
}

const customerCookie = "neuralhunt_customer_session"
const csAdminCookie = "neuralhunt_customer_admin"

func (s *Service) cookieSameSite(name string) http.SameSite {
	// The public customer portal may return from external payment providers. Lax
	// still blocks cross-site fetch/POST cookie use while allowing normal top-level
	// navigation back to the portal. Keep the private admin cookie Strict.
	if name == customerCookie {
		return http.SameSiteLaxMode
	}
	return http.SameSiteStrictMode
}

func (s *Service) cookieSecure(r *http.Request, name string) bool {
	if name != csAdminCookie {
		// The public customer portal should normally stay HTTPS-only.
		return s.cfg.CookieSecure
	}

	// The Customer-Service admin listener is commonly reachable only through an
	// encrypted VPN but over plain HTTP. Keep its cookie policy independent from
	// the public customer portal. "auto" means Secure behind HTTPS and non-Secure
	// for a direct HTTP/VPN connection.
	switch strings.ToLower(strings.TrimSpace(s.cfg.AdminCookieSecureMode)) {
	case "0", "false", "no", "off":
		return false
	case "1", "true", "yes", "on":
		return true
	case "", "auto":
		return requestIsHTTPS(r)
	default:
		return true
	}
}

func (s *Service) setCookie(w http.ResponseWriter, r *http.Request, name, value string, ttl time.Duration) {
	now := time.Now().UTC()
	http.SetCookie(w, &http.Cookie{
		Name:     name,
		Value:    value,
		Path:     "/",
		HttpOnly: true,
		Secure:   s.cookieSecure(r, name),
		SameSite: s.cookieSameSite(name),
		MaxAge:   int(ttl.Seconds()),
		Expires:  now.Add(ttl),
	})
}
func (s *Service) clearCookie(w http.ResponseWriter, r *http.Request, name string) {
	http.SetCookie(w, &http.Cookie{
		Name:     name,
		Value:    "",
		Path:     "/",
		HttpOnly: true,
		Secure:   s.cookieSecure(r, name),
		SameSite: s.cookieSameSite(name),
		MaxAge:   -1,
		Expires:  time.Unix(1, 0).UTC(),
	})
}
func (s *Service) customerID(r *http.Request) (string, error) {
	c, err := r.Cookie(customerCookie)
	if err != nil {
		return "", err
	}
	return s.store.SessionCustomer(r.Context(), c.Value)
}
func (s *Service) requireCustomer(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		cid, err := s.customerID(r)
		if err != nil {
			jsonOut(w, 401, map[string]string{"error": "unauthorized"})
			return
		}
		ctx := context.WithValue(r.Context(), ctxKey("customer"), cid)
		next.ServeHTTP(w, r.WithContext(ctx))
	})
}

type ctxKey string

func customerID(r *http.Request) string {
	v, _ := r.Context().Value(ctxKey("customer")).(string)
	return v
}

func (s *Service) PublicRoutes(ui http.Handler) http.Handler {
	r := chi.NewRouter()
	r.Use(s.security)
	r.Get("/api/healthz", func(w http.ResponseWriter, r *http.Request) { jsonOut(w, 200, map[string]bool{"ok": true}) })
	r.Get("/api/registration/config", s.registrationConfig)
	r.Post("/api/register/challenge", s.registrationChallenge)
	r.Post("/api/register", s.register)
	r.Post("/api/login", s.login)
	r.Post("/api/logout", s.logout)
	r.Post("/api/paypal/webhook", s.paypalWebhook)
	r.Group(func(r chi.Router) {
		r.Use(s.requireCustomer)
		r.Get("/api/me", s.me)
		r.Get("/api/ledger", s.ledger)
		r.Put("/api/reward-identity", s.rewardIdentity)
		r.Get("/api/tasks", s.tasks)
		r.Get("/api/workers", s.workers)
		r.Post("/api/workers", s.createWorker)
		r.Put("/api/workers/{id}", s.updateWorker)
		r.Post("/api/workers/{id}/start", s.startWorker)
		r.Post("/api/workers/{id}/stop", s.stopWorker)
		r.Delete("/api/workers/{id}", s.deleteWorker)
		r.Get("/api/workers/{id}/identity", s.workerIdentityGet)
		r.Put("/api/workers/{id}/identity", s.workerIdentityPut)
		r.Get("/api/billing/packages", s.billingPackages)
		r.Post("/api/billing/paypal/order", s.paypalCreateOrder)
		r.Post("/api/billing/paypal/capture", s.paypalCapture)
	})
	r.Mount("/", ui)
	return r
}
func (s *Service) AdminRoutes(ui http.Handler) http.Handler {
	r := chi.NewRouter()
	r.Use(s.security)
	r.Post("/api/admin/login", s.adminLogin)
	r.Post("/api/admin/logout", s.adminLogout)
	r.Group(func(r chi.Router) {
		r.Use(s.requireAdmin)
		r.Get("/api/admin/overview", s.adminOverview)
		r.Get("/api/admin/settings", s.adminSettings)
		r.Put("/api/admin/settings", s.adminUpdateSettings)
		r.Post("/api/admin/credits/grant", s.adminCreditGrant)
		r.Post("/api/admin/customers/{id}/block", s.adminBlockCustomer)
		r.Post("/api/admin/customers/{id}/unblock", s.adminUnblockCustomer)
		r.Post("/api/admin/customers/{id}/workers/stop", s.adminStopCustomerWorkers)
		r.Put("/api/admin/customers/{id}/worker-limit", s.adminSetCustomerWorkerLimit)
		r.Get("/api/admin/tasks", s.adminTaskCatalog)
		r.Post("/api/admin/customers/{id}/workers", s.adminCreateWorkerForCustomer)
		r.Post("/api/admin/workers/{id}/stop", s.adminStopAnyWorker)
		r.Post("/api/admin/workers/{id}/update-image", s.adminUpdateWorkerImage)
		r.Post("/api/admin/workers/update-image", s.adminUpdateAllWorkerImages)
		r.Put("/api/admin/controllers/{id}", s.adminUpdateControllerPolicy)
		r.Post("/api/admin/invites", s.adminCreateInvite)
	})
	r.Mount("/", ui)
	return r
}
func (s *Service) InternalRoutes() http.Handler {
	r := chi.NewRouter()
	r.Use(s.security)
	r.Post("/internal/workers/register", s.internalWorkerRegister)
	r.Post("/internal/workers/lease", s.internalWorkerLease)
	r.Post("/internal/game/positive-tip", s.internalGamePositiveTip)
	r.Post("/internal/controllers/register", s.internalControllerRegister)
	r.Get("/api/healthz", func(w http.ResponseWriter, r *http.Request) { jsonOut(w, 200, map[string]bool{"ok": true}) })
	return r
}

func (s *Service) register(w http.ResponseWriter, r *http.Request) {
	cfg := s.portalSettings(r.Context())
	if !cfg.RegistrationEnabled {
		jsonOut(w, http.StatusServiceUnavailable, map[string]string{"error": "Registrierungen sind derzeit deaktiviert."})
		return
	}
	var in struct {
		Username     string `json:"Username"`
		Password     string `json:"Password"`
		Challenge    string `json:"Challenge"`
		ProofCounter uint64 `json:"ProofCounter"`
		InviteCode   string `json:"InviteCode"`
	}
	if decode(r, &in) != nil {
		jsonOut(w, 400, map[string]string{"error": "bad json"})
		return
	}
	in.Username = strings.TrimSpace(in.Username)
	if len(in.Username) < 3 || len(in.Username) > 80 {
		jsonOut(w, 400, map[string]string{"error": "username must be 3..80 characters"})
		return
	}
	if cfg.RegistrationPOWBits > 0 {
		if !VerifyRegistrationChallenge(s.cfg.SharedSecret, in.Username, in.Challenge) || !VerifyRegistrationProof(in.Username, in.Challenge, in.ProofCounter, cfg.RegistrationPOWBits) {
			jsonOut(w, http.StatusBadRequest, map[string]string{"error": "Registrierungs-Proof-of-Work ist ungültig oder abgelaufen."})
			return
		}
	}
	inviteHash := ""
	if cfg.InviteRequired {
		if strings.TrimSpace(in.InviteCode) == "" {
			jsonOut(w, http.StatusForbidden, map[string]string{"error": "Für die Registrierung ist ein einmaliger Invite-Code erforderlich."})
			return
		}
		inviteHash = HashInviteCode(in.InviteCode)
	}
	salt, hash, err := NewPasswordHash(in.Password)
	if err != nil {
		jsonOut(w, 400, map[string]string{"error": err.Error()})
		return
	}
	cid := "cust_" + RandomToken(16)
	if err := s.store.CreateCustomer(r.Context(), cid, in.Username, salt, hash, cfg.SignupBonusMicros, inviteHash); err != nil {
		if errors.Is(err, ErrInvalidInvite) {
			jsonOut(w, http.StatusForbidden, map[string]string{"error": "Invite-Code ist ungültig, abgelaufen oder bereits verwendet."})
			return
		}
		jsonOut(w, 409, map[string]string{"error": "username already exists"})
		return
	}
	sid := RandomToken(32)
	if err := s.store.CreateSession(r.Context(), sid, cid, s.cfg.SessionTTL); err != nil {
		jsonOut(w, 500, map[string]string{"error": "session failed"})
		return
	}
	s.setCookie(w, r, customerCookie, sid, s.cfg.SessionTTL)
	jsonOut(w, 201, map[string]any{"id": cid, "username": in.Username, "signup_bonus_micros": cfg.SignupBonusMicros})
}

func (s *Service) login(w http.ResponseWriter, r *http.Request) {
	if !s.portalSettings(r.Context()).LoginEnabled {
		jsonOut(w, http.StatusServiceUnavailable, map[string]string{"error": "Anmeldung ist momentan administrativ deaktiviert."})
		return
	}
	var in struct{ Username, Password string }
	if decode(r, &in) != nil {
		jsonOut(w, 400, map[string]string{"error": "bad json"})
		return
	}
	c, salt, hash, err := s.store.CustomerByUsername(r.Context(), in.Username)
	if err != nil || !VerifyPassword(in.Password, salt, hash) {
		time.Sleep(250 * time.Millisecond)
		jsonOut(w, 401, map[string]string{"error": "invalid credentials"})
		return
	}
	if c.Blocked {
		jsonOut(w, http.StatusForbidden, map[string]string{"error": "Dieses Kundenkonto ist gesperrt."})
		return
	}
	sid := RandomToken(32)
	if err := s.store.CreateSession(r.Context(), sid, c.ID, s.cfg.SessionTTL); err != nil {
		jsonOut(w, 500, map[string]string{"error": "session failed"})
		return
	}
	s.setCookie(w, r, customerCookie, sid, s.cfg.SessionTTL)
	jsonOut(w, 200, map[string]any{"ok": true})
}
func (s *Service) logout(w http.ResponseWriter, r *http.Request) {
	if c, err := r.Cookie(customerCookie); err == nil {
		s.store.DeleteSession(r.Context(), c.Value)
	}
	s.clearCookie(w, r, customerCookie)
	jsonOut(w, 200, map[string]bool{"ok": true})
}
func (s *Service) me(w http.ResponseWriter, r *http.Request) {
	cid := customerID(r)
	c, err := s.store.CustomerByID(r.Context(), cid)
	if err != nil {
		jsonOut(w, 404, map[string]string{"error": "customer missing"})
		return
	}
	bal, _ := s.store.BalanceMicros(r.Context(), cid)
	jsonOut(w, 200, map[string]any{"customer": c, "balance_micros": bal, "worker_rate_micros_per_minute": s.cfg.WorkerRateMicrosPerMinute, "paypal_enabled": s.paypalAllowed()})
}
func (s *Service) ledger(w http.ResponseWriter, r *http.Request) {
	x, err := s.store.LedgerSummary(r.Context(), customerID(r), 60)
	if err != nil {
		jsonOut(w, 500, map[string]string{"error": err.Error()})
		return
	}
	jsonOut(w, 200, x)
}

func (s *Service) rewardIdentity(w http.ResponseWriter, r *http.Request) {
	cid := customerID(r)
	var in struct {
		LinkCode string `json:"link_code"`
		Clear    bool   `json:"clear"`
	}
	if decode(r, &in) != nil {
		jsonOut(w, 400, map[string]string{"error": "bad json"})
		return
	}

	workers, _ := s.store.Workers(r.Context(), cid)
	if in.Clear {
		// Clearing must reach the game first; otherwise old worker delegations
		// could silently remain active while the portal already looks unlinked.
		if err := s.checkGameControlPlane(r.Context()); err != nil {
			jsonOut(w, http.StatusBadGateway, map[string]string{"error": err.Error()})
			return
		}
		for _, wk := range workers {
			if wk.WorkerClientID == "" {
				continue
			}
			if err := s.syncDelegation(r.Context(), wk.WorkerClientID, ""); err != nil {
				jsonOut(w, http.StatusBadGateway, map[string]string{"error": "Reward-Kopplung konnte im Spiel nicht entfernt werden: " + err.Error()})
				return
			}
		}
		if err := s.store.SetRewardClientID(r.Context(), cid, ""); err != nil {
			jsonOut(w, 500, map[string]string{"error": err.Error()})
			return
		}
		jsonOut(w, 200, map[string]any{"ok": true, "reward_client_id": ""})
		return
	}

	if strings.TrimSpace(in.LinkCode) == "" {
		jsonOut(w, 400, map[string]string{"error": "Hosted-Code fehlt. Erzeuge ihn im Neural-Hunt-Spiel mit der gewünschten Haupt-Identität."})
		return
	}
	clientID, err := s.redeemRewardLink(r.Context(), in.LinkCode)
	if err != nil {
		log.Printf("customer reward identity %s pairing failed: %v", cid, err)
		jsonOut(w, http.StatusBadGateway, map[string]string{"error": "Hosted-Code konnte nicht gekoppelt werden: " + err.Error()})
		return
	}

	// The one-shot code has now been consumed. Persist the proven owner before
	// syncing already-known workers so a transient delegation failure does not
	// destroy a valid pairing and force the customer to generate another code.
	if err := s.store.SetRewardClientID(r.Context(), cid, clientID); err != nil {
		jsonOut(w, 500, map[string]string{"error": "Hosted-Code wurde bestätigt, aber die Haupt-Identität konnte lokal nicht gespeichert werden: " + err.Error()})
		return
	}

	var warnings []string
	for _, wk := range workers {
		if wk.WorkerClientID == "" {
			continue
		}
		if err := s.syncDelegation(r.Context(), wk.WorkerClientID, clientID); err != nil {
			warnings = append(warnings, fmt.Sprintf("Worker %s: %v", wk.ID, err))
		}
	}
	out := map[string]any{"ok": true, "reward_client_id": clientID}
	if len(warnings) > 0 {
		out["warning"] = "Haupt-Identität ist gekoppelt; bestehende Worker konnten noch nicht vollständig synchronisiert werden: " + strings.Join(warnings, "; ")
		log.Printf("customer reward identity %s coupled with delegation warnings: %s", cid, strings.Join(warnings, "; "))
	}
	jsonOut(w, 200, out)
}

func (s *Service) proxyGET(ctx context.Context, path string, out any) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimRight(s.cfg.GamePublicURL, "/")+path, nil)
	if err != nil {
		return err
	}
	resp, err := s.hc.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	b, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
	if resp.StatusCode/100 != 2 {
		return fmt.Errorf("game HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(b)))
	}
	return json.Unmarshal(b, out)
}
func (s *Service) tasks(w http.ResponseWriter, r *http.Request) {
	var out any
	if err := s.proxyGET(r.Context(), "/api/public/tasks", &out); err != nil {
		jsonOut(w, 502, map[string]string{"error": err.Error()})
		return
	}
	jsonOut(w, 200, out)
}
func (s *Service) taskAvailable(ctx context.Context, taskID string) (bool, error) {
	var tasks []struct {
		ID     string `json:"id"`
		Paused bool   `json:"paused"`
	}
	if err := s.proxyGET(ctx, "/api/public/tasks", &tasks); err != nil {
		return false, err
	}
	for _, t := range tasks {
		if t.ID == strings.TrimSpace(taskID) {
			return true, nil
		}
	}
	return false, nil
}

func normalizePath(v string) string {
	v = strings.ToLower(strings.TrimSpace(v))
	switch v {
	case "pulse", "flux", "orbit", "auto":
		return v
	default:
		return "auto"
	}
}
func (s *Service) workers(w http.ResponseWriter, r *http.Request) {
	x, err := s.store.Workers(r.Context(), customerID(r))
	if err != nil {
		jsonOut(w, 500, map[string]string{"error": err.Error()})
		return
	}
	// Physical host/controller IDs are private infrastructure metadata. The
	// customer-facing API deliberately keeps placement transparent.
	for i := range x {
		x[i].ControllerID = ""
		x[i].ContainerID = ""
		x[i].Volume = ""
		x[i].LastLeaseAt = nil
	}
	jsonOut(w, 200, x)
}
func (s *Service) createWorkerRecord(ctx context.Context, cid, taskID, beaconPath string, ignoreCustomerLimit bool) (Worker, error) {
	cust, err := s.store.CustomerByID(ctx, cid)
	if err != nil {
		return Worker{}, fmt.Errorf("customer not found")
	}
	if !ignoreCustomerLimit && !cust.WorkerLimitBypass {
		n, err := s.store.WorkerCount(ctx, cid, false)
		if err != nil {
			return Worker{}, err
		}
		if n >= s.cfg.MaxWorkersPerCustomer {
			return Worker{}, fmt.Errorf("worker limit reached (%d)", s.cfg.MaxWorkersPerCustomer)
		}
	}
	n, err := s.store.TotalWorkerCount(ctx)
	if err != nil {
		return Worker{}, err
	}
	if n >= s.cfg.MaxWorkersGlobal {
		return Worker{}, fmt.Errorf("global hosted worker inventory limit reached")
	}
	available, err := s.taskAvailable(ctx, taskID)
	if err != nil {
		return Worker{}, fmt.Errorf("task catalog: %w", err)
	}
	if !available {
		return Worker{}, fmt.Errorf("task is not active/available")
	}
	wid := "wrk_" + RandomToken(12)
	wk := Worker{ID: wid, CustomerID: cid, TaskID: strings.TrimSpace(taskID), BeaconPath: normalizePath(beaconPath), Volume: "nh_identity_" + strings.ReplaceAll(wid, "-", "_"), RegisterToken: RandomToken(24), RateMicrosPerMinute: s.cfg.WorkerRateMicrosPerMinute}
	if err := s.store.CreateWorker(ctx, wk); err != nil {
		return Worker{}, err
	}
	return wk, nil
}

func (s *Service) createWorker(w http.ResponseWriter, r *http.Request) {
	var in struct{ TaskID, BeaconPath string }
	if decode(r, &in) != nil || strings.TrimSpace(in.TaskID) == "" {
		jsonOut(w, 400, map[string]string{"error": "task_id required"})
		return
	}
	wk, err := s.createWorkerRecord(r.Context(), customerID(r), in.TaskID, in.BeaconPath, false)
	if err != nil {
		status := http.StatusBadRequest
		if strings.Contains(err.Error(), "limit reached") {
			status = http.StatusConflict
		}
		if strings.Contains(err.Error(), "global hosted") || strings.Contains(err.Error(), "task catalog") {
			status = http.StatusServiceUnavailable
		}
		jsonOut(w, status, map[string]string{"error": err.Error()})
		return
	}
	wk.ControllerID = ""
	wk.ContainerID = ""
	wk.Volume = ""
	wk.LastLeaseAt = nil
	jsonOut(w, 201, wk)
}

func (s *Service) adminTaskCatalog(w http.ResponseWriter, r *http.Request) {
	var out any
	if err := s.proxyGET(r.Context(), "/api/public/tasks", &out); err != nil {
		jsonOut(w, http.StatusBadGateway, map[string]string{"error": err.Error()})
		return
	}
	jsonOut(w, http.StatusOK, out)
}

func (s *Service) adminCreateWorkerForCustomer(w http.ResponseWriter, r *http.Request) {
	cid := chi.URLParam(r, "id")
	var in struct {
		TaskID     string `json:"task_id"`
		BeaconPath string `json:"beacon_path"`
	}
	if decode(r, &in) != nil || strings.TrimSpace(in.TaskID) == "" {
		jsonOut(w, http.StatusBadRequest, map[string]string{"error": "task_id required"})
		return
	}
	wk, err := s.createWorkerRecord(r.Context(), cid, in.TaskID, in.BeaconPath, true)
	if err != nil {
		status := http.StatusBadRequest
		if strings.Contains(err.Error(), "global hosted") {
			status = http.StatusServiceUnavailable
		}
		if strings.Contains(err.Error(), "customer not found") {
			status = http.StatusNotFound
		}
		jsonOut(w, status, map[string]string{"error": err.Error()})
		return
	}
	jsonOut(w, http.StatusCreated, wk)
}
func (s *Service) updateWorker(w http.ResponseWriter, r *http.Request) {
	cid, wid := customerID(r), chi.URLParam(r, "id")
	wk, err := s.store.Worker(r.Context(), cid, wid)
	if err != nil {
		jsonOut(w, 404, map[string]string{"error": "worker not found"})
		return
	}
	var in struct{ TaskID, BeaconPath string }
	if decode(r, &in) != nil || strings.TrimSpace(in.TaskID) == "" {
		jsonOut(w, 400, map[string]string{"error": "task_id required"})
		return
	}
	available, err := s.taskAvailable(r.Context(), in.TaskID)
	if err != nil {
		jsonOut(w, 502, map[string]string{"error": "task catalog: " + err.Error()})
		return
	}
	if !available {
		jsonOut(w, 400, map[string]string{"error": "task is not active/available"})
		return
	}
	if wk.ContainerID != "" {
		// Revoke centrally first. If the remote host is temporarily unavailable,
		// the worker loses its next lease and exits on its own.
		_ = s.store.SetWorkerRuntime(r.Context(), wid, "stopped", wk.ContainerID, "")
		if rt, rtErr := s.runtimeForWorker(r.Context(), &wk, false); rtErr == nil {
			_ = rt.Stop(r.Context(), wk.ContainerID, 5)
			_ = rt.Remove(r.Context(), wk.ContainerID)
			_ = s.store.SetWorkerRuntime(r.Context(), wid, "stopped", "", "")
		}
	}
	if err := s.store.UpdateWorkerConfig(r.Context(), cid, wid, strings.TrimSpace(in.TaskID), normalizePath(in.BeaconPath)); err != nil {
		jsonOut(w, 500, map[string]string{"error": err.Error()})
		return
	}
	jsonOut(w, 200, map[string]bool{"ok": true})
}

func (s *Service) ensurePaidSlice(ctx context.Context, wk Worker) (ok bool, newlyCharged bool, chargedAt time.Time, err error) {
	now := time.Now().UTC()
	if wk.LastChargeAt != nil && now.Sub(*wk.LastChargeAt) < time.Minute {
		return true, false, time.Time{}, nil
	}
	ok, err = s.store.ChargeWorkerMinute(ctx, wk, now)
	return ok, ok && err == nil, now, err
}
func (s *Service) workerContainerConfig(wk Worker, name string) WorkerContainerConfig {
	registerURL := strings.TrimSpace(s.cfg.WorkerRegisterURL)
	if registerURL == "" {
		registerURL = "http://customer-service:8092/internal/workers/register"
	}
	return WorkerContainerConfig{Image: s.cfg.WorkerImage, Entrypoint: s.cfg.WorkerEntrypoint, Network: s.cfg.WorkerNetwork, GameURL: s.cfg.GamePublicURL, RegisterURL: registerURL, WorkerID: wk.ID, RegisterToken: wk.RegisterToken, TaskID: wk.TaskID, BeaconPath: wk.BeaconPath, Volume: wk.Volume, Name: name}
}

func (s *Service) controllerOnline(c ServiceController) bool {
	return c.Enabled && !c.Draining && time.Since(c.LastSeen) <= s.cfg.ControllerOfflineAfter
}

func (s *Service) controllerRuntime(ctx context.Context, controllerID string) (WorkerRuntime, error) {
	c, err := s.store.ServiceController(ctx, strings.TrimSpace(controllerID))
	if err != nil {
		return nil, fmt.Errorf("service controller %q not found", controllerID)
	}
	if !c.Enabled {
		return nil, fmt.Errorf("service controller %s is disabled", c.ID)
	}
	if time.Since(c.LastSeen) > s.cfg.ControllerOfflineAfter {
		return nil, fmt.Errorf("service controller %s is offline (last heartbeat %s ago)", c.ID, time.Since(c.LastSeen).Round(time.Second))
	}
	return NewRemoteControllerClient(c.BaseURL, s.cfg.ControllerSharedSecret)
}

func (s *Service) chooseController(ctx context.Context) (ServiceController, error) {
	controllers, err := s.store.ServiceControllers(ctx)
	if err != nil {
		return ServiceController{}, err
	}
	var best ServiceController
	bestScore := 1e18
	found := false
	for _, c := range controllers {
		if !s.controllerOnline(c) {
			continue
		}
		total, running, err := s.store.ControllerWorkerCounts(ctx, c.ID)
		if err != nil {
			continue
		}
		if c.MaxWorkers > 0 && (total >= c.MaxWorkers || c.ReportedWorkers >= c.MaxWorkers) {
			continue
		}
		maxW := c.MaxWorkers
		if maxW <= 0 {
			maxW = 1
		}
		maxR := c.MaxRunning
		if maxR <= 0 {
			maxR = 1
		}
		// Prefer the least-loaded controller. Local DB assignment counts are the
		// source of truth; reported counts add a small penalty for emergency/manual
		// containers that the Master may not know about yet.
		score := float64(running)/float64(maxR)*10 + float64(total)/float64(maxW)
		if c.ReportedRunning > running {
			score += float64(c.ReportedRunning-running) / float64(maxR) * 4
		}
		if c.ReportedWorkers > total {
			score += float64(c.ReportedWorkers-total) / float64(maxW)
		}
		if !found || score < bestScore {
			best, bestScore, found = c, score, true
		}
	}
	if !found {
		return ServiceController{}, errors.New("no online Service Controller with free capacity")
	}
	return best, nil
}

// runtimeForWorker resolves the worker host. Existing assignments are sticky so
// identity volumes never jump hosts accidentally. New workers are placed on a
// healthy controller in controller/hybrid mode; direct mode keeps the legacy
// local docker.sock behavior.
func (s *Service) runtimeForWorker(ctx context.Context, wk *Worker, assign bool) (WorkerRuntime, error) {
	if wk == nil {
		return nil, errors.New("worker required")
	}
	if strings.EqualFold(strings.TrimSpace(wk.ControllerID), "local") {
		if s.docker == nil {
			return nil, errors.New("worker is assigned to local Docker but local runtime is disabled")
		}
		return s.docker, nil
	}
	if strings.TrimSpace(wk.ControllerID) != "" {
		return s.controllerRuntime(ctx, wk.ControllerID)
	}
	mode := s.cfg.WorkerOrchestrationMode
	if mode == "direct" {
		if s.docker == nil {
			return nil, errors.New("local Docker runtime unavailable")
		}
		return s.docker, nil
	}
	// Backward compatibility: workers created before Service Controllers existed
	// have no controller_id. If they already have a local container, never move
	// their named identity volume implicitly to another host.
	if wk.ContainerID != "" || wk.WorkerClientID != "" {
		if s.docker != nil {
			_ = s.store.SetWorkerController(ctx, wk.ID, "local")
			wk.ControllerID = "local"
			return s.docker, nil
		}
		return nil, errors.New("legacy local worker has an existing runtime identity but no controller assignment; use direct/hybrid mode to export or stop it before moving to controller-only mode")
	}
	if assign {
		c, err := s.chooseController(ctx)
		if err == nil {
			rt, rtErr := NewRemoteControllerClient(c.BaseURL, s.cfg.ControllerSharedSecret)
			if rtErr != nil {
				return nil, rtErr
			}
			if err := s.store.SetWorkerController(ctx, wk.ID, c.ID); err != nil {
				return nil, err
			}
			wk.ControllerID = c.ID
			return rt, nil
		}
		if mode == "controller" {
			return nil, err
		}
	}
	if mode == "hybrid" && s.docker != nil {
		_ = s.store.SetWorkerController(ctx, wk.ID, "local")
		wk.ControllerID = "local"
		return s.docker, nil
	}
	return nil, errors.New("worker has no Service Controller assignment")
}

func (s *Service) ensureWorkerImageOn(ctx context.Context, rt WorkerRuntime) error {
	if rt == nil {
		return errors.New("worker runtime unavailable")
	}
	auth, err := RegistryAuthHeader(s.cfg.WorkerRegistryUsername, s.cfg.WorkerRegistryPassword, s.cfg.WorkerRegistryServer)
	if err != nil {
		return err
	}
	return rt.EnsureImage(ctx, s.cfg.WorkerImage, s.cfg.WorkerAutoPull, auth)
}

func (s *Service) pullWorkerImageOn(ctx context.Context, rt WorkerRuntime) error {
	if rt == nil {
		return errors.New("worker runtime unavailable")
	}
	auth, err := RegistryAuthHeader(s.cfg.WorkerRegistryUsername, s.cfg.WorkerRegistryPassword, s.cfg.WorkerRegistryServer)
	if err != nil {
		return err
	}
	if err := rt.PullImage(ctx, s.cfg.WorkerImage, auth); err != nil {
		return fmt.Errorf("pull worker image %q: %w", s.cfg.WorkerImage, err)
	}
	return nil
}

// recreateWorkerWithCurrentImage preserves the named identity volume and the
// worker's task/beacon configuration. Running workers are resumed after the
// container was recreated; stopped workers remain stopped. Pulling is done by
// the caller so bulk updates need only one registry request.
func (s *Service) recreateWorkerWithCurrentImage(ctx context.Context, wk Worker) error {
	rt, err := s.runtimeForWorker(ctx, &wk, true)
	if err != nil {
		return err
	}
	resume := wk.Status == "running" || wk.Status == "starting"
	if resume {
		if cust, err := s.store.CustomerByID(ctx, wk.CustomerID); err != nil || cust.Blocked {
			resume = false
		}
	}
	oldID := wk.ContainerID
	if oldID != "" {
		_ = rt.Stop(ctx, oldID, 5)
		if err := rt.Remove(ctx, oldID); err != nil {
			return fmt.Errorf("remove old worker container: %w", err)
		}
	}
	name := "neuralhunt-worker-" + strings.TrimPrefix(wk.ID, "wrk_")
	_ = rt.Remove(ctx, name)
	containerID, err := rt.CreateWorker(ctx, s.workerContainerConfig(wk, name))
	if err != nil {
		_ = s.store.SetWorkerRuntime(ctx, wk.ID, "error", "", "image update: "+err.Error())
		return err
	}
	if resume {
		if err := rt.Start(ctx, containerID); err != nil {
			_ = rt.Remove(ctx, containerID)
			_ = s.store.SetWorkerRuntime(ctx, wk.ID, "error", "", "image update start: "+err.Error())
			return err
		}
		if err := s.store.SetWorkerRuntime(ctx, wk.ID, "running", containerID, ""); err != nil {
			_ = rt.Stop(ctx, containerID, 2)
			return err
		}
		return nil
	}
	return s.store.SetWorkerRuntime(ctx, wk.ID, "stopped", containerID, "")
}

// ensureWorkerContainer makes the worker identity volume addressable through
// Docker's archive API even when a previous config change removed the old
// container. The helper container remains stopped until the customer starts it.
func (s *Service) ensureWorkerContainer(ctx context.Context, wk Worker) (Worker, error) {
	if wk.ContainerID != "" {
		return wk, nil
	}
	rt, err := s.runtimeForWorker(ctx, &wk, true)
	if err != nil {
		return wk, err
	}
	s.workerImageMu.Lock()
	err = s.ensureWorkerImageOn(ctx, rt)
	s.workerImageMu.Unlock()
	if err != nil {
		return wk, err
	}
	name := "neuralhunt-worker-" + strings.TrimPrefix(wk.ID, "wrk_")
	id, err := rt.CreateWorker(ctx, s.workerContainerConfig(wk, name))
	if err != nil {
		return wk, err
	}
	wk.ContainerID = id
	if err := s.store.SetWorkerRuntime(ctx, wk.ID, "stopped", id, ""); err != nil {
		_ = rt.Remove(ctx, id)
		return wk, err
	}
	return wk, nil
}

func (s *Service) startWorker(w http.ResponseWriter, r *http.Request) {
	cid, wid := customerID(r), chi.URLParam(r, "id")
	wk, err := s.store.Worker(r.Context(), cid, wid)
	if err != nil {
		jsonOut(w, 404, map[string]string{"error": "worker not found"})
		return
	}
	if wk.Status != "running" {
		cust, custErr := s.store.CustomerByID(r.Context(), cid)
		if custErr != nil {
			jsonOut(w, 404, map[string]string{"error": "customer not found"})
			return
		}
		if !cust.WorkerLimitBypass {
			if n, _ := s.store.WorkerCount(r.Context(), cid, true); n >= s.cfg.MaxRunningPerCustomer {
				jsonOut(w, 409, map[string]string{"error": fmt.Sprintf("running worker limit reached (%d)", s.cfg.MaxRunningPerCustomer)})
				return
			}
		}
		if n, _ := s.store.RunningWorkerCount(r.Context()); n >= s.cfg.MaxRunningGlobal {
			jsonOut(w, 503, map[string]string{"error": "hosted worker capacity reached"})
			return
		}
	}
	available, err := s.taskAvailable(r.Context(), wk.TaskID)
	if err != nil {
		jsonOut(w, 502, map[string]string{"error": "task catalog: " + err.Error()})
		return
	}
	if !available {
		jsonOut(w, 409, map[string]string{"error": "assigned task is no longer active; choose another task"})
		return
	}
	cust, err := s.store.CustomerByID(r.Context(), cid)
	if err != nil || strings.TrimSpace(cust.RewardClientID) == "" {
		jsonOut(w, 409, map[string]string{"error": "configure and verify a main reward identity before starting hosted workers"})
		return
	}

	rt, err := s.runtimeForWorker(r.Context(), &wk, true)
	if err != nil {
		jsonOut(w, 503, map[string]string{"error": err.Error()})
		return
	}
	s.workerImageMu.Lock()
	err = s.ensureWorkerImageOn(r.Context(), rt)
	s.workerImageMu.Unlock()
	if err != nil {
		jsonOut(w, 502, map[string]string{"error": err.Error()})
		return
	}

	claimed, err := s.store.ClaimWorkerStart(r.Context(), cid, wid)
	if err != nil {
		jsonOut(w, 500, map[string]string{"error": err.Error()})
		return
	}
	if !claimed {
		jsonOut(w, 409, map[string]string{"error": "worker is already running or starting"})
		return
	}
	ok, newlyCharged, chargedAt, err := s.ensurePaidSlice(r.Context(), wk)
	if err != nil {
		_ = s.store.SetWorkerRuntime(r.Context(), wid, "error", wk.ContainerID, err.Error())
		jsonOut(w, 500, map[string]string{"error": err.Error()})
		return
	}
	if !ok {
		_ = s.store.SetWorkerRuntime(r.Context(), wid, "stopped", wk.ContainerID, "prepaid balance exhausted")
		jsonOut(w, 402, map[string]string{"error": "prepaid balance exhausted"})
		return
	}
	if wk.ContainerID != "" {
		_ = rt.Remove(r.Context(), wk.ContainerID)
	}
	name := "neuralhunt-worker-" + strings.TrimPrefix(wid, "wrk_")
	_ = rt.Remove(r.Context(), name)
	containerID, err := rt.CreateWorker(r.Context(), s.workerContainerConfig(wk, name))
	if err == nil {
		err = rt.Start(r.Context(), containerID)
	}
	if err != nil {
		if containerID != "" {
			_ = rt.Stop(r.Context(), containerID, 2)
			_ = rt.Remove(r.Context(), containerID)
		}
		if newlyCharged {
			_ = s.store.RefundWorkerStartMinute(r.Context(), wk, chargedAt, "runtime_start_failed")
		}
		_ = s.store.SetWorkerRuntime(r.Context(), wid, "error", "", err.Error())
		jsonOut(w, 502, map[string]string{"error": err.Error()})
		return
	}
	if err := s.store.SetWorkerRuntime(r.Context(), wid, "running", containerID, ""); err != nil {
		_ = rt.Stop(r.Context(), containerID, 2)
		_ = rt.Remove(r.Context(), containerID)
		if newlyCharged {
			_ = s.store.RefundWorkerStartMinute(r.Context(), wk, chargedAt, "runtime_state_failed")
		}
		jsonOut(w, 500, map[string]string{"error": "worker state could not be persisted"})
		return
	}
	jsonOut(w, 200, map[string]bool{"ok": true})
}
func (s *Service) stopWorker(w http.ResponseWriter, r *http.Request) {
	cid, wid := customerID(r), chi.URLParam(r, "id")
	wk, err := s.store.Worker(r.Context(), cid, wid)
	if err != nil {
		jsonOut(w, 404, map[string]string{"error": "worker not found"})
		return
	}
	if err := s.stopWorkerInternal(r.Context(), wk, "stopped"); err != nil {
		jsonOut(w, 502, map[string]string{"error": err.Error()})
		return
	}
	jsonOut(w, 200, map[string]bool{"ok": true})
}
func (s *Service) stopWorkerInternal(ctx context.Context, wk Worker, status string) error {
	// Revoke the central lease first. Even when a remote controller is offline,
	// the agent can no longer renew its lease and will self-terminate.
	if err := s.store.SetWorkerRuntime(ctx, wk.ID, status, wk.ContainerID, ""); err != nil {
		return err
	}
	if wk.ContainerID == "" {
		return nil
	}
	rt, err := s.runtimeForWorker(ctx, &wk, false)
	if err != nil {
		return err
	}
	if err := rt.Stop(ctx, wk.ContainerID, 5); err != nil && !strings.Contains(err.Error(), "304") {
		return err
	}
	return nil
}
func (s *Service) deleteWorker(w http.ResponseWriter, r *http.Request) {
	cid, wid := customerID(r), chi.URLParam(r, "id")
	wk, err := s.store.Worker(r.Context(), cid, wid)
	if err != nil {
		jsonOut(w, 404, map[string]string{"error": "worker not found"})
		return
	}
	rt, rtErr := s.runtimeForWorker(r.Context(), &wk, false)
	if wk.ContainerID != "" {
		if rtErr != nil {
			jsonOut(w, 502, map[string]string{"error": "worker host unavailable; refusing to orphan identity volume: " + rtErr.Error()})
			return
		}
		_ = rt.Stop(r.Context(), wk.ContainerID, 3)
		if err := rt.Remove(r.Context(), wk.ContainerID); err != nil {
			jsonOut(w, 502, map[string]string{"error": err.Error()})
			return
		}
	}
	if wk.WorkerClientID != "" {
		_ = s.syncDelegation(r.Context(), wk.WorkerClientID, "")
	}
	if rtErr == nil && rt != nil {
		if err := rt.RemoveVolume(r.Context(), wk.Volume); err != nil {
			jsonOut(w, 502, map[string]string{"error": "identity volume removal failed: " + err.Error()})
			return
		}
	}
	if err := s.store.DeleteWorker(r.Context(), cid, wid); err != nil {
		jsonOut(w, 500, map[string]string{"error": err.Error()})
		return
	}
	jsonOut(w, 200, map[string]bool{"ok": true})
}
func (s *Service) workerIdentityGet(w http.ResponseWriter, r *http.Request) {
	wk, err := s.store.Worker(r.Context(), customerID(r), chi.URLParam(r, "id"))
	if err != nil {
		jsonOut(w, 404, map[string]string{"error": "worker not found"})
		return
	}
	if wk.ContainerID == "" {
		jsonOut(w, 409, map[string]string{"error": "worker identity does not exist yet; start the worker once or upload an identity first"})
		return
	}
	rt, err := s.runtimeForWorker(r.Context(), &wk, false)
	if err != nil {
		jsonOut(w, 502, map[string]string{"error": err.Error()})
		return
	}
	b, err := rt.GetFile(r.Context(), wk.ContainerID, "/identity/identity.json")
	if err != nil {
		jsonOut(w, 502, map[string]string{"error": err.Error()})
		return
	}
	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("Content-Disposition", `attachment; filename="neuralhunt-worker-identity.json"`)
	w.Header().Set("Cache-Control", "no-store")
	_, _ = w.Write(b)
}
func (s *Service) workerIdentityPut(w http.ResponseWriter, r *http.Request) {
	wk, err := s.store.Worker(r.Context(), customerID(r), chi.URLParam(r, "id"))
	if err != nil {
		jsonOut(w, 404, map[string]string{"error": "worker not found"})
		return
	}
	wk, err = s.ensureWorkerContainer(r.Context(), wk)
	if err != nil {
		jsonOut(w, 502, map[string]string{"error": err.Error()})
		return
	}
	rt, err := s.runtimeForWorker(r.Context(), &wk, false)
	if err != nil {
		jsonOut(w, 502, map[string]string{"error": err.Error()})
		return
	}
	b, err := io.ReadAll(io.LimitReader(r.Body, 256<<10))
	if err != nil || len(b) == 0 {
		jsonOut(w, 400, map[string]string{"error": "identity JSON required"})
		return
	}
	if err := ValidateRawIdentity(b); err != nil {
		jsonOut(w, 400, map[string]string{"error": err.Error()})
		return
	}
	if wk.Status == "running" {
		_ = rt.Stop(r.Context(), wk.ContainerID, 5)
	}
	if err := rt.PutFile(r.Context(), wk.ContainerID, "/identity", "identity.json", b); err != nil {
		jsonOut(w, 502, map[string]string{"error": err.Error()})
		return
	}
	_ = s.store.SetWorkerClient(r.Context(), wk.ID, "")
	_ = s.store.SetWorkerRuntime(r.Context(), wk.ID, "stopped", wk.ContainerID, "")
	jsonOut(w, 200, map[string]bool{"ok": true})
}

func (s *Service) authenticateWorkerInternal(r *http.Request, workerID string) (Worker, bool) {
	wk, err := s.store.WorkerByID(r.Context(), workerID)
	if err != nil {
		return Worker{}, false
	}
	provided := strings.TrimSpace(strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer "))
	if len(provided) != len(wk.RegisterToken) || subtle.ConstantTimeCompare([]byte(provided), []byte(wk.RegisterToken)) != 1 {
		return Worker{}, false
	}
	return wk, true
}

func (s *Service) internalWorkerLease(w http.ResponseWriter, r *http.Request) {
	var in struct {
		WorkerID string `json:"worker_id"`
	}
	if decode(r, &in) != nil || strings.TrimSpace(in.WorkerID) == "" {
		jsonOut(w, 400, map[string]string{"error": "worker_id required"})
		return
	}
	wk, ok := s.authenticateWorkerInternal(r, in.WorkerID)
	if !ok {
		jsonOut(w, 401, map[string]string{"error": "unauthorized"})
		return
	}
	if wk.Status != "running" && wk.Status != "starting" {
		jsonOut(w, 409, map[string]string{"error": "worker lease revoked"})
		return
	}
	if cust, err := s.store.CustomerByID(r.Context(), wk.CustomerID); err != nil || cust.Blocked {
		jsonOut(w, 409, map[string]string{"error": "worker lease revoked"})
		return
	}
	_ = s.store.TouchWorkerLease(r.Context(), wk.ID)
	jsonOut(w, 200, map[string]any{"ok": true, "lease_sec": 45})
}

func (s *Service) internalControllerRegister(w http.ResponseWriter, r *http.Request) {
	secret := strings.TrimSpace(s.cfg.ControllerSharedSecret)
	provided := strings.TrimSpace(strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer "))
	if len(secret) < 32 || len(provided) != len(secret) || subtle.ConstantTimeCompare([]byte(provided), []byte(secret)) != 1 {
		jsonOut(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"})
		return
	}
	var in struct {
		ID              string `json:"id"`
		Name            string `json:"name"`
		BaseURL         string `json:"base_url"`
		ProtocolVersion int    `json:"protocol_version"`
		MaxWorkers      int    `json:"max_workers"`
		MaxRunning      int    `json:"max_running"`
		ReportedWorkers int    `json:"reported_workers"`
		ReportedRunning int    `json:"reported_running"`
	}
	if decode(r, &in) != nil || strings.TrimSpace(in.ID) == "" || strings.TrimSpace(in.BaseURL) == "" {
		jsonOut(w, 400, map[string]string{"error": "id and base_url required"})
		return
	}
	u, err := url.Parse(strings.TrimSpace(in.BaseURL))
	if err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") {
		jsonOut(w, 400, map[string]string{"error": "invalid base_url"})
		return
	}
	if in.ProtocolVersion != 1 {
		jsonOut(w, http.StatusConflict, map[string]string{"error": fmt.Sprintf("unsupported Service Controller protocol version %d (Master supports 1)", in.ProtocolVersion)})
		return
	}
	if in.MaxWorkers <= 0 || in.MaxWorkers > 100000 || in.MaxRunning <= 0 || in.MaxRunning > 100000 || in.ReportedWorkers < 0 || in.ReportedRunning < 0 {
		jsonOut(w, 400, map[string]string{"error": "invalid capacity"})
		return
	}
	baseURL := strings.TrimRight(strings.TrimSpace(in.BaseURL), "/")
	// Registration is bidirectional: accept a heartbeat only when the Master can
	// reach the advertised private control endpoint as well.
	rt, rtErr := NewRemoteControllerClient(baseURL, s.cfg.ControllerSharedSecret)
	if rtErr != nil {
		jsonOut(w, 400, map[string]string{"error": rtErr.Error()})
		return
	}
	pingCtx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
	pingErr := rt.Ping(pingCtx)
	cancel()
	if pingErr != nil {
		jsonOut(w, http.StatusBadGateway, map[string]string{"error": "Master cannot reach advertised controller URL: " + pingErr.Error()})
		return
	}
	name := strings.TrimSpace(in.Name)
	if name == "" {
		name = in.ID
	}
	if err := s.store.UpsertServiceController(r.Context(), ServiceController{ID: strings.TrimSpace(in.ID), Name: name, BaseURL: baseURL, ProtocolVersion: in.ProtocolVersion, MaxWorkers: in.MaxWorkers, MaxRunning: in.MaxRunning, ReportedWorkers: in.ReportedWorkers, ReportedRunning: in.ReportedRunning}); err != nil {
		jsonOut(w, 500, map[string]string{"error": err.Error()})
		return
	}
	jsonOut(w, 200, map[string]any{"ok": true, "controller_id": strings.TrimSpace(in.ID), "master_mode": s.cfg.WorkerOrchestrationMode})
}

func (s *Service) internalWorkerRegister(w http.ResponseWriter, r *http.Request) {
	var in struct {
		WorkerID string `json:"worker_id"`
		ClientID string `json:"client_id"`
	}
	if decode(r, &in) != nil {
		jsonOut(w, 400, map[string]string{"error": "bad json"})
		return
	}
	wk, ok := s.authenticateWorkerInternal(r, in.WorkerID)
	if !ok {
		jsonOut(w, 401, map[string]string{"error": "unauthorized"})
		return
	}
	// RestartPolicy=unless-stopped is a recovery mechanism, never a lease bypass.
	// A container that was explicitly revoked/blocked must fail before it can
	// reconnect to the game and submit guesses.
	if wk.Status != "running" && wk.Status != "starting" {
		jsonOut(w, 409, map[string]string{"error": "worker lease revoked"})
		return
	}
	if cust, err := s.store.CustomerByID(r.Context(), wk.CustomerID); err != nil || cust.Blocked {
		jsonOut(w, 409, map[string]string{"error": "worker lease revoked"})
		return
	}
	if err := s.store.SetWorkerClient(r.Context(), wk.ID, in.ClientID); err != nil {
		jsonOut(w, 500, map[string]string{"error": err.Error()})
		return
	}
	_ = s.store.TouchWorkerLease(r.Context(), wk.ID)
	cust, _ := s.store.CustomerByID(r.Context(), wk.CustomerID)
	if err := s.syncDelegation(r.Context(), in.ClientID, cust.RewardClientID); err != nil {
		_ = s.store.SetWorkerRuntime(r.Context(), wk.ID, "error", wk.ContainerID, "delegation failed: "+err.Error())
		jsonOut(w, 502, map[string]string{"error": "delegation failed: " + err.Error()})
		return
	}
	jsonOut(w, 200, map[string]bool{"ok": true})
}

// CheckGameControlPlane verifies that the configured private game listener is
// reachable and that both services use the same hosted-service secret. It is
// safe to call at startup and does not consume a Hosted Code.
func (s *Service) CheckGameControlPlane(ctx context.Context) error {
	return s.checkGameControlPlane(ctx)
}

func (s *Service) checkGameControlPlane(ctx context.Context) error {
	if strings.TrimSpace(s.cfg.SharedSecret) == "" {
		return errors.New("CUSTOMER_SERVICE_SHARED_SECRET fehlt")
	}
	base := strings.TrimRight(strings.TrimSpace(s.cfg.GameAdminURL), "/")
	if base == "" {
		return errors.New("CS_GAME_ADMIN_URL fehlt")
	}
	endpoint := base + "/api/internal/customer-service/health"
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
	if err != nil {
		return err
	}
	req.Header.Set("Authorization", "Bearer "+s.cfg.SharedSecret)
	resp, err := s.hc.Do(req)
	if err != nil {
		return fmt.Errorf("Game-Control-Plane unter %s nicht erreichbar: %w. Bei lokalem 'go run' normalerweise CS_GAME_ADMIN_URL=http://127.0.0.1:8081; im Compose-Netz http://app:8081", base, err)
	}
	defer resp.Body.Close()
	b, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
	if resp.StatusCode == http.StatusUnauthorized {
		return errors.New("Game-Control-Plane erreichbar, aber CUSTOMER_SERVICE_SHARED_SECRET stimmt zwischen Game Server und Customer Service nicht überein")
	}
	if resp.StatusCode == http.StatusNotFound {
		return fmt.Errorf("Game-Control-Plane antwortet mit HTTP 404. CS_GAME_ADMIN_URL=%s zeigt wahrscheinlich auf den öffentlichen Listener/Reverse-Proxy oder auf ein älteres Game-Image ohne Hosted-Control-Plane", base)
	}
	if resp.StatusCode/100 != 2 {
		return fmt.Errorf("Game-Control-Plane HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(b)))
	}
	var out struct {
		OK                    bool   `json:"ok"`
		Service               string `json:"service"`
		CustomerLinkSupported bool   `json:"customer_link_supported"`
	}
	if err := json.Unmarshal(b, &out); err != nil {
		return fmt.Errorf("Game-Control-Plane liefert keine gültige JSON-Antwort: %w", err)
	}
	if !out.OK || out.Service != "neuralhunt-game-control-plane" || !out.CustomerLinkSupported {
		return errors.New("Game-Control-Plane ist erreichbar, unterstützt aber die erwartete Hosted-Code-Schnittstelle nicht")
	}
	return nil
}

func (s *Service) redeemRewardLink(ctx context.Context, code string) (string, error) {
	if err := s.checkGameControlPlane(ctx); err != nil {
		return "", err
	}
	body, _ := json.Marshal(map[string]string{"code": strings.TrimSpace(code)})
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(s.cfg.GameAdminURL, "/")+"/api/internal/customer-link/consume", strings.NewReader(string(body)))
	if err != nil {
		return "", err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+s.cfg.SharedSecret)
	resp, err := s.hc.Do(req)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	b, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
	if resp.StatusCode == http.StatusUnauthorized {
		return "", errors.New("Game Server lehnt CUSTOMER_SERVICE_SHARED_SECRET ab")
	}
	if resp.StatusCode == http.StatusNotFound {
		return "", errors.New("Hosted-Code ist abgelaufen, ungültig oder bereits verwendet")
	}
	if resp.StatusCode/100 != 2 {
		return "", fmt.Errorf("Game-Pairing HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(b)))
	}
	var out struct {
		ClientID string `json:"client_id"`
	}
	if err := json.Unmarshal(b, &out); err != nil {
		return "", err
	}
	if strings.TrimSpace(out.ClientID) == "" {
		return "", errors.New("game returned empty reward identity")
	}
	return strings.TrimSpace(out.ClientID), nil
}

func (s *Service) rewardIdentityExists(ctx context.Context, clientID string) (bool, error) {
	if strings.TrimSpace(s.cfg.SharedSecret) == "" {
		return false, errors.New("CUSTOMER_SERVICE_SHARED_SECRET missing")
	}
	body, _ := json.Marshal(map[string]string{"client_id": clientID})
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(s.cfg.GameAdminURL, "/")+"/api/internal/identity-exists", strings.NewReader(string(body)))
	if err != nil {
		return false, err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+s.cfg.SharedSecret)
	resp, err := s.hc.Do(req)
	if err != nil {
		return false, err
	}
	defer resp.Body.Close()
	b, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
	if resp.StatusCode/100 != 2 {
		return false, fmt.Errorf("game identity HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(b)))
	}
	var out struct {
		Exists bool `json:"exists"`
	}
	if err := json.Unmarshal(b, &out); err != nil {
		return false, err
	}
	return out.Exists, nil
}

func (s *Service) syncDelegation(ctx context.Context, worker, owner string) error {
	if strings.TrimSpace(s.cfg.SharedSecret) == "" {
		return errors.New("CUSTOMER_SERVICE_SHARED_SECRET missing")
	}
	body, _ := json.Marshal(map[string]string{"worker_client_id": worker, "owner_client_id": owner})
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(s.cfg.GameAdminURL, "/")+"/api/internal/delegations", strings.NewReader(string(body)))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+s.cfg.SharedSecret)
	resp, err := s.hc.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	b, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
	if resp.StatusCode/100 != 2 {
		return fmt.Errorf("game delegation HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(b)))
	}
	return nil
}

func (s *Service) RunBilling(ctx context.Context) {
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			s.billOnce(ctx)
		}
	}
}
func (s *Service) billOnce(ctx context.Context) {
	workers, err := s.store.RunningWorkers(ctx)
	if err != nil {
		log.Printf("customer billing: %v", err)
		return
	}
	now := time.Now().UTC()
	for _, wk := range workers {
		if cust, err := s.store.CustomerByID(ctx, wk.CustomerID); err == nil && cust.Blocked {
			_ = s.stopWorkerInternal(ctx, wk, "stopped")
			log.Printf("billing worker %s stopped: customer account blocked", wk.ID)
			continue
		}
		if wk.ContainerID == "" {
			continue
		}
		// A fresh worker lease is the most reliable liveness signal in distributed
		// mode because it reaches the Master independently of the controller API.
		alive := wk.LastLeaseAt != nil && now.Sub(*wk.LastLeaseAt) <= 90*time.Second
		if !alive {
			rt, rtErr := s.runtimeForWorker(ctx, &wk, false)
			if rtErr == nil {
				running, runErr := rt.Running(ctx, wk.ContainerID)
				if runErr == nil && running {
					alive = true
				}
			}
		}
		if !alive {
			_ = s.store.SetWorkerRuntime(ctx, wk.ID, "stopped", wk.ContainerID, "worker lease/liveness expired")
			continue
		}
		last := wk.LastChargeAt
		if last == nil {
			t := now
			last = &t
		}
		for now.Sub(*last) >= time.Minute {
			next := last.Add(time.Minute)
			ok, err := s.store.ChargeWorkerMinute(ctx, wk, next)
			if err != nil {
				log.Printf("billing worker %s: %v", wk.ID, err)
				break
			}
			if !ok {
				_ = s.stopWorkerInternal(ctx, wk, "stopped")
				log.Printf("billing worker %s stopped: prepaid balance exhausted", wk.ID)
				break
			}
			last = &next
			wk.LastChargeAt = last
		}
	}
}

func (s *Service) billingPackages(w http.ResponseWriter, r *http.Request) {
	enabled := s.paypalAllowed()
	packages := s.cfg.Packages
	if !enabled {
		packages = nil
	}
	jsonOut(w, 200, map[string]any{"paypal_enabled": enabled, "environment": s.cfg.PayPalEnvironment, "packages": packages})
}
func (s *Service) paypalAllowed() bool {
	if !s.cfg.PayPalEnabled || s.paypal == nil || !s.paypal.Ready() {
		return false
	}
	if strings.EqualFold(s.cfg.PayPalEnvironment, "live") && s.cfg.PayPalLiveApprovalAck != "I_HAVE_PAYPAL_APPROVAL" {
		return false
	}
	return true
}
func (s *Service) packageByID(id string) (CreditPackage, bool) {
	for _, p := range s.cfg.Packages {
		if p.ID == id {
			return p, true
		}
	}
	return CreditPackage{}, false
}
func centsValue(c int64) string {
	return fmt.Sprintf("%d.%02d", c/100, int64(math.Abs(float64(c%100))))
}
func parseMoneyCents(v string) (int64, error) {
	v = strings.TrimSpace(v)
	if v == "" || strings.HasPrefix(v, "-") || strings.HasPrefix(v, "+") {
		return 0, errors.New("invalid money value")
	}
	whole, frac, hasDot := strings.Cut(v, ".")
	if whole == "" {
		whole = "0"
	}
	if hasDot {
		if len(frac) > 2 {
			return 0, errors.New("money value has more than two decimals")
		}
		for len(frac) < 2 {
			frac += "0"
		}
	} else {
		frac = "00"
	}
	w, err := strconv.ParseInt(whole, 10, 64)
	if err != nil || w < 0 {
		return 0, errors.New("invalid money value")
	}
	f, err := strconv.ParseInt(frac, 10, 64)
	if err != nil || f < 0 || f > 99 {
		return 0, errors.New("invalid money value")
	}
	if w > (math.MaxInt64-f)/100 {
		return 0, errors.New("money value too large")
	}
	return w*100 + f, nil
}
func (s *Service) paypalCreateOrder(w http.ResponseWriter, r *http.Request) {
	if !s.paypalAllowed() {
		jsonOut(w, 503, map[string]string{"error": "PayPal is disabled or live approval acknowledgement is missing"})
		return
	}
	var in struct {
		PackageID string `json:"package_id"`
	}
	if decode(r, &in) != nil {
		jsonOut(w, 400, map[string]string{"error": "bad json"})
		return
	}
	pkg, ok := s.packageByID(in.PackageID)
	if !ok {
		jsonOut(w, 400, map[string]string{"error": "unknown package"})
		return
	}
	cid := customerID(r)
	base := strings.TrimRight(s.cfg.PublicBaseURL, "/")
	order, err := s.paypal.CreateOrder(r.Context(), "credits:"+cid+":"+pkg.ID, "Neural Hunt hosted worker credits", centsValue(pkg.AmountCents), pkg.Currency, base+"/?paypal=return", base+"/?paypal=cancel")
	if err != nil {
		jsonOut(w, 502, map[string]string{"error": err.Error()})
		return
	}
	approve := ApprovalURL(order)
	if order.ID == "" || approve == "" {
		jsonOut(w, 502, map[string]string{"error": "PayPal did not return an approval URL"})
		return
	}
	if err := s.store.UpsertPayPalOrder(r.Context(), order.ID, cid, pkg.ID, pkg.AmountCents, pkg.Currency, pkg.CreditsMicros, order.Status); err != nil {
		jsonOut(w, 500, map[string]string{"error": err.Error()})
		return
	}
	jsonOut(w, 201, map[string]string{"order_id": order.ID, "approval_url": approve})
}
func (s *Service) paypalCapture(w http.ResponseWriter, r *http.Request) {
	if !s.paypalAllowed() {
		jsonOut(w, 503, map[string]string{"error": "PayPal disabled"})
		return
	}
	var in struct {
		OrderID string `json:"order_id"`
	}
	if decode(r, &in) != nil {
		jsonOut(w, 400, map[string]string{"error": "bad json"})
		return
	}
	stored, err := s.store.PayPalOrder(r.Context(), in.OrderID)
	if err != nil || stored.CustomerID != customerID(r) {
		jsonOut(w, 404, map[string]string{"error": "order not found"})
		return
	}
	view, err := s.paypal.CaptureOrder(r.Context(), stored.OrderID)
	if err != nil {
		view, err = s.paypal.GetOrder(r.Context(), stored.OrderID)
	}
	if err != nil {
		jsonOut(w, 502, map[string]string{"error": err.Error()})
		return
	}
	if err := s.reconcilePayPal(r.Context(), stored, view); err != nil {
		jsonOut(w, 409, map[string]string{"error": err.Error()})
		return
	}
	bal, _ := s.store.BalanceMicros(r.Context(), stored.CustomerID)
	jsonOut(w, 200, map[string]any{"ok": true, "balance_micros": bal})
}
func (s *Service) reconcilePayPal(ctx context.Context, stored PayPalOrder, view PayPalOrderView) error {
	if !strings.EqualFold(view.Status, "COMPLETED") {
		return fmt.Errorf("PayPal order status is %s", view.Status)
	}
	if len(view.PurchaseUnits) < 1 {
		return errors.New("PayPal order has no purchase unit")
	}
	cents, err := parseMoneyCents(view.PurchaseUnits[0].Amount.Value)
	if err != nil {
		return err
	}
	if cents != stored.AmountCents || !strings.EqualFold(view.PurchaseUnits[0].Amount.CurrencyCode, stored.Currency) {
		return errors.New("PayPal captured amount/currency does not match requested credit package")
	}
	captureID := ""
	for _, c := range view.PurchaseUnits[0].Payments.Captures {
		if !strings.EqualFold(c.Status, "COMPLETED") {
			continue
		}
		cc, err := parseMoneyCents(c.Amount.Value)
		if err == nil && cc == stored.AmountCents && strings.EqualFold(c.Amount.CurrencyCode, stored.Currency) {
			captureID = c.ID
			break
		}
	}
	if captureID == "" {
		return errors.New("PayPal order has no matching completed capture")
	}
	return s.store.CompletePayPalOrder(ctx, stored.OrderID, captureID)
}
func findOrderID(v any) string {
	switch x := v.(type) {
	case map[string]any:
		if sd, ok := x["supplementary_data"].(map[string]any); ok {
			if rel, ok := sd["related_ids"].(map[string]any); ok {
				if id, ok := rel["order_id"].(string); ok {
					return id
				}
			}
		}
		for k, v := range x {
			if k == "order_id" {
				if id, ok := v.(string); ok {
					return id
				}
			}
			if id := findOrderID(v); id != "" {
				return id
			}
		}
	case []any:
		for _, v := range x {
			if id := findOrderID(v); id != "" {
				return id
			}
		}
	}
	return ""
}
func (s *Service) paypalWebhook(w http.ResponseWriter, r *http.Request) {
	if !s.paypalAllowed() {
		jsonOut(w, 503, map[string]string{"error": "PayPal disabled"})
		return
	}
	raw, err := io.ReadAll(io.LimitReader(r.Body, 2<<20))
	if err != nil {
		jsonOut(w, 400, map[string]string{"error": "bad body"})
		return
	}
	ok, err := s.paypal.VerifyWebhook(r.Context(), s.cfg.PayPalWebhookID, r.Header, raw)
	if err != nil || !ok {
		jsonOut(w, 401, map[string]string{"error": "invalid PayPal webhook signature"})
		return
	}
	var ev any
	if json.Unmarshal(raw, &ev) != nil {
		jsonOut(w, 400, map[string]string{"error": "bad event"})
		return
	}
	orderID := findOrderID(ev)
	if orderID != "" {
		if stored, err := s.store.PayPalOrder(r.Context(), orderID); err == nil {
			if view, err := s.paypal.GetOrder(r.Context(), orderID); err == nil {
				_ = s.reconcilePayPal(r.Context(), stored, view)
			}
		}
	}
	jsonOut(w, 200, map[string]bool{"ok": true})
}

func (s *Service) adminLogin(w http.ResponseWriter, r *http.Request) {
	var in struct{ Username, Password string }
	if decode(r, &in) != nil {
		jsonOut(w, 400, map[string]string{"error": "bad json"})
		return
	}
	uok := len(in.Username) == len(s.cfg.AdminUser) && subtle.ConstantTimeCompare([]byte(in.Username), []byte(s.cfg.AdminUser)) == 1
	pok := len(in.Password) == len(s.cfg.AdminPassword) && subtle.ConstantTimeCompare([]byte(in.Password), []byte(s.cfg.AdminPassword)) == 1
	if !uok || !pok {
		time.Sleep(300 * time.Millisecond)
		jsonOut(w, 401, map[string]string{"error": "invalid credentials"})
		return
	}
	sid := RandomToken(32)
	s.adminSessions.Store(sid, time.Now().Add(12*time.Hour))
	s.setCookie(w, r, csAdminCookie, sid, 12*time.Hour)
	jsonOut(w, 200, map[string]bool{"ok": true})
}
func (s *Service) requireAdmin(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		c, err := r.Cookie(csAdminCookie)
		if err != nil {
			jsonOut(w, 401, map[string]string{"error": "unauthorized"})
			return
		}
		v, ok := s.adminSessions.Load(c.Value)
		exp, _ := v.(time.Time)
		if !ok || time.Now().After(exp) {
			s.adminSessions.Delete(c.Value)
			jsonOut(w, 401, map[string]string{"error": "unauthorized"})
			return
		}
		next.ServeHTTP(w, r)
	})
}
func (s *Service) adminLogout(w http.ResponseWriter, r *http.Request) {
	if c, err := r.Cookie(csAdminCookie); err == nil {
		s.adminSessions.Delete(c.Value)
	}
	s.clearCookie(w, r, csAdminCookie)
	jsonOut(w, 200, map[string]bool{"ok": true})
}
func (s *Service) adminOverview(w http.ResponseWriter, r *http.Request) {
	rows, err := s.store.DB.QueryContext(r.Context(), `SELECT c.id,c.username,c.reward_client_id,c.blocked,c.blocked_reason,c.blocked_at,c.worker_limit_bypass,c.created_at,COALESCE((SELECT sum(delta_micros) FROM credit_ledger l WHERE l.customer_id=c.id),0) FROM customers c ORDER BY c.created_at DESC LIMIT 500`)
	if err != nil {
		jsonOut(w, 500, map[string]string{"error": err.Error()})
		return
	}
	defer rows.Close()
	var out []map[string]any
	for rows.Next() {
		var id, user, reward, blockedReason string
		var blocked, workerLimitBypass int
		var blockedAt sql.NullInt64
		var created, bal int64
		if err := rows.Scan(&id, &user, &reward, &blocked, &blockedReason, &blockedAt, &workerLimitBypass, &created, &bal); err != nil {
			continue
		}
		workers, _ := s.store.Workers(r.Context(), id)
		running := 0
		for _, wk := range workers {
			if wk.Status == "running" {
				running++
			}
		}
		item := map[string]any{
			"id": id, "username": user, "reward_client_id": reward,
			"blocked": blocked != 0, "blocked_reason": blockedReason, "worker_limit_bypass": workerLimitBypass != 0,
			"created_at": time.UnixMilli(created).UTC(), "balance_micros": bal,
			"workers": len(workers), "running": running, "worker_items": workers,
		}
		if blockedAt.Valid {
			item["blocked_at"] = time.UnixMilli(blockedAt.Int64).UTC()
		}
		out = append(out, item)
	}
	controllers, _ := s.store.ServiceControllers(r.Context())
	controllerItems := make([]map[string]any, 0, len(controllers))
	for _, c := range controllers {
		total, running, _ := s.store.ControllerWorkerCounts(r.Context(), c.ID)
		controllerItems = append(controllerItems, map[string]any{"id": c.ID, "name": c.Name, "base_url": c.BaseURL, "protocol_version": c.ProtocolVersion, "max_workers": c.MaxWorkers, "max_running": c.MaxRunning, "reported_workers": c.ReportedWorkers, "reported_running": c.ReportedRunning, "assigned_workers": total, "assigned_running": running, "enabled": c.Enabled, "draining": c.Draining, "online": time.Since(c.LastSeen) <= s.cfg.ControllerOfflineAfter, "last_seen": c.LastSeen, "last_error": c.LastError})
	}
	jsonOut(w, 200, map[string]any{
		"manual_credits_enabled": s.cfg.AllowManualCredits,
		"paypal_enabled":         s.paypalAllowed(),
		"worker_image":           s.cfg.WorkerImage,
		"orchestration_mode":     s.cfg.WorkerOrchestrationMode,
		"controllers":            controllerItems,
		"settings":               s.portalSettings(r.Context()),
		"customers":              out,
	})
}

func (s *Service) adminUpdateWorkerImage(w http.ResponseWriter, r *http.Request) {
	wid := strings.TrimSpace(chi.URLParam(r, "id"))
	wk, err := s.store.WorkerByID(r.Context(), wid)
	if err != nil {
		jsonOut(w, http.StatusNotFound, map[string]string{"error": "worker not found"})
		return
	}
	rt, err := s.runtimeForWorker(r.Context(), &wk, true)
	if err != nil {
		jsonOut(w, http.StatusServiceUnavailable, map[string]string{"error": err.Error()})
		return
	}
	s.workerImageMu.Lock()
	err = s.pullWorkerImageOn(r.Context(), rt)
	s.workerImageMu.Unlock()
	if err != nil {
		jsonOut(w, http.StatusBadGateway, map[string]string{"error": err.Error()})
		return
	}
	if err := s.recreateWorkerWithCurrentImage(r.Context(), wk); err != nil {
		jsonOut(w, http.StatusBadGateway, map[string]string{"error": "worker image update: " + err.Error()})
		return
	}
	jsonOut(w, http.StatusOK, map[string]any{"ok": true, "worker_id": wk.ID, "image": s.cfg.WorkerImage, "controller_id": wk.ControllerID, "resumed": wk.Status == "running" || wk.Status == "starting"})
}

func (s *Service) adminUpdateAllWorkerImages(w http.ResponseWriter, r *http.Request) {
	rows, err := s.store.DB.QueryContext(r.Context(), `SELECT `+workerCols+` FROM workers ORDER BY created_at`)
	if err != nil {
		jsonOut(w, 500, map[string]string{"error": err.Error()})
		return
	}
	var workers []Worker
	for rows.Next() {
		wk, scanErr := scanWorker(rows)
		if scanErr != nil {
			_ = rows.Close()
			jsonOut(w, 500, map[string]string{"error": scanErr.Error()})
			return
		}
		workers = append(workers, wk)
	}
	_ = rows.Close()
	// Pull once per distinct worker host, then recreate each worker on the host
	// that owns its persistent identity volume.
	pullState := map[string]error{}
	updated := 0
	var warnings []string
	for i := range workers {
		wk := workers[i]
		rt, rtErr := s.runtimeForWorker(r.Context(), &wk, true)
		if rtErr != nil {
			warnings = append(warnings, wk.ID+": "+rtErr.Error())
			continue
		}
		key := wk.ControllerID
		if key == "" {
			key = "local"
		}
		pullErr, seen := pullState[key]
		if !seen {
			s.workerImageMu.Lock()
			pullErr = s.pullWorkerImageOn(r.Context(), rt)
			s.workerImageMu.Unlock()
			pullState[key] = pullErr
			if pullErr != nil {
				warnings = append(warnings, "host "+key+": "+pullErr.Error())
			}
		}
		if pullErr != nil {
			continue
		}
		if err := s.recreateWorkerWithCurrentImage(r.Context(), wk); err != nil {
			warnings = append(warnings, wk.ID+": "+err.Error())
			continue
		}
		updated++
	}
	jsonOut(w, http.StatusOK, map[string]any{"ok": len(warnings) == 0, "image": s.cfg.WorkerImage, "updated": updated, "total": len(workers), "hosts": len(pullState), "warnings": warnings})
}

func (s *Service) adminUpdateControllerPolicy(w http.ResponseWriter, r *http.Request) {
	id := strings.TrimSpace(chi.URLParam(r, "id"))
	if id == "" {
		jsonOut(w, 400, map[string]string{"error": "controller id required"})
		return
	}
	var in struct {
		Enabled  bool `json:"enabled"`
		Draining bool `json:"draining"`
	}
	if decode(r, &in) != nil {
		jsonOut(w, 400, map[string]string{"error": "bad json"})
		return
	}
	if err := s.store.SetServiceControllerPolicy(r.Context(), id, in.Enabled, in.Draining); err != nil {
		jsonOut(w, 500, map[string]string{"error": err.Error()})
		return
	}
	jsonOut(w, 200, map[string]any{"ok": true, "id": id, "enabled": in.Enabled, "draining": in.Draining})
}

func (s *Service) adminCreditGrant(w http.ResponseWriter, r *http.Request) {
	if !s.cfg.AllowManualCredits {
		jsonOut(w, 403, map[string]string{"error": "manual credit bypass disabled; set CS_ALLOW_MANUAL_CREDITS=1 on the private admin service"})
		return
	}
	var in struct {
		CustomerID string  `json:"customer_id"`
		Credits    float64 `json:"credits"`
		Reason     string  `json:"reason"`
	}
	if decode(r, &in) != nil || in.Credits <= 0 || in.Credits > 1_000_000 {
		jsonOut(w, 400, map[string]string{"error": "invalid grant"})
		return
	}
	micros := int64(math.Round(in.Credits * 1_000_000))
	ref := "manual:" + RandomToken(12)
	reason := "manual_test_grant"
	if strings.TrimSpace(in.Reason) != "" {
		reason += ":" + strings.TrimSpace(in.Reason)
	}
	if err := s.store.AddLedger(r.Context(), strings.TrimSpace(in.CustomerID), micros, reason, ref); err != nil {
		jsonOut(w, 400, map[string]string{"error": err.Error()})
		return
	}
	jsonOut(w, 200, map[string]any{"ok": true, "reference": ref})
}

// Keep sql imported in this file because several handlers intentionally treat
// sql.ErrNoRows as a 404 without leaking database details.
var _ = sql.ErrNoRows
var _ = url.PathEscape