From 3a5a9941793f6e931177ac7421ce4316558a42ec Mon Sep 17 00:00:00 2001 From: rob Date: Fri, 2 Oct 2026 09:39:41 +0200 Subject: [PATCH 1/2] broadcaster: /pool_status, a pin switch, and the last block in snapshots MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three additions to the broadcaster. BROADCASTER_LIVE_PIN=0 posts the live message without pinning it. A bot that may post in a group but not pin (a plain member of a forum group, for example) otherwise logs "not enough rights to manage pinned messages" on every new live message. Default 1, so nothing changes unless it is set. BROADCASTER_COMMANDS=1 answers /pool_status in TELEGRAM_CHAT_ID with the current stats, as a reply in the topic it was typed in. lib/commands.js runs its own getUpdates long poll beside the poster, so a held poll never delays a post and a failure there never stops one. getUpdates bypasses the send queue for the same reason. Details: - Only the configured chat is answered (numeric id or @username), so adding the bot elsewhere does not make it reply there. - "/pool_status@other_bot" is ignored. - At most one reply per chat per BROADCASTER_COMMAND_COOLDOWN_SEC (default 30). - The command is registered in the bot's "/" menu, because a bot in privacy mode only reliably sees the addressed "/pool_status@bot" form a menu tap sends. - The first start skips the backlog. The offset is kept in broadcaster.json, so a restart answers nothing twice. - A dashboard failure gets a short "can't be read right now" reply. - Startup failures are retried rather than switching the command off. - Off under BROADCASTER_DRY_RUN, which has no bot to read with. The live message, the periodic update and the /pool_status reply now show "Last block: # · ago": the newest pending or confirmed block, never an orphaned or rejected one. index.js now builds one state object and one persist() for both loops, so neither can save over the other with a stale copy. Co-Authored-By: Claude Opus 5.5 (1M context) --- broadcaster/README.md | 22 ++- broadcaster/index.js | 32 ++-- broadcaster/lib/broadcaster.js | 26 ++-- broadcaster/lib/commands.js | 143 ++++++++++++++++++ broadcaster/lib/config.js | 11 ++ broadcaster/lib/messages.js | 25 ++- broadcaster/lib/state.js | 5 +- broadcaster/lib/telegram.js | 71 ++++++--- broadcaster/test/broadcaster.test.js | 21 +++ broadcaster/test/commands.test.js | 115 ++++++++++++++ broadcaster/test/telegram.test.js | 31 ++++ deploy/docker/.env.example | 4 + deploy/docker/docker-compose.yml | 2 + deploy/systemd/simplepool-broadcaster.service | 2 + 14 files changed, 468 insertions(+), 42 deletions(-) create mode 100644 broadcaster/lib/commands.js create mode 100644 broadcaster/test/commands.test.js diff --git a/broadcaster/README.md b/broadcaster/README.md index df84da1..c79158f 100644 --- a/broadcaster/README.md +++ b/broadcaster/README.md @@ -22,7 +22,8 @@ if Telegram or the service goes down, the pool is unaffected. | **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..0e9fa7e 100644 --- a/broadcaster/lib/messages.js +++ b/broadcaster/lib/messages.js @@ -98,9 +98,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 +118,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 +134,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..14c1597 100644 --- a/broadcaster/test/broadcaster.test.js +++ b/broadcaster/test/broadcaster.test.js @@ -269,3 +269,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..0900e50 --- /dev/null +++ b/broadcaster/test/commands.test.js @@ -0,0 +1,115 @@ +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, workers_active: 4, accepted: 1000, + 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, /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 From af1254c085381effc356eaebcff71242398a9415 Mon Sep 17 00:00:00 2001 From: rob Date: Fri, 2 Oct 2026 09:42:04 +0200 Subject: [PATCH 2/2] broadcaster: current hashrate and rejected share count in every snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The stat block shared by the digest, the live message, the periodic update and /pool_status now leads with the current rate, the dashboard's 5-minute hashrate_5m: Hashrate now: 3.33 PH/s (5m) Hashrate: 2.33 PH/s (1h) · 2.00 PH/s (24h) The shares line gives the rejected count beside the accepted one, keeping the rate: Shares (24h): 34,763 accepted · 76 rejected (0.22%) Both fields were already in /api/status. Co-Authored-By: Claude Opus 5.5 (1M context) --- broadcaster/README.md | 2 +- broadcaster/lib/messages.js | 9 ++++++--- broadcaster/test/broadcaster.test.js | 9 ++++++--- broadcaster/test/commands.test.js | 6 ++++-- 4 files changed, 17 insertions(+), 9 deletions(-) diff --git a/broadcaster/README.md b/broadcaster/README.md index c79158f..cc5cd16 100644 --- a/broadcaster/README.md +++ b/broadcaster/README.md @@ -17,7 +17,7 @@ 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 | diff --git a/broadcaster/lib/messages.js b/broadcaster/lib/messages.js index 0e9fa7e..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)}`, ]; } diff --git a/broadcaster/test/broadcaster.test.js b/broadcaster/test/broadcaster.test.js index 14c1597..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 diff --git a/broadcaster/test/commands.test.js b/broadcaster/test/commands.test.js index 0900e50..dc4376b 100644 --- a/broadcaster/test/commands.test.js +++ b/broadcaster/test/commands.test.js @@ -11,8 +11,8 @@ const CHAT = -1001518607784; function status() { return { pool: { - db_ready: true, hashrate: 1e12, hashrate_1h: 2e12, workers_active: 4, accepted: 1000, - reject_rate_pct: 0.5, best_share_24h: 123456, blocks_lifetime: 7, last_share_ts: T0 / 1000 - 5, + 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, }, }; } @@ -67,6 +67,8 @@ test('/pool_status is answered in its topic, as a reply, with the last block', a 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, /^🟢/); });