Merge from develop

.github/workflows/e2e-linux.yml

@@ -16,6 +16,7 @@ jobs:
       ELECTRON_CUSTOM_VERSION: 23.0.0
       DISPLAY: ":99.0"
       TEST_MODE: true
+      IS_CI: true
 
     steps:
       - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1

.github/workflows/e2e-mac.yml

@@ -9,6 +9,7 @@ jobs:
       ELECTRON_CUSTOM_VERSION: 23.0.0
       TEST_MODE: true
       IS_E2E: true
+      IS_CI: true
 
     steps:
       - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1

.github/workflows/e2e-win.yml

@@ -11,6 +11,7 @@ jobs:
       ELECTRON_CUSTOM_VERSION: 23.0.0
       TEST_MODE: true
       E2E: true
+      IS_CI: true
 
     steps:
       - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1

CHANGELOG.md

@@ -2,10 +2,18 @@
 
 # New features:
 
+* Adds connection status information to messages panel on desktop when no peers are connected ([#1706](https://github.com/TryQuiet/quiet/issues/1706)
+
 # Refactorings:
 
+* Logging from all sources can be written to node console
+
 # Fixes:
 
+# Chores
+
+* Cleanup data directory at end of e2e tests
+
 [2.2.0]
 
 # New features:
@@ -15,20 +23,18 @@
 # Refactorings:
 
 * Use ack for CREATE_NETWORK and simplify
-* Logging from all sources can be written to node console
 * Move Community model to the backend
 
 # Fixes:
 
 * Allow JPEG and GIF files as profile photos ([#2332](https://github.com/TryQuiet/quiet/issues/2332))
 * Fixes issues with recreating general channel when deleted while offline ([#2334](https://github.com/TryQuiet/quiet/issues/2334))
+* Fix issues with recreating general channel when deleted while offline ([#2334](https://github.com/TryQuiet/quiet/issues/2334))
+* Fix package.json license inconsistency
 
 # New Features
 
-* Adds connection status information to messages panel on desktop when no peers are connected ([#1706](https://github.com/TryQuiet/quiet/
 * Add utilities for emoji detection in messages and make all-emoji message larger font size ([#519](https://github.com/TryQuiet/quiet/issues/519))
-* Fix issues with recreating general channel when deleted while offline ([#2334](https://github.com/TryQuiet/quiet/issues/2334))
-* Fix package.json license inconsistency
 
 [2.1.2]
 

packages/backend/package.json

@@ -108,6 +108,7 @@
     "dotenv": "8.2.0",
     "events": "^3.2.0",
     "express": "^4.17.1",
+    "fastq": "^1.17.1",
     "get-port": "^5.1.1",
     "go-ipfs": "npm:[email protected]",
     "http-server": "^0.12.3",

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

@@ -31,6 +31,7 @@ import waitForExpect from 'wait-for-expect'
 import { Libp2pEvents } from '../libp2p/libp2p.types'
 import { sleep } from '../common/sleep'
 import { createLibp2pAddress } from '@quiet/common'
+import { lib } from 'crypto-js'
 
 jest.setTimeout(100_000)
 

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

@@ -519,7 +519,7 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI
       agent: this.socksProxyAgent,
       localAddress: this.libp2pService.createLibp2pAddress(onionAddress, peerId.toString()),
       targetPort: this.ports.libp2pHiddenService,
-      peers: peers ?? [],
+      peers: peers ? peers.slice(1) : [],
       psk: Libp2pService.generateLibp2pPSK(community.psk).fullKey,
     }
     await this.libp2pService.createInstance(params)

packages/backend/src/nest/libp2p/libp2p.service.spec.ts

@@ -8,7 +8,7 @@ import { Libp2pEvents, Libp2pNodeParams } from './libp2p.types'
 import { toString as uint8ArrayToString } from 'uint8arrays/to-string'
 import validator from 'validator'
 import waitForExpect from 'wait-for-expect'
-import { ProcessInChunksService } from './process-in-chunks.service'
+import { DEFAULT_NUM_TRIES, ProcessInChunksService } from './process-in-chunks.service'
 
 describe('Libp2pService', () => {
   let module: TestingModule
@@ -93,10 +93,12 @@ describe('Libp2pService', () => {
     await libp2pService.createInstance(params)
     expect(libp2pService.libp2pInstance).not.toBeNull()
     // @ts-expect-error processItem is private
-    const dialPeerSpy = jest.spyOn(processInChunks, 'processItem')
+    const processItemSpy = jest.spyOn(processInChunks, 'processItem')
+    const dialSpy = jest.spyOn(libp2pService.libp2pInstance!, 'dial')
     libp2pService.emit(Libp2pEvents.DIAL_PEERS, addresses)
     await waitForExpect(async () => {
-      expect(dialPeerSpy).toBeCalledTimes(1)
+      expect(processItemSpy).toBeCalledTimes(2 * DEFAULT_NUM_TRIES)
+      expect(dialSpy).toBeCalledTimes(1)
     })
   })
 })

packages/backend/src/nest/libp2p/process-in-chunks.service.ts

@@ -1,56 +1,85 @@
 import { EventEmitter } from 'events'
+import fastq from 'fastq'
+import type { queue, done } from 'fastq'
+
 import Logger from '../common/logger'
 
 const DEFAULT_CHUNK_SIZE = 10
+export const DEFAULT_NUM_TRIES = 2
+
+type ProcessTask<T> = {
+  data: T
+  tries: number
+}
 
 export class ProcessInChunksService<T> extends EventEmitter {
   private isActive: boolean
-  private data: T[]
+  private data: Set<T> = new Set()
   private chunkSize: number
+  private taskQueue: queue<ProcessTask<T>>
   private processItem: (arg: T) => Promise<any>
   private readonly logger = Logger(ProcessInChunksService.name)
   constructor() {
     super()
   }
 
   public init(data: T[], processItem: (arg: T) => Promise<any>, chunkSize: number = DEFAULT_CHUNK_SIZE) {
-    this.data = data
+    this.logger(`Initializing process-in-chunks.service with peers ${JSON.stringify(data, null, 2)}`)
     this.processItem = processItem
     this.chunkSize = chunkSize
+    this.taskQueue = fastq(this, this.processOneItem, this.chunkSize)
+    this.updateData(data)
+    this.addToTaskQueue()
   }
 
-  updateData(items: T[]) {
+  public updateData(items: T[]) {
     this.logger(`Updating data with ${items.length} items`)
-    this.data = [...new Set(this.data.concat(items))]
+    this.taskQueue.pause()
+    items.forEach(item => this.data.add(item))
+    this.addToTaskQueue()
   }
 
-  public async processOneItem() {
-    const toProcess = this.data.shift()
-    if (toProcess) {
-      try {
-        await this.processItem(toProcess)
-      } catch (e) {
-        this.logger(`Processing ${toProcess} failed, message:`, e.message)
-      } finally {
-        process.nextTick(async () => {
-          await this.processOneItem()
-        })
+  private addToTaskQueue() {
+    this.logger(`Adding ${this.data.size} items to the task queue`)
+    for (const item of this.data) {
+      if (item) {
+        this.logger(`Adding data ${item} to the task queue`)
+        this.data.delete(item)
+        try {
+          this.taskQueue.push({ data: item, tries: 0 } as ProcessTask<T>)
+        } catch (e) {
+          this.logger.error(`Error occurred while adding new task for item ${item} to the queue`, e)
+          this.data.add(item)
+        }
       }
     }
   }
 
-  public async process() {
-    this.logger(`Processing ${this.data.length} items`)
-    for (let i = 0; i < this.chunkSize; i++) {
-      // Do not wait for this promise as items should be processed simultineously
-      void this.processOneItem()
+  public async processOneItem(task: ProcessTask<T>) {
+    try {
+      this.logger(`Processing task with data ${task.data}`)
+      await this.processItem(task.data)
+    } catch (e) {
+      this.logger.error(`Processing task with data ${task.data} failed`, e)
+      if (task.tries + 1 < DEFAULT_NUM_TRIES) {
+        this.logger(`Will try to re-attempt task with data ${task.data}`)
+        this.taskQueue.push({ ...task, tries: task.tries + 1 })
+      }
+    } finally {
+      this.logger(`Done attempting to process task with data ${task.data}`)
     }
   }
 
+  public async process() {
+    this.logger(`Processing ${this.taskQueue.length} items`)
+    this.taskQueue.resume()
+  }
+
   public stop() {
     if (this.isActive) {
       this.logger('Stopping initial dial')
       this.isActive = false
+      this.taskQueue.pause()
     }
   }
 }

packages/backend/src/nest/libp2p/process-in-chunks.spec.ts

@@ -27,7 +27,7 @@ describe('ProcessInChunks', () => {
     processInChunks.init(['a', 'b', 'c', 'd'], mockProcessItem)
     await processInChunks.process()
     await waitForExpect(() => {
-      expect(mockProcessItem).toBeCalledTimes(4)
+      expect(mockProcessItem).toBeCalledTimes(6)
     })
   })
 
@@ -43,7 +43,7 @@ describe('ProcessInChunks', () => {
     processInChunks.updateData(['e', 'f'])
     await processInChunks.process()
     await waitForExpect(() => {
-      expect(mockProcessItem).toBeCalledTimes(4)
+      expect(mockProcessItem).toBeCalledTimes(5)
     })
   })
 
@@ -60,7 +60,7 @@ describe('ProcessInChunks', () => {
     processInChunks.init(['a', 'b', 'c', 'd'], mockProcessItem, chunkSize)
     await processInChunks.process()
     await waitForExpect(() => {
-      expect(mockProcessItem).toBeCalledTimes(4)
+      expect(mockProcessItem).toBeCalledTimes(2)
     })
   })
 

packages/backend/src/nest/local-db/local-db.service.ts

@@ -95,17 +95,31 @@ export class LocalDbService {
     }
   }
 
-  public async getSortedPeers(peers: string[] = []): Promise<string[]> {
+  public async getSortedPeers(
+    peers?: string[] | undefined,
+    includeLocalPeerAddress: boolean = true
+  ): Promise<string[]> {
+    if (!peers) {
+      const currentCommunity = await this.getCurrentCommunity()
+      if (!currentCommunity) {
+        throw new Error('No peers were provided and no community was found to extract peers from')
+      }
+      peers = currentCommunity.peerList
+      if (!peers) {
+        throw new Error('No peers provided and no peers found on current stored community')
+      }
+    }
+
     const peersStats = (await this.get(LocalDBKeys.PEERS)) || {}
     const stats: NetworkStats[] = Object.values(peersStats)
     const network = await this.getNetworkInfo()
 
     if (network) {
       const localPeerAddress = createLibp2pAddress(network.hiddenService.onionAddress, network.peerId.id)
       this.logger('Local peer', localPeerAddress)
-      return filterAndSortPeers(peers, stats, localPeerAddress)
+      return filterAndSortPeers(peers, stats, localPeerAddress, includeLocalPeerAddress)
     } else {
-      return filterAndSortPeers(peers, stats)
+      return filterAndSortPeers(peers, stats, undefined, includeLocalPeerAddress)
     }
   }
 

packages/common/src/dir.ts

@@ -0,0 +1,19 @@
+import path from 'path'
+import { DESKTOP_DATA_DIR, DESKTOP_DEV_DATA_DIR } from './static'
+
+export type GetDataAppPathDefaults = {
+  appDataPath?: string
+  dataDir?: string
+}
+
+export const getAppDataPath = (defaults: GetDataAppPathDefaults = {}): string => {
+  const defaultAppDataPath = defaults.appDataPath || process.env.APPDATA
+  const defaultDataDir = defaults.dataDir || process.env.DATA_DIR
+
+  const dataPath =
+    defaultAppDataPath ||
+    (process.platform === 'darwin' ? process.env.HOME + '/Library/Application Support' : process.env.HOME + '/.config')
+  const appPath = defaultDataDir || (process.env.NODE_ENV === 'development' ? DESKTOP_DEV_DATA_DIR : DESKTOP_DATA_DIR)
+
+  return path.join(dataPath, appPath)
+}

packages/common/src/index.ts

@@ -11,3 +11,4 @@ export * from './libp2p'
 export * from './tests'
 export * from './auth'
 export * from './messages'
+export * from './dir'