Merge branch 'main' into opencode/quick-comet

go-backend/internal/store/db.go

@@ -0,0 +1,466 @@
+// Package store provides a thin dialect-aware wrapper around database/sql,
+// enabling transparent use of both SQLite and PostgreSQL.
+package store
+
+import (
+	"database/sql"
+	"strconv"
+	"strings"
+)
+
+// Dialect identifies the underlying database engine.
+type Dialect int
+
+const (
+	DialectSQLite Dialect = iota
+	DialectPostgres
+)
+
+// String returns a human-readable dialect name.
+func (d Dialect) String() string {
+	switch d {
+	case DialectSQLite:
+		return "sqlite"
+	case DialectPostgres:
+		return "postgres"
+	default:
+		return "unknown"
+	}
+}
+
+// DB wraps *sql.DB with dialect awareness.
+type DB struct {
+	raw     *sql.DB
+	dialect Dialect
+}
+
+// Wrap creates a new dialect-aware DB from an existing *sql.DB.
+func Wrap(raw *sql.DB, dialect Dialect) *DB {
+	return &DB{raw: raw, dialect: dialect}
+}
+
+// Dialect returns the database dialect.
+func (db *DB) Dialect() Dialect {
+	if db == nil {
+		return DialectSQLite
+	}
+	return db.dialect
+}
+
+// RawDB returns the underlying *sql.DB.
+func (db *DB) RawDB() *sql.DB {
+	if db == nil {
+		return nil
+	}
+	return db.raw
+}
+
+// Close closes the underlying connection.
+func (db *DB) Close() error {
+	if db == nil || db.raw == nil {
+		return nil
+	}
+	return db.raw.Close()
+}
+
+// Ping verifies the connection is alive.
+func (db *DB) Ping() error {
+	return db.raw.Ping()
+}
+
+// Exec executes a query with transparent placeholder and syntax rewriting.
+func (db *DB) Exec(query string, args ...any) (sql.Result, error) {
+	return db.raw.Exec(db.rewrite(query), args...)
+}
+
+// Query executes a query that returns rows, with transparent rewriting.
+func (db *DB) Query(query string, args ...any) (*sql.Rows, error) {
+	return db.raw.Query(db.rewrite(query), args...)
+}
+
+// QueryRow executes a query that returns at most one row, with transparent rewriting.
+func (db *DB) QueryRow(query string, args ...any) *sql.Row {
+	return db.raw.QueryRow(db.rewrite(query), args...)
+}
+
+// Begin starts a transaction, returning a dialect-aware Tx.
+func (db *DB) Begin() (*Tx, error) {
+	tx, err := db.raw.Begin()
+	if err != nil {
+		return nil, err
+	}
+	return &Tx{raw: tx, dialect: db.dialect}, nil
+}
+
+// ExecReturningID executes an INSERT and returns the auto-generated id.
+//   - SQLite: uses LastInsertId()
+//   - PostgreSQL: appends RETURNING id and uses QueryRow().Scan()
+func (db *DB) ExecReturningID(query string, args ...any) (int64, error) {
+	q := db.rewrite(query)
+	if db.dialect == DialectPostgres {
+		q = ensureReturningID(q)
+		var id int64
+		if err := db.raw.QueryRow(q, args...).Scan(&id); err != nil {
+			return 0, err
+		}
+		return id, nil
+	}
+	res, err := db.raw.Exec(q, args...)
+	if err != nil {
+		return 0, err
+	}
+	return res.LastInsertId()
+}
+
+// Tx wraps *sql.Tx with dialect awareness.
+type Tx struct {
+	raw     *sql.Tx
+	dialect Dialect
+}
+
+// Exec executes a query inside the transaction with transparent rewriting.
+func (tx *Tx) Exec(query string, args ...any) (sql.Result, error) {
+	return tx.raw.Exec(rewriteQuery(tx.dialect, query), args...)
+}
+
+// Query executes a query that returns rows inside the transaction.
+func (tx *Tx) Query(query string, args ...any) (*sql.Rows, error) {
+	return tx.raw.Query(rewriteQuery(tx.dialect, query), args...)
+}
+
+// QueryRow executes a query that returns at most one row inside the transaction.
+func (tx *Tx) QueryRow(query string, args ...any) *sql.Row {
+	return tx.raw.QueryRow(rewriteQuery(tx.dialect, query), args...)
+}
+
+// Commit commits the transaction.
+func (tx *Tx) Commit() error { return tx.raw.Commit() }
+
+// Rollback aborts the transaction.
+func (tx *Tx) Rollback() error { return tx.raw.Rollback() }
+
+// ExecReturningID executes an INSERT inside the transaction and returns the id.
+func (tx *Tx) ExecReturningID(query string, args ...any) (int64, error) {
+	q := rewriteQuery(tx.dialect, query)
+	if tx.dialect == DialectPostgres {
+		q = ensureReturningID(q)
+		var id int64
+		if err := tx.raw.QueryRow(q, args...).Scan(&id); err != nil {
+			return 0, err
+		}
+		return id, nil
+	}
+	res, err := tx.raw.Exec(q, args...)
+	if err != nil {
+		return 0, err
+	}
+	return res.LastInsertId()
+}
+
+func (db *DB) rewrite(query string) string {
+	return rewriteQuery(db.dialect, query)
+}
+
+func rewriteQuery(dialect Dialect, query string) string {
+	if dialect != DialectPostgres {
+		return query
+	}
+	query = rewriteUserIdentifier(query)
+	query = rewriteInsertOrIgnore(query)
+	query = rewritePlaceholders(query)
+	return query
+}
+
+func rewriteUserIdentifier(query string) string {
+	var buf strings.Builder
+	buf.Grow(len(query) + 16)
+	i := 0
+	for i < len(query) {
+		if end, ok := skipSQLProtectedSegment(query, i); ok {
+			buf.WriteString(query[i:end])
+			i = end
+			continue
+		}
+
+		ch := query[i]
+		if isIdentifierChar(ch) {
+			j := i + 1
+			for j < len(query) && isIdentifierChar(query[j]) {
+				j++
+			}
+			tok := query[i:j]
+			if strings.EqualFold(tok, "user") {
+				buf.WriteString(`"user"`)
+			} else {
+				buf.WriteString(tok)
+			}
+			i = j
+			continue
+		}
+
+		buf.WriteByte(ch)
+		i++
+	}
+	return buf.String()
+}
+
+func isIdentifierChar(ch byte) bool {
+	if ch >= 'a' && ch <= 'z' {
+		return true
+	}
+	if ch >= 'A' && ch <= 'Z' {
+		return true
+	}
+	if ch >= '0' && ch <= '9' {
+		return true
+	}
+	return ch == '_'
+}
+
+func rewriteInsertOrIgnore(query string) string {
+	start, end, ok := findKeywordSequenceOutside(query, []string{"INSERT", "OR", "IGNORE", "INTO"}, 0)
+	if !ok {
+		return query
+	}
+
+	rewritten := query[:start] + "INSERT INTO" + query[end:]
+	rewritten = strings.TrimRight(rewritten, "; \t\n")
+
+	insertIntoEnd := start + len("INSERT INTO")
+	if _, _, hasOnConflict := findKeywordSequenceOutside(rewritten, []string{"ON", "CONFLICT"}, insertIntoEnd); hasOnConflict {
+		return rewritten
+	}
+
+	if retStart, _, hasReturning := findKeywordSequenceOutside(rewritten, []string{"RETURNING"}, insertIntoEnd); hasReturning {
+		prefix := strings.TrimRight(rewritten[:retStart], " \t\n")
+		suffix := strings.TrimLeft(rewritten[retStart:], " \t\n")
+		return prefix + " ON CONFLICT DO NOTHING " + suffix
+	}
+
+	return rewritten + " ON CONFLICT DO NOTHING"
+}
+
+func rewritePlaceholders(query string) string {
+	var buf strings.Builder
+	buf.Grow(len(query) + 16)
+	n := 1
+	for i := 0; i < len(query); i++ {
+		if end, ok := skipSQLProtectedSegment(query, i); ok {
+			buf.WriteString(query[i:end])
+			i = end - 1
+			continue
+		}
+
+		ch := query[i]
+		if ch == '?' {
+			buf.WriteByte('$')
+			buf.WriteString(strconv.Itoa(n))
+			n++
+			continue
+		}
+		buf.WriteByte(ch)
+	}
+	return buf.String()
+}
+
+func ensureReturningID(query string) string {
+	trimmed := strings.TrimRight(query, "; \t\n")
+	if _, _, ok := findKeywordSequenceOutside(trimmed, []string{"RETURNING"}, 0); ok {
+		return trimmed
+	}
+	return trimmed + " RETURNING id"
+}
+
+func findKeywordSequenceOutside(query string, keywords []string, from int) (int, int, bool) {
+	if len(keywords) == 0 {
+		return 0, 0, false
+	}
+	if from < 0 {
+		from = 0
+	}
+	if from >= len(query) {
+		return 0, 0, false
+	}
+
+	matched := 0
+	seqStart := -1
+
+	for i := from; i < len(query); {
+		if end, ok := skipSQLProtectedSegment(query, i); ok {
+			i = end
+			continue
+		}
+
+		ch := query[i]
+		if isIdentifierChar(ch) {
+			j := i + 1
+			for j < len(query) && isIdentifierChar(query[j]) {
+				j++
+			}
+			tok := query[i:j]
+
+			if strings.EqualFold(tok, keywords[matched]) {
+				if matched == 0 {
+					seqStart = i
+				}
+				matched++
+				if matched == len(keywords) {
+					return seqStart, j, true
+				}
+			} else if strings.EqualFold(tok, keywords[0]) {
+				seqStart = i
+				matched = 1
+			} else {
+				matched = 0
+				seqStart = -1
+			}
+
+			i = j
+			continue
+		}
+
+		if !isSQLSpace(ch) {
+			matched = 0
+			seqStart = -1
+		}
+		i++
+	}
+
+	return 0, 0, false
+}
+
+func skipSQLProtectedSegment(query string, i int) (int, bool) {
+	if i < 0 || i >= len(query) {
+		return 0, false
+	}
+
+	switch query[i] {
+	case '\'':
+		return skipSingleQuotedLiteral(query, i), true
+	case '"':
+		return skipDoubleQuotedIdentifier(query, i), true
+	case '-':
+		if i+1 < len(query) && query[i+1] == '-' {
+			return skipLineComment(query, i), true
+		}
+	case '/':
+		if i+1 < len(query) && query[i+1] == '*' {
+			return skipBlockComment(query, i), true
+		}
+	case '$':
+		if end, ok := skipDollarQuotedLiteral(query, i); ok {
+			return end, true
+		}
+	}
+
+	return 0, false
+}
+
+func skipSingleQuotedLiteral(query string, i int) int {
+	for j := i + 1; j < len(query); j++ {
+		if query[j] != '\'' {
+			continue
+		}
+		if j+1 < len(query) && query[j+1] == '\'' {
+			j++
+			continue
+		}
+		return j + 1
+	}
+	return len(query)
+}
+
+func skipDoubleQuotedIdentifier(query string, i int) int {
+	for j := i + 1; j < len(query); j++ {
+		if query[j] != '"' {
+			continue
+		}
+		if j+1 < len(query) && query[j+1] == '"' {
+			j++
+			continue
+		}
+		return j + 1
+	}
+	return len(query)
+}
+
+func skipLineComment(query string, i int) int {
+	for j := i + 2; j < len(query); j++ {
+		if query[j] == '\n' {
+			return j
+		}
+	}
+	return len(query)
+}
+
+func skipBlockComment(query string, i int) int {
+	depth := 1
+	for j := i + 2; j < len(query)-1; j++ {
+		if query[j] == '/' && query[j+1] == '*' {
+			depth++
+			j++
+			continue
+		}
+		if query[j] == '*' && query[j+1] == '/' {
+			depth--
+			j++
+			if depth == 0 {
+				return j + 1
+			}
+		}
+	}
+	return len(query)
+}
+
+func skipDollarQuotedLiteral(query string, i int) (int, bool) {
+	if i < 0 || i >= len(query) || query[i] != '$' {
+		return 0, false
+	}
+
+	if i+1 >= len(query) {
+		return 0, false
+	}
+
+	var endTag int
+	if query[i+1] == '$' {
+		endTag = i + 1
+	} else {
+		if !isDollarTagStart(query[i+1]) {
+			return 0, false
+		}
+		j := i + 2
+		for j < len(query) && isDollarTagChar(query[j]) {
+			j++
+		}
+		if j >= len(query) || query[j] != '$' {
+			return 0, false
+		}
+		endTag = j
+	}
+
+	tag := query[i : endTag+1]
+	if closeIdx := strings.Index(query[endTag+1:], tag); closeIdx >= 0 {
+		return endTag + 1 + closeIdx + len(tag), true
+	}
+	return len(query), true
+}
+
+func isDollarTagStart(ch byte) bool {
+	return ch == '_' || (ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z')
+}
+
+func isDollarTagChar(ch byte) bool {
+	if isDollarTagStart(ch) {
+		return true
+	}
+	return ch >= '0' && ch <= '9'
+}
+
+func isSQLSpace(ch byte) bool {
+	switch ch {
+	case ' ', '\t', '\n', '\r', '\f':
+		return true
+	default:
+		return false
+	}
+}

go-backend/internal/store/db_test.go

@@ -0,0 +1,116 @@
+package store
+
+import "testing"
+
+func TestRewritePlaceholdersSkipsProtectedSegments(t *testing.T) {
+	q := `SELECT ?, '?', "id?", $$body ? $$, $tag$X?$tag$, col -- comment ?
+FROM t /* block ? */ WHERE id = ?`
+	got := rewritePlaceholders(q)
+	want := `SELECT $1, '?', "id?", $$body ? $$, $tag$X?$tag$, col -- comment ?
+FROM t /* block ? */ WHERE id = $2`
+	if got != want {
+		t.Fatalf("unexpected rewrite\nwant: %s\ngot:  %s", want, got)
+	}
+}
+
+func TestRewriteInsertOrIgnoreBasic(t *testing.T) {
+	q := `INSERT OR IGNORE INTO user_group_user(user_group_id, user_id, created_time) VALUES(?, ?, ?)`
+	got := rewriteInsertOrIgnore(q)
+	want := `INSERT INTO user_group_user(user_group_id, user_id, created_time) VALUES(?, ?, ?) ON CONFLICT DO NOTHING`
+	if got != want {
+		t.Fatalf("unexpected rewrite\nwant: %s\ngot:  %s", want, got)
+	}
+}
+
+func TestRewriteInsertOrIgnoreBeforeReturning(t *testing.T) {
+	q := `INSERT OR IGNORE INTO x(a) VALUES(?) RETURNING id`
+	got := rewriteInsertOrIgnore(q)
+	want := `INSERT INTO x(a) VALUES(?) ON CONFLICT DO NOTHING RETURNING id`
+	if got != want {
+		t.Fatalf("unexpected rewrite\nwant: %s\ngot:  %s", want, got)
+	}
+}
+
+func TestRewriteInsertOrIgnoreNotDuplicatingOnConflict(t *testing.T) {
+	q := `INSERT OR IGNORE INTO x(a) VALUES(?) ON CONFLICT(a) DO UPDATE SET a=excluded.a`
+	got := rewriteInsertOrIgnore(q)
+	want := `INSERT INTO x(a) VALUES(?) ON CONFLICT(a) DO UPDATE SET a=excluded.a`
+	if got != want {
+		t.Fatalf("unexpected rewrite\nwant: %s\ngot:  %s", want, got)
+	}
+}
+
+func TestEnsureReturningID(t *testing.T) {
+	if got := ensureReturningID(`INSERT INTO x(a) VALUES($1)`); got != `INSERT INTO x(a) VALUES($1) RETURNING id` {
+		t.Fatalf("missing RETURNING append: %s", got)
+	}
+	if got := ensureReturningID(`INSERT INTO x(a) VALUES($1) RETURNING other_id`); got != `INSERT INTO x(a) VALUES($1) RETURNING other_id` {
+		t.Fatalf("RETURNING should not be duplicated: %s", got)
+	}
+}
+
+func TestRewriteUserIdentifierSafety(t *testing.T) {
+	q := `SELECT user, user_id, 'user', "user", note FROM user -- user
+WHERE owner='user'`
+	got := rewriteUserIdentifier(q)
+	want := `SELECT "user", user_id, 'user', "user", note FROM "user" -- user
+WHERE owner='user'`
+	if got != want {
+		t.Fatalf("unexpected rewrite\nwant: %s\ngot:  %s", want, got)
+	}
+}
+
+func TestRewriteQueryPostgresPipeline(t *testing.T) {
+	q := `INSERT OR IGNORE INTO user(name, note) VALUES(?, '?')`
+	got := rewriteQuery(DialectPostgres, q)
+	want := `INSERT INTO "user"(name, note) VALUES($1, '?') ON CONFLICT DO NOTHING`
+	if got != want {
+		t.Fatalf("unexpected rewrite\nwant: %s\ngot:  %s", want, got)
+	}
+}
+
+func TestRewriteInsertOrIgnoreSkipsStringLiteral(t *testing.T) {
+	q := `SELECT 'INSERT OR IGNORE INTO t(a) VALUES(?)' AS q`
+	got := rewriteInsertOrIgnore(q)
+	if got != q {
+		t.Fatalf("string literal should stay unchanged\nwant: %s\ngot:  %s", q, got)
+	}
+}
+
+func TestRewriteInsertOrIgnoreSkipsCommentedKeyword(t *testing.T) {
+	q := `-- INSERT OR IGNORE INTO ignored(a) VALUES(?)
+INSERT OR IGNORE INTO real_t(a) VALUES(?)`
+	got := rewriteInsertOrIgnore(q)
+	want := `-- INSERT OR IGNORE INTO ignored(a) VALUES(?)
+INSERT INTO real_t(a) VALUES(?) ON CONFLICT DO NOTHING`
+	if got != want {
+		t.Fatalf("unexpected rewrite\nwant: %s\ngot:  %s", want, got)
+	}
+}
+
+func TestRewritePlaceholdersSkipsNestedBlockComment(t *testing.T) {
+	q := `SELECT ? /* outer ? /* inner ? */ still_outer ? */ FROM t WHERE id = ?`
+	got := rewritePlaceholders(q)
+	want := `SELECT $1 /* outer ? /* inner ? */ still_outer ? */ FROM t WHERE id = $2`
+	if got != want {
+		t.Fatalf("unexpected rewrite\nwant: %s\ngot:  %s", want, got)
+	}
+}
+
+func TestRewritePlaceholdersSkipsUnterminatedBlockComment(t *testing.T) {
+	q := `SELECT ? /* unterminated ? comment`
+	got := rewritePlaceholders(q)
+	want := `SELECT $1 /* unterminated ? comment`
+	if got != want {
+		t.Fatalf("unexpected rewrite\nwant: %s\ngot:  %s", want, got)
+	}
+}
+
+func TestRewriteUserIdentifierSkipsDollarQuotedAndComment(t *testing.T) {
+	q := `SELECT user, $$user ?$$ AS body, col FROM user /* user */ -- user`
+	got := rewriteUserIdentifier(q)
+	want := `SELECT "user", $$user ?$$ AS body, col FROM "user" /* user */ -- user`
+	if got != want {
+		t.Fatalf("unexpected rewrite\nwant: %s\ngot:  %s", want, got)
+	}
+}

go-backend/internal/store/postgres/embed.go

@@ -0,0 +1,9 @@
+package postgres
+
+import _ "embed"
+
+//go:embed sql/schema.sql
+var EmbeddedSchema string
+
+//go:embed sql/data.sql
+var EmbeddedSeedData string

go-backend/internal/store/postgres/sql/data.sql

@@ -0,0 +1,18 @@
+INSERT INTO "user" (id, "user", pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status)
+VALUES (1, 'admin_user', '3c85cdebade1c51cf64ca9f3c09d182d', 0, 2727251700000, 99999, 0, 0, 1, 99999, 1748914865000, 1754011744252, 1)
+ON CONFLICT DO NOTHING;
+
+INSERT INTO vite_config (id, name, value, time)
+VALUES (1, 'app_name', 'flux', 1755147963000)
+ON CONFLICT DO NOTHING;
+
+DO $$
+BEGIN
+    IF to_regclass('public.user_id_seq') IS NOT NULL THEN
+        PERFORM setval('user_id_seq', (SELECT COALESCE(MAX(id), 0) FROM "user"));
+    END IF;
+    IF to_regclass('public.vite_config_id_seq') IS NOT NULL THEN
+        PERFORM setval('vite_config_id_seq', (SELECT COALESCE(MAX(id), 0) FROM vite_config));
+    END IF;
+END
+$$;

go-backend/internal/store/postgres/sql/schema.sql

@@ -0,0 +1,241 @@
+CREATE TABLE IF NOT EXISTS forward (
+  id SERIAL PRIMARY KEY,
+  user_id INTEGER NOT NULL,
+  user_name VARCHAR(100) NOT NULL,
+  name VARCHAR(100) NOT NULL,
+  tunnel_id INTEGER NOT NULL,
+  remote_addr TEXT NOT NULL,
+  strategy VARCHAR(100) NOT NULL DEFAULT 'fifo',
+  in_flow BIGINT NOT NULL DEFAULT 0,
+  out_flow BIGINT NOT NULL DEFAULT 0,
+  created_time BIGINT NOT NULL,
+  updated_time BIGINT NOT NULL,
+  status INTEGER NOT NULL,
+  inx INTEGER NOT NULL DEFAULT 0
+);
+
+CREATE TABLE IF NOT EXISTS forward_port (
+  id SERIAL PRIMARY KEY,
+  forward_id INTEGER NOT NULL,
+  node_id INTEGER NOT NULL,
+  port INTEGER NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS node (
+  id SERIAL PRIMARY KEY,
+  name VARCHAR(100) NOT NULL,
+  secret VARCHAR(100) NOT NULL,
+  server_ip VARCHAR(100) NOT NULL,
+  server_ip_v4 VARCHAR(100),
+  server_ip_v6 VARCHAR(100),
+  port TEXT NOT NULL,
+  interface_name VARCHAR(200),
+  version VARCHAR(100),
+  http INTEGER NOT NULL DEFAULT 0,
+  tls INTEGER NOT NULL DEFAULT 0,
+  socks INTEGER NOT NULL DEFAULT 0,
+  created_time BIGINT NOT NULL,
+  updated_time BIGINT,
+  status INTEGER NOT NULL,
+  tcp_listen_addr VARCHAR(100) NOT NULL DEFAULT '[::]',
+  udp_listen_addr VARCHAR(100) NOT NULL DEFAULT '[::]',
+  inx INTEGER NOT NULL DEFAULT 0,
+  is_remote INTEGER DEFAULT 0,
+  remote_url TEXT,
+  remote_token TEXT,
+  remote_config TEXT
+);
+
+CREATE TABLE IF NOT EXISTS speed_limit (
+  id SERIAL PRIMARY KEY,
+  name VARCHAR(100) NOT NULL,
+  speed INTEGER NOT NULL,
+  tunnel_id INTEGER NOT NULL,
+  tunnel_name VARCHAR(100) NOT NULL,
+  created_time BIGINT NOT NULL,
+  updated_time BIGINT,
+  status INTEGER NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS statistics_flow (
+  id SERIAL PRIMARY KEY,
+  user_id INTEGER NOT NULL,
+  flow BIGINT NOT NULL,
+  total_flow BIGINT NOT NULL,
+  time VARCHAR(100) NOT NULL,
+  created_time BIGINT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS tunnel (
+  id SERIAL PRIMARY KEY,
+  name VARCHAR(100) NOT NULL,
+  traffic_ratio DOUBLE PRECISION NOT NULL DEFAULT 1.0,
+  type INTEGER NOT NULL,
+  protocol VARCHAR(10) NOT NULL DEFAULT 'tls',
+  flow BIGINT NOT NULL,
+  created_time BIGINT NOT NULL,
+  updated_time BIGINT NOT NULL,
+  status INTEGER NOT NULL,
+  in_ip TEXT,
+  inx INTEGER NOT NULL DEFAULT 0
+);
+
+CREATE TABLE IF NOT EXISTS chain_tunnel (
+    id SERIAL PRIMARY KEY,
+    tunnel_id INTEGER NOT NULL,
+    chain_type VARCHAR(10) NOT NULL,
+    node_id INTEGER NOT NULL,
+    port INTEGER,
+    strategy VARCHAR(10),
+    inx INTEGER,
+    protocol VARCHAR(10)
+);
+
+CREATE TABLE IF NOT EXISTS "user" (
+  id SERIAL PRIMARY KEY,
+  "user" VARCHAR(100) NOT NULL,
+  pwd VARCHAR(100) NOT NULL,
+  role_id INTEGER NOT NULL,
+  exp_time BIGINT NOT NULL,
+  flow BIGINT NOT NULL,
+  in_flow BIGINT NOT NULL DEFAULT 0,
+  out_flow BIGINT NOT NULL DEFAULT 0,
+  flow_reset_time BIGINT NOT NULL,
+  num INTEGER NOT NULL,
+  created_time BIGINT NOT NULL,
+  updated_time BIGINT,
+  status INTEGER NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS user_tunnel (
+  id SERIAL PRIMARY KEY,
+  user_id INTEGER NOT NULL,
+  tunnel_id INTEGER NOT NULL,
+  speed_id INTEGER,
+  num INTEGER NOT NULL,
+  flow BIGINT NOT NULL,
+  in_flow BIGINT NOT NULL DEFAULT 0,
+  out_flow BIGINT NOT NULL DEFAULT 0,
+  flow_reset_time BIGINT NOT NULL,
+  exp_time BIGINT NOT NULL,
+  status INTEGER NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS tunnel_group (
+  id SERIAL PRIMARY KEY,
+  name VARCHAR(100) NOT NULL,
+  created_time BIGINT NOT NULL,
+  updated_time BIGINT NOT NULL,
+  status INTEGER NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS user_group (
+  id SERIAL PRIMARY KEY,
+  name VARCHAR(100) NOT NULL,
+  created_time BIGINT NOT NULL,
+  updated_time BIGINT NOT NULL,
+  status INTEGER NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS tunnel_group_tunnel (
+  id SERIAL PRIMARY KEY,
+  tunnel_group_id INTEGER NOT NULL,
+  tunnel_id INTEGER NOT NULL,
+  created_time BIGINT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS user_group_user (
+  id SERIAL PRIMARY KEY,
+  user_group_id INTEGER NOT NULL,
+  user_id INTEGER NOT NULL,
+  created_time BIGINT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS group_permission (
+  id SERIAL PRIMARY KEY,
+  user_group_id INTEGER NOT NULL,
+  tunnel_group_id INTEGER NOT NULL,
+  created_time BIGINT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS group_permission_grant (
+  id SERIAL PRIMARY KEY,
+  user_group_id INTEGER NOT NULL,
+  tunnel_group_id INTEGER NOT NULL,
+  user_tunnel_id INTEGER NOT NULL,
+  created_by_group INTEGER NOT NULL DEFAULT 0,
+  created_time BIGINT NOT NULL
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_tunnel_group_name ON tunnel_group(name);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_user_group_name ON user_group(name);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_tunnel_group_tunnel_unique ON tunnel_group_tunnel(tunnel_group_id, tunnel_id);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_user_group_user_unique ON user_group_user(user_group_id, user_id);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_group_permission_unique ON group_permission(user_group_id, tunnel_group_id);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_group_permission_grant_unique ON group_permission_grant(user_group_id, tunnel_group_id, user_tunnel_id);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_user_tunnel_unique ON user_tunnel(user_id, tunnel_id);
+
+CREATE TABLE IF NOT EXISTS vite_config (
+  id SERIAL PRIMARY KEY,
+  name VARCHAR(200) NOT NULL UNIQUE,
+  value VARCHAR(200) NOT NULL,
+  time BIGINT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS peer_share (
+    id SERIAL PRIMARY KEY,
+    name TEXT NOT NULL,
+    node_id INTEGER NOT NULL,
+    token TEXT NOT NULL UNIQUE,
+    max_bandwidth INTEGER DEFAULT 0,
+    expiry_time BIGINT DEFAULT 0,
+    port_range_start INTEGER DEFAULT 0,
+    port_range_end INTEGER DEFAULT 0,
+    current_flow BIGINT DEFAULT 0,
+    is_active INTEGER DEFAULT 1,
+    created_time BIGINT NOT NULL,
+    updated_time BIGINT NOT NULL,
+    allowed_domains TEXT DEFAULT '',
+    allowed_ips TEXT DEFAULT ''
+);
+
+CREATE TABLE IF NOT EXISTS peer_share_runtime (
+    id SERIAL PRIMARY KEY,
+    share_id INTEGER NOT NULL,
+    node_id INTEGER NOT NULL,
+    reservation_id TEXT NOT NULL UNIQUE,
+    resource_key TEXT NOT NULL UNIQUE,
+    binding_id TEXT NOT NULL DEFAULT '',
+    role TEXT NOT NULL DEFAULT '',
+    chain_name TEXT NOT NULL DEFAULT '',
+    service_name TEXT NOT NULL DEFAULT '',
+    protocol TEXT NOT NULL DEFAULT 'tls',
+    strategy TEXT NOT NULL DEFAULT 'round',
+    port INTEGER NOT NULL DEFAULT 0,
+    target TEXT NOT NULL DEFAULT '',
+    applied INTEGER NOT NULL DEFAULT 0,
+    status INTEGER NOT NULL DEFAULT 1,
+    created_time BIGINT NOT NULL,
+    updated_time BIGINT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_peer_share_runtime_share_node_status ON peer_share_runtime(share_id, node_id, status);
+CREATE INDEX IF NOT EXISTS idx_peer_share_runtime_binding_id ON peer_share_runtime(binding_id);
+
+CREATE TABLE IF NOT EXISTS federation_tunnel_binding (
+    id SERIAL PRIMARY KEY,
+    tunnel_id INTEGER NOT NULL,
+    node_id INTEGER NOT NULL,
+    chain_type INTEGER NOT NULL,
+    hop_inx INTEGER NOT NULL DEFAULT 0,
+    remote_url TEXT NOT NULL,
+    resource_key TEXT NOT NULL UNIQUE,
+    remote_binding_id TEXT NOT NULL,
+    allocated_port INTEGER NOT NULL,
+    status INTEGER NOT NULL DEFAULT 1,
+    created_time BIGINT NOT NULL,
+    updated_time BIGINT NOT NULL
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_federation_tunnel_binding_unique ON federation_tunnel_binding(tunnel_id, node_id, chain_type, hop_inx);
+CREATE INDEX IF NOT EXISTS idx_federation_tunnel_binding_tunnel ON federation_tunnel_binding(tunnel_id, status);

go-backend/internal/store/sqlite/repository.go

@@ -12,6 +12,9 @@ import (
 	"strings"
 	"time"
 
+	_ "github.com/jackc/pgx/v5/stdlib"
+	"go-backend/internal/store"
+	pgstore "go-backend/internal/store/postgres"
 	_ "modernc.org/sqlite"
 )
 
@@ -22,10 +25,10 @@ var embeddedSchema string
 var embeddedSeedData string
 
 type Repository struct {
-	db *sql.DB
+	db *store.DB
 }
 
-func (r *Repository) DB() *sql.DB {
+func (r *Repository) DB() *store.DB {
 	if r == nil {
 		return nil
 	}
@@ -165,17 +168,53 @@ func Open(path string) (*Repository, error) {
 		return nil, err
 	}
 
-	db, err := sql.Open("sqlite", path)
+	// Use _pragma DSN parameters so every connection from the pool gets
+	// the same settings (busy_timeout and synchronous are per-connection).
+	dsn := "file:" + path +
+		"?_pragma=busy_timeout(5000)" +
+		"&_pragma=journal_mode(WAL)" +
+		"&_pragma=synchronous(NORMAL)"
+	raw, err := sql.Open("sqlite", dsn)
 	if err != nil {
 		return nil, err
 	}
+	db := store.Wrap(raw, store.DialectSQLite)
 
 	if err := db.Ping(); err != nil {
 		_ = db.Close()
 		return nil, err
 	}
 
-	if err := bootstrapSchema(db); err != nil {
+	if err := bootstrapSchema(db, embeddedSchema, embeddedSeedData); err != nil {
+		_ = db.Close()
+		return nil, err
+	}
+
+	if err := migrateSchema(db); err != nil {
+		_ = db.Close()
+		return nil, err
+	}
+
+	return &Repository{db: db}, nil
+}
+
+func OpenPostgres(dsn string) (*Repository, error) {
+	if strings.TrimSpace(dsn) == "" {
+		return nil, fmt.Errorf("empty postgres dsn")
+	}
+
+	raw, err := sql.Open("pgx", dsn)
+	if err != nil {
+		return nil, err
+	}
+	db := store.Wrap(raw, store.DialectPostgres)
+
+	if err := db.Ping(); err != nil {
+		_ = db.Close()
+		return nil, err
+	}
+
+	if err := bootstrapSchema(db, pgstore.EmbeddedSchema, pgstore.EmbeddedSeedData); err != nil {
 		_ = db.Close()
 		return nil, err
 	}
@@ -361,7 +400,7 @@ func (r *Repository) GetUserPackageForwards(userID int64) ([]UserForwardDetail,
 	}
 
 	rows, err := r.db.Query(`
-		SELECT f.id, f.name, f.tunnel_id, t.name, f.remote_addr, f.in_flow, f.out_flow, f.status, f.created_time
+		SELECT f.id, f.name, f.tunnel_id, COALESCE(t.name, ''), f.remote_addr, f.in_flow, f.out_flow, f.status, f.created_time
 		FROM forward f
 		LEFT JOIN tunnel t ON t.id = f.tunnel_id
 		WHERE f.user_id = ?
@@ -680,7 +719,7 @@ func (r *Repository) ListForwards() ([]map[string]interface{}, error) {
 	}
 
 	rows, err := r.db.Query(`
-		SELECT f.id, f.user_id, f.user_name, f.name, f.tunnel_id, t.name, f.remote_addr, f.strategy,
+		SELECT f.id, f.user_id, f.user_name, f.name, f.tunnel_id, COALESCE(t.name, ''), f.remote_addr, f.strategy,
 		       f.in_flow, f.out_flow, f.created_time, f.status, f.inx
 		FROM forward f
 		LEFT JOIN tunnel t ON t.id = f.tunnel_id
@@ -858,9 +897,9 @@ func (r *Repository) ListTunnels() ([]map[string]interface{}, error) {
 	}
 
 	chainRows, err := r.db.Query(`
-		SELECT tunnel_id, chain_type, node_id, protocol, strategy, COALESCE(inx, 0)
+		SELECT tunnel_id, CAST(chain_type AS INTEGER), node_id, protocol, strategy, COALESCE(inx, 0)
 		FROM chain_tunnel
-		ORDER BY tunnel_id ASC, chain_type ASC, inx ASC, id ASC
+		ORDER BY tunnel_id ASC, CAST(chain_type AS INTEGER) ASC, inx ASC, id ASC
 	`)
 	if err != nil {
 		return nil, err
@@ -1141,7 +1180,7 @@ func nullableForwardIngress(v string) interface{} {
 	return v
 }
 
-func resolveForwardIngress(db *sql.DB, forwardID int64, tunnelID int64) (string, sql.NullInt64, error) {
+func resolveForwardIngress(db *store.DB, forwardID int64, tunnelID int64) (string, sql.NullInt64, error) {
 	var tunnelInIP sql.NullString
 	if err := db.QueryRow(`SELECT in_ip FROM tunnel WHERE id = ? LIMIT 1`, tunnelID).Scan(&tunnelInIP); err != nil {
 		if !errors.Is(err, sql.ErrNoRows) {
@@ -1243,33 +1282,54 @@ func ensureParentDir(dbPath string) error {
 	return osMkdirAll(dir)
 }
 
-func bootstrapSchema(db *sql.DB) error {
+func bootstrapSchema(db *store.DB, schemaSQL, seedSQL string) error {
 	if db == nil {
 		return errors.New("nil db")
 	}
 
-	if _, err := db.Exec(embeddedSchema); err != nil {
+	if _, err := db.Exec(schemaSQL); err != nil {
 		return fmt.Errorf("apply schema.sql: %w", err)
 	}
 
-	if _, err := db.Exec(embeddedSeedData); err != nil {
+	if _, err := db.Exec(seedSQL); err != nil {
 		return fmt.Errorf("apply data.sql: %w", err)
 	}
 	return nil
 }
 
-func migrateSchema(db *sql.DB) error {
+const currentSchemaVersion = 1
+
+func getSchemaVersion(db *store.DB) int {
+	_, _ = db.Exec(`CREATE TABLE IF NOT EXISTS schema_version (version INTEGER NOT NULL DEFAULT 0)`)
+	var v int
+	if err := db.QueryRow(`SELECT version FROM schema_version LIMIT 1`).Scan(&v); err != nil {
+		_, _ = db.Exec(`INSERT INTO schema_version(version) VALUES(0)`)
+		return 0
+	}
+	return v
+}
+
+func setSchemaVersion(db *store.DB, v int) {
+	_, _ = db.Exec(`UPDATE schema_version SET version = ?`, v)
+}
+
+func migrateSchema(db *store.DB) error {
 	if db == nil {
 		return errors.New("nil db")
 	}
 
+	ver := getSchemaVersion(db)
+	if ver >= currentSchemaVersion {
+		return nil
+	}
+
 	ensureColumn := func(table, col, typ string) {
 		var dummy interface{}
 		err := db.QueryRow(fmt.Sprintf("SELECT %s FROM %s LIMIT 1", col, table)).Scan(&dummy)
 		if err == nil || errors.Is(err, sql.ErrNoRows) {
 			return
 		}
-		if strings.Contains(err.Error(), "no such column") {
+		if isMissingColumnError(db.Dialect(), err) {
 			if _, alterErr := db.Exec(fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s", table, col, typ)); alterErr != nil {
 				log.Printf("failed to add column %s to %s: %v", col, table, alterErr)
 			}
@@ -1306,9 +1366,177 @@ func migrateSchema(db *sql.DB) error {
 			ensureColumn(table, col, typ)
 		}
 	}
+
+	if db.Dialect() == store.DialectPostgres {
+		if err := ensurePostgresIDDefaults(db); err != nil {
+			return err
+		}
+	}
+	setSchemaVersion(db, currentSchemaVersion)
 	return nil
 }
 
+func ensurePostgresIDDefaults(db *store.DB) error {
+	rows, err := db.Query(`
+		SELECT c.table_schema, c.table_name
+		FROM information_schema.table_constraints tc
+		JOIN information_schema.key_column_usage kcu
+		  ON tc.constraint_name = kcu.constraint_name
+		 AND tc.table_schema = kcu.table_schema
+		JOIN information_schema.columns c
+		  ON c.table_schema = kcu.table_schema
+		 AND c.table_name = kcu.table_name
+		 AND c.column_name = kcu.column_name
+		WHERE tc.constraint_type = 'PRIMARY KEY'
+		  AND kcu.column_name = 'id'
+		  AND c.data_type IN ('integer', 'bigint')
+		  AND c.is_identity = 'NO'
+		  AND c.table_schema = current_schema()
+		ORDER BY c.table_name ASC
+	`)
+	if err != nil {
+		return fmt.Errorf("discover postgres id columns: %w", err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var schemaName string
+		var tableName string
+		if err := rows.Scan(&schemaName, &tableName); err != nil {
+			return fmt.Errorf("scan postgres id table row: %w", err)
+		}
+		if err := ensurePostgresTableIDDefault(db, schemaName, tableName); err != nil {
+			return fmt.Errorf("repair %s.%s id default: %w", schemaName, tableName, err)
+		}
+	}
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("iterate postgres id tables: %w", err)
+	}
+
+	return nil
+}
+
+func ensurePostgresTableIDDefault(db *store.DB, schemaName, tableName string) error {
+	var defaultExpr sql.NullString
+	if err := db.QueryRow(`
+		SELECT column_default
+		FROM information_schema.columns
+		WHERE table_schema = ?
+		  AND table_name = ?
+		  AND column_name = 'id'
+		LIMIT 1
+	`, schemaName, tableName).Scan(&defaultExpr); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil
+		}
+		return err
+	}
+
+	hasNextvalDefault := defaultExpr.Valid && strings.Contains(strings.ToLower(defaultExpr.String), "nextval(")
+
+	var serialSeq sql.NullString
+	if err := db.QueryRow(`
+		SELECT pg_get_serial_sequence(quote_ident(?) || '.' || quote_ident(?), 'id')
+	`, schemaName, tableName).Scan(&serialSeq); err != nil {
+		return err
+	}
+
+	seqRef := strings.TrimSpace(serialSeq.String)
+	if seqRef == "" && hasNextvalDefault {
+		seqRef = extractNextvalRegclass(defaultExpr.String)
+	}
+
+	if !hasNextvalDefault || seqRef == "" {
+		seqName := tableName + "_id_seq"
+		if _, err := db.Exec(fmt.Sprintf("CREATE SEQUENCE IF NOT EXISTS %s.%s", quoteSQLIdentifier(schemaName), quoteSQLIdentifier(seqName))); err != nil {
+			return err
+		}
+
+		seqRef = schemaName + "." + seqName
+		if _, err := db.Exec(fmt.Sprintf(
+			"ALTER TABLE %s.%s ALTER COLUMN id SET DEFAULT nextval(%s::regclass)",
+			quoteSQLIdentifier(schemaName),
+			quoteSQLIdentifier(tableName),
+			quoteSQLLiteral(seqRef),
+		)); err != nil {
+			return err
+		}
+
+		if _, err := db.Exec(fmt.Sprintf(
+			"ALTER SEQUENCE %s.%s OWNED BY %s.%s.id",
+			quoteSQLIdentifier(schemaName),
+			quoteSQLIdentifier(seqName),
+			quoteSQLIdentifier(schemaName),
+			quoteSQLIdentifier(tableName),
+		)); err != nil {
+			return err
+		}
+	}
+
+	return syncPostgresTableIDSequence(db, schemaName, tableName, seqRef)
+}
+
+func syncPostgresTableIDSequence(db *store.DB, schemaName, tableName, seqRef string) error {
+	var maxID int64
+	if err := db.QueryRow(fmt.Sprintf(
+		"SELECT COALESCE(MAX(id), 0) FROM %s.%s",
+		quoteSQLIdentifier(schemaName),
+		quoteSQLIdentifier(tableName),
+	)).Scan(&maxID); err != nil {
+		return err
+	}
+
+	setVal := maxID
+	isCalled := true
+	if maxID <= 0 {
+		setVal = 1
+		isCalled = false
+	}
+
+	if _, err := db.Exec(`SELECT setval(?::regclass, ?, ?)`, seqRef, setVal, isCalled); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func extractNextvalRegclass(defaultExpr string) string {
+	nextvalIdx := strings.Index(strings.ToLower(defaultExpr), "nextval(")
+	if nextvalIdx < 0 {
+		return ""
+	}
+	expr := defaultExpr[nextvalIdx:]
+	firstQuote := strings.Index(expr, "'")
+	if firstQuote < 0 {
+		return ""
+	}
+	expr = expr[firstQuote+1:]
+	secondQuote := strings.Index(expr, "'")
+	if secondQuote < 0 {
+		return ""
+	}
+	return strings.TrimSpace(expr[:secondQuote])
+}
+
+func quoteSQLIdentifier(ident string) string {
+	return `"` + strings.ReplaceAll(ident, `"`, `""`) + `"`
+}
+
+func quoteSQLLiteral(value string) string {
+	return "'" + strings.ReplaceAll(value, "'", "''") + "'"
+}
+
+func isMissingColumnError(dialect store.Dialect, err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := strings.ToLower(err.Error())
+	if dialect == store.DialectPostgres {
+		return strings.Contains(msg, "column") && strings.Contains(msg, "does not exist")
+	}
+	return strings.Contains(msg, "no such column")
+}
+
 func (r *Repository) CreatePeerShare(share *PeerShare) error {
 	if r == nil || r.db == nil {
 		return errors.New("repository not initialized")