feat: add pending RCA proposal consumer and migration

packages/indexer-agent/src/agent.ts

@@ -716,6 +716,11 @@ export class Agent {
               if (!operator.dipsManager) {
                 throw new Error('DipsManager is not available')
               }
+
+              await operator.dipsManager.acceptPendingProposals(
+                activeAllocations,
+              )
+
               this.logger.debug(
                 `Matching agreement allocations for network ${network.specification.networkIdentifier}`,
               )

packages/indexer-agent/src/commands/start.ts

@@ -15,6 +15,7 @@ import {
   createIndexerManagementClient,
   createIndexerManagementServer,
   defineIndexerManagementModels,
+  definePendingRcaProposalModel,
   defineQueryFeeModels,
   GraphNode,
   indexerError,
@@ -670,6 +671,9 @@ export async function run(
   await sequelize.sync()
   logger.info(`Successfully synced database models`)
 
+  // Define after sync so Sequelize won't try to create/alter this indexer-rs-owned table
+  const pendingRcaModel = definePendingRcaProposalModel(sequelize)
+
   // --------------------------------------------------------------------------------
   // * Networks
   // --------------------------------------------------------------------------------
@@ -710,6 +714,7 @@ export async function run(
       },
     },
     multiNetworks,
+    pendingRcaModel,
   })
 
   // --------------------------------------------------------------------------------

packages/indexer-agent/src/db/migrations/23-add-pending-rca-proposals.ts

@@ -0,0 +1,74 @@
+import { Logger } from '@graphprotocol/common-ts'
+
+import { QueryInterface, DataTypes } from 'sequelize'
+
+interface MigrationContext {
+  queryInterface: QueryInterface
+  logger: Logger
+}
+
+interface Context {
+  context: MigrationContext
+}
+
+export async function up({ context }: Context): Promise<void> {
+  const { queryInterface, logger } = context
+
+  const tables = await queryInterface.showAllTables()
+  logger.debug('Checking if pending_rca_proposals table exists', { tables })
+
+  if (tables.includes('pending_rca_proposals')) {
+    logger.debug(
+      'pending_rca_proposals already exists, migration not necessary',
+    )
+    return
+  }
+
+  logger.info('Create pending_rca_proposals')
+  await queryInterface.createTable('pending_rca_proposals', {
+    id: {
+      type: DataTypes.UUID,
+      primaryKey: true,
+    },
+    signed_payload: {
+      type: DataTypes.BLOB,
+      allowNull: false,
+    },
+    version: {
+      type: DataTypes.SMALLINT,
+      allowNull: false,
+      defaultValue: 2,
+    },
+    status: {
+      type: DataTypes.STRING(20),
+      allowNull: false,
+      defaultValue: 'pending',
+    },
+    created_at: {
+      type: DataTypes.DATE,
+      allowNull: false,
+    },
+    updated_at: {
+      type: DataTypes.DATE,
+      allowNull: false,
+    },
+  })
+
+  await queryInterface.addIndex(
+    'pending_rca_proposals',
+    ['status', 'created_at'],
+    {
+      name: 'idx_pending_rca_status',
+    },
+  )
+  await queryInterface.addIndex('pending_rca_proposals', {
+    fields: [{ name: 'created_at', order: 'DESC' }],
+    name: 'idx_pending_rca_created',
+  })
+}
+
+export async function down({ context }: Context): Promise<void> {
+  const { queryInterface, logger } = context
+  logger.info('Drop pending_rca_proposals')
+  await queryInterface.dropTable('pending_rca_proposals')
+}

packages/indexer-common/src/indexer-management/actions.ts

@@ -24,6 +24,7 @@ import {
 import { Order, Transaction } from 'sequelize'
 import { Eventual, join, Logger } from '@graphprotocol/common-ts'
 import groupBy from 'lodash.groupby'
+import { PendingRcaProposal } from './models/pending-rca-proposal'
 
 export class ActionManager {
   declare multiNetworks: MultiNetworks<Network>
@@ -38,6 +39,7 @@ export class ActionManager {
     logger: Logger,
     models: IndexerManagementModels,
     graphNode: GraphNode,
+    pendingRcaModel?: typeof PendingRcaProposal,
   ): Promise<ActionManager> {
     const actionManager = new ActionManager()
     actionManager.multiNetworks = multiNetworks
@@ -52,6 +54,7 @@ export class ActionManager {
         models,
         graphNode,
         network,
+        pendingRcaModel,
       )
     })
 

packages/indexer-common/src/indexer-management/allocations.ts

@@ -47,6 +47,7 @@ import {
   encodeStopServiceData,
   PaymentTypes,
 } from '@graphprotocol/toolshed'
+import { PendingRcaProposal } from './models/pending-rca-proposal'
 import {
   encodeCollectIndexingRewardsData,
   encodePOIMetadata,
@@ -166,9 +167,16 @@ export class AllocationManager {
     private models: IndexerManagementModels,
     private graphNode: GraphNode,
     private network: Network,
+    private pendingRcaModel?: typeof PendingRcaProposal,
   ) {
     if (this.network.specification.indexerOptions.dipperEndpoint) {
-      this.dipsManager = new DipsManager(this.logger, this.models, this.network, this)
+      this.dipsManager = new DipsManager(
+        this.logger,
+        this.models,
+        this.network,
+        this,
+        this.pendingRcaModel,
+      )
     }
   }
 

packages/indexer-common/src/indexer-management/client.ts

@@ -20,6 +20,7 @@ import {
   Network,
   RulesManager,
 } from '@graphprotocol/indexer-common'
+import { PendingRcaProposal } from './models/pending-rca-proposal'
 
 export interface IndexerManagementResolverContext {
   models: IndexerManagementModels
@@ -574,6 +575,7 @@ export interface IndexerManagementClientOptions {
   multiNetworks: MultiNetworks<Network> | undefined
   defaults: IndexerManagementDefaults
   actionManager?: ActionManager | undefined
+  pendingRcaModel?: typeof PendingRcaProposal
 }
 
 export class IndexerManagementClient extends Client {
@@ -595,7 +597,7 @@ export class IndexerManagementClient extends Client {
 export const createIndexerManagementClient = async (
   options: IndexerManagementClientOptions,
 ): Promise<IndexerManagementClient> => {
-  const { models, graphNode, logger, defaults, multiNetworks } = options
+  const { models, graphNode, logger, defaults, multiNetworks, pendingRcaModel } = options
   const schema = buildSchema(print(SCHEMA_SDL))
   const resolvers = {
     ...indexingRuleResolvers,
@@ -608,7 +610,13 @@ export const createIndexerManagementClient = async (
   }
 
   const actionManager = multiNetworks
-    ? await ActionManager.create(multiNetworks, logger, models, graphNode)
+    ? await ActionManager.create(
+        multiNetworks,
+        logger,
+        models,
+        graphNode,
+        pendingRcaModel,
+      )
     : undefined
 
   const rulesManager = multiNetworks

packages/indexer-common/src/indexer-management/models/index.ts

@@ -11,6 +11,7 @@ export * from './indexing-rule'
 export * from './poi-dispute'
 export * from './action'
 export * from './indexing-agreement'
+export * from './pending-rca-proposal'
 
 export type IndexerManagementModels = IndexingRuleModels &
   CostModelModels &

packages/indexer-common/src/indexer-management/models/pending-rca-proposal.ts

@@ -0,0 +1,55 @@
+import { DataTypes, Sequelize, Model, InferAttributes } from 'sequelize'
+
+export class PendingRcaProposal extends Model<InferAttributes<PendingRcaProposal>> {
+  declare id: string
+  declare signed_payload: Buffer
+  declare version: number
+  declare status: string
+  declare created_at: Date
+  declare updated_at: Date
+}
+
+export const definePendingRcaProposalModel = (
+  sequelize: Sequelize,
+): typeof PendingRcaProposal => {
+  PendingRcaProposal.init(
+    {
+      id: {
+        type: DataTypes.UUID,
+        primaryKey: true,
+      },
+      signed_payload: {
+        type: DataTypes.BLOB,
+        allowNull: false,
+      },
+      version: {
+        type: DataTypes.SMALLINT,
+        allowNull: false,
+        defaultValue: 2,
+      },
+      status: {
+        type: DataTypes.STRING(20),
+        allowNull: false,
+        defaultValue: 'pending',
+      },
+      created_at: {
+        type: DataTypes.DATE,
+        allowNull: false,
+      },
+      updated_at: {
+        type: DataTypes.DATE,
+        allowNull: false,
+      },
+    },
+    {
+      modelName: 'PendingRcaProposal',
+      sequelize,
+      tableName: 'pending_rca_proposals',
+      timestamps: true,
+      createdAt: 'created_at',
+      updatedAt: 'updated_at',
+    },
+  )
+
+  return PendingRcaProposal
+}