const LOGGED_OUT_STATUS = DisconnectReason?.loggedOut ?? 401;

const RECONNECT_IN_PROGRESS_ERROR = "no active socket - reconnection in progress";

const GROUP_META_TTL_MS = 5 * 60 * 1000; // 5 minutes

const INBOUND_CLOSE_DRAIN_TIMEOUT_MS = 5_000;

export const WHATSAPP_GROUP_METADATA_CACHE_MAX_ENTRIES = 500;

type WhatsAppGroupMetadataCacheEntry = {

const self = selfIdentity.identity;

type QueuedInboundMessage = WebInboundMessage & {

dedupeKey?: string;

debounceKey?: string;

};

const inboundDebounceMs = Math.max(0, Math.trunc(options.debounceMs ?? 0));

const pendingDebounceKeys = new Set<string>();

const activeInboundFlushes = new Set<Promise<void>>();

const buildInboundDebounceKey = (msg: WebInboundMessage): string | null => {

const sender = msg.sender;

const senderKey =

msg.chatType === "group"

? (getPrimaryIdentityId(sender ?? null) ??

msg.senderJid ??

msg.senderE164 ??

msg.senderName ??

msg.from)

: msg.from;

if (!senderKey) {

return null;

}

const conversationKey = msg.chatType === "group" ? msg.chatId : msg.from;

return `${msg.accountId}:${conversationKey}:${senderKey}`;

};

const shouldDebounceInboundMessage = (msg: WebInboundMessage): boolean =>

options.shouldDebounce?.(msg) ?? true;

const finalizeInboundDedupe = async (

entries: QueuedInboundMessage[],

};

const debouncer = createInboundDebouncer<QueuedInboundMessage>({

debounceMs: options.debounceMs ?? 0,

buildKey: (msg) => {

const sender = msg.sender;

const senderKey =

msg.chatType === "group"

? (getPrimaryIdentityId(sender ?? null) ??

msg.senderJid ??

msg.senderE164 ??

msg.senderName ??

msg.from)

: msg.from;

if (!senderKey) {

return null;

}

const conversationKey = msg.chatType === "group" ? msg.chatId : msg.from;

return `${msg.accountId}:${conversationKey}:${senderKey}`;

},

shouldDebounce: options.shouldDebounce,

debounceMs: inboundDebounceMs,

buildKey: (msg) => msg.debounceKey ?? buildInboundDebounceKey(msg),

shouldDebounce: shouldDebounceInboundMessage,

onFlush: async (entries) => {

const last = entries.at(-1);

if (!last) {

return;

}

let finishFlush!: () => void;

const flushTask = new Promise<void>((resolve) => {

finishFlush = resolve;

});

activeInboundFlushes.add(flushTask);

try {

if (entries.length === 1) {

await options.onMessage(last);

await finalizeInboundDedupe(entries);

const last = entries.at(-1);

if (!last) {

return;

}

const mentioned = new Set<string>();

try {

if (entries.length === 1) {

await options.onMessage(last);

await finalizeInboundDedupe(entries);

return;

}

const mentioned = new Set<string>();

for (const entry of entries) {

for (const jid of entry.mentions ?? entry.mentionedJids ?? []) {

mentioned.add(jid);

}

}

const combinedBody = entries

.map((entry) => entry.body)

.filter(Boolean)

.join("\n");

const combinedMessage: WebInboundMessage = {

...last,

body: combinedBody,

mentions: mentioned.size > 0 ? Array.from(mentioned) : undefined,

mentionedJids: mentioned.size > 0 ? Array.from(mentioned) : undefined,

isBatched: true,

};

await options.onMessage(combinedMessage);

await finalizeInboundDedupe(entries);

} catch (error) {

await finalizeInboundDedupe(entries, error);

throw error;

}

} finally {

for (const entry of entries) {

for (const jid of entry.mentions ?? entry.mentionedJids ?? []) {

mentioned.add(jid);

if (entry.debounceKey) {

pendingDebounceKeys.delete(entry.debounceKey);

}

}

const combinedBody = entries

.map((entry) => entry.body)

.filter(Boolean)

.join("\n");

const combinedMessage: WebInboundMessage = {

...last,

body: combinedBody,

mentions: mentioned.size > 0 ? Array.from(mentioned) : undefined,

mentionedJids: mentioned.size > 0 ? Array.from(mentioned) : undefined,

isBatched: true,

};

await options.onMessage(combinedMessage);

await finalizeInboundDedupe(entries);

} catch (error) {

await finalizeInboundDedupe(entries, error);

throw error;

activeInboundFlushes.delete(flushTask);

finishFlush();

}

},

onError: (err) => {

mediaFileName: enriched.mediaFileName,

dedupeKey: inbound.id ? `${options.accountId}:${inbound.remoteJid}:${inbound.id}` : undefined,

};

const debounceKey = buildInboundDebounceKey(inboundMessage);

if (debounceKey) {

inboundMessage.debounceKey = debounceKey;

if (inboundDebounceMs > 0 && shouldDebounceInboundMessage(inboundMessage)) {

pendingDebounceKeys.add(debounceKey);

}

}

if (inboundMessage.id) {

cacheInboundMessageMeta(inboundMessage.accountId, inboundMessage.chatId, inboundMessage.id, {

participant: inboundMessage.senderJid,

}

};

const pendingMessageHandlers = new Set<Promise<void>>();

const handleMessagesUpsert = async (upsert: { type?: string; messages?: Array<WAMessage> }) => {

if (upsert.type !== "notify" && upsert.type !== "append") {

return;

await enqueueInboundMessage(msg, inbound, enriched);

}

};

const handleMessagesUpsertEvent = (upsert: { type?: string; messages?: Array<WAMessage> }) => {

const task = handleMessagesUpsert(upsert).catch((err) => {

inboundLogger.error({ error: String(err) }, "messages.upsert handler error");

inboundConsoleLog.error(`Messages upsert handler error: ${String(err)}`);

});

pendingMessageHandlers.add(task);

void task.finally(() => {

pendingMessageHandlers.delete(task);

});

};

const waitForPendingMessageHandlers = async () => {

while (pendingMessageHandlers.size > 0) {

await Promise.all(Array.from(pendingMessageHandlers));

}

};

const drainDebouncedInboundMessages = async () => {

while (pendingDebounceKeys.size > 0 || activeInboundFlushes.size > 0) {

const debounceKeys = Array.from(pendingDebounceKeys);

if (debounceKeys.length > 0) {

await Promise.all(debounceKeys.map((key) => debouncer.flushKey(key)));

}

const flushes = Array.from(activeInboundFlushes);

if (flushes.length > 0) {

await Promise.allSettled(flushes);

}

await Promise.resolve();

}

};

const drainInboundBeforeSocketClose = async () => {

await waitForPendingMessageHandlers();

await drainDebouncedInboundMessages();

};

const drainInboundBeforeSocketCloseWithTimeout = async () => {

let timeout: ReturnType<typeof setTimeout> | null = null;

try {

await Promise.race([

drainInboundBeforeSocketClose(),

new Promise<void>((_, reject) => {

timeout = setTimeout(() => {

reject(

new Error(

`Timed out draining WhatsApp inbound debounce after ${INBOUND_CLOSE_DRAIN_TIMEOUT_MS}ms`,

),

);

}, INBOUND_CLOSE_DRAIN_TIMEOUT_MS);

timeout.unref?.();

}),

]);

} finally {

if (timeout) {

clearTimeout(timeout);

}

}

};

const handleConnectionUpdate = (update: Partial<import("baileys").ConnectionState>) => {

try {

if (update.connection === "close") {

removeListener?: (event: string, listener: (...args: unknown[]) => void) => void;

},

"messages.upsert",

handleMessagesUpsert as unknown as (...args: unknown[]) => void,

handleMessagesUpsertEvent as unknown as (...args: unknown[]) => void,

);

const detachConnectionUpdate = attachEmitterListener(

sock.ev as unknown as {

try {

detachMessagesUpsert();

detachConnectionUpdate();

await drainInboundBeforeSocketCloseWithTimeout();

} catch (err) {

logWhatsAppVerbose(options.verbose, `Inbound close drain failed: ${String(err)}`);

}

try {

closeInboundMonitorSocket(sock);

} catch (err) {

logWhatsAppVerbose(options.verbose, `Socket close failed: ${String(err)}`);