chore(2631): Refactor storage service to separate channel and message logic

packages/backend/package.json

@@ -175,8 +175,8 @@
     "socket.io-client": "^4.7.5",
     "string-replace-loader": "3.1.0",
     "ts-jest-resolver": "^2.0.0",
-    "uint8arrays": "^5.1.0",
     "uint8arraylist": "^2.4.8",
+    "uint8arrays": "^5.1.0",
     "utf-8-validate": "^5.0.2",
     "validator": "^13.11.0",
     "ws": "^8.18.0"

packages/backend/src/nest/auth/sigchain.service.ts

@@ -60,9 +60,9 @@ export class SigChainService implements OnModuleInit {
    * @param setActive Whether to set the chain as active
    * @returns Whether the chain was set as active
    */
-  addChain(chain: SigChain, setActive: boolean): boolean {
+  addChain(chain: SigChain, setActive: boolean, teamName?: string): boolean {
     if (this.chains.has(chain.team.teamName)) {
-      throw new Error(`Chain for team ${chain.team.teamName} already exists`)
+      throw new Error(`Chain for team ${teamName} already exists`)
     }
     this.chains.set(chain.team.teamName, chain)
     if (setActive) {

packages/backend/src/nest/common/types.ts

@@ -1,8 +1,8 @@
-import { type EventsType } from '@orbitdb/core'
-import { type ChannelMessage, type PublicChannel } from '@quiet/types'
+import { type PublicChannel } from '@quiet/types'
+import { ChannelStore } from '../storage/channels/channel.store'
 
 export interface PublicChannelsRepo {
-  db: EventsType<ChannelMessage>
+  store: ChannelStore
   eventsAttached: boolean
 }
 

packages/backend/src/nest/common/utils.ts

@@ -1,4 +1,5 @@
 import fs from 'fs'
+import fsAsync from 'fs/promises'
 import getPort from 'get-port'
 import path from 'path'
 import { Server } from 'socket.io'
@@ -266,19 +267,16 @@ export async function createPeerId(): Promise<CreatedLibp2pPeerId> {
   }
 }
 
-export const createArbitraryFile = (filePath: string, sizeBytes: number) => {
-  const stream = fs.createWriteStream(filePath)
+export const createArbitraryFile = async (filePath: string, sizeBytes: number) => {
   const maxChunkSize = 1048576 // 1MB
 
   let remainingSize = sizeBytes
 
   while (remainingSize > 0) {
     const chunkSize = Math.min(maxChunkSize, remainingSize)
-    stream.write(crypto.randomBytes(chunkSize))
+    await fsAsync.appendFile(filePath, crypto.randomBytes(chunkSize))
     remainingSize -= chunkSize
   }
-
-  stream.end()
 }
 
 export async function* asyncGeneratorFromIterator<T>(asyncIterator: AsyncIterable<T>): AsyncGenerator<T> {

packages/backend/src/nest/connections-manager/connections-manager.service.ts

@@ -240,6 +240,7 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI
 
     if (community.name) {
       try {
+        this.logger.info(`Loading chain for team name ${community.name}`)
         await this.sigChainService.loadChain(community.name, true)
       } catch (e) {
         this.logger.warn('Failed to load sigchain', e)
@@ -706,6 +707,7 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI
 
     const community = {
       id: payload.id,
+      name: payload.name,
       peerList: [...new Set([localAddress, ...metadata.peers])],
       psk: metadata.psk,
       ownerOrbitDbIdentity: metadata.ownerOrbitDbIdentity,
@@ -899,7 +901,7 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI
         this.serverIoProvider.io.emit(SocketActionTypes.CERTIFICATES_STORED, {
           certificates: await this.storageService?.loadAllCertificates(),
         })
-        await this.storageService?.loadAllChannels()
+        await this.storageService?.channels.loadAllChannels()
       }
     })
     this.socketService.on(
@@ -954,7 +956,7 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI
     this.socketService.on(
       SocketActionTypes.CREATE_CHANNEL,
       async (args: CreateChannelPayload, callback: (response?: CreateChannelResponse) => void) => {
-        callback(await this.storageService?.subscribeToChannel(args.channel))
+        callback(await this.storageService?.channels.subscribeToChannel(args.channel))
       }
     )
     this.socketService.on(
@@ -963,39 +965,39 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI
         payload: { channelId: string; ownerPeerId: string },
         callback: (response: DeleteChannelResponse) => void
       ) => {
-        callback(await this.storageService?.deleteChannel(payload))
+        callback(await this.storageService?.channels.deleteChannel(payload))
       }
     )
     this.socketService.on(
       SocketActionTypes.DELETE_FILES_FROM_CHANNEL,
       async (payload: DeleteFilesFromChannelSocketPayload) => {
         this.logger.info(`socketService - ${SocketActionTypes.DELETE_FILES_FROM_CHANNEL}`)
-        await this.storageService?.deleteFilesFromChannel(payload)
+        await this.storageService?.channels.deleteFilesFromChannel(payload)
         // await this.deleteFilesFromTemporaryDir() //crashes on mobile, will be fixes in next versions
       }
     )
     this.socketService.on(SocketActionTypes.SEND_MESSAGE, async (args: SendMessagePayload) => {
-      await this.storageService?.sendMessage(args.message)
+      await this.storageService?.channels.sendMessage(args.message)
     })
     this.socketService.on(
       SocketActionTypes.GET_MESSAGES,
       async (payload: GetMessagesPayload, callback: (response?: MessagesLoadedPayload) => void) => {
-        callback(await this.storageService?.getMessages(payload.channelId, payload.ids))
+        callback(await this.storageService?.channels.getMessages(payload.channelId, payload.ids))
       }
     )
 
     // Files
     this.socketService.on(SocketActionTypes.DOWNLOAD_FILE, async (metadata: FileMetadata) => {
-      await this.storageService?.downloadFile(metadata)
+      await this.storageService?.channels.downloadFile(metadata)
     })
     this.socketService.on(SocketActionTypes.UPLOAD_FILE, async (metadata: FileMetadata) => {
-      await this.storageService?.uploadFile(metadata)
+      await this.storageService?.channels.uploadFile(metadata)
     })
     this.socketService.on(SocketActionTypes.FILE_UPLOADED, async (args: FileMetadata) => {
-      await this.storageService?.uploadFile(args)
+      await this.storageService?.channels.uploadFile(args)
     })
     this.socketService.on(SocketActionTypes.CANCEL_DOWNLOAD, mid => {
-      this.storageService?.cancelDownload(mid)
+      this.storageService?.channels.cancelDownload(mid)
     })
 
     // System
@@ -1011,46 +1013,48 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI
 
   private attachStorageListeners() {
     if (!this.storageService) return
-    this.storageService.on(SocketActionTypes.CONNECTION_PROCESS_INFO, data => {
-      this.serverIoProvider.io.emit(SocketActionTypes.CONNECTION_PROCESS_INFO, data)
-    })
-    this.storageService.on(StorageEvents.CERTIFICATES_STORED, (payload: SendCertificatesResponse) => {
-      this.logger.info(`Storage - ${StorageEvents.CERTIFICATES_STORED}`)
-      this.serverIoProvider.io.emit(SocketActionTypes.CERTIFICATES_STORED, payload)
-    })
-    this.storageService.on(StorageEvents.CHANNELS_STORED, (payload: ChannelsReplicatedPayload) => {
+    // Channel and Message Events
+    this.storageService.channels.on(StorageEvents.CHANNELS_STORED, (payload: ChannelsReplicatedPayload) => {
       this.serverIoProvider.io.emit(SocketActionTypes.CHANNELS_STORED, payload)
     })
-    this.storageService.on(StorageEvents.MESSAGES_STORED, (payload: MessagesLoadedPayload) => {
+    this.storageService.channels.on(StorageEvents.MESSAGES_STORED, (payload: MessagesLoadedPayload) => {
       this.serverIoProvider.io.emit(SocketActionTypes.MESSAGES_STORED, payload)
     })
-    this.storageService.on(StorageEvents.MESSAGE_IDS_STORED, (payload: ChannelMessageIdsResponse) => {
+    this.storageService.channels.on(StorageEvents.MESSAGE_IDS_STORED, (payload: ChannelMessageIdsResponse) => {
       if (payload.ids.length === 0) {
         return
       }
       this.serverIoProvider.io.emit(SocketActionTypes.MESSAGE_IDS_STORED, payload)
     })
-    this.storageService.on(StorageEvents.CHANNEL_SUBSCRIBED, (payload: ChannelSubscribedPayload) => {
+    this.storageService.channels.on(StorageEvents.CHANNEL_SUBSCRIBED, (payload: ChannelSubscribedPayload) => {
       this.serverIoProvider.io.emit(SocketActionTypes.CHANNEL_SUBSCRIBED, payload)
     })
-    this.storageService.on(StorageEvents.REMOVE_DOWNLOAD_STATUS, (payload: RemoveDownloadStatus) => {
+    this.storageService.channels.on(StorageEvents.REMOVE_DOWNLOAD_STATUS, (payload: RemoveDownloadStatus) => {
       this.serverIoProvider.io.emit(SocketActionTypes.REMOVE_DOWNLOAD_STATUS, payload)
     })
-    this.storageService.on(StorageEvents.FILE_UPLOADED, (payload: UploadFilePayload) => {
+    this.storageService.channels.on(StorageEvents.FILE_UPLOADED, (payload: UploadFilePayload) => {
       this.serverIoProvider.io.emit(SocketActionTypes.FILE_UPLOADED, payload)
     })
-    this.storageService.on(StorageEvents.DOWNLOAD_PROGRESS, (payload: DownloadStatus) => {
+    this.storageService.channels.on(StorageEvents.DOWNLOAD_PROGRESS, (payload: DownloadStatus) => {
       this.serverIoProvider.io.emit(SocketActionTypes.DOWNLOAD_PROGRESS, payload)
     })
-    this.storageService.on(StorageEvents.MESSAGE_MEDIA_UPDATED, (payload: FileMetadata) => {
+    this.storageService.channels.on(StorageEvents.MESSAGE_MEDIA_UPDATED, (payload: FileMetadata) => {
       this.serverIoProvider.io.emit(SocketActionTypes.MESSAGE_MEDIA_UPDATED, payload)
     })
+    this.storageService.channels.on(StorageEvents.SEND_PUSH_NOTIFICATION, (payload: PushNotificationPayload) => {
+      this.serverIoProvider.io.emit(SocketActionTypes.PUSH_NOTIFICATION, payload)
+    })
+    // Other Events
+    this.storageService.on(SocketActionTypes.CONNECTION_PROCESS_INFO, data => {
+      this.serverIoProvider.io.emit(SocketActionTypes.CONNECTION_PROCESS_INFO, data)
+    })
+    this.storageService.on(StorageEvents.CERTIFICATES_STORED, (payload: SendCertificatesResponse) => {
+      this.logger.info(`Storage - ${StorageEvents.CERTIFICATES_STORED}`)
+      this.serverIoProvider.io.emit(SocketActionTypes.CERTIFICATES_STORED, payload)
+    })
     this.storageService.on(StorageEvents.COMMUNITY_UPDATED, (payload: Community) => {
       this.serverIoProvider.io.emit(SocketActionTypes.COMMUNITY_UPDATED, payload)
     })
-    this.storageService.on(StorageEvents.SEND_PUSH_NOTIFICATION, (payload: PushNotificationPayload) => {
-      this.serverIoProvider.io.emit(SocketActionTypes.PUSH_NOTIFICATION, payload)
-    })
     this.storageService.on(StorageEvents.CSRS_STORED, async (payload: { csrs: string[] }) => {
       this.logger.info(`Storage - ${StorageEvents.CSRS_STORED}`)
       const users = await getUsersFromCsrs(payload.csrs)

packages/backend/src/nest/ipfs-file-manager/big-files.long.spec.ts

@@ -33,7 +33,7 @@ describe('IpfsFileManagerService', () => {
     tmpDir = createTmpDir()
     filePath = new URL('./testUtils/large-file.txt', import.meta.url).pathname
     // Generate 2.1GB file
-    createArbitraryFile(filePath, BIG_FILE_SIZE)
+    await createArbitraryFile(filePath, BIG_FILE_SIZE)
     module = await Test.createTestingModule({
       imports: [TestModule, IpfsFileManagerModule, IpfsModule, SocketModule, Libp2pModule],
     }).compile()

packages/backend/src/nest/storage/base.store.ts

@@ -24,7 +24,7 @@ abstract class StoreBase<V, S extends KeyValueType<V> | EventsType<V>> extends E
     logger.info('Closed', this.getAddress())
   }
 
-  abstract init(): Promise<void>
+  abstract init(...args: any[]): Promise<void> | Promise<StoreBase<V, S>>
   abstract clean(): void
 }