feat: 接入管理 Secret 凭据与连接刷新

internal/database/adapter/kubernetes/credentials.go

@@ -0,0 +1,68 @@
+/*
+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 kubernetes 提供 Database 所需的 Kubernetes API 薄适配。
+package kubernetes
+
+import (
+	"context"
+	"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"
+
+	"git.ddupan.top/panxiao81/ayatori/internal/database/application"
+	"git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance"
+)
+
+// SecretCredentials 直接读取 API server，不将 Secret 数据纳入共享 informer cache。
+// namespace 在装配时固定，Instance 不能选择跨 namespace 读取。
+type SecretCredentials struct {
+	secrets typedcore.SecretInterface
+}
+
+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")
+	}
+	client, err := typedcore.NewForConfig(config)
+	if err != nil {
+		return nil, application.ErrCredentialsUnavailable
+	}
+	return &SecretCredentials{secrets: client.Secrets(namespace)}, nil
+}
+
+func (r *SecretCredentials) Read(ctx context.Context, ref instance.CredentialReference) (application.Credentials, error) {
+	if err := ref.Validate(); err != nil {
+		return application.Credentials{}, application.ErrCredentialsInvalid
+	}
+	keys := ref.Values()
+	secret, err := r.secrets.Get(ctx, keys.Name, metav1.GetOptions{})
+	if err != nil {
+		return application.Credentials{}, application.ErrCredentialsUnavailable
+	}
+	return decode(secret, keys)
+}
+
+func decode(secret *corev1.Secret, keys instance.CredentialReferenceValues) (application.Credentials, error) {
+	if secret.DeletionTimestamp != nil {
+		return application.Credentials{}, application.ErrCredentialsUnavailable
+	}
+	return application.NewCredentials(string(secret.Data[keys.UsernameKey]), string(secret.Data[keys.PasswordKey]))
+}

internal/database/adapter/postgresql/connector.go

@@ -0,0 +1,116 @@
+/*
+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 postgresql 使用 pgxpool 提供 PostgreSQL 能力的薄适配。
+package postgresql
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"net"
+	"net/url"
+	"strconv"
+
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"git.ddupan.top/panxiao81/ayatori/internal/database/application"
+	"git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance"
+)
+
+// Connector 不读取 Secret、不决定连接何时替换；池本身由 pgxpool 实现。
+type Connector struct {
+	RootCert string
+}
+
+type database struct {
+	pool *pgxpool.Pool
+}
+
+func (*database) String() string     { return "[redacted PostgreSQL database]" }
+func (d *database) GoString() string { return d.String() }
+func (d *database) Close() {
+	d.pool.Close()
+}
+
+func (d *database) Version(ctx context.Context) (string, error) {
+	var version string
+	if err := d.pool.QueryRow(ctx, "SHOW server_version").Scan(&version); err != nil {
+		return "", safeError(err, application.ErrObservation)
+	}
+	return version, nil
+}
+
+func (c Connector) Connect(ctx context.Context, endpoint instance.Endpoint, credentials application.Credentials) (application.Database, error) {
+	if err := endpoint.Validate(); err != nil {
+		return nil, err
+	}
+	if credentials.Username() == "" || credentials.Password() == "" {
+		return nil, application.ErrCredentialsInvalid
+	}
+	endpointValues := endpoint.Values()
+	query := url.Values{
+		"sslmode":          {string(endpointValues.TLSMode)},
+		"connect_timeout":  {"5"},
+		"application_name": {"ayatori-database-management"},
+	}
+	if c.RootCert != "" {
+		query.Set("sslrootcert", c.RootCert)
+	}
+	connectionURL := url.URL{
+		Scheme:   "postgresql",
+		Host:     net.JoinHostPort(endpointValues.Host, strconv.Itoa(endpointValues.Port)),
+		Path:     "/" + endpointValues.ManagementDatabase,
+		User:     url.UserPassword(credentials.Username(), credentials.Password()),
+		RawQuery: query.Encode(),
+	}
+	config, err := pgxpool.ParseConfig(connectionURL.String())
+	if err != nil {
+		return nil, application.ErrConnection
+	}
+	// pgx 不实现 libpq hostaddr；复用其 LookupFunc 扩展点，TLS 验证身份仍采用 host。
+	config.ConnConfig.LookupFunc = func(context.Context, string) ([]string, error) {
+		return []string{endpointValues.HostAddr}, nil
+	}
+	config.ConnConfig.Fallbacks = nil
+	pool, err := pgxpool.NewWithConfig(ctx, config)
+	if err != nil {
+		return nil, safeError(err, application.ErrConnection)
+	}
+	if err := pool.Ping(ctx); err != nil {
+		pool.Close()
+		return nil, safeError(err, application.ErrConnection)
+	}
+	return &database{pool: pool}, nil
+}
+
+func safeError(err, fallback error) error {
+	if errors.Is(err, context.Canceled) {
+		return context.Canceled
+	}
+	if errors.Is(err, context.DeadlineExceeded) {
+		return context.DeadlineExceeded
+	}
+	var pgerr *pgconn.PgError
+	if errors.As(err, &pgerr) && (pgerr.Code == "28P01" || pgerr.Code == "28000") {
+		return application.ErrAuthentication
+	}
+	if _, ok := errors.AsType[*tls.CertificateVerificationError](err); ok {
+		return application.ErrAuthentication
+	}
+	return fallback
+}

internal/database/adapter/postgresql/credentials_integration_test.go

@@ -0,0 +1,227 @@
+//go:build integration
+
+/*
+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 postgresql_test
+
+import (
+	"context"
+	"errors"
+	"os/exec"
+	"sync"
+	"testing"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"git.ddupan.top/panxiao81/ayatori/internal/database/adapter/postgresql"
+	"git.ddupan.top/panxiao81/ayatori/internal/database/application"
+	"git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance"
+)
+
+func TestManagementSecretScopeAndMissingDependencyRecovery(t *testing.T) {
+	fixture := newCredentialFixture(t)
+	reference := fixture.target.Definition().AdminCredential()
+
+	fixture.createSecret(t, "unrelated")
+	if _, err := fixture.reader.Read(fixture.ctx, reference); !errors.Is(err, application.ErrCredentialsUnavailable) {
+		t.Fatal("a Secret in another namespace satisfied the reference")
+	}
+	if _, err := fixture.service.ObserveVersion(fixture.ctx, fixture.target); !errors.Is(err, application.ErrCredentialsUnavailable) {
+		t.Fatal("missing Secret did not fail closed")
+	}
+
+	fixture.createSecret(t, controllerNamespace)
+	if _, err := fixture.deniedReader.Read(fixture.ctx, reference); !errors.Is(err, application.ErrCredentialsUnavailable) {
+		t.Fatal("API server did not enforce Secret RBAC")
+	}
+	fixture.observeVersion(t)
+
+	fixture.updateSecret(t, func(secret *corev1.Secret) {
+		delete(secret.Data, "credential")
+	})
+	if _, err := fixture.service.ObserveVersion(fixture.ctx, fixture.target); !errors.Is(err, application.ErrCredentialsInvalid) {
+		t.Fatal("missing credential field reused a cached connection")
+	}
+	fixture.updateSecret(t, func(secret *corev1.Secret) {
+		secret.Data["credential"] = []byte(fixturePassword)
+	})
+	fixture.observeVersion(t)
+
+	err := fixture.client.CoreV1().Secrets(controllerNamespace).Delete(fixture.ctx, secretName, metav1.DeleteOptions{})
+	if err != nil {
+		t.Fatal("cannot delete fixture Secret")
+	}
+	if _, err := fixture.service.ObserveVersion(fixture.ctx, fixture.target); !errors.Is(err, application.ErrCredentialsUnavailable) {
+		t.Fatal("deleted Secret retained access")
+	}
+}
+
+func TestEffectiveCredentialChangesReplaceConnection(t *testing.T) {
+	fixture := newCredentialFixture(t)
+	fixture.createSecret(t, controllerNamespace)
+	fixture.observeVersion(t)
+	originalBackend := fixture.backendIDs(t)
+	if originalBackend == "" {
+		t.Fatal("management connection not visible in PostgreSQL")
+	}
+
+	fixture.updateSecret(t, func(secret *corev1.Secret) {
+		secret.Labels = map[string]string{"changed": "true"}
+		secret.Data["unrelated"] = []byte("ignored")
+	})
+	fixture.observeVersion(t)
+	if fixture.backendIDs(t) != originalBackend {
+		t.Fatal("metadata or unrelated fields rebuilt the connection")
+	}
+
+	// 先改变 Secret、暂不改变服务器密码：旧连接必须失效，新认证必须失败。
+	fixture.updateSecret(t, func(secret *corev1.Secret) {
+		secret.Data["credential"] = []byte(rotatedPassword)
+	})
+	version, err := fixture.service.ObserveVersion(fixture.ctx, fixture.target)
+	if !errors.Is(err, application.ErrAuthentication) || version != "" {
+		t.Fatal("old connection bypassed changed credentials")
+	}
+
+	fixture.queryPostgres(t, "ALTER ROLE postgres PASSWORD '"+rotatedPassword+"'")
+	fixture.observeVersion(t)
+	if fixture.backendIDs(t) == originalBackend {
+		t.Fatal("password rotation reused the old backend")
+	}
+
+	fixture.updateSecret(t, func(secret *corev1.Secret) {
+		secret.Data["login"] = []byte("nonexistent")
+	})
+	if _, err := fixture.service.ObserveVersion(fixture.ctx, fixture.target); !errors.Is(err, application.ErrAuthentication) {
+		t.Fatal("username change did not require a new authentication")
+	}
+	fixture.updateSecret(t, func(secret *corev1.Secret) {
+		secret.Data["login"] = []byte(fixtureUser)
+	})
+	fixture.observeVersion(t)
+}
+
+func TestObservationDiscardsResultWhenCredentialsChange(t *testing.T) {
+	fixture := newCredentialFixture(t)
+	fixture.createSecret(t, controllerNamespace)
+
+	reads := 0
+	fixture.gate.beforeRead = func() {
+		reads++
+		if reads == 2 {
+			fixture.updateSecret(t, func(secret *corev1.Secret) {
+				secret.Data["credential"] = []byte(rotatedPassword)
+			})
+		}
+	}
+
+	version, err := fixture.service.ObserveVersion(fixture.ctx, fixture.target)
+	if !errors.Is(err, application.ErrCredentialsChanged) {
+		t.Fatal("in-flight rotation was not detected")
+	}
+	if version != "" {
+		t.Fatal("observation returned data obtained with stale credentials")
+	}
+	if fixture.backendIDs(t) != "" {
+		t.Fatal("stale connection was retained after rotation")
+	}
+}
+
+func TestConnectionReleaseAndServiceRestart(t *testing.T) {
+	fixture := newCredentialFixture(t)
+	fixture.createSecret(t, controllerNamespace)
+	fixture.observeVersion(t)
+
+	fixture.service.Forget(fixture.target.Identity().Name())
+	if fixture.backendIDs(t) != "" {
+		t.Fatal("Forget retained a connection")
+	}
+	fixture.observeVersion(t)
+
+	fixture.service.Close()
+	fixture.service.Close()
+	if _, err := fixture.service.ObserveVersion(fixture.ctx, fixture.target); !errors.Is(err, application.ErrClosed) {
+		t.Fatal("closed service accepted work")
+	}
+
+	restarted, err := application.NewInstanceService(fixture.reader, postgresql.Connector{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(restarted.Close)
+	if _, err := restarted.ObserveVersion(fixture.ctx, fixture.target); err != nil {
+		t.Fatal("new service could not recover from stored Secret", err)
+	}
+
+	// 此 fixture 未启用 TLS；各加密模式均不得偷偷回退到明文连接。
+	for _, mode := range []instance.TLSMode{instance.TLSRequire, instance.TLSVerifyCA, instance.TLSVerifyFull} {
+		securedTarget := target(t, fixture.port, mode)
+		if _, err := restarted.ObserveVersion(fixture.ctx, securedTarget); err == nil {
+			t.Fatal("TLS policy silently downgraded to plaintext")
+		}
+	}
+}
+
+func TestConcurrentVersionObservations(t *testing.T) {
+	fixture := newCredentialFixture(t)
+	fixture.createSecret(t, controllerNamespace)
+
+	var workers sync.WaitGroup
+	for range 4 {
+		workers.Go(func() {
+			version, err := fixture.service.ObserveVersion(fixture.ctx, fixture.target)
+			if err != nil || version == "" {
+				t.Error("concurrent observation failed", err)
+			}
+		})
+	}
+	workers.Go(func() {
+		fixture.service.Forget(fixture.target.Identity().Name())
+	})
+	workers.Wait()
+	fixture.observeVersion(t)
+}
+
+func TestManagementConnectionRecoversAfterTimeout(t *testing.T) {
+	fixture := newCredentialFixture(t)
+	fixture.createSecret(t, controllerNamespace)
+	fixture.observeVersion(t)
+
+	if err := exec.CommandContext(fixture.ctx, "docker", "pause", fixture.containerID).Run(); err != nil {
+		t.Fatal("cannot pause isolated PostgreSQL fixture")
+	}
+	// 即使断言失败，也先恢复容器，再由 fixture 按原 ID 清理。
+	t.Cleanup(func() {
+		cleanupContext, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		_ = exec.CommandContext(cleanupContext, "docker", "unpause", fixture.containerID).Run()
+	})
+
+	queryContext, cancel := context.WithTimeout(fixture.ctx, 500*time.Millisecond)
+	version, err := fixture.service.ObserveVersion(queryContext, fixture.target)
+	cancel()
+	if err == nil || version != "" {
+		t.Fatal("timed out PostgreSQL observation returned a successful result")
+	}
+
+	if err := exec.CommandContext(fixture.ctx, "docker", "unpause", fixture.containerID).Run(); err != nil {
+		t.Fatal("cannot resume isolated PostgreSQL fixture")
+	}
+	fixture.observeVersion(t)
+}

internal/database/adapter/postgresql/fixture_integration_test.go

@@ -0,0 +1,278 @@
+//go:build integration
+
+/*
+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 postgresql_test
+
+import (
+	"context"
+	"os/exec"
+	"regexp"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+
+	secretadapter "git.ddupan.top/panxiao81/ayatori/internal/database/adapter/kubernetes"
+	"git.ddupan.top/panxiao81/ayatori/internal/database/adapter/postgresql"
+	"git.ddupan.top/panxiao81/ayatori/internal/database/application"
+	"git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance"
+)
+
+const (
+	fixtureHost         = "fixture.invalid"
+	fixtureUser         = "postgres"
+	dockerExec          = "exec"
+	fixtureImage        = "postgres@sha256:18cfe3ef5e6815560c98237d6216d1e5119702fb0f3894c8785dd58b8bbe5d73"
+	fixturePassword     = "AYATORI-TEST-ONLY-initial-password"
+	rotatedPassword     = "AYATORI-TEST-ONLY-rotated-password"
+	controllerNamespace = "database-controller"
+	secretName          = "management"
+)
+
+// fixture 不接受外部 DSN，只创建自己的临时容器并按确切 ID 清理。
+func postgresFixture(t *testing.T, ctx context.Context) (string, int) {
+	t.Helper()
+	output, err := exec.CommandContext(ctx, "docker", "run", "--rm", "-d", "-p", "127.0.0.1::5432",
+		"-e", "POSTGRES_PASSWORD="+fixturePassword, fixtureImage).Output()
+	if err != nil {
+		t.Fatal("cannot start isolated PostgreSQL fixture")
+	}
+	id := strings.TrimSpace(string(output))
+	if !regexp.MustCompile(`^[a-f0-9]{64}$`).MatchString(id) {
+		t.Fatal("unexpected container identifier")
+	}
+	t.Cleanup(func() {
+		cleanup, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		if exec.CommandContext(cleanup, "docker", "rm", "-f", id).Run() != nil {
+			t.Error("fixture cleanup failed")
+		}
+	})
+	output, err = exec.CommandContext(ctx, "docker", "inspect", "--format", `{{(index (index .NetworkSettings.Ports "5432/tcp") 0).HostPort}}`, id).Output()
+	if err != nil {
+		t.Fatal("cannot inspect fixture port")
+	}
+	port, err := strconv.Atoi(strings.TrimSpace(string(output)))
+	if err != nil {
+		t.Fatal("invalid fixture port")
+	}
+	// 初次 init 的临时服务器只监听 Unix socket，必须等最终 TCP listener。
+	for exec.CommandContext(ctx, "docker", dockerExec, id, "pg_isready", "-h", "127.0.0.1", "-U", fixtureUser).Run() != nil {
+		select {
+		case <-ctx.Done():
+			t.Fatal("fixture startup timed out")
+		case <-time.After(200 * time.Millisecond):
+		}
+	}
+	return id, port
+}
+
+func target(t *testing.T, port int, mode instance.TLSMode) instance.ObservationTarget {
+	t.Helper()
+	id, err := instance.NewIdentity("fixture-uid", "fixture")
+	if err != nil {
+		t.Fatal(err)
+	}
+	revision, err := instance.NewRevision(1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	endpoint, err := instance.NewEndpoint(instance.EndpointValues{
+		Host:               fixtureHost,
+		HostAddr:           "127.0.0.1",
+		Port:               port,
+		ManagementDatabase: fixtureUser,
+		TLSMode:            mode,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	ref, err := instance.NewCredentialReference(instance.CredentialReferenceValues{
+		Name:        secretName,
+		UsernameKey: "login",
+		PasswordKey: "credential",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	definition, err := instance.NewDefinition(endpoint, ref)
+	if err != nil {
+		t.Fatal(err)
+	}
+	value, err := instance.NewObservationTarget(id, revision, definition)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return value
+}
+
+// 在真实读取前设置屏障，确定性验证观测期间 Secret 变化；实际数据仍来自 API server。
+type gatedReader struct {
+	application.CredentialReader
+	beforeRead func()
+}
+
+func (r *gatedReader) Read(ctx context.Context, ref instance.CredentialReference) (application.Credentials, error) {
+	if r.beforeRead != nil {
+		r.beforeRead()
+	}
+	return r.CredentialReader.Read(ctx, ref)
+}
+
+// credentialFixture 为每个场景创建独立 API server、PostgreSQL 和应用服务。
+type credentialFixture struct {
+	ctx          context.Context
+	client       *kubernetes.Clientset
+	reader       *secretadapter.SecretCredentials
+	deniedReader *secretadapter.SecretCredentials
+	gate         *gatedReader
+	service      *application.InstanceService
+	target       instance.ObservationTarget
+	containerID  string
+	port         int
+}
+
+func newCredentialFixture(t *testing.T) *credentialFixture {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	t.Cleanup(cancel)
+
+	environment := &envtest.Environment{}
+	config, err := environment.Start()
+	if err != nil {
+		t.Fatal("envtest startup failed", err)
+	}
+	t.Cleanup(func() {
+		if err := environment.Stop(); err != nil {
+			t.Error("envtest cleanup failed", err)
+		}
+	})
+
+	client, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		t.Fatal("cannot create test client")
+	}
+	for _, namespace := range []string{controllerNamespace, "unrelated"} {
+		_, err := client.CoreV1().Namespaces().Create(
+			ctx,
+			&corev1.Namespace{Name: namespace},
+			metav1.CreateOptions{},
+		)
+		if err != nil {
+			t.Fatal("cannot create fixture namespace")
+		}
+	}
+
+	reader, err := secretadapter.NewSecretCredentials(config, controllerNamespace)
+	if err != nil {
+		t.Fatal(err)
+	}
+	user, err := environment.AddUser(envtest.User{Name: "without-secret-access"}, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	deniedReader, err := secretadapter.NewSecretCredentials(user.Config(), controllerNamespace)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	containerID, port := postgresFixture(t, ctx)
+	gate := &gatedReader{CredentialReader: reader}
+	service, err := application.NewInstanceService(gate, postgresql.Connector{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(service.Close)
+
+	return &credentialFixture{
+		ctx:          ctx,
+		client:       client,
+		reader:       reader,
+		deniedReader: deniedReader,
+		gate:         gate,
+		service:      service,
+		target:       target(t, port, instance.TLSDisable),
+		containerID:  containerID,
+		port:         port,
+	}
+}
+
+func (f *credentialFixture) createSecret(t *testing.T, namespace string) {
+	t.Helper()
+	secret := &corev1.Secret{
+		Name: secretName,
+		Data: map[string][]byte{
+			"login":      []byte(fixtureUser),
+			"credential": []byte(fixturePassword),
+		},
+	}
+	if _, err := f.client.CoreV1().Secrets(namespace).Create(f.ctx, secret, metav1.CreateOptions{}); err != nil {
+		t.Fatal("cannot create fixture Secret")
+	}
+}
+
+func (f *credentialFixture) updateSecret(t *testing.T, change func(*corev1.Secret)) {
+	t.Helper()
+	secrets := f.client.CoreV1().Secrets(controllerNamespace)
+	secret, err := secrets.Get(f.ctx, secretName, metav1.GetOptions{})
+	if err != nil {
+		t.Fatal("cannot read fixture Secret")
+	}
+	change(secret)
+	if _, err := secrets.Update(f.ctx, secret, metav1.UpdateOptions{}); err != nil {
+		t.Fatal("cannot update fixture Secret")
+	}
+}
+
+func (f *credentialFixture) observeVersion(t *testing.T) {
+	t.Helper()
+	version, err := f.service.ObserveVersion(f.ctx, f.target)
+	if err != nil {
+		t.Fatal("version observation failed", err)
+	}
+	if version == "" {
+		t.Fatal("successful observation returned an empty version")
+	}
+}
+
+func (f *credentialFixture) queryPostgres(t *testing.T, sql string) string {
+	t.Helper()
+	output, err := exec.CommandContext(
+		f.ctx, "docker", dockerExec, f.containerID,
+		"psql", "-U", fixtureUser, "-tAc", sql,
+	).Output()
+	if err != nil {
+		t.Fatal("fixture SQL failed")
+	}
+	return strings.TrimSpace(string(output))
+}
+
+func (f *credentialFixture) backendIDs(t *testing.T) string {
+	t.Helper()
+	return f.queryPostgres(t, `
+		SELECT pid
+		FROM pg_stat_activity
+		WHERE application_name = 'ayatori-database-management'
+		ORDER BY pid
+	`)
+}

internal/database/adapter/postgresql/tls_integration_test.go

@@ -0,0 +1,140 @@
+//go:build integration
+
+/*
+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 postgresql_test
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"math/big"
+	"net"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"git.ddupan.top/panxiao81/ayatori/internal/database/adapter/postgresql"
+	"git.ddupan.top/panxiao81/ayatori/internal/database/application"
+	"git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance"
+)
+
+func fixtureCertificate(t *testing.T) (string, string) {
+	t.Helper()
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cert := &x509.Certificate{
+		SerialNumber:          big.NewInt(1),
+		Subject:               pkix.Name{CommonName: fixtureHost},
+		NotBefore:             time.Now().Add(-time.Hour),
+		NotAfter:              time.Now().Add(time.Hour),
+		DNSNames:              []string{fixtureHost},
+		IPAddresses:           []net.IP{net.ParseIP("127.0.0.1")},
+		IsCA:                  true,
+		BasicConstraintsValid: true,
+		KeyUsage:              x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+	}
+	der, err := x509.CreateCertificate(rand.Reader, cert, cert, &key.PublicKey, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	encodedKey, err := x509.MarshalECPrivateKey(key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	dir := t.TempDir()
+	certPath := filepath.Join(dir, "server.crt")
+	keyPath := filepath.Join(dir, "server.key")
+	if err := os.WriteFile(certPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0600); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(keyPath, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: encodedKey}), 0600); err != nil {
+		t.Fatal(err)
+	}
+	return certPath, keyPath
+}
+
+func TestPostgreSQLTLSHostIdentity(t *testing.T) {
+	const psql = "psql"
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	id, port := postgresFixture(t, ctx)
+	certPath, keyPath := fixtureCertificate(t)
+	commands := [][]string{
+		{"cp", certPath, id + ":/tmp/server.crt"},
+		{"cp", keyPath, id + ":/tmp/server.key"},
+		{dockerExec, "-u", "0", id, "chown", "postgres:postgres", "/tmp/server.crt", "/tmp/server.key"},
+		{dockerExec, id, psql, "-U", fixtureUser, "-c", "ALTER SYSTEM SET ssl_cert_file='/tmp/server.crt'"},
+		{dockerExec, id, psql, "-U", fixtureUser, "-c", "ALTER SYSTEM SET ssl_key_file='/tmp/server.key'"},
+		{dockerExec, id, psql, "-U", fixtureUser, "-c", "ALTER SYSTEM SET ssl=on"},
+		{dockerExec, id, psql, "-U", fixtureUser, "-c", "SELECT pg_reload_conf()"},
+	}
+	for _, args := range commands {
+		if exec.CommandContext(ctx, "docker", args...).Run() != nil {
+			t.Fatal("TLS fixture setup failed")
+		}
+	}
+	credentials, err := application.NewCredentials(fixtureUser, fixturePassword)
+	if err != nil {
+		t.Fatal(err)
+	}
+	connector := postgresql.Connector{RootCert: certPath}
+	endpoint := target(t, port, instance.TLSVerifyFull).Definition().Endpoint()
+	db, err := connector.Connect(ctx, endpoint, credentials)
+	if err != nil {
+		t.Fatal("trusted DNS SAN connection failed", err)
+	}
+	if version, err := db.Version(ctx); err != nil || version == "" {
+		db.Close()
+		t.Fatal("TLS metadata read failed", err)
+	}
+	db.Close()
+	values := endpoint.Values()
+	values.Host = "127.0.0.1"
+	ipEndpoint, err := instance.NewEndpoint(values)
+	if err != nil {
+		t.Fatal(err)
+	}
+	db, err = connector.Connect(ctx, ipEndpoint, credentials)
+	if err != nil {
+		t.Fatal("trusted IP SAN connection failed", err)
+	}
+	db.Close()
+	values.Host = "wrong.invalid"
+	wrongEndpoint, err := instance.NewEndpoint(values)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if db, err := connector.Connect(ctx, wrongEndpoint, credentials); err == nil {
+		db.Close()
+		t.Fatal("wrong TLS hostname accepted")
+	}
+	otherCA, _ := fixtureCertificate(t)
+	if db, err := (postgresql.Connector{RootCert: otherCA}).Connect(ctx, endpoint, credentials); err == nil {
+		db.Close()
+		t.Fatal("wrong CA accepted")
+	}
+}

internal/database/application/credentials.go

@@ -0,0 +1,58 @@
+/*
+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 application 定义 Database 用例与适配器之间的边界。
+package application
+
+import (
+	"context"
+	"errors"
+
+	"git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance"
+)
+
+var (
+	ErrCredentialsUnavailable = errors.New("management credentials unavailable")
+	ErrCredentialsInvalid     = errors.New("management credentials invalid")
+)
+
+// Credentials 只存在于应用与连接适配器内存，不进入领域对象或持久化状态。
+type Credentials struct {
+	username string
+	password string
+}
+
+func NewCredentials(username, password string) (Credentials, error) {
+	if username == "" || password == "" {
+		return Credentials{}, ErrCredentialsInvalid
+	}
+	return Credentials{username: username, password: password}, nil
+}
+
+func (c Credentials) Username() string { return c.username }
+func (c Credentials) Password() string { return c.password }
+func (c Credentials) String() string   { return "[redacted management credentials]" }
+func (c Credentials) GoString() string { return c.String() }
+
+// MarshalJSON 显式隐藏内容，避免未来字段调整意外改变日志或序列化行为。
+func (c Credentials) MarshalJSON() ([]byte, error) {
+	return []byte(`"[redacted management credentials]"`), nil
+}
+
+// CredentialReader 返回本次读取的有效值；metadata 不参与凭据相等比较。
+type CredentialReader interface {
+	Read(context.Context, instance.CredentialReference) (Credentials, error)
+}

internal/database/application/credentials_test.go

@@ -0,0 +1,70 @@
+/*
+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 application
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+)
+
+const testUsername = "test-user"
+
+func TestCredentialsRejectEmptyValues(t *testing.T) {
+	for _, values := range [][2]string{
+		{"", "test-password"},
+		{testUsername, ""},
+		{"", ""},
+	} {
+		if _, err := NewCredentials(values[0], values[1]); err != ErrCredentialsInvalid {
+			t.Fatal("empty credential was accepted")
+		}
+	}
+}
+
+func TestCredentialAndServiceFormattingIsRedacted(t *testing.T) {
+	const canary = "SECRET-CANARY-never-log-this"
+	credentials, err := NewCredentials(canary, canary)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if credentials.Username() != canary || credentials.Password() != canary {
+		t.Fatal("explicit credential access changed values")
+	}
+
+	service, err := NewInstanceService(&sourceStub{credentials: credentials}, &connectorStub{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(service.Close)
+
+	encoded, err := json.Marshal(credentials)
+	if err != nil {
+		t.Fatal(err)
+	}
+	outputs := []string{
+		string(encoded),
+		fmt.Sprintf("%v %+v %#v", credentials, credentials, credentials),
+		fmt.Sprintf("%v %+v %#v", service, service, service),
+	}
+	for _, output := range outputs {
+		if strings.Contains(output, canary) {
+			t.Fatal("formatting leaked credential data")
+		}
+	}
+}

internal/database/application/instance_service.go

@@ -0,0 +1,171 @@
+/*
+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 application
+
+import (
+	"context"
+	"errors"
+	"sync"
+
+	"git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance"
+)
+
+var (
+	ErrConnection         = errors.New("management connection unavailable")
+	ErrAuthentication     = errors.New("management authentication failed")
+	ErrObservation        = errors.New("management observation failed")
+	ErrCredentialsChanged = errors.New("management credentials changed during observation")
+	ErrClosed             = errors.New("instance service closed")
+)
+
+// Database 与 Connector 复用原项目 internal/instance/service.go 的能力边界。
+// 版本查询只是本切片的连通性观察，不能产生领域 Ready。
+type Database interface {
+	Version(context.Context) (string, error)
+	Close()
+}
+
+type Connector interface {
+	Connect(context.Context, instance.Endpoint, Credentials) (Database, error)
+}
+
+type entry struct {
+	target      instance.ObservationTarget
+	credentials Credentials
+	database    Database
+}
+
+// InstanceService 由原 Service 迁移：连接复用与释放属于应用装配，不属于 SQL adapter。
+// 保留原实现串行操作的约束，防止 Close 与查询并发；controller 停止 worker 后调用 Close。
+// 不缓存能力观察，不把连接存活等同于 Ready。凭据每轮重新读取，而非只在引用变化时读取。
+type InstanceService struct {
+	mu        sync.Mutex
+	source    CredentialReader
+	connector Connector
+	entries   map[string]*entry
+	closed    bool
+}
+
+func NewInstanceService(source CredentialReader, connector Connector) (*InstanceService, error) {
+	if source == nil || connector == nil {
+		return nil, errors.New("credential source and connector required")
+	}
+	return &InstanceService{
+		source:    source,
+		connector: connector,
+		entries:   make(map[string]*entry),
+	}, nil
+}
+
+func (s *InstanceService) String() string   { return "[redacted instance service]" }
+func (s *InstanceService) GoString() string { return s.String() }
+
+// ObserveVersion 返回当前目标和凭据下的版本；任何失败均返回空结果。
+// 调用者仍需使用 CR resourceVersion 保存前提防止 spec 并发修改；本方法不建立跨系统事务。
+func (s *InstanceService) ObserveVersion(ctx context.Context, target instance.ObservationTarget) (string, error) {
+	if err := target.Validate(); err != nil {
+		return "", err
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.closed {
+		return "", ErrClosed
+	}
+	if err := ctx.Err(); err != nil {
+		return "", err
+	}
+
+	// 先读取有效凭据。读取失败时不得继续使用缓存中的旧连接。
+	name := target.Identity().Name()
+	credentials, err := s.source.Read(ctx, target.Definition().AdminCredential())
+	if err != nil {
+		s.release(name)
+		return "", credentialError(err)
+	}
+	if credentials.username == "" || credentials.password == "" {
+		s.release(name)
+		return "", ErrCredentialsInvalid
+	}
+
+	// 连接身份与有效值均未变化时复用 pgxpool；generation 本身不要求换池。
+	current := s.entries[name]
+	if current != nil && (current.target.Identity() != target.Identity() ||
+		current.target.Definition() != target.Definition() || current.credentials != credentials) {
+		s.release(name)
+		current = nil
+	}
+	if current == nil {
+		database, err := s.connector.Connect(ctx, target.Definition().Endpoint(), credentials)
+		if err != nil {
+			return "", err
+		}
+		current = &entry{
+			target:      target,
+			credentials: credentials,
+			database:    database,
+		}
+		s.entries[name] = current
+	}
+
+	version, err := current.database.Version(ctx)
+	if err != nil {
+		s.release(name)
+		return "", err
+	}
+
+	// 回读后再检查凭据，避免把轮换前取得的结果交给新凭据的调用链。
+	latest, err := s.source.Read(ctx, target.Definition().AdminCredential())
+	if err != nil {
+		s.release(name)
+		return "", credentialError(err)
+	}
+	if latest != credentials {
+		s.release(name)
+		return "", ErrCredentialsChanged
+	}
+	return version, nil
+}
+
+func credentialError(err error) error {
+	if errors.Is(err, ErrCredentialsInvalid) {
+		return ErrCredentialsInvalid
+	}
+	return ErrCredentialsUnavailable
+}
+
+// Forget 只释放本地连接；不删除数据库或 registry，不替代 Instance finalizer。
+func (s *InstanceService) Forget(name string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.release(name)
+}
+
+func (s *InstanceService) release(name string) {
+	if current := s.entries[name]; current != nil {
+		current.database.Close()
+	}
+	delete(s.entries, name)
+}
+
+func (s *InstanceService) Close() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.closed = true
+	for name := range s.entries {
+		s.release(name)
+	}
+}

internal/database/application/instance_service_test.go

@@ -0,0 +1,182 @@
+/*
+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 application
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance"
+)
+
+// 延续源项目 Service 测试，用于穷举身份与装配失败；真实行为由 adapter 集成测试验证。
+type sourceStub struct {
+	credentials Credentials
+	err         error
+}
+
+func (s *sourceStub) Read(context.Context, instance.CredentialReference) (Credentials, error) {
+	return s.credentials, s.err
+}
+
+type databaseStub struct {
+	closes int
+	err    error
+}
+
+func (d *databaseStub) Version(context.Context) (string, error) { return "17", d.err }
+func (d *databaseStub) Close() {
+	d.closes++
+}
+
+type connectorStub struct {
+	databases []*databaseStub
+	err       error
+}
+
+func (c *connectorStub) Connect(context.Context, instance.Endpoint, Credentials) (Database, error) {
+	if c.err != nil {
+		return nil, c.err
+	}
+	db := &databaseStub{}
+	c.databases = append(c.databases, db)
+	return db, nil
+}
+
+func serviceTarget(t *testing.T, uid, host, secret string, generation int64) instance.ObservationTarget {
+	t.Helper()
+	id, err := instance.NewIdentity(uid, "shared")
+	if err != nil {
+		t.Fatal(err)
+	}
+	revision, err := instance.NewRevision(generation)
+	if err != nil {
+		t.Fatal(err)
+	}
+	endpoint, err := instance.NewEndpoint(instance.EndpointValues{
+		Host:               host,
+		HostAddr:           "127.0.0.1",
+		Port:               5432,
+		ManagementDatabase: "postgres",
+		TLSMode:            instance.TLSDisable,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	ref, err := instance.NewCredentialReference(instance.CredentialReferenceValues{
+		Name:        secret,
+		UsernameKey: "user",
+		PasswordKey: "pass",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	definition, err := instance.NewDefinition(endpoint, ref)
+	if err != nil {
+		t.Fatal(err)
+	}
+	target, err := instance.NewObservationTarget(id, revision, definition)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return target
+}
+
+func TestInstanceConnectionIdentity(t *testing.T) {
+	source := &sourceStub{credentials: Credentials{username: testUsername, password: "test-only"}}
+	connector := &connectorStub{}
+	service, err := NewInstanceService(source, connector)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer service.Close()
+	ctx := context.Background()
+	cases := []struct {
+		name            string
+		target          instance.ObservationTarget
+		wantConnections int
+	}{
+		{"initial connection", serviceTarget(t, "uid-1", "first", "admin", 1), 1},
+		{"generation alone", serviceTarget(t, "uid-1", "first", "admin", 2), 1},
+		{"endpoint changed", serviceTarget(t, "uid-1", "second", "admin", 3), 2},
+		{"reference changed", serviceTarget(t, "uid-1", "second", "replacement", 4), 3},
+		{"same name with new UID", serviceTarget(t, "uid-2", "second", "replacement", 1), 4},
+	}
+	for _, testCase := range cases {
+		if _, err := service.ObserveVersion(ctx, testCase.target); err != nil {
+			t.Fatal(err)
+		}
+		if len(connector.databases) != testCase.wantConnections {
+			t.Fatalf("%s: got %d connections, want %d", testCase.name, len(connector.databases), testCase.wantConnections)
+		}
+	}
+	for _, db := range connector.databases[:3] {
+		if db.closes != 1 {
+			t.Fatal("replaced connection not closed exactly once")
+		}
+	}
+	service.Forget("shared")
+	service.Forget("shared")
+	if connector.databases[3].closes != 1 {
+		t.Fatal("forget did not close exactly once")
+	}
+}
+
+func TestInstanceAssemblyFailureRecovery(t *testing.T) {
+	ctx := context.Background()
+	target := serviceTarget(t, "uid-1", "first", "admin", 1)
+	source := &sourceStub{
+		credentials: Credentials{username: testUsername, password: "test-only"},
+		err:         errors.New("unsafe source error"),
+	}
+	connector := &connectorStub{err: ErrConnection}
+	service, err := NewInstanceService(source, connector)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer service.Close()
+	if version, err := service.ObserveVersion(ctx, target); version != "" || !errors.Is(err, ErrCredentialsUnavailable) {
+		t.Fatal("unsafe source error escaped")
+	}
+	source.err = nil
+	if version, err := service.ObserveVersion(ctx, target); version != "" || !errors.Is(err, ErrConnection) {
+		t.Fatal("connection failure returned evidence")
+	}
+	connector.err = nil
+	if _, err := service.ObserveVersion(ctx, target); err != nil {
+		t.Fatal(err)
+	}
+	connector.databases[0].err = ErrObservation
+	if version, err := service.ObserveVersion(ctx, target); version != "" || !errors.Is(err, ErrObservation) {
+		t.Fatal("failed query returned evidence")
+	}
+	if connector.databases[0].closes != 1 {
+		t.Fatal("failed connection retained")
+	}
+	if _, err := service.ObserveVersion(ctx, target); err != nil {
+		t.Fatal("retry failed", err)
+	}
+	service.Close()
+	service.Close()
+	if connector.databases[1].closes != 1 {
+		t.Fatal("shutdown did not close once")
+	}
+	if _, err := service.ObserveVersion(ctx, target); !errors.Is(err, ErrClosed) {
+		t.Fatal("closed service accepted work")
+	}
+}