214 lines
13 KiB
TypeScript
214 lines
13 KiB
TypeScript
// `storebound watch`: store events for every app on this machine, for people and agents.
|
|
// storebound watch [--dir <folder>] [--every 15m] [--notify os,ntfy,telegram,webhook] [--once] [--dry] [--json]
|
|
// storebound watch test [--notify ...] [--dry] sends one test event to every configured sink
|
|
// Apps: the machine's app folder list (~/.storebound/projects.json: folders and store-only apps, the same list as the
|
|
// desktop app) or --dir for one folder. Each pass reads release state (like `release status`), crash / ANR health
|
|
// (like `health`, at most hourly) and new reviews (like `reviews list`) per app with the credentials those commands
|
|
// use, compares with ~/.storebound/watch-state.json and emits events: rejected, approved, live, rollout_halted,
|
|
// crash_threshold, review_low (<= 2 stars), review_needs_reply (3-5 stars, no reply). The first pass of an app only
|
|
// records the baseline. Every event is appended to ~/.storebound/events.jsonl (`storebound events` reads it) and sent
|
|
// to the sinks (docs/watch.md). Read-only on the stores. STOREBOUND_HOME moves ~/.storebound.
|
|
// --every loop period, 5m or more (a bare number is minutes), with +-10 % jitter; errors back off up to 4 h
|
|
// --once one pass and exit (cron, PM2 restarts) --dry print what would be sent; nothing is sent or saved
|
|
// --json events as JSON lines on stdout
|
|
import { existsSync } from 'node:fs';
|
|
import { join, resolve } from 'node:path';
|
|
import { emptyStatus, fillAppStatus, lastLine, type Exec } from './lib/app-status';
|
|
import type { Args } from './lib/args';
|
|
import { CONFIG_NAMES } from './lib/config';
|
|
import { consoleIdsFromConfig } from './lib/console-links';
|
|
import { profileEnv } from './lib/credentials';
|
|
import { done, log, warn } from './lib/log';
|
|
import { type IgnoreStore, ignoredStores, isIgnored, readRegistry } from './lib/registry';
|
|
import { runCommand } from './lib/runner';
|
|
import type { HealthSnapshot } from './lib/health';
|
|
import {
|
|
type AppRead, type AppSnap, appendEvents, durationMs, eventsFile, healthSnap, type PassTarget, readState, releaseMap, type ReviewSnap, runPass, stateFile,
|
|
testEvent, type WatchEvent, writeState,
|
|
} from './lib/watch';
|
|
import { configuredSinks, parseNotify, sendEvent, sinkConfig, type SinkConfig } from './lib/watch-sinks';
|
|
|
|
const TAG = 'watch';
|
|
const USAGE = 'usage: storebound watch [--dir <folder>] [--every 15m] [--notify os,ntfy,telegram,webhook] [--once] [--dry] [--json] | watch test [--notify ...] [--dry]';
|
|
export const HEALTH_EVERY_MS = 60 * 60_000;
|
|
const MIN_EVERY_MS = 5 * 60_000;
|
|
const MAX_BACKOFF_MS = 4 * 60 * 60_000;
|
|
const CHILD_TIMEOUT_MS = 5 * 60_000;
|
|
const REVIEW_DAYS = 3;
|
|
|
|
interface Target extends PassTarget { dir: string; virtual: boolean; env: Record<string, string>; /** store sides the registry ignores: not read, no events */ skip: IgnoreStore[] }
|
|
|
|
/** Runs a storebound child command; a hung one is cancelled after the timeout. */
|
|
const exec = (env: Record<string, string>): Exec => async (argv, cwd) => {
|
|
const run = runCommand(argv, { cwd, env, machine: true });
|
|
const timer = setTimeout(() => run.cancel(), CHILD_TIMEOUT_MS);
|
|
try { return await run.result; } finally { clearTimeout(timer); }
|
|
};
|
|
|
|
/** The apps to watch: --dir, or the registry's folders and store-only apps (a store-only app that has a folder here is read once). */
|
|
export async function listTargets(args: Args): Promise<Target[]> {
|
|
const dir = args.get('dir');
|
|
if (dir) {
|
|
const d = resolve(dir);
|
|
if (!CONFIG_NAMES.some(n => existsSync(join(d, n)))) throw new Error(`--dir: no storebound.config.ts in ${d}`);
|
|
return [{ key: `dir:${d}`, name: d.split(/[\\/]/).pop() ?? d, dir: d, virtual: false, env: {}, skip: [] }];
|
|
}
|
|
const reg = readRegistry();
|
|
const out: Target[] = [];
|
|
for (const p of reg.projects) {
|
|
if (isIgnored(p)) continue; // ignored apps are not watched at all (storebound apps ignore / unignore)
|
|
if (!CONFIG_NAMES.some(n => existsSync(join(p.path, n)))) { warn(TAG, `${p.name}: no storebound.config.ts in ${p.path}, skipped (storebound apps remove ${p.path})`); continue; }
|
|
out.push({ key: `dir:${p.path}`, name: p.name, dir: p.path, virtual: false, env: {}, skip: ignoredStores(p) });
|
|
}
|
|
if (reg.storeApps?.length) {
|
|
const { checkVirtual, virtualDir } = await import('./serve');
|
|
for (const s of reg.storeApps) {
|
|
if (isIgnored(s)) continue;
|
|
try {
|
|
// a store-only App Store app added without --bundle-id: read it from the API with the app's own key
|
|
const bundleId = s.store === 'apple' ? (s.bundleId || await appleBundleId(s.appId, s.account ? profileEnv(s.account) : {})) : null;
|
|
if (s.store === 'apple' && !bundleId) throw new Error('no bundle id (storebound apps add --store apple --app-id <id> --bundle-id <id>)');
|
|
const v = checkVirtual({ name: s.name, ...(s.store === 'apple' ? { apple: { bundleId: bundleId!, appId: s.appId } } : { play: { packageName: s.appId } }) });
|
|
out.push({ key: `store:${s.id}`, name: s.name, dir: virtualDir(v), virtual: true, env: s.account ? profileEnv(s.account) : {}, skip: [] });
|
|
} catch (e) { warn(TAG, `${s.name}: ${(e as Error).message}, skipped`); }
|
|
}
|
|
}
|
|
return out;
|
|
}
|
|
|
|
/** The bundle id of an App Store app by its numeric id, read with the given credential env; null when it cannot be read. */
|
|
async function appleBundleId(appId: string, env: Record<string, string>): Promise<string | null> {
|
|
const saved = Object.fromEntries(Object.keys(env).map(k => [k, process.env[k]]));
|
|
Object.assign(process.env, env);
|
|
try {
|
|
const { ascClient } = await import('./apple/asc');
|
|
const r = await (await ascClient(process.cwd(), '')).asc('GET', `/v1/apps/${appId}?fields[apps]=bundleId`) as { data?: { attributes?: { bundleId?: string } } };
|
|
return r.data?.attributes?.bundleId ?? null;
|
|
} catch { return null; } finally {
|
|
for (const [k, v] of Object.entries(saved)) { if (v === undefined) delete process.env[k]; else process.env[k] = v; }
|
|
}
|
|
}
|
|
|
|
/** Reads one app: releases of both stores, health when due, new reviews. A part that cannot be read is left out. */
|
|
async function readApp(t: Target, prev: AppSnap | undefined, now: number): Promise<AppRead> {
|
|
const run = exec(t.env);
|
|
const errors: string[] = [];
|
|
// virtual: true skips fillAppStatus's offline `check` (only the store reads are needed here)
|
|
const e = emptyStatus(t.dir);
|
|
try { await fillAppStatus(e, t.dir, { live: true, virtual: true, exec: run }); } catch (err) { return { snap: null, errors: [(err as Error).message] }; }
|
|
const s = e.summary!;
|
|
const useApple = !!s.apple && !t.skip.includes('apple'), usePlay = !!s.play && !t.skip.includes('play');
|
|
const snap: AppSnap = { name: s.name || t.name, ids: { appleAppId: s.apple?.appId ?? null } };
|
|
if (usePlay && !t.virtual && !prev?.ids.playDeveloperId) {
|
|
try { const { loadConfig } = await import('./lib/config'); const { playIdsOf } = await import('./lib/play-ids'); const l = await loadConfig(undefined, t.dir); const play = playIdsOf(l.config, l.root); Object.assign(snap.ids, consoleIdsFromConfig(l.config), { playDeveloperId: play.developerId, playAppId: play.appId }); } catch { /* links fall back to the app list / developer chooser */ }
|
|
}
|
|
const all = e.release.all ?? [];
|
|
snap.releases = {};
|
|
if (useApple) { if (e.storeErrors.ios) errors.push(`App Store: ${e.storeErrors.ios}`); else snap.releases.apple = releaseMap(all, 'ios'); }
|
|
if (usePlay) { if (e.storeErrors.android) errors.push(`Google Play: ${e.storeErrors.android}`); else snap.releases.play = releaseMap(all, 'android'); }
|
|
|
|
const due = !prev?.healthAt || now - Date.parse(prev.healthAt) >= HEALTH_EVERY_MS;
|
|
if (due && (useApple || usePlay)) {
|
|
const h = await run(['health', '--json'], t.dir);
|
|
if (h.data && typeof h.data === 'object') { snap.health = healthSnap({ ...(h.data as HealthSnapshot), ...(useApple ? {} : { apple: undefined }), ...(usePlay ? {} : { play: undefined }) } as HealthSnapshot); snap.healthAt = new Date(now).toISOString(); }
|
|
else errors.push(`health: ${lastLine(h.out) || `exit ${h.code}`}`);
|
|
}
|
|
|
|
snap.reviews = {};
|
|
for (const [store, on] of [['apple', useApple], ['play', usePlay]] as const) {
|
|
if (!on) continue;
|
|
const r = await run(['reviews', 'list', '--store', store, '--days', String(REVIEW_DAYS), '--json'], t.dir);
|
|
if (r.code !== 0 || !Array.isArray(r.data)) { errors.push(`reviews ${store}: ${lastLine(r.out) || `exit ${r.code}`}`); continue; }
|
|
snap.reviews[store] = (r.data as { id: string; rating: number; title?: string; body?: string; date: string; reply?: unknown }[]).map((x): ReviewSnap =>
|
|
({ id: String(x.id), rating: x.rating, replied: !!x.reply, date: x.date, text: `${x.title ? x.title + ': ' : ''}${x.body ?? ''}`.slice(0, 300) }));
|
|
}
|
|
return { snap, errors };
|
|
}
|
|
|
|
const line = (e: WatchEvent): string => `${e.at.slice(0, 19)}Z ${e.app} [${e.store}] ${e.kind}${e.version ? ` ${e.version}` : ''}: ${e.detail} ${e.url}`;
|
|
|
|
function chosenSinks(args: Args): SinkConfig {
|
|
const only = args.has('notify') ? parseNotify(args.list('notify')) : undefined;
|
|
if (only) {
|
|
const have = sinkConfig();
|
|
for (const n of only) if (n !== 'os' && !have[n]) warn(TAG, `--notify ${n}: not configured (docs/watch.md), skipped`);
|
|
}
|
|
return sinkConfig(process.env, undefined, only);
|
|
}
|
|
|
|
/** Sends to the sinks; a failing sink is logged (the sink name only) and never stops anything. */
|
|
async function deliver(e: WatchEvent, c: SinkConfig, args: Args): Promise<void> {
|
|
if (args.has('dry')) { log(TAG, `dry run: would send ${e.kind} of ${e.app} to ${configuredSinks(c).join(', ') || 'no sink'}`); return; }
|
|
for (const r of await sendEvent(e, c)) if (!r.ok) warn(TAG, `${r.sink}: not sent (${r.error})`);
|
|
}
|
|
|
|
function emit(e: WatchEvent, args: Args): void {
|
|
if (args.has('json')) process.stdout.write(JSON.stringify(e) + '\n');
|
|
else console.log(line(e));
|
|
}
|
|
|
|
async function testCmd(args: Args): Promise<void> {
|
|
const c = chosenSinks(args);
|
|
const names = configuredSinks(c);
|
|
if (!names.length) { warn(TAG, 'no sink configured: set STOREBOUND_NTFY_TOPIC, STOREBOUND_TELEGRAM_TOKEN + STOREBOUND_TELEGRAM_CHAT, STOREBOUND_WEBHOOK_URL, or pass --notify os (docs/watch.md)'); process.exitCode = 1; return; }
|
|
const e = testEvent();
|
|
emit(e, args);
|
|
if (args.has('dry')) { done(TAG, `dry run: would send the test event to ${names.join(', ')}`); return; }
|
|
const rep = await sendEvent(e, c);
|
|
for (const r of rep) (r.ok ? log : warn)(TAG, `${r.sink}: ${r.ok ? 'sent' : `not sent (${r.error})`}`);
|
|
const missing = names.filter(n => !rep.some(r => r.sink === n));
|
|
if (missing.length) log(TAG, `${missing.join(', ')}: unavailable on this machine, skipped`);
|
|
if (rep.some(r => !r.ok)) process.exitCode = 1;
|
|
done(TAG, `test event: ${rep.filter(r => r.ok).length}/${names.length} sink(s) delivered`);
|
|
}
|
|
|
|
export function everyMs(args: Args): number {
|
|
const ms = durationMs(args.str('every', '15m'), 'm');
|
|
if (ms < MIN_EVERY_MS) throw new Error('--every: 5m at least (the stores rate-limit their APIs)');
|
|
return ms;
|
|
}
|
|
|
|
export async function run(args: Args): Promise<void> {
|
|
const sub = args._[0];
|
|
if (sub === 'test') return testCmd(args);
|
|
if (sub) throw new Error(USAGE);
|
|
const every = everyMs(args);
|
|
const sinks = chosenSinks(args);
|
|
const once = args.has('once');
|
|
let stopping = false, wake: (() => void) | null = null;
|
|
const stop = (sig: string): void => {
|
|
if (stopping) process.exit(130);
|
|
stopping = true;
|
|
log(TAG, `${sig}: finishing the pass in progress, then stopping`);
|
|
wake?.();
|
|
};
|
|
process.on('SIGINT', () => stop('SIGINT'));
|
|
process.on('SIGTERM', () => stop('SIGTERM'));
|
|
|
|
let n = 0, failures = 0;
|
|
log(TAG, `${once ? 'one pass' : `every ${Math.round(every / 60_000)}m`}, sinks: ${configuredSinks(sinks).join(', ') || 'none (events go to the log and stdout)'}${args.has('dry') ? ', dry run' : ''}`);
|
|
while (!stopping) {
|
|
n++;
|
|
let ok = false, count = 0, events = 0, errs = 0;
|
|
try {
|
|
const targets = await listTargets(args);
|
|
if (!targets.length) warn(TAG, 'no app folders: storebound apps add <folder> (or --dir <folder>)');
|
|
const now = Date.now();
|
|
const r = await runPass({ targets, prev: readState(), at: new Date(now).toISOString(), concurrency: 2, read: (t, prev) => readApp(t as Target, prev, now) });
|
|
count = targets.length; events = r.events.length; errs = r.errors.length;
|
|
for (const x of r.errors) warn(TAG, `${x.app}: ${x.error.slice(0, 200)}`);
|
|
if (r.baseline.length) log(TAG, `baseline recorded for ${r.baseline.join(', ')} (no events on a first read)`);
|
|
for (const e of r.events) { emit(e, args); await deliver(e, sinks, args); }
|
|
if (!args.has('dry')) { appendEvents(r.events); writeState(r.state); }
|
|
ok = !(count > 0 && errs >= count);
|
|
} catch (e) { errs++; warn(TAG, `pass failed: ${(e as Error).message.slice(0, 300)}`); }
|
|
failures = ok ? 0 : failures + 1;
|
|
const wait = Math.min(MAX_BACKOFF_MS, every * 2 ** Math.min(failures, 6)) * (failures ? 1 : 0.9 + Math.random() * 0.2);
|
|
log(TAG, `pass ${n}: ${count} app(s), ${events} event(s), ${errs} error(s)${once || stopping ? '' : `, next in ${Math.round(wait / 60_000)}m${failures ? ` (backoff after ${failures} failed pass(es))` : ''}`}`);
|
|
if (once || stopping) break;
|
|
await new Promise<void>(res => { const t = setTimeout(res, wait); wake = () => { clearTimeout(t); res(); }; });
|
|
wake = null;
|
|
}
|
|
if (once && failures) process.exitCode = 1;
|
|
log(TAG, `stopped (state ${stateFile()}, events ${eventsFile()})`);
|
|
}
|