[OTE-784] Limit addresses for compliance check to dydx wallets with d…

…eposit

indexer/packages/compliance/src/clients/elliptic-provider.ts

@@ -26,6 +26,8 @@ export const API_PATH: string = '/v2/wallet/synchronous';
 export const API_URI: string = `https://aml-api.elliptic.co${API_PATH}`;
 export const RISK_SCORE_KEY: string = 'risk_score';
 export const NO_RULES_TRIGGERED_RISK_SCORE: number = -1;
+// We use different negative values of risk score to represent different elliptic response states
+export const NOT_IN_BLOCKCHAIN_RISK_SCORE: number = -2;
 
 export class EllipticProviderClient extends ComplianceClient {
   private apiKey: string;
@@ -98,7 +100,7 @@ export class EllipticProviderClient extends ComplianceClient {
           `${config.SERVICE_NAME}.get_elliptic_risk_score.status_code`,
           { status: '404' },
         );
-        return NO_RULES_TRIGGERED_RISK_SCORE;
+        return NOT_IN_BLOCKCHAIN_RISK_SCORE;
       }
 
       if (error?.response?.status === 429) {

indexer/packages/compliance/src/index.ts

@@ -5,3 +5,4 @@ export * from './geoblocking/util';
 export * from './types';
 export * from './config';
 export * from './constants';
+export * from './clients/elliptic-provider';

indexer/packages/postgres/__tests__/stores/compliance-data-table.test.ts

@@ -1,5 +1,6 @@
 import { ComplianceDataFromDatabase, ComplianceProvider } from '../../src/types';
 import * as ComplianceDataTable from '../../src/stores/compliance-table';
+import * as WalletTable from '../../src/stores/wallet-table';
 import {
   clearData,
   migrate,
@@ -9,6 +10,7 @@ import {
   blockedComplianceData,
   blockedAddress,
   nonBlockedComplianceData,
+  defaultWallet,
 } from '../helpers/constants';
 import { DateTime } from 'luxon';
 
@@ -139,6 +141,29 @@ describe('Compliance data store', () => {
     expect(complianceData).toEqual(blockedComplianceData);
   });
 
+  it('Successfully filters by onlyDydxAddressWithDeposit', async () => {
+    // Create two compliance entries, one with a corresponding wallet entry and another without
+    await Promise.all([
+      WalletTable.create(defaultWallet),
+      ComplianceDataTable.create(nonBlockedComplianceData),
+      ComplianceDataTable.create({
+        ...nonBlockedComplianceData,
+        address: 'not_dydx_address',
+      }),
+    ]);
+
+    const complianceData: ComplianceDataFromDatabase[] = await ComplianceDataTable.findAll(
+      {
+        addressInWalletsTable: true,
+      },
+      [],
+      { readReplica: true },
+    );
+
+    expect(complianceData.length).toEqual(1);
+    expect(complianceData[0]).toEqual(nonBlockedComplianceData);
+  });
+
   it('Unable finds compliance data', async () => {
     const complianceData:
     ComplianceDataFromDatabase | undefined = await ComplianceDataTable.findByAddressAndProvider(

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

@@ -13,6 +13,7 @@ import {
 } from '../helpers/stores-helpers';
 import Transaction from '../helpers/transaction';
 import ComplianceDataModel from '../models/compliance-data-model';
+import WalletModel from '../models/wallet-model';
 import {
   ComplianceDataFromDatabase,
   ComplianceDataQueryConfig,
@@ -34,6 +35,7 @@ export async function findAll(
     provider,
     blocked,
     limit,
+    addressInWalletsTable,
   }: ComplianceDataQueryConfig,
   requiredFields: QueryableField[],
   options: Options = DEFAULT_POSTGRES_OPTIONS,
@@ -45,6 +47,7 @@ export async function findAll(
       provider,
       blocked,
       limit,
+      addressInWalletsTable,
     } as QueryConfig,
     requiredFields,
   );
@@ -70,6 +73,14 @@ export async function findAll(
     baseQuery = baseQuery.where(ComplianceDataColumns.blocked, blocked);
   }
 
+  if (addressInWalletsTable === true) {
+    baseQuery = baseQuery.innerJoin(
+      WalletModel.tableName,
+      `${ComplianceDataModel.tableName}.${ComplianceDataColumns.address}`,
+      '=',
+      `${WalletModel.tableName}.${WalletModel.idColumn}`);
+  }
+
   if (options.orderBy !== undefined) {
     for (const [column, order] of options.orderBy) {
       baseQuery = baseQuery.orderBy(

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

@@ -92,6 +92,7 @@ export enum QueryableField {
   REFEREE_ADDRESS = 'refereeAddress',
   KEY = 'key',
   TOKEN = 'token',
+  ADDRESS_IN_WALLETS_TABLE = 'addressInWalletsTable',
 }
 
 export interface QueryConfig {
@@ -291,6 +292,7 @@ export interface ComplianceDataQueryConfig extends QueryConfig {
   [QueryableField.UPDATED_BEFORE_OR_AT]?: string,
   [QueryableField.PROVIDER]?: string,
   [QueryableField.BLOCKED]?: boolean,
+  [QueryableField.ADDRESS_IN_WALLETS_TABLE]?: boolean,
 }
 
 export interface ComplianceStatusQueryConfig extends QueryConfig {

indexer/services/comlink/__tests__/controllers/api/v4/compliance-controller.test.ts

@@ -9,7 +9,11 @@ import {
 } from '@dydxprotocol-indexer/postgres';
 import { stats } from '@dydxprotocol-indexer/base';
 import { complianceProvider } from '../../../../src/helpers/compliance/compliance-clients';
-import { ComplianceClientResponse, INDEXER_COMPLIANCE_BLOCKED_PAYLOAD } from '@dydxprotocol-indexer/compliance';
+import {
+  ComplianceClientResponse,
+  INDEXER_COMPLIANCE_BLOCKED_PAYLOAD,
+  NOT_IN_BLOCKCHAIN_RISK_SCORE,
+} from '@dydxprotocol-indexer/compliance';
 import { ratelimitRedis } from '../../../../src/caches/rate-limiters';
 import { redis } from '@dydxprotocol-indexer/redis';
 import { DateTime } from 'luxon';
@@ -257,5 +261,33 @@ describe('compliance-controller#V4', () => {
         { provider: complianceProvider.provider },
       );
     });
+
+    it('GET /screen for invalid address does not upsert compliance data', async () => {
+      const invalidAddress: string = 'invalidAddress';
+      const notInBlockchainRiskScore: string = NOT_IN_BLOCKCHAIN_RISK_SCORE.toString();
+
+      jest.spyOn(complianceProvider.client, 'getComplianceResponse').mockImplementation(
+        (address: string): Promise<ComplianceClientResponse> => {
+          return Promise.resolve({
+            address,
+            blocked,
+            riskScore: notInBlockchainRiskScore,
+          });
+        },
+      );
+
+      const response: any = await sendRequest({
+        type: RequestMethod.GET,
+        path: `/v4/screen?address=${invalidAddress}`,
+      });
+
+      expect(response.body).toEqual({
+        restricted: false,
+        reason: undefined,
+      });
+
+      const data = await ComplianceTable.findAll({}, [], {});
+      expect(data).toHaveLength(0);
+    });
   });
 });

indexer/services/comlink/src/controllers/api/v4/compliance-controller.ts

@@ -1,5 +1,9 @@
 import { logger, stats, TooManyRequestsError } from '@dydxprotocol-indexer/base';
-import { ComplianceClientResponse, INDEXER_COMPLIANCE_BLOCKED_PAYLOAD } from '@dydxprotocol-indexer/compliance';
+import {
+  ComplianceClientResponse,
+  INDEXER_COMPLIANCE_BLOCKED_PAYLOAD,
+  NOT_IN_BLOCKCHAIN_RISK_SCORE,
+} from '@dydxprotocol-indexer/compliance';
 import { ComplianceDataCreateObject, ComplianceDataFromDatabase, ComplianceTable } from '@dydxprotocol-indexer/postgres';
 import express from 'express';
 import { checkSchema, matchedData } from 'express-validator';
@@ -85,6 +89,17 @@ export class ComplianceControllerHelper extends Controller {
       ComplianceClientResponse = await complianceProvider.client.getComplianceResponse(
         address,
       );
+      // Don't upsert invalid addresses (address causing ellitic error) to compliance table.
+      // When the elliptic request fails with 404, getComplianceResponse returns
+      // riskScore=NOT_IN_BLOCKCHAIN_RISK_SCORE
+      if (response.riskScore === undefined ||
+        Number(response.riskScore) === NOT_IN_BLOCKCHAIN_RISK_SCORE) {
+        return {
+          restricted: false,
+          reason: undefined,
+        };
+      }
+
       complianceData = await ComplianceTable.upsert({
         ..._.omitBy(response, _.isUndefined) as ComplianceDataCreateObject,
         provider: complianceProvider.provider,

indexer/services/roundtable/__tests__/tasks/update-compliance-data.test.ts

@@ -678,6 +678,66 @@ describe('update-compliance-data', () => {
 
     config.MAX_COMPLIANCE_DATA_QUERY_PER_LOOP = defaultMaxQueries;
   });
+
+  it('Only updates old addresses that are in wallets table', async () => {
+    const rogueWallet: string = 'address_not_in_wallets';
+    // Seed database with old compliance data, and set up subaccounts to not be active
+    // Create a compliance dataentry that is not in the wallets table
+    await Promise.all([
+      setupComplianceData(config.MAX_COMPLIANCE_DATA_AGE_SECONDS * 2),
+      setupInitialSubaccounts(config.MAX_ACTIVE_COMPLIANCE_DATA_AGE_SECONDS * 2),
+    ]);
+    await ComplianceTable.create({
+      ...testConstants.nonBlockedComplianceData,
+      address: rogueWallet,
+    });
+
+    const riskScore: string = '75.00';
+    setupMockProvider(
+      mockProvider,
+      { [testConstants.defaultAddress]: { blocked: true, riskScore } },
+    );
+
+    await updateComplianceDataTask(mockProvider);
+
+    const updatedCompliancnceData: ComplianceDataFromDatabase[] = await ComplianceTable.findAll({
+      address: [testConstants.defaultAddress],
+    }, [], {});
+    const unchangedComplianceData: ComplianceDataFromDatabase[] = await ComplianceTable.findAll({
+      address: [rogueWallet],
+    }, [], {});
+
+    expectUpdatedCompliance(
+      updatedCompliancnceData[0],
+      {
+        address: testConstants.defaultAddress,
+        blocked: true,
+        riskScore,
+      },
+      mockProvider.provider,
+    );
+    expectUpdatedCompliance(
+      unchangedComplianceData[0],
+      {
+        address: rogueWallet,
+        blocked: testConstants.nonBlockedComplianceData.blocked,
+        riskScore: testConstants.nonBlockedComplianceData.riskScore,
+      },
+      mockProvider.provider,
+    );
+    expectGaugeStats({
+      activeAddresses: 0,
+      newAddresses: 0,
+      oldAddresses: 1,
+      addressesScreened: 1,
+      upserted: 1,
+      statusUpserted: 1,
+      activeAddressesWithStaleCompliance: 0,
+      inactiveAddressesWithStaleCompliance: 1,
+    },
+    mockProvider.provider,
+    );
+  });
 });
 
 async function setupComplianceData(

indexer/services/roundtable/src/tasks/update-compliance-data.ts

@@ -1,7 +1,7 @@
 import {
   STATS_NO_SAMPLING, delay, logger, stats,
 } from '@dydxprotocol-indexer/base';
-import { ComplianceClientResponse } from '@dydxprotocol-indexer/compliance';
+import { ComplianceClientResponse, NOT_IN_BLOCKCHAIN_RISK_SCORE } from '@dydxprotocol-indexer/compliance';
 import {
   ComplianceDataColumns,
   ComplianceDataCreateObject,
@@ -151,6 +151,7 @@ export default async function runTask(
         blocked: false,
         provider: complianceProvider.provider,
         updatedBeforeOrAt: ageThreshold,
+        addressInWalletsTable: true,
       },
       [],
       { readReplica: true },
@@ -318,10 +319,19 @@ async function getComplianceData(
         return result.value;
       },
     ));
+    const addressNotFoundResponses:
+    PromiseFulfilledResult<ComplianceClientResponse>[] = successResponses.filter(
+      (result: PromiseSettledResult<ComplianceClientResponse>):
+      result is PromiseFulfilledResult<ComplianceClientResponse> => {
+        // riskScore = NOT_IN_BLOCKCHAIN_RISK_SCORE denotes elliptic 404 responses
+        return result.status === 'fulfilled' && result.value.riskScore === NOT_IN_BLOCKCHAIN_RISK_SCORE.toString();
+      },
+    );
 
     if (failedResponses.length > 0) {
       const addressesWithoutResponses: string[] = _.without(
         addresses,
+        // complianceResponses includes 404 responses
         ..._.map(complianceResponses, 'address'),
       );
       stats.increment(
@@ -337,6 +347,22 @@ async function getComplianceData(
         errors: failedResponses,
       });
     }
+
+    if (addressNotFoundResponses.length > 0) {
+      const notFoundAddresses = addressNotFoundResponses.map((result) => result.value.address);
+
+      stats.increment(
+        `${config.SERVICE_NAME}.${taskName}.get_compliance_data_404`,
+        1,
+        undefined,
+        { provider: complianceProvider.provider },
+      );
+      logger.error({
+        at: 'updated-compliance-data#getComplianceData',
+        message: 'Failed to retrieve compliance data for the addresses due to elliptic 404',
+        addresses: notFoundAddresses,
+      });
+    }
     stats.timing(
       `${config.SERVICE_NAME}.${taskName}.get_batch_compliance_data`,
       Date.now() - startBatch,