From 7efe8792ab22019690457c0a4e640aa9b6564720 Mon Sep 17 00:00:00 2001
From: Eddy de Vink <github@eddydev.ink>
Date: Wed, 22 Jul 2026 22:10:06 +0200
Subject: [PATCH] feat: job queue (FIFO concurrency=1) for all Appium/ADB
 operations

- 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
---
 README.md               |  23 ++
 wa-appium-api/server.js | 638 ++++++++++++++++------------------------
 2 files changed, 284 insertions(+), 377 deletions(-)

diff --git a/README.md b/README.md
index 7a32e7f..7f8c659 100644
--- a/README.md
+++ b/README.md
@@ -347,6 +347,29 @@ curl -X POST http://127.0.0.1:3010/webhooks/whatsapp/incoming \
 - `openChat(d, query)` — zoek chat via search, open 'm
 - Finders gebruiken UiAutomator2 `new UiSelector()` via WebDriverIO
 
+**Job queue (FIFO, concurrency=1):**
+Alle Appium/ADB-operaties gaan door één centrale queue, omdat er maar één
+fysieke telefoon is. Geen twee acties tegelijk op dezelfde UI-sessie.
+
+- **Aanbieding:** endpoints geven meteen `202 { jobId }` terug, job wordt
+  achter de schermen verwerkt
+- **Status:** `GET /jobs/:jobId` — pending / in_progress / done / failed
+- **Overzicht:** `GET /jobs` — laatste 50 jobs
+- **Webhook:** ontvangst gaat direct (200), alleen media-pull gaat via queue
+- **Retry:** bij falen blijft de job op `failed`. De aanroeper moet zelf
+  opnieuw proberen.
+- **Logging:** `[queue] +jobId (label) — N jobs waiting`, dan `▶` en `✔`/`✗`
+- **Breaking change:** `getChats` en `fetchMessages` waren synchroon (data
+  direct in response). Nu geven ze ook `202 + jobId`. Poll `GET /jobs/:jobId`
+  tot `done` en check `result` voor de data.
+
+```bash
+# voorbeeld: job aanmaken, status pollen, resultaat uitlezen
+JOB=$(curl -s http://localhost:3010/client/getChats/EdGpt | python3 -c "import sys,json; print(json.load(sys.stdin)['jobId'])")
+sleep 5
+curl -s http://localhost:3010/jobs/$JOB | python3 -m json.tool
+```
+
 ---
 
 ## Troubleshooting
diff --git a/wa-appium-api/server.js b/wa-appium-api/server.js
index 8cb1b9e..9ebf8c3 100644
--- a/wa-appium-api/server.js
+++ b/wa-appium-api/server.js
@@ -1,9 +1,10 @@
 const express = require('express');
 const { remote } = require('webdriverio');
 const multer = require('multer');
-const { execSync, exec } = require('child_process');
+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);
@@ -16,9 +17,7 @@ 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'));
-    }
+    if (!file.mimetype.startsWith('image/')) return cb(new Error('Only image files are supported'));
     cb(null, true);
   },
 });
@@ -26,21 +25,101 @@ const upload = multer({
 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.getSession(); return _driver; } catch {
       try { await _driver.deleteSession(); } catch {}
       _driver = null;
     }
   }
   _driver = await remote({
-    hostname: APPIUM_HOST,
-    port: APPIUM_PORT,
+    hostname: APPIUM_HOST, port: APPIUM_PORT,
     capabilities: {
       platformName: 'Android',
       'appium:automationName': 'UiAutomator2',
@@ -53,27 +132,20 @@ async function getDriver() {
   return _driver;
 }
 
-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));
-    }
-  }
+function adb(args) {
+  return execSync(`adb ${args}`, { timeout: 30000, encoding: 'utf8' }).trim();
 }
 
-function uia(selector) {
-  return `android=new UiSelector().${selector}`;
-}
+function uia(s) { return `android=new UiSelector().${s}`; }
 
-function find(d, selector, timeout = 5000) {
+async function find(d, selector, timeout = 5000) {
   return d.$(selector).waitForExist({ timeout }).then(() => d.$(selector), () => null);
 }
 
-function adb(args) {
-  return execSync(`adb ${args}`, { timeout: 30000, encoding: 'utf8' }).trim();
+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) {
@@ -93,417 +165,144 @@ async function goHome(d) {
 
 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 searchBar = await find(
-    d,
-    uia('resourceId("com.whatsapp:id/search_bar_inner_layout")'),
-    3000
-  );
-  if (!searchBar) throw new Error('Could not find search bar');
-  await searchBar.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 searchInput = await find(d, uia('resourceId("com.whatsapp:id/search_input")'), 3000);
-  if (!searchInput) throw new Error('Could not find search input');
-  await searchInput.clearValue();
-  await d.pause(300);
-  await searchInput.setValue(query);
-  await d.pause(2000);
-
-  const chats = await d.$$(uia('resourceId("com.whatsapp:id/contact_row_container")'));
-  for (const row of chats) {
-    const nameEl = await row.$(
-      uia('resourceId("com.whatsapp:id/conversations_row_contact_name")')
-    );
-    const name = await nameEl.getText().catch(() => '');
-    if (name.toLowerCase().includes(query.toLowerCase())) {
-      await row.click();
-      await d.pause(1500);
-      return;
-    }
-  }
-  if (chats.length > 0) {
-    await chats[0].click();
-    await d.pause(1500);
-    return;
+  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(localPath, remotePath) {
-  return adb(`push "${localPath}" "${remotePath}"`);
-}
-
-function adbRm(remotePath) {
-  return adb(`shell rm -f "${remotePath}"`);
-}
-
-function adbShell(cmd) {
-  return adb(`shell ${cmd}`);
-}
-
-function adbLatestFile(dir) {
-  const out = adb(`shell ls -t "${dir}/" 2>/dev/null | head -1`);
-  return out ? `${dir}/${out}` : null;
-}
-
-function adbPull(remotePath, localPath) {
-  return adb(`pull "${remotePath}" "${localPath}"`);
-}
+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 remoteName = `wa_${Date.now()}${ext}`;
-  const remotePath = `/sdcard/Pictures/${remoteName}`;
-
-  adbPush(localPath, remotePath);
-
+  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://${remotePath} ` +
-      `-p com.whatsapp`
-    );
+    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")')
-    );
+    const contacts = await d.$$(uia('resourceId("com.whatsapp:id/contactpicker_row_name")'));
     let clicked = false;
     for (const c of contacts) {
-      const name = await c.getText().catch(() => '');
-      if (name.toLowerCase().includes(chatName.toLowerCase())) {
-        await c.click();
-        clicked = true;
-        break;
-      }
-    }
-    if (!clicked && contacts.length > 0) {
-      await contacts[0].click();
-      clicked = true;
+      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 captionInput = await find(
-        d,
-        uia('resourceId("com.whatsapp:id/caption")'),
-        3000
-      );
-      if (captionInput) {
-        await captionInput.clearValue();
-        await d.pause(200);
-        await captionInput.setValue(caption);
-        await d.pause(500);
-      } else {
-        const entry = await find(d, uia('resourceId("com.whatsapp:id/entry")'), 2000);
-        if (entry) {
-          await entry.setValue(caption);
-          await d.pause(500);
-        }
-      }
+      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); } }
     }
-
-    let sendBtn = await find(d, uia('resourceId("com.whatsapp:id/send")'), 5000);
-    if (!sendBtn) {
-      sendBtn = await find(d, uia('descriptionContains("Send")'), 3000);
-    }
-    if (!sendBtn) throw new Error('Could not find send button after selecting contact');
-
-    await sendBtn.click();
-    await d.pause(2000);
-
+    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(remotePath);
-  }
+  } finally { adbRm(rPath); }
 }
 
 // ── Media receive ───────────────────────────────────────────────
 
 function waitForNewImage(beforeTimestamp) {
   const t0 = beforeTimestamp || Date.now();
-  const maxWait = MEDIA_WAIT_MS;
-  const pollInterval = 1000;
   let waited = 0;
-
-  while (waited < maxWait) {
+  while (waited < MEDIA_WAIT_MS) {
     const latest = adbLatestFile(WHATSAPP_IMAGES_DIR);
     if (latest) {
-      const tsMatch = adbShell(`stat -c '%Y' "${latest}" 2>/dev/null || echo 0`);
-      const ts = parseInt(tsMatch, 10) * 1000;
+      const ts = parseInt(adbShell(`stat -c '%Y' "${latest}" 2>/dev/null || echo 0`), 10) * 1000;
       if (ts >= t0) return latest;
     }
-    new (require('events').EventEmitter)().setMaxListeners(100);
-    const start = Date.now();
-    while (Date.now() - start < pollInterval) {}
-    waited += pollInterval;
+    new EventEmitter().setMaxListeners(100);
+    const s = Date.now(); while (Date.now() - s < 1000) {} // poll interval
+    waited += 1000;
   }
   return null;
 }
 
 function pullImageFile(remotePath) {
-  const filename = path.basename(remotePath);
-  const localPath = `/tmp/wa-downloads/${filename}`;
-  fs.mkdirSync('/tmp/wa-downloads', { recursive: true });
-  adbPull(remotePath, localPath);
-  return localPath;
+  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) {
+function fanoutMedia(event, localPath, subscribers) {
   const token = `media_${Date.now()}_${++mediaTokenCounter}`;
-  const filename = path.basename(localPath);
-  mediaDownloads.set(token, { path: localPath, filename, createdAt: Date.now() });
-
-  const downloadUrl = `http://127.0.0.1:${PORT}/download/${token}`;
-  const eventWithUrl = { ...event, mediaUrl: downloadUrl, mediaFilename: filename };
-
-  for (const sub of subscribers) {
-    fetch(sub.url, {
-      method: 'POST',
-      headers: { 'Content-Type': 'application/json' },
-      body: JSON.stringify(eventWithUrl),
-    }).then((r) => {
-      console.log(`  media fanout to ${sub.url}: ${r.status}`);
-    }).catch((e) => {
-      console.log(`  media fanout to ${sub.url} failed: ${e.message}`);
-    });
+  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}`));
   }
 }
 
-// Cleanup old downloads every 5 min
 setInterval(() => {
   const cutoff = Date.now() - 10 * 60 * 1000;
-  for (const [token, entry] of mediaDownloads) {
-    if (entry.createdAt < cutoff) {
-      try { fs.unlinkSync(entry.path); } catch {}
-      mediaDownloads.delete(token);
-    }
-  }
+  for (const [t, e] of mediaDownloads) { if (e.createdAt < cutoff) { try { fs.unlinkSync(e.path); } catch {} mediaDownloads.delete(t); } }
 }, 5 * 60 * 1000);
 
-// ── Endpoints (Appium) ─────────────────────────────────────────
-
-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.get('/client/getChats/:session', async (req, res) => {
-  try {
-    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 nameEl = await row.$(
-          uia('resourceId("com.whatsapp:id/conversations_row_contact_name")')
-        );
-        const name = await nameEl.getText();
-
-        const msgEl = await row.$(uia('resourceId("com.whatsapp:id/single_msg_tv")'));
-        let lastMessage = '';
-        if (await msgEl.isExisting()) {
-          lastMessage = await msgEl.getText();
-        } else {
-          const msgFromEl = await row.$(uia('resourceId("com.whatsapp:id/msg_from_tv")'));
-          if (await msgFromEl.isExisting()) lastMessage = await msgFromEl.getText();
-        }
-
-        const timeEl = await row.$(
-          uia('resourceId("com.whatsapp:id/conversations_row_date")')
-        );
-        const timestamp = (await timeEl.isExisting()) ? await timeEl.getText() : '';
-
-        const unreadEl = await row.$(
-          uia('resourceId("com.whatsapp:id/conversations_row_unread_indicator")')
-        );
-        let unread = 0;
-        if (await unreadEl.isExisting()) {
-          const desc = await unreadEl.getAttribute('content-desc');
-          const m = desc ? desc.match(/(\d+)/) : null;
-          if (m) unread = parseInt(m[1], 10);
-        }
-
-        chats.push({ name, lastMessage, timestamp, unread });
-      } catch {}
-    }
-
-    res.json({ chats });
-  } catch (e) {
-    res.status(500).json({ error: e.message });
-  }
-});
-
-app.post('/chat/fetchMessages/:session', async (req, res) => {
-  try {
-    const { chatId } = req.body;
-    if (!chatId) return res.status(400).json({ error: 'chatId required' });
-
-    const d = await getDriver();
-    await ensureWhatsApp(d);
-    const query = chatId.replace(/@[cg]\.us$/, '');
-    await openChat(d, query);
-    await d.pause(2000);
-
-    const layoutEls = await d.$$(uia('resourceId("com.whatsapp:id/main_layout")'));
-    const messages = [];
-
-    for (const layout of layoutEls) {
-      try {
-        const textEl = await layout.$(uia('resourceId("com.whatsapp:id/message_text")'));
-        const dateEl = await layout.$(uia('resourceId("com.whatsapp:id/date")'));
-        if (await textEl.isExisting()) {
-          messages.push({
-            text: await textEl.getText(),
-            timestamp: (await dateEl.isExisting()) ? await dateEl.getText() : '',
-          });
-        }
-      } catch {}
-    }
-
-    messages.reverse();
-    goHome(d).catch(() => {});
-    res.json({ messages });
-  } catch (e) {
-    res.status(500).json({ error: e.message });
-  }
-});
-
-app.post('/client/sendMessage/:session', async (req, res) => {
-  try {
-    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 d = await getDriver();
-    await ensureWhatsApp(d);
-    const query = chatId.replace(/@[cg]\.us$/, '');
-    await openChat(d, query);
-
-    const input = await find(d, uia('resourceId("com.whatsapp:id/entry")'), 5000);
-    if (!input) throw new Error('Could not find message input');
-
-    await input.setValue(content);
-    await d.pause(500);
-
-    let sendBtn = await find(d, uia('resourceId("com.whatsapp:id/send")'), 3000);
-    if (!sendBtn) {
-      sendBtn = await find(d, uia('descriptionContains("Send")'), 2000);
-    }
-    if (!sendBtn) throw new Error('Could not find send button');
-    await sendBtn.click();
-
-    await d.pause(1000);
-    goHome(d).catch(() => {});
-    res.json({ status: 'sent', content });
-  } catch (e) {
-    res.status(500).json({ error: e.message });
-  }
-});
-
-// ── Media send endpoint ────────────────────────────────────────
-
-app.post('/messages/send-media', upload.single('image'), async (req, res) => {
-  let localPath = null;
-  try {
-    const { chatName, caption } = req.body;
-    const file = req.file;
-    if (!chatName) return res.status(400).json({ error: 'chatName required' });
-    if (!file) return res.status(400).json({ error: 'image file required' });
-
-    localPath = file.path;
-    const d = await getDriver();
-
-    await sendMediaViaShareIntent(d, chatName, localPath, file.mimetype, caption || '');
-
-    res.json({ status: 'sent', chatName, caption: caption || '' });
-  } catch (e) {
-    res.status(500).json({ error: e.message });
-  } finally {
-    if (localPath) {
-      try { fs.unlinkSync(localPath); } catch {}
-    }
-  }
-});
-
 // ── Webhook subscriber store ─────────────────────────────────────
 
 const subscribers = [];
 
 function fanout(event) {
-  for (const sub of subscribers) {
-    fetch(sub.url, {
-      method: 'POST',
-      headers: { 'Content-Type': 'application/json' },
-      body: JSON.stringify(event),
-    }).then((r) => {
-      console.log(`  webhook fanout to ${sub.url}: ${r.status}`);
-    }).catch((e) => {
-      console.log(`  webhook fanout to ${sub.url} failed: ${e.message}`);
-    });
+  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}`));
   }
 }
 
-// ── Webhook endpoints ─────────────────────────────────────────
+// ── 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' });
-  }
-
+  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' });
-  }
+  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',
-  };
+  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' });
-
-    setImmediate(async () => {
+    queue.add('media-pull', async () => {
       try {
-        const imagePath = waitForNewImage(event.timestamp);
-        if (!imagePath) {
-          console.log(`  media: no new image found for ${chatId} from ${sender}`);
-          fanout({ ...event, mediaStatus: 'not-found' });
-          return;
-        }
-        const localPath = pullImageFile(imagePath);
-        console.log(`  media: pulled ${imagePath} → ${localPath}`);
-        fanoutMedia({ ...event, mediaStatus: 'downloaded' }, localPath);
-      } catch (e) {
-        console.log(`  media processing error: ${e.message}`);
-        fanout({ ...event, mediaStatus: 'error', error: e.message });
-      }
+        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' });
@@ -514,10 +313,7 @@ app.post('/webhooks/whatsapp/incoming', (req, res) => {
 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() });
-  }
+  if (!subscribers.find((s) => s.url === url)) subscribers.push({ url, createdAt: new Date().toISOString() });
   res.json({ status: 'subscribed', url, activeSubscribers: subscribers.length });
 });
 
@@ -525,12 +321,101 @@ app.get('/webhooks/subscribers', (_req, res) => {
   res.json({ subscribers: subscribers.map((s) => ({ url: s.url, createdAt: s.createdAt })) });
 });
 
-// ── Media download endpoint ──────────────────────────────────
-
 app.get('/download/:token', (req, res) => {
-  const entry = mediaDownloads.get(req.params.token);
-  if (!entry) return res.status(404).json({ error: 'download expired or not found' });
-  res.download(entry.path, entry.filename);
+  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 ─────────────────────────────────────────────
@@ -539,8 +424,7 @@ fs.mkdirSync('/tmp/wa-uploads', { recursive: true });
 fs.mkdirSync('/tmp/wa-downloads', { recursive: true });
 
 app.listen(PORT, '0.0.0.0', () => {
-  const host = require('os').networkInterfaces();
   console.log(`wa-appium-api listening on http://0.0.0.0:${PORT}`);
-  console.log(`  Appium server: ${APPIUM_HOST}:${APPIUM_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'}`);
 });