[FIX] Implement client errors on ddp-streamer

ee/server/broker.ts

@@ -1,11 +1,12 @@
-import { ServiceBroker, Context, ServiceSchema, Serializers } from 'moleculer';
+import { ServiceBroker, Context, ServiceSchema, Serializers, Errors } from 'moleculer';
 import EJSON from 'ejson';
 
 import { asyncLocalStorage, License } from '../../server/sdk';
 import { api } from '../../server/sdk/api';
 import { IBroker, IBrokerNode, IServiceMetrics } from '../../server/sdk/types/IBroker';
 import { ServiceClass } from '../../server/sdk/types/ServiceClass';
 import { EventSignatures } from '../../server/sdk/lib/Events';
+import { isClientSafeError, ClientSafeError, MethodError } from '../../server/sdk/errors';
 import { LocalBroker } from '../../server/sdk/lib/LocalBroker';
 
 const events: { [k: string]: string } = {
@@ -288,6 +289,35 @@ const {
 	SKIP_PROCESS_EVENT_REGISTRATION = 'true',
 } = process.env;
 
+class CustomRegenerator extends Errors.Regenerator {
+	restoreCustomError(plainError: any): Error | undefined {
+		const { message, reason, details, errorType, isClientSafe } = plainError;
+
+		if (errorType === 'Meteor.Error') {
+			return new MethodError(message, reason, details);
+		}
+
+		if (isClientSafe) {
+			return new ClientSafeError(message, reason, details);
+		}
+
+		return undefined;
+	}
+
+	extractPlainError(err: Error | ClientSafeError): Errors.PlainMoleculerError {
+		console.log('extractPlainError ->', err);
+		return {
+			...super.extractPlainError(err),
+			...(isClientSafeError(err) && {
+				isClientSafe: err.isClientSafe,
+				errorType: err.errorType,
+				reason: err.reason,
+				details: err.details,
+			}),
+		};
+	}
+}
+
 // only starts network broker if transporter properly configured
 if (TRANSPORTER.match(/^(?:nats|TCP)/)) {
 	const network = new ServiceBroker({
@@ -376,6 +406,7 @@ if (TRANSPORTER.match(/^(?:nats|TCP)/)) {
 				},
 			},
 		},
+		errorRegenerator: new CustomRegenerator(),
 	});
 
 	new NetworkBroker(network);

ee/server/services/account/Account.ts

@@ -10,6 +10,7 @@ import { getCollection, Collections } from '../mongo';
 import { IUser } from '../../../../definition/IUser';
 import { ServiceClass } from '../../../../server/sdk/types/ServiceClass';
 import { IAccount, ILoginResult } from '../../../../server/sdk/types/IAccount';
+import { ClientSafeError } from '../../../../server/sdk/errors';
 
 const saveSession = async (uid: string, newToken: IHashedStampedToken): Promise<void> => {
 	const Users = await getCollection<IUser>(Collections.User);
@@ -56,12 +57,21 @@ const loginViaResume = async (resume: string): Promise<false | ILoginResult> =>
 	}
 
 	const { when } = user.services?.resume?.loginTokens?.find((token) => token.hashedToken === hashedToken) || {};
+	if (!when) {
+		throw new ClientSafeError(403, 'Your session has expired. Please log in again.');
+	}
+
+	const tokenExpires = _tokenExpiration(when);
+
+	if (new Date() >= tokenExpires) {
+		throw new ClientSafeError(403, 'Your session has expired. Please log in again.');
+	}
 
 	return {
 		uid: user._id,
 		token: resume,
-		hashedToken,
-		tokenExpires: when ? _tokenExpiration(when) : undefined,
+		hashedToken, // TODO should we return the hashed token? Meteor does not.
+		tokenExpires,
 		type: 'resume',
 	};
 };

ee/server/services/ddp-streamer/Client.ts

@@ -209,6 +209,3 @@ export class Client extends EventEmitter {
 		return this.ws.send(this.encodePayload(payload));
 	}
 }
-
-// TODO implement meteor errors
-// a["{\"msg\":\"result\",\"id\":\"12\",\"error\":{\"isClientSafe\":true,\"error\":403,\"reason\":\"User has no password set\",\"message\":\"User has no password set [403]\",\"errorType\":\"Meteor.Error\"}}"]

ee/server/services/ddp-streamer/Server.ts

@@ -8,6 +8,10 @@ import { Publication } from './Publication';
 import { Client } from './Client';
 import { IPacket } from './types/IPacket';
 import { MeteorService } from '../../../../server/sdk';
+import { ClientSafeError, MethodError } from '../../../../server/sdk/errors';
+import { Logger } from '../../../../server/lib/logger/Logger';
+
+const logger = new Logger('DDP-Streamer');
 
 type SubscriptionFn = (this: Publication, eventName: string, options: object) => void;
 type MethodFn = (this: Client, ...args: any[]) => any;
@@ -34,24 +38,38 @@ export class Server extends EventEmitter {
 
 	async call(client: Client, packet: IPacket): Promise<void> {
 		try {
+			// if method was not defined on DDP Streamer we fall back to Meteor
 			if (!this._methods.has(packet.method)) {
 				const result = await MeteorService.callMethodWithToken(client.userId, client.userToken, packet.method, packet.params);
 				if (result?.result) {
 					return this.result(client, packet, result.result);
 				}
 
-				throw new Error(`Method '${packet.method}' doesn't exist`);
+				throw new MethodError(404, `Method '${packet.method}' not found`);
 			}
-			const fn = this._methods.get(packet.method);
 
+			const fn = this._methods.get(packet.method);
 			if (!fn) {
-				throw Error('method not found');
+				throw new MethodError(404, `Method '${packet.method}' not found`);
 			}
 
 			const result = await fn.apply(client, packet.params);
 			return this.result(client, packet, result);
-		} catch (error) {
-			return this.result(client, packet, null, error.toString());
+		} catch (error: any) {
+			if (error instanceof ClientSafeError) {
+				return this.result(client, packet, null, error.toJSON());
+			}
+
+			// default errors are logged to the console and redacted from the client
+			logger.error({ msg: 'Method call error', error });
+
+			return this.result(client, packet, null, {
+				isClientSafe: true,
+				error: 500,
+				reason: 'Internal server error',
+				message: 'Internal server error [500]',
+				errorType: 'Meteor.Error', // TODO should we use Meteor.Error?
+			});
 		}
 	}
 
@@ -67,18 +85,34 @@ export class Server extends EventEmitter {
 	async subscribe(client: Client, packet: IPacket): Promise<void> {
 		try {
 			if (!this._subscriptions.has(packet.name)) {
-				throw new Error(`Subscription '${packet.name}' doesn't exist`);
+				// TODO should not be MethorError
+				throw new MethodError(404, `Subscription '${packet.name}' not found`);
 			}
 			const fn = this._subscriptions.get(packet.name);
 			if (!fn) {
-				throw new Error('subscription not found');
+				throw new MethodError(404, `Subscription '${packet.name}' not found`);
 			}
 
 			const publication = new Publication(client, packet, this);
 			const [eventName, options] = packet.params;
 			await fn.call(publication, eventName, options);
-		} catch (error) {
+		} catch (error: any) {
 			this.nosub(client, packet, error.toString());
+
+			if (error instanceof ClientSafeError) {
+				return this.nosub(client, packet, error.toJSON());
+			}
+
+			// default errors are logged to the console and redacted from the client
+			logger.error({ msg: 'Subscribe error', error });
+
+			return this.nosub(client, packet, {
+				isClientSafe: true,
+				error: 500,
+				reason: 'Internal server error',
+				message: 'Internal server error [500]',
+				errorType: 'Meteor.Error', // TODO should we use Meteor.Error?
+			});
 		}
 	}
 
@@ -93,7 +127,7 @@ export class Server extends EventEmitter {
 		return this.publish(`stream-${stream}`, fn);
 	}
 
-	result(client: Client, { id }: IPacket, result?: any, error?: string): void {
+	result(client: Client, { id }: IPacket, result?: any, error?: string | Record<string, any>): void {
 		client.send(
 			this.serialize({
 				[DDP_EVENTS.MSG]: DDP_EVENTS.RESULT,
@@ -110,7 +144,7 @@ export class Server extends EventEmitter {
 		);
 	}
 
-	nosub(client: Client, { id }: IPacket, error?: string): void {
+	nosub(client: Client, { id }: IPacket, error?: string | Record<string, any>): void {
 		return client.send(
 			this.serialize({
 				[DDP_EVENTS.MSG]: DDP_EVENTS.NO_SUBSCRIBE,

ee/server/services/ddp-streamer/configureServer.ts

@@ -6,6 +6,7 @@ import { UserStatus } from '../../../../definition/UserStatus';
 import { Server } from './Server';
 import { AutoUpdateRecord } from '../../../../server/sdk/types/IMeteor';
 import { api } from '../../../../server/sdk/api';
+import { MethodError } from '../../../../server/sdk/errors';
 
 export const server = new Server();
 
@@ -70,7 +71,7 @@ server.methods({
 	async 'login'({ resume, user, password }: { resume: string; user: { username: string }; password: string }) {
 		const result = await Account.login({ resume, user, password });
 		if (!result) {
-			throw new Error('login error');
+			throw new MethodError(403, "You've been logged out by the server. Please log in again");
 		}
 
 		this.userId = result.uid;

ee/server/services/package.json

@@ -36,9 +36,9 @@
 		"fibers": "^5.0.0",
 		"jaeger-client": "^3.18.1",
 		"mem": "^8.1.1",
-		"moleculer": "^0.14.14",
+		"moleculer": "^0.14.19",
 		"mongodb": "^3.6.10",
-		"nats": "^1.4.12",
+		"nats": "^2.4.0",
 		"pino": "^7.6.4",
 		"sodium-native": "^3.2.1",
 		"sodium-plus": "^0.9.0",