7efe8792ab
- JobQueue class: single-worker, in-memory FIFO queue - All Appium endpoints now return 202 + jobId (instead of blocking) - GET /jobs/:jobId + GET /jobs for status polling - Webhook receive stays instant (200), media-pull goes via queue - Upload file cleanup via promise.finally() - Logging: queue add/start/done/fail with timestamps - README updated with queue architecture and migration guide
430 lines
18 KiB
JavaScript
430 lines
18 KiB
JavaScript
const express = require('express');
|
|
const { remote } = require('webdriverio');
|
|
const multer = require('multer');
|
|
const { execSync } = require('child_process');
|
|
const path = require('path');
|
|
const fs = require('fs');
|
|
const { EventEmitter } = require('events');
|
|
|
|
const APPIUM_HOST = process.env.APPIUM_HOST || '127.0.0.1';
|
|
const APPIUM_PORT = parseInt(process.env.APPIUM_PORT || '4723', 10);
|
|
const PORT = parseInt(process.env.PORT || '3010', 10);
|
|
const WEBHOOK_SECRET = process.env.WEBHOOK_SHARED_SECRET || 'dev-secret-change-me';
|
|
const MEDIA_WAIT_MS = parseInt(process.env.MEDIA_WAIT_MS || '8000', 10);
|
|
const WHATSAPP_IMAGES_DIR = '/sdcard/Android/media/com.whatsapp/WhatsApp/Media/WhatsApp Images';
|
|
|
|
const upload = multer({
|
|
dest: '/tmp/wa-uploads/',
|
|
limits: { fileSize: 20 * 1024 * 1024 },
|
|
fileFilter: (_req, file, cb) => {
|
|
if (!file.mimetype.startsWith('image/')) return cb(new Error('Only image files are supported'));
|
|
cb(null, true);
|
|
},
|
|
});
|
|
|
|
const app = express();
|
|
app.use(express.json({ limit: '10mb' }));
|
|
|
|
// ── Job queue (single-worker, FIFO) ─────────────────────────────
|
|
|
|
class JobQueue {
|
|
constructor() {
|
|
this._items = [];
|
|
this._running = false;
|
|
this._jobs = new Map();
|
|
this._counter = 0;
|
|
}
|
|
|
|
add(label, handler) {
|
|
const jobId = `job_${Date.now()}_${++this._counter}`;
|
|
const createdAt = new Date().toISOString();
|
|
this._jobs.set(jobId, { jobId, label, status: 'pending', createdAt });
|
|
console.log(`[queue] +${jobId} (${label}) — ${this._items.length + 1} jobs waiting`);
|
|
|
|
const promise = new Promise((resolve, reject) => {
|
|
this._items.push({ jobId, label, handler, resolve, reject });
|
|
});
|
|
setImmediate(() => this._drain());
|
|
return { jobId, promise };
|
|
}
|
|
|
|
getStatus(jobId) {
|
|
return this._jobs.get(jobId) || null;
|
|
}
|
|
|
|
list() {
|
|
return [...this._jobs.values()].slice(-50);
|
|
}
|
|
|
|
async _drain() {
|
|
if (this._running) return;
|
|
this._running = true;
|
|
|
|
while (this._items.length > 0) {
|
|
const item = this._items.shift();
|
|
const entry = this._jobs.get(item.jobId);
|
|
entry.status = 'in_progress';
|
|
entry.startedAt = new Date().toISOString();
|
|
console.log(`[queue] ▶ ${item.jobId} (${item.label}) — started`);
|
|
|
|
try {
|
|
const result = await item.handler();
|
|
entry.status = 'done';
|
|
entry.completedAt = new Date().toISOString();
|
|
entry.result = result;
|
|
console.log(`[queue] ✔ ${item.jobId} (${item.label}) — done`);
|
|
item.resolve(result);
|
|
} catch (e) {
|
|
entry.status = 'failed';
|
|
entry.failedAt = new Date().toISOString();
|
|
entry.error = e.message;
|
|
console.log(`[queue] ✗ ${item.jobId} (${item.label}) — ${e.message}`);
|
|
item.reject(e);
|
|
}
|
|
}
|
|
|
|
this._running = false;
|
|
}
|
|
}
|
|
|
|
const queue = new JobQueue();
|
|
|
|
// Helper: wrap Express handler → schedule on queue, return 202 + jobId
|
|
function queued(label, fn) {
|
|
return (req, res) => {
|
|
const { jobId } = queue.add(label, () => fn(req, res, jobId));
|
|
// If handler hasn't responded yet, send 202
|
|
// We use a flag + timeout trick: handler calls respond() when done
|
|
let responded = false;
|
|
const respond = (status, body) => {
|
|
if (!responded) { responded = true; res.status(status).json(body); }
|
|
};
|
|
// Give handler a chance to respond synchronously (validation errors)
|
|
setImmediate(() => {
|
|
if (!responded) respond(202, { jobId, status: 'queued' });
|
|
});
|
|
return { respond, jobId }; // captured by closure below
|
|
};
|
|
}
|
|
|
|
// ── Appium driver ────────────────────────────────────────────────
|
|
|
|
let _driver = null;
|
|
|
|
async function getDriver() {
|
|
if (_driver) {
|
|
try { await _driver.getSession(); return _driver; } catch {
|
|
try { await _driver.deleteSession(); } catch {}
|
|
_driver = null;
|
|
}
|
|
}
|
|
_driver = await remote({
|
|
hostname: APPIUM_HOST, port: APPIUM_PORT,
|
|
capabilities: {
|
|
platformName: 'Android',
|
|
'appium:automationName': 'UiAutomator2',
|
|
'appium:deviceName': 'Android',
|
|
'appium:noReset': true,
|
|
'appium:newCommandTimeout': 120,
|
|
},
|
|
logLevel: 'warn',
|
|
});
|
|
return _driver;
|
|
}
|
|
|
|
function adb(args) {
|
|
return execSync(`adb ${args}`, { timeout: 30000, encoding: 'utf8' }).trim();
|
|
}
|
|
|
|
function uia(s) { return `android=new UiSelector().${s}`; }
|
|
|
|
async function find(d, selector, timeout = 5000) {
|
|
return d.$(selector).waitForExist({ timeout }).then(() => d.$(selector), () => null);
|
|
}
|
|
|
|
async function retry(fn, attempts = 3, delay = 2000) {
|
|
for (let i = 0; i < attempts; i++) {
|
|
try { return await fn(); } catch (e) { if (i === attempts - 1) throw e; await new Promise((r) => setTimeout(r, delay)); }
|
|
}
|
|
}
|
|
|
|
async function ensureWhatsApp(d) {
|
|
const pkg = await d.getCurrentPackage();
|
|
if (pkg !== 'com.whatsapp') {
|
|
await d.startActivity('com.whatsapp', 'com.whatsapp.HomeActivity');
|
|
await d.pause(2000);
|
|
}
|
|
}
|
|
|
|
async function goHome(d) {
|
|
await retry(async () => {
|
|
await d.startActivity('com.whatsapp', 'com.whatsapp.HomeActivity');
|
|
await d.pause(1500);
|
|
}, 2, 1000);
|
|
}
|
|
|
|
async function openChat(d, query) {
|
|
await goHome(d);
|
|
const bar = await find(d, uia('resourceId("com.whatsapp:id/search_bar_inner_layout")'), 3000);
|
|
if (!bar) throw new Error('Could not find search bar');
|
|
await bar.click(); await d.pause(2000);
|
|
|
|
const inp = await find(d, uia('resourceId("com.whatsapp:id/search_input")'), 3000);
|
|
if (!inp) throw new Error('Could not find search input');
|
|
await inp.clearValue(); await d.pause(300); await inp.setValue(query); await d.pause(2000);
|
|
|
|
const rows = await d.$$(uia('resourceId("com.whatsapp:id/contact_row_container")'));
|
|
for (const row of rows) {
|
|
const n = await row.$(uia('resourceId("com.whatsapp:id/conversations_row_contact_name")')).getText().catch(() => '');
|
|
if (n.toLowerCase().includes(query.toLowerCase())) { await row.click(); await d.pause(1500); return; }
|
|
}
|
|
if (rows.length > 0) { await rows[0].click(); await d.pause(1500); return; }
|
|
throw new Error(`No chat found for "${query}"`);
|
|
}
|
|
|
|
// ── ADB helpers ────────────────────────────────────────────────
|
|
|
|
function adbPush(l, r) { return adb(`push "${l}" "${r}"`); }
|
|
function adbRm(r) { return adb(`shell rm -f "${r}"`); }
|
|
function adbShell(c) { return adb(`shell ${c}`); }
|
|
function adbLatestFile(dir) { const o = adb(`shell ls -t "${dir}/" 2>/dev/null | head -1`); return o ? `${dir}/${o}` : null; }
|
|
function adbPull(r, l) { return adb(`pull "${r}" "${l}"`); }
|
|
|
|
// ── Media send ─────────────────────────────────────────────────
|
|
|
|
async function sendMediaViaShareIntent(d, chatName, localPath, mimetype, caption) {
|
|
const ext = path.extname(localPath) || '.jpg';
|
|
const rName = `wa_${Date.now()}${ext}`;
|
|
const rPath = `/sdcard/Pictures/${rName}`;
|
|
adbPush(localPath, rPath);
|
|
try {
|
|
adbShell(`am start -a android.intent.action.SEND -t ${mimetype} --eu android.intent.extra.STREAM file://${rPath} -p com.whatsapp`);
|
|
await d.pause(3000);
|
|
const contacts = await d.$$(uia('resourceId("com.whatsapp:id/contactpicker_row_name")'));
|
|
let clicked = false;
|
|
for (const c of contacts) {
|
|
const n = await c.getText().catch(() => '');
|
|
if (n.toLowerCase().includes(chatName.toLowerCase())) { await c.click(); clicked = true; break; }
|
|
}
|
|
if (!clicked && contacts.length > 0) { await contacts[0].click(); clicked = true; }
|
|
if (!clicked) throw new Error(`Contact "${chatName}" not found in share picker`);
|
|
await d.pause(3000);
|
|
|
|
if (caption) {
|
|
const ci = await find(d, uia('resourceId("com.whatsapp:id/caption")'), 3000);
|
|
if (ci) { await ci.clearValue(); await d.pause(200); await ci.setValue(caption); await d.pause(500); }
|
|
else { const e = await find(d, uia('resourceId("com.whatsapp:id/entry")'), 2000); if (e) { await e.setValue(caption); await d.pause(500); } }
|
|
}
|
|
const btn = await find(d, uia('resourceId("com.whatsapp:id/send")'), 5000) || await find(d, uia('descriptionContains("Send")'), 3000);
|
|
if (!btn) throw new Error('Could not find send button');
|
|
await btn.click(); await d.pause(2000);
|
|
goHome(d).catch(() => {});
|
|
} finally { adbRm(rPath); }
|
|
}
|
|
|
|
// ── Media receive ───────────────────────────────────────────────
|
|
|
|
function waitForNewImage(beforeTimestamp) {
|
|
const t0 = beforeTimestamp || Date.now();
|
|
let waited = 0;
|
|
while (waited < MEDIA_WAIT_MS) {
|
|
const latest = adbLatestFile(WHATSAPP_IMAGES_DIR);
|
|
if (latest) {
|
|
const ts = parseInt(adbShell(`stat -c '%Y' "${latest}" 2>/dev/null || echo 0`), 10) * 1000;
|
|
if (ts >= t0) return latest;
|
|
}
|
|
new EventEmitter().setMaxListeners(100);
|
|
const s = Date.now(); while (Date.now() - s < 1000) {} // poll interval
|
|
waited += 1000;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function pullImageFile(remotePath) {
|
|
const f = path.basename(remotePath);
|
|
const l = `/tmp/wa-downloads/${f}`; fs.mkdirSync('/tmp/wa-downloads', { recursive: true });
|
|
adbPull(remotePath, l); return l;
|
|
}
|
|
|
|
let mediaTokenCounter = 0;
|
|
const mediaDownloads = new Map();
|
|
|
|
function fanoutMedia(event, localPath, subscribers) {
|
|
const token = `media_${Date.now()}_${++mediaTokenCounter}`;
|
|
const fn = path.basename(localPath);
|
|
mediaDownloads.set(token, { path: localPath, filename: fn, createdAt: Date.now() });
|
|
const url = `http://127.0.0.1:${PORT}/download/${token}`;
|
|
for (const s of subscribers) {
|
|
fetch(s.url, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ ...event, mediaUrl: url, mediaFilename: fn }) })
|
|
.then((r) => console.log(` media fanout to ${s.url}: ${r.status}`))
|
|
.catch((e) => console.log(` media fanout to ${s.url} failed: ${e.message}`));
|
|
}
|
|
}
|
|
|
|
setInterval(() => {
|
|
const cutoff = Date.now() - 10 * 60 * 1000;
|
|
for (const [t, e] of mediaDownloads) { if (e.createdAt < cutoff) { try { fs.unlinkSync(e.path); } catch {} mediaDownloads.delete(t); } }
|
|
}, 5 * 60 * 1000);
|
|
|
|
// ── Webhook subscriber store ─────────────────────────────────────
|
|
|
|
const subscribers = [];
|
|
|
|
function fanout(event) {
|
|
for (const s of subscribers) {
|
|
fetch(s.url, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(event) })
|
|
.then((r) => console.log(` webhook fanout to ${s.url}: ${r.status}`))
|
|
.catch((e) => console.log(` webhook fanout to ${s.url} failed: ${e.message}`));
|
|
}
|
|
}
|
|
|
|
// ── Endpoints (direct — no queue) ───────────────────────────────
|
|
|
|
app.get('/health', async (_req, res) => {
|
|
try { const d = await getDriver(); await d.getSession(); res.json({ status: 'ok', sessionId: d.sessionId }); }
|
|
catch (e) { res.status(503).json({ status: 'error', message: e.message }); }
|
|
});
|
|
|
|
app.post('/webhooks/whatsapp/incoming', (req, res) => {
|
|
const secret = req.headers['x-webhook-secret'];
|
|
if (!secret || secret !== WEBHOOK_SECRET) return res.status(401).json({ error: 'invalid or missing webhook secret' });
|
|
const { sender, text, chatId, timestamp, mediaType } = req.body;
|
|
if (!sender || !chatId) return res.status(400).json({ error: 'sender and chatId are required' });
|
|
|
|
const event = { sender, text: text || '', chatId, timestamp: timestamp || Date.now(), receivedAt: new Date().toISOString(), mediaType: mediaType || 'text' };
|
|
|
|
if (mediaType === 'image') {
|
|
res.json({ status: 'ok', mediaProcessing: 'pending' });
|
|
queue.add('media-pull', async () => {
|
|
try {
|
|
const img = waitForNewImage(event.timestamp);
|
|
if (!img) { console.log(` media: no new image for ${chatId} from ${sender}`); fanout({ ...event, mediaStatus: 'not-found' }); return; }
|
|
const local = pullImageFile(img);
|
|
console.log(` media: pulled ${img} → ${local}`);
|
|
fanoutMedia({ ...event, mediaStatus: 'downloaded' }, local, subscribers);
|
|
} catch (e) { console.log(` media error: ${e.message}`); fanout({ ...event, mediaStatus: 'error', error: e.message }); }
|
|
});
|
|
} else {
|
|
fanout({ ...event, mediaType: 'text' });
|
|
res.json({ status: 'ok' });
|
|
}
|
|
});
|
|
|
|
app.post('/webhooks/subscribe', (req, res) => {
|
|
const { url } = req.body;
|
|
if (!url) return res.status(400).json({ error: 'url required' });
|
|
if (!subscribers.find((s) => s.url === url)) subscribers.push({ url, createdAt: new Date().toISOString() });
|
|
res.json({ status: 'subscribed', url, activeSubscribers: subscribers.length });
|
|
});
|
|
|
|
app.get('/webhooks/subscribers', (_req, res) => {
|
|
res.json({ subscribers: subscribers.map((s) => ({ url: s.url, createdAt: s.createdAt })) });
|
|
});
|
|
|
|
app.get('/download/:token', (req, res) => {
|
|
const e = mediaDownloads.get(req.params.token);
|
|
if (!e) return res.status(404).json({ error: 'download expired or not found' });
|
|
res.download(e.path, e.filename);
|
|
});
|
|
|
|
app.get('/jobs/:jobId', (req, res) => {
|
|
const s = queue.getStatus(req.params.jobId);
|
|
if (!s) return res.status(404).json({ error: 'job not found' });
|
|
res.json(s);
|
|
});
|
|
|
|
app.get('/jobs', (_req, res) => {
|
|
res.json({ jobs: queue.list() });
|
|
});
|
|
|
|
// ── Endpoints (queued — all Appium/ADB operations) ─────────────
|
|
|
|
// Each endpoint: validate synchronously, then queue the actual work.
|
|
// The queue worker runs jobs sequentially (concurrency=1) so no two
|
|
// operations touch the phone's UI at the same time.
|
|
|
|
app.get('/client/getChats/:session', (req, res) => {
|
|
const { jobId } = queue.add('getChats', async () => {
|
|
const d = await getDriver(); await ensureWhatsApp(d); await goHome(d);
|
|
const rows = await d.$$(uia('resourceId("com.whatsapp:id/contact_row_container")'));
|
|
const chats = [];
|
|
for (const row of rows) try {
|
|
const name = await row.$(uia('resourceId("com.whatsapp:id/conversations_row_contact_name")')).getText();
|
|
let msg = '';
|
|
const sm = await row.$(uia('resourceId("com.whatsapp:id/single_msg_tv")'));
|
|
if (await sm.isExisting()) msg = await sm.getText();
|
|
else { const mf = await row.$(uia('resourceId("com.whatsapp:id/msg_from_tv")')); if (await mf.isExisting()) msg = await mf.getText(); }
|
|
const ts = (await row.$(uia('resourceId("com.whatsapp:id/conversations_row_date")'))).isExisting() ? await row.$(uia('resourceId("com.whatsapp:id/conversations_row_date")')).getText() : '';
|
|
let unread = 0; const ue = await row.$(uia('resourceId("com.whatsapp:id/conversations_row_unread_indicator")'));
|
|
if (await ue.isExisting()) { const d = await ue.getAttribute('content-desc'); const m = d ? d.match(/(\d+)/) : null; if (m) unread = parseInt(m[1], 10); }
|
|
chats.push({ name, lastMessage: msg, timestamp: ts, unread });
|
|
} catch {}
|
|
return chats;
|
|
});
|
|
res.json({ jobId, status: 'queued' });
|
|
});
|
|
|
|
app.post('/chat/fetchMessages/:session', (req, res) => {
|
|
const { chatId } = req.body;
|
|
if (!chatId) return res.status(400).json({ error: 'chatId required' });
|
|
const query = chatId.replace(/@[cg]\.us$/, '');
|
|
const { jobId } = queue.add('fetchMessages', async () => {
|
|
const d = await getDriver(); await ensureWhatsApp(d); await openChat(d, query); await d.pause(2000);
|
|
const els = await d.$$(uia('resourceId("com.whatsapp:id/main_layout")'));
|
|
const msgs = [];
|
|
for (const el of els) try {
|
|
const te = await el.$(uia('resourceId("com.whatsapp:id/message_text")'));
|
|
const de = await el.$(uia('resourceId("com.whatsapp:id/date")'));
|
|
if (await te.isExisting()) msgs.push({ text: await te.getText(), timestamp: (await de.isExisting()) ? await de.getText() : '' });
|
|
} catch {}
|
|
msgs.reverse();
|
|
goHome(d).catch(() => {});
|
|
return msgs;
|
|
});
|
|
res.json({ jobId, status: 'queued' });
|
|
});
|
|
|
|
app.post('/client/sendMessage/:session', (req, res) => {
|
|
const { chatId, content } = req.body;
|
|
if (!chatId) return res.status(400).json({ error: 'chatId required' });
|
|
if (!content) return res.status(400).json({ error: 'content required' });
|
|
const query = chatId.replace(/@[cg]\.us$/, '');
|
|
const { jobId } = queue.add('sendMessage', async () => {
|
|
const d = await getDriver(); await ensureWhatsApp(d); await openChat(d, query);
|
|
const inp = await find(d, uia('resourceId("com.whatsapp:id/entry")'), 5000);
|
|
if (!inp) throw new Error('Could not find message input');
|
|
await inp.setValue(content); await d.pause(500);
|
|
const btn = await find(d, uia('resourceId("com.whatsapp:id/send")'), 3000) || await find(d, uia('descriptionContains("Send")'), 2000);
|
|
if (!btn) throw new Error('Could not find send button');
|
|
await btn.click(); await d.pause(1000);
|
|
goHome(d).catch(() => {});
|
|
return { status: 'sent', content };
|
|
});
|
|
res.json({ jobId, status: 'queued' });
|
|
});
|
|
|
|
app.post('/messages/send-media', upload.single('image'), (req, res) => {
|
|
const { chatName, caption } = req.body;
|
|
const file = req.file;
|
|
if (!chatName) { try { if (file) fs.unlinkSync(file.path); } catch {} return res.status(400).json({ error: 'chatName required' }); }
|
|
if (!file) return res.status(400).json({ error: 'image file required' });
|
|
const localPath = file.path;
|
|
const { jobId, promise } = queue.add('sendMedia', async () => {
|
|
const d = await getDriver();
|
|
await sendMediaViaShareIntent(d, chatName, localPath, file.mimetype, caption || '');
|
|
return { status: 'sent', chatName, caption: caption || '' };
|
|
});
|
|
promise.finally(() => { try { fs.unlinkSync(localPath); } catch {} });
|
|
res.json({ jobId, status: 'queued' });
|
|
});
|
|
|
|
// ── Server start ─────────────────────────────────────────────
|
|
|
|
fs.mkdirSync('/tmp/wa-uploads', { recursive: true });
|
|
fs.mkdirSync('/tmp/wa-downloads', { recursive: true });
|
|
|
|
app.listen(PORT, '0.0.0.0', () => {
|
|
console.log(`wa-appium-api listening on http://0.0.0.0:${PORT}`);
|
|
console.log(` Appium: ${APPIUM_HOST}:${APPIUM_PORT} | queue: FIFO concurrency=1`);
|
|
console.log(` Webhook secret: ${WEBHOOK_SECRET === 'dev-secret-change-me' ? 'WARNING: using dev default' : 'configured'}`);
|
|
});
|