Init ws server (#6)

* Init ws server

.npmrc

@@ -0,0 +1,2 @@
+@hcengineering:registry=https://npm.pkg.github.com/
+//npm.pkg.github.com/:_authToken=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

.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
+    }
+  ]
+}

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'}},
 ];
 

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
+}

packages/sdk-types/src/index.ts

@@ -1 +1,3 @@
 export * from './db'
+export * from './event'
+export * from './ws'

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
+}

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"
+  }
+}

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<Config> = {
+    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

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<Result> {
+    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<Result> {
+    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<Result> {
+    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<Result> {
+    await this.db.removeMessage(event.message)
+
+    const broadcastEvent: MessageRemovedEvent = {
+      type: EventType.MessageRemoved,
+      message: event.message
+    }
+
+    return {
+      broadcastEvent,
+      result: {}
+    }
+  }
+
+  private async createReaction(event: CreateReactionEvent): Promise<Result> {
+    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<Result> {
+    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<Result> {
+    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<Result> {
+    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<Result> {
+    await this.db.createNotification(event.message, event.context)
+
+    return {
+      result: {}
+    }
+  }
+
+  private async removeNotification(event: RemoveNotificationEvent): Promise<Result> {
+    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<Result> {
+    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<Result> {
+    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<Result> {
+    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: {}
+    }
+  }
+}

packages/server/src/index.ts

@@ -0,0 +1,6 @@
+import { config } from 'dotenv'
+
+import { main } from './main.ts'
+
+config()
+void main()

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<void> => {
+  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<ConnectionInfo> {
+  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 }
+}

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<QueryId, FindMessagesParams>
+  notificationQueries: Map<QueryId, FindNotificationsParams>
+  contextQueries: Map<QueryId, FindNotificationContextParams>
+}
+
+export class Manager {
+  private sessionsByWorkspace: Map<string, SessionInfo[]> = 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<string, any>
+  ): 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<void> {
+    await this.broadcast(event, workspace)
+    const derived = await this.triggers.process(event, workspace)
+    const derivedPromises: Promise<void>[] = []
+    for (const d of derived) {
+      derivedPromises.push(this.next(d, workspace))
+    }
+    await Promise.all(derivedPromises)
+  }
+
+  private async broadcast(event: BroadcastEvent, workspace: string): Promise<void> {
+    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
+  }
+}

packages/server/src/server/error.ts

@@ -0,0 +1,8 @@
+export class ApiError extends Error {
+    constructor (
+        readonly code: string,
+        readonly message: string
+    ) {
+        super(message)
+    }
+}

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)
+}

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<Message[]> {
+    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<void> {
+    this.manager.unsubscribeQuery(this.id, this.info.workspace, id)
+  }
+
+  async findNotifications(params: FindNotificationsParams, queryId?: number): Promise<Notification[]> {
+    //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<NotificationContext[]> {
+    //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<EventResult> {
+    const { result, broadcastEvent } = await this.eventProcessor.process(event)
+    if (broadcastEvent !== undefined) {
+      void this.manager.next(broadcastEvent, this.info.workspace)
+    }
+    return result
+  }
+}

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<BroadcastEvent[]> {
+    switch (event.type) {
+      case EventType.MessageCreated:
+        return this.createNotifications(event, workspace)
+    }
+
+    return []
+  }
+
+  private async createNotifications(event: MessageCreatedEvent, workspace: string): Promise<BroadcastEvent[]> {
+    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<ContextID> {
+    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<BroadcastEvent[]> {
+    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
+  }
+}

packages/server/src/types.ts

@@ -0,0 +1,7 @@
+import type { SocialID } from '@communication/types'
+
+export interface ConnectionInfo {
+  workspace: string
+  personWorkspace: string
+  socialId: SocialID
+}

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<WorkspaceInfo | undefined> {
+    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
+}

packages/server/src/utils/logger.ts

@@ -0,0 +1,24 @@
+export interface Logger {
+    log: (message: string, data?: Record<string, any>) => void
+    warn: (message: string, data?: Record<string, any>) => void
+    error: (message: string, data?: Record<string, any>) => void
+    debug: (message: string, data?: Record<string, any>) => void
+}
+
+export class ConsoleLogger implements Logger {
+    log (message: string, data?: Record<string, any>): void {
+        console.log({ message, ...data })
+    }
+
+    warn (message: string, data?: Record<string, any>): void {
+        console.warn({ message, ...data })
+    }
+
+    error (message: string, data?: Record<string, any>): void {
+        console.error({ message, ...data })
+    }
+
+    debug (message: string, data?: Record<string, any>): void {
+        console.debug({ message, ...data })
+    }
+}

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)
+}

packages/server/tsconfig.json

@@ -0,0 +1,8 @@
+{
+  "extends": "../../tsconfig.json",
+  "compilerOptions": {
+    "outDir": "./dist",
+    "rootDir": "./src"
+  },
+  "include": ["src"]
+}