feat: implement on-chain collect for DIPs agreements

packages/indexer-agent/src/agent.ts

@@ -727,6 +727,8 @@ export class Agent {
               await operator.dipsManager.matchAgreementAllocations(
                 activeAllocations,
               )
+
+              await operator.dipsManager.collectAgreementPayments()
             }
           },
         )
@@ -1130,15 +1132,14 @@ export class Agent {
               forceAction,
             )
           } else {
-            const expiringAllocations =
-              await this.identifyExpiringAllocations(
-                logger,
-                activeDeploymentAllocations,
-                deploymentAllocationDecision,
-                epoch,
-                maxAllocationDuration,
-                network,
-              )
+            const expiringAllocations = await this.identifyExpiringAllocations(
+              logger,
+              activeDeploymentAllocations,
+              deploymentAllocationDecision,
+              epoch,
+              maxAllocationDuration,
+              network,
+            )
             if (expiringAllocations.length > 0) {
               if (isHorizon) {
                 // Horizon allocations don't need the close/reopen cycle.

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

@@ -296,8 +296,7 @@ export const start = {
         group: 'Query Fees',
       })
       .option('rav-check-interval', {
-        description:
-          'How often the RAV processing loop runs, in seconds',
+        description: 'How often the RAV processing loop runs, in seconds',
         type: 'number',
         default: 900,
         group: 'Query Fees',
@@ -396,6 +395,15 @@ export const start = {
         required: false,
         group: 'Indexing Fees ("DIPs")',
       })
+      .option('dips-collection-target', {
+        description:
+          'Target collection point within the agreement window as a percentage (1-90). ' +
+          'Lower values collect sooner (safer), higher values collect later (fewer txs).',
+        type: 'number',
+        default: 50,
+        required: false,
+        group: 'Indexing Fees ("DIPs")',
+      })
       .check(argv => {
         if (
           !argv['network-subgraph-endpoint'] &&
@@ -472,6 +480,7 @@ export async function createNetworkSpecification(
     ravCollectionInterval: argv.ravCollectionInterval,
     ravCheckInterval: argv.ravCheckInterval,
     dipsEpochsMargin: argv.dipsEpochsMargin,
+    dipsCollectionTarget: argv.dipsCollectionTarget,
   }
 
   const transactionMonitoring = {

packages/indexer-common/src/allocations/graph-tally-collector.ts

@@ -423,11 +423,16 @@ export class GraphTallyCollector {
             const allocationId = collectionIdToAllocationId(rav.rav.rav.collectionId)
             const isActive = await this.isActiveAllocation(allocationId)
             if (isActive && !this.isCooldownExpired(rav.rav.rav.collectionId)) {
-              this.logger.trace('[TAPv2] Skipping active allocation RAV: cooldown not expired', {
-                collectionId: rav.rav.rav.collectionId,
-                lastCollectedAt: this.lastCollectedAt.get(rav.rav.rav.collectionId.toLowerCase()),
-                ravCollectionInterval: this.ravCollectionInterval,
-              })
+              this.logger.trace(
+                '[TAPv2] Skipping active allocation RAV: cooldown not expired',
+                {
+                  collectionId: rav.rav.rav.collectionId,
+                  lastCollectedAt: this.lastCollectedAt.get(
+                    rav.rav.rav.collectionId.toLowerCase(),
+                  ),
+                  ravCollectionInterval: this.ravCollectionInterval,
+                },
+              )
               results.belowThreshold.push(rav)
               return results
             }

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

@@ -174,6 +174,7 @@ export class AllocationManager {
         this.logger,
         this.models,
         this.network,
+        this.graphNode,
         this,
         this.pendingRcaModel,
       )

packages/indexer-common/src/indexing-fees/__tests__/accept-proposals.test.ts

@@ -168,7 +168,8 @@ function createDipsManager(
   models: IndexerManagementModels,
   consumer: PendingRcaConsumer,
 ): DipsManager {
-  const dm = new DipsManager(logger, models, network, null)
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  const dm = new DipsManager(logger, models, network, {} as any, null)
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   ;(dm as any).pendingRcaConsumer = consumer
   return dm
@@ -246,7 +247,8 @@ describe('DipsManager.acceptPendingProposals', () => {
   test('returns early when pendingRcaConsumer is null', async () => {
     const models = createMockModels()
     const network = createMockNetwork()
-    const dm = new DipsManager(logger, models, network, null)
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const dm = new DipsManager(logger, models, network, {} as any, null)
 
     // Should not throw
     await dm.acceptPendingProposals([])

packages/indexer-common/src/indexing-fees/__tests__/agreement-monitor.test.ts

@@ -0,0 +1,103 @@
+/* eslint-disable @typescript-eslint/no-explicit-any,@typescript-eslint/no-unused-vars */
+import {
+  fetchCollectableAgreements,
+  SubgraphIndexingAgreement,
+} from '../agreement-monitor'
+
+const mockQuery = jest.fn()
+const mockNetworkSubgraph = { query: mockQuery } as any
+
+const INDEXER_ADDRESS = '0x1234567890abcdef1234567890abcdef12345678'
+
+describe('fetchCollectableAgreements', () => {
+  beforeEach(() => {
+    jest.clearAllMocks()
+  })
+
+  test('returns agreements in Accepted and CanceledByPayer states', async () => {
+    mockQuery.mockResolvedValueOnce({
+      data: {
+        indexingAgreements: [
+          {
+            id: '0x00000000000000000000000000000001',
+            allocationId: '0xaaaa',
+            subgraphDeploymentId: '0xbbbb',
+            state: 1,
+            lastCollectionAt: '1000',
+            endsAt: '9999999999',
+            maxInitialTokens: '1000000',
+            maxOngoingTokensPerSecond: '100',
+            tokensPerSecond: '50',
+            tokensPerEntityPerSecond: '10',
+            minSecondsPerCollection: 3600,
+            maxSecondsPerCollection: 86400,
+            canceledAt: '0',
+          },
+        ],
+      },
+    })
+
+    const result = await fetchCollectableAgreements(mockNetworkSubgraph, INDEXER_ADDRESS)
+
+    expect(result).toHaveLength(1)
+    expect(result[0].id).toBe('0x00000000000000000000000000000001')
+    expect(result[0].state).toBe(1)
+    expect(mockQuery).toHaveBeenCalledTimes(1)
+  })
+
+  test('returns empty array when no agreements exist', async () => {
+    mockQuery.mockResolvedValueOnce({
+      data: { indexingAgreements: [] },
+    })
+
+    const result = await fetchCollectableAgreements(mockNetworkSubgraph, INDEXER_ADDRESS)
+
+    expect(result).toHaveLength(0)
+  })
+
+  test('paginates through large result sets', async () => {
+    // First page: 1000 results
+    const page1 = Array.from({ length: 1000 }, (_, i) => ({
+      id: `0x${i.toString(16).padStart(32, '0')}`,
+      allocationId: '0xaaaa',
+      subgraphDeploymentId: '0xbbbb',
+      state: 1,
+      lastCollectionAt: '1000',
+      endsAt: '9999999999',
+      maxInitialTokens: '1000000',
+      maxOngoingTokensPerSecond: '100',
+      tokensPerSecond: '50',
+      tokensPerEntityPerSecond: '10',
+      minSecondsPerCollection: 3600,
+      maxSecondsPerCollection: 86400,
+      canceledAt: '0',
+    }))
+    // Second page: 1 result
+    const page2 = [
+      {
+        id: '0x' + 'f'.repeat(32),
+        allocationId: '0xaaaa',
+        subgraphDeploymentId: '0xbbbb',
+        state: 1,
+        lastCollectionAt: '1000',
+        endsAt: '9999999999',
+        maxInitialTokens: '1000000',
+        maxOngoingTokensPerSecond: '100',
+        tokensPerSecond: '50',
+        tokensPerEntityPerSecond: '10',
+        minSecondsPerCollection: 3600,
+        maxSecondsPerCollection: 86400,
+        canceledAt: '0',
+      },
+    ]
+
+    mockQuery
+      .mockResolvedValueOnce({ data: { indexingAgreements: page1 } })
+      .mockResolvedValueOnce({ data: { indexingAgreements: page2 } })
+
+    const result = await fetchCollectableAgreements(mockNetworkSubgraph, INDEXER_ADDRESS)
+
+    expect(result).toHaveLength(1001)
+    expect(mockQuery).toHaveBeenCalledTimes(2)
+  })
+})