JianFeeeee
a8d6811cd7
## 主体是 gofmt
`server/` 下 20 个文件的改动里，**142/147 行是注释与缩进** ——
Go 1.19+ 的 gofmt 把块注释里的 `// x` 规范成 `//\tx`。
功能零变化。
## 一处实质：init.sql 补 parent_session_id
`ALTER TABLE sessions ADD COLUMN IF NOT EXISTS parent_session_id
TEXT REFERENCES sessions(session_id)`
远端已在 `db/migrate.go` 的 `sqliteAddColumns` 结构迁移通道里加了同一列；
init.sql 是**全新库**的建库脚本，缺这列会让新库少一个字段。
- `IF NOT EXISTS` ⇒ 与结构迁移通道幂等，两者并存无害
- 默认空 = 根节点；**不据历史数据猜父子**（猜错会让线索拓扑失真，比没有更糟）
## 新增 handler/ratelimit_test.go（162 行）
速率限制的判据。此前这一块没有独立判据。
## 验证（本机实跑，非推断）
go build ./... 通过
go vet ./internal/{handler,repo,db,sse}/ 通过
go test -count=1（handler/repo/db/sse/blob）
handler 41.3s ok
repo 101.9s ok
db 1.1s ok
sse 0.01s ok
blob 0.08s ok
`-count=1` 是刻意的：第一遍跑出来全是 `(cached)`，那种绿灯不算数。
109 lines
3.1 KiB
Go
109 lines
3.1 KiB
Go
package sse
|
||
|
||
import (
|
||
"net/http/httptest"
|
||
"strings"
|
||
"testing"
|
||
)
|
||
|
||
// 一个可 Flush 的 ResponseWriter，供集成测试用。
|
||
type flushWriter struct {
|
||
*httptest.ResponseRecorder
|
||
flushed chan bool
|
||
}
|
||
|
||
func (f *flushWriter) Flush() {
|
||
select {
|
||
case f.flushed <- true:
|
||
default:
|
||
}
|
||
}
|
||
|
||
// 端到端验证：Manager 完整走一遍「事件入缓冲区 → 新连接带 Last-Event-ID 重连 → 补投」。
|
||
// 这是生产里最关键的可靠性路径 —— 断线期间收的邮件，重连后必须能看到。
|
||
func TestManagerReplayOnReconnect(t *testing.T) {
|
||
m := &Manager{
|
||
clients: make(map[string]*Client),
|
||
eventBuffer: make(map[string]*eventRing),
|
||
}
|
||
|
||
// 1) 用户 alice 发来一封信（无人连接也入缓冲区）
|
||
m.SendToUser("alice", "new_mail", map[string]string{"mail_id": "m1"})
|
||
m.SendToUser("alice", "new_mail", map[string]string{"mail_id": "m2"})
|
||
m.SendToUser("alice", "new_mail", map[string]string{"mail_id": "m3"})
|
||
|
||
ring := m.eventBuffer["u:alice"]
|
||
if ring == nil {
|
||
t.Fatal("alice 的缓冲区应该已创建")
|
||
}
|
||
|
||
// 2) 带 Last-Event-ID=1 重连，应补投 m2、m3（跳过 m1）
|
||
req := httptest.NewRequest("GET", "/events/stream", nil)
|
||
req.Header.Set("Last-Event-ID", "1")
|
||
w := &flushWriter{httptest.NewRecorder(), make(chan bool, 10)}
|
||
|
||
client := m.AddClient(w, req, "", "alice")
|
||
if client == nil {
|
||
t.Fatal("AddClient 返回 nil（flushWriter 应支持 Flush）")
|
||
}
|
||
defer m.RemoveClient(client.ID)
|
||
|
||
body := w.Body.String()
|
||
if !strings.Contains(body, `"mail_id":"m2"`) {
|
||
t.Error("重连后应补投 m2，实际 body:", body)
|
||
}
|
||
if !strings.Contains(body, `"mail_id":"m3"`) {
|
||
t.Error("重连后应补投 m3，实际 body:", body)
|
||
}
|
||
if strings.Contains(body, `"mail_id":"m1"`) {
|
||
t.Error("已确认的 m1 不应重放（Last-Event-ID=1），实际 body:", body)
|
||
}
|
||
|
||
// 3) 连接期间新来一封信，实时推送
|
||
m.SendToUser("alice", "new_mail", map[string]string{"mail_id": "m4"})
|
||
body = w.Body.String()
|
||
if !strings.Contains(body, `"mail_id":"m4"`) {
|
||
t.Error("在线连接应实时收到 m4，实际 body:", body)
|
||
}
|
||
}
|
||
|
||
// 序列号全局递增，两条不同事件不同 ID。
|
||
func TestEventIDMonotonic(t *testing.T) {
|
||
m := &Manager{
|
||
clients: make(map[string]*Client),
|
||
eventBuffer: make(map[string]*eventRing),
|
||
}
|
||
a := m.nextEventID()
|
||
b := m.nextEventID()
|
||
if a == b {
|
||
t.Fatalf("两个连续事件 ID 相同: %q", a)
|
||
}
|
||
if a > b {
|
||
t.Fatalf("事件 ID 应递增: %q > %q", a, b)
|
||
}
|
||
}
|
||
|
||
// 确保事件 ID 写进了 SSE 帧（EventSource 靠 id: 行记住位置）
|
||
func TestSendWritesIDField(t *testing.T) {
|
||
fw := &flushWriter{httptest.NewRecorder(), make(chan bool, 5)}
|
||
c := &Client{
|
||
ID: "c1",
|
||
UserName: "alice",
|
||
Res: fw,
|
||
Flusher: fw,
|
||
done: make(chan struct{}),
|
||
}
|
||
c.SendWithID("42", "new_mail", map[string]string{"x": "y"})
|
||
|
||
body := fw.Body.String()
|
||
if !strings.Contains(body, "id: 42\n") {
|
||
t.Error("帧里应有 id: 42 行，实际:", body)
|
||
}
|
||
if !strings.Contains(body, "event: new_mail") {
|
||
t.Error("帧里应有 event: new_mail，实际:", body)
|
||
}
|
||
if !strings.Contains(body, `data: {"x":"y"}`) {
|
||
t.Error("帧里应有 data，实际:", body)
|
||
}
|
||
}
|