Merge pull request #12754 from immobiliare/fix/search-stream-does-not-await-callbacks

fix(ldap): search stream does not await callbacks

.changeset/slimy-zebras-reply.md

@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-catalog-backend-module-ldap': patch
+---
+
+Now the `searchStream` method in LDAP client awaits the callbacks

plugins/catalog-backend-module-ldap/api-report.md

@@ -90,7 +90,7 @@ export class LdapClient {
   searchStreaming(
     dn: string,
     options: SearchOptions,
-    f: (entry: SearchEntry) => void,
+    f: (entry: SearchEntry) => Promise<void> | void,
   ): Promise<void>;
 }
 

plugins/catalog-backend-module-ldap/src/ldap/client.ts

@@ -14,12 +14,12 @@
  * limitations under the License.
  */
 
-import { ForwardedError } from '@backstage/errors';
+import { ForwardedError, stringifyError } from '@backstage/errors';
 import ldap, { Client, SearchEntry, SearchOptions } from 'ldapjs';
 import { cloneDeep } from 'lodash';
 import { Logger } from 'winston';
 import { BindConfig, TLSConfig } from './config';
-import { errorString } from './util';
+import { createOptions, errorString } from './util';
 import {
   ActiveDirectoryVendor,
   DefaultLdapVendor,
@@ -147,23 +147,45 @@ export class LdapClient {
   async searchStreaming(
     dn: string,
     options: SearchOptions,
-    f: (entry: SearchEntry) => void,
+    f: (entry: SearchEntry) => Promise<void> | void,
   ): Promise<void> {
     try {
       return await new Promise<void>((resolve, reject) => {
         // Note that we clone the (frozen) options, since ldapjs rudely tries to
         // overwrite parts of them
-        this.client.search(dn, cloneDeep(options), (err, res) => {
+        this.client.search(dn, createOptions(options), (err, res) => {
           if (err) {
             reject(new Error(errorString(err)));
           }
+          let awaitList: Array<Promise<void> | void> = [];
+          let transformError = false;
+
+          const transformReject = (e: Error) => {
+            transformError = true;
+            reject(
+              new Error(
+                `Transform function threw an exception, ${stringifyError(e)}`,
+              ),
+            );
+          };
 
           res.on('searchReference', () => {
             this.logger.warn('Received unsupported search referral');
           });
 
           res.on('searchEntry', entry => {
-            f(entry);
+            if (!transformError) awaitList.push(f(entry));
+          });
+
+          res.on('page', (_, cb) => {
+            // awaits completion before fetching next page
+            Promise.all(awaitList)
+              .then(() => {
+                // flush list
+                awaitList = [];
+                if (cb) cb();
+              })
+              .catch(transformReject);
           });
 
           res.on('error', e => {
@@ -176,7 +198,9 @@ export class LdapClient {
             } else if (r.status !== 0) {
               throw new Error(`Got status ${r.status}: ${r.errorMessage}`);
             } else {
-              resolve();
+              Promise.all(awaitList)
+                .then(() => resolve())
+                .catch(transformReject);
             }
           });
         });

plugins/catalog-backend-module-ldap/src/ldap/util.test.ts

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-import { errorString } from './util';
+import { errorString, createOptions } from './util';
 
 describe('errorString', () => {
   it('formats', () => {
@@ -22,3 +22,52 @@ describe('errorString', () => {
     expect(errorString(e)).toEqual('1 n: m');
   });
 });
+
+describe('createOptions', () => {
+  it('should add pagePause', () => {
+    const options = {
+      filter: 'f',
+      paged: true,
+      timeLimit: 42,
+      sizeLimit: 100,
+      derefAliases: 0,
+      typesOnly: false,
+    };
+    expect(createOptions(options)).toEqual({
+      filter: 'f',
+      paged: { pagePause: true },
+      timeLimit: 42,
+      sizeLimit: 100,
+      derefAliases: 0,
+      typesOnly: false,
+    });
+    const options2 = {
+      filter: 'f',
+      paged: {},
+      timeLimit: 42,
+      sizeLimit: 100,
+      derefAliases: 0,
+      typesOnly: false,
+    };
+    expect(createOptions(options2)).toEqual({
+      filter: 'f',
+      paged: { pagePause: true },
+      timeLimit: 42,
+      sizeLimit: 100,
+      derefAliases: 0,
+      typesOnly: false,
+    });
+  });
+
+  it('should not add pagePause', () => {
+    const options = {
+      filter: 'f',
+      paged: false,
+      timeLimit: 42,
+      sizeLimit: 100,
+      derefAliases: 0,
+      typesOnly: false,
+    };
+    expect(createOptions(options)).toEqual(options);
+  });
+});

plugins/catalog-backend-module-ldap/src/ldap/util.ts

@@ -14,7 +14,8 @@
  * limitations under the License.
  */
 
-import { Error as LDAPError, SearchEntry } from 'ldapjs';
+import { Error as LDAPError, SearchEntry, SearchOptions } from 'ldapjs';
+import { cloneDeep } from 'lodash';
 import { LdapVendor } from './vendors';
 
 /**
@@ -54,6 +55,18 @@ export function mapStringAttr(
   }
 }
 
+export function createOptions(inputOptions: SearchOptions): SearchOptions {
+  const result = cloneDeep(inputOptions);
+
+  if (result.paged === true) {
+    result.paged = { pagePause: true };
+  } else if (typeof result.paged === 'object') {
+    result.paged.pagePause = true;
+  }
+
+  return result;
+}
+
 export type RecursivePartial<T> = {
   [P in keyof T]?: T[P] extends (infer U)[]
     ? RecursivePartial<U>[]