Compare commits

..

2 Commits

Author SHA1 Message Date
Renato Alcara 0bc9f4a34e fix: flatten carousel thumbnail fallback 2026-03-16 22:07:28 -03:00
Renato Alcara 2d07ddcc7a fix: harden carousel live rendering path 2026-03-16 21:57:28 -03:00
56 changed files with 2915 additions and 11065 deletions
+3 -268
View File
@@ -1,7 +1,7 @@
syntax = "proto3";
package proto;
/// WhatsApp Version: 2.3000.1038569661
/// WhatsApp Version: 2.3000.1035216863
message ADVDeviceIdentity {
optional uint32 rawId = 1;
@@ -14,7 +14,6 @@ message ADVDeviceIdentity {
enum ADVEncryptionType {
E2EE = 0;
HOSTED = 1;
NON_E2EE = 2;
}
message ADVKeyIndexList {
optional uint32 rawId = 1;
@@ -62,7 +61,6 @@ message AIHomeState {
ANIMATE_PHOTO = 2;
ANALYZE_FILE = 3;
COLLABORATE = 4;
OPEN_GREETING_CARD = 5;
}
}
@@ -249,16 +247,6 @@ message AIRichResponseUnifiedResponse {
optional bytes data = 1;
}
enum AISubscriptionRequestType {
UNSPECIFIED = 0;
THINK_HARD = 1;
IMAGE_GEN = 2;
VIDEO_GEN = 3;
}
message AISubscriptionUpsellMetadata {
optional AISubscriptionRequestType requestType = 1;
}
message AIThreadInfo {
optional AIThreadServerInfo serverInfo = 1;
optional AIThreadClientInfo clientInfo = 2;
@@ -429,16 +417,9 @@ message BotCapabilityMetadata {
JSON_PATCH_STREAMING = 58;
AI_TAB_FORCE_CLIPPY = 59;
UNIFIED_RESPONSE_EMBEDDED_SCREENS = 60;
AI_SUBSCRIPTION_ENABLED = 61;
}
}
message BotCommandMetadata {
optional string commandName = 1;
optional string commandDescription = 2;
optional string commandPrompt = 3;
}
message BotDocumentMessageMetadata {
optional DocumentPluginType pluginType = 1;
enum DocumentPluginType {
@@ -670,9 +651,6 @@ message BotMetadata {
optional BotRenderingConfigMetadata botRenderingConfigMetadata = 36;
optional BotInfrastructureDiagnostics botInfrastructureDiagnostics = 37;
optional AIMediaCollectionMetadata aiMediaCollectionMetadata = 38;
optional BotCommandMetadata commandMetadata = 39;
optional BotResolvedToolCallMetadata resolvedToolCallMetadata = 40;
optional AISubscriptionUpsellMetadata subscriptionUpsellMetadata = 41;
optional bytes internalMetadata = 999;
}
@@ -724,7 +702,6 @@ enum BotMetricsEntryPoint {
WEB_NAVIGATION_BAR = 47;
GROUP_MEMBER = 54;
CHATLIST_SEARCH = 55;
NEW_CHAT_LIST = 56;
}
message BotMetricsMetadata {
optional string destinationId = 1;
@@ -911,11 +888,6 @@ message BotRenderingMetadata {
}
message BotResolvedToolCallMetadata {
optional string toolCallId = 1;
optional string resolutionDataSerialized = 2;
}
message BotSessionMetadata {
optional string sessionId = 1;
optional BotSessionSource sessionSource = 2;
@@ -1151,7 +1123,6 @@ message ClientPayload {
optional int32 preacksCount = 45;
optional int32 processingQueueSize = 46;
repeated string pairedPeripherals = 47;
optional bytes testIsolationId = 48;
enum AccountType {
DEFAULT = 0;
GUEST = 1;
@@ -1247,7 +1218,6 @@ message ClientPayload {
optional string deviceExpId = 14;
optional DeviceType deviceType = 15;
optional string deviceModelType = 16;
optional DistributionChannel distributionChannel = 17;
message AppVersion {
optional uint32 primary = 1;
optional uint32 secondary = 2;
@@ -1263,12 +1233,6 @@ message ClientPayload {
WEARABLE = 3;
VR = 4;
}
enum DistributionChannel {
APPSTORE = 0;
WEBSITE = 1;
TESTFLIGHT = 2;
INTERNAL = 3;
}
enum Platform {
ANDROID = 0;
IOS = 1;
@@ -1438,9 +1402,6 @@ message ContextInfo {
optional MediaDomainInfo mediaDomainInfo = 74;
optional PartiallySelectedContent partiallySelectedContent = 75;
optional uint32 afterReadDuration = 76;
optional CrossAppSource crossAppSource = 77;
optional BusinessInteractionPills businessInteractionPills = 78;
optional string posterStatusId = 79;
message AdReplyInfo {
optional string advertiserName = 1;
optional MediaType mediaType = 2;
@@ -1453,47 +1414,10 @@ message ContextInfo {
}
}
message BusinessInteractionPills {
optional string businessJid = 1;
repeated Pill pills = 2;
optional EntryPoint entryPoint = 3;
enum EntryPoint {
ENTRY_POINT_UNKNOWN = 0;
P2P_LINK_SHARE = 1;
CONTACT_CARD_SHARING = 2;
PHONE_NUMBER = 3;
STATUS = 4;
IN_THREAD_CONTEXT_CARD = 5;
}
message Pill {
optional ContextInfo.BusinessInteractionPills.PillType pillType = 1;
optional string actionUrl = 2;
}
enum PillType {
UNKNOWN = 0;
VIEW_BUSINESS = 1;
CHAT = 2;
CALL = 3;
CATALOG = 4;
CHANNEL = 5;
BOOK_APPOINTMENT = 6;
OFFERS = 7;
BESTSELLERS = 8;
MENU = 9;
ABOUT = 10;
}
}
message BusinessMessageForwardInfo {
optional string businessOwnerJid = 1;
}
enum CrossAppSource {
CROSS_APP_SOURCE_UNKNOWN = 0;
CROSS_APP_SOURCE_INSTAGRAM = 1;
CROSS_APP_SOURCE_FACEBOOK = 2;
}
message DataSharingContext {
optional bool showMmDisclosure = 1;
optional string encryptedSignalTokenConsented = 2;
@@ -1542,10 +1466,6 @@ message ContextInfo {
optional string wtwaWebsiteUrl = 26;
optional string adPreviewUrl = 27;
optional bool containsCtwaFlowsAutoReply = 28;
optional int32 agmThumbnailStrategy = 29;
optional int32 agmTitleStrategy = 30;
optional int32 agmSubtitleStrategy = 31;
optional int32 agmHeaderInteractionStrategy = 32;
enum AdType {
CTWA = 0;
CAWC = 1;
@@ -1702,7 +1622,6 @@ message Conversation {
optional bool isMarketingMessageThread = 55;
optional bool isSenderNewAccount = 56;
optional uint32 afterReadDuration = 57;
optional bool isSenderSuspicious = 58;
enum EndOfHistoryTransferType {
COMPLETE_BUT_MORE_MESSAGES_REMAIN_ON_PRIMARY = 0;
COMPLETE_AND_NO_MORE_MESSAGE_REMAIN_ON_PRIMARY = 1;
@@ -1732,7 +1651,6 @@ message DeviceCapabilities {
optional bool companionSupportEnabled = 2;
optional bool campaignSyncEnabled = 3;
optional bool insightsSyncEnabled = 4;
optional int32 recipientLimit = 5;
}
enum ChatLockSupportLevel {
@@ -1808,8 +1726,6 @@ message DeviceProps {
optional uint32 initialSyncMaxMessagesPerChat = 20;
optional bool supportManusHistory = 21;
optional bool supportHatchHistory = 22;
repeated string supportedBotChannelFbids = 23;
optional bool supportInlineContacts = 24;
}
enum PlatformType {
@@ -1996,16 +1912,6 @@ message GroupParticipant {
}
}
message GroupRootKeyShare {
repeated GroupRootKeyShareEntry keys = 1;
}
message GroupRootKeyShareEntry {
optional bytes groupRootKey = 1;
optional string keyId = 2;
optional int64 expiryTimestampMs = 3;
}
message HandshakeMessage {
optional ClientHello clientHello = 2;
optional ServerHello serverHello = 3;
@@ -2027,28 +1933,14 @@ message HandshakeMessage {
optional bytes paddedBytes = 6;
optional bool sendServerHelloPaddedBytes = 7;
optional bool simulateXxkemFs = 8;
optional HandshakeMessage.HandshakePqMode pqMode = 9;
optional bytes extendedEphemeral = 10;
}
enum HandshakePqMode {
HANDSHAKE_PQ_MODE_UNKNOWN = 0;
XXKEM = 1;
XXKEM_FS = 2;
WA_CLASSICAL = 3;
WA_PQ = 4;
IKKEM = 5;
IKKEM_FS = 6;
XXKEM_2 = 7;
IKKEM_2 = 8;
}
message ServerHello {
optional bytes ephemeral = 1;
optional bytes static = 2;
optional bytes payload = 3;
optional bytes extendedStatic = 4;
optional bytes paddingBytes = 5;
optional bytes extendedCiphertext = 6;
}
}
@@ -2072,8 +1964,6 @@ message HistorySync {
optional bytes shareableChatIdentifierEncryptionKey = 17;
repeated Account accounts = 18;
optional bytes nctSalt = 19;
repeated InlineContact inlineContacts = 20;
optional bool inlineContactsProvided = 21;
enum BotAIWaitListState {
IN_WAITLIST = 0;
AI_AVAILABLE = 1;
@@ -2168,14 +2058,6 @@ message InThreadSurveyMetadata {
}
message InlineContact {
optional string pnJid = 1;
optional string lidJid = 2;
optional string fullName = 3;
optional string firstName = 4;
optional string username = 5;
}
message InteractiveAnnotation {
repeated Point polygonVertices = 1;
optional bool shouldSkipConfirmation = 4;
@@ -2254,6 +2136,7 @@ message LimitSharing {
CHAT_SETTING = 1;
BIZ_SUPPORTS_FB_HOSTING = 2;
UNKNOWN_GROUP = 3;
DEPRECATION = 4;
}
}
@@ -2415,9 +2298,6 @@ message Message {
optional ConditionalRevealMessage conditionalRevealMessage = 120;
optional PollAddOptionMessage pollAddOptionMessage = 121;
optional EventInviteMessage eventInviteMessage = 122;
optional GroupRootKeyShare groupRootKeyShare = 123;
optional P2PPaymentReminderNotification p2PPaymentReminderNotification = 124;
optional SplitPaymentMessage splitPaymentMessage = 125;
message AlbumMessage {
optional uint32 expectedImageCount = 2;
optional uint32 expectedVideoCount = 3;
@@ -2598,41 +2478,6 @@ message Message {
optional string id = 2;
}
message ChatCustomImageWallpaper {
optional string directPath = 1;
optional bytes mediaKey = 2;
optional bytes fileEncSha256 = 3;
optional bytes fileSha256 = 4;
optional float dimLevel = 5;
}
message ChatDefaultWallpaper {
optional bool isDoodleEnabled = 1;
}
message ChatSolidColorWallpaper {
optional string colorLight = 1;
optional string colorDark = 2;
optional bool isDoodleEnabled = 3;
}
message ChatStockImageWallpaper {
optional string stockImageId = 1;
optional float dimLevel = 2;
}
message ChatThemeSetting {
optional int64 settingTimestampMs = 1;
optional bool clearTheme = 2;
optional string colorSchemeId = 3;
oneof wallpaper {
Message.ChatDefaultWallpaper defaultWallpaper = 10;
Message.ChatSolidColorWallpaper solidColor = 11;
Message.ChatStockImageWallpaper stockImage = 12;
Message.ChatCustomImageWallpaper customImage = 13;
}
}
message CloudAPIThreadControlNotification {
optional CloudAPIThreadControl status = 1;
optional int64 senderNotificationTimestampMs = 2;
@@ -2644,7 +2489,6 @@ message Message {
UNKNOWN = 0;
CONTROL_PASSED = 1;
CONTROL_TAKEN = 2;
INFO = 3;
}
message CloudAPIThreadControlNotificationContent {
optional string handoffNotificationText = 1;
@@ -2742,7 +2586,6 @@ message Message {
optional int64 startTime = 5;
optional string caption = 6;
optional bool isCanceled = 7;
optional int64 endTime = 8;
}
message EventMessage {
@@ -3325,35 +3168,6 @@ message Message {
}
}
message P2PPaymentReminderNotification {
optional string reminderId = 1;
optional Money amount = 2;
optional ReminderFrequency frequency = 3;
optional int64 nextReminderTimestamp = 4;
optional int64 expiryTimestamp = 5;
optional ReminderState state = 6;
optional string description = 7;
optional string creatorJid = 8;
optional string receiverJid = 9;
optional string upiId = 10;
optional int64 createdTimestamp = 11;
enum ReminderFrequency {
UNKNOWN_FREQUENCY = 0;
WEEKLY = 1;
BIWEEKLY = 2;
MONTHLY = 3;
CUSTOM = 4;
}
enum ReminderState {
UNKNOWN_STATE = 0;
ACTIVE = 1;
PAUSED = 2;
STOPPED = 3;
EXPIRED = 4;
CANCELLED = 5;
}
}
message PaymentExtendedMetadata {
optional uint32 type = 1;
optional string platform = 2;
@@ -3456,7 +3270,6 @@ message Message {
optional int32 onDemandMsgCount = 4;
optional int64 oldestMsgTimestampMs = 5;
optional string accountLid = 6;
optional bool supportInlineResponse = 7;
}
message PlaceholderMessageResendRequest {
@@ -3543,7 +3356,6 @@ message Message {
ERROR_REQUEST_ON_NON_SMB_PRIMARY = 4;
ERROR_HOSTED_DEVICE_NOT_CONNECTED = 5;
ERROR_HOSTED_DEVICE_LOGIN_TIME_NOT_SET = 6;
ERROR_MULTI_PROVIDER_NOT_CONFIGURED = 7;
}
message HistorySyncChunkRetryResponse {
optional Message.HistorySyncType syncType = 1;
@@ -3763,7 +3575,6 @@ message Message {
optional MemberLabel memberLabel = 27;
optional AIMediaCollectionMessage aiMediaCollectionMessage = 28;
optional uint32 afterReadDuration = 29;
optional Message.ChatThemeSetting chatThemeSetting = 30;
enum Type {
REVOKE = 0;
EPHEMERAL_SETTING = 3;
@@ -3793,7 +3604,7 @@ message Message {
GROUP_MEMBER_LABEL_CHANGE = 30;
AI_MEDIA_COLLECTION_MESSAGE = 31;
MESSAGE_UNSCHEDULE = 32;
CHAT_THEME_SETTING = 34;
BOT_UNLINK_MESSAGE = 33;
}
}
@@ -3885,26 +3696,6 @@ message Message {
optional bytes axolotlSenderKeyDistributionMessage = 2;
}
message SplitPaymentMessage {
optional string splitId = 1;
optional Money totalAmount = 2;
optional string description = 3;
optional string requesterJid = 4;
repeated Message.SplitPaymentParticipant participants = 5;
optional int64 createdAtMs = 6;
optional ContextInfo contextInfo = 17;
}
message SplitPaymentParticipant {
optional string jid = 1;
optional Money amount = 2;
optional SplitPaymentStatus status = 3;
enum SplitPaymentStatus {
PENDING = 0;
PAID = 1;
}
}
message StatusNotificationMessage {
optional MessageKey responseMessageKey = 1;
optional MessageKey originalMessageKey = 2;
@@ -3964,7 +3755,6 @@ message Message {
optional bool isLottie = 21;
optional string accessibilityLabel = 22;
optional int32 premium = 24;
optional string emojis = 25;
}
message StickerPackMessage {
@@ -4185,7 +3975,6 @@ message MessageContextInfo {
optional LimitSharing limitSharingV2 = 14;
repeated ThreadID threadId = 15;
optional WebLinkRenderConfig weblinkRenderConfig = 16;
optional bytes teeBotMetadata = 17;
enum MessageAddonExpiryType {
STATIC = 1;
DEPENDENT_ON_PARENT = 2;
@@ -4378,8 +4167,6 @@ enum MutationProps {
CUSTOMER_DATA_ACTION = 83;
SUBSCRIPTIONS_SYNC_V2_ACTION = 84;
THREAD_PIN_ACTION = 85;
AUTO_ORGANIZE_BUSINESS_CHAT_SETTING = 86;
BIZ_AI_SETTINGS_NUDGE_ACTION = 87;
SHARE_OWN_PN = 10001;
BUSINESS_BROADCAST_ACTION = 10002;
AI_THREAD_DELETE_ACTION = 10003;
@@ -4693,12 +4480,6 @@ message ReportingTokenInfo {
optional bytes reportingTag = 1;
}
message ScheduledMessageMetadata {
optional string revealKeyId = 1;
optional bytes revealKey = 2;
optional uint64 scheduledTime = 3;
}
message SenderKeyDistributionMessage {
optional uint32 id = 1;
optional uint32 iteration = 2;
@@ -4855,7 +4636,6 @@ message StatusAttribution {
SHARECHAT = 9;
GOOGLE_PHOTOS = 10;
SOUNDCLOUD = 11;
SHAZAM = 12;
}
}
@@ -4913,7 +4693,6 @@ message StatusAttribution {
LAYOUTS = 8;
NEWSLETTER_STATUS = 9;
STATUS_CLOSE_SHARING = 10;
PAID_PARTNERSHIP = 11;
}
}
@@ -5027,8 +4806,6 @@ message SyncActionValue {
optional CustomerDataAction customerDataAction = 83;
optional SubscriptionsSyncV2Action subscriptionsSyncV2Action = 84;
optional ThreadPinAction threadPinAction = 85;
optional AutoOrganizeBusinessChatSetting autoOrganizeBusinessChatSetting = 86;
optional BizAISettingsNudgeAction bizAiSettingsNudgeAction = 87;
message AgentAction {
optional string name = 1;
optional int32 deviceID = 2;
@@ -5048,10 +4825,6 @@ message SyncActionValue {
optional SyncActionValue.SyncActionMessageRange messageRange = 2;
}
message AutoOrganizeBusinessChatSetting {
optional bool autoOrganize = 1;
}
message AvatarUpdatedAction {
optional AvatarEventType eventType = 1;
repeated SyncActionValue.StickerAction recentAvatarStickers = 2;
@@ -5062,20 +4835,6 @@ message SyncActionValue {
}
}
message BizAISettingsNudgeAction {
optional BizAISettingsCategory category = 1;
optional int64 version = 2;
optional int64 updatedAtMs = 3;
enum BizAISettingsCategory {
UNKNOWN = 0;
INSTRUCTIONS = 1;
RESPONSE_SETTINGS = 2;
EXAMPLE_RESPONSES = 3;
KNOWLEDGE = 4;
LEAD_GEN = 5;
}
}
message BotWelcomeRequestAction {
optional bool isSent = 1;
}
@@ -5121,7 +4880,6 @@ message SyncActionValue {
repeated SyncActionValue.BroadcastListParticipant participants = 2;
optional string listName = 3;
repeated string labelIds = 4;
optional string audienceExpression = 5;
}
message CallLogAction {
@@ -5251,7 +5009,6 @@ message SyncActionValue {
DRAFTED = 8;
AI_HANDOFF = 9;
CHANNELS = 10;
AI_RESPONDING = 11;
}
}
@@ -5425,7 +5182,6 @@ message SyncActionValue {
repeated string keywords = 3;
optional int32 count = 4;
optional bool deleted = 5;
repeated string associatedLabelIds = 6;
}
message RecentEmojiWeightsAction {
@@ -5468,8 +5224,6 @@ message SyncActionValue {
optional bool isStatusNotificationEnabled = 29;
optional int32 statusNotificationToneId = 30;
optional bool shouldPlaySoundForCallNotification = 31;
optional string chatThemeId = 32;
optional string colorSchemeId = 33;
enum DisplayMode {
DISPLAY_MODE_UNKNOWN = 0;
ALWAYS = 1;
@@ -5514,8 +5268,6 @@ message SyncActionValue {
IS_STATUS_NOTIFICATION_ENABLED = 29;
STATUS_NOTIFICATION_TONE_ID = 30;
SHOULD_PLAY_SOUND_FOR_CALL_NOTIFICATION = 31;
CHAT_THEME_ID = 32;
COLOR_SCHEME_ID = 33;
}
enum SettingPlatform {
PLATFORM_UNKNOWN = 0;
@@ -5539,22 +5291,11 @@ message SyncActionValue {
repeated string userJid = 2;
optional bool shareToFB = 3;
optional bool shareToIG = 4;
repeated CustomList customLists = 5;
repeated StatusDistributionMode modes = 6;
message CustomList {
optional string listId = 1;
optional string name = 2;
optional string emoji = 3;
optional bool isSelected = 4;
repeated string userJid = 5;
}
enum StatusDistributionMode {
ALLOW_LIST = 0;
DENY_LIST = 1;
CONTACTS = 2;
CLOSE_FRIENDS = 3;
CUSTOM_LIST = 4;
}
}
@@ -5941,10 +5682,6 @@ message WebMessageInfo {
optional QuarantinedMessage quarantinedMessage = 77;
optional uint32 nonJidMentions = 78;
optional string hsmTag = 79;
optional uint64 ephemeralExpirationTimestamp = 80;
optional ScheduledMessageMetadata scheduledMessageMetadata = 81;
optional string decisionId = 82;
repeated string decisionSources = 83;
enum BizPrivacyStatus {
E2EE = 0;
FB = 2;
@@ -6184,8 +5921,6 @@ message WebMessageInfo {
GROUP_MEMBER_SHARE_GROUP_HISTORY_MODE = 221;
GROUP_OPEN_BOT_ADDED = 222;
GROUP_TEE_BOT_ADDED = 223;
CONTACT_INFO = 224;
SCHEDULED_MESSAGE_CREATED = 225;
}
}
+18 -26
View File
@@ -18,21 +18,14 @@ try {
'\tif (typeof value === "number") {\n' +
'\t\treturn String(value);\n' +
'\t}\n' +
'\t// Fast path: convert Long {low, high} directly via native BigInt\n' +
'\t// BigInt.toString() is a native C++ operation, much faster than Long\'s pure JS division loops\n' +
'\tif (value && typeof value.low === "number" && typeof value.high === "number") {\n' +
'\t\t// Normalize high to signed int32 so the sign-bit check works for inputs\n' +
'\t\t// where high is stored unsigned (e.g. raw {low, high} JSON).\n' +
'\t\tconst high = value.high | 0;\n' +
'\t\tconst lo = BigInt(value.low >>> 0);\n' +
'\t\tconst hi = BigInt(value.high >>> 0);\n' +
'\t\tconst combined = (hi << 32n) | lo;\n' +
'\t\tif (!unsigned && high < 0) {\n' +
'\t\t\treturn (combined - (1n << 64n)).toString();\n' +
'\t\t}\n' +
'\t\treturn combined.toString();\n' +
'\tif (!$util.Long) {\n' +
'\t\treturn String(value);\n' +
'\t}\n' +
'\treturn String(value);\n' +
'\tconst normalized = $util.Long.fromValue(value);\n' +
'\tconst prepared = unsigned && normalized && typeof normalized.toUnsigned === "function"\n' +
'\t\t? normalized.toUnsigned()\n' +
'\t\t: normalized;\n' +
'\treturn prepared.toString();\n' +
'}\n\n'
const longToNumberHelper =
'function longToNumber(value, unsigned) {\n' +
@@ -40,20 +33,19 @@ try {
'\t\treturn value;\n' +
'\t}\n' +
'\tif (typeof value === "string") {\n' +
'\t\tconst numeric = Number(value);\n' +
'\t\treturn numeric;\n' +
'\t}\n' +
'\tif (!$util.Long) {\n' +
'\t\treturn Number(value);\n' +
'\t}\n' +
'\t// Fast path: convert Long {low, high} directly via native BigInt\n' +
'\tif (value && typeof value.low === "number" && typeof value.high === "number") {\n' +
'\t\tconst high = value.high | 0;\n' +
'\t\tconst lo = BigInt(value.low >>> 0);\n' +
'\t\tconst hi = BigInt(value.high >>> 0);\n' +
'\t\tconst combined = (hi << 32n) | lo;\n' +
'\t\tif (!unsigned && high < 0) {\n' +
'\t\t\treturn Number(combined - (1n << 64n));\n' +
'\t\t}\n' +
'\t\treturn Number(combined);\n' +
'\t}\n' +
'\treturn Number(value);\n' +
'\tconst normalized = $util.Long.fromValue(value);\n' +
'\tconst prepared = unsigned && normalized && typeof normalized.toUnsigned === "function"\n' +
'\t\t? normalized.toUnsigned()\n' +
'\t\t: typeof normalized.toSigned === "function"\n' +
'\t\t\t? normalized.toSigned()\n' +
'\t\t\t: normalized;\n' +
'\treturn prepared.toNumber();\n' +
'}\n\n'
if (!content.includes('function longToString(')) {
+15 -633
View File
File diff suppressed because it is too large Load Diff
+31 -4457
View File
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
-531
View File
@@ -1,531 +0,0 @@
# Mensagens View-Once (Visualização Única) — Guia de Implementação
**Data:** 2026-03-23
**Versão:** 1.0
---
## 1. O que é View-Once
Mensagem de visualização única é um tipo especial de mídia (imagem, vídeo ou áudio) que o destinatário pode abrir apenas uma vez. Após abrir, o conteúdo some permanentemente.
---
> ## ⚠️ RESTRIÇÃO CRÍTICA — Contas Hosted (Meta Business Cloud API)
>
> **Contas registradas como "hosted" na Meta Business Cloud API têm view-once bloqueado pelo servidor do WhatsApp.**
>
> Essa restrição se aplica a **todos os clientes** (Android, iOS, WA Web) e **não há como contornar por código** — é uma política imposta diretamente pelo servidor WA para contas hosted.
>
> **Como identificar se sua conta é hosted:**
> - A conta foi criada via Meta Business Cloud API (WABA)
> - Aparece como "Meta" ou "Business Cloud" nas configurações do WhatsApp Business Manager
>
> **Contas não-hosted** (registradas diretamente via QR code ou pair code) **funcionam normalmente** com view-once.
---
**Comportamento em cada dispositivo após envio via API (contas não-hosted):**
| Dispositivo | O que aparece |
|---|---|
| Destinatário (Android/iOS) | Recebe a mídia com ícone de olhinho — pode abrir uma vez |
| Android principal do remetente | Mostra na conversa que a mensagem foi enviada (ícone de olhinho) |
| WA Web do remetente | "Aguardando mensagem. Abra o WhatsApp no seu celular." |
> **Nota:** A mensagem "Aguardando mensagem..." no WA Web é o comportamento correto e esperado quando o envio vem de um companion (API). É uma limitação do protocolo WhatsApp — o servidor só permite que o celular principal notifique companions com o conteúdo da view-once.
---
## 2. Como Enviar via API
### 2.1 Imagem
```bash
curl -X POST https://sua-api.com/v1/messages/send_image \
-H "Content-Type: application/json" \
-H "x-api-key: SEU_API_KEY" \
-d '{
"instance": "nome-da-instancia",
"to": "5511999999999",
"caption": "Olha só isso!",
"imageUrl": "https://exemplo.com/foto.jpg",
"viewOnce": true
}'
```
**Resposta de sucesso:**
```json
{ "ok": true }
```
### 2.2 Vídeo
```bash
curl -X POST https://sua-api.com/v1/messages/send_video \
-H "Content-Type: application/json" \
-H "x-api-key: SEU_API_KEY" \
-d '{
"instance": "nome-da-instancia",
"to": "5511999999999",
"caption": "Assista uma vez",
"videoUrl": "https://exemplo.com/video.mp4",
"viewOnce": true
}'
```
### 2.3 Áudio
```bash
curl -X POST https://sua-api.com/v1/messages/send_audio \
-H "Content-Type: application/json" \
-H "x-api-key: SEU_API_KEY" \
-d '{
"instance": "nome-da-instancia",
"to": "5511999999999",
"audioUrl": "https://exemplo.com/audio.ogg",
"ptt": false,
"viewOnce": true
}'
```
### 2.4 Usando Base64 (sem URL pública)
```bash
curl -X POST https://sua-api.com/v1/messages/send_image \
-H "Content-Type: application/json" \
-H "x-api-key: SEU_API_KEY" \
-d '{
"instance": "nome-da-instancia",
"to": "5511999999999",
"imageBase64": "data:image/jpeg;base64,/9j/4AAQSkZJRgAB...",
"viewOnce": true
}'
```
---
## 3. Como Receber via Webhook
Quando um contato envia uma mensagem de visualização única para o número conectado na API, o webhook recebe o evento com `key.isViewOnce = true`.
### 3.1 Estrutura do evento recebido
```json
{
"event": "messages.upsert",
"instance": "nome-da-instancia",
"data": {
"messages": [
{
"key": {
"remoteJid": "5511999999999@s.whatsapp.net",
"fromMe": false,
"id": "3EB0ABCDEF123456",
"isViewOnce": true
},
"message": {
"viewOnceMessageV2": {
"message": {
"imageMessage": {
"url": "https://mmg.whatsapp.net/...",
"mimetype": "image/jpeg",
"mediaKey": "base64...",
"fileEncSha256": "base64...",
"directPath": "/v/...",
"viewOnce": true
}
}
}
},
"messageTimestamp": 1742700000,
"pushName": "João Silva"
}
],
"type": "notify"
}
}
```
### 3.2 Como identificar que é view-once
O campo principal para identificar é `key.isViewOnce = true`. Existem dois caminhos para confirmar:
```javascript
function isViewOnce(msg) {
// Caminho 1: flag direta (mais simples)
if (msg.key?.isViewOnce) return true
// Caminho 2: verificar o wrapper da mensagem
const inner =
msg.message?.viewOnceMessageV2?.message ||
msg.message?.viewOnceMessage?.message ||
msg.message?.viewOnceMessageV2Extension?.message
return !!(
inner?.imageMessage?.viewOnce ||
inner?.videoMessage?.viewOnce ||
inner?.audioMessage?.viewOnce
)
}
```
### 3.3 Como acessar a mídia
```javascript
function getViewOnceMedia(msg) {
const inner =
msg.message?.viewOnceMessageV2?.message ||
msg.message?.viewOnceMessage?.message ||
msg.message?.viewOnceMessageV2Extension?.message
if (!inner) return null
if (inner.imageMessage) return { type: 'image', media: inner.imageMessage }
if (inner.videoMessage) return { type: 'video', media: inner.videoMessage }
if (inner.audioMessage) return { type: 'audio', media: inner.audioMessage }
return null
}
// Uso:
const media = getViewOnceMedia(msg)
if (media) {
console.log(`View-once ${media.type}`)
console.log('URL:', media.media.url)
console.log('DirectPath:', media.media.directPath)
console.log('MediaKey:', media.media.mediaKey) // necessário para download
}
```
---
## 4. Como Exibir numa Ferramenta (Frontend/CRM)
### 4.1 Lógica de detecção e exibição
```javascript
function renderMessage(msg) {
if (!isViewOnce(msg)) {
// Renderizar mensagem normal
return renderNormalMessage(msg)
}
const media = getViewOnceMedia(msg)
if (!media) return
// Verificar se já foi aberta (controle local da ferramenta)
const alreadyOpened = localStorage.getItem(`vo_${msg.key.id}`)
if (alreadyOpened) {
// Mostrar placeholder de "já visualizado"
return renderViewOncePlaceholder(media.type, msg.key.fromMe)
}
if (msg.key.fromMe) {
// Mensagem enviada por mim — nunca mostrar conteúdo
return renderSentViewOnce(media.type)
}
// Mensagem recebida — mostrar botão para "abrir uma vez"
return renderViewOnceButton(media.type, msg)
}
```
### 4.2 Componente de exibição — mensagem recebida
```javascript
function renderViewOnceButton(type, msg) {
const icons = { image: '📷', video: '🎥', audio: '🎵' }
const labels = { image: 'Foto', video: 'Vídeo', áudio: 'Áudio' }
return `
<div class="view-once-bubble received">
<span class="view-once-icon">${icons[type]}</span>
<span class="view-once-label">${labels[type]} de visualização única</span>
<button onclick="openViewOnce('${msg.key.id}', '${type}')">
Abrir
</button>
</div>
`
}
async function openViewOnce(msgId, type) {
// 1. Marcar como aberta ANTES de mostrar (só uma chance)
localStorage.setItem(`vo_${msgId}`, '1')
// 2. Baixar e mostrar a mídia
const mediaUrl = await downloadViewOnceMedia(msgId)
showMedia(type, mediaUrl)
// 3. Após fechar/visualizar, substituir pelo placeholder
onMediaClosed(() => {
replaceWithPlaceholder(msgId, type)
})
}
```
### 4.3 Componente de exibição — mensagem enviada
```javascript
function renderSentViewOnce(type) {
const labels = { image: 'Foto', video: 'Vídeo', audio: 'Áudio' }
return `
<div class="view-once-bubble sent">
<span class="view-once-icon">👁️</span>
<span>${labels[type]} de visualização única</span>
<span class="view-once-status">Enviada</span>
</div>
`
}
```
### 4.4 Placeholder após visualização
```javascript
function renderViewOncePlaceholder(type, fromMe) {
const labels = { image: 'foto', video: 'vídeo', audio: 'áudio' }
const text = fromMe
? `Você enviou uma ${labels[type]} de visualização única`
: `${labels[type]} de visualização única aberta`
return `
<div class="view-once-bubble placeholder">
<span class="view-once-icon">🚫</span>
<span>${text}</span>
</div>
`
}
```
### 4.5 CSS sugerido
```css
.view-once-bubble {
display: flex;
align-items: center;
gap: 8px;
padding: 10px 14px;
border-radius: 12px;
max-width: 280px;
}
.view-once-bubble.received {
background: #f0f0f0;
color: #333;
}
.view-once-bubble.sent {
background: #dcf8c6;
color: #333;
align-self: flex-end;
}
.view-once-bubble.placeholder {
background: #e8e8e8;
color: #999;
font-style: italic;
}
.view-once-bubble button {
background: #25d366;
color: white;
border: none;
border-radius: 8px;
padding: 6px 12px;
cursor: pointer;
font-size: 13px;
}
```
---
## 5. Código Implementado na API
### 5.1 Geração da mensagem — `src/Utils/messages.ts`
Quando `viewOnce: true` é passado, a mídia é encapsulada no wrapper moderno `viewOnceMessageV2` (campo 55 do proto) e o flag `viewOnce=true` é setado na mídia interna:
```typescript
if (hasOptionalProperty(message, 'viewOnce') && !!message.viewOnce) {
// Seta viewOnce=true na mensagem de mídia interna
if (m.imageMessage) m.imageMessage.viewOnce = true
else if (m.videoMessage) m.videoMessage.viewOnce = true
else if (m.audioMessage) m.audioMessage.viewOnce = true
// Usa viewOnceMessageV2 (campo 55) — formato atual do WA
m = { viewOnceMessageV2: { message: m } }
}
```
### 5.2 Envio — `src/Socket/messages-send.ts`
**Detecção de view-once real** (não confunde com botões/listas que também usam o wrapper `viewOnceMessage`):
```typescript
// Detecta view-once real pela flag viewOnce na mídia interna.
// viewOnceMessage* wrappers são também usados por botões/listas/carrosséis
// (que carregam interactiveMessage dentro) — esses NÃO têm viewOnce=true na mídia.
const viewOnceInner =
message.viewOnceMessageV2?.message ||
message.viewOnceMessage?.message ||
message.viewOnceMessageV2Extension?.message
const isViewOnceMsg = !!(
viewOnceInner?.imageMessage?.viewOnce ||
viewOnceInner?.videoMessage?.viewOnce ||
viewOnceInner?.audioMessage?.viewOnce
)
// Para view-once: DSM enviado apenas para o celular principal (device=0).
// Companions (device>0) são omitidos — o servidor WA gera
// <unavailable type="view_once"/> automaticamente para eles.
const viewOnceMeRecipients = isViewOnceMsg
? meRecipients.filter(jid => !jidDecode(jid)?.device)
: meRecipients
// assertSessions apenas para quem vai realmente receber
await assertSessions([...viewOnceMeRecipients, ...otherRecipients])
const [meNodes, otherNodes] = await Promise.all([
createParticipantNodes(viewOnceMeRecipients, meMsg || message, extraAttrs),
createParticipantNodes(otherRecipients, message, extraAttrs)
])
participants.push(...meNodes)
participants.push(...otherNodes)
// phash recalculado com os participantes reais
const phashRecipients = isViewOnceMsg
? [...viewOnceMeRecipients, ...otherRecipients]
: [...meRecipients, ...otherRecipients]
if (phashRecipients.length > 0) {
extraAttrs['phash'] = generateParticipantHashV2(phashRecipients)
}
```
**`getMediaType()` — garante que vídeo e áudio view-once sejam entregues:**
```typescript
const getMediaType = (message: proto.IMessage) => {
// Desencapsula wrapper view-once antes de verificar o tipo de mídia.
// Sem isso, message.videoMessage e message.audioMessage são undefined
// (estão dentro do wrapper) e o atributo mediatype ficaria ausente no
// enc node — o servidor WA descarta silenciosamente vídeo/áudio view-once.
const inner =
message.viewOnceMessage?.message ||
message.viewOnceMessageV2?.message ||
message.viewOnceMessageV2Extension?.message
if (inner) {
return getMediaType(inner)
}
if (message.imageMessage) return 'image'
else if (message.videoMessage) return 'video'
// ...
}
```
### 5.3 Recepção — `src/Utils/decode-wa-message.ts`
Quando a API recebe uma view-once como companion (linked device), o servidor envia dois stanzas:
- **Stanza 1** (`enc`): conteúdo completo com metadados da mídia
- **Stanza 2** (`unavailable type="view_once"`): sinal de sync
O stanza 1 é detectado e marcado corretamente:
```typescript
// Detecta view-once na stanza 1 recebida por linked device.
// O wrapper viewOnceMessage também é usado por mensagens interativas
// (interactiveMessage, listMessage) — esses NÃO têm viewOnce=true na mídia.
const viewOnceInner =
msg.viewOnceMessage?.message ||
msg.viewOnceMessageV2?.message ||
msg.viewOnceMessageV2Extension?.message
if (
viewOnceInner?.imageMessage?.viewOnce ||
viewOnceInner?.videoMessage?.viewOnce ||
viewOnceInner?.audioMessage?.viewOnce
) {
fullMessage.key.isViewOnce = true
}
```
---
## 6. Fluxo Completo
```
API envia view-once para destinatário
├── Protobuf gerado:
│ viewOnceMessageV2 {
│ message {
│ imageMessage { viewOnce: true, url, mediaKey, ... }
│ }
│ }
├── Stanza enviado ao servidor WA:
│ <message to="destinatario@s.whatsapp.net" type="media">
│ <enc type="pkmsg">...viewOnceMessageV2 cifrado...</enc> ← para o destinatário
│ <participants>
│ <to jid="meu-numero:0@lid"> ← celular principal
│ <enc type="msg">...DSM{viewOnceMessageV2}...</enc>
│ </to>
│ <!-- device>0 omitidos — servidor gera unavailable -->
│ </participants>
│ </message>
└── Distribuição pelo servidor:
├── Destinatário → recebe enc → abre uma vez ✅
├── Celular principal → recebe DSM → mostra "você enviou" ✅
└── WA Web → recebe <unavailable> → "Aguardando..." ✅
Destinatário envia view-once de volta para a API
├── Stanza recebido pela API (stanza 1 — enc com conteúdo):
│ <message from="destinatario@s.whatsapp.net" type="media">
│ <enc type="msg">...viewOnceMessageV2 cifrado...</enc>
│ </message>
├── decode-wa-message.ts:
│ └── Descriptografa → detecta viewOnce=true → seta key.isViewOnce=true
├── Stanza recebido (stanza 2 — unavailable, ignorado):
│ <message from="...">
│ <unavailable type="view_once"/>
│ </message>
└── Webhook emitido:
messages.upsert → msg.key.isViewOnce = true ✅
```
---
## 7. Limitações
| Situação | Comportamento | Motivo |
|---|---|---|
| **Conta hosted (Meta Business Cloud API)** | **View-once bloqueado para todos os clientes** | **Política do servidor WA para contas hosted — sem workaround** |
| WA Web do remetente | "Aguardando mensagem. Abra no celular." | Servidor WA só aceita `<unavailable>` vindo do celular principal, não de companions |
| View-once de documento/sticker | Não suportado | Limitação do protocolo WA — só imagem, vídeo e áudio |
| `error_479` nos logs | Normal, não é erro | TcToken é atualizado após o envio de view-once |
| Companion não consegue abrir | Por design | O conteúdo não é entregue para linked devices |
---
## 8. Checklist de Validação
**Envio:**
- [ ] Destinatário recebe com ícone de olhinho
- [ ] Destinatário consegue abrir a mídia
- [ ] Após abrir, a mídia some
- [ ] Celular principal do remetente mostra "você enviou" com ícone de olhinho
- [ ] WA Web do remetente mostra "Aguardando mensagem..."
- [ ] Funciona para imagem, vídeo e áudio
- [ ] Logs mostram `📤 Message sent` (error_479 é normal)
**Recepção:**
- [ ] Webhook recebe `key.isViewOnce = true`
- [ ] `viewOnceMessageV2.message.imageMessage` (ou video/audio) está presente
- [ ] API não crasha ao receber view-once de outro dispositivo
- [ ] Mensagem interativa (botão/lista) enviada após view-once não é afetada
+1 -1
View File
@@ -1 +1 @@
{"version":[2,3000,1038585534]}
{"version":[2,3000,1035223526]}
+9 -43
View File
@@ -31,26 +31,9 @@ export const STATUS_EXPIRY_SECONDS = 24 * 60 * 60
export const PLACEHOLDER_MAX_AGE_SECONDS = 7 * 24 * 60 * 60
export const NOISE_MODE = 'Noise_XX_25519_AESGCM_SHA256\0\0\0\0'
/**
* Protocol mode: 'web' (default) or 'native' (experimental)
* - web: WA\x06\x03 — standard WhatsApp Web protocol
* - native: WAM\x05 — native Android protocol (may enable view-once media on companions)
*
* Set via BAILEYS_PROTOCOL env var:
* BAILEYS_PROTOCOL=native
*/
const PROTOCOL_MODE = process.env.BAILEYS_PROTOCOL?.trim().toLowerCase() === 'native' ? 'native' : 'web'
export const DICT_VERSION = PROTOCOL_MODE === 'native' ? 5 : 3
export const DICT_VERSION = 3
export const KEY_BUNDLE_TYPE = Buffer.from([5])
// WA\x06\x03 = web protocol, WAM\x05 = native Android protocol
export const NOISE_WA_HEADER = PROTOCOL_MODE === 'native'
? Buffer.from([87, 65, 77, DICT_VERSION]) // WAM\x05
: Buffer.from([87, 65, 6, DICT_VERSION]) // WA\x06\x03
export { PROTOCOL_MODE }
export const NOISE_WA_HEADER = Buffer.from([87, 65, 6, DICT_VERSION]) // last is "DICT_VERSION"
/** from: https://stackoverflow.com/questions/3809401/what-is-a-good-regular-expression-to-match-a-url */
export const URL_REGEX = /https:\/\/(?![^:@\/\s]+:[^:@\/\s]+@)[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}(:\d+)?(\/[^\s]*)?/g
@@ -142,6 +125,8 @@ export const DEFAULT_CONNECTION_CONFIG: SocketConfig = {
getMessage: async () => undefined,
cachedGroupMetadata: async () => undefined,
makeSignalRepository: makeLibSignalRepository,
// Circuit breaker configuration
enableCircuitBreaker: true,
// Listener limits (memory leak prevention)
// WebSocket: 8 core events (open, close, error, message, ping, pong, upgrade, unexpected-response)
// + 10 dynamic listeners (reconnect handlers, custom events)
@@ -169,20 +154,7 @@ export const MEDIA_PATH_MAP: { [T in MediaType]?: string } = {
'md-msg-hist': '/mms/md-app-state',
'biz-cover-photo': '/pps/biz-cover-photo',
'sticker-pack': '/mms/sticker-pack',
'thumbnail-sticker-pack': '/mms/thumbnail-sticker-pack',
// View-once types — same upload path as regular media (server CDN routing TBD)
'viewonce-image': '/mms/image',
'viewonce-video': '/mms/video',
'viewonce-audio': '/mms/audio'
}
export const NEWSLETTER_MEDIA_PATH_MAP: { [T in MediaType]?: string } = {
image: '/newsletter/newsletter-image',
video: '/newsletter/newsletter-video',
document: '/newsletter/newsletter-document',
audio: '/newsletter/newsletter-audio',
sticker: '/newsletter/newsletter-image',
'thumbnail-link': '/newsletter/newsletter-image'
'thumbnail-sticker-pack': '/mms/thumbnail-sticker-pack'
}
export const MEDIA_HKDF_KEY_MAPPING = {
@@ -206,11 +178,7 @@ export const MEDIA_HKDF_KEY_MAPPING = {
ptv: 'Video',
'biz-cover-photo': 'Image',
'sticker-pack': 'Sticker Pack',
'thumbnail-sticker-pack': 'Sticker Pack Thumbnail',
// View-once types — same HKDF keys as regular types (recipients decrypt with normal keys)
'viewonce-image': 'Image',
'viewonce-video': 'Video',
'viewonce-audio': 'Audio'
'thumbnail-sticker-pack': 'Sticker Pack Thumbnail'
}
export type MediaType = keyof typeof MEDIA_HKDF_KEY_MAPPING
@@ -220,12 +188,10 @@ export const MEDIA_KEYS = Object.keys(MEDIA_PATH_MAP) as MediaType[]
/** 120s timeout for history sync stall detection, same as WA Web's handleChunkProgress / restartPausedTimer (g = 120) */
export const HISTORY_SYNC_PAUSED_TIMEOUT_MS = 120_000
// Replenishment threshold: when server count drops below this, top-up back to INITIAL_PREKEY_COUNT
export const MIN_PREKEY_COUNT = 200
export const MIN_PREKEY_COUNT = 5
// Initial pool size matching WA Business (CDP IDB capture: prekey-store = 812 on registration)
// Rounded to 800 for cleanliness; replenishment always tops up to this value
export const INITIAL_PREKEY_COUNT = 800
// Moderate prekey count (upstream uses 812, reduced to balance rate limiting and availability)
export const INITIAL_PREKEY_COUNT = 200
export const UPLOAD_TIMEOUT = 30000 // 30 seconds
// Moderate upload interval to balance rate limiting and responsiveness (was 5000)
+68 -92
View File
@@ -6,6 +6,7 @@ import type { LIDMapping, SignalAuthState, SignalKeyStoreWithTransaction } from
import type { BaileysEventEmitter } from '../Types/Events'
import type { SignalRepositoryWithLIDStore } from '../Types/Signal'
import { generateSignalPubKey } from '../Utils'
import { CircuitBreaker } from '../Utils/circuit-breaker.js'
import type { ILogger } from '../Utils/logger'
import { metrics } from '../Utils/prometheus-metrics.js'
import { isAnyLidUser, isAnyPnUser, jidDecode, transferDevice, WAJIDDomains } from '../WABinary'
@@ -66,6 +67,8 @@ export interface IdentitySaveResult {
export interface LibSignalRepositoryOptions {
/** Event emitter for broadcasting identity changes */
ev?: BaileysEventEmitter
/** Circuit breaker for prekey operations (optional) */
preKeyCircuitBreaker?: CircuitBreaker
}
// ============================================
@@ -253,7 +256,7 @@ export function makeLibSignalRepository(
pnToLIDFunc?: (jids: string[]) => Promise<LIDMapping[] | undefined>,
options?: LibSignalRepositoryOptions
): SignalRepositoryWithLIDStore {
const { ev } = options || {}
const { ev, preKeyCircuitBreaker } = options || {}
const lidMapping = new LIDMappingStore(auth.keys as SignalKeyStoreWithTransaction, logger, pnToLIDFunc)
// Identity key cache to avoid repeated storage reads
@@ -274,7 +277,7 @@ export function makeLibSignalRepository(
cacheMetricsInterval.unref()
}
const storage = signalStorage(auth, lidMapping, identityKeyCache, ev, logger)
const storage = signalStorage(auth, lidMapping, identityKeyCache, ev, preKeyCircuitBreaker, logger)
const parsedKeys = auth.keys as SignalKeyStoreWithTransaction
const migratedSessionCache = new LRUCache<string, true>({
@@ -299,6 +302,25 @@ export function makeLibSignalRepository(
// Promise instead of each spawning their own DB transactions.
const migrationInFlight = new Map<string, Promise<{ migrated: number; skipped: number; total: number }>>()
// Resolve PN JID to its canonical LID JID for transaction locking.
// This prevents PN/LID race conditions where concurrent operations for the
// same logical contact acquire different mutex locks because one uses PN
// and the other uses LID. (Aligned with WABA behavior — all operations use LID internally.)
const resolveCanonicalJid = async(jid: string): Promise<string> => {
if (isAnyLidUser(jid)) {
return jid
}
if (isAnyPnUser(jid)) {
const lid = await lidMapping.getLIDForPN(jid)
if (lid) {
return lid
}
}
return jid
}
const repository: SignalRepositoryWithLIDStore = {
decryptGroupMessage({ group, authorJid, msg }) {
const senderName = jidToSignalSenderKeyName(group, authorJid)
@@ -341,14 +363,6 @@ export function makeLibSignalRepository(
},
async decryptMessage({ jid, type, ciphertext }) {
const addr = jidToSignalProtocolAddress(jid)
// Wire address = the EXACT key that signalStorage's loadSession/storeSession
// uses internally (it canonicalizes PN→LID via resolveLIDSignalAddress).
// We MUST lock on this same key — locking on the raw `jid` lets two
// concurrent ops for the same logical contact (one arriving as PN, one as
// LID) acquire DIFFERENT mutex slots and interleave on the same session
// record, corrupting the ratchet → perpetual `Bad MAC` on own DSM after
// the WhatsApp PN→LID DSM rollout.
const wireAddr = await resolveLIDSignalAddress(addr.toString(), lidMapping)
const session = new libsignal.SessionCipher(storage, addr)
async function doDecrypt() {
@@ -371,9 +385,13 @@ export function makeLibSignalRepository(
'Identity key changed - contact may have reinstalled WhatsApp, session will be re-established'
)
// (No circuit breaker to reset — bounded-retry is stateless,
// each retry is independent so identity-change recovery
// happens automatically on the next operation.)
// Reset prekey circuit breaker since we identified the cause
// Reset regardless of state (could be open, half-open, or closed with accumulated failures)
// eslint-disable-next-line max-depth
if (preKeyCircuitBreaker) {
preKeyCircuitBreaker.reset()
logger.debug({ jid }, 'Reset prekey circuit breaker after identity key change detection')
}
} else if (saveResult.isNew) {
logger.debug(
{ jid, addr: addrStr, fingerprint: saveResult.currentFingerprint },
@@ -397,23 +415,24 @@ export function makeLibSignalRepository(
return result
}
// Use canonical JID (PN→LID resolved) as transaction key to prevent
// PN/LID race conditions on the same logical session.
const canonicalJid = await resolveCanonicalJid(jid)
return parsedKeys.transaction(async () => {
return await doDecrypt()
}, wireAddr)
}, canonicalJid)
},
async encryptMessage({ jid, data }) {
const addr = jidToSignalProtocolAddress(jid)
// See decryptMessage above for the rationale: lock on the wire address,
// not the raw JID, so concurrent PN/LID ops on the same session serialize.
const wireAddr = await resolveLIDSignalAddress(addr.toString(), lidMapping)
const cipher = new libsignal.SessionCipher(storage, addr)
const canonicalJid = await resolveCanonicalJid(jid)
return parsedKeys.transaction(async () => {
const { type: sigType, body } = await cipher.encrypt(data)
const type = sigType === 3 ? 'pkmsg' : 'msg'
return { type, ciphertext: Buffer.from(body, 'binary') }
}, wireAddr)
}, canonicalJid)
},
async encryptGroupMessage({ group, meId, data }) {
@@ -441,16 +460,10 @@ export function makeLibSignalRepository(
async injectE2ESession({ jid, session }) {
logger.trace({ jid }, 'injecting E2EE session')
const addr = jidToSignalProtocolAddress(jid)
// Same wire-address locking as encrypt/decrypt — `SessionBuilder.initOutgoing`
// writes via `storeSession` which canonicalizes, so the lock key must be
// the wire address for the mutex to actually serialize concurrent ops on
// the same logical session.
const wireAddr = await resolveLIDSignalAddress(addr.toString(), lidMapping)
const cipher = new libsignal.SessionBuilder(storage, addr)
const cipher = new libsignal.SessionBuilder(storage, jidToSignalProtocolAddress(jid))
return parsedKeys.transaction(async () => {
await cipher.initOutgoing(session)
}, wireAddr)
}, jid)
},
jidToSignalProtocolAddress(jid) {
return jidToSignalProtocolAddress(jid).toString()
@@ -727,65 +740,6 @@ const jidToSignalSenderKeyName = (group: string, user: string): SenderKeyName =>
return new SenderKeyName(group, jidToSignalProtocolAddress(user))
}
/**
* Resolve a Signal protocol address (string form `signalUser.device`, where
* `signalUser` is `user` for the WhatsApp domain or `user_domainType` for any
* other domain — e.g. `12345_1.0` for a LID device 0) to its canonical *wire*
* address — the exact key that `loadSession`/`storeSession` uses when
* reading/writing `keys.get('session', [...])`.
*
* This is the SAME key the storage layer uses internally, so callers that need
* a transaction lock around session mutations MUST lock on this resolved
* address (not the raw JID), otherwise the lock key and storage key drift
* apart and concurrent ops on the same session interleave → ratchet
* corruption → perpetual `Bad MAC` errors.
*
* Behavior: if the input already targets a LID domain, returns it unchanged
* (cheap fast path — no awaits). Otherwise looks up the PN→LID mapping; if
* found, returns the LID-form Signal address. If no mapping exists, returns
* the input unchanged.
*
* Residual race window: the lock and the storage's internal re-resolution
* each call this function once. If a PN→LID mapping is added between those
* two calls (microsecond window), the lock could be on the PN address while
* storage canonicalizes to LID. In production this is mitigated upstream:
* `messages-recv.ts` runs `migrateSession()` synchronously BEFORE
* `decryptMessage()` is called, so the mapping is already populated by the
* time we lock. Eliminating the residual race entirely would require storage
* to NOT re-resolve, which is a bigger refactor (other call sites depend on
* the current contract).
*
* Top-level (not inside `signalStorage`) so `makeLibSignalRepository` can
* compute the same wire address for transaction locking.
*/
const resolveLIDSignalAddress = async (id: string, lidMapping: LIDMappingStore): Promise<string> => {
if (id.includes('.')) {
const [deviceId, device] = id.split('.')
if (!deviceId) {
throw new Error(`Malformed signal address (empty user portion before '.'): "${id}"`)
}
if (device === undefined || device === '') {
throw new Error(`Malformed signal address (empty device portion after '.'): "${id}"`)
}
const [user, domainType_] = deviceId.split('_')
const domainType = parseInt(domainType_ || '0')
if (domainType === WAJIDDomains.LID || domainType === WAJIDDomains.HOSTED_LID) return id
const pnJid = `${user!}${device !== '0' ? `:${device}` : ''}@${domainType === WAJIDDomains.HOSTED ? 'hosted' : 's.whatsapp.net'}`
const lidForPN = await lidMapping.getLIDForPN(pnJid)
if (lidForPN) {
const lidAddr = jidToSignalProtocolAddress(lidForPN)
return lidAddr.toString()
}
}
return id
}
/**
* Extended SignalStorage with identity key management
* This type adds identity key operations to the standard Signal storage
@@ -815,11 +769,33 @@ function signalStorage(
lidMapping: LIDMappingStore,
identityKeyCache: LRUCache<string, Uint8Array>,
ev?: BaileysEventEmitter,
preKeyCircuitBreaker?: CircuitBreaker,
logger?: ILogger
): ExtendedSignalStorage {
// Bind the top-level `resolveLIDSignalAddress` to this instance's lidMapping
// so call sites below stay readable.
const resolveAddr = (id: string) => resolveLIDSignalAddress(id, lidMapping)
// Shared function to resolve PN signal address to LID if mapping exists
const resolveLIDSignalAddress = async (id: string): Promise<string> => {
if (id.includes('.')) {
const [deviceId, device] = id.split('.')
if (!deviceId) {
throw new Error('Missing device ID')
}
const [user, domainType_] = deviceId.split('_')
const domainType = parseInt(domainType_ || '0')
if (domainType === WAJIDDomains.LID || domainType === WAJIDDomains.HOSTED_LID) return id
const pnJid = `${user!}${device !== '0' ? `:${device}` : ''}@${domainType === WAJIDDomains.HOSTED ? 'hosted' : 's.whatsapp.net'}`
const lidForPN = await lidMapping.getLIDForPN(pnJid)
if (lidForPN) {
const lidAddr = jidToSignalProtocolAddress(lidForPN)
return lidAddr.toString()
}
}
return id
}
// Delayed PreKey deletion: grace period to handle race conditions
// where two pkmsg with the same preKeyId arrive nearly simultaneously.
@@ -831,7 +807,7 @@ function signalStorage(
return {
loadSession: async (id: string) => {
try {
const wireJid = await resolveAddr(id)
const wireJid = await resolveLIDSignalAddress(id)
const { [wireJid]: sess } = await keys.get('session', [wireJid])
if (sess) {
@@ -844,7 +820,7 @@ function signalStorage(
return null
},
storeSession: async (id: string, session: libsignal.SessionRecord) => {
const wireJid = await resolveAddr(id)
const wireJid = await resolveLIDSignalAddress(id)
await keys.set({ session: { [wireJid]: session.serialize() } })
},
isTrustedIdentity: () => {
@@ -921,7 +897,7 @@ function signalStorage(
const timer = metrics.signalIdentityKeyOperations?.startTimer({ operation: 'load' })
try {
const wireJid = await resolveAddr(id)
const wireJid = await resolveLIDSignalAddress(id)
// Check cache first
const cached = identityKeyCache.get(wireJid)
@@ -955,7 +931,7 @@ function signalStorage(
const timer = metrics.signalIdentityKeyOperations?.startTimer({ operation: 'save' })
try {
const wireJid = await resolveAddr(id)
const wireJid = await resolveLIDSignalAddress(id)
const currentFingerprint = generateKeyFingerprint(identityKey)
// Load existing key (from cache or storage)
+17 -182
View File
@@ -54,7 +54,7 @@ import {
resolveLidToPn
} from '../Utils'
import { makeKeyedMutex, makeMutex } from '../Utils/make-mutex'
import processMessage, { getChatId } from '../Utils/process-message'
import processMessage from '../Utils/process-message'
import { buildTcTokenFromJid } from '../Utils/tc-token-utils'
import {
type BinaryNode,
@@ -98,24 +98,6 @@ export const makeChatsSocket = (config: SocketConfig) => {
let privacySettings: { [_: string]: string } | undefined
/**
* Server-assigned AB props that gate tctoken-related protocol behavior.
* Defaults match WA Web (safe — avoids spurious error 463 if the prop never
* arrives). Populated from `fetchProps()` on connection.
*
* - `privacyTokenOn1to1` (AB prop 10518 / `privacy_token_sending_on_all_1_on_1_messages`):
* include tctoken in 1:1 messages.
* - `profilePicPrivacyToken` (AB prop 9666 / `profile_scraping_privacy_token_in_photo_iq`):
* include tctoken in profile picture IQs.
* - `lidTrustedTokenIssueToLid` (AB prop 14303 / `lid_trusted_token_issue_to_lid`):
* issue privacy tokens to the contact's LID instead of the PN.
*/
const serverProps = {
privacyTokenOn1to1: true,
profilePicPrivacyToken: true,
lidTrustedTokenIssueToLid: false
}
let syncState: SyncState = SyncState.Connecting
/** this mutex ensures that messages from the same chat are processed in order, while allowing parallel processing of messages from different chats */
@@ -130,18 +112,6 @@ export const makeChatsSocket = (config: SocketConfig) => {
/** this mutex ensures that notifications from the same chat are processed in order, while allowing parallel processing across chats */
const notificationMutex = makeKeyedMutex()
/**
* Per-chat mutex dedicated to post-upsert work (history app-state sync +
* processMessage side effects). Kept separate from `messageMutex` because
* the inbound caller already holds `messageMutex(chatId)` while running
* decrypt + upsertMessage; sharing the same mutex would let a concurrently-
* arrived message N+1 enqueue *between* msg N's outer callback and msg N's
* post-upsert task, so msg N+1's processMessage could run before msg N's
* (breaking per-chat ordering of side effects). With a separate mutex,
* post-upsert tasks enqueue strictly in upsertMessage call order.
*/
const postUpsertMutex = makeKeyedMutex()
// Timeout for AwaitingInitialSync state
let awaitingSyncTimeout: NodeJS.Timeout | undefined
@@ -821,10 +791,7 @@ export const makeChatsSocket = (config: SocketConfig) => {
me && (normalizedJid === jidNormalizedUser(me.id) || (me.lid && normalizedJid === jidNormalizedUser(me.lid)))
let content: BinaryNode[] | undefined = baseContent
// Gate inclusion on AB prop 9666 (profile_scraping_privacy_token_in_photo_iq).
// WA Web defaults to true; if the server flips it off, we mirror that to
// avoid divergence with the spec-compliant client.
if (serverProps.profilePicPrivacyToken && isUserJid && !isSelf) {
if (isUserJid && !isSelf) {
content = await buildTcTokenFromJid({
authState,
jid: normalizedJid,
@@ -1093,21 +1060,22 @@ export const makeChatsSocket = (config: SocketConfig) => {
}
}
/** fetch AB props */
/** sending non-abt props may fix QR scan fail if server expects */
const fetchProps = async () => {
//TODO: implement both protocol 1 and protocol 2 prop fetching, specially for abKey for WM
const resultNode = await query({
tag: 'iq',
attrs: {
to: S_WHATSAPP_NET,
xmlns: 'abt',
xmlns: 'w',
type: 'get'
},
content: [
{
tag: 'props',
attrs: {
protocol: '1',
...(authState?.creds?.lastPropHash ? { hash: authState.creds.lastPropHash } : {})
protocol: '2',
hash: authState?.creds?.lastPropHash || ''
}
}
]
@@ -1126,25 +1094,7 @@ export const makeChatsSocket = (config: SocketConfig) => {
props = reduceBinaryNodeToDictionary(propsNode, 'prop')
}
// Extract protocol-relevant AB props (defaults match WA Web; see serverProps doc).
// We accept both numeric IDs and human-readable names so the parser is resilient
// to upstream renaming and to future-versioned WA Web servers.
const privacyTokenProp = props['10518'] ?? props['privacy_token_sending_on_all_1_on_1_messages']
if (privacyTokenProp !== undefined) {
serverProps.privacyTokenOn1to1 = privacyTokenProp === 'true' || privacyTokenProp === '1'
}
const profilePicProp = props['9666'] ?? props['profile_scraping_privacy_token_in_photo_iq']
if (profilePicProp !== undefined) {
serverProps.profilePicPrivacyToken = profilePicProp === 'true' || profilePicProp === '1'
}
const lidIssueProp = props['14303'] ?? props['lid_trusted_token_issue_to_lid']
if (lidIssueProp !== undefined) {
serverProps.lidTrustedTokenIssueToLid = lidIssueProp === 'true' || lidIssueProp === '1'
}
logger.debug({ serverProps }, 'fetched props')
logger.debug('fetched props')
return props
}
@@ -1411,19 +1361,7 @@ export const makeChatsSocket = (config: SocketConfig) => {
blockedCollections.clear()
logger.info('Doing app state sync')
try {
await resyncAppState(ALL_WA_PATCH_NAMES, true)
} catch (err) {
// Failure recovery: without this, syncState would stay at Syncing
// and ev.flush() would never run, leaving the event buffer pinned
// until the buffer's own safety timeout expires. Force the state
// machine forward so live inbound events can flow even if the
// app-state resync failed (collections are already cleared, so
// blocked patches will be retried on the next creds.update tick).
syncState = SyncState.Online
ev.flush()
throw err
}
// Sync is complete, go online and flush everything
syncState = SyncState.Online
@@ -1435,25 +1373,12 @@ export const makeChatsSocket = (config: SocketConfig) => {
}
}
// Post-upsert work: history app-state sync + processMessage side effects.
// Awaiting here keeps `messages.upsert` pinned in the event buffer
// (createBufferedFunction only schedules flush after work() resolves), so
// the hot path detaches this work to release the emit on the next debounce
// tick.
//
// Use Promise.allSettled so the combined promise only settles after BOTH
// tasks finish. With a plain Promise.all, an early rejection from one task
// would release the keyed mutex while the other task is still mutating
// chat state — letting the next message of the same chat overtake it and
// break per-chat ordering.
//
// Returns the per-task settle status so the keyShare branch can know
// whether processMessage actually persisted the new app-state-sync key
// before triggering doAppStateSync (otherwise the sync would hit
// isMissingKeyError and park collections in blockedCollections).
const postUpsertTasks = async (): Promise<{ processMessageOk: boolean }> => {
const [historyResult, processResult] = await Promise.allSettled([
shouldProcessHistoryMsg ? doAppStateSync() : Promise.resolve(),
await Promise.all([
(async () => {
if (shouldProcessHistoryMsg) {
await doAppStateSync()
}
})(),
processMessage(msg, {
signalRepository,
shouldProcessHistoryMsg,
@@ -1467,99 +1392,10 @@ export const makeChatsSocket = (config: SocketConfig) => {
})
])
if (historyResult.status === 'rejected') {
logger?.warn(
{ err: historyResult.reason, messageId: msg.key?.id, remoteJid: msg.key?.remoteJid },
'history doAppStateSync failed'
)
}
if (processResult.status === 'rejected') {
logger?.warn(
{ err: processResult.reason, messageId: msg.key?.id, remoteJid: msg.key?.remoteJid },
'processMessage failed'
)
}
return { processMessageOk: processResult.status === 'fulfilled' }
}
// Use getChatId + jidNormalizedUser so the mutex key matches the chat-id
// scheme processMessage uses for chat updates (broadcasts target the
// participant). When getChatId/jidNormalizedUser yields nothing usable
// (missing or malformed JID), prefer a message-derived fallback over a
// single global 'unknown' bucket — that bucket would head-of-line block
// every malformed message behind a shared queue. msg.key.id is unique
// per message so unrelated malformed inputs no longer serialize together;
// for valid messages we still hit the normalized chat-id path, so the
// per-chat ordering guarantee is unchanged where it matters.
const rawChatId = getChatId(msg.key)
const normalizedChatId = rawChatId ? jidNormalizedUser(rawChatId) : ''
const postUpsertChatId = normalizedChatId || msg.key?.id || 'unknown'
// Wrap in `postUpsertMutex(chatId)` (a SEPARATE keyed mutex from the outer
// `messageMutex` held by the inbound caller) so per-chat ordering of
// processMessage side effects (chat.unreadCount, LID/PN mapping,
// messages.update, history downloads) is preserved across messages of the
// same chat.
//
// Why a separate mutex: if we re-used messageMutex, a concurrently-arrived
// message N+1 could enqueue on the outer mutex BEFORE msg N's post-upsert
// task gets enqueued (because N's outer callback yields on `await decrypt()`
// before reaching the inner enqueue site). The queue would then be
// [OuterN+1, InnerN, ...], so InnerN+1 would beat InnerN to processMessage.
// With its own mutex, post-upsert tasks enqueue strictly in upsertMessage
// call order (which IS message arrival order because the outer
// messageMutex serializes the upserts per-chat).
const postUpsertWork = postUpsertMutex.mutex(postUpsertChatId, postUpsertTasks)
const isKeyShareDuringSync =
!!msg.message?.protocolMessage?.appStateSyncKeyShare && syncState === SyncState.Syncing
if (isKeyShareDuringSync) {
// appStateSyncKeyShare path: processMessage persists the new app-state-sync
// key in its APP_STATE_SYNC_KEY_SHARE handler (via keyStore.transaction).
// The follow-up doAppStateSync() needs that key to decrypt patches, so
// we MUST wait for processMessage to actually succeed before kicking off
// the sync — otherwise it would hit isMissingKeyError and park
// collections in blockedCollections, regressing the very issue this
// branch was added to fix.
//
// No deadlock with the inbound caller's messageMutex because
// postUpsertMutex is a different mutex instance.
logger.info('App state sync key arrived, awaiting persistence before triggering sync')
const { processMessageOk } = await postUpsertWork
if (!processMessageOk) {
logger?.warn(
{ messageId: msg.key?.id, remoteJid: msg.key?.remoteJid },
'processMessage failed during key-share — skipping doAppStateSync to avoid isMissingKeyError'
)
} else {
try {
// If the app state key arrives and we are waiting to sync, trigger the sync now.
if (msg.message?.protocolMessage?.appStateSyncKeyShare && syncState === SyncState.Syncing) {
logger.info('App state sync key arrived, triggering app state sync')
await doAppStateSync()
} catch (err) {
logger?.warn(
{ err, messageId: msg.key?.id, remoteJid: msg.key?.remoteJid },
'doAppStateSync failed after key-share persistence'
)
}
}
} else {
// `postUpsertWork` is not expected to reject — `Promise.allSettled`
// inside `postUpsertTasks` never rejects, and per-task failures are
// already logged inline. The defensive catch routes any truly
// unexpected rejection (e.g. `postUpsertMutex` internal corruption,
// future synchronous throws inside processMessage) through
// `onUnexpectedError` instead of letting it surface as an
// UnhandledPromiseRejection — which on Node ≥15 can terminate the
// long-running socket process.
postUpsertWork.catch(err =>
onUnexpectedError(
err,
`processing post-upsert work for message ${msg.key?.id || 'unknown'} on ${msg.key?.remoteJid || 'unknown chat'}`
)
)
}
})
@@ -1776,7 +1612,6 @@ export const makeChatsSocket = (config: SocketConfig) => {
return {
...sock,
serverProps,
createCallLink,
getBotListV2,
messageMutex,
-6
View File
@@ -343,13 +343,11 @@ export const extractGroupMetadata = (result: BinaryNode) => {
let descId: string | undefined
let descOwner: string | undefined
let descOwnerPn: string | undefined
let descOwnerUsername: string | undefined
let descTime: number | undefined
if (descChild) {
desc = getBinaryNodeChildString(descChild, 'body')
descOwner = descChild.attrs.participant ? jidNormalizedUser(descChild.attrs.participant) : undefined
descOwnerPn = descChild.attrs.participant_pn ? jidNormalizedUser(descChild.attrs.participant_pn) : undefined
descOwnerUsername = descChild.attrs.participant_username || undefined
descTime = +descChild.attrs.t!
descId = descChild.attrs.id
}
@@ -364,19 +362,16 @@ export const extractGroupMetadata = (result: BinaryNode) => {
subject: group.attrs.subject!,
subjectOwner: group.attrs.s_o,
subjectOwnerPn: group.attrs.s_o_pn,
subjectOwnerUsername: group.attrs.s_o_username,
subjectTime: +(group.attrs.s_t ?? '0'),
size: group.attrs.size ? +group.attrs.size : getBinaryNodeChildren(group, 'participant').length,
creation: +(group.attrs.creation ?? '0'),
owner: group.attrs.creator ? jidNormalizedUser(group.attrs.creator) : undefined,
ownerPn: group.attrs.creator_pn ? jidNormalizedUser(group.attrs.creator_pn) : undefined,
ownerUsername: group.attrs.creator_username || undefined,
owner_country_code: group.attrs.creator_country_code,
desc,
descId,
descOwner,
descOwnerPn,
descOwnerUsername,
descTime,
linkedParent: getBinaryNodeChild(group, 'linked_parent')?.attrs.jid || undefined,
restrict: !!getBinaryNodeChild(group, 'locked'),
@@ -391,7 +386,6 @@ export const extractGroupMetadata = (result: BinaryNode) => {
id: attrs.jid!,
phoneNumber: isLidUser(attrs.jid) && isPnUser(attrs.phone_number) ? attrs.phone_number : undefined,
lid: isPnUser(attrs.jid) && isLidUser(attrs.lid) ? attrs.lid : undefined,
username: attrs.participant_username || attrs.username || undefined,
admin: (attrs.type || null) as GroupParticipant['admin']
}
}),
+151 -284
View File
@@ -7,7 +7,6 @@ import { proto } from '../../WAProto/index.js'
import {
DEFAULT_CACHE_TTLS,
DEFAULT_SESSION_CLEANUP_CONFIG,
INITIAL_PREKEY_COUNT,
KEY_BUNDLE_TYPE,
MIN_PREKEY_COUNT,
PLACEHOLDER_MAX_AGE_SECONDS,
@@ -49,12 +48,9 @@ import {
getStatusFromReceiptType,
handleIdentityChange,
hkdf,
BAD_MAC_ERROR_TEXT,
DECRYPTION_RETRY_CONFIG,
MISSING_KEYS_ERROR_TEXT,
NACK_REASONS,
NO_MESSAGE_FOUND_ERROR_TEXT,
RetryReason,
normalizeKeyLidToPn,
normalizeMessageJids,
resolveLidToPn,
@@ -73,14 +69,7 @@ import {
recordMessageReceived,
recordMessageRetry
} from '../Utils/prometheus-metrics.js'
import {
buildMergedTcTokenIndexWrite,
isTcTokenExpired,
resolveIssuanceJid,
resolveTcTokenJid,
storeTcTokensFromIqResult,
TC_TOKEN_INDEX_KEY
} from '../Utils/tc-token-utils'
import { isTcTokenExpired, resolveTcTokenJid, storeTcTokensFromIqResult } from '../Utils/tc-token-utils'
import {
areJidsSameUser,
type BinaryNode,
@@ -171,12 +160,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
let sendActiveReceipts = false
// ======= tctoken index tracking for cross-session pruning =======
// TC_TOKEN_INDEX_KEY is imported from tc-token-utils so the value stays in sync
// with messages-send/process-message writes (avoids string drift on rename).
// Race note: this prune-driven index write may interleave with
// buildMergedTcTokenIndexWrite calls from issuance/history-sync paths. Worst
// case: a JID resurrected by a stale read gets pruned again on the next 24h
// sweep — no data loss, just one extra cycle.
const TC_TOKEN_INDEX_KEY = '__index'
const TC_TOKEN_PRUNE_TS_KEY = '__prune_ts'
const tcTokenKnownJids = new Set<string>()
const tcTokenRetriedMsgIds = new Set<string>()
@@ -210,25 +194,16 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
}
})()
/**
* Debounced save of the tctoken JID index (5s).
*
* Merges with the persisted index instead of overwriting — other layers
* (messages-send fire-and-forget issuance, process-message history sync) may
* write JIDs to the index via `buildMergedTcTokenIndexWrite` without updating
* `tcTokenKnownJids`. Without the merge those JIDs would be silently dropped
* the next time this debounced save fires.
*/
/** Debounced save of the tctoken JID index (5s) */
const scheduleTcTokenIndexSave = () => {
if (tcTokenIndexSaveTimer) clearTimeout(tcTokenIndexSaveTimer)
tcTokenIndexSaveTimer = setTimeout(async () => {
try {
const merged = await buildMergedTcTokenIndexWrite(authState.keys, tcTokenKnownJids)
const arr = Array.from(tcTokenKnownJids)
await authState.keys.set({
tctoken: {
...merged,
[TC_TOKEN_INDEX_KEY]: {
...merged[TC_TOKEN_INDEX_KEY],
token: Buffer.from(JSON.stringify(arr), 'utf8'),
timestamp: unixTimestampSeconds().toString()
}
}
@@ -1237,7 +1212,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
return await query(stanza)
}
const sendRetryRequest = async (node: BinaryNode, forceIncludeKeys = false, decryptionError?: string) => {
const sendRetryRequest = async (node: BinaryNode, forceIncludeKeys = false) => {
const { fullMessage } = decodeMessageNode(node, authState.creds.me!.id, authState.creds.me!.lid || '')
const { key: msgKey } = fullMessage
const msgId = msgKey.id!
@@ -1335,39 +1310,39 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
const { account, signedPreKey, signedIdentityKey: identityKey } = authState.creds
const fromJid = node.attrs.from!
// Derive the Signal error code from the actual decryption failure message.
// Sent in the retry receipt so the peer (even another InfiniteAPI instance)
// knows the exact failure type and can recreate the session immediately
// instead of falling back to the 1-hour timeout.
//
// Codes mirror RetryReason enum in message-retry-manager.ts:
// 0 = UnknownError | 1 = NoSession | 2 = InvalidKey
// 3 = InvalidKeyId | 4 = InvalidMessage | 7 = BadMac
//
// Uses DECRYPTION_RETRY_CONFIG error lists (single source of truth in
// decode-wa-message.ts) so additions to those lists are picked up here
// automatically.
//
// NOTE: We do NOT delete the session here (receiver side). The Signal Protocol
// recovers automatically when the sender's pkmsg arrives — it overwrites the
// corrupted session. Deleting prematurely creates a race window where no session
// exists, which can cause "No Session" errors on concurrent messages.
const retryErrorCode = (() => {
if (!decryptionError) return RetryReason.UnknownError
// Bad MAC must be checked first — it is also in corruptedSessionErrors
// but warrants the more specific code 7 over the generic code 4.
if (decryptionError.includes(BAD_MAC_ERROR_TEXT)) return RetryReason.SignalErrorBadMac
// MessageCounterError and other corrupted-session variants
if (DECRYPTION_RETRY_CONFIG.corruptedSessionErrors.some(e => decryptionError.includes(e))) return RetryReason.SignalErrorInvalidMessage
// Missing / invalid session record
if (DECRYPTION_RETRY_CONFIG.sessionRecordErrors.some(e => decryptionError.includes(e)) ||
/no\s+(open\s+)?sessions?/i.test(decryptionError)) return RetryReason.SignalErrorNoSession
// PreKey / key-id errors
if (/pre\s*key/i.test(decryptionError)) return RetryReason.SignalErrorInvalidKeyId
// Identity / key errors
if (/invalid\s*key|untrusted\s*identity/i.test(decryptionError)) return RetryReason.SignalErrorInvalidKey
return RetryReason.UnknownError
})()
// Check if we should recreate the session
let shouldRecreateSession = false
let recreateReason = ''
if (enableAutoSessionRecreation && messageRetryManager && retryCount >= 1) {
try {
// Check if we have a session with this JID
const sessionId = signalRepository.jidToSignalProtocolAddress(fromJid)
const hasSession = await signalRepository.validateSession(fromJid)
// Extract error code from retry node if present (for MAC error detection)
const retryNode = getBinaryNodeChild(node, 'retry')
const errorAttr = retryNode?.attrs?.error
const errorCode = messageRetryManager.parseRetryErrorCode(errorAttr)
const result = messageRetryManager.shouldRecreateSession(fromJid, hasSession.exists, errorCode)
shouldRecreateSession = result.recreate
recreateReason = result.reason
if (shouldRecreateSession) {
logger.debug({ fromJid, retryCount, reason: recreateReason, errorCode }, 'recreating session for retry')
// Delete existing session to force recreation
// CRITICAL: Use same transaction key as encrypt/decrypt operations to prevent race
// Using meId ensures this delete serializes with sendMessage() and other session operations
await authState.keys.transaction(async () => {
await authState.keys.set({ session: { [sessionId]: null } })
}, authState.creds.me?.id || 'session-operation')
forceIncludeKeys = true
}
} catch (error) {
logger.warn({ error, fromJid }, 'failed to check session recreation')
}
}
if (retryCount <= 2) {
// Use new retry manager for phone requests if available
@@ -1408,7 +1383,8 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
id: node.attrs.id!,
t: node.attrs.t!,
v: '1',
error: retryErrorCode.toString()
// ADD ERROR FIELD
error: '0'
}
},
{
@@ -1427,7 +1403,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
receipt.attrs.participant = node.attrs.participant
}
if (retryCount > 1 || forceIncludeKeys) {
if (retryCount > 1 || forceIncludeKeys || shouldRecreateSession) {
const { update, preKeys } = await getNextPreKeys(authState, 1)
const [keyId] = Object.keys(preKeys)
@@ -1455,58 +1431,6 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
}, authState?.creds?.me?.id || 'sendRetryRequest')
}
/**
* Fire-and-forget tctoken re-issuance after a peer's device identity changed.
* Mirrors WAWebSendTcTokenWhenDeviceIdentityChange — runs in PARALLEL with the
* session refresh (not after it).
*
* Why parallel and not sequential:
* - WA Web invokes this BEFORE assertSessions to maximise the chance the contact
* has a fresh tctoken by the time the next outbound send executes.
* - Running after assertSessions (the fork's previous behaviour) races with the
* next send and risks error 463 when the contact reinstalls and the user replies
* immediately afterwards.
*
* Gated on `entry.senderTimestamp` (we previously issued a token to this peer in the
* current bucket window). Preserves the existing senderTimestamp instead of issuing
* a fresh one — keeps WA Web's bucket coalescing semantics: same token, same window.
*/
const reissueTcTokenAfterIdentityChange = (from: string): void => {
void (async () => {
const normalizedJid = jidNormalizedUser(from)
const tcJid = await resolveTcTokenJid(normalizedJid, getLIDForPN)
const tcTokenData = await authState.keys.get('tctoken', [tcJid])
const senderTs = tcTokenData?.[tcJid]?.senderTimestamp
if (senderTs === null || senderTs === undefined || isTcTokenExpired(senderTs)) {
return
}
logTcToken('reissue', { jid: normalizedJid, reason: 'identity_changed', senderTimestamp: senderTs })
const getPNForLID = signalRepository.lidMapping.getPNForLID.bind(signalRepository.lidMapping)
const issueJid = await resolveIssuanceJid(
normalizedJid,
sock.serverProps.lidTrustedTokenIssueToLid,
getLIDForPN,
getPNForLID
)
const result = await getPrivacyTokens([issueJid], senderTs)
await storeTcTokensFromIqResult({
result,
fallbackJid: tcJid,
keys: authState.keys,
getLIDForPN,
onNewJidStored: storedJid => {
tcTokenKnownJids.add(storedJid)
scheduleTcTokenIndexSave()
}
})
logTcToken('reissue_ok', { jid: normalizedJid, reason: 'identity_changed' })
})().catch(err => {
logTcToken('reissue_fail', { jid: from, error: err?.message })
})
}
const handleEncryptNotification = async (node: BinaryNode) => {
const from = node.attrs.from
if (from === S_WHATSAPP_NET) {
@@ -1516,8 +1440,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
logger.debug({ count, shouldUploadMorePreKeys }, 'recv pre-key count')
if (shouldUploadMorePreKeys) {
// Top-up back to INITIAL_PREKEY_COUNT so the pool is always restored to full size
await uploadPreKeys(Math.max(1, INITIAL_PREKEY_COUNT - count))
await uploadPreKeys()
}
} else {
const result = await handleIdentityChange(node, {
@@ -1526,12 +1449,57 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
validateSession: signalRepository.validateSession,
assertSessions,
debounceCache: identityAssertDebounce,
logger,
// Fire reissue in parallel with the session refresh, NOT after it.
// See reissueTcTokenAfterIdentityChange doc for the rationale.
onBeforeSessionRefresh: reissueTcTokenAfterIdentityChange
logger
})
// When a session is refreshed (identity change), re-issue tctoken fire-and-forget
// WABA Android: reissue stores senderTimestamp + realIssueTimestamp after IQ success
if (result.action === 'session_refreshed') {
const normalizedJid = jidNormalizedUser(from)
resolveTcTokenJid(normalizedJid, getLIDForPN)
.then(async tcJid => {
const tcData = await authState.keys.get('tctoken', [tcJid])
const entry = tcData[tcJid]
if (entry?.token?.length && !isTcTokenExpired(entry.timestamp)) {
const senderTs = unixTimestampSeconds()
logTcToken('reissue', { jid: normalizedJid, reason: 'session_refreshed' })
getPrivacyTokens([normalizedJid], senderTs)
.then(async (iqResult) => {
await storeTcTokensFromIqResult({
result: iqResult,
fallbackJid: normalizedJid,
keys: authState.keys,
getLIDForPN,
onNewJidStored: (storedJid) => {
tcTokenKnownJids.add(storedJid)
scheduleTcTokenIndexSave()
}
})
// Persist senderTimestamp + realIssueTimestamp after IQ success
const currentData = await authState.keys.get('tctoken', [tcJid])
const currentEntry = currentData[tcJid]
await authState.keys.set({
tctoken: {
[tcJid]: {
...currentEntry,
token: currentEntry?.token ?? Buffer.alloc(0),
senderTimestamp: senderTs,
realIssueTimestamp: 0
}
}
})
logTcToken('reissue_ok', { jid: normalizedJid, reason: 'session_refreshed' })
})
.catch(err => {
logTcToken('reissue_fail', { jid: normalizedJid, error: err?.message })
})
}
})
.catch(() => {
/* ignore resolution errors */
})
}
if (result.action === 'no_identity_node') {
logger.info({ node }, 'unknown encrypt notification')
}
@@ -2152,6 +2120,12 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
if (resolvedRemoteJid) key.remoteJid = resolvedRemoteJid
if (resolvedParticipant) key.participant = resolvedParticipant
if (shouldIgnoreJid(remoteJid!) && remoteJid !== S_WHATSAPP_NET) {
logger.trace({ remoteJid }, 'ignoring receipt from jid')
await sendMessageAck(node)
return
}
const ids = [attrs.id!]
if (Array.isArray(content)) {
const items = getBinaryNodeChildren(content[0], 'item')
@@ -2227,6 +2201,11 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
const handleNotification = async (node: BinaryNode) => {
const remoteJid = node.attrs.from
if (shouldIgnoreJid(remoteJid!) && remoteJid !== S_WHATSAPP_NET) {
logger.trace({ remoteJid }, 'ignored notification')
await sendMessageAck(node)
return
}
try {
await Promise.all([
@@ -2240,7 +2219,6 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
fromMe,
participant: node.attrs.participant,
participantAlt,
participantUsername: node.attrs.participant_username,
addressingMode,
id: node.attrs.id,
...(msg.key || {})
@@ -2249,10 +2227,6 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
msg.messageTimestamp = +node.attrs.t!
const fullMsg = proto.WebMessageInfo.fromObject(msg) as WAMessage
// Preserve custom WAMessageKey fields (participantAlt, participantUsername,
// addressingMode) that proto.WebMessageInfo.fromObject strips because
// they aren't part of the proto.MessageKey schema.
fullMsg.key = msg.key
await upsertMessage(fullMsg, 'append')
}
})
@@ -2263,59 +2237,23 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
}
const handleMessage = async (node: BinaryNode) => {
const encNode = getBinaryNodeChild(node, 'enc')
const unavailableNode = getBinaryNodeChild(node, 'unavailable')
// msmsg block removed — allow msmsg decryption (required for view-once on linked devices)
if (unavailableNode?.attrs?.type === 'view_once') {
const { fullMessage: voMsg } = decryptMessageNode(
node,
authState.creds.me!.id,
authState.creds.me!.lid || '',
signalRepository,
logger,
)
voMsg.key.isViewOnce = true
// Try to request the actual view-once content from the primary phone via PDO
// The phone may resend the full media (url, mediaKey, directPath, etc.)
console.log(`[VO_PDO] Requesting view-once content via PDO for ${voMsg.key.id} from ${voMsg.key.remoteJid}`)
try {
const requestId = await requestPlaceholderResend(voMsg.key, {
key: { ...voMsg.key },
pushName: voMsg.pushName,
messageTimestamp: voMsg.messageTimestamp,
participant: voMsg.key.participant,
participantAlt: voMsg.key.participantAlt
})
if (requestId && requestId !== 'RESOLVED') {
console.log(`[VO_PDO] PDO request sent! requestId=${requestId} — waiting for phone to resend`)
} else if (requestId === 'RESOLVED') {
console.log(`[VO_PDO] Message arrived during PDO delay!`)
} else {
console.log(`[VO_PDO] Already requested or no response`)
}
} catch (err: any) {
console.log(`[VO_PDO] PDO request failed: ${err.message} — falling back to placeholder`)
}
// Emit placeholder immediately so consumers see the notification
// If PDO succeeds, the real message will arrive via messages.upsert later
voMsg.message = {
viewOnceMessage: {
message: {
imageMessage: {
viewOnce: true,
}
}
}
}
const upsertType = node.attrs.offline ? 'append' : 'notify'
await upsertMessage(voMsg, upsertType)
if (shouldIgnoreJid(node.attrs.from!) && node.attrs.from !== S_WHATSAPP_NET) {
logger.trace({ from: node.attrs.from }, 'ignored message')
// Send a clean ACK (no error code) so the server considers the
// message delivered. Using error 500 (UnhandledError) previously
// caused the server to retry delivery, generating duplicate traffic.
await sendMessageAck(node)
return
}
const encNode = getBinaryNodeChild(node, 'enc')
// TODO: temporary fix for crashes and issues resulting of failed msmsg decryption
if (encNode?.attrs.type === 'msmsg') {
logger.debug({ key: node.attrs.key }, 'ignored msmsg')
await sendMessageAck(node, NACK_REASONS.MissingMessageSecret)
return
}
const {
fullMessage: msg,
category,
@@ -2330,54 +2268,44 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
)
const alt = msg.key.participantAlt || msg.key.remoteJidAlt
// Handle LID/PN mappings with optimized hot-path:
// - storeLIDPNMappings is fire-and-forget (background) — does NOT block decrypt
// - migrateSession is SYNC (await) — REQUIRED for decrypt to find session
//
// SAFETY: normalizeMessageJids has a fast-path that uses key.*Alt directly without
// hitting the store, so the just-arrived message normalizes correctly even before
// the background store completes. Subsequent messages in the same chat hit the
// store after the background write is done (ms later).
//
// Pre-check (getPNForLID/getLIDForPN) was removed — storeLIDPNMappings has internal
// LRU cache + dedup, the pre-check was a redundant store round-trip per inbound
// message that added latency under load.
//
// HISTORICAL: this restores the intent of d73cd28d39 (2026-02-03) which was
// partially reverted by c3fc792351 the same day due to a race-condition concern
// with migrateSession (kept sync here). storeLIDPNMappings was over-protected:
// it persists a mapping that downstream consumers can re-derive from key.*Alt,
// while migrateSession actually moves the Signal session record that decrypt()
// will load microseconds later — those two have very different criticality.
//
// DO NOT make migrateSession async — decrypt() depends on the session being at
// the correct identifier (LID vs PN) when it runs. Other code paths (USync
// device lookup in messages-send.ts) create LID/PN mappings without migrating
// the session, so we cannot skip migration even when the mapping already exists.
// Handle LID/PN mappings with hybrid approach:
// - Store mapping operation runs in background (non-critical for decrypt)
// - Session migration MUST complete before decrypt() to avoid "No session record" errors
// This addresses Codex/Copilot review concerns about race conditions with decrypt()
if (!!alt) {
const altServer = jidDecode(alt)?.server
const primaryJid = msg.key.participant || msg.key.remoteJid!
if (altServer === 'lid') {
// Fire-and-forget: storeLIDPNMappings has internal cache+dedup,
// pre-check (getPNForLID) was redundant.
signalRepository.lidMapping
// Check if mapping already exists to avoid unnecessary storage operations
const existingMapping = await signalRepository.lidMapping.getPNForLID(alt)
if (!existingMapping) {
// MUST await: normalizeMessageJids() runs after this and needs the mapping
// in the LIDMappingStore to resolve LID→PN for events delivered to consumers
await signalRepository.lidMapping
.storeLIDPNMappings([{ lid: alt, pn: primaryJid }])
.catch(error => logger.warn({ error, alt, primaryJid }, 'background LID mapping store failed'))
.catch(error => logger.warn({ error, alt, primaryJid }, 'LID mapping storage failed'))
}
// CRITICAL: ALWAYS migrate session SYNC, even if mapping exists.
// Other code paths (e.g., USync device lookup in messages-send.ts) may create
// mappings via storeLIDPNMappings() without calling migrateSession(). This
// leaves sessions under PN format while decrypt() expects LID format.
// Skipping migration based on mapping existence causes "No session record" errors.
// CRITICAL: ALWAYS migrate session, even if mapping exists
// Other code paths (e.g., USync device lookup in messages-send.ts:310-319)
// may create mappings via storeLIDPNMappings() without calling migrateSession()
// This leaves sessions under PN format while decrypt() expects LID format
// Skipping migration based on mapping existence causes "No session record" errors
await signalRepository.migrateSession(primaryJid, alt)
} else {
// Fire-and-forget: same rationale as above.
signalRepository.lidMapping
// Check if reverse mapping exists
const existingMapping = await signalRepository.lidMapping.getLIDForPN(alt)
if (!existingMapping) {
// MUST await: normalizeMessageJids() runs after this and needs the mapping
// in the LIDMappingStore to resolve LID→PN for events delivered to consumers
await signalRepository.lidMapping
.storeLIDPNMappings([{ lid: primaryJid, pn: alt }])
.catch(error => logger.warn({ error, alt, primaryJid }, 'background LID mapping store failed'))
.catch(error => logger.warn({ error, alt, primaryJid }, 'LID mapping storage failed'))
}
// CRITICAL: ALWAYS migrate session SYNC.
// CRITICAL: ALWAYS migrate session, even if mapping exists
// Same reasoning as above - mapping existence doesn't guarantee session migration
await signalRepository.migrateSession(alt, primaryJid)
}
}
@@ -2439,23 +2367,6 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
{ msgId: msg.key?.id, messageType },
'CTWA: Skipping placeholder resend for unavailable fanout type'
)
// For view-once: emit the message with a viewOnceMessage placeholder
// so consumers (zpro, etc.) can detect and display it.
// Include a message wrapper so isValidMsg accepts it.
if (messageType === 'view_once_unavailable_fanout') {
msg.key.isViewOnce = true
msg.message = {
viewOnceMessage: {
message: {
imageMessage: {
viewOnce: true,
}
}
}
}
const upsertType = node.attrs.offline ? 'append' : 'notify'
await upsertMessage(msg, upsertType)
}
metrics.ctwaRecoveryFailures.inc({ reason: 'unavailable_fanout' })
return sendMessageAck(node)
}
@@ -2594,7 +2505,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
}
const encNode = getBinaryNodeChild(node, 'enc')
await sendRetryRequest(node, !encNode, errorMessage)
await sendRetryRequest(node, !encNode)
if (retryRequestDelayMs) {
await delay(retryRequestDelayMs)
}
@@ -2603,7 +2514,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
// Still attempt retry even if pre-key upload failed
try {
const encNode = getBinaryNodeChild(node, 'enc')
await sendRetryRequest(node, !encNode, errorMessage)
await sendRetryRequest(node, !encNode)
} catch (retryErr) {
logger.error({ retryErr }, 'Failed to send retry after error handling')
}
@@ -2637,11 +2548,6 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
await sendReceipt(msg.key.remoteJid!, participant!, [msg.key.id!], type)
// NOTE: do NOT send view_once_read from a linked device (API).
// WA shows view-once as view_once_unavailable_fanout on linked devices — only the
// primary phone sends view_once_read when the user actually opens it.
// Sending it from a linked device causes WA to flag the account and disables view-once.
// send ack for history message
const isAnyHistoryMsg = getHistoryMsg(msg.message!)
if (isAnyHistoryMsg) {
@@ -3126,40 +3032,6 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
identifier: string,
exec: (node: BinaryNode) => Promise<void>
) => {
// Fast path: ack and drop ignored JIDs before entering the buffer/queue.
// Skips type='call' so call events are never silently dropped via
// shouldIgnoreJid (preserves InfiniteAPI's pre-existing behavior).
// Wrapped in try/catch so a throw from shouldIgnoreJid (user callback)
// or sendMessageAck (e.g. websocket closed) is routed through
// onUnexpectedError instead of becoming an unhandled rejection —
// matches the protection processNodeWithBuffer provides.
if (type !== 'call') {
try {
const from = node.attrs.from
let ignoreJid = from
if (type === 'receipt' && from) {
const attrs = node.attrs
const isLid = attrs.from!.includes('lid')
const isNodeFromMe = areJidsSameUser(
attrs.participant || attrs.from,
isLid ? authState.creds.me?.lid : authState.creds.me?.id
)
ignoreJid = !isNodeFromMe || isJidGroup(attrs.from) ? attrs.from : attrs.recipient
}
if (ignoreJid && ignoreJid !== S_WHATSAPP_NET && shouldIgnoreJid(ignoreJid)) {
// Plain ack (no NACK error code) preserves InfiniteAPI's prior
// behavior — ignored stanzas are an intentional drop, not a
// processing failure, so we don't want server-side retries.
await sendMessageAck(node)
return
}
} catch (error) {
onUnexpectedError(error as Error, identifier)
return
}
}
const isOffline = !!node.attrs.offline
if (isOffline) {
@@ -3258,24 +3130,19 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
// Await index load first — prevents overwriting a more complete persisted index
// if the connection closes before the initial load finishes.
tcTokenIndexLoaded
.then(async () => {
try {
// Same merge-with-persisted invariant as scheduleTcTokenIndexSave —
// other layers may have written cross-layer JIDs to the index since
// our in-memory set was last updated.
const merged = await buildMergedTcTokenIndexWrite(authState.keys, tcTokenKnownJids)
await authState.keys.set({
.then(() => {
Promise.resolve(
authState.keys.set({
tctoken: {
...merged,
[TC_TOKEN_INDEX_KEY]: {
...merged[TC_TOKEN_INDEX_KEY],
token: Buffer.from(JSON.stringify([...tcTokenKnownJids]), 'utf8'),
timestamp: unixTimestampSeconds().toString()
}
}
})
} catch {
).catch(() => {
/* non-critical */
}
})
})
.catch(() => {
/* non-critical */
+192 -1202
View File
File diff suppressed because it is too large Load Diff
+2 -2
View File
@@ -123,11 +123,11 @@ export const makeNewsletterSocket = (config: SocketConfig) => {
},
newsletterMute: (jid: string) => {
return executeWMexQuery({ input: { newsletter_id: jid, type: 'MUTE_ADMIN_ACTIVITY', value: 'OFF' } }, QueryIds.MUTE, XWAPaths.xwa2_newsletter_mute_v2)
return executeWMexQuery({ newsletter_id: jid }, QueryIds.MUTE, XWAPaths.xwa2_newsletter_mute_v2)
},
newsletterUnmute: (jid: string) => {
return executeWMexQuery({ input: { newsletter_id: jid, type: 'MUTE_ADMIN_ACTIVITY', value: 'ON' } }, QueryIds.UNMUTE, XWAPaths.xwa2_newsletter_unmute_v2)
return executeWMexQuery({ newsletter_id: jid }, QueryIds.UNMUTE, XWAPaths.xwa2_newsletter_unmute_v2)
},
newsletterUpdateName: async (jid: string, name: string) => {
+171 -119
View File
@@ -39,7 +39,12 @@ import {
xmppSignedPreKey
} from '../Utils'
import { getPlatformId, isAndroidBrowser } from '../Utils/browser-utils'
import { withBoundedRetry } from '../Utils/bounded-retry'
import {
CircuitBreaker,
CircuitOpenError,
createConnectionCircuitBreaker,
createPreKeyCircuitBreaker
} from '../Utils/circuit-breaker'
import {
decrementActiveConnections,
incrementActiveConnections,
@@ -89,6 +94,10 @@ export const makeSocket = (config: SocketConfig) => {
transactionOpts,
qrTimeout,
makeSignalRepository,
enableCircuitBreaker = true,
queryCircuitBreaker: queryCircuitBreakerConfig,
connectionCircuitBreaker: connectionCircuitBreakerConfig,
preKeyCircuitBreaker: preKeyCircuitBreakerConfig,
// If enableUnifiedSession is explicitly set (true/false), use it
// Otherwise (undefined), check env var, then default to true
enableUnifiedSession: enableUnifiedSessionConfig
@@ -98,6 +107,57 @@ export const makeSocket = (config: SocketConfig) => {
const enableUnifiedSession =
enableUnifiedSessionConfig !== undefined ? enableUnifiedSessionConfig : shouldEnableUnifiedSession()
// Initialize circuit breakers if enabled
let queryCircuitBreaker: CircuitBreaker | undefined
let connectionCircuitBreaker: CircuitBreaker | undefined
let preKeyCircuitBreaker: CircuitBreaker | undefined
if (enableCircuitBreaker) {
// Circuit breaker for query operations (most critical)
queryCircuitBreaker = createConnectionCircuitBreaker({
name: 'socket-query',
failureThreshold: 5,
failureWindow: 60000,
resetTimeout: 30000,
successThreshold: 2,
timeout: defaultQueryTimeoutMs || 60000,
onStateChange: (from, to) => {
logger.info({ from, to }, 'Query circuit breaker state changed')
},
onOpen: () => {
logger.warn('Query circuit breaker OPENED - blocking requests')
},
onClose: () => {
logger.info('Query circuit breaker CLOSED - resuming normal operation')
},
...queryCircuitBreakerConfig
})
// Circuit breaker for connection operations
connectionCircuitBreaker = createConnectionCircuitBreaker({
name: 'socket-connection',
failureThreshold: 3,
failureWindow: 30000,
resetTimeout: 60000,
successThreshold: 1,
onStateChange: (from, to) => {
logger.info({ from, to }, 'Connection circuit breaker state changed')
},
...connectionCircuitBreakerConfig
})
// Circuit breaker for pre-key operations
preKeyCircuitBreaker = createPreKeyCircuitBreaker({
name: 'socket-prekey',
onStateChange: (from, to) => {
logger.info({ from, to }, 'PreKey circuit breaker state changed')
},
...preKeyCircuitBreakerConfig
})
logger.info('Circuit breakers initialized for socket operations')
}
// Unified Session Manager will be initialized after sendNode is defined
let unifiedSessionManager: UnifiedSessionManager | undefined
@@ -178,17 +238,20 @@ export const makeSocket = (config: SocketConfig) => {
})
}
/**
* Send a raw buffer over the WebSocket.
*
* Replaces the previous connectionCircuitBreaker wrapping. WebSocket
* lifecycle errors (Connection Closed / Lost / Replaced) were already
* filtered out of the circuit's failure count, so the only failures it
* tripped on were rare native send() errors for which a state-machine
* gate adds no value over a fast propagated error. Reconnection logic
* in makeSocket already handles WS recovery independently.
*/
/** send a raw buffer with circuit breaker protection */
const sendRawMessage = async (data: Uint8Array | Buffer) => {
if (connectionCircuitBreaker) {
try {
return await connectionCircuitBreaker.execute(() => sendRawMessageInternal(data))
} catch (error) {
if (error instanceof CircuitOpenError) {
logger.warn({ circuitName: error.circuitName }, 'Send blocked by connection circuit breaker')
}
throw error
}
}
return sendRawMessageInternal(data)
}
@@ -203,7 +266,7 @@ export const makeSocket = (config: SocketConfig) => {
}
// Initialize Unified Session Manager now that sendNode is defined
// (single initialization to avoid duplicating manager state)
// (single initialization to avoid duplicating circuit breakers and state)
if (enableUnifiedSession) {
const sendNodeForSession = async (node: BinaryNode): Promise<void> => {
await sendNode(node)
@@ -212,6 +275,7 @@ export const makeSocket = (config: SocketConfig) => {
unifiedSessionManager = createUnifiedSessionManager({
enabled: true,
logger,
enableCircuitBreaker,
sendNode: sendNodeForSession
})
logger.info('Unified session manager initialized')
@@ -282,8 +346,11 @@ export const makeSocket = (config: SocketConfig) => {
// Register the response listener BEFORE sending — avoids a race where the server
// responds before we start listening. waitForMessage already handles its own
// timeout (returns undefined) and connection-close errors (throws), so we do NOT
// wrap it in a second promiseTimeout. The outer wrapper would race the inner
// timeout near the deadline and throw a spurious Boom('Timed Out').
// wrap it in a second promiseTimeout. The outer wrapper caused a race condition
// where both timers fired at ~the same deadline: the outer one threw
// Boom('Timed Out') while sendNode was still pending, producing a spurious error
// whose message contained "socket-query" → matched the circuit-breaker's
// "socket" pattern → incorrectly tripped the breaker after 5 timeouts.
const responsePromise = waitForMessage<any>(msgId, timeoutMs)
// Prevent unhandled-rejection if sendNode throws before we reach
// `await responsePromise` below. The error from sendNode still propagates
@@ -305,17 +372,23 @@ export const makeSocket = (config: SocketConfig) => {
return result
}
/**
* Send a query (iq stanza) and wait for its response.
*
* Replaces the previous queryCircuitBreaker wrapping (state-machine that
* blocked ALL queries for 30s after 5 failures). The cascade behavior
* was incompatible with WhatsApp Android's empirical retry pattern:
* each operation retries independently with exponential backoff, no
* global state. Callers that need retry semantics wrap query() with
* withBoundedRetry() at their level (assertSessions, etc.).
*/
/** send a query with circuit breaker protection */
const query = async (node: BinaryNode, timeoutMs?: number) => {
// If circuit breaker is enabled, wrap the query
if (queryCircuitBreaker) {
try {
return await queryCircuitBreaker.execute(() => queryInternal(node, timeoutMs))
} catch (error) {
// If circuit is open, log and rethrow with context
if (error instanceof CircuitOpenError) {
logger.warn({ circuitName: error.circuitName, state: error.state }, 'Query blocked by circuit breaker')
}
throw error
}
}
// Fallback to direct query if circuit breaker is disabled
return queryInternal(node, timeoutMs)
}
@@ -612,33 +685,30 @@ export const makeSocket = (config: SocketConfig) => {
let lastUploadTime = 0
/** generates and uploads a set of pre-keys to the server */
const uploadPreKeys = async (count = MIN_PREKEY_COUNT) => {
// Check minimum interval. (Previously this was guarded by `retryCount === 0`
// because the function recursed on retry; with bounded-retry handling
// retries internally there is no recursion and the guard is implicit.)
const uploadPreKeys = async (count = MIN_PREKEY_COUNT, retryCount = 0) => {
// Check if pre-key circuit breaker is open
if (preKeyCircuitBreaker?.isOpen()) {
logger.warn('PreKey circuit breaker is open, skipping upload')
throw new CircuitOpenError('socket-prekey', 'open')
}
// Check minimum interval (except for retries)
if (retryCount === 0) {
const timeSinceLastUpload = Date.now() - lastUploadTime
if (timeSinceLastUpload < MIN_UPLOAD_INTERVAL) {
logger.debug(`Skipping upload, only ${timeSinceLastUpload}ms since last upload`)
return
}
}
// Prevent multiple concurrent uploads — if one is already running, wait for it and return:
// the concurrent upload already replenished the pool, so there is nothing left to do.
// Prevent multiple concurrent uploads
if (uploadPreKeysPromise) {
logger.debug('Pre-key upload already in progress, waiting for completion')
await uploadPreKeysPromise
return
}
// Shared abort controller so the outer Promise.race can cancel the
// in-flight bounded-retry loop if its own timeout fires first.
// Without this, the outer race rejects but the bounded-retry loop
// keeps running in the background, mutating lastUploadTime / logs
// after the caller has given up (CodeRabbit + Copilot reviews).
const uploadAbort = new AbortController()
const uploadLogic = async () => {
logger.info({ count }, 'uploading pre-keys')
logger.info({ count, retryCount }, 'uploading pre-keys')
// Generate and save pre-keys atomically (prevents ID collisions on retry)
const node = await keys.transaction(async () => {
@@ -649,69 +719,45 @@ export const makeSocket = (config: SocketConfig) => {
return node // Only return node since update is already used
}, creds?.me?.id || 'upload-pre-keys')
// Upload to server. Bounded-retry handles backoff (replaces both
// the previous circuit breaker AND the manual exponential backoff
// retry loop with maxRetries=3).
//
// CRITICAL: query() returns `undefined` on timeout (it does NOT
// throw — see waitForMessage). If we just `await query(node)`,
// a timed-out upload would resolve with undefined and bounded-retry
// would treat it as success. Validate the response shape so a
// timeout actually triggers a retry (CodeRabbit review on PR #393).
//
// bounded-retry's ttlMs MUST be < UPLOAD_TIMEOUT (the outer
// Promise.race below). With margin (28s vs 30s) bounded-retry
// always reaches its natural give-up first. We additionally pass
// an AbortController signal so that if the outer race ever fires
// first, the bounded-retry loop is aborted and does not keep
// running in the background (CodeRabbit + Copilot reviews).
const PER_ATTEMPT_TIMEOUT_MS = 8_000
const RETRY_TTL_MS = UPLOAD_TIMEOUT - 2_000
const uploadToServer = async (signal?: AbortSignal) => {
const result = await query(node, PER_ATTEMPT_TIMEOUT_MS)
if (signal?.aborted) {
throw new Error('aborted')
}
if (!result) {
// query() returned undefined → underlying iq response
// timed out. Throw so bounded-retry retries.
throw new Boom('Pre-key upload query timed out (no response)', { statusCode: 408 })
}
// Upload to server with circuit breaker protection
const uploadToServer = async () => {
await query(node)
logger.info({ count }, 'uploaded pre-keys successfully')
lastUploadTime = Date.now()
}
try {
await withBoundedRetry(uploadToServer, {
name: 'uploadPreKeys',
// 1s -> 2s -> 4s -> 8s (cap). Max 4 attempts fit comfortably
// within RETRY_TTL_MS (28s) given perAttemptTimeoutMs=8s.
delays: [1000, 2000, 4000, 8000],
ttlMs: RETRY_TTL_MS,
perAttemptTimeoutMs: PER_ATTEMPT_TIMEOUT_MS,
signal: uploadAbort.signal,
logger
})
// Use circuit breaker if available
if (preKeyCircuitBreaker) {
await preKeyCircuitBreaker.execute(uploadToServer)
} else {
await uploadToServer()
}
} catch (uploadError) {
logger.error({ uploadError: (uploadError as Error).toString(), count }, 'Failed to upload pre-keys to server')
// Don't retry if circuit breaker is open
if (uploadError instanceof CircuitOpenError) {
throw uploadError
}
// Exponential backoff retry (max 3 retries)
if (retryCount < 3) {
const backoffDelay = Math.min(1000 * Math.pow(2, retryCount), 10000)
logger.info(`Retrying pre-key upload in ${backoffDelay}ms`)
await new Promise(resolve => setTimeout(resolve, backoffDelay))
return uploadPreKeys(count, retryCount + 1)
}
throw uploadError
}
}
// Outer timeout protection. With ttlMs=28s vs UPLOAD_TIMEOUT=30s,
// bounded-retry should always win — but if anything ever blocks
// uploadLogic outside bounded-retry's reach, this race aborts the
// whole thing cleanly via uploadAbort.
// Add timeout protection
uploadPreKeysPromise = Promise.race([
uploadLogic(),
new Promise<void>((_, reject) =>
setTimeout(() => {
uploadAbort.abort()
reject(new Boom('Pre-key upload timeout', { statusCode: 408 }))
}, UPLOAD_TIMEOUT)
setTimeout(() => reject(new Boom('Pre-key upload timeout', { statusCode: 408 })), UPLOAD_TIMEOUT)
)
])
@@ -738,15 +784,14 @@ export const makeSocket = (config: SocketConfig) => {
try {
let count = 0
const preKeyCount = await getAvailablePreKeysOnServer()
// How many to upload: top-up to INITIAL_PREKEY_COUNT from whatever remains on server
count = Math.max(0, INITIAL_PREKEY_COUNT - preKeyCount)
if (preKeyCount === 0) count = INITIAL_PREKEY_COUNT
else count = MIN_PREKEY_COUNT
const { exists: currentPreKeyExists, currentPreKeyId } = await verifyCurrentPreKeyExists()
logger.info(`${preKeyCount} pre-keys found on server`)
logger.info(`Current prekey ID: ${currentPreKeyId}, exists in storage: ${currentPreKeyExists}`)
// Trigger upload when below the replenishment threshold, not when count < topUp amount
const lowServerCount = preKeyCount < MIN_PREKEY_COUNT
const lowServerCount = preKeyCount <= count
const missingCurrentPreKey = !currentPreKeyExists && currentPreKeyId > 0
const shouldUpload = lowServerCount || missingCurrentPreKey
@@ -1106,7 +1151,7 @@ export const makeSocket = (config: SocketConfig) => {
// Decrement active connections
decrementActiveConnections()
clearTimeout(keepAliveReq)
clearInterval(keepAliveReq)
clearTimeout(qrTimer)
// Clear offline-buffer safety timer so its callback cannot call ev.flush()
@@ -1183,6 +1228,12 @@ export const makeSocket = (config: SocketConfig) => {
cleanupPreKeyAutoSync()
cleanupSessionTTL()
// CRITICAL: Destroy circuit breakers AFTER cleanup functions complete
// This ensures cleanup functions can still use circuit breakers if needed
queryCircuitBreaker?.destroy()
connectionCircuitBreaker?.destroy()
preKeyCircuitBreaker?.destroy()
// IMPORTANT: Do NOT use removeAllListeners('connection.update')
// It would remove consumer listeners, breaking their reconnection logic
}
@@ -1211,16 +1262,8 @@ export const makeSocket = (config: SocketConfig) => {
})
}
const startKeepAliveRequest = () => {
// Use recursive setTimeout with ±15% jitter to match WA Desktop behaviour
// (WA Business Desktop: ~25-30s intervals with natural variance)
const scheduleNextKeepAlive = () => {
const jitter = keepAliveIntervalMs * 0.15
const delay = keepAliveIntervalMs + Math.floor((Math.random() * 2 - 1) * jitter)
keepAliveReq = setTimeout(onKeepAliveTick, delay)
}
const onKeepAliveTick = () => {
const startKeepAliveRequest = () =>
(keepAliveReq = setInterval(() => {
if (!lastDateRecv) {
lastDateRecv = new Date()
}
@@ -1232,15 +1275,13 @@ export const makeSocket = (config: SocketConfig) => {
*/
if (diff > keepAliveIntervalMs + 5000) {
void end(new Boom('Connection was lost', { statusCode: DisconnectReason.connectionLost }))
return // connection closing — do not reschedule
} else if (ws.isOpen) {
// Send keep-alive ping via sendNode() (fire-and-forget) instead of query().
// We don't need the response correlator overhead — WA's pong arrives as an
// incoming frame and updates lastDateRecv via the standard receive path.
// (Historical context: this used to bypass the query circuit breaker,
// which would block pings during cascade failures. Circuit breaker has
// since been replaced by per-operation bounded-retry — but the
// fire-and-forget pattern stays correct on its own merits.)
// query() wraps the ping in the query circuit breaker — when that breaker is
// open or timing out, the ping is never sent, WA never responds, lastDateRecv
// goes stale, and the diff check above wrongly fires "Connection was lost".
// sendNode() bypasses the query circuit breaker entirely; WA's ping response
// still arrives as an incoming frame and updates lastDateRecv normally.
sendNode({
tag: 'iq',
attrs: {
@@ -1256,15 +1297,7 @@ export const makeSocket = (config: SocketConfig) => {
} else {
logger.warn('keep alive called when WS not open')
}
// Do not reschedule once shutdown has started (closed set by end() on any concurrent path)
if (!closed) {
scheduleNextKeepAlive()
}
}
scheduleNextKeepAlive()
}
}, keepAliveIntervalMs))
/** i have no idea why this exists. pls enlighten me */
const sendPassiveIq = (tag: 'passive' | 'active') =>
query({
@@ -1756,6 +1789,25 @@ export const makeSocket = (config: SocketConfig) => {
sendWAMBuffer,
executeUSyncQuery,
onWhatsApp,
// Circuit breaker utilities
circuitBreakers: {
query: queryCircuitBreaker,
connection: connectionCircuitBreaker,
preKey: preKeyCircuitBreaker
},
/** Get circuit breaker statistics */
getCircuitBreakerStats: () => ({
query: queryCircuitBreaker?.getStats(),
connection: connectionCircuitBreaker?.getStats(),
preKey: preKeyCircuitBreaker?.getStats()
}),
/** Reset all circuit breakers to closed state */
resetCircuitBreakers: () => {
queryCircuitBreaker?.reset()
connectionCircuitBreaker?.reset()
preKeyCircuitBreaker?.reset()
logger.info('All circuit breakers reset to closed state')
},
// Unified Session Telemetry
/** Send unified_session telemetry manually */
sendUnifiedSession,
-2
View File
@@ -9,8 +9,6 @@ export interface Contact {
name?: string
/** name of the contact, the contact has set on their own on WA */
notify?: string
/** username associated with this contact, when provided by WA */
username?: string
/** I have no idea */
verifiedName?: string
// Baileys Added
-6
View File
@@ -27,8 +27,6 @@ export type BaileysEventMap = {
chats: Chat[]
contacts: Contact[]
messages: WAMessage[]
/** Past participants for group chats (people who left/were removed). userJid is always a phone number (PN), never a LID. */
pastParticipants?: proto.IPastParticipants[] | null
isLatest?: boolean
progress?: number | null
syncType?: proto.HistorySync.HistorySyncType | null
@@ -81,7 +79,6 @@ export type BaileysEventMap = {
id: string
author: string
authorPn?: string
authorUsername?: string
participants: GroupParticipant[]
action: ParticipantAction
}
@@ -89,7 +86,6 @@ export type BaileysEventMap = {
id: string
author: string
authorPn?: string
authorUsername?: string
participant: string
participantPn?: string
action: RequestJoinAction
@@ -189,8 +185,6 @@ export type BufferedEventData = {
chats: { [jid: string]: Chat }
contacts: { [jid: string]: Contact }
messages: { [uqId: string]: WAMessage }
/** Keyed by groupJid for O(1) deduplication across chunks */
pastParticipants: { [groupJid: string]: proto.IPastParticipant[] }
empty: boolean
isLatest: boolean
progress?: number | null
-4
View File
@@ -20,20 +20,17 @@ export interface GroupMetadata {
addressingMode?: WAMessageAddressingMode
owner: string | undefined
ownerPn?: string | undefined
ownerUsername?: string | undefined
owner_country_code?: string | undefined
subject: string
/** group subject owner */
subjectOwner?: string
subjectOwnerPn?: string
subjectOwnerUsername?: string
/** group subject modification date */
subjectTime?: number
creation?: number
desc?: string
descOwner?: string
descOwnerPn?: string
descOwnerUsername?: string
descId?: string
descTime?: number
/** if this group is part of a community, it returns the jid of the community to which it belongs */
@@ -59,7 +56,6 @@ export interface GroupMetadata {
/** the person who added you to group or changed some setting in group */
author?: string
authorPn?: string
authorUsername?: string
}
export interface WAGroupCreateResponse {
+2 -15
View File
@@ -19,9 +19,7 @@ export type WAContactMessage = proto.Message.IContactMessage
export type WAContactsArrayMessage = proto.Message.IContactsArrayMessage
export type WAMessageKey = proto.IMessageKey & {
remoteJidAlt?: string
remoteJidUsername?: string
participantAlt?: string
participantUsername?: string
server_id?: string
addressingMode?: string
isViewOnce?: boolean // TODO: remove out of the message key, place in WebMessageInfo
@@ -157,7 +155,6 @@ export type MessageReceiptType =
| 'sender'
| 'inactive'
| 'played'
| 'view_once_read'
| undefined
export type MediaConnInfo = {
@@ -181,8 +178,6 @@ export interface WAUrlInfo {
type Mentionable = {
/** list of jids that are mentioned in the accompanying text */
mentions?: string[]
/** mention all participants in the group */
mentionAll?: boolean
}
type Contextable = {
/** add contextInfo to the message */
@@ -995,16 +990,8 @@ export type MessageGenerationOptionsFromContent = MiscMessageGenerationOptions &
export type WAMediaUploadFunction = (
encFilePath: string,
opts: { fileEncSha256B64: string; mediaType: MediaType; timeoutMs?: number; newsletter?: boolean }
) => Promise<{
mediaUrl: string | undefined
directPath: string
meta_hmac?: string
ts?: number
fbid?: number
thumbnailDirectPath?: string
thumbnailSha256?: string
}>
opts: { fileEncSha256B64: string; mediaType: MediaType; timeoutMs?: number }
) => Promise<{ mediaUrl: string; directPath: string; meta_hmac?: string; ts?: number; fbid?: number }>
export type MediaGenerationOptions = {
logger?: ILogger
+8 -8
View File
@@ -4,10 +4,10 @@ export enum XWAPaths {
xwa2_newsletter_view = 'xwa2_newsletter_view',
xwa2_newsletter_metadata = 'xwa2_newsletter',
xwa2_newsletter_admin_count = 'xwa2_newsletter_admin',
xwa2_newsletter_mute_v2 = 'xwa2_newsletter_update_user_setting',
xwa2_newsletter_unmute_v2 = 'xwa2_newsletter_update_user_setting',
xwa2_newsletter_follow = 'xwa2_newsletter_join_v2',
xwa2_newsletter_unfollow = 'xwa2_newsletter_leave_v2',
xwa2_newsletter_mute_v2 = 'xwa2_newsletter_mute_v2',
xwa2_newsletter_unmute_v2 = 'xwa2_newsletter_unmute_v2',
xwa2_newsletter_follow = 'xwa2_newsletter_follow',
xwa2_newsletter_unfollow = 'xwa2_newsletter_unfollow',
xwa2_newsletter_change_owner = 'xwa2_newsletter_change_owner',
xwa2_newsletter_demote = 'xwa2_newsletter_demote',
xwa2_newsletter_delete_v2 = 'xwa2_newsletter_delete_v2'
@@ -17,10 +17,10 @@ export enum QueryIds {
UPDATE_METADATA = '24250201037901610',
METADATA = '6563316087068696',
SUBSCRIBERS = '9783111038412085',
FOLLOW = '24404358912487870',
UNFOLLOW = '9767147403369991',
MUTE = '31938993655691868',
UNMUTE = '31938993655691868',
FOLLOW = '7871414976211147',
UNFOLLOW = '7238632346214362',
MUTE = '29766401636284406',
UNMUTE = '9864994326891137',
ADMIN_COUNT = '7130823597031706',
CHANGE_OWNER = '7341777602580933',
DEMOTE = '6551828931592903',
+18
View File
@@ -1,6 +1,7 @@
import type { Agent } from 'https'
import type { URL } from 'url'
import { proto } from '../../WAProto/index.js'
import type { CircuitBreakerOptions } from '../Utils/circuit-breaker'
import type { ILogger } from '../Utils/logger'
import type { AuthenticationState, LIDMapping, SignalAuthState, TransactionCapabilityOptions } from './Auth'
import type { GroupMetadata } from './GroupMetadata'
@@ -202,6 +203,23 @@ export type SocketConfig = {
pnToLIDFunc?: (jids: string[]) => Promise<LIDMapping[] | undefined>
) => SignalRepositoryWithLIDStore
// === Circuit Breaker Configuration ===
/** Enable circuit breaker protection for socket operations (default: true) */
enableCircuitBreaker?: boolean
/** Circuit breaker configuration for query operations */
queryCircuitBreaker?: Partial<CircuitBreakerOptions>
/** Circuit breaker configuration for connection operations */
connectionCircuitBreaker?: Partial<CircuitBreakerOptions>
/** Circuit breaker configuration for pre-key operations */
preKeyCircuitBreaker?: Partial<CircuitBreakerOptions>
/** Circuit breaker configuration for message operations */
messageCircuitBreaker?: Partial<CircuitBreakerOptions>
// === Listener Limits (Memory Leak Prevention) ===
/**
-11
View File
@@ -341,18 +341,7 @@ export const addTransactionCapability = (
return result
} catch (error) {
// SessionError is part of the normal Bad MAC recovery flow
// (retry receipt → sender resends as pkmsg → new session within ~1.3s).
// Logging it as ERROR creates 2 noise lines per recoverable Bad MAC cycle.
// Downgrade to debug for SessionError; keep ERROR for everything else.
// The error is still re-thrown — recovery behavior is unchanged.
const errName = (error as { name?: string })?.name
if (errName === 'SessionError') {
logger.debug({ error }, 'transaction failed (SessionError — recoverable via retry receipt)')
} else {
logger.error({ error }, 'transaction failed, rolling back')
}
throw error
}
})
-393
View File
@@ -1,393 +0,0 @@
/**
* Bounded retry WhatsApp-aligned per-operation retry without global state.
*
* # Why this exists
*
* Replaces the circuit breaker pattern that was causing cascading failures
* in production: 5 timeouts in 60s would trip the global breaker, blocking
* EVERY socket query (typing indicators, profile pic fetches, contact
* validation, presence updates) for 30s even queries to peers that were
* perfectly healthy.
*
* # Empirical justification
*
* Captured WhatsApp Android's actual retry behavior via Frida hooks
* (`hook-circuit-breaker-re-v2.js`, `hook-retry-bounds.js`) on a real
* device under controlled WiFi off/on cycles. Findings:
*
* 1. Delay sequence (per-operation): 3s -> 10s -> 60s -> ~64s -> 120s
* (cap at 2 min). Last value reused for further attempts.
* 2. Memory profile during 5-min network outage: PSS dropped 53MB and
* stabilised. FDs closed (305 -> 295). NO unbounded accumulation.
* 3. Recovery on reconnect: ~10s. No retry storm.
* 4. NO global state machine observed. Each operation has its own timer.
* Failures of operation A do NOT block operation B.
*
* The default delays in WHATSAPP_BACKOFF_DELAYS below match the captured
* sequence directly. The default 10-min TTL is empirical 3-5 min stability
* + safety buffer.
*
* # Design properties
*
* - Per-operation isolation: each call to `withBoundedRetry` has its own
* timer. No shared state. Operation A failing has zero effect on B.
* - Bounded by `ttlMs` (wall-clock budget). The TTL is enforced strictly
* at three points: (a) before each attempt, (b) cap on per-attempt
* timeout, (c) cap on retry delay.
* - Per-attempt timeout: each attempt has its own deadline (default 30s),
* automatically capped to remaining TTL budget so total runtime never
* exceeds ttlMs.
* - AbortSignal cancellation: external cancellation supported, both for
* the sleep between retries AND (optionally) the in-flight operation
* when the operation accepts a signal parameter.
* - Memory bound: at most one outstanding retry timer per call. Listeners
* are explicitly removed when timers settle. Once the call resolves /
* rejects / aborts, all state is freed.
*
* # Logging
*
* Pass `logger` in options to get structured logs at each retry attempt,
* give-up, and post-failure recovery. The module emits Prometheus metrics
* regardless of whether a logger is provided.
*
* # When to use this vs. plain query()
*
* - Use plain `query()` when you want fast-fail semantics (caller decides
* what to do on failure). Most call sites in InfiniteAPI use this.
* - Use `withBoundedRetry(() => query(...), { name: 'X' })` when the
* operation is "must-eventually-succeed" with no upstream retry e.g.
* `uploadPreKeys` or other write paths where the alternative is data loss.
*
* # Examples
*
* ```ts
* // Single attempt with 5-min TTL, give up after that:
* await withBoundedRetry(
* () => assertSessions([jid], true),
* { name: 'assertSessions', ttlMs: 5 * 60_000, logger }
* )
*
* // Tight deadline, fast give-up:
* await withBoundedRetry(
* () => sendNode(node),
* { name: 'send', ttlMs: 30_000, perAttemptTimeoutMs: 5_000 }
* )
*
* // Operation that respects abort signal (best practice):
* await withBoundedRetry(
* (signal) => fetchSomething({ signal }),
* { name: 'fetch', logger }
* )
* ```
*
* @module Utils/bounded-retry
*/
import type { ILogger } from './logger'
import { metrics } from './prometheus-metrics.js'
/**
* Default delay sequence (milliseconds) matches WhatsApp Android empirical
* behavior. After exhausting the sequence, the last value is used (cap).
*/
export const WHATSAPP_BACKOFF_DELAYS = [3000, 10000, 60000, 60000, 120000] as const
/**
* Default jitter (+/- 15%) to prevent thundering-herd retries.
*/
export const DEFAULT_JITTER_FACTOR = 0.15 as const
/**
* Default time-to-live for retries: 10 minutes.
*
* Empirical justification: WhatsApp Android stabilises memory in ~3-5 min
* during a network outage; 10 min gives a generous safety buffer while
* preventing unbounded retry accumulation.
*/
export const DEFAULT_TTL_MS = 10 * 60 * 1000
/**
* Default per-attempt timeout: 30 seconds.
*
* Most WhatsApp queries respond within seconds. A 30s timeout is generous
* enough for slow networks but prevents a single hang from blocking retries.
*/
export const DEFAULT_PER_ATTEMPT_TIMEOUT_MS = 30000
export interface BoundedRetryOptions {
/** Operation name for logging/metrics */
name?: string
/** Sequence of delays (ms). Last value is used as cap. */
delays?: readonly number[]
/** Jitter factor 0..1 (default 0.15) */
jitter?: number
/** Total wall-clock budget — gives up after this. Default 10 min. */
ttlMs?: number
/** Per-attempt timeout (ms). Default 30s. Capped by remaining TTL. */
perAttemptTimeoutMs?: number
/** Predicate: should we retry on this error? Default: always */
shouldRetry?: (err: Error, attempt: number) => boolean
/** Hook fired before each retry */
onRetry?: (err: Error, attempt: number, delayMs: number) => void
/**
* AbortSignal when fired, the loop stops at the next observation point:
* - If the loop is sleeping between retries, the sleep rejects immediately.
* - If an attempt is in flight, the abort is forwarded to the operation
* via the per-attempt signal. The operation must ITSELF observe the
* signal (e.g. `(signal) => fetch({ signal })`) for cancellation to
* be truly immediate; otherwise it cancels at the next per-attempt
* timeout boundary.
* - Either way, on the next loop iteration BoundedRetryAbortedError is
* thrown, so the caller never sees more than one trailing attempt
* after the abort.
*/
signal?: AbortSignal
/** Optional logger for structured retry/give-up/recovery logs */
logger?: ILogger
}
export class BoundedRetryGiveUpError extends Error {
constructor(
public readonly opName: string,
public readonly attempts: number,
public readonly elapsedMs: number,
public readonly lastError: Error
) {
super(
`bounded-retry "${opName}" gave up after ${attempts} attempts ` +
`(${elapsedMs}ms elapsed). Last error: ${lastError.message}`
)
this.name = 'BoundedRetryGiveUpError'
}
}
export class BoundedRetryAbortedError extends Error {
constructor(public readonly opName: string) {
super(`bounded-retry "${opName}" aborted via signal`)
this.name = 'BoundedRetryAbortedError'
}
}
/**
* Apply jitter to a delay: returns delay * (1 +/- jitter)
*/
function withJitter(delayMs: number, jitter: number): number {
if (jitter <= 0) return delayMs
const factor = 1 + (Math.random() * 2 - 1) * jitter
return Math.max(0, Math.round(delayMs * factor))
}
/**
* Pick the delay for a given attempt index. Falls back to the last value
* (cap) once the sequence is exhausted.
*/
function pickDelay(attempt: number, delays: readonly number[]): number {
if (delays.length === 0) return 0
if (attempt < delays.length) return delays[attempt]!
return delays[delays.length - 1]!
}
/**
* Wrap a promise with a per-attempt timeout. Aborts the supplied controller
* when the timeout fires so the caller can cancel any in-flight work
* (operations that accept the signal). Always clears the timer on settlement.
*/
function withTimeout<T>(
promise: Promise<T>,
timeoutMs: number,
name: string,
abortOnTimeout: AbortController
): Promise<T> {
return new Promise<T>((resolve, reject) => {
const timer = setTimeout(() => {
abortOnTimeout.abort()
reject(new Error(`bounded-retry "${name}" attempt timed out after ${timeoutMs}ms`))
}, timeoutMs)
promise
.then(value => {
clearTimeout(timer)
resolve(value)
})
.catch(err => {
clearTimeout(timer)
reject(err as Error)
})
})
}
/**
* Sleep with abort support. Always removes the abort listener on settlement
* so listeners do not accumulate on long-lived signals.
*/
function sleep(ms: number, signal?: AbortSignal): Promise<void> {
return new Promise((resolve, reject) => {
if (signal?.aborted) {
reject(new Error('aborted'))
return
}
let onAbort: (() => void) | undefined
const cleanup = () => {
if (onAbort && signal) {
signal.removeEventListener('abort', onAbort)
}
}
const timer = setTimeout(() => {
cleanup()
resolve()
}, ms)
if (signal) {
onAbort = () => {
clearTimeout(timer)
cleanup()
reject(new Error('aborted'))
}
signal.addEventListener('abort', onAbort, { once: true })
}
})
}
/**
* Run an async operation with bounded exponential-backoff retry.
*
* Independent per-call: no global state, no cross-operation interaction.
* Memory bound: at most one outstanding retry timer per call.
*
* The operation may optionally accept an `AbortSignal` parameter when the
* per-attempt timeout fires (or the outer signal is aborted), the inner
* signal is aborted so the operation can stop in-flight work cleanly.
*
* @example
* ```ts
* const result = await withBoundedRetry(
* (signal) => assertSessions([jid], true, { signal }),
* { name: 'assertSessions', ttlMs: 5 * 60_000, logger }
* )
* ```
*/
export async function withBoundedRetry<T>(
operation: (signal?: AbortSignal) => Promise<T>,
options: BoundedRetryOptions = {}
): Promise<T> {
const name = options.name ?? 'bounded-retry'
const delays = options.delays ?? WHATSAPP_BACKOFF_DELAYS
const jitter = options.jitter ?? DEFAULT_JITTER_FACTOR
const ttlMs = options.ttlMs ?? DEFAULT_TTL_MS
const perAttemptTimeoutMs = options.perAttemptTimeoutMs ?? DEFAULT_PER_ATTEMPT_TIMEOUT_MS
const shouldRetry = options.shouldRetry ?? (() => true)
const logger = options.logger
const start = Date.now()
let lastError: Error = new Error('unknown')
let attempt = 0
while (true) {
// Check abort + TTL BEFORE starting an attempt — strict wall-clock
// budget. Without this check the loop could begin a new attempt with
// 0ms remaining and then run for up to `perAttemptTimeoutMs`.
if (options.signal?.aborted) {
throw new BoundedRetryAbortedError(name)
}
const elapsedBeforeAttempt = Date.now() - start
const remainingBudget = ttlMs - elapsedBeforeAttempt
if (remainingBudget <= 0) {
metrics.errors?.inc({ category: 'bounded_retry', code: 'ttl_exceeded' })
logger?.warn?.(
{ op: name, attempts: attempt, elapsedMs: elapsedBeforeAttempt, ttlMs, lastError: lastError.message },
'bounded-retry: TTL exceeded before next attempt — giving up'
)
throw new BoundedRetryGiveUpError(name, attempt, elapsedBeforeAttempt, lastError)
}
// Cap the per-attempt timeout to the remaining TTL budget so a single
// attempt cannot run past the wall-clock deadline.
const attemptTimeoutMs = Math.min(perAttemptTimeoutMs, remainingBudget)
const attemptAbort = new AbortController()
// Forward outer abort to the per-attempt controller so the in-flight
// operation is cancelled when the user aborts. Cleaned up below.
let onOuterAbort: (() => void) | undefined
if (options.signal) {
onOuterAbort = () => attemptAbort.abort()
if (options.signal.aborted) {
attemptAbort.abort()
} else {
options.signal.addEventListener('abort', onOuterAbort, { once: true })
}
}
try {
const result = await withTimeout(operation(attemptAbort.signal), attemptTimeoutMs, name, attemptAbort)
if (attempt > 0) {
metrics.socketEvents?.inc({ event: 'bounded_retry_recovered' })
logger?.info?.(
{ op: name, attempts: attempt + 1, elapsedMs: Date.now() - start },
'bounded-retry: operation succeeded after retries'
)
}
return result
} catch (err) {
lastError = err as Error
attempt++
// (outer-abort listener detachment happens in the finally below —
// avoiding a double-remove that breaks listener-count assertions.)
// If the outer signal aborted us mid-attempt, surface that explicitly
// rather than as a generic operation failure.
if (options.signal?.aborted) {
throw new BoundedRetryAbortedError(name)
}
const elapsed = Date.now() - start
if (!shouldRetry(lastError, attempt)) {
metrics.errors?.inc({ category: 'bounded_retry', code: 'predicate_no_retry' })
logger?.warn?.(
{ op: name, attempts: attempt, elapsedMs: elapsed, error: lastError.message },
'bounded-retry: shouldRetry returned false — giving up'
)
throw lastError
}
// Re-check budget after the failure so we do not sleep past TTL.
const remainingAfterFailure = ttlMs - elapsed
if (remainingAfterFailure <= 0) {
metrics.errors?.inc({ category: 'bounded_retry', code: 'ttl_exceeded' })
logger?.warn?.(
{ op: name, attempts: attempt, elapsedMs: elapsed, ttlMs, lastError: lastError.message },
'bounded-retry: TTL exceeded after attempt — giving up'
)
throw new BoundedRetryGiveUpError(name, attempt, elapsed, lastError)
}
const baseDelay = pickDelay(attempt - 1, delays)
const delayMs = Math.min(withJitter(baseDelay, jitter), remainingAfterFailure)
options.onRetry?.(lastError, attempt, delayMs)
metrics.socketEvents?.inc({ event: 'bounded_retry_attempt' })
logger?.debug?.(
{ op: name, attempt, delayMs, elapsedMs: elapsed, error: lastError.message },
'bounded-retry: scheduling next attempt'
)
try {
await sleep(delayMs, options.signal)
} catch {
throw new BoundedRetryAbortedError(name)
}
} finally {
// Belt-and-suspenders: ensure the outer-abort listener is always
// removed even on early throws.
if (options.signal && onOuterAbort) {
options.signal.removeEventListener('abort', onOuterAbort)
}
}
}
}
-1
View File
@@ -1083,7 +1083,6 @@ export const processSyncAction = (
action.lidContactAction.firstName ||
action.lidContactAction.username ||
undefined,
username: action.lidContactAction.username || undefined,
lid: id!,
phoneNumber: undefined
}
+766
View File
@@ -0,0 +1,766 @@
/**
* Circuit Breaker Pattern Implementation
*
* Provides protection against cascading failures by monitoring operation outcomes
* and temporarily blocking requests when failure thresholds are exceeded.
*
* States:
* - CLOSED: Normal operation, requests pass through
* - OPEN: Failures exceeded threshold, requests are blocked
* - HALF_OPEN: Testing recovery, limited requests allowed
*
* @module Utils/circuit-breaker
*/
import { EventEmitter } from 'events'
import { metrics } from './prometheus-metrics.js'
/**
* Circuit breaker operational states
*/
export type CircuitState = 'closed' | 'open' | 'half-open'
/**
* Failure record with timestamp for sliding window tracking
*/
export interface FailureRecord {
timestamp: number
error: Error
}
/**
* Circuit breaker configuration options
*/
export interface CircuitBreakerOptions {
/** Unique identifier for this circuit breaker (used in metrics/logging) */
name: string
/** Number of failures within the window to trigger OPEN state (default: 5) */
failureThreshold?: number
/** Time window in ms for counting failures (default: 60000) */
failureWindow?: number
/** Number of successes required in HALF_OPEN to return to CLOSED (default: 2) */
successThreshold?: number
/** Time in ms to wait before transitioning from OPEN to HALF_OPEN (default: 30000) */
resetTimeout?: number
/** Timeout for individual operations in ms (default: 10000) */
timeout?: number
/** Minimum number of requests before circuit can trip (default: 5) */
volumeThreshold?: number
/** Predicate to determine if an error should count as a failure */
shouldCountError?: (error: Error) => boolean
/** Whether to collect Prometheus metrics (default: true) */
collectMetrics?: boolean
/** Fallback function when circuit is OPEN */
fallback?: <T>() => T | Promise<T>
/** Callback when state changes */
onStateChange?: (from: CircuitState, to: CircuitState) => void
/** Callback on failure */
onFailure?: (error: Error) => void
/** Callback on success */
onSuccess?: () => void
/** Callback when circuit opens */
onOpen?: () => void
/** Callback when circuit closes */
onClose?: () => void
/** Callback when circuit enters half-open */
onHalfOpen?: () => void
}
/**
* Circuit breaker statistics
*/
export interface CircuitBreakerStats {
state: CircuitState
failures: number
successes: number
consecutiveFailures: number
consecutiveSuccesses: number
totalCalls: number
totalFailures: number
totalSuccesses: number
totalRejected: number
failureRate: number
lastFailureTime?: number
lastSuccessTime?: number
lastStateChange?: number
isOpen: boolean
isClosed: boolean
isHalfOpen: boolean
}
/**
* Error thrown when circuit is OPEN and request is rejected
*/
export class CircuitOpenError extends Error {
constructor(
public readonly circuitName: string,
public readonly state: CircuitState
) {
super(`Circuit breaker "${circuitName}" is ${state}`)
this.name = 'CircuitOpenError'
}
}
/**
* Error thrown when operation exceeds timeout
*/
export class CircuitTimeoutError extends Error {
constructor(
public readonly circuitName: string,
public readonly timeoutMs: number
) {
super(`Circuit breaker "${circuitName}" operation timed out after ${timeoutMs}ms`)
this.name = 'CircuitTimeoutError'
}
}
/**
* Circuit Breaker implementation with sliding window failure tracking
*
* @example
* ```typescript
* const breaker = new CircuitBreaker({
* name: 'whatsapp-api',
* failureThreshold: 5,
* resetTimeout: 30000
* })
*
* try {
* const result = await breaker.execute(() => sendMessage(msg))
* } catch (error) {
* if (error instanceof CircuitOpenError) {
* // Circuit is open, use fallback
* }
* }
* ```
*/
export class CircuitBreaker extends EventEmitter {
private state: CircuitState = 'closed'
private failureRecords: FailureRecord[] = []
private consecutiveFailures = 0
private consecutiveSuccesses = 0
private totalCalls = 0
private totalFailures = 0
private totalSuccesses = 0
private totalRejected = 0
private lastFailureTime?: number
private lastSuccessTime?: number
private lastStateChange: number
private resetTimer?: ReturnType<typeof setTimeout>
private readonly options: Required<CircuitBreakerOptions>
constructor(options: CircuitBreakerOptions) {
super()
this.options = {
name: options.name,
failureThreshold: options.failureThreshold ?? 5,
failureWindow: options.failureWindow ?? 60000,
successThreshold: options.successThreshold ?? 2,
resetTimeout: options.resetTimeout ?? 30000,
timeout: options.timeout ?? 10000,
volumeThreshold: options.volumeThreshold ?? 5,
shouldCountError: options.shouldCountError ?? (() => true),
collectMetrics: options.collectMetrics ?? true,
fallback:
options.fallback ??
(() => {
throw new CircuitOpenError(this.options.name, this.state)
}),
onStateChange: options.onStateChange ?? (() => {}),
onFailure: options.onFailure ?? (() => {}),
onSuccess: options.onSuccess ?? (() => {}),
onOpen: options.onOpen ?? (() => {}),
onClose: options.onClose ?? (() => {}),
onHalfOpen: options.onHalfOpen ?? (() => {})
}
this.lastStateChange = Date.now()
}
/**
* Check if the circuit allows execution
*/
canExecute(): boolean {
if (this.state === 'closed') {
return true
}
if (this.state === 'open') {
return false
}
// HALF_OPEN: allow limited requests for testing
return true
}
/**
* Execute an async operation with circuit breaker protection
*/
async execute<T>(operation: () => T | Promise<T>): Promise<T> {
this.totalCalls++
// Check if circuit allows execution
if (this.state === 'open') {
this.totalRejected++
if (this.options.collectMetrics) {
metrics.errors.inc({ category: 'circuit_breaker', code: 'rejected' })
}
return this.options.fallback() as T
}
// Execute with timeout protection
try {
const result = await this.executeWithTimeout(operation)
this.recordSuccess()
return result
} catch (error) {
this.recordFailure(error as Error)
throw error
}
}
/**
* Execute a synchronous operation with circuit breaker protection
*/
executeSync<T>(operation: () => T): T {
this.totalCalls++
if (this.state === 'open') {
this.totalRejected++
if (this.options.collectMetrics) {
metrics.errors.inc({ category: 'circuit_breaker', code: 'rejected' })
}
return this.options.fallback() as T
}
try {
const result = operation()
this.recordSuccess()
return result
} catch (error) {
this.recordFailure(error as Error)
throw error
}
}
/**
* Execute operation with timeout wrapper
*/
private async executeWithTimeout<T>(operation: () => T | Promise<T>): Promise<T> {
return new Promise<T>((resolve, reject) => {
const timer = setTimeout(() => {
reject(new CircuitTimeoutError(this.options.name, this.options.timeout))
}, this.options.timeout)
Promise.resolve(operation())
.then(result => {
clearTimeout(timer)
resolve(result)
})
.catch(error => {
clearTimeout(timer)
reject(error)
})
})
}
/**
* Record a successful operation
*/
private recordSuccess(): void {
this.totalSuccesses++
this.lastSuccessTime = Date.now()
this.consecutiveSuccesses++
this.consecutiveFailures = 0
if (this.options.collectMetrics) {
metrics.socketEvents.inc({ event: 'circuit_success' })
}
this.options.onSuccess()
this.emit('success')
// In HALF_OPEN state, check if we can close the circuit
if (this.state === 'half-open') {
if (this.consecutiveSuccesses >= this.options.successThreshold) {
this.transitionTo('closed')
}
}
}
/**
* Record a failed operation
*/
private recordFailure(error: Error): void {
// Check if this error should count as a failure
if (!this.options.shouldCountError(error)) {
return
}
const now = Date.now()
this.totalFailures++
this.lastFailureTime = now
this.consecutiveFailures++
this.consecutiveSuccesses = 0
// Add to failure records for sliding window
this.failureRecords.push({ timestamp: now, error })
// Clean old failures outside the window
this.cleanOldFailures()
if (this.options.collectMetrics) {
metrics.errors.inc({ category: 'circuit_breaker', code: 'failure' })
}
this.options.onFailure(error)
this.emit('failure', error)
// State transition logic
if (this.state === 'half-open') {
// Any failure in HALF_OPEN immediately reopens the circuit
this.transitionTo('open')
} else if (this.state === 'closed') {
// Check if we should trip the circuit
const recentFailures = this.failureRecords.length
if (this.totalCalls >= this.options.volumeThreshold && recentFailures >= this.options.failureThreshold) {
this.transitionTo('open')
}
}
}
/**
* Remove failure records outside the sliding window
*/
private cleanOldFailures(): void {
const cutoff = Date.now() - this.options.failureWindow
this.failureRecords = this.failureRecords.filter(record => record.timestamp > cutoff)
}
/**
* Transition to a new state
*/
private transitionTo(newState: CircuitState): void {
const oldState = this.state
if (oldState === newState) {
return
}
this.state = newState
this.lastStateChange = Date.now()
// Clear existing reset timer
if (this.resetTimer) {
clearTimeout(this.resetTimer)
this.resetTimer = undefined
}
// State-specific actions
switch (newState) {
case 'closed':
this.consecutiveFailures = 0
this.consecutiveSuccesses = 0
this.failureRecords = []
this.options.onClose()
this.emit('close')
break
case 'open':
this.consecutiveSuccesses = 0
this.options.onOpen()
this.emit('open')
// Record circuit breaker trip metric
if (this.options.collectMetrics) {
metrics.circuitBreakerTrips?.inc({ name: this.options.name })
}
// Schedule transition to HALF_OPEN
this.resetTimer = setTimeout(() => {
this.transitionTo('half-open')
}, this.options.resetTimeout)
break
case 'half-open':
this.consecutiveSuccesses = 0
this.consecutiveFailures = 0
this.options.onHalfOpen()
this.emit('half-open')
break
}
this.options.onStateChange(oldState, newState)
this.emit('state-change', { from: oldState, to: newState })
}
/**
* Manually trip the circuit to OPEN state
*/
trip(): void {
this.transitionTo('open')
}
/**
* Manually reset the circuit to CLOSED state
*/
reset(): void {
this.transitionTo('closed')
}
/**
* Get current circuit state
*/
getState(): CircuitState {
return this.state
}
/**
* Check if circuit is OPEN
*/
isOpen(): boolean {
return this.state === 'open'
}
/**
* Check if circuit is CLOSED
*/
isClosed(): boolean {
return this.state === 'closed'
}
/**
* Check if circuit is HALF_OPEN
*/
isHalfOpen(): boolean {
return this.state === 'half-open'
}
/**
* Get circuit breaker statistics
*/
getStats(): CircuitBreakerStats {
this.cleanOldFailures()
const failureRate = this.totalCalls > 0 ? (this.totalFailures / this.totalCalls) * 100 : 0
return {
state: this.state,
failures: this.failureRecords.length,
successes: this.consecutiveSuccesses,
consecutiveFailures: this.consecutiveFailures,
consecutiveSuccesses: this.consecutiveSuccesses,
totalCalls: this.totalCalls,
totalFailures: this.totalFailures,
totalSuccesses: this.totalSuccesses,
totalRejected: this.totalRejected,
failureRate,
lastFailureTime: this.lastFailureTime,
lastSuccessTime: this.lastSuccessTime,
lastStateChange: this.lastStateChange,
isOpen: this.isOpen(),
isClosed: this.isClosed(),
isHalfOpen: this.isHalfOpen()
}
}
/**
* Get circuit breaker name
*/
getName(): string {
return this.options.name
}
/**
* Destroy circuit breaker and clean up resources
*/
destroy(): void {
if (this.resetTimer) {
clearTimeout(this.resetTimer)
}
this.failureRecords = []
this.removeAllListeners()
}
}
/**
* Factory function to create a circuit breaker
*/
export function createCircuitBreaker(options: CircuitBreakerOptions): CircuitBreaker {
return new CircuitBreaker(options)
}
/**
* Registry for managing multiple circuit breakers
*/
export class CircuitBreakerRegistry {
private breakers: Map<string, CircuitBreaker> = new Map()
/**
* Get or create a circuit breaker by name
*/
get(name: string, options?: Omit<CircuitBreakerOptions, 'name'>): CircuitBreaker {
if (!this.breakers.has(name)) {
const breaker = new CircuitBreaker({ ...options, name })
this.breakers.set(name, breaker)
}
return this.breakers.get(name)!
}
/**
* Check if a circuit breaker exists
*/
has(name: string): boolean {
return this.breakers.has(name)
}
/**
* Remove a circuit breaker
*/
remove(name: string): boolean {
const breaker = this.breakers.get(name)
if (breaker) {
breaker.destroy()
return this.breakers.delete(name)
}
return false
}
/**
* Get all circuit breakers
*/
getAll(): Map<string, CircuitBreaker> {
return new Map(this.breakers)
}
/**
* Get statistics for all circuit breakers
*/
getAllStats(): Record<string, CircuitBreakerStats> {
const stats: Record<string, CircuitBreakerStats> = {}
for (const [name, breaker] of this.breakers) {
stats[name] = breaker.getStats()
}
return stats
}
/**
* Reset all circuit breakers to CLOSED state
*/
resetAll(): void {
for (const breaker of this.breakers.values()) {
breaker.reset()
}
}
/**
* Destroy all circuit breakers
*/
destroyAll(): void {
for (const breaker of this.breakers.values()) {
breaker.destroy()
}
this.breakers.clear()
}
}
/**
* Global circuit breaker registry instance
*/
export const globalCircuitRegistry = new CircuitBreakerRegistry()
/**
* Decorator to protect a method with circuit breaker
*
* @example
* ```typescript
* class MyService {
* @circuitBreaker({ failureThreshold: 3 })
* async fetchData() {
* return await api.getData()
* }
* }
* ```
*/
export function circuitBreaker(options: Omit<CircuitBreakerOptions, 'name'> & { name?: string } = {}) {
return function (
_target: unknown,
propertyKey: string,
descriptor: TypedPropertyDescriptor<(...args: unknown[]) => unknown>
) {
const originalMethod = descriptor.value
if (!originalMethod) return descriptor
const name = options.name || propertyKey
const breaker = globalCircuitRegistry.get(name, options)
descriptor.value = async function (...args: unknown[]): Promise<unknown> {
return breaker.execute(() => originalMethod.apply(this, args))
}
return descriptor
}
}
/**
* Wrap a function with circuit breaker protection
*
* @example
* ```typescript
* const protectedFetch = withCircuitBreaker(
* fetchData,
* { name: 'api-fetch', failureThreshold: 5 }
* )
* ```
*/
// eslint-disable-next-line space-before-function-paren
export function withCircuitBreaker<T extends (...args: unknown[]) => unknown>(
fn: T,
options: CircuitBreakerOptions
): T {
const breaker = new CircuitBreaker(options)
return (async (...args: Parameters<T>): Promise<ReturnType<T>> => {
return breaker.execute(() => fn(...args)) as Promise<ReturnType<T>>
}) as unknown as T
}
/**
* Get health status of all circuit breakers
*/
export function getCircuitHealth(): {
healthy: boolean
openCircuits: string[]
stats: Record<string, CircuitBreakerStats>
} {
const stats = globalCircuitRegistry.getAllStats()
const openCircuits: string[] = []
for (const [name, stat] of Object.entries(stats)) {
if (stat.isOpen) {
openCircuits.push(name)
}
}
return {
healthy: openCircuits.length === 0,
openCircuits,
stats
}
}
/**
* Create a pre-configured circuit breaker for WhatsApp PreKey operations
*
* This circuit breaker is optimized for handling encryption/session errors
* that commonly occur with WhatsApp's Signal protocol implementation.
*
* @example
* ```typescript
* const preKeyBreaker = createPreKeyCircuitBreaker()
*
* async function sendEncryptedMessage(msg) {
* return preKeyBreaker.execute(async () => {
* return await encryptAndSend(msg)
* })
* }
* ```
*/
export function createPreKeyCircuitBreaker(customOptions?: Partial<CircuitBreakerOptions>): CircuitBreaker {
const preKeyErrorPatterns = ['prekey', 'pre-key', 'session', 'signal', 'encrypt', 'decrypt', 'cipher', 'key']
return new CircuitBreaker({
name: 'prekey-operations',
failureThreshold: 5,
failureWindow: 60000,
resetTimeout: 30000,
successThreshold: 2,
shouldCountError: (error: Error) => {
const message = error.message.toLowerCase()
return preKeyErrorPatterns.some(pattern => message.includes(pattern))
},
...customOptions
})
}
/**
* Create a circuit breaker for WebSocket connection operations
*/
export function createConnectionCircuitBreaker(customOptions?: Partial<CircuitBreakerOptions>): CircuitBreaker {
const connectionErrorPatterns = [
'econnrefused',
'econnreset',
'etimedout',
'enotfound',
'socket',
'websocket',
'connection',
'network'
]
// Normal WS lifecycle status codes that must NEVER trip the query circuit breaker.
// These are transient events that happen on every reconnect and carry no information
// about persistent server-side failures. The reconnection logic in makeSocket handles
// them independently.
const WS_LIFECYCLE_STATUS_CODES = new Set([
428, // connectionClosed
408, // connectionLost / timedOut
440, // connectionReplaced
])
return new CircuitBreaker({
name: 'connection-operations',
failureThreshold: 3,
failureWindow: 30000,
resetTimeout: 60000,
successThreshold: 1,
shouldCountError: (error: Error) => {
// CircuitTimeoutError messages embed the circuit name (e.g. "socket-query"), which
// accidentally matches the "socket" pattern below and causes a self-reinforcing loop
// where the breaker's own timeouts keep tripping the breaker. Exclude them explicitly.
if (error instanceof CircuitTimeoutError) return false
// Exclude normal WS reconnect events (Connection Closed, Connection Lost, Timed Out, etc.).
// These are not server-side failures — they happen on every restart/redeploy and should
// never cause the circuit to open. If we counted them, the 3 concurrent parallel
// post-login queries (sendPassiveIq, uploadPreKeysToServerIfRequired, digestKeyBundle) could all
// fail simultaneously when the WS drops, instantly opening the circuit and blocking
// profile-picture fetches and message delivery for the next 30 s.
const statusCode = (error as { output?: { statusCode?: number } })?.output?.statusCode
if (statusCode !== undefined && WS_LIFECYCLE_STATUS_CODES.has(statusCode)) return false
const message = error.message.toLowerCase()
const code = (error as NodeJS.ErrnoException).code?.toLowerCase() || ''
return connectionErrorPatterns.some(pattern => message.includes(pattern) || code.includes(pattern))
},
...customOptions
})
}
/**
* Create a circuit breaker for message sending operations
*/
export function createMessageCircuitBreaker(customOptions?: Partial<CircuitBreakerOptions>): CircuitBreaker {
return new CircuitBreaker({
name: 'message-operations',
failureThreshold: 5,
failureWindow: 60000,
resetTimeout: 15000,
successThreshold: 2,
timeout: 30000,
...customOptions
})
}
export default CircuitBreaker
+2 -44
View File
@@ -58,11 +58,7 @@ export const BAD_MAC_ERROR_TEXT = 'Bad MAC'
export const DECRYPTION_RETRY_CONFIG = {
maxRetries: 3,
baseDelayMs: 100,
// 'No matching sessions found' is the libsignal error when decryptWithSessions exhausts
// all stored sessions for a JID. Same recovery flow (retry receipt → pkmsg → new session)
// — categorise it as session-record so the caller logs DEBUG on retry, ERROR only when
// retries are exhausted (instead of dumping the full stack as an unknown error).
sessionRecordErrors: ['No session record', 'SessionError: No session record', 'No matching sessions found'],
sessionRecordErrors: ['No session record', 'SessionError: No session record'],
corruptedSessionErrors: ['Bad MAC', 'MessageCounterError', MISSING_KEYS_ERROR_TEXT]
}
@@ -246,14 +242,10 @@ export function decodeMessageNode(stanza: BinaryNode, meId: string, meLid: strin
const key: WAMessageKey = {
remoteJid: chatId,
remoteJidAlt: !isJidGroup(chatId) ? addressingContext.senderAlt : undefined,
remoteJidUsername: !isJidGroup(chatId)
? stanza.attrs.peer_recipient_username || stanza.attrs.recipient_username
: undefined,
fromMe,
id: msgId,
participant,
participantAlt: isJidGroup(chatId) ? addressingContext.senderAlt : undefined,
participantUsername: stanza.attrs.participant ? stanza.attrs.participant_username : undefined,
addressingMode: addressingContext.addressingMode,
...(msgType === 'newsletter' && stanza.attrs.server_id ? { server_id: stanza.attrs.server_id } : {})
}
@@ -400,23 +392,6 @@ export const decryptMessageNode = (
} else {
fullMessage.message = msg
}
// Detect view-once media on stanza 1 received by linked device.
// viewOnceMessage wrapper is also used for interactive messages
// (interactiveMessage, listMessage, nativeFlowMessage) -- those do NOT have
// imageMessage.viewOnce / videoMessage.viewOnce / audioMessage.viewOnce = true.
// Only real view-once media carries viewOnce: true on the inner media message.
const viewOnceInner =
msg.viewOnceMessage?.message ||
msg.viewOnceMessageV2?.message ||
msg.viewOnceMessageV2Extension?.message
if (
viewOnceInner?.imageMessage?.viewOnce ||
viewOnceInner?.videoMessage?.viewOnce ||
viewOnceInner?.audioMessage?.viewOnce
) {
fullMessage.key.isViewOnce = true
}
} catch (err: any) {
// Check if this is a final failure after all retries exhausted
const isRetryExhausted = err instanceof RetryExhaustedError
@@ -425,26 +400,9 @@ export const decryptMessageNode = (
const isCorrupted = isCorruptedSessionError(originalError)
const isSessionRecord = isSessionRecordError(originalError)
// Slim error projection — keep name/message/type for diagnosis,
// drop `stack` which adds 4-5 lines of node_modules paths per log
// for known-recoverable libsignal errors.
//
// CRITICAL: only slim for KNOWN-RECOVERABLE categories (corrupted /
// session-record). The unknown-error branch keeps the full Error so
// protobuf/parsing/runtime bugs still emit a stack trace where it
// matters most. Catches Copilot/Codex P2 review on PR #391.
const slimErr = originalError
? {
name: (originalError as { name?: string }).name,
message: (originalError as { message?: string }).message,
type: (originalError as { type?: string }).type
}
: undefined
const isRecoverableCategory = isCorrupted || isSessionRecord
const errorContext = {
key: fullMessage.key,
err: isRecoverableCategory ? slimErr : originalError,
err: originalError,
messageType: tag === 'plaintext' ? 'plaintext' : attrs.type,
sender,
author,
-28
View File
@@ -1,5 +1,4 @@
import EventEmitter from 'events'
import { proto } from '../../WAProto/index.js'
import type {
BaileysEvent,
BaileysEventEmitter,
@@ -891,7 +890,6 @@ const makeBufferData = (): BufferedEventData => {
chats: {},
messages: {},
contacts: {},
pastParticipants: {},
isLatest: false,
empty: true
},
@@ -969,28 +967,6 @@ function append<E extends BufferableEvent>(
}
}
// Merge pastParticipants with deduplication by groupJid and by userJid within each group.
// Multiple HistorySync chunks can carry the same group -- we merge rather than concatenate
// to avoid duplicate entries in the final event delivered to the consumer.
for (const group of (eventData.pastParticipants ?? []) as proto.IPastParticipants[]) {
const groupJid = group.groupJid
if (!groupJid) continue
if (!data.historySets.pastParticipants[groupJid]) {
data.historySets.pastParticipants[groupJid] = []
}
const existing = data.historySets.pastParticipants[groupJid]
const seenJids = new Set(existing.map(p => p.userJid).filter(Boolean))
for (const participant of (group.pastParticipants ?? []) as proto.IPastParticipant[]) {
if (participant.userJid && !seenJids.has(participant.userJid)) {
existing.push(participant)
seenJids.add(participant.userJid)
}
}
}
data.historySets.empty = false
data.historySets.syncType = eventData.syncType
data.historySets.progress = eventData.progress
@@ -1316,10 +1292,6 @@ function consolidateEvents(data: BufferedEventData) {
chats: Object.values(data.historySets.chats),
messages: Object.values(data.historySets.messages),
contacts: Object.values(data.historySets.contacts),
// Convert dedup map back to array. Each entry has groupJid + participants (all unique userJids).
pastParticipants: Object.entries(data.historySets.pastParticipants).map(
([groupJid, participants]) => ({ groupJid, pastParticipants: participants })
),
syncType: data.historySets.syncType,
progress: data.historySets.progress,
isLatest: data.historySets.isLatest,
+39 -6
View File
@@ -6,12 +6,11 @@
* @module Utils/health-status
*/
import { globalCircuitRegistry } from './circuit-breaker.js'
import { getVersionCacheStatus } from './version-cache.js'
/**
* Circuit breaker health information (kept for k8s probe schema stability
* always reports an empty list now that circuit breakers were removed in
* favour of bounded-retry).
* Circuit breaker health information
*/
export interface CircuitBreakerHealth {
name: string
@@ -106,10 +105,44 @@ export function getHealthStatus(): HealthStatus {
})
}
// 2. Circuit breakers were removed in favour of bounded-retry.
// Keep the field for backward-compat with k8s probe schema, always empty.
// 2. Check circuit breakers
const circuitBreakers: CircuitBreakerHealth[] = []
checks.push({ name: 'circuit_breakers', status: 'pass' })
let openCircuits = 0
for (const [name, breaker] of globalCircuitRegistry.getAll()) {
const stats = breaker.getStats()
const state = breaker.getState()
circuitBreakers.push({
name,
state,
failures: stats.totalFailures,
successes: stats.totalSuccesses,
totalCalls: stats.totalCalls
})
if (state === 'open') {
openCircuits++
}
}
if (openCircuits > 0) {
checks.push({
name: 'circuit_breakers',
status: openCircuits > 2 ? 'fail' : 'warn',
message: `${openCircuits} circuit breaker(s) are open`
})
if (openCircuits > 2) {
overallStatus = 'unhealthy'
} else if (overallStatus === 'healthy') {
overallStatus = 'degraded'
}
} else {
checks.push({
name: 'circuit_breakers',
status: 'pass'
})
}
// 3. Check memory usage (warn if > 90%)
const memUsage = process.memoryUsage()
+13 -46
View File
@@ -1,6 +1,5 @@
import { pipeline } from 'stream/promises'
import { promisify } from 'util'
import { createInflate, inflate } from 'zlib'
import { inflate } from 'zlib'
import { proto } from '../../WAProto/index.js'
import type { Chat, Contact, LIDMapping, WAMessage } from '../Types'
import { WAMessageStubType } from '../Types'
@@ -16,8 +15,7 @@ import {
import { toNumber } from './generics'
import type { ILogger } from './logger.js'
import { normalizeMessageContent } from './messages'
import { DEFAULT_ORIGIN } from '../Defaults'
import { downloadContentFromMessage, getUrlFromDirectPath } from './messages-media'
import { downloadContentFromMessage } from './messages-media'
const inflatePromise = promisify(inflate)
@@ -30,15 +28,16 @@ const inflatePromise = promisify(inflate)
*/
export const downloadHistory = async (msg: proto.Message.IHistorySyncNotification, options: RequestInit) => {
const stream = await downloadContentFromMessage(msg, 'md-msg-hist', { options })
// Pipe decrypted stream directly through zlib inflate.
// Avoids allocating an intermediate buffer for the compressed payload —
// memory peaks during 50MB history syncs drop ~50% (PR upstream #2333).
const inflater = createInflate()
const chunks: Buffer[] = []
inflater.on('data', (chunk: Buffer) => chunks.push(chunk))
await pipeline(stream, inflater)
const bufferArray: Buffer[] = []
for await (const chunk of stream) {
bufferArray.push(chunk)
}
let buffer: Buffer = Buffer.concat(bufferArray)
// decompress buffer
buffer = await inflatePromise(buffer)
const buffer = Buffer.concat(chunks)
const syncData = proto.HistorySync.decode(buffer)
return syncData
}
@@ -315,7 +314,6 @@ export const processHistoryMessage = (item: proto.IHistorySync, logger?: ILogger
contacts.push({
id: chatId,
name: chat.displayName || chat.name || chat.username || undefined,
username: chat.username || undefined,
lid: chat.lidJid || chat.accountLid || undefined,
phoneNumber: chat.pnJid || undefined
})
@@ -361,7 +359,7 @@ export const processHistoryMessage = (item: proto.IHistorySync, logger?: ILogger
}
}
chats.push(chat)
chats.push({ ...chat })
}
break
@@ -376,25 +374,10 @@ export const processHistoryMessage = (item: proto.IHistorySync, logger?: ILogger
// Convert Map back to array for return
const lidPnMappings = Array.from(lidPnMap.values())
// Normalize pastParticipants: resolve LID userJids → PN so the consumer always
// receives a phone number, never an opaque LID identifier.
// Uses the lidPnMap built above (populated from phoneNumberToLidMappings + conversations).
// If a LID cannot be resolved, the original value is kept rather than dropping the participant.
const pastParticipants: proto.IPastParticipants[] = (item.pastParticipants ?? []).map(group => ({
groupJid: group.groupJid,
pastParticipants: (group.pastParticipants ?? []).map(participant => {
const userJid = participant.userJid
if (!userJid || !isAnyLidUser(userJid)) return participant
const mapping = lidPnMap.get(jidNormalizedUser(userJid))
return mapping?.pn ? { ...participant, userJid: mapping.pn } : participant
})
}))
return {
chats,
contacts,
messages,
pastParticipants,
lidPnMappings,
syncType: item.syncType,
progress: item.progress
@@ -425,23 +408,7 @@ export const downloadAndProcessHistorySyncNotification = async (
historyMsg = await downloadHistory(msg, options)
}
const result = processHistoryMessage(historyMsg, logger)
// Mirror WA Desktop behaviour: DELETE the CDN blob only after processing succeeds.
// Doing this earlier (e.g. inside downloadHistory) risks permanent history loss if
// processing throws — the server copy would be gone and retry after reconnect would fail.
if (msg.directPath) {
const cdnUrl = getUrlFromDirectPath(msg.directPath)
fetch(cdnUrl, {
...options,
method: 'DELETE',
headers: { ...((options as RequestInit).headers ?? {}), Origin: DEFAULT_ORIGIN }
}).catch(() => {
// non-fatal — server will expire it anyway
})
}
return result
return processHistoryMessage(historyMsg, logger)
}
/**
-23
View File
@@ -57,16 +57,6 @@ export type IdentityChangeContext = {
debounceCache: NodeCache<boolean>
/** Logger instance for debugging and monitoring */
logger: ILogger
/**
* Invoked right before `assertSessions` is called for an existing-session identity
* change. Used to kick off fire-and-forget side effects (e.g. tctoken re-issuance)
* in the same order WA Web does i.e. before the E2E session is re-established.
* Must not throw; implementations are responsible for their own error handling.
*
* Skipped when the refresh itself is skipped (no_identity_node, invalid_notification,
* skipped_companion_device, skipped_self_primary, debounced, skipped_offline).
*/
onBeforeSessionRefresh?: (jid: string) => void
}
// ============================================================================
@@ -180,19 +170,6 @@ export async function handleIdentityChange(
// This ensures we don't incorrectly debounce when we exit early (offline, etc.)
ctx.debounceCache.set(from, true)
// Fire-and-forget side effects (e.g. tctoken re-issuance) BEFORE the session is
// re-established. WA Web runs these in parallel with the session refresh —
// running afterwards would race with the next outbound send and risk error 463.
//
// Wrapped in try/catch so a misbehaving consumer callback cannot abort identity
// change recovery. We log and continue — assertSessions still runs so the E2E
// session always gets refreshed.
try {
ctx.onBeforeSessionRefresh?.(from)
} catch (error) {
ctx.logger.warn({ error, jid: from }, 'onBeforeSessionRefresh callback threw — continuing with session refresh')
}
// Attempt session refresh/creation
try {
await ctx.assertSessions([from], true)
+1 -1
View File
@@ -33,8 +33,8 @@ export * from './trace-context'
export * from './prometheus-metrics'
// Resilience and performance
export * from './bounded-retry'
export * from './cache-utils'
export * from './circuit-breaker'
export * from './retry-utils'
// Telemetry and detection mitigation
+8 -34
View File
@@ -10,13 +10,7 @@ import { join } from 'path'
import { Readable, Transform } from 'stream'
import { URL } from 'url'
import { proto } from '../../WAProto/index.js'
import {
DEFAULT_ORIGIN,
MEDIA_HKDF_KEY_MAPPING,
MEDIA_PATH_MAP,
type MediaType,
NEWSLETTER_MEDIA_PATH_MAP
} from '../Defaults'
import { DEFAULT_ORIGIN, MEDIA_HKDF_KEY_MAPPING, MEDIA_PATH_MAP, type MediaType } from '../Defaults'
import type {
BaileysEventMap,
DownloadableMessage,
@@ -166,7 +160,7 @@ export const extractImageThumb = async (bufferOrFilePath: Readable | Buffer | st
height: dimensions.height
}
}
} else if ('jimp' in lib && typeof lib.jimp?.Jimp === 'function') {
} else if ('jimp' in lib && (typeof lib.jimp?.Jimp === 'object' || typeof lib.jimp?.Jimp === 'function')) {
const jimp = await (lib.jimp.Jimp as any).read(bufferOrFilePath)
const dimensions = {
width: jimp.width,
@@ -618,7 +612,7 @@ export const downloadEncryptedContent = async (
const output = new Transform({
transform(chunk, _, callback) {
let data = remainingBytes.length ? Buffer.concat([remainingBytes, chunk]) : chunk
let data = Buffer.concat([remainingBytes, chunk])
const decryptLength = toSmallestChunkSize(data.length)
remainingBytes = data.slice(decryptLength)
@@ -687,10 +681,6 @@ type MediaUploadResult = {
meta_hmac?: string
ts?: number
fbid?: number
thumbnail_info?: {
thumbnail_sha256?: string
thumbnail_direct_path?: string
}
}
export type UploadParams = {
@@ -835,21 +825,11 @@ export const getWAUploadToServer = (
{ customUploadHosts, fetchAgent, logger, options }: SocketConfig,
refreshMediaConn: (force: boolean) => Promise<MediaConnInfo>
): WAMediaUploadFunction => {
return async (filePath, { mediaType, fileEncSha256B64, timeoutMs, newsletter }) => {
return async (filePath, { mediaType, fileEncSha256B64, timeoutMs }) => {
// send a query JSON to obtain the url & auth token to upload our media
let uploadInfo = await refreshMediaConn(false)
let urls:
| {
mediaUrl: string
directPath: string
meta_hmac?: string
ts?: number
fbid?: number
thumbnailDirectPath?: string
thumbnailSha256?: string
}
| undefined
let urls: { mediaUrl: string; directPath: string; meta_hmac?: string; ts?: number; fbid?: number } | undefined
const hosts = [...customUploadHosts, ...uploadInfo.hosts]
fileEncSha256B64 = encodeBase64EncodedStringForUpload(fileEncSha256B64)
@@ -871,11 +851,7 @@ export const getWAUploadToServer = (
logger.debug(`uploading to "${hostname}"`)
const auth = encodeURIComponent(uploadInfo.auth)
const mediaPath = (newsletter ? NEWSLETTER_MEDIA_PATH_MAP[mediaType] : undefined) || MEDIA_PATH_MAP[mediaType]
let url = `https://${hostname}${mediaPath}/${fileEncSha256B64}?auth=${auth}&token=${fileEncSha256B64}`
if (newsletter) {
url += '&server_thumb_gen=1'
}
const url = `https://${hostname}${MEDIA_PATH_MAP[mediaType]}/${fileEncSha256B64}?auth=${auth}&token=${fileEncSha256B64}`
let result: MediaUploadResult | undefined
try {
@@ -892,13 +868,11 @@ export const getWAUploadToServer = (
if (result?.url || result?.direct_path) {
urls = {
mediaUrl: result.url || result.direct_path!,
mediaUrl: result.url!,
directPath: result.direct_path!,
meta_hmac: result.meta_hmac,
fbid: result.fbid,
ts: result.ts,
thumbnailDirectPath: result.thumbnail_info?.thumbnail_direct_path,
thumbnailSha256: result.thumbnail_info?.thumbnail_sha256
ts: result.ts
}
break
} else {
+107 -115
View File
@@ -45,8 +45,8 @@ import {
generateThumbnail,
getAudioDuration,
getAudioWaveform,
getStream,
getRawMediaUploadData,
getStream,
type MediaDownloadOptions
} from './messages-media'
import { shouldIncludeReportingToken } from './reporting-utils'
@@ -196,11 +196,10 @@ export const prepareWAMessageMedia = async (
)
const fileSha256B64 = fileSha256.toString('base64')
const { directPath, thumbnailDirectPath, thumbnailSha256 } = await options.upload(filePath, {
const { mediaUrl, directPath } = await options.upload(filePath, {
fileEncSha256B64: fileSha256B64,
mediaType: mediaType,
timeoutMs: options.mediaUploadTimeoutMs,
newsletter: true
timeoutMs: options.mediaUploadTimeoutMs
})
await fs.unlink(filePath)
@@ -208,12 +207,10 @@ export const prepareWAMessageMedia = async (
const obj = WAProto.Message.fromObject({
// todo: add more support here
[`${mediaType}Message`]: (MessageTypeProto as any)[mediaType].fromObject({
// url intentionally omitted — newsletters use directPath only
url: mediaUrl,
directPath,
fileSha256,
fileLength,
thumbnailDirectPath,
thumbnailSha256: thumbnailSha256 ? Buffer.from(thumbnailSha256, 'base64') : undefined,
...uploadData,
media: undefined
})
@@ -239,7 +236,7 @@ export const prepareWAMessageMedia = async (
const requiresDurationComputation = mediaType === 'audio' && typeof uploadData.seconds === 'undefined'
const requiresThumbnailComputation =
(mediaType === 'image' || mediaType === 'video') && typeof uploadData['jpegThumbnail'] === 'undefined'
const requiresWaveformProcessing = mediaType === 'audio' && uploadData.ptt === true && typeof uploadData.waveform === 'undefined'
const requiresWaveformProcessing = mediaType === 'audio' && uploadData.ptt === true
const requiresAudioBackground = options.backgroundColor && mediaType === 'audio' && uploadData.ptt === true
const requiresOriginalForSomeProcessing = requiresDurationComputation || requiresThumbnailComputation
const { mediaKey, encFilePath, originalFilePath, fileEncSha256, fileSha256, fileLength } = await encryptedStream(
@@ -257,9 +254,7 @@ export const prepareWAMessageMedia = async (
(async () => {
const result = await options.upload(encFilePath, {
fileEncSha256B64,
// Use mediaTypeOverride for upload path (e.g. 'viewonce-image' → /o1/mms/image)
// Encryption already used the override via encryptedStream above
mediaType: options.mediaTypeOverride || mediaType,
mediaType,
timeoutMs: options.mediaUploadTimeoutMs
})
logger?.debug({ mediaType, cacheableKey }, 'uploaded media')
@@ -297,7 +292,7 @@ export const prepareWAMessageMedia = async (
logger?.debug('computed audio duration')
}
if (requiresWaveformProcessing && !uploadData.waveform) {
if (requiresWaveformProcessing) {
uploadData.waveform = await getAudioWaveform(filePath, logger)
logger?.debug('processed waveform')
}
@@ -480,6 +475,48 @@ export const formatNativeFlowButton = (button: NativeButton): NativeFlowButton =
}
}
const recoverCarouselImageMetadata = async (
card: CarouselMessageOptions['cards'][number],
imageMessage: proto.Message.IImageMessage,
mediaOptions: MessageContentGenerationOptions
) => {
if (imageMessage.jpegThumbnail && imageMessage.height && imageMessage.width) {
return
}
try {
const { stream } = await getStream(card.image!, mediaOptions.options)
const thumb = await extractImageThumb(stream)
if (!imageMessage.jpegThumbnail) {
imageMessage.jpegThumbnail = thumb.buffer
}
if (!imageMessage.width && thumb.original.width) {
imageMessage.width = thumb.original.width
}
if (!imageMessage.height && thumb.original.height) {
imageMessage.height = thumb.original.height
}
mediaOptions.logger?.info(
{
cardTitle: card.title,
hasJpegThumbnail: !!imageMessage.jpegThumbnail,
width: imageMessage.width,
height: imageMessage.height
},
'[CAROUSEL] Recovered image thumbnail/dimensions from source media'
)
} catch (error) {
mediaOptions.logger?.warn(
{ cardTitle: card.title, error },
'[CAROUSEL] Failed to recover image thumbnail/dimensions from source media'
)
}
}
/**
* Generates a button message using Native Flow format wrapped in viewOnceMessage
* This is the modern approach for button messages that works on iOS and Android
@@ -640,59 +677,6 @@ export const generateCarouselMessage = async (
)
}
const recoverCarouselImageMetadata = async (
imageMessage: proto.Message.IImageMessage | null | undefined,
cardImage: WAMediaUpload,
cardTitle: string | undefined,
uploadOptions: MessageContentGenerationOptions
) => {
if (!imageMessage) {
return
}
const missingThumbnail = !imageMessage.jpegThumbnail
const missingWidth = !imageMessage.width
const missingHeight = !imageMessage.height
if (!missingThumbnail && !missingWidth && !missingHeight) {
return
}
try {
const { stream } = await getStream(cardImage, uploadOptions.options)
const { buffer, original } = await extractImageThumb(stream)
if (missingThumbnail) {
imageMessage.jpegThumbnail = buffer.toString('base64')
}
if (missingWidth && original.width) {
imageMessage.width = original.width
}
if (missingHeight && original.height) {
imageMessage.height = original.height
}
uploadOptions.logger?.info(
{
cardTitle,
recoveredThumbnail: !!imageMessage.jpegThumbnail,
width: imageMessage.width,
height: imageMessage.height
},
'[CAROUSEL] Recovered image metadata from source media'
)
} catch (error) {
uploadOptions.logger?.warn(
{
cardTitle,
trace: error instanceof Error ? error.stack : String(error)
},
'[CAROUSEL] Failed source-media thumbnail fallback'
)
}
}
// Map cards to the carousel format (processing media)
const carouselCards = await Promise.all(
cards.map(async card => {
@@ -708,12 +692,10 @@ export const generateCarouselMessage = async (
if (hasMedia && mediaOptions) {
if (card.image) {
const { imageMessage } = await prepareWAMessageMedia({ image: card.image }, mediaOptions)
if (imageMessage) {
await recoverCarouselImageMetadata(card, imageMessage, mediaOptions)
}
// Mirror the working Pastorini-style result: every carousel image card should
// carry a jpegThumbnail and dimensions before it reaches the Web live renderer.
await recoverCarouselImageMetadata(imageMessage, card.image, card.title, mediaOptions)
// Validate image fields needed for WhatsApp rendering
if (imageMessage && !imageMessage.jpegThumbnail) {
mediaOptions.logger?.warn(
{ cardTitle: card.title },
@@ -1293,20 +1275,57 @@ export const generateWAMessageContent = async (
options.logger?.info('Sending CTA buttons as nativeFlowMessage with viewOnceMessage wrapper')
}
}
// Check for nativeCarousel
// Check for nativeCarousel — inline handler (validated on Android, iOS, Web)
// Direct interactiveMessage at root (field 45), NO viewOnceMessage wrapper,
// NO messageContextInfo, NO biz/bot stanza nodes needed
else if (hasNonNullishProperty(message, 'nativeCarousel')) {
const carouselMsg = message as any
const carouselOptions: CarouselMessageOptions = {
cards: carouselMsg.nativeCarousel.cards,
title: carouselMsg.nativeCarousel.title || carouselMsg.title,
text: carouselMsg.text,
footer: carouselMsg.footer
const cards = carouselMsg.nativeCarousel.cards || []
const title = carouselMsg.nativeCarousel.title || carouselMsg.title
const text = carouselMsg.text
const footer = carouselMsg.footer
const carouselCards = await Promise.all(
cards.map(async (card: any) => {
const hasMedia = !!(card.image || card.video)
const header: any = {
title: card.title || '',
subtitle: card.footer || '',
hasMediaAttachment: hasMedia
}
if (hasMedia && card.image) {
const { imageMessage } = await prepareWAMessageMedia({ image: card.image }, options)
if (imageMessage && !imageMessage.height) imageMessage.height = 500
if (imageMessage && !imageMessage.width) imageMessage.width = 500
header.imageMessage = imageMessage
}
return {
header,
body: { text: card.body || '' },
footer: card.footer ? { text: card.footer } : undefined,
nativeFlowMessage: {
buttons: (card.buttons || []).map((btn: any) => {
switch (btn.type) {
case 'url': return { name: 'cta_url', buttonParamsJson: JSON.stringify({ display_text: btn.text, url: btn.url, merchant_url: btn.url }) }
case 'copy': return { name: 'cta_copy', buttonParamsJson: JSON.stringify({ display_text: btn.text, copy_code: btn.copyText }) }
case 'call': return { name: 'cta_call', buttonParamsJson: JSON.stringify({ display_text: btn.text, phone_number: btn.phoneNumber }) }
default: return { name: 'quick_reply', buttonParamsJson: JSON.stringify({ display_text: btn.text, id: btn.id }) }
}
})
}
}
})
)
m.interactiveMessage = {
header: { title: title || ' ', hasMediaAttachment: false },
body: { text: text || '' },
footer: footer ? { text: footer } : undefined,
carouselMessage: {
cards: carouselCards,
messageVersion: 1
}
}
// Pass options for media processing if cards have images/videos
const generated = await generateCarouselMessage(carouselOptions, options)
// Frida capture shows interactiveMessage DIRECT (field 45) in DSM — no viewOnceMessage wrapper
// Testing without wrapper: biz node + quality_control already match Pastorini CDP stanza
m.interactiveMessage = generated.interactiveMessage
return m
}
// Check for nativeList
@@ -1752,49 +1771,22 @@ export const generateWAMessageContent = async (
}
}
} else {
// For view-once media, upload to the one-time CDN (/o1/mms/*) instead of regular CDN
// WA Web view-once messages always have directPath: /o1/v/t24/... (not /v/t62.7118-24/...)
// Using the wrong CDN causes WA server to flag the account and breaks view-once capability
const isViewOnce = hasOptionalProperty(message, 'viewOnce') && !!message.viewOnce
const viewOnceTypeMap: Record<string, string> = {
image: 'viewonce-image',
video: 'viewonce-video',
audio: 'viewonce-audio'
}
const viewOnceOverride = isViewOnce
? viewOnceTypeMap[Object.keys(message).find(k => k in viewOnceTypeMap) || '']
: undefined
m = await prepareWAMessageMedia(message, {
...options,
mediaTypeOverride: (viewOnceOverride as any) || options.mediaTypeOverride
})
m = await prepareWAMessageMedia(message, options)
}
if (hasOptionalProperty(message, 'viewOnce') && !!message.viewOnce) {
// Also set viewOnce=true on the inner media message (required by WA app to render as view-once)
if (m.imageMessage) m.imageMessage.viewOnce = true
else if (m.videoMessage) m.videoMessage.viewOnce = true
else if (m.audioMessage) m.audioMessage.viewOnce = true
// Use viewOnceMessageV2 (field 55) — modern WA clients use this wrapper
// viewOnceMessage (field 37) is the old format which WA server no longer accepts with view_once_write
m = { viewOnceMessageV2: { message: m } }
m = { viewOnceMessage: { message: m } }
}
if (
(hasOptionalProperty(message, 'mentions') && message.mentions?.length) ||
(hasOptionalProperty(message, 'mentionAll') && message.mentionAll)
) {
if (hasOptionalProperty(message, 'mentions') && message.mentions?.length) {
const messageType = Object.keys(m)[0] as Extract<keyof proto.IMessage, MessageWithContextInfo>
if (messageType) {
const key = m[messageType]
if (key && 'contextInfo' in key) {
key.contextInfo = key.contextInfo || {}
if (message.mentions?.length) {
if (key && 'contextInfo' in key && !!key.contextInfo) {
key.contextInfo.mentionedJid = message.mentions
}
if (message.mentionAll) {
key.contextInfo.nonJidMentions = 1
} else if (key) {
key.contextInfo = {
mentionedJid: message.mentions
}
}
}
+1 -196
View File
@@ -40,7 +40,6 @@ import { getKeyAuthor, toNumber } from './generics'
import { downloadAndProcessHistorySyncNotification } from './history'
import type { ILogger } from './logger'
import { metrics, recordHistorySyncMessages } from './prometheus-metrics.js'
import { buildMergedTcTokenIndexWrite } from './tc-token-utils'
type ProcessMessageContext = {
shouldProcessHistoryMsg: boolean
@@ -63,163 +62,6 @@ const REAL_MSG_STUB_TYPES = new Set([
const REAL_MSG_REQ_ME_STUB_TYPES = new Set([WAMessageStubType.GROUP_PARTICIPANT_ADD])
/**
* Extract tctoken / tcTokenTimestamp / tcTokenSenderTimestamp from history-sync chats
* and persist them to the `tctoken` store. Mirrors WA Web's `bulkCreateOrMerge` pass
* over the chat table during history sync.
*
* Why this matters: when a user logs in on a new device, the multi-device history sync
* is the only way that device learns about tctokens issued/received on the original
* device. Without this pass, the new device sends 1:1 messages with no tctoken until
* the contact triggers a fresh notification which surfaces as error 463 in production.
*
* Monotonicity: we only overwrite an existing entry if the incoming timestamp is
* STRICTLY newer (`incoming > existing`). Equal timestamps are skipped to avoid
* reverting senderTimestamp / realIssueTimestamp set by other layers (e.g. a
* reissue that fired between history-sync chunks).
*
* Index hygiene: every JID we write here is added to the persistent prune index
* (TC_TOKEN_INDEX_KEY) via buildMergedTcTokenIndexWrite, so the 24h prune sweep in
* messages-recv picks them up across sessions.
*/
/**
* Single-concurrency queue for `storeTcTokensFromHistorySync` calls.
*
* Why: the function does read-then-write merges (`keyStore.get('tctoken', ...)`
* compute `keyStore.set(...)`) which are NOT atomic at the store level. If two
* history-sync chunks invoke this concurrently (common during reconnect / QR
* scan), an older chunk that started first can `keyStore.set` AFTER a newer
* chunk, overwriting the newer entry and worse, the merged `__index` write
* can drop JIDs the other chunk just added. Result: stale tcTokens / repeat 463
* sends until the next opportunistic refetch.
*
* Serialising via a chained Promise keeps the runs ordered while still freeing
* the calling `processMessage` to emit `messaging-history.set` immediately
* (the chain is fire-and-forget at the call site). Errors don't break the chain
* each `catch` resets it to `Promise.resolve()` so a single failure can't
* stall future runs.
*
* The chain is module-scoped (one per Node process). Multiple Baileys instances
* sharing this module will serialise across instances too, but their writes
* target different keyStores so there's no correctness gain only a tiny loss
* of inter-instance parallelism for tcToken syncs, which is acceptable given
* how rarely this runs vs. how rare cross-instance contention is.
*/
let historyTcTokenChain: Promise<void> = Promise.resolve()
function scheduleHistoryTcTokenSync(
chats: Chat[],
signalRepository: SignalRepositoryWithLIDStore,
keyStore: SignalKeyStoreWithTransaction,
logger?: ILogger
): void {
historyTcTokenChain = historyTcTokenChain
.catch(() => {
/* swallow prior error so chain stays alive */
})
.then(() => storeTcTokensFromHistorySync(chats, signalRepository, keyStore, logger))
.catch(err => {
logger?.warn({ err }, 'background tctoken history-sync persistence failed')
})
}
async function storeTcTokensFromHistorySync(
chats: Chat[],
signalRepository: SignalRepositoryWithLIDStore,
keyStore: SignalKeyStoreWithTransaction,
logger?: ILogger
) {
// Cheap filter first — most chats in a sync chunk don't carry tcToken at all,
// and we want to avoid spinning up promises for them.
const tokenChats = chats.filter(chat => {
const ts = chat.tcTokenTimestamp ? toNumber(chat.tcTokenTimestamp) : 0
return !!chat.tcToken?.length && ts > 0
})
if (!tokenChats.length) {
return
}
// Pre-normalize so the rest of the pipeline is a synchronous join.
const normalized = tokenChats.map(chat => ({
chat,
ts: toNumber(chat.tcTokenTimestamp!),
jid: jidNormalizedUser(chat.id!)
}))
// BATCHED LID resolution. The previous shape called getLIDForPN once per
// chat (sequential await inside a for-of), which became the bottleneck
// during heavy history sync — every cold-cache hit was a DB round-trip,
// stalling messaging-history.set and spilling into the event-buffer.
// `getLIDsForPNs` resolves a deduped list in ONE batched query (and shares
// USync retry across PNs that miss cache), turning O(N) round-trips into 1.
//
// LID inputs (and `@hosted.lid`) skip the lookup entirely — they're already
// the storage form. Failures degrade gracefully: a missing mapping just
// stores under the original jid, matching `resolveTcTokenJid`'s null branch.
const pnsToResolve = [...new Set(normalized.filter(({ jid }) => !isLidUser(jid)).map(({ jid }) => jid))]
const pnToLid = new Map<string, string>()
if (pnsToResolve.length) {
try {
const mappings = await signalRepository.lidMapping.getLIDsForPNs(pnsToResolve)
// Flat loop (continue-on-skip) keeps max nesting depth at 4 for lint.
for (const { pn, lid } of mappings ?? []) {
if (!pn || !lid) continue
pnToLid.set(jidNormalizedUser(pn), lid)
}
} catch (err) {
// Per-chat fallback below (storageJid := jid). Don't abort the chunk —
// CodeRabbit noted that all-or-nothing rejection here would drop every
// tctoken in the batch AND prevent messaging-history.set from firing.
logger?.warn({ err }, 'storeTcTokensFromHistorySync: getLIDsForPNs batch failed; falling back to per-chat jid')
}
}
const candidates = normalized.map(({ chat, ts, jid }) => ({
storageJid: pnToLid.get(jid) ?? jid,
token: Buffer.from(chat.tcToken!),
ts,
senderTs: chat.tcTokenSenderTimestamp ? toNumber(chat.tcTokenSenderTimestamp) : undefined
}))
const jids = candidates.map(c => c.storageJid)
const existing = await keyStore.get('tctoken', jids)
const entries: Record<string, { token: Buffer; timestamp?: string; senderTimestamp?: number }> = {}
for (const c of candidates) {
// Same-batch dedup: when two chats resolve to the same storageJid (e.g. PN+LID
// aliases collapsing through resolveTcTokenJid, or duplicate chunks across
// retries), prefer the value already written by an earlier iteration so a
// lower-ts entry can't overwrite a higher-ts one captured from `existing`.
const existingEntry = entries[c.storageJid] ?? existing[c.storageJid]
const existingTs = existingEntry?.timestamp ? Number(existingEntry.timestamp) : 0
// Strict > guard: equal timestamps are skipped so we never clobber
// senderTimestamp written by other layers (issuance after send, etc).
if (existingTs > 0 && existingTs >= c.ts) {
continue
}
entries[c.storageJid] = {
...existingEntry,
token: c.token,
timestamp: String(c.ts),
...(c.senderTs !== undefined ? { senderTimestamp: c.senderTs } : {})
}
}
if (Object.keys(entries).length) {
logger?.debug({ count: Object.keys(entries).length }, 'storing tctokens from history sync')
try {
// Include updated __index so cross-session pruning picks these JIDs up.
const indexWrite = await buildMergedTcTokenIndexWrite(keyStore, Object.keys(entries))
await keyStore.set({ tctoken: { ...entries, ...indexWrite } })
} catch (err) {
logger?.warn({ err }, 'failed to store tctokens from history sync')
}
}
}
/** Cleans a received message to further processing */
export const cleanMessage = (message: WAMessage, meId: string, meLid: string) => {
// ensure remoteJid and participant doesn't have device or agent in it
@@ -576,12 +418,6 @@ const processMessage = async (
// Emit LID-PN mappings from history sync
// This is how WhatsApp Web learns mappings for chats with non-contacts
//
// MUST run BEFORE storeTcTokensFromHistorySync — otherwise resolveTcTokenJid()
// can't resolve PN→LID for fresh-device chats (mapping cache is empty), tokens
// get persisted under PN keys, and the send path (which resolves to LID first)
// misses them — exactly the error 463 scenario this whole change is meant to
// prevent. Catches Codex P1 review on PR #386.
if (data.lidPnMappings?.length) {
logger?.debug({ count: data.lidPnMappings.length }, 'processing LID-PN mappings from history sync')
// eslint-disable-next-line max-depth
@@ -606,29 +442,6 @@ const processMessage = async (
}
}
// Persist tctokens carried by history-sync chats in BACKGROUND, serialised.
//
// Originally awaited (PR #386) to avoid 463 on first multi-device send, but in
// production this drained the event buffer per-chunk and added visible delivery
// latency (especially after restart / QR scan when many chunks arrived at once).
//
// `scheduleHistoryTcTokenSync` enqueues onto a single-concurrency promise chain
// (see definition above) — chunks persist sequentially in the order they were
// emitted, preserving timestamp monotonicity AND keeping the `__index` write
// safe from concurrent merge clobbers. The call returns immediately so the
// `messaging-history.set` emit is not blocked.
//
// TRADE-OFF: a listener that fires an outbound send IMMEDIATELY after the emit
// may race the still-pending persistence and get a 463 on that specific send.
// The existing 463 handler in messages-recv.ts triggers a getPrivacyTokens()
// refetch that auto-recovers within seconds. Net result is much better UX than
// per-chunk stalls.
//
// DO NOT add `await` back here without re-evaluating production latency, AND
// DO NOT call storeTcTokensFromHistorySync directly — it must go through the
// chain to preserve write ordering across overlapping chunks.
scheduleHistoryTcTokenSync(data.chats, signalRepository, keyStore, logger)
ev.emit('messaging-history.set', {
...data,
isLatest: histNotification.syncType !== proto.HistorySync.HistorySyncType.ON_DEMAND ? isLatest : undefined,
@@ -950,19 +763,12 @@ const processMessage = async (
id: jid,
author: message.key.participant!,
authorPn: message.key.participantAlt!,
authorUsername: message.key.participantUsername!,
participants,
action
})
const emitGroupUpdate = (update: Partial<GroupMetadata>) => {
ev.emit('groups.update', [
{
id: jid,
...update,
author: message.key.participant ?? undefined,
authorPn: message.key.participantAlt,
authorUsername: message.key.participantUsername
}
{ id: jid, ...update, author: message.key.participant ?? undefined, authorPn: message.key.participantAlt }
])
}
@@ -971,7 +777,6 @@ const processMessage = async (
id: jid,
author: message.key.participant!,
authorPn: message.key.participantAlt!,
authorUsername: message.key.participantUsername!,
participant: participant.lid,
participantPn: participant.pn,
action,
+23
View File
@@ -1570,6 +1570,26 @@ export const metrics = {
new Counter('socket_reconnects_total', 'Total socket reconnection attempts', ['reason'])
),
// ========== Circuit Breaker Metrics ==========
circuitBreakerState: baileysMetrics.register(
new Gauge('circuit_breaker_state', 'Circuit breaker state (0=closed, 1=open, 2=half-open)', ['name'])
),
circuitBreakerTrips: baileysMetrics.register(
new Counter('circuit_breaker_trips_total', 'Total circuit breaker trips', ['name'])
),
circuitBreakerRecoveries: baileysMetrics.register(
new Counter('circuit_breaker_recoveries_total', 'Total circuit breaker recoveries', ['name'])
),
circuitBreakerRejections: baileysMetrics.register(
new Counter('circuit_breaker_rejections_total', 'Total requests rejected by circuit breaker', ['name'])
),
circuitBreakerSuccesses: baileysMetrics.register(
new Counter('circuit_breaker_successes_total', 'Total successful requests through circuit breaker', ['name'])
),
circuitBreakerFailures: baileysMetrics.register(
new Counter('circuit_breaker_failures_total', 'Total failed requests through circuit breaker', ['name'])
),
// ========== Encryption Metrics ==========
encryptionOperations: baileysMetrics.register(
new Counter('encryption_operations_total', 'Total encryption operations', ['operation'])
@@ -2147,6 +2167,9 @@ function initializeMetricsWithLabels(): void {
metrics.messageFailures?.inc({ type: 'text', reason: 'max_retries' }, 0)
metrics.messageFailures?.inc({ type: 'other', reason: 'max_retries' }, 0)
// Circuit breaker metric
metrics.circuitBreakerTrips?.inc({ name: 'main' }, 0)
console.log('[Prometheus] Initialized metrics with labels')
} catch (error) {
console.error('[Prometheus] Error initializing metrics with labels:', error)
+148 -13
View File
@@ -6,21 +6,15 @@
* - Jitter to avoid thundering herd
* - Configurable max attempts
* - Customizable retry predicates
* - Cancellation support (AbortSignal)
*
* Used by `decode-wa-message.ts` for Bad MAC retry recovery on the decrypt
* path. For socket-operation retries (uploadPreKeys etc.) prefer
* `withBoundedRetry` from `./bounded-retry.ts`, which is empirically
* aligned with WhatsApp Android's per-operation backoff.
*
* NOTE: the previous `circuitBreaker` integration option, the `withRetry`
* decorator, the `retryable` wrapper and the `RetryManager` class were
* removed when the socket-level circuit breaker was retired see PR #393
* for the rationale.
* - Circuit breaker integration
* - Event hooks
* - Cancellation support
*
* @module Utils/retry-utils
*/
import { EventEmitter } from 'events'
import type { CircuitBreaker } from './circuit-breaker.js'
import { metrics } from './prometheus-metrics.js'
/**
@@ -65,6 +59,8 @@ export interface RetryOptions {
operationName?: string
/** Collect metrics */
collectMetrics?: boolean
/** Circuit breaker for integration */
circuitBreaker?: CircuitBreaker
/** Callback before each retry */
onRetry?: (error: Error, attempt: number, delay: number) => void | Promise<void>
/** Callback on success */
@@ -287,6 +283,7 @@ export async function retry<T>(
timeout: options.timeout,
operationName: options.operationName ?? 'operation',
collectMetrics: options.collectMetrics ?? true,
circuitBreaker: options.circuitBreaker,
onRetry: options.onRetry ?? (() => {}),
onSuccess: options.onSuccess ?? (() => {}),
onFailure: options.onFailure ?? (() => {}),
@@ -316,6 +313,11 @@ export async function retry<T>(
throw new RetryAbortedError(attempt)
}
// Check circuit breaker
if (config.circuitBreaker?.isOpen()) {
throw new Error(`Circuit breaker "${config.circuitBreaker.getName()}" is open`)
}
try {
// Execute operation
let result: T
@@ -423,6 +425,138 @@ export function createRetrier(defaultOptions: RetryOptions = {}) {
}
}
/**
* Decorator to add retry to method
*/
export function withRetry(options: RetryOptions = {}) {
return function (
_target: unknown,
propertyKey: string,
descriptor: TypedPropertyDescriptor<(...args: unknown[]) => unknown>
) {
const originalMethod = descriptor.value
if (!originalMethod) return descriptor
descriptor.value = async function (...args: unknown[]): Promise<unknown> {
return retry(() => originalMethod.apply(this, args), {
...options,
operationName: options.operationName || propertyKey
})
}
return descriptor
}
}
/**
* Wrapper for function with retry
*/
// eslint-disable-next-line space-before-function-paren
export function retryable<T extends (...args: unknown[]) => unknown>(
fn: T,
options: RetryOptions = {}
): (...args: Parameters<T>) => Promise<ReturnType<T>> {
return async (...args: Parameters<T>): Promise<ReturnType<T>> => {
return retry(() => fn(...args), options) as Promise<ReturnType<T>>
}
}
/**
* Class to manage retries with state
*/
export class RetryManager extends EventEmitter {
private activeRetries: Map<string, { cancel: () => void; context: RetryContext }> = new Map()
private defaultOptions: RetryOptions
constructor(defaultOptions: RetryOptions = {}) {
super()
this.defaultOptions = defaultOptions
}
/**
* Execute operation with retry
*/
async execute<T>(
id: string,
operation: (context: RetryContext) => T | Promise<T>,
options?: RetryOptions
): Promise<T> {
// Cancel previous retry with same ID
this.cancel(id)
const abortController = new AbortController()
const mergedOptions = { ...this.defaultOptions, ...options, abortSignal: abortController.signal }
const retryPromise = retry(context => {
this.activeRetries.set(id, {
cancel: () => abortController.abort(),
context
})
this.emit('attempt', { id, attempt: context.attempt })
return operation(context)
}, mergedOptions)
try {
const result = await retryPromise
this.emit('success', { id })
return result
} catch (error) {
this.emit('failure', { id, error })
throw error
} finally {
this.activeRetries.delete(id)
}
}
/**
* Cancel in-progress retry
*/
cancel(id: string): boolean {
const active = this.activeRetries.get(id)
if (active) {
active.cancel()
this.activeRetries.delete(id)
this.emit('cancelled', { id })
return true
}
return false
}
/**
* Cancel all retries
*/
cancelAll(): void {
for (const [id, active] of this.activeRetries) {
active.cancel()
this.emit('cancelled', { id })
}
this.activeRetries.clear()
}
/**
* Check if there is an active retry
*/
isActive(id: string): boolean {
return this.activeRetries.has(id)
}
/**
* Return active retry context
*/
getContext(id: string): RetryContext | undefined {
return this.activeRetries.get(id)?.context
}
/**
* Return active retry IDs
*/
getActiveIds(): string[] {
return Array.from(this.activeRetries.keys())
}
}
/**
* Common predicates for shouldRetry
*/
@@ -526,8 +660,9 @@ export const retryConfigs = {
* Uses fixed delay array: 1s, 2s, 5s, 10s, 20s (with ±15% jitter)
*
* NOTE: Values are hardcoded instead of referencing RETRY_BACKOFF_DELAYS/RETRY_JITTER_FACTOR
* to prevent "Cannot access before initialization" errors in ESM environments
* caused by indirect circular imports through prometheus-metrics.ts.
* to prevent "Cannot access before initialization" errors in ESM environments.
* This occurs when modules are loaded in specific orders due to indirect circular imports
* (e.g., via prometheus-metrics.ts -> circuit-breaker.ts chain).
* Keep these values in sync with the constants above (lines 25, 31).
*/
rsocket: {
+94 -20
View File
@@ -6,9 +6,7 @@
* - Configurable log levels (trace, debug, info, warn, error, fatal)
* - JSON formatting for log analysis
* - Hierarchical context with child loggers
* - External system integration via fire-and-forget hooks (failures
* counted in hookFailures metric; no per-hook circuit breaker chronic
* breakage is visible to operators via the metric)
* - External system integration via hooks with circuit breaker
* - Logging metrics with Prometheus integration
* - Sensitive data sanitization
* - Log buffering for batch writes
@@ -75,6 +73,10 @@ export interface StructuredLoggerConfig {
maxLogsPerSecond?: number
/** Enable async logging queue (default: false) */
enableAsyncQueue?: boolean
/** Circuit breaker failure threshold for external hooks (default: 5) */
circuitBreakerThreshold?: number
/** Circuit breaker reset timeout in ms (default: 30000) */
circuitBreakerResetMs?: number
/** Enable Prometheus metrics integration (default: false) */
enableMetrics?: boolean
}
@@ -153,6 +155,8 @@ export interface LoggerMetrics {
bufferFlushes: number
/** External hook failures */
hookFailures: number
/** Circuit breaker trips */
circuitBreakerTrips: number
/** Average log processing time in ms */
avgProcessingTimeMs: number
}
@@ -165,6 +169,8 @@ export interface LoggerStatistics extends LoggerMetrics {
bufferSize: number
/** Rate limiter tokens available */
rateLimiterTokens: number
/** Circuit breaker state */
circuitBreakerState: 'closed' | 'open' | 'half-open'
/** Queue size (if async enabled) */
queueSize: number
/** Created timestamp */
@@ -236,6 +242,69 @@ class RateLimiter {
}
}
// ============================================================================
// CIRCUIT BREAKER
// ============================================================================
/**
* Circuit breaker for external hook protection
*/
class CircuitBreaker {
private failures = 0
private lastFailure = 0
private state: 'closed' | 'open' | 'half-open' = 'closed'
constructor(
private readonly threshold: number,
private readonly resetTimeoutMs: number
) {}
async execute<T>(operation: () => Promise<T>): Promise<T | null> {
if (this.state === 'open') {
if (Date.now() - this.lastFailure > this.resetTimeoutMs) {
this.state = 'half-open'
} else {
return null // Circuit is open, skip
}
}
try {
const result = await operation()
this.onSuccess()
return result
} catch (error) {
this.onFailure()
throw error
}
}
private onSuccess(): void {
this.failures = 0
this.state = 'closed'
}
private onFailure(): void {
this.failures++
this.lastFailure = Date.now()
if (this.failures >= this.threshold) {
this.state = 'open'
}
}
getState(): 'closed' | 'open' | 'half-open' {
// Check if should transition from open to half-open
if (this.state === 'open' && Date.now() - this.lastFailure > this.resetTimeoutMs) {
this.state = 'half-open'
}
return this.state
}
getFailures(): number {
return this.failures
}
}
// ============================================================================
// ASYNC LOG QUEUE
// ============================================================================
@@ -334,6 +403,8 @@ export class StructuredLogger implements ILogger {
private rateLimiter: RateLimiter | null = null
// Circuit breaker for external hook
private circuitBreaker: CircuitBreaker | null = null
// Async queue
private asyncQueue: AsyncLogQueue | null = null
@@ -364,6 +435,8 @@ export class StructuredLogger implements ILogger {
enableRateLimiting: config.enableRateLimiting ?? envConfig.enableRateLimiting ?? false,
maxLogsPerSecond: config.maxLogsPerSecond ?? envConfig.maxLogsPerSecond ?? 1000,
enableAsyncQueue: config.enableAsyncQueue ?? envConfig.enableAsyncQueue ?? false,
circuitBreakerThreshold: config.circuitBreakerThreshold ?? 5,
circuitBreakerResetMs: config.circuitBreakerResetMs ?? 30000,
enableMetrics: config.enableMetrics ?? envConfig.enableMetrics ?? false
}
@@ -382,6 +455,7 @@ export class StructuredLogger implements ILogger {
droppedLogs: 0,
bufferFlushes: 0,
hookFailures: 0,
circuitBreakerTrips: 0,
avgProcessingTimeMs: 0
}
@@ -390,6 +464,11 @@ export class StructuredLogger implements ILogger {
this.rateLimiter = new RateLimiter(this.config.maxLogsPerSecond)
}
// Initialize circuit breaker for external hook
if (this.config.externalHook) {
this.circuitBreaker = new CircuitBreaker(this.config.circuitBreakerThreshold, this.config.circuitBreakerResetMs)
}
// Initialize async queue
if (this.config.enableAsyncQueue) {
this.asyncQueue = new AsyncLogQueue()
@@ -490,27 +569,19 @@ export class StructuredLogger implements ILogger {
this.output(entry)
}
// External hook — fire-and-forget. Failures are counted in
// hookFailures metric so chronic breakage is visible to operators
// who can fix their hook. We do not gate further calls — that is
// the consumer's responsibility (consistent with the rest of the
// codebase after the circuit-breaker removal).
// External hook with circuit breaker
const externalHook = this.config.externalHook
if (externalHook) {
try {
const hookResult = externalHook(entry) as unknown
// Use Promise.resolve().catch() so any thenable (including
// Promise-likes that only implement .then) is handled
// safely — calling .catch directly would throw on such
// objects (Copilot review on PR #393).
if (hookResult && typeof (hookResult as PromiseLike<unknown>).then === 'function') {
void Promise.resolve(hookResult as PromiseLike<unknown>).catch(() => {
this.metrics.hookFailures++
if (externalHook && this.circuitBreaker) {
this.circuitBreaker
.execute(async () => {
await Promise.resolve(externalHook(entry))
})
}
} catch {
.catch(() => {
this.metrics.hookFailures++
if (this.circuitBreaker?.getState() === 'open') {
this.metrics.circuitBreakerTrips++
}
})
}
// Track processing time
@@ -763,6 +834,7 @@ export class StructuredLogger implements ILogger {
...this.metrics,
bufferSize: this.buffer.length,
rateLimiterTokens: this.rateLimiter?.getTokens() ?? 0,
circuitBreakerState: this.circuitBreaker?.getState() ?? 'closed',
queueSize: this.asyncQueue?.getSize() ?? 0,
createdAt: this.createdAt,
uptimeMs: Date.now() - this.createdAt
@@ -788,6 +860,7 @@ export class StructuredLogger implements ILogger {
droppedLogs: 0,
bufferFlushes: 0,
hookFailures: 0,
circuitBreakerTrips: 0,
avgProcessingTimeMs: 0
}
this.totalProcessingTime = 0
@@ -820,6 +893,7 @@ export class StructuredLogger implements ILogger {
// Clear references
this.rateLimiter = null
this.circuitBreaker = null
this.metricsModule = null
}
}
-1
View File
@@ -46,7 +46,6 @@ export const processContactAction = (
{
id,
name: action.fullName || action.firstName || action.username || undefined,
username: action.username || undefined,
lid: lidJid || undefined,
phoneNumber
}
+3 -122
View File
@@ -1,86 +1,12 @@
import type { SignalKeyStoreWithTransaction } from '../Types'
import type { BinaryNode } from '../WABinary'
import {
getBinaryNodeChild,
getBinaryNodeChildren,
isAnyLidUser,
isAnyPnUser,
isHostedLidUser,
isHostedPnUser,
isJidMetaAI,
isLidUser,
isPnUser,
jidNormalizedUser
} from '../WABinary'
import { getBinaryNodeChild, getBinaryNodeChildren, isLidUser, jidNormalizedUser } from '../WABinary'
/** 7 days in seconds — matches WA Web AB prop tctoken_duration */
const TC_TOKEN_BUCKET_DURATION = 604800
/** 4 buckets → ~28-day rolling window — matches WA Web AB prop tctoken_num_buckets */
const TC_TOKEN_NUM_BUCKETS = 4
/**
* Sentinel key under the `tctoken` store holding a JSON array of tracked storage JIDs
* for cross-session pruning. Mirrors WA Web's CLEAN_TC_TOKENS index lookup.
*
* Exported so other modules (messages-recv, messages-send, process-message) reference
* the same constant. The fork previously inlined this string in messages-recv.ts;
* keep the value identical (`'__index'`) for backward compatibility with persisted state.
*/
export const TC_TOKEN_INDEX_KEY = '__index'
// Phone-number pattern matching WABinary's isJidBot, applied against the user part so
// the check is invariant to @c.us ↔ @s.whatsapp.net normalization.
const BOT_PHONE_REGEX = /^1313555\d{4}$|^131655500\d{2}$/
/**
* Mirrors WA Web's `Wid.isRegularUser()` (user ¬PSA ¬Bot). Used to gate tctoken
* storage against malformed notifications WA Web filters server-side but we
* defend here for parity with `WAWebSetTcTokenChatAction.handleIncomingTcToken`.
* Works for both pre- and post-normalized JIDs (`@c.us` vs `@s.whatsapp.net`).
*/
export function isRegularUser(jid: string | undefined): boolean {
if (!jid) return false
const user = jid.split('@')[0] ?? ''
if (user === '0') return false // PSA
if (BOT_PHONE_REGEX.test(user)) return false // Bot by phone pattern
if (isJidMetaAI(jid)) return false // MetaAI (@bot server)
return !!(isPnUser(jid) || isLidUser(jid) || isHostedPnUser(jid) || isHostedLidUser(jid) || jid.endsWith('@c.us'))
}
/** Read the persisted tctoken JID index and return its entries (never contains the sentinel key itself). */
export async function readTcTokenIndex(keys: SignalKeyStoreWithTransaction): Promise<string[]> {
const data = await keys.get('tctoken', [TC_TOKEN_INDEX_KEY])
const entry = data[TC_TOKEN_INDEX_KEY]
if (!entry?.token?.length) return []
try {
const parsed = JSON.parse(Buffer.from(entry.token).toString())
if (!Array.isArray(parsed)) return []
return parsed.filter((j): j is string => typeof j === 'string' && j.length > 0 && j !== TC_TOKEN_INDEX_KEY)
} catch {
return []
}
}
/**
* Build a SignalDataSet fragment that writes the merged index (persisted added)
* under the sentinel key. Lets callers update the index without clobbering writes
* made by other layers (history sync, concurrent sessions on the same store).
*/
export async function buildMergedTcTokenIndexWrite(
keys: SignalKeyStoreWithTransaction,
addedJids: Iterable<string>
): Promise<{ [TC_TOKEN_INDEX_KEY]: { token: Buffer } }> {
const persisted = await readTcTokenIndex(keys)
const merged = new Set(persisted)
for (const jid of addedJids) {
if (jid && jid !== TC_TOKEN_INDEX_KEY) merged.add(jid)
}
return {
[TC_TOKEN_INDEX_KEY]: { token: Buffer.from(JSON.stringify([...merged])) }
}
}
/**
* Check if a received token is expired using WA Web's rolling bucket algorithm.
* Reference: WAWebTrustedContactsUtils.isTokenExpired
@@ -137,43 +63,6 @@ export async function resolveTcTokenJid(
return lid ?? normalized
}
/**
* Resolve target JID for issuing privacy token based on AB prop 14303
* (`lid_trusted_token_issue_to_lid`). When the prop is on, issuance goes to
* the LID; when off, it goes to the PN. Returns the original JID if no
* mapping is found in either direction.
*
* Normalizes the JID upfront and uses the `isAny*` helpers so callers can pass
* `@c.us`, `@s.whatsapp.net`, `@hosted`, `@hosted.lid`, `@lid` or device-specific
* forms `LIDMappingStore.getLIDForPN` early-returns unless `isAnyPnUser`,
* so unnormalized inputs would silently bypass routing.
*
* Reference: WAWebTrustedContactsManager.issuePrivacyTokens
*/
export async function resolveIssuanceJid(
jid: string,
issueToLid: boolean,
getLIDForPN: (pn: string) => Promise<string | null>,
getPNForLID?: (lid: string) => Promise<string | null>
): Promise<string> {
const normalized = jidNormalizedUser(jid)
if (issueToLid) {
if (isAnyLidUser(normalized)) return normalized
if (!isAnyPnUser(normalized)) return normalized
const lid = await getLIDForPN(normalized)
return lid ?? normalized
}
if (!isAnyLidUser(normalized)) return normalized
if (getPNForLID) {
const pn = await getPNForLID(normalized)
return pn ?? normalized
}
return normalized
}
type TcTokenParams = {
jid: string
baseContent?: BinaryNode[]
@@ -246,13 +135,7 @@ export async function storeTcTokensFromIqResult({
continue
}
// In notifications, tokenNode.attrs.jid is OUR own device JID, not the sender's.
// Prefer fallbackJid (resolved from notification's `from` / `sender_lid`) so the
// token is stored under the peer's JID, never under self.
const rawJid = jidNormalizedUser(fallbackJid || tokenNode.attrs.jid)
// Defense against malformed notifications (PSA WID '0', bots, MetaAI). WA Web
// filters these server-side; we mirror Wid.isRegularUser() locally.
if (!isRegularUser(rawJid)) continue
const rawJid = jidNormalizedUser(tokenNode.attrs.jid || fallbackJid)
const storageJid = await resolveTcTokenJid(rawJid, getLIDForPN)
const existingTcData = await keys.get('tctoken', [storageJid])
const existingEntry = existingTcData[storageJid]
@@ -275,13 +158,11 @@ export async function storeTcTokensFromIqResult({
...existingEntry,
token: Buffer.from(tokenNode.content),
timestamp: tokenNode.attrs.t,
// WABA Android: resets real_issue_timestamp to null when storing a new token
// (UPDATE wa_trusted_contacts_send SET real_issue_timestamp=null)
// Resets real_issue_timestamp to null when storing a new token
realIssueTimestamp: null
}
// Store under resolved storageJid AND under fallbackJid (PN) for reliable lookup
// The read path may resolve to a different LID than the store path
const normalizedFallback = jidNormalizedUser(fallbackJid)
const keysToStore: Record<string, typeof tokenEntry | null> = {
[storageJid]: tokenEntry
+37 -5
View File
@@ -15,6 +15,7 @@
*/
import type { BinaryNode } from '../WABinary/types.js'
import { CircuitBreaker, CircuitOpenError, createConnectionCircuitBreaker } from './circuit-breaker.js'
import type { ILogger } from './logger.js'
import { metrics } from './prometheus-metrics.js'
@@ -57,6 +58,8 @@ export interface UnifiedSessionOptions {
enabled?: boolean
/** Logger instance for debugging */
logger?: ILogger
/** Enable circuit breaker protection for send operations */
enableCircuitBreaker?: boolean
/** Function to send binary nodes to WhatsApp */
sendNode?: (node: BinaryNode) => Promise<void>
}
@@ -90,9 +93,7 @@ export type UnifiedSessionTrigger = 'login' | 'pairing' | 'presence' | 'manual'
* Manages the unified_session telemetry feature with:
* - Server time synchronization
* - Rate limiting (prevents spam)
* - Best-effort send semantics (failures are swallowed and counted
* telemetry is non-critical so a single attempt with the underlying
* sendNode timeout is enough; no retry, no circuit breaker)
* - Circuit breaker protection
* - Prometheus metrics integration
* - Structured logging
*
@@ -120,6 +121,7 @@ export class UnifiedSessionManager {
}
private readonly options: Required<UnifiedSessionOptions>
private circuitBreaker: CircuitBreaker | null = null
/** Minimum interval between unified_session sends (1 minute) */
private static readonly MIN_SEND_INTERVAL_MS = TimeMs.Minute
@@ -128,6 +130,7 @@ export class UnifiedSessionManager {
this.options = {
enabled: options.enabled ?? true,
logger: options.logger ?? (console as unknown as ILogger),
enableCircuitBreaker: options.enableCircuitBreaker ?? true,
sendNode:
options.sendNode ??
(async () => {
@@ -135,6 +138,25 @@ export class UnifiedSessionManager {
})
}
// Initialize circuit breaker if enabled
if (this.options.enableCircuitBreaker) {
this.circuitBreaker = createConnectionCircuitBreaker({
name: 'unified-session',
failureThreshold: 3,
failureWindow: 60000,
resetTimeout: 30000,
successThreshold: 1,
timeout: 10000,
onStateChange: (from, to) => {
this.options.logger.debug?.({ from, to }, 'Unified session circuit breaker state changed')
},
onOpen: () => {
this.options.logger.warn?.('Unified session circuit breaker OPENED')
metrics.circuitBreakerTrips?.inc({ name: 'unified-session' })
}
})
}
this.state.isInitialized = true
this.options.logger.debug?.('UnifiedSessionManager initialized')
}
@@ -270,12 +292,20 @@ export class UnifiedSessionManager {
}
try {
// Telemetry is non-critical: a single send attempt with the
// underlying sendNode timeout is enough. No retry/circuit needed.
// Execute with circuit breaker if available
if (this.circuitBreaker) {
await this.circuitBreaker.execute(sendOperation)
} else {
await sendOperation()
}
} catch (error) {
const errorMessage = error instanceof Error ? error.message : String(error)
if (error instanceof CircuitOpenError) {
this.options.logger.warn?.({ trigger, circuitState: error.state }, 'Unified session blocked by circuit breaker')
} else {
this.options.logger.warn?.({ trigger, error: errorMessage }, 'Failed to send unified session telemetry')
}
// Record failure metric
metrics.errors?.inc({ category: 'unified_session', code: 'send_failed' })
@@ -301,6 +331,7 @@ export class UnifiedSessionManager {
sendCount: 0,
isInitialized: true
}
this.circuitBreaker?.reset()
this.options.logger.debug?.('UnifiedSessionManager reset')
}
@@ -308,6 +339,7 @@ export class UnifiedSessionManager {
* Destroy the manager and clean up resources.
*/
destroy(): void {
this.circuitBreaker?.destroy()
this.state.isInitialized = false
this.options.logger.debug?.('UnifiedSessionManager destroyed')
}
-4
View File
@@ -3,7 +3,6 @@ import { createHash } from 'crypto'
import { proto } from '../../WAProto/index.js'
import {
KEY_BUNDLE_TYPE,
PROTOCOL_MODE,
WA_ADV_ACCOUNT_SIG_PREFIX,
WA_ADV_DEVICE_SIG_PREFIX,
WA_ADV_HOSTED_ACCOUNT_SIG_PREFIX
@@ -59,10 +58,7 @@ const getClientPayload = (config: SocketConfig) => {
userAgent: getUserAgent(config)
}
// Native protocol (WAM\x05) does not use WebInfo — only web protocol does
if (PROTOCOL_MODE !== 'native') {
payload.webInfo = getWebInfo(config)
}
return payload
}
+1 -27
View File
@@ -13,7 +13,7 @@ export class USyncContactProtocol implements USyncQueryProtocol {
}
getUserElement(user: USyncUser): BinaryNode {
if (user.phone) {
//TODO: Implement type / username fields (not yet supported)
return {
tag: 'contact',
attrs: {},
@@ -21,32 +21,6 @@ export class USyncContactProtocol implements USyncQueryProtocol {
}
}
if (user.username) {
return {
tag: 'contact',
attrs: {
username: user.username,
...(user.usernameKey ? { pin: user.usernameKey } : {}),
...(user.lid ? { lid: user.lid } : {})
}
}
}
if (user.type) {
return {
tag: 'contact',
attrs: {
type: user.type
}
}
}
return {
tag: 'contact',
attrs: {}
}
}
parser(node: BinaryNode): boolean {
if (node.tag === 'contact') {
assertNodeErrorFree(node)
@@ -1,28 +0,0 @@
import type { USyncQueryProtocol } from '../../Types/USync'
import { assertNodeErrorFree, type BinaryNode } from '../../WABinary'
import { USyncUser } from '../USyncUser'
export class USyncUsernameProtocol implements USyncQueryProtocol {
name = 'username'
getQueryElement(): BinaryNode {
return {
tag: 'username',
attrs: {}
}
}
getUserElement(user: USyncUser): BinaryNode | null {
void user
return null
}
parser(node: BinaryNode): string | null {
if (node.tag === 'username') {
assertNodeErrorFree(node)
return typeof node.content === 'string' ? node.content : null
}
return null
}
}
-1
View File
@@ -2,4 +2,3 @@ export * from './USyncDeviceProtocol'
export * from './USyncContactProtocol'
export * from './USyncStatusProtocol'
export * from './USyncDisappearingModeProtocol'
export * from './USyncUsernameProtocol'
+9 -22
View File
@@ -6,12 +6,11 @@ import {
USyncContactProtocol,
USyncDeviceProtocol,
USyncDisappearingModeProtocol,
USyncStatusProtocol,
USyncUsernameProtocol
USyncStatusProtocol
} from './Protocols'
import { USyncUser } from './USyncUser'
export type USyncQueryResultList = { [protocol: string]: unknown; id: string; rawId?: number }
export type USyncQueryResultList = { [protocol: string]: unknown; id: string }
export type USyncQueryResult = {
list: USyncQueryResultList[]
@@ -69,8 +68,10 @@ export class USyncQuery {
//TODO: see if there are any errors in the result node
//const resultNode = getBinaryNodeChild(usyncNode, 'result')
const parseNodeList = (content: BinaryNode[]): USyncQueryResultList[] =>
content.reduce((acc: USyncQueryResultList[], node) => {
const listNode = usyncNode ? getBinaryNodeChild(usyncNode, 'list') : undefined
if (listNode?.content && Array.isArray(listNode.content)) {
queryResult.list = listNode.content.reduce((acc: USyncQueryResultList[], node) => {
const id = node?.attrs.jid
if (id) {
const data = Array.isArray(node?.content)
@@ -88,24 +89,15 @@ export class USyncQuery {
.filter(([, b]) => b !== null) as [string, unknown][]
)
: {}
const rawIdAttr = node?.attrs?.['raw_id']
const rawId = rawIdAttr !== undefined ? Number(rawIdAttr) : undefined
acc.push({ ...data, id, ...(rawId !== undefined && !isNaN(rawId) ? { rawId } : {}) })
acc.push({ ...data, id })
}
return acc
}, [])
const listNode = usyncNode ? getBinaryNodeChild(usyncNode, 'list') : undefined
if (listNode?.content && Array.isArray(listNode.content)) {
queryResult.list = parseNodeList(listNode.content)
}
const sideListNode = usyncNode ? getBinaryNodeChild(usyncNode, 'side_list') : undefined
if (sideListNode?.content && Array.isArray(sideListNode.content)) {
queryResult.sideList = parseNodeList(sideListNode.content)
}
//TODO: implement side list
//const sideListNode = getBinaryNodeChild(usyncNode, 'side_list')
return queryResult
}
@@ -138,9 +130,4 @@ export class USyncQuery {
this.protocols.push(new USyncLIDProtocol())
return this
}
withUsernameProtocol() {
this.protocols.push(new USyncUsernameProtocol())
return this
}
}
-12
View File
@@ -2,8 +2,6 @@ export class USyncUser {
id?: string
lid?: string
phone?: string
username?: string
usernameKey?: string
type?: string
personaId?: string
@@ -22,16 +20,6 @@ export class USyncUser {
return this
}
withUsername(username: string) {
this.username = username
return this
}
withUsernameKey(usernameKey: string) {
this.usernameKey = usernameKey
return this
}
withType(type: string) {
this.type = type
return this
@@ -297,52 +297,4 @@ describe('Identity Change Handling', () => {
expect(result.device).toBe(5)
})
})
describe('onBeforeSessionRefresh callback', () => {
it('fires before assertSessions when a session refresh is about to run', async () => {
mockValidateSession.mockResolvedValue({ exists: true })
const callOrder: string[] = []
mockAssertSessions.mockImplementation(async () => {
callOrder.push('assertSessions')
return true
})
const onBeforeSessionRefresh = jest.fn((jid: string) => {
callOrder.push(`before:${jid}`)
})
const node = createIdentityChangeNode('user@s.whatsapp.net')
const ctx = { ...createContext(), onBeforeSessionRefresh }
const result = await handleIdentityChange(node, ctx)
expect(result.action).toBe('session_refreshed')
expect(callOrder).toEqual(['before:user@s.whatsapp.net', 'assertSessions'])
})
it('does not fire when the refresh is skipped (no_identity / offline / self)', async () => {
const onBeforeSessionRefresh = jest.fn()
// no identity node
const noIdentityNode: BinaryNode = {
tag: 'notification',
attrs: { from: 'a@s.whatsapp.net', type: 'encrypt' },
content: []
}
await handleIdentityChange(noIdentityNode, { ...createContext(), onBeforeSessionRefresh })
// offline notification
mockValidateSession.mockResolvedValue({ exists: true })
await handleIdentityChange(createIdentityChangeNode('b@s.whatsapp.net', '0'), {
...createContext(),
onBeforeSessionRefresh
})
// self primary
await handleIdentityChange(createIdentityChangeNode(mockMeId!), {
...createContext(),
onBeforeSessionRefresh
})
expect(onBeforeSessionRefresh).not.toHaveBeenCalled()
})
})
})
-391
View File
@@ -1,391 +0,0 @@
import { jest } from '@jest/globals'
import {
withBoundedRetry,
BoundedRetryGiveUpError,
BoundedRetryAbortedError,
WHATSAPP_BACKOFF_DELAYS,
DEFAULT_TTL_MS
} from '../../Utils/bounded-retry'
// All tests use very short real delays (1-50ms) so they run fast.
// This sidesteps the fragility of jest fake timers + async chains.
describe('bounded-retry — WhatsApp-aligned per-operation retry', () => {
test('default delay sequence matches WhatsApp Android empirical capture', () => {
// Frida trace: 3000 -> 10000 -> 60000 -> ~64000 -> 120000 ms
expect(WHATSAPP_BACKOFF_DELAYS).toEqual([3000, 10000, 60000, 60000, 120000])
})
test('default TTL is 10 minutes (matches empirical stabilisation + safety buffer)', () => {
expect(DEFAULT_TTL_MS).toBe(600_000)
})
test('first-attempt success returns immediately, no retries', async () => {
const op = jest.fn<() => Promise<string>>().mockResolvedValue('ok')
const result = await withBoundedRetry(op, { name: 'fast-success' })
expect(result).toBe('ok')
expect(op).toHaveBeenCalledTimes(1)
})
test('retries with the configured delay sequence after failures', async () => {
const op = jest
.fn<() => Promise<number>>()
.mockRejectedValueOnce(new Error('try1'))
.mockRejectedValueOnce(new Error('try2'))
.mockResolvedValueOnce(42)
const onRetry = jest.fn<(err: Error, attempt: number, delayMs: number) => void>()
const result = await withBoundedRetry(op, {
name: 'three-tries',
delays: [10, 20, 40],
jitter: 0,
onRetry
})
expect(result).toBe(42)
expect(op).toHaveBeenCalledTimes(3)
expect(onRetry).toHaveBeenCalledTimes(2)
expect(onRetry.mock.calls[0]![2]).toBe(10)
expect(onRetry.mock.calls[1]![2]).toBe(20)
})
test('TTL exhaustion throws BoundedRetryGiveUpError (no infinite retries)', async () => {
const op = jest.fn<() => Promise<string>>().mockRejectedValue(new Error('always-fails'))
await expect(
withBoundedRetry(op, {
name: 'ttl-test',
delays: [10],
jitter: 0,
ttlMs: 50
})
).rejects.toBeInstanceOf(BoundedRetryGiveUpError)
// Should have tried multiple times before giving up
expect(op.mock.calls.length).toBeGreaterThanOrEqual(2)
})
test('per-attempt timeout fires for slow operations', async () => {
const slowOp = jest.fn<() => Promise<string>>(() => new Promise(() => {})) // hangs
await expect(
withBoundedRetry(slowOp, {
name: 'slow',
delays: [5],
jitter: 0,
ttlMs: 200,
perAttemptTimeoutMs: 30
})
).rejects.toBeInstanceOf(BoundedRetryGiveUpError)
// Multiple attempts should have been tried (each timed out at 30ms)
expect(slowOp.mock.calls.length).toBeGreaterThan(1)
})
test('shouldRetry predicate can short-circuit retries', async () => {
const op = jest
.fn<() => Promise<string>>()
.mockRejectedValueOnce(new Error('transient'))
.mockRejectedValueOnce(new Error('FATAL: do not retry'))
await expect(
withBoundedRetry(op, {
name: 'predicate-test',
delays: [1],
jitter: 0,
shouldRetry: err => !err.message.startsWith('FATAL')
})
).rejects.toThrow('FATAL: do not retry')
expect(op).toHaveBeenCalledTimes(2)
})
test('AbortSignal cancels in-flight retry sleep', async () => {
const op = jest.fn<() => Promise<string>>().mockRejectedValue(new Error('fail'))
const ctrl = new AbortController()
const promise = withBoundedRetry(op, {
name: 'abort-test',
delays: [10_000], // long sleep so abort can happen during it
jitter: 0,
signal: ctrl.signal
})
// First attempt fires immediately; sleep starts
await new Promise(resolve => setTimeout(resolve, 10))
ctrl.abort()
await expect(promise).rejects.toBeInstanceOf(BoundedRetryAbortedError)
})
test('two concurrent retry chains do NOT interfere with each other (per-op isolation)', async () => {
// Key advantage over circuit breaker: failures in one chain do not
// block another chain. With circuit breaker, 5 failures from chain A
// would block chain B too. With bounded-retry, B is independent.
const flakyOp = jest
.fn<() => Promise<string>>()
.mockRejectedValueOnce(new Error('a1'))
.mockResolvedValueOnce('a-ok')
const fastOp = jest.fn<() => Promise<string>>().mockResolvedValue('b-ok')
// Start a (will retry once after 30ms) and b (immediate)
const aPromise = withBoundedRetry(flakyOp, { name: 'a', delays: [30], jitter: 0 })
const bPromise = withBoundedRetry(fastOp, { name: 'b', delays: [30], jitter: 0 })
// b should resolve right away
const bResult = await bPromise
expect(bResult).toBe('b-ok')
expect(fastOp).toHaveBeenCalledTimes(1)
// a still running — wait for it
const aResult = await aPromise
expect(aResult).toBe('a-ok')
expect(flakyOp).toHaveBeenCalledTimes(2)
})
test('delay sequence caps at last value (no unbounded growth)', async () => {
const op = jest.fn<() => Promise<string>>().mockRejectedValue(new Error('fail'))
const onRetry = jest.fn<(err: Error, attempt: number, delayMs: number) => void>()
await expect(
withBoundedRetry(op, {
name: 'cap-test',
delays: [5, 10, 15], // 3 distinct steps; 4th+ retry should reuse 15
jitter: 0,
ttlMs: 300,
onRetry
})
).rejects.toBeInstanceOf(BoundedRetryGiveUpError)
// Find delays >= 4th retry — all should be capped at 15
const lateRetries = onRetry.mock.calls.filter(c => c[1] >= 4)
expect(lateRetries.length).toBeGreaterThan(0)
for (const call of lateRetries) {
expect(call[2]).toBeLessThanOrEqual(15)
}
})
test('jitter randomises delay within +/- factor', async () => {
const op = jest.fn<() => Promise<string>>().mockRejectedValue(new Error('fail'))
const observed: number[] = []
// Single run, capture multiple onRetry calls (each with different jitter)
await withBoundedRetry(op, {
name: 'jitter-test',
delays: [50], // base delay
jitter: 0.3, // +/- 30% -> [35, 65]
ttlMs: 1000, // ttl high enough for ~10 retries
perAttemptTimeoutMs: 5,
onRetry: (_, __, d) => observed.push(d)
}).catch(() => {})
expect(observed.length).toBeGreaterThanOrEqual(3)
const unique = new Set(observed)
expect(unique.size).toBeGreaterThan(1)
// Every observed delay must be within jitter window — but the LAST one
// may be capped by remaining ttl budget, so check all but the last
const inWindow = observed.slice(0, -1)
for (const d of inWindow) {
expect(d).toBeGreaterThanOrEqual(35)
expect(d).toBeLessThanOrEqual(65)
}
})
test('delay does not exceed remaining TTL budget', async () => {
const op = jest.fn<() => Promise<string>>().mockRejectedValue(new Error('fail'))
const onRetry = jest.fn<(err: Error, attempt: number, delayMs: number) => void>()
await withBoundedRetry(op, {
name: 'budget-cap',
delays: [200], // would exceed ttl
jitter: 0,
ttlMs: 50, // very short
onRetry
}).catch(() => {})
// All scheduled delays should be capped to remaining budget
for (const call of onRetry.mock.calls) {
expect(call[2]).toBeLessThanOrEqual(50)
}
})
// ─── Tests for fixes from PR #393 review ──────────────────────────────
test('TTL is enforced BEFORE the next attempt (not just after)', async () => {
// Regression: previously TTL was only checked after a failure, so a
// new attempt could start with 0ms remaining and run for the full
// per-attempt timeout, overshooting wall-clock budget by that much.
const slowOp = jest.fn<() => Promise<string>>(() => new Promise(resolve => setTimeout(() => resolve('late'), 200)))
const start = Date.now()
await expect(
withBoundedRetry(slowOp, {
name: 'ttl-strict',
delays: [10],
jitter: 0,
ttlMs: 50,
perAttemptTimeoutMs: 1000 // huge, must be capped by TTL
})
).rejects.toBeInstanceOf(BoundedRetryGiveUpError)
const elapsed = Date.now() - start
// With strict TTL enforcement, total runtime should be very close to ttlMs.
// Allow up to 100ms slack for Node timer + microtask scheduling.
expect(elapsed).toBeLessThan(150)
})
test('per-attempt timeout is capped by remaining TTL budget', async () => {
// If user passes perAttemptTimeoutMs > ttlMs, the cap should apply.
const slowOp = jest.fn<() => Promise<string>>(() => new Promise(() => {})) // hangs forever
const start = Date.now()
await expect(
withBoundedRetry(slowOp, {
name: 'attempt-cap',
delays: [1],
jitter: 0,
ttlMs: 30,
perAttemptTimeoutMs: 5000 // way bigger than ttl
})
).rejects.toBeInstanceOf(BoundedRetryGiveUpError)
const elapsed = Date.now() - start
expect(elapsed).toBeLessThan(150) // not 5000ms!
})
test('operation receives an AbortSignal that is aborted on per-attempt timeout', async () => {
// New API: operation can opt into cancellation via the signal arg.
const aborts: boolean[] = []
const opThatRespectsSignal = jest.fn(
(signal?: AbortSignal) =>
new Promise<string>((resolve, reject) => {
const timer = setTimeout(() => resolve('late'), 100)
signal?.addEventListener('abort', () => {
clearTimeout(timer)
aborts.push(true)
reject(new Error('aborted by signal'))
})
})
)
await withBoundedRetry(opThatRespectsSignal, {
name: 'signal-aware',
delays: [1],
jitter: 0,
ttlMs: 80,
perAttemptTimeoutMs: 20 // smaller than 100ms op duration
}).catch(() => {})
expect(aborts.length).toBeGreaterThan(0) // at least one attempt was aborted
})
test('sleep listener is removed when timeout completes normally (no leak)', async () => {
// Regression: sleep used signal.addEventListener with { once: true } but
// never removed the listener on timer-resolve. With many retries on a
// shared signal, listeners would accumulate.
//
// Make the failure mode observable by spying on add/removeEventListener
// directly so the assertion fails if cleanup is removed.
const ctrl = new AbortController()
const realAdd = ctrl.signal.addEventListener.bind(ctrl.signal)
const realRemove = ctrl.signal.removeEventListener.bind(ctrl.signal)
let addCount = 0
let removeCount = 0
ctrl.signal.addEventListener = ((type: string, listener: never, opts?: never) => {
if (type === 'abort') addCount++
return realAdd(type, listener, opts)
}) as typeof ctrl.signal.addEventListener
ctrl.signal.removeEventListener = ((type: string, listener: never, opts?: never) => {
if (type === 'abort') removeCount++
return realRemove(type, listener, opts)
}) as typeof ctrl.signal.removeEventListener
const op = jest
.fn<() => Promise<string>>()
.mockRejectedValueOnce(new Error('1'))
.mockRejectedValueOnce(new Error('2'))
.mockRejectedValueOnce(new Error('3'))
.mockResolvedValueOnce('ok')
const result = await withBoundedRetry(op, {
name: 'no-leak',
delays: [5],
jitter: 0,
ttlMs: 500,
signal: ctrl.signal
})
expect(result).toBe('ok')
expect(op).toHaveBeenCalledTimes(4)
// If sleep() leaks listeners, addCount > removeCount. Each retry adds
// listeners (1 sleep + 1 outer-abort forwarding) and must remove the
// same number on settlement.
expect(addCount).toBeGreaterThan(0)
expect(addCount).toBe(removeCount)
})
test('logger receives structured logs for retry, give-up, and recovery', async () => {
const debugCalls: unknown[][] = []
const infoCalls: unknown[][] = []
const warnCalls: unknown[][] = []
const fakeLogger = {
debug: (...args: unknown[]) => debugCalls.push(args),
info: (...args: unknown[]) => infoCalls.push(args),
warn: (...args: unknown[]) => warnCalls.push(args),
error: () => {},
fatal: () => {},
trace: () => {},
child: () => fakeLogger,
level: 'debug'
}
const op = jest
.fn<() => Promise<string>>()
.mockRejectedValueOnce(new Error('boom'))
.mockResolvedValueOnce('ok')
const result = await withBoundedRetry(op, {
name: 'logged-op',
delays: [5],
jitter: 0,
ttlMs: 200,
logger: fakeLogger as never
})
expect(result).toBe('ok')
// Should have logged the scheduling at debug level
expect(debugCalls.length).toBeGreaterThanOrEqual(1)
// Should have logged the recovery at info level
expect(infoCalls.length).toBeGreaterThanOrEqual(1)
})
test('logger reports give-up via warn when TTL exceeded', async () => {
const warnCalls: unknown[][] = []
const fakeLogger = {
debug: () => {},
info: () => {},
warn: (...args: unknown[]) => warnCalls.push(args),
error: () => {},
fatal: () => {},
trace: () => {},
child: () => fakeLogger,
level: 'debug'
}
const op = jest.fn<() => Promise<string>>().mockRejectedValue(new Error('always-fails'))
await expect(
withBoundedRetry(op, {
name: 'logged-fail',
delays: [5],
jitter: 0,
ttlMs: 30,
logger: fakeLogger as never
})
).rejects.toBeInstanceOf(BoundedRetryGiveUpError)
expect(warnCalls.length).toBeGreaterThanOrEqual(1)
})
})
+371
View File
@@ -0,0 +1,371 @@
/**
* Testes unitários para circuit-breaker.ts
*/
import { afterEach, beforeEach, describe, expect, it } from '@jest/globals'
import {
CircuitBreaker,
CircuitBreakerRegistry,
CircuitOpenError,
type CircuitState,
CircuitTimeoutError,
createCircuitBreaker,
getCircuitHealth,
globalCircuitRegistry,
withCircuitBreaker
} from '../../Utils/circuit-breaker.js'
describe('CircuitBreaker', () => {
let breaker: CircuitBreaker
beforeEach(() => {
breaker = createCircuitBreaker({
name: 'test',
failureThreshold: 3,
successThreshold: 2,
resetTimeout: 100,
timeout: 1000,
volumeThreshold: 3,
collectMetrics: false
})
})
afterEach(() => {
breaker.destroy()
})
describe('initial state', () => {
it('should start in closed state', () => {
expect(breaker.getState()).toBe('closed')
expect(breaker.isClosed()).toBe(true)
expect(breaker.isOpen()).toBe(false)
expect(breaker.isHalfOpen()).toBe(false)
})
})
describe('successful operations', () => {
it('should execute successful operations', async () => {
const result = await breaker.execute(() => 'success')
expect(result).toBe('success')
})
it('should execute async operations', async () => {
const result = await breaker.execute(async () => {
await new Promise(resolve => setTimeout(resolve, 10))
return 'async success'
})
expect(result).toBe('async success')
})
it('should track successful calls in stats', async () => {
await breaker.execute(() => 'success')
await breaker.execute(() => 'success')
const stats = breaker.getStats()
expect(stats.totalCalls).toBe(2)
expect(stats.totalSuccesses).toBe(2)
expect(stats.totalFailures).toBe(0)
})
})
describe('failure handling', () => {
it('should open after reaching failure threshold', async () => {
const failingOp = () => {
throw new Error('Failure')
}
// Reach failure threshold
for (let i = 0; i < 3; i++) {
await expect(breaker.execute(failingOp)).rejects.toThrow('Failure')
}
expect(breaker.isOpen()).toBe(true)
})
it('should not open before reaching threshold', async () => {
const failingOp = () => {
throw new Error('Failure')
}
// Below threshold
for (let i = 0; i < 2; i++) {
await expect(breaker.execute(failingOp)).rejects.toThrow('Failure')
}
expect(breaker.isClosed()).toBe(true)
})
it('should reset failure count on success', async () => {
const failingOp = () => {
throw new Error('Failure')
}
await expect(breaker.execute(failingOp)).rejects.toThrow()
await expect(breaker.execute(failingOp)).rejects.toThrow()
// Success resets failures
await breaker.execute(() => 'success')
// Only 1 more failure needed (already have 2 in window, need 3 total)
await expect(breaker.execute(failingOp)).rejects.toThrow()
// Should open now (3 failures in window)
expect(breaker.isOpen()).toBe(true)
})
})
describe('open state', () => {
beforeEach(async () => {
// Force open
breaker.trip()
})
it('should reject operations when open', async () => {
await expect(breaker.execute(() => 'success')).rejects.toThrow(CircuitOpenError)
})
it('should transition to half-open after reset timeout', async () => {
await new Promise(resolve => setTimeout(resolve, 150))
expect(breaker.isHalfOpen()).toBe(true)
})
})
describe('half-open state', () => {
beforeEach(async () => {
breaker.trip()
await new Promise(resolve => setTimeout(resolve, 150))
})
it('should close after success threshold', async () => {
expect(breaker.isHalfOpen()).toBe(true)
// Success threshold is 2
await breaker.execute(() => 'success')
await breaker.execute(() => 'success')
expect(breaker.isClosed()).toBe(true)
})
it('should reopen on failure', async () => {
expect(breaker.isHalfOpen()).toBe(true)
await expect(
breaker.execute(() => {
throw new Error('Failure')
})
).rejects.toThrow()
expect(breaker.isOpen()).toBe(true)
})
})
describe('manual controls', () => {
it('should force open with trip()', () => {
breaker.trip()
expect(breaker.isOpen()).toBe(true)
})
it('should force close with reset()', async () => {
breaker.trip()
breaker.reset()
expect(breaker.isClosed()).toBe(true)
})
})
describe('timeout', () => {
it('should timeout slow operations', async () => {
const slowBreaker = createCircuitBreaker({
name: 'slow',
timeout: 50,
collectMetrics: false
})
await expect(slowBreaker.execute(() => new Promise(resolve => setTimeout(resolve, 200)))).rejects.toThrow(
CircuitTimeoutError
)
slowBreaker.destroy()
})
})
describe('events', () => {
it('should emit state-change event', async () => {
const stateChanges: Array<{ from: CircuitState; to: CircuitState }> = []
breaker.on('state-change', change => {
stateChanges.push(change)
})
breaker.trip()
breaker.reset()
expect(stateChanges).toHaveLength(2)
expect(stateChanges[0]).toEqual({ from: 'closed', to: 'open' })
expect(stateChanges[1]).toEqual({ from: 'open', to: 'closed' })
})
it('should emit success and failure events', async () => {
let successCount = 0
let failureCount = 0
breaker.on('success', () => successCount++)
breaker.on('failure', () => failureCount++)
await breaker.execute(() => 'success')
await expect(
breaker.execute(() => {
throw new Error()
})
).rejects.toThrow()
expect(successCount).toBe(1)
expect(failureCount).toBe(1)
})
})
describe('isFailure predicate', () => {
it('should use custom isFailure predicate', async () => {
const customBreaker = createCircuitBreaker({
name: 'custom',
failureThreshold: 1,
volumeThreshold: 1,
shouldCountError: (error: any) => error.message !== 'Ignored',
collectMetrics: false
})
// This error should be ignored
await expect(
customBreaker.execute(() => {
throw new Error('Ignored')
})
).rejects.toThrow()
expect(customBreaker.isClosed()).toBe(true)
// This should trip the breaker
await expect(
customBreaker.execute(() => {
throw new Error('Real failure')
})
).rejects.toThrow()
expect(customBreaker.isOpen()).toBe(true)
customBreaker.destroy()
})
})
describe('sync operations', () => {
it('should execute sync operations', () => {
const result = breaker.executeSync(() => 'sync result')
expect(result).toBe('sync result')
})
it('should handle sync failures', () => {
expect(() =>
breaker.executeSync(() => {
throw new Error('Sync failure')
})
).toThrow('Sync failure')
})
})
})
describe('CircuitBreakerRegistry', () => {
let registry: CircuitBreakerRegistry
beforeEach(() => {
registry = new CircuitBreakerRegistry()
})
afterEach(() => {
registry.destroyAll()
})
it('should create and retrieve circuit breakers', () => {
const breaker1 = registry.get('test1')
const breaker2 = registry.get('test1')
expect(breaker1).toBe(breaker2)
})
it('should check if breaker exists', () => {
registry.get('exists')
expect(registry.has('exists')).toBe(true)
expect(registry.has('notexists')).toBe(false)
})
it('should remove breaker', () => {
registry.get('toRemove')
expect(registry.remove('toRemove')).toBe(true)
expect(registry.has('toRemove')).toBe(false)
})
it('should get all stats', () => {
registry.get('breaker1')
registry.get('breaker2')
const stats = registry.getAllStats()
expect(stats).toHaveProperty('breaker1')
expect(stats).toHaveProperty('breaker2')
})
it('should reset all breakers', async () => {
const breaker = registry.get('test')
breaker.trip()
registry.resetAll()
expect(breaker.isClosed()).toBe(true)
})
})
describe('withCircuitBreaker', () => {
it('should wrap function with circuit breaker', async () => {
let callCount = 0
const protectedFn = withCircuitBreaker(
async () => {
callCount++
return 'result'
},
{
name: 'wrapped-fn',
collectMetrics: false
}
)
const result = await protectedFn()
expect(result).toBe('result')
expect(callCount).toBe(1)
})
})
describe('getCircuitHealth', () => {
beforeEach(() => {
globalCircuitRegistry.destroyAll()
})
it('should report healthy when all circuits closed', () => {
globalCircuitRegistry.get('healthy1')
globalCircuitRegistry.get('healthy2')
const health = getCircuitHealth()
expect(health.healthy).toBe(true)
expect(health.openCircuits).toHaveLength(0)
})
it('should report unhealthy when circuit is open', () => {
const breaker = globalCircuitRegistry.get('unhealthy')
breaker.trip()
const health = getCircuitHealth()
expect(health.healthy).toBe(false)
expect(health.openCircuits).toContain('unhealthy')
})
})
+100 -1
View File
@@ -2,7 +2,7 @@
* Testes unitários para retry-utils.ts
*/
import { describe, expect, it, jest } from '@jest/globals'
import { beforeEach, describe, expect, it, jest } from '@jest/globals'
import {
calculateDelay,
createRetrier,
@@ -11,10 +11,12 @@ import {
retry,
RETRY_BACKOFF_DELAYS,
RETRY_JITTER_FACTOR,
retryable,
RetryAbortedError,
retryConfigs,
type RetryContext,
RetryExhaustedError,
RetryManager,
retryPredicates,
retryWithResult
} from '../../Utils/retry-utils.js'
@@ -313,6 +315,103 @@ describe('createRetrier', () => {
})
})
describe('retryable', () => {
it('should wrap function with retry', async () => {
let attempts = 0
const fn = retryable(
() => {
attempts++
if (attempts < 2) throw new Error('Failing')
return 'wrapped result'
},
{
maxAttempts: 5,
baseDelay: 10,
collectMetrics: false
}
)
const result = await fn()
expect(result).toBe('wrapped result')
expect(attempts).toBe(2)
})
})
describe('RetryManager', () => {
let manager: RetryManager
beforeEach(() => {
manager = new RetryManager({ baseDelay: 200, collectMetrics: false })
})
it('should execute operation with id', async () => {
const result = await manager.execute('op1', () => 'result')
expect(result).toBe('result')
})
it('should cancel active retry', async () => {
let attempt = 0
const promise = manager.execute('cancelable', async () => {
attempt++
if (attempt === 1) {
throw new Error('Temporary failure')
}
await new Promise(resolve => setTimeout(resolve, 1000))
return 'result'
})
// Cancel during retry delay
setTimeout(() => manager.cancel('cancelable'), 100)
await expect(promise).rejects.toThrow()
})
it('should check if operation is active', async () => {
let resolve: () => void
const promise = manager.execute(
'active',
() =>
new Promise<string>(r => {
resolve = () => r('done')
})
)
expect(manager.isActive('active')).toBe(true)
resolve!()
await promise
expect(manager.isActive('active')).toBe(false)
})
it('should cancel all operations', async () => {
const promises = [
manager.execute('op1', () => new Promise((_, reject) => setTimeout(() => reject(new Error()), 1000))),
manager.execute('op2', () => new Promise((_, reject) => setTimeout(() => reject(new Error()), 1000)))
]
setTimeout(() => manager.cancelAll(), 20)
await expect(Promise.all(promises)).rejects.toThrow()
})
it('should emit events', async () => {
const events: string[] = []
manager.on('attempt', () => events.push('attempt'))
manager.on('success', () => events.push('success'))
await manager.execute('test', () => 'result')
expect(events).toContain('attempt')
expect(events).toContain('success')
})
})
describe('retryPredicates', () => {
describe('always', () => {
it('should always return true', () => {
+1 -348
View File
@@ -1,17 +1,7 @@
import { jest } from '@jest/globals'
import { DisconnectReason, type SignalKeyStoreWithTransaction } from '../../Types'
import { getErrorCodeFromStreamError, SERVER_ERROR_CODES } from '../../Utils'
import {
buildMergedTcTokenIndexWrite,
buildTcTokenFromJid,
isRegularUser,
isTcTokenExpired,
readTcTokenIndex,
resolveIssuanceJid,
shouldSendNewTcToken,
storeTcTokensFromIqResult,
TC_TOKEN_INDEX_KEY
} from '../../Utils/tc-token-utils'
import { buildTcTokenFromJid, isTcTokenExpired, shouldSendNewTcToken } from '../../Utils/tc-token-utils'
import type { BinaryNode } from '../../WABinary'
/** 7 days in seconds — matches WA Web tctoken_duration */
@@ -858,340 +848,3 @@ describe('tctoken integration scenarios', () => {
})
})
})
// ─── isRegularUser (PSA / bot / MetaAI gating) ─────────────────────────
describe('isRegularUser', () => {
it('rejects undefined / empty', () => {
expect(isRegularUser(undefined)).toBe(false)
expect(isRegularUser('')).toBe(false)
})
it('rejects PSA WID (user "0")', () => {
expect(isRegularUser('0@s.whatsapp.net')).toBe(false)
expect(isRegularUser('0@c.us')).toBe(false)
})
it('rejects bot phone numbers (1313555XXXX, 131655500XX)', () => {
expect(isRegularUser('13135550000@s.whatsapp.net')).toBe(false)
expect(isRegularUser('13135559999@s.whatsapp.net')).toBe(false)
// /^131655500\d{2}$/ — 11 digits: 131655500 + 2 trailing
expect(isRegularUser('13165550000@s.whatsapp.net')).toBe(false)
expect(isRegularUser('13165550099@s.whatsapp.net')).toBe(false)
})
it('rejects MetaAI (@bot server)', () => {
expect(isRegularUser('foo@bot')).toBe(false)
})
it('accepts regular PN users (@s.whatsapp.net, @c.us)', () => {
expect(isRegularUser('5511999999999@s.whatsapp.net')).toBe(true)
expect(isRegularUser('5511999999999@c.us')).toBe(true)
})
it('accepts LID users', () => {
expect(isRegularUser('123456@lid')).toBe(true)
})
it('accepts hosted PN/LID users', () => {
expect(isRegularUser('5511999999999@hosted')).toBe(true)
expect(isRegularUser('123456@hosted.lid')).toBe(true)
})
})
// ─── resolveIssuanceJid (AB prop 14303 routing) ────────────────────────
describe('resolveIssuanceJid', () => {
const PN = '5511999999999@s.whatsapp.net'
const LID = '123456@lid'
const PN_NO_LID = '5511888888888@s.whatsapp.net'
const LID_NO_PN = '999999@lid'
const getLIDForPN = jest.fn<(pn: string) => Promise<string | null>>(async (pn: string) => {
if (pn === PN) return LID
return null
})
const getPNForLID = jest.fn<(lid: string) => Promise<string | null>>(async (lid: string) => {
if (lid === LID) return PN
return null
})
beforeEach(() => {
getLIDForPN.mockClear()
getPNForLID.mockClear()
})
describe('AB prop 14303 ON (issueToLid=true)', () => {
it('returns LID unchanged when input is LID', async () => {
expect(await resolveIssuanceJid(LID, true, getLIDForPN, getPNForLID)).toBe(LID)
expect(getLIDForPN).not.toHaveBeenCalled()
})
it('resolves PN to LID via lidMapping', async () => {
expect(await resolveIssuanceJid(PN, true, getLIDForPN, getPNForLID)).toBe(LID)
expect(getLIDForPN).toHaveBeenCalledWith(PN)
})
it('falls back to original PN when no LID mapping exists', async () => {
expect(await resolveIssuanceJid(PN_NO_LID, true, getLIDForPN, getPNForLID)).toBe(PN_NO_LID)
})
})
describe('AB prop 14303 OFF (issueToLid=false)', () => {
it('returns PN unchanged when input is PN', async () => {
expect(await resolveIssuanceJid(PN, false, getLIDForPN, getPNForLID)).toBe(PN)
expect(getPNForLID).not.toHaveBeenCalled()
})
it('resolves LID to PN via lidMapping', async () => {
expect(await resolveIssuanceJid(LID, false, getLIDForPN, getPNForLID)).toBe(PN)
expect(getPNForLID).toHaveBeenCalledWith(LID)
})
it('falls back to original LID when no PN mapping exists', async () => {
expect(await resolveIssuanceJid(LID_NO_PN, false, getLIDForPN, getPNForLID)).toBe(LID_NO_PN)
})
it('returns LID unchanged when getPNForLID is omitted', async () => {
expect(await resolveIssuanceJid(LID, false, getLIDForPN)).toBe(LID)
})
})
describe('JID normalization (handles @c.us and hosted forms)', () => {
const PN_CUS = '5511999999999@c.us'
const PN_NORMALIZED = '5511999999999@s.whatsapp.net'
const HOSTED_LID = '999999@hosted.lid'
it('normalizes @c.us before resolving (issueToLid=true)', async () => {
const fn = jest.fn<(pn: string) => Promise<string | null>>(async (pn: string) => {
if (pn === PN_NORMALIZED) return LID
return null
})
const result = await resolveIssuanceJid(PN_CUS, true, fn, getPNForLID)
expect(result).toBe(LID)
// Mapping store called with NORMALIZED form
expect(fn).toHaveBeenCalledWith(PN_NORMALIZED)
})
it('treats hosted LID as LID input (issueToLid=true returns it unchanged)', async () => {
expect(await resolveIssuanceJid(HOSTED_LID, true, getLIDForPN, getPNForLID)).toBe(HOSTED_LID)
expect(getLIDForPN).not.toHaveBeenCalled()
})
it('treats hosted LID as LID input (issueToLid=false converts via getPNForLID)', async () => {
const fn = jest.fn<(lid: string) => Promise<string | null>>(async () => null)
await resolveIssuanceJid(HOSTED_LID, false, getLIDForPN, fn)
expect(fn).toHaveBeenCalledWith(HOSTED_LID)
})
})
})
// ─── readTcTokenIndex / buildMergedTcTokenIndexWrite ───────────────────
describe('tctoken cross-session prune index', () => {
it('exports the sentinel key as "__index"', () => {
// Persisted state compatibility — tests freeze the value to catch unintended renames.
expect(TC_TOKEN_INDEX_KEY).toBe('__index')
})
describe('readTcTokenIndex', () => {
it('returns [] when the index entry is absent', async () => {
const keys = createMockKeys()
;(keys.get as any).mockResolvedValue({})
expect(await readTcTokenIndex(keys)).toEqual([])
})
it('returns [] when the index token buffer is empty', async () => {
const keys = createMockKeys()
;(keys.get as any).mockResolvedValue({ [TC_TOKEN_INDEX_KEY]: { token: Buffer.alloc(0) } })
expect(await readTcTokenIndex(keys)).toEqual([])
})
it('returns [] when the JSON payload is corrupted', async () => {
const keys = createMockKeys()
;(keys.get as any).mockResolvedValue({ [TC_TOKEN_INDEX_KEY]: { token: Buffer.from('not-json') } })
expect(await readTcTokenIndex(keys)).toEqual([])
})
it('returns [] when the JSON payload is not an array', async () => {
const keys = createMockKeys()
;(keys.get as any).mockResolvedValue({ [TC_TOKEN_INDEX_KEY]: { token: Buffer.from('{"a":1}') } })
expect(await readTcTokenIndex(keys)).toEqual([])
})
it('returns the persisted JIDs', async () => {
const keys = createMockKeys()
const stored = ['a@lid', 'b@lid', 'c@s.whatsapp.net']
;(keys.get as any).mockResolvedValue({
[TC_TOKEN_INDEX_KEY]: { token: Buffer.from(JSON.stringify(stored)) }
})
expect(await readTcTokenIndex(keys)).toEqual(stored)
})
it('filters out the sentinel key itself, empty strings, and non-strings', async () => {
const keys = createMockKeys()
const stored = ['a@lid', '', TC_TOKEN_INDEX_KEY, 42, null, 'b@lid']
;(keys.get as any).mockResolvedValue({
[TC_TOKEN_INDEX_KEY]: { token: Buffer.from(JSON.stringify(stored)) }
})
expect(await readTcTokenIndex(keys)).toEqual(['a@lid', 'b@lid'])
})
})
describe('buildMergedTcTokenIndexWrite', () => {
it('merges added JIDs with the persisted set (de-duplicated)', async () => {
const keys = createMockKeys()
;(keys.get as any).mockResolvedValue({
[TC_TOKEN_INDEX_KEY]: { token: Buffer.from(JSON.stringify(['a@lid', 'b@lid'])) }
})
const write = await buildMergedTcTokenIndexWrite(keys, ['b@lid', 'c@lid'])
const decoded = JSON.parse(write[TC_TOKEN_INDEX_KEY].token.toString()) as string[]
expect(decoded.sort()).toEqual(['a@lid', 'b@lid', 'c@lid'])
})
it('drops the sentinel key from the added set', async () => {
const keys = createMockKeys()
;(keys.get as any).mockResolvedValue({})
const write = await buildMergedTcTokenIndexWrite(keys, [TC_TOKEN_INDEX_KEY, 'a@lid'])
const decoded = JSON.parse(write[TC_TOKEN_INDEX_KEY].token.toString()) as string[]
expect(decoded).toEqual(['a@lid'])
})
it('handles empty inputs', async () => {
const keys = createMockKeys()
;(keys.get as any).mockResolvedValue({})
const write = await buildMergedTcTokenIndexWrite(keys, [])
const decoded = JSON.parse(write[TC_TOKEN_INDEX_KEY].token.toString()) as string[]
expect(decoded).toEqual([])
})
})
})
// ─── storeTcTokensFromIqResult — isRegularUser gating + monotonicity ───
describe('storeTcTokensFromIqResult — gating and monotonicity', () => {
const PEER_PN = '5511999999999@s.whatsapp.net'
const PEER_LID = '123456@lid'
const getLIDForPN = jest.fn<(pn: string) => Promise<string | null>>(async (pn: string) => {
if (pn === PEER_PN) return PEER_LID
return null
})
beforeEach(() => {
getLIDForPN.mockClear()
})
const buildIqResult = (jid: string, tokenBytes: Uint8Array, t: string): BinaryNode => ({
tag: 'iq',
attrs: {},
content: [
{
tag: 'tokens',
attrs: {},
content: [
{
tag: 'token',
attrs: { jid, type: 'trusted_contact', t },
content: tokenBytes
}
]
}
]
})
it('skips PSA WID (jid="0") even if returned in tokens block', async () => {
const keys = createMockKeys()
;(keys.get as any).mockResolvedValue({})
await storeTcTokensFromIqResult({
result: buildIqResult('myDeviceJid@s.whatsapp.net', Buffer.from([0x01, 0x02]), '1700000000'),
fallbackJid: '0@s.whatsapp.net',
keys,
getLIDForPN
})
expect(keys.set).not.toHaveBeenCalled()
})
it('skips bot phone numbers', async () => {
const keys = createMockKeys()
;(keys.get as any).mockResolvedValue({})
await storeTcTokensFromIqResult({
result: buildIqResult('myDeviceJid@s.whatsapp.net', Buffer.from([0x01, 0x02]), '1700000000'),
fallbackJid: '13135550000@s.whatsapp.net',
keys,
getLIDForPN
})
expect(keys.set).not.toHaveBeenCalled()
})
it('skips MetaAI (@bot server)', async () => {
const keys = createMockKeys()
;(keys.get as any).mockResolvedValue({})
await storeTcTokensFromIqResult({
result: buildIqResult('myDeviceJid@s.whatsapp.net', Buffer.from([0x01, 0x02]), '1700000000'),
fallbackJid: 'meta-ai@bot',
keys,
getLIDForPN
})
expect(keys.set).not.toHaveBeenCalled()
})
it('uses fallbackJid (NOT the token node jid which is own device) as storage source', async () => {
const keys = createMockKeys()
;(keys.get as any).mockResolvedValue({})
const onNewJidStored = jest.fn()
await storeTcTokensFromIqResult({
result: buildIqResult('myDeviceJid:1@s.whatsapp.net', Buffer.from([0xaa, 0xbb]), '1700000000'),
fallbackJid: PEER_PN,
keys,
getLIDForPN,
onNewJidStored
})
expect(keys.set).toHaveBeenCalled()
const setArgs = (keys.set.mock.calls[0]?.[0] as { tctoken: Record<string, unknown> }).tctoken
expect(Object.keys(setArgs)).toContain(PEER_LID)
expect(onNewJidStored).toHaveBeenCalledWith(PEER_LID)
})
it('skips when incoming timestamp is older than existing (monotonicity)', async () => {
const keys = createMockKeys()
;(keys.get as any).mockResolvedValue({
[PEER_LID]: { token: Buffer.from([0x99]), timestamp: '1800000000' }
})
await storeTcTokensFromIqResult({
result: buildIqResult('myDeviceJid@s.whatsapp.net', Buffer.from([0xaa]), '1700000000'),
fallbackJid: PEER_PN,
keys,
getLIDForPN
})
expect(keys.set).not.toHaveBeenCalled()
})
it('skips timestamp-less tokens (would be immediately expired)', async () => {
const keys = createMockKeys()
;(keys.get as any).mockResolvedValue({})
await storeTcTokensFromIqResult({
result: {
tag: 'iq',
attrs: {},
content: [
{
tag: 'tokens',
attrs: {},
content: [
{
tag: 'token',
attrs: { jid: 'd@s.whatsapp.net', type: 'trusted_contact' },
content: Buffer.from([0x01])
}
]
}
]
},
fallbackJid: PEER_PN,
keys,
getLIDForPN
})
expect(keys.set).not.toHaveBeenCalled()
})
})
+19 -9
View File
@@ -29,7 +29,8 @@ const TimeMs = {
jest.mock('../../Utils/prometheus-metrics.js', () => ({
metrics: {
socketEvents: { inc: jest.fn() },
errors: { inc: jest.fn() }
errors: { inc: jest.fn() },
circuitBreakerTrips: { inc: jest.fn() }
}
}))
@@ -58,6 +59,7 @@ describe('UnifiedSessionManager', () => {
manager = createUnifiedSessionManager({
enabled: true,
logger: mockLogger as any,
enableCircuitBreaker: false, // Disable for simpler testing
sendNode: mockSendNode
})
})
@@ -354,11 +356,7 @@ describe('shouldEnableUnifiedSession', () => {
})
})
describe('UnifiedSessionManager — non-critical send semantics', () => {
// The previous "with CircuitBreaker" suite is gone — telemetry no longer
// uses a circuit breaker. Verify equivalent behavior still holds:
// failures are swallowed (telemetry is non-critical) and don't propagate.
describe('UnifiedSessionManager with CircuitBreaker', () => {
let manager: UnifiedSessionManager
let mockSendNode: MockSendNode
@@ -368,6 +366,7 @@ describe('UnifiedSessionManager — non-critical send semantics', () => {
manager = createUnifiedSessionManager({
enabled: true,
logger: createMockLogger() as any,
enableCircuitBreaker: true,
sendNode: mockSendNode
})
})
@@ -376,13 +375,24 @@ describe('UnifiedSessionManager — non-critical send semantics', () => {
manager.destroy()
})
it('should send telemetry once per call', async () => {
it('should work with circuit breaker enabled', async () => {
await manager.send('login')
expect(mockSendNode).toHaveBeenCalledTimes(1)
})
it('should swallow send failures gracefully (telemetry is non-critical)', async () => {
mockSendNode.mockRejectedValueOnce(new Error('boom'))
it('should handle circuit breaker failures gracefully', async () => {
// Make send fail multiple times to trigger circuit breaker
let callCount = 0
mockSendNode.mockImplementation(async () => {
callCount++
if (callCount <= 3) {
throw new Error(`Fail ${callCount}`)
}
})
// These should not throw even with failures
await expect(manager.send('login')).resolves.toBeUndefined()
await expect(manager.send('login')).resolves.toBeUndefined()
await expect(manager.send('login')).resolves.toBeUndefined()
})
})
-209
View File
@@ -1,209 +0,0 @@
import Long from 'long'
import $protobuf from 'protobufjs/minimal.js'
const $util = $protobuf.util
// proto implementation
function longToStringOld(value: any, unsigned?: boolean): string {
if (typeof value === 'string') return value
if (typeof value === 'number') return String(value)
if (!$util.Long) return String(value)
const normalized = ($util.Long as any).fromValue(value)
const prepared =
unsigned && normalized && typeof normalized.toUnsigned === 'function' ? normalized.toUnsigned() : normalized
return prepared.toString()
}
// bigint implementation
function longToStringNew(value: any, unsigned?: boolean): string {
if (typeof value === 'string') return value
if (typeof value === 'number') return String(value)
if (value && typeof value.low === 'number' && typeof value.high === 'number') {
const high = value.high | 0
const lo = BigInt(value.low >>> 0)
const hi = BigInt(value.high >>> 0)
const combined = (hi << 32n) | lo
if (!unsigned && high < 0) {
return (combined - (1n << 64n)).toString()
}
return combined.toString()
}
return String(value)
}
function longToNumberOld(value: any, unsigned?: boolean): number {
if (typeof value === 'number') return value
if (typeof value === 'string') return Number(value)
if (!$util.Long) return Number(value)
const normalized = ($util.Long as any).fromValue(value)
const prepared =
unsigned && normalized && typeof normalized.toUnsigned === 'function'
? normalized.toUnsigned()
: typeof normalized.toSigned === 'function'
? normalized.toSigned()
: normalized
return prepared.toNumber()
}
function longToNumberNew(value: any, unsigned?: boolean): number {
if (typeof value === 'number') return value
if (typeof value === 'string') return Number(value)
if (value && typeof value.low === 'number' && typeof value.high === 'number') {
const high = value.high | 0
const lo = BigInt(value.low >>> 0)
const hi = BigInt(value.high >>> 0)
const combined = (hi << 32n) | lo
if (!unsigned && high < 0) {
return Number(combined - (1n << 64n))
}
return Number(combined)
}
return Number(value)
}
describe('BigInt vs Long equivalence validation', () => {
// Test cases: [description, Long value, unsigned flag]
const testCases: [string, Long, boolean][] = [
// Basic values
['zero unsigned', Long.fromNumber(0, true), true],
['zero signed', Long.fromNumber(0, false), false],
['one unsigned', Long.fromNumber(1, true), true],
['one signed', Long.fromNumber(1, false), false],
// Typical WhatsApp timestamps (seconds since epoch)
['timestamp 2023', Long.fromNumber(1700000000, true), true],
['timestamp 2025', Long.fromNumber(1750000000, true), true],
['timestamp as signed', Long.fromNumber(1700000000, false), false],
// File sizes
['small file', Long.fromNumber(1024, true), true],
['medium file 10MB', Long.fromNumber(10485760, true), true],
['large file 2GB', Long.fromNumber(2147483648, true), true],
// Boundary values - 32-bit
['max int32', Long.fromNumber(2147483647, false), false],
['min int32', Long.fromNumber(-2147483648, false), false],
['max uint32', Long.fromNumber(4294967295, true), true],
// Around MAX_SAFE_INTEGER
['MAX_SAFE_INTEGER', Long.fromString('9007199254740991', false), false],
['MAX_SAFE_INTEGER + 1', Long.fromString('9007199254740992', false), false],
['MAX_SAFE_INTEGER unsigned', Long.fromString('9007199254740991', true), true],
// Large unsigned values
['large unsigned', Long.fromString('9999999999999999999', true), true],
['max uint64', Long.fromString('18446744073709551615', true), true],
['max uint64 - 1', Long.fromString('18446744073709551614', true), true],
// Signed negative values
['negative one', Long.fromNumber(-1, false), false],
['negative small', Long.fromNumber(-100, false), false],
['negative large', Long.fromNumber(-2147483648, false), false],
['min int64', Long.fromString('-9223372036854775808', false), false],
['max int64', Long.fromString('9223372036854775807', false), false],
['-1 as signed Long', Long.fromBits(-1, -1, false), false],
// Tricky bit patterns
['high=1, low=0', Long.fromBits(0, 1, true), true],
['high=0, low=-1 (0xFFFFFFFF)', Long.fromBits(-1, 0, true), true],
['high=-1, low=-1 unsigned (max uint64)', Long.fromBits(-1, -1, true), true],
['high=0x7FFFFFFF, low=0xFFFFFFFF (max int64)', Long.fromBits(-1, 0x7fffffff, false), false],
['high=0x80000000, low=0 (min int64)', Long.fromBits(0, -2147483648, false), false],
// Powers of 2
['2^32', Long.fromString('4294967296', true), true],
['2^32 signed', Long.fromString('4294967296', false), false],
['2^48', Long.fromString('281474976710656', true), true],
['2^63', Long.fromString('9223372036854775808', true), true]
]
describe('longToString equivalence', () => {
for (const [desc, longVal, unsigned] of testCases) {
it(`${desc}: Long(${longVal.low}, ${longVal.high}, ${unsigned})`, () => {
const oldResult = longToStringOld(longVal, unsigned)
const newResult = longToStringNew(longVal, unsigned)
expect(newResult).toBe(oldResult)
})
}
})
describe('longToNumber equivalence', () => {
for (const [desc, longVal, unsigned] of testCases) {
it(`${desc}: Long(${longVal.low}, ${longVal.high}, ${unsigned})`, () => {
const oldResult = longToNumberOld(longVal, unsigned)
const newResult = longToNumberNew(longVal, unsigned)
// For large values, both may lose precision equally (Number limitation)
// So we compare string representations
expect(newResult.toString()).toBe(oldResult.toString())
})
}
})
describe('non-Long inputs (fallback paths)', () => {
it('string passthrough', () => {
expect(longToStringNew('12345')).toBe('12345')
expect(longToStringNew('0')).toBe('0')
})
it('number passthrough', () => {
expect(longToStringNew(42)).toBe('42')
expect(longToStringNew(0)).toBe('0')
expect(longToStringNew(-1)).toBe('-1')
})
it('null/undefined fallback', () => {
expect(longToStringNew(null)).toBe('null')
expect(longToStringNew(undefined)).toBe('undefined')
})
it('object without low/high', () => {
expect(longToStringNew({ foo: 'bar' })).toBe('[object Object]')
})
// Defensive: a raw {low, high} JSON object can carry `high` as an
// unsigned 32-bit value (Long.fromBits would normalize via `| 0`,
// but plain JSON deserialization does not). Without `value.high | 0`
// the sign-bit check fails and the result is interpreted as unsigned.
// Upstream Baileys PR #2333 has this latent bug; InfiniteAPI fixes it.
it('signed sign-bit detection on raw unsigned high (-1 as {low: -1, high: 4294967295})', () => {
const raw = { low: -1, high: 0xffffffff }
// signed: -1
expect(longToStringNew(raw, false)).toBe('-1')
// unsigned: max uint64
expect(longToStringNew(raw, true)).toBe('18446744073709551615')
})
it('signed sign-bit detection on raw unsigned high (-4294967296 as {low: 0, high: 4294967295})', () => {
const raw = { low: 0, high: 0xffffffff }
// signed: -4294967296 (high word = -1 after normalization)
expect(longToStringNew(raw, false)).toBe('-4294967296')
// unsigned: 0xFFFFFFFF00000000 = 18446744069414584320
expect(longToStringNew(raw, true)).toBe('18446744069414584320')
})
})
describe('fuzz: random Long values', () => {
it('100 random unsigned values match', () => {
for (let i = 0; i < 100; i++) {
const low = (Math.random() * 0xffffffff) | 0
const high = (Math.random() * 0xffffffff) | 0
const longVal = Long.fromBits(low, high, true)
const oldResult = longToStringOld(longVal, true)
const newResult = longToStringNew(longVal, true)
expect(newResult).toBe(oldResult)
}
})
it('100 random signed values match', () => {
for (let i = 0; i < 100; i++) {
const low = (Math.random() * 0xffffffff) | 0
const high = (Math.random() * 0xffffffff) | 0
const longVal = Long.fromBits(low, high, false)
const oldResult = longToStringOld(longVal, false)
const newResult = longToStringNew(longVal, false)
expect(newResult).toBe(oldResult)
}
})
})
})
@@ -1,146 +0,0 @@
import { jest } from '@jest/globals'
import { proto, type WAMessage } from '../..'
import { DEFAULT_CONNECTION_CONFIG } from '../../Defaults'
import makeWASocket from '../../Socket'
import { makeSession, mockWebSocket } from '../TestUtils/session'
mockWebSocket()
// chats.ts treats a connection as a reconnection when EITHER signal is true:
// 1. authState.creds.accountSyncCounter > 0
// (at least one full history sync completed in a previous session)
// 2. socketSkippedOfflineBuffer (forwarded from socket.ts as `skipOfflineBuffer`)
// (socket.ts already decided to skip the offline buffer, e.g. because
// routingInfo was stale and was discarded on startup)
// On reconnection, AwaitingInitialSync is skipped and the buffer is flushed
// immediately so live messages are not held in the initial-sync window.
describe('Reconnection Sync Skip', () => {
it('should skip the history sync wait on reconnection (accountSyncCounter > 0)', async () => {
const { state, clear } = await makeSession()
state.creds.me = { id: '1234567890:1@s.whatsapp.net', name: 'Test User' }
// Simulate a session that has already synced before
state.creds.accountSyncCounter = 1
const sock = makeWASocket({
...DEFAULT_CONNECTION_CONFIG,
auth: state
})
const messageListener = jest.fn()
sock.ev.on('messages.upsert', messageListener)
// Simulate receiving pending notifications (triggers AwaitingInitialSync)
sock.ev.emit('connection.update', { receivedPendingNotifications: true })
// Emit a message immediately after — if the wait is skipped,
// the buffer should already be flushed and this message should be delivered.
const msg = proto.WebMessageInfo.fromObject({
key: { remoteJid: '1234567890@s.whatsapp.net', fromMe: false, id: 'MSG_AFTER_RECONNECT' },
messageTimestamp: Date.now() / 1000,
message: { conversation: 'Hello after reconnect' }
}) as WAMessage
sock.ev.emit('messages.upsert', { messages: [msg], type: 'notify' })
await new Promise(resolve => setTimeout(resolve, 50))
// Message should be delivered immediately, NOT buffered
expect(messageListener).toHaveBeenCalledTimes(1)
expect(messageListener).toHaveBeenCalledWith(
expect.objectContaining({
messages: expect.arrayContaining([expect.objectContaining({ key: msg.key })]),
type: 'notify'
})
)
await sock.end(new Error('Test completed'))
await clear()
})
it('should skip the history sync wait when socketSkippedOfflineBuffer is true (stale routingInfo, accountSyncCounter === 0)', async () => {
const { state, clear } = await makeSession()
state.creds.me = { id: '1234567890:1@s.whatsapp.net', name: 'Test User' }
// Fresh-looking session from the counter perspective
state.creds.accountSyncCounter = 0
// But routingInfo is present and we ask the socket to discard it on start.
// socket.ts will set hadStaleRoutingInfo=true → skipOfflineBuffer=true,
// which chats.ts forwards as socketSkippedOfflineBuffer. Without this
// branch, the buffers would be misaligned (offline buffer skipped while
// AwaitingInitialSync still waits) and live messages would stall.
state.creds.routingInfo = Buffer.from([1, 2, 3, 4])
const sock = makeWASocket({
...DEFAULT_CONNECTION_CONFIG,
auth: state,
clearRoutingInfoOnStart: true
})
const messageListener = jest.fn()
sock.ev.on('messages.upsert', messageListener)
sock.ev.emit('connection.update', { receivedPendingNotifications: true })
const msg = proto.WebMessageInfo.fromObject({
key: { remoteJid: '1234567890@s.whatsapp.net', fromMe: false, id: 'MSG_AFTER_STALE_ROUTING' },
messageTimestamp: Date.now() / 1000,
message: { conversation: 'Hello after stale routing reconnect' }
}) as WAMessage
sock.ev.emit('messages.upsert', { messages: [msg], type: 'notify' })
await new Promise(resolve => setTimeout(resolve, 50))
expect(messageListener).toHaveBeenCalledTimes(1)
expect(messageListener).toHaveBeenCalledWith(
expect.objectContaining({
messages: expect.arrayContaining([expect.objectContaining({ key: msg.key })]),
type: 'notify'
})
)
await sock.end(new Error('Test completed'))
await clear()
})
it('should still wait for history sync on fresh pairing (both signals false)', async () => {
const { state, clear } = await makeSession()
state.creds.me = { id: '1234567890:1@s.whatsapp.net', name: 'Test User' }
// Fresh pairing — both reconnect signals are false:
// accountSyncCounter is 0 (default) and no stale routingInfo to clear.
state.creds.accountSyncCounter = 0
const sock = makeWASocket({
...DEFAULT_CONNECTION_CONFIG,
auth: state
})
const messageListener = jest.fn()
sock.ev.on('messages.upsert', messageListener)
sock.ev.emit('connection.update', { receivedPendingNotifications: true })
const msg = proto.WebMessageInfo.fromObject({
key: { remoteJid: '1234567890@s.whatsapp.net', fromMe: false, id: 'MSG_DURING_INITIAL_SYNC' },
messageTimestamp: Date.now() / 1000,
message: { conversation: 'Hello during initial sync' }
}) as WAMessage
sock.ev.emit('messages.upsert', { messages: [msg], type: 'notify' })
await new Promise(resolve => setTimeout(resolve, 50))
// Message should be BUFFERED (not delivered) because we're waiting for history sync
expect(messageListener).toHaveBeenCalledTimes(0)
// Drain the 2s AwaitingInitialSync timeout (chats.ts:1522) so its callback
// doesn't fire after Jest considers the test done — otherwise we get a
// "Cannot log after tests are done" warning from the post-teardown flush
// and the suite is flagged with --detectOpenHandles. After the timer
// fires, syncState becomes Online and the buffer flushes; that delivery
// is post-assertion and irrelevant to the test goal.
await new Promise(resolve => setTimeout(resolve, 2_100))
await sock.end(new Error('Test completed'))
await clear()
}, 10_000)
})
-88
View File
@@ -1,4 +1,3 @@
import Long from 'long'
import '../index.js'
import { proto } from '../../WAProto/index.js'
@@ -29,91 +28,4 @@ describe('proto serialization', () => {
const json = message.toJSON()
expect(json.message?.imageMessage?.fileLength).toBe('1234567890123456789')
})
it('converts Long objects to strings correctly via BigInt fast path', () => {
const message = proto.WebMessageInfo.fromObject({
key: {
remoteJid: '123@s.whatsapp.net',
id: 'ABC123',
fromMe: false
},
messageTimestamp: Long.fromNumber(1700000000, true),
message: {
imageMessage: {
fileLength: Long.fromString('9876543210', true)
}
}
})
const json = message.toJSON()
expect(json.messageTimestamp).toBe('1700000000')
expect(json.message?.imageMessage?.fileLength).toBe('9876543210')
})
it('handles large unsigned Long values (> MAX_SAFE_INTEGER)', () => {
const message = proto.WebMessageInfo.fromObject({
key: {
remoteJid: '123@s.whatsapp.net',
id: 'ABC123',
fromMe: false
},
messageTimestamp: Long.fromString('18446744073709551615', true), // max uint64
message: {
imageMessage: {
fileLength: Long.fromString('9999999999999999999', true)
}
}
})
const json = message.toJSON()
expect(json.messageTimestamp).toBe('18446744073709551615')
expect(json.message?.imageMessage?.fileLength).toBe('9999999999999999999')
})
it('handles zero and small Long values', () => {
const message = proto.WebMessageInfo.fromObject({
key: {
remoteJid: '123@s.whatsapp.net',
id: 'ABC123',
fromMe: false
},
messageTimestamp: Long.fromNumber(0, true),
message: {
imageMessage: {
fileLength: Long.fromNumber(1, true)
}
}
})
const json = message.toJSON()
expect(json.messageTimestamp).toBe('0')
expect(json.message?.imageMessage?.fileLength).toBe('1')
})
it('roundtrips encode/decode with Long fields preserving values (> MAX_SAFE_INTEGER)', () => {
// Use uint64 values above Number.MAX_SAFE_INTEGER (2^53 - 1) so the
// roundtrip actually exercises the BigInt fast path on decode->toJSON.
// Safe integers would silently work even if longToString fell back to
// Number(value).
const original = proto.WebMessageInfo.fromObject({
key: {
remoteJid: '123@s.whatsapp.net',
id: 'ABC123',
fromMe: false
},
messageTimestamp: Long.fromString('18446744073709551615', true),
message: {
imageMessage: {
fileLength: Long.fromString('9999999999999999999', true)
}
}
})
const encoded = proto.WebMessageInfo.encode(original).finish()
const decoded = proto.WebMessageInfo.decode(encoded)
const json = decoded.toJSON()
expect(json.messageTimestamp).toBe('18446744073709551615')
expect(json.message?.imageMessage?.fileLength).toBe('9999999999999999999')
})
})
+2 -11
View File
@@ -28,16 +28,7 @@ console.info = function (...args: unknown[]) {
// Track errors by type + JID to avoid duplicates (using Map for better performance)
const _errorTimestamps = new Map<string, number>()
// Dedup window for repeated decrypt-error console lines (Bad MAC / Counter / etc).
// Was 150ms, but retry attempts of the SAME message are typically ~300-1000ms apart,
// so the second attempt fell outside the window and double-printed.
//
// TRADE-OFF: dedup key is `errorType + JID` (no message-id). With 5s, a burst of
// errors for the SAME JID — even of slightly different categories or different
// messages — collapses to one log line every 5s. This is intentional for a noisy
// production stream; if you need per-message visibility, set BAILEYS_LOG_LEVEL=debug
// to bypass this console-side dedup and see the structured pino logs in full.
const DEDUP_WINDOW_MS = 5000
const DEDUP_WINDOW_MS = 150
console.error = function (...args: unknown[]) {
if (args.length > 0 && typeof args[0] === 'string') {
@@ -79,7 +70,7 @@ console.error = function (...args: unknown[]) {
const lastTime = _errorTimestamps.get(dedupeKey)
if (lastTime && now - lastTime < DEDUP_WINDOW_MS) {
return // Skip duplicate within DEDUP_WINDOW_MS window
return // Skip duplicate within 150ms window
}
_errorTimestamps.set(dedupeKey, now)