274 lines
11 KiB
Go
274 lines
11 KiB
Go
package repository
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.pcux.cn/xuans/Scert/backend/internal/deployment"
|
|
)
|
|
|
|
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)
|
|
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)
|
|
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()
|
|
}
|
|
result, err := r.db.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 = ? WHERE 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.ID)
|
|
if err != nil {
|
|
return fmt.Errorf("更新服务器失败：%w", err)
|
|
}
|
|
count, err := result.RowsAffected()
|
|
if err != nil {
|
|
return fmt.Errorf("读取服务器更新结果失败：%w", err)
|
|
}
|
|
if count == 0 {
|
|
return sql.ErrNoRows
|
|
}
|
|
return nil
|
|
}
|
|
|
|
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 {
|
|
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)
|
|
}
|
|
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 domain_id = ?"), domainID); err != nil {
|
|
return rollback(fmt.Errorf("清理域名服务器关联失败：%w", err))
|
|
}
|
|
for _, serverID := range cleaned {
|
|
var exists int
|
|
if err := tx.QueryRowContext(ctx, r.query("SELECT COUNT(*) FROM servers WHERE id = ?"), serverID).Scan(&exists); err != nil {
|
|
return rollback(fmt.Errorf("校验部署服务器失败：%w", err))
|
|
}
|
|
if exists == 0 {
|
|
return rollback(fmt.Errorf("部署服务器不存在：%s", serverID))
|
|
}
|
|
if _, err := tx.ExecContext(ctx, r.query("INSERT INTO domain_servers (domain_id, server_id) VALUES (?, ?)"), domainID, serverID); 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 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 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); 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
|
|
}
|