internal/store/postgres/channel_store.go

Rama main · réplica de lectura del repositorio original.

package postgres

import (
	"go.uber.org/zap"

	"telesrv/internal/domain"
	"telesrv/internal/store"
	"telesrv/internal/store/postgres/sqlcgen"
)

const channelDialogQueryLimit = 500

const channelDialogCandidateLimit = 10000

const channelMemberFilterBatch = 1000

const retryableChannelTxAttempts = 3

// ChannelStore implements store.ChannelStore with PostgreSQL.
type ChannelStore struct {
	db              sqlcgen.DBTX
	ids             store.ChannelIDAllocator
	msgIDs          store.ChannelMessageIDAllocator
	log             *zap.Logger
	rowCache        *ChannelRowCache
	topMsgCache     *ChannelTopMessageCache
	memberCache     *ChannelMemberCache
	dialogCache     *ChannelDialogCache
	boostCache      *ChannelBoostCache
	differenceCache *ChannelDifferenceBaseCache
	// paidReactionStartingGrant mirrors app/stars lazy grant policy inside the
	// paid reaction settlement transaction.
	paidReactionStartingGrant int64
}

// ChannelStoreOption adjusts a PostgreSQL ChannelStore dependency.
type ChannelStoreOption func(*ChannelStore)

// WithChannelAllocators injects the Redis-backed channel id / message id allocators.
func WithChannelAllocators(ids store.ChannelIDAllocator, msgIDs store.ChannelMessageIDAllocator) ChannelStoreOption {
	return func(s *ChannelStore) {
		s.ids = ids
		s.msgIDs = msgIDs
	}
}

// WithChannelLogger injects the channel store logger, used to trace channel pts allocation.
func WithChannelLogger(log *zap.Logger) ChannelStoreOption {
	return func(s *ChannelStore) {
		s.log = log
	}
}

// WithChannelRowCache injects the "shared channel row" in-process cache (invalidated in real
// time by ReadModelChangeListener). Passing nil is equivalent to disabling it and does not
// affect any read-path behavior.
func WithChannelRowCache(cache *ChannelRowCache) ChannelStoreOption {
	return func(s *ChannelStore) {
		s.rowCache = cache
	}
}

// WithChannelTopMessageCache injects the shared dialog-top message cache. The
// cache never contains viewer reaction/read overlays.
func WithChannelTopMessageCache(cache *ChannelTopMessageCache) ChannelStoreOption {
	return func(s *ChannelStore) {
		s.topMsgCache = cache
	}
}

// WithChannelMemberCache injects the "channel member/access state" in-process cache.
// Passing nil is equivalent to disabling it; transactions still bypass it, and the read model
// listener invalidates it on commit.
func WithChannelMemberCache(cache *ChannelMemberCache) ChannelStoreOption {
	return func(s *ChannelStore) {
		s.memberCache = cache
	}
}

// WithChannelDialogCache injects the viewer-scoped channel dialog projection cache.
// Passing nil is equivalent to disabling it; transactions still bypass it, avoiding reading a
// stale dialog/read boundary after a write.
func WithChannelDialogCache(cache *ChannelDialogCache) ChannelStoreOption {
	return func(s *ChannelStore) {
		s.dialogCache = cache
	}
}

// WithChannelBoostCache injects the channel boost read-model projection cache (SelfBoostsApplied + peer total).
// Passing nil is equivalent to disabling it; transactions still bypass it, avoiding the send-permission check reading a stale value.
func WithChannelBoostCache(cache *ChannelBoostCache) ChannelStoreOption {
	return func(s *ChannelStore) {
		s.boostCache = cache
	}
}

// WithChannelDifferenceBaseCache injects the shared immutable event/message
// page cache used by updates.getChannelDifference. Viewer access and overlays
// remain outside this cache.
func WithChannelDifferenceBaseCache(cache *ChannelDifferenceBaseCache) ChannelStoreOption {
	return func(s *ChannelStore) {
		s.differenceCache = cache
	}
}

// WithChannelStarsStartingGrant keeps paid reaction's atomic debit aligned
// with the configured app/stars lazy first-use grant.
func WithChannelStarsStartingGrant(amount int64) ChannelStoreOption {
	return func(s *ChannelStore) {
		if amount > 0 {
			s.paidReactionStartingGrant = amount
		} else {
			s.paidReactionStartingGrant = 0
		}
	}
}

// cacheActive reports whether the channel row cache is usable for the current handle: only when
// the cache is enabled and the pool (not a transaction) is used. Inside a transaction
// (db != s.db) the cache is always bypassed for live reads, guaranteeing transactional
// read-your-write.
func (s *ChannelStore) cacheActive(db sqlcgen.DBTX) bool {
	return s.rowCache != nil && db == s.db
}

func (s *ChannelStore) topMessageCacheActive(db sqlcgen.DBTX) bool {
	return s.topMsgCache != nil && db == s.db
}

func (s *ChannelStore) memberCacheActive(db sqlcgen.DBTX) bool {
	return s.memberCache != nil && db == s.db
}

func (s *ChannelStore) dialogCacheActive(db sqlcgen.DBTX) bool {
	return s.dialogCache != nil && db == s.db
}

func (s *ChannelStore) boostCacheActive(db sqlcgen.DBTX) bool {
	return s.boostCache != nil && db == s.db
}

// NewChannelStore creates a ChannelStore backed by a pgx pool (or transaction).
func NewChannelStore(db sqlcgen.DBTX, opts ...ChannelStoreOption) *ChannelStore {
	s := &ChannelStore{db: db, paidReactionStartingGrant: domain.DefaultStarsStartingGrant}
	for _, opt := range opts {
		opt(s)
	}
	if s.log == nil {
		s.log = zap.NewNop()
	}
	if s.ids == nil {
		s.ids = pgChannelIDAllocator{db: db}
	}
	if s.msgIDs == nil {
		s.msgIDs = pgChannelMessageIDAllocator{db: db}
	}
	return s
}

const channelColumns = `c.id, c.access_hash, c.creator_user_id, c.title, c.about, COALESCE(c.username, ''), c.verified, c.scam, c.fake, c.gigagroup,
c.broadcast, c.megagroup, c.forum, c.forum_tabs, c.autotranslation, c.restricted_sponsored, c.broadcast_messages_allowed, c.send_paid_messages_stars, c.noforwards, c.join_to_send, c.join_request, c.signatures, c.signature_profiles, c.pre_history_hidden, c.participants_hidden, c.antispam,
EXISTS (SELECT 1 FROM channel_invites ci WHERE ci.channel_id = c.id AND NOT ci.revoked) AS has_link,
c.linked_chat_id, c.linked_community_id, c.monoforum, c.linked_monoforum_id, c.slowmode_seconds, c.boosts_unrestrict, c.default_banned_rights::text,
c.available_reactions::text, c.color_set, c.color, c.color_background_emoji_id, c.profile_color_set, c.profile_color, c.profile_color_background_emoji_id, c.emoji_status_document_id, c.emoji_status_until,
c.wallpaper::text, c.participants_count, c.admins_count, c.kicked_count, c.banned_count, c.top_message_id, c.pinned_message_id, c.pts,
c.ttl_period, c.date, c.deleted, c.photo_id, c.photo_dc_id, c.photo_stripped,
c.active_call_id, c.active_call_access_hash, c.active_call_not_empty`

const channelMessageColumns = `channel_id, id, random_id, sender_user_id, from_peer_type, from_peer_id,
send_as_peer_type, send_as_peer_id, message_date, edit_date, post, silent, noforwards, body,
entities::text, reply_to::text, reply_to_msg_id, reply_to_peer_type, reply_to_peer_id, reply_to_top_id,
fwd_from::text, discussion_channel_id, discussion_message_id, action::text, pts, deleted, media::text,
reply_markup::text, rich_message::text, ttl_period, expires_at, views_count, post_author, pinned, via_bot_id, grouped_id, from_boosts_applied, saved_peer_type, saved_peer_id, paid_message_stars, suggested_post::text, show_message_sender, factcheck::text`

const channelForumTopicColumns = `channel_id, topic_id, creator_user_id, title, icon_color, icon_emoji_id,
title_missing, closed, hidden, pinned, pinned_order, date, top_message_id, read_inbox_max_id,
read_outbox_max_id, unread_count, unread_mentions_count, unread_reactions_count,
unread_poll_votes_count`

const (
	messageHistoryLoadBackward messageHistoryLoad = iota
	messageHistoryLoadForward
	messageHistoryLoadAround
)