Merge pull request #300 from Canner/feature/remote-activity-log

Feature(CORE): Activity log extension

Author: onlyjackfrost

packages/core/src/containers/modules/extension.ts

@@ -3,6 +3,7 @@ import { ExtensionLoader } from '../../lib/extension-loader';
 import { ICoreOptions } from '../../models/coreOptions';
 import templateEngineModules from '../../lib/template-engine/built-in-extensions';
 import validatorModule from '../../lib/validators/built-in-validators';
+import LoggerModule from '../../lib/loggers';
 import {
   builtInCodeLoader,
   builtInTemplateProvider,
@@ -23,6 +24,7 @@ export const extensionModule = (options: ICoreOptions) =>
     for (const templateEngineModule of templateEngineModules) {
       loader.loadInternalExtensionModule(templateEngineModule);
     }
+    loader.loadInternalExtensionModule(LoggerModule);
     // Validator (single module)
     loader.loadInternalExtensionModule(validatorModule);
     // Template provider (single module)

packages/core/src/containers/types.ts

@@ -52,4 +52,6 @@ export const TYPES = {
   Extension_CompilerLoader: Symbol.for('Extension_CompilerLoader'),
   Extension_DataSource: Symbol.for('Extension_DataSource'),
   Extension_ProfileReader: Symbol.for('ProfileReader'),
+  // Logger
+  Extension_ActivityLogger: Symbol.for('Extension_ActivityLogger'),
 };

packages/core/src/index.ts

@@ -1,5 +1,6 @@
 export * from './lib/utils';
 export * from './lib/validators';
+export * from './lib/loggers';
 export * from './lib/template-engine';
 export * from './lib/artifact-builder';
 export * from './lib/data-query';

packages/core/src/lib/cache-layer/cacheLayerLoader.ts

@@ -22,7 +22,6 @@ export class CacheLayerLoader implements ICacheLayerLoader {
   private options: ICacheLayerOptions;
   private cacheStorage: DataSource;
   private logger = getLogger({ scopeName: 'CORE' });
-
   constructor(
     @inject(TYPES.CacheLayerOptions) options: CacheLayerOptions,
     @inject(TYPES.Factory_DataSource)

packages/core/src/lib/cache-layer/cacheLayerRefresher.ts

@@ -1,12 +1,24 @@
 import ms, { StringValue } from 'ms';
 import { uniq } from 'lodash';
 import { ToadScheduler, SimpleIntervalJob, AsyncTask } from 'toad-scheduler';
-import { inject, injectable } from 'inversify';
+import { inject, injectable, multiInject } from 'inversify';
 import { TYPES } from '@vulcan-sql/core/types';
-import { APISchema } from '@vulcan-sql/core/models';
+import {
+  APISchema,
+  ActivityLogContentOptions,
+  ActivityLogType,
+  CacheLayerInfo,
+  IActivityLogger,
+} from '@vulcan-sql/core/models';
 import { ConfigurationError } from '../utils/errors';
 import { ICacheLayerLoader } from './cacheLayerLoader';
+import { getLogger } from '../utils';
+import moment = require('moment');
 
+enum RefreshResult {
+  SUCCESS = 'SUCCESS',
+  FAILED = 'FAILED',
+}
 export interface ICacheLayerRefresher {
   /**
    * Start the job to load the data source to cache storage and created tables from cache settings in schemas
@@ -22,9 +34,16 @@ export interface ICacheLayerRefresher {
 export class CacheLayerRefresher implements ICacheLayerRefresher {
   private cacheLoader: ICacheLayerLoader;
   private scheduler = new ToadScheduler();
+  private activityLoggers: IActivityLogger[];
+  private logger = getLogger({ scopeName: 'CORE' });
 
-  constructor(@inject(TYPES.CacheLayerLoader) loader: ICacheLayerLoader) {
+  constructor(
+    @inject(TYPES.CacheLayerLoader) loader: ICacheLayerLoader,
+    @multiInject(TYPES.Extension_ActivityLogger)
+    activityLoggers: IActivityLogger[]
+  ) {
     this.cacheLoader = loader;
+    this.activityLoggers = activityLoggers;
   }
 
   public async start(
@@ -53,16 +72,14 @@ export class CacheLayerRefresher implements ICacheLayerRefresher {
               const refreshJob = new SimpleIntervalJob(
                 { milliseconds, runImmediately },
                 new AsyncTask(workerId, async () => {
-                  // load data the to cache storage
-
-                  await this.cacheLoader.load(templateName, cache);
+                  await this.loadCacheAndSendActivityLog(schema, cache);
                 }),
                 { preventOverrun: true, id: workerId }
               );
               // add the job to schedule cache refresh task
               this.scheduler.addIntervalJob(refreshJob);
             } else {
-              await this.cacheLoader.load(templateName, cache);
+              await this.loadCacheAndSendActivityLog(schema, cache);
             }
           })
         );
@@ -77,6 +94,44 @@ export class CacheLayerRefresher implements ICacheLayerRefresher {
     this.scheduler.stop();
   }
 
+  private async loadCacheAndSendActivityLog(
+    schema: APISchema,
+    cache: CacheLayerInfo
+  ) {
+    const { urlPath } = schema;
+    const { sql } = cache;
+    let refreshResult = RefreshResult.SUCCESS;
+    const now = moment.utc().format('YYYY-MM-DD HH:mm:ss');
+    const templateName = schema.templateSource.replace('/', '_');
+    try {
+      // get the current time in format of UTC
+      await this.cacheLoader.load(templateName, cache);
+    } catch (error: any) {
+      refreshResult = RefreshResult.FAILED;
+      this.logger.debug(`Failed to refresh cache: ${error}`);
+    } finally {
+      // send activity log
+      const content = {
+        isSuccess: refreshResult === RefreshResult.SUCCESS ? true : false,
+        activityLogType: ActivityLogType.CACHE_REFRESH,
+        logTime: now,
+        urlPath,
+        sql,
+      } as ActivityLogContentOptions;
+      const activityLoggers = this.getActivityLoggers();
+      for (const activityLogger of activityLoggers)
+        activityLogger.log(content).catch((err: any) => {
+          this.logger.debug(
+            `Failed to log activity after refreshing cache: ${err}`
+          );
+        });
+    }
+  }
+
+  private getActivityLoggers(): IActivityLogger[] {
+    return this.activityLoggers.filter((logger) => logger.isEnabled());
+  }
+
   private checkDuplicateCacheTableName(schemas: APISchema[]) {
     const tableNames = schemas
       // => [[table1, table2], [table1, table3], [table4]]

packages/core/src/lib/loggers/httpLogger.ts

@@ -0,0 +1,54 @@
+import {
+  BaseActivityLogger,
+  ActivityLoggerType,
+} from '../../models/extensions/logger';
+import {
+  VulcanExtensionId,
+  VulcanInternalExtension,
+} from '../../models/extensions';
+import axios, { AxiosRequestHeaders } from 'axios';
+import { ConnectionConfig, getUrl } from '../utils/url';
+
+export interface HttpLoggerConfig {
+  connection?: HttpLoggerConnectionConfig;
+}
+
+export interface HttpLoggerConnectionConfig extends ConnectionConfig {
+  headers?: Record<string, string | number | boolean> | undefined;
+}
+
+@VulcanInternalExtension('activity-log')
+@VulcanExtensionId(ActivityLoggerType.HTTP_LOGGER)
+export class HttpLogger extends BaseActivityLogger<HttpLoggerConfig> {
+  private logger = this.getLogger();
+
+  public async log(payload: any): Promise<void> {
+    if (!this.isEnabled()) return;
+    const option = this.getOptions();
+    if (!option?.connection) {
+      throw new Error('Http logger connection should be provided');
+    }
+    const headers = option.connection.headers;
+    const url = getUrl(option.connection);
+    try {
+      // get connection info from option and use axios to send a post requet to the endpoint
+      await this.sendActivityLog(url, payload, headers);
+      this.logger.debug(`Activity log sent`);
+    } catch (err) {
+      this.logger.debug(
+        `Failed to send activity log to http logger, url: ${url}`
+      );
+      throw err;
+    }
+  }
+
+  protected async sendActivityLog(
+    url: string,
+    payload: any,
+    headers: AxiosRequestHeaders | undefined
+  ): Promise<void> {
+    await axios.post(url, payload, {
+      headers: headers,
+    });
+  }
+}

packages/core/src/lib/loggers/index.ts

@@ -0,0 +1,4 @@
+import { HttpLogger } from './httpLogger';
+export * from './httpLogger';
+
+export default [HttpLogger];

packages/core/src/lib/template-engine/compiler-environment/base.ts

@@ -7,6 +7,7 @@ import * as nunjucks from 'nunjucks';
 export abstract class BaseCompilerEnvironment extends nunjucks.Environment {
   abstract getExtensions(): ExtensionBase[];
 
+  // initialize template engines extensions
   public async initializeExtensions() {
     const extensions = this.getExtensions();
     for (const extension of extensions) {

packages/core/src/lib/utils/url.ts

@@ -0,0 +1,14 @@
+export interface ConnectionConfig {
+  ssl?: boolean;
+  host?: string;
+  port?: number | string;
+  path?: string;
+}
+
+export const getUrl = (connection: ConnectionConfig): string => {
+  const { ssl, host, port, path = '' } = connection;
+  const protocol = ssl ? 'https' : 'http';
+  let urlbase = `${protocol}://${host}`;
+  urlbase = port ? `${urlbase}:${port}` : urlbase;
+  return new URL(path, urlbase).href;
+};

packages/core/src/models/coreOptions.ts

@@ -1,6 +1,7 @@
 import { IArtifactBuilderOptions } from './artifactBuilderOptions';
 import { ICacheLayerOptions } from './cacheLayerOptions';
 import { IDocumentOptions } from './documentOptions';
+import { IActivityLoggerOptions } from './loggerOptions';
 import { IProfilesLookupOptions } from './profilesLookupOptions';
 import { ITemplateEngineOptions } from './templateEngineOptions';
 
@@ -24,6 +25,7 @@ export interface ICoreOptions {
   extensions?: ExtensionAliases;
   document?: IDocumentOptions;
   profiles?: IProfilesLookupOptions;
+  'activity-log'?: IActivityLoggerOptions;
   cache?: ICacheLayerOptions;
   [moduleAlias: string]: any;
 }