cc-backend/pkg/nats/client.go

// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
// All rights reserved. This file is part of cc-backend.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.

// Package nats provides a generic NATS messaging client for publish/subscribe communication.
//
// The package wraps the nats.go library with connection management, automatic reconnection
// handling, and subscription tracking. It supports multiple authentication methods including
// username/password and credential files.
//
// # Configuration
//
// Configure the client via JSON in the application config:
//
//	{
//	  "nats": {
//	    "address": "nats://localhost:4222",
//	    "username": "user",
//	    "password": "secret"
//	  }
//	}
//
// Or using a credentials file:
//
//	{
//	  "nats": {
//	    "address": "nats://localhost:4222",
//	    "creds-file-path": "/path/to/creds.json"
//	  }
//	}
//
// # Usage
//
// The package provides a singleton client initialized once and retrieved globally:
//
//	nats.Init(rawConfig)
//	nats.Connect()
//
//	client := nats.GetClient()
//	client.Subscribe("events", func(subject string, data []byte) {
//	    fmt.Printf("Received: %s\n", data)
//	})
//
//	client.Publish("events", []byte("hello"))
//
// # Thread Safety
//
// All Client methods are safe for concurrent use.
package nats

import (
	"context"
	"fmt"
	"sync"

	cclog "github.com/ClusterCockpit/cc-lib/ccLogger"
	"github.com/nats-io/nats.go"
)

var (
	clientOnce     sync.Once
	clientInstance *Client
)

// Client wraps a NATS connection with subscription management.
type Client struct {
	conn          *nats.Conn
	subscriptions []*nats.Subscription
	mu            sync.Mutex
}

// MessageHandler is a callback function for processing received messages.
type MessageHandler func(subject string, data []byte)

// Connect initializes the singleton NATS client using the global Keys config.
func Connect() {
	clientOnce.Do(func() {
		if Keys.Address == "" {
			cclog.Warn("NATS: no address configured, skipping connection")
			return
		}

		client, err := NewClient(nil)
		if err != nil {
			cclog.Errorf("NATS connection failed: %v", err)
			return
		}

		clientInstance = client
	})
}

// GetClient returns the singleton NATS client instance.
func GetClient() *Client {
	if clientInstance == nil {
		cclog.Warn("NATS client not initialized")
	}
	return clientInstance
}

// NewClient creates a new NATS client. If cfg is nil, uses the global Keys config.
func NewClient(cfg *NatsConfig) (*Client, error) {
	if cfg == nil {
		cfg = &Keys
	}

	if cfg.Address == "" {
		return nil, fmt.Errorf("NATS address is required")
	}

	var opts []nats.Option

	if cfg.Username != "" && cfg.Password != "" {
		opts = append(opts, nats.UserInfo(cfg.Username, cfg.Password))
	}

	if cfg.CredsFilePath != "" {
		opts = append(opts, nats.UserCredentials(cfg.CredsFilePath))
	}

	opts = append(opts, nats.DisconnectErrHandler(func(_ *nats.Conn, err error) {
		if err != nil {
			cclog.Warnf("NATS disconnected: %v", err)
		}
	}))

	opts = append(opts, nats.ReconnectHandler(func(nc *nats.Conn) {
		cclog.Infof("NATS reconnected to %s", nc.ConnectedUrl())
	}))

	opts = append(opts, nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
		cclog.Errorf("NATS error: %v", err)
	}))

	nc, err := nats.Connect(cfg.Address, opts...)
	if err != nil {
		return nil, fmt.Errorf("NATS connect failed: %w", err)
	}

	cclog.Infof("NATS connected to %s", cfg.Address)

	return &Client{
		conn:          nc,
		subscriptions: make([]*nats.Subscription, 0),
	}, nil
}

// Subscribe registers a handler for messages on the given subject.
func (c *Client) Subscribe(subject string, handler MessageHandler) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	sub, err := c.conn.Subscribe(subject, func(msg *nats.Msg) {
		handler(msg.Subject, msg.Data)
	})
	if err != nil {
		return fmt.Errorf("NATS subscribe to '%s' failed: %w", subject, err)
	}

	c.subscriptions = append(c.subscriptions, sub)
	cclog.Infof("NATS subscribed to '%s'", subject)
	return nil
}

// SubscribeQueue registers a handler with queue group for load-balanced message processing.
func (c *Client) SubscribeQueue(subject, queue string, handler MessageHandler) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	sub, err := c.conn.QueueSubscribe(subject, queue, func(msg *nats.Msg) {
		handler(msg.Subject, msg.Data)
	})
	if err != nil {
		return fmt.Errorf("NATS queue subscribe to '%s' (queue: %s) failed: %w", subject, queue, err)
	}

	c.subscriptions = append(c.subscriptions, sub)
	cclog.Infof("NATS queue subscribed to '%s' (queue: %s)", subject, queue)
	return nil
}

// SubscribeChan subscribes to a subject and delivers messages to the provided channel.
func (c *Client) SubscribeChan(subject string, ch chan *nats.Msg) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	sub, err := c.conn.ChanSubscribe(subject, ch)
	if err != nil {
		return fmt.Errorf("NATS chan subscribe to '%s' failed: %w", subject, err)
	}

	c.subscriptions = append(c.subscriptions, sub)
	cclog.Infof("NATS chan subscribed to '%s'", subject)
	return nil
}

// Publish sends data to the specified subject.
func (c *Client) Publish(subject string, data []byte) error {
	if err := c.conn.Publish(subject, data); err != nil {
		return fmt.Errorf("NATS publish to '%s' failed: %w", subject, err)
	}
	return nil
}

// Request sends a request and waits for a response with the given context timeout.
func (c *Client) Request(subject string, data []byte, timeout context.Context) ([]byte, error) {
	msg, err := c.conn.RequestWithContext(timeout, subject, data)
	if err != nil {
		return nil, fmt.Errorf("NATS request to '%s' failed: %w", subject, err)
	}
	return msg.Data, nil
}

// Flush flushes the connection buffer to ensure all published messages are sent.
func (c *Client) Flush() error {
	return c.conn.Flush()
}

// Close unsubscribes all subscriptions and closes the NATS connection.
func (c *Client) Close() {
	c.mu.Lock()
	defer c.mu.Unlock()

	for _, sub := range c.subscriptions {
		if err := sub.Unsubscribe(); err != nil {
			cclog.Warnf("NATS unsubscribe failed: %v", err)
		}
	}
	c.subscriptions = nil

	if c.conn != nil {
		c.conn.Close()
		cclog.Info("NATS connection closed")
	}
}

// IsConnected returns true if the client has an active connection.
func (c *Client) IsConnected() bool {
	return c.conn != nil && c.conn.IsConnected()
}

// Connection returns the underlying NATS connection for advanced usage.
func (c *Client) Connection() *nats.Conn {
	return c.conn
}