10474844847

Committed 20 Aug 2024 03:45PM UTC coverage: 15.293% (-0.2%) from 15.5%

Build # 10474844847

Build Type

Pull #8915

github

Committed by

web-flow

Commit Message

Merge 653af5951 into e90d468fa

Pull Request Pull Request #8915: Fixed community switching for auth user

Run Details

3508 of 21338 branches covered (16.44%)

Branch coverage included in aggregate %.

0 of 1 new or added line in 1 file covered. (0.0%)

81 existing lines in 5 files now uncovered.

4521 of 31163 relevant lines covered (14.51%)

5.45 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

16.13

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

import {
  ChainEventSigs,
  ContestContentAdded,
  ContestContentUpvoted,
  ContestStarted,
  EventNames,
  EvmEventSignatures,
  OneOffContestManagerDeployed,
  RecurringContestManagerDeployed,
  events as coreEvents,
  logger,
  parseEvmEventToContestEvent,
  stats,
} from '@hicommonwealth/core';
import { DB, emitEvent } from '@hicommonwealth/model';
import { ethers } from 'ethers';
import { z } from 'zod';
import { getEventSources } from './getEventSources';
import { getEvents, getProvider, migrateEvents } from './logProcessing';
import { EvmEvent, EvmSource } from './types';

const log = logger(import.meta);

/**
 * Given a ChainNode id and event sources, this function fetches all events parsed since
 * the last stored block number and then emits notifications for each of them before updating
 * the last fetched block number. This function will never throw an error.
 */
export async function processChainNode(
  models: DB,
  chainNodeId: number,
  evmSource: EvmSource,
): Promise<void> {
  try {
    log.info(
      'Processing:\n' +
        `\tchainNodeId: ${chainNodeId}\n` +
        `\tcontracts: ${JSON.stringify(Object.keys(evmSource.contracts))}`,
    );
    stats().increment('ce.evm.chain_node_id', {
      chainNodeId: String(chainNodeId),
    });

    const lastProcessedBlock = await models.LastProcessedEvmBlock.findOne({
      where: {
        chain_node_id: chainNodeId,
      },
    });

    const provider = getProvider(evmSource.rpc);
    const currentBlock = await provider.getBlockNumber();

    let startBlockNum: number;
    // @ts-expect-error StrictNullChecks
    if (lastProcessedBlock?.block_number < currentBlock) {
      // @ts-expect-error StrictNullChecks
      startBlockNum = lastProcessedBlock?.block_number + 1;
    } else if (lastProcessedBlock?.block_number === undefined) {
      startBlockNum = currentBlock - 10;
    } else {
      // the last processed block is the same as the current block
      // this occurs if the poll interval is shorter than the block time
      return;
    }

    const allEvents: EvmEvent[] = [];
    const migratedData = await migrateEvents(evmSource, startBlockNum);
    // @ts-expect-error StrictNullChecks
    if (migratedData?.events?.length > 0)
      // @ts-expect-error StrictNullChecks
      allEvents.push(...migratedData.events);

    const { events, lastBlockNum } = await getEvents(evmSource, startBlockNum);
    allEvents.push(...events);

    await models.sequelize.transaction(async (transaction) => {
      if (!lastProcessedBlock) {
        await models.LastProcessedEvmBlock.create(
          {
            chain_node_id: chainNodeId,
            block_number: lastBlockNum,
          },
          { transaction },
        );
      } else if (lastProcessedBlock.block_number !== lastBlockNum) {
        lastProcessedBlock.block_number = lastBlockNum;
        await lastProcessedBlock.save({ transaction });
      }

      if (allEvents.length === 0) {
        log.info(`Processed 0 events for chainNodeId ${chainNodeId}`);
        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);

        switch (event.eventSource.eventSignature) {
          case EvmEventSignatures.NamespaceFactory.ContestManagerDeployed:
            return parseContestEvent('NewContest') as
              | {
                  event_name: EventNames.RecurringContestManagerDeployed;
                  event_payload: z.infer<
                    typeof RecurringContestManagerDeployed
                  >;
                }
              | {
                  event_name: EventNames.OneOffContestManagerDeployed;
                  event_payload: z.infer<typeof OneOffContestManagerDeployed>;
                };
          case EvmEventSignatures.Contests.RecurringContestStarted:
            return parseContestEvent('NewRecurringContestStarted') as {
              event_name: EventNames.ContestStarted;
              event_payload: z.infer<typeof ContestStarted>;
            };
          case EvmEventSignatures.Contests.SingleContestStarted:
            return parseContestEvent('NewSingleContestStarted') as {
              event_name: EventNames.ContestStarted;
              event_payload: z.infer<typeof ContestStarted>;
            };
          case EvmEventSignatures.Contests.ContentAdded:
            return parseContestEvent('ContentAdded') as {
              event_name: EventNames.ContestContentAdded;
              event_payload: z.infer<typeof ContestContentAdded>;
            };
          case EvmEventSignatures.Contests.RecurringContestVoterVoted:
            return parseContestEvent('VoterVotedRecurring') as {
              event_name: EventNames.ContestContentUpvoted;
              event_payload: z.infer<typeof ContestContentUpvoted>;
            };
          case EvmEventSignatures.Contests.SingleContestVoterVoted:
            return parseContestEvent('VoterVotedOneOff') as {
              event_name: EventNames.ContestContentUpvoted;
              event_payload: z.infer<typeof ContestContentUpvoted>;
            };
        }

        // fallback to generic chain event
        return {
          event_name: EventNames.ChainEventCreated,
          event_payload: event as z.infer<typeof coreEvents.ChainEventCreated>,
        } as {
          event_name: EventNames.ChainEventCreated;
          event_payload: z.infer<typeof coreEvents.ChainEventCreated>;
        };
      });

      await emitEvent(models.Outbox, records, transaction);
    });

    log.info(
      `Processed ${allEvents.length} events for chainNodeId ${chainNodeId}`,
    );
  } catch (e) {
    const msg = `Error occurred while processing chainNodeId ${chainNodeId}`;
    log.error(msg, e);
  }
}

/**
 * Schedules processFn execution for each chainNode RPC in eventRpcSources. processFn execution is scheduled
 * evenly across the interval time so that blocks are not fetched all at once for all chainNodes. For example,
 * if there are 2 chainNodes and the interval is 4000ms, then processFn will be called for chainNode1 at T=0ms
 * and at T=2000ms and for chainNode2.
 * @param interval Time in milliseconds between each fetch of a ChainNode's blocks
 * @param processFn WARNING: must never throw an error. Errors thrown by processFn will not be caught.
 */
export async function scheduleNodeProcessing(
  models: DB,
  interval: number,
  processFn: (
    models: DB,
    chainNodeId: number,
    sources: EvmSource,
  ) => Promise<void>,
) {
  const evmSources = await getEventSources(models);

  const numEvmSources = Object.keys(evmSources).length;
  if (!numEvmSources) {
    return;
  }

  const chainNodeIds = Object.keys(evmSources);
  const betweenInterval = interval / numEvmSources;

  chainNodeIds.forEach((chainNodeId, index) => {
    const delay = index * betweenInterval;

    setTimeout(async () => {
      await processFn(models, +chainNodeId, evmSources[chainNodeId]);
    }, delay);
  });
}

1	import {
2	ChainEventSigs,
3	ContestContentAdded,
4	ContestContentUpvoted,
5	ContestStarted,
6	EventNames,
7	EvmEventSignatures,
8	OneOffContestManagerDeployed,
9	RecurringContestManagerDeployed,
10	events as coreEvents,
11	logger,
12	parseEvmEventToContestEvent,
13	stats,
14	} from '@hicommonwealth/core';
15	import { DB, emitEvent } from '@hicommonwealth/model';
16	import { ethers } from 'ethers';
17	import { z } from 'zod';
18	import { getEventSources } from './getEventSources';
19	import { getEvents, getProvider, migrateEvents } from './logProcessing';
20	import { EvmEvent, EvmSource } from './types';
21
22	const log = logger(import.meta);	1✔
23
24	/**
25	* Given a ChainNode id and event sources, this function fetches all events parsed since
26	* the last stored block number and then emits notifications for each of them before updating
27	* the last fetched block number. This function will never throw an error.
28	*/
29	export async function processChainNode(
30	models: DB,
31	chainNodeId: number,
32	evmSource: EvmSource,
33	): Promise<void> {
UNCOV 34	try {	×
UNCOV 35	log.info(	×
36	'Processing:\n' +
37	`\tchainNodeId: ${chainNodeId}\n` +
38	`\tcontracts: ${JSON.stringify(Object.keys(evmSource.contracts))}`,
39	);
UNCOV 40	stats().increment('ce.evm.chain_node_id', {	×
41	chainNodeId: String(chainNodeId),
42	});
43
UNCOV 44	const lastProcessedBlock = await models.LastProcessedEvmBlock.findOne({	×
45	where: {
46	chain_node_id: chainNodeId,
47	},
48	});
49
UNCOV 50	const provider = getProvider(evmSource.rpc);	×
UNCOV 51	const currentBlock = await provider.getBlockNumber();	×
52
53	let startBlockNum: number;
54	// @ts-expect-error StrictNullChecks
UNCOV 55	if (lastProcessedBlock?.block_number < currentBlock) {	×
56	// @ts-expect-error StrictNullChecks
57	startBlockNum = lastProcessedBlock?.block_number + 1;	×
UNCOV 58	} else if (lastProcessedBlock?.block_number === undefined) {	×
UNCOV 59	startBlockNum = currentBlock - 10;	×
60	} else {
61	// the last processed block is the same as the current block
62	// this occurs if the poll interval is shorter than the block time
63	return;	×
64	}
65
UNCOV 66	const allEvents: EvmEvent[] = [];	×
UNCOV 67	const migratedData = await migrateEvents(evmSource, startBlockNum);	×
68	// @ts-expect-error StrictNullChecks
UNCOV 69	if (migratedData?.events?.length > 0)	×
70	// @ts-expect-error StrictNullChecks
71	allEvents.push(...migratedData.events);	×
72
UNCOV 73	const { events, lastBlockNum } = await getEvents(evmSource, startBlockNum);	×
UNCOV 74	allEvents.push(...events);	×
75
UNCOV 76	await models.sequelize.transaction(async (transaction) => {	×
UNCOV 77	if (!lastProcessedBlock) {	×
UNCOV 78	await models.LastProcessedEvmBlock.create(	×
79	{
80	chain_node_id: chainNodeId,
81	block_number: lastBlockNum,
82	},
83	{ transaction },
84	);
85	} else if (lastProcessedBlock.block_number !== lastBlockNum) {	×
86	lastProcessedBlock.block_number = lastBlockNum;	×
87	await lastProcessedBlock.save({ transaction });	×
88	}
89
UNCOV 90	if (allEvents.length === 0) {	×
91	log.info(`Processed 0 events for chainNodeId ${chainNodeId}`);	×
92	return;	×
93	}
94
UNCOV 95	const records = allEvents.map((event) => {	×
UNCOV 96	const contractAddress = ethers.utils.getAddress(event.rawLog.address);	×
97
UNCOV 98	const parseContestEvent = (e: keyof typeof ChainEventSigs) =>	×
99	parseEvmEventToContestEvent(e, contractAddress, event.parsedArgs);	×
100
UNCOV 101	switch (event.eventSource.eventSignature) {	×
102	case EvmEventSignatures.NamespaceFactory.ContestManagerDeployed:
103	return parseContestEvent('NewContest') as	×
104	\| {
105	event_name: EventNames.RecurringContestManagerDeployed;
106	event_payload: z.infer<
107	typeof RecurringContestManagerDeployed
108	>;
109	}
110	\| {
111	event_name: EventNames.OneOffContestManagerDeployed;
112	event_payload: z.infer<typeof OneOffContestManagerDeployed>;
113	};
114	case EvmEventSignatures.Contests.RecurringContestStarted:
115	return parseContestEvent('NewRecurringContestStarted') as {	×
116	event_name: EventNames.ContestStarted;
117	event_payload: z.infer<typeof ContestStarted>;
118	};
119	case EvmEventSignatures.Contests.SingleContestStarted:
120	return parseContestEvent('NewSingleContestStarted') as {	×
121	event_name: EventNames.ContestStarted;
122	event_payload: z.infer<typeof ContestStarted>;
123	};
124	case EvmEventSignatures.Contests.ContentAdded:
125	return parseContestEvent('ContentAdded') as {	×
126	event_name: EventNames.ContestContentAdded;
127	event_payload: z.infer<typeof ContestContentAdded>;
128	};
129	case EvmEventSignatures.Contests.RecurringContestVoterVoted:
130	return parseContestEvent('VoterVotedRecurring') as {	×
131	event_name: EventNames.ContestContentUpvoted;
132	event_payload: z.infer<typeof ContestContentUpvoted>;
133	};
134	case EvmEventSignatures.Contests.SingleContestVoterVoted:
135	return parseContestEvent('VoterVotedOneOff') as {	×
136	event_name: EventNames.ContestContentUpvoted;
137	event_payload: z.infer<typeof ContestContentUpvoted>;
138	};
139	}
140
141	// fallback to generic chain event
UNCOV 142	return {	×
143	event_name: EventNames.ChainEventCreated,
144	event_payload: event as z.infer<typeof coreEvents.ChainEventCreated>,
145	} as {
146	event_name: EventNames.ChainEventCreated;
147	event_payload: z.infer<typeof coreEvents.ChainEventCreated>;
148	};
149	});
150
UNCOV 151	await emitEvent(models.Outbox, records, transaction);	×
152	});
153
UNCOV 154	log.info(	×
155	`Processed ${allEvents.length} events for chainNodeId ${chainNodeId}`,
156	);
157	} catch (e) {
158	const msg = `Error occurred while processing chainNodeId ${chainNodeId}`;	×
159	log.error(msg, e);	×
160	}
161	}
162
163	/**
164	* Schedules processFn execution for each chainNode RPC in eventRpcSources. processFn execution is scheduled
165	* evenly across the interval time so that blocks are not fetched all at once for all chainNodes. For example,
166	* if there are 2 chainNodes and the interval is 4000ms, then processFn will be called for chainNode1 at T=0ms
167	* and at T=2000ms and for chainNode2.
168	* @param interval Time in milliseconds between each fetch of a ChainNode's blocks
169	* @param processFn WARNING: must never throw an error. Errors thrown by processFn will not be caught.
170	*/
171	export async function scheduleNodeProcessing(
172	models: DB,
173	interval: number,
174	processFn: (
175	models: DB,
176	chainNodeId: number,
177	sources: EvmSource,
178	) => Promise<void>,
179	) {
180	const evmSources = await getEventSources(models);	3✔
181
182	const numEvmSources = Object.keys(evmSources).length;	3✔
183	if (!numEvmSources) {	3✔
184	return;	1✔
185	}
186
187	const chainNodeIds = Object.keys(evmSources);	2✔
188	const betweenInterval = interval / numEvmSources;	2✔
189
190	chainNodeIds.forEach((chainNodeId, index) => {	2✔
191	const delay = index * betweenInterval;	3✔
192
193	setTimeout(async () => {	3✔
194	await processFn(models, +chainNodeId, evmSources[chainNodeId]);	3✔
195	}, delay);
196	});
197	}

hicommonwealth / commonwealth / 10474844847

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous