feat: server-state baseline (Appium-fallback, event-dedupe, retry-semantics, webhook-reg, unlimited-limit)
This commit is contained in:
parent
c97d662856
commit
1d5a233f68
4 changed files with 129 additions and 8 deletions
12
src/index.js
12
src/index.js
|
|
@ -17,6 +17,17 @@ async function notifyTelegramOnline() {
|
||||||
console.error('Telegram startup notification failed:', err.message)
|
console.error('Telegram startup notification failed:', err.message)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
async function registerWhatsAppWebhook() {
|
||||||
|
const subscribeURL = process.env.WHATSAPP_WEBHOOK_SUBSCRIBE_URL
|
||||||
|
const targetURL = process.env.WHATSAPP_WEBHOOK_TARGET_URL
|
||||||
|
if (!subscribeURL || !targetURL) return
|
||||||
|
try {
|
||||||
|
await axios.post(subscribeURL, { url: targetURL }, { timeout: 5000 })
|
||||||
|
console.log(`Registered WhatsApp fallback webhook: ${targetURL}`)
|
||||||
|
} catch (err) {
|
||||||
|
console.error('WhatsApp fallback webhook registration failed:', err.message)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
const app = express()
|
const app = express()
|
||||||
app.use(express.json({ limit: '10mb' }))
|
app.use(express.json({ limit: '10mb' }))
|
||||||
|
|
@ -84,6 +95,7 @@ async function startServer() {
|
||||||
console.log(`Bot name: ${process.env.BOT_NAME}`)
|
console.log(`Bot name: ${process.env.BOT_NAME}`)
|
||||||
console.log('Daily summary cron job scheduled for 10:00')
|
console.log('Daily summary cron job scheduled for 10:00')
|
||||||
notifyTelegramOnline()
|
notifyTelegramOnline()
|
||||||
|
registerWhatsAppWebhook()
|
||||||
})
|
})
|
||||||
|
|
||||||
function shutdown(signal) {
|
function shutdown(signal) {
|
||||||
|
|
|
||||||
|
|
@ -1,23 +1,88 @@
|
||||||
|
const fs = require('fs')
|
||||||
|
const path = require('path')
|
||||||
|
const API_BASE = process.env.WHATSAPP_API_ENDPOINT
|
||||||
const { findDienst, getPngFilePaths } = require('./dienstService')
|
const { findDienst, getPngFilePaths } = require('./dienstService')
|
||||||
const { sendTextMessage, sendImageMessage, notifyAdmin, sendChatSeen } = require('./whatsappApi')
|
const { sendTextMessage, sendImageMessage, notifyAdmin, sendChatSeen } = require('./whatsappApi')
|
||||||
const { checkLimit, increment, MAX_REQUESTS_PER_DAY } = require('./rateLimiter')
|
const { checkLimit, increment, MAX_REQUESTS_PER_DAY } = require('./rateLimiter')
|
||||||
const { increment: incrementDaily } = require('./dailyStats')
|
const { increment: incrementDaily } = require('./dailyStats')
|
||||||
|
|
||||||
const ALLOWED_GROUP = process.env.ALLOWED_GROUP
|
const ALLOWED_GROUP = process.env.ALLOWED_GROUP
|
||||||
|
const ALLOWED_GROUP_NAMES = (process.env.ALLOWED_GROUP_NAME || '')
|
||||||
|
.split(',')
|
||||||
|
.map(name => name.trim())
|
||||||
|
.filter(Boolean)
|
||||||
|
const ALLOWED_CHAT_IDS = new Set([ALLOWED_GROUP, ...ALLOWED_GROUP_NAMES].filter(Boolean))
|
||||||
|
const appiumEventCooldown = 24 * 60 * 60 * 1000
|
||||||
|
const appiumEventStatePath = process.env.APPIUM_EVENT_STATE_PATH ||
|
||||||
|
path.join(__dirname, '..', 'appium-event-state.json')
|
||||||
|
let appiumEventState = { events: [] }
|
||||||
|
try {
|
||||||
|
appiumEventState = JSON.parse(fs.readFileSync(appiumEventStatePath, 'utf8'))
|
||||||
|
if (!Array.isArray(appiumEventState.events)) appiumEventState = { events: [] }
|
||||||
|
appiumEventState.events = appiumEventState.events.map(event => {
|
||||||
|
const parts = event.key.split('\u0000')
|
||||||
|
return {
|
||||||
|
key: [normalizeChatId(parts[0]), parts[1], parts[2]].join('\u0000'),
|
||||||
|
processedAt: event.processedAt || event.timestamp || 0,
|
||||||
|
}
|
||||||
|
})
|
||||||
|
} catch {
|
||||||
|
appiumEventState = { events: [] }
|
||||||
|
}
|
||||||
|
|
||||||
|
function normalizeChatId(chatId) {
|
||||||
|
return (chatId || '')
|
||||||
|
.replace(/^\u200e/, '')
|
||||||
|
.replace(/\s*\(\d+ berichten?\)$/, '')
|
||||||
|
.trim()
|
||||||
|
}
|
||||||
|
|
||||||
|
function getAppiumEventKey(payload) {
|
||||||
|
return [normalizeChatId(payload.chatId), payload.sender, payload.text].join('\u0000')
|
||||||
|
}
|
||||||
|
|
||||||
|
function isDuplicateAppiumEvent(payload) {
|
||||||
|
const now = Date.now()
|
||||||
|
appiumEventState.events = appiumEventState.events.filter(event =>
|
||||||
|
now - (event.processedAt || event.timestamp || 0) < appiumEventCooldown
|
||||||
|
)
|
||||||
|
return appiumEventState.events.some(event =>
|
||||||
|
event.key === getAppiumEventKey(payload)
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
function markAppiumEventProcessed(payload) {
|
||||||
|
appiumEventState.events.push({
|
||||||
|
key: getAppiumEventKey(payload),
|
||||||
|
processedAt: Date.now(),
|
||||||
|
})
|
||||||
|
appiumEventState.events = appiumEventState.events.slice(-500)
|
||||||
|
const temporaryPath = `${appiumEventStatePath}.tmp`
|
||||||
|
fs.writeFileSync(temporaryPath, JSON.stringify(appiumEventState))
|
||||||
|
fs.renameSync(temporaryPath, appiumEventStatePath)
|
||||||
|
}
|
||||||
function extractDienstnummer(body) {
|
function extractDienstnummer(body) {
|
||||||
const match = body.match(/^\s*(\d{4,5})\s*$/)
|
const match = body.match(/^\s*(\d{4,5})\s*$/)
|
||||||
return match ? match[1] : null
|
return match ? match[1] : null
|
||||||
}
|
}
|
||||||
|
|
||||||
async function handleWebhook(payload) {
|
async function handleWebhook(payload) {
|
||||||
if (payload.dataType !== 'message') return
|
const isWWebEvent = payload.dataType === 'message'
|
||||||
if (payload.sessionId !== process.env.WHATSAPP_SESSION) return
|
const isAppiumEvent = Boolean(payload.sender && payload.chatId)
|
||||||
|
if (!isWWebEvent && !isAppiumEvent) return
|
||||||
const message = payload.data?.message
|
if (isAppiumEvent && isDuplicateAppiumEvent(payload)) return
|
||||||
|
if (isWWebEvent && payload.sessionId !== process.env.WHATSAPP_SESSION) return
|
||||||
|
const message = isWWebEvent
|
||||||
|
? payload.data?.message
|
||||||
|
: {
|
||||||
|
fromMe: false,
|
||||||
|
from: payload.chatId,
|
||||||
|
author: payload.sender,
|
||||||
|
body: payload.text || '',
|
||||||
|
}
|
||||||
if (!message) return
|
if (!message) return
|
||||||
if (message.fromMe) return
|
if (message.fromMe) return
|
||||||
if (message.from !== ALLOWED_GROUP) return
|
if (!ALLOWED_CHAT_IDS.has(normalizeChatId(message.from))) return
|
||||||
|
|
||||||
const dienstnummer = extractDienstnummer(message.body || '')
|
const dienstnummer = extractDienstnummer(message.body || '')
|
||||||
if (!dienstnummer) return
|
if (!dienstnummer) return
|
||||||
|
|
@ -30,7 +95,9 @@ async function handleWebhook(payload) {
|
||||||
|
|
||||||
console.log(`Processing dienst ${dienstnummer} for ${message.author}`)
|
console.log(`Processing dienst ${dienstnummer} for ${message.author}`)
|
||||||
|
|
||||||
const author = message.author
|
const author = isWWebEvent
|
||||||
|
? message.author || message.from
|
||||||
|
: message.author
|
||||||
if (!author) return
|
if (!author) return
|
||||||
|
|
||||||
const limit = checkLimit(author)
|
const limit = checkLimit(author)
|
||||||
|
|
@ -45,25 +112,32 @@ async function handleWebhook(payload) {
|
||||||
const dienst = findDienst(dienstnummer)
|
const dienst = findDienst(dienstnummer)
|
||||||
if (!dienst) {
|
if (!dienst) {
|
||||||
await sendTextMessage(author, `Dienstnummer ${dienstnummer} niet gevonden.`)
|
await sendTextMessage(author, `Dienstnummer ${dienstnummer} niet gevonden.`)
|
||||||
|
if (isAppiumEvent) markAppiumEventProcessed(payload)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
const pngPaths = getPngFilePaths(dienst)
|
const pngPaths = getPngFilePaths(dienst)
|
||||||
if (!pngPaths || pngPaths.length === 0) {
|
if (!pngPaths || pngPaths.length === 0) {
|
||||||
await sendTextMessage(author, `Geen pagina's gevonden voor dienst ${dienstnummer}.`)
|
await sendTextMessage(author, `Geen pagina's gevonden voor dienst ${dienstnummer}.`)
|
||||||
|
if (isAppiumEvent) markAppiumEventProcessed(payload)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let sendFailed = false
|
||||||
for (const pngPath of pngPaths) {
|
for (const pngPath of pngPaths) {
|
||||||
const filename = pngPath.split('/').pop()
|
const filename = pngPath.split('/').pop()
|
||||||
try {
|
try {
|
||||||
await sendImageMessage(author, pngPath, filename)
|
await sendImageMessage(author, pngPath, filename)
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
|
sendFailed = true
|
||||||
const errorMsg = `Fout bij verzenden ${filename} naar ${author}: ${err.message}`
|
const errorMsg = `Fout bij verzenden ${filename} naar ${author}: ${err.message}`
|
||||||
console.error(errorMsg)
|
console.error(errorMsg)
|
||||||
await notifyAdmin(errorMsg)
|
if (API_BASE.includes(':3001')) {
|
||||||
|
await notifyAdmin(errorMsg)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if (!sendFailed && isAppiumEvent) markAppiumEventProcessed(payload)
|
||||||
}
|
}
|
||||||
|
|
||||||
module.exports = { handleWebhook, extractDienstnummer }
|
module.exports = { handleWebhook, extractDienstnummer }
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,7 @@
|
||||||
const MAX_REQUESTS_PER_DAY = parseInt(process.env.MAX_REQUESTS_PER_DAY, 10) || 5
|
const configuredLimit = parseInt(process.env.MAX_REQUESTS_PER_DAY, 10)
|
||||||
|
const MAX_REQUESTS_PER_DAY = Number.isFinite(configuredLimit) && configuredLimit > 0
|
||||||
|
? configuredLimit
|
||||||
|
: Infinity
|
||||||
|
|
||||||
const userCounts = new Map()
|
const userCounts = new Map()
|
||||||
let currentDate = new Date().toDateString()
|
let currentDate = new Date().toDateString()
|
||||||
|
|
|
||||||
|
|
@ -46,7 +46,38 @@ async function sendTextMessage(chatId, text) {
|
||||||
}, `sendTextMessage to ${chatId}`)
|
}, `sendTextMessage to ${chatId}`)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async function waitForAppiumJob(jobId) {
|
||||||
|
for (let attempt = 0; attempt < 120; attempt++) {
|
||||||
|
const response = await fetch(`${API_BASE}/jobs/${jobId}`)
|
||||||
|
if (!response.ok) throw new Error(`Appium job lookup failed: ${response.status}`)
|
||||||
|
const job = await response.json()
|
||||||
|
if (job.status === 'done') return job.result
|
||||||
|
if (job.status === 'failed') throw new Error(job.error || 'Appium job failed')
|
||||||
|
await sleep(1000)
|
||||||
|
}
|
||||||
|
throw new Error(`Appium job timed out: ${jobId}`)
|
||||||
|
}
|
||||||
|
|
||||||
async function sendImageMessage(chatId, filePath, filename) {
|
async function sendImageMessage(chatId, filePath, filename) {
|
||||||
|
if (API_BASE.includes(':3010')) {
|
||||||
|
return withRetry(async () => {
|
||||||
|
const fileBuffer = fs.readFileSync(filePath)
|
||||||
|
const form = new FormData()
|
||||||
|
form.append('image', new Blob([fileBuffer], { type: 'image/png' }), filename)
|
||||||
|
form.append('chatName', chatId.replace(/@[cg]\.us$/, ''))
|
||||||
|
const response = await fetch(`${API_BASE}/messages/send-media`, {
|
||||||
|
method: 'POST',
|
||||||
|
body: form
|
||||||
|
})
|
||||||
|
if (!response.ok) {
|
||||||
|
throw new Error(`sendImageMessage failed: ${response.status} ${response.statusText}`)
|
||||||
|
}
|
||||||
|
const queued = await response.json()
|
||||||
|
if (!queued.jobId) throw new Error('Appium did not return a job ID')
|
||||||
|
return waitForAppiumJob(queued.jobId)
|
||||||
|
}, `sendImageMessage ${filename} to ${chatId}`)
|
||||||
|
}
|
||||||
|
|
||||||
return withRetry(async () => {
|
return withRetry(async () => {
|
||||||
const fileBuffer = fs.readFileSync(filePath)
|
const fileBuffer = fs.readFileSync(filePath)
|
||||||
const base64Data = fileBuffer.toString('base64')
|
const base64Data = fileBuffer.toString('base64')
|
||||||
|
|
@ -86,6 +117,7 @@ async function notifyAdmin(errorText) {
|
||||||
}
|
}
|
||||||
|
|
||||||
async function sendChatSeen(chatId) {
|
async function sendChatSeen(chatId) {
|
||||||
|
if (API_BASE.includes(':3010')) return null
|
||||||
const url = `${API_BASE}/client/sendSeen/${SESSION}`
|
const url = `${API_BASE}/client/sendSeen/${SESSION}`
|
||||||
const response = await axios.post(url, { chatId })
|
const response = await axios.post(url, { chatId })
|
||||||
return response.data
|
return response.data
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue