go-sip/internal/mq/broker.go

package mq

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"sync"
	"time"

	"git.ipao.vip/rogee/go-sip/internal/contract"
	"git.ipao.vip/rogee/go-sip/internal/logging"
	"git.ipao.vip/rogee/go-sip/internal/tenant"
	amqp "github.com/rabbitmq/amqp091-go"
)

const (
	CommandsExchange = "agent-call.dispatchers.v1"
	ResultsExchange  = "agent-call.saas.v1"
	MaxMessageBytes  = 8 << 20
)

type Broker struct {
	conn           *amqp.Connection
	invalidHandler InvalidMessageHandler
	dispatcherID   string
	prefetch       int
	controlQueue   string
	resultRoute    string
	closed         <-chan *amqp.Error
	mu             sync.Mutex
}

// ControlQueueName returns this Dispatcher's SaaS-provisioned control queue.
func ControlQueueName(dispatcherID string) string {
	return "agent-call.d." + dispatcherID + ".control.v1"
}

// Open verifies SaaS-provisioned topology passively. It never declares,
// binds, or deletes exchanges or queues.
func Open(url, dispatcherID string, prefetch int) (*Broker, error) {
	if strings.TrimSpace(url) == "" {
		return nil, errors.New("rabbitmq URL is required")
	}
	if prefetch <= 0 {
		return nil, errors.New("prefetch must be positive")
	}
	if err := tenant.ValidateDispatcherID(dispatcherID); err != nil {
		return nil, err
	}
	conn, err := amqp.Dial(url)
	if err != nil {
		return nil, fmt.Errorf("dial rabbitmq: %w", err)
	}
	channel, err := conn.Channel()
	if err != nil {
		_ = conn.Close()
		return nil, fmt.Errorf("open rabbitmq current channel: %w", err)
	}
	for _, exchange := range []string{CommandsExchange, ResultsExchange} {
		if err := channel.ExchangeDeclarePassive(exchange, "topic", true, false, false, false, nil); err != nil {
			_ = channel.Close()
			_ = conn.Close()
			return nil, fmt.Errorf("required SaaS exchange %s unavailable: %w", exchange, err)
		}
	}
	controlQueue := ControlQueueName(dispatcherID)
	if _, err := channel.QueueDeclarePassive(controlQueue, true, false, false, false, nil); err != nil {
		_ = channel.Close()
		_ = conn.Close()
		return nil, fmt.Errorf("required SaaS control queue unavailable: %w", err)
	}
	// The shared SaaS result queue is not readable by Dispatcher credentials.
	// mandatory + return + confirm and a SaaS-owned Mock consumer prove routing.
	_ = channel.Close()
	return &Broker{
		conn:         conn,
		dispatcherID: dispatcherID,
		prefetch:     prefetch,
		controlQueue: controlQueue,
		resultRoute:  "d." + dispatcherID + ".out",
		closed:       conn.NotifyClose(make(chan *amqp.Error, 1)),
	}, nil
}

func (b *Broker) Close() error {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.conn == nil {
		return nil
	}
	err := b.conn.Close()
	b.conn = nil
	return err
}

func (b *Broker) Done() <-chan *amqp.Error { return b.closed }

func (b *Broker) Publish(ctx context.Context, exchange, routingKey string, body []byte) (publishErr error) {
	started := time.Now()
	attrs := append(messageIdentity(body), "transport", "mq", "exchange", exchange, "routing_key", routingKey)
	slog.InfoContext(ctx, "MQ publish requested", append(attrs, "direction", "dispatcher->mq", "stage", "publish_requested", "body_bytes", len(body))...)
	logging.Body(ctx, "MQ publish data", body, append(attrs, "direction", "dispatcher->mq", "stage", "publish_requested")...)
	defer func() {
		result := append(append([]any{}, attrs...), "duration_ms", time.Since(started).Milliseconds())
		if publishErr != nil {
			slog.ErrorContext(ctx, "MQ publish failed", append(result, "direction", "dispatcher->mq", "stage", "publish_failed", "error", publishErr)...)
		} else {
			slog.InfoContext(ctx, "MQ publish confirmed", append(result, "direction", "mq->dispatcher", "stage", "publisher_confirmed", "saas_processed", false)...)
		}
	}()
	if exchange != ResultsExchange || routingKey != b.resultRoute || len(body) == 0 || len(body) > MaxMessageBytes {
		return fmt.Errorf("publish requires result exchange, Dispatcher route and 1..%d body bytes", MaxMessageBytes)
	}
	if err := contract.ValidateCurrent("mq", body); err != nil {
		return fmt.Errorf("outbound MQ contract: %w", err)
	}
	var identity struct {
		EventID      string `json:"event_id"`
		EventType    string `json:"event_type"`
		DispatcherID string `json:"dispatcher_id"`
		Payload      struct {
			Status string `json:"status"`
		} `json:"payload"`
	}
	if err := json.Unmarshal(body, &identity); err != nil {
		return fmt.Errorf("decode outbound MQ identity: %w", err)
	}
	if identity.DispatcherID != b.dispatcherID {
		return errors.New("outbound MQ dispatcher owner mismatch")
	}
	if identity.EventType != "call.execute.result" &&
		!((identity.EventType == "task.control" || identity.EventType == "call.execute") && identity.Payload.Status != "") &&
		!(identity.EventType == "sip.config" && identity.Payload.Status == "rejected") {
		return fmt.Errorf("unapproved outbound MQ event %q", identity.EventType)
	}
	if len(identity.EventID) > 255 {
		return errors.New("outbound event identity exceeds AMQP limit")
	}
	slog.InfoContext(ctx, "MQ outbound validated", append(attrs, "direction", "dispatcher->mq", "stage", "validated")...)
	messageID := identity.EventID
	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	b.mu.Lock()
	defer b.mu.Unlock()
	if err := ctx.Err(); err != nil {
		return err
	}
	if b.conn == nil || b.conn.IsClosed() {
		return errors.New("rabbitmq current connection is closed")
	}
	channel, err := b.conn.Channel()
	if err != nil {
		return fmt.Errorf("open current publication channel: %w", err)
	}
	defer channel.Close()
	if err := channel.Confirm(false); err != nil {
		return fmt.Errorf("enable current publisher confirms: %w", err)
	}
	returned := channel.NotifyReturn(make(chan amqp.Return, 1))
	slog.InfoContext(ctx, "MQ publish", append(attrs, "direction", "dispatcher->mq", "stage", "publish_start")...)
	confirmation, err := channel.PublishWithDeferredConfirmWithContext(ctx, exchange, routingKey, true, false, amqp.Publishing{
		ContentType: "application/json", DeliveryMode: amqp.Persistent, Body: body, MessageId: messageID,
	})
	if err != nil {
		return fmt.Errorf("publish current result: %w", err)
	}
	if confirmation == nil {
		return errors.New("rabbitmq current publisher confirmation unavailable")
	}
	acked, err := confirmation.WaitContext(ctx)
	if err != nil {
		return fmt.Errorf("wait for current publisher confirmation: %w", err)
	}
	select {
	case result, ok := <-returned:
		if !ok {
			return errors.New("current publication channel closed before routing was established")
		}
		return fmt.Errorf("current publication returned: code=%d", result.ReplyCode)
	default:
	}
	if !acked {
		return errors.New("rabbitmq current publisher was negatively acknowledged")
	}
	return nil
}

type Consumer struct {
	cancel context.CancelFunc
	done   chan struct{}
	mu     sync.Mutex
	err    error
}

func (c *Consumer) Wait(ctx context.Context) error {
	select {
	case <-c.done:
		c.mu.Lock()
		defer c.mu.Unlock()
		return c.err
	case <-ctx.Done():
		return ctx.Err()
	}
}

// RequestStop cancels consumption without waiting for the current handler.
// Calling Stop from inside that handler would deadlock on its own completion.
func (c *Consumer) RequestStop() {
	c.cancel()
}

func (c *Consumer) Stop(ctx context.Context) error {
	c.cancel()
	err := c.Wait(ctx)
	if errors.Is(err, context.Canceled) {
		return nil
	}
	return err
}

func (b *Broker) ConsumePredeclared(ctx context.Context, queue string, handler MessageHandler) error {
	consumer, err := b.StartPredeclaredConsumer(ctx, queue, handler)
	if err != nil {
		return err
	}
	err = consumer.Wait(context.Background())
	if err != nil {
		return err
	}
	return ctx.Err()
}

// StartPredeclaredConsumer subscribes only to an existing SaaS-owned queue.
// Stopping it closes the channel so every unacknowledged delivery is requeued.
func (b *Broker) StartPredeclaredConsumer(ctx context.Context, queue string, handler MessageHandler) (*Consumer, error) {
	if handler == nil {
		return nil, errors.New("predeclared handler is required")
	}
	if err := b.validateQueue(queue, true); err != nil {
		return nil, err
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	b.mu.Lock()
	conn, prefetch, dispatcherID := b.conn, b.prefetch, b.dispatcherID
	closed := conn == nil || conn.IsClosed()
	b.mu.Unlock()
	if closed {
		return nil, errors.New("rabbitmq current connection is closed")
	}
	channel, err := conn.Channel()
	if err != nil {
		return nil, fmt.Errorf("open current consumer channel: %w", err)
	}
	if _, err := channel.QueueDeclarePassive(queue, true, false, false, false, nil); err != nil {
		_ = channel.Close()
		return nil, fmt.Errorf("required SaaS-owned queue %s unavailable: %w", queue, err)
	}
	if err := channel.Qos(prefetch, 0, false); err != nil {
		_ = channel.Close()
		return nil, fmt.Errorf("set current consumer prefetch: %w", err)
	}
	consumerTag := fmt.Sprintf("sip-go-agent-%d", consumerSequence.Add(1))
	deliveries, err := channel.Consume(queue, consumerTag, false, false, false, false, nil)
	if err != nil {
		_ = channel.Close()
		return nil, fmt.Errorf("consume SaaS-owned current queue: %w", err)
	}
	consumerCtx, cancel := context.WithCancel(ctx)
	consumer := &Consumer{cancel: cancel, done: make(chan struct{})}
	go func() {
		consumeErr := b.consumeDeliveries(consumerCtx, dispatcherID, queue, deliveries, handler)
		if errors.Is(consumeErr, context.Canceled) && consumerCtx.Err() != nil {
			consumeErr = nil
		}
		if err := channel.Cancel(consumerTag, false); err != nil {
			consumeErr = errors.Join(consumeErr, fmt.Errorf("cancel current consumer: %w", err))
		}
		if err := channel.Close(); err != nil {
			consumeErr = errors.Join(consumeErr, fmt.Errorf("close current consumer channel: %w", err))
		}
		consumer.mu.Lock()
		consumer.err = consumeErr
		consumer.mu.Unlock()
		close(consumer.done)
	}()
	return consumer, nil
}

// PurgeTaskQueue removes Ready messages from only this Dispatcher's task queue.
// The caller must first stop its consumer so unacknowledged deliveries are
// requeued before the purge; SaaS still owns queue creation and bindings.
func (b *Broker) PurgeTaskQueue(ctx context.Context, taskID string) (int, error) {
	route, err := tenant.TaskRoute(b.dispatcherID, taskID)
	if err != nil {
		return 0, err
	}
	if err := ctx.Err(); err != nil {
		return 0, err
	}
	b.mu.Lock()
	conn := b.conn
	closed := conn == nil || conn.IsClosed()
	b.mu.Unlock()
	if closed {
		return 0, errors.New("rabbitmq current connection is closed")
	}
	channel, err := conn.Channel()
	if err != nil {
		return 0, fmt.Errorf("open current purge channel: %w", err)
	}
	defer channel.Close()
	if _, err := channel.QueueDeclarePassive(route.Queue, true, false, false, false, nil); err != nil {
		return 0, fmt.Errorf("required SaaS-owned queue %s unavailable: %w", route.Queue, err)
	}
	if err := ctx.Err(); err != nil {
		return 0, err
	}
	count, err := channel.QueuePurge(route.Queue, false)
	if err != nil {
		return 0, fmt.Errorf("purge SaaS-owned task queue %s: %w", route.Queue, err)
	}
	return count, nil
}

// DrainControlPredeclared processes the already queued controls before task admission.
// Control deliveries must pass through the handler before ACK; a transient
// failure is requeued and closes startup admission.
func (b *Broker) DrainControlPredeclared(ctx context.Context, queue string, handler MessageHandler) (int, error) {
	if queue == "" || queue != b.controlQueue || handler == nil {
		return 0, errors.New("configured SaaS-owned control queue and handler are required")
	}
	if err := ctx.Err(); err != nil {
		return 0, err
	}
	b.mu.Lock()
	conn := b.conn
	closed := conn == nil || conn.IsClosed()
	b.mu.Unlock()
	if closed {
		return 0, errors.New("rabbitmq current connection is closed")
	}
	channel, err := conn.Channel()
	if err != nil {
		return 0, fmt.Errorf("open control backlog channel: %w", err)
	}
	defer channel.Close()
	if _, err := channel.QueueDeclarePassive(queue, true, false, false, false, nil); err != nil {
		return 0, fmt.Errorf("required SaaS-owned control queue %s unavailable: %w", queue, err)
	}
	processed := 0
	for {
		if err := ctx.Err(); err != nil {
			return processed, err
		}
		delivery, ok, err := channel.Get(queue, false)
		if err != nil {
			return processed, fmt.Errorf("read SaaS-owned control backlog %s: %w", queue, err)
		}
		if !ok {
			return processed, nil
		}
		observeDelivery(ctx, queue, &delivery)
		if err := b.validateInbound(queue, delivery.RoutingKey, delivery.Body); err != nil {
			if err := b.rejectInbound(ctx, queue, delivery, err); err != nil {
				return processed, err
			}
			processed++
			continue
		}
		slog.InfoContext(ctx, "MQ inbound validated", append(messageIdentity(delivery.Body), "direction", "mq->dispatcher", "queue", queue, "stage", "validated")...)
		if err := handler(ctx, delivery.RoutingKey, delivery.Body); err != nil {
			slog.ErrorContext(ctx, "MQ handler failed", append(messageIdentity(delivery.Body), "stage", "handler_failed", "error", err, "permanent", IsPermanent(err))...)
			if IsPermanent(err) {
				if rejectErr := delivery.Reject(false); rejectErr != nil {
					return processed, fmt.Errorf("control handler error %v; reject: %w", err, rejectErr)
				}
				processed++
				return processed, fmt.Errorf("control backlog rejected permanently; admission stays closed: %w", err)
			}
			if nackErr := delivery.Nack(false, true); nackErr != nil {
				return processed, fmt.Errorf("control handler error %v; nack: %w", err, nackErr)
			}
			return processed, fmt.Errorf("control backlog handler failed, admission stays closed: %w", err)
		}
		if err := delivery.Ack(false); err != nil {
			return processed, fmt.Errorf("ack processed control backlog: %w", err)
		}
		processed++
	}
}

func (b *Broker) ControlQueue() string { return b.controlQueue }

// validateInbound prevents a wrongly bound queue, foreign D message, old
// envelope, or outbound acknowledgment from entering the command handler.
func (b *Broker) validateInbound(queue, routingKey string, body []byte) error {
	if len(body) == 0 || len(body) > MaxMessageBytes {
		return errors.New("invalid MQ message size")
	}
	if err := b.validateQueue(queue, true); err != nil {
		return err
	}
	if err := contract.ValidateCurrent("mq", body); err != nil {
		return fmt.Errorf("inbound MQ contract: %w", err)
	}
	var identity struct {
		DispatcherID string `json:"dispatcher_id"`
		EventType    string `json:"event_type"`
		Payload      struct {
			TaskID   string `json:"task_id"`
			Action   string `json:"action"`
			Callee   string `json:"callee"`
			Status   string `json:"status"`
			TrunkID  string `json:"trunk_id"`
			Revision int64  `json:"revision"`
		} `json:"payload"`
	}
	if err := json.Unmarshal(body, &identity); err != nil {
		return fmt.Errorf("decode MQ message identity: %w", err)
	}
	if identity.Payload.Status != "" {
		return errors.New("outbound receipt cannot enter an inbound command queue")
	}
	if identity.DispatcherID != b.dispatcherID {
		return errors.New("MQ message belongs to another Dispatcher")
	}
	if queue == b.controlQueue {
		if routingKey != "d."+b.dispatcherID+".control.in" {
			return errors.New("control message used wrong routing key")
		}
		if (identity.EventType == "sip.config" && identity.Payload.TrunkID != "" && identity.Payload.Revision > 0) || (identity.EventType == "task.control" && identity.Payload.TaskID != "" && identity.Payload.Action != "") {
			return nil
		}
		return errors.New("unapproved control-queue event")
	}
	prefix := "agent-call.d." + b.dispatcherID + ".task."
	taskID := strings.TrimSuffix(strings.TrimPrefix(queue, prefix), ".v1")
	route, err := tenant.TaskRoute(b.dispatcherID, taskID)
	if err != nil || routingKey != route.BindingKey || identity.EventType != "call.execute" || identity.Payload.TaskID != taskID || identity.Payload.Callee == "" {
		return errors.New("unapproved task-queue event or routing key")
	}
	return nil
}

func (b *Broker) validateQueue(queue string, allowControl bool) error {
	if allowControl && queue == b.controlQueue {
		return nil
	}
	prefix := "agent-call.d." + b.dispatcherID + ".task."
	if !strings.HasPrefix(queue, prefix) || !strings.HasSuffix(queue, ".v1") {
		return errors.New("queue is not owned by this Dispatcher task")
	}
	taskID := strings.TrimSuffix(strings.TrimPrefix(queue, prefix), ".v1")
	route, err := tenant.TaskRoute(b.dispatcherID, taskID)
	if err != nil || route.Queue != queue {
		return errors.New("invalid assigned task queue")
	}
	return nil
}

func (b *Broker) consumeDeliveries(ctx context.Context, dispatcherID, queue string, deliveries <-chan amqp.Delivery, handler MessageHandler) error {
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case delivery, ok := <-deliveries:
			if !ok {
				return errors.New("rabbitmq current delivery channel closed")
			}
			observeDelivery(ctx, queue, &delivery)
			if err := b.validateInbound(queue, delivery.RoutingKey, delivery.Body); err != nil {
				if err := b.rejectInbound(ctx, queue, delivery, err); err != nil {
					return err
				}
				continue
			}
			slog.InfoContext(ctx, "MQ inbound validated", append(messageIdentity(delivery.Body), "direction", "mq->dispatcher", "queue", queue, "stage", "validated")...)
			if err := handler(ctx, delivery.RoutingKey, delivery.Body); err != nil {
				slog.ErrorContext(ctx, "MQ handler failed", append(messageIdentity(delivery.Body), "stage", "handler_failed", "error", err, "permanent", IsPermanent(err))...)
				if IsPermanent(err) {
					if rejectErr := delivery.Reject(false); rejectErr != nil {
						return fmt.Errorf("permanent current handler error %v; reject: %w", err, rejectErr)
					}
					if queue == b.controlQueue {
						return fmt.Errorf("control handler rejected permanently; admission stays closed: %w", err)
					}
					continue
				}
				if nackErr := delivery.Nack(false, true); nackErr != nil {
					return fmt.Errorf("current handler error %v; nack: %w", err, nackErr)
				}
				if queue == b.controlQueue {
					return fmt.Errorf("control handler failed; admission stays closed: %w", err)
				}
				continue
			}
			if err := delivery.Ack(false); err != nil {
				return fmt.Errorf("ack current delivery: %w", err)
			}
		}
	}
}