fix(v2/telegram): await pairing interceptor work to serialize DB commits
The Telegram pairing interceptor fired DB writes (createMessagingGroup,
upsertUser, grantRole) and the pairing-success confirmation inside an
unawaited `void (async () => {...})()`. Recent changes (0d3326a user
privilege model, c483860 pairing confirmation) widened the work done
inside this closure to include an extra two DB writes and a Telegram
API round-trip, making the race between match and commit reproducible
— a paired message could appear "lost" until a second send.
Change onInbound to optionally return a Promise, await it in the
chat-sdk-bridge dispatch callbacks, and make the pairing interceptor
async so its DB writes + confirmation send complete before the handler
resolves.
Note: the upstream @chat-adapter/telegram SDK itself does not await
processUpdate in its polling loop, so the adapter's getUpdates offset
still advances before our handler resolves. A true restart-safe fix
needs a corresponding change in chat-adapter.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -20,7 +20,11 @@ export interface ChannelSetup {
|
|||||||
conversations: ConversationConfig[];
|
conversations: ConversationConfig[];
|
||||||
|
|
||||||
/** Called when an inbound message arrives from the platform. */
|
/** Called when an inbound message arrives from the platform. */
|
||||||
onInbound(platformId: string, threadId: string | null, message: InboundMessage): void;
|
onInbound(
|
||||||
|
platformId: string,
|
||||||
|
threadId: string | null,
|
||||||
|
message: InboundMessage,
|
||||||
|
): void | Promise<void>;
|
||||||
|
|
||||||
/** Called when the adapter discovers metadata about a conversation. */
|
/** Called when the adapter discovers metadata about a conversation. */
|
||||||
onMetadata(platformId: string, name?: string, isGroup?: boolean): void;
|
onMetadata(platformId: string, name?: string, isGroup?: boolean): void;
|
||||||
|
|||||||
@@ -165,20 +165,20 @@ export function createChatSdkBridge(config: ChatSdkBridgeConfig): ChannelAdapter
|
|||||||
// Subscribed threads — forward all messages
|
// Subscribed threads — forward all messages
|
||||||
chat.onSubscribedMessage(async (thread, message) => {
|
chat.onSubscribedMessage(async (thread, message) => {
|
||||||
const channelId = adapter.channelIdFromThreadId(thread.id);
|
const channelId = adapter.channelIdFromThreadId(thread.id);
|
||||||
setupConfig.onInbound(channelId, thread.id, await messageToInbound(message));
|
await setupConfig.onInbound(channelId, thread.id, await messageToInbound(message));
|
||||||
});
|
});
|
||||||
|
|
||||||
// @mention in unsubscribed thread — forward + subscribe
|
// @mention in unsubscribed thread — forward + subscribe
|
||||||
chat.onNewMention(async (thread, message) => {
|
chat.onNewMention(async (thread, message) => {
|
||||||
const channelId = adapter.channelIdFromThreadId(thread.id);
|
const channelId = adapter.channelIdFromThreadId(thread.id);
|
||||||
setupConfig.onInbound(channelId, thread.id, await messageToInbound(message));
|
await setupConfig.onInbound(channelId, thread.id, await messageToInbound(message));
|
||||||
await thread.subscribe();
|
await thread.subscribe();
|
||||||
});
|
});
|
||||||
|
|
||||||
// DMs — always forward + subscribe
|
// DMs — always forward + subscribe
|
||||||
chat.onDirectMessage(async (thread, message) => {
|
chat.onDirectMessage(async (thread, message) => {
|
||||||
const channelId = adapter.channelIdFromThreadId(thread.id);
|
const channelId = adapter.channelIdFromThreadId(thread.id);
|
||||||
setupConfig.onInbound(channelId, null, await messageToInbound(message));
|
await setupConfig.onInbound(channelId, null, await messageToInbound(message));
|
||||||
await thread.subscribe();
|
await thread.subscribe();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
@@ -115,8 +115,8 @@ function createPairingInterceptor(
|
|||||||
hostOnInbound: ChannelSetup['onInbound'],
|
hostOnInbound: ChannelSetup['onInbound'],
|
||||||
token: string,
|
token: string,
|
||||||
): ChannelSetup['onInbound'] {
|
): ChannelSetup['onInbound'] {
|
||||||
return (platformId, threadId, message) => {
|
return async (platformId, threadId, message) => {
|
||||||
void (async () => {
|
try {
|
||||||
const botUsername = await botUsernamePromise;
|
const botUsername = await botUsernamePromise;
|
||||||
if (!botUsername) {
|
if (!botUsername) {
|
||||||
hostOnInbound(platformId, threadId, message);
|
hostOnInbound(platformId, threadId, message);
|
||||||
@@ -187,11 +187,11 @@ function createPairingInterceptor(
|
|||||||
});
|
});
|
||||||
|
|
||||||
await sendPairingConfirmation(token, platformId);
|
await sendPairingConfirmation(token, platformId);
|
||||||
})().catch((err) => {
|
} catch (err) {
|
||||||
log.error('Telegram pairing interceptor error', { err });
|
log.error('Telegram pairing interceptor error', { err });
|
||||||
// Fail open: pass through so a pairing bug doesn't break normal traffic.
|
// Fail open: pass through so a pairing bug doesn't break normal traffic.
|
||||||
hostOnInbound(platformId, threadId, message);
|
hostOnInbound(platformId, threadId, message);
|
||||||
});
|
}
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user