From 7582e8b55ecb8c7336b4f452e964c66017b4864d Mon Sep 17 00:00:00 2001 From: Nixevol Date: Wed, 30 Sep 2026 05:59:51 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=20SQLite=20=E8=BF=81?= =?UTF-8?q?=E7=A7=BB=E5=A4=87=E4=BB=BD=E3=80=81=E5=AE=8C=E6=95=B4=E8=A1=A8?= =?UTF-8?q?=E7=BB=93=E6=9E=84=E4=B8=8E=E7=AE=80=E5=8D=95=E5=86=99=E5=85=A5?= =?UTF-8?q?=E9=98=9F=E5=88=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/nixmsg/serve.go | 6 +- docs/DEVIATIONS.md | 24 ++++ internal/store/db.go | 112 +++++++++++++-- internal/store/db_test.go | 159 ++++++++++++++++++++-- internal/store/migrate.go | 111 +++++++++++---- internal/store/migrations/0002_schema.sql | 131 ++++++++++++++++++ internal/store/queue.go | 62 +++++++++ 7 files changed, 549 insertions(+), 56 deletions(-) create mode 100644 internal/store/migrations/0002_schema.sql create mode 100644 internal/store/queue.go diff --git a/cmd/nixmsg/serve.go b/cmd/nixmsg/serve.go index f9d53b6..debb3f7 100644 --- a/cmd/nixmsg/serve.go +++ b/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) diff --git a/docs/DEVIATIONS.md b/docs/DEVIATIONS.md index 7c6f149..ede7a2a 100644 --- a/docs/DEVIATIONS.md +++ b/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 则写入 `/backup/pre-migrate-.db`。迁移失败返回错误，不自动从备份恢复。 + - 原因：空库备份无意义；失败退出与文档一致，恢复交给运维。 + - 备选方案：失败时自动还原备份再退出。 + - 影响：与任务说明一致；运维需知备份路径。 + + ## 平台 P 暂无。 diff --git a/internal/store/db.go b/internal/store/db.go index d4ce685..85395d4 100644 --- a/internal/store/db.go +++ b/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")) - return 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", + dbPath := filepath.ToSlash(filepath.Join(dataDir, dbFileName)) + dsn := fmt.Sprintf( + "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 } diff --git a/internal/store/db_test.go b/internal/store/db_test.go index a43e472..9fc1198 100644 --- a/internal/store/db_test.go +++ b/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 { - t.Fatal(migErr) + assertMigrationCount(t, db.Write, 2) + 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 - if scanErr := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations`).Scan(&n); scanErr != nil { + var value string + if scanErr := db.Read.QueryRow(`SELECT value FROM settings WHERE key = ?`, "registration_enabled").Scan(&value); scanErr != nil { t.Fatal(scanErr) } - if n != 1 { - t.Fatalf("want 1 migration row, got %d", n) + if value != "0" { + 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() } diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 5212480..0ae347b 100644 --- a/internal/store/migrate.go +++ b/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。 -func Migrate(db *sql.DB) error { +// Migrate 应用尚未执行的嵌入迁移。 +// 若 dbExisted 为 true（调用 Open/Migrate 前已有 nixmsg.db）且存在未应用版本， +// 先 VACUUM INTO /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,36 +83,58 @@ 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)) - tx, err := db.Begin() - if err != nil { - return fmt.Errorf("begin migration %d: %w", version, err) - } - if sqlText != "" { - if _, err := tx.Exec(sqlText); err != nil { - _ = tx.Rollback() - return fmt.Errorf("apply migration %d: %w", version, err) - } - } - if _, err := tx.Exec( - `INSERT INTO schema_migrations(version, applied_at) VALUES(?, ?)`, - version, - time.Now().UnixMilli(), - ); err != nil { + 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", m.version, err) + } + if m.body != "" { + if _, err := tx.Exec(m.body); err != nil { _ = tx.Rollback() - return fmt.Errorf("record migration %d: %w", version, err) - } - if err := tx.Commit(); err != nil { - return fmt.Errorf("commit 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(?, ?)`, + m.version, + time.Now().UnixMilli(), + ); err != nil { + _ = tx.Rollback() + return fmt.Errorf("record migration %d: %w", m.version, err) + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit migration %d: %w", m.version, err) + } return nil } diff --git a/internal/store/migrations/0002_schema.sql b/internal/store/migrations/0002_schema.sql new file mode 100644 index 0000000..31b1cb1 --- /dev/null +++ b/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); diff --git a/internal/store/queue.go b/internal/store/queue.go new file mode 100644 index 0000000..2f7f487 --- /dev/null +++ b/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 +}