go-sip/internal/mq/integration_test.go

//go:build integration

package mq

import (
	"context"
	"errors"
	"fmt"
	"os"
	"testing"
	"time"

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

func TestLocalRabbitMQConfirmAckAndDeadLetter(t *testing.T) {
	url := os.Getenv("RABBITMQ_URL")
	if url == "" {
		t.Skip("RABBITMQ_URL is not configured")
	}
	broker, err := OpenWithPrefetch(url, uuid.NewString(), 1)
	if err != nil {
		t.Fatal(err)
	}
	defer broker.Close()

	tenantKey := fmt.Sprintf("integration-%d", time.Now().UnixNano())
	queue, err := broker.DeclareTenantQueue(tenantKey)
	if err != nil {
		t.Fatal(err)
	}
	route, err := tenant.NewDispatcherRoute(broker.dispatcherID, tenantKey)
	if err != nil {
		t.Fatal(err)
	}
	routingKey := route.InboundKey
	deadLetterQueue := route.DeadLetterQueue
	body := []byte(`{"command_id":"integration-command"}`)

	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	publishCommand := func(ctx context.Context) error {
		channel, err := broker.conn.Channel()
		if err != nil {
			return err
		}
		defer channel.Close()
		if err := channel.Confirm(false); err != nil {
			return err
		}
		returned := channel.NotifyReturn(make(chan amqp.Return, 1))
		confirmation, err := channel.PublishWithDeferredConfirmWithContext(ctx, DefaultExchange, routingKey, true, false, amqp.Publishing{
			DeliveryMode: amqp.Persistent,
			Body:         body,
		})
		if err != nil {
			return err
		}
		acked, err := confirmation.WaitContext(ctx)
		if err != nil {
			return err
		}
		select {
		case result := <-returned:
			return fmt.Errorf("command publication returned: code=%d", result.ReplyCode)
		default:
		}
		if !acked {
			return errors.New("command publication was negatively acknowledged")
		}
		return nil
	}
	acked := make(chan struct{}, 1)
	consumeDone := make(chan error, 1)
	go func() {
		consumeDone <- broker.Consume(ctx, queue, func(_ context.Context, gotRoutingKey string, gotBody []byte) error {
			if gotRoutingKey != routingKey || string(gotBody) != string(body) {
				return fmt.Errorf("delivery mismatch: key=%q body=%q", gotRoutingKey, gotBody)
			}
			acked <- struct{}{}
			return nil
		})
	}()
	if err := publishCommand(ctx); err != nil {
		t.Fatal(err)
	}
	select {
	case <-acked:
	case <-ctx.Done():
		t.Fatal(ctx.Err())
	}
	cancel()
	select {
	case <-consumeDone:
	case <-time.After(3 * time.Second):
		t.Fatal("consumer did not stop")
	}

	permanentCtx, permanentCancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer permanentCancel()
	permanentSeen := make(chan struct{}, 1)
	permanentDone := make(chan error, 1)
	go func() {
		permanentDone <- broker.Consume(permanentCtx, queue, func(_ context.Context, _ string, _ []byte) error {
			permanentSeen <- struct{}{}
			return Permanent(errors.New("synthetic permanent command error"))
		})
	}()
	if err := publishCommand(permanentCtx); err != nil {
		t.Fatal(err)
	}
	select {
	case <-permanentSeen:
	case <-permanentCtx.Done():
		t.Fatal(permanentCtx.Err())
	}
	permanentCancel()
	select {
	case <-permanentDone:
	case <-time.After(3 * time.Second):
		t.Fatal("permanent consumer did not stop")
	}

	dlqChannel, err := broker.conn.Channel()
	if err != nil {
		t.Fatal(err)
	}
	defer dlqChannel.Close()
	if _, err := dlqChannel.QueueDeclarePassive(deadLetterQueue, true, false, false, false, nil); err != nil {
		t.Fatal(err)
	}
	dlqDeliveries, err := dlqChannel.Consume(deadLetterQueue, "sip-go-agent-dlq", false, false, false, false, nil)
	if err != nil {
		t.Fatal(err)
	}
	deadLettered := make(chan struct{}, 1)
	dlqDone := make(chan error, 1)
	go func() {
		for delivery := range dlqDeliveries {
			if string(delivery.Body) != string(body) {
				dlqDone <- fmt.Errorf("dead-letter body mismatch: %q", delivery.Body)
				return
			}
			if err := delivery.Ack(false); err != nil {
				dlqDone <- err
				return
			}
			deadLettered <- struct{}{}
			dlqDone <- nil
			return
		}
		dlqDone <- errors.New("dead-letter delivery channel closed")
	}()
	select {
	case <-deadLettered:
		_ = dlqChannel.Cancel("sip-go-agent-dlq", false)
	case err := <-dlqDone:
		t.Fatal(err)
	case <-time.After(15 * time.Second):
		t.Fatal("dead-letter message did not arrive")
	}
	select {
	case err := <-dlqDone:
		if err != nil {
			t.Fatal(err)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("dead-letter consumer did not stop")
	}
}