go-sip/cmd/saas-server/mq.go

package main

import (
	"context"
	"errors"
	"fmt"
	"time"

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

const feedbackQueue = "agent-call.saas.events.v1"

type broker struct{ conn *amqp.Connection }

func connectBroker(address string, data snapshots) (*broker, error) {
	conn, err := amqp.DialConfig(address, amqp.Config{Dial: amqp.DefaultDial(5 * time.Second)})
	if err != nil {
		return nil, fmt.Errorf("connect test MQ: %w", err)
	}
	b := &broker{conn: conn}
	if err = b.provision(data); err != nil {
		conn.Close()
		return nil, err
	}
	return b, nil
}

func (b *broker) provision(data snapshots) error {
	ch, err := b.conn.Channel()
	if err != nil {
		return fmt.Errorf("open topology channel: %w", err)
	}
	defer ch.Close()
	for _, exchange := range []string{mq.CommandsExchange, mq.ResultsExchange} {
		if err = ch.ExchangeDeclare(exchange, "topic", true, false, false, false, nil); err != nil {
			return fmt.Errorf("declare %s: %w", exchange, err)
		}
	}
	bind := func(queue, route, exchange string) error {
		if _, e := ch.QueueDeclare(queue, true, false, false, false, nil); e != nil {
			return fmt.Errorf("declare %s: %w", queue, e)
		}
		if e := ch.QueueBind(queue, route, exchange, false, nil); e != nil {
			return fmt.Errorf("bind %s: %w", queue, e)
		}
		return nil
	}
	if err = bind(mq.ControlQueueName(data.dispatcherID), "d."+data.dispatcherID+".control.in", mq.CommandsExchange); err != nil {
		return err
	}
	for id := range data.tasks {
		if err = bind("agent-call.d."+data.dispatcherID+".task."+id+".v1", "d."+data.dispatcherID+".task."+id+".in", mq.CommandsExchange); err != nil {
			return err
		}
	}
	return bind(feedbackQueue, "d."+data.dispatcherID+".out", mq.ResultsExchange)
}

func (b *broker) publish(ctx context.Context, route string, raw []byte) (string, error) {
	ch, err := b.conn.Channel()
	if err != nil {
		return "not_sent", fmt.Errorf("open publisher: %w", err)
	}
	defer ch.Close()
	if err = ch.Confirm(false); err != nil {
		return "not_sent", fmt.Errorf("enable confirmation: %w", err)
	}
	confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1))
	returns := ch.NotifyReturn(make(chan amqp.Return, 1))
	if err = ch.PublishWithContext(ctx, mq.CommandsExchange, route, true, false, amqp.Publishing{ContentType: "application/json", DeliveryMode: amqp.Persistent, Body: raw}); err != nil {
		return "unknown", fmt.Errorf("MQ publish outcome unknown: %w", err)
	}
	return awaitConfirm(ctx, confirms, returns)
}

func awaitConfirm(ctx context.Context, confirms <-chan amqp.Confirmation, returns <-chan amqp.Return) (string, error) {
	returned := false
	for {
		select {
		case _, ok := <-returns:
			if !ok {
				returns = nil
			} else {
				returned = true
			}
		case confirm, ok := <-confirms:
			if !ok {
				return "unknown", errors.New("MQ confirmation stream closed")
			}
			// RabbitMQ emits return before confirmation; drain the buffered notification.
			select {
			case _, ok := <-returns:
				if ok {
					returned = true
				}
			default:
			}
			if returned {
				return "unroutable", errors.New("MQ returned unroutable message")
			}
			if !confirm.Ack {
				return "rejected", errors.New("MQ publisher NACK")
			}
			return "confirmed", nil
		case <-ctx.Done():
			return "unknown", errors.New("MQ confirmation deadline exceeded; do not resend")
		}
	}
}

func (b *broker) consume(ctx context.Context, s *service) error {
	ch, err := b.conn.Channel()
	if err != nil {
		return fmt.Errorf("open feedback consumer: %w", err)
	}
	defer ch.Close()
	if err = ch.Qos(1, 0, false); err != nil {
		return fmt.Errorf("feedback prefetch: %w", err)
	}
	deliveries, err := ch.Consume(feedbackQueue, "", false, true, false, false, nil)
	if err != nil {
		return fmt.Errorf("start exclusive feedback consumer: %w", err)
	}
	for {
		select {
		case <-ctx.Done():
			return nil
		case delivery, ok := <-deliveries:
			if !ok {
				return errors.New("MQ feedback stream closed")
			}
			if err = s.receive(delivery.Body, func() error { return delivery.Ack(false) }); err != nil {
				return fmt.Errorf("feedback save/ACK failed: %w", err)
			}
		}
	}
}