beaconchain: stream transactions and blobs

examples/beaconchain-client/src/main.ts

@@ -1,5 +1,5 @@
+import { BeaconChainStream, Filter } from "@apibara/beaconchain";
 import { createClient } from "@apibara/protocol";
-import { Filter, BeaconChainStream } from "@apibara/beaconchain";
 import { defineCommand, runMain } from "citty";
 import consola from "consola";
 
@@ -33,26 +33,23 @@ const command = defineCommand({
     console.log(response);
 
     const filter = Filter.make({
-      // header: {
-      //   always: true,
-      // }
-      validators: [
+      blobs: [
         {
-          // validatorIndex: 1000,
-          status: "withdrawal_done",
+          includeTransaction: true,
         },
-        // {
-        //   // validatorIndex: 1000,
-        //   status: "withdrawal_possible",
-        // }
       ],
+      // transactions: [{
+      //   to: "0xff00000000000000000000000000000000042069",
+      //   includeBlob: true,
+      // }]
+      // validators: [{}],
     });
 
     const request = BeaconChainStream.Request.make({
       filter: [filter],
       finality: "accepted",
       startingCursor: {
-        orderKey: 5_000_000n,
+        orderKey: 5_200_000n,
       },
     });
 
@@ -61,8 +58,22 @@ const command = defineCommand({
         case "data": {
           consola.info("Data", message.data.endCursor?.orderKey);
 
+          const maxBlobSize = 131_072; // bytes
           for (const block of message.data.data) {
-            console.log(block);
+            const transactions = block.transactions ?? [];
+            for (const blob of block.blobs) {
+              if (blob.blob === undefined) continue;
+              let i = blob.blob.length - 1;
+              while (i >= 0 && blob.blob[i] === 0) i--;
+              const utilization = (i / maxBlobSize) * 100;
+              const tx = transactions.find(
+                (tx) => tx.transactionIndex === blob.transactionIndex,
+              );
+              consola.info(
+                `${blob.transactionHash} - ${utilization.toFixed(2)}%`,
+              );
+              consola.info(`    ${tx?.to}`);
+            }
           }
 
           break;

examples/evm-client/src/main.ts

@@ -2,7 +2,7 @@ import { EvmStream, Filter } from "@apibara/evm";
 import { createClient } from "@apibara/protocol";
 import { defineCommand, runMain } from "citty";
 import consola from "consola";
-import { decodeEventLog, encodeEventTopics, parseAbi } from "viem";
+import { encodeEventTopics, parseAbi } from "viem";
 
 const abi = parseAbi([
   "event Transfer(address indexed from, address indexed to, uint256 value)",

packages/beaconchain/package.json

@@ -31,7 +31,6 @@
     "@types/long": "^5.0.0",
     "@types/node": "^20.12.12",
     "unbuild": "^2.0.0",
-    "viem": "^2.13.8",
     "vitest": "^1.6.0"
   },
   "dependencies": {
@@ -41,6 +40,7 @@
     "effect": "^3.2.6",
     "long": "^5.2.1",
     "nice-grpc-common": "^2.0.2",
-    "protobufjs": "^7.1.2"
+    "protobufjs": "^7.1.2",
+    "viem": "^2.13.8"
   }
 }

packages/beaconchain/proto/data.proto

@@ -34,11 +34,11 @@ message BlockHeader {
 
 message Transaction {
   // Transaction hash.
-  B256 hash = 1;
+  B256 transaction_hash = 1;
   // Nonce.
   uint64 nonce = 2;
   // Transaction index in the block.
-  uint64 transaction_index = 3;
+  uint32 transaction_index = 3;
   // Sender.
   Address from = 4;
   // Recipient.
@@ -48,7 +48,7 @@ message Transaction {
   // Gas price.
   U128 gas_price = 7;
   // Gas amount.
-  U256 gas = 8;
+  U128 gas_limit = 8;
   // Max base fee per gas the sender is willing to pay.
   U128 max_fee_per_gas = 9;
   // Miner's tip.
@@ -58,7 +58,7 @@ message Transaction {
   // The signature's r,s,v,yParity values.
   Signature signature = 12;
   // Chain ID.
-  uint64 chain_id = 13;
+  optional uint64 chain_id = 13;
   // EIP-2930 access list.
   repeated AccessListItem access_list = 14;
   // EIP-2718 transaction type.
@@ -96,6 +96,10 @@ message Blob {
   repeated B256 kzg_commitment_inclusion_proof = 5;
   // Blob hash.
   B256 blob_hash = 6;
+  // Index of the transaction that posted the blob.
+  uint32 transaction_index = 7;
+  // Hash of the transaction that posted the blob.
+  B256 transaction_hash = 8;
 }
 
 message ExecutionPayload {

packages/beaconchain/proto/filter.proto

@@ -37,4 +37,7 @@ message ValidatorFilter {
   ValidatorStatus status = 2;
 }
 
-message BlobFilter {}
+message BlobFilter {
+  // Include the transaction that posted the blob.
+  optional bool include_transaction = 1;
+}

packages/beaconchain/src/block.ts

@@ -1,17 +1,18 @@
+import { AccessListItem, Signature } from "@apibara/evm";
+import { BytesFromUint8Array } from "@apibara/protocol";
 import { Schema } from "@effect/schema";
-import { Address, B256 } from "@apibara/evm";
 
-import { ValidatorStatus } from "./common";
+import { Address, B256, B384, U128, U256, ValidatorStatus } from "./common";
 import * as proto from "./proto";
 
 export const ExecutionPayload = Schema.Struct({
   parentHash: Schema.optional(B256),
   feeRecipient: Schema.optional(Address),
   stateRoot: Schema.optional(B256),
   receiptsRoot: Schema.optional(B256),
-  // TODO: logs bloom
+  logsBloom: BytesFromUint8Array,
   prevRandao: Schema.optional(B256),
-  blockNumber: Schema.BigIntFromSelf,
+  blockNumber: Schema.optional(Schema.BigIntFromSelf),
   timestamp: Schema.optional(Schema.DateFromSelf),
 });
 
@@ -22,13 +23,13 @@ export const BlockHeader = Schema.Struct({
   proposerIndex: Schema.Number,
   parentRoot: Schema.optional(B256),
   stateRoot: Schema.optional(B256),
-  // TODO: randao reveal
-  depositCount: Schema.BigIntFromSelf,
+  randaoReveal: BytesFromUint8Array,
+  depositCount: Schema.optional(Schema.BigIntFromSelf),
   depositRoot: Schema.optional(B256),
   blockHash: Schema.optional(B256),
   graffiti: Schema.optional(B256),
   executionPayload: Schema.optional(ExecutionPayload),
-  // TODO: blob kzg commitments
+  blobKzgCommitments: Schema.optional(Schema.Array(B384)),
 });
 
 export type BlockHeader = typeof BlockHeader.Type;
@@ -37,7 +38,7 @@ export const Validator = Schema.Struct({
   validatorIndex: Schema.optional(Schema.Number),
   balance: Schema.optional(Schema.BigIntFromSelf),
   status: Schema.optional(ValidatorStatus),
-  // TODO: pubkey
+  pubkey: Schema.optional(B384),
   withdrawalCredentials: Schema.optional(B256),
   effectiveBalance: Schema.optional(Schema.BigIntFromSelf),
   slashed: Schema.optional(Schema.Boolean),
@@ -47,9 +48,44 @@ export const Validator = Schema.Struct({
   withdrawableEpoch: Schema.optional(Schema.BigIntFromSelf),
 });
 
+export const Blob = Schema.Struct({
+  blobIndex: Schema.optional(Schema.Number),
+  blob: Schema.optional(Schema.Uint8ArrayFromSelf),
+  kzgCommitment: Schema.optional(B384),
+  kzgProof: Schema.optional(B384),
+  kzgCommitmentInclusionProof: Schema.optional(Schema.Array(B256)),
+  blobHash: Schema.optional(B256),
+  transactionIndex: Schema.optional(Schema.Number),
+  transactionHash: Schema.optional(B256),
+});
+
+export const Transaction = Schema.Struct({
+  transactionHash: Schema.optional(B256),
+  nonce: Schema.optional(Schema.BigIntFromSelf),
+  transactionIndex: Schema.optional(Schema.Number),
+  from: Schema.optional(Address),
+  to: Schema.optional(Address),
+  value: Schema.optional(U256),
+  gasPrice: Schema.optional(U128),
+  gasLimit: Schema.optional(U128),
+  maxFeePerGas: Schema.optional(U128),
+  maxPriorityFeePerGas: Schema.optional(U128),
+  input: Schema.optional(Schema.Uint8ArrayFromSelf),
+  signature: Schema.optional(Signature),
+  chainId: Schema.optional(Schema.BigIntFromSelf),
+  accessList: Schema.optional(Schema.Array(AccessListItem)),
+  transactionType: Schema.optional(Schema.BigIntFromSelf),
+  maxFeePerBlobGas: Schema.optional(U128),
+  blobVersionedHashes: Schema.optional(Schema.Array(B256)),
+});
+
+export type Transaction = typeof Transaction.Type;
+
 export const Block = Schema.Struct({
   header: Schema.optional(BlockHeader),
-  validators: Schema.optional(Schema.Array(Validator)),
+  validators: Schema.Array(Validator),
+  blobs: Schema.Array(Blob),
+  transactions: Schema.Array(Transaction),
 });
 
 export type Block = typeof Block.Type;

packages/beaconchain/src/common.test.ts

@@ -0,0 +1,21 @@
+import { describe, expect, it } from "vitest";
+
+import { Schema } from "@effect/schema";
+import { pad } from "viem";
+
+import { B384 } from "./common";
+
+describe("B384", () => {
+  const encode = Schema.encodeSync(B384);
+  const decode = Schema.decodeSync(B384);
+
+  it("should convert from and to proto", () => {
+    const value =
+      "0x9df92d765b5aa041fd4bbe8d5878eb89290efa78e444c1a603eecfae2ea05fa4";
+
+    const message = encode(value);
+
+    const back = decode(message);
+    expect(back).toEqual(pad(value, { size: 48 }));
+  });
+});

packages/beaconchain/src/common.ts

@@ -1,4 +1,5 @@
 import { Schema } from "@effect/schema";
+import { hexToBytes, pad } from "viem";
 
 import * as proto from "./proto";
 
@@ -41,21 +42,54 @@ export const ValidatorStatus = Schema.transform(
     },
     encode(value) {
       const enumMap = {
-        ["pending_initialized"]:
-          proto.common.ValidatorStatus.PENDING_INITIALIZED,
-        ["pending_queued"]: proto.common.ValidatorStatus.PENDING_QUEUED,
-        ["active_ongoing"]: proto.common.ValidatorStatus.ACTIVE_ONGOING,
-        ["active_exiting"]: proto.common.ValidatorStatus.ACTIVE_EXITING,
-        ["active_slashed"]: proto.common.ValidatorStatus.ACTIVE_SLASHED,
-        ["exited_unslashed"]: proto.common.ValidatorStatus.EXITED_UNSLASHED,
-        ["exited_slashed"]: proto.common.ValidatorStatus.EXITED_SLASHED,
-        ["withdrawal_possible"]:
-          proto.common.ValidatorStatus.WITHDRAWAL_POSSIBLE,
-        ["withdrawal_done"]: proto.common.ValidatorStatus.WITHDRAWAL_DONE,
-        ["unknown"]: proto.common.ValidatorStatus.UNKNOWN,
+        pending_initialized: proto.common.ValidatorStatus.PENDING_INITIALIZED,
+        pending_queued: proto.common.ValidatorStatus.PENDING_QUEUED,
+        active_ongoing: proto.common.ValidatorStatus.ACTIVE_ONGOING,
+        active_exiting: proto.common.ValidatorStatus.ACTIVE_EXITING,
+        active_slashed: proto.common.ValidatorStatus.ACTIVE_SLASHED,
+        exited_unslashed: proto.common.ValidatorStatus.EXITED_UNSLASHED,
+        exited_slashed: proto.common.ValidatorStatus.EXITED_SLASHED,
+        withdrawal_possible: proto.common.ValidatorStatus.WITHDRAWAL_POSSIBLE,
+        withdrawal_done: proto.common.ValidatorStatus.WITHDRAWAL_DONE,
+        unknown: proto.common.ValidatorStatus.UNKNOWN,
       } as const;
 
       return enumMap[value] ?? proto.common.ValidatorStatus.UNKNOWN;
     },
   },
 );
+
+const _B384 = Schema.TemplateLiteral(Schema.Literal("0x"), Schema.String);
+const B384Proto = Schema.Struct({
+  x0: Schema.BigIntFromSelf,
+  x1: Schema.BigIntFromSelf,
+  x2: Schema.BigIntFromSelf,
+  x3: Schema.BigIntFromSelf,
+  x4: Schema.BigIntFromSelf,
+  x5: Schema.BigIntFromSelf,
+});
+
+export const B384 = Schema.transform(B384Proto, _B384, {
+  decode(value) {
+    const x0 = value.x0.toString(16).padStart(16, "0");
+    const x1 = value.x1.toString(16).padStart(16, "0");
+    const x2 = value.x2.toString(16).padStart(16, "0");
+    const x3 = value.x3.toString(16).padStart(16, "0");
+    const x4 = value.x4.toString(16).padStart(16, "0");
+    const x5 = value.x5.toString(16).padStart(16, "0");
+    return `0x${x0}${x1}${x2}${x3}${x4}${x5}` as `0x${string}`;
+  },
+  encode(value) {
+    const bytes = hexToBytes(pad(value, { size: 48, dir: "left" }));
+    const dv = new DataView(bytes.buffer);
+    const x0 = dv.getBigUint64(0);
+    const x1 = dv.getBigUint64(8);
+    const x2 = dv.getBigUint64(16);
+    const x3 = dv.getBigUint64(24);
+    const x4 = dv.getBigUint64(32);
+    const x5 = dv.getBigUint64(40);
+    return { x0, x1, x2, x3, x4, x5 };
+  },
+});
+
+export type B384 = typeof B384.Type;

packages/beaconchain/src/filter.ts

@@ -1,7 +1,7 @@
 import { Schema } from "@effect/schema";
 
+import { Address, ValidatorStatus } from "./common";
 import * as proto from "./proto";
-import { ValidatorStatus } from "./common";
 
 /** Header options.
  *
@@ -13,16 +13,52 @@ export const HeaderFilter = Schema.Struct({
 
 export type HeaderFilter = typeof HeaderFilter.Type;
 
+/** Filter transactions.
+ *
+ * @prop from Filter transactions by the sender address.
+ * @prop to Filter transactions by the target address.
+ * @prop includeBlob Include any blob posted by the transaction..
+ */
+export const TransactionFilter = Schema.Struct({
+  from: Schema.optional(Address),
+  to: Schema.optional(Address),
+  includeBlob: Schema.optional(Schema.Boolean),
+});
+
+export type TransactionFilter = typeof TransactionFilter.Type;
+
+/** Filter validators.
+ *
+ * @prop validatorIndex Filter validators by their index.
+ * @prop status Filter validators by their status.
+ */
 export const ValidatorFilter = Schema.Struct({
   validatorIndex: Schema.optional(Schema.Number),
   status: Schema.optional(ValidatorStatus),
 });
 
 export type ValidatorFilter = typeof ValidatorFilter.Type;
 
+/** Filter blobs.
+ *
+ * @prop includeTransaction Include the transaction that posted the blob.
+ */
+export const BlobFilter = Schema.Struct({
+  includeTransaction: Schema.optional(Schema.Boolean),
+});
+
+export type BlobFilter = typeof BlobFilter.Type;
+
+/** Filter block data.
+ *
+ * @prop header Change how block headers are returned.
+ * @prop validators Filter validators.
+ */
 export const Filter = Schema.Struct({
   header: Schema.optional(HeaderFilter),
+  transactions: Schema.optional(Schema.Array(TransactionFilter)),
   validators: Schema.optional(Schema.Array(ValidatorFilter)),
+  blobs: Schema.optional(Schema.Array(BlobFilter)),
 });
 
 export type Filter = typeof Filter.Type;