hyapp-server/services/room-service/internal/room/service/im_moderation.go

package service

import (
	"context"
	"fmt"
	"strings"
	"time"

	"google.golang.org/protobuf/proto"
	roomeventsv1 "hyapp.local/api/proto/events/room/v1"
	"hyapp/pkg/appcode"
)

const (
	// forbid_send_msg 官方单接口上限为 200 QPS；双节点各 100 QPS，避免共享 SDKAppID 被单服务打满。
	imMuteProjectionInterval = 10 * time.Millisecond
	// 云 API 执行期间若 Room Cell 状态继续变化，最多在同一门闩内追赶三次，之后交给 outbox 退避重试。
	imMuteProjectionMaxAttempts = 3
	// REST client 单次上限为 5s；总预算额外覆盖进程内门闩、限速和最新快照复核。
	imMuteProjectionTimeout = 6 * time.Second
)

type imMuteProjectionKey struct {
	appCode string
	roomID  string
	userID  int64
}

// imMuteProjectionGate 让同一 app/room/user 的旧 mute 和新 unmute 在单进程内串行，
// 同时允许等待者遵守 context deadline；Room Cell 最新状态仍是每次云端写入的唯一输入。
type imMuteProjectionGate struct {
	token chan struct{}
	refs  int
}

// syncIMUserMuteFromOutbox 只消费已经随 Room Cell 命令原子提交的 RoomUserMuted 事实。
// 普通群消息不会调用该路径，因此腾讯 IM 回调关闭后，共享 SDKAppID 的其它项目不再依赖本服务延迟或可用性。
func (s *Service) syncIMUserMuteFromOutbox(ctx context.Context, envelope *roomeventsv1.EventEnvelope) error {
	if s == nil || s.imUserMuter == nil || envelope == nil || envelope.GetEventType() != "RoomUserMuted" {
		return nil
	}

	var event roomeventsv1.RoomUserMuted
	if err := proto.Unmarshal(envelope.GetBody(), &event); err != nil {
		return fmt.Errorf("decode im mute outbox event: %w", err)
	}
	if event.GetTargetUserId() <= 0 || strings.TrimSpace(envelope.GetRoomId()) == "" {
		return fmt.Errorf("im mute outbox event is incomplete")
	}
	return s.reconcileIMUserMute(ctx, envelope.GetRoomId(), event.GetTargetUserId())
}

// reconcileIMUserMute 在每次 forbid_send_msg 前后重读 Room Cell。
// outbox 可能因网络失败重放旧 mute 事件，不能直接采用历史 payload.muted，否则会覆盖较新的 unmute。
func (s *Service) reconcileIMUserMute(parent context.Context, roomID string, userID int64) error {
	if s == nil || s.imUserMuter == nil {
		return nil
	}
	roomID = strings.TrimSpace(roomID)
	if roomID == "" || userID <= 0 {
		return fmt.Errorf("im mute reconcile target is incomplete")
	}

	ctx, cancel := context.WithTimeout(parent, imMuteProjectionTimeout)
	defer cancel()
	release, err := s.acquireIMMuteProjection(ctx, imMuteProjectionKey{
		appCode: appcode.FromContext(ctx),
		roomID:  roomID,
		userID:  userID,
	})
	if err != nil {
		return err
	}
	defer release()

	for attempt := 1; attempt <= imMuteProjectionMaxAttempts; attempt++ {
		before, err := s.currentSnapshot(ctx, roomID)
		if err != nil {
			return err
		}
		if before == nil {
			// 房间已被物理清理时群也不再是活跃投递通道，历史禁言投影可以直接结束。
			return nil
		}
		desiredMuted := containsID(before.GetMuteUserIds(), userID)
		if err := s.waitIMMuteProjectionRate(ctx); err != nil {
			return err
		}
		if err := s.imUserMuter.SetRoomGroupMemberMuted(ctx, roomID, userID, desiredMuted); err != nil {
			return err
		}

		after, err := s.currentSnapshot(ctx, roomID)
		if err != nil {
			return err
		}
		if after == nil || containsID(after.GetMuteUserIds(), userID) == desiredMuted {
			return nil
		}
		// 状态在腾讯请求飞行期间变化；保持同一 keyed gate 并立即投影最新值，避免本轮留下旧云端状态。
	}
	return fmt.Errorf("im mute state kept changing during %d reconcile attempts", imMuteProjectionMaxAttempts)
}

func (s *Service) acquireIMMuteProjection(ctx context.Context, key imMuteProjectionKey) (func(), error) {
	s.imMuteGateMu.Lock()
	if s.imMuteGates == nil {
		s.imMuteGates = make(map[imMuteProjectionKey]*imMuteProjectionGate)
	}
	gate := s.imMuteGates[key]
	if gate == nil {
		gate = &imMuteProjectionGate{token: make(chan struct{}, 1)}
		gate.token <- struct{}{}
		s.imMuteGates[key] = gate
	}
	gate.refs++
	s.imMuteGateMu.Unlock()

	select {
	case <-ctx.Done():
		s.releaseIMMuteProjectionRef(key, gate)
		return nil, ctx.Err()
	case <-gate.token:
		return func() {
			gate.token <- struct{}{}
			s.releaseIMMuteProjectionRef(key, gate)
		}, nil
	}
}

func (s *Service) releaseIMMuteProjectionRef(key imMuteProjectionKey, gate *imMuteProjectionGate) {
	s.imMuteGateMu.Lock()
	defer s.imMuteGateMu.Unlock()
	gate.refs--
	if gate.refs == 0 && s.imMuteGates[key] == gate {
		delete(s.imMuteGates, key)
	}
}

func (s *Service) waitIMMuteProjectionRate(ctx context.Context) error {
	s.imMuteRateMu.Lock()
	now := time.Now()
	slot := now
	if s.imMuteNextCall.After(slot) {
		slot = s.imMuteNextCall
	}
	s.imMuteNextCall = slot.Add(imMuteProjectionInterval)
	s.imMuteRateMu.Unlock()

	wait := time.Until(slot)
	if wait <= 0 {
		return nil
	}
	timer := time.NewTimer(wait)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-timer.C:
		return nil
	}
}