Scert/backend/internal/repository/server_repository.go

package repository

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"git.pcux.cn/xuans/Scert/backend/internal/deployment"
	"git.pcux.cn/xuans/Scert/backend/internal/domain"
)

type ServerRepository interface {
	CreateServer(ctx context.Context, server deployment.Server) error
	UpdateServer(ctx context.Context, server deployment.Server) error
	GetServerByID(ctx context.Context, id string) (deployment.Server, error)
	ListServers(ctx context.Context) ([]deployment.Server, error)
	DeleteServer(ctx context.Context, id string) error
	IsPrivateKeyRefUsed(ctx context.Context, reference string) (bool, error)
	ListServersByDomainID(ctx context.Context, domainID string) ([]deployment.Server, error)
	ReplaceDomainServers(ctx context.Context, domainID string, serverIDs []string) error
}

type SQLServerRepository struct {
	db     *sql.DB
	driver string
}

func NewSQLServerRepository(db *sql.DB, driver ...string) *SQLServerRepository {
	return &SQLServerRepository{db: db, driver: repositoryDriver(driver)}
}

func (r *SQLServerRepository) query(value string) string {
	return Rebind(r.driver, value)
}

func (r *SQLServerRepository) CreateServer(ctx context.Context, server deployment.Server) error {
	server = server.Normalize()
	if err := validateServer(server); err != nil {
		return err
	}
	command, err := json.Marshal(server.ReloadCommand)
	if err != nil {
		return fmt.Errorf("编码 reload 命令失败：%w", err)
	}
	validationCommand, err := json.Marshal(server.ValidationCommand)
	if err != nil {
		return fmt.Errorf("编码配置校验命令失败：%w", err)
	}
	createdAt := server.CreatedAt
	if createdAt.IsZero() {
		createdAt = time.Now().UTC()
	}
	updatedAt := server.UpdatedAt
	if updatedAt.IsZero() {
		updatedAt = createdAt
	}
	_, err = r.db.ExecContext(ctx, r.query(`INSERT INTO servers
		(id, name, address, port, username, auth_method, private_key_ref, password_ref, host_key_fingerprint, ssh_key_status, ssh_key_fingerprint, ssh_key_deployed_at, ssh_key_tested_at, ssh_key_error, certificate_path, private_key_path, certificate_chain_path, web_server, validation_command_json, reload_command_json, created_at, updated_at, project_id, owner_id, created_by, updated_by)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`), server.ID, server.Name, server.Address, server.Port, server.Username, server.AuthMethod, server.PrivateKeyRef, server.PasswordRef,
		server.HostKeyFingerprint, server.SSHKeyStatus, server.SSHKeyFingerprint, server.SSHKeyDeployedAt, server.SSHKeyTestedAt, server.SSHKeyError, server.CertificatePath, server.PrivateKeyPath, server.CertificateChainPath, server.WebServer, string(validationCommand), string(command), createdAt, updatedAt,
		server.ProjectID, server.OwnerID, server.CreatedBy, server.UpdatedBy)
	if err != nil {
		return fmt.Errorf("保存服务器失败：%w", err)
	}
	return nil
}

func (r *SQLServerRepository) UpdateServer(ctx context.Context, server deployment.Server) error {
	server = server.Normalize()
	if err := validateServer(server); err != nil {
		return err
	}
	command, err := json.Marshal(server.ReloadCommand)
	if err != nil {
		return fmt.Errorf("编码 reload 命令失败：%w", err)
	}
	validationCommand, err := json.Marshal(server.ValidationCommand)
	if err != nil {
		return fmt.Errorf("编码配置校验命令失败：%w", err)
	}
	updatedAt := server.UpdatedAt
	if updatedAt.IsZero() {
		updatedAt = time.Now().UTC()
	}
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	var projectID string
	if err := tx.QueryRowContext(ctx, r.query(`SELECT project_id FROM servers WHERE id = ?`+serverResourceLock(r.driver)), server.ID).Scan(&projectID); err != nil {
		return err
	}
	if projectID != server.ProjectID {
		return sql.ErrNoRows
	}
	if err := validateServerDomainPaths(ctx, tx, r.driver, server, nil); err != nil {
		return err
	}
	_, err = tx.ExecContext(ctx, r.query(`UPDATE servers SET name = ?, address = ?, port = ?, username = ?, auth_method = ?, private_key_ref = ?, password_ref = ?, host_key_fingerprint = ?, ssh_key_status = ?, ssh_key_fingerprint = ?, ssh_key_deployed_at = ?, ssh_key_tested_at = ?, ssh_key_error = ?, certificate_path = ?, private_key_path = ?, certificate_chain_path = ?, web_server = ?, validation_command_json = ?, reload_command_json = ?, updated_at = ?, updated_by = ? WHERE id = ? AND project_id = ?`),
		server.Name, server.Address, server.Port, server.Username, server.AuthMethod, server.PrivateKeyRef, server.PasswordRef, server.HostKeyFingerprint, server.SSHKeyStatus, server.SSHKeyFingerprint, server.SSHKeyDeployedAt, server.SSHKeyTestedAt, server.SSHKeyError, server.CertificatePath, server.PrivateKeyPath, server.CertificateChainPath, server.WebServer, string(validationCommand), string(command), updatedAt, server.UpdatedBy, server.ID, server.ProjectID)
	if err != nil {
		return fmt.Errorf("更新服务器失败：%w", err)
	}
	return tx.Commit()
}

func (r *SQLServerRepository) GetServerByID(ctx context.Context, id string) (deployment.Server, error) {
	row := r.db.QueryRowContext(ctx, r.query(serverSelect+` WHERE id = ?`), strings.TrimSpace(id))
	return scanServer(row)
}

func (r *SQLServerRepository) ListServers(ctx context.Context) ([]deployment.Server, error) {
	rows, err := r.db.QueryContext(ctx, r.query(serverSelect+` ORDER BY name, id`))
	if err != nil {
		return nil, fmt.Errorf("查询服务器失败：%w", err)
	}
	defer rows.Close()
	result := make([]deployment.Server, 0)
	for rows.Next() {
		server, err := scanServer(rows)
		if err != nil {
			return nil, err
		}
		result = append(result, server)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("读取服务器失败：%w", err)
	}
	return result, nil
}

func (r *SQLServerRepository) DeleteServer(ctx context.Context, id string) error {
	id = strings.TrimSpace(id)
	if id == "" {
		return errors.New("服务器 ID 不能为空")
	}
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("开始删除服务器失败：%w", err)
	}
	rollback := func(cause error) error {
		_ = tx.Rollback()
		return cause
	}
	if _, err := tx.ExecContext(ctx, r.query("DELETE FROM domain_servers WHERE server_id = ?"), id); err != nil {
		return rollback(fmt.Errorf("删除服务器关联失败：%w", err))
	}
	result, err := tx.ExecContext(ctx, r.query("DELETE FROM servers WHERE id = ?"), id)
	if err != nil {
		return rollback(fmt.Errorf("删除服务器失败：%w", err))
	}
	count, err := result.RowsAffected()
	if err != nil {
		return rollback(fmt.Errorf("读取服务器删除结果失败：%w", err))
	}
	if count == 0 {
		return rollback(sql.ErrNoRows)
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("提交服务器删除失败：%w", err)
	}
	return nil
}

func (r *SQLServerRepository) IsPrivateKeyRefUsed(ctx context.Context, reference string) (bool, error) {
	reference = strings.TrimSpace(reference)
	if reference == "" {
		return false, errors.New("SSH 私钥引用不能为空")
	}
	var count int
	if err := r.db.QueryRowContext(ctx, r.query("SELECT COUNT(*) FROM servers WHERE private_key_ref = ?"), reference).Scan(&count); err != nil {
		return false, fmt.Errorf("检查 SSH 私钥引用占用状态失败：%w", err)
	}
	return count > 0, nil
}

func (r *SQLServerRepository) ListServersByDomainID(ctx context.Context, domainID string) ([]deployment.Server, error) {
	rows, err := r.db.QueryContext(ctx, r.query(serverSelectQualified+` INNER JOIN domain_servers ds ON ds.server_id = servers.id WHERE ds.domain_id = ? ORDER BY servers.name, servers.id`), strings.TrimSpace(domainID))
	if err != nil {
		return nil, fmt.Errorf("查询域名部署服务器失败：%w", err)
	}
	defer rows.Close()
	result := make([]deployment.Server, 0)
	for rows.Next() {
		server, err := scanServer(rows)
		if err != nil {
			return nil, err
		}
		result = append(result, server)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("读取域名部署服务器失败：%w", err)
	}
	return result, nil
}

func (r *SQLServerRepository) ReplaceDomainServers(ctx context.Context, domainID string, serverIDs []string) error {
	domainID = strings.TrimSpace(domainID)
	if strings.TrimSpace(domainID) == "" {
		return errors.New("域名 ID 不能为空")
	}
	cleaned := make([]string, 0, len(serverIDs))
	seen := make(map[string]struct{}, len(serverIDs))
	for _, id := range serverIDs {
		id = strings.TrimSpace(id)
		if id == "" {
			return errors.New("服务器 ID 不能为空")
		}
		if _, ok := seen[id]; ok {
			continue
		}
		seen[id] = struct{}{}
		cleaned = append(cleaned, id)
	}
	sort.Strings(cleaned)
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("开始更新域名服务器关联失败：%w", err)
	}
	defer tx.Rollback()
	rollback := func(cause error) error {
		_ = tx.Rollback()
		return cause
	}
	var candidate domain.Domain
	if err := tx.QueryRowContext(ctx, r.query(`SELECT id, project_id, name, certificate_type FROM domains WHERE id = ?`+serverResourceLock(r.driver)), domainID).Scan(&candidate.ID, &candidate.ProjectID, &candidate.Name, &candidate.CertificateType); err != nil {
		return rollback(fmt.Errorf("读取绑定域名项目失败：%w", err))
	}
	projectID := candidate.ProjectID
	if _, err := tx.ExecContext(ctx, r.query("DELETE FROM domain_servers WHERE domain_id = ?"), domainID); err != nil {
		return rollback(fmt.Errorf("清理域名服务器关联失败：%w", err))
	}
	for _, serverID := range cleaned {
		server, err := scanServer(tx.QueryRowContext(ctx, r.query(serverSelectQualified+` WHERE servers.id = ? AND servers.project_id = ?`+serverResourceLock(r.driver)), serverID, projectID))
		if err != nil {
			return rollback(fmt.Errorf("校验部署服务器失败：%w", err))
		}
		if err := validateServerDomainPaths(ctx, tx, r.driver, server, &candidate); err != nil {
			return rollback(err)
		}
		if _, err := tx.ExecContext(ctx, r.query("INSERT INTO domain_servers (domain_id, server_id, project_id) VALUES (?, ?, ?)"), domainID, serverID, projectID); err != nil {
			return rollback(fmt.Errorf("保存域名服务器关联失败：%w", err))
		}
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("提交域名服务器关联失败：%w", err)
	}
	return nil
}

const serverSelect = `SELECT id, name, address, port, username, auth_method, private_key_ref, password_ref, host_key_fingerprint, ssh_key_status, ssh_key_fingerprint, ssh_key_deployed_at, ssh_key_tested_at, ssh_key_error, certificate_path, private_key_path, certificate_chain_path, web_server, validation_command_json, reload_command_json, created_at, updated_at, project_id, owner_id, created_by, updated_by FROM servers`
const serverSelectQualified = `SELECT servers.id, servers.name, servers.address, servers.port, servers.username, servers.auth_method, servers.private_key_ref, servers.password_ref, servers.host_key_fingerprint, servers.ssh_key_status, servers.ssh_key_fingerprint, servers.ssh_key_deployed_at, servers.ssh_key_tested_at, servers.ssh_key_error, servers.certificate_path, servers.private_key_path, servers.certificate_chain_path, servers.web_server, servers.validation_command_json, servers.reload_command_json, servers.created_at, servers.updated_at, servers.project_id, servers.owner_id, servers.created_by, servers.updated_by FROM servers`

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

func scanServer(row rowScanner) (deployment.Server, error) {
	var server deployment.Server
	var webServer string
	var authMethod string
	var validationCommandJSON string
	var commandJSON string
	if err := row.Scan(&server.ID, &server.Name, &server.Address, &server.Port, &server.Username, &authMethod, &server.PrivateKeyRef, &server.PasswordRef, &server.HostKeyFingerprint, &server.SSHKeyStatus, &server.SSHKeyFingerprint, &server.SSHKeyDeployedAt, &server.SSHKeyTestedAt, &server.SSHKeyError, &server.CertificatePath, &server.PrivateKeyPath, &server.CertificateChainPath, &webServer, &validationCommandJSON, &commandJSON, &server.CreatedAt, &server.UpdatedAt, &server.ProjectID, &server.OwnerID, &server.CreatedBy, &server.UpdatedBy); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return deployment.Server{}, sql.ErrNoRows
		}
		return deployment.Server{}, fmt.Errorf("读取服务器失败：%w", err)
	}
	server.AuthMethod = deployment.SSHAuthMethod(authMethod)
	server.WebServer = deployment.WebServer(webServer)
	if err := json.Unmarshal([]byte(validationCommandJSON), &server.ValidationCommand); err != nil {
		return deployment.Server{}, fmt.Errorf("解析配置校验命令失败：%w", err)
	}
	if err := json.Unmarshal([]byte(commandJSON), &server.ReloadCommand); err != nil {
		return deployment.Server{}, fmt.Errorf("解析 reload 命令失败：%w", err)
	}
	return server.Normalize(), nil
}

func validateServer(server deployment.Server) error {
	if err := server.Validate(); err != nil {
		return err
	}
	return nil
}