From 7e431a25fb2c80078d55771467ada5b6df3ce496 Mon Sep 17 00:00:00 2001 From: std-external <201375710+std-external@users.noreply.github.com> Date: Sun, 18 Oct 2026 14:32:08 +0800 Subject: [PATCH] bridge: schedule text messages in the database and deliver them on a timer --- packages/bridge/package.json | 1 + packages/bridge/src/dialogs.ts | 223 ++++++++- packages/bridge/src/index.ts | 153 ++++-- packages/bridge/src/models.ts | 54 ++ .../bridge/src/scheduled-messages.test.ts | 363 ++++++++++++++ packages/bridge/src/scheduled-messages.ts | 466 ++++++++++++++++++ yarn.lock | 25 + 7 files changed, 1229 insertions(+), 56 deletions(-) create mode 100644 packages/bridge/src/scheduled-messages.test.ts create mode 100644 packages/bridge/src/scheduled-messages.ts diff --git a/packages/bridge/package.json b/packages/bridge/package.json index 54d79815..d7cb4cd0 100644 --- a/packages/bridge/package.json +++ b/packages/bridge/package.json @@ -56,6 +56,7 @@ "@mtproto-relay/mtproto": "workspace:*", "@mtproto-relay/update-store": "workspace:*", "@mtproto-relay/update-store-memory": "workspace:*", + "cron": "^4.4.0", "jsqr": "^1.4.0", "linkify-it": "^6.1.0", "long": "^5.3.2", diff --git a/packages/bridge/src/dialogs.ts b/packages/bridge/src/dialogs.ts index a86b1373..d18149f5 100644 --- a/packages/bridge/src/dialogs.ts +++ b/packages/bridge/src/dialogs.ts @@ -38,6 +38,7 @@ import { probeImageDimensions } from './image-dimensions.js' import { withAutoLinkEntities } from './message-entities.js' import { getCardThumbnailFile, makeCardThumbnailPhoto, storageFileType } from './card-thumbnail.js' import type { DraftStore, StoredDraft } from './draft-store.js' +import type { ScheduledMessage, ScheduledMessageStore } from './scheduled-messages.js' import { MUTE_FOREVER, type NotificationSettingsStore, type NotificationTarget } from './notification-settings.js' import type { BlockedPeerChange, BlockedPeerStore } from './blocked-peers.js' import { SystemPeerCallbackError, type SystemPeerResolution, type SystemPeerService } from './system-peer.js' @@ -330,6 +331,8 @@ export class DialogRpc { private readonly _systemPeers?: SystemPeerService, private readonly _messageProjection?: MessageProjectionPipeline, private readonly _sendEchoes?: SendEchoes, + private readonly _scheduled?: ScheduledMessageStore, + private readonly _onScheduledChange?: () => void, ) { this._actions = new PlatformMessageActions(_platform, _session) // 有 `MessageStore` 时淘汰旧条目并由数据库兜底；无存储模式不可淘汰，避免重用已分配的 ID。 @@ -1335,7 +1338,221 @@ export class DialogRpc { req: tl.messages.RawGetScheduledHistoryRequest, ): Promise { await this._hydratePeers() - return this._emptyMessages(this._conversation(this._resolvePeer(req.peer))) + const conversation = this._conversation(this._resolvePeer(req.peer)) + const rows = await this._scheduledMessages(conversation.id) + return this._scheduledMessagesReply(conversation, rows) + } + + async getScheduledMessages( + req: tl.messages.RawGetScheduledMessagesRequest, + ): Promise { + await this._hydratePeers() + const conversation = this._conversation(this._resolvePeer(req.peer)) + if (!this._scheduled) throw new RpcError(400, 'SCHEDULED_MESSAGES_UNAVAILABLE') + const rows = await this._scheduled.byTlIds( + this._session.platformSessionId, conversation.id, req.id, + ) + return this._scheduledMessagesReply(conversation, rows) + } + + /** + * Send scheduled messages now. + * + * Each row goes through the ordinary send path, so the resulting message is + * persisted and published exactly like a normal send; the scheduled entries + * are then removed from the client-visible list. + */ + async sendScheduledMessages( + req: tl.messages.RawSendScheduledMessagesRequest, + ): Promise { + await this._hydratePeers() + const conversation = this._conversation(this._resolvePeer(req.peer)) + if (!this._scheduled) throw new RpcError(400, 'SCHEDULED_MESSAGES_UNAVAILABLE') + const rows = await this._scheduled.byTlIds( + this._session.platformSessionId, conversation.id, req.id, + ) + if (!rows.length) throw new RpcError(400, 'MESSAGE_ID_INVALID') + const updates: tl.TypeUpdate[] = [] + const sentIds: number[] = [] + for (const row of rows) { + const sent = await this._sendMessage(this._scheduledSendRequest(row)) + const sentUpdates = sent._ === 'updates' ? sent.updates : [] + updates.push(...sentUpdates) + const newId = sentUpdates + .map((update) => update._ === 'updateNewMessage' || update._ === 'updateNewChannelMessage' + ? update.message.id + : undefined) + .find((id): id is number => id !== undefined) + if (newId !== undefined) sentIds.push(newId) + await this._scheduled.cancel( + this._session.platformSessionId, conversation.id, row.tlMessageId, + ) + } + this._onScheduledChange?.() + updates.push({ + _: 'updateDeleteScheduledMessages', + peer: this._conversationPeer(conversation), + messages: rows.map((row) => row.tlMessageId), + }) + return { + _: 'updates', updates, + users: [this._makeSelfUser()], + chats: conversation.kind === 'direct' ? [] : [this._makeChat(conversation)], + date: Math.floor(Date.now() / 1_000), seq: 0, + } + } + + async deleteScheduledMessages( + req: tl.messages.RawDeleteScheduledMessagesRequest, + ): Promise { + await this._hydratePeers() + const conversation = this._conversation(this._resolvePeer(req.peer)) + if (!this._scheduled) throw new RpcError(400, 'SCHEDULED_MESSAGES_UNAVAILABLE') + const removed: number[] = [] + for (const tlMessageId of req.id) { + if (await this._scheduled.cancel( + this._session.platformSessionId, conversation.id, tlMessageId, + )) removed.push(tlMessageId) + } + if (removed.length) this._onScheduledChange?.() + return { + _: 'updates', updates: [{ + _: 'updateDeleteScheduledMessages', + peer: this._conversationPeer(conversation), + messages: removed, + }], + users: [this._makeSelfUser()], + chats: conversation.kind === 'direct' ? [] : [this._makeChat(conversation)], + date: Math.floor(Date.now() / 1_000), seq: 0, + } + } + + /** + * Deliver one stored row when its time comes (called by the dispatcher). + * + * The peer is rebuilt from the stored conversation id rather than from the + * creating client's input peer, so a restart cannot leave a stale peer behind. + */ + async deliverScheduled(message: ScheduledMessage): Promise<{ + sentTlMessageId?: number + peer: tl.TypePeer + }> { + const conversationId = message.platformConversationId + if (!this._conversations.has(conversationId)) { + const stored = await this._store?.getConversation( + this._session.platformSessionId, conversationId, + ) + if (stored) this._conversations.set(conversationId, stored) + else this._conversations.set(conversationId, { + id: conversationId, kind: message.conversationKind, title: conversationId, + }) + } + const conversation = this._conversation(conversationId) + if (conversation.kind === 'direct') await this._getPeerUser(conversationId) + const result = await this._sendMessage(this._scheduledSendRequest(message)) + const resultUpdates = result._ === 'updates' ? result.updates : [] + const sentTlMessageId = resultUpdates + .map((update) => update._ === 'updateNewMessage' || update._ === 'updateNewChannelMessage' + ? update.message.id + : undefined) + .find((id): id is number => id !== undefined) + return { sentTlMessageId, peer: this._conversationPeer(conversation) } + } + + private _scheduledSendRequest(message: ScheduledMessage): SendMessageRequest { + const conversation = this._conversation(message.platformConversationId) + const peer: tl.TypeInputPeer = conversation.kind === 'direct' + ? { _: 'inputPeerUser', userId: this._userId(conversation.id), accessHash: Long.ZERO } + : { _: 'inputPeerChannel', channelId: this._peerId(conversation.id), accessHash: Long.ONE } + return { + _: 'messages.sendMessage', + peer, + message: message.text, + randomId: Long.fromString(message.randomId), + entities: (message.entities ?? undefined) as unknown as tl.TypeMessageEntity[] | undefined, + replyTo: message.replyToMsgId + ? { _: 'inputReplyToMessage', replyToMsgId: message.replyToMsgId } + : undefined, + } + } + + private async _scheduleTextMessage(req: SendMessageRequest): Promise { + if (!this._scheduled || !this._store) throw new RpcError(400, 'SCHEDULED_MESSAGES_UNAVAILABLE') + await this._hydratePeers() + const conversationId = this._resolveMessageTarget(req.peer, req.replyTo) + this._assertWritableConversation(conversationId) + const conversation = this._conversation(conversationId) + if (conversation.kind === 'channel' && conversation.parentId) { + throw new RpcError(400, 'SCHEDULED_MESSAGES_UNAVAILABLE') + } + const dueAt = req.scheduleDate! + if (!Number.isSafeInteger(dueAt) || dueAt <= 0) throw new RpcError(400, 'SCHEDULE_DATE_INVALID') + const idScope = `scheduled:${this._session.platformSessionId}:${conversationId}` + const [tlMessageId] = await this._store.allocateIds(idScope, 1) + const replyToMsgId = req.replyTo?._ === 'inputReplyToMessage' + ? req.replyTo.replyToMsgId + : undefined + const row = await this._scheduled.create({ + platformSessionId: this._session.platformSessionId, + platformConversationId: conversationId, + conversationKind: conversation.kind, + tlMessageId, + text: req.message, + entities: (req.entities as unknown as JsonValue[] | undefined) ?? null, + replyToMsgId, + dueAt, + randomId: req.randomId.toString(), + }) + this._onScheduledChange?.() + return { + _: 'updates', + updates: [{ + _: 'updateNewScheduledMessage', + message: this._scheduledMessageTl(row, conversation), + }], + users: [this._makeSelfUser()], + chats: conversation.kind === 'direct' ? [] : [this._makeChat(conversation)], + date: Math.floor(Date.now() / 1_000), seq: 0, + } + } + + private async _scheduledMessages(conversationId: string): Promise { + if (!this._scheduled) return [] + return this._scheduled.list(this._session.platformSessionId, conversationId) + } + + private _scheduledMessagesReply( + conversation: import('./platform.js').IMConversation, + rows: ScheduledMessage[], + ): tl.messages.RawMessages { + return { + _: 'messages.messages', + messages: rows + .sort((left, right) => left.dueAt - right.dueAt) + .map((row) => this._scheduledMessageTl(row, conversation)), + topics: [], + chats: conversation.kind === 'direct' ? [] : [this._makeChat(conversation)], + users: [this._makeSelfUser()], + } + } + + private _scheduledMessageTl( + row: ScheduledMessage, + conversation: import('./platform.js').IMConversation, + ): tl.RawMessage { + return { + _: 'message', + id: row.tlMessageId, + peerId: this._conversationPeer(conversation), + fromId: { _: 'peerUser', userId: this._selfId }, + date: row.dueAt, + message: row.text, + entities: (row.entities ?? undefined) as unknown as tl.TypeMessageEntity[] | undefined, + out: true, + replyTo: row.replyToMsgId + ? { _: 'messageReplyHeader', replyToScheduled: true, replyToMsgId: row.replyToMsgId } + : undefined, + } } async getMessages(req: GetMessagesRequest): Promise { @@ -2222,6 +2439,7 @@ export class DialogRpc { } async sendMedia(req: SendMediaRequest, excludeConnection?: ServerConnection): Promise { + if (req.scheduleDate !== undefined) throw new RpcError(400, 'SCHEDULED_MESSAGES_UNAVAILABLE') return this._sendMediaOnce(req.randomId.toString(), async () => { const startedAt = performance.now() const resolved = await this._resolveSendMedia(req.media) @@ -2257,6 +2475,7 @@ export class DialogRpc { } async sendMultiMedia(req: SendMultiMediaRequest, excludeConnection?: ServerConnection): Promise { + if (req.scheduleDate !== undefined) throw new RpcError(400, 'SCHEDULED_MESSAGES_UNAVAILABLE') const randomId = req.multiMedia.map((item) => item.randomId.toString()).join(':') return this._sendMediaOnce(randomId, async () => { if (!req.multiMedia.length) throw new RpcError(400, 'MEDIA_EMPTY') @@ -3497,7 +3716,7 @@ export class DialogRpc { if (Array.from(req.message).length > this._platform.capabilities.send.maxTextLength) { throw new RpcError(400, 'MESSAGE_TOO_LONG') } - if (req.scheduleDate !== undefined) throw new RpcError(400, 'SCHEDULED_MESSAGES_UNAVAILABLE') + if (req.scheduleDate !== undefined) return this._scheduleTextMessage(req) await this._hydratePeers() const peerId = this._resolveMessageTarget(req.peer, req.replyTo) diff --git a/packages/bridge/src/index.ts b/packages/bridge/src/index.ts index 2c75ee82..ef53c44e 100644 --- a/packages/bridge/src/index.ts +++ b/packages/bridge/src/index.ts @@ -38,6 +38,9 @@ import { SRP_PRIME_HEX, SrpChallengeStore, generateSrpVerifier, isValidSrpVerifier, verifySrpChallenge, } from './login-srp.js' import { DraftStore } from './draft-store.js' +import { + ScheduledMessageDispatcher, ScheduledMessageStore, type ScheduledMessage, +} from './scheduled-messages.js' import { NotificationSettingsStore } from './notification-settings.js' import { makeCrossGramServerConfig, makePlatformAccountView, makeUnavailableAccountView, @@ -271,6 +274,7 @@ export function apply(ctx: Context, config: BridgeConfig = {}): void { } }, 'mtproto-bridge.search-rebuild') const drafts = new DraftStore(ctx.database) + const scheduledStore = new ScheduledMessageStore(ctx.database) const notificationSettings = new NotificationSettingsStore( ctx.database, config.autoMuteGroupChats ?? true, ) @@ -468,6 +472,7 @@ export function apply(ctx: Context, config: BridgeConfig = {}): void { bridgeLogger.warn('system peer bootstrap failed: %s', String(error)) }) }) + let scheduledDispatcher: ScheduledMessageDispatcher | undefined const resolveBridgeSession = createSessionResolver( ctx, registry, stickerRpcFor, reactionRpcFor, resources, store, drafts, notificationSettings, blockedPeers, dialogFolders, @@ -483,12 +488,39 @@ export function apply(ctx: Context, config: BridgeConfig = {}): void { }, messageProjection, sendEchoes, + scheduledStore, + () => { void scheduledDispatcher?.rescan() }, ) const requireBridgeSession: typeof resolveBridgeSession = async (rpcContext, ...rest) => { const state = await resolveBridgeSession(rpcContext, ...rest) historySync.touch(state.session.platformSessionId, rpcContext.mtprotoRpc?.request._) return state } + requireBridgeSession.buildState = resolveBridgeSession.buildState + const deliverScheduledMessage = async (message: ScheduledMessage): Promise => { + const [row] = await ctx.database.get('mtproto_platform_session', { + id: message.platformSessionId, active: true, + }) + if (!row) throw new Error(`platform session is not active: ${message.platformSessionId}`) + const state = await resolveBridgeSession.buildState({ + platformId: row.platformId, platformSessionId: message.platformSessionId, + }) + const delivered = await state.dialogs.deliverScheduled(message) + await updates.publishAccountUpdates(state.session, [{ + _: 'updateDeleteScheduledMessages', + peer: delivered.peer, + messages: [message.tlMessageId], + sentMessages: delivered.sentTlMessageId === undefined ? undefined : [delivered.sentTlMessageId], + } as tl.TypeUpdate]) + return delivered.sentTlMessageId + } + scheduledDispatcher = new ScheduledMessageDispatcher(scheduledStore, deliverScheduledMessage) + ctx.effect(() => { + void scheduledDispatcher!.start().catch((error) => { + bridgeLogger.warn('scheduled message dispatcher failed to start: %s', String(error)) + }) + return () => scheduledDispatcher!.stop() + }, 'mtproto-bridge.scheduled-messages') new MtprotoBridgeService(ctx, requireBridgeSession, dcId) registerGroupFilesMiniApp(ctx, platforms, requireBridgeSession, config.groupFilesMiniApp) @@ -1057,6 +1089,12 @@ export function apply(ctx: Context, config: BridgeConfig = {}): void { )) rpc.register('messages.getScheduledHistory', async (rpc, req) => (await requireBridgeSession(rpc)).dialogs.getScheduledHistory(req as tl.messages.RawGetScheduledHistoryRequest)) + rpc.register('messages.getScheduledMessages', async (rpc, req) => + (await requireBridgeSession(rpc)).dialogs.getScheduledMessages(req as tl.messages.RawGetScheduledMessagesRequest)) + rpc.register('messages.sendScheduledMessages', async (rpc, req) => + (await requireBridgeSession(rpc)).dialogs.sendScheduledMessages(req as tl.messages.RawSendScheduledMessagesRequest)) + rpc.register('messages.deleteScheduledMessages', async (rpc, req) => + (await requireBridgeSession(rpc)).dialogs.deleteScheduledMessages(req as tl.messages.RawDeleteScheduledMessagesRequest)) rpc.register('messages.getPinnedDialogs', async (rpc, req) => (await requireBridgeSession(rpc)).dialogs.getPinnedDialogs((req as tl.messages.RawGetPinnedDialogsRequest).folderId)) rpc.register('messages.toggleDialogPin', async (rpc, req) => { @@ -1562,6 +1600,8 @@ export function createSessionResolver( ) => void | Promise, messageProjection?: MessageProjectionPipeline, sendEchoes?: SendEchoes, + scheduled?: ScheduledMessageStore, + onScheduledChange?: () => void, ) { const projection = messageProjection ?? new MessageProjectionPipeline(ctx) const loading = new Map>() @@ -1583,7 +1623,60 @@ export function createSessionResolver( } } - return async ( + const makeState = async ( + identity: { platformId: string, platformSessionId: string }, + options: { notifyRpc?: ServerRpcContext, authKeyId?: string } = {}, + ) => { + const platform = registry.get(identity.platformId) + if (!platform) throw new RpcError(500, 'PLATFORM_NOT_AVAILABLE') + const [row] = await ctx.database.get('mtproto_platform_session', { + id: identity.platformSessionId, + active: true, + }) + if (!row) throw new RpcError(401, 'PLATFORM_SESSION_REVOKED') + const [auth] = await ctx.database.get('mtproto_auth_session', { + platformId: identity.platformId, platformSessionId: identity.platformSessionId, + }) + if (!auth) throw new RpcError(401, 'AUTH_KEY_UNREGISTERED') + const session = sessionFromRow(row, auth.virtualPhone) + const selfRow = await store.getUser(session.platformId, session.userId) + ?? await store.upsertUser(session, { + id: session.userId, + firstName: (row.metadata.firstName as string) ?? 'Bridge', + lastName: row.metadata.lastName as string | undefined, + username: row.metadata.username as string | undefined, + metadata: row.metadata, + }) + await subscriptions.ensure(session) + if (options.notifyRpc) await notifyAuthorizedSession(session, options.notifyRpc) + const state: BridgeSessionState = { + generation, platform, session, projection, + stickers: stickerRpcFor(platform, session), + dialogs: undefined as never, + } + state.dialogs = new DialogRpc( + platform, session, store, uploads, onTransferProgress, dcId, state.stickers, + reactionRpcFor(platform, session), + resources, + (localSession, event, options) => subscriptions.ingestLocalEvent(localSession, event, options), + options.authKeyId, + historyTrace, + drafts, + onDraftUpdate, + notificationSettings, + selfRow.id, + blockedPeers, + dialogFolders, + systemPeers, + projection, + sendEchoes, + scheduled, + onScheduledChange, + ) + return state + } + + const requireSession = async ( rpc: ServerRpcContext, provisionalIdentity?: { platformId: string, platformSessionId: string }, cache = true, @@ -1599,60 +1692,9 @@ export function createSessionResolver( if (!rpc.authKeyId) throw new RpcError(401, 'AUTH_KEY_UNREGISTERED') const authKeyId = authKeyHex(rpc.authKeyId) - const makeState = async ( - identity: { platformId: string, platformSessionId: string }, - notify = true, - ) => { - const platform = registry.get(identity.platformId) - if (!platform) throw new RpcError(500, 'PLATFORM_NOT_AVAILABLE') - const [row] = await ctx.database.get('mtproto_platform_session', { - id: identity.platformSessionId, - active: true, - }) - if (!row) throw new RpcError(401, 'PLATFORM_SESSION_REVOKED') - const [auth] = await ctx.database.get('mtproto_auth_session', { - platformId: identity.platformId, platformSessionId: identity.platformSessionId, - }) - if (!auth) throw new RpcError(401, 'AUTH_KEY_UNREGISTERED') - const session = sessionFromRow(row, auth.virtualPhone) - const selfRow = await store.getUser(session.platformId, session.userId) - ?? await store.upsertUser(session, { - id: session.userId, - firstName: (row.metadata.firstName as string) ?? 'Bridge', - lastName: row.metadata.lastName as string | undefined, - username: row.metadata.username as string | undefined, - metadata: row.metadata, - }) - await subscriptions.ensure(session) - if (notify) await notifyAuthorizedSession(session, rpc) - const state: BridgeSessionState = { - generation, platform, session, projection, - stickers: stickerRpcFor(platform, session), - dialogs: undefined as never, - } - state.dialogs = new DialogRpc( - platform, session, store, uploads, onTransferProgress, dcId, state.stickers, - reactionRpcFor(platform, session), - resources, - (localSession, event, options) => subscriptions.ingestLocalEvent(localSession, event, options), - authKeyId, - historyTrace, - drafts, - onDraftUpdate, - notificationSettings, - selfRow.id, - blockedPeers, - dialogFolders, - systemPeers, - projection, - sendEchoes, - ) - return state - } - if (!cache) { if (!provisionalIdentity) throw new RpcError(401, 'AUTH_KEY_UNREGISTERED') - return makeState(provisionalIdentity, false) + return makeState(provisionalIdentity) } while (true) { @@ -1661,7 +1703,7 @@ export function createSessionResolver( pending = (async () => { const [binding] = await ctx.database.get('mtproto_auth_binding', { authKeyId }) if (!binding) throw new RpcError(401, 'AUTH_KEY_UNREGISTERED') - return makeState(binding) + return makeState(binding, { notifyRpc: rpc, authKeyId }) })() loading.set(authKeyId, pending) pending.finally(() => { @@ -1679,6 +1721,9 @@ export function createSessionResolver( return state } } + requireSession.buildState = async (identity: { platformId: string, platformSessionId: string }) => + makeState(identity) + return requireSession } function authKeyHex(authKeyId: Uint8Array): string { diff --git a/packages/bridge/src/models.ts b/packages/bridge/src/models.ts index 4f4118c9..9a215b83 100644 --- a/packages/bridge/src/models.ts +++ b/packages/bridge/src/models.ts @@ -281,6 +281,33 @@ export interface DialogFilterRow { updatedAt: Date } + +export interface ScheduledMessageRow { + id: number + platformSessionId: string + platformConversationId: string + conversationKind: IMConversationKind + tlMessageId: number + text: string + entities: JsonValue[] | null + replyToMsgId: number | null + createdAt: Date +} + +export interface ScheduledMessageStateRow { + messageId: number + dueAt: number + status: 'pending' | 'sending' | 'sent' | 'failed' | 'cancelled' + attempts: number + nextAttemptAt: number + leaseOwner: string | null + leaseUntil: number | null + randomId: string + sentTlMessageId: number | null + lastError: string | null + updatedAt: Date +} + export interface DialogFolderPeerRow { id: string platformSessionId: string @@ -320,6 +347,8 @@ declare module '@cordisjs/plugin-database' { mtproto_blocked_peer: BlockedPeerRow mtproto_dialog_filter: DialogFilterRow mtproto_dialog_folder_peer: DialogFolderPeerRow + mtproto_scheduled_message: ScheduledMessageRow + mtproto_scheduled_state: ScheduledMessageStateRow } } @@ -560,4 +589,29 @@ export function defineModels(ctx: Context): void { unique: [['platformSessionId', 'platformConversationId']], indexes: [['platformSessionId', 'folderId'], ['platformSessionId', 'pinnedOrder']], }) + + ctx.model.extend('mtproto_scheduled_message', { + id: 'unsigned', platformSessionId: 'string', platformConversationId: 'text', + conversationKind: 'string', tlMessageId: 'unsigned', text: 'text', + entities: { type: 'json', nullable: true }, + replyToMsgId: { type: 'unsigned', nullable: true }, + createdAt: 'timestamp', + }, { + primary: 'id', autoInc: true, + unique: [['platformSessionId', 'platformConversationId', 'tlMessageId']], + indexes: [['platformSessionId', 'platformConversationId']], + }) + + ctx.model.extend('mtproto_scheduled_state', { + messageId: 'unsigned', dueAt: 'integer', status: 'string', attempts: 'unsigned', + nextAttemptAt: 'integer', leaseOwner: { type: 'text', nullable: true }, + leaseUntil: { type: 'integer', nullable: true }, randomId: 'text', + sentTlMessageId: { type: 'unsigned', nullable: true }, + lastError: { type: 'text', nullable: true }, updatedAt: 'timestamp', + }, { + primary: 'messageId', + unique: [['randomId']], + indexes: [['status', 'nextAttemptAt'], ['status', 'leaseUntil']], + }) } + diff --git a/packages/bridge/src/scheduled-messages.test.ts b/packages/bridge/src/scheduled-messages.test.ts new file mode 100644 index 00000000..35e2d561 --- /dev/null +++ b/packages/bridge/src/scheduled-messages.test.ts @@ -0,0 +1,363 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { Context } from 'cordis' +import Database from '@cordisjs/plugin-database' +import SQLiteDriver from '@cordisjs/plugin-database-sqlite' +import Long from 'long' +import type { tl } from '@mtcute/core' +import { defineModels } from './models.js' +import { DialogRpc } from './dialogs.js' +import { MessageStore } from './message-store.js' +import type { IMMessageInput, IMPlatform } from './platform.js' +import { + clampScheduleDelay, SCHEDULED_MESSAGE_MAX_DELAY_MS, ScheduledMessageDispatcher, + ScheduledMessageStore, type ScheduledMessage, +} from './scheduled-messages.js' + +const disposals: Array<() => Promise> = [] + +afterEach(async () => { + await Promise.all(disposals.splice(0).map((dispose) => dispose())) +}) + +async function createDatabase() { + const ctx = new Context() + const fibers = [ctx.plugin(Database), ctx.plugin(SQLiteDriver, { path: ':memory:' })] + await Promise.all(fibers) + await new Promise((resolve) => setTimeout(resolve, 25)) + defineModels(ctx) + await ctx.database.prepared() + disposals.push(async () => { + for (const fiber of fibers.reverse()) await Promise.resolve((fiber as any).dispose?.()) + }) + return ctx.database +} + +function nowSeconds(): number { + return Math.floor(Date.now() / 1_000) +} + +async function createDue( + store: ScheduledMessageStore, + options: { dueAt?: number, text?: string, tlMessageId?: number, randomId?: string } = {}, +): Promise { + return store.create({ + platformSessionId: 'session', + platformConversationId: 'peer', + conversationKind: 'direct', + tlMessageId: options.tlMessageId ?? 1, + text: options.text ?? 'later', + dueAt: options.dueAt ?? nowSeconds() + 60, + randomId: options.randomId ?? String(options.tlMessageId ?? 1), + }) +} + +interface ArmedTimer { + dueMs: number + fire: () => void +} + +function recordingScheduler(): { armed: ArmedTimer[], schedule: (dueMs: number, callback: () => void) => () => void } { + const armed: ArmedTimer[] = [] + return { + armed, + schedule: (dueMs, callback) => { + armed.push({ dueMs, fire: callback }) + return () => undefined + }, + } +} + +describe('scheduled message store', () => { + it('creates a message and its scheduling state in one row pair', async () => { + const database = await createDatabase() + const store = new ScheduledMessageStore(database) + const created = await createDue(store, { dueAt: 1_800_000_000 }) + expect(created).toMatchObject({ + platformSessionId: 'session', + platformConversationId: 'peer', + tlMessageId: 1, + status: 'pending', + attempts: 0, + dueAt: 1_800_000_000, + }) + const list = await store.list('session', 'peer') + expect(list.map((row) => row.text)).toEqual(['later']) + }) + + it('cancels a pending message and hides it from the list', async () => { + const database = await createDatabase() + const store = new ScheduledMessageStore(database) + const created = await createDue(store) + expect(await store.cancel('session', 'peer', created.tlMessageId)).toBe(true) + expect(await store.list('session', 'peer')).toHaveLength(0) + expect(await store.cancel('session', 'peer', created.tlMessageId)).toBe(false) + }) +}) + +describe('scheduled message dispatcher', () => { + it('sends a due message and marks it sent', async () => { + const database = await createDatabase() + const store = new ScheduledMessageStore(database) + const message = await createDue(store, { dueAt: nowSeconds() - 5 }) + const sent: number[] = [] + const timer = recordingScheduler() + const dispatcher = new ScheduledMessageDispatcher(store, async (row) => { + sent.push(row.tlMessageId) + return 900 + }, { schedule: timer.schedule }) + await dispatcher.start() + await dispatcher.sweep() + expect(sent).toEqual([message.tlMessageId]) + const [state] = await database.get('mtproto_scheduled_state', { messageId: message.id }) + expect(state.status).toBe('sent') + expect(state.sentTlMessageId).toBe(900) + dispatcher.stop() + }) + + it('recovers a message whose sender crashed with an expired lease', async () => { + const database = await createDatabase() + const store = new ScheduledMessageStore(database) + const message = await createDue(store, { dueAt: nowSeconds() - 60 }) + await database.set('mtproto_scheduled_state', { messageId: message.id }, { + status: 'sending', leaseOwner: 'crashed', leaseUntil: nowSeconds() - 30, + }) + const sent: number[] = [] + const dispatcher = new ScheduledMessageDispatcher(store, async (row) => { + sent.push(row.tlMessageId) + return 1 + }, { schedule: recordingScheduler().schedule }) + await dispatcher.start() + await dispatcher.sweep() + expect(sent).toEqual([message.tlMessageId]) + dispatcher.stop() + }) + + it('sends once when two dispatchers race for the same row', async () => { + const database = await createDatabase() + const store = new ScheduledMessageStore(database) + await createDue(store, { dueAt: nowSeconds() - 5 }) + let sends = 0 + const send = async () => { + sends++ + await new Promise((resolve) => setTimeout(resolve, 10)) + return 7 + } + const first = new ScheduledMessageDispatcher(store, send, { + owner: 'first', schedule: recordingScheduler().schedule, + }) + const second = new ScheduledMessageDispatcher(store, send, { + owner: 'second', schedule: recordingScheduler().schedule, + }) + await Promise.all([first.sweep(), second.sweep()]) + expect(sends).toBe(1) + first.stop() + second.stop() + }) + + it('does not send a cancelled message', async () => { + const database = await createDatabase() + const store = new ScheduledMessageStore(database) + const message = await createDue(store, { dueAt: nowSeconds() - 5 }) + await store.cancel('session', 'peer', message.tlMessageId) + let sends = 0 + const dispatcher = new ScheduledMessageDispatcher(store, async () => { + sends++ + return 1 + }, { schedule: recordingScheduler().schedule }) + await dispatcher.start() + await dispatcher.sweep() + expect(sends).toBe(0) + dispatcher.stop() + }) + + it('keeps the current timer when a later message is added', async () => { + const database = await createDatabase() + const store = new ScheduledMessageStore(database) + const timer = recordingScheduler() + const dispatcher = new ScheduledMessageDispatcher(store, async () => 1, { + schedule: timer.schedule, + }) + await createDue(store, { dueAt: nowSeconds() + 120, tlMessageId: 10, randomId: '10' }) + await dispatcher.start() + expect(timer.armed).toHaveLength(1) + + await createDue(store, { dueAt: nowSeconds() + 300, tlMessageId: 11, randomId: '11' }) + await dispatcher.rescan() + expect(timer.armed).toHaveLength(1) + + await createDue(store, { dueAt: nowSeconds() + 30, tlMessageId: 12, randomId: '12' }) + await dispatcher.rescan() + expect(timer.armed).toHaveLength(2) + dispatcher.stop() + }) + + it('parks a message after repeated send failures', async () => { + const database = await createDatabase() + const store = new ScheduledMessageStore(database) + const message = await createDue(store, { dueAt: nowSeconds() - 5 }) + const dispatcher = new ScheduledMessageDispatcher(store, async () => { + throw new Error('platform offline') + }, { schedule: recordingScheduler().schedule }) + await dispatcher.start() + for (let attempt = 0; attempt < 7; attempt++) { + await database.set('mtproto_scheduled_state', { messageId: message.id }, { + status: 'pending', nextAttemptAt: nowSeconds() - 1, + }) + await dispatcher.sweep() + } + const [state] = await database.get('mtproto_scheduled_state', { messageId: message.id }) + expect(state.status).toBe('failed') + expect(state.lastError).toBe('platform offline') + dispatcher.stop() + }) +}) + +describe('scheduled timer cap', () => { + it('never asks a timer to wait longer than the 24 day cap', () => { + const thirtyDays = 30 * 24 * 60 * 60 * 1_000 + expect(clampScheduleDelay(thirtyDays)).toBe(SCHEDULED_MESSAGE_MAX_DELAY_MS) + expect(clampScheduleDelay(-5)).toBe(0) + expect(clampScheduleDelay(Number.POSITIVE_INFINITY)).toBe(SCHEDULED_MESSAGE_MAX_DELAY_MS) + expect(SCHEDULED_MESSAGE_MAX_DELAY_MS).toBeLessThan(2 ** 31 - 1) + }) +}) + +describe('scheduled message RPCs', () => { + it('creates, lists, delivers and deletes a scheduled text message', async () => { + const database = await createDatabase() + const store = new MessageStore(database) + const scheduled = new ScheduledMessageStore(database) + const session = { + platformSessionId: 'rpc-session', platformId: 'rpc-platform', userId: 'self', + credentials: {}, metadata: { firstName: 'Self' }, + } + let lastInput: IMMessageInput | undefined + const platform: IMPlatform = { + capabilities: { + history: true, + send: { text: true, images: false, files: false, mixed: false, maxTextLength: 4096, maxMedia: 0 }, + conversations: { groups: false, channels: false, subchannels: false }, + }, + async subscribe() { return () => {} }, + async sendMessage(_session, conversation, content) { + lastInput = content + return { + id: String(Date.now()), conversationId: conversation.id, senderId: 'self', + timestamp: Math.floor(Date.now() / 1_000), outgoing: true, content, + } + }, + async getDialogs() { + return { + dialogs: [{ + conversation: { id: 'alice', kind: 'direct' as const, title: 'Alice' }, + unreadCount: 0, + }], + } + }, + async getHistory() { return { messages: [] } }, + async getUser(_session, id) { + return id === 'alice' ? { id, firstName: 'Alice' } : null + }, + async getContacts() { return { users: [{ id: 'alice', firstName: 'Alice' }] } }, + } as unknown as IMPlatform + const rpc = new DialogRpc( + platform, session, store, undefined, undefined, 1, undefined, undefined, undefined, + undefined, undefined, undefined, undefined, undefined, undefined, 1, undefined, + undefined, undefined, undefined, undefined, scheduled, + ) + await rpc.getContacts() + const aliceId = rpc.peerTlId('alice') + const peer = { _: 'inputPeerUser' as const, userId: aliceId, accessHash: Long.ZERO } + const created = await rpc.sendMessage({ + _: 'messages.sendMessage', peer, message: 'send this later', + randomId: Long.fromNumber(41), scheduleDate: nowSeconds() + 120, + }) + expect(created).toMatchObject({ + _: 'updates', + updates: [{ _: 'updateNewScheduledMessage', message: { message: 'send this later' } }], + }) + + const listed = await rpc.getScheduledHistory({ + _: 'messages.getScheduledHistory', peer, hash: Long.ZERO, + }) + expect(listed).toMatchObject({ _: 'messages.messages' }) + const listedMessages = 'messages' in listed ? listed.messages : [] + expect(listedMessages).toHaveLength(1) + const scheduledMessage = listedMessages[0] as tl.RawMessage + expect(scheduledMessage).toMatchObject({ message: 'send this later', out: true }) + + const row = (await scheduled.list(session.platformSessionId, 'alice'))[0] + const delivered = await rpc.deliverScheduled(row) + expect(lastInput?.parts[0]).toMatchObject({ type: 'text', text: 'send this later' }) + expect(delivered.sentTlMessageId).toBeGreaterThan(0) + + const deleted = await rpc.deleteScheduledMessages({ + _: 'messages.deleteScheduledMessages', peer, id: [scheduledMessage.id], + }) + expect(deleted).toMatchObject({ + _: 'updates', + updates: [{ _: 'updateDeleteScheduledMessages', messages: [scheduledMessage.id] }], + }) + expect(await scheduled.list(session.platformSessionId, 'alice')).toHaveLength(0) + }) + + it('sends a scheduled message immediately when asked', async () => { + const database = await createDatabase() + const store = new MessageStore(database) + const scheduled = new ScheduledMessageStore(database) + const session = { + platformSessionId: 'rpc-session-2', platformId: 'rpc-platform', userId: 'self', + credentials: {}, metadata: { firstName: 'Self' }, + } + let lastInput: IMMessageInput | undefined + const platform: IMPlatform = { + capabilities: { + history: true, + send: { text: true, images: false, files: false, mixed: false, maxTextLength: 4096, maxMedia: 0 }, + conversations: { groups: false, channels: false, subchannels: false }, + }, + async subscribe() { return () => {} }, + async sendMessage(_session, conversation, content) { + lastInput = content + return { + id: String(Date.now()), conversationId: conversation.id, senderId: 'self', + timestamp: Math.floor(Date.now() / 1_000), outgoing: true, content, + } + }, + async getDialogs() { + return { + dialogs: [{ + conversation: { id: 'alice', kind: 'direct' as const, title: 'Alice' }, + unreadCount: 0, + }], + } + }, + async getHistory() { return { messages: [] } }, + async getUser(_session, id) { return { id, firstName: id } }, + async getContacts() { return { users: [{ id: 'alice', firstName: 'Alice' }] } }, + } as unknown as IMPlatform + const rpc = new DialogRpc( + platform, session, store, undefined, undefined, 1, undefined, undefined, undefined, + undefined, undefined, undefined, undefined, undefined, undefined, 1, undefined, + undefined, undefined, undefined, undefined, scheduled, + ) + await rpc.getContacts() + const peer = { _: 'inputPeerUser' as const, userId: rpc.peerTlId('alice'), accessHash: Long.ZERO } + await rpc.sendMessage({ + _: 'messages.sendMessage', peer, message: 'right now', + randomId: Long.fromNumber(77), scheduleDate: nowSeconds() + 600, + }) + const listed = await rpc.getScheduledHistory({ + _: 'messages.getScheduledHistory', peer, hash: Long.ZERO, + }) + const listedMessages = 'messages' in listed ? listed.messages : [] + const scheduledId = (listedMessages[0] as tl.RawMessage).id + const sent = await rpc.sendScheduledMessages({ + _: 'messages.sendScheduledMessages', peer, id: [scheduledId], + }) + expect(lastInput?.parts[0]).toMatchObject({ type: 'text', text: 'right now' }) + expect((sent._ === 'updates' ? sent.updates : []) + .some((update) => update._ === 'updateDeleteScheduledMessages')).toBe(true) + expect(await scheduled.list(session.platformSessionId, 'alice')).toHaveLength(0) + }) +}) diff --git a/packages/bridge/src/scheduled-messages.ts b/packages/bridge/src/scheduled-messages.ts new file mode 100644 index 00000000..9b70ee75 --- /dev/null +++ b/packages/bridge/src/scheduled-messages.ts @@ -0,0 +1,466 @@ +import type { Database } from '@cordisjs/plugin-database' +import { CronJob } from 'cron' +import type { + ScheduledMessageRow, ScheduledMessageStateRow, +} from './models.js' +import type { IMConversationKind, JsonValue } from './platform.js' + +/** How long one dispatcher may hold a row before another pass may steal it. */ +export const SCHEDULED_MESSAGE_LEASE_MS = 60_000 +/** + * Node timers cannot wait longer than 2^31-1 ms (about 24.8 days) and fire + * immediately when they overflow, so long schedule dates are truncated and + * re-armed when the truncated timer wakes. + */ +export const SCHEDULED_MESSAGE_MAX_DELAY_MS = 24 * 60 * 60 * 1_000 +/** Retry delays after a failed send; the sixth failure parks the row. */ +export const SCHEDULED_MESSAGE_RETRY_MS = [ + 30_000, 120_000, 600_000, 3_600_000, 21_600_000, 86_400_000, +] +/** How often the safety sweep looks for rows the timer missed. */ +export const SCHEDULED_MESSAGE_FALLBACK_MS = 120_000 +/** Rows claimed by one pass; keeps one burst from blocking the event loop. */ +export const SCHEDULED_MESSAGE_BATCH = 50 + +export type ScheduledMessageStatus = ScheduledMessageStateRow['status'] + +/** One scheduled message joined with its scheduling state. */ +export interface ScheduledMessage extends ScheduledMessageRow { + dueAt: number + status: ScheduledMessageStatus + attempts: number + nextAttemptAt: number + leaseOwner: string | null + leaseUntil: number | null + randomId: string + sentTlMessageId: number | null + lastError: string | null +} + +export interface CreateScheduledMessageInput { + platformSessionId: string + platformConversationId: string + conversationKind: IMConversationKind + tlMessageId: number + text: string + entities?: JsonValue[] | null + replyToMsgId?: number + /** Absolute send time in Unix seconds. */ + dueAt: number + randomId: string +} + +/** + * Durable half of scheduled messages. + * + * The message body and its scheduling state live in separate tables: editing a + * body never touches the state machine, and retries/leases never rewrite the + * body. + */ +export class ScheduledMessageStore { + constructor(private readonly _database: Database) {} + + async create(input: CreateScheduledMessageInput): Promise { + const now = new Date() + return this._database.withTransaction(async (database) => { + const message = await database.create('mtproto_scheduled_message', { + platformSessionId: input.platformSessionId, + platformConversationId: input.platformConversationId, + conversationKind: input.conversationKind, + tlMessageId: input.tlMessageId, + text: input.text, + entities: input.entities ?? null, + replyToMsgId: input.replyToMsgId ?? null, + createdAt: now, + }) + await database.create('mtproto_scheduled_state', { + messageId: message.id, + dueAt: input.dueAt, + status: 'pending', + attempts: 0, + nextAttemptAt: input.dueAt, + leaseOwner: null, + leaseUntil: null, + randomId: input.randomId, + sentTlMessageId: null, + lastError: null, + updatedAt: now, + }) + const [row] = await this._join(database, { messageId: message.id }) + return row + }) + } + + async list( + platformSessionId: string, + platformConversationId: string, + ): Promise { + const states = await this._database.select('mtproto_scheduled_state').execute() + const rows = await this._joinRows(states.filter((state) => + state.status !== 'sent' && state.status !== 'cancelled') as unknown as ScheduledMessageStateRow[]) + return rows.filter((row) => row.platformSessionId === platformSessionId + && row.platformConversationId === platformConversationId) + } + + async byTlIds( + platformSessionId: string, + platformConversationId: string, + tlMessageIds: readonly number[], + ): Promise { + const wanted = new Set(tlMessageIds) + return (await this.list(platformSessionId, platformConversationId)) + .filter((row) => wanted.has(row.tlMessageId)) + } + + async get(messageId: number): Promise { + const [row] = await this._join(this._database, { messageId }) + return row + } + + /** Earliest row that may still be sent, across every session. */ + async nextPending(now = Math.floor(Date.now() / 1_000)): Promise { + const states = await this._database.select('mtproto_scheduled_state', { status: 'pending' }) + .execute() + const rows = await this._joinRows(states as unknown as ScheduledMessageStateRow[]) + return rows.sort((left, right) => left.nextAttemptAt - right.nextAttemptAt)[0] + } + + /** Rows due at `now`, oldest first. */ + async due(now: number, limit = SCHEDULED_MESSAGE_BATCH): Promise { + const states = await this._database.select('mtproto_scheduled_state', { status: 'pending' }) + .execute() + const rows = (await this._joinRows(states as unknown as ScheduledMessageStateRow[])) + .filter((row) => row.nextAttemptAt <= now) + return rows.sort((left, right) => left.nextAttemptAt - right.nextAttemptAt).slice(0, limit) + } + + /** + * Atomically move one due row to `sending`. + * + * A single conditional UPDATE is the portable lock: SQLite has no + * `SELECT ... FOR UPDATE SKIP LOCKED`, and this statement is atomic on every + * supported driver. `modified === 1` means this process won the row. + */ + async claim( + messageId: number, + owner: string, + now: number, + leaseMs = SCHEDULED_MESSAGE_LEASE_MS, + ): Promise { + // `@cordisjs/plugin-database` implements `set()` as SELECT-then-UPDATE, so + // two in-process claimers can both observe the same row. Serialize claims + // per database instance; the lease columns still guard crashed senders. + return serializeClaim(this._database, () => this._claimUnlocked(messageId, owner, now, leaseMs)) + } + + private async _claimUnlocked( + messageId: number, + owner: string, + now: number, + leaseMs: number, + ): Promise { + const claimLease = { + status: 'sending' as const, + leaseOwner: owner, + leaseUntil: now + Math.ceil(leaseMs / 1_000), + updatedAt: new Date(), + } + // Two single-statement conditional updates instead of one `$or`: each is a + // plain `UPDATE ... WHERE` and therefore atomic on every driver. + const due = await this._database.set('mtproto_scheduled_state', { + messageId, status: 'pending', nextAttemptAt: { $lte: now }, + }, claimLease) + if (writeCount(due) > 0) return true + const expired = await this._database.set('mtproto_scheduled_state', { + messageId, status: 'sending', leaseUntil: { $lt: now }, + }, claimLease) + return writeCount(expired) > 0 + } + + async markSent(messageId: number, sentTlMessageId?: number): Promise { + await this._database.set('mtproto_scheduled_state', { messageId }, { + status: 'sent', + sentTlMessageId: sentTlMessageId ?? null, + leaseOwner: null, + leaseUntil: null, + lastError: null, + updatedAt: new Date(), + }) + } + + /** Put a row back for a later attempt (or park it after too many). */ + async reschedule( + messageId: number, + attempts: number, + nextAttemptAt: number, + status: ScheduledMessageStatus, + error?: string, + ): Promise { + await this._database.set('mtproto_scheduled_state', { messageId }, { + status, + attempts, + nextAttemptAt, + leaseOwner: null, + leaseUntil: null, + lastError: error ?? null, + updatedAt: new Date(), + }) + } + + async cancel(platformSessionId: string, platformConversationId: string, tlMessageId: number): Promise { + const [message] = await this._database.get('mtproto_scheduled_message', { + platformSessionId, platformConversationId, tlMessageId, + }) + if (!message) return false + let modified = 0 + for (const status of ['pending', 'sending', 'failed'] as const) { + const result = await this._database.set('mtproto_scheduled_state', { + messageId: message.id, status, + }, { + status: 'cancelled', leaseOwner: null, leaseUntil: null, updatedAt: new Date(), + }) + modified += writeCount(result) + } + return modified > 0 + } + + /** Rows whose lease expired while their sender was running (crash recovery). */ + async requeueExpired(now: number): Promise { + const result = await this._database.set('mtproto_scheduled_state', { + status: 'sending', + leaseUntil: { $lt: now }, + }, { + status: 'pending', leaseOwner: null, leaseUntil: null, updatedAt: new Date(), + }) + return writeCount(result) + } + + private async _join( + database: Database, + query: { messageId: number }, + ): Promise { + const [state] = await database.get('mtproto_scheduled_state', query) + if (!state) return [] + return this._joinRows([state as unknown as ScheduledMessageStateRow]) + } + + private async _joinRows(states: ScheduledMessageStateRow[]): Promise { + if (!states.length) return [] + const ids = new Set(states.map((state) => state.messageId)) + const byId = new Map((await this._database.select('mtproto_scheduled_message').execute()) + .filter((message) => ids.has(message.id)) + .map((message) => [message.id, message])) + return states.flatMap((state): ScheduledMessage[] => { + const message = byId.get(state.messageId) + if (!message) return [] + return [{ + ...message, + dueAt: state.dueAt, + status: state.status, + attempts: state.attempts, + nextAttemptAt: state.nextAttemptAt, + leaseOwner: state.leaseOwner, + leaseUntil: state.leaseUntil, + randomId: state.randomId, + sentTlMessageId: state.sentTlMessageId, + lastError: state.lastError, + }] + }) + } +} + +const claimLocks = new WeakMap>() + +/** Run `fn` after every earlier claim on the same database settled. */ +async function serializeClaim(key: object, fn: () => Promise): Promise { + const previous = claimLocks.get(key) ?? Promise.resolve() + let release!: () => void + const current = new Promise((resolve) => { release = resolve }) + claimLocks.set(key, previous.then(() => current)) + await previous + try { + return await fn() + } finally { + release() + } +} + +export interface ScheduledDispatcherOptions { + owner?: string + fallbackMs?: number + maxDelayMs?: number + leaseMs?: number + now?: () => number + /** Injected in tests; production uses `cron` + `setTimeout`. */ + schedule?: (dueMs: number, callback: () => void) => () => void +} + +/** + * Arms one timer for the earliest due message and falls back to a periodic + * sweep. + * + * Polling every second is unnecessary: the timer fires exactly at the due + * second, and the sweep exists for restart, clock jumps and a throwing + * callback. Both paths go through `claim()`, so they cannot double-send. + */ +export class ScheduledMessageDispatcher { + private readonly _owner: string + private readonly _fallbackMs: number + private readonly _maxDelayMs: number + private readonly _leaseMs: number + private readonly _now: () => number + private readonly _schedule: (dueMs: number, callback: () => void) => () => void + private _cancelTimer?: () => void + private _fallback?: ReturnType + private _armedAt?: number + private _running = false + private _stopped = false + + constructor( + private readonly _store: ScheduledMessageStore, + private readonly _send: (message: ScheduledMessage) => Promise, + options: ScheduledDispatcherOptions = {}, + ) { + this._owner = options.owner ?? `scheduler-${process.pid}` + this._fallbackMs = options.fallbackMs ?? SCHEDULED_MESSAGE_FALLBACK_MS + this._maxDelayMs = options.maxDelayMs ?? SCHEDULED_MESSAGE_MAX_DELAY_MS + this._leaseMs = options.leaseMs ?? SCHEDULED_MESSAGE_LEASE_MS + this._now = options.now ?? (() => Date.now()) + this._schedule = options.schedule ?? defaultSchedule() + } + + async start(): Promise { + if (this._running) return + this._stopped = false + this._running = true + await this._store.requeueExpired(Math.floor(this._now() / 1_000)) + await this.rescan() + this._fallback = setInterval(() => { + void this.sweep().catch(() => undefined) + }, this._fallbackMs) + this._fallback.unref?.() + } + + stop(): void { + this._stopped = true + this._running = false + this._clearTimer() + if (this._fallback) clearInterval(this._fallback) + this._fallback = undefined + } + + /** + * Re-arm only when the new earliest row is earlier than the armed one. + * + * Re-arming on every change is harmless but wakes the process for nothing; + * equal or later dates leave the current timer alone. + */ + async rescan(): Promise { + if (this._stopped) return + const next = await this._store.nextPending(this._dueNow()) + if (!next) { + this._clearTimer() + this._armedAt = undefined + return + } + if (this._cancelTimer && this._armedAt !== undefined && this._armedAt <= next.nextAttemptAt) return + this._arm(next) + } + + /** Claim and send everything that is due, then re-arm. */ + async sweep(): Promise { + if (this._stopped) return + this._clearTimer() + this._armedAt = undefined + const now = this._dueNow() + const due = await this._store.due(now) + for (const message of due) { + if (this._stopped) break + await this._dispatch(message, now) + } + await this.rescan() + } + + private async _dispatch(message: ScheduledMessage, now: number): Promise { + const claimed = await this._store.claim(message.id, this._owner, now, this._leaseMs) + if (!claimed) return + try { + const sentTlMessageId = await this._send(message) + await this._store.markSent(message.id, sentTlMessageId) + } catch (error) { + const attempts = message.attempts + 1 + const retry = SCHEDULED_MESSAGE_RETRY_MS[Math.min(attempts - 1, SCHEDULED_MESSAGE_RETRY_MS.length - 1)] + const retryAt = now + Math.ceil(retry / 1_000) + const parked = attempts > SCHEDULED_MESSAGE_RETRY_MS.length + await this._store.reschedule( + message.id, attempts, parked ? now : retryAt, + parked ? 'failed' : 'pending', + error instanceof Error ? error.message : String(error), + ) + } + } + + private _arm(message: ScheduledMessage): void { + this._clearTimer() + this._armedAt = message.nextAttemptAt + const delay = message.nextAttemptAt * 1_000 - this._now() + if (delay > this._maxDelayMs) { + // Re-arm when the truncated wait expires; the row is still in the future. + const timer = setTimeout(() => { void this.sweep() }, this._maxDelayMs) + timer.unref?.() + this._cancelTimer = () => clearTimeout(timer) + return + } + this._cancelTimer = this._schedule(this._now() + Math.max(0, delay), () => { + this._cancelTimer = undefined + this._armedAt = undefined + void this.sweep() + }) + } + + private _clearTimer(): void { + this._cancelTimer?.() + this._cancelTimer = undefined + this._armedAt = undefined + } + + private _dueNow(): number { + return Math.floor(this._now() / 1_000) + } +} + +/** + * Clamp one timer wait to what Node can represent. + * + * `setTimeout` fires immediately when the delay exceeds 2^31-1 ms, so a date + * months away must be cut down and re-armed when the truncated wait expires. + */ +export function clampScheduleDelay(delayMs: number, maxMs = SCHEDULED_MESSAGE_MAX_DELAY_MS): number { + if (!Number.isFinite(delayMs)) return maxMs + return Math.min(Math.max(0, Math.trunc(delayMs)), maxMs) +} + +/** Drivers report a conditional UPDATE as `matched` or `modified`. */ +function writeCount(result: { modified?: number, matched?: number }): number { + return result.modified ?? result.matched ?? 0 +} + +function defaultSchedule(): (dueMs: number, callback: () => void) => () => void { + return (dueMs, callback) => { + const delay = clampScheduleDelay(dueMs - Date.now()) + // Very short waits are not worth a cron job; `cron` also rounds dates to + // whole seconds, which would make sub-second tests and immediate rows late. + if (delay < 1_000) { + const timer = setTimeout(callback, Math.max(0, delay)) + timer.unref?.() + return () => clearTimeout(timer) + } + const job = CronJob.from({ + cronTime: new Date(dueMs), + onTick: callback, + start: true, + unrefTimeout: true, + }) + return () => job.stop() + } +} diff --git a/yarn.lock b/yarn.lock index 1275ed62..5c7b3ea7 100644 --- a/yarn.lock +++ b/yarn.lock @@ -1886,6 +1886,7 @@ __metadata: "@testing-library/react": "npm:^16.3.0" "@types/react": "npm:^19.2.0" "@types/react-dom": "npm:^19.2.0" + cron: "npm:^4.4.0" happy-dom: "npm:^20.11.2" jsqr: "npm:^1.4.0" linkify-it: "npm:^6.1.0" @@ -2950,6 +2951,13 @@ __metadata: languageName: node linkType: hard +"@types/luxon@npm:~3.7.0": + version: 3.7.6 + resolution: "@types/luxon@npm:3.7.6" + checksum: 10c0/135c55910d8f4fb615cbfaa93b173e57e3eed9731da683005acdb84dba5ece5dbbf56cc82330015ccf5247a6e84fefbb2e2ebe255c957636f604f79efacb4214 + languageName: node + linkType: hard + "@types/node@npm:*": version: 26.1.1 resolution: "@types/node@npm:26.1.1" @@ -3550,6 +3558,16 @@ __metadata: languageName: node linkType: hard +"cron@npm:^4.4.0": + version: 4.4.0 + resolution: "cron@npm:4.4.0" + dependencies: + "@types/luxon": "npm:~3.7.0" + luxon: "npm:~3.7.0" + checksum: 10c0/e3762fda4a9a2404ac2e54e1104a3c6c695747e14b2321bbc5531aa1192c75a319075695fcaf68b6927e0ffd877b5454f314c680f6586db1a78b2c25b76b8a32 + languageName: node + linkType: hard + "cross-env@npm:^10.1.0": version: 10.1.0 resolution: "cross-env@npm:10.1.0" @@ -4834,6 +4852,13 @@ __metadata: languageName: node linkType: hard +"luxon@npm:~3.7.0": + version: 3.7.2 + resolution: "luxon@npm:3.7.2" + checksum: 10c0/ed8f0f637826c08c343a29dd478b00628be93bba6f068417b1d8896b61cb61c6deacbe1df1e057dbd9298334044afa150f9aaabbeb3181418ac8520acfdc2ae2 + languageName: node + linkType: hard + "lz-string@npm:^1.5.0": version: 1.5.0 resolution: "lz-string@npm:1.5.0"