Cyrene_v2/internal/services/gateway/history_write_broadcast_test.go

package gateway

import (
	"context"
	"encoding/json"
	"net/http"
	"testing"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
)

// 这些测试覆盖 D3 实时同步契约：一次成功的消息编辑或删除必须以
// WS 帧到达该账号的其他浏览器，而不仅发起 REST 调用的客户端
// （否则其他每个浏览器都得重读才能收敛）。

// TestPatchMessageBroadcastsEdited 覆盖 ①：PATCH /api/v1/messages/{id}
// 返回 200，并恰好推送一个 `message_edited` 帧，其作用域为被编辑消息
// 自身的会话，携带新内容（以及客户端渲染所用的 mode/createdAt）。该帧
// 的 data 是抹除了 thinking 的更新后消息 DTO，与 message_persisted
// 契约一致；REST 回复保留 thinking（仅 owner/admin 可读）。
func TestPatchMessageBroadcastsEdited(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	srv, ts, jwt, st := newHistoryEnv(t)
	c := newOwnerConn(srv.hub, 32)
	srv.hub.register(c)
	sub, err := st.NewSubSession(ctx, "owner")
	if err != nil {
		t.Fatalf("new sub: %v", err)
	}
	tok := signAny(t, jwt, "acct", auth.RoleOwner)

	seedMessage := func(sessionID string) history.Message {
		t.Helper()
		return seedFullMessage(t, st, history.Message{
			SessionID: sessionID, Role: history.RoleUser, Content: "old body",
			Mode: "chat", Thinking: "internal chain", CreatedAt: 1000,
		})
	}
	for _, tc := range []struct {
		name      string
		sessionID string
	}{
		{name: "main session", sessionID: history.MainSessionID},
		{name: "sub session", sessionID: sub},
	} {
		t.Run(tc.name, func(t *testing.T) {
			msg := seedMessage(tc.sessionID)
			resp, body := doReq(t, http.MethodPatch, ts.URL+"/api/v1/messages/"+msg.ID, map[string]any{"content": "new body"}, tok)
			if resp.StatusCode != http.StatusOK {
				t.Fatalf("patch status = %d, want 200 (body=%s)", resp.StatusCode, body)
			}
			// REST 回复是发给发起方客户端的完整 DTO。
			var reply historyMessage
			if err := json.Unmarshal(decodeBody(t, body)["data"], &reply); err != nil {
				t.Fatalf("decode reply: %v (body=%s)", err, body)
			}
			if reply.Content != "new body" || reply.Thinking != "internal chain" {
				t.Fatalf("reply = %+v, want edited content + kept thinking", reply)
			}

			ev := readFakeFrame(t, c)
			if ev.Type != "message_edited" {
				t.Fatalf("frame type = %q, want message_edited (frame=%+v)", ev.Type, ev)
			}
			if ev.SessionID != tc.sessionID {
				t.Fatalf("frame sessionId = %q, want %q", ev.SessionID, tc.sessionID)
			}
			data, ok := ev.Data.(map[string]any)
			if !ok {
				t.Fatalf("frame data = %+v, want object", ev.Data)
			}
			if data["id"] != msg.ID || data["content"] != "new body" {
				t.Fatalf("frame data = %+v, want id=%s content=new body", data, msg.ID)
			}
			if data["sessionId"] != tc.sessionID {
				t.Fatalf("frame data.sessionId = %v, want %q", data["sessionId"], tc.sessionID)
			}
			if data["mode"] != "chat" {
				t.Fatalf("frame data.mode = %v, want chat", data["mode"])
			}
			if _, isNum := data["createdAt"].(float64); !isNum {
				t.Fatalf("frame data.createdAt = %v, want a number", data["createdAt"])
			}
			if _, leaked := data["thinking"]; leaked {
				t.Fatalf("frame data = %+v, thinking must be stripped from the broadcast", data)
			}
			if err := connHasNoFrame(c); err != nil {
				t.Fatal(err)
			}
		})
	}
}

// TestDeleteMessageBroadcastsDeleted 覆盖 ②：DELETE /api/v1/messages/{id}
// 返回 200，真正删除该行，并推送一个 `message_deleted` 帧，携带 id 以及
// 所属会话（在该行消失前读出），使远端浏览器无需重读即可移除它。
func TestDeleteMessageBroadcastsDeleted(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	srv, ts, jwt, st := newHistoryEnv(t)
	c := newOwnerConn(srv.hub, 32)
	srv.hub.register(c)
	sub, err := st.NewSubSession(ctx, "owner")
	if err != nil {
		t.Fatalf("new sub: %v", err)
	}
	msg := seedFullMessage(t, st, history.Message{SessionID: sub, Role: history.RoleUser, Content: "to be removed", CreatedAt: 2000})
	tok := signAny(t, jwt, "acct", auth.RoleOwner)

	resp, body := doReq(t, http.MethodDelete, ts.URL+"/api/v1/messages/"+msg.ID, nil, tok)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("delete status = %d, want 200 (body=%s)", resp.StatusCode, body)
	}

	ev := readFakeFrame(t, c)
	if ev.Type != "message_deleted" {
		t.Fatalf("frame type = %q, want message_deleted (frame=%+v)", ev.Type, ev)
	}
	if ev.SessionID != sub {
		t.Fatalf("frame sessionId = %q, want %q", ev.SessionID, sub)
	}
	data, ok := ev.Data.(map[string]any)
	if !ok {
		t.Fatalf("frame data = %+v, want object", ev.Data)
	}
	if data["id"] != msg.ID {
		t.Fatalf("frame data.id = %v, want %s", data["id"], msg.ID)
	}
	if err := connHasNoFrame(c); err != nil {
		t.Fatal(err)
	}

	// 广播不得宣称一次并未发生的删除。
	remaining, err := st.List(ctx, sub, 0, 50, "")
	if err != nil {
		t.Fatalf("list after delete: %v", err)
	}
	for _, m := range remaining {
		if m.ID == msg.ID {
			t.Fatalf("message %s still stored after a successful delete", msg.ID)
		}
	}
}

// TestDeleteMessageUnknownIDBroadcastsNothing 钉住失败侧：404 的删除
// （未知 id）不得推送 `message_deleted` 帧，否则客户端会丢掉一条仍然
// 存在的消息。
func TestDeleteMessageUnknownIDBroadcastsNothing(t *testing.T) {
	t.Parallel()
	srv, ts, jwt, _ := newHistoryEnv(t)
	c := newOwnerConn(srv.hub, 32)
	srv.hub.register(c)
	tok := signAny(t, jwt, "acct", auth.RoleOwner)

	resp, body := doReq(t, http.MethodDelete, ts.URL+"/api/v1/messages/msg-does-not-exist", nil, tok)
	if resp.StatusCode != http.StatusNotFound {
		t.Fatalf("delete status = %d, want 404 (body=%s)", resp.StatusCode, body)
	}
	if err := connHasNoFrame(c); err != nil {
		t.Fatal(err)
	}
}