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}) }