#!/usr/bin/env node
/**
 * A reference annotator for an ordner.fyi install: it signs a person in,
 * reads their files and writes back what it found. It uses Node's built-ins
 * only (Node.js 22.18 or later runs this file directly), so it can be read and
 * copied without anything else.
 *
 *   node annotate-example.ts --jmap https://example.org \
 *     [--issuer https://example.org/auth] [--state ./annotate-state.json] \
 *     [--follow 0] [--name "Example Annotator"]
 *
 * --jmap    the install's origin, its /.well-known/jmap, or a JMAP session URL
 * --issuer  the authorization server, when the JMAP answer does not name it
 * --state   where the registration and the refresh token are kept between runs
 * --follow  seconds to keep listening for new files after the queue is empty
 * --name    the client_name the person sees on the approval page
 *
 * The steps, in the order the functions below appear:
 *
 *   discover → register → signIn → session → defineField → defineView
 *   → annotateQueue → follow
 *
 * Two functions stand in for real intelligence and are marked as such:
 * `label` (where a model reads a file and says what it is) and `embed`
 * (where an embedding model turns content into a vector).
 */

import { createHash } from "node:crypto";
import { readFile, rename, writeFile } from "node:fs/promises";
import { setTimeout as sleep } from "node:timers/promises";
import { parseArgs } from "node:util";

// --- constants --------------------------------------------------------------

const CORE = "urn:ietf:params:jmap:core";
const FILENODE = "urn:ietf:params:jmap:filenode";
/** The install's own additions to FileNode: annotations, fields, views, vectors. */
const FILES = "https://ns.ordner.fyi/svc/jmap/files";
/** Pictures: the `photo` property, with its faces. Absent on an install without Photos. */
const PHOTOS = "https://ns.ordner.fyi/svc/jmap/photos";

const DEVICE_GRANT = "urn:ietf:params:oauth:grant-type:device_code";
/**
 * What this client asks for. `drive:read` reads every file, `drive:annotate`
 * writes annotations (never content, names or places), `drive:faces` reads
 * and writes face embeddings, which the person may untick on the approval
 * page, and `offline_access` gives a refresh token so the next run needs no
 * approval.
 */
const SCOPE = "openid offline_access drive:read drive:annotate drive:faces";

/** The text of a file is uploaded only up to this size; larger is refused. */
const MAX_TEXT = 1024 * 1024;
/** How many queue entries one request takes. */
const PAGE = 50;

// --- small helpers ----------------------------------------------------------

type Json = null | boolean | number | string | Json[] | { [k: string]: Json };
type Obj = { [k: string]: Json };

const say = (line: string): void => {
	process.stdout.write(`${line}\n`);
};

class Failure extends Error {}

async function json(res: Response, what: string): Promise<Obj> {
	const text = await res.text();
	if (!res.ok)
		throw new Failure(`${what}: HTTP ${res.status} ${text.slice(0, 500)}`);
	return JSON.parse(text) as Obj;
}

/** RFC 6570 §3.2.2 (level 1): each value percent-encoded but for unreserved characters. */
function expand(template: string, vars: Record<string, string>): string {
	return template.replace(/\{([A-Za-z0-9_]+)\}/g, (_, name: string) =>
		encodeURIComponent(vars[name] ?? "").replace(
			/[!'()*]/g,
			(c) => `%${c.charCodeAt(0).toString(16).toUpperCase()}`,
		),
	);
}

const MONTHS = "JanFebMarAprMayJunJulAugSepOctNovDec";

/** RFC 9110 §10.2.3: seconds, or an HTTP date (§5.6.7's IMF-fixdate); one second otherwise. */
function retryAfter(res: Response): number {
	const v = res.headers.get("retry-after")?.trim();
	if (v === undefined) return 1000;
	if (/^\d+$/.test(v)) return Number(v) * 1000;
	const m =
		/^[A-Z][a-z]{2}, (\d{2}) ([A-Z][a-z]{2}) (\d{4}) (\d{2}):(\d{2}):(\d{2}) GMT$/.exec(
			v,
		);
	const month = m === null ? -1 : MONTHS.indexOf(m[2] as string) / 3;
	if (m === null || !Number.isInteger(month)) return 1000;
	const [, day, , year, h, min, sec] = m.map(Number) as number[];
	const at = Date.UTC(year as number, month, day, h, min, sec);
	return Math.max(0, at - Date.now());
}

// --- the state kept between runs -----------------------------------------

/**
 * One registration per installation of the application, kept with the
 * refresh token. Registering on every run would leave a trail of clients
 * the person has to approve and clean up.
 */
interface State {
	issuer?: string;
	client?: {
		client_id: string;
		/** RFC 7592 §3: where the registration is read, changed or deleted. */
		registration_client_uri?: string;
		registration_access_token?: string;
	};
	refresh_token?: string;
	scope?: string;
}

async function loadState(path: string): Promise<State> {
	try {
		return JSON.parse(await readFile(path, "utf8")) as State;
	} catch (err) {
		if ((err as NodeJS.ErrnoException).code === "ENOENT") return {};
		throw err;
	}
}

/** Written whole and renamed into place, readable by its owner only: it holds a refresh token. */
async function saveState(path: string, state: State): Promise<void> {
	const tmp = `${path}.tmp`;
	await writeFile(tmp, `${JSON.stringify(state, null, "\t")}\n`, {
		mode: 0o600,
	});
	await rename(tmp, path);
}

// --- 1. discover ----------------------------------------------------------

interface Discovered {
	/** The JMAP session resource, RFC 8620 §2. */
	sessionUrl: string;
	issuer: string;
	registration_endpoint: string;
	device_authorization_endpoint: string;
	token_endpoint: string;
}

/**
 * From the JMAP URL alone to the authorization server's endpoints.
 *
 * 1. The session is asked for without credentials. The 401 answer's
 *    `WWW-Authenticate: Bearer … resource_metadata="…"` names the protected
 *    resource's metadata (RFC 9728 §5.1).
 * 2. That document lists `authorization_servers` (RFC 9728 §2).
 * 3. The authorization server's metadata (RFC 8414 §3) names the
 *    registration, device authorization and token endpoints.
 *
 * `--issuer` stands in for steps 1 and 2 when a server's 401 names no
 * metadata.
 */
async function discover(
	jmap: string,
	issuerOverride: string | undefined,
): Promise<Discovered> {
	const given = new URL(jmap);
	// RFC 8620 §2.2: an origin alone means its well-known session location.
	const asked =
		given.pathname === "/" || given.pathname === ""
			? new URL("/.well-known/jmap", given)
			: given;
	const anonymous = await fetch(asked, { redirect: "follow" });
	await anonymous.body?.cancel();
	const sessionUrl = anonymous.url || asked.toString();

	let issuer = issuerOverride;
	if (issuer === undefined) {
		const challenge = anonymous.headers.get("www-authenticate") ?? "";
		// RFC 9728 §5.1: an auth-param of the Bearer challenge.
		const m = /resource_metadata="([^"]+)"/i.exec(challenge);
		if (anonymous.status !== 401 || m === null)
			throw new Failure(
				`${sessionUrl} answered ${anonymous.status} without resource_metadata; pass --issuer`,
			);
		const resource = await json(
			await fetch(new URL(m[1] as string, sessionUrl)),
			"protected resource metadata",
		);
		const servers = resource.authorization_servers;
		if (!Array.isArray(servers) || typeof servers[0] !== "string")
			throw new Failure("the protected resource names no authorization server");
		issuer = servers[0];
	}
	issuer = issuer.replace(/\/+$/, "");

	const meta = await authorizationServerMetadata(issuer);
	// RFC 8414 §3.3: the metadata must be the issuer's own.
	if (meta.issuer !== issuer)
		throw new Failure(`metadata names issuer ${meta.issuer}, not ${issuer}`);
	for (const k of [
		"registration_endpoint",
		"device_authorization_endpoint",
		"token_endpoint",
	])
		if (typeof meta[k] !== "string")
			throw new Failure(`the authorization server lists no ${k}`);
	return {
		sessionUrl,
		issuer,
		registration_endpoint: meta.registration_endpoint as string,
		device_authorization_endpoint: meta.device_authorization_endpoint as string,
		token_endpoint: meta.token_endpoint as string,
	};
}

/**
 * RFC 8414 §3.1 puts the well-known part between the host and the issuer's
 * path (`https://h/.well-known/oauth-authorization-server/auth`); an OpenID
 * provider also answers OpenID Connect Discovery 1.0 §4, which appends it
 * (`https://h/auth/.well-known/openid-configuration`). The first is asked
 * first.
 */
async function authorizationServerMetadata(issuer: string): Promise<Obj> {
	const u = new URL(issuer);
	const path = u.pathname === "/" ? "" : u.pathname;
	const rfc8414 = `${u.origin}/.well-known/oauth-authorization-server${path}`;
	const res = await fetch(rfc8414);
	if (res.ok) return json(res, rfc8414);
	await res.body?.cancel();
	const oidc = `${issuer}/.well-known/openid-configuration`;
	return json(await fetch(oidc), oidc);
}

// --- 2. register ----------------------------------------------------------

/**
 * Dynamic client registration (RFC 7591 §3.1), once per installation.
 *
 * The install registers native applications only, so `redirect_uris` must
 * be loopback (RFC 8252 §7.3) even for a client that signs in with the
 * device grant alone, and `grant_types` must include `authorization_code`.
 * `token_endpoint_auth_method: "none"`: a public client holds no secret
 * (RFC 8252 §8.4). `client_name` is shown to the person, marked as not
 * verified.
 *
 * A kept registration is checked first with RFC 7592 §2.1's read: an
 * install removes a registration that is never used, and an administrator
 * may ban one; either answers 401, after which the client registers again.
 */
async function register(
	d: Discovered,
	state: State,
	name: string,
	statePath: string,
): Promise<string> {
	if (state.client !== undefined && state.issuer === d.issuer) {
		const { registration_client_uri: uri, registration_access_token: token } =
			state.client;
		if (uri === undefined || token === undefined) return state.client.client_id;
		const res = await fetch(uri, {
			headers: { authorization: `Bearer ${token}` },
		});
		await res.body?.cancel();
		if (res.ok) {
			say(`using client ${state.client.client_id}`);
			return state.client.client_id;
		}
		if (res.status !== 401 && res.status !== 404)
			throw new Failure(`reading the registration: HTTP ${res.status}`);
		say(`client ${state.client.client_id} is gone; registering again`);
		delete state.refresh_token;
	}
	const answer = await json(
		await fetch(d.registration_endpoint, {
			method: "POST",
			headers: { "content-type": "application/json" },
			body: JSON.stringify({
				client_name: name,
				redirect_uris: ["http://127.0.0.1/callback"],
				token_endpoint_auth_method: "none",
				grant_types: ["authorization_code", "refresh_token", DEVICE_GRANT],
				scope: SCOPE,
			}),
		}),
		"registration",
	);
	state.issuer = d.issuer;
	state.client = {
		client_id: answer.client_id as string,
		registration_client_uri: answer.registration_client_uri as string,
		registration_access_token: answer.registration_access_token as string,
	};
	delete state.refresh_token;
	await saveState(statePath, state);
	say(`registered client ${state.client.client_id}`);
	return state.client.client_id;
}

// --- 3. sign in (device grant) -------------------------------------------

interface Tokens {
	access: string;
	/** The scopes granted: the person may have unticked some (RFC 6749 §5.1). */
	scopes: Set<string>;
}

async function tokenRequest(
	d: Discovered,
	form: Record<string, string>,
): Promise<{ status: number; body: Obj }> {
	const res = await fetch(d.token_endpoint, {
		method: "POST",
		headers: { "content-type": "application/x-www-form-urlencoded" },
		body: new URLSearchParams(form).toString(),
	});
	const text = await res.text();
	let body: Obj = {};
	try {
		body = JSON.parse(text) as Obj;
	} catch {
		body = { error: "invalid_response", error_description: text.slice(0, 200) };
	}
	return { status: res.status, body };
}

/** Keeps a rotated refresh token and the granted scope, and answers the access token. */
async function took(
	body: Obj,
	state: State,
	statePath: string,
): Promise<Tokens> {
	if (typeof body.refresh_token === "string")
		state.refresh_token = body.refresh_token;
	// RFC 6749 §5.1: absent, the scope is the one asked.
	const scope =
		typeof body.scope === "string" ? body.scope : (state.scope ?? SCOPE);
	state.scope = scope;
	await saveState(statePath, state);
	return {
		access: body.access_token as string,
		scopes: new Set(scope.split(" ")),
	};
}

/** RFC 6749 §6: a new access token from the kept refresh token, or null when it no longer works. */
async function refresh(
	d: Discovered,
	clientId: string,
	state: State,
	statePath: string,
): Promise<Tokens | null> {
	if (state.refresh_token === undefined) return null;
	const { status, body } = await tokenRequest(d, {
		grant_type: "refresh_token",
		refresh_token: state.refresh_token,
		client_id: clientId,
	});
	if (status === 200) return took(body, state, statePath);
	// RFC 6749 §5.2: the person revoked the application, or the chain ended.
	if (body.error === "invalid_grant") {
		delete state.refresh_token;
		await saveState(statePath, state);
		return null;
	}
	throw new Failure(`refresh: ${body.error} ${body.error_description ?? ""}`);
}

/**
 * The device authorization grant (RFC 8628), for an application without a
 * browser of its own: it shows a code, the person opens the address on any
 * device, signs in, reads what the application asks for and approves.
 * Meanwhile the application polls the token endpoint.
 */
async function signIn(
	d: Discovered,
	clientId: string,
	state: State,
	statePath: string,
): Promise<Tokens> {
	const again = await refresh(d, clientId, state, statePath);
	if (again !== null) {
		say("signed in with the kept refresh token");
		return again;
	}
	// RFC 8628 §3.1
	const started = await json(
		await fetch(d.device_authorization_endpoint, {
			method: "POST",
			headers: { "content-type": "application/x-www-form-urlencoded" },
			body: new URLSearchParams({
				client_id: clientId,
				scope: SCOPE,
			}).toString(),
		}),
		"device authorization",
	);
	// RFC 8628 §3.3: what the person is told.
	say(
		`Open ${started.verification_uri} and enter the code ${started.user_code}`,
	);
	if (typeof started.verification_uri_complete === "string")
		say(`(or open ${started.verification_uri_complete})`);

	// RFC 8628 §3.2: `interval` defaults to 5 seconds.
	let interval =
		typeof started.interval === "number" ? started.interval * 1000 : 5000;
	const until = Date.now() + (started.expires_in as number) * 1000;
	while (Date.now() < until) {
		await sleep(interval);
		// RFC 8628 §3.4
		const { status, body } = await tokenRequest(d, {
			grant_type: DEVICE_GRANT,
			device_code: started.device_code as string,
			client_id: clientId,
		});
		if (status === 200) {
			const tokens = await took(body, state, statePath);
			say(`approved; granted ${[...tokens.scopes].join(" ")}`);
			return tokens;
		}
		// RFC 8628 §3.5
		switch (body.error) {
			case "authorization_pending":
				continue;
			case "slow_down":
				interval += 5000;
				continue;
			case "access_denied":
				throw new Failure("the person declined");
			case "expired_token":
				throw new Failure("the code expired before it was approved; run again");
			default:
				throw new Failure(
					`token: ${body.error} ${body.error_description ?? ""}`,
				);
		}
	}
	throw new Failure("the code expired before it was approved; run again");
}

// --- 4. the JMAP session --------------------------------------------------

interface Session {
	apiUrl: string;
	downloadUrl: string;
	uploadUrl: string;
	eventSourceUrl: string;
	accountId: string;
	using: string[];
	/** Whether pictures and their faces are served. */
	photos: boolean;
	maxObjectsInGet: number;
	maxObjectsInSet: number;
	maxCallsInRequest: number;
}

/**
 * A JMAP client over one access token: a 401 refreshes it once and asks
 * again, a 429 waits as long as `Retry-After` says (the person's request
 * limits apply to their applications).
 */
class Client {
	readonly #d: Discovered;
	readonly clientId: string;
	readonly #state: State;
	readonly #statePath: string;
	tokens: Tokens;
	session!: Session;

	constructor(
		d: Discovered,
		clientId: string,
		state: State,
		statePath: string,
		tokens: Tokens,
	) {
		this.#d = d;
		this.clientId = clientId;
		this.#state = state;
		this.#statePath = statePath;
		this.tokens = tokens;
	}

	/** Any authorised request; `init` may be a function so a retried body is made again. */
	async fetch(
		url: string,
		init: () => RequestInit = () => ({}),
	): Promise<Response> {
		for (let attempt = 0; ; attempt++) {
			const i = init();
			const res = await fetch(url, {
				...i,
				headers: {
					...(i.headers as Record<string, string>),
					authorization: `Bearer ${this.tokens.access}`,
				},
			});
			if (res.status === 401 && attempt === 0) {
				await res.body?.cancel();
				const fresh = await refresh(
					this.#d,
					this.clientId,
					this.#state,
					this.#statePath,
				);
				if (fresh === null)
					throw new Failure(
						"the person revoked this application; run again to sign in",
					);
				this.tokens = fresh;
				continue;
			}
			if (res.status === 429 && attempt < 10) {
				await res.body?.cancel();
				await sleep(retryAfter(res));
				continue;
			}
			return res;
		}
	}

	/** RFC 8620 §3.3: one request of method calls; a method-level error throws. */
	async call(calls: [string, Obj, string][]): Promise<Map<string, Obj>> {
		if (calls.length > this.session.maxCallsInRequest)
			throw new Failure(
				"more method calls than the server takes in one request",
			);
		const res = await this.fetch(this.session.apiUrl, () => ({
			method: "POST",
			headers: { "content-type": "application/json" },
			body: JSON.stringify({ using: this.session.using, methodCalls: calls }),
		}));
		const body = await json(res, "JMAP request");
		const out = new Map<string, Obj>();
		for (const [name, args, tag] of body.methodResponses as [
			string,
			Obj,
			string,
		][]) {
			// RFC 8620 §3.6.2
			if (name === "error")
				throw new Failure(`${tag}: ${args.type} ${args.description ?? ""}`);
			out.set(tag, args);
		}
		return out;
	}
}

/**
 * RFC 8620 §2: the session names the API, upload, download and push URLs
 * (relative ones resolve against the session's own URL), the accounts and
 * the server's limits. The person's drive is the primary account of the
 * FileNode capability.
 */
async function session(c: Client, sessionUrl: string): Promise<Session> {
	const s = await json(await c.fetch(sessionUrl), "session");
	const caps = s.capabilities as Obj;
	const primary = s.primaryAccounts as Record<string, string>;
	const accountId = primary[FILENODE];
	if (accountId === undefined) throw new Failure("no drive is reachable");
	if (caps[FILES] === undefined)
		throw new Failure(`this server does not speak ${FILES}`);
	const core = caps[CORE] as Record<string, number>;
	const photos = caps[PHOTOS] !== undefined;
	// The URLs may be relative to the session's; resolved as text, so that
	// URL() does not percent-encode a template's braces.
	const at = (u: unknown): string => {
		const t = u as string;
		if (/^[a-z][a-z0-9+.-]*:/i.test(t)) return t;
		const base = new URL(sessionUrl);
		return t.startsWith("/")
			? `${base.origin}${t}`
			: `${base.origin}${base.pathname.replace(/[^/]*$/, "")}${t}`;
	};
	return {
		apiUrl: at(s.apiUrl),
		downloadUrl: at(s.downloadUrl),
		uploadUrl: at(s.uploadUrl),
		eventSourceUrl: at(s.eventSourceUrl),
		accountId,
		using: [CORE, FILENODE, FILES, ...(photos ? [PHOTOS] : [])],
		photos,
		maxObjectsInGet: core.maxObjectsInGet ?? 50,
		maxObjectsInSet: core.maxObjectsInSet ?? 50,
		maxCallsInRequest: core.maxCallsInRequest ?? 4,
	};
}

// --- 5. a field of its own ------------------------------------------------

/**
 * Beyond a description and labels, an application declares typed fields
 * of its own. The server indexes them by type, never by meaning: `facet`
 * offers the field as a filter with counts, `searchable` puts its text in
 * the full-text search, `shown` shows it in the file's details. The call
 * replaces the application's whole schema, so it names every field.
 */
async function defineField(c: Client): Promise<string> {
	await c.call([
		[
			"AnnotationSchema/set",
			{
				accountId: c.session.accountId,
				schema: {
					fields: {
						category: {
							type: "string",
							title: { "x-default": "Category", de: "Kategorie" },
							shown: true,
							facet: true,
							searchable: true,
						},
					},
				},
			},
			"s",
		],
	]);
	say("field category defined");
	return ownSource(c);
}

/**
 * What the application writes is kept under its source: who signed it in
 * and which application it is. The server makes the id from the
 * credential, never from anything the client claims:
 * `<actorId>.app.<clientId>`, the person's id and this registration's
 * `client_id`. Two people using the same application are two sources. The
 * schema just written is the one whose `source.app.id` is this client.
 */
async function ownSource(c: Client): Promise<string> {
	const got = await c.call([
		[
			"AnnotationSchema/get",
			{ accountId: c.session.accountId, ids: null },
			"g",
		],
	]);
	const list = (got.get("g") as Obj).list as Obj[];
	const mine = list.find(
		(s) => ((s.source as Obj).app as Obj | null)?.id === c.clientId,
	);
	if (mine === undefined)
		throw new Failure("the schema just written is not listed");
	return mine.id as string;
}

// --- 6. a view of its own -------------------------------------------------

/**
 * A view is a saved query the person's Files lists beside its own places.
 * It is not a folder: it lists what the query answers now, so writing the
 * field keeps it current. `key` is the application's own name for it: a
 * create with a key it used before updates that view.
 */
async function defineView(c: Client, source: string): Promise<void> {
	const out = await c.call([
		[
			"AnnotationView/set",
			{
				accountId: c.session.accountId,
				create: {
					v: {
						key: "documents",
						title: { "x-default": "Documents", de: "Dokumente" },
						in: "files",
						layout: "list",
						filter: { field: { source, name: "category", equals: "document" } },
						columns: ["category"],
					},
				},
			},
			"v",
		],
	]);
	const refused = ((out.get("v") as Obj).notCreated as Obj | null)?.v;
	if (refused) throw new Failure(`view: ${JSON.stringify(refused)}`);
	say("view documents defined");
}

// --- 7. the queue ---------------------------------------------------------

interface FileNode {
	id: string;
	name: string;
	type: string | null;
	size: number | null;
	blobId: string | null;
	/** A read picture's metadata, or null (absent without the Photos capability). */
	photo?: Obj | null;
}

interface Found {
	category: "picture" | "document" | "audio" | "video" | "other";
	description: string;
	labels: { name: string; confidence: number }[];
}

/*
 * ===================================================================
 *  THIS IS WHERE A MODEL GOES.
 *
 *  `label` stands in for whatever reads a file and says what it is: a
 *  vision model for a picture, a classifier for a document. It is given
 *  the node (and could download its content, or a picture's smaller
 *  renditions) and answers a description, labels with a confidence, and
 *  this application's own field. This one only looks at the media type.
 * ===================================================================
 */
function label(file: FileNode): Found {
	const type = (file.type ?? "").split(";", 1)[0]?.trim().toLowerCase() ?? "";
	const top = type.split("/", 1)[0];
	const category: Found["category"] =
		top === "image"
			? "picture"
			: top === "audio"
				? "audio"
				: top === "video"
					? "video"
					: top === "text" ||
							type === "application/pdf" ||
							type === "application/rtf" ||
							type === "application/msword" ||
							type.startsWith(
								"application/vnd.openxmlformats-officedocument.",
							) ||
							type.startsWith("application/vnd.oasis.opendocument.")
						? "document"
						: "other";
	return {
		category,
		description: `A ${category} (${type || "unknown type"}), by its media type.`,
		labels: [{ name: category, confidence: 0.5 }],
	};
}

/*
 * ===================================================================
 *  THIS IS WHERE AN EMBEDDING MODEL GOES.
 *
 *  `embed` stands in for a text or image embedding model. A real one
 *  answers vectors whose closeness means similar content, so that
 *  FileNode/query's `similarTo` finds related files. This one makes a
 *  deterministic 64-dimension vector from a SHA-256 of what it is given,
 *  which is close only to itself. The model name says what made the
 *  vector and its version: vectors of different models are never compared.
 * ===================================================================
 */
function embed(bytes: Uint8Array): Float32Array {
	const v = new Float32Array(64);
	for (let block = 0; block < 2; block++) {
		const h = createHash("sha256").update(bytes).update(String(block)).digest();
		for (let i = 0; i < 32; i++)
			v[block * 32 + i] = ((h[i] as number) - 127.5) / 127.5;
	}
	return unit(v);
}

function unit(v: Float32Array): Float32Array {
	let n = 0;
	for (const x of v) n += x * x;
	n = Math.sqrt(n) || 1;
	return v.map((x) => x / n);
}

/** A vector as the server takes it: base64 of little-endian float32. */
function vectorText(v: Float32Array): string {
	const out = Buffer.alloc(v.length * 4);
	for (let i = 0; i < v.length; i++) out.writeFloatLE(v[i] as number, i * 4);
	return out.toString("base64");
}

/** A dummy face embedding: a unit vector along the first axis. */
const DUMMY_FACE = (() => {
	const v = new Float32Array(64);
	v[0] = 1;
	return vectorText(v);
})();

const essence = (type: string | null) =>
	(type ?? "").split(";", 1)[0]?.trim().toLowerCase() ?? "";

/** RFC 8620 §6.2: a blob by the download URL template. */
async function download(
	c: Client,
	f: FileNode,
): Promise<Uint8Array<ArrayBuffer>> {
	const url = expand(c.session.downloadUrl, {
		accountId: c.session.accountId,
		blobId: f.blobId as string,
		name: f.name,
		type: essence(f.type) || "application/octet-stream",
	});
	const res = await c.fetch(url);
	if (!res.ok) throw new Failure(`download ${f.name}: HTTP ${res.status}`);
	return new Uint8Array(await res.arrayBuffer());
}

/** RFC 8620 §6.1: the file's text, uploaded; the annotation names the upload's blobId. */
async function uploadText(
	c: Client,
	text: Uint8Array<ArrayBuffer>,
): Promise<string> {
	const url = expand(c.session.uploadUrl, { accountId: c.session.accountId });
	const res = await c.fetch(url, () => ({
		method: "POST",
		headers: { "content-type": "text/plain; charset=utf-8" },
		body: text,
	}));
	return (await json(res, "upload")).blobId as string;
}

/**
 * Annotates every file this application has not annotated yet, a page at a
 * time, until none is left.
 *
 * `isAnnotated: false` is the application's own queue: the files whose
 * current content it has not annotated (none, or an annotation of older
 * content). Query and get go in one request: the get's `#ids` takes the
 * query's ids by back-reference (RFC 8620 §3.7).
 *
 * Every write names the `blobId` it was made from. When the file changed
 * since it was read the write is refused with `contentChanged`, and the
 * file comes back in the queue with its new content.
 */
async function annotateQueue(c: Client): Promise<number> {
	const s = c.session;
	const faces = s.photos;
	const mayEmbedFaces = c.tokens.scopes.has("drive:faces");
	const limit = Math.min(PAGE, s.maxObjectsInGet, s.maxObjectsInSet);
	const properties = [
		"id",
		"name",
		"type",
		"size",
		"blobId",
		...(faces ? ["photo"] : []),
	];
	let done = 0;
	/** Files refused for another reason than contentChanged: marked read with nothing to say. */
	const giveUp = new Set<string>();
	for (;;) {
		const page = await c.call([
			[
				"FileNode/query",
				{ accountId: s.accountId, filter: { isAnnotated: false }, limit },
				"q",
			],
			[
				"FileNode/get",
				{
					accountId: s.accountId,
					"#ids": { resultOf: "q", name: "FileNode/query", path: "/ids" },
					properties,
				},
				"g",
			],
		]);
		const files = ((page.get("g") as Obj).list as unknown as FileNode[]).filter(
			(f) => f.blobId !== null,
		);
		if (files.length === 0) return done;

		const annotations: Obj = {};
		const vectors: Obj = {};
		for (const f of files) {
			const blobId = f.blobId as string;
			if (giveUp.has(f.id)) {
				// `{ blobId }` alone records "read, nothing to say": the queue moves on.
				annotations[f.id] = { blobId };
				continue;
			}
			const found = label(f);
			const update: Obj = {
				blobId,
				description: found.description,
				labels: found.labels,
				fields: { category: found.category },
			};
			let bytes: Uint8Array<ArrayBuffer> = new TextEncoder().encode(
				`${f.name}\n${f.type}`,
			);
			if (essence(f.type) === "text/plain" && (f.size ?? 0) <= MAX_TEXT) {
				bytes = await download(c, f);
				// The text must be UTF-8; other bytes are left without text.
				try {
					new TextDecoder("utf-8", { fatal: true }).decode(bytes);
					update.text = await uploadText(c, bytes);
				} catch (err) {
					if (err instanceof Failure) throw err;
				}
			}
			if (faces && f.photo != null) {
				// One region per face: its centre and size as fractions of the
				// picture as displayed. The embedding needs drive:faces; without
				// it a face is written without one.
				update.faces = [
					{
						area: { x: 0.5, y: 0.4, w: 0.2, h: 0.25 },
						confidence: 0.5,
						...(mayEmbedFaces
							? {
									embedding: { model: "example-face-64@1", vector: DUMMY_FACE },
								}
							: {}),
					},
				];
			}
			annotations[f.id] = update;
			vectors[f.id] = {
				blobId,
				vectors: [
					{
						model: "example-hash-64@1",
						kind: found.category === "picture" ? "image" : "text",
						vector: vectorText(embed(bytes)),
					},
				],
			};
		}

		const out = await c.call([
			[
				"FileAnnotation/set",
				{ accountId: s.accountId, update: annotations },
				"a",
			],
			...(Object.keys(vectors).length > 0
				? ([
						[
							"FileVector/set",
							{ accountId: s.accountId, update: vectors },
							"v",
						],
					] as [string, Obj, string][])
				: []),
		]);
		const a = out.get("a") as Obj;
		const refused = (a.notUpdated ?? {}) as Record<string, Obj>;
		const vRefused = (out.get("v")?.notUpdated ?? {}) as Record<string, Obj>;
		let progress = false;
		for (const f of files) {
			const why = refused[f.id];
			if (why === undefined) {
				progress = true;
				if (!giveUp.has(f.id)) {
					done++;
					const v = vRefused[f.id];
					const written = (a.updated as Record<string, Obj | null> | null)?.[
						f.id
					];
					const faceNote =
						written?.faces !== undefined
							? `, ${(written.faces as Json[]).filter((x) => x !== null).length} face(s)`
							: "";
					say(
						`annotated ${f.name} (${label(f).category}${faceNote}${v ? `; vector refused: ${v.type}` : ""})`,
					);
				}
			} else if (why.type === "contentChanged") {
				// Changed since it was read: it is in the queue again with its new content.
				say(`${f.name} changed while it was read; it comes back`);
				progress = true;
			} else if (!giveUp.has(f.id)) {
				say(`${f.name} refused: ${why.type} ${why.description ?? ""}`);
				giveUp.add(f.id);
				progress = true;
			} else throw new Failure(`${f.name} cannot be marked read: ${why.type}`);
		}
		if (!progress) throw new Failure("the queue does not move");
	}
}

// --- 8. follow push -------------------------------------------------------

/**
 * The JMAP EventSource (RFC 8620 §7.3) tells of every state change of the
 * account; a new or changed file changes the `FileNode` state, and the
 * queue is read again. The stream is read with fetch, because a browser's
 * EventSource cannot send an Authorization header; its format is the HTML
 * standard's text/event-stream (WHATWG HTML §9.2). After a dropped
 * connection the client reconnects with `Last-Event-ID`.
 */
async function follow(c: Client, seconds: number): Promise<void> {
	const until = Date.now() + seconds * 1000;
	const url = expand(c.session.eventSourceUrl, {
		types: "FileNode",
		closeafter: "no",
		ping: "30",
	});
	let lastEventId: string | undefined;
	let retry = 1000;
	let known: string | undefined;
	let running: Promise<void> | null = null;
	let again = false;
	const run = () => {
		if (running !== null) {
			again = true;
			return;
		}
		running = (async () => {
			do {
				again = false;
				const n = await annotateQueue(c);
				if (n > 0) say(`queue empty after ${n} file(s)`);
			} while (again);
		})().finally(() => {
			running = null;
		});
	};

	say(`following changes for ${seconds} s`);
	while (Date.now() < until) {
		const stop = AbortSignal.timeout(Math.max(1, until - Date.now()));
		try {
			const res = await c.fetch(url, () => ({
				headers: {
					accept: "text/event-stream",
					...(lastEventId !== undefined
						? { "last-event-id": lastEventId }
						: {}),
				},
				signal: stop,
			}));
			if (!res.ok || res.body === null)
				throw new Failure(`event source: HTTP ${res.status}`);
			// The stream tells of changes from now on: what changed before it
			// was open (since the last queue run, or while it was down) is
			// caught up by reading the queue once more.
			run();
			say("listening for changes");
			const decoder = new TextDecoder();
			let buffer = "";
			let event = {
				type: "message",
				data: [] as string[],
				id: undefined as string | undefined,
			};
			for await (const chunk of res.body) {
				buffer += decoder.decode(chunk, { stream: true });
				let nl = buffer.search(/\r\n|\r|\n/);
				while (nl >= 0) {
					const line = buffer.slice(0, nl);
					buffer = buffer.slice(nl + (buffer.startsWith("\r\n", nl) ? 2 : 1));
					nl = buffer.search(/\r\n|\r|\n/);
					if (line === "") {
						// A blank line dispatches the event.
						if (event.id !== undefined) lastEventId = event.id;
						if (event.type === "state" && event.data.length > 0) {
							const change = JSON.parse(event.data.join("\n")) as Obj;
							const changed = (change.changed as Record<string, Obj>)?.[
								c.session.accountId
							];
							const now = changed?.FileNode as string | undefined;
							if (now !== undefined && now !== known) {
								known = now;
								say("files changed");
								run();
							}
						}
						event = { type: "message", data: [], id: undefined };
						continue;
					}
					if (line.startsWith(":")) continue; // a comment, e.g. a keep-alive
					const colon = line.indexOf(":");
					const field = colon < 0 ? line : line.slice(0, colon);
					let value = colon < 0 ? "" : line.slice(colon + 1);
					if (value.startsWith(" ")) value = value.slice(1);
					if (field === "event") event.type = value;
					else if (field === "data") event.data.push(value);
					else if (field === "id") event.id = value;
					else if (field === "retry" && /^\d+$/.test(value))
						retry = Number(value);
				}
			}
		} catch (err) {
			if (stop.aborted) break;
			if (err instanceof Failure && /revoked/.test(err.message)) throw err;
			say(`event source dropped (${(err as Error).message}); reconnecting`);
		}
		if (Date.now() < until)
			await sleep(Math.min(retry, Math.max(0, until - Date.now())));
	}
	while (running !== null) await running;
}

// --- main -----------------------------------------------------------------

async function main(): Promise<void> {
	const { values } = parseArgs({
		options: {
			jmap: { type: "string" },
			issuer: { type: "string" },
			state: { type: "string", default: "annotate-state.json" },
			follow: { type: "string", default: "0" },
			name: { type: "string", default: "Example Annotator" },
		},
	});
	if (values.jmap === undefined) throw new Failure("--jmap <URL> is required");
	const followFor = Number(values.follow);
	if (!Number.isFinite(followFor) || followFor < 0)
		throw new Failure("--follow takes a number of seconds");
	const statePath = values.state as string;
	const state = await loadState(statePath);

	const d = await discover(values.jmap, values.issuer);
	const clientId = await register(d, state, values.name as string, statePath);
	const tokens = await signIn(d, clientId, state, statePath);
	const c = new Client(d, clientId, state, statePath, tokens);
	c.session = await session(c, d.sessionUrl);
	// `photos:annotate` when the person narrowed the grant to their Pictures:
	// the same calls then reach that folder alone.
	if (
		!c.tokens.scopes.has("drive:annotate") &&
		!c.tokens.scopes.has("photos:annotate")
	)
		throw new Failure("the person did not allow annotating; nothing to do");

	const source = await defineField(c);
	say(`writing as source ${source}`);
	await defineView(c, source);
	const n = await annotateQueue(c);
	say(`queue empty after ${n} file(s)`);
	if (followFor > 0) await follow(c, followFor);
	say("done");
}

main().then(
	() => process.exit(0),
	(err: unknown) => {
		process.stderr.write(
			`${err instanceof Failure ? err.message : (err as Error).stack}\n`,
		);
		process.exit(1);
	},
);