From f786d464ff6d817cd98a29f3d7b3c5fcc2452614 Mon Sep 17 00:00:00 2001 From: Eric Doughty-Papassideris Date: Tue, 26 Mar 2024 12:06:22 +0100 Subject: [PATCH] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Moving=20re-index=20comman?= =?UTF-8?q?ds=20to=20a=20class=20based=20system=20(#438)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/cli/cmds/search_cmds/index_all.ts | 180 +++++++++++------- .../node/src/cli/lib/run_with_platform.ts | 2 +- 2 files changed, 107 insertions(+), 75 deletions(-) diff --git a/tdrive/backend/node/src/cli/cmds/search_cmds/index_all.ts b/tdrive/backend/node/src/cli/cmds/search_cmds/index_all.ts index efc2b774..8c7c3136 100644 --- a/tdrive/backend/node/src/cli/cmds/search_cmds/index_all.ts +++ b/tdrive/backend/node/src/cli/cmds/search_cmds/index_all.ts @@ -14,10 +14,10 @@ import Repository, { import { EntityTarget, SearchServiceAPI } from "../../../core/platform/services/search/api"; import CompanyUser, { TYPE as CompanyUserTYPE } from "../../../services/user/entities/company_user"; import runWithPlatform from "../../lib/run_with_platform"; +import SearchRepository from "src/core/platform/services/search/repository"; type Options = { spinner: ora.Ora; - repositoryName?: string; repairEntities: boolean; }; @@ -40,90 +40,125 @@ async function iterateOverRepoPages( } while (page.page_token); } -type RepositoryConstructor = (database: DatabaseServiceAPI) => Promise>; - -const makeRepoConstructor = - (table: string, entity: EntityTarget): RepositoryConstructor => - database => - database.getRepository(table, entity); - -/** The commandline name for the user repository. It has specific code for the repairEntities so is defined here. */ -const UsersRepoName = "users"; - -class SearchIndexAll { - database: DatabaseServiceAPI; - search: SearchServiceAPI; - - constructor(readonly platform: TdrivePlatform) { +abstract class ReindexerCLICommand { + protected readonly database: DatabaseServiceAPI; + protected readonly search: SearchServiceAPI; + constructor( + protected readonly platform: TdrivePlatform, + private readonly options: Options, + private readonly table: string, + private readonly entity: EntityTarget, + ) { this.database = this.platform.getProvider("database"); this.search = this.platform.getProvider("search"); } + private _dbRepo: Repository; + protected async dbRepository() { + return (this._dbRepo = + this._dbRepo || (await this.database.getRepository(this.table, this.entity))); + } + private _searchRepo: SearchRepository; + protected async searchRepository() { + return (this._searchRepo = + this._searchRepo || this.search.getRepository(this.table, this.entity)); + } - private static readonly supportedRepos = new Map>([ - [UsersRepoName, makeRepoConstructor(UserTYPE, User)], - ["documents", makeRepoConstructor(DriveFileTYPE, DriveFile)], - ]); - public static isRepoSupported = (repositoryName: string) => - this.supportedRepos.has(repositoryName); - public static getSupportedRepoNames = () => [...this.supportedRepos.keys()]; - private getRepository = (repositoryName: string) => - SearchIndexAll.supportedRepos.get(repositoryName)(this.database); + protected statusStart(info: string) { + this.options.spinner.start(`${this.table} > ${info}`); + } + protected statusSucceed(info: string) { + this.options.spinner.succeed(`${this.table} > ${info}`); + } + protected statusWarn(info: string) { + this.options.spinner.warn(`${this.table} > ${info}`); + } + protected statusFail(info: string) { + this.options.spinner.fail(`${this.table} > ${info}`); + } + protected statusInfo(info: string) { + this.options.spinner.info(`${this.table} > ${info}`); + } - private async repairEntitiesInUsers( - options: Options, - repository: Repository, - ): Promise { - // Complete user with companies in cache - options.spinner.start("Adding companies to cache of user"); + protected async repairEntities(): Promise { + this.statusWarn(`repairEntities > No repair action for ${this.table}`); + } + + protected async mapEntitiesToReIndexFromDBToSearch(entities: Entity[]): Promise { + return entities; + } + + protected async reindexFromDBToSearch(): Promise { + const repository = await this.dbRepository(); + this.statusStart("Start indexing..."); + let count = 0; + await iterateOverRepoPages(repository, async entities => { + entities = await this.mapEntitiesToReIndexFromDBToSearch(entities); + await this.search.upsert(entities); + count += entities.length; + this.statusStart(`Indexed ${count} ${this.table}...`); + }); + if (count === 0) this.statusWarn(`Index ${this.table} finished; but 0 items included`); + else this.statusSucceed(`${count} ${this.table} indexed`); + const giveFlushAChanceDurationMS = 10000; + this.statusStart(`Emptying flush (${giveFlushAChanceDurationMS / 1000}s)...`); + await waitTimeoutMS(giveFlushAChanceDurationMS); + this.statusSucceed("Done!"); + } + public async run(): Promise { + if (this.options.repairEntities) await this.repairEntities(); + await this.reindexFromDBToSearch(); + } +} + +/** Serves as an index of classes for the repos to reindex; to specialise bits of behaviour and what not */ +const RepositoryNameToCTOR = new Map< + string, + (platform: TdrivePlatform, options: Options) => ReindexerCLICommand +>(); + +class UserReindexerCLICommand extends ReindexerCLICommand { + constructor(platform: TdrivePlatform, options: Options) { + super(platform, options, UserTYPE, User); + } + + protected override async repairEntities(): Promise { + this.statusStart("repairEntities > Adding companies to cache of user"); const companiesUsersRepository = await this.database.getRepository( CompanyUserTYPE, CompanyUser, ); let count = 0; + const repository = await this.dbRepository(); await iterateOverRepoPages(repository, async entities => { for (const user of entities) { const companies = await companiesUsersRepository.find({ user_id: user.id }, {}, undefined); + const prevCache = JSON.stringify(user.cache); user.cache ||= { companies: [] }; user.cache.companies = companies.getEntities().map(company => company.group_id); - await repository.save(user, undefined); + const newCache = JSON.stringify(user.cache); + if (prevCache != newCache) await repository.save(user, undefined); } count += entities.length; - options.spinner.start(`Adding companies to cache of ${count} users...`); + this.statusStart(`repairEntities > Adding companies to cache of ${count} users...`); }); - options.spinner.succeed(`Added companies to cache of ${count} users`); - } - - private repairEntities(options: Options, repository: Repository): Promise { - switch (options.repositoryName) { - case UsersRepoName: - return this.repairEntitiesInUsers(options, repository); - default: - options.spinner.warn(`No repair action for repository ${options.repositoryName}`); - break; - } - } - - public async run(options: Options): Promise { - const repository = await this.getRepository(options.repositoryName); - - if (options.repairEntities) await this.repairEntities(options, repository); - - options.spinner.start(`Start indexing ${options.repositoryName}...`); - let count = 0; - await iterateOverRepoPages(repository, async entities => { - await this.search.upsert(entities); - count += entities.length; - options.spinner.start(`Indexed ${count} ${options.repositoryName}...`); - }); - if (count === 0) - options.spinner.warn(`Index ${options.repositoryName} finished; but 0 items included`); - else options.spinner.succeed(`${count} ${options.repositoryName} indexed`); - const giveFlushAChanceDurationMS = 10000; - options.spinner.start(`Emptying flush (${giveFlushAChanceDurationMS / 1000}s)...`); - await waitTimeoutMS(giveFlushAChanceDurationMS); - options.spinner.succeed("Done!"); + this.statusSucceed(`repairEntities > Added companies to cache of ${count} users`); } } +RepositoryNameToCTOR.set( + "users", + (platform, options) => new UserReindexerCLICommand(platform, options), +); + +class DocumentsReindexerCLICommand extends ReindexerCLICommand { + constructor(platform: TdrivePlatform, options: Options) { + super(platform, options, DriveFileTYPE, DriveFile); + } +} +RepositoryNameToCTOR.set( + "documents", + (platform, options) => new DocumentsReindexerCLICommand(platform, options), +); + const reindexingArgumentGroupTitle = "Re-indexing options"; const repositoryArgumentName = "repository"; const command: yargs.CommandModule = { @@ -133,7 +168,7 @@ const command: yargs.CommandModule = { [repositoryArgumentName]: { type: "string", description: "Repository to re-index.", - choices: SearchIndexAll.getSupportedRepoNames(), + choices: [...RepositoryNameToCTOR.keys()], demandOption: true, group: reindexingArgumentGroupTitle, }, @@ -158,18 +193,15 @@ const command: yargs.CommandModule = { } } - runWithPlatform("Re-index " + argv.repository, async ({ spinner, platform }) => { + runWithPlatform("Re-index", async ({ spinner, platform }) => { try { - const migrator = new SearchIndexAll(platform); - for (const repositoryName of eachOnlyOnce(repositories)) { - await migrator.run({ - repositoryName, + for (const repositoryName of eachOnlyOnce(repositories)) + await RepositoryNameToCTOR.get(repositoryName)(platform, { spinner, repairEntities: !!argv.repairEntities, - }); - } + }).run(); } catch (err) { - spinner.fail(`Error indexing: ${err.stack}`); + spinner.fail(err.stack || err); return 1; } }); diff --git a/tdrive/backend/node/src/cli/lib/run_with_platform.ts b/tdrive/backend/node/src/cli/lib/run_with_platform.ts index 0d497785..35059f30 100644 --- a/tdrive/backend/node/src/cli/lib/run_with_platform.ts +++ b/tdrive/backend/node/src/cli/lib/run_with_platform.ts @@ -22,7 +22,7 @@ export default async function runWithPlatform( platform: TdrivePlatform; }) => Promise | Promise, ) { - const spinner = ora({ prefixText: prefix + " -" }); + const spinner = ora({ prefixText: prefix + " >" }); spinner.start("Platform: starting..."); const platform = await tdrive.run(config.get("services")); await gr.doInit(platform);