[READY FOR MERGE] Implement newsletter (#1532)

* feat: implement basic newsletter functionality with socket integration and event handling

* feat: enhance media handling for newsletters with raw media upload support

* feat: working updatePicture, removePicure, adminCount, mute, Unmute

* fix: fetchMessages

* chore: cleanup

* fix: update newsletter metadata path and query ID for consistency. newsletterMetadata works now

* chore: enhance newsletter metadata parsing and error handling

* fix: correct DELETE QueryId value in Newsletter.ts

* chore: split mex stuffs to own file

* chore: remove as any

src/Utils/decode-wa-message.ts

@@ -114,7 +114,8 @@ export function decodeMessageNode(stanza: BinaryNode, meId: string, meLid: strin
 		senderPn: stanza?.attrs?.sender_pn,
 		participant,
 		participantPn: stanza?.attrs?.participant_pn,
-		participantLid: stanza?.attrs?.participant_lid
+		participantLid: stanza?.attrs?.participant_lid,
+		...(msgType === 'newsletter' && stanza.attrs.server_id ? { server_id: stanza.attrs.server_id } : {})
 	}
 
 	const fullMessage: proto.IWebMessageInfo = {

src/Utils/generics.ts

@@ -443,3 +443,7 @@ export function bytesToCrockford(buffer: Buffer): string {
 
 	return crockford.join('')
 }
+
+export function encodeNewsletterMessage(message: proto.IMessage): Uint8Array {
+	return proto.Message.encode(message).finish()
+}

src/Utils/messages-media.ts

@@ -60,6 +60,47 @@ export const hkdfInfoKey = (type: MediaType) => {
 	return `WhatsApp ${hkdfInfo} Keys`
 }
 
+export const getRawMediaUploadData = async (media: WAMediaUpload, mediaType: MediaType, logger?: ILogger) => {
+	const { stream } = await getStream(media)
+	logger?.debug('got stream for raw upload')
+
+	const hasher = Crypto.createHash('sha256')
+	const filePath = join(tmpdir(), mediaType + generateMessageIDV2())
+	const fileWriteStream = createWriteStream(filePath)
+
+	let fileLength = 0
+	try {
+		for await (const data of stream) {
+			fileLength += data.length
+			hasher.update(data)
+			if (!fileWriteStream.write(data)) {
+				await once(fileWriteStream, 'drain')
+			}
+		}
+
+		fileWriteStream.end()
+		await once(fileWriteStream, 'finish')
+		stream.destroy()
+		const fileSha256 = hasher.digest()
+		logger?.debug('hashed data for raw upload')
+		return {
+			filePath: filePath,
+			fileSha256,
+			fileLength
+		}
+	} catch (error) {
+		fileWriteStream.destroy()
+		stream.destroy()
+		try {
+			await fs.unlink(filePath)
+		} catch {
+			//
+		}
+
+		throw error
+	}
+}
+
 /** generates all the keys required to encrypt/decrypt & sign a media message */
 export async function getMediaKeys(
 	buffer: Uint8Array | string | null | undefined,
@@ -143,22 +184,24 @@ export const generateProfilePicture = async (
 	mediaUpload: WAMediaUpload,
 	dimensions?: { width: number; height: number }
 ) => {
+	let buffer: Buffer
+
 	const { width: w = 640, height: h = 640 } = dimensions || {}
 
-	let bufferOrFilePath: Buffer | string
 	if (Buffer.isBuffer(mediaUpload)) {
-		bufferOrFilePath = mediaUpload
-	} else if ('url' in mediaUpload) {
-		bufferOrFilePath = mediaUpload.url.toString()
+		buffer = mediaUpload
 	} else {
-		bufferOrFilePath = await toBuffer(mediaUpload.stream)
+		// Use getStream to handle all WAMediaUpload types (Buffer, Stream, URL)
+		const { stream } = await getStream(mediaUpload)
+		// Convert the resulting stream to a buffer
+		buffer = await toBuffer(stream)
 	}
 
 	const lib = await getImageProcessingLibrary()
 	let img: Promise<Buffer>
 	if ('sharp' in lib && typeof lib.sharp?.default === 'function') {
 		img = lib.sharp
-			.default(bufferOrFilePath)
+			.default(buffer)
 			.resize(w, h)
 			.jpeg({
 				quality: 50
@@ -166,7 +209,7 @@ export const generateProfilePicture = async (
 			.toBuffer()
 	} else if ('jimp' in lib && typeof lib.jimp?.read === 'function') {
 		const { read, MIME_JPEG, RESIZE_BILINEAR } = lib.jimp
-		const jimp = await read(bufferOrFilePath as string)
+		const jimp = await read(buffer)
 		const min = Math.min(jimp.getWidth(), jimp.getHeight())
 		const cropped = jimp.crop(0, 0, min, min)
 

src/Utils/messages.ts

@@ -9,7 +9,6 @@ import {
 	AnyMediaMessageContent,
 	AnyMessageContent,
 	DownloadableMessage,
-	MediaGenerationOptions,
 	MediaType,
 	MessageContentGenerationOptions,
 	MessageGenerationOptions,
@@ -23,7 +22,7 @@ import {
 	WAProto,
 	WATextMessage
 } from '../Types'
-import { isJidGroup, isJidStatusBroadcast, jidNormalizedUser } from '../WABinary'
+import { isJidGroup, isJidNewsletter, isJidStatusBroadcast, jidNormalizedUser } from '../WABinary'
 import { sha256 } from './crypto'
 import { generateMessageIDV2, getKeyAuthor, unixTimestampSeconds } from './generics'
 import { ILogger } from './logger'
@@ -33,6 +32,7 @@ import {
 	generateThumbnail,
 	getAudioDuration,
 	getAudioWaveform,
+	getRawMediaUploadData,
 	MediaDownloadOptions
 } from './messages-media'
 
@@ -108,7 +108,10 @@ const assertColor = async color => {
 	}
 }
 
-export const prepareWAMessageMedia = async (message: AnyMediaMessageContent, options: MediaGenerationOptions) => {
+export const prepareWAMessageMedia = async (
+	message: AnyMediaMessageContent,
+	options: MessageContentGenerationOptions
+) => {
 	const logger = options.logger
 
 	let mediaType: (typeof MEDIA_KEYS)[number] | undefined
@@ -127,13 +130,12 @@ export const prepareWAMessageMedia = async (message: AnyMediaMessageContent, opt
 		media: message[mediaType]
 	}
 	delete uploadData[mediaType]
-	// check if cacheable + generate cache key
+
 	const cacheableKey =
 		typeof uploadData.media === 'object' &&
 		'url' in uploadData.media &&
 		!!uploadData.media.url &&
 		!!options.mediaCache &&
-		// generate the key
 		mediaType + ':' + uploadData.media.url.toString()
 
 	if (mediaType === 'document' && !uploadData.fileName) {
@@ -144,7 +146,6 @@ export const prepareWAMessageMedia = async (message: AnyMediaMessageContent, opt
 		uploadData.mimetype = MIMETYPE_MAP[mediaType]
 	}
 
-	// check for cache hit
 	if (cacheableKey) {
 		const mediaBuff = options.mediaCache!.get<Buffer>(cacheableKey)
 		if (mediaBuff) {
@@ -159,6 +160,48 @@ export const prepareWAMessageMedia = async (message: AnyMediaMessageContent, opt
 		}
 	}
 
+	const isNewsletter = !!options.jid && isJidNewsletter(options.jid)
+	if (isNewsletter) {
+		logger?.info({ key: cacheableKey }, 'Preparing raw media for newsletter')
+		const { filePath, fileSha256, fileLength } = await getRawMediaUploadData(
+			uploadData.media,
+			options.mediaTypeOverride || mediaType,
+			logger
+		)
+
+		const fileSha256B64 = fileSha256.toString('base64')
+		const { mediaUrl, directPath } = await options.upload(filePath, {
+			fileEncSha256B64: fileSha256B64,
+			mediaType: mediaType,
+			timeoutMs: options.mediaUploadTimeoutMs
+		})
+
+		await fs.unlink(filePath)
+
+		const obj = WAProto.Message.fromObject({
+			[`${mediaType}Message`]: MessageTypeProto[mediaType].fromObject({
+				url: mediaUrl,
+				directPath,
+				fileSha256,
+				fileLength,
+				...uploadData,
+				media: undefined
+			})
+		})
+
+		if (uploadData.ptv) {
+			obj.ptvMessage = obj.videoMessage
+			delete obj.videoMessage
+		}
+
+		if (cacheableKey) {
+			logger?.debug({ cacheableKey }, 'set cache')
+			options.mediaCache!.set(cacheableKey, WAProto.Message.encode(obj).finish())
+		}
+
+		return obj
+	}
+
 	const requiresDurationComputation = mediaType === 'audio' && typeof uploadData.seconds === 'undefined'
 	const requiresThumbnailComputation =
 		(mediaType === 'image' || mediaType === 'video') && typeof uploadData['jpegThumbnail'] === 'undefined'
@@ -174,7 +217,7 @@ export const prepareWAMessageMedia = async (message: AnyMediaMessageContent, opt
 			opts: options.options
 		}
 	)
-	// url safe Base64 encode the SHA256 hash of the body
+
 	const fileEncSha256B64 = fileEncSha256.toString('base64')
 	const [{ mediaUrl, directPath }] = await Promise.all([
 		(async () => {
@@ -539,7 +582,7 @@ export const generateWAMessageFromContent = (
 	const timestamp = unixTimestampSeconds(options.timestamp)
 	const { quoted, userJid } = options
 
-	if (quoted) {
+	if (quoted && !isJidNewsletter(jid)) {
 		const participant = quoted.key.fromMe
 			? userJid
 			: quoted.participant || quoted.key.participant || quoted.key.remoteJid
@@ -574,7 +617,9 @@ export const generateWAMessageFromContent = (
 		// and it's not a protocol message -- delete, toggle disappear message
 		key !== 'protocolMessage' &&
 		// already not converted to disappearing message
-		key !== 'ephemeralMessage'
+		key !== 'ephemeralMessage' &&
+		// newsletters don't support ephemeral messages
+		!isJidNewsletter(jid)
 	) {
 		innerMessage[key].contextInfo = {
 			...(innerMessage[key].contextInfo || {}),
@@ -603,7 +648,8 @@ export const generateWAMessageFromContent = (
 export const generateWAMessage = async (jid: string, content: AnyMessageContent, options: MessageGenerationOptions) => {
 	// ensure msg ID is with every log
 	options.logger = options?.logger?.child({ msgId: options.messageId })
-	return generateWAMessageFromContent(jid, await generateWAMessageContent(content, options), options)
+	// Pass jid in the options to generateWAMessageContent
+	return generateWAMessageFromContent(jid, await generateWAMessageContent(content, { ...options, jid }), options)
 }
 
 /** Get the key to access the true type of content */