feat(natsstore): select the NATS store when NATS_URL is set

cmd/server/backend.go

@@ -14,6 +14,7 @@ import (
 	"github.com/router-for-me/CLIProxyAPI/v8/internal/config"
 	"github.com/router-for-me/CLIProxyAPI/v8/internal/misc"
 	"github.com/router-for-me/CLIProxyAPI/v8/internal/store"
+	"github.com/router-for-me/CLIProxyAPI/v8/internal/store/natsstore"
 	coreauth "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/auth"
 	log "github.com/sirupsen/logrus"
 )
@@ -65,16 +66,29 @@ var storeBackends = []storeBackend{
 	{name: "postgres", detect: detectPostgresBackend},
 	{name: "object", detect: detectObjectBackend},
 	{name: "git", detect: detectGitBackend},
+	{name: "nats", detect: detectNatsBackend},
 }
 
-// selectBackend returns the opener of the first backend selected by the environment, or nil.
+// selectBackend returns the opener of the first backend selected by the
+// environment, or nil. Later backends that are also configured are ignored
+// with a warning, since a stray variable is easy to leave behind.
 func selectBackend(env backendEnv) openBackendFunc {
+	var (
+		selected openBackendFunc
+		chosen   string
+	)
 	for _, backend := range storeBackends {
-		if open := backend.detect(env); open != nil {
-			return open
+		open := backend.detect(env)
+		if open == nil {
+			continue
 		}
+		if selected == nil {
+			selected, chosen = open, backend.name
+			continue
+		}
+		log.Warnf("%s store settings are present but ignored because the %s store is selected", backend.name, chosen)
 	}
-	return nil
+	return selected
 }
 
 func detectPostgresBackend(env backendEnv) openBackendFunc {
@@ -223,6 +237,50 @@ func detectGitBackend(env backendEnv) openBackendFunc {
 	}
 }
 
+func detectNatsBackend(env backendEnv) openBackendFunc {
+	if _, ok := env.lookup(natsstore.EnvURL); !ok {
+		return nil
+	}
+	lookup := func(key string) (string, bool) { return env.lookup(key) }
+	localPath, _ := env.lookup("NATS_STORE_LOCAL_PATH", "nats_store_local_path")
+	spool := filepath.Join(env.spoolBase(localPath), "natsstore")
+	return func() (*openedBackend, error) {
+		settings, err := natsstore.StoreSettingsFromEnv(lookup)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read nats store settings: %w", err)
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		nc, err := natsstore.Connect(ctx, natsstore.SettingsFromEnv(lookup).Resolve(), lookup, 0)
+		if err != nil {
+			return nil, fmt.Errorf("failed to connect nats token store: %w", err)
+		}
+		boot, err := natsstore.Bootstrap(ctx, nc, settings)
+		if err != nil {
+			nc.Close()
+			return nil, fmt.Errorf("failed to bootstrap nats store buckets: %w", err)
+		}
+		inst, err := natsstore.NewStore(nc, boot, settings, spool)
+		if err != nil {
+			nc.Close()
+			return nil, fmt.Errorf("failed to initialize nats token store: %w", err)
+		}
+		examplePath := filepath.Join(env.workDir, "config.example.yaml")
+		if errBootstrap := inst.Bootstrap(ctx, examplePath); errBootstrap != nil {
+			_ = inst.Close()
+			return nil, fmt.Errorf("failed to bootstrap nats-backed config: %w", errBootstrap)
+		}
+		return &openedBackend{
+			tokenStore: inst,
+			configPath: inst.ConfigPath(),
+			authDir:    inst.AuthDir(),
+			ready: func() {
+				log.Infof("nats-backed token store enabled, bucket prefix: %s, workspace path: %s", settings.Prefix, inst.WorkDir())
+			},
+		}, nil
+	}
+}
+
 // loadBackendConfig loads the config a backend placed on disk and points it at the backend's auth directory.
 func loadBackendConfig(opened *openedBackend, isCloudDeploy bool) (*config.Config, error) {
 	cfg, err := config.LoadConfigOptional(opened.configPath, isCloudDeploy)

cmd/server/backend_nats_test.go

@@ -0,0 +1,117 @@
+package main
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/router-for-me/CLIProxyAPI/v8/internal/store/natsstore/natstest"
+	coreauth "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/auth"
+)
+
+// storePersister mirrors what the watcher requires of a remote store.
+type storePersister interface {
+	PersistConfig(ctx context.Context) error
+	PersistAuthFiles(ctx context.Context, message string, paths ...string) error
+}
+
+func TestNatsBackendIsSelectedByURL(t *testing.T) {
+	if selectBackend(backendEnv{lookup: envLookup(map[string]string{"NATS_CREDS": "x"}), workDir: t.TempDir()}) != nil {
+		t.Fatal("selected nats without NATS_URL")
+	}
+	if selectBackend(backendEnv{lookup: envLookup(map[string]string{"NATS_URL": "nats://127.0.0.1:1"}), workDir: t.TempDir()}) == nil {
+		t.Fatal("NATS_URL did not select a backend")
+	}
+	both := envLookup(map[string]string{"PGSTORE_DSN": "postgres://example.invalid/db", "NATS_URL": "nats://127.0.0.1:1"})
+	var got string
+	for _, backend := range storeBackends {
+		if backend.detect(backendEnv{lookup: both, workDir: t.TempDir()}) != nil {
+			got = backend.name
+			break
+		}
+	}
+	if got != "postgres" {
+		t.Fatalf("first selected backend = %q, want postgres to keep priority", got)
+	}
+}
+
+func TestNatsBackendSharesCredentialsAcrossReplicas(t *testing.T) {
+	srv := natstest.Start(t)
+	workDir := t.TempDir()
+	if err := os.WriteFile(filepath.Join(workDir, "config.example.yaml"), []byte("port: 8317\nhost: \"\"\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	openReplica := func() *openedBackend {
+		t.Helper()
+		env := backendEnv{
+			workDir: workDir,
+			lookup: envLookup(map[string]string{
+				"NATS_URL":              srv.URL,
+				"NATS_STORE_PREFIX":     "e2e",
+				"NATS_STORE_CREATE":     "auto",
+				"NATS_STORE_LOCAL_PATH": t.TempDir(),
+			}),
+		}
+		open := selectBackend(env)
+		if open == nil {
+			t.Fatal("nats backend was not selected")
+		}
+		opened, err := open()
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(func() { _ = opened.tokenStore.(interface{ Close() error }).Close() })
+		return opened
+	}
+
+	first := openReplica()
+	cfg, err := loadBackendConfig(first, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if cfg.Port != 8317 || cfg.AuthDir != first.authDir {
+		t.Fatalf("config = port %d, auth dir %q", cfg.Port, cfg.AuthDir)
+	}
+	if _, ok := first.tokenStore.(storePersister); !ok {
+		t.Fatal("token store does not implement the watcher persister")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+	auth := &coreauth.Auth{
+		ID: "team/user@example.com.json", FileName: "team/user@example.com.json", Provider: "test",
+		Metadata: map[string]any{"type": "test", "email": "user@example.com", "access_token": "tok-e2e"},
+	}
+	if _, err := first.tokenStore.Save(ctx, auth); err != nil {
+		t.Fatal(err)
+	}
+
+	second := openReplica()
+	if second.authDir == first.authDir {
+		t.Fatal("replicas share a spool")
+	}
+	auths, err := second.tokenStore.List(ctx)
+	if err != nil || len(auths) != 1 || auths[0].ID != "team/user@example.com.json" || auths[0].Metadata["access_token"] != "tok-e2e" {
+		t.Fatalf("second replica List = %v, %v", auths, err)
+	}
+	if _, err := os.Stat(filepath.Join(second.authDir, "team", "user@example.com.json")); err != nil {
+		t.Fatalf("credential was not mirrored into the second spool: %v", err)
+	}
+
+	// An edit that reaches the first spool, as the management API would make it, goes back to the buckets.
+	edited := filepath.Join(first.authDir, "team", "user@example.com.json")
+	if err := os.WriteFile(edited, []byte(`{"type":"test","email":"user@example.com","access_token":"tok-e2e","note":"edited"}`), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if err := first.tokenStore.(storePersister).PersistAuthFiles(ctx, "", edited); err != nil {
+		t.Fatal(err)
+	}
+	third := openReplica()
+	got, err := os.ReadFile(filepath.Join(third.authDir, "team", "user@example.com.json"))
+	if err != nil || !strings.Contains(string(got), `"note":"edited"`) {
+		t.Fatalf("third replica spool = %s, %v", got, err)
+	}
+}