From 6c41eae0f0bd8ea0d80663721d4d29b1d41e6d54 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sun, 30 Aug 2026 13:31:41 -0400 Subject: [PATCH] feat(store): reserved DM group + upsert + create-guard + convert-on-add (RIG-2963) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Peer-DM T2 (design product/compass-agent-peer-dm §Plan T2, Decisions R3/R4). Store layer for agent peer-DMs, stacked on the T1 proto contract (#738). - New go/internal/store/dm.go: EnsureOwnerDMGroupTx (per-owner reserved __dm__ group, visibility-discriminated, mirrors EnsureOwnerCoordinationGroupTx); UpsertDMChannelTx (deterministic sorted-handle name, no suffix search, ON CONFLICT DO NOTHING + re-SELECT resume loop, born kind=DM + OPEN ownerless + mandatory, expand owner membership + seed cursors); verifyReconcileDMTx (R3 belt: wrong-kind squat -> ErrNotFound, mandatory + missing-member drift reconciled in-tx); LockOwnerDMTx (distinct 'dm:' advisory-lock key domain); isReservedDMGroupTx (create-guard discriminator). - channels.go: R3 create-guard in CreateChannel (reject a manual create into a reserved DM group with merged ErrNotFound); R4 convert-on-add in UpdateChannelMembers (a genuine member ADD on kind=DM requires ConvertChannelName, flips kind=CHANNEL + rename + group_id=NULL freeing the DM name, before the add) + two-party floor (a remove may not strand a DM below two agent parties). FOR UPDATE read now also returns kind. - inputs.go: MemberUpdatesOptions{ConvertChannelName}; UpdateChannelMembers signature widened; comms.go threads req.Msg.GetConvertChannelName(). - dm_pgtest_test.go: 9 pgtest cases (create invariants, idempotent resume either direction, concurrent-open race -> one channel, delivery-target predicate, manual create into reserved group -> not_found, squat belt, convert with/without name, fresh pair after convert, remove-below-two rejected). Co-authored-by: Matt Wilkinson --- .../comms/agent_conversation_pgtest_test.go | 2 +- go/internal/comms/comms.go | 1 + go/internal/comms/pinned_board_pgtest_test.go | 2 +- .../pre_settle_closure_pgtest_test.go | 2 +- go/internal/store/authz_test.go | 4 +- .../store/channel_policy_pgtest_test.go | 8 +- go/internal/store/channels.go | 202 ++++++- go/internal/store/channels_test.go | 38 +- go/internal/store/delivery_cursors_test.go | 8 +- go/internal/store/dm.go | 266 +++++++++ go/internal/store/dm_pgtest_test.go | 523 ++++++++++++++++++ go/internal/store/inputs.go | 11 + .../store/messages_answer_post_test.go | 2 +- .../store/messages_authored_update_test.go | 2 +- go/server/offline_mention_e2e_pgtest_test.go | 2 +- 15 files changed, 1017 insertions(+), 56 deletions(-) create mode 100644 go/internal/store/dm.go create mode 100644 go/internal/store/dm_pgtest_test.go diff --git a/go/internal/comms/agent_conversation_pgtest_test.go b/go/internal/comms/agent_conversation_pgtest_test.go index 39d3f8cc..2eaeac4d 100644 --- a/go/internal/comms/agent_conversation_pgtest_test.go +++ b/go/internal/comms/agent_conversation_pgtest_test.go @@ -280,7 +280,7 @@ func TestCommitAgentUpdateRevokedMemberIsNotFound(t *testing.T) { t.Fatalf("the agent cannot edit while still a member: %v", err) } - if _, _, err := st.UpdateChannelMembers(ctx, owner.ID, ch.ID, []store.MemberUpdate{{AccountID: agent.ID, Remove: true}}); err != nil { + if _, _, err := st.UpdateChannelMembers(ctx, owner.ID, ch.ID, []store.MemberUpdate{{AccountID: agent.ID, Remove: true}}, store.MemberUpdatesOptions{}); err != nil { t.Fatalf("UpdateChannelMembers(remove): %v", err) } diff --git a/go/internal/comms/comms.go b/go/internal/comms/comms.go index 29996940..02ad2dfc 100644 --- a/go/internal/comms/comms.go +++ b/go/internal/comms/comms.go @@ -266,6 +266,7 @@ func (c *Comms) UpdateChannelMembers( caller, store.ChannelID(req.Msg.GetChannelId()), updates, + store.MemberUpdatesOptions{ConvertChannelName: req.Msg.GetConvertChannelName()}, ) if err != nil { return nil, edgeError(err) diff --git a/go/internal/comms/pinned_board_pgtest_test.go b/go/internal/comms/pinned_board_pgtest_test.go index faef3770..4e348035 100644 --- a/go/internal/comms/pinned_board_pgtest_test.go +++ b/go/internal/comms/pinned_board_pgtest_test.go @@ -318,7 +318,7 @@ func TestUpdatePinnedBoardMembershipRevokedMidFlightIsNotFound(t *testing.T) { // `other` is a member first — a pin would succeed. Now revoke and commit. if _, _, err := st.UpdateChannelMembers(ctx, owner.ID, ch.ID, []store.MemberUpdate{ {AccountID: other.ID, Remove: true}, - }); err != nil { + }, store.MemberUpdatesOptions{}); err != nil { t.Fatalf("UpdateChannelMembers(remove other): %v", err) } diff --git a/go/internal/delivery/pre_settle_closure_pgtest_test.go b/go/internal/delivery/pre_settle_closure_pgtest_test.go index 6668dea1..1320f776 100644 --- a/go/internal/delivery/pre_settle_closure_pgtest_test.go +++ b/go/internal/delivery/pre_settle_closure_pgtest_test.go @@ -109,7 +109,7 @@ func mustRoomWithMembers(t *testing.T, ctx context.Context, s *store.Store, owne // member-update path (so it can post / receive live delivers). func subscribeMember(t *testing.T, ctx context.Context, s *store.Store, owner store.AccountID, ch store.ChannelID, agent store.AccountID) { t.Helper() - if _, _, err := s.UpdateChannelMembers(ctx, owner, ch, []store.MemberUpdate{{AccountID: agent, Subscribed: true}}); err != nil { + if _, _, err := s.UpdateChannelMembers(ctx, owner, ch, []store.MemberUpdate{{AccountID: agent, Subscribed: true}}, store.MemberUpdatesOptions{}); err != nil { t.Fatalf("UpdateChannelMembers(subscribe %s): %v", agent, err) } } diff --git a/go/internal/store/authz_test.go b/go/internal/store/authz_test.go index 6eec760e..60e76fd7 100644 --- a/go/internal/store/authz_test.go +++ b/go/internal/store/authz_test.go @@ -70,7 +70,7 @@ func TestUpdateChannelMembersNonMemberRefused(t *testing.T) { // Self-add escalation by a non-member is refused. _, _, err := s.UpdateChannelMembers(ctx, outsider.ID, ch.ID, []MemberUpdate{ {AccountID: outsider.ID}, - }) + }, MemberUpdatesOptions{}) sentinelIs(t, err, ErrNotFound, "non-member self-add") // The outsider gained no membership: the refusal did not write. @@ -85,7 +85,7 @@ func TestUpdateChannelMembersNonMemberRefused(t *testing.T) { // The member (owner) still mutates: adding a newcomer succeeds. updated, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: newcomer.ID}, - }) + }, MemberUpdatesOptions{}) if err != nil { t.Fatalf("member UpdateChannelMembers: %v", err) } diff --git a/go/internal/store/channel_policy_pgtest_test.go b/go/internal/store/channel_policy_pgtest_test.go index c3700fb2..0fab8b6f 100644 --- a/go/internal/store/channel_policy_pgtest_test.go +++ b/go/internal/store/channel_policy_pgtest_test.go @@ -89,7 +89,7 @@ func TestUpdateChannelMembersUnsubscribeRejectedOnMandatory(t *testing.T) { // An explicit unsubscribe of member is refused. _, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: member.ID, Unsubscribe: true}, - }) + }, MemberUpdatesOptions{}) sentinelIs(t, err, ErrInvalidArgument, "unsubscribe on mandatory channel") // A plain add (not an unsubscribe) still works: the guard is scoped to the @@ -97,7 +97,7 @@ func TestUpdateChannelMembersUnsubscribeRejectedOnMandatory(t *testing.T) { newMember := mustUser(t, s, "newcomer") if _, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: newMember.ID}, - }); err != nil { + }, MemberUpdatesOptions{}); err != nil { t.Fatalf("plain add on mandatory channel: %v", err) } } @@ -410,7 +410,7 @@ func TestUpdateChannelMembersSeedsUnsubscribedAddOnMandatory(t *testing.T) { // Plain add (subscribed defaults false) of the agent to the mandatory channel. if _, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: late.ID}, - }); err != nil { + }, MemberUpdatesOptions{}); err != nil { t.Fatalf("UpdateChannelMembers(plain add on mandatory): %v", err) } @@ -548,7 +548,7 @@ func TestUpdateChannelMembersConcurrentFlipSeedsLateMember(t *testing.T) { // under test. With the fix its FOR UPDATE read blocks on B's lock. done := make(chan error, 1) go func() { - _, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{{AccountID: late.ID}}) + _, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{{AccountID: late.ID}}, MemberUpdatesOptions{}) done <- err }() diff --git a/go/internal/store/channels.go b/go/internal/store/channels.go index 7df50099..09cef737 100644 --- a/go/internal/store/channels.go +++ b/go/internal/store/channels.go @@ -120,6 +120,22 @@ func (s *Store) CreateChannel(ctx context.Context, actor AccountID, c NewChannel } } + // R3 primary defense: the manual create path is server-forbidden from + // targeting a reserved per-owner DM group. Only the OpenDM path may write + // there (UpsertDMChannelTx), so rejecting a create here makes squatting a + // deterministic dm--… name impossible — no in-advance existence check + // needed. The rejection is the merged ErrNotFound (never confirms the group + // exists, so a stranger cannot probe the reserved namespace). + if c.GroupID != "" { + reserved, err := isReservedDMGroupTx(ctx, tx, c.GroupID) + if err != nil { + return Channel{}, err + } + if reserved { + return Channel{}, fmt.Errorf("%w: group %q", ErrNotFound, c.GroupID) + } + } + if _, err := tx.Exec(ctx, "INSERT INTO channels (id, name, group_id, kind, post_policy, owner_account_id, mandatory_subscription) "+ "VALUES ($1, $2, NULLIF($3, ''), $4, $5, NULLIF($6, ''), $7)", @@ -448,7 +464,7 @@ func (s *Store) ChannelByNameForViewer(ctx context.Context, viewer AccountID, na // D9 write-authz is enforced here in the store: the actor must be a member of // the channel to mutate it, so an unknown channel and a non-member both return // ErrNotFound (the not-found/forbidden merge). -func (s *Store) UpdateChannelMembers(ctx context.Context, actor AccountID, channelID ChannelID, updates []MemberUpdate) (Channel, []AccountID, error) { +func (s *Store) UpdateChannelMembers(ctx context.Context, actor AccountID, channelID ChannelID, updates []MemberUpdate, opts MemberUpdatesOptions) (Channel, []AccountID, error) { tx, err := s.pool.Begin(ctx) if err != nil { return Channel{}, nil, fmt.Errorf("store: begin update members: %w", err) @@ -465,29 +481,34 @@ func (s *Store) UpdateChannelMembers(ctx context.Context, actor AccountID, chann return Channel{}, nil, err } - // T4: read the channel's mandatory_subscription flag once under the tx. It - // serves two purposes: (1) an explicit unsubscribe on a mandatory channel is - // refused with InvalidArgument (membership implies a non-togglable - // subscription there); (2) a plain add to a mandatory channel must seed the - // new member's delivery cursor (below), because a mandatory channel makes - // every member a delivery target regardless of the subscribed flag. The read - // is FOR UPDATE so it serializes against a concurrent SetChannelPolicy - // mandatory flip, which takes the same channels-row lock before seeding all - // current members. Without the lock a member added concurrently with a flip - // can be dropped by both writers — B's seed-all runs before A inserts M, and - // A reads the stale mandatory=false and skips M — leaving M an unseeded - // member of a mandatory channel (the absent cursor coalesces to live head, - // so M is permanently caught-up and silently receives nothing). The lock - // guarantees the seed-presence invariant: whichever writer commits first is - // observed by the second, so M is seeded by exactly one of them, never zero. - // owner_account_id/policy fields are server-set and never mutated through - // this path — UpdateChannelMembers only ever touches membership rows. - var mandatory bool + // T4/R4: read the channel's mandatory_subscription flag AND kind once under + // the tx. mandatory serves two purposes: (1) an explicit unsubscribe on a + // mandatory channel is refused; (2) a plain add to a mandatory channel must + // seed the new member's delivery cursor (else it mints an un-seeded delivery + // target, the fail-DANGEROUS D2 hazard). The read is FOR UPDATE so it + // serializes against a concurrent SetChannelPolicy mandatory flip (same + // channels-row lock), guaranteeing a member added concurrently with a flip is + // seeded by exactly one writer, never zero. kind drives the R4 DM guards: a + // genuine member ADD on a kind=DM channel is a conversion, and a remove may + // not strand a DM below two agent parties. policy/owner fields are server-set + // and never mutated through this path. + var ( + mandatory bool + kindRaw int32 + ) if err := tx.QueryRow(ctx, - "SELECT mandatory_subscription FROM channels WHERE id = $1 FOR UPDATE", string(channelID), - ).Scan(&mandatory); err != nil { + "SELECT mandatory_subscription, kind FROM channels WHERE id = $1 FOR UPDATE", string(channelID), + ).Scan(&mandatory, &kindRaw); err != nil { return Channel{}, nil, fmt.Errorf("store: read channel mandatory flag: %w", err) } + kind := ChannelKind(kindRaw) + // The unsubscribe guard reads the PRE-convert mandatory state: a DM is + // born-mandatory, so an unsubscribe batched with a genuine convert-add is + // rejected here even though the post-convert channel is non-mandatory and + // would permit it. This mid-batch ambiguity is not reachable through the RPC + // surface (open_dm and member edits are distinct calls) and convert+unsubscribe + // is not a real use case; evaluating against the pre-convert state is the + // conservative choice. for _, u := range updates { if u.Unsubscribe { if mandatory { @@ -497,6 +518,25 @@ func (s *Store) UpdateChannelMembers(ctx context.Context, actor AccountID, chann } } + // R4 convert-on-add: a genuine member ADD (not a remove, not an unsubscribe, + // naming an account not already a member) on a kind=DM channel converts the + // two-party DM into a named CHANNEL before the add path runs. maybeConvertDM + // returns the channel's kind after any conversion (unchanged when no genuine + // add, or the channel was never a DM) and whether it converted this call. A + // convert clears mandatory_subscription in the DB (the result is a normal + // opt-in channel), so the caller's `mandatory` local — read pre-convert as + // TRUE for a born-mandatory DM — MUST be refreshed to FALSE, or the add loop + // below would seed the opt-in third member's delivery cursor as if the channel + // were still mandatory (a spurious seed: an unsubscribed add on a normal + // channel owes no cursor until it subscribes — the D2 seed-at-subscribe rule). + kind, converted, err := maybeConvertDM(ctx, tx, channelID, kind, updates, opts) + if err != nil { + return Channel{}, nil, err + } + if converted { + mandatory = false + } + var removed []AccountID for _, u := range updates { if u.AccountID == "" { @@ -510,6 +550,11 @@ func (s *Store) UpdateChannelMembers(ctx context.Context, actor AccountID, chann if deleted { removed = append(removed, u.AccountID) } + // R4: a DM is a fixed two-party surface — a remove may not strand it + // below two agent parties (teardown, not member surgery, ends a DM). + if err := requireDMTwoParties(ctx, tx, channelID, kind); err != nil { + return Channel{}, nil, err + } continue } if err := addOrUpdateMember(ctx, tx, channelID, u, mandatory); err != nil { @@ -527,6 +572,121 @@ func (s *Store) UpdateChannelMembers(ctx context.Context, actor AccountID, chann return ch, removed, nil } +// maybeConvertDM applies the R4 DM-to-CHANNEL conversion when a genuine member +// ADD targets a kind=DM channel, and returns the channel's kind afterward. A +// genuine add on a DM MUST supply opts.ConvertChannelName (else +// ErrInvalidArgument); with it, one tx converts the two-party DM into a normal +// opt-in channel before the caller's add path runs: +// +// - kind=CHANNEL, name=ConvertChannelName, group_id=NULL — leaving the reserved +// DM group frees the deterministic dm--a--b name so a future open_dm mints a +// FRESH pair DM. +// - mandatory_subscription=FALSE — the result is a genuine normal channel +// (opt-in), not a force-subscribe surface (Matt's ruling: a converted DM is a +// normal channel; a normal channel is opt-in). Without this the converted +// channel would keep the DM's born-mandatory flag and NO member could ever +// unsubscribe (the unsubscribe guard rejects it on mandatory channels). +// - the two original DM parties are flipped subscribed=TRUE. A DM member carries +// subscribed=FALSE (it was a delivery target only via the mandatory flag we +// just cleared), so without this they would SILENTLY stop receiving the very +// conversation they were mid-way through — and nothing notifies an agent it +// lost delivery (there is no add/drop notification; delivery is the only +// signal). Flipping the incumbents subscribed keeps them in the conversation; +// the newly-added third member joins opt-in (subscribed per its own update), +// the normal-channel default. +// +// When the channel is not a DM, or the batch has no genuine add (a pure +// subscribe-flip/remove adds no party), the kind is returned unchanged and no +// name is required. +func maybeConvertDM(ctx context.Context, tx pgx.Tx, channelID ChannelID, kind ChannelKind, updates []MemberUpdate, opts MemberUpdatesOptions) (ChannelKind, bool, error) { + if kind != ChannelKindDM { + return kind, false, nil + } + genuineAdd, err := hasGenuineAdd(ctx, tx, channelID, updates) + if err != nil { + return kind, false, err + } + if !genuineAdd { + return kind, false, nil + } + if opts.ConvertChannelName == "" { + return kind, false, fmt.Errorf("%w: adding a third member converts a DM to a channel and requires a channel name", ErrInvalidArgument) + } + // The convert sets group_id=NULL, and the channel-name unique index is + // partial on group_id IS NOT NULL — an ungrouped channel is exempt, so this + // UPDATE cannot raise a (group_id, name) unique violation. Ungrouped channel + // names are deliberately not constrained (mirrors home-channel dup behavior). + if _, err := tx.Exec(ctx, + "UPDATE channels SET kind = $1, name = $2, group_id = NULL, mandatory_subscription = FALSE WHERE id = $3", + int32(ChannelKindChannel), opts.ConvertChannelName, string(channelID), + ); err != nil { + return kind, false, fmt.Errorf("store: convert dm channel: %w", err) + } + // Keep the two incumbent DM parties in the conversation: flip every current + // AGENT member subscribed (a human owner member is left as-is — subscription + // is an agent-delivery concept). They already have a seeded delivery cursor + // from the DM's born-mandatory create, so no seed is owed here. + if _, err := tx.Exec(ctx, + "UPDATE channel_members cm SET subscribed = TRUE "+ + "FROM agent_accounts aa WHERE aa.account_id = cm.account_id AND cm.channel_id = $1", + string(channelID), + ); err != nil { + return kind, false, fmt.Errorf("store: subscribe converted dm parties: %w", err) + } + return ChannelKindChannel, true, nil +} + +// hasGenuineAdd reports whether updates contain at least one genuine member ADD +// against channelID: an update that is not a remove and not an unsubscribe, +// naming an account not already a member. A subscribe-flip of an existing member +// (or a re-add of a current member) is NOT a genuine add — it adds no party — so +// it does not trigger the R4 DM conversion. The membership probe reads this tx's +// snapshot, so a member added earlier in the same batch counts as present. +func hasGenuineAdd(ctx context.Context, tx pgx.Tx, channelID ChannelID, updates []MemberUpdate) (bool, error) { + for _, u := range updates { + if u.Remove || u.Unsubscribe || u.AccountID == "" { + continue + } + var exists bool + if err := tx.QueryRow(ctx, + "SELECT EXISTS (SELECT 1 FROM channel_members WHERE channel_id = $1 AND account_id = $2)", + string(channelID), string(u.AccountID), + ).Scan(&exists); err != nil { + return false, fmt.Errorf("store: probe member presence: %w", err) + } + if !exists { + return true, nil + } + } + return false, nil +} + +// requireDMTwoParties enforces the R4 two-party floor after a remove: on a +// kind=DM channel it counts the AGENT members remaining in this tx's snapshot and +// rejects the remove (ErrInvalidArgument, rolling it back) if fewer than two +// remain — a one-party DM is not a thing; teardown, not member surgery, ends a +// DM. Human owner members are excluded from the count: the DM's parties are its +// two agents; their pulled-in owners are membership bookkeeping, not parties. A +// non-DM channel has no such floor and returns nil immediately. +func requireDMTwoParties(ctx context.Context, tx pgx.Tx, channelID ChannelID, kind ChannelKind) error { + if kind != ChannelKindDM { + return nil + } + var parties int + if err := tx.QueryRow(ctx, + "SELECT COUNT(*) FROM channel_members cm "+ + "JOIN agent_accounts aa ON aa.account_id = cm.account_id "+ + "WHERE cm.channel_id = $1", + string(channelID), + ).Scan(&parties); err != nil { + return fmt.Errorf("store: count agent members: %w", err) + } + if parties < 2 { + return fmt.Errorf("%w: a DM must keep two agent parties; convert or tear it down instead", ErrInvalidArgument) + } + return nil +} + // removeMember deletes one member row, preserving transitive owner-membership // (design.md:231-234) symmetrically with creation: a user must stay while any // agent it owns remains in the channel, so removing such an owner is rejected as diff --git a/go/internal/store/channels_test.go b/go/internal/store/channels_test.go index 51fa2e57..05931475 100644 --- a/go/internal/store/channels_test.go +++ b/go/internal/store/channels_test.go @@ -372,7 +372,7 @@ func TestUpdateChannelMembersMutations(t *testing.T) { // Add a member, subscribed. updated, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: newcomer.ID, Subscribed: true}, - }) + }, MemberUpdatesOptions{}) if err != nil { t.Fatalf("UpdateChannelMembers(add): %v", err) } @@ -386,7 +386,7 @@ func TestUpdateChannelMembersMutations(t *testing.T) { // Flip subscribed off. if _, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: newcomer.ID, Subscribed: false}, - }); err != nil { + }, MemberUpdatesOptions{}); err != nil { t.Fatalf("UpdateChannelMembers(unsubscribe): %v", err) } if memberSubscribed(t, s, ch.ID, newcomer.ID) { @@ -396,7 +396,7 @@ func TestUpdateChannelMembersMutations(t *testing.T) { // Remove the member. afterRemove, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: newcomer.ID, Remove: true}, - }) + }, MemberUpdatesOptions{}) if err != nil { t.Fatalf("UpdateChannelMembers(remove): %v", err) } @@ -424,7 +424,7 @@ func TestUpdateChannelMembersAddingAgentPullsOwner(t *testing.T) { // Adding the agent must also pull its owner into the channel. updated, _, err := s.UpdateChannelMembers(ctx, actor.ID, ch.ID, []MemberUpdate{ {AccountID: agent.ID}, - }) + }, MemberUpdatesOptions{}) if err != nil { t.Fatalf("UpdateChannelMembers(add agent): %v", err) } @@ -449,7 +449,7 @@ func TestUpdateChannelMembersPreservesOwnerSubscription(t *testing.T) { } if _, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: owner.ID, Subscribed: true}, - }); err != nil { + }, MemberUpdatesOptions{}); err != nil { t.Fatalf("UpdateChannelMembers(subscribe owner): %v", err) } if !memberSubscribed(t, s, ch.ID, owner.ID) { @@ -463,7 +463,7 @@ func TestUpdateChannelMembersPreservesOwnerSubscription(t *testing.T) { // clobbered the already-subscribed owner to FALSE on the agent join. updated, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: agent.ID}, - }) + }, MemberUpdatesOptions{}) if err != nil { t.Fatalf("UpdateChannelMembers(add agent): %v", err) } @@ -484,7 +484,7 @@ func TestUpdateChannelMembersUnknownChannelNotFound(t *testing.T) { actor := mustUser(t, s, "actor") _, _, err := s.UpdateChannelMembers(ctx, actor.ID, ChannelID("ghost"), []MemberUpdate{ {AccountID: actor.ID}, - }) + }, MemberUpdatesOptions{}) sentinelIs(t, err, ErrNotFound, "unknown channel") } @@ -541,7 +541,7 @@ func TestUpdateChannelMembersRejectsOwnerRemovalWithAgentPresent(t *testing.T) { } if _, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: agent.ID}, - }); err != nil { + }, MemberUpdatesOptions{}); err != nil { t.Fatalf("UpdateChannelMembers(add agent): %v", err) } @@ -551,7 +551,7 @@ func TestUpdateChannelMembersRejectsOwnerRemovalWithAgentPresent(t *testing.T) { // a user can always read anything its agent is party to. _, _, err = s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: owner.ID, Remove: true}, - }) + }, MemberUpdatesOptions{}) sentinelIs(t, err, ErrInvalidArgument, "remove owner while its agent remains") // Positive companion, proving the invariant is about the DEPENDENT agent, @@ -559,7 +559,7 @@ func TestUpdateChannelMembersRejectsOwnerRemovalWithAgentPresent(t *testing.T) { // it, so this succeeds)... afterAgent, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: agent.ID, Remove: true}, - }) + }, MemberUpdatesOptions{}) if err != nil { t.Fatalf("UpdateChannelMembers(remove agent): %v", err) } @@ -571,7 +571,7 @@ func TestUpdateChannelMembersRejectsOwnerRemovalWithAgentPresent(t *testing.T) { // the channel to depend on its membership. afterOwner, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: owner.ID, Remove: true}, - }) + }, MemberUpdatesOptions{}) if err != nil { t.Fatalf("UpdateChannelMembers(remove owner after agent gone): %v", err) } @@ -608,7 +608,7 @@ func TestUpdateChannelMembersRemovalScansAllOwnedAgents(t *testing.T) { {AccountID: agent1.ID}, {AccountID: agent2.ID}, {AccountID: otherAgent.ID}, - }) + }, MemberUpdatesOptions{}) if err != nil { t.Fatalf("UpdateChannelMembers(add agents): %v", err) } @@ -621,7 +621,7 @@ func TestUpdateChannelMembersRemovalScansAllOwnedAgents(t *testing.T) { // (a) Both owned agents remain: removing the owner is rejected. _, _, err = s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: owner.ID, Remove: true}, - }) + }, MemberUpdatesOptions{}) sentinelIs(t, err, ErrInvalidArgument, "remove owner while both its agents remain") // (b) Remove the FIRST-added owned agent. If the EXISTS scanned only the @@ -629,7 +629,7 @@ func TestUpdateChannelMembersRemovalScansAllOwnedAgents(t *testing.T) { // wrongly succeed. agent2 still remains, so removal must STILL be rejected. afterAgent1, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: agent1.ID, Remove: true}, - }) + }, MemberUpdatesOptions{}) if err != nil { t.Fatalf("UpdateChannelMembers(remove agent1): %v", err) } @@ -641,7 +641,7 @@ func TestUpdateChannelMembersRemovalScansAllOwnedAgents(t *testing.T) { } _, _, err = s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: owner.ID, Remove: true}, - }) + }, MemberUpdatesOptions{}) sentinelIs(t, err, ErrInvalidArgument, "remove owner while its second agent remains") // (c) Remove the second owned agent. No agent the owner owns remains, so @@ -649,7 +649,7 @@ func TestUpdateChannelMembersRemovalScansAllOwnedAgents(t *testing.T) { // a channel member, confirming the gate is scoped to owner_user_id=$removed. afterAgent2, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: agent2.ID, Remove: true}, - }) + }, MemberUpdatesOptions{}) if err != nil { t.Fatalf("UpdateChannelMembers(remove agent2): %v", err) } @@ -661,7 +661,7 @@ func TestUpdateChannelMembersRemovalScansAllOwnedAgents(t *testing.T) { } afterOwner, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: owner.ID, Remove: true}, - }) + }, MemberUpdatesOptions{}) if err != nil { t.Fatalf("UpdateChannelMembers(remove owner after all its agents gone): %v", err) } @@ -728,7 +728,7 @@ func TestSubscriberAccountIDsToggle(t *testing.T) { // Add a member subscribed: it appears in both member and subscriber sets. subbed, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: newcomer.ID, Subscribed: true}, - }) + }, MemberUpdatesOptions{}) if err != nil { t.Fatalf("UpdateChannelMembers(add subscribed): %v", err) } @@ -743,7 +743,7 @@ func TestSubscriberAccountIDsToggle(t *testing.T) { // subscriber set — the join/subscribe tier split. unsubbed, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{ {AccountID: newcomer.ID, Subscribed: false}, - }) + }, MemberUpdatesOptions{}) if err != nil { t.Fatalf("UpdateChannelMembers(unsubscribe): %v", err) } diff --git a/go/internal/store/delivery_cursors_test.go b/go/internal/store/delivery_cursors_test.go index c8def065..2150c5b9 100644 --- a/go/internal/store/delivery_cursors_test.go +++ b/go/internal/store/delivery_cursors_test.go @@ -68,7 +68,7 @@ func subscribeAgent(t *testing.T, s *Store, actor AccountID, ch ChannelID, agent t.Helper() if _, _, err := s.UpdateChannelMembers(context.Background(), actor, ch, []MemberUpdate{ {AccountID: agent, Subscribed: true}, - }); err != nil { + }, MemberUpdatesOptions{}); err != nil { t.Fatalf("UpdateChannelMembers(subscribe agent): %v", err) } } @@ -283,7 +283,7 @@ func TestSeedDeliveryCursorRidesMemberInsert(t *testing.T) { newcomer := mustUser(t, s, "newcomer") if _, _, err := s.UpdateChannelMembers(context.Background(), owner.ID, other.ID, []MemberUpdate{ {AccountID: newcomer.ID, Subscribed: true}, - }); err != nil { + }, MemberUpdatesOptions{}); err != nil { t.Fatalf("UpdateChannelMembers(subscribe user): %v", err) } if _, _, ok := readCursor(t, s, newcomer.ID, other.ID); ok { @@ -443,7 +443,7 @@ func TestUndeliveredMessagesHomeChannelSweepsWhenUnsubscribed(t *testing.T) { // UPDATE SET subscribed). The owner is a home-channel member and may mutate. if _, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch, []MemberUpdate{ {AccountID: agent.ID, Subscribed: false}, - }); err != nil { + }, MemberUpdatesOptions{}); err != nil { t.Fatalf("UpdateChannelMembers(unsubscribe agent home): %v", err) } if memberSubscribed(t, s, ch, agent.ID) { @@ -625,7 +625,7 @@ func TestReSubscribeDoesNotResetCursor(t *testing.T) { // Unsubscribe, post M2 during the unsubscribed window, then re-subscribe. if _, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch, []MemberUpdate{ {AccountID: agent.ID, Subscribed: false}, - }); err != nil { + }, MemberUpdatesOptions{}); err != nil { t.Fatalf("UpdateChannelMembers(unsubscribe): %v", err) } m2, _ := postAs(t, s, ch, owner.ID, "owed M2 during unsub") diff --git a/go/internal/store/dm.go b/go/internal/store/dm.go new file mode 100644 index 00000000..446dbae3 --- /dev/null +++ b/go/internal/store/dm.go @@ -0,0 +1,266 @@ +package store + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5" +) + +// dmGroupName is the fixed reserved name of the per-owner DM group — the +// namespace every one of an owner's peer-DM channels lives in (design R3). It is +// distinct from the coordination group's __coordination__ so the two reserved +// namespaces never collide, and it is the discriminator (paired with +// VisibilityOwner) both the CreateChannel create-guard and EnsureOwnerDMGroupTx +// key on. +const dmGroupName = "__dm__" + +// DMChannelSpec is the resolved identity + membership for one peer-DM channel, +// computed by the OpenDM path (which owns the deterministic sorted-handle NAME +// and the two agent parties) and handed to UpsertDMChannelTx (which owns the +// SQL). The store never encodes the name derivation or the DM policy — it only +// persists what the OpenDM edge decides. Members are the two agent parties, +// pre-resolved to account ids; UpsertDMChannelTx pulls in each party's owning +// user(s) itself (transitive owner-membership). +type DMChannelSpec struct { + GroupID ChannelGroupID + Name string + Members []AccountID +} + +// EnsureOwnerDMGroupTx get-or-creates the single per-owner namespace group that +// hosts every one of ownerUserID's peer-DM channels, on tx, and returns its id. +// It mirrors EnsureOwnerCoordinationGroupTx exactly — deterministic + idempotent +// on owner_user_id + the fixed reserved name, VisibilityOwner (DMs are +// owner-private, never lattice-shared), un-parented — differing only in the +// reserved name (__dm__ vs __coordination__) so the two reserved namespaces stay +// disjoint. The get-half is VISIBILITY-DISCRIMINATED (AND visibility = $3, bound +// to VisibilityOwner): CreateChannelGroup has no reserved-name guard, so a user +// CAN plant a top-level group named __dm__ at any visibility; a wider +// (VisibilityShared) planted group must NEVER be adopted (it would host +// owner-private DMs in a shared group — a cross-tenant leak), so the +// discriminator excludes it and the create-half INSERTs the correct +// owner-visible group. The caller holds the per-owner DM advisory lock +// (LockOwnerDMTx), so the SELECT-then-INSERT cannot race a concurrent first-open +// for the same owner into two groups. +func (s *Store) EnsureOwnerDMGroupTx(ctx context.Context, tx pgx.Tx, ownerUserID AccountID) (ChannelGroupID, error) { + if ownerUserID == "" { + return "", fmt.Errorf("%w: owner user id is required", ErrInvalidArgument) + } + + var existing string + switch err := tx.QueryRow(ctx, + `SELECT id FROM channel_groups WHERE owner_user_id = $1 AND name = $2 AND parent_group_id IS NULL AND visibility = $3`, + string(ownerUserID), dmGroupName, int32(VisibilityOwner), + ).Scan(&existing); { + case err == nil: + return ChannelGroupID(existing), nil + case !noRows(err): + return "", fmt.Errorf("store: resolve dm group: %w", err) + } + + id := newID() + if _, err := tx.Exec(ctx, + `INSERT INTO channel_groups (id, name, parent_group_id, owner_user_id, visibility) VALUES ($1, $2, NULL, $3, $4)`, + id, dmGroupName, string(ownerUserID), int32(VisibilityOwner), + ); err != nil { + return "", fmt.Errorf("store: insert dm group: %w", err) + } + return ChannelGroupID(id), nil +} + +// UpsertDMChannelTx get-or-creates spec's peer-DM channel on tx and returns its +// id plus whether it was created this call. Unlike UpsertCoordinationChannelTx +// there is NO suffix search: a DM name is the deterministic sorted-handle pair +// key, so a collision on (group, name) is EITHER the caller's own prior open (a +// RESUME) or a squat the R3 belt rejects — never a name to route around. +// +// Resolution loop (the caller holds the per-owner DM advisory lock, so no +// concurrent OPEN for the same owner interleaves — but the loop still tolerates a +// concurrent writer for defense-in-depth): +// - A channel exists at (group, name): RESUME. Run the R3 belt verify-reconcile +// (kind=DM ∧ mandatory ∧ {members} ⊆ actual, reconciling recoverable drift; +// a wrong kind → ErrNotFound, never adopted) and return (id, false, nil). +// - No channel: INSERT it born kind=DM, zero-value policy (OPEN, ownerless) + +// mandatory_subscription=true, poison-free via ON CONFLICT (group_id, name) +// DO NOTHING on the partial unique index. On a returned row → created path: +// insert the expanded member set (both parties + their owners) and seed every +// agent member's delivery cursor in this tx (the born-mandatory discipline), +// return (id, true, nil). On no row (a concurrent open won the race) → loop +// back to re-SELECT, which now resumes the committed row. +func (s *Store) UpsertDMChannelTx(ctx context.Context, tx pgx.Tx, spec DMChannelSpec) (ChannelID, bool, error) { + if spec.GroupID == "" { + return "", false, fmt.Errorf("%w: dm channel group is required", ErrInvalidArgument) + } + if spec.Name == "" { + return "", false, fmt.Errorf("%w: dm channel name is required", ErrInvalidArgument) + } + if len(spec.Members) < 2 { + return "", false, fmt.Errorf("%w: dm channel requires two agent parties", ErrInvalidArgument) + } + + // The final member set: both agent parties plus each party's owning user(s) + // — the transitive owner-membership invariant (design.md:231-234), computed + // identically on the create and resume paths so a reconcile converges on + // exactly what a create would have produced. One party is the actor, the + // rest are the requested set. + members, err := expandOwnerMembership(ctx, tx, spec.Members[0], spec.Members[1:]) + if err != nil { + return "", false, err + } + + for { + var ( + existingID string + existingKind int32 + ) + switch err := tx.QueryRow(ctx, + `SELECT id, kind FROM channels WHERE group_id = $1 AND name = $2`, + string(spec.GroupID), spec.Name, + ).Scan(&existingID, &existingKind); { + case err == nil: + // Resume: the R3 belt verifies + reconciles the resolved row before + // adopting it, and returns ErrNotFound on a wrong-kind squat. + if err := verifyReconcileDMTx(ctx, tx, ChannelID(existingID), ChannelKind(existingKind), members); err != nil { + return "", false, err + } + return ChannelID(existingID), false, nil + case !noRows(err): + return "", false, fmt.Errorf("store: resolve dm channel: %w", err) + } + + // Free name (as of our SELECT): INSERT born kind=DM, zero-value policy + // (OPEN, no owner) + mandatory. ON CONFLICT DO NOTHING absorbs a row a + // concurrent open committed between our SELECT and this INSERT — zero + // rows returned rather than a raised unique-violation — so the tx is + // never poisoned; we loop and resume the committed row. + id := newID() + switch err := tx.QueryRow(ctx, + `INSERT INTO channels (id, name, group_id, kind, post_policy, owner_account_id, mandatory_subscription) `+ + `VALUES ($1, $2, $3, $4, $5, NULL, $6) `+ + `ON CONFLICT (group_id, name) WHERE group_id IS NOT NULL DO NOTHING `+ + `RETURNING id`, + id, spec.Name, string(spec.GroupID), int32(ChannelKindDM), + int32(ChannelPostPolicyOpen), true, + ).Scan(&id); { + case err == nil: + for _, m := range members { + if _, err := tx.Exec(ctx, + `INSERT INTO channel_members (channel_id, account_id, subscribed) VALUES ($1, $2, FALSE) `+ + `ON CONFLICT (channel_id, account_id) DO NOTHING`, + id, string(m), + ); err != nil { + return "", false, upsertMemberErr(err, m) + } + } + // Born mandatory ⇒ every member is a delivery target regardless of + // the subscribed flag (the D1 disjunct), so each agent member's + // delivery cursor MUST be seeded in this same tx — an un-seeded + // delivery target is the fail-DANGEROUS D2 hazard. Self-guarding + // (agent-only) and idempotent, so human members are a no-op. + if err := seedChannelDeliveryCursors(ctx, tx, ChannelID(id)); err != nil { + return "", false, err + } + return ChannelID(id), true, nil + case noRows(err): + // A concurrent open won the (group, name) race. Re-SELECT resolves it + // as a resume on the next iteration. + continue + default: + return "", false, fmt.Errorf("store: insert dm channel: %w", err) + } + } +} + +// LockOwnerDMTx takes the per-owner advisory lock that serializes every peer-DM +// open under one owner's DM namespace, on tx (auto-released at tx end). It +// mirrors LockOwnerCoordinationTx with a DISTINCT key domain ('dm:' vs +// 'coordination:') so DM opens never serialize behind coordination reconciles +// (or the per-owner-tree reparent lock). hashtext widens the text key to the int +// the advisory lock takes; a hash collision across two owners is a benign +// redundant wait, never a wrong result. +func LockOwnerDMTx(ctx context.Context, tx pgx.Tx, ownerUserID AccountID) error { + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext('dm:' || $1))`, string(ownerUserID)); err != nil { + return fmt.Errorf("store: lock owner dm: %w", err) + } + return nil +} + +// verifyReconcileDMTx is the R3 belt (design.md:355-361, T2:708-710): on a resume +// it asserts the resolved channel is a real DM and reconciles recoverable drift +// in-tx, or refuses to adopt a wrong-kind row. +// +// - A wrong kind (a hand-planted squat, or drift) → ErrNotFound. The +// CreateChannel create-guard makes this unreachable via the manual path, but +// the belt defends any other write into the reserved group rather than +// resolving every future open to a hostile row. +// - mandatory drift (a DM found non-mandatory) is recoverable: re-assert it, +// then seed every current agent member's cursor (the SetChannelPolicy +// flip discipline — a FALSE→TRUE re-assert makes every member a delivery +// target, so none may be left un-seeded, the fail-DANGEROUS D2 hazard). +// - a missing wanted member is recoverable: re-add its row, then seed cursors +// again so the re-added member is a seeded delivery target. Both seeds are +// self-guarding (agent-only) and idempotent — a present member and an +// already-seeded cursor are no-ops. +func verifyReconcileDMTx(ctx context.Context, tx pgx.Tx, channelID ChannelID, kind ChannelKind, wanted []AccountID) error { + if kind != ChannelKindDM { + return fmt.Errorf("%w: dm channel %q", ErrNotFound, channelID) + } + if _, err := tx.Exec(ctx, + `UPDATE channels SET mandatory_subscription = TRUE WHERE id = $1 AND mandatory_subscription = FALSE`, + string(channelID), + ); err != nil { + return fmt.Errorf("store: reassert dm mandatory: %w", err) + } + // Seed EVERY current agent member's delivery cursor (not only re-added ones), + // matching SetChannelPolicy's mandatory-flip discipline: a FALSE→TRUE + // re-assert makes every member a delivery target, so a pre-existing member + // that somehow lacked a cursor must not be left an un-seeded delivery target + // (the fail-DANGEROUS D2 hazard). Self-guarding (agent-only) and idempotent, + // so an already-seeded member is a no-op. + if err := seedChannelDeliveryCursors(ctx, tx, channelID); err != nil { + return err + } + for _, m := range wanted { + if _, err := tx.Exec(ctx, + `INSERT INTO channel_members (channel_id, account_id, subscribed) VALUES ($1, $2, FALSE) `+ + `ON CONFLICT (channel_id, account_id) DO NOTHING`, + string(channelID), string(m), + ); err != nil { + return upsertMemberErr(err, m) + } + } + // Re-added members get their cursor from the seed-all above only if they were + // present at that statement; a member added just now by the loop needs its own + // seed, so seed once more after the adds (idempotent). + if err := seedChannelDeliveryCursors(ctx, tx, channelID); err != nil { + return err + } + return nil +} + +// isReservedDMGroupTx reports whether groupID is a reserved per-owner DM group — +// the discriminator the CreateChannel create-guard (R3 primary defense) keys on: +// the fixed reserved name AND VisibilityOwner (symmetric with the coordination +// group's visibility-discriminated get-half). A group id that names no group is +// not reserved (false, nil) — the caller's own not-found handling covers an +// unknown group. Because only the OpenDM path writes into a group this predicate +// matches, guarding CreateChannel against it makes squatting a dm--… name +// impossible with no in-advance existence check. +func isReservedDMGroupTx(ctx context.Context, tx pgx.Tx, groupID ChannelGroupID) (bool, error) { + var ( + name string + vis int32 + ) + switch err := tx.QueryRow(ctx, + `SELECT name, visibility FROM channel_groups WHERE id = $1`, + string(groupID), + ).Scan(&name, &vis); { + case err == nil: + return name == dmGroupName && ChannelGroupVisibility(vis) == VisibilityOwner, nil + case noRows(err): + return false, nil + default: + return false, fmt.Errorf("store: resolve dm group discriminator: %w", err) + } +} diff --git a/go/internal/store/dm_pgtest_test.go b/go/internal/store/dm_pgtest_test.go new file mode 100644 index 00000000..f370fa9a --- /dev/null +++ b/go/internal/store/dm_pgtest_test.go @@ -0,0 +1,523 @@ +//go:build pgtest + +package store + +// Peer-DM store contracts (RIG-2963 T2, design.md:691-741, Decisions R3/R4): +// the reserved per-owner DM group, the deterministic-name upsert (create / +// idempotent resume / concurrent-open race), the R3 create-guard + squat belt, +// and the R4 convert-on-add / two-party floor. These are properties only a real +// Postgres proves (the partial unique index, the in-tx cursor seeds, the +// delivery predicate, a concurrent race), so the file is pgtest-tagged. + +import ( + "context" + "slices" + "sync" + "testing" + + "github.com/jackc/pgx/v5" +) + +// openDM drives one peer-DM open the way the T3 OpenDM path will: under the +// per-owner DM advisory lock, ensure the owner's reserved group, then upsert the +// deterministic-name channel for the two agent parties. It runs the whole open in +// one WithTx so the lock, group, channel, members, and cursor seeds commit +// atomically — the same shape the real edge uses. Returns the channel id and +// whether it was created this call. +func openDM(t *testing.T, s *Store, ownerUserID AccountID, name string, members []AccountID) (ChannelID, bool) { + t.Helper() + var ( + id ChannelID + created bool + ) + if err := s.WithTx(context.Background(), func(tx pgx.Tx) error { + if err := LockOwnerDMTx(context.Background(), tx, ownerUserID); err != nil { + return err + } + gid, err := s.EnsureOwnerDMGroupTx(context.Background(), tx, ownerUserID) + if err != nil { + return err + } + id, created, err = s.UpsertDMChannelTx(context.Background(), tx, DMChannelSpec{ + GroupID: gid, Name: name, Members: members, + }) + return err + }); err != nil { + t.Fatalf("openDM(%q): %v", name, err) + } + return id, created +} + +// dmGroupIDFor resolves the owner's reserved DM group id (the group openDM +// ensured), for tests that plant a row directly into it. +func dmGroupIDFor(t *testing.T, s *Store, ownerUserID AccountID) ChannelGroupID { + t.Helper() + var gid string + if err := s.pool.QueryRow(context.Background(), + `SELECT id FROM channel_groups WHERE owner_user_id = $1 AND name = $2 AND parent_group_id IS NULL AND visibility = $3`, + string(ownerUserID), dmGroupName, int32(VisibilityOwner), + ).Scan(&gid); err != nil { + t.Fatalf("resolve dm group for %s: %v", ownerUserID, err) + } + return ChannelGroupID(gid) +} + +// cursorExists reports whether an agent has a seeded delivery cursor on channel. +func cursorExists(t *testing.T, s *Store, agent AccountID, channel ChannelID) bool { + t.Helper() + var exists bool + if err := s.pool.QueryRow(context.Background(), + `SELECT EXISTS(SELECT 1 FROM agent_delivery_cursors WHERE agent_account_id = $1 AND channel_id = $2)`, + string(agent), string(channel), + ).Scan(&exists); err != nil { + t.Fatalf("cursor probe (%s,%s): %v", agent, channel, err) + } + return exists +} + +// TestUpsertDMChannelCreateInvariants pins the created DM's exact shape: kind=DM, +// zero/OPEN ownerless policy, mandatory_subscription=true, members = both agent +// parties + the pulled-in owner, and a seeded delivery cursor for each agent. +func TestUpsertDMChannelCreateInvariants(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "owner") + a := mustAgent(t, s, owner.ID, "alice") + b := mustAgent(t, s, owner.ID, "bob") + + id, created := openDM(t, s, owner.ID, "dm--alice--bob", []AccountID{a.ID, b.ID}) + if !created { + t.Fatal("first open of a fresh pair reported created=false, want true") + } + + ch, err := s.GetChannel(context.Background(), id) + if err != nil { + t.Fatalf("GetChannel: %v", err) + } + if ch.Kind != ChannelKindDM { + t.Fatalf("kind = %d, want ChannelKindDM", ch.Kind) + } + if ch.Policy.PostPolicy != ChannelPostPolicyOpen || ch.Policy.OwnerAccountID != "" { + t.Fatalf("policy = %+v, want OPEN + ownerless (zero policy)", ch.Policy) + } + if !ch.Policy.MandatorySubscription { + t.Fatal("mandatory_subscription = false, want true (DM is born mandatory)") + } + got := memberSet(ch) + for _, want := range []AccountID{a.ID, b.ID, owner.ID} { + if !got[want] { + t.Fatalf("members %v missing %s", ch.MemberAccountIDs, want) + } + } + if len(ch.MemberAccountIDs) != 3 { + t.Fatalf("members = %v, want exactly both parties + owner", ch.MemberAccountIDs) + } + for _, agent := range []AccountID{a.ID, b.ID} { + if !cursorExists(t, s, agent, id) { + t.Fatalf("agent party %s has no seeded delivery cursor on the DM", agent) + } + } +} + +// TestUpsertDMChannelResumeIsIdempotent pins that a second open of the same pair +// (in EITHER member order — the name is the direction-independent sorted-handle +// key, so the store takes a pre-sorted name and either member order resolves the +// same channel) resumes the SAME id with created=false, minting no second row. +func TestUpsertDMChannelResumeIsIdempotent(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "owner") + a := mustAgent(t, s, owner.ID, "alice") + b := mustAgent(t, s, owner.ID, "bob") + + first, created := openDM(t, s, owner.ID, "dm--alice--bob", []AccountID{a.ID, b.ID}) + if !created { + t.Fatal("first open reported created=false, want true") + } + // Re-open with the members in the REVERSED order — same deterministic name. + second, created := openDM(t, s, owner.ID, "dm--alice--bob", []AccountID{b.ID, a.ID}) + if created { + t.Fatal("re-open reported created=true, want false (resume)") + } + if second != first { + t.Fatalf("re-open resolved a different channel: %s vs %s", second, first) + } + + var count int + if err := s.pool.QueryRow(context.Background(), + `SELECT COUNT(*) FROM channels c JOIN channel_groups g ON g.id = c.group_id + WHERE g.owner_user_id = $1 AND g.name = $2`, + string(owner.ID), dmGroupName, + ).Scan(&count); err != nil { + t.Fatalf("count dm channels: %v", err) + } + if count != 1 { + t.Fatalf("owner has %d DM channels, want exactly 1", count) + } +} + +// TestUpsertDMChannelConcurrentOpenRace pins that two concurrent opens of the +// same pair mint exactly one channel and both return the same id — the ON +// CONFLICT DO NOTHING + re-SELECT resume loop absorbs the race. Each open runs on +// its own tx (its own advisory-lock take), so the loser blocks on the winner's +// (group, name) tuple, then resumes it. +func TestUpsertDMChannelConcurrentOpenRace(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "owner") + a := mustAgent(t, s, owner.ID, "alice") + b := mustAgent(t, s, owner.ID, "bob") + + const name = "dm--alice--bob" + members := []AccountID{a.ID, b.ID} + + type result struct { + id ChannelID + err error + } + results := make([]result, 2) + var wg sync.WaitGroup + start := make(chan struct{}) + wg.Add(2) + for i := range results { + go func(i int) { + defer wg.Done() + <-start + var r result + r.err = s.WithTx(context.Background(), func(tx pgx.Tx) error { + if err := LockOwnerDMTx(context.Background(), tx, owner.ID); err != nil { + return err + } + gid, err := s.EnsureOwnerDMGroupTx(context.Background(), tx, owner.ID) + if err != nil { + return err + } + r.id, _, err = s.UpsertDMChannelTx(context.Background(), tx, DMChannelSpec{ + GroupID: gid, Name: name, Members: members, + }) + return err + }) + results[i] = r + }(i) + } + close(start) + wg.Wait() + + for i, r := range results { + if r.err != nil { + t.Fatalf("concurrent open %d errored: %v", i, r.err) + } + } + if results[0].id != results[1].id { + t.Fatalf("concurrent opens minted different channels: %s vs %s", results[0].id, results[1].id) + } + + var count int + if err := s.pool.QueryRow(context.Background(), + `SELECT COUNT(*) FROM channels c JOIN channel_groups g ON g.id = c.group_id + WHERE g.owner_user_id = $1 AND g.name = $2`, + string(owner.ID), dmGroupName, + ).Scan(&count); err != nil { + t.Fatalf("count dm channels: %v", err) + } + if count != 1 { + t.Fatalf("concurrent open produced %d channels, want exactly 1", count) + } +} + +// TestUpsertDMChannelIsDeliveryTarget pins that the created DM satisfies the D1 +// delivery predicate purely through its mandatory flag: an agent party who has +// NEVER subscribed (subscribed=false, the DM is not its home channel) is still a +// delivery target, because the predicate disjoins ch.mandatory_subscription. This +// is the same predicate SubscribedAgents / SweepChannels enforce, asserted here +// via SweepChannels (the channel enumeration) and SubscribedAgents (the recipient +// resolution). +func TestUpsertDMChannelIsDeliveryTarget(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "owner") + a := mustAgent(t, s, owner.ID, "alice") + b := mustAgent(t, s, owner.ID, "bob") + + id, _ := openDM(t, s, owner.ID, "dm--alice--bob", []AccountID{a.ID, b.ID}) + + // Neither party subscribed and the DM is not either party's home channel, so + // only the mandatory disjunct can put them in the sweep/deliver sets. + swept, err := s.SweepChannels(context.Background(), a.ID) + if err != nil { + t.Fatalf("SweepChannels(alice): %v", err) + } + if !slices.Contains(swept, id) { + t.Fatalf("DM %s not in alice's sweep set %v (mandatory disjunct failed)", id, swept) + } + + // When bob posts, alice is a delivery target of the DM via the same predicate. + recips, err := s.SubscribedAgents(context.Background(), id, b.ID) + if err != nil { + t.Fatalf("SubscribedAgents: %v", err) + } + if !containsAccount(recips, a.ID) { + t.Fatalf("alice %s not a delivery target of the DM (recipients %v)", a.ID, recips) + } +} + +// TestCreateChannelIntoReservedDMGroupIsNotFound pins the R3 primary defense: a +// manual CreateChannel targeting the reserved DM group is rejected with the +// merged ErrNotFound (never confirming the group exists), so a squat is +// impossible via the create path. +func TestCreateChannelIntoReservedDMGroupIsNotFound(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "owner") + a := mustAgent(t, s, owner.ID, "alice") + b := mustAgent(t, s, owner.ID, "bob") + + // Materialize the reserved DM group by opening a real DM first. + openDM(t, s, owner.ID, "dm--alice--bob", []AccountID{a.ID, b.ID}) + gid := dmGroupIDFor(t, s, owner.ID) + + _, err := s.CreateChannel(context.Background(), owner.ID, NewChannel{ + Name: "dm--alice--carol", GroupID: gid, Kind: ChannelKindChannel, + }) + sentinelIs(t, err, ErrNotFound, "manual create into the reserved DM group") +} + +// TestUpsertDMChannelSquatBeltRejectsWrongKind pins the R3 belt: if a wrong-kind +// row is planted at the deterministic name in the reserved group (impossible via +// the create-guard, but the belt defends any other write path), a resume returns +// ErrNotFound rather than adopting it. +func TestUpsertDMChannelSquatBeltRejectsWrongKind(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "owner") + a := mustAgent(t, s, owner.ID, "alice") + b := mustAgent(t, s, owner.ID, "bob") + + // Open a real DM to materialize the group, then hand-plant a squat under a + // DIFFERENT deterministic name (a raw INSERT bypasses the create-guard). + openDM(t, s, owner.ID, "dm--alice--bob", []AccountID{a.ID, b.ID}) + gid := dmGroupIDFor(t, s, owner.ID) + const squatName = "dm--alice--carol" + if _, err := s.pool.Exec(context.Background(), + `INSERT INTO channels (id, name, group_id, kind, post_policy, owner_account_id, mandatory_subscription) + VALUES ($1, $2, $3, $4, $5, NULL, $6)`, + newID(), squatName, string(gid), int32(ChannelKindChannel), int32(ChannelPostPolicyOpen), true, + ); err != nil { + t.Fatalf("plant squat row: %v", err) + } + + // A resume that resolves the squat name must refuse the wrong-kind adoptee. + err := s.WithTx(context.Background(), func(tx pgx.Tx) error { + if err := LockOwnerDMTx(context.Background(), tx, owner.ID); err != nil { + return err + } + _, _, e := s.UpsertDMChannelTx(context.Background(), tx, DMChannelSpec{ + GroupID: gid, Name: squatName, Members: []AccountID{a.ID, b.ID}, + }) + return e + }) + sentinelIs(t, err, ErrNotFound, "resume onto a wrong-kind squat row") +} + +// TestUpsertDMChannelResumeReconcilesDrift pins the R3 belt's RECOVERABLE-drift +// path (the half TestUpsertDMChannelSquatBeltRejectsWrongKind does not cover): a +// resume onto a real DM that has drifted — a wanted member row deleted AND the +// mandatory flag flipped to FALSE — restores the member row, re-seeds its +// delivery cursor, and re-asserts mandatory, all in the resume tx. This is the +// D2-hazard-adjacent path: a regression that skipped the seed-on-readd would +// silently mint an un-seeded delivery target on a resumed DM. +func TestUpsertDMChannelResumeReconcilesDrift(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "owner") + a := mustAgent(t, s, owner.ID, "alice") + b := mustAgent(t, s, owner.ID, "bob") + + const name = "dm--alice--bob" + id, created := openDM(t, s, owner.ID, name, []AccountID{a.ID, b.ID}) + if !created { + t.Fatal("first open reported created=false, want true") + } + + // Simulate recoverable drift directly in the DB (bypassing the API, which has + // no path to produce it): drop agent b's member row + its cursor, and flip the + // channel non-mandatory. + ctx := context.Background() + if _, err := s.pool.Exec(ctx, + `DELETE FROM channel_members WHERE channel_id = $1 AND account_id = $2`, + string(id), string(b.ID), + ); err != nil { + t.Fatalf("drop member row: %v", err) + } + if _, err := s.pool.Exec(ctx, + `DELETE FROM agent_delivery_cursors WHERE agent_account_id = $1 AND channel_id = $2`, + string(b.ID), string(id), + ); err != nil { + t.Fatalf("drop cursor: %v", err) + } + if _, err := s.pool.Exec(ctx, + `UPDATE channels SET mandatory_subscription = FALSE WHERE id = $1`, string(id), + ); err != nil { + t.Fatalf("flip non-mandatory: %v", err) + } + + // Re-open the same pair: a resume that runs the belt reconcile. + gotID, created := openDM(t, s, owner.ID, name, []AccountID{a.ID, b.ID}) + if created { + t.Fatal("resume onto a drifted DM reported created=true, want false") + } + if gotID != id { + t.Fatalf("resume minted a new channel %s, want the drifted %s", gotID, id) + } + + ch, err := s.GetChannel(ctx, id) + if err != nil { + t.Fatalf("GetChannel: %v", err) + } + if !ch.Policy.MandatorySubscription { + t.Fatal("mandatory_subscription still FALSE after resume, want re-asserted TRUE") + } + if got := memberSet(ch); !got[b.ID] { + t.Fatalf("dropped member %s not restored on resume; members = %v", b.ID, ch.MemberAccountIDs) + } + if !cursorExists(t, s, b.ID, id) { + t.Fatalf("restored member %s has no re-seeded delivery cursor (D2 hazard)", b.ID) + } +} + +// TestConvertOnAddRequiresNameAndConverts pins R4: adding a third agent to a DM +// without a convert name is invalid_argument; with a name the channel becomes +// kind=CHANNEL, takes the new name, leaves the reserved DM group (group_id NULL), +// and the third member is added. Per Matt's M1 ruling the converted channel is a +// normal opt-in channel: mandatory_subscription is cleared, the two incumbent DM +// parties are flipped subscribed=TRUE (so they keep receiving the conversation — +// there is no add/drop notification, delivery is the only signal), and the newly +// added third member joins opt-in (subscribed=FALSE unless its update said so). +func TestConvertOnAddRequiresNameAndConverts(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "owner") + a := mustAgent(t, s, owner.ID, "alice") + b := mustAgent(t, s, owner.ID, "bob") + c := mustAgent(t, s, owner.ID, "carol") + + id, _ := openDM(t, s, owner.ID, "dm--alice--bob", []AccountID{a.ID, b.ID}) + + // Add the third agent WITHOUT a convert name → invalid_argument. + _, _, err := s.UpdateChannelMembers(context.Background(), a.ID, id, + []MemberUpdate{{AccountID: c.ID}}, MemberUpdatesOptions{}) + sentinelIs(t, err, ErrInvalidArgument, "third-member add without a convert name") + + // The DM is untouched by the rejected add. + ch, err := s.GetChannel(context.Background(), id) + if err != nil { + t.Fatalf("GetChannel after rejected add: %v", err) + } + if ch.Kind != ChannelKindDM || containsAccount(ch.MemberAccountIDs, c.ID) { + t.Fatalf("rejected add mutated the DM: kind=%d members=%v", ch.Kind, ch.MemberAccountIDs) + } + + // Add the third agent WITH a convert name → conversion. + converted, _, err := s.UpdateChannelMembers(context.Background(), a.ID, id, + []MemberUpdate{{AccountID: c.ID}}, MemberUpdatesOptions{ConvertChannelName: "war-room"}) + if err != nil { + t.Fatalf("convert-on-add: %v", err) + } + if converted.Kind != ChannelKindChannel { + t.Fatalf("post-convert kind = %d, want ChannelKindChannel", converted.Kind) + } + if converted.Name != "war-room" { + t.Fatalf("post-convert name = %q, want war-room", converted.Name) + } + if converted.GroupID != "" { + t.Fatalf("post-convert group = %q, want empty (left the reserved DM group)", converted.GroupID) + } + if !containsAccount(converted.MemberAccountIDs, c.ID) { + t.Fatalf("third member %s not added: %v", c.ID, converted.MemberAccountIDs) + } + for _, m := range []AccountID{a.ID, b.ID} { + if !containsAccount(converted.MemberAccountIDs, m) { + t.Fatalf("existing member %s dropped by conversion: %v", m, converted.MemberAccountIDs) + } + } + // M1: the converted channel is a normal opt-in channel, not force-subscribe. + if converted.Policy.MandatorySubscription { + t.Fatal("converted channel still mandatory_subscription=true, want FALSE (normal opt-in channel)") + } + // The two incumbent DM parties are kept subscribed so they keep receiving. + for _, m := range []AccountID{a.ID, b.ID} { + if !memberSubscribed(t, s, id, m) { + t.Fatalf("incumbent DM party %s not subscribed after convert — would silently drop from the conversation", m) + } + } + // The newly-added third member joins opt-in (its update set no subscribe flag). + if memberSubscribed(t, s, id, c.ID) { + t.Fatalf("third member %s force-subscribed; a normal-channel add is opt-in", c.ID) + } + // M1 seed discipline: an opt-in (unsubscribed) add on the now-non-mandatory + // channel owes NO delivery cursor — it is seeded only when it subscribes (D2 + // seed-at-subscribe). A spurious seed here (the stale-mandatory bug) would + // later replay backlog from convert-time, so assert the cursor is absent. + if cursorExists(t, s, c.ID, id) { + t.Fatalf("third member %s has a delivery cursor after an opt-in add on a non-mandatory channel — spurious seed (stale mandatory flag)", c.ID) + } + // A member CAN now unsubscribe (the DM's force-subscribe is gone). + if _, _, err := s.UpdateChannelMembers(context.Background(), a.ID, id, + []MemberUpdate{{AccountID: a.ID, Unsubscribe: true}}, MemberUpdatesOptions{}); err != nil { + t.Fatalf("unsubscribe on converted channel rejected, want allowed: %v", err) + } +} + +// TestOpenDMAfterConvertMintsFreshPair pins the R4 open_dm semantics after a +// convert: the conversion freed the deterministic name, so a subsequent open of +// the same pair mints a FRESH pair DM at that name (created=true, a new id). +func TestOpenDMAfterConvertMintsFreshPair(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "owner") + a := mustAgent(t, s, owner.ID, "alice") + b := mustAgent(t, s, owner.ID, "bob") + c := mustAgent(t, s, owner.ID, "carol") + + const name = "dm--alice--bob" + first, _ := openDM(t, s, owner.ID, name, []AccountID{a.ID, b.ID}) + + // Convert the DM away by adding a third member with a name — frees `name`. + if _, _, err := s.UpdateChannelMembers(context.Background(), a.ID, first, + []MemberUpdate{{AccountID: c.ID}}, MemberUpdatesOptions{ConvertChannelName: "war-room"}); err != nil { + t.Fatalf("convert: %v", err) + } + + // Re-open the same pair: the freed name is available, so this is a FRESH DM. + second, created := openDM(t, s, owner.ID, name, []AccountID{a.ID, b.ID}) + if !created { + t.Fatal("re-open after convert reported created=false, want true (fresh pair DM)") + } + if second == first { + t.Fatal("re-open after convert resolved the converted channel, want a fresh id") + } + ch, err := s.GetChannel(context.Background(), second) + if err != nil { + t.Fatalf("GetChannel(fresh): %v", err) + } + if ch.Kind != ChannelKindDM || ch.Name != name { + t.Fatalf("fresh DM shape wrong: kind=%d name=%q", ch.Kind, ch.Name) + } +} + +// TestRemoveBelowTwoPartiesRejected pins R4: a remove that would strand a DM +// below two agent parties is invalid_argument (a one-party DM is not a thing). +func TestRemoveBelowTwoPartiesRejected(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "owner") + a := mustAgent(t, s, owner.ID, "alice") + b := mustAgent(t, s, owner.ID, "bob") + + id, _ := openDM(t, s, owner.ID, "dm--alice--bob", []AccountID{a.ID, b.ID}) + + _, _, err := s.UpdateChannelMembers(context.Background(), a.ID, id, + []MemberUpdate{{AccountID: b.ID, Remove: true}}, MemberUpdatesOptions{}) + sentinelIs(t, err, ErrInvalidArgument, "remove leaving a DM below two agent parties") + + // The remove rolled back: both parties remain. + ch, err := s.GetChannel(context.Background(), id) + if err != nil { + t.Fatalf("GetChannel after rejected remove: %v", err) + } + for _, m := range []AccountID{a.ID, b.ID} { + if !containsAccount(ch.MemberAccountIDs, m) { + t.Fatalf("party %s dropped by a rejected remove: %v", m, ch.MemberAccountIDs) + } + } +} diff --git a/go/internal/store/inputs.go b/go/internal/store/inputs.go index 4f6e9462..03d26d20 100644 --- a/go/internal/store/inputs.go +++ b/go/internal/store/inputs.go @@ -78,3 +78,14 @@ type MemberUpdate struct { // mandatory_subscription channel (T4) — a plain add is unaffected. Unsubscribe bool } + +// MemberUpdatesOptions carries the non-per-member scalars for +// UpdateChannelMembers. ConvertChannelName is the caller-supplied name a +// DM-to-CHANNEL conversion (R4) takes: a genuine member ADD on a kind=DM channel +// requires it (the third party converts the two-party DM into a named channel, +// freeing the DM's deterministic name), and it is ignored for a non-DM channel +// or a pure subscribe/remove. It is a separate scalar from the add/remove/ +// subscribe lists, so it rides here rather than in MemberUpdate. +type MemberUpdatesOptions struct { + ConvertChannelName string +} diff --git a/go/internal/store/messages_answer_post_test.go b/go/internal/store/messages_answer_post_test.go index 2e08685c..ac87858c 100644 --- a/go/internal/store/messages_answer_post_test.go +++ b/go/internal/store/messages_answer_post_test.go @@ -47,7 +47,7 @@ func TestAnswerAskPostsAnswerMessage(t *testing.T) { answerer := mustUser(t, s, "answerer") ch := mustNamedChannel(t, s, agent.ID, "room") // The answerer must be a member of the ask's channel to answer it. - if _, _, err := s.UpdateChannelMembers(ctx, agent.ID, ch.ID, []MemberUpdate{{AccountID: answerer.ID}}); err != nil { + if _, _, err := s.UpdateChannelMembers(ctx, agent.ID, ch.ID, []MemberUpdate{{AccountID: answerer.ID}}, MemberUpdatesOptions{}); err != nil { t.Fatalf("add answerer as member: %v", err) } diff --git a/go/internal/store/messages_authored_update_test.go b/go/internal/store/messages_authored_update_test.go index 55c1d495..0c990742 100644 --- a/go/internal/store/messages_authored_update_test.go +++ b/go/internal/store/messages_authored_update_test.go @@ -147,7 +147,7 @@ func TestUpdateMessageBlocksAsAuthorRevokedMemberIsNotFound(t *testing.T) { t.Fatalf("the author cannot edit while still a member: %v", err) } - if _, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{{AccountID: author.ID, Remove: true}}); err != nil { + if _, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{{AccountID: author.ID, Remove: true}}, MemberUpdatesOptions{}); err != nil { t.Fatalf("UpdateChannelMembers(remove): %v", err) } diff --git a/go/server/offline_mention_e2e_pgtest_test.go b/go/server/offline_mention_e2e_pgtest_test.go index 5c54b382..a1d171fe 100644 --- a/go/server/offline_mention_e2e_pgtest_test.go +++ b/go/server/offline_mention_e2e_pgtest_test.go @@ -177,7 +177,7 @@ func (w *mentionE2EWire) seedAgentMember(t *testing.T, handle string, subscribed } if _, _, err := w.store.UpdateChannelMembers(w.ctx, w.adminID, w.channel, []store.MemberUpdate{ {AccountID: agent.ID, Subscribed: subscribed}, - }); err != nil { + }, store.MemberUpdatesOptions{}); err != nil { t.Fatalf("UpdateChannelMembers(add %q, subscribed=%v): %v", handle, subscribed, err) } if err := w.store.RecordAgentPlacement(w.ctx, agent.ID, fakeRunnerID, containerFor(handle)); err != nil {