From c84f42a1926cd454d2bd69dec1b44d73dd0c259f Mon Sep 17 00:00:00 2001 From: mintaka Date: Mon, 31 Aug 2026 12:00:33 -0400 Subject: [PATCH] feat(store): migrate messages + topics + delivery to sqlc (T4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Migrate the message, topic, and delivery-consumer store domains from inline pgx SQL to sqlc-generated queries (RIG-3034 T4, design record §T4). The hand-written Store methods keep their exact exported signatures and wrap the generated calls; error wrapping, domain-type mapping, tx orchestration, the D9 not-found/forbidden merge, and the D2 seed self-guard all stay hand-written. Files migrated (queries/*.sql compiled into internal/store/db/): - messages.go (16 sites): post-policy read, insert (:one, ON CONFLICT dedup signalled via errMessageInsertConflict), topic get-or-create, list/search (:many), ask-containment find (FOR UPDATE), authored update (:one RETURNING), block update (:execrows), dedup read. - topics.go (9 sites): list, resolve-for-update, rename/merge resolution loop, archive, get. - delivery_cursors.go (15 sites): seed (per-agent + per-channel), owed-mention record/read/clear (:execrows)/count, mention-routed mark + recovery scan, AckDelivery resolve/load(FOR UPDATE)/self-authored/advance, undelivered sweep, in-sweep-set. - delivery_reads.go (7 sites): subscribed agents, channel agent members, is-agent, message-by-id, message-channel, topic/channel names, sweep channels. - presence_reads.go (2 sites): open-ask JSONB path probe, shared-channel EXISTS. Const-hoisted, never in the gate allowlist — no entry to remove. Each migrated statement is semantically identical to the inline SQL it replaces; RowsAffected-branching sites use :execrows. scanMessages/scanTopics/ scanMessagesByChannel and the now-dead execer interface are removed in favor of generated-row mappers (messageFromParts, topicFromRow); the fail-loud decode error on a malformed block set is preserved. applyAskAnswer takes a nil-guard on its *Message parameter: messageFromParts returns a value Message flowing in via &msg, and the module's nilaway gate (now gating, not advisory) cannot prove the pointer non-nil across the call, so it reports a potential nil-panic at the msg.Blocks deref. The guard is the clean suppression (nilaway ignores //nolint); the branch is a defensive invariant — the sole caller passes the address of a freshly-built local — so the gating nilaway job is its regression guard. inline-sql-gate allowlist: 21 -> 17 (drop messages.go, topics.go, delivery_cursors.go, delivery_reads.go). Verified: go build ./... clean; go test ./internal/store/ ok; full pgtest lane (podman postgres:16) ok; golangci-lint 0 issues; sqlc generate drift-clean; GATE_ROOT=. bun tools/inline-sql-gate/index.ts exits 0 (allowlist 17). Co-authored-by: Matt Wilkinson --- go/internal/store/db/delivery_cursors.sql.go | 385 ++++++++++++++ go/internal/store/db/delivery_reads.sql.go | 182 +++++++ go/internal/store/db/messages.sql.go | 477 ++++++++++++++++++ go/internal/store/db/presence_reads.sql.go | 53 ++ go/internal/store/db/querier.go | 81 +++ go/internal/store/db/topics.sql.go | 181 +++++++ go/internal/store/delivery_cursors.go | 301 ++++------- go/internal/store/delivery_reads.go | 136 ++--- go/internal/store/messages.go | 329 ++++++------ go/internal/store/presence_reads.go | 27 +- .../store/queries/delivery_cursors.sql | 97 ++++ go/internal/store/queries/delivery_reads.sql | 46 ++ go/internal/store/queries/messages.sql | 99 ++++ go/internal/store/queries/presence_reads.sql | 21 + go/internal/store/queries/topics.sql | 43 ++ go/internal/store/store.go | 7 - go/internal/store/topics.go | 103 ++-- tools/inline-sql-gate/index.ts | 10 +- 18 files changed, 2015 insertions(+), 563 deletions(-) create mode 100644 go/internal/store/db/delivery_cursors.sql.go create mode 100644 go/internal/store/db/delivery_reads.sql.go create mode 100644 go/internal/store/db/messages.sql.go create mode 100644 go/internal/store/db/presence_reads.sql.go create mode 100644 go/internal/store/db/topics.sql.go create mode 100644 go/internal/store/queries/delivery_cursors.sql create mode 100644 go/internal/store/queries/delivery_reads.sql create mode 100644 go/internal/store/queries/messages.sql create mode 100644 go/internal/store/queries/presence_reads.sql create mode 100644 go/internal/store/queries/topics.sql diff --git a/go/internal/store/db/delivery_cursors.sql.go b/go/internal/store/db/delivery_cursors.sql.go new file mode 100644 index 000000000..2eabb0497 --- /dev/null +++ b/go/internal/store/db/delivery_cursors.sql.go @@ -0,0 +1,385 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: delivery_cursors.sql + +package db + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const advanceDeliveryCursor = `-- name: AdvanceDeliveryCursor :exec +UPDATE agent_delivery_cursors +SET acked_seq = $3, above_seqs = $4, acked_at = now() +WHERE agent_account_id = $1 AND channel_id = $2 +` + +type AdvanceDeliveryCursorParams struct { + AgentAccountID string + ChannelID string + AckedSeq int64 + AboveSeqs []int64 +} + +func (q *Queries) AdvanceDeliveryCursor(ctx context.Context, arg AdvanceDeliveryCursorParams) error { + _, err := q.db.Exec(ctx, advanceDeliveryCursor, + arg.AgentAccountID, + arg.ChannelID, + arg.AckedSeq, + arg.AboveSeqs, + ) + return err +} + +const clearOwedMention = `-- name: ClearOwedMention :execrows +DELETE FROM owed_mentions WHERE agent_account_id = $1 AND message_id = $2 +` + +type ClearOwedMentionParams struct { + AgentAccountID string + MessageID string +} + +func (q *Queries) ClearOwedMention(ctx context.Context, arg ClearOwedMentionParams) (int64, error) { + result, err := q.db.Exec(ctx, clearOwedMention, arg.AgentAccountID, arg.MessageID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const countOwedMentions = `-- name: CountOwedMentions :one +SELECT COUNT(*) FROM owed_mentions +` + +func (q *Queries) CountOwedMentions(ctx context.Context) (int64, error) { + row := q.db.QueryRow(ctx, countOwedMentions) + var count int64 + err := row.Scan(&count) + return count, err +} + +const inSweepSet = `-- name: InSweepSet :one +SELECT EXISTS( + SELECT 1 + FROM channel_members cm + JOIN agent_accounts aa ON aa.account_id = cm.account_id + JOIN channels ch ON ch.id = cm.channel_id + WHERE cm.account_id = $1 + AND cm.channel_id = $2 + AND (cm.subscribed OR cm.channel_id = aa.home_channel_id OR ch.mandatory_subscription)) +` + +type InSweepSetParams struct { + AccountID string + ChannelID string +} + +func (q *Queries) InSweepSet(ctx context.Context, arg InSweepSetParams) (bool, error) { + row := q.db.QueryRow(ctx, inSweepSet, arg.AccountID, arg.ChannelID) + var exists bool + err := row.Scan(&exists) + return exists, err +} + +const loadDeliveryCursor = `-- name: LoadDeliveryCursor :one +SELECT acked_seq, above_seqs FROM agent_delivery_cursors +WHERE agent_account_id = $1 AND channel_id = $2 +FOR UPDATE +` + +type LoadDeliveryCursorParams struct { + AgentAccountID string + ChannelID string +} + +type LoadDeliveryCursorRow struct { + AckedSeq int64 + AboveSeqs []int64 +} + +func (q *Queries) LoadDeliveryCursor(ctx context.Context, arg LoadDeliveryCursorParams) (LoadDeliveryCursorRow, error) { + row := q.db.QueryRow(ctx, loadDeliveryCursor, arg.AgentAccountID, arg.ChannelID) + var i LoadDeliveryCursorRow + err := row.Scan(&i.AckedSeq, &i.AboveSeqs) + return i, err +} + +const markMentionsRouted = `-- name: MarkMentionsRouted :exec +UPDATE messages SET mentions_routed_at = $1 WHERE id = $2 +` + +type MarkMentionsRoutedParams struct { + MentionsRoutedAt pgtype.Int8 + ID string +} + +func (q *Queries) MarkMentionsRouted(ctx context.Context, arg MarkMentionsRoutedParams) error { + _, err := q.db.Exec(ctx, markMentionsRouted, arg.MentionsRoutedAt, arg.ID) + return err +} + +const owedMentions = `-- name: OwedMentions :many +SELECT m.id, m.topic_id, t.channel_id, m.author_account_id, m.at_unix_ms, m.blocks +FROM owed_mentions om +JOIN messages m ON m.id = om.message_id +JOIN topics t ON t.id = m.topic_id +WHERE om.agent_account_id = $1 +ORDER BY t.channel_id, m.seq ASC +` + +type OwedMentionsRow struct { + ID string + TopicID string + ChannelID string + AuthorAccountID string + AtUnixMs int64 + Blocks []byte +} + +func (q *Queries) OwedMentions(ctx context.Context, agentAccountID string) ([]OwedMentionsRow, error) { + rows, err := q.db.Query(ctx, owedMentions, agentAccountID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []OwedMentionsRow + for rows.Next() { + var i OwedMentionsRow + if err := rows.Scan( + &i.ID, + &i.TopicID, + &i.ChannelID, + &i.AuthorAccountID, + &i.AtUnixMs, + &i.Blocks, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const recordOwedMention = `-- name: RecordOwedMention :exec +INSERT INTO owed_mentions (agent_account_id, message_id, channel_id, recorded_at_unix_ms) +VALUES ($1, $2, $3, $4) +ON CONFLICT (agent_account_id, message_id) DO NOTHING +` + +type RecordOwedMentionParams struct { + AgentAccountID string + MessageID string + ChannelID string + RecordedAtUnixMs int64 +} + +func (q *Queries) RecordOwedMention(ctx context.Context, arg RecordOwedMentionParams) error { + _, err := q.db.Exec(ctx, recordOwedMention, + arg.AgentAccountID, + arg.MessageID, + arg.ChannelID, + arg.RecordedAtUnixMs, + ) + return err +} + +const resolveAckMessage = `-- name: ResolveAckMessage :one +SELECT m.seq FROM messages m JOIN topics t ON t.id = m.topic_id WHERE m.id = $1 AND t.channel_id = $2 +` + +type ResolveAckMessageParams struct { + ID string + ChannelID string +} + +func (q *Queries) ResolveAckMessage(ctx context.Context, arg ResolveAckMessageParams) (int64, error) { + row := q.db.QueryRow(ctx, resolveAckMessage, arg.ID, arg.ChannelID) + var seq int64 + err := row.Scan(&seq) + return seq, err +} + +const seedChannelDeliveryCursors = `-- name: SeedChannelDeliveryCursors :exec +INSERT INTO agent_delivery_cursors (agent_account_id, channel_id, acked_seq) +SELECT cm.account_id, $1, + COALESCE((SELECT MAX(m.seq) FROM messages m JOIN topics t ON t.id = m.topic_id WHERE t.channel_id = $1), 0) +FROM channel_members cm +JOIN agent_accounts aa ON aa.account_id = cm.account_id +WHERE cm.channel_id = $1 +ON CONFLICT (agent_account_id, channel_id) DO NOTHING +` + +func (q *Queries) SeedChannelDeliveryCursors(ctx context.Context, channelID string) error { + _, err := q.db.Exec(ctx, seedChannelDeliveryCursors, channelID) + return err +} + +const seedDeliveryCursor = `-- name: SeedDeliveryCursor :exec + +INSERT INTO agent_delivery_cursors (agent_account_id, channel_id, acked_seq) +SELECT $1, $2, COALESCE((SELECT MAX(m.seq) FROM messages m JOIN topics t ON t.id = m.topic_id WHERE t.channel_id = $2), 0) +WHERE EXISTS (SELECT 1 FROM agent_accounts WHERE account_id = $1) +ON CONFLICT (agent_account_id, channel_id) DO NOTHING +` + +type SeedDeliveryCursorParams struct { + AgentAccountID string + ChannelID string +} + +// Delivery-cursor queries (sqlc adoption T4, RIG-3034). These replace the inline +// SQL literals in internal/store/delivery_cursors.go; the hand-written Store +// methods keep their signatures, the AckDelivery tx orchestration (the owed-clear +// FIRST, the commit-if-cleared arm, the contiguous-advance loop in Go), and the +// D2 seed self-guard/idempotency contract. The two message-fanout reads +// (OwedMentions, UndeliveredMessages) share the per-channel projection the Go +// drains with an inline loop calling messageFromParts. +func (q *Queries) SeedDeliveryCursor(ctx context.Context, arg SeedDeliveryCursorParams) error { + _, err := q.db.Exec(ctx, seedDeliveryCursor, arg.AgentAccountID, arg.ChannelID) + return err +} + +const selfAuthoredSeqsAbove = `-- name: SelfAuthoredSeqsAbove :many +SELECT m.seq FROM messages m JOIN topics t ON t.id = m.topic_id +WHERE t.channel_id = $1 AND m.seq > $2 AND m.author_account_id = $3 +` + +type SelfAuthoredSeqsAboveParams struct { + ChannelID string + Seq int64 + AuthorAccountID string +} + +func (q *Queries) SelfAuthoredSeqsAbove(ctx context.Context, arg SelfAuthoredSeqsAboveParams) ([]int64, error) { + rows, err := q.db.Query(ctx, selfAuthoredSeqsAbove, arg.ChannelID, arg.Seq, arg.AuthorAccountID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []int64 + for rows.Next() { + var seq int64 + if err := rows.Scan(&seq); err != nil { + return nil, err + } + items = append(items, seq) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const undeliveredMessages = `-- name: UndeliveredMessages :many +SELECT m.id, m.topic_id, t.channel_id, m.author_account_id, m.at_unix_ms, m.blocks +FROM channel_members cm +JOIN agent_accounts aa ON aa.account_id = cm.account_id +JOIN topics t ON t.channel_id = cm.channel_id +JOIN messages m ON m.topic_id = t.id +JOIN channels ch ON ch.id = cm.channel_id +LEFT JOIN agent_delivery_cursors dc + ON dc.agent_account_id = cm.account_id AND dc.channel_id = cm.channel_id +WHERE cm.account_id = $1 + AND (cm.subscribed OR cm.channel_id = aa.home_channel_id OR ch.mandatory_subscription) + AND m.author_account_id <> $1 + AND m.seq > COALESCE( + dc.acked_seq, + (SELECT COALESCE(MAX(mh.seq), 0) FROM messages mh JOIN topics th ON th.id = mh.topic_id WHERE th.channel_id = cm.channel_id)) + AND m.seq <> ALL(COALESCE(dc.above_seqs, '{}'::BIGINT[])) +ORDER BY t.channel_id, m.seq ASC +` + +type UndeliveredMessagesRow struct { + ID string + TopicID string + ChannelID string + AuthorAccountID string + AtUnixMs int64 + Blocks []byte +} + +func (q *Queries) UndeliveredMessages(ctx context.Context, accountID string) ([]UndeliveredMessagesRow, error) { + rows, err := q.db.Query(ctx, undeliveredMessages, accountID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []UndeliveredMessagesRow + for rows.Next() { + var i UndeliveredMessagesRow + if err := rows.Scan( + &i.ID, + &i.TopicID, + &i.ChannelID, + &i.AuthorAccountID, + &i.AtUnixMs, + &i.Blocks, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const unroutedMentionMessages = `-- name: UnroutedMentionMessages :many +SELECT m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks, t.channel_id, m.seq +FROM messages m +JOIN topics t ON t.id = m.topic_id +WHERE m.mentions_routed_at IS NULL AND m.seq > $1 +ORDER BY m.seq ASC +LIMIT $2 +` + +type UnroutedMentionMessagesParams struct { + Seq int64 + Limit int32 +} + +type UnroutedMentionMessagesRow struct { + ID string + TopicID string + AuthorAccountID string + AtUnixMs int64 + Blocks []byte + ChannelID string + Seq int64 +} + +func (q *Queries) UnroutedMentionMessages(ctx context.Context, arg UnroutedMentionMessagesParams) ([]UnroutedMentionMessagesRow, error) { + rows, err := q.db.Query(ctx, unroutedMentionMessages, arg.Seq, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []UnroutedMentionMessagesRow + for rows.Next() { + var i UnroutedMentionMessagesRow + if err := rows.Scan( + &i.ID, + &i.TopicID, + &i.AuthorAccountID, + &i.AtUnixMs, + &i.Blocks, + &i.ChannelID, + &i.Seq, + ); 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/delivery_reads.sql.go b/go/internal/store/db/delivery_reads.sql.go new file mode 100644 index 000000000..b3430f62d --- /dev/null +++ b/go/internal/store/db/delivery_reads.sql.go @@ -0,0 +1,182 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: delivery_reads.sql + +package db + +import ( + "context" +) + +const channelAgentMembers = `-- name: ChannelAgentMembers :many +SELECT aa.account_id +FROM channel_members cm +JOIN agent_accounts aa ON aa.account_id = cm.account_id +WHERE cm.channel_id = $1 + AND cm.account_id <> $2 +ORDER BY aa.account_id +` + +type ChannelAgentMembersParams struct { + ChannelID string + AccountID string +} + +func (q *Queries) ChannelAgentMembers(ctx context.Context, arg ChannelAgentMembersParams) ([]string, error) { + rows, err := q.db.Query(ctx, channelAgentMembers, arg.ChannelID, arg.AccountID) + 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 isAgentAccount = `-- name: IsAgentAccount :one +SELECT EXISTS (SELECT 1 FROM agent_accounts WHERE account_id = $1) +` + +func (q *Queries) IsAgentAccount(ctx context.Context, accountID string) (bool, error) { + row := q.db.QueryRow(ctx, isAgentAccount, accountID) + var exists bool + err := row.Scan(&exists) + return exists, err +} + +const messageByID = `-- name: MessageByID :one +SELECT id, topic_id, author_account_id, at_unix_ms, blocks +FROM messages +WHERE id = $1 +` + +type MessageByIDRow struct { + ID string + TopicID string + AuthorAccountID string + AtUnixMs int64 + Blocks []byte +} + +func (q *Queries) MessageByID(ctx context.Context, id string) (MessageByIDRow, error) { + row := q.db.QueryRow(ctx, messageByID, id) + var i MessageByIDRow + err := row.Scan( + &i.ID, + &i.TopicID, + &i.AuthorAccountID, + &i.AtUnixMs, + &i.Blocks, + ) + return i, err +} + +const messageChannel = `-- name: MessageChannel :one +SELECT t.channel_id FROM messages m JOIN topics t ON t.id = m.topic_id WHERE m.id = $1 +` + +func (q *Queries) MessageChannel(ctx context.Context, id string) (string, error) { + row := q.db.QueryRow(ctx, messageChannel, id) + var channel_id string + err := row.Scan(&channel_id) + return channel_id, err +} + +const subscribedAgents = `-- name: SubscribedAgents :many + +SELECT aa.account_id +FROM channel_members cm +JOIN agent_accounts aa ON aa.account_id = cm.account_id +JOIN channels ch ON ch.id = cm.channel_id +WHERE cm.channel_id = $1 + AND (cm.subscribed OR cm.channel_id = aa.home_channel_id OR ch.mandatory_subscription) + AND cm.account_id <> $2 +ORDER BY aa.account_id +` + +type SubscribedAgentsParams struct { + ChannelID string + AccountID string +} + +// Delivery-consumer read queries (sqlc adoption T4, RIG-3034). These replace the +// inline SQL literals in internal/store/delivery_reads.go; the hand-written Store +// methods keep their signatures, the D1 sweep-set disjunct (kept textually in +// sync with delivery_cursors.sql UndeliveredMessages/InSweepSet), and the D9 +// error mapping. MessageByID shares the message projection the Go drains via +// messageFromParts. +func (q *Queries) SubscribedAgents(ctx context.Context, arg SubscribedAgentsParams) ([]string, error) { + rows, err := q.db.Query(ctx, subscribedAgents, arg.ChannelID, arg.AccountID) + 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 sweepChannels = `-- name: SweepChannels :many +SELECT cm.channel_id +FROM channel_members cm +JOIN agent_accounts aa ON aa.account_id = cm.account_id +JOIN channels ch ON ch.id = cm.channel_id +WHERE cm.account_id = $1 + AND (cm.subscribed OR cm.channel_id = aa.home_channel_id OR ch.mandatory_subscription) +ORDER BY cm.channel_id +` + +func (q *Queries) SweepChannels(ctx context.Context, accountID string) ([]string, error) { + rows, err := q.db.Query(ctx, sweepChannels, accountID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []string + for rows.Next() { + var channel_id string + if err := rows.Scan(&channel_id); err != nil { + return nil, err + } + items = append(items, channel_id) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const topicChannelNames = `-- name: TopicChannelNames :one +SELECT t.name AS topic_name, c.name AS channel_name FROM topics t JOIN channels c ON c.id = t.channel_id WHERE t.id = $1 +` + +type TopicChannelNamesRow struct { + TopicName string + ChannelName string +} + +func (q *Queries) TopicChannelNames(ctx context.Context, id string) (TopicChannelNamesRow, error) { + row := q.db.QueryRow(ctx, topicChannelNames, id) + var i TopicChannelNamesRow + err := row.Scan(&i.TopicName, &i.ChannelName) + return i, err +} diff --git a/go/internal/store/db/messages.sql.go b/go/internal/store/db/messages.sql.go new file mode 100644 index 000000000..e16695d7c --- /dev/null +++ b/go/internal/store/db/messages.sql.go @@ -0,0 +1,477 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: messages.sql + +package db + +import ( + "context" +) + +const findAskMessage = `-- name: FindAskMessage :many +SELECT m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks +FROM messages m +JOIN topics t ON t.id = m.topic_id +JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 +WHERE m.blocks @> $2::jsonb +FOR UPDATE OF m +` + +type FindAskMessageParams struct { + AccountID string + Column2 []byte +} + +type FindAskMessageRow struct { + ID string + TopicID string + AuthorAccountID string + AtUnixMs int64 + Blocks []byte +} + +func (q *Queries) FindAskMessage(ctx context.Context, arg FindAskMessageParams) ([]FindAskMessageRow, error) { + rows, err := q.db.Query(ctx, findAskMessage, arg.AccountID, arg.Column2) + if err != nil { + return nil, err + } + defer rows.Close() + var items []FindAskMessageRow + for rows.Next() { + var i FindAskMessageRow + if err := rows.Scan( + &i.ID, + &i.TopicID, + &i.AuthorAccountID, + &i.AtUnixMs, + &i.Blocks, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getChannelPostPolicy = `-- name: GetChannelPostPolicy :one + +SELECT post_policy, COALESCE(owner_account_id, '') AS owner_account_id +FROM channels WHERE id = $1 +` + +type GetChannelPostPolicyRow struct { + PostPolicy int16 + OwnerAccountID string +} + +// Message-domain queries (sqlc adoption T4, RIG-3034). These replace the inline +// SQL literals in internal/store/messages.go; the hand-written Store methods keep +// their exact signatures, their tx orchestration (AppendMessage/AnswerAsk begin +// and commit their own txns), the ON CONFLICT idempotency signalling +// (errMessageInsertConflict), the JSONB block (de)serialization, and the D9 +// not-found/forbidden error mapping — all hand-written around these generated +// calls. Every message read shares the id/topic_id/author_account_id/at_unix_ms/ +// blocks projection (the former scanMessages order) so the Go maps each row the +// same way via messageFromParts. +func (q *Queries) GetChannelPostPolicy(ctx context.Context, id string) (GetChannelPostPolicyRow, error) { + row := q.db.QueryRow(ctx, getChannelPostPolicy, id) + var i GetChannelPostPolicyRow + err := row.Scan(&i.PostPolicy, &i.OwnerAccountID) + return i, err +} + +const getMessageBlocks = `-- name: GetMessageBlocks :one +SELECT blocks FROM messages WHERE id = $1 +` + +func (q *Queries) GetMessageBlocks(ctx context.Context, id string) ([]byte, error) { + row := q.db.QueryRow(ctx, getMessageBlocks, id) + var blocks []byte + err := row.Scan(&blocks) + return blocks, err +} + +const getMessageByRequestID = `-- name: GetMessageByRequestID :many +SELECT id, topic_id, author_account_id, at_unix_ms, blocks +FROM messages +WHERE author_account_id = $1 AND client_request_id = $2 +` + +type GetMessageByRequestIDParams struct { + AuthorAccountID string + ClientRequestID string +} + +type GetMessageByRequestIDRow struct { + ID string + TopicID string + AuthorAccountID string + AtUnixMs int64 + Blocks []byte +} + +func (q *Queries) GetMessageByRequestID(ctx context.Context, arg GetMessageByRequestIDParams) ([]GetMessageByRequestIDRow, error) { + rows, err := q.db.Query(ctx, getMessageByRequestID, arg.AuthorAccountID, arg.ClientRequestID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetMessageByRequestIDRow + for rows.Next() { + var i GetMessageByRequestIDRow + if err := rows.Scan( + &i.ID, + &i.TopicID, + &i.AuthorAccountID, + &i.AtUnixMs, + &i.Blocks, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getPageCursorSeq = `-- name: GetPageCursorSeq :one +SELECT m.seq FROM messages m +JOIN topics t ON t.id = m.topic_id +JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 +WHERE m.id = $2 AND t.channel_id = $3 +` + +type GetPageCursorSeqParams struct { + AccountID string + ID string + ChannelID string +} + +func (q *Queries) GetPageCursorSeq(ctx context.Context, arg GetPageCursorSeqParams) (int64, error) { + row := q.db.QueryRow(ctx, getPageCursorSeq, arg.AccountID, arg.ID, arg.ChannelID) + var seq int64 + err := row.Scan(&seq) + return seq, err +} + +const getTopicByName = `-- name: GetTopicByName :one +SELECT id, archived FROM topics WHERE channel_id = $1 AND lower(name) = lower($2) +` + +type GetTopicByNameParams struct { + ChannelID string + Lower string +} + +type GetTopicByNameRow struct { + ID string + Archived bool +} + +func (q *Queries) GetTopicByName(ctx context.Context, arg GetTopicByNameParams) (GetTopicByNameRow, error) { + row := q.db.QueryRow(ctx, getTopicByName, arg.ChannelID, arg.Lower) + var i GetTopicByNameRow + err := row.Scan(&i.ID, &i.Archived) + return i, err +} + +const getTopicChannel = `-- name: GetTopicChannel :one +SELECT channel_id FROM topics WHERE id = $1 +` + +func (q *Queries) GetTopicChannel(ctx context.Context, id string) (string, error) { + row := q.db.QueryRow(ctx, getTopicChannel, id) + var channel_id string + err := row.Scan(&channel_id) + return channel_id, err +} + +const insertMessage = `-- name: InsertMessage :one +INSERT INTO messages (id, topic_id, author_account_id, at_unix_ms, blocks, text_content, client_request_id) +VALUES ($1, $2, $3, $4, $5, $6, $7) +ON CONFLICT (author_account_id, client_request_id) WHERE client_request_id <> '' +DO NOTHING +RETURNING id, at_unix_ms, seq +` + +type InsertMessageParams struct { + ID string + TopicID string + AuthorAccountID string + AtUnixMs int64 + Blocks []byte + TextContent string + ClientRequestID string +} + +type InsertMessageRow struct { + ID string + AtUnixMs int64 + Seq int64 +} + +func (q *Queries) InsertMessage(ctx context.Context, arg InsertMessageParams) (InsertMessageRow, error) { + row := q.db.QueryRow(ctx, insertMessage, + arg.ID, + arg.TopicID, + arg.AuthorAccountID, + arg.AtUnixMs, + arg.Blocks, + arg.TextContent, + arg.ClientRequestID, + ) + var i InsertMessageRow + err := row.Scan(&i.ID, &i.AtUnixMs, &i.Seq) + return i, err +} + +const insertTopicIgnore = `-- name: InsertTopicIgnore :exec +INSERT INTO topics (id, channel_id, name, created_by_account_id, created_at_unix_ms) +VALUES ($1, $2, $3, $4, $5) +ON CONFLICT (channel_id, lower(name)) DO NOTHING +` + +type InsertTopicIgnoreParams struct { + ID string + ChannelID string + Name string + CreatedByAccountID string + CreatedAtUnixMs int64 +} + +func (q *Queries) InsertTopicIgnore(ctx context.Context, arg InsertTopicIgnoreParams) error { + _, err := q.db.Exec(ctx, insertTopicIgnore, + arg.ID, + arg.ChannelID, + arg.Name, + arg.CreatedByAccountID, + arg.CreatedAtUnixMs, + ) + return err +} + +const listMessages = `-- name: ListMessages :many +SELECT m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks +FROM messages m +JOIN topics t ON t.id = m.topic_id +JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 +WHERE t.channel_id = $2 AND ($3 = 0 OR m.seq < $3) AND ($5 = 0 OR m.seq <= $5) + AND ($6 = '' OR m.topic_id = $6) +ORDER BY m.seq DESC +LIMIT $4 +` + +type ListMessagesParams struct { + AccountID string + ChannelID string + Column3 interface{} + Limit int32 + Column5 interface{} + Column6 interface{} +} + +type ListMessagesRow struct { + ID string + TopicID string + AuthorAccountID string + AtUnixMs int64 + Blocks []byte +} + +func (q *Queries) ListMessages(ctx context.Context, arg ListMessagesParams) ([]ListMessagesRow, error) { + rows, err := q.db.Query(ctx, listMessages, + arg.AccountID, + arg.ChannelID, + arg.Column3, + arg.Limit, + arg.Column5, + arg.Column6, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListMessagesRow + for rows.Next() { + var i ListMessagesRow + if err := rows.Scan( + &i.ID, + &i.TopicID, + &i.AuthorAccountID, + &i.AtUnixMs, + &i.Blocks, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const messagesHeadSeq = `-- name: MessagesHeadSeq :one +SELECT COALESCE(MAX(seq), 0)::BIGINT AS head FROM messages +` + +func (q *Queries) MessagesHeadSeq(ctx context.Context) (int64, error) { + row := q.db.QueryRow(ctx, messagesHeadSeq) + var head int64 + err := row.Scan(&head) + return head, err +} + +const reviveTopic = `-- name: ReviveTopic :exec +UPDATE topics SET archived = FALSE WHERE id = $1 +` + +func (q *Queries) ReviveTopic(ctx context.Context, id string) error { + _, err := q.db.Exec(ctx, reviveTopic, id) + return err +} + +const searchMessages = `-- name: SearchMessages :many +SELECT m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks +FROM messages m +JOIN topics t ON t.id = m.topic_id +JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 +WHERE m.search_tsv @@ websearch_to_tsquery('english', $2) + AND ($3 = '' OR t.channel_id = $3) + AND ($5 = 0 OR m.seq <= $5) +ORDER BY ts_rank(m.search_tsv, websearch_to_tsquery('english', $2)) DESC, m.seq DESC +LIMIT $4 +` + +type SearchMessagesParams struct { + AccountID string + WebsearchToTsquery string + Column3 interface{} + Limit int32 + Column5 interface{} +} + +type SearchMessagesRow struct { + ID string + TopicID string + AuthorAccountID string + AtUnixMs int64 + Blocks []byte +} + +func (q *Queries) SearchMessages(ctx context.Context, arg SearchMessagesParams) ([]SearchMessagesRow, error) { + rows, err := q.db.Query(ctx, searchMessages, + arg.AccountID, + arg.WebsearchToTsquery, + arg.Column3, + arg.Limit, + arg.Column5, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []SearchMessagesRow + for rows.Next() { + var i SearchMessagesRow + if err := rows.Scan( + &i.ID, + &i.TopicID, + &i.AuthorAccountID, + &i.AtUnixMs, + &i.Blocks, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const updateMessageBlocks = `-- name: UpdateMessageBlocks :execrows +UPDATE messages SET blocks = $1, text_content = $2 WHERE id = $3 +` + +type UpdateMessageBlocksParams struct { + Blocks []byte + TextContent string + ID string +} + +func (q *Queries) UpdateMessageBlocks(ctx context.Context, arg UpdateMessageBlocksParams) (int64, error) { + result, err := q.db.Exec(ctx, updateMessageBlocks, arg.Blocks, arg.TextContent, arg.ID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const updateMessageBlocksAsAuthor = `-- name: UpdateMessageBlocksAsAuthor :one +UPDATE messages m +SET blocks = $1, text_content = $2 +FROM topics t +WHERE m.id = $3 + AND t.id = m.topic_id + AND m.author_account_id = $4 + AND EXISTS ( + SELECT 1 FROM channel_members cm + WHERE cm.channel_id = t.channel_id AND cm.account_id = $4 + ) +RETURNING m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks +` + +type UpdateMessageBlocksAsAuthorParams struct { + Blocks []byte + TextContent string + ID string + AuthorAccountID string +} + +type UpdateMessageBlocksAsAuthorRow struct { + ID string + TopicID string + AuthorAccountID string + AtUnixMs int64 + Blocks []byte +} + +func (q *Queries) UpdateMessageBlocksAsAuthor(ctx context.Context, arg UpdateMessageBlocksAsAuthorParams) (UpdateMessageBlocksAsAuthorRow, error) { + row := q.db.QueryRow(ctx, updateMessageBlocksAsAuthor, + arg.Blocks, + arg.TextContent, + arg.ID, + arg.AuthorAccountID, + ) + var i UpdateMessageBlocksAsAuthorRow + err := row.Scan( + &i.ID, + &i.TopicID, + &i.AuthorAccountID, + &i.AtUnixMs, + &i.Blocks, + ) + return i, err +} + +const updateTopicLastSeq = `-- name: UpdateTopicLastSeq :exec +UPDATE topics SET last_seq = GREATEST(last_seq, $2) WHERE id = $1 +` + +type UpdateTopicLastSeqParams struct { + ID string + LastSeq int64 +} + +func (q *Queries) UpdateTopicLastSeq(ctx context.Context, arg UpdateTopicLastSeqParams) error { + _, err := q.db.Exec(ctx, updateTopicLastSeq, arg.ID, arg.LastSeq) + return err +} diff --git a/go/internal/store/db/presence_reads.sql.go b/go/internal/store/db/presence_reads.sql.go new file mode 100644 index 000000000..1e0389362 --- /dev/null +++ b/go/internal/store/db/presence_reads.sql.go @@ -0,0 +1,53 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: presence_reads.sql + +package db + +import ( + "context" +) + +const agentHasOpenAsk = `-- name: AgentHasOpenAsk :one + +SELECT EXISTS ( + SELECT 1 FROM messages + WHERE author_account_id = $1 + AND blocks @? '$[*] ? (@.kind == "ask" && (!exists(@.ask.answered) || @.ask.answered == false))' +) +` + +// Presence-component read queries (sqlc adoption T4, RIG-3034). These replace the +// const-hoisted SQL in internal/store/presence_reads.go (it was never in the +// inline-sql-gate allowlist — the gate is literal-at-callsite scoped and this +// file passed its SQL as a const identifier). The hand-written Store methods keep +// their signatures and error mapping. +func (q *Queries) AgentHasOpenAsk(ctx context.Context, authorAccountID string) (bool, error) { + row := q.db.QueryRow(ctx, agentHasOpenAsk, authorAccountID) + var exists bool + err := row.Scan(&exists) + return exists, err +} + +const sharesVisibleChannel = `-- name: SharesVisibleChannel :one +SELECT EXISTS ( + SELECT 1 + FROM channel_members cm1 + JOIN channel_members cm2 ON cm2.channel_id = cm1.channel_id + WHERE cm1.account_id = $1 + AND cm2.account_id = $2 +) +` + +type SharesVisibleChannelParams struct { + AccountID string + AccountID_2 string +} + +func (q *Queries) SharesVisibleChannel(ctx context.Context, arg SharesVisibleChannelParams) (bool, error) { + row := q.db.QueryRow(ctx, sharesVisibleChannel, arg.AccountID, arg.AccountID_2) + var exists bool + err := row.Scan(&exists) + return exists, err +} diff --git a/go/internal/store/db/querier.go b/go/internal/store/db/querier.go index f60f0e3c4..192274be0 100644 --- a/go/internal/store/db/querier.go +++ b/go/internal/store/db/querier.go @@ -13,6 +13,13 @@ import ( type Querier interface { AccountVisibleTo(ctx context.Context, arg AccountVisibleToParams) (bool, error) AcquireOwnerTreeLock(ctx context.Context, hashtext string) error + AdvanceDeliveryCursor(ctx context.Context, arg AdvanceDeliveryCursorParams) error + // Presence-component read queries (sqlc adoption T4, RIG-3034). These replace the + // const-hoisted SQL in internal/store/presence_reads.go (it was never in the + // inline-sql-gate allowlist — the gate is literal-at-callsite scoped and this + // file passed its SQL as a const identifier). The hand-written Store methods keep + // their signatures and error mapping. + AgentHasOpenAsk(ctx context.Context, authorAccountID string) (bool, error) // Agent-tree queries (sqlc adoption T2, RIG-3034). These replace the // const-hoisted `agentTreeProjection` + per-caller WHERE composition that lived // in internal/store/agent_tree.go and was run through the `queryAgents` helper — @@ -28,21 +35,26 @@ type Querier interface { 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) + ChannelAgentMembers(ctx context.Context, arg ChannelAgentMembersParams) ([]string, 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) + ClearOwedMention(ctx context.Context, arg ClearOwedMentionParams) (int64, 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) + CountOwedMentions(ctx context.Context) (int64, 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) + DeleteTopic(ctx context.Context, id string) error EnsureChannelMember(ctx context.Context, arg EnsureChannelMemberParams) error + FindAskMessage(ctx context.Context, arg FindAskMessageParams) ([]FindAskMessageRow, 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) @@ -51,6 +63,16 @@ type Querier interface { GetAgentWorkspaceID(ctx context.Context, agentAccountID string) (string, error) GetChannel(ctx context.Context, id string) (GetChannelRow, error) GetChannelGroupVisibility(ctx context.Context, id string) (int16, error) + // Message-domain queries (sqlc adoption T4, RIG-3034). These replace the inline + // SQL literals in internal/store/messages.go; the hand-written Store methods keep + // their exact signatures, their tx orchestration (AppendMessage/AnswerAsk begin + // and commit their own txns), the ON CONFLICT idempotency signalling + // (errMessageInsertConflict), the JSONB block (de)serialization, and the D9 + // not-found/forbidden error mapping — all hand-written around these generated + // calls. Every message read shares the id/topic_id/author_account_id/at_unix_ms/ + // blocks projection (the former scanMessages order) so the Go maps each row the + // same way via messageFromParts. + GetChannelPostPolicy(ctx context.Context, id string) (GetChannelPostPolicyRow, 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 @@ -60,8 +82,15 @@ type Querier interface { // DeleteChannelMember (channels.sql) — the statements are identical. GetCoordinationGroup(ctx context.Context, arg GetCoordinationGroupParams) (string, error) GetGlobalHandleID(ctx context.Context, handle string) (string, error) + GetMessageBlocks(ctx context.Context, id string) ([]byte, error) + GetMessageByRequestID(ctx context.Context, arg GetMessageByRequestIDParams) ([]GetMessageByRequestIDRow, error) + GetPageCursorSeq(ctx context.Context, arg GetPageCursorSeqParams) (int64, error) + GetTopic(ctx context.Context, id string) (Topic, error) + GetTopicByName(ctx context.Context, arg GetTopicByNameParams) (GetTopicByNameRow, error) + GetTopicChannel(ctx context.Context, id string) (string, error) GetVisibleAgentHandleID(ctx context.Context, arg GetVisibleAgentHandleIDParams) (string, error) GetVisibleGlobalHandleID(ctx context.Context, arg GetVisibleGlobalHandleIDParams) (string, error) + InSweepSet(ctx context.Context, arg InSweepSetParams) (bool, error) // Account-domain queries (sqlc adoption T2, RIG-3034). These replace the inline // SQL literals that lived in internal/store/accounts.go; the hand-written Store // methods keep their exact signatures and wrap these generated calls, mapping the @@ -98,11 +127,23 @@ type Querier interface { InsertCoordinationChannel(ctx context.Context, arg InsertCoordinationChannelParams) (string, error) InsertCoordinationGroup(ctx context.Context, arg InsertCoordinationGroupParams) error InsertHomeChannel(ctx context.Context, arg InsertHomeChannelParams) error + InsertMessage(ctx context.Context, arg InsertMessageParams) (InsertMessageRow, error) InsertSystemAccount(ctx context.Context, accountID string) error + InsertTopicIgnore(ctx context.Context, arg InsertTopicIgnoreParams) error InsertUserAccount(ctx context.Context, arg InsertUserAccountParams) error + IsAgentAccount(ctx context.Context, accountID string) (bool, error) ListChannelGroups(ctx context.Context, accountID string) ([]ListChannelGroupsRow, error) ListChannels(ctx context.Context, accountID string) ([]ListChannelsRow, error) + ListMessages(ctx context.Context, arg ListMessagesParams) ([]ListMessagesRow, error) + // Topic-domain queries (sqlc adoption T4, RIG-3034). These replace the inline + // SQL literals in internal/store/topics.go; the hand-written Store methods keep + // their signatures, the UpdateTopic tx orchestration, the rename/merge resolution + // loop, and the D9 not-found/forbidden error mapping. The topic projection + // (id, channel_id, name, created_by_account_id, created_at_unix_ms, archived, + // last_seq) matches the former scanTopics order so the Go maps each row to Topic. + ListTopics(ctx context.Context, arg ListTopicsParams) ([]Topic, error) ListVisibleAccounts(ctx context.Context, id string) ([]ListVisibleAccountsRow, error) + LoadDeliveryCursor(ctx context.Context, arg LoadDeliveryCursorParams) (LoadDeliveryCursorRow, 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 @@ -111,11 +152,24 @@ type Querier interface { 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 + MarkMentionsRouted(ctx context.Context, arg MarkMentionsRoutedParams) error + MergeTopicLastSeq(ctx context.Context, arg MergeTopicLastSeqParams) error + MessageByID(ctx context.Context, id string) (MessageByIDRow, error) + MessageChannel(ctx context.Context, id string) (string, error) MessageInChannel(ctx context.Context, arg MessageInChannelParams) (int32, error) + MessagesHeadSeq(ctx context.Context) (int64, error) + MoveMessagesToTopic(ctx context.Context, arg MoveMessagesToTopicParams) error + OwedMentions(ctx context.Context, agentAccountID string) ([]OwedMentionsRow, error) OwnerHasPresentAgent(ctx context.Context, arg OwnerHasPresentAgentParams) (bool, error) PinnedEntries(ctx context.Context, channelID string) ([]PinnedEntriesRow, error) + RecordOwedMention(ctx context.Context, arg RecordOwedMentionParams) error + RenameTopic(ctx context.Context, arg RenameTopicParams) error + ResolveAckMessage(ctx context.Context, arg ResolveAckMessageParams) (int64, error) ResolveCoordinationManager(ctx context.Context, id string) (ResolveCoordinationManagerRow, error) ResolveOwner(ctx context.Context, accountID string) (string, error) + ResolveTopicForUpdate(ctx context.Context, arg ResolveTopicForUpdateParams) (string, error) + ResolveTopicRenameTarget(ctx context.Context, arg ResolveTopicRenameTargetParams) (string, error) + ReviveTopic(ctx context.Context, id string) error // Scaffold-only query proving sqlc generation works end to end (T1). // // This is NOT a real store query: the per-domain query files land in T2..T6 as @@ -124,10 +178,37 @@ type Querier interface { // selects a genuinely existing column from a genuinely existing table // (tenants, 0001_init.sql), so sqlc compiles it against the real schema. ScaffoldGetTenant(ctx context.Context, id string) (Tenant, error) + SearchMessages(ctx context.Context, arg SearchMessagesParams) ([]SearchMessagesRow, error) + SeedChannelDeliveryCursors(ctx context.Context, channelID string) error + // Delivery-cursor queries (sqlc adoption T4, RIG-3034). These replace the inline + // SQL literals in internal/store/delivery_cursors.go; the hand-written Store + // methods keep their signatures, the AckDelivery tx orchestration (the owed-clear + // FIRST, the commit-if-cleared arm, the contiguous-advance loop in Go), and the + // D2 seed self-guard/idempotency contract. The two message-fanout reads + // (OwedMentions, UndeliveredMessages) share the per-channel projection the Go + // drains with an inline loop calling messageFromParts. + SeedDeliveryCursor(ctx context.Context, arg SeedDeliveryCursorParams) error SeedHomeChannelMembers(ctx context.Context, arg SeedHomeChannelMembersParams) error + SelfAuthoredSeqsAbove(ctx context.Context, arg SelfAuthoredSeqsAboveParams) ([]int64, error) + SetTopicArchived(ctx context.Context, arg SetTopicArchivedParams) error + SharesVisibleChannel(ctx context.Context, arg SharesVisibleChannelParams) (bool, error) SubscribeConvertedDMParties(ctx context.Context, channelID string) error + // Delivery-consumer read queries (sqlc adoption T4, RIG-3034). These replace the + // inline SQL literals in internal/store/delivery_reads.go; the hand-written Store + // methods keep their signatures, the D1 sweep-set disjunct (kept textually in + // sync with delivery_cursors.sql UndeliveredMessages/InSweepSet), and the D9 + // error mapping. MessageByID shares the message projection the Go drains via + // messageFromParts. + SubscribedAgents(ctx context.Context, arg SubscribedAgentsParams) ([]string, error) + SweepChannels(ctx context.Context, accountID string) ([]string, error) + TopicChannelNames(ctx context.Context, id string) (TopicChannelNamesRow, error) + UndeliveredMessages(ctx context.Context, accountID string) ([]UndeliveredMessagesRow, error) + UnroutedMentionMessages(ctx context.Context, arg UnroutedMentionMessagesParams) ([]UnroutedMentionMessagesRow, error) UpdateAgentParent(ctx context.Context, arg UpdateAgentParentParams) error UpdateChannelPolicy(ctx context.Context, arg UpdateChannelPolicyParams) error + UpdateMessageBlocks(ctx context.Context, arg UpdateMessageBlocksParams) (int64, error) + UpdateMessageBlocksAsAuthor(ctx context.Context, arg UpdateMessageBlocksAsAuthorParams) (UpdateMessageBlocksAsAuthorRow, error) + UpdateTopicLastSeq(ctx context.Context, arg UpdateTopicLastSeqParams) error UpsertChannelMember(ctx context.Context, arg UpsertChannelMemberParams) error } diff --git a/go/internal/store/db/topics.sql.go b/go/internal/store/db/topics.sql.go new file mode 100644 index 000000000..f7ce114be --- /dev/null +++ b/go/internal/store/db/topics.sql.go @@ -0,0 +1,181 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: topics.sql + +package db + +import ( + "context" +) + +const deleteTopic = `-- name: DeleteTopic :exec +DELETE FROM topics WHERE id = $1 +` + +func (q *Queries) DeleteTopic(ctx context.Context, id string) error { + _, err := q.db.Exec(ctx, deleteTopic, id) + return err +} + +const getTopic = `-- name: GetTopic :one +SELECT id, channel_id, name, created_by_account_id, created_at_unix_ms, archived, last_seq +FROM topics WHERE id = $1 +` + +func (q *Queries) GetTopic(ctx context.Context, id string) (Topic, error) { + row := q.db.QueryRow(ctx, getTopic, id) + var i Topic + err := row.Scan( + &i.ID, + &i.ChannelID, + &i.Name, + &i.CreatedByAccountID, + &i.CreatedAtUnixMs, + &i.Archived, + &i.LastSeq, + ) + return i, err +} + +const listTopics = `-- name: ListTopics :many + +SELECT id, channel_id, name, created_by_account_id, created_at_unix_ms, archived, last_seq +FROM topics +WHERE channel_id = $1 AND ($2 OR NOT archived) +ORDER BY last_seq DESC, created_at_unix_ms DESC, id +` + +type ListTopicsParams struct { + ChannelID string + Column2 interface{} +} + +// Topic-domain queries (sqlc adoption T4, RIG-3034). These replace the inline +// SQL literals in internal/store/topics.go; the hand-written Store methods keep +// their signatures, the UpdateTopic tx orchestration, the rename/merge resolution +// loop, and the D9 not-found/forbidden error mapping. The topic projection +// (id, channel_id, name, created_by_account_id, created_at_unix_ms, archived, +// last_seq) matches the former scanTopics order so the Go maps each row to Topic. +func (q *Queries) ListTopics(ctx context.Context, arg ListTopicsParams) ([]Topic, error) { + rows, err := q.db.Query(ctx, listTopics, arg.ChannelID, arg.Column2) + if err != nil { + return nil, err + } + defer rows.Close() + var items []Topic + for rows.Next() { + var i Topic + if err := rows.Scan( + &i.ID, + &i.ChannelID, + &i.Name, + &i.CreatedByAccountID, + &i.CreatedAtUnixMs, + &i.Archived, + &i.LastSeq, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const mergeTopicLastSeq = `-- name: MergeTopicLastSeq :exec +UPDATE topics dst SET last_seq = GREATEST(dst.last_seq, src.last_seq) +FROM topics src WHERE dst.id = $1 AND src.id = $2 +` + +type MergeTopicLastSeqParams struct { + ID string + ID_2 string +} + +func (q *Queries) MergeTopicLastSeq(ctx context.Context, arg MergeTopicLastSeqParams) error { + _, err := q.db.Exec(ctx, mergeTopicLastSeq, arg.ID, arg.ID_2) + return err +} + +const moveMessagesToTopic = `-- name: MoveMessagesToTopic :exec +UPDATE messages SET topic_id = $1 WHERE topic_id = $2 +` + +type MoveMessagesToTopicParams struct { + TopicID string + TopicID_2 string +} + +func (q *Queries) MoveMessagesToTopic(ctx context.Context, arg MoveMessagesToTopicParams) error { + _, err := q.db.Exec(ctx, moveMessagesToTopic, arg.TopicID, arg.TopicID_2) + return err +} + +const renameTopic = `-- name: RenameTopic :exec +UPDATE topics SET name = $2 WHERE id = $1 +` + +type RenameTopicParams struct { + ID string + Name string +} + +func (q *Queries) RenameTopic(ctx context.Context, arg RenameTopicParams) error { + _, err := q.db.Exec(ctx, renameTopic, arg.ID, arg.Name) + return err +} + +const resolveTopicForUpdate = `-- name: ResolveTopicForUpdate :one +SELECT t.channel_id FROM topics t +JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 +WHERE t.id = $2 +FOR UPDATE OF t +` + +type ResolveTopicForUpdateParams struct { + AccountID string + ID string +} + +func (q *Queries) ResolveTopicForUpdate(ctx context.Context, arg ResolveTopicForUpdateParams) (string, error) { + row := q.db.QueryRow(ctx, resolveTopicForUpdate, arg.AccountID, arg.ID) + var channel_id string + err := row.Scan(&channel_id) + return channel_id, err +} + +const resolveTopicRenameTarget = `-- name: ResolveTopicRenameTarget :one +SELECT id FROM topics +WHERE channel_id = $1 AND lower(name) = lower($2) AND id <> $3 +FOR UPDATE +` + +type ResolveTopicRenameTargetParams struct { + ChannelID string + Lower string + ID string +} + +func (q *Queries) ResolveTopicRenameTarget(ctx context.Context, arg ResolveTopicRenameTargetParams) (string, error) { + row := q.db.QueryRow(ctx, resolveTopicRenameTarget, arg.ChannelID, arg.Lower, arg.ID) + var id string + err := row.Scan(&id) + return id, err +} + +const setTopicArchived = `-- name: SetTopicArchived :exec +UPDATE topics SET archived = $2 WHERE id = $1 +` + +type SetTopicArchivedParams struct { + ID string + Archived bool +} + +func (q *Queries) SetTopicArchived(ctx context.Context, arg SetTopicArchivedParams) error { + _, err := q.db.Exec(ctx, setTopicArchived, arg.ID, arg.Archived) + return err +} diff --git a/go/internal/store/delivery_cursors.go b/go/internal/store/delivery_cursors.go index a6cadefb4..03306d2a1 100644 --- a/go/internal/store/delivery_cursors.go +++ b/go/internal/store/delivery_cursors.go @@ -6,60 +6,47 @@ import ( "time" "github.com/jackc/pgx/v5" -) + "github.com/jackc/pgx/v5/pgtype" -// seedDeliveryCursorSQL seeds a per-(agent, channel) delivery cursor to the -// current channel head — MAX(seq) over the channel's messages, 0 if empty — with -// NO history replay (design record D2). It is self-guarding and idempotent in one -// race-free statement: the WHERE EXISTS admits the row only for an agent account -// (a user id yields zero rows, so a non-agent member is a silent no-op rather -// than an FK violation), and ON CONFLICT DO NOTHING means a re-subscribe never -// resets an existing cursor. $1 is the agent account id, $2 the channel id. -const seedDeliveryCursorSQL = ` - INSERT INTO agent_delivery_cursors (agent_account_id, channel_id, acked_seq) - SELECT $1, $2, COALESCE((SELECT MAX(m.seq) FROM messages m JOIN topics t ON t.id = m.topic_id WHERE t.channel_id = $2), 0) - WHERE EXISTS (SELECT 1 FROM agent_accounts WHERE account_id = $1) - ON CONFLICT (agent_account_id, channel_id) DO NOTHING` + "github.com/RigelBuild/compass/go/internal/store/db" +) // seedDeliveryCursor is the shared in-txn seed: it rides the caller's existing // transaction (the channel_members insert txn) so a missed seed is a loud // constraint failure in that same commit, never a silent skip. Called by the two // membership-insert sites (CreateAgent home-channel seed, addOrUpdateMember -// subscribe upsert) and by the exported SeedDeliveryCursor wrapper. Self-guarding -// (see seedDeliveryCursorSQL), so it is safe to call unconditionally for a member -// whose agent-ness is not separately known. +// subscribe upsert) and by the exported SeedDeliveryCursor wrapper. The seed is +// self-guarding and idempotent in one race-free statement (queries/ +// delivery_cursors.sql SeedDeliveryCursor): the WHERE EXISTS admits the row only +// for an agent account (a user id yields zero rows, so a non-agent member is a +// silent no-op rather than an FK violation), it seeds acked_seq to the current +// channel head — MAX(seq) over the channel's messages, 0 if empty — with NO +// history replay (design record D2), and ON CONFLICT DO NOTHING means a +// re-subscribe never resets an existing cursor. Safe to call unconditionally for +// a member whose agent-ness is not separately known. func seedDeliveryCursor(ctx context.Context, tx pgx.Tx, agent AccountID, channel ChannelID) error { - if _, err := tx.Exec(ctx, seedDeliveryCursorSQL, string(agent), string(channel)); err != nil { + if err := db.New(tx).SeedDeliveryCursor(ctx, db.SeedDeliveryCursorParams{ + AgentAccountID: string(agent), + ChannelID: string(channel), + }); err != nil { return fmt.Errorf("store: seed delivery cursor: %w", err) } return nil } -// seedChannelDeliveryCursorsSQL seeds EVERY agent member of the channel to the -// current channel head in one statement — MAX(seq) over the channel's messages, -// 0 if empty, a same-channel constant across all seeded rows — with NO history -// replay (design record D2). The JOIN agent_accounts is the agent-only guard -// (the set form of the per-row WHERE EXISTS in seedDeliveryCursorSQL): a human -// member has no agent_accounts row and so is a silent no-op rather than an FK -// violation. ON CONFLICT DO NOTHING keeps a re-subscribe / re-run idempotent — -// it never resets an existing cursor. $1 is the channel id. -const seedChannelDeliveryCursorsSQL = ` - INSERT INTO agent_delivery_cursors (agent_account_id, channel_id, acked_seq) - SELECT cm.account_id, $1, - COALESCE((SELECT MAX(m.seq) FROM messages m JOIN topics t ON t.id = m.topic_id WHERE t.channel_id = $1), 0) - FROM channel_members cm - JOIN agent_accounts aa ON aa.account_id = cm.account_id - WHERE cm.channel_id = $1 - ON CONFLICT (agent_account_id, channel_id) DO NOTHING` - // seedChannelDeliveryCursors is the set-based counterpart to seedDeliveryCursor: -// it seeds all agent members of the channel in a single statement (collapsing the -// per-member seed loop), riding the caller's existing transaction so a missed -// seed is a loud failure in that same commit. Self-guarding (agent-only, see -// seedChannelDeliveryCursorsSQL) and idempotent, so it is safe to call for a -// channel whose member set includes humans. +// it seeds EVERY agent member of the channel in a single statement (queries/ +// delivery_cursors.sql SeedChannelDeliveryCursors — the set form of the per-row +// seed, collapsing the per-member seed loop), riding the caller's existing +// transaction so a missed seed is a loud failure in that same commit. The JOIN +// agent_accounts is the agent-only guard (a human member has no agent_accounts +// row and so is a silent no-op rather than an FK violation), each row is seeded +// to the current channel head with NO history replay (design record D2), and +// ON CONFLICT DO NOTHING keeps a re-subscribe / re-run idempotent — it never +// resets an existing cursor. Safe to call for a channel whose member set includes +// humans. func seedChannelDeliveryCursors(ctx context.Context, tx pgx.Tx, channel ChannelID) error { - if _, err := tx.Exec(ctx, seedChannelDeliveryCursorsSQL, string(channel)); err != nil { + if err := db.New(tx).SeedChannelDeliveryCursors(ctx, string(channel)); err != nil { return fmt.Errorf("store: seed channel delivery cursors: %w", err) } return nil @@ -88,12 +75,12 @@ func (s *Store) SeedDeliveryCursor(ctx context.Context, tx pgx.Tx, agent Account // persist a silently-inconsistent row. The settle-edge caller already holds the // message's channel, so this is an assertion, not a lookup. func (s *Store) RecordOwedMention(ctx context.Context, agent AccountID, channel ChannelID, messageID string) error { - if _, err := s.pool.Exec(ctx, - `INSERT INTO owed_mentions (agent_account_id, message_id, channel_id, recorded_at_unix_ms) - VALUES ($1, $2, $3, $4) - ON CONFLICT (agent_account_id, message_id) DO NOTHING`, - string(agent), messageID, string(channel), time.Now().UnixMilli(), - ); err != nil { + if err := s.q.RecordOwedMention(ctx, db.RecordOwedMentionParams{ + AgentAccountID: string(agent), + MessageID: messageID, + ChannelID: string(channel), + RecordedAtUnixMs: time.Now().UnixMilli(), + }); err != nil { return fmt.Errorf("store: record owed mention: %w", err) } return nil @@ -106,53 +93,17 @@ func (s *Store) RecordOwedMention(ctx context.Context, agent AccountID, channel // also removes it, so this is belt-and-suspenders). Channels with no owed // messages are omitted from the map. func (s *Store) OwedMentions(ctx context.Context, agent AccountID) (map[ChannelID][]Message, error) { - const q = ` - SELECT m.id, m.topic_id, t.channel_id, m.author_account_id, m.at_unix_ms, m.blocks - FROM owed_mentions om - JOIN messages m ON m.id = om.message_id - JOIN topics t ON t.id = m.topic_id - WHERE om.agent_account_id = $1 - ORDER BY t.channel_id, m.seq ASC` - rows, err := s.pool.Query(ctx, q, string(agent)) + rows, err := s.q.OwedMentions(ctx, string(agent)) if err != nil { return nil, fmt.Errorf("store: read owed mentions: %w", err) } - defer rows.Close() - return scanMessagesByChannel(rows, "owed mention") -} - -// scanMessagesByChannel drains rows of the shared per-channel message projection -// (m.id, m.topic_id, t.channel_id, m.author_account_id, m.at_unix_ms, m.blocks, -// ordered by channel then seq) into a channel-keyed map — the scan half shared by -// UndeliveredMessages (the cursor sweep) and OwedMentions (the mention-gap -// backstop), which differ only in their query. `what` names the row in error -// messages ("undelivered message" / "owed mention"). Channels with no rows are -// absent from the map. -func scanMessagesByChannel(rows pgx.Rows, what string) (map[ChannelID][]Message, error) { out := make(map[ChannelID][]Message) - for rows.Next() { - var ( - id, topicID, channelID, author string - atMS int64 - blocksJSON []byte - ) - if err := rows.Scan(&id, &topicID, &channelID, &author, &atMS, &blocksJSON); err != nil { - return nil, fmt.Errorf("store: scan %s: %w", what, err) - } - blocks, err := unmarshalBlocks(blocksJSON) + for _, r := range rows { + m, err := messageFromParts(r.ID, r.TopicID, r.AuthorAccountID, r.AtUnixMs, r.Blocks) if err != nil { return nil, err } - out[ChannelID(channelID)] = append(out[ChannelID(channelID)], Message{ - ID: MessageID(id), - TopicID: topicID, - AuthorAccountID: AccountID(author), - At: time.UnixMilli(atMS).UTC(), - Blocks: blocks, - }) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate %ss: %w", what, err) + out[ChannelID(r.ChannelID)] = append(out[ChannelID(r.ChannelID)], m) } return out, nil } @@ -162,14 +113,14 @@ func scanMessagesByChannel(rows pgx.Rows, what string) (map[ChannelID][]Message, // state, and reports whether a row was actually deleted. Clearing an absent row // is a no-op (cleared=false, err=nil). func (s *Store) clearOwedMention(ctx context.Context, tx pgx.Tx, agent AccountID, messageID string) (cleared bool, err error) { - tag, err := tx.Exec(ctx, - `DELETE FROM owed_mentions WHERE agent_account_id = $1 AND message_id = $2`, - string(agent), messageID, - ) + affected, err := db.New(tx).ClearOwedMention(ctx, db.ClearOwedMentionParams{ + AgentAccountID: string(agent), + MessageID: messageID, + }) if err != nil { return false, fmt.Errorf("store: clear owed mention: %w", err) } - return tag.RowsAffected() > 0, nil + return affected > 0, nil } // ClearOwedMention deletes the owed row for (agent, messageID) using the pool @@ -178,10 +129,10 @@ func (s *Store) clearOwedMention(ctx context.Context, tx pgx.Tx, agent AccountID // owed message this way so a vanished message stops re-logging on every start. // Clearing an absent row is a no-op. func (s *Store) ClearOwedMention(ctx context.Context, agent AccountID, messageID string) error { - if _, err := s.pool.Exec(ctx, - `DELETE FROM owed_mentions WHERE agent_account_id = $1 AND message_id = $2`, - string(agent), messageID, - ); err != nil { + if _, err := s.q.ClearOwedMention(ctx, db.ClearOwedMentionParams{ + AgentAccountID: string(agent), + MessageID: messageID, + }); err != nil { return fmt.Errorf("store: clear owed mention: %w", err) } return nil @@ -191,11 +142,11 @@ func (s *Store) ClearOwedMention(ctx context.Context, agent AccountID, messageID // agents — the startup visibility count (RIG-1641 T2 observability) so a // silently-growing owed table is surfaced. func (s *Store) CountOwedMentions(ctx context.Context) (int, error) { - var n int - if err := s.pool.QueryRow(ctx, `SELECT COUNT(*) FROM owed_mentions`).Scan(&n); err != nil { + n, err := s.q.CountOwedMentions(ctx) + if err != nil { return 0, fmt.Errorf("store: count owed mentions: %w", err) } - return n, nil + return int(n), nil } // MessageWithChannel is a message plus its resolved channel and store-space seq @@ -214,10 +165,10 @@ type MessageWithChannel struct { // timestamp; the contract readers rely on is NULL vs non-NULL only (RIG-2490 // T1). Marking an unknown id is a no-op. func (s *Store) MarkMentionsRouted(ctx context.Context, messageID string) error { - if _, err := s.pool.Exec(ctx, - `UPDATE messages SET mentions_routed_at = $1 WHERE id = $2`, - time.Now().UnixMilli(), messageID, - ); err != nil { + if err := s.q.MarkMentionsRouted(ctx, db.MarkMentionsRoutedParams{ + MentionsRoutedAt: pgtype.Int8{Int64: time.Now().UnixMilli(), Valid: true}, + ID: messageID, + }); err != nil { return fmt.Errorf("store: mark mentions routed: %w", err) } return nil @@ -233,47 +184,25 @@ func (s *Store) MarkMentionsRouted(ctx context.Context, messageID string) error // the caller stays NULL and is re-scanned from 0 at the next recovery point, so // this is not the killed high-water. func (s *Store) UnroutedMentionMessages(ctx context.Context, afterSeq int64, limit int) ([]MessageWithChannel, error) { - const q = ` - SELECT m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks, t.channel_id, m.seq - FROM messages m - JOIN topics t ON t.id = m.topic_id - WHERE m.mentions_routed_at IS NULL AND m.seq > $1 - ORDER BY m.seq ASC - LIMIT $2` - rows, err := s.pool.Query(ctx, q, afterSeq, limit) + rows, err := s.q.UnroutedMentionMessages(ctx, db.UnroutedMentionMessagesParams{ + Seq: afterSeq, + Limit: int32(limit), //nolint:gosec // G115: caller passes a small fixed batch size + }) if err != nil { return nil, fmt.Errorf("store: read unrouted mention messages: %w", err) } - defer rows.Close() var out []MessageWithChannel - for rows.Next() { - var ( - id, topicID, author, channelID string - atMS, seq int64 - blocksJSON []byte - ) - if err := rows.Scan(&id, &topicID, &author, &atMS, &blocksJSON, &channelID, &seq); err != nil { - return nil, fmt.Errorf("store: scan unrouted mention message: %w", err) - } - blocks, err := unmarshalBlocks(blocksJSON) + for _, r := range rows { + m, err := messageFromParts(r.ID, r.TopicID, r.AuthorAccountID, r.AtUnixMs, r.Blocks) if err != nil { return nil, err } out = append(out, MessageWithChannel{ - Message: Message{ - ID: MessageID(id), - TopicID: topicID, - AuthorAccountID: AccountID(author), - At: time.UnixMilli(atMS).UTC(), - Blocks: blocks, - }, - Channel: ChannelID(channelID), - Seq: seq, + Message: m, + Channel: ChannelID(r.ChannelID), + Seq: r.Seq, }) } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate unrouted mention messages: %w", err) - } return out, nil } @@ -314,11 +243,9 @@ func (s *Store) AckDelivery(ctx context.Context, agent AccountID, channel Channe // Resolve messageID → seq scoped to THIS channel. A message id that names no // row in this channel (fabricated, foreign, or in another channel) resolves // to no row: the overshoot clamp — a fabricated id cannot advance the cursor. - var seq int64 - switch err := tx.QueryRow(ctx, - `SELECT m.seq FROM messages m JOIN topics t ON t.id = m.topic_id WHERE m.id = $1 AND t.channel_id = $2`, - messageID, string(channel), - ).Scan(&seq); { + qtx := db.New(tx) + seq, err := qtx.ResolveAckMessage(ctx, db.ResolveAckMessageParams{ID: messageID, ChannelID: string(channel)}) + switch { case noRows(err): // Never dispatched to this (agent, channel): a no-op. The owed row (if // any) is keyed (agent, message_id) for a valid ack of THIS channel, and @@ -352,21 +279,18 @@ func (s *Store) AckDelivery(ctx context.Context, agent AccountID, channel Channe // Load the current cursor. An absent row means no cursor was seeded for this // (agent, channel) — the mention-gap population — so there is nothing to // advance; commit the owed-clear if it did work, otherwise no-op. - var ( - ackedSeq int64 - aboveSeqs []int64 - ) - switch err := tx.QueryRow(ctx, - `SELECT acked_seq, above_seqs FROM agent_delivery_cursors - WHERE agent_account_id = $1 AND channel_id = $2 - FOR UPDATE`, - string(agent), string(channel), - ).Scan(&ackedSeq, &aboveSeqs); { + cursor, err := qtx.LoadDeliveryCursor(ctx, db.LoadDeliveryCursorParams{ + AgentAccountID: string(agent), + ChannelID: string(channel), + }) + switch { case noRows(err): return commitIfCleared() // no cursor to advance; commit the clear if any. case err != nil: return fmt.Errorf("store: load delivery cursor: %w", err) } + ackedSeq := cursor.AckedSeq + aboveSeqs := cursor.AboveSeqs // A duplicate or reordered ack (at or below the contiguous cursor) advances // nothing; commit the owed-clear if it did work, otherwise no-op. @@ -392,27 +316,18 @@ func (s *Store) AckDelivery(ctx context.Context, agent AccountID, channel Channe // multi-channel deployment acked seqs above such a gap remain in above_seqs // rather than draining. That boundedness gap is the parked design question // (PR #55 Open Questions); correctness (no message loss) is unaffected. - rows, err := tx.Query(ctx, - `SELECT m.seq FROM messages m JOIN topics t ON t.id = m.topic_id - WHERE t.channel_id = $1 AND m.seq > $2 AND m.author_account_id = $3`, - string(channel), ackedSeq, string(agent), - ) + ownSeqList, err := qtx.SelfAuthoredSeqsAbove(ctx, db.SelfAuthoredSeqsAboveParams{ + ChannelID: string(channel), + Seq: ackedSeq, + AuthorAccountID: string(agent), + }) if err != nil { return fmt.Errorf("store: load self-authored seqs: %w", err) } - ownSeqs := make(map[int64]bool) - for rows.Next() { - var s int64 - if err := rows.Scan(&s); err != nil { - rows.Close() - return fmt.Errorf("store: scan self-authored seq: %w", err) - } + ownSeqs := make(map[int64]bool, len(ownSeqList)) + for _, s := range ownSeqList { ownSeqs[s] = true } - rows.Close() - if err := rows.Err(); err != nil { - return fmt.Errorf("store: iterate self-authored seqs: %w", err) - } for { next := ackedSeq + 1 @@ -433,12 +348,12 @@ func (s *Store) AckDelivery(ctx context.Context, agent AccountID, channel Channe } } - if _, err := tx.Exec(ctx, - `UPDATE agent_delivery_cursors - SET acked_seq = $3, above_seqs = $4, acked_at = now() - WHERE agent_account_id = $1 AND channel_id = $2`, - string(agent), string(channel), ackedSeq, remaining, - ); err != nil { + if err := qtx.AdvanceDeliveryCursor(ctx, db.AdvanceDeliveryCursorParams{ + AgentAccountID: string(agent), + ChannelID: string(channel), + AckedSeq: ackedSeq, + AboveSeqs: remaining, + }); err != nil { return fmt.Errorf("store: advance delivery cursor: %w", err) } @@ -470,29 +385,19 @@ func (s *Store) UndeliveredMessages(ctx context.Context, agent AccountID) (map[C // cursor predicate admits nothing (caught-up, no replay). author_account_id // <> agent excludes the agent's own posts; the array predicate excludes the // retained above-set. - const q = ` - SELECT m.id, m.topic_id, t.channel_id, m.author_account_id, m.at_unix_ms, m.blocks - FROM channel_members cm - JOIN agent_accounts aa ON aa.account_id = cm.account_id - JOIN topics t ON t.channel_id = cm.channel_id - JOIN messages m ON m.topic_id = t.id - JOIN channels ch ON ch.id = cm.channel_id - LEFT JOIN agent_delivery_cursors dc - ON dc.agent_account_id = cm.account_id AND dc.channel_id = cm.channel_id - WHERE cm.account_id = $1 - AND (cm.subscribed OR cm.channel_id = aa.home_channel_id OR ch.mandatory_subscription) - AND m.author_account_id <> $1 - AND m.seq > COALESCE( - dc.acked_seq, - (SELECT COALESCE(MAX(mh.seq), 0) FROM messages mh JOIN topics th ON th.id = mh.topic_id WHERE th.channel_id = cm.channel_id)) - AND m.seq <> ALL(COALESCE(dc.above_seqs, '{}'::BIGINT[])) - ORDER BY t.channel_id, m.seq ASC` - rows, err := s.pool.Query(ctx, q, string(agent)) + rows, err := s.q.UndeliveredMessages(ctx, string(agent)) if err != nil { return nil, fmt.Errorf("store: sweep undelivered messages: %w", err) } - defer rows.Close() - return scanMessagesByChannel(rows, "undelivered message") + out := make(map[ChannelID][]Message) + for _, r := range rows { + m, err := messageFromParts(r.ID, r.TopicID, r.AuthorAccountID, r.AtUnixMs, r.Blocks) + if err != nil { + return nil, err + } + out[ChannelID(r.ChannelID)] = append(out[ChannelID(r.ChannelID)], m) + } + return out, nil } // InSweepSet reports whether agent is in channel's D2 sweep set — subscribed OR @@ -501,17 +406,11 @@ func (s *Store) UndeliveredMessages(ctx context.Context, agent AccountID) (map[C // backstop, so an offline mention to it needs a durable owed_mentions row (T1). // The disjunct mirrors UndeliveredMessages EXACTLY so the two never drift. func (s *Store) InSweepSet(ctx context.Context, agent AccountID, channel ChannelID) (bool, error) { - const q = ` - SELECT EXISTS( - SELECT 1 - FROM channel_members cm - JOIN agent_accounts aa ON aa.account_id = cm.account_id - JOIN channels ch ON ch.id = cm.channel_id - WHERE cm.account_id = $1 - AND cm.channel_id = $2 - AND (cm.subscribed OR cm.channel_id = aa.home_channel_id OR ch.mandatory_subscription))` - var in bool - if err := s.pool.QueryRow(ctx, q, string(agent), string(channel)).Scan(&in); err != nil { + in, err := s.q.InSweepSet(ctx, db.InSweepSetParams{ + AccountID: string(agent), + ChannelID: string(channel), + }) + if err != nil { return false, fmt.Errorf("store: in sweep set: %w", err) } return in, nil diff --git a/go/internal/store/delivery_reads.go b/go/internal/store/delivery_reads.go index 7764a14d5..c079add3b 100644 --- a/go/internal/store/delivery_reads.go +++ b/go/internal/store/delivery_reads.go @@ -3,6 +3,8 @@ package store import ( "context" "fmt" + + "github.com/RigelBuild/compass/go/internal/store/db" ) // The delivery consumer's read side (RIG-1569 T3, design record D1). These live @@ -27,33 +29,14 @@ import ( // and is excluded, so a deliver is only ever dispatched to an agent session. $1 // is the channel, $2 the author account excluded from the result. func (s *Store) SubscribedAgents(ctx context.Context, channel ChannelID, author AccountID) ([]AccountID, error) { - const q = ` - SELECT aa.account_id - FROM channel_members cm - JOIN agent_accounts aa ON aa.account_id = cm.account_id - JOIN channels ch ON ch.id = cm.channel_id - WHERE cm.channel_id = $1 - AND (cm.subscribed OR cm.channel_id = aa.home_channel_id OR ch.mandatory_subscription) - AND cm.account_id <> $2 - ORDER BY aa.account_id` - rows, err := s.pool.Query(ctx, q, string(channel), string(author)) + rows, err := s.q.SubscribedAgents(ctx, db.SubscribedAgentsParams{ + ChannelID: string(channel), + AccountID: string(author), + }) if err != nil { return nil, fmt.Errorf("store: resolve subscribed agents: %w", err) } - defer rows.Close() - - var agents []AccountID - for rows.Next() { - var acct string - if err := rows.Scan(&acct); err != nil { - return nil, fmt.Errorf("store: scan subscribed agent: %w", err) - } - agents = append(agents, AccountID(acct)) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate subscribed agents: %w", err) - } - return agents, nil + return accountIDs(rows), nil } // ChannelAgentMembers resolves every AGENT member of a channel, author excluded, @@ -65,31 +48,14 @@ func (s *Store) SubscribedAgents(ctx context.Context, channel ChannelID, author // members (a human member has no agent_accounts row); $1 is the channel, $2 the // author excluded (an agent's own `@agents` / self-mention never steers itself). func (s *Store) ChannelAgentMembers(ctx context.Context, channel ChannelID, author AccountID) ([]AccountID, error) { - const q = ` - SELECT aa.account_id - FROM channel_members cm - JOIN agent_accounts aa ON aa.account_id = cm.account_id - WHERE cm.channel_id = $1 - AND cm.account_id <> $2 - ORDER BY aa.account_id` - rows, err := s.pool.Query(ctx, q, string(channel), string(author)) + rows, err := s.q.ChannelAgentMembers(ctx, db.ChannelAgentMembersParams{ + ChannelID: string(channel), + AccountID: string(author), + }) if err != nil { return nil, fmt.Errorf("store: resolve channel agent members: %w", err) } - defer rows.Close() - - var agents []AccountID - for rows.Next() { - var acct string - if err := rows.Scan(&acct); err != nil { - return nil, fmt.Errorf("store: scan channel agent member: %w", err) - } - agents = append(agents, AccountID(acct)) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate channel agent members: %w", err) - } - return agents, nil + return accountIDs(rows), nil } // IsAgentAccount reports whether account is an owned agent (has an agent_accounts @@ -100,11 +66,8 @@ func (s *Store) ChannelAgentMembers(ctx context.Context, channel ChannelID, auth // caller (the consumer, resolving a message's author) treats "not an agent" and // "unknown" identically: deliver at post. func (s *Store) IsAgentAccount(ctx context.Context, account AccountID) (bool, error) { - var exists bool - if err := s.pool.QueryRow(ctx, - `SELECT EXISTS (SELECT 1 FROM agent_accounts WHERE account_id = $1)`, - string(account), - ).Scan(&exists); err != nil { + exists, err := s.q.IsAgentAccount(ctx, string(account)) + if err != nil { return false, fmt.Errorf("store: check agent account: %w", err) } return exists, nil @@ -116,23 +79,14 @@ func (s *Store) IsAgentAccount(ctx context.Context, account AccountID) (bool, er // stale in-memory copy — the commit-lag-safe read the settle gate rides. An // unknown id is ErrNotFound. func (s *Store) MessageByID(ctx context.Context, messageID string) (Message, error) { - const q = ` - SELECT id, topic_id, author_account_id, at_unix_ms, blocks - FROM messages - WHERE id = $1` - rows, err := s.pool.Query(ctx, q, messageID) + row, err := s.q.MessageByID(ctx, messageID) if err != nil { + if noRows(err) { + return Message{}, fmt.Errorf("%w: message %q", ErrNotFound, messageID) + } return Message{}, fmt.Errorf("store: read message by id: %w", err) } - defer rows.Close() - msgs, err := scanMessages(rows) - if err != nil { - return Message{}, err - } - if len(msgs) == 0 { - return Message{}, fmt.Errorf("%w: message %q", ErrNotFound, messageID) - } - return msgs[0], nil + return messageFromParts(row.ID, row.TopicID, row.AuthorAccountID, row.AtUnixMs, row.Blocks) } // MessageChannel resolves a message id to its channel — the ack arm's channel @@ -144,10 +98,8 @@ func (s *Store) MessageByID(ctx context.Context, messageID string) (Message, err // ErrNotFound, which the ack arm treats as a fail-closed no-op (a foreign or // fabricated ack never advances a cursor). func (s *Store) MessageChannel(ctx context.Context, messageID string) (ChannelID, error) { - var channel string - if err := s.pool.QueryRow(ctx, - `SELECT t.channel_id FROM messages m JOIN topics t ON t.id = m.topic_id WHERE m.id = $1`, messageID, - ).Scan(&channel); err != nil { + channel, err := s.q.MessageChannel(ctx, messageID) + if err != nil { if noRows(err) { return "", fmt.Errorf("%w: message %q", ErrNotFound, messageID) } @@ -164,15 +116,14 @@ func (s *Store) MessageChannel(ctx context.Context, messageID string) (ChannelID // author from_handle (GetAccount). An unknown topic id is ErrNotFound, which the // caller logs and treats as empty names: a name miss never blocks a delivery. func (s *Store) TopicChannelNames(ctx context.Context, topicID string) (topicName, channelName string, err error) { - if err := s.pool.QueryRow(ctx, - `SELECT t.name, c.name FROM topics t JOIN channels c ON c.id = t.channel_id WHERE t.id = $1`, topicID, - ).Scan(&topicName, &channelName); err != nil { + row, err := s.q.TopicChannelNames(ctx, topicID) + if err != nil { if noRows(err) { return "", "", fmt.Errorf("%w: topic %q", ErrNotFound, topicID) } return "", "", fmt.Errorf("store: resolve topic channel names: %w", err) } - return topicName, channelName, nil + return row.TopicName, row.ChannelName, nil } // SweepChannels returns the D1 disjunct channel set an agent sweeps: every @@ -188,30 +139,29 @@ func (s *Store) TopicChannelNames(ctx context.Context, topicID string) (topicNam // the cursor sweep's cannot drift. $1 is always an agent, so the JOIN to // agent_accounts matches exactly one row and yields its home_channel_id. func (s *Store) SweepChannels(ctx context.Context, agent AccountID) ([]ChannelID, error) { - const q = ` - SELECT cm.channel_id - FROM channel_members cm - JOIN agent_accounts aa ON aa.account_id = cm.account_id - JOIN channels ch ON ch.id = cm.channel_id - WHERE cm.account_id = $1 - AND (cm.subscribed OR cm.channel_id = aa.home_channel_id OR ch.mandatory_subscription) - ORDER BY cm.channel_id` - rows, err := s.pool.Query(ctx, q, string(agent)) + rows, err := s.q.SweepChannels(ctx, string(agent)) if err != nil { return nil, fmt.Errorf("store: resolve sweep channels: %w", err) } - defer rows.Close() - - var channels []ChannelID - for rows.Next() { - var ch string - if err := rows.Scan(&ch); err != nil { - return nil, fmt.Errorf("store: scan sweep channel: %w", err) - } - channels = append(channels, ChannelID(ch)) + if len(rows) == 0 { + return nil, nil } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate sweep channels: %w", err) + channels := make([]ChannelID, 0, len(rows)) + for _, ch := range rows { + channels = append(channels, ChannelID(ch)) } return channels, nil } + +// accountIDs maps a generated string-column read to the domain AccountID slice, +// preserving the former nil-on-empty result the pgx scan loops returned. +func accountIDs(rows []string) []AccountID { + if len(rows) == 0 { + return nil + } + out := make([]AccountID, 0, len(rows)) + for _, a := range rows { + out = append(out, AccountID(a)) + } + return out +} diff --git a/go/internal/store/messages.go b/go/internal/store/messages.go index 8255a42ab..dce2ffb21 100644 --- a/go/internal/store/messages.go +++ b/go/internal/store/messages.go @@ -8,6 +8,8 @@ import ( "time" "github.com/jackc/pgx/v5" + + "github.com/RigelBuild/compass/go/internal/store/db" ) // AppendMessage stores a new message under a topic in channelID, assigning the @@ -63,21 +65,15 @@ func (s *Store) AppendMessage(ctx context.Context, m Message, channelID string, // not-found/forbidden merge), so the policy leaks no oracle: a member who // may not post is indistinguishable from a non-member. Checked in this same // tx as the membership gate and the insert, under the committed policy. - var ( - postPolicy int32 - ownerAcct string - ) - if err := tx.QueryRow(ctx, - "SELECT post_policy, COALESCE(owner_account_id, '') FROM channels WHERE id = $1", - channelID, - ).Scan(&postPolicy, &ownerAcct); err != nil { + policy, err := db.New(tx).GetChannelPostPolicy(ctx, channelID) + if err != nil { if noRows(err) { return Message{}, false, fmt.Errorf("%w: channel %q", ErrNotFound, channelID) } return Message{}, false, fmt.Errorf("store: read channel post policy: %w", err) } - if ChannelPostPolicy(postPolicy) == ChannelPostPolicyOwnerOnly && - string(m.AuthorAccountID) != ownerAcct { + if ChannelPostPolicy(policy.PostPolicy) == ChannelPostPolicyOwnerOnly && + string(m.AuthorAccountID) != policy.OwnerAccountID { return Message{}, false, fmt.Errorf("%w: channel %q", ErrNotFound, channelID) } @@ -138,21 +134,16 @@ func insertMessageTx(ctx context.Context, tx pgx.Tx, m Message, topicID string, } id := newID() at := time.Now().UTC() - const q = ` - INSERT INTO messages (id, topic_id, author_account_id, at_unix_ms, blocks, text_content, client_request_id) - VALUES ($1, $2, $3, $4, $5, $6, $7) - ON CONFLICT (author_account_id, client_request_id) WHERE client_request_id <> '' - DO NOTHING - RETURNING id, at_unix_ms, seq` - var ( - storedID string - atMS int64 - seq int64 - ) - err = tx.QueryRow(ctx, q, - id, topicID, string(m.AuthorAccountID), - at.UnixMilli(), blocksJSON, textContent(m.Blocks), clientRequestID, - ).Scan(&storedID, &atMS, &seq) + q := db.New(tx) + row, err := q.InsertMessage(ctx, db.InsertMessageParams{ + ID: id, + TopicID: topicID, + AuthorAccountID: string(m.AuthorAccountID), + AtUnixMs: at.UnixMilli(), + Blocks: blocksJSON, + TextContent: textContent(m.Blocks), + ClientRequestID: clientRequestID, + }) switch { case noRows(err): return Message{}, errMessageInsertConflict @@ -163,16 +154,13 @@ func insertMessageTx(ctx context.Context, tx pgx.Tx, m Message, topicID string, // Maintain the topic's denormalized activity marker in the same tx. GREATEST // so two concurrent appends to one topic converge on the higher seq // regardless of commit order (the row-lock serializes the two updates). - if _, err := tx.Exec(ctx, - `UPDATE topics SET last_seq = GREATEST(last_seq, $2) WHERE id = $1`, - topicID, seq, - ); err != nil { + if err := q.UpdateTopicLastSeq(ctx, db.UpdateTopicLastSeqParams{ID: topicID, LastSeq: row.Seq}); err != nil { return Message{}, fmt.Errorf("store: update topic last_seq: %w", err) } - m.ID = MessageID(storedID) + m.ID = MessageID(row.ID) m.TopicID = topicID - m.At = time.UnixMilli(atMS).UTC() + m.At = time.UnixMilli(row.AtUnixMs).UTC() return m, nil } @@ -192,11 +180,10 @@ func insertMessageTx(ctx context.Context, tx pgx.Tx, m Message, topicID string, // archived topic clears its archived flag in the same tx (archive is a tidiness // flag, not a lock — a post at a tidied-away name revives the conversation). func resolveTopicForAppend(ctx context.Context, tx pgx.Tx, channelID string, topic TopicRef, author AccountID, atMS int64) (string, error) { + q := db.New(tx) if topic.ID != "" { - var topicChannelID string - switch err := tx.QueryRow(ctx, - `SELECT channel_id FROM topics WHERE id = $1`, topic.ID, - ).Scan(&topicChannelID); { + topicChannelID, err := q.GetTopicChannel(ctx, topic.ID) + switch { case noRows(err): return "", fmt.Errorf("%w: topic %q is not in this channel", ErrInvalidArgument, topic.ID) case err != nil: @@ -215,26 +202,21 @@ func resolveTopicForAppend(ctx context.Context, tx pgx.Tx, channelID string, top // When unset, the mint is skipped entirely — a name that resolves to no row // below is ErrNotFound. if topic.Create { - if _, err := tx.Exec(ctx, - `INSERT INTO topics (id, channel_id, name, created_by_account_id, created_at_unix_ms) - VALUES ($1, $2, $3, $4, $5) - ON CONFLICT (channel_id, lower(name)) DO NOTHING`, - newID(), channelID, topic.Name, string(author), atMS, - ); err != nil { + if err := q.InsertTopicIgnore(ctx, db.InsertTopicIgnoreParams{ + ID: newID(), + ChannelID: channelID, + Name: topic.Name, + CreatedByAccountID: string(author), + CreatedAtUnixMs: atMS, + }); err != nil { if pgErrIs(err, pgForeignKeyViolation) { return "", fmt.Errorf("%w: unknown channel %q or author %q", ErrInvalidArgument, channelID, author) } return "", fmt.Errorf("store: get-or-create topic: %w", err) } } - var ( - topicID string - archived bool - ) - if err := tx.QueryRow(ctx, - `SELECT id, archived FROM topics WHERE channel_id = $1 AND lower(name) = lower($2)`, - channelID, topic.Name, - ).Scan(&topicID, &archived); err != nil { + row, err := q.GetTopicByName(ctx, db.GetTopicByNameParams{ChannelID: channelID, Lower: topic.Name}) + if err != nil { if noRows(err) { // Create was unset (or a racing delete removed the row) and no topic // carries this name: an in-band ErrNotFound, never a silent mint. The @@ -243,14 +225,12 @@ func resolveTopicForAppend(ctx context.Context, tx pgx.Tx, channelID string, top } return "", fmt.Errorf("store: resolve topic: %w", err) } - if archived { - if _, err := tx.Exec(ctx, - `UPDATE topics SET archived = FALSE WHERE id = $1`, topicID, - ); err != nil { + if row.Archived { + if err := q.ReviveTopic(ctx, row.ID); err != nil { return "", fmt.Errorf("store: revive archived topic: %w", err) } } - return topicID, nil + return row.ID, nil } // MessagesHeadSeq returns the current head of the message sequence — the @@ -273,13 +253,11 @@ func resolveTopicForAppend(ctx context.Context, tx pgx.Tx, channelID string, top // count-metadata exposure is accepted as within the threat model, not a leak to // close by scoping the boundary — that would be a different token (RIG-1333 OQ4). func (s *Store) MessagesHeadSeq(ctx context.Context) (uint64, error) { - var head uint64 - if err := s.pool.QueryRow(ctx, - `SELECT COALESCE(MAX(seq), 0) FROM messages`, - ).Scan(&head); err != nil { + head, err := s.q.MessagesHeadSeq(ctx) + if err != nil { return 0, fmt.Errorf("store: read messages head seq: %w", err) } - return head, nil + return uint64(head), nil //nolint:gosec // G115: head is COALESCE(MAX(seq), 0), always >= 0 } // updateMessageBlocksExec is the shared block-write core, run against @@ -293,7 +271,7 @@ func (s *Store) MessagesHeadSeq(ctx context.Context) (uint64, error) { // // It performs NO membership or authorship check, so it is deliberately // unexported: every exported write path must authorize before reaching it. -func updateMessageBlocksExec(ctx context.Context, db execer, id MessageID, blocks []MessageBlock) error { +func updateMessageBlocksExec(ctx context.Context, dbtx db.DBTX, id MessageID, blocks []MessageBlock) error { if len(blocks) == 0 { return fmt.Errorf("%w: message has no blocks", ErrInvalidArgument) } @@ -306,14 +284,15 @@ func updateMessageBlocksExec(ctx context.Context, db execer, id MessageID, block if err != nil { return err } - tag, err := db.Exec(ctx, - "UPDATE messages SET blocks = $1, text_content = $2 WHERE id = $3", - blocksJSON, textContent(blocks), string(id), - ) + affected, err := db.New(dbtx).UpdateMessageBlocks(ctx, db.UpdateMessageBlocksParams{ + Blocks: blocksJSON, + TextContent: textContent(blocks), + ID: string(id), + }) if err != nil { return fmt.Errorf("store: update message blocks: %w", err) } - if tag.RowsAffected() == 0 { + if affected == 0 { return fmt.Errorf("%w: message %q", ErrNotFound, id) } return nil @@ -382,33 +361,21 @@ func (s *Store) UpdateMessageBlocksAsAuthor(ctx context.Context, actor AccountID // matches no row) or after it, never between a separate check and the write. // The EXISTS subquery is the membership half and the author_account_id // equality the authorship half; both must hold for the row to match. - const q = ` - UPDATE messages m - SET blocks = $1, text_content = $2 - FROM topics t - WHERE m.id = $3 - AND t.id = m.topic_id - AND m.author_account_id = $4 - AND EXISTS ( - SELECT 1 FROM channel_members cm - WHERE cm.channel_id = t.channel_id AND cm.account_id = $4 - ) - RETURNING m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks` - rows, err := s.pool.Query(ctx, q, blocksJSON, textContent(blocks), string(id), string(actor)) + row, err := s.q.UpdateMessageBlocksAsAuthor(ctx, db.UpdateMessageBlocksAsAuthorParams{ + Blocks: blocksJSON, + TextContent: textContent(blocks), + ID: string(id), + AuthorAccountID: string(actor), + }) if err != nil { + if noRows(err) { + // Unknown id, not the author, or no longer a member — one answer for + // all three, so a refusal enumerates nothing. + return Message{}, fmt.Errorf("%w: message %q", ErrNotFound, id) + } return Message{}, fmt.Errorf("store: update message blocks as author: %w", err) } - defer rows.Close() - msgs, err := scanMessages(rows) - if err != nil { - return Message{}, err - } - if len(msgs) == 0 { - // Unknown id, not the author, or no longer a member — one answer for all - // three, so a refusal enumerates nothing. - return Message{}, fmt.Errorf("%w: message %q", ErrNotFound, id) - } - return msgs[0], nil + return messageFromParts(row.ID, row.TopicID, row.AuthorAccountID, row.AtUnixMs, row.Blocks) } // MessageAskIDs returns the ask_id of every ask block on the message, in block @@ -429,10 +396,8 @@ func (s *Store) UpdateMessageBlocksAsAuthor(ctx context.Context, actor AccountID // single-statement UpdateMessageBlocksAsAuthor that follows, and its result is // never derived from what this read returned. func (s *Store) MessageAskIDs(ctx context.Context, id MessageID) ([]string, error) { - var blocksJSON []byte - if err := s.pool.QueryRow(ctx, - `SELECT blocks FROM messages WHERE id = $1`, string(id), - ).Scan(&blocksJSON); err != nil { + blocksJSON, err := s.q.GetMessageBlocks(ctx, string(id)) + if err != nil { if noRows(err) { return nil, nil } @@ -480,19 +445,18 @@ func (s *Store) ListMessages(ctx context.Context, q ListMessagesQuery) ([]Messag // visibility boundary (the D9 not-found/forbidden merge the main query and // AnswerAsk also apply). The channel is the cursor message's topic's // channel. - err := s.pool.QueryRow(ctx, - `SELECT m.seq FROM messages m - JOIN topics t ON t.id = m.topic_id - JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 - WHERE m.id = $2 AND t.channel_id = $3`, - string(q.Actor), string(q.Page.BeforeMessageID), string(q.ChannelID), - ).Scan(&beforeSeq) + seq, err := s.q.GetPageCursorSeq(ctx, db.GetPageCursorSeqParams{ + AccountID: string(q.Actor), + ID: string(q.Page.BeforeMessageID), + ChannelID: string(q.ChannelID), + }) if err != nil { if noRows(err) { return nil, fmt.Errorf("%w: before-cursor %q not in channel", ErrInvalidArgument, q.Page.BeforeMessageID) } return nil, fmt.Errorf("store: resolve page cursor: %w", err) } + beforeSeq = seq } // A zero beforeSeq (no cursor) reads the newest page; a positive one pages @@ -511,26 +475,22 @@ func (s *Store) ListMessages(ctx context.Context, q ListMessagesQuery) ([]Messag // id-deduping client converges to current content (last-write-wins). // Freezing content too would need an update/change-seq and a larger schema // change; membership-only is the ratified scope (RIG-1333 OQ5). - const query = ` - SELECT m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks - FROM messages m - JOIN topics t ON t.id = m.topic_id - JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 - WHERE t.channel_id = $2 AND ($3 = 0 OR m.seq < $3) AND ($5 = 0 OR m.seq <= $5) - AND ($6 = '' OR m.topic_id = $6) - ORDER BY m.seq DESC - LIMIT $4` - // seq is BIGSERIAL (the int64 domain) and SnapshotSeq is a server-issued - // boundary the client echoes back, so the value is in range by construction; - // an out-of-range client value degrades to an empty page (m.seq <= a negative - // bound matches nothing), never a fault. - snap := int64(q.Page.SnapshotSeq) //nolint:gosec // G115: see the note above — server-issued seq, int64 domain - rows, err := s.pool.Query(ctx, query, string(q.Actor), string(q.ChannelID), beforeSeq, int64(limit), snap, q.TopicID) + // SnapshotSeq is a server-issued boundary the client echoes back, so the + // value is in range by construction; an out-of-range client value degrades + // to an empty page (m.seq <= a negative bound matches nothing), never a fault. + snap := int64(q.Page.SnapshotSeq) //nolint:gosec // G115: server-issued seq, int64 domain + rows, err := s.q.ListMessages(ctx, db.ListMessagesParams{ + AccountID: string(q.Actor), + ChannelID: string(q.ChannelID), + Column3: beforeSeq, + Limit: int32(limit), //nolint:gosec // G115: clampLimit bounds this to maxPageLimit (200) + Column5: snap, + Column6: q.TopicID, + }) if err != nil { return nil, fmt.Errorf("store: list messages: %w", err) } - defer rows.Close() - return scanMessages(rows) + return messagesFromListRows(rows) } // SearchMessages runs a Postgres full-text search over message text, scoped to @@ -551,24 +511,19 @@ func (s *Store) SearchMessages(ctx context.Context, actor AccountID, scope Searc // rows rather than erroring. Visibility: the message's channel (resolved // through the topic join) must be one the actor is a member of; the optional // scope narrows within that set. - const q = ` - SELECT m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks - FROM messages m - JOIN topics t ON t.id = m.topic_id - JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 - WHERE m.search_tsv @@ websearch_to_tsquery('english', $2) - AND ($3 = '' OR t.channel_id = $3) - AND ($5 = 0 OR m.seq <= $5) - ORDER BY ts_rank(m.search_tsv, websearch_to_tsquery('english', $2)) DESC, m.seq DESC - LIMIT $4` // SnapshotSeq is int64-safe by construction; see ListMessages. snap := int64(page.SnapshotSeq) //nolint:gosec // G115: server-issued seq, int64 domain (see ListMessages) - rows, err := s.pool.Query(ctx, q, string(actor), query, string(scope.ChannelID), int64(limit), snap) + rows, err := s.q.SearchMessages(ctx, db.SearchMessagesParams{ + AccountID: string(actor), + WebsearchToTsquery: query, + Column3: string(scope.ChannelID), + Limit: int32(limit), //nolint:gosec // G115: clampLimit bounds this to maxPageLimit (200) + Column5: snap, + }) if err != nil { return nil, fmt.Errorf("store: search messages: %w", err) } - defer rows.Close() - return scanMessages(rows) + return messagesFromSearchRows(rows) } // AnswerAsk records a participant's atomic answer to a pending structured ask @@ -620,33 +575,24 @@ func (s *Store) AnswerAsk(ctx context.Context, actor AccountID, askID string, an // Zero rows -> ErrNotFound (never a distinct not-authorized), so ask // existence cannot leak across a membership boundary. FOR UPDATE OF m locks // the message row (not the membership row) for the transaction's duration. - const q = ` - SELECT m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks - FROM messages m - JOIN topics t ON t.id = m.topic_id - JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 - WHERE m.blocks @> $2::jsonb - FOR UPDATE OF m` filter, err := askIDContainmentFilter(askID) if err != nil { return Message{}, Message{}, fmt.Errorf("store: marshal ask filter: %w", err) } - rows, err := tx.Query(ctx, q, string(actor), filter) + found, err := db.New(tx).FindAskMessage(ctx, db.FindAskMessageParams{ + AccountID: string(actor), + Column2: filter, + }) if err != nil { return Message{}, Message{}, fmt.Errorf("store: find ask: %w", err) } - defer rows.Close() - msgs, err := scanMessages(rows) + if len(found) == 0 { + return Message{}, Message{}, fmt.Errorf("%w: ask %q", ErrNotFound, askID) + } + msg, err := messageFromParts(found[0].ID, found[0].TopicID, found[0].AuthorAccountID, found[0].AtUnixMs, found[0].Blocks) if err != nil { return Message{}, Message{}, err } - if len(msgs) == 0 { - return Message{}, Message{}, fmt.Errorf("%w: ask %q", ErrNotFound, askID) - } - msg := msgs[0] - // rows must be closed before issuing further queries on the same tx (pgx - // serializes a connection: an open rows cursor blocks the answer insert). - rows.Close() // Locate the ask block and validate the answers cover its questions exactly, // each answer against its question's offered options and arity, then record @@ -723,6 +669,9 @@ func findAsk(blocks []MessageBlock, askID string) *Ask { // ErrInvalidArgument. ask_id is immutable (preserved by the update write path), // so recording an answer never re-mints it. func applyAskAnswer(msg *Message, askID string, answers []AskAnswer) error { + if msg == nil { + return errors.New("store: apply ask answer: nil message") + } for i := range msg.Blocks { ask := msg.Blocks[i].Ask if ask == nil || ask.AskID != askID { @@ -811,52 +760,68 @@ func validateQuestionAnswer(q *AskQuestion, a AskAnswer, askID string) error { // getMessageByRequestID returns the message stored under an author's idempotency // key — the dedup path for a retried AppendMessage. func (s *Store) getMessageByRequestID(ctx context.Context, author AccountID, clientRequestID string) (Message, error) { - const q = ` - SELECT id, topic_id, author_account_id, at_unix_ms, blocks - FROM messages - WHERE author_account_id = $1 AND client_request_id = $2` - rows, err := s.pool.Query(ctx, q, string(author), clientRequestID) + rows, err := s.q.GetMessageByRequestID(ctx, db.GetMessageByRequestIDParams{ + AuthorAccountID: string(author), + ClientRequestID: clientRequestID, + }) if err != nil { return Message{}, fmt.Errorf("store: read deduped message: %w", err) } - defer rows.Close() - msgs, err := scanMessages(rows) + if len(rows) == 0 { + return Message{}, fmt.Errorf("%w: deduped message for key %q", ErrNotFound, clientRequestID) + } + return messageFromParts(rows[0].ID, rows[0].TopicID, rows[0].AuthorAccountID, rows[0].AtUnixMs, rows[0].Blocks) +} + +// messageFromParts reconstructs a domain Message from the shared five-column +// projection (id, topic_id, author_account_id, at_unix_ms, blocks) every message +// read returns, decoding the JSONB block set. It replaces the former scanMessages +// helper now that sqlc emits a typed row per query rather than a pgx.Rows cursor. +func messageFromParts(id, topicID, author string, atMS int64, blocksJSON []byte) (Message, error) { + blocks, err := unmarshalBlocks(blocksJSON) if err != nil { return Message{}, err } - if len(msgs) == 0 { - return Message{}, fmt.Errorf("%w: deduped message for key %q", ErrNotFound, clientRequestID) - } - return msgs[0], nil + return Message{ + ID: MessageID(id), + TopicID: topicID, + AuthorAccountID: AccountID(author), + At: time.UnixMilli(atMS).UTC(), + Blocks: blocks, + }, nil } -// scanMessages reads message rows into Messages, decoding the JSONB block set on -// each. Shared by the list, search, and dedup reads. -func scanMessages(rows pgx.Rows) ([]Message, error) { - var msgs []Message - for rows.Next() { - var ( - id, topicID, author string - atMS int64 - blocksJSON []byte - ) - if err := rows.Scan(&id, &topicID, &author, &atMS, &blocksJSON); err != nil { - return nil, fmt.Errorf("store: scan message: %w", err) - } - blocks, err := unmarshalBlocks(blocksJSON) +// messagesFromListRows / messagesFromSearchRows map the generated per-query rows +// (structurally identical, but distinct Go types) through messageFromParts. A +// decode failure on any row is propagated (fail-loud), matching the former +// scanMessages: a malformed block set is a store invariant violation, never a +// silently-dropped row. +func messagesFromListRows(rows []db.ListMessagesRow) ([]Message, error) { + if len(rows) == 0 { + return nil, nil + } + out := make([]Message, 0, len(rows)) + for _, r := range rows { + m, err := messageFromParts(r.ID, r.TopicID, r.AuthorAccountID, r.AtUnixMs, r.Blocks) if err != nil { return nil, err } - msgs = append(msgs, Message{ - ID: MessageID(id), - TopicID: topicID, - AuthorAccountID: AccountID(author), - At: time.UnixMilli(atMS).UTC(), - Blocks: blocks, - }) + out = append(out, m) + } + return out, nil +} + +func messagesFromSearchRows(rows []db.SearchMessagesRow) ([]Message, error) { + if len(rows) == 0 { + return nil, nil } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate messages: %w", err) + out := make([]Message, 0, len(rows)) + for _, r := range rows { + m, err := messageFromParts(r.ID, r.TopicID, r.AuthorAccountID, r.AtUnixMs, r.Blocks) + if err != nil { + return nil, err + } + out = append(out, m) } - return msgs, nil + return out, nil } diff --git a/go/internal/store/presence_reads.go b/go/internal/store/presence_reads.go index d2b721253..e4da41940 100644 --- a/go/internal/store/presence_reads.go +++ b/go/internal/store/presence_reads.go @@ -3,6 +3,8 @@ package store import ( "context" "fmt" + + "github.com/RigelBuild/compass/go/internal/store/db" ) // The presence component's read side (RIG-1569 T8, design record D4). Two pure @@ -32,14 +34,8 @@ import ( // An agent with no authored asks, or only answered asks, is false — never an // error for the ordinary "no open ask" case. func (s *Store) AgentHasOpenAsk(ctx context.Context, agent AccountID) (bool, error) { - const q = ` - SELECT EXISTS ( - SELECT 1 FROM messages - WHERE author_account_id = $1 - AND blocks @? '$[*] ? (@.kind == "ask" && (!exists(@.ask.answered) || @.ask.answered == false))' - )` - var open bool - if err := s.pool.QueryRow(ctx, q, string(agent)).Scan(&open); err != nil { + open, err := s.q.AgentHasOpenAsk(ctx, string(agent)) + if err != nil { return false, fmt.Errorf("store: check agent open ask: %w", err) } return open, nil @@ -60,16 +56,11 @@ func (s *Store) AgentHasOpenAsk(ctx context.Context, agent AccountID) (bool, err // co-inhabits a channel with) and is the MVP the brief pins, with the broader // visibility nuance parked for the driver. func (s *Store) SharesVisibleChannel(ctx context.Context, actor AccountID, agent AccountID) (bool, error) { - const q = ` - SELECT EXISTS ( - SELECT 1 - FROM channel_members cm1 - JOIN channel_members cm2 ON cm2.channel_id = cm1.channel_id - WHERE cm1.account_id = $1 - AND cm2.account_id = $2 - )` - var shares bool - if err := s.pool.QueryRow(ctx, q, string(actor), string(agent)).Scan(&shares); err != nil { + shares, err := s.q.SharesVisibleChannel(ctx, db.SharesVisibleChannelParams{ + AccountID: string(actor), + AccountID_2: string(agent), + }) + if err != nil { return false, fmt.Errorf("store: check shared visible channel: %w", err) } return shares, nil diff --git a/go/internal/store/queries/delivery_cursors.sql b/go/internal/store/queries/delivery_cursors.sql new file mode 100644 index 000000000..51bc063cf --- /dev/null +++ b/go/internal/store/queries/delivery_cursors.sql @@ -0,0 +1,97 @@ +-- Delivery-cursor queries (sqlc adoption T4, RIG-3034). These replace the inline +-- SQL literals in internal/store/delivery_cursors.go; the hand-written Store +-- methods keep their signatures, the AckDelivery tx orchestration (the owed-clear +-- FIRST, the commit-if-cleared arm, the contiguous-advance loop in Go), and the +-- D2 seed self-guard/idempotency contract. The two message-fanout reads +-- (OwedMentions, UndeliveredMessages) share the per-channel projection the Go +-- drains with an inline loop calling messageFromParts. + +-- name: SeedDeliveryCursor :exec +INSERT INTO agent_delivery_cursors (agent_account_id, channel_id, acked_seq) +SELECT $1, $2, COALESCE((SELECT MAX(m.seq) FROM messages m JOIN topics t ON t.id = m.topic_id WHERE t.channel_id = $2), 0) +WHERE EXISTS (SELECT 1 FROM agent_accounts WHERE account_id = $1) +ON CONFLICT (agent_account_id, channel_id) DO NOTHING; + +-- name: SeedChannelDeliveryCursors :exec +INSERT INTO agent_delivery_cursors (agent_account_id, channel_id, acked_seq) +SELECT cm.account_id, $1, + COALESCE((SELECT MAX(m.seq) FROM messages m JOIN topics t ON t.id = m.topic_id WHERE t.channel_id = $1), 0) +FROM channel_members cm +JOIN agent_accounts aa ON aa.account_id = cm.account_id +WHERE cm.channel_id = $1 +ON CONFLICT (agent_account_id, channel_id) DO NOTHING; + +-- name: RecordOwedMention :exec +INSERT INTO owed_mentions (agent_account_id, message_id, channel_id, recorded_at_unix_ms) +VALUES ($1, $2, $3, $4) +ON CONFLICT (agent_account_id, message_id) DO NOTHING; + +-- name: OwedMentions :many +SELECT m.id, m.topic_id, t.channel_id, m.author_account_id, m.at_unix_ms, m.blocks +FROM owed_mentions om +JOIN messages m ON m.id = om.message_id +JOIN topics t ON t.id = m.topic_id +WHERE om.agent_account_id = $1 +ORDER BY t.channel_id, m.seq ASC; + +-- name: ClearOwedMention :execrows +DELETE FROM owed_mentions WHERE agent_account_id = $1 AND message_id = $2; + +-- name: CountOwedMentions :one +SELECT COUNT(*) FROM owed_mentions; + +-- name: MarkMentionsRouted :exec +UPDATE messages SET mentions_routed_at = $1 WHERE id = $2; + +-- name: UnroutedMentionMessages :many +SELECT m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks, t.channel_id, m.seq +FROM messages m +JOIN topics t ON t.id = m.topic_id +WHERE m.mentions_routed_at IS NULL AND m.seq > $1 +ORDER BY m.seq ASC +LIMIT $2; + +-- name: ResolveAckMessage :one +SELECT m.seq FROM messages m JOIN topics t ON t.id = m.topic_id WHERE m.id = $1 AND t.channel_id = $2; + +-- name: LoadDeliveryCursor :one +SELECT acked_seq, above_seqs FROM agent_delivery_cursors +WHERE agent_account_id = $1 AND channel_id = $2 +FOR UPDATE; + +-- name: SelfAuthoredSeqsAbove :many +SELECT m.seq FROM messages m JOIN topics t ON t.id = m.topic_id +WHERE t.channel_id = $1 AND m.seq > $2 AND m.author_account_id = $3; + +-- name: AdvanceDeliveryCursor :exec +UPDATE agent_delivery_cursors +SET acked_seq = $3, above_seqs = $4, acked_at = now() +WHERE agent_account_id = $1 AND channel_id = $2; + +-- name: UndeliveredMessages :many +SELECT m.id, m.topic_id, t.channel_id, m.author_account_id, m.at_unix_ms, m.blocks +FROM channel_members cm +JOIN agent_accounts aa ON aa.account_id = cm.account_id +JOIN topics t ON t.channel_id = cm.channel_id +JOIN messages m ON m.topic_id = t.id +JOIN channels ch ON ch.id = cm.channel_id +LEFT JOIN agent_delivery_cursors dc + ON dc.agent_account_id = cm.account_id AND dc.channel_id = cm.channel_id +WHERE cm.account_id = $1 + AND (cm.subscribed OR cm.channel_id = aa.home_channel_id OR ch.mandatory_subscription) + AND m.author_account_id <> $1 + AND m.seq > COALESCE( + dc.acked_seq, + (SELECT COALESCE(MAX(mh.seq), 0) FROM messages mh JOIN topics th ON th.id = mh.topic_id WHERE th.channel_id = cm.channel_id)) + AND m.seq <> ALL(COALESCE(dc.above_seqs, '{}'::BIGINT[])) +ORDER BY t.channel_id, m.seq ASC; + +-- name: InSweepSet :one +SELECT EXISTS( + SELECT 1 + FROM channel_members cm + JOIN agent_accounts aa ON aa.account_id = cm.account_id + JOIN channels ch ON ch.id = cm.channel_id + WHERE cm.account_id = $1 + AND cm.channel_id = $2 + AND (cm.subscribed OR cm.channel_id = aa.home_channel_id OR ch.mandatory_subscription)); diff --git a/go/internal/store/queries/delivery_reads.sql b/go/internal/store/queries/delivery_reads.sql new file mode 100644 index 000000000..2d2d25758 --- /dev/null +++ b/go/internal/store/queries/delivery_reads.sql @@ -0,0 +1,46 @@ +-- Delivery-consumer read queries (sqlc adoption T4, RIG-3034). These replace the +-- inline SQL literals in internal/store/delivery_reads.go; the hand-written Store +-- methods keep their signatures, the D1 sweep-set disjunct (kept textually in +-- sync with delivery_cursors.sql UndeliveredMessages/InSweepSet), and the D9 +-- error mapping. MessageByID shares the message projection the Go drains via +-- messageFromParts. + +-- name: SubscribedAgents :many +SELECT aa.account_id +FROM channel_members cm +JOIN agent_accounts aa ON aa.account_id = cm.account_id +JOIN channels ch ON ch.id = cm.channel_id +WHERE cm.channel_id = $1 + AND (cm.subscribed OR cm.channel_id = aa.home_channel_id OR ch.mandatory_subscription) + AND cm.account_id <> $2 +ORDER BY aa.account_id; + +-- name: ChannelAgentMembers :many +SELECT aa.account_id +FROM channel_members cm +JOIN agent_accounts aa ON aa.account_id = cm.account_id +WHERE cm.channel_id = $1 + AND cm.account_id <> $2 +ORDER BY aa.account_id; + +-- name: IsAgentAccount :one +SELECT EXISTS (SELECT 1 FROM agent_accounts WHERE account_id = $1); + +-- name: MessageByID :one +SELECT id, topic_id, author_account_id, at_unix_ms, blocks +FROM messages +WHERE id = $1; + +-- name: MessageChannel :one +SELECT t.channel_id FROM messages m JOIN topics t ON t.id = m.topic_id WHERE m.id = $1; + +-- name: TopicChannelNames :one +SELECT t.name AS topic_name, c.name AS channel_name FROM topics t JOIN channels c ON c.id = t.channel_id WHERE t.id = $1; +-- name: SweepChannels :many +SELECT cm.channel_id +FROM channel_members cm +JOIN agent_accounts aa ON aa.account_id = cm.account_id +JOIN channels ch ON ch.id = cm.channel_id +WHERE cm.account_id = $1 + AND (cm.subscribed OR cm.channel_id = aa.home_channel_id OR ch.mandatory_subscription) +ORDER BY cm.channel_id; diff --git a/go/internal/store/queries/messages.sql b/go/internal/store/queries/messages.sql new file mode 100644 index 000000000..35c3f5df5 --- /dev/null +++ b/go/internal/store/queries/messages.sql @@ -0,0 +1,99 @@ +-- Message-domain queries (sqlc adoption T4, RIG-3034). These replace the inline +-- SQL literals in internal/store/messages.go; the hand-written Store methods keep +-- their exact signatures, their tx orchestration (AppendMessage/AnswerAsk begin +-- and commit their own txns), the ON CONFLICT idempotency signalling +-- (errMessageInsertConflict), the JSONB block (de)serialization, and the D9 +-- not-found/forbidden error mapping — all hand-written around these generated +-- calls. Every message read shares the id/topic_id/author_account_id/at_unix_ms/ +-- blocks projection (the former scanMessages order) so the Go maps each row the +-- same way via messageFromParts. + +-- name: GetChannelPostPolicy :one +SELECT post_policy, COALESCE(owner_account_id, '') AS owner_account_id +FROM channels WHERE id = $1; + +-- name: InsertMessage :one +INSERT INTO messages (id, topic_id, author_account_id, at_unix_ms, blocks, text_content, client_request_id) +VALUES ($1, $2, $3, $4, $5, $6, $7) +ON CONFLICT (author_account_id, client_request_id) WHERE client_request_id <> '' +DO NOTHING +RETURNING id, at_unix_ms, seq; + +-- name: UpdateTopicLastSeq :exec +UPDATE topics SET last_seq = GREATEST(last_seq, $2) WHERE id = $1; + +-- name: GetTopicChannel :one +SELECT channel_id FROM topics WHERE id = $1; + +-- name: InsertTopicIgnore :exec +INSERT INTO topics (id, channel_id, name, created_by_account_id, created_at_unix_ms) +VALUES ($1, $2, $3, $4, $5) +ON CONFLICT (channel_id, lower(name)) DO NOTHING; + +-- name: GetTopicByName :one +SELECT id, archived FROM topics WHERE channel_id = $1 AND lower(name) = lower($2); + +-- name: ReviveTopic :exec +UPDATE topics SET archived = FALSE WHERE id = $1; + +-- name: MessagesHeadSeq :one +SELECT COALESCE(MAX(seq), 0)::BIGINT AS head FROM messages; + +-- name: UpdateMessageBlocks :execrows +UPDATE messages SET blocks = $1, text_content = $2 WHERE id = $3; + +-- name: UpdateMessageBlocksAsAuthor :one +UPDATE messages m +SET blocks = $1, text_content = $2 +FROM topics t +WHERE m.id = $3 + AND t.id = m.topic_id + AND m.author_account_id = $4 + AND EXISTS ( + SELECT 1 FROM channel_members cm + WHERE cm.channel_id = t.channel_id AND cm.account_id = $4 + ) +RETURNING m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks; + +-- name: GetMessageBlocks :one +SELECT blocks FROM messages WHERE id = $1; + +-- name: GetPageCursorSeq :one +SELECT m.seq FROM messages m +JOIN topics t ON t.id = m.topic_id +JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 +WHERE m.id = $2 AND t.channel_id = $3; + +-- name: ListMessages :many +SELECT m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks +FROM messages m +JOIN topics t ON t.id = m.topic_id +JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 +WHERE t.channel_id = $2 AND ($3 = 0 OR m.seq < $3) AND ($5 = 0 OR m.seq <= $5) + AND ($6 = '' OR m.topic_id = $6) +ORDER BY m.seq DESC +LIMIT $4; + +-- name: SearchMessages :many +SELECT m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks +FROM messages m +JOIN topics t ON t.id = m.topic_id +JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 +WHERE m.search_tsv @@ websearch_to_tsquery('english', $2) + AND ($3 = '' OR t.channel_id = $3) + AND ($5 = 0 OR m.seq <= $5) +ORDER BY ts_rank(m.search_tsv, websearch_to_tsquery('english', $2)) DESC, m.seq DESC +LIMIT $4; + +-- name: FindAskMessage :many +SELECT m.id, m.topic_id, m.author_account_id, m.at_unix_ms, m.blocks +FROM messages m +JOIN topics t ON t.id = m.topic_id +JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 +WHERE m.blocks @> $2::jsonb +FOR UPDATE OF m; + +-- name: GetMessageByRequestID :many +SELECT id, topic_id, author_account_id, at_unix_ms, blocks +FROM messages +WHERE author_account_id = $1 AND client_request_id = $2; diff --git a/go/internal/store/queries/presence_reads.sql b/go/internal/store/queries/presence_reads.sql new file mode 100644 index 000000000..19c17c7e5 --- /dev/null +++ b/go/internal/store/queries/presence_reads.sql @@ -0,0 +1,21 @@ +-- Presence-component read queries (sqlc adoption T4, RIG-3034). These replace the +-- const-hoisted SQL in internal/store/presence_reads.go (it was never in the +-- inline-sql-gate allowlist — the gate is literal-at-callsite scoped and this +-- file passed its SQL as a const identifier). The hand-written Store methods keep +-- their signatures and error mapping. + +-- name: AgentHasOpenAsk :one +SELECT EXISTS ( + SELECT 1 FROM messages + WHERE author_account_id = $1 + AND blocks @? '$[*] ? (@.kind == "ask" && (!exists(@.ask.answered) || @.ask.answered == false))' +); + +-- name: SharesVisibleChannel :one +SELECT EXISTS ( + SELECT 1 + FROM channel_members cm1 + JOIN channel_members cm2 ON cm2.channel_id = cm1.channel_id + WHERE cm1.account_id = $1 + AND cm2.account_id = $2 +); diff --git a/go/internal/store/queries/topics.sql b/go/internal/store/queries/topics.sql new file mode 100644 index 000000000..a00e7e74a --- /dev/null +++ b/go/internal/store/queries/topics.sql @@ -0,0 +1,43 @@ +-- Topic-domain queries (sqlc adoption T4, RIG-3034). These replace the inline +-- SQL literals in internal/store/topics.go; the hand-written Store methods keep +-- their signatures, the UpdateTopic tx orchestration, the rename/merge resolution +-- loop, and the D9 not-found/forbidden error mapping. The topic projection +-- (id, channel_id, name, created_by_account_id, created_at_unix_ms, archived, +-- last_seq) matches the former scanTopics order so the Go maps each row to Topic. + +-- name: ListTopics :many +SELECT id, channel_id, name, created_by_account_id, created_at_unix_ms, archived, last_seq +FROM topics +WHERE channel_id = $1 AND ($2 OR NOT archived) +ORDER BY last_seq DESC, created_at_unix_ms DESC, id; + +-- name: ResolveTopicForUpdate :one +SELECT t.channel_id FROM topics t +JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 +WHERE t.id = $2 +FOR UPDATE OF t; + +-- name: SetTopicArchived :exec +UPDATE topics SET archived = $2 WHERE id = $1; + +-- name: GetTopic :one +SELECT id, channel_id, name, created_by_account_id, created_at_unix_ms, archived, last_seq +FROM topics WHERE id = $1; + +-- name: ResolveTopicRenameTarget :one +SELECT id FROM topics +WHERE channel_id = $1 AND lower(name) = lower($2) AND id <> $3 +FOR UPDATE; + +-- name: RenameTopic :exec +UPDATE topics SET name = $2 WHERE id = $1; + +-- name: MoveMessagesToTopic :exec +UPDATE messages SET topic_id = $1 WHERE topic_id = $2; + +-- name: MergeTopicLastSeq :exec +UPDATE topics dst SET last_seq = GREATEST(dst.last_seq, src.last_seq) +FROM topics src WHERE dst.id = $1 AND src.id = $2; + +-- name: DeleteTopic :exec +DELETE FROM topics WHERE id = $1; diff --git a/go/internal/store/store.go b/go/internal/store/store.go index de256b2ab..6e921fd27 100644 --- a/go/internal/store/store.go +++ b/go/internal/store/store.go @@ -73,13 +73,6 @@ type querier interface { QueryRow(ctx context.Context, sql string, args ...any) pgx.Row } -// execer is the write surface shared by the pool and a transaction, so a write -// helper (updateMessageBlocksExec) can run against either. Both *pgxpool.Pool -// and pgx.Tx satisfy it. -type execer interface { - Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) -} - // Open connects to Postgres at dsn (a pgx pool), applies any pending embedded // migrations under an advisory lock, and verifies the resulting schema version // matches what this binary expects — refusing to serve on a failed migration diff --git a/go/internal/store/topics.go b/go/internal/store/topics.go index 4aea116b2..665f68893 100644 --- a/go/internal/store/topics.go +++ b/go/internal/store/topics.go @@ -5,6 +5,8 @@ import ( "fmt" "github.com/jackc/pgx/v5" + + "github.com/RigelBuild/compass/go/internal/store/db" ) // ListTopics returns the topics in channelID, newest-activity-first (last_seq @@ -28,17 +30,11 @@ func (s *Store) ListTopics(ctx context.Context, callerAccountID, channelID strin return nil, fmt.Errorf("%w: channel %q", ErrNotFound, channelID) } - const q = ` - SELECT id, channel_id, name, created_by_account_id, created_at_unix_ms, archived, last_seq - FROM topics - WHERE channel_id = $1 AND ($2 OR NOT archived) - ORDER BY last_seq DESC, created_at_unix_ms DESC, id` - rows, err := s.pool.Query(ctx, q, channelID, includeArchived) + rows, err := s.q.ListTopics(ctx, db.ListTopicsParams{ChannelID: channelID, Column2: includeArchived}) if err != nil { return nil, fmt.Errorf("store: list topics: %w", err) } - defer rows.Close() - return scanTopics(rows) + return topicsFromRows(rows), nil } // UpdateTopic renames and/or archives a topic under an acting account, or — @@ -71,14 +67,12 @@ func (s *Store) UpdateTopic(ctx context.Context, callerAccountID, topicID string // member of the topic's channel. Zero rows (unknown topic OR non-member) -> // ErrNotFound. FOR UPDATE OF t locks the source topic row for the tx so a // concurrent rename/merge serializes. - var channelID string - switch err := tx.QueryRow(ctx, - `SELECT t.channel_id FROM topics t - JOIN channel_members cm ON cm.channel_id = t.channel_id AND cm.account_id = $1 - WHERE t.id = $2 - FOR UPDATE OF t`, - callerAccountID, topicID, - ).Scan(&channelID); { + q := db.New(tx) + channelID, err := q.ResolveTopicForUpdate(ctx, db.ResolveTopicForUpdateParams{ + AccountID: callerAccountID, + ID: topicID, + }) + switch { case noRows(err): return Topic{}, fmt.Errorf("%w: topic %q", ErrNotFound, topicID) case err != nil: @@ -97,25 +91,20 @@ func (s *Store) UpdateTopic(ctx context.Context, callerAccountID, topicID string } if archived != nil { - if _, err := tx.Exec(ctx, - `UPDATE topics SET archived = $2 WHERE id = $1`, surviving, *archived, - ); err != nil { + if err := q.SetTopicArchived(ctx, db.SetTopicArchivedParams{ID: surviving, Archived: *archived}); err != nil { return Topic{}, fmt.Errorf("store: set topic archived: %w", err) } } - var topic Topic - if err := tx.QueryRow(ctx, - `SELECT id, channel_id, name, created_by_account_id, created_at_unix_ms, archived, last_seq - FROM topics WHERE id = $1`, surviving, - ).Scan(&topic.ID, &topic.ChannelID, &topic.Name, &topic.CreatedByAccountID, &topic.CreatedAtUnixMS, &topic.Archived, &topic.LastSeq); err != nil { + topic, err := q.GetTopic(ctx, surviving) + if err != nil { return Topic{}, fmt.Errorf("store: read updated topic: %w", err) } if err := tx.Commit(ctx); err != nil { return Topic{}, fmt.Errorf("store: commit update topic: %w", err) } - return topic, nil + return topicFromRow(topic), nil } // applyTopicRename renames topic topicID (in channelID) to newName, or merges it @@ -126,21 +115,19 @@ func (s *Store) UpdateTopic(ctx context.Context, callerAccountID, topicID string func (s *Store) applyTopicRename(ctx context.Context, tx pgx.Tx, channelID, topicID, newName string) (string, error) { // A same-channel topic already holding the target name (excluding the source // itself) is a merge target. - var targetID string - switch err := tx.QueryRow(ctx, - `SELECT id FROM topics - WHERE channel_id = $1 AND lower(name) = lower($2) AND id <> $3 - FOR UPDATE`, - channelID, newName, topicID, - ).Scan(&targetID); { + q := db.New(tx) + targetID, err := q.ResolveTopicRenameTarget(ctx, db.ResolveTopicRenameTargetParams{ + ChannelID: channelID, + Lower: newName, + ID: topicID, + }) + switch { case noRows(err): // No collision: rename in place. The unique index still guards a race // with a concurrent create of the same name (that transaction holds its // own row lock); such a rename fails the constraint and surfaces as a // conflict rather than corrupting the index. - if _, err := tx.Exec(ctx, - `UPDATE topics SET name = $2 WHERE id = $1`, topicID, newName, - ); err != nil { + if err := q.RenameTopic(ctx, db.RenameTopicParams{ID: topicID, Name: newName}); err != nil { if pgErrIs(err, pgUniqueViolation) { return "", fmt.Errorf("%w: topic name %q already exists in this channel", ErrConflict, newName) } @@ -154,36 +141,40 @@ func (s *Store) applyTopicRename(ctx context.Context, tx pgx.Tx, channelID, topi // Collision: merge the source into the target. Every source message carries // the target's topic_id, the target absorbs the source's activity marker, // and the emptied source row is deleted — all in this tx. - if _, err := tx.Exec(ctx, - `UPDATE messages SET topic_id = $1 WHERE topic_id = $2`, targetID, topicID, - ); err != nil { + if err := q.MoveMessagesToTopic(ctx, db.MoveMessagesToTopicParams{TopicID: targetID, TopicID_2: topicID}); err != nil { return "", fmt.Errorf("store: move messages on topic merge: %w", err) } - if _, err := tx.Exec(ctx, - `UPDATE topics dst SET last_seq = GREATEST(dst.last_seq, src.last_seq) - FROM topics src WHERE dst.id = $1 AND src.id = $2`, - targetID, topicID, - ); err != nil { + if err := q.MergeTopicLastSeq(ctx, db.MergeTopicLastSeqParams{ID: targetID, ID_2: topicID}); err != nil { return "", fmt.Errorf("store: merge topic last_seq: %w", err) } - if _, err := tx.Exec(ctx, `DELETE FROM topics WHERE id = $1`, topicID); err != nil { + if err := q.DeleteTopic(ctx, topicID); err != nil { return "", fmt.Errorf("store: delete merged topic: %w", err) } return targetID, nil } -// scanTopics reads topic rows into Topics. -func scanTopics(rows pgx.Rows) ([]Topic, error) { - var topics []Topic - for rows.Next() { - var t Topic - if err := rows.Scan(&t.ID, &t.ChannelID, &t.Name, &t.CreatedByAccountID, &t.CreatedAtUnixMS, &t.Archived, &t.LastSeq); err != nil { - return nil, fmt.Errorf("store: scan topic: %w", err) - } - topics = append(topics, t) +// topicFromRow maps the generated db.Topic (the shared topic projection) to the +// domain Topic; topicsFromRows applies it across a list read. They replace the +// former scanTopics pgx.Rows helper. +func topicFromRow(t db.Topic) Topic { + return Topic{ + ID: t.ID, + ChannelID: t.ChannelID, + Name: t.Name, + CreatedByAccountID: t.CreatedByAccountID, + CreatedAtUnixMS: t.CreatedAtUnixMs, + Archived: t.Archived, + LastSeq: t.LastSeq, + } +} + +func topicsFromRows(rows []db.Topic) []Topic { + if len(rows) == 0 { + return nil } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate topics: %w", err) + out := make([]Topic, 0, len(rows)) + for _, t := range rows { + out = append(out, topicFromRow(t)) } - return topics, nil + return out } diff --git a/tools/inline-sql-gate/index.ts b/tools/inline-sql-gate/index.ts index 6540eb04c..990856b20 100644 --- a/tools/inline-sql-gate/index.ts +++ b/tools/inline-sql-gate/index.ts @@ -84,12 +84,10 @@ export const ALLOWLIST: string[] = [ // shape this gate flags): the record's 24-file list minus agent_tree.go + // presence_reads.go (const-hoisted, not literal-at-callsite — see above), // 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/messages.go", - "go/internal/store/topics.go", - "go/internal/store/delivery_cursors.go", - "go/internal/store/delivery_reads.go", + // accounts.go migrated in T2; channels/channel_pins/coordination in T3; + // messages/topics/delivery_cursors/delivery_reads in T4 (RIG-3034). + // agent_tree.go + presence_reads.go were never seeded here (const-hoisted SQL, + // so the literal-scoped gate produced no finding). "go/internal/store/agent_sessions.go", "go/internal/store/agent_transcripts.go", "go/internal/store/agent_activity.go",