[PM-15001] Replace throttle decorator (#15015)

* Add comments to AuditService Abstraction

* Replace throttle usage with rxjs mergeMap with concurrent limit

* Add test cases for audit service

* Remove throttle

libs/common/src/abstractions/audit.service.ts

@@ -1,8 +1,17 @@
-// FIXME: Update this file to be type safe and remove this and next line
-// @ts-strict-ignore
 import { BreachAccountResponse } from "../models/response/breach-account.response";
 
 export abstract class AuditService {
-  passwordLeaked: (password: string) => Promise<number>;
-  breachedAccounts: (username: string) => Promise<BreachAccountResponse[]>;
+  /**
+   * Checks how many times a password has been leaked.
+   * @param password The password to check.
+   * @returns A promise that resolves to the number of times the password has been leaked.
+   */
+  abstract passwordLeaked: (password: string) => Promise<number>;
+
+  /**
+   * Retrieves accounts that have been breached for a given username.
+   * @param username The username to check for breaches.
+   * @returns A promise that resolves to an array of BreachAccountResponse objects.
+   */
+  abstract breachedAccounts: (username: string) => Promise<BreachAccountResponse[]>;
 }

libs/common/src/platform/misc/throttle.spec.ts

@@ -1,97 +0,0 @@
-import { throttle } from "./throttle";
-
-describe("throttle decorator", () => {
-  it("should call the function once at a time", async () => {
-    const foo = new Foo();
-    const promises = [];
-    for (let i = 0; i < 10; i++) {
-      promises.push(foo.bar(1));
-    }
-    await Promise.all(promises);
-
-    expect(foo.calls).toBe(10);
-  });
-
-  it("should call the function once at a time for each object", async () => {
-    const foo = new Foo();
-    const foo2 = new Foo();
-    const promises = [];
-    for (let i = 0; i < 10; i++) {
-      promises.push(foo.bar(1));
-      promises.push(foo2.bar(1));
-    }
-    await Promise.all(promises);
-
-    expect(foo.calls).toBe(10);
-    expect(foo2.calls).toBe(10);
-  });
-
-  it("should call the function limit at a time", async () => {
-    const foo = new Foo();
-    const promises = [];
-    for (let i = 0; i < 10; i++) {
-      promises.push(foo.baz(1));
-    }
-    await Promise.all(promises);
-
-    expect(foo.calls).toBe(10);
-  });
-
-  it("should call the function limit at a time for each object", async () => {
-    const foo = new Foo();
-    const foo2 = new Foo();
-    const promises = [];
-    for (let i = 0; i < 10; i++) {
-      promises.push(foo.baz(1));
-      promises.push(foo2.baz(1));
-    }
-    await Promise.all(promises);
-
-    expect(foo.calls).toBe(10);
-    expect(foo2.calls).toBe(10);
-  });
-});
-
-class Foo {
-  calls = 0;
-  inflight = 0;
-
-  @throttle(1, () => "bar")
-  bar(a: number) {
-    this.calls++;
-    this.inflight++;
-    return new Promise((res) => {
-      setTimeout(() => {
-        expect(this.inflight).toBe(1);
-        this.inflight--;
-        res(a * 2);
-      }, Math.random() * 10);
-    });
-  }
-
-  @throttle(5, () => "baz")
-  baz(a: number) {
-    this.calls++;
-    this.inflight++;
-    return new Promise((res) => {
-      setTimeout(() => {
-        expect(this.inflight).toBeLessThanOrEqual(5);
-        this.inflight--;
-        res(a * 3);
-      }, Math.random() * 10);
-    });
-  }
-
-  @throttle(1, () => "qux")
-  qux(a: number) {
-    this.calls++;
-    this.inflight++;
-    return new Promise((res) => {
-      setTimeout(() => {
-        expect(this.inflight).toBe(1);
-        this.inflight--;
-        res(a * 3);
-      }, Math.random() * 10);
-    });
-  }
-}

libs/common/src/platform/misc/throttle.ts

@@ -1,71 +0,0 @@
-// FIXME: Update this file to be type safe and remove this and next line
-// @ts-strict-ignore
-/**
- * Use as a Decorator on async functions, it will limit how many times the function can be
- * in-flight at a time.
- *
- * Calls beyond the limit will be queued, and run when one of the active calls finishes
- */
-export function throttle(limit: number, throttleKey: (args: any[]) => string) {
-  return <T>(
-    target: any,
-    propertyKey: string | symbol,
-    descriptor: TypedPropertyDescriptor<(...args: any[]) => Promise<T>>,
-  ) => {
-    const originalMethod: () => Promise<T> = descriptor.value;
-    const allThrottles = new Map<any, Map<string, (() => void)[]>>();
-
-    const getThrottles = (obj: any) => {
-      let throttles = allThrottles.get(obj);
-      if (throttles != null) {
-        return throttles;
-      }
-      throttles = new Map<string, (() => void)[]>();
-      allThrottles.set(obj, throttles);
-      return throttles;
-    };
-
-    return {
-      value: function (...args: any[]) {
-        const throttles = getThrottles(this);
-        const argsThrottleKey = throttleKey(args);
-        let queue = throttles.get(argsThrottleKey);
-        if (queue == null) {
-          queue = [];
-          throttles.set(argsThrottleKey, queue);
-        }
-
-        return new Promise<T>((resolve, reject) => {
-          const exec = () => {
-            const onFinally = () => {
-              queue.splice(queue.indexOf(exec), 1);
-              if (queue.length >= limit) {
-                queue[limit - 1]();
-              } else if (queue.length === 0) {
-                throttles.delete(argsThrottleKey);
-                if (throttles.size === 0) {
-                  allThrottles.delete(this);
-                }
-              }
-            };
-            originalMethod
-              .apply(this, args)
-              .then((val: any) => {
-                onFinally();
-                return val;
-              })
-              .catch((err: any) => {
-                onFinally();
-                throw err;
-              })
-              .then(resolve, reject);
-          };
-          queue.push(exec);
-          if (queue.length <= limit) {
-            exec();
-          }
-        });
-      },
-    };
-  };
-}

libs/common/src/services/audit.service.spec.ts

@@ -0,0 +1,81 @@
+import { ApiService } from "../abstractions/api.service";
+import { CryptoFunctionService } from "../key-management/crypto/abstractions/crypto-function.service";
+import { ErrorResponse } from "../models/response/error.response";
+
+import { AuditService } from "./audit.service";
+
+jest.useFakeTimers();
+
+// Polyfill global Request for Jest environment if not present
+if (typeof global.Request === "undefined") {
+  global.Request = jest.fn((input: string | URL, init?: RequestInit) => {
+    return { url: typeof input === "string" ? input : input.toString(), ...init };
+  }) as any;
+}
+
+describe("AuditService", () => {
+  let auditService: AuditService;
+  let mockCrypto: jest.Mocked<CryptoFunctionService>;
+  let mockApi: jest.Mocked<ApiService>;
+
+  beforeEach(() => {
+    mockCrypto = {
+      hash: jest.fn().mockResolvedValue(Buffer.from("AABBCCDDEEFF", "hex")),
+    } as unknown as jest.Mocked<CryptoFunctionService>;
+
+    mockApi = {
+      nativeFetch: jest.fn().mockResolvedValue({
+        text: jest.fn().mockResolvedValue(`CDDEEFF:4\nDDEEFF:2\n123456:1`),
+      }),
+      getHibpBreach: jest.fn(),
+    } as unknown as jest.Mocked<ApiService>;
+
+    auditService = new AuditService(mockCrypto, mockApi, 2);
+  });
+
+  it("should not exceed max concurrent passwordLeaked requests", async () => {
+    const inFlight: string[] = [];
+    const maxInFlight: number[] = [];
+
+    // Patch fetchLeakedPasswordCount to track concurrency
+    const origFetch = (auditService as any).fetchLeakedPasswordCount.bind(auditService);
+    jest
+      .spyOn(auditService as any, "fetchLeakedPasswordCount")
+      .mockImplementation(async (password: string) => {
+        inFlight.push(password);
+        maxInFlight.push(inFlight.length);
+        // Simulate async work to allow concurrency limiter to take effect
+        await new Promise((resolve) => setTimeout(resolve, 100));
+        inFlight.splice(inFlight.indexOf(password), 1);
+        return origFetch(password);
+      });
+
+    const p1 = auditService.passwordLeaked("password1");
+    const p2 = auditService.passwordLeaked("password2");
+    const p3 = auditService.passwordLeaked("password3");
+    const p4 = auditService.passwordLeaked("password4");
+
+    jest.advanceTimersByTime(250);
+
+    // Flush all pending timers and microtasks
+    await jest.runAllTimersAsync();
+    await Promise.all([p1, p2, p3, p4]);
+
+    // The max value in maxInFlight should not exceed 2 (the concurrency limit)
+    expect(Math.max(...maxInFlight)).toBeLessThanOrEqual(2);
+    expect((auditService as any).fetchLeakedPasswordCount).toHaveBeenCalledTimes(4);
+    expect(mockCrypto.hash).toHaveBeenCalledTimes(4);
+    expect(mockApi.nativeFetch).toHaveBeenCalledTimes(4);
+  });
+
+  it("should return empty array for breachedAccounts on 404", async () => {
+    mockApi.getHibpBreach.mockRejectedValueOnce({ statusCode: 404 } as ErrorResponse);
+    const result = await auditService.breachedAccounts("user@example.com");
+    expect(result).toEqual([]);
+  });
+
+  it("should throw error for breachedAccounts on non-404 error", async () => {
+    mockApi.getHibpBreach.mockRejectedValueOnce({ statusCode: 500 } as ErrorResponse);
+    await expect(auditService.breachedAccounts("user@example.com")).rejects.toThrow();
+  });
+});

libs/common/src/services/audit.service.ts

@@ -1,21 +1,58 @@
+import { Subject } from "rxjs";
+import { mergeMap } from "rxjs/operators";
+
 import { ApiService } from "../abstractions/api.service";
 import { AuditService as AuditServiceAbstraction } from "../abstractions/audit.service";
 import { CryptoFunctionService } from "../key-management/crypto/abstractions/crypto-function.service";
 import { BreachAccountResponse } from "../models/response/breach-account.response";
 import { ErrorResponse } from "../models/response/error.response";
-import { throttle } from "../platform/misc/throttle";
 import { Utils } from "../platform/misc/utils";
 
 const PwnedPasswordsApi = "https://api.pwnedpasswords.com/range/";
 
 export class AuditService implements AuditServiceAbstraction {
+  private passwordLeakedSubject = new Subject<{
+    password: string;
+    resolve: (count: number) => void;
+    reject: (err: any) => void;
+  }>();
+
   constructor(
     private cryptoFunctionService: CryptoFunctionService,
     private apiService: ApiService,
-  ) {}
+    private readonly maxConcurrent: number = 100, // default to 100, can be overridden
+  ) {
+    this.maxConcurrent = maxConcurrent;
+    this.passwordLeakedSubject
+      .pipe(
+        mergeMap(
+          // Handle each password leak request, resolving or rejecting the associated promise.
+          async (req) => {
+            try {
+              const count = await this.fetchLeakedPasswordCount(req.password);
+              req.resolve(count);
+            } catch (err) {
+              req.reject(err);
+            }
+          },
+          this.maxConcurrent, // Limit concurrent API calls
+        ),
+      )
+      .subscribe();
+  }
 
-  @throttle(100, () => "passwordLeaked")
   async passwordLeaked(password: string): Promise<number> {
+    return new Promise<number>((resolve, reject) => {
+      this.passwordLeakedSubject.next({ password, resolve, reject });
+    });
+  }
+
+  /**
+   * Fetches the count of leaked passwords from the Pwned Passwords API.
+   * @param password The password to check.
+   * @returns A promise that resolves to the number of times the password has been leaked.
+   */
+  protected async fetchLeakedPasswordCount(password: string): Promise<number> {
     const hashBytes = await this.cryptoFunctionService.hash(password, "sha1");
     const hash = Utils.fromBufferToHex(hashBytes).toUpperCase();
     const hashStart = hash.substr(0, 5);