|
|
|
|
@@ -3,10 +3,14 @@ import { getJob } from '../../services/storage/jobStorage.js';
|
|
|
|
|
import fetch from 'node-fetch';
|
|
|
|
|
import pThrottle from 'p-throttle';
|
|
|
|
|
import { normalizeImageUrl } from '../../utils.js';
|
|
|
|
|
import logger from '../../services/logger.js';
|
|
|
|
|
|
|
|
|
|
const RATE_LIMIT_INTERVAL = 1000;
|
|
|
|
|
const chatThrottleMap = new Map();
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Removes stale throttled call entries to keep memory bounded.
|
|
|
|
|
*/
|
|
|
|
|
function cleanupOldThrottles() {
|
|
|
|
|
const now = Date.now();
|
|
|
|
|
const maxAge = RATE_LIMIT_INTERVAL + 1000;
|
|
|
|
|
@@ -17,6 +21,15 @@ function cleanupOldThrottles() {
|
|
|
|
|
for (const chatId of toBeDeleted) chatThrottleMap.delete(chatId);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Return a throttled wrapper for a chatId to limit Telegram API calls.
|
|
|
|
|
* Uses p-throttle with 1 request per RATE_LIMIT_INTERVAL per chat.
|
|
|
|
|
*
|
|
|
|
|
* @template {Function} T
|
|
|
|
|
* @param {string|number} chatId
|
|
|
|
|
* @param {T} call - async function (endpoint: string, body: any) => Promise<Response>
|
|
|
|
|
* @returns {T}
|
|
|
|
|
*/
|
|
|
|
|
function getThrottled(chatId, call) {
|
|
|
|
|
cleanupOldThrottles();
|
|
|
|
|
const now = Date.now();
|
|
|
|
|
@@ -30,15 +43,38 @@ function getThrottled(chatId, call) {
|
|
|
|
|
return throttled;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Shorten a string to a maximum length with an ellipsis suffix.
|
|
|
|
|
* @param {string} str
|
|
|
|
|
* @param {number} [len=90]
|
|
|
|
|
* @returns {string}
|
|
|
|
|
*/
|
|
|
|
|
function shorten(str, len = 90) {
|
|
|
|
|
if (!str) return '';
|
|
|
|
|
return str.length > len ? str.substring(0, len).trim() + '...' : str;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Escape basic HTML entities for Telegram HTML parse mode.
|
|
|
|
|
* @param {string} [s='']
|
|
|
|
|
* @returns {string}
|
|
|
|
|
*/
|
|
|
|
|
function escapeHtml(s = '') {
|
|
|
|
|
return s.replace(/&/g, '&').replace(/</g, '<').replace(/>/g, '>');
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Build a Telegram photo caption (max 1024 characters) using HTML parse mode.
|
|
|
|
|
* @param {string} jobName
|
|
|
|
|
* @param {string} serviceName
|
|
|
|
|
* @param {Object} o - Listing object
|
|
|
|
|
* @param {string} [o.title]
|
|
|
|
|
* @param {string} [o.address]
|
|
|
|
|
* @param {string|number} [o.price]
|
|
|
|
|
* @param {string|number} [o.size]
|
|
|
|
|
* @param {string} [o.link]
|
|
|
|
|
* @returns {string}
|
|
|
|
|
*/
|
|
|
|
|
function buildCaption(jobName, serviceName, o) {
|
|
|
|
|
const title = shorten((o.title || '').replace(/\*/g, ''), 90);
|
|
|
|
|
const meta = [o.address, o.price, o.size].filter(Boolean).join(' | ');
|
|
|
|
|
@@ -47,6 +83,13 @@ function buildCaption(jobName, serviceName, o) {
|
|
|
|
|
)}'><b>${escapeHtml(title)}</b></a>\n${escapeHtml(meta)}`.slice(0, 1024);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Build a Telegram message text using HTML parse mode.
|
|
|
|
|
* @param {string} jobName
|
|
|
|
|
* @param {string} serviceName
|
|
|
|
|
* @param {Object} o - Listing object
|
|
|
|
|
* @returns {string}
|
|
|
|
|
*/
|
|
|
|
|
function buildText(jobName, serviceName, o) {
|
|
|
|
|
const title = shorten((o.title || '').replace(/\*/g, ''), 90);
|
|
|
|
|
const meta = [o.address, o.price, o.size].filter(Boolean).join(' | ');
|
|
|
|
|
@@ -57,8 +100,27 @@ function buildText(jobName, serviceName, o) {
|
|
|
|
|
);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
export const send = ({ serviceName, newListings, notificationConfig, jobKey }) => {
|
|
|
|
|
const { token, chatId } = notificationConfig.find((adapter) => adapter.id === config.id).fields;
|
|
|
|
|
/**
|
|
|
|
|
* Send new listings to Telegram.
|
|
|
|
|
* - Respects per-chat Telegram rate limits using a lightweight throttle cache.
|
|
|
|
|
* - Falls back to sendMessage when sendPhoto fails or image is missing.
|
|
|
|
|
*
|
|
|
|
|
* @param {Object} params
|
|
|
|
|
* @param {string} params.serviceName - Name of the crawler/service producing the listings.
|
|
|
|
|
* @param {Array<Object>} params.newListings - Array of new listing objects.
|
|
|
|
|
* @param {Array<Object>} params.notificationConfig - Notification adapters configuration array.
|
|
|
|
|
* @param {string} params.jobKey - Storage job key to resolve the human readable job name.
|
|
|
|
|
* @returns {Promise<Array<Response>>} Promise resolving when all send operations complete.
|
|
|
|
|
*/
|
|
|
|
|
export const send = ({ serviceName, newListings = [], notificationConfig, jobKey }) => {
|
|
|
|
|
const adapterCfg = notificationConfig.find((adapter) => adapter.id === config.id);
|
|
|
|
|
if (!adapterCfg || !adapterCfg.fields) {
|
|
|
|
|
throw new Error(`Telegram adapter configuration missing for job '${jobKey || ''}'`);
|
|
|
|
|
}
|
|
|
|
|
const { token, chatId } = adapterCfg.fields;
|
|
|
|
|
if (!token || !chatId) {
|
|
|
|
|
throw new Error("Telegram 'token' and 'chatId' must be provided in notification config");
|
|
|
|
|
}
|
|
|
|
|
const job = getJob(jobKey);
|
|
|
|
|
const jobName = job == null ? jobKey : job.name;
|
|
|
|
|
|
|
|
|
|
@@ -68,9 +130,16 @@ export const send = ({ serviceName, newListings, notificationConfig, jobKey }) =
|
|
|
|
|
body: JSON.stringify(body),
|
|
|
|
|
headers: { 'Content-Type': 'application/json' },
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
if (!res.ok) {
|
|
|
|
|
const errorBody = await res.text();
|
|
|
|
|
throw new Error(`API error for '${jobName}'. '${endpoint}' returned ${errorBody}`);
|
|
|
|
|
}
|
|
|
|
|
return res;
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
if (!Array.isArray(newListings) || newListings.length === 0) return Promise.resolve([]);
|
|
|
|
|
|
|
|
|
|
const promises = newListings.map(async (o) => {
|
|
|
|
|
const img = normalizeImageUrl(o.image);
|
|
|
|
|
const textPayload = {
|
|
|
|
|
@@ -81,28 +150,32 @@ export const send = ({ serviceName, newListings, notificationConfig, jobKey }) =
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
if (!img) {
|
|
|
|
|
return throttledCall('sendMessage', textPayload);
|
|
|
|
|
return await throttledCall('sendMessage', textPayload).catch(async (e) => {
|
|
|
|
|
logger.error(`Error sending message to Telegram: ${e.message}`);
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
try {
|
|
|
|
|
return await throttledCall('sendPhoto', {
|
|
|
|
|
chat_id: chatId,
|
|
|
|
|
photo: img,
|
|
|
|
|
caption: buildCaption(jobName, serviceName, o),
|
|
|
|
|
parse_mode: 'HTML',
|
|
|
|
|
return await throttledCall('sendPhoto', {
|
|
|
|
|
chat_id: chatId,
|
|
|
|
|
photo: img,
|
|
|
|
|
caption: buildCaption(jobName, serviceName, o),
|
|
|
|
|
parse_mode: 'HTML',
|
|
|
|
|
}).catch(async (e) => {
|
|
|
|
|
logger.error(`Error sending photo to Telegram and use a fallback: ${e.message}`);
|
|
|
|
|
return await throttledCall('sendMessage', textPayload).catch((e) => {
|
|
|
|
|
logger.error(`Error sending message to Telegram: ${e.message}`);
|
|
|
|
|
throw e;
|
|
|
|
|
});
|
|
|
|
|
} catch (e) {
|
|
|
|
|
// If we see a timeout due to sending an image, try sending it without
|
|
|
|
|
if (e && (e.code === 'ETIMEDOUT' || e.errno === 'ETIMEDOUT')) {
|
|
|
|
|
return throttledCall('sendMessage', textPayload);
|
|
|
|
|
}
|
|
|
|
|
throw e;
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
return Promise.all(promises);
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Telegram notification adapter configuration schema.
|
|
|
|
|
* @type {{id:string,name:string,readme:string,description:string,fields:{token:{type:string,label:string,description:string},chatId:{type:string,label:string,description:string}}}}
|
|
|
|
|
*/
|
|
|
|
|
export const config = {
|
|
|
|
|
id: 'telegram',
|
|
|
|
|
name: 'Telegram',
|
|
|
|
|
|