zot/pkg/extensions/events/nats_sink.go

//go:build events

package events

import (
	"context"
	"fmt"

	cenats "github.com/cloudevents/sdk-go/protocol/nats/v2"
	cloudevents "github.com/cloudevents/sdk-go/v2"
	"github.com/nats-io/nats.go"

	zerr "zotregistry.dev/zot/v2/errors"
	eventsconf "zotregistry.dev/zot/v2/pkg/extensions/config/events"
)

// NATSSink implements a CloudEvents sink that publishes to NATS.
type NATSSink struct {
	cloudevents.Client

	conn   *nats.Conn
	config eventsconf.SinkConfig
}

// NewNATSSink creates a new NATS sink.
func NewNATSSink(config eventsconf.SinkConfig) (*NATSSink, error) {
	if config.Type != eventsconf.NATS {
		return nil, zerr.ErrInvalidEventSinkType
	}

	if config.Address == "" {
		return nil, zerr.ErrEventSinkAddressEmpty
	}

	opts := []nats.Option{
		nats.Name(EventSource),
		nats.Timeout(config.Timeout),
	}

	if config.Credentials != nil {
		if config.Credentials.File != nil && *config.Credentials.File != "" {
			opts = append(opts, nats.UserCredentials(*config.Credentials.File))
		} else if config.Credentials.Username != "" {
			opts = append(opts, nats.UserInfo(
				config.Credentials.Username,
				config.Credentials.Password,
			))
		}
	}

	if config.TLSConfig != nil && (config.TLSConfig.CACertFile != "" || config.TLSConfig.CertFile != "") {
		tlsConfig, err := getTLSConfig(config)
		if err != nil {
			return nil, err
		}

		opts = append(opts, nats.Secure(tlsConfig))
	}

	sender, err := cenats.NewSender(config.Address, config.Channel, opts)
	if err != nil {
		return nil, fmt.Errorf("failed to create NATS protocol: %w", err)
	}

	ceClient, err := cloudevents.NewClient(sender)
	if err != nil {
		return nil, fmt.Errorf("failed to create CloudEvents client: %w", err)
	}

	return &NATSSink{
		Client: ceClient,
		conn:   sender.Conn,
		config: config,
	}, nil
}

// Emit sends a CloudEvent to NATS.
func (s *NATSSink) Emit(event *cloudevents.Event) cloudevents.Result {
	if err := event.Validate(); err != nil {
		return err
	}

	ctx, cancel := context.WithTimeout(context.Background(), s.config.Timeout)
	defer cancel()

	return s.Send(ctx, *event)
}

// Close closes the NATS connection.
func (s *NATSSink) Close() error {
	if s.conn != nil {
		s.conn.Close()
	}

	return nil
}