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
}