From 86b67c557b7776bf2704c88304e0b3834b44bf17 Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Thu, 18 Jun 2026 00:01:26 +0800 Subject: [PATCH] =?UTF-8?q?feat(server/db):=20=E5=A4=9A=E5=BA=93(3/4)?= =?UTF-8?q?=E2=80=94=20Dialect=20=E6=8A=BD=E8=B1=A1(upsert/=E9=94=81)+=20?= =?UTF-8?q?=E5=90=88=E5=B9=B6=20directory=5Fversion?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - db.Dialect:LockForUpdate(mysql "FOR UPDATE" / sqlite "")、Upsert(中性 EXCLUDED.col → mysql VALUES()/ sqlite excluded.);DialectForDB 从连接驱动推导 - 9 处 ON DUPLICATE KEY、11 处 FOR UPDATE 全走 dialect;sqlite 靠 _txlock= immediate 取得 BEGIN IMMEDIATE 悲观锁等价语义 - directory_version 三处重复合并为 store.BumpDirectoryVersion(dialect 感知) - 这些文件同时含(2/4)的 UTC→Go 改动(与 upsert/锁同语句交错,无法拆分) - 顺带:usage 的 FIELD()、codes 的 DATE_ADD/GREATEST 续期、nodes 的 UNIX_TIMESTAMP、NULLIF 等 MySQL 专属构造一并退回 Go/可移植写法 Co-Authored-By: Claude Opus 4.8 --- server/internal/codes/store.go | 63 +++++++----- server/internal/db/dialect.go | 96 +++++++++++++++++++ server/internal/devices/store.go | 24 +++-- server/internal/nodes/lifecycle.go | 39 ++++---- server/internal/nodes/store.go | 44 +++++---- server/internal/provision/mysqlstore.go | 31 +++--- server/internal/scheduler/wiring_lifecycle.go | 18 ++-- server/internal/store/directory.go | 28 ++++++ server/internal/usage/store.go | 35 ++++--- 9 files changed, 265 insertions(+), 113 deletions(-) create mode 100644 server/internal/db/dialect.go create mode 100644 server/internal/store/directory.go diff --git a/server/internal/codes/store.go b/server/internal/codes/store.go index e4f24c3..e0feaa3 100644 --- a/server/internal/codes/store.go +++ b/server/internal/codes/store.go @@ -6,6 +6,8 @@ import ( "fmt" "strings" "time" + + dbx "github.com/wangjia/pangolin/server/internal/db" ) // PlanCode represents a plan tier. @@ -76,11 +78,12 @@ type BatchRow struct { // outside an explicit transaction; methods that accept a *sql.Tx run within // that transaction. type Store struct { - db *sql.DB + db *sql.DB + dialect dbx.Dialect } -// NewStore creates a Store backed by the given MySQL connection pool. -func NewStore(db *sql.DB) *Store { return &Store{db: db} } +// NewStore creates a Store backed by the given connection pool (MySQL or SQLite). +func NewStore(db *sql.DB) *Store { return &Store{db: db, dialect: dbx.DialectForDB(db)} } // -------------------------------------------------------------------------- // Batch and code creation (used by Service.CreateBatch and webhook) @@ -88,10 +91,14 @@ func NewStore(db *sql.DB) *Store { return &Store{db: db} } // CreateBatch inserts a new code_batches row and returns its ID. func (s *Store) CreateBatch(ctx context.Context, channel BatchChannel, createdBy, note string) (int64, error) { + var notePtr *string + if note != "" { + notePtr = ¬e + } res, err := s.db.ExecContext(ctx, `INSERT INTO code_batches (channel, created_by, note, created_at) - VALUES (?, ?, NULLIF(?, ''), UTC_TIMESTAMP(6))`, - string(channel), createdBy, note) + VALUES (?, ?, ?, ?)`, + string(channel), createdBy, notePtr, time.Now().UTC()) if err != nil { return 0, fmt.Errorf("store.CreateBatch: %w", err) } @@ -133,7 +140,7 @@ func (s *Store) FindCodeByHashForUpdate(ctx context.Context, tx *sql.Tx, hash st FROM codes c JOIN plans p ON p.id = c.plan_id WHERE c.code_hash = ? - FOR UPDATE`, + `+s.dialect.LockForUpdate(), hash) var cr CodeRow @@ -151,9 +158,9 @@ func (s *Store) FindCodeByHashForUpdate(ctx context.Context, tx *sql.Tx, hash st // MarkRedeemed updates a codes row to status='redeemed' within tx. func (s *Store) MarkRedeemed(ctx context.Context, tx *sql.Tx, codeID, userID int64) error { _, err := tx.ExecContext(ctx, - `UPDATE codes SET status='redeemed', redeemed_by=?, redeemed_at=UTC_TIMESTAMP(6) + `UPDATE codes SET status='redeemed', redeemed_by=?, redeemed_at=? WHERE id=? AND status='unused'`, - userID, codeID) + userID, time.Now().UTC(), codeID) if err != nil { return fmt.Errorf("store.MarkRedeemed: %w", err) } @@ -181,9 +188,9 @@ func (s *Store) GetActiveSubscriptions(ctx context.Context, tx *sql.Tx, userID i `SELECT s.id, s.user_id, s.plan_id, p.code, s.expires_at, s.source FROM subscriptions s JOIN plans p ON p.id = s.plan_id - WHERE s.user_id=? AND s.expires_at > UTC_TIMESTAMP(6) + WHERE s.user_id=? AND s.expires_at > ? ORDER BY s.expires_at DESC`, - userID) + userID, time.Now().UTC()) if err != nil { return nil, fmt.Errorf("store.GetActiveSubscriptions: %w", err) } @@ -201,15 +208,24 @@ func (s *Store) GetActiveSubscriptions(ctx context.Context, tx *sql.Tx, userID i } // ExtendSubscription sets expires_at to max(current_expires_at, now) + duration. -// Called inside the redeem transaction. +// Called inside the redeem transaction. The base date is computed in Go (the +// subscription row is read inside the locked redeem tx, then updated) — portable +// across engines and avoids DB-side DATE_ADD/GREATEST. func (s *Store) ExtendSubscription(ctx context.Context, tx *sql.Tx, subID int64, durationDays int) error { + var current time.Time + if err := tx.QueryRowContext(ctx, + `SELECT expires_at FROM subscriptions WHERE id=?`, subID).Scan(¤t); err != nil { + return fmt.Errorf("store.ExtendSubscription read: %w", err) + } + base := time.Now().UTC() + if current.After(base) { + base = current.UTC() + } + newExpiry := base.AddDate(0, 0, durationDays) + _, err := tx.ExecContext(ctx, - `UPDATE subscriptions - SET expires_at = DATE_ADD( - GREATEST(expires_at, UTC_TIMESTAMP(6)), - INTERVAL ? DAY) - WHERE id=?`, - durationDays, subID) + `UPDATE subscriptions SET expires_at = ? WHERE id=?`, + newExpiry, subID) if err != nil { return fmt.Errorf("store.ExtendSubscription: %w", err) } @@ -235,8 +251,8 @@ func (s *Store) CreateSubscription( res, err := tx.ExecContext(ctx, `INSERT INTO subscriptions (user_id, plan_id, expires_at, source, created_at) - VALUES (?, ?, ?, 'code', UTC_TIMESTAMP(6))`, - userID, planID, expiresAt) + VALUES (?, ?, ?, 'code', ?)`, + userID, planID, expiresAt, now) if err != nil { return 0, fmt.Errorf("store.CreateSubscription: %w", err) } @@ -255,14 +271,15 @@ func (s *Store) WriteAuditLog(ctx context.Context, tx *sql.Tx, actor, action, ta if metaJSON == "" { metaJSON = "null" } + now := time.Now().UTC() if tx != nil { _, err = tx.ExecContext(ctx, - `INSERT INTO audit_log (actor, action, target, meta, at) VALUES (?, ?, ?, ?, UTC_TIMESTAMP(6))`, - actor, action, target, metaJSON) + `INSERT INTO audit_log (actor, action, target, meta, at) VALUES (?, ?, ?, ?, ?)`, + actor, action, target, metaJSON, now) } else { _, err = s.db.ExecContext(ctx, - `INSERT INTO audit_log (actor, action, target, meta, at) VALUES (?, ?, ?, ?, UTC_TIMESTAMP(6))`, - actor, action, target, metaJSON) + `INSERT INTO audit_log (actor, action, target, meta, at) VALUES (?, ?, ?, ?, ?)`, + actor, action, target, metaJSON, now) } if err != nil { return fmt.Errorf("store.WriteAuditLog: %w", err) diff --git a/server/internal/db/dialect.go b/server/internal/db/dialect.go new file mode 100644 index 0000000..e16727c --- /dev/null +++ b/server/internal/db/dialect.go @@ -0,0 +1,96 @@ +package db + +import ( + "database/sql" + "fmt" + "regexp" + "strings" +) + +// Dialect captures the small set of SQL constructs that differ between MySQL +// and SQLite. The ~95% of the server's queries are plain CRUD with "?" +// placeholders and need no dialect handling; only upserts and row locks do. +type Dialect interface { + // Name reports "mysql" or "sqlite". + Name() string + + // LockForUpdate returns the pessimistic row-lock clause to append to a + // SELECT inside a transaction: + // mysql → "FOR UPDATE" + // sqlite → "" (single-writer engine; the BEGIN IMMEDIATE transaction — + // enabled via _txlock=immediate in the DSN — already serializes + // writers, giving equivalent exclusivity). + LockForUpdate() string + + // Upsert builds the conflict-resolution tail for an INSERT. conflictCols is + // the unique/primary key that triggers the conflict. setExprs are assignment + // expressions; refer to the would-be-inserted value with the sentinel + // "EXCLUDED." (rewritten to VALUES(col) on MySQL, kept as excluded.col + // on SQLite). With no setExprs the conflict is a no-op (insert-or-ignore). + // + // Example (atomic accumulate): + // d.Upsert([]string{"user_id","date"}, + // "bytes_up = bytes_up + EXCLUDED.bytes_up") + Upsert(conflictCols []string, setExprs ...string) string +} + +// DialectForDB derives the Dialect from the driver backing an open *sql.DB, so +// the dialect always matches the actual connection (no global state, and store +// constructors keep their (db *sql.DB) signatures). +func DialectForDB(database *sql.DB) Dialect { + if database == nil { + return MySQLDialect{} + } + name := strings.ToLower(fmt.Sprintf("%T", database.Driver())) + if strings.Contains(name, "sqlite") { + return SQLiteDialect{} + } + return MySQLDialect{} +} + +// DialectFor returns the Dialect for a driver name ("mysql"|"sqlite"). +func DialectFor(driver string) Dialect { + if Normalize(driver) == "sqlite" { + return SQLiteDialect{} + } + return MySQLDialect{} +} + +var excludedRef = regexp.MustCompile(`EXCLUDED\.(\w+)`) + +// MySQLDialect implements Dialect for MySQL. +type MySQLDialect struct{} + +func (MySQLDialect) Name() string { return "mysql" } +func (MySQLDialect) LockForUpdate() string { return "FOR UPDATE" } + +func (MySQLDialect) Upsert(conflictCols []string, setExprs ...string) string { + if len(setExprs) == 0 { + // No-op upsert (insert-or-ignore): assign a key column to itself. + col := "id" + if len(conflictCols) > 0 { + col = conflictCols[0] + } + return fmt.Sprintf("ON DUPLICATE KEY UPDATE %s = %s", col, col) + } + rewritten := make([]string, len(setExprs)) + for i, e := range setExprs { + rewritten[i] = excludedRef.ReplaceAllString(e, "VALUES($1)") + } + return "ON DUPLICATE KEY UPDATE " + strings.Join(rewritten, ", ") +} + +// SQLiteDialect implements Dialect for SQLite. +type SQLiteDialect struct{} + +func (SQLiteDialect) Name() string { return "sqlite" } +func (SQLiteDialect) LockForUpdate() string { return "" } + +func (SQLiteDialect) Upsert(conflictCols []string, setExprs ...string) string { + target := "ON CONFLICT(" + strings.Join(conflictCols, ", ") + ") " + if len(setExprs) == 0 { + return target + "DO NOTHING" + } + // SQLite accepts the EXCLUDED. reference verbatim. + return target + "DO UPDATE SET " + strings.Join(setExprs, ", ") +} diff --git a/server/internal/devices/store.go b/server/internal/devices/store.go index 2f6fc9a..a1d16e5 100644 --- a/server/internal/devices/store.go +++ b/server/internal/devices/store.go @@ -5,6 +5,8 @@ import ( "database/sql" "fmt" "time" + + dbx "github.com/wangjia/pangolin/server/internal/db" ) // DeviceRow mirrors a `devices` table row. @@ -33,11 +35,12 @@ type effSub struct { // Store wraps a *sql.DB and exposes the database operations the devices module // needs. Methods that take a *sql.Tx run within that transaction. type Store struct { - db *sql.DB + db *sql.DB + dialect dbx.Dialect } -// NewStore creates a Store backed by the given MySQL connection pool. -func NewStore(db *sql.DB) *Store { return &Store{db: db} } +// NewStore creates a Store backed by the given connection pool (MySQL or SQLite). +func NewStore(db *sql.DB) *Store { return &Store{db: db, dialect: dbx.DialectForDB(db)} } // BeginTx starts a transaction at Read Committed isolation. Per-user // serialization for device mutations is achieved by locking the users row @@ -77,7 +80,7 @@ func (s *Store) ListByUser(ctx context.Context, userID int64) ([]DeviceRow, erro func (s *Store) findDeviceByUUIDTx(ctx context.Context, tx *sql.Tx, uuid string) (*DeviceRow, error) { row := tx.QueryRowContext(ctx, `SELECT id, uuid, user_id, name, platform, last_seen, created_at - FROM devices WHERE uuid=? FOR UPDATE`, uuid) + FROM devices WHERE uuid=? `+s.dialect.LockForUpdate(), uuid) var d DeviceRow if err := row.Scan(&d.ID, &d.UUID, &d.UserID, &d.Name, &d.Platform, &d.LastSeen, &d.CreatedAt); err == sql.ErrNoRows { return nil, nil @@ -90,7 +93,7 @@ func (s *Store) findDeviceByUUIDTx(ctx context.Context, tx *sql.Tx, uuid string) // lockUser locks the users row to serialize per-user device mutations and // returns the user's status. Returns (false, "", nil) when the user is absent. func (s *Store) lockUser(ctx context.Context, tx *sql.Tx, userID int64) (exists bool, status string, err error) { - row := tx.QueryRowContext(ctx, `SELECT status FROM users WHERE id=? FOR UPDATE`, userID) + row := tx.QueryRowContext(ctx, `SELECT status FROM users WHERE id=? `+s.dialect.LockForUpdate(), userID) if e := row.Scan(&status); e == sql.ErrNoRows { return false, "", nil } else if e != nil { @@ -110,10 +113,11 @@ func (s *Store) countDevicesTx(ctx context.Context, tx *sql.Tx, userID int64) (i // insertDeviceTx inserts a new device row inside tx and returns it. func (s *Store) insertDeviceTx(ctx context.Context, tx *sql.Tx, uuid string, userID int64, name, platform string) (*DeviceRow, error) { + now := time.Now().UTC() res, err := tx.ExecContext(ctx, `INSERT INTO devices (uuid, user_id, name, platform, last_seen, created_at) - VALUES (?, ?, ?, ?, UTC_TIMESTAMP(6), UTC_TIMESTAMP(6))`, - uuid, userID, name, platform) + VALUES (?, ?, ?, ?, ?, ?)`, + uuid, userID, name, platform, now, now) if err != nil { return nil, fmt.Errorf("store.insertDeviceTx: %w", err) } @@ -124,7 +128,7 @@ func (s *Store) insertDeviceTx(ctx context.Context, tx *sql.Tx, uuid string, use // touchLastSeenTx updates a device's last_seen to now inside tx. func (s *Store) touchLastSeenTx(ctx context.Context, tx *sql.Tx, deviceID int64) error { _, err := tx.ExecContext(ctx, - `UPDATE devices SET last_seen=UTC_TIMESTAMP(6) WHERE id=?`, deviceID) + `UPDATE devices SET last_seen=? WHERE id=?`, time.Now().UTC(), deviceID) if err != nil { return fmt.Errorf("store.touchLastSeenTx: %w", err) } @@ -212,8 +216,8 @@ func (s *Store) writeAuditLogTx(ctx context.Context, tx *sql.Tx, actor, action, } _, err := tx.ExecContext(ctx, `INSERT INTO audit_log (actor, action, target, meta, at) - VALUES (?, ?, ?, ?, UTC_TIMESTAMP(6))`, - actor, action, target, metaJSON) + VALUES (?, ?, ?, ?, ?)`, + actor, action, target, metaJSON, time.Now().UTC()) if err != nil { return fmt.Errorf("store.writeAuditLogTx: %w", err) } diff --git a/server/internal/nodes/lifecycle.go b/server/internal/nodes/lifecycle.go index 1a0f4a4..aa8b343 100644 --- a/server/internal/nodes/lifecycle.go +++ b/server/internal/nodes/lifecycle.go @@ -6,10 +6,13 @@ import ( "encoding/json" "errors" "fmt" + "time" "github.com/redis/go-redis/v9" + dbx "github.com/wangjia/pangolin/server/internal/db" "github.com/wangjia/pangolin/server/internal/mtls" + "github.com/wangjia/pangolin/server/internal/store" ) // ErrInvalidTransition is returned when the requested status transition is not @@ -56,9 +59,10 @@ type transitionSpec struct { // // Called by #15 (block-detection / drain scheduler) and #14 (provisioning). type Lifecycle struct { - db *sql.DB - crl *mtls.CRL // for MarkDestroyed post-commit hook - rdb redis.Cmdable // for MarkDestroyed post-commit cleanup + db *sql.DB + dialect dbx.Dialect + crl *mtls.CRL // for MarkDestroyed post-commit hook + rdb redis.Cmdable // for MarkDestroyed post-commit cleanup } // NewLifecycle constructs a Lifecycle. @@ -66,22 +70,13 @@ type Lifecycle struct { // - crl: mTLS revocation manager (task 5b, mtls.NewCRL) // - rdb: Redis client or Cmdable (for post-destroy key cleanup) func NewLifecycle(db *sql.DB, crl *mtls.CRL, rdb redis.Cmdable) *Lifecycle { - return &Lifecycle{db: db, crl: crl, rdb: rdb} + return &Lifecycle{db: db, dialect: dbx.DialectForDB(db), crl: crl, rdb: rdb} } // BumpVersion atomically increments the directory_version singleton within tx. -// -// Signature matches the one defined by task 5d (BumpVersion(ctx, tx)); if 5d -// has merged, remove this copy and update callers to use the 5d version. -func BumpVersion(ctx context.Context, tx *sql.Tx) error { - _, err := tx.ExecContext(ctx, - `INSERT INTO directory_version (id, version) VALUES (1, 1) - ON DUPLICATE KEY UPDATE version = version + 1`, - ) - if err != nil { - return fmt.Errorf("nodes.BumpVersion: %w", err) - } - return nil +// Delegates to the canonical store.BumpDirectoryVersion (dialect-aware). +func BumpVersion(ctx context.Context, tx *sql.Tx, d dbx.Dialect) error { + return store.BumpDirectoryVersion(ctx, tx, d) } // MarkProbing transitions a node from provisioning → probing. @@ -197,14 +192,14 @@ func (l *Lifecycle) MarkBlockedSuspect(ctx context.Context, nodeUUID string, det return fmt.Errorf("lifecycle.MarkBlockedSuspect: marshal detail: %w", err) } if _, err := tx.ExecContext(ctx, - `INSERT INTO node_events (node_id, event, detail, at) VALUES (?, ?, ?, UTC_TIMESTAMP(6))`, - nodeID, string(eventBlockedSuspect), detailJSON, + `INSERT INTO node_events (node_id, event, detail, at) VALUES (?, ?, ?, ?)`, + nodeID, string(eventBlockedSuspect), detailJSON, time.Now().UTC(), ); err != nil { return fmt.Errorf("lifecycle.MarkBlockedSuspect: insert event: %w", err) } // Bump directory version. - if err := BumpVersion(ctx, tx); err != nil { + if err := BumpVersion(ctx, tx, l.dialect); err != nil { return err } @@ -265,15 +260,15 @@ func (l *Lifecycle) transact( // Insert the event record. if _, err := tx.ExecContext(ctx, - `INSERT INTO node_events (node_id, event, detail, at) VALUES (?, ?, ?, UTC_TIMESTAMP(6))`, - nodeID, string(spec.event), detailJSON, + `INSERT INTO node_events (node_id, event, detail, at) VALUES (?, ?, ?, ?)`, + nodeID, string(spec.event), detailJSON, time.Now().UTC(), ); err != nil { return fmt.Errorf("lifecycle.transact: insert event [%s→%s]: %w", spec.from, spec.to, err) } // Bump the global directory version. - if err := BumpVersion(ctx, tx); err != nil { + if err := BumpVersion(ctx, tx, l.dialect); err != nil { return err } diff --git a/server/internal/nodes/store.go b/server/internal/nodes/store.go index 448959b..01f997c 100644 --- a/server/internal/nodes/store.go +++ b/server/internal/nodes/store.go @@ -6,6 +6,7 @@ import ( "fmt" "time" + dbx "github.com/wangjia/pangolin/server/internal/db" agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" ) @@ -76,14 +77,15 @@ type NodeStore interface { bytesUp, bytesDown int64, minutes int64) error } -// SQLNodeStore implements NodeStore against a MySQL 8.x database. +// SQLNodeStore implements NodeStore against a SQL database (MySQL or SQLite). type SQLNodeStore struct { - db *sql.DB + db *sql.DB + dialect dbx.Dialect } // NewSQLNodeStore creates a SQLNodeStore backed by db. func NewSQLNodeStore(db *sql.DB) *SQLNodeStore { - return &SQLNodeStore{db: db} + return &SQLNodeStore{db: db, dialect: dbx.DialectForDB(db)} } // NodeByUUID looks up a node by UUID. Returns (nil, nil) when not found. @@ -160,12 +162,12 @@ func (s *SQLNodeStore) EntitlementForUser(ctx context.Context, userID int64) (*E SELECT p.code, p.ad_gate, p.daily_minutes, s.expires_at FROM subscriptions s JOIN plans p ON p.id = s.plan_id - WHERE s.user_id = ? AND s.expires_at > UTC_TIMESTAMP() + WHERE s.user_id = ? AND s.expires_at > ? ORDER BY s.expires_at DESC LIMIT 1 ` e := &Entitlement{DpUUID: dpUUID} - err := s.db.QueryRowContext(ctx, q, userID).Scan( + err := s.db.QueryRowContext(ctx, q, userID, time.Now().UTC()).Scan( &e.PlanCode, &e.AdGate, &e.DailyMinutes, &e.ExpiresAt, ) if err == sql.ErrNoRows { @@ -219,12 +221,12 @@ func (s *SQLNodeStore) ActiveNodeUUIDs(ctx context.Context) ([]string, error) { // CredentialsForNode returns active (non-expired) credentials for nodeUUID. func (s *SQLNodeStore) CredentialsForNode(ctx context.Context, nodeUUID string) ([]*agentv1.Credential, error) { const q = ` - SELECT cc.dp_uuid, cc.protocol, cc.flow, UNIX_TIMESTAMP(cc.expires_at) + SELECT cc.dp_uuid, cc.protocol, cc.flow, cc.expires_at FROM connect_credentials cc JOIN nodes n ON n.id = cc.node_id - WHERE n.uuid = ? AND cc.expires_at > UTC_TIMESTAMP() + WHERE n.uuid = ? AND cc.expires_at > ? ` - rows, err := s.db.QueryContext(ctx, q, nodeUUID) + rows, err := s.db.QueryContext(ctx, q, nodeUUID, time.Now().UTC()) if err != nil { return nil, fmt.Errorf("nodes.SQLNodeStore.CredentialsForNode: %w", err) } @@ -233,9 +235,11 @@ func (s *SQLNodeStore) CredentialsForNode(ctx context.Context, nodeUUID string) var out []*agentv1.Credential for rows.Next() { var c agentv1.Credential - if err := rows.Scan(&c.DpUUID, &c.Protocol, &c.Flow, &c.ExpiresAtUnix); err != nil { + var expiresAt time.Time + if err := rows.Scan(&c.DpUUID, &c.Protocol, &c.Flow, &expiresAt); err != nil { return nil, fmt.Errorf("nodes.SQLNodeStore.CredentialsForNode scan: %w", err) } + c.ExpiresAtUnix = expiresAt.Unix() out = append(out, &c) } return out, rows.Err() @@ -243,14 +247,13 @@ func (s *SQLNodeStore) CredentialsForNode(ctx context.Context, nodeUUID string) // PersistCredential upserts the credential into connect_credentials. func (s *SQLNodeStore) PersistCredential(ctx context.Context, nodeID int64, cred *agentv1.Credential, expiresAt time.Time) error { - const q = ` + q := ` INSERT INTO connect_credentials (node_id, dp_uuid, protocol, flow, expires_at) VALUES (?, ?, ?, ?, ?) - ON DUPLICATE KEY UPDATE - protocol = VALUES(protocol), - flow = VALUES(flow), - expires_at = VALUES(expires_at) - ` + ` + s.dialect.Upsert([]string{"node_id", "dp_uuid"}, + "protocol = EXCLUDED.protocol", + "flow = EXCLUDED.flow", + "expires_at = EXCLUDED.expires_at") if _, err := s.db.ExecContext(ctx, q, nodeID, cred.DpUUID, int32(cred.Protocol), cred.Flow, expiresAt.UTC(), ); err != nil { @@ -290,14 +293,13 @@ func (s *SQLNodeStore) AccumulateUsage( ctx context.Context, userID int64, date time.Time, bytesUp, bytesDown int64, minutes int64, ) error { - const q = ` + q := ` INSERT INTO usage_daily (user_id, date, bytes_up, bytes_down, minutes_used) VALUES (?, ?, ?, ?, ?) - ON DUPLICATE KEY UPDATE - bytes_up = bytes_up + VALUES(bytes_up), - bytes_down = bytes_down + VALUES(bytes_down), - minutes_used = minutes_used + VALUES(minutes_used) - ` + ` + s.dialect.Upsert([]string{"user_id", "date"}, + "bytes_up = bytes_up + EXCLUDED.bytes_up", + "bytes_down = bytes_down + EXCLUDED.bytes_down", + "minutes_used = minutes_used + EXCLUDED.minutes_used") if _, err := s.db.ExecContext(ctx, q, userID, date.Format("2006-01-02"), bytesUp, bytesDown, minutes, ); err != nil { diff --git a/server/internal/provision/mysqlstore.go b/server/internal/provision/mysqlstore.go index a088086..5dc5ee4 100644 --- a/server/internal/provision/mysqlstore.go +++ b/server/internal/provision/mysqlstore.go @@ -6,15 +6,22 @@ import ( "encoding/json" "fmt" "strings" + "time" + + dbx "github.com/wangjia/pangolin/server/internal/db" ) // MySQLStore is the production Store backed by the shared *sql.DB pool. +// (Despite the name it is dialect-aware and also works against SQLite.) type MySQLStore struct { - db *sql.DB + db *sql.DB + dialect dbx.Dialect } -// NewMySQLStore wraps a MySQL connection pool. -func NewMySQLStore(db *sql.DB) *MySQLStore { return &MySQLStore{db: db} } +// NewMySQLStore wraps a database connection pool (MySQL or SQLite). +func NewMySQLStore(db *sql.DB) *MySQLStore { + return &MySQLStore{db: db, dialect: dbx.DialectForDB(db)} +} var _ Store = (*MySQLStore)(nil) @@ -54,10 +61,10 @@ func (s *MySQLStore) InsertNode(ctx context.Context, n *Node) (int64, error) { `INSERT INTO nodes (uuid, region, name_zh, name_en, role, tier, endpoint, hy2_port, reality_pbk, reality_sni, provider_id, tags, status, weight, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, UTC_TIMESTAMP(6))`, + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, n.UUID, n.Region, n.NameZH, n.NameEn, string(n.Role), string(n.Tier), endpoint, nullInt(n.HY2Port), n.RealityPBK, n.RealitySNI, n.ProviderID, - marshalJSONList(n.Tags), string(status), weight) + marshalJSONList(n.Tags), string(status), weight, time.Now().UTC()) if err != nil { return 0, fmt.Errorf("store.InsertNode: %w", err) } @@ -236,8 +243,8 @@ func (s *MySQLStore) WriteNodeEvent(ctx context.Context, nodeID int64, event Eve detailJSON = "null" } _, err := s.db.ExecContext(ctx, - `INSERT INTO node_events (node_id, event, detail, at) VALUES (?, ?, ?, UTC_TIMESTAMP(6))`, - nodeID, string(event), detailJSON) + `INSERT INTO node_events (node_id, event, detail, at) VALUES (?, ?, ?, ?)`, + nodeID, string(event), detailJSON, time.Now().UTC()) if err != nil { return fmt.Errorf("store.WriteNodeEvent: %w", err) } @@ -249,8 +256,8 @@ func (s *MySQLStore) WriteAuditLog(ctx context.Context, actor, action, target, m metaJSON = "null" } _, err := s.db.ExecContext(ctx, - `INSERT INTO audit_log (actor, action, target, meta, at) VALUES (?, ?, ?, ?, UTC_TIMESTAMP(6))`, - actor, action, target, metaJSON) + `INSERT INTO audit_log (actor, action, target, meta, at) VALUES (?, ?, ?, ?, ?)`, + actor, action, target, metaJSON, time.Now().UTC()) if err != nil { return fmt.Errorf("store.WriteAuditLog: %w", err) } @@ -288,9 +295,9 @@ func (s *MySQLStore) LookupIdempotency(ctx context.Context, key string) (string, func (s *MySQLStore) SaveIdempotency(ctx context.Context, key, nodeUUID string) error { _, err := s.db.ExecContext(ctx, `INSERT INTO provision_idempotency (idempotency_key, node_uuid, created_at) - VALUES (?, ?, UTC_TIMESTAMP(6)) - ON DUPLICATE KEY UPDATE idempotency_key = idempotency_key`, - key, nodeUUID) + VALUES (?, ?, ?) `+ + s.dialect.Upsert([]string{"idempotency_key"}), + key, nodeUUID, time.Now().UTC()) if err != nil { return fmt.Errorf("store.SaveIdempotency: %w", err) } diff --git a/server/internal/scheduler/wiring_lifecycle.go b/server/internal/scheduler/wiring_lifecycle.go index 818355a..8ca2fae 100644 --- a/server/internal/scheduler/wiring_lifecycle.go +++ b/server/internal/scheduler/wiring_lifecycle.go @@ -8,9 +8,11 @@ import ( "strings" "time" + dbx "github.com/wangjia/pangolin/server/internal/db" "github.com/wangjia/pangolin/server/internal/nodes" "github.com/wangjia/pangolin/server/internal/scheduler/detect" "github.com/wangjia/pangolin/server/internal/scheduler/orchestrate" + "github.com/wangjia/pangolin/server/internal/store" ) // SQLLifecycle is a MySQL-backed lifecycle store. The detect and orchestrate @@ -30,13 +32,14 @@ import ( // — provision.DestroyNode tears down the VM, and the agent's mTLS cert can be // revoked separately. Wire CRL revocation when the real fleet exists (TODO). type SQLLifecycle struct { - db *sql.DB - loads *nodes.LoadCache + db *sql.DB + dialect dbx.Dialect + loads *nodes.LoadCache } // NewSQLLifecycle wires the store. loads may be nil (load reads then return zero). func NewSQLLifecycle(db *sql.DB, loads *nodes.LoadCache) *SQLLifecycle { - return &SQLLifecycle{db: db, loads: loads} + return &SQLLifecycle{db: db, dialect: dbx.DialectForDB(db), loads: loads} } const nodeSelectCols = `id, uuid, status, region, tier, role, reality_sni, reality_pbk, hy2_port, provider_id, weight, name_zh, name_en` @@ -134,9 +137,7 @@ func (l *SQLLifecycle) transition(ctx context.Context, nodeID, from, to string, _, _ = tx.ExecContext(ctx, `INSERT INTO node_events (node_id, event, detail) VALUES (?, ?, ?)`, nodeID, ev, dj) } - if _, err := tx.ExecContext(ctx, - `INSERT INTO directory_version (id, version) VALUES (1, 1) - ON DUPLICATE KEY UPDATE version = version + 1`); err != nil { + if err := store.BumpDirectoryVersion(ctx, tx, l.dialect); err != nil { return 0, err } if err := tx.Commit(); err != nil { @@ -151,10 +152,7 @@ func (l *SQLLifecycle) setWeight(ctx context.Context, nodeID string, weight int) } func (l *SQLLifecycle) bumpVersion(ctx context.Context) error { - _, err := l.db.ExecContext(ctx, - `INSERT INTO directory_version (id, version) VALUES (1, 1) - ON DUPLICATE KEY UPDATE version = version + 1`) - return err + return store.BumpDirectoryVersion(ctx, l.db, l.dialect) } func (l *SQLLifecycle) version(ctx context.Context) int64 { diff --git a/server/internal/store/directory.go b/server/internal/store/directory.go new file mode 100644 index 0000000..7aa83b6 --- /dev/null +++ b/server/internal/store/directory.go @@ -0,0 +1,28 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + + "github.com/wangjia/pangolin/server/internal/db" +) + +// Execer is satisfied by both *sql.DB and *sql.Tx. +type Execer interface { + ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) +} + +// BumpDirectoryVersion atomically increments the directory_version singleton +// (the row is seeded by migration 7). This is the single canonical +// implementation; nodes.Lifecycle and the scheduler both call it instead of +// each carrying their own copy of the upsert. +func BumpDirectoryVersion(ctx context.Context, q Execer, d db.Dialect) error { + _, err := q.ExecContext(ctx, + `INSERT INTO directory_version (id, version) VALUES (1, 1) `+ + d.Upsert([]string{"id"}, "version = version + 1")) + if err != nil { + return fmt.Errorf("store.BumpDirectoryVersion: %w", err) + } + return nil +} diff --git a/server/internal/usage/store.go b/server/internal/usage/store.go index 6483fe0..b808ac7 100644 --- a/server/internal/usage/store.go +++ b/server/internal/usage/store.go @@ -5,6 +5,8 @@ import ( "database/sql" "fmt" "time" + + dbx "github.com/wangjia/pangolin/server/internal/db" ) // Unlimited is the sentinel returned by CheckFreeConnect for plans with no @@ -37,11 +39,12 @@ type DailyUsage struct { // Store wraps a *sql.DB and exposes the usage_daily / plans / users queries the // usage package needs. type Store struct { - db *sql.DB + db *sql.DB + dialect dbx.Dialect } -// NewStore creates a Store backed by the given MySQL connection pool. -func NewStore(db *sql.DB) *Store { return &Store{db: db} } +// NewStore creates a Store backed by the given connection pool (MySQL or SQLite). +func NewStore(db *sql.DB) *Store { return &Store{db: db, dialect: dbx.DialectForDB(db)} } // AggregateUsage accumulates one usage delta into usage_daily for (userID, day) // using INSERT … ON DUPLICATE KEY UPDATE so concurrent reports from multiple @@ -51,10 +54,10 @@ func (s *Store) AggregateUsage(ctx context.Context, userID int64, day time.Time, _, err := s.db.ExecContext(ctx, `INSERT INTO usage_daily (user_id, date, bytes_up, bytes_down, minutes_used) VALUES (?, ?, ?, ?, ?) - ON DUPLICATE KEY UPDATE - bytes_up = bytes_up + VALUES(bytes_up), - bytes_down = bytes_down + VALUES(bytes_down), - minutes_used = minutes_used + VALUES(minutes_used)`, + `+s.dialect.Upsert([]string{"user_id", "date"}, + "bytes_up = bytes_up + EXCLUDED.bytes_up", + "bytes_down = bytes_down + EXCLUDED.bytes_down", + "minutes_used = minutes_used + EXCLUDED.minutes_used"), userID, d, bytesUp, bytesDown, minutes) if err != nil { return fmt.Errorf("store.AggregateUsage: %w", err) @@ -128,6 +131,7 @@ func (s *Store) GetDay(ctx context.Context, userID int64, day time.Time) (*Daily // SELECT … FOR UPDATE to be safe under concurrent unlock attempts. func (s *Store) MarkAdUnlocked(ctx context.Context, userID int64, day time.Time) (alreadyUnlocked bool, err error) { d := day.UTC().Format(dateLayout) + now := time.Now().UTC() tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) if err != nil { @@ -142,7 +146,7 @@ func (s *Store) MarkAdUnlocked(ctx context.Context, userID int64, day time.Time) var unlockedAt sql.NullTime row := tx.QueryRowContext(ctx, - `SELECT ad_unlocked_at FROM usage_daily WHERE user_id = ? AND date = ? FOR UPDATE`, + `SELECT ad_unlocked_at FROM usage_daily WHERE user_id = ? AND date = ? `+s.dialect.LockForUpdate(), userID, d) switch err := row.Scan(&unlockedAt); err { case nil: @@ -155,16 +159,16 @@ func (s *Store) MarkAdUnlocked(ctx context.Context, userID int64, day time.Time) return true, nil } if _, uErr := tx.ExecContext(ctx, - `UPDATE usage_daily SET ad_unlocked_at = UTC_TIMESTAMP(6) + `UPDATE usage_daily SET ad_unlocked_at = ? WHERE user_id = ? AND date = ? AND ad_unlocked_at IS NULL`, - userID, d); uErr != nil { + now, userID, d); uErr != nil { return false, fmt.Errorf("store.MarkAdUnlocked update: %w", uErr) } case sql.ErrNoRows: if _, iErr := tx.ExecContext(ctx, `INSERT INTO usage_daily (user_id, date, ad_unlocked_at) - VALUES (?, ?, UTC_TIMESTAMP(6))`, - userID, d); iErr != nil { + VALUES (?, ?, ?)`, + userID, d, now); iErr != nil { return false, fmt.Errorf("store.MarkAdUnlocked insert: %w", iErr) } default: @@ -187,10 +191,11 @@ func (s *Store) EffectivePlan(ctx context.Context, userID int64) (*Plan, error) `SELECT p.code, p.daily_minutes, p.ad_gate FROM subscriptions s JOIN plans p ON p.id = s.plan_id - WHERE s.user_id = ? AND s.expires_at > UTC_TIMESTAMP(6) - ORDER BY FIELD(p.code, 'team', 'pro', 'free'), s.expires_at DESC + WHERE s.user_id = ? AND s.expires_at > ? + ORDER BY CASE p.code WHEN 'team' THEN 0 WHEN 'pro' THEN 1 WHEN 'free' THEN 2 ELSE 3 END, + s.expires_at DESC LIMIT 1`, - userID).Scan(&p.Code, &p.DailyMinutes, &p.AdGate) + userID, time.Now().UTC()).Scan(&p.Code, &p.DailyMinutes, &p.AdGate) if err == nil { return &p, nil }