refactor: 分离公共基础设施与 Database 适配器

internal/database/adapter/kubernetes/credentials.go

@@ -22,10 +22,8 @@ import (
 	"errors"
 
 	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/validation"
-	typedcore "k8s.io/client-go/kubernetes/typed/core/v1"
-	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"git.ddupan.top/panxiao81/ayatori/internal/database/application"
 	"git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance"
@@ -34,18 +32,16 @@ import (
 // SecretCredentials 直接读取 API server，不将 Secret 数据纳入共享 informer cache。
 // namespace 在装配时固定，Instance 不能选择跨 namespace 读取。
 type SecretCredentials struct {
-	secrets typedcore.SecretInterface
+	reader    client.Reader
+	namespace string
 }
 
-func NewSecretCredentials(config *rest.Config, namespace string) (*SecretCredentials, error) {
-	if config == nil || len(validation.IsDNS1123Label(namespace)) != 0 {
-		return nil, errors.New("valid controller namespace and API configuration required")
+// NewSecretCredentials 要求注入 manager.GetAPIReader() 或等价直连 reader，不可使用缓存 reader。
+func NewSecretCredentials(reader client.Reader, namespace string) (*SecretCredentials, error) {
+	if reader == nil || len(validation.IsDNS1123Label(namespace)) != 0 {
+		return nil, errors.New("valid controller namespace and API reader required")
 	}
-	client, err := typedcore.NewForConfig(config)
-	if err != nil {
-		return nil, application.ErrCredentialsUnavailable
-	}
-	return &SecretCredentials{secrets: client.Secrets(namespace)}, nil
+	return &SecretCredentials{reader: reader, namespace: namespace}, nil
 }
 
 func (r *SecretCredentials) Read(ctx context.Context, ref instance.CredentialReference) (application.Credentials, error) {
@@ -53,7 +49,8 @@ func (r *SecretCredentials) Read(ctx context.Context, ref instance.CredentialRef
 		return application.Credentials{}, application.ErrCredentialsInvalid
 	}
 	keys := ref.Values()
-	secret, err := r.secrets.Get(ctx, keys.Name, metav1.GetOptions{})
+	secret := &corev1.Secret{}
+	err := r.reader.Get(ctx, client.ObjectKey{Namespace: r.namespace, Name: keys.Name}, secret)
 	if err != nil {
 		return application.Credentials{}, application.ErrCredentialsUnavailable
 	}

internal/database/adapter/openbao/authentication.go

@@ -1,180 +0,0 @@
-/*
-Copyright 2026.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package openbao
-
-import (
-	"context"
-	"errors"
-	"strings"
-	"sync"
-	"sync/atomic"
-	"time"
-
-	kubernetesauth "github.com/openbao/openbao/api/auth/kubernetes/v2"
-	bao "github.com/openbao/openbao/api/v2"
-	authenticationv1 "k8s.io/api/authentication/v1"
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/util/validation"
-	kubeclient "sigs.k8s.io/controller-runtime/pkg/client"
-)
-
-var ErrAuthenticationConfiguration = errors.New("invalid OpenBao Kubernetes authentication configuration")
-
-// KubernetesSession 为专用 SDK client 维护短期登录，不持久化或对外返回 token。
-// 每次登录通过 manager 的 Kubernetes 身份申请新的 SA JWT，不依赖 controller 的部署位置。
-// 续期调度由官方 LifetimeWatcher 负责，不实现自己的 lease 算法。
-type KubernetesSession struct {
-	client     *bao.Client
-	mount      string
-	role       string
-	kubernetes kubeclient.Client
-	identity   KubernetesIdentity
-	running    sync.Mutex
-	ready      atomic.Bool
-}
-
-// KubernetesIdentity 是部署固定的登录目标，不由 Tenant 选择。
-type KubernetesIdentity struct {
-	Namespace      string
-	ServiceAccount string
-	Audience       string
-}
-
-func NewKubernetesSession(
-	client *bao.Client,
-	kubernetes kubeclient.Client,
-	mount, role string,
-	identity KubernetesIdentity,
-) (*KubernetesSession, error) {
-	if client == nil || kubernetes == nil || !validPath(mount) || !pathSegment.MatchString(role) ||
-		len(validation.IsDNS1123Label(identity.Namespace)) != 0 ||
-		len(validation.IsDNS1123Subdomain(identity.ServiceAccount)) != 0 || strings.TrimSpace(identity.Audience) == "" {
-		return nil, ErrAuthenticationConfiguration
-	}
-	client.ClearToken()
-	client.SetMaxRetries(0)
-	client.SetClientTimeout(15 * time.Second)
-	return &KubernetesSession{
-		client:     client,
-		mount:      mount,
-		role:       role,
-		kubernetes: kubernetes,
-		identity:   identity,
-	}, nil
-}
-
-// Ready 仅代表当前登录 lease 正由 SDK 管理，不保证下一次后端请求一定成功。
-func (s *KubernetesSession) Ready() bool { return s.ready.Load() }
-
-// Start 可交给 manager 管理；关闭时清空本地 token，不撤销共享后端数据。
-// SDK 的 Stop 不取消已发出的续期 HTTP 请求，因此等待该请求结束后才退出，最长受 client timeout 限制。
-func (s *KubernetesSession) Start(ctx context.Context) error {
-	if !s.running.TryLock() {
-		return errors.New("OpenBao authentication is already running")
-	}
-	defer s.running.Unlock()
-	defer s.clear()
-	for ctx.Err() == nil {
-		s.clear()
-		secret := s.login(ctx)
-		if secret != nil {
-			s.watch(ctx, secret)
-		}
-		s.clear()
-		// 登录失败及不能继续续期均有限速，防止依赖故障时形成请求忙循环。
-		select {
-		case <-ctx.Done():
-			return nil
-		case <-time.After(5 * time.Second):
-		}
-	}
-	return nil
-}
-
-func (s *KubernetesSession) clear() {
-	s.ready.Store(false)
-	s.client.ClearToken()
-}
-
-func (s *KubernetesSession) login(ctx context.Context) *bao.Secret {
-	ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
-	defer cancel()
-	// JWT 只用于本次登录，不缓存或自行解析 kubeconfig 中的凭据。
-	// client-go 负责 kubeconfig/in-cluster 身份与凭据更新；API server 按 RBAC 签发。
-	expirationSeconds := int64(600)
-	account := &corev1.ServiceAccount{
-		Namespace: s.identity.Namespace,
-		Name:      s.identity.ServiceAccount,
-	}
-	token := &authenticationv1.TokenRequest{
-		Spec: authenticationv1.TokenRequestSpec{
-			Audiences:         []string{s.identity.Audience},
-			ExpirationSeconds: &expirationSeconds,
-		},
-	}
-	// 子资源写入直接请求 API server，不读 cache，也不需要额外的 ServiceAccount get 权限。
-	err := s.kubernetes.SubResource("token").Create(ctx, account, token)
-	if err != nil || strings.TrimSpace(token.Status.Token) == "" ||
-		!token.Status.ExpirationTimestamp.After(time.Now()) {
-		return nil
-	}
-	// helper 会缓存 token，不能跨登录轮次复用。
-	method, err := kubernetesauth.NewKubernetesAuth(s.role,
-		kubernetesauth.WithMountPath(s.mount),
-		kubernetesauth.WithServiceAccountToken(token.Status.Token),
-	)
-	if err != nil {
-		return nil
-	}
-	// 先检查短期 lease，再发布到共享 client，避免暴露不合规的登录结果。
-	secret, err := method.Login(ctx, s.client)
-	if err != nil || secret == nil || secret.Auth == nil || secret.Auth.ClientToken == "" || secret.Auth.LeaseDuration <= 0 {
-		return nil
-	}
-	if ctx.Err() != nil {
-		return nil
-	}
-	s.client.SetToken(secret.Auth.ClientToken)
-	return secret
-}
-
-func (s *KubernetesSession) watch(ctx context.Context, secret *bao.Secret) {
-	behavior := bao.RenewBehaviorErrorOnErrors
-	if !secret.Auth.Renewable {
-		behavior = bao.RenewBehaviorRenewDisabled
-	}
-	watcher, err := s.client.NewLifetimeWatcher(&bao.LifetimeWatcherInput{Secret: secret, RenewBehavior: behavior})
-	if err != nil {
-		return
-	}
-	s.ready.Store(true)
-	go watcher.Start()
-	for {
-		select {
-		case <-ctx.Done():
-			s.clear()
-			watcher.Stop()
-			<-watcher.DoneCh()
-			return
-		case <-watcher.DoneCh():
-			watcher.Stop()
-			return
-		case <-watcher.RenewCh():
-			// 不记录 SDK Secret 或 token；无需复制 SDK 已处理的续期数据。
-		}
-	}
-}

internal/database/adapter/openbao/authentication_integration_test.go

@@ -51,7 +51,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 
-	"git.ddupan.top/panxiao81/ayatori/internal/database/adapter/openbao"
+	"git.ddupan.top/panxiao81/ayatori/internal/infra/openbao"
 )
 
 const (
@@ -232,6 +232,40 @@ func newKubernetesAuthFixture(t *testing.T) *kubernetesAuthFixture {
 	}
 }
 
+const authRole = "controller"
+
+var testIdentity = openbao.KubernetesIdentity{Namespace: authNamespace, ServiceAccount: authRole, Audience: authAudience}
+
+func waitForAuthentication(t *testing.T, check func() bool) {
+	t.Helper()
+	deadline := time.NewTimer(20 * time.Second)
+	defer deadline.Stop()
+	for !check() {
+		select {
+		case <-deadline.C:
+			t.Fatal("authentication condition timed out")
+		case <-time.After(20 * time.Millisecond):
+		}
+	}
+}
+
+func startSession(t *testing.T, session interface{ Start(context.Context) error }) context.CancelFunc {
+	t.Helper()
+	ctx, cancel := context.WithCancel(t.Context())
+	done := make(chan error, 1)
+	go func() { done <- session.Start(ctx) }()
+	t.Cleanup(func() {
+		cancel()
+		select {
+		case <-done:
+		case <-time.After(20 * time.Second):
+			t.Error("authentication did not stop")
+		}
+	})
+	return cancel
+}
+
+// 此处验证公共认证会话与 Database 凭据适配器的跨层集成。
 func TestKubernetesSessionWithRealTokenReview(t *testing.T) {
 	api := newKubernetesAuthFixture(t)
 	ctx := t.Context()

internal/database/adapter/openbao/authentication_test.go

@@ -1,219 +0,0 @@
-/*
-Copyright 2026.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package openbao_test
-
-import (
-	"context"
-	"encoding/json"
-	"net/http"
-	"net/http/httptest"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	authenticationv1 "k8s.io/api/authentication/v1"
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/meta"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime/schema"
-	"k8s.io/client-go/rest"
-	kubeclient "sigs.k8s.io/controller-runtime/pkg/client"
-
-	"git.ddupan.top/panxiao81/ayatori/internal/database/adapter/openbao"
-)
-
-const (
-	authRole = "controller"
-)
-
-var testIdentity = openbao.KubernetesIdentity{Namespace: "bao-controller", ServiceAccount: authRole, Audience: "openbao"}
-
-func authenticationClient(t *testing.T, address string) kubeclient.Client {
-	t.Helper()
-	// HTTP 单元 fixture 只提供 TokenRequest；静态映射避免额外模拟 discovery API。
-	mapper := meta.NewDefaultRESTMapper([]schema.GroupVersion{corev1.SchemeGroupVersion})
-	mapper.Add(corev1.SchemeGroupVersion.WithKind("ServiceAccount"), meta.RESTScopeNamespace)
-	client, err := kubeclient.New(&rest.Config{Host: address, ContentType: "application/json"}, kubeclient.Options{
-		Mapper: mapper,
-	})
-	if err != nil {
-		t.Fatal("cannot construct fixture Kubernetes client")
-	}
-	return client
-}
-
-func authenticationServer(t *testing.T, token func() (string, bool), login http.HandlerFunc) *httptest.Server {
-	t.Helper()
-	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/api/v1/namespaces/bao-controller/serviceaccounts/controller/token" {
-			login(w, r)
-			return
-		}
-		var request authenticationv1.TokenRequest
-		if json.NewDecoder(r.Body).Decode(&request) != nil || r.Method != http.MethodPost ||
-			len(request.Spec.Audiences) != 1 || request.Spec.Audiences[0] != testIdentity.Audience ||
-			request.Spec.ExpirationSeconds == nil || *request.Spec.ExpirationSeconds != 600 {
-			t.Error("unexpected TokenRequest target or lifetime")
-			w.WriteHeader(http.StatusBadRequest)
-			return
-		}
-		jwt, allowed := token()
-		if !allowed {
-			w.WriteHeader(http.StatusForbidden)
-			return
-		}
-		w.Header().Set("Content-Type", "application/json")
-		if err := json.NewEncoder(w).Encode(authenticationv1.TokenRequest{
-			APIVersion: "authentication.k8s.io/v1", Kind: "TokenRequest",
-			Status: authenticationv1.TokenRequestStatus{Token: jwt, ExpirationTimestamp: metav1.NewTime(time.Now().Add(10 * time.Minute))},
-		}); err != nil {
-			t.Error("cannot encode fixture TokenRequest response")
-		}
-	}))
-}
-
-func TestKubernetesSessionDoesNotFallbackFromMissingToken(t *testing.T) {
-	var requests atomic.Int32
-	for _, missing := range []bool{true, false} {
-		server := authenticationServer(t, func() (string, bool) { return "", !missing }, func(w http.ResponseWriter, _ *http.Request) {
-			requests.Add(1)
-			w.WriteHeader(http.StatusForbidden)
-		})
-		defer server.Close()
-		client := fixtureClient(t, server.URL)
-		session, err := openbao.NewKubernetesSession(client, authenticationClient(t, server.URL), "kubernetes", authRole, testIdentity)
-		if err != nil {
-			t.Fatal(err)
-		}
-		ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
-		err = session.Start(ctx)
-		cancel()
-		if err != nil || session.Ready() || client.Token() != "" || requests.Load() != 0 {
-			t.Fatal("denied or empty TokenRequest must not fall back to another identity")
-		}
-	}
-}
-
-func waitForAuthentication(t *testing.T, check func() bool) {
-	t.Helper()
-	deadline := time.NewTimer(20 * time.Second)
-	defer deadline.Stop()
-	for !check() {
-		select {
-		case <-deadline.C:
-			t.Fatal("authentication condition timed out")
-		case <-time.After(20 * time.Millisecond):
-		}
-	}
-}
-
-func startSession(t *testing.T, session interface{ Start(context.Context) error }) context.CancelFunc {
-	t.Helper()
-	ctx, cancel := context.WithCancel(t.Context())
-	done := make(chan error, 1)
-	go func() { done <- session.Start(ctx) }()
-	t.Cleanup(func() {
-		cancel()
-		select {
-		case <-done:
-		case <-time.After(20 * time.Second):
-			t.Error("authentication did not stop")
-		}
-	})
-	return cancel
-}
-
-func TestKubernetesSessionRequestsNewTokenAfterFailure(t *testing.T) {
-	var attempts atomic.Int32
-	var accepted atomic.Bool
-	var rotated atomic.Bool
-	server := authenticationServer(t, func() (string, bool) {
-		if rotated.Load() {
-			return "rotated-test-jwt", true
-		}
-		return "expired-test-jwt", true
-	}, func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/v1/auth/kubernetes/login" {
-			w.WriteHeader(http.StatusNotFound)
-			return
-		}
-		var request struct{ JWT, Role string }
-		if json.NewDecoder(r.Body).Decode(&request) != nil || request.Role != authRole {
-			t.Error("unexpected authentication request")
-			w.WriteHeader(http.StatusBadRequest)
-			return
-		}
-		attempts.Add(1)
-		if request.JWT != "rotated-test-jwt" {
-			w.WriteHeader(http.StatusForbidden)
-			return
-		}
-		accepted.Store(true)
-		if err := json.NewEncoder(w).Encode(map[string]any{"auth": map[string]any{
-			"client_token": fixtureToken, "lease_duration": 60, "renewable": false,
-		}}); err != nil {
-			t.Error("cannot encode authentication fixture response")
-		}
-	})
-	defer server.Close()
-	client := fixtureClient(t, server.URL)
-	session, err := openbao.NewKubernetesSession(client, authenticationClient(t, server.URL), "kubernetes", authRole, testIdentity)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if client.Token() != "" {
-		t.Fatal("constructor retained preexisting static token")
-	}
-	cancel := startSession(t, session)
-	waitForAuthentication(t, func() bool { return attempts.Load() > 0 })
-	if session.Ready() || client.Token() != "" {
-		t.Fatal("failed login retained credentials")
-	}
-	rotated.Store(true)
-	waitForAuthentication(t, func() bool { return accepted.Load() && session.Ready() })
-	if client.Token() != fixtureToken {
-		t.Fatal("successful login did not configure the client")
-	}
-	if session.Start(t.Context()) == nil {
-		t.Fatal("allowed concurrent lifecycle owners")
-	}
-	cancel()
-	waitForAuthentication(t, func() bool { return !session.Ready() && client.Token() == "" })
-}
-
-func TestKubernetesSessionRejectsUnboundedLease(t *testing.T) {
-	var attempts atomic.Int32
-	server := authenticationServer(t, func() (string, bool) { return "test-jwt", true }, func(w http.ResponseWriter, _ *http.Request) {
-		attempts.Add(1)
-		if err := json.NewEncoder(w).Encode(map[string]any{"auth": map[string]any{
-			"client_token": fixtureToken, "lease_duration": 0,
-		}}); err != nil {
-			t.Error("cannot encode authentication fixture response")
-		}
-	})
-	defer server.Close()
-	client := fixtureClient(t, server.URL)
-	session, err := openbao.NewKubernetesSession(client, authenticationClient(t, server.URL), "kubernetes", authRole, testIdentity)
-	if err != nil {
-		t.Fatal(err)
-	}
-	startSession(t, session)
-	waitForAuthentication(t, func() bool { return attempts.Load() > 0 })
-	if session.Ready() || client.Token() != "" {
-		t.Fatal("accepted a token without a finite lease")
-	}
-}

internal/database/adapter/openbao/credentials.go

@@ -49,12 +49,11 @@ type Credentials struct {
 }
 
 // NewCredentials 不登录、不读取环境 token。调用方必须提供专用的已认证 client。
-// 禁用 SDK 写入重试，防止第一次结果丢失后被 CAS 错误掩盖。
+// client 由公共 infra 禁用自动重试，防止第一次结果丢失后被 CAS 错误掩盖。
 func NewCredentials(client *bao.Client, mount, basePath string) (*Credentials, error) {
 	if client == nil || !validPath(mount) || !validPath(basePath) {
 		return nil, ErrInvalidLocation
 	}
-	client.SetMaxRetries(0)
 	return &Credentials{kv: client.KVv2(mount), basePath: basePath}, nil
 }
 

internal/database/adapter/openbao/credentials_test.go

@@ -104,6 +104,7 @@ func fixtureClient(t *testing.T, address string) *bao.Client {
 	t.Helper()
 	config := bao.DefaultConfig()
 	config.Address = address
+	config.MaxRetries = 0
 	client, err := bao.NewClient(config)
 	if err != nil {
 		t.Fatal("cannot construct fixture client")

internal/database/adapter/postgresql/fixture_integration_test.go

@@ -32,6 +32,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+	kubeclient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 
 	secretadapter "git.ddupan.top/panxiao81/ayatori/internal/database/adapter/kubernetes"
@@ -244,7 +245,11 @@ func newCredentialFixture(t *testing.T) *credentialFixture {
 		}
 	}
 
-	reader, err := secretadapter.NewSecretCredentials(config, controllerNamespace)
+	apiReader, err := kubeclient.New(config, kubeclient.Options{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	reader, err := secretadapter.NewSecretCredentials(apiReader, controllerNamespace)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -252,7 +257,11 @@ func newCredentialFixture(t *testing.T) *credentialFixture {
 	if err != nil {
 		t.Fatal(err)
 	}
-	deniedReader, err := secretadapter.NewSecretCredentials(user.Config(), controllerNamespace)
+	deniedAPIReader, err := kubeclient.New(user.Config(), kubeclient.Options{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	deniedReader, err := secretadapter.NewSecretCredentials(deniedAPIReader, controllerNamespace)
 	if err != nil {
 		t.Fatal(err)
 	}

internal/database/adapter/postgresql/instance_controller_integration_test.go

@@ -50,14 +50,6 @@ func TestInstanceControllerWithRealPostgreSQL(t *testing.T) {
 		t.Fatal(err)
 	}
 	restricted := instanceControllerRBAC(t, f, apiClient)
-	credentials, err := secretadapter.NewSecretCredentials(restricted, controllerNamespace)
-	if err != nil {
-		t.Fatal(err)
-	}
-	service, err := application.NewInstanceService(credentials, postgresql.Connector{})
-	if err != nil {
-		t.Fatal(err)
-	}
 	// 同进程 -count 重复启动测试 manager；生产继续校验 controller 名称唯一。
 	skipRepeatedName := true
 	manager, err := ctrl.NewManager(restricted, ctrl.Options{
@@ -68,6 +60,14 @@ func TestInstanceControllerWithRealPostgreSQL(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	credentials, err := secretadapter.NewSecretCredentials(manager.GetAPIReader(), controllerNamespace)
+	if err != nil {
+		t.Fatal(err)
+	}
+	service, err := application.NewInstanceService(credentials, postgresql.Connector{})
+	if err != nil {
+		t.Fatal(err)
+	}
 	reconciler := &databasecontroller.InstanceReconciler{Observer: service, SecretNamespace: controllerNamespace}
 	if err := reconciler.SetupWithManager(manager); err != nil {
 		t.Fatal(err)