chore: implement dipper service client

pcarranzav · pcarranzav · commit 919567a24f31 · 2025-02-14T18:01:09.000-03:00
diff --git a/packages/indexer-common/package.json b/packages/indexer-common/package.json
@@ -22,8 +22,11 @@
     "clean": "rm -rf ./node_modules ./dist ./tsconfig.tsbuildinfo"
   },
   "dependencies": {
+    "@bufbuild/protobuf": "2.2.3",
     "@graphprotocol/common-ts": "2.0.11",
     "@graphprotocol/cost-model": "0.1.18",
+    "@graphprotocol/dips-proto": "0.2.0",
+    "@grpc/grpc-js": "^1.12.6",
     "@semiotic-labs/tap-contracts-bindings": "^1.2.1",
     "@thi.ng/heaps": "1.2.38",
     "@types/lodash.clonedeep": "^4.5.7",
diff --git a/packages/indexer-common/src/indexing-fees/dipper-service-client.ts b/packages/indexer-common/src/indexing-fees/dipper-service-client.ts
@@ -0,0 +1,162 @@
+import { Client, credentials } from '@grpc/grpc-js'
+import { UnaryCallback } from '@grpc/grpc-js/build/src/client'
+import { DipperServiceClientImpl } from '@graphprotocol/dips-proto/generated/gateway'
+import { Wallet } from 'ethers'
+import {
+  _TypedDataEncoder,
+  arrayify,
+  defaultAbiCoder,
+  recoverAddress,
+} from 'ethers/lib/utils'
+import { toAddress } from '@graphprotocol/common-ts'
+
+type RpcImpl = (service: string, method: string, data: Uint8Array) => Promise<Uint8Array>
+
+interface Rpc {
+  request: RpcImpl
+}
+
+export const domainSalt =
+  '0xb4632c657c26dce5d4d7da1d65bda185b14ff8f905ddbb03ea0382ed06c5ef28'
+export const chainId = 0xa4b1 // 42161
+export const cancelAgreementDomain = {
+  name: 'Graph Protocol Indexing Agreement Cancellation',
+  version: '0',
+  chainId: chainId,
+  salt: domainSalt,
+}
+export const cancelAgreementTypes = {
+  CancellationRequest: [{ name: 'agreement_id', type: 'bytes16' }],
+}
+
+export const collectPaymentsDomain = {
+  name: 'Graph Protocol Indexing Agreement Collection',
+  version: '0',
+  chainId: chainId,
+  salt: domainSalt,
+}
+export const collectPaymentsTypes = {
+  CollectionRequest: [
+    { name: 'agreement_id', type: 'bytes16' },
+    { name: 'allocation_id', type: 'address' },
+    { name: 'entity_count', type: 'uint64' },
+  ],
+}
+
+export const createSignedCancellationRequest = async (
+  agreementId: string,
+  wallet: Wallet,
+): Promise<Uint8Array> => {
+  const signature = await wallet._signTypedData(
+    cancelAgreementDomain,
+    cancelAgreementTypes,
+    { agreement_id: agreementId },
+  )
+  return arrayify(
+    defaultAbiCoder.encode(['tuple(bytes16)', 'bytes'], [[agreementId], signature]),
+  )
+}
+
+export const createSignedCollectionRequest = async (
+  agreementId: string,
+  allocationId: string,
+  entityCount: number,
+  wallet: Wallet,
+): Promise<Uint8Array> => {
+  const signature = await wallet._signTypedData(
+    collectPaymentsDomain,
+    collectPaymentsTypes,
+    { agreement_id: agreementId, allocation_id: allocationId, entity_count: entityCount },
+  )
+  return arrayify(
+    defaultAbiCoder.encode(
+      ['tuple(bytes16, address, uint64)', 'bytes'],
+      [[agreementId, allocationId, entityCount], signature],
+    ),
+  )
+}
+
+export const decodeTapReceipt = (receipt: Uint8Array, verifyingContract: string) => {
+  const [message, signature] = defaultAbiCoder.decode(
+    ['tuple(address,uint64,uint64,uint128)', 'bytes'],
+    receipt,
+  )
+
+  const [allocationId, timestampNs, nonce, value] = message
+
+  // Recover the signer address from the signature
+  // compute the EIP-712 digest of the message
+  const domain = {
+    name: 'TAP',
+    version: '1',
+    chainId: chainId,
+    verifyingContract,
+  }
+
+  const types = {
+    Receipt: [
+      { name: 'allocation_id', type: 'address' },
+      { name: 'timestamp_ns', type: 'uint64' },
+      { name: 'nonce', type: 'uint64' },
+      { name: 'value', type: 'uint128' },
+    ],
+  }
+
+  const digest = _TypedDataEncoder.hash(domain, types, {
+    allocation_id: allocationId,
+    timestamp_ns: timestampNs,
+    nonce: nonce,
+    value: value,
+  })
+  const signerAddress = recoverAddress(digest, signature)
+  return {
+    allocation_id: allocationId,
+    signer_address: toAddress(signerAddress),
+    signature: signature,
+    timestamp_ns: timestampNs,
+    nonce: nonce,
+    value: value,
+  }
+}
+
+export const createRpc = (url: string): Rpc => {
+  const client = new Client(url, credentials.createInsecure())
+  const request: RpcImpl = (service, method, data) => {
+    // Conventionally in gRPC, the request path looks like
+    //   "package.names.ServiceName/MethodName",
+    // we therefore construct such a string
+    const path = `/${service}/${method}`
+
+    return new Promise((resolve, reject) => {
+      // makeUnaryRequest transmits the result (and error) with a callback
+      // transform this into a promise!
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      const resultCallback: UnaryCallback<any> = (err, res) => {
+        if (err) {
+          return reject(err)
+        }
+        resolve(res)
+      }
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      function passThrough(argument: any) {
+        return argument
+      }
+
+      // Using passThrough as the deserialize functions
+      client.makeUnaryRequest(
+        path,
+        (d) => Buffer.from(d),
+        passThrough,
+        data,
+        resultCallback,
+      )
+    })
+  }
+
+  return { request }
+}
+
+export const createDipperServiceClient = (url: string) => {
+  const rpc = createRpc(url)
+  return new DipperServiceClientImpl(rpc)
+}
diff --git a/packages/indexer-common/src/indexing-fees/dips.ts b/packages/indexer-common/src/indexing-fees/dips.ts
@@ -11,14 +11,35 @@ import {
 } from '@graphprotocol/indexer-common'
 import { Op } from 'sequelize'
 
+import {
+  createDipperServiceClient,
+  createSignedCancellationRequest,
+  createSignedCollectionRequest,
+  decodeTapReceipt,
+} from './dipper-service-client'
+import {
+  CollectPaymentStatus,
+  DipperServiceClientImpl,
+} from '@graphprotocol/dips-proto/generated/gateway'
+import { IndexingAgreement } from '../indexer-management/models/indexing-agreement'
+
 export class DipsManager {
+  private dipperServiceClient: DipperServiceClientImpl
+
   constructor(
     private logger: Logger,
     private models: IndexerManagementModels,
     private graphNode: GraphNode,
     private network: Network,
     private parent: AllocationManager | null,
-  ) {}
+  ) {
+    if (!this.network.specification.indexerOptions.dipperEndpoint) {
+      throw new Error('dipperEndpoint is not set')
+    }
+    this.dipperServiceClient = createDipperServiceClient(
+      this.network.specification.indexerOptions.dipperEndpoint,
+    )
+  }
   // Cancel an agreement associated to an allocation if it exists
   async tryCancelAgreement(allocationId: string) {
     const agreement = await this.models.IndexingAgreement.findOne({
@@ -28,8 +49,22 @@ export class DipsManager {
       },
     })
     if (agreement) {
-      // TODO use dips-proto to cancel agreement via grpc
-      // Mark the agreement as cancelled
+      try {
+        const cancellation = await createSignedCancellationRequest(
+          agreement.id,
+          this.network.wallet,
+        )
+        await this.dipperServiceClient.CancelAgreement({
+          version: 1,
+          signedCancellation: cancellation,
+        })
+
+        // Mark the agreement as cancelled
+        agreement.cancelled_at = new Date()
+        await agreement.save()
+      } catch (error) {
+        this.logger.error(`Error cancelling agreement ${agreement.id}`, { error })
+      }
     }
   }
   // Update the current and last allocation ids for an agreement if it exists
@@ -62,18 +97,50 @@ export class DipsManager {
     })
     for (const agreement of outstandingAgreements) {
       if (agreement.last_allocation_id) {
-        await this.tryCollectPayment(agreement.last_allocation_id)
+        await this.tryCollectPayment(agreement)
       } else {
         // This should never happen as we check for this in the query
         this.logger.error(`Agreement ${agreement.id} has no last allocation id`)
       }
     }
   }
-  async tryCollectPayment(lastAllocationId: string) {
-    // TODO: use dips-proto to collect payment via grpc
-
-    // TODO: store the receipt in the database
-    // (tap-agent will take care of aggregating it into a RAV)
+  async tryCollectPayment(agreement: IndexingAgreement) {
+    if (!agreement.last_allocation_id) {
+      this.logger.error(`Agreement ${agreement.id} has no last allocation id`)
+      return
+    }
+    const entityCount = 0 // TODO: get entity count from graph node
+    const collection = await createSignedCollectionRequest(
+      agreement.id,
+      agreement.last_allocation_id,
+      entityCount,
+      this.network.wallet,
+    )
+    try {
+      const response = await this.dipperServiceClient.CollectPayment({
+        version: 1,
+        signedCollection: collection,
+      })
+      if (response.status === CollectPaymentStatus.ACCEPT) {
+        if (!this.network.tapCollector) {
+          throw new Error('TapCollector not initialized')
+        }
+        // Store the tap receipt in the database
+        const tapReceipt = decodeTapReceipt(
+          response.tapReceipt,
+          this.network.tapCollector?.tapContracts.tapVerifier.address,
+        )
+        await this.network.queryFeeModels.scalarTapReceipts.create(tapReceipt)
+      } else {
+        this.logger.error(`Error collecting payment for agreement ${agreement.id}`, {
+          error: response.status,
+        })
+      }
+    } catch (error) {
+      this.logger.error(`Error collecting payment for agreement ${agreement.id}`, {
+        error,
+      })
+    }
 
     // Mark the agreement as having had a payment collected
     await this.models.IndexingAgreement.update(
@@ -82,7 +149,7 @@ export class DipsManager {
       },
       {
         where: {
-          last_allocation_id: lastAllocationId,
+          id: agreement.id,
         },
       },
     )
diff --git a/packages/indexer-common/src/network.ts b/packages/indexer-common/src/network.ts
@@ -52,7 +52,7 @@ export class Network {
   specification: spec.NetworkSpecification
   paused: Eventual<boolean>
   isOperator: Eventual<boolean>
-
+  queryFeeModels: QueryFeeModels
   private constructor(
     logger: Logger,
     contracts: NetworkContracts,
@@ -66,6 +66,7 @@ export class Network {
     specification: spec.NetworkSpecification,
     paused: Eventual<boolean>,
     isOperator: Eventual<boolean>,
+    queryFeeModels: QueryFeeModels,
   ) {
     this.logger = logger
     this.contracts = contracts
@@ -79,6 +80,7 @@ export class Network {
     this.specification = specification
     this.paused = paused
     this.isOperator = isOperator
+    this.queryFeeModels = queryFeeModels
   }
 
   static async create(
@@ -345,6 +347,7 @@ export class Network {
       specification,
       paused,
       isOperator,
+      queryFeeModels,
     )
   }
 
diff --git a/packages/indexer-common/src/query-fees/models.ts b/packages/indexer-common/src/query-fees/models.ts
@@ -13,11 +13,20 @@ export interface ScalarTapReceiptsAttributes {
   value: bigint
   error_log?: string
 }
+export interface ScalarTapReceiptsCreationAttributes {
+  allocation_id: Address
+  signer_address: Address
+  signature: Uint8Array
+  timestamp_ns: bigint
+  nonce: bigint
+  value: bigint
+}
+
 export class ScalarTapReceipts
-  extends Model<ScalarTapReceiptsAttributes>
+  extends Model<ScalarTapReceiptsAttributes, ScalarTapReceiptsCreationAttributes>
   implements ScalarTapReceiptsAttributes
 {
-  public id!: number
+  public id!: CreationOptional<number>
   public allocation_id!: Address
   public signer_address!: Address
   public signature!: Uint8Array
diff --git a/yarn.lock b/yarn.lock
