package postgres import ( "context" "errors" "fmt" "github.com/jackc/pgx/v5" "telesrv/internal/domain" "telesrv/internal/store/postgres/sqlcgen" ) // topicReadFallbackInbox is the viewer's initial read watermark for a topic with no read row: // messages before the topic root are treated as read (root-1), and it is not below available_min // (invisible before the trim line). For General (topic 1), root-1=0, i.e. "unread from the start". // This way, when the migration goes live it does not turn all existing topics unread in one jump. func topicReadFallbackInbox(topicID, availableMinID int) int { base := topicID - 1 if availableMinID > base { base = availableMinID } if base < 0 { base = 0 } return base } // channelTopicMessageCond returns the message-belonging predicate for a topic, with the placeholder carrying topicID. // General(=1) merges reply_to_top_id IN (placeholder, 0): both new messages inside General (=1) and // pre-forum historical messages (=0) count as General. func channelTopicMessageCond(topicID int, placeholder string) string { if topicID == domain.ForumGeneralTopicID { // General merges reply_to_top_id ∈ {placeholder,0}, but excludes other topics' root // service messages (their reply_to_top_id=0, but their id is some topic root, so they // are not part of General). return "(cm.reply_to_top_id = " + placeholder + " OR cm.reply_to_top_id = 0)" + " AND NOT EXISTS (SELECT 1 FROM channel_forum_topics gt WHERE gt.channel_id = cm.channel_id AND gt.topic_id = cm.id AND NOT gt.deleted)" } return "cm.reply_to_top_id = " + placeholder } // channelTopicReadInbox returns the viewer's read watermark for a topic, falling back when there is no row. func (s *ChannelStore) channelTopicReadInbox(ctx context.Context, db sqlcgen.DBTX, channelID, userID int64, topicID, availableMinID int) (int, error) { var v int err := db.QueryRow(ctx, `SELECT read_inbox_max_id FROM channel_topic_read WHERE channel_id = $1 AND user_id = $2 AND topic_id = $3`, channelID, userID, topicID).Scan(&v) if errors.Is(err, pgx.ErrNoRows) { return topicReadFallbackInbox(topicID, availableMinID), nil } if err != nil { return 0, fmt.Errorf("select channel topic read: %w", err) } if fb := topicReadFallbackInbox(topicID, availableMinID); v < fb { v = fb } return v, nil } // topicReadWater is the viewer's inbox/outbox read watermark for a topic. type topicReadWater struct { Inbox int Outbox int } // channelTopicReadBatch fetches the per-viewer read watermarks for multiple topics in a batch (used by getForumTopics for on-the-fly computing and delivery); // topics with no row fall back (inbox=root-1, outbox=0). Returns topicID -> watermark. func (s *ChannelStore) channelTopicReadBatch(ctx context.Context, channelID, userID int64, topicIDs []int32, availableMinID int) (map[int]topicReadWater, error) { out := make(map[int]topicReadWater, len(topicIDs)) for _, t := range topicIDs { out[int(t)] = topicReadWater{Inbox: topicReadFallbackInbox(int(t), availableMinID)} } if len(topicIDs) == 0 { return out, nil } rows, err := s.db.Query(ctx, `SELECT topic_id, read_inbox_max_id, read_outbox_max_id FROM channel_topic_read WHERE channel_id = $1 AND user_id = $2 AND topic_id = ANY($3::int[])`, channelID, userID, topicIDs) if err != nil { return nil, fmt.Errorf("batch channel topic read: %w", err) } defer rows.Close() for rows.Next() { var topicID, inbox, outbox int if err := rows.Scan(&topicID, &inbox, &outbox); err != nil { return nil, err } w := out[topicID] if inbox > w.Inbox { w.Inbox = inbox } w.Outbox = outbox out[topicID] = w } return out, rows.Err() } // channelTopicTopMessageID returns a topic's current latest visible message id, as the upper bound for advancing readDiscussion. func (s *ChannelStore) channelTopicTopMessageID(ctx context.Context, channelID int64, topicID, availableMinID int) (int, error) { var v int err := s.db.QueryRow(ctx, ` SELECT COALESCE(MAX(cm.id), 0)::int FROM channel_messages cm WHERE cm.channel_id = $1 AND `+channelTopicMessageCond(topicID, "$2")+` AND cm.id > $3 AND NOT cm.deleted`, channelID, topicID, availableMinID).Scan(&v) if err != nil { return 0, fmt.Errorf("channel topic top message: %w", err) } return v, nil } // ReadChannelTopicHistory advances the viewer's per-topic read watermark for a single forum topic (messages.readDiscussion), // without touching the channel-level channel_members.read_inbox_max_id (eliminating cross-topic read crosstalk), and returns the senders whose outbox needs advancing. func (s *ChannelStore) ReadChannelTopicHistory(ctx context.Context, req domain.ReadChannelTopicHistoryRequest) (domain.ReadChannelTopicHistoryResult, error) { if req.UserID == 0 || req.ChannelID == 0 || req.TopicID <= 0 { return domain.ReadChannelTopicHistoryResult{}, domain.ErrChannelInvalid } channel, member, err := s.getChannelForMemberOrLinkedGuest(ctx, s.db, req.UserID, req.ChannelID) if err != nil { return domain.ReadChannelTopicHistoryResult{}, err } if !channel.Forum { return domain.ReadChannelTopicHistoryResult{}, domain.ErrChannelForumMissing } if member.Guest { return domain.ReadChannelTopicHistoryResult{}, domain.ErrChannelPrivate } topMax, err := s.channelTopicTopMessageID(ctx, req.ChannelID, req.TopicID, member.AvailableMinID) if err != nil { return domain.ReadChannelTopicHistoryResult{}, err } maxID := req.MaxID if maxID <= 0 || maxID > topMax { maxID = topMax } prev, err := s.channelTopicReadInbox(ctx, s.db, req.ChannelID, req.UserID, req.TopicID, member.AvailableMinID) if err != nil { return domain.ReadChannelTopicHistoryResult{}, err } if maxID <= prev { return domain.ReadChannelTopicHistoryResult{Channel: channel, TopicID: req.TopicID, MaxID: prev, Changed: false, Pts: channel.Pts}, nil } if _, err := s.db.Exec(ctx, ` INSERT INTO channel_topic_read (channel_id, user_id, topic_id, read_inbox_max_id, read_inbox_date, updated_at) VALUES ($1, $2, $3, $4, $5, now()) ON CONFLICT (channel_id, user_id, topic_id) DO UPDATE SET read_inbox_date = CASE WHEN channel_topic_read.read_inbox_max_id < EXCLUDED.read_inbox_max_id THEN EXCLUDED.read_inbox_date ELSE channel_topic_read.read_inbox_date END, read_inbox_max_id = GREATEST(channel_topic_read.read_inbox_max_id, EXCLUDED.read_inbox_max_id), updated_at = now()`, req.ChannelID, req.UserID, req.TopicID, maxID, req.Date); err != nil { return domain.ReadChannelTopicHistoryResult{}, fmt.Errorf("upsert channel topic read inbox: %w", err) } outbox, err := s.advanceChannelTopicReadOutbox(ctx, req.ChannelID, req.UserID, req.TopicID, prev, maxID) if err != nil { return domain.ReadChannelTopicHistoryResult{}, err } return domain.ReadChannelTopicHistoryResult{Channel: channel, TopicID: req.TopicID, MaxID: maxID, Changed: true, Pts: channel.Pts, OutboxUpdates: outbox}, nil } // advanceChannelTopicReadOutbox advances the per-topic read_outbox watermark of the senders // covered by this read within the topic, and returns those senders for online delivery of // updateReadChannelDiscussionOutbox (read receipt ✓✓). The receipt UPSERT is idempotent and does // not need to be strongly atomic with the inbox. func (s *ChannelStore) advanceChannelTopicReadOutbox(ctx context.Context, channelID, readerUserID int64, topicID, prev, maxID int) ([]domain.ChannelReadOutboxUpdate, error) { rows, err := s.db.Query(ctx, ` SELECT DISTINCT cm.sender_user_id FROM channel_messages cm WHERE cm.channel_id = $1 AND `+channelTopicMessageCond(topicID, "$2")+` AND cm.id > $3 AND cm.id <= $4 AND cm.sender_user_id <> $5 AND cm.sender_user_id <> 0 AND NOT cm.deleted`, channelID, topicID, prev, maxID, readerUserID) if err != nil { return nil, fmt.Errorf("scan channel topic outbox senders: %w", err) } senders := make([]int64, 0, 8) for rows.Next() { var sender int64 if err := rows.Scan(&sender); err != nil { rows.Close() return nil, err } senders = append(senders, sender) } rows.Close() if err := rows.Err(); err != nil { return nil, err } updates := make([]domain.ChannelReadOutboxUpdate, 0, len(senders)) for _, sender := range senders { if _, err := s.db.Exec(ctx, ` INSERT INTO channel_topic_read (channel_id, user_id, topic_id, read_outbox_max_id, updated_at) VALUES ($1, $2, $3, $4, now()) ON CONFLICT (channel_id, user_id, topic_id) DO UPDATE SET read_outbox_max_id = GREATEST(channel_topic_read.read_outbox_max_id, EXCLUDED.read_outbox_max_id), updated_at = now()`, channelID, sender, topicID, maxID); err != nil { return nil, fmt.Errorf("upsert channel topic read outbox: %w", err) } updates = append(updates, domain.ChannelReadOutboxUpdate{UserID: sender, MaxID: maxID}) } return updates, nil } // GeneralForumTopic computes on the fly the state of the General topic (id=1) for the viewer: it belongs to reply_to_top_id ∈ {0,1} and // excludes other topics' root service messages; unread uses General's own per-topic watermark (not crosstalked by ordinary topic reads). func (s *ChannelStore) GeneralForumTopic(ctx context.Context, viewerUserID, channelID int64) (domain.ChannelForumTopic, error) { if viewerUserID == 0 || channelID == 0 { return domain.ChannelForumTopic{}, domain.ErrChannelInvalid } channel, member, err := s.getChannelForMember(ctx, s.db, viewerUserID, channelID) if err != nil { return domain.ChannelForumTopic{}, err } if !channel.Forum { return domain.ChannelForumTopic{}, domain.ErrChannelForumMissing } gid := domain.ForumGeneralTopicID waters, err := s.channelTopicReadBatch(ctx, channelID, viewerUserID, []int32{int32(gid)}, member.AvailableMinID) if err != nil { return domain.ChannelForumTopic{}, err } water := waters[gid] cond := channelTopicMessageCond(gid, "$2") var topMsg int if err := s.db.QueryRow(ctx, ` SELECT COALESCE(MAX(cm.id), 0)::int FROM channel_messages cm WHERE cm.channel_id = $1 AND `+cond+` AND cm.id > $3 AND NOT cm.deleted`, channelID, gid, member.AvailableMinID).Scan(&topMsg); err != nil { return domain.ChannelForumTopic{}, fmt.Errorf("general top message: %w", err) } var unread int if err := s.db.QueryRow(ctx, ` SELECT COUNT(*)::int FROM ( SELECT 1 FROM channel_messages cm WHERE cm.channel_id = $1 AND `+cond+` AND cm.id > $3 AND cm.sender_user_id <> $4 AND NOT cm.deleted LIMIT $5 ) g`, channelID, gid, water.Inbox, viewerUserID, domain.MaxDialogUnreadCount).Scan(&unread); err != nil { return domain.ChannelForumTopic{}, fmt.Errorf("general unread: %w", err) } return domain.ChannelForumTopic{ ChannelID: channelID, TopicID: gid, Title: "General", CreatorUserID: channel.CreatorUserID, Date: channel.Date, TopMessageID: topMsg, ReadInboxMaxID: water.Inbox, ReadOutboxMaxID: water.Outbox, UnreadCount: unread, UnreadMentionsCount: s.countChannelUnreadMentionsForTop(ctx, viewerUserID, channelID, gid), UnreadReactionsCount: s.countChannelUnreadReactionsForTop(ctx, viewerUserID, channelID, gid, member.AvailableMinID), }, nil }