CRT/storebound
1
0
Fork 0
Code Issues Pull requests Releases 1 Packages 1 Activity Actions

watch: ingest sink (--notify ingest) posting events to a push ingest endpoint

Browse source
This commit is contained in:
CRTkafa 2026-10-23 15:30:18 +08:00
parent 5b3905679b
commit 487e138f9d
9 changed files with 236 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

2
cli/commands.ts
View file

@ -11,7 +11,7 @@ export const PLAIN: Record<string, [() => Promise<Plain>, string]> = {
'import': [() => import('./import'), 'fastlane [--dir fastlane] [--profile app|game|subscription-app] [--dry]: Appfile, deliver + supply metadata and screenshots -> config, store/*.md, shots frames (read-only on fastlane, never overwrites)'],
'install': [() => import('./install'), 'skills + MCP server for --agent claude|codex|cursor|copilot|windsurf|gemini|opencode'],
'mcp': [async () => ({ run: async () => (await import('./mcp')).serve() }), 'stdio MCP server wrapping these commands (for any MCP-capable AI tool)'],
'watch': [() => import('./watch'), 'store events for every app on this machine: rejected, approved, live, rollout halted, crash threshold, low reviews; loops every --every 15m or --once; sinks --notify os|ntfy|telegram|webhook; test sends a test event; --dry, --dir <folder>, --json (read-only on the stores)'],
'watch': [() => import('./watch'), 'store events for every app on this machine: rejected, approved, live, rollout halted, crash threshold, low reviews; loops every --every 15m or --once; sinks --notify os|ntfy|telegram|webhook|ingest; test sends a test event; --dry, --dir <folder>, --json (read-only on the stores)'],
'status': [() => import('./status'), '[--app <name|store id|folder>] [--json]: per platform the live and next version, build, track, review state and rollout %, without uploading (read-only; read by ship pipelines, docs/ship-step.md)'],
'events': [() => import('./events'), '[--since 24h] [--app <name>] [--kind rejected] [--json]: the events watch recorded in ~/.storebound/events.jsonl (read-only)'],
'serve': [() => import('./serve'), '--stdio: JSON-RPC session for the storebound desktop app (dry runs; writes need a host confirm token)'],

2
cli/data/flags.json
View file

@ -1696,7 +1696,7 @@
"storebound validate [--apple] [--play] [--json] [--src store/renders] [--no-local] [--no-health] [--no-scan]"
],
"watch": [
"storebound watch [--dir <folder>] [--every 15m] [--notify os,ntfy,telegram,webhook] [--once] [--dry] [--json]",
"storebound watch [--dir <folder>] [--every 15m] [--notify os,ntfy,telegram,webhook,ingest] [--once] [--dry] [--json]",
"storebound watch test [--notify ...] [--dry] sends one test event to every configured sink"
],
"web": [

103
cli/lib/watch-sinks.ts
View file

@ -1,29 +1,33 @@
// Where `storebound watch` sends events: OS notification, ntfy, Telegram, a webhook. All optional, configured from the
// 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
// watch.json: { "notify": ["os"], "ntfy": { "url", "topic", "token" }, "telegram": { "token", "chat" }, "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 } from 'node:fs';
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';
export const SINK_NAMES: readonly SinkName[] = ['os', 'ntfy', 'telegram', 'webhook'];
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> } = {};
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;
@ -32,6 +36,8 @@ export function sinkConfig(env: Env = process.env, file: string | null = watchCo
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 = {};
@ -96,7 +102,7 @@ export function osCommand(platform: string, title: string, message: string): { c
return null;
}
export type FetchLike = (url: string, init: { method: string; headers: Record<string, string>; body: string; signal?: AbortSignal }) => Promise<{ ok: boolean; status: number }>;
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 => {
@ -118,6 +124,88 @@ export async function post(fetchFn: FetchLike, r: HttpRequest, o: { timeoutMs?:
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. */
@ -133,5 +221,6 @@ export async function sendEvent(e: WatchEvent, c: SinkConfig, d: { fetch?: Fetch
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;
}

10
cli/lib/watch.ts
View file

@ -25,6 +25,8 @@ export interface WatchEvent {
detail: string;
/** The absolute console deep link for the app (App Store Connect / Play Console). */
url: string;
/** Review events only: the store's review id (dedupe key of the ingest sink). */
ref?: string;
}
/** One version on one store, reduced to what the diff needs. */
@ -189,8 +191,8 @@ export function diffApp(prev: AppSnap | undefined, next: AppSnap, at: string): W
if (!prev) return [];
const out: WatchEvent[] = [];
const ids = { ...prev.ids, ...next.ids };
const ev = (store: WatchStore, kind: EventKind, version: string | null, from: string | null, to: string | null, detail: string): void => {
out.push({ at, app: next.name, store, kind, version, from, to, detail, url: eventUrl(kind, store, ids) });
const ev = (store: WatchStore, kind: EventKind, version: string | null, from: string | null, to: string | null, detail: string, ref?: string): void => {
out.push({ at, app: next.name, store, kind, version, from, to, detail, url: eventUrl(kind, store, ids), ...(ref ? { ref } : {}) });
};
for (const store of ['apple', 'play'] as const) {
@ -225,8 +227,8 @@ export function diffApp(prev: AppSnap | undefined, next: AppSnap, at: string): W
const seen = new Set(p.map(r => r.id));
for (const r of n) {
if (seen.has(r.id)) continue;
if (r.rating >= 1 && r.rating <= LOW_STARS) ev(store, 'review_low', null, null, `${r.rating} stars`, `${r.rating}-star review${r.replied ? '' : ', no reply yet'}: ${snippet(r.text)}`);
else if (!r.replied) ev(store, 'review_needs_reply', null, null, `${r.rating} stars`, `${r.rating}-star review without a reply: ${snippet(r.text)}`);
if (r.rating >= 1 && r.rating <= LOW_STARS) ev(store, 'review_low', null, null, `${r.rating} stars`, `${r.rating}-star review${r.replied ? '' : ', no reply yet'}: ${snippet(r.text)}`, r.id);
else if (!r.replied) ev(store, 'review_needs_reply', null, null, `${r.rating} stars`, `${r.rating}-star review without a reply: ${snippet(r.text)}`, r.id);
}
}
return out;

6
cli/watch.ts
View file

@ -1,5 +1,5 @@
// `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 [--dir <folder>] [--every 15m] [--notify os,ntfy,telegram,webhook,ingest] [--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
@ -29,7 +29,7 @@ import {
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]';
const USAGE = 'usage: storebound watch [--dir <folder>] [--every 15m] [--notify os,ntfy,telegram,webhook,ingest] [--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;
@ -150,7 +150,7 @@ function emit(e: WatchEvent, args: Args): void {
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; }
if (!names.length) { warn(TAG, 'no sink configured: set STOREBOUND_NTFY_TOPIC, STOREBOUND_TELEGRAM_TOKEN + STOREBOUND_TELEGRAM_CHAT, STOREBOUND_WEBHOOK_URL, STOREBOUND_INGEST_URL + STOREBOUND_INGEST_TOKEN_FILE, 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; }