go-sip/internal/logging/http.go

package logging

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"mime"
	"net/http"
	"strings"
	"sync"
	"time"
)

// HTTPClient clones the client, retaining its TLS, redirect, timeout and custom
// transport settings. Only bytes actually read at the boundary are observed.
func HTTPClient(client *http.Client, local, remote string) *http.Client {
	if client == nil {
		client = http.DefaultClient
	}
	clone := *client
	transport := client.Transport
	if transport == nil {
		transport = http.DefaultTransport
	}
	if observed, ok := transport.(*httpObserver); ok {
		transport = observed.next
	}
	clone.Transport = &httpObserver{next: transport, local: local, remote: remote}
	return &clone
}

// ObserveDefaultHTTP covers library-owned HTTP clients (including native ARI)
// whose supported API does not expose a transport. Call only at service startup.
func ObserveDefaultHTTP(local string) {
	next := http.DefaultTransport
	if observed, ok := next.(*httpObserver); ok {
		next = observed.next
	}
	http.DefaultTransport = &httpObserver{next: next, local: local}
}

// CloneDefaultTransport retains explicit transport settings used by isolated
// TLS tests. A custom non-HTTP transport is an error, not a silent fallback.
func CloneDefaultTransport() (*http.Transport, error) {
	base := http.DefaultTransport
	if observed, ok := base.(*httpObserver); ok {
		base = observed.next
	}
	transport, ok := base.(*http.Transport)
	if !ok {
		return nil, fmt.Errorf("default HTTP transport cannot be cloned: %T", base)
	}
	return transport.Clone(), nil
}

type httpObserver struct {
	next          http.RoundTripper
	local, remote string
}

func (t *httpObserver) RoundTrip(req *http.Request) (*http.Response, error) {
	started := time.Now()
	remote := t.remote
	if remote == "" {
		remote = req.URL.Host
		if remote == "127.0.0.1:8088" {
			remote = "asterisk"
		}
	}
	attrs := []any{"transport", "http", "method", req.Method, "url", req.URL.String()}
	ctx := req.Context()
	slog.InfoContext(ctx, "HTTP request", append(attrs, "direction", t.local+"->"+remote, "stage", "request_start")...)
	clone := req.Clone(ctx)
	if req.Body != nil {
		clone.Body = newObservedBody(ctx, req.Body, req.Header.Get("Content-Type"), "HTTP request data", append(attrs, "direction", t.local+"->"+remote)...)
	}
	res, err := t.next.RoundTrip(clone)
	result := append(attrs, "direction", remote+"->"+t.local, "duration_ms", time.Since(started).Milliseconds())
	if err != nil {
		slog.ErrorContext(ctx, "HTTP failed", append(result, "stage", "transport_failed", "error", err)...)
		return res, err
	}
	slog.InfoContext(ctx, "HTTP response", append(result, "stage", "response_received", "http_status", res.StatusCode, "content_type", res.Header.Get("Content-Type"))...)
	// HTTP upgrades require the original read/write stream (ASR/TTS WebSocket).
	if res.StatusCode == http.StatusSwitchingProtocols {
		return res, nil
	}
	if res.Body != nil {
		res.Body = newObservedBody(ctx, res.Body, res.Header.Get("Content-Type"), "HTTP response data", append(result, "http_status", res.StatusCode)...)
	}
	return res, nil
}

type observedBody struct {
	source  io.ReadCloser
	ctx     context.Context
	message string
	attrs   []any
	collect bool
	mu      sync.Mutex
	data    bytes.Buffer
	count   int64
	done    bool
	started time.Time
}

func newObservedBody(ctx context.Context, source io.ReadCloser, contentType, message string, attrs ...any) *observedBody {
	media, _, err := mime.ParseMediaType(contentType)
	isJSON := err == nil && (media == "application/json" || strings.HasSuffix(media, "+json"))
	mode := "binary_omitted"
	if isJSON {
		mode = "json"
	}
	return &observedBody{source: source, ctx: ctx, message: message, attrs: append(append([]any{}, attrs...), "content_type", contentType, "body_mode", mode), collect: isJSON && slog.Default().Enabled(ctx, slog.LevelDebug), started: time.Now()}
}
func (r *observedBody) Read(p []byte) (int, error) {
	n, err := r.source.Read(p)
	r.mu.Lock()
	defer r.mu.Unlock()
	r.count += int64(n)
	if r.collect && !r.done {
		r.data.Write(p[:n])
	}
	if errors.Is(err, io.EOF) {
		r.finishLocked(true, nil)
	} else if err != nil {
		r.finishLocked(false, err)
	}
	return n, err
}
func (r *observedBody) Close() error {
	err := r.source.Close()
	r.mu.Lock()
	defer r.mu.Unlock()
	r.finishLocked(false, err)
	if err != nil {
		slog.ErrorContext(r.ctx, "HTTP body close failed", append(r.attrs, "stage", "body_close_failed", "error", err)...)
	}
	return err
}
func (r *observedBody) finishLocked(complete bool, err error) {
	if r.done {
		return
	}
	r.done = true
	stage := "body_read"
	if !complete {
		stage = "body_read_incomplete"
	}
	attrs := append(append([]any{}, r.attrs...), "stage", stage, "bytes_read", r.count, "body_duration_ms", time.Since(r.started).Milliseconds())
	if err != nil {
		slog.ErrorContext(r.ctx, "HTTP body read failed", append(attrs, "error", err)...)
	} else {
		slog.InfoContext(r.ctx, "HTTP body observation", attrs...)
	}
	if r.collect {
		// A decoder may close after consuming a valid JSON value without reading
		// EOF. Log observed bytes without claiming a complete wire body; malformed
		// prefixes still produce an explicit safe redaction diagnostic.
		Body(r.ctx, r.message, r.data.Bytes(), append(attrs, "body_complete", complete)...)
	}
	r.data.Reset()
}