feat: event based history tables, dbs and importer

prisma/migrations/20240404105443_history_v2/migration.sql

@@ -0,0 +1,47 @@
+-- CreateTable
+CREATE TABLE "PairLiquidityInfoHistoryV2" (
+    "id" SERIAL NOT NULL,
+    "pairId" INTEGER NOT NULL,
+    "eventType" TEXT NOT NULL,
+    "height" INTEGER NOT NULL,
+    "microBlockTime" BIGINT NOT NULL,
+    "logIndex" INTEGER NOT NULL,
+    "microBlockHash" TEXT NOT NULL,
+    "transactionHash" TEXT NOT NULL,
+    "totalSupply" TEXT,
+    "reserve0" TEXT,
+    "reserve1" TEXT,
+    "deltaReserve0" TEXT,
+    "deltaReserve1" TEXT,
+    "fiatPrice" TEXT,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "PairLiquidityInfoHistoryV2_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "PairLiquidityInfoHistoryV2Error" (
+    "id" SERIAL NOT NULL,
+    "pairId" INTEGER NOT NULL,
+    "microBlockHash" TEXT NOT NULL,
+    "logIndex" INTEGER NOT NULL,
+    "error" TEXT NOT NULL,
+    "timesOccurred" INTEGER NOT NULL DEFAULT 1,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "PairLiquidityInfoHistoryV2Error_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "PairLiquidityInfoHistoryV2_pairId_microBlockHash_logIndex_key" ON "PairLiquidityInfoHistoryV2"("pairId", "microBlockHash", "logIndex");
+
+-- CreateIndex
+CREATE UNIQUE INDEX "PairLiquidityInfoHistoryV2Error_pairId_microBlockHash_logIn_key" ON "PairLiquidityInfoHistoryV2Error"("pairId", "microBlockHash", "logIndex", "error");
+
+-- AddForeignKey
+ALTER TABLE "PairLiquidityInfoHistoryV2" ADD CONSTRAINT "PairLiquidityInfoHistoryV2_pairId_fkey" FOREIGN KEY ("pairId") REFERENCES "Pair"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "PairLiquidityInfoHistoryV2Error" ADD CONSTRAINT "PairLiquidityInfoHistoryV2Error_pairId_fkey" FOREIGN KEY ("pairId") REFERENCES "Pair"("id") ON DELETE RESTRICT ON UPDATE CASCADE;

prisma/schema.prisma

@@ -25,16 +25,18 @@ model Token {
 }
 
 model Pair {
-  id                        Int                             @id @default(autoincrement())
-  address                   String                          @unique
-  token0                    Token                           @relation("Pair_Token0", fields: [t0], references: [id])
-  t0                        Int
-  token1                    Token                           @relation("Pair_Token1", fields: [t1], references: [id])
-  t1                        Int
-  liquidityInfo             PairLiquidityInfo?              @relation("Pair_Info")
-  synchronized              Boolean
-  liquidityInfoHistory      PairLiquidityInfoHistory[]
-  liquidityInfoHistoryError PairLiquidityInfoHistoryError[]
+  id                          Int                               @id @default(autoincrement())
+  address                     String                            @unique
+  token0                      Token                             @relation("Pair_Token0", fields: [t0], references: [id])
+  t0                          Int
+  token1                      Token                             @relation("Pair_Token1", fields: [t1], references: [id])
+  t1                          Int
+  liquidityInfo               PairLiquidityInfo?                @relation("Pair_Info")
+  synchronized                Boolean
+  liquidityInfoHistory        PairLiquidityInfoHistory[]
+  liquidityInfoHistoryV2      PairLiquidityInfoHistoryV2[]
+  liquidityInfoHistoryError   PairLiquidityInfoHistoryError[]
+  liquidityInfoHistoryV2Error PairLiquidityInfoHistoryV2Error[]
 }
 
 model PairLiquidityInfo {
@@ -73,3 +75,39 @@ model PairLiquidityInfoHistoryError {
 
   @@unique(name: "pairIdMicroBlockHashErrorUniqueIndex", [pairId, microBlockHash, error])
 }
+
+model PairLiquidityInfoHistoryV2 {
+  id              Int      @id @default(autoincrement())
+  pair            Pair     @relation(fields: [pairId], references: [id])
+  pairId          Int
+  eventType       String
+  height          Int
+  microBlockTime  BigInt
+  logIndex        Int
+  microBlockHash  String
+  transactionHash String
+  totalSupply     String?
+  reserve0        String?
+  reserve1        String?
+  deltaReserve0   String?
+  deltaReserve1   String?
+  fiatPrice       String?
+  createdAt       DateTime @default(now())
+  updatedAt       DateTime @default(now()) @updatedAt
+
+  @@unique(name: "pairIdMicroBlockHashLogIndexUniqueIndex", [pairId, microBlockHash, logIndex])
+}
+
+model PairLiquidityInfoHistoryV2Error {
+  id             Int      @id @default(autoincrement())
+  pair           Pair     @relation(fields: [pairId], references: [id])
+  pairId         Int
+  microBlockHash String
+  logIndex       Int
+  error          String
+  timesOccurred  Int      @default(1)
+  createdAt      DateTime @default(now())
+  updatedAt      DateTime @default(now()) @updatedAt
+
+  @@unique(name: "pairIdMicroBlockHashLogIndexUniqueIndex", [pairId, microBlockHash, logIndex, error])
+}

src/app.module.ts

@@ -5,11 +5,12 @@ import { TokensService } from './api/tokens/tokens.service';
 import { PairsService } from './api/pairs/pairs.service';
 import { ClientsModule } from './clients/clients.module';
 import { ApiModule } from './api/api.module';
-import { PairSyncService } from './tasks/pair-sync.service';
+import { PairSyncService } from './tasks/pair-sync/pair-sync.service';
 import { MdwWsClientService } from './clients/mdw-ws-client.service';
+import { TasksModule } from './tasks/tasks.module';
 
 @Module({
-  imports: [ApiModule, ClientsModule, DatabaseModule],
+  imports: [ApiModule, ClientsModule, DatabaseModule, TasksModule],
   controllers: [AppController],
   providers: [MdwWsClientService, PairsService, TokensService, PairSyncService],
 })

src/clients/clients.module.ts

@@ -1,9 +1,10 @@
 import { Module } from '@nestjs/common';
 import { MdwHttpClientService } from './mdw-http-client.service';
 import { SdkClientService } from './sdk-client.service';
+import { MdwWsClientService } from './mdw-ws-client.service';
 
 @Module({
-  providers: [MdwHttpClientService, MdwHttpClientService, SdkClientService],
-  exports: [MdwHttpClientService, MdwHttpClientService, SdkClientService],
+  providers: [MdwHttpClientService, MdwWsClientService, SdkClientService],
+  exports: [MdwHttpClientService, MdwWsClientService, SdkClientService],
 })
 export class ClientsModule {}

src/database/database.module.ts

@@ -4,19 +4,25 @@ import { PairDbService } from './pair/pair-db.service';
 import { PairLiquidityInfoHistoryDbService } from './pair-liquidity-info-history/pair-liquidity-info-history-db.service';
 import { PairLiquidityInfoHistoryErrorDbService } from './pair-liquidity-info-history-error/pair-liquidity-info-history-error-db.service';
 import { TokenDbService } from './token/token-db.service';
+import { PairLiquidityInfoHistoryV2DbService } from './pair-liquidity-info-history/pair-liquidity-info-history-v2-db.service';
+import { PairLiquidityInfoHistoryV2ErrorDbService } from './pair-liquidity-info-history-error/pair-liquidity-info-history-v2-error-db.service';
 
 @Module({
   providers: [
     PrismaService,
     PairDbService,
     PairLiquidityInfoHistoryDbService,
     PairLiquidityInfoHistoryErrorDbService,
+    PairLiquidityInfoHistoryV2DbService,
+    PairLiquidityInfoHistoryV2ErrorDbService,
     TokenDbService,
   ],
   exports: [
     PairDbService,
     PairLiquidityInfoHistoryDbService,
     PairLiquidityInfoHistoryErrorDbService,
+    PairLiquidityInfoHistoryV2DbService,
+    PairLiquidityInfoHistoryV2ErrorDbService,
     TokenDbService,
   ],
 })

src/database/pair-liquidity-info-history-error/pair-liquidity-info-history-v2-error-db.service.ts

@@ -0,0 +1,51 @@
+import { Injectable } from '@nestjs/common';
+import { PrismaService } from '../prisma.service';
+import { PairLiquidityInfoHistoryV2Error } from '@prisma/client';
+
+@Injectable()
+export class PairLiquidityInfoHistoryV2ErrorDbService {
+  constructor(private prisma: PrismaService) {}
+
+  getErrorByPairIdAndMicroBlockHashWithinHours(
+    pairId: number,
+    microBlockHash: string,
+    logIndex: number,
+    withinHours: number,
+  ): Promise<PairLiquidityInfoHistoryV2Error | null> {
+    return this.prisma.pairLiquidityInfoHistoryV2Error.findFirst({
+      where: {
+        pairId: pairId,
+        microBlockHash: microBlockHash,
+        logIndex: logIndex,
+        updatedAt: {
+          gt: new Date(Date.now() - withinHours * 60 * 60 * 1000),
+        },
+      },
+    });
+  }
+
+  upsert(
+    data: Omit<
+      PairLiquidityInfoHistoryV2Error,
+      'id' | 'timesOccurred' | 'createdAt' | 'updatedAt'
+    >,
+  ) {
+    return this.prisma.pairLiquidityInfoHistoryV2Error.upsert({
+      where: {
+        pairIdMicroBlockHashLogIndexUniqueIndex: {
+          pairId: data.pairId,
+          microBlockHash: data.microBlockHash,
+          logIndex: data.logIndex,
+          error: data.error,
+        },
+      },
+      update: { timesOccurred: { increment: 1 } },
+      create: {
+        pairId: data.pairId,
+        microBlockHash: data.microBlockHash,
+        logIndex: data.logIndex,
+        error: data.error,
+      },
+    });
+  }
+}

src/database/pair-liquidity-info-history/pair-liquidity-info-history-db.service.ts

@@ -80,7 +80,7 @@ export class PairLiquidityInfoHistoryDbService {
           microBlockHash: data.microBlockHash,
         },
       },
-      update: {},
+      update: data,
       create: data,
     });
   }

src/database/pair-liquidity-info-history/pair-liquidity-info-history-v2-db.service.ts

@@ -0,0 +1,33 @@
+import { Injectable } from '@nestjs/common';
+import { PrismaService } from '../prisma.service';
+import { PairLiquidityInfoHistoryV2 } from '@prisma/client';
+
+@Injectable()
+export class PairLiquidityInfoHistoryV2DbService {
+  constructor(private prisma: PrismaService) {}
+
+  upsert(
+    data: Omit<PairLiquidityInfoHistoryV2, 'id' | 'updatedAt' | 'createdAt'>,
+  ) {
+    return this.prisma.pairLiquidityInfoHistoryV2.upsert({
+      where: {
+        pairIdMicroBlockHashLogIndexUniqueIndex: {
+          pairId: data.pairId,
+          microBlockHash: data.microBlockHash,
+          logIndex: data.logIndex,
+        },
+      },
+      update: data,
+      create: data,
+    });
+  }
+
+  getLastlySyncedLogByPairId(pairId: number) {
+    return this.prisma.pairLiquidityInfoHistoryV2.findFirst({
+      where: {
+        pairId,
+      },
+      orderBy: [{ microBlockTime: 'desc' }, { logIndex: 'desc' }],
+    });
+  }
+}

src/main.ts

@@ -2,7 +2,7 @@ import { NestFactory } from '@nestjs/core';
 import { AppModule } from './app.module';
 import { DocumentBuilder, SwaggerModule } from '@nestjs/swagger';
 import { nonNullable } from './lib/utils';
-import { PairSyncService } from './tasks/pair-sync.service';
+import { PairSyncService } from './tasks/pair-sync/pair-sync.service';
 
 const version = nonNullable(process.env.npm_package_version);
 async function bootstrap() {