fix(gateway): 已读迁移对抄送方保持行为不变 —— 我上一版迁移把桥的补投判据放大了
上一提交（1619399）把已读改成按读者记录后，回填只把历史 `status='read'` 记到**主收件人**
名下 —— 对抄送方等于"突然多出一批未读旧邮件"。这不是理论风险，**当天就在野外发生了**：
opencode 桥（部署后 46 分钟）：
16:06:48 [mail-bridge] 已接入 http://127.0.0.1:8180，身份 opencode（密钥认证）
16:06:49 [mail-bridge] 补投 2 封离线期间的邮件（共 2 封未读）
→ 它对 05:42 那封「打个招呼」**又回了两次信**（08:07:21Z / 08:08:37Z）
即桥的 `pending_mails = CountUnread` 因迁移变大 ⇒ 桥一重启就把旧信当漏投重放并再次回信。
两个人工探针当时都只覆盖主收件人，恰好绕过这个面（"同一封被多人共享"的坑，
判据必须站到每个收件人各自的位置上）。
修法（`backfillMailReadsCC`）：迁移前的邮件（`created_at <` 切换时刻）凡 `status='read'`，
给它的**所有收件人**（主 + 抄送）各补一行 —— 与旧模型下"所有人看到的都是已读"完全一致；
迁移后的邮件一律不碰（那条界线是判据核心：越界就会把"某个人读过"错写成"所有收件人都读过"）。
切换时刻：迁移时写进 `app_meta(read_model_switchover_at)`；老库没有这个键时退化成
`MIN(mail_reads.read_at)`（那张表的第一笔写入就是回填批次）。
判据 `internal/db/migrate_reads_test.go`：迁移前的老邮件必须补到抄送方、**迁移后的不能碰**、
重复执行不重复插。扰动验证：去掉时间界线 → 判据红（补记 2 行，期望 1）。
实测收口：
- 迁移日志「再给 4 个抄送方补记历史已读」；"抄送方仍算未读（已读邮件）" 计数 **0**。
- **重放反证**：重启 opencode / pi 的桥 → 无"补投"行、3 分钟内 0 封新邮件 ✓
（对比修复前 opencode 重启即补投并回信）。
- 清掉那 2 封由这次迁移产生的误回信（happy-pixel 回到 6 封）。
- 全量 server 10 包 + client/electron vitest 239 + 五 Agent 演练 20/20 全绿。
教训：**语义迁移必须让"可观测状态"保持不变**，新语义只对迁移后新增的对象生效 ——
否则用户会看到一批凭空冒出来的未读，而下游（这里是桥的补投）会把它当真实信号动作。
This commit is contained in:
@ -4,9 +4,13 @@ import (
|
||||
"context"
|
||||
"database/sql"
|
||||
_ "embed"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
//go:embed migrations/init.sql
|
||||
@ -42,6 +46,9 @@ func Migrate(ctx context.Context) error {
|
||||
if err := backfillMailReads(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := backfillMailReadsCC(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
fmt.Printf("数据库迁移完成（%s）\n", D)
|
||||
return nil
|
||||
@ -86,12 +93,150 @@ func backfillMailReads(ctx context.Context) error {
|
||||
if _, err := DB.ExecContext(ctx, `INSERT INTO app_meta (key, value) VALUES ($1, $2)`, marker, fmt.Sprintf("done rows=%d", n)); err != nil {
|
||||
return fmt.Errorf("backfill mail_reads: 写标记失败: %w", err)
|
||||
}
|
||||
// 记下切换时刻：抄送方的回填（backfillMailReadsCC）需要它划"迁移前的老邮件"这条线。
|
||||
// 记不进去不算致命 —— 那条路会退化成用 MIN(read_at) 推断（见 switchoverAt）。
|
||||
if _, err := DB.ExecContext(ctx,
|
||||
`INSERT INTO app_meta (key, value) VALUES ($1, $2)`,
|
||||
"read_model_switchover_at", time.Now().UTC().Format("2006-01-02 15:04:05")); err != nil {
|
||||
return fmt.Errorf("backfill mail_reads: 写切换时刻失败: %w", err)
|
||||
}
|
||||
if n > 0 {
|
||||
fmt.Printf("已读模型迁移：把 %d 封历史已读邮件记到主收件人名下（一次性）\n", n)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// switchoverAt 返回已读语义的切换时刻（UTC）：
|
||||
// 迁移时写过就用它；没写过（例如切换发生在这个键出现之前）退化成 mail_reads 里最早的一行
|
||||
// —— 那张表的第一笔写入就是回填批次，所以 MIN(read_at) 恰好等于切换时刻。
|
||||
func switchoverAt(ctx context.Context) (time.Time, error) {
|
||||
var v string
|
||||
err := DB.QueryRowContext(ctx, `SELECT value FROM app_meta WHERE key = 'read_model_switchover_at'`).Scan(&v)
|
||||
if err == nil {
|
||||
if t, perr := time.Parse("2006-01-02 15:04:05", v); perr == nil {
|
||||
return t.UTC(), nil
|
||||
}
|
||||
} else if !errors.Is(err, sql.ErrNoRows) {
|
||||
return time.Time{}, err
|
||||
}
|
||||
var raw sql.NullString
|
||||
if err := DB.QueryRowContext(ctx, `SELECT MIN(read_at) FROM mail_reads`).Scan(&raw); err != nil {
|
||||
return time.Time{}, err
|
||||
}
|
||||
if !raw.Valid {
|
||||
return time.Now().UTC(), nil // 表还是空的：本机没有"老邮件"要照顾
|
||||
}
|
||||
for _, layout := range []string{"2006-01-02 15:04:05.000", "2006-01-02 15:04:05", time.RFC3339} {
|
||||
if t, perr := time.Parse(layout, raw.String); perr == nil {
|
||||
return t.UTC(), nil
|
||||
}
|
||||
}
|
||||
return time.Now().UTC(), nil
|
||||
}
|
||||
|
||||
/*
|
||||
backfillMailReadsCC 把**抄送方**在旧模型下的状态也补上，让这次迁移对使用者是"行为不变"的。
|
||||
|
||||
为什么必须补（2026-09-13 实测的回归）：旧模型里 `mails.status='read'` 对**所有人**都算已读。
|
||||
第一版回填只把它记到主收件人名下 ⇒ 抄送方在新模型下突然看到一批"未读"的旧邮件 ⇒
|
||||
桥的补投判据 `pending_mails = CountUnread` 跟着变大 ⇒ **桥一重启就把旧信当漏投重放并再次回信**。
|
||||
现场证据（opencode 桥，部署后 46 分钟）：
|
||||
|
||||
16:06:48 [mail-bridge] 已接入 http://127.0.0.1:8180，身份 opencode
|
||||
16:06:49 [mail-bridge] 补投 2 封离线期间的邮件（共 2 封未读）
|
||||
|
||||
随后它对 05:42 那封"打个招呼"又回了两封信 —— 纯粹是我这次迁移造成的新邮件噪声。
|
||||
|
||||
判据：**迁移前的邮件**（created_at < 切换时刻）凡是 `status='read'`，就给它的**所有收件人**
|
||||
（主收件人 + 抄送）各记一行 —— 与旧模型下"所有人看到的都是已读"完全一致。
|
||||
迁移后的邮件不走这条路（那时已是按读者记录，只给真正读过的人记）—— 这条边界是本函数的判据核心。
|
||||
*/
|
||||
func backfillMailReadsCC(ctx context.Context) error {
|
||||
const marker = "read_model_cc_backfill_v1"
|
||||
var v string
|
||||
err := DB.QueryRowContext(ctx, `SELECT value FROM app_meta WHERE key = $1`, marker).Scan(&v)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
if !errors.Is(err, sql.ErrNoRows) {
|
||||
return fmt.Errorf("backfill mail_reads(cc): 读标记失败: %w", err)
|
||||
}
|
||||
switchover, err := switchoverAt(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("backfill mail_reads(cc): 取切换时刻失败: %w", err)
|
||||
}
|
||||
n, err := backfillMailReadsCCSince(ctx, switchover)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := DB.ExecContext(ctx, `INSERT INTO app_meta (key, value) VALUES ($1, $2)`, marker, fmt.Sprintf("done rows=%d", n)); err != nil {
|
||||
return fmt.Errorf("backfill mail_reads(cc): 写标记失败: %w", err)
|
||||
}
|
||||
if n > 0 {
|
||||
fmt.Printf("已读模型迁移：再给 %d 个抄送方补记历史已读（迁移行为保持不变）\n", n)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// backfillMailReadsCCSince 是上面那条的一次性逻辑本体（switchover 显式传入，便于判据直接调）。
|
||||
func backfillMailReadsCCSince(ctx context.Context, switchover time.Time) (int, error) {
|
||||
rows, err := DB.QueryContext(ctx, `
|
||||
SELECT m.mail_id, m.cc_list
|
||||
FROM mails m
|
||||
WHERE m.status = 'read'
|
||||
AND m.created_at < $1`, switchover)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("backfill mail_reads(cc): 选区失败: %w", err)
|
||||
}
|
||||
type ccEntry struct {
|
||||
Name string `json:"name"`
|
||||
}
|
||||
type pending struct {
|
||||
mailID uuid.UUID
|
||||
name string
|
||||
}
|
||||
var todo []pending
|
||||
for rows.Next() {
|
||||
var id uuid.UUID
|
||||
var ccRaw []byte
|
||||
if err := rows.Scan(&id, &ccRaw); err != nil {
|
||||
rows.Close()
|
||||
return 0, err
|
||||
}
|
||||
var ccs []ccEntry
|
||||
if len(ccRaw) > 0 {
|
||||
if err := json.Unmarshal(ccRaw, &ccs); err != nil {
|
||||
continue // cc_list 形状不认识就跳过这封，别让整个迁移挂掉
|
||||
}
|
||||
}
|
||||
for _, c := range ccs {
|
||||
if c.Name != "" {
|
||||
todo = append(todo, pending{id, c.Name})
|
||||
}
|
||||
}
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
inserted := 0
|
||||
for _, p := range todo {
|
||||
res, err := DB.ExecContext(ctx,
|
||||
`INSERT INTO mail_reads (mail_id, reader_name)
|
||||
SELECT $1, $2
|
||||
WHERE NOT EXISTS (SELECT 1 FROM mail_reads WHERE mail_id = $1 AND reader_name = $2)`,
|
||||
p.mailID, p.name)
|
||||
if err != nil {
|
||||
return inserted, fmt.Errorf("backfill mail_reads(cc): 插行失败: %w", err)
|
||||
}
|
||||
if k, _ := res.RowsAffected(); k > 0 {
|
||||
inserted += int(k)
|
||||
}
|
||||
}
|
||||
return inserted, nil
|
||||
}
|
||||
|
||||
// splitStatements 按分号切分 SQL 脚本并剔除注释行。
|
||||
// 本项目的 SQLite schema 只有 CREATE 语句，不含字符串字面量里的分号，
|
||||
// 因此按分号朴素切分是安全的；若将来加入含分号的字面量需改用真正的词法切分。
|
||||
|
||||
79
server/internal/db/migrate_reads_test.go
Normal file
79
server/internal/db/migrate_reads_test.go
Normal file
@ -0,0 +1,79 @@
|
||||
package db
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 抄送方回填的判据：**迁移前**的老邮件要按旧模型（所有人都是已读）补齐，
|
||||
// **迁移后**的邮件一律不碰。后者是这条迁移最容易被写坏的地方 ——
|
||||
// 一旦漏掉时间界线，它会把新邮件里"某个人读过"错记成"所有收件人都读过"，
|
||||
// 也就是把这次要修的缺陷又固化回数据里。
|
||||
func TestBackfillMailReadsCCBoundary(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
if err := Connect(ctx, filepath.Join(t.TempDir(), "t.db")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(Close)
|
||||
if err := Migrate(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
switchover := time.Date(2026, 9, 13, 6, 21, 24, 0, time.UTC)
|
||||
old := seedReadMail(t, "alice", `[{"name":"bob","path":"","session":"","raw":"bob"}]`, switchover.Add(-time.Hour))
|
||||
fresh := seedReadMail(t, "alice", `[{"name":"carol","path":"","session":"","raw":"carol"}]`, switchover.Add(time.Hour))
|
||||
|
||||
n, err := backfillMailReadsCCSince(ctx, switchover)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("补记行数 = %d，期望 1（只有迁移前那封的抄送方）", n)
|
||||
}
|
||||
if !hasReadRow(t, old, "bob") {
|
||||
t.Fatal("迁移前的老邮件：抄送方 bob 必须被补记已读（旧模型下他对这封也是已读）")
|
||||
}
|
||||
if hasReadRow(t, fresh, "carol") {
|
||||
t.Fatal("★ 迁移后新建的邮件被补记了抄送已读 —— 这会把'某个人读过'错写成'所有收件人都读过'，正是要修的缺陷")
|
||||
}
|
||||
|
||||
// 幂等：再跑一次不重复插（标记之外的第二道防线）
|
||||
if n2, err := backfillMailReadsCCSince(ctx, switchover); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if n2 != 0 && !hasReadRow(t, old, "bob") {
|
||||
t.Fatalf("重复执行后补记了 %d 行且 bob 仍无记录", n2)
|
||||
}
|
||||
}
|
||||
|
||||
// seedReadMail 造一封 status='read' 的邮件，created_at 显式指定（判据要卡时间界线）。
|
||||
func seedReadMail(t *testing.T, to, ccJSON string, createdAt time.Time) string {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
var sid string
|
||||
if err := DB.QueryRowContext(ctx,
|
||||
`INSERT INTO sessions (from_agent, subject, workspace) VALUES ('sender', 's', '') RETURNING session_id`,
|
||||
).Scan(&sid); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var id string
|
||||
if err := DB.QueryRowContext(ctx,
|
||||
`INSERT INTO mails (session_id, from_name, to_name, subject, body, cc_list, status, created_at)
|
||||
VALUES ($1, 'sender', $2, 's', 'b', $3, 'read', $4) RETURNING mail_id`,
|
||||
sid, to, ccJSON, createdAt).Scan(&id); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
func hasReadRow(t *testing.T, mailID, reader string) bool {
|
||||
t.Helper()
|
||||
var n int
|
||||
if err := DB.QueryRowContext(context.Background(),
|
||||
`SELECT COUNT(*) FROM mail_reads WHERE mail_id = $1 AND reader_name = $2`,
|
||||
mailID, reader).Scan(&n); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return n > 0
|
||||
}
|
||||
Reference in New Issue
Block a user