go-sip/internal/mq/current.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/tenant"
	amqp "github.com/rabbitmq/amqp091-go"
)

const (
	CommandsExchangeCurrent   = "agent-call.dispatchers.v1"
	ResultsExchangeCurrent    = "agent-call.saas.v1"
	DeadLetterExchangeCurrent = "agent-call.dead-letter.v1"
	MaxCurrentMessageBytes    = 8 << 20
)

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

// OpenCurrent verifies SaaS-provisioned topology passively. It never declares,
// binds, or deletes exchanges or queues.
func CurrentControlQueueName(dispatcherID string) string {
	return "agent-call.d." + dispatcherID + ".control.v1"
}

func OpenCurrent(url, dispatcherID string, prefetch int) (*CurrentBroker, 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{CommandsExchangeCurrent, ResultsExchangeCurrent, DeadLetterExchangeCurrent} {
		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 := CurrentControlQueueName(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 &CurrentBroker{
		conn:         conn,
		dispatcherID: dispatcherID,
		prefetch:     prefetch,
		controlQueue: controlQueue,
		resultRoute:  "d." + dispatcherID + ".out",
		closed:       conn.NotifyClose(make(chan *amqp.Error, 1)),
	}, nil
}

func (b *CurrentBroker) 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 *CurrentBroker) Done() <-chan *amqp.Error { return b.closed }

func (b *CurrentBroker) Publish(ctx context.Context, exchange, routingKey string, body []byte) error {
	if exchange != ResultsExchangeCurrent || routingKey != b.resultRoute || len(body) == 0 || len(body) > MaxCurrentMessageBytes {
		return fmt.Errorf("current publish requires result exchange, Dispatcher route and 1..%d body bytes", MaxCurrentMessageBytes)
	}
	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 != "") {
		return fmt.Errorf("unapproved outbound MQ event %q", identity.EventType)
	}
	if len(identity.EventID) > 255 {
		return errors.New("outbound event identity exceeds AMQP limit")
	}
	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))
	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 CurrentConsumer struct {
	cancel context.CancelFunc
	done   chan struct{}
	mu     sync.Mutex
	err    error
}

func (c *CurrentConsumer) 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 *CurrentConsumer) RequestStop() {
	c.cancel()
}

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

func (b *CurrentBroker) 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 *CurrentBroker) StartPredeclaredConsumer(ctx context.Context, queue string, handler MessageHandler) (*CurrentConsumer, 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-current-%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 := &CurrentConsumer{cancel: cancel, done: make(chan struct{})}
	go func() {
		consumeErr := b.consumeCurrentDeliveries(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
}

func (b *CurrentBroker) DrainPredeclared(ctx context.Context, queue string) (int, error) {
	if err := b.validateQueue(queue, false); 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 drain 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 queue %s unavailable: %w", queue, err)
	}
	drained := 0
	for {
		if err := ctx.Err(); err != nil {
			return drained, err
		}
		delivery, ok, err := channel.Get(queue, false)
		if err != nil {
			return drained, fmt.Errorf("read SaaS-owned queue %s for stop drain: %w", queue, err)
		}
		if !ok {
			return drained, nil
		}
		if err := delivery.Ack(false); err != nil {
			return drained, fmt.Errorf("ack stopped task backlog: %w", err)
		}
		drained++
	}
}

// DrainControlPredeclared processes the already queued controls before task admission.
// Unlike stopped task backlogs, control deliveries must pass through the handler
// before ACK; a transient failure is requeued and closes startup admission.
func (b *CurrentBroker) 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
		}
		if len(delivery.Body) == 0 || len(delivery.Body) > MaxCurrentMessageBytes || !json.Valid(delivery.Body) {
			if err := delivery.Reject(false); err != nil {
				return processed, fmt.Errorf("reject invalid control backlog message: %w", err)
			}
			slog.Error("MQ control backlog invalid; admission stays closed", "dispatcher_id", b.dispatcherID, "delivery_tag", delivery.DeliveryTag, "reason", "invalid_size_or_json", "bytes", len(delivery.Body))
			processed++
			return processed, errors.New("invalid control backlog message; admission stays closed")
		}
		if err := b.validateInbound(queue, delivery.RoutingKey, delivery.Body); err != nil {
			if rejectErr := delivery.Reject(false); rejectErr != nil {
				return processed, fmt.Errorf("invalid control message %v; reject: %w", err, rejectErr)
			}
			slog.Error("MQ control backlog invalid; admission stays closed", "dispatcher_id", b.dispatcherID, "delivery_tag", delivery.DeliveryTag, "reason", err)
			processed++
			return processed, fmt.Errorf("invalid control backlog message; admission stays closed: %w", err)
		}
		if err := handler(ctx, delivery.RoutingKey, delivery.Body); err != nil {
			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 *CurrentBroker) 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 *CurrentBroker) validateInbound(queue, routingKey string, body []byte) error {
	if len(body) == 0 || len(body) > MaxCurrentMessageBytes {
		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"`
		} `json:"payload"`
	}
	if err := json.Unmarshal(body, &identity); err != nil {
		return fmt.Errorf("decode MQ message identity: %w", err)
	}
	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.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.CurrentTaskRoute(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 *CurrentBroker) 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.CurrentTaskRoute(b.dispatcherID, taskID)
	if err != nil || route.Queue != queue {
		return errors.New("invalid assigned task queue")
	}
	return nil
}

func (b *CurrentBroker) consumeCurrentDeliveries(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")
			}
			if len(delivery.Body) == 0 || len(delivery.Body) > MaxCurrentMessageBytes {
				if err := delivery.Reject(false); err != nil {
					return fmt.Errorf("reject invalid current message size: %w", err)
				}
				slog.Warn("MQ current message rejected", "dispatcher_id", dispatcherID, "delivery_tag", delivery.DeliveryTag, "reason", "invalid_message_size", "bytes", len(delivery.Body))
				if queue == b.controlQueue {
					return errors.New("invalid control message size; admission stays closed")
				}
				continue
			}
			if !json.Valid(delivery.Body) {
				if err := delivery.Reject(false); err != nil {
					return fmt.Errorf("reject invalid current JSON: %w", err)
				}
				slog.Warn("MQ current message rejected", "dispatcher_id", dispatcherID, "delivery_tag", delivery.DeliveryTag, "reason", "invalid_json")
				if queue == b.controlQueue {
					return errors.New("invalid control JSON; admission stays closed")
				}
				continue
			}
			if err := b.validateInbound(queue, delivery.RoutingKey, delivery.Body); err != nil {
				if rejectErr := delivery.Reject(false); rejectErr != nil {
					return fmt.Errorf("invalid inbound MQ message %v; reject: %w", err, rejectErr)
				}
				slog.Warn("MQ current message rejected", "dispatcher_id", dispatcherID, "delivery_tag", delivery.DeliveryTag, "reason", err)
				if queue == b.controlQueue {
					return fmt.Errorf("invalid control message; admission stays closed: %w", err)
				}
				continue
			}
			if err := handler(ctx, delivery.RoutingKey, delivery.Body); err != nil {
				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)
			}
		}
	}
}