refactor(apps): drop Deno specific APIs in deno-runtime

Author: d-gubert

packages/apps/deno-runtime/deno.jsonc

@@ -6,7 +6,6 @@
 		"@rocket.chat/apps/": "../",
 		"@std/cli": "jsr:@std/cli@^1.0.9",
 		"@std/io": "jsr:@std/io@^0.225.3",
-		"@std/streams": "jsr:@std/streams@^1.0.16",
 		"acorn": "npm:acorn@8.10.0",
 		"acorn-walk": "npm:acorn-walk@8.2.0",
 		"astring": "npm:astring@1.8.6",
@@ -16,7 +15,7 @@
 	},
 	"unstable": ["detect-cjs","sloppy-imports"],
 	"tasks": {
-		"test": "deno test --no-check --allow-read --allow-write"
+		"test": "deno test --no-check --allow-read --allow-write --allow-env"
 	},
 	"fmt": {
 		"lineWidth": 160,

packages/apps/deno-runtime/deno.lock

@@ -1,10 +1,9 @@
-import { Buffer } from 'node:buffer';
+import { open } from 'node:fs/promises';
 
 import type { App } from '@rocket.chat/apps-engine/definition/App';
 import { AppsEngineException } from '@rocket.chat/apps-engine/definition/exceptions/AppsEngineException';
 import type { IFileUploadContext } from '@rocket.chat/apps-engine/definition/uploads/IFileUploadContext'
 import type { IUploadDetails } from '@rocket.chat/apps-engine/definition/uploads/IUploadDetails'
-import { toArrayBuffer } from '@std/streams';
 import { Defined, JsonRpcError } from 'jsonrpc-lite';
 
 import { AppObjectRegistry } from '../../AppObjectRegistry';
@@ -42,25 +41,29 @@ export default async function handleUploadEvents(request: RequestContext): Promi
 		assertIsUpload(file);
 		assertString(path);
 
-		using tempFile = await Deno.open(path, { read: true, create: false });
-		let context: IFileUploadContext;
+		const tempFile = await open(path, 'r');
 
-		switch (method) {
-			case 'executePreFileUpload': {
-				const fileContents = await toArrayBuffer(tempFile.readable);
-				context = { file, content: Buffer.from(fileContents) };
-				break;
+		try {
+			let context: IFileUploadContext;
+
+			switch (method) {
+				case 'executePreFileUpload': {
+					context = { file, content: await tempFile.readFile() };
+					break;
+				}
 			}
-		}
 
-		return await handlerFunction.call(
-			wrapAppForRequest(app, request),
-			context,
-			AppAccessorsInstance.getReader(),
-			AppAccessorsInstance.getHttp(),
-			AppAccessorsInstance.getPersistence(),
-			AppAccessorsInstance.getModifier(),
-		);
+			return await handlerFunction.call(
+				wrapAppForRequest(app, request),
+				context,
+				AppAccessorsInstance.getReader(),
+				AppAccessorsInstance.getHttp(),
+				AppAccessorsInstance.getPersistence(),
+				AppAccessorsInstance.getModifier(),
+			);
+		} finally {
+			await tempFile.close();
+		}
 	} catch(e) {
 		if (e?.name === AppsEngineException.name) {
 			return new JsonRpcError(e.message, AppsEngineException.JSONRPC_ERROR_CODE, { name: e.name });

packages/apps/deno-runtime/handlers/app/handleUploadEvents.ts

@@ -1,5 +1,8 @@
 // deno-lint-ignore-file no-explicit-any
 import { Buffer } from 'node:buffer';
+import { mkdtemp, rm, writeFile } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
 
 import type { App } from '@rocket.chat/apps-engine/definition/App';
 import type { IPreFileUpload } from '@rocket.chat/apps-engine/definition/uploads/IPreFileUpload';
@@ -16,13 +19,15 @@ import { AppObjectRegistry } from '../../AppObjectRegistry';
 
 describe('handlers > upload', () => {
 	let app: App & IPreFileUpload;
+	let tempDir: string;
 	let path: string;
 	let file: IUploadDetails;
 
 	beforeEach(async () => {
 		AppObjectRegistry.clear();
 
-		path = await Deno.makeTempFile();
+		tempDir = await mkdtemp(join(tmpdir(), 'rc-apps-upload-'));
+		path = join(tempDir, 'tempfile');
 
 		app = {
 			extendConfiguration: () => {},
@@ -33,7 +38,7 @@ describe('handlers > upload', () => {
 
 		const content = 'Temp file for testing';
 
-		await Deno.writeTextFile(path, content);
+		await writeFile(path, content);
 
 		file = {
 			name: 'TempFile.txt',
@@ -45,7 +50,7 @@ describe('handlers > upload', () => {
 	});
 
 	afterEach(async () => {
-		await Deno.remove(path).catch((e) => e?.code !== 'ENOENT' && console.warn(`Failed to remove temp file at ${path}`, e));
+		await rm(tempDir, { recursive: true, force: true }).catch((e) => console.warn(`Failed to remove temp dir at ${tempDir}`, e));
 	});
 
 	it('correctly handles valid parameters', async () => {
@@ -97,7 +102,7 @@ describe('handlers > upload', () => {
 	});
 
 	it('fails when "path" is not a readable file path', async () => {
-		await Deno.remove(path);
+		await rm(path);
 
 		const result = await handleUploadEvents(createMockRequest({ method: 'app:executePreFileUpload', params: [{ file, path }] }));
 

packages/apps/deno-runtime/handlers/tests/upload-event-handler.test.ts

@@ -1,5 +1,3 @@
-import { writeAll } from '@std/io';
-
 import * as jsonrpc from 'jsonrpc-lite';
 
 import { encoder } from './codec';
@@ -47,7 +45,7 @@ export const Queue = new (class Queue {
 			const message = this.queue.shift();
 
 			if (message) {
-				await Transport.send(message);
+				await transport.send(message);
 			}
 		}
 
@@ -64,34 +62,36 @@ export const Queue = new (class Queue {
 	}
 })();
 
-export const Transport = new (class Transporter {
-	private selectedTransport: Transporter['stdoutTransport'] | Transporter['noopTransport'];
-
-	constructor() {
-		this.selectedTransport = this.stdoutTransport.bind(this);
-	}
-
-	private async stdoutTransport(message: Uint8Array): Promise<void> {
-		await writeAll(Deno.stdout, message);
-	}
-
-	private async noopTransport(_message: Uint8Array): Promise<void> {}
-
-	public selectTransport(transport: 'stdout' | 'noop'): void {
-		switch (transport) {
-			case 'stdout':
-				this.selectedTransport = this.stdoutTransport.bind(this);
-				break;
-			case 'noop':
-				this.selectedTransport = this.noopTransport.bind(this);
-				break;
-		}
-	}
-
-	public send(message: Uint8Array): Promise<void> {
-		return this.selectedTransport(message);
-	}
-})();
+/**
+ * A platform-dependent component responsible for delivering encoded messages to
+ * the host that controls this runtime.
+ *
+ * Each runtime platform is expected to provide its own implementation and
+ * inject it via {@link setTransport}.
+ */
+export type Transport = {
+	send(message: Uint8Array): Promise<void>;
+};
+
+/**
+ * The default transport. It discards every message, and is used until a
+ * platform injects its own transport via {@link setTransport}.
+ */
+export const noopTransport: Transport = {
+	send: () => Promise.resolve(),
+};
+
+let transport: Transport = noopTransport;
+
+/**
+ * Injects the transport implementation to be used when sending messages.
+ *
+ * Platforms must call this during bootstrap to wire up the appropriate
+ * transport. Until then, messages are discarded by the default no-op transport.
+ */
+export function setTransport(newTransport: Transport): void {
+	transport = newTransport;
+}
 
 export function parseMessage(message: string | Record<string, unknown>) {
 	let parsed: jsonrpc.IParsedObject | jsonrpc.IParsedObject[];

packages/apps/deno-runtime/lib/messenger.ts

@@ -1,9 +1,10 @@
-import { writeAll } from '@std/io';
+import process from 'node:process';
+
 import { Queue } from './messenger';
 
 export function collectMetrics() {
 	return {
-		pid: Deno.pid,
+		pid: process.pid,
 		queueSize: Queue.getCurrentSize(),
 	};
 }
@@ -16,5 +17,7 @@ const encoder = new TextEncoder();
 export async function sendMetrics() {
 	const metrics = collectMetrics();
 
-	await writeAll(Deno.stderr, encoder.encode(JSON.stringify(metrics)));
+	await new Promise<void>((resolve, reject) => {
+		process.stderr.write(encoder.encode(JSON.stringify(metrics)), (error) => (error ? reject(error) : resolve()));
+	});
 }

packages/apps/deno-runtime/lib/metricsCollector.ts

@@ -3,6 +3,7 @@ import { afterAll, beforeEach, describe, it } from 'https://deno.land/std@0.203.
 import { spy } from 'https://deno.land/std@0.203.0/testing/mock.ts';
 
 import * as Messenger from '../messenger';
+import { stdoutTransport } from '../transports/stdoutTransport';
 import { AppObjectRegistry } from '../../AppObjectRegistry';
 import { createMockRequest } from '../../handlers/tests/helpers/mod';
 import { RequestContext } from '../requestContext';
@@ -14,14 +15,14 @@ describe('Messenger', () => {
 	beforeEach(() => {
 		AppObjectRegistry.clear();
 		AppObjectRegistry.set('id', 'test');
-		Messenger.Transport.selectTransport('noop');
+		Messenger.setTransport(Messenger.noopTransport);
 
 		context = createMockRequest({ method: 'test', params: [] });
 	});
 
 	afterAll(() => {
 		AppObjectRegistry.clear();
-		Messenger.Transport.selectTransport('stdout');
+		Messenger.setTransport(stdoutTransport);
 	});
 
 	it('should add logs to success responses', async () => {

packages/apps/deno-runtime/lib/tests/messenger.test.ts

@@ -0,0 +1,16 @@
+import { writeAll } from '@std/io';
+
+import type { Transport } from '../messenger';
+
+/**
+ * Transport that writes messages to the process' standard output.
+ *
+ * This is the transport used when the runtime is executed as a subprocess by
+ * the Apps-Engine framework, and it is specific to platforms that expose a
+ * Node-compatible `process.stdout`.
+ */
+export const stdoutTransport: Transport = {
+	send(message: Uint8Array): Promise<void> {
+		return writeAll(Deno.stdout, message);
+	},
+};

packages/apps/deno-runtime/lib/transports/stdoutTransport.ts

@@ -1,16 +1,17 @@
-if (!Deno.args.includes('--subprocess')) {
-	Deno.stderr.writeSync(
-		new TextEncoder().encode(`
+import process from 'node:process';
+
+if (!process.argv.includes('--subprocess')) {
+	process.stderr.write(`
             This is a Deno wrapper for Rocket.Chat Apps. It is not meant to be executed stand-alone;
             It is instead meant to be executed as a subprocess by the Apps-Engine framework.
-       `),
-	);
-	Deno.exit(1001);
+       `);
+	process.exit(1001);
 }
 
 import { JsonRpcError } from 'jsonrpc-lite';
 
 import * as Messenger from './lib/messenger';
+import { stdoutTransport } from './lib/transports/stdoutTransport';
 import { decoder } from './lib/codec';
 import { Logger } from './lib/logger';
 
@@ -101,7 +102,7 @@ function handleResponse(response: Messenger.JsonRpcResponse): void {
 async function main() {
 	Messenger.sendNotification({ method: 'ready' });
 
-	for await (const message of decoder.decodeStream(Deno.stdin.readable)) {
+	for await (const message of decoder.decodeStream(process.stdin)) {
 		try {
 			// Process PING command first as it is not JSON RPC
 			if (message === COMMAND_PING) {
@@ -130,6 +131,9 @@ async function main() {
 	}
 }
 
+// This runtime communicates with the Apps-Engine host through stdout
+Messenger.setTransport(stdoutTransport);
+
 registerErrorListeners();
 
 main();