feat: add storing paging progress in redis

.pnp.cjs

@@ -6156,6 +6156,7 @@ const RAW_RUNTIME_STATE =
           ["@types/cors", "npm:2.8.13"],\
           ["@types/dotenv", "npm:8.2.0"],\
           ["@types/express", "npm:4.17.17"],\
+          ["@types/ioredis", "npm:5.0.0"],\
           ["@types/jest", "npm:29.5.2"],\
           ["@types/newrelic", "npm:9.14.0"],\
           ["@types/node", "npm:20.5.7"],\
@@ -6168,6 +6169,7 @@ const RAW_RUNTIME_STATE =
           ["express", "npm:4.18.2"],\
           ["inversify", "npm:6.0.1"],\
           ["inversify-express-utils", "npm:6.4.3"],\
+          ["ioredis", "npm:5.3.2"],\
           ["jest", "virtual:fd909b174d079e30b336c4ce72c38a88c1e447767b1a8dd7655e07719a1e31b97807f0931368724fc78897ff15e6a6d00b83316c0f76d11f85111f342e08bb79#npm:29.5.0"],\
           ["mongodb", "virtual:365b8c88cdf194291829ee28b79556e2328175d26a621363e703848100bea0042e9500db2a1206c9bbc3a4a76a1d169639ef774b2ea3a1a98584a9936b58c6be#npm:6.0.0"],\
           ["mysql2", "npm:3.3.3"],\
@@ -6340,6 +6342,7 @@ const RAW_RUNTIME_STATE =
           ["@types/cors", "npm:2.8.13"],\
           ["@types/dotenv", "npm:8.2.0"],\
           ["@types/express", "npm:4.17.17"],\
+          ["@types/ioredis", "npm:5.0.0"],\
           ["@types/jest", "npm:29.5.2"],\
           ["@types/jsonwebtoken", "npm:9.0.2"],\
           ["@types/newrelic", "npm:9.14.0"],\
@@ -6359,6 +6362,7 @@ const RAW_RUNTIME_STATE =
           ["helmet", "npm:7.0.0"],\
           ["inversify", "npm:6.0.1"],\
           ["inversify-express-utils", "npm:6.4.3"],\
+          ["ioredis", "npm:5.3.2"],\
           ["jest", "virtual:fd909b174d079e30b336c4ce72c38a88c1e447767b1a8dd7655e07719a1e31b97807f0931368724fc78897ff15e6a6d00b83316c0f76d11f85111f342e08bb79#npm:29.5.0"],\
           ["jsonwebtoken", "npm:9.0.0"],\
           ["mongodb", "virtual:365b8c88cdf194291829ee28b79556e2328175d26a621363e703848100bea0042e9500db2a1206c9bbc3a4a76a1d169639ef774b2ea3a1a98584a9936b58c6be#npm:6.0.0"],\
@@ -16772,7 +16776,7 @@ const RAW_RUNTIME_STATE =
           ["@types/better-sqlite3", null],\
           ["@types/google-cloud__spanner", null],\
           ["@types/hdb-pool", null],\
-          ["@types/ioredis", null],\
+          ["@types/ioredis", "npm:5.0.0"],\
           ["@types/mongodb", null],\
           ["@types/mssql", null],\
           ["@types/mysql2", null],\
@@ -16796,7 +16800,7 @@ const RAW_RUNTIME_STATE =
           ["dotenv", "npm:16.1.3"],\
           ["glob", "npm:8.1.0"],\
           ["hdb-pool", null],\
-          ["ioredis", null],\
+          ["ioredis", "npm:5.3.2"],\
           ["mkdirp", "npm:2.1.6"],\
           ["mongodb", "virtual:365b8c88cdf194291829ee28b79556e2328175d26a621363e703848100bea0042e9500db2a1206c9bbc3a4a76a1d169639ef774b2ea3a1a98584a9936b58c6be#npm:6.0.0"],\
           ["mssql", null],\

packages/revisions/package.json

@@ -41,6 +41,7 @@
     "express": "^4.18.2",
     "inversify": "^6.0.1",
     "inversify-express-utils": "^6.4.3",
+    "ioredis": "^5.3.2",
     "mongodb": "^6.0.0",
     "mysql2": "^3.0.1",
     "reflect-metadata": "0.1.13",
@@ -52,6 +53,7 @@
     "@types/cors": "^2.8.9",
     "@types/dotenv": "^8.2.0",
     "@types/express": "^4.17.14",
+    "@types/ioredis": "^5.0.0",
     "@types/jest": "^29.5.1",
     "@types/node": "^20.5.7",
     "@typescript-eslint/eslint-plugin": "^6.5.0",

packages/revisions/src/Bootstrap/Container.ts

@@ -1,4 +1,5 @@
 import { ControllerContainer, ControllerContainerInterface, MapperInterface } from '@standardnotes/domain-core'
+import Redis from 'ioredis'
 import { Container, interfaces } from 'inversify'
 import { MongoRepository, Repository } from 'typeorm'
 import * as winston from 'winston'
@@ -68,6 +69,8 @@ import { RemoveRevisionsFromSharedVault } from '../Domain/UseCase/RemoveRevision
 import { ItemRemovedFromSharedVaultEventHandler } from '../Domain/Handler/ItemRemovedFromSharedVaultEventHandler'
 import { TransitionRequestedEventHandler } from '../Domain/Handler/TransitionRequestedEventHandler'
 import { SharedVaultRemovedEventHandler } from '../Domain/Handler/SharedVaultRemovedEventHandler'
+import { TransitionRepositoryInterface } from '../Domain/Transition/TransitionRepositoryInterface'
+import { RedisTransitionRepository } from '../Infra/Redis/RedisTransitionRepository'
 
 export class ContainerConfigLoader {
   constructor(private mode: 'server' | 'worker' = 'server') {}
@@ -88,11 +91,28 @@ export class ContainerConfigLoader {
     const isConfiguredForSelfHosting = env.get('MODE', true) === 'self-hosted'
     const isConfiguredForHomeServerOrSelfHosting = isConfiguredForHomeServer || isConfiguredForSelfHosting
     const isSecondaryDatabaseEnabled = env.get('SECONDARY_DB_ENABLED', true) === 'true'
+    const isConfiguredForInMemoryCache = env.get('CACHE_TYPE', true) === 'memory'
 
     const container = new Container({
       defaultScope: 'Singleton',
     })
 
+    if (!isConfiguredForInMemoryCache) {
+      const redisUrl = env.get('REDIS_URL')
+      const isRedisInClusterMode = redisUrl.indexOf(',') > 0
+      let redis
+      if (isRedisInClusterMode) {
+        redis = new Redis.Cluster(redisUrl.split(','))
+      } else {
+        redis = new Redis(redisUrl)
+      }
+
+      container.bind(TYPES.Revisions_Redis).toConstantValue(redis)
+      container
+        .bind<TransitionRepositoryInterface>(TYPES.Revisions_TransitionStatusRepository)
+        .toConstantValue(new RedisTransitionRepository(container.get<Redis>(TYPES.Revisions_Redis)))
+    }
+
     let logger: winston.Logger
     if (configuration?.logger) {
       logger = configuration.logger as winston.Logger
@@ -348,6 +368,9 @@ export class ContainerConfigLoader {
           isSecondaryDatabaseEnabled
             ? container.get<RevisionRepositoryInterface>(TYPES.Revisions_MongoDBRevisionRepository)
             : null,
+          isConfiguredForInMemoryCache
+            ? null
+            : container.get<TransitionRepositoryInterface>(TYPES.Revisions_TransitionStatusRepository),
           container.get<TimerInterface>(TYPES.Revisions_Timer),
           container.get<winston.Logger>(TYPES.Revisions_Logger),
           env.get('MIGRATION_BATCH_SIZE', true) ? +env.get('MIGRATION_BATCH_SIZE', true) : 100,

packages/revisions/src/Bootstrap/Types.ts

@@ -1,6 +1,7 @@
 const TYPES = {
   Revisions_DBConnection: Symbol.for('Revisions_DBConnection'),
   Revisions_Logger: Symbol.for('Revisions_Logger'),
+  Revisions_Redis: Symbol.for('Revisions_Redis'),
   Revisions_SQS: Symbol.for('Revisions_SQS'),
   Revisions_SNS: Symbol.for('Revisions_SNS'),
   Revisions_S3: Symbol.for('Revisions_S3'),
@@ -27,6 +28,7 @@ const TYPES = {
   Revisions_MongoDBRevisionRepository: Symbol.for('Revisions_MongoDBRevisionRepository'),
   Revisions_DumpRepository: Symbol.for('Revisions_DumpRepository'),
   Revisions_RevisionRepositoryResolver: Symbol.for('Revisions_RevisionRepositoryResolver'),
+  Revisions_TransitionStatusRepository: Symbol.for('Revisions_TransitionStatusRepository'),
   // env vars
   Revisions_AUTH_JWT_SECRET: Symbol.for('Revisions_AUTH_JWT_SECRET'),
   Revisions_SQS_QUEUE_URL: Symbol.for('Revisions_SQS_QUEUE_URL'),

packages/revisions/src/Domain/Transition/TransitionRepositoryInterface.ts

@@ -0,0 +1,4 @@
+export interface TransitionRepositoryInterface {
+  getPagingProgress(userUuid: string): Promise<number>
+  setPagingProgress(userUuid: string, progress: number): Promise<void>
+}

packages/revisions/src/Domain/UseCase/Transition/TransitionRevisionsFromPrimaryToSecondaryDatabaseForUser/TransitionRevisionsFromPrimaryToSecondaryDatabaseForUser.ts

@@ -5,13 +5,13 @@ import { Logger } from 'winston'
 
 import { TransitionRevisionsFromPrimaryToSecondaryDatabaseForUserDTO } from './TransitionRevisionsFromPrimaryToSecondaryDatabaseForUserDTO'
 import { RevisionRepositoryInterface } from '../../../Revision/RevisionRepositoryInterface'
+import { TransitionRepositoryInterface } from '../../../Transition/TransitionRepositoryInterface'
 
 export class TransitionRevisionsFromPrimaryToSecondaryDatabaseForUser implements UseCaseInterface<void> {
-  private readonly pagingProgress: Map<string, number> = new Map()
-
   constructor(
     private primaryRevisionsRepository: RevisionRepositoryInterface,
     private secondRevisionsRepository: RevisionRepositoryInterface | null,
+    private transitionStatusRepository: TransitionRepositoryInterface | null,
     private timer: TimerInterface,
     private logger: Logger,
     private pageSize: number,
@@ -24,6 +24,10 @@ export class TransitionRevisionsFromPrimaryToSecondaryDatabaseForUser implements
       return Result.fail('Secondary revision repository is not set')
     }
 
+    if (this.transitionStatusRepository === null) {
+      return Result.fail('Transition status repository is not set')
+    }
+
     const userUuidOrError = Uuid.create(dto.userUuid)
     if (userUuidOrError.isFailed()) {
       return Result.fail(userUuidOrError.getError())
@@ -73,10 +77,9 @@ export class TransitionRevisionsFromPrimaryToSecondaryDatabaseForUser implements
 
   private async migrateRevisionsForUser(userUuid: Uuid): Promise<Result<string[]>> {
     try {
-      if (!this.pagingProgress.has(userUuid.value)) {
-        this.pagingProgress.set(userUuid.value, 1)
-      }
-      const initialPage = this.pagingProgress.get(userUuid.value) as number
+      const initialPage = await (this.transitionStatusRepository as TransitionRepositoryInterface).getPagingProgress(
+        userUuid.value,
+      )
 
       this.logger.info(`[${userUuid.value}] Migrating from page ${initialPage}`)
 
@@ -84,7 +87,10 @@ export class TransitionRevisionsFromPrimaryToSecondaryDatabaseForUser implements
       const totalPages = Math.ceil(totalRevisionsCountForUser / this.pageSize)
       const revisionsToSkipInIntegrityCheck = []
       for (let currentPage = initialPage; currentPage <= totalPages; currentPage++) {
-        this.pagingProgress.set(userUuid.value, currentPage)
+        await (this.transitionStatusRepository as TransitionRepositoryInterface).setPagingProgress(
+          userUuid.value,
+          currentPage,
+        )
 
         const query = {
           userUuid: userUuid,

packages/revisions/src/Infra/Redis/RedisTransitionRepository.ts

@@ -0,0 +1,23 @@
+import * as IORedis from 'ioredis'
+
+import { TransitionRepositoryInterface } from '../../Domain/Transition/TransitionRepositoryInterface'
+
+export class RedisTransitionRepository implements TransitionRepositoryInterface {
+  private readonly PREFIX = 'transition-revisions-paging-progress'
+
+  constructor(private redisClient: IORedis.Redis) {}
+
+  async getPagingProgress(userUuid: string): Promise<number> {
+    const progress = await this.redisClient.get(`${this.PREFIX}:${userUuid}`)
+
+    if (progress === null) {
+      return 1
+    }
+
+    return parseInt(progress)
+  }
+
+  async setPagingProgress(userUuid: string, progress: number): Promise<void> {
+    await this.redisClient.setex(`${this.PREFIX}:${userUuid}`, 172_800, progress.toString())
+  }
+}

packages/syncing-server/package.json

@@ -47,6 +47,7 @@
     "helmet": "^7.0.0",
     "inversify": "^6.0.1",
     "inversify-express-utils": "^6.4.3",
+    "ioredis": "^5.3.2",
     "jsonwebtoken": "^9.0.0",
     "mongodb": "^6.0.0",
     "mysql2": "^3.0.1",
@@ -63,6 +64,7 @@
     "@types/cors": "^2.8.9",
     "@types/dotenv": "^8.2.0",
     "@types/express": "^4.17.14",
+    "@types/ioredis": "^5.0.0",
     "@types/jest": "^29.5.1",
     "@types/jsonwebtoken": "^9.0.1",
     "@types/node": "^20.5.7",

packages/syncing-server/src/Bootstrap/Container.ts

@@ -1,4 +1,5 @@
 import * as winston from 'winston'
+import Redis from 'ioredis'
 import { Container, interfaces } from 'inversify'
 
 import { Env } from './Env'
@@ -171,6 +172,8 @@ import { SharedVaultRemovedEventHandler } from '../Domain/Handler/SharedVaultRem
 import { DesignateSurvivor } from '../Domain/UseCase/SharedVaults/DesignateSurvivor/DesignateSurvivor'
 import { RemoveUserFromSharedVaults } from '../Domain/UseCase/SharedVaults/RemoveUserFromSharedVaults/RemoveUserFromSharedVaults'
 import { TransferSharedVault } from '../Domain/UseCase/SharedVaults/TransferSharedVault/TransferSharedVault'
+import { TransitionRepositoryInterface } from '../Domain/Transition/TransitionRepositoryInterface'
+import { RedisTransitionRepository } from '../Infra/Redis/RedisTransitionRepository'
 
 export class ContainerConfigLoader {
   private readonly DEFAULT_CONTENT_SIZE_TRANSFER_LIMIT = 10_000_000
@@ -228,6 +231,23 @@ export class ContainerConfigLoader {
     const isConfiguredForSelfHosting = env.get('MODE', true) === 'self-hosted'
     const isConfiguredForHomeServerOrSelfHosting = isConfiguredForHomeServer || isConfiguredForSelfHosting
     const isSecondaryDatabaseEnabled = env.get('SECONDARY_DB_ENABLED', true) === 'true'
+    const isConfiguredForInMemoryCache = env.get('CACHE_TYPE', true) === 'memory'
+
+    if (!isConfiguredForInMemoryCache) {
+      const redisUrl = env.get('REDIS_URL')
+      const isRedisInClusterMode = redisUrl.indexOf(',') > 0
+      let redis
+      if (isRedisInClusterMode) {
+        redis = new Redis.Cluster(redisUrl.split(','))
+      } else {
+        redis = new Redis(redisUrl)
+      }
+
+      container.bind(TYPES.Sync_Redis).toConstantValue(redis)
+      container
+        .bind<TransitionRepositoryInterface>(TYPES.Sync_TransitionStatusRepository)
+        .toConstantValue(new RedisTransitionRepository(container.get<Redis>(TYPES.Sync_Redis)))
+    }
 
     container.bind<Env>(TYPES.Sync_Env).toConstantValue(env)
 
@@ -833,6 +853,9 @@ export class ContainerConfigLoader {
         new TransitionItemsFromPrimaryToSecondaryDatabaseForUser(
           container.get<ItemRepositoryInterface>(TYPES.Sync_SQLItemRepository),
           isSecondaryDatabaseEnabled ? container.get<ItemRepositoryInterface>(TYPES.Sync_MongoDBItemRepository) : null,
+          isConfiguredForInMemoryCache
+            ? null
+            : container.get<TransitionRepositoryInterface>(TYPES.Sync_TransitionStatusRepository),
           container.get<TimerInterface>(TYPES.Sync_Timer),
           container.get<Logger>(TYPES.Sync_Logger),
           env.get('MIGRATION_BATCH_SIZE', true) ? +env.get('MIGRATION_BATCH_SIZE', true) : 100,

packages/syncing-server/src/Bootstrap/Types.ts

@@ -15,6 +15,7 @@ const TYPES = {
   Sync_SharedVaultUserRepository: Symbol.for('Sync_SharedVaultUserRepository'),
   Sync_NotificationRepository: Symbol.for('Sync_NotificationRepository'),
   Sync_MessageRepository: Symbol.for('Sync_MessageRepository'),
+  Sync_TransitionStatusRepository: Symbol.for('Sync_TransitionStatusRepository'),
   // ORM
   Sync_ORMItemRepository: Symbol.for('Sync_ORMItemRepository'),
   Sync_ORMLegacyItemRepository: Symbol.for('Sync_ORMLegacyItemRepository'),

packages/syncing-server/src/Domain/Transition/TransitionRepositoryInterface.ts

@@ -0,0 +1,4 @@
+export interface TransitionRepositoryInterface {
+  getPagingProgress(userUuid: string): Promise<number>
+  setPagingProgress(userUuid: string, progress: number): Promise<void>
+}

packages/syncing-server/src/Domain/UseCase/Transition/TransitionItemsFromPrimaryToSecondaryDatabaseForUser/TransitionItemsFromPrimaryToSecondaryDatabaseForUser.ts

@@ -6,13 +6,13 @@ import { Logger } from 'winston'
 import { TransitionItemsFromPrimaryToSecondaryDatabaseForUserDTO } from './TransitionItemsFromPrimaryToSecondaryDatabaseForUserDTO'
 import { ItemRepositoryInterface } from '../../../Item/ItemRepositoryInterface'
 import { ItemQuery } from '../../../Item/ItemQuery'
+import { TransitionRepositoryInterface } from '../../../Transition/TransitionRepositoryInterface'
 
 export class TransitionItemsFromPrimaryToSecondaryDatabaseForUser implements UseCaseInterface<void> {
-  private readonly pagingProgress: Map<string, number> = new Map()
-
   constructor(
     private primaryItemRepository: ItemRepositoryInterface,
     private secondaryItemRepository: ItemRepositoryInterface | null,
+    private transitionStatusRepository: TransitionRepositoryInterface | null,
     private timer: TimerInterface,
     private logger: Logger,
     private pageSize: number,
@@ -25,6 +25,10 @@ export class TransitionItemsFromPrimaryToSecondaryDatabaseForUser implements Use
       return Result.fail('Secondary item repository is not set')
     }
 
+    if (this.transitionStatusRepository === null) {
+      return Result.fail('Transition status repository is not set')
+    }
+
     const userUuidOrError = Uuid.create(dto.userUuid)
     if (userUuidOrError.isFailed()) {
       return Result.fail(userUuidOrError.getError())
@@ -79,10 +83,9 @@ export class TransitionItemsFromPrimaryToSecondaryDatabaseForUser implements Use
 
   private async migrateItemsForUser(userUuid: Uuid): Promise<Result<string[]>> {
     try {
-      if (!this.pagingProgress.has(userUuid.value)) {
-        this.pagingProgress.set(userUuid.value, 1)
-      }
-      const initialPage = this.pagingProgress.get(userUuid.value) as number
+      const initialPage = await (this.transitionStatusRepository as TransitionRepositoryInterface).getPagingProgress(
+        userUuid.value,
+      )
 
       this.logger.info(`[${userUuid.value}] Migrating from page ${initialPage}`)
 
@@ -90,7 +93,10 @@ export class TransitionItemsFromPrimaryToSecondaryDatabaseForUser implements Use
       const totalPages = Math.ceil(totalItemsCountForUser / this.pageSize)
       const itemsToSkipInIntegrityCheck = []
       for (let currentPage = initialPage; currentPage <= totalPages; currentPage++) {
-        this.pagingProgress.set(userUuid.value, currentPage)
+        await (this.transitionStatusRepository as TransitionRepositoryInterface).setPagingProgress(
+          userUuid.value,
+          currentPage,
+        )
 
         const query: ItemQuery = {
           userUuid: userUuid.value,

packages/syncing-server/src/Infra/Redis/RedisTransitionRepository.ts

@@ -0,0 +1,23 @@
+import * as IORedis from 'ioredis'
+
+import { TransitionRepositoryInterface } from '../../Domain/Transition/TransitionRepositoryInterface'
+
+export class RedisTransitionRepository implements TransitionRepositoryInterface {
+  private readonly PREFIX = 'transition-items-paging-progress'
+
+  constructor(private redisClient: IORedis.Redis) {}
+
+  async getPagingProgress(userUuid: string): Promise<number> {
+    const progress = await this.redisClient.get(`${this.PREFIX}:${userUuid}`)
+
+    if (progress === null) {
+      return 1
+    }
+
+    return parseInt(progress)
+  }
+
+  async setPagingProgress(userUuid: string, progress: number): Promise<void> {
+    await this.redisClient.setex(`${this.PREFIX}:${userUuid}`, 172_800, progress.toString())
+  }
+}

yarn.lock

@@ -5045,6 +5045,7 @@ __metadata:
     "@types/cors": "npm:^2.8.9"
     "@types/dotenv": "npm:^8.2.0"
     "@types/express": "npm:^4.17.14"
+    "@types/ioredis": "npm:^5.0.0"
     "@types/jest": "npm:^29.5.1"
     "@types/newrelic": "npm:^9.14.0"
     "@types/node": "npm:^20.5.7"
@@ -5057,6 +5058,7 @@ __metadata:
     express: "npm:^4.18.2"
     inversify: "npm:^6.0.1"
     inversify-express-utils: "npm:^6.4.3"
+    ioredis: "npm:^5.3.2"
     jest: "npm:^29.5.0"
     mongodb: "npm:^6.0.0"
     mysql2: "npm:^3.0.1"
@@ -5235,6 +5237,7 @@ __metadata:
     "@types/cors": "npm:^2.8.9"
     "@types/dotenv": "npm:^8.2.0"
     "@types/express": "npm:^4.17.14"
+    "@types/ioredis": "npm:^5.0.0"
     "@types/jest": "npm:^29.5.1"
     "@types/jsonwebtoken": "npm:^9.0.1"
     "@types/newrelic": "npm:^9.14.0"
@@ -5254,6 +5257,7 @@ __metadata:
     helmet: "npm:^7.0.0"
     inversify: "npm:^6.0.1"
     inversify-express-utils: "npm:^6.4.3"
+    ioredis: "npm:^5.3.2"
     jest: "npm:^29.5.0"
     jsonwebtoken: "npm:^9.0.0"
     mongodb: "npm:^6.0.0"
@@ -9531,7 +9535,7 @@ __metadata:
   languageName: node
   linkType: hard
 
-"ioredis@npm:*, ioredis@npm:^5.2.4":
+"ioredis@npm:*, ioredis@npm:^5.2.4, ioredis@npm:^5.3.2":
   version: 5.3.2
   resolution: "ioredis@npm:5.3.2"
   dependencies: