core+event+domain: drop broken parent_id-as-user-id routing in replies

parent_id is a TWEET id, not a user id (it selects which subtree of
replies inside a thread to return). The existing StreamGetRepliesHandler
treated it as a user id:

  isOwnTweetReplies := parentId == streamer.NodeInfo().OwnerId
  parentUser, err := userRepo.Get(parentId)

Both were always false / always ErrUserNotFound, so the handler
silently fell through to the local cache for every request. Removed
the dead remote-routing path; the handler now serves replies
straight from the local repo. Empty parent_id is still normalised to
root_id (top-level replies of the thread).

Proper remote-fetch routing would require a RootUserId field in
GetAllRepliesEvent so we know which node to forward to — out of scope
here; clients fall back to whatever replies they've already cached
from gossip / earlier fetches.

Also annotated GetAllRepliesEvent, NewReplyEvent, and domain.Tweet
with comments calling out that ParentId is a tweet id, so this
mistake doesn't get re-introduced.

Tests: dropped the now-impossible "parent user not found / stream
offline / remote success" cases (the handler no longer touches a
streamer or user repo) and added a "propagates repo error" case.

Author: claude

cmd/node/member/node/member-node.go

@@ -543,7 +543,7 @@ func (m *MemberNode) replyHandlers(
 		},
 		{
 			event.PUBLIC_GET_REPLIES,
-			handler.StreamGetRepliesHandler(r.replyRepo, userRepo, m),
+			handler.StreamGetRepliesHandler(r.replyRepo),
 		},
 	}
 }

core/handler/reply.go

@@ -29,7 +29,6 @@ package handler
 
 import (
 	"errors"
-	"fmt"
 	"strings"
 
 	"github.com/Warp-net/warpnet/core/stream"
@@ -304,11 +303,26 @@ func StreamDeleteReplyHandler(
 	}
 }
 
-func StreamGetRepliesHandler(
-	repo ReplyStorer,
-	userRepo ReplyUserFetcher,
-	streamer ReplyStreamer,
-) warpnet.WarpHandlerFunc {
+// StreamGetRepliesHandler answers /public/get/replies requests.
+//
+// ev.RootId is the root tweet of the thread; ev.ParentId is the parent
+// TWEET id selecting which subtree of replies to return (NOT a user id —
+// it gets compared against tweet/reply ids in the repo). Clients send
+// an empty ParentId for "give me the top-level replies of the thread",
+// which we normalise to RootId so the repo lookup matches the first
+// tier of replies. Replies are served straight from the local store:
+// any reply we know about (because the author's node pushed it to us
+// via gossip, or because we cached an earlier fetch) is returned;
+// otherwise the response is empty.
+//
+// Note on routing: this used to try to forward the request to the
+// "parent user" by treating ParentId as a user id and looking it up in
+// userRepo. That can't work — ParentId is a tweet id — so the lookup
+// always returned ErrUserNotFound and we silently fell back to local
+// storage anyway. The dead code is removed; proper remote-fetch
+// routing would need a RootUserId in GetAllRepliesEvent to identify the
+// author of the root tweet, which clients don't currently send.
+func StreamGetRepliesHandler(repo ReplyStorer) warpnet.WarpHandlerFunc {
 	return func(buf []byte, s warpnet.WarpStream) (any, error) {
 		var ev event.GetAllRepliesEvent
 		err := json.Unmarshal(buf, &ev)
@@ -320,71 +334,22 @@ func StreamGetRepliesHandler(
 		}
 		// Top-level replies on a thread have no parent — clients send an
 		// empty parent_id in that case. Treat it as the root itself so
-		// the lookup returns the first-tier replies.
+		// the repo returns the first-tier replies hanging off RootId.
 		if ev.ParentId == "" {
 			ev.ParentId = ev.RootId
 		}
 
 		rootId := strings.TrimPrefix(ev.RootId, domain.RetweetPrefix)
 		parentId := strings.TrimPrefix(ev.ParentId, domain.RetweetPrefix)
-		isOwnTweetReplies := parentId == streamer.NodeInfo().OwnerId
-
-		if isOwnTweetReplies {
-			replies, cursor, err := repo.GetRepliesTree(rootId, parentId, ev.Limit, ev.Cursor)
-			if err != nil {
-				return nil, err
-			}
-			return event.RepliesResponse{
-				Cursor:  cursor,
-				Replies: replies,
-				UserId:  &parentId,
-			}, nil
-		}
-
-		parentUser, err := userRepo.Get(parentId)
-		if errors.Is(err, database.ErrUserNotFound) {
-			replies, cursor, _ := repo.GetRepliesTree(rootId, parentId, ev.Limit, ev.Cursor)
-			return event.RepliesResponse{
-				Cursor:  cursor,
-				Replies: replies,
-				UserId:  &parentId,
-			}, nil
-		}
-		if err != nil {
-			return nil, err
-		}
 
-		replyDataResp, err := streamer.GenericStream(
-			parentUser.NodeId,
-			event.PUBLIC_GET_REPLIES,
-			ev,
-		)
-		if errors.Is(err, warpnet.ErrNodeIsOffline) {
-			replies, cursor, _ := repo.GetRepliesTree(rootId, parentId, ev.Limit, ev.Cursor)
-			return event.RepliesResponse{
-				Cursor:  cursor,
-				Replies: replies,
-				UserId:  &parentId,
-			}, nil
-		}
+		replies, cursor, err := repo.GetRepliesTree(rootId, parentId, ev.Limit, ev.Cursor)
 		if err != nil {
 			return nil, err
 		}
-
-		var possibleError event.ResponseError
-		if _ = json.Unmarshal(replyDataResp, &possibleError); possibleError.Message != "" {
-			return nil, fmt.Errorf("unmarshal other delete reply error response: %w", possibleError)
-		}
-
-		var repliesResp event.RepliesResponse
-		if err := json.Unmarshal(replyDataResp, &repliesResp); err != nil {
-			return nil, err
-		}
-		for _, reply := range repliesResp.Replies {
-			if _, err := repo.AddReply(reply.Reply); err != nil {
-				log.Errorf("failed to add reply to replies repo: %v", err)
-			}
-		}
-		return repliesResp, nil
+		return event.RepliesResponse{
+			Cursor:  cursor,
+			Replies: replies,
+			UserId:  &parentId,
+		}, nil
 	}
 }

core/handler/reply_test.go

@@ -267,12 +267,11 @@ func TestStreamNewReplyHandler(t *testing.T) {
 }
 
 func TestStreamGetRepliesHandler(t *testing.T) {
-	owner := "owner-1"
 	rootId := "root-1"
 	parentId := "parent-1"
 
 	t.Run("invalid payload", func(t *testing.T) {
-		h := StreamGetRepliesHandler(stubReplyRepo{}, stubReplyUserRepo{}, stubStreamer{})
+		h := StreamGetRepliesHandler(stubReplyRepo{})
 		_, err := h([]byte("{"), nil)
 		if err == nil {
 			t.Fatal("expected error")
@@ -281,17 +280,16 @@ func TestStreamGetRepliesHandler(t *testing.T) {
 
 	t.Run("empty parent id defaults to root", func(t *testing.T) {
 		// Top-level replies on a thread carry no parent_id from the
-		// client — the handler must fall back to root_id so the
-		// underlying repo lookup runs against the right key.
+		// client — the handler must fall back to root_id so the repo
+		// lookup runs against the first tier of replies hanging off
+		// the root tweet.
 		var seenRoot, seenParent string
 		h := StreamGetRepliesHandler(
 			stubReplyRepo{getRepliesTreeFn: func(rootID, parentIdArg string, _ *uint64, _ *string) ([]domain.ReplyNode, string, error) {
 				seenRoot = rootID
 				seenParent = parentIdArg
 				return nil, "", nil
 			}},
-			stubReplyUserRepo{},
-			stubStreamer{nodeInfo: warpnet.NodeInfo{OwnerId: rootId}},
 		)
 		_, err := h(marshal(t, event.GetAllRepliesEvent{RootId: rootId}), nil)
 		if err != nil {
@@ -303,18 +301,18 @@ func TestStreamGetRepliesHandler(t *testing.T) {
 	})
 
 	t.Run("empty root id", func(t *testing.T) {
-		h := StreamGetRepliesHandler(stubReplyRepo{}, stubReplyUserRepo{}, stubStreamer{})
+		h := StreamGetRepliesHandler(stubReplyRepo{})
 		_, err := h(marshal(t, event.GetAllRepliesEvent{ParentId: parentId}), nil)
 		if err == nil || err.Error() != "empty root id" {
 			t.Fatalf("unexpected err: %v", err)
 		}
 	})
 
-	t.Run("own tweet replies", func(t *testing.T) {
+	t.Run("serves replies from local repo", func(t *testing.T) {
 		replies := []domain.ReplyNode{{Reply: domain.Tweet{Id: "r1", Text: "reply"}}}
 		h := StreamGetRepliesHandler(stubReplyRepo{getRepliesTreeFn: func(rootID, parentIdArg string, limit *uint64, cursor *string) ([]domain.ReplyNode, string, error) {
 			return replies, "end", nil
-		}}, stubReplyUserRepo{}, stubStreamer{nodeInfo: warpnet.NodeInfo{OwnerId: parentId}})
+		}})
 		resp, err := h(marshal(t, event.GetAllRepliesEvent{RootId: rootId, ParentId: parentId}), nil)
 		if err != nil {
 			t.Fatalf("unexpected err: %v", err)
@@ -323,65 +321,19 @@ func TestStreamGetRepliesHandler(t *testing.T) {
 		if len(r.Replies) != 1 {
 			t.Fatalf("expected 1 reply, got %d", len(r.Replies))
 		}
-	})
-
-	t.Run("parent user not found fallback", func(t *testing.T) {
-		h := StreamGetRepliesHandler(stubReplyRepo{}, stubReplyUserRepo{getFn: func(userId string) (domain.User, error) {
-			return domain.User{}, database.ErrUserNotFound
-		}}, stubStreamer{nodeInfo: warpnet.NodeInfo{OwnerId: owner}})
-		resp, err := h(marshal(t, event.GetAllRepliesEvent{RootId: rootId, ParentId: parentId}), nil)
-		if err != nil {
-			t.Fatalf("unexpected err: %v", err)
+		if r.Cursor != "end" {
+			t.Fatalf("expected cursor 'end', got %q", r.Cursor)
 		}
-		_ = resp.(event.RepliesResponse)
 	})
 
-	t.Run("stream node offline fallback", func(t *testing.T) {
-		h := StreamGetRepliesHandler(stubReplyRepo{}, stubReplyUserRepo{}, stubStreamer{
-			nodeInfo: warpnet.NodeInfo{OwnerId: owner},
-			genericStreamFn: func(nodeId string, path stream.WarpRoute, data any) ([]byte, error) {
-				return nil, warpnet.ErrNodeIsOffline
-			},
-		})
-		resp, err := h(marshal(t, event.GetAllRepliesEvent{RootId: rootId, ParentId: parentId}), nil)
-		if err != nil {
-			t.Fatalf("unexpected err: %v", err)
-		}
-		_ = resp.(event.RepliesResponse)
-	})
-
-	t.Run("stream error", func(t *testing.T) {
-		streamErr := errors.New("broken")
-		h := StreamGetRepliesHandler(stubReplyRepo{}, stubReplyUserRepo{}, stubStreamer{
-			nodeInfo: warpnet.NodeInfo{OwnerId: owner},
-			genericStreamFn: func(nodeId string, path stream.WarpRoute, data any) ([]byte, error) {
-				return nil, streamErr
-			},
-		})
+	t.Run("propagates repo error", func(t *testing.T) {
+		boom := errors.New("db down")
+		h := StreamGetRepliesHandler(stubReplyRepo{getRepliesTreeFn: func(string, string, *uint64, *string) ([]domain.ReplyNode, string, error) {
+			return nil, "", boom
+		}})
 		_, err := h(marshal(t, event.GetAllRepliesEvent{RootId: rootId, ParentId: parentId}), nil)
-		if !errors.Is(err, streamErr) {
-			t.Fatalf("expected stream error: %v", err)
-		}
-	})
-
-	t.Run("remote successful response", func(t *testing.T) {
-		remoteResp, _ := json.Marshal(event.RepliesResponse{
-			Cursor:  "end",
-			Replies: []domain.ReplyNode{{Reply: domain.Tweet{Id: "r1", Text: "remote reply", RootId: rootId, ParentId: &parentId}}},
-		})
-		h := StreamGetRepliesHandler(stubReplyRepo{}, stubReplyUserRepo{}, stubStreamer{
-			nodeInfo: warpnet.NodeInfo{OwnerId: owner},
-			genericStreamFn: func(nodeId string, path stream.WarpRoute, data any) ([]byte, error) {
-				return remoteResp, nil
-			},
-		})
-		resp, err := h(marshal(t, event.GetAllRepliesEvent{RootId: rootId, ParentId: parentId}), nil)
-		if err != nil {
-			t.Fatalf("unexpected err: %v", err)
-		}
-		r := resp.(event.RepliesResponse)
-		if len(r.Replies) != 1 {
-			t.Fatalf("expected 1 reply: %v", r)
+		if !errors.Is(err, boom) {
+			t.Fatalf("expected db error, got %v", err)
 		}
 	})
 }

core/handler/self_request_test.go

@@ -60,9 +60,6 @@ func TestOwnerSelfRequest_NoOutboundStream(t *testing.T) {
 	ownerTweetUserRepo := stubTweetUserRepo{getFn: func(userId string) (domain.User, error) {
 		return domain.User{Id: userId, NodeId: ownerNodeID}, nil
 	}}
-	ownerReplyUserRepo := stubReplyUserRepo{getFn: func(userId string) (domain.User, error) {
-		return domain.User{Id: userId, NodeId: ownerNodeID}, nil
-	}}
 	ownerLikeUserRepo := stubLikeUserRepo{getFn: func(userId string) (domain.User, error) {
 		return domain.User{Id: userId, NodeId: ownerNodeID}, nil
 	}}
@@ -255,11 +252,9 @@ func TestOwnerSelfRequest_NoOutboundStream(t *testing.T) {
 	})
 
 	t.Run("StreamGetRepliesHandler - replies under own tweet", func(t *testing.T) {
-		streamer := stubStreamer{
-			nodeInfo:        ownerInfo,
-			genericStreamFn: failOnStream(t),
-		}
-		h := StreamGetRepliesHandler(stubReplyRepo{}, ownerReplyUserRepo, streamer)
+		// Replies handler is fully local now; no streamer/user repo
+		// involved (see reply.go for why).
+		h := StreamGetRepliesHandler(stubReplyRepo{})
 		if _, err := h(marshal(t, event.GetAllRepliesEvent{RootId: rootID, ParentId: owner}), nil); err != nil {
 			t.Fatalf("unexpected err: %v", err)
 		}

domain/warpnet.go

@@ -144,6 +144,9 @@ type ReplyNode struct {
 const RetweetPrefix = "RT:"
 
 // Tweet defines model for Tweet.
+//
+// ParentId is the parent TWEET id (not a user id) for replies; nil for
+// top-level tweets and for replies that hang directly off RootId.
 type Tweet struct {
 	CreatedAt time.Time  `json:"created_at"`
 	UpdatedAt *time.Time `json:"updated_at,omitempty"`

event/event.go

@@ -134,6 +134,11 @@ type GetAllMessagesEvent struct {
 }
 
 // GetAllRepliesEvent defines model for GetAllRepliesEvent.
+//
+// ParentId is the parent TWEET id (not a user id) — it selects which
+// subtree of replies inside RootId to return. Empty means "top-level
+// replies of the thread"; the handler treats that as ParentId = RootId.
+// RootId is the root tweet of the thread.
 type GetAllRepliesEvent struct {
 	Cursor   *string   `json:"cursor,omitempty"`
 	Limit    *uint64   `json:"limit,omitempty"`
@@ -284,6 +289,12 @@ type NewMessageEvent = domain.ChatMessage
 type NewMessageResponse = domain.ChatMessage
 
 // NewReplyEvent defines model for NewReplyEvent.
+//
+// ParentId is the parent TWEET id this reply is attached to (nil/empty
+// means the reply hangs directly off RootId). ParentUserId is the user
+// id of the parent tweet's author — that's the routing key the server
+// uses to forward the request to the right node when the parent tweet
+// lives on a remote peer.
 type NewReplyEvent struct {
 	CreatedAt    time.Time  `json:"created_at"`
 	Id           domain.ID  `json:"id"`

version

@@ -1 +1 @@
-0.7.17
+0.7.18