internal/store/postgres/channel_polls.go

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

package postgres

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"

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

// Channel/supergroup message poll voting/closing: membership and message visibility reuse the same validation as reactions,
// and poll-level semantics are delegated to poll.go's shared SQL.

func (s *ChannelStore) VoteChannelMessagePoll(ctx context.Context, req domain.VoteChannelMessagePollRequest) (domain.ChannelMessagePollResult, error) {
	return s.mutateChannelMessagePoll(ctx, req.UserID, req.ChannelID, req.MessageID, req.Date, func(ctx context.Context, tx pgx.Tx, def domain.PollDefinition, date int) error {
		return applyPollVote(ctx, tx, def, req.UserID, req.Options, date)
	})
}

func (s *ChannelStore) CloseChannelMessagePoll(ctx context.Context, req domain.CloseChannelMessagePollRequest) (domain.ChannelMessagePollResult, error) {
	return s.mutateChannelMessagePoll(ctx, req.UserID, req.ChannelID, req.MessageID, req.Date, func(ctx context.Context, tx pgx.Tx, def domain.PollDefinition, _ int) error {
		return closePollAsCreator(ctx, tx, def, req.UserID)
	})
}

func (s *ChannelStore) mutateChannelMessagePoll(
	ctx context.Context,
	userID, channelID int64,
	messageID int,
	date int,
	mutate func(ctx context.Context, tx pgx.Tx, def domain.PollDefinition, date int) error,
) (domain.ChannelMessagePollResult, error) {
	if userID == 0 || channelID == 0 || messageID <= 0 || messageID > domain.MaxMessageBoxID {
		return domain.ChannelMessagePollResult{}, domain.ErrChannelInvalid
	}
	if date == 0 {
		date = int(time.Now().Unix())
	}
	beginner, ok := s.db.(txBeginner)
	if !ok {
		return domain.ChannelMessagePollResult{}, fmt.Errorf("mutate channel message poll: db does not support transactions")
	}
	tx, err := beginner.Begin(ctx)
	if err != nil {
		return domain.ChannelMessagePollResult{}, fmt.Errorf("begin channel poll tx: %w", err)
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback(ctx)
		}
	}()
	channel, member, err := s.getChannelForMember(ctx, tx, userID, channelID)
	if err != nil {
		return domain.ChannelMessagePollResult{}, err
	}
	msg, err := s.getChannelMessage(ctx, tx, channelID, messageID)
	if err != nil {
		return domain.ChannelMessagePollResult{}, err
	}
	if msg.Deleted || msg.Action != nil || msg.ID <= member.AvailableMinID {
		return domain.ChannelMessagePollResult{}, domain.ErrMessageIDInvalid
	}
	if msg.Media == nil || msg.Media.Kind != domain.MessageMediaKindPoll || msg.Media.Poll == nil || msg.Media.Poll.ID == 0 {
		return domain.ChannelMessagePollResult{}, domain.ErrMessageIDInvalid
	}
	pollID := msg.Media.Poll.ID
	defs, err := loadPollDefinitions(ctx, tx, []int64{pollID}, true)
	if err != nil {
		return domain.ChannelMessagePollResult{}, err
	}
	def, ok := defs[pollID]
	if !ok {
		return domain.ChannelMessagePollResult{}, domain.ErrPollNotFound
	}
	if err := mutate(ctx, tx, def, date); err != nil {
		return domain.ChannelMessagePollResult{}, err
	}
	if err := enrichPollMediaRefs(ctx, tx, []pollMediaRef{{media: msg.Media, viewer: userID}}); err != nil {
		return domain.ChannelMessagePollResult{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return domain.ChannelMessagePollResult{}, fmt.Errorf("commit channel poll tx: %w", err)
	}
	committed = true
	return domain.ChannelMessagePollResult{
		PollID:     pollID,
		Channel:    channel,
		Message:    msg,
		Recipients: []int64{userID, msg.SenderUserID},
	}, nil
}

// ChannelPollFanoutViews batch-loads the per-viewer enrich of one poll message over a set of viewers (removing the fan-out
// N+1 of per-viewer GetChannelMessages): viewer-invariant aggregates (counts/total/recent) are computed once via
// pollViewerAggregates(viewer=0) + batch viewerOptions + batch member visibility, and each viewer is composed with the exact same
// domain.ResolvePollResults as the per-viewer path, so it is byte-identical. Member/AvailableMinID visibility replicates
// GetChannelMessages here (active member && msgID>available_min_id); the bot history filter is layered on by the app layer.
func (s *ChannelStore) ChannelPollFanoutViews(ctx context.Context, channelID int64, msgID int, viewers []int64, now int) (domain.ChannelPollFanoutViews, error) {
	out := domain.ChannelPollFanoutViews{Polls: map[int64]*domain.MessagePoll{}}
	if channelID == 0 || msgID <= 0 || len(viewers) == 0 {
		return out, nil
	}
	msg, err := s.getChannelMessage(ctx, s.db, channelID, msgID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return out, nil
		}
		return domain.ChannelPollFanoutViews{}, err
	}
	if msg.Deleted || msg.Media == nil || msg.Media.Kind != domain.MessageMediaKindPoll || msg.Media.Poll == nil || msg.Media.Poll.ID == 0 {
		return out, nil
	}
	pollID := msg.Media.Poll.ID
	defs, err := loadPollDefinitions(ctx, s.db, []int64{pollID}, false)
	if err != nil {
		return domain.ChannelPollFanoutViews{}, err
	}
	def, ok := defs[pollID]
	if !ok {
		return out, nil
	}
	out.Found = true
	out.Message = msg
	// viewer-invariant template: viewerUserID=0 → pollViewerAggregates skips the ViewerOptions block and emits only counts/total/recent.
	tmplAggs, err := pollViewerAggregates(ctx, s.db, 0, []int64{pollID})
	if err != nil {
		return domain.ChannelPollFanoutViews{}, err
	}
	tmpl := tmplAggs[pollID]
	viewerOpts, err := s.batchPollViewerOptions(ctx, pollID, viewers)
	if err != nil {
		return domain.ChannelPollFanoutViews{}, err
	}
	members, err := s.batchChannelMemberAvailableMinID(ctx, channelID, viewers)
	if err != nil {
		return domain.ChannelPollFanoutViews{}, err
	}
	for _, viewer := range viewers {
		if viewer == 0 {
			continue
		}
		availMinID, isMember := members[viewer]
		if !isMember || (availMinID > 0 && msgID <= availMinID) {
			out.Polls[viewer] = nil // evaluated but not visible (not an active member / pre-history hidden)
			continue
		}
		// agg.Counts/RecentVoters are shared with tmpl (ResolvePollResults is read-only and RecentVoters is copied internally), so it is safe.
		agg := domain.PollAggregates{
			Counts:       tmpl.Counts,
			TotalVoters:  tmpl.TotalVoters,
			RecentVoters: tmpl.RecentVoters,
			ViewerOptions: viewerOpts[viewer],
		}
		results := domain.ResolvePollResults(def, agg, viewer, now)
		pollCopy := *msg.Media.Poll // shallow copy: ApplyPollState only writes Closed + a new Results pointer and does not touch the shared def slice
		domain.ApplyPollState(&pollCopy, def, results, now)
		out.Polls[viewer] = &pollCopy
	}
	return out, nil
}

// batchPollViewerOptions fetches a set of viewers' vote options in one poll at once (replacing per-viewer single queries).
func (s *ChannelStore) batchPollViewerOptions(ctx context.Context, pollID int64, viewers []int64) (map[int64][][]byte, error) {
	out := make(map[int64][][]byte, len(viewers))
	if pollID == 0 || len(viewers) == 0 {
		return out, nil
	}
	rows, err := s.db.Query(ctx, `SELECT user_id, options::text FROM poll_votes WHERE poll_id = $1 AND user_id = ANY($2)`, pollID, viewers)
	if err != nil {
		return nil, fmt.Errorf("batch poll viewer options: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var userID int64
		var optionsJSON string
		if err := rows.Scan(&userID, &optionsJSON); err != nil {
			return nil, fmt.Errorf("scan poll viewer option: %w", err)
		}
		options, err := decodePollOptions(optionsJSON)
		if err != nil {
			return nil, err
		}
		out[userID] = options
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate poll viewer options: %w", err)
	}
	return out, nil
}

// batchChannelMemberAvailableMinID fetches a set of viewers' visible membership + available_min_id at once; presence in the map
// = visible for that channel's messages (active member && not restricted by ViewMessages, replicating validateChannelMemberVisible),
// the value being their available_min_id; absence = not visible (not active / left / banned / ViewMessages-restricted).
func (s *ChannelStore) batchChannelMemberAvailableMinID(ctx context.Context, channelID int64, viewers []int64) (map[int64]int, error) {
	out := make(map[int64]int, len(viewers))
	if channelID == 0 || len(viewers) == 0 {
		return out, nil
	}
	rows, err := s.db.Query(ctx, `SELECT user_id, available_min_id, banned_rights::text FROM channel_members WHERE channel_id = $1 AND user_id = ANY($2) AND status = 'active'`, channelID, viewers)
	if err != nil {
		return nil, fmt.Errorf("batch channel member available_min_id: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var userID int64
		var availableMinID int
		var bannedRights string
		if err := rows.Scan(&userID, &availableMinID, &bannedRights); err != nil {
			return nil, fmt.Errorf("scan channel member available_min_id: %w", err)
		}
		var rights domain.ChannelBannedRights
		_ = json.Unmarshal([]byte(bannedRights), &rights)
		if rights.ViewMessages {
			continue // active but restricted from viewing messages → not visible (consistent with validateChannelMemberVisible)
		}
		out[userID] = availableMinID
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate channel member available_min_id: %w", err)
	}
	return out, nil
}

// populateChannelMessagesPolls enriches all poll media in the page from the requesting viewer's perspective;
// it is uniformly mounted by populateChannelMessagesReactions (all channel message read paths share one choke point).
func (s *ChannelStore) populateChannelMessagesPolls(ctx context.Context, db sqlcgen.DBTX, viewerUserID int64, messages []domain.ChannelMessage) error {
	refs := make([]pollMediaRef, 0, 2)
	for i := range messages {
		refs = append(refs, pollMediaRef{media: messages[i].Media, viewer: viewerUserID})
	}
	return enrichPollMediaRefs(ctx, db, refs)
}