diff --git a/cmd/telesrv/main.go b/cmd/telesrv/main.go index 60c33051..a4a4558e 100644 --- a/cmd/telesrv/main.go +++ b/cmd/telesrv/main.go @@ -305,6 +305,7 @@ func run(logger *zap.Logger) error { updateEventStore := postgres.NewUpdateEventStore(pool, postgres.WithUpdateEventLogger(logger.Named("store").Named("updates"))) readModelVersionStore := storepkg.NewCachedReadModelVersionStore(postgres.NewReadModelVersionStore(pool), 0, 0) dispatchOutboxStore := postgres.NewDispatchOutboxStore(pool, postgres.WithLeaseTimeout(cfg.OutboxLeaseTimeout)) + bootstrapUpdateStore := postgres.NewBootstrapUpdateJobStore(pool) boxIDAllocator := redisstore.NewBoxIDAllocator(rdb, postgres.NewMessageBoxCounterSource(pool)) channelIDAllocator := redisstore.NewChannelIDAllocator(rdb, postgres.NewChannelIDCounterSource(pool)) channelMessageIDAllocator := redisstore.NewChannelMessageIDAllocator(rdb, postgres.NewChannelMessageIDCounterSource(pool)) @@ -595,34 +596,35 @@ func run(logger *zap.Logger) error { TempKeyResolveCacheTTL: 5 * time.Second, TempKeyResolveCacheMaxEntries: cfg.TempKeyResolveCacheMaxEntries, }, rpc.Deps{ - Auth: authService, - Account: accountService, - Privacy: privacyService, - Help: help.NewService(helpStore, helpStore, help.WithMapboxToken(cfg.MapboxToken)), - AICompose: aiComposeService, - Users: usersService, - Updates: updatesService, - Contacts: contactsService, - Dialogs: dialogsService, - Messages: messagesService, - Channels: channelsService, - Files: filesService, - Bots: botsService, - Polls: pollsapp.NewService(pollStore), - Stories: storiesapp.NewService(storyStore, storiesapp.WithChannelStoryAccess(channelsService)), - Phone: phoneService, - SecretChats: secretChatService, - Stars: starsService, - Gifts: giftsService, - Passkey: passkeyService, - Themes: themeService, - GroupCalls: groupCallsService, - SFU: sfuService, - TURN: turnService, - LangPack: langPackService, - Sessions: activeSessions, - Inline: inlineRegistryStore, - Limiter: rateLimiter, + Auth: authService, + Account: accountService, + Privacy: privacyService, + Help: help.NewService(helpStore, helpStore, help.WithMapboxToken(cfg.MapboxToken)), + AICompose: aiComposeService, + Users: usersService, + Updates: updatesService, + BootstrapUpdates: bootstrapUpdateStore, + Contacts: contactsService, + Dialogs: dialogsService, + Messages: messagesService, + Channels: channelsService, + Files: filesService, + Bots: botsService, + Polls: pollsapp.NewService(pollStore), + Stories: storiesapp.NewService(storyStore, storiesapp.WithChannelStoryAccess(channelsService)), + Phone: phoneService, + SecretChats: secretChatService, + Stars: starsService, + Gifts: giftsService, + Passkey: passkeyService, + Themes: themeService, + GroupCalls: groupCallsService, + SFU: sfuService, + TURN: turnService, + LangPack: langPackService, + Sessions: activeSessions, + Inline: inlineRegistryStore, + Limiter: rateLimiter, }, logger.Named("rpc"), clock.System) readModelListener := postgres.NewReadModelChangeListener(cfg.PostgresDSN, postgres.ReadModelCacheSet{ ReadModelVersions: readModelVersionStore, @@ -664,6 +666,7 @@ func run(logger *zap.Logger) error { rpc.WithOutboxPushTimeout(cfg.OutboundPushTimeout), rpc.WithOutboxUpdateBuilder(router.BuildOutboxUpdates), ).Run(ctx) + go rpc.NewBootstrapUpdateDispatcher(router, logger.Named("rpc").Named("bootstrap")).Run(ctx) go rpc.NewScheduledDispatcher(router, logger.Named("rpc").Named("scheduled")).Run(ctx) go rpc.NewExpiryDispatcher(router, logger.Named("rpc").Named("expiry")).Run(ctx) go rpc.NewPhoneExpiryDispatcher(router, logger.Named("rpc").Named("phone-expiry"), cfg.CallExpiryInterval).Run(ctx) diff --git a/deploy/migrations/0052_bootstrap_update_jobs.down.sql b/deploy/migrations/0052_bootstrap_update_jobs.down.sql new file mode 100644 index 00000000..ee66703c --- /dev/null +++ b/deploy/migrations/0052_bootstrap_update_jobs.down.sql @@ -0,0 +1,2 @@ +DROP INDEX IF EXISTS public.user_update_events_new_message_box_idx; +DROP TABLE IF EXISTS public.bootstrap_update_jobs; diff --git a/deploy/migrations/0052_bootstrap_update_jobs.up.sql b/deploy/migrations/0052_bootstrap_update_jobs.up.sql new file mode 100644 index 00000000..8d8bc9a2 --- /dev/null +++ b/deploy/migrations/0052_bootstrap_update_jobs.up.sql @@ -0,0 +1,34 @@ +CREATE TABLE IF NOT EXISTS public.bootstrap_update_jobs ( + id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + kind character varying(32) NOT NULL, + user_id bigint NOT NULL REFERENCES public.users(id) ON DELETE CASCADE, + auth_key_id bigint DEFAULT 0 NOT NULL, + session_id bigint DEFAULT 0 NOT NULL, + message_box_id integer NOT NULL, + status character varying(16) DEFAULT 'pending'::character varying NOT NULL, + attempts integer DEFAULT 0 NOT NULL, + last_error text DEFAULT ''::text NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL, + ready_at timestamp with time zone, + published_at timestamp with time zone, + CONSTRAINT bootstrap_update_jobs_kind_check CHECK ((kind)::text = ANY (ARRAY[('login_message'::character varying)::text])), + CONSTRAINT bootstrap_update_jobs_status_check CHECK ((status)::text = ANY (ARRAY[('pending'::character varying)::text, ('ready'::character varying)::text, ('publishing'::character varying)::text, ('published'::character varying)::text, ('failed'::character varying)::text])), + CONSTRAINT bootstrap_update_jobs_message_box_fkey FOREIGN KEY (user_id, message_box_id) REFERENCES public.message_boxes(owner_user_id, box_id) ON DELETE CASCADE +); + +CREATE UNIQUE INDEX IF NOT EXISTS bootstrap_update_jobs_unique_login + ON public.bootstrap_update_jobs (kind, user_id, message_box_id); + +CREATE INDEX IF NOT EXISTS bootstrap_update_jobs_ready_idx + ON public.bootstrap_update_jobs (COALESCE(ready_at, created_at), user_id, id) + WHERE (status)::text = 'ready'::text; + +CREATE INDEX IF NOT EXISTS bootstrap_update_jobs_publishing_stale_idx + ON public.bootstrap_update_jobs (updated_at, user_id, id) + WHERE (status)::text = 'publishing'::text; + +CREATE INDEX IF NOT EXISTS user_update_events_new_message_box_idx + ON public.user_update_events (user_id, message_box_id, pts) + WHERE (event_type)::text = 'new_message'::text + AND message_box_id IS NOT NULL; diff --git a/internal/app/updates/service.go b/internal/app/updates/service.go index d05af621..0b0a7c21 100644 --- a/internal/app/updates/service.go +++ b/internal/app/updates/service.go @@ -22,6 +22,10 @@ type dispatchingEventAppender interface { AppendAllocatedWithDispatch(ctx context.Context, userID int64, event domain.UpdateEvent, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, error) } +type newMessageEventFinder interface { + FindNewMessageEvent(ctx context.Context, userID int64, messageBoxID int) (domain.UpdateEvent, bool, error) +} + // ServiceOption 调整 updates 服务的运行时依赖。 type ServiceOption func(*Service) @@ -259,6 +263,37 @@ func (s *Service) RecordNewMessage(ctx context.Context, authKeyID [8]byte, userI }, false, 0) } +// PublishNewMessage appends an account-visible message update and enqueues +// online dispatch without acknowledging any device-local update state. +func (s *Service) PublishNewMessage(ctx context.Context, userID int64, msg domain.Message) (domain.UpdateEvent, domain.UpdateState, error) { + if userID == 0 { + userID = msg.OwnerUserID + } + if finder, ok := s.events.(newMessageEventFinder); ok && msg.ID > 0 { + event, found, err := finder.FindNewMessageEvent(ctx, userID, msg.ID) + if err != nil { + return domain.UpdateEvent{}, domain.UpdateState{}, err + } + if found { + st, err := s.currentState(ctx, userID) + if err != nil { + return domain.UpdateEvent{}, domain.UpdateState{}, err + } + return event, st, nil + } + } + date := msg.Date + if date == 0 { + date = int(time.Now().Unix()) + } + return s.recordEventCore(ctx, [8]byte{}, userID, domain.UpdateEvent{ + Type: domain.UpdateEventNewMessage, + Date: date, + Message: msg, + PtsCount: 1, + }, true, 0, false) +} + // RecordMessageReactions records a durable marker for message reaction changes. // // updateMessageReactions has no pts fields in Layer 225, but TDesktop still diff --git a/internal/app/updates/service_test.go b/internal/app/updates/service_test.go index e7bcb66c..ad118dea 100644 --- a/internal/app/updates/service_test.go +++ b/internal/app/updates/service_test.go @@ -48,6 +48,45 @@ func TestRecordNewMessageFeedsGetDifference(t *testing.T) { } } +func TestPublishNewMessageIsIdempotentByMessageBoxID(t *testing.T) { + ctx := context.Background() + var authKeyID [8]byte + authKeyID[0] = 11 + events := memory.NewUpdateEventStore() + svc := NewService(memory.NewUpdateStateStore(), events) + msg := domain.Message{ + ID: 10, + OwnerUserID: 1000000001, + Peer: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID}, + From: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID}, + Date: 1700000000, + Body: "Login code: 12345", + } + + firstEvent, firstState, err := svc.PublishNewMessage(ctx, msg.OwnerUserID, msg) + if err != nil { + t.Fatalf("PublishNewMessage first: %v", err) + } + secondEvent, secondState, err := svc.PublishNewMessage(ctx, msg.OwnerUserID, msg) + if err != nil { + t.Fatalf("PublishNewMessage retry: %v", err) + } + if firstEvent.Pts != 1 || firstState.Pts != 1 { + t.Fatalf("first event/state = %+v / %+v, want pts=1", firstEvent, firstState) + } + if secondEvent.Pts != firstEvent.Pts || secondState.Pts != firstState.Pts { + t.Fatalf("retry event/state = %+v / %+v, want same pts as first %+v / %+v", secondEvent, secondState, firstEvent, firstState) + } + + diff, err := svc.GetDifference(ctx, authKeyID, msg.OwnerUserID, domain.UpdateState{}) + if err != nil { + t.Fatalf("GetDifference: %v", err) + } + if diff.State.Pts != 1 || len(diff.Events) != 1 || diff.Events[0].Message.ID != msg.ID { + t.Fatalf("diff = %+v, want one durable login message event", diff) + } +} + func TestRecordReadHistoryFeedsGetDifference(t *testing.T) { ctx := context.Background() var authKeyID [8]byte diff --git a/internal/domain/bootstrap_update.go b/internal/domain/bootstrap_update.go new file mode 100644 index 00000000..12f1732f --- /dev/null +++ b/internal/domain/bootstrap_update.go @@ -0,0 +1,39 @@ +package domain + +import "time" + +type BootstrapUpdateJobKind string + +const ( + BootstrapUpdateJobLoginMessage BootstrapUpdateJobKind = "login_message" +) + +type BootstrapUpdateJobStatus string + +const ( + BootstrapUpdateJobPending BootstrapUpdateJobStatus = "pending" + BootstrapUpdateJobReady BootstrapUpdateJobStatus = "ready" + BootstrapUpdateJobPublishing BootstrapUpdateJobStatus = "publishing" + BootstrapUpdateJobPublished BootstrapUpdateJobStatus = "published" + BootstrapUpdateJobFailed BootstrapUpdateJobStatus = "failed" +) + +const BootstrapUpdateMaxAttempts = 5 + +// BootstrapUpdateJob defers first-session account updates until the client has +// received its initial updates baseline. Pending jobs do not occupy pts. +type BootstrapUpdateJob struct { + ID int64 + Kind BootstrapUpdateJobKind + UserID int64 + AuthKeyID [8]byte + SessionID int64 + MessageBoxID int + Status BootstrapUpdateJobStatus + Attempts int + LastError string + CreatedAt time.Time + UpdatedAt time.Time + ReadyAt time.Time + PublishedAt time.Time +} diff --git a/internal/rpc/auth.go b/internal/rpc/auth.go index 22858c13..1ce2ac65 100644 --- a/internal/rpc/auth.go +++ b/internal/rpc/auth.go @@ -20,8 +20,6 @@ import ( // devCodeLength 是开发固定验证码长度，写入 auth.sentCode 的 type.length。 const devCodeLength = 5 -const loginMessagePushDelay = 2 * time.Second - // registerAuth 注册 auth.* RPC handler。 func (r *Router) registerAuth(d *tg.ServerDispatcher) { d.OnAuthBindTempAuthKey(r.onAuthBindTempAuthKey) @@ -329,7 +327,7 @@ func (r *Router) onAuthSignIn(ctx context.Context, req *tg.AuthSignInRequest) (t r.setAuthUserCache(id, u.ID, true) } r.bindSessionUser(ctx, u.ID) - r.recordAndScheduleLoginMessagePush(ctx, loginMessage) + r.enqueueLoginMessageBootstrap(ctx, loginMessage) r.pushSignInServiceNotificationToOthers(ctx, u) return &tg.AuthAuthorization{User: r.tgSelfUser(u)}, nil } @@ -598,7 +596,7 @@ func (r *Router) onAuthSignUp(ctx context.Context, req *tg.AuthSignUpRequest) (t r.setAuthUserCache(id, u.ID, true) } r.bindSessionUser(ctx, u.ID) - r.recordAndScheduleLoginMessagePush(ctx, loginMessage) + r.enqueueLoginMessageBootstrap(ctx, loginMessage) return &tg.AuthAuthorization{User: r.tgSelfUser(u)}, nil } @@ -734,84 +732,6 @@ func (r *Router) pushSignInServiceNotificationToOthers(ctx context.Context, u do }() } -func (r *Router) recordAndScheduleLoginMessagePush(ctx context.Context, msg domain.Message) { - authKeyID, hasAuthKeyID := AuthKeyIDFrom(ctx) - sessionID, hasSessionID := SessionIDFrom(ctx) - if !hasAuthKeyID || !hasSessionID || msg.ID == 0 { - return - } - event := domain.UpdateEvent{Type: domain.UpdateEventNewMessage, Pts: 1, PtsCount: 1, Date: msg.Date, Message: msg} - state := domain.UpdateState{Pts: 1, Date: msg.Date, Seq: 0} - if r.deps.Updates != nil { - recorded, st, err := r.deps.Updates.RecordNewMessage(ctx, authKeyID, msg.OwnerUserID, msg) - if err != nil { - r.log.Warn("record login message update", zap.Error(err)) - return - } - event = recorded - state = st - } - if r.deps.Sessions == nil { - return - } - // 提前从请求 ctx 取出 rawAuthKeyID（值类型），闭包只捕获该值、不捕获请求 ctx—— - // 避免延迟推送的 AfterFunc 在 loginMessagePushDelay 期间延长请求 ctx 链路的存活。 - rawAuthKeyID, _ := RawAuthKeyIDFrom(ctx) - time.AfterFunc(loginMessagePushDelay, func() { - pushCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - r.pushLoginMessage(pushCtx, rawAuthKeyID, sessionID, event, state) - }) -} - -func (r *Router) pushLoginMessage(ctx context.Context, rawAuthKeyID [8]byte, sessionID int64, event domain.UpdateEvent, state domain.UpdateState) { - if r.deps.Sessions == nil || event.Message.ID == 0 { - return - } - updates := tgLoginMessageUpdates(event, state) - if updates == nil { - return - } - var err error - if scoped, ok := r.scopedSessions(); ok && rawAuthKeyID != ([8]byte{}) { - err = scoped.PushToSessionForAuthKey(ctx, rawAuthKeyID, sessionID, proto.MessageFromServer, updates) - } else { - err = r.deps.Sessions.PushToSession(ctx, sessionID, proto.MessageFromServer, updates) - } - if err != nil { - r.log.Debug("push login message", zap.Int64("session_id", sessionID), zap.Error(err)) - return - } - r.log.Debug("pushed login message", - zap.Int64("session_id", sessionID), - zap.Int("message_id", event.Message.ID), - zap.Int("pts", event.Pts), - zap.Int("seq", state.Seq), - ) -} - -func tgLoginMessageUpdates(event domain.UpdateEvent, state domain.UpdateState) *tg.Updates { - item := tgMessage(event.Message) - if item == nil { - return nil - } - if state.Date == 0 { - state.Date = event.Date - } - return &tg.Updates{ - Updates: []tg.UpdateClass{ - &tg.UpdateNewMessage{ - Message: item, - Pts: event.Pts, - PtsCount: event.PtsCount, - }, - }, - Users: []tg.UserClass{tgUser(domain.OfficialSystemUser())}, - Date: state.Date, - Seq: state.Seq, - } -} - func (r *Router) tgSignInServiceNotification(ctx context.Context, u domain.User, authKeyID [8]byte) *tg.Updates { now := r.clock.Now() client := "Unknown device" diff --git a/internal/rpc/bootstrap_updates.go b/internal/rpc/bootstrap_updates.go new file mode 100644 index 00000000..91ae2f9b --- /dev/null +++ b/internal/rpc/bootstrap_updates.go @@ -0,0 +1,185 @@ +package rpc + +import ( + "context" + "fmt" + "time" + + "go.uber.org/zap" + + "telesrv/internal/domain" + "telesrv/internal/postresponse" +) + +const ( + defaultBootstrapBatch = 100 + defaultBootstrapInterval = 200 * time.Millisecond + defaultBootstrapLease = 30 * time.Second +) + +type BootstrapUpdateDispatcher struct { + router *Router + log *zap.Logger + batch int + interval time.Duration + maxIdleInterval time.Duration + leaseTimeout time.Duration +} + +type BootstrapUpdateOption func(*BootstrapUpdateDispatcher) + +func WithBootstrapUpdateBatch(n int) BootstrapUpdateOption { + return func(d *BootstrapUpdateDispatcher) { + if n > 0 { + d.batch = n + } + } +} + +func WithBootstrapUpdateInterval(interval time.Duration) BootstrapUpdateOption { + return func(d *BootstrapUpdateDispatcher) { + if interval > 0 { + d.interval = interval + } + } +} + +func WithBootstrapUpdateLease(timeout time.Duration) BootstrapUpdateOption { + return func(d *BootstrapUpdateDispatcher) { + if timeout > 0 { + d.leaseTimeout = timeout + } + } +} + +func NewBootstrapUpdateDispatcher(router *Router, log *zap.Logger, opts ...BootstrapUpdateOption) *BootstrapUpdateDispatcher { + if log == nil { + log = zap.NewNop() + } + d := &BootstrapUpdateDispatcher{ + router: router, + log: log, + batch: defaultBootstrapBatch, + interval: defaultBootstrapInterval, + maxIdleInterval: defaultIdleDispatchMaxInterval, + leaseTimeout: defaultBootstrapLease, + } + for _, opt := range opts { + if opt != nil { + opt(d) + } + } + return d +} + +func (d *BootstrapUpdateDispatcher) Run(ctx context.Context) { + if d == nil || d.router == nil { + return + } + runIdleBackoffLoop(ctx, d.interval, d.maxIdleInterval, d.DispatchOnce) +} + +func (d *BootstrapUpdateDispatcher) DispatchOnce(ctx context.Context) bool { + return d.router.publishReadyBootstrapUpdates(ctx, d.batch, d.leaseTimeout, d.log) > 0 +} + +func (r *Router) enqueueLoginMessageBootstrap(ctx context.Context, msg domain.Message) { + if r.deps.BootstrapUpdates == nil || msg.OwnerUserID == 0 || msg.ID == 0 { + return + } + authKeyID, hasAuthKeyID := AuthKeyIDFrom(ctx) + sessionID, hasSessionID := SessionIDFrom(ctx) + if !hasAuthKeyID || !hasSessionID { + return + } + if _, err := r.deps.BootstrapUpdates.EnqueueLoginMessage(ctx, domain.BootstrapUpdateJob{ + Kind: domain.BootstrapUpdateJobLoginMessage, + UserID: msg.OwnerUserID, + AuthKeyID: authKeyID, + SessionID: sessionID, + MessageBoxID: msg.ID, + Status: domain.BootstrapUpdateJobPending, + }); err != nil { + r.log.Warn("enqueue bootstrap login message", zap.Int64("user_id", msg.OwnerUserID), zap.Int("message_id", msg.ID), zap.Error(err)) + } +} + +func (r *Router) registerBootstrapAfterBaseline(ctx context.Context, userID int64) { + if r.deps.BootstrapUpdates == nil || userID == 0 { + return + } + authKeyID, hasAuthKeyID := AuthKeyIDFrom(ctx) + sessionID, hasSessionID := SessionIDFrom(ctx) + if !hasAuthKeyID || !hasSessionID { + return + } + postresponse.Register(ctx, func() { + cbCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + ready, err := r.deps.BootstrapUpdates.MarkReadyForSession(cbCtx, userID, authKeyID, sessionID) + if err != nil { + r.log.Warn("mark bootstrap updates ready", zap.Int64("user_id", userID), zap.Int64("session_id", sessionID), zap.Error(err)) + return + } + if ready == 0 { + return + } + r.publishReadyBootstrapUpdates(cbCtx, ready, defaultBootstrapLease, r.log.Named("bootstrap")) + }) +} + +func (r *Router) publishReadyBootstrapUpdates(ctx context.Context, batch int, leaseTimeout time.Duration, log *zap.Logger) int { + if r.deps.BootstrapUpdates == nil || r.deps.Updates == nil || r.deps.Messages == nil { + return 0 + } + if log == nil { + log = zap.NewNop() + } + jobs, err := r.deps.BootstrapUpdates.ClaimReady(ctx, batch, leaseTimeout) + if err != nil { + log.Warn("claim bootstrap updates", zap.Error(err)) + return 0 + } + for _, job := range jobs { + if err := r.publishBootstrapUpdateJob(ctx, job); err != nil { + log.Warn("publish bootstrap update", + zap.Int64("job_id", job.ID), + zap.Int64("user_id", job.UserID), + zap.String("kind", string(job.Kind)), + zap.Error(err), + ) + _ = r.deps.BootstrapUpdates.MarkFailed(ctx, job.ID, err.Error()) + } + } + return len(jobs) +} + +func (r *Router) publishBootstrapUpdateJob(ctx context.Context, job domain.BootstrapUpdateJob) error { + switch job.Kind { + case domain.BootstrapUpdateJobLoginMessage: + return r.publishBootstrapLoginMessage(ctx, job) + default: + return fmt.Errorf("unknown bootstrap update kind %q", job.Kind) + } +} + +func (r *Router) publishBootstrapLoginMessage(ctx context.Context, job domain.BootstrapUpdateJob) error { + list, err := r.deps.Messages.GetMessages(ctx, job.UserID, []int{job.MessageBoxID}) + if err != nil { + return fmt.Errorf("load bootstrap login message: %w", err) + } + if len(list.Messages) == 0 { + return fmt.Errorf("bootstrap login message missing: box_id=%d", job.MessageBoxID) + } + msg := list.Messages[0] + if msg.OwnerUserID != job.UserID || msg.ID != job.MessageBoxID { + return fmt.Errorf("bootstrap login message mismatch: owner=%d id=%d", msg.OwnerUserID, msg.ID) + } + if _, _, err := r.deps.Updates.PublishNewMessage(ctx, job.UserID, msg); err != nil { + return fmt.Errorf("publish bootstrap login message update: %w", err) + } + if err := r.deps.BootstrapUpdates.MarkPublished(ctx, job.ID); err != nil { + return err + } + return nil +} diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index 0ab9334e..b3406169 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -315,6 +315,7 @@ type UpdatesService interface { GetDifference(ctx context.Context, authKeyID [8]byte, userID int64, from domain.UpdateState) (domain.UpdateDifference, error) ClearAuthKey(ctx context.Context, authKeyID [8]byte) error RecordNewMessage(ctx context.Context, authKeyID [8]byte, userID int64, msg domain.Message) (domain.UpdateEvent, domain.UpdateState, error) + PublishNewMessage(ctx context.Context, userID int64, msg domain.Message) (domain.UpdateEvent, domain.UpdateState, error) RecordStory(ctx context.Context, authKeyID [8]byte, userID int64, story domain.Story, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) RecordStoryFanout(ctx context.Context, userID int64, story domain.Story) (domain.UpdateEvent, domain.UpdateState, error) RecordReadStories(ctx context.Context, authKeyID [8]byte, userID int64, read domain.StoryReadResult, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) @@ -674,35 +675,36 @@ type AIComposeService interface { // Deps 按业务域注入服务接口。各域的 handler 注册见对应文件（auth.go / users.go / updates.go）。 type Deps struct { - Auth AuthService - Account AccountService - Privacy PrivacyService - Help HelpService - AICompose AIComposeService - Users UsersService - Updates UpdatesService - Contacts ContactsService - Dialogs DialogsService - Messages MessagesService - Stories StoriesService - Channels ChannelsService - Files FilesService - Bots BotsService - Polls PollsService - Phone PhoneService - GroupCalls GroupCallsService - SFU sfu.Service - TURN turnsrv.Service - LangPack LangPackService - Sessions SessionBinder - Inline store.InlineRegistryStore - Limiter RateLimiter - Metrics Metrics - SecretChats SecretChatService - Stars StarsService - Gifts GiftsService - Passkey PasskeyService - Themes ThemeService + Auth AuthService + Account AccountService + Privacy PrivacyService + Help HelpService + AICompose AIComposeService + Users UsersService + Updates UpdatesService + BootstrapUpdates store.BootstrapUpdateJobStore + Contacts ContactsService + Dialogs DialogsService + Messages MessagesService + Stories StoriesService + Channels ChannelsService + Files FilesService + Bots BotsService + Polls PollsService + Phone PhoneService + GroupCalls GroupCallsService + SFU sfu.Service + TURN turnsrv.Service + LangPack LangPackService + Sessions SessionBinder + Inline store.InlineRegistryStore + Limiter RateLimiter + Metrics Metrics + SecretChats SecretChatService + Stars StarsService + Gifts GiftsService + Passkey PasskeyService + Themes ThemeService } // ThemeService 抽象自定义云主题(app/themes):创建/更新/查询主题 + 维护每用户已安装列表。 diff --git a/internal/rpc/rpc_testkit_updates_test.go b/internal/rpc/rpc_testkit_updates_test.go index 0b8eab06..8c7e7eb5 100644 --- a/internal/rpc/rpc_testkit_updates_test.go +++ b/internal/rpc/rpc_testkit_updates_test.go @@ -71,6 +71,19 @@ func (s *captureUpdates) RecordNewMessage(_ context.Context, authKeyID [8]byte, return event, s.state, nil } +func (s *captureUpdates) PublishNewMessage(_ context.Context, userID int64, msg domain.Message) (domain.UpdateEvent, domain.UpdateState, error) { + s.userID = userID + s.date = msg.Date + event := domain.UpdateEvent{Type: domain.UpdateEventNewMessage, Pts: s.state.Pts + 1, PtsCount: 1, Date: msg.Date, Message: msg} + s.events = append(s.events, event) + st := s.state + st.Pts = event.Pts + if st.Date == 0 { + st.Date = msg.Date + } + return event, st, nil +} + func (s *captureUpdates) RecordStory(_ context.Context, authKeyID [8]byte, userID int64, story domain.Story, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) { s.excludeSessionID = excludeSessionID return s.recordCapturedEvent(authKeyID, userID, domain.UpdateEvent{ diff --git a/internal/rpc/updates.go b/internal/rpc/updates.go index 1d397e38..b9c18530 100644 --- a/internal/rpc/updates.go +++ b/internal/rpc/updates.go @@ -34,6 +34,7 @@ func (r *Router) onUpdatesGetState(ctx context.Context) (*tg.UpdatesState, error return nil, internalErr() } r.markSessionReceivesUpdates(ctx, userID) + r.registerBootstrapAfterBaseline(ctx, userID) // 密聊 qts 是设备级、独立于账号级 pts 引擎：注入当前设备已分配的 qts（无密聊设备为 0）。 out := tgUpdateState(st) out.Qts = r.deviceEncryptedQts(ctx) @@ -76,6 +77,7 @@ func (r *Router) onUpdatesGetDifference(ctx context.Context, req *tg.UpdatesGetD // 密聊握手/已读状态事件（无 qts）：按未投递标记补回 OtherUpdates。 stateUpdates, statePeerUserIDs, stateEventIDs := r.encryptedStateUpdates(ctx, userID) if len(st.Events) == 0 && len(st.ChannelNudges) == 0 && len(encMsgs) == 0 && len(stateUpdates) == 0 { + r.registerBootstrapAfterBaseline(ctx, userID) return &tg.UpdatesDifferenceEmpty{Date: st.State.Date, Seq: st.State.Seq}, nil } st.Events = r.enrichUpdateEvents(ctx, userID, st.Events) @@ -87,6 +89,7 @@ func (r *Router) onUpdatesGetDifference(ctx context.Context, req *tg.UpdatesGetD _ = r.deps.SecretChats.MarkStateEventsDelivered(ctx, deviceKey, stateEventIDs) } } + r.registerBootstrapAfterBaseline(ctx, userID) return diff, nil } diff --git a/internal/rpc/updates_rpc_test.go b/internal/rpc/updates_rpc_test.go index 963f614f..42a69c0d 100644 --- a/internal/rpc/updates_rpc_test.go +++ b/internal/rpc/updates_rpc_test.go @@ -7,19 +7,31 @@ import ( "time" "github.com/gotd/td/clock" - "github.com/gotd/td/proto" "github.com/gotd/td/tg" "go.uber.org/zap/zaptest" appchannels "telesrv/internal/app/channels" appupdates "telesrv/internal/app/updates" "telesrv/internal/domain" + "telesrv/internal/postresponse" "telesrv/internal/store/memory" ) -func TestPushLoginMessageSendsUpdateNewMessage(t *testing.T) { - sessions := &captureSessions{} - r := New(Config{}, Deps{Sessions: sessions}, zaptest.NewLogger(t), clock.System) +func TestBootstrapLoginMessagePublishesNewMessageAfterReady(t *testing.T) { + bootstrap := memory.NewBootstrapUpdateJobStore() + updates := &captureUpdates{state: domain.UpdateState{Pts: 3, Date: 1700000000}} + messages := &captureMessages{ + list: domain.MessageList{Messages: []domain.Message{{ + ID: 99, + OwnerUserID: 1000000001, + Peer: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID}, + From: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID}, + Date: 1700000100, + Body: "Login code: 12345", + Entities: []domain.MessageEntity{{Type: domain.MessageEntityBold, Offset: 0, Length: 11}}, + }}}, + } + r := New(Config{}, Deps{BootstrapUpdates: bootstrap, Updates: updates, Messages: messages}, zaptest.NewLogger(t), clock.System) msg := domain.Message{ ID: 99, OwnerUserID: 1000000001, @@ -29,39 +41,78 @@ func TestPushLoginMessageSendsUpdateNewMessage(t *testing.T) { Body: "Login code: 12345", Entities: []domain.MessageEntity{{Type: domain.MessageEntityBold, Offset: 0, Length: 11}}, } - event := domain.UpdateEvent{Type: domain.UpdateEventNewMessage, Pts: 4, PtsCount: 1, Date: msg.Date, Message: msg} - state := domain.UpdateState{Pts: 4, Date: 1700000100, Seq: 2} + authKeyID := [8]byte{1, 2, 3} + sessionID := int64(55) + ctx := WithSessionID(WithAuthKeyID(context.Background(), authKeyID), sessionID) + r.enqueueLoginMessageBootstrap(ctx, msg) + claimed := r.publishReadyBootstrapUpdates(context.Background(), 10, time.Second, zaptest.NewLogger(t)) + if claimed != 0 || len(updates.events) != 0 { + t.Fatalf("published before ready = claimed %d events %d, want none", claimed, len(updates.events)) + } - r.pushLoginMessage(context.Background(), [8]byte{}, 55, event, state) + ready, err := bootstrap.MarkReadyForSession(context.Background(), msg.OwnerUserID, authKeyID, sessionID) + if err != nil { + t.Fatalf("mark ready: %v", err) + } + if ready != 1 { + t.Fatalf("ready jobs = %d, want 1", ready) + } + claimed = r.publishReadyBootstrapUpdates(context.Background(), 10, time.Second, zaptest.NewLogger(t)) + if claimed != 1 { + t.Fatalf("claimed jobs = %d, want 1", claimed) + } + if len(updates.events) != 1 { + t.Fatalf("published events = %d, want 1", len(updates.events)) + } + event := updates.events[0] + if event.Type != domain.UpdateEventNewMessage || event.Message.ID != msg.ID || event.Pts != 4 { + t.Fatalf("event = %+v, want login message pts 4", event) + } +} - gotSession := sessions.snapshot() - if gotSession.sessionID != 55 || gotSession.messageType != proto.MessageFromServer { - t.Fatalf("push target = session %d type %v, want session 55 MessageFromServer", gotSession.sessionID, gotSession.messageType) +func TestUpdatesGetStatePublishesBootstrapAfterRPCResult(t *testing.T) { + bootstrap := memory.NewBootstrapUpdateJobStore() + updates := &captureUpdates{state: domain.UpdateState{Pts: 0, Date: 1700000000}} + msg := domain.Message{ + ID: 1, + OwnerUserID: 1780243001, + Peer: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID}, + From: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID}, + Date: 1700000100, + Body: "Login code: 12345", } - got, ok := gotSession.message.(*tg.Updates) - if !ok { - t.Fatalf("pushed message = %T, want *tg.Updates", gotSession.message) + messages := &captureMessages{list: domain.MessageList{Messages: []domain.Message{msg}}} + r := New(Config{}, Deps{BootstrapUpdates: bootstrap, Updates: updates, Messages: messages}, zaptest.NewLogger(t), clock.System) + authKeyID := [8]byte{9, 8, 7} + sessionID := int64(5723482677041206318) + ctx := postresponse.WithCallbacks( + WithUserID( + WithSessionID(WithAuthKeyID(context.Background(), authKeyID), sessionID), + msg.OwnerUserID, + ), + ) + r.enqueueLoginMessageBootstrap(ctx, msg) + + state, err := r.onUpdatesGetState(ctx) + if err != nil { + t.Fatalf("getState: %v", err) } - if len(got.Updates) != 1 || len(got.Users) != 1 { - t.Fatalf("updates payload = %+v, want one update and official user", got) + if state.Pts != 0 { + t.Fatalf("state pts = %d, want 0 before bootstrap publish", state.Pts) } - update, ok := got.Updates[0].(*tg.UpdateNewMessage) - if !ok { - t.Fatalf("update = %T, want *tg.UpdateNewMessage", got.Updates[0]) + if claimed := r.publishReadyBootstrapUpdates(context.Background(), 10, time.Second, zaptest.NewLogger(t)); claimed != 0 { + t.Fatalf("bootstrap published before post-response callback: %d", claimed) } - if update.Pts != event.Pts || update.PtsCount != event.PtsCount || got.Seq != state.Seq { - t.Fatalf("update state = pts %d pts_count %d seq %d, want %d/%d/%d", update.Pts, update.PtsCount, got.Seq, event.Pts, event.PtsCount, state.Seq) + if len(updates.events) != 0 { + t.Fatalf("events before post-response = %d, want 0", len(updates.events)) } - message, ok := update.Message.(*tg.Message) - if !ok { - t.Fatalf("update message = %T, want *tg.Message", update.Message) + + postresponse.Run(ctx) + if len(updates.events) != 1 { + t.Fatalf("events after post-response = %d, want 1", len(updates.events)) } - if message.ID != msg.ID || message.Message != msg.Body { - t.Fatalf("message = %+v, want login message %+v", message, msg) - } - user, ok := got.Users[0].(*tg.User) - if !ok || user.ID != domain.OfficialSystemUserID || !user.Verified || !user.Support { - t.Fatalf("user = %#v, want verified official system user", got.Users[0]) + if got := updates.events[0]; got.Type != domain.UpdateEventNewMessage || got.Message.ID != msg.ID { + t.Fatalf("event after post-response = %+v, want login message", got) } } diff --git a/internal/store/bootstrap_update_job.go b/internal/store/bootstrap_update_job.go new file mode 100644 index 00000000..f6693a5a --- /dev/null +++ b/internal/store/bootstrap_update_job.go @@ -0,0 +1,16 @@ +package store + +import ( + "context" + "time" + + "telesrv/internal/domain" +) + +type BootstrapUpdateJobStore interface { + EnqueueLoginMessage(ctx context.Context, job domain.BootstrapUpdateJob) (domain.BootstrapUpdateJob, error) + MarkReadyForSession(ctx context.Context, userID int64, authKeyID [8]byte, sessionID int64) (int, error) + ClaimReady(ctx context.Context, limit int, leaseTimeout time.Duration) ([]domain.BootstrapUpdateJob, error) + MarkPublished(ctx context.Context, id int64) error + MarkFailed(ctx context.Context, id int64, lastError string) error +} diff --git a/internal/store/memory/bootstrap_update_job.go b/internal/store/memory/bootstrap_update_job.go new file mode 100644 index 00000000..7132b7df --- /dev/null +++ b/internal/store/memory/bootstrap_update_job.go @@ -0,0 +1,144 @@ +package memory + +import ( + "context" + "sync" + "time" + + "telesrv/internal/domain" +) + +type BootstrapUpdateJobStore struct { + mu sync.Mutex + nextID int64 + jobs map[int64]domain.BootstrapUpdateJob + uniq map[bootstrapUpdateJobKey]int64 +} + +type bootstrapUpdateJobKey struct { + kind domain.BootstrapUpdateJobKind + userID int64 + messageBoxID int +} + +func NewBootstrapUpdateJobStore() *BootstrapUpdateJobStore { + return &BootstrapUpdateJobStore{ + nextID: 1, + jobs: make(map[int64]domain.BootstrapUpdateJob), + uniq: make(map[bootstrapUpdateJobKey]int64), + } +} + +func (s *BootstrapUpdateJobStore) EnqueueLoginMessage(_ context.Context, job domain.BootstrapUpdateJob) (domain.BootstrapUpdateJob, error) { + if job.Kind == "" { + job.Kind = domain.BootstrapUpdateJobLoginMessage + } + if job.Status == "" { + job.Status = domain.BootstrapUpdateJobPending + } + now := time.Now() + s.mu.Lock() + defer s.mu.Unlock() + key := bootstrapUpdateJobKey{kind: job.Kind, userID: job.UserID, messageBoxID: job.MessageBoxID} + if id, ok := s.uniq[key]; ok { + existing := s.jobs[id] + if existing.Status == domain.BootstrapUpdateJobFailed { + existing.Status = domain.BootstrapUpdateJobPending + existing.LastError = "" + } + existing.AuthKeyID = job.AuthKeyID + existing.SessionID = job.SessionID + existing.UpdatedAt = now + s.jobs[id] = existing + return existing, nil + } + job.ID = s.nextID + s.nextID++ + job.CreatedAt = now + job.UpdatedAt = now + s.jobs[job.ID] = job + s.uniq[key] = job.ID + return job, nil +} + +func (s *BootstrapUpdateJobStore) MarkReadyForSession(_ context.Context, userID int64, authKeyID [8]byte, sessionID int64) (int, error) { + now := time.Now() + count := 0 + s.mu.Lock() + defer s.mu.Unlock() + for id, job := range s.jobs { + if job.UserID != userID || job.AuthKeyID != authKeyID || job.SessionID != sessionID || job.Status != domain.BootstrapUpdateJobPending { + continue + } + job.Status = domain.BootstrapUpdateJobReady + job.ReadyAt = now + job.UpdatedAt = now + s.jobs[id] = job + count++ + } + return count, nil +} + +func (s *BootstrapUpdateJobStore) ClaimReady(_ context.Context, limit int, leaseTimeout time.Duration) ([]domain.BootstrapUpdateJob, error) { + if limit <= 0 { + limit = 100 + } + if leaseTimeout <= 0 { + leaseTimeout = 30 * time.Second + } + now := time.Now() + out := make([]domain.BootstrapUpdateJob, 0, limit) + s.mu.Lock() + defer s.mu.Unlock() + for id, job := range s.jobs { + if len(out) >= limit { + break + } + ready := job.Status == domain.BootstrapUpdateJobReady && (job.ReadyAt.IsZero() || !job.ReadyAt.After(now)) + stalePublishing := job.Status == domain.BootstrapUpdateJobPublishing && now.Sub(job.UpdatedAt) > leaseTimeout + if !ready && !stalePublishing { + continue + } + job.Status = domain.BootstrapUpdateJobPublishing + job.Attempts++ + job.UpdatedAt = now + s.jobs[id] = job + out = append(out, job) + } + return out, nil +} + +func (s *BootstrapUpdateJobStore) MarkPublished(_ context.Context, id int64) error { + now := time.Now() + s.mu.Lock() + if job, ok := s.jobs[id]; ok { + job.Status = domain.BootstrapUpdateJobPublished + job.PublishedAt = now + job.UpdatedAt = now + s.jobs[id] = job + } + s.mu.Unlock() + return nil +} + +func (s *BootstrapUpdateJobStore) MarkFailed(_ context.Context, id int64, lastError string) error { + now := time.Now() + s.mu.Lock() + if job, ok := s.jobs[id]; ok { + if job.Attempts >= domain.BootstrapUpdateMaxAttempts { + job.Status = domain.BootstrapUpdateJobFailed + } else { + job.Status = domain.BootstrapUpdateJobReady + delay := time.Duration(job.Attempts*job.Attempts) * time.Second + if delay > time.Minute { + delay = time.Minute + } + job.ReadyAt = now.Add(delay) + } + job.LastError = lastError + job.UpdatedAt = now + s.jobs[id] = job + } + s.mu.Unlock() + return nil +} diff --git a/internal/store/memory/bootstrap_update_job_test.go b/internal/store/memory/bootstrap_update_job_test.go new file mode 100644 index 00000000..5380b4e9 --- /dev/null +++ b/internal/store/memory/bootstrap_update_job_test.go @@ -0,0 +1,58 @@ +package memory + +import ( + "context" + "testing" + "time" + + "telesrv/internal/domain" +) + +func TestBootstrapUpdateJobRetriesBeforeFailed(t *testing.T) { + ctx := context.Background() + store := NewBootstrapUpdateJobStore() + job, err := store.EnqueueLoginMessage(ctx, domain.BootstrapUpdateJob{ + Kind: domain.BootstrapUpdateJobLoginMessage, + UserID: 1000000001, + AuthKeyID: [8]byte{1, 2, 3}, + SessionID: 77, + MessageBoxID: 10, + }) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + if ready, err := store.MarkReadyForSession(ctx, job.UserID, job.AuthKeyID, job.SessionID); err != nil || ready != 1 { + t.Fatalf("mark ready = %d, %v; want 1 nil", ready, err) + } + claimed, err := store.ClaimReady(ctx, 10, time.Second) + if err != nil { + t.Fatalf("claim ready: %v", err) + } + if len(claimed) != 1 || claimed[0].Attempts != 1 { + t.Fatalf("claimed = %+v, want one first attempt", claimed) + } + if err := store.MarkFailed(ctx, claimed[0].ID, "temporary"); err != nil { + t.Fatalf("mark failed: %v", err) + } + retry, err := store.ClaimReady(ctx, 10, time.Second) + if err != nil { + t.Fatalf("claim retry before backoff: %v", err) + } + if len(retry) != 0 { + t.Fatalf("retry before backoff = %+v, want none", retry) + } + + store.mu.Lock() + job = store.jobs[claimed[0].ID] + job.ReadyAt = time.Now().Add(-time.Second) + store.jobs[claimed[0].ID] = job + store.mu.Unlock() + + retry, err = store.ClaimReady(ctx, 10, time.Second) + if err != nil { + t.Fatalf("claim retry after backoff: %v", err) + } + if len(retry) != 1 || retry[0].Attempts != 2 { + t.Fatalf("retry after backoff = %+v, want second attempt", retry) + } +} diff --git a/internal/store/memory/updates.go b/internal/store/memory/updates.go index a348142e..9465e34d 100644 --- a/internal/store/memory/updates.go +++ b/internal/store/memory/updates.go @@ -46,15 +46,7 @@ func (s *UpdateEventStore) append(userID int64, event domain.UpdateEvent, alloca event.PtsCount = 1 } event.UserID = userID - event.Message = cloneMessage(event.Message) - event.Story = cloneUpdateStory(event.Story) - event.MessageIDs = append([]int(nil), event.MessageIDs...) - event.Peers = append([]domain.Peer(nil), event.Peers...) - event.Users = append([]domain.User(nil), event.Users...) - event.Channels = append([]domain.Channel(nil), event.Channels...) - event.Reaction = cloneUpdateReaction(event.Reaction) - event.QuickReplies = cloneUpdateQuickReplies(event.QuickReplies) - event.QuickReplyMessage = cloneUpdateQuickReplyMessage(event.QuickReplyMessage) + event = cloneUpdateEvent(event) s.mu.Lock() if allocate { current := 0 @@ -79,16 +71,7 @@ func (s *UpdateEventStore) ListAfter(_ context.Context, userID int64, pts, limit if event.Pts <= pts { continue } - event.Message = cloneMessage(event.Message) - event.Story = cloneUpdateStory(event.Story) - event.MessageIDs = append([]int(nil), event.MessageIDs...) - event.Peers = append([]domain.Peer(nil), event.Peers...) - event.Users = append([]domain.User(nil), event.Users...) - event.Channels = append([]domain.Channel(nil), event.Channels...) - event.Reaction = cloneUpdateReaction(event.Reaction) - event.QuickReplies = cloneUpdateQuickReplies(event.QuickReplies) - event.QuickReplyMessage = cloneUpdateQuickReplyMessage(event.QuickReplyMessage) - out = append(out, event) + out = append(out, cloneUpdateEvent(event)) if limit > 0 && len(out) >= limit { break } @@ -96,6 +79,33 @@ func (s *UpdateEventStore) ListAfter(_ context.Context, userID int64, pts, limit return out, nil } +func (s *UpdateEventStore) FindNewMessageEvent(_ context.Context, userID int64, messageBoxID int) (domain.UpdateEvent, bool, error) { + if userID == 0 || messageBoxID <= 0 { + return domain.UpdateEvent{}, false, nil + } + s.mu.RLock() + defer s.mu.RUnlock() + for _, event := range s.events[userID] { + if event.Type == domain.UpdateEventNewMessage && event.Message.ID == messageBoxID { + return cloneUpdateEvent(event), true, nil + } + } + return domain.UpdateEvent{}, false, nil +} + +func cloneUpdateEvent(event domain.UpdateEvent) domain.UpdateEvent { + event.Message = cloneMessage(event.Message) + event.Story = cloneUpdateStory(event.Story) + event.MessageIDs = append([]int(nil), event.MessageIDs...) + event.Peers = append([]domain.Peer(nil), event.Peers...) + event.Users = append([]domain.User(nil), event.Users...) + event.Channels = append([]domain.Channel(nil), event.Channels...) + event.Reaction = cloneUpdateReaction(event.Reaction) + event.QuickReplies = cloneUpdateQuickReplies(event.QuickReplies) + event.QuickReplyMessage = cloneUpdateQuickReplyMessage(event.QuickReplyMessage) + return event +} + func cloneUpdateStory(story domain.Story) domain.Story { story.Entities = append([]domain.MessageEntity(nil), story.Entities...) story.Views.Reactions = append([]domain.ChannelMessageReactionCount(nil), story.Views.Reactions...) @@ -122,7 +132,6 @@ func cloneUpdateQuickReplyMessage(in domain.QuickReplyMessage) domain.QuickReply return out } - // MaxContiguousPts 返回从 1 起无空洞的最大 pts（内存版按 pts_count 连续扫描）。 func (s *UpdateEventStore) MaxContiguousPts(_ context.Context, userID int64) (int, error) { s.mu.RLock() diff --git a/internal/store/postgres/bootstrap_update_job.go b/internal/store/postgres/bootstrap_update_job.go new file mode 100644 index 00000000..173b44d8 --- /dev/null +++ b/internal/store/postgres/bootstrap_update_job.go @@ -0,0 +1,189 @@ +package postgres + +import ( + "context" + "database/sql" + "fmt" + "time" + + "telesrv/internal/domain" + "telesrv/internal/store/postgres/sqlcgen" +) + +type BootstrapUpdateJobStore struct { + db sqlcgen.DBTX +} + +func NewBootstrapUpdateJobStore(db sqlcgen.DBTX) *BootstrapUpdateJobStore { + return &BootstrapUpdateJobStore{db: db} +} + +func (s *BootstrapUpdateJobStore) EnqueueLoginMessage(ctx context.Context, job domain.BootstrapUpdateJob) (domain.BootstrapUpdateJob, error) { + if job.Kind == "" { + job.Kind = domain.BootstrapUpdateJobLoginMessage + } + if job.Status == "" { + job.Status = domain.BootstrapUpdateJobPending + } + row := s.db.QueryRow(ctx, ` +INSERT INTO bootstrap_update_jobs ( + kind, user_id, auth_key_id, session_id, message_box_id, status +) VALUES ($1, $2, $3, $4, $5, $6) +ON CONFLICT (kind, user_id, message_box_id) DO UPDATE SET + auth_key_id = EXCLUDED.auth_key_id, + session_id = EXCLUDED.session_id, + status = CASE + WHEN bootstrap_update_jobs.status = 'failed' THEN 'pending' + ELSE bootstrap_update_jobs.status + END, + last_error = CASE + WHEN bootstrap_update_jobs.status = 'failed' THEN '' + ELSE bootstrap_update_jobs.last_error + END, + updated_at = now() +RETURNING id, kind, user_id, auth_key_id, session_id, message_box_id, status, attempts, last_error, created_at, updated_at, ready_at, published_at`, + string(job.Kind), + job.UserID, + authKeyIDToInt64(job.AuthKeyID), + job.SessionID, + job.MessageBoxID, + string(job.Status), + ) + out, err := scanBootstrapUpdateJob(row) + if err != nil { + return domain.BootstrapUpdateJob{}, fmt.Errorf("enqueue bootstrap login message: %w", err) + } + return out, nil +} + +func (s *BootstrapUpdateJobStore) MarkReadyForSession(ctx context.Context, userID int64, authKeyID [8]byte, sessionID int64) (int, error) { + tag, err := s.db.Exec(ctx, ` +UPDATE bootstrap_update_jobs +SET status = 'ready', + ready_at = now(), + updated_at = now() +WHERE user_id = $1 + AND auth_key_id = $2 + AND session_id = $3 + AND status = 'pending'`, + userID, authKeyIDToInt64(authKeyID), sessionID) + if err != nil { + return 0, fmt.Errorf("mark bootstrap jobs ready: %w", err) + } + return int(tag.RowsAffected()), nil +} + +func (s *BootstrapUpdateJobStore) ClaimReady(ctx context.Context, limit int, leaseTimeout time.Duration) ([]domain.BootstrapUpdateJob, error) { + if limit <= 0 { + limit = 100 + } + if limit > 1000 { + limit = 1000 + } + leaseSeconds := int(leaseTimeout / time.Second) + if leaseSeconds <= 0 { + leaseSeconds = int(defaultDispatchLease / time.Second) + } + rows, err := s.db.Query(ctx, ` +WITH picked AS ( + SELECT id + FROM bootstrap_update_jobs + WHERE (status = 'ready' AND COALESCE(ready_at, created_at) <= now()) + OR (status = 'publishing' AND updated_at < now() - make_interval(secs => $1::int)) + ORDER BY COALESCE(ready_at, created_at) ASC, user_id ASC, id ASC + LIMIT $2 + FOR UPDATE SKIP LOCKED +) +UPDATE bootstrap_update_jobs j +SET status = 'publishing', + attempts = j.attempts + 1, + updated_at = now() +FROM picked p +WHERE j.id = p.id +RETURNING j.id, j.kind, j.user_id, j.auth_key_id, j.session_id, j.message_box_id, j.status, j.attempts, j.last_error, j.created_at, j.updated_at, j.ready_at, j.published_at`, + leaseSeconds, limit) + if err != nil { + return nil, fmt.Errorf("claim bootstrap jobs: %w", err) + } + defer rows.Close() + out := make([]domain.BootstrapUpdateJob, 0) + for rows.Next() { + job, err := scanBootstrapUpdateJob(rows) + if err != nil { + return nil, err + } + out = append(out, job) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("claim bootstrap jobs rows: %w", err) + } + return out, nil +} + +func (s *BootstrapUpdateJobStore) MarkPublished(ctx context.Context, id int64) error { + if _, err := s.db.Exec(ctx, ` +UPDATE bootstrap_update_jobs +SET status = 'published', + published_at = now(), + updated_at = now() +WHERE id = $1`, id); err != nil { + return fmt.Errorf("mark bootstrap job published: %w", err) + } + return nil +} + +func (s *BootstrapUpdateJobStore) MarkFailed(ctx context.Context, id int64, lastError string) error { + if _, err := s.db.Exec(ctx, ` +UPDATE bootstrap_update_jobs +SET status = CASE WHEN attempts >= $3 THEN 'failed' ELSE 'ready' END, + ready_at = CASE + WHEN attempts >= $3 THEN ready_at + ELSE now() + make_interval(secs => LEAST(60, attempts * attempts)) + END, + last_error = $2, + updated_at = now() +WHERE id = $1`, id, lastError, domain.BootstrapUpdateMaxAttempts); err != nil { + return fmt.Errorf("mark bootstrap job failed: %w", err) + } + return nil +} + +type bootstrapJobScanner interface { + Scan(dest ...any) error +} + +func scanBootstrapUpdateJob(row bootstrapJobScanner) (domain.BootstrapUpdateJob, error) { + var ( + job domain.BootstrapUpdateJob + kind, status string + authKeyID int64 + readyAt, published sql.NullTime + ) + if err := row.Scan( + &job.ID, + &kind, + &job.UserID, + &authKeyID, + &job.SessionID, + &job.MessageBoxID, + &status, + &job.Attempts, + &job.LastError, + &job.CreatedAt, + &job.UpdatedAt, + &readyAt, + &published, + ); err != nil { + return domain.BootstrapUpdateJob{}, fmt.Errorf("scan bootstrap job: %w", err) + } + job.Kind = domain.BootstrapUpdateJobKind(kind) + job.Status = domain.BootstrapUpdateJobStatus(status) + job.AuthKeyID = authKeyIDFromInt64(authKeyID) + if readyAt.Valid { + job.ReadyAt = readyAt.Time + } + if published.Valid { + job.PublishedAt = published.Time + } + return job, nil +} diff --git a/internal/store/postgres/update_event.go b/internal/store/postgres/update_event.go index 200954f6..60a63da8 100644 --- a/internal/store/postgres/update_event.go +++ b/internal/store/postgres/update_event.go @@ -457,6 +457,34 @@ func (s *UpdateEventStore) ListAfter(ctx context.Context, userID int64, pts, lim return out, nil } +func (s *UpdateEventStore) FindNewMessageEvent(ctx context.Context, userID int64, messageBoxID int) (domain.UpdateEvent, bool, error) { + if userID == 0 || messageBoxID <= 0 { + return domain.UpdateEvent{}, false, nil + } + var pts int + if err := s.db.QueryRow(ctx, ` +SELECT pts +FROM user_update_events +WHERE user_id = $1 + AND event_type = $2 + AND message_box_id = $3 +ORDER BY pts ASC +LIMIT 1`, userID, string(domain.UpdateEventNewMessage), messageBoxID).Scan(&pts); err != nil { + if err == pgx.ErrNoRows { + return domain.UpdateEvent{}, false, nil + } + return domain.UpdateEvent{}, false, fmt.Errorf("find new message update event: %w", err) + } + events, err := s.ListAfter(ctx, userID, pts-1, 1) + if err != nil { + return domain.UpdateEvent{}, false, err + } + if len(events) == 0 || events[0].Pts != pts { + return domain.UpdateEvent{}, false, fmt.Errorf("hydrate new message update event: expected pts %d", pts) + } + return events[0], true, nil +} + // MaxContiguousPts 见 store.UpdateEventStore 接口说明。PG 写路径保证水位与 // durable event 同事务提交；缺行代表该账号还没有 update。 func (s *UpdateEventStore) MaxContiguousPts(ctx context.Context, userID int64) (int, error) {