refactor(query-orchestrator): Migrate QueryQueue to TS

Author: ovr

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

@@ -86,7 +86,7 @@ export interface QueueDriverConnectionInterface {
   // Trying to acquire a lock for processing a queue item, this method can return null when
   // multiple nodes tries to process the same query
   retrieveForProcessing(hash: QueryKeyHash, processingId: ProcessingId): Promise<RetrieveForProcessingResponse>;
-  freeProcessingLock(hash: QueryKeyHash, processingId: ProcessingId, activated: unknown): Promise<void>;
+  freeProcessingLock(hash: QueryKeyHash, processingId: ProcessingId, activated: unknown): Promise<void | string>;
   optimisticQueryUpdate(hash: QueryKeyHash, toUpdate: unknown, processingId: ProcessingId, queueId: QueueId | null): Promise<boolean>;
   cancelQuery(queryKey: QueryKey): Promise<QueryDef | null>;
   getQueryAndRemove(hash: QueryKeyHash): Promise<[QueryDef]>;

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

@@ -865,7 +865,7 @@ export class PreAggregationLoader {
     const queue = await this.preAggregations.getQueue(this.preAggregation.dataSource);
     return queue.executeInQueue(
       'query',
-      this.preAggregationQueryKey(invalidationKeys),
+      <any> this.preAggregationQueryKey(invalidationKeys),
       {
         preAggregation: this.preAggregation,
         preAggregationsTablesToTempTables: this.preAggregationsTablesToTempTables,
@@ -2061,7 +2061,7 @@ export class PreAggregations {
     tables = tables.filter(row => `${schema}.${row.table_name}` === table);
 
     // fetching query result
-    const { queueDriver } = this.queue[dataSource];
+    const queueDriver = this.queue[dataSource].getQueueDriver();
     const conn = await queueDriver.createConnection();
     const result = await conn.getResult(key);
     queueDriver.release(conn);

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

@@ -468,7 +468,7 @@ export class QueryCache {
     };
 
     if (!persistent) {
-      return queue.executeInQueue('query', cacheKey, _query, priority, opt);
+      return queue.executeInQueue('query', cacheKey as any, _query, priority, opt);
     } else {
       return queue.executeInQueue('stream', cacheKey, {
         ..._query,
@@ -636,11 +636,11 @@ export class QueryCache {
                 source.once('end', () => cleanup(undefined));
                 source.once('error', cleanup);
                 source.once('close', () => cleanup(undefined));
-      
+
                 target.once('end', () => cleanup(undefined));
                 target.once('error', cleanup);
                 target.once('close', () => cleanup(undefined));
-      
+
                 source.pipe(target);
               })
               .catch((reason) => {