From 5766d2684bf6b20801ef3e2ab781a39c6c03c750 Mon Sep 17 00:00:00 2001
From: Eric Doughty-Papassideris <edoughtypapassideris@linagora.com>
Date: Tue, 23 Apr 2024 14:45:50 +0200
Subject: [PATCH] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20=20cli:=20buffering=20rein?=
 =?UTF-8?q?dexing=20of=20documents?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../index-all/documents-reindexer.ts          |  5 ++-
 .../node/src/cli/utils/buffered-action.ts     | 43 +++++++++++++++++++
 2 files changed, 47 insertions(+), 1 deletion(-)
 create mode 100644 tdrive/backend/node/src/cli/utils/buffered-action.ts

diff --git a/tdrive/backend/node/src/cli/cmds/search_cmds/index-all/documents-reindexer.ts b/tdrive/backend/node/src/cli/cmds/search_cmds/index-all/documents-reindexer.ts
index 035379b5..579ccf42 100644
--- a/tdrive/backend/node/src/cli/cmds/search_cmds/index-all/documents-reindexer.ts
+++ b/tdrive/backend/node/src/cli/cmds/search_cmds/index-all/documents-reindexer.ts
@@ -13,6 +13,7 @@ import iterateOverRepoPages from "../../../utils/iterate-over-repository-pages";
 
 import BaseReindexer from "./base-reindexer";
 import type ReindexerOptions from "./reindexer-options";
+import BufferedAction from "../../../utils/buffered-action";
 
 export default class DocumentsReindexer extends BaseReindexer<DriveFile> {
   constructor(platform: TdrivePlatform, options: ReindexerOptions) {
@@ -62,6 +63,7 @@ export default class DocumentsReindexer extends BaseReindexer<DriveFile> {
     const repository = await this.dbRepository();
     let count = 0,
       countChanged = 0;
+    const bufferedWriter = new BufferedAction<DriveFile>(100, items => repository.saveAll(items));
     await iterateOverRepoPages(
       repository,
       async entities => {
@@ -99,12 +101,13 @@ export default class DocumentsReindexer extends BaseReindexer<DriveFile> {
               );
             if (content_keywords !== entity.content_keywords) {
               entity.content_keywords = content_keywords;
-              repository.save(entity);
+              bufferedWriter.do(entity);
               countChanged++;
             }
           } catch (err) {
             this.statusFail(err.stack);
           }
+          bufferedWriter.flush();
         }
         this.statusStart(`Repaired ${count} documents (${countChanged} changed)`);
       },
diff --git a/tdrive/backend/node/src/cli/utils/buffered-action.ts b/tdrive/backend/node/src/cli/utils/buffered-action.ts
new file mode 100644
index 00000000..60e96d49
--- /dev/null
+++ b/tdrive/backend/node/src/cli/utils/buffered-action.ts
@@ -0,0 +1,43 @@
+/** Batch operations on objects into groups of `minimumBatchSize`
+ * provided to the constructor's callback.
+ *
+ * Call `do` or `doAll` to enqueue items, don't forget to:
+ * - call flush after all items are enqueued
+ * - await all the methods (do* and flush)
+ *
+ * The callback can be called with more than `minimumBatchSize`
+ * when `doAll` is used - see note on that method.
+ */
+export default class BufferedAction<T> {
+  // TODO: low and high marks
+  private buffer: T[] = [];
+  constructor(
+    private readonly minimumBatchSize: number,
+    private readonly action: (items: T[]) => Promise<void>,
+  ) {}
+  /** Call to empty all items by calling the callback with all items if at least one is queued */
+  public async flush() {
+    const buffer = this.buffer;
+    if (!buffer.length) return 0;
+    this.buffer = new Array(this.minimumBatchSize);
+    this.action(buffer);
+    return buffer.length;
+  }
+  private async flushIfShould() {
+    return this.buffer.length >= this.minimumBatchSize ? await this.flush() : 0;
+  }
+  /** Enqueue a single item to process in the batch later. Returns items actually processed. */
+  public async do(item: T) {
+    this.buffer.push(item);
+    return await this.flushIfShould();
+  }
+  /** Enqueue an array of items to process in the batch later.
+   * If the additional items goes beyond `minimumBatchSize`,
+   * all items will be flushed at once, however many was added
+   * by `doAll`.
+   * Returns items actually processed. */
+  public async doAll(items: T[]) {
+    this.buffer = this.buffer.concat(items);
+    return await this.flushIfShould();
+  }
+}