xuans/Scert
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

暂存 PostgreSQL 数据库兼容实现

Browse Source
This commit is contained in:
xuans 2026-09-16 13:14:50 +08:00
parent 959d14eb2f
commit eaf474c8e7
14 changed files with 326 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
backend/internal/database/database.go
View File

@ -4,11 +4,14 @@ import (
"context"
"database/sql"
"fmt"
"net/url"
"path/filepath"
"strings"
"time"
"github.com/go-sql-driver/mysql"
"github.com/jackc/pgx/v5"
_ "github.com/jackc/pgx/v5/stdlib"
_ "modernc.org/sqlite"
"git.pcux.cn/xuans/Scert/backend/internal/config"
@ -19,25 +22,37 @@ type Database struct {
}
func Open(ctx context.Context, cfg config.DatabaseConfig) (*Database, error) {
if cfg.Driver != "sqlite" && cfg.Driver != "mysql" {
driver := strings.ToLower(strings.TrimSpace(cfg.Driver))
if driver != "sqlite" && driver != "mysql" && driver != "postgres" {
return nil, fmt.Errorf("数据库驱动无效：%s", cfg.Driver)
}
dsn := cfg.DSN
if cfg.Driver == "mysql" {
dsn := strings.TrimSpace(cfg.DSN)
if driver == "mysql" {
var err error
dsn, err = normalizeMySQLDSN(dsn)
if err != nil {
return nil, err
}
}
if cfg.Driver == "sqlite" {
if driver == "postgres" {
var err error
dsn, err = normalizePostgresDSN(dsn)
if err != nil {
return nil, err
}
}
if driver == "sqlite" {
if cfg.DSN != ":memory:" && len(cfg.DSN) > 0 && cfg.DSN != "file::memory:?cache=shared" {
if err := ensureSQLiteDir(cfg.DSN); err != nil {
return nil, err
}
}
}
db, err := sql.Open(cfg.Driver, dsn)
sqlDriver := driver
if driver == "postgres" {
sqlDriver = "pgx"
}
db, err := sql.Open(sqlDriver, dsn)
if err != nil {
return nil, fmt.Errorf("打开数据库失败：%w", err)
}
@ -50,7 +65,7 @@ func Open(ctx context.Context, cfg config.DatabaseConfig) (*Database, error) {
_ = db.Close()
return nil, fmt.Errorf("连接数据库失败：%w", err)
}
if cfg.Driver == "sqlite" {
if driver == "sqlite" {
if err := configureSQLite(pingCtx, db); err != nil {
_ = db.Close()
return nil, err
@ -86,9 +101,42 @@ func normalizeMySQLDSN(dsn string) (string, error) {
return cfg.FormatDSN(), nil
}
func normalizePostgresDSN(dsn string) (string, error) {
parsed, err := pgx.ParseConfig(dsn)
if err != nil {
return "", fmt.Errorf("解析 PostgreSQL 连接信息失败：%w", err)
}
if strings.TrimSpace(parsed.Host) == "" || strings.TrimSpace(parsed.Database) == "" || strings.TrimSpace(parsed.User) == "" {
return "", fmt.Errorf("PostgreSQL 连接信息必须包含主机、数据库和用户名")
}
sslMode := postgresSSLMode(dsn)
if sslMode != "verify-full" {
return "", fmt.Errorf("PostgreSQL 连接必须使用 sslmode=verify-full")
}
if parsed.TLSConfig == nil || parsed.TLSConfig.InsecureSkipVerify {
return "", fmt.Errorf("PostgreSQL TLS 校验配置无效")
}
return dsn, nil
}
func postgresSSLMode(dsn string) string {
dsn = strings.TrimSpace(dsn)
if parsed, err := url.Parse(dsn); err == nil && (parsed.Scheme == "postgres" || parsed.Scheme == "postgresql") {
return strings.ToLower(strings.TrimSpace(parsed.Query().Get("sslmode")))
}
for _, field := range strings.Fields(dsn) {
key, value, found := strings.Cut(field, "=")
if found && strings.EqualFold(strings.TrimSpace(key), "sslmode") {
return strings.ToLower(strings.Trim(strings.TrimSpace(value), "'\""))
}
}
return ""
}
// SanitizeDSN 返回可写入日志的数据库连接信息，不包含 MySQL 密码。
func SanitizeDSN(driver, dsn string) string {
if strings.EqualFold(strings.TrimSpace(driver), "mysql") {
switch strings.ToLower(strings.TrimSpace(driver)) {
case "mysql":
cfg, err := mysql.ParseDSN(dsn)
if err != nil {
return "mysql://[连接信息无效]"
@ -97,10 +145,33 @@ func SanitizeDSN(driver, dsn string) string {
cfg.Passwd = "********"
}
return cfg.FormatDSN()
case "postgres":
parsed, err := url.Parse(dsn)
if err == nil && (parsed.Scheme == "postgres" || parsed.Scheme == "postgresql") {
if parsed.User != nil {
username := parsed.User.Username()
if _, hasPassword := parsed.User.Password(); hasPassword {
parsed.User = url.UserPassword(username, "********")
}
}
return parsed.String()
}
return sanitizePostgresKeywordDSN(dsn)
}
return dsn
}
func sanitizePostgresKeywordDSN(dsn string) string {
fields := strings.Fields(dsn)
for index, field := range fields {
key, _, found := strings.Cut(field, "=")
if found && strings.EqualFold(strings.TrimSpace(key), "password") {
fields[index] = key + "=********"
}
}
return strings.Join(fields, " ")
}
func ensureSQLiteDir(dsn string) error {
path := dsn
if strings.HasPrefix(path, "file:") {

16
backend/internal/database/migrate.go
View File

@ -9,14 +9,14 @@ import (
"strings"
)
//go:embed migrations/sqlite/0001_initial.sql migrations/sqlite/0002_domain_options.sql migrations/sqlite/0003_deployment.sql migrations/sqlite/0004_task_attempts.sql migrations/sqlite/0005_users.sql migrations/sqlite/0006_server_validation.sql migrations/mysql/0001_initial.sql migrations/mysql/0002_domain_options.sql migrations/mysql/0003_deployment.sql migrations/mysql/0004_task_attempts.sql migrations/mysql/0005_users.sql migrations/mysql/0006_server_validation.sql
//go:embed migrations/sqlite/0001_initial.sql migrations/sqlite/0002_domain_options.sql migrations/sqlite/0003_deployment.sql migrations/sqlite/0004_task_attempts.sql migrations/sqlite/0005_users.sql migrations/sqlite/0006_server_validation.sql migrations/mysql/0001_initial.sql migrations/mysql/0002_domain_options.sql migrations/mysql/0003_deployment.sql migrations/mysql/0004_task_attempts.sql migrations/mysql/0005_users.sql migrations/mysql/0006_server_validation.sql migrations/postgres/0001_initial.sql migrations/postgres/0002_domain_options.sql migrations/postgres/0003_deployment.sql migrations/postgres/0004_task_attempts.sql migrations/postgres/0005_users.sql migrations/postgres/0006_server_validation.sql
var migrationFiles embed.FS
func Migrate(ctx context.Context, db *sql.DB, driver string) error {
if db == nil {
return fmt.Errorf("数据库连接不能为空")
}
if driver != "sqlite" && driver != "mysql" {
if driver != "sqlite" && driver != "mysql" && driver != "postgres" {
return fmt.Errorf("数据库驱动无效：%s", driver)
}
if _, err := db.ExecContext(ctx, migrationTableSQL(driver)); err != nil {
@ -75,7 +75,11 @@ func applyMigration(ctx context.Context, db *sql.DB, driver string, version int)
return fmt.Errorf("执行数据库迁移失败：%w", err)
}
}
if _, err := tx.ExecContext(ctx, "INSERT INTO schema_migrations (version) VALUES (?)", version); err != nil {
migrationVersionInsert := "INSERT INTO schema_migrations (version) VALUES (?)"
if driver == "postgres" {
migrationVersionInsert = "INSERT INTO schema_migrations (version) VALUES ($1)"
}
if _, err := tx.ExecContext(ctx, migrationVersionInsert, version); err != nil {
_ = tx.Rollback()
return fmt.Errorf("记录迁移版本失败：%w", err)
}
@ -150,6 +154,12 @@ func migrationTableSQL(driver string) string {
applied_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
)`
}
if driver == "postgres" {
return `CREATE TABLE IF NOT EXISTS schema_migrations (
version INTEGER NOT NULL PRIMARY KEY,
applied_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
)`
}
return `CREATE TABLE IF NOT EXISTS schema_migrations (
version INTEGER NOT NULL PRIMARY KEY,
applied_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP

41
backend/internal/database/migrations/postgres/0001_initial.sql Normal file
View File

@ -0,0 +1,41 @@
CREATE TABLE IF NOT EXISTS settings (
key TEXT NOT NULL PRIMARY KEY,
value TEXT NOT NULL,
updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS domains (
id TEXT NOT NULL PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
certificate_type TEXT NOT NULL,
sans_json TEXT NOT NULL,
status TEXT NOT NULL,
expires_at TIMESTAMPTZ NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS tasks (
id TEXT NOT NULL PRIMARY KEY,
domain_id TEXT NOT NULL,
task_type TEXT NOT NULL,
status TEXT NOT NULL,
message TEXT NOT NULL,
started_at TIMESTAMPTZ NULL,
finished_at TIMESTAMPTZ NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_tasks_domain_id ON tasks(domain_id);
CREATE TABLE IF NOT EXISTS audit_logs (
id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
action TEXT NOT NULL,
resource_type TEXT NOT NULL,
resource_id TEXT NOT NULL,
status TEXT NOT NULL,
message TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_audit_logs_created_at ON audit_logs(created_at);

3
backend/internal/database/migrations/postgres/0002_domain_options.sql Normal file
View File

@ -0,0 +1,3 @@
ALTER TABLE domains ADD COLUMN auto_renew BOOLEAN NOT NULL DEFAULT TRUE;
ALTER TABLE domains ADD COLUMN acme_environment TEXT NOT NULL DEFAULT 'staging';
ALTER TABLE domains ADD COLUMN dns_provider TEXT NOT NULL DEFAULT '';

27
backend/internal/database/migrations/postgres/0003_deployment.sql Normal file
View File

@ -0,0 +1,27 @@
CREATE TABLE IF NOT EXISTS servers (
id TEXT NOT NULL PRIMARY KEY,
name TEXT NOT NULL,
address TEXT NOT NULL,
port INTEGER NOT NULL,
username TEXT NOT NULL,
private_key_ref TEXT NOT NULL,
host_key_fingerprint TEXT NOT NULL,
certificate_path TEXT NOT NULL,
private_key_path TEXT NOT NULL,
certificate_chain_path TEXT NOT NULL,
web_server TEXT NOT NULL,
reload_command_json TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_servers_name ON servers(name);
CREATE TABLE IF NOT EXISTS domain_servers (
domain_id TEXT NOT NULL,
server_id TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (domain_id, server_id)
);
CREATE INDEX IF NOT EXISTS idx_domain_servers_server_id ON domain_servers(server_id);

20
backend/internal/database/migrations/postgres/0004_task_attempts.sql Normal file
View File

@ -0,0 +1,20 @@
ALTER TABLE tasks ADD COLUMN stage TEXT NOT NULL DEFAULT 'queued';
ALTER TABLE tasks ADD COLUMN operator_id TEXT NOT NULL DEFAULT '';
ALTER TABLE tasks ADD COLUMN certificate_version_id TEXT NOT NULL DEFAULT '';
ALTER TABLE tasks ADD COLUMN idempotency_key TEXT NOT NULL DEFAULT '';
ALTER TABLE tasks ADD COLUMN updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP;
CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status);
CREATE INDEX IF NOT EXISTS idx_tasks_idempotency ON tasks(idempotency_key);
CREATE TABLE IF NOT EXISTS task_attempts (
id TEXT NOT NULL PRIMARY KEY,
task_id TEXT NOT NULL,
attempt_number INTEGER NOT NULL,
stage TEXT NOT NULL,
status TEXT NOT NULL,
message TEXT NOT NULL,
started_at TIMESTAMPTZ NULL,
finished_at TIMESTAMPTZ NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE (task_id, attempt_number)
);
CREATE INDEX IF NOT EXISTS idx_task_attempts_task_id ON task_attempts(task_id);

12
backend/internal/database/migrations/postgres/0005_users.sql Normal file
View File

@ -0,0 +1,12 @@
CREATE TABLE IF NOT EXISTS users (
id TEXT NOT NULL PRIMARY KEY,
username TEXT NOT NULL UNIQUE,
email TEXT NULL UNIQUE,
password_hash TEXT NOT NULL,
role TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'active',
created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_users_role_status ON users(role, status);

1
backend/internal/database/migrations/postgres/0006_server_validation.sql Normal file
View File

@ -0,0 +1 @@
ALTER TABLE servers ADD COLUMN validation_command_json TEXT NOT NULL DEFAULT '[]';

2
backend/internal/repository/audit_reader.go
View File

@ -18,7 +18,7 @@ func (r *SQLAuditRepository) ListAuditLogs(ctx context.Context, limit int) ([]Au
if limit <= 0 || limit > 200 {
limit = 50
}
rows, err := r.db.QueryContext(ctx, `SELECT action, resource_type, resource_id, status, message, created_at FROM audit_logs ORDER BY created_at DESC, id DESC LIMIT `+fmt.Sprintf("%d", limit))
rows, err := r.db.QueryContext(ctx, r.query(`SELECT action, resource_type, resource_id, status, message, created_at FROM audit_logs ORDER BY created_at DESC, id DESC LIMIT `+fmt.Sprintf("%d", limit)))
if err != nil {
return nil, fmt.Errorf("查询审计日志失败：%w", err)
}

17
backend/internal/repository/audit_repository.go
View File

@ -23,11 +23,16 @@ type AuditRepository interface {
}
type SQLAuditRepository struct {
db *sql.DB
db *sql.DB
driver string
}
func NewSQLAuditRepository(db *sql.DB) *SQLAuditRepository {
return &SQLAuditRepository{db: db}
func NewSQLAuditRepository(db *sql.DB, driver ...string) *SQLAuditRepository {
return &SQLAuditRepository{db: db, driver: repositoryDriver(driver)}
}
func (r *SQLAuditRepository) query(value string) string {
return Rebind(r.driver, value)
}
func (r *SQLAuditRepository) CreateAuditLog(ctx context.Context, item AuditLog) error {
@ -44,9 +49,9 @@ func (r *SQLAuditRepository) CreateAuditLog(ctx context.Context, item AuditLog)
if item.CreatedAt.IsZero() {
item.CreatedAt = time.Now().UTC()
}
_, err := r.db.ExecContext(ctx, `INSERT INTO audit_logs
_, err := r.db.ExecContext(ctx, r.query(`INSERT INTO audit_logs
(action, resource_type, resource_id, status, message, created_at)
VALUES (?, ?, ?, ?, ?, ?)`, item.Action, item.ResourceType, item.ResourceID, item.Status, item.Message, item.CreatedAt.UTC())
VALUES (?, ?, ?, ?, ?, ?)`), item.Action, item.ResourceType, item.ResourceID, item.Status, item.Message, item.CreatedAt.UTC())
if err != nil {
return fmt.Errorf("保存审计日志失败：%w", err)
}
@ -60,7 +65,7 @@ func (r *SQLAuditRepository) PruneExpiredAuditLogs(ctx context.Context, before t
if before.IsZero() {
return 0, errors.New("审计日志清理时间不能为空")
}
result, err := r.db.ExecContext(ctx, `DELETE FROM audit_logs WHERE created_at < ?`, before.UTC())
result, err := r.db.ExecContext(ctx, r.query(`DELETE FROM audit_logs WHERE created_at < ?`), before.UTC())
if err != nil {
return 0, fmt.Errorf("清理过期审计日志失败：%w", err)
}

65
backend/internal/repository/rebind.go Normal file
View File

@ -0,0 +1,65 @@
package repository
import (
"strconv"
"strings"
)
// Rebind 将内部统一使用的参数占位符转换为目标数据库驱动所需格式。
// 它只处理 SQL 字符串字面量和标识符之外的问号，避免改写业务数据。
func Rebind(driver, query string) string {
if strings.ToLower(strings.TrimSpace(driver)) != "postgres" || !strings.Contains(query, "?") {
return query
}
var result strings.Builder
result.Grow(len(query) + 8)
parameter := 0
inSingleQuote := false
inDoubleQuote := false
inBacktick := false
for index := 0; index < len(query); index++ {
char := query[index]
switch char {
case '\'':
if !inDoubleQuote && !inBacktick {
if inSingleQuote && index+1 < len(query) && query[index+1] == '\'' {
result.WriteByte(char)
index++
result.WriteByte(query[index])
continue
}
inSingleQuote = !inSingleQuote
}
case '"':
if !inSingleQuote && !inBacktick {
inDoubleQuote = !inDoubleQuote
}
case '`':
if !inSingleQuote && !inDoubleQuote {
inBacktick = !inBacktick
}
case '?':
if !inSingleQuote && !inDoubleQuote && !inBacktick {
parameter++
result.WriteByte('$')
result.WriteString(strconv.Itoa(parameter))
continue
}
}
result.WriteByte(char)
}
return result.String()
}
func repositoryDriver(driver []string) string {
if len(driver) == 0 {
return "sqlite"
}
value := strings.ToLower(strings.TrimSpace(driver[0]))
if value == "" {
return "sqlite"
}
return value
}

35
backend/internal/repository/server_repository.go
View File

@ -24,11 +24,16 @@ type ServerRepository interface {
}
type SQLServerRepository struct {
db *sql.DB
db *sql.DB
driver string
}
func NewSQLServerRepository(db *sql.DB) *SQLServerRepository {
return &SQLServerRepository{db: db}
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 {
@ -52,9 +57,9 @@ func (r *SQLServerRepository) CreateServer(ctx context.Context, server deploymen
if updatedAt.IsZero() {
updatedAt = createdAt
}
_, err = r.db.ExecContext(ctx, `INSERT INTO servers
_, err = r.db.ExecContext(ctx, r.query(`INSERT INTO servers
(id, name, address, port, username, private_key_ref, host_key_fingerprint, 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.PrivateKeyRef,
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`), server.ID, server.Name, server.Address, server.Port, server.Username, server.PrivateKeyRef,
server.HostKeyFingerprint, server.CertificatePath, server.PrivateKeyPath, server.CertificateChainPath, server.WebServer, string(validationCommand), string(command), createdAt, updatedAt)
if err != nil {
return fmt.Errorf("保存服务器失败：%w", err)
@ -79,7 +84,7 @@ func (r *SQLServerRepository) UpdateServer(ctx context.Context, server deploymen
if updatedAt.IsZero() {
updatedAt = time.Now().UTC()
}
result, err := r.db.ExecContext(ctx, `UPDATE servers SET name = ?, address = ?, port = ?, username = ?, private_key_ref = ?, host_key_fingerprint = ?, certificate_path = ?, private_key_path = ?, certificate_chain_path = ?, web_server = ?, validation_command_json = ?, reload_command_json = ?, updated_at = ? WHERE id = ?`,
result, err := r.db.ExecContext(ctx, r.query(`UPDATE servers SET name = ?, address = ?, port = ?, username = ?, private_key_ref = ?, host_key_fingerprint = ?, 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.PrivateKeyRef, server.HostKeyFingerprint, server.CertificatePath, server.PrivateKeyPath, server.CertificateChainPath, server.WebServer, string(validationCommand), string(command), updatedAt, server.ID)
if err != nil {
return fmt.Errorf("更新服务器失败：%w", err)
@ -95,12 +100,12 @@ func (r *SQLServerRepository) UpdateServer(ctx context.Context, server deploymen
}
func (r *SQLServerRepository) GetServerByID(ctx context.Context, id string) (deployment.Server, error) {
row := r.db.QueryRowContext(ctx, serverSelect+` WHERE id = ?`, strings.TrimSpace(id))
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, serverSelect+` ORDER BY name, id`)
rows, err := r.db.QueryContext(ctx, r.query(serverSelect+` ORDER BY name, id`))
if err != nil {
return nil, fmt.Errorf("查询服务器失败：%w", err)
}
@ -132,10 +137,10 @@ func (r *SQLServerRepository) DeleteServer(ctx context.Context, id string) error
_ = tx.Rollback()
return cause
}
if _, err := tx.ExecContext(ctx, "DELETE FROM domain_servers WHERE server_id = ?", id); err != nil {
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, "DELETE FROM servers WHERE id = ?", id)
result, err := tx.ExecContext(ctx, r.query("DELETE FROM servers WHERE id = ?"), id)
if err != nil {
return rollback(fmt.Errorf("删除服务器失败：%w", err))
}
@ -158,14 +163,14 @@ func (r *SQLServerRepository) IsPrivateKeyRefUsed(ctx context.Context, reference
return false, errors.New("SSH 私钥引用不能为空")
}
var count int
if err := r.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM servers WHERE private_key_ref = ?", reference).Scan(&count); err != nil {
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, 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))
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)
}
@ -209,18 +214,18 @@ func (r *SQLServerRepository) ReplaceDomainServers(ctx context.Context, domainID
_ = tx.Rollback()
return cause
}
if _, err := tx.ExecContext(ctx, "DELETE FROM domain_servers WHERE domain_id = ?", domainID); err != nil {
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, "SELECT COUNT(*) FROM servers WHERE id = ?", serverID).Scan(&exists); err != nil {
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, "INSERT INTO domain_servers (domain_id, server_id) VALUES (?, ?)", domainID, serverID); err != nil {
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))
}
}

33
backend/internal/repository/sql_domain_repository.go
View File

@ -13,11 +13,16 @@ import (
)
type SQLDomainRepository struct {
db *sql.DB
db *sql.DB
driver string
}
func NewSQLDomainRepository(db *sql.DB) *SQLDomainRepository {
return &SQLDomainRepository{db: db}
func NewSQLDomainRepository(db *sql.DB, driver ...string) *SQLDomainRepository {
return &SQLDomainRepository{db: db, driver: repositoryDriver(driver)}
}
func (r *SQLDomainRepository) query(value string) string {
return Rebind(r.driver, value)
}
func (r *SQLDomainRepository) Create(ctx context.Context, item domain.Domain) error {
@ -28,9 +33,9 @@ func (r *SQLDomainRepository) Create(ctx context.Context, item domain.Domain) er
if err != nil {
return fmt.Errorf("编码 SAN 域名失败：%w", err)
}
_, err = r.db.ExecContext(ctx, `INSERT INTO domains
_, err = r.db.ExecContext(ctx, r.query(`INSERT INTO domains
(id, name, certificate_type, sans_json, auto_renew, acme_environment, dns_provider, status, expires_at, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, item.ID, item.Name, item.CertificateType, string(payload), item.AutoRenew,
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`), item.ID, item.Name, item.CertificateType, string(payload), item.AutoRenew,
item.ACMEEnvironment, item.DNSProvider, item.Status, nullableTime(item.ExpiresAt), item.CreatedAt, item.UpdatedAt)
if err != nil {
return fmt.Errorf("保存域名失败：%w", err)
@ -46,9 +51,9 @@ func (r *SQLDomainRepository) Update(ctx context.Context, item domain.Domain) er
if err != nil {
return fmt.Errorf("编码 SAN 域名失败：%w", err)
}
result, err := r.db.ExecContext(ctx, `UPDATE domains SET
result, err := r.db.ExecContext(ctx, r.query(`UPDATE domains SET
name = ?, certificate_type = ?, sans_json = ?, auto_renew = ?, acme_environment = ?, dns_provider = ?, updated_at = ?
WHERE id = ?`, item.Name, item.CertificateType, string(payload), item.AutoRenew, item.ACMEEnvironment,
WHERE id = ?`), item.Name, item.CertificateType, string(payload), item.AutoRenew, item.ACMEEnvironment,
item.DNSProvider, item.UpdatedAt, item.ID)
if err != nil {
return fmt.Errorf("更新域名失败：%w", err)
@ -67,8 +72,8 @@ func (r *SQLDomainRepository) GetByID(ctx context.Context, id string) (domain.Do
if r == nil || r.db == nil {
return domain.Domain{}, errors.New("数据库连接不能为空")
}
row := r.db.QueryRowContext(ctx, `SELECT id, name, certificate_type, sans_json, auto_renew, acme_environment, dns_provider, status, expires_at, created_at, updated_at
FROM domains WHERE id = ?`, id)
row := r.db.QueryRowContext(ctx, r.query(`SELECT id, name, certificate_type, sans_json, auto_renew, acme_environment, dns_provider, status, expires_at, created_at, updated_at
FROM domains WHERE id = ?`), id)
return scanDomain(row)
}
@ -76,8 +81,8 @@ func (r *SQLDomainRepository) List(ctx context.Context) ([]domain.Domain, error)
if r == nil || r.db == nil {
return nil, errors.New("数据库连接不能为空")
}
rows, err := r.db.QueryContext(ctx, `SELECT id, name, certificate_type, sans_json, auto_renew, acme_environment, dns_provider, status, expires_at, created_at, updated_at
FROM domains ORDER BY name`)
rows, err := r.db.QueryContext(ctx, r.query(`SELECT id, name, certificate_type, sans_json, auto_renew, acme_environment, dns_provider, status, expires_at, created_at, updated_at
FROM domains ORDER BY name`))
if err != nil {
return nil, fmt.Errorf("查询域名失败：%w", err)
}
@ -108,10 +113,10 @@ func (r *SQLDomainRepository) Delete(ctx context.Context, id string) error {
_ = tx.Rollback()
return cause
}
if _, err := tx.ExecContext(ctx, "DELETE FROM domain_servers WHERE domain_id = ?", id); err != nil {
if _, err := tx.ExecContext(ctx, r.query("DELETE FROM domain_servers WHERE domain_id = ?"), id); err != nil {
return rollback(fmt.Errorf("删除域名服务器关联失败：%w", err))
}
result, err := tx.ExecContext(ctx, "DELETE FROM domains WHERE id = ?", id)
result, err := tx.ExecContext(ctx, r.query("DELETE FROM domains WHERE id = ?"), id)
if err != nil {
return rollback(fmt.Errorf("删除域名失败：%w", err))
}
@ -136,7 +141,7 @@ func (r *SQLDomainRepository) UpdateCertificateState(ctx context.Context, id, st
if status == "" {
return errors.New("证书状态不能为空")
}
result, err := r.db.ExecContext(ctx, `UPDATE domains SET status = ?, expires_at = ?, updated_at = ? WHERE id = ?`,
result, err := r.db.ExecContext(ctx, r.query(`UPDATE domains SET status = ?, expires_at = ?, updated_at = ? WHERE id = ?`),
status, nullableTime(expiresAt), time.Now().UTC(), id)
if err != nil {
return fmt.Errorf("更新证书状态失败：%w", err)

26
backend/internal/repository/user_repository.go
View File

@ -85,7 +85,7 @@ func (r *SQLUserRepository) FindByIdentifier(ctx context.Context, identifier str
if err != nil {
return User{}, err
}
row := r.db.QueryRowContext(ctx, userSelect+` WHERE LOWER(username) = ? OR (email IS NOT NULL AND LOWER(email) = ?) LIMIT 1`, normalized, normalized)
row := r.db.QueryRowContext(ctx, r.query(userSelect+` WHERE LOWER(username) = ? OR (email IS NOT NULL AND LOWER(email) = ?) LIMIT 1`), normalized, normalized)
return scanUser(row)
}
@ -93,7 +93,7 @@ func (r *SQLUserRepository) ListUsers(ctx context.Context) ([]User, error) {
if err := r.ensureDB(); err != nil {
return nil, err
}
rows, err := r.db.QueryContext(ctx, userSelect+` ORDER BY role, username, id`)
rows, err := r.db.QueryContext(ctx, r.query(userSelect+` ORDER BY role, username, id`))
if err != nil {
return nil, fmt.Errorf("查询用户失败：%w", err)
}
@ -120,7 +120,7 @@ func (r *SQLUserRepository) GetUserByID(ctx context.Context, id string) (User, e
if id == "" {
return User{}, errors.New("用户 ID 不能为空")
}
return scanUser(r.db.QueryRowContext(ctx, userSelect+` WHERE id = ?`, id))
return scanUser(r.db.QueryRowContext(ctx, r.query(userSelect+` WHERE id = ?`), id))
}
func (r *SQLUserRepository) CreateUser(ctx context.Context, item User) error {
@ -137,9 +137,9 @@ func (r *SQLUserRepository) CreateUser(ctx context.Context, item User) error {
if item.UpdatedAt.IsZero() {
item.UpdatedAt = item.CreatedAt
}
_, err := r.db.ExecContext(ctx, `INSERT INTO users
_, err := r.db.ExecContext(ctx, r.query(`INSERT INTO users
(id, username, email, password_hash, role, status, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, item.ID, item.Username, r.emailValue(item.Email), item.PasswordHash,
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`), item.ID, item.Username, r.emailValue(item.Email), item.PasswordHash,
item.Role, item.Status, item.CreatedAt, item.UpdatedAt)
if err != nil {
return fmt.Errorf("保存用户失败：%w", err)
@ -155,7 +155,7 @@ func (r *SQLUserRepository) UpdateUser(ctx context.Context, item User) error {
if err := validateUser(item, false); err != nil {
return err
}
result, err := r.db.ExecContext(ctx, `UPDATE users SET username = ?, email = ?, password_hash = ?, role = ?, status = ?, updated_at = ? WHERE id = ?`,
result, err := r.db.ExecContext(ctx, r.query(`UPDATE users SET username = ?, email = ?, password_hash = ?, role = ?, status = ?, updated_at = ? WHERE id = ?`),
item.Username, r.emailValue(item.Email), item.PasswordHash, item.Role, item.Status, time.Now().UTC(), item.ID)
if err != nil {
return fmt.Errorf("更新用户失败：%w", err)
@ -181,7 +181,7 @@ func (r *SQLUserRepository) DeleteUser(ctx context.Context, id string) error {
if id == "primary-super-admin" {
return ErrSuperAdminProtected
}
result, err := r.db.ExecContext(ctx, `DELETE FROM users WHERE id = ?`, id)
result, err := r.db.ExecContext(ctx, r.query(`DELETE FROM users WHERE id = ?`), id)
if err != nil {
return fmt.Errorf("删除用户失败：%w", err)
}
@ -208,7 +208,7 @@ func (r *SQLUserRepository) upsert(ctx context.Context, item User) error {
item.CreatedAt = now
}
item.UpdatedAt = now
result, err := r.db.ExecContext(ctx, `UPDATE users SET username = ?, email = ?, password_hash = ?, role = ?, status = ?, updated_at = ? WHERE id = ?`,
result, err := r.db.ExecContext(ctx, r.query(`UPDATE users SET username = ?, email = ?, password_hash = ?, role = ?, status = ?, updated_at = ? WHERE id = ?`),
item.Username, r.emailValue(item.Email), item.PasswordHash, item.Role, item.Status, item.UpdatedAt, item.ID)
if err != nil {
return err
@ -220,20 +220,24 @@ func (r *SQLUserRepository) upsert(ctx context.Context, item User) error {
if count > 0 {
return nil
}
_, err = r.db.ExecContext(ctx, `INSERT INTO users
_, err = r.db.ExecContext(ctx, r.query(`INSERT INTO users
(id, username, email, password_hash, role, status, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, item.ID, item.Username, r.emailValue(item.Email), item.PasswordHash,
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`), item.ID, item.Username, r.emailValue(item.Email), item.PasswordHash,
item.Role, item.Status, item.CreatedAt, item.UpdatedAt)
return err
}
func (r *SQLUserRepository) emailValue(email string) any {
if email == "" && r.driver == "mysql" {
if email == "" && (r.driver == "mysql" || r.driver == "postgres") {
return nil
}
return email
}
func (r *SQLUserRepository) query(value string) string {
return Rebind(r.driver, value)
}
func normalizeUser(item User) User {
item.ID = strings.TrimSpace(item.ID)
item.Username = strings.TrimSpace(strings.ToLower(item.Username))