Refactoring

packages/common/src/types/index.ts

@@ -10,4 +10,11 @@ export {
 export { DataRecordNested, DataRecordValue, FailureData, IndexResult } from './dataRecord.js';
 export { ConsoleLike } from './logger.js';
 export { Repository, RepositoryIndexingOperations } from './repository.js';
-export { ElasticsearchService } from './service.js';
+export {
+	BulkAction,
+	CreateBulkRequest,
+	DeleteBulkRequest,
+	ElasticsearchService,
+	UpdateBulkRequest,
+	UpsertBulkRequest,
+} from './service.js';

packages/common/src/types/service.ts

@@ -1,9 +1,55 @@
 import type { DataRecordNested, IndexResult } from './dataRecord.js';
 
+export const BulkAction = {
+	DELETE: 'delete',
+	UPSERT: 'upsert',
+	CREATE: 'create',
+	UPDATE: 'update',
+} as const;
+
+export type CreateBulkRequest = {
+	action: typeof BulkAction.CREATE;
+	dataSet: DataRecordNested;
+};
+
+export type DeleteBulkRequest = {
+	action: typeof BulkAction.DELETE;
+	id: string;
+};
+
+export type UpdateBulkRequest = {
+	action: typeof BulkAction.UPDATE;
+	dataSet: DataRecordNested;
+};
+
+export type UpsertBulkRequest = {
+	action: typeof BulkAction.UPSERT;
+	dataSet: DataRecordNested;
+};
+
 /**
  * Interface defining the contract for Elasticsearch service operations.
  */
 export interface ElasticsearchService {
+	/**
+	 * Indexes data into a specified Elasticsearch index.
+	 *
+	 * @param index - The name of the index where the data will be stored.
+	 * @param data - The data to be indexed.
+	 * @returns A promise that resolves to the result of the indexing operation.
+	 */
+	addData(index: string, data: DataRecordNested): Promise<IndexResult>;
+
+	/**
+	 * Performs a bulk upsert operation to index or update multiple documents in the specified index.
+	 * @param index The name of the index where the documents will be upserted
+	 * @param data An array of data records to be upserted.
+	 */
+	bulk(
+		index: string,
+		request: (CreateBulkRequest | UpdateBulkRequest | DeleteBulkRequest | UpsertBulkRequest)[],
+	): Promise<IndexResult>;
+
 	/**
 	 * Creates an index in Elasticsearch.
 	 *
@@ -13,13 +59,13 @@ export interface ElasticsearchService {
 	createIndex(index: string): Promise<boolean>;
 
 	/**
-	 * Indexes data into a specified Elasticsearch index.
+	 * Deletes a document from a specified Elasticsearch index.
 	 *
-	 * @param index - The name of the index where the data will be stored.
-	 * @param data - The data to be indexed.
-	 * @returns A promise that resolves to the result of the indexing operation.
+	 * @param index - The name of the index from which the document will be deleted.
+	 * @param id - The ID of the document to delete.
+	 * @returns A promise that resolves to the result of the deletion operation.
 	 */
-	addData(index: string, data: DataRecordNested): Promise<IndexResult>;
+	deleteData(index: string, id: string): Promise<IndexResult>;
 
 	/**
 	 * Checks the availability of the Elasticsearch service.
@@ -37,20 +83,4 @@ export interface ElasticsearchService {
 	 * @returns A promise that resolves to the result of the update operation.
 	 */
 	updateData(index: string, id: string, data: DataRecordNested): Promise<IndexResult>;
-
-	/**
-	 * Deletes a document from a specified Elasticsearch index.
-	 *
-	 * @param index - The name of the index from which the document will be deleted.
-	 * @param id - The ID of the document to delete.
-	 * @returns A promise that resolves to the result of the deletion operation.
-	 */
-	deleteData(index: string, id: string): Promise<IndexResult>;
-
-	/**
-	 * Performs a bulk upsert operation to index or update multiple documents in the specified index.
-	 * @param index The name of the index where the documents will be upserted
-	 * @param data An array of data records to be upserted.
-	 */
-	bulkUpsert(index: string, data: DataRecordNested[]): Promise<IndexResult>;
 }

packages/indexer-client/src/client/index.ts

@@ -1,8 +1,12 @@
 import {
+	type CreateBulkRequest,
 	type DataRecordNested,
+	type DeleteBulkRequest,
 	ElasticSearchConfig,
 	ElasticsearchService,
 	ElasticSearchSupportedVersions,
+	type UpdateBulkRequest,
+	type UpsertBulkRequest,
 } from '@overture-stack/maestro-common';
 
 import { es7 } from './v7/client.js';
@@ -28,7 +32,8 @@ export const clientProvider = (elasticSearchConfig: ElasticSearchConfig): Elasti
 
 	return {
 		addData: (index: string, data: DataRecordNested) => service.addData(index, data),
-		bulkUpsert: (index: string, data: DataRecordNested[]) => service.bulkUpsert(index, data),
+		bulk: (index: string, request: (CreateBulkRequest | UpdateBulkRequest | DeleteBulkRequest | UpsertBulkRequest)[]) =>
+			service.bulk(index, request),
 		createIndex: (index: string) => service.createIndex(index),
 		deleteData: (index: string, id: string) => service.deleteData(index, id),
 		ping: () => service.ping(),

packages/indexer-client/src/client/v7/client.ts

@@ -1,15 +1,19 @@
 import { Client } from 'es7';
 
 import {
+	type CreateBulkRequest,
 	type DataRecordNested,
+	type DeleteBulkRequest,
 	type ElasticSearchConfig,
 	type ElasticsearchService,
 	ElasticSearchSupportedVersions,
 	type IndexResult,
+	type UpdateBulkRequest,
+	type UpsertBulkRequest,
 } from '@overture-stack/maestro-common';
 
 import { getAuth } from '../../common/config.js';
-import { bulkUpsert, createIndexIfNotExists, deleteData, indexData, ping, updateData } from './operations.js';
+import { bulk, createIndexIfNotExists, deleteData, indexData, ping, updateData } from './operations.js';
 
 /**
  *  Creates an instance of the Elasticsearch service for version 7.
@@ -41,8 +45,11 @@ export const es7 = (config: ElasticSearchConfig): ElasticsearchService => {
 			return indexData(client, index, data);
 		},
 
-		async bulkUpsert(index: string, data: DataRecordNested[]): Promise<IndexResult> {
-			return bulkUpsert(client, index, data);
+		async bulk(
+			index: string,
+			request: (CreateBulkRequest | UpdateBulkRequest | DeleteBulkRequest | UpsertBulkRequest)[],
+		): Promise<IndexResult> {
+			return bulk(client, index, request);
 		},
 
 		async createIndex(index: string): Promise<boolean> {

packages/indexer-client/src/client/v7/operations.ts

@@ -1,23 +1,60 @@
 import type { Client } from 'es7';
 import type { BulkOperationType, BulkResponseItem } from 'es7/api/types';
 
-import { type DataRecordNested, type FailureData, IndexResult, logger } from '@overture-stack/maestro-common';
+import {
+	BulkAction,
+	type CreateBulkRequest,
+	type DataRecordNested,
+	type DeleteBulkRequest,
+	type FailureData,
+	IndexResult,
+	logger,
+	type UpdateBulkRequest,
+	type UpsertBulkRequest,
+} from '@overture-stack/maestro-common';
 
 /**
- * Indexes the specified document. If the document exists, replaces the document and increments the version.
+ * Performs bulk operations (e.g., delete, upsert, create, update) on a specified index in Elasticsearch.
  *
- * @param client An instance of the Elasticsearch `Client` used to perform the indexing operation
- * @param index The name of the Elasticsearch index to create
- * @param dataSet The actual data to be stored in the document
- * @returns
+ * @param client  The Elasticsearch client used to perform the bulk operations
+ * @param index The name of the index on which to perform the bulk operations
+ * @param bulkRequest An array of bulk request objects representing the actions to be performed.
+ * The array can include upsert, delete, create, and update actions. Each action contains specific data
+ * that will be used for the corresponding operation.
+ *
+ * @returns A Promise that resolves to an `IndexResult` object. The result includes:
+ * - `indexName`: The name of the index on which the operations were performed.
+ * - `successful`: A boolean indicating whether all operations were successful.
+ * - `failureData`: An object that maps the index of failed operations to the error details. If no operations fail, this object is empty.
  */
-export const bulkUpsert = async (client: Client, index: string, dataSet: DataRecordNested[]) => {
+export const bulk = async (
+	client: Client,
+	index: string,
+	bulkRequest: (UpsertBulkRequest | DeleteBulkRequest | CreateBulkRequest | UpdateBulkRequest)[],
+): Promise<IndexResult> => {
 	try {
-		const body = dataSet.flatMap((doc) => [{ index: { _index: index, _id: doc?.['id'] } }, doc]);
+		const body = bulkRequest.flatMap((val) => {
+			switch (val.action) {
+				case BulkAction.DELETE:
+					return [{ delete: { _index: index, _id: val.id } }];
+				case BulkAction.UPSERT: {
+					const doc = val.dataSet;
+					return [{ index: { _index: index, _id: doc?.['id'] } }, doc];
+				}
+				case BulkAction.CREATE: {
+					const doc = val.dataSet;
+					return [{ create: { _index: index, _id: doc?.['id'] } }, doc];
+				}
+				case BulkAction.UPDATE: {
+					const doc = val.dataSet;
+					return [{ update: { _index: index, _id: doc?.['id'] } }, { doc }];
+				}
+			}
+		});
 
 		const response = await client.bulk({ refresh: true, body });
 
-		logger.debug(`Bulk upsert in index:'${index}'`, `# of documents:'${dataSet.length}'`, response.statusCode);
+		logger.debug(`Bulk actions in index:'${index}'`, `# of documents:'${bulkRequest.length}'`, response.statusCode);
 
 		const failureData: FailureData = {};
 		if (response.body.errors) {
@@ -40,7 +77,7 @@ export const bulkUpsert = async (client: Client, index: string, dataSet: DataRec
 	} catch (error) {
 		let errorMessage = JSON.stringify(error);
 
-		logger.error(`Error update doc: ${errorMessage}`);
+		logger.error(`Error bulk action: ${errorMessage}`);
 
 		if (typeof error === 'object' && error && 'name' in error && typeof error.name === 'string') {
 			errorMessage = error.name;

packages/indexer-client/src/client/v8/client.ts

@@ -1,15 +1,19 @@
 import { Client } from 'es8';
 
 import {
+	type CreateBulkRequest,
 	type DataRecordNested,
+	type DeleteBulkRequest,
 	type ElasticSearchConfig,
 	type ElasticsearchService,
 	ElasticSearchSupportedVersions,
 	type IndexResult,
+	type UpdateBulkRequest,
+	type UpsertBulkRequest,
 } from '@overture-stack/maestro-common';
 
 import { getAuth } from '../../common/config.js';
-import { bulkUpsert, createIndexIfNotExists, deleteData, indexData, ping, updateData } from './operations.js';
+import { bulk, createIndexIfNotExists, deleteData, indexData, ping, updateData } from './operations.js';
 
 /**
  *  Creates an instance of the Elasticsearch service for version 8.
@@ -40,8 +44,11 @@ export const es8 = (config: ElasticSearchConfig): ElasticsearchService => {
 		async addData(index: string, data: DataRecordNested): Promise<IndexResult> {
 			return indexData(client, index, data);
 		},
-		async bulkUpsert(index: string, data: DataRecordNested[]): Promise<IndexResult> {
-			return bulkUpsert(client, index, data);
+		async bulk(
+			index: string,
+			request: (CreateBulkRequest | UpdateBulkRequest | DeleteBulkRequest | UpsertBulkRequest)[],
+		): Promise<IndexResult> {
+			return bulk(client, index, request);
 		},
 
 		async createIndex(index: string): Promise<boolean> {

packages/indexer-client/src/client/v8/operations.ts

@@ -1,15 +1,60 @@
 import type { Client } from 'es8';
 import type { BulkOperationType, BulkResponseItem } from 'es8/lib/api/types.js';
 
-import { type DataRecordNested, FailureData, IndexResult, logger } from '@overture-stack/maestro-common';
+import {
+	BulkAction,
+	type CreateBulkRequest,
+	type DataRecordNested,
+	type DeleteBulkRequest,
+	FailureData,
+	IndexResult,
+	logger,
+	type UpdateBulkRequest,
+	type UpsertBulkRequest,
+} from '@overture-stack/maestro-common';
 
-export const bulkUpsert = async (client: Client, index: string, dataSet: DataRecordNested[]) => {
+/**
+ * Performs bulk operations (e.g., delete, upsert, create, update) on a specified index in Elasticsearch.
+ *
+ * @param client  The Elasticsearch client used to perform the bulk operations
+ * @param index The name of the index on which to perform the bulk operations
+ * @param bulkRequest An array of bulk request objects representing the actions to be performed.
+ * The array can include upsert, delete, create, and update actions. Each action contains specific data
+ * that will be used for the corresponding operation.
+ *
+ * @returns A Promise that resolves to an `IndexResult` object. The result includes:
+ * - `indexName`: The name of the index on which the operations were performed.
+ * - `successful`: A boolean indicating whether all operations were successful.
+ * - `failureData`: An object that maps the index of failed operations to the error details. If no operations fail, this object is empty.
+ */
+export const bulk = async (
+	client: Client,
+	index: string,
+	bulkRequest: (UpsertBulkRequest | DeleteBulkRequest | CreateBulkRequest | UpdateBulkRequest)[],
+): Promise<IndexResult> => {
 	try {
-		const body = dataSet.flatMap((doc) => [{ index: { _index: index, _id: doc?.['id'] } }, doc]);
+		const body = bulkRequest.flatMap((val) => {
+			switch (val.action) {
+				case BulkAction.DELETE:
+					return [{ delete: { _index: index, _id: val.id } }];
+				case BulkAction.UPSERT: {
+					const doc = val.dataSet;
+					return [{ index: { _index: index, _id: doc?.['id'] } }, doc];
+				}
+				case BulkAction.CREATE: {
+					const doc = val.dataSet;
+					return [{ create: { _index: index, _id: doc?.['id'] } }, doc];
+				}
+				case BulkAction.UPDATE: {
+					const doc = val.dataSet;
+					return [{ update: { _index: index, _id: doc?.['id'] } }, { doc }];
+				}
+			}
+		});
 
 		const response = await client.bulk({ refresh: true, body });
 
-		logger.debug(`Bulk upsert in index:'${index}'`, `# of documents: ${response.items.length}`);
+		logger.debug(`Bulk action in index:'${index}'`, `# of documents: ${response.items.length}`);
 
 		const failureData: FailureData = {};