@@ -26,6 +26,7 @@ import (
2626 "github.com/benbjohnson/litestream/file"
2727 "github.com/benbjohnson/litestream/gs"
2828 "github.com/benbjohnson/litestream/internal"
29+ "github.com/benbjohnson/litestream/nats"
2930 "github.com/benbjohnson/litestream/s3"
3031 "github.com/benbjohnson/litestream/sftp"
3132)
@@ -559,6 +560,19 @@ type ReplicaConfig struct {
559560 Password string `yaml:"password"`
560561 KeyPath string `yaml:"key-path"`
561562
563+ // NATS settings
564+ JWT string `yaml:"jwt"`
565+ Seed string `yaml:"seed"`
566+ Creds string `yaml:"creds"`
567+ NKey string `yaml:"nkey"`
568+ Username string `yaml:"username"`
569+ Token string `yaml:"token"`
570+ TLS bool `yaml:"tls"`
571+ RootCAs []string `yaml:"root-cas"`
572+ MaxReconnects * int `yaml:"max-reconnects"`
573+ ReconnectWait * time.Duration `yaml:"reconnect-wait"`
574+ Timeout * time.Duration `yaml:"timeout"`
575+
562576 // Encryption identities and recipients
563577 Age struct {
564578 Identities []string `yaml:"identities"`
@@ -617,6 +631,10 @@ func NewReplicaFromConfig(c *ReplicaConfig, db *litestream.DB) (_ *litestream.Re
617631 if r .Client , err = newSFTPReplicaClientFromConfig (c , r ); err != nil {
618632 return nil , err
619633 }
634+ case "nats" :
635+ if r .Client , err = newNATSReplicaClientFromConfig (c , r ); err != nil {
636+ return nil , err
637+ }
620638 default :
621639 return nil , fmt .Errorf ("unknown replica type in config: %q" , c .Type )
622640 }
@@ -850,6 +868,72 @@ func newSFTPReplicaClientFromConfig(c *ReplicaConfig, _ *litestream.Replica) (_
850868 return client , nil
851869}
852870
871+ // newNATSReplicaClientFromConfig returns a new instance of nats.ReplicaClient built from config.
872+ func newNATSReplicaClientFromConfig (c * ReplicaConfig , _ * litestream.Replica ) (_ * nats.ReplicaClient , err error ) {
873+ // Parse URL if provided to extract bucket name and server URL
874+ var url , bucket string
875+ if c .URL != "" {
876+ scheme , host , path , err := ParseReplicaURL (c .URL )
877+ if err != nil {
878+ return nil , fmt .Errorf ("invalid NATS URL: %w" , err )
879+ }
880+ if scheme != "nats" {
881+ return nil , fmt .Errorf ("invalid scheme for NATS replica: %s" , scheme )
882+ }
883+
884+ // Reconstruct URL without bucket path
885+ if host != "" {
886+ url = fmt .Sprintf ("nats://%s" , host )
887+ }
888+
889+ // Extract bucket name from path
890+ if path != "" {
891+ bucket = strings .Trim (path , "/" )
892+ }
893+ }
894+
895+ // Use bucket from config if not extracted from URL
896+ if bucket == "" {
897+ bucket = c .Bucket
898+ }
899+
900+ // Ensure required settings are set
901+ if bucket == "" {
902+ return nil , fmt .Errorf ("bucket required for NATS replica" )
903+ }
904+
905+ // Build replica client
906+ client := nats .NewReplicaClient ()
907+ client .URL = url
908+ client .BucketName = bucket
909+
910+ // Set authentication options
911+ client .JWT = c .JWT
912+ client .Seed = c .Seed
913+ client .Creds = c .Creds
914+ client .NKey = c .NKey
915+ client .Username = c .Username
916+ client .Password = c .Password
917+ client .Token = c .Token
918+
919+ // Set TLS options
920+ client .TLS = c .TLS
921+ client .RootCAs = c .RootCAs
922+
923+ // Set connection options with defaults
924+ if c .MaxReconnects != nil {
925+ client .MaxReconnects = * c .MaxReconnects
926+ }
927+ if c .ReconnectWait != nil {
928+ client .ReconnectWait = * c .ReconnectWait
929+ }
930+ if c .Timeout != nil {
931+ client .Timeout = * c .Timeout
932+ }
933+
934+ return client , nil
935+ }
936+
853937// applyLitestreamEnv copies "LITESTREAM" prefixed environment variables to
854938// their AWS counterparts as the "AWS" prefix can be confusing when using a
855939// non-AWS S3-compatible service.
0 commit comments