feat: create requests

packages/automated-dispute/src/interfaces/protocolProvider.ts

@@ -1,9 +1,7 @@
 import { Caip2ChainId } from "@ebo-agent/blocknumber/dist/types.js";
-import { Timestamp } from "@ebo-agent/shared";
 import { Address } from "viem";
 
-import type { EboEvent, EboEventName } from "../types/events.js";
-import type { Dispute, Request, Response } from "../types/prophet.js";
+import type { Dispute, EboEvent, EboEventName, Epoch, Request, Response } from "../types/index.js";
 import { ProtocolContractsNames } from "../constants.js";
 
 export type ProtocolContract = (typeof ProtocolContractsNames)[number];
@@ -18,11 +16,7 @@ export interface IReadProvider {
      *
      * @returns A promise that resolves with the current epoch, block number, and timestamp.
      */
-    getCurrentEpoch(): Promise<{
-        currentEpoch: bigint;
-        currentEpochBlockNumber: bigint;
-        currentEpochTimestamp: Timestamp;
-    }>;
+    getCurrentEpoch(): Promise<Epoch>;
 
     /**
      * Gets the last finalized block number.

packages/automated-dispute/src/providers/protocolProvider.ts

@@ -1,5 +1,4 @@
 import { Caip2ChainId } from "@ebo-agent/blocknumber/dist/types.js";
-import { Timestamp } from "@ebo-agent/shared";
 import {
     Address,
     BaseError,
@@ -19,8 +18,7 @@ import {
 import { privateKeyToAccount } from "viem/accounts";
 import { arbitrum } from "viem/chains";
 
-import type { EboEvent, EboEventName } from "../types/events.js";
-import type { Dispute, Request, Response } from "../types/prophet.js";
+import type { Dispute, EboEvent, EboEventName, Epoch, Request, Response } from "../types/index.js";
 import { eboRequestCreatorAbi, epochManagerAbi, oracleAbi } from "../abis/index.js";
 import { RpcUrlsEmpty } from "../exceptions/rpcUrlsEmpty.exception.js";
 import {
@@ -138,24 +136,20 @@ export class ProtocolProvider implements IProtocolProvider {
      *
      * @returns The current epoch, its block number and its timestamp
      */
-    async getCurrentEpoch(): Promise<{
-        currentEpoch: bigint;
-        currentEpochBlockNumber: bigint;
-        currentEpochTimestamp: Timestamp;
-    }> {
-        const [currentEpoch, currentEpochBlockNumber] = await Promise.all([
+    async getCurrentEpoch(): Promise<Epoch> {
+        const [epoch, epochFirstBlockNumber] = await Promise.all([
             this.epochManagerContract.read.currentEpoch(),
             this.epochManagerContract.read.currentEpochBlock(),
         ]);
 
-        const currentEpochBlock = await this.readClient.getBlock({
-            blockNumber: currentEpochBlockNumber,
+        const epochFirstBlock = await this.readClient.getBlock({
+            blockNumber: epochFirstBlockNumber,
         });
 
         return {
-            currentEpoch,
-            currentEpochBlockNumber,
-            currentEpochTimestamp: currentEpochBlock.timestamp,
+            epoch,
+            epochFirstBlockNumber,
+            epochStartTimestamp: epochFirstBlock.timestamp,
         };
     }
 
@@ -242,7 +236,7 @@ export class ProtocolProvider implements IProtocolProvider {
     }
 
     // TODO: use Caip2 Chain ID instead of string in return type
-    async getAvailableChains(): Promise<string[]> {
+    async getAvailableChains(): Promise<Caip2ChainId[]> {
         // TODO: implement actual method
         return ["eip155:1", "eip155:42161"];
     }

packages/automated-dispute/src/services/eboActor.ts

@@ -5,6 +5,17 @@ import { Mutex } from "async-mutex";
 import { Heap } from "heap-js";
 import { ContractFunctionRevertedError } from "viem";
 
+import type {
+    Dispute,
+    DisputeStatus,
+    EboEvent,
+    EboEventName,
+    Epoch,
+    Request,
+    RequestId,
+    Response,
+    ResponseBody,
+} from "../types/index.js";
 import {
     DisputeWithoutResponse,
     EBORequestCreator_ChainNotAdded,
@@ -27,16 +38,6 @@ import {
     FinalizeRequest,
     UpdateDisputeStatus,
 } from "../services/index.js";
-import {
-    Dispute,
-    DisputeStatus,
-    EboEvent,
-    EboEventName,
-    Request,
-    RequestId,
-    Response,
-    ResponseBody,
-} from "../types/index.js";
 
 /**
  * Compare function to sort events chronologically in ascending order by block number
@@ -53,6 +54,8 @@ const EBO_EVENT_COMPARATOR = (e1: EboEvent<EboEventName>, e2: EboEvent<EboEventN
     return e1.logIndex - e2.logIndex;
 };
 
+export type ActorRequest = { id: RequestId; epoch: bigint; chainId: Caip2ChainId };
+
 /**
  * Actor that handles a singular Prophet's request asking for the block number that corresponds
  * to an instant on an indexed chain.
@@ -78,10 +81,7 @@ export class EboActor {
      * @param logger an `ILogger` instance
      */
     constructor(
-        private readonly actorRequest: {
-            id: RequestId;
-            epoch: bigint;
-        },
+        public readonly actorRequest: ActorRequest,
         private readonly protocolProvider: ProtocolProvider,
         private readonly blockNumberService: BlockNumberService,
         private readonly registry: EboRegistry,
@@ -432,15 +432,19 @@ export class EboActor {
      *
      * Be aware that a request can be finalized but some of its disputes can still be pending resolution.
      *
+     * At last, actors must be kept alive until their epoch concludes, to ensure no actor/request duplication.
+     *
+     * @param currentEpoch the epoch to check against actor termination
      * @param blockNumber block number to check entities at
      * @returns `true` if all entities are settled, otherwise `false`
      */
-    public canBeTerminated(blockNumber: bigint): boolean {
+    public canBeTerminated(currentEpoch: Epoch["epoch"], blockNumber: bigint): boolean {
         const request = this.getActorRequest();
+        const isPastEpoch = currentEpoch > request.epoch;
         const isRequestFinalized = request.status === "Finalized";
         const nonSettledProposals = this.activeProposals(blockNumber);
 
-        return isRequestFinalized && nonSettledProposals.length === 0;
+        return isPastEpoch && isRequestFinalized && nonSettledProposals.length === 0;
     }
 
     /**
@@ -556,10 +560,10 @@ export class EboActor {
     private async buildResponse(chainId: Caip2ChainId): Promise<ResponseBody> {
         // FIXME(non-current epochs): adapt this code to fetch timestamps corresponding
         //  to the first block of any epoch, not just the current epoch
-        const { currentEpochTimestamp } = await this.protocolProvider.getCurrentEpoch();
+        const { epochStartTimestamp } = await this.protocolProvider.getCurrentEpoch();
 
         const epochBlockNumber = await this.blockNumberService.getEpochBlockNumber(
-            currentEpochTimestamp,
+            epochStartTimestamp,
             chainId,
         );
 

packages/automated-dispute/src/services/eboActorsManager.ts

@@ -4,8 +4,8 @@ import { Mutex } from "async-mutex";
 
 import { RequestAlreadyHandled } from "../exceptions/index.js";
 import { ProtocolProvider } from "../providers/protocolProvider.js";
-import { RequestId } from "../types/prophet.js";
-import { EboActor } from "./eboActor.js";
+import { RequestId } from "../types/index.js";
+import { ActorRequest, EboActor } from "./eboActor.js";
 import { EboMemoryRegistry } from "./eboRegistry/eboMemoryRegistry.js";
 
 export class EboActorsManager {
@@ -21,7 +21,11 @@ export class EboActorsManager {
      * @returns array of normalized request IDs
      */
     public getRequestIds(): RequestId[] {
-        return [...this.requestActorMap.entries()].map((entry) => Address.normalize(entry[0]));
+        return [...this.requestActorMap.keys()].map((requestId) => Address.normalize(requestId));
+    }
+
+    public getActorsRequests(): ActorRequest[] {
+        return [...this.requestActorMap.values()].map((actor) => actor.actorRequest);
     }
 
     /**
@@ -30,10 +34,7 @@ export class EboActorsManager {
      * @param actor an `EboActor` instance that handles a request.
      */
     public createActor(
-        actorRequest: {
-            id: RequestId;
-            epoch: bigint;
-        },
+        actorRequest: ActorRequest,
         protocolProvider: ProtocolProvider,
         blockNumberService: BlockNumberService,
         logger: ILogger,

packages/automated-dispute/src/services/eboProcessor.ts

@@ -1,12 +1,13 @@
 import { isNativeError } from "util/types";
 import { BlockNumberService } from "@ebo-agent/blocknumber";
+import { Caip2ChainId } from "@ebo-agent/blocknumber/dist/types.js";
 import { Address, ILogger } from "@ebo-agent/shared";
 
 import { ProcessorAlreadyStarted } from "../exceptions/index.js";
 import { ProtocolProvider } from "../providers/protocolProvider.js";
 import { alreadyDeletedActorWarning, droppingUnhandledEventsWarning } from "../templates/index.js";
-import { EboEvent, EboEventName } from "../types/events.js";
-import { RequestId } from "../types/prophet.js";
+import { EboEvent, EboEventName, Epoch, RequestId } from "../types/index.js";
+import { ActorRequest } from "./eboActor.js";
 import { EboActorsManager } from "./eboActorsManager.js";
 
 const DEFAULT_MS_BETWEEN_CHECKS = 10 * 60 * 1000; // 10 minutes
@@ -49,13 +50,11 @@ export class EboProcessor {
 
     /** Sync new blocks and their events with their corresponding actors. */
     private async sync() {
-        // TODO: detect new epoch by comparing subgraph's data with EpochManager's current epoch
-        //  and trigger a request creation if there's no actor handling an <epoch, chain> request.
-        //  This process should somehow check if there's already a request created for the epoch
-        //  and chain that has no agent assigned and create it if that's the case.
         try {
+            const currentEpoch = await this.getCurrentEpoch();
+
             if (!this.lastCheckedBlock) {
-                this.lastCheckedBlock = await this.getEpochStartBlock();
+                this.lastCheckedBlock = currentEpoch.epochFirstBlockNumber;
             }
 
             const lastBlock = await this.getLastFinalizedBlock();
@@ -74,7 +73,7 @@ export class EboProcessor {
                 try {
                     const events = eventsByRequestId.get(requestId) ?? [];
 
-                    await this.syncRequest(requestId, events, lastBlock);
+                    await this.syncRequest(requestId, events, currentEpoch.epoch, lastBlock);
                 } catch (err) {
                     this.onActorError(requestId, err as Error);
                 }
@@ -84,6 +83,8 @@ export class EboProcessor {
 
             this.logger.info(`Consumed events up to block ${lastBlock}.`);
 
+            this.createMissingRequests(currentEpoch.epoch);
+
             this.lastCheckedBlock = lastBlock;
         } catch (err) {
             if (isNativeError(err)) {
@@ -97,18 +98,18 @@ export class EboProcessor {
     }
 
     /**
-     * Fetches the first block of the current epoch.
+     * Fetches the current epoch for the protocol chain.
      *
-     * @returns the first block of the current epoch
+     * @returns the current epoch properties of the protocol chain.
      */
-    private async getEpochStartBlock(): Promise<bigint> {
-        this.logger.info("Fetching current epoch start block...");
+    private async getCurrentEpoch(): Promise<Epoch> {
+        this.logger.info("Fetching current epoch...");
 
-        const { currentEpochBlockNumber } = await this.protocolProvider.getCurrentEpoch();
+        const currentEpoch = await this.protocolProvider.getCurrentEpoch();
 
-        this.logger.info(`Current epoch start block ${currentEpochBlockNumber} fetched.`);
+        this.logger.info(`Current epoch fetched.`);
 
-        return currentEpochBlockNumber;
+        return currentEpoch;
     }
 
     /**
@@ -182,9 +183,15 @@ export class EboProcessor {
      *
      * @param requestId the ID of the `Request`
      * @param events a stream of consumed events
+     * @param currentEpoch the current epoch based on the last block
      * @param lastBlock the last block checked
      */
-    private async syncRequest(requestId: RequestId, events: EboEventStream, lastBlock: bigint) {
+    private async syncRequest(
+        requestId: RequestId,
+        events: EboEventStream,
+        currentEpoch: Epoch["epoch"],
+        lastBlock: bigint,
+    ) {
         const firstEvent = events[0];
         const actor = this.getOrCreateActor(requestId, firstEvent);
 
@@ -199,7 +206,7 @@ export class EboProcessor {
         await actor.processEvents();
         await actor.onLastBlockUpdated(lastBlock);
 
-        if (actor.canBeTerminated(lastBlock)) {
+        if (actor.canBeTerminated(currentEpoch, lastBlock)) {
             this.terminateActor(requestId);
         }
     }
@@ -235,6 +242,7 @@ export class EboProcessor {
         const actorRequest = {
             id: Address.normalize(event.requestId),
             epoch: event.metadata.epoch,
+            chainId: event.metadata.chainId,
         };
 
         const actor = this.actorsManager.createActor(
@@ -259,6 +267,72 @@ export class EboProcessor {
         this.terminateActor(requestId);
     }
 
+    /**
+     * Creates missing requests for the specified epoch, based on the
+     * available chains and the currently being handled requests.
+     *
+     * @param epoch the epoch number
+     */
+    private async createMissingRequests(epoch: Epoch["epoch"]): Promise<void> {
+        try {
+            const handledEpochChains = this.actorsManager
+                .getActorsRequests()
+                .reduce((actorRequestMap, actorRequest: ActorRequest) => {
+                    const epochRequests = actorRequestMap.get(actorRequest.epoch) ?? new Set();
+
+                    epochRequests.add(actorRequest.chainId);
+
+                    return actorRequestMap.set(actorRequest.epoch, epochRequests);
+                }, new Map<Epoch["epoch"], Set<Caip2ChainId>>());
+
+            this.logger.info("Fetching available chains...");
+
+            const availableChains: Caip2ChainId[] =
+                await this.protocolProvider.getAvailableChains();
+
+            this.logger.info("Available chains fetched.");
+
+            const unhandledEpochChain = availableChains.filter((chain) => {
+                const epochRequests = handledEpochChains.get(epoch);
+                const isHandled = epochRequests && epochRequests.has(chain);
+
+                return !isHandled;
+            });
+
+            this.logger.info("Creating missing requests...");
+
+            const epochChainRequests = unhandledEpochChain.map(async (chain) => {
+                try {
+                    this.logger.info(`Creating request for chain ${chain} and epoch ${epoch}...`);
+
+                    await this.protocolProvider.createRequest(epoch, [chain]);
+
+                    this.logger.info(`Request created for chain ${chain} and epoch ${epoch}`);
+                } catch (err) {
+                    // Request creation must be notified but it's not critical, as it will be
+                    // retried during next sync.
+
+                    // TODO: warn when getting a EBORequestCreator_RequestAlreadyCreated
+                    // TODO: notify under any other error
+
+                    this.logger.error(
+                        `Could not create a request for epoch ${epoch} and chain ${chain}.`,
+                    );
+                }
+            });
+
+            await Promise.all(epochChainRequests);
+
+            this.logger.info("Missing requests created.");
+        } catch (err) {
+            // TODO: notify
+
+            this.logger.error(
+                `Requests creation missing: ${isNativeError(err) ? err.message : err}`,
+            );
+        }
+    }
+
     /**
      * Removes the actor from tracking the request.
      *