// Package rabbitmq consumes ingest-service's "new reading" events and
// publishes rule-trigger events for notification-service. One connection is
// shared, with separate channels for consuming and publishing, since that's
// the amqp best practice (a channel is not meant to be used concurrently for
// unrelated flows).
package rabbitmq

import (
	"fmt"

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

const (
	NewReadingQueue    = "telemetry.new_reading" // declared by ingest-service; declared here too, idempotently
	RuleTriggeredQueue = "automation.rule_triggered"
)

type Client struct {
	conn      *amqp.Connection
	consumeCh *amqp.Channel
	publishCh *amqp.Channel
}

func Connect(url string) (*Client, error) {
	conn, err := amqp.Dial(url)
	if err != nil {
		return nil, fmt.Errorf("dial rabbitmq: %w", err)
	}

	consumeCh, err := conn.Channel()
	if err != nil {
		conn.Close()
		return nil, fmt.Errorf("open consume channel: %w", err)
	}
	// Process one reading at a time: simple, demonstrable backpressure —
	// the next delivery isn't sent until the current one is acked/nacked.
	if err := consumeCh.Qos(1, 0, false); err != nil {
		conn.Close()
		return nil, fmt.Errorf("set qos: %w", err)
	}
	if _, err := consumeCh.QueueDeclare(NewReadingQueue, true, false, false, false, nil); err != nil {
		conn.Close()
		return nil, fmt.Errorf("declare queue %s: %w", NewReadingQueue, err)
	}

	publishCh, err := conn.Channel()
	if err != nil {
		conn.Close()
		return nil, fmt.Errorf("open publish channel: %w", err)
	}
	if _, err := publishCh.QueueDeclare(RuleTriggeredQueue, true, false, false, false, nil); err != nil {
		conn.Close()
		return nil, fmt.Errorf("declare queue %s: %w", RuleTriggeredQueue, err)
	}

	return &Client{conn: conn, consumeCh: consumeCh, publishCh: publishCh}, nil
}

func (c *Client) Close() error {
	return c.conn.Close()
}