feat(subgraph): restore immutable transition log entities

PR #5 replaces the immutable IndexingAgreementAccepted/Canceled/Updated
event-log entities with a mutable aggregated IndexingAgreement. Current-state
queries improve, but event-sourcing consumers break: dipper's chain_listener
polls indexingAgreementAccepteds/indexingAgreementCanceleds since-block to
replay state transitions into its local DB. Restore the logs alongside the
aggregated entity so snapshot and streaming consumers share one subgraph.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: MoonBoi9001

schema.graphql

@@ -72,6 +72,46 @@ type IndexerDeploymentLatest @entity(immutable: false) {
   blockTimestamp: BigInt!
 }
 
+# Immutable transition logs. The aggregated IndexingAgreement entity above
+# serves current-state queries; these append-only logs serve event-sourcing
+# consumers (e.g. dipper's chain_listener polls them since-block to replay
+# state transitions into its local DB). Preserving both views lets snapshot
+# and streaming consumers share one subgraph.
+type IndexingAgreementAccepted @entity(immutable: true) {
+  id: ID!
+  indexer: Bytes!
+  payer: Bytes!
+  agreementId: Bytes!
+  allocationId: Bytes!
+  blockNumber: BigInt!
+  blockTimestamp: BigInt!
+  transactionHash: Bytes!
+}
+
+type IndexingAgreementCanceled @entity(immutable: true) {
+  id: ID!
+  indexer: Bytes!
+  payer: Bytes!
+  agreementId: Bytes!
+  "Address that initiated the cancel (from SubgraphService canceledOnBehalfOf param)"
+  canceledBy: Bytes!
+  blockNumber: BigInt!
+  blockTimestamp: BigInt!
+  transactionHash: Bytes!
+}
+
+type IndexingAgreementUpdated @entity(immutable: true) {
+  id: ID!
+  indexer: Bytes!
+  payer: Bytes!
+  agreementId: Bytes!
+  allocationId: Bytes!
+  version: Int!
+  blockNumber: BigInt!
+  blockTimestamp: BigInt!
+  transactionHash: Bytes!
+}
+
 # Immutable log of every OfferStored event emitted by RecurringCollector.
 # OFFER_TYPE_NEW = 0, OFFER_TYPE_UPDATE = 1.
 type OfferStored @entity(immutable: true) {

src/subgraphService.ts

@@ -1,10 +1,17 @@
 import { ethereum } from '@graphprotocol/graph-ts'
 import {
   IndexingAgreementAccepted as AcceptedEvent,
+  IndexingAgreementCanceled as CanceledEvent,
   IndexingAgreementUpdated as UpdatedEvent,
   IndexingFeesCollectedV1 as FeesCollectedEvent,
 } from '../generated/SubgraphService/SubgraphService'
-import { IndexerDeploymentLatest, IndexingFeeCollection } from '../generated/schema'
+import {
+  IndexerDeploymentLatest,
+  IndexingFeeCollection,
+  IndexingAgreementAccepted,
+  IndexingAgreementCanceled,
+  IndexingAgreementUpdated,
+} from '../generated/schema'
 import { createOrLoadIndexingAgreement } from './helpers'
 
 export function handleIndexingAgreementAccepted(event: AcceptedEvent): void {
@@ -20,6 +27,34 @@ export function handleIndexingAgreementAccepted(event: AcceptedEvent): void {
   }
 
   agreement.save()
+
+  let logId = event.transaction.hash.concatI32(event.logIndex.toI32()).toHexString()
+  let log = new IndexingAgreementAccepted(logId)
+  log.indexer = event.params.indexer
+  log.payer = event.params.payer
+  log.agreementId = event.params.agreementId
+  log.allocationId = event.params.allocationId
+  log.blockNumber = event.block.number
+  log.blockTimestamp = event.block.timestamp
+  log.transactionHash = event.transaction.hash
+  log.save()
+}
+
+export function handleIndexingAgreementCanceled(event: CanceledEvent): void {
+  // State and canceledAt are set by RecurringCollector.handleAgreementCanceled,
+  // which emits the canonical cancel event with the canceledBy enum. This
+  // handler just writes the immutable transition log for event-sourcing
+  // consumers (dipper's chain_listener).
+  let logId = event.transaction.hash.concatI32(event.logIndex.toI32()).toHexString()
+  let log = new IndexingAgreementCanceled(logId)
+  log.indexer = event.params.indexer
+  log.payer = event.params.payer
+  log.agreementId = event.params.agreementId
+  log.canceledBy = event.params.canceledOnBehalfOf
+  log.blockNumber = event.block.number
+  log.blockTimestamp = event.block.timestamp
+  log.transactionHash = event.transaction.hash
+  log.save()
 }
 
 export function handleIndexingAgreementUpdated(event: UpdatedEvent): void {
@@ -34,6 +69,18 @@ export function handleIndexingAgreementUpdated(event: UpdatedEvent): void {
   }
 
   agreement.save()
+
+  let logId = event.transaction.hash.concatI32(event.logIndex.toI32()).toHexString()
+  let log = new IndexingAgreementUpdated(logId)
+  log.indexer = event.params.indexer
+  log.payer = event.params.payer
+  log.agreementId = event.params.agreementId
+  log.allocationId = event.params.allocationId
+  log.version = event.params.version
+  log.blockNumber = event.block.number
+  log.blockTimestamp = event.block.timestamp
+  log.transactionHash = event.transaction.hash
+  log.save()
 }
 
 export function handleIndexingFeesCollectedV1(event: FeesCollectedEvent): void {

subgraph.template.yaml

@@ -19,12 +19,17 @@ dataSources:
         - IndexingAgreement
         - IndexingFeeCollection
         - IndexerDeploymentLatest
+        - IndexingAgreementAccepted
+        - IndexingAgreementCanceled
+        - IndexingAgreementUpdated
       abis:
         - name: SubgraphService
           file: ./abis/SubgraphService.json
       eventHandlers:
         - event: IndexingAgreementAccepted(indexed address,indexed address,indexed bytes16,address,bytes32,uint8,bytes)
           handler: handleIndexingAgreementAccepted
+        - event: IndexingAgreementCanceled(indexed address,indexed address,indexed bytes16,address)
+          handler: handleIndexingAgreementCanceled
         - event: IndexingAgreementUpdated(indexed address,indexed address,indexed bytes16,address,uint8,bytes)
           handler: handleIndexingAgreementUpdated
         - event: IndexingFeesCollectedV1(indexed address,indexed address,indexed bytes16,address,bytes32,uint256,uint256,uint256,bytes32,uint256,bytes)

tests/subgraphService.test.ts

@@ -2,11 +2,13 @@ import { assert, describe, test, clearStore, afterEach } from 'matchstick-as'
 import { Address, Bytes, BigInt, ethereum } from '@graphprotocol/graph-ts'
 import {
   handleIndexingAgreementAccepted,
+  handleIndexingAgreementCanceled,
   handleIndexingAgreementUpdated,
   handleIndexingFeesCollectedV1,
 } from '../src/subgraphService'
 import {
   IndexingAgreementAccepted as AcceptedEvent,
+  IndexingAgreementCanceled as CanceledEvent,
   IndexingAgreementUpdated as UpdatedEvent,
   IndexingFeesCollectedV1 as FeesCollectedEvent,
 } from '../generated/SubgraphService/SubgraphService'
@@ -46,6 +48,30 @@ function createAcceptedEvent(
   return event
 }
 
+function createCanceledEvent(
+  indexer: Address,
+  payer: Address,
+  agreementId: Bytes,
+  canceledOnBehalfOf: Address,
+): CanceledEvent {
+  let event = changetype<CanceledEvent>(newMockEvent())
+
+  event.parameters = new Array()
+  event.parameters.push(new ethereum.EventParam('indexer', ethereum.Value.fromAddress(indexer)))
+  event.parameters.push(new ethereum.EventParam('payer', ethereum.Value.fromAddress(payer)))
+  event.parameters.push(
+    new ethereum.EventParam('agreementId', ethereum.Value.fromFixedBytes(agreementId)),
+  )
+  event.parameters.push(
+    new ethereum.EventParam(
+      'canceledOnBehalfOf',
+      ethereum.Value.fromAddress(canceledOnBehalfOf),
+    ),
+  )
+
+  return event
+}
+
 function createUpdatedEvent(
   indexer: Address,
   payer: Address,
@@ -177,6 +203,29 @@ describe('handleIndexingAgreementAccepted', () => {
     )
     // State remains NotAccepted until RC handler fires
     assert.fieldEquals('IndexingAgreement', agreementId.toHexString(), 'state', 'NotAccepted')
+
+    // Immutable transition log emitted for event-sourcing consumers
+    assert.entityCount('IndexingAgreementAccepted', 1)
+  })
+})
+
+describe('handleIndexingAgreementCanceled', () => {
+  afterEach(() => {
+    clearStore()
+  })
+
+  test('emits immutable IndexingAgreementCanceled log with canceledBy address', () => {
+    let indexer = Address.fromString('0x0000000000000000000000000000000000000001')
+    let payer = Address.fromString('0x0000000000000000000000000000000000000002')
+    let agreementId = Bytes.fromHexString('0x0102030405060708090a0b0c0d0e0f10')
+    let canceledOnBehalfOf = Address.fromString('0x0000000000000000000000000000000000000004')
+
+    let event = createCanceledEvent(indexer, payer, agreementId, canceledOnBehalfOf)
+    handleIndexingAgreementCanceled(event)
+
+    assert.entityCount('IndexingAgreementCanceled', 1)
+    // Aggregated entity state is owned by the RC handler — SS handler must not create one
+    assert.entityCount('IndexingAgreement', 0)
   })
 })
 
@@ -232,6 +281,10 @@ describe('handleIndexingAgreementUpdated', () => {
       'tokensPerEntityPerSecond',
       '100',
     )
+
+    // One log per event: the accept and the update each emit their own immutable log
+    assert.entityCount('IndexingAgreementAccepted', 1)
+    assert.entityCount('IndexingAgreementUpdated', 1)
   })
 })