Skip to content

Commit

Permalink
feat: integrate v3 fill events (#339)
Browse files Browse the repository at this point in the history 
* fix: replace save with update (#338)

Co-authored-by: amateima <[email protected]>

* fix: replace save with update (#340)

* fix: replace save with update

* Fix

---------

Co-authored-by: amateima <[email protected]>

* feat: integrate v3 fill events

* Add fill events consumer

* Finish v3 fill events

* Dix

* feat: implement speedup v3 events (#341)

* feat: implement speedup v3 events

* Fix bugs

* Fix import

* Fix

---------

Co-authored-by: amateima <[email protected]>

---------

Co-authored-by: amateima <[email protected]>
  • Loading branch information
@amateima
amateima and amateima authored Feb 20, 2024
1 parent 4886720 commit 20e328d
Show file tree
Hide file tree
Showing 20 changed files with 533 additions and 65 deletions.
2 changes: 1 addition & 1 deletion src/modules/configuration/index.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -182,7 +182,7 @@ export const configValues = () => ({
[ChainIds.sepolia]: [
{
address: "0x622d59F3dbD28fcFE746E0d2f83ebdC286E89A3c",
startBlockNumber: 14819486,
startBlockNumber: 5146129,
abi: JSON.stringify(SpokePoolV3Abi),
acrossVersion: AcrossContractsVersion.V3,
},
Expand Down
23 changes: 20 additions & 3 deletions src/modules/deposit/model/deposit.entity.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,14 @@ export type DepositFillTx2 = {
relayerFeePct: string;
date?: string;
};
export type DepositFillTxV3 = {
updatedRecipient: string;
updatedMessage: string;
updatedOutputAmount: string;
fillType: number;
hash: string;
date?: string;
};
export type RequestedSpeedUpDepositTx = {
hash: string;
blockNumber: number;
Expand All @@ -37,6 +45,13 @@ export type RequestedSpeedUpDepositTx = {
updatedRecipient?: string;
updatedMessage?: string;
};
export type RequestedSpeedUpDepositTxV3 = {
hash: string;
blockNumber: number;
updatedOutputAmount: string;
updatedRecipient: string;
updatedMessage: string;
};
export type FeeBreakdown = {
// lp fee
lpFeeUsd: string;
Expand Down Expand Up @@ -147,7 +162,9 @@ export class Deposit {
outputTokenId?: number;

@ManyToOne(() => Token)
@JoinColumn([{ name: "tokenId", referencedColumnName: "id", foreignKeyConstraintName: "FK_deposit_outputTokenId" }])
@JoinColumn([
{ name: "outputTokenId", referencedColumnName: "id", foreignKeyConstraintName: "FK_deposit_outputTokenId" },
])
outputToken?: Token;

@Column({ nullable: true })
Expand All @@ -167,10 +184,10 @@ export class Deposit {
depositTxHash: string;

@Column({ type: "jsonb", default: [] })
fillTxs: (DepositFillTx | DepositFillTx2)[];
fillTxs: (DepositFillTx | DepositFillTx2 | DepositFillTxV3)[];

@Column({ type: "jsonb", default: [] })
speedUps: RequestedSpeedUpDepositTx[];
speedUps: RequestedSpeedUpDepositTx[] | RequestedSpeedUpDepositTxV3[];

@Column({ type: "jsonb", default: {} })
feeBreakdown: FeeBreakdown;
Expand Down
4 changes: 4 additions & 0 deletions src/modules/rewards/services/op-rebate-service.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,10 @@ export class OpRebateService {

this.assertDepositKeys(deposit, ["price", "token", "depositDate", "feeBreakdown"]);

if (deposit.outputTokenAddress) {
this.assertDepositKeys(deposit, ["outputTokenPrice", "outputToken"]);
}

if (Object.keys(deposit.feeBreakdown).length === 0) {
throw new Error(`Deposit with id ${depositPrimaryKey} is missing fee breakdown`);
}
Expand Down
114 changes: 101 additions & 13 deletions src/modules/scraper/adapter/messaging/BlocksEventsConsumer.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,10 @@ import {
BlocksEventsQueueMessage,
FillEventsQueueMessage,
FillEventsQueueMessage2,
FillEventsV3QueueMessage,
ScraperQueue,
SpeedUpEventsQueueMessage,
SpeedUpEventsV3QueueMessage,
} from ".";
import { Deposit } from "../../../deposit/model/deposit.entity";
import { ScraperQueuesService } from "../../service/ScraperQueuesService";
Expand All @@ -27,6 +29,8 @@ import {
RequestedSpeedUpDepositEvent2,
RequestedSpeedUpDepositEvent2_5,
FundsDepositedV3Event,
FilledV3RelayEvent,
RequestedSpeedUpV3DepositEvent,
} from "../../../web3/model";
import { AppConfig } from "../../../configuration/configuration.service";
import { splitBlockRanges } from "../../utils";
Expand Down Expand Up @@ -59,19 +63,24 @@ export class BlocksEventsConsumer {
// Split the block range in case multiple SpokePool contracts need to be queried
const blocksToQuery = splitBlockRanges(ascSpokePoolConfigs, from, to);
// Get the events from the SpokePool contracts
const { depositEvents, depositV3Events, fillEvents, speedUpEvents } = await this.getEvents(blocksToQuery, chainId);
const { depositEvents, depositV3Events, fillEvents, fillV3Events, speedUpEvents, speedUpV3Events } =
await this.getEvents(blocksToQuery, chainId);
const eventsCount = {
depositEvents: depositEvents.length,
depositV3Events: depositV3Events.length,
fillEvents: fillEvents.length,
fillV3Events: fillV3Events.length,
speedUpEvents: speedUpEvents.length,
speedUpV3Events: speedUpV3Events.length,
};
this.logger.log(`${from}-${to} - chainId ${chainId} - ${JSON.stringify(eventsCount)}`);

await this.processDepositEvents(chainId, depositEvents);
await this.processDepositV3Events(chainId, depositV3Events);
await this.processFillEvents(chainId, fillEvents);
await this.processFillEvents(chainId, fillV3Events);
await this.processSpeedUpEvents(chainId, speedUpEvents);
await this.processSpeedUpV3Events(chainId, speedUpV3Events);
}

private async getEvents(
Expand All @@ -82,11 +91,15 @@ export class BlocksEventsConsumer {
const depositEvents: Event[] = [];
const depositV3Events: Event[] = [];
const fillEvents: Event[] = [];
const fillV3Events: Event[] = [];
const speedUpEvents: Event[] = [];
const speedUpV3Events: Event[] = [];

for (const blocks of blocksToQuery) {
const spokePoolEventQuerier = this.providers.getSpokePoolEventQuerier(chainId, blocks.address);
let depositEventsPromises = [];
let fillEventsPromises = [];
let speedUpEventsPromises = [];

if (blocks.acrossVersion === AcrossContractsVersion.V2) {
depositEventsPromises = [
Expand All @@ -95,41 +108,80 @@ export class BlocksEventsConsumer {
res([]);
}),
];
fillEventsPromises = [
spokePoolEventQuerier.getFilledRelayEvents(blocks.from, blocks.to),
new Promise((res) => {
res([]);
}),
];
speedUpEventsPromises = [
spokePoolEventQuerier.getRequestedSpeedUpDepositEvents(blocks.from, blocks.to),
new Promise((res) => {
res([]);
}),
];
} else if (blocks.acrossVersion === AcrossContractsVersion.V2_5) {
depositEventsPromises = [
spokePoolEventQuerier.getFundsDepositEvents(blocks.from, blocks.to),
new Promise((res) => {
res([]);
}),
];
fillEventsPromises = [
spokePoolEventQuerier.getFilledRelayEvents(blocks.from, blocks.to),
new Promise((res) => {
res([]);
}),
];
speedUpEventsPromises = [
spokePoolEventQuerier.getRequestedSpeedUpDepositEvents(blocks.from, blocks.to),
new Promise((res) => {
res([]);
}),
];
} else if (blocks.acrossVersion === AcrossContractsVersion.V3) {
depositEventsPromises = [
new Promise((res) => {
res([]);
}),
spokePoolEventQuerier.getFundsDepositedV3Events(blocks.from, blocks.to),
];
fillEventsPromises = [
spokePoolEventQuerier.getFilledRelayEvents(blocks.from, blocks.to),
spokePoolEventQuerier.getFilledV3RelayEvents(blocks.from, blocks.to),
];
speedUpEventsPromises = [
new Promise((res) => {
res([]);
}),
spokePoolEventQuerier.getRequestedSpeedUpV3DepositEvents(blocks.from, blocks.to),
];
}
const promises = [
...depositEventsPromises,
spokePoolEventQuerier.getFilledRelayEvents(blocks.from, blocks.to),
spokePoolEventQuerier.getRequestedSpeedUpDepositEvents(blocks.from, blocks.to),
];
const [depositEventsChunk, depositV3EventsChunk, fillEventsChunk, speedUpEventsChunk] = await Promise.all(
promises,
);
const promises = [...depositEventsPromises, ...fillEventsPromises, ...speedUpEventsPromises];
const [
depositEventsChunk,
depositV3EventsChunk,
fillEventsChunk,
fillV3EventsChunk,
speedUpEventsChunk,
speedUpV3EventsChunk,
] = await Promise.all(promises);

depositEvents.push(...depositEventsChunk);
depositV3Events.push(...depositV3EventsChunk);
fillEvents.push(...fillEventsChunk);
fillV3Events.push(...fillV3EventsChunk);
speedUpEvents.push(...speedUpEventsChunk);
speedUpV3Events.push(...speedUpV3EventsChunk);
}

return {
depositEvents,
depositV3Events,
fillEvents,
fillV3Events,
speedUpEvents,
speedUpV3Events,
};
}

Expand Down Expand Up @@ -182,7 +234,7 @@ export class BlocksEventsConsumer {
(contract) => contract.address === address,
)[0];

if (acrossVersion === AcrossContractsVersion.V2) {
if (event.args.appliedRelayerFeePct) {
const typedEvent = event as FilledRelayEvent2;
const message: FillEventsQueueMessage = {
depositId: typedEvent.args.depositId,
Expand All @@ -195,7 +247,7 @@ export class BlocksEventsConsumer {
destinationToken: typedEvent.args.destinationToken,
};
await this.scraperQueuesService.publishMessage<FillEventsQueueMessage>(ScraperQueue.FillEvents, message);
} else if (acrossVersion === AcrossContractsVersion.V2_5) {
} else if (event.args.updatableRelayData) {
const typedEvent = event as FilledRelayEvent2_5;
const message: FillEventsQueueMessage2 = {
depositId: typedEvent.args.depositId,
Expand All @@ -208,6 +260,20 @@ export class BlocksEventsConsumer {
destinationToken: typedEvent.args.destinationToken,
};
await this.scraperQueuesService.publishMessage<FillEventsQueueMessage2>(ScraperQueue.FillEvents2, message);
} else if (event.args.relayExecutionInfo) {
const typedEvent = event as FilledV3RelayEvent;
const message: FillEventsV3QueueMessage = {
updatedRecipient: typedEvent.args.relayExecutionInfo.updatedRecipient,
updatedMessage: typedEvent.args.relayExecutionInfo.updatedMessage,
updatedOutputAmount: typedEvent.args.relayExecutionInfo.updatedOutputAmount.toString(),
fillType: typedEvent.args.relayExecutionInfo.fillType,
depositId: typedEvent.args.depositId,
originChainId: typedEvent.args.originChainId.toNumber(),
transactionHash: typedEvent.transactionHash,
};
await this.scraperQueuesService.publishMessage<FillEventsV3QueueMessage>(ScraperQueue.FillEventsV3, message);
} else {
throw new Error("Unknown fill event type");
}
}
}
Expand Down Expand Up @@ -255,6 +321,28 @@ export class BlocksEventsConsumer {
}
}

private async processSpeedUpV3Events(chainId: number, events: Event[]) {
for (const event of events) {
const { transactionHash, blockNumber, args } = event as RequestedSpeedUpV3DepositEvent;
const message: SpeedUpEventsV3QueueMessage = {
depositSourceChainId: chainId,
depositId: args.depositId,
transactionHash,
blockNumber,
depositor: args.depositor,
depositorSignature: args.depositorSignature,
updatedOutputAmount: args.updatedOutputAmount.toString(),
updatedRecipient: args.updatedRecipient,
updatedMessage: args.updatedMessage,
};

await this.scraperQueuesService.publishMessage<SpeedUpEventsV3QueueMessage>(
ScraperQueue.SpeedUpEventsV3,
message,
);
}
}

private async fromFundsDepositedEventToDeposit(chainId: number, event: Event) {
const typedEvent = event as FundsDepositedEvent2 | FundsDepositedEvent2_5;
const { transactionHash, blockNumber } = typedEvent;
Expand Down Expand Up @@ -325,8 +413,8 @@ export class BlocksEventsConsumer {
// v3 properties
outputAmount: outputAmount.toString(),
outputTokenAddress: outputToken,
fillDeadline: fillDeadline,
exclusivityDeadline: exclusivityDeadline,
fillDeadline: new Date(fillDeadline * 1000),
exclusivityDeadline: new Date(exclusivityDeadline * 1000),
relayer,
});
}
Expand Down
17 changes: 11 additions & 6 deletions src/modules/scraper/adapter/messaging/DepositFilledDateConsumer.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ import { Repository } from "typeorm";
import { DateTime } from "luxon";
import { EthProvidersService } from "../../../web3/services/EthProvidersService";
import { DepositFilledDateQueueMessage, ScraperQueue, TrackFillEventQueueMessage } from ".";
import { Deposit, DepositFillTx, DepositFillTx2 } from "../../../deposit/model/deposit.entity";
import { Deposit, DepositFillTx, DepositFillTx2, DepositFillTxV3 } from "../../../deposit/model/deposit.entity";
import { ScraperQueuesService } from "../../service/ScraperQueuesService";

@Processor(ScraperQueue.DepositFilledDate)
Expand All @@ -27,9 +27,7 @@ export class DepositFilledDateConsumer {
if (!deposit) return;
if (deposit.status !== "filled") return;

if (!deposit.depositDate) {
throw new Error("Wait for deposit date");
}
if (!deposit.depositDate) throw new Error("Wait for deposit date");

const fillTxsWithoutDate = deposit.fillTxs.filter((fillTx) => !fillTx.date).length;

Expand All @@ -51,7 +49,14 @@ export class DepositFilledDateConsumer {
}

deposit.filledDate = filledDate;
await this.depositRepository.save(deposit);

await this.depositRepository.update(
{ id: deposit.id },
{
filledDate: deposit.filledDate,
fillTxs: deposit.fillTxs,
},
);

this.scraperQueuesService.publishMessage<TrackFillEventQueueMessage>(ScraperQueue.TrackFillEvent, {
depositId: deposit.id,
Expand All @@ -60,7 +65,7 @@ export class DepositFilledDateConsumer {
});
}

private async fillDateForFillTx(chainId: number, fillTx: DepositFillTx | DepositFillTx2) {
private async fillDateForFillTx(chainId: number, fillTx: DepositFillTx | DepositFillTx2 | DepositFillTxV3) {
if (fillTx.date) return fillTx;

const tx = await this.providers.getCachedTransaction(chainId, fillTx.hash);
Expand Down
Loading

0 comments on commit 20e328d

Please sign in to comment.