astra/owpengram-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(messages): sync saved message tags

Browse source
This commit is contained in:
iamxvbaba 2026-07-25 19:23:14 +08:00
parent a785ae7491
commit 6b3eba6c5d
42 changed files with 1581 additions and 551 deletions
Download patch file Download diff file Expand all files Collapse all files

29
internal/rpc/channels_passive_stubs_rpc_test.go
View file

@ -4,6 +4,7 @@ import (
"github.com/iamxvbaba/td/tg"
"strings"
apptelemetry "telesrv/internal/app/clienttelemetry"
appmessages "telesrv/internal/app/messages"
appmoderation "telesrv/internal/app/moderation"
"telesrv/internal/domain"
"telesrv/internal/store/memory"
@ -865,28 +866,22 @@ func TestTDesktopPassiveChannelStubs(t *testing.T) {
if !ok || staleEmptyPage.Hash != 0 || len(staleEmptyPage.Tags) != 0 {
t.Fatalf("messages.getSavedReactionTags stale empty hash = %#v, want empty page hash 0", staleEmptyTags)
}
r.deps.Messages = appmessages.NewService(memory.NewMessageStore(), nil)
if _, err := f.users.SetPremiumUntil(ownerCtx, owner.ID, int(time.Now().Add(time.Hour).Unix())); err != nil {
t.Fatalf("grant owner premium for saved tag rename: %v", err)
}
updateTagReq := &tg.MessagesUpdateSavedReactionTagRequest{Reaction: &tg.ReactionEmoji{Emoticon: "ok"}}
updateTagReq.SetTitle("Work")
if ok, err := r.onMessagesUpdateSavedReactionTag(ownerCtx, updateTagReq); err != nil || !ok {
t.Fatalf("messages.updateSavedReactionTag = ok %v err %v, want true nil", ok, err)
if _, err := r.onMessagesUpdateSavedReactionTag(ownerCtx, updateTagReq); err == nil || !strings.Contains(err.Error(), "REACTION_INVALID") {
t.Fatalf("messages.updateSavedReactionTag unassigned err = %v, want REACTION_INVALID", err)
}
globalTags, err := r.onMessagesGetSavedReactionTags(ownerCtx, &tg.MessagesGetSavedReactionTagsRequest{})
if err != nil {
t.Fatalf("messages.getSavedReactionTags global: %v", err)
}
globalPage, ok := globalTags.(*tg.MessagesSavedReactionTags)
if !ok || globalPage.Hash == 0 || len(globalPage.Tags) != 1 {
t.Fatalf("messages.getSavedReactionTags global = %#v, want one hashable tag", globalTags)
}
if emoji, ok := globalPage.Tags[0].Reaction.(*tg.ReactionEmoji); !ok || emoji.Emoticon != "ok" || globalPage.Tags[0].Title != "Work" || globalPage.Tags[0].Count != 0 {
t.Fatalf("messages.getSavedReactionTags tag = %+v, want ok/Work/count0", globalPage.Tags[0])
}
globalNotModified, err := r.onMessagesGetSavedReactionTags(ownerCtx, &tg.MessagesGetSavedReactionTagsRequest{Hash: globalPage.Hash})
if err != nil {
t.Fatalf("messages.getSavedReactionTags hash: %v", err)
}
if _, ok := globalNotModified.(*tg.MessagesSavedReactionTagsNotModified); !ok {
t.Fatalf("messages.getSavedReactionTags hash = %#v, want notModified", globalNotModified)
if !ok || globalPage.Hash != 0 || len(globalPage.Tags) != 0 {
t.Fatalf("messages.getSavedReactionTags global = %#v, want empty", globalTags)
}
peerTagsAfterUpdate, err := r.onMessagesGetSavedReactionTags(ownerCtx, savedTagsReq)
if err != nil {
@ -909,9 +904,11 @@ func TestTDesktopPassiveChannelStubs(t *testing.T) {
if err != nil {
t.Fatalf("messages.getDefaultTagReactions: %v", err)
}
if got := tagReactions.(*tg.MessagesReactions).Reactions; len(got) != 0 {
t.Fatalf("messages.getDefaultTagReactions = %+v, want empty", got)
defaultPage, ok := tagReactions.(*tg.MessagesReactions)
if !ok || defaultPage.Hash == 0 || len(defaultPage.Reactions) == 0 {
t.Fatalf("messages.getDefaultTagReactions = %#v, want non-empty hashable catalog", tagReactions)
}
r.deps.Messages = nil
// poll 链路已是真实现：对非 poll 消息一律 MESSAGE_ID_INVALID（与官方一致）。
if _, err := r.onMessagesSendVote(ownerCtx, &tg.MessagesSendVoteRequest{
Peer: inputPeerChannel(channel),

3
internal/rpc/convert_messages.go
View file

@ -444,6 +444,9 @@ func tgMessageReactions(viewerUserID int64, in *domain.ChannelMessageReactions)
if in.CanSeeList {
out.SetCanSeeList(true)
}
if in.AsTags {
out.SetReactionsAsTags(true)
}
for _, item := range in.Results {
reaction := tgMessageReaction(item.Reaction)
if reaction == nil || item.Count <= 0 {

28
internal/rpc/convert_updates.go
View file

@ -35,7 +35,7 @@ func tgUpdatesDifference(viewerUserID int64, diff domain.UpdateDifference) tg.Up
if update := tgReadHistoryOutboxUpdate(event); update != nil {
out.OtherUpdates = append(out.OtherUpdates, update)
}
case domain.UpdateEventMessageReactions, domain.UpdateEventMessagePoll:
case domain.UpdateEventMessagePoll:
// 同时下发消息快照（含最新聚合）与对应通知 update；事件无 TL pts，
// pts 推进靠 difference state 本身。
if msg := tgMessage(event.Message); msg != nil {
@ -464,32 +464,6 @@ func tgOtherUpdateFromEvent(event domain.UpdateEvent) tg.UpdateClass {
return nil
}
return tgUpdateMessagePoll(pollPeer, event.Message.ID, media.Poll)
case domain.UpdateEventMessageReactions:
if event.Message.ID <= 0 || event.Message.ID > domain.MaxMessageBoxID {
return nil
}
peer := event.Message.Peer
if peer.Type == "" || peer.ID == 0 {
peer = event.Peer
}
outPeer := tgPeer(peer)
if outPeer == nil {
return nil
}
reactions := event.Message.Reactions
if reactions == nil {
empty := domain.ChannelMessageReactions{CanSeeList: true, Results: []domain.ChannelMessageReactionCount{}, Recent: []domain.ChannelMessagePeerReaction{}}
reactions = &empty
}
converted := tgMessageReactions(event.UserID, reactions)
if converted == nil {
converted = &tg.MessageReactions{Results: []tg.ReactionCount{}}
}
return &tg.UpdateMessageReactions{
Peer: outPeer,
MsgID: event.Message.ID,
Reactions: *converted,
}
case domain.UpdateEventDialogFilter:
update := &tg.UpdateDialogFilter{ID: event.FilterID}
if event.DialogFilter != nil {

4
internal/rpc/deps.go
View file

@ -564,6 +564,8 @@ type MessagesService interface {
GetOutboxReadDate(ctx context.Context, userID int64, req domain.OutboxReadDateRequest) (int, error)
SetMessageReactions(ctx context.Context, userID int64, req domain.SetPrivateMessageReactionsRequest) (domain.PrivateMessageReactionsResult, error)
GetMessageReactions(ctx context.Context, userID int64, req domain.PrivateMessageReactionsRequest) (domain.PrivateMessageReactionsResult, error)
SavedReactionTags(ctx context.Context, userID int64, savedPeer domain.Peer, limit int) ([]domain.SavedReactionTag, error)
UpdateSavedReactionTag(ctx context.Context, userID int64, tag domain.SavedReactionTag) error
VoteMessagePoll(ctx context.Context, userID int64, req domain.VotePrivateMessagePollRequest) (domain.PrivateMessagePollResult, error)
CloseMessagePoll(ctx context.Context, userID int64, req domain.ClosePrivateMessagePollRequest) (domain.PrivateMessagePollResult, error)
ListUnreadReactionMessages(ctx context.Context, userID int64, peer domain.Peer, limit int) ([]domain.Message, error)
@ -691,8 +693,6 @@ type ChannelsService interface {
TopReactions(ctx context.Context, userID int64, limit int) ([]domain.MessageReaction, error)
RecentReactions(ctx context.Context, userID int64, limit int) ([]domain.MessageReaction, error)
ClearRecentReactions(ctx context.Context, userID int64) error
SavedReactionTags(ctx context.Context, userID int64, limit int) ([]domain.SavedReactionTag, error)
UpdateSavedReactionTag(ctx context.Context, userID int64, tag domain.SavedReactionTag) error
GetPremiumBoostStatus(ctx context.Context, userID, channelID int64, now int) (domain.PremiumBoostStatus, error)
ListPremiumBoosts(ctx context.Context, userID, channelID int64, gifts bool, offset string, limit, now int) (domain.PremiumBoostList, error)
GetPremiumMyBoosts(ctx context.Context, userID int64, now, premiumUntil int) (domain.PremiumMyBoosts, error)

4
internal/rpc/messages_contracts.go
View file

@ -19,10 +19,6 @@ type accountPaidReactionPrivacyService interface {
SetPaidReactionPrivacy(ctx context.Context, userID int64, privacy domain.PaidReactionPrivacy) (domain.AccountReactionSettings, error)
}
type messageReactionUpdateRecorder interface {
RecordMessageReactions(ctx context.Context, authKeyID [8]byte, userID int64, msg domain.Message) (domain.UpdateEvent, domain.UpdateState, error)
}
type messagePollUpdateRecorder interface {
RecordMessagePoll(ctx context.Context, authKeyID [8]byte, userID int64, msg domain.Message) (domain.UpdateEvent, domain.UpdateState, error)
}

41
internal/rpc/messages_history.go
View file

@ -766,7 +766,7 @@ func (r *Router) messageFilterFromHistoryRequest(userID int64, req *tg.MessagesG
}, true
}
func (r *Router) messageFilterFromSearchRequest(userID int64, req *tg.MessagesSearchRequest) domain.MessageFilter {
func (r *Router) messageFilterFromSearchRequest(ctx context.Context, userID int64, req *tg.MessagesSearchRequest) (domain.MessageFilter, error) {
limit := req.Limit
if limit > 500 {
limit = 500
@ -774,6 +774,8 @@ func (r *Router) messageFilterFromSearchRequest(userID int64, req *tg.MessagesSe
filter := domain.MessageFilter{
Query: req.Q,
OffsetID: req.OffsetID,
MinDate: req.MinDate,
MaxDate: req.MaxDate,
AddOffset: domain.ClampMessageHistoryAddOffset(req.AddOffset),
Limit: limit,
MaxID: req.MaxID,
@ -786,7 +788,42 @@ func (r *Router) messageFilterFromSearchRequest(userID int64, req *tg.MessagesSe
filter.HasPeer = true
filter.Peer = peer
}
return filter
savedReactions, hasSavedReactions := req.GetSavedReaction()
savedPeerInput, hasSavedPeer := req.GetSavedPeerID()
if hasSavedReactions || hasSavedPeer {
if !filter.HasPeer ||
filter.Peer != (domain.Peer{Type: domain.PeerTypeUser, ID: userID}) {
return domain.MessageFilter{}, peerIDInvalidErr()
}
}
if hasSavedPeer {
if savedPeerInput == nil {
return domain.MessageFilter{}, peerIDInvalidErr()
}
savedPeer, err := r.checkedDomainPeerFromInputPeer(ctx, userID, savedPeerInput)
if err != nil {
return domain.MessageFilter{}, err
}
filter.SavedPeer = savedPeer
}
if hasSavedReactions {
if len(savedReactions) == 0 || len(savedReactions) > maxReactionVector {
return domain.MessageFilter{}, reactionInvalidErr()
}
seen := make(map[string]struct{}, len(savedReactions))
for _, item := range savedReactions {
reaction, err := domainMessageReactionFromTL(item)
if err != nil {
return domain.MessageFilter{}, err
}
if _, ok := seen[reaction.Key()]; ok {
continue
}
seen[reaction.Key()] = struct{}{}
filter.SavedReactions = append(filter.SavedReactions, reaction)
}
}
return filter, nil
}
func (r *Router) channelHistoryFilterFromSearchRequest(userID int64, req *tg.MessagesSearchRequest, channelID int64) (domain.ChannelHistoryFilter, bool) {

106
internal/rpc/messages_reactions_catalog.go
View file

@ -2,10 +2,13 @@ package rpc
import (
"context"
"github.com/iamxvbaba/td/tg"
"hash/fnv"
"strconv"
"crypto/md5"
"encoding/binary"
"sort"
"strings"
"github.com/iamxvbaba/td/tg"
"telesrv/internal/compat/tdesktop"
"telesrv/internal/domain"
)
@ -69,33 +72,34 @@ func (r *Router) onMessagesGetSavedReactionTags(ctx context.Context, req *tg.Mes
if err != nil {
return nil, internalErr()
}
var savedPeer domain.Peer
if peer, ok := req.GetPeer(); ok && peer != nil {
if _, err := r.checkedDomainPeerFromInputPeer(ctx, userID, peer); err != nil {
savedPeer, err = r.checkedDomainPeerFromInputPeer(ctx, userID, peer)
if err != nil {
return nil, err
}
}
if r.deps.Messages == nil {
return savedReactionTagsEmpty(req.Hash), nil
}
if r.deps.Channels == nil {
return savedReactionTagsEmpty(req.Hash), nil
}
tags, err := r.deps.Channels.SavedReactionTags(ctx, userID, domain.MaxSavedReactionTags)
tags, err := r.deps.Messages.SavedReactionTags(ctx, userID, savedPeer, domain.MaxSavedReactionTags)
if err != nil {
return nil, channelInvalidErr(err)
return nil, messageReactionErr(err)
}
return savedReactionTagsFromDomain(tags, req.Hash), nil
return savedReactionTagsFromDomain(tags, req.Hash, savedPeer.ID == 0), nil
}
func (r *Router) onMessagesGetDefaultTagReactions(ctx context.Context, hash int64) (tg.MessagesReactionsClass, error) {
if _, _, err := r.currentUserID(ctx); err != nil {
return nil, internalErr()
}
return messagesReactionsEmpty(hash), nil
return messagesReactionsFromDomain(
r.reactionsWithCatalogFallback(ctx, nil, domain.MaxChannelMessageReactionsPerUser),
hash,
), nil
}
func messagesReactionsEmpty(hash int64) tg.MessagesReactionsClass {
if hash != 0 {
return &tg.MessagesReactionsNotModified{}
}
func messagesReactionsEmpty(_ int64) tg.MessagesReactionsClass {
return &tg.MessagesReactions{
Hash: 0,
Reactions: []tg.ReactionClass{},
@ -127,8 +131,15 @@ func savedReactionTagsEmpty(_ int64) tg.MessagesSavedReactionTagsClass {
}
}
func savedReactionTagsFromDomain(tags []domain.SavedReactionTag, requestHash int64) tg.MessagesSavedReactionTagsClass {
hash := savedReactionTagListHash(tags)
func savedReactionTagsFromDomain(tags []domain.SavedReactionTag, requestHash int64, includeTitles bool) tg.MessagesSavedReactionTagsClass {
tags = append([]domain.SavedReactionTag(nil), tags...)
sort.SliceStable(tags, func(i, j int) bool {
if tags[i].Count != tags[j].Count {
return tags[i].Count > tags[j].Count
}
return savedReactionTagLongID(tags[i].Reaction) > savedReactionTagLongID(tags[j].Reaction)
})
hash := savedReactionTagListHash(tags, includeTitles)
if hash != 0 && requestHash == hash {
return &tg.MessagesSavedReactionTagsNotModified{}
}
@ -142,7 +153,7 @@ func savedReactionTagsFromDomain(tags []domain.SavedReactionTag, requestHash int
Reaction: reaction,
Count: tag.Count,
}
if tag.Title != "" {
if includeTitles && tag.Title != "" {
item.SetTitle(tag.Title)
}
out = append(out, item)
@ -239,38 +250,47 @@ func messageReactionListHash(reactions []domain.MessageReaction) int64 {
if len(reactions) == 0 {
return 0
}
h := fnv.New64a()
var hash uint64
for _, reaction := range reactions {
_, _ = h.Write([]byte(reaction.Type))
_, _ = h.Write([]byte{0})
_, _ = h.Write([]byte(reaction.Value()))
_, _ = h.Write([]byte{0xff})
hash = telegramListHashNext(hash, savedReactionTagLongID(reaction))
}
sum := int64(h.Sum64() & 0x7fffffffffffffff)
if sum == 0 {
return 1
}
return sum
return int64(hash)
}
func savedReactionTagListHash(tags []domain.SavedReactionTag) int64 {
func savedReactionTagListHash(tags []domain.SavedReactionTag, includeTitles bool) int64 {
if len(tags) == 0 {
return 0
}
h := fnv.New64a()
var hash uint64
for _, tag := range tags {
_, _ = h.Write([]byte(tag.Reaction.Type))
_, _ = h.Write([]byte{0})
_, _ = h.Write([]byte(tag.Reaction.Value()))
_, _ = h.Write([]byte{0})
_, _ = h.Write([]byte(tag.Title))
_, _ = h.Write([]byte{0})
_, _ = h.Write([]byte(strconv.Itoa(tag.Count)))
_, _ = h.Write([]byte{0xff})
hash = telegramListHashNext(hash, savedReactionTagLongID(tag.Reaction))
if includeTitles && tag.Title != "" {
hash = telegramListHashNext(hash, md5LongID(tag.Title))
}
hash = telegramListHashNext(hash, uint64(tag.Count))
}
sum := int64(h.Sum64() & 0x7fffffffffffffff)
if sum == 0 {
return 1
}
return sum
return int64(hash)
}
func savedReactionTagLongID(reaction domain.MessageReaction) uint64 {
switch reaction.Type {
case domain.MessageReactionEmoji:
return md5LongID(strings.ReplaceAll(reaction.Emoticon, "\ufe0f", ""))
case domain.MessageReactionCustomEmoji:
return uint64(reaction.DocumentID)
default:
return 0
}
}
func md5LongID(value string) uint64 {
sum := md5.Sum([]byte(value))
return binary.BigEndian.Uint64(sum[:8])
}
func telegramListHashNext(hash, id uint64) uint64 {
hash ^= hash >> 21
hash ^= hash << 35
hash ^= hash >> 4
return hash + id
}

18
internal/rpc/messages_reactions_helpers.go
View file

@ -3,10 +3,12 @@ package rpc
import (
"context"
"errors"
"github.com/iamxvbaba/td/tg"
"strings"
"telesrv/internal/domain"
"unicode/utf8"
"github.com/iamxvbaba/td/tg"
"telesrv/internal/domain"
)
func (r *Router) onMessagesUpdateSavedReactionTag(ctx context.Context, req *tg.MessagesUpdateSavedReactionTagRequest) (bool, error) {
@ -18,8 +20,8 @@ func (r *Router) onMessagesUpdateSavedReactionTag(ctx context.Context, req *tg.M
if err != nil {
return false, err
}
if reaction.Type != domain.MessageReactionEmoji {
return false, reactionInvalidErr()
if !r.viewerPremium(ctx, userID) {
return false, premiumAccountRequiredErr()
}
title, ok := req.GetTitle()
if !ok {
@ -28,13 +30,13 @@ func (r *Router) onMessagesUpdateSavedReactionTag(ctx context.Context, req *tg.M
if utf8.RuneCountInString(title) > maxSavedReactionTagTitle {
return false, limitInvalidErr()
}
if r.deps.Channels != nil {
if err := r.deps.Channels.UpdateSavedReactionTag(ctx, userID, domain.SavedReactionTag{
if r.deps.Messages != nil {
if err := r.deps.Messages.UpdateSavedReactionTag(ctx, userID, domain.SavedReactionTag{
UserID: userID,
Reaction: reaction,
Title: title,
}); err != nil {
return false, channelInvalidErr(err)
return false, messageReactionErr(err)
}
}
r.pushUserUpdates(ctx, userID, &tg.Updates{
@ -166,6 +168,8 @@ func messageReactionErr(err error) error {
switch {
case errors.Is(err, domain.ErrMessageIDInvalid):
return messageIDInvalidErr()
case errors.Is(err, domain.ErrReactionInvalid):
return reactionInvalidErr()
default:
return internalErr()
}

55
internal/rpc/messages_reactions_send.go
View file

@ -29,6 +29,10 @@ func (r *Router) onMessagesSendReaction(ctx context.Context, req *tg.MessagesSen
// reactions_user_max_premium=3），否则客户端允许的多 reaction 会被静默裁剪。
perUserMax := domain.MessageReactionsUserMax(r.viewerPremium(ctx, userID))
reactions = domain.TrimMessageReactionsToUserMax(reactions, perUserMax)
if peer.Type == domain.PeerTypeUser && peer.ID == userID &&
len(reactions) > 0 && !r.viewerPremium(ctx, userID) {
return nil, premiumAccountRequiredErr()
}
date := int(r.clock.Now().Unix())
if peer.Type == domain.PeerTypeChannel && r.deps.Channels != nil {
res, err := r.deps.Channels.SetMessageReactions(ctx, userID, domain.SetChannelMessageReactionsRequest{
@ -54,7 +58,7 @@ func (r *Router) onMessagesSendReaction(ctx context.Context, req *tg.MessagesSen
return updates, nil
}
if peer.Type == domain.PeerTypeUser && r.deps.Messages != nil {
if len(reactions) == 0 && r.shouldSuppressTransientPrivateReactionClear(userID, peer, req.MsgID, date) {
if peer.ID != userID && len(reactions) == 0 && r.shouldSuppressTransientPrivateReactionClear(userID, peer, req.MsgID, date) {
res, err := r.deps.Messages.GetMessageReactions(ctx, userID, domain.PrivateMessageReactionsRequest{
OwnerUserID: userID,
Peer: peer,
@ -65,7 +69,7 @@ func (r *Router) onMessagesSendReaction(ctx context.Context, req *tg.MessagesSen
}
return r.privateMessagesReactionsUpdates(ctx, userID, peer, res, []int{req.MsgID}), nil
}
if req.Big && len(reactions) > 0 {
if peer.ID != userID && req.Big && len(reactions) > 0 {
r.rememberTransientPrivateBigReaction(userID, peer, req.MsgID, date)
}
res, err := r.deps.Messages.SetMessageReactions(ctx, userID, domain.SetPrivateMessageReactionsRequest{
@ -84,19 +88,13 @@ func (r *Router) onMessagesSendReaction(ctx context.Context, req *tg.MessagesSen
if len(reactions) == 0 {
r.forgetTransientPrivateBigReaction(userID, peer, req.MsgID)
}
if err := r.recordMessageReactionUse(ctx, userID, reactions, req.GetAddToRecent(), date); err != nil {
return nil, internalErr()
// Saved Messages reactions are private tags, not ordinary reaction usage.
if peer.ID != userID {
if err := r.recordMessageReactionUse(ctx, userID, reactions, req.GetAddToRecent(), date); err != nil {
return nil, internalErr()
}
}
recordedEvents, err := r.recordPrivateMessageReactionEvents(ctx, userID, res)
if err != nil {
return nil, internalErr()
}
// reaction 事件占双方账号 pts 但 updateMessageReactions 不带 pts；
// 在线直推必须附 pts 簿记，否则双方下一条带 pts 的更新被判空洞。
updates := r.privateMessageReactionsUpdates(ctx, userID, peer, res)
if updates != nil {
updates.Updates = appendAuxPtsBookkeeping(updates.Updates, recordedEvents[userID])
}
r.pushUserUpdates(ctx, userID, updates)
for _, msg := range res.Messages {
if msg.OwnerUserID == 0 || msg.OwnerUserID == userID {
@ -104,9 +102,6 @@ func (r *Router) onMessagesSendReaction(ctx context.Context, req *tg.MessagesSen
}
viewerPeer := msg.Peer
viewerUpdates := r.privateMessageReactionsUpdates(ctx, msg.OwnerUserID, viewerPeer, res)
if viewerUpdates != nil {
viewerUpdates.Updates = appendAuxPtsBookkeeping(viewerUpdates.Updates, recordedEvents[msg.OwnerUserID])
}
r.pushUserUpdates(ctx, msg.OwnerUserID, viewerUpdates)
}
return updates, nil
@ -223,33 +218,6 @@ func (r *Router) recordMessageReactionUse(ctx context.Context, userID int64, rea
return recorder.RecordMessageReactionUse(ctx, userID, reactions, addToRecent, date)
}
func (r *Router) recordPrivateMessageReactionEvents(ctx context.Context, requestUserID int64, res domain.PrivateMessageReactionsResult) (map[int64]domain.UpdateEvent, error) {
if r.deps.Updates == nil {
return nil, nil
}
recorder, ok := r.deps.Updates.(messageReactionUpdateRecorder)
if !ok {
return nil, nil
}
authKeyID, _ := AuthKeyIDFrom(ctx)
events := make(map[int64]domain.UpdateEvent, len(res.Messages))
for _, msg := range res.Messages {
if msg.OwnerUserID == 0 || msg.ID == 0 {
continue
}
eventAuthKeyID := [8]byte{}
if msg.OwnerUserID == requestUserID {
eventAuthKeyID = authKeyID
}
event, _, err := recorder.RecordMessageReactions(ctx, eventAuthKeyID, msg.OwnerUserID, msg)
if err != nil {
return nil, err
}
events[msg.OwnerUserID] = event
}
return events, nil
}
func (r *Router) channelMessageReactionsUpdates(ctx context.Context, viewerUserID int64, res domain.ChannelMessageReactionsResult) *tg.Updates {
ids := []int{res.Message.ID}
if res.Message.ID <= 0 && len(res.Messages) > 0 {
@ -311,6 +279,7 @@ func minifyChannelReactionsResult(res domain.ChannelMessageReactionsResult) doma
}
out := domain.ChannelMessageReactions{
CanSeeList: in.CanSeeList,
AsTags: in.AsTags,
Results: make([]domain.ChannelMessageReactionCount, 0, len(in.Results)),
Recent: make([]domain.ChannelMessagePeerReaction, 0, len(in.Recent)),
}

267
internal/rpc/messages_reactions_test.go
View file

@ -5,82 +5,25 @@ import (
"github.com/iamxvbaba/td/clock"
"github.com/iamxvbaba/td/proto"
"github.com/iamxvbaba/td/tg"
"github.com/iamxvbaba/td/tgerr"
"go.uber.org/zap/zaptest"
appchannels "telesrv/internal/app/channels"
appusers "telesrv/internal/app/users"
"telesrv/internal/domain"
"telesrv/internal/store/memory"
"testing"
"time"
)
func TestUpdatesDifferenceIncludesReactionMessageAndUpdate(t *testing.T) {
const (
aliceID = int64(1000000001)
bobID = int64(1000000002)
)
reaction := domain.MessageReaction{Type: domain.MessageReactionEmoji, Emoticon: "\U0001f44d"}
reactions := domain.ChannelMessageReactions{
CanSeeList: true,
Results: []domain.ChannelMessageReactionCount{{
Reaction: reaction,
Count: 1,
ChosenOrder: 1,
}},
Recent: []domain.ChannelMessagePeerReaction{{
UserID: bobID,
Reaction: reaction,
My: true,
ChosenOrder: 1,
Date: 1700000310,
}},
}
msg := domain.Message{
ID: 68,
UID: 7001,
OwnerUserID: aliceID,
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: bobID},
From: domain.Peer{Type: domain.PeerTypeUser, ID: aliceID},
Date: 1700000300,
Body: "rx",
Reactions: &reactions,
}
got, ok := tgUpdatesDifference(0, domain.UpdateDifference{
State: domain.UpdateState{Pts: 9, Date: 1700000310},
Events: []domain.UpdateEvent{{
UserID: aliceID,
Type: domain.UpdateEventMessageReactions,
Pts: 9,
PtsCount: 1,
Date: 1700000310,
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: bobID},
Message: msg,
}},
}).(*tg.UpdatesDifference)
if !ok {
t.Fatalf("difference = %T, want *tg.UpdatesDifference", got)
}
if len(got.NewMessages) != 1 || len(got.OtherUpdates) != 1 {
t.Fatalf("difference messages/updates = %d/%d, want 1/1", len(got.NewMessages), len(got.OtherUpdates))
}
wireMsg, ok := got.NewMessages[0].(*tg.Message)
if !ok || wireMsg.ID != msg.ID {
t.Fatalf("message = %T %+v, want message %d", got.NewMessages[0], got.NewMessages[0], msg.ID)
}
msgReactions, ok := wireMsg.GetReactions()
if !ok || len(msgReactions.Results) != 1 || msgReactions.Results[0].Count != 1 || msgReactions.Results[0].ChosenOrder != 1 {
t.Fatalf("message reactions = %+v set=%v, want chosen reaction", msgReactions, ok)
}
update, ok := got.OtherUpdates[0].(*tg.UpdateMessageReactions)
if !ok || update.MsgID != msg.ID || len(update.Reactions.Results) != 1 || update.Reactions.Results[0].ChosenOrder != 1 {
t.Fatalf("reaction update = %T %+v, want update for msg %d", got.OtherUpdates[0], got.OtherUpdates[0], msg.ID)
}
}
func TestMessagesUpdateSavedReactionTagPersistsAndPushesRefresh(t *testing.T) {
const userID = int64(1000000001)
userID, users := newReactionTestUsers(t, true)
sessions := &captureSessions{}
reaction := domain.MessageReaction{Type: domain.MessageReactionEmoji, Emoticon: "\U0001f44d"}
messages := &captureMessages{savedTags: []domain.SavedReactionTag{{
UserID: userID, Reaction: reaction, Count: 1,
}}}
r := New(Config{}, Deps{
Channels: appchannels.NewService(memory.NewChannelStore()),
Messages: messages,
Users: users,
Sessions: sessions,
}, zaptest.NewLogger(t), clock.System)
@ -121,6 +64,198 @@ func TestMessagesUpdateSavedReactionTagPersistsAndPushesRefresh(t *testing.T) {
}
}
func TestMessagesUpdateSavedReactionTagAcceptsCustomEmoji(t *testing.T) {
userID, users := newReactionTestUsers(t, true)
custom := domain.MessageReaction{Type: domain.MessageReactionCustomEmoji, DocumentID: 90001}
messages := &captureMessages{savedTags: []domain.SavedReactionTag{{
UserID: userID, Reaction: custom, Count: 1,
}}}
r := New(Config{}, Deps{Messages: messages, Users: users}, zaptest.NewLogger(t), clock.System)
req := &tg.MessagesUpdateSavedReactionTagRequest{
Reaction: &tg.ReactionCustomEmoji{DocumentID: custom.DocumentID},
}
req.SetTitle("Work")
ok, err := r.onMessagesUpdateSavedReactionTag(WithUserID(context.Background(), userID), req)
if err != nil || !ok {
t.Fatalf("rename custom saved tag = %v, %v", ok, err)
}
if messages.updatedSavedTag.Reaction.Key() != custom.Key() || messages.updatedSavedTag.Title != "Work" {
t.Fatalf("updated custom tag = %+v", messages.updatedSavedTag)
}
}
func newReactionTestUsers(t *testing.T, premium bool) (int64, UsersService) {
t.Helper()
users := memory.NewUserStore()
user, err := users.Create(context.Background(), domain.User{
Phone: "+15550000001",
FirstName: "Reaction",
})
if err != nil {
t.Fatalf("create reaction test user: %v", err)
}
if premium {
if _, err := users.SetPremiumUntil(context.Background(), user.ID, int(time.Now().Add(time.Hour).Unix())); err != nil {
t.Fatalf("set reaction test premium: %v", err)
}
}
return user.ID, appusers.NewService(users)
}
func TestMessagesSendReactionSavedMessageUsesTagsWithoutPTSBookkeeping(t *testing.T) {
userID, users := newReactionTestUsers(t, true)
messages := &captureMessages{}
sessions := &captureSessions{}
r := New(Config{}, Deps{
Messages: messages,
Users: users,
Sessions: sessions,
}, zaptest.NewLogger(t), fixedClock{now: time.Unix(1_700_000_200, 0)})
req := &tg.MessagesSendReactionRequest{
Peer: &tg.InputPeerSelf{},
MsgID: 7,
Reaction: []tg.ReactionClass{&tg.ReactionCustomEmoji{DocumentID: 90001}},
}
req.SetReaction(req.Reaction)
got, err := r.onMessagesSendReaction(
WithSessionID(WithUserID(context.Background(), userID), 72),
req,
)
if err != nil {
t.Fatalf("send saved tag: %v", err)
}
updates, ok := got.(*tg.Updates)
if !ok || len(updates.Updates) != 1 {
t.Fatalf("saved tag result = %T %+v, want one update", got, got)
}
update, ok := updates.Updates[0].(*tg.UpdateMessageReactions)
if !ok || !update.Reactions.ReactionsAsTags || len(update.Reactions.Results) != 1 {
t.Fatalf("saved tag update = %T %+v, want reactions_as_tags", updates.Updates[0], updates.Updates[0])
}
for _, item := range updates.Updates {
if deleted, ok := item.(*tg.UpdateDeleteMessages); ok {
t.Fatalf("saved tag emitted fake delete pts bookkeeping: %+v", deleted)
}
}
if messages.setReactionReq.Peer != (domain.Peer{Type: domain.PeerTypeUser, ID: userID}) {
t.Fatalf("saved tag peer = %+v, want self", messages.setReactionReq.Peer)
}
push := sessions.snapshot()
pushed, ok := push.message.(*tg.Updates)
if push.userID != userID || push.sessionID != 72 || !ok || len(pushed.Updates) != 1 {
t.Fatalf("saved tag push = user %d exclude %d %T %+v", push.userID, push.sessionID, push.message, push.message)
}
pushedReaction, ok := pushed.Updates[0].(*tg.UpdateMessageReactions)
if !ok || !pushedReaction.Reactions.ReactionsAsTags {
t.Fatalf("saved tag pushed update = %T %+v, want reactions_as_tags", pushed.Updates[0], pushed.Updates[0])
}
}
func TestMessagesSendReactionSavedMessageRequiresPremiumButAllowsClear(t *testing.T) {
userID, users := newReactionTestUsers(t, false)
messages := &captureMessages{}
r := New(Config{}, Deps{
Messages: messages,
Users: users,
}, zaptest.NewLogger(t), clock.System)
add := &tg.MessagesSendReactionRequest{
Peer: &tg.InputPeerSelf{},
MsgID: 8,
Reaction: []tg.ReactionClass{&tg.ReactionEmoji{Emoticon: "👍"}},
}
add.SetReaction(add.Reaction)
if _, err := r.onMessagesSendReaction(WithUserID(context.Background(), userID), add); !tgerr.Is(err, "PREMIUM_ACCOUNT_REQUIRED") {
t.Fatalf("non-premium add err = %v, want PREMIUM_ACCOUNT_REQUIRED", err)
}
clear := &tg.MessagesSendReactionRequest{Peer: &tg.InputPeerSelf{}, MsgID: 8}
if _, err := r.onMessagesSendReaction(WithUserID(context.Background(), userID), clear); err != nil {
t.Fatalf("non-premium clear: %v", err)
}
}
func TestSavedReactionTagHashMatchesClientShape(t *testing.T) {
plain := domain.MessageReaction{Type: domain.MessageReactionEmoji, Emoticon: "❤️"}
withoutVariation := domain.MessageReaction{Type: domain.MessageReactionEmoji, Emoticon: "❤"}
if got, want := messageReactionListHash([]domain.MessageReaction{plain}), messageReactionListHash([]domain.MessageReaction{withoutVariation}); got != want {
t.Fatalf("emoji variation-selector hash = %d, want normalized %d", got, want)
}
tags := []domain.SavedReactionTag{{
Reaction: plain,
Title: "Love",
Count: 3,
}}
full := savedReactionTagsFromDomain(tags, 0, true)
page, ok := full.(*tg.MessagesSavedReactionTags)
if !ok || page.Hash == 0 || len(page.Tags) != 1 || page.Tags[0].Title != "Love" {
t.Fatalf("saved tag page = %T %+v", full, full)
}
if page.Hash != -4770309592622053821 {
t.Fatalf("saved tag client hash = %d, want -4770309592622053821", page.Hash)
}
if cached := savedReactionTagsFromDomain(tags, page.Hash, true); cached == nil {
t.Fatal("cached saved tag result is nil")
} else if _, ok := cached.(*tg.MessagesSavedReactionTagsNotModified); !ok {
t.Fatalf("cached saved tag result = %T, want not modified", cached)
}
perPeer := savedReactionTagsFromDomain(tags, 0, false)
peerPage, ok := perPeer.(*tg.MessagesSavedReactionTags)
if !ok || peerPage.Tags[0].Title != "" || peerPage.Hash == page.Hash {
t.Fatalf("per-peer saved tags = %T %+v, want title omitted and scope hash", perPeer, perPeer)
}
}
func TestMessageFilterFromSearchRequestParsesSavedTagsAndPeer(t *testing.T) {
const userID = int64(1000000001)
r := New(Config{}, Deps{}, zaptest.NewLogger(t), clock.System)
req := &tg.MessagesSearchRequest{
Peer: &tg.InputPeerSelf{},
Q: "needle",
MinDate: 100,
MaxDate: 200,
Limit: 50,
Filter: &tg.InputMessagesFilterEmpty{},
}
req.SetSavedPeerID(&tg.InputPeerSelf{})
req.SetSavedReaction([]tg.ReactionClass{
&tg.ReactionEmoji{Emoticon: "👍"},
&tg.ReactionCustomEmoji{DocumentID: 90001},
})
filter, err := r.messageFilterFromSearchRequest(WithUserID(context.Background(), userID), userID, req)
if err != nil {
t.Fatalf("parse saved search filter: %v", err)
}
if filter.Peer != (domain.Peer{Type: domain.PeerTypeUser, ID: userID}) ||
filter.SavedPeer != filter.Peer || len(filter.SavedReactions) != 2 ||
filter.MinDate != 100 || filter.MaxDate != 200 {
t.Fatalf("saved search filter = %+v", filter)
}
req.Peer = &tg.InputPeerUser{UserID: userID + 1, AccessHash: 1}
if _, err := r.messageFilterFromSearchRequest(WithUserID(context.Background(), userID), userID, req); !tgerr.Is(err, "PEER_ID_INVALID") {
t.Fatalf("non-self saved search err = %v, want PEER_ID_INVALID", err)
}
}
func TestMessagesGetDefaultTagReactionsReturnsHashableCatalog(t *testing.T) {
r := New(Config{}, Deps{}, zaptest.NewLogger(t), clock.System)
ctx := WithUserID(context.Background(), 1000000001)
got, err := r.onMessagesGetDefaultTagReactions(ctx, 0)
if err != nil {
t.Fatalf("get default tag reactions: %v", err)
}
page, ok := got.(*tg.MessagesReactions)
if !ok || page.Hash == 0 || len(page.Reactions) == 0 {
t.Fatalf("default tags = %T %+v, want non-empty hashable catalog", got, got)
}
cached, err := r.onMessagesGetDefaultTagReactions(ctx, page.Hash)
if err != nil {
t.Fatalf("get cached default tags: %v", err)
}
if _, ok := cached.(*tg.MessagesReactionsNotModified); !ok {
t.Fatalf("cached default tags = %T, want not modified", cached)
}
}
func TestMessagesSendReactionPrivatePeerReturnsReactionUpdate(t *testing.T) {
const (
userID = int64(1000000001)

5
internal/rpc/messages_register.go
View file

@ -702,7 +702,10 @@ func (r *Router) registerMessages(d *tlprofile.Dispatcher) {
if err != nil {
return nil, internalErr()
}
filter := r.messageFilterFromSearchRequest(userID, req)
filter, err := r.messageFilterFromSearchRequest(ctx, userID, req)
if err != nil {
return nil, err
}
if filter.HasPeer && filter.Peer.Type == domain.PeerTypeChannel {
if r.deps.Channels == nil {
return messagesNotModifiedOrEmpty(req.Hash), nil

36
internal/rpc/rpc_testkit_messages_test.go
View file

@ -27,6 +27,10 @@ type captureMessages struct {
setReactionRes domain.PrivateMessageReactionsResult
getReactionReq domain.PrivateMessageReactionsRequest
getReactionRes domain.PrivateMessageReactionsResult
savedTagPeer domain.Peer
savedTags []domain.SavedReactionTag
updatedSavedTag domain.SavedReactionTag
savedTagErr error
getMessagesCalls int
getMessagesIDs [][]int
getMessagesListed bool
@ -432,7 +436,12 @@ func (s *captureMessages) SetMessageReactions(_ context.Context, userID int64, r
s.setReactionReq = req
if len(s.setReactionRes.Messages) == 0 {
if len(req.Reactions) == 0 {
reactions := domain.ChannelMessageReactions{CanSeeList: true, Results: []domain.ChannelMessageReactionCount{}, Recent: []domain.ChannelMessagePeerReaction{}}
reactions := domain.ChannelMessageReactions{
CanSeeList: req.Peer.ID != userID,
AsTags: req.Peer.ID == userID,
Results: []domain.ChannelMessageReactionCount{},
Recent: []domain.ChannelMessagePeerReaction{},
}
s.setReactionRes = domain.PrivateMessageReactionsResult{
Messages: []domain.Message{{
ID: req.MessageID,
@ -447,20 +456,23 @@ func (s *captureMessages) SetMessageReactions(_ context.Context, userID int64, r
return s.setReactionRes, nil
}
reactions := domain.ChannelMessageReactions{
CanSeeList: true,
CanSeeList: req.Peer.ID != userID,
AsTags: req.Peer.ID == userID,
Results: []domain.ChannelMessageReactionCount{{
Reaction: req.Reactions[0],
Count: 1,
ChosenOrder: 1,
}},
Recent: []domain.ChannelMessagePeerReaction{{
}
if req.Peer.ID != userID {
reactions.Recent = []domain.ChannelMessagePeerReaction{{
UserID: userID,
Reaction: req.Reactions[0],
My: true,
Big: req.Big,
ChosenOrder: 1,
Date: req.Date,
}},
}}
}
s.setReactionRes = domain.PrivateMessageReactionsResult{
Messages: []domain.Message{{
@ -503,6 +515,22 @@ func (s *captureMessages) GetMessageReactions(_ context.Context, userID int64, r
return s.getReactionRes, nil
}
func (s *captureMessages) SavedReactionTags(_ context.Context, _ int64, savedPeer domain.Peer, _ int) ([]domain.SavedReactionTag, error) {
s.savedTagPeer = savedPeer
return append([]domain.SavedReactionTag(nil), s.savedTags...), s.savedTagErr
}
func (s *captureMessages) UpdateSavedReactionTag(_ context.Context, _ int64, tag domain.SavedReactionTag) error {
s.updatedSavedTag = tag
for i := range s.savedTags {
if s.savedTags[i].Reaction.Key() == tag.Reaction.Key() {
s.savedTags[i].Title = tag.Title
return s.savedTagErr
}
}
return s.savedTagErr
}
func (s *captureMessages) EditMessage(_ context.Context, userID int64, req domain.EditMessageRequest) (domain.EditMessageResult, error) {
s.editReq = req
if s.editRes.OwnerUserID == 0 {

33
internal/rpc/update_peer_refs.go
View file

@ -45,9 +45,6 @@ func (r *Router) enrichUpdateEventsWithPeerCache(ctx context.Context, viewerUser
}
}
}
if out[i].Type == domain.UpdateEventMessageReactions {
out[i] = r.enrichMessageReactionEvent(ctx, viewerUserID, out[i])
}
if out[i].Type == domain.UpdateEventMessagePoll {
out[i] = r.enrichMessagePollEvent(ctx, viewerUserID, out[i])
}
@ -115,36 +112,6 @@ type updateEventPeerRefs struct {
channelIDs map[int64]struct{}
}
func (r *Router) enrichMessageReactionEvent(ctx context.Context, viewerUserID int64, event domain.UpdateEvent) domain.UpdateEvent {
if r.deps.Messages == nil || event.Message.ID <= 0 {
return event
}
peer := event.Message.Peer
if peer.Type == "" || peer.ID == 0 {
peer = event.Peer
}
if peer.Type != domain.PeerTypeUser || peer.ID == 0 {
return event
}
res, err := r.deps.Messages.GetMessageReactions(ctx, viewerUserID, domain.PrivateMessageReactionsRequest{
OwnerUserID: viewerUserID,
Peer: peer,
IDs: []int{event.Message.ID},
})
if err != nil {
return event
}
for _, msg := range res.Messages {
if msg.OwnerUserID == viewerUserID && msg.ID == event.Message.ID {
msg.Pts = event.Pts
event.Message = msg
event.Peer = msg.Peer
return event
}
}
return event
}
// enrichMessagePollEvent 在 difference 重放时按 viewer 重载消息（media 含最新 poll 权威态与
// viewer 门控），与 reaction 事件 enrich 同构。
func (r *Router) enrichMessagePollEvent(ctx context.Context, viewerUserID int64, event domain.UpdateEvent) domain.UpdateEvent {