Implement notifications service with shared messaging and email drivers (#36)
This commit is contained in:
Stephan D committed
1 parent
36277b6c45
commit
5c2d36281b
54 files changed
+3746
-6
No files matched your search
@@ -0,0 +1,304 @@
|
||||
package natsb
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"net/url"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.sendico.io/tech/sxchnge/shared/errors"
|
||||
"git.sendico.io/tech/sxchnge/shared/logging"
|
||||
nc "git.sendico.io/tech/sxchnge/shared/messaging/internal/natsb/config"
|
||||
"github.com/nats-io/nats.go"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
type NatsBroker struct {
|
||||
logger mlogger.Logger
|
||||
nc *nats.Conn
|
||||
js nats.JetStreamContext
|
||||
}
|
||||
|
||||
type envConfig struct {
|
||||
User, Password, Host string
|
||||
Port int
|
||||
}
|
||||
|
||||
const redactedNATSPassword = "xxxxx"
|
||||
|
||||
func buildSafePublishableNATSURL(rawURL string) string {
|
||||
if rawURL == "" {
|
||||
return rawURL
|
||||
}
|
||||
|
||||
parts := strings.Split(rawURL, ",")
|
||||
safe := make([]string, 0, len(parts))
|
||||
for _, part := range parts {
|
||||
trimmed := strings.TrimSpace(part)
|
||||
if trimmed == "" {
|
||||
continue
|
||||
}
|
||||
built, ok := buildSafePublishableNATSEntry(trimmed)
|
||||
if !ok {
|
||||
safe = append(safe, "<invalid NATS URL>")
|
||||
continue
|
||||
}
|
||||
safe = append(safe, built)
|
||||
}
|
||||
|
||||
if len(safe) == 0 {
|
||||
return "<invalid NATS URL>"
|
||||
}
|
||||
return strings.Join(safe, ",")
|
||||
}
|
||||
|
||||
func buildSafePublishableNATSEntry(raw string) (string, bool) {
|
||||
parsed, err := url.Parse(raw)
|
||||
if err == nil && parsed.Host != "" {
|
||||
safe := &url.URL{
|
||||
Scheme: parsed.Scheme,
|
||||
Host: parsed.Host,
|
||||
Path: parsed.Path,
|
||||
RawPath: parsed.RawPath,
|
||||
RawQuery: "",
|
||||
Fragment: "",
|
||||
}
|
||||
if safe.Scheme == "" {
|
||||
safe.Scheme = "nats"
|
||||
}
|
||||
if parsed.User != nil {
|
||||
username := redactedNATSPassword
|
||||
if username == "" {
|
||||
username = redactedNATSPassword
|
||||
}
|
||||
safe.User = url.UserPassword(redactedNATSPassword, redactedNATSPassword)
|
||||
}
|
||||
return safe.String(), true
|
||||
}
|
||||
|
||||
return buildSafePublishableFromAuthority(raw)
|
||||
}
|
||||
|
||||
func buildSafePublishableFromAuthority(raw string) (string, bool) {
|
||||
scheme := "nats"
|
||||
authorityAndSuffix := raw
|
||||
if schemeIndex := strings.Index(raw, "://"); schemeIndex >= 0 {
|
||||
if candidate := strings.TrimSpace(raw[:schemeIndex]); candidate != "" {
|
||||
scheme = candidate
|
||||
}
|
||||
authorityAndSuffix = raw[schemeIndex+3:]
|
||||
}
|
||||
|
||||
authorityEnd := strings.IndexAny(authorityAndSuffix, "/?#")
|
||||
if authorityEnd < 0 {
|
||||
authorityEnd = len(authorityAndSuffix)
|
||||
}
|
||||
|
||||
authority := authorityAndSuffix[:authorityEnd]
|
||||
suffix := authorityAndSuffix[authorityEnd:]
|
||||
atIndex := strings.LastIndex(authority, "@")
|
||||
hostPort := authority
|
||||
username := ""
|
||||
if atIndex >= 0 {
|
||||
userInfo := authority[:atIndex]
|
||||
hostPort = authority[atIndex+1:]
|
||||
if hostPort == "" {
|
||||
return "", false
|
||||
}
|
||||
username = userInfo
|
||||
if colonIndex := strings.Index(userInfo, ":"); colonIndex >= 0 {
|
||||
username = userInfo[:colonIndex]
|
||||
}
|
||||
if username == "" {
|
||||
username = redactedNATSPassword
|
||||
}
|
||||
}
|
||||
|
||||
if hostPort == "" || strings.ContainsAny(hostPort, " \t\r\n") {
|
||||
return "", false
|
||||
}
|
||||
|
||||
safe := &url.URL{
|
||||
Scheme: scheme,
|
||||
Host: hostPort,
|
||||
}
|
||||
if username != "" {
|
||||
safe.User = url.UserPassword(redactedNATSPassword, redactedNATSPassword)
|
||||
}
|
||||
_ = suffix
|
||||
return safe.String(), true
|
||||
}
|
||||
|
||||
// loadEnv gathers and validates connection details from environment variables
|
||||
// listed in the Settings struct. Invalid or missing values surface as a typed
|
||||
// InvalidArgument error so callers can decide how to handle them.
|
||||
func loadEnv(settings *nc.Settings, l mlogger.Logger) (*envConfig, error) {
|
||||
get := func(key, label string) (string, error) {
|
||||
if v := os.Getenv(key); v != "" {
|
||||
return v, nil
|
||||
}
|
||||
l.Error(fmt.Sprintf("NATS %s not found in environment", label), zap.String("env_var", key))
|
||||
return "", merrors.InvalidArgument(fmt.Sprintf("NATS %s not found in environment variable: %s", label, key), key)
|
||||
}
|
||||
|
||||
user, err := get(settings.UsernameEnv, "user name")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
password, err := get(settings.PasswordEnv, "password")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
host, err := get(settings.HostEnv, "host")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
portStr, err := get(settings.PortEnv, "port")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
port, err := strconv.Atoi(portStr)
|
||||
if err != nil || port <= 0 || port > 65535 {
|
||||
l.Error("Invalid NATS port value", zap.String("port", portStr))
|
||||
return nil, merrors.InvalidArgument("Invalid NATS port: "+portStr, settings.PortEnv)
|
||||
}
|
||||
|
||||
return &envConfig{
|
||||
User: user,
|
||||
Password: password,
|
||||
Host: host,
|
||||
Port: port,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func NewNatsBroker(logger mlogger.Logger, settings *nc.Settings) (*NatsBroker, error) {
|
||||
if settings == nil {
|
||||
return nil, merrors.InvalidArgument("NATS settings are required")
|
||||
}
|
||||
l := logger.Named("broker")
|
||||
var err error
|
||||
var cfg *envConfig
|
||||
var natsURL string
|
||||
if settings != nil && strings.TrimSpace(settings.URLEnv) != "" {
|
||||
urlVal := strings.TrimSpace(os.Getenv(settings.URLEnv))
|
||||
if urlVal != "" {
|
||||
natsURL = urlVal
|
||||
}
|
||||
}
|
||||
if natsURL == "" {
|
||||
// Helper function to get environment variables
|
||||
cfg, err = loadEnv(settings, l)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
u := &url.URL{
|
||||
Scheme: "nats",
|
||||
Host: net.JoinHostPort(cfg.Host, strconv.Itoa(cfg.Port)),
|
||||
}
|
||||
natsURL = u.String()
|
||||
}
|
||||
publishableNATSURL := buildSafePublishableNATSURL(natsURL)
|
||||
|
||||
opts := []nats.Option{
|
||||
nats.Name(settings.NATSName),
|
||||
nats.MaxReconnects(settings.MaxReconnects),
|
||||
nats.ReconnectWait(time.Duration(settings.ReconnectWait) * time.Second),
|
||||
nats.RetryOnFailedConnect(true),
|
||||
nats.DisconnectErrHandler(func(conn *nats.Conn, err error) {
|
||||
fields := []zap.Field{
|
||||
zap.String("broker", settings.NATSName),
|
||||
}
|
||||
if conn != nil {
|
||||
fields = append(fields, zap.String("connected_url", buildSafePublishableNATSURL(conn.ConnectedUrl())))
|
||||
}
|
||||
if err != nil {
|
||||
fields = append(fields, zap.Error(err))
|
||||
}
|
||||
l.Warn("Disconnected from NATS", fields...)
|
||||
}),
|
||||
nats.ReconnectHandler(func(conn *nats.Conn) {
|
||||
fields := []zap.Field{
|
||||
zap.String("broker", settings.NATSName),
|
||||
}
|
||||
if conn != nil {
|
||||
fields = append(fields, zap.String("connected_url", buildSafePublishableNATSURL(conn.ConnectedUrl())))
|
||||
}
|
||||
l.Info("Reconnected to NATS", fields...)
|
||||
}),
|
||||
nats.ClosedHandler(func(conn *nats.Conn) {
|
||||
fields := []zap.Field{
|
||||
zap.String("broker", settings.NATSName),
|
||||
}
|
||||
if conn != nil {
|
||||
if url := conn.ConnectedUrl(); url != "" {
|
||||
fields = append(fields, zap.String("connected_url", buildSafePublishableNATSURL(url)))
|
||||
}
|
||||
if err := conn.LastError(); err != nil {
|
||||
fields = append(fields, zap.Error(err))
|
||||
}
|
||||
}
|
||||
l.Warn("NATS connection closed", fields...)
|
||||
}),
|
||||
}
|
||||
if cfg != nil {
|
||||
opts = append(opts, nats.UserInfo(cfg.User, cfg.Password))
|
||||
} else if settings != nil {
|
||||
userEnv := strings.TrimSpace(settings.UsernameEnv)
|
||||
passEnv := strings.TrimSpace(settings.PasswordEnv)
|
||||
if userEnv != "" && passEnv != "" {
|
||||
user := strings.TrimSpace(os.Getenv(userEnv))
|
||||
pass := strings.TrimSpace(os.Getenv(passEnv))
|
||||
if user != "" || pass != "" {
|
||||
opts = append(opts, nats.UserInfo(user, pass))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
res := &NatsBroker{
|
||||
logger: l.Named("nats"),
|
||||
}
|
||||
|
||||
if res.nc, err = nats.Connect(natsURL, opts...); err != nil {
|
||||
l.Error("Failed to connect to NATS", zap.String("url", publishableNATSURL), zap.Error(err))
|
||||
return nil, err
|
||||
}
|
||||
if res.js, err = res.nc.JetStream(); err != nil {
|
||||
l.Warn("Failed to initialise JetStream context", zap.Error(err))
|
||||
}
|
||||
|
||||
logger.Info("Connected to NATS", zap.String("broker", settings.NATSName),
|
||||
zap.String("url", publishableNATSURL))
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (b *NatsBroker) JetStream() nats.JetStreamContext {
|
||||
if b == nil {
|
||||
return nil
|
||||
}
|
||||
return b.js
|
||||
}
|
||||
|
||||
func (b *NatsBroker) HasJetStream() bool {
|
||||
return b != nil && b.js != nil
|
||||
}
|
||||
|
||||
func (b *NatsBroker) Logger() mlogger.Logger {
|
||||
if b == nil {
|
||||
return nil
|
||||
}
|
||||
return b.logger
|
||||
}
|
||||
|
||||
func (b *NatsBroker) Close() {
|
||||
if b != nil && b.nc != nil {
|
||||
b.nc.Close()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
package natsb
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestLogsRedactNATSCredentials(t *testing.T) {
|
||||
for _, url := range []string{"nats://secret-user:secret-password@localhost:4222", "nats://secret-token@localhost:4222", "nats://localhost:4222?token=secret-token", "invalid secret-password"} {
|
||||
safe := buildSafePublishableNATSURL(url)
|
||||
if strings.Contains(safe, "secret") {
|
||||
t.Fatalf("credential leaked: %q", safe)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
package natsb
|
||||
|
||||
type Settings struct {
|
||||
URLEnv string `mapstructure:"url_env" yaml:"url_env"`
|
||||
HostEnv string `mapstructure:"host_env" yaml:"host_env"`
|
||||
PortEnv string `mapstructure:"port_env" yaml:"port_env"`
|
||||
UsernameEnv string `mapstructure:"username_env" yaml:"username_env"`
|
||||
PasswordEnv string `mapstructure:"password_env" yaml:"password_env"`
|
||||
NATSName string `mapstructure:"broker_name" yaml:"broker_name"`
|
||||
MaxReconnects int `mapstructure:"max_reconnects" yaml:"max_reconnects"`
|
||||
ReconnectWait int `mapstructure:"reconnect_wait" yaml:"reconnect_wait"`
|
||||
BufferSize int `mapstructure:"buffer_size" yaml:"buffer_size"`
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
package natscap
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
|
||||
"git.sendico.io/tech/sxchnge/shared/messaging/capability"
|
||||
basenats "git.sendico.io/tech/sxchnge/shared/messaging/internal/natsb"
|
||||
"git.sendico.io/tech/sxchnge/shared/messaging/reliable"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
type DurableAdapter struct {
|
||||
broker *basenats.NatsBroker
|
||||
}
|
||||
|
||||
type durableAdminContext interface {
|
||||
AddStream(cfg *nats.StreamConfig, opts ...nats.JSOpt) (*nats.StreamInfo, error)
|
||||
UpdateStream(cfg *nats.StreamConfig, opts ...nats.JSOpt) (*nats.StreamInfo, error)
|
||||
StreamInfo(stream string, opts ...nats.JSOpt) (*nats.StreamInfo, error)
|
||||
AddConsumer(stream string, cfg *nats.ConsumerConfig, opts ...nats.JSOpt) (*nats.ConsumerInfo, error)
|
||||
UpdateConsumer(stream string, cfg *nats.ConsumerConfig, opts ...nats.JSOpt) (*nats.ConsumerInfo, error)
|
||||
ConsumerInfo(stream, consumer string, opts ...nats.JSOpt) (*nats.ConsumerInfo, error)
|
||||
}
|
||||
|
||||
func NewDurableAdapter(broker *basenats.NatsBroker) *DurableAdapter {
|
||||
if broker == nil || broker.JetStream() == nil {
|
||||
return nil
|
||||
}
|
||||
return &DurableAdapter{broker: broker}
|
||||
}
|
||||
|
||||
func (a *DurableAdapter) Publish(ctx context.Context, record reliable.Record) error {
|
||||
if err := reliable.ValidatePublicRecord(record); err != nil {
|
||||
return err
|
||||
}
|
||||
normalized, err := reliable.NormalizePublicRecord(record)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return a.publishRecord(ctx, normalized)
|
||||
}
|
||||
|
||||
func (a *DurableAdapter) PublishRuntime(ctx context.Context, record reliable.Record) error {
|
||||
normalized, err := reliable.NormalizeRuntimeRecord(record)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return a.publishRecord(ctx, normalized)
|
||||
}
|
||||
|
||||
func (a *DurableAdapter) publishRecord(ctx context.Context, record reliable.Record) error {
|
||||
js := a.jetStream()
|
||||
if js == nil {
|
||||
return capability.Unavailable("durable transport")
|
||||
}
|
||||
msg := &nats.Msg{
|
||||
Subject: strings.TrimSpace(record.Subject),
|
||||
Data: append([]byte(nil), record.Payload...),
|
||||
Header: nats.Header{},
|
||||
}
|
||||
for key, values := range record.Headers {
|
||||
for _, value := range values {
|
||||
msg.Header.Add(key, value)
|
||||
}
|
||||
}
|
||||
if record.MessageID != "" {
|
||||
msg.Header.Set(nats.MsgIdHdr, record.MessageID)
|
||||
}
|
||||
_, err := js.PublishMsg(msg, nats.Context(ctx))
|
||||
return err
|
||||
}
|
||||
|
||||
func (a *DurableAdapter) OpenConsumer(ctx context.Context, settings reliable.DurableSettings, consumer reliable.DurableConsumerSettings) (reliable.Subscription, error) {
|
||||
js := a.jetStream()
|
||||
if js == nil {
|
||||
return nil, capability.Unavailable("durable transport")
|
||||
}
|
||||
sub, err := js.PullSubscribe(
|
||||
consumer.Subject,
|
||||
consumer.Durable,
|
||||
nats.Bind(settings.Stream.Name, consumer.Durable),
|
||||
nats.Context(ctx),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &natsSubscription{sub: sub}, nil
|
||||
}
|
||||
|
||||
func (a *DurableAdapter) EnsureProducerResources(_ context.Context, settings reliable.DurableSettings) error {
|
||||
if a == nil || a.broker == nil {
|
||||
return capability.Unavailable("durable admin")
|
||||
}
|
||||
return ensureDurableStream(a.broker.Logger(), a.jetStream(), settings)
|
||||
}
|
||||
|
||||
func (a *DurableAdapter) EnsureConsumerResources(_ context.Context, settings reliable.DurableSettings, consumer reliable.DurableConsumerSettings) error {
|
||||
if a == nil || a.broker == nil {
|
||||
return capability.Unavailable("durable admin")
|
||||
}
|
||||
js := a.jetStream()
|
||||
if err := ensureDurableStream(a.broker.Logger(), js, settings); err != nil {
|
||||
return err
|
||||
}
|
||||
return ensureDurableConsumerConfig(a.broker.Logger(), js, settings, consumer)
|
||||
}
|
||||
|
||||
func (a *DurableAdapter) jetStream() nats.JetStreamContext {
|
||||
if a == nil || a.broker == nil {
|
||||
return nil
|
||||
}
|
||||
return a.broker.JetStream()
|
||||
}
|
||||
@@ -0,0 +1,282 @@
|
||||
package natscap
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
||||
"git.sendico.io/tech/sxchnge/shared/messaging/capability"
|
||||
"git.sendico.io/tech/sxchnge/shared/messaging/reliable"
|
||||
"github.com/nats-io/nats.go"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
func ensureDurableStream(logger *zap.Logger, js durableAdminContext, settings reliable.DurableSettings) error {
|
||||
if js == nil {
|
||||
return capability.Unavailable("durable admin")
|
||||
}
|
||||
expected := &nats.StreamConfig{
|
||||
Name: settings.Stream.Name,
|
||||
Subjects: append([]string(nil), settings.Stream.Subjects...),
|
||||
Retention: nats.LimitsPolicy,
|
||||
MaxAge: settings.StreamMaxAge(),
|
||||
Replicas: settings.Stream.Replicas,
|
||||
}
|
||||
if strings.EqualFold(strings.TrimSpace(settings.Stream.Storage), "memory") {
|
||||
expected.Storage = nats.MemoryStorage
|
||||
} else {
|
||||
expected.Storage = nats.FileStorage
|
||||
}
|
||||
|
||||
info, err := js.StreamInfo(expected.Name)
|
||||
if err != nil {
|
||||
if errors.Is(err, nats.ErrStreamNotFound) {
|
||||
if !settings.AllowsAutoProvision() {
|
||||
if logger != nil {
|
||||
logger.Warn("Durable stream is missing and auto provisioning is disabled", zap.String("stream", expected.Name))
|
||||
}
|
||||
return err
|
||||
}
|
||||
_, err = js.AddStream(expected)
|
||||
return err
|
||||
}
|
||||
return err
|
||||
}
|
||||
current := info.Config
|
||||
if err := validateStreamConfig(current, *expected, settings.IsProductionValidation()); err != nil {
|
||||
if settings.AllowsAutoProvision() {
|
||||
reconciled, reconcileErr := reconcileDurableStream(logger, js, current, *expected, settings)
|
||||
if reconciled {
|
||||
return reconcileErr
|
||||
}
|
||||
if reconcileErr != nil {
|
||||
return reconcileErr
|
||||
}
|
||||
}
|
||||
return err
|
||||
}
|
||||
if settings.AllowsAutoProvision() && current.MaxAge != expected.MaxAge {
|
||||
current.MaxAge = expected.MaxAge
|
||||
_, err = js.UpdateStream(¤t)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func reconcileDurableStream(logger *zap.Logger, js durableAdminContext, current nats.StreamConfig, expected nats.StreamConfig, settings reliable.DurableSettings) (bool, error) {
|
||||
if js == nil {
|
||||
return false, capability.Unavailable("durable admin")
|
||||
}
|
||||
if strings.TrimSpace(current.Name) != strings.TrimSpace(expected.Name) {
|
||||
return false, merrors.DataConflictf("durable stream name drift: got=%s want=%s", current.Name, expected.Name)
|
||||
}
|
||||
if current.Storage != expected.Storage {
|
||||
return false, merrors.DataConflictf("durable stream storage drift: got=%s want=%s", current.Storage.String(), expected.Storage.String())
|
||||
}
|
||||
if current.Retention != expected.Retention {
|
||||
return false, merrors.DataConflictf("durable stream retention drift: got=%s want=%s", current.Retention.String(), expected.Retention.String())
|
||||
}
|
||||
if current.Replicas != expected.Replicas {
|
||||
return false, merrors.DataConflictf("durable stream replicas drift: got=%d want=%d", current.Replicas, expected.Replicas)
|
||||
}
|
||||
missing, extra := stringSetDiff(expected.Subjects, current.Subjects), stringSetDiff(current.Subjects, expected.Subjects)
|
||||
if len(extra) > 0 {
|
||||
return false, merrors.DataConflict("durable stream has unexpected subjects")
|
||||
}
|
||||
|
||||
if len(missing) == 0 {
|
||||
return false, nil
|
||||
}
|
||||
current.Subjects = append([]string(nil), expected.Subjects...)
|
||||
current.MaxAge = expected.MaxAge
|
||||
_, err := js.UpdateStream(¤t)
|
||||
if err != nil {
|
||||
return true, err
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func ensureDurableConsumerConfig(logger *zap.Logger, js durableAdminContext, settings reliable.DurableSettings, consumer reliable.DurableConsumerSettings) error {
|
||||
if js == nil {
|
||||
return capability.Unavailable("durable admin")
|
||||
}
|
||||
deliverPolicy, err := natsDeliverPolicy(consumer.DeliverPolicy)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
expected := &nats.ConsumerConfig{
|
||||
Durable: consumer.Durable,
|
||||
DeliverPolicy: deliverPolicy,
|
||||
AckPolicy: nats.AckExplicitPolicy,
|
||||
AckWait: time.Duration(consumer.AckWaitSeconds) * time.Second,
|
||||
MaxDeliver: consumer.MaxDeliver,
|
||||
FilterSubject: consumer.Subject,
|
||||
ReplayPolicy: nats.ReplayInstantPolicy,
|
||||
}
|
||||
if consumer.InactiveThresholdSeconds > 0 {
|
||||
expected.InactiveThreshold = time.Duration(consumer.InactiveThresholdSeconds) * time.Second
|
||||
}
|
||||
info, err := js.ConsumerInfo(settings.Stream.Name, consumer.Durable)
|
||||
if err != nil {
|
||||
if errors.Is(err, nats.ErrConsumerNotFound) {
|
||||
if !settings.AllowsAutoProvision() {
|
||||
if logger != nil {
|
||||
logger.Warn("Durable consumer is missing and auto provisioning is disabled", zap.String("stream", settings.Stream.Name), zap.String("subject", consumer.Subject), zap.String("durable", consumer.Durable))
|
||||
}
|
||||
return err
|
||||
}
|
||||
_, err = js.AddConsumer(settings.Stream.Name, expected)
|
||||
return err
|
||||
}
|
||||
return err
|
||||
}
|
||||
current := info.Config
|
||||
if current.DeliverPolicy != expected.DeliverPolicy && logger != nil {
|
||||
logger.Warn("Durable consumer deliver policy drift requires delete and recreate",
|
||||
zap.String("stream", settings.Stream.Name),
|
||||
zap.String("subject", consumer.Subject),
|
||||
zap.String("durable", consumer.Durable),
|
||||
zap.String("current_deliver_policy", natsDeliverPolicyName(current.DeliverPolicy)),
|
||||
zap.String("expected_deliver_policy", natsDeliverPolicyName(expected.DeliverPolicy)),
|
||||
zap.Bool("recreate_required", true),
|
||||
)
|
||||
}
|
||||
if err := validateConsumerConfig(current, *expected, settings.IsProductionValidation()); err != nil {
|
||||
return err
|
||||
}
|
||||
if settings.AllowsAutoProvision() && (current.AckWait != expected.AckWait || current.MaxDeliver != expected.MaxDeliver || current.InactiveThreshold != expected.InactiveThreshold) {
|
||||
current.AckWait = expected.AckWait
|
||||
current.MaxDeliver = expected.MaxDeliver
|
||||
current.InactiveThreshold = expected.InactiveThreshold
|
||||
_, err = js.UpdateConsumer(settings.Stream.Name, ¤t)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func natsDeliverPolicy(deliveryMode reliable.DurableDeliverPolicy) (nats.DeliverPolicy, error) {
|
||||
switch deliveryMode {
|
||||
case "", reliable.DurableDeliverAll:
|
||||
return nats.DeliverAllPolicy, nil
|
||||
case reliable.DurableDeliverNew:
|
||||
return nats.DeliverNewPolicy, nil
|
||||
default:
|
||||
return nats.DeliverAllPolicy, merrors.InvalidArgumentf("unsupported durable consumer deliver policy: %q", deliveryMode)
|
||||
}
|
||||
}
|
||||
|
||||
func natsDeliverPolicyName(deliveryMode nats.DeliverPolicy) string {
|
||||
switch deliveryMode {
|
||||
case nats.DeliverAllPolicy:
|
||||
return "all"
|
||||
case nats.DeliverLastPolicy:
|
||||
return "last"
|
||||
case nats.DeliverNewPolicy:
|
||||
return "new"
|
||||
case nats.DeliverByStartSequencePolicy:
|
||||
return "by_start_sequence"
|
||||
case nats.DeliverByStartTimePolicy:
|
||||
return "by_start_time"
|
||||
case nats.DeliverLastPerSubjectPolicy:
|
||||
return "last_per_subject"
|
||||
default:
|
||||
return "unknown"
|
||||
}
|
||||
}
|
||||
|
||||
func validateStreamConfig(current nats.StreamConfig, expected nats.StreamConfig, strict bool) error {
|
||||
if strings.TrimSpace(current.Name) != strings.TrimSpace(expected.Name) {
|
||||
return merrors.DataConflictf("durable stream name drift: got=%s want=%s", current.Name, expected.Name)
|
||||
}
|
||||
if !sameStrings(current.Subjects, expected.Subjects) {
|
||||
return merrors.DataConflictf("durable stream subjects drift: current=%v expected=%v", current.Subjects, expected.Subjects)
|
||||
}
|
||||
if current.Storage != expected.Storage {
|
||||
return merrors.DataConflict("durable stream storage drift")
|
||||
}
|
||||
if current.Retention != expected.Retention {
|
||||
return merrors.DataConflict("durable stream retention drift")
|
||||
}
|
||||
if current.Replicas != expected.Replicas {
|
||||
return merrors.DataConflict("durable stream replicas drift")
|
||||
}
|
||||
if strict && current.MaxAge != expected.MaxAge {
|
||||
return nil
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateConsumerConfig(current nats.ConsumerConfig, expected nats.ConsumerConfig, strict bool) error {
|
||||
if strings.TrimSpace(current.Durable) != strings.TrimSpace(expected.Durable) {
|
||||
return merrors.DataConflict("durable consumer name drift")
|
||||
}
|
||||
if strings.TrimSpace(current.FilterSubject) != strings.TrimSpace(expected.FilterSubject) {
|
||||
return merrors.DataConflict("durable consumer filter subject drift")
|
||||
}
|
||||
if current.DeliverPolicy != expected.DeliverPolicy {
|
||||
return merrors.DataConflict("durable consumer deliver policy drift; delete and recreate consumer to apply deliver policy")
|
||||
}
|
||||
if current.AckPolicy != expected.AckPolicy {
|
||||
return merrors.DataConflict("durable consumer ack policy drift")
|
||||
}
|
||||
if strict && current.AckWait != expected.AckWait {
|
||||
return merrors.DataConflict("durable consumer ack wait drift")
|
||||
}
|
||||
if strict && current.MaxDeliver != expected.MaxDeliver {
|
||||
return merrors.DataConflict("durable consumer max deliver drift")
|
||||
}
|
||||
if strict && current.InactiveThreshold != expected.InactiveThreshold {
|
||||
return merrors.DataConflict("durable consumer inactive threshold drift")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func sameStrings(left, right []string) bool {
|
||||
if len(left) != len(right) {
|
||||
return false
|
||||
}
|
||||
seen := map[string]int{}
|
||||
for _, value := range left {
|
||||
seen[strings.TrimSpace(value)]++
|
||||
}
|
||||
for _, value := range right {
|
||||
key := strings.TrimSpace(value)
|
||||
if seen[key] == 0 {
|
||||
return false
|
||||
}
|
||||
seen[key]--
|
||||
}
|
||||
for _, count := range seen {
|
||||
if count != 0 {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func stringSetDiff(left, right []string) []string {
|
||||
if len(left) == 0 {
|
||||
return nil
|
||||
}
|
||||
seen := map[string]int{}
|
||||
for _, value := range right {
|
||||
seen[strings.TrimSpace(value)]++
|
||||
}
|
||||
result := make([]string, 0)
|
||||
for _, value := range left {
|
||||
key := strings.TrimSpace(value)
|
||||
if key == "" {
|
||||
continue
|
||||
}
|
||||
if seen[key] > 0 {
|
||||
seen[key]--
|
||||
continue
|
||||
}
|
||||
result = append(result, key)
|
||||
}
|
||||
if len(result) == 0 {
|
||||
return nil
|
||||
}
|
||||
return result
|
||||
}
|
||||
@@ -0,0 +1,178 @@
|
||||
package natscap
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
||||
"git.sendico.io/tech/sxchnge/shared/messaging/capability"
|
||||
"git.sendico.io/tech/sxchnge/shared/messaging/reliable"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
type natsSubscription struct {
|
||||
mu sync.Mutex
|
||||
sub *nats.Subscription
|
||||
closed bool
|
||||
}
|
||||
|
||||
func fetchPollContext(ctx context.Context, wait time.Duration) (context.Context, context.CancelFunc, bool) {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
if wait <= 0 {
|
||||
return ctx, nil, false
|
||||
}
|
||||
if deadline, hasDeadline := ctx.Deadline(); hasDeadline && time.Until(deadline) <= wait {
|
||||
return ctx, nil, false
|
||||
}
|
||||
nextCtx, cancel := context.WithTimeout(ctx, wait)
|
||||
return nextCtx, cancel, true
|
||||
}
|
||||
|
||||
func (s *natsSubscription) Fetch(ctx context.Context, limit int, wait time.Duration) ([]reliable.Delivery, error) {
|
||||
if limit <= 0 {
|
||||
return nil, merrors.InvalidArgument("fetch limit must be greater than zero", "limit")
|
||||
}
|
||||
s.mu.Lock()
|
||||
closed := s.closed
|
||||
sub := s.sub
|
||||
s.mu.Unlock()
|
||||
if closed || sub == nil {
|
||||
return nil, capability.ErrClosed
|
||||
}
|
||||
fetchCtx, cancel, derivedDeadline := fetchPollContext(ctx, wait)
|
||||
if cancel != nil {
|
||||
defer cancel()
|
||||
}
|
||||
msgs, err := sub.Fetch(limit, nats.Context(fetchCtx))
|
||||
if err != nil {
|
||||
switch {
|
||||
case errors.Is(err, nats.ErrTimeout):
|
||||
return nil, nil
|
||||
case derivedDeadline && errors.Is(err, context.DeadlineExceeded):
|
||||
return nil, nil
|
||||
default:
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
s.mu.Lock()
|
||||
closed = s.closed
|
||||
s.mu.Unlock()
|
||||
if closed {
|
||||
return nil, capability.ErrClosed
|
||||
}
|
||||
result := make([]reliable.Delivery, 0, len(msgs))
|
||||
for _, msg := range msgs {
|
||||
if msg == nil {
|
||||
continue
|
||||
}
|
||||
result = append(result, &natsDelivery{sub: s, msg: msg, record: messageRecord(msg), attempt: deliveryAttempt(msg)})
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (s *natsSubscription) Close() error {
|
||||
s.mu.Lock()
|
||||
if s.closed {
|
||||
s.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
s.closed = true
|
||||
sub := s.sub
|
||||
s.sub = nil
|
||||
s.mu.Unlock()
|
||||
if sub == nil {
|
||||
return nil
|
||||
}
|
||||
if err := sub.Unsubscribe(); err != nil && !errors.Is(err, nats.ErrBadSubscription) {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *natsSubscription) isClosed() bool {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.closed
|
||||
}
|
||||
|
||||
type natsDelivery struct {
|
||||
sub *natsSubscription
|
||||
msg *nats.Msg
|
||||
record reliable.Record
|
||||
attempt int
|
||||
|
||||
mu sync.Mutex
|
||||
acked bool
|
||||
}
|
||||
|
||||
func (d *natsDelivery) Record() reliable.Record {
|
||||
return d.record
|
||||
}
|
||||
|
||||
func (d *natsDelivery) Attempt() int {
|
||||
if d.attempt <= 0 {
|
||||
return 1
|
||||
}
|
||||
return d.attempt
|
||||
}
|
||||
|
||||
func (d *natsDelivery) Ack(ctx context.Context) error {
|
||||
d.mu.Lock()
|
||||
if d.acked {
|
||||
d.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
if d.sub != nil && d.sub.isClosed() {
|
||||
d.mu.Unlock()
|
||||
return capability.ErrClosed
|
||||
}
|
||||
msg := d.msg
|
||||
d.mu.Unlock()
|
||||
if msg == nil {
|
||||
return capability.ErrClosed
|
||||
}
|
||||
if err := msg.AckSync(nats.Context(ctx)); err != nil {
|
||||
return err
|
||||
}
|
||||
d.mu.Lock()
|
||||
d.acked = true
|
||||
d.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func messageRecord(msg *nats.Msg) reliable.Record {
|
||||
headers := make(map[string][]string, len(msg.Header)+1)
|
||||
for key, values := range msg.Header {
|
||||
headers[strings.ToLower(strings.TrimSpace(key))] = append([]string(nil), values...)
|
||||
}
|
||||
messageID := strings.TrimSpace(msg.Header.Get(nats.MsgIdHdr))
|
||||
if businessID := reliable.HeaderValue(headers, reliable.HeaderOriginalBusinessMsgID); businessID != "" {
|
||||
messageID = businessID
|
||||
}
|
||||
headers[reliable.HeaderTransportMessageID] = []string{strings.TrimSpace(msg.Header.Get(nats.MsgIdHdr))}
|
||||
return reliable.Record{
|
||||
Subject: strings.TrimSpace(msg.Subject),
|
||||
Payload: append([]byte(nil), msg.Data...),
|
||||
Headers: headers,
|
||||
MessageID: strings.TrimSpace(messageID),
|
||||
}
|
||||
}
|
||||
|
||||
func deliveryAttempt(msg *nats.Msg) int {
|
||||
if msg == nil {
|
||||
return 1
|
||||
}
|
||||
meta, err := msg.Metadata()
|
||||
if err != nil || meta == nil || meta.NumDelivered == 0 {
|
||||
return 1
|
||||
}
|
||||
if meta.NumDelivered > uint64(^uint(0)>>1) {
|
||||
return int(^uint(0) >> 1)
|
||||
}
|
||||
return int(meta.NumDelivered)
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
package natscap
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.sendico.io/tech/sxchnge/shared/messaging/capability"
|
||||
"git.sendico.io/tech/sxchnge/shared/messaging/reliable"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
func TestFetchPollDeadlineIsBounded(t *testing.T) {
|
||||
parent, cancel := context.WithTimeout(context.Background(), time.Hour)
|
||||
defer cancel()
|
||||
ctx, stop, derived := fetchPollContext(parent, time.Second)
|
||||
if stop != nil {
|
||||
defer stop()
|
||||
}
|
||||
deadline, ok := ctx.Deadline()
|
||||
if !ok || !derived || time.Until(deadline) > time.Second {
|
||||
t.Fatal("fetch escaped poll deadline")
|
||||
}
|
||||
}
|
||||
func TestClosedSubscriptionAndAck(t *testing.T) {
|
||||
s := &natsSubscription{}
|
||||
if err := s.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := s.Fetch(context.Background(), 1, time.Second); !errors.Is(err, capability.ErrClosed) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
d := &natsDelivery{sub: s}
|
||||
if err := d.Ack(context.Background()); !errors.Is(err, capability.ErrClosed) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
d.acked = true
|
||||
if err := d.Ack(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
func TestConsumerDriftRejected(t *testing.T) {
|
||||
want := nats.ConsumerConfig{Durable: "email", FilterSubject: "notifications.email", AckPolicy: nats.AckExplicitPolicy, DeliverPolicy: nats.DeliverAllPolicy, AckWait: time.Minute}
|
||||
current := want
|
||||
current.FilterSubject = "other"
|
||||
if err := validateConsumerConfig(current, want, true); err == nil {
|
||||
t.Fatal("subject drift accepted")
|
||||
}
|
||||
current = want
|
||||
current.AckWait = time.Second
|
||||
if err := validateConsumerConfig(current, want, true); err == nil {
|
||||
t.Fatal("ack drift accepted")
|
||||
}
|
||||
if _, err := natsDeliverPolicy(reliable.DurableDeliverPolicy("invalid")); err == nil {
|
||||
t.Fatal("invalid deliver policy accepted")
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user