Something went wrong. Try again.
A fully featured CLI used as an alternative to AppView applications (Bsky.app/mu.social)
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114import { readState, writeState } from './state.ts';import { CliError, EXIT } from './errors.ts';import type { Context } from './context.ts';import type { OptionSpec } from './command.ts';import { parseDuration } from './time.ts';import { info } from './output.ts';
// Watching: a command that keeps running and prints what happens as it does.//// Bluesky has no push channel a client like this can use — direct messages// are not on the firehose, and push notifications go only to the official// apps — so watching is polling, the way the app itself does it: cheaply,// with a cursor, so each poll asks only for what is new. The pace starts at// --interval and slows, up to six times that, while nothing happens; anything// happening brings it straight back.//// A watch ends when told to (--count, --timeout, --once), on Ctrl-C, or when// the session can no longer be used (exit 3). A network hiccup, or a server// error, is retried with the pace slowed, and said on stderr: one bad poll// does not end a watch that may have been running for days.//// With --resume, where it got to is kept after every poll — per account, in// atcli's state file beside the config — so the next watch carries on from there,// neither repeating nor missing anything.
export const WATCH_OPTIONS: OptionSpec[] = [ { name: 'interval', type: 'string', value: 'duration', description: 'how often to look, at the most (default: 10s for messages, 30s for notifications, as the Bluesky app does); it slows to six times that while nothing happens' }, { name: 'count', type: 'number', value: 'n', description: 'stop after this many events' }, { name: 'timeout', type: 'string', value: 'duration', description: 'stop after this long, events or not (e.g. 10m)' }, { name: 'once', type: 'boolean', description: 'print what is new since the cursor (or --resume) and stop, rather than keep watching' }, { name: 'resume', type: 'boolean', description: 'carry on from where the last --resume watch of this account stopped, and keep track for the next' }, { name: 'cursor', type: 'string', value: 'cursor', description: 'carry on from here: the "cursor" of an event a watch printed with --json' },];
export interface WatchEvent { /** What --json prints for it, one line each. */ data: unknown; /** What a person sees. */ text: () => string; /** Where a watch that stopped right after this event carries on from. */ cursor: string;}
export interface Poller { /** Find out where "now" is, before the first poll (when not resuming or given a cursor). */ start(): Promise<void>; /** What is new since the last poll. */ poll(): Promise<WatchEvent[]>; /** Where it has got to, for --resume and for each event's JSON. */ cursor(): string | undefined;}
export function savedCursor(key: string): string | undefined { return readState(`watch:${key}`);}
function saveCursor(key: string, cursor: string): void { writeState(`watch:${key}`, cursor);}
/** Run a watch until it is told to stop. Returns how many events it printed. */export async function watch(ctx: Context, poller: Poller, { key, interval: fallback, fresh }: { key: string; interval: number; fresh: boolean }): Promise<number> { if (ctx.format === 'markdown' || ctx.format === 'html') throw ctx.usage('a watch prints text, or JSON with one event per line (--json); not Markdown or HTML'); const durationOf = (name: string) => { const text = ctx.str(name); try { return text === undefined ? undefined : parseDuration(text); } catch (error) { throw ctx.usage(`--${name}: ${(error as Error).message}`); } }; const base = Math.max(100, durationOf('interval') ?? fallback); const slowest = base * 6; const limit = ctx.num('count'); const timeout = durationOf('timeout'); const deadline = timeout === undefined ? Infinity : Date.now() + timeout; const once = ctx.flag('once'); const resume = ctx.flag('resume');
if (fresh) await poller.start(); if (!ctx.json && !once && process.stderr.isTTY) info(`watching; Ctrl-C to stop`);
let printed = 0; let pace = base; for (;;) { let events: WatchEvent[] = []; try { events = await poller.poll(); } catch (error) { // An ended session, or a mistake on the command line, is not something to wait out. if (error instanceof CliError && (error.exit === EXIT.auth || error.exit === EXIT.usage || error.exit === EXIT.notFound)) throw error; if (once) throw error; info(`watch: ${(error as Error).message}; trying again`); pace = Math.min(slowest, pace * 2); } let cursor = poller.cursor(); for (const event of events) { ctx.out(ctx.json ? `${JSON.stringify(event.data)}\n` : `${event.text()}\n`); printed++; // Stopping part way through what one poll brought: carry on after // the last event printed, not after the last one fetched. if (limit !== undefined && printed >= limit) { cursor = event.cursor; break; } } if (resume && cursor) saveCursor(key, cursor); if (once || (limit !== undefined && printed >= limit) || Date.now() >= deadline) return printed; pace = events.length ? base : Math.min(slowest, Math.round(pace * 1.5)); await new Promise((resolve) => setTimeout(resolve, Math.max(0, Math.min(pace, deadline - Date.now())))); if (Date.now() >= deadline) return printed; }}