diff options
Diffstat (limited to 'src/daemon.js')
| -rw-r--r-- | src/daemon.js | 452 |
1 files changed, 452 insertions, 0 deletions
diff --git a/src/daemon.js b/src/daemon.js new file mode 100644 index 0000000..f74d7cd --- /dev/null +++ b/src/daemon.js @@ -0,0 +1,452 @@ +// Daemon: keeps the WhatsApp Web session open (headless Chrome through whatsapp-web.js) +// and exposes it on a Unix socket. WhatsApp Web takes 10-30 s to start and only receives +// messages while it is open, so one process per command does not work. + +import fs from 'node:fs'; +import net from 'node:net'; +import path from 'node:path'; +import wweb from 'whatsapp-web.js'; +import { getMethod, validate } from './api/index.js'; +import { patchClient, patchPage, patchStructures } from './compat.js'; +import { paths, loadConfig, VERSION } from './config.js'; +import { Hooks } from './hooks.js'; +import { WaError, onLines, send, ping } from './rpc.js'; +import * as ser from './serialize.js'; + +const { Client, LocalAuth } = wweb; +const STREAM = Symbol('stream'); +const KNOWN_CODES = new Set(['EINVAL', 'EAUTH', 'ENOTFOUND', 'EAMBIGUOUS', 'ETIMEOUT', 'EAPI', 'EDAEMON', 'EINTERNAL']); +const JID_RE = /^[^\s@]+@(c\.us|g\.us|lid|s\.whatsapp\.net|newsletter|broadcast)$/; + +const normalize = (s) => String(s).normalize('NFD').replace(/\p{M}/gu, '').toLowerCase().trim(); + +// The same person can show up as a number (@c.us) and as a LID (@lid): prefer the number. +function dedupe(entries) { + if (entries.length < 2) return entries; + const phone = entries.filter(([id]) => !id.endsWith('@lid')); + return phone.length === 1 ? phone : entries; +} + +function toRpcError(e) { + if (KNOWN_CODES.has(e.code)) return { code: e.code, message: e.message, ...(e.data ? { data: e.data } : {}) }; + return { code: 'EAPI', message: e.message || String(e) }; +} + +function processAlive(pid, needle) { + try { + process.kill(pid, 0); + } catch (e) { + if (e.code !== 'EPERM') return false; + } + try { + return fs.readFileSync(`/proc/${pid}/cmdline`, 'utf8').includes(needle); + } catch { + return true; + } +} + +function acquirePidFile(file) { + for (let i = 0; i < 3; i++) { + try { + fs.writeFileSync(file, `${process.pid}\n`, { flag: 'wx' }); + return; + } catch (e) { + if (e.code !== 'EEXIST') throw e; + const pid = Number.parseInt(fs.readFileSync(file, 'utf8'), 10); + if (pid && pid !== process.pid && processAlive(pid, 'daemon')) { + throw new WaError('EDAEMON', `a daemon is already running for this account (pid ${pid})`); + } + fs.rmSync(file, { force: true }); + } + } +} + +// If a previous daemon died from SIGKILL, its (detached) Chrome may still be alive +// holding the profile. It is detected through the profile SingletonLock. +function killOrphanChrome(profileDir, log) { + let target; + try { + target = fs.readlinkSync(path.join(profileDir, 'SingletonLock')); + } catch { + return; + } + const pid = Number.parseInt(target.split('-').pop(), 10); + if (pid && processAlive(pid, profileDir)) { + log(`killing orphan Chrome from a previous run (pid ${pid})`); + try { + process.kill(pid, 'SIGKILL'); + } catch { /* already dead */ } + } +} + +export async function runDaemon(account) { + const P = paths(account); + const cfg = loadConfig(account); + const log = (...a) => console.error(new Date().toISOString(), `[${account}]`, ...a); + + for (const d of [P.stateDir, P.dataDir]) fs.mkdirSync(d, { recursive: true, mode: 0o700 }); + fs.mkdirSync(path.dirname(P.socket), { recursive: true, mode: 0o700 }); + acquirePidFile(P.pidFile); + if (await ping(P.socket)) { + fs.rmSync(P.pidFile, { force: true }); + throw new WaError('EDAEMON', `a daemon is already listening on ${P.socket}`); + } + fs.rmSync(P.socket, { force: true }); + killOrphanChrome(path.join(P.dataDir, 'session'), log); + patchStructures(); + patchClient(Client); + if (cfg.trace || process.env.WA_CLI_TRACE) (await import('./trace.js')).enableTrace(log); + + const startedAt = new Date(); + const hooks = new Hooks({ + dir: P.hooksDir, account, socket: P.socket, timeoutMs: cfg.hookTimeoutMs, enabled: cfg.hooks, log, + }); + + // ---------------------------------------------------------------- session state + + const state = { state: 'starting', since: Date.now(), progress: Date.now(), qr: null, code: null, reason: null }; + const waiters = new Set(); + let wwebVersion = null; + + function setState(s, extra = {}) { + Object.assign(state, { state: s, since: Date.now(), progress: Date.now() }, extra); + for (const w of waiters) if (w.states.includes(s)) w.done(); + } + + function waitState(states, timeoutMs) { + if (states.includes(state.state)) return Promise.resolve(); + return new Promise((resolve, reject) => { + const w = { + states, + done: () => { + clearTimeout(w.timer); + waiters.delete(w); + resolve(); + }, + }; + w.timer = setTimeout(() => { + waiters.delete(w); + reject(new WaError('ETIMEOUT', `timed out waiting for state ${states.join('/')} (current: ${state.state})`)); + }, timeoutMs); + waiters.add(w); + }); + } + + async function ensureReady() { + const needAuth = () => new WaError('EAUTH', 'no linked session: run `wa auth login`'); + if (state.state === 'ready') return; + if (state.state === 'qr' || state.state === 'code') throw needAuth(); + await waitState(['ready', 'qr', 'code'], cfg.readyTimeoutMs); + if (state.state !== 'ready') throw needAuth(); + } + + // ---------------------------------------------------------------- client + + const client = new Client({ + authStrategy: new LocalAuth({ dataPath: P.dataDir }), + // No cache: WhatsApp Web changes version often and the local cache listener does + // `await res.text()` without a catch (it used to crash the process). + webVersionCache: { type: 'none' }, + puppeteer: { + ...cfg.puppeteer, + // No site isolation: the cross-origin WhatsApp Web iframe (pdf-viewer) runs in the + // same process. Otherwise it is a separate target, and when it navigates while the + // bindings are registered, exposeFunction fails with "Target closed" and `ready` never comes. + args: [...(cfg.puppeteer.args || []), '--disable-site-isolation-trials', '--disable-features=IsolateOrigins,site-per-process'], + }, + deviceName: cfg.deviceName, + browserName: cfg.browserName, + ...(cfg.webVersion ? { webVersion: cfg.webVersion } : {}), + }); + + const me = () => (client.info?.wid + ? { id: client.info.wid._serialized, name: client.info.pushname ?? null } + : null); + + // Contact names (address book or pushname) and chat names, for senderName/chatName. + // Loaded in the background once ready. Chats cover the @lid IDs, which + // getContacts() returns translated to @c.us. + const names = new Map(); + async function refreshNames() { + try { + for (const c of await client.getContacts()) { + const n = c.name || c.pushname || c.shortName || c.verifiedName; + if (n) names.set(c.id._serialized, n); + } + for (const c of await client.getChats()) { + if (c.name && (c.isGroup || !names.has(c.id._serialized))) names.set(c.id._serialized, c.name); + } + } catch (e) { + log('could not load contacts:', e.message); + } + } + + async function resolveChat(query) { + const q = String(query).trim(); + if (JID_RE.test(q)) return q.replace(/@s\.whatsapp\.net$/, '@c.us'); + if (/^\+?[\d\s().-]{6,}$/.test(q)) { + const id = await client.getNumberId(q.replace(/\D/g, '')); + if (!id) throw new WaError('ENOTFOUND', `the number ${q} is not on WhatsApp`); + return id._serialized; + } + // Exact name: first the contacts/groups cache (fast); otherwise getChats(), + // which can take several seconds on accounts with many groups. + const n = normalize(q); + const exactIn = (entries) => dedupe(entries.filter(([, name]) => normalize(name || '') === n)); + let exact = exactIn([...names]); + let chats = null; + if (!exact.length) { + chats = await client.getChats(); + exact = exactIn(chats.map((c) => [c.id._serialized, c.name])); + } + if (exact.length === 1) return exact[0][0]; + if (exact.length > 1) { + throw new WaError('EAMBIGUOUS', `"${q}" matches several chats; use the JID`, + exact.map(([id, name]) => ({ id, name }))); + } + const partial = chats.filter((c) => normalize(c.name || '').includes(n)).slice(0, 10) + .map((c) => ({ id: c.id._serialized, name: c.name })); + throw new WaError('ENOTFOUND', `no chat or contact named "${q}"` + + (partial.length ? `; similar: ${partial.map((c) => `${c.name} <${c.id}>`).join(', ')}` : ''), + partial.length ? partial : undefined); + } + + const ctx = { + client, account, paths: P, config: cfg, WaError, STREAM, normalize, waitState, resolveChat, + names: (id) => names.get(id), + get ser() { + return { names: ctx.names, me: me() }; + }, + status: () => ({ + account, + state: state.state, + since: new Date(state.since).toISOString(), + me: me(), + wwebVersion, + ...(state.state === 'qr' ? { qr: state.qr } : {}), + ...(state.state === 'code' ? { code: state.code } : {}), + ...(state.reason ? { reason: state.reason } : {}), + }), + daemonInfo: () => ({ + running: true, + pid: process.pid, + account, + version: VERSION, + startedAt: startedAt.toISOString(), + uptimeSec: Math.round(process.uptime()), + state: state.state, + socket: P.socket, + logFile: P.logFile, + dataDir: P.dataDir, + hooksDir: P.hooksDir, + subscribers: subs.size, + }), + async getChat(id) { + const c = await client.getChatById(id); + if (!c) throw new WaError('ENOTFOUND', `chat not found: ${id}`); + return c; + }, + async getMessage(id) { + const m = await client.getMessageById(id); + if (!m) throw new WaError('ENOTFOUND', `message not found: ${id}`); + return m; + }, + subscribe(conn, filter, replay) { + const sub = { ...conn, filter }; + subs.add(sub); + conn.conn.on('close', () => subs.delete(sub)); + if (!replay) return; + const snap = { + qr: ['qr', { qr: state.qr }], + code: ['code', { code: state.code }], + authenticated: ['authenticated', {}], + ready: ['ready', { me: me(), wwebVersion }], + }[state.state]; + if (snap && (!filter || filter.has(snap[0]))) send(sub.conn, { id: sub.id, event: envelope(...snap) }); + }, + shutdown, + }; + + // ---------------------------------------------------------------- events + + const subs = new Set(); + const envelope = (event, data) => ({ event, account, time: new Date().toISOString(), data }); + + function emit(event, data) { + const ev = envelope(event, data); + for (const s of subs) if (!s.filter || s.filter.has(event)) send(s.conn, { id: s.id, event: ev }); + hooks.run(event, ev); + } + + const msg = (m) => ser.message(m, ctx.ser); + const forward = { + qr: (qr) => { + setState('qr', { qr, code: null, reason: null }); + return { qr }; + }, + code: (code) => { + setState('code', { code, qr: null }); + return { code }; + }, + authenticated: () => { + if (state.state !== 'authenticated') log('authenticated, syncing'); + setState('authenticated', { qr: null, code: null }); + return {}; + }, + auth_failure: (message) => { + log('authentication failed:', message); + return { message }; + }, + ready: async () => { + // Compatibility patches before accepting commands (see src/compat.js). + try { + const applied = await patchPage(client.pupPage); + if (applied.length) log('compat:', applied.join(', ')); + } catch (e) { + log('compat: could not patch the page:', e.message); + } + wwebVersion = await client.getWWebVersion().catch(() => null); + refreshNames(); + setState('ready', { reason: null }); + log(`ready: ${me()?.name} <${me()?.id}> (WhatsApp Web ${wwebVersion})`); + return { me: me() }; + }, + disconnected: (reason) => { + log('disconnected:', reason); + setState('disconnected', { reason }); + // LOGOUT: whatsapp-web.js goes back to the QR screen by itself. Any other reason: + // exit and let systemd (or the next `wa`) start it again. + if (reason !== 'LOGOUT') setImmediate(() => shutdown(`disconnected (${reason})`, 1)); + return { reason }; + }, + change_state: (s) => ({ state: s }), + loading_screen: (percent, message) => { + state.progress = Date.now(); + return { percent, message }; + }, + message: msg, + message_create: msg, + media_uploaded: msg, + message_revoke_me: msg, + message_ack: (m, ack) => ({ message: msg(m), ack }), + message_edit: (m, newBody, prevBody) => ({ message: msg(m), newBody, prevBody }), + message_revoke_everyone: (m, revoked) => ({ message: msg(m), revoked: revoked ? msg(revoked) : null }), + message_reaction: ser.reaction, + unread_count: (c) => ser.chat(c, ctx.ser), + chat_archived: (c, archived) => ({ chat: ser.chat(c, ctx.ser), archived }), + chat_removed: (c) => ser.chat(c, ctx.ser), + contact_changed: (m, oldId, newId, isContact) => { + refreshNames(); + return { message: msg(m), oldId, newId, isContact }; + }, + group_join: ser.groupNotification, + group_leave: ser.groupNotification, + group_update: ser.groupNotification, + group_admin_changed: ser.groupNotification, + group_membership_request: ser.groupNotification, + call: ser.call, + vote_update: (v) => ({ + voter: v.voter, + selectedOptions: v.selectedOptions, + messageId: v.parentMessage?.id?._serialized ?? null, + timestamp: v.interractedAtTs, + }), + }; + for (const [event, fn] of Object.entries(forward)) { + client.on(event, async (...args) => { + let data; + try { + data = await fn(...args); + } catch (e) { + log(`error handling event ${event}:`, e.message); + return; + } + emit(event, data); + }); + } + + // ---------------------------------------------------------------- socket + + const server = net.createServer((conn) => { + conn.on('error', () => {}); + onLines(conn, async (line) => { + let req; + try { + req = JSON.parse(line); + } catch { + return send(conn, { id: null, error: { code: 'EINVAL', message: 'invalid JSON' } }); + } + const { id = null, method, params } = req; + try { + const def = getMethod(method); + const p = validate(method, params ?? {}); + if (def.ready !== false) await ensureReady(); + for (const [k, spec] of Object.entries(def.params || {})) { + if (spec.type === 'chat' && p[k] !== undefined) p[k] = await resolveChat(p[k]); + } + const result = await def.handler(ctx, p, { conn, id }); + if (result !== STREAM) send(conn, { id, result: result ?? null }); + } catch (e) { + if (!KNOWN_CODES.has(e.code)) log(`${method}:`, e.stack || e.message); + send(conn, { id, error: toRpcError(e) }); + } + }); + }); + + let stopping = false; + async function shutdown(reason, code = 0) { + if (stopping) return; + stopping = true; + log('stopping:', reason); + setTimeout(() => process.exit(code), 15000).unref(); + server.close(); + for (const s of subs) s.conn.end(); + fs.rmSync(P.socket, { force: true }); + try { + await client.destroy(); + } catch (e) { + log('error closing the browser:', e.message); + } + fs.rmSync(P.pidFile, { force: true }); + process.exit(code); + } + // whatsapp-web.js has async listeners without a catch (e.g. `await res.text()` on the + // web.whatsapp.com response, which fails if Chrome already dropped the body). Without + // this, any of those rejections kills the daemon. Real failures (browser closed, + // disconnection, stuck startup) have their own handling. + process.on('unhandledRejection', (e) => log('unhandled promise rejection:', e?.stack || e)); + process.on('SIGTERM', () => shutdown('SIGTERM')); + process.on('SIGINT', () => shutdown('SIGINT')); + process.on('SIGHUP', () => shutdown('SIGHUP')); + + await new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(P.socket, () => { + fs.chmodSync(P.socket, 0o600); + resolve(); + }); + }); + log(`wa-cli ${VERSION}: listening on ${P.socket} (pid ${process.pid})`); + + // Watchdog: whatsapp-web.js has awaits without a timeout while loading the page and + // injecting itself (page.goto, exposeFunction). If it hangs, it stays in + // starting/authenticated forever and never emits `ready`. With no progress within + // startTimeoutMs, exit: systemd (or the next `wa`) restarts it. loading_screen counts + // as progress (long sync). From authenticated to ready usually takes a few seconds + // (the heavy sync happens before). + setInterval(() => { + const stalled = Date.now() - state.progress; + const limit = state.state === 'authenticated' ? Math.min(cfg.startTimeoutMs, 60000) : cfg.startTimeoutMs; + if ((state.state === 'starting' || state.state === 'authenticated') && stalled > limit) { + shutdown(`no progress in ${Math.round(stalled / 1000)} s (state ${state.state})`, 1); + } + }, 5000).unref(); + + try { + await client.initialize(); + client.pupBrowser?.on('disconnected', () => shutdown('the browser closed', 1)); + } catch (e) { + log('error initializing WhatsApp Web:', e.stack || e.message); + await shutdown('initialization error', 1); + } +} |