feat: 实现 PostgreSQL 所有权 registry

internal/postgresql/registry/registry.go

@@ -0,0 +1,307 @@
+/*
+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 registry persists controller ownership evidence in the PostgreSQL
+// management database. It deliberately does not persist reconciliation phases;
+// those belong to the Kubernetes resource status.
+package registry
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+const schemaVersion = 1
+
+var (
+	// ErrNotFound indicates that the registry has no matching ownership record.
+	ErrNotFound = errors.New("registry ownership record not found")
+	// ErrConflict indicates that a requested name or path belongs to another Tenant UID.
+	ErrConflict = errors.New("registry ownership conflict")
+)
+
+// Beginner is implemented by pgx.Conn and pgxpool.Pool.
+type Beginner interface {
+	Begin(context.Context) (pgx.Tx, error)
+}
+
+// Store manages ownership records in one PostgreSQLInstance management database.
+type Store struct {
+	db Beginner
+}
+
+// NewStore creates a registry store backed by a PostgreSQL connection or pool.
+func NewStore(db Beginner) *Store {
+	return &Store{db: db}
+}
+
+// Ownership identifies every external resource reserved for one Tenant UID.
+type Ownership struct {
+	InstanceUID     string
+	TenantUID       string
+	TenantNamespace string
+	TenantName      string
+	DatabaseName    string
+	RoleName        string
+	CredentialPath  string
+}
+
+// Record is the persisted ownership state.
+type Record struct {
+	Ownership
+	Managed    bool
+	CreatedAt  time.Time
+	UpdatedAt  time.Time
+	RetainedAt *time.Time
+}
+
+// ClaimResult reports whether Claim inserted a new record or found the same claim.
+type ClaimResult string
+
+const (
+	ClaimCreated ClaimResult = "Created"
+	ClaimOwned   ClaimResult = "Owned"
+)
+
+// Bootstrap creates the private schema and the current registry table idempotently.
+func (s *Store) Bootstrap(ctx context.Context) error {
+	if s == nil || s.db == nil {
+		return errors.New("bootstrap registry: nil database")
+	}
+
+	tx, err := s.db.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("begin registry bootstrap: %w", err)
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	for _, statement := range bootstrapStatements {
+		if _, err := tx.Exec(ctx, statement); err != nil {
+			return fmt.Errorf("bootstrap registry schema version %d: %w", schemaVersion, err)
+		}
+	}
+	var version int
+	if err := tx.QueryRow(ctx, `SELECT version FROM postgresql_tenant_operator.schema_version WHERE singleton`).Scan(&version); err != nil {
+		return fmt.Errorf("read registry schema version: %w", err)
+	}
+	if version != schemaVersion {
+		return fmt.Errorf("registry schema version %d is unsupported; expected %d", version, schemaVersion)
+	}
+
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("commit registry bootstrap: %w", err)
+	}
+	return nil
+}
+
+// Claim reserves all names for an owner. Repeating an identical claim is idempotent.
+func (s *Store) Claim(ctx context.Context, owner Ownership) (ClaimResult, error) {
+	if s == nil || s.db == nil {
+		return "", errors.New("claim registry ownership: nil database")
+	}
+	if err := owner.validate(); err != nil {
+		return "", err
+	}
+
+	tx, err := s.db.Begin(ctx)
+	if err != nil {
+		return "", fmt.Errorf("begin registry claim: %w", err)
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	commandTag, err := tx.Exec(ctx, claimStatement,
+		owner.InstanceUID,
+		owner.TenantUID,
+		owner.TenantNamespace,
+		owner.TenantName,
+		owner.DatabaseName,
+		owner.RoleName,
+		owner.CredentialPath,
+	)
+	if err != nil {
+		return "", fmt.Errorf("insert registry claim: %w", err)
+	}
+
+	record, err := getByTenantUID(ctx, tx, owner.InstanceUID, owner.TenantUID)
+	if err != nil {
+		if errors.Is(err, ErrNotFound) {
+			return "", fmt.Errorf("%w: database, role, tenant identity, or credential path is already reserved", ErrConflict)
+		}
+		return "", err
+	}
+	if !record.equal(owner) || !record.Managed {
+		return "", fmt.Errorf("%w: existing record does not match the requested managed owner", ErrConflict)
+	}
+
+	if err := tx.Commit(ctx); err != nil {
+		return "", fmt.Errorf("commit registry claim: %w", err)
+	}
+	if commandTag.RowsAffected() == 1 {
+		return ClaimCreated, nil
+	}
+	return ClaimOwned, nil
+}
+
+// Get returns the ownership record for an Instance UID and Tenant UID.
+func (s *Store) Get(ctx context.Context, instanceUID, tenantUID string) (Record, error) {
+	if s == nil || s.db == nil {
+		return Record{}, errors.New("get registry record: nil database")
+	}
+	if instanceUID == "" || tenantUID == "" {
+		return Record{}, errors.New("get registry record: instance UID and tenant UID are required")
+	}
+
+	tx, err := s.db.Begin(ctx)
+	if err != nil {
+		return Record{}, fmt.Errorf("begin registry read: %w", err)
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	record, err := getByTenantUID(ctx, tx, instanceUID, tenantUID)
+	if err != nil {
+		return Record{}, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return Record{}, fmt.Errorf("commit registry read: %w", err)
+	}
+	return record, nil
+}
+
+// MarkRetained changes a matching managed ownership record into an unmanaged tombstone.
+// Repeating the operation for the same tombstone is safe.
+func (s *Store) MarkRetained(ctx context.Context, owner Ownership) error {
+	return s.changeOwnership(ctx, owner, "mark registry record retained", func(ctx context.Context, tx pgx.Tx, record Record) error {
+		if !record.Managed {
+			return nil
+		}
+		tag, err := tx.Exec(ctx, markRetainedStatement, owner.InstanceUID, owner.TenantUID)
+		if err != nil {
+			return err
+		}
+		if tag.RowsAffected() != 1 {
+			return ErrConflict
+		}
+		return nil
+	})
+}
+
+// Delete removes a matching managed record after its external resources have been deleted.
+// An already absent record is treated as a successful retry; a retained record is never deleted.
+func (s *Store) Delete(ctx context.Context, owner Ownership) error {
+	return s.changeOwnership(ctx, owner, "delete registry record", func(ctx context.Context, tx pgx.Tx, record Record) error {
+		if !record.Managed {
+			return ErrConflict
+		}
+		tag, err := tx.Exec(ctx, deleteStatement, owner.InstanceUID, owner.TenantUID)
+		if err != nil {
+			return err
+		}
+		if tag.RowsAffected() != 1 {
+			return ErrConflict
+		}
+		return nil
+	})
+}
+
+func (s *Store) changeOwnership(
+	ctx context.Context,
+	owner Ownership,
+	operation string,
+	change func(context.Context, pgx.Tx, Record) error,
+) error {
+	if s == nil || s.db == nil {
+		return fmt.Errorf("%s: nil database", operation)
+	}
+	if err := owner.validate(); err != nil {
+		return fmt.Errorf("%s: %w", operation, err)
+	}
+
+	tx, err := s.db.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("begin %s: %w", operation, err)
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	record, err := getByTenantUIDForUpdate(ctx, tx, owner.InstanceUID, owner.TenantUID)
+	if errors.Is(err, ErrNotFound) && operation == "delete registry record" {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("%s: %w", operation, err)
+	}
+	if !record.equal(owner) {
+		return fmt.Errorf("%s: %w", operation, ErrConflict)
+	}
+	if err := change(ctx, tx, record); err != nil {
+		return fmt.Errorf("%s: %w", operation, err)
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("commit %s: %w", operation, err)
+	}
+	return nil
+}
+
+func getByTenantUIDForUpdate(ctx context.Context, tx pgx.Tx, instanceUID, tenantUID string) (Record, error) {
+	return scanRecord(tx.QueryRow(ctx, getByTenantUIDForUpdateStatement, instanceUID, tenantUID))
+}
+
+func getByTenantUID(ctx context.Context, tx pgx.Tx, instanceUID, tenantUID string) (Record, error) {
+	return scanRecord(tx.QueryRow(ctx, getByTenantUIDStatement, instanceUID, tenantUID))
+}
+
+type rowScanner interface {
+	Scan(...any) error
+}
+
+func scanRecord(row rowScanner) (Record, error) {
+	var record Record
+	err := row.Scan(
+		&record.InstanceUID,
+		&record.TenantUID,
+		&record.TenantNamespace,
+		&record.TenantName,
+		&record.DatabaseName,
+		&record.RoleName,
+		&record.CredentialPath,
+		&record.Managed,
+		&record.CreatedAt,
+		&record.UpdatedAt,
+		&record.RetainedAt,
+	)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return Record{}, ErrNotFound
+	}
+	if err != nil {
+		return Record{}, fmt.Errorf("read registry record: %w", err)
+	}
+	return record, nil
+}
+
+func (o Ownership) validate() error {
+	if o.InstanceUID == "" || o.TenantUID == "" || o.TenantNamespace == "" || o.TenantName == "" ||
+		o.DatabaseName == "" || o.RoleName == "" || o.CredentialPath == "" {
+		return errors.New("claim registry ownership: all ownership fields are required")
+	}
+	return nil
+}
+
+func (o Ownership) equal(other Ownership) bool {
+	return o == other
+}

internal/postgresql/registry/registry_integration_test.go

@@ -0,0 +1,132 @@
+/*
+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 registry
+
+import (
+	"context"
+	"errors"
+	"os"
+	"testing"
+
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+func TestRegistryLifecycle(t *testing.T) {
+	dsn := os.Getenv("POSTGRES_TEST_DSN")
+	if dsn == "" {
+		t.Skip("POSTGRES_TEST_DSN is not set")
+	}
+
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("create PostgreSQL pool: %v", err)
+	}
+	t.Cleanup(pool.Close)
+	if err := pool.Ping(ctx); err != nil {
+		t.Fatalf("ping PostgreSQL: %v", err)
+	}
+
+	dropRegistrySchema(t, ctx, pool)
+	t.Cleanup(func() { dropRegistrySchema(t, ctx, pool) })
+
+	store := NewStore(pool)
+	if err := store.Bootstrap(ctx); err != nil {
+		t.Fatalf("Bootstrap() error = %v", err)
+	}
+	if err := store.Bootstrap(ctx); err != nil {
+		t.Fatalf("second Bootstrap() error = %v", err)
+	}
+
+	owner := testOwnership("tenant-uid-1", "netbox", "netbox", "netbox", "postgresql-tenants/netbox/netbox")
+	result, err := store.Claim(ctx, owner)
+	if err != nil || result != ClaimCreated {
+		t.Fatalf("first Claim() = %q, %v; want %q, nil", result, err, ClaimCreated)
+	}
+	result, err = store.Claim(ctx, owner)
+	if err != nil || result != ClaimOwned {
+		t.Fatalf("second Claim() = %q, %v; want %q, nil", result, err, ClaimOwned)
+	}
+
+	record, err := store.Get(ctx, owner.InstanceUID, owner.TenantUID)
+	if err != nil {
+		t.Fatalf("Get() error = %v", err)
+	}
+	if record.Ownership != owner || !record.Managed || record.RetainedAt != nil {
+		t.Fatalf("Get() = %#v; want matching managed owner", record)
+	}
+
+	conflict := testOwnership("tenant-uid-2", "other", owner.DatabaseName, "other", "postgresql-tenants/other/other")
+	if _, err := store.Claim(ctx, conflict); !errors.Is(err, ErrConflict) {
+		t.Fatalf("conflicting Claim() error = %v, want ErrConflict", err)
+	}
+
+	if err := store.MarkRetained(ctx, owner); err != nil {
+		t.Fatalf("MarkRetained() error = %v", err)
+	}
+	if err := store.MarkRetained(ctx, owner); err != nil {
+		t.Fatalf("second MarkRetained() error = %v", err)
+	}
+	record, err = store.Get(ctx, owner.InstanceUID, owner.TenantUID)
+	if err != nil || record.Managed || record.RetainedAt == nil {
+		t.Fatalf("retained Get() = %#v, %v; want unmanaged tombstone", record, err)
+	}
+	if err := store.Delete(ctx, owner); !errors.Is(err, ErrConflict) {
+		t.Fatalf("Delete(retained) error = %v, want ErrConflict", err)
+	}
+	if _, err := store.Claim(ctx, owner); !errors.Is(err, ErrConflict) {
+		t.Fatalf("Claim(retained) error = %v, want ErrConflict", err)
+	}
+
+	deletable := testOwnership("tenant-uid-3", "gitea", "gitea", "gitea", "postgresql-tenants/gitea/gitea")
+	if _, err := store.Claim(ctx, deletable); err != nil {
+		t.Fatalf("Claim(deletable) error = %v", err)
+	}
+	if err := store.Delete(ctx, deletable); err != nil {
+		t.Fatalf("Delete() error = %v", err)
+	}
+	if err := store.Delete(ctx, deletable); err != nil {
+		t.Fatalf("second Delete() error = %v", err)
+	}
+	if _, err := store.Get(ctx, deletable.InstanceUID, deletable.TenantUID); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("Get(deleted) error = %v, want ErrNotFound", err)
+	}
+}
+
+func testOwnership(tenantUID, tenantName, databaseName, roleName, credentialPath string) Ownership {
+	return Ownership{
+		InstanceUID:     "instance-uid-1",
+		TenantUID:       tenantUID,
+		TenantNamespace: tenantName,
+		TenantName:      tenantName,
+		DatabaseName:    databaseName,
+		RoleName:        roleName,
+		CredentialPath:  credentialPath,
+	}
+}
+
+type schemaDropper interface {
+	Exec(context.Context, string, ...any) (pgconn.CommandTag, error)
+}
+
+func dropRegistrySchema(t *testing.T, ctx context.Context, db schemaDropper) {
+	t.Helper()
+	if _, err := db.Exec(ctx, `DROP SCHEMA IF EXISTS postgresql_tenant_operator CASCADE`); err != nil {
+		t.Fatalf("drop registry schema: %v", err)
+	}
+}

internal/postgresql/registry/sql.go

@@ -0,0 +1,86 @@
+/*
+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 registry
+
+var bootstrapStatements = []string{
+	`CREATE SCHEMA IF NOT EXISTS postgresql_tenant_operator`,
+	`CREATE TABLE IF NOT EXISTS postgresql_tenant_operator.schema_version (
+    singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton),
+    version integer NOT NULL
+)`,
+	`INSERT INTO postgresql_tenant_operator.schema_version (singleton, version)
+VALUES (true, 1)
+ON CONFLICT (singleton) DO NOTHING`,
+	`CREATE TABLE IF NOT EXISTS postgresql_tenant_operator.tenant_ownership (
+    instance_uid text NOT NULL,
+    tenant_uid text NOT NULL,
+    tenant_namespace text NOT NULL,
+    tenant_name text NOT NULL,
+    database_name text NOT NULL,
+    role_name text NOT NULL,
+    credential_path text NOT NULL,
+    managed boolean NOT NULL DEFAULT true,
+    created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
+    updated_at timestamptz NOT NULL DEFAULT clock_timestamp(),
+    retained_at timestamptz,
+    PRIMARY KEY (instance_uid, tenant_uid),
+    UNIQUE (instance_uid, tenant_namespace, tenant_name),
+    UNIQUE (instance_uid, database_name),
+    UNIQUE (instance_uid, role_name),
+    UNIQUE (credential_path),
+    CHECK (managed OR retained_at IS NOT NULL)
+)`,
+}
+
+const claimStatement = `
+INSERT INTO postgresql_tenant_operator.tenant_ownership (
+    instance_uid,
+    tenant_uid,
+    tenant_namespace,
+    tenant_name,
+    database_name,
+    role_name,
+    credential_path
+) VALUES ($1, $2, $3, $4, $5, $6, $7)
+ON CONFLICT DO NOTHING`
+
+const getByTenantUIDStatement = `
+SELECT
+    instance_uid,
+    tenant_uid,
+    tenant_namespace,
+    tenant_name,
+    database_name,
+    role_name,
+    credential_path,
+    managed,
+    created_at,
+    updated_at,
+    retained_at
+FROM postgresql_tenant_operator.tenant_ownership
+WHERE instance_uid = $1 AND tenant_uid = $2`
+
+const getByTenantUIDForUpdateStatement = getByTenantUIDStatement + ` FOR UPDATE`
+
+const markRetainedStatement = `
+UPDATE postgresql_tenant_operator.tenant_ownership
+SET managed = false, retained_at = clock_timestamp(), updated_at = clock_timestamp()
+WHERE instance_uid = $1 AND tenant_uid = $2 AND managed = true`
+
+const deleteStatement = `
+DELETE FROM postgresql_tenant_operator.tenant_ownership
+WHERE instance_uid = $1 AND tenant_uid = $2 AND managed = true`