Add Channel Metadata Validation

CHANGELOG.md

@@ -12,7 +12,6 @@
 * Don't send deletion message for private channels [#3273](https://github.com/TryQuiet/quiet/issues/3273)
 * Ensure notification registration waits for auth handshake [#3289](https://github.com/TryQuiet/quiet/issues/3289)
 * Fixed a race condition that can cause stale data to remain after leaving community [#3253](https://github.com/TryQuiet/quiet/issues/3253)
-* Temporarily disable private channels
 * Validate user ID on decrypted message matches the signature [#3334](https://github.com/TryQuiet/quiet/issues/3334)
 
 ### Chores
@@ -25,6 +24,7 @@
 * Include team ID and createdAt in message encryption and validate on consume [#3304](https://github.com/TryQuiet/quiet/issues/3304)
 * Require team ID on invite links, use team ID for all chain operations, and hash team name on sigchains [#3296](https://github.com/TryQuiet/quiet/issues/3296)
 * Use randomly generated Base58 usernames on sigchain [#3321](https://github.com/TryQuiet/quiet/issues/3321)
+* Tighten controls on channel metadata DB operations, move private channels to separate metadata DB [#3329](https://github.com/TryQuiet/quiet/issues/3329)
 
 ## [7.3.0]
 

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

@@ -4,6 +4,8 @@ import { ChannelStore } from '../storage/channels/channel.store'
 export interface ChannelRepo {
   store: ChannelStore
   eventsAttached: boolean
+  subscribed: boolean
+  subscriptionPromise?: Promise<void>
   public: boolean
 }
 

packages/backend/src/nest/libp2p/integration-tests/orbitdb-message-fanout.spec.ts

@@ -462,7 +462,7 @@ describe(`OrbitDB Syncing with ${N_PEERS} peers`, () => {
     expect(channel).toBeDefined()
     const getChannels = await channelsService.getChannels()
     expect(getChannels.length).toBe(2)
-    expect(getChannels[1].id).toBe(newChannel.id)
+    expect(getChannels.find(channel => channel.id === newChannel.id)).toBeDefined()
 
     // Send a message in the new channel
     const message = await factory.build<ChannelMessage>('ChannelMessage', {
@@ -494,7 +494,7 @@ describe(`OrbitDB Syncing with ${N_PEERS} peers`, () => {
       const channelsService = modules[i].get(ChannelsService)
       const channels = await channelsService.getChannels()
       expect(channels.length).toBe(1) // Only the first channel should be present
-      expect(channels[0].id).toBe(publicChannels[0].id)
+      expect(channels.find(channel => channel.id === publicChannels[0].id)).toBeDefined()
     }
   })
 

packages/backend/src/nest/storage/channels/channel.store.spec.ts

@@ -0,0 +1,76 @@
+import EventEmitter from 'node:events'
+import { jest } from '@jest/globals'
+
+import { ChannelStore } from './channel.store'
+import { StorageEvents } from '../storage.types'
+
+describe('ChannelStore', () => {
+  const makeLogger = () => ({
+    debug: jest.fn(),
+    error: jest.fn(),
+    info: jest.fn(),
+    trace: jest.fn(),
+    warn: jest.fn(),
+  })
+
+  async function* emptyIterator() {}
+
+  it('ignores update entries without a matching channel id', async () => {
+    const channelId = 'channel-1'
+    const storeEvents = new EventEmitter()
+    const messagesService = {
+      onConsume: jest.fn(),
+    }
+    const localDbService = {
+      getCurrentCommunity: jest.fn(async () => ({ id: 'community-1' })),
+    }
+    const auth = Object.assign(new EventEmitter(), {
+      team: { id: 'team-1' },
+    })
+    const channelStore = new ChannelStore(
+      {} as any,
+      localDbService as any,
+      messagesService as any,
+      {} as any,
+      {} as any,
+      auth as any,
+      {} as any,
+      {} as any
+    )
+    ;(channelStore as any).channelData = {
+      id: channelId,
+      name: 'general',
+      public: true,
+      teamId: 'team-1',
+    }
+    ;(channelStore as any).logger = makeLogger()
+    ;(channelStore as any)._messagesService = messagesService
+    ;(channelStore as any).store = {
+      events: storeEvents,
+      iterator: emptyIterator,
+      sync: {
+        start: jest.fn(async () => {}),
+      },
+    }
+
+    const messageIdsListener = jest.fn()
+    channelStore.on(StorageEvents.MESSAGE_IDS_STORED, messageIdsListener)
+    await channelStore.subscribe()
+    messageIdsListener.mockClear()
+    localDbService.getCurrentCommunity.mockClear()
+
+    storeEvents.emit('update', {
+      hash: 'non-message-entry',
+      payload: {
+        value: {
+          id: 'non-message-value',
+        },
+      },
+    })
+    await new Promise(resolve => setImmediate(resolve))
+
+    expect(messagesService.onConsume).not.toHaveBeenCalled()
+    expect(localDbService.getCurrentCommunity).not.toHaveBeenCalled()
+    expect(messageIdsListener).not.toHaveBeenCalled()
+  })
+})

packages/backend/src/nest/storage/channels/channel.store.ts

@@ -144,9 +144,9 @@ export class ChannelStore extends EventStoreBase<EncryptedMessage, ConsumedChann
     this.getStore().events.on('update', async (entry: LogEntry<EncryptedMessage>) => {
       const entryChannelId = entry.payload.value?.channelId
       // TODO: seperate event bus for each channel so we don't have to check this on every update
-      if (entryChannelId != null && entryChannelId !== this.channelData.id) {
+      if (entryChannelId !== this.channelData.id) {
         this.logger.debug(
-          `Ignoring database update for different channel`,
+          `Ignoring database update without matching channel`,
           entry.hash,
           entryChannelId,
           this.channelData.id
@@ -374,8 +374,20 @@ export class ChannelStore extends EventStoreBase<EncryptedMessage, ConsumedChann
    */
   public async close(): Promise<void> {
     this.logger.info(`Closing channel store`)
+    const store = this.store
+    if (store == null) {
+      this.logger.warn(`Store is already undefined, nothing to close`)
+      return
+    }
+
     await this.stopSync()
-    await this.getStore().close()
+    await store.close()
+    if (this.authListenerAttached) {
+      this.auth.removeListener(SigchainEvents.UPDATED, this.handleAuthUpdated)
+      this.authListenerAttached = false
+    }
+    this.store = undefined
+    this._subscribing = false
   }
 
   /**
@@ -393,6 +405,7 @@ export class ChannelStore extends EventStoreBase<EncryptedMessage, ConsumedChann
    */
   public async clean(): Promise<void> {
     this.logger.info(`Cleaning channel store`, this.channelData.id, this.channelData.name)
+    const store = this.store
     try {
       await this.stopSync()
     } catch (e) {
@@ -402,11 +415,16 @@ export class ChannelStore extends EventStoreBase<EncryptedMessage, ConsumedChann
       if (!this.store) {
         this.logger.warn(`Store is already undefined, nothing to drop`)
       } else {
-        await this.getStore().drop()
+        await store!.drop()
       }
     } catch (e) {
       this.logger.error(`Failed to drop store`, e)
     }
+    try {
+      await store?.close()
+    } catch (e) {
+      this.logger.error(`Failed to close store after drop`, e)
+    }
     if (this.authListenerAttached) {
       this.auth.removeListener(SigchainEvents.UPDATED, this.handleAuthUpdated)
       this.authListenerAttached = false