♻️ Factor out platform startup/shutdown, make reindexable more obviously extensible (#438)

tdrive/backend/node/src/cli/cmds/search_cmds/index_all.ts

@@ -1,24 +1,32 @@
 import yargs from "yargs";
-import tdrive from "../../../tdrive";
 import ora from "ora";
 
-import config from "../../../core/config";
 import { TdrivePlatform } from "../../../core/platform/platform";
 import { DatabaseServiceAPI } from "../../../core/platform/services/database/api";
 import { Pagination } from "../../../core/platform/framework/api/crud-service";
 
 import User, { TYPE as UserTYPE } from "../../../services/user/entities/user";
 import Repository from "../../../core/platform/services/database/services/orm/repository/repository";
-import { SearchServiceAPI } from "../../../core/platform/services/search/api";
+import { EntityTarget, SearchServiceAPI } from "../../../core/platform/services/search/api";
 import CompanyUser, { TYPE as CompanyUserTYPE } from "../../../services/user/entities/company_user";
-import gr from "../../../services/global-resolver";
+import runWithPlatform from "../../lib/run_with_platform";
 
 type Options = {
   spinner: ora.Ora;
-  repository?: string;
-  repairEntities?: boolean;
+  repositoryName?: string;
+  repairEntities: boolean;
 };
 
+type RepositoryConstructor<Entity> = (database: DatabaseServiceAPI) => Promise<Repository<Entity>>;
+
+const makeRepoConstructor =
+  <Entity>(table: string, entity: EntityTarget<Entity>): RepositoryConstructor<Entity> =>
+  database =>
+    database.getRepository<Entity>(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;
@@ -28,47 +36,60 @@ class SearchIndexAll {
     this.search = this.platform.getProvider<SearchServiceAPI>("search");
   }
 
-  public async run(options: Options): Promise<void> {
-    const repositories: Map<string, Repository<any>> = new Map();
-    repositories.set("users", await this.database.getRepository(UserTYPE, User));
-
-    const repository = repositories.get(options.repository);
-    if (!repository) {
-      throw (
-        "No repository set (or valid) ready for indexation, available are: " +
-        Array.from(repositories.keys()).join(", ")
-      );
-    }
+  private static readonly supportedRepos = new Map<string, RepositoryConstructor<any>>([
+    [UsersRepoName, makeRepoConstructor(UserTYPE, User)],
+  ]);
+  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);
 
+  private async repairEntitiesInUsers(
+    options: Options,
+    repository: Repository<User>,
+  ): Promise<void> {
     // Complete user with companies in cache
-    if (options.repository === "users" && options.repairEntities) {
-      options.spinner.info("Complete user with companies in cache");
-      const companiesUsersRepository = await this.database.getRepository(
-        CompanyUserTYPE,
-        CompanyUser,
-      );
-      const userRepository = await this.database.getRepository(UserTYPE, User);
-      let page: Pagination = { limitStr: "100" };
-      // For each rows
-      do {
-        const list = await userRepository.find({}, { pagination: page }, undefined);
+    options.spinner.start("Complete user with companies in cache");
+    const companiesUsersRepository = await this.database.getRepository(
+      CompanyUserTYPE,
+      CompanyUser,
+    );
+    let page: Pagination = { limitStr: "100" };
+    let count = 0;
+    do {
+      const list = await repository.find({}, { pagination: page }, undefined);
 
-        for (const user of list.getEntities()) {
-          const companies = await companiesUsersRepository.find(
-            { user_id: user.id },
-            {},
-            undefined,
-          );
+      for (const user of list.getEntities()) {
+        const companies = await companiesUsersRepository.find({ user_id: user.id }, {}, undefined);
 
-          user.cache ||= { companies: [] };
-          user.cache.companies = companies.getEntities().map(company => company.group_id);
-          await repositories.get("users").save(user, undefined);
-        }
+        user.cache ||= { companies: [] };
+        user.cache.companies = companies.getEntities().map(company => company.group_id);
+        await repository.save(user, undefined);
+      }
+      count += list.getEntities().length;
+      options.spinner.start(`Completed ${count} users with companies in cache...`);
 
-        page = list.nextPage as Pagination;
-        await new Promise(r => setTimeout(r, 200));
-      } while (page.page_token);
+      page = list.nextPage as Pagination;
+      await new Promise(r => setTimeout(r, 2000));
+    } while (page.page_token);
+    options.spinner.succeed(`Completed ${count} users with companies in cache`);
+  }
+
+  private repairEntities(options: Options, repository: Repository<any>): Promise<void> {
+    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<void> {
+    const repository = await this.getRepository(options.repositoryName);
+
+    if (options.repairEntities) await this.repairEntities(options, repository);
 
     options.spinner.start("Start indexing...");
     let count = 0;
@@ -82,44 +103,54 @@ class SearchIndexAll {
       options.spinner.start("Indexed " + count + " items...");
       await new Promise(r => setTimeout(r, 200));
     } while (page.page_token);
-
-    options.spinner.succeed(`${count} items indexed`);
+    if (count === 0) options.spinner.warn("Index finieshed; but 0 items included");
+    else options.spinner.succeed(`${count} items indexed`);
     options.spinner.start("Emptying flush (10s)...");
     await new Promise(r => setTimeout(r, 10000));
     options.spinner.succeed("Done!");
   }
 }
 
+const repositoryArgumentName = "repository";
 const command: yargs.CommandModule<unknown, unknown> = {
   command: "index",
   describe: "command to reindex search middleware from db entities",
   builder: {
-    repository: {
+    [repositoryArgumentName]: {
       type: "string",
-      description: "Choose a repository to reindex",
+      description: `Repository name: ${SearchIndexAll.getSupportedRepoNames().join(", ")}`,
     },
     repairEntities: {
       default: false,
       type: "boolean",
-      description: "Choose to repair entities too when possible",
+      description: "Repair entities too when possible",
     },
   },
   handler: async argv => {
-    const spinner = ora({ text: "Reindex repository - " }).start();
-    const platform = await tdrive.run(config.get("services"));
-    await gr.doInit(platform);
-    const migrator = new SearchIndexAll(platform);
-    const repository = (argv.repository || "") as string;
+    const repositoryName = (argv[repositoryArgumentName] || "") as string;
+    if (!(repositoryName && SearchIndexAll.isRepoSupported(repositoryName)))
+      throw new Error(
+        `${
+          repositoryName ? `Invalid (${JSON.stringify(repositoryName)})` : "Missing"
+        } repository.\n` +
+          `\tSet with --${repositoryArgumentName} .\n` +
+          `\tPossible values: ${SearchIndexAll.getSupportedRepoNames().join(", ")}.`,
+      );
 
-    // Let this run even without a repository as its error message includes valid repository names
-    await migrator.run({
-      repository,
-      spinner,
+    runWithPlatform("Re-index " + argv.repository, async ({ spinner, platform }) => {
+      try {
+        const migrator = new SearchIndexAll(platform);
+        await migrator.run({
+          repositoryName,
+          spinner,
+          repairEntities: !!argv.repairEntities,
+        });
+        return 0;
+      } catch (err) {
+        spinner.fail(`Error indexing: ${err.stack}`);
+        return 1;
+      }
     });
-    spinner.start("Shutting down platform...");
-    await platform.stop();
-    spinner.succeed("Platform shutdown");
-    spinner.stop();
   },
 };
 

tdrive/backend/node/src/cli/lib/run_with_platform.ts

@@ -0,0 +1,36 @@
+import ora from "ora";
+
+import config from "../../core/config";
+import tdrive from "../../tdrive";
+import gr from "../../services/global-resolver";
+import type { TdrivePlatform } from "../../core/platform/platform";
+
+/**
+ * Start the platform and its services, run the command (passed as
+ * the `handler` callback), then cleanly shut down the platform.
+ * @param prefix Prefix text to set on the Ora spinner
+ * @param handler Callback to run the actual command.
+ *    - If it returns a number, that will be the exit code of the process.
+ *    - No attempt to catch errors is done here. If it throws, the platform
+ *      shutdown will be skipped and normal unhandled exception stuff will go on.
+ */
+export default async function runWithPlatform(
+  prefix: string,
+  handler: (args: {
+    spinner: ora.Ora;
+    config: config.IConfig;
+    platform: TdrivePlatform;
+  }) => Promise<number | undefined> | Promise<void>,
+) {
+  const spinner = ora({ prefixText: prefix + " -" });
+  spinner.start("Platform: starting...");
+  const platform = await tdrive.run(config.get("services"));
+  await gr.doInit(platform);
+  spinner.succeed("Platform: started");
+  const exitCode = await handler({ spinner, config, platform });
+  if (typeof exitCode === "number") process.exitCode = exitCode;
+  spinner.start("Platform: shutting down...");
+  await platform.stop();
+  spinner.succeed("Platform: shutdown");
+  spinner.stop();
+}