fix: sync account freeze peer visibility
This commit is contained in:
A
parent
d9875b5caa
commit
eba402946a
26 changed files with 1034 additions and 19 deletions
107
internal/rpc/account_freeze_worker.go
Normal file
107
internal/rpc/account_freeze_worker.go
Normal file
|
|
@ -0,0 +1,107 @@
|
|||
package rpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/iamxvbaba/td/tg"
|
||||
"go.uber.org/zap"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
)
|
||||
|
||||
type accountFreezeNotificationService interface {
|
||||
ClaimAccountFreezeNotifications(ctx context.Context, now time.Time, limit int, lease time.Duration) ([]domain.AccountFreezeNotification, error)
|
||||
CompleteAccountFreezeNotification(ctx context.Context, id, version int64, now time.Time) error
|
||||
}
|
||||
|
||||
// RunAccountFreezeNotifications drains the crash-safe, coalesced non-pts
|
||||
// updateUser queue. One attempt is enough for online delivery; offline clients
|
||||
// recover the current state from viewer-scoped user hydration.
|
||||
func (r *Router) RunAccountFreezeNotifications(ctx context.Context, interval time.Duration, batch int) {
|
||||
if interval <= 0 {
|
||||
interval = time.Minute
|
||||
}
|
||||
if batch <= 0 {
|
||||
batch = 500
|
||||
}
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
r.drainAccountFreezeNotifications(ctx, batch)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
case <-r.accountFreezeWake:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Router) drainAccountFreezeNotifications(ctx context.Context, batch int) {
|
||||
svc, ok := r.deps.AccountFreeze.(accountFreezeNotificationService)
|
||||
if !ok || r.deps.Users == nil {
|
||||
return
|
||||
}
|
||||
for {
|
||||
now := r.clock.Now().UTC()
|
||||
claimCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
notifications, err := svc.ClaimAccountFreezeNotifications(claimCtx, now, batch, 2*time.Minute)
|
||||
cancel()
|
||||
if err != nil {
|
||||
r.log.Warn("claim account freeze notifications failed", zap.Error(err))
|
||||
return
|
||||
}
|
||||
for _, notification := range notifications {
|
||||
r.dispatchAccountFreezeNotification(ctx, svc, notification)
|
||||
}
|
||||
if len(notifications) < batch {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Router) dispatchAccountFreezeNotification(ctx context.Context, svc accountFreezeNotificationService, notification domain.AccountFreezeNotification) {
|
||||
peer := domain.Peer{Type: domain.PeerTypeUser, ID: notification.FrozenUserID}
|
||||
if contacts, ok := r.deps.Contacts.(interface{ InvalidateViewers(...int64) }); ok {
|
||||
contacts.InvalidateViewers(notification.TargetUserID)
|
||||
}
|
||||
if dialogs, ok := r.deps.Dialogs.(interface {
|
||||
InvalidateDialog(int64, domain.Peer)
|
||||
}); ok {
|
||||
dialogs.InvalidateDialog(notification.TargetUserID, peer)
|
||||
}
|
||||
r.invalidateRPCProjectionForPeer(notification.TargetUserID, peer)
|
||||
|
||||
loadCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
user, found, err := r.deps.Users.ByID(loadCtx, notification.TargetUserID, notification.FrozenUserID)
|
||||
cancel()
|
||||
if err != nil {
|
||||
r.log.Warn("load frozen user projection for notification failed",
|
||||
zap.Int64("target_user_id", notification.TargetUserID),
|
||||
zap.Int64("frozen_user_id", notification.FrozenUserID),
|
||||
zap.Int64("version", notification.Version),
|
||||
zap.Error(err))
|
||||
return
|
||||
}
|
||||
if !found {
|
||||
user = domain.User{ID: notification.FrozenUserID, Deleted: true}
|
||||
}
|
||||
pushCtx, pushCancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
r.pushUserUpdates(pushCtx, notification.TargetUserID, &tg.Updates{
|
||||
Updates: []tg.UpdateClass{&tg.UpdateUser{UserID: notification.FrozenUserID}},
|
||||
Users: r.tgUsersForViewer(notification.TargetUserID, []domain.User{user}),
|
||||
Date: int(r.clock.Now().Unix()),
|
||||
})
|
||||
pushCancel()
|
||||
|
||||
completeCtx, completeCancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
err = svc.CompleteAccountFreezeNotification(completeCtx, notification.ID, notification.Version, r.clock.Now().UTC())
|
||||
completeCancel()
|
||||
if err != nil {
|
||||
r.log.Warn("complete account freeze notification failed",
|
||||
zap.Int64("notification_id", notification.ID),
|
||||
zap.Int64("version", notification.Version),
|
||||
zap.Error(err))
|
||||
}
|
||||
}
|
||||
136
internal/rpc/account_freeze_worker_test.go
Normal file
136
internal/rpc/account_freeze_worker_test.go
Normal file
|
|
@ -0,0 +1,136 @@
|
|||
package rpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/iamxvbaba/td/clock"
|
||||
"github.com/iamxvbaba/td/tg"
|
||||
"go.uber.org/zap/zaptest"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
)
|
||||
|
||||
func TestAccountFreezeNotificationPushesCurrentViewerProjection(t *testing.T) {
|
||||
const (
|
||||
viewerID = int64(1001)
|
||||
frozenID = int64(1002)
|
||||
)
|
||||
sessions := &captureSessions{}
|
||||
freezeSvc := &freezeWorkerService{}
|
||||
users := &freezeWorkerUsers{user: domain.User{
|
||||
ID: frozenID,
|
||||
FirstName: "Frozen",
|
||||
RestrictionReasons: domain.AccountFrozenRestrictionReasons(),
|
||||
}}
|
||||
r := New(Config{}, Deps{
|
||||
AccountFreeze: freezeSvc,
|
||||
Users: users,
|
||||
Sessions: sessions,
|
||||
}, zaptest.NewLogger(t), clock.System)
|
||||
|
||||
r.dispatchAccountFreezeNotification(context.Background(), freezeSvc, domain.AccountFreezeNotification{
|
||||
ID: 7, TargetUserID: viewerID, FrozenUserID: frozenID, Version: 4, Frozen: true,
|
||||
})
|
||||
|
||||
if len(freezeSvc.completed) != 1 || freezeSvc.completed[0] != [2]int64{7, 4} {
|
||||
t.Fatalf("completed = %v, want [[7 4]]", freezeSvc.completed)
|
||||
}
|
||||
if got := sessions.pushedUserIDs(); len(got) != 1 || got[0] != viewerID {
|
||||
t.Fatalf("pushed user IDs = %v, want [%d]", got, viewerID)
|
||||
}
|
||||
updates, ok := sessions.lastUserPush().(*tg.Updates)
|
||||
if !ok || len(updates.Updates) != 1 || len(updates.Users) != 1 {
|
||||
t.Fatalf("push = %#v, want updateUser plus projected user", sessions.lastUserPush())
|
||||
}
|
||||
if update, ok := updates.Updates[0].(*tg.UpdateUser); !ok || update.UserID != frozenID {
|
||||
t.Fatalf("update = %#v, want updateUser(%d)", updates.Updates[0], frozenID)
|
||||
}
|
||||
projected, ok := updates.Users[0].(*tg.User)
|
||||
if !ok || !projected.Restricted {
|
||||
t.Fatalf("projected user = %#v, want restricted user", updates.Users[0])
|
||||
}
|
||||
reasons, ok := projected.GetRestrictionReason()
|
||||
if !ok || len(reasons) != 1 || reasons[0].Reason != "frozen" {
|
||||
t.Fatalf("projected restriction = %+v ok=%v", reasons, ok)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAccountFreezeNotificationLoadsCurrentStateAndRetriesLoadFailure(t *testing.T) {
|
||||
const (
|
||||
viewerID = int64(2001)
|
||||
frozenID = int64(2002)
|
||||
)
|
||||
sessions := &captureSessions{}
|
||||
freezeSvc := &freezeWorkerService{}
|
||||
users := &freezeWorkerUsers{err: errors.New("projection unavailable")}
|
||||
r := New(Config{}, Deps{
|
||||
AccountFreeze: freezeSvc,
|
||||
Users: users,
|
||||
Sessions: sessions,
|
||||
}, zaptest.NewLogger(t), clock.System)
|
||||
notification := domain.AccountFreezeNotification{
|
||||
ID: 8, TargetUserID: viewerID, FrozenUserID: frozenID, Version: 5, Frozen: true,
|
||||
}
|
||||
|
||||
r.dispatchAccountFreezeNotification(context.Background(), freezeSvc, notification)
|
||||
if len(freezeSvc.completed) != 0 || len(sessions.pushedUserIDs()) != 0 {
|
||||
t.Fatalf("failed load completed=%v pushes=%v, want retry without push", freezeSvc.completed, sessions.pushedUserIDs())
|
||||
}
|
||||
|
||||
// The queued payload may say frozen, but delivery must hydrate the latest
|
||||
// viewer projection so a newer unfreeze can never be overwritten by stale work.
|
||||
users.err = nil
|
||||
users.user = domain.User{ID: frozenID, FirstName: "Active"}
|
||||
r.dispatchAccountFreezeNotification(context.Background(), freezeSvc, notification)
|
||||
updates, ok := sessions.lastUserPush().(*tg.Updates)
|
||||
if !ok || len(updates.Users) != 1 {
|
||||
t.Fatalf("push = %#v", sessions.lastUserPush())
|
||||
}
|
||||
projected, ok := updates.Users[0].(*tg.User)
|
||||
if !ok || projected.Restricted {
|
||||
t.Fatalf("latest projected user = %#v, want unrestricted", updates.Users[0])
|
||||
}
|
||||
if len(freezeSvc.completed) != 1 || freezeSvc.completed[0] != [2]int64{8, 5} {
|
||||
t.Fatalf("completed = %v, want [[8 5]]", freezeSvc.completed)
|
||||
}
|
||||
}
|
||||
|
||||
type freezeWorkerService struct {
|
||||
completed [][2]int64
|
||||
}
|
||||
|
||||
func (*freezeWorkerService) AccountFreeze(context.Context, int64) (domain.AccountFreeze, bool, error) {
|
||||
return domain.AccountFreeze{}, false, nil
|
||||
}
|
||||
|
||||
func (*freezeWorkerService) ClaimAccountFreezeNotifications(context.Context, time.Time, int, time.Duration) ([]domain.AccountFreezeNotification, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (s *freezeWorkerService) CompleteAccountFreezeNotification(_ context.Context, id, version int64, _ time.Time) error {
|
||||
s.completed = append(s.completed, [2]int64{id, version})
|
||||
return nil
|
||||
}
|
||||
|
||||
type freezeWorkerUsers struct {
|
||||
user domain.User
|
||||
err error
|
||||
}
|
||||
|
||||
func (s *freezeWorkerUsers) Self(context.Context, int64) (domain.User, error) {
|
||||
return s.user, s.err
|
||||
}
|
||||
|
||||
func (s *freezeWorkerUsers) ByID(context.Context, int64, int64) (domain.User, bool, error) {
|
||||
return s.user, s.err == nil, s.err
|
||||
}
|
||||
|
||||
func (s *freezeWorkerUsers) ByIDs(context.Context, int64, []int64) ([]domain.User, error) {
|
||||
if s.err != nil {
|
||||
return nil, s.err
|
||||
}
|
||||
return []domain.User{s.user}, nil
|
||||
}
|
||||
|
|
@ -46,3 +46,20 @@ func (r *Router) NotifyStarsBalanceChanged(ctx context.Context, balance domain.S
|
|||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
// NotifyAccountFreezeChanged invalidates target-scoped projections immediately
|
||||
// and wakes the durable audience nudge worker. Cross-instance cache invalidation
|
||||
// is also carried by the committed user_visibility read-model notification.
|
||||
func (r *Router) NotifyAccountFreezeChanged(_ context.Context, freeze domain.AccountFreeze) error {
|
||||
if r == nil || freeze.UserID == 0 {
|
||||
return nil
|
||||
}
|
||||
r.invalidateRPCProjectionForUser(freeze.UserID)
|
||||
if r.accountFreezeWake != nil {
|
||||
select {
|
||||
case r.accountFreezeWake <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ func tgSelfUser(u domain.User) *tg.User {
|
|||
applyTgUserBotFields(out, u)
|
||||
applyTgUserPremiumFields(out, u)
|
||||
applyTgUserColorFields(out, u)
|
||||
applyTgUserRestrictionFields(out, u)
|
||||
if u.LinkedCommunityID != 0 {
|
||||
out.SetLinkedCommunityID(u.LinkedCommunityID)
|
||||
}
|
||||
|
|
@ -60,6 +61,7 @@ func tgUser(u domain.User) *tg.User {
|
|||
applyTgUserBotFields(out, u)
|
||||
applyTgUserPremiumFields(out, u)
|
||||
applyTgUserColorFields(out, u)
|
||||
applyTgUserRestrictionFields(out, u)
|
||||
if u.LinkedCommunityID != 0 {
|
||||
out.SetLinkedCommunityID(u.LinkedCommunityID)
|
||||
}
|
||||
|
|
@ -69,6 +71,28 @@ func tgUser(u domain.User) *tg.User {
|
|||
return out
|
||||
}
|
||||
|
||||
func applyTgUserRestrictionFields(out *tg.User, u domain.User) {
|
||||
if out == nil || len(u.RestrictionReasons) == 0 {
|
||||
return
|
||||
}
|
||||
reasons := make([]tg.RestrictionReason, 0, len(u.RestrictionReasons))
|
||||
for _, reason := range u.RestrictionReasons {
|
||||
if reason.Platform == "" || reason.Reason == "" || reason.Text == "" {
|
||||
continue
|
||||
}
|
||||
reasons = append(reasons, tg.RestrictionReason{
|
||||
Platform: reason.Platform,
|
||||
Reason: reason.Reason,
|
||||
Text: reason.Text,
|
||||
})
|
||||
}
|
||||
if len(reasons) == 0 {
|
||||
return
|
||||
}
|
||||
out.Restricted = true
|
||||
out.SetRestrictionReason(reasons)
|
||||
}
|
||||
|
||||
// applyTgUserPremiumFields 由到期时间即时派生 premium flag（bit28，独立位）与
|
||||
// emoji status。判断用真实时钟：premium 的权威来源是 premium_expires_at 本身，
|
||||
// 到期即停发，正确性不依赖后台 sweeper（它只负责清理与 updateUser 通知）；
|
||||
|
|
|
|||
62
internal/rpc/convert_users_restriction_test.go
Normal file
62
internal/rpc/convert_users_restriction_test.go
Normal file
|
|
@ -0,0 +1,62 @@
|
|||
package rpc
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/iamxvbaba/td/bin"
|
||||
"github.com/iamxvbaba/td/tg"
|
||||
"github.com/iamxvbaba/td/tlprofile"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
)
|
||||
|
||||
func TestTgUserEncodesFrozenRestriction(t *testing.T) {
|
||||
user := tgUser(domain.User{
|
||||
ID: 1001,
|
||||
FirstName: "Frozen",
|
||||
RestrictionReasons: domain.AccountFrozenRestrictionReasons(),
|
||||
})
|
||||
if !user.Restricted {
|
||||
t.Fatal("tg user restricted=false, want true")
|
||||
}
|
||||
reasons, ok := user.GetRestrictionReason()
|
||||
if !ok || len(reasons) != 1 {
|
||||
t.Fatalf("restriction_reason = %+v ok=%v, want one reason", reasons, ok)
|
||||
}
|
||||
if got := reasons[0]; got.Platform != "all" || got.Reason != "frozen" || got.Text != "This account is frozen." {
|
||||
t.Fatalf("restriction_reason = %+v", got)
|
||||
}
|
||||
|
||||
for profile := tlprofile.Profile225; profile <= tlprofile.Profile228; profile++ {
|
||||
wire := &bin.Buffer{}
|
||||
if err := tlprofile.EncodeObject(profile, user, wire); err != nil {
|
||||
t.Fatalf("encode layer %d frozen user: %v", profile, err)
|
||||
}
|
||||
decoded, err := tlprofile.DecodeObject(profile, &bin.Buffer{Buf: wire.Copy()}, tlprofile.Limits{})
|
||||
if err != nil {
|
||||
t.Fatalf("decode layer %d frozen user: %v", profile, err)
|
||||
}
|
||||
exact, ok := decoded.(*tg.User)
|
||||
if !ok || !exact.Restricted {
|
||||
t.Fatalf("layer %d user = %#v, want restricted", profile, decoded)
|
||||
}
|
||||
exactReasons, ok := exact.GetRestrictionReason()
|
||||
if !ok || len(exactReasons) != 1 || exactReasons[0].Reason != "frozen" {
|
||||
t.Fatalf("layer %d restriction = %+v ok=%v", profile, exactReasons, ok)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestTgUserSkipsIncompleteRestriction(t *testing.T) {
|
||||
user := tgUser(domain.User{
|
||||
ID: 1001,
|
||||
FirstName: "Active",
|
||||
RestrictionReasons: []domain.UserRestrictionReason{{Platform: "all", Reason: "frozen"}},
|
||||
})
|
||||
if user.Restricted {
|
||||
t.Fatal("incomplete restriction was encoded")
|
||||
}
|
||||
if reasons, ok := user.GetRestrictionReason(); ok || len(reasons) != 0 {
|
||||
t.Fatalf("restriction_reason = %+v ok=%v, want omitted", reasons, ok)
|
||||
}
|
||||
}
|
||||
|
|
@ -164,6 +164,7 @@ type Router struct {
|
|||
stickerCatalog *stickerCatalogCache
|
||||
transientPrivateBigReactions transientPrivateBigReactionCache
|
||||
accountSettings *accountSettingsCache
|
||||
accountFreezeWake chan struct{}
|
||||
// webPageResolveSem 是链接预览异步解析的并发信号量（有界）：发送后把 pending 占位
|
||||
// 解析为卡片并就地替换。满则丢弃任务（消息留 pending）。nil=未启用（测试可直接调
|
||||
// resolvePendingWebPage 同步验证）。
|
||||
|
|
@ -237,7 +238,7 @@ func New(cfg Config, deps Deps, log *zap.Logger, clk clock.Clock) *Router {
|
|||
if instanceID == "" {
|
||||
instanceID = fmt.Sprintf("%016x", randomNonZeroInt64())
|
||||
}
|
||||
r := &Router{cfg: cfg, log: log, clock: clk, deps: deps, exactProfiles: make(map[clientInfoSessionKey]exactSessionProfileEntry), authLayerEvidence: make(map[[8]byte]authLayerDefaultEvidence), presence: newPresenceTracker(), callbacks: newCallbackRegistry(deps.BotCallbacks), inlines: newInlineRegistry(botInlineQueryTTL, deps.Inline), webviews: newWebViewRegistry(webViewSessionTTL, deps.Inline), loginTokens: newLoginTokenRegistry(), botAPIUpdates: newBotAPIUpdateNotifier(), tempKeyResolveCache: newTempKeyResolveCache(cfg.TempKeyResolveCacheMaxEntries), storyProjectionCache: newStoryProjectionCache(clk.Now), storyPinnedCache: newStoryPinnedAvailableCache(clk.Now), storyPinnedListCache: newStoryPinnedStoriesCache(clk.Now), channelFullBotCache: newChannelFullBotInfoCache(clk.Now), userFullProjectionCache: newUserFullProjectionCache(clk.Now), peerSettingsProjectionCache: newPeerSettingsProjectionCache(clk.Now), channelFullProjectionCache: newChannelFullProjectionCache(clk.Now), emojiStickers: newEmojiStickerIndex(clk.Now), notifySettings: newNotifySettingsCache(clk.Now), stickerCatalog: newStickerCatalogCache(clk.Now), accountSettings: newAccountSettingsCache(clk.Now), instanceID: instanceID}
|
||||
r := &Router{cfg: cfg, log: log, clock: clk, deps: deps, exactProfiles: make(map[clientInfoSessionKey]exactSessionProfileEntry), authLayerEvidence: make(map[[8]byte]authLayerDefaultEvidence), presence: newPresenceTracker(), callbacks: newCallbackRegistry(deps.BotCallbacks), inlines: newInlineRegistry(botInlineQueryTTL, deps.Inline), webviews: newWebViewRegistry(webViewSessionTTL, deps.Inline), loginTokens: newLoginTokenRegistry(), botAPIUpdates: newBotAPIUpdateNotifier(), tempKeyResolveCache: newTempKeyResolveCache(cfg.TempKeyResolveCacheMaxEntries), storyProjectionCache: newStoryProjectionCache(clk.Now), storyPinnedCache: newStoryPinnedAvailableCache(clk.Now), storyPinnedListCache: newStoryPinnedStoriesCache(clk.Now), channelFullBotCache: newChannelFullBotInfoCache(clk.Now), userFullProjectionCache: newUserFullProjectionCache(clk.Now), peerSettingsProjectionCache: newPeerSettingsProjectionCache(clk.Now), channelFullProjectionCache: newChannelFullProjectionCache(clk.Now), emojiStickers: newEmojiStickerIndex(clk.Now), notifySettings: newNotifySettingsCache(clk.Now), stickerCatalog: newStickerCatalogCache(clk.Now), accountSettings: newAccountSettingsCache(clk.Now), accountFreezeWake: make(chan struct{}, 1), instanceID: instanceID}
|
||||
r.channelFanout = newChannelFanoutDispatcher(r, defaultChannelFanoutShards, defaultChannelFanoutBuffer)
|
||||
r.botAPIEnqueueQueue = newBotAPIEnqueueDispatcher(log, defaultBotAPIEnqueueBuffer)
|
||||
r.webPageResolveSem = make(chan struct{}, webPageResolveConcurrency)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue