mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
786569eac0a9231db11c8d043a53e6c3ce233e13
pikasTech-HWLAB/internal/userbilling/service.go
T
lyon 786569eac0 fix: trace auth login with otel
2026-06-20 08:37:56 +08:00

2881 lines
106 KiB
Go
Raw Blame History

package userbilling
import (
"context"
"crypto/hmac"
"crypto/rand"
"crypto/sha256"
"crypto/subtle"
"database/sql"
"embed"
"encoding/base64"
"encoding/binary"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io/fs"
"log"
"math"
"net/http"
"os"
"os/signal"
"sort"
"strconv"
"strings"
"syscall"
"time"
_ "github.com/jackc/pgx/v4/stdlib"
)
const serviceID = "hwlab-user-billing"
//go:embed migrations/*.sql
var migrationFS embed.FS
type Config struct {
Addr string
DatabaseURL string
RedisURL string
InitialCredits int64
CreditPerThousandTok int64
SessionTTL time.Duration
ReservationTTL time.Duration
RegistrationEnabled bool
InternalToken string
RequireInternalToken bool
MigrateOnStart bool
StateAuthority string
RuntimeNamespace string
ExternalDatabaseLabel string
}
type Server struct {
config Config
db *sql.DB
mux *http.ServeMux
}
type User struct {
ID string `json:"id"`
Email string `json:"email"`
Username string `json:"username"`
DisplayName string `json:"displayName"`
Status string `json:"status"`
Role string `json:"role"`
EmailVerified bool `json:"emailVerified"`
CreatedAt time.Time `json:"createdAt"`
}
type Principal struct {
UserID string `json:"userId"`
Email string `json:"email"`
Username string `json:"username"`
Role string `json:"role"`
Scopes []string `json:"scopes"`
AuthType string `json:"authType"`
KeyID string `json:"keyId,omitempty"`
}
type serviceUsageSummary struct {
ServiceID string `json:"serviceId"`
ResourceType string `json:"resourceType"`
Credits int64 `json:"credits"`
Quantity int64 `json:"quantity"`
RecordCount int64 `json:"recordCount"`
LastUsedAt time.Time `json:"lastUsedAt"`
}
type ledgerSummaryRow struct {
ID string `json:"id"`
Kind string `json:"kind"`
Reason string `json:"reason"`
DeltaCredits int64 `json:"deltaCredits"`
BalanceBefore *int64 `json:"balanceBefore,omitempty"`
BalanceAfter int64 `json:"balanceAfter"`
Source string `json:"source"`
Status string `json:"status"`
IdempotencyKey string `json:"idempotencyKey,omitempty"`
OperatorUserID string `json:"operatorUserId,omitempty"`
OperatorUsername string `json:"operatorUsername,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
CreatedAt time.Time `json:"createdAt"`
}
type adminUsageRow struct {
ID string `json:"id"`
ReservationID string `json:"reservationId,omitempty"`
UserID string `json:"userId"`
Email string `json:"email"`
Username string `json:"username"`
APIKeyID string `json:"apiKeyId,omitempty"`
ServiceID string `json:"serviceId"`
ResourceType string `json:"resourceType"`
Unit string `json:"unit"`
Quantity int64 `json:"quantity"`
Credits int64 `json:"credits"`
Status string `json:"status"`
IdempotencyKey string `json:"idempotencyKey,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
CreatedAt time.Time `json:"createdAt"`
}
type adminLedgerRow struct {
ledgerSummaryRow
UserID string `json:"userId"`
Email string `json:"email"`
Username string `json:"username"`
}
type adminAuditQuery struct {
Page int
PageSize int
UserID string
Search string
ServiceID string
ResourceType string
APIKeyID string
Status string
Kind string
Source string
IdempotencyKey string
From string
To string
}
type creditAdjustResult struct {
UserID string `json:"userId"`
BalanceBefore int64 `json:"balanceBefore"`
BalanceAfter int64 `json:"balanceAfter"`
DeltaCredits int64 `json:"deltaCredits"`
LedgerID string `json:"ledgerId"`
Kind string `json:"kind"`
Reason string `json:"reason"`
Source string `json:"source"`
Status string `json:"status"`
IdempotencyKey string `json:"idempotencyKey,omitempty"`
Idempotent bool `json:"idempotent"`
}
type reservationSummaryRow struct {
ReservationID string `json:"reservationId"`
ServiceID string `json:"serviceId"`
ResourceType string `json:"resourceType"`
EstimatedCredits int64 `json:"estimatedCredits"`
EstimatedTokens int64 `json:"estimatedTokens"`
Status string `json:"status"`
ExpiresAt time.Time `json:"expiresAt"`
CreatedAt time.Time `json:"createdAt"`
}
type billingPlanSummary struct {
ID string `json:"id"`
DisplayName string `json:"displayName"`
Status string `json:"status"`
Description string `json:"description"`
Metadata map[string]any `json:"metadata,omitempty"`
}
type resourceEntitlementSummary struct {
ID string `json:"id,omitempty"`
PlanID string `json:"planId,omitempty"`
ResourceType string `json:"resourceType"`
Enabled bool `json:"enabled"`
MonthlyQuotaCredits int64 `json:"monthlyQuotaCredits"`
MonthlyQuotaUnlimited bool `json:"monthlyQuotaUnlimited"`
MonthlyUsageCredits int64 `json:"monthlyUsageCredits"`
MonthlyRemainingCredits int64 `json:"monthlyRemainingCredits"`
MaxConcurrentReservations int64 `json:"maxConcurrentReservations"`
ConcurrencyUnlimited bool `json:"concurrencyUnlimited"`
ActiveReservations int64 `json:"activeReservations"`
RPMLimit int64 `json:"rpmLimit"`
RPMUnlimited bool `json:"rpmUnlimited"`
RPMUsed int64 `json:"rpmUsed"`
LimitSemantics string `json:"limitSemantics"`
Metadata map[string]any `json:"metadata,omitempty"`
}
type adminBillingUserSummary struct {
User User `json:"user"`
PlanID string `json:"planId"`
Plan billingPlanSummary `json:"plan,omitempty"`
Entitlements []resourceEntitlementSummary `json:"entitlements,omitempty"`
Credits creditSummary `json:"credits"`
Usage userUsageAggregate `json:"usage"`
APIKeys apiKeyAggregate `json:"apiKeys"`
Reservations reservationAdminSummary `json:"reservations"`
}
type creditSummary struct {
Balance int64 `json:"balance"`
Reserved int64 `json:"reserved"`
Available int64 `json:"available"`
}
type userUsageAggregate struct {
TotalCredits int64 `json:"totalCredits"`
TotalQuantity int64 `json:"totalQuantity"`
RecordCount int64 `json:"recordCount"`
LastUsedAt *time.Time `json:"lastUsedAt"`
}
type apiKeyAggregate struct {
ActiveCount int64 `json:"activeCount"`
TotalCount int64 `json:"totalCount"`
LastUsedAt *time.Time `json:"lastUsedAt"`
}
type apiKeySummary struct {
ID string `json:"id"`
Name string `json:"name"`
Prefix string `json:"prefix"`
Scopes []string `json:"scopes"`
Status string `json:"status"`
CreatedAt time.Time `json:"createdAt"`
LastUsedAt *time.Time `json:"lastUsedAt"`
RevokedAt *time.Time `json:"revokedAt"`
}
type adminUsersQuery struct {
Page int
PageSize int
Search string
Status string
Role string
}
type adminUserDetail struct {
Summary adminBillingUserSummary `json:"summary"`
Ledger []ledgerSummaryRow `json:"ledger"`
APIKeys []apiKeySummary `json:"apiKeys"`
UsageBy []serviceUsageSummary `json:"usageByService"`
Reservations []reservationSummaryRow `json:"reservations"`
}
type billingPlanDetail struct {
Plan billingPlanSummary `json:"plan"`
Entitlements []resourceEntitlementSummary `json:"entitlements"`
}
type reservationAdminSummary struct {
ActiveCount int64 `json:"activeCount"`
Active []reservationSummaryRow `json:"active"`
}
type apiError struct {
Code string `json:"code"`
Message string `json:"message"`
}
func Run(parent context.Context) error {
cfg := LoadConfig(os.Environ())
server, err := NewServer(parent, cfg)
if err != nil {
return err
}
defer server.Close()
ctx, stop := signal.NotifyContext(parent, syscall.SIGINT, syscall.SIGTERM)
defer stop()
httpServer := &http.Server{
Addr: cfg.Addr,
Handler: server,
ReadHeaderTimeout: 10 * time.Second,
}
errCh := make(chan error, 1)
go func() {
log.Printf("%s listening on %s", serviceID, cfg.Addr)
errCh <- httpServer.ListenAndServe()
}()
select {
case <-ctx.Done():
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
_ = httpServer.Shutdown(shutdownCtx)
return nil
case err := <-errCh:
if errors.Is(err, http.ErrServerClosed) {
return nil
}
return err
}
}
func LoadConfig(env []string) Config {
values := map[string]string{}
for _, item := range env {
name, value, ok := strings.Cut(item, "=")
if ok {
values[name] = value
}
}
port := first(values["HWLAB_USER_BILLING_PORT"], values["PORT"], "6670")
cfg := Config{
Addr: first(values["HWLAB_USER_BILLING_ADDR"], ":"+port),
DatabaseURL: first(values["HWLAB_USER_BILLING_DB_URL"], values["DATABASE_URL"]),
RedisURL: values["HWLAB_USER_BILLING_REDIS_URL"],
InitialCredits: parseInt64(values["HWLAB_USER_BILLING_INITIAL_CREDITS"], 0),
CreditPerThousandTok: parseInt64(values["HWLAB_USER_BILLING_CREDIT_PER_1K_TOKENS"], 1),
SessionTTL: parseDurationHours(values["HWLAB_USER_BILLING_SESSION_TTL_HOURS"], 720),
ReservationTTL: parseDurationMinutes(values["HWLAB_USER_BILLING_RESERVATION_TTL_MINUTES"], 30),
RegistrationEnabled: values["HWLAB_USER_BILLING_REGISTRATION_ENABLED"] != "0",
InternalToken: values["HWLAB_USER_BILLING_INTERNAL_TOKEN"],
RequireInternalToken: values["HWLAB_USER_BILLING_REQUIRE_INTERNAL_TOKEN"] == "1",
MigrateOnStart: values["HWLAB_USER_BILLING_MIGRATE_ON_START"] != "0",
StateAuthority: first(values["HWLAB_USER_BILLING_STATE_AUTHORITY"], "pk01-postgres"),
RuntimeNamespace: first(values["POD_NAMESPACE"], values["HWLAB_NAMESPACE"], "hwlab-v03"),
ExternalDatabaseLabel: first(values["HWLAB_USER_BILLING_EXTERNAL_DB_LABEL"], "pk01-external-postgres"),
}
return cfg
}
func NewServer(ctx context.Context, cfg Config) (*Server, error) {
server := &Server{config: cfg, mux: http.NewServeMux()}
if cfg.DatabaseURL != "" {
db, err := sql.Open("pgx", cfg.DatabaseURL)
if err != nil {
return nil, err
}
db.SetMaxOpenConns(8)
db.SetMaxIdleConns(4)
db.SetConnMaxLifetime(30 * time.Minute)
server.db = db
if cfg.MigrateOnStart {
migrateCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
defer cancel()
if err := server.migrate(migrateCtx); err != nil {
return nil, fmt.Errorf("migrate user billing schema: %w", err)
}
}
}
server.routes()
return server, nil
}
func (s *Server) Close() {
if s.db != nil {
_ = s.db.Close()
}
}
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
s.mux.ServeHTTP(w, r)
}
func (s *Server) routes() {
s.route(http.MethodGet, "/health/live", s.handleLive)
s.route(http.MethodGet, "/health/ready", s.handleReady)
s.route(http.MethodPost, "/v1/auth/register", s.handleRegister)
s.route(http.MethodPost, "/v1/auth/login", s.handleLogin)
s.route(http.MethodPost, "/v1/auth/logout", s.handleLogout)
s.route(http.MethodPost, "/v1/auth/refresh", s.handleRefresh)
s.route(http.MethodGet, "/v1/me", s.handleMe)
s.route(http.MethodPatch, "/v1/me/profile", s.handleUpdateProfile)
s.route(http.MethodPost, "/v1/me/password", s.handleChangePassword)
s.route(http.MethodGet, "/v1/billing/summary", s.handleBillingSummary)
s.route(http.MethodPost, "/v1/redeem", s.handleRedeem)
s.route(http.MethodGet, "/v1/subscription/summary", s.handleSubscriptionSummary)
s.route(http.MethodGet, "/v1/payments/summary", s.handlePaymentSummary)
s.mux.HandleFunc("/v1/api-keys", s.handleAPIKeys)
s.mux.HandleFunc("/v1/api-keys/", s.handleAPIKeyByID)
s.route(http.MethodPost, "/internal/auth/introspect", s.handleIntrospect)
s.route(http.MethodPost, "/internal/billing/preflight", s.handleBillingPreflight)
s.route(http.MethodPost, "/internal/billing/record", s.handleBillingRecord)
s.route(http.MethodPost, "/internal/billing/release", s.handleBillingRelease)
s.route(http.MethodGet, "/internal/admin/billing/summary", s.handleAdminBillingSummary)
s.route(http.MethodGet, "/internal/admin/billing/plans", s.handleAdminBillingPlans)
s.route(http.MethodGet, "/internal/admin/usage", s.handleAdminUsage)
s.route(http.MethodGet, "/internal/admin/usage/export", s.handleAdminUsageExport)
s.route(http.MethodGet, "/internal/admin/ledger", s.handleAdminLedger)
s.route(http.MethodGet, "/internal/admin/ledger/export", s.handleAdminLedgerExport)
s.mux.HandleFunc("/internal/admin/redeem-codes", s.handleAdminRedeemCodes)
s.mux.HandleFunc("/internal/admin/redeem-codes/", s.handleAdminRedeemCodeByID)
s.route(http.MethodGet, "/internal/admin/redeem-redemptions", s.handleAdminRedeemRedemptions)
s.route(http.MethodPost, "/internal/admin/credits/adjust", s.handleAdminCreditAdjust)
s.mux.HandleFunc("/internal/admin/users", s.handleAdminUsers)
s.mux.HandleFunc("/internal/admin/users/", s.handleAdminUserByID)
s.route(http.MethodPost, "/internal/admin/users/status", s.handleAdminUserStatus)
}
func (s *Server) route(method, path string, handler http.HandlerFunc) {
s.mux.HandleFunc(path, func(w http.ResponseWriter, r *http.Request) {
if r.Method != method {
w.Header().Set("Allow", method)
writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
return
}
handler(w, r)
})
}
func (s *Server) migrate(ctx context.Context) error {
if s.db == nil {
return nil
}
files, err := fs.Glob(migrationFS, "migrations/*.sql")
if err != nil {
return err
}
sort.Strings(files)
for _, name := range files {
body, err := migrationFS.ReadFile(name)
if err != nil {
return err
}
for _, statement := range splitSQLStatements(string(body)) {
if _, err := s.db.ExecContext(ctx, statement); err != nil {
return fmt.Errorf("%s: %w", name, err)
}
}
}
return nil
}
func splitSQLStatements(text string) []string {
parts := strings.Split(text, ";\n")
statements := make([]string, 0, len(parts))
for _, part := range parts {
stmt := strings.TrimSpace(part)
if stmt == "" {
continue
}
if !strings.HasSuffix(stmt, ";") {
stmt += ";"
}
statements = append(statements, stmt)
}
return statements
}
func (s *Server) handleLive(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]any{
"serviceId": serviceID,
"status": "ok",
"stateless": true,
"stateAuthority": s.config.StateAuthority,
"database": map[string]any{"configured": s.db != nil, "authority": s.config.ExternalDatabaseLabel},
"redis": map[string]any{"configured": s.config.RedisURL != "", "role": "cache-only"},
"namespace": s.config.RuntimeNamespace,
"registrationEnabled": s.config.RegistrationEnabled,
"internalTokenRequired": s.config.RequireInternalToken,
})
}
func (s *Server) handleReady(w http.ResponseWriter, r *http.Request) {
if s.db == nil {
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"serviceId": serviceID, "status": "degraded", "reason": "database-url-not-configured"})
return
}
ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
defer cancel()
if err := s.db.PingContext(ctx); err != nil {
s.logDatabaseError("ready_ping", err)
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"serviceId": serviceID, "status": "degraded", "reason": "database-ping-failed", "database": map[string]any{"configured": true, "authority": s.config.ExternalDatabaseLabel, "errorClass": classifyDatabaseError(err)}})
return
}
if err := s.readyWriteProbe(ctx); err != nil {
s.logDatabaseError("ready_write_probe", err)
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"serviceId": serviceID, "status": "degraded", "reason": "database-transaction-probe-failed", "database": map[string]any{"configured": true, "authority": s.config.ExternalDatabaseLabel, "errorClass": classifyDatabaseError(err)}})
return
}
writeJSON(w, http.StatusOK, map[string]any{"serviceId": serviceID, "status": "ready", "database": map[string]any{"configured": true, "authority": s.config.ExternalDatabaseLabel, "transactionProbe": "write-rollback-ok"}})
}
func (s *Server) handleRegister(w http.ResponseWriter, r *http.Request) {
if !s.registrationAvailable(w) {
return
}
var req struct {
Email string `json:"email"`
Username string `json:"username"`
DisplayName string `json:"displayName"`
Password string `json:"password"`
}
if !decodeJSON(w, r, &req) {
return
}
email := strings.ToLower(strings.TrimSpace(req.Email))
username := normalizeUsername(req.Username)
if email == "" || !strings.Contains(email, "@") || username == "" || len(req.Password) < 10 {
writeAPIError(w, http.StatusBadRequest, "invalid_registration", "email, username and a password with at least 10 characters are required")
return
}
passwordHash, err := hashPassword(req.Password)
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "password_hash_failed", "could not hash password")
return
}
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
if err != nil {
s.logDatabaseError("register_begin_tx", err)
writeAPIError(w, http.StatusInternalServerError, "transaction_failed", "could not start transaction")
return
}
defer tx.Rollback()
user := User{ID: newID("usr"), Email: email, Username: username, DisplayName: strings.TrimSpace(req.DisplayName), Status: "active", Role: "user"}
err = tx.QueryRowContext(ctx, `INSERT INTO hwlab_users (id, email, username, display_name, password_hash, status, role) VALUES ($1, $2, $3, $4, $5, 'active', 'user') RETURNING created_at`, user.ID, user.Email, user.Username, user.DisplayName, passwordHash).Scan(&user.CreatedAt)
if err != nil {
if isDuplicate(err) {
writeAPIError(w, http.StatusConflict, "user_exists", "email or username already exists")
return
}
writeAPIError(w, http.StatusInternalServerError, "user_create_failed", "could not create user")
return
}
_, err = tx.ExecContext(ctx, `INSERT INTO hwlab_credit_accounts (user_id, balance_credits, reserved_credits, plan_id) VALUES ($1, $2, 0, 'default')`, user.ID, s.config.InitialCredits)
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "credit_account_failed", "could not create credit account")
return
}
if s.config.InitialCredits > 0 {
_, err = tx.ExecContext(ctx, `INSERT INTO hwlab_credit_ledger (id, user_id, delta_credits, balance_before, balance_after, kind, reason, source, status, metadata) VALUES ($1, $2, $3, 0, $3, 'grant', 'registration_initial_credit', 'registration', 'applied', '{}'::jsonb)`, newID("led"), user.ID, s.config.InitialCredits)
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "credit_ledger_failed", "could not write initial credit ledger")
return
}
}
if err := tx.Commit(); err != nil {
writeAPIError(w, http.StatusInternalServerError, "transaction_commit_failed", "could not commit registration")
return
}
writeJSON(w, http.StatusCreated, map[string]any{"user": user, "initialCredits": s.config.InitialCredits})
}
func (s *Server) handleLogin(w http.ResponseWriter, r *http.Request) {
trace := newOtelTraceContext(r.Header.Get("traceparent"))
if traceparent := trace.traceparent(); traceparent != "" {
w.Header().Set("traceparent", traceparent)
w.Header().Set("x-hwlab-otel-trace-id", trace.TraceID)
}
startedAt := time.Now().UTC()
status := http.StatusInternalServerError
outcome := "error"
errorCode := ""
defer func() {
attrs := map[string]any{
"http.method": http.MethodPost,
"http.route": "/v1/auth/login",
"http.status_code": status,
"auth.outcome": outcome,
"user_billing.service_id": serviceID,
"user_billing.values_redacted": true,
}
s.emitOtelSpanAsync("user-billing.auth.login", trace, trace.ServerSpanID, trace.ParentSpanID, otelSpanKindServer, startedAt, attrs, status, errorCode)
}()
if !s.databaseAvailable(w) {
status = http.StatusServiceUnavailable
outcome = "state_backend_not_configured"
errorCode = "state_backend_not_configured"
return
}
var req struct {
Email string `json:"email"`
Username string `json:"username"`
Password string `json:"password"`
}
if !decodeJSON(w, r, &req) {
status = http.StatusBadRequest
outcome = "invalid_json"
errorCode = "invalid_json"
return
}
login := strings.ToLower(strings.TrimSpace(first(req.Email, req.Username)))
if login == "" || req.Password == "" {
status = http.StatusBadRequest
outcome = "invalid_login"
errorCode = "invalid_login"
writeAPIError(w, http.StatusBadRequest, "invalid_login", "email or username and password are required")
return
}
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
var user User
var passwordHash string
dbStartedAt := time.Now().UTC()
err := s.db.QueryRowContext(ctx, `SELECT id, email, username, display_name, password_hash, status, role, email_verified, created_at FROM hwlab_users WHERE lower(email) = $1 OR lower(username) = $1`, login).Scan(&user.ID, &user.Email, &user.Username, &user.DisplayName, &passwordHash, &user.Status, &user.Role, &user.EmailVerified, &user.CreatedAt)
dbErrorCode := ""
if err != nil && !errors.Is(err, sql.ErrNoRows) {
dbErrorCode = first(sqlErrorCode(err), "db_query_failed")
}
s.emitOtelSpanAsync("user-billing.postgres.login_user", trace, newOtelSpanID(), trace.ServerSpanID, otelSpanKindInternal, dbStartedAt, map[string]any{
"db.system": "postgresql",
"db.operation": "SELECT",
"db.sql.table": "hwlab_users",
"db.rows_found": err == nil,
"user_billing.values_redacted": true,
"db.error_code": dbErrorCode,
}, 0, dbErrorCode)
if errors.Is(err, sql.ErrNoRows) {
status = http.StatusUnauthorized
outcome = "invalid_credentials"
errorCode = "invalid_credentials"
writeAPIError(w, http.StatusUnauthorized, "invalid_credentials", "invalid email, username or password")
return
}
if err != nil {
status = http.StatusInternalServerError
outcome = "db_error"
errorCode = first(dbErrorCode, "login_failed")
writeAPIError(w, http.StatusInternalServerError, errorCode, "could not load user")
return
}
if !verifyPassword(passwordHash, req.Password) {
status = http.StatusUnauthorized
outcome = "invalid_credentials"
errorCode = "invalid_credentials"
writeAPIError(w, http.StatusUnauthorized, "invalid_credentials", "invalid email, username or password")
return
}
if user.Status != "active" {
status = http.StatusForbidden
outcome = "user_disabled"
errorCode = "user_disabled"
writeAPIError(w, http.StatusForbidden, "user_disabled", "user is not active")
return
}
sessionToken := "hws_" + randomToken(32)
expiresAt := time.Now().UTC().Add(s.config.SessionTTL)
_, err = s.db.ExecContext(ctx, `INSERT INTO hwlab_user_sessions (id, user_id, token_hash, expires_at) VALUES ($1, $2, $3, $4)`, newID("ses"), user.ID, tokenHash(sessionToken), expiresAt)
if err != nil {
status = http.StatusInternalServerError
outcome = "session_create_failed"
errorCode = first(sqlErrorCode(err), "session_create_failed")
writeAPIError(w, http.StatusInternalServerError, errorCode, "could not create session")
return
}
status = http.StatusOK
outcome = "session_created"
writeJSON(w, http.StatusOK, map[string]any{"token": sessionToken, "tokenType": "Bearer", "expiresAt": expiresAt, "user": user})
}
func (s *Server) handleLogout(w http.ResponseWriter, r *http.Request) {
if !s.databaseAvailable(w) {
return
}
token := bearerToken(r)
if token == "" {
writeAPIError(w, http.StatusUnauthorized, "missing_bearer", "authorization bearer token is required")
return
}
ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
defer cancel()
_, _ = s.db.ExecContext(ctx, `UPDATE hwlab_user_sessions SET revoked_at = now() WHERE token_hash = $1 AND revoked_at IS NULL`, tokenHash(token))
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
func (s *Server) handleRefresh(w http.ResponseWriter, r *http.Request) {
principal, ok := s.requirePrincipal(w, r)
if !ok {
return
}
if principal.AuthType != "session" {
writeAPIError(w, http.StatusForbidden, "session_required", "only session tokens can be refreshed")
return
}
expiresAt := time.Now().UTC().Add(s.config.SessionTTL)
ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
defer cancel()
_, err := s.db.ExecContext(ctx, `UPDATE hwlab_user_sessions SET expires_at = $1 WHERE token_hash = $2 AND revoked_at IS NULL`, expiresAt, tokenHash(bearerToken(r)))
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "session_refresh_failed", "could not refresh session")
return
}
writeJSON(w, http.StatusOK, map[string]any{"expiresAt": expiresAt})
}
func (s *Server) handleMe(w http.ResponseWriter, r *http.Request) {
principal, ok := s.requirePrincipal(w, r)
if !ok {
return
}
balance, reserved, _ := s.accountBalance(r.Context(), principal.UserID)
writeJSON(w, http.StatusOK, map[string]any{"principal": principal, "credits": map[string]any{"balance": balance, "reserved": reserved, "available": balance - reserved}})
}
func (s *Server) handleUpdateProfile(w http.ResponseWriter, r *http.Request) {
principal, ok := s.requirePrincipal(w, r)
if !ok {
return
}
var req struct {
Username *string `json:"username"`
DisplayName *string `json:"displayName"`
}
if !decodeJSON(w, r, &req) {
return
}
updates := []string{}
args := []any{principal.UserID}
if req.Username != nil {
username := normalizeUsername(*req.Username)
if strings.TrimSpace(*req.Username) != "" && username == "" {
writeAPIError(w, http.StatusBadRequest, "invalid_username", "username must be 3-64 characters and contain only lowercase letters, digits, '-' or '_'")
return
}
if username != "" && username != principal.Username {
args = append(args, username)
updates = append(updates, "username = $"+strconv.Itoa(len(args)))
}
}
if req.DisplayName != nil {
args = append(args, strings.TrimSpace(*req.DisplayName))
updates = append(updates, "display_name = $"+strconv.Itoa(len(args)))
}
if len(updates) == 0 {
writeAPIError(w, http.StatusBadRequest, "no_profile_fields", "at least one profile field must be provided")
return
}
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
var user User
query := `UPDATE hwlab_users SET ` + strings.Join(updates, ", ") + `, updated_at = now() WHERE id = $1 RETURNING id, email, username, display_name, status, role, email_verified, created_at`
err := s.db.QueryRowContext(ctx, query, args...).Scan(&user.ID, &user.Email, &user.Username, &user.DisplayName, &user.Status, &user.Role, &user.EmailVerified, &user.CreatedAt)
if err != nil {
if isDuplicate(err) {
writeAPIError(w, http.StatusConflict, "username_exists", "username already exists")
return
}
writeAPIError(w, http.StatusInternalServerError, "profile_update_failed", "could not update profile")
return
}
writeJSON(w, http.StatusOK, map[string]any{"contractVersion": "user-profile-v1", "updated": true, "user": user, "valuesRedacted": true})
}
func (s *Server) handleChangePassword(w http.ResponseWriter, r *http.Request) {
principal, ok := s.requirePrincipal(w, r)
if !ok {
return
}
if principal.AuthType != "session" {
writeAPIError(w, http.StatusForbidden, "session_required", "only session-authenticated users can change password")
return
}
var req struct {
CurrentPassword string `json:"currentPassword"`
Password string `json:"password"`
NewPassword string `json:"newPassword"`
}
if !decodeJSON(w, r, &req) {
return
}
newPassword := first(req.NewPassword, req.Password)
if req.CurrentPassword == "" || len(newPassword) < 10 {
writeAPIError(w, http.StatusBadRequest, "invalid_password_change", "currentPassword and a new password with at least 10 characters are required")
return
}
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
var currentHash string
if err := s.db.QueryRowContext(ctx, `SELECT password_hash FROM hwlab_users WHERE id = $1 AND status = 'active'`, principal.UserID).Scan(&currentHash); err != nil {
writeAPIError(w, http.StatusUnauthorized, "invalid_subject", "active user was not found")
return
}
if !verifyPassword(currentHash, req.CurrentPassword) {
writeAPIError(w, http.StatusUnauthorized, "invalid_current_password", "current password is invalid")
return
}
passwordHash, err := hashPassword(newPassword)
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "password_hash_failed", "could not hash password")
return
}
if _, err := s.db.ExecContext(ctx, `UPDATE hwlab_users SET password_hash = $2, updated_at = now() WHERE id = $1`, principal.UserID, passwordHash); err != nil {
writeAPIError(w, http.StatusInternalServerError, "password_update_failed", "could not update password")
return
}
writeJSON(w, http.StatusOK, map[string]any{"contractVersion": "user-profile-v1", "changed": true, "valuesRedacted": true})
}
func (s *Server) handleBillingSummary(w http.ResponseWriter, r *http.Request) {
principal, ok := s.requirePrincipal(w, r)
if !ok {
return
}
limit := queryLimit(r, "limit", 20, 100)
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
balance, reserved, err := s.accountBalance(ctx, principal.UserID)
if err != nil {
s.logDatabaseError("billing_summary_account", err)
writeAPIError(w, http.StatusInternalServerError, "billing_summary_failed", "could not load credit account")
return
}
totalCredits, totalQuantity, recordCount, err := s.usageTotals(ctx, principal.UserID)
if err != nil {
s.logDatabaseError("billing_summary_usage_totals", err)
writeAPIError(w, http.StatusInternalServerError, "billing_summary_failed", "could not load usage totals")
return
}
byService, err := s.usageByService(ctx, principal.UserID, 20)
if err != nil {
s.logDatabaseError("billing_summary_usage_by_service", err)
writeAPIError(w, http.StatusInternalServerError, "billing_summary_failed", "could not load service usage")
return
}
ledgerRows, err := s.ledgerRows(ctx, principal.UserID, limit)
if err != nil {
s.logDatabaseError("billing_summary_ledger", err)
writeAPIError(w, http.StatusInternalServerError, "billing_summary_failed", "could not load credit ledger")
return
}
reservations, err := s.activeReservations(ctx, principal.UserID, 20)
if err != nil {
s.logDatabaseError("billing_summary_reservations", err)
writeAPIError(w, http.StatusInternalServerError, "billing_summary_failed", "could not load active reservations")
return
}
planID, plan, entitlements, err := s.accountPlanContext(ctx, principal.UserID)
if err != nil {
s.logDatabaseError("billing_summary_entitlements", err)
writeAPIError(w, http.StatusInternalServerError, "billing_summary_failed", "could not load billing plan entitlements")
return
}
writeJSON(w, http.StatusOK, map[string]any{
"contractVersion": "user-billing-summary-v1",
"serviceId": serviceID,
"status": "ok",
"principal": principal,
"credits": map[string]any{
"balance": balance,
"reserved": reserved,
"available": balance - reserved,
},
"planId": planID,
"plan": plan,
"entitlements": entitlements,
"usage": map[string]any{
"totalCredits": totalCredits,
"totalQuantity": totalQuantity,
"recordCount": recordCount,
"byService": byService,
},
"ledger": map[string]any{
"count": len(ledgerRows),
"limit": limit,
"rows": ledgerRows,
},
"reservations": map[string]any{
"activeCount": len(reservations),
"active": reservations,
},
"state": map[string]any{
"stateless": true,
"stateAuthority": s.config.StateAuthority,
"database": map[string]any{"authority": s.config.ExternalDatabaseLabel},
"redis": map[string]any{"role": "cache-only"},
},
"valuesRedacted": true,
})
}
func (s *Server) handleAPIKeys(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet:
s.handleListAPIKeys(w, r)
case http.MethodPost:
s.handleCreateAPIKey(w, r)
default:
w.Header().Set("Allow", "GET, POST")
writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
}
}
func (s *Server) handleCreateAPIKey(w http.ResponseWriter, r *http.Request) {
principal, ok := s.requirePrincipal(w, r)
if !ok {
return
}
var req struct {
Name string `json:"name"`
Scopes []string `json:"scopes"`
}
if !decodeJSON(w, r, &req) {
return
}
name := strings.TrimSpace(first(req.Name, "default"))
scopes := req.Scopes
if len(scopes) == 0 {
scopes = []string{"api"}
}
scopesJSON, _ := json.Marshal(scopes)
key := "hwl_" + randomToken(32)
keyID := newID("key")
prefix := key[:min(len(key), 16)]
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
_, err := s.db.ExecContext(ctx, `INSERT INTO hwlab_api_keys (id, user_id, name, key_prefix, key_hash, scopes_json) VALUES ($1, $2, $3, $4, $5, $6::jsonb)`, keyID, principal.UserID, name, prefix, tokenHash(key), string(scopesJSON))
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "api_key_create_failed", "could not create API key")
return
}
summary := apiKeySummary{ID: keyID, Name: name, Prefix: prefix, Scopes: scopes, Status: "active", CreatedAt: time.Now().UTC()}
writeJSON(w, http.StatusCreated, map[string]any{"contractVersion": "user-api-key-v1", "id": keyID, "key": key, "secret": key, "apiKey": summary, "prefix": prefix, "scopes": scopes, "created": true, "valuesRedacted": true})
}
func (s *Server) handleListAPIKeys(w http.ResponseWriter, r *http.Request) {
principal, ok := s.requirePrincipal(w, r)
if !ok {
return
}
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
keys, err := s.apiKeysForUser(ctx, principal.UserID)
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "api_key_list_failed", "could not list API keys")
return
}
writeJSON(w, http.StatusOK, map[string]any{"contractVersion": "user-api-key-v1", "keys": keys, "count": len(keys), "valuesRedacted": true})
}
func (s *Server) handleAPIKeyByID(w http.ResponseWriter, r *http.Request) {
principal, ok := s.requirePrincipal(w, r)
if !ok {
return
}
keyID := strings.TrimPrefix(r.URL.Path, "/v1/api-keys/")
keyID = strings.TrimSpace(keyID)
if keyID == "" || strings.Contains(keyID, "/") {
writeAPIError(w, http.StatusNotFound, "api_key_not_found", "API key was not found")
return
}
switch r.Method {
case http.MethodPatch:
s.handleUpdateAPIKey(w, r, principal.UserID, keyID)
case http.MethodDelete:
s.handleRevokeAPIKey(w, r, principal.UserID, keyID)
default:
w.Header().Set("Allow", "PATCH, DELETE")
writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
}
}
func (s *Server) handleUpdateAPIKey(w http.ResponseWriter, r *http.Request, userID, keyID string) {
var req struct {
Name string `json:"name"`
Scopes []string `json:"scopes"`
Status string `json:"status"`
}
if !decodeJSON(w, r, &req) {
return
}
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
updates := []string{}
args := []any{keyID, userID}
if strings.TrimSpace(req.Name) != "" {
args = append(args, strings.TrimSpace(req.Name))
updates = append(updates, "name = $"+strconv.Itoa(len(args)))
}
if len(req.Scopes) > 0 {
scopesJSON, _ := json.Marshal(req.Scopes)
args = append(args, string(scopesJSON))
updates = append(updates, "scopes_json = $"+strconv.Itoa(len(args))+"::jsonb")
}
status := strings.TrimSpace(strings.ToLower(req.Status))
if status != "" {
if status != "revoked" {
writeAPIError(w, http.StatusBadRequest, "invalid_status", "only revoked status can be set through user API key update")
return
}
updates = append(updates, "status = 'revoked'", "revoked_at = now()")
}
if len(updates) == 0 {
key, err := s.apiKeyForUser(ctx, userID, keyID)
if err != nil {
writeAPIError(w, http.StatusNotFound, "api_key_not_found", "API key was not found")
return
}
writeJSON(w, http.StatusOK, map[string]any{"contractVersion": "user-api-key-v1", "key": key, "updated": false, "valuesRedacted": true})
return
}
query := `UPDATE hwlab_api_keys SET ` + strings.Join(updates, ", ") + ` WHERE id = $1 AND user_id = $2 RETURNING id, name, key_prefix, scopes_json::text, status, created_at, last_used_at, revoked_at`
key, err := scanAPIKey(s.db.QueryRowContext(ctx, query, args...))
if errors.Is(err, sql.ErrNoRows) {
writeAPIError(w, http.StatusNotFound, "api_key_not_found", "API key was not found")
return
}
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "api_key_update_failed", "could not update API key")
return
}
writeJSON(w, http.StatusOK, map[string]any{"contractVersion": "user-api-key-v1", "key": key, "updated": true, "valuesRedacted": true})
}
func (s *Server) handleRevokeAPIKey(w http.ResponseWriter, r *http.Request, userID, keyID string) {
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
key, err := scanAPIKey(s.db.QueryRowContext(ctx, `UPDATE hwlab_api_keys SET status = 'revoked', revoked_at = COALESCE(revoked_at, now()) WHERE id = $1 AND user_id = $2 RETURNING id, name, key_prefix, scopes_json::text, status, created_at, last_used_at, revoked_at`, keyID, userID))
if errors.Is(err, sql.ErrNoRows) {
writeAPIError(w, http.StatusNotFound, "api_key_not_found", "API key was not found")
return
}
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "api_key_revoke_failed", "could not revoke API key")
return
}
writeJSON(w, http.StatusOK, map[string]any{"contractVersion": "user-api-key-v1", "key": key, "revoked": true, "valuesRedacted": true})
}
func (s *Server) handleIntrospect(w http.ResponseWriter, r *http.Request) {
if !s.internalAllowed(w, r) || !s.databaseAvailable(w) {
return
}
var req struct {
Token string `json:"token"`
APIKey string `json:"apiKey"`
}
if !decodeJSON(w, r, &req) {
return
}
token := first(req.Token, req.APIKey)
principal, err := s.principalFromToken(r.Context(), token)
if err != nil {
writeJSON(w, http.StatusOK, map[string]any{"active": false})
return
}
writeJSON(w, http.StatusOK, map[string]any{"active": true, "principal": principal})
}
func (s *Server) handleBillingPreflight(w http.ResponseWriter, r *http.Request) {
if !s.internalAllowed(w, r) || !s.databaseAvailable(w) {
return
}
var req struct {
Token string `json:"token"`
APIKey string `json:"apiKey"`
UserID string `json:"userId"`
ServiceID string `json:"serviceId"`
EstimatedCredits int64 `json:"estimatedCredits"`
EstimatedTokens int64 `json:"estimatedTokens"`
IdempotencyKey string `json:"idempotencyKey"`
Metadata map[string]any `json:"metadata"`
}
if !decodeJSON(w, r, &req) {
return
}
principal, err := s.principalForBilling(r.Context(), r, req.Token, req.APIKey, req.UserID)
if err != nil {
writeAPIError(w, http.StatusUnauthorized, "invalid_subject", "billing subject is not active")
return
}
service := strings.TrimSpace(req.ServiceID)
if service == "" {
writeAPIError(w, http.StatusBadRequest, "missing_service", "serviceId is required")
return
}
estimated := s.estimateCredits(req.EstimatedCredits, req.EstimatedTokens)
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
reservation, err := s.reserveCredits(ctx, principal.UserID, principal.KeyID, service, estimated, req.EstimatedTokens, strings.TrimSpace(req.IdempotencyKey), req.Metadata)
if err != nil {
if errors.Is(err, errInsufficientCredits) {
writeJSON(w, http.StatusPaymentRequired, map[string]any{"allowed": false, "reason": "insufficient_credits"})
return
}
var limitErr billingLimitError
if errors.As(err, &limitErr) {
writeJSON(w, limitErr.Status, map[string]any{"allowed": false, "reason": limitErr.Code, "error": apiError{Code: limitErr.Code, Message: limitErr.Message}})
return
}
writeAPIError(w, http.StatusInternalServerError, "billing_preflight_failed", "could not reserve credits")
return
}
writeJSON(w, http.StatusOK, reservation)
}
func (s *Server) handleBillingRecord(w http.ResponseWriter, r *http.Request) {
if !s.internalAllowed(w, r) || !s.databaseAvailable(w) {
return
}
var req struct {
ReservationID string `json:"reservationId"`
Token string `json:"token"`
APIKey string `json:"apiKey"`
UserID string `json:"userId"`
ServiceID string `json:"serviceId"`
UsedCredits int64 `json:"usedCredits"`
UsedTokens int64 `json:"usedTokens"`
IdempotencyKey string `json:"idempotencyKey"`
Metadata map[string]any `json:"metadata"`
}
if !decodeJSON(w, r, &req) {
return
}
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
result, err := s.recordUsage(ctx, r, req.ReservationID, req.Token, req.APIKey, req.UserID, req.ServiceID, req.UsedCredits, req.UsedTokens, req.IdempotencyKey, req.Metadata)
if err != nil {
writeAPIError(w, http.StatusBadRequest, "billing_record_failed", err.Error())
return
}
writeJSON(w, http.StatusOK, result)
}
func (s *Server) handleBillingRelease(w http.ResponseWriter, r *http.Request) {
if !s.internalAllowed(w, r) || !s.databaseAvailable(w) {
return
}
var req struct {
ReservationID string `json:"reservationId"`
ServiceID string `json:"serviceId"`
Reason string `json:"reason"`
IdempotencyKey string `json:"idempotencyKey"`
Metadata map[string]any `json:"metadata"`
}
if !decodeJSON(w, r, &req) {
return
}
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
result, err := s.releaseReservation(ctx, req.ReservationID, req.ServiceID, req.Reason, req.IdempotencyKey, req.Metadata)
if err != nil {
writeAPIError(w, http.StatusBadRequest, "billing_release_failed", err.Error())
return
}
writeJSON(w, http.StatusOK, result)
}
func (s *Server) handleAdminUsers(w http.ResponseWriter, r *http.Request) {
if !s.internalAllowed(w, r) || !s.databaseAvailable(w) {
return
}
switch r.Method {
case http.MethodGet:
s.handleAdminUserList(w, r)
case http.MethodPost:
s.handleAdminCreateUser(w, r)
default:
w.Header().Set("Allow", "GET, POST")
writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
}
}
func (s *Server) handleAdminUserByID(w http.ResponseWriter, r *http.Request) {
if !s.internalAllowed(w, r) || !s.databaseAvailable(w) {
return
}
userID := strings.TrimSpace(strings.TrimPrefix(r.URL.Path, "/internal/admin/users/"))
if userID == "" || strings.Contains(userID, "/") {
writeAPIError(w, http.StatusNotFound, "user_not_found", "user not found")
return
}
switch r.Method {
case http.MethodGet:
s.handleAdminUserDetail(w, r, userID)
case http.MethodPatch:
s.handleAdminUpdateUser(w, r, userID)
default:
w.Header().Set("Allow", "GET, PATCH")
writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
}
}
func (s *Server) handleAdminUserList(w http.ResponseWriter, r *http.Request) {
query := adminUsersQueryFromRequest(r)
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
users, total, err := s.adminUsers(ctx, query)
if err != nil {
s.logDatabaseError("admin_users_list", err)
writeAPIError(w, http.StatusInternalServerError, "admin_users_failed", "could not load users")
return
}
writeJSON(w, http.StatusOK, map[string]any{
"contractVersion": "user-billing-admin-users-v1",
"users": users,
"count": len(users),
"total": total,
"pagination": map[string]any{
"page": query.Page,
"pageSize": query.PageSize,
"offset": (query.Page - 1) * query.PageSize,
},
"filters": map[string]any{"search": query.Search, "status": query.Status, "role": query.Role},
"state": map[string]any{
"stateless": true,
"stateAuthority": s.config.StateAuthority,
"database": map[string]any{"authority": s.config.ExternalDatabaseLabel},
"redis": map[string]any{"role": "cache-only"},
},
"valuesRedacted": true,
})
}
func (s *Server) handleAdminUserDetail(w http.ResponseWriter, r *http.Request, userID string) {
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
detail, err := s.adminUserDetail(ctx, userID)
if errors.Is(err, sql.ErrNoRows) {
writeAPIError(w, http.StatusNotFound, "user_not_found", "user not found")
return
}
if err != nil {
s.logDatabaseError("admin_user_detail", err)
writeAPIError(w, http.StatusInternalServerError, "admin_user_detail_failed", "could not load user detail")
return
}
writeJSON(w, http.StatusOK, map[string]any{"contractVersion": "user-billing-admin-user-v1", "detail": detail, "valuesRedacted": true})
}
func (s *Server) handleAdminCreateUser(w http.ResponseWriter, r *http.Request) {
var req struct {
Email string `json:"email"`
Username string `json:"username"`
DisplayName string `json:"displayName"`
Password string `json:"password"`
Role string `json:"role"`
Status string `json:"status"`
InitialCredits int64 `json:"initialCredits"`
}
if !decodeJSON(w, r, &req) {
return
}
email := strings.ToLower(strings.TrimSpace(req.Email))
username := normalizeUsername(req.Username)
role := normalizeRole(req.Role, "user")
status := normalizeUserStatus(req.Status, "active")
if email == "" || !strings.Contains(email, "@") || username == "" || len(req.Password) < 10 || role == "" || status == "" {
writeAPIError(w, http.StatusBadRequest, "invalid_admin_user", "email, username, role, status and a password with at least 10 characters are required")
return
}
passwordHash, err := hashPassword(req.Password)
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "password_hash_failed", "could not hash password")
return
}
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "transaction_failed", "could not start transaction")
return
}
defer tx.Rollback()
user := User{ID: newID("usr"), Email: email, Username: username, DisplayName: strings.TrimSpace(req.DisplayName), Status: status, Role: role}
err = tx.QueryRowContext(ctx, `INSERT INTO hwlab_users (id, email, username, display_name, password_hash, status, role) VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING email_verified, created_at`, user.ID, user.Email, user.Username, user.DisplayName, passwordHash, status, role).Scan(&user.EmailVerified, &user.CreatedAt)
if err != nil {
if isDuplicate(err) {
writeAPIError(w, http.StatusConflict, "user_exists", "email or username already exists")
return
}
writeAPIError(w, http.StatusInternalServerError, "user_create_failed", "could not create user")
return
}
_, err = tx.ExecContext(ctx, `INSERT INTO hwlab_credit_accounts (user_id, balance_credits, reserved_credits, plan_id) VALUES ($1, $2, 0, 'default')`, user.ID, nonNegative(req.InitialCredits))
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "credit_account_failed", "could not create credit account")
return
}
if req.InitialCredits > 0 {
_, err = tx.ExecContext(ctx, `INSERT INTO hwlab_credit_ledger (id, user_id, delta_credits, balance_before, balance_after, kind, reason, source, status, metadata) VALUES ($1, $2, $3, 0, $3, 'admin_grant', 'admin_create_user_initial_credit', 'admin', 'applied', '{}'::jsonb)`, newID("led"), user.ID, req.InitialCredits)
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "credit_ledger_failed", "could not write initial credit ledger")
return
}
}
if err := tx.Commit(); err != nil {
writeAPIError(w, http.StatusInternalServerError, "transaction_commit_failed", "could not commit user creation")
return
}
writeJSON(w, http.StatusCreated, map[string]any{"contractVersion": "user-billing-admin-user-v1", "created": true, "user": user, "valuesRedacted": true})
}
func (s *Server) handleAdminUpdateUser(w http.ResponseWriter, r *http.Request, userID string) {
var req struct {
Email *string `json:"email"`
Username *string `json:"username"`
DisplayName *string `json:"displayName"`
Password *string `json:"password"`
Role *string `json:"role"`
Status *string `json:"status"`
PlanID *string `json:"planId"`
}
if !decodeJSON(w, r, &req) {
return
}
updates := []string{}
args := []any{userID}
if req.Email != nil {
email := strings.ToLower(strings.TrimSpace(*req.Email))
if email == "" || !strings.Contains(email, "@") {
writeAPIError(w, http.StatusBadRequest, "invalid_email", "email is invalid")
return
}
args = append(args, email)
updates = append(updates, "email = $"+strconv.Itoa(len(args)))
}
if req.Username != nil {
username := normalizeUsername(*req.Username)
if username == "" {
writeAPIError(w, http.StatusBadRequest, "invalid_username", "username is invalid")
return
}
args = append(args, username)
updates = append(updates, "username = $"+strconv.Itoa(len(args)))
}
if req.DisplayName != nil {
args = append(args, strings.TrimSpace(*req.DisplayName))
updates = append(updates, "display_name = $"+strconv.Itoa(len(args)))
}
if req.Role != nil {
role := normalizeRole(*req.Role, "")
if role == "" {
writeAPIError(w, http.StatusBadRequest, "invalid_role", "role must be user or admin")
return
}
args = append(args, role)
updates = append(updates, "role = $"+strconv.Itoa(len(args)))
}
if req.Status != nil {
status := normalizeUserStatus(*req.Status, "")
if status == "" {
writeAPIError(w, http.StatusBadRequest, "invalid_status", "status must be active, disabled or pending")
return
}
args = append(args, status)
updates = append(updates, "status = $"+strconv.Itoa(len(args)))
}
if req.Password != nil {
if len(*req.Password) < 10 {
writeAPIError(w, http.StatusBadRequest, "invalid_password", "password must be at least 10 characters")
return
}
passwordHash, err := hashPassword(*req.Password)
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "password_hash_failed", "could not hash password")
return
}
args = append(args, passwordHash)
updates = append(updates, "password_hash = $"+strconv.Itoa(len(args)))
}
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
planID := ""
if req.PlanID != nil {
planID = strings.TrimSpace(*req.PlanID)
if planID == "" {
writeAPIError(w, http.StatusBadRequest, "invalid_plan", "planId is required when updating a billing plan")
return
}
}
if len(updates) == 0 && planID == "" {
detail, err := s.adminUserDetail(ctx, userID)
if errors.Is(err, sql.ErrNoRows) {
writeAPIError(w, http.StatusNotFound, "user_not_found", "user not found")
return
}
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "admin_user_detail_failed", "could not load user detail")
return
}
writeJSON(w, http.StatusOK, map[string]any{"contractVersion": "user-billing-admin-user-v1", "updated": false, "detail": detail, "valuesRedacted": true})
return
}
if len(updates) == 0 {
detail, err := s.updateUserPlan(ctx, userID, planID)
if errors.Is(err, sql.ErrNoRows) {
writeAPIError(w, http.StatusNotFound, "user_or_plan_not_found", "user or plan not found")
return
}
if err != nil {
writeAPIError(w, http.StatusBadRequest, "plan_update_failed", err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"contractVersion": "user-billing-admin-user-v1", "updated": true, "detail": detail, "valuesRedacted": true})
return
}
query := `UPDATE hwlab_users SET ` + strings.Join(updates, ", ") + `, updated_at = now() WHERE id = $1 RETURNING id, email, username, display_name, status, role, email_verified, created_at`
var user User
err := s.db.QueryRowContext(ctx, query, args...).Scan(&user.ID, &user.Email, &user.Username, &user.DisplayName, &user.Status, &user.Role, &user.EmailVerified, &user.CreatedAt)
if errors.Is(err, sql.ErrNoRows) {
writeAPIError(w, http.StatusNotFound, "user_not_found", "user not found")
return
}
if err != nil {
if isDuplicate(err) {
writeAPIError(w, http.StatusConflict, "user_exists", "email or username already exists")
return
}
writeAPIError(w, http.StatusInternalServerError, "user_update_failed", "could not update user")
return
}
if planID != "" {
if _, err := s.updateUserPlan(ctx, userID, planID); err != nil {
writeAPIError(w, http.StatusBadRequest, "plan_update_failed", err.Error())
return
}
}
writeJSON(w, http.StatusOK, map[string]any{"contractVersion": "user-billing-admin-user-v1", "updated": true, "user": user, "valuesRedacted": true})
}
func (s *Server) handleAdminCreditAdjust(w http.ResponseWriter, r *http.Request) {
if !s.internalAllowed(w, r) || !s.databaseAvailable(w) {
return
}
var req struct {
UserID string `json:"userId"`
DeltaCredits int64 `json:"deltaCredits"`
Reason string `json:"reason"`
IdempotencyKey string `json:"idempotencyKey"`
Metadata map[string]any `json:"metadata"`
}
if !decodeJSON(w, r, &req) {
return
}
if strings.TrimSpace(req.UserID) == "" || req.DeltaCredits == 0 {
writeAPIError(w, http.StatusBadRequest, "invalid_adjustment", "userId and non-zero deltaCredits are required")
return
}
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
adjustment, err := s.adjustCredits(ctx, req.UserID, req.DeltaCredits, "admin_adjust", req.Reason, req.IdempotencyKey, req.Metadata)
if err != nil {
writeAPIError(w, http.StatusBadRequest, "credit_adjust_failed", err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"userId": req.UserID, "balance": adjustment.BalanceAfter, "adjustment": adjustment, "valuesRedacted": true})
}
func (s *Server) handleAdminUserStatus(w http.ResponseWriter, r *http.Request) {
if !s.internalAllowed(w, r) || !s.databaseAvailable(w) {
return
}
var req struct {
UserID string `json:"userId"`
Status string `json:"status"`
}
if !decodeJSON(w, r, &req) {
return
}
if req.Status != "active" && req.Status != "disabled" && req.Status != "pending" {
writeAPIError(w, http.StatusBadRequest, "invalid_status", "status must be active, disabled or pending")
return
}
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
res, err := s.db.ExecContext(ctx, `UPDATE hwlab_users SET status = $2, updated_at = now() WHERE id = $1`, req.UserID, req.Status)
if err != nil {
writeAPIError(w, http.StatusInternalServerError, "status_update_failed", "could not update user status")
return
}
affected, _ := res.RowsAffected()
if affected == 0 {
writeAPIError(w, http.StatusNotFound, "user_not_found", "user not found")
return
}
writeJSON(w, http.StatusOK, map[string]any{"userId": req.UserID, "status": req.Status})
}
func (s *Server) handleAdminBillingSummary(w http.ResponseWriter, r *http.Request) {
if !s.internalAllowed(w, r) || !s.databaseAvailable(w) {
return
}
limit := queryLimit(r, "limit", 50, 100)
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
users, err := s.adminBillingUsers(ctx, limit)
if err != nil {
s.logDatabaseError("admin_billing_summary_users", err)
writeAPIError(w, http.StatusInternalServerError, "admin_billing_summary_failed", "could not load admin billing summary")
return
}
totals, err := s.adminBillingTotals(ctx)
if err != nil {
s.logDatabaseError("admin_billing_summary_totals", err)
writeAPIError(w, http.StatusInternalServerError, "admin_billing_summary_failed", "could not load admin billing totals")
return
}
writeJSON(w, http.StatusOK, map[string]any{
"contractVersion": "user-billing-admin-summary-v1",
"serviceId": serviceID,
"status": "ok",
"users": users,
"count": len(users),
"limit": limit,
"totals": totals,
"state": map[string]any{
"stateless": true,
"stateAuthority": s.config.StateAuthority,
"database": map[string]any{"authority": s.config.ExternalDatabaseLabel},
"redis": map[string]any{"role": "cache-only"},
},
"valuesRedacted": true,
})
}
func (s *Server) handleAdminBillingPlans(w http.ResponseWriter, r *http.Request) {
if !s.internalAllowed(w, r) || !s.databaseAvailable(w) {
return
}
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
plans, err := s.billingPlans(ctx)
if err != nil {
s.logDatabaseError("admin_billing_plans", err)
writeAPIError(w, http.StatusInternalServerError, "admin_billing_plans_failed", "could not load billing plans")
return
}
writeJSON(w, http.StatusOK, map[string]any{"contractVersion": "user-billing-admin-plans-v1", "plans": plans, "count": len(plans), "limitSemantics": "zero-means-unlimited", "valuesRedacted": true})
}
var errInsufficientCredits = errors.New("insufficient credits")
type billingLimitError struct {
Code string
Message string
Status int
}
func (err billingLimitError) Error() string { return err.Message }
func (s *Server) reserveCredits(ctx context.Context, userID, apiKeyID, service string, credits, tokens int64, idempotencyKey string, metadata map[string]any) (map[string]any, error) {
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
if err != nil {
return nil, err
}
defer tx.Rollback()
if idempotencyKey != "" {
var existingID, status string
var estimated int64
var resourceType string
err := tx.QueryRowContext(ctx, `SELECT id, status, estimated_credits, resource_type FROM hwlab_billing_reservations WHERE idempotency_key = $1`, idempotencyKey).Scan(&existingID, &status, &estimated, &resourceType)
if err == nil {
return map[string]any{"allowed": status == "reserved", "reservationId": existingID, "estimatedCredits": estimated, "resourceType": resourceType, "status": status, "idempotent": true}, tx.Commit()
}
if !errors.Is(err, sql.ErrNoRows) {
return nil, err
}
}
var balance, reserved int64
var planID string
if err := tx.QueryRowContext(ctx, `SELECT balance_credits, reserved_credits, plan_id FROM hwlab_credit_accounts WHERE user_id = $1 FOR UPDATE`, userID).Scan(&balance, &reserved, &planID); err != nil {
return nil, err
}
resourceType := resourceTypeForService(service)
entitlement, err := s.evaluateResourceEntitlement(ctx, tx, userID, planID, resourceType, credits)
if err != nil {
return nil, err
}
available := balance - reserved
if available < credits {
return nil, errInsufficientCredits
}
reservationID := newID("res")
metadataJSON := jsonObject(metadata)
expiresAt := time.Now().UTC().Add(s.config.ReservationTTL)
_, err = tx.ExecContext(ctx, `INSERT INTO hwlab_billing_reservations (id, user_id, service_id, resource_type, estimated_credits, estimated_tokens, api_key_id, idempotency_key, metadata, expires_at) VALUES ($1, $2, $3, $4, $5, $6, nullif($7, ''), nullif($8, ''), $9::jsonb, $10)`, reservationID, userID, service, resourceType, credits, nonNegative(tokens), apiKeyID, idempotencyKey, metadataJSON, expiresAt)
if err != nil {
return nil, err
}
_, err = tx.ExecContext(ctx, `UPDATE hwlab_credit_accounts SET reserved_credits = reserved_credits + $2, updated_at = now() WHERE user_id = $1`, userID, credits)
if err != nil {
return nil, err
}
if err := tx.Commit(); err != nil {
return nil, err
}
return map[string]any{"allowed": true, "reservationId": reservationID, "estimatedCredits": credits, "estimatedTokens": tokens, "resourceType": resourceType, "planId": planID, "entitlement": entitlement, "availableBefore": available, "expiresAt": expiresAt}, nil
}
func (s *Server) recordUsage(ctx context.Context, r *http.Request, reservationID, token, apiKey, userID, serviceID string, usedCredits, usedTokens int64, idempotencyKey string, metadata map[string]any) (map[string]any, error) {
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
if err != nil {
return nil, err
}
defer tx.Rollback()
if idempotencyKey != "" {
var existingID string
err := tx.QueryRowContext(ctx, `SELECT id FROM hwlab_usage_records WHERE idempotency_key = $1`, idempotencyKey).Scan(&existingID)
if err == nil {
return map[string]any{"recordId": existingID, "idempotent": true}, tx.Commit()
}
if !errors.Is(err, sql.ErrNoRows) {
return nil, err
}
}
var reserved int64
apiKeyID := ""
resourceType := ""
if reservationID != "" {
var status string
err := tx.QueryRowContext(ctx, `SELECT user_id, service_id, resource_type, estimated_credits, status, COALESCE(api_key_id, '') FROM hwlab_billing_reservations WHERE id = $1 FOR UPDATE`, reservationID).Scan(&userID, &serviceID, &resourceType, &reserved, &status, &apiKeyID)
if err != nil {
return nil, err
}
if status != "reserved" {
return nil, fmt.Errorf("reservation status is %s", status)
}
} else {
principal, err := s.principalForBilling(ctx, r, token, apiKey, userID)
if err != nil {
return nil, err
}
userID = principal.UserID
apiKeyID = principal.KeyID
serviceID = strings.TrimSpace(serviceID)
resourceType = resourceTypeForService(serviceID)
}
if serviceID == "" {
return nil, errors.New("serviceId is required")
}
if resourceType == "" {
resourceType = resourceTypeForService(serviceID)
}
credits := s.estimateCredits(usedCredits, usedTokens)
var balance, accountReserved int64
var planID string
if err := tx.QueryRowContext(ctx, `SELECT balance_credits, reserved_credits, plan_id FROM hwlab_credit_accounts WHERE user_id = $1 FOR UPDATE`, userID).Scan(&balance, &accountReserved, &planID); err != nil {
return nil, err
}
if reservationID == "" {
if _, err := s.evaluateResourceEntitlement(ctx, tx, userID, planID, resourceType, credits); err != nil {
return nil, err
}
}
if reservationID == "" && balance-accountReserved < credits {
return nil, errInsufficientCredits
}
newReserved := accountReserved - reserved
if newReserved < 0 {
newReserved = 0
}
newBalance := balance - credits
if newBalance < 0 {
return nil, errInsufficientCredits
}
_, err = tx.ExecContext(ctx, `UPDATE hwlab_credit_accounts SET balance_credits = $2, reserved_credits = $3, updated_at = now() WHERE user_id = $1`, userID, newBalance, newReserved)
if err != nil {
return nil, err
}
recordID := newID("use")
metadataJSON := jsonObject(metadata)
_, err = tx.ExecContext(ctx, `INSERT INTO hwlab_usage_records (id, reservation_id, user_id, service_id, resource_type, api_key_id, quantity, credits, status, idempotency_key, metadata) VALUES ($1, nullif($2, ''), $3, $4, $5, nullif($6, ''), $7, $8, 'recorded', nullif($9, ''), $10::jsonb)`, recordID, reservationID, userID, serviceID, resourceType, apiKeyID, nonNegative(usedTokens), credits, idempotencyKey, metadataJSON)
if err != nil {
return nil, err
}
usageLedgerKey := ""
if idempotencyKey != "" {
usageLedgerKey = "usage:" + idempotencyKey
}
_, err = tx.ExecContext(ctx, `INSERT INTO hwlab_credit_ledger (id, user_id, delta_credits, balance_before, balance_after, kind, reason, source, status, idempotency_key, metadata) VALUES ($1, $2, $3, $4, $5, 'usage', $6, 'usage', 'applied', nullif($7, ''), $8::jsonb)`, newID("led"), userID, -credits, balance, newBalance, serviceID, usageLedgerKey, metadataJSON)
if err != nil {
return nil, err
}
if reservationID != "" {
_, err = tx.ExecContext(ctx, `UPDATE hwlab_billing_reservations SET status = 'completed', updated_at = now() WHERE id = $1`, reservationID)
if err != nil {
return nil, err
}
}
if err := tx.Commit(); err != nil {
return nil, err
}
return map[string]any{"recordId": recordID, "userId": userID, "serviceId": serviceID, "resourceType": resourceType, "credits": credits, "balance": newBalance}, nil
}
func (s *Server) releaseReservation(ctx context.Context, reservationID, serviceID, reason, idempotencyKey string, metadata map[string]any) (map[string]any, error) {
reservationID = strings.TrimSpace(reservationID)
if reservationID == "" {
return nil, errors.New("reservationId is required")
}
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
if err != nil {
return nil, err
}
defer tx.Rollback()
var userID, actualServiceID, resourceType, status string
var reserved int64
err = tx.QueryRowContext(ctx, `SELECT user_id, service_id, resource_type, estimated_credits, status FROM hwlab_billing_reservations WHERE id = $1 FOR UPDATE`, reservationID).Scan(&userID, &actualServiceID, &resourceType, &reserved, &status)
if err != nil {
return nil, err
}
serviceID = strings.TrimSpace(serviceID)
if serviceID != "" && serviceID != actualServiceID {
return nil, fmt.Errorf("reservation service is %s", actualServiceID)
}
if status == "cancelled" || status == "expired" {
return map[string]any{"released": true, "reservationId": reservationID, "userId": userID, "serviceId": actualServiceID, "resourceType": resourceType, "releasedCredits": int64(0), "status": status, "idempotent": true}, tx.Commit()
}
if status != "reserved" {
return nil, fmt.Errorf("reservation status is %s", status)
}
var accountReserved int64
if err := tx.QueryRowContext(ctx, `SELECT reserved_credits FROM hwlab_credit_accounts WHERE user_id = $1 FOR UPDATE`, userID).Scan(&accountReserved); err != nil {
return nil, err
}
newReserved := accountReserved - reserved
if newReserved < 0 {
newReserved = 0
}
_, err = tx.ExecContext(ctx, `UPDATE hwlab_credit_accounts SET reserved_credits = $2, updated_at = now() WHERE user_id = $1`, userID, newReserved)
if err != nil {
return nil, err
}
releaseMetadata := map[string]any{
"releaseReason": strings.TrimSpace(reason),
"releaseIdempotency": strings.TrimSpace(idempotencyKey),
"releaseValuesRedacted": true,
}
for key, value := range metadata {
releaseMetadata[key] = value
}
_, err = tx.ExecContext(ctx, `UPDATE hwlab_billing_reservations SET status = 'cancelled', metadata = COALESCE(metadata, '{}'::jsonb) || $2::jsonb, updated_at = now() WHERE id = $1`, reservationID, jsonObject(releaseMetadata))
if err != nil {
return nil, err
}
if err := tx.Commit(); err != nil {
return nil, err
}
return map[string]any{"released": true, "reservationId": reservationID, "userId": userID, "serviceId": actualServiceID, "resourceType": resourceType, "releasedCredits": reserved, "status": "cancelled"}, nil
}
func (s *Server) adjustCredits(ctx context.Context, userID string, delta int64, kind, reason, idempotencyKey string, metadata map[string]any) (creditAdjustResult, error) {
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
if err != nil {
return creditAdjustResult{}, err
}
defer tx.Rollback()
operatorUserID, operatorUsername, _ := operatorFields(metadata)
if idempotencyKey != "" {
var result creditAdjustResult
var before sql.NullInt64
err := tx.QueryRowContext(ctx, `SELECT id, user_id, COALESCE(balance_before, balance_after - delta_credits), balance_after, delta_credits, kind, reason, source, status FROM hwlab_credit_ledger WHERE idempotency_key = $1`, idempotencyKey).Scan(&result.LedgerID, &result.UserID, &before, &result.BalanceAfter, &result.DeltaCredits, &result.Kind, &result.Reason, &result.Source, &result.Status)
if err == nil {
if before.Valid {
result.BalanceBefore = before.Int64
}
result.IdempotencyKey = idempotencyKey
result.Idempotent = true
return result, tx.Commit()
}
if !errors.Is(err, sql.ErrNoRows) {
return creditAdjustResult{}, err
}
}
var balance int64
if err := tx.QueryRowContext(ctx, `SELECT balance_credits FROM hwlab_credit_accounts WHERE user_id = $1 FOR UPDATE`, userID).Scan(&balance); err != nil {
return creditAdjustResult{}, err
}
newBalance := balance + delta
if newBalance < 0 {
return creditAdjustResult{}, errInsufficientCredits
}
_, err = tx.ExecContext(ctx, `UPDATE hwlab_credit_accounts SET balance_credits = $2, updated_at = now() WHERE user_id = $1`, userID, newBalance)
if err != nil {
return creditAdjustResult{}, err
}
ledgerID := newID("led")
_, err = tx.ExecContext(ctx, `INSERT INTO hwlab_credit_ledger (id, user_id, delta_credits, balance_before, balance_after, kind, reason, source, status, idempotency_key, operator_user_id, operator_username, metadata) VALUES ($1, $2, $3, $4, $5, $6, $7, 'admin', 'applied', nullif($8, ''), nullif($9, ''), nullif($10, ''), $11::jsonb)`, ledgerID, userID, delta, balance, newBalance, kind, reason, idempotencyKey, operatorUserID, operatorUsername, jsonObject(metadata))
if err != nil {
return creditAdjustResult{}, err
}
result := creditAdjustResult{UserID: userID, BalanceBefore: balance, BalanceAfter: newBalance, DeltaCredits: delta, LedgerID: ledgerID, Kind: kind, Reason: reason, Source: "admin", Status: "applied", IdempotencyKey: idempotencyKey, Idempotent: false}
return result, tx.Commit()
}
type txQueryer interface {
QueryContext(context.Context, string, ...any) (*sql.Rows, error)
QueryRowContext(context.Context, string, ...any) *sql.Row
}
func (s *Server) evaluateResourceEntitlement(ctx context.Context, tx txQueryer, userID, planID, resourceType string, requestedCredits int64) (resourceEntitlementSummary, error) {
plan, err := s.planByID(ctx, tx, planID)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return resourceEntitlementSummary{}, billingLimitError{Code: "plan_not_found", Message: "billing plan is not active", Status: http.StatusForbidden}
}
return resourceEntitlementSummary{}, err
}
if plan.Status != "active" {
return resourceEntitlementSummary{}, billingLimitError{Code: "plan_disabled", Message: "billing plan is disabled", Status: http.StatusForbidden}
}
entitlements, err := s.resourceEntitlements(ctx, tx, planID, userID)
if err != nil {
return resourceEntitlementSummary{}, err
}
var entitlement resourceEntitlementSummary
found := false
for _, item := range entitlements {
if item.ResourceType == resourceType {
entitlement = item
found = true
break
}
}
if !found || !entitlement.Enabled {
return resourceEntitlementSummary{}, billingLimitError{Code: "resource_not_entitled", Message: "resource is not enabled for this billing plan", Status: http.StatusForbidden}
}
if entitlement.MaxConcurrentReservations > 0 && entitlement.ActiveReservations >= entitlement.MaxConcurrentReservations {
return entitlement, billingLimitError{Code: "resource_concurrency_exceeded", Message: "resource concurrency limit exceeded", Status: http.StatusTooManyRequests}
}
if entitlement.RPMLimit > 0 && entitlement.RPMUsed >= entitlement.RPMLimit {
return entitlement, billingLimitError{Code: "resource_rpm_exceeded", Message: "resource RPM limit exceeded", Status: http.StatusTooManyRequests}
}
if entitlement.MonthlyQuotaCredits > 0 && entitlement.MonthlyUsageCredits+requestedCredits > entitlement.MonthlyQuotaCredits {
return entitlement, billingLimitError{Code: "resource_monthly_quota_exceeded", Message: "resource monthly quota exceeded", Status: http.StatusPaymentRequired}
}
return entitlement, nil
}
func (s *Server) planByID(ctx context.Context, queryer txQueryer, planID string) (billingPlanSummary, error) {
planID = strings.TrimSpace(planID)
if planID == "" {
planID = "default"
}
var plan billingPlanSummary
var metadataRaw string
err := queryer.QueryRowContext(ctx, `SELECT id, display_name, status, description, metadata::text FROM hwlab_billing_plans WHERE id = $1`, planID).Scan(&plan.ID, &plan.DisplayName, &plan.Status, &plan.Description, &metadataRaw)
if err != nil {
return plan, err
}
plan.Metadata = parseJSONMap(metadataRaw)
return plan, nil
}
func (s *Server) billingPlans(ctx context.Context) ([]billingPlanDetail, error) {
rows, err := s.db.QueryContext(ctx, `SELECT id, display_name, status, description, metadata::text FROM hwlab_billing_plans ORDER BY id ASC`)
if err != nil {
return nil, err
}
defer rows.Close()
plans := []billingPlanDetail{}
for rows.Next() {
var detail billingPlanDetail
var metadataRaw string
if err := rows.Scan(&detail.Plan.ID, &detail.Plan.DisplayName, &detail.Plan.Status, &detail.Plan.Description, &metadataRaw); err != nil {
return nil, err
}
detail.Plan.Metadata = parseJSONMap(metadataRaw)
plans = append(plans, detail)
}
if err := rows.Err(); err != nil {
return nil, err
}
for i := range plans {
entitlements, err := s.resourceEntitlements(ctx, s.db, plans[i].Plan.ID, "")
if err != nil {
return nil, err
}
plans[i].Entitlements = entitlements
}
return plans, nil
}
func (s *Server) updateUserPlan(ctx context.Context, userID, planID string) (adminUserDetail, error) {
planID = strings.TrimSpace(planID)
if planID == "" {
return adminUserDetail{}, errors.New("planId is required")
}
if _, err := s.planByID(ctx, s.db, planID); err != nil {
return adminUserDetail{}, err
}
res, err := s.db.ExecContext(ctx, `UPDATE hwlab_credit_accounts SET plan_id = $2, updated_at = now() WHERE user_id = $1`, userID, planID)
if err != nil {
return adminUserDetail{}, err
}
affected, _ := res.RowsAffected()
if affected == 0 {
return adminUserDetail{}, sql.ErrNoRows
}
return s.adminUserDetail(ctx, userID)
}
func (s *Server) resourceEntitlements(ctx context.Context, queryer txQueryer, planID, userID string) ([]resourceEntitlementSummary, error) {
rows, err := queryer.QueryContext(ctx, `
SELECT id, plan_id, resource_type, enabled, monthly_quota_credits, max_concurrent_reservations, rpm_limit, metadata::text
FROM hwlab_resource_entitlements
WHERE plan_id = $1
ORDER BY resource_type ASC`, planID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []resourceEntitlementSummary{}
for rows.Next() {
var item resourceEntitlementSummary
var metadataRaw string
if err := rows.Scan(&item.ID, &item.PlanID, &item.ResourceType, &item.Enabled, &item.MonthlyQuotaCredits, &item.MaxConcurrentReservations, &item.RPMLimit, &metadataRaw); err != nil {
return nil, err
}
item.Metadata = parseJSONMap(metadataRaw)
item.LimitSemantics = "zero-means-unlimited"
item.MonthlyQuotaUnlimited = item.MonthlyQuotaCredits == 0
item.ConcurrencyUnlimited = item.MaxConcurrentReservations == 0
item.RPMUnlimited = item.RPMLimit == 0
items = append(items, item)
}
if err := rows.Err(); err != nil {
return nil, err
}
if strings.TrimSpace(userID) == "" || len(items) == 0 {
return items, nil
}
for i := range items {
if err := s.populateEntitlementUsage(ctx, queryer, userID, &items[i]); err != nil {
return nil, err
}
}
return items, nil
}
func (s *Server) populateEntitlementUsage(ctx context.Context, queryer txQueryer, userID string, item *resourceEntitlementSummary) error {
monthStart := time.Now().UTC().Format("2006-01-02")
if err := queryer.QueryRowContext(ctx, `SELECT COALESCE(SUM(credits), 0) FROM hwlab_usage_records WHERE user_id = $1 AND resource_type = $2 AND created_at >= date_trunc('month', $3::timestamptz)`, userID, item.ResourceType, monthStart).Scan(&item.MonthlyUsageCredits); err != nil {
return err
}
if err := queryer.QueryRowContext(ctx, `SELECT COUNT(*) FROM hwlab_billing_reservations WHERE user_id = $1 AND resource_type = $2 AND status = 'reserved' AND expires_at > now()`, userID, item.ResourceType).Scan(&item.ActiveReservations); err != nil {
return err
}
if err := queryer.QueryRowContext(ctx, `SELECT COUNT(*) FROM hwlab_billing_reservations WHERE user_id = $1 AND resource_type = $2 AND created_at >= now() - interval '1 minute'`, userID, item.ResourceType).Scan(&item.RPMUsed); err != nil {
return err
}
if item.MonthlyQuotaCredits > 0 {
item.MonthlyRemainingCredits = item.MonthlyQuotaCredits - item.MonthlyUsageCredits
if item.MonthlyRemainingCredits < 0 {
item.MonthlyRemainingCredits = 0
}
}
return nil
}
func resourceTypeForService(serviceID string) string {
value := strings.ToLower(strings.TrimSpace(serviceID))
switch {
case strings.Contains(value, "code-agent") || strings.Contains(value, "code_agent") || strings.Contains(value, "codex"):
return "code_agent"
case strings.Contains(value, "aipod") || strings.Contains(value, "ai-pod"):
return "aipod"
case strings.Contains(value, "hwpod") || strings.Contains(value, "hw-pod"):
return "hwpod"
case value != "":
return strings.Map(func(r rune) rune {
if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') || r == '_' {
return r
}
if r == '-' || r == '.' || r == '/' {
return '_'
}
return -1
}, value)
default:
return "generic"
}
}
func parseJSONMap(raw string) map[string]any {
result := map[string]any{}
_ = json.Unmarshal([]byte(raw), &result)
return result
}
func operatorFields(metadata map[string]any) (string, string, string) {
operator, _ := metadata["operator"].(map[string]any)
if operator == nil {
return "", "", ""
}
return stringFromAny(operator["id"]), stringFromAny(operator["username"]), stringFromAny(operator["role"])
}
func stringFromAny(value any) string {
if value == nil {
return ""
}
return strings.TrimSpace(fmt.Sprint(value))
}
func (s *Server) principalForBilling(ctx context.Context, r *http.Request, token, apiKey, userID string) (*Principal, error) {
if token == "" && apiKey == "" {
token = bearerToken(r)
}
if token != "" || apiKey != "" {
return s.principalFromToken(ctx, first(token, apiKey))
}
userID = strings.TrimSpace(userID)
if userID == "" {
return nil, errors.New("billing subject is required")
}
var principal Principal
err := s.db.QueryRowContext(ctx, `SELECT id, email, username, role FROM hwlab_users WHERE id = $1 AND status = 'active'`, userID).Scan(&principal.UserID, &principal.Email, &principal.Username, &principal.Role)
if err != nil {
return nil, err
}
principal.AuthType = "internal-user-id"
principal.Scopes = []string{"api"}
return &principal, nil
}
func (s *Server) principalFromToken(ctx context.Context, token string) (*Principal, error) {
token = strings.TrimSpace(token)
if token == "" {
return nil, errors.New("empty token")
}
hash := tokenHash(token)
var principal Principal
if strings.HasPrefix(token, "hwl_") {
var scopesRaw string
err := s.db.QueryRowContext(ctx, `SELECT u.id, u.email, u.username, u.role, k.id, k.scopes_json::text FROM hwlab_api_keys k JOIN hwlab_users u ON u.id = k.user_id WHERE k.key_hash = $1 AND k.status = 'active' AND k.revoked_at IS NULL AND u.status = 'active'`, hash).Scan(&principal.UserID, &principal.Email, &principal.Username, &principal.Role, &principal.KeyID, &scopesRaw)
if err != nil {
return nil, err
}
_ = json.Unmarshal([]byte(scopesRaw), &principal.Scopes)
if len(principal.Scopes) == 0 {
principal.Scopes = []string{"api"}
}
principal.AuthType = "api-key"
_, _ = s.db.ExecContext(ctx, `UPDATE hwlab_api_keys SET last_used_at = now() WHERE id = $1`, principal.KeyID)
return &principal, nil
}
err := s.db.QueryRowContext(ctx, `SELECT u.id, u.email, u.username, u.role FROM hwlab_user_sessions s JOIN hwlab_users u ON u.id = s.user_id WHERE s.token_hash = $1 AND s.revoked_at IS NULL AND s.expires_at > now() AND u.status = 'active'`, hash).Scan(&principal.UserID, &principal.Email, &principal.Username, &principal.Role)
if err != nil {
return nil, err
}
principal.AuthType = "session"
principal.Scopes = []string{"session"}
return &principal, nil
}
func (s *Server) requirePrincipal(w http.ResponseWriter, r *http.Request) (*Principal, bool) {
if !s.databaseAvailable(w) {
return nil, false
}
principal, err := s.principalFromToken(r.Context(), bearerToken(r))
if err != nil {
writeAPIError(w, http.StatusUnauthorized, "unauthorized", "valid bearer session or API key is required")
return nil, false
}
return principal, true
}
func (s *Server) internalAllowed(w http.ResponseWriter, r *http.Request) bool {
if s.config.InternalToken == "" {
if s.config.RequireInternalToken {
writeAPIError(w, http.StatusServiceUnavailable, "internal_token_missing", "internal token is required but not configured")
return false
}
return true
}
candidate := strings.TrimSpace(r.Header.Get("X-HWLAB-Internal-Token"))
if candidate == "" {
candidate = bearerToken(r)
}
if subtle.ConstantTimeCompare([]byte(candidate), []byte(s.config.InternalToken)) != 1 {
writeAPIError(w, http.StatusUnauthorized, "internal_auth_failed", "valid internal token is required")
return false
}
return true
}
func (s *Server) databaseAvailable(w http.ResponseWriter) bool {
if s.db != nil {
return true
}
writeAPIError(w, http.StatusServiceUnavailable, "state_backend_not_configured", "HWLAB_USER_BILLING_DB_URL must point to the PK01 external PostgreSQL authority")
return false
}
func (s *Server) registrationAvailable(w http.ResponseWriter) bool {
if !s.RegistrationEnabled() {
writeAPIError(w, http.StatusForbidden, "registration_disabled", "registration is disabled")
return false
}
return s.databaseAvailable(w)
}
func (s *Server) RegistrationEnabled() bool {
return s.config.RegistrationEnabled
}
func (s *Server) readyWriteProbe(ctx context.Context) error {
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
if err != nil {
return err
}
probeID := newID("usr_probe")
_, err = tx.ExecContext(ctx, `INSERT INTO hwlab_users (id, email, username, display_name, password_hash, status, role) VALUES ($1, $2, $3, 'readiness probe', 'readiness-probe', 'pending', 'user')`, probeID, probeID+"@readiness.invalid", probeID)
if err != nil {
_ = tx.Rollback()
return err
}
return tx.Rollback()
}
func (s *Server) logDatabaseError(operation string, err error) {
log.Printf("%s database operation=%s errorClass=%s errorCode=%s valuesRedacted=true", serviceID, operation, classifyDatabaseError(err), databaseErrorCode(err))
}
func classifyDatabaseError(err error) string {
if err == nil {
return "none"
}
text := strings.ToLower(err.Error())
code := strings.ToUpper(databaseErrorCode(err))
if code == "57014" || errors.Is(err, context.DeadlineExceeded) || strings.Contains(text, "deadline exceeded") || strings.Contains(text, "timeout") {
return "timeout"
}
if strings.Contains(text, "certificate") || strings.Contains(text, "ssl") || strings.Contains(text, "tls") {
return "ssl"
}
if strings.Contains(text, "password") || strings.Contains(text, "authentication") || strings.Contains(text, "pg_hba") || code == "28P01" || strings.HasPrefix(code, "28") {
return "auth"
}
if strings.Contains(text, "too many connections") || strings.Contains(text, "connection refused") || strings.Contains(text, "connection reset") || strings.Contains(text, "broken pipe") {
return "connection"
}
return "database"
}
func databaseErrorCode(err error) string {
if err == nil {
return ""
}
type sqlState interface{ SQLState() string }
var state sqlState
if errors.As(err, &state) {
return state.SQLState()
}
type coder interface{ Code() string }
var coded coder
if errors.As(err, &coded) {
return coded.Code()
}
return ""
}
func (s *Server) accountBalance(ctx context.Context, userID string) (int64, int64, error) {
var balance, reserved int64
err := s.db.QueryRowContext(ctx, `SELECT balance_credits, reserved_credits FROM hwlab_credit_accounts WHERE user_id = $1`, userID).Scan(&balance, &reserved)
return balance, reserved, err
}
func (s *Server) accountPlanContext(ctx context.Context, userID string) (string, billingPlanSummary, []resourceEntitlementSummary, error) {
planID := "default"
if err := s.db.QueryRowContext(ctx, `SELECT COALESCE(plan_id, 'default') FROM hwlab_credit_accounts WHERE user_id = $1`, userID).Scan(&planID); err != nil {
return "", billingPlanSummary{}, nil, err
}
plan, err := s.planByID(ctx, s.db, planID)
if err != nil {
return planID, billingPlanSummary{}, nil, err
}
entitlements, err := s.resourceEntitlements(ctx, s.db, planID, userID)
return planID, plan, entitlements, err
}
func (s *Server) usageTotals(ctx context.Context, userID string) (int64, int64, int64, error) {
var credits, quantity, count int64
err := s.db.QueryRowContext(ctx, `SELECT COALESCE(SUM(credits), 0), COALESCE(SUM(quantity), 0), COUNT(*) FROM hwlab_usage_records WHERE user_id = $1`, userID).Scan(&credits, &quantity, &count)
return credits, quantity, count, err
}
func (s *Server) usageByService(ctx context.Context, userID string, limit int) ([]serviceUsageSummary, error) {
rows, err := s.db.QueryContext(ctx, `SELECT service_id, resource_type, COALESCE(SUM(credits), 0), COALESCE(SUM(quantity), 0), COUNT(*), MAX(created_at) FROM hwlab_usage_records WHERE user_id = $1 GROUP BY service_id, resource_type ORDER BY MAX(created_at) DESC NULLS LAST, service_id ASC LIMIT $2`, userID, limit)
if err != nil {
return nil, err
}
defer rows.Close()
items := []serviceUsageSummary{}
for rows.Next() {
var item serviceUsageSummary
if err := rows.Scan(&item.ServiceID, &item.ResourceType, &item.Credits, &item.Quantity, &item.RecordCount, &item.LastUsedAt); err != nil {
return nil, err
}
items = append(items, item)
}
return items, rows.Err()
}
func (s *Server) ledgerRows(ctx context.Context, userID string, limit int) ([]ledgerSummaryRow, error) {
rows, err := s.db.QueryContext(ctx, `SELECT id, kind, reason, delta_credits, balance_before, balance_after, source, status, COALESCE(idempotency_key, ''), COALESCE(operator_user_id, ''), COALESCE(operator_username, ''), metadata::text, created_at FROM hwlab_credit_ledger WHERE user_id = $1 ORDER BY created_at DESC, id DESC LIMIT $2`, userID, limit)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ledgerSummaryRow{}
for rows.Next() {
var item ledgerSummaryRow
var before sql.NullInt64
var metadataRaw string
if err := rows.Scan(&item.ID, &item.Kind, &item.Reason, &item.DeltaCredits, &before, &item.BalanceAfter, &item.Source, &item.Status, &item.IdempotencyKey, &item.OperatorUserID, &item.OperatorUsername, &metadataRaw, &item.CreatedAt); err != nil {
return nil, err
}
if before.Valid {
value := before.Int64
item.BalanceBefore = &value
}
item.Metadata = parseJSONMap(metadataRaw)
items = append(items, item)
}
return items, rows.Err()
}
func (s *Server) activeReservations(ctx context.Context, userID string, limit int) ([]reservationSummaryRow, error) {
rows, err := s.db.QueryContext(ctx, `SELECT id, service_id, resource_type, estimated_credits, estimated_tokens, status, expires_at, created_at FROM hwlab_billing_reservations WHERE user_id = $1 AND status = 'reserved' AND expires_at > now() ORDER BY expires_at ASC, created_at DESC LIMIT $2`, userID, limit)
if err != nil {
return nil, err
}
defer rows.Close()
items := []reservationSummaryRow{}
for rows.Next() {
var item reservationSummaryRow
if err := rows.Scan(&item.ReservationID, &item.ServiceID, &item.ResourceType, &item.EstimatedCredits, &item.EstimatedTokens, &item.Status, &item.ExpiresAt, &item.CreatedAt); err != nil {
return nil, err
}
items = append(items, item)
}
return items, rows.Err()
}
func (s *Server) adminBillingUsers(ctx context.Context, limit int) ([]adminBillingUserSummary, error) {
rows, err := s.db.QueryContext(ctx, `
SELECT
u.id,
u.email,
u.username,
u.display_name,
u.status,
u.role,
u.email_verified,
u.created_at,
COALESCE(a.plan_id, 'default'),
COALESCE(a.balance_credits, 0),
COALESCE(a.reserved_credits, 0),
COALESCE(usage.total_credits, 0),
COALESCE(usage.total_quantity, 0),
COALESCE(usage.record_count, 0),
usage.last_used_at,
COALESCE(keys.active_count, 0),
COALESCE(keys.total_count, 0),
keys.last_key_used_at,
COALESCE(res.active_count, 0)
FROM hwlab_users u
LEFT JOIN hwlab_credit_accounts a ON a.user_id = u.id
LEFT JOIN (
SELECT user_id, SUM(credits) AS total_credits, SUM(quantity) AS total_quantity, COUNT(*) AS record_count, MAX(created_at) AS last_used_at
FROM hwlab_usage_records
GROUP BY user_id
) usage ON usage.user_id = u.id
LEFT JOIN (
SELECT user_id, COUNT(*) FILTER (WHERE status = 'active' AND revoked_at IS NULL) AS active_count, COUNT(*) AS total_count, MAX(last_used_at) AS last_key_used_at
FROM hwlab_api_keys
GROUP BY user_id
) keys ON keys.user_id = u.id
LEFT JOIN (
SELECT user_id, COUNT(*) AS active_count
FROM hwlab_billing_reservations
WHERE status = 'reserved' AND expires_at > now()
GROUP BY user_id
) res ON res.user_id = u.id
ORDER BY usage.last_used_at DESC NULLS LAST, u.created_at DESC, u.username ASC
LIMIT $1`, limit)
if err != nil {
return nil, err
}
defer rows.Close()
users := []adminBillingUserSummary{}
userIDs := []string{}
for rows.Next() {
var item adminBillingUserSummary
var lastUsed sql.NullTime
var lastKeyUsed sql.NullTime
if err := rows.Scan(&item.User.ID, &item.User.Email, &item.User.Username, &item.User.DisplayName, &item.User.Status, &item.User.Role, &item.User.EmailVerified, &item.User.CreatedAt, &item.PlanID, &item.Credits.Balance, &item.Credits.Reserved, &item.Usage.TotalCredits, &item.Usage.TotalQuantity, &item.Usage.RecordCount, &lastUsed, &item.APIKeys.ActiveCount, &item.APIKeys.TotalCount, &lastKeyUsed, &item.Reservations.ActiveCount); err != nil {
return nil, err
}
if lastUsed.Valid {
value := lastUsed.Time
item.Usage.LastUsedAt = &value
}
if lastKeyUsed.Valid {
value := lastKeyUsed.Time
item.APIKeys.LastUsedAt = &value
}
item.Credits.Available = item.Credits.Balance - item.Credits.Reserved
userIDs = append(userIDs, item.User.ID)
users = append(users, item)
}
if err := rows.Err(); err != nil {
return nil, err
}
reservations, err := s.activeReservationsForUsers(ctx, userIDs, 5)
if err != nil {
return nil, err
}
for i := range users {
users[i].Reservations.Active = reservations[users[i].User.ID]
}
if err := s.enrichAdminBillingUsers(ctx, users); err != nil {
return nil, err
}
return users, nil
}
func (s *Server) activeReservationsForUsers(ctx context.Context, userIDs []string, perUserLimit int) (map[string][]reservationSummaryRow, error) {
items := map[string][]reservationSummaryRow{}
if len(userIDs) == 0 {
return items, nil
}
placeholders := make([]string, len(userIDs))
args := make([]any, 0, len(userIDs)+1)
for i, userID := range userIDs {
placeholders[i] = "$" + strconv.Itoa(i+1)
args = append(args, userID)
}
limitParam := "$" + strconv.Itoa(len(args)+1)
args = append(args, perUserLimit)
query := `
SELECT user_id, id, service_id, resource_type, estimated_credits, estimated_tokens, status, expires_at, created_at
FROM (
SELECT user_id, id, service_id, resource_type, estimated_credits, estimated_tokens, status, expires_at, created_at,
row_number() OVER (PARTITION BY user_id ORDER BY expires_at ASC, created_at DESC) AS rn
FROM hwlab_billing_reservations
WHERE user_id IN (` + strings.Join(placeholders, ", ") + `) AND status = 'reserved' AND expires_at > now()
) ranked
WHERE rn <= ` + limitParam + `
ORDER BY expires_at ASC, created_at DESC`
rows, err := s.db.QueryContext(ctx, query, args...)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var userID string
var item reservationSummaryRow
if err := rows.Scan(&userID, &item.ReservationID, &item.ServiceID, &item.ResourceType, &item.EstimatedCredits, &item.EstimatedTokens, &item.Status, &item.ExpiresAt, &item.CreatedAt); err != nil {
return nil, err
}
items[userID] = append(items[userID], item)
}
return items, rows.Err()
}
func (s *Server) adminBillingTotals(ctx context.Context) (map[string]any, error) {
var totals struct {
Users int64
ActiveUsers int64
DisabledUsers int64
PendingUsers int64
BalanceCredits int64
ReservedCredits int64
AvailableCredits int64
UsageCredits int64
UsageQuantity int64
UsageRecords int64
ActiveReservations int64
}
err := s.db.QueryRowContext(ctx, `
SELECT
(SELECT COUNT(*) FROM hwlab_users),
(SELECT COUNT(*) FROM hwlab_users WHERE status = 'active'),
(SELECT COUNT(*) FROM hwlab_users WHERE status = 'disabled'),
(SELECT COUNT(*) FROM hwlab_users WHERE status = 'pending'),
COALESCE((SELECT SUM(balance_credits) FROM hwlab_credit_accounts), 0),
COALESCE((SELECT SUM(reserved_credits) FROM hwlab_credit_accounts), 0),
COALESCE((SELECT SUM(balance_credits - reserved_credits) FROM hwlab_credit_accounts), 0),
COALESCE((SELECT SUM(credits) FROM hwlab_usage_records), 0),
COALESCE((SELECT SUM(quantity) FROM hwlab_usage_records), 0),
(SELECT COUNT(*) FROM hwlab_usage_records),
(SELECT COUNT(*) FROM hwlab_billing_reservations WHERE status = 'reserved' AND expires_at > now())`).Scan(&totals.Users, &totals.ActiveUsers, &totals.DisabledUsers, &totals.PendingUsers, &totals.BalanceCredits, &totals.ReservedCredits, &totals.AvailableCredits, &totals.UsageCredits, &totals.UsageQuantity, &totals.UsageRecords, &totals.ActiveReservations)
if err != nil {
return nil, err
}
return map[string]any{
"users": totals.Users,
"activeUsers": totals.ActiveUsers,
"disabledUsers": totals.DisabledUsers,
"pendingUsers": totals.PendingUsers,
"balanceCredits": totals.BalanceCredits,
"reservedCredits": totals.ReservedCredits,
"availableCredits": totals.AvailableCredits,
"usageCredits": totals.UsageCredits,
"usageQuantity": totals.UsageQuantity,
"usageRecords": totals.UsageRecords,
"activeReservations": totals.ActiveReservations,
}, nil
}
func (s *Server) apiKeysForUser(ctx context.Context, userID string) ([]apiKeySummary, error) {
rows, err := s.db.QueryContext(ctx, `SELECT id, name, key_prefix, scopes_json::text, status, created_at, last_used_at, revoked_at FROM hwlab_api_keys WHERE user_id = $1 ORDER BY created_at DESC, id DESC`, userID)
if err != nil {
return nil, err
}
defer rows.Close()
keys := []apiKeySummary{}
for rows.Next() {
key, err := scanAPIKey(rows)
if err != nil {
return nil, err
}
keys = append(keys, key)
}
return keys, rows.Err()
}
func (s *Server) apiKeyForUser(ctx context.Context, userID, keyID string) (apiKeySummary, error) {
return scanAPIKey(s.db.QueryRowContext(ctx, `SELECT id, name, key_prefix, scopes_json::text, status, created_at, last_used_at, revoked_at FROM hwlab_api_keys WHERE user_id = $1 AND id = $2`, userID, keyID))
}
type apiKeyScanner interface {
Scan(dest ...any) error
}
func scanAPIKey(scanner apiKeyScanner) (apiKeySummary, error) {
var key apiKeySummary
var scopesRaw string
var lastUsed sql.NullTime
var revoked sql.NullTime
if err := scanner.Scan(&key.ID, &key.Name, &key.Prefix, &scopesRaw, &key.Status, &key.CreatedAt, &lastUsed, &revoked); err != nil {
return key, err
}
_ = json.Unmarshal([]byte(scopesRaw), &key.Scopes)
if len(key.Scopes) == 0 {
key.Scopes = []string{"api"}
}
if lastUsed.Valid {
value := lastUsed.Time
key.LastUsedAt = &value
}
if revoked.Valid {
value := revoked.Time
key.RevokedAt = &value
}
return key, nil
}
func adminUsersQueryFromRequest(r *http.Request) adminUsersQuery {
page := queryLimit(r, "page", 1, 100000)
pageSize := queryLimit(r, "pageSize", queryLimit(r, "limit", 50, 100), 100)
status := normalizeUserStatus(r.URL.Query().Get("status"), "")
role := normalizeRole(r.URL.Query().Get("role"), "")
return adminUsersQuery{
Page: page,
PageSize: pageSize,
Search: strings.TrimSpace(r.URL.Query().Get("search")),
Status: status,
Role: role,
}
}
func (s *Server) adminUsers(ctx context.Context, query adminUsersQuery) ([]adminBillingUserSummary, int64, error) {
where, args := adminUsersWhere(query)
var total int64
countQuery := `SELECT COUNT(*) FROM hwlab_users u ` + where
if err := s.db.QueryRowContext(ctx, countQuery, args...).Scan(&total); err != nil {
return nil, 0, err
}
offset := (query.Page - 1) * query.PageSize
args = append(args, query.PageSize, offset)
limitParam := "$" + strconv.Itoa(len(args)-1)
offsetParam := "$" + strconv.Itoa(len(args))
rows, err := s.db.QueryContext(ctx, `
SELECT
u.id,
u.email,
u.username,
u.display_name,
u.status,
u.role,
u.email_verified,
u.created_at,
COALESCE(a.plan_id, 'default'),
COALESCE(a.balance_credits, 0),
COALESCE(a.reserved_credits, 0),
COALESCE(usage.total_credits, 0),
COALESCE(usage.total_quantity, 0),
COALESCE(usage.record_count, 0),
usage.last_used_at,
COALESCE(keys.active_count, 0),
COALESCE(keys.total_count, 0),
keys.last_key_used_at,
COALESCE(res.active_count, 0)
FROM hwlab_users u
LEFT JOIN hwlab_credit_accounts a ON a.user_id = u.id
LEFT JOIN (
SELECT user_id, SUM(credits) AS total_credits, SUM(quantity) AS total_quantity, COUNT(*) AS record_count, MAX(created_at) AS last_used_at
FROM hwlab_usage_records
GROUP BY user_id
) usage ON usage.user_id = u.id
LEFT JOIN (
SELECT user_id, COUNT(*) FILTER (WHERE status = 'active' AND revoked_at IS NULL) AS active_count, COUNT(*) AS total_count, MAX(last_used_at) AS last_key_used_at
FROM hwlab_api_keys
GROUP BY user_id
) keys ON keys.user_id = u.id
LEFT JOIN (
SELECT user_id, COUNT(*) AS active_count
FROM hwlab_billing_reservations
WHERE status = 'reserved' AND expires_at > now()
GROUP BY user_id
) res ON res.user_id = u.id
`+where+`
ORDER BY usage.last_used_at DESC NULLS LAST, u.created_at DESC, u.username ASC
LIMIT `+limitParam+` OFFSET `+offsetParam, args...)
if err != nil {
return nil, 0, err
}
defer rows.Close()
users, userIDs, err := scanAdminBillingUsers(rows)
if err != nil {
return nil, 0, err
}
reservations, err := s.activeReservationsForUsers(ctx, userIDs, 5)
if err != nil {
return nil, 0, err
}
for i := range users {
users[i].Reservations.Active = reservations[users[i].User.ID]
}
if err := s.enrichAdminBillingUsers(ctx, users); err != nil {
return nil, 0, err
}
return users, total, nil
}
func (s *Server) enrichAdminBillingUsers(ctx context.Context, users []adminBillingUserSummary) error {
for i := range users {
plan, err := s.planByID(ctx, s.db, users[i].PlanID)
if err != nil {
return err
}
entitlements, err := s.resourceEntitlements(ctx, s.db, users[i].PlanID, users[i].User.ID)
if err != nil {
return err
}
users[i].Plan = plan
users[i].Entitlements = entitlements
}
return nil
}
func adminUsersWhere(query adminUsersQuery) (string, []any) {
clauses := []string{}
args := []any{}
if query.Status != "" {
args = append(args, query.Status)
clauses = append(clauses, "u.status = $"+strconv.Itoa(len(args)))
}
if query.Role != "" {
args = append(args, query.Role)
clauses = append(clauses, "u.role = $"+strconv.Itoa(len(args)))
}
if query.Search != "" {
args = append(args, "%"+strings.ToLower(query.Search)+"%")
param := "$" + strconv.Itoa(len(args))
clauses = append(clauses, "(lower(u.id) LIKE "+param+" OR lower(u.email) LIKE "+param+" OR lower(u.username) LIKE "+param+" OR lower(u.display_name) LIKE "+param+")")
}
if len(clauses) == 0 {
return "", args
}
return "WHERE " + strings.Join(clauses, " AND "), args
}
func (s *Server) adminUserDetail(ctx context.Context, userID string) (adminUserDetail, error) {
users, _, err := s.adminUsers(ctx, adminUsersQuery{Page: 1, PageSize: 1, Search: userID})
if err != nil {
return adminUserDetail{}, err
}
var summary adminBillingUserSummary
found := false
for _, user := range users {
if user.User.ID == userID {
summary = user
found = true
break
}
}
if !found {
return adminUserDetail{}, sql.ErrNoRows
}
ledger, err := s.ledgerRows(ctx, userID, 50)
if err != nil {
return adminUserDetail{}, err
}
keys, err := s.apiKeysForUser(ctx, userID)
if err != nil {
return adminUserDetail{}, err
}
usageBy, err := s.usageByService(ctx, userID, 50)
if err != nil {
return adminUserDetail{}, err
}
reservations, err := s.activeReservations(ctx, userID, 50)
if err != nil {
return adminUserDetail{}, err
}
return adminUserDetail{Summary: summary, Ledger: ledger, APIKeys: keys, UsageBy: usageBy, Reservations: reservations}, nil
}
func scanAdminBillingUsers(rows *sql.Rows) ([]adminBillingUserSummary, []string, error) {
users := []adminBillingUserSummary{}
userIDs := []string{}
for rows.Next() {
var item adminBillingUserSummary
var lastUsed sql.NullTime
var lastKeyUsed sql.NullTime
if err := rows.Scan(&item.User.ID, &item.User.Email, &item.User.Username, &item.User.DisplayName, &item.User.Status, &item.User.Role, &item.User.EmailVerified, &item.User.CreatedAt, &item.PlanID, &item.Credits.Balance, &item.Credits.Reserved, &item.Usage.TotalCredits, &item.Usage.TotalQuantity, &item.Usage.RecordCount, &lastUsed, &item.APIKeys.ActiveCount, &item.APIKeys.TotalCount, &lastKeyUsed, &item.Reservations.ActiveCount); err != nil {
return nil, nil, err
}
if lastUsed.Valid {
value := lastUsed.Time
item.Usage.LastUsedAt = &value
}
if lastKeyUsed.Valid {
value := lastKeyUsed.Time
item.APIKeys.LastUsedAt = &value
}
item.Credits.Available = item.Credits.Balance - item.Credits.Reserved
userIDs = append(userIDs, item.User.ID)
users = append(users, item)
}
if err := rows.Err(); err != nil {
return nil, nil, err
}
return users, userIDs, nil
}
func normalizeRole(value, fallback string) string {
role := strings.TrimSpace(strings.ToLower(value))
if role == "" {
return fallback
}
if role == "user" || role == "admin" {
return role
}
return ""
}
func normalizeUserStatus(value, fallback string) string {
status := strings.TrimSpace(strings.ToLower(value))
if status == "" {
return fallback
}
if status == "active" || status == "disabled" || status == "pending" {
return status
}
return ""
}
func queryLimit(r *http.Request, name string, fallback, max int) int {
value := strings.TrimSpace(r.URL.Query().Get(name))
if value == "" {
return fallback
}
parsed, err := strconv.Atoi(value)
if err != nil || parsed <= 0 {
return fallback
}
if parsed > max {
return max
}
return parsed
}
func (s *Server) estimateCredits(explicit, tokens int64) int64 {
if explicit > 0 {
return explicit
}
if tokens <= 0 {
return 1
}
per := s.config.CreditPerThousandTok
if per <= 0 {
per = 1
}
return int64(math.Ceil(float64(tokens*per) / 1000.0))
}
func decodeJSON(w http.ResponseWriter, r *http.Request, target any) bool {
defer r.Body.Close()
decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20))
decoder.DisallowUnknownFields()
if err := decoder.Decode(target); err != nil {
writeAPIError(w, http.StatusBadRequest, "invalid_json", "request body must be valid JSON")
return false
}
return true
}
func writeAPIError(w http.ResponseWriter, status int, code, message string) {
writeJSON(w, status, map[string]any{"error": apiError{Code: code, Message: message}})
}
func writeJSON(w http.ResponseWriter, status int, payload any) {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(payload)
}
func bearerToken(r *http.Request) string {
value := strings.TrimSpace(r.Header.Get("Authorization"))
if strings.HasPrefix(strings.ToLower(value), "bearer ") {
return strings.TrimSpace(value[7:])
}
return ""
}
func normalizeUsername(value string) string {
value = strings.ToLower(strings.TrimSpace(value))
value = strings.Map(func(r rune) rune {
if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') || r == '-' || r == '_' {
return r
}
return -1
}, value)
if len(value) < 3 || len(value) > 64 {
return ""
}
return value
}
func hashPassword(password string) (string, error) {
salt, err := randomBytes(16)
if err != nil {
return "", err
}
iterations := 210000
digest := pbkdf2SHA256([]byte(password), salt, iterations, 32)
return fmt.Sprintf("pbkdf2_sha256$%d$%s$%s", iterations, b64(salt), b64(digest)), nil
}
func verifyPassword(encoded, password string) bool {
parts := strings.Split(encoded, "$")
if len(parts) != 4 || parts[0] != "pbkdf2_sha256" {
return false
}
iterations, err := strconv.Atoi(parts[1])
if err != nil || iterations <= 0 {
return false
}
salt, err := base64.RawURLEncoding.DecodeString(parts[2])
if err != nil {
return false
}
expected, err := base64.RawURLEncoding.DecodeString(parts[3])
if err != nil {
return false
}
actual := pbkdf2SHA256([]byte(password), salt, iterations, len(expected))
return subtle.ConstantTimeCompare(actual, expected) == 1
}
func pbkdf2SHA256(password, salt []byte, iterations, keyLen int) []byte {
hLen := sha256.Size
numBlocks := (keyLen + hLen - 1) / hLen
var output []byte
for block := 1; block <= numBlocks; block++ {
u := pbkdf2Block(password, salt, block)
t := append([]byte(nil), u...)
for i := 1; i < iterations; i++ {
u = hmacSHA256(password, u)
for j := range t {
t[j] ^= u[j]
}
}
output = append(output, t...)
}
return output[:keyLen]
}
func pbkdf2Block(password, salt []byte, block int) []byte {
buf := make([]byte, len(salt)+4)
copy(buf, salt)
binary.BigEndian.PutUint32(buf[len(salt):], uint32(block))
return hmacSHA256(password, buf)
}
func hmacSHA256(key, data []byte) []byte {
h := hmac.New(sha256.New, key)
_, _ = h.Write(data)
return h.Sum(nil)
}
func tokenHash(token string) string {
sum := sha256.Sum256([]byte(token))
return hex.EncodeToString(sum[:])
}
func newID(prefix string) string {
return prefix + "_" + randomToken(16)
}
func randomToken(size int) string {
buf, err := randomBytes(size)
if err != nil {
panic(err)
}
return b64(buf)
}
func randomBytes(size int) ([]byte, error) {
buf := make([]byte, size)
_, err := rand.Read(buf)
return buf, err
}
func b64(value []byte) string {
return base64.RawURLEncoding.EncodeToString(value)
}
func jsonObject(value map[string]any) string {
if value == nil {
return "{}"
}
encoded, err := json.Marshal(value)
if err != nil {
return "{}"
}
return string(encoded)
}
func parseInt64(value string, fallback int64) int64 {
if strings.TrimSpace(value) == "" {
return fallback
}
parsed, err := strconv.ParseInt(value, 10, 64)
if err != nil {
return fallback
}
return parsed
}
func parseDurationHours(value string, fallbackHours int64) time.Duration {
return time.Duration(parseInt64(value, fallbackHours)) * time.Hour
}
func parseDurationMinutes(value string, fallbackMinutes int64) time.Duration {
return time.Duration(parseInt64(value, fallbackMinutes)) * time.Minute
}
func first(values ...string) string {
for _, value := range values {
if strings.TrimSpace(value) != "" {
return strings.TrimSpace(value)
}
}
return ""
}
func nonNegative(value int64) int64 {
if value < 0 {
return 0
}
return value
}
func min(a, b int) int {
if a < b {
return a
}
return b
}
func isDuplicate(err error) bool {
return err != nil && strings.Contains(strings.ToLower(err.Error()), "duplicate")
}
Reference in New Issue View Git Blame Copy Permalink