Feat/xodo 1088

.github/workflows/wf-cube.yml

@@ -0,0 +1,33 @@
+on:
+  push:
+    branches:
+      - master
+
+env:
+  REGISTRY: ghcr.io
+  IMAGE_NAME: ${{ github.repository }}
+  tag: 'transcube-${{ github.sha }}'
+
+jobs:
+  build-and-push-image:
+    name: Build Latest - TransAI Cube
+    runs-on: ubuntu-latest
+
+    permissions:
+      contents: read
+      packages: write
+      attestations: write
+      id-token: write
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+      - name: Copy yarn.lock file
+        run: cp yarn.lock packages/cubejs-docker
+      - name: Build & push
+        uses: xip-online-applications/gha-workflow-templates/ghcr-build-push@main
+        with:
+          repository: '${{ github.repository }}/cube'
+          tag: ${{ env.tag }}
+          context: ./
+          dockerfile: ./packages/cubejs-docker/transai.Dockerfile

package.json

@@ -22,7 +22,8 @@
     "lint:fix": "lerna run lint:fix",
     "tsc": "tsc --build",
     "tsc:watch": "tsc --build --watch",
-    "clean": "rimraf packages/*/{tsconfig.tsbuildinfo,lib,dist}"
+    "clean": "rimraf packages/*/{tsconfig.tsbuildinfo,lib,dist}",
+    "link:transai": "cd ./packages/cubejs-backend-shared && yarn link && yarn install && cd ../cubejs-backend-cloud && yarn link && yarn install && cd ../cubejs-backend-native && yarn link && yarn install && cd ../cubejs-server && yarn link && yarn install && yarn install && cd ../cubejs-server-core && yarn link && yarn install && cd ../cubejs-api-gateway && yarn link && yarn install && cd ../cubejs-schema-compiler && yarn link && yarn install && cd ../cubejs-query-orchestrator && yarn link && yarn install && cd ../cubejs-postgres-driver && yarn link && yarn install && cd ../cubejs-cubestore-driver && yarn link && yarn install && cd ../cubejs-event-emitter && yarn link && yarn install"
   },
   "author": "Cube Dev, Inc.",
   "dependencies": {

packages/cubejs-api-gateway/package.json

@@ -27,6 +27,7 @@
     "dist/src/*"
   ],
   "dependencies": {
+    "@cubejs-backend/event-emitter": "file:../cubejs-event-emitter",
     "@cubejs-backend/native": "1.3.78",
     "@cubejs-backend/shared": "1.3.78",
     "@ungap/structured-clone": "^0.3.4",
@@ -48,6 +49,7 @@
     "nexus": "^1.1.0",
     "node-fetch": "^2.6.1",
     "ramda": "^0.27.0",
+    "rxjs": "^7.8.1",
     "uuid": "^8.3.2"
   },
   "devDependencies": {

packages/cubejs-api-gateway/src/LocalSubscriptionStore.ts

@@ -2,6 +2,24 @@ interface LocalSubscriptionStoreOptions {
   heartBeatInterval?: number;
 }
 
+const haveCommonElement = (arr1: string[], arr2: string[]): boolean => {
+  return arr1.some(element => arr2.includes(element));
+}
+
+// TODO: Check whether this is the correct way to get cube names
+const getCubeNames = (query) => {
+  if (!query) {
+    return [];
+  }
+
+  const allColumns = [
+    ...(query.measures || []).map(m => m.split('.')[0]),
+    ...(query.dimensions || []).map(d => d.split('.')[0]),
+  ];
+
+  return Array.from(new Set(allColumns));
+};
+
 export class LocalSubscriptionStore {
   protected connections = {};
 
@@ -20,6 +38,7 @@ export class LocalSubscriptionStore {
     const connection = this.getConnection(connectionId);
     connection.subscriptions[subscriptionId] = {
       ...subscription,
+      cubes: getCubeNames(subscription.message?.params?.query),
       timestamp: new Date()
     };
   }
@@ -45,6 +64,11 @@ export class LocalSubscriptionStore {
     }).reduce((a, b) => a.concat(b), []);
   }
 
+  public async getSubscriptionsByCubeName(cubes: Array<string>) {
+    // TODO: Implement cube filtering by auth context
+    return (await this.getAllSubscriptions()).filter(subscription => haveCommonElement(cubes, subscription.cubes));
+  }
+
   public async cleanupSubscriptions(connectionId: string) {
     delete this.connections[connectionId];
   }

packages/cubejs-api-gateway/src/SubscriptionServer.ts

@@ -1,5 +1,7 @@
 import { v4 as uuidv4 } from 'uuid';
-
+import { EventEmitterInterface } from '@cubejs-backend/event-emitter';
+import { Subject } from 'rxjs';
+import { map, filter, bufferTime, tap } from 'rxjs/operators';
 import { UserError } from './UserError';
 import type { ApiGateway } from './gateway';
 import type { LocalSubscriptionStore } from './LocalSubscriptionStore';
@@ -20,13 +22,35 @@ const calcMessageLength = (message: unknown) => Buffer.byteLength(
 
 export type WebSocketSendMessageFn = (connectionId: string, message: any) => Promise<void>;
 
+const ensureArray = (value: any) => (Array.isArray(value) ? value : [value]);
+
 export class SubscriptionServer {
+  readonly #cubeRenewSubject = new Subject<unknown>();
+
+  readonly #cubeRenewedPipe = this.#cubeRenewSubject.pipe(
+    map((val) => ensureArray(val)),
+    // Map only the renewedCube property
+    map((val) => val as Array<{ renewedCube: string | undefined }>),
+    map((val) => val.map(v => v.renewedCube)),
+    // Buffer the values for 300ms
+    bufferTime(100),
+    map((renewedCubes) => renewedCubes.flat()),
+    map((val) => val.filter(v => v !== undefined)),
+    // Filter out any empty arrays
+    filter((renewedCubes) => renewedCubes.length > 0),
+    // Convert the array of arrays to an array of unique arrays
+    map((renewedCubes) => Array.from(new Set(renewedCubes))),
+  );
+
   public constructor(
     protected readonly apiGateway: ApiGateway,
     protected readonly sendMessage: WebSocketSendMessageFn,
     protected readonly subscriptionStore: LocalSubscriptionStore,
     protected readonly contextAcceptor: ContextAcceptorFn,
+    protected readonly eventEmitter: EventEmitterInterface
   ) {
+    this.eventEmitter.on('cubeRenewed', (val: unknown) => this.#cubeRenewSubject.next(val));
+    this.#cubeRenewedPipe.subscribe(this.renewCubes.bind(this));
   }
 
   public resultFn(connectionId: string, messageId: string, requestId: string | undefined) {
@@ -154,4 +178,21 @@ export class SubscriptionServer {
   public clear() {
     this.subscriptionStore.clear();
   }
+
+  public async renewCubes(cubes) {
+    if (cubes.length === 0) {
+      return;
+    }
+
+    const subs = await this.subscriptionStore.getSubscriptionsByCubeName(cubes);
+
+    if (subs.length === 0) {
+      return;
+    }
+
+    console.log('Renewing subs based on changed cubes', cubes, subs.length);
+    subs.map(async subscription => {
+      this.processMessage(subscription.connectionId, subscription.message, true);
+    });
+  }
 }

packages/cubejs-api-gateway/src/gateway.ts

@@ -1,5 +1,6 @@
 /* eslint-disable no-restricted-syntax */
 import * as stream from 'stream';
+import { EventEmitterInterface } from '@cubejs-backend/event-emitter';
 import { assertNever } from 'assert-never';
 import jwt, { Algorithm as JWTAlgorithm } from 'jsonwebtoken';
 import R from 'ramda';
@@ -180,6 +181,7 @@ class ApiGateway {
     protected readonly compilerApi: (ctx: RequestContext) => Promise<any>,
     protected readonly adapterApi: (ctx: RequestContext) => Promise<any>,
     protected readonly logger: any,
+    protected readonly eventEmitter: EventEmitterInterface,
     protected readonly options: ApiGatewayOptions,
   ) {
     this.dataSourceStorage = options.dataSourceStorage;
@@ -465,6 +467,18 @@ class ApiGateway {
      * jobs scope                                                    *
      *************************************************************** */
 
+    app.get(
+      `${this.basePath}/v1/run-scheduled-refresh`,
+      userMiddlewares,
+      userAsyncHandler(async (req, res) => {
+        await this.runScheduledRefresh({
+          queryingOptions: req.query.queryingOptions,
+          context: req.context,
+          res: this.resToResultFn(res)
+        });
+      })
+    );
+
     app.post(
       `${this.basePath}/v1/pre-aggregations/jobs`,
       userMiddlewares,
@@ -569,13 +583,35 @@ class ApiGateway {
   }
 
   public initSubscriptionServer(sendMessage: WebSocketSendMessageFn) {
-    return new SubscriptionServer(this, sendMessage, this.subscriptionStore, this.wsContextAcceptor);
+    return new SubscriptionServer(this, sendMessage, this.subscriptionStore, this.wsContextAcceptor, this.eventEmitter);
   }
 
   protected duration(requestStarted) {
     return requestStarted && (new Date().getTime() - requestStarted.getTime());
   }
 
+  public async runScheduledRefresh({ context, res, queryingOptions }: {
+    context: RequestContext,
+    res: ResponseResultFn,
+    queryingOptions: any
+  }) {
+    const requestStarted = new Date();
+    try {
+      await this.assertApiScope('jobs', context.securityContext);
+      const refreshScheduler = this.refreshScheduler();
+      res(await refreshScheduler.runScheduledRefresh(context, {
+        ...this.parseQueryParam(queryingOptions || {}),
+        throwErrors: true,
+        forceNoCache: true,
+        isJob: true,
+      }));
+    } catch (e: any) {
+      this.handleError({
+        e, context, res, requestStarted
+      });
+    }
+  }
+
   private filterVisibleItemsInMeta(context: RequestContext, cubes: any[]) {
     const isDevMode = getEnv('devMode');
     function visibilityFilter(item) {
@@ -2090,6 +2126,7 @@ class ApiGateway {
         return;
       }
 
+      // TODO: RENE
       // TODO subscribe to refreshKeys instead of constantly firing load
       await this.load({
         query,

packages/cubejs-api-gateway/test/auth.test.ts

@@ -3,6 +3,7 @@ import express, { Application as ExpressApplication, RequestHandler } from 'expr
 // eslint-disable-next-line import/no-extraneous-dependencies
 import request from 'supertest';
 import jwt from 'jsonwebtoken';
+import { DefaultEventEmitter } from '@cubejs-backend/event-emitter';
 import { pausePromise } from '@cubejs-backend/shared';
 import { resetLogger } from '@cubejs-backend/native';
 
@@ -60,7 +61,9 @@ function createApiGateway(handler: RequestHandler, logger: () => any, options: P
     }
   }
 
-  const apiGateway = new ApiGatewayFake('secret', <any>null, () => adapterApi, logger, {
+  const eventEmitter = new DefaultEventEmitter();
+
+  const apiGateway = new ApiGatewayFake('secret', <any>null, () => adapterApi, logger, eventEmitter, {
     standalone: true,
     dataSourceStorage,
     basePath: '/cubejs-api',

packages/cubejs-api-gateway/test/index.test.ts

@@ -5,6 +5,7 @@ import express from 'express';
 // eslint-disable-next-line import/no-extraneous-dependencies
 import request from 'supertest';
 import jwt from 'jsonwebtoken';
+import { DefaultEventEmitter } from '@cubejs-backend/event-emitter';
 
 import * as console from 'console';
 import { ApiGateway, ApiGatewayOptions, Query, QueryRequest, Request } from '../src';
@@ -52,7 +53,8 @@ async function createApiGateway(
 ) {
   process.env.NODE_ENV = 'production';
 
-  const apiGateway = new ApiGateway(API_SECRET, compilerApi, async () => adapterApi, logger, {
+  const eventEmitter = new DefaultEventEmitter();
+  const apiGateway = new ApiGateway(API_SECRET, compilerApi, async () => adapterApi, logger, eventEmitter, {
     standalone: true,
     dataSourceStorage,
     basePath: '/cubejs-api',

packages/cubejs-api-gateway/test/permissions.test.ts

@@ -6,6 +6,7 @@ import {
   DataSourceStorageMock,
   AdapterApiMock
 } from './mocks';
+import { DefaultEventEmitter } from '@cubejs-backend/event-emitter';
 
 const API_SECRET = 'secret';
 const AUTH_TOKEN = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.e30.t-IDcSemACt8x4iTMCda8Yhe3iZaWbvV5XKSTbuAn0M';
@@ -18,7 +19,9 @@ function createApiGateway(
   const app = express();
   const adapterApi: any = new AdapterApiMock();
   const dataSourceStorage: any = new DataSourceStorageMock();
-  const apiGateway = new ApiGateway(API_SECRET, compilerApi, () => adapterApi, logger, {
+  const eventEmitter = new DefaultEventEmitter();
+
+  const apiGateway = new ApiGateway(API_SECRET, compilerApi, () => adapterApi, logger, eventEmitter, {
     standalone: true,
     dataSourceStorage,
     basePath: '/cubejs-api',

packages/cubejs-backend-shared/src/env.ts

@@ -142,7 +142,7 @@ const variables: Record<string, (...args: any) => any> = {
     .default('false')
     .asBoolStrict(),
   webSockets: () => get('CUBEJS_WEB_SOCKETS')
-    .default('false')
+    .default('true')
     .asBoolStrict(),
   serverHeadersTimeout: () => get('CUBEJS_SERVER_HEADERS_TIMEOUT')
     .asInt(),
@@ -2020,6 +2020,28 @@ const variables: Record<string, (...args: any) => any> = {
   cubeStoreNoHeartBeatTimeout: () => get('CUBEJS_CUBESTORE_NO_HEART_BEAT_TIMEOUT')
     .default('30')
     .asInt(),
+
+  // EventEmitter
+  eventEmitterSetting: () => get('CUBEJS_EVENT_EMITTER_TYPE')
+    .default('memory')
+    .asString(),
+  eventEmitterRedisUrl: () => get('CUBEJS_EVENT_EMITTER_REDIS_URL')
+    .default('')
+    .asString(),
+
+  // Redis
+  redisPoolMin: () => get('CUBEJS_REDIS_POOL_MIN')
+    .default('2')
+    .asInt(),
+  redisPoolMax: () => get('CUBEJS_REDIS_POOL_MAX')
+    .default('1000')
+    .asInt(),
+  redisUseIORedis: () => get('CUBEJS_REDIS_USE_IOREDIS')
+    .default('false')
+    .asBoolStrict(),
+  redisAcquireTimeout: () => get('CUBEJS_REDIS_ACQUIRE_TIMEOUT')
+    .default('5000')
+    .asInt(),
   cubeStoreRollingWindowJoin: () => get('CUBEJS_CUBESTORE_ROLLING_WINDOW_JOIN')
     .default('false')
     .asBoolStrict(),
@@ -2144,10 +2166,6 @@ const variables: Record<string, (...args: any) => any> = {
   // Support for Redis as queue & cache driver was removed in 0.36
   // This code is used to detect Redis and throw an error
   // TODO(ovr): Remove in after 1.0 + LTS
-  redisUseIORedis: () => get('CUBEJS_REDIS_USE_IOREDIS')
-    .default('false')
-    .asBoolStrict(),
-  // TODO(ovr): Remove in after 1.0 + LTS
   redisUrl: () => {
     const redisUrl = get('CUBEJS_REDIS_URL')
       .asString();

packages/cubejs-docker/package.json.transai

@@ -0,0 +1,28 @@
+{
+  "name": "@cubejs-backend/docker",
+  "version": "0.31.20",
+  "description": "Cube.js In Docker (virtual package)",
+  "author": "Cube Dev, Inc.",
+  "license": "Apache-2.0",
+  "private": true,
+  "engines": {
+    "node": "^14.0.0 || ^16.0.0 || >=17.0.0"
+  },
+  "scripts": {
+      "link:dev": "yarn link @cubejs-backend/shared @cubejs-backend/server @cubejs-backend/server-core @cubejs-backend/api-gateway @cubejs-backend/schema-compiler @cubejs-backend/query-orchestrator @cubejs-backend/postgres-driver @cubejs-backend/event-emitter",
+      "link:transai": "cd ./packages/cubejs-backend-shared && yarn link && cd ../cubejs-backend-cloud && yarn link && cd ../cubejs-backend-native && yarn link && cd ../cubejs-server && yarn link && cd ../cubejs-server-core && yarn link && cd ../cubejs-api-gateway && yarn link && cd ../cubejs-schema-compiler && yarn link && cd ../cubejs-query-orchestrator && yarn link && cd ../cubejs-postgres-driver && yarn link && cd ../cubejs-cubestore-driver && yarn link && cd ../cubejs-client-core && yarn link && cd ../cubejs-client-ws-transport && yarn link && cd ../cubejs-event-emitter && yarn link"
+  },
+  "dependencies": {
+    "@cubejs-backend/postgres-driver": "file:/cube-build/packages/cubejs-postgres-driver",
+    "@cubejs-backend/server": "file:/cube-build/packages/cubejs-server",
+    "cubejs-cli": "file:/cube-build/packages/cubejs-cli",
+    "typescript": "~4.1.5",
+    "@typescript-eslint/eslint-plugin": "^4.17.0",
+    "core-js": "^3.34.0",
+    "lerna": "^4.0.0",
+    "redis": "^4.6.14"
+  },
+  "resolutions": {
+    "colors": "1.4.0"
+  }
+}