send activity log after refreshing cache

onlyjackfrost · onlyjackfrost · commit a874313d4d2b · 2023-09-11T17:54:28.000+08:00
diff --git a/packages/core/src/lib/cache-layer/cacheLayerLoader.ts b/packages/core/src/lib/cache-layer/cacheLayerLoader.ts
@@ -23,19 +23,15 @@ export class CacheLayerLoader implements ICacheLayerLoader {
   private options: ICacheLayerOptions;
   private cacheStorage: DataSource;
   private logger = getLogger({ scopeName: 'CORE' });
-  private activityLoggers: IActivityLogger;
   constructor(
     @inject(TYPES.CacheLayerOptions) options: CacheLayerOptions,
     @inject(TYPES.Factory_DataSource)
-    dataSourceFactory: interfaces.SimpleFactory<DataSource>,
-    @inject(TYPES.Extension_ActivityLogger)
-    activityLogger: IActivityLogger
+    dataSourceFactory: interfaces.SimpleFactory<DataSource>
   ) {
     this.dataSourceFactory = dataSourceFactory;
     this.options = options;
     // prepare cache data source
     this.cacheStorage = this.dataSourceFactory(cacheProfileName);
-    this.activityLoggers = activityLogger;
   }
 
   /**
@@ -50,7 +46,6 @@ export class CacheLayerLoader implements ICacheLayerLoader {
     const { cacheTableName, sql, profile, indexes, folderSubpath } = cache;
     const type = this.options.type!;
     const dataSource = this.dataSourceFactory(profile);
-    await this.activityLoggers.log({ a: 1 });
 
     // generate directory for cache file path to export
     // format => [folderPath]/[schema.templateSource]/[profileName]/[cacheTableName]]/[timestamp]
diff --git a/packages/core/src/lib/cache-layer/cacheLayerRefresher.ts b/packages/core/src/lib/cache-layer/cacheLayerRefresher.ts
@@ -3,10 +3,16 @@ import { uniq } from 'lodash';
 import { ToadScheduler, SimpleIntervalJob, AsyncTask } from 'toad-scheduler';
 import { inject, injectable } from 'inversify';
 import { TYPES } from '@vulcan-sql/core/types';
-import { APISchema } from '@vulcan-sql/core/models';
+import { APISchema, 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 +28,15 @@ export interface ICacheLayerRefresher {
 export class CacheLayerRefresher implements ICacheLayerRefresher {
   private cacheLoader: ICacheLayerLoader;
   private scheduler = new ToadScheduler();
+  private activityLogger: IActivityLogger;
+  private logger = getLogger({ scopeName: 'CORE' });
 
-  constructor(@inject(TYPES.CacheLayerLoader) loader: ICacheLayerLoader) {
+  constructor(
+    @inject(TYPES.CacheLayerLoader) loader: ICacheLayerLoader,
+    @inject(TYPES.Extension_ActivityLogger) activityLogger: IActivityLogger
+  ) {
     this.cacheLoader = loader;
+    this.activityLogger = activityLogger;
   }
 
   public async start(
@@ -40,9 +52,10 @@ export class CacheLayerRefresher implements ICacheLayerRefresher {
       schemas.map(async (schema) => {
         // skip the schema by return if not set the cache
         if (!schema.cache) return;
+        const { urlPath } = schema;
         return await Promise.all(
           schema.cache.map(async (cache) => {
-            const { cacheTableName, profile, refreshTime } = cache;
+            const { cacheTableName, profile, refreshTime, sql } = cache;
             // replace the '/' tp '_' to avoid the file path issue for templateSource
             const templateName = schema.templateSource.replace('/', '_');
             // If refresh time is set, use the scheduler to schedule the load task for refresh
@@ -54,15 +67,56 @@ export class CacheLayerRefresher implements ICacheLayerRefresher {
                 { milliseconds, runImmediately },
                 new AsyncTask(workerId, async () => {
                   // load data the to cache storage
-
-                  await this.cacheLoader.load(templateName, cache);
+                  let refreshResult = RefreshResult.SUCCESS;
+                  const now = moment.utc().format('YYYY-MM-DD HH:mm:ss');
+                  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 = {
+                      logTime: now,
+                      urlPath,
+                      sql,
+                      refreshResult,
+                    };
+                    await this.activityLogger.log(content).catch((err: any) => {
+                      this.logger.debug(
+                        `Failed to log activity after refreshing cache: ${err}`
+                      );
+                    });
+                  }
                 }),
                 { preventOverrun: true, id: workerId }
               );
               // add the job to schedule cache refresh task
               this.scheduler.addIntervalJob(refreshJob);
             } else {
-              await this.cacheLoader.load(templateName, cache);
+              let refreshResult = RefreshResult.SUCCESS;
+              const now = moment.utc().format('YYYY-MM-DD HH:mm:ss');
+              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 = {
+                  logTime: now,
+                  urlPath,
+                  sql,
+                  refreshResult,
+                };
+                await this.activityLogger.log(content).catch((err: any) => {
+                  this.logger.debug(
+                    `Failed to log activity after refreshing cache: ${err}`
+                  );
+                });
+              }
             }
           })
         );
diff --git a/packages/core/test/cache-layer/cacheLayerRefresher.spec.ts b/packages/core/test/cache-layer/cacheLayerRefresher.spec.ts
@@ -12,12 +12,32 @@ import {
   vulcanCacheSchemaName,
 } from '@vulcan-sql/core';
 import { MockDataSource, getQueryResults } from './mockDataSource';
+import { HttpLogger } from '../../src/lib/loggers/httpLogger';
 
 // This is a helper function that will flush all pending promises in the event loop when use the setInterval and the callback is promise (jest > 27 version).
 // reference: https://gist.github.com/apieceofbart/e6dea8d884d29cf88cdb54ef14ddbcc4
 const flushPromises = () =>
   new Promise(jest.requireActual('timers').setImmediate);
 
+jest.mock('../../src/lib/loggers/httpLogger', () => {
+  const originalModule = jest.requireActual('../../src/lib/loggers/httpLogger');
+  return {
+    ...originalModule,
+    HttpLogger: jest.fn().mockImplementation(() => {
+      return {
+        log: jest.fn().mockResolvedValue(true), // Spy on the add method
+      };
+    }),
+  };
+});
+const mockLogger = new HttpLogger(
+  {
+    enabled: true,
+    options: { 'http-logger': { connection: { host: 'localhost' } } },
+  },
+  'http-logger'
+);
+
 describe('Test cache layer refresher', () => {
   const folderPath = 'refresher-test-exported-parquets';
   const profiles = [
@@ -65,6 +85,10 @@ describe('Test cache layer refresher', () => {
     fs.rmSync(folderPath, { recursive: true, force: true });
   });
 
+  afterEach(() => {
+    jest.clearAllMocks();
+  });
+
   it('Should fail to start when exist duplicate cache table name over than one API schema', async () => {
     // Arrange
     const schemas: Array<APISchema> = [
@@ -98,7 +122,7 @@ describe('Test cache layer refresher', () => {
         ] as Array<CacheLayerInfo>,
       },
     ];
-    const refresher = new CacheLayerRefresher(stubCacheLoader);
+    const refresher = new CacheLayerRefresher(stubCacheLoader, mockLogger);
 
     // Act, Assert
     await expect(() => refresher.start(schemas)).rejects.toThrow(
@@ -149,7 +173,7 @@ describe('Test cache layer refresher', () => {
         ] as Array<CacheLayerInfo>,
       },
     ];
-    const refresher = new CacheLayerRefresher(stubCacheLoader);
+    const refresher = new CacheLayerRefresher(stubCacheLoader, mockLogger);
 
     // Act, Assert
     await expect(() => refresher.start(schemas)).rejects.toThrow(
@@ -195,7 +219,7 @@ describe('Test cache layer refresher', () => {
       ];
       // Act
       const loader = new CacheLayerLoader(options, stubFactory as any);
-      const refresher = new CacheLayerRefresher(loader);
+      const refresher = new CacheLayerRefresher(loader, mockLogger);
       await refresher.start(schemas);
 
       // Assert
@@ -271,7 +295,7 @@ describe('Test cache layer refresher', () => {
 
     // Stub the load method to not do any thing.
     stubCacheLoader.load.resolves();
-    const refresher = new CacheLayerRefresher(stubCacheLoader);
+    const refresher = new CacheLayerRefresher(stubCacheLoader, mockLogger);
     // Act
     await refresher.start(schemas);
 
@@ -304,4 +328,98 @@ describe('Test cache layer refresher', () => {
     refresher.stop();
     jest.clearAllTimers();
   });
+
+  it(
+    'Should send activity log after cacheLoader execute "load" successfully',
+    async () => {
+      // Arrange
+      const schemas: Array<APISchema> = [
+        {
+          ...sinon.stubInterface<APISchema>(),
+          templateSource: 'template-1',
+          profiles: [profiles[0].name, profiles[1].name],
+          cache: [
+            {
+              cacheTableName: 'orders',
+              sql: sinon.default.stub() as any,
+              profile: profiles[0].name,
+            },
+            {
+              cacheTableName: 'products',
+              sql: sinon.default.stub() as any,
+              profile: profiles[1].name,
+            },
+          ] as Array<CacheLayerInfo>,
+        },
+        {
+          ...sinon.stubInterface<APISchema>(),
+          templateSource: 'template-2',
+          profiles: [profiles[2].name],
+          cache: [
+            {
+              cacheTableName: 'users',
+              sql: sinon.default.stub() as any,
+              profile: profiles[2].name,
+            },
+          ] as Array<CacheLayerInfo>,
+        },
+      ];
+      // Act
+      const loader = new CacheLayerLoader(options, stubFactory as any);
+      const refresher = new CacheLayerRefresher(loader, mockLogger);
+      await refresher.start(schemas);
+
+      // Assert
+      expect(mockLogger.log).toHaveBeenCalledTimes(3);
+      refresher.stop();
+    },
+    100 * 1000
+  );
+  // Should send activity log when cacheLoader failed on executing "load"
+  it(
+    'Should send activity log after cacheLoader execute "load" failed',
+    async () => {
+      const schemas: Array<APISchema> = [
+        {
+          ...sinon.stubInterface<APISchema>(),
+          templateSource: 'template-1',
+          profiles: [profiles[0].name, profiles[1].name],
+          cache: [
+            {
+              cacheTableName: 'orders',
+              sql: sinon.default.stub() as any,
+              profile: profiles[0].name,
+            },
+            {
+              cacheTableName: 'products',
+              sql: sinon.default.stub() as any,
+              profile: profiles[1].name,
+            },
+          ] as Array<CacheLayerInfo>,
+        },
+        {
+          ...sinon.stubInterface<APISchema>(),
+          templateSource: 'template-2',
+          profiles: [profiles[2].name],
+          cache: [
+            {
+              cacheTableName: 'users',
+              sql: sinon.default.stub() as any,
+              profile: profiles[2].name,
+            },
+          ] as Array<CacheLayerInfo>,
+        },
+      ];
+      // Act
+      const loader = new CacheLayerLoader(options, stubFactory as any);
+      stubCacheLoader.load.throws();
+      const refresher = new CacheLayerRefresher(loader, mockLogger);
+      await refresher.start(schemas);
+
+      // Assert
+      expect(mockLogger.log).toHaveBeenCalledTimes(3);
+      refresher.stop();
+    },
+    100 * 1000
+  );
 });
