Merge pull request #10899 from hicommonwealth/tim/event-migration-fixes

libs/core/src/integration/events.utils.ts

@@ -42,6 +42,7 @@ type EvmMapper<Input extends string, Output extends ZodSchema> = {
   mapEvmToSchema: (
     contestAddress: string | null,
     evmInput: ParseSignature<Input>,
+    blockNumber: number,
   ) => {
     event_name: Events;
     event_payload: z.infer<Output>;
@@ -58,12 +59,14 @@ const RecurringContestManagerDeployedMapper: EvmMapper<
   mapEvmToSchema: (
     contestAddress,
     { contest, namespace, interval, oneOff: _ },
+    blockNumber,
   ) => ({
     event_name: 'RecurringContestManagerDeployed',
     event_payload: {
       contest_address: contest,
       namespace: namespace,
       interval: BigNumber.from(interval).toNumber(),
+      block_number: blockNumber,
     },
   }),
 };
@@ -78,12 +81,14 @@ const OneOffContestManagerDeployedMapper: EvmMapper<
   mapEvmToSchema: (
     contestAddress,
     { contest, namespace, interval, oneOff: _ },
+    blockNumber,
   ) => ({
     event_name: 'OneOffContestManagerDeployed',
     event_payload: {
       contest_address: contest,
       namespace: namespace,
       length: BigNumber.from(interval).toNumber(),
+      block_number: blockNumber,
     },
   }),
 };
@@ -244,6 +249,7 @@ export const parseEvmEventToContestEvent = <
   chainEventName: Event,
   contestAddress: string | null,
   evmParsedArgs: Result,
+  blockNumber: number,
 ): ContestOutboxEvent<Event> => {
   const m = EvmMappers[chainEventName];
   if (!m) {
@@ -254,7 +260,7 @@ export const parseEvmEventToContestEvent = <
   for (const mapper of mappers) {
     const evmInput = parseEthersResult(mapper.signature, evmParsedArgs);
     if (!mapper.condition || mapper.condition(evmInput)) {
-      return mapper.mapEvmToSchema(contestAddress, evmInput);
+      return mapper.mapEvmToSchema(contestAddress, evmInput, blockNumber);
     }
   }
   throw new Error(`No valid mapper found for event: ${chainEventName}`);

libs/core/test/utils/parseEvmEventToContestEvent.spec.ts

@@ -17,6 +17,7 @@ describe('parseEvmEventToContestEvent', () => {
         ethers.BigNumber.from(7), // interval
         false, // oneOff
       ],
+      1,
     );
     expect(event_name).to.eq('RecurringContestManagerDeployed');
     const parsedEvent =
@@ -37,6 +38,7 @@ describe('parseEvmEventToContestEvent', () => {
         ethers.BigNumber.from(7), // interval is same as length
         true, // oneOff
       ],
+      1,
     );
     expect(event_name).to.eq('OneOffContestManagerDeployed');
     const parsedEvent =
@@ -56,6 +58,7 @@ describe('parseEvmEventToContestEvent', () => {
         ethers.BigNumber.from(1000), // startTime
         ethers.BigNumber.from(1001), // endTime
       ],
+      1,
     );
     expect(event_name).to.eq('ContestStarted');
     const parsedEvent = events.ContestStarted.parse(event_payload);
@@ -79,6 +82,7 @@ describe('parseEvmEventToContestEvent', () => {
         '0x1', // creator
         '/threads/1', // url
       ],
+      1,
     );
     expect(event_name).to.eq('ContestContentAdded');
     const parsedEvent = events.ContestContentAdded.parse(event_payload);
@@ -99,6 +103,7 @@ describe('parseEvmEventToContestEvent', () => {
         ethers.BigNumber.from(888), // contestId
         ethers.BigNumber.from(9000), // votingPower
       ],
+      1,
     );
     expect(event_name).to.eq('ContestContentUpvoted');
     const parsedEvent = events.ContestContentUpvoted.parse(event_payload);
@@ -119,6 +124,7 @@ describe('parseEvmEventToContestEvent', () => {
         ethers.BigNumber.from(10), // contentId
         ethers.BigNumber.from(9000), // votingPower
       ],
+      1,
     );
     expect(event_name).to.eq('ContestContentUpvoted');
     const parsedEvent = events.ContestContentUpvoted.parse(event_payload);
@@ -131,9 +137,14 @@ describe('parseEvmEventToContestEvent', () => {
 
   test('should throw if the wrong number of args are used outbox shape', () => {
     expect(() => {
-      parseEvmEventToContestEvent('VoterVotedRecurring', contestAddress, [
-        '0x2', // voterAddress
-      ]);
+      parseEvmEventToContestEvent(
+        'VoterVotedRecurring',
+        contestAddress,
+        [
+          '0x2', // voterAddress
+        ],
+        1,
+      );
     }).to.throw('evm parsed args does not match signature');
   });
 });

libs/evm-protocols/src/common-protocol/utils.ts

@@ -50,6 +50,17 @@ export const getBlock = async ({
   };
 };
 
+export const getBlockNumber = async ({
+  evmClient,
+  rpc,
+}: {
+  evmClient?: EvmClientType;
+  rpc: string;
+}): Promise<number> => {
+  const web3 = evmClient || new Web3(rpc);
+  return Number(await web3.eth.getBlockNumber());
+};
+
 export const getTransactionReceipt = async ({
   evmClient,
   rpc,

libs/model/src/contest/Contests.projection.ts

@@ -53,6 +53,7 @@ async function updateOrCreateWithAlert(
   contest_address: string,
   interval: number,
   isOneOff: boolean,
+  blockNumber: number,
 ) {
   const community = await models.Community.findOne({
     where: { namespace_address: namespace },
@@ -157,7 +158,7 @@ async function updateOrCreateWithAlert(
           event_signature: eventSignature,
           contract_name: childContractName,
           parent_contract_address: cp.factoryContracts[ethChainId].factory,
-          // TODO: add created_at_block so EVM CE runs the migrateEvents func
+          created_at_block: blockNumber,
         };
       },
     );
@@ -244,6 +245,7 @@ export function Contests(): Projection<typeof inputs> {
           payload.contest_address,
           payload.interval,
           false,
+          payload.block_number,
         );
       },
 
@@ -254,6 +256,7 @@ export function Contests(): Projection<typeof inputs> {
           payload.contest_address,
           0,
           true,
+          payload.block_number,
         );
       },
 

libs/model/src/models/evmEventSource.ts

@@ -33,8 +33,12 @@ export default (
         type: Sequelize.STRING,
         allowNull: false,
       },
-      created_at_block: { type: Sequelize.INTEGER, allowNull: true },
-      events_migrated: { type: Sequelize.BOOLEAN, allowNull: true },
+      created_at_block: { type: Sequelize.INTEGER, allowNull: false },
+      events_migrated: {
+        type: Sequelize.BOOLEAN,
+        allowNull: false,
+        defaultValue: false,
+      },
     },
     {
       tableName: 'EvmEventSources',

libs/model/test/contest/contests-projection-lifecycle.spec.ts

@@ -211,6 +211,7 @@ describe('Contests projection lifecycle', () => {
         namespace,
         contest_address: recurring,
         interval: 10,
+        block_number: 1,
       },
     });
 
@@ -231,6 +232,7 @@ describe('Contests projection lifecycle', () => {
         namespace,
         contest_address: oneoff,
         length: 1,
+        block_number: 1,
       },
     });
 

libs/schemas/src/entities/evm-event-source.schemas.ts

@@ -8,8 +8,6 @@ export const EvmEventSource = z.object({
   event_signature: z.string(),
   contract_name: z.nativeEnum(ChildContractNames),
   parent_contract_address: EVM_ADDRESS,
-
-  // TODO: this should be required
-  created_at_block: z.number().optional(),
+  created_at_block: z.number(),
   events_migrated: z.boolean().optional(),
 });

libs/schemas/src/events/events.schemas.ts

@@ -280,12 +280,22 @@ export const events = {
       .int()
       .positive()
       .describe('Recurring constest interval'),
+    block_number: z
+      .number()
+      .int()
+      .positive()
+      .describe('The block number in which the contest was created'),
   }).describe('When a new recurring contest manager gets deployed'),
 
   OneOffContestManagerDeployed: EventMetadata.extend({
     namespace: z.string().describe('Community namespace'),
     contest_address: z.string().describe('Contest manager address'),
     length: z.number().int().positive().describe('Length of contest in days'),
+    block_number: z
+      .number()
+      .int()
+      .positive()
+      .describe('The block number in which the contest was created'),
   }).describe('When a new one-off contest manager gets deployed'),
 
   // Contest Events

packages/commonwealth/server/migrations/20250207092628-migrated-evm-sources.js

@@ -0,0 +1,48 @@
+'use strict';
+
+/** @type {import('sequelize-cli').Migration} */
+module.exports = {
+  async up(queryInterface, Sequelize) {
+    return queryInterface.sequelize.transaction(async (transaction) => {
+      await queryInterface.sequelize.query(
+        `
+        UPDATE "EvmEventSources"
+        SET created_at_block = 1,
+            events_migrated = true;
+      `,
+        { transaction },
+      );
+      await queryInterface.sequelize.query(
+        `
+        ALTER TABLE "EvmEventSources"
+        ALTER COLUMN created_at_block SET NOT NULL,
+        ALTER COLUMN events_migrated SET NOT NULL,
+        ALTER COLUMN events_migrated SET DEFAULT FALSE;
+      `,
+        { transaction },
+      );
+    });
+  },
+
+  async down(queryInterface, Sequelize) {
+    return queryInterface.sequelize.transaction(async (transaction) => {
+      await queryInterface.sequelize.query(
+        `
+            ALTER TABLE "EvmEventSources"
+                ALTER COLUMN created_at_block DROP NOT NULL,
+                ALTER COLUMN events_migrated DROP NOT NULL,
+                ALTER COLUMN events_migrated DROP DEFAULT;
+        `,
+        { transaction },
+      );
+      await queryInterface.sequelize.query(
+        `
+        UPDATE "EvmEventSources"
+        SET created_at_block = NULL,
+            events_migrated = NULL;
+      `,
+        { transaction },
+      );
+    });
+  },
+};

packages/commonwealth/server/workers/evmChainEvents/logProcessing.ts

@@ -234,8 +234,11 @@ export async function getEvents(
 export async function migrateEvents(
   evmSource: EvmSource,
   endingBlockNum: number,
-): Promise<{ events: EvmEvent[]; lastBlockNum: number } | undefined> {
-  let oldestBlock: number;
+): Promise<
+  | { events: EvmEvent[]; lastBlockNum: number; contracts: ContractSources }
+  | { contracts: ContractSources }
+> {
+  let oldestBlock: number | undefined;
   const contracts: ContractSources = {};
   for (const [contractAddress, abiSignature] of Object.entries(
     evmSource.contracts,
@@ -249,34 +252,35 @@ export async function migrateEvents(
           };
         }
         contracts[contractAddress].sources.push(source);
-        // @ts-expect-error StrictNullChecks
         if (!oldestBlock || oldestBlock > source.created_at_block) {
           oldestBlock = source.created_at_block;
         }
       }
     }
   }
 
-  if (Object.keys(contracts).length > 0) {
+  if (Object.keys(contracts).length > 0 && oldestBlock) {
     const result = await getEvents(
       {
         rpc: evmSource.rpc,
         maxBlockRange: evmSource.maxBlockRange,
         contracts,
       },
-      // @ts-expect-error StrictNullChecks
       oldestBlock,
       endingBlockNum,
     );
     config.WORKERS.EVM_CE_TRACE &&
       logger.warn('Events migrated', {
-        // @ts-expect-error StrictNullChecks
         startingBlockNum: oldestBlock,
         endingBlockNum,
       });
-    return result;
+    return {
+      events: result.events,
+      lastBlockNum: result.lastBlockNum,
+      contracts,
+    };
   } else {
-    // logger.info('No events to migrate');
-    return;
+    config.WORKERS.EVM_CE_TRACE && logger.info('No events to migrate');
+    return { contracts };
   }
 }

packages/commonwealth/server/workers/evmChainEvents/nodeProcessing.ts

@@ -15,6 +15,7 @@ import { config } from '../../config';
 import { getEventSources } from './getEventSources';
 import { getEvents, getProvider, migrateEvents } from './logProcessing';
 import { EvmEvent, EvmSource } from './types';
+import { updateMigratedEvmEventSources } from './utils';
 
 const log = logger(import.meta);
 
@@ -75,8 +76,8 @@ export async function processChainNode(
     }
 
     const allEvents: EvmEvent[] = [];
-    const migratedData = await migrateEvents(evmSource, startBlockNum);
-    if (migratedData && migratedData?.events?.length > 0)
+    const migratedData = await migrateEvents(evmSource, startBlockNum - 1);
+    if ('events' in migratedData && migratedData.events?.length > 0)
       allEvents.push(...migratedData.events);
 
     const { events, lastBlockNum } = await getEvents(
@@ -101,15 +102,24 @@ export async function processChainNode(
       }
 
       if (allEvents.length === 0) {
-        // log.info(`Processed 0 events for chainNodeId ${ethChainId}`);
+        await updateMigratedEvmEventSources(
+          ethChainId,
+          migratedData,
+          transaction,
+        );
         return;
       }
 
       const records = allEvents.map((event) => {
         const contractAddress = ethers.utils.getAddress(event.rawLog.address);
 
         const parseContestEvent = (e: keyof typeof ChainEventSigs) =>
-          parseEvmEventToContestEvent(e, contractAddress, event.parsedArgs);
+          parseEvmEventToContestEvent(
+            e,
+            contractAddress,
+            event.parsedArgs,
+            event.rawLog.blockNumber,
+          );
 
         switch (event.eventSource.eventSignature) {
           case EvmEventSignatures.NamespaceFactory.ContestManagerDeployed:
@@ -176,6 +186,11 @@ export async function processChainNode(
           )}`,
         );
         await emitEvent(models.Outbox, records, transaction);
+        await updateMigratedEvmEventSources(
+          ethChainId,
+          migratedData,
+          transaction,
+        );
       }
     });
   } catch (e) {