chore: remove rocketchat:streamer meteor package (#38494)

Author: cardoso

.worktrees/replies-refactor

@@ -0,0 +1 @@
+Subproject commit e5b749fabc58569dd3d3d059ca1745d9606b661d

apps/meteor/.meteor/packages

@@ -5,7 +5,6 @@
 
 rocketchat:mongo-config
 rocketchat:livechat
-rocketchat:streamer
 rocketchat:version
 
 accounts-base@3.1.2

apps/meteor/.meteor/versions

@@ -70,7 +70,6 @@ reload@1.3.2
 retry@1.1.1
 rocketchat:livechat@0.0.1
 rocketchat:mongo-config@0.0.1
-rocketchat:streamer@1.1.0
 rocketchat:version@1.0.0
 routepolicy@1.1.2
 service-configuration@1.3.5

apps/meteor/app/importer/server/classes/ImporterWebsocket.ts

@@ -1,6 +1,6 @@
 import type { IImportProgress } from '@rocket.chat/core-typings';
-import type { IStreamer } from 'meteor/rocketchat:streamer';
 
+import type { IStreamer } from '../../../../server/modules/streamer/types';
 import notifications from '../../../notifications/server/lib/Notifications';
 
 class ImporterWebsocketDef {

apps/meteor/app/notifications/client/lib/Presence.ts

@@ -2,16 +2,17 @@ import { UserStatus } from '@rocket.chat/core-typings';
 import { Meteor } from 'meteor/meteor';
 
 import { Presence } from '../../../../client/lib/presence';
+import { streamerCentral } from '../../../../client/lib/streamer';
 
 // TODO implement API on Streamer to be able to listen to all streamed data
 // this is a hacky way to listen to all streamed data from user-presence Streamer
 
-new Meteor.Streamer('user-presence');
+streamerCentral.getStreamer('user-presence', { ddpConnection: Meteor.connection });
 
 type args = [username: string, statusChanged?: UserStatus, statusText?: string];
 
 export const STATUS_MAP = [UserStatus.OFFLINE, UserStatus.ONLINE, UserStatus.AWAY, UserStatus.BUSY, UserStatus.DISABLED];
 
-Meteor.StreamerCentral.on('stream-user-presence', (uid: string, [username, statusChanged, statusText]: args) => {
+streamerCentral.on('stream-user-presence', (uid: string, [username, statusChanged, statusText]: args) => {
 	Presence.notify({ _id: uid, username, status: STATUS_MAP[statusChanged as any], statusText });
 });

apps/meteor/app/notifications/server/lib/Presence.ts

@@ -1,7 +1,8 @@
 import type { IUser } from '@rocket.chat/core-typings';
 import type { StreamerEvents } from '@rocket.chat/ddp-client';
 import { Emitter } from '@rocket.chat/emitter';
-import type { IPublication, IStreamerConstructor, Connection, IStreamer } from 'meteor/rocketchat:streamer';
+
+import type { IPublication, IStreamerConstructor, Connection, IStreamer } from '../../../../server/modules/streamer/types';
 
 type UserPresenceStreamProps = {
 	added: IUser['_id'][];

apps/meteor/client/lib/streamer/ddp.ts

@@ -0,0 +1,172 @@
+import * as z from 'zod/mini';
+
+/**
+ * DDP Message Specification
+ * https://github.com/meteor/meteor/blob/devel/packages/ddp/DDP.md
+ */
+
+// Establishing connection
+const ConnectSchema = z.strictObject({
+	msg: z.literal('connect'),
+	session: z.optional(z.string()),
+	version: z.string(),
+	support: z.array(z.string()),
+});
+
+const ConnectedSchema = z.strictObject({
+	msg: z.literal('connected'),
+	session: z.string(),
+});
+
+const FailedSchema = z.strictObject({
+	msg: z.literal('failed'),
+	version: z.string(),
+});
+
+// Heartbeats
+const PingSchema = z.strictObject({
+	msg: z.literal('ping'),
+	id: z.optional(z.string()),
+});
+
+const PongSchema = z.strictObject({
+	msg: z.literal('pong'),
+	id: z.optional(z.string()),
+});
+
+// Managing data
+const AddedSchema = z.strictObject({
+	msg: z.literal('added'),
+	collection: z.string(),
+	id: z.json(),
+	fields: z.optional(z.record(z.string(), z.json())),
+	// FIXME: Apparently Meteor can send `cleared` in `added`...
+	cleared: z.optional(z.array(z.string())),
+});
+
+const AddedBeforeSchema = z.strictObject({
+	msg: z.literal('addedBefore'),
+	collection: z.string(),
+	id: z.json(),
+	fields: z.optional(z.record(z.string(), z.json())),
+	before: z.optional(z.string()),
+});
+
+const ChangedSchema = z.strictObject({
+	msg: z.literal('changed'),
+	collection: z.string(),
+	id: z.json(),
+	fields: z.optional(z.record(z.string(), z.json())),
+	cleared: z.optional(z.array(z.string())),
+});
+
+const MovedBeforeSchema = z.strictObject({
+	msg: z.literal('movedBefore'),
+	collection: z.string(),
+	id: z.json(),
+	before: z.optional(z.string()),
+});
+
+const RemovedSchema = z.strictObject({
+	msg: z.literal('removed'),
+	collection: z.string(),
+	id: z.json(),
+});
+// Managing subscriptions
+const NosubSchema = z.strictObject({
+	msg: z.literal('nosub'),
+	id: z.string(),
+	error: z.optional(z.json()),
+});
+const ReadySchema = z.strictObject({
+	msg: z.literal('ready'),
+	subs: z.array(z.string()),
+});
+const SubSchema = z.strictObject({
+	msg: z.literal('sub'),
+	id: z.string(),
+	name: z.string(),
+	params: z.optional(z.array(z.json())),
+});
+
+const UnsubSchema = z.strictObject({
+	msg: z.literal('unsub'),
+	id: z.string(),
+});
+
+// Remote procedure calls
+const MethodSchema = z.strictObject({
+	msg: z.literal('method'),
+	id: z.string(),
+	method: z.string(),
+	params: z.optional(z.array(z.json())),
+	randomSeed: z.optional(z.string()),
+});
+
+const ResultSchema = z.strictObject({
+	msg: z.literal('result'),
+	id: z.string(),
+	error: z.optional(z.json()),
+	result: z.optional(z.json()),
+});
+
+const UpdatedSchema = z.strictObject({
+	msg: z.literal('updated'),
+	methods: z.array(z.string()),
+});
+
+const ServerIdSchema = z.strictObject({
+	msg: z.literal('server_id'),
+	server_id: z.string(),
+});
+
+/**
+ * Main DDP Message Schema
+ */
+const DDPMessageSchema = z.discriminatedUnion('msg', [
+	// Connection
+	ConnectSchema,
+	ConnectedSchema,
+	FailedSchema,
+
+	// Heartbeats
+	PingSchema,
+	PongSchema,
+
+	// Data
+	AddedSchema,
+	AddedBeforeSchema,
+	ChangedSchema,
+	MovedBeforeSchema,
+	RemovedSchema,
+
+	// Subscriptions
+	NosubSchema,
+	ReadySchema,
+	SubSchema,
+	UnsubSchema,
+
+	// RPC
+	MethodSchema,
+	ResultSchema,
+	UpdatedSchema,
+
+	// Server ID
+	ServerIdSchema,
+]);
+
+/**
+ * Inferred TypeScript Type
+ * This is "erasable" as it compiles away completely.
+ */
+export type DDPMessage = z.infer<typeof DDPMessageSchema>;
+
+/**
+ * Parses a DDP message from JSON string
+ * @param json The JSON string to parse
+ * @returns The parsed DDP message
+ * @throws If the JSON is invalid or does not conform to the DDP message schema
+ */
+export const parseDDPMessage = (json: string): DDPMessage => {
+	return DDPMessageSchema.parse(JSON.parse(json));
+};

apps/meteor/client/lib/streamer/emitter.ts

@@ -0,0 +1,52 @@
+export type StreamArgs = unknown[];
+export type EventHandler<TArgs extends StreamArgs = any> = (...args: TArgs) => void;
+
+type HandlerMap = {
+	[event: string]: EventHandler[] | undefined;
+};
+
+export class EV {
+	private handlers: HandlerMap = {};
+
+	emit(event: string, ...args: StreamArgs): void {
+		this.handlers[event]?.forEach((handler) => handler.apply(this, args));
+	}
+
+	emitWithScope(event: string, scope: unknown, ...args: StreamArgs): void {
+		this.handlers[event]?.forEach((handler) => handler.apply(scope, args));
+	}
+
+	listenerCount(event: string): number {
+		return this.handlers[event]?.length ?? 0;
+	}
+
+	on<TArgs extends StreamArgs>(event: string, callback: EventHandler<TArgs>): void {
+		if (!this.handlers[event]) {
+			this.handlers[event] = [];
+		}
+		this.handlers[event].push(callback);
+	}
+
+	once(event: string, callback: EventHandler): void {
+		const onetimeCallback: EventHandler = (...args) => {
+			this.removeListener(event, onetimeCallback);
+			callback.apply(this, args);
+		};
+		this.on(event, onetimeCallback);
+	}
+
+	removeListener(event: string, callback: EventHandler): void {
+		const handlers = this.handlers[event];
+		if (!handlers) {
+			return;
+		}
+		const index = handlers.indexOf(callback);
+		if (index > -1) {
+			handlers.splice(index, 1);
+		}
+	}
+
+	removeAllListeners(event: string): void {
+		this.handlers[event] = undefined;
+	}
+}

apps/meteor/client/lib/streamer/index.ts

@@ -0,0 +1,3 @@
+import { StreamerCentral } from './streamer';
+
+export const streamerCentral = new StreamerCentral();