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

feat: sync monoforum and collectible emoji status

Browse source 
Sync telesrv bd15657 (feat(account): implement collectible emoji status).

Skipped telesrv docs changes per public sync rules.
This commit is contained in:
A 2026-07-19 20:37:10 +08:00
parent edb7057757
commit 0c99ae0a9d
91 changed files with 4061 additions and 693 deletions
Download patch file Download diff file Expand all files Collapse all files

153
internal/rpc/account.go
View file

@ -119,11 +119,7 @@ func (r *Router) registerAccount(d *tlprofile.Dispatcher) {
Hash)
})
registerRPC[*tg.AccountGetCollectibleEmojiStatusesRequest](d, tlprofile.SemanticMethodAccountGetCollectibleEmojiStatuses, func(ctx context.Context, layerRequest *tg.AccountGetCollectibleEmojiStatusesRequest) (any, error) {
hash := layerRequest.
Hash
_ = hash
return tdesktop.CollectibleEmojiStatuses(), nil
return r.onAccountGetCollectibleEmojiStatuses(ctx, layerRequest.Hash)
})
registerRPC[*tg.AccountGetDefaultGroupPhotoEmojisRequest](d, tlprofile.SemanticMethodAccountGetDefaultGroupPhotoEmojis, func(ctx context.Context, layerRequest *tg.AccountGetDefaultGroupPhotoEmojisRequest) (any, error) {
hash := layerRequest.
@ -1611,10 +1607,10 @@ func (r *Router) onAccountUpdatePersonalChannel(ctx context.Context, channel tg.
return true, nil
}
// onAccountUpdateEmojiStatus 持久化用户自定义 emoji status（premium 专属）。
// emojiStatusEmpty 与未支持的 collectible 类型按清除处理（collectible 依赖
// Stars 礼物模型，范围外，记兼容矩阵）；变更经 updateUserEmojiStatus 推给
// 本人全部在线 session（self user 对象同时携带最新 emoji_status 字段）。
// onAccountUpdateEmojiStatus persists either a normal custom emoji or a
// complete collectible snapshot. Collectibles must still be locally owned by
// the actor; unsupported constructors are rejected instead of being mistaken
// for a clear operation.
func (r *Router) onAccountUpdateEmojiStatus(ctx context.Context, status tg.EmojiStatusClass) (bool, error) {
userID, _, err := r.currentUserID(ctx)
if err != nil {
@ -1624,33 +1620,105 @@ func (r *Router) onAccountUpdateEmojiStatus(ctx context.Context, status tg.Emoji
if !ok {
return true, nil // 服务未接通（精简测试装配）时保持旧 stub 语义
}
var documentID int64
var until int
if s, ok := status.(*tg.EmojiStatus); ok {
documentID = s.DocumentID
if v, ok := s.GetUntil(); ok {
until = v
}
value, err := r.domainUserEmojiStatus(ctx, userID, status)
if err != nil {
return false, err
}
var (
u domain.User
event domain.UpdateEvent
durableWrite bool
)
authKeyID, _ := AuthKeyIDFrom(ctx)
sessionID, _ := SessionIDFrom(ctx)
if durable, ok := r.deps.Users.(UserEmojiStatusDurableService); ok {
u, event, durableWrite, err = durable.UpdateEmojiStatusWithEvent(
ctx, userID, value, int(r.clock.Now().Unix()), rawAuthKeyIDForOrigin(ctx), sessionID,
)
} else {
u, err = svc.UpdateEmojiStatus(ctx, userID, value)
}
u, err := svc.UpdateEmojiStatus(ctx, userID, documentID, until)
if err != nil {
if errors.Is(err, domain.ErrPremiumRequired) {
return false, tgerr400("PREMIUM_ACCOUNT_REQUIRED")
}
if errors.Is(err, domain.ErrStarGiftCollectibleInvalid) {
return false, tgerr400("COLLECTIBLE_INVALID")
}
return false, internalErr()
}
r.invalidateRPCProjectionForUser(u.ID)
r.pushUserUpdates(ctx, u.ID, &tg.Updates{
Updates: []tg.UpdateClass{&tg.UpdateUserEmojiStatus{
UserID: u.ID,
EmojiStatus: tgUserEmojiStatus(u, r.clock.Now().Unix()),
}},
Users: []tg.UserClass{r.tgSelfUser(u)},
Date: int(r.clock.Now().Unix()),
})
update := &tg.UpdateUserEmojiStatus{UserID: u.ID, EmojiStatus: tgUserEmojiStatusValue(value)}
if durableWrite {
if sessionID != 0 {
r.bookkeepAuxPtsForCurrentSession(ctx, event)
}
r.pushUserUpdatesIfNoReliableDispatch(ctx, u.ID, &tg.Updates{
Updates: []tg.UpdateClass{update}, Users: []tg.UserClass{r.tgSelfUser(u)}, Date: event.Date,
})
} else if updates, ok := r.deps.Updates.(UserEmojiStatusUpdatesService); ok {
event, _, recordErr := updates.RecordUserEmojiStatus(ctx, authKeyID, userID, value, rawAuthKeyIDForOrigin(ctx), sessionID)
if recordErr != nil {
return false, internalErr()
}
if sessionID != 0 {
r.bookkeepAuxPtsForCurrentSession(ctx, event)
}
r.pushUserUpdatesIfNoReliableDispatch(ctx, u.ID, &tg.Updates{
Updates: []tg.UpdateClass{update}, Users: []tg.UserClass{r.tgSelfUser(u)}, Date: event.Date,
})
} else {
// Lightweight test deployments without the durable extension retain the
// previous online-only behavior; production wiring implements it.
r.pushUserUpdates(ctx, u.ID, &tg.Updates{
Updates: []tg.UpdateClass{update}, Users: []tg.UserClass{r.tgSelfUser(u)}, Date: int(r.clock.Now().Unix()),
})
}
return true, nil
}
func (r *Router) domainUserEmojiStatus(ctx context.Context, userID int64, input tg.EmojiStatusClass) (domain.UserEmojiStatus, error) {
switch status := input.(type) {
case *tg.EmojiStatusEmpty:
return domain.UserEmojiStatus{}, nil
case *tg.EmojiStatus:
value := domain.UserEmojiStatus{DocumentID: status.DocumentID}
if until, ok := status.GetUntil(); ok {
value.Until = until
}
if !value.Valid() {
return domain.UserEmojiStatus{}, tgerr400("EMOJI_STATUS_INVALID")
}
return value, nil
case *tg.InputEmojiStatusCollectible:
if r.deps.Gifts == nil || status.CollectibleID <= 0 {
return domain.UserEmojiStatus{}, tgerr400("COLLECTIBLE_INVALID")
}
gift, found, err := r.deps.Gifts.UniqueByID(ctx, status.CollectibleID)
if err != nil {
return domain.UserEmojiStatus{}, internalErr()
}
owner := domain.Peer{Type: domain.PeerTypeUser, ID: userID}
if !found || gift.Owner != owner || gift.Burned || gift.OwnerAddress != "" {
return domain.UserEmojiStatus{}, tgerr400("COLLECTIBLE_INVALID")
}
collectible, valid := domain.CollectibleEmojiStatus(gift)
if !valid {
return domain.UserEmojiStatus{}, tgerr400("COLLECTIBLE_INVALID")
}
value := domain.UserEmojiStatus{DocumentID: collectible.DocumentID, Collectible: collectible}
if until, ok := status.GetUntil(); ok {
value.Until = until
}
if !value.Valid() {
return domain.UserEmojiStatus{}, tgerr400("COLLECTIBLE_INVALID")
}
return value, nil
default:
return domain.UserEmojiStatus{}, inputConstructorInvalidErr()
}
}
// onAccountUpdateColor 持久化当前用户的消息 accent 或资料页背景色。
// 普通 peerColor 可清除（color flag absent）、可显式设置 color=0；collectible
// 颜色依赖礼物资产模型，当前阶段按范围外能力拒绝并记录在兼容矩阵。
@ -1746,6 +1814,41 @@ func (r *Router) onAccountGetDefaultEmojiStatuses(ctx context.Context, hash int6
return &tg.AccountEmojiStatuses{Hash: catalogHash, Statuses: statuses}, nil
}
// onAccountGetCollectibleEmojiStatuses returns the actor's active locally
// owned unique gifts as complete emojiStatusCollectible values. The bounded
// list order and hash are stable, so Android can safely reuse its cache.
func (r *Router) onAccountGetCollectibleEmojiStatuses(ctx context.Context, hash int64) (tg.AccountEmojiStatusesClass, error) {
userID, _, err := r.currentUserID(ctx)
if err != nil {
return nil, internalErr()
}
if r.deps.Gifts == nil {
return tdesktop.CollectibleEmojiStatuses(), nil
}
gifts, err := r.deps.Gifts.ListUniqueByOwner(ctx, domain.Peer{Type: domain.PeerTypeUser, ID: userID}, domain.MaxSavedStarGiftsLimit)
if err != nil {
return nil, internalErr()
}
ids := make([]int64, 0, len(gifts))
statuses := make([]tg.EmojiStatusClass, 0, len(gifts))
for _, gift := range gifts {
collectible, ok := domain.CollectibleEmojiStatus(gift)
if !ok {
continue
}
ids = append(ids, collectible.CollectibleID)
statuses = append(statuses, tgUserEmojiStatusValue(domain.UserEmojiStatus{
DocumentID: collectible.DocumentID,
Collectible: collectible,
}))
}
catalogHash := mediaCatalogHash(ids)
if hash != 0 && hash == catalogHash {
return &tg.AccountEmojiStatusesNotModified{}, nil
}
return &tg.AccountEmojiStatuses{Hash: catalogHash, Statuses: statuses}, nil
}
func (r *Router) pushUsernameUpdate(ctx context.Context, u domain.User) {
if u.ID == 0 {
return

134
internal/rpc/account_collectible_emoji_status_test.go Normal file
View file

@ -0,0 +1,134 @@
package rpc
import (
"context"
"testing"
"time"
"github.com/iamxvbaba/td/clock"
"github.com/iamxvbaba/td/tg"
"github.com/iamxvbaba/td/tgerr"
"go.uber.org/zap/zaptest"
appusers "telesrv/internal/app/users"
"telesrv/internal/domain"
"telesrv/internal/store/memory"
)
type collectibleEmojiGiftService struct {
GiftsService
gifts map[int64]domain.UniqueStarGift
}
func (s *collectibleEmojiGiftService) UniqueByID(_ context.Context, id int64) (domain.UniqueStarGift, bool, error) {
gift, ok := s.gifts[id]
return gift, ok, nil
}
func (s *collectibleEmojiGiftService) ListUniqueByOwner(_ context.Context, owner domain.Peer, limit int) ([]domain.UniqueStarGift, error) {
out := make([]domain.UniqueStarGift, 0, len(s.gifts))
for _, gift := range s.gifts {
if gift.Owner == owner && !gift.Burned && gift.OwnerAddress == "" {
out = append(out, gift)
}
}
if len(out) > limit {
out = out[:limit]
}
return out, nil
}
func collectibleEmojiTestGift(ownerID int64) domain.UniqueStarGift {
return domain.UniqueStarGift{
ID: 9001, Title: "Plush Pepe", Slug: "PlushPepe-1",
Owner: domain.Peer{Type: domain.PeerTypeUser, ID: ownerID},
Model: domain.StarGiftCollectibleAttribute{Document: &domain.Document{ID: 7101}},
Pattern: domain.StarGiftCollectibleAttribute{Document: &domain.Document{ID: 7201}},
Backdrop: domain.StarGiftCollectibleAttribute{
CenterColor: 0x102030, EdgeColor: 0x405060,
PatternColor: 0x708090, TextColor: 0xa0b0c0,
},
}
}
func TestAccountCollectibleEmojiStatusListSetAndRejectNonOwner(t *testing.T) {
ctx := context.Background()
userStore := memory.NewUserStore()
owner, err := userStore.Create(ctx, domain.User{AccessHash: 1, Phone: "15550009101", FirstName: "Owner"})
if err != nil {
t.Fatal(err)
}
other, err := userStore.Create(ctx, domain.User{AccessHash: 2, Phone: "15550009102", FirstName: "Other"})
if err != nil {
t.Fatal(err)
}
users := appusers.NewService(userStore)
if _, err := users.GrantPremium(ctx, owner.ID, 1); err != nil {
t.Fatalf("grant premium: %v", err)
}
gift := collectibleEmojiTestGift(owner.ID)
gifts := &collectibleEmojiGiftService{gifts: map[int64]domain.UniqueStarGift{gift.ID: gift}}
r := New(Config{}, Deps{Users: users, Gifts: gifts}, zaptest.NewLogger(t), clock.System)
ownerCtx := WithUserID(ctx, owner.ID)
listed, err := r.onAccountGetCollectibleEmojiStatuses(ownerCtx, 0)
if err != nil {
t.Fatalf("get collectible statuses: %v", err)
}
statuses, ok := listed.(*tg.AccountEmojiStatuses)
if !ok || len(statuses.Statuses) != 1 || statuses.Hash == 0 {
t.Fatalf("collectible list = %T %#v", listed, listed)
}
collectible, ok := statuses.Statuses[0].(*tg.EmojiStatusCollectible)
if !ok || collectible.CollectibleID != gift.ID || collectible.DocumentID != gift.Model.Document.ID ||
collectible.PatternDocumentID != gift.Pattern.Document.ID || collectible.PatternColor != gift.Backdrop.PatternColor {
t.Fatalf("collectible status = %T %#v", statuses.Statuses[0], statuses.Statuses[0])
}
if cached, err := r.onAccountGetCollectibleEmojiStatuses(ownerCtx, statuses.Hash); err != nil {
t.Fatalf("get cached collectible statuses: %v", err)
} else if _, ok := cached.(*tg.AccountEmojiStatusesNotModified); !ok {
t.Fatalf("cached collectible statuses = %T, want notModified", cached)
}
input := &tg.InputEmojiStatusCollectible{CollectibleID: gift.ID}
input.SetUntil(2_000_000_000)
if ok, err := r.onAccountUpdateEmojiStatus(ownerCtx, input); err != nil || !ok {
t.Fatalf("set collectible status: ok=%v err=%v", ok, err)
}
self, err := users.Self(ctx, owner.ID)
if err != nil {
t.Fatal(err)
}
if !self.EmojiStatusCollectible.Valid() || self.EmojiStatusCollectible.CollectibleID != gift.ID ||
self.EmojiStatusUntil != 2_000_000_000 {
t.Fatalf("persisted collectible status = %+v", self.EmojiStatus())
}
wire, ok := tgUserEmojiStatus(self, time.Now().Unix()).(*tg.EmojiStatusCollectible)
if !ok || wire.Slug != gift.Slug || wire.TextColor != gift.Backdrop.TextColor {
t.Fatalf("wire collectible = %T %#v", tgUserEmojiStatus(self, time.Now().Unix()), wire)
}
stolen := gift
stolen.Owner = domain.Peer{Type: domain.PeerTypeUser, ID: other.ID}
gifts.gifts[gift.ID] = stolen
if ok, err := r.onAccountUpdateEmojiStatus(ownerCtx, &tg.InputEmojiStatusCollectible{CollectibleID: gift.ID}); ok || !tgerr.Is(err, "COLLECTIBLE_INVALID") {
t.Fatalf("set non-owned collectible: ok=%v err=%v", ok, err)
}
}
func TestCollectibleEmojiStatusDurableUpdateProjection(t *testing.T) {
collectible, ok := domain.CollectibleEmojiStatus(collectibleEmojiTestGift(1))
if !ok {
t.Fatal("test gift should project")
}
value := domain.UserEmojiStatus{DocumentID: collectible.DocumentID, Collectible: collectible}
update, ok := tgOtherUpdateFromEvent(domain.UpdateEvent{
UserID: 1, Type: domain.UpdateEventUserEmojiStatus, EmojiStatus: value,
}).(*tg.UpdateUserEmojiStatus)
if !ok {
t.Fatal("durable event did not produce updateUserEmojiStatus")
}
if status, ok := update.EmojiStatus.(*tg.EmojiStatusCollectible); !ok || status.PatternDocumentID != collectible.PatternDocumentID {
t.Fatalf("durable wire status = %T %#v", update.EmojiStatus, update.EmojiStatus)
}
}

2
internal/rpc/bots_longtail.go
View file

@ -422,7 +422,7 @@ func (r *Router) onBotsUpdateUserEmojiStatus(ctx context.Context, req *tg.BotsUp
if !ok {
return false, userPermissionDeniedErr()
}
u, err := svc.UpdateEmojiStatus(ctx, target.ID, documentID, until)
u, err := svc.UpdateEmojiStatus(ctx, target.ID, domain.UserEmojiStatus{DocumentID: documentID, Until: until})
if err != nil {
if errors.Is(err, domain.ErrPremiumRequired) {
return false, tgerr400("PREMIUM_ACCOUNT_REQUIRED")

16
internal/rpc/channel_fanout_dispatcher.go
View file

@ -993,6 +993,22 @@ func (r *Router) enqueueChannelMessageFanout(ctx context.Context, originUserID i
})
}
// enqueueMonoforumMessageFanout only targets the subscriber sub-dialog and active parent-channel
// admins. A monoforum has no ordinary members, so member recomputation would either drop the
// message or leak it to an invalid historical membership.
func (r *Router) enqueueMonoforumMessageFanout(ctx context.Context, originUserID int64, mono domain.Channel, savedPeer domain.Peer, res domain.SendChannelMessageResult) {
fanoutCache := newViewerPeerCache(r)
ownerIDs := channelMessageFanoutOwnerIDs(res, []int64{savedPeer.ID})
r.enqueueChannelFanoutWithPrefetch(ctx, channelFanoutExplicit, originUserID, mono.ID, res.Event.Pts, res.Recipients,
0,
func(bgCtx context.Context, viewers []int64) {
r.prefetchChannelFanoutUsers(bgCtx, fanoutCache, viewers, ownerIDs)
},
func(bgCtx context.Context, viewerUserID int64) *tg.Updates {
return r.monoforumDeliveryUpdates(bgCtx, viewerUserID, mono, savedPeer, res)
})
}
// skipDeliverySet 把 SkipDeliveryUserIDs 切片转成查找集合（nil 表示无排除）。
func skipDeliverySet(ids []int64) map[int64]struct{} {
if len(ids) == 0 {

2
internal/rpc/channels_stubs.go
View file

@ -669,6 +669,8 @@ func channelInvalidErr(err error) error {
return tgerr400("CHAT_WRITE_FORBIDDEN")
case errors.Is(err, domain.ErrChannelAdminRequired):
return tgerr400("CHAT_ADMIN_REQUIRED")
case errors.Is(err, domain.ErrChannelMonoforumUnsupported):
return tgerr400("CHANNEL_MONOFORUM_UNSUPPORTED")
case errors.Is(err, domain.ErrUserAlreadyParticipant):
return tgerr400("USER_ALREADY_PARTICIPANT")
case errors.Is(err, domain.ErrReplyMessageIDInvalid):

8
internal/rpc/convert_channels_core.go
View file

@ -82,7 +82,7 @@ func tgChannelMessage(viewerUserID int64, m domain.ChannelMessage) tg.MessageCla
return nil
}
peer := &tg.PeerChannel{ChannelID: m.ChannelID}
outgoing := m.SenderUserID == viewerUserID && viewerUserID != 0 && m.From.Type != domain.PeerTypeChannel
outgoing := m.SenderUserID == viewerUserID && viewerUserID != 0 && (m.SavedPeer.ID != 0 || m.From.Type != domain.PeerTypeChannel)
from := tg.PeerClass(nil)
if !m.Post && m.SendAs != nil && m.SendAs.ID != 0 {
from = tgPeer(*m.SendAs)
@ -139,6 +139,12 @@ func tgChannelMessage(viewerUserID int64, m domain.ChannelMessage) tg.MessageCla
// 频道私信(monoforum):saved_peer_id 让客户端把消息归入对应订阅者子会话。
msg.SetSavedPeerID(tgPeer(m.SavedPeer))
}
if suggested, ok := tgSuggestedPost(m.SuggestedPost); ok {
msg.SetSuggestedPost(suggested)
}
if m.PaidMessageStars > 0 {
msg.SetPaidMessageStars(m.PaidMessageStars)
}
if m.Pinned {
msg.SetPinned(true)
}

3
internal/rpc/convert_dialogs.go
View file

@ -165,6 +165,9 @@ func tgDialogDraft(d domain.DialogDraft) tg.DraftMessageClass {
if rich := mustTGRichMessage(d.RichMessage); rich != nil {
out.SetRichMessage(*rich)
}
if suggested, ok := tgSuggestedPost(d.SuggestedPost); ok {
out.SetSuggestedPost(suggested)
}
return out
}

5
internal/rpc/convert_updates.go
View file

@ -237,6 +237,11 @@ func tgOtherUpdateFromEvent(event domain.UpdateEvent) tg.UpdateClass {
return nil
}
return &tg.UpdateUserPhone{UserID: event.UserID, Phone: event.Phone}
case domain.UpdateEventUserEmojiStatus:
if event.UserID == 0 || !event.EmojiStatus.Valid() {
return nil
}
return &tg.UpdateUserEmojiStatus{UserID: event.UserID, EmojiStatus: tgUserEmojiStatusValue(event.EmojiStatus)}
case domain.UpdateEventChannelState:
if event.Peer.Type != domain.PeerTypeChannel || event.Peer.ID == 0 {
return nil

32
internal/rpc/convert_users.go
View file

@ -85,9 +85,35 @@ func tgUserEmojiStatus(u domain.User, now int64) tg.EmojiStatusClass {
if !u.EmojiStatusActiveAt(now) {
return &tg.EmojiStatusEmpty{}
}
status := &tg.EmojiStatus{DocumentID: u.EmojiStatusDocumentID}
if u.EmojiStatusUntil > 0 {
status.SetUntil(u.EmojiStatusUntil)
return tgUserEmojiStatusValue(u.EmojiStatus())
}
// tgUserEmojiStatusValue converts an already validated absolute snapshot. It
// is shared by inline user projections and durable updateUserEmojiStatus.
func tgUserEmojiStatusValue(value domain.UserEmojiStatus) tg.EmojiStatusClass {
if !value.Valid() || value.Empty() {
return &tg.EmojiStatusEmpty{}
}
if collectible := value.Collectible; !collectible.Empty() {
status := &tg.EmojiStatusCollectible{
CollectibleID: collectible.CollectibleID,
DocumentID: collectible.DocumentID,
Title: collectible.Title,
Slug: collectible.Slug,
PatternDocumentID: collectible.PatternDocumentID,
CenterColor: collectible.CenterColor,
EdgeColor: collectible.EdgeColor,
PatternColor: collectible.PatternColor,
TextColor: collectible.TextColor,
}
if value.Until > 0 {
status.SetUntil(value.Until)
}
return status
}
status := &tg.EmojiStatus{DocumentID: value.DocumentID}
if value.Until > 0 {
status.SetUntil(value.Until)
}
return status
}

17
internal/rpc/deps.go
View file

@ -298,7 +298,14 @@ type UserIdentityService interface {
type UserPremiumService interface {
GrantPremium(ctx context.Context, userID int64, months int) (domain.User, error)
SweepExpiredPremium(ctx context.Context, now int64, limit int) ([]domain.User, error)
UpdateEmojiStatus(ctx context.Context, userID int64, documentID int64, until int) (domain.User, error)
UpdateEmojiStatus(ctx context.Context, userID int64, status domain.UserEmojiStatus) (domain.User, error)
}
// UserEmojiStatusDurableService exposes the aggregate state+event write used
// by account.updateEmojiStatus. The bool is false for lightweight stores that
// require the RPC Updates service to append the event separately.
type UserEmojiStatusDurableService interface {
UpdateEmojiStatusWithEvent(ctx context.Context, userID int64, status domain.UserEmojiStatus, date int, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.User, domain.UpdateEvent, bool, error)
}
// UserColorService 是 UsersService 的个人色板扩展能力。用于 account.updateColor
@ -429,6 +436,13 @@ type UpdatesService interface {
RecordDraftMessage(ctx context.Context, stateAuthKeyID [8]byte, userID int64, peer domain.Peer, topMsgID int, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error)
}
// UserEmojiStatusUpdatesService is the optional durable settings-update
// extension used by account.updateEmojiStatus. Keeping it separate preserves
// lightweight test/service implementations of the core UpdatesService.
type UserEmojiStatusUpdatesService interface {
RecordUserEmojiStatus(ctx context.Context, stateAuthKeyID [8]byte, userID int64, status domain.UserEmojiStatus, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error)
}
// ContactsService 抽象通讯录查询。
type ContactsService interface {
GetContacts(ctx context.Context, userID int64, hash int64) (domain.ContactList, bool, error)
@ -871,6 +885,7 @@ type GiftsService interface {
UniqueBySlug(ctx context.Context, slug string) (domain.UniqueStarGift, bool, error)
UniqueByID(ctx context.Context, uniqueGiftID int64) (domain.UniqueStarGift, bool, error)
UniqueByIDs(ctx context.Context, uniqueGiftIDs []int64) (map[int64]domain.UniqueStarGift, error)
ListUniqueByOwner(ctx context.Context, owner domain.Peer, limit int) ([]domain.UniqueStarGift, error)
Upgrade(ctx context.Context, req domain.StarGiftUpgradeRequest) (domain.StarGiftUpgradeResult, error)
UpgradeReceipt(ctx context.Context, userID int64, commandKey string) (domain.StarGiftUpgradeReceipt, bool, error)
RecordSavedGift(ctx context.Context, gift domain.SavedStarGift) (int64, error)

4
internal/rpc/errors.go
View file

@ -129,6 +129,10 @@ func effectIDInvalidErr() error { return tgerr.New(400, "EFFECT_ID_INVALID") }
func paymentUnsupportedErr() error { return tgerr.New(406, "PAYMENT_UNSUPPORTED") }
func allowPaymentRequiredErr(stars int64) error {
return tgerr.New(403, fmt.Sprintf("ALLOW_PAYMENT_REQUIRED_%d", stars))
}
func balanceTooLowErr() error { return tgerr.New(400, "BALANCE_TOO_LOW") }
func starsAmountInvalidErr() error { return tgerr.New(400, "STARS_AMOUNT_INVALID") }

41
internal/rpc/messages_dialogs.go
View file

@ -3,10 +3,12 @@ package rpc
import (
"context"
"errors"
"unicode/utf8"
"github.com/iamxvbaba/td/tg"
"go.uber.org/zap"
"telesrv/internal/domain"
"unicode/utf8"
)
func (r *Router) onMessagesSaveDraft(ctx context.Context, req *tg.MessagesSaveDraftRequest) (bool, error) {
@ -159,8 +161,18 @@ func (r *Router) dialogDraftFromSaveDraft(ctx context.Context, userID int64, pee
if len(req.Entities) > maxMessageEntityCount {
return domain.DialogDraft{}, limitInvalidErr()
}
if !req.SuggestedPost.Zero() {
return domain.DialogDraft{}, suggestedPostPeerInvalidErr()
suggestedInput, hasSuggestedPost := req.GetSuggestedPost()
suggestedPost, err := domainSuggestedPost(suggestedInput, hasSuggestedPost)
if err != nil {
return domain.DialogDraft{}, err
}
if hasSuggestedPost {
if peer.Type != domain.PeerTypeChannel || r.deps.Channels == nil {
return domain.DialogDraft{}, suggestedPostPeerInvalidErr()
}
if _, _, resolveErr := r.deps.Channels.ResolveMonoforumSend(ctx, userID, peer.ID); resolveErr != nil {
return domain.DialogDraft{}, suggestedPostPeerInvalidErr()
}
}
replyTo, err := r.messageReplyFromInput(ctx, userID, peer, req.ReplyTo)
if err != nil {
@ -182,17 +194,18 @@ func (r *Router) dialogDraftFromSaveDraft(ctx context.Context, userID int64, pee
topMessageID = replyTo.TopMessageID
}
return domain.DialogDraft{
Peer: peer,
TopMessageID: topMessageID,
Date: date,
NoWebpage: req.NoWebpage,
InvertMedia: req.InvertMedia,
Message: req.Message,
Entities: domainMessageEntities(req.Entities),
ReplyTo: replyTo,
WebPage: webpage,
Effect: req.Effect,
RichMessage: richMessage,
Peer: peer,
TopMessageID: topMessageID,
Date: date,
NoWebpage: req.NoWebpage,
InvertMedia: req.InvertMedia,
Message: req.Message,
Entities: domainMessageEntities(req.Entities),
ReplyTo: replyTo,
WebPage: webpage,
Effect: req.Effect,
SuggestedPost: suggestedPost,
RichMessage: richMessage,
}, nil
}

107
internal/rpc/messages_monoforum.go
View file

@ -103,8 +103,8 @@ func (r *Router) monoforumSavedHistory(ctx context.Context, userID int64, mono d
}, nil
}
// monoforumChats 投影客户端 materialize monoforum 私信所需的频道:monoforum 自身(直接投影,管理员
// 非其成员故不能走可见性受限的 GetChannels)+ 母广播频道(管理员是其成员)。
// monoforumChats 投影客户端 materialize monoforum 私信所需的频道:monoforum 自身直接投影，
// 再按 viewer 补母广播频道。订阅者没有 monoforum member row，管理员身份也只来自母频道。
func (r *Router) monoforumChats(ctx context.Context, userID int64, mono domain.Channel) []tg.ChatClass {
chats := []tg.ChatClass{tgChannelChatForView(userID, domain.ChannelView{Channel: mono})}
if mono.LinkedMonoforumID != 0 && r.deps.Channels != nil {
@ -151,8 +151,8 @@ func (r *Router) monoforumSubscriberUsers(ctx context.Context, userID int64, dia
return r.tgUsers(found)
}
// monoforumReplyPresent 判断 sendMessage 的 reply_to 是否带 monoforum_peer_id(频道私信发送的唯一标志)。
// 普通发送恒不带,故据此 gate monoforum 分支,普通发送热路径零额外成本。
// monoforumReplyPresent 判断 sendMessage 的 reply_to 是否显式携带 monoforum_peer_id。
// 管理员回复必须带目标订阅者；普通订阅者按官方 TDesktop 行为不携带 reply_to，目标由调用者推导。
func monoforumReplyPresent(input tg.InputReplyToClass) bool {
switch v := input.(type) {
case *tg.InputReplyToMonoForum:
@ -189,46 +189,75 @@ func (r *Router) monoforumReplyTargetPeer(userID int64, input tg.InputReplyToCla
return r.domainPeerFromInputPeer(userID, inputPeer)
}
func (r *Router) monoforumSavedPeerForSender(userID int64, isAdmin bool, replyTo tg.InputReplyToClass) (domain.Peer, error) {
savedPeer := domain.Peer{Type: domain.PeerTypeUser, ID: userID}
if monoforumReplyPresent(replyTo) {
var valid bool
savedPeer, valid = r.monoforumReplyTargetPeer(userID, replyTo)
if !valid || savedPeer.Type != domain.PeerTypeUser || savedPeer.ID == 0 {
return domain.Peer{}, replyToMonoforumPeerInvalidErr()
}
} else if isAdmin {
return domain.Peer{}, replyToMonoforumPeerInvalidErr()
}
if !isAdmin && savedPeer.ID != userID {
return domain.Peer{}, replyToMonoforumPeerInvalidErr()
}
return savedPeer, nil
}
// monoforumMessageReplyFromInput separates the sub-dialog selector from the actual message reply.
// InputReplyToMonoForum only selects a subscriber; InputReplyToMessage may carry both the selector
// and a real reply_to_msg_id, so clear flags.5 before reusing the common structural validator.
func (r *Router) monoforumMessageReplyFromInput(ctx context.Context, userID int64, peer domain.Peer, input tg.InputReplyToClass) (*domain.MessageReply, error) {
switch value := input.(type) {
case nil, *tg.InputReplyToMonoForum:
return nil, nil
case *tg.InputReplyToMessage:
if value == nil {
return nil, nil
}
clean := *value
clean.Flags.Unset(5)
clean.MonoforumPeerID = nil
return r.messageReplyFromInput(ctx, userID, peer, &clean)
default:
return r.messageReplyFromInput(ctx, userID, peer, input)
}
}
// sendMonoforumMessage 处理向频道私信(monoforum)发送:订阅者发到自己的子会话,管理员回复到目标订阅者。
// saved_peer 来自 reply_to 的 monoforum_peer_id;管理员可写任意订阅者子会话,普通订阅者只能写自己的。
func (r *Router) sendMonoforumMessage(ctx context.Context, userID int64, peer domain.Peer, req *tg.MessagesSendMessageRequest, fingerprint []byte, preflighted bool) (tg.UpdatesClass, error) {
// saved_peer 对订阅者由调用者推导、对管理员来自 reply_to;管理员可写任意订阅者子会话,订阅者只能写自己的。
func (r *Router) sendMonoforumMessage(ctx context.Context, userID int64, peer domain.Peer, mono domain.Channel, isAdmin bool, req domain.SendMonoforumMessageRequest) (tg.UpdatesClass, error) {
if r.deps.Channels == nil {
return nil, notImplementedErr()
}
if peer.Type != domain.PeerTypeChannel || peer.ID == 0 {
return nil, peerIDInvalidErr()
}
mono, isAdmin, err := r.deps.Channels.ResolveMonoforumSend(ctx, userID, peer.ID)
if err != nil {
if errors.Is(err, domain.ErrChannelInvalid) {
// 带 monoforum_peer_id 却不是 monoforum 频道。
return nil, tgerr400("CHANNEL_MONOFORUM_UNSUPPORTED")
}
return nil, internalErr()
}
savedPeer, ok := r.monoforumReplyTargetPeer(userID, req.ReplyTo)
if !ok || savedPeer.Type != domain.PeerTypeUser || savedPeer.ID == 0 {
if mono.ID != peer.ID || !mono.Monoforum || req.SavedPeer.Type != domain.PeerTypeUser || req.SavedPeer.ID == 0 {
return nil, replyToMonoforumPeerInvalidErr()
}
if !isAdmin && savedPeer.ID != userID {
if !isAdmin && req.SavedPeer.ID != userID {
// 普通订阅者只能写自己的子会话,不能写他人的。
return nil, replyToMonoforumPeerInvalidErr()
}
res, err := r.deps.Channels.SendMonoforumMessage(ctx, domain.SendMonoforumMessageRequest{
MonoforumID: mono.ID,
SenderUserID: userID,
SavedPeer: savedPeer,
RandomID: req.RandomID,
IdempotencyFingerprint: fingerprint,
IdempotencyPreflighted: preflighted,
Message: req.Message,
Entities: domainMessageEntities(req.Entities),
Date: int(r.clock.Now().Unix()),
})
req.MonoforumID = mono.ID
req.SenderUserID = userID
if req.Date == 0 {
req.Date = int(r.clock.Now().Unix())
}
res, err := r.deps.Channels.SendMonoforumMessage(ctx, req)
if err != nil {
return nil, messageSendErr(err)
}
return r.monoforumSendUpdates(ctx, userID, mono, savedPeer, res), nil
if req.ClearDraft {
r.clearDraftAfterSend(ctx, userID, peer, req.ReplyTo)
}
if !res.Duplicate {
r.enqueueMonoforumMessageFanout(ctx, userID, mono, req.SavedPeer, res)
}
return r.monoforumSendUpdates(ctx, userID, mono, req.SavedPeer, res), nil
}
// monoforumSendUpdates 给发送者构造回声 Updates:updateMessageID(关联 random_id)+ updateNewChannelMessage
@ -245,6 +274,11 @@ func (r *Router) monoforumSendUpdates(ctx context.Context, userID int64, mono do
newMsg.Message = &tg.MessageEmpty{ID: res.Message.ID}
}
updates = append(updates, newMsg)
if res.SenderStarsBalance != nil && res.Message.SenderUserID == userID {
updates = append(updates, &tg.UpdateStarsBalance{
Balance: &tg.StarsAmount{Amount: res.SenderStarsBalance.Balance},
})
}
date := int(r.clock.Now().Unix())
if res.Duplicate && res.ReplayDeleteEvent != nil {
if deleted := tgChannelUpdate(userID, *res.ReplayDeleteEvent); deleted != nil {
@ -261,3 +295,18 @@ func (r *Router) monoforumSendUpdates(ctx context.Context, userID int64, mono do
Date: date,
}
}
func (r *Router) monoforumDeliveryUpdates(ctx context.Context, userID int64, mono domain.Channel, savedPeer domain.Peer, res domain.SendChannelMessageResult) *tg.Updates {
updates, _ := r.monoforumSendUpdates(ctx, userID, mono, savedPeer, res).(*tg.Updates)
if updates == nil {
return nil
}
filtered := make([]tg.UpdateClass, 0, len(updates.Updates))
for _, update := range updates.Updates {
if _, randomMapping := update.(*tg.UpdateMessageID); !randomMapping {
filtered = append(filtered, update)
}
}
updates.Updates = filtered
return updates
}

227
internal/rpc/messages_monoforum_rpc_test.go
View file

@ -2,6 +2,7 @@ package rpc
import (
"context"
"strings"
"testing"
"github.com/iamxvbaba/td/bin"
@ -10,6 +11,7 @@ import (
"go.uber.org/zap/zaptest"
appchannels "telesrv/internal/app/channels"
appdialogs "telesrv/internal/app/dialogs"
appusers "telesrv/internal/app/users"
"telesrv/internal/domain"
"telesrv/internal/store/memory"
@ -17,7 +19,7 @@ import (
// TestMonoforumSavedDialogsAndHistory 验证频道私信(monoforum)读侧 RPC:管理员经
// getSavedDialogs(parent_peer=monoforum) 看订阅者子会话列表、经 getSavedHistory 看某订阅者历史
// (消息带 saved_peer_id);非管理员被拒。
// (消息带 saved_peer_id);订阅者经普通 getHistory 只看自己的子会话。
func TestMonoforumSavedDialogsAndHistory(t *testing.T) {
ctx := context.Background()
userStore := memory.NewUserStore()
@ -98,12 +100,31 @@ func TestMonoforumSavedDialogsAndHistory(t *testing.T) {
if !seenChats[monoID] || !seenChats[created.Channel.ID] {
t.Fatalf("main monoforum chats = %+v, want monoforum %d and parent %d", seenChats, monoID, created.Channel.ID)
}
var deniedRaw bin.Buffer
if err := (&tg.MessagesGetHistoryRequest{Peer: monoInput, Limit: 20}).Encode(&deniedRaw); err != nil {
var subscriberRaw bin.Buffer
if err := (&tg.MessagesGetHistoryRequest{Peer: monoInput, Limit: 20}).Encode(&subscriberRaw); err != nil {
t.Fatalf("encode non-admin getHistory(monoforum): %v", err)
}
if _, err := r.Dispatch(WithUserID(ctx, sub.ID), [8]byte{}, 0, &deniedRaw); err == nil {
t.Fatalf("non-admin getHistory(monoforum) = nil err, want denied")
subscriberEnc, err := r.Dispatch(WithUserID(ctx, sub.ID), [8]byte{}, 0, &subscriberRaw)
if err != nil {
t.Fatalf("non-admin getHistory(monoforum): %v", err)
}
subscriberHistory, ok := subscriberEnc.(*tg.MessagesChannelMessages)
if !ok {
t.Fatalf("non-admin getHistory(monoforum) = %T, want *tg.MessagesChannelMessages", subscriberEnc)
}
if len(subscriberHistory.Messages) != 1 {
t.Fatalf("non-admin getHistory(monoforum) = %d msgs, want own sublist message", len(subscriberHistory.Messages))
}
subscriberMessage, ok := subscriberHistory.Messages[0].(*tg.Message)
if !ok || subscriberMessage.Message != "hello channel" {
t.Fatalf("non-admin history[0] = %#v, want own 'hello channel'", subscriberHistory.Messages[0])
}
subscriberSavedPeer, ok := subscriberMessage.GetSavedPeerID()
if !ok {
t.Fatalf("non-admin history message missing saved_peer_id")
}
if peer, ok := subscriberSavedPeer.(*tg.PeerUser); !ok || peer.UserID != sub.ID {
t.Fatalf("non-admin history saved_peer_id = %#v, want self %d", subscriberSavedPeer, sub.ID)
}
// 管理员看私信列表。
@ -179,9 +200,9 @@ func TestMonoforumSavedDialogsAndHistory(t *testing.T) {
}
}
// TestMonoforumSendMessageWritePath 验证写侧:订阅者经 sendMessage(peer=monoforum,
// reply_to=InputReplyToMonoForum{自己}) 发私信;管理员回复到目标订阅者;订阅者不能写他人子会话;
// 普通发送(无 monoforum_peer_id)不受影响。
// TestMonoforumSendMessageWritePath 验证写侧:订阅者按 TDesktop 实际请求仅以
// peer=monoforum 发到自己的子会话;管理员必须显式指定目标订阅者;suggested_post 被持久化返回;
// 订阅者不能写他人子会话。
func TestMonoforumSendMessageWritePath(t *testing.T) {
ctx := context.Background()
userStore := memory.NewUserStore()
@ -196,39 +217,160 @@ func TestMonoforumSendMessageWritePath(t *testing.T) {
channelStore := memory.NewChannelStore()
channelSvc := appchannels.NewService(channelStore)
dialogSvc := appdialogs.NewService(memory.NewDialogStore(), channelStore)
r := New(Config{}, Deps{
Users: appusers.NewService(userStore),
Channels: channelSvc,
Dialogs: dialogSvc,
}, zaptest.NewLogger(t), clock.System)
created, err := channelSvc.CreateChannel(ctx, owner.ID, domain.CreateChannelRequest{Title: "DM Broadcast", Broadcast: true, Date: 1000})
if err != nil {
t.Fatalf("create channel: %v", err)
}
enabled, err := channelStore.SetPaidMessagesPrice(ctx, owner.ID, created.Channel.ID, 0, true)
enabled, err := channelStore.SetPaidMessagesPrice(ctx, owner.ID, created.Channel.ID, 10, true)
if err != nil {
t.Fatalf("enable DM: %v", err)
}
monoID := enabled.Channel.LinkedMonoforumID
monoInput := &tg.InputPeerChannel{ChannelID: monoID}
mono, err := channelStore.GetChannelByID(ctx, monoID)
if err != nil {
t.Fatalf("get monoforum: %v", err)
}
monoInput := &tg.InputPeerChannel{ChannelID: monoID, AccessHash: mono.AccessHash}
monoChannelInput := &tg.InputChannel{ChannelID: monoID, AccessHash: mono.AccessHash}
// 订阅者发私信到自己的子会话。
// 订阅者不是 monoforum 成员，但 TDesktop 打开会话时必须能读取 full channel shell。
full, err := r.onChannelsGetFullChannel(WithUserID(ctx, sub.ID), monoChannelInput)
if err != nil {
t.Fatalf("subscriber getFullChannel(monoforum): %v", err)
}
if full == nil || full.FullChat == nil {
t.Fatalf("subscriber getFullChannel(monoforum) = %#v, want full chat", full)
}
// monoforum 永远不能通过 join 变成普通频道成员，否则会生成错误的 joined service message。
if _, err := r.onChannelsJoinChannel(WithUserID(ctx, sub.ID), monoChannelInput); err == nil || !strings.Contains(err.Error(), "CHANNEL_MONOFORUM_UNSUPPORTED") {
t.Fatalf("subscriber joinChannel(monoforum) err = %v, want CHANNEL_MONOFORUM_UNSUPPORTED", err)
}
// TDesktop 在发送前保存相同 suggested_post 草稿；它必须可写、可恢复，不能变成 CHANNEL_PRIVATE。
draftSuggested := tg.SuggestedPost{}
draftSuggested.SetPrice(&tg.StarsAmount{Amount: 10})
draftSuggested.SetScheduleDate(1_700_100_000)
draftReq := &tg.MessagesSaveDraftRequest{Peer: monoInput, Message: "pending suggested post"}
draftReq.SetSuggestedPost(draftSuggested)
if ok, err := r.onMessagesSaveDraft(WithUserID(ctx, sub.ID), draftReq); err != nil || !ok {
t.Fatalf("subscriber saveDraft(monoforum) = %v, %v; want true, nil", ok, err)
}
storedDraft, found, err := dialogSvc.GetDraft(ctx, sub.ID, domain.Peer{Type: domain.PeerTypeChannel, ID: monoID}, 0)
if err != nil || !found {
t.Fatalf("get persisted monoforum draft = %+v, %v, %v; want found", storedDraft, found, err)
}
if storedDraft.Message != "pending suggested post" || storedDraft.SuggestedPost == nil || storedDraft.SuggestedPost.Price == nil || storedDraft.SuggestedPost.Price.Amount != 10 || storedDraft.SuggestedPost.ScheduleDate != 1_700_100_000 {
t.Fatalf("persisted monoforum draft = %+v, want suggested post content", storedDraft)
}
tooLow := &tg.MessagesSendMessageRequest{Peer: monoInput, Message: "under-authorized", RandomID: 554}
tooLow.SetAllowPaidStars(9)
if _, err := r.onMessagesSendMessage(WithUserID(ctx, sub.ID), tooLow); err == nil || !strings.Contains(err.Error(), "ALLOW_PAYMENT_REQUIRED") || !strings.Contains(err.Error(), "(10)") {
t.Fatalf("under-authorized paid message err = %v, want ALLOW_PAYMENT_REQUIRED_10", err)
}
// TDesktop 的订阅者请求不携带 InputReplyToMonoForum;服务端必须从调用者推导 saved_peer=self。
subReq := &tg.MessagesSendMessageRequest{Peer: monoInput, Message: "hi from sub", RandomID: 555}
subReq.SetReplyTo(&tg.InputReplyToMonoForum{MonoforumPeerID: &tg.InputPeerUser{UserID: sub.ID}})
subReq.ClearDraft = true
subReq.SetAllowPaidStars(20)
suggestedInput := tg.SuggestedPost{}
suggestedInput.SetPrice(&tg.StarsAmount{Amount: 10})
suggestedInput.SetScheduleDate(1_700_100_000)
subReq.SetSuggestedPost(suggestedInput)
subUpd, err := r.onMessagesSendMessage(WithUserID(ctx, sub.ID), subReq)
if err != nil {
t.Fatalf("subscriber sendMessage(monoforum): %v", err)
}
if _, ok := subUpd.(*tg.Updates); !ok {
subUpdates, ok := subUpd.(*tg.Updates)
if !ok {
t.Fatalf("subscriber send updates = %T, want *tg.Updates", subUpd)
}
var subMessageID int
var subPaidStars, subBalance int64
for _, update := range subUpdates.Updates {
if newMessage, ok := update.(*tg.UpdateNewChannelMessage); ok {
if message, ok := newMessage.Message.(*tg.Message); ok {
subMessageID = message.ID
subPaidStars, _ = message.GetPaidMessageStars()
}
}
if balance, ok := update.(*tg.UpdateStarsBalance); ok {
if amount, ok := balance.Balance.(*tg.StarsAmount); ok {
subBalance = amount.Amount
}
}
}
if subMessageID == 0 || subPaidStars != 10 || subBalance != 990 {
t.Fatalf("subscriber send updates id/paid/balance = %d/%d/%d, want id>0/10/990: %#v", subMessageID, subPaidStars, subBalance, subUpdates.Updates)
}
if _, found, err := dialogSvc.GetDraft(ctx, sub.ID, domain.Peer{Type: domain.PeerTypeChannel, ID: monoID}, 0); err != nil || found {
t.Fatalf("clear_draft after paid send found/err = %v/%v, want false/nil", found, err)
}
duplicateUpd, err := r.onMessagesSendMessage(WithUserID(ctx, sub.ID), subReq)
if err != nil {
t.Fatalf("subscriber paid replay: %v", err)
}
duplicateUpdates, ok := duplicateUpd.(*tg.Updates)
if !ok {
t.Fatalf("subscriber paid replay = %T, want *tg.Updates", duplicateUpd)
}
var duplicateBalance int64
for _, update := range duplicateUpdates.Updates {
if balance, ok := update.(*tg.UpdateStarsBalance); ok {
if amount, ok := balance.Balance.(*tg.StarsAmount); ok {
duplicateBalance = amount.Amount
}
}
}
if duplicateBalance != 990 {
t.Fatalf("subscriber paid replay balance = %d, want 990 without a second debit", duplicateBalance)
}
// 管理员回复到该订阅者的子会话。
// 管理员回复到该订阅者的子会话：同一个 inputReplyToMessage 同时携带真实 reply id
// 和 monoforum target，两部分都必须保留。
adminReq := &tg.MessagesSendMessageRequest{Peer: monoInput, Message: "admin reply", RandomID: 556}
adminReq.SetReplyTo(&tg.InputReplyToMonoForum{MonoforumPeerID: &tg.InputPeerUser{UserID: sub.ID}})
if _, err := r.onMessagesSendMessage(WithUserID(ctx, owner.ID), adminReq); err != nil {
adminReply := &tg.InputReplyToMessage{ReplyToMsgID: subMessageID}
adminReply.SetMonoforumPeerID(&tg.InputPeerUser{UserID: sub.ID})
adminReq.SetReplyTo(adminReply)
adminReq.SetAllowPaidStars(100)
adminUpd, err := r.onMessagesSendMessage(WithUserID(ctx, owner.ID), adminReq)
if err != nil {
t.Fatalf("admin reply sendMessage(monoforum): %v", err)
}
if updates, ok := adminUpd.(*tg.Updates); ok {
for _, update := range updates.Updates {
if _, balance := update.(*tg.UpdateStarsBalance); balance {
t.Fatalf("admin free reply emitted a balance debit: %#v", updates.Updates)
}
if newMessage, ok := update.(*tg.UpdateNewChannelMessage); ok {
if message, ok := newMessage.Message.(*tg.Message); ok {
if stars, paid := message.GetPaidMessageStars(); paid || stars != 0 {
t.Fatalf("admin reply paid_message_stars = %d/%v, want 0/false", stars, paid)
}
}
}
}
}
// 带媒体的 suggested post 必须走同一 monoforum 子会话，不能退化为普通频道消息或丢 flags。
mediaSuggested := tg.SuggestedPost{}
mediaSuggested.SetPrice(&tg.StarsAmount{Amount: 15})
mediaReq := &tg.MessagesSendMediaRequest{
Peer: monoInput, RandomID: 558, Message: "media suggestion",
Media: &tg.InputMediaContact{PhoneNumber: "+15550003003", FirstName: "Media", LastName: "Contact", Vcard: ""},
}
mediaReq.SetSuggestedPost(mediaSuggested)
mediaReq.SetAllowPaidStars(15)
if _, err := r.onMessagesSendMedia(WithUserID(ctx, sub.ID), mediaReq); err != nil {
t.Fatalf("subscriber sendMedia(monoforum): %v", err)
}
// 订阅者不能写他人(owner)的子会话。
sneaky := &tg.MessagesSendMessageRequest{Peer: monoInput, Message: "sneaky", RandomID: 557}
@ -237,7 +379,7 @@ func TestMonoforumSendMessageWritePath(t *testing.T) {
t.Fatalf("subscriber writing another's sublist = nil err, want REPLY_TO_MONOFORUM_PEER_INVALID")
}
// 经管理员读历史:子会话含两条(订阅者发 + 管理员回复),倒序。
// 经管理员读历史:子会话含三条(订阅者文本 + 管理员回复 + 订阅者媒体),倒序。
hreq := &tg.MessagesGetSavedHistoryRequest{Peer: &tg.InputPeerUser{UserID: sub.ID}}
hreq.SetParentPeer(monoInput)
hres, err := r.onMessagesGetSavedHistory(WithUserID(ctx, owner.ID), hreq)
@ -248,11 +390,54 @@ func TestMonoforumSendMessageWritePath(t *testing.T) {
if !ok {
t.Fatalf("getSavedHistory = %T, want *tg.MessagesMessagesSlice", hres)
}
if len(slice.Messages) != 2 {
t.Fatalf("history = %d msgs, want 2 (sub + admin)", len(slice.Messages))
if len(slice.Messages) != 3 {
t.Fatalf("history = %d msgs, want 3 (sub text + admin + sub media)", len(slice.Messages))
}
top, ok := slice.Messages[0].(*tg.Message)
if !ok || top.Message != "admin reply" {
t.Fatalf("history[0] = %#v, want newest 'admin reply'", slice.Messages[0])
if !ok || top.Message != "media suggestion" {
t.Fatalf("history[0] = %#v, want newest media suggestion", slice.Messages[0])
}
if _, ok := top.Media.(*tg.MessageMediaContact); !ok {
t.Fatalf("history[0] media = %T, want MessageMediaContact", top.Media)
}
if paid, ok := top.GetPaidMessageStars(); !ok || paid != 10 {
t.Fatalf("media paid_message_stars = %d/%v, want actual configured price 10", paid, ok)
}
topSuggested, ok := top.GetSuggestedPost()
if !ok {
t.Fatalf("media message missing suggested_post")
}
topPrice, ok := topSuggested.GetPrice()
if !ok {
t.Fatalf("media suggested_post missing price")
}
if stars, ok := topPrice.(*tg.StarsAmount); !ok || stars.Amount != 15 {
t.Fatalf("media suggested_post price = %#v, want 15 Stars", topPrice)
}
adminMessage, ok := slice.Messages[1].(*tg.Message)
if !ok || adminMessage.Message != "admin reply" {
t.Fatalf("history[1] = %#v, want admin reply", slice.Messages[1])
}
if header, ok := adminMessage.ReplyTo.(*tg.MessageReplyHeader); !ok || header.ReplyToMsgID != subMessageID {
t.Fatalf("admin reply header = %#v, want reply_to_msg_id %d", adminMessage.ReplyTo, subMessageID)
}
suggestedMessage, ok := slice.Messages[2].(*tg.Message)
if !ok {
t.Fatalf("history[2] = %T, want *tg.Message", slice.Messages[2])
}
suggested, ok := suggestedMessage.GetSuggestedPost()
if !ok {
t.Fatalf("subscriber message missing suggested_post")
}
price, ok := suggested.GetPrice()
if !ok {
t.Fatalf("suggested_post missing price")
}
stars, ok := price.(*tg.StarsAmount)
if !ok || stars.Amount != 10 {
t.Fatalf("suggested_post price = %#v, want 10 Stars", price)
}
if scheduleDate, ok := suggested.GetScheduleDate(); !ok || scheduleDate != 1_700_100_000 {
t.Fatalf("suggested_post schedule = %d/%v, want 1700100000/true", scheduleDate, ok)
}
}

76
internal/rpc/messages_send.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) onMessagesSendMessage(ctx context.Context, req *tg.MessagesSendMessageRequest) (tg.UpdatesClass, error) {
@ -63,12 +65,39 @@ func (r *Router) onMessagesSendMessage(ctx context.Context, req *tg.MessagesSend
sendErr = internalErr()
return nil, sendErr
}
// 频道私信(monoforum):仅当 reply_to 带 monoforum_peer_id 时走专用发送路径(普通发送恒不带,
// 故此 gate 对普通发送零额外成本)。peer 解析为 monoforum 频道时按订阅者子会话发送。
if monoforumReplyPresent(req.ReplyTo) {
savedPeer, valid := r.monoforumReplyTargetPeer(userID, req.ReplyTo)
if !valid || savedPeer.Type != domain.PeerTypeUser || savedPeer.ID == 0 {
sendErr = replyToMonoforumPeerInvalidErr()
suggestedInput, hasSuggestedPost := req.GetSuggestedPost()
// monoforum 普通用户发送不带 reply_to，saved_peer 必须由服务端推导为自己；管理员回复才必须
// 显式携带 monoforum_peer_id。仅凭 reply_to 判路由会把用户请求误送进普通 megagroup 路径。
var mono domain.Channel
var monoforum, monoforumAdmin bool
if peer.Type == domain.PeerTypeChannel && r.deps.Channels != nil {
mono, monoforumAdmin, err = r.deps.Channels.ResolveMonoforumSend(ctx, userID, peer.ID)
switch {
case err == nil:
monoforum = true
case !errors.Is(err, domain.ErrChannelInvalid):
sendErr = internalErr()
return nil, sendErr
}
}
if hasSuggestedPost && !monoforum {
sendErr = suggestedPostPeerInvalidErr()
return nil, sendErr
}
if monoforum {
suggestedPost, suggestedErr := domainSuggestedPost(suggestedInput, hasSuggestedPost)
if suggestedErr != nil {
sendErr = suggestedErr
return nil, sendErr
}
savedPeer, err := r.monoforumSavedPeerForSender(userID, monoforumAdmin, req.ReplyTo)
if err != nil {
sendErr = err
return nil, sendErr
}
replyTo, err := r.monoforumMessageReplyFromInput(ctx, userID, peer, req.ReplyTo)
if err != nil {
sendErr = err
return nil, sendErr
}
replay, err := r.lookupChannelSendReplay(ctx, userID, peer.ID, savedPeer, req.RandomID, idempotencyFingerprint)
@ -78,6 +107,9 @@ func (r *Router) onMessagesSendMessage(ctx context.Context, req *tg.MessagesSend
}
if replay.found {
duplicate = true
if req.ClearDraft {
r.clearDraftAfterSend(ctx, userID, peer, replyTo)
}
return r.monoforumSendUpdates(ctx, userID, replay.channel.Channel, savedPeer, replay.channel), nil
}
if err := r.checkSendRateLimit(ctx, userID, 1); err != nil {
@ -89,13 +121,30 @@ func (r *Router) onMessagesSendMessage(ctx context.Context, req *tg.MessagesSend
sendErr = err
return nil, sendErr
}
updates, err := r.sendMonoforumMessage(ctx, userID, checkedPeer, req, idempotencyFingerprint, replay.checked)
updates, err := r.sendMonoforumMessage(ctx, userID, checkedPeer, mono, monoforumAdmin, domain.SendMonoforumMessageRequest{
SavedPeer: savedPeer,
RandomID: req.RandomID,
IdempotencyFingerprint: idempotencyFingerprint,
IdempotencyPreflighted: replay.checked,
Message: req.Message,
Entities: domainMessageEntities(req.Entities),
ReplyTo: replyTo,
Silent: req.Silent,
NoForwards: req.Noforwards,
SuggestedPost: suggestedPost,
AllowPaidStars: req.AllowPaidStars,
ClearDraft: req.ClearDraft,
})
if err != nil {
sendErr = err
return nil, sendErr
}
return updates, nil
}
if req.AllowPaidStars > 0 {
sendErr = paymentUnsupportedErr()
return nil, sendErr
}
replay, err := r.lookupOutgoingReplay(ctx, userID, peer, req.RandomID, idempotencyFingerprint)
if err != nil {
sendErr = err
@ -201,7 +250,12 @@ func (r *Router) onMessagesSendMessage(ctx context.Context, req *tg.MessagesSend
}
func messageSendErr(err error) error {
var paymentRequired *domain.StarsPaymentRequiredError
switch {
case errors.As(err, &paymentRequired) && paymentRequired.Stars > 0:
return allowPaymentRequiredErr(paymentRequired.Stars)
case errors.Is(err, domain.ErrStarsInsufficient):
return balanceTooLowErr()
case errors.Is(err, domain.ErrUserFrozen):
return frozenMethodInvalidErr()
case errors.Is(err, domain.ErrReplyMessageIDInvalid):
@ -314,10 +368,8 @@ func sendMessageUnsupportedOptionErr(req *tg.MessagesSendMessageRequest) error {
// req.Effect 不再一律拒绝：消息特效已实现，合法性在 messageEffectInvalid 单独校验。
case req.AllowPaidStars < 0:
return starsAmountInvalidErr()
case req.AllowPaidStars > 0 || req.AllowPaidFloodskip:
case req.AllowPaidFloodskip:
return paymentUnsupportedErr()
case !req.SuggestedPost.Zero():
return suggestedPostPeerInvalidErr()
default:
return nil
}

73
internal/rpc/messages_suggested_post.go Normal file
View file

@ -0,0 +1,73 @@
package rpc
import (
"github.com/iamxvbaba/td/tg"
"telesrv/internal/domain"
)
const (
minSuggestedPostStars int64 = 5
maxSuggestedPostStars int64 = 100_000
minSuggestedPostNanoTON int64 = 10_000_000
maxSuggestedPostNanoTON int64 = 10_000_000_000_000
)
func domainSuggestedPost(input tg.SuggestedPost, present bool) (*domain.SuggestedPost, error) {
if !present {
return nil, nil
}
if input.GetAccepted() || input.GetRejected() {
return nil, tgerr400("SUGGESTED_POST_AMOUNT_INVALID")
}
out := &domain.SuggestedPost{}
if date, ok := input.GetScheduleDate(); ok {
if date <= 0 {
return nil, scheduleDateInvalidErr()
}
out.ScheduleDate = date
}
if price, ok := input.GetPrice(); ok {
switch value := price.(type) {
case *tg.StarsAmount:
if value == nil || value.Amount < minSuggestedPostStars || value.Amount > maxSuggestedPostStars ||
value.Nanos < 0 || value.Nanos >= 1_000_000_000 || value.Amount == maxSuggestedPostStars && value.Nanos != 0 {
return nil, tgerr400("SUGGESTED_POST_AMOUNT_INVALID")
}
out.Price = &domain.SuggestedPostPrice{Kind: domain.SuggestedPostPriceStars, Amount: value.Amount, Nanos: value.Nanos}
case *tg.StarsTonAmount:
if value == nil || value.Amount < minSuggestedPostNanoTON || value.Amount > maxSuggestedPostNanoTON {
return nil, tgerr400("SUGGESTED_POST_AMOUNT_INVALID")
}
out.Price = &domain.SuggestedPostPrice{Kind: domain.SuggestedPostPriceTON, Amount: value.Amount}
default:
return nil, tgerr400("SUGGESTED_POST_AMOUNT_INVALID")
}
}
return out, nil
}
func tgSuggestedPost(input *domain.SuggestedPost) (tg.SuggestedPost, bool) {
if input == nil {
return tg.SuggestedPost{}, false
}
out := tg.SuggestedPost{}
if input.Accepted {
out.SetAccepted(true)
}
if input.Rejected {
out.SetRejected(true)
}
if input.ScheduleDate > 0 {
out.SetScheduleDate(input.ScheduleDate)
}
if input.Price != nil {
switch input.Price.Kind {
case domain.SuggestedPostPriceStars:
out.SetPrice(&tg.StarsAmount{Amount: input.Price.Amount, Nanos: input.Price.Nanos})
case domain.SuggestedPostPriceTON:
out.SetPrice(&tg.StarsTonAmount{Amount: input.Price.Amount})
}
}
return out, true
}

2
internal/rpc/payments.go
View file

@ -453,6 +453,8 @@ func tgStarsTransactions(in []domain.StarsTransaction) []tg.StarsTransaction {
switch t.Reason {
case domain.StarsReasonReaction:
item.Reaction = true
case domain.StarsReasonPaidMessage:
item.SetPaidMessages(1)
case domain.StarsReasonGift:
item.Gift = true
case domain.StarsReasonGiftUpgrade:

16
internal/rpc/payments_stars_rpc_test.go
View file

@ -85,6 +85,22 @@ func TestOnPaymentsGetStarsTransactions(t *testing.T) {
}
}
func TestTGStarsTransactionsPaidMessage(t *testing.T) {
out := tgStarsTransactions([]domain.StarsTransaction{{
ID: 1, UserID: 42, Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: 50},
Amount: -10, Date: 1700002002, Reason: domain.StarsReasonPaidMessage, Title: "Paid message",
}})
if len(out) != 1 {
t.Fatalf("paid-message transactions = %d, want 1", len(out))
}
if paid, ok := out[0].GetPaidMessages(); !ok || paid != 1 {
t.Fatalf("paid_messages = %d/%v, want 1/true", paid, ok)
}
if amount, ok := out[0].Amount.(*tg.StarsAmount); !ok || amount.Amount != -10 {
t.Fatalf("paid-message amount = %#v, want -10", out[0].Amount)
}
}
// deps.Stars==nil 兜底：返回合法的空 starsStatus（余额 0），不崩。
func TestOnPaymentsGetStarsStatusNilDeps(t *testing.T) {
r := New(Config{}, Deps{}, zaptest.NewLogger(t), clock.System)

83
internal/rpc/send_media.go
View file

@ -274,6 +274,89 @@ func (r *Router) onMessagesSendMedia(ctx context.Context, req *tg.MessagesSendMe
if !ok || peer.ID == 0 {
return nil, peerIDInvalidErr()
}
suggestedInput, hasSuggestedPost := req.GetSuggestedPost()
var mono domain.Channel
var monoforum, monoforumAdmin bool
if peer.Type == domain.PeerTypeChannel && r.deps.Channels != nil {
mono, monoforumAdmin, err = r.deps.Channels.ResolveMonoforumSend(ctx, userID, peer.ID)
switch {
case err == nil:
monoforum = true
case !errors.Is(err, domain.ErrChannelInvalid):
return nil, internalErr()
}
}
if hasSuggestedPost && !monoforum {
return nil, suggestedPostPeerInvalidErr()
}
if monoforum {
if req.AllowPaidStars < 0 {
return nil, starsAmountInvalidErr()
}
if req.AllowPaidFloodskip {
return nil, paymentUnsupportedErr()
}
if req.ScheduleDate != 0 && !scheduleDateIsImmediate(req.ScheduleDate, int(r.clock.Now().Unix())) {
return nil, scheduleDateInvalidErr()
}
suggestedPost, err := domainSuggestedPost(suggestedInput, hasSuggestedPost)
if err != nil {
return nil, err
}
savedPeer, err := r.monoforumSavedPeerForSender(userID, monoforumAdmin, req.ReplyTo)
if err != nil {
return nil, err
}
replyTo, err := r.monoforumMessageReplyFromInput(ctx, userID, peer, req.ReplyTo)
if err != nil {
return nil, err
}
replay, err := r.lookupChannelSendReplay(ctx, userID, peer.ID, savedPeer, req.RandomID, idempotencyFingerprint)
if err != nil {
return nil, err
}
if replay.found {
if req.ClearDraft {
r.clearDraftAfterSend(ctx, userID, peer, replyTo)
}
return r.monoforumSendUpdates(ctx, userID, replay.channel.Channel, savedPeer, replay.channel), nil
}
if r.messageEffectInvalid(ctx, req.Effect) {
return nil, effectIDInvalidErr()
}
if err := r.checkSendRateLimit(ctx, userID, 1); err != nil {
return nil, err
}
checkedPeer, err := r.checkedDomainPeerFromInputPeer(ctx, userID, req.Peer)
if err != nil {
return nil, err
}
media, err := r.resolveInputMedia(ctx, userID, req.Media)
if err != nil {
return nil, err
}
if media == nil {
return nil, mediaInvalidErr()
}
return r.sendMonoforumMessage(ctx, userID, checkedPeer, mono, monoforumAdmin, domain.SendMonoforumMessageRequest{
SavedPeer: savedPeer,
RandomID: req.RandomID,
IdempotencyFingerprint: idempotencyFingerprint,
IdempotencyPreflighted: replay.checked,
Message: req.Message,
Entities: domainMessageEntities(req.Entities),
Media: media,
ReplyTo: replyTo,
Silent: req.Silent,
NoForwards: req.Noforwards,
SuggestedPost: suggestedPost,
AllowPaidStars: req.AllowPaidStars,
ClearDraft: req.ClearDraft,
})
}
if req.AllowPaidStars > 0 || req.AllowPaidFloodskip {
return nil, paymentUnsupportedErr()
}
replay, err := r.lookupOutgoingReplay(ctx, userID, peer, req.RandomID, idempotencyFingerprint)
if err != nil {
return nil, err