go-sip/internal/mq/amqp.go

// Package mq contains the RabbitMQ adapter. Business state remains in the
// Dispatcher store; this package only declares topology and transports bytes.
package mq

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

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

var consumerSequence atomic.Uint64

const (
	DefaultExchange    = "agent-call.dispatchers.v2"
	EventExchange      = "agent-call.saas.v2"
	DeadLetterExchange = "agent-call.dead-letter.v2"
	SaaSQueue          = "agent-call.saas.events.v2"
	DefaultPrefetch    = 1
	MaxMessageBytes    = 256 << 10
)

type Publisher interface {
	Publish(context.Context, string, string, []byte) error
}

type permanentError struct{ err error }

func (e permanentError) Error() string { return e.err.Error() }
func (e permanentError) Unwrap() error { return e.err }

func Permanent(err error) error {
	if err == nil {
		return nil
	}
	return permanentError{err: err}
}

func IsPermanent(err error) bool {
	var target permanentError
	return errors.As(err, &target)
}

type Broker struct {
	conn         *amqp.Connection
	channel      *amqp.Channel
	dispatcherID string
	prefetch     int
	outbound     map[string]bool
	inboxes      map[string]bool
	closed       <-chan *amqp.Error
	mu           sync.Mutex
}

func Open(url, dispatcherID string) (*Broker, error) {
	return OpenWithPrefetch(url, dispatcherID, DefaultPrefetch)
}

func OpenWithPrefetch(url, dispatcherID string, prefetch int) (*Broker, error) {
	if 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 channel: %w", err)
	}
	owner := "agent-call.d." + dispatcherID + ".owner.v2"
	if _, err := channel.QueueDeclare(owner, false, false, true, false, nil); err != nil {
		_ = conn.Close()
		return nil, fmt.Errorf("claim dispatcher identity %s: %w", dispatcherID, err)
	}
	for _, exchange := range []string{DefaultExchange, EventExchange, DeadLetterExchange} {
		if err := channel.ExchangeDeclare(exchange, "topic", true, false, false, false, nil); err != nil {
			_ = conn.Close()
			return nil, fmt.Errorf("declare topic exchange %s: %w", exchange, err)
		}
	}
	return &Broker{conn: conn, channel: channel, dispatcherID: dispatcherID, prefetch: prefetch,
		outbound: make(map[string]bool), inboxes: make(map[string]bool), 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.channel, b.conn = nil, nil
	return err
}

// Done reports loss of the connection that owns this Dispatcher identity.
// The caller must stop admission when that ownership connection is lost.
func (b *Broker) Done() <-chan *amqp.Error { return b.closed }

func (b *Broker) Publish(ctx context.Context, exchange, routingKey string, body []byte) error {
	if exchange != EventExchange || routingKey == "" || len(body) == 0 || len(body) > MaxMessageBytes {
		return errors.New("publish requires SaaS exchange, declared route and 1..262144 body bytes")
	}
	var identity struct {
		EventID   string `json:"event_id"`
		MessageID string `json:"message_id"`
	}
	if err := json.Unmarshal(body, &identity); err != nil {
		return fmt.Errorf("decode outbound message identity: %w", err)
	}
	if (identity.EventID == "") == (identity.MessageID == "") {
		return errors.New("outbound message requires exactly one event_id or message_id")
	}
	messageID := identity.EventID + identity.MessageID
	if len(messageID) > 255 {
		return errors.New("outbound message identity exceeds AMQP limit")
	}
	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 connection is closed")
	}
	if !b.outbound[routingKey] {
		return errors.New("outbound route does not belong to a declared Dispatcher/tenant")
	}
	// A separate channel isolates late returns/confirms after an ambiguous send.
	// Connections are reused; a timed-out publication can never acknowledge the next one.
	channel, err := b.conn.Channel()
	if err != nil {
		return fmt.Errorf("open publication channel: %w", err)
	}
	defer channel.Close()
	if _, err := channel.QueueDeclarePassive(SaaSQueue, true, false, false, false, nil); err != nil {
		return fmt.Errorf("required SaaS queue unavailable: %w", err)
	}
	if err := channel.Confirm(false); err != nil {
		return fmt.Errorf("enable 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: %w", err)
	}
	if confirmation == nil {
		return errors.New("rabbitmq publisher confirmation unavailable")
	}
	acked, err := confirmation.WaitContext(ctx)
	if err != nil {
		return fmt.Errorf("wait publisher confirmation: %w", err)
	}
	// RabbitMQ sends basic.return before its confirm; the SDK dispatches it
	// before completing the deferred confirmation. A positive confirm alone is insufficient.
	select {
	case result, ok := <-returned:
		if !ok {
			return errors.New("publication channel closed before routing was established")
		}
		return fmt.Errorf("publication returned: code=%d", result.ReplyCode)
	default:
	}
	if !acked {
		return errors.New("rabbitmq publisher was negatively acknowledged")
	}
	return nil
}

func (b *Broker) DeclareTenantQueue(tenantKey string) (string, error) {
	route, err := tenant.NewDispatcherRoute(b.dispatcherID, tenantKey)
	if err != nil {
		return "", err
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.channel == nil {
		return "", errors.New("rabbitmq channel is closed")
	}
	queue, routingKey, deadLetterQueue := route.InboxQueue, route.InboundKey, route.DeadLetterQueue
	queueArgs := amqp.Table{
		"x-dead-letter-exchange":    DeadLetterExchange,
		"x-dead-letter-routing-key": routingKey,
	}
	if _, err := b.channel.QueueDeclare(queue, true, false, false, false, queueArgs); err != nil {
		return "", fmt.Errorf("declare tenant queue: %w", err)
	}
	if _, err := b.channel.QueueDeclare(deadLetterQueue, true, false, false, false, nil); err != nil {
		return "", fmt.Errorf("declare tenant dead-letter queue: %w", err)
	}
	if err := b.channel.QueueBind(queue, routingKey, DefaultExchange, false, nil); err != nil {
		return "", fmt.Errorf("bind tenant queue: %w", err)
	}
	if err := b.channel.QueueBind(deadLetterQueue, routingKey, DeadLetterExchange, false, nil); err != nil {
		return "", fmt.Errorf("bind tenant dead-letter queue: %w", err)
	}
	if _, err := b.channel.QueueDeclare(SaaSQueue, true, false, false, false, nil); err != nil {
		return "", fmt.Errorf("declare SaaS queue: %w", err)
	}
	if err := b.channel.QueueBind(SaaSQueue, route.OutboundKey, EventExchange, false, nil); err != nil {
		return "", fmt.Errorf("bind SaaS queue: %w", err)
	}
	b.outbound[route.OutboundKey], b.inboxes[queue] = true, true
	return queue, nil
}

type MessageHandler func(context.Context, string, []byte) error

// Consume ACKs only after the handler returns nil. A transient handler error
// requeues; malformed or unauthorized messages can be rejected by the caller
// with Permanent, which RabbitMQ dead-letters through the tenant queue policy.
func (b *Broker) Consume(ctx context.Context, queue string, handler MessageHandler) error {
	if queue == "" || handler == nil {
		return errors.New("queue and handler are required")
	}
	b.mu.Lock()
	if b.channel == nil {
		b.mu.Unlock()
		return errors.New("rabbitmq channel is closed")
	}
	if !b.inboxes[queue] {
		b.mu.Unlock()
		return errors.New("queue does not belong to a declared Dispatcher/tenant")
	}
	prefetch := b.prefetch
	if err := b.channel.Qos(prefetch, 0, false); err != nil {
		b.mu.Unlock()
		return fmt.Errorf("set tenant prefetch: %w", err)
	}
	consumerTag := fmt.Sprintf("sip-go-agent-%d", consumerSequence.Add(1))
	deliveries, err := b.channel.Consume(queue, consumerTag, false, false, false, false, nil)
	b.mu.Unlock()
	if err != nil {
		return fmt.Errorf("consume tenant queue: %w", err)
	}
	defer func() {
		b.mu.Lock()
		if b.channel != nil {
			_ = b.channel.Cancel(consumerTag, false)
		}
		b.mu.Unlock()
	}()
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case d, ok := <-deliveries:
			if !ok {
				return errors.New("rabbitmq delivery channel closed")
			}
			if len(d.Body) == 0 || len(d.Body) > MaxMessageBytes {
				if err := d.Reject(false); err != nil {
					return fmt.Errorf("reject invalid message size: %w", err)
				}
				slog.Warn("MQ message rejected", "dispatcher_id", b.dispatcherID, "delivery_tag", d.DeliveryTag, "reason", "invalid_message_size", "bytes", len(d.Body))
				continue
			}
			if err := handler(ctx, d.RoutingKey, d.Body); err != nil {
				if IsPermanent(err) {
					if rejectErr := d.Reject(false); rejectErr != nil {
						return fmt.Errorf("permanent handler error %v; reject: %w", err, rejectErr)
					}
					continue
				}
				if nackErr := d.Nack(false, true); nackErr != nil {
					return fmt.Errorf("handler error %v; nack: %w", err, nackErr)
				}
				continue
			}
			if err := d.Ack(false); err != nil {
				return fmt.Errorf("ack delivery: %w", err)
			}
		}
	}
}