feat: gateway events for messages (#44)

* feat: add new methods to GatewayController

* feat: create message create/delete gateway packets

* feat: getChannel middleware

* refactor: use getChannel middleware

* refactor: send gateway events on message create/delete

* test: getChannel middleware

* test: GatewayController broadcast/sendMessage

* refactor: use getMessages middleware for getMessages route

* refactor: add generic type to sendMessage

Author: amishshah

src/controllers/GatewayController.ts

@@ -53,6 +53,15 @@ export default class GatewayController {
 		}
 	}
 
+	public async broadcast<T extends GatewayPacket>(message: T) {
+		await this.gatewayService.send([...this.authenticatedClients.keys()], message);
+	}
+
+	public async sendMessage<T extends GatewayPacket>(to: string[], message: T) {
+		const recipients = [...this.authenticatedClients.entries()].filter(([,id]) => to.includes(id)).map(entry => entry[0]);
+		await this.gatewayService.send(recipients, message);
+	}
+
 	public async onAuthenticate(ws: WebSocket, packet: IdentifyGatewayPacket) {
 		if (this.authenticatedClients.has(ws)) throw new GatewayError('Already authenticated!');
 		const { token } = packet.data;

src/controllers/MessageController.ts

@@ -4,18 +4,32 @@ import { AuthenticatedResponse } from '../routes/middleware/getUser';
 import MessageService from '../services/MessageService';
 import { AccountType } from '../entities/User';
 import { HttpCode } from '../util/errors';
+import { ChannelResponse } from '../routes/middleware/getChannel';
+import GatewayController from './GatewayController';
+import { MessageCreateGatewayPacket, GatewayPacketType, MessageDeleteGatewayPacket } from '../util/gateway';
+import { EventChannel } from '../entities/Channel';
 
 @injectable()
 export class MessageController {
 	private readonly messageService: MessageService;
+	private readonly gatewayController: GatewayController;
 
-	public constructor(@inject(MessageService) messageService: MessageService) {
+	public constructor(@inject(MessageService) messageService: MessageService, @inject(GatewayController) gatewayController: GatewayController) {
 		this.messageService = messageService;
+		this.gatewayController = gatewayController;
 	}
 
-	public async createMessage(req: Request & { params: { channelID: string } }, res: AuthenticatedResponse, next: NextFunction): Promise<void> {
+	public async createMessage(req: Request, res: ChannelResponse, next: NextFunction): Promise<void> {
 		try {
-			const message = await this.messageService.createMessage({ ...req.body, channelID: req.params.channelID, authorID: res.locals.user.id });
+			const message = await this.messageService.createMessage({ ...req.body, channel: res.locals.channel, author: res.locals.user });
+			if (res.locals.channel instanceof EventChannel) {
+				await this.gatewayController.broadcast<MessageCreateGatewayPacket>({
+					type: GatewayPacketType.MessageCreate,
+					data: {
+						message
+					}
+				});
+			}
 			res.json({ message });
 		} catch (error) {
 			next(error);
@@ -31,10 +45,10 @@ export class MessageController {
 		}
 	}
 
-	public async getMessages(req: Request & { params: { channelID: string; page: number } }, res: AuthenticatedResponse, next: NextFunction): Promise<void> {
+	public async getMessages(req: Request & { params: { channelID: string; page: number } }, res: ChannelResponse, next: NextFunction): Promise<void> {
 		try {
 			const messages = await this.messageService.getMessages({
-				channelID: req.params.channelID,
+				channel: res.locals.channel,
 				page: Number(req.query.page),
 				count: 50
 			});
@@ -44,13 +58,22 @@ export class MessageController {
 		}
 	}
 
-	public async deleteMessage(req: Request & { params: { channelID: string; messageID: string } }, res: AuthenticatedResponse, next: NextFunction): Promise<void> {
+	public async deleteMessage(req: Request & { params: { messageID: string } }, res: ChannelResponse, next: NextFunction): Promise<void> {
 		try {
 			await this.messageService.deleteMessage({
 				id: req.params.messageID,
-				channelID: req.params.channelID,
+				channelID: res.locals.channel.id,
 				authorID: res.locals.user.accountType === AccountType.Admin ? undefined : res.locals.user.id
 			});
+			if (res.locals.channel instanceof EventChannel) {
+				await this.gatewayController.broadcast<MessageDeleteGatewayPacket>({
+					type: GatewayPacketType.MessageDelete,
+					data: {
+						messageID: req.params.messageID,
+						channelID: res.locals.channel.id
+					}
+				});
+			}
 			res.status(HttpCode.NoContent).end();
 		} catch (error) {
 			next(error);

src/routes/MessageRoutes.ts

@@ -2,6 +2,7 @@ import { Router } from 'express';
 import { inject, injectable } from 'tsyringe';
 import { getUser, isVerified } from './middleware';
 import { MessageController } from '../controllers/MessageController';
+import getChannel from './middleware/getChannel';
 
 @injectable()
 export class MessageRoutes {
@@ -12,9 +13,9 @@ export class MessageRoutes {
 	}
 
 	public routes(router: Router): void {
-		router.post('/channels/:channelID/messages', getUser, isVerified, this.messageController.createMessage.bind(this.messageController));
-		router.get('/channels/:channelID/messages', getUser, isVerified, this.messageController.getMessages.bind(this.messageController));
+		router.post('/channels/:channelID/messages', getUser, isVerified, getChannel, this.messageController.createMessage.bind(this.messageController));
+		router.get('/channels/:channelID/messages', getUser, isVerified, getChannel, this.messageController.getMessages.bind(this.messageController));
 		router.get('/channels/:channelID/messages/:messageID', getUser, isVerified, this.messageController.getMessage.bind(this.messageController));
-		router.delete('/channels/:channelID/messages/:messageID', getUser, isVerified, this.messageController.deleteMessage.bind(this.messageController));
+		router.delete('/channels/:channelID/messages/:messageID', getUser, isVerified, getChannel, this.messageController.deleteMessage.bind(this.messageController));
 	}
 }

src/routes/middleware/getChannel.ts

@@ -0,0 +1,25 @@
+import { NextFunction, Request } from 'express';
+import { Channel, EventChannel } from '../../entities/Channel';
+import { APIError, HttpCode } from '../../util/errors';
+import { AuthenticatedResponse } from './getUser';
+import { container } from 'tsyringe';
+import ChannelService from '../../services/ChannelService';
+
+enum GetChannelError {
+	NotAllowed = 'You are not allowed to view this channel',
+	NotFound = 'Channel not found'
+}
+
+export type ChannelResponse = AuthenticatedResponse & { locals: { channel: Channel } };
+
+export default async function getChannel(req: Request, res: AuthenticatedResponse, next: NextFunction) {
+	const channelService = container.resolve(ChannelService);
+	if (!req.params.channelID) return next(new APIError(HttpCode.NotFound, GetChannelError.NotFound));
+	const channel = await channelService.findOne({ id: req.params.channelID });
+	if (!channel) return next(new APIError(HttpCode.NotFound, GetChannelError.NotFound));
+
+	if (channel instanceof EventChannel) {
+		(res as ChannelResponse).locals.channel = channel;
+	}
+	next();
+}

src/routes/middleware/index.ts

@@ -1,3 +1,4 @@
 export { default as getUser } from './getUser';
 export { default as isVerified } from './isVerified';
 export { default as isAdmin } from './isAdmin';
+export { default as getChannel } from './getChannel';

src/services/ChannelService.ts

@@ -0,0 +1,10 @@
+import { singleton } from 'tsyringe';
+import { getRepository, FindOneOptions, FindConditions } from 'typeorm';
+import { Channel } from '../entities/Channel';
+
+@singleton()
+export default class ChannelService {
+	public async findOne(findConditions: FindConditions<Channel>, options?: FindOneOptions) {
+		return getRepository(Channel).findOne(findConditions, options);
+	}
+}

src/services/MessageService.ts

@@ -4,8 +4,9 @@ import { getRepository } from 'typeorm';
 import { APIError, formatValidationErrors, HttpCode } from '../util/errors';
 import { validateOrReject } from 'class-validator';
 import { Channel } from '../entities/Channel';
+import { User } from '../entities/User';
 
-type MessageCreationData = Omit<APIMessage, 'id'>;
+type MessageCreationData = Omit<APIMessage, 'id' | 'channelID' | 'authorID'> & { channel: Channel; author: User };
 
 enum GetMessageError {
 	NotFound = 'Message not found',
@@ -18,18 +19,14 @@ enum DeleteMessageError {
 	NotAuthor = 'You are not the author of this message'
 }
 
-enum GetMessagesError {
-	ChannelNotFound = 'Channel does not exist'
-}
-
 @singleton()
 export default class MessageService {
 	public async createMessage(data: MessageCreationData): Promise<APIMessage> {
 		const message = new Message();
 		message.content = data.content;
 		message.time = new Date(data.time);
-		message.author = { id: data.authorID } as any;
-		message.channel = { id: data.channelID } as any;
+		message.author = data.author;
+		message.channel = data.channel;
 		await validateOrReject(message).catch(e => Promise.reject(formatValidationErrors(e)));
 		return (await getRepository(Message).save(message)).toJSON();
 	}
@@ -41,13 +38,11 @@ export default class MessageService {
 		return message.toJSON();
 	}
 
-	public async getMessages(data: { channelID: string; page?: number; count: number }): Promise<APIMessage[]> {
+	public async getMessages(data: { channel: Channel; page?: number; count: number }): Promise<APIMessage[]> {
 		if (!data.page || isNaN(data.page)) data.page = 0;
-		if (!data.channelID) throw new APIError(HttpCode.NotFound, GetMessagesError.ChannelNotFound);
-		const channel = await getRepository(Channel).findOneOrFail(data.channelID).catch(() => Promise.reject(new APIError(HttpCode.NotFound, GetMessagesError.ChannelNotFound)));
 		const messages = await getRepository(Message)
 			.find({
-				where: { channel },
+				where: { channel: data.channel },
 				order: { time: 'DESC' },
 				skip: data.count * data.page,
 				take: data.count

src/util/gateway/index.ts

@@ -1,8 +1,12 @@
+import { APIMessage } from '../../entities/Message';
+
 export class GatewayError extends Error {}
 
 export enum GatewayPacketType {
 	Identify = 'IDENTIFY',
-	Hello = 'HELLO'
+	Hello = 'HELLO',
+	MessageCreate = 'MESSAGE_CREATE',
+	MessageDelete = 'MESSAGE_DELETE'
 }
 
 export interface GatewayPacket {
@@ -19,3 +23,18 @@ export interface IdentifyGatewayPacket extends GatewayPacket {
 export interface HelloGatewayPacket extends GatewayPacket {
 	type: GatewayPacketType.Hello;
 }
+
+export interface MessageCreateGatewayPacket extends GatewayPacket {
+	type: GatewayPacketType.MessageCreate;
+	data: {
+		message: APIMessage;
+	};
+}
+
+export interface MessageDeleteGatewayPacket extends GatewayPacket {
+	type: GatewayPacketType.MessageDelete;
+	data: {
+		messageID: string;
+		channelID: string;
+	};
+}

tests/fixtures/messages.ts

@@ -6,7 +6,7 @@ import users from './users';
 import events from './events';
 import { v4 as uuidv4 } from 'uuid';
 
-export function createMessage(data: { author?: User; channel?: Channel; content?: string; time?: Date }): [Pick<APIMessage, 'content' | 'authorID' | 'channelID' | 'time'>, Message] {
+export function createMessage(data: { author?: User; channel?: Channel; content?: string; time?: Date }): [Pick<APIMessage, 'content' | 'time'> & { author: User; channel: Channel }, Message] {
 	const concreteData = {
 		author: data.author ?? users.find(user => user.accountStatus === AccountStatus.Verified)!,
 		channel: data.channel ?? events[0].channel,
@@ -25,8 +25,8 @@ export function createMessage(data: { author?: User; channel?: Channel; content?
 	*/
 	return [
 		{
-			authorID: concreteData.author.id,
-			channelID: concreteData.channel.id,
+			author: concreteData.author,
+			channel: concreteData.channel,
 			content: concreteData.content,
 			time: message.time.toISOString()
 		},

tests/integration/controllers/gatewayController.test.ts

@@ -5,18 +5,20 @@ import { UserService } from '../../../src/services/UserService';
 import { container } from 'tsyringe';
 import { mock, instance, reset, spy, when, objectContaining, verify, anything } from 'ts-mockito';
 import * as auth from '../../../src/util/auth';
+import GatewayController from '../../../src/controllers/GatewayController';
 
 let spiedVerifyJWT: typeof auth;
 let wss: MockWebSocketServer;
 let mockedUserService: UserService;
+let gatewayController: GatewayController;
 
 beforeAll(() => {
 	spiedVerifyJWT = spy(auth);
 	mockedUserService = mock(UserService);
 	container.clearInstances();
 	container.register<UserService>(UserService, { useValue: instance(mockedUserService) });
 	wss = new MockWebSocketServer();
-	createGateway(wss);
+	gatewayController = createGateway(wss);
 });
 
 function createWebSocket(): Promise<MockWebSocket> {
@@ -27,18 +29,18 @@ function createWebSocket(): Promise<MockWebSocket> {
 	});
 }
 
-let ws: MockWebSocket;
+let sockets: MockWebSocket[];
 
 beforeEach(async () => {
 	reset(spiedVerifyJWT);
 	spiedVerifyJWT = spy(auth);
 	reset(mockedUserService);
-	ws = await createWebSocket();
+	sockets = await Promise.all([0, 0].map(() => createWebSocket()));
 	await wait();
 });
 
 afterEach(async () => {
-	await ws.close();
+	await Promise.all(sockets.map(ws => ws.close()));
 	await wait();
 });
 
@@ -51,18 +53,62 @@ function wait() {
 describe('GatewayController', () => {
 	describe('general', () => {
 		test('Closes for unknown JSON', async () => {
-			await ws.send(JSON.stringify({ z: 2 }));
-			await expect(ws.nextMessage).rejects.toThrow();
+			await sockets[0].send(JSON.stringify({ z: 2 }));
+			await expect(sockets[0].nextMessage).rejects.toThrow();
 		});
 
 		test('Closes for unknown packet', async () => {
-			await ws.send(JSON.stringify({ type: -1 }));
-			await expect(ws.nextMessage).rejects.toThrow();
+			await sockets[0].send(JSON.stringify({ type: -1 }));
+			await expect(sockets[0].nextMessage).rejects.toThrow();
 		});
 
 		test('Closes for invalid packet', async () => {
-			await ws.send('garbage');
-			await expect(ws.nextMessage).rejects.toThrow();
+			await sockets[0].send('garbage');
+			await expect(sockets[0].nextMessage).rejects.toThrow();
+		});
+
+		test('broadcast', async () => {
+			// Authenticate the first socket
+			gatewayController.authenticatedClients.set(sockets[0].mirror, '');
+
+			const payload = { time: Date.now() };
+			const stringPayload = JSON.stringify(payload);
+
+			await gatewayController.broadcast(payload as any);
+			await wait();
+			await expect(sockets[0].nextMessage).resolves.toEqual(stringPayload);
+			expect(sockets[1].allMessages.length).toEqual(0);
+
+			// Authenticate the second socket
+			gatewayController.authenticatedClients.set(sockets[1].mirror, '');
+			await gatewayController.broadcast(payload as any);
+			await expect(sockets[0].nextMessage).resolves.toEqual(stringPayload);
+			await expect(sockets[1].nextMessage).resolves.toEqual(stringPayload);
+		});
+
+		test('sendTo', async () => {
+			// Authenticate the first socket
+			gatewayController.authenticatedClients.set(sockets[0].mirror, 'banana');
+			gatewayController.authenticatedClients.set(sockets[1].mirror, 'apple');
+
+			const payload = { time: Date.now() };
+			const stringPayload = JSON.stringify(payload);
+
+			await gatewayController.sendMessage(['banana'], payload as any);
+			await expect(sockets[0].nextMessage).resolves.toEqual(stringPayload);
+			expect(sockets[1].allMessages.length).toEqual(0);
+
+			await gatewayController.sendMessage(['apple'], payload as any);
+			await expect(sockets[1].nextMessage).resolves.toEqual(stringPayload);
+			expect(sockets[0].allMessages.length).toEqual(1);
+
+			await gatewayController.sendMessage(['apple', 'banana'], payload as any);
+			await expect(sockets[0].nextMessage).resolves.toEqual(stringPayload);
+			await expect(sockets[1].nextMessage).resolves.toEqual(stringPayload);
+
+			await gatewayController.sendMessage(['apple', 'banana', 'ghost'], payload as any);
+			await expect(sockets[0].nextMessage).resolves.toEqual(stringPayload);
+			await expect(sockets[1].nextMessage).resolves.toEqual(stringPayload);
 		});
 	});
 
@@ -77,8 +123,8 @@ describe('GatewayController', () => {
 
 			when(spiedVerifyJWT.verifyJWT('123')).thenResolve({ id: '456' });
 			when(mockedUserService.findOne(objectContaining({ id: '456' }))).thenResolve({} as any);
-			await ws.send(JSON.stringify(payload));
-			expect(JSON.parse(await ws.nextMessage)).toMatchObject({ type: GatewayPacketType.Hello });
+			await sockets[0].send(JSON.stringify(payload));
+			expect(JSON.parse(await sockets[0].nextMessage)).toMatchObject({ type: GatewayPacketType.Hello });
 			verify(spiedVerifyJWT.verifyJWT('123')).once();
 			verify(mockedUserService.findOne(objectContaining({ id: '456' }))).once();
 		});
@@ -92,8 +138,8 @@ describe('GatewayController', () => {
 			};
 
 			when(spiedVerifyJWT.verifyJWT('123')).thenReject(new Error('Test Error'));
-			await ws.send(JSON.stringify(payload));
-			await expect(ws.nextMessage).rejects.toThrow();
+			await sockets[0].send(JSON.stringify(payload));
+			await expect(sockets[0].nextMessage).rejects.toThrow();
 			verify(spiedVerifyJWT.verifyJWT('123')).once();
 			verify(mockedUserService.findOne(anything())).never();
 		});
@@ -108,8 +154,8 @@ describe('GatewayController', () => {
 
 			when(spiedVerifyJWT.verifyJWT('123')).thenResolve({ id: '456' });
 			when(mockedUserService.findOne(objectContaining({ id: '456' }))).thenReject(new Error('Test Error'));
-			await ws.send(JSON.stringify(payload));
-			await expect(ws.nextMessage).rejects.toThrow();
+			await sockets[0].send(JSON.stringify(payload));
+			await expect(sockets[0].nextMessage).rejects.toThrow();
 			verify(spiedVerifyJWT.verifyJWT('123')).once();
 			verify(mockedUserService.findOne(objectContaining({ id: '456' }))).once();
 		});