feat(store): authorize channel writes and reads by tree participation (RIG-4187)

go/internal/store/authz.go

@@ -7,26 +7,14 @@ import (
 	"github.com/RigelBuild/compass/go/internal/store/db"
 )
 
-// requireChannelMember is the D9 write-authorization primitive: it verifies the
-// actor is a member of channelID and returns ErrNotFound if not. This mirrors
-// the read paths' membership gate (ListMessages/SearchMessages/AnswerAsk JOIN
-// channel_members) so a write authorizes against the same visible set a read
-// does — a caller who cannot see a channel cannot mutate it either, and the
-// refusal is the not-found/forbidden merge (a non-member cannot tell an
-// unauthorized channel apart from a nonexistent one, so a probe enumerates
-// nothing).
-//
-// It runs against either the pool or an open transaction (querier), so a
-// mutation can gate inside its own tx before touching state — the D9 discipline
-// the design record requires on every write RPC ("authorized server-side
-// against the authenticated account's visible set", design.md:1101-1102).
+// requireChannelMember is the D9 write gate: the actor must participate in
+// channelID (isChannelMember), else ErrNotFound, so a probe cannot tell a
+// forbidden channel from a missing one. It takes the pool or an open tx so a
+// mutation gates inside its own transaction.
 func requireChannelMember(ctx context.Context, q db.DBTX, actor AccountID, channelID ChannelID) error {
-	member, err := db.New(q).ChannelMemberExists(ctx, db.ChannelMemberExistsParams{
-		ChannelID: string(channelID),
-		AccountID: string(actor),
-	})
+	member, err := isChannelMember(ctx, q, actor, channelID)
 	if err != nil {
-		return fmt.Errorf("store: check channel membership: %w", err)
+		return err
 	}
 	if !member {
 		// The not-found/forbidden merge: a non-member is told the channel does
@@ -36,46 +24,40 @@ func requireChannelMember(ctx context.Context, q db.DBTX, actor AccountID, chann
 	return nil
 }
 
-// IsChannelMember reports whether actor is a member of channelID. It is the
-// exported form used by the SubscribeComms stream edge to filter each
-// fanned-out event by the subscriber's visible set (a non-member never receives
-// an event for a channel it cannot see) without turning a non-visible event
-// into an error — the D9 discipline extended from the read RPCs to the live
-// stream (design.md:446-447: the fan-out is visibility-scoped).
+// IsChannelMember reports whether actor participates in channelID (a member row
+// or TREE derivation). The SubscribeComms edge uses it to drop events the
+// subscriber cannot see instead of returning an error.
 func (s *Store) IsChannelMember(ctx context.Context, actor AccountID, channelID ChannelID) (bool, error) {
 	return isChannelMember(ctx, s.scopedPool(), actor, channelID)
 }
 
-// isChannelMember reports whether actor is a member of channelID (the
-// package-internal form IsChannelMember exports and requireChannelMember wraps).
+// isChannelMember reports whether actor participates in channelID: a member
+// row, or for a TREE channel the anchor's subtree or owner (ChannelParticipant).
+// It is the package-internal form IsChannelMember exports and
+// requireChannelMember wraps.
 func isChannelMember(ctx context.Context, q db.DBTX, actor AccountID, channelID ChannelID) (bool, error) {
-	member, err := db.New(q).ChannelMemberExists(ctx, db.ChannelMemberExistsParams{
+	participant, err := db.New(q).ChannelParticipant(ctx, db.ChannelParticipantParams{
 		ChannelID: string(channelID),
 		AccountID: string(actor),
 	})
 	if err != nil {
 		return false, fmt.Errorf("store: check channel membership: %w", err)
 	}
-	return member, nil
+	return participant.Valid && participant.Bool, nil
 }
 
-// IsTopicChannelMember reports whether actor is a member of the channel that
-// owns topicID. It is the topic-scoped form the SubscribeComms stream edge uses
-// to gate MessagePosted/MessageUpdated now that a wire message carries only a
-// topic, not a channel: the channel is resolved through topics.channel_id (the
-// design record's "a consumer that needs the channel resolves it through the
-// topic"), so the per-event filter stays at read-parity with ListMessages
-// (which JOINs channel_members on the topic's channel). An unknown topic yields
-// false (not visible) — the not-found/forbidden merge extended to the stream.
+// IsTopicChannelMember reports whether actor participates in the channel that
+// owns topicID, resolved through topics.channel_id, for stream events that carry
+// only a topic. An unknown topic yields false.
 func (s *Store) IsTopicChannelMember(ctx context.Context, actor AccountID, topicID string) (bool, error) {
-	member, err := s.q.TopicChannelMemberExists(ctx, db.TopicChannelMemberExistsParams{
+	participant, err := s.q.TopicChannelParticipant(ctx, db.TopicChannelParticipantParams{
 		ID:        topicID,
 		AccountID: string(actor),
 	})
 	if err != nil {
 		return false, fmt.Errorf("store: check topic channel membership: %w", err)
 	}
-	return member, nil
+	return participant.Valid && participant.Bool, nil
 }
 
 // requireGroupCreateAuthz authorizes creating a channel inside groupID. The

go/internal/store/channel_participant_pgtest_test.go

@@ -0,0 +1,179 @@
+//go:build pgtest
+
+package store
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func appendAsParticipant(t *testing.T, s *Store, author AccountID, channel ChannelID, text string) (Message, error) {
+	t.Helper()
+	m, _, err := s.AppendMessage(t.Context(), Message{
+		AuthorAccountID: author, Blocks: []MessageBlock{textBlock(text)},
+	}, string(channel), TopicRef{Name: "general", Create: true}, "")
+	return m, err
+}
+
+func countChannelMembers(t *testing.T, s *Store, id ChannelID) int {
+	t.Helper()
+	var n int
+	if err := s.pool.QueryRow(t.Context(),
+		`SELECT count(*) FROM channel_members WHERE channel_id = $1`, string(id)).Scan(&n); err != nil {
+		t.Fatalf("count channel members: %v", err)
+	}
+	return n
+}
+
+// treeFixture: owner → root → mid → leaf, plus a sibling outside root's subtree
+// and a foreign user; the TREE channel is anchored at root.
+type treeFixture struct {
+	owner, foreign           Account
+	root, mid, leaf, sibling Account
+	channel                  Channel
+}
+
+func newTreeFixture(t *testing.T, s *Store) treeFixture {
+	t.Helper()
+	owner := mustUser(t, s, "probe-owner")
+	root := mustAgent(t, s, owner.ID, "probe-root")
+	mid := mustAgentWithParent(t, s, owner.ID, root.ID, "probe-mid")
+	return treeFixture{
+		owner:   owner,
+		foreign: mustUser(t, s, "probe-foreign"),
+		root:    root,
+		mid:     mid,
+		leaf:    mustAgentWithParent(t, s, owner.ID, mid.ID, "probe-leaf"),
+		sibling: mustAgent(t, s, owner.ID, "probe-sibling"),
+		channel: mustAttachedChannel(t, s, owner.ID, root.ID, "probe-tree", ChannelMembershipModeTree),
+	}
+}
+
+func TestChannelParticipantTreeArm(t *testing.T) {
+	s := newTestStore(t)
+	f := newTreeFixture(t, s)
+
+	for _, author := range []Account{f.owner, f.root, f.mid, f.leaf} {
+		if _, err := appendAsParticipant(t, s, author.ID, f.channel.ID, "from "+author.Handle); err != nil {
+			t.Fatalf("AppendMessage by participant %s: %v", author.Handle, err)
+		}
+	}
+	for _, outsider := range []Account{f.sibling, f.foreign} {
+		_, err := appendAsParticipant(t, s, outsider.ID, f.channel.ID, "intrusion")
+		sentinelIs(t, err, ErrNotFound, "post by non-participant "+outsider.Handle)
+	}
+
+	topics, err := s.ListTopics(t.Context(), string(f.leaf.ID), string(f.channel.ID), false)
+	if err != nil {
+		t.Fatalf("ListTopics by a subtree agent: %v", err)
+	}
+	if len(topics) != 1 || topics[0].Name != "general" {
+		t.Fatalf("ListTopics by a subtree agent = %+v, want the general topic", topics)
+	}
+	_, err = s.ListTopics(t.Context(), string(f.sibling.ID), string(f.channel.ID), false)
+	sentinelIs(t, err, ErrNotFound, "ListTopics by an agent outside the subtree")
+
+	for _, tc := range []struct {
+		who  Account
+		want bool
+	}{{f.leaf, true}, {f.owner, true}, {f.sibling, false}, {f.foreign, false}} {
+		got, err := s.IsChannelMember(t.Context(), tc.who.ID, f.channel.ID)
+		if err != nil || got != tc.want {
+			t.Fatalf("IsChannelMember(%s) = %v, %v; want %v", tc.who.Handle, got, err, tc.want)
+		}
+		got, err = s.IsTopicChannelMember(t.Context(), tc.who.ID, topics[0].ID)
+		if err != nil || got != tc.want {
+			t.Fatalf("IsTopicChannelMember(%s) = %v, %v; want %v", tc.who.Handle, got, err, tc.want)
+		}
+	}
+	if got, err := s.IsTopicChannelMember(t.Context(), f.owner.ID, "missing-topic"); err != nil || got {
+		t.Fatalf("IsTopicChannelMember(unknown topic) = %v, %v; want false", got, err)
+	}
+	if n := countChannelMembers(t, s, f.channel.ID); n != 0 {
+		t.Fatalf("tree channel member rows = %d, want 0 (participation is derived)", n)
+	}
+}
+
+func TestChannelParticipantFollowsReparentAgent(t *testing.T) {
+	s := newTestStore(t)
+	f := newTreeFixture(t, s)
+	if _, err := appendAsParticipant(t, s, f.leaf.ID, f.channel.ID, "before move"); err != nil {
+		t.Fatalf("post before move: %v", err)
+	}
+	before := countChannelMembers(t, s, f.channel.ID)
+
+	if _, err := s.ReparentAgent(t.Context(), f.owner.ID, f.leaf.ID, f.sibling.ID); err != nil {
+		t.Fatalf("ReparentAgent(leaf → sibling): %v", err)
+	}
+	_, err := appendAsParticipant(t, s, f.leaf.ID, f.channel.ID, "after move")
+	sentinelIs(t, err, ErrNotFound, "post after the agent left the subtree")
+	if after := countChannelMembers(t, s, f.channel.ID); after != before {
+		t.Fatalf("member rows changed on agent reparent: %d → %d, want no reconcile", before, after)
+	}
+}
+
+func TestChannelParticipantExplicitChannelIgnoresTree(t *testing.T) {
+	s := newTestStore(t)
+	f := newTreeFixture(t, s)
+	explicit := mustAttachedChannel(t, s, f.owner.ID, f.root.ID, "probe-explicit", ChannelMembershipModeExplicit, f.root.ID)
+
+	_, err := appendAsParticipant(t, s, f.mid.ID, explicit.ID, "not a member")
+	sentinelIs(t, err, ErrNotFound, "post by a subtree agent into an EXPLICIT channel")
+	posted, err := appendAsParticipant(t, s, f.root.ID, explicit.ID, "member row")
+	if err != nil {
+		t.Fatalf("post by the explicit member anchor: %v", err)
+	}
+	if got, err := s.IsTopicChannelMember(t.Context(), f.mid.ID, posted.TopicID); err != nil || got {
+		t.Fatalf("IsTopicChannelMember(subtree agent, explicit topic) = %v, %v; want false", got, err)
+	}
+}
+
+// hasGenuineAdd needs stored-row semantics: a derived participant with no
+// member row is still a genuine add.
+func TestHasGenuineAddIgnoresDerivedParticipation(t *testing.T) {
+	s := newTestStore(t)
+	f := newTreeFixture(t, s)
+	tx, err := s.beginTenantTx(t.Context())
+	if err != nil {
+		t.Fatalf("begin: %v", err)
+	}
+	defer func() { _ = tx.Rollback(t.Context()) }()
+
+	add, err := hasGenuineAdd(t.Context(), tx, f.channel.ID, []MemberUpdate{{AccountID: f.leaf.ID, Subscribed: true}})
+	if err != nil {
+		t.Fatalf("hasGenuineAdd: %v", err)
+	}
+	if !add {
+		t.Fatal("hasGenuineAdd(derived participant without a row) = false, want true")
+	}
+}
+
+// The schema does not forbid parent cycles (the store refuses them on write), so
+// a raw-SQL cycle proves the UNION walk in both probes terminates.
+func TestChannelParticipantTerminatesOnParentCycle(t *testing.T) {
+	s := newTestStore(t)
+	f := newTreeFixture(t, s)
+	if _, err := appendAsParticipant(t, s, f.owner.ID, f.channel.ID, "seed"); err != nil {
+		t.Fatalf("seed post: %v", err)
+	}
+	topics, err := s.ListTopics(t.Context(), string(f.owner.ID), string(f.channel.ID), false)
+	if err != nil || len(topics) != 1 {
+		t.Fatalf("ListTopics = %v, %v; want one topic", topics, err)
+	}
+	a := mustAgent(t, s, f.owner.ID, "cycle-a")
+	b := mustAgentWithParent(t, s, f.owner.ID, a.ID, "cycle-b")
+	if _, err := s.pool.Exec(t.Context(),
+		`UPDATE agent_accounts SET parent_agent_id = $2 WHERE account_id = $1`, string(a.ID), string(b.ID)); err != nil {
+		t.Fatalf("close the parent cycle: %v", err)
+	}
+
+	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+	defer cancel()
+	if got, err := s.IsChannelMember(ctx, b.ID, f.channel.ID); err != nil || got {
+		t.Fatalf("IsChannelMember(cyclic agent) = %v, %v; want false", got, err)
+	}
+	if got, err := s.IsTopicChannelMember(ctx, b.ID, topics[0].ID); err != nil || got {
+		t.Fatalf("IsTopicChannelMember(cyclic agent) = %v, %v; want false", got, err)
+	}
+}

go/internal/store/channels.go

@@ -289,7 +289,7 @@ func (s *Store) ReparentChannel(ctx context.Context, actor AccountID, channelID
 	defer func() { _ = tx.Rollback(ctx) }()
 	qtx := s.q.WithTx(tx)
 
-	if err := requireChannelParticipant(ctx, qtx, actor, channelID); err != nil {
+	if err := requireChannelMember(ctx, tx, actor, channelID); err != nil {
 		return Channel{}, err
 	}
 	row, err := qtx.LockChannelForReparent(ctx, string(channelID))
@@ -299,7 +299,7 @@ func (s *Store) ReparentChannel(ctx context.Context, actor AccountID, channelID
 		}
 		return Channel{}, fmt.Errorf("store: lock channel for reparent: %w", err)
 	}
-	if err := requireChannelParticipant(ctx, qtx, actor, channelID); err != nil {
+	if err := requireChannelMember(ctx, tx, actor, channelID); err != nil {
 		return Channel{}, err
 	}
 
@@ -349,22 +349,6 @@ func (s *Store) ReparentChannel(ctx context.Context, actor AccountID, channelID
 	return s.getChannel(ctx, channelID)
 }
 
-// requireChannelParticipant is the ReparentChannel gate: non-participant and
-// unknown channel both merge to ErrNotFound.
-func requireChannelParticipant(ctx context.Context, q *db.Queries, actor AccountID, channelID ChannelID) error {
-	participant, err := q.ChannelParticipant(ctx, db.ChannelParticipantParams{
-		ChannelID: string(channelID),
-		AccountID: string(actor),
-	})
-	if err != nil {
-		return fmt.Errorf("store: check channel participation: %w", err)
-	}
-	if !participant.Valid || !participant.Bool {
-		return fmt.Errorf("%w: channel %q", ErrNotFound, channelID)
-	}
-	return nil
-}
-
 // expandOwnerMembership computes the final member set for a new channel: the
 // requested members, plus the actor, plus the owning user of every agent in the
 // set — the transitive owner-membership invariant (design.md:231-234),