package otelrecorder

import (
        "context"
        "fmt"
        "sync"
        "time"

        "go.opentelemetry.io/otel/attribute"
        "go.opentelemetry.io/otel/metric"

        recorder "github.com/mgurevin/recorder"
)

type asyncRecorderMetrics struct {
        registration metric.Registration
        closeOnce    sync.Once
        closeErr     error
}

func newAsyncRecorderMetrics(meter metric.Meter, asyncRecorder *recorder.AsyncRecorder) (*asyncRecorderMetrics, error) {
        queueDepth, err := meter.Int64ObservableGauge("recorder.async.queue.depth",
                metric.WithUnit("{entry}"),
                metric.WithDescription("Entries waiting in the bounded asynchronous recorder queue"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async queue depth gauge: %w", err)
        }

        queueCapacity, err := meter.Int64ObservableGauge("recorder.async.queue.capacity",
                metric.WithUnit("{entry}"),
                metric.WithDescription("Configured asynchronous recorder queue capacity"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async queue capacity gauge: %w", err)
        }

        blockedNow, err := meter.Int64ObservableGauge("recorder.async.producers.blocked",
                metric.WithUnit("{producer}"),
                metric.WithDescription("Producers currently blocked waiting for queue capacity"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async blocked producers gauge: %w", err)
        }

        inFlight, err := meter.Int64ObservableGauge("recorder.async.entries.in_flight",
                metric.WithUnit("{entry}"),
                metric.WithDescription("Entries currently being processed by the downstream recorder"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async in-flight gauge: %w", err)
        }

        maxBlockTime, err := meter.Float64ObservableGauge("recorder.async.block.max",
                metric.WithUnit("ms"),
                metric.WithDescription("Longest observed producer block duration"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async maximum block time gauge: %w", err)
        }

        oldestBlockAge, err := meter.Float64ObservableGauge("recorder.async.block.oldest_age",
                metric.WithUnit("ms"),
                metric.WithDescription("Age of the oldest producer currently blocked for queue capacity"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async oldest block age gauge: %w", err)
        }

        maxBatchSize, err := meter.Int64ObservableGauge("recorder.async.batch.size.max",
                metric.WithUnit("{entry}"),
                metric.WithDescription("Largest asynchronous delivery batch observed"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async maximum batch size gauge: %w", err)
        }

        accepted, err := meter.Int64ObservableCounter("recorder.async.entries.accepted",
                metric.WithUnit("{entry}"),
                metric.WithDescription("Entries accepted by the asynchronous recorder"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async accepted counter: %w", err)
        }

        processed, err := meter.Int64ObservableCounter("recorder.async.entries.processed",
                metric.WithUnit("{entry}"),
                metric.WithDescription("Downstream Record calls attempted by the asynchronous recorder"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async processed counter: %w", err)
        }

        batches, err := meter.Int64ObservableCounter("recorder.async.batches.processed",
                metric.WithUnit("{batch}"),
                metric.WithDescription("Batches attempted by the asynchronous recorder"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async processed batches counter: %w", err)
        }

        blocked, err := meter.Int64ObservableCounter("recorder.async.records.blocked",
                metric.WithUnit("{record}"),
                metric.WithDescription("Record calls that waited for asynchronous queue capacity"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async blocked records counter: %w", err)
        }

        blockTime, err := meter.Float64ObservableCounter("recorder.async.block.duration",
                metric.WithUnit("ms"),
                metric.WithDescription("Cumulative producer time spent waiting for queue capacity"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async block duration counter: %w", err)
        }

        dropped, err := meter.Int64ObservableCounter("recorder.async.entries.dropped",
                metric.WithUnit("{entry}"),
                metric.WithDescription("Entries dropped by the asynchronous recorder, by bounded reason"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async dropped counter: %w", err)
        }

        sinkPanics, err := meter.Int64ObservableCounter("recorder.async.sink.panics",
                metric.WithUnit("{panic}"),
                metric.WithDescription("Panics recovered from the downstream recorder"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async sink panic counter: %w", err)
        }

        sinkErrors, err := meter.Int64ObservableCounter("recorder.async.sink.errors",
                metric.WithUnit("{error}"),
                metric.WithDescription("Observable downstream recorder and close errors"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async sink error counter: %w", err)
        }

        dropHandlerPanics, err := meter.Int64ObservableCounter("recorder.async.drop_handler.panics",
                metric.WithUnit("{panic}"),
                metric.WithDescription("Panics recovered from the asynchronous recorder drop handler"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async drop handler panic counter: %w", err)
        }

        dropHandlerErrors, err := meter.Int64ObservableCounter("recorder.async.drop_handler.errors",
                metric.WithUnit("{error}"),
                metric.WithDescription("Errors returned by the asynchronous recorder drop handler"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create async drop handler error counter: %w", err)
        }

        instruments := []metric.Observable{
                queueDepth, queueCapacity, blockedNow, inFlight, maxBlockTime, oldestBlockAge, maxBatchSize,
                accepted, processed, batches, blocked, blockTime, dropped, sinkPanics, sinkErrors,
                dropHandlerPanics, dropHandlerErrors,
        }

        registration, err := meter.RegisterCallback(func(_ context.Context, observer metric.Observer) error {
                stats := asyncRecorder.Stats()
                observer.ObserveInt64(queueDepth, int64(stats.Pending))
                observer.ObserveInt64(queueCapacity, int64(stats.Capacity))
                observer.ObserveInt64(blockedNow, int64(stats.CurrentlyBlocked))
                observer.ObserveInt64(inFlight, int64(stats.InFlight))
                observer.ObserveFloat64(maxBlockTime, float64(stats.MaxBlockTime)/float64(time.Millisecond))
                observer.ObserveFloat64(oldestBlockAge, float64(stats.OldestBlockAge)/float64(time.Millisecond))
                observer.ObserveInt64(maxBatchSize, int64(stats.MaxBatchSize))
                observer.ObserveInt64(accepted, int64(stats.Accepted))
                observer.ObserveInt64(processed, int64(stats.Processed))
                observer.ObserveInt64(batches, int64(stats.BatchesProcessed))
                observer.ObserveInt64(blocked, int64(stats.BlockedRecords))
                observer.ObserveFloat64(blockTime, float64(stats.TotalBlockTime)/float64(time.Millisecond))
                observer.ObserveInt64(dropped, int64(stats.DroppedNewest), metric.WithAttributes(attribute.String("recorder.async.drop.reason", string(recorder.AsyncDropPolicyNewest))))
                observer.ObserveInt64(dropped, int64(stats.DroppedOldest), metric.WithAttributes(attribute.String("recorder.async.drop.reason", string(recorder.AsyncDropPolicyOldest))))
                observer.ObserveInt64(dropped, int64(stats.DroppedTimeoutNewest), metric.WithAttributes(attribute.String("recorder.async.drop.reason", string(recorder.AsyncDropTimeoutNewest))))
                observer.ObserveInt64(dropped, int64(stats.DroppedTimeoutOldest), metric.WithAttributes(attribute.String("recorder.async.drop.reason", string(recorder.AsyncDropTimeoutOldest))))
                observer.ObserveInt64(dropped, int64(stats.DroppedClosed), metric.WithAttributes(attribute.String("recorder.async.drop.reason", string(recorder.AsyncDropClosed))))
                observer.ObserveInt64(sinkPanics, int64(stats.SinkPanics))
                observer.ObserveInt64(sinkErrors, int64(stats.SinkErrors))
                observer.ObserveInt64(dropHandlerPanics, int64(stats.DropHandlerPanics))
                observer.ObserveInt64(dropHandlerErrors, int64(stats.DropHandlerErrors))

                return nil
        }, instruments...)
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: register async recorder metrics: %w", err)
        }

        return &asyncRecorderMetrics{registration: registration}, nil
}

func (m *asyncRecorderMetrics) close() error {
        m.closeOnce.Do(func() {
                m.closeErr = m.registration.Unregister()
        })

        return m.closeErr
}

package otelrecorder

import (
        "net/url"
        "strconv"
        "strings"

        "go.opentelemetry.io/otel/attribute"

        recorder "github.com/mgurevin/recorder"
)

func recorderExtension(entry *recorder.Entry) *recorder.RecorderEntryExtension {
        if entry != nil && entry.Recorder != nil {
                return entry.Recorder
        }

        return &recorder.RecorderEntryExtension{}
}

func entryDispositionAttribute(disposition recorder.EntryDisposition) attribute.KeyValue {
        value := "keep"
        if disposition == recorder.EntryDispositionDiscard {
                value = "discard"
        }

        return attribute.String("recorder.entry.disposition", value)
}

// baseAttributes is the shared low-cardinality set used on synthesized spans
// and inside span events. It never contains URLs beyond scheme+host, header
// or body material, or correlation IDs.
func (e *Exporter) baseAttributes(entry *recorder.Entry) []attribute.KeyValue {
        attrs := make([]attribute.KeyValue, 0, 16)

        if entry.Request != nil {
                if entry.Request.Method != "" {
                        attrs = append(attrs, attribute.String("http.request.method", e.clamp(entry.Request.Method)))
                }

                if scheme, host := schemeAndHost(entry.Request.URL); scheme != "" || host != "" {
                        if scheme != "" {
                                attrs = append(attrs, attribute.String("url.scheme", scheme))
                        }

                        if host != "" {
                                attrs = append(attrs, attribute.String("server.address", e.clamp(host)))
                        }
                }
        }

        if entry.Response != nil {
                attrs = append(attrs, attribute.Int("http.response.status_code", entry.Response.Status))
        }

        if name, version := protocol(entry); name != "" {
                attrs = append(attrs, attribute.String("network.protocol.name", name))
                if version != "" {
                        attrs = append(attrs, attribute.String("network.protocol.version", version))
                }
        }

        attrs = append(attrs, attribute.String("recorder.state", e.clamp(recorderExtension(entry).State)))
        if recorderExtension(entry).Error != nil {
                attrs = append(attrs, attribute.String("recorder.error.phase", e.clamp(recorderExtension(entry).Error.Phase)))
        }

        if recorderExtension(entry).ResponseBodyDecoded {
                attrs = append(attrs, attribute.Bool("recorder.response.decoded", true))
        }

        if recorderExtension(entry).RequestBody != nil && recorderExtension(entry).RequestBody.Truncated {
                attrs = append(attrs, attribute.Bool("recorder.request.body.truncated", true))
        }

        if recorderExtension(entry).ResponseBody != nil && recorderExtension(entry).ResponseBody.Truncated {
                attrs = append(attrs, attribute.Bool("recorder.response.body.truncated", true))
        }

        return attrs
}

// eventAttributes builds the full "recorder.http.exchange" span event set:
// base attributes plus numeric timings, body sizes, network and TLS summary.
func (e *Exporter) eventAttributes(entry *recorder.Entry) []attribute.KeyValue {
        attrs := e.baseAttributes(entry)

        attrs = append(attrs, attribute.Float64("recorder.duration_ms", entry.Time))
        if t := entry.Timings; t != nil {
                for _, tv := range []struct {
                        name string
                        v    float64
                }{
                        {"blocked", t.Blocked},
                        {"dns", t.DNS},
                        {"connect", t.Connect},
                        {"ssl", t.SSL},
                        {"send", t.Send},
                        {"wait", t.Wait},
                        {"receive", t.Receive},
                } {
                        if tv.v >= 0 { // -1 means unmeasured; omit rather than mislead
                                attrs = append(attrs, attribute.Float64("recorder.timings."+tv.name, tv.v))
                        }
                }
        }

        if rb := recorderExtension(entry).RequestBody; rb != nil {
                attrs = append(attrs, attribute.Int64("recorder.request.body.bytes", rb.TotalBytes))
        }

        if rb := recorderExtension(entry).ResponseBody; rb != nil {
                attrs = append(attrs, attribute.Int64("recorder.response.body.bytes", rb.TotalBytes))
        }

        if closedEarly(entry) {
                attrs = append(attrs, attribute.Bool("recorder.response.closed_early", true))
        }

        if n := recorderExtension(entry).Network; n != nil {
                attrs = append(attrs,
                        attribute.Bool("recorder.network.reused", n.ConnectionReused),
                        attribute.Bool("recorder.network.http2", n.HTTP2),
                )
                if n.WasIdle {
                        attrs = append(attrs, attribute.Bool("recorder.network.was_idle", true))
                }

                if n.DNSCoalesced {
                        attrs = append(attrs, attribute.Bool("recorder.network.dns_coalesced", true))
                }
        }

        if tl := recorderExtension(entry).TLS; tl != nil {
                attrs = append(attrs,
                        attribute.String("recorder.tls.version", e.clamp(tl.Version)),
                        attribute.String("recorder.tls.cipher_suite", e.clamp(tl.CipherSuite)),
                )
                if tl.DidResume {
                        attrs = append(attrs, attribute.Bool("recorder.tls.resumed", true))
                }
        }

        if e.cfg.IncludeIDs {
                if recorderExtension(entry).TraceID != "" {
                        attrs = append(attrs, attribute.String("recorder.trace_id", e.clamp(recorderExtension(entry).TraceID)))
                }

                if recorderExtension(entry).ExchangeID != "" {
                        attrs = append(attrs, attribute.String("recorder.exchange_id", e.clamp(recorderExtension(entry).ExchangeID)))
                }
        }

        return e.appendCustom(attrs, e.cfg.SpanEventAttributes, entry)
}

// metricAttributes is the deliberately minimal label set for every metric.
func (e *Exporter) metricAttributes(entry *recorder.Entry) []attribute.KeyValue {
        attrs := make([]attribute.KeyValue, 0, 8)
        if entry.Request != nil && entry.Request.Method != "" {
                attrs = append(attrs, attribute.String("http.request.method", e.clamp(entry.Request.Method)))
        }

        status := 0
        if entry.Response != nil {
                status = entry.Response.Status
        }

        attrs = append(attrs, attribute.String("http.response.status_class", statusClass(status)))

        attrs = append(attrs, attribute.String("recorder.state", e.clamp(recorderExtension(entry).State)))
        if entry.Request != nil {
                if scheme, _ := schemeAndHost(entry.Request.URL); scheme != "" {
                        attrs = append(attrs, attribute.String("url.scheme", scheme))
                }
        }

        if name, version := protocol(entry); name != "" {
                attrs = append(attrs, attribute.String("network.protocol.name", name))
                if version != "" {
                        attrs = append(attrs, attribute.String("network.protocol.version", version))
                }
        }

        return e.appendCustom(attrs, e.cfg.MetricAttributes, entry)
}

func (e *Exporter) appendCustom(dst []attribute.KeyValue,
        fn func(*recorder.Entry) []attribute.KeyValue, entry *recorder.Entry,
) []attribute.KeyValue {
        if fn == nil {
                return dst
        }

        extra := fn(entry)
        if len(extra) > maxCustomAttributes {
                extra = extra[:maxCustomAttributes]
        }

        for _, kv := range extra {
                if kv.Value.Type() == attribute.STRING {
                        kv = attribute.String(string(kv.Key), e.clamp(kv.Value.AsString()))
                }

                dst = append(dst, kv)
        }

        return dst
}

// clamp bounds a string attribute value to the configured maximum length.
func (e *Exporter) clamp(s string) string {
        if e.cfg.MaxAttributeLength > 0 && len(s) > e.cfg.MaxAttributeLength {
                return s[:e.cfg.MaxAttributeLength]
        }

        return s
}

// statusClass folds a status code into a bounded label: "0" for exchanges
// without a response, otherwise "1xx".."5xx".
func statusClass(status int) string {
        if status <= 0 {
                return "0"
        }

        return strconv.Itoa(status/100) + "xx"
}

// schemeAndHost extracts only the scheme and host from the recorded URL —
// path, query, fragment and userinfo never leave the adapter.
func schemeAndHost(raw string) (scheme, host string) {
        u, err := url.Parse(raw)
        if err != nil {
                return "", ""
        }

        return u.Scheme, u.Hostname()
}

// protocol maps the recorded HTTP version onto network.protocol.name/version
// ("HTTP/2.0" -> "http"/"2"). Unknown versions yield empty values rather
// than guesses.
func protocol(entry *recorder.Entry) (name, version string) {
        v := ""
        if entry.Response != nil && entry.Response.HTTPVersion != "" {
                v = entry.Response.HTTPVersion
        } else if entry.Request != nil {
                v = entry.Request.HTTPVersion
        }

        if v == "" {
                return "", ""
        }

        rest, ok := strings.CutPrefix(v, "HTTP/")
        if !ok {
                return "", ""
        }

        rest = strings.TrimSuffix(rest, ".0")

        return "http", rest
}

// Package otelrecorder exports finished recorder entries as OpenTelemetry
// span events and metrics. It lives in its own Go module so the core
// recorder package stays dependency-free.
//
// The adapter plugs into recorder.Config.OnEntryCompleted:
//
//        exporter, err := otelrecorder.NewExporter(otelrecorder.DefaultConfig())
//        if err != nil { ... }
//        config := recorder.DefaultConfig()
//        config.OnEntryCompleted = exporter.OnEntryCompleted
//        client := &http.Client{
//                Transport: recorder.NewTransport(http.DefaultTransport, rec, config),
//        }
//
// When the request context carries an active (recording) span, the entry is
// attached to it as a span event named "recorder.http.exchange". Without an
// active span nothing is traced by default; Config.CreateSpanIfNone makes
// the exporter synthesize a client span covering the exchange instead.
// Metrics are always recorded.
//
// # Cardinality and safety
//
// Only low-cardinality, bounded fields ever leave this adapter. It never
// exports full URLs, paths, query strings, header or cookie values, body
// content, or raw HAR JSON — not even in redacted form. server.address
// carries the host only. High-cardinality correlation IDs from _recorder are
// excluded by default and can be opted into as span event
// attributes only (Config.IncludeIDs); they are never metric attributes.
// Metric attributes are limited to method, status code *class* ("2xx",
// "0"), state, entry disposition, scheme and protocol. Specialized instruments
// add only bounded dimensions: HTTP phase, body direction/capture outcome,
// protection mode, fixed fail-closed reason, body-redactor kind/outcome, and
// fixed async drop reason. Numeric facts
// (timings, body sizes and protection counts) are measurements, never labels.
// String attribute values are clamped to MaxAttributeLength.
//
// In addition to total duration, streamed body sizes and exchange failures,
// the exporter reports per-phase latency, retained body bytes, capture
// outcomes, redacted/encrypted/tokenized value counts, fail-closed fallbacks,
// body-redactor outcomes, and optional AsyncRecorder queue/backpressure health.
// No rule names, key IDs, protected values, error messages, body content,
// storage paths or sink identities become attributes.
//
// Custom attributes (Config.SpanEventAttributes / Config.MetricAttributes) are the
// intended hook for user-controlled low-cardinality dimensions such as a URL
// path *template* ("/users/{id}"). Never derive them from raw request data:
// every distinct metric attribute value creates a new time series, and
// unbounded values (IDs, full paths, tokens) will blow up your metrics
// backend.
//
// When the full HAR entry is needed, write it to a dedicated sink
// (recorder.HARFileRecorder, recorder.JSONStreamRecorder) — an OTel
// attribute is the wrong place for a document.
package otelrecorder

import (
        "context"
        "fmt"
        "time"

        "go.opentelemetry.io/otel"
        "go.opentelemetry.io/otel/attribute"
        "go.opentelemetry.io/otel/codes"
        "go.opentelemetry.io/otel/metric"
        "go.opentelemetry.io/otel/trace"

        recorder "github.com/mgurevin/recorder"
)

const (
        instrumentationName = "github.com/mgurevin/recorder/otelrecorder"

        // EventName is the span event name used for every exchange.
        EventName = "recorder.http.exchange"

        // maxCustomAttributes bounds user-supplied attribute lists.
        maxCustomAttributes = 16

        // defaultMaxAttributeLength clamps string attribute values.
        defaultMaxAttributeLength = 128
)

// Config defines providers, span behavior, custom attributes, and optional
// process-health metric sources for NewExporter.
type Config struct {
        // TracerProvider defaults to OpenTelemetry's global tracer provider.
        TracerProvider trace.TracerProvider
        // MeterProvider defaults to OpenTelemetry's global meter provider.
        MeterProvider metric.MeterProvider
        // CreateSpanIfNone synthesizes a short client span when none is recording.
        CreateSpanIfNone bool
        // SetSpanErrorStatus marks the touched span when the exchange failed.
        SetSpanErrorStatus bool
        // IncludeIDs adds high-cardinality recorder IDs to span events, never metrics.
        IncludeIDs bool
        // MaxAttributeLength bounds custom and built-in string attributes.
        MaxAttributeLength int
        // SpanEventAttributes supplies at most 16 application-controlled attributes.
        SpanEventAttributes func(*recorder.Entry) []attribute.KeyValue
        // MetricAttributes supplies at most 16 bounded, low-cardinality attributes.
        MetricAttributes func(*recorder.Entry) []attribute.KeyValue
        // AsyncRecorder enables queue, backpressure, drop, and sink-health metrics.
        AsyncRecorder *recorder.AsyncRecorder
        // FileBodyStore enables capacity and asset-lifecycle metrics.
        FileBodyStore *recorder.FileBodyStore
        // SamplingTransport enables sampling, retention, and cleanup metrics.
        SamplingTransport *recorder.Transport
}

// DefaultConfig returns the bounded, low-cardinality exporter defaults.
func DefaultConfig() Config { return Config{MaxAttributeLength: defaultMaxAttributeLength} }

// Exporter converts finished entries into OTel span events and metrics.
// Safe for concurrent use; entries arrive from whichever goroutine finished
// the exchange.
type Exporter struct {
        cfg    Config
        tracer trace.Tracer

        duration        metric.Float64Histogram
        phase           metric.Float64Histogram
        reqSize         metric.Int64Histogram
        respSize        metric.Int64Histogram
        capturedSize    metric.Int64Histogram
        failures        metric.Int64Counter
        closedEarly     metric.Int64Counter
        truncated       metric.Int64Counter
        captures        metric.Int64Counter
        redacted        metric.Int64Counter
        fallbacks       metric.Int64Counter
        bodyRedactions  metric.Int64Counter
        asyncMetrics    *asyncRecorderMetrics
        fileBodyMetrics *fileBodyStoreMetrics
        samplingMetrics *samplingMetrics
}

// NewExporter builds an Exporter. Instrument creation errors (invalid meter
// implementations) are returned rather than silently dropped.
func NewExporter(cfg Config) (*Exporter, error) {
        if cfg.TracerProvider == nil {
                cfg.TracerProvider = otel.GetTracerProvider()
        }

        if cfg.MeterProvider == nil {
                cfg.MeterProvider = otel.GetMeterProvider()
        }

        e := &Exporter{cfg: cfg, tracer: cfg.TracerProvider.Tracer(instrumentationName)}
        meter := cfg.MeterProvider.Meter(instrumentationName)

        var err error
        if e.duration, err = meter.Float64Histogram("recorder.http.client.duration",
                metric.WithUnit("ms"),
                metric.WithDescription("Total duration of recorded HTTP exchanges")); err != nil {
                return nil, fmt.Errorf("otelrecorder: create duration histogram: %w", err)
        }

        if e.phase, err = meter.Float64Histogram("recorder.http.client.phase.duration",
                metric.WithUnit("ms"),
                metric.WithDescription("Duration of measured HTTP exchange phases")); err != nil {
                return nil, fmt.Errorf("otelrecorder: create phase duration histogram: %w", err)
        }

        if e.reqSize, err = meter.Int64Histogram("recorder.http.client.request.body.size",
                metric.WithUnit("By"),
                metric.WithDescription("Request body bytes that flowed through the stream")); err != nil {
                return nil, fmt.Errorf("otelrecorder: create request size histogram: %w", err)
        }

        if e.respSize, err = meter.Int64Histogram("recorder.http.client.response.body.size",
                metric.WithUnit("By"),
                metric.WithDescription("Response body bytes that flowed through the stream")); err != nil {
                return nil, fmt.Errorf("otelrecorder: create response size histogram: %w", err)
        }

        if e.capturedSize, err = meter.Int64Histogram("recorder.body.captured.size",
                metric.WithUnit("By"),
                metric.WithDescription("Body bytes retained by the configured capture policy")); err != nil {
                return nil, fmt.Errorf("otelrecorder: create captured body size histogram: %w", err)
        }

        if e.failures, err = meter.Int64Counter("recorder.http.client.failures",
                metric.WithDescription("Exchanges that failed at the transport or body layer, by phase")); err != nil {
                return nil, fmt.Errorf("otelrecorder: create failure counter: %w", err)
        }

        if e.closedEarly, err = meter.Int64Counter("recorder.http.client.closed_early",
                metric.WithDescription("Response bodies closed before EOF")); err != nil {
                return nil, fmt.Errorf("otelrecorder: create closed-early counter: %w", err)
        }

        if e.truncated, err = meter.Int64Counter("recorder.http.client.body.truncated",
                metric.WithDescription("Body captures truncated by the configured limit")); err != nil {
                return nil, fmt.Errorf("otelrecorder: create truncated counter: %w", err)
        }

        if e.captures, err = meter.Int64Counter("recorder.body.capture.operations",
                metric.WithDescription("Body capture outcomes by direction")); err != nil {
                return nil, fmt.Errorf("otelrecorder: create body capture counter: %w", err)
        }

        if e.redacted, err = meter.Int64Counter("recorder.redaction.values",
                metric.WithDescription("Recorded sensitive values by protection mode and direction")); err != nil {
                return nil, fmt.Errorf("otelrecorder: create redaction value counter: %w", err)
        }

        if e.fallbacks, err = meter.Int64Counter("recorder.redaction.fallbacks",
                metric.WithDescription("Fail-closed protection fallbacks by fixed reason and direction")); err != nil {
                return nil, fmt.Errorf("otelrecorder: create redaction fallback counter: %w", err)
        }

        if e.bodyRedactions, err = meter.Int64Counter("recorder.body.redaction.operations",
                metric.WithDescription("Body redactor executions by bounded kind, outcome and direction")); err != nil {
                return nil, fmt.Errorf("otelrecorder: create body redaction counter: %w", err)
        }

        if cfg.AsyncRecorder != nil {
                e.asyncMetrics, err = newAsyncRecorderMetrics(meter, cfg.AsyncRecorder)
                if err != nil {
                        return nil, err
                }
        }

        if cfg.FileBodyStore != nil {
                e.fileBodyMetrics, err = newFileBodyStoreMetrics(meter, cfg.FileBodyStore)
                if err != nil {
                        if e.asyncMetrics != nil {
                                _ = e.asyncMetrics.close()
                        }

                        return nil, err
                }
        }

        if cfg.SamplingTransport != nil {
                e.samplingMetrics, err = newSamplingMetrics(meter, cfg.SamplingTransport)
                if err != nil {
                        if e.asyncMetrics != nil {
                                _ = e.asyncMetrics.close()
                        }

                        if e.fileBodyMetrics != nil {
                                _ = e.fileBodyMetrics.close()
                        }

                        return nil, err
                }
        }

        return e, nil
}

// Close unregisters observable metric callbacks owned by the exporter. It is
// safe to call more than once. Close does not close the AsyncRecorder.
func (e *Exporter) Close() error {
        if e == nil {
                return nil
        }

        var firstErr error
        if e.asyncMetrics != nil {
                firstErr = e.asyncMetrics.close()
        }

        if e.fileBodyMetrics != nil {
                if err := e.fileBodyMetrics.close(); firstErr == nil {
                        firstErr = err
                }
        }

        if e.samplingMetrics != nil {
                if err := e.samplingMetrics.close(); firstErr == nil {
                        firstErr = err
                }
        }

        return firstErr
}

// OnEntryCompleted implements recorder.OnEntryCompleted. Wire it up with
// recorder.Config.OnEntryCompleted. The effective keep/discard disposition is
// exported as a bounded span-event and metric attribute.
func (e *Exporter) OnEntryCompleted(
        ctx context.Context,
        entry *recorder.Entry,
        disposition recorder.EntryDisposition,
) {
        if entry == nil {
                return
        }

        e.recordMetrics(ctx, entry, disposition)
        e.recordSpan(ctx, entry, disposition)
}

func (e *Exporter) recordSpan(
        ctx context.Context,
        entry *recorder.Entry,
        disposition recorder.EntryDisposition,
) {
        span := trace.SpanFromContext(ctx)
        created := false

        if !span.IsRecording() {
                if !e.cfg.CreateSpanIfNone {
                        return
                }

                start := entry.StartTime()
                _, span = e.tracer.Start(ctx, spanName(entry),
                        trace.WithSpanKind(trace.SpanKindClient),
                        trace.WithTimestamp(start))
                created = true
        }

        end := entry.StartTime().Add(time.Duration(entry.Time * float64(time.Millisecond)))
        dispositionAttr := entryDispositionAttribute(disposition)
        span.AddEvent(EventName,
                trace.WithTimestamp(end),
                trace.WithAttributes(append(e.eventAttributes(entry), dispositionAttr)...))

        if e.cfg.SetSpanErrorStatus && recorderExtension(entry).Error != nil {
                span.SetStatus(codes.Error, e.clamp(recorderExtension(entry).Error.Phase))
        }

        if created {
                span.SetAttributes(append(e.baseAttributes(entry), dispositionAttr)...)
                span.End(trace.WithTimestamp(end))
        }
}

func (e *Exporter) recordMetrics(
        ctx context.Context,
        entry *recorder.Entry,
        disposition recorder.EntryDisposition,
) {
        attrs := append(e.metricAttributes(entry), entryDispositionAttribute(disposition))
        opt := metric.WithAttributes(attrs...)

        e.duration.Record(ctx, entry.Time, opt)
        e.recordPhaseMetrics(ctx, entry, attrs)

        if rb := recorderExtension(entry).RequestBody; rb != nil {
                e.reqSize.Record(ctx, rb.TotalBytes, opt)
                e.recordBodyCapture(ctx, rb, "request", attrs)

                if rb.Truncated {
                        e.truncated.Add(ctx, 1, metric.WithAttributes(append(attrs,
                                attribute.String("recorder.body.direction", "request"))...))
                }
        }

        if rb := recorderExtension(entry).ResponseBody; rb != nil {
                e.respSize.Record(ctx, rb.TotalBytes, opt)
                e.recordBodyCapture(ctx, rb, "response", attrs)

                if rb.Truncated {
                        e.truncated.Add(ctx, 1, metric.WithAttributes(append(attrs,
                                attribute.String("recorder.body.direction", "response"))...))
                }
        }

        if recorderExtension(entry).Error != nil {
                e.failures.Add(ctx, 1, metric.WithAttributes(append(attrs,
                        attribute.String("recorder.error.phase", e.clamp(recorderExtension(entry).Error.Phase)))...))
        }

        if closedEarly(entry) {
                e.closedEarly.Add(ctx, 1, opt)
        }

        e.recordRedactionMetrics(ctx, recorderExtension(entry).Redaction, attrs)
}

func (e *Exporter) recordPhaseMetrics(ctx context.Context, entry *recorder.Entry, attrs []attribute.KeyValue) {
        if entry.Timings == nil {
                return
        }

        for _, phase := range []struct {
                name  string
                value float64
        }{
                {"blocked", entry.Timings.Blocked},
                {"dns", entry.Timings.DNS},
                {"connect", entry.Timings.Connect},
                {"tls", entry.Timings.SSL},
                {"send", entry.Timings.Send},
                {"wait", entry.Timings.Wait},
                {"receive", entry.Timings.Receive},
        } {
                if phase.value >= 0 {
                        e.phase.Record(ctx, phase.value, metric.WithAttributes(metricAttrs(attrs,
                                attribute.String("recorder.http.phase", phase.name))...))
                }
        }
}

func (e *Exporter) recordBodyCapture(ctx context.Context, body *recorder.BodyInfo,
        direction string, attrs []attribute.KeyValue,
) {
        if !body.Present {
                return
        }

        bodyAttrs := metricAttrs(attrs, attribute.String("recorder.body.direction", direction))
        e.capturedSize.Record(ctx, body.CapturedBytes, metric.WithAttributes(bodyAttrs...))
        e.captures.Add(ctx, 1, metric.WithAttributes(metricAttrs(bodyAttrs,
                attribute.String("recorder.body.capture.outcome", bodyCaptureOutcome(body)))...))
}

func bodyCaptureOutcome(body *recorder.BodyInfo) string {
        switch {
        case body.ReadError != "" || body.CloseError != "":
                return "failed"

        case body.ClosedEarly:
                return "closed_early"

        case body.Truncated:
                return "truncated"

        case body.Complete:
                return "complete"

        default:
                return "incomplete"
        }
}

func (e *Exporter) recordRedactionMetrics(ctx context.Context, info *recorder.RedactionInfo,
        attrs []attribute.KeyValue,
) {
        if info == nil {
                return
        }

        e.recordRedactionScope(ctx, info.Request, "request", attrs)
        e.recordRedactionScope(ctx, info.Response, "response", attrs)
}

func (e *Exporter) recordRedactionScope(ctx context.Context, scope *recorder.RedactionScopeInfo,
        direction string, attrs []attribute.KeyValue,
) {
        if scope == nil {
                return
        }

        e.recordProtectionCounts(ctx, scope.Protection, direction, attrs)

        if scope.Body == nil {
                return
        }

        bodyAttrs := metricAttrs(attrs,
                attribute.String("recorder.body.direction", direction),
                attribute.String("recorder.body.redaction.kind", bodyRedactionKind(scope.Body.Kind)),
                attribute.String("recorder.body.redaction.outcome", bodyRedactionOutcome(scope.Body.Outcome)),
        )
        e.bodyRedactions.Add(ctx, 1, metric.WithAttributes(bodyAttrs...))
        e.recordProtectionCounts(ctx, scope.Body.Protection, direction, attrs)
}

func (e *Exporter) recordProtectionCounts(ctx context.Context, counts *recorder.ProtectionCounts,
        direction string, attrs []attribute.KeyValue,
) {
        if counts == nil {
                return
        }

        for _, mode := range []struct {
                name  string
                count int64
        }{
                {"redacted", counts.Redacted},
                {"encrypted", counts.Encrypted},
                {"tokenized", counts.Tokenized},
        } {
                if mode.count > 0 {
                        e.redacted.Add(ctx, mode.count, metric.WithAttributes(metricAttrs(attrs,
                                attribute.String("recorder.redaction.direction", direction),
                                attribute.String("recorder.protection.mode", mode.name))...))
                }
        }

        for reason, count := range counts.Fallbacks {
                if count > 0 {
                        e.fallbacks.Add(ctx, count, metric.WithAttributes(metricAttrs(attrs,
                                attribute.String("recorder.redaction.direction", direction),
                                attribute.String("recorder.protection.reason", protectionFallbackReason(reason)))...))
                }
        }
}

func metricAttrs(base []attribute.KeyValue, extra ...attribute.KeyValue) []attribute.KeyValue {
        attrs := make([]attribute.KeyValue, 0, len(base)+len(extra))
        attrs = append(attrs, base...)
        attrs = append(attrs, extra...)

        return attrs
}

func protectionFallbackReason(reason string) string {
        switch reason {
        case "value_too_large", "encryption_failed", "tokenization_failed":
                return reason

        default:
                return "other"
        }
}

func bodyRedactionKind(kind string) string {
        switch kind {
        case "custom", "builtin:multipart", "builtin:form", "builtin:json", "builtin:xml", "builtin:sniff":
                return kind

        default:
                return "other"
        }
}

func bodyRedactionOutcome(outcome string) string {
        switch outcome {
        case recorder.BodyRedactionRedacted, recorder.BodyRedactionUnchanged, recorder.BodyRedactionFailed:
                return outcome

        default:
                return "other"
        }
}

func spanName(entry *recorder.Entry) string {
        if entry.Request != nil && entry.Request.Method != "" {
                return "HTTP " + entry.Request.Method
        }

        return "HTTP"
}

func closedEarly(entry *recorder.Entry) bool {
        if recorderExtension(entry).State == recorder.StateClosedEarly {
                return true
        }

        return recorderExtension(entry).ResponseBody != nil && recorderExtension(entry).ResponseBody.ClosedEarly
}

package otelrecorder

import (
        "context"
        "fmt"
        "sync"

        "go.opentelemetry.io/otel/attribute"
        "go.opentelemetry.io/otel/metric"

        recorder "github.com/mgurevin/recorder"
)

type fileBodyStoreMetrics struct {
        registration metric.Registration
        closeOnce    sync.Once
        closeErr     error
}

func newFileBodyStoreMetrics(meter metric.Meter, store *recorder.FileBodyStore) (*fileBodyStoreMetrics, error) {
        files, err := meter.Int64ObservableGauge("recorder.body.store.files",
                metric.WithUnit("{file}"),
                metric.WithDescription("Body files owned by the managed store, by bounded state"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create body store files gauge: %w", err)
        }

        bytesUsed, err := meter.Int64ObservableGauge("recorder.body.store.bytes",
                metric.WithUnit("By"),
                metric.WithDescription("Body bytes owned by the managed store, by bounded state"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create body store bytes gauge: %w", err)
        }

        maxFiles, err := meter.Int64ObservableGauge("recorder.body.store.capacity.files",
                metric.WithUnit("{file}"),
                metric.WithDescription("Configured managed body store file capacity"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create body store file capacity gauge: %w", err)
        }

        maxBytes, err := meter.Int64ObservableGauge("recorder.body.store.capacity.bytes",
                metric.WithUnit("By"),
                metric.WithDescription("Configured managed body store byte capacity"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create body store byte capacity gauge: %w", err)
        }

        operations, err := meter.Int64ObservableCounter("recorder.body.store.operations",
                metric.WithUnit("{operation}"),
                metric.WithDescription("Managed body store lifecycle outcomes"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create body store operation counter: %w", err)
        }

        stateKey := attribute.Key("recorder.body.store.state")
        outcomeKey := attribute.Key("recorder.body.store.outcome")

        registration, err := meter.RegisterCallback(func(_ context.Context, observer metric.Observer) error {
                stats := store.Stats()
                observer.ObserveInt64(files, int64(stats.PartialFiles), metric.WithAttributes(stateKey.String("partial")))
                observer.ObserveInt64(files, int64(stats.CommittedFiles), metric.WithAttributes(stateKey.String("committed")))
                observer.ObserveInt64(bytesUsed, stats.PartialBytes, metric.WithAttributes(stateKey.String("partial")))
                observer.ObserveInt64(bytesUsed, stats.CommittedBytes, metric.WithAttributes(stateKey.String("committed")))
                observer.ObserveInt64(maxFiles, int64(stats.MaxFiles))
                observer.ObserveInt64(maxBytes, stats.MaxBytes)
                observer.ObserveInt64(operations, int64(stats.CommittedTotal), metric.WithAttributes(outcomeKey.String("committed")))
                observer.ObserveInt64(operations, int64(stats.AbortedTotal), metric.WithAttributes(outcomeKey.String("aborted")))
                observer.ObserveInt64(operations, int64(stats.ReleasedTotal), metric.WithAttributes(outcomeKey.String("released")))
                observer.ObserveInt64(operations, int64(stats.QuotaRejected), metric.WithAttributes(outcomeKey.String("quota_rejected")))
                observer.ObserveInt64(operations, int64(stats.RecoveredPartials), metric.WithAttributes(outcomeKey.String("recovered_partial")))
                observer.ObserveInt64(operations, int64(stats.WriteFailures), metric.WithAttributes(outcomeKey.String("write_failed")))
                observer.ObserveInt64(operations, int64(stats.CommitFailures), metric.WithAttributes(outcomeKey.String("commit_failed")))
                observer.ObserveInt64(operations, int64(stats.AbortFailures), metric.WithAttributes(outcomeKey.String("abort_failed")))
                observer.ObserveInt64(operations, int64(stats.ReleaseFailures), metric.WithAttributes(outcomeKey.String("release_failed")))

                return nil
        }, files, bytesUsed, maxFiles, maxBytes, operations)
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: register body store metrics: %w", err)
        }

        return &fileBodyStoreMetrics{registration: registration}, nil
}

func (m *fileBodyStoreMetrics) close() error {
        m.closeOnce.Do(func() {
                m.closeErr = m.registration.Unregister()
        })

        return m.closeErr
}

package otelrecorder

import (
        "context"
        "fmt"
        "sync"

        "go.opentelemetry.io/otel/attribute"
        "go.opentelemetry.io/otel/metric"

        recorder "github.com/mgurevin/recorder"
)

type samplingMetrics struct {
        registration metric.Registration
        closeOnce    sync.Once
        closeErr     error
}

func newSamplingMetrics(meter metric.Meter, transport *recorder.Transport) (*samplingMetrics, error) {
        head, err := meter.Int64ObservableCounter("recorder.sampling.head.decisions",
                metric.WithUnit("{decision}"),
                metric.WithDescription("Head sampling outcomes by bounded decision"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create head sampling decision counter: %w", err)
        }

        retention, err := meter.Int64ObservableCounter("recorder.sampling.retention.outcomes",
                metric.WithUnit("{decision}"),
                metric.WithDescription("Finalized-entry retention outcomes after managed-asset cleanup"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create retention decision counter: %w", err)
        }

        policyFailures, err := meter.Int64ObservableCounter("recorder.sampling.policy.failures",
                metric.WithUnit("{failure}"),
                metric.WithDescription("Sampling policy panics and invalid decisions by bounded stage and reason"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create sampling policy failure counter: %w", err)
        }

        assetReleaseFailures, err := meter.Int64ObservableCounter("recorder.sampling.asset_release.failures",
                metric.WithUnit("{failure}"),
                metric.WithDescription("Failures releasing managed assets for discarded entries"))
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: create sampling asset release failure counter: %w", err)
        }

        decisionKey := attribute.Key("recorder.sampling.decision")
        stageKey := attribute.Key("recorder.sampling.stage")
        reasonKey := attribute.Key("recorder.sampling.failure.reason")

        registration, err := meter.RegisterCallback(func(_ context.Context, observer metric.Observer) error {
                stats := transport.SamplingStats()
                observer.ObserveInt64(head, int64(stats.HeadFull), metric.WithAttributes(decisionKey.String("full")))
                observer.ObserveInt64(head, int64(stats.HeadMetadataOnly), metric.WithAttributes(decisionKey.String("metadata_only")))
                observer.ObserveInt64(head, int64(stats.HeadDropped), metric.WithAttributes(decisionKey.String("drop")))
                observer.ObserveInt64(retention, int64(stats.Retained), metric.WithAttributes(decisionKey.String("retained")))
                observer.ObserveInt64(retention, int64(stats.Discarded), metric.WithAttributes(decisionKey.String("discarded")))
                observer.ObserveInt64(policyFailures, int64(stats.HeadPanics), metric.WithAttributes(stageKey.String("head"), reasonKey.String("panic")))
                observer.ObserveInt64(policyFailures, int64(stats.HeadInvalidDecisions), metric.WithAttributes(stageKey.String("head"), reasonKey.String("invalid_decision")))
                observer.ObserveInt64(policyFailures, int64(stats.RetentionPanics), metric.WithAttributes(stageKey.String("retention"), reasonKey.String("panic")))
                observer.ObserveInt64(policyFailures, int64(stats.RetentionInvalidDecisions), metric.WithAttributes(stageKey.String("retention"), reasonKey.String("invalid_decision")))
                observer.ObserveInt64(assetReleaseFailures, int64(stats.AssetReleaseFailures))

                return nil
        }, head, retention, policyFailures, assetReleaseFailures)
        if err != nil {
                return nil, fmt.Errorf("otelrecorder: register sampling metrics: %w", err)
        }

        return &samplingMetrics{registration: registration}, nil
}

func (m *samplingMetrics) close() error {
        m.closeOnce.Do(func() {
                m.closeErr = m.registration.Unregister()
        })

        return m.closeErr
}