add cluster matching adapters

migration/1728554628004-AddEstimatedClusterMatching.ts

@@ -1,34 +1,34 @@
-import {MigrationInterface, QueryRunner} from "typeorm";
+import { MigrationInterface, QueryRunner } from 'typeorm';
 
-export class AddEstimatedClusterMatching1728554628004 implements MigrationInterface {
-
-    public async up(queryRunner: QueryRunner): Promise<void> {
-        await queryRunner.query(`
+export class AddEstimatedClusterMatching1728554628004
+  implements MigrationInterface
+{
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
             CREATE TABLE estimated_cluster_matching (
               id SERIAL PRIMARY KEY,
               project_id INT NOT NULL,
               qf_round_id INT NOT NULL,
               matching DOUBLE PRECISION NOT NULL
             );
           `);
-      
-          // Create indexes on the new table
-          await queryRunner.query(`
+
+    // Create indexes on the new table
+    await queryRunner.query(`
             CREATE INDEX estimated_cluster_matching_project_id_qfround_id
             ON estimated_cluster_matching (project_id, qf_round_id);
           `);
-      
-          await queryRunner.query(`
+
+    await queryRunner.query(`
             CREATE INDEX estimated_cluster_matching_matching
             ON estimated_cluster_matching (matching);
           `);
-    }
+  }
 
-    public async down(queryRunner: QueryRunner): Promise<void> {
-        // Revert changes if necessary by dropping the table and restoring the view
-      await queryRunner.query(`
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    // Revert changes if necessary by dropping the table and restoring the view
+    await queryRunner.query(`
         DROP TABLE IF EXISTS estimated_cluster_matching;
       `);
-    }
-
+  }
 }

src/adapters/adaptersFactory.ts

@@ -22,6 +22,9 @@ import { DonationSaveBackupMockAdapter } from './donationSaveBackup/DonationSave
 import { SuperFluidAdapter } from './superFluid/superFluidAdapter';
 import { SuperFluidMockAdapter } from './superFluid/superFluidMockAdapter';
 import { SuperFluidAdapterInterface } from './superFluid/superFluidAdapterInterface';
+import { CocmAdapter } from './cocmAdapter/cocmAdapter';
+import { CocmMockAdapter } from './cocmAdapter/cocmMockAdapter';
+import { CocmAdapterInterface } from './cocmAdapter/cocmAdapterInterface';
 
 const discordAdapter = new DiscordAdapter();
 const googleAdapter = new GoogleAdapter();
@@ -147,3 +150,17 @@ export const getSuperFluidAdapter = (): SuperFluidAdapterInterface => {
       return superFluidMockAdapter;
   }
 };
+
+const clusterMatchingAdapter = new CocmAdapter();
+const clusterMatchingMockAdapter = new CocmMockAdapter();
+
+export const getClusterMatchingAdapter = (): CocmAdapterInterface => {
+  switch (process.env.CLUSTER_MATCHING_ADAPTER) {
+    case 'clusterMatching':
+      return clusterMatchingAdapter;
+    case 'mock':
+      return clusterMatchingMockAdapter;
+    default:
+      return clusterMatchingMockAdapter;
+  }
+};

src/adapters/cocmAdapter/cocmAdapter.ts

@@ -0,0 +1,46 @@
+import axios from 'axios';
+import {
+  CocmAdapterInterface,
+  EstimatedMatchingInput,
+  ProjectsEstimatedMatchings,
+} from './cocmAdapterInterface';
+import { logger } from '../../utils/logger';
+import { i18n, translationErrorMessagesKeys } from '../../utils/errorMessages';
+
+export class CocmAdapter implements CocmAdapterInterface {
+  private ClusterMatchingURL;
+
+  constructor() {
+    this.ClusterMatchingURL =
+      process.env.CLUSTER_MATCHING_API_URL || 'localhost';
+  }
+
+  async fetchEstimatedClusterMatchings(
+    matchingDataInput: EstimatedMatchingInput,
+  ): Promise<ProjectsEstimatedMatchings> {
+    try {
+      const result = await axios.post(
+        this.ClusterMatchingURL,
+        matchingDataInput,
+        {
+          headers: {
+            Accept: 'application/json',
+            'Content-Type': 'application/json',
+          },
+        },
+      );
+      if (result?.data?.error !== null) {
+        logger.error('clusterMatchingApi error', result.data.error);
+        throw new Error(
+          i18n.__(translationErrorMessagesKeys.CLUSTER_MATCHING_API_ERROR),
+        );
+      }
+      return result.data;
+    } catch (e) {
+      logger.error('clusterMatchingApi error', e);
+      throw new Error(
+        i18n.__(translationErrorMessagesKeys.CLUSTER_MATCHING_API_ERROR),
+      );
+    }
+  }
+}

src/adapters/cocmAdapter/cocmAdapterInterface.ts

@@ -0,0 +1,49 @@
+// Example Data
+// {
+// 	"matching_data": [
+// 		{
+// 			"matching_amount": 83.25,
+// 			"matching_percent": 50.0,
+// 			"project_name": "Test1",
+// 			"strategy": "COCM"
+// 		},
+// 		{
+// 			"matching_amount": 83.25,
+// 			"matching_percent": 50.0,
+// 			"project_name": "Test3",
+// 			"strategy": "COCM"
+// 		}
+// 	]
+// }
+
+export interface ProjectsEstimatedMatchings {
+  matching_data: {
+    matching_amount: number;
+    matching_percent: number;
+    project_name: string;
+    strategy: string;
+  }[];
+}
+
+export interface EstimatedMatchingInput {
+  votes_data: [
+    {
+      voter: string;
+      payoutAddress: string;
+      amountUSD: number;
+      project_name: string;
+      score: number;
+    },
+  ];
+  strategy: string;
+  min_donation_threshold_amount: number;
+  matching_cap_amount: number;
+  matching_amount: number;
+  passport_threshold: number;
+}
+
+export interface CocmAdapterInterface {
+  fetchEstimatedClusterMatchings(
+    matchingDataInput: EstimatedMatchingInput,
+  ): Promise<ProjectsEstimatedMatchings>;
+}

src/adapters/cocmAdapter/cocmMockAdapter.ts

@@ -0,0 +1,30 @@
+import axios from 'axios';
+import {
+  CocmAdapterInterface,
+  ProjectsEstimatedMatchings,
+} from './cocmAdapterInterface';
+import { i18n, translationErrorMessagesKeys } from '../../utils/errorMessages';
+import { logger } from '../../utils/logger';
+
+export class CocmMockAdapter implements CocmAdapterInterface {
+  async fetchEstimatedClusterMatchings(
+    _matchingDataInput,
+  ): Promise<ProjectsEstimatedMatchings> {
+    return {
+      matching_data: [
+        {
+          matching_amount: 83.25,
+          matching_percent: 50.0,
+          project_name: 'Test1',
+          strategy: 'COCM',
+        },
+        {
+          matching_amount: 83.25,
+          matching_percent: 50.0,
+          project_name: 'Test3',
+          strategy: 'COCM',
+        },
+      ],
+    };
+  }
+}

src/server/bootstrap.ts

@@ -339,20 +339,6 @@ export async function bootstrap() {
         logger.error('Enabling power boosting snapshot ', e);
       }
     }
-
-    if (!isTestEnv) {
-      // They will fail in test env, because we run migrations after bootstrap so refreshing them will cause this error
-      // relation "project_estimated_matching_view" does not exist
-      logger.debug(
-        'continueDbSetup() before refreshProjectEstimatedMatchingView() ',
-        new Date(),
-      );
-      await refreshProjectEstimatedMatchingView();
-      logger.debug(
-        'continueDbSetup() after refreshProjectEstimatedMatchingView() ',
-        new Date(),
-      );
-    }
     logger.debug('continueDbSetup() end of function', new Date());
   }
 

src/services/cronJobs/syncEstimatedClusterMatchingJob.ts

@@ -0,0 +1,50 @@
+import { schedule } from 'node-cron';
+import { spawn, Worker, Thread } from 'threads';
+import config from '../../config';
+import { logger } from '../../utils/logger';
+import {
+  findActiveQfRound,
+  findUsersWithoutMBDScoreInActiveAround,
+} from '../../repositories/qfRoundRepository';
+import { findUserById } from '../../repositories/userRepository';
+import { UserQfRoundModelScore } from '../../entities/userQfRoundModelScore';
+
+const cronJobTime =
+  (config.get('SYNC_ESTIMATED_CLUSTED_MATCHING_CRONJOB_EXPRESSION') as string) ||
+  '0 * * * * *';
+
+export const runSyncEstimatedClusterMatchingCronjob = () => {
+  logger.debug(
+    'runSyncEstimatedClusterMatchingCronjob() has been called, cronJobTime',
+    cronJobTime,
+  );
+  schedule(cronJobTime, async () => {
+    await fetchAndUpdateClusterEstimatedMatching();
+  });
+};
+
+export const fetchAndUpdateClusterEstimatedMatching = async () => {
+  const fetchWorker = await spawn(
+    new Worker('../../workers/cocm/fetchEstimatedClusterMtchingWorker'),
+  );
+
+  const updateWorker = await spawn(
+    new Worker('../../workers/cocm/updateProjectsEstimatedClusterMatchingWorker')
+  );
+  const activeQfRoundId =
+    (await findActiveQfRound())?.id;
+  if (!activeQfRoundId || activeQfRoundId === 0) return;
+
+  for (const projectId of []) {
+    try {
+
+    //   const userScore = await worker.syncUserScore({
+    //     userWallet: user?.walletAddress,
+    //   });
+    } catch (e) {
+      logger.info(`User with Id ${1} did not sync MBD score this batch`);
+    }
+  }
+  await Thread.terminate(fetchWorker);
+  await Thread.terminate(updateWorker);
+};

src/utils/errorMessages.ts

@@ -19,6 +19,7 @@ export const setI18nLocaleForRequest = async (req, _res, next) => {
 };
 
 export const errorMessages = {
+  CLUSTER_MATCHING_API_ERROR: 'Error in the cluster matching api, check logs',
   FIAT_DONATION_ALREADY_EXISTS: 'Onramper donation already exists',
   CAMPAIGN_NOT_FOUND: 'Campaign not found',
   QF_ROUND_NOT_FOUND: 'qf round not found',
@@ -208,6 +209,7 @@ export const errorMessages = {
 };
 
 export const translationErrorMessagesKeys = {
+  CLUSTER_MATCHING_API_ERROR: 'CLUSTER_MATCHING_API_ERROR',
   GITCOIN_ERROR_FETCHING_DATA: 'GITCOIN_ERROR_FETCHING_DATA',
   TX_NOT_FOUND: 'TX_NOT_FOUND',
   INVALID_PROJECT_ID: 'INVALID_PROJECT_ID',

src/workers/cocm/fetchEstimatedClusterMatchingWorker.ts

@@ -0,0 +1,17 @@
+// workers/auth.js
+import { expose } from 'threads/worker';
+import { WorkerModule } from 'threads/dist/types/worker';
+import { getClusterMatchingAdapter } from '../../adapters/adaptersFactory';
+
+type FetchEstimatedClusterMatchingWorkerFunctions = 'fetchEstimatedClusterMatching';
+
+export type FetchEstimatedClusterMatchingWorker =
+  WorkerModule<FetchEstimatedClusterMatchingWorkerFunctions>;
+
+const worker: FetchEstimatedClusterMatchingWorker = {
+  async fetchEstimatedClusterMatching(matchingDataInput: any) {
+    return await getClusterMatchingAdapter().fetchEstimatedClusterMatchings(matchingDataInput);
+  },
+};
+
+expose(worker);