feat: 实现 SQLite 迁移备份、完整表结构与简单写入队列

cmd/nixmsg/serve.go

@@ -35,16 +35,12 @@ func runServe(ctx context.Context, cfg config.Config) error {
 		return fmt.Errorf("mkdir data_dir: %w", err)
 	}
 
-	db, err := store.OpenWriter(cfg.DataDir, cfg.SQLiteSynchronous)
+	db, err := store.Open(cfg.DataDir, cfg.SQLiteSynchronous)
 	if err != nil {
 		return err
 	}
 	defer func() { _ = db.Close() }()
 
-	if migErr := store.Migrate(db); migErr != nil {
-		return migErr
-	}
-
 	mux := http.NewServeMux()
 	mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
 		w.WriteHeader(http.StatusOK)

docs/DEVIATIONS.md

@@ -73,6 +73,30 @@
    - 备选方案：协议包只做编解码，校验留给各 app 模块。  
    - 影响：服务端应复用本包 `Validate`，避免重复规则。
 
+### T0.3 2026-09-30
+
+1. **完整表放在 0002，不改已发布的 0001**  
+   - 原条款：TASKS T0.3 / 4.2「`0001_init.sql` 包含第 7.7 节全部表」；T0.1 偏差曾写「T0.3 需替换 0001 正文」。  
+   - 实际做法：保留 `0001_init.sql` 为 `SELECT 1;`；新增 `0002_schema.sql` 写入 DEVELOPMENT 7.7 全部业务表与索引（含 `api_tokens`、`settings`、`session_hash` 等）。`schema_migrations` 仍由迁移执行器 `CREATE TABLE IF NOT EXISTS` 维护，不放入 0002。  
+   - 原因：T0.1 的 0001 可能已记入已有库的 `schema_migrations`；改写已发布迁移语义会导致「版本已应用但表不存在」。  
+   - 备选方案：对未迁移库特殊检测并改写 0001（复杂且易错）。  
+   - 影响：新库会有版本 1+2 两行；与 TASKS「表在 0001」字面不一致，与「不改已发布迁移」一致。
+
+2. **写入队列先做一操作一事务**  
+   - 原条款：DEVELOPMENT 7.2 合并提交（最多 256 或凑满 2ms，SAVEPOINT）；TASKS T0.3 允许简单实现，P2 换合并。  
+   - 实际做法：`store.Queue` 用互斥锁串行，每请求一个事务；注释与本条标明 P2 再改为写 goroutine 合并提交。  
+   - 原因：本任务范围；合并留给平台 P2。  
+   - 备选方案：T0.3 直接做合并（抢 P2 范围）。  
+   - 影响：高并发写入落盘次数偏多，正式压测前需完成 P2。
+
+3. **空库不备份；仅已有 db 文件且有未应用版本时 VACUUM INTO**  
+   - 原条款：DEVELOPMENT 7.7「有未应用版本时先 VACUUM INTO」；未区分空库。  
+   - 实际做法：`Open` 在打开前检查 `nixmsg.db` 是否已存在；不存在则跳过备份；存在且有 pending 则写入 `<data_dir>/backup/pre-migrate-<UTC时间>.db`。迁移失败返回错误，不自动从备份恢复。  
+   - 原因：空库备份无意义；失败退出与文档一致，恢复交给运维。  
+   - 备选方案：失败时自动还原备份再退出。  
+   - 影响：与任务说明一致；运维需知备份路径。
+
+
 ## 平台 P
 
 暂无。

internal/store/db.go

@@ -10,29 +10,105 @@ import (
 	_ "modernc.org/sqlite"
 )
 
-// OpenWriter 按 DEVELOPMENT 7.7 打开写连接（带 _txlock=immediate）。
+const dbFileName = "nixmsg.db"
+
+// DB 持有读写连接与写入队列。
+type DB struct {
+	Write *sql.DB
+	Read  *sql.DB
+	Queue *Queue
+}
+
+// Open 打开数据目录下的库：写连接、读连接池、迁移，并启动简单写入队列。
+// 若 nixmsg.db 尚不存在则跳过迁移前备份。
+func Open(dataDir, synchronous string) (*DB, error) {
+	if err := os.MkdirAll(dataDir, 0o755); err != nil {
+		return nil, fmt.Errorf("mkdir data_dir: %w", err)
+	}
+	dbPath := filepath.Join(dataDir, dbFileName)
+	existed, err := fileExists(dbPath)
+	if err != nil {
+		return nil, err
+	}
+
+	write, err := OpenWriter(dataDir, synchronous)
+	if err != nil {
+		return nil, err
+	}
+	if migErr := Migrate(write, dataDir, existed); migErr != nil {
+		_ = write.Close()
+		return nil, migErr
+	}
+	read, err := OpenReader(dataDir, synchronous)
+	if err != nil {
+		_ = write.Close()
+		return nil, err
+	}
+	q := NewQueue(write)
+	return &DB{Write: write, Read: read, Queue: q}, nil
+}
+
+// Close 停止写入队列并关闭连接。
+func (d *DB) Close() error {
+	var first error
+	if d.Queue != nil {
+		if err := d.Queue.Close(); err != nil {
+			first = err
+		}
+	}
+	if d.Read != nil {
+		if err := d.Read.Close(); err != nil && first == nil {
+			first = err
+		}
+	}
+	if d.Write != nil {
+		if err := d.Write.Close(); err != nil && first == nil {
+			first = err
+		}
+	}
+	return first
+}
+
+// OpenWriter 按 DEVELOPMENT 7.7 打开写连接（带 _txlock=immediate，MaxOpenConns=1）。
 func OpenWriter(dataDir, synchronous string) (*sql.DB, error) {
 	if err := os.MkdirAll(dataDir, 0o755); err != nil {
 		return nil, fmt.Errorf("mkdir data_dir: %w", err)
 	}
-	dsn, err := writeDSN(dataDir, synchronous)
+	dsn, err := buildDSN(dataDir, synchronous, true)
 	if err != nil {
 		return nil, err
 	}
 	db, err := sql.Open("sqlite", dsn)
 	if err != nil {
-		return nil, fmt.Errorf("open sqlite: %w", err)
+		return nil, fmt.Errorf("open sqlite writer: %w", err)
 	}
 	db.SetMaxOpenConns(1)
 	db.SetMaxIdleConns(1)
 	if err := db.Ping(); err != nil {
 		_ = db.Close()
-		return nil, fmt.Errorf("ping sqlite: %w", err)
+		return nil, fmt.Errorf("ping sqlite writer: %w", err)
 	}
 	return db, nil
 }
 
-func writeDSN(dataDir, synchronous string) (string, error) {
+// OpenReader 打开读连接池（DSN 不含 _txlock=immediate）。
+func OpenReader(dataDir, synchronous string) (*sql.DB, error) {
+	dsn, err := buildDSN(dataDir, synchronous, false)
+	if err != nil {
+		return nil, err
+	}
+	db, err := sql.Open("sqlite", dsn)
+	if err != nil {
+		return nil, fmt.Errorf("open sqlite reader: %w", err)
+	}
+	if err := db.Ping(); err != nil {
+		_ = db.Close()
+		return nil, fmt.Errorf("ping sqlite reader: %w", err)
+	}
+	return db, nil
+}
+
+func buildDSN(dataDir, synchronous string, writer bool) (string, error) {
 	sync := strings.ToUpper(strings.TrimSpace(synchronous))
 	if sync == "" {
 		sync = "FULL"
@@ -40,10 +116,28 @@ func writeDSN(dataDir, synchronous string) (string, error) {
 	if sync != "FULL" && sync != "NORMAL" {
 		return "", fmt.Errorf("invalid sqlite_synchronous: %s", synchronous)
 	}
-	dbPath := filepath.ToSlash(filepath.Join(dataDir, "nixmsg.db"))
+	dbPath := filepath.ToSlash(filepath.Join(dataDir, dbFileName))
-	return fmt.Sprintf(
+	dsn := fmt.Sprintf(
-		"file:%s?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=synchronous(%s)&_pragma=foreign_keys(ON)&_pragma=secure_delete(ON)&_txlock=immediate",
+		"file:%s?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=synchronous(%s)&_pragma=foreign_keys(ON)&_pragma=secure_delete(ON)",
 		dbPath,
 		sync,
-	), nil
+	)
+	if writer {
+		dsn += "&_txlock=immediate"
+	}
+	return dsn, nil
+}
+
+func fileExists(path string) (bool, error) {
+	st, err := os.Stat(path)
+	if err == nil {
+		if st.IsDir() {
+			return false, fmt.Errorf("db path is a directory: %s", path)
+		}
+		return true, nil
+	}
+	if os.IsNotExist(err) {
+		return false, nil
+	}
+	return false, err
 }

internal/store/db_test.go

@@ -1,36 +1,165 @@
 package store
 
 import (
+	"context"
+	"database/sql"
+	"os"
 	"path/filepath"
+	"strings"
 	"testing"
+	"time"
 )
 
-func TestOpenAndMigrate(t *testing.T) {
+func TestOpenEmptyDirCreatesTables(t *testing.T) {
 	t.Parallel()
 	dir := t.TempDir()
-	db, err := OpenWriter(dir, "FULL")
+
+	db, err := Open(dir, "FULL")
 	if err != nil {
 		t.Fatal(err)
 	}
 	defer func() { _ = db.Close() }()
 
-	if migErr := Migrate(db); migErr != nil {
+	assertMigrationCount(t, db.Write, 2)
-		t.Fatal(migErr)
+	for _, table := range []string{
+		"endpoints", "settings", "talk_grants", "groups", "group_members",
+		"messages", "message_bodies", "deliveries", "receipts", "send_keys",
+		"admin_sessions", "api_tokens", "schema_migrations",
+	} {
+		assertTableExists(t, db.Write, table)
 	}
-	if migErr := Migrate(db); migErr != nil {
+	// 空目录首次启动不应产生迁移备份。
-		t.Fatal(migErr)
+	if entries, readErr := os.ReadDir(filepath.Join(dir, "backup")); readErr == nil && len(entries) > 0 {
+		t.Fatalf("unexpected backup files on empty start: %d", len(entries))
+	}
+}
+
+func TestOpenIdempotentNoRemigrate(t *testing.T) {
+	t.Parallel()
+	dir := t.TempDir()
+
+	db1, err := Open(dir, "FULL")
+	if err != nil {
+		t.Fatal(err)
+	}
+	assertMigrationCount(t, db1.Write, 2)
+	_ = db1.Close()
+
+	db2, err := Open(dir, "FULL")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = db2.Close() }()
+	assertMigrationCount(t, db2.Write, 2)
+
+	if entries, readErr := os.ReadDir(filepath.Join(dir, "backup")); readErr == nil && len(entries) > 0 {
+		t.Fatalf("idempotent reopen should not backup: %v", entries)
+	}
+}
+
+func TestMigrateBackupWhenNewVersion(t *testing.T) {
+	t.Parallel()
+	dir := t.TempDir()
+
+	// 模拟仅应用了 0001 的旧库。
+	w, err := OpenWriter(dir, "FULL")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, execErr := w.Exec(`
+CREATE TABLE schema_migrations (
+  version INTEGER PRIMARY KEY,
+  applied_at INTEGER NOT NULL
+)`); execErr != nil {
+		t.Fatal(execErr)
+	}
+	if _, execErr := w.Exec(`INSERT INTO schema_migrations(version, applied_at) VALUES(1, ?)`, time.Now().UnixMilli()); execErr != nil {
+		t.Fatal(execErr)
+	}
+	_ = w.Close()
+
+	db, err := Open(dir, "FULL")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = db.Close() }()
+
+	assertMigrationCount(t, db.Write, 2)
+	assertTableExists(t, db.Write, "endpoints")
+	assertTableExists(t, db.Write, "api_tokens")
+
+	backupDir := filepath.Join(dir, "backup")
+	entries, err := os.ReadDir(backupDir)
+	if err != nil {
+		t.Fatalf("backup dir missing: %v", err)
+	}
+	found := false
+	for _, e := range entries {
+		if strings.HasPrefix(e.Name(), "pre-migrate-") && strings.HasSuffix(e.Name(), ".db") {
+			found = true
+			info, statErr := e.Info()
+			if statErr != nil {
+				t.Fatal(statErr)
+			}
+			if info.Size() == 0 {
+				t.Fatalf("backup file empty: %s", e.Name())
+			}
+		}
+	}
+	if !found {
+		t.Fatal("expected pre-migrate-*.db backup")
+	}
+}
+
+func TestWriteQueueOneOpOneTx(t *testing.T) {
+	t.Parallel()
+	dir := t.TempDir()
+	db, err := Open(dir, "FULL")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = db.Close() }()
+
+	ctx := context.Background()
+	err = db.Queue.Do(ctx, func(tx *sql.Tx) error {
+		_, execErr := tx.Exec(
+			`INSERT INTO settings(key, value, updated_at) VALUES(?, ?, ?)`,
+			"registration_enabled", "0", time.Now().UnixMilli(),
+		)
+		return execErr
+	})
+	if err != nil {
+		t.Fatal(err)
 	}
 
-	var n int
+	var value string
-	if scanErr := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations`).Scan(&n); scanErr != nil {
+	if scanErr := db.Read.QueryRow(`SELECT value FROM settings WHERE key = ?`, "registration_enabled").Scan(&value); scanErr != nil {
 		t.Fatal(scanErr)
 	}
-	if n != 1 {
+	if value != "0" {
-		t.Fatalf("want 1 migration row, got %d", n)
+		t.Fatalf("want 0, got %q", value)
+	}
+}
+
+func assertMigrationCount(t *testing.T, db *sql.DB, want int) {
+	t.Helper()
+	var n int
+	if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations`).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != want {
+		t.Fatalf("migration rows: want %d got %d", want, n)
+	}
+}
+
+func assertTableExists(t *testing.T, db *sql.DB, name string) {
+	t.Helper()
+	var got string
+	err := db.QueryRow(
+		`SELECT name FROM sqlite_master WHERE type='table' AND name=?`,
+		name,
+	).Scan(&got)
+	if err != nil {
+		t.Fatalf("table %s missing: %v", name, err)
 	}
-	nested, openErr := OpenWriter(filepath.Join(dir, "nested"), "NORMAL")
-	if openErr != nil {
-		t.Fatal(openErr)
-	}
-	_ = nested.Close()
 }

internal/store/migrate.go

@@ -5,6 +5,8 @@ import (
 	"embed"
 	"fmt"
 	"io/fs"
+	"os"
+	"path/filepath"
 	"sort"
 	"strconv"
 	"strings"
@@ -14,8 +16,11 @@ import (
 //go:embed migrations/*.sql
 var migrationFS embed.FS
 
-// Migrate 应用尚未执行的迁移。T0.1 仅为空执行器加 0001 占位；完整备份与表结构见 T0.3。
+// Migrate 应用尚未执行的嵌入迁移。
-func Migrate(db *sql.DB) error {
+// 若 dbExisted 为 true（调用 Open/Migrate 前已有 nixmsg.db）且存在未应用版本，
+// 先 VACUUM INTO <data_dir>/backup/pre-migrate-<时间>.db，再迁移。
+// 任一版本失败则返回错误，调用方不得继续带半新半旧库提供服务。
+func Migrate(db *sql.DB, dataDir string, dbExisted bool) error {
 	if _, err := db.Exec(`
 CREATE TABLE IF NOT EXISTS schema_migrations (
   version INTEGER PRIMARY KEY,
@@ -24,9 +29,38 @@ CREATE TABLE IF NOT EXISTS schema_migrations (
 		return fmt.Errorf("ensure schema_migrations: %w", err)
 	}
 
+	pending, err := pendingMigrations(db)
+	if err != nil {
+		return err
+	}
+	if len(pending) == 0 {
+		return nil
+	}
+
+	if dbExisted {
+		if err := backupBeforeMigrate(db, dataDir); err != nil {
+			return err
+		}
+	}
+
+	for _, m := range pending {
+		if err := applyMigration(db, m); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+type migrationFile struct {
+	version int
+	name    string
+	body    string
+}
+
+func pendingMigrations(db *sql.DB) ([]migrationFile, error) {
 	entries, err := fs.ReadDir(migrationFS, "migrations")
 	if err != nil {
-		return fmt.Errorf("read migrations: %w", err)
+		return nil, fmt.Errorf("read migrations: %w", err)
 	}
 	var names []string
 	for _, e := range entries {
@@ -37,10 +71,11 @@ CREATE TABLE IF NOT EXISTS schema_migrations (
 	}
 	sort.Strings(names)
 
+	var pending []migrationFile
 	for _, name := range names {
 		version, err := parseMigrationVersion(name)
 		if err != nil {
-			return err
+			return nil, err
 		}
 		var exists int
 		err = db.QueryRow(`SELECT 1 FROM schema_migrations WHERE version = ?`, version).Scan(&exists)
@@ -48,35 +83,57 @@ CREATE TABLE IF NOT EXISTS schema_migrations (
 			continue
 		}
 		if err != sql.ErrNoRows {
-			return fmt.Errorf("check migration %d: %w", version, err)
+			return nil, fmt.Errorf("check migration %d: %w", version, err)
 		}
-
 		body, err := migrationFS.ReadFile("migrations/" + name)
 		if err != nil {
-			return fmt.Errorf("read migration %s: %w", name, err)
+			return nil, fmt.Errorf("read migration %s: %w", name, err)
 		}
-		sqlText := strings.TrimSpace(string(body))
+		pending = append(pending, migrationFile{
+			version: version,
+			name:    name,
+			body:    strings.TrimSpace(string(body)),
+		})
+	}
+	return pending, nil
+}
+
+func backupBeforeMigrate(db *sql.DB, dataDir string) error {
+	backupDir := filepath.Join(dataDir, "backup")
+	if err := os.MkdirAll(backupDir, 0o755); err != nil {
+		return fmt.Errorf("mkdir backup: %w", err)
+	}
+	stamp := time.Now().UTC().Format("20060102T150405")
+	backupPath := filepath.Join(backupDir, "pre-migrate-"+stamp+".db")
+	// SQLite VACUUM INTO 需要字面量路径；统一用斜杠，并对单引号转义。
+	quoted := strings.ReplaceAll(filepath.ToSlash(backupPath), "'", "''")
+	if _, err := db.Exec("VACUUM INTO '" + quoted + "'"); err != nil {
+		return fmt.Errorf("vacuum into backup %s: %w", backupPath, err)
+	}
+	return nil
+}
+
+func applyMigration(db *sql.DB, m migrationFile) error {
 	tx, err := db.Begin()
 	if err != nil {
-			return fmt.Errorf("begin migration %d: %w", version, err)
+		return fmt.Errorf("begin migration %d: %w", m.version, err)
 	}
-		if sqlText != "" {
+	if m.body != "" {
-			if _, err := tx.Exec(sqlText); err != nil {
+		if _, err := tx.Exec(m.body); err != nil {
 			_ = tx.Rollback()
-				return fmt.Errorf("apply migration %d: %w", version, err)
+			return fmt.Errorf("apply migration %d (%s): %w", m.version, m.name, err)
 		}
 	}
 	if _, err := tx.Exec(
 		`INSERT INTO schema_migrations(version, applied_at) VALUES(?, ?)`,
-			version,
+		m.version,
 		time.Now().UnixMilli(),
 	); err != nil {
 		_ = tx.Rollback()
-			return fmt.Errorf("record migration %d: %w", version, err)
+		return fmt.Errorf("record migration %d: %w", m.version, err)
 	}
 	if err := tx.Commit(); err != nil {
-			return fmt.Errorf("commit migration %d: %w", version, err)
+		return fmt.Errorf("commit migration %d: %w", m.version, err)
-		}
 	}
 	return nil
 }

internal/store/migrations/0002_schema.sql

@@ -0,0 +1,131 @@
+-- 完整业务表（T0.1 的 0001 仅为 SELECT 1 占位且可能已记入 schema_migrations，故放在 0002）。
+CREATE TABLE endpoints (
+  id TEXT PRIMARY KEY,
+  name TEXT NOT NULL DEFAULT '',
+  remark TEXT NOT NULL DEFAULT '',
+  source TEXT NOT NULL DEFAULT 'admin',
+  login_hash TEXT NOT NULL,
+  talk_hash TEXT,
+  talk_version INTEGER NOT NULL DEFAULT 0,
+  default_delay_ms INTEGER NOT NULL DEFAULT 0,
+  enabled INTEGER NOT NULL DEFAULT 1,
+  created_at INTEGER NOT NULL,
+  online_since INTEGER,
+  offline_since INTEGER,
+  session_hash TEXT,
+  session_issued_at INTEGER,
+  session_used_at INTEGER
+);
+
+CREATE TABLE settings (
+  key TEXT PRIMARY KEY,
+  value TEXT NOT NULL,
+  updated_at INTEGER NOT NULL
+);
+
+CREATE TABLE talk_grants (
+  sender_id TEXT NOT NULL,
+  target_id TEXT NOT NULL,
+  target_talk_version INTEGER NOT NULL,
+  kind TEXT NOT NULL,
+  created_at INTEGER NOT NULL,
+  PRIMARY KEY (sender_id, target_id)
+);
+
+CREATE TABLE groups (
+  id TEXT PRIMARY KEY,
+  name TEXT NOT NULL,
+  owner_id TEXT NOT NULL,
+  created_at INTEGER NOT NULL
+);
+
+CREATE TABLE group_members (
+  group_id TEXT NOT NULL,
+  endpoint_id TEXT NOT NULL,
+  joined_at INTEGER NOT NULL,
+  PRIMARY KEY (group_id, endpoint_id)
+);
+
+CREATE TABLE messages (
+  seq INTEGER PRIMARY KEY,
+  id TEXT NOT NULL,
+  sender_id TEXT NOT NULL,
+  dest_kind TEXT NOT NULL,
+  dest_id TEXT NOT NULL,
+  meta TEXT NOT NULL DEFAULT '{}',
+  content_type TEXT NOT NULL,
+  body_enc TEXT NOT NULL,
+  send_at INTEGER NOT NULL,
+  keep INTEGER NOT NULL,
+  ttl_seconds INTEGER NOT NULL DEFAULT 0,
+  receipt INTEGER NOT NULL,
+  state TEXT NOT NULL,
+  reason TEXT NOT NULL DEFAULT '',
+  created_at INTEGER NOT NULL,
+  UNIQUE (sender_id, id)
+);
+
+CREATE TABLE message_bodies (
+  seq INTEGER PRIMARY KEY REFERENCES messages(seq) ON DELETE CASCADE,
+  body BLOB NOT NULL
+);
+
+CREATE TABLE deliveries (
+  seq INTEGER NOT NULL REFERENCES messages(seq) ON DELETE CASCADE,
+  endpoint_id TEXT NOT NULL,
+  send_at INTEGER NOT NULL,
+  keep INTEGER NOT NULL,
+  state TEXT NOT NULL,
+  reason TEXT NOT NULL DEFAULT '',
+  expire_at INTEGER,
+  pushed_conn TEXT,
+  pushed_at INTEGER,
+  attempts INTEGER NOT NULL DEFAULT 0,
+  updated_at INTEGER NOT NULL,
+  PRIMARY KEY (seq, endpoint_id)
+);
+
+CREATE TABLE receipts (
+  receipt_id INTEGER PRIMARY KEY,
+  sender_id TEXT NOT NULL,
+  msg_id TEXT NOT NULL,
+  endpoint_id TEXT NOT NULL DEFAULT '',
+  state TEXT NOT NULL,
+  reason TEXT NOT NULL DEFAULT '',
+  created_at INTEGER NOT NULL,
+  acked INTEGER NOT NULL DEFAULT 0
+);
+
+CREATE TABLE send_keys (
+  sender_id TEXT NOT NULL,
+  msg_id TEXT NOT NULL,
+  request_sha256 BLOB NOT NULL,
+  created_at INTEGER NOT NULL,
+  PRIMARY KEY (sender_id, msg_id)
+);
+
+CREATE TABLE admin_sessions (
+  token_hash TEXT PRIMARY KEY,
+  created_at INTEGER NOT NULL,
+  expires_at INTEGER NOT NULL
+);
+
+CREATE TABLE api_tokens (
+  id TEXT PRIMARY KEY,
+  name TEXT NOT NULL,
+  token_hash TEXT NOT NULL UNIQUE,
+  enabled INTEGER NOT NULL DEFAULT 1,
+  created_at INTEGER NOT NULL,
+  last_used_at INTEGER
+);
+
+CREATE INDEX idx_messages_due ON messages(state, send_at);
+CREATE INDEX idx_messages_dest ON messages(dest_kind, dest_id, state);
+CREATE INDEX idx_messages_created ON messages(created_at);
+CREATE INDEX idx_messages_sender ON messages(sender_id, created_at);
+CREATE INDEX idx_messages_sender_state ON messages(sender_id, state);
+CREATE INDEX idx_deliveries_outbox ON deliveries(endpoint_id, state, send_at, seq);
+CREATE INDEX idx_deliveries_expire ON deliveries(state, expire_at);
+CREATE INDEX idx_receipts_outbox ON receipts(sender_id, acked, receipt_id);
+CREATE INDEX idx_talk_grants_target ON talk_grants(target_id);
+CREATE INDEX idx_group_members_endpoint ON group_members(endpoint_id);

internal/store/queue.go

@@ -0,0 +1,62 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"sync"
+)
+
+// ErrQueueClosed 表示写入队列已关闭。
+var ErrQueueClosed = errors.New("store: write queue closed")
+
+// WriteFunc 在单个写事务中执行的操作。
+type WriteFunc func(tx *sql.Tx) error
+
+// Queue 写入队列：提交一个写操作并拿到结果。
+//
+// 本任务（T0.3）实现为互斥串行的一操作一事务，不做合并；DEVELOPMENT 7.2
+// 要求的写 goroutine 合并提交（最多 256 个或凑满 2ms、SAVEPOINT 隔离失败）留给 P2。
+type Queue struct {
+	db *sql.DB
+
+	mu     sync.Mutex
+	closed bool
+}
+
+// NewQueue 创建简单写入队列（一操作一事务）。
+func NewQueue(db *sql.DB) *Queue {
+	return &Queue{db: db}
+}
+
+// Do 提交写操作并等待提交结果。
+func (q *Queue) Do(ctx context.Context, fn WriteFunc) error {
+	if fn == nil {
+		return errors.New("store: nil write func")
+	}
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	if q.closed {
+		return ErrQueueClosed
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	tx, err := q.db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	if err := fn(tx); err != nil {
+		_ = tx.Rollback()
+		return err
+	}
+	return tx.Commit()
+}
+
+// Close 关闭队列，之后 Do 返回 ErrQueueClosed。
+func (q *Queue) Close() error {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	q.closed = true
+	return nil
+}