feat(store): owner-set visibility and derived TREE members on channel reads (RIG-4188)

docs/specs/product/compass.md

@@ -539,9 +539,11 @@ carries a caller identity.
 
 The server SHALL scope every listing, read, and search to the channels, groups,
 and accounts the caller may see, enforced in the store (SQL), not at the RPC
-edge. A message read for a channel the caller is not a member of SHALL return
-nothing rather than the channel's contents — a non-member cannot read a private
-channel's history by naming its id.
+edge. An attached channel is visible to its members and to the anchor agent's
+owner set. That visibility does not grant history access: message reads, search,
+topic reads, and writes require channel participation. TREE participation is the
+anchor agent, its owner, and the agents in the anchor's subtree. A caller who
+does not participate SHALL receive no channel history by naming its id.
 
 #### Scenario: A non-member lists a private channel's messages
 

go/internal/comms/channel_attach_pgtest_test.go

@@ -0,0 +1,49 @@
+//go:build pgtest
+
+package comms
+
+import (
+	"testing"
+
+	compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1"
+	"github.com/RigelBuild/compass/go/internal/store"
+)
+
+func TestPublishChannelChangedCarriesTreeAttachment(t *testing.T) {
+	h := newStreamHarness(t)
+	owner := mustUser(t, h.store, "tree-event-owner")
+	anchor := mustAgent(t, h.store, owner.ID, "tree-event-anchor")
+	channel, err := h.store.CreateChannel(t.Context(), owner.ID, store.NewChannel{
+		Name: "tree-event", Kind: store.ChannelKindChannel,
+		ParentAgentID: anchor.ID, MembershipMode: store.ChannelMembershipModeTree,
+	})
+	if err != nil {
+		t.Fatalf("CreateChannel: %v", err)
+	}
+
+	sub, err := h.bus.Subscribe(0, 0)
+	if err != nil {
+		t.Fatalf("subscribe event bus: %v", err)
+	}
+	defer sub.Cancel()
+
+	h.svc.publishChannelChanged(channel, nil)
+	select {
+	case event := <-sub.Live:
+		changed := event.Payload.GetChannelChanged()
+		if changed == nil {
+			t.Fatalf("published payload = %T, want ChannelChanged", event.Payload.GetPayload())
+		}
+		got := changed.GetChannel()
+		if got.GetParentAgentId() != string(anchor.ID) || got.GetMembershipMode() != compassv1.ChannelMembershipMode_CHANNEL_MEMBERSHIP_MODE_TREE {
+			t.Fatalf("ChannelChanged channel parent=%q mode=%v, want parent=%q TREE", got.GetParentAgentId(), got.GetMembershipMode(), anchor.ID)
+		}
+	default:
+		t.Fatal("publishChannelChanged emitted no event")
+	}
+	select {
+	case event := <-sub.Live:
+		t.Fatalf("publishChannelChanged emitted duplicate event %T", event.Payload.GetPayload())
+	default:
+	}
+}

go/internal/comms/channel_attach_wire_test.go

@@ -7,6 +7,7 @@ import (
 	"google.golang.org/protobuf/reflect/protoreflect"
 
 	compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1"
+	"github.com/RigelBuild/compass/go/internal/store"
 )
 
 // The channel-attach fields are additive: their numbers are the wire contract
@@ -69,3 +70,13 @@ func TestChannelAttachRoundTrip(t *testing.T) {
 		t.Fatalf("round trip: got %v, want %v", &out, in)
 	}
 }
+
+func TestChannelToWireCarriesTreeAttachment(t *testing.T) {
+	got := channelToWire(store.Channel{
+		ParentAgentID:  "agent-anchor",
+		MembershipMode: store.ChannelMembershipModeTree,
+	})
+	if got.GetParentAgentId() != "agent-anchor" || got.GetMembershipMode() != compassv1.ChannelMembershipMode_CHANNEL_MEMBERSHIP_MODE_TREE {
+		t.Fatalf("channelToWire attachment = parent %q mode %v, want agent-anchor and TREE", got.GetParentAgentId(), got.GetMembershipMode())
+	}
+}

go/internal/comms/mapping.go

@@ -77,9 +77,17 @@ func channelToWire(c store.Channel) *compassv1.Channel {
 		PostPolicy:            channelPostPolicyToWire(c.Policy.PostPolicy),
 		OwnerAccountId:        string(c.Policy.OwnerAccountID),
 		MandatorySubscription: c.Policy.MandatorySubscription,
+		ParentAgentId:         string(c.ParentAgentID),
+		MembershipMode:        channelMembershipModeToWire(c.MembershipMode),
 	}
 }
 
+func channelMembershipModeToWire(m store.ChannelMembershipMode) compassv1.ChannelMembershipMode {
+	if m == store.ChannelMembershipModeTree {
+		return compassv1.ChannelMembershipMode_CHANNEL_MEMBERSHIP_MODE_TREE
+	}
+	return compassv1.ChannelMembershipMode_CHANNEL_MEMBERSHIP_MODE_EXPLICIT
+}
 func channelKindToWire(k store.ChannelKind) compassv1.ChannelKind {
 	switch k {
 	case store.ChannelKindDM:

go/internal/store/channel_by_name_pgtest_test.go

@@ -109,3 +109,38 @@ func TestChannelByNameForViewerAmbiguousIsInvalidArgument(t *testing.T) {
 		t.Fatalf("ambiguous resolve returned ErrNotFound, want ErrInvalidArgument only")
 	}
 }
+
+func TestChannelByNameForViewerNarrowsOwnerSetDuplicatesToParticipants(t *testing.T) {
+	s := newTestStore(t)
+	owner := mustUser(t, s, "standup-owner")
+	anchorA := mustAgent(t, s, owner.ID, "standup-a")
+	anchorB := mustAgent(t, s, owner.ID, "standup-b")
+	channelA, err := s.CreateChannel(t.Context(), owner.ID, NewChannel{
+		Name: "standup", Kind: ChannelKindChannel,
+		ParentAgentID: anchorA.ID, MembershipMode: ChannelMembershipModeTree,
+	})
+	if err != nil {
+		t.Fatalf("CreateChannel(A): %v", err)
+	}
+	if _, err := s.CreateChannel(t.Context(), owner.ID, NewChannel{
+		Name: "standup", Kind: ChannelKindChannel,
+		ParentAgentID: anchorB.ID, MembershipMode: ChannelMembershipModeTree,
+	}); err != nil {
+		t.Fatalf("CreateChannel(B): %v", err)
+	}
+
+	got, err := s.ChannelByNameForViewer(t.Context(), anchorA.ID, "standup")
+	if err != nil {
+		t.Fatalf("ChannelByNameForViewer(A1): %v", err)
+	}
+	if got.ID != channelA.ID {
+		t.Fatalf("A1 resolved channel %q, want its own %q", got.ID, channelA.ID)
+	}
+	_, err = s.ChannelByNameForViewer(t.Context(), owner.ID, "standup")
+	sentinelIs(t, err, ErrInvalidArgument, "owner resolves both standup channels")
+	// A sibling that sees both but participates in neither keeps the visible-set
+	// ambiguity rather than a not-found for a name its list shows.
+	sibling := mustAgent(t, s, owner.ID, "standup-c")
+	_, err = s.ChannelByNameForViewer(t.Context(), sibling.ID, "standup")
+	sentinelIs(t, err, ErrInvalidArgument, "non-participant sibling resolves both standup channels")
+}

go/internal/store/channel_participant_pgtest_test.go

@@ -4,8 +4,11 @@ package store
 
 import (
 	"context"
+	"reflect"
 	"testing"
 	"time"
+
+	"github.com/RigelBuild/compass/go/internal/store/db"
 )
 
 func appendAsParticipant(t *testing.T, s *Store, author AccountID, channel ChannelID, text string) (Message, error) {
@@ -95,6 +98,225 @@ func TestChannelParticipantTreeArm(t *testing.T) {
 	}
 }
 
+func TestChannelVisibilityOwnerSetParity(t *testing.T) {
+	s := newTestStore(t)
+	owner := mustUser(t, s, "visibility-owner")
+	anchor := mustAgent(t, s, owner.ID, "visibility-anchor")
+	sibling := mustAgent(t, s, owner.ID, "visibility-sibling")
+	foreign := mustUser(t, s, "visibility-foreign")
+	channel := mustAttachedChannel(t, s, owner.ID, anchor.ID, "visibility-tree", ChannelMembershipModeTree)
+
+	for _, tc := range []struct {
+		name string
+		who  AccountID
+		want bool
+	}{
+		{name: "owner", who: owner.ID, want: true},
+		{name: "same-owner sibling", who: sibling.ID, want: true},
+		{name: "foreign user", who: foreign.ID},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			channels, err := s.ListChannels(t.Context(), tc.who)
+			if err != nil {
+				t.Fatalf("ListChannels: %v", err)
+			}
+			listed := false
+			for _, got := range channels {
+				if got.ID == channel.ID {
+					listed = true
+				}
+			}
+			visible, err := s.ChannelVisibleTo(t.Context(), tc.who, channel.ID)
+			if err != nil {
+				t.Fatalf("ChannelVisibleTo: %v", err)
+			}
+			if listed != tc.want || visible != tc.want || visible != listed {
+				t.Fatalf("owner-set visibility: ListChannels=%v ChannelVisibleTo=%v, want %v", listed, visible, tc.want)
+			}
+		})
+	}
+}
+
+func TestChannelTreeParticipantReadsAndWrites(t *testing.T) {
+	s := newTestStore(t)
+	f := newTreeFixture(t, s)
+
+	message, err := appendAsParticipant(t, s, f.leaf.ID, f.channel.ID, "subtree searchable history")
+	if err != nil {
+		t.Fatalf("AppendMessage by subtree agent: %v", err)
+	}
+
+	listed, err := s.ListMessages(t.Context(), ListMessagesQuery{Actor: f.leaf.ID, ChannelID: f.channel.ID})
+	if err != nil || len(listed) != 1 || listed[0].ID != message.ID {
+		t.Fatalf("ListMessages for subtree agent = %v, %v; want message %q", listed, err, message.ID)
+	}
+
+	searched, err := s.SearchMessages(t.Context(), f.leaf.ID, SearchScope{ChannelID: f.channel.ID}, "subtree searchable", Page{})
+	if err != nil || len(searched) != 1 || searched[0].ID != message.ID {
+		t.Fatalf("SearchMessages for subtree agent = %v, %v; want message %q", searched, err, message.ID)
+	}
+	ownerSearch, err := s.SearchMessages(t.Context(), f.owner.ID, SearchScope{ChannelID: f.channel.ID}, "subtree searchable", Page{})
+	if err != nil || len(ownerSearch) != 1 || ownerSearch[0].ID != message.ID {
+		t.Fatalf("SearchMessages for anchor owner = %v, %v; want message %q", ownerSearch, err, message.ID)
+	}
+	unscoped, err := s.SearchMessages(t.Context(), f.leaf.ID, SearchScope{}, "subtree searchable", Page{})
+	if err != nil || len(unscoped) != 1 || unscoped[0].ID != message.ID {
+		t.Fatalf("unscoped SearchMessages for subtree agent = %v, %v; want message %q", unscoped, err, message.ID)
+	}
+	ownerListed, err := s.ListMessages(t.Context(), ListMessagesQuery{Actor: f.owner.ID, ChannelID: f.channel.ID})
+	if err != nil || len(ownerListed) != 1 || ownerListed[0].ID != message.ID {
+		t.Fatalf("ListMessages for anchor owner = %v, %v; want message %q", ownerListed, err, message.ID)
+	}
+
+	cursorSeq, err := s.q.GetPageCursorSeq(t.Context(), db.GetPageCursorSeqParams{
+		AccountID: string(f.leaf.ID), ID: string(message.ID), ChannelID: string(f.channel.ID),
+	})
+	if err != nil || cursorSeq == 0 {
+		t.Fatalf("GetPageCursorSeq for subtree agent = %d, %v; want message sequence", cursorSeq, err)
+	}
+	page, err := s.ListMessages(t.Context(), ListMessagesQuery{Actor: f.leaf.ID, ChannelID: f.channel.ID, Page: Page{BeforeMessageID: message.ID}})
+	if err != nil || len(page) != 0 {
+		t.Fatalf("ListMessages cursor for subtree agent = %v, %v; want empty page", page, err)
+	}
+
+	if _, err := s.UpdateMessageBlocksAsAuthor(t.Context(), f.leaf.ID, message.ID, []MessageBlock{textBlock("edited by subtree author")}); err != nil {
+		t.Fatalf("UpdateMessageBlocksAsAuthor by subtree author: %v", err)
+	}
+	if _, err := s.UpdateTopic(t.Context(), string(f.leaf.ID), listed[0].TopicID, nil, nil); err != nil {
+		t.Fatalf("UpdateTopic by subtree agent: %v", err)
+	}
+
+	if _, err := s.ReparentAgent(t.Context(), f.owner.ID, f.leaf.ID, ""); err != nil {
+		t.Fatalf("ReparentAgent(leaf out of channel subtree): %v", err)
+	}
+	if _, err := s.UpdateMessageBlocksAsAuthor(t.Context(), f.leaf.ID, message.ID, []MessageBlock{textBlock("reparented author edit")}); err == nil {
+		t.Fatal("UpdateMessageBlocksAsAuthor by reparented author succeeded")
+	} else {
+		sentinelIs(t, err, ErrNotFound, "reparented author UpdateMessageBlocksAsAuthor")
+	}
+}
+
+func TestChannelTreeOwnerSetSiblingReadWriteDenials(t *testing.T) {
+	s := newTestStore(t)
+	f := newTreeFixture(t, s)
+	message, err := appendAsParticipant(t, s, f.leaf.ID, f.channel.ID, "subtree searchable history")
+	if err != nil {
+		t.Fatalf("AppendMessage by subtree agent: %v", err)
+	}
+
+	ownerOnly, err := s.ListMessages(t.Context(), ListMessagesQuery{Actor: f.sibling.ID, ChannelID: f.channel.ID})
+	if err != nil || len(ownerOnly) != 0 {
+		t.Fatalf("owner-set sibling history = %v, %v; want no messages", ownerOnly, err)
+	}
+	ownerSearch, err := s.SearchMessages(t.Context(), f.sibling.ID, SearchScope{ChannelID: f.channel.ID}, "subtree searchable", Page{})
+	if err != nil || len(ownerSearch) != 0 {
+		t.Fatalf("owner-set sibling search = %v, %v; want no messages", ownerSearch, err)
+	}
+	unscopedSearch, err := s.SearchMessages(t.Context(), f.sibling.ID, SearchScope{}, "subtree searchable", Page{})
+	if err != nil || len(unscopedSearch) != 0 {
+		t.Fatalf("owner-set sibling unscoped search = %v, %v; want no messages", unscopedSearch, err)
+	}
+	ownerCursorSeq, err := s.q.GetPageCursorSeq(t.Context(), db.GetPageCursorSeqParams{
+		AccountID: string(f.sibling.ID), ID: string(message.ID), ChannelID: string(f.channel.ID),
+	})
+	if !noRows(err) {
+		t.Fatalf("GetPageCursorSeq for owner-set sibling = %d, %v; want no rows", ownerCursorSeq, err)
+	}
+	_, err = s.UpdateTopic(t.Context(), string(f.sibling.ID), message.TopicID, nil, nil)
+	sentinelIs(t, err, ErrNotFound, "owner-set sibling UpdateTopic")
+
+	outsiderMessageID := newID()
+	if _, err := s.scopedPool().Exec(t.Context(), `
+		INSERT INTO messages (id, topic_id, author_account_id, at_unix_ms, blocks, text_content)
+		VALUES ($1, $2, $3, 1, '[{"kind":"text","text":"outsider authored"}]'::jsonb, 'outsider authored')`,
+		outsiderMessageID, message.TopicID, string(f.sibling.ID)); err != nil {
+		t.Fatalf("insert outsider-authored message: %v", err)
+	}
+	if _, err := s.UpdateMessageBlocksAsAuthor(t.Context(), f.sibling.ID, MessageID(outsiderMessageID), []MessageBlock{textBlock("unauthorized edit")}); err == nil {
+		t.Fatal("owner-set sibling edited a message without channel participation")
+	} else {
+		sentinelIs(t, err, ErrNotFound, "owner-set sibling UpdateMessageBlocksAsAuthor")
+	}
+
+	_, _, err = s.AppendMessage(t.Context(), Message{AuthorAccountID: f.leaf.ID, Blocks: []MessageBlock{pendingAsk("tree-ask", false)}}, string(f.channel.ID), TopicRef{Name: "asks", Create: true}, "")
+	if err != nil {
+		t.Fatalf("AppendMessage(ask): %v", err)
+	}
+	askFilter, err := askIDContainmentFilter("tree-ask")
+	if err != nil {
+		t.Fatalf("askIDContainmentFilter: %v", err)
+	}
+	if rows, err := s.q.FindAskMessage(t.Context(), db.FindAskMessageParams{AccountID: string(f.sibling.ID), Column2: askFilter}); err != nil || len(rows) != 0 {
+		t.Fatalf("FindAskMessage for owner-set sibling = %d rows, %v; want none", len(rows), err)
+	}
+	_, _, err = s.AnswerAsk(t.Context(), f.sibling.ID, "tree-ask", []AskAnswer{{QuestionID: "q1", ChosenOptionIDs: []string{"opt-a"}}})
+	sentinelIs(t, err, ErrNotFound, "owner-set sibling AnswerAsk")
+
+	if rows, err := s.q.FindAskMessage(t.Context(), db.FindAskMessageParams{AccountID: string(f.leaf.ID), Column2: askFilter}); err != nil || len(rows) != 1 {
+		t.Fatalf("FindAskMessage for subtree agent = %d rows, %v; want one ask", len(rows), err)
+	}
+	if _, _, err := s.AnswerAsk(t.Context(), f.leaf.ID, "tree-ask", []AskAnswer{{QuestionID: "q1", ChosenOptionIDs: []string{"opt-a"}}}); err != nil {
+		t.Fatalf("AnswerAsk by subtree agent: %v", err)
+	}
+}
+
+func TestChannelTreeMembersAreAttributedAndSubscriptionsIntersect(t *testing.T) {
+	s := newTestStore(t)
+	owner := mustUser(t, s, "materialize-owner")
+	rootA := mustAgent(t, s, owner.ID, "materialize-a")
+	childA := mustAgentWithParent(t, s, owner.ID, rootA.ID, "materialize-a-child")
+	rootB := mustAgent(t, s, owner.ID, "materialize-b")
+	childB := mustAgentWithParent(t, s, owner.ID, rootB.ID, "materialize-b-child")
+	otherOwner := mustUser(t, s, "materialize-other")
+	outside := mustAgent(t, s, otherOwner.ID, "materialize-outside")
+	channelA := mustAttachedChannel(t, s, owner.ID, rootA.ID, "materialize-tree-a", ChannelMembershipModeTree)
+	channelB := mustAttachedChannel(t, s, owner.ID, rootB.ID, "materialize-tree-b", ChannelMembershipModeTree)
+
+	if _, err := s.ReparentAgent(t.Context(), owner.ID, childA.ID, rootB.ID); err != nil {
+		t.Fatalf("ReparentAgent(child A out): %v", err)
+	}
+	if _, err := s.scopedPool().Exec(t.Context(),
+		`INSERT INTO channel_subscriptions (channel_id, account_id, subscribed) VALUES ($1, $2, TRUE)`,
+		string(channelA.ID), string(childA.ID)); err != nil {
+		t.Fatalf("insert stale subscription: %v", err)
+	}
+	if _, err := s.scopedPool().Exec(t.Context(),
+		`INSERT INTO channel_subscriptions (channel_id, account_id, subscribed) VALUES ($1, $2, TRUE)`,
+		string(channelA.ID), string(rootA.ID)); err != nil {
+		t.Fatalf("insert live subscription: %v", err)
+	}
+	if _, err := s.scopedPool().Exec(t.Context(),
+		`INSERT INTO channel_subscriptions (channel_id, account_id, subscribed) VALUES ($1, $2, TRUE)`,
+		string(channelA.ID), string(outside.ID)); err != nil {
+		t.Fatalf("insert outside stale subscription: %v", err)
+	}
+
+	channels, err := s.ListChannels(t.Context(), owner.ID)
+	if err != nil {
+		t.Fatalf("ListChannels: %v", err)
+	}
+	got := make(map[ChannelID]map[AccountID]bool)
+	for _, channel := range channels {
+		if channel.ID == channelA.ID || channel.ID == channelB.ID {
+			got[channel.ID] = memberSet(channel)
+			if channel.ID == channelA.ID && !reflect.DeepEqual(channel.SubscriberAccountIDs, []AccountID{rootA.ID}) {
+				t.Fatalf("channel A subscribers = %v, want only %s", channel.SubscriberAccountIDs, rootA.ID)
+			}
+			if channel.ID == channelB.ID && len(channel.SubscriberAccountIDs) != 0 {
+				t.Fatalf("channel B subscribers = %v, want none", channel.SubscriberAccountIDs)
+			}
+		}
+	}
+	wantA := map[AccountID]bool{owner.ID: true, rootA.ID: true}
+	wantB := map[AccountID]bool{owner.ID: true, rootB.ID: true, childA.ID: true, childB.ID: true}
+	if !reflect.DeepEqual(got[channelA.ID], wantA) || !reflect.DeepEqual(got[channelB.ID], wantB) {
+		t.Fatalf("derived participants are misattributed: A=%v want %v; B=%v want %v", got[channelA.ID], wantA, got[channelB.ID], wantB)
+	}
+	if n := countChannelMembers(t, s, channelA.ID); n != 0 {
+		t.Fatalf("TREE channel has %d stored member rows, want 0", n)
+	}
+}
+
 func TestChannelParticipantFollowsReparentAgent(t *testing.T) {
 	s := newTestStore(t)
 	f := newTreeFixture(t, s)