diff --git a/cmd/server/backend.go b/cmd/server/backend.go index 52ffe14a15b..ee12e853eec 100644 --- a/cmd/server/backend.go +++ b/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) diff --git a/cmd/server/backend_nats_test.go b/cmd/server/backend_nats_test.go new file mode 100644 index 00000000000..793b7488739 --- /dev/null +++ b/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) + } +}