♻️ cli: buffering reindexing of documents

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)`);
       },

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();
+  }
+}