Implement OP Stack batch parsing for L2 block range extraction

- Add full batch data parsing with zlib/brotli decompression
- Parse span batch and singular batch formats per OP Stack derivation spec
- Extract L2 block count and calculate block range from timestamps
- Add finalizePendingOpBatches job to confirm batches on safe block
- Add l2BlockTime field to OpChainConfig model
- Fix finalizePendingOrbitBatches to include orbitChildConfigs association
- Fix hex-to-integer conversion for l1BlockNumber in batch creation

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Author: Antoine de Chevigné

run/jobs/blockSync.js

@@ -22,6 +22,7 @@ module.exports = async job => {
         return 'Missing parameter';
 
     const workspace = await Workspace.findOne({
+        subQuery: false,
         where: {
             name: data.workspace,
             '$user.firebaseUserId$': data.userId
@@ -226,11 +227,19 @@ module.exports = async job => {
                 isBatchTransaction(tx, opConfig.batchInboxAddress)
             );
 
+            // Get L1 block timestamp for batch parsing
+            const l1Timestamp = typeof processedBlock.timestamp === 'string' && processedBlock.timestamp.startsWith('0x')
+                ? parseInt(processedBlock.timestamp, 16)
+                : Number(processedBlock.timestamp);
+
             for (const tx of batchTxs) {
                 try {
                     const batchInfo = await getBatchInfo(tx, {
                         batchInboxAddress: opConfig.batchInboxAddress,
-                        beaconUrl: workspace.beaconUrl
+                        beaconUrl: workspace.beaconUrl,
+                        workspaceId: opConfig.workspaceId,
+                        l1Timestamp: l1Timestamp,
+                        l2BlockTime: opConfig.l2BlockTime || 2
                     });
 
                     if (batchInfo) {
@@ -247,16 +256,24 @@ module.exports = async job => {
                             });
                             const nextBatchIndex = lastBatch ? lastBatch.batchIndex + 1 : 0;
 
+                            // Convert hex values to integers
+                            const l1BlockNumber = typeof batchInfo.l1BlockNumber === 'string' && batchInfo.l1BlockNumber.startsWith('0x')
+                                ? parseInt(batchInfo.l1BlockNumber, 16)
+                                : Number(batchInfo.l1BlockNumber);
+                            const l1TransactionIndex = typeof batchInfo.l1TransactionIndex === 'string' && batchInfo.l1TransactionIndex.startsWith('0x')
+                                ? parseInt(batchInfo.l1TransactionIndex, 16)
+                                : Number(batchInfo.l1TransactionIndex);
+
                             await OpBatch.create({
                                 workspaceId: opConfig.workspaceId,
                                 batchIndex: nextBatchIndex,
-                                l1BlockNumber: batchInfo.l1BlockNumber,
+                                l1BlockNumber: l1BlockNumber,
                                 l1TransactionHash: batchInfo.l1TransactionHash,
                                 l1TransactionId: l1Transaction ? l1Transaction.id : null,
-                                l1TransactionIndex: batchInfo.l1TransactionIndex,
-                                epochNumber: batchInfo.l1BlockNumber, // Epoch is typically the L1 block
+                                l1TransactionIndex: l1TransactionIndex,
+                                epochNumber: l1BlockNumber, // Epoch is typically the L1 block
                                 timestamp: tx.timestamp ? new Date(tx.timestamp * 1000) : new Date(),
-                                txCount: batchInfo.estimatedBlockCount || null,
+                                txCount: batchInfo.blockCount || null,
                                 l2BlockStart: batchInfo.l2BlockStart,
                                 l2BlockEnd: batchInfo.l2BlockEnd,
                                 blobHash: batchInfo.blobHash,

run/jobs/finalizePendingOpBatches.js

@@ -0,0 +1,59 @@
+/**
+ * @fileoverview OP Stack batch finalization job.
+ * Confirms pending batches once parent chain reaches safe block.
+ * @module jobs/finalizePendingOpBatches
+ */
+
+const logger = require('../lib/logger');
+const { OpBatch, OpChainConfig, Workspace } = require('../models');
+const { Op } = require('sequelize');
+
+module.exports = async () => {
+    // Find all OP child configs and their parent workspaces
+    const opConfigs = await OpChainConfig.findAll({
+        include: [{
+            model: Workspace,
+            as: 'parentWorkspace',
+            required: true
+        }]
+    });
+
+    let allConfirmedBatches = [];
+
+    for (const opConfig of opConfigs) {
+        try {
+            const parentWorkspace = opConfig.parentWorkspace;
+            if (!parentWorkspace || !parentWorkspace.rpcServer) continue;
+
+            const client = parentWorkspace.getViemPublicClient();
+            const safeBlock = await client.getBlock({ blockTag: 'safe' });
+
+            logger.info(`Validating OP batches for workspace ${opConfig.workspaceId} against safe block ${safeBlock.number}`);
+
+            const pendingBatches = await OpBatch.findAll({
+                where: {
+                    workspaceId: opConfig.workspaceId,
+                    status: 'pending',
+                    l1BlockNumber: {
+                        [Op.lte]: Number(safeBlock.number)
+                    }
+                }
+            });
+
+            for (const batch of pendingBatches) {
+                await batch.update({ status: 'confirmed' });
+                logger.info(`Confirmed OP batch ${batch.batchIndex} for workspace ${opConfig.workspaceId}`);
+            }
+
+            allConfirmedBatches = allConfirmedBatches.concat(pendingBatches);
+        } catch (error) {
+            logger.error(`Error finalizing OP batches for config ${opConfig.id}: ${error.message}`, {
+                location: 'jobs.finalizePendingOpBatches',
+                error,
+                configId: opConfig.id
+            });
+        }
+    }
+
+    return `Confirmed ${allConfirmedBatches.length} OP batches`;
+};

run/jobs/finalizePendingOrbitBatches.js

@@ -10,7 +10,10 @@ const { Op } = require('sequelize');
 
 module.exports = async () => {
 
-    const workspaces = await Workspace.findAll({ where: { isTopL1Parent: true } });
+    const workspaces = await Workspace.findAll({
+        where: { isTopL1Parent: true },
+        include: ['orbitChildConfigs']
+    });
 
     let allPendingBatches = [];
     for (const workspace of workspaces) {

run/jobs/index.js

@@ -20,6 +20,7 @@ module.exports = {
     backfillOrbitMessageDeliveredLogs: require('./backfillOrbitMessageDeliveredLogs'),
     storeOrbitDeposit: require('./storeOrbitDeposit'),
     finalizePendingOpOutputs: require('./finalizePendingOpOutputs'),
+    finalizePendingOpBatches: require('./finalizePendingOpBatches'),
     linkOpDepositsToL2Txs: require('./linkOpDepositsToL2Txs'),
 
     // Medium Priority