Merge pull request 'Implement messages' (#25) from implement-messages into main

Reviewed-on: https://git.verdigado.com/NB-Public/rocketchat2matrix/pulls/25

.env.example

@@ -1,2 +1,3 @@
 REGISTRATION_SHARED_SECRET='look in your synapses homeserver.yaml'
+AS_TOKEN='look in the app-service.yaml'
 EXCLUDED_USERS='rocket.cat' # Comma-separated list of usernames or IDs

README.md

@@ -14,13 +14,15 @@ mongoexport --collection=users --db=rocketchat --out=users.json
 
 Export them to `inputs/`
 
-## Running the Matrix Dev Server
+## Configuring the Matrix Dev Server
+
+Generate a Synapse homeserver config with the following command (you might change `my.matrix.host` for the actual server name, as it can't be changed afterwards):
 
 ```shell
 docker-compose run --rm -e SYNAPSE_SERVER_NAME=my.matrix.host -e SYNAPSE_REPORT_STATS=no synapse generate
 ```
 
-To run the script without hitting rate limiting, you SHOULD add the following options to the freshly generated `files/homeserver.yaml`. **Do not leave these in the production setup!**
+To run the script without hitting rate limiting and activating an *Application Service* to send messages by different users with our desired timestamps, you MUST add the following options to the freshly generated `files/homeserver.yaml`. **Do not leave these in the production setup!**
 
 ```yaml
 rc_joins:
@@ -43,9 +45,17 @@ rc_invites:
   per_issuer:
     per_second: 1024
     burst_count: 2048
+app_service_config_files:
+  - /data/app-service.yaml
 ```
 
-Continue setting up the server:
+Now edit `app-service.example.yaml` and save it at `files/app-service.yaml`, changing the tokens.
+
+Copy over `.env.example` to `.env` and insert your values.
+
+## Starting the Matrix Dev Server
+
+Boot up the container and (for the first time starting the server or after resetting it manually) create an admin user:
 
 ```shell
 docker-compose up -d
@@ -65,12 +75,12 @@ curl --request POST \
 > src/config/synapse_access_token.json
 ```
 
+## Installing and Running the Script
+
+Install NodeJS and npm on your system, install the script's dependencies via `npm install`.
+
 To finally run the script, execute it via `npm start`.
 
-## Configuration
-
-Copy over `.env.example` to `.env` and insert your values.
-
 ## Running Tests
 
 `npm test`.
@@ -85,7 +95,7 @@ sudo rm files/homeserver.db
 rm db.sqlite
 ```
 
-Then you can restart with an empty but quite equal server, following the instructions above, excluding the `generate` command.
+Then you can restart with an empty but quite equal server, following the instructions above to start the dev server.
 
 ## Design Decisions
 

app-service.example.yaml

@@ -0,0 +1,9 @@
+id: "rc2matrix migration"
+url: "http://127.0.0.1:1234"
+as_token: "30c05ae90a248a4188e620216fa72e349803310ec83e2a77b34fe90be6081f46" # Change me to something different and random
+hs_token: "312df522183efd404ec1cd22d2ffa4bbc76a8c1ccf541dd692eef281356bb74e" # Change me to something different and random
+sender_localpart: "_rc_migration_bot"
+namespaces:
+   users:
+    - exclusive: false
+      regex: "@.*"

src/app.ts

@@ -4,10 +4,12 @@ import lineByLine from 'n-readlines'
 import 'reflect-metadata'
 import { handle as handleRoom } from './handlers/rooms'
 import { handle as handleUser } from './handlers/users'
+import { handle as handleMessage } from './handlers/messages'
 import log from './helpers/logger'
 import { initStorage } from './helpers/storage'
 import { whoami } from './helpers/synapse'
 import { Entity, entities } from './Entities'
+import { AxiosError } from 'axios'
 
 log.info('rocketchat2matrix starts.')
 
@@ -27,7 +29,7 @@ async function loadRcExport(entity: Entity) {
         break
 
       case Entity.Messages:
-        log.debug(`Message: ${item.name}`)
+        await handleMessage(item)
         break
 
       default:
@@ -44,9 +46,17 @@ async function main() {
     await loadRcExport(Entity.Users)
     log.info('Parsing rooms')
     await loadRcExport(Entity.Rooms)
+    log.info('Parsing messages')
+    await loadRcExport(Entity.Messages)
     log.info('Done.')
   } catch (error) {
-    log.error(`Encountered an error while booting up: ${error}`, error)
+    if (error instanceof AxiosError) {
+      log.error(`Error during request: ${error.message}`)
+      log.error(`Request: ${error.request?.method} ${error.request?.path}`)
+      log.error(`Response: ${error.response?.status}`, error.response?.data)
+    } else {
+      log.error(`Encountered an error while booting up: ${error}`, error)
+    }
   }
 }
 

src/handlers/messages.test.ts

@@ -0,0 +1,87 @@
+process.env.AS_TOKEN = 'ApplicationSecretToken'
+process.env.EXCLUDED_USERS = 'excludedUser1,excludedUser2'
+import { expect, jest, test } from '@jest/globals'
+import axios from 'axios'
+import * as storage from '../helpers/storage'
+import {
+  MatrixMessage,
+  RcMessage,
+  createMessage,
+  handle,
+  mapMessage,
+} from './messages'
+
+jest.mock('axios')
+const mockedAxios = axios as jest.Mocked<typeof axios>
+
+jest.mock('../helpers/storage')
+const mockedStorage = storage as jest.Mocked<typeof storage>
+
+const rcMessage: RcMessage = {
+  _id: 'testMessage',
+  rid: 'testRoom',
+  msg: 'Test',
+  u: {
+    _id: 'testUser',
+  },
+  ts: {
+    $date: '1970-01-02T06:51:51.0Z', // UNIX-TS: 111111000
+  },
+}
+
+const matrixMessage: MatrixMessage = {
+  body: 'Test',
+  msgtype: 'm.text',
+  type: 'm.room.message',
+}
+
+test('mapping messages', () => {
+  expect(mapMessage(rcMessage)).toStrictEqual(matrixMessage)
+})
+
+test('creating messages', async () => {
+  mockedAxios.put.mockResolvedValue({ data: { event_id: 'message@matrix' } })
+
+  await expect(
+    createMessage(matrixMessage, 'roomID', 'userID', 42, 'transactionId')
+  ).resolves.toBe('message@matrix')
+
+  expect(mockedAxios.put).toHaveBeenCalledWith(
+    '/_matrix/client/v3/rooms/roomID/send/m.room.message/transactionId?user_id=userID&ts=42',
+    matrixMessage,
+    { headers: { Authorization: 'Bearer ApplicationSecretToken' } }
+  )
+  mockedAxios.put.mockClear()
+})
+
+test('handling messages', async () => {
+  mockedAxios.put.mockResolvedValue({ data: { event_id: 'test@matrix' } })
+  mockedStorage.getRoomId.mockResolvedValue('testMatrixRoom')
+  mockedStorage.getUserId.mockResolvedValue('testMatrixUser')
+  mockedStorage.getMessageId.mockResolvedValueOnce(undefined) // For checking if the Message already exists
+  mockedStorage.getMessageId.mockResolvedValue('testMatrixMessage') // For checking the parent message
+
+  await expect(handle({ ...rcMessage, tmid: 'threadId' })).resolves.toBe(
+    undefined
+  )
+
+  expect(mockedAxios.put).toHaveBeenLastCalledWith(
+    '/_matrix/client/v3/rooms/testMatrixRoom/send/m.room.message/testMessage?user_id=testMatrixUser&ts=111111000',
+    {
+      ...matrixMessage,
+      'm.relates_to': {
+        rel_type: 'm.thread',
+        event_id: 'testMatrixMessage',
+        is_falling_back: true,
+        'm.in_reply_to': {
+          event_id: 'testMatrixMessage',
+        },
+      },
+    },
+    { headers: { Authorization: 'Bearer ApplicationSecretToken' } }
+  )
+  expect(mockedStorage.getRoomId).toHaveBeenLastCalledWith('testRoom')
+  expect(mockedStorage.getUserId).toHaveBeenLastCalledWith('testUser')
+  expect(mockedStorage.getMessageId).toHaveBeenLastCalledWith('threadId')
+  mockedAxios.put.mockClear()
+})

src/handlers/messages.ts

@@ -0,0 +1,146 @@
+import { Entity, entities } from '../Entities'
+import { IdMapping } from '../entity/IdMapping'
+import log from '../helpers/logger'
+import { getMessageId, getRoomId, getUserId, save } from '../helpers/storage'
+import { axios, formatUserSessionOptions } from '../helpers/synapse'
+
+const applicationServiceToken = process.env.AS_TOKEN || ''
+if (!applicationServiceToken) {
+  const message = 'No AS_TOKEN found in .env.'
+  log.error(message)
+  throw new Error(message)
+}
+
+export type RcMessage = {
+  _id: string
+  t?: string // Event type
+  rid: string // The unique id for the room
+  msg: string // The content of the message.
+  tmid?: string
+  ts: {
+    $date: string
+  }
+  mentions?: string[]
+  u: {
+    _id: string
+    username?: string
+    name?: string
+  }
+  // md?: any // The message's content in a markdown format.
+  pinned?: boolean
+  drid?: string // The direct room id (if belongs to a direct room).
+  // attachments?: any[] // An array of attachment objects, available only when the message has at least one attachment.
+  reactions?: object // Object containing reaction information associated with the message.
+}
+
+export type MatrixMessage = {
+  body: string
+  msgtype: 'm.text'
+  type: 'm.room.message'
+  'm.relates_to'?: {
+    rel_type: 'm.thread'
+    event_id: string
+    is_falling_back: true
+    'm.in_reply_to': {
+      event_id: string
+    }
+  }
+}
+
+export function mapMessage(rcMessage: RcMessage): MatrixMessage {
+  return {
+    body: rcMessage.msg,
+    msgtype: 'm.text',
+    type: 'm.room.message',
+  }
+}
+
+export async function createMapping(
+  rcId: string,
+  matrixId: string
+): Promise<void> {
+  const messageMapping = new IdMapping()
+  messageMapping.rcId = rcId
+  messageMapping.matrixId = matrixId
+  messageMapping.type = entities[Entity.Messages].mappingType
+
+  await save(messageMapping)
+  log.debug('Mapping added:', messageMapping)
+}
+
+export async function createMessage(
+  matrixMessage: MatrixMessage,
+  room_id: string,
+  user_id: string,
+  ts: number,
+  transactionId: string
+): Promise<string> {
+  return (
+    await axios.put(
+      `/_matrix/client/v3/rooms/${room_id}/send/m.room.message/${transactionId}?user_id=${user_id}&ts=${ts}`,
+      matrixMessage,
+      formatUserSessionOptions(applicationServiceToken)
+    )
+  ).data.event_id
+}
+
+export async function handle(rcMessage: RcMessage): Promise<void> {
+  log.info(`Parsing message with ID: ${rcMessage._id}`)
+
+  const matrixId = await getMessageId(rcMessage._id)
+  if (matrixId) {
+    log.debug(`Mapping exists: ${rcMessage._id} -> ${matrixId}`)
+    return
+  }
+
+  if (rcMessage.t) {
+    log.warn(`Message ${rcMessage._id} is of type ${rcMessage.t}, skipping.`)
+    return
+  }
+
+  const room_id = await getRoomId(rcMessage.rid)
+  if (!room_id) {
+    log.warn(
+      `Could not find room ${rcMessage.rid} for message ${rcMessage._id}, skipping.`
+    )
+    return
+  }
+
+  const user_id = await getUserId(rcMessage.u._id)
+  if (!user_id) {
+    log.warn(
+      `Could not find author ${rcMessage.u.username} for message ${rcMessage._id}, skipping.`
+    )
+    return
+  }
+
+  const matrixMessage = mapMessage(rcMessage)
+
+  const ts = new Date(rcMessage.ts.$date).valueOf()
+  if (rcMessage.tmid) {
+    const event_id = await getMessageId(rcMessage.tmid)
+    if (!event_id) {
+      log.warn(`Related message ${rcMessage.tmid} missing, skipping.`)
+      return
+    } else {
+      matrixMessage['m.relates_to'] = {
+        rel_type: 'm.thread',
+        event_id,
+        is_falling_back: true,
+        'm.in_reply_to': {
+          event_id,
+        },
+      }
+    }
+  }
+
+  const event_id = await createMessage(
+    matrixMessage,
+    room_id,
+    user_id,
+    ts,
+    rcMessage._id
+  )
+
+  createMapping(rcMessage._id, event_id)
+}