[IND-499]: Add AggregateTradingRewardsProcessedCache with roundtable …

…task (#927)

indexer/packages/base/src/date-helpers.ts

@@ -0,0 +1,23 @@
+/**
+ * Floor of the date to the timeInMilliseconds.
+ * For example with ONE_MINUTE_IN_MILLISECONDS, 12:02:33 will be rounded to 12:02:00
+ */
+export function floorDate(date: Date, timeInMilliseconds: number): Date {
+  return new Date(
+    Math.floor(
+      date.getTime() / timeInMilliseconds,
+    ) * timeInMilliseconds,
+  );
+}
+
+/**
+ * Ceiling of the date to the timeInMilliseconds.
+ * For example with ONE_MINUTE_IN_MILLISECONDS, 12:02:33 will be rounded to 12:03:00
+ */
+export function ceilingDate(date: Date, timeInMilliseconds: number): Date {
+  return new Date(
+    Math.ceil(
+      date.getTime() / timeInMilliseconds,
+    ) * timeInMilliseconds,
+  );
+}

indexer/packages/base/src/index.ts

@@ -12,6 +12,7 @@ export * from './axios';
 export * from './constants';
 export * from './bugsnag';
 export * from './stats-util';
+export * from './date-helpers';
 
 // Do this outside logger.ts to avoid a dependency cycle with logger transports that may trigger
 // additional logging.

indexer/packages/postgres/__tests__/helpers/mock-generators.ts

@@ -5,6 +5,7 @@ import * as MarketTable from '../../src/stores/market-table';
 import * as PerpetualMarketTable from '../../src/stores/perpetual-market-table';
 import * as SubaccountTable from '../../src/stores/subaccount-table';
 import * as TendermintEventTable from '../../src/stores/tendermint-event-table';
+import * as WalletTable from '../../src/stores/wallet-table';
 import {
   defaultAsset,
   defaultAsset2,
@@ -25,6 +26,7 @@ import {
   defaultTendermintEvent2,
   defaultTendermintEvent3,
   defaultTendermintEvent4,
+  defaultWallet,
 } from './constants';
 
 export async function seedData() {
@@ -61,4 +63,7 @@ export async function seedData() {
     AssetTable.create(defaultAsset2),
     AssetTable.create(defaultAsset3),
   ]);
+  await Promise.all([
+    WalletTable.create(defaultWallet),
+  ]);
 }

indexer/packages/postgres/__tests__/stores/trading-reward-aggregation-table.test.ts

@@ -1,13 +1,13 @@
-import { TradingRewardAggregationFromDatabase } from '../../src/types';
+import { TradingRewardAggregationFromDatabase, TradingRewardAggregationPeriod } from '../../src/types';
 import { clearData, migrate, teardown } from '../../src/helpers/db-helpers';
 import {
+  createdDateTime,
   defaultSubaccountId,
   defaultTradingRewardAggregation,
   defaultTradingRewardAggregationId,
-  defaultWallet,
 } from '../helpers/constants';
 import * as TradingRewardAggregationTable from '../../src/stores/trading-reward-aggregation-table';
-import { WalletTable } from '../../src';
+import { BlockTable } from '../../src';
 import { seedData } from '../helpers/mock-generators';
 import { denomToHumanReadableConversion } from '../helpers/conversion-helpers';
 
@@ -18,7 +18,6 @@ describe('TradingRewardAggregation store', () => {
 
   beforeEach(async () => {
     await seedData();
-    await WalletTable.create(defaultWallet);
   });
 
   afterEach(async () => {
@@ -59,6 +58,39 @@ describe('TradingRewardAggregation store', () => {
     );
   });
 
+  it('Successfully finds latest monthly TradingRewardAggregation', async () => {
+    await Promise.all([
+      BlockTable.create({
+        blockHeight: '100',
+        time: createdDateTime.toISO(),
+      }),
+    ]);
+
+    await Promise.all([
+      TradingRewardAggregationTable.create({
+        ...defaultTradingRewardAggregation,
+        period: TradingRewardAggregationPeriod.MONTHLY,
+      }),
+      TradingRewardAggregationTable.create({
+        ...defaultTradingRewardAggregation,
+        startedAtHeight: '100',
+        period: TradingRewardAggregationPeriod.MONTHLY,
+      }),
+    ]);
+
+    const tradingRewardAggregation:
+    TradingRewardAggregationFromDatabase | undefined = await TradingRewardAggregationTable
+      .getLatestAggregatedTradeReward(TradingRewardAggregationPeriod.MONTHLY);
+
+    expect(tradingRewardAggregation).toEqual(
+      expect.objectContaining({
+        ...defaultTradingRewardAggregation,
+        startedAtHeight: '100',
+        period: TradingRewardAggregationPeriod.MONTHLY,
+      }),
+    );
+  });
+
   it('Successfully finds a TradingRewardAggregation', async () => {
     await TradingRewardAggregationTable.create(defaultTradingRewardAggregation);
 
@@ -101,4 +133,44 @@ describe('TradingRewardAggregation store', () => {
       amount,
     });
   });
+
+  it('Successfully deleted trading reward aggregations after a certain height', async () => {
+    await Promise.all([
+      BlockTable.create({
+        blockHeight: '100',
+        time: createdDateTime.toISO(),
+      }),
+      BlockTable.create({
+        blockHeight: '101',
+        time: createdDateTime.toISO(),
+      }),
+    ]);
+
+    await Promise.all([
+      TradingRewardAggregationTable.create(defaultTradingRewardAggregation),
+      TradingRewardAggregationTable.create({
+        ...defaultTradingRewardAggregation,
+        startedAtHeight: '100',
+      }),
+      TradingRewardAggregationTable.create({
+        ...defaultTradingRewardAggregation,
+        startedAtHeight: '101',
+      }),
+    ]);
+
+    await TradingRewardAggregationTable.deleteAll({
+      startedAtHeightOrAfter: '100',
+    });
+
+    const tradingRewardAggregations:
+    TradingRewardAggregationFromDatabase[] = await TradingRewardAggregationTable.findAll(
+      {},
+      [],
+    );
+
+    expect(tradingRewardAggregations.length).toEqual(1);
+    expect(tradingRewardAggregations[0]).toEqual(
+      expect.objectContaining(defaultTradingRewardAggregation),
+    );
+  });
 });

indexer/packages/postgres/__tests__/stores/wallet-table.test.ts

@@ -1,6 +1,6 @@
 import { WalletFromDatabase } from '../../src/types';
 import { clearData, migrate, teardown } from '../../src/helpers/db-helpers';
-import { defaultAddress, defaultWallet } from '../helpers/constants';
+import { defaultWallet2 } from '../helpers/constants';
 import * as WalletTable from '../../src/stores/wallet-table';
 
 describe('Wallet store', () => {
@@ -17,27 +17,25 @@ describe('Wallet store', () => {
   });
 
   it('Successfully creates a Wallet', async () => {
-    await WalletTable.create(defaultWallet);
+    await WalletTable.create(defaultWallet2);
   });
 
   it('Successfully upserts a Wallet multiple times', async () => {
-    await WalletTable.upsert(defaultWallet);
+    await WalletTable.upsert(defaultWallet2);
     let wallet: WalletFromDatabase | undefined = await WalletTable.findById(
-      defaultAddress,
+      defaultWallet2.address,
     );
 
-    expect(wallet).toEqual(expect.objectContaining(defaultWallet));
-    await WalletTable.upsert(defaultWallet);
-    wallet = await WalletTable.findById(
-      defaultAddress,
-    );
+    expect(wallet).toEqual(expect.objectContaining(defaultWallet2));
+    await WalletTable.upsert(defaultWallet2);
+    wallet = await WalletTable.findById(defaultWallet2.address);
 
-    expect(wallet).toEqual(expect.objectContaining(defaultWallet));
+    expect(wallet).toEqual(expect.objectContaining(defaultWallet2));
   });
 
   it('Successfully finds all Wallets', async () => {
     await Promise.all([
-      WalletTable.create(defaultWallet),
+      WalletTable.create(defaultWallet2),
       WalletTable.create({
         address: 'fake_address',
         totalTradingRewards: '0',
@@ -51,19 +49,19 @@ describe('Wallet store', () => {
     );
 
     expect(wallets.length).toEqual(2);
-    expect(wallets[0]).toEqual(expect.objectContaining(defaultWallet));
+    expect(wallets[0]).toEqual(expect.objectContaining(defaultWallet2));
     expect(wallets[1]).toEqual(expect.objectContaining({
       address: 'fake_address',
     }));
   });
 
   it('Successfully finds a Wallet', async () => {
-    await WalletTable.create(defaultWallet);
+    await WalletTable.create(defaultWallet2);
 
     const wallet: WalletFromDatabase | undefined = await WalletTable.findById(
-      defaultAddress,
+      defaultWallet2.address,
     );
 
-    expect(wallet).toEqual(expect.objectContaining(defaultWallet));
+    expect(wallet).toEqual(expect.objectContaining(defaultWallet2));
   });
 });

indexer/packages/postgres/src/models/trading-reward-aggregation-model.ts

@@ -51,7 +51,7 @@ export default class TradingRewardAggregationModel extends Model {
         address: { type: 'string' },
         startedAt: { type: 'string', format: 'date-time' }, // Inclusive
         startedAtHeight: { type: 'string', pattern: IntegerPattern }, // Inclusive
-        endedAt: { type: ['string', 'null'], format: 'date-time' }, // Inclusive
+        endedAt: { type: ['string', 'null'], format: 'date-time' }, // Exclusive
         endedAtHeight: { type: ['string', 'null'], pattern: IntegerPattern }, // Inclusive
         period: { type: 'string', enum: [...Object.values(TradingRewardAggregationPeriod)] },
         amount: { type: 'string', pattern: NonNegativeNumericPattern },

indexer/packages/postgres/src/stores/trading-reward-aggregation-table.ts

@@ -92,6 +92,22 @@ export async function findAll(
   return baseQuery.returning('*');
 }
 
+export async function getLatestAggregatedTradeReward(
+  period: TradingRewardAggregationPeriod,
+  options: Options = DEFAULT_POSTGRES_OPTIONS,
+): Promise<TradingRewardAggregationFromDatabase | undefined> {
+  const baseQuery:
+  QueryBuilder<TradingRewardAggregationModel> = setupBaseQuery<TradingRewardAggregationModel>(
+    TradingRewardAggregationModel,
+    options,
+  );
+
+  return baseQuery
+    .where(TradingRewardAggregationColumns.period, period)
+    .orderBy(TradingRewardAggregationColumns.startedAtHeight, Ordering.DESC)
+    .first();
+}
+
 export async function create(
   aggregationToCreate: TradingRewardAggregationCreateObject,
   options: Options = { txId: undefined },
@@ -126,7 +142,7 @@ export async function update(
   {
     ...fields
   }: TradingRewardAggregationUpdateObject,
-  options: Options = { txId: undefined },
+  options: Options = DEFAULT_POSTGRES_OPTIONS,
 ): Promise<TradingRewardAggregationFromDatabase | undefined> {
   const aggregation = await TradingRewardAggregationModel.query(
     Transaction.get(options.txId),
@@ -136,3 +152,27 @@ export async function update(
   // The objection types mistakenly think the query returns an array of orders.
   return aggregation as unknown as (TradingRewardAggregationFromDatabase | undefined);
 }
+
+export async function deleteAll(
+  {
+    period,
+    startedAtHeightOrAfter,
+  }: TradingRewardAggregationQueryConfig,
+  options: Options = DEFAULT_POSTGRES_OPTIONS,
+): Promise<void> {
+  let baseQuery:
+  QueryBuilder<TradingRewardAggregationModel> = setupBaseQuery<TradingRewardAggregationModel>(
+    TradingRewardAggregationModel,
+    options,
+  );
+
+  if (period) {
+    baseQuery = baseQuery.where(TradingRewardAggregationColumns.period, period);
+  }
+
+  if (startedAtHeightOrAfter) {
+    baseQuery = baseQuery.where(TradingRewardAggregationColumns.startedAtHeight, '>=', startedAtHeightOrAfter);
+  }
+
+  await baseQuery.delete();
+}

indexer/packages/postgres/src/stores/wallet-table.ts

@@ -1,4 +1,4 @@
-import { QueryBuilder } from 'objection';
+import { PartialModelObject, QueryBuilder } from 'objection';
 
 import { DEFAULT_POSTGRES_OPTIONS } from '../constants';
 import { setupBaseQuery, verifyAllRequiredFields } from '../helpers/stores-helpers';
@@ -13,6 +13,7 @@ import {
   WalletCreateObject,
   WalletFromDatabase,
   WalletQueryConfig,
+  WalletUpdateObject,
 } from '../types';
 
 export async function findAll(
@@ -70,6 +71,21 @@ export async function create(
   ).insert(walletToCreate).returning('*');
 }
 
+export async function update(
+  {
+    address,
+    ...fields
+  }: WalletUpdateObject,
+  options: Options = { txId: undefined },
+): Promise<WalletFromDatabase | undefined> {
+  const wallet = await WalletModel.query(
+    Transaction.get(options.txId),
+  ).findById(address);
+  const updatedWallet = await wallet.$query().patch(fields as PartialModelObject<WalletModel>).returning('*');
+  // The objection types mistakenly think the query returns an array of Wallets.
+  return updatedWallet as unknown as (WalletFromDatabase | undefined);
+}
+
 export async function upsert(
   walletToUpsert: WalletCreateObject,
   options: Options = { txId: undefined },

indexer/packages/postgres/src/types/query-types.ts

@@ -75,6 +75,7 @@ export enum QueryableField {
   BLOCK_TIME_BEFORE_OR_AT = 'blockTimeBeforeOrAt',
   STARTED_AT_HEIGHT = 'startedAtHeight',
   PERIOD = 'period',
+  STARTED_AT_HEIGHT_OR_AFTER = 'startedAtHeightOrAfter',
 }
 
 export interface QueryConfig {
@@ -279,5 +280,6 @@ export interface TradingRewardQueryConfig extends QueryConfig {
 export interface TradingRewardAggregationQueryConfig extends QueryConfig {
   [QueryableField.ADDRESS]?: string;
   [QueryableField.STARTED_AT_HEIGHT]?: string;
+  [QueryableField.STARTED_AT_HEIGHT_OR_AFTER]?: string;
   [QueryableField.PERIOD]?: TradingRewardAggregationPeriod;
 }

indexer/packages/postgres/src/types/wallet-types.ts

@@ -5,6 +5,11 @@ export interface WalletCreateObject {
   totalTradingRewards: string,
 }
 
+export interface WalletUpdateObject {
+  address: string,
+  totalTradingRewards: string,
+}
+
 export enum WalletColumns {
   address = 'address',
   totalTradingRewards = 'totalTradingRewards',