fix: await credentials when subscribing

src/AsyncUtils.ts

@@ -11,10 +11,10 @@ export async function* map<T, U>(
 
 export async function* filter<T>(
   iterable: AsyncIterable<T>,
-  predicate: (value: T) => boolean,
+  predicate: (value: T) => Promise<boolean> | boolean,
 ): AsyncGenerator<T, void, void> {
   for await (const value of iterable) {
-    if (predicate(value)) yield value;
+    if (await predicate(value)) yield value;
   }
 }
 

src/AuthorizedSocketConnection.ts

@@ -94,11 +94,11 @@ export default class AuthorizedSocketConnection<TContext, TCredentials> {
     this.socket.emit('app_error', error);
   }
 
-  isAuthorized(
+  async isAuthorized(
     data: any,
     hasPermission: (data: any, credentials: TCredentials) => boolean,
   ) {
-    const credentials = this.config.credentialsManager.getCredentials();
+    const credentials = await this.config.credentialsManager.getCredentials();
     const isAuthorized = !!credentials && hasPermission(data, credentials);
     if (!isAuthorized) {
       this.log('info', 'unauthorized', {
@@ -239,7 +239,7 @@ export default class AuthorizedSocketConnection<TContext, TCredentials> {
     const stream: AsyncIterable<unknown> = resultOrStream as any;
 
     for await (const payload of stream) {
-      const credentials = this.config.credentialsManager.getCredentials();
+      const credentials = await this.config.credentialsManager.getCredentials();
 
       let response;
       try {

src/CredentialsManager.ts

@@ -1,5 +1,5 @@
 export interface CredentialsManager<TCredentials> {
-  getCredentials(): TCredentials | null | undefined;
+  getCredentials(): Promise<TCredentials | null | undefined>;
   authenticate(authorization: string): void | Promise<void>;
   unauthenticate(): void | Promise<void>; // allow for redis etc down the line
 }

src/JwtCredentialsManager.ts

@@ -19,19 +19,21 @@ export default abstract class JwtCredentialsManager<
 
   token: string | null | undefined;
 
-  credentials: TCredentials | null | undefined;
+  credentialsPromise: Promise<TCredentials | null | undefined> | null;
 
   renewHandle: NodeJS.Timeout | null | undefined;
 
   constructor(config: JwtCredentialsManagerConfig) {
     this.config = config;
 
     this.token = null;
-    this.credentials = null;
+    this.credentialsPromise = null;
   }
 
-  getCredentials(): TCredentials | null | undefined {
-    const { credentials } = this;
+  async getCredentials(): Promise<TCredentials | null | undefined> {
+    if (!this.credentialsPromise) return null;
+
+    const credentials = await this.credentialsPromise;
     if (credentials && Date.now() >= credentials.exp * SECONDS_TO_MS) {
       return null;
     }
@@ -56,7 +58,7 @@ export default abstract class JwtCredentialsManager<
     }
 
     this.token = null;
-    this.credentials = null;
+    this.credentialsPromise = null;
   }
 
   async updateCredentials() {
@@ -65,21 +67,22 @@ export default abstract class JwtCredentialsManager<
       throw new Error('JwtCredentialManager: Unauthenticated');
     }
 
-    const credentials = await this.getCredentialsFromAuthorization(token);
+    this.credentialsPromise = Promise.resolve(
+      this.getCredentialsFromAuthorization(token),
+    );
+    await this.credentialsPromise;
 
     // Avoid race conditions with multiple updates.
     if (this.token !== token) {
       return;
     }
 
-    this.credentials = credentials;
-
     // TODO: Don't schedule renewal if the new credentials are expired or
     // almost expired.
     this.scheduleRenewCredentials();
   }
 
-  scheduleRenewCredentials() {
+  async scheduleRenewCredentials() {
     if (this.renewHandle) {
       clearTimeout(this.renewHandle);
     }
@@ -89,7 +92,7 @@ export default abstract class JwtCredentialsManager<
       return;
     }
 
-    const { credentials } = this;
+    const credentials = await this.credentialsPromise;
     if (!credentials) {
       return;
     }