diff --git a/broadcaster/README.md b/broadcaster/README.md index df84da1..cc5cd16 100644 --- a/broadcaster/README.md +++ b/broadcaster/README.md @@ -17,12 +17,13 @@ if Telegram or the service goes down, the pool is unaffected. | | when | |---|---| -| **Daily digest** | once per UTC day at `DIGEST_UTC_HOUR` (default 12): hashrate 1h/24h, active workers, shares and reject rate, best share, blocks in the last 24h and all time, mode and fee | +| **Daily digest** | once per UTC day at `DIGEST_UTC_HOUR` (default 12): hashrate now (5m), 1h and 24h, active workers, accepted and rejected shares, best share, blocks in the last 24h and all time, mode and fee | | **Block found** | the first time a block appears in `/api/blocks` as `pending` or `confirmed` | | **Block lost** | a block it announced turns `orphaned` or `rejected` | | **Pool quiet / back** | no share accepted for `STALE_SHARES_SEC` (default 15m), and when shares resume | | **Health failing / recovered** | `/health` changes state, with the failing checks listed | -| **Pinned live message** (`BROADCASTER_LIVE=1`) | one message, pinned, edited every `BROADCASTER_LIVE_MS` (default 5m) with current stats | +| **Pinned live message** (`BROADCASTER_LIVE=1`) | one message, pinned unless `BROADCASTER_LIVE_PIN=0`, edited every `BROADCASTER_LIVE_MS` (default 5m) with current stats and the last block | +| **`/pool_status`** (`BROADCASTER_COMMANDS=1`) | when someone sends it in the chat: the current stats, as a reply in the same topic, at most once per `BROADCASTER_COMMAND_COOLDOWN_SEC` (default 30) | | **Periodic update** (`BROADCASTER_SUMMARY_HOURS=N`) | the same stats as a new message every N hours, on UTC boundaries (6 → 00, 06, 12, 18 UTC) | - **Edited is not posted.** Telegram does not notify anyone of an edit or @@ -58,6 +59,23 @@ appear under the channel's name and avatar. (`23563`). In a group, the bot posts as itself and needs no admin rights when members may send (and, for the live message, pin) messages. +### `/pool_status` + +With `BROADCASTER_COMMANDS=1` the service also reads the chat, with +`getUpdates`, and answers `/pool_status` with the live message's numbers +plus the last block. It answers only in `TELEGRAM_CHAT_ID`, so adding the +bot to another group does not make it answer there. + +- It registers the command in the bot's `/` menu. A bot in privacy mode + (BotFather's default) always sees `/pool_status@yourbot`, which is what a + tap in that menu sends, but not always a bare `/pool_status`. Turn privacy + off (`/setprivacy` → Disable) if people type it by hand. +- Telegram allows one `getUpdates` reader per bot and none while a webhook + is set. Run one broadcaster per bot, and don't poll the bot from anywhere + else while it runs. +- The first start skips commands sent before it, and the read position is + kept in `broadcaster.json`, so a restart answers nothing twice. + Try it without a token first: ```bash @@ -82,7 +100,9 @@ Environment only. The full list with defaults is in | `BROADCASTER_POLL_MS` | default 60000 | | `DIGEST_UTC_HOUR` | 0–23, or `off` (default 12) | | `BROADCASTER_LIVE`, `BROADCASTER_LIVE_MS` | pinned live message (default off, 300000) | +| `BROADCASTER_LIVE_PIN` | `0` = post the live message without pinning it (default 1) | | `BROADCASTER_SUMMARY_HOURS` | 1–168, or `off` (default off) | +| `BROADCASTER_COMMANDS`, `BROADCASTER_COMMAND_COOLDOWN_SEC` | answer `/pool_status` (default off, 30) | | `STALE_SHARES_SEC` | default 900 | | `HEALTH_DEBOUNCE` | default 3 | diff --git a/broadcaster/index.js b/broadcaster/index.js index 7687b0e..a65d259 100644 --- a/broadcaster/index.js +++ b/broadcaster/index.js @@ -3,6 +3,7 @@ * Reads the dashboard's public JSON API and posts to one channel: a daily * digest, found / orphaned blocks, the pool going quiet and coming back, * ledger health changes, and optionally a pinned message kept current. + * With BROADCASTER_COMMANDS=1 it also answers /pool_status in that chat. * * Config is environment-only — see lib/config.js for the full list. * @@ -17,10 +18,11 @@ import { readFileSync } from 'node:fs'; import { loadConfig } from './lib/config.js'; -import { loadState } from './lib/state.js'; +import { loadState, saveState } from './lib/state.js'; import { DashboardClient } from './lib/dashboard.js'; import { TelegramClient, DryRunClient } from './lib/telegram.js'; import { Broadcaster } from './lib/broadcaster.js'; +import { Commands } from './lib/commands.js'; const cfg = loadConfig(); const { version } = JSON.parse(readFileSync(new URL('./package.json', import.meta.url), 'utf8')); @@ -36,17 +38,29 @@ log.info(`simplepool-broadcaster ${version} starting ` + `${cfg.threadId != null ? ` topic=${cfg.threadId}` : ''} state=${cfg.statePath}` + `${cfg.dryRun ? ' DRY RUN' : ''})`); log.info(` poll ${cfg.pollMs}ms, digest ${cfg.digestHour == null ? 'off' : `${cfg.digestHour}:00 UTC`}, ` + - `live ${cfg.live ? `every ${cfg.liveMs}ms` : 'off'}, ` + + `live ${cfg.live ? `every ${cfg.liveMs}ms${cfg.livePin ? '' : ' (unpinned)'}` : 'off'}, ` + `summary ${cfg.summaryHours == null ? 'off' : `every ${cfg.summaryHours}h`}, ` + `stale after ${cfg.staleSharesSec}s`); -const broadcaster = new Broadcaster({ - dashboard: new DashboardClient({ url: cfg.dashboardUrl }), - telegram: cfg.dryRun ? new DryRunClient() : new TelegramClient({ token: cfg.token, chatId: cfg.chatId, threadId: cfg.threadId }), - state: loadState(cfg.statePath), - cfg, - log, -}); +// One state object and one writer for both loops, so neither can save over +// the other's changes with a stale copy. +const state = loadState(cfg.statePath); +const persist = () => saveState(cfg.statePath, state); +const dashboard = new DashboardClient({ url: cfg.dashboardUrl }); +const telegram = cfg.dryRun + ? new DryRunClient() + : new TelegramClient({ token: cfg.token, chatId: cfg.chatId, threadId: cfg.threadId }); + +const broadcaster = new Broadcaster({ dashboard, telegram, state, cfg, log, persist }); + +if (cfg.commands && cfg.dryRun) { + log.info(' commands: off under BROADCASTER_DRY_RUN (they need a real bot to read messages)'); +} else if (cfg.commands) { + const commands = new Commands({ dashboard, telegram, state, cfg, log, persist }); + // Its own loop: a long poll must never delay a post, and a failure here + // must never stop the posts. + commands.run().catch((e) => log.error(`commands stopped: ${e.message}`)); +} let lastTickError = null; async function loop() { diff --git a/broadcaster/lib/broadcaster.js b/broadcaster/lib/broadcaster.js index 33bcf26..2441772 100644 --- a/broadcaster/lib/broadcaster.js +++ b/broadcaster/lib/broadcaster.js @@ -7,7 +7,8 @@ * recovering, once the new state has held for `debounce` polls * - posts the daily digest once per UTC day, at or after DIGEST_UTC_HOUR * - posts a stats summary once per BROADCASTER_SUMMARY_HOURS period - * - refreshes the pinned live message (BROADCASTER_LIVE=1) + * - refreshes the live message (BROADCASTER_LIVE=1), pinned unless + * BROADCASTER_LIVE_PIN=0 * * State is saved after every post, so a failure halfway through a tick * retries only what was not sent. @@ -45,9 +46,9 @@ export class Broadcaster { await this.#liveness(status, nowSec); await this.#health(status); await this.#digest(status, blocks, nowMs); - await this.#summary(status, nowMs); + await this.#summary(status, blocks, nowMs); } - await this.#live(status, nowMs); + await this.#live(status, blocks, nowMs); } /* First run: remember everything as already said. */ @@ -115,13 +116,14 @@ export class Broadcaster { /* A new message per period, unlike the live one, which is edited. The * first period seen (first run, or the option just turned on) is * recorded without a post, so enabling it does not post on the spot. */ - async #summary(status, nowMs) { + async #summary(status, blocks, nowMs) { if (this.cfg.summaryHours == null) return; const slot = this.#summarySlot(nowMs); if (this.state.lastSummarySlot === slot) return; if (this.state.lastSummarySlot != null) { await this.#post(msg.summary({ status, + blocks, nowSec: Math.floor(nowMs / 1000), flowing: this.state.sharesFlowing !== false, ...this.#ctx(), @@ -135,10 +137,11 @@ export class Broadcaster { return Math.floor(nowMs / (this.cfg.summaryHours * 3_600_000)); } - async #live(status, nowMs) { + async #live(status, blocks, nowMs) { if (!this.cfg.live || nowMs - this.state.liveUpdatedAt < this.cfg.liveMs) return; const html = msg.live({ status, + blocks, nowSec: Math.floor(nowMs / 1000), flowing: this.state.sharesFlowing !== false, ...this.#ctx(), @@ -156,10 +159,15 @@ export class Broadcaster { if (this.state.liveMessageId == null) { this.state.liveMessageId = await this.telegram.send(html); this.persist(); - try { - await this.telegram.pin(this.state.liveMessageId); - } catch (e) { - this.log.warn(`could not pin the live message (does the bot have "Pin messages"?): ${e.message}`); + // BROADCASTER_LIVE_PIN=0: a chat where the bot may post but not + // pin keeps the message, without a warning on every new one. + if (this.cfg.livePin !== false) { + try { + await this.telegram.pin(this.state.liveMessageId); + } catch (e) { + this.log.warn(`could not pin the live message (does the bot have "Pin messages"? ` + + `BROADCASTER_LIVE_PIN=0 stops trying): ${e.message}`); + } } } this.state.liveUpdatedAt = nowMs; diff --git a/broadcaster/lib/commands.js b/broadcaster/lib/commands.js new file mode 100644 index 0000000..d1f9ea8 --- /dev/null +++ b/broadcaster/lib/commands.js @@ -0,0 +1,143 @@ +/* /pool_status: answer a command in the chat with the current stats. + * + * Runs beside the poster, on its own long-poll loop (BROADCASTER_COMMANDS=1). + * Only TELEGRAM_CHAT_ID is answered: the bot can be added anywhere, and a + * stranger's group must not get a reply. The answer goes to the topic the + * command was typed in, as a reply to it, and at most once per + * BROADCASTER_COMMAND_COOLDOWN_SEC per chat, so the command cannot be used + * to flood the chat. + * + * A bot in privacy mode (the default) is shown "/pool_status@" but not + * always a bare "/pool_status". Registering the command puts it in the "/" + * menu, and a tap there sends the addressed form. + * + * The update offset is saved in broadcaster.json, so a restart neither + * answers a command twice nor answers commands typed while it was down: the + * very first start skips the backlog. + */ + +import * as msg from './messages.js'; + +export const COMMANDS = [ + { command: 'pool_status', description: 'Current pool stats' }, +]; + +const RETRY_MS = 5000; + +export class Commands { + constructor({ dashboard, telegram, state, cfg, log, persist, sleep = defaultSleep }) { + this.dashboard = dashboard; + this.telegram = telegram; + this.state = state; + this.cfg = cfg; + this.log = log; + this.persist = persist; + this.sleep = sleep; + this.username = null; + this.lastReply = new Map(); // chat id -> ms of the last reply + this.stopped = false; + } + + /* getMe, register the menu, skip the backlog on a first start. */ + async init() { + this.username = (await this.telegram.getMe()).username ?? null; + try { + await this.telegram.setMyCommands(COMMANDS); + } catch (e) { + this.log.warn(`could not register the command menu: ${e.message}`); + } + if (this.state.updateOffset == null) { + const pending = await this.telegram.getUpdates({ offset: -1, timeoutSec: 0 }); + this.state.updateOffset = pending.length ? pending[pending.length - 1].update_id + 1 : 0; + this.persist(); + } + this.log.info(`commands: answering /pool_status in ${this.cfg.chatId} as @${this.username}`); + } + + /* One long poll, and every message it brought. */ + async pollOnce(clock = Date.now) { + const updates = await this.telegram.getUpdates({ offset: this.state.updateOffset, timeoutSec: 25 }); + for (const u of updates) { + // Advanced before handling: a reply that fails is not retried, + // because the person can ask again and a stale answer is worse. + this.state.updateOffset = u.update_id + 1; + this.persist(); + if (u.message) await this.handle(u.message, clock()); + } + } + + async run() { + let ready = false; + let lastError = null; + while (!this.stopped) { + try { + // Retried like a poll: Telegram being unreachable at start + // must not switch the command off until the next restart. + if (!ready) { await this.init(); ready = true; } + await this.pollOnce(); + lastError = null; + } catch (e) { + if (e.message !== lastError) this.log.warn(`commands: ${e.message}`); + lastError = e.message; + await this.sleep(RETRY_MS); + } + } + } + + async handle(m, nowMs) { + if (!this.#isPoolStatus(m.text) || !this.#allowedChat(m.chat)) return; + + const chatKey = String(m.chat.id); + const last = this.lastReply.get(chatKey); + if (last != null && nowMs - last < this.cfg.commandCooldownSec * 1000) return; + this.lastReply.set(chatKey, nowMs); + + const where = { + chatId: m.chat.id, + // In a forum, a command typed in a topic is answered there; one + // typed in General carries no topic. + threadId: m.is_topic_message ? m.message_thread_id : null, + replyTo: m.message_id, + }; + let html; + try { + const [status, blocks] = await Promise.all([this.dashboard.status(), this.dashboard.blocks()]); + const nowSec = Math.floor(nowMs / 1000); + html = msg.poolStatus({ + status, blocks, nowSec, + flowing: this.#flowing(status, nowSec), + poolName: this.cfg.poolName, publicUrl: this.cfg.publicUrl, + }); + } catch (e) { + this.log.warn(`commands: /pool_status could not read the dashboard: ${e.message}`); + html = msg.statusUnavailable({ poolName: this.cfg.poolName }); + } + await this.telegram.send(html, where); + } + + /* "/pool_status", "/pool_status@this_bot", with or without arguments; + * never "/pool_status@another_bot". */ + #isPoolStatus(text) { + const match = /^\/pool_status(?:@(\w+))?(?:\s|$)/i.exec(text ?? ''); + if (!match) return false; + return match[1] == null || this.username == null + || match[1].toLowerCase() === this.username.toLowerCase(); + } + + #allowedChat(chat) { + const want = String(this.cfg.chatId); + if (String(chat.id) === want) return true; + return chat.username != null && want.toLowerCase() === `@${chat.username}`.toLowerCase(); + } + + /* Same rule as the poster's liveness check, without its debounce: this + * answers about right now. */ + #flowing(status, nowSec) { + const last = status.pool?.last_share_ts; + return last != null && nowSec - last < this.cfg.staleSharesSec; + } +} + +function defaultSleep(ms) { + return new Promise((r) => setTimeout(r, ms)); +} diff --git a/broadcaster/lib/config.js b/broadcaster/lib/config.js index d8391c8..4841bf3 100644 --- a/broadcaster/lib/config.js +++ b/broadcaster/lib/config.js @@ -31,6 +31,14 @@ * with current stats (default off) * BROADCASTER_LIVE_MS how often the pinned message is refreshed * (default 300000) + * BROADCASTER_LIVE_PIN '0' = post the live message without pinning it, + * for a chat where the bot may not pin (default 1) + * BROADCASTER_COMMANDS '1' = answer /pool_status in TELEGRAM_CHAT_ID with + * the current stats (default off). Reads messages + * with getUpdates, so nothing else may poll this + * bot and it must have no webhook. + * BROADCASTER_COMMAND_COOLDOWN_SEC at most one reply per chat this often + * (default 30), so the command cannot flood a chat * BROADCASTER_SUMMARY_HOURS also post the current stats as a NEW message * every N hours (1-168), on UTC boundaries: 6 means * 00, 06, 12 and 18 UTC (exact for any N that @@ -90,6 +98,9 @@ export function loadConfig() { pollMs: num('BROADCASTER_POLL_MS', 60000, { min: 1000 }), digestHour: digestRaw === 'off' ? null : num('DIGEST_UTC_HOUR', 12, { min: 0, max: 23 }), live: process.env.BROADCASTER_LIVE === '1', + livePin: process.env.BROADCASTER_LIVE_PIN !== '0', + commands: process.env.BROADCASTER_COMMANDS === '1', + commandCooldownSec: num('BROADCASTER_COMMAND_COOLDOWN_SEC', 30, { min: 0 }), liveMs: num('BROADCASTER_LIVE_MS', 300000, { min: 60000 }), summaryHours: summaryRaw == null || summaryRaw === 'off' || summaryRaw === '0' ? null : num('BROADCASTER_SUMMARY_HOURS', null, { min: 1, max: 168 }), diff --git a/broadcaster/lib/messages.js b/broadcaster/lib/messages.js index 3bb4d5a..84c32a7 100644 --- a/broadcaster/lib/messages.js +++ b/broadcaster/lib/messages.js @@ -63,12 +63,15 @@ function lines(...xs) { return xs.filter((x) => x != null && x !== false).join('\n'); } -/* The core stat block shared by the digest and the pinned live message. */ +/* The core stat block shared by the digest, the live message, the periodic + * update and /pool_status. "Current" is the dashboard's 5-minute rate. */ function statLines(p) { return [ - `Hashrate: ${fmtHashrate(p.hashrate_1h)} (1h) · ${fmtHashrate(p.hashrate)} (24h)`, + `Hashrate now: ${fmtHashrate(p.hashrate_5m)} (5m)`, + `Hashrate: ${fmtHashrate(p.hashrate_1h)} (1h) · ${fmtHashrate(p.hashrate)} (24h)`, `Active workers (24h): ${fmtN(p.workers_active)}`, - `Shares (24h): ${fmtN(p.accepted)} accepted · ${(p.reject_rate_pct ?? 0).toFixed(2)}% rejected`, + `Shares (24h): ${fmtN(p.accepted)} accepted · ${fmtN(p.rejected)} rejected ` + + `(${(p.reject_rate_pct ?? 0).toFixed(2)}%)`, `Best share (24h): ${fmtDiff(p.best_share_24h)}`, ]; } @@ -98,9 +101,19 @@ export function digest({ status, blocks, poolName, publicUrl, nowSec }) { ); } -/* The pinned message and the periodic summary show the same numbers; only - * the heading and the time line differ. */ -function snapshot({ status, poolName, publicUrl, nowSec, flowing, title, timeLabel }) { +/* The newest block that still counts: pending or confirmed. An orphaned or + * rejected one is not the pool's last block. `blocks` is newest first. */ +export function lastBlockLine(blocks, nowSec) { + const b = (blocks ?? []).find((x) => x.status === 'pending' || x.status === 'confirmed'); + if (!b) return 'Last block: none yet'; + const ago = b.ts != null ? ` · ${fmtAgo(nowSec - b.ts)} ago` : ''; + return `Last block: #${fmtN(b.height)}${ago}`; +} + +/* The pinned message, the periodic summary and the /pool_status reply show + * the same numbers; only the heading and the time line differ. The last + * block line appears whenever the blocks list was read. */ +function snapshot({ status, blocks, poolName, publicUrl, nowSec, flowing, title, timeLabel }) { const p = status.pool; const lastShare = p.last_share_ts ? fmtAgo(nowSec - p.last_share_ts) + ' ago' : 'never'; return lines( @@ -108,6 +121,7 @@ function snapshot({ status, poolName, publicUrl, nowSec, flowing, title, timeLab '', ...statLines(p), `Last share: ${lastShare}`, + blocks !== undefined ? lastBlockLine(blocks, nowSec) : null, `Blocks (all time): ${fmtN(p.blocks_lifetime)}`, '', `${timeLabel} ${new Date(nowSec * 1000).toISOString().slice(11, 16)} UTC`, @@ -123,6 +137,14 @@ export function summary({ flowing = true, ...rest }) { return snapshot({ ...rest, flowing, title: 'update', timeLabel: 'As of' }); } +export function poolStatus({ flowing = true, ...rest }) { + return snapshot({ ...rest, flowing, title: 'status', timeLabel: 'As of' }); +} + +export function statusUnavailable({ poolName }) { + return `⚠️ ${esc(poolName)}: the pool's status can't be read right now. Try again in a minute.`; +} + export function blockFound({ block, poolName, publicUrl }) { const reward = block.reward_sats != null ? ` · reward ${fmtBtc(block.reward_sats)}` : ''; return lines( diff --git a/broadcaster/lib/state.js b/broadcaster/lib/state.js index 88bc23d..dabc15e 100644 --- a/broadcaster/lib/state.js +++ b/broadcaster/lib/state.js @@ -9,7 +9,9 @@ * lastDigestDate 'YYYY-MM-DD' (UTC) of the last daily digest * lastSummarySlot index of the last BROADCASTER_SUMMARY_HOURS period a * summary was posted for (ms since epoch / period) - * liveMessageId the pinned message edited in place + * liveMessageId the live message edited in place + * updateOffset next Telegram update id to read (BROADCASTER_COMMANDS=1); + * null until the first start skips the backlog * liveUpdatedAt ms of its last refresh * sharesFlowing last announced liveness (true/false/null) * healthOk last announced ledger-health state (true/false/null) @@ -32,6 +34,7 @@ export function emptyState() { lastDigestDate: null, lastSummarySlot: null, liveMessageId: null, + updateOffset: null, liveUpdatedAt: 0, sharesFlowing: null, healthOk: null, diff --git a/broadcaster/lib/telegram.js b/broadcaster/lib/telegram.js index 4b262e5..0f84f56 100644 --- a/broadcaster/lib/telegram.js +++ b/broadcaster/lib/telegram.js @@ -1,4 +1,5 @@ -/* Minimal Telegram Bot API client: the three calls a channel poster needs. +/* Minimal Telegram Bot API client: what a channel poster needs, plus the + * long poll that brings in /pool_status (BROADCASTER_COMMANDS=1). * * Every call goes through one queue, spaced MIN_GAP_MS apart. Telegram allows * about 20 posts a minute into one chat; the broadcaster never comes close, @@ -33,11 +34,17 @@ export class TelegramClient { } /* Returns the new message's id. Only a send names the topic: edits and - * pins address a message id, which already belongs to one. */ - async send(html) { + * pins address a message id, which already belongs to one. + * + * A command reply overrides the destination: `chatId` and `threadId` are + * where the command was typed, and `replyTo` quotes it. */ + async send(html, { chatId = this.chatId, threadId = this.threadId, replyTo = null } = {}) { const r = await this.call('sendMessage', { - chat_id: this.chatId, - ...(this.threadId != null && { message_thread_id: this.threadId }), + chat_id: chatId, + ...(threadId != null && { message_thread_id: threadId }), + ...(replyTo != null && { + reply_parameters: { message_id: replyTo, allow_sending_without_reply: true }, + }), text: html, parse_mode: 'HTML', link_preview_options: { is_disabled: true }, @@ -69,6 +76,28 @@ export class TelegramClient { }); } + async getMe() { + return this.call('getMe', {}); + } + + /* Puts the commands in the "/" menu, so a tap sends "/pool_status@bot", + * which a bot in privacy mode is always shown. */ + async setMyCommands(commands) { + await this.call('setMyCommands', { commands }); + } + + /* Long poll for new messages. NOT queued: it holds the connection open + * for up to `timeoutSec`, and a post must not wait behind it. Telegram + * allows one getUpdates at a time per bot, and none while a webhook is + * set (409 Conflict). */ + async getUpdates({ offset, timeoutSec = 25 }) { + const { res, json } = await this.#request('getUpdates', { + offset, timeout: timeoutSec, allowed_updates: ['message'], + }, AbortSignal.timeout((timeoutSec + 15) * 1000)); + if (json.ok) return json.result; + throw new TelegramError(json.error_code ?? res.status, json.description ?? 'unknown error'); + } + call(method, body) { const run = this.queue.then(() => this.#callNow(method, body)); this.queue = run.catch(() => {}); @@ -81,18 +110,7 @@ export class TelegramClient { if (wait > 0) await this.sleep(wait); this.lastSent = Date.now(); - let res, json; - try { - res = await this.fetch(`${this.base}/${method}`, { - method: 'POST', - headers: { 'content-type': 'application/json' }, - body: JSON.stringify(body), - }); - json = await res.json(); - } catch (e) { - // fetch errors can embed the URL; keep only the cause. - throw new TelegramError('network', e.cause?.code || e.name || 'request failed'); - } + const { res, json } = await this.#request(method, body); if (json.ok) return json.result; const retryAfter = json.parameters?.retry_after; @@ -103,6 +121,21 @@ export class TelegramClient { throw new TelegramError(json.error_code ?? res.status, json.description ?? 'unknown error'); } } + + async #request(method, body, signal) { + try { + const res = await this.fetch(`${this.base}/${method}`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(body), + ...(signal && { signal }), + }); + return { res, json: await res.json() }; + } catch (e) { + // fetch errors can embed the URL; keep only the cause. + throw new TelegramError('network', e.cause?.code || e.name || 'request failed'); + } + } } /* Stands in for TelegramClient under BROADCASTER_DRY_RUN=1. */ @@ -111,9 +144,9 @@ export class DryRunClient { this.out = out; this.nextId = 1; } - async send(html) { + async send(html, { replyTo = null } = {}) { const id = this.nextId++; - this.out(`--- post #${id} ---\n${html}\n`); + this.out(`--- post #${id}${replyTo != null ? ` (reply to #${replyTo})` : ''} ---\n${html}\n`); return id; } async edit(messageId, html) { this.out(`--- edit #${messageId} ---\n${html}\n`); } diff --git a/broadcaster/test/broadcaster.test.js b/broadcaster/test/broadcaster.test.js index 8cd3a0d..7624ee8 100644 --- a/broadcaster/test/broadcaster.test.js +++ b/broadcaster/test/broadcaster.test.js @@ -29,7 +29,7 @@ function status({ lastShareTs, healthOk = true, healthStatus } = {}) { pool: { db_ready: true, hashrate: 1e12, hashrate_1h: 2e12, hashrate_5m: 3e12, - workers_active: 4, accepted: 1000, reject_rate_pct: 0.5, + workers_active: 4, accepted: 1000, rejected: 5, reject_rate_pct: 0.5, best_share_24h: 123456, blocks_lifetime: 7, last_share_ts: lastShareTs, mode: 'pps-classic', fee_bps: 100, }, @@ -179,7 +179,8 @@ test('digest posts once per UTC day at or after the configured hour', async () = const digests = telegram.posts.filter((p) => p.includes('daily report')); assert.equal(digests.length, 1); assert.match(digests[0], /2026-09-30/); - assert.match(digests[0], /2\.00 TH\/s<\/b> \(1h\)/); + assert.match(digests[0], /Hashrate now: 3\.00 TH\/s<\/b> \(5m\)/); + assert.match(digests[0], /2\.00 TH\/s \(1h\)/); assert.match(digests[0], /pps-classic · fee 1\.00%/); assert.match(digests[0], /href="https:\/\/pool\.example\/"/); @@ -244,7 +245,9 @@ test('summary posts a new message once per period, on UTC boundaries', async () assert.equal(telegram.posts.length, 1); assert.match(telegram.posts[0], /testpool — update/); assert.match(telegram.posts[0], /As of 12:00 UTC/); - assert.match(telegram.posts[0], /Hashrate: 2\.00 TH\/s<\/b>/); + assert.match(telegram.posts[0], /Hashrate now: 3\.00 TH\/s<\/b> \(5m\)/); + assert.match(telegram.posts[0], /Hashrate: 2\.00 TH\/s \(1h\) · 1\.00 TH\/s \(24h\)/); + assert.match(telegram.posts[0], /Shares \(24h\): 1,000 accepted · 5 rejected \(0\.50%\)/); await tick(T0 + 5 * H); // 15:00, still 12-18 assert.equal(telegram.posts.length, 1); await tick(T0 + 8 * H); // 18:00 @@ -269,3 +272,24 @@ test('summary is off by default and leaves the live message alone', async () => assert.equal(telegram.posts.length, 1); // the live message, posted once assert.equal(telegram.edits.length, 1); }); + +test('BROADCASTER_LIVE_PIN=0 posts the live message without pinning it', async () => { + const { b, telegram } = setup({ live: true, livePin: false, digestHour: null }); + await b.tick(T0); + assert.equal(telegram.posts.length, 1); + assert.deepEqual(telegram.pins, []); +}); + +test('live message and summary show the last counted block', async () => { + const { b, dashboard, telegram } = setup({ live: true, summaryHours: 1, digestHour: null }); + dashboard.b = [ + block('old', 100, 'confirmed', T0 / 1000 - 7200), + block('lost', 101, 'orphaned', T0 / 1000 - 600), + ]; + await b.tick(T0); + assert.match(telegram.posts[0], /Last block: #100<\/b> · 2h 0m ago/); + dashboard.s = status({ lastShareTs: (T0 + 60 * MIN) / 1000 }); + await b.tick(T0 + 60 * MIN); + assert.match(telegram.posts[1], /— update/); + assert.match(telegram.posts[1], /Last block: #100<\/b>/); +}); diff --git a/broadcaster/test/commands.test.js b/broadcaster/test/commands.test.js new file mode 100644 index 0000000..dc4376b --- /dev/null +++ b/broadcaster/test/commands.test.js @@ -0,0 +1,117 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; + +import { Commands, COMMANDS } from '../lib/commands.js'; +import { emptyState } from '../lib/state.js'; + +const quietLog = { info() {}, warn() {}, error() {} }; +const T0 = Date.UTC(2026, 9, 2, 10, 0, 0); +const CHAT = -1001518607784; + +function status() { + return { + pool: { + db_ready: true, hashrate: 1e12, hashrate_1h: 2e12, hashrate_5m: 3e12, workers_active: 4, + accepted: 1000, rejected: 5, reject_rate_pct: 0.5, best_share_24h: 123456, blocks_lifetime: 7, last_share_ts: T0 / 1000 - 5, + }, + }; +} + +class FakeTelegram { + constructor(pending = []) { this.sent = []; this.menus = []; this.pending = pending; this.polls = []; } + async getMe() { return { username: 'avonpool_bot' }; } + async setMyCommands(c) { this.menus.push(c); } + async getUpdates(args) { this.polls.push(args); const p = this.pending; this.pending = []; return p; } + async send(html, opts) { this.sent.push({ html, opts }); return this.sent.length; } +} + +function setup({ pending, dashboardFails = false, state = emptyState() } = {}) { + const telegram = new FakeTelegram(pending); + const dashboard = { + async status() { if (dashboardFails) throw new Error('HTTP 502'); return status(); }, + async blocks() { return [{ hash: 'h', height: 970802, status: 'confirmed', ts: T0 / 1000 - 3600 }]; }, + }; + const cfg = { chatId: String(CHAT), poolName: 'avonpool_beta', publicUrl: null, + commandCooldownSec: 30, staleSharesSec: 900 }; + const c = new Commands({ dashboard, telegram, state, cfg, log: quietLog, persist() {} }); + return { c, telegram, state }; +} + +const cmd = (text, extra = {}) => ({ + message_id: 77, text, chat: { id: CHAT, type: 'supergroup' }, + is_topic_message: true, message_thread_id: 23563, ...extra, +}); + +test('first start registers the menu and skips the backlog', async () => { + const { c, telegram, state } = setup({ pending: [{ update_id: 40, message: cmd('/pool_status') }] }); + await c.init(); + assert.deepEqual(telegram.menus, [COMMANDS]); + assert.deepEqual(telegram.polls[0], { offset: -1, timeoutSec: 0 }); + assert.equal(state.updateOffset, 41); + assert.equal(telegram.sent.length, 0); +}); + +test('a restart keeps its offset instead of skipping again', async () => { + const { c, telegram } = setup({ state: { ...emptyState(), updateOffset: 12 } }); + await c.init(); + assert.equal(telegram.polls.length, 0); +}); + +test('/pool_status is answered in its topic, as a reply, with the last block', async () => { + const { c, telegram, state } = setup({ state: { ...emptyState(), updateOffset: 5 } }); + await c.init(); + telegram.pending = [{ update_id: 5, message: cmd('/pool_status@avonpool_bot') }]; + await c.pollOnce(() => T0); + assert.equal(state.updateOffset, 6); + assert.equal(telegram.sent.length, 1); + const { html, opts } = telegram.sent[0]; + assert.deepEqual(opts, { chatId: CHAT, threadId: 23563, replyTo: 77 }); + assert.match(html, /avonpool_beta — status/); + assert.match(html, /Hashrate now: 3\.00 TH\/s<\/b> \(5m\)/); + assert.match(html, /1,000 accepted · 5 rejected \(0\.50%\)/); + assert.match(html, /Last block: #970,802<\/b> · 1h 0m ago/); + assert.match(html, /^🟢/); +}); + +test('General topic: no thread id', async () => { + const { c, telegram } = setup(); + await c.init(); + await c.handle(cmd('/pool_status', { is_topic_message: undefined, message_thread_id: undefined }), T0); + assert.equal(telegram.sent[0].opts.threadId, null); +}); + +test('other chats, other bots and other text are ignored', async () => { + const { c, telegram } = setup(); + await c.init(); + await c.handle(cmd('/pool_status', { chat: { id: -100999 } }), T0); + await c.handle(cmd('/pool_status@some_other_bot'), T0); + await c.handle(cmd('/pool_statusx'), T0); + await c.handle(cmd('what is the pool_status'), T0); + await c.handle(cmd(undefined), T0); + assert.equal(telegram.sent.length, 0); +}); + +test('a public chat configured by @username matches', async () => { + const { c, telegram } = setup(); + c.cfg.chatId = '@AvonPool'; + await c.init(); + await c.handle(cmd('/pool_status', { chat: { id: -1002, username: 'avonpool' } }), T0); + assert.equal(telegram.sent.length, 1); +}); + +test('one reply per chat per cooldown', async () => { + const { c, telegram } = setup(); + await c.init(); + await c.handle(cmd('/pool_status'), T0); + await c.handle(cmd('/pool_status'), T0 + 10_000); + assert.equal(telegram.sent.length, 1); + await c.handle(cmd('/pool_status'), T0 + 31_000); + assert.equal(telegram.sent.length, 2); +}); + +test('a dashboard failure is answered, not swallowed', async () => { + const { c, telegram } = setup({ dashboardFails: true }); + await c.init(); + await c.handle(cmd('/pool_status'), T0); + assert.match(telegram.sent[0].html, /can't be read right now/); +}); diff --git a/broadcaster/test/telegram.test.js b/broadcaster/test/telegram.test.js index f78b15d..37e4555 100644 --- a/broadcaster/test/telegram.test.js +++ b/broadcaster/test/telegram.test.js @@ -86,3 +86,34 @@ test('calls are spaced minGapMs apart', async () => { assert.equal(slept.length, 1); assert.ok(slept[0] > 2900 && slept[0] <= 3000); }); + +test('a command reply overrides chat and topic and quotes the command', async () => { + const f = fakeFetch([[200, { ok: true, result: { message_id: 9 } }]]); + const t = new TelegramClient({ token: 'x', chatId: 'c', threadId: 1, fetchImpl: f, sleep: noSleep, minGapMs: 0 }); + await t.send('a', { chatId: -100, threadId: null, replyTo: 77 }); + const body = f.calls[0].body; + assert.equal(body.chat_id, -100); + assert.equal('message_thread_id' in body, false); + assert.deepEqual(body.reply_parameters, { message_id: 77, allow_sending_without_reply: true }); +}); + +test('getUpdates is not queued behind posts and asks only for messages', async () => { + let release; + const gate = new Promise((r) => { release = r; }); + const calls = []; + let updatesBody; + const fetchImpl = async (url, init) => { + calls.push(url.split('/').pop()); + if (url.endsWith('getUpdates')) updatesBody = JSON.parse(init.body); + if (url.endsWith('sendMessage')) await gate; + return { status: 200, json: async () => ({ ok: true, result: url.endsWith('getUpdates') ? [] : { message_id: 1 } }) }; + }; + const t = new TelegramClient({ token: 'x', chatId: 'c', fetchImpl, sleep: noSleep, minGapMs: 0 }); + const posting = t.send('a'); + await new Promise((r) => setImmediate(r)); + assert.deepEqual(await t.getUpdates({ offset: 3, timeoutSec: 0 }), []); + assert.deepEqual(calls, ['sendMessage', 'getUpdates']); + assert.deepEqual(updatesBody.allowed_updates, ['message']); + release(); + await posting; +}); diff --git a/deploy/docker/.env.example b/deploy/docker/.env.example index cec16e4..5f0733f 100644 --- a/deploy/docker/.env.example +++ b/deploy/docker/.env.example @@ -78,7 +78,11 @@ THUNDER_SIDECHAIN_ID=9 DIGEST_UTC_HOUR=12 # 1 = keep one pinned message in the channel edited with current stats. BROADCASTER_LIVE=0 +# 0 = post the live message without pinning it (the bot may not pin). +BROADCASTER_LIVE_PIN=1 # Also post the stats as a new message every N hours (1-168), or 'off'. BROADCASTER_SUMMARY_HOURS=off +# 1 = answer /pool_status in TELEGRAM_CHAT_ID (reads the chat via getUpdates). +BROADCASTER_COMMANDS=0 # 1 = print posts to the container log instead of sending them. BROADCASTER_DRY_RUN=0 diff --git a/deploy/docker/docker-compose.yml b/deploy/docker/docker-compose.yml index f21f678..28e5777 100644 --- a/deploy/docker/docker-compose.yml +++ b/deploy/docker/docker-compose.yml @@ -175,7 +175,9 @@ services: BROADCASTER_STATE_PATH: /data/broadcaster.json DIGEST_UTC_HOUR: ${DIGEST_UTC_HOUR:-12} BROADCASTER_LIVE: ${BROADCASTER_LIVE:-0} + BROADCASTER_LIVE_PIN: ${BROADCASTER_LIVE_PIN:-1} BROADCASTER_SUMMARY_HOURS: ${BROADCASTER_SUMMARY_HOURS:-off} + BROADCASTER_COMMANDS: ${BROADCASTER_COMMANDS:-0} BROADCASTER_DRY_RUN: ${BROADCASTER_DRY_RUN:-0} volumes: # Writes only broadcaster.json. diff --git a/deploy/systemd/simplepool-broadcaster.service b/deploy/systemd/simplepool-broadcaster.service index dd95329..3ccbc29 100644 --- a/deploy/systemd/simplepool-broadcaster.service +++ b/deploy/systemd/simplepool-broadcaster.service @@ -14,7 +14,9 @@ Environment=DASHBOARD_URL=http://127.0.0.1:8081 Environment=BROADCASTER_STATE_PATH=@ROOT@/data/broadcaster.json Environment=DIGEST_UTC_HOUR=12 # Environment=BROADCASTER_LIVE=1 +# Environment=BROADCASTER_LIVE_PIN=0 # Environment=BROADCASTER_SUMMARY_HOURS=6 +# Environment=BROADCASTER_COMMANDS=1 # Environment=PUBLIC_DASHBOARD_URL=https://pool.example # Environment=POOL_NAME= # The bot token and channel go in a 0600 drop-in, not here: this file is