// prover/broadcaster.js // CommonJS — reads signed attestations, broadcasts to CircuitBreaker on Amoy // SafeKrypte signs attestations; oracle private key signs transaction envelopes const { ethers } = require("ethers"); const fs = require("node:fs"); const path = require("node:path"); const notifier = require("./notifier"); // --------------------------------------------------------------------------- // Files // --------------------------------------------------------------------------- const ATTESTATIONS_FILE = path.resolve( __dirname, "..", ".local", "state", "submitter-attestations.json" ); const RECEIPTS_FILE = path.resolve( __dirname, "..", ".local", "state", "broadcast-receipts.json" ); // --------------------------------------------------------------------------- // ABI — exactly matching CircuitBreaker.sol (MVP) const CIRCUIT_BREAKER_ABI = [ "function updateProof(bytes32 assetId, bytes32 deedHash) external", "function tripCircuit(string calldata reason) external", "function circuitOpen() external view returns (bool)", "function latestProof(bytes32 assetId) external view returns (bytes32)", ]; // --------------------------------------------------------------------------- // Idempotency: receipt store // --------------------------------------------------------------------------- function loadReceipts() { if (!fs.existsSync(RECEIPTS_FILE)) return {}; try { return JSON.parse(fs.readFileSync(RECEIPTS_FILE, "utf-8")); } catch { console.warn("[broadcaster] Could not parse receipts — starting fresh."); return {}; } } function saveReceipts(receipts) { const dir = path.dirname(RECEIPTS_FILE); if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true }); fs.writeFileSync(RECEIPTS_FILE, JSON.stringify(receipts, null, 2)); } // --------------------------------------------------------------------------- // Attestation loader — handles { attestations: [...] } or raw array // --------------------------------------------------------------------------- function loadAttestations() { if (!fs.existsSync(ATTESTATIONS_FILE)) return []; const raw = JSON.parse(fs.readFileSync(ATTESTATIONS_FILE, "utf-8")); const attestations = Array.isArray(raw) ? raw : raw.attestations; if (!Array.isArray(attestations)) { throw new Error("Attestations file must be an array or { attestations: [...] }"); } return attestations; } // --------------------------------------------------------------------------- // Validation // --------------------------------------------------------------------------- function validateAttestation(att) { const required = ["digest", "signature", "keyId", "action"]; for (const f of required) { if (!att[f]) throw new Error(`Missing field: ${f}`); } const { kind, assetId, deedHash, reason } = att.action; if (!["updateProof", "tripCircuit"].includes(kind)) { throw new Error(`Unknown action kind: ${kind}`); } if (kind === "updateProof") { if (!ethers.isHexString(assetId, 32)) throw new Error("Invalid assetId"); if (!ethers.isHexString(deedHash, 32)) throw new Error("Invalid deedHash"); } if (kind === "tripCircuit") { if (!reason || typeof reason !== "string") throw new Error("tripCircuit requires reason"); } } // --------------------------------------------------------------------------- // Single attestation broadcast // --------------------------------------------------------------------------- async function broadcastOne(contract, att, receipts, options = {}) { const digest = att.digest; // 1. Receipt guard if (receipts[digest]) { console.log(`[broadcaster] SKIP ${digest.slice(0, 16)}... — already broadcast (tx: ${receipts[digest].txHash})`); return { skipped: true, digest, txHash: receipts[digest].txHash }; } if (options.dryRun) { console.log(`[broadcaster] DRY-RUN ${digest.slice(0, 16)}... — would send ${att.action.kind}`); return { dryRun: true, digest }; } const { kind, assetId, deedHash, reason } = att.action; let tx; if (kind === "updateProof") { // On-chain idempotency: skip if deedHash already set try { const onChain = await contract.latestProof(assetId); if (onChain.toLowerCase() === deedHash.toLowerCase()) { console.log("[broadcaster] SKIP updateProof — deedHash already on-chain"); receipts[digest] = { txHash: "already-on-chain", timestamp: new Date().toISOString() }; saveReceipts(receipts); return { skipped: true, digest, reason: "already-on-chain" }; } } catch (err) { console.warn(`[broadcaster] Could not read latestProof (continuing): ${err.message}`); } console.log(`[broadcaster] SEND updateProof assetId=${assetId.slice(0, 16)}... digest=${digest.slice(0, 16)}...`); tx = await contract.updateProof(assetId, deedHash); // MVP: no signature } if (kind === "tripCircuit") { // Circuit state guard try { const open = await contract.circuitOpen(); if (!open) { console.log("[broadcaster] SKIP tripCircuit — circuit already tripped"); receipts[digest] = { txHash: "circuit-already-tripped", timestamp: new Date().toISOString() }; saveReceipts(receipts); return { skipped: true, digest, reason: "circuit-already-tripped" }; } } catch (err) { console.warn(`[broadcaster] Could not read circuitOpen (continuing): ${err.message}`); } console.log(`[broadcaster] SEND tripCircuit reason="${reason}" digest=${digest.slice(0, 16)}...`); tx = await contract.tripCircuit(reason); // MVP: no signature } console.log(`[broadcaster] WAIT ${tx.hash}...`); const receipt = await tx.wait(); receipts[digest] = { txHash: receipt.hash, blockNumber: receipt.blockNumber, gasUsed: receipt.gasUsed.toString(), timestamp: new Date().toISOString(), attestationDigest: digest, action: att.action, keyId: att.keyId, }; saveReceipts(receipts); console.log(`[broadcaster] CONFIRMED block ${receipt.blockNumber} gas ${receipt.gasUsed} tx ${receipt.hash}`); // Notify success try { notifier.broadcastSuccess({ txHash: receipt.hash, block: receipt.blockNumber, gasUsed: receipt.gasUsed.toString(), action: att.action.kind, assetId: att.action.assetId, }); // Extra alert for circuit trip — high-severity event if (att.action.kind === 'tripCircuit') { notifier.circuitTrip({ reason: att.action.reason, txHash: receipt.hash, assetId: att.action.assetId, }); } } catch (_) {} return { success: true, digest, txHash: receipt.hash, blockNumber: receipt.blockNumber }; } // --------------------------------------------------------------------------- // Main // --------------------------------------------------------------------------- async function broadcast(options = {}) { // Notify pipeline start try { notifier.broadcastStart({}); } catch (_) {} const attestations = loadAttestations(); if (!attestations.length) { console.log("[broadcaster] No attestations — nothing to broadcast."); return []; } console.log(`[broadcaster] Loaded ${attestations.length} attestation(s)`); // Env check const rpcUrl = process.env.POLYGON_AMOY_RPC_URL || process.env.RPC_URL; const oracleKey = process.env.ORACLE_PRIVATE_KEY; const contractAddr = process.env.CIRCUIT_BREAKER_ADDRESS; if (!rpcUrl) throw new Error("Missing RPC_URL / POLYGON_AMOY_RPC_URL"); if (!oracleKey) throw new Error("Missing ORACLE_PRIVATE_KEY"); if (!contractAddr) throw new Error("Missing CIRCUIT_BREAKER_ADDRESS"); if (!ethers.isAddress(contractAddr)) throw new Error("Invalid CIRCUIT_BREAKER_ADDRESS"); const provider = new ethers.JsonRpcProvider(rpcUrl); const signer = new ethers.Wallet(oracleKey, provider); const contract = new ethers.Contract(contractAddr, CIRCUIT_BREAKER_ABI, signer); // For V2, no oracle check needed; threshold signatures handle auth console.log(`[broadcaster] Connected to CircuitBreakerV2 at ${contractAddr}`); const receipts = loadReceipts(); const results = []; for (const att of attestations) { try { validateAttestation(att); const r = await broadcastOne(contract, att, receipts, options); results.push(r); } catch (err) { console.error(`[broadcaster] FAILED ${att.digest?.slice(0, 16) || "unknown"}: ${err.message}`); // Notify failure try { notifier.broadcastFailure({ digest: att.digest, action: att.action?.kind, assetId: att.action?.assetId, error: err.message, }); } catch (_) {} results.push({ error: true, digest: att.digest, message: err.message }); } } const succeeded = results.filter(r => r.success).length; const skipped = results.filter(r => r.skipped).length; const dryRuns = results.filter(r => r.dryRun).length; const failed = results.filter(r => r.error).length; console.log(`[broadcaster] Done — ${succeeded} sent, ${skipped} skipped, ${dryRuns} dry-run, ${failed} failed`); return results; } module.exports = { broadcast, broadcastOne, loadAttestations, loadReceipts, saveReceipts, validateAttestation, }; // --------------------------------------------------------------------------- // CLI // --------------------------------------------------------------------------- if (require.main === module) { const dryRun = process.argv.includes("--dry-run") || process.argv.includes("--dry"); broadcast({ dryRun }) .then(() => { process.exit(0); }) .catch(err => { console.error(`[broadcaster] Fatal: ${err.message}`); process.exit(1); }); }