telephony-inspector/internal/sip/callflow.go

package sip

import (
	"sync"
	"time"
)

// CallFlow represents a SIP call with all its packets
type CallFlow struct {
	CallID    string
	Packets   []*Packet
	StartTime time.Time
	EndTime   time.Time

	// Summary info
	From  string
	To    string
	State CallState
}

// CallState represents the current state of a call
type CallState string

const (
	CallStateInitial    CallState = "Initial"
	CallStateRinging    CallState = "Ringing"
	CallStateConnected  CallState = "Connected"
	CallStateTerminated CallState = "Terminated"
	CallStateFailed     CallState = "Failed"
)

// CallFlowStore stores and manages call flows
type CallFlowStore struct {
	mu    sync.RWMutex
	flows map[string]*CallFlow
}

// NewCallFlowStore creates a new call flow store
func NewCallFlowStore() *CallFlowStore {
	return &CallFlowStore{
		flows: make(map[string]*CallFlow),
	}
}

// AddPacket adds a packet to the appropriate call flow
func (s *CallFlowStore) AddPacket(p *Packet) *CallFlow {
	s.mu.Lock()
	defer s.mu.Unlock()

	if p.CallID == "" {
		return nil
	}

	flow, exists := s.flows[p.CallID]

	// Determine timestamp to use
	ts := p.Timestamp
	if ts.IsZero() {
		ts = time.Now()
	}

	if !exists {
		flow = &CallFlow{
			CallID:    p.CallID,
			Packets:   make([]*Packet, 0),
			StartTime: ts,
			From:      p.From,
			To:        p.To,
			State:     CallStateInitial,
		}
		s.flows[p.CallID] = flow
	}

	flow.Packets = append(flow.Packets, p)
	// Always update EndTime to the latest packet's timestamp
	if ts.After(flow.EndTime) {
		flow.EndTime = ts
	}

	// Update call state based on packet
	s.updateState(flow, p)

	return flow
}

// updateState updates the call state based on the packet
func (s *CallFlowStore) updateState(flow *CallFlow, p *Packet) {
	if p.IsRequest {
		switch p.Method {
		case MethodINVITE:
			if flow.State == CallStateInitial {
				flow.State = CallStateInitial
			}
		case MethodBYE, MethodCANCEL:
			flow.State = CallStateTerminated
		}
	} else {
		// Response
		switch {
		case p.StatusCode >= 100 && p.StatusCode < 200:
			if p.StatusCode == 180 || p.StatusCode == 183 {
				flow.State = CallStateRinging
			}
		case p.StatusCode >= 200 && p.StatusCode < 300:
			flow.State = CallStateConnected
		case p.StatusCode >= 400:
			flow.State = CallStateFailed
		}
	}
}

// GetFlow returns a call flow by Call-ID
func (s *CallFlowStore) GetFlow(callID string) *CallFlow {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.flows[callID]
}

// GetAllFlows returns all call flows
func (s *CallFlowStore) GetAllFlows() []*CallFlow {
	s.mu.RLock()
	defer s.mu.RUnlock()

	flows := make([]*CallFlow, 0, len(s.flows))
	for _, f := range s.flows {
		flows = append(flows, f)
	}
	return flows
}

// GetRecentFlows returns the N most recent call flows sorted by StartTime (oldest first)
func (s *CallFlowStore) GetRecentFlows(n int) []*CallFlow {
	flows := s.GetAllFlows()

	// Sort by start time ascending (oldest first), then by CallID for stable order
	for i := 0; i < len(flows)-1; i++ {
		for j := i + 1; j < len(flows); j++ {
			// Compare by StartTime first
			if flows[i].StartTime.After(flows[j].StartTime) {
				flows[i], flows[j] = flows[j], flows[i]
			} else if flows[i].StartTime.Equal(flows[j].StartTime) {
				// If same time, sort by CallID for stable order
				if flows[i].CallID > flows[j].CallID {
					flows[i], flows[j] = flows[j], flows[i]
				}
			}
		}
	}

	if len(flows) > n {
		flows = flows[:n]
	}
	return flows
}

// GetSortedFlows returns all call flows sorted by StartTime (oldest first)
func (s *CallFlowStore) GetSortedFlows() []*CallFlow {
	flows := s.GetAllFlows()

	// Sort by start time ascending (oldest first), then by CallID for stable order
	for i := 0; i < len(flows)-1; i++ {
		for j := i + 1; j < len(flows); j++ {
			// Compare by StartTime first
			if flows[i].StartTime.After(flows[j].StartTime) {
				flows[i], flows[j] = flows[j], flows[i]
			} else if flows[i].StartTime.Equal(flows[j].StartTime) {
				// If same time, sort by CallID for stable order
				if flows[i].CallID > flows[j].CallID {
					flows[i], flows[j] = flows[j], flows[i]
				}
			}
		}
	}
	return flows
}

// Count returns the number of call flows
func (s *CallFlowStore) Count() int {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return len(s.flows)
}

// Summary returns a string summary of a packet for display
func (p *Packet) Summary() string {
	if p.IsRequest {
		return string(p.Method)
	}
	return formatStatusCode(p.StatusCode, p.StatusText)
}

func formatStatusCode(code int, text string) string {
	if text != "" {
		return text
	}
	switch code {
	case 100:
		return "100 Trying"
	case 180:
		return "180 Ringing"
	case 183:
		return "183 Session Progress"
	case 200:
		return "200 OK"
	case 400:
		return "400 Bad Request"
	case 401:
		return "401 Unauthorized"
	case 403:
		return "403 Forbidden"
	case 404:
		return "404 Not Found"
	case 408:
		return "408 Request Timeout"
	case 480:
		return "480 Temporarily Unavailable"
	case 486:
		return "486 Busy Here"
	case 487:
		return "487 Request Terminated"
	case 488:
		return "488 Not Acceptable Here"
	case 500:
		return "500 Server Internal Error"
	case 503:
		return "503 Service Unavailable"
	default:
		return string(rune('0'+code/100)) + "xx"
	}
}