pulse-lets-go/internal/esl/client.go

// Package esl реализует клиент FreeSWITCH Event Socket Library (ESL).
// Использует сырой TCP — протокол ESL прост (HTTP-подобные заголовки +
// Content-Length тело), не нужна внешняя библиотека.
//
// Клиент обеспечивает:
//   - Подключение и аутентификацию
//   - Синхронные команды (api)
//   - Подписку на события
//   - Автоматический реконнект с exponential backoff
//   - Потокобезопасность (mutex на отправку команд)
package esl

import (
	"bufio"
	"fmt"
	"io"
	"log"
	"net"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

// Event — событие, полученное от FreeSWITCH ESL.
type Event struct {
	Name    string
	Headers map[string]string
	Body    string
}

// Client — ESL-клиент для управления FreeSWITCH.
type Client struct {
	host     string
	port     int
	password string

	conn   net.Conn
	reader *bufio.Reader
	mu     sync.Mutex // защищает отправку синхронных команд (auth, subscribe)
	sendMu sync.Mutex // защищает запись в сокет (async gateway commands)

	connected atomic.Bool
	eventsCh  chan *Event
	closeCh   chan struct{}

	// Переподключение
	shouldReconn atomic.Bool
	backoff      time.Duration

	// Статистика
	reconnects atomic.Int64
	gatewayOps atomic.Int64
	eventsRecv atomic.Int64
	startTime  time.Time
}

// NewClient создаёт нового ESL-клиента.
func NewClient(host string, port int, password string) *Client {
	return &Client{
		host:     host,
		port:     port,
		password: password,
		eventsCh: make(chan *Event, 100),
		closeCh:  make(chan struct{}),
		backoff:  1 * time.Second,
	}
}

// Connect подключается к FreeSWITCH и запускает фоновый реконнект.
// Блокируется до первой успешной аутентификации.
func (c *Client) Connect() error {
	c.shouldReconn.Store(true)
	if err := c.tryConnect(); err != nil {
		return fmt.Errorf("esl connect: %w", err)
	}
	return nil
}

// ConnectWithRetry подключается в фоне с бесконечными попытками.
func (c *Client) ConnectWithRetry() {
	c.shouldReconn.Store(true)
	for {
		err := c.tryConnect()
		if err == nil {
			return // успех
		}
		if !c.shouldReconn.Load() {
			return // остановка
		}
		log.Printf("[esl] ошибка подключения: %v, повтор через %v", err, c.backoff)
		time.Sleep(c.backoff)
		c.backoff *= 2
		if c.backoff > 60*time.Second {
			c.backoff = 60 * time.Second
		}
	}
}

// tryConnect выполняет одну попытку подключения и аутентификации.
func (c *Client) tryConnect() error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Закрываем предыдущее соединение если было
	if c.conn != nil {
		c.conn.Close()
		c.closeCh = make(chan struct{})
	}

	addr := net.JoinHostPort(c.host, fmt.Sprintf("%d", c.port))
	dialer := net.Dialer{Timeout: 5 * time.Second}
	conn, err := dialer.Dial("tcp", addr)
	if err != nil {
		return fmt.Errorf("dial %s: %w", addr, err)
	}
	c.conn = conn
	c.reader = bufio.NewReader(conn)

	// 1. Читаем auth/request
	headers, _, err := c.readMessage()
	if err != nil {
		conn.Close()
		return fmt.Errorf("чтение auth/request: %w", err)
	}
	if headers["Content-Type"] != "auth/request" {
		conn.Close()
		return fmt.Errorf("ожидался auth/request, получен %s", headers["Content-Type"])
	}

	// 2. Отправляем пароль
	authCmd := fmt.Sprintf("auth %s\n\n", c.password)
	if _, err := conn.Write([]byte(authCmd)); err != nil {
		conn.Close()
		return fmt.Errorf("отправка auth: %w", err)
	}

	// 3. Читаем auth response
	headers, _, err = c.readMessage()
	if err != nil {
		conn.Close()
		return fmt.Errorf("чтение auth response: %w", err)
	}
	if reply := headers["Reply-Text"]; !strings.HasPrefix(reply, "+OK") {
		conn.Close()
		return fmt.Errorf("auth rejected: %s", reply)
	}

	c.connected.Store(true)
	c.startTime = time.Now()
	c.backoff = 1 * time.Second // сброс backoff после успешного коннекта

	endpoint := fmt.Sprintf("%s:%d", c.host, c.port)
	log.Printf("[esl] подключён к %s", endpoint)

	return nil
}

// Disconnect отключает клиент и останавливает реконнект.
func (c *Client) Disconnect() {
	c.shouldReconn.Store(false)
	c.mu.Lock()
	c.connected.Store(false)
	if c.conn != nil {
		c.conn.Close()
		c.conn = nil
	}
	c.mu.Unlock()
	// Безопасное закрытие closeCh (может быть уже закрыт)
	select {
	case <-c.closeCh:
		// уже закрыт
	default:
		close(c.closeCh)
	}
}

// IsConnected возвращает статус подключения.
func (c *Client) IsConnected() bool {
	return c.connected.Load()
}

// Send отправляет синхронную команду и возвращает тело ответа.
// Пример: Send("api sofia profile external gwlist") → список gateway.
func (c *Client) Send(command string) (map[string]string, string, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if !c.connected.Load() || c.conn == nil {
		return nil, "", fmt.Errorf("esl не подключён")
	}

	// Отправляем команду
	fullCmd := command + "\n\n"
	if _, err := c.conn.Write([]byte(fullCmd)); err != nil {
		c.connected.Store(false)
		return nil, "", fmt.Errorf("отправка команды: %w", err)
	}

	// Читаем ответ
	return c.readMessage()
}

// Subscribe подписывается на события ESL.
// events — имена событий для фильтра: "SOFIA::gateway_register SOFIA::gateway_unregister".
func (c *Client) Subscribe(events string) error {
	cmd := fmt.Sprintf("event plain %s", events)
	_, _, err := c.Send(cmd)
	if err != nil {
		return fmt.Errorf("подписка на события: %w", err)
	}
	log.Printf("[esl] подписан на события: %s", events)
	return nil
}

// Events возвращает канал с входящими событиями.
func (c *Client) Events() <-chan *Event {
	return c.eventsCh
}

// StartReadingEvents запускает горутину чтения асинхронных событий FS.
// Должна вызываться после Connect().
func (c *Client) StartReadingEvents() {
	if !c.shouldReconn.Load() {
		return
	}
	go c.readEventsLoop()
}

// SendAsync отправляет асинхронную команду (bgapi) без ожидания ответа.
// Ответ придёт как BACKGROUND_JOB событие в readEventsLoop.
// Использует sendMu для защиты записи в сокет.
func (c *Client) SendAsync(command string) error {
	c.sendMu.Lock()
	defer c.sendMu.Unlock()

	if !c.connected.Load() || c.conn == nil {
		return fmt.Errorf("esl не подключён")
	}

	fullCmd := command + "\n\n"
	_, err := c.conn.Write([]byte(fullCmd))
	if err != nil {
		return fmt.Errorf("отправка команды: %w", err)
	}
	return nil
}

// readEventsLoop читает асинхронные сообщения (события) от FS.
// Не держит мьютекс во время блокирующего чтения — синхронные команды
// (Send) используют c.mu, асинхронные (SendAsync) используют sendMu.
func (c *Client) readEventsLoop() {
	for {
		select {
		case <-c.closeCh:
			return
		default:
		}

		if !c.connected.Load() || c.conn == nil {
			time.Sleep(500 * time.Millisecond)
			continue
		}

		headers, body, err := c.readMessage()

		if err != nil {
			if err == io.EOF || strings.Contains(err.Error(), "use of closed network connection") {
				c.connected.Store(false)
				log.Printf("[esl] соединение разорвано, реконнект...")
				c.reconnects.Add(1)
				c.reconnect()
				continue
			}
			log.Printf("[esl] ошибка чтения события: %v", err)
			time.Sleep(100 * time.Millisecond)
			continue
		}

		eventName := headers["Event-Name"]
		if eventName == "" {
			continue // не событие (например, ответ на команду)
		}

		evt := &Event{
			Name:    eventName,
			Headers: headers,
			Body:    body,
		}
		c.eventsRecv.Add(1)

		select {
		case c.eventsCh <- evt:
		default:
			log.Printf("[esl] канал событий переполнен, пропускаем %s", eventName)
		}
	}
}

// reconnect выполняет реконнект с backoff.
func (c *Client) reconnect() {
	if !c.shouldReconn.Load() {
		return
	}
	log.Printf("[esl] реконнект через %v...", c.backoff)
	time.Sleep(c.backoff)

	c.mu.Lock()
	if c.conn != nil {
		c.conn.Close()
		c.conn = nil
	}
	c.mu.Unlock()

	if err := c.tryConnect(); err != nil {
		log.Printf("[esl] реконнект не удался: %v", err)
		c.backoff *= 2
		if c.backoff > 60*time.Second {
			c.backoff = 60 * time.Second
		}
		return
	}

	// Успешный реконнект
	c.backoff = 1 * time.Second
	log.Printf("[esl] реконнект успешен, переподписываемся на события")
	// TODO: gatewaySyncAll() и resubscribe на события вызываются снаружи
}

// --- Приватные методы ---

// readMessage читает одно ESL-сообщение: заголовки + тело.
// Вызывающий должен гарантировать монопольный доступ к c.reader.
func (c *Client) readMessage() (map[string]string, string, error) {
	headers := make(map[string]string)

	for {
		line, err := c.reader.ReadString('\n')
		if err != nil {
			return nil, "", err
		}
		line = strings.TrimRight(line, "\r\n")
		if line == "" {
			break // конец заголовков
		}

		parts := strings.SplitN(line, ":", 2)
		if len(parts) == 2 {
			headers[strings.TrimSpace(parts[0])] = strings.TrimSpace(parts[1])
		}
	}

	// Читаем тело согласно Content-Length
	cl := headers["Content-Length"]
	if cl == "" {
		return headers, "", nil
	}
	length, err := strconv.Atoi(cl)
	if err != nil {
		return headers, "", fmt.Errorf("некорректный Content-Length: %s", cl)
	}

	body := make([]byte, length)
	if _, err := io.ReadFull(c.reader, body); err != nil {
		return headers, "", fmt.Errorf("чтение тела (Content-Length=%d): %w", length, err)
	}

	return headers, strings.TrimSpace(string(body)), nil
}

// --- Метрики ---

// Stats возвращает статистику ESL-клиента для health endpoint.
type Stats struct {
	Status     string `json:"status"`
	Host       string `json:"host"`
	Reconnects int64  `json:"reconnects"`
	GatewayOps int64  `json:"gateway_ops_total"`
	EventsRecv int64  `json:"events_received"`
	UptimeSec  int64  `json:"uptime_seconds"`
}

// GetStats возвращает текущую статистику.
func (c *Client) GetStats() Stats {
	st := "connected"
	if !c.connected.Load() {
		st = "disconnected"
	}
	return Stats{
		Status:     st,
		Host:       fmt.Sprintf("%s:%d", c.host, c.port),
		Reconnects: c.reconnects.Load(),
		GatewayOps: c.gatewayOps.Load(),
		EventsRecv: c.eventsRecv.Load(),
		UptimeSec:  int64(time.Since(c.startTime).Seconds()),
	}
}

// IncGatewayOps увеличивает счётчик операций gateway.
func (c *Client) IncGatewayOps() {
	c.gatewayOps.Add(1)
}