internal/store/postgres/channel_helpers.go
Rama main · réplica de lectura del repositorio original.
package postgres
import (
"context"
"database/sql"
"errors"
"fmt"
"github.com/jackc/pgx/v5"
"go.uber.org/zap"
"sort"
"strings"
"telesrv/internal/domain"
"telesrv/internal/store/postgres/sqlcgen"
)
func (s *ChannelStore) SaveChannelDefaultSendAs(ctx context.Context, req domain.SaveChannelDefaultSendAsRequest) (domain.ChannelView, error) {
if req.UserID == 0 || req.ChannelID == 0 {
return domain.ChannelView{}, domain.ErrChannelInvalid
}
var sendAsType sql.NullString
var sendAsID sql.NullInt64
if req.SendAs != nil {
if req.SendAs.Type != domain.PeerTypeUser && req.SendAs.Type != domain.PeerTypeChannel {
return domain.ChannelView{}, domain.ErrChannelInvalid
}
sendAsType = sql.NullString{String: string(req.SendAs.Type), Valid: true}
sendAsID = sql.NullInt64{Int64: req.SendAs.ID, Valid: true}
}
channel, member, err := s.getChannelForMember(ctx, s.db, req.UserID, req.ChannelID)
if err != nil {
return domain.ChannelView{}, err
}
topMessageID := channel.TopMessageID
if topMessageID <= member.AvailableMinID {
topMessageID = member.HistoryClearAnchorID
if topMessageID != member.AvailableMinID {
topMessageID = 0
}
}
if _, err := s.db.Exec(ctx, `
INSERT INTO channel_dialogs (
user_id, channel_id, top_message_id, top_message_date,
read_inbox_max_id, read_outbox_max_id,
default_send_as_peer_type, default_send_as_peer_id
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
ON CONFLICT (user_id, channel_id) DO UPDATE SET
default_send_as_peer_type = EXCLUDED.default_send_as_peer_type,
default_send_as_peer_id = EXCLUDED.default_send_as_peer_id,
updated_at = now()`,
req.UserID,
req.ChannelID,
topMessageID,
channel.Date,
member.ReadInboxMaxID,
member.ReadOutboxMaxID,
sendAsType,
sendAsID,
); err != nil {
return domain.ChannelView{}, fmt.Errorf("save channel default send as: %w", err)
}
// Invalidate this process's cache right after the pool write, so the getChannelDialog below reads the just-written value
// (dialog_light NOTIFY only asynchronously invalidates other instances; this instance's read-your-write must be covered synchronously).
if s.dialogCacheActive(s.db) {
s.dialogCache.delete(req.UserID, req.ChannelID)
}
dialog, err := s.getChannelDialog(ctx, s.db, req.UserID, channel)
if err != nil {
return domain.ChannelView{}, err
}
return domain.ChannelView{Channel: channel, Self: member, Dialog: dialog}, nil
}
func (s *ChannelStore) DeleteChannel(ctx context.Context, req domain.DeleteChannelRequest) (domain.DeleteChannelResult, error) {
if req.UserID == 0 || req.ChannelID == 0 {
return domain.DeleteChannelResult{}, domain.ErrChannelInvalid
}
beginner, ok := s.db.(txBeginner)
if !ok {
return domain.DeleteChannelResult{}, fmt.Errorf("delete channel: db does not support transactions")
}
tx, err := beginner.Begin(ctx)
if err != nil {
return domain.DeleteChannelResult{}, fmt.Errorf("begin delete channel: %w", err)
}
committed := false
defer func() {
if !committed {
_ = tx.Rollback(ctx)
}
}()
channel, member, err := s.getChannelForMember(ctx, tx, req.UserID, req.ChannelID)
if err != nil {
return domain.DeleteChannelResult{}, err
}
if member.Role != domain.ChannelRoleCreator {
return domain.DeleteChannelResult{}, domain.ErrChannelAdminRequired
}
recipients, _ := s.ListActiveChannelMemberIDs(ctx, req.UserID, req.ChannelID, 0)
if _, err := tx.Exec(ctx, `UPDATE channels SET deleted = true, username = NULL, updated_at = now() WHERE id = $1`, req.ChannelID); err != nil {
return domain.DeleteChannelResult{}, fmt.Errorf("mark channel deleted: %w", err)
}
if err := deletePeerUsernameTx(ctx, tx, peerUsernameTypeChannel, req.ChannelID); err != nil {
return domain.DeleteChannelResult{}, err
}
if err := markUserChannelMemberIndexDeletedTx(ctx, tx, req.ChannelID, true); err != nil {
return domain.DeleteChannelResult{}, err
}
channel.Deleted = true
channel.Username = ""
// Also soft-delete the linked monoforum (the channel direct-messages container). Cascade only when the counterpart is
// itself a monoforum—so deleting a parent broadcast channel clears its virtual mono, while (defensively) deleting a mono never
// reverse-deletes the real parent channel. Not cascading would leave an orphan with monoforum=true pointing at a deleted parent (client render crash + DB garbage accumulating with each delete).
var linkedMono *domain.Channel
if channel.LinkedMonoforumID != 0 {
mono, err := getChannelByID(ctx, tx, channel.LinkedMonoforumID)
switch {
case errors.Is(err, domain.ErrChannelInvalid):
// The linked mono no longer exists/is already deleted; no cascade needed.
case err != nil:
return domain.DeleteChannelResult{}, fmt.Errorf("load linked monoforum: %w", err)
case mono.Monoforum:
if _, err := tx.Exec(ctx, `UPDATE channels SET deleted = true, username = NULL, updated_at = now() WHERE id = $1`, mono.ID); err != nil {
return domain.DeleteChannelResult{}, fmt.Errorf("mark linked monoforum deleted: %w", err)
}
if err := deletePeerUsernameTx(ctx, tx, peerUsernameTypeChannel, mono.ID); err != nil {
return domain.DeleteChannelResult{}, err
}
if err := markUserChannelMemberIndexDeletedTx(ctx, tx, mono.ID, true); err != nil {
return domain.DeleteChannelResult{}, err
}
mono.Deleted = true
mono.Username = ""
linkedMono = &mono
}
}
if err := deleteChannelWelcomeMessageDeliveriesTx(ctx, tx, channel.ID); err != nil {
return domain.DeleteChannelResult{}, err
}
if linkedMono != nil {
if err := deleteChannelWelcomeMessageDeliveriesTx(ctx, tx, linkedMono.ID); err != nil {
return domain.DeleteChannelResult{}, err
}
}
if err := tx.Commit(ctx); err != nil {
return domain.DeleteChannelResult{}, fmt.Errorf("commit delete channel: %w", err)
}
committed = true
return domain.DeleteChannelResult{Channel: channel, Recipients: recipients, LinkedMonoforum: linkedMono}, nil
}
func (s *ChannelStore) SearchPublicChannels(ctx context.Context, viewerUserID int64, query string, limit int) (domain.PublicChannelSearchResult, error) {
if viewerUserID == 0 {
return domain.PublicChannelSearchResult{}, domain.ErrChannelInvalid
}
if limit <= 0 || limit > domain.MaxPublicChannelSearchLimit {
limit = domain.MaxPublicChannelSearchLimit
}
queryLower := strings.ToLower(strings.TrimSpace(strings.TrimPrefix(query, "@")))
if queryLower == "" {
return domain.PublicChannelSearchResult{}, nil
}
queryPrefix := escapeLike(queryLower) + "%"
queryLike := "%" + escapeLike(queryLower) + "%"
rows, err := s.db.Query(ctx, `
WITH username_matches AS (
SELECT
peer_id,
MIN(CASE
WHEN username_lower = $2 THEN 0
ELSE 1
END) AS rank
FROM peer_usernames
WHERE peer_type = 'channel'
AND active
AND collectible_id IS NOT NULL
AND (
username_lower = $2
OR username_lower LIKE $3 ESCAPE '\'
)
GROUP BY peer_id
)
SELECT `+channelColumns+`
FROM channels c
LEFT JOIN username_matches um ON um.peer_id = c.id
WHERE NOT c.deleted
AND (c.broadcast OR c.megagroup)
AND NOT EXISTS (
SELECT 1
FROM channel_members m
WHERE m.channel_id = c.id
AND m.user_id = $1
AND m.status = 'active'
)
AND (
um.peer_id IS NOT NULL
OR lower(c.username) = $2
OR lower(c.username) LIKE $3 ESCAPE '\'
OR lower(c.title) LIKE $3 ESCAPE '\'
OR lower(c.username) LIKE $4 ESCAPE '\'
OR lower(c.title) LIKE $4 ESCAPE '\'
)
ORDER BY CASE
WHEN um.rank = 0 OR lower(c.username) = $2 THEN 0
WHEN um.rank = 1 OR lower(c.username) LIKE $3 ESCAPE '\' THEN 1
WHEN lower(c.username) LIKE $4 ESCAPE '\' THEN 2
WHEN lower(c.title) LIKE $3 ESCAPE '\' THEN 3
ELSE 4
END,
c.participants_count DESC,
c.date DESC,
c.id DESC
LIMIT $5`, viewerUserID, queryLower, queryPrefix, queryLike, limit)
if err != nil {
return domain.PublicChannelSearchResult{}, fmt.Errorf("search public channels: %w", err)
}
defer rows.Close()
out := domain.PublicChannelSearchResult{
Results: make([]domain.Channel, 0, limit),
}
for rows.Next() {
ch, err := scanChannel(rows)
if err != nil {
return domain.PublicChannelSearchResult{}, err
}
out.Results = append(out.Results, ch)
}
if err := rows.Err(); err != nil {
return domain.PublicChannelSearchResult{}, err
}
return out, nil
}
func (s *ChannelStore) SearchPublicPosts(ctx context.Context, viewerUserID int64, req domain.ChannelSearchPostsRequest) (domain.ChannelHistory, error) {
query := strings.TrimSpace(req.Query)
hashtag := strings.TrimSpace(req.Hashtag)
if (query == "") == (hashtag == "") {
return domain.ChannelHistory{}, domain.ErrChannelInvalid
}
limit := req.Limit
if limit <= 0 || limit > domain.MaxChannelSearchPostsLimit {
limit = domain.MaxChannelSearchPostsLimit
}
args := []any{}
where := `NOT deleted
AND body <> ''
AND EXISTS (
SELECT 1
FROM channels c
WHERE c.id = channel_messages.channel_id
AND NOT c.deleted
AND COALESCE(c.username, '') <> ''
)`
if query != "" {
args = append(args, "%"+escapeLike(query)+"%")
where += fmt.Sprintf(" AND body ILIKE $%d ESCAPE '\\'", len(args))
}
if hashtag != "" {
args = append(args, "%#"+escapeLike(hashtag)+"%")
where += fmt.Sprintf(" AND body ILIKE $%d ESCAPE '\\'", len(args))
}
switch {
case req.OffsetRate > 0 && req.OffsetChannelID > 0 && req.OffsetID > 0:
args = append(args, req.OffsetRate, req.OffsetChannelID, req.OffsetID)
n := len(args)
where += fmt.Sprintf(" AND (message_date < $%d OR (message_date = $%d AND (channel_id < $%d OR (channel_id = $%d AND id < $%d))))", n-2, n-2, n-1, n-1, n)
case req.OffsetRate > 0:
args = append(args, req.OffsetRate)
where += fmt.Sprintf(" AND message_date < $%d", len(args))
case req.OffsetChannelID > 0 && req.OffsetID > 0:
args = append(args, req.OffsetChannelID, req.OffsetID)
n := len(args)
where += fmt.Sprintf(" AND (channel_id < $%d OR (channel_id = $%d AND id < $%d))", n-1, n-1, n)
case req.OffsetID > 0:
args = append(args, req.OffsetID)
where += fmt.Sprintf(" AND id < $%d", len(args))
}
queryLimit := limit + 1
args = append(args, queryLimit)
rows, err := s.db.Query(ctx, `
SELECT `+channelMessageColumns+`
FROM channel_messages
WHERE `+where+`
ORDER BY message_date DESC, channel_id DESC, id DESC
LIMIT $`+fmt.Sprint(len(args)), args...)
if err != nil {
return domain.ChannelHistory{}, fmt.Errorf("search public channel posts: %w", err)
}
defer rows.Close()
out := domain.ChannelHistory{}
channelRefs := make(map[int64]struct{})
for rows.Next() {
msg, err := scanChannelMessage(rows)
if err != nil {
return domain.ChannelHistory{}, err
}
out.Messages = append(out.Messages, msg)
channelRefs[msg.ChannelID] = struct{}{}
}
if err := rows.Err(); err != nil {
return domain.ChannelHistory{}, err
}
if len(out.Messages) > limit {
out.Messages = out.Messages[:limit]
out.Count = limit + 1
channelRefs = make(map[int64]struct{}, len(out.Messages))
for _, msg := range out.Messages {
channelRefs[msg.ChannelID] = struct{}{}
}
} else {
out.Count = len(out.Messages)
}
channels, err := listChannelsByIDs(ctx, s.db, mapKeysInt64(channelRefs))
if err != nil {
return domain.ChannelHistory{}, err
}
out.Channels = channels
if err := s.populateChannelMessagesReactions(ctx, s.db, viewerUserID, out.Channels, out.Messages); err != nil {
return domain.ChannelHistory{}, err
}
return out, nil
}
func (s *ChannelStore) ListActiveChannelIDsForUser(ctx context.Context, userID, afterChannelID int64, limit int) ([]int64, error) {
if userID == 0 || afterChannelID < 0 {
return nil, domain.ErrChannelInvalid
}
if limit <= 0 || limit > domain.MaxSynchronousChannelDialogFanout {
limit = domain.MaxSynchronousChannelDialogFanout
}
rows, err := s.db.Query(ctx, `
WITH visible_channels AS (
SELECT channel_id
FROM user_channel_member_index
WHERE user_id = $1 AND status = 'active' AND NOT deleted
UNION
SELECT mono.id
FROM channels mono
JOIN channels parent ON parent.id = mono.linked_monoforum_id
AND NOT parent.deleted AND parent.broadcast_messages_allowed AND parent.linked_monoforum_id = mono.id
WHERE mono.monoforum AND NOT mono.deleted
AND (EXISTS (
SELECT 1 FROM channel_members admin
WHERE admin.channel_id = parent.id AND admin.user_id = $1 AND admin.status = 'active'
AND (admin.role = 'creator' OR (admin.role = 'admin' AND COALESCE((admin.admin_rights->>'ManageDirectMessages')::boolean, false)))
) OR EXISTS (
SELECT 1 FROM channel_messages message
WHERE message.channel_id = mono.id AND message.saved_peer_type = 'user' AND message.saved_peer_id = $1 AND NOT message.deleted
))
)
SELECT channel_id FROM visible_channels
WHERE channel_id > $2
ORDER BY channel_id
LIMIT $3`, userID, afterChannelID, limit)
if err != nil {
return nil, fmt.Errorf("list active channel ids for user: %w", err)
}
defer rows.Close()
out := make([]int64, 0, limit)
for rows.Next() {
var channelID int64
if err := rows.Scan(&channelID); err != nil {
return nil, err
}
out = append(out, channelID)
}
return out, rows.Err()
}
func (s *ChannelStore) ListDirtyActiveChannelsForUser(ctx context.Context, userID int64, sinceDate int, afterChannelID int64, limit int) ([]domain.DirtyChannel, error) {
if userID == 0 || sinceDate <= 0 || afterChannelID < 0 {
return nil, domain.ErrChannelInvalid
}
if limit <= 0 || limit > domain.MaxChannelDifferenceLimit {
limit = domain.MaxChannelDifferenceLimit
}
rows, err := s.db.Query(ctx, `
WITH visible_channels AS (
SELECT channel_id
FROM user_channel_member_index
WHERE user_id = $1 AND status = 'active' AND NOT deleted
UNION
SELECT mono.id
FROM channels mono
JOIN channels parent ON parent.id = mono.linked_monoforum_id
AND NOT parent.deleted AND parent.broadcast_messages_allowed AND parent.linked_monoforum_id = mono.id
WHERE mono.monoforum AND NOT mono.deleted
AND (EXISTS (
SELECT 1 FROM channel_members admin
WHERE admin.channel_id = parent.id AND admin.user_id = $1 AND admin.status = 'active'
AND (admin.role = 'creator' OR (admin.role = 'admin' AND COALESCE((admin.admin_rights->>'ManageDirectMessages')::boolean, false)))
) OR EXISTS (
SELECT 1 FROM channel_messages message
WHERE message.channel_id = mono.id AND message.saved_peer_type = 'user' AND message.saved_peer_id = $1 AND NOT message.deleted
))
)
SELECT visible.channel_id, c.pts
FROM visible_channels visible
JOIN channels c ON c.id = visible.channel_id AND NOT c.deleted
JOIN channel_update_checkpoints cp ON cp.channel_id = visible.channel_id
WHERE visible.channel_id > $3
AND cp.latest_event_date > $2
ORDER BY visible.channel_id ASC
LIMIT $4`, userID, sinceDate, afterChannelID, limit)
if err != nil {
return nil, fmt.Errorf("list dirty active channels for user: %w", err)
}
defer rows.Close()
byChannelID := make(map[int64]domain.DirtyChannel, limit*2)
for rows.Next() {
var item domain.DirtyChannel
if err := rows.Scan(&item.ChannelID, &item.Pts); err != nil {
return nil, err
}
item.ChannelUpdatesDirty = true
byChannelID[item.ChannelID] = item
}
if err := rows.Err(); err != nil {
return nil, err
}
clearRows, err := s.db.Query(ctx, `
SELECT i.channel_id, c.pts, i.available_min_id, i.history_clear_updated_at
FROM user_channel_member_index i
JOIN channels c ON c.id = i.channel_id AND NOT c.deleted
WHERE i.user_id = $1
AND i.status = 'active'
AND NOT i.deleted
AND i.channel_id > $3
AND i.history_clear_anchor_id > 0
AND i.history_clear_anchor_id = i.available_min_id
AND i.history_clear_updated_at >= $2
ORDER BY i.channel_id ASC
LIMIT $4`, userID, sinceDate, afterChannelID, limit)
if err != nil {
return nil, fmt.Errorf("list owner-local channel history clears for user: %w", err)
}
defer clearRows.Close()
for clearRows.Next() {
var item domain.DirtyChannel
if err := clearRows.Scan(&item.ChannelID, &item.Pts, &item.AvailableMinID, &item.HistoryClearDate); err != nil {
return nil, err
}
if existing, ok := byChannelID[item.ChannelID]; ok {
item.ChannelUpdatesDirty = existing.ChannelUpdatesDirty
}
byChannelID[item.ChannelID] = item
}
if err := clearRows.Err(); err != nil {
return nil, err
}
out := make([]domain.DirtyChannel, 0, len(byChannelID))
for _, item := range byChannelID {
out = append(out, item)
}
sort.Slice(out, func(i, j int) bool { return out[i].ChannelID < out[j].ChannelID })
if len(out) > limit {
out = out[:limit]
}
return out, nil
}
type rowScanner interface {
Scan(dest ...any) error
}
func (s *ChannelStore) getChannelForViewer(ctx context.Context, db sqlcgen.DBTX, viewerUserID, channelID int64) (domain.Channel, domain.ChannelMember, bool, error) {
ch, member, err := s.getChannelForMember(ctx, db, viewerUserID, channelID)
if err == nil {
return ch, member, false, nil
}
if !errors.Is(err, domain.ErrChannelPrivate) {
return domain.Channel{}, domain.ChannelMember{}, false, err
}
ch, err = s.channelByID(ctx, db, channelID)
if err != nil {
return domain.Channel{}, domain.ChannelMember{}, false, err
}
if guest, ok, guestErr := s.getLinkedDiscussionGuest(ctx, db, viewerUserID, ch); guestErr != nil {
return domain.Channel{}, domain.ChannelMember{}, false, guestErr
} else if ok {
return ch, guest, true, nil
}
if member, _, ok, err := s.monoforumAdminPreview(ctx, db, viewerUserID, ch); err != nil {
return domain.Channel{}, domain.ChannelMember{}, false, err
} else if ok {
return ch, member, true, nil
}
if ch.Monoforum && ch.LinkedMonoforumID != 0 {
parent, parentErr := s.channelByID(ctx, db, ch.LinkedMonoforumID)
if parentErr != nil {
return domain.Channel{}, domain.ChannelMember{}, false, parentErr
}
if parent.BroadcastMessagesAllowed && parent.LinkedMonoforumID == ch.ID {
return ch, syntheticMonoforumUserMember(ch, viewerUserID), true, nil
}
}
publicUsernameIDs, err := activeCollectibleUsernamePeerIDs(ctx, db, peerUsernameTypeChannel, []int64{ch.ID})
if err != nil {
return domain.Channel{}, domain.ChannelMember{}, false, err
}
_, hasActiveUsername := publicUsernameIDs[ch.ID]
if !publicPreviewableChannel(ch, hasActiveUsername) {
return domain.Channel{}, domain.ChannelMember{}, false, domain.ErrChannelPrivate
}
member, err = s.getPublicPreviewMember(ctx, db, viewerUserID, ch)
if err != nil {
return domain.Channel{}, domain.ChannelMember{}, false, err
}
return ch, member, true, nil
}
// channelMessageVisibleToViewer applies the message-level half of synthetic monoforum access.
// Subscribers do not have channel_members rows and may only address saved_peer=self; a synthetic
// manager view may address every subscriber sub-dialog.
func channelMessageVisibleToViewer(channel domain.Channel, member domain.ChannelMember, viewerUserID int64, msg domain.ChannelMessage) bool {
if !channel.Monoforum {
return true
}
if member.CanManageDirectMessages() {
return true
}
return msg.SavedPeer == (domain.Peer{Type: domain.PeerTypeUser, ID: viewerUserID})
}
func getChannelByID(ctx context.Context, db sqlcgen.DBTX, channelID int64) (domain.Channel, error) {
ch, err := scanChannel(db.QueryRow(ctx, `SELECT `+channelColumns+` FROM channels c WHERE c.id = $1 AND NOT c.deleted`, channelID))
if errors.Is(err, pgx.ErrNoRows) {
return domain.Channel{}, domain.ErrChannelInvalid
}
return ch, err
}
// channelByID is the cache read-through version of getChannelByID, consuming the cache only on a pool handle (bypassed inside transactions).
// A hit returns the cached channel row; a miss queries PG and backfills. Invalidation is driven in real time by ReadModelChangeListener.
func (s *ChannelStore) channelByID(ctx context.Context, db sqlcgen.DBTX, channelID int64) (domain.Channel, error) {
if s.cacheActive(db) {
return s.rowCache.getOrLoad(ctx, channelID, func() (domain.Channel, error) {
return getChannelByID(ctx, db, channelID)
})
}
return getChannelByID(ctx, db, channelID)
}
func listChannelsByIDs(ctx context.Context, db sqlcgen.DBTX, ids []int64) ([]domain.Channel, error) {
if len(ids) == 0 {
return nil, nil
}
rows, err := db.Query(ctx, `SELECT `+channelColumns+` FROM channels c WHERE c.id = ANY($1::bigint[]) AND NOT c.deleted ORDER BY c.id ASC`, ids)
if err != nil {
return nil, fmt.Errorf("list channels by ids: %w", err)
}
defer rows.Close()
out := make([]domain.Channel, 0, len(ids))
for rows.Next() {
ch, err := scanChannel(rows)
if err != nil {
return nil, err
}
out = append(out, ch)
}
if err := rows.Err(); err != nil {
return nil, err
}
return out, nil
}
// channelsByIDs hydrates viewer-independent channel rows in one batch and
// reuses them across owners. The per-viewer member/dialog state stays outside
// this cache and is read by its own owner-scoped query.
func (s *ChannelStore) channelsByIDs(ctx context.Context, db sqlcgen.DBTX, ids []int64) (map[int64]domain.Channel, error) {
load := func(ctx context.Context, missing []int64) (map[int64]domain.Channel, error) {
channels, err := listChannelsByIDs(ctx, db, missing)
if err != nil {
return nil, err
}
out := make(map[int64]domain.Channel, len(channels))
for _, channel := range channels {
out[channel.ID] = channel
}
return out, nil
}
if s.cacheActive(db) {
return s.rowCache.getOrLoadBatch(ctx, ids, load)
}
return load(ctx, ids)
}
func listChannelsByIDsInOrder(ctx context.Context, db sqlcgen.DBTX, ids []int64) ([]domain.Channel, error) {
channels, err := listChannelsByIDs(ctx, db, ids)
if err != nil {
return nil, err
}
byID := make(map[int64]domain.Channel, len(channels))
for _, channel := range channels {
byID[channel.ID] = channel
}
out := make([]domain.Channel, 0, len(ids))
for _, id := range ids {
if channel, ok := byID[id]; ok {
out = append(out, channel)
}
}
return out, nil
}
func listUsersByIDs(ctx context.Context, db sqlcgen.DBTX, ids []int64) ([]domain.User, error) {
if len(ids) == 0 {
return nil, nil
}
rows, err := db.Query(ctx, `
SELECT id, access_hash, phone, first_name, last_name, username, country_code, verified, support, is_bot, bot_info_version,
COALESCE(EXTRACT(EPOCH FROM premium_expires_at), 0)::bigint AS premium_until,
emoji_status_document_id,
emoji_status_until
FROM users
WHERE id = ANY($1::bigint[])
ORDER BY id ASC`, ids)
if err != nil {
return nil, fmt.Errorf("list users by ids: %w", err)
}
defer rows.Close()
out := make([]domain.User, 0, len(ids))
for rows.Next() {
var u domain.User
var premiumUntil, emojiStatusUntil int64
if err := rows.Scan(&u.ID, &u.AccessHash, &u.Phone, &u.FirstName, &u.LastName, &u.Username, &u.CountryCode, &u.Verified, &u.Support, &u.Bot, &u.BotInfoVersion, &premiumUntil, &u.EmojiStatusDocumentID, &emojiStatusUntil); err != nil {
return nil, err
}
u.PremiumUntil = int(premiumUntil)
u.EmojiStatusUntil = int(emojiStatusUntil)
out = append(out, u)
}
if err := rows.Err(); err != nil {
return nil, err
}
return out, nil
}
type channelReplyStatKey struct {
channelID int64
rootID int
}
func (s *ChannelStore) topicWithViewerCounters(ctx context.Context, viewerUserID, channelID int64, topic domain.ChannelForumTopic, readMaxID, availableMinID int) domain.ChannelForumTopic {
topic.UnreadCount = s.channelThreadUnreadCount(ctx, channelID, topic.TopicID, viewerUserID, readMaxID)
topic.UnreadMentionsCount = s.countChannelUnreadMentionsForTop(ctx, viewerUserID, channelID, topic.TopicID)
topic.UnreadReactionsCount = s.countChannelUnreadReactionsForTop(ctx, viewerUserID, channelID, topic.TopicID, availableMinID)
return topic
}
func (s *ChannelStore) populateForumTopicViewerCounters(ctx context.Context, viewerUserID, channelID int64, topics []domain.ChannelForumTopic, availableMinID int) error {
if len(topics) == 0 || viewerUserID == 0 || channelID == 0 {
return nil
}
roots := make([]int32, 0, len(topics))
seen := make(map[int]struct{}, len(topics))
indexes := make(map[int][]int, len(topics))
for i := range topics {
topics[i].UnreadCount = 0
topics[i].UnreadMentionsCount = 0
topics[i].UnreadReactionsCount = 0
rootID := topics[i].TopicID
indexes[rootID] = append(indexes[rootID], i)
if rootID <= 0 {
continue
}
if _, ok := seen[rootID]; ok {
continue
}
seen[rootID] = struct{}{}
roots = append(roots, int32(rootID))
}
if len(roots) == 0 {
return nil
}
// per-topic read watermark: each topic computes unread from its own read_inbox_max_id and sends down the real read positions (removing the dead column),
// eliminating channel-level single-watermark crosstalk.
waters, err := s.channelTopicReadBatch(ctx, channelID, viewerUserID, roots, availableMinID)
if err != nil {
return err
}
inbox := make(map[int]int, len(waters))
for i := range topics {
w := waters[topics[i].TopicID]
inbox[topics[i].TopicID] = w.Inbox
topics[i].ReadInboxMaxID = w.Inbox
topics[i].ReadOutboxMaxID = w.Outbox
}
if err := s.populateForumTopicUnreadCounts(ctx, channelID, roots, viewerUserID, inbox, indexes, topics); err != nil {
return err
}
if err := s.populateForumTopicUnreadMentionCounts(ctx, viewerUserID, channelID, roots, indexes, topics); err != nil {
return err
}
if err := s.populateForumTopicUnreadReactionCounts(ctx, viewerUserID, channelID, roots, availableMinID, indexes, topics); err != nil {
return err
}
return nil
}
func (s *ChannelStore) populateForumTopicUnreadCounts(ctx context.Context, channelID int64, roots []int32, viewerUserID int64, waters map[int]int, indexes map[int][]int, topics []domain.ChannelForumTopic) error {
// Pair each root with its own per-topic read watermark; after unnesting the pairs, COUNT per topic by its own threshold.
readMaxes := make([]int32, len(roots))
for i, r := range roots {
readMaxes[i] = int32(waters[int(r)])
}
rows, err := s.db.Query(ctx, `
SELECT cm.reply_to_top_id, COUNT(*)::int
FROM channel_messages cm
JOIN unnest($2::int[], $3::int[]) AS w(topic_id, read_max) ON cm.reply_to_top_id = w.topic_id
WHERE cm.channel_id = $1
AND cm.id > w.read_max
AND cm.sender_user_id <> $4
AND NOT cm.deleted
GROUP BY cm.reply_to_top_id`, channelID, roots, readMaxes, viewerUserID)
if err != nil {
return fmt.Errorf("count forum topic unread messages: %w", err)
}
defer rows.Close()
for rows.Next() {
var rootID, count int
if err := rows.Scan(&rootID, &count); err != nil {
return err
}
for _, idx := range indexes[rootID] {
topics[idx].UnreadCount = count
}
}
if err := rows.Err(); err != nil {
return err
}
return nil
}
func (s *ChannelStore) populateForumTopicUnreadMentionCounts(ctx context.Context, userID, channelID int64, roots []int32, indexes map[int][]int, topics []domain.ChannelForumTopic) error {
rows, err := s.db.Query(ctx, `
SELECT top_message_id, COUNT(*)::int
FROM channel_unread_mentions
WHERE user_id = $1
AND channel_id = $2
AND top_message_id = ANY($3::int[])
AND unread
GROUP BY top_message_id`, userID, channelID, roots)
if err != nil {
return fmt.Errorf("count forum topic unread mentions: %w", err)
}
defer rows.Close()
for rows.Next() {
var rootID, count int
if err := rows.Scan(&rootID, &count); err != nil {
return err
}
for _, idx := range indexes[rootID] {
topics[idx].UnreadMentionsCount = count
}
}
if err := rows.Err(); err != nil {
return err
}
return nil
}
func (s *ChannelStore) populateForumTopicUnreadReactionCounts(ctx context.Context, userID, channelID int64, roots []int32, availableMinID int, indexes map[int][]int, topics []domain.ChannelForumTopic) error {
rows, err := s.db.Query(ctx, `
WITH reaction_messages AS (
SELECT
r.message_id,
CASE
WHEN cm.id = ANY($3::int[]) THEN cm.id
ELSE COALESCE(NULLIF(cm.reply_to_top_id, 0), NULLIF(cm.reply_to_msg_id, 0), 0)
END AS topic_id
FROM channel_message_reactions r
JOIN channel_messages cm ON cm.channel_id = r.channel_id AND cm.id = r.message_id
WHERE r.sender_user_id = $1
AND r.channel_id = $2
AND r.unread
AND r.reacted_user_id <> $1
AND cm.id > $4
AND NOT cm.deleted
AND (
cm.id = ANY($3::int[])
OR COALESCE(NULLIF(cm.reply_to_top_id, 0), NULLIF(cm.reply_to_msg_id, 0), 0) = ANY($3::int[])
)
)
SELECT topic_id, COUNT(DISTINCT message_id)::int
FROM reaction_messages
WHERE topic_id <> 0
GROUP BY topic_id`, userID, channelID, roots, availableMinID)
if err != nil {
return fmt.Errorf("count forum topic unread reactions: %w", err)
}
defer rows.Close()
for rows.Next() {
var rootID, count int
if err := rows.Scan(&rootID, &count); err != nil {
return err
}
for _, idx := range indexes[rootID] {
topics[idx].UnreadReactionsCount = count
}
}
if err := rows.Err(); err != nil {
return err
}
return nil
}
func (s *ChannelStore) resolveChannelReply(ctx context.Context, db sqlcgen.DBTX, req domain.SendChannelMessageRequest, member domain.ChannelMember, channel domain.Channel, selfBoostsApplied int) (*domain.MessageReply, error) {
if req.ReplyTo == nil {
return nil, nil
}
if err := domain.ValidateMessageReplyBounds(req.ReplyTo); err != nil {
return nil, err
}
peer := req.ReplyTo.Peer
channelPeer := domain.Peer{Type: domain.PeerTypeChannel, ID: req.ChannelID}
if peer.ID == 0 {
peer = channelPeer
}
if peer != channelPeer {
// inputReplyToMessage.reply_to_peer_id may deliberately reference a
// message from another dialog (the official clients expose this as
// "Reply in another chat"). Keep that source pair intact instead of
// resolving it as a destination-channel thread reply.
if req.ReplyTo.MessageID <= 0 {
return nil, domain.ErrReplyMessageIDInvalid
}
switch peer.Type {
case domain.PeerTypeUser:
var exists bool
if err := db.QueryRow(ctx, `SELECT EXISTS (
SELECT 1 FROM message_boxes
WHERE owner_user_id=$1 AND peer_type='user' AND peer_id=$2 AND box_id=$3 AND NOT deleted
)`, req.UserID, peer.ID, req.ReplyTo.MessageID).Scan(&exists); err != nil || !exists {
return nil, domain.ErrReplyMessageIDInvalid
}
case domain.PeerTypeChannel:
target, err := s.getChannelMessage(ctx, db, peer.ID, req.ReplyTo.MessageID)
if err != nil || target.Deleted {
return nil, domain.ErrReplyMessageIDInvalid
}
default:
return nil, domain.ErrReplyMessageIDInvalid
}
reply := cloneMessageReply(req.ReplyTo)
reply.Peer = peer
return reply, nil
}
if req.ReplyTo.MessageID == 0 {
if req.ReplyTo.TopMessageID <= 0 {
return nil, domain.ErrReplyMessageIDInvalid
}
if !channel.Forum {
root, err := s.getChannelMessage(ctx, db, req.ChannelID, req.ReplyTo.TopMessageID)
if err != nil {
if errors.Is(err, domain.ErrMessageIDInvalid) || errors.Is(err, pgx.ErrNoRows) {
return nil, domain.ErrReplyMessageIDInvalid
}
return nil, err
}
if root.ID <= member.AvailableMinID || channel.LinkedChatID == 0 {
return nil, domain.ErrReplyMessageIDInvalid
}
source, err := getChannelByID(ctx, db, channel.LinkedChatID)
if err != nil {
if errors.Is(err, domain.ErrChannelInvalid) || errors.Is(err, pgx.ErrNoRows) {
return nil, domain.ErrReplyMessageIDInvalid
}
return nil, err
}
if root.Forward == nil || root.Forward.ChannelPost <= 0 {
return nil, domain.ErrReplyMessageIDInvalid
}
post, err := s.getChannelMessage(ctx, db, source.ID, root.Forward.ChannelPost)
if err != nil {
if errors.Is(err, domain.ErrMessageIDInvalid) || errors.Is(err, pgx.ErrNoRows) {
return nil, domain.ErrReplyMessageIDInvalid
}
return nil, err
}
if !domain.LinkedDiscussionRootMatches(channel, source, root, post) {
return nil, domain.ErrReplyMessageIDInvalid
}
reply := cloneMessageReply(req.ReplyTo)
reply.MessageID, reply.TopMessageID, reply.Peer = root.ID, root.ID, channelPeer
return reply, nil
}
topic, err := s.getForumTopic(ctx, db, req.ChannelID, req.ReplyTo.TopMessageID)
if err != nil {
return nil, domain.ErrReplyMessageIDInvalid
}
if topic.Hidden {
return nil, domain.ErrReplyMessageIDInvalid
}
if topic.Closed && !canManageForumTopic(channel, member, topic, req.UserID, selfBoostsApplied) {
return nil, domain.ErrChannelWriteForbidden
}
reply := cloneMessageReply(req.ReplyTo)
reply.MessageID = 0
reply.Peer = channelPeer
reply.TopMessageID = topic.TopicID
reply.ForumTopic = true
return reply, nil
}
target, err := s.getChannelMessage(ctx, db, req.ChannelID, req.ReplyTo.MessageID)
if err != nil {
if errors.Is(err, domain.ErrMessageIDInvalid) || errors.Is(err, pgx.ErrNoRows) {
return nil, domain.ErrReplyMessageIDInvalid
}
return nil, err
}
if target.Deleted || target.ID <= member.AvailableMinID {
return nil, domain.ErrReplyMessageIDInvalid
}
reply := cloneMessageReply(req.ReplyTo)
reply.MessageID = target.ID
reply.Peer = channelPeer
reply.TopMessageID = target.ID
if target.ReplyTo != nil && target.ReplyTo.TopMessageID > 0 {
reply.TopMessageID = target.ReplyTo.TopMessageID
}
// inputReplyToMessage.top_msg_id is not a criterion for reply validity. Per the TL constructor definition, it "carries
// the topic id only when replying to a forum topic message other than General; if the replied-to message is deleted before the method executes, the server uses
// this value to route the message into the correct topic"—that is, it is a hint for the topic landing spot, not a validation field that must match the server-derived value.
// reply_to_msg_id already uniquely points to an existing and visible message, and the thread root is also derived by the server from the target message,
// so a client sending an unrelated top_msg_id is not enough to reject the whole reply: official clients send
// top_msg_id when replying with media in a non-forum channel (measured: TDesktop 7.0.2 layer 228 sends top_msg_id=1),
// and the old implementation hard-failed on it, making "reply with photo" (via sendMedia, where the field is sent) consistently return 400
// REPLY_MESSAGE_ID_INVALID, while the same reply via sendMessage (which omits the field) succeeded.
if req.ReplyTo.TopMessageID > 0 && req.ReplyTo.TopMessageID != reply.TopMessageID {
s.log.Info("channel reply adopts the server-derived top_msg_id, ignoring the client hint value",
zap.Int64("channel_id", req.ChannelID),
zap.Int("client_top_msg_id", req.ReplyTo.TopMessageID),
zap.Int("computed_top_msg_id", reply.TopMessageID),
zap.Int("reply_to_msg_id", req.ReplyTo.MessageID),
)
}
if channel.Forum && reply.TopMessageID > 0 {
if topic, err := s.getForumTopic(ctx, db, req.ChannelID, reply.TopMessageID); err == nil && !topic.Hidden {
if topic.Closed && !canManageForumTopic(channel, member, topic, req.UserID, selfBoostsApplied) {
return nil, domain.ErrChannelWriteForbidden
}
reply.ForumTopic = true
} else if err != nil && !errors.Is(err, domain.ErrMessageIDInvalid) {
return nil, err
}
}
return reply, nil
}
func visibleChannelTopAfter(ctx context.Context, db sqlcgen.DBTX, channelID int64, availableMinID int, fallbackDate int) (int, int, error) {
var id, date int
err := db.QueryRow(ctx, `
SELECT id, message_date
FROM channel_messages
WHERE channel_id = $1 AND id > $2 AND NOT deleted
ORDER BY id DESC
LIMIT 1`, channelID, availableMinID).Scan(&id, &date)
if errors.Is(err, pgx.ErrNoRows) {
return 0, fallbackDate, nil
}
if err != nil {
return 0, 0, fmt.Errorf("select visible channel top: %w", err)
}
return id, date, nil
}
func discussionGroupUpdateResult(changed map[int64]domain.Channel) domain.DiscussionGroupUpdateResult {
ids := make([]int64, 0, len(changed))
for id := range changed {
ids = append(ids, id)
}
sort.Slice(ids, func(i, j int) bool {
return ids[i] < ids[j]
})
out := domain.DiscussionGroupUpdateResult{Channels: make([]domain.Channel, 0, len(ids))}
for _, id := range ids {
out.Channels = append(out.Channels, changed[id])
}
return out
}
func canChangeChannelInfo(member domain.ChannelMember) bool {
return member.Role == domain.ChannelRoleCreator || (member.Role == domain.ChannelRoleAdmin && member.AdminRights.ChangeInfo)
}
func boolPtr(v bool) *bool {
return &v
}
func channelInitialAvailableMinID(channel domain.Channel) int {
if channel.PreHistoryHidden {
return channel.TopMessageID
}
return 0
}
func adminRightsSubset(want, have domain.ChannelAdminRights) bool {
return (!want.ChangeInfo || have.ChangeInfo) &&
(!want.PostMessages || have.PostMessages) &&
(!want.EditMessages || have.EditMessages) &&
(!want.DeleteMessages || have.DeleteMessages) &&
(!want.PostStories || have.PostStories) &&
(!want.EditStories || have.EditStories) &&
(!want.DeleteStories || have.DeleteStories) &&
(!want.BanUsers || have.BanUsers) &&
(!want.InviteUsers || have.InviteUsers) &&
(!want.PinMessages || have.PinMessages) &&
(!want.AddAdmins || have.AddAdmins) &&
(!want.ManageCall || have.ManageCall) &&
(!want.Anonymous || have.Anonymous) &&
(!want.ManageRanks || have.ManageRanks) &&
(!want.ManageDirectMessages || have.ManageDirectMessages)
}
// checkEditMemberRank validates a rank-only (member tag) edit: creator edits
// anyone but no one else edits the creator; admins always edit their own tag,
// and with ManageRanks edit plain members plus admins they promoted; plain
// members edit only their own tag and only while neither the channel default
// nor their personal banned rights set edit_rank. Member tags exist only in
// megagroups: broadcast participants must keep an empty rank so the admins
// participant filter stays a pure admin list there.
func checkEditMemberRank(channel domain.Channel, actor, target domain.ChannelMember) error {
if !channel.Megagroup {
return domain.ErrMegagroupIDInvalid
}
if actor.UserID == target.UserID {
if actor.Role == domain.ChannelRoleCreator || actor.Role == domain.ChannelRoleAdmin {
return nil
}
if channel.DefaultBannedRights.EditRank || actor.BannedRights.EditRank {
return domain.ErrChannelRightForbidden
}
return nil
}
if target.Role == domain.ChannelRoleCreator {
return domain.ErrChannelUserCreator
}
if actor.Role == domain.ChannelRoleCreator {
return nil
}
if actor.Role != domain.ChannelRoleAdmin || !actor.AdminRights.ManageRanks {
return domain.ErrChannelAdminRequired
}
if target.Role == domain.ChannelRoleAdmin && target.InviterUserID != actor.UserID {
return domain.ErrChannelRightForbidden
}
return nil
}
func adminLogBanType(previous, next domain.ChannelMember) domain.ChannelAdminLogEventType {
if next.Status == domain.ChannelMemberKicked || next.BannedRights.ViewMessages {
return domain.ChannelAdminLogParticipantKick
}
if previous.Status == domain.ChannelMemberKicked || previous.BannedRights.ViewMessages {
return domain.ChannelAdminLogParticipantUnkick
}
if !zeroChannelBannedRights(next.BannedRights) {
return domain.ChannelAdminLogParticipantBan
}
return domain.ChannelAdminLogParticipantUnban
}
func adminLogSearchText(event domain.ChannelAdminLogEvent) string {
parts := []string{
event.Query,
event.PrevString,
event.NewString,
}
for _, msg := range []*domain.ChannelMessage{event.Message, event.PrevMessage, event.NewMessage} {
if msg != nil {
parts = append(parts, msg.Body)
}
}
return strings.ToLower(strings.TrimSpace(strings.Join(parts, " ")))
}
func adminLogLikePattern(query string) string {
query = strings.ReplaceAll(query, `\`, `\\`)
query = strings.ReplaceAll(query, `%`, `\%`)
query = strings.ReplaceAll(query, `_`, `\_`)
return "%" + query + "%"
}
func (a pgChannelIDAllocator) NextChannelID(ctx context.Context) (int64, error) {
current, err := a.CurrentChannelID(ctx)
if err != nil {
return 0, err
}
return current + 1, nil
}
func (a pgChannelIDAllocator) CurrentChannelID(ctx context.Context) (int64, error) {
var id int64
err := a.db.QueryRow(ctx, `SELECT COALESCE(MAX(id), 0) FROM channels`).Scan(&id)
return id, err
}