feat(server/db): 多库(3/4)— Dialect 抽象(upsert/锁)+ 合并 directory_version

- 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 <noreply@anthropic.com>

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
 		}
 

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 {