Main package

import "go.digitalxero.dev/mqutils/v2"

// Create a consumer
config := viper.New()
config.Set("url", "amqp://localhost:5672")
config.Set("queue", "my-queue")

consumer, err := mqutils.NewConsumer(ctx, config)
if err != nil {
    log.Fatal(err)
}

// Register a handler
mqutils.RegisterHandler("my-queue", myHandler)

// Run the consumer
if err := consumer.Run(ctx); err != nil {
    log.Printf("Consumer error: %v", err)
}

mqutils.RegisterBatchHandler("metrics.batch", func(ctx context.Context, msgs []types.Message) error {
    // Process batch of metrics
    var metrics []Metric
    for _, msg := range msgs {
        var metric Metric
        if err := json.Unmarshal(msg.Body(), &metric); err != nil {
            _ = msg.Nack() // reject just this message
            continue
        }
        metrics = append(metrics, metric)
    }

    // Bulk insert metrics; an error return nacks every message the
    // handler did not already acknowledge, a nil return acks them.
    return db.BulkInsertMetrics(ctx, metrics)
})

var RegisterBatchHandler = types.RegisterBatchHandler

mqutils.RegisterHandler("user.events", func(ctx context.Context, msg types.Message) error {
    // Process user event
    var event UserEvent
    if err := json.Unmarshal(msg.Body(), &event); err != nil {
        return err // the consumer runtime nacks the message
    }
    // Handle the event... a nil return acknowledges the message
    return nil
})

var RegisterHandler = types.RegisterHandler

func NewConsumer(ctx context.Context, config *viper.Viper) (types.Consumer, error)

config := viper.New()
config.Set("url", "kafka://localhost:9092")
config.Set("queue", "events")
config.Set("consumer_group", "my-service")

consumer, err := mqutils.NewConsumer(ctx, config)
if err != nil {
    return fmt.Errorf("failed to create consumer: %w", err)
}

func NewProducer(ctx context.Context, config *viper.Viper) (types.Producer, error)

config := viper.New()
config.Set("url", "amqp://localhost:5672")

producer, err := mqutils.NewProducer(ctx, config)
if err != nil {
    return fmt.Errorf("failed to create producer: %w", err)
}

// Some producers need explicit start
if err := producer.Start(ctx); err != nil {
    return fmt.Errorf("failed to start producer: %w", err)
}

// Publish a message
err = producer.Publish(ctx, "correlation-123", "events", "user.created",
    "application/json", []byte(`{"id": 123}`))

type ConsumerBuilder interface {
    // WithURL sets the broker connection URL. Required. The URL scheme
    // selects the backend (amqp://, kafka://, nats://, sqs://, pubsub://,
    // redis://, ...).
    WithURL(url string) ConsumerBuilder
    // WithDestination sets the consume target using the canonical
    // "destination" key (queue, topic, subject, stream — backend-dependent).
    WithDestination(destination string) ConsumerBuilder
    // WithConsumerGroup sets the competing-consumer group identity using the
    // canonical "consumer_group" key.
    WithConsumerGroup(group string) ConsumerBuilder
    // WithHandler supplies the message handler directly — no separate
    // RegisterHandler call is needed.
    WithHandler(handler types.HandlerFunc) ConsumerBuilder
    // WithHandlerName references a handler previously registered with
    // RegisterHandler. Mutually exclusive with WithHandler.
    WithHandlerName(name string) ConsumerBuilder
    // WithBatchHandler enables batch processing with the supplied handler,
    // batch size, and flush timeout.
    WithBatchHandler(handler types.BatchHandlerFunc, size int, timeout time.Duration) ConsumerBuilder
    // WithMaxRetries sets the retry budget using the canonical "max_retries"
    // key; messages beyond it are dropped or dead-lettered by the backend.
    WithMaxRetries(n int) ConsumerBuilder
    // WithGracefulShutdown enables graceful shutdown with the given drain
    // timeout.
    WithGracefulShutdown(timeout time.Duration) ConsumerBuilder
    // WithAutoReconnect toggles automatic reconnection on connection loss.
    WithAutoReconnect(enabled bool) ConsumerBuilder
    // WithPrefetch sets the message channel buffer / prefetch depth.
    WithPrefetch(n int) ConsumerBuilder
    // WithConcurrency limits active handlers or batches. When omitted, the backend
    // defaults to its effective message_channel_buffer, including WithPrefetch.
    WithConcurrency(n int) ConsumerBuilder
    // WithTLSSkipVerify disables TLS certificate verification (use with
    // caution; intended for local brokers and test endpoints).
    WithTLSSkipVerify(skip bool) ConsumerBuilder
    // WithTLSServerName sets the TLS ServerName (SNI) used when it differs
    // from the URL host. Maps to the sni_hostname config key.
    WithTLSServerName(name string) ConsumerBuilder
    // WithTLSClientCert sets the client certificate and key PEM paths for
    // mTLS. Both are required if either is set. Maps to tls_cert and tls_key.
    WithTLSClientCert(certFile, keyFile string) ConsumerBuilder
    // WithTLSCA sets an optional CA PEM path. System roots are used if unset.
    // Maps to tls_ca.
    WithTLSCA(caFile string) ConsumerBuilder
    // WithConfig merges a viper configuration for backend-specific keys.
    // Typed builder values take precedence over keys already set on it.
    WithConfig(v *viper.Viper) ConsumerBuilder
    // Build resolves the backend from the URL and constructs the consumer.
    Build(ctx context.Context) (types.Consumer, error)
}

func NewConsumerBuilder() ConsumerBuilder

type ProducerBuilder interface {
    // WithURL sets the broker connection URL. Required. The URL scheme
    // selects the backend.
    WithURL(url string) ProducerBuilder
    // WithDestination sets the publish target using the canonical
    // "destination" key (exchange, topic, subject, stream — backend-dependent).
    WithDestination(destination string) ProducerBuilder
    // WithTLSSkipVerify disables TLS certificate verification.
    WithTLSSkipVerify(skip bool) ProducerBuilder
    // WithTLSServerName sets the TLS ServerName (SNI) used when it differs
    // from the URL host. Maps to the sni_hostname config key.
    WithTLSServerName(name string) ProducerBuilder
    // WithTLSClientCert sets the client certificate and key PEM paths for
    // mTLS. Both are required if either is set. Maps to tls_cert and tls_key.
    WithTLSClientCert(certFile, keyFile string) ProducerBuilder
    // WithTLSCA sets an optional CA PEM path. System roots are used if unset.
    // Maps to tls_ca.
    WithTLSCA(caFile string) ProducerBuilder
    // WithConfig merges a viper configuration for backend-specific keys.
    WithConfig(v *viper.Viper) ProducerBuilder
    // Build resolves the backend from the URL and constructs the producer.
    Build(ctx context.Context) (types.Producer, error)
}

func NewProducerBuilder() ProducerBuilder