refactor(query-orchestrator): RedisQueueDriver - migrate to TypeScript (#6528)

Author: ovr

packages/cubejs-base-driver/src/queue-driver.interface.ts

@@ -6,6 +6,7 @@ export interface QueryKeyHash extends String {
   __type: 'QueryKeyHash'
 }
 
+export type GetActiveAndToProcessResponse = [active: string[], toProcess: string[]];
 export type AddToQueueResponse = [added: number, _b: any, _c: any, queueSize: number, addedToQueueTime: number];
 export type QueryStageStateResponse = [active: string[], toProcess: string[]] | [active: string[], toProcess: string[], defs: Record<string, QueryDef>];
 export type RetrieveForProcessingSuccess = [
@@ -51,7 +52,19 @@ export interface QueueDriverConnectionInterface {
   redisHash(queryKey: QueryKey): QueryKeyHash;
   getResultBlocking(queryKey: QueryKey): Promise<unknown>;
   getResult(queryKey: QueryKey): Promise<any>;
-  addToQueue(keyScore: number, queryKey: QueryKey, orphanedTime: any, queryHandler: any, query: AddToQueueQuery, priority: number, options: AddToQueueOptions): Promise<AddToQueueResponse>;
+  /**
+   * Adds specified by the queryKey query to the queue, returns tuple
+   * with the operation result.
+   *
+   * @param keyScore Redis specific thing
+   * @param queryKey
+   * @param orphanedTime
+   * @param queryHandler Our queue allow to use different handlers. For example query, cvsQuery, etc.
+   * @param query
+   * @param priority
+   * @param options
+   */
+  addToQueue(keyScore: number, queryKey: QueryKey, orphanedTime: number, queryHandler: string, query: AddToQueueQuery, priority: number, options: AddToQueueOptions): Promise<AddToQueueResponse>;
   // Return query keys which was sorted by priority and time
   getToProcessQueries(): Promise<string[]>;
   getActiveQueries(): Promise<string[]>;
@@ -74,7 +87,7 @@ export interface QueueDriverConnectionInterface {
   release(): void;
   //
   getQueriesToCancel(): Promise<string[]>
-  getActiveAndToProcess(): Promise<[active: string[], toProcess: string[]]>;
+  getActiveAndToProcess(): Promise<GetActiveAndToProcessResponse>;
 }
 
 export interface QueueDriverInterface {

packages/cubejs-cubestore-driver/src/CubeStoreQueueDriver.ts

@@ -44,7 +44,7 @@ class CubestoreQueueDriverConnection implements QueueDriverConnectionInterface {
   public async addToQueue(
     keyScore: number,
     queryKey: QueryKey,
-    orphanedTime: any,
+    orphanedTime: number,
     queryHandler: string,
     query: AddToQueueQuery,
     priority: number,

packages/cubejs-query-orchestrator/src/orchestrator/AsyncRedisClient.ts

@@ -11,7 +11,7 @@ interface Multi extends Commands<Multi> {
   execAsync: <T = any>() => Promise<T>,
 }
 
-interface AsyncRedisClient extends RedisClient {
+export interface AsyncRedisClient extends RedisClient {
   evalAsync: Commands<Promise<any>>['eval'],
   brpopAsync: Commands<Promise<any>>['brpop'],
   delAsync: Commands<Promise<any>>['del'],

packages/cubejs-query-orchestrator/src/orchestrator/BaseQueueDriver.ts

@@ -16,5 +16,5 @@ export abstract class BaseQueueDriver implements QueueDriverInterface {
 
   abstract createConnection(): Promise<QueueDriverConnectionInterface>;
 
-  abstract release(connection: QueueDriverConnectionInterface): Promise<void>;
+  abstract release(connection: QueueDriverConnectionInterface): void;
 }

packages/cubejs-query-orchestrator/src/orchestrator/RedisPool.ts

@@ -81,7 +81,9 @@ export class RedisPool {
 
   public release(client: AsyncRedisClient) {
     if (this.pool) {
-      this.pool.release(client);
+      this.pool.release(client).catch(() => {
+        // nothing to do
+      });
     } else if (client) {
       client.quit();
     }