agent-call/management/backend/stats.go

package management

import (
	"encoding/base64"
	"encoding/json"
	"errors"
	"sort"
	"strconv"
	"strings"
	"time"

	"database/sql"
	"github.com/gofiber/fiber/v3"
)

type statsQuery struct {
	From         *time.Time
	To           *time.Time
	Mode         string
	ProviderID   string
	TrunkID      string
	CellID       string
	EgressPoolID string
}

type metricAccumulator struct {
	Attempts, StartedConfirmed, StartedUncertain, NotStarted, Rejected, Answered, TerminalNotAnswered, PendingNotAnswered, ActiveAnswered, AIFailed int
	TalkSeconds                                                                                                                                     []float64
	WaitSeconds                                                                                                                                     []float64
	FailureReasons                                                                                                                                  map[string]int
	Starts                                                                                                                                          []time.Time
	Answers                                                                                                                                         []time.Time
}

func parseStatsQuery(c fiber.Ctx, serverMode string) (statsQuery, *AppError) {
	mode := strings.ToLower(strings.TrimSpace(c.Query("mode")))
	if mode == "" {
		mode = serverMode
	}
	if mode == "all" {
		mode = "mixed"
	}
	if mode != "mock" && mode != "real" && mode != "mixed" {
		return statsQuery{}, newAppError(422, "INVALID_MODE", "mode must be mock, mixed, or real", nil)
	}
	to := time.Now().UTC()
	from := to.Add(-24 * time.Hour)
	var err error
	if raw := c.Query("from"); raw != "" {
		from, err = time.Parse(time.RFC3339, raw)
		if err != nil {
			return statsQuery{}, newAppError(422, "INVALID_FROM", "from must be RFC3339 UTC", nil)
		}
	}
	if raw := c.Query("to"); raw != "" {
		to, err = time.Parse(time.RFC3339, raw)
		if err != nil {
			return statsQuery{}, newAppError(422, "INVALID_TO", "to must be RFC3339 UTC", nil)
		}
	}
	from = from.UTC()
	to = to.UTC()
	if !to.After(from) {
		return statsQuery{}, newAppError(422, "INVALID_RANGE", "to must be after from", nil)
	}
	if to.Sub(from) > 31*24*time.Hour {
		return statsQuery{}, newAppError(422, "RANGE_TOO_LARGE", "statistics range cannot exceed 31 days", nil)
	}
	return statsQuery{From: &from, To: &to, Mode: mode, ProviderID: c.Query("provider_id"), TrunkID: c.Query("trunk_id"), CellID: c.Query("cell_id"), EgressPoolID: c.Query("egress_pool_id")}, nil
}

func (s *Server) outboundSummary(c fiber.Ctx) error {
	p, authErr := s.authorize(c, "admin", "sip.statistics.read", "", "")
	if authErr != nil {
		return s.fail(c, authErr)
	}
	if err := s.checkFilterResources(c, p); err != nil {
		return s.fail(c, err)
	}
	q, appErr := parseStatsQuery(c, s.cfg.PublicMode())
	if appErr != nil {
		return s.fail(c, appErr)
	}
	ctx, cancel := s.context(c)
	defer cancel()
	attempts, err := s.store.ListAttempts(ctx, AttemptFilter{From: q.From, To: q.To, ProviderID: q.ProviderID, TrunkID: q.TrunkID, CellID: q.CellID, EgressPoolID: q.EgressPoolID, Mode: modeFilter(q.Mode)})
	if err != nil {
		return s.fail(c, asAppError(err))
	}
	timeline, err := s.store.ListAttempts(ctx, AttemptFilter{From: q.From, To: q.To, ProviderID: q.ProviderID, TrunkID: q.TrunkID, CellID: q.CellID, EgressPoolID: q.EgressPoolID, Mode: modeFilter(q.Mode), Timeline: true})
	if err != nil {
		return s.fail(c, asAppError(err))
	}
	now := time.Now().UTC()
	current, err := s.store.ListAttempts(ctx, AttemptFilter{ProviderID: q.ProviderID, TrunkID: q.TrunkID, CellID: q.CellID, EgressPoolID: q.EgressPoolID, Mode: modeFilter(q.Mode), Timeline: true})
	if err != nil {
		return s.fail(c, asAppError(err))
	}
	cpsFrom := now.Truncate(time.Second).Add(-time.Second)
	cpsTo := cpsFrom.Add(time.Second)
	recent, err := s.store.ListAttempts(ctx, AttemptFilter{From: &cpsFrom, To: &cpsTo, ProviderID: q.ProviderID, TrunkID: q.TrunkID, CellID: q.CellID, EgressPoolID: q.EgressPoolID, Mode: modeFilter(q.Mode)})
	if err != nil {
		return s.fail(c, asAppError(err))
	}
	attempts = filterAttemptResources(p, attempts)
	timeline = filterAttemptResources(p, timeline)
	current = filterAttemptResources(p, current)
	recent = filterAttemptResources(p, recent)
	current = activeAttempts(current, now)
	result := summaryResult(q, attempts, timeline, current, recent, now, cpsFrom, cpsTo)
	return c.JSON(result)
}

func filterAttemptResources(p principal, attempts []Attempt) []Attempt {
	if len(p.Resources) == 0 {
		return attempts
	}
	out := attempts[:0]
	for _, attempt := range attempts {
		if resourceAllowed(p.TokenSpec, "provider", attempt.ProviderID) && resourceAllowed(p.TokenSpec, "trunk", attempt.TrunkID) && resourceAllowed(p.TokenSpec, "cell", attempt.CellID) && resourceAllowed(p.TokenSpec, "egress_pool", attempt.EgressPoolID) {
			out = append(out, attempt)
		}
	}
	return out
}

func activeAttempts(attempts []Attempt, asOf time.Time) []Attempt {
	active := make([]Attempt, 0, len(attempts))
	for _, attempt := range attempts {
		if attempt.AnsweredAt != nil && (attempt.EndedAt == nil || attempt.EndedAt.After(asOf)) {
			active = append(active, attempt)
		}
	}
	return active
}

func modeFilter(mode string) string {
	if mode == "mixed" {
		return ""
	}
	return mode
}

func summaryResult(q statsQuery, attempts, timeline, current, recent []Attempt, sampledAt, cpsFrom, cpsTo time.Time) map[string]any {
	m := accumulate(attempts, *q.To)
	complete := m.StartedUncertain == 0 && validFacts(attempts)
	missing := []string{}
	if len(attempts) == 0 {
		missing = append(missing, "attempt_facts")
	}
	if m.StartedUncertain > 0 {
		missing = append(missing, "origin_confirmation")
	}
	if !validFacts(attempts) {
		missing = append(missing, "fact_order")
	}
	dataAsOf := any(nil)
	for _, a := range attempts {
		if !a.UpdatedAt.IsZero() && (dataAsOf == nil || a.UpdatedAt.Format(time.RFC3339Nano) > dataAsOf.(string)) {
			dataAsOf = utcString(a.UpdatedAt)
		}
	}
	var answerRate any
	if m.StartedConfirmed > 0 {
		answerRate = float64(m.Answered) / float64(m.StartedConfirmed)
	}
	realtime := accumulate(current, sampledAt)
	talkStats := durationStats(m.TalkSeconds)
	metrics := map[string]any{"attempts_total": m.Attempts, "started_confirmed": m.StartedConfirmed, "started_uncertain": m.StartedUncertain, "not_started": m.NotStarted, "rejected": m.Rejected, "answered": m.Answered, "terminal_not_answered": m.TerminalNotAnswered, "pending_not_answered": m.PendingNotAnswered, "active_answered": realtime.ActiveAnswered, "ai_failed_after_answer": m.AIFailed, "answer_rate": answerRate, "talk_duration_seconds": talkStats, "acd_seconds": talkStats["avg"], "total_talk_seconds": talkStats["total"], "wait_duration_seconds": durationStats(m.WaitSeconds), "peak_concurrency": peakConcurrency(timeline, *q.From, *q.To), "peak_cps": peakCPS(m.Starts, *q.From, *q.To)}
	return map[string]any{"mode": q.Mode, "from": utcString(*q.From), "to": utcString(*q.To), "timezone": "UTC", "definition_version": "sip-statistics.v1", "filters": map[string]string{"provider_id": q.ProviderID, "trunk_id": q.TrunkID, "cell_id": q.CellID, "egress_pool_id": q.EgressPoolID}, "complete": complete, "missing_sources": missing, "unresolved_count": m.StartedUncertain + m.PendingNotAnswered, "data_as_of": dataAsOf, "generated_at": utcString(time.Now()), "metrics": metrics, "realtime": map[string]any{"current_answered": realtime.ActiveAnswered, "current_cps": peakCPS(accumulate(recent, cpsTo).Starts, cpsFrom, cpsTo), "cps_bucket_start": utcString(cpsFrom), "cps_bucket_end": utcString(cpsTo), "sampled_at": utcString(sampledAt), "complete": validFacts(current) && validFacts(recent), "source": "attempt_facts"}, "failure_reasons": failureReasonList(m.FailureReasons), "semantics": map[string]string{"started_confirmed": "only attempts with a persisted confirmed origin fact", "answered": "answered_at is present and not after ended_at", "terminal_not_answered": "ended_at is present while answered_at is absent", "complete": "false when origin or time-order facts are uncertain"}}
}

func accumulate(attempts []Attempt, to time.Time) metricAccumulator {
	m := metricAccumulator{FailureReasons: map[string]int{}}
	for _, a := range attempts {
		if a.OriginStatus == "rejected" {
			m.Rejected++
			continue
		}
		m.Attempts++
		switch a.OriginStatus {
		case "confirmed":
			m.StartedConfirmed++
		case "uncertain":
			m.StartedUncertain++
		case "not_started":
			m.NotStarted++
		}
		if a.AttemptStartedAt != nil {
			m.Starts = append(m.Starts, *a.AttemptStartedAt)
		}
		if a.AnsweredAt != nil {
			m.Answers = append(m.Answers, *a.AnsweredAt)
			m.Answered++
			if a.EndedAt == nil || a.EndedAt.After(to) {
				m.ActiveAnswered++
			}
		} else if a.EndedAt != nil {
			m.TerminalNotAnswered++
			reason := strings.ToLower(strings.TrimSpace(a.TerminationReason))
			if reason == "" {
				reason = "unknown"
			}
			m.FailureReasons[reason]++
		} else if a.OriginStatus == "confirmed" || a.OriginStatus == "uncertain" {
			m.PendingNotAnswered++
		}
		if a.AIStatus == "failed" && a.AnsweredAt != nil {
			m.AIFailed++
		}
		if a.AnsweredAt != nil && a.EndedAt != nil && !a.EndedAt.Before(*a.AnsweredAt) {
			m.TalkSeconds = append(m.TalkSeconds, a.EndedAt.Sub(*a.AnsweredAt).Seconds())
			if a.AttemptStartedAt != nil && !a.AnsweredAt.Before(*a.AttemptStartedAt) {
				m.WaitSeconds = append(m.WaitSeconds, a.AnsweredAt.Sub(*a.AttemptStartedAt).Seconds())
			}
		}
	}
	return m
}
func validFacts(attempts []Attempt) bool {
	for _, a := range attempts {
		if a.AnsweredAt != nil && a.AttemptStartedAt != nil && a.AnsweredAt.Before(*a.AttemptStartedAt) {
			return false
		}
		if a.EndedAt != nil && a.AnsweredAt != nil && a.EndedAt.Before(*a.AnsweredAt) {
			return false
		}
	}
	return true
}
func durationStats(values []float64) map[string]any {
	if len(values) == 0 {
		return map[string]any{"count": 0, "total": 0, "p50": nil, "p95": nil, "avg": nil}
	}
	sort.Float64s(values)
	sum := 0.0
	for _, v := range values {
		sum += v
	}
	return map[string]any{"count": len(values), "total": sum, "p50": percentile(values, .50), "p95": percentile(values, .95), "avg": sum / float64(len(values))}
}
func percentile(v []float64, p float64) float64 {
	if len(v) == 0 {
		return 0
	}
	i := int(float64(len(v)-1)*p + 0.5)
	if i < 0 {
		i = 0
	}
	if i >= len(v) {
		i = len(v) - 1
	}
	return v[i]
}
func failureReasonList(values map[string]int) []map[string]any {
	keys := make([]string, 0, len(values))
	for k := range values {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	out := make([]map[string]any, 0, len(keys))
	for _, k := range keys {
		out = append(out, map[string]any{"reason": k, "count": values[k]})
	}
	return out
}

func peakConcurrency(attempts []Attempt, from, to time.Time) int {
	type event struct {
		t     time.Time
		delta int
	}
	events := []event{}
	for _, a := range attempts {
		if a.AnsweredAt == nil {
			continue
		}
		start := *a.AnsweredAt
		if start.Before(from) {
			start = from
		}
		if !start.Before(to) {
			continue
		}
		end := to
		if a.EndedAt != nil && a.EndedAt.Before(end) {
			end = *a.EndedAt
		}
		if !end.After(start) {
			continue
		}
		events = append(events, event{start, 1}, event{end, -1})
	}
	sort.Slice(events, func(i, j int) bool {
		if events[i].t.Equal(events[j].t) {
			return events[i].delta < events[j].delta
		}
		return events[i].t.Before(events[j].t)
	})
	current, max := 0, 0
	for _, e := range events {
		current += e.delta
		if current > max {
			max = current
		}
	}
	return max
}
func peakCPS(starts []time.Time, from, to time.Time) int {
	counts := map[string]int{}
	for _, t := range starts {
		if t.Before(from) || !t.Before(to) {
			continue
		}
		key := t.UTC().Truncate(time.Second).Format(time.RFC3339)
		counts[key]++
	}
	max := 0
	for _, v := range counts {
		if v > max {
			max = v
		}
	}
	return max
}

func (s *Server) outboundTimeseries(c fiber.Ctx) error {
	p, authErr := s.authorize(c, "admin", "sip.statistics.read", "", "")
	if authErr != nil {
		return s.fail(c, authErr)
	}
	if err := s.checkFilterResources(c, p); err != nil {
		return s.fail(c, err)
	}
	q, appErr := parseStatsQuery(c, s.cfg.PublicMode())
	if appErr != nil {
		return s.fail(c, appErr)
	}
	gran := strings.ToLower(c.Query("granularity"))
	if gran == "" {
		gran = "hour"
	}
	step := time.Hour
	switch gran {
	case "minute":
		step = time.Minute
		if q.To.Sub(*q.From) > 24*time.Hour {
			return s.fail(c, newAppError(422, "GRANULARITY_TOO_FINE", "minute series cannot exceed 24 hours", nil))
		}
	case "hour":
		step = time.Hour
	case "day":
		step = 24 * time.Hour
	default:
		return s.fail(c, newAppError(422, "INVALID_GRANULARITY", "granularity must be minute, hour, or day", nil))
	}
	ctx, cancel := s.context(c)
	defer cancel()
	attempts, err := s.store.ListAttempts(ctx, AttemptFilter{From: q.From, To: q.To, ProviderID: q.ProviderID, TrunkID: q.TrunkID, CellID: q.CellID, EgressPoolID: q.EgressPoolID, Mode: modeFilter(q.Mode)})
	if err != nil {
		return s.fail(c, asAppError(err))
	}
	timeline, err := s.store.ListAttempts(ctx, AttemptFilter{From: q.From, To: q.To, ProviderID: q.ProviderID, TrunkID: q.TrunkID, CellID: q.CellID, EgressPoolID: q.EgressPoolID, Mode: modeFilter(q.Mode), Timeline: true})
	if err != nil {
		return s.fail(c, asAppError(err))
	}
	attempts = filterAttemptResources(p, attempts)
	timeline = filterAttemptResources(p, timeline)
	buckets := make([]map[string]any, 0)
	start := q.From.Truncate(step)
	for bucket := start; bucket.Before(*q.To); bucket = bucket.Add(step) {
		end := bucket.Add(step)
		if end.After(*q.To) {
			end = *q.To
		}
		windowFrom := bucket
		if windowFrom.Before(*q.From) {
			windowFrom = *q.From
		}
		var subset []Attempt
		for _, a := range attempts {
			if a.AttemptStartedAt != nil && !a.AttemptStartedAt.Before(windowFrom) && a.AttemptStartedAt.Before(end) {
				subset = append(subset, a)
			}
		}
		m := accumulate(subset, end)
		buckets = append(buckets, map[string]any{"bucket_start": utcString(bucket), "bucket_end": utcString(end), "attempts_total": m.Attempts, "started_confirmed": m.StartedConfirmed, "started_uncertain": m.StartedUncertain, "answered": m.Answered, "terminal_not_answered": m.TerminalNotAnswered, "pending_not_answered": m.PendingNotAnswered, "peak_concurrency": peakConcurrency(timelineForBucket(timeline, windowFrom, end), windowFrom, end), "peak_cps": peakCPS(m.Starts, windowFrom, end)})
	}
	return c.JSON(map[string]any{"mode": q.Mode, "from": utcString(*q.From), "to": utcString(*q.To), "timezone": "UTC", "definition_version": "sip-statistics.v1", "filters": map[string]string{"provider_id": q.ProviderID, "trunk_id": q.TrunkID, "cell_id": q.CellID, "egress_pool_id": q.EgressPoolID}, "granularity": gran, "complete": validFacts(attempts) && countUncertain(attempts) == 0, "series": buckets})
}
func timelineForBucket(attempts []Attempt, from, to time.Time) []Attempt {
	out := make([]Attempt, 0, len(attempts))
	for _, attempt := range attempts {
		if attempt.AnsweredAt != nil && attempt.AnsweredAt.Before(to) && (attempt.EndedAt == nil || attempt.EndedAt.After(from)) {
			out = append(out, attempt)
		}
	}
	return out
}

func countUncertain(v []Attempt) int {
	n := 0
	for _, a := range v {
		if a.OriginStatus == "uncertain" {
			n++
		}
	}
	return n
}

func (s *Server) callAttempts(c fiber.Ctx) error {
	p, authErr := s.authorize(c, "admin", "sip.calls.read", "", "")
	if authErr != nil {
		return s.fail(c, authErr)
	}
	if err := s.checkFilterResources(c, p); err != nil {
		return s.fail(c, err)
	}
	q, appErr := parseStatsQuery(c, s.cfg.PublicMode())
	if appErr != nil {
		return s.fail(c, appErr)
	}
	limit := 50
	if raw := c.Query("limit"); raw != "" {
		n, err := strconv.Atoi(raw)
		if err != nil || n < 1 || n > 200 {
			return s.fail(c, newAppError(422, "INVALID_LIMIT", "limit must be between 1 and 200", nil))
		}
		limit = n
	}
	ctx, cancel := s.context(c)
	defer cancel()
	attempts, err := s.store.ListAttempts(ctx, AttemptFilter{From: q.From, To: q.To, ProviderID: q.ProviderID, TrunkID: q.TrunkID, CellID: q.CellID, EgressPoolID: q.EgressPoolID, Mode: modeFilter(q.Mode), Limit: limit + 1})
	if err != nil {
		return s.fail(c, asAppError(err))
	}
	attempts = filterAttemptResources(p, attempts)
	fingerprint := statsFingerprint(q)
	cursor := c.Query("cursor")
	if cursor != "" {
		last, err := decodeCursor(cursor)
		if err != nil || last.Filter != fingerprint {
			return s.fail(c, newAppError(400, "INVALID_CURSOR", "cursor is invalid for the current filters", nil))
		}
		filtered := attempts[:0]
		for _, a := range attempts {
			current := cursorTime(a)
			if current.After(last.Time) || (current.Equal(last.Time) && a.AttemptID > last.ID) {
				filtered = append(filtered, a)
			}
		}
		attempts = filtered
	}
	next := ""
	if len(attempts) > limit {
		last := attempts[limit-1]
		next = encodeCursor(last, fingerprint)
		attempts = attempts[:limit]
	}
	out := make([]map[string]any, 0, len(attempts))
	for _, a := range attempts {
		out = append(out, attemptView(a, false))
	}
	return c.JSON(map[string]any{"mode": q.Mode, "from": utcString(*q.From), "to": utcString(*q.To), "complete": validFacts(attempts) && countUncertain(attempts) == 0, "attempts": out, "next_cursor": next})
}

type cursorValue struct {
	Time       time.Time
	ID, Filter string
}

func cursorTime(a Attempt) time.Time {
	if a.AttemptStartedAt != nil {
		return *a.AttemptStartedAt
	}
	return a.UpdatedAt
}
func statsFingerprint(q statsQuery) string {
	b, _ := json.Marshal(map[string]any{"from": q.From, "to": q.To, "mode": q.Mode, "provider_id": q.ProviderID, "trunk_id": q.TrunkID, "cell_id": q.CellID, "egress_pool_id": q.EgressPoolID})
	return hashBytes(b)
}
func encodeCursor(a Attempt, filter string) string {
	b, _ := json.Marshal(map[string]string{"time": utcString(cursorTime(a)), "id": a.AttemptID, "filter": filter})
	return base64.RawURLEncoding.EncodeToString(b)
}
func decodeCursor(raw string) (cursorValue, error) {
	b, err := base64.RawURLEncoding.DecodeString(raw)
	if err != nil {
		return cursorValue{}, err
	}
	var v struct {
		Time   string `json:"time"`
		ID     string `json:"id"`
		Filter string `json:"filter"`
	}
	if err := json.Unmarshal(b, &v); err != nil {
		return cursorValue{}, err
	}
	t, err := time.Parse(time.RFC3339Nano, v.Time)
	return cursorValue{Time: t, ID: v.ID, Filter: v.Filter}, err
}
func attemptView(a Attempt, detail bool) map[string]any {
	out := map[string]any{"attempt_id": a.AttemptID, "execution_id": a.ExecutionID, "call_id": a.CallID, "tenant_key": a.TenantKey, "provider_id": a.ProviderID, "trunk_id": a.TrunkID, "cell_id": a.CellID, "egress_pool_id": a.EgressPoolID, "config_revision": a.ConfigRevision, "mode": a.Mode, "dialed_number": maskNumber(a.DialedNumber), "origin_status": a.OriginStatus, "termination_reason": a.TerminationReason, "sip_code": a.SIPCode, "q850": a.Q850, "ai_status": a.AIStatus, "source": a.Source, "fact_version": a.FactVersion, "updated_at": utcString(a.UpdatedAt)}
	if a.AttemptStartedAt != nil {
		out["attempt_started_at"] = utcString(*a.AttemptStartedAt)
	} else {
		out["attempt_started_at"] = nil
	}
	if a.AnsweredAt != nil {
		out["answered_at"] = utcString(*a.AnsweredAt)
	} else {
		out["answered_at"] = nil
	}
	if a.EndedAt != nil {
		out["ended_at"] = utcString(*a.EndedAt)
	} else {
		out["ended_at"] = nil
	}
	if !detail {
		delete(out, "execution_id")
		delete(out, "call_id")
	}
	return out
}
func (s *Server) callAttempt(c fiber.Ctx) error {
	id := c.Params("attempt_id")
	p, authErr := s.authorize(c, "admin", "sip.calls.read", "", "")
	if authErr != nil {
		return s.fail(c, authErr)
	}
	_ = p
	ctx, cancel := s.context(c)
	defer cancel()
	a, err := s.store.GetAttempt(ctx, id)
	if errors.Is(err, sql.ErrNoRows) {
		return s.fail(c, newAppError(404, "ATTEMPT_NOT_FOUND", "call attempt does not exist", nil))
	}
	if err != nil {
		return s.fail(c, asAppError(err))
	}
	return c.JSON(map[string]any{"mode": s.cfg.PublicMode(), "attempt": attemptView(a, true)})
}

func (s *Server) audit(c fiber.Ctx) error {
	p, authErr := s.authorize(c, "admin", "sip.audit.read", "", "")
	if authErr != nil {
		return s.fail(c, authErr)
	}
	if err := s.checkFilterResources(c, p); err != nil {
		return s.fail(c, err)
	}
	limit := 50
	if raw := c.Query("limit"); raw != "" {
		n, err := strconv.Atoi(raw)
		if err != nil || n < 1 || n > 200 {
			return s.fail(c, newAppError(422, "INVALID_LIMIT", "limit must be between 1 and 200", nil))
		}
		limit = n
	}
	ctx, cancel := s.context(c)
	defer cancel()
	rows, err := s.store.ListAudit(ctx, c.Query("resource_id"), c.Query("request_id"), c.Query("actor"), limit)
	if err != nil {
		return s.fail(c, asAppError(err))
	}
	visible := rows[:0]
	for _, row := range rows {
		if resourceAllowed(p.TokenSpec, row.ResourceType, row.ResourceID) {
			visible = append(visible, row)
		}
	}
	return c.JSON(map[string]any{"mode": s.cfg.PublicMode(), "audit": visible})
}