fix: rename to collectorId and use collectorId on subgraph service

Author: Maikol

packages/horizon/contracts/interfaces/IPaymentsCollector.sol

@@ -24,7 +24,7 @@ interface IPaymentsCollector {
      */
     event PaymentCollected(
         IGraphPayments.PaymentTypes paymentType,
-        address indexed paymentId,
+        bytes32 indexed collectorId,
         address indexed payer,
         address receiver,
         address indexed dataService,

packages/horizon/contracts/interfaces/ITAPCollector.sol

@@ -26,7 +26,7 @@ interface ITAPCollector is IPaymentsCollector {
     /// @notice The Receipt Aggregate Voucher (RAV) struct
     struct ReceiptAggregateVoucher {
         // The ID of the payment
-        address paymentId;
+        bytes32 collectorId;
         // The address of the payer the RAV was issued by
         address payer;
         // The address of the data service the RAV was issued to

packages/horizon/contracts/payments/collectors/TAPCollector.sol

@@ -36,7 +36,7 @@ contract TAPCollector is EIP712, GraphDirectory, ITAPCollector {
     mapping(address signer => PayerAuthorization authorizedSigner) public authorizedSigners;
 
     /// @notice Tracks the amount of tokens already collected by a data service from a payer to a receiver
-    mapping(address dataService => mapping(address paymentId => mapping(address receiver => mapping(address payer => uint256 tokens))))
+    mapping(address dataService => mapping(bytes32 collectorId => mapping(address receiver => mapping(address payer => uint256 tokens))))
         public tokensCollected;
 
     /// @notice The duration (in seconds) in which a signer is thawing before they can be revoked
@@ -182,7 +182,7 @@ contract TAPCollector is EIP712, GraphDirectory, ITAPCollector {
         address payer = authorizedSigners[signer].payer;
         require(signedRAV.rav.payer == payer, TAPCollectorInvalidRAVPayer(payer, signedRAV.rav.payer));
 
-        address paymentId = signedRAV.rav.paymentId;
+        bytes32 collectorId = signedRAV.rav.collectorId;
         address dataService = signedRAV.rav.dataService;
         address receiver = signedRAV.rav.serviceProvider;
 
@@ -200,7 +200,7 @@ contract TAPCollector is EIP712, GraphDirectory, ITAPCollector {
         uint256 tokensToCollect = 0;
         {
             uint256 tokensRAV = signedRAV.rav.valueAggregate;
-            uint256 tokensAlreadyCollected = tokensCollected[dataService][paymentId][receiver][payer];
+            uint256 tokensAlreadyCollected = tokensCollected[dataService][collectorId][receiver][payer];
             require(
                 tokensRAV > tokensAlreadyCollected,
                 TAPCollectorInconsistentRAVTokens(tokensRAV, tokensAlreadyCollected)
@@ -218,11 +218,11 @@ contract TAPCollector is EIP712, GraphDirectory, ITAPCollector {
         }
 
         if (tokensToCollect > 0) {
-            tokensCollected[dataService][paymentId][receiver][payer] += tokensToCollect;
+            tokensCollected[dataService][collectorId][receiver][payer] += tokensToCollect;
             _graphPaymentsEscrow().collect(_paymentType, payer, receiver, tokensToCollect, dataService, dataServiceCut);
         }
 
-        emit PaymentCollected(_paymentType, paymentId, payer, receiver, dataService, tokensToCollect);
+        emit PaymentCollected(_paymentType, collectorId, payer, receiver, dataService, tokensToCollect);
         emit RAVCollected(
             payer,
             dataService,

packages/horizon/test/payments/tap-collector/TAPCollector.t.sol

@@ -142,7 +142,7 @@ contract TAPCollectorTest is HorizonStakingSharedTest, PaymentsEscrowSharedTest
         (address _payer, , ) = tapCollector.authorizedSigners(_signer);
         uint256 tokensAlreadyCollected = tapCollector.tokensCollected(
             signedRAV.rav.dataService,
-            signedRAV.rav.paymentId,
+            signedRAV.rav.collectorId,
             signedRAV.rav.serviceProvider,
             _payer
         );
@@ -153,7 +153,7 @@ contract TAPCollectorTest is HorizonStakingSharedTest, PaymentsEscrowSharedTest
         vm.expectEmit(address(tapCollector));
         emit IPaymentsCollector.PaymentCollected(
             _paymentType,
-            signedRAV.rav.paymentId,
+            signedRAV.rav.collectorId,
             _payer,
             signedRAV.rav.serviceProvider,
             signedRAV.rav.dataService,
@@ -175,7 +175,7 @@ contract TAPCollectorTest is HorizonStakingSharedTest, PaymentsEscrowSharedTest
 
         uint256 tokensCollectedAfter = tapCollector.tokensCollected(
             signedRAV.rav.dataService,
-            signedRAV.rav.paymentId,
+            signedRAV.rav.collectorId,
             signedRAV.rav.serviceProvider,
             _payer
         );

packages/horizon/test/payments/tap-collector/collect/collect.t.sol

@@ -48,7 +48,7 @@ contract TAPCollectorCollectTest is TAPCollectorTest {
     ) private pure returns (ITAPCollector.ReceiptAggregateVoucher memory rav) {
         return
             ITAPCollector.ReceiptAggregateVoucher({
-                paymentId: _allocationId,
+                collectorId: bytes32(uint256(uint160(_allocationId))),
                 payer: _payer,
                 dataService: _collector,
                 serviceProvider: _indexer,
@@ -408,7 +408,12 @@ contract TAPCollectorCollectTest is TAPCollectorTest {
         bytes memory data = _getQueryFeeEncodedData(signerPrivateKey, params);
 
         resetPrank(users.verifier);
-        uint256 tokensAlreadyCollected = tapCollector.tokensCollected(users.verifier, _allocationId, users.indexer, users.gateway);
+        uint256 tokensAlreadyCollected = tapCollector.tokensCollected(
+            users.verifier,
+            bytes32(uint256(uint160(_allocationId))),
+            users.indexer,
+            users.gateway
+        );
         tokensToCollect = bound(tokensToCollect, tokens - tokensAlreadyCollected + 1, type(uint128).max);
 
         vm.expectRevert(
@@ -448,7 +453,12 @@ contract TAPCollectorCollectTest is TAPCollectorTest {
 
         for (uint256 i = 0; i < numAllocations; i++) {
             assertEq(
-                tapCollector.tokensCollected(collectTestParams[i].collector, collectTestParams[i].allocationId, collectTestParams[i].indexer, collectTestParams[i].payer),
+                tapCollector.tokensCollected(
+                    collectTestParams[i].collector,
+                    bytes32(uint256(uint160(collectTestParams[i].allocationId))),
+                    collectTestParams[i].indexer,
+                    collectTestParams[i].payer
+                ),
                 collectTestParams[i].tokens,
                 "Incorrect tokens collected for allocation"
             );

packages/subgraph-service/contracts/SubgraphService.sol

@@ -536,7 +536,13 @@ contract SubgraphService is
      */
     function _collectQueryFees(ITAPCollector.SignedRAV memory _signedRav) private returns (uint256 feesCollected) {
         address indexer = _signedRav.rav.serviceProvider;
-        address allocationId = abi.decode(_signedRav.rav.metadata, (address));
+
+        // Check that the collectorId fits in an address (160 bits)
+        require(
+            uint256(_signedRav.rav.collectorId) < (1 << 160),
+            SubgraphServiceInvalidCollectorId(_signedRav.rav.collectorId)
+        );
+        address allocationId = address(uint160(uint256(_signedRav.rav.collectorId)));
         Allocation.State memory allocation = _allocations.get(allocationId);
 
         // Check RAV is consistent - RAV indexer must match the allocation's indexer

packages/subgraph-service/contracts/interfaces/ISubgraphService.sol

@@ -124,6 +124,12 @@ interface ISubgraphService is IDataServiceFees {
      */
     error SubgraphServiceInvalidZeroStakeToFeesRatio();
 
+    /**
+     * @notice Thrown when collectorId is too large for an address
+     * @param collectorId The collectorId that is too large
+     */
+    error SubgraphServiceInvalidCollectorId(bytes32 collectorId);
+
     /**
      * @notice Force close a stale allocation
      * @dev This function can be permissionlessly called when the allocation is stale. This