feat: Introducing Logging Class (Disabled Usage of Logger)

runner/src/hasura-client/hasura-client.ts

@@ -129,6 +129,75 @@ export default class HasuraClient {
     });
   }
 
+  async createLogsTable (source: string, schemaName: string): Promise<any> {
+    return await this.executeSql(
+      `
+      set schema '${schemaName}';
+
+      CREATE TABLE __logs (
+          id BIGSERIAL NOT NULL,
+          block_height NUMERIC(20) NOT NULL,
+          log_date DATE NOT NULL,
+          log_timestamp TIMESTAMP NOT NULL,
+          log_type TEXT NOT NULL,
+          log_level TEXT NOT NULL,
+          message TEXT NOT NULL,
+          PRIMARY KEY (log_date, id)
+      ) PARTITION BY RANGE (log_date);
+
+      CREATE INDEX logs_log_timestamp_idx ON __logs USING btree (log_timestamp);
+      CREATE INDEX logs_log_type_idx ON __logs USING btree (log_type);
+      CREATE INDEX logs_log_level_idx ON __logs USING btree (log_level);
+      CREATE INDEX logs_block_height_idx ON __logs USING btree (block_height);
+      CREATE INDEX logs_search_vector_idx ON __logs USING GIN (to_tsvector('english', message));
+
+
+      CREATE OR REPLACE FUNCTION fn_create_partition(_tbl text, _date date, _interval_start text, _interval_end text)
+      RETURNS void
+      LANGUAGE plpgsql AS
+    $func$
+    DECLARE
+      _start text;
+      _end text;
+      _partition_name text;
+    BEGIN
+      _start := TO_CHAR(date_trunc('day', _date + (_interval_start)::interval), 'YYYY-MM-DD');
+        _end := TO_CHAR(date_trunc('day', _date + (_interval_end)::interval), 'YYYY-MM-DD');
+      _partition_name := TO_CHAR(date_trunc('day', _date + (_interval_start)::interval), 'YYYYMMDD');
+      -- Create partition 
+      EXECUTE 'CREATE TABLE IF NOT EXISTS ' || _tbl || '_p' || _partition_name || ' PARTITION OF ' || _tbl || ' FOR VALUES FROM (''' || _start || ''') TO (''' || _end || ''')';
+    END
+    $func$;
+
+    SELECT fn_create_partition('${schemaName}.__logs', CURRENT_DATE, '0 day', '1 day');
+    SELECT fn_create_partition('${schemaName}.__logs', CURRENT_DATE, '1 day', '2 day');
+
+    CREATE OR REPLACE FUNCTION fn_delete_partition(_tbl text, _date date, _interval_start text, _interval_end text)
+    RETURNS void
+    LANGUAGE plpgsql AS
+  $func$
+  DECLARE
+    _start text;
+    _end text;
+    _partition_name text;
+  BEGIN
+    _start := TO_CHAR(date_trunc('day', _date + (_interval_start)::interval), 'YYYY-MM-DD');
+      _end := TO_CHAR(date_trunc('day', _date + (_interval_end)::interval), 'YYYY-MM-DD');
+    _partition_name := TO_CHAR(date_trunc('day', _date + (_interval_start)::interval), 'YYYYMMDD');
+    -- Detach partition 
+    EXECUTE 'ALTER TABLE ' || _tbl || ' DETACH PARTITION ' || _tbl || '_p' || _partition_name;
+    EXECUTE 'DROP TABLE '  || _tbl || '_p' || _partition_name;
+  END
+  $func$;
+
+  -- GRANT CREATE, USAGE ON SCHEMA cron TO ${source};
+  -- SELECT cron.schedule('${schemaName}___logs_create_partition', '0 1 * * *', $$SELECT fn_create_partition('${schemaName}.logs', CURRENT_DATE, '1 day', '2 day')$$);
+  -- SELECT cron.schedule('${schemaName}___logs_delete_partition', '0 2 * * *', $$SELECT fn_delete_partition('${schemaName}.logs', CURRENT_DATE, '-15 day', '-14 day')$$);
+      `,
+      { source, readOnly: false }
+    );
+  }
+
   async runMigrations (source: string, schemaName: string, migration: string): Promise<any> {
     return await this.executeSql(
       `
@@ -146,7 +215,6 @@ export default class HasuraClient {
         source,
       }
     );
-
     return tablesInSource
       .filter(({ schema }: { schema: string }) => schema === schemaName)
       .map(({ name }: { name: string }) => name);

runner/src/indexer-logger/indexer-logger.test.ts

@@ -0,0 +1,83 @@
+import pgFormat from 'pg-format';
+import IndexerLogger from './indexer-logger';
+import PgClient from '../pg-client';
+import { LogLevel } from '../stream-handler/stream-handler';
+
+describe('IndexerLogger', () => {
+    let pgClient: PgClient;
+    let query: any;
+
+    beforeEach(() => {
+        query = jest.fn().mockReturnValue({ rows: [] });
+        pgClient = {
+          query,
+          format: pgFormat
+        } as unknown as PgClient;
+      });
+
+  const mockDatabaseConnectionParameters = {
+    username: 'test_user',
+    password: 'test_password',
+    host: 'test_host',
+    port: 5432,
+    database: 'test_database'
+  };
+
+  describe('create', () => {
+    it('should create an instance of IndexerLogger with provided PgClient instance', () => {
+      const indexerLogger = IndexerLogger.create(mockDatabaseConnectionParameters, pgClient);
+      expect(indexerLogger).toBeInstanceOf(IndexerLogger);
+    });
+
+    it('should create an instance of IndexerLogger with a new PgClient instance if none provided', () => {
+      const indexerLogger = IndexerLogger.create(mockDatabaseConnectionParameters);
+      expect(indexerLogger).toBeInstanceOf(IndexerLogger);
+    });
+  });
+
+  describe('writeLog', () => {
+        it('should call the query method with the correct parameters', async () => {
+
+            const spyFormatDate = jest.spyOn(IndexerLogger.prototype, 'formatDate');
+
+            const indexerLogger = IndexerLogger.create(mockDatabaseConnectionParameters, pgClient);
+            const blockHeight = 123;
+            const functionName = 'testFunction';
+            const logDate = new Date();
+            const logTimestamp = new Date();
+            const logType = 'system';
+            const logLevel = LogLevel.INFO;
+            const message = 'Test message';
+
+            await indexerLogger.writeLog(
+                blockHeight,
+                functionName,
+                logDate,
+                logTimestamp,
+                logType,
+                logLevel,
+                message
+            );
+
+            expect(typeof blockHeight).toBe('number');
+            expect(typeof functionName).toBe('string');
+            expect(logDate).toBeInstanceOf(Date);
+            expect(logTimestamp).toBeInstanceOf(Date);
+            expect(typeof logType).toBe('string');
+            expect(typeof logLevel).toBe('number'); 
+            expect(typeof message).toBe('string');
+
+            expect(spyFormatDate).toHaveBeenCalledWith(logDate);
+            expect(query.mock.calls.length).toBe(1);
+            expect(query.mock.calls[0][0]).toEqual('INSERT INTO testFunction.__logs (block_height, log_date, log_timestamp, log_type, log_level, message) VALUES ($1, $2, $3, $4, $5, $6)');
+            expect(query.mock.calls[0][1]).toEqual([
+                blockHeight,
+                expect.any(Date), 
+                logTimestamp,
+                logType,
+                LogLevel[logLevel],
+                message
+            ]);
+        });
+    });
+});

runner/src/indexer-logger/indexer-logger.ts

@@ -0,0 +1,60 @@
+import { wrapError } from '../utility';
+import PgClient from '../pg-client';
+import { type DatabaseConnectionParameters } from '../provisioner/provisioner';
+import { LogLevel } from '../stream-handler/stream-handler';
+
+export default class IndexerLogger {
+
+  private constructor (
+    private readonly pgClient: PgClient
+  ) {}
+
+  private extractDateParts(date: Date): { year: number, month: number, day: number } {
+    return {
+        year: date.getFullYear(),
+        month: date.getMonth(),
+        day: date.getDate()
+    };
+  }
+
+  formatDate(date: Date): Date {
+    const { year, month, day } = this.extractDateParts(date);
+    return new Date(year, month, day);
+  }
+
+    static create (
+        databaseConnectionParameters: DatabaseConnectionParameters,
+        pgClientInstance: PgClient | undefined = undefined
+      ): IndexerLogger {
+        const pgClient = pgClientInstance ?? new PgClient({
+          user: databaseConnectionParameters.username,
+          password: databaseConnectionParameters.password,
+          host: process.env.PGHOST,
+          port: Number(databaseConnectionParameters.port),
+          database: databaseConnectionParameters.database,
+        });
+        return new IndexerLogger(pgClient);
+      }
+
+  async writeLog(
+    blockHeight: number,
+    functionName: string,
+    logDate: Date,
+    logTimestamp: Date,
+    logType: string,
+    logLevel: LogLevel,
+    message: string,
+    ): Promise<void> {
+    const schemaName = functionName.replace(/[^a-zA-Z0-9]/g, '_');
+
+    const formattedLogDate = this.formatDate(logDate);
+    const logLevelString = LogLevel[logLevel];
+
+    const query = 
+      `INSERT INTO ${schemaName}.__logs (block_height, log_date, log_timestamp, log_type, log_level, message) VALUES ($1, $2, $3, $4, $5, $6)`;
+
+    const values = [blockHeight, formattedLogDate, logTimestamp, logType, logLevelString, message];
+
+    await wrapError(async () => await this.pgClient.query(query, values), `Failed to execute '${query}' on ${schemaName}`);
+  }
+}