package db import ( "context" "database/sql" "testing" ) // channelEventsMigrationID 是渠道事件留存两张表的迁移 id(design/26 §4.1,切片 E1)。 const channelEventsMigrationID = "0021_channel_events" // schemaObjectExists 报告 sqlite_master 里是否存在某类对象(table / index)。 // 与 columnExists 同族,都只读元数据、不碰 data/cyrene.db。 func schemaObjectExists(t *testing.T, ctx context.Context, conn *sql.DB, kind, name string) bool { t.Helper() var n int if err := conn.QueryRowContext(ctx, "SELECT count(*) FROM sqlite_master WHERE type = ? AND name = ?", kind, name).Scan(&n); err != nil { t.Fatalf("sqlite_master(%s %s): %v", kind, name, err) } return n == 1 } // TestMigrateChannelEventsTablesReversible 钉住 0021(channel_events + // channel_event_counters,design/26 §4.1):up 建两张表与全部索引、重复 up 是空操作、 // down 整表消失、再 up 恢复端到端可用。它用自己的内存库,绝不碰 data/cyrene.db。 // // 断言的是**库里的结构事实**(表/列/索引在不在、默认值读出来是什么),不按日志文本 // 断言 —— 与 migrate_turn_id_test.go / migrate_document_test.go 同一口径。 func TestMigrateChannelEventsTablesReversible(t *testing.T) { t.Parallel() ctx := context.Background() conn := openMemory(t) if err := MigrateUp(ctx, conn, testLogger()); err != nil { t.Fatalf("MigrateUp: %v", err) } // 两张表与它们的列必须逐个到位(列名错一个,写入面就会静默写错地方)。 columns := map[string][]string{ "channel_events": { "id", "platform", "account_id", "tier", "category", "platform_type", "channel_id", "channel_type", "group_id", "actor_id", "actor_role", "target_id", "self_related", "message_id", "summary", "created_at", "updated_at", }, "channel_event_counters": { "id", "platform", "account_id", "tier", "platform_type", "category", "count", "first_seen_at", "last_seen_at", }, } for table, cols := range columns { if !schemaObjectExists(t, ctx, conn, "table", table) { t.Fatalf("%s should exist after MigrateUp", table) } for _, col := range cols { if !columnExists(t, ctx, conn, table, col) { t.Fatalf("%s.%s should exist after MigrateUp", table, col) } } } // 索引:查询面与 GC 依赖 idx_channel_events_created_at,upsert 依赖计数表的唯一键。 for _, idx := range []string{ "idx_channel_events_created_at", "idx_channel_events_platform_category", "idx_channel_events_channel_id", "idx_channel_events_tier", "idx_channel_event_counters_key", "idx_channel_event_counters_category", } { if !schemaObjectExists(t, ctx, conn, "index", idx) { t.Fatalf("%s should exist after MigrateUp", idx) } } // 默认值:只写必填列也能落行,可选列读回来是 ''/0,时间戳由 SQLite 默认填。 if _, err := conn.ExecContext(ctx, `INSERT INTO channel_events (platform) VALUES (?)`, "qq-obv11"); err != nil { t.Fatalf("insert minimal channel event: %v", err) } var ( tier, category, summary, createdAt string selfRelated int ) if err := conn.QueryRowContext(ctx, `SELECT tier, category, summary, self_related, created_at FROM channel_events WHERE platform = ?`, "qq-obv11").Scan(&tier, &category, &summary, &selfRelated, &createdAt); err != nil { t.Fatalf("read minimal channel event: %v", err) } if tier != "" || category != "" || summary != "" || selfRelated != 0 { t.Fatalf("defaults = (%q, %q, %q, %d), want ('', '', '', 0)", tier, category, summary, selfRelated) } if createdAt == "" { t.Fatal("created_at should be filled by the migration default") } // 重复 up:已记录的迁移被跳过,结构仍在且不报错(幂等)。 if err := MigrateUp(ctx, conn, testLogger()); err != nil { t.Fatalf("MigrateUp second call: %v", err) } if !schemaObjectExists(t, ctx, conn, "table", "channel_events") { t.Fatal("channel_events should still exist after idempotent re-run") } // 恰好回滚到 0021(含它之后新加的迁移),日后再加迁移也不会让本测试失效。 ids := upMigrationIDs(t) idx := -1 for i, id := range ids { if id == channelEventsMigrationID { idx = i } } if idx < 0 { t.Fatalf("%s not found in %v", channelEventsMigrationID, ids) } if err := MigrateDown(ctx, conn, len(ids)-idx, testLogger()); err != nil { t.Fatalf("MigrateDown(%d): %v", len(ids)-idx, err) } for _, table := range []string{"channel_events", "channel_event_counters"} { if schemaObjectExists(t, ctx, conn, "table", table) { t.Fatalf("%s should be dropped by the 0021 down", table) } } // 重新 up 恢复端到端可用。 if err := MigrateUp(ctx, conn, testLogger()); err != nil { t.Fatalf("re-MigrateUp: %v", err) } for _, table := range []string{"channel_events", "channel_event_counters"} { if !schemaObjectExists(t, ctx, conn, "table", table) { t.Fatalf("%s should exist again after re-MigrateUp", table) } } } // TestMigrate0021CountersUniqueKey_正常 钉住计数表唯一键的**行为**(不只是「索引 // 存在」):同一 `(platform, account_id, platform_type)` 二次 upsert 必须累加成 // **一行**,否则「上游推过几次」的答案会分裂成多行、计数上界也随之失控 // (design/26 §4.1 表二的全部理由)。 func TestMigrate0021CountersUniqueKey_正常(t *testing.T) { t.Parallel() ctx := context.Background() conn := openMemory(t) if err := MigrateUp(ctx, conn, testLogger()); err != nil { t.Fatalf("MigrateUp: %v", err) } const upsert = `INSERT INTO channel_event_counters (platform, account_id, tier, platform_type, category, count, first_seen_at, last_seen_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(platform, account_id, platform_type) DO UPDATE SET count = count + excluded.count, last_seen_at = excluded.last_seen_at` for _, count := range []int{1, 2} { if _, err := conn.ExecContext(ctx, upsert, "qq-obv11", "bot1", "t1", "heartbeat", "heartbeat", count, "2026-10-05T00:00:00.000Z", "2026-10-05T00:00:00.000Z"); err != nil { t.Fatalf("upsert counter (%d): %v", count, err) } } var rows, total int if err := conn.QueryRowContext(ctx, `SELECT count(*), COALESCE(sum(count), 0) FROM channel_event_counters WHERE platform = ?`, "qq-obv11").Scan(&rows, &total); err != nil { t.Fatalf("read counters: %v", err) } if rows != 1 { t.Fatalf("counter rows = %d, want 1 (the unique key must merge, not duplicate)", rows) } if total != 3 { t.Fatalf("counter total = %d, want 3 (1 + 2)", total) } }