From c068b3a7347e9c4527bc5ee895a5c548e0e64ba6 Mon Sep 17 00:00:00 2001 From: mintaka Date: Mon, 31 Aug 2026 02:13:07 -0400 Subject: [PATCH] feat(store): migrate channels + channel_pins + coordination to sqlc (T3) Migrate the channels, channel_pins, and coordination store domains from inline pgx SQL to sqlc-generated typed queries (RIG-3034, design record sqlc-adoption T3), following the T2 recipe exactly. Pool-scoped calls go through s.q; tx-scoped through s.q.WithTx(tx) / db.New(tx). Every migrated statement is semantically identical to the inline SQL it replaced (same columns, WHERE, JOINs, ORDER BY, null-handling); the CTE-based channel and group visibility reads move into queries/channels.sql with their predicates inlined per query (sqlc has no fragment composition), textually identical so the stream-edge single-id checks cannot drift from the list reads. The DELETE-and-check-rows-affected member removal uses the :execrows kind so removeMember/SetCoordinationMembersTx keep their RowsAffected semantics. The coordination per-owner advisory lock pg_advisory_xact_lock(hashtext('coordination:' || $1)) is now a tx-bound LockOwnerCoordination call inside its transaction; the suffix-search channel-provision loop, the WithTx seam, error wrapping, domain-type mapping, and the FOR UPDATE lock / cap-check control flow all stay hand-written. Regenerated internal/store/db (drift-clean) and dropped the three migrated inline-sql-gate allowlist entries (24 -> 21). Co-authored-by: Matt Wilkinson --- go/internal/store/channel_pins.go | 89 ++- go/internal/store/channels.go | 462 ++++++--------- go/internal/store/coordination.go | 128 ++--- go/internal/store/db/channel_pins.sql.go | 157 +++++ go/internal/store/db/channels.sql.go | 640 +++++++++++++++++++++ go/internal/store/db/coordination.sql.go | 189 ++++++ go/internal/store/db/querier.go | 60 ++ go/internal/store/queries/channel_pins.sql | 30 + go/internal/store/queries/channels.sql | 224 ++++++++ go/internal/store/queries/coordination.sql | 39 ++ tools/inline-sql-gate/index.ts | 3 - 11 files changed, 1607 insertions(+), 414 deletions(-) create mode 100644 go/internal/store/db/channel_pins.sql.go create mode 100644 go/internal/store/db/channels.sql.go create mode 100644 go/internal/store/db/coordination.sql.go create mode 100644 go/internal/store/queries/channel_pins.sql create mode 100644 go/internal/store/queries/channels.sql create mode 100644 go/internal/store/queries/coordination.sql diff --git a/go/internal/store/channel_pins.go b/go/internal/store/channel_pins.go index 06d682a7d..59b81dedc 100644 --- a/go/internal/store/channel_pins.go +++ b/go/internal/store/channel_pins.go @@ -6,6 +6,8 @@ import ( "time" "github.com/jackc/pgx/v5" + + "github.com/RigelBuild/compass/go/internal/store/db" ) // maxChannelPins is the per-channel cap on the pinned board: at most this many @@ -126,10 +128,10 @@ func (s *Store) UnpinMessage(ctx context.Context, ch ChannelID, msg MessageID, b if err := requireBoardMutator(ctx, tx, ch, by, postPolicy, ownerAcct); err != nil { return nil, err } - if _, err := tx.Exec(ctx, - `DELETE FROM channel_pins WHERE channel_id = $1 AND message_id = $2`, - string(ch), string(msg), - ); err != nil { + if err := db.New(tx).DeleteChannelPin(ctx, db.DeleteChannelPinParams{ + ChannelID: string(ch), + MessageID: string(msg), + }); err != nil { return nil, fmt.Errorf("store: delete channel pin: %w", err) } @@ -156,17 +158,14 @@ func (s *Store) PinnedEntries(ctx context.Context, ch ChannelID) ([]PinnedEntry, // post_policy and owner so the caller can gate the mutator under the lock without // a second query. An unknown channel matches zero rows and is ErrNotFound. func lockChannelForPins(ctx context.Context, tx pgx.Tx, ch ChannelID) (postPolicy int32, ownerAcct string, err error) { - err = tx.QueryRow(ctx, - `SELECT post_policy, COALESCE(owner_account_id, '') FROM channels WHERE id = $1 FOR UPDATE`, - string(ch), - ).Scan(&postPolicy, &ownerAcct) + row, err := db.New(tx).LockChannelForPins(ctx, string(ch)) if err != nil { if noRows(err) { return 0, "", fmt.Errorf("%w: channel %q", ErrNotFound, ch) } return 0, "", fmt.Errorf("store: lock channel for pins: %w", err) } - return postPolicy, ownerAcct, nil + return int32(row.PostPolicy), row.OwnerAccountID, nil } // requireBoardMutator enforces board-mutation authz for `by` under the caller's @@ -193,12 +192,10 @@ func requireBoardMutator(ctx context.Context, tx pgx.Tx, ch ChannelID, by Accoun // another channel or no message at all is ErrNotFound (the not-found/forbidden // merge). func requireMessageInChannel(ctx context.Context, tx pgx.Tx, ch ChannelID, msg MessageID) error { - var one int - err := tx.QueryRow(ctx, - `SELECT 1 FROM messages m JOIN topics t ON t.id = m.topic_id - WHERE m.id = $1 AND t.channel_id = $2`, - string(msg), string(ch), - ).Scan(&one) + _, err := db.New(tx).MessageInChannel(ctx, db.MessageInChannelParams{ + ID: string(msg), + ChannelID: string(ch), + }) if err != nil { if noRows(err) { return fmt.Errorf("%w: message %q not in channel %q", ErrNotFound, msg, ch) @@ -215,14 +212,12 @@ func requireMessageInChannel(ctx context.Context, tx pgx.Tx, ch ChannelID, msg M func pinFresh(ctx context.Context, tx pgx.Tx, ch ChannelID, msg MessageID, by AccountID) error { // Count and next-position under the lock: the lock makes this read-modify // -write race-free, so the cap cannot be exceeded by a concurrent pin. - var count int - var nextPos int32 - if err := tx.QueryRow(ctx, - `SELECT count(*), COALESCE(MAX(position), -1) + 1 FROM channel_pins WHERE channel_id = $1`, - string(ch), - ).Scan(&count, &nextPos); err != nil { + counts, err := db.New(tx).CountChannelPins(ctx, string(ch)) + if err != nil { return fmt.Errorf("store: count channel pins: %w", err) } + count := counts.Count + nextPos := counts.NextPosition if count >= maxChannelPins { return fmt.Errorf("%w: channel %q already has the maximum of %d pins", ErrFailedPrecondition, ch, maxChannelPins) } @@ -239,11 +234,10 @@ func pinFresh(ctx context.Context, tx pgx.Tx, ch ChannelID, msg MessageID, by Ac func pinRepoint(ctx context.Context, tx pgx.Tx, ch ChannelID, msg, replace MessageID, by AccountID) error { // Delete the replaced entry and capture its position in one statement; zero // rows means replace was not pinned — the CAS is lost. - var pos int32 - err := tx.QueryRow(ctx, - `DELETE FROM channel_pins WHERE channel_id = $1 AND message_id = $2 RETURNING position`, - string(ch), string(replace), - ).Scan(&pos) + pos, err := db.New(tx).DeleteChannelPinReturningPosition(ctx, db.DeleteChannelPinReturningPositionParams{ + ChannelID: string(ch), + MessageID: string(replace), + }) if err != nil { if noRows(err) { return fmt.Errorf("%w: pin %q is no longer on channel %q's board, re-read", ErrConflict, replace, ch) @@ -259,11 +253,13 @@ func pinRepoint(ctx context.Context, tx pgx.Tx, ch ChannelID, msg, replace Messa // insertPin writes one channel_pins pointer, mapping a primary-key conflict (the // message already pinned in ch) to ErrConflict. func insertPin(ctx context.Context, tx pgx.Tx, ch ChannelID, msg MessageID, pos int32, by AccountID) error { - if _, err := tx.Exec(ctx, - `INSERT INTO channel_pins (channel_id, message_id, position, pinned_at_unix_ms, pinned_by_account_id) - VALUES ($1, $2, $3, $4, $5)`, - string(ch), string(msg), pos, time.Now().UTC().UnixMilli(), string(by), - ); err != nil { + if err := db.New(tx).InsertChannelPin(ctx, db.InsertChannelPinParams{ + ChannelID: string(ch), + MessageID: string(msg), + Position: pos, + PinnedAtUnixMs: time.Now().UTC().UnixMilli(), + PinnedByAccountID: string(by), + }); err != nil { if pgErrIs(err, pgUniqueViolation) { return fmt.Errorf("%w: message %q is already pinned in channel %q", ErrConflict, msg, ch) } @@ -274,30 +270,19 @@ func insertPin(ctx context.Context, tx pgx.Tx, ch ChannelID, msg MessageID, pos // pinnedEntriesTx reads ch's board ordered by position from any querier (the pool // for the read-only PinnedEntries, or the open tx for the mutating returns). -func pinnedEntriesTx(ctx context.Context, q querier, ch ChannelID) ([]PinnedEntry, error) { - rows, err := q.Query(ctx, - `SELECT message_id, position, pinned_at_unix_ms, pinned_by_account_id - FROM channel_pins WHERE channel_id = $1 ORDER BY position`, - string(ch), - ) +func pinnedEntriesTx(ctx context.Context, q db.DBTX, ch ChannelID) ([]PinnedEntry, error) { + rows, err := db.New(q).PinnedEntries(ctx, string(ch)) if err != nil { return nil, fmt.Errorf("store: query channel pins: %w", err) } - defer rows.Close() - - var entries []PinnedEntry - for rows.Next() { - var e PinnedEntry - var msgID, by string - if err := rows.Scan(&msgID, &e.Position, &e.PinnedAtUnixMs, &by); err != nil { - return nil, fmt.Errorf("store: scan channel pin: %w", err) - } - e.MessageID = MessageID(msgID) - e.PinnedByAccountID = AccountID(by) - entries = append(entries, e) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate channel pins: %w", err) + entries := make([]PinnedEntry, 0, len(rows)) + for _, row := range rows { + entries = append(entries, PinnedEntry{ + MessageID: MessageID(row.MessageID), + Position: row.Position, + PinnedAtUnixMs: row.PinnedAtUnixMs, + PinnedByAccountID: AccountID(row.PinnedByAccountID), + }) } return entries, nil } diff --git a/go/internal/store/channels.go b/go/internal/store/channels.go index 09cef7373..c83335a6c 100644 --- a/go/internal/store/channels.go +++ b/go/internal/store/channels.go @@ -6,6 +6,8 @@ import ( "slices" "github.com/jackc/pgx/v5" + + "github.com/RigelBuild/compass/go/internal/store/db" ) // CreateChannelGroup inserts a namespace group owned by ownerUserID. When a @@ -37,25 +39,26 @@ func (s *Store) CreateChannelGroup(ctx context.Context, ownerUserID AccountID, g if err := requireGroupCreateAuthz(ctx, tx, ownerUserID, g.ParentGroupID); err != nil { return ChannelGroup{}, err } - var parentVis int32 - if err := tx.QueryRow(ctx, - "SELECT visibility FROM channel_groups WHERE id = $1", string(g.ParentGroupID), - ).Scan(&parentVis); err != nil { + parentVis, err := s.q.WithTx(tx).GetChannelGroupVisibility(ctx, string(g.ParentGroupID)) + if err != nil { return ChannelGroup{}, fmt.Errorf("store: read parent group: %w", err) } // A higher enum value is more open (OWNER=0 < SHARED=1), so the child's // value must not exceed the parent's. - if int32(g.Visibility) > parentVis { + if int32(g.Visibility) > int32(parentVis) { return ChannelGroup{}, fmt.Errorf( "%w: group visibility %d wider than parent %d", ErrInvalidArgument, g.Visibility, parentVis) } } id := newID() - if _, err := tx.Exec(ctx, - "INSERT INTO channel_groups (id, name, parent_group_id, owner_user_id, visibility) VALUES ($1, $2, NULLIF($3, ''), $4, $5)", - id, g.Name, string(g.ParentGroupID), string(ownerUserID), int32(g.Visibility), - ); err != nil { + if err := s.q.WithTx(tx).InsertChannelGroup(ctx, db.InsertChannelGroupParams{ + ID: id, + Name: g.Name, + Column3: string(g.ParentGroupID), + OwnerUserID: string(ownerUserID), + Visibility: int16(g.Visibility), //nolint:gosec // G115: ChannelGroupVisibility is a CHECK-constrained 0/1 enum (channel_groups.visibility), always within int16 + }); err != nil { return ChannelGroup{}, fmt.Errorf("store: insert channel group: %w", err) } if err := tx.Commit(ctx); err != nil { @@ -136,12 +139,15 @@ func (s *Store) CreateChannel(ctx context.Context, actor AccountID, c NewChannel } } - 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)", - id, c.Name, string(c.GroupID), int32(c.Kind), - int32(c.Policy.PostPolicy), string(c.Policy.OwnerAccountID), c.Policy.MandatorySubscription, - ); err != nil { + if err := s.q.WithTx(tx).InsertChannel(ctx, db.InsertChannelParams{ + ID: id, + Name: c.Name, + Column3: string(c.GroupID), + Kind: int16(c.Kind), //nolint:gosec // G115: ChannelKind is a CHECK-constrained 0/1/2 enum (channels.kind), always within int16 + PostPolicy: int16(c.Policy.PostPolicy), + Column6: string(c.Policy.OwnerAccountID), + MandatorySubscription: c.Policy.MandatorySubscription, + }); err != nil { if pgErrIs(err, pgUniqueViolation) { return Channel{}, fmt.Errorf("%w: channel %q already exists in group %q", ErrConflict, c.Name, c.GroupID) } @@ -164,12 +170,12 @@ func (s *Store) CreateChannel(ctx context.Context, actor AccountID, c NewChannel if c.Policy.OwnerAccountID != "" && !slices.Contains(members, c.Policy.OwnerAccountID) { return Channel{}, fmt.Errorf("%w: owner account %q must be a channel member", ErrInvalidArgument, c.Policy.OwnerAccountID) } + qtx := s.q.WithTx(tx) 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 { + if err := qtx.EnsureChannelMember(ctx, db.EnsureChannelMemberParams{ + ChannelID: id, + AccountID: string(m), + }); err != nil { if pgErrIs(err, pgForeignKeyViolation) { return Channel{}, fmt.Errorf("%w: unknown member account %q", ErrInvalidArgument, m) } @@ -231,121 +237,32 @@ func expandOwnerMembership(ctx context.Context, tx pgx.Tx, actor AccountID, requ for _, m := range ordered { ids = append(ids, string(m)) } - rows, err := tx.Query(ctx, - "SELECT owner_user_id FROM agent_accounts WHERE account_id = ANY($1)", ids, - ) + owners, err := db.New(tx).AgentOwnersByIDs(ctx, ids) if err != nil { return nil, fmt.Errorf("store: resolve agent owners: %w", err) } - defer rows.Close() - - var owners []AccountID - for rows.Next() { - var owner string - if err := rows.Scan(&owner); err != nil { - return nil, fmt.Errorf("store: scan agent owner: %w", err) - } - owners = append(owners, AccountID(owner)) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate agent owners: %w", err) - } for _, o := range owners { - add(o) + add(AccountID(o)) } return ordered, nil } -// effectiveVisibilityCTE computes each channel group's effective visibility — -// the most restrictive value on its path to the root (D9) — by walking the -// parent chain. Every read and stream-filter predicate that gates on group -// visibility opens with this identical CTE so they compute `effective(id, -// eff_vis)` the same way and cannot drift. eff_vis = 1 is SHARED. -const effectiveVisibilityCTE = ` - WITH RECURSIVE ancestry AS ( - SELECT id, parent_group_id, visibility AS min_vis - FROM channel_groups - UNION ALL - SELECT a.id, g.parent_group_id, LEAST(a.min_vis, g.visibility) - FROM ancestry a - JOIN channel_groups g ON g.id = a.parent_group_id - ), - effective AS ( - SELECT id, MIN(min_vis) AS eff_vis - FROM ancestry - GROUP BY id - )` - -// viewerCTE resolves the set of user ids a caller ($1) views as: itself, plus — -// for an agent caller — its owning user, so an agent sees its owner's groups. -// Appended after effectiveVisibilityCTE (hence the leading comma). Referenced by -// the group predicate's owner check. -const viewerCTE = `, - viewer AS ( - SELECT owner_user_id AS uid FROM agent_accounts WHERE account_id = $1 - UNION ALL - SELECT $1 AS uid - )` - -// channelVisiblePredicate is the ListChannels visibility rule as a reusable -// boolean over a channel row aliased `c` and the `effective` CTE, viewer $1: the -// caller is a member (which governs DM/GROUP_DM directly, design.md:235-243), or -// it is a plain grouped channel (kind=0) whose group's effective visibility is -// SHARED. An ungrouped channel is owner-scoped, visible only through membership. -// Shared by ListChannels and ChannelVisibleTo so the stream edge cannot drift -// from the read. -const channelVisiblePredicate = `( - EXISTS ( - SELECT 1 FROM channel_members cm - WHERE cm.channel_id = c.id AND cm.account_id = $1 - ) - OR ( - c.kind = 0 AND c.group_id IS NOT NULL AND EXISTS ( - SELECT 1 FROM effective e WHERE e.id = c.group_id AND e.eff_vis = 1 - ) - ) - )` - -// groupVisiblePredicate is the ListChannelGroups visibility rule as a reusable -// boolean over a group row aliased `g`, the `effective` CTE and `viewer` CTE: the -// group's effective visibility is SHARED, or the caller (or its owning user) -// owns it. Shared by ListChannelGroups and ChannelGroupVisibleTo. -const groupVisiblePredicate = `(e.eff_vis = 1 OR g.owner_user_id IN (SELECT uid FROM viewer))` - // ListChannelGroups returns the channel groups visible to visibleTo (see // groupVisiblePredicate / effectiveVisibilityCTE for the rule). func (s *Store) ListChannelGroups(ctx context.Context, visibleTo AccountID) ([]ChannelGroup, error) { - const q = effectiveVisibilityCTE + viewerCTE + ` - SELECT g.id, g.name, COALESCE(g.parent_group_id, ''), g.owner_user_id, g.visibility - FROM channel_groups g - JOIN effective e ON e.id = g.id - WHERE ` + groupVisiblePredicate + ` - ORDER BY g.name` - rows, err := s.pool.Query(ctx, q, string(visibleTo)) + rows, err := s.q.ListChannelGroups(ctx, string(visibleTo)) if err != nil { return nil, fmt.Errorf("store: list channel groups: %w", err) } - defer rows.Close() - var groups []ChannelGroup - for rows.Next() { - var ( - g ChannelGroup - id, name, parent, owner string - visibility int32 - ) - if err := rows.Scan(&id, &name, &parent, &owner, &visibility); err != nil { - return nil, fmt.Errorf("store: scan channel group: %w", err) - } - g.ID = ChannelGroupID(id) - g.Name = name - g.ParentGroupID = ChannelGroupID(parent) - g.OwnerUserID = AccountID(owner) - g.Visibility = ChannelGroupVisibility(visibility) - groups = append(groups, g) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate channel groups: %w", err) + for _, row := range rows { + groups = append(groups, ChannelGroup{ + ID: ChannelGroupID(row.ID), + Name: row.Name, + ParentGroupID: ChannelGroupID(row.ParentGroupID), + OwnerUserID: AccountID(row.OwnerUserID), + Visibility: ChannelGroupVisibility(row.Visibility), + }) } return groups, nil } @@ -355,14 +272,11 @@ func (s *Store) ListChannelGroups(ctx context.Context, visibleTo AccountID) ([]C // edge to filter ChannelGroupChanged at read-parity. Shares the CTEs + predicate // with the list read so the two cannot drift. func (s *Store) ChannelGroupVisibleTo(ctx context.Context, actor AccountID, groupID ChannelGroupID) (bool, error) { - const q = effectiveVisibilityCTE + viewerCTE + ` - SELECT EXISTS ( - SELECT 1 FROM channel_groups g - JOIN effective e ON e.id = g.id - WHERE g.id = $2 AND ` + groupVisiblePredicate + ` - )` - var visible bool - if err := s.pool.QueryRow(ctx, q, string(actor), string(groupID)).Scan(&visible); err != nil { + visible, err := s.q.ChannelGroupVisibleTo(ctx, db.ChannelGroupVisibleToParams{ + AccountID: string(actor), + ID: string(groupID), + }) + if err != nil { return false, fmt.Errorf("store: check group visibility: %w", err) } return visible, nil @@ -371,19 +285,15 @@ func (s *Store) ChannelGroupVisibleTo(ctx context.Context, actor AccountID, grou // ListChannels returns the channels visible to visibleTo (see // channelVisiblePredicate / effectiveVisibilityCTE for the rule). func (s *Store) ListChannels(ctx context.Context, visibleTo AccountID) ([]Channel, error) { - const q = effectiveVisibilityCTE + ` - SELECT c.id, c.name, COALESCE(c.group_id, ''), c.kind, c.post_policy, COALESCE(c.owner_account_id, ''), c.mandatory_subscription - FROM channels c - WHERE ` + channelVisiblePredicate + ` - ORDER BY c.name` - rows, err := s.pool.Query(ctx, q, string(visibleTo)) + rows, err := s.q.ListChannels(ctx, string(visibleTo)) if err != nil { return nil, fmt.Errorf("store: list channels: %w", err) } - defer rows.Close() - - channels, err := scanChannels(ctx, s.pool, rows) - if err != nil { + var channels []Channel + for _, row := range rows { + channels = append(channels, channelFromRow(row.ID, row.Name, row.GroupID, row.Kind, row.PostPolicy, row.OwnerAccountID, row.MandatorySubscription)) + } + if err := loadChannelMembers(ctx, s.pool, channels); err != nil { return nil, err } return channels, nil @@ -395,13 +305,11 @@ func (s *Store) ListChannels(ctx context.Context, visibleTo AccountID) ([]Channe // non-member viewer (which bare membership would wrongly drop) while a private // channel's does not. Shares the CTE + predicate with the list read. func (s *Store) ChannelVisibleTo(ctx context.Context, actor AccountID, channelID ChannelID) (bool, error) { - const q = effectiveVisibilityCTE + ` - SELECT EXISTS ( - SELECT 1 FROM channels c - WHERE c.id = $2 AND ` + channelVisiblePredicate + ` - )` - var visible bool - if err := s.pool.QueryRow(ctx, q, string(actor), string(channelID)).Scan(&visible); err != nil { + visible, err := s.q.ChannelVisibleTo(ctx, db.ChannelVisibleToParams{ + AccountID: string(actor), + ID: string(channelID), + }) + if err != nil { return false, fmt.Errorf("store: check channel visibility: %w", err) } return visible, nil @@ -426,19 +334,18 @@ func (s *Store) ChannelVisibleTo(ctx context.Context, actor AccountID, channelID // Match is exact-case, mirroring the channels_group_name_key uniqueness the store // enforces on writes. func (s *Store) ChannelByNameForViewer(ctx context.Context, viewer AccountID, name string) (Channel, error) { - const q = effectiveVisibilityCTE + ` - SELECT c.id, c.name, COALESCE(c.group_id, ''), c.kind, c.post_policy, COALESCE(c.owner_account_id, ''), c.mandatory_subscription - FROM channels c - WHERE c.name = $2 AND ` + channelVisiblePredicate + ` - ORDER BY c.id` - rows, err := s.pool.Query(ctx, q, string(viewer), name) + rows, err := s.q.ChannelsByNameForViewer(ctx, db.ChannelsByNameForViewerParams{ + AccountID: string(viewer), + Name: name, + }) if err != nil { return Channel{}, fmt.Errorf("store: resolve channel by name: %w", err) } - defer rows.Close() - - channels, err := scanChannels(ctx, s.pool, rows) - if err != nil { + var channels []Channel + for _, row := range rows { + channels = append(channels, channelFromRow(row.ID, row.Name, row.GroupID, row.Kind, row.PostPolicy, row.OwnerAccountID, row.MandatorySubscription)) + } + if err := loadChannelMembers(ctx, s.pool, channels); err != nil { return Channel{}, err } switch len(channels) { @@ -492,16 +399,12 @@ func (s *Store) UpdateChannelMembers(ctx context.Context, actor AccountID, chann // 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, kind FROM channels WHERE id = $1 FOR UPDATE", string(channelID), - ).Scan(&mandatory, &kindRaw); err != nil { + lock, err := s.q.WithTx(tx).LockChannelMandatoryKind(ctx, string(channelID)) + if err != nil { return Channel{}, nil, fmt.Errorf("store: read channel mandatory flag: %w", err) } - kind := ChannelKind(kindRaw) + mandatory := lock.MandatorySubscription + kind := ChannelKind(lock.Kind) // 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 @@ -616,21 +519,18 @@ func maybeConvertDM(ctx context.Context, tx pgx.Tx, channelID ChannelID, kind Ch // 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 { + if err := db.New(tx).ConvertDMChannel(ctx, db.ConvertDMChannelParams{ + Kind: int16(ChannelKindChannel), + Name: opts.ConvertChannelName, + ID: 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 { + if err := db.New(tx).SubscribeConvertedDMParties(ctx, string(channelID)); err != nil { return kind, false, fmt.Errorf("store: subscribe converted dm parties: %w", err) } return ChannelKindChannel, true, nil @@ -647,11 +547,11 @@ func hasGenuineAdd(ctx context.Context, tx pgx.Tx, channelID ChannelID, 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 { + exists, err := db.New(tx).ChannelMemberExists(ctx, db.ChannelMemberExistsParams{ + ChannelID: string(channelID), + AccountID: string(u.AccountID), + }) + if err != nil { return false, fmt.Errorf("store: probe member presence: %w", err) } if !exists { @@ -672,13 +572,8 @@ func requireDMTwoParties(ctx context.Context, tx pgx.Tx, channelID ChannelID, ki 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 { + parties, err := db.New(tx).CountAgentMembers(ctx, string(channelID)) + if err != nil { return fmt.Errorf("store: count agent members: %w", err) } if parties < 2 { @@ -694,27 +589,25 @@ func requireDMTwoParties(ctx context.Context, tx pgx.Tx, channelID ChannelID, ki // read. Reports whether a row was actually deleted — removing an account that // was not a member is a no-op that owes no ChannelChanged to anyone. func removeMember(ctx context.Context, tx pgx.Tx, channelID ChannelID, accountID AccountID) (bool, error) { - var ownsPresentAgent bool - if err := tx.QueryRow(ctx, - "SELECT EXISTS ("+ - "SELECT 1 FROM agent_accounts aa "+ - "JOIN channel_members cm ON cm.account_id = aa.account_id "+ - "WHERE aa.owner_user_id = $1 AND cm.channel_id = $2 AND aa.account_id <> $1)", - string(accountID), string(channelID), - ).Scan(&ownsPresentAgent); err != nil { + qtx := db.New(tx) + ownsPresentAgent, err := qtx.OwnerHasPresentAgent(ctx, db.OwnerHasPresentAgentParams{ + OwnerUserID: string(accountID), + ChannelID: string(channelID), + }) + if err != nil { return false, fmt.Errorf("store: check dependent agents: %w", err) } if ownsPresentAgent { return false, fmt.Errorf("%w: cannot remove %q while an agent it owns remains in the channel", ErrInvalidArgument, accountID) } - tag, err := tx.Exec(ctx, - "DELETE FROM channel_members WHERE channel_id = $1 AND account_id = $2", - string(channelID), string(accountID), - ) + rowsAffected, err := qtx.DeleteChannelMember(ctx, db.DeleteChannelMemberParams{ + ChannelID: string(channelID), + AccountID: string(accountID), + }) if err != nil { return false, fmt.Errorf("store: remove member: %w", err) } - return tag.RowsAffected() > 0, nil + return rowsAffected > 0, nil } // addOrUpdateMember adds (or subscribe-flips) the directly-named member, then @@ -735,13 +628,14 @@ func addOrUpdateMember(ctx context.Context, tx pgx.Tx, channelID ChannelID, u Me if err != nil { return err } + qtx := db.New(tx) for i, m := range toAdd { if i == 0 { - if _, err := tx.Exec(ctx, - "INSERT INTO channel_members (channel_id, account_id, subscribed) VALUES ($1, $2, $3) "+ - "ON CONFLICT (channel_id, account_id) DO UPDATE SET subscribed = EXCLUDED.subscribed", - string(channelID), string(m), u.Subscribed, - ); err != nil { + if err := qtx.UpsertChannelMember(ctx, db.UpsertChannelMemberParams{ + ChannelID: string(channelID), + AccountID: string(m), + Subscribed: u.Subscribed, + }); err != nil { return upsertMemberErr(err, m) } // Seed this member's delivery cursor in the SAME txn as the member @@ -757,11 +651,10 @@ func addOrUpdateMember(ctx context.Context, tx pgx.Tx, channelID ChannelID, u Me } continue } - 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 { + if err := qtx.EnsureChannelMember(ctx, db.EnsureChannelMemberParams{ + ChannelID: string(channelID), + AccountID: string(m), + }); err != nil { return upsertMemberErr(err, m) } } @@ -816,14 +709,8 @@ func (s *Store) SetChannelPolicy(ctx context.Context, actor AccountID, channelID // lock so both the newly-mandatory transition and the owner-authz gate are // computed against the committed state, serialized against a concurrent // policy change on the same channel. - var ( - wasMandatory bool - currentOwner string - ) - if err := tx.QueryRow(ctx, - "SELECT mandatory_subscription, COALESCE(owner_account_id, '') FROM channels WHERE id = $1 FOR UPDATE", - string(channelID), - ).Scan(&wasMandatory, ¤tOwner); err != nil { + lock, err := s.q.WithTx(tx).LockChannelPolicy(ctx, string(channelID)) + if err != nil { // Defensive/unreachable: requireChannelMember above already proved the // channel exists (a nonexistent channel has no members), so this // FOR UPDATE cannot return no-rows. Kept for symmetry with messages.go. @@ -832,6 +719,8 @@ func (s *Store) SetChannelPolicy(ctx context.Context, actor AccountID, channelID } return Channel{}, fmt.Errorf("store: lock channel for policy: %w", err) } + wasMandatory := lock.MandatorySubscription + currentOwner := lock.OwnerAccountID // T4 owner-only policy gate. SetChannelPolicy is create-or-update of policy: // an ownerless channel (empty owner, the only legal state when OPEN) has no @@ -873,11 +762,11 @@ func (s *Store) SetChannelPolicy(ctx context.Context, actor AccountID, channelID // existing owner) reaches this after the owner gate, so the membership EXISTS // reveals nothing an authorized caller should not already know. if p.OwnerAccountID != "" { - var ownerIsMember bool - if err := tx.QueryRow(ctx, - "SELECT EXISTS (SELECT 1 FROM channel_members WHERE channel_id = $1 AND account_id = $2)", - string(channelID), string(p.OwnerAccountID), - ).Scan(&ownerIsMember); err != nil { + ownerIsMember, err := s.q.WithTx(tx).ChannelMemberExists(ctx, db.ChannelMemberExistsParams{ + ChannelID: string(channelID), + AccountID: string(p.OwnerAccountID), + }) + if err != nil { return Channel{}, fmt.Errorf("store: check owner membership: %w", err) } if !ownerIsMember { @@ -885,10 +774,12 @@ func (s *Store) SetChannelPolicy(ctx context.Context, actor AccountID, channelID } } - if _, err := tx.Exec(ctx, - "UPDATE channels SET post_policy = $2, owner_account_id = NULLIF($3, ''), mandatory_subscription = $4 WHERE id = $1", - string(channelID), int32(p.PostPolicy), string(p.OwnerAccountID), p.MandatorySubscription, - ); err != nil { + if err := s.q.WithTx(tx).UpdateChannelPolicy(ctx, db.UpdateChannelPolicyParams{ + ID: string(channelID), + PostPolicy: int16(p.PostPolicy), //nolint:gosec // G115: ChannelPostPolicy is a CHECK-constrained 0/1 enum (channels.post_policy), always within int16 + Column3: string(p.OwnerAccountID), + MandatorySubscription: p.MandatorySubscription, + }); err != nil { if pgErrIs(err, pgForeignKeyViolation) { return Channel{}, fmt.Errorf("%w: unknown owner account %q", ErrInvalidArgument, p.OwnerAccountID) } @@ -927,89 +818,90 @@ func (s *Store) GetChannel(ctx context.Context, id ChannelID) (Channel, error) { // getChannel loads one channel with its member set, or ErrNotFound. func (s *Store) getChannel(ctx context.Context, id ChannelID) (Channel, error) { - rows, err := s.pool.Query(ctx, - "SELECT id, name, COALESCE(group_id, ''), kind, post_policy, COALESCE(owner_account_id, ''), mandatory_subscription FROM channels WHERE id = $1", string(id)) + row, err := s.q.GetChannel(ctx, string(id)) if err != nil { + if noRows(err) { + return Channel{}, fmt.Errorf("%w: channel %q", ErrNotFound, id) + } return Channel{}, fmt.Errorf("store: get channel: %w", err) } - defer rows.Close() - channels, err := scanChannels(ctx, s.pool, rows) - if err != nil { + channels := []Channel{channelFromRow(row.ID, row.Name, row.GroupID, row.Kind, row.PostPolicy, row.OwnerAccountID, row.MandatorySubscription)} + if err := loadChannelMembers(ctx, s.pool, channels); err != nil { return Channel{}, err } - if len(channels) == 0 { - return Channel{}, fmt.Errorf("%w: channel %q", ErrNotFound, id) - } return channels[0], nil } // scanChannels reads channel rows and populates each channel's member set with // one follow-up query over the whole id set, so member loading is O(1) // round-trips rather than one per channel. -func scanChannels(ctx context.Context, q querier, rows pgx.Rows) ([]Channel, error) { - var ( - channels []Channel - ids []string - ) - byID := make(map[ChannelID]int) +func scanChannels(ctx context.Context, q db.DBTX, rows pgx.Rows) ([]Channel, error) { //nolint:unused // called by the pgtest-tagged test helper coordChannels (coordination_pgtest_test.go); the untagged lint build excludes that file and reads this as dead + var channels []Channel for rows.Next() { var ( - c Channel id, name, groupID string - kind int32 - postPolicy int32 + kind int16 + postPolicy int16 ownerAccountID string mandatorySubscription bool ) if err := rows.Scan(&id, &name, &groupID, &kind, &postPolicy, &ownerAccountID, &mandatorySubscription); err != nil { return nil, fmt.Errorf("store: scan channel: %w", err) } - c.ID = ChannelID(id) - c.Name = name - c.GroupID = ChannelGroupID(groupID) - c.Kind = ChannelKind(kind) - c.Policy = ChannelPolicy{ - PostPolicy: ChannelPostPolicy(postPolicy), - OwnerAccountID: AccountID(ownerAccountID), - MandatorySubscription: mandatorySubscription, - } - byID[c.ID] = len(channels) - channels = append(channels, c) - ids = append(ids, id) + channels = append(channels, channelFromRow(id, name, groupID, kind, postPolicy, ownerAccountID, mandatorySubscription)) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("store: iterate channels: %w", err) } - if len(channels) == 0 { - return channels, nil + if err := loadChannelMembers(ctx, q, channels); err != nil { + return nil, err + } + return channels, nil +} + +// channelFromRow builds the base Channel (id, name, group, kind, policy) from +// the shared seven-column channel projection every channel read selects; the +// caller populates the member/subscriber sets with loadChannelMembers. +func channelFromRow(id, name, groupID string, kind, postPolicy int16, ownerAccountID string, mandatorySubscription bool) Channel { + return Channel{ + ID: ChannelID(id), + Name: name, + GroupID: ChannelGroupID(groupID), + Kind: ChannelKind(kind), + Policy: ChannelPolicy{ + PostPolicy: ChannelPostPolicy(postPolicy), + OwnerAccountID: AccountID(ownerAccountID), + MandatorySubscription: mandatorySubscription, + }, } +} - memRows, err := q.Query(ctx, - "SELECT channel_id, account_id, subscribed FROM channel_members WHERE channel_id = ANY($1) ORDER BY account_id", - ids, - ) +// loadChannelMembers populates each channel's member and subscriber sets with +// one follow-up query over the whole id set, so member loading is O(1) +// round-trips rather than one per channel. Runs against the pool or a tx (any +// db.DBTX), mirroring the former scanChannels member follow-up. +func loadChannelMembers(ctx context.Context, q db.DBTX, channels []Channel) error { + if len(channels) == 0 { + return nil + } + byID := make(map[ChannelID]int, len(channels)) + ids := make([]string, len(channels)) + for i := range channels { + byID[channels[i].ID] = i + ids[i] = string(channels[i].ID) + } + members, err := db.New(q).ChannelMembersByChannelIDs(ctx, ids) if err != nil { - return nil, fmt.Errorf("store: load channel members: %w", err) + return fmt.Errorf("store: load channel members: %w", err) } - defer memRows.Close() - for memRows.Next() { - var ( - channelID, accountID string - subscribed bool - ) - if err := memRows.Scan(&channelID, &accountID, &subscribed); err != nil { - return nil, fmt.Errorf("store: scan channel member: %w", err) - } - idx := byID[ChannelID(channelID)] - channels[idx].MemberAccountIDs = append(channels[idx].MemberAccountIDs, AccountID(accountID)) - if subscribed { - channels[idx].SubscriberAccountIDs = append(channels[idx].SubscriberAccountIDs, AccountID(accountID)) + for _, m := range members { + idx := byID[ChannelID(m.ChannelID)] + channels[idx].MemberAccountIDs = append(channels[idx].MemberAccountIDs, AccountID(m.AccountID)) + if m.Subscribed { + channels[idx].SubscriberAccountIDs = append(channels[idx].SubscriberAccountIDs, AccountID(m.AccountID)) } } - if err := memRows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate channel members: %w", err) - } - return channels, nil + return nil } // OpenAgentWorkspace returns the agent's observation-pane workspace, creating it @@ -1047,21 +939,19 @@ func (s *Store) OpenAgentWorkspace(ctx context.Context, actor AccountID, agentAc // Insert-or-return: create the workspace on first open, else return the // existing row. ON CONFLICT DO NOTHING then a read covers the concurrent // case without a unique-violation surfacing to the caller. - if _, err := tx.Exec(ctx, - "INSERT INTO agent_workspaces (id, agent_account_id) VALUES ($1, $2) "+ - "ON CONFLICT (agent_account_id) DO NOTHING", - id, string(agentAccountID), - ); err != nil { + qtx := s.q.WithTx(tx) + if err := qtx.InsertAgentWorkspaceIgnore(ctx, db.InsertAgentWorkspaceIgnoreParams{ + ID: id, + AgentAccountID: string(agentAccountID), + }); err != nil { if pgErrIs(err, pgForeignKeyViolation) { return AgentWorkspace{}, fmt.Errorf("%w: unknown agent %q", ErrInvalidArgument, agentAccountID) } return AgentWorkspace{}, fmt.Errorf("store: open workspace: %w", err) } - var wsID string - if err := tx.QueryRow(ctx, - "SELECT id FROM agent_workspaces WHERE agent_account_id = $1", string(agentAccountID), - ).Scan(&wsID); err != nil { + wsID, err := qtx.GetAgentWorkspaceID(ctx, string(agentAccountID)) + if err != nil { return AgentWorkspace{}, fmt.Errorf("store: read workspace: %w", err) } if err := tx.Commit(ctx); err != nil { diff --git a/go/internal/store/coordination.go b/go/internal/store/coordination.go index 8e5b29097..a17815183 100644 --- a/go/internal/store/coordination.go +++ b/go/internal/store/coordination.go @@ -5,6 +5,9 @@ import ( "fmt" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + + "github.com/RigelBuild/compass/go/internal/store/db" ) // CoordinationHook is the manager-comms coordination-channel reconcile the comms @@ -104,11 +107,13 @@ func (s *Store) EnsureOwnerCoordinationGroupTx(ctx context.Context, tx pgx.Tx, o // discriminated SELECT would wrongly adopt. const coordinationGroupName = "__coordination__" - 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), coordinationGroupName, int32(VisibilityOwner), - ).Scan(&existing); { + qtx := db.New(tx) + existing, err := qtx.GetCoordinationGroup(ctx, db.GetCoordinationGroupParams{ + OwnerUserID: string(ownerUserID), + Name: coordinationGroupName, + Visibility: int16(VisibilityOwner), + }) + switch { case err == nil: return ChannelGroupID(existing), nil case !noRows(err): @@ -116,10 +121,12 @@ func (s *Store) EnsureOwnerCoordinationGroupTx(ctx context.Context, tx pgx.Tx, o } 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, coordinationGroupName, string(ownerUserID), int32(VisibilityOwner), - ); err != nil { + if err := qtx.InsertCoordinationGroup(ctx, db.InsertCoordinationGroupParams{ + ID: id, + Name: coordinationGroupName, + OwnerUserID: string(ownerUserID), + Visibility: int16(VisibilityOwner), + }); err != nil { return "", fmt.Errorf("store: insert coordination group: %w", err) } return ChannelGroupID(id), nil @@ -166,26 +173,24 @@ func (s *Store) UpsertCoordinationChannelTx(ctx context.Context, tx pgx.Tx, spec return "", fmt.Errorf("%w: coordination channel name is required", ErrInvalidArgument) } + qtx := db.New(tx) for suffix := 1; ; suffix++ { name := spec.BaseName if suffix > 1 { name = fmt.Sprintf("%s-%d", spec.BaseName, suffix) } - var ( - existingID string - existingOwner string - ) - switch err := tx.QueryRow(ctx, - `SELECT id, COALESCE(owner_account_id, '') FROM channels WHERE group_id = $1 AND name = $2`, - string(spec.GroupID), name, - ).Scan(&existingID, &existingOwner); { + existing, err := qtx.GetCoordinationChannelByName(ctx, db.GetCoordinationChannelByNameParams{ + GroupID: pgtype.Text{String: string(spec.GroupID), Valid: true}, + Name: name, + }) + switch { case err == nil: // A channel with this name already exists. Resume only when the // manager owns it; otherwise it is a user's channel we must never // adopt — advance to the next suffix. - if AccountID(existingOwner) == spec.OwnerAccountID { - return ChannelID(existingID), nil + if AccountID(existing.OwnerAccountID) == spec.OwnerAccountID { + return ChannelID(existing.ID), nil } continue case !noRows(err): @@ -201,16 +206,18 @@ func (s *Store) UpsertCoordinationChannelTx(ctx context.Context, tx pgx.Tx, spec // row. On that no-row case we loop back to the SELECT, which now sees the // concurrently-committed row and resumes-or-suffixes it. 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, NULLIF($6, ''), $7) `+ - `ON CONFLICT (group_id, name) WHERE group_id IS NOT NULL DO NOTHING `+ - `RETURNING id`, - id, name, string(spec.GroupID), int32(ChannelKindChannel), - int32(spec.Policy.PostPolicy), string(spec.Policy.OwnerAccountID), spec.Policy.MandatorySubscription, - ).Scan(&id); { + insertedID, err := qtx.InsertCoordinationChannel(ctx, db.InsertCoordinationChannelParams{ + ID: id, + Name: name, + GroupID: pgtype.Text{String: string(spec.GroupID), Valid: true}, + Kind: int16(ChannelKindChannel), + PostPolicy: int16(spec.Policy.PostPolicy), //nolint:gosec // G115: ChannelPostPolicy is a CHECK-constrained 0/1 enum (channels.post_policy), always within int16 + Column6: string(spec.Policy.OwnerAccountID), + MandatorySubscription: spec.Policy.MandatorySubscription, + }) + switch { case err == nil: - return ChannelID(id), nil + return ChannelID(insertedID), nil case noRows(err): // A concurrent writer won the (group, name) race between our SELECT // and this INSERT. Re-resolve the SAME name (undo the loop's suffix @@ -243,24 +250,15 @@ func (s *Store) SetCoordinationMembersTx(ctx context.Context, tx pgx.Tx, channel wantSet[m] = true } - rows, err := tx.Query(ctx, - `SELECT account_id FROM channel_members WHERE channel_id = $1`, string(channelID)) + qtx := db.New(tx) + memberIDs, err := qtx.ChannelMemberIDs(ctx, string(channelID)) if err != nil { return nil, fmt.Errorf("store: list coordination members: %w", err) } - current := make(map[AccountID]bool) - for rows.Next() { - var m string - if err := rows.Scan(&m); err != nil { - rows.Close() - return nil, fmt.Errorf("store: scan coordination member: %w", err) - } + current := make(map[AccountID]bool, len(memberIDs)) + for _, m := range memberIDs { current[AccountID(m)] = true } - rows.Close() - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate coordination members: %w", err) - } // Add every wanted member missing a row, seeding its delivery cursor in this // tx. seedDeliveryCursor is self-guarding (agent-only) and idempotent, so a @@ -270,11 +268,10 @@ func (s *Store) SetCoordinationMembersTx(ctx context.Context, tx pgx.Tx, channel if current[m] { continue } - 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 { + if err := qtx.EnsureChannelMember(ctx, db.EnsureChannelMemberParams{ + ChannelID: string(channelID), + AccountID: string(m), + }); err != nil { if pgErrIs(err, pgForeignKeyViolation) { return nil, fmt.Errorf("%w: unknown coordination member %q", ErrInvalidArgument, m) } @@ -293,10 +290,10 @@ func (s *Store) SetCoordinationMembersTx(ctx context.Context, tx pgx.Tx, channel if wantSet[m] { continue } - if _, err := tx.Exec(ctx, - `DELETE FROM channel_members WHERE channel_id = $1 AND account_id = $2`, - string(channelID), string(m), - ); err != nil { + if _, err := qtx.DeleteChannelMember(ctx, db.DeleteChannelMemberParams{ + ChannelID: string(channelID), + AccountID: string(m), + }); err != nil { return nil, fmt.Errorf("store: remove coordination member: %w", err) } removed = append(removed, m) @@ -312,24 +309,15 @@ func (s *Store) SetCoordinationMembersTx(ctx context.Context, tx pgx.Tx, channel // a stable set. Runs on the passed tx so it reads the tree state the parent-edge // write just committed within this same transaction. func (s *Store) CoordinationReports(ctx context.Context, tx pgx.Tx, managerAgentID AccountID) ([]AccountID, error) { - rows, err := tx.Query(ctx, - `SELECT account_id FROM agent_accounts WHERE parent_agent_id = $1 ORDER BY account_id`, - string(managerAgentID)) + reports, err := db.New(tx).CoordinationReports(ctx, pgtype.Text{String: string(managerAgentID), Valid: true}) if err != nil { return nil, fmt.Errorf("store: list coordination reports: %w", err) } - defer rows.Close() - members := []AccountID{managerAgentID} - for rows.Next() { - var m string - if err := rows.Scan(&m); err != nil { - return nil, fmt.Errorf("store: scan coordination report: %w", err) - } + members := make([]AccountID, 0, len(reports)+1) + members = append(members, managerAgentID) + for _, m := range reports { members = append(members, AccountID(m)) } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate coordination reports: %w", err) - } return members, nil } @@ -338,16 +326,10 @@ func (s *Store) CoordinationReports(ctx context.Context, tx pgx.Tx, managerAgent // name derives from the handle; the group from the owner). An id that names no // agent account is ErrNotFound. Runs on the passed tx (mid-parent-edge-write). func (s *Store) ResolveCoordinationManagerTx(ctx context.Context, tx pgx.Tx, managerAgentID AccountID) (handle string, ownerUserID AccountID, err error) { - var owner string - switch scanErr := tx.QueryRow(ctx, - `SELECT a.handle, ag.owner_user_id - FROM accounts a - JOIN agent_accounts ag ON ag.account_id = a.id - WHERE a.id = $1`, - string(managerAgentID), - ).Scan(&handle, &owner); { + row, scanErr := db.New(tx).ResolveCoordinationManager(ctx, string(managerAgentID)) + switch { case scanErr == nil: - return handle, AccountID(owner), nil + return row.Handle, AccountID(row.OwnerUserID), nil case noRows(scanErr): return "", "", fmt.Errorf("%w: coordination manager %q", ErrNotFound, managerAgentID) default: @@ -369,7 +351,7 @@ func LockOwnerCoordinationTx(ctx context.Context, tx pgx.Tx, ownerUserID Account // reconcile runs INSIDE a parent-edge write that may itself hold the tree // lock, and a distinct key avoids a self-deadlock-adjacent double-take while // still serializing coordination reconciles against each other. - if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext('coordination:' || $1))`, string(ownerUserID)); err != nil { + if err := db.New(tx).LockOwnerCoordination(ctx, pgtype.Text{String: string(ownerUserID), Valid: true}); err != nil { return fmt.Errorf("store: lock owner coordination: %w", err) } return nil diff --git a/go/internal/store/db/channel_pins.sql.go b/go/internal/store/db/channel_pins.sql.go new file mode 100644 index 000000000..ccd6e9aa8 --- /dev/null +++ b/go/internal/store/db/channel_pins.sql.go @@ -0,0 +1,157 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: channel_pins.sql + +package db + +import ( + "context" +) + +const countChannelPins = `-- name: CountChannelPins :one +SELECT count(*) AS count, COALESCE(MAX(position), -1) + 1 AS next_position +FROM channel_pins WHERE channel_id = $1 +` + +type CountChannelPinsRow struct { + Count int64 + NextPosition int32 +} + +func (q *Queries) CountChannelPins(ctx context.Context, channelID string) (CountChannelPinsRow, error) { + row := q.db.QueryRow(ctx, countChannelPins, channelID) + var i CountChannelPinsRow + err := row.Scan(&i.Count, &i.NextPosition) + return i, err +} + +const deleteChannelPin = `-- name: DeleteChannelPin :exec +DELETE FROM channel_pins WHERE channel_id = $1 AND message_id = $2 +` + +type DeleteChannelPinParams struct { + ChannelID string + MessageID string +} + +func (q *Queries) DeleteChannelPin(ctx context.Context, arg DeleteChannelPinParams) error { + _, err := q.db.Exec(ctx, deleteChannelPin, arg.ChannelID, arg.MessageID) + return err +} + +const deleteChannelPinReturningPosition = `-- name: DeleteChannelPinReturningPosition :one +DELETE FROM channel_pins WHERE channel_id = $1 AND message_id = $2 RETURNING position +` + +type DeleteChannelPinReturningPositionParams struct { + ChannelID string + MessageID string +} + +func (q *Queries) DeleteChannelPinReturningPosition(ctx context.Context, arg DeleteChannelPinReturningPositionParams) (int32, error) { + row := q.db.QueryRow(ctx, deleteChannelPinReturningPosition, arg.ChannelID, arg.MessageID) + var position int32 + err := row.Scan(&position) + return position, err +} + +const insertChannelPin = `-- name: InsertChannelPin :exec +INSERT INTO channel_pins (channel_id, message_id, position, pinned_at_unix_ms, pinned_by_account_id) +VALUES ($1, $2, $3, $4, $5) +` + +type InsertChannelPinParams struct { + ChannelID string + MessageID string + Position int32 + PinnedAtUnixMs int64 + PinnedByAccountID string +} + +func (q *Queries) InsertChannelPin(ctx context.Context, arg InsertChannelPinParams) error { + _, err := q.db.Exec(ctx, insertChannelPin, + arg.ChannelID, + arg.MessageID, + arg.Position, + arg.PinnedAtUnixMs, + arg.PinnedByAccountID, + ) + return err +} + +const lockChannelForPins = `-- name: LockChannelForPins :one + +SELECT post_policy, COALESCE(owner_account_id, '') AS owner_account_id +FROM channels WHERE id = $1 FOR UPDATE +` + +type LockChannelForPinsRow struct { + PostPolicy int16 + OwnerAccountID string +} + +// Channel-pins (pinned board) queries (sqlc adoption T3, RIG-3034). These +// replace the inline SQL literals in internal/store/channel_pins.go; the +// hand-written Store methods and the in-tx FOR UPDATE lock / cap-check control +// flow stay exactly as they were and wrap these generated calls. +func (q *Queries) LockChannelForPins(ctx context.Context, id string) (LockChannelForPinsRow, error) { + row := q.db.QueryRow(ctx, lockChannelForPins, id) + var i LockChannelForPinsRow + err := row.Scan(&i.PostPolicy, &i.OwnerAccountID) + return i, err +} + +const messageInChannel = `-- name: MessageInChannel :one +SELECT 1 FROM messages m JOIN topics t ON t.id = m.topic_id +WHERE m.id = $1 AND t.channel_id = $2 +` + +type MessageInChannelParams struct { + ID string + ChannelID string +} + +func (q *Queries) MessageInChannel(ctx context.Context, arg MessageInChannelParams) (int32, error) { + row := q.db.QueryRow(ctx, messageInChannel, arg.ID, arg.ChannelID) + var column_1 int32 + err := row.Scan(&column_1) + return column_1, err +} + +const pinnedEntries = `-- name: PinnedEntries :many +SELECT message_id, position, pinned_at_unix_ms, pinned_by_account_id +FROM channel_pins WHERE channel_id = $1 ORDER BY position +` + +type PinnedEntriesRow struct { + MessageID string + Position int32 + PinnedAtUnixMs int64 + PinnedByAccountID string +} + +func (q *Queries) PinnedEntries(ctx context.Context, channelID string) ([]PinnedEntriesRow, error) { + rows, err := q.db.Query(ctx, pinnedEntries, channelID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []PinnedEntriesRow + for rows.Next() { + var i PinnedEntriesRow + if err := rows.Scan( + &i.MessageID, + &i.Position, + &i.PinnedAtUnixMs, + &i.PinnedByAccountID, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} diff --git a/go/internal/store/db/channels.sql.go b/go/internal/store/db/channels.sql.go new file mode 100644 index 000000000..e6419844e --- /dev/null +++ b/go/internal/store/db/channels.sql.go @@ -0,0 +1,640 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: channels.sql + +package db + +import ( + "context" +) + +const agentOwnersByIDs = `-- name: AgentOwnersByIDs :many +SELECT owner_user_id FROM agent_accounts WHERE account_id = ANY($1::text[]) +` + +func (q *Queries) AgentOwnersByIDs(ctx context.Context, dollar_1 []string) ([]string, error) { + rows, err := q.db.Query(ctx, agentOwnersByIDs, dollar_1) + if err != nil { + return nil, err + } + defer rows.Close() + var items []string + for rows.Next() { + var owner_user_id string + if err := rows.Scan(&owner_user_id); err != nil { + return nil, err + } + items = append(items, owner_user_id) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const channelGroupVisibleTo = `-- name: ChannelGroupVisibleTo :one +WITH RECURSIVE ancestry AS ( + SELECT id, parent_group_id, visibility AS min_vis + FROM channel_groups + UNION ALL + SELECT a.id, g.parent_group_id, LEAST(a.min_vis, g.visibility) + FROM ancestry a + JOIN channel_groups g ON g.id = a.parent_group_id +), +effective AS ( + SELECT id, MIN(min_vis) AS eff_vis + FROM ancestry + GROUP BY id +), +viewer AS ( + SELECT owner_user_id AS uid FROM agent_accounts WHERE account_id = $1 + UNION ALL + SELECT $1 AS uid +) +SELECT EXISTS ( + SELECT 1 FROM channel_groups g + JOIN effective e ON e.id = g.id + WHERE g.id = $2 AND (e.eff_vis = 1 OR g.owner_user_id IN (SELECT uid FROM viewer)) +) +` + +type ChannelGroupVisibleToParams struct { + AccountID string + ID string +} + +func (q *Queries) ChannelGroupVisibleTo(ctx context.Context, arg ChannelGroupVisibleToParams) (bool, error) { + row := q.db.QueryRow(ctx, channelGroupVisibleTo, arg.AccountID, arg.ID) + var exists bool + err := row.Scan(&exists) + return exists, err +} + +const channelMemberExists = `-- name: ChannelMemberExists :one +SELECT EXISTS (SELECT 1 FROM channel_members WHERE channel_id = $1 AND account_id = $2) +` + +type ChannelMemberExistsParams struct { + ChannelID string + AccountID string +} + +func (q *Queries) ChannelMemberExists(ctx context.Context, arg ChannelMemberExistsParams) (bool, error) { + row := q.db.QueryRow(ctx, channelMemberExists, arg.ChannelID, arg.AccountID) + var exists bool + err := row.Scan(&exists) + return exists, err +} + +const channelMembersByChannelIDs = `-- name: ChannelMembersByChannelIDs :many +SELECT channel_id, account_id, subscribed +FROM channel_members +WHERE channel_id = ANY($1::text[]) +ORDER BY account_id +` + +func (q *Queries) ChannelMembersByChannelIDs(ctx context.Context, dollar_1 []string) ([]ChannelMember, error) { + rows, err := q.db.Query(ctx, channelMembersByChannelIDs, dollar_1) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ChannelMember + for rows.Next() { + var i ChannelMember + if err := rows.Scan(&i.ChannelID, &i.AccountID, &i.Subscribed); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const channelVisibleTo = `-- name: ChannelVisibleTo :one +WITH RECURSIVE ancestry AS ( + SELECT id, parent_group_id, visibility AS min_vis + FROM channel_groups + UNION ALL + SELECT a.id, g.parent_group_id, LEAST(a.min_vis, g.visibility) + FROM ancestry a + JOIN channel_groups g ON g.id = a.parent_group_id +), +effective AS ( + SELECT id, MIN(min_vis) AS eff_vis + FROM ancestry + GROUP BY id +) +SELECT EXISTS ( + SELECT 1 FROM channels c + WHERE c.id = $2 AND ( + EXISTS ( + SELECT 1 FROM channel_members cm + WHERE cm.channel_id = c.id AND cm.account_id = $1 + ) + OR ( + c.kind = 0 AND c.group_id IS NOT NULL AND EXISTS ( + SELECT 1 FROM effective e WHERE e.id = c.group_id AND e.eff_vis = 1 + ) + ) + ) +) +` + +type ChannelVisibleToParams struct { + AccountID string + ID string +} + +func (q *Queries) ChannelVisibleTo(ctx context.Context, arg ChannelVisibleToParams) (bool, error) { + row := q.db.QueryRow(ctx, channelVisibleTo, arg.AccountID, arg.ID) + var exists bool + err := row.Scan(&exists) + return exists, err +} + +const channelsByNameForViewer = `-- name: ChannelsByNameForViewer :many +WITH RECURSIVE ancestry AS ( + SELECT id, parent_group_id, visibility AS min_vis + FROM channel_groups + UNION ALL + SELECT a.id, g.parent_group_id, LEAST(a.min_vis, g.visibility) + FROM ancestry a + JOIN channel_groups g ON g.id = a.parent_group_id +), +effective AS ( + SELECT id, MIN(min_vis) AS eff_vis + FROM ancestry + GROUP BY id +) +SELECT c.id, c.name, COALESCE(c.group_id, '') AS group_id, c.kind, c.post_policy, + COALESCE(c.owner_account_id, '') AS owner_account_id, c.mandatory_subscription +FROM channels c +WHERE c.name = $2 AND ( + EXISTS ( + SELECT 1 FROM channel_members cm + WHERE cm.channel_id = c.id AND cm.account_id = $1 + ) + OR ( + c.kind = 0 AND c.group_id IS NOT NULL AND EXISTS ( + SELECT 1 FROM effective e WHERE e.id = c.group_id AND e.eff_vis = 1 + ) + ) + ) +ORDER BY c.id +` + +type ChannelsByNameForViewerParams struct { + AccountID string + Name string +} + +type ChannelsByNameForViewerRow struct { + ID string + Name string + GroupID string + Kind int16 + PostPolicy int16 + OwnerAccountID string + MandatorySubscription bool +} + +func (q *Queries) ChannelsByNameForViewer(ctx context.Context, arg ChannelsByNameForViewerParams) ([]ChannelsByNameForViewerRow, error) { + rows, err := q.db.Query(ctx, channelsByNameForViewer, arg.AccountID, arg.Name) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ChannelsByNameForViewerRow + for rows.Next() { + var i ChannelsByNameForViewerRow + if err := rows.Scan( + &i.ID, + &i.Name, + &i.GroupID, + &i.Kind, + &i.PostPolicy, + &i.OwnerAccountID, + &i.MandatorySubscription, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const convertDMChannel = `-- name: ConvertDMChannel :exec +UPDATE channels SET kind = $1, name = $2, group_id = NULL, mandatory_subscription = FALSE WHERE id = $3 +` + +type ConvertDMChannelParams struct { + Kind int16 + Name string + ID string +} + +func (q *Queries) ConvertDMChannel(ctx context.Context, arg ConvertDMChannelParams) error { + _, err := q.db.Exec(ctx, convertDMChannel, arg.Kind, arg.Name, arg.ID) + return err +} + +const countAgentMembers = `-- name: CountAgentMembers :one +SELECT COUNT(*) FROM channel_members cm +JOIN agent_accounts aa ON aa.account_id = cm.account_id +WHERE cm.channel_id = $1 +` + +func (q *Queries) CountAgentMembers(ctx context.Context, channelID string) (int64, error) { + row := q.db.QueryRow(ctx, countAgentMembers, channelID) + var count int64 + err := row.Scan(&count) + return count, err +} + +const deleteChannelMember = `-- name: DeleteChannelMember :execrows +DELETE FROM channel_members WHERE channel_id = $1 AND account_id = $2 +` + +type DeleteChannelMemberParams struct { + ChannelID string + AccountID string +} + +func (q *Queries) DeleteChannelMember(ctx context.Context, arg DeleteChannelMemberParams) (int64, error) { + result, err := q.db.Exec(ctx, deleteChannelMember, arg.ChannelID, arg.AccountID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const getAgentWorkspaceID = `-- name: GetAgentWorkspaceID :one +SELECT id FROM agent_workspaces WHERE agent_account_id = $1 +` + +func (q *Queries) GetAgentWorkspaceID(ctx context.Context, agentAccountID string) (string, error) { + row := q.db.QueryRow(ctx, getAgentWorkspaceID, agentAccountID) + var id string + err := row.Scan(&id) + return id, err +} + +const getChannel = `-- name: GetChannel :one +SELECT id, name, COALESCE(group_id, '') AS group_id, kind, post_policy, + COALESCE(owner_account_id, '') AS owner_account_id, mandatory_subscription +FROM channels WHERE id = $1 +` + +type GetChannelRow struct { + ID string + Name string + GroupID string + Kind int16 + PostPolicy int16 + OwnerAccountID string + MandatorySubscription bool +} + +func (q *Queries) GetChannel(ctx context.Context, id string) (GetChannelRow, error) { + row := q.db.QueryRow(ctx, getChannel, id) + var i GetChannelRow + err := row.Scan( + &i.ID, + &i.Name, + &i.GroupID, + &i.Kind, + &i.PostPolicy, + &i.OwnerAccountID, + &i.MandatorySubscription, + ) + return i, err +} + +const getChannelGroupVisibility = `-- name: GetChannelGroupVisibility :one +SELECT visibility FROM channel_groups WHERE id = $1 +` + +func (q *Queries) GetChannelGroupVisibility(ctx context.Context, id string) (int16, error) { + row := q.db.QueryRow(ctx, getChannelGroupVisibility, id) + var visibility int16 + err := row.Scan(&visibility) + return visibility, err +} + +const insertAgentWorkspaceIgnore = `-- name: InsertAgentWorkspaceIgnore :exec +INSERT INTO agent_workspaces (id, agent_account_id) +VALUES ($1, $2) +ON CONFLICT (agent_account_id) DO NOTHING +` + +type InsertAgentWorkspaceIgnoreParams struct { + ID string + AgentAccountID string +} + +func (q *Queries) InsertAgentWorkspaceIgnore(ctx context.Context, arg InsertAgentWorkspaceIgnoreParams) error { + _, err := q.db.Exec(ctx, insertAgentWorkspaceIgnore, arg.ID, arg.AgentAccountID) + return err +} + +const insertChannel = `-- name: InsertChannel :exec +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) +` + +type InsertChannelParams struct { + ID string + Name string + Column3 interface{} + Kind int16 + PostPolicy int16 + Column6 interface{} + MandatorySubscription bool +} + +func (q *Queries) InsertChannel(ctx context.Context, arg InsertChannelParams) error { + _, err := q.db.Exec(ctx, insertChannel, + arg.ID, + arg.Name, + arg.Column3, + arg.Kind, + arg.PostPolicy, + arg.Column6, + arg.MandatorySubscription, + ) + return err +} + +const insertChannelGroup = `-- name: InsertChannelGroup :exec + +INSERT INTO channel_groups (id, name, parent_group_id, owner_user_id, visibility) +VALUES ($1, $2, NULLIF($3, ''), $4, $5) +` + +type InsertChannelGroupParams struct { + ID string + Name string + Column3 interface{} + OwnerUserID string + Visibility int16 +} + +// Channel-domain queries (sqlc adoption T3, RIG-3034). These replace the inline +// SQL literals that lived in internal/store/channels.go; the hand-written Store +// methods keep their exact signatures and wrap these generated calls, mapping +// the generated row structs back to the domain Channel / ChannelGroup. +// +// The channel-visibility CTE (effective(id, eff_vis)) and the channel/group +// visibility predicates are repeated per read because sqlc has no query-fragment +// composition — they replace the former channels.go const fragments +// (effectiveVisibilityCTE / viewerCTE / channelVisiblePredicate / +// groupVisiblePredicate). The copies MUST stay textually identical so the stream +// edge's single-id visibility check cannot drift from the list read (the +// anti-drift guarantee the frozen record requires). +func (q *Queries) InsertChannelGroup(ctx context.Context, arg InsertChannelGroupParams) error { + _, err := q.db.Exec(ctx, insertChannelGroup, + arg.ID, + arg.Name, + arg.Column3, + arg.OwnerUserID, + arg.Visibility, + ) + return err +} + +const listChannelGroups = `-- name: ListChannelGroups :many +WITH RECURSIVE ancestry AS ( + SELECT id, parent_group_id, visibility AS min_vis + FROM channel_groups + UNION ALL + SELECT a.id, g.parent_group_id, LEAST(a.min_vis, g.visibility) + FROM ancestry a + JOIN channel_groups g ON g.id = a.parent_group_id +), +effective AS ( + SELECT id, MIN(min_vis) AS eff_vis + FROM ancestry + GROUP BY id +), +viewer AS ( + SELECT owner_user_id AS uid FROM agent_accounts WHERE account_id = $1 + UNION ALL + SELECT $1 AS uid +) +SELECT g.id, g.name, COALESCE(g.parent_group_id, '') AS parent_group_id, g.owner_user_id, g.visibility +FROM channel_groups g +JOIN effective e ON e.id = g.id +WHERE (e.eff_vis = 1 OR g.owner_user_id IN (SELECT uid FROM viewer)) +ORDER BY g.name +` + +type ListChannelGroupsRow struct { + ID string + Name string + ParentGroupID string + OwnerUserID string + Visibility int16 +} + +func (q *Queries) ListChannelGroups(ctx context.Context, accountID string) ([]ListChannelGroupsRow, error) { + rows, err := q.db.Query(ctx, listChannelGroups, accountID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListChannelGroupsRow + for rows.Next() { + var i ListChannelGroupsRow + if err := rows.Scan( + &i.ID, + &i.Name, + &i.ParentGroupID, + &i.OwnerUserID, + &i.Visibility, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listChannels = `-- name: ListChannels :many +WITH RECURSIVE ancestry AS ( + SELECT id, parent_group_id, visibility AS min_vis + FROM channel_groups + UNION ALL + SELECT a.id, g.parent_group_id, LEAST(a.min_vis, g.visibility) + FROM ancestry a + JOIN channel_groups g ON g.id = a.parent_group_id +), +effective AS ( + SELECT id, MIN(min_vis) AS eff_vis + FROM ancestry + GROUP BY id +) +SELECT c.id, c.name, COALESCE(c.group_id, '') AS group_id, c.kind, c.post_policy, + COALESCE(c.owner_account_id, '') AS owner_account_id, c.mandatory_subscription +FROM channels c +WHERE ( + EXISTS ( + SELECT 1 FROM channel_members cm + WHERE cm.channel_id = c.id AND cm.account_id = $1 + ) + OR ( + c.kind = 0 AND c.group_id IS NOT NULL AND EXISTS ( + SELECT 1 FROM effective e WHERE e.id = c.group_id AND e.eff_vis = 1 + ) + ) + ) +ORDER BY c.name +` + +type ListChannelsRow struct { + ID string + Name string + GroupID string + Kind int16 + PostPolicy int16 + OwnerAccountID string + MandatorySubscription bool +} + +func (q *Queries) ListChannels(ctx context.Context, accountID string) ([]ListChannelsRow, error) { + rows, err := q.db.Query(ctx, listChannels, accountID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListChannelsRow + for rows.Next() { + var i ListChannelsRow + if err := rows.Scan( + &i.ID, + &i.Name, + &i.GroupID, + &i.Kind, + &i.PostPolicy, + &i.OwnerAccountID, + &i.MandatorySubscription, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const lockChannelMandatoryKind = `-- name: LockChannelMandatoryKind :one +SELECT mandatory_subscription, kind FROM channels WHERE id = $1 FOR UPDATE +` + +type LockChannelMandatoryKindRow struct { + MandatorySubscription bool + Kind int16 +} + +func (q *Queries) LockChannelMandatoryKind(ctx context.Context, id string) (LockChannelMandatoryKindRow, error) { + row := q.db.QueryRow(ctx, lockChannelMandatoryKind, id) + var i LockChannelMandatoryKindRow + err := row.Scan(&i.MandatorySubscription, &i.Kind) + return i, err +} + +const lockChannelPolicy = `-- name: LockChannelPolicy :one +SELECT mandatory_subscription, COALESCE(owner_account_id, '') AS owner_account_id +FROM channels WHERE id = $1 FOR UPDATE +` + +type LockChannelPolicyRow struct { + MandatorySubscription bool + OwnerAccountID string +} + +func (q *Queries) LockChannelPolicy(ctx context.Context, id string) (LockChannelPolicyRow, error) { + row := q.db.QueryRow(ctx, lockChannelPolicy, id) + var i LockChannelPolicyRow + err := row.Scan(&i.MandatorySubscription, &i.OwnerAccountID) + return i, err +} + +const ownerHasPresentAgent = `-- name: OwnerHasPresentAgent :one +SELECT EXISTS ( + SELECT 1 FROM agent_accounts aa + JOIN channel_members cm ON cm.account_id = aa.account_id + WHERE aa.owner_user_id = $1 AND cm.channel_id = $2 AND aa.account_id <> $1 +) +` + +type OwnerHasPresentAgentParams struct { + OwnerUserID string + ChannelID string +} + +func (q *Queries) OwnerHasPresentAgent(ctx context.Context, arg OwnerHasPresentAgentParams) (bool, error) { + row := q.db.QueryRow(ctx, ownerHasPresentAgent, arg.OwnerUserID, arg.ChannelID) + var exists bool + err := row.Scan(&exists) + return exists, err +} + +const subscribeConvertedDMParties = `-- name: SubscribeConvertedDMParties :exec +UPDATE channel_members cm SET subscribed = TRUE +FROM agent_accounts aa WHERE aa.account_id = cm.account_id AND cm.channel_id = $1 +` + +func (q *Queries) SubscribeConvertedDMParties(ctx context.Context, channelID string) error { + _, err := q.db.Exec(ctx, subscribeConvertedDMParties, channelID) + return err +} + +const updateChannelPolicy = `-- name: UpdateChannelPolicy :exec +UPDATE channels SET post_policy = $2, owner_account_id = NULLIF($3, ''), mandatory_subscription = $4 WHERE id = $1 +` + +type UpdateChannelPolicyParams struct { + ID string + PostPolicy int16 + Column3 interface{} + MandatorySubscription bool +} + +func (q *Queries) UpdateChannelPolicy(ctx context.Context, arg UpdateChannelPolicyParams) error { + _, err := q.db.Exec(ctx, updateChannelPolicy, + arg.ID, + arg.PostPolicy, + arg.Column3, + arg.MandatorySubscription, + ) + return err +} + +const upsertChannelMember = `-- name: UpsertChannelMember :exec +INSERT INTO channel_members (channel_id, account_id, subscribed) +VALUES ($1, $2, $3) +ON CONFLICT (channel_id, account_id) DO UPDATE SET subscribed = EXCLUDED.subscribed +` + +type UpsertChannelMemberParams struct { + ChannelID string + AccountID string + Subscribed bool +} + +func (q *Queries) UpsertChannelMember(ctx context.Context, arg UpsertChannelMemberParams) error { + _, err := q.db.Exec(ctx, upsertChannelMember, arg.ChannelID, arg.AccountID, arg.Subscribed) + return err +} diff --git a/go/internal/store/db/coordination.sql.go b/go/internal/store/db/coordination.sql.go new file mode 100644 index 000000000..35f4d1a99 --- /dev/null +++ b/go/internal/store/db/coordination.sql.go @@ -0,0 +1,189 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: coordination.sql + +package db + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const channelMemberIDs = `-- name: ChannelMemberIDs :many +SELECT account_id FROM channel_members WHERE channel_id = $1 +` + +func (q *Queries) ChannelMemberIDs(ctx context.Context, channelID string) ([]string, error) { + rows, err := q.db.Query(ctx, channelMemberIDs, channelID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []string + for rows.Next() { + var account_id string + if err := rows.Scan(&account_id); err != nil { + return nil, err + } + items = append(items, account_id) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const coordinationReports = `-- name: CoordinationReports :many +SELECT account_id FROM agent_accounts WHERE parent_agent_id = $1 ORDER BY account_id +` + +func (q *Queries) CoordinationReports(ctx context.Context, parentAgentID pgtype.Text) ([]string, error) { + rows, err := q.db.Query(ctx, coordinationReports, parentAgentID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []string + for rows.Next() { + var account_id string + if err := rows.Scan(&account_id); err != nil { + return nil, err + } + items = append(items, account_id) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getCoordinationChannelByName = `-- name: GetCoordinationChannelByName :one +SELECT id, COALESCE(owner_account_id, '') AS owner_account_id +FROM channels WHERE group_id = $1 AND name = $2 +` + +type GetCoordinationChannelByNameParams struct { + GroupID pgtype.Text + Name string +} + +type GetCoordinationChannelByNameRow struct { + ID string + OwnerAccountID string +} + +func (q *Queries) GetCoordinationChannelByName(ctx context.Context, arg GetCoordinationChannelByNameParams) (GetCoordinationChannelByNameRow, error) { + row := q.db.QueryRow(ctx, getCoordinationChannelByName, arg.GroupID, arg.Name) + var i GetCoordinationChannelByNameRow + err := row.Scan(&i.ID, &i.OwnerAccountID) + return i, err +} + +const getCoordinationGroup = `-- name: GetCoordinationGroup :one + +SELECT id FROM channel_groups +WHERE owner_user_id = $1 AND name = $2 AND parent_group_id IS NULL AND visibility = $3 +` + +type GetCoordinationGroupParams struct { + OwnerUserID string + Name string + Visibility int16 +} + +// Coordination-store queries (sqlc adoption T3, RIG-3034). These replace the +// inline SQL literals in internal/store/coordination.go; the hand-written Store +// methods keep their signatures, the per-owner advisory-lock discipline, the +// suffix-search resolution loop, and the WithTx seam (which stays hand-written). +// The member INSERT/DELETE reuse EnsureChannelMember (accounts.sql) and +// DeleteChannelMember (channels.sql) — the statements are identical. +func (q *Queries) GetCoordinationGroup(ctx context.Context, arg GetCoordinationGroupParams) (string, error) { + row := q.db.QueryRow(ctx, getCoordinationGroup, arg.OwnerUserID, arg.Name, arg.Visibility) + var id string + err := row.Scan(&id) + return id, err +} + +const insertCoordinationChannel = `-- name: InsertCoordinationChannel :one +INSERT INTO channels (id, name, group_id, kind, post_policy, owner_account_id, mandatory_subscription) +VALUES ($1, $2, $3, $4, $5, NULLIF($6, ''), $7) +ON CONFLICT (group_id, name) WHERE group_id IS NOT NULL DO NOTHING +RETURNING id +` + +type InsertCoordinationChannelParams struct { + ID string + Name string + GroupID pgtype.Text + Kind int16 + PostPolicy int16 + Column6 interface{} + MandatorySubscription bool +} + +func (q *Queries) InsertCoordinationChannel(ctx context.Context, arg InsertCoordinationChannelParams) (string, error) { + row := q.db.QueryRow(ctx, insertCoordinationChannel, + arg.ID, + arg.Name, + arg.GroupID, + arg.Kind, + arg.PostPolicy, + arg.Column6, + arg.MandatorySubscription, + ) + var id string + err := row.Scan(&id) + return id, err +} + +const insertCoordinationGroup = `-- name: InsertCoordinationGroup :exec +INSERT INTO channel_groups (id, name, parent_group_id, owner_user_id, visibility) +VALUES ($1, $2, NULL, $3, $4) +` + +type InsertCoordinationGroupParams struct { + ID string + Name string + OwnerUserID string + Visibility int16 +} + +func (q *Queries) InsertCoordinationGroup(ctx context.Context, arg InsertCoordinationGroupParams) error { + _, err := q.db.Exec(ctx, insertCoordinationGroup, + arg.ID, + arg.Name, + arg.OwnerUserID, + arg.Visibility, + ) + return err +} + +const lockOwnerCoordination = `-- name: LockOwnerCoordination :exec +SELECT pg_advisory_xact_lock(hashtext('coordination:' || $1)) +` + +func (q *Queries) LockOwnerCoordination(ctx context.Context, dollar_1 pgtype.Text) error { + _, err := q.db.Exec(ctx, lockOwnerCoordination, dollar_1) + return err +} + +const resolveCoordinationManager = `-- name: ResolveCoordinationManager :one +SELECT a.handle, ag.owner_user_id +FROM accounts a +JOIN agent_accounts ag ON ag.account_id = a.id +WHERE a.id = $1 +` + +type ResolveCoordinationManagerRow struct { + Handle string + OwnerUserID string +} + +func (q *Queries) ResolveCoordinationManager(ctx context.Context, id string) (ResolveCoordinationManagerRow, error) { + row := q.db.QueryRow(ctx, resolveCoordinationManager, id) + var i ResolveCoordinationManagerRow + err := row.Scan(&i.Handle, &i.OwnerUserID) + return i, err +} diff --git a/go/internal/store/db/querier.go b/go/internal/store/db/querier.go index c6bf14dc3..f60f0e3c4 100644 --- a/go/internal/store/db/querier.go +++ b/go/internal/store/db/querier.go @@ -25,15 +25,40 @@ type Querier interface { // Agent subtype set. The two `role` columns are aliased (user_role / agent_role) // so the generated row fields do not collide. AgentNeighborhood(ctx context.Context, id string) ([]AgentNeighborhoodRow, error) + AgentOwnersByIDs(ctx context.Context, dollar_1 []string) ([]string, error) AgentSubtree(ctx context.Context, accountID string) ([]AgentSubtreeRow, error) AgentsByOwner(ctx context.Context, ownerUserID string) ([]AgentsByOwnerRow, error) + ChannelGroupVisibleTo(ctx context.Context, arg ChannelGroupVisibleToParams) (bool, error) + ChannelMemberExists(ctx context.Context, arg ChannelMemberExistsParams) (bool, error) + ChannelMemberIDs(ctx context.Context, channelID string) ([]string, error) + ChannelMembersByChannelIDs(ctx context.Context, dollar_1 []string) ([]ChannelMember, error) + ChannelVisibleTo(ctx context.Context, arg ChannelVisibleToParams) (bool, error) + ChannelsByNameForViewer(ctx context.Context, arg ChannelsByNameForViewerParams) ([]ChannelsByNameForViewerRow, error) + ConvertDMChannel(ctx context.Context, arg ConvertDMChannelParams) error + CoordinationReports(ctx context.Context, parentAgentID pgtype.Text) ([]string, error) + CountAgentMembers(ctx context.Context, channelID string) (int64, error) + CountChannelPins(ctx context.Context, channelID string) (CountChannelPinsRow, error) CountRootAgents(ctx context.Context, ownerUserID string) (int64, error) + DeleteChannelMember(ctx context.Context, arg DeleteChannelMemberParams) (int64, error) + DeleteChannelPin(ctx context.Context, arg DeleteChannelPinParams) error + DeleteChannelPinReturningPosition(ctx context.Context, arg DeleteChannelPinReturningPositionParams) (int32, error) EnsureChannelMember(ctx context.Context, arg EnsureChannelMemberParams) error GetAccount(ctx context.Context, id string) (GetAccountRow, error) GetAccountByGlobalHandle(ctx context.Context, handle string) (GetAccountByGlobalHandleRow, error) GetAccountByOwnerHandle(ctx context.Context, arg GetAccountByOwnerHandleParams) (GetAccountByOwnerHandleRow, error) GetAgentOwner(ctx context.Context, accountID string) (string, error) GetAgentParent(ctx context.Context, accountID string) (pgtype.Text, error) + GetAgentWorkspaceID(ctx context.Context, agentAccountID string) (string, error) + GetChannel(ctx context.Context, id string) (GetChannelRow, error) + GetChannelGroupVisibility(ctx context.Context, id string) (int16, error) + GetCoordinationChannelByName(ctx context.Context, arg GetCoordinationChannelByNameParams) (GetCoordinationChannelByNameRow, error) + // Coordination-store queries (sqlc adoption T3, RIG-3034). These replace the + // inline SQL literals in internal/store/coordination.go; the hand-written Store + // methods keep their signatures, the per-owner advisory-lock discipline, the + // suffix-search resolution loop, and the WithTx seam (which stays hand-written). + // The member INSERT/DELETE reuse EnsureChannelMember (accounts.sql) and + // DeleteChannelMember (channels.sql) — the statements are identical. + GetCoordinationGroup(ctx context.Context, arg GetCoordinationGroupParams) (string, error) GetGlobalHandleID(ctx context.Context, handle string) (string, error) GetVisibleAgentHandleID(ctx context.Context, arg GetVisibleAgentHandleIDParams) (string, error) GetVisibleGlobalHandleID(ctx context.Context, arg GetVisibleGlobalHandleIDParams) (string, error) @@ -54,10 +79,42 @@ type Querier interface { InsertAccount(ctx context.Context, arg InsertAccountParams) error InsertAccountHandle(ctx context.Context, arg InsertAccountHandleParams) error InsertAgentAccount(ctx context.Context, arg InsertAgentAccountParams) error + InsertAgentWorkspaceIgnore(ctx context.Context, arg InsertAgentWorkspaceIgnoreParams) error + InsertChannel(ctx context.Context, arg InsertChannelParams) error + // Channel-domain queries (sqlc adoption T3, RIG-3034). These replace the inline + // SQL literals that lived in internal/store/channels.go; the hand-written Store + // methods keep their exact signatures and wrap these generated calls, mapping + // the generated row structs back to the domain Channel / ChannelGroup. + // + // The channel-visibility CTE (effective(id, eff_vis)) and the channel/group + // visibility predicates are repeated per read because sqlc has no query-fragment + // composition — they replace the former channels.go const fragments + // (effectiveVisibilityCTE / viewerCTE / channelVisiblePredicate / + // groupVisiblePredicate). The copies MUST stay textually identical so the stream + // edge's single-id visibility check cannot drift from the list read (the + // anti-drift guarantee the frozen record requires). + InsertChannelGroup(ctx context.Context, arg InsertChannelGroupParams) error + InsertChannelPin(ctx context.Context, arg InsertChannelPinParams) error + InsertCoordinationChannel(ctx context.Context, arg InsertCoordinationChannelParams) (string, error) + InsertCoordinationGroup(ctx context.Context, arg InsertCoordinationGroupParams) error InsertHomeChannel(ctx context.Context, arg InsertHomeChannelParams) error InsertSystemAccount(ctx context.Context, accountID string) error InsertUserAccount(ctx context.Context, arg InsertUserAccountParams) error + ListChannelGroups(ctx context.Context, accountID string) ([]ListChannelGroupsRow, error) + ListChannels(ctx context.Context, accountID string) ([]ListChannelsRow, error) ListVisibleAccounts(ctx context.Context, id string) ([]ListVisibleAccountsRow, error) + // Channel-pins (pinned board) queries (sqlc adoption T3, RIG-3034). These + // replace the inline SQL literals in internal/store/channel_pins.go; the + // hand-written Store methods and the in-tx FOR UPDATE lock / cap-check control + // flow stay exactly as they were and wrap these generated calls. + LockChannelForPins(ctx context.Context, id string) (LockChannelForPinsRow, error) + LockChannelMandatoryKind(ctx context.Context, id string) (LockChannelMandatoryKindRow, error) + LockChannelPolicy(ctx context.Context, id string) (LockChannelPolicyRow, error) + LockOwnerCoordination(ctx context.Context, dollar_1 pgtype.Text) error + MessageInChannel(ctx context.Context, arg MessageInChannelParams) (int32, error) + OwnerHasPresentAgent(ctx context.Context, arg OwnerHasPresentAgentParams) (bool, error) + PinnedEntries(ctx context.Context, channelID string) ([]PinnedEntriesRow, error) + ResolveCoordinationManager(ctx context.Context, id string) (ResolveCoordinationManagerRow, error) ResolveOwner(ctx context.Context, accountID string) (string, error) // Scaffold-only query proving sqlc generation works end to end (T1). // @@ -68,7 +125,10 @@ type Querier interface { // (tenants, 0001_init.sql), so sqlc compiles it against the real schema. ScaffoldGetTenant(ctx context.Context, id string) (Tenant, error) SeedHomeChannelMembers(ctx context.Context, arg SeedHomeChannelMembersParams) error + SubscribeConvertedDMParties(ctx context.Context, channelID string) error UpdateAgentParent(ctx context.Context, arg UpdateAgentParentParams) error + UpdateChannelPolicy(ctx context.Context, arg UpdateChannelPolicyParams) error + UpsertChannelMember(ctx context.Context, arg UpsertChannelMemberParams) error } var _ Querier = (*Queries)(nil) diff --git a/go/internal/store/queries/channel_pins.sql b/go/internal/store/queries/channel_pins.sql new file mode 100644 index 000000000..67a99c93f --- /dev/null +++ b/go/internal/store/queries/channel_pins.sql @@ -0,0 +1,30 @@ +-- Channel-pins (pinned board) queries (sqlc adoption T3, RIG-3034). These +-- replace the inline SQL literals in internal/store/channel_pins.go; the +-- hand-written Store methods and the in-tx FOR UPDATE lock / cap-check control +-- flow stay exactly as they were and wrap these generated calls. + +-- name: LockChannelForPins :one +SELECT post_policy, COALESCE(owner_account_id, '') AS owner_account_id +FROM channels WHERE id = $1 FOR UPDATE; + +-- name: MessageInChannel :one +SELECT 1 FROM messages m JOIN topics t ON t.id = m.topic_id +WHERE m.id = $1 AND t.channel_id = $2; + +-- name: CountChannelPins :one +SELECT count(*) AS count, COALESCE(MAX(position), -1) + 1 AS next_position +FROM channel_pins WHERE channel_id = $1; + +-- name: DeleteChannelPin :exec +DELETE FROM channel_pins WHERE channel_id = $1 AND message_id = $2; + +-- name: DeleteChannelPinReturningPosition :one +DELETE FROM channel_pins WHERE channel_id = $1 AND message_id = $2 RETURNING position; + +-- name: InsertChannelPin :exec +INSERT INTO channel_pins (channel_id, message_id, position, pinned_at_unix_ms, pinned_by_account_id) +VALUES ($1, $2, $3, $4, $5); + +-- name: PinnedEntries :many +SELECT message_id, position, pinned_at_unix_ms, pinned_by_account_id +FROM channel_pins WHERE channel_id = $1 ORDER BY position; diff --git a/go/internal/store/queries/channels.sql b/go/internal/store/queries/channels.sql new file mode 100644 index 000000000..5cf461c1c --- /dev/null +++ b/go/internal/store/queries/channels.sql @@ -0,0 +1,224 @@ +-- Channel-domain queries (sqlc adoption T3, RIG-3034). These replace the inline +-- SQL literals that lived in internal/store/channels.go; the hand-written Store +-- methods keep their exact signatures and wrap these generated calls, mapping +-- the generated row structs back to the domain Channel / ChannelGroup. +-- +-- The channel-visibility CTE (effective(id, eff_vis)) and the channel/group +-- visibility predicates are repeated per read because sqlc has no query-fragment +-- composition — they replace the former channels.go const fragments +-- (effectiveVisibilityCTE / viewerCTE / channelVisiblePredicate / +-- groupVisiblePredicate). The copies MUST stay textually identical so the stream +-- edge's single-id visibility check cannot drift from the list read (the +-- anti-drift guarantee the frozen record requires). + +-- name: InsertChannelGroup :exec +INSERT INTO channel_groups (id, name, parent_group_id, owner_user_id, visibility) +VALUES ($1, $2, NULLIF($3, ''), $4, $5); + +-- name: GetChannelGroupVisibility :one +SELECT visibility FROM channel_groups WHERE id = $1; + +-- name: InsertChannel :exec +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); + +-- name: UpsertChannelMember :exec +INSERT INTO channel_members (channel_id, account_id, subscribed) +VALUES ($1, $2, $3) +ON CONFLICT (channel_id, account_id) DO UPDATE SET subscribed = EXCLUDED.subscribed; + +-- name: DeleteChannelMember :execrows +DELETE FROM channel_members WHERE channel_id = $1 AND account_id = $2; + +-- name: AgentOwnersByIDs :many +SELECT owner_user_id FROM agent_accounts WHERE account_id = ANY($1::text[]); + +-- name: ChannelMemberExists :one +SELECT EXISTS (SELECT 1 FROM channel_members WHERE channel_id = $1 AND account_id = $2); + +-- name: LockChannelMandatoryKind :one +SELECT mandatory_subscription, kind FROM channels WHERE id = $1 FOR UPDATE; + +-- name: ConvertDMChannel :exec +UPDATE channels SET kind = $1, name = $2, group_id = NULL, mandatory_subscription = FALSE WHERE id = $3; + +-- name: SubscribeConvertedDMParties :exec +UPDATE channel_members cm SET subscribed = TRUE +FROM agent_accounts aa WHERE aa.account_id = cm.account_id AND cm.channel_id = $1; + +-- name: CountAgentMembers :one +SELECT COUNT(*) FROM channel_members cm +JOIN agent_accounts aa ON aa.account_id = cm.account_id +WHERE cm.channel_id = $1; + +-- name: OwnerHasPresentAgent :one +SELECT EXISTS ( + SELECT 1 FROM agent_accounts aa + JOIN channel_members cm ON cm.account_id = aa.account_id + WHERE aa.owner_user_id = $1 AND cm.channel_id = $2 AND aa.account_id <> $1 +); + +-- name: LockChannelPolicy :one +SELECT mandatory_subscription, COALESCE(owner_account_id, '') AS owner_account_id +FROM channels WHERE id = $1 FOR UPDATE; + +-- name: UpdateChannelPolicy :exec +UPDATE channels SET post_policy = $2, owner_account_id = NULLIF($3, ''), mandatory_subscription = $4 WHERE id = $1; + +-- name: GetChannel :one +SELECT id, name, COALESCE(group_id, '') AS group_id, kind, post_policy, + COALESCE(owner_account_id, '') AS owner_account_id, mandatory_subscription +FROM channels WHERE id = $1; + +-- name: ChannelMembersByChannelIDs :many +SELECT channel_id, account_id, subscribed +FROM channel_members +WHERE channel_id = ANY($1::text[]) +ORDER BY account_id; + +-- name: ListChannelGroups :many +WITH RECURSIVE ancestry AS ( + SELECT id, parent_group_id, visibility AS min_vis + FROM channel_groups + UNION ALL + SELECT a.id, g.parent_group_id, LEAST(a.min_vis, g.visibility) + FROM ancestry a + JOIN channel_groups g ON g.id = a.parent_group_id +), +effective AS ( + SELECT id, MIN(min_vis) AS eff_vis + FROM ancestry + GROUP BY id +), +viewer AS ( + SELECT owner_user_id AS uid FROM agent_accounts WHERE account_id = $1 + UNION ALL + SELECT $1 AS uid +) +SELECT g.id, g.name, COALESCE(g.parent_group_id, '') AS parent_group_id, g.owner_user_id, g.visibility +FROM channel_groups g +JOIN effective e ON e.id = g.id +WHERE (e.eff_vis = 1 OR g.owner_user_id IN (SELECT uid FROM viewer)) +ORDER BY g.name; + +-- name: ChannelGroupVisibleTo :one +WITH RECURSIVE ancestry AS ( + SELECT id, parent_group_id, visibility AS min_vis + FROM channel_groups + UNION ALL + SELECT a.id, g.parent_group_id, LEAST(a.min_vis, g.visibility) + FROM ancestry a + JOIN channel_groups g ON g.id = a.parent_group_id +), +effective AS ( + SELECT id, MIN(min_vis) AS eff_vis + FROM ancestry + GROUP BY id +), +viewer AS ( + SELECT owner_user_id AS uid FROM agent_accounts WHERE account_id = $1 + UNION ALL + SELECT $1 AS uid +) +SELECT EXISTS ( + SELECT 1 FROM channel_groups g + JOIN effective e ON e.id = g.id + WHERE g.id = $2 AND (e.eff_vis = 1 OR g.owner_user_id IN (SELECT uid FROM viewer)) +); + +-- name: ListChannels :many +WITH RECURSIVE ancestry AS ( + SELECT id, parent_group_id, visibility AS min_vis + FROM channel_groups + UNION ALL + SELECT a.id, g.parent_group_id, LEAST(a.min_vis, g.visibility) + FROM ancestry a + JOIN channel_groups g ON g.id = a.parent_group_id +), +effective AS ( + SELECT id, MIN(min_vis) AS eff_vis + FROM ancestry + GROUP BY id +) +SELECT c.id, c.name, COALESCE(c.group_id, '') AS group_id, c.kind, c.post_policy, + COALESCE(c.owner_account_id, '') AS owner_account_id, c.mandatory_subscription +FROM channels c +WHERE ( + EXISTS ( + SELECT 1 FROM channel_members cm + WHERE cm.channel_id = c.id AND cm.account_id = $1 + ) + OR ( + c.kind = 0 AND c.group_id IS NOT NULL AND EXISTS ( + SELECT 1 FROM effective e WHERE e.id = c.group_id AND e.eff_vis = 1 + ) + ) + ) +ORDER BY c.name; + +-- name: ChannelVisibleTo :one +WITH RECURSIVE ancestry AS ( + SELECT id, parent_group_id, visibility AS min_vis + FROM channel_groups + UNION ALL + SELECT a.id, g.parent_group_id, LEAST(a.min_vis, g.visibility) + FROM ancestry a + JOIN channel_groups g ON g.id = a.parent_group_id +), +effective AS ( + SELECT id, MIN(min_vis) AS eff_vis + FROM ancestry + GROUP BY id +) +SELECT EXISTS ( + SELECT 1 FROM channels c + WHERE c.id = $2 AND ( + EXISTS ( + SELECT 1 FROM channel_members cm + WHERE cm.channel_id = c.id AND cm.account_id = $1 + ) + OR ( + c.kind = 0 AND c.group_id IS NOT NULL AND EXISTS ( + SELECT 1 FROM effective e WHERE e.id = c.group_id AND e.eff_vis = 1 + ) + ) + ) +); + +-- name: ChannelsByNameForViewer :many +WITH RECURSIVE ancestry AS ( + SELECT id, parent_group_id, visibility AS min_vis + FROM channel_groups + UNION ALL + SELECT a.id, g.parent_group_id, LEAST(a.min_vis, g.visibility) + FROM ancestry a + JOIN channel_groups g ON g.id = a.parent_group_id +), +effective AS ( + SELECT id, MIN(min_vis) AS eff_vis + FROM ancestry + GROUP BY id +) +SELECT c.id, c.name, COALESCE(c.group_id, '') AS group_id, c.kind, c.post_policy, + COALESCE(c.owner_account_id, '') AS owner_account_id, c.mandatory_subscription +FROM channels c +WHERE c.name = $2 AND ( + EXISTS ( + SELECT 1 FROM channel_members cm + WHERE cm.channel_id = c.id AND cm.account_id = $1 + ) + OR ( + c.kind = 0 AND c.group_id IS NOT NULL AND EXISTS ( + SELECT 1 FROM effective e WHERE e.id = c.group_id AND e.eff_vis = 1 + ) + ) + ) +ORDER BY c.id; + +-- name: InsertAgentWorkspaceIgnore :exec +INSERT INTO agent_workspaces (id, agent_account_id) +VALUES ($1, $2) +ON CONFLICT (agent_account_id) DO NOTHING; + +-- name: GetAgentWorkspaceID :one +SELECT id FROM agent_workspaces WHERE agent_account_id = $1; diff --git a/go/internal/store/queries/coordination.sql b/go/internal/store/queries/coordination.sql new file mode 100644 index 000000000..543d38a9e --- /dev/null +++ b/go/internal/store/queries/coordination.sql @@ -0,0 +1,39 @@ +-- Coordination-store queries (sqlc adoption T3, RIG-3034). These replace the +-- inline SQL literals in internal/store/coordination.go; the hand-written Store +-- methods keep their signatures, the per-owner advisory-lock discipline, the +-- suffix-search resolution loop, and the WithTx seam (which stays hand-written). +-- The member INSERT/DELETE reuse EnsureChannelMember (accounts.sql) and +-- DeleteChannelMember (channels.sql) — the statements are identical. + +-- name: GetCoordinationGroup :one +SELECT id FROM channel_groups +WHERE owner_user_id = $1 AND name = $2 AND parent_group_id IS NULL AND visibility = $3; + +-- name: InsertCoordinationGroup :exec +INSERT INTO channel_groups (id, name, parent_group_id, owner_user_id, visibility) +VALUES ($1, $2, NULL, $3, $4); + +-- name: GetCoordinationChannelByName :one +SELECT id, COALESCE(owner_account_id, '') AS owner_account_id +FROM channels WHERE group_id = $1 AND name = $2; + +-- name: InsertCoordinationChannel :one +INSERT INTO channels (id, name, group_id, kind, post_policy, owner_account_id, mandatory_subscription) +VALUES ($1, $2, $3, $4, $5, NULLIF($6, ''), $7) +ON CONFLICT (group_id, name) WHERE group_id IS NOT NULL DO NOTHING +RETURNING id; + +-- name: ChannelMemberIDs :many +SELECT account_id FROM channel_members WHERE channel_id = $1; + +-- name: CoordinationReports :many +SELECT account_id FROM agent_accounts WHERE parent_agent_id = $1 ORDER BY account_id; + +-- name: ResolveCoordinationManager :one +SELECT a.handle, ag.owner_user_id +FROM accounts a +JOIN agent_accounts ag ON ag.account_id = a.id +WHERE a.id = $1; + +-- name: LockOwnerCoordination :exec +SELECT pg_advisory_xact_lock(hashtext('coordination:' || $1)); diff --git a/tools/inline-sql-gate/index.ts b/tools/inline-sql-gate/index.ts index 7218ae83b..6540eb04c 100644 --- a/tools/inline-sql-gate/index.ts +++ b/tools/inline-sql-gate/index.ts @@ -86,9 +86,6 @@ export const ALLOWLIST: string[] = [ // plus dm.go (added post-record). Each drops as its domain migrates. // accounts.go migrated in T2 (RIG-3034); agent_tree.go was never seeded here // (its SQL was const-hoisted, so the literal-scoped gate produced no finding). - "go/internal/store/channels.go", - "go/internal/store/channel_pins.go", - "go/internal/store/coordination.go", "go/internal/store/messages.go", "go/internal/store/topics.go", "go/internal/store/delivery_cursors.go",