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 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 *DockerClient paypal *PayPalClient cfg Config adminSessions sync.Map workerImageMu sync.Mutex hc *http.Client } func NewService(store *Store, docker *DockerClient, 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 } 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.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.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 } 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 } 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 != "" { _ = s.docker.Stop(r.Context(), wk.ContainerID, 5) _ = s.docker.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) ensureWorkerImage(ctx context.Context) error { // Avoid several customer requests triggering concurrent pulls of the same // image. The local inspect is cheap once the image is available. s.workerImageMu.Lock() defer s.workerImageMu.Unlock() auth, err := RegistryAuthHeader(s.cfg.WorkerRegistryUsername, s.cfg.WorkerRegistryPassword, s.cfg.WorkerRegistryServer) if err != nil { return err } return s.docker.EnsureImage(ctx, s.cfg.WorkerImage, s.cfg.WorkerAutoPull, auth) } // pullWorkerImage always asks Docker to refresh CS_WORKER_IMAGE, even if the // tag already exists locally. This is intentionally admin-only behavior for // mutable release tags such as worker_latest; normal worker starts keep the // cheaper EnsureImage behavior. func (s *Service) pullWorkerImage(ctx context.Context) error { s.workerImageMu.Lock() defer s.workerImageMu.Unlock() auth, err := RegistryAuthHeader(s.cfg.WorkerRegistryUsername, s.cfg.WorkerRegistryPassword, s.cfg.WorkerRegistryServer) if err != nil { return err } if err := s.docker.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 { 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 != "" { _ = s.docker.Stop(ctx, oldID, 5) if err := s.docker.Remove(ctx, oldID); err != nil { return fmt.Errorf("remove old worker container: %w", err) } } name := "neuralhunt-worker-" + strings.TrimPrefix(wk.ID, "wrk_") // Clear a crash-orphan with the deterministic name. Remove is idempotent for // 404s and never removes the named identity volume (v=false). _ = s.docker.Remove(ctx, name) containerID, err := s.docker.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 := s.docker.Start(ctx, containerID); err != nil { _ = s.docker.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 { _ = s.docker.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 } if err := s.ensureWorkerImage(ctx); err != nil { return wk, err } name := "neuralhunt-worker-" + strings.TrimPrefix(wk.ID, "wrk_") id, err := s.docker.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 { _ = s.docker.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 } if err := s.ensureWorkerImage(r.Context()); 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 != "" { _ = s.docker.Remove(r.Context(), wk.ContainerID) } name := "neuralhunt-worker-" + strings.TrimPrefix(wid, "wrk_") _ = s.docker.Remove(r.Context(), name) // clears a crash-orphan with the deterministic name containerID, err := s.docker.CreateWorker(r.Context(), s.workerContainerConfig(wk, name)) if err == nil { err = s.docker.Start(r.Context(), containerID) } if err != nil { if containerID != "" { _ = s.docker.Stop(r.Context(), containerID, 2) _ = s.docker.Remove(r.Context(), containerID) } if newlyCharged { _ = s.store.RefundWorkerStartMinute(r.Context(), wk, chargedAt, "docker_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 { _ = s.docker.Stop(r.Context(), containerID, 2) _ = s.docker.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]any{"ok": true, "container_id": containerID}) } 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 lease in SQLite first. Even if Docker itself is temporarily // unavailable, the worker's next lease renewal will fail and its agent will // self-terminate instead of continuing with a blocked/stopped account. if err := s.store.SetWorkerRuntime(ctx, wk.ID, status, wk.ContainerID, ""); err != nil { return err } if wk.ContainerID != "" { if err := s.docker.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 } if wk.ContainerID != "" { _ = s.docker.Stop(r.Context(), wk.ContainerID, 3) _ = s.docker.Remove(r.Context(), wk.ContainerID) } if wk.WorkerClientID != "" { _ = s.syncDelegation(r.Context(), wk.WorkerClientID, "") } if err := s.store.DeleteWorker(r.Context(), cid, wid); err != nil { jsonOut(w, 500, map[string]string{"error": err.Error()}) return } _ = s.docker.RemoveVolume(r.Context(), wk.Volume) 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 } b, err := s.docker.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 } 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" { _ = s.docker.Stop(r.Context(), wk.ContainerID, 5) } if err := s.docker.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 } jsonOut(w, 200, map[string]any{"ok": true, "lease_sec": 45}) } 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 } 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 } running, err := s.docker.Running(ctx, wk.ContainerID) if err != nil || !running { _ = s.store.SetWorkerRuntime(ctx, wk.ID, "stopped", wk.ContainerID, "") 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) } jsonOut(w, 200, map[string]any{ "manual_credits_enabled": s.cfg.AllowManualCredits, "paypal_enabled": s.paypalAllowed(), "worker_image": s.cfg.WorkerImage, "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 } if err := s.pullWorkerImage(r.Context()); 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, "resumed": wk.Status == "running" || wk.Status == "starting"}) } func (s *Service) adminUpdateAllWorkerImages(w http.ResponseWriter, r *http.Request) { if err := s.pullWorkerImage(r.Context()); err != nil { jsonOut(w, http.StatusBadGateway, map[string]string{"error": err.Error()}) return } rows, err := s.store.DB.QueryContext(r.Context(), `SELECT `+workerCols+` FROM workers ORDER BY created_at`) if err != nil { jsonOut(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) return } var workers []Worker for rows.Next() { wk, scanErr := scanWorker(rows) if scanErr != nil { _ = rows.Close() jsonOut(w, http.StatusInternalServerError, map[string]string{"error": scanErr.Error()}) return } workers = append(workers, wk) } _ = rows.Close() updated := 0 var warnings []string for _, wk := range workers { 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), "warnings": warnings}) } 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