test(e2e): test harness + signal/prekey fixes uncovered along the way

.github/workflows/e2e.yml

@@ -0,0 +1,74 @@
+name: E2E Tests
+
+on:
+  push:
+    branches:
+      - master
+  pull_request_target:
+    branches:
+      - master
+
+permissions:
+  contents: read
+
+jobs:
+  e2e:
+    name: E2E Tests
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    services:
+      mock-server:
+        image: ghcr.io/whiskeysockets-devtools/bartender:latest
+        credentials:
+          username: ${{ github.actor }}
+          password: ${{ secrets.BARTENDER_GHCR_TOKEN }}
+        ports:
+          - 8080:8080
+        env:
+          CHATSTATE_TTL_SECS: "3"
+          ADV_SECRET_KEY: "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
+        options: --log-driver none
+
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          ref: ${{ github.event.pull_request.head.sha || github.sha }}
+
+      - name: Setup Node.js and Corepack
+        uses: actions/setup-node@v4
+        with:
+          node-version: 20.x
+
+      - name: Enable Corepack and Set Yarn Version
+        run: |
+          corepack enable
+          corepack prepare yarn@4.x --activate
+
+      - name: Restore Yarn Cache
+        uses: actions/cache@v4
+        with:
+          path: .yarn/cache
+          key: ${{ runner.os }}-yarn-${{ hashFiles('**/yarn.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-yarn-
+
+      - name: Install dependencies
+        run: yarn install --immutable
+
+      - name: Wait for mock server
+        run: |
+          for i in $(seq 1 30); do
+            if curl -sk https://localhost:8080/ > /dev/null 2>&1; then
+              echo "Mock server is ready"
+              exit 0
+            fi
+            sleep 1
+          done
+          echo "Mock server failed to become ready"
+          exit 1
+
+      - name: Run E2E tests
+        env:
+          SOCKET_URL: "wss://127.0.0.1:8080/ws/chat"
+          ADV_SECRET_KEY: "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
+        run: yarn test:e2e

package.json

@@ -37,7 +37,7 @@
     "preinstall": "node ./engine-requirements.js",
     "release": "release-it",
     "test": "node --experimental-vm-modules ./node_modules/.bin/jest --testMatch '**/*.test.ts'",
-    "test:e2e": "node --experimental-vm-modules ./node_modules/.bin/jest --testMatch '**/*.test-e2e.ts'",
+    "test:e2e": "NODE_TLS_REJECT_UNAUTHORIZED=0 node --experimental-vm-modules ./node_modules/.bin/jest --runInBand --testMatch '**/*.test-e2e.ts'",
     "update:version": "tsx ./scripts/update-version.ts"
   },
   "dependencies": {

src/Defaults/index.ts

@@ -135,7 +135,6 @@ export const MIN_PREKEY_COUNT = 5
 export const INITIAL_PREKEY_COUNT = 812
 
 export const UPLOAD_TIMEOUT = 30000 // 30 seconds
-export const MIN_UPLOAD_INTERVAL = 5000 // 5 seconds minimum between uploads
 
 export const DEFAULT_CACHE_TTLS = {
 	SIGNAL_STORE: 5 * 60, // 5 minutes

src/Socket/chats.ts

@@ -52,12 +52,12 @@
 	type BinaryNode,
 	getBinaryNodeChild,
 	getBinaryNodeChildren,
+	isHostedLidUser,
+	isHostedPnUser,
 	isLidUser,
 	isPnUser,
 	jidDecode,
 	jidNormalizedUser,
-	isHostedLidUser,
-	isHostedPnUser,
 	reduceBinaryNodeToDictionary,
 	S_WHATSAPP_NET
 } from '../WABinary'
@@ -683,7 +683,7 @@
 								// collection is done with sync
 								collectionsToHandle.delete(name)
 							}
 						} catch (error: any) {
 							attemptsMap[name] = (attemptsMap[name] || 0) + 1
 
 							const logData = {

src/Socket/messages-recv.ts

@@ -213,7 +213,7 @@
 			return
 		}
 
 		let data: any
 		try {
 			data = JSON.parse(mexNode.content.toString())
 		} catch (error) {
@@ -309,7 +309,7 @@
 			case 'update':
 				const settingsNode = getBinaryNodeChild(child, 'settings')
 				if (settingsNode) {
 					const update: Record<string, any> = {}
 					const nameNode = getBinaryNodeChild(settingsNode, 'name')
 					if (nameNode?.content) update.name = nameNode.content.toString()
 
@@ -534,6 +534,9 @@
 		}, authState?.creds?.me?.id || 'sendRetryRequest')
 	}
 
+	// Mirrors WAWeb/Handle/PreKeyLow.js: skip a re-issued notification with the same stanza id.
+	const inFlightPreKeyLow = new Set<string>()
+
 	/**
 	 * Fire-and-forget tctoken re-issuance after a peer's device identity changed.
 	 * Mirrors WAWebSendTcTokenWhenDeviceIdentityChange — runs in parallel with
@@ -574,13 +577,23 @@
 	const handleEncryptNotification = async (node: BinaryNode) => {
 		const from = node.attrs.from
 		if (from === S_WHATSAPP_NET) {
+			const stanzaId = node.attrs.id
+			if (stanzaId && inFlightPreKeyLow.has(stanzaId)) {
+				return
+			}
+
 			const countChild = getBinaryNodeChild(node, 'count')
 			const count = +countChild!.attrs.value!
 			const shouldUploadMorePreKeys = count < MIN_PREKEY_COUNT
 
 			logger.debug({ count, shouldUploadMorePreKeys }, 'recv pre-key count')
 			if (shouldUploadMorePreKeys) {
-				await uploadPreKeys()
+				if (stanzaId) inFlightPreKeyLow.add(stanzaId)
+				try {
+					await uploadPreKeys()
+				} finally {
+					if (stanzaId) inFlightPreKeyLow.delete(stanzaId)
+				}
 			}
 		} else {
 			const result = await handleIdentityChange(node, {
@@ -929,7 +942,7 @@
 			const jids = await readTcTokenIndex(authState.keys)
 			for (const jid of jids) tcTokenKnownJids.add(jid)
 			logger.debug({ count: tcTokenKnownJids.size }, 'loaded tctoken index')
 		} catch (err: any) {
 			logger.warn({ err: err?.message }, 'failed to load tctoken index')
 		}
 	})()
@@ -1365,47 +1378,23 @@
 							}
 						}
 
-						const errorMessage = msg?.messageStubParameters?.[0] || ''
-						const isPreKeyError = errorMessage.includes('PreKey')
-
-						logger.debug(`[handleMessage] Attempting retry request for failed decryption`)
+						logger.debug('[handleMessage] Attempting retry request for failed decryption')
 
-						// Handle both pre-key and normal retries in single mutex
+						// WAWeb only retry-receipts here; server emits PreKeyLow if prekeys run low.
 						await retryMutex.mutex(async () => {
 							try {
 								if (!ws.isOpen) {
 									logger.debug({ node }, 'Connection closed, skipping retry')
 									return
 								}
 
-								// Handle pre-key errors with upload and delay
-								if (isPreKeyError) {
-									logger.info({ error: errorMessage }, 'PreKey error detected, uploading and retrying')
-
-									try {
-										logger.debug('Uploading pre-keys for error recovery')
-										await uploadPreKeys(5)
-										logger.debug('Waiting for server to process new pre-keys')
-										await delay(1000)
-									} catch (uploadErr) {
-										logger.error({ uploadErr }, 'Pre-key upload failed, proceeding with retry anyway')
-									}
-								}
-
 								const encNode = getBinaryNodeChild(node, 'enc')
 								await sendRetryRequest(node, !encNode)
 								if (retryRequestDelayMs) {
 									await delay(retryRequestDelayMs)
 								}
 							} catch (err) {
-								logger.error({ err, isPreKeyError }, 'Failed to handle retry, attempting basic retry')
-								// Still attempt retry even if pre-key upload failed
-								try {
-									const encNode = getBinaryNodeChild(node, 'enc')
-									await sendRetryRequest(node, !encNode)
-								} catch (retryErr) {
-									logger.error({ retryErr }, 'Failed to send retry after error handling')
-								}
+								logger.error({ err }, 'Failed to send retry')
 							}
 
 							acked = true
@@ -1490,14 +1479,14 @@
 				status
 			}
 
-      if (status === 'relaylatency') {
-        const latencyValue = infoChild.attrs.latency || infoChild.attrs['latency_ms'] || infoChild.attrs['latency-ms']
-        const latencyMs = latencyValue ? Number(latencyValue) : undefined
-        if (Number.isFinite(latencyMs)) {
-          call.latencyMs = latencyMs
-        }
-      }
-      
+			if (status === 'relaylatency') {
+				const latencyValue = infoChild.attrs.latency || infoChild.attrs['latency_ms'] || infoChild.attrs['latency-ms']
+				const latencyMs = latencyValue ? Number(latencyValue) : undefined
+				if (Number.isFinite(latencyMs)) {
+					call.latencyMs = latencyMs
+				}
+			}
+
 			if (status === 'offer') {
 				call.isVideo = !!getBinaryNodeChild(infoChild, 'video')
 				call.isGroup = infoChild.attrs.type === 'group' || !!infoChild.attrs['group-jid']
@@ -1625,6 +1614,7 @@
 			)
 			ignoreJid = !isNodeFromMe || isJidGroup(attrs.from) ? attrs.from : attrs.recipient
 		}
+
 		if (ignoreJid && ignoreJid !== S_WHATSAPP_NET && shouldIgnoreJid(ignoreJid)) {
 			await sendMessageAck(node, type === 'message' ? NACK_REASONS.UnhandledError : undefined)
 			return
@@ -1786,7 +1776,7 @@
 			for (const jid of survivors) tcTokenKnownJids.add(jid)
 
 			logger.debug({ mutated, remaining: survivors.size }, 'pruned expired tctokens')
 		} catch (err: any) {
 			logger.warn({ err: err?.message }, 'failed to prune expired tctokens')
 		}
 	}

src/Socket/messages-send.ts

@@ -25,12 +25,14 @@
 	generateMessageIDV2,
 	generateParticipantHashV2,
 	generateWAMessage,
+	getMediaHost,
 	getStatusCodeForMediaRetry,
 	getUrlFromDirectPath,
 	getWAUploadToServer,
 	MessageRetryManager,
 	normalizeMessageContent,
 	parseAndInjectE2ESessions,
+	setMediaHost,
 	unixTimestampSeconds
 } from '../Utils'
 import { getUrlInfo } from '../Utils/link-preview'
@@ -109,11 +111,6 @@
 			useClones: false
 		})
 
-	const peerSessionsCache = new NodeCache<boolean>({
-		stdTTL: DEFAULT_CACHE_TTLS.USER_DEVICES,
-		useClones: false
-	})
-
 	// Initialize message retry manager if enabled
 	const messageRetryManager = enableRecentMessageCache ? new MessageRetryManager(logger, maxMsgRetryCount) : null
 
@@ -146,6 +143,10 @@
 					fetchDate: new Date()
 				}
 				logger.debug('fetched media conn')
+				if (node.hosts[0]) {
+					setMediaHost(node.hosts[0].hostname)
+				}
+
 				return node
 			})()
 		}
@@ -436,24 +437,15 @@
 
 	const assertSessions = async (jids: string[], force?: boolean) => {
 		let didFetchNewSession = false
-		const uniqueJids = [...new Set(jids)] // Deduplicate JIDs
+		const uniqueJids = [...new Set(jids)]
 		const jidsRequiringFetch: string[] = []
 
 		logger.debug({ jids }, 'assertSessions call with jids')
 
-		// Check peerSessionsCache and validate sessions using libsignal loadSession
 		for (const jid of uniqueJids) {
-			const signalId = signalRepository.jidToSignalProtocolAddress(jid)
-			const cachedSession = peerSessionsCache.get(signalId)
-			if (cachedSession !== undefined) {
-				if (cachedSession && !force) {
-					continue // Session exists in cache
-				}
-			} else {
+			if (!force) {
 				const sessionValidation = await signalRepository.validateSession(jid)
-				const hasSession = sessionValidation.exists
-				peerSessionsCache.set(signalId, hasSession)
-				if (hasSession && !force) {
+				if (sessionValidation.exists) {
 					continue
 				}
 			}
@@ -494,12 +486,6 @@
 			})
 			await parseAndInjectE2ESessions(result, signalRepository)
 			didFetchNewSession = true
-
-			// Cache fetched sessions using wire JIDs
-			for (const wireJid of wireJids) {
-				const signalId = signalRepository.jidToSignalProtocolAddress(wireJid)
-				peerSessionsCache.set(signalId, true)
-			}
 		}
 
 		return didFetchNewSession
@@ -559,8 +545,8 @@
 		const meLid = authState.creds.me?.lid
 		const meLidUser = meLid ? jidDecode(meLid)?.user : null
 
 		const encryptionPromises = (patchedMessages as any).map(
 			async ({ recipientJid: jid, message: patchedMessage }: any) => {
 				try {
 					if (!jid) return null
 
@@ -1028,7 +1014,7 @@
 						;(stanza.content as BinaryNode[]).push(reportingNode)
 						logger.trace({ jid }, 'added reporting token to message')
 					}
 				} catch (error: any) {
 					logger.warn({ jid, trace: error?.stack }, 'failed to attach reporting token')
 				}
 			}
@@ -1054,7 +1040,7 @@
 						: null
 				try {
 					await authState.keys.set({ tctoken: { [tcTokenJid]: cleared } })
 				} catch (err: any) {
 					logger.debug({ jid: destinationJid, err: err?.message }, 'failed to persist tctoken expiry cleanup')
 				}
 			}
@@ -1235,6 +1221,9 @@
 		sendReceipts,
 		readMessages,
 		refreshMediaConn,
+		get mediaHost() {
+			return getMediaHost()
+		},
 		waUploadToServer,
 		fetchPrivacySettings,
 		sendPeerDataOperationMessage,
@@ -1271,7 +1260,7 @@
 								content.url = getUrlFromDirectPath(content.directPath!)
 
 								logger.debug({ directPath: media.directPath, key: result.key }, 'media update successful')
 							} catch (err: any) {
 								error = err
 							}
 						}

src/Socket/socket.ts

@@ -8,7 +8,6 @@ import {
 	DEF_TAG_PREFIX,
 	INITIAL_PREKEY_COUNT,
 	MIN_PREKEY_COUNT,
-	MIN_UPLOAD_INTERVAL,
 	NOISE_WA_HEADER,
 	PROCESSABLE_HISTORY_TYPES,
 	TimeMs,
@@ -468,22 +467,11 @@ export const makeSocket = (config: SocketConfig) => {
 		return +countChild.attrs.value!
 	}
 
-	// Pre-key upload state management
+	// WAWeb has no time throttle here; the server drives uploads via PreKeyLow notifications.
 	let uploadPreKeysPromise: Promise<void> | null = null
-	let lastUploadTime = 0
 
 	/** generates and uploads a set of pre-keys to the server */
 	const uploadPreKeys = async (count = MIN_PREKEY_COUNT, retryCount = 0) => {
-		// 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 (uploadPreKeysPromise) {
 			logger.debug('Pre-key upload already in progress, waiting for completion')
 			await uploadPreKeysPromise
@@ -505,7 +493,6 @@ export const makeSocket = (config: SocketConfig) => {
 			try {
 				await query(node)
 				logger.info({ count }, 'uploaded pre-keys successfully')
-				lastUploadTime = Date.now()
 			} catch (uploadError) {
 				logger.error({ uploadError: (uploadError as Error).toString(), count }, 'Failed to upload pre-keys to server')
 

src/Types/Socket.ts

@@ -50,6 +50,8 @@ export type SocketConfig = {
 	version: WAVersion
 	/** override browser config */
 	browser: WABrowserDescription
+	/** Initial pushName carried in the registration ClientPayload (used by mock servers for deterministic phone assignment). */
+	pushName?: string
 	/** agent used for fetch requests -- uploading/downloading media */
 	fetchAgent?: Agent
 	/** should the QR be printed in the terminal

src/Utils/messages-media.ts

@@ -500,7 +500,18 @@ export const encryptedStream = async (
 	}
 }
 
-const DEF_HOST = 'mmg.whatsapp.net'
+export const DEF_MEDIA_HOST = 'mmg.whatsapp.net'
+
+let _mediaHost = DEF_MEDIA_HOST
+
+/** Set the media host used for downloading media. Called internally by the socket when media_conn is fetched. */
+export const setMediaHost = (host: string) => {
+	_mediaHost = host
+}
+
+/** Get the current media host used for downloading media. */
+export const getMediaHost = () => _mediaHost
+
 const AES_CHUNK_SIZE = 16
 
 const toSmallestChunkSize = (num: number) => {
@@ -513,15 +524,14 @@ export type MediaDownloadOptions = {
 	options?: RequestInit
 }
 
-export const getUrlFromDirectPath = (directPath: string) => `https://${DEF_HOST}${directPath}`
+export const getUrlFromDirectPath = (directPath: string, host: string = _mediaHost) => `https://${host}${directPath}`
 
 export const downloadContentFromMessage = async (
 	{ mediaKey, directPath, url }: DownloadableMessage,
 	type: MediaType,
 	opts: MediaDownloadOptions = {}
 ) => {
-	const isValidMediaUrl = url?.startsWith('https://mmg.whatsapp.net/')
-	const downloadUrl = isValidMediaUrl ? url : getUrlFromDirectPath(directPath!)
+	const downloadUrl = directPath ? getUrlFromDirectPath(directPath) : url
 	if (!downloadUrl) {
 		throw new Boom('No valid media URL or directPath present in message', { statusCode: 400 })
 	}