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

feat: defer login bootstrap updates

Browse source
This commit is contained in:
A 2026-07-05 03:28:19 +08:00
parent a246fd5417
commit 4d9f1e271d
18 changed files with 958 additions and 188 deletions
Download patch file Download diff file Expand all files Collapse all files

59
cmd/telesrv/main.go
View file

@ -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)

2
deploy/migrations/0052_bootstrap_update_jobs.down.sql Normal file
View file

@ -0,0 +1,2 @@
DROP INDEX IF EXISTS public.user_update_events_new_message_box_idx;
DROP TABLE IF EXISTS public.bootstrap_update_jobs;

34
deploy/migrations/0052_bootstrap_update_jobs.up.sql Normal file
View file

@ -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;

35
internal/app/updates/service.go
View file

@ -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

39
internal/app/updates/service_test.go
View file

@ -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

39
internal/domain/bootstrap_update.go Normal file
View file

@ -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
}

84
internal/rpc/auth.go
View file

@ -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"

185
internal/rpc/bootstrap_updates.go Normal file
View file

@ -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
}

60
internal/rpc/deps.go
View file

@ -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):创建/更新/查询主题 + 维护每用户已安装列表。

13
internal/rpc/rpc_testkit_updates_test.go
View file

@ -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{

3
internal/rpc/updates.go
View file

@ -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
}

109
internal/rpc/updates_rpc_test.go
View file

@ -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)
}
}

16
internal/store/bootstrap_update_job.go Normal file
View file

@ -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
}

144
internal/store/memory/bootstrap_update_job.go Normal file
View file

@ -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
}

58
internal/store/memory/bootstrap_update_job_test.go Normal file
View file

@ -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)
}
}

49
internal/store/memory/updates.go
View file

@ -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()

189
internal/store/postgres/bootstrap_update_job.go Normal file
View file

@ -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
}

28
internal/store/postgres/update_event.go
View file

@ -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) {