Files

2985 lines
110 KiB
Go

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
MigrationTimeout time.Duration
BootstrapTimeout time.Duration
StateAuthority string
RuntimeNamespace string
ExternalDatabaseLabel string
BootstrapAdminID string
BootstrapAdminEmail string
BootstrapAdminUsername string
BootstrapAdminDisplayName string
BootstrapAdminPasswordHash string
BootstrapAdminPassword 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",
MigrationTimeout: parseDurationSeconds(values["HWLAB_USER_BILLING_MIGRATION_TIMEOUT_SECONDS"], 20),
BootstrapTimeout: parseDurationSeconds(values["HWLAB_USER_BILLING_BOOTSTRAP_TIMEOUT_SECONDS"], 10),
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"),
BootstrapAdminID: first(values["HWLAB_BOOTSTRAP_ADMIN_ID"], "usr_bootstrap_admin"),
BootstrapAdminEmail: values["HWLAB_BOOTSTRAP_ADMIN_EMAIL"],
BootstrapAdminUsername: first(values["HWLAB_BOOTSTRAP_ADMIN_USERNAME"], "admin"),
BootstrapAdminDisplayName: first(values["HWLAB_BOOTSTRAP_ADMIN_DISPLAY_NAME"], "HWLAB Admin"),
BootstrapAdminPasswordHash: values["HWLAB_BOOTSTRAP_ADMIN_PASSWORD_HASH"],
BootstrapAdminPassword: values["HWLAB_BOOTSTRAP_ADMIN_PASSWORD"],
}
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(envInt("HWLAB_USER_BILLING_DB_POOL_MAX", 4))
db.SetMaxIdleConns(4)
db.SetConnMaxLifetime(30 * time.Minute)
server.db = db
if cfg.MigrateOnStart {
migrateCtx, cancel := context.WithTimeout(ctx, durationOrDefault(cfg.MigrationTimeout, 20*time.Second))
defer cancel()
if err := server.migrate(migrateCtx); err != nil {
return nil, fmt.Errorf("migrate user billing schema: %w", err)
}
}
bootstrapCtx, cancel := context.WithTimeout(ctx, durationOrDefault(cfg.BootstrapTimeout, 10*time.Second))
defer cancel()
if err := server.ensureBootstrapAdmin(bootstrapCtx); err != nil {
return nil, fmt.Errorf("ensure bootstrap admin: %w", err)
}
}
server.routes()
return server, nil
}
func (s *Server) ensureBootstrapAdmin(ctx context.Context) error {
if s.db == nil {
return nil
}
username := normalizeUsername(s.config.BootstrapAdminUsername)
if username == "" {
return nil
}
passwordHash := strings.TrimSpace(s.config.BootstrapAdminPasswordHash)
if passwordHash == "" && strings.TrimSpace(s.config.BootstrapAdminPassword) != "" {
hash, err := hashPassword(s.config.BootstrapAdminPassword)
if err != nil {
return err
}
passwordHash = hash
}
if passwordHash == "" {
return nil
}
adminID := strings.TrimSpace(first(s.config.BootstrapAdminID, "usr_bootstrap_admin"))
email := strings.ToLower(strings.TrimSpace(s.config.BootstrapAdminEmail))
if email == "" {
email = username + "@bootstrap.hwlab.local"
}
displayName := strings.TrimSpace(first(s.config.BootstrapAdminDisplayName, "HWLAB Admin"))
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
if err != nil {
return err
}
defer tx.Rollback()
result, err := tx.ExecContext(ctx, `UPDATE hwlab_users SET email = $3, display_name = $4, password_hash = $5, status = 'active', role = 'admin', email_verified = true, updated_at = now() WHERE id = $1 OR username = $2`, adminID, username, email, displayName, passwordHash)
if err != nil {
return err
}
rows, err := result.RowsAffected()
if err != nil {
return err
}
if rows == 0 {
_, err = tx.ExecContext(ctx, `INSERT INTO hwlab_users (id, email, username, display_name, password_hash, status, role, email_verified) VALUES ($1, $2, $3, $4, $5, 'active', 'admin', true)`, adminID, email, username, displayName, passwordHash)
if err != nil {
return err
}
}
_, err = tx.ExecContext(ctx, `INSERT INTO hwlab_credit_accounts (user_id, balance_credits, reserved_credits, plan_id) VALUES ($1, $2, 0, 'default') ON CONFLICT (user_id) DO NOTHING`, adminID, nonNegative(s.config.InitialCredits))
if err != nil {
return err
}
return tx.Commit()
}
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 {
if strings.HasPrefix(encoded, "sha256:") {
parts := strings.Split(encoded, ":")
if len(parts) != 3 || parts[1] == "" || parts[2] == "" {
return false
}
digest := sha256.Sum256([]byte(parts[1] + ":" + password))
actual := hex.EncodeToString(digest[:])
return subtle.ConstantTimeCompare([]byte(actual), []byte(parts[2])) == 1
}
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 envInt(name string, fallback int) int {
value := strings.TrimSpace(os.Getenv(name))
if value == "" {
return fallback
}
parsed, err := strconv.Atoi(value)
if err != nil || parsed <= 0 {
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 parseDurationSeconds(value string, fallbackSeconds int64) time.Duration {
return time.Duration(parseInt64(value, fallbackSeconds)) * time.Second
}
func durationOrDefault(value time.Duration, fallback time.Duration) time.Duration {
if value <= 0 {
return fallback
}
return value
}
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")
}