platform/rabbit/conn.go

// Package rabbit holds the shared RabbitMQ transport mechanics — connection
// setup plus topic declare/publish/consume helpers — that every module used to
// hand-roll. It carries NO event vocabulary: topic and routing names are plain
// strings supplied by the caller, so each module keeps its own naming policy
// (finder's "<country>_<topic>", CMS's "kind.dataType", etc.). See
// docs/platform-uniforming-plan.md (platform/rabbit seam, migration done).
package rabbit

import (
	"log"
	"sync"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

// Default backoff bounds for the automatic reconnect loop.
const (
	backoffMin = 1 * time.Second
	backoffMax = 30 * time.Second
)

// Connect dials url with a consistent configuration, tagging the connection with
// name so it is identifiable in the RabbitMQ management UI and metrics. It
// returns a bare *amqp.Connection and does NOT reconnect on failure.
//
// Prefer Dial instead: it returns a *Conn wrapper that handles automatic
// reconnection and notifies consumers when a reconnect completes so they can
// re-declare their topology.
//
// This replaces the 14 inconsistent hand-rolled dial sites (some with
// connection Properties, some bare amqp.Dial with no name).
func Connect(url, name string) (*amqp.Connection, error) {
	return dial(url, name)
}

// dial is the shared dial implementation used by both Connect and Dial.
func dial(url, name string) (*amqp.Connection, error) {
	props := amqp.NewConnectionProperties()
	if name != "" {
		props["connection_name"] = name
	}
	return amqp.DialConfig(url, amqp.Config{Properties: props})
}

// ---------------------------------------------------------------------------

// Conn wraps *amqp.Connection with automatic reconnection. After Dial, it
// monitors NotifyClose and re-dials with exponential backoff (1s→30s). All
// exported methods are safe for concurrent use.
//
// Usage:
//
//	conn, err := rabbit.Dial(url, "my-service")
//	if err != nil { return err }
//	defer conn.Close()
//
//	// Register consumer setup: called immediately AND on each reconnect.
//	conn.NotifyOnReconnect(func() {
//	    ch, err := conn.Channel()
//	    if err != nil { return }
//	    rabbit.ListenToTopic(ch, prefix, topic, handler)
//	})
type Conn struct {
	url  string
	name string

	mu   sync.Mutex
	conn *amqp.Connection
	done chan struct{} // closed by Close(); stops the reconnect loop

	onReconn chan struct{} // closed on each reconnect to wake up listeners

	notifyFns    []func() // registered via NotifyOnReconnect
	reconnectFns []func() // registered via NotifyReconnect
}

// Dial connects to RabbitMQ and returns a *Conn that automatically reconnects
// on connection loss. The name is shown in the RabbitMQ management UI.
func Dial(url, name string) (*Conn, error) {
	c := &Conn{
		url:      url,
		name:     name,
		done:     make(chan struct{}),
		onReconn: make(chan struct{}),
	}
	var raw *amqp.Connection
	var err error
	for attempts := 0; attempts < 10; attempts++ {
		raw, err = dial(url, name)
		if err == nil {
			break
		}
		if attempts < 9 {
			time.Sleep(500 * time.Millisecond)
		}
	}
	if err != nil {
		return nil, err
	}
	c.conn = raw

	// Fire initial connect signal so a consumer registered before this runs
	// (e.g. via NotifyOnReconnect above) gets called. Since onReconn starts
	// open, we close it here and replace it with a fresh one for reconnect.
	close(c.onReconn)
	c.onReconn = make(chan struct{})

	go c.reconnectLoop()
	return c, nil
}

// Connection returns the current *amqp.Connection. Safe for concurrent use.
// Use this to pass the underlying connection to functions that expect
// *amqp.Connection, such as SendChange and the finder's messaging wrappers.
func (c *Conn) Connection() *amqp.Connection {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.conn
}

// Channel opens a channel on the current underlying connection. It is safe
// for concurrent use; the error path matches *amqp.Connection.Channel.
func (c *Conn) Channel() (*amqp.Channel, error) {
	return c.Connection().Channel()
}

// OnReconnect returns a channel that is closed each time a reconnect completes.
// Use range to re-run consumer setup:
//
//	go func() {
//	    for range conn.OnReconnect() {
//	        ch, _ := conn.Channel()
//	        rabbit.ListenToTopic(ch, prefix, topic, handler)
//	    }
//	}()
//
// The channel is NOT signalled on initial connect — only on reconnects.
// For a function that fires on both initial connect and reconnect, use
// NotifyOnReconnect.
func (c *Conn) OnReconnect() <-chan struct{} {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.onReconn
}

// NotifyOnReconnect registers a function that is called immediately (on the
// current goroutine) and again on each subsequent reconnect (on the reconnect
// loop goroutine). The function should be idempotent — it may be called while
// the old listeners are still shutting down from the previous connection drop.
// Multiple functions can be registered and are called in registration order.
func (c *Conn) NotifyOnReconnect(fn func()) {
	// Call immediately with the current (live) connection.
	fn()

	c.mu.Lock()
	c.notifyFns = append(c.notifyFns, fn)
	c.mu.Unlock()
}

// NotifyReconnect registers a function that is called on each subsequent
// reconnect, but NOT immediately on registration. Use this when the caller
// needs to perform first-time setup separately and only wants the reconnect
// hook for re-declaring consumers or topology.
func (c *Conn) NotifyReconnect(fn func()) {
	c.mu.Lock()
	c.reconnectFns = append(c.reconnectFns, fn)
	c.mu.Unlock()
}

// Done returns a channel that is closed when the connection is permanently
// closed (Close was called).
func (c *Conn) Done() <-chan struct{} {
	return c.done
}

// Close permanently closes the connection and stops the reconnect loop.
// It is safe to call multiple times.
func (c *Conn) Close() error {
	select {
	case <-c.done:
		return nil
	default:
	}
	close(c.done)

	c.mu.Lock()
	defer c.mu.Unlock()
	if c.conn != nil {
		return c.conn.Close()
	}
	return nil
}

// reconnectLoop runs in a background goroutine. It waits for the connection
// to close (either normally or with an error), then re-dials with exponential
// backoff and notifies registered callbacks.
func (c *Conn) reconnectLoop() {
	for {
		// Wait for the connection to break (or Close to be called).
		errCh := make(chan *amqp.Error)
		c.mu.Lock()
		conn := c.conn
		c.mu.Unlock()

		conn.NotifyClose(errCh)

		select {
		case <-c.done:
			return
		case err, ok := <-errCh:
			if !ok || err == nil {
				// Channel closed or clean shutdown (Close called on the underlying
				// connection). Either way, stop reconnecting.
				return
			}
			log.Printf("rabbit: connection %s lost: %v (reconnecting)", c.name, err)
		}

		// Reconnect with exponential backoff.
		c.reconnect()
	}
}

// reconnect attempts to re-dial until successful or Close is called.
func (c *Conn) reconnect() {
	duration := backoffMin
	for {
		select {
		case <-c.done:
			return
		default:
		}

		newConn, err := dial(c.url, c.name)
		if err == nil {
			c.mu.Lock()
			old := c.conn
			c.conn = newConn
			c.mu.Unlock()
			if old != nil {
				_ = old.Close()
			}

			log.Printf("rabbit: connection %s re-established", c.name)

			// Notify consumers: close the current signal channel so blocked
			// OnReconnect calls advance; replace with a fresh one.
			c.mu.Lock()
			close(c.onReconn)
			c.onReconn = make(chan struct{})
			fns := append([]func(){}, c.notifyFns...)
			fns = append(fns, c.reconnectFns...)
			c.mu.Unlock()

			for _, fn := range fns {
				fn()
			}
			return
		}

		log.Printf("rabbit: reconnect for %s failed: %v (retry in %v)", c.name, err, duration)

		select {
		case <-c.done:
			return
		case <-time.After(duration):
		}

		duration *= 2
		if duration > backoffMax {
			duration = backoffMax
		}
	}
}