internal/store/postgres/channel_membership_batch.go

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

package postgres

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

	"github.com/jackc/pgx/v5"

	"telesrv/internal/domain"
)

func channelMembersForUpdateBatchTx(ctx context.Context, tx pgx.Tx, channelID int64, userIDs []int64) (map[int64]domain.ChannelMember, error) {
	rows, err := tx.Query(ctx, `
SELECT channel_id, user_id, inviter_user_id, role, status, joined_at, left_at,
       admin_rights::text, banned_rights::text, rank, available_min_id, available_min_pts,
       history_clear_anchor_id, history_clear_anchor_date,
       read_inbox_max_id, read_outbox_max_id, unread_mark, slowmode_last_send_date
FROM channel_members
WHERE channel_id = $1 AND user_id = ANY($2::bigint[])
ORDER BY user_id
FOR UPDATE`, channelID, userIDs)
	if err != nil {
		return nil, fmt.Errorf("lock channel invite members: %w", err)
	}
	defer rows.Close()
	out := make(map[int64]domain.ChannelMember, len(userIDs))
	for rows.Next() {
		member, err := scanChannelMember(rows)
		if err != nil {
			return nil, err
		}
		out[member.UserID] = member
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("lock channel invite members: %w", err)
	}
	return out, nil
}

func enableChannelMembershipBatchTx(ctx context.Context, tx pgx.Tx) error {
	if _, err := tx.Exec(ctx, `SELECT set_config('telesrv.membership_batch_mode', 'on', true)`); err != nil {
		return fmt.Errorf("enable channel membership batch invalidation: %w", err)
	}
	return nil
}

func upsertChannelMembersBatchTx(ctx context.Context, tx pgx.Tx, channel domain.Channel, members []domain.ChannelMember) error {
	if len(members) == 0 {
		return nil
	}
	userIDs := make([]int64, len(members))
	inviterIDs := make([]int64, len(members))
	joinedAt := make([]int32, len(members))
	availableMinIDs := make([]int32, len(members))
	availableMinPts := make([]int32, len(members))
	readInboxMaxIDs := make([]int32, len(members))
	for i, member := range members {
		userIDs[i] = member.UserID
		inviterIDs[i] = member.InviterUserID
		joinedAt[i] = int32(member.JoinedAt)
		availableMinIDs[i] = int32(member.AvailableMinID)
		availableMinPts[i] = int32(member.AvailableMinPts)
		readInboxMaxIDs[i] = int32(member.ReadInboxMaxID)
	}
	if _, err := tx.Exec(ctx, `
WITH input AS MATERIALIZED (
    SELECT user_id, inviter_user_id, joined_at, available_min_id, available_min_pts, read_inbox_max_id
    FROM unnest(
        $2::bigint[], $3::bigint[], $4::integer[], $5::integer[], $6::integer[], $7::integer[]
    ) AS value(user_id, inviter_user_id, joined_at, available_min_id, available_min_pts, read_inbox_max_id)
)
INSERT INTO channel_members (
    channel_id, user_id, inviter_user_id, role, status, joined_at, left_at,
    admin_rights, banned_rights, rank, available_min_id, available_min_pts,
    read_inbox_max_id, read_outbox_max_id, unread_mark, slowmode_last_send_date
)
SELECT $1, user_id, inviter_user_id, 'member', 'active', joined_at, 0,
       '{}'::jsonb, '{}'::jsonb, '', available_min_id, available_min_pts,
       read_inbox_max_id, 0, false, 0
FROM input
ORDER BY user_id
ON CONFLICT (channel_id, user_id) DO UPDATE SET
    inviter_user_id = EXCLUDED.inviter_user_id,
    role = EXCLUDED.role,
    status = EXCLUDED.status,
    joined_at = EXCLUDED.joined_at,
    left_at = EXCLUDED.left_at,
    admin_rights = EXCLUDED.admin_rights,
    banned_rights = EXCLUDED.banned_rights,
    rank = EXCLUDED.rank,
    available_min_id = GREATEST(channel_members.available_min_id, EXCLUDED.available_min_id),
    available_min_pts = GREATEST(channel_members.available_min_pts, EXCLUDED.available_min_pts),
    read_inbox_max_id = GREATEST(channel_members.read_inbox_max_id, EXCLUDED.read_inbox_max_id),
    updated_at = now()`, channel.ID, userIDs, inviterIDs, joinedAt, availableMinIDs, availableMinPts, readInboxMaxIDs); err != nil {
		return fmt.Errorf("batch upsert channel members: %w", err)
	}

	if _, err := tx.Exec(ctx, `
WITH input AS MATERIALIZED (
    SELECT user_id
    FROM unnest($2::bigint[]) AS value(user_id)
)
INSERT INTO user_channel_member_index (
    user_id, channel_id, status, megagroup, broadcast, deleted,
    role, left_at, forum, public_username, can_pin_messages
)
SELECT user_id, $1, 'active', $3, $4, $5, 'member', 0, $6, $7, false
FROM input
ORDER BY user_id
ON CONFLICT (user_id, channel_id) DO UPDATE SET
    status = EXCLUDED.status,
    megagroup = EXCLUDED.megagroup,
    broadcast = EXCLUDED.broadcast,
    deleted = EXCLUDED.deleted,
    role = EXCLUDED.role,
    left_at = EXCLUDED.left_at,
    forum = EXCLUDED.forum,
    public_username = EXCLUDED.public_username,
    can_pin_messages = EXCLUDED.can_pin_messages,
    updated_at = now()`, channel.ID, userIDs, channel.Megagroup, channel.Broadcast, channel.Deleted, channel.Forum, channel.Username != ""); err != nil {
		return fmt.Errorf("batch upsert user channel member index: %w", err)
	}
	return nil
}

func insertChannelInviteAdminLogsBatchTx(ctx context.Context, tx pgx.Tx, channelID, inviterUserID int64, date int, members []domain.ChannelMember) error {
	if len(members) == 0 {
		return nil
	}
	type row struct {
		Ordinal     int                  `json:"ordinal"`
		Participant domain.ChannelMember `json:"participant"`
	}
	input := make([]row, len(members))
	for i, member := range members {
		input[i] = row{Ordinal: i + 1, Participant: member}
	}
	payload, err := json.Marshal(input)
	if err != nil {
		return fmt.Errorf("marshal channel invite admin logs: %w", err)
	}
	if _, err := tx.Exec(ctx, `
WITH input AS MATERIALIZED (
    SELECT ordinal, participant
    FROM jsonb_to_recordset($5::jsonb) AS value(ordinal integer, participant jsonb)
), allocated AS MATERIALIZED (
    UPDATE channels
    SET admin_log_seq = admin_log_seq + $4, updated_at = now()
    WHERE id = $1
    RETURNING admin_log_seq
)
INSERT INTO channel_admin_log_events (
    channel_id, id, actor_user_id, event_date, event_type, participant, query
)
SELECT $1, allocated.admin_log_seq - $4 + input.ordinal, $2, $3,
       'participant_invite', input.participant, ''
FROM input
CROSS JOIN allocated
ORDER BY input.ordinal`, channelID, inviterUserID, date, len(members), string(payload)); err != nil {
		return fmt.Errorf("batch insert channel invite admin logs: %w", err)
	}
	return nil
}

func upsertChannelDialogsBatchTx(ctx context.Context, tx pgx.Tx, channel domain.Channel, top domain.ChannelMessage, members []domain.ChannelMember) error {
	if len(members) == 0 {
		return nil
	}
	topDate := top.Date
	if topDate == 0 {
		topDate = channel.Date
	}
	userIDs := make([]int64, len(members))
	readInboxMaxIDs := make([]int32, len(members))
	readOutboxMaxIDs := make([]int32, len(members))
	for i, member := range members {
		userIDs[i] = member.UserID
		readInboxMaxIDs[i] = int32(member.ReadInboxMaxID)
		readOutboxMaxIDs[i] = int32(member.ReadOutboxMaxID)
	}
	if _, err := tx.Exec(ctx, `
WITH input AS MATERIALIZED (
    SELECT user_id, read_inbox_max_id, read_outbox_max_id
    FROM unnest($4::bigint[], $5::integer[], $6::integer[])
         AS value(user_id, read_inbox_max_id, read_outbox_max_id)
)
INSERT INTO channel_dialogs (
    user_id, channel_id, top_message_id, top_message_date,
    read_inbox_max_id, read_outbox_max_id, unread_count, unread_mark
)
SELECT user_id, $1, $2, $3, read_inbox_max_id, read_outbox_max_id, 0, false
FROM input
ORDER BY user_id
ON CONFLICT (user_id, channel_id) DO UPDATE SET
    top_message_id = GREATEST(channel_dialogs.top_message_id, EXCLUDED.top_message_id),
    top_message_date = GREATEST(channel_dialogs.top_message_date, EXCLUDED.top_message_date),
    read_inbox_max_id = GREATEST(channel_dialogs.read_inbox_max_id, EXCLUDED.read_inbox_max_id),
    read_outbox_max_id = GREATEST(channel_dialogs.read_outbox_max_id, EXCLUDED.read_outbox_max_id),
    unread_mark = false,
    updated_at = now()`, channel.ID, channel.TopMessageID, topDate, userIDs, readInboxMaxIDs, readOutboxMaxIDs); err != nil {
		return fmt.Errorf("batch upsert channel dialogs: %w", err)
	}
	if _, err := tx.Exec(ctx, `
UPDATE channel_dialogs AS dialog
SET unread_count = (
        SELECT COUNT(*)::int
        FROM (
            SELECT 1
            FROM channel_messages AS message
            WHERE message.channel_id = dialog.channel_id
              AND message.id > dialog.read_inbox_max_id
              AND message.id <= dialog.top_message_id
              AND message.sender_user_id <> dialog.user_id
              AND NOT message.deleted
            LIMIT $3
        ) AS capped
    ),
    updated_at = now()
WHERE dialog.channel_id = $1
  AND dialog.user_id = ANY($2::bigint[])`, channel.ID, userIDs, domain.MaxDialogUnreadCount); err != nil {
		return fmt.Errorf("batch refresh channel dialog unread count: %w", err)
	}
	return nil
}

func refreshChannelUnreadReactionsCountsBatchTx(ctx context.Context, tx pgx.Tx, channelID int64, userIDs []int64) error {
	if len(userIDs) == 0 {
		return nil
	}
	if _, err := tx.Exec(ctx, `
WITH input AS MATERIALIZED (
    SELECT user_id FROM unnest($2::bigint[]) AS value(user_id)
), counts AS MATERIALIZED (
    SELECT input.user_id,
           (
               SELECT COUNT(DISTINCT reaction.message_id)::int
               FROM channel_message_reactions AS reaction
               JOIN channel_messages AS message
                 ON message.channel_id = reaction.channel_id AND message.id = reaction.message_id
               JOIN channel_members AS member
                 ON member.channel_id = reaction.channel_id AND member.user_id = input.user_id
               WHERE reaction.sender_user_id = input.user_id
                 AND reaction.channel_id = $1
                 AND reaction.unread
                 AND reaction.reacted_user_id <> input.user_id
                 AND message.id > member.available_min_id
                 AND NOT message.deleted
                 AND member.status = 'active'
                 AND NOT COALESCE((member.banned_rights->>'ViewMessages')::boolean, false)
           ) AS count
    FROM input
)
INSERT INTO channel_dialogs (user_id, channel_id, unread_reactions_count)
SELECT user_id, $1, count
FROM counts
ORDER BY user_id
ON CONFLICT (user_id, channel_id) DO UPDATE SET
    unread_reactions_count = EXCLUDED.unread_reactions_count,
    updated_at = now()`, channelID, userIDs); err != nil {
		return fmt.Errorf("batch refresh channel unread reactions count: %w", err)
	}
	return nil
}

func bumpChannelMembershipReadModelsBatchTx(ctx context.Context, tx pgx.Tx, channelID int64, userIDs []int64) error {
	if len(userIDs) == 0 {
		return nil
	}
	if _, err := tx.Exec(ctx, `SELECT public.telesrv_bump_channel_membership_read_models($1, $2::bigint[])`, channelID, userIDs); err != nil {
		return fmt.Errorf("batch bump channel membership read models: %w", err)
	}
	return nil
}