feat: index bulks

Author: PierreJeanjacquot

schema.graphql

@@ -338,6 +338,24 @@ type Deal @entity {
   workerpoolorder: WorkerpoolOrder # todo: not available if not broadcasted
   requestorder: RequestOrder # todo: not available if not broadcasted
   events: [DealEvent!]! @derivedFrom(field: "deal")
+  bulk: Bulk
+}
+
+type Bulk @entity(immutable: true) {
+  id: ID!
+  hash: String!
+  content: String!
+  slices: [BulkSlice!] @derivedFrom(field: "bulk")
+  deals: [Deal!]! @derivedFrom(field: "bulk")
+}
+
+type BulkSlice @entity(immutable: true) {
+  id: ID!
+  hash: String!
+  content: String!
+  bulk: Bulk!
+  index: BigInt!
+  datasets: [Dataset!]
 }
 
 enum TaskStatus {

src/Modules/Bulk.ts

@@ -0,0 +1,103 @@
+import {
+    BigInt,
+    Bytes,
+    dataSource,
+    DataSourceContext,
+    DataSourceTemplate,
+    json,
+    JSONValueKind,
+} from '@graphprotocol/graph-ts';
+import { Bulk, BulkSlice } from '../../generated/schema';
+import { CONTEXT_BULK, CONTEXT_INDEX, CONTEXT_REQUESTHASH, createBulkSliceID } from '../utils';
+
+function isIntegerString(str: string): boolean {
+    // empty string is not valid
+    if (str.length == 0) {
+        return false;
+    }
+    // 0 prefixed string is not valid
+    if (str[0] === '0' && str.length > 1) {
+        return false;
+    }
+    // non numeric character is not valid
+    for (let i = 0; i < str.length; i++) {
+        if (!['0', '1', '2', '3', '4', '5', '6', '7', '8', '9'].includes(str[i])) {
+            return false;
+        }
+    }
+    return true;
+}
+
+export function handleBulk(content: Bytes): void {
+    const hash = dataSource.stringParam();
+    const context = dataSource.context();
+    const requestorder = context.getString(CONTEXT_REQUESTHASH);
+
+    // multiple deals using same requestorder use the same bulk, we use requestorderHash as bulk ID
+    const bulkid = requestorder;
+
+    let bulk = new Bulk(bulkid);
+    bulk.hash = hash;
+    bulk.content = content.toString();
+
+    const jsonContent = json.try_fromBytes(content);
+    if (jsonContent.isOk) {
+        const contentObject = jsonContent.value.toObject();
+
+        for (let i = 0; i < contentObject.entries.length; i++) {
+            const entry = contentObject.entries[i];
+
+            if (isIntegerString(entry.key)) {
+                const index = BigInt.fromString(entry.key);
+                if (entry.value.kind == JSONValueKind.OBJECT) {
+                    let sliceCid = entry.value.toObject().getEntry('datasets');
+                    if (sliceCid != null && sliceCid.value.kind == JSONValueKind.STRING) {
+                        let sliceContext = new DataSourceContext();
+                        sliceContext.setString(CONTEXT_BULK, bulkid);
+                        sliceContext.setBigInt(CONTEXT_INDEX, index);
+                        DataSourceTemplate.createWithContext(
+                            'BulkSlice',
+                            [sliceCid.value.toString()],
+                            sliceContext,
+                        );
+                    }
+                }
+            }
+        }
+    }
+
+    bulk.save();
+}
+
+export function handleBulkSlice(content: Bytes): void {
+    const hash = dataSource.stringParam();
+    const context = dataSource.context();
+    const bulk = context.getString(CONTEXT_BULK);
+    const index = context.getBigInt(CONTEXT_INDEX);
+
+    let bulkSlice = new BulkSlice(createBulkSliceID(bulk, index));
+    bulkSlice.hash = hash;
+    bulkSlice.bulk = bulk;
+    bulkSlice.index = index;
+    bulkSlice.content = content.toString();
+
+    const jsonContent = json.try_fromBytes(content);
+    if (jsonContent.isOk && jsonContent.value.kind == JSONValueKind.ARRAY) {
+        const datasetArray = jsonContent.value.toArray();
+        for (let i = 0; i < datasetArray.length; i++) {
+            const dataset = datasetArray[i];
+            if (dataset.kind == JSONValueKind.STRING) {
+                let datasetAddress = dataset.toString().toLowerCase();
+                let datasets = bulkSlice.datasets;
+                if (datasets == null) {
+                    datasets = new Array<string>();
+                }
+                // dataset address may be invalid, this is not an issue, the model will prune it
+                datasets.push(datasetAddress);
+                bulkSlice.datasets = datasets;
+            }
+        }
+    }
+
+    bulkSlice.save();
+}

src/Modules/IexecPoco.ts

@@ -1,7 +1,14 @@
 // SPDX-FileCopyrightText: 2020-2025 IEXEC BLOCKCHAIN TECH <[email protected]>
 // SPDX-License-Identifier: Apache-2.0
 
-import { Address, BigInt, dataSource } from '@graphprotocol/graph-ts';
+import {
+    Address,
+    BigInt,
+    dataSource,
+    DataSourceContext,
+    DataSourceTemplate,
+    json,
+} from '@graphprotocol/graph-ts';
 const chainName = dataSource.network();
 
 import {
@@ -21,6 +28,7 @@ import {
 
 import {
     AccurateContribution,
+    Bulk,
     FaultyContribution,
     OrdersMatched,
     SchedulerNotice,
@@ -34,6 +42,7 @@ import {
 } from '../../generated/schema';
 
 import {
+    CONTEXT_REQUESTHASH,
     createContributionID,
     createEventID,
     fetchAccount,
@@ -94,6 +103,26 @@ export function handleOrdersMatched(event: OrdersMatchedEvent): void {
     deal.timestamp = event.block.timestamp;
     deal.save();
 
+    // params including bulk_cid reference a dataset bulk
+    const params = json.try_fromString(viewedDeal.params);
+    if (params.isOk) {
+        const bulkCid = params.value.toObject().getEntry('bulk_cid');
+        if (bulkCid) {
+            // the same bulk is used by any deal using the same requestorder => we use requestorderHash as bulk ID
+            const bulkId = event.params.requestHash.toHex();
+            // create the bulk if not existing yet
+            const indexedBulk = Bulk.load(bulkId);
+            if (!indexedBulk) {
+                let context = new DataSourceContext();
+                context.setString(CONTEXT_REQUESTHASH, bulkId);
+                DataSourceTemplate.createWithContext('Bulk', [bulkCid.value.toString()], context);
+            }
+            // bulk may not be indexed, this is not an issue, the model will prune it
+            deal.bulk = bulkId;
+            deal.save();
+        }
+    }
+
     const dataset = deal.dataset;
 
     let apporder = fetchApporder(event.params.appHash.toHex());