// Server-side host for atlibs wasm modules, with no emdash and no npm deps (Node >= 25
// built-ins: node:sqlite, node:crypto, node:worker_threads, fetch). The same pieces as the
// browser engine: a PluginContext (javascript/src/plugin-context's surface), a session
// (app password), and the run-module host call() (own PDS, proxy, service, attest).
import { createECDH, createPrivateKey, randomBytes, randomUUID, sign } from "node:crypto";
import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
import { readFile } from "node:fs/promises";
import { availableParallelism } from "node:os";
import { dirname, join, resolve } from "node:path";
import { DatabaseSync } from "node:sqlite";
import { Worker } from "node:worker_threads";

import { createDidKey } from "../didkey/index.js";
import { policyCall } from "../oauth/index.js";
import { compareBy, matches, settingsAccess, wasmAccess } from "../plugin-context/index.js";
import { json, xrpcBody, xrpcUrl } from "../run-module/host-call.js";
import { pdsOf } from "../run-module/index.js";

// The run-module host call: one implementation for Node and pages.
export { hostCall } from "../run-module/host-call.js";

// One sqlite row per (namespace, id). Every read-modify-write runs in BEGIN IMMEDIATE, so
// compare-and-set is atomic across processes sharing the state dir, as IndexedDB's is across tabs.
function collection(db, namespace) {
	const select = db.prepare("SELECT value, revision FROM rows WHERE ns = ? AND id = ?");
	const selectAll = db.prepare("SELECT id, value FROM rows WHERE ns = ?");
	const upsert = db.prepare("INSERT INTO rows (ns, id, value, revision) VALUES (?, ?, ?, ?) ON CONFLICT DO UPDATE SET value = excluded.value, revision = excluded.revision");
	const remove = db.prepare("DELETE FROM rows WHERE ns = ? AND id = ?");
	const find = (id) => {
		const found = select.get(namespace, id);
		return found && { value: JSON.parse(found.value), revision: found.revision };
	};
	const rows = () => selectAll.all(namespace).map(({ id, value }) => ({ id, value: JSON.parse(value) }));
	const write = (id, data) => {
		const revision = randomUUID();
		upsert.run(namespace, id, JSON.stringify(data ?? null), revision);
		return revision;
	};
	const tx = (fn) => {
		db.exec("BEGIN IMMEDIATE");
		try {
			const result = fn();
			db.exec("COMMIT");
			return result;
		} catch (error) {
			db.exec("ROLLBACK");
			throw error;
		}
	};

	const self = {
		get: async (id) => find(id)?.value ?? null,
		getVersioned: async (id) => find(id) ?? null,
		exists: async (id) => !!find(id),
		put: async (id, data) => void write(id, data),
		delete: async (id) => remove.run(namespace, id).changes > 0,
		compareAndSet: async (id, expectedRevision, data) =>
			tx(() => ((find(id)?.revision ?? null) !== expectedRevision ? { applied: false } : { applied: true, revision: write(id, data) })),
		compareAndDelete: async (id, expectedRevision) =>
			tx(() => {
				if (find(id)?.revision !== expectedRevision) return { applied: false };
				remove.run(namespace, id);
				return { applied: true };
			}),
		getMany: async (ids) => new Map(ids.map((id) => [id, find(id)]).filter(([, found]) => found).map(([id, found]) => [id, found.value])),
		putMany: async (items) => tx(() => void items.forEach(({ id, data }) => write(id, data))),
		deleteMany: async (ids) => tx(() => ids.reduce((sum, id) => sum + Number(remove.run(namespace, id).changes), 0)),
		// ponytail: filters and pages in JS like the browser context; push where/orderBy into SQL if tables grow.
		query: async ({ where, orderBy, limit = 50, cursor } = {}) => {
			const all = rows().filter((item) => matches(item.value, where)).sort(compareBy(orderBy));
			const start = cursor ? Number(cursor) : 0;
			const end = start + Math.min(Math.max(limit, 1), 100);
			const hasMore = end < all.length;
			return { items: all.slice(start, end).map(({ id, value }) => ({ id, data: value })), hasMore, ...(hasMore && { cursor: String(end) }) };
		},
		count: async (where) => rows().filter((item) => matches(item.value, where)).length,
		updateIf: async (id, { where, set = {}, delta = {} }) => {
			const changes = JSON.parse(JSON.stringify(set));
			const steps = Object.entries(delta).map(([field, spec]) => {
				if (field in changes) throw new TypeError("Storage field appears in both set and delta");
				const amount = "inc" in spec ? spec.inc : -spec.dec;
				if (!Number.isSafeInteger(amount)) throw new TypeError("Storage delta must be a safe integer");
				return [field, amount];
			});
			return tx(() => {
				const data = find(id)?.value;
				if (typeof data !== "object" || data === null || Array.isArray(data) || !matches(data, where)) return { applied: false };
				const next = { ...data, ...changes };
				for (const [field, amount] of steps) {
					next[field] = data[field] + amount;
					if (!Number.isSafeInteger(data[field]) || !Number.isSafeInteger(next[field])) return { applied: false };
				}
				write(id, next);
				return { applied: true, data: next };
			});
		},
		// SettingsAccess / KVAccess shape
		set: (id, value) => self.put(id, value),
		list: async (prefix = "") => rows().filter((item) => item.id.startsWith(prefix)).map(({ id, value }) => ({ key: id, value })),
	};
	return self;
}

// Locks across every process sharing the state dir (and every run in this one): a lease row
// (<id>:locks / name -> { owner, until }) taken by compare-and-set (BEGIN IMMEDIATE); a holder that
// died loses it once `until` passes. rust/host store.rs (Store::lock, try_lease) takes the same rows, so
// the JS and Rust CLIs exclude each other too.
//   lock(name, fn, ttlMs?)  waits for the lease (polling), runs fn, gives it back
//   lease(name, ttlMs?)     takes it now or answers null; held (renewed every ttlMs / 3) until release()
function locker(rows) {
	const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
	const take = async (name, ttlMs) => {
		const held = await rows.getVersioned(name);
		const now = Date.now();
		if (held && held.value?.until > now) return null;
		return (await rows.compareAndSet(name, held?.revision ?? null, { owner: randomUUID(), until: now + ttlMs })).revision ?? null;
	};
	const lock = async (name, fn, ttlMs = 30000) => {
		let revision;
		while (!(revision = await take(name, ttlMs))) await sleep(25);
		try {
			return await fn();
		} finally {
			await rows.compareAndDelete(name, revision);
		}
	};
	const lease = async (name, ttlMs = 30000) => {
		let revision = await take(name, ttlMs);
		if (!revision) return null;
		const renew = setInterval(async () => {
			try {
				const renewed = await rows.compareAndSet(name, revision, { owner: revision, until: Date.now() + ttlMs });
				if (renewed.applied) revision = renewed.revision;
				else clearInterval(renew); // it ran out and another holder took it
			} catch {
				clearInterval(renew); // the context closed
			}
		}, ttlMs / 3);
		renew.unref?.();
		let released = false;
		return () => {
			if (released) return;
			released = true;
			clearInterval(renew);
			rows.compareAndDelete(name, revision).catch(() => {});
		};
	};
	return { lock, lease };
}

// The AES-GCM key secret settings are sealed with: 32 bytes in <stateDir>/secret.key (0600),
// created on first use. Keep it apart from backups of store.db.
function secretKey(path) {
	let key;
	return () => {
		if (!key) {
			let raw;
			try {
				raw = readFileSync(path);
			} catch (error) {
				if (error.code !== "ENOENT") throw error;
				raw = randomBytes(32);
				try {
					writeFileSync(path, raw, { mode: 0o600, flag: "wx" });
				} catch (error) {
					if (error.code !== "EEXIST") throw error;
					raw = readFileSync(path); // another process made it first
				}
			}
			if (raw.length !== 32) throw new Error(`${path} must hold 32 bytes`);
			key = crypto.subtle.importKey("raw", raw, "AES-GCM", false, ["encrypt", "decrypt"]);
		}
		return key;
	};
}

// wasm-worker.js is a browser module worker (self.onmessage / self.postMessage); this runs it
// unchanged in a worker_thread, behind the slice of the Worker interface wasmAccess uses.
const WORKER_SHIM = `const { parentPort, workerData } = require("node:worker_threads");
globalThis.self = { postMessage: (data) => parentPort.postMessage(data), set onmessage(fn) { parentPort.on("message", (data) => fn({ data })); } };
import(workerData);`;

/** ATLIBS_WASM_CONCURRENCY: wasm runs executing at once in this process (default: its cores). */
export function wasmConcurrency(value = process.env.ATLIBS_WASM_CONCURRENCY) {
	if (value === undefined || value.trim() === "") return availableParallelism();
	if (!/^\d+$/.test(value.trim()) || Number(value) < 1) throw new Error(`ATLIBS_WASM_CONCURRENCY must be a positive integer, not ${JSON.stringify(value)}`);
	return Number(value);
}

// One queue per process, as the Rust host's Wasm semaphore: overlapping requests cannot multiply per-instance
// memory without bound. FIFO; a finished run hands its slot to the next waiter. Checked at import, so a bad
// ATLIBS_WASM_CONCURRENCY stops the process at startup. The browser engine has none: a tab runs one user's work.
const wasmSlot = ((max) => {
	let active = 0;
	const waiting = [];
	return async (fn) => {
		if (active < max) active++;
		else await new Promise((resolve) => waiting.push(resolve));
		try {
			return await fn();
		} finally {
			const next = waiting.shift();
			next ? next() : active--;
		}
	};
})(wasmConcurrency());

function spawnWorker(url) {
	const worker = new Worker(WORKER_SHIM, { eval: true, workerData: url.href });
	return {
		postMessage: (data) => worker.postMessage(data),
		terminate: () => void worker.terminate(),
		set onmessage(fn) {
			worker.on("message", (data) => fn({ data }));
		},
		set onerror(fn) {
			worker.on("error", (error) => fn({ message: error.message }));
		},
	};
}

/**
 * The browser createPluginContext's surface on Node: storage collections, settings (schema
 * type "secret" sealed with <stateDir>/secret.key), kv, http (fetch), wasm (worker_thread, 5 s,
 * sha256 checked, ATLIBS_WASM_CONCURRENCY at once per process), log, site, url. State lives in <stateDir>/store.db. Hosts also get lock(name, fn,
 * ttlMs?) and lease(name, ttlMs?): locks every process sharing the state dir respects (locker).
 * @param {{ id: string, stateDir: string, version?: string, storage?: Record<string, unknown>, settingsSchema?: Record<string, { type: string }>, siteUrl?: string, siteName?: string, fetch?: typeof fetch, wasmTimeoutMs?: number }} options
 */
export function createNodeContext({ id, stateDir, version = "0.0.0", storage = {}, settingsSchema = {}, siteUrl = "http://localhost/", siteName = "", fetch: fetchImpl, wasmTimeoutMs = 5000 }) {
	mkdirSync(stateDir, { recursive: true, mode: 0o700 });
	const db = new DatabaseSync(join(stateDir, "store.db"), { timeout: 5000 });
	db.exec("CREATE TABLE IF NOT EXISTS rows (ns TEXT NOT NULL, id TEXT NOT NULL, value TEXT NOT NULL, revision TEXT NOT NULL, PRIMARY KEY (ns, id))");
	const key = secretKey(join(stateDir, "secret.key"));
	// stderr, so a CLI's stdout stays the result.
	const wasm = wasmAccess(wasmTimeoutMs, spawnWorker);
	const log = Object.fromEntries(
		["debug", "info", "warn", "error"].map((level) => [level, (message, data) => console.error(`[${id}] ${level}: ${message}`, ...(data === undefined ? [] : [data]))]),
	);
	return {
		plugin: { id, version },
		storage: Object.fromEntries(Object.keys(storage).map((name) => [name, collection(db, `${id}:storage:${name}`)])),
		settings: settingsAccess(collection(db, `${id}:settings`), settingsSchema, id, key),
		// settings sealed whatever their name (OAuth sessions per DID: javascript/src/oauth-server): the same rows and
		// additional data as settings, so rust/host Store::setting reads them too
		secrets: settingsAccess(collection(db, `${id}:settings`), new Proxy({}, { get: () => ({ type: "secret" }) }), id, key),
		kv: collection(db, `${id}:kv`),
		http: { fetch: (url, init) => (fetchImpl ?? globalThis.fetch)(url, init) },
		// the timeout starts inside run, once a slot is taken: queued time is not wall time
		wasm: { run: (args) => wasmSlot(() => wasm.run(args)) },
		log,
		site: { name: siteName, url: siteUrl, locale: "en", trailingSlash: "ignore" },
		url: (path) => new URL(path, siteUrl).href,
		...locker(collection(db, `${id}:locks`)),
		close: () => db.close(),
	};
}

/** fetch for file: URLs, which Node's fetch does not take (localModule of a local build). */
export const fileFetch = async (url) => new Response(await readFile(url));

// The context the CLI and Node servers (examples/node-hono) run modules in. managed_records is
// run-module's rkey ledger: the records this host saw a module create, which that module
// (moduleOwner: its id) may later update or delete (cleanup). It persists in the state dir across runs.
export const contextOptions = (stateDir) => ({
	id: "atlibs-server",
	stateDir,
	storage: { managed_records: {} },
	settingsSchema: { attestKeys: { type: "secret" }, session: { type: "secret" }, didkeys: { type: "secret" } },
});

/**
 * The did:key sealed under `id` in the "didkeys" setting, made and saved on first use. The
 * read-modify-write runs under the "didkeys" lock (rust/host Store::didkey takes it too), so runs and
 * processes that first use different modules at once keep every key.
 */
export const didkeyOf = (ctx, id) =>
	ctx.lock(
		"didkeys",
		async () => {
			const keys = (await ctx.settings.get("didkeys")) ?? {};
			const key = await createDidKey(keys[id]);
			if (!keys[id]) await ctx.settings.set("didkeys", { ...keys, [id]: key.jwk });
			return key;
		},
		10000,
	);

// secp256k1 for javascript/src/attest from node:crypto: the slice of @noble/curves 2.x it uses
// (compact r||s, low-S, sha256 prehash).
const N = 0xfffffffffffffffffffffffffffffffebaaedce6af48a03bbfd25e8cd0364141n;
const b64u = (bytes) => Buffer.from(bytes).toString("base64url");
export const secp256k1 = {
	utils: { randomSecretKey: () => new Uint8Array(randomBytes(32)) },
	getPublicKey(secretKey, compressed) {
		const ecdh = createECDH("secp256k1");
		ecdh.setPrivateKey(secretKey);
		return new Uint8Array(ecdh.getPublicKey(null, compressed ? "compressed" : "uncompressed"));
	},
	sign(message, secretKey) {
		const point = secp256k1.getPublicKey(secretKey, false);
		const jwk = { kty: "EC", crv: "secp256k1", x: b64u(point.subarray(1, 33)), y: b64u(point.subarray(33)), d: b64u(secretKey) };
		const rs = new Uint8Array(sign("sha256", message, { key: createPrivateKey({ key: jwk, format: "jwk" }), dsaEncoding: "ieee-p1363" }));
		const s = BigInt(`0x${Buffer.from(rs.subarray(32)).toString("hex")}`);
		if (s > N / 2n) rs.set(Buffer.from((N - s).toString(16).padStart(64, "0"), "hex"), 32);
		return rs;
	},
};

/**
 * App-password session: createSession at `service` (an entryway or PDS), then XRPC to the
 * account's own PDS (#atproto_pds of the session's DID document). createSession is rate limited
 * (30 per 5 min per account), so a saved session of this identifier (`load`) is used instead, and
 * `onSession` is called with every new one to save it (refresh tokens rotate).
 *
 * Runs share one session, in this process and (with `lock`, sessionStore) across processes sharing
 * the state dir. A request that comes back ExpiredToken or InvalidToken renews it once for all of
 * them: one renewal in flight here, under `lock` across processes; it first re-reads the saved
 * session (another run may have refreshed it: use that), else refreshes with the newest refresh
 * token, and logs in again only if the refresh is refused.
 * @param {{ identifier: string, password?: string, service?: string, fetch?: typeof fetch,
 *   load?: () => Promise<object | undefined>, onSession?: (session: object) => unknown, lock?: <T>(fn: () => Promise<T>) => Promise<T> }} options
 * @returns {Promise<{ did: string, handle: string, pds: string, xrpc: (nsid: string, init?: { params?: Record<string, string>, input?: unknown, headers?: Record<string, string> }) => Promise<any> }>}
 */
export async function login({ identifier, password, service = "https://bsky.social", fetch = globalThis.fetch, load, onSession, lock = (fn) => fn() }) {
	const post = async (url, nsid, headers, input) =>
		json(nsid, await fetch(url, { method: "POST", headers: { ...headers, ...(input && { "content-type": "application/json" }) }, body: input && JSON.stringify(input) }));
	const pdsOfSession = (session) => session.didDoc?.service?.find((entry) => entry.id === "#atproto_pds" || entry.id === `${session.did}#atproto_pds`)?.serviceEndpoint ?? service;
	const saved = async () => {
		const found = await load?.();
		return found?.identifier === identifier && found.refreshJwt ? found : undefined;
	};
	let tokens = await saved();
	const save = async (session) => {
		tokens = { identifier, ...session };
		await onSession?.(tokens);
	};
	const create = async () => save(await post(xrpcUrl(service, "com.atproto.server.createSession"), "com.atproto.server.createSession", {}, { identifier, password }));
	const refresh = () =>
		post(xrpcUrl(pdsOfSession(tokens), "com.atproto.server.refreshSession"), "com.atproto.server.refreshSession", { authorization: `Bearer ${tokens.refreshJwt}` }).then(save, create);
	// Several processes starting on an empty state dir log in once.
	if (!tokens) await lock(async () => (tokens = await saved()) ?? create());
	let renewing;
	const renew = (used) =>
		tokens.accessJwt !== used
			? undefined // another run here renewed it already
			: (renewing ??= lock(async () => {
					const latest = await saved();
					if (latest) tokens = latest;
					if (tokens.accessJwt === used) await refresh(); // else another process renewed it
				}).finally(() => (renewing = undefined)));
	const xrpc = async (nsid, { params, input, encoding, headers } = {}, retried = false) => {
		const body = xrpcBody(input, encoding);
		const used = tokens.accessJwt;
		const response = await fetch(xrpcUrl(pdsOfSession(tokens), nsid, params), {
			method: input !== undefined ? "POST" : "GET",
			headers: { ...headers, authorization: `Bearer ${used}`, ...body.headers },
			body: body.body,
		});
		try {
			return await json(nsid, response);
		} catch (error) {
			if (retried || (error.error !== "ExpiredToken" && error.error !== "InvalidToken")) throw error;
			await renew(used);
			return xrpc(nsid, { params, input, encoding, headers }, true);
		}
	};
	return { did: tokens.did, handle: tokens.handle, pds: pdsOfSession(tokens), xrpc };
}

/** login's load, onSession and lock for a createNodeContext: the session sealed in the state dir, shared by its processes. */
export const sessionStore = (ctx) => ({
	load: () => ctx.settings.get("session"),
	onSession: (session) => ctx.settings.set("session", session),
	lock: (fn) => ctx.lock("session", fn),
});

/**
 * Who a run acts as, from explicit config only, never the environment (README "Who a run acts as"):
 *   { appPassword: { handle, password?, service? } }  an app password (the simple / testing case; its session
 *                                                     is saved in the state dir, so password may be omitted then)
 *   { oauth: { did } }                                a user's OAuth session (oauthServer from javascript/src/oauth-server)
 *   undefined                                         no session: signed out
 * @param {{ oauth?: { session: (did: string) => Promise<object> } }} [hosts]
 */
export async function openSession(ctx, config, { oauth } = {}) {
	if (config === undefined || config === null) return undefined;
	if (config.appPassword) {
		const { handle, password, service = "https://bsky.social" } = config.appPassword;
		if (!handle) throw new Error("appPassword needs a handle");
		return login({ identifier: handle, password, service, ...sessionStore(ctx) });
	}
	if (config.oauth) {
		if (!oauth) throw new Error("an oauth session needs an oauthServer (javascript/src/oauth-server)");
		return oauth.session(config.oauth.did);
	}
	throw new Error("a session config is { appPassword: { handle, password } } or { oauth: { did } }");
}

/**
 * A command line's session flags as openSession's config, and the other words:
 *   --app-password <handle> (--password-file <path|-> | --password-env <NAME>) [--pds <url>]
 * The password is never in argv: a file (- is stdin, first line) or the variable NAMED by --password-env.
 * No flags: no config (signed out); the environment is read only for the variable a flag names.
 */
export function sessionFlags(argv, { env = process.env, read = (path) => readFileSync(path === "-" ? 0 : path, "utf8") } = {}) {
	const FLAGS = ["app-password", "password-file", "password-env", "pds"];
	const found = {};
	const rest = [];
	for (let i = 0; i < argv.length; i++) {
		const [flag, inline] = argv[i].startsWith("--") ? argv[i].slice(2).split(/=(.*)/s) : [];
		if (!FLAGS.includes(flag)) {
			rest.push(argv[i]);
			continue;
		}
		const value = inline ?? argv[++i];
		if (value === undefined) throw new Error(`--${flag} needs a value`);
		if (flag in found) throw new Error(`--${flag} given twice`);
		found[flag] = value;
	}
	const file = found["password-file"], name = found["password-env"];
	if (file !== undefined && name !== undefined) throw new Error("give one of --password-file and --password-env");
	const password = file !== undefined ? read(file).split(/\r?\n/)[0].trim() : name !== undefined ? env[name] : undefined;
	if (name !== undefined && password === undefined) throw new Error(`--password-env ${name}: ${name} is not set`);
	if (password === "") throw new Error("the app password is empty");
	if (!found["app-password"]) {
		if (password !== undefined || found.pds) throw new Error("--password-file / --password-env / --pds go with --app-password <handle>");
		return { rest };
	}
	return { config: { appPassword: { handle: found["app-password"], password, ...(found.pds && { service: found.pds }) } }, rest };
}

const CLIENT = "com.fedfork.exec.oauth.clientMetadata";

/**
 * The policy (javascript/src/oauth policyCall's) module `key` runs under on a server: its session, the
 * clientMetadata record its module record names, with that record's scope. A published module's `session`
 * ({ uri, cid }) is fetched from its publisher; a local module.json's (a name) is records/<name>.json in `dir`
 * or a directory above it (rust/cli registry local_lib). No record: no session of its own.
 */
export async function modulePolicy(ctx, key, module, dir = ".") {
	const ref = module.session;
	let session, record;
	if (typeof ref === "string") {
		session = ref;
		for (let at = resolve(dir); !record; at = dirname(at)) {
			const file = join(at, "records", `${ref}.json`);
			if (existsSync(file)) record = JSON.parse(readFileSync(file, "utf8"));
			else if (dirname(at) === at) break;
		}
	} else if (ref?.uri !== undefined) {
		const [, did, collection, rkey] = /^at:\/\/(did:[^/]+)\/([^/]+)\/([^/]+)$/.exec(ref.uri) ?? [];
		if (collection !== CLIENT) throw new Error(`${key}: session is not a ${CLIENT} record`);
		session = rkey;
		const found = await ctx.http.fetch(xrpcUrl(await pdsOf(ctx, did), "com.atproto.repo.getRecord", { repo: did, collection, rkey }));
		if (!found.ok) throw new Error(`${key}: session record ${ref.uri} not found`);
		record = (await found.json()).value;
	}
	const scope = record?.clientMetadata?.scope;
	return { sessions: typeof scope === "string" ? { [session]: { scope } } : {}, modules: { [key]: { session } } };
}

/**
 * A signed-in run's call held to its module's own session: the account the
 * run acts as (an OAuth user or an app password, whatever it was granted) is the one connected session, and
 * policyCall allows only what the module's session record's scope allows; calls needing no scope go through.
 * rust/host oauth::ScopeCall is the same rule and error text.
 * @param {{ sessions: Record<string, { scope?: string, record?: object }>, modules?: Record<string, { session?: string }> }} policy
 *   modulePolicy's, or a stage.json (`--target spa` sessions carry their scope, `--target server` ones their record)
 */
export function sessionCall(policy, module, call) {
	const own = policy.modules?.[module]?.session;
	const scope = policy.sessions?.[own]?.scope ?? policy.sessions?.[own]?.record?.clientMetadata?.scope;
	const has = own && typeof scope === "string";
	return policyCall({ sessions: has ? { [own]: { scope } } : {}, modules: { [module]: { session: own } } }, module, { [has ? own : ""]: call });
}