netfelix-audio-fix/server/services/ssh.ts

import { Client } from 'ssh2';
import type { Node } from '../types';

export interface ExecResult {
	exitCode: number;
	output: string;
}

/** Test SSH connectivity to a node. Returns ok + optional error message. */
export function testConnection(node: Node): Promise<{ ok: boolean; error?: string }> {
	return new Promise((resolve) => {
		const conn = new Client();
		const timeout = setTimeout(() => {
			conn.destroy();
			resolve({ ok: false, error: 'Connection timed out' });
		}, 10_000);

		conn.on('ready', () => {
			clearTimeout(timeout);
			conn.exec('echo ok', (err, stream) => {
				if (err) {
					conn.end();
					resolve({ ok: false, error: err.message });
					return;
				}
				stream.on('close', () => {
					conn.end();
					resolve({ ok: true });
				});
			});
		});

		conn.on('error', (err) => {
			clearTimeout(timeout);
			resolve({ ok: false, error: err.message });
		});

		conn.connect(buildConnectConfig(node));
	});
}

/**
 * Execute a command on a remote node and stream output lines.
 * Yields lines as they arrive. Throws if connection fails.
 */
export async function* execStream(
	node: Node,
	command: string
): AsyncGenerator<string> {
	// Collect lines via a promise-based queue
	const queue: string[] = [];
	const resolvers: Array<(value: IteratorResult<string>) => void> = [];
	let done = false;
	let errorVal: Error | null = null;

	const push = (line: string) => {
		if (resolvers.length > 0) {
			resolvers.shift()!({ value: line, done: false });
		} else {
			queue.push(line);
		}
	};

	const finish = (err?: Error) => {
		done = true;
		errorVal = err ?? null;
		while (resolvers.length > 0) {
			resolvers.shift()!({ value: undefined as unknown as string, done: true });
		}
	};

	const conn = new Client();

	conn.on('ready', () => {
		conn.exec(command, { pty: false }, (err, stream) => {
			if (err) {
				conn.end();
				finish(err);
				return;
			}

			stream.stdout.on('data', (data: Buffer) => {
				const lines = data.toString('utf8').split('\n');
				for (const line of lines) {
					if (line) push(line);
				}
			});

			stream.stderr.on('data', (data: Buffer) => {
				const lines = data.toString('utf8').split('\n');
				for (const line of lines) {
					if (line) push(`[stderr] ${line}`);
				}
			});

			stream.on('close', (code: number) => {
				if (code !== 0) {
					push(`[exit code ${code}]`);
				}
				conn.end();
				finish();
			});
		});
	});

	conn.on('error', (err) => finish(err));
	conn.connect(buildConnectConfig(node));

	// Yield from the queue
	while (true) {
		if (queue.length > 0) {
			yield queue.shift()!;
		} else if (done) {
			if (errorVal) throw errorVal;
			return;
		} else {
			await new Promise<IteratorResult<string>>((resolve) => {
				resolvers.push(resolve);
			});
		}
	}
}

/**
 * Execute a command on a remote node and return full output + exit code.
 * For use when streaming isn't needed.
 */
export function execOnce(node: Node, command: string): Promise<ExecResult> {
	return new Promise((resolve, reject) => {
		const conn = new Client();
		let output = '';

		conn.on('ready', () => {
			conn.exec(command, (err, stream) => {
				if (err) {
					conn.end();
					reject(err);
					return;
				}

				stream.stdout.on('data', (d: Buffer) => { output += d.toString(); });
				stream.stderr.on('data', (d: Buffer) => { output += d.toString(); });
				stream.on('close', (code: number) => {
					conn.end();
					resolve({ exitCode: code ?? 0, output });
				});
			});
		});

		conn.on('error', reject);
		conn.connect(buildConnectConfig(node));
	});
}

function buildConnectConfig(node: Node): Parameters<Client['connect']>[0] {
	return {
		host: node.host,
		port: node.port,
		username: node.username,
		privateKey: node.private_key,
		readyTimeout: 10_000,
	};
}