feat(gateway+web): D 消息编辑/删除 WS 实时广播 — message_edited/message_deleted
- 后端: ws_hub.go broadcastMessageEdited(完整DTO,thinking剥离)/broadcastMessageDeleted(全广播+sessionId,前端过滤); history_write.go PATCH/DELETE成功后触发(404/400/501不广播); history/message_ops.go GetMessage(导出壳,取删除前sessionId) - 前端: useAppStore.updateMessageById/removeMessageById + messageEventApplies(同会话过滤); ws.ts 消费两事件(缺content丢弃); dispatchFrame导出(可测) - 真机e2e: PATCH/DELETE → WS收到 message_edited/message_deleted, 二次删404无帧 - Docs/api-client v0.34 §3.8; 测试 history_write_broadcast_test + messageSync.test(14 checks) - 遗留: 前端编辑/删除UI入口+api客户端函数未接(lib/api 无 PATCH/DELETE messages); ws_hub.go 589行>500(既有)
This commit is contained in:
1 parent
4ea68d17f3
commit
2720d77331
8 files changed
+485
-15
No files matched your search
@@ -9,10 +9,11 @@ import (
|
||||
)
|
||||
|
||||
// This file owns the single-message operations behind the Android collaboration batch
|
||||
// ③a (D2 search, D3 edit/delete). Both are strictly session-scoped or id-scoped: a
|
||||
// message is only ever reached through the session it belongs to (Search) or its own
|
||||
// opaque id (UpdateMessage/DeleteMessage), so an isolated group/stranger session can
|
||||
// never surface in another session's results (project red line).
|
||||
// ③a (D2 search, D3 edit/delete) plus the id-scoped read the D delete broadcast needs.
|
||||
// All are strictly session-scoped or id-scoped: a message is only ever reached through
|
||||
// the session it belongs to (Search) or its own opaque id (GetMessage/UpdateMessage/
|
||||
// DeleteMessage), so an isolated group/stranger session can never surface in another
|
||||
// session's results (project red line).
|
||||
|
||||
// Search returns up to limit messages for sessionID whose content contains query,
|
||||
// newest first. The match is a case-insensitive ASCII LIKE (SQLite's default for
|
||||
@@ -91,6 +92,14 @@ func (s *Store) DeleteMessage(ctx context.Context, id string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetMessage reads the message with id, returning ErrMessageNotFound when no such
|
||||
// row exists. It is the id-scoped read the transport uses before removing a
|
||||
// message: DeleteMessage leaves nothing to ask afterwards, yet the broadcast that
|
||||
// tells other clients to drop the row is session-scoped.
|
||||
func (s *Store) GetMessage(ctx context.Context, id string) (Message, error) {
|
||||
return s.messageByID(ctx, id)
|
||||
}
|
||||
|
||||
// messageByID reads one message by id, returning ErrMessageNotFound when the row is
|
||||
// absent.
|
||||
func (s *Store) messageByID(ctx context.Context, id string) (Message, error) {
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
@@ -14,7 +15,9 @@ import (
|
||||
// This file owns the session/message write endpoints added for the Android
|
||||
// collaboration batch ③a: PATCH /sessions/{id} (A4 rename + D5 pin), PATCH
|
||||
// /messages/{id} (D3 edit) and DELETE /messages/{id} (D3 delete). All are owner/admin
|
||||
// history writes; the reads and DTOs live in history.go.
|
||||
// history writes; the reads and DTOs live in history.go. The D3 edit/delete paths also
|
||||
// broadcast their outcome over the WS hub (message_edited / message_deleted, see
|
||||
// ws_hub.go) so the account's other browsers stay in sync without a manual re-read.
|
||||
|
||||
// patchSessionRequest is the PATCH /api/v1/sessions/{id} body. Both fields are pointers
|
||||
// so an absent field is distinguishable from an explicit zero value: title renames the
|
||||
@@ -145,11 +148,20 @@ func (s *Server) handlePatchMessage(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(w, apperr.Wrap(apperr.ErrInternal, "message_failed", "failed to update message"))
|
||||
return
|
||||
}
|
||||
// Broadcast the edit so every other browser of the account re-renders the same
|
||||
// bubble without a refresh (the message_persisted realtime-sync contract, D3 edit
|
||||
// path). Thinking is stripped from the frame to match the persisted/replay
|
||||
// contract; the REST reply keeps it because the messages routes are owner/admin-only.
|
||||
edited := toHistoryMessage(msg)
|
||||
edited.Thinking = ""
|
||||
s.hub.broadcastMessageEdited(edited)
|
||||
writeJSON(w, http.StatusOK, toHistoryMessage(msg))
|
||||
}
|
||||
|
||||
// handleDeleteMessage removes a single message (D3). An unknown id answers 404, a
|
||||
// missing HistoryStore 501. It is owner/admin-gated.
|
||||
// missing HistoryStore 501. It is owner/admin-gated. A successful delete is
|
||||
// broadcast as a `message_deleted` frame so every browser of the account drops the
|
||||
// row without a refresh; the owning session is read before the row disappears.
|
||||
func (s *Server) handleDeleteMessage(w http.ResponseWriter, r *http.Request) {
|
||||
if s.history == nil {
|
||||
writeError(w, &apperr.AppError{Code: "not_implemented", Message: "会话历史服务未配置(未注入 HistoryStore)", HTTPCode: http.StatusNotImplemented})
|
||||
@@ -160,6 +172,7 @@ func (s *Server) handleDeleteMessage(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "message id is required"))
|
||||
return
|
||||
}
|
||||
sessionID := s.messageSession(r.Context(), id)
|
||||
if err := s.history.DeleteMessage(r.Context(), id); err != nil {
|
||||
if errors.Is(err, history.ErrMessageNotFound) {
|
||||
writeError(w, apperr.Wrap(apperr.ErrNotFound, "not_found", "message %q not found", id))
|
||||
@@ -169,5 +182,31 @@ func (s *Server) handleDeleteMessage(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(w, apperr.Wrap(apperr.ErrInternal, "message_failed", "failed to delete message"))
|
||||
return
|
||||
}
|
||||
s.hub.broadcastMessageDeleted(sessionID, id)
|
||||
writeJSON(w, http.StatusOK, map[string]any{"deleted": id})
|
||||
}
|
||||
|
||||
// messageLookup is the optional narrower view of HistoryStore used by DELETE to learn
|
||||
// a message's owning session before it is removed (afterwards there is nothing left to
|
||||
// ask). *history.Store implements it; a store that does not simply yields an unscoped
|
||||
// frame, which the client resolves by the globally unique message id.
|
||||
type messageLookup interface {
|
||||
GetMessage(ctx context.Context, id string) (history.Message, error)
|
||||
}
|
||||
|
||||
// messageSession returns the session id owning message id, or "" when the store cannot
|
||||
// answer (no lookup surface, unknown id, or a read error). A blank session is safe for
|
||||
// the broadcast: the message id is globally unique, so a client that falls back to an
|
||||
// id-only filter can only ever touch that message's own conversation.
|
||||
func (s *Server) messageSession(ctx context.Context, id string) string {
|
||||
lookup, ok := s.history.(messageLookup)
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
msg, err := lookup.GetMessage(ctx, id)
|
||||
if err != nil {
|
||||
s.log.Debug("history message session lookup", "err", err, "message", id)
|
||||
return ""
|
||||
}
|
||||
return msg.SessionID
|
||||
}
|
||||
@@ -0,0 +1,165 @@
|
||||
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"
|
||||
)
|
||||
|
||||
// These tests cover the D3 realtime-sync contract: a successful message edit or
|
||||
// delete must reach the account's other browsers as a WS frame, not only the client
|
||||
// that issued the REST call (otherwise every other browser has to re-read to converge).
|
||||
|
||||
// TestPatchMessageBroadcastsEdited covers ①: PATCH /api/v1/messages/{id} answers 200
|
||||
// and pushes exactly one `message_edited` frame scoped to the edited message's own
|
||||
// session, carrying the new content (and the mode/createdAt a client renders with).
|
||||
// The frame's data is the updated message DTO with thinking stripped, matching the
|
||||
// message_persisted contract; the REST reply keeps thinking (owner/admin-only read).
|
||||
func TestPatchMessageBroadcastsEdited(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
srv, ts, jwt, st := newHistoryEnv(t)
|
||||
c := newFakeConn(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)
|
||||
}
|
||||
// The REST reply is the full DTO for the issuing client.
|
||||
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 covers ②: DELETE /api/v1/messages/{id} answers
|
||||
// 200, really removes the row, and pushes one `message_deleted` frame carrying the id
|
||||
// plus the owning session (read before the row disappears) so a remote browser can
|
||||
// drop it without re-reading.
|
||||
func TestDeleteMessageBroadcastsDeleted(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
srv, ts, jwt, st := newHistoryEnv(t)
|
||||
c := newFakeConn(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)
|
||||
}
|
||||
|
||||
// The broadcast must not claim a delete that did not happen.
|
||||
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 pins the failure side: a 404 delete
|
||||
// (unknown id) must not push a `message_deleted` frame, or a client would drop a
|
||||
// message that still exists.
|
||||
func TestDeleteMessageUnknownIDBroadcastsNothing(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv, ts, jwt, _ := newHistoryEnv(t)
|
||||
c := newFakeConn(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)
|
||||
}
|
||||
}
|
||||
@@ -188,6 +188,35 @@ func (h *hub) broadcast(ev wsEvent) {
|
||||
}
|
||||
}
|
||||
|
||||
// wsMessageEdited / wsMessageDeleted are the downstream event types of the D3
|
||||
// history writes: a message's content was edited, or a message was removed. Both
|
||||
// reuse the wsEvent envelope and the same all-connections fan-out as
|
||||
// `message_persisted` — the hub has no per-session subscription, so the frame
|
||||
// carries SessionID and a client filters it by its active session, exactly as it
|
||||
// does for message_persisted.
|
||||
const (
|
||||
wsMessageEdited = "message_edited"
|
||||
wsMessageDeleted = "message_deleted"
|
||||
)
|
||||
|
||||
// broadcastMessageEdited fans a content edit out to every client as a
|
||||
// `message_edited` frame whose data is the updated message DTO. The frame is a
|
||||
// render-only sync: the edited content's single source of truth stays the history
|
||||
// store, so a client that missed the frame converges again on its next read. The
|
||||
// caller strips Thinking to match the message_persisted/replay contract.
|
||||
func (h *hub) broadcastMessageEdited(msg historyMessage) {
|
||||
h.broadcast(wsEvent{Type: wsMessageEdited, SessionID: msg.SessionID, Data: msg})
|
||||
}
|
||||
|
||||
// broadcastMessageDeleted fans a removal out to every client as a
|
||||
// `message_deleted` frame carrying the removed id (plus the owning sessionId for
|
||||
// client-side session filtering), so every browser of the account drops the row
|
||||
// without a refresh. sessionID may be empty when the store could not report it;
|
||||
// clients then fall back to the globally unique id.
|
||||
func (h *hub) broadcastMessageDeleted(sessionID, id string) {
|
||||
h.broadcast(wsEvent{Type: wsMessageDeleted, SessionID: sessionID, Data: map[string]any{"id": id}})
|
||||
}
|
||||
|
||||
// relay is the bus handler for the ws.> pattern and the producer bridge
|
||||
// patterns. The ws.chat subject is a streaming trigger (run the loop and emit
|
||||
// chunks); a ws.* subject becomes an outbound event of its design type; a
|
||||
|
||||
Reference in new issue
Block a user