// Stat returns pool statistics (connection counts, acquire waits) for the metrics collector.
func (d *DB) Stat() *pgxpool.Stat { return d.pool.Stat() }

// Healthy reports whether the blob store is reachable (a cheap BucketExists round-trip). Used by
// the dependency probe backing /readyz.
func (s *MinIOStore) Healthy(ctx context.Context) error {
    if _, err := s.client.BucketExists(ctx, s.bucket); err != nil {
        return fmt.Errorf("blob: bucket check: %w", err)
    }
    return nil
}

    depUp         *prometheus.GaugeVec     // label: dep — 1 up / 0 down (set by the DependencyProbe)
    schedRuns     *prometheus.CounterVec   // labels: task, result (ok|error)
    schedDuration *prometheus.HistogramVec // label: task
    rateRejected  *prometheus.CounterVec   // label: limiter (user|auth_ip|tsa|public_sign)
    // rateRejectedN mirrors rateRejected as plain atomics so the admin rate-limits view can read
    // live counts back without a Prometheus gather.
    rateRejectedN map[string]*atomic.Int64

        depUp: f.NewGaugeVec(prometheus.GaugeOpts{
            Name: "obscura_dependency_up",
            Help: "External dependency health (1 = reachable, 0 = down), labelled by dependency.",
        }, []string{"dep"}),
        schedRuns: f.NewCounterVec(prometheus.CounterOpts{
            Name: "obscura_scheduler_runs_total",
            Help: "Scheduler task executions, labelled by task and result (ok|error).",
        }, []string{"task", "result"}),
        schedDuration: f.NewHistogramVec(prometheus.HistogramOpts{
            Name:    "obscura_scheduler_run_duration_seconds",
            Help:    "Scheduler task run duration in seconds, labelled by task.",
            Buckets: prometheus.DefBuckets,
        }, []string{"task"}),
        rateRejected: f.NewCounterVec(prometheus.CounterOpts{
            Name: "obscura_rate_limit_rejected_total",
            Help: "Requests rejected by a rate limiter, labelled by limiter.",
        }, []string{"limiter"}),
        rateRejectedN: map[string]*atomic.Int64{
            "user": {}, "auth_ip": {}, "tsa": {}, "public_sign": {},
        },

    for _, name := range []string{"user", "auth_ip", "tsa", "public_sign"} {
        m.rateRejected.WithLabelValues(name)
    }

// SetDependencyUp records a dependency's health (1 up / 0 down); called by the DependencyProbe.
func (m *Metrics) SetDependencyUp(dep string, up bool) {
    v := 0.0
    if up {
        v = 1
    }
    m.depUp.WithLabelValues(dep).Set(v)
}

// RecordSchedulerRun records one scheduler task execution (result ok|error) + its duration.
func (m *Metrics) RecordSchedulerRun(task string, d time.Duration, err error) {
    result := "ok"
    if err != nil {
        result = "error"
    }
    m.schedRuns.WithLabelValues(task, result).Inc()
    m.schedDuration.WithLabelValues(task).Observe(d.Seconds())
}

// RecordRateLimitRejected bumps the rejection counter for a limiter (user|auth_ip|tsa|public_sign).
func (m *Metrics) RecordRateLimitRejected(limiter string) {
    m.rateRejected.WithLabelValues(limiter).Inc()
    if n := m.rateRejectedN[limiter]; n != nil {
        n.Add(1)
    }
}

// RateLimitRejections returns the current rejection counts per limiter (for the admin view).
func (m *Metrics) RateLimitRejections() map[string]int64 {
    out := make(map[string]int64, len(m.rateRejectedN))
    for k, v := range m.rateRejectedN {
        out[k] = v.Load()
    }
    return out
}

// Register adds a custom collector (the DB-pool + AI-tokens scrape collectors) to the private
// registry so they appear on /metrics.
func (m *Metrics) Register(c prometheus.Collector) error { return m.reg.Register(c) }

package httpapi

import (
    "context"
    "time"

    "github.com/jackc/pgx/v5/pgxpool"
    "github.com/prometheus/client_golang/prometheus"

    "github.com/Virtue-Digital-Indonesia/obscura/internal/platform/db"
)

// ---------------------------------------------------------------------------
// DB pool collector — reads pool.Stat() at scrape time (no background goroutine).
// ---------------------------------------------------------------------------

var (
    dbPoolTotal        = prometheus.NewDesc("obscura_db_pool_total_conns", "Total connections currently in the pgx pool.", nil, nil)
    dbPoolAcquired     = prometheus.NewDesc("obscura_db_pool_acquired_conns", "Currently acquired (in-use) connections.", nil, nil)
    dbPoolIdle         = prometheus.NewDesc("obscura_db_pool_idle_conns", "Currently idle connections.", nil, nil)
    dbPoolMax          = prometheus.NewDesc("obscura_db_pool_max_conns", "Configured maximum pool size.", nil, nil)
    dbPoolAcquireTotal = prometheus.NewDesc("obscura_db_pool_acquire_total", "Cumulative count of successful connection acquires.", nil, nil)
    dbPoolAcquireWait  = prometheus.NewDesc("obscura_db_pool_acquire_wait_seconds_total", "Cumulative time spent blocked waiting for a connection.", nil, nil)
)

// DBPoolCollector exports pgxpool statistics. It reads Stat() on each scrape, so there is no
// background work and the numbers are always current.
type DBPoolCollector struct{ stat func() *pgxpool.Stat }

// NewDBPoolCollector builds the collector over a pool Stat() accessor (pass database.Stat).
func NewDBPoolCollector(stat func() *pgxpool.Stat) *DBPoolCollector { return &DBPoolCollector{stat: stat} }

// Describe implements prometheus.Collector.
func (c *DBPoolCollector) Describe(ch chan<- *prometheus.Desc) {
    ch <- dbPoolTotal
    ch <- dbPoolAcquired
    ch <- dbPoolIdle
    ch <- dbPoolMax
    ch <- dbPoolAcquireTotal
    ch <- dbPoolAcquireWait
}

// Collect implements prometheus.Collector.
func (c *DBPoolCollector) Collect(ch chan<- prometheus.Metric) {
    s := c.stat()
    if s == nil {
        return
    }
    ch <- prometheus.MustNewConstMetric(dbPoolTotal, prometheus.GaugeValue, float64(s.TotalConns()))
    ch <- prometheus.MustNewConstMetric(dbPoolAcquired, prometheus.GaugeValue, float64(s.AcquiredConns()))
    ch <- prometheus.MustNewConstMetric(dbPoolIdle, prometheus.GaugeValue, float64(s.IdleConns()))
    ch <- prometheus.MustNewConstMetric(dbPoolMax, prometheus.GaugeValue, float64(s.MaxConns()))
    ch <- prometheus.MustNewConstMetric(dbPoolAcquireTotal, prometheus.CounterValue, float64(s.AcquireCount()))
    ch <- prometheus.MustNewConstMetric(dbPoolAcquireWait, prometheus.CounterValue, s.AcquireDuration().Seconds())
}

// ---------------------------------------------------------------------------
// AI-tokens collector — sums ai_usage per feature at scrape time.
// ---------------------------------------------------------------------------

// AITokenRow is one feature's cumulative token totals.
type AITokenRow struct {
    Feature   string
    TokensIn  int64
    TokensOut int64
}

var aiTokensDesc = prometheus.NewDesc(
    "obscura_ai_tokens_total",
    "Cumulative AI tokens by feature and direction (in|out), summed from ai_usage.",
    []string{"feature", "direction"}, nil,
)

// AITokensCollector exports AI token totals. The DB read is bounded and best-effort: a query error
// omits the ai series for that scrape rather than failing the whole /metrics response.
type AITokensCollector struct {
    query func(ctx context.Context) ([]AITokenRow, error)
}

// NewAITokensCollector builds the collector over an ai_usage aggregate reader (use AITokenQuery).
func NewAITokensCollector(q func(ctx context.Context) ([]AITokenRow, error)) *AITokensCollector {
    return &AITokensCollector{query: q}
}

// Describe implements prometheus.Collector.
func (c *AITokensCollector) Describe(ch chan<- *prometheus.Desc) { ch <- aiTokensDesc }

// Collect implements prometheus.Collector.
func (c *AITokensCollector) Collect(ch chan<- prometheus.Metric) {
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()
    rows, err := c.query(ctx)
    if err != nil {
        return
    }
    for _, r := range rows {
        ch <- prometheus.MustNewConstMetric(aiTokensDesc, prometheus.CounterValue, float64(r.TokensIn), r.Feature, "in")
        ch <- prometheus.MustNewConstMetric(aiTokensDesc, prometheus.CounterValue, float64(r.TokensOut), r.Feature, "out")
    }
}

// AITokenQuery builds the ai_usage aggregate reader over the DB (wired in wire.go).
func AITokenQuery(database *db.DB) func(ctx context.Context) ([]AITokenRow, error) {
    return func(ctx context.Context) ([]AITokenRow, error) {
        rows, err := database.Exec(ctx).Query(ctx,
            `SELECT feature, COALESCE(sum(tokens_in), 0), COALESCE(sum(tokens_out), 0) FROM ai_usage GROUP BY feature`)
        if err != nil {
            return nil, err
        }
        defer rows.Close()
        var out []AITokenRow
        for rows.Next() {
            var r AITokenRow
            if err := rows.Scan(&r.Feature, &r.TokensIn, &r.TokensOut); err != nil {
                return nil, err
            }
            out = append(out, r)
        }
        return out, rows.Err()
    }
}

    onRun func(task string, d time.Duration, err error) // optional metrics observer (nil = no-op)

// SetObserver registers a callback invoked after each task run (task name, duration, handler
// error). Used to feed scheduler metrics. nil disables it.
func (s *Service) SetObserver(fn func(task string, d time.Duration, err error)) { s.onRun = fn }

        start := s.clock()
        herr := r.handler(ctx)
        if s.onRun != nil {
            s.onRun(name, s.clock().Sub(start), herr)
        }
        if herr != nil && err == nil {
            err = fmt.Errorf("scheduler task %q: %w", name, herr)
        }
        ran++

func (s *Server) rateLimit(rl *rateLimiter, name string, keyFn func(*http.Request) (string, bool), max int) func(http.Handler) http.Handler {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            key, limited := keyFn(r)
            if limited && !rl.allow(key, max, time.Now()) {
                s.metrics.RecordRateLimitRejected(name)
                w.Header().Set("Retry-After", strconv.Itoa(int(rl.window.Seconds())))
                writeProblem(w, &kernel.Error{Kind: kernel.ErrRateLimited, Code: "api.rate_limited", Message: "too many requests — please slow down"})
                return
            }
            next.ServeHTTP(w, r)
        })
    }
}

        authRL := s.rateLimit(authLimiter, "auth_ip", ipKeyFn, apiRateAuthRouteMax)
        tsaRL := s.rateLimit(tsaLimiter, "tsa", tsaKeyFn, apiRateTSAMax)
        r.Use(s.rateLimit(userLimiter, "user", principalKeyFn, apiRateUserMax))

        s.metrics.RecordRateLimitRejected("public_sign")

    // Phase-1 observability plumbing: pool + AI-token scrape collectors and the scheduler run
    // observer, all on the same private registry served at /metrics.
    if err := metrics.Register(httpapi.NewDBPoolCollector(database.Stat)); err != nil {
        return fmt.Errorf("register db pool collector: %w", err)
    }
    if err := metrics.Register(httpapi.NewAITokensCollector(httpapi.AITokenQuery(database))); err != nil {
        return fmt.Errorf("register ai tokens collector: %w", err)
    }
    schedulerSvc.SetObserver(metrics.RecordSchedulerRun)

git add go/internal/platform/db/db.go go/internal/platform/blob/minio.go go/internal/httpapi/metrics.go go/internal/httpapi/metrics_collectors.go go/internal/scheduler/app/service.go go/internal/httpapi/ratelimit.go go/internal/httpapi/server.go go/internal/httpapi/handlers_public_sign.go go/cmd/obscura-server/wire.go
git commit -m "feat(metrics): db-pool + dependency_up + scheduler + rate-limit-rejected + ai-tokens series"

package httpapi

import (
    "context"
    "encoding/json"
    "fmt"
    "net/http"
    "sync"
    "time"
)

// DependencyProbe checks each configured external dependency with a short per-dep timeout and
// (a) backs the deep /readyz response and (b) refreshes the obscura_dependency_up gauges. It is
// constructed in wire.go with closures that reach the real dependencies (DB, blob store, sidecars,
// Gotenberg, protection engine). ONLY dependencies actually configured for THIS deployment are
// added, so e.g. a build with EXTRACT_PROVIDER=none has no "extract" check and /readyz never 503s
// on a component the deployment does not run.
type DependencyProbe struct {
    metrics *Metrics
    timeout time.Duration
    checks  []depCheck
}

type depCheck struct {
    name  string
    check func(ctx context.Context) error
}

// NewDependencyProbe builds an empty probe. perDepTimeout bounds each individual check (~2s) so a
// hung dependency cannot stall the probe; checks run concurrently, so the whole probe is ~one
// timeout worst-case.
func NewDependencyProbe(m *Metrics, perDepTimeout time.Duration) *DependencyProbe {
    return &DependencyProbe{metrics: m, timeout: perDepTimeout}
}

// Add registers a dependency check (chainable). Order is preserved in the deps map.
func (p *DependencyProbe) Add(name string, check func(ctx context.Context) error) *DependencyProbe {
    p.checks = append(p.checks, depCheck{name: name, check: check})
    return p
}

// run executes every check concurrently under its own timeout, updates the gauges, and returns the
// aggregate readiness + a per-dep status map ("ok" or the error text).
func (p *DependencyProbe) run(ctx context.Context) (ready bool, deps map[string]string) {
    deps = make(map[string]string, len(p.checks))
    errs := make([]error, len(p.checks))
    var wg sync.WaitGroup
    for i, c := range p.checks {
        wg.Add(1)
        go func(i int, c depCheck) {
            defer wg.Done()
            cctx, cancel := context.WithTimeout(ctx, p.timeout)
            defer cancel()
            errs[i] = c.check(cctx)
        }(i, c)
    }
    wg.Wait()
    ready = true
    for i, c := range p.checks {
        up := errs[i] == nil
        p.metrics.SetDependencyUp(c.name, up)
        if up {
            deps[c.name] = "ok"
        } else {
            ready = false
            deps[c.name] = errs[i].Error()
        }
    }
    return ready, deps
}

// Refresh runs the probe and updates gauges only (background tick — keeps dependency_up fresh
// without depending on a /readyz hit or a scrape).
func (p *DependencyProbe) Refresh(ctx context.Context) { p.run(ctx) }

// Readyz is the deep readiness handler: 200 {ready:true, deps:{...}} when every configured
// dependency is up, else 503 {ready:false, deps:{dep:err}}. It is mounted on the OUTER mux
// (wire.go), so it is not wrapped by the chi middleware — scrapes and probes never pollute the
// RED metrics. Bounded by the per-dep timeouts so a hung dependency can't stall a load balancer.
func (p *DependencyProbe) Readyz(w http.ResponseWriter, r *http.Request) {
    ready, deps := p.run(r.Context())
    status := http.StatusOK
    if !ready {
        status = http.StatusServiceUnavailable
    }
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(status)
    _ = json.NewEncoder(w).Encode(map[string]any{"ready": ready, "deps": deps})
}

// HTTPHealthCheck returns a dependency check that GETs url with the given client and treats any
// response below 500 as healthy (the endpoint answered), a transport error or a 5xx as down.
func HTTPHealthCheck(client *http.Client, url string) func(ctx context.Context) error {
    return func(ctx context.Context) error {
        req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
        if err != nil {
            return err
        }
        resp, err := client.Do(req)
        if err != nil {
            return err
        }
        defer resp.Body.Close()
        if resp.StatusCode >= 500 {
            return fmt.Errorf("status %d", resp.StatusCode)
        }
        return nil
    }
}

    // Deep dependency probe backing /readyz + the obscura_dependency_up gauges. Only deps actually
    // configured for THIS deployment are added, so /readyz never 503s on a component the deployment
    // does not run. Each check is bounded to 2s; the probe runs them concurrently.
    depProbe := httpapi.NewDependencyProbe(metrics, 2*time.Second)
    healthClient := &http.Client{Timeout: 2 * time.Second}
    depProbe.Add("postgres", database.Ping)
    depProbe.Add("minio", blobStore.Healthy)
    depProbe.Add("engine", engine.Ready)
    depProbe.Add("gotenberg", httpapi.HTTPHealthCheck(healthClient, strings.TrimRight(cfg.GotenbergURL, "/")+"/health"))
    if cfg.Extract.Provider == "sidecar" {
        depProbe.Add("extract", httpapi.HTTPHealthCheck(healthClient, strings.TrimRight(cfg.Extract.SidecarURL, "/")+"/healthz"))
    }
    if cfg.Embed.Provider == "sidecar" {
        depProbe.Add("embed", httpapi.HTTPHealthCheck(healthClient, strings.TrimRight(cfg.Embed.SidecarURL, "/")+"/healthz"))
    }
    // AI is probed only when it is a real HTTP backend WITH a configured base URL (self-hosted
    // OpenAI-compatible / custom endpoint). A managed provider with no base URL, or the mock, is
    // in-process/undialable-by-default and is left unprobed — no new egress on the air-gapped path.
    if (cfg.AI.Provider == "openai" || cfg.AI.Provider == "anthropic") && cfg.AI.BaseURL != "" {
        depProbe.Add("ai", httpapi.HTTPHealthCheck(healthClient, strings.TrimRight(cfg.AI.BaseURL, "/")))
    }
    // Keep the dependency_up gauges fresh without depending on a /readyz hit or a scrape.
    go loop(ctx, logger, "dependency-probe", 15*time.Second, func(ctx context.Context) error {
        depProbe.Refresh(ctx)
        return nil
    })

    mux.HandleFunc("GET /readyz", depProbe.Readyz)

git add go/internal/httpapi/dependency.go go/cmd/obscura-server/wire.go
git commit -m "feat(readyz): dependency-aware deep /readyz probe + dependency_up gauge tick"

go get go.opentelemetry.io/otel@latest \
       go.opentelemetry.io/otel/sdk@latest \
       go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp@latest
go mod tidy

// OTELConfig configures opt-in OpenTelemetry tracing. Empty Endpoint = tracing OFF (the default,
// air-gapped-safe): no spans created or exported, nothing leaves the box. When set, spans export
// to the OTLP/HTTP collector at Endpoint.
type OTELConfig struct {
    Endpoint    string `env:"OTEL_EXPORTER_OTLP_ENDPOINT"`
    Headers     string `env:"OTEL_EXPORTER_OTLP_HEADERS"`
    ServiceName string `env:"OTEL_SERVICE_NAME" envDefault:"obscura"`
    Sampler     string `env:"OTEL_TRACES_SAMPLER"`
    SamplerArg  string `env:"OTEL_TRACES_SAMPLER_ARG"`
}

func (o OTELConfig) validate() error {
    if strings.TrimSpace(o.Endpoint) == "" {
        return nil // tracing off — ignore the rest, never fail closed on an unset endpoint
    }
    if o.SamplerArg != "" {
        if f, err := strconv.ParseFloat(o.SamplerArg, 64); err != nil || f < 0 || f > 1 {
            return fmt.Errorf("config: invalid OTEL_TRACES_SAMPLER_ARG %q (want a ratio in [0,1])", o.SamplerArg)
        }
    }
    switch o.Sampler {
    case "", "always_on", "always_off", "traceidratio",
        "parentbased_always_on", "parentbased_always_off", "parentbased_traceidratio":
    default:
        return fmt.Errorf("config: invalid OTEL_TRACES_SAMPLER %q", o.Sampler)
    }
    if o.Headers != "" && !strings.Contains(o.Headers, "=") {
        return fmt.Errorf("config: invalid OTEL_EXPORTER_OTLP_HEADERS (want k=v[,k2=v2])")
    }
    return nil
}

// Package obs wires opt-in OpenTelemetry tracing. It is a no-op unless an OTLP endpoint is
// configured, preserving Obscura's air-gapped-by-default posture: with no endpoint the global OTel
// TracerProvider stays the built-in no-op (zero span allocation, nothing leaves the box).
package obs

import (
    "context"
    "fmt"
    "strconv"
    "strings"

    "github.com/jackc/pgx/v5"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
    "go.opentelemetry.io/otel/trace"
)

// Config is the parsed OTEL_* configuration (from platform/config). Endpoint empty = off.
type Config struct {
    Endpoint    string
    Headers     string
    ServiceName string
    Sampler     string
    SamplerArg  string
}

func noopShutdown(context.Context) error { return nil }

// SetupTracing installs a global TracerProvider + W3C propagator when cfg.Endpoint is set, and
// returns a shutdown func that flushes pending spans on exit. When cfg.Endpoint is empty it
// installs nothing and returns a no-op shutdown. A setup error with an endpoint set is returned to
// the caller (which logs loudly and continues with the no-op provider) — tracing never blocks boot.
func SetupTracing(ctx context.Context, cfg Config) (func(context.Context) error, error) {
    if strings.TrimSpace(cfg.Endpoint) == "" {
        return noopShutdown, nil
    }
    var opts []otlptracehttp.Option
    if strings.Contains(cfg.Endpoint, "://") {
        opts = append(opts, otlptracehttp.WithEndpointURL(cfg.Endpoint))
        if strings.HasPrefix(cfg.Endpoint, "http://") {
            opts = append(opts, otlptracehttp.WithInsecure())
        }
    } else {
        opts = append(opts, otlptracehttp.WithEndpoint(cfg.Endpoint), otlptracehttp.WithInsecure())
    }
    if h := parseHeaders(cfg.Headers); len(h) > 0 {
        opts = append(opts, otlptracehttp.WithHeaders(h))
    }
    exp, err := otlptracehttp.New(ctx, opts...)
    if err != nil {
        return noopShutdown, fmt.Errorf("otel exporter: %w", err)
    }
    name := cfg.ServiceName
    if name == "" {
        name = "obscura"
    }
    res, err := resource.New(ctx, resource.WithAttributes(semconv.ServiceName(name)))
    if err != nil {
        return noopShutdown, fmt.Errorf("otel resource: %w", err)
    }
    tp := sdktrace.NewTracerProvider(
        sdktrace.WithBatcher(exp),
        sdktrace.WithResource(res),
        sdktrace.WithSampler(sampler(cfg.Sampler, cfg.SamplerArg)),
    )
    otel.SetTracerProvider(tp)
    otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{}))
    return tp.Shutdown, nil
}

// sampler maps OTEL_TRACES_SAMPLER[/_ARG] to an sdktrace.Sampler. Default: parent-based, always
// sample the root (the OTel spec default).
func sampler(name, arg string) sdktrace.Sampler {
    ratio := 1.0
    if arg != "" {
        if f, err := strconv.ParseFloat(arg, 64); err == nil {
            ratio = f
        }
    }
    switch name {
    case "always_off":
        return sdktrace.NeverSample()
    case "always_on":
        return sdktrace.AlwaysSample()
    case "traceidratio":
        return sdktrace.TraceIDRatioBased(ratio)
    case "parentbased_always_off":
        return sdktrace.ParentBased(sdktrace.NeverSample())
    case "parentbased_traceidratio":
        return sdktrace.ParentBased(sdktrace.TraceIDRatioBased(ratio))
    default: // "" or parentbased_always_on
        return sdktrace.ParentBased(sdktrace.AlwaysSample())
    }
}

func parseHeaders(s string) map[string]string {
    out := map[string]string{}
    for _, kv := range strings.Split(s, ",") {
        kv = strings.TrimSpace(kv)
        if kv == "" {
            continue
        }
        if i := strings.IndexByte(kv, '='); i > 0 {
            out[strings.TrimSpace(kv[:i])] = strings.TrimSpace(kv[i+1:])
        }
    }
    return out
}

// --- pgx query tracer (DB spans) ---

type pgxSpanKey struct{}

type pgxTracer struct{ tracer trace.Tracer }

// NewPgxTracer returns a pgx.QueryTracer that emits one client span per query. Attach it to the
// pool ONLY when tracing is enabled (a nil tracer = zero overhead on the default air-gapped path).
func NewPgxTracer() pgx.QueryTracer { return &pgxTracer{tracer: otel.Tracer("obscura/pgx")} }

func (t *pgxTracer) TraceQueryStart(ctx context.Context, _ *pgx.Conn, data pgx.TraceQueryStartData) context.Context {
    ctx, span := t.tracer.Start(ctx, "pg.query", trace.WithSpanKind(trace.SpanKindClient))
    stmt := data.SQL
    if len(stmt) > 512 {
        stmt = stmt[:512]
    }
    span.SetAttributes(attribute.String("db.system", "postgresql"), attribute.String("db.statement", stmt))
    return context.WithValue(ctx, pgxSpanKey{}, span)
}

func (t *pgxTracer) TraceQueryEnd(ctx context.Context, _ *pgx.Conn, data pgx.TraceQueryEndData) {
    span, ok := ctx.Value(pgxSpanKey{}).(trace.Span)
    if !ok {
        return
    }
    if data.Err != nil {
        span.RecordError(data.Err)
    }
    span.End()
}

// Option configures Open.
type Option func(*pgxpool.Config)

// WithQueryTracer attaches a pgx QueryTracer to every pooled connection (DB spans). A nil tracer is
// a no-op, so the default (tracing-off) path pays nothing.
func WithQueryTracer(t pgx.QueryTracer) Option {
    return func(c *pgxpool.Config) {
        if t != nil {
            c.ConnConfig.Tracer = t
        }
    }
}

// Open connects and verifies the pool. Options (e.g. WithQueryTracer) tune the pool config before
// connect. The variadic signature keeps existing db.Open(ctx, dsn) callers (incl. tests) valid.
func Open(ctx context.Context, dsn string, opts ...Option) (*DB, error) {
    poolCfg, err := pgxpool.ParseConfig(dsn)
    if err != nil {
        return nil, fmt.Errorf("db parse config: %w", err)
    }
    for _, o := range opts {
        o(poolCfg)
    }
    pool, err := pgxpool.NewWithConfig(ctx, poolCfg)
    if err != nil {
        return nil, fmt.Errorf("db open: %w", err)
    }
    if err := pool.Ping(ctx); err != nil {
        pool.Close()
        return nil, fmt.Errorf("db ping: %w", err)
    }
    return &DB{pool: pool}, nil
}

    // Opt-in OpenTelemetry tracing. OFF by default (no OTEL endpoint) → a no-op provider: nothing
    // leaves the box. A setup error with an endpoint set is logged loudly; we continue with the
    // no-op provider so a misconfigured collector never blocks boot.
    traceShutdown, err := obs.SetupTracing(ctx, obs.Config{
        Endpoint:    cfg.OTEL.Endpoint,
        Headers:     cfg.OTEL.Headers,
        ServiceName: cfg.OTEL.ServiceName,
        Sampler:     cfg.OTEL.Sampler,
        SamplerArg:  cfg.OTEL.SamplerArg,
    })
    if err != nil {
        logger.Error("tracing setup failed; continuing without tracing", "err", err)
    }
    defer func() {
        sctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        defer cancel()
        _ = traceShutdown(sctx)
    }()

    var queryTracer pgx.QueryTracer
    if cfg.OTEL.Endpoint != "" {
        queryTracer = obs.NewPgxTracer()
    }

    database, err := db.Open(ctx, cfg.DatabaseURL, db.WithQueryTracer(queryTracer))

git add go/go.mod go/go.sum go/internal/platform/config/config.go go/internal/platform/obs/tracing.go go/internal/platform/db/db.go go/cmd/obscura-server/wire.go
git commit -m "feat(tracing): opt-in OTel setup (no-op by default) + pgx query tracer + db.Open option"

package httpapi

import (
    "net/http"

    "github.com/go-chi/chi/v5"
    "github.com/go-chi/chi/v5/middleware"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/trace"
)

// tracing starts one server span per request. When tracing is disabled the global tracer is a
// no-op (a non-recording span with an invalid SpanContext), so this costs almost nothing. It runs
// OUTSIDE requestLog so the span id is already in ctx when the access line is written. The span is
// (re)named with the bounded chi route PATTERN — resolved only after routing — to keep span-name
// cardinality flat, mirroring the RED metrics.
func (s *Server) tracing(next http.Handler) http.Handler {
    tracer := otel.Tracer("obscura/http")
    prop := otel.GetTextMapPropagator()
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        ctx := prop.Extract(r.Context(), propagation.HeaderCarrier(r.Header))
        ctx, span := tracer.Start(ctx, r.Method, trace.WithSpanKind(trace.SpanKindServer))
        defer span.End()
        ww := middleware.NewWrapResponseWriter(w, r.ProtoMajor)
        r = r.WithContext(ctx)
        next.ServeHTTP(ww, r)
        route := chi.RouteContext(ctx).RoutePattern()
        if route == "" {
            route = "other"
        }
        span.SetName(r.Method + " " + route)
        span.SetAttributes(
            attribute.String("http.request.method", r.Method),
            attribute.String("http.route", route),
            attribute.Int("http.response.status_code", ww.Status()),
        )
        if p, ok := PrincipalFrom(ctx); ok {
            span.SetAttributes(attribute.String("user.id", string(p.UserID)))
        }
    })
}

    r.Use(middleware.RequestID)
    r.Use(s.tracing)
    r.Use(requestLog(s.logger))
    r.Use(s.metrics.instrument)
    r.Use(middleware.Recoverer)

            attrs := []slog.Attr{
                slog.String("method", r.Method),
                slog.String("route", route),
                slog.String("path", r.URL.Path),
                slog.Int("status", ww.Status()),
                slog.Int64("latency_ms", time.Since(start).Milliseconds()),
                slog.String("request_id", reqID),
                slog.String("remote", r.RemoteAddr),
            }
            if sc := trace.SpanContextFromContext(r.Context()); sc.IsValid() {
                attrs = append(attrs,
                    slog.String("trace_id", sc.TraceID().String()),
                    slog.String("span_id", sc.SpanID().String()),
                )
            }
            logger.LogAttrs(r.Context(), slog.LevelInfo, "http_request", attrs...)

git add go/internal/httpapi/tracing.go go/internal/httpapi/server.go go/internal/httpapi/logging.go
git commit -m "feat(tracing): HTTP server span middleware + trace_id/span_id log correlation"

cat > /tmp/otelcol.yaml <<'EOF'
receivers:
  otlp:
    protocols:
      http:
        endpoint: 0.0.0.0:4318
exporters:
  debug:
    verbosity: detailed
service:
  pipelines:
    traces:
      receivers: [otlp]
      exporters: [debug]
EOF
docker run -d --name otelcol --network <compose-net> -v /tmp/otelcol.yaml:/etc/otelcol/config.yaml otel/opentelemetry-collector:latest

-- +goose Up
-- Admin-tunable rate limits (singleton row id=1). Seeded with today's hardcoded constants so
-- behavior is byte-unchanged until an admin edits. The in-memory limiters read these live via an
-- atomic.Pointer refreshed on save (no restart). A safety floor is enforced in app validation so an
-- admin can't lock everyone (including themselves) out, and the auth-IP brute-force floor can't be
-- disabled to 0. public_sign_per_min governs the OTP-SEND budget (the abuse-sensitive one shown in
-- the UI); the OTP-submit budget stays a fixed constant.
CREATE TABLE rate_limit_settings (
    id                  int PRIMARY KEY DEFAULT 1 CHECK (id = 1),
    user_per_min        int NOT NULL DEFAULT 600,
    auth_ip_per_min     int NOT NULL DEFAULT 60,
    tsa_per_min         int NOT NULL DEFAULT 600,
    public_sign_per_min int NOT NULL DEFAULT 5,
    updated_at          timestamptz NOT NULL DEFAULT now()
);
INSERT INTO rate_limit_settings (id) VALUES (1);

-- +goose Down
DROP TABLE rate_limit_settings;

// Package domain holds the rate-limit configuration value type shared by the httpapi limiters and
// the admin editor. The four limits mirror the four in-memory limiters; per-limit safety floors
// prevent an admin from locking everyone out.
package domain

import "fmt"

// Config is the live rate-limit configuration (requests per minute per key).
type Config struct {
    UserPerMin       int // protected API, per authed user
    AuthIPPerMin     int // unauthenticated credential routes, per client IP (brute-force floor)
    TSAPerMin        int // RFC3161 TSA, per external client IP (loopback self-call exempt)
    PublicSignPerMin int // public OTP-SEND budget, per token+IP
}

// Per-limit safety floors + an upper ceiling. user/tsa are floored generously so the SPA and the
// signer keep working; auth-IP keeps a meaningful brute-force minimum; public-sign's floor is 1
// (its default is 5 and it is not an admin-lockout vector).
const (
    FloorUser       = 60
    FloorAuthIP     = 10
    FloorTSA        = 30
    FloorPublicSign = 1
    Ceil            = 100000
)

// Defaults returns the seeded constants (matches migration 00086 + the pre-existing consts).
func Defaults() Config {
    return Config{UserPerMin: 600, AuthIPPerMin: 60, TSAPerMin: 600, PublicSignPerMin: 5}
}

// Validate enforces the safety floors + ceiling. A rejected save returns a clear, admin-facing error.
func (c Config) Validate() error {
    for _, f := range []struct {
        name  string
        val   int
        floor int
    }{
        {"user_per_min", c.UserPerMin, FloorUser},
        {"auth_ip_per_min", c.AuthIPPerMin, FloorAuthIP},
        {"tsa_per_min", c.TSAPerMin, FloorTSA},
        {"public_sign_per_min", c.PublicSignPerMin, FloorPublicSign},
    } {
        if f.val < f.floor {
            return fmt.Errorf("%s must be at least %d (safety floor to prevent lockout)", f.name, f.floor)
        }
        if f.val > Ceil {
            return fmt.Errorf("%s must be at most %d", f.name, Ceil)
        }
    }
    return nil
}

// Package adapters persists the singleton rate_limit_settings row.
package adapters

import (
    "context"
    "errors"
    "fmt"

    "github.com/jackc/pgx/v5"

    "github.com/Virtue-Digital-Indonesia/obscura/internal/platform/db"
    "github.com/Virtue-Digital-Indonesia/obscura/internal/ratelimit/domain"
)

// Store reads/writes the singleton rate_limit_settings row.
type Store struct{ db *db.DB }

// NewStore builds the rate-limit store.
func NewStore(d *db.DB) *Store { return &Store{db: d} }

// Load reads the singleton config, falling back to seeded defaults when the row is missing or
// out-of-range (a corrupt row must never brick the limiters).
func (s *Store) Load(ctx context.Context) (domain.Config, error) {
    var c domain.Config
    err := s.db.Exec(ctx).QueryRow(ctx,
        `SELECT user_per_min, auth_ip_per_min, tsa_per_min, public_sign_per_min FROM rate_limit_settings WHERE id = 1`).
        Scan(&c.UserPerMin, &c.AuthIPPerMin, &c.TSAPerMin, &c.PublicSignPerMin)
    if errors.Is(err, pgx.ErrNoRows) {
        return domain.Defaults(), nil
    }
    if err != nil {
        return domain.Defaults(), fmt.Errorf("rate limit load: %w", err)
    }
    if c.Validate() != nil {
        return domain.Defaults(), nil
    }
    return c, nil
}

// Save writes the singleton config.
func (s *Store) Save(ctx context.Context, c domain.Config) error {
    if _, err := s.db.Exec(ctx).Exec(ctx,
        `UPDATE rate_limit_settings
            SET user_per_min = $1, auth_ip_per_min = $2, tsa_per_min = $3, public_sign_per_min = $4, updated_at = now()
          WHERE id = 1`,
        c.UserPerMin, c.AuthIPPerMin, c.TSAPerMin, c.PublicSignPerMin); err != nil {
        return fmt.Errorf("rate limit save: %w", err)
    }
    return nil
}