configcenter/internal/store/postgres/postgres.go

package postgres

import (
	"context"
	"embed"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/longpeng/configcenter/internal/domain"
	storepkg "github.com/longpeng/configcenter/internal/store"
)

//go:embed migrations/*.sql
var migrations embed.FS

type Store struct {
	pool *pgxpool.Pool
}

func Open(ctx context.Context, databaseURL string) (*Store, error) {
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		return nil, fmt.Errorf("create postgres pool: %w", err)
	}
	if err := pool.Ping(ctx); err != nil {
		pool.Close()
		return nil, fmt.Errorf("ping postgres: %w", err)
	}
	return &Store{pool: pool}, nil
}

func (s *Store) Close() { s.pool.Close() }

func (s *Store) Ping(ctx context.Context) error { return s.pool.Ping(ctx) }

func (s *Store) Migrate(ctx context.Context) error {
	entries, err := migrations.ReadDir("migrations")
	if err != nil {
		return fmt.Errorf("list migrations: %w", err)
	}
	for _, entry := range entries {
		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".sql") {
			continue
		}
		content, err := migrations.ReadFile("migrations/" + entry.Name())
		if err != nil {
			return fmt.Errorf("read migration %s: %w", entry.Name(), err)
		}
		if _, err := s.pool.Exec(ctx, string(content)); err != nil {
			return fmt.Errorf("apply migration %s: %w", entry.Name(), err)
		}
	}
	return nil
}

func (s *Store) ListApplications(ctx context.Context) ([]domain.Application, error) {
	rows, err := s.pool.Query(ctx, `SELECT id, app_code, name, department, owner, description, created_at, updated_at FROM applications ORDER BY app_code`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]domain.Application, 0)
	for rows.Next() {
		var item domain.Application
		if err := rows.Scan(&item.ID, &item.Code, &item.Name, &item.Department, &item.Owner, &item.Desc, &item.CreatedAt, &item.UpdatedAt); err != nil {
			return nil, err
		}
		result = append(result, item)
	}
	return result, rows.Err()
}

func (s *Store) CreateApplication(ctx context.Context, input domain.Application) (domain.Application, error) {
	err := s.pool.QueryRow(ctx, `
		INSERT INTO applications(app_code, name, department, owner, description)
		VALUES($1, $2, $3, $4, $5)
		RETURNING id, app_code, name, department, owner, description, created_at, updated_at`,
		input.Code, input.Name, input.Department, input.Owner, input.Desc,
	).Scan(&input.ID, &input.Code, &input.Name, &input.Department, &input.Owner, &input.Desc, &input.CreatedAt, &input.UpdatedAt)
	if err != nil {
		return domain.Application{}, classify(err)
	}
	s.audit(ctx, actor(input.UpdatedBy), "create", "app", input.ID, input)
	return input, nil
}

func (s *Store) UpdateApplication(ctx context.Context, id int64, input domain.Application) (domain.Application, error) {
	err := s.pool.QueryRow(ctx, `
		UPDATE applications SET name=$2, department=$3, owner=$4, description=$5, updated_at=now()
		WHERE id=$1
		RETURNING id, app_code, name, department, owner, description, created_at, updated_at`,
		id, input.Name, input.Department, input.Owner, input.Desc,
	).Scan(&input.ID, &input.Code, &input.Name, &input.Department, &input.Owner, &input.Desc, &input.CreatedAt, &input.UpdatedAt)
	if err != nil {
		return domain.Application{}, classify(err)
	}
	s.audit(ctx, actor(input.UpdatedBy), "update", "app", id, input)
	return input, nil
}

func (s *Store) DeleteApplication(ctx context.Context, id int64, actor string) error {
	return s.deleteAndAudit(ctx, "applications", "app", id, actor)
}

func (s *Store) ListEnvironments(ctx context.Context) ([]domain.Environment, error) {
	rows, err := s.pool.Query(ctx, `SELECT id, env_code, name, sort_order, description FROM environments ORDER BY sort_order, env_code`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]domain.Environment, 0)
	for rows.Next() {
		var item domain.Environment
		if err := rows.Scan(&item.ID, &item.Code, &item.Name, &item.Order, &item.Desc); err != nil {
			return nil, err
		}
		result = append(result, item)
	}
	return result, rows.Err()
}

func (s *Store) CreateEnvironment(ctx context.Context, input domain.Environment) (domain.Environment, error) {
	input.Code = strings.ToUpper(input.Code)
	err := s.pool.QueryRow(ctx, `
		INSERT INTO environments(env_code, name, sort_order, description) VALUES($1, $2, $3, $4)
		RETURNING id, env_code, name, sort_order, description`, input.Code, input.Name, input.Order, input.Desc,
	).Scan(&input.ID, &input.Code, &input.Name, &input.Order, &input.Desc)
	if err != nil {
		return domain.Environment{}, classify(err)
	}
	s.audit(ctx, actor(input.UpdatedBy), "create", "env", input.ID, input)
	return input, nil
}

func (s *Store) UpdateEnvironment(ctx context.Context, id int64, input domain.Environment) (domain.Environment, error) {
	err := s.pool.QueryRow(ctx, `
		UPDATE environments SET name=$2, sort_order=$3, description=$4 WHERE id=$1
		RETURNING id, env_code, name, sort_order, description`, id, input.Name, input.Order, input.Desc,
	).Scan(&input.ID, &input.Code, &input.Name, &input.Order, &input.Desc)
	if err != nil {
		return domain.Environment{}, classify(err)
	}
	s.audit(ctx, actor(input.UpdatedBy), "update", "env", id, input)
	return input, nil
}

func (s *Store) DeleteEnvironment(ctx context.Context, id int64, actor string) error {
	return s.deleteAndAudit(ctx, "environments", "env", id, actor)
}

func (s *Store) ListNamespaces(ctx context.Context, appID *int64) ([]domain.Namespace, error) {
	query := `SELECT id, app_id, name, format, description FROM namespaces`
	args := []any{}
	if appID != nil {
		query += ` WHERE app_id=$1`
		args = append(args, *appID)
	}
	query += ` ORDER BY app_id, name`
	rows, err := s.pool.Query(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]domain.Namespace, 0)
	for rows.Next() {
		var item domain.Namespace
		if err := rows.Scan(&item.ID, &item.AppID, &item.Name, &item.Type, &item.Desc); err != nil {
			return nil, err
		}
		result = append(result, item)
	}
	return result, rows.Err()
}

func (s *Store) GetNamespace(ctx context.Context, id int64) (domain.Namespace, error) {
	var item domain.Namespace
	err := s.pool.QueryRow(ctx, `SELECT id, app_id, name, format, description FROM namespaces WHERE id=$1`, id).
		Scan(&item.ID, &item.AppID, &item.Name, &item.Type, &item.Desc)
	return item, classify(err)
}

func (s *Store) CreateNamespace(ctx context.Context, input domain.Namespace) (domain.Namespace, error) {
	err := s.pool.QueryRow(ctx, `
		INSERT INTO namespaces(app_id, name, format, description) VALUES($1, $2, $3, $4)
		RETURNING id, app_id, name, format, description`, input.AppID, input.Name, input.Type, input.Desc,
	).Scan(&input.ID, &input.AppID, &input.Name, &input.Type, &input.Desc)
	if err != nil {
		return domain.Namespace{}, classify(err)
	}
	s.audit(ctx, actor(input.UpdatedBy), "create", "namespace", input.ID, input)
	return input, nil
}

func (s *Store) UpdateNamespace(ctx context.Context, id int64, input domain.Namespace) (domain.Namespace, error) {
	err := s.pool.QueryRow(ctx, `
		UPDATE namespaces SET format=$2, description=$3 WHERE id=$1
		RETURNING id, app_id, name, format, description`, id, input.Type, input.Desc,
	).Scan(&input.ID, &input.AppID, &input.Name, &input.Type, &input.Desc)
	if err != nil {
		return domain.Namespace{}, classify(err)
	}
	s.audit(ctx, actor(input.UpdatedBy), "update", "namespace", id, input)
	return input, nil
}

func (s *Store) DeleteNamespace(ctx context.Context, id int64, actor string) error {
	return s.deleteAndAudit(ctx, "namespaces", "namespace", id, actor)
}

func (s *Store) ListConfigItems(ctx context.Context, appID, namespaceID, envID int64) ([]domain.ConfigItem, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT ci.id, n.app_id, ci.namespace_id, ci.env_id, ci.key, ci.value, ci.released_value,
		       ci.pending_delete, ci.comment, ci.updated_by, ci.updated_at
		FROM config_items ci JOIN namespaces n ON n.id=ci.namespace_id
		WHERE n.app_id=$1 AND ci.namespace_id=$2 AND ci.env_id=$3 ORDER BY ci.key`, appID, namespaceID, envID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]domain.ConfigItem, 0)
	for rows.Next() {
		item, err := scanConfig(rows)
		if err != nil {
			return nil, err
		}
		result = append(result, item)
	}
	return result, rows.Err()
}

func (s *Store) GetConfigItem(ctx context.Context, id int64) (domain.ConfigItem, error) {
	row := s.pool.QueryRow(ctx, `
		SELECT ci.id, n.app_id, ci.namespace_id, ci.env_id, ci.key, ci.value, ci.released_value,
		       ci.pending_delete, ci.comment, ci.updated_by, ci.updated_at
		FROM config_items ci JOIN namespaces n ON n.id=ci.namespace_id WHERE ci.id=$1`, id)
	item, err := scanConfig(row)
	return item, classify(err)
}

func (s *Store) CreateConfigItem(ctx context.Context, input domain.ConfigItem) (domain.ConfigItem, error) {
	err := s.pool.QueryRow(ctx, `
		INSERT INTO config_items(namespace_id, env_id, key, value, comment, updated_by)
		SELECT n.id, $3, $4, $5, $6, $7 FROM namespaces n WHERE n.id=$2 AND n.app_id=$1
		RETURNING id, namespace_id, env_id, key, value, released_value, pending_delete, comment, updated_by, updated_at`,
		input.AppID, input.NamespaceID, input.EnvironmentID, input.Key, input.Value, input.Comment, actor(input.UpdatedBy),
	).Scan(&input.ID, &input.NamespaceID, &input.EnvironmentID, &input.Key, &input.Value, &input.ReleasedValue, &input.PendingDelete, &input.Comment, &input.UpdatedBy, &input.UpdatedAt)
	if err != nil {
		return domain.ConfigItem{}, classify(err)
	}
	s.audit(ctx, input.UpdatedBy, "create", "config", input.ID, input)
	return input, nil
}

func (s *Store) UpdateConfigItem(ctx context.Context, id int64, input domain.ConfigItem) (domain.ConfigItem, error) {
	err := s.pool.QueryRow(ctx, `
		UPDATE config_items SET value=$2, comment=$3, updated_by=$4, updated_at=now(), pending_delete=false WHERE id=$1
		RETURNING id, namespace_id, env_id, key, value, released_value, pending_delete, comment, updated_by, updated_at`,
		id, input.Value, input.Comment, actor(input.UpdatedBy),
	).Scan(&input.ID, &input.NamespaceID, &input.EnvironmentID, &input.Key, &input.Value, &input.ReleasedValue, &input.PendingDelete, &input.Comment, &input.UpdatedBy, &input.UpdatedAt)
	if err != nil {
		return domain.ConfigItem{}, classify(err)
	}
	if err := s.pool.QueryRow(ctx, `SELECT app_id FROM namespaces WHERE id=$1`, input.NamespaceID).Scan(&input.AppID); err != nil {
		return domain.ConfigItem{}, err
	}
	s.audit(ctx, input.UpdatedBy, "update", "config", id, input)
	return input, nil
}

func (s *Store) SetConfigItemPendingDelete(ctx context.Context, id int64, pending bool, updatedBy string) (domain.ConfigItem, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return domain.ConfigItem{}, err
	}
	defer tx.Rollback(ctx) //nolint:errcheck
	var released *string
	if err := tx.QueryRow(ctx, `SELECT released_value FROM config_items WHERE id=$1 FOR UPDATE`, id).Scan(&released); err != nil {
		return domain.ConfigItem{}, classify(err)
	}
	if released == nil && pending {
		if _, err := tx.Exec(ctx, `DELETE FROM config_items WHERE id=$1`, id); err != nil {
			return domain.ConfigItem{}, err
		}
		if err := insertAudit(ctx, tx, actor(updatedBy), "delete", "config", id, map[string]bool{"draftOnly": true}); err != nil {
			return domain.ConfigItem{}, err
		}
		if err := tx.Commit(ctx); err != nil {
			return domain.ConfigItem{}, err
		}
		return domain.ConfigItem{ID: id}, nil
	}
	var item domain.ConfigItem
	err = tx.QueryRow(ctx, `
		UPDATE config_items SET pending_delete=$2, updated_by=$3, updated_at=now() WHERE id=$1
		RETURNING id, namespace_id, env_id, key, value, released_value, pending_delete, comment, updated_by, updated_at`,
		id, pending, actor(updatedBy),
	).Scan(&item.ID, &item.NamespaceID, &item.EnvironmentID, &item.Key, &item.Value, &item.ReleasedValue, &item.PendingDelete, &item.Comment, &item.UpdatedBy, &item.UpdatedAt)
	if err != nil {
		return domain.ConfigItem{}, classify(err)
	}
	if err := tx.QueryRow(ctx, `SELECT app_id FROM namespaces WHERE id=$1`, item.NamespaceID).Scan(&item.AppID); err != nil {
		return domain.ConfigItem{}, err
	}
	if err := insertAudit(ctx, tx, item.UpdatedBy, "update", "config", id, map[string]bool{"pendingDelete": pending}); err != nil {
		return domain.ConfigItem{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return domain.ConfigItem{}, err
	}
	return item, nil
}

func (s *Store) Publish(ctx context.Context, request domain.PublishRequest) (domain.Release, error) {
	tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted})
	if err != nil {
		return domain.Release{}, err
	}
	defer tx.Rollback(ctx) //nolint:errcheck
	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, advisoryKey(request.NamespaceID, request.EnvironmentID)); err != nil {
		return domain.Release{}, err
	}
	appCode, namespaceName, envCode, err := loadScope(ctx, tx, request.AppID, request.NamespaceID, request.EnvironmentID)
	if err != nil {
		return domain.Release{}, classify(err)
	}
	items, err := listConfigTx(ctx, tx, request.AppID, request.NamespaceID, request.EnvironmentID, true)
	if err != nil {
		return domain.Release{}, err
	}
	snapshot := make(map[string]string)
	added, modified, removed := 0, 0, 0
	for _, item := range items {
		if item.PendingDelete {
			removed++
			continue
		}
		snapshot[item.Key] = item.Value
		if item.ReleasedValue == nil {
			added++
		} else if *item.ReleasedValue != item.Value {
			modified++
		}
	}
	if added+modified+removed == 0 {
		return domain.Release{}, storepkg.ErrNoPendingChange
	}
	version, err := nextVersion(ctx, tx, request.NamespaceID, request.EnvironmentID)
	if err != nil {
		return domain.Release{}, err
	}
	release, err := insertRelease(ctx, tx, request.AppID, request.NamespaceID, request.EnvironmentID, version, snapshot, added, modified, removed, request.Comment, actor(request.Operator))
	if err != nil {
		return domain.Release{}, err
	}
	key := fmt.Sprintf("/config/%s/%s/%s", strings.ToLower(envCode), appCode, namespaceName)
	if _, err := tx.Exec(ctx, `INSERT INTO release_outbox(release_id, etcd_key, payload) VALUES($1, $2, $3)`, release.ID, key, snapshot); err != nil {
		return domain.Release{}, err
	}
	if _, err := tx.Exec(ctx, `DELETE FROM config_items WHERE namespace_id=$1 AND env_id=$2 AND pending_delete=true`, request.NamespaceID, request.EnvironmentID); err != nil {
		return domain.Release{}, err
	}
	if _, err := tx.Exec(ctx, `UPDATE config_items SET released_value=value, pending_delete=false WHERE namespace_id=$1 AND env_id=$2`, request.NamespaceID, request.EnvironmentID); err != nil {
		return domain.Release{}, err
	}
	if err := insertAudit(ctx, tx, release.Operator, "publish", "release", release.ID, release); err != nil {
		return domain.Release{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return domain.Release{}, err
	}
	return release, nil
}

func (s *Store) Rollback(ctx context.Context, request domain.RollbackRequest) (domain.Release, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return domain.Release{}, err
	}
	defer tx.Rollback(ctx) //nolint:errcheck
	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, advisoryKey(request.NamespaceID, request.EnvironmentID)); err != nil {
		return domain.Release{}, err
	}
	appCode, namespaceName, envCode, err := loadScope(ctx, tx, request.AppID, request.NamespaceID, request.EnvironmentID)
	if err != nil {
		return domain.Release{}, classify(err)
	}
	var targetSnapshot, currentSnapshot map[string]string
	var targetVersion int
	err = tx.QueryRow(ctx, `SELECT version, snapshot FROM releases WHERE namespace_id=$1 AND env_id=$2 AND version=$3`, request.NamespaceID, request.EnvironmentID, request.TargetVersion).Scan(&targetVersion, &targetSnapshot)
	if err != nil {
		return domain.Release{}, classify(err)
	}
	var latestVersion int
	err = tx.QueryRow(ctx, `SELECT version, snapshot FROM releases WHERE namespace_id=$1 AND env_id=$2 ORDER BY version DESC LIMIT 1`, request.NamespaceID, request.EnvironmentID).Scan(&latestVersion, &currentSnapshot)
	if err != nil {
		return domain.Release{}, classify(err)
	}
	if latestVersion == targetVersion {
		return domain.Release{}, storepkg.ErrInvalidRollback
	}
	added, modified, removed := diff(currentSnapshot, targetSnapshot)
	if _, err := tx.Exec(ctx, `DELETE FROM config_items WHERE namespace_id=$1 AND env_id=$2`, request.NamespaceID, request.EnvironmentID); err != nil {
		return domain.Release{}, err
	}
	for key, value := range targetSnapshot {
		if _, err := tx.Exec(ctx, `INSERT INTO config_items(namespace_id, env_id, key, value, released_value, updated_by) VALUES($1,$2,$3,$4,$4,$5)`, request.NamespaceID, request.EnvironmentID, key, value, actor(request.Operator)); err != nil {
			return domain.Release{}, err
		}
	}
	version := latestVersion + 1
	comment := fmt.Sprintf("回滚至版本 v%d", targetVersion)
	release, err := insertRelease(ctx, tx, request.AppID, request.NamespaceID, request.EnvironmentID, version, targetSnapshot, added, modified, removed, comment, actor(request.Operator))
	if err != nil {
		return domain.Release{}, err
	}
	key := fmt.Sprintf("/config/%s/%s/%s", strings.ToLower(envCode), appCode, namespaceName)
	if _, err := tx.Exec(ctx, `INSERT INTO release_outbox(release_id, etcd_key, payload) VALUES($1,$2,$3)`, release.ID, key, targetSnapshot); err != nil {
		return domain.Release{}, err
	}
	if err := insertAudit(ctx, tx, release.Operator, "rollback", "release", release.ID, map[string]int{"targetVersion": targetVersion, "newVersion": version}); err != nil {
		return domain.Release{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return domain.Release{}, err
	}
	return release, nil
}

func (s *Store) ListReleases(ctx context.Context, appID, namespaceID, envID int64) ([]domain.Release, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT r.id, n.app_id, r.namespace_id, r.env_id, r.version, r.snapshot, r.diff_added, r.diff_modified,
		       r.diff_removed, r.comment, r.operator, r.etcd_revision, r.status, r.created_at
		FROM releases r JOIN namespaces n ON n.id=r.namespace_id
		WHERE n.app_id=$1 AND r.namespace_id=$2 AND r.env_id=$3 ORDER BY r.version DESC`, appID, namespaceID, envID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]domain.Release, 0)
	for rows.Next() {
		item, err := scanRelease(rows)
		if err != nil {
			return nil, err
		}
		result = append(result, item)
	}
	return result, rows.Err()
}

func (s *Store) GetRelease(ctx context.Context, id int64) (domain.Release, error) {
	row := s.pool.QueryRow(ctx, `
		SELECT r.id, n.app_id, r.namespace_id, r.env_id, r.version, r.snapshot, r.diff_added, r.diff_modified,
		       r.diff_removed, r.comment, r.operator, r.etcd_revision, r.status, r.created_at
		FROM releases r JOIN namespaces n ON n.id=r.namespace_id WHERE r.id=$1`, id)
	item, err := scanRelease(row)
	return item, classify(err)
}

func (s *Store) ListAuditLogs(ctx context.Context, limit int) ([]domain.AuditLog, error) {
	if limit <= 0 || limit > 500 {
		limit = 100
	}
	rows, err := s.pool.Query(ctx, `SELECT id, actor, action, target_type, target_id, detail, created_at FROM audit_logs ORDER BY created_at DESC LIMIT $1`, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]domain.AuditLog, 0)
	for rows.Next() {
		var item domain.AuditLog
		if err := rows.Scan(&item.ID, &item.Actor, &item.Action, &item.TargetType, &item.TargetID, &item.Detail, &item.CreatedAt); err != nil {
			return nil, err
		}
		result = append(result, item)
	}
	return result, rows.Err()
}

func (s *Store) EnsureBootstrapAdmin(ctx context.Context, username, passwordHash, displayName string) error {
	_, err := s.pool.Exec(ctx, `
		INSERT INTO users(username, password_hash, display_name, is_admin)
		VALUES($1,$2,$3,true) ON CONFLICT(username) DO NOTHING`, username, passwordHash, displayName)
	return classify(err)
}

func (s *Store) FindUserByUsername(ctx context.Context, username string) (domain.UserCredential, error) {
	var item domain.UserCredential
	err := s.pool.QueryRow(ctx, `
		SELECT id, username, password_hash, display_name, is_admin, disabled, created_at
		FROM users WHERE lower(username)=lower($1)`, username).
		Scan(&item.ID, &item.Username, &item.PasswordHash, &item.DisplayName, &item.IsAdmin, &item.Disabled, &item.CreatedAt)
	return item, classify(err)
}

func (s *Store) ListUsers(ctx context.Context) ([]domain.User, error) {
	rows, err := s.pool.Query(ctx, `SELECT id, username, display_name, is_admin, disabled, created_at FROM users ORDER BY username`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]domain.User, 0)
	for rows.Next() {
		var item domain.User
		if err := rows.Scan(&item.ID, &item.Username, &item.DisplayName, &item.IsAdmin, &item.Disabled, &item.CreatedAt); err != nil {
			return nil, err
		}
		result = append(result, item)
	}
	return result, rows.Err()
}

func (s *Store) CreateUser(ctx context.Context, input domain.User, passwordHash string) (domain.User, error) {
	err := s.pool.QueryRow(ctx, `
		INSERT INTO users(username, password_hash, display_name, is_admin, disabled)
		VALUES($1,$2,$3,$4,$5)
		RETURNING id, username, display_name, is_admin, disabled, created_at`,
		input.Username, passwordHash, input.DisplayName, input.IsAdmin, input.Disabled,
	).Scan(&input.ID, &input.Username, &input.DisplayName, &input.IsAdmin, &input.Disabled, &input.CreatedAt)
	if err != nil {
		return domain.User{}, classify(err)
	}
	s.audit(ctx, actor(input.UpdatedBy), "create", "user", input.ID, input)
	return input, nil
}

func (s *Store) SetUserAppRole(ctx context.Context, input domain.UserAppRole) error {
	command, err := s.pool.Exec(ctx, `
		INSERT INTO user_app_roles(user_id, app_id, role_id)
		SELECT $1, $2, r.id FROM roles r WHERE r.name=$3
		ON CONFLICT(user_id, app_id) DO UPDATE SET role_id=EXCLUDED.role_id`, input.UserID, input.AppID, input.Role)
	if err != nil {
		return classify(err)
	}
	if command.RowsAffected() == 0 {
		return storepkg.ErrNotFound
	}
	s.audit(ctx, actor(input.UpdatedBy), "update", "user_app_role", input.UserID, input)
	return nil
}

func (s *Store) DeleteUserAppRole(ctx context.Context, userID, appID int64, updatedBy string) error {
	command, err := s.pool.Exec(ctx, `DELETE FROM user_app_roles WHERE user_id=$1 AND app_id=$2`, userID, appID)
	if err != nil {
		return err
	}
	if command.RowsAffected() == 0 {
		return storepkg.ErrNotFound
	}
	s.audit(ctx, actor(updatedBy), "delete", "user_app_role", userID, map[string]int64{"appId": appID})
	return nil
}

func (s *Store) GetUserAppRole(ctx context.Context, userID, appID int64) (string, error) {
	var role string
	err := s.pool.QueryRow(ctx, `
		SELECT r.name FROM user_app_roles uar JOIN roles r ON r.id=uar.role_id
		WHERE uar.user_id=$1 AND uar.app_id=$2`, userID, appID).Scan(&role)
	return role, classify(err)
}

func (s *Store) ListUserAppRoles(ctx context.Context, userID int64) ([]domain.UserAppRole, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT uar.user_id, uar.app_id, r.name
		FROM user_app_roles uar JOIN roles r ON r.id=uar.role_id
		WHERE uar.user_id=$1 ORDER BY uar.app_id`, userID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]domain.UserAppRole, 0)
	for rows.Next() {
		var item domain.UserAppRole
		if err := rows.Scan(&item.UserID, &item.AppID, &item.Role); err != nil {
			return nil, err
		}
		result = append(result, item)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	if len(result) == 0 {
		var exists bool
		if err := s.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM users WHERE id=$1)`, userID).Scan(&exists); err != nil {
			return nil, err
		}
		if !exists {
			return nil, storepkg.ErrNotFound
		}
	}
	return result, nil
}

func (s *Store) ResolveScope(ctx context.Context, envCode, appCode, namespaceName string) (int64, int64, int64, error) {
	var appID, namespaceID, envID int64
	err := s.pool.QueryRow(ctx, `
		SELECT a.id, n.id, e.id
		FROM applications a JOIN namespaces n ON n.app_id=a.id CROSS JOIN environments e
		WHERE a.app_code=$1 AND n.name=$2 AND lower(e.env_code)=lower($3)`, appCode, namespaceName, envCode).
		Scan(&appID, &namespaceID, &envID)
	return appID, namespaceID, envID, classify(err)
}

func (s *Store) ListGrayRules(ctx context.Context, appID, namespaceID, envID int64) ([]domain.GrayRule, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT g.id, n.app_id, g.namespace_id, g.env_id, g.rule_type, g.rule_value, g.overrides,
		       g.enabled, g.priority, g.description, g.created_at
		FROM gray_rules g JOIN namespaces n ON n.id=g.namespace_id
		WHERE n.app_id=$1 AND g.namespace_id=$2 AND g.env_id=$3
		ORDER BY g.priority, g.id`, appID, namespaceID, envID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]domain.GrayRule, 0)
	for rows.Next() {
		item, err := scanGrayRule(rows)
		if err != nil {
			return nil, err
		}
		result = append(result, item)
	}
	return result, rows.Err()
}

func (s *Store) GetGrayRule(ctx context.Context, id int64) (domain.GrayRule, error) {
	row := s.pool.QueryRow(ctx, `
		SELECT g.id, n.app_id, g.namespace_id, g.env_id, g.rule_type, g.rule_value, g.overrides,
		       g.enabled, g.priority, g.description, g.created_at
		FROM gray_rules g JOIN namespaces n ON n.id=g.namespace_id WHERE g.id=$1`, id)
	item, err := scanGrayRule(row)
	return item, classify(err)
}

func (s *Store) CreateGrayRule(ctx context.Context, input domain.GrayRule, updatedBy string) (domain.GrayRule, error) {
	err := s.pool.QueryRow(ctx, `
		INSERT INTO gray_rules(namespace_id, env_id, rule_type, rule_value, overrides, enabled, priority, description)
		SELECT n.id, $3, $4, $5, $6, $7, $8, $9 FROM namespaces n
		WHERE n.id=$2 AND n.app_id=$1
		RETURNING id, namespace_id, env_id, rule_type, rule_value, overrides, enabled, priority, description, created_at`,
		input.AppID, input.NamespaceID, input.EnvironmentID, input.RuleType, input.RuleValue, input.Overrides, input.Enabled, input.Priority, input.Desc,
	).Scan(&input.ID, &input.NamespaceID, &input.EnvironmentID, &input.RuleType, &input.RuleValue, &input.Overrides, &input.Enabled, &input.Priority, &input.Desc, &input.CreatedAt)
	if err != nil {
		return domain.GrayRule{}, classify(err)
	}
	s.audit(ctx, actor(updatedBy), "create", "gray_rule", input.ID, input)
	return input, nil
}

func (s *Store) UpdateGrayRule(ctx context.Context, id int64, input domain.GrayRule, updatedBy string) (domain.GrayRule, error) {
	err := s.pool.QueryRow(ctx, `
		UPDATE gray_rules SET rule_type=$2, rule_value=$3, overrides=$4, enabled=$5, priority=$6, description=$7
		WHERE id=$1
		RETURNING id, namespace_id, env_id, rule_type, rule_value, overrides, enabled, priority, description, created_at`,
		id, input.RuleType, input.RuleValue, input.Overrides, input.Enabled, input.Priority, input.Desc,
	).Scan(&input.ID, &input.NamespaceID, &input.EnvironmentID, &input.RuleType, &input.RuleValue, &input.Overrides, &input.Enabled, &input.Priority, &input.Desc, &input.CreatedAt)
	if err != nil {
		return domain.GrayRule{}, classify(err)
	}
	if err := s.pool.QueryRow(ctx, `SELECT app_id FROM namespaces WHERE id=$1`, input.NamespaceID).Scan(&input.AppID); err != nil {
		return domain.GrayRule{}, classify(err)
	}
	s.audit(ctx, actor(updatedBy), "update", "gray_rule", id, input)
	return input, nil
}

func (s *Store) DeleteGrayRule(ctx context.Context, id int64, updatedBy string) error {
	command, err := s.pool.Exec(ctx, `DELETE FROM gray_rules WHERE id=$1`, id)
	if err != nil {
		return err
	}
	if command.RowsAffected() == 0 {
		return storepkg.ErrNotFound
	}
	s.audit(ctx, actor(updatedBy), "delete", "gray_rule", id, nil)
	return nil
}

func (s *Store) ClaimOutbox(ctx context.Context, limit int) ([]domain.OutboxEntry, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx) //nolint:errcheck
	rows, err := tx.Query(ctx, `
		SELECT id, release_id, etcd_key, payload, retry_count
		FROM release_outbox
		WHERE (status='pending' AND next_attempt_at <= now())
		   OR (status='processing' AND locked_at < now() - interval '60 seconds')
		ORDER BY id FOR UPDATE SKIP LOCKED LIMIT $1`, limit)
	if err != nil {
		return nil, err
	}
	entries := make([]domain.OutboxEntry, 0)
	ids := make([]int64, 0)
	for rows.Next() {
		var entry domain.OutboxEntry
		if err := rows.Scan(&entry.ID, &entry.ReleaseID, &entry.EtcdKey, &entry.Payload, &entry.RetryCount); err != nil {
			rows.Close()
			return nil, err
		}
		entries, ids = append(entries, entry), append(ids, entry.ID)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return nil, err
	}
	for _, id := range ids {
		if _, err := tx.Exec(ctx, `UPDATE release_outbox SET status='processing', locked_at=now() WHERE id=$1`, id); err != nil {
			return nil, err
		}
	}
	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}
	return entries, nil
}

func (s *Store) MarkOutboxDone(ctx context.Context, id, releaseID, revision int64) error {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback(ctx) //nolint:errcheck
	if _, err := tx.Exec(ctx, `UPDATE release_outbox SET status='done', locked_at=NULL WHERE id=$1`, id); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `UPDATE releases SET status='applied', etcd_revision=$2 WHERE id=$1`, releaseID, revision); err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (s *Store) MarkOutboxFailed(ctx context.Context, id, releaseID int64, message string, maxRetry int) error {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback(ctx) //nolint:errcheck
	var status string
	err = tx.QueryRow(ctx, `
		UPDATE release_outbox
		SET retry_count=retry_count+1,
		    status=CASE WHEN retry_count+1 >= $3 THEN 'failed' ELSE 'pending' END,
		    last_error=$2,
		    next_attempt_at=now() + make_interval(secs => power(2, LEAST(retry_count+1, 6))::int),
		    locked_at=NULL
		WHERE id=$1 RETURNING status`, id, message, maxRetry).Scan(&status)
	if err != nil {
		return classify(err)
	}
	if status == "failed" {
		if _, err := tx.Exec(ctx, `UPDATE releases SET status='failed' WHERE id=$1`, releaseID); err != nil {
			return err
		}
	}
	return tx.Commit(ctx)
}

func (s *Store) OutboxStats(ctx context.Context) (domain.OutboxStats, error) {
	var result domain.OutboxStats
	err := s.pool.QueryRow(ctx, `
		SELECT COUNT(*) FILTER (WHERE status='pending'),
		       COUNT(*) FILTER (WHERE status='processing'),
		       COUNT(*) FILTER (WHERE status='failed')
		FROM release_outbox`).Scan(&result.Pending, &result.Processing, &result.Failed)
	return result, err
}

type rowScanner interface {
	Scan(...any) error
}

func scanConfig(row rowScanner) (domain.ConfigItem, error) {
	var item domain.ConfigItem
	err := row.Scan(&item.ID, &item.AppID, &item.NamespaceID, &item.EnvironmentID, &item.Key, &item.Value, &item.ReleasedValue, &item.PendingDelete, &item.Comment, &item.UpdatedBy, &item.UpdatedAt)
	return item, err
}

func scanRelease(row rowScanner) (domain.Release, error) {
	var item domain.Release
	err := row.Scan(&item.ID, &item.AppID, &item.NamespaceID, &item.EnvironmentID, &item.Version, &item.Snapshot, &item.Added, &item.Modified, &item.Removed, &item.Comment, &item.Operator, &item.EtcdRevision, &item.Status, &item.Time)
	return item, err
}

func scanGrayRule(row rowScanner) (domain.GrayRule, error) {
	var item domain.GrayRule
	err := row.Scan(&item.ID, &item.AppID, &item.NamespaceID, &item.EnvironmentID, &item.RuleType, &item.RuleValue, &item.Overrides, &item.Enabled, &item.Priority, &item.Desc, &item.CreatedAt)
	return item, err
}

func listConfigTx(ctx context.Context, tx pgx.Tx, appID, namespaceID, envID int64, lock bool) ([]domain.ConfigItem, error) {
	query := `
		SELECT ci.id, n.app_id, ci.namespace_id, ci.env_id, ci.key, ci.value, ci.released_value,
		       ci.pending_delete, ci.comment, ci.updated_by, ci.updated_at
		FROM config_items ci JOIN namespaces n ON n.id=ci.namespace_id
		WHERE n.app_id=$1 AND ci.namespace_id=$2 AND ci.env_id=$3 ORDER BY ci.key`
	if lock {
		query += ` FOR UPDATE OF ci`
	}
	rows, err := tx.Query(ctx, query, appID, namespaceID, envID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := make([]domain.ConfigItem, 0)
	for rows.Next() {
		item, err := scanConfig(rows)
		if err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func loadScope(ctx context.Context, tx pgx.Tx, appID, namespaceID, envID int64) (string, string, string, error) {
	var appCode, namespaceName, envCode string
	err := tx.QueryRow(ctx, `
		SELECT a.app_code, n.name, e.env_code
		FROM applications a JOIN namespaces n ON n.app_id=a.id CROSS JOIN environments e
		WHERE a.id=$1 AND n.id=$2 AND e.id=$3`, appID, namespaceID, envID).Scan(&appCode, &namespaceName, &envCode)
	return appCode, namespaceName, envCode, err
}

func nextVersion(ctx context.Context, tx pgx.Tx, namespaceID, envID int64) (int, error) {
	var version int
	err := tx.QueryRow(ctx, `SELECT COALESCE(MAX(version),0)+1 FROM releases WHERE namespace_id=$1 AND env_id=$2`, namespaceID, envID).Scan(&version)
	return version, err
}

func insertRelease(ctx context.Context, tx pgx.Tx, appID, namespaceID, envID int64, version int, snapshot map[string]string, added, modified, removed int, comment, operator string) (domain.Release, error) {
	item := domain.Release{AppID: appID, NamespaceID: namespaceID, EnvironmentID: envID, Version: version, Snapshot: snapshot, Added: added, Modified: modified, Removed: removed, Comment: comment, Operator: operator}
	err := tx.QueryRow(ctx, `
		INSERT INTO releases(namespace_id, env_id, version, snapshot, diff_added, diff_modified, diff_removed, comment, operator)
		VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)
		RETURNING id, status, created_at`, namespaceID, envID, version, snapshot, added, modified, removed, comment, operator,
	).Scan(&item.ID, &item.Status, &item.Time)
	return item, err
}

func (s *Store) deleteAndAudit(ctx context.Context, table, targetType string, id int64, updatedBy string) error {
	if table != "applications" && table != "environments" && table != "namespaces" {
		return errors.New("unsupported delete target")
	}
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback(ctx)                                                             //nolint:errcheck
	command, err := tx.Exec(ctx, fmt.Sprintf("DELETE FROM %s WHERE id=$1", table), id) // #nosec G201 -- table is allow-listed above.
	if err != nil {
		return classify(err)
	}
	if command.RowsAffected() == 0 {
		return storepkg.ErrNotFound
	}
	if err := insertAudit(ctx, tx, actor(updatedBy), "delete", targetType, id, nil); err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (s *Store) audit(ctx context.Context, updatedBy, action, targetType string, targetID int64, detail any) {
	_, _ = s.pool.Exec(ctx, `INSERT INTO audit_logs(actor, action, target_type, target_id, detail) VALUES($1,$2,$3,$4,$5)`, actor(updatedBy), action, targetType, targetID, toJSON(detail))
}

func insertAudit(ctx context.Context, tx pgx.Tx, updatedBy, action, targetType string, targetID int64, detail any) error {
	_, err := tx.Exec(ctx, `INSERT INTO audit_logs(actor, action, target_type, target_id, detail) VALUES($1,$2,$3,$4,$5)`, actor(updatedBy), action, targetType, targetID, toJSON(detail))
	return err
}

func toJSON(value any) json.RawMessage {
	payload, _ := json.Marshal(value)
	return payload
}

func classify(err error) error {
	if err == nil {
		return nil
	}
	if errors.Is(err, pgx.ErrNoRows) {
		return storepkg.ErrNotFound
	}
	var pgErr *pgconn.PgError
	if errors.As(err, &pgErr) {
		switch pgErr.Code {
		case "23505":
			return storepkg.ErrConflict
		case "23503", "23514", "23502":
			return fmt.Errorf("invalid data: %s", pgErr.Message)
		}
	}
	return err
}

func actor(value string) string {
	if strings.TrimSpace(value) == "" {
		return "admin"
	}
	return strings.TrimSpace(value)
}

func advisoryKey(namespaceID, envID int64) int64 {
	return namespaceID*1_000_000_000 + envID
}

func diff(current, target map[string]string) (added, modified, removed int) {
	for key, value := range target {
		old, exists := current[key]
		if !exists {
			added++
		} else if old != value {
			modified++
		}
	}
	for key := range current {
		if _, exists := target[key]; !exists {
			removed++
		}
	}
	return
}

var _ = time.Second