Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ This subgraph tracks the full lifecycle of [Direct Indexing Payments](https://gi
<tr><td><code>IndexingAgreementCanceled</code></td><td>Agreement canceled by payer or indexer</td></tr>
<tr><td><code>IndexingAgreementUpdated</code></td><td>Agreement terms or allocation changed</td></tr>
<tr><td><code>IndexingFeesCollectedV1</code></td><td>Fees collected against an active agreement</td></tr>
<tr><td><code>ServiceProviderRegistered</code></td><td>Indexer registers (or re-registers) its service URL on-chain</td></tr>
</table>

<br>
Expand Down
9 changes: 9 additions & 0 deletions abis/SubgraphService.json
Original file line number Diff line number Diff line change
Expand Up @@ -54,5 +54,14 @@
],
"name": "IndexingFeesCollectedV1",
"type": "event"
},
{
"anonymous": false,
"inputs": [
{ "indexed": true, "internalType": "address", "name": "serviceProvider", "type": "address" },
{ "indexed": false, "internalType": "bytes", "name": "data", "type": "bytes" }
],
"name": "ServiceProviderRegistered",
"type": "event"
}
]
4 changes: 2 additions & 2 deletions config/arbitrum-one.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"network": "arbitrum-one",
"subgraphServiceAddress": "0x0000000000000000000000000000000000000000",
"subgraphServiceAddress": "0xb2Bb92d0DE618878E438b55D5846cfecD9301105",
"recurringCollectorAddress": "0x0000000000000000000000000000000000000000",
"startBlock": 0
"startBlock": 397492865
}
6 changes: 3 additions & 3 deletions config/arbitrum-sepolia.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"network": "arbitrum-sepolia",
"subgraphServiceAddress": "0x0000000000000000000000000000000000000000",
"recurringCollectorAddress": "0x0000000000000000000000000000000000000000",
"startBlock": 0
"subgraphServiceAddress": "0xc24A3dAC5d06d771f657A48B20cE1a671B78f26b",
"recurringCollectorAddress": "0x0b18befc60455121ad66ae6e4a647955fcde3900",
"startBlock": 201859812
}
2 changes: 1 addition & 1 deletion config/hardhat.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"network": "hardhat",
"subgraphServiceAddress": "0x0000000000000000000000000000000000000000",
"subgraphServiceAddress": "0xCf7Ed3AccA5a467e9e704C703E8D87F634fB0Fc9",
"recurringCollectorAddress": "0x0000000000000000000000000000000000000000",
"startBlock": 0
}
24 changes: 24 additions & 0 deletions schema.graphql
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@ type IndexingAgreement @entity(immutable: false) {
payer: Bytes!
"The indexer address"
indexer: Bytes!
"Link to the indexer's registration record (URL etc.). Null until the indexer's ServiceProviderRegistered event has been indexed; resolves automatically once it is. Lets consumers read the indexer URL alongside the agreement in one query instead of cross-querying the network subgraph."
indexerInfo: Indexer
"The allocation linked to this agreement"
allocationId: Bytes!
"The subgraph deployment ID"
Expand Down Expand Up @@ -82,6 +84,28 @@ type IndexerDeploymentLatest @entity(immutable: false) {
blockTimestamp: BigInt!
}

# Indexer (service provider) registration record, keyed by the indexer address.
#
# Sourced from the SubgraphService contract's ServiceProviderRegistered event,
# whose payload decodes to (string url, string geohash, address paymentsDestination).
# We store the url so consumers that hold an agreement's indexer address can read
# the indexer's service URL from this same subgraph instead of cross-querying the
# network subgraph.
#
# Mutable because an indexer can re-register at any time to change its URL; the
# contract overwrites its stored record and re-emits the event, so we treat this
# as last-write-wins.
type Indexer @entity(immutable: false) {
"The indexer (service provider) address"
id: Bytes!
"Service URL from the latest ServiceProviderRegistered event"
url: String!
"Block number of the registration that last set this URL. Consumers polling for URL changes filter with `lastUpdatedAtBlock_gt` since their last seen block."
lastUpdatedAtBlock: BigInt!
"Transaction hash of the registration that last set this URL. Auditable link back to the on-chain register call."
lastUpdatedAtTx: Bytes!
}

# Latest stored offer per agreementId, keyed by bytes16 agreement ID.
# Dipper queries this entity as an idempotency gate -- avoids re-submitting
# an offer after a crashed-mid-flight restart where the on-chain tx landed
Expand Down
29 changes: 20 additions & 9 deletions src/helpers.ts
Original file line number Diff line number Diff line change
@@ -1,11 +1,10 @@
import { Address, BigInt, Bytes } from '@graphprotocol/graph-ts'
import { Address, BigInt, ByteArray, Bytes } from '@graphprotocol/graph-ts'
import { IndexingAgreement } from '../generated/schema'

export const BIGINT_ZERO = BigInt.fromI32(0)
// 32-byte zero — same rationale as the canceledBy = Address.zero() pattern
// below. Bytes.empty() serializes with unpredictable padding on non-nullable
// fields; a fixed-length sentinel gives consumers a deterministic value to
// check against when no on-chain tx exists yet.
// 32-byte zero sentinel. Bytes.empty() serializes with unpredictable padding on
// non-nullable fields; a fixed-length value gives consumers something
// deterministic to check against when no on-chain tx exists yet.
export const BYTES32_ZERO = Bytes.fromHexString('0x' + '00'.repeat(32)) as Bytes

export function createOrLoadIndexingAgreement(agreementId: Bytes): IndexingAgreement {
Expand All @@ -30,13 +29,25 @@ export function createOrLoadIndexingAgreement(agreementId: Bytes): IndexingAgree
agreement.lastUpdatedAt = BIGINT_ZERO
agreement.canceledAt = BIGINT_ZERO
agreement.canceledAtTx = BYTES32_ZERO
// Default to 20-byte zero address rather than Bytes.empty(). Graph-node
// serializes empty Bytes on non-nullable fields with unpredictable
// padding (observed as "0x00000000" in practice), which breaks strict
// 20-byte-address parsers on the consumer side.
// Default to 20-byte zero address rather than Bytes.empty(): graph-node
// serializes empty Bytes on non-nullable fields with unpredictable padding
// (observed as "0x00000000"), breaking strict 20-byte-address parsers downstream.
agreement.canceledBy = Address.zero() as Bytes
agreement.tokensCollected = BIGINT_ZERO
agreement.lastStateChangeBlock = BIGINT_ZERO
}
return agreement
}

// ethereum.decode expects a standalone dynamic tuple to begin with a 32-byte
// offset word, but abi.encode(url, geohash, dest) on-chain omits that outer
// offset. Prepend it so the registration payload decodes as (string,string,address).
export function tuplePrefixBytes(input: Bytes): Bytes {
let tuplePrefix = ByteArray.fromHexString(
'0x0000000000000000000000000000000000000000000000000000000000000020',
)
let inputAsTuple = new Uint8Array(tuplePrefix.length + input.length)
inputAsTuple.set(tuplePrefix, 0)
inputAsTuple.set(input.subarray(0), tuplePrefix.length)
return Bytes.fromUint8Array(inputAsTuple)
}
40 changes: 15 additions & 25 deletions src/recurringCollector.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,13 +9,9 @@ import {
} from '../generated/RecurringCollector/RecurringCollector'
import { createOrLoadIndexingAgreement, BIGINT_ZERO } from './helpers'

// CancelAgreementBy enum from IRecurringCollector.sol:
// 0 = ServiceProvider, 1 = Payer, 2 = ThirdParty
// The contract treats anything that isn't Payer as ServiceProvider when
// emitting the SubgraphService-side IndexingAgreementCanceled event, so we
// mirror that mapping here. ThirdParty (2) is currently unreachable from
// SubgraphService — adding the explicit branch documents the contract's
// intent and keeps the mapping correct if a future data service surfaces it.
// CancelAgreementBy enum from IRecurringCollector.sol: 0 = ServiceProvider,
// 1 = Payer, 2 = ThirdParty. The contract treats anything that isn't Payer as
// ServiceProvider; ThirdParty is currently unreachable from SubgraphService.
const CANCEL_BY_PAYER: i32 = 1

export function handleAgreementAccepted(event: AgreementAccepted): void {
Expand All @@ -25,6 +21,9 @@ export function handleAgreementAccepted(event: AgreementAccepted): void {
// accept(), so the event's block timestamp is the canonical value.
agreement.payer = event.params.payer
agreement.indexer = event.params.serviceProvider
// Link to the Indexer registration record by address. Safe to set even if that
// record isn't indexed yet — the reference resolves once the indexer registers.
agreement.indexerInfo = event.params.serviceProvider
agreement.state = 'Accepted'
agreement.acceptedAt = event.block.timestamp
agreement.lastCollectionAt = event.block.timestamp
Expand All @@ -44,12 +43,9 @@ export function handleAgreementCanceled(event: AgreementCanceled): void {
let agreement = IndexingAgreement.load(event.params.agreementId)
if (agreement == null) return

// The actual canceler address is written by
// subgraphService.handleIndexingAgreementCanceled, which fires in the
// same transaction and reads the SubgraphService event's
// canceledOnBehalfOf parameter. The contract sets
// `agreement.canceledAt = uint64(block.timestamp)` inside cancel(), so
// the event's block timestamp is the canonical value.
// The actual canceler address is written by handleIndexingAgreementCanceled,
// which fires in the same tx and reads canceledOnBehalfOf. The contract sets
// canceledAt = block.timestamp inside cancel(), so the event timestamp is canonical.
if (event.params.canceledBy == CANCEL_BY_PAYER) {
agreement.state = 'CanceledByPayer'
} else {
Expand Down Expand Up @@ -85,13 +81,9 @@ export function handleRCACollected(event: RCACollected): void {
}

export function handleOfferStored(event: OfferStoredEvent): void {
// OfferStored fires once per agreementId for OFFER_TYPE_NEW and again
// for each OFFER_TYPE_UPDATE that changes the stored offer hash. The
// contract overwrites $.rcaOffers / $.rcauOffers in-place, so dipper's
// idempotency gate has to see the latest terms — keep the entity
// mutable and refresh offerType / offerHash on every event. `createdAt`
// fields stay pinned to the first OFFER_TYPE_NEW so consumers can
// distinguish initial offer from subsequent updates.
// Fires for OFFER_TYPE_NEW and again for each OFFER_TYPE_UPDATE that changes the
// stored hash. The contract overwrites in place, so refresh offerType/offerHash
// every event; createdAt stays pinned to the first NEW to mark the initial offer.
let offer = Offer.load(event.params.agreementId)
if (offer == null) {
offer = new Offer(event.params.agreementId)
Expand All @@ -107,11 +99,9 @@ export function handleOfferStored(event: OfferStoredEvent): void {
}

export function handleOfferCancelled(event: OfferCancelledEvent): void {
// OfferCancelled fires when a payer (or any signer at SCOPE_SIGNED)
// cancels a stored RCA/RCAU offer. The contract deletes the on-chain
// entry, so dipper's idempotency gate must treat the Offer as no longer
// live. Set canceledAt to the event's block timestamp; consumers query
// `canceledAt > 0` to decide "safe to re-submit".
// Fires when a signer cancels a stored RCA/RCAU offer; the contract deletes the
// on-chain entry, so mark the Offer not-live by stamping canceledAt with the event
// timestamp. Consumers query `canceledAt > 0` to decide "safe to re-submit".
let offer = Offer.load(event.params.agreementId)
if (offer == null) return
offer.canceledAt = event.block.timestamp
Expand Down
48 changes: 41 additions & 7 deletions src/subgraphService.ts
Original file line number Diff line number Diff line change
@@ -1,15 +1,22 @@
import { ethereum } from '@graphprotocol/graph-ts'
import { ethereum, log } from '@graphprotocol/graph-ts'
import {
IndexingAgreementAccepted as AcceptedEvent,
IndexingAgreementCanceled as CanceledEvent,
IndexingAgreementUpdated as UpdatedEvent,
IndexingFeesCollectedV1 as FeesCollectedEvent,
ServiceProviderRegistered as RegisteredEvent,
} from '../generated/SubgraphService/SubgraphService'
import { IndexerDeploymentLatest, IndexingFeeCollection } from '../generated/schema'
import { createOrLoadIndexingAgreement } from './helpers'
import { Indexer, IndexerDeploymentLatest, IndexingFeeCollection } from '../generated/schema'
import { createOrLoadIndexingAgreement, tuplePrefixBytes } from './helpers'

export function handleIndexingAgreementAccepted(event: AcceptedEvent): void {
let agreement = createOrLoadIndexingAgreement(event.params.agreementId)
// The RecurringCollector's AgreementAccepted handler sets these too (same tx,
// same values); setting them here as well keeps the agreement's identity and
// its indexer link intact even if the collector data source is misconfigured.
agreement.indexer = event.params.indexer
agreement.indexerInfo = event.params.indexer
agreement.payer = event.params.payer
agreement.allocationId = event.params.allocationId
agreement.subgraphDeploymentId = event.params.subgraphDeploymentId
agreement.acceptedAtTx = event.transaction.hash
Expand All @@ -27,10 +34,9 @@ export function handleIndexingAgreementAccepted(event: AcceptedEvent): void {

export function handleIndexingAgreementCanceled(event: CanceledEvent): void {
let agreement = createOrLoadIndexingAgreement(event.params.agreementId)
// canceledOnBehalfOf is the actual signer that initiated the cancel. For
// operator-initiated cancels this is the operator, not the payer/indexer
// directly. Dipper's chain_listener compares this to its own signer
// address to decide CanceledByRequester vs CanceledByIndexer.
// canceledOnBehalfOf is the signer that initiated the cancel — for
// operator-initiated cancels the operator, not the payer/indexer. Dipper
// compares it to its own signer to decide CanceledByRequester vs CanceledByIndexer.
agreement.canceledBy = event.params.canceledOnBehalfOf
agreement.canceledAtTx = event.transaction.hash
agreement.lastStateChangeBlock = event.block.number
Expand Down Expand Up @@ -82,3 +88,31 @@ export function handleIndexingFeesCollectedV1(event: FeesCollectedEvent): void {
latest.blockTimestamp = event.block.timestamp
latest.save()
}

export function handleServiceProviderRegistered(event: RegisteredEvent): void {
// The url/geohash/paymentsDestination ride inside the opaque `data` blob the
// contract re-emits; decode it the way the contract encoded it. We only keep
// the url (element 0); geohash and paymentsDestination are intentionally dropped.
let decoded = ethereum.decode('(string,string,address)', tuplePrefixBytes(event.params.data))
// decode() of a tuple type returns either null or a tuple-kind value, so a null
// check is sufficient — matching the (uint256,uint256) decode sites above.
if (decoded == null) {
// Surface a malformed payload rather than swallowing it; the indexer simply
// keeps whatever url was last decoded (or none).
log.warning('ServiceProviderRegistered failed to decode for {} (data: {})', [
event.params.serviceProvider.toHexString(),
event.params.data.toHexString(),
])
return
}

let indexer = Indexer.load(event.params.serviceProvider)
if (indexer == null) {
indexer = new Indexer(event.params.serviceProvider)
}
// Re-registration overwrites the on-chain record, so this is last-write-wins.
indexer.url = decoded.toTuple()[0].toString()
indexer.lastUpdatedAtBlock = event.block.number
indexer.lastUpdatedAtTx = event.transaction.hash
indexer.save()
}
3 changes: 3 additions & 0 deletions subgraph.template.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ dataSources:
- IndexingAgreement
- IndexingFeeCollection
- IndexerDeploymentLatest
- Indexer
abis:
- name: SubgraphService
file: ./abis/SubgraphService.json
Expand All @@ -31,6 +32,8 @@ dataSources:
handler: handleIndexingAgreementUpdated
- event: IndexingFeesCollectedV1(indexed address,indexed address,indexed bytes16,address,bytes32,uint256,uint256,uint256,bytes32,uint256,bytes)
handler: handleIndexingFeesCollectedV1
- event: ServiceProviderRegistered(indexed address,bytes)
handler: handleServiceProviderRegistered
file: ./src/subgraphService.ts
- kind: ethereum
name: RecurringCollector
Expand Down
74 changes: 73 additions & 1 deletion tests/recurringCollector.test.ts
Original file line number Diff line number Diff line change
@@ -1,13 +1,20 @@
import { assert, describe, test, clearStore, afterEach, newMockEvent } from 'matchstick-as'
import { Address, Bytes, BigInt, ethereum } from '@graphprotocol/graph-ts'
import { handleOfferStored, handleOfferCancelled } from '../src/recurringCollector'
import {
handleAgreementAccepted,
handleOfferStored,
handleOfferCancelled,
} from '../src/recurringCollector'
import {
AgreementAccepted as AgreementAcceptedEvent,
OfferStored as OfferStoredEvent,
OfferCancelled as OfferCancelledEvent,
} from '../generated/RecurringCollector/RecurringCollector'

const PAYER = Address.fromString('0x0000000000000000000000000000000000000002')
const CALLER = Address.fromString('0x0000000000000000000000000000000000000003')
const SERVICE_PROVIDER = Address.fromString('0x0000000000000000000000000000000000000001')
const DATA_SERVICE = Address.fromString('0x0000000000000000000000000000000000000005')
const AGREEMENT_ID = Bytes.fromHexString('0x0102030405060708090a0b0c0d0e0f10')

// OFFER_TYPE_NEW from IAgreementCollector.sol after the audit reshuffle
Expand Down Expand Up @@ -41,6 +48,54 @@ function createOfferStoredEvent(
return event
}

function createAgreementAcceptedEvent(
serviceProvider: Address,
agreementId: Bytes,
): AgreementAcceptedEvent {
let event = changetype<AgreementAcceptedEvent>(newMockEvent())

event.parameters = new Array()
event.parameters.push(
new ethereum.EventParam('dataService', ethereum.Value.fromAddress(DATA_SERVICE)),
)
event.parameters.push(new ethereum.EventParam('payer', ethereum.Value.fromAddress(PAYER)))
event.parameters.push(
new ethereum.EventParam('serviceProvider', ethereum.Value.fromAddress(serviceProvider)),
)
event.parameters.push(
new ethereum.EventParam('agreementId', ethereum.Value.fromFixedBytes(agreementId)),
)
event.parameters.push(
new ethereum.EventParam('endsAt', ethereum.Value.fromUnsignedBigInt(BigInt.fromI32(0))),
)
event.parameters.push(
new ethereum.EventParam(
'maxInitialTokens',
ethereum.Value.fromUnsignedBigInt(BigInt.fromI32(0)),
),
)
event.parameters.push(
new ethereum.EventParam(
'maxOngoingTokensPerSecond',
ethereum.Value.fromUnsignedBigInt(BigInt.fromI32(0)),
),
)
event.parameters.push(
new ethereum.EventParam(
'minSecondsPerCollection',
ethereum.Value.fromUnsignedBigInt(BigInt.fromI32(0)),
),
)
event.parameters.push(
new ethereum.EventParam(
'maxSecondsPerCollection',
ethereum.Value.fromUnsignedBigInt(BigInt.fromI32(0)),
),
)

return event
}

function createOfferCancelledEvent(
caller: Address,
agreementId: Bytes,
Expand Down Expand Up @@ -123,3 +178,20 @@ describe('handleOfferStored', () => {
assert.fieldEquals('Offer', id, 'canceledAt', '0')
})
})

describe('handleAgreementAccepted', () => {
afterEach(() => {
clearStore()
})

test('links the agreement to the indexer registration record by address', () => {
let event = createAgreementAcceptedEvent(SERVICE_PROVIDER, AGREEMENT_ID)
handleAgreementAccepted(event)

let id = AGREEMENT_ID.toHexString()
assert.fieldEquals('IndexingAgreement', id, 'indexer', SERVICE_PROVIDER.toHexString())
// indexerInfo is the typed link consumers follow to read the indexer URL in
// one query; it stores the same address as the Indexer entity id.
assert.fieldEquals('IndexingAgreement', id, 'indexerInfo', SERVICE_PROVIDER.toHexString())
})
})
Loading