226 lines
14 KiB
TypeScript
226 lines
14 KiB
TypeScript
// Where `storebound watch` sends events: OS notification, ntfy, Telegram, a webhook, a push ingest endpoint. All optional, configured from the
|
|
// environment or ~/.storebound/watch.json (never printed; logs name the sink, not its URL, topic or token):
|
|
// STOREBOUND_NTFY_URL (default https://ntfy.sh) + STOREBOUND_NTFY_TOPIC [+ STOREBOUND_NTFY_TOKEN]
|
|
// STOREBOUND_TELEGRAM_TOKEN + STOREBOUND_TELEGRAM_CHAT
|
|
// STOREBOUND_WEBHOOK_URL
|
|
// STOREBOUND_INGEST_URL + STOREBOUND_INGEST_TOKEN_FILE [+ STOREBOUND_INGEST_CATEGORY]
|
|
// watch.json: { "notify": ["os"], "ntfy": { "url", "topic", "token" }, "telegram": { "token", "chat" }, "webhook": { "url" },
|
|
// "ingest": { "url", "tokenFile", "category" } }
|
|
// Builders are pure (tests check them); `send` takes the fetch and the OS runner as parameters.
|
|
import { spawnSync } from 'node:child_process';
|
|
import { readFileSync, statSync } from 'node:fs';
|
|
import { homedir } from 'node:os';
|
|
import { type EventKind, watchConfigFile, type WatchEvent } from './watch';
|
|
|
|
export type SinkName = 'os' | 'ntfy' | 'telegram' | 'webhook' | 'ingest';
|
|
export const SINK_NAMES: readonly SinkName[] = ['os', 'ntfy', 'telegram', 'webhook', 'ingest'];
|
|
|
|
export interface SinkConfig {
|
|
os?: true;
|
|
ntfy?: { url: string; topic: string; token?: string };
|
|
telegram?: { token: string; chat: string };
|
|
webhook?: { url: string };
|
|
ingest?: { url: string; tokenFile: string; category: string };
|
|
}
|
|
|
|
type Env = Record<string, string | undefined>;
|
|
|
|
/** The sinks that are configured (os only when the file lists it). `only` limits and, for os, enables. */
|
|
export function sinkConfig(env: Env = process.env, file: string | null = watchConfigFile(env), only?: SinkName[]): SinkConfig {
|
|
let f: { notify?: string[]; ntfy?: Record<string, string>; telegram?: Record<string, string>; webhook?: Record<string, string>; ingest?: Record<string, string> } = {};
|
|
if (file) { try { f = JSON.parse(readFileSync(file, 'utf8')); } catch { /* no file */ } }
|
|
const out: SinkConfig = {};
|
|
const topic = env.STOREBOUND_NTFY_TOPIC || f.ntfy?.topic;
|
|
if (topic) out.ntfy = { url: (env.STOREBOUND_NTFY_URL || f.ntfy?.url || 'https://ntfy.sh').replace(/\/+$/, ''), topic, ...(env.STOREBOUND_NTFY_TOKEN || f.ntfy?.token ? { token: (env.STOREBOUND_NTFY_TOKEN || f.ntfy?.token)! } : {}) };
|
|
const tg = { token: env.STOREBOUND_TELEGRAM_TOKEN || f.telegram?.token, chat: env.STOREBOUND_TELEGRAM_CHAT || f.telegram?.chat };
|
|
if (tg.token && tg.chat) out.telegram = { token: tg.token, chat: tg.chat };
|
|
const hook = env.STOREBOUND_WEBHOOK_URL || f.webhook?.url;
|
|
if (hook) out.webhook = { url: hook };
|
|
const iUrl = env.STOREBOUND_INGEST_URL || f.ingest?.url, iFile = env.STOREBOUND_INGEST_TOKEN_FILE || f.ingest?.tokenFile;
|
|
if (iUrl && iFile) out.ingest = { url: iUrl, tokenFile: iFile, category: env.STOREBOUND_INGEST_CATEGORY || f.ingest?.category || 'storebound' };
|
|
if (Array.isArray(f.notify) && f.notify.includes('os')) out.os = true;
|
|
if (only) {
|
|
const picked: SinkConfig = {};
|
|
for (const n of only) if (n === 'os') picked.os = true; else if (out[n]) (picked as Record<string, unknown>)[n] = out[n];
|
|
return picked;
|
|
}
|
|
return out;
|
|
}
|
|
|
|
export const configuredSinks = (c: SinkConfig): SinkName[] => SINK_NAMES.filter(n => !!c[n]);
|
|
|
|
/** `--notify os,ntfy` -> names; throws on an unknown one. */
|
|
export function parseNotify(list: string[]): SinkName[] {
|
|
for (const n of list) if (!(SINK_NAMES as readonly string[]).includes(n)) throw new Error(`--notify: ${n} is not a sink (${SINK_NAMES.join(', ')})`);
|
|
return list as SinkName[];
|
|
}
|
|
|
|
const LABEL: Record<EventKind, string> = {
|
|
rejected: 'Rejected', approved: 'Approved', live: 'Live', rollout_halted: 'Rollout halted', crash_threshold: 'Crash threshold',
|
|
review_low: 'Low review', review_needs_reply: 'Review needs a reply',
|
|
};
|
|
const TAGS: Record<EventKind, string> = {
|
|
rejected: 'x', approved: 'white_check_mark', live: 'rocket', rollout_halted: 'octagonal_sign', crash_threshold: 'warning', review_low: 'star', review_needs_reply: 'speech_balloon',
|
|
};
|
|
const URGENT: ReadonlySet<EventKind> = new Set<EventKind>(['rejected', 'rollout_halted']);
|
|
|
|
/** HTTP header values are Latin-1 text on one line. */
|
|
const header = (s: string): string => s.replace(/[\r\n]+/g, ' ').replace(/[^\x20-\x7e]/g, '?').slice(0, 250);
|
|
export const titleOf = (e: WatchEvent): string => `${e.app}${e.version ? ` ${e.version}` : ''}: ${LABEL[e.kind]}`;
|
|
export const textOf = (e: WatchEvent): string => `${e.detail}\n${e.url}`;
|
|
|
|
export interface HttpRequest { url: string; headers: Record<string, string>; body: string }
|
|
|
|
export function ntfyRequest(e: WatchEvent, c: NonNullable<SinkConfig['ntfy']>): HttpRequest {
|
|
return { url: `${c.url}/${encodeURIComponent(c.topic)}`, body: e.detail,
|
|
headers: { Title: header(titleOf(e)), Tags: `${TAGS[e.kind]},${e.store}`, Click: e.url, Priority: URGENT.has(e.kind) ? 'high' : 'default', ...(c.token ? { Authorization: `Bearer ${c.token}` } : {}) } };
|
|
}
|
|
export function telegramRequest(e: WatchEvent, c: NonNullable<SinkConfig['telegram']>): HttpRequest {
|
|
return { url: `https://api.telegram.org/bot${c.token}/sendMessage`, headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify({ chat_id: c.chat, text: `${titleOf(e)}\n${textOf(e)}`.slice(0, 4000), disable_web_page_preview: true }) };
|
|
}
|
|
export function webhookRequest(e: WatchEvent, c: NonNullable<SinkConfig['webhook']>): HttpRequest {
|
|
return { url: c.url, headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(e) };
|
|
}
|
|
|
|
/** The command of an OS notification (title and message travel in the environment, never inside a script string). */
|
|
export function osCommand(platform: string, title: string, message: string): { cmd: string; args: string[]; env: Record<string, string> } | null {
|
|
const env = { SB_TITLE: title.slice(0, 120), SB_MSG: message.slice(0, 400) };
|
|
if (platform === 'win32') {
|
|
const script = [
|
|
'[Windows.UI.Notifications.ToastNotificationManager, Windows.UI.Notifications, ContentType = WindowsRuntime] > $null',
|
|
'$x = [Windows.UI.Notifications.ToastNotificationManager]::GetTemplateContent([Windows.UI.Notifications.ToastTemplateType]::ToastText02)',
|
|
'$t = $x.GetElementsByTagName(\'text\')',
|
|
'$t.Item(0).AppendChild($x.CreateTextNode($env:SB_TITLE)) > $null',
|
|
'$t.Item(1).AppendChild($x.CreateTextNode($env:SB_MSG)) > $null',
|
|
'[Windows.UI.Notifications.ToastNotificationManager]::CreateToastNotifier(\'storebound\').Show([Windows.UI.Notifications.ToastNotification]::new($x))',
|
|
].join('; ');
|
|
return { cmd: 'powershell', args: ['-NoProfile', '-NonInteractive', '-Command', script], env };
|
|
}
|
|
if (platform === 'darwin') return { cmd: 'osascript', args: ['-e', 'display notification (system attribute "SB_MSG") with title (system attribute "SB_TITLE")'], env };
|
|
if (platform === 'linux') return { cmd: 'notify-send', args: [env.SB_TITLE, env.SB_MSG], env };
|
|
return null;
|
|
}
|
|
|
|
export type FetchLike = (url: string, init: { method: string; headers: Record<string, string>; body: string; signal?: AbortSignal }) => Promise<{ ok: boolean; status: number; text?: () => Promise<string> }>;
|
|
export type OsRunner = (c: { cmd: string; args: string[]; env: Record<string, string> }) => boolean;
|
|
|
|
export const runOs: OsRunner = c => {
|
|
try { return spawnSync(c.cmd, c.args, { env: { ...process.env, ...c.env }, timeout: 10_000, stdio: 'ignore', windowsHide: true }).status === 0; } catch { return false; }
|
|
};
|
|
|
|
/** POST with a timeout and one retry (network error, 429 or 5xx). Resolves to an error text, or null when sent. */
|
|
export async function post(fetchFn: FetchLike, r: HttpRequest, o: { timeoutMs?: number; retryMs?: number } = {}): Promise<string | null> {
|
|
let last = '';
|
|
for (let attempt = 0; attempt < 2; attempt++) {
|
|
if (attempt) await new Promise(res => setTimeout(res, o.retryMs ?? 1500));
|
|
try {
|
|
const res = await fetchFn(r.url, { method: 'POST', headers: r.headers, body: r.body, signal: AbortSignal.timeout(o.timeoutMs ?? 10_000) });
|
|
if (res.ok) return null;
|
|
last = `HTTP ${res.status}`;
|
|
if (res.status < 500 && res.status !== 429) return last;
|
|
} catch (e) { last = (e as Error).name === 'TimeoutError' ? 'timed out' : 'network error'; }
|
|
}
|
|
return last;
|
|
}
|
|
|
|
// ---- ingest: one JSON POST per event to a push ingest endpoint (Bearer token read from a file at send time) ----
|
|
export const INGEST_EVENTS: Partial<Record<EventKind, string>> = {
|
|
rejected: 'review-rejected', approved: 'review-approved', live: 'live', rollout_halted: 'rollout-halted',
|
|
crash_threshold: 'crash-threshold', review_low: 'low-reviews', review_needs_reply: 'review-reply',
|
|
};
|
|
export const INGEST_MAX_BODY = 8 * 1024;
|
|
export const INGEST_DETAIL_MAX = 300;
|
|
export const INGEST_PER_MINUTE = 60;
|
|
export const slugOf = (s: string): string => s.toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/^-+|-+$/g, '') || 'app';
|
|
|
|
function ingestDetail(e: WatchEvent): string {
|
|
if (e.kind === 'review_low' || e.kind === 'review_needs_reply') return e.to ?? e.detail;
|
|
if (e.kind === 'crash_threshold' && e.store === 'play') {
|
|
const m = /([\d.]+)\s*%/.exec(e.to ?? '');
|
|
if (m) return `crash-free ${(100 - Number(m[1])).toFixed(1)}%`;
|
|
}
|
|
return e.detail;
|
|
}
|
|
|
|
/** The exact JSON body the endpoint accepts (no other fields: unknown ones are a 400); null for an event with no mapping. */
|
|
export function ingestBody(e: WatchEvent, c: { category: string }): Record<string, string> | null {
|
|
const event = INGEST_EVENTS[e.kind];
|
|
if (!event) return null;
|
|
const app = slugOf(e.app);
|
|
const body: Record<string, string> = { category: c.category, event, app, platform: e.store === 'apple' ? 'ios' : 'android' };
|
|
if (e.version) body.version = e.version;
|
|
body.detail = ingestDetail(e).replace(/\s+/g, ' ').trim().slice(0, INGEST_DETAIL_MAX);
|
|
body.url = e.url;
|
|
body.dedupeKey = `${app}:${event}:${e.version ?? ''}${e.ref ? `:${e.ref}` : ''}`;
|
|
// keep the whole body within 8 KiB (only detail is free text)
|
|
while (Buffer.byteLength(JSON.stringify(body)) > INGEST_MAX_BODY && body.detail!.length > 0) body.detail = body.detail!.slice(0, Math.max(0, body.detail!.length - 100));
|
|
return body;
|
|
}
|
|
|
|
const expandHome = (p: string): string => (p === '~' || /^~[\\/]/.test(p) ? homedir() + p.slice(1) : p);
|
|
|
|
/** The token from its file (trimmed), or an error text that names the file, never the token. */
|
|
export function readIngestToken(file: string): { token: string } | { error: string } {
|
|
const path = expandHome(file);
|
|
try {
|
|
const st = statSync(path);
|
|
if (process.platform !== 'win32' && (st.mode & 0o077)) return { error: `token file ${file} is readable by group or others, refusing to use it (chmod 600 ${file})` };
|
|
const token = readFileSync(path, 'utf8').trim();
|
|
if (!token) return { error: `token file ${file} is empty` };
|
|
return { token };
|
|
} catch { return { error: `token file ${file} cannot be read` }; }
|
|
}
|
|
|
|
let sentAt: number[] = [];
|
|
/** Client-side limit: INGEST_PER_MINUTE sends in any 60 s window; true = allowed (and counted). */
|
|
export function ingestAllow(now = Date.now()): boolean {
|
|
sentAt = sentAt.filter(t => now - t < 60_000);
|
|
if (sentAt.length >= INGEST_PER_MINUTE) return false;
|
|
sentAt.push(now);
|
|
return true;
|
|
}
|
|
export const resetIngestLimit = (): void => { sentAt = []; };
|
|
|
|
/** 202 = sent. 5xx and network errors are retried once; 4xx never (status and the server's message are the error). Resolves to an error text or null. */
|
|
export async function sendIngest(e: WatchEvent, c: NonNullable<SinkConfig['ingest']>, d: { fetch: FetchLike; timeoutMs?: number; retryMs?: number }): Promise<string | null> {
|
|
const body = ingestBody(e, c);
|
|
if (!body) return null;
|
|
const t = readIngestToken(c.tokenFile);
|
|
if ('error' in t) return `not sent, ${t.error}`;
|
|
if (!ingestAllow()) return `not sent, rate limit of ${INGEST_PER_MINUTE}/min reached (event dropped)`;
|
|
const scrub = (s: string): string => s.split(t.token).join('***').replace(/\s+/g, ' ').slice(0, 200);
|
|
const init = { method: 'POST', headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${t.token}` }, body: JSON.stringify(body) };
|
|
let last = '';
|
|
for (let attempt = 0; attempt < 2; attempt++) {
|
|
if (attempt) await new Promise(res => setTimeout(res, d.retryMs ?? 1500));
|
|
try {
|
|
const res = await d.fetch(c.url, { ...init, signal: AbortSignal.timeout(d.timeoutMs ?? 10_000) });
|
|
if (res.status === 202) return null;
|
|
let msg = '';
|
|
try { msg = scrub(await res.text?.() ?? ''); } catch { /* no body */ }
|
|
last = `HTTP ${res.status}${msg ? `: ${msg}` : ''}`;
|
|
if (res.status < 500) return last;
|
|
} catch (err) { last = (err as Error).name === 'TimeoutError' ? 'timed out' : 'network error'; }
|
|
}
|
|
return last;
|
|
}
|
|
|
|
export interface SendReport { sink: SinkName; ok: boolean; error?: string }
|
|
|
|
/** Sends one event to every sink in the config. A failing sink is a report row, never an exception. */
|
|
export async function sendEvent(e: WatchEvent, c: SinkConfig, d: { fetch?: FetchLike; os?: OsRunner; platform?: string; timeoutMs?: number; retryMs?: number } = {}): Promise<SendReport[]> {
|
|
const fetchFn = d.fetch ?? (fetch as unknown as FetchLike);
|
|
const out: SendReport[] = [];
|
|
const http = async (sink: SinkName, r: HttpRequest): Promise<void> => { const err = await post(fetchFn, r, d); out.push(err ? { sink, ok: false, error: err } : { sink, ok: true }); };
|
|
if (c.os) {
|
|
const cmd = osCommand(d.platform ?? process.platform, titleOf(e), e.detail);
|
|
// unavailable (no notifier on this machine): skipped quietly, not a failure
|
|
if (cmd) { const ok = (d.os ?? runOs)(cmd); if (ok) out.push({ sink: 'os', ok: true }); }
|
|
}
|
|
if (c.ntfy) await http('ntfy', ntfyRequest(e, c.ntfy));
|
|
if (c.telegram) await http('telegram', telegramRequest(e, c.telegram));
|
|
if (c.webhook) await http('webhook', webhookRequest(e, c.webhook));
|
|
if (c.ingest && INGEST_EVENTS[e.kind]) { const err = await sendIngest(e, c.ingest, { fetch: fetchFn, timeoutMs: d.timeoutMs, retryMs: d.retryMs }); out.push(err ? { sink: 'ingest', ok: false, error: err } : { sink: 'ingest', ok: true }); }
|
|
return out;
|
|
}
|