diff --git a/.npmrc b/.npmrc new file mode 100644 index 0000000000..07f82b7293 --- /dev/null +++ b/.npmrc @@ -0,0 +1,2 @@ +@hcengineering:registry=https://npm.pkg.github.com/ +//npm.pkg.github.com/:_authToken=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 \ No newline at end of file diff --git a/.vscode/launch.json b/.vscode/launch.json new file mode 100644 index 0000000000..65434e460a --- /dev/null +++ b/.vscode/launch.json @@ -0,0 +1,24 @@ +{ + "version": "0.2.0", + "configurations": [ + { + "type": "node", + "request": "launch", + "name": "Debug Server", + "env": { + "DB_URL": "postgresql://root@127.0.0.1:26257/my_database?sslmode=disable", + "ACCOUNTS_URL": "http://localhost:3000", + "SECRET": "secret" + }, + "runtimeExecutable": "bun", + "runtimeArgs": ["run"], + "args": ["src/index.ts"], + "cwd": "${workspaceFolder}/packages/server", + "protocol": "inspector", + "runtimeVersion": "20", + "showAsyncStacks": true, + "outputCapture": "std", + "sourceMaps": true + } + ] +} diff --git a/bun.lockb b/bun.lockb index 76fa802047..bcfd9769cc 100755 Binary files a/bun.lockb and b/bun.lockb differ diff --git a/eslint.config.js b/eslint.config.js index 0a7a95e275..ab02536846 100644 --- a/eslint.config.js +++ b/eslint.config.js @@ -5,10 +5,10 @@ import configPrettier from "eslint-config-prettier"; /** @type {import('eslint').Linter.Config[]} */ export default [ - {files: ["**/*.ts"]}, pluginJs.configs.recommended, prettierRecommended, configPrettier, ...tseslint.configs.recommended, + {files: ["**/*.ts"], rules: {"@typescript-eslint/no-explicit-any": "off", '@typescript-eslint/no-empty-object-type': 'off'}}, ]; diff --git a/packages/sdk-types/src/event.ts b/packages/sdk-types/src/event.ts new file mode 100644 index 0000000000..00f49b6714 --- /dev/null +++ b/packages/sdk-types/src/event.ts @@ -0,0 +1,227 @@ +import type { + Attachment, + CardID, + ContextID, + Message, + MessageID, + NotificationContext, + NotificationContextUpdate, + Patch, + Reaction, + RichText, + SocialID, + Notification +} from '@communication/types' + +export enum EventType { + CreateMessage = 'createMessage', + RemoveMessage = 'removeMessage', + CreatePatch = 'createPatch', + CreateReaction = 'createReaction', + RemoveReaction = 'removeReaction', + CreateAttachment = 'createAttachment', + RemoveAttachment = 'removeAttachment', + CreateNotification = 'createNotification', + RemoveNotification = 'removeNotification', + CreateNotificationContext = 'createNotificationContext', + RemoveNotificationContext = 'removeNotificationContext', + UpdateNotificationContext = 'updateNotificationContext', + + MessageCreated = 'messageCreated', + MessageRemoved = 'messageRemoved', + PatchCreated = 'patchCreated', + ReactionCreated = 'reactionCreated', + ReactionRemoved = 'reactionRemoved', + AttachmentCreated = 'attachmentCreated', + AttachmentRemoved = 'attachmentRemoved', + NotificationCreated = 'notificationCreated', + NotificationRemoved = 'notificationRemoved', + NotificationContextCreated = 'notificationContextCreated', + NotificationContextRemoved = 'notificationContextRemoved', + NotificationContextUpdated = 'notificationContextUpdated' +} + +export type Event = + | CreateMessageEvent + | RemoveMessageEvent + | CreatePatchEvent + | CreateReactionEvent + | RemoveReactionEvent + | CreateAttachmentEvent + | RemoveAttachmentEvent + | CreateNotificationEvent + | RemoveNotificationEvent + | CreateNotificationContextEvent + | RemoveNotificationContextEvent + | UpdateNotificationContextEvent + +export interface CreateMessageEvent { + type: EventType.CreateMessage + card: CardID + content: RichText + creator: SocialID +} + +export interface RemoveMessageEvent { + type: EventType.RemoveMessage + message: MessageID +} + +export interface CreatePatchEvent { + type: EventType.CreatePatch + message: MessageID + content: RichText + creator: SocialID +} + +export interface CreateReactionEvent { + type: EventType.CreateReaction + message: MessageID + reaction: string + creator: SocialID +} + +export interface RemoveReactionEvent { + type: EventType.RemoveReaction + message: MessageID + reaction: string + creator: SocialID +} + +export interface CreateAttachmentEvent { + type: EventType.CreateAttachment + message: MessageID + card: CardID + creator: SocialID +} + +export interface RemoveAttachmentEvent { + type: EventType.RemoveAttachment + message: MessageID + card: CardID +} + +export interface CreateNotificationEvent { + type: EventType.CreateNotification + message: MessageID + context: ContextID +} + +export interface RemoveNotificationEvent { + type: EventType.RemoveNotification + message: MessageID + context: ContextID +} + +export interface CreateNotificationContextEvent { + type: EventType.CreateNotificationContext + card: CardID + lastView?: Date + lastUpdate?: Date +} + +export interface RemoveNotificationContextEvent { + type: EventType.RemoveNotificationContext + context: ContextID +} + +export interface UpdateNotificationContextEvent { + type: EventType.UpdateNotificationContext + context: ContextID + update: NotificationContextUpdate +} + +export type EventResult = CreateMessageResult | CreateNotificationContextResult | {} + +export interface CreateMessageResult { + id: MessageID +} + +export interface CreateNotificationContextResult { + id: ContextID +} + +//TODO: THINK ABOUT BETTER NAMES +export type BroadcastEvent = + | MessageCreatedEvent + | MessageRemovedEvent + | PatchCreatedEvent + | ReactionCreatedEvent + | ReactionRemovedEvent + | AttachmentCreatedEvent + | AttachmentRemovedEvent + | NotificationCreatedEvent + | NotificationRemovedEvent + | NotificationContextCreatedEvent + | NotificationContextRemovedEvent + | NotificationContextUpdatedEvent + +export interface MessageCreatedEvent { + type: EventType.MessageCreated + card: CardID + message: Message +} + +export interface MessageRemovedEvent { + type: EventType.MessageRemoved + message: MessageID +} + +export interface PatchCreatedEvent { + type: EventType.PatchCreated + patch: Patch +} + +export interface ReactionCreatedEvent { + type: EventType.ReactionCreated + reaction: Reaction +} + +export interface ReactionRemovedEvent { + type: EventType.ReactionRemoved + message: MessageID + reaction: string + creator: SocialID +} + +export interface AttachmentCreatedEvent { + type: EventType.AttachmentCreated + attachment: Attachment +} + +export interface AttachmentRemovedEvent { + type: EventType.AttachmentRemoved + message: MessageID + card: CardID +} + +export interface NotificationCreatedEvent { + type: EventType.NotificationCreated + personWorkspace: string + notification: Notification +} + +export interface NotificationRemovedEvent { + type: EventType.NotificationRemoved + personWorkspace: string + message: MessageID + context: ContextID +} + +export interface NotificationContextCreatedEvent { + type: EventType.NotificationContextCreated + context: NotificationContext +} + +export interface NotificationContextRemovedEvent { + type: EventType.NotificationContextRemoved + personWorkspace: string + context: ContextID +} + +export interface NotificationContextUpdatedEvent { + type: EventType.NotificationContextUpdated + personWorkspace: string + context: ContextID + update: NotificationContextUpdate +} diff --git a/packages/sdk-types/src/index.ts b/packages/sdk-types/src/index.ts index 1beb455f5e..084f5c7e90 100644 --- a/packages/sdk-types/src/index.ts +++ b/packages/sdk-types/src/index.ts @@ -1 +1,3 @@ export * from './db' +export * from './event' +export * from './ws' diff --git a/packages/sdk-types/src/ws.ts b/packages/sdk-types/src/ws.ts new file mode 100644 index 0000000000..ed91b811b7 --- /dev/null +++ b/packages/sdk-types/src/ws.ts @@ -0,0 +1,17 @@ +export type RequestId = string + +export interface Response { + id?: RequestId + result?: any + error?: string //TODO: Use platform error +} + +export interface Request { + id?: RequestId + method: string + params: any[] +} + +export interface HelloRequest extends Request { + binary?: boolean +} diff --git a/packages/server/package.json b/packages/server/package.json new file mode 100644 index 0000000000..43c5e0cd85 --- /dev/null +++ b/packages/server/package.json @@ -0,0 +1,27 @@ +{ + "name": "@communication/server", + "version": "0.1.0", + "main": "src/index.ts", + "module": "src/index.ts", + "type": "module", + "devDependencies": { + "@types/bun": "^1.1.14", + "@types/express": "^5.0.0", + "@types/cors": "^2.8.17", + "@types/ws": "^8.5.13" + }, + "dependencies": { + "@hcengineering/server-token": "^0.6.377", + "@communication/cockroach": "workspace:*", + "@communication/sdk-types": "workspace:*", + "@communication/types": "workspace:*", + "cors": "^2.8.5", + "dotenv": "^16.4.7", + "express": "^4.21.2", + "msgpackr": "^1.11.2", + "ws": "^8.18.0" + }, + "peerDependencies": { + "typescript": "^5.6.3" + } +} diff --git a/packages/server/src/config.ts b/packages/server/src/config.ts new file mode 100644 index 0000000000..2f835bc72a --- /dev/null +++ b/packages/server/src/config.ts @@ -0,0 +1,29 @@ +interface Config { + Port: number + DbUrl: string + AccountsUrl: string + Secret: string +} + +const parseNumber = (str: string | undefined): number | undefined => (str !== undefined ? Number(str) : undefined) + +const config: Config = (() => { + const params: Partial = { + Port: parseNumber(process.env.PORT) ?? 8090, + DbUrl: process.env.DB_URL, + AccountsUrl: process.env.ACCOUNTS_URL, + Secret: process.env.SECRET + } + + const missingEnv = Object.entries(params) + .filter(([, value]) => value === undefined) + .map(([key]) => key) + + if (missingEnv.length > 0) { + throw Error(`Missing env variables: ${missingEnv.join(', ')}`) + } + + return params as Config +})() + +export default config diff --git a/packages/server/src/eventProcessor.ts b/packages/server/src/eventProcessor.ts new file mode 100644 index 0000000000..ecdb2d9a5d --- /dev/null +++ b/packages/server/src/eventProcessor.ts @@ -0,0 +1,275 @@ +import { type Message, type Patch, type Reaction, type Attachment } from '@communication/types' +import { + EventType, + type CreateAttachmentEvent, + type AttachmentCreatedEvent, + type CreateMessageEvent, + type MessageCreatedEvent, + type CreatePatchEvent, + type PatchCreatedEvent, + type CreateReactionEvent, + type ReactionCreatedEvent, + type Event, + type BroadcastEvent, + type RemoveAttachmentEvent, + type AttachmentRemovedEvent, + type RemoveMessageEvent, + type MessageRemovedEvent, + type RemoveReactionEvent, + type ReactionRemovedEvent, + type EventResult, + type DbAdapter, + type CreateNotificationEvent, + type RemoveNotificationEvent, + type CreateNotificationContextEvent, + type RemoveNotificationContextEvent, + type UpdateNotificationContextEvent, + type NotificationRemovedEvent, + type NotificationContextCreatedEvent, + type NotificationContextRemovedEvent, + type NotificationContextUpdatedEvent +} from '@communication/sdk-types' + +type Result = { + broadcastEvent?: BroadcastEvent + result: EventResult +} + +export class EventProcessor { + constructor( + private readonly db: DbAdapter, + private readonly workspace: string, + private readonly personWorkspace: string + ) {} + + async process(event: Event): Promise { + switch (event.type) { + case EventType.CreateMessage: + return await this.createMessage(event) + case EventType.RemoveMessage: + return await this.removeMessage(event) + case EventType.CreatePatch: + return await this.createPatch(event) + case EventType.CreateReaction: + return await this.createReaction(event) + case EventType.RemoveReaction: + return await this.removeReaction(event) + case EventType.CreateAttachment: + return await this.createAttachment(event) + case EventType.RemoveAttachment: + return await this.removeAttachment(event) + case EventType.CreateNotification: + return await this.createNotification(event) + case EventType.RemoveNotification: + return await this.removeNotification(event) + case EventType.CreateNotificationContext: + return await this.createNotificationContext(event) + case EventType.RemoveNotificationContext: + return await this.removeNotificationContext(event) + case EventType.UpdateNotificationContext: + return await this.updateNotificationContext(event) + } + } + + private async createMessage(event: CreateMessageEvent): Promise { + const created = new Date() + const id = await this.db.createMessage(event.content, event.creator, created) + await this.db.placeMessage(id, event.card, this.workspace) + const message: Message = { + id, + content: event.content, + creator: event.creator, + created: created, + edited: created, + reactions: [], + attachments: [] + } + const broadcastEvent: MessageCreatedEvent = { + type: EventType.MessageCreated, + card: event.card, + message + } + return { + broadcastEvent, + result: { id } + } + } + + private async createPatch(event: CreatePatchEvent): Promise { + const created = new Date() + await this.db.createPatch(event.message, event.content, event.creator, created) + + const patch: Patch = { + message: event.message, + content: event.content, + creator: event.creator, + created: created + } + const broadcastEvent: PatchCreatedEvent = { + type: EventType.PatchCreated, + patch + } + return { + broadcastEvent, + result: {} + } + } + + private async removeMessage(event: RemoveMessageEvent): Promise { + await this.db.removeMessage(event.message) + + const broadcastEvent: MessageRemovedEvent = { + type: EventType.MessageRemoved, + message: event.message + } + + return { + broadcastEvent, + result: {} + } + } + + private async createReaction(event: CreateReactionEvent): Promise { + const created = new Date() + await this.db.createReaction(event.message, event.reaction, event.creator, created) + + const reaction: Reaction = { + message: event.message, + reaction: event.reaction, + creator: event.creator, + created: created + } + const broadcastEvent: ReactionCreatedEvent = { + type: EventType.ReactionCreated, + reaction + } + return { + broadcastEvent, + result: {} + } + } + + private async removeReaction(event: RemoveReactionEvent): Promise { + await this.db.removeReaction(event.message, event.reaction, event.creator) + const broadcastEvent: ReactionRemovedEvent = { + type: EventType.ReactionRemoved, + message: event.message, + reaction: event.reaction, + creator: event.creator + } + return { + broadcastEvent, + result: {} + } + } + + private async createAttachment(event: CreateAttachmentEvent): Promise { + const created = new Date() + await this.db.createAttachment(event.message, event.card, event.creator, created) + + const attachment: Attachment = { + message: event.message, + card: event.card, + creator: event.creator, + created: created + } + const broadcastEvent: AttachmentCreatedEvent = { + type: EventType.AttachmentCreated, + attachment + } + + return { + broadcastEvent, + result: {} + } + } + + private async removeAttachment(event: RemoveAttachmentEvent): Promise { + await this.db.removeAttachment(event.message, event.card) + const broadcastEvent: AttachmentRemovedEvent = { + type: EventType.AttachmentRemoved, + message: event.message, + card: event.card + } + return { + broadcastEvent, + result: {} + } + } + + private async createNotification(event: CreateNotificationEvent): Promise { + await this.db.createNotification(event.message, event.context) + + return { + result: {} + } + } + + private async removeNotification(event: RemoveNotificationEvent): Promise { + await this.db.removeNotification(event.message, event.context) + + const broadcastEvent: NotificationRemovedEvent = { + type: EventType.NotificationRemoved, + personWorkspace: this.personWorkspace, + message: event.message, + context: event.context + } + return { + broadcastEvent, + result: {} + } + } + + private async createNotificationContext(event: CreateNotificationContextEvent): Promise { + const id = await this.db.createContext( + this.workspace, + event.card, + this.personWorkspace, + event.lastView, + event.lastUpdate + ) + const broadcastEvent: NotificationContextCreatedEvent = { + type: EventType.NotificationContextCreated, + context: { + id, + workspace: this.workspace, + personWorkspace: this.personWorkspace, + card: event.card, + lastView: event.lastView, + lastUpdate: event.lastUpdate + } + } + return { + broadcastEvent, + result: { id } + } + } + + private async removeNotificationContext(event: RemoveNotificationContextEvent): Promise { + await this.db.removeContext(event.context) + const broadcastEvent: NotificationContextRemovedEvent = { + type: EventType.NotificationContextRemoved, + personWorkspace: this.personWorkspace, + context: event.context + } + return { + broadcastEvent, + result: {} + } + } + + async updateNotificationContext(event: UpdateNotificationContextEvent): Promise { + await this.db.updateContext(event.context, event.update) + + const broadcastEvent: NotificationContextUpdatedEvent = { + type: EventType.NotificationContextUpdated, + personWorkspace: this.personWorkspace, + context: event.context, + update: event.update + } + return { + broadcastEvent, + result: {} + } + } +} diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts new file mode 100644 index 0000000000..f959c1eb96 --- /dev/null +++ b/packages/server/src/index.ts @@ -0,0 +1,6 @@ +import { config } from 'dotenv' + +import { main } from './main.ts' + +config() +void main() diff --git a/packages/server/src/main.ts b/packages/server/src/main.ts new file mode 100644 index 0000000000..8d22d60e43 --- /dev/null +++ b/packages/server/src/main.ts @@ -0,0 +1,141 @@ +import WebSocket, { WebSocketServer, type RawData } from 'ws' +import { createDbAdapter } from '@communication/cockroach' +import type { Response, HelloRequest } from '@communication/sdk-types' +import { decodeToken } from '@hcengineering/server-token' + +import type { ConnectionInfo } from './types.ts' +import { deserializeRequest, serializeResponse } from './utils/serialize.ts' +import config from './config' +import { listen, createServer } from './server/server' +import { ConsoleLogger } from './utils/logger' +import { Manager } from './manager.ts' +import type { Session } from './session.ts' +import { getWorkspaceInfo } from './utils/account.ts' + +const logger = new ConsoleLogger() + +const pingTimeout = 10000 +const requestTimeout = 60 * 1000 + +//TODO: use platform errors +const UNAUTHORIZED_ERROR = 'Unauthorized' +const UNKNOWN_ERROR = 'Unknown' + +export const main = async (): Promise => { + const server = listen(createServer(), config.Port) + const wss = new WebSocketServer({ noServer: true }) + const db = await createDbAdapter(config.DbUrl) + const manager = new Manager(db) + + server.on('upgrade', async (req, socket, head) => { + const url = new URL('http://localhost' + (req.url ?? '')) + const token = url.searchParams.get('token') ?? '' + + try { + const info = await validateToken(token) + wss.handleUpgrade(req, socket, head, (ws) => { + handleConnection(ws, manager, info) + }) + } catch (error: any) { + logger.error('Invalid token', { error }) + wss.handleUpgrade(req, socket, head, (ws) => { + const resp: Response = { + result: UNAUTHORIZED_ERROR, + error + } + sendResponse(ws, resp, false) + socket.destroy() + }) + } + }) + + const shutdown = (): void => { + db.close() + server.close(() => { + process.exit() + }) + } + + process.on('SIGINT', shutdown) + process.on('SIGTERM', shutdown) + process.on('uncaughtException', (e) => { + console.error(e) + }) + process.on('unhandledRejection', (e) => { + console.error(e) + }) +} + +function handleConnection(ws: WebSocket, manager: Manager, info: ConnectionInfo) { + const session = manager.createSession(ws, info) + const pingInterval = setInterval(() => { + const now = Date.now() + const lastRequestDiff = now - session.lastRequest + + if (lastRequestDiff > requestTimeout) { + console.warn(`Connection inactive for ${lastRequestDiff}ms, closing`, info) + manager.closeSession(ws, info.workspace) + ws.close() + clearInterval(pingInterval) + return + } + sendResponse(ws, { id: 'ping', result: 'ping' }, session.binary) + }, pingTimeout) + + ws.on('message', async (raw: RawData) => { + try { + await handleRequest(raw, session, ws) + } catch (err: any) { + logger.error('Error during message handling', { err }) + } + }) + + ws.on('close', () => { + manager.closeSession(ws, info.workspace) + clearInterval(pingInterval) + }) + + ws.on('error', (error) => { + logger.log('Error', { error, ...info }) + }) +} + +function sendResponse(ws: WebSocket, resp: Response, binary: boolean) { + ws.send(serializeResponse(resp, binary), { binary }) +} + +async function handleRequest(raw: RawData, session: Session, ws: WebSocket) { + const request = deserializeRequest(raw, session.binary) + if (request === undefined) return + + if (request.id === 'hello') { + const hello = request as HelloRequest + session.binary = hello.binary ?? false + sendResponse(ws, { id: 'hello', result: 'hello' }, false) + return + } + + try { + const fn = (session as any)[request.method] + const params = [...request.params] + const result = await fn.apply(session, params) + const response: Response = { id: request.id, result } + sendResponse(ws, response, session.binary) + } catch (err: any) { + const response: Response = { id: request.id, result: UNKNOWN_ERROR, error: err } + sendResponse(ws, response, session.binary) + } +} + +//TODO: decodeToken or authorize with account service or both +async function validateToken(token: string): Promise { + const { email } = decodeToken(token, true, config.Secret) + const info = await getWorkspaceInfo(token) + + if (info === undefined) { + throw new Error('No workspace info') + } + + const personWorkspace = 'cd0aba36-1c4f-4170-95f2-27a12a5415f7' + return { workspace: info.workspaceId, personWorkspace, socialId: email } +} diff --git a/packages/server/src/manager.ts b/packages/server/src/manager.ts new file mode 100644 index 0000000000..3bb640cc36 --- /dev/null +++ b/packages/server/src/manager.ts @@ -0,0 +1,179 @@ +import WebSocket from 'ws' +import { + type BroadcastEvent, + type DbAdapter, + EventType, + type MessageCreatedEvent, + type NotificationContextCreatedEvent, + type NotificationCreatedEvent, + type Response +} from '@communication/sdk-types' +import type { FindMessagesParams, FindNotificationContextParams, FindNotificationsParams } from '@communication/types' + +import { Session } from './session' +import type { ConnectionInfo } from './types' +import { serializeResponse } from './utils/serialize.ts' +import { Triggers } from './triggers.ts' + +type QueryId = number | string +type QueryType = 'message' | 'notification' | 'context' + +type SessionInfo = { + session: Session + ws: WebSocket + messageQueries: Map + notificationQueries: Map + contextQueries: Map +} + +export class Manager { + private sessionsByWorkspace: Map = new Map() + private triggers: Triggers + private lastSessionId: number = 0 + + constructor(private readonly db: DbAdapter) { + this.triggers = new Triggers(db) + } + + createSession(ws: WebSocket, info: ConnectionInfo): Session { + const current = this.sessionsByWorkspace.get(info.workspace) ?? [] + this.lastSessionId++ + const session = new Session(this.lastSessionId, info, this.db, this) + current.push({ session, ws, messageQueries: new Map(), notificationQueries: new Map(), contextQueries: new Map() }) + this.sessionsByWorkspace.set(info.workspace, current) + return session + } + + closeSession(ws: WebSocket, workspace: string): void { + const sessions = this.sessionsByWorkspace.get(workspace) ?? [] + if (sessions.length === 0) return + const newSessions = sessions.filter((it) => it.ws !== ws) + if (newSessions.length === 0) { + this.sessionsByWorkspace.delete(workspace) + } else { + this.sessionsByWorkspace.set(workspace, newSessions) + } + } + + getSessionInfo(sessionId: number, workspace: string): SessionInfo | undefined { + const sessions = this.sessionsByWorkspace.get(workspace) ?? [] + return sessions.find((it) => it.session.id === sessionId) + } + + subscribeQuery( + sessionId: number, + workspace: string, + type: QueryType, + queryId: number, + params: Record + ): void { + const info = this.getSessionInfo(sessionId, workspace) + if (info == null) return + + if (type === 'message') { + info.messageQueries.set(queryId, params) + } else if (type === 'notification') { + info.notificationQueries.set(queryId, params) + } else if (type === 'context') { + info.contextQueries.set(queryId, params) + } + } + + unsubscribeQuery(sessionId: number, workspace: string, queryId: number): void { + const info = this.getSessionInfo(sessionId, workspace) + if (info == null) return + + info.messageQueries.delete(queryId) + info.notificationQueries.delete(queryId) + info.contextQueries.delete(queryId) + } + + async next(event: BroadcastEvent, workspace: string): Promise { + await this.broadcast(event, workspace) + const derived = await this.triggers.process(event, workspace) + const derivedPromises: Promise[] = [] + for (const d of derived) { + derivedPromises.push(this.next(d, workspace)) + } + await Promise.all(derivedPromises) + } + + private async broadcast(event: BroadcastEvent, workspace: string): Promise { + const sessions = this.sessionsByWorkspace.get(workspace) ?? [] + const response: Response = { result: event } + for (const session of sessions) { + const msg = serializeResponse(response, session.session.binary) + if (this.match(event, session)) { + session.ws.send(msg) + } + } + } + + private match(event: BroadcastEvent, info: SessionInfo): boolean { + switch (event.type) { + case EventType.MessageCreated: + return this.matchMessagesQuery(event, Array.from(info.messageQueries.values())) + case EventType.PatchCreated: + case EventType.MessageRemoved: + case EventType.ReactionCreated: + case EventType.ReactionRemoved: + case EventType.AttachmentCreated: + case EventType.AttachmentRemoved: + return info.messageQueries.size > 0 + case EventType.NotificationCreated: + return ( + info.session.info.personWorkspace === event.personWorkspace && + this.matchNotificationQuery(event, Array.from(info.notificationQueries.values())) + ) + case EventType.NotificationRemoved: + return info.session.info.personWorkspace === event.personWorkspace && info.notificationQueries.size > 0 + case EventType.NotificationContextCreated: + return ( + info.session.info.personWorkspace === event.context.personWorkspace && + this.matchContextQuery(event, Array.from(info.contextQueries.values())) + ) + case EventType.NotificationContextRemoved: + return info.session.info.personWorkspace === event.personWorkspace && info.contextQueries.size > 0 + case EventType.NotificationContextUpdated: + return info.session.info.personWorkspace === event.personWorkspace && info.contextQueries.size > 0 + } + } + + private matchMessagesQuery(event: MessageCreatedEvent, queries: FindMessagesParams[]): boolean { + if (queries.length === 0) return false + + for (const query of queries) { + if (query.id != null && query.id !== event.message.id) continue + if (query.card != null && query.card !== event.card) continue + return true + } + + return false + } + + private matchNotificationQuery(event: NotificationCreatedEvent, queries: FindNotificationsParams[]): boolean { + if (queries.length === 0) return false + + for (const query of queries) { + if (query.context != null && query.context !== event.notification.context) continue + if (query.message != null && query.message !== event.notification.message.id) continue + if (query.read != null && query.read !== event.notification.read) continue + if (query.archived != null && query.archived !== event.notification.archived) continue + return true + } + + return false + } + + private matchContextQuery(event: NotificationContextCreatedEvent, queries: FindNotificationContextParams[]): boolean { + if (queries.length === 0) return false + + for (const query of queries) { + if (query.id != null && query.id !== event.context.id) continue + if (query.card != null && query.card !== event.context.card) continue + return true + } + + return false + } +} diff --git a/packages/server/src/server/error.ts b/packages/server/src/server/error.ts new file mode 100644 index 0000000000..55af5f0946 --- /dev/null +++ b/packages/server/src/server/error.ts @@ -0,0 +1,8 @@ +export class ApiError extends Error { + constructor ( + readonly code: string, + readonly message: string + ) { + super(message) + } +} \ No newline at end of file diff --git a/packages/server/src/server/server.ts b/packages/server/src/server/server.ts new file mode 100644 index 0000000000..0941596c97 --- /dev/null +++ b/packages/server/src/server/server.ts @@ -0,0 +1,26 @@ +import cors from 'cors' +import express, { type Express } from 'express' +import { Server } from 'http' + +import { ApiError } from './error' + +export function createServer (): Express { + const app = express() + + app.use(cors()) + app.use(express.json()) + + app.use((_req, res, _next) => { + res.status(404).send({ message: 'Not found' }) + }) + + return app +} + +export function listen (e: Express, port: number, host?: string): Server { + const cb = (): void => { + console.log(`Communication server has been started at ${host ?? '*'}:${port}`) + } + + return host !== undefined ? e.listen(port, host, cb) : e.listen(port, cb) +} diff --git a/packages/server/src/session.ts b/packages/server/src/session.ts new file mode 100644 index 0000000000..3efd0a1a86 --- /dev/null +++ b/packages/server/src/session.ts @@ -0,0 +1,76 @@ +import type { DbAdapter, Event, EventResult } from '@communication/sdk-types' +import type { + FindMessagesParams, + FindNotificationContextParams, + FindNotificationsParams, + Message, + Notification, + NotificationContext +} from '@communication/types' + +import type { ConnectionInfo } from './types' +import { EventProcessor } from './eventProcessor.ts' +import type { Manager } from './manager.ts' + +export class Session { + binary: boolean = false + lastRequest: number = Date.now() + + private readonly eventProcessor: EventProcessor + + constructor( + readonly id: number, + readonly info: ConnectionInfo, + private readonly db: DbAdapter, + private readonly manager: Manager + ) { + this.eventProcessor = new EventProcessor(db, info.workspace, info.personWorkspace) + } + + ping(): string { + this.lastRequest = Date.now() + return 'pong' + } + + async findMessages(params: FindMessagesParams, queryId?: number): Promise { + const result = await this.db.findMessages(this.info.workspace, params) + if (queryId != null) { + this.manager.subscribeQuery(this.id, this.info.workspace, 'message', queryId, params) + } + return result + } + + async unsubscribeQuery(id: number): Promise { + this.manager.unsubscribeQuery(this.id, this.info.workspace, id) + } + + async findNotifications(params: FindNotificationsParams, queryId?: number): Promise { + //TODO: do we need filter by workspace by default? + const result = await this.db.findNotifications(params, this.info.personWorkspace) + if (queryId != null) { + this.manager.subscribeQuery(this.id, this.info.workspace, 'notification', queryId, params) + } + return result + } + + async findNotificationContexts( + params: FindNotificationContextParams, + queryId?: number + ): Promise { + //TODO: do we need filter by workspace by default? + const result = await this.db.findContexts(params, [this.info.personWorkspace]) + if (queryId != null) { + this.manager.subscribeQuery(this.id, this.info.workspace, 'context', queryId, params) + } + + return result + } + + async event(event: Event): Promise { + const { result, broadcastEvent } = await this.eventProcessor.process(event) + if (broadcastEvent !== undefined) { + void this.manager.next(broadcastEvent, this.info.workspace) + } + return result + } +} diff --git a/packages/server/src/triggers.ts b/packages/server/src/triggers.ts new file mode 100644 index 0000000000..4f131853c6 --- /dev/null +++ b/packages/server/src/triggers.ts @@ -0,0 +1,112 @@ +import { + type BroadcastEvent, + type DbAdapter, + EventType, + type MessageCreatedEvent, + type NotificationContextCreatedEvent, + type NotificationCreatedEvent +} from '@communication/sdk-types' +import type { NotificationContext, ContextID } from '@communication/types' + +export class Triggers { + constructor(private readonly db: DbAdapter) {} + + async process(event: BroadcastEvent, workspace: string): Promise { + switch (event.type) { + case EventType.MessageCreated: + return this.createNotifications(event, workspace) + } + + return [] + } + + private async createNotifications(event: MessageCreatedEvent, workspace: string): Promise { + const card = event.card + const subscribedPersonWorkspaces = ['cd0aba36-1c4f-4170-95f2-27a12a5415f7', 'cd0aba36-1c4f-4170-95f2-27a12a5415f8'] + + const res: BroadcastEvent[] = [] + const contexts = await this.db.findContexts({ card }, [], workspace) + + res.push(...(await this.updateNotificationContexts(event.message.created, contexts))) + + for (const personWorkspace of subscribedPersonWorkspaces) { + const existsContext = contexts.find( + (it) => it.card === card && it.personWorkspace === personWorkspace && workspace === it.workspace + ) + const contextId = await this.getOrCreateContextId( + workspace, + card, + personWorkspace, + res, + event.message.created, + existsContext + ) + + await this.db.createNotification(event.message.id, contextId) + + const resultEvent: NotificationCreatedEvent = { + type: EventType.NotificationCreated, + personWorkspace, + notification: { + context: contextId, + message: event.message, + read: false, + archived: false + } + } + res.push(resultEvent) + } + + return res + } + + private async getOrCreateContextId( + workspace: string, + card: string, + personWorkspace: string, + res: BroadcastEvent[], + lastUpdate: Date, + context?: NotificationContext + ): Promise { + if (context !== undefined) { + return context.id + } else { + const contextId = await this.db.createContext(workspace, card, personWorkspace, undefined, lastUpdate) + const newContext = { + id: contextId, + card, + workspace, + personWorkspace + } + const resultEvent: NotificationContextCreatedEvent = { + type: EventType.NotificationContextCreated, + context: newContext + } + + res.push(resultEvent) + + return contextId + } + } + + private async updateNotificationContexts( + lastUpdate: Date, + contexts: NotificationContext[] + ): Promise { + const res: BroadcastEvent[] = [] + for (const context of contexts) { + if (context.lastUpdate === undefined || context.lastUpdate < lastUpdate) { + await this.db.updateContext(context.id, { lastUpdate }) + res.push({ + type: EventType.NotificationContextUpdated, + personWorkspace: context.personWorkspace, + context: context.id, + update: { + lastUpdate + } + }) + } + } + return res + } +} diff --git a/packages/server/src/types.ts b/packages/server/src/types.ts new file mode 100644 index 0000000000..8d46b65816 --- /dev/null +++ b/packages/server/src/types.ts @@ -0,0 +1,7 @@ +import type { SocialID } from '@communication/types' + +export interface ConnectionInfo { + workspace: string + personWorkspace: string + socialId: SocialID +} diff --git a/packages/server/src/utils/account.ts b/packages/server/src/utils/account.ts new file mode 100644 index 0000000000..882e437f25 --- /dev/null +++ b/packages/server/src/utils/account.ts @@ -0,0 +1,23 @@ +import config from "../config.ts"; + + +type WorkspaceInfo = { +workspaceId: string +} + +export async function getWorkspaceInfo (token: string): Promise { + const accountsUrl = config.AccountsUrl + const response = await fetch(accountsUrl, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Authorization: 'Bearer ' + token + }, + body: JSON.stringify({ + method: 'getWorkspaceInfo', + params: [] + }) + }) + const result = await response.json() + return result.result as WorkspaceInfo | undefined +} \ No newline at end of file diff --git a/packages/server/src/utils/logger.ts b/packages/server/src/utils/logger.ts new file mode 100644 index 0000000000..828c27e432 --- /dev/null +++ b/packages/server/src/utils/logger.ts @@ -0,0 +1,24 @@ +export interface Logger { + log: (message: string, data?: Record) => void + warn: (message: string, data?: Record) => void + error: (message: string, data?: Record) => void + debug: (message: string, data?: Record) => void +} + +export class ConsoleLogger implements Logger { + log (message: string, data?: Record): void { + console.log({ message, ...data }) + } + + warn (message: string, data?: Record): void { + console.warn({ message, ...data }) + } + + error (message: string, data?: Record): void { + console.error({ message, ...data }) + } + + debug (message: string, data?: Record): void { + console.debug({ message, ...data }) + } +} \ No newline at end of file diff --git a/packages/server/src/utils/serialize.ts b/packages/server/src/utils/serialize.ts new file mode 100644 index 0000000000..1aa35209da --- /dev/null +++ b/packages/server/src/utils/serialize.ts @@ -0,0 +1,41 @@ +import { Packr } from 'msgpackr' +import type {Response, Request} from '@communication/sdk-types' + +import type {RawData} from "ws"; + +const packr = new Packr({ structuredClone: true, bundleStrings: true, copyBuffers: false }) + +export function serializeResponse(resp: Response, binary: boolean) { + return binary ? serializeBinary(resp) : serializeJson(resp) +} + +export function deserializeRequest(raw: RawData, binary: boolean): Request | undefined { + let buff: Buffer | undefined + if (raw instanceof Buffer) { + buff = raw + } else if (Array.isArray(raw)) { + buff = Buffer.concat(raw.map(it => new Uint8Array(it))) + } + + if(buff === undefined) { + return undefined + } + + return binary ? deserializeBinary(buff) : deserializeJson(buff) +} + +function deserializeBinary(data: any): any { + return packr.decode(data) +} + +function deserializeJson(data: any): any { + return JSON.parse(data.toString()) +} + +function serializeBinary(data: any) { + return new Uint8Array(packr.encode(data)) +} + +function serializeJson(data: any) { + return JSON.stringify(data) +} \ No newline at end of file diff --git a/packages/server/tsconfig.json b/packages/server/tsconfig.json new file mode 100644 index 0000000000..49e05cea1e --- /dev/null +++ b/packages/server/tsconfig.json @@ -0,0 +1,8 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { + "outDir": "./dist", + "rootDir": "./src" + }, + "include": ["src"] +}