feat: add reindex logic

packages/core/src/helpers/fp-ts/delay-task-either.ts

@@ -0,0 +1,12 @@
+import * as TE from 'fp-ts/TaskEither';
+import { timeout } from '../timeout';
+
+export const timeoutTE = (ms: number) => TE.fromTask(async () => timeout(ms));
+
+export const delayTaskEither =
+  <TE>(time: number) =>
+  (task: TE): TE =>
+    TE.chainFirst(() => async () => {
+      await timeout(time);
+      return TE.of(undefined) as any;
+    })(task as any) as TE;

packages/core/src/helpers/fp-ts/index.ts

@@ -1,5 +1,7 @@
 export * from './catch-task-either-tagged-error-te';
+export * from './delay-task-either';
 export * from './tap-task-either-error-te';
 export * from './tap-task-either-error';
 export * from './tap-task-either';
+export * from './try-or-throw';
 export * from './try-tagged-error-task';

packages/core/src/helpers/fp-ts/try-or-throw.ts

@@ -0,0 +1,16 @@
+import { task as T, either as E } from 'fp-ts';
+import { pipe } from 'fp-ts/function';
+
+import type { TaskEither } from 'fp-ts/TaskEither';
+
+export const tryOrThrow = <E, A>(taskEither: TaskEither<E, A>): T.Task<A> =>
+  pipe(
+    taskEither,
+    T.map(either => {
+      if (E.isLeft(either)) {
+        throw either.left;
+      }
+
+      return either.right;
+    }),
+  );

packages/core/src/helpers/index.ts

@@ -2,3 +2,4 @@ export * from './fp-ts';
 export * from './get-first-obj-key-value';
 export * from './nop';
 export * from './reject-falsy-values';
+export * from './timeout';

packages/core/src/helpers/timeout.ts

@@ -0,0 +1,4 @@
+export const timeout = async (ms: number) =>
+  new Promise(resolve => {
+    setTimeout(resolve, ms);
+  });

packages/core/src/types/can-be/can-be-promise.types.ts

@@ -0,0 +1 @@
+export type CanBePromise<T> = T | Promise<T>;

packages/core/src/types/can-be/index.ts

@@ -0,0 +1 @@
+export * from './can-be-promise.types';

packages/core/src/types/entity.types.ts

@@ -0,0 +1 @@
+export type ID = string | number;

packages/core/src/types/fp-ts/index.ts

@@ -1,2 +1,2 @@
-export * from './tagged-task-either.types';
 export * from './tagged-error.types';
+export * from './tagged-task-either.types';

packages/core/src/types/index.ts

@@ -1,2 +1,4 @@
+export * from './can-be';
+export * from './entity.types';
 export * from './fp-ts';
 export * from './value-object.types';

packages/core/src/types/value-object.types.ts

@@ -1,7 +1,5 @@
 export abstract class ValueObject<T> {
-  constructor(readonly props: T) {}
-
-  static of: <A>(props: A) => ValueObject<A>;
+  constructor(readonly props: Readonly<T>) {}
 
   unwrap() {
     return this.props;

packages/reindex/package.json

@@ -35,7 +35,9 @@
     "@elastic/elasticsearch": "^8.12.1",
     "@searchpunch/core": "*",
     "@searchpunch/logger": "*",
+    "async-await-queue": "^2.1.4",
     "elastic-builder": "^2.24.0",
+    "fast-deep-equal": "^3.1.3",
     "fp-ts": "^2.16.2",
     "object-hash": "^3.0.0"
   },

packages/reindex/src/+internal/client/es-monadic-client.error.ts

@@ -6,6 +6,10 @@ export class EsUnableSetAliasError extends TaggedError.ofLiteral(
 
 export class EsNotFoundError extends TaggedError.ofLiteral('EsNotFound') {}
 
+export class EsBulkReindexError extends TaggedError.ofLiteral(
+  'EsCannotBulkReindex',
+) {}
+
 export class EsConnectionRefused extends TaggedError.ofLiteral(
   'EsConnectionRefused',
 ) {}
@@ -16,4 +20,5 @@ export type EsTaggedError =
   | EsConnectionRefused
   | EsUnableSetAliasError
   | EsNotFoundError
+  | EsBulkReindexError
   | EsInternalError;

packages/reindex/src/+internal/client/es-monadic-client.ts

@@ -1,18 +1,20 @@
 import { pipe } from 'fp-ts/function';
 import * as es from '@elastic/elasticsearch';
+import * as esb from 'elastic-builder';
 
 import { PinoLogger, type AbstractLogger } from '@searchpunch/logger';
-import { getFirstObjKeyValue } from '@searchpunch/core';
 
-import type { EsDocId } from './es-monadic-client.types';
+import type { ID } from '@searchpunch/core';
+import type { EsDoc } from './es-monadic-client.types';
 
 import { tryEsTask } from './try-es-task';
 import {
+  EsBulkReindexError,
   EsNotFoundError,
   EsUnableSetAliasError,
 } from './es-monadic-client.error';
 
-type AttrWithEsIndex<F = unknown> = F & {
+export type AttrWithEsIndex<F = unknown> = F & {
   index: string;
 };
 
@@ -28,7 +30,7 @@ export class EsMonadicClient {
 
   constructor(options: EsMonadicClientOptions) {
     this.rawClient = options.client;
-    this.logger = options.logger ?? new PinoLogger('EsMonadicDecorator');
+    this.logger = options.logger ?? new PinoLogger('EsMonadicClient');
   }
 
   static ofConnection = (
@@ -47,7 +49,7 @@ export class EsMonadicClient {
           index,
         });
 
-        return getFirstObjKeyValue(response.body).mappings;
+        return response.body.mappings;
       }, tryEsTask(EsNotFoundError)),
 
     delete: (names: string[]) =>
@@ -84,8 +86,33 @@ export class EsMonadicClient {
       ),
   };
 
+  readonly reindex = {
+    bulk: <Doc extends EsDoc<unknown>>({
+      index,
+      docs,
+    }: AttrWithEsIndex<{ docs: Doc[] }>) =>
+      pipe(
+        async () => {
+          await this.rawClient.bulk({
+            refresh: true,
+            body: docs.flatMap(({ _id, ...doc }) => [
+              {
+                index: {
+                  _index: index,
+                  _id: _id.toString(),
+                },
+              },
+              doc,
+            ]),
+          });
+        },
+        tryEsTask(EsBulkReindexError),
+        this.logger.fp.logTaskEitherError(() => 'Cannot bulk reindex records!'),
+      ),
+  };
+
   readonly record = {
-    get: ({ id, index }: AttrWithEsIndex<{ id: EsDocId }>) =>
+    get: ({ id, index }: AttrWithEsIndex<{ id: ID }>) =>
       pipe(
         async () => {
           const response = await this.rawClient.get({
@@ -99,7 +126,7 @@ export class EsMonadicClient {
         this.logger.fp.logTaskEitherError(() => `Record with ${id} not found!`),
       ),
 
-    delete: ({ id, index }: AttrWithEsIndex<{ id: EsDocId }>) =>
+    delete: ({ id, index }: AttrWithEsIndex<{ id: ID }>) =>
       pipe(
         async () =>
           this.rawClient.delete({
@@ -109,6 +136,32 @@ export class EsMonadicClient {
         tryEsTask(),
         this.logger.fp.logTaskEitherError(() => `Record with ${id} not found!`),
       ),
+
+    deleteByIds: ({ ids, index }: AttrWithEsIndex<{ ids: ID[] }>) =>
+      pipe(
+        async () => {
+          if (!ids.length) {
+            return;
+          }
+
+          await this.rawClient.deleteByQuery({
+            index,
+            body: esb
+              .requestBodySearch()
+              .query(
+                esb.termsQuery(
+                  'id',
+                  ids.map(id => id.toString()),
+                ),
+              )
+              .toJSON(),
+          });
+        },
+        tryEsTask(),
+        this.logger.fp.logTaskEitherError(
+          () => `Cannot delete records ${ids.join(', ')}!`,
+        ),
+      ),
   };
 
   readonly alias = {

packages/reindex/src/+internal/client/es-monadic-client.types.ts

@@ -1 +1,5 @@
-export type EsDocId = string | number;
+import type { ID } from '@searchpunch/core';
+
+export type EsDoc<I> = I & {
+  _id: ID;
+};

packages/reindex/src/+internal/client/index.ts

@@ -1 +1,4 @@
+export * from './es-monadic-client.error';
 export * from './es-monadic-client';
+export * from './es-monadic-client.types';
+export * from './try-es-task';

packages/reindex/src/+internal/helpers/create-index-schema-with-magic-meta.ts

@@ -21,3 +21,15 @@ export const createIndexSchemaWithMagicMeta = <S extends { mappings?: object }>(
     ...schema.mappings,
   },
 });
+
+export const getIndexSchemaMagicMeta = (
+  mappings: any,
+): EsIndexMagicMeta | null => {
+  const maybeMappings = mappings?._meta;
+
+  if (maybeMappings?.tag === MAGIC_ES_SCHEMA_TAG) {
+    return maybeMappings;
+  }
+
+  return null;
+};

packages/reindex/src/es-index/es-index-db-adapter.ts

@@ -0,0 +1,9 @@
+import type { EsDoc } from '+internal';
+import type { CanBePromise, ID } from '@searchpunch/core';
+
+export type EsDbAsyncIdIterator = AsyncIterableIterator<ID[]>;
+
+export type EsIndexDbAdapter<Doc> = {
+  createAllEntitiesIdsIterator: () => EsDbAsyncIdIterator;
+  findEntities: (ids: ID[]) => CanBePromise<Array<EsDoc<Doc>>>;
+};

packages/reindex/src/es-index/es-index.error.ts

@@ -0,0 +1,13 @@
+import { TaggedError } from '@searchpunch/core';
+
+export class EsIndexBulkReindexError extends TaggedError.ofLiteral(
+  'EsIndexBulkReindexError',
+) {}
+
+export class EsIndexStreamReindexError extends TaggedError.ofLiteral(
+  'EsIndexStreamReindexError',
+) {}
+
+export class EsIndexSingleReindexError extends TaggedError.ofLiteral(
+  'EsIndexSingleReindexError',
+) {}