fix: inline

Author: shortcuts

specs/search/helpers/chunkedPush.yml

@@ -8,7 +8,7 @@ import { {{{dependencyName}}}Client } from '{{{dependencyPackage}}}';
 import type { {{#lambda.titlecase}}{{{dependencyName}}}{{/lambda.titlecase}}Client } from '{{{dependencyPackage}}}';
 {{/dependencies}}
 
-import type { ChunkedBatchOptions, ReplaceAllObjectsOptions, ReplaceAllObjectsWithTransformationResponse } from '@algolia/client-search';
+import type { ReplaceAllObjectsOptions, ReplaceAllObjectsWithTransformationResponse } from '@algolia/client-search';
 import type { PartialUpdateObjectsOptions, SaveObjectsOptions } from '@algolia/client-search';
 import type { PushTaskRecords, WatchResponse } from '@algolia/ingestion';
 
@@ -75,20 +75,6 @@ export type Algoliasearch = SearchClient & {
     options: ReplaceAllObjectsOptions,
     requestOptions?: RequestOptions | undefined,
   ) => Promise<ReplaceAllObjectsWithTransformationResponse>;
-
-  /**
-   * Helper: Chunks the given `objects` list in subset of 1000 elements max in order to make it fit in `push` requests by leveraging the Transformation pipeline setup in the Push connector (https://www.algolia.com/doc/guides/sending-and-managing-data/send-and-update-your-data/connectors/push/).
-   *
-   * @summary Helper: Chunks the given `objects` list in subset of 1000 elements max in order to make it fit in `batch` requests.
-   * @param chunkedPush - The `chunkedPush` object.
-   * @param chunkedPush.indexName - The `indexName` to replace `objects` in.
-   * @param chunkedPush.objects - The array of `objects` to store in the given Algolia `indexName`.
-   * @param chunkedPush.action - The `batch` `action` to perform on the given array of `objects`, defaults to `addObject`.
-   * @param chunkedPush.waitForTasks - Whether or not we should wait until every `batch` tasks has been processed, this operation may slow the total execution time of this method but is more reliable.
-   * @param chunkedPush.batchSize - The size of the chunk of `objects`. The number of `batch` calls will be equal to `length(objects) / batchSize`. Defaults to 1000.
-   * @param requestOptions - The requestOptions to send along with the query, they will be forwarded to the `getTask` method and merged with the transporter requestOptions.
-   */
-  chunkedPush: (options: ChunkedBatchOptions, requestOptions?: RequestOptions) => Promise<Array<WatchResponse>>;
 };
 
 export type TransformationOptions = {
@@ -174,71 +160,6 @@ export function algoliasearch(
       );
     },
 
-    async chunkedPush(
-      { indexName, objects, action = 'addObject', waitForTasks, batchSize = 1000 }: ChunkedBatchOptions,
-      requestOptions?: RequestOptions,
-    ): Promise<Array<WatchResponse>> {
-      if (!ingestionTransporter) {
-        throw new Error('`transformation.region` must be provided at client instantiation before calling this method.');
-      }
-
-      if (!options?.transformation?.region) {
-        throw new Error('`region` must be provided when leveraging the transformation pipeline');
-      }
-
-      let records: Array<PushTaskRecords> = [];
-      const responses: Array<WatchResponse> = [];
-
-      const objectEntries = objects.entries();
-      for (const [i, obj] of objectEntries) {
-        records.push(obj as PushTaskRecords);
-        if (records.length === batchSize || i === objects.length - 1) {
-          responses.push(
-            await ingestionTransporter.push(
-              { indexName, pushTaskPayload: { action, records }, watch: waitForTasks },
-              requestOptions,
-            ),
-          );
-          records = [];
-        }
-      }
-
-      let retryCount = 0;
-
-      if (waitForTasks) {
-        for (const resp of responses) {
-          if (resp.eventID === undefined || !resp.eventID) {
-            throw new Error('received unexpected response from the push endpoint, eventID must not be undefined');
-          }
-
-          await createIterablePromise({
-            func: async () => {
-              if (resp.eventID === undefined || !resp.eventID) {
-                throw new Error('received unexpected response from the push endpoint, eventID must not be undefined');
-              }
-
-              return ingestionTransporter.getEvent({ runID: resp.runID, eventID: resp.eventID }).catch((error: ApiError) => {
-                if (error.status === 404) {
-                  return undefined;
-                }
-
-                throw error;
-              })
-            },
-            validate: (response) => response !== undefined,
-            aggregator: () => (retryCount += 1),
-            error: {
-              validate: () => retryCount >= 50,
-              message: () => `The maximum number of retries exceeded. (${retryCount}/${50})`,
-            },
-            timeout: (): number => Math.min(retryCount * 200, 5000),
-          });
-        }
-      }
-
-      return responses;
-    },
-
     async replaceAllObjectsWithTransformation(
       { indexName, objects, batchSize, scopes }: ReplaceAllObjectsOptions,
       requestOptions?: RequestOptions | undefined,
@@ -271,10 +192,59 @@ export function algoliasearch(
           requestOptions,
         );
 
-        const watchResponses = await this.chunkedPush(
-          { indexName: tmpIndexName, objects, waitForTasks: true, batchSize },
-          requestOptions,
-        );
+        let records: Array<PushTaskRecords> = [];
+        const watchResponses: Array<WatchResponse> = [];
+
+        const objectEntries = objects.entries();
+        for (const [i, obj] of objectEntries) {
+          records.push(obj as PushTaskRecords);
+          if (records.length === batchSize || i === objects.length - 1) {
+            watchResponses.push(
+              await ingestionTransporter.push(
+                {
+                  indexName: tmpIndexName,
+                  pushTaskPayload: { action: 'addObject', records },
+                  referenceIndexName: indexName,
+                },
+                requestOptions,
+              ),
+            );
+            records = [];
+          }
+        }
+
+        let retryCount = 0;
+
+        for (const resp of watchResponses) {
+          if (resp.eventID === undefined || !resp.eventID) {
+            throw new Error('received unexpected response from the push endpoint, eventID must not be undefined');
+          }
+
+          await createIterablePromise({
+            func: async () => {
+              if (resp.eventID === undefined || !resp.eventID) {
+                throw new Error('received unexpected response from the push endpoint, eventID must not be undefined');
+              }
+
+              return ingestionTransporter
+                .getEvent({ runID: resp.runID, eventID: resp.eventID })
+                .catch((error: ApiError) => {
+                  if (error.status === 404) {
+                    return undefined;
+                  }
+
+                  throw error;
+                });
+            },
+            validate: (response) => response !== undefined,
+            aggregator: () => (retryCount += 1),
+            error: {
+              validate: () => retryCount >= 50,
+              message: () => `The maximum number of retries exceeded. (${retryCount}/${50})`,
+            },
+            timeout: (): number => Math.min(retryCount * 200, 5000),
+          });
+        }
 
         await this.waitForTask({
           indexName: tmpIndexName,