Add Oracle Aggregator and CCIP Integration
- Introduced Aggregator.sol for Chainlink-compatible oracle functionality, including round-based updates and access control. - Added OracleWithCCIP.sol to extend Aggregator with CCIP cross-chain messaging capabilities. - Created .gitmodules to include OpenZeppelin contracts as a submodule. - Developed a comprehensive deployment guide in NEXT_STEPS_COMPLETE_GUIDE.md for Phase 2 and smart contract deployment. - Implemented Vite configuration for the orchestration portal, supporting both Vue and React frameworks. - Added server-side logic for the Multi-Cloud Orchestration Portal, including API endpoints for environment management and monitoring. - Created scripts for resource import and usage validation across non-US regions. - Added tests for CCIP error handling and integration to ensure robust functionality. - Included various new files and directories for the orchestration portal and deployment scripts.
This commit is contained in:
defiQUG
290
watcher/src/index.ts
Normal file
290
watcher/src/index.ts
Normal file
@@ -0,0 +1,290 @@
|
||||
/**
|
||||
* CCIP Watcher and Relayer
|
||||
* Production-grade service to watch Chain-138 and relay transactions to Ethereum via CCIP
|
||||
*/
|
||||
|
||||
import { ethers } from "ethers";
|
||||
import { Pool } from "pg";
|
||||
import * as dotenv from "dotenv";
|
||||
import winston from "winston";
|
||||
|
||||
dotenv.config();
|
||||
|
||||
// Configuration
|
||||
const CONFIG = {
|
||||
// Chain-138 RPC
|
||||
CHAIN138_RPC_WS: process.env.CHAIN138_RPC_WS || process.env.CHAIN138_RPC_URL || "",
|
||||
CHAIN138_RPC_HTTP: process.env.CHAIN138_RPC_HTTP || process.env.CHAIN138_RPC_URL || "",
|
||||
|
||||
// Ethereum RPC
|
||||
ETHEREUM_RPC: process.env.ETHEREUM_MAINNET_RPC || process.env.ETHEREUM_RPC || "",
|
||||
|
||||
// Contract addresses
|
||||
CCIP_REPORTER_ADDRESS: process.env.CCIP_REPORTER_CHAIN138_ADDRESS || "",
|
||||
CCIP_LOGGER_ADDRESS: process.env.CCIP_LOGGER_ETH_ADDRESS || "",
|
||||
|
||||
// Chain selectors (from CCIP Directory)
|
||||
CHAIN138_SELECTOR: process.env.CHAIN138_SELECTOR || "0x000000000000008a",
|
||||
ETH_MAINNET_SELECTOR: process.env.ETH_MAINNET_SELECTOR || "0x500147",
|
||||
|
||||
// Relayer configuration
|
||||
RELAYER_PRIVATE_KEY: process.env.RELAYER_PRIVATE_KEY || "",
|
||||
BATCH_SIGNER_PRIVATE_KEY: process.env.BATCH_SIGNER_PRIVATE_KEY || "",
|
||||
|
||||
// Batching configuration
|
||||
BATCH_SIZE: parseInt(process.env.BATCH_SIZE || "10"),
|
||||
BATCH_INTERVAL_MS: parseInt(process.env.BATCH_INTERVAL_MS || "60000"), // 1 minute
|
||||
|
||||
// Confirmation requirements
|
||||
CONFIRMATIONS_REQUIRED: parseInt(process.env.CONFIRMATIONS_REQUIRED || "12"),
|
||||
|
||||
// Database
|
||||
DATABASE_URL: process.env.DATABASE_URL || "postgresql://user:password@localhost:5432/ccip_relayer",
|
||||
|
||||
// Monitoring
|
||||
METRICS_PORT: parseInt(process.env.METRICS_PORT || "9090"),
|
||||
};
|
||||
|
||||
// Logger setup
|
||||
const logger = winston.createLogger({
|
||||
level: process.env.LOG_LEVEL || "info",
|
||||
format: winston.format.combine(
|
||||
winston.format.timestamp(),
|
||||
winston.format.json()
|
||||
),
|
||||
transports: [
|
||||
new winston.transports.Console({
|
||||
format: winston.format.simple(),
|
||||
}),
|
||||
],
|
||||
});
|
||||
|
||||
// Database connection
|
||||
const pool = new Pool({
|
||||
connectionString: CONFIG.DATABASE_URL,
|
||||
});
|
||||
|
||||
// Initialize database schema
|
||||
async function initDatabase() {
|
||||
await pool.query(`
|
||||
CREATE TABLE IF NOT EXISTS outbox (
|
||||
id SERIAL PRIMARY KEY,
|
||||
tx_hash VARCHAR(66) UNIQUE NOT NULL,
|
||||
from_address VARCHAR(42) NOT NULL,
|
||||
to_address VARCHAR(42) NOT NULL,
|
||||
value NUMERIC NOT NULL,
|
||||
block_number BIGINT NOT NULL,
|
||||
observed_at TIMESTAMP DEFAULT NOW(),
|
||||
confirmed_at TIMESTAMP,
|
||||
relayed BOOLEAN DEFAULT FALSE,
|
||||
relayed_txhash VARCHAR(66),
|
||||
batch_id VARCHAR(66),
|
||||
attempts INT DEFAULT 0,
|
||||
last_error TEXT
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_outbox_relayed ON outbox(relayed);
|
||||
CREATE INDEX IF NOT EXISTS idx_outbox_confirmed ON outbox(confirmed_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_outbox_batch ON outbox(batch_id);
|
||||
`);
|
||||
|
||||
logger.info("Database initialized");
|
||||
}
|
||||
|
||||
// Watch Chain-138 for transactions
|
||||
async function watchChain138() {
|
||||
if (!CONFIG.CHAIN138_RPC_WS && !CONFIG.CHAIN138_RPC_HTTP) {
|
||||
throw new Error("Chain-138 RPC URL not configured");
|
||||
}
|
||||
|
||||
const provider = CONFIG.CHAIN138_RPC_WS
|
||||
? new ethers.WebSocketProvider(CONFIG.CHAIN138_RPC_WS)
|
||||
: new ethers.JsonRpcProvider(CONFIG.CHAIN138_RPC_HTTP);
|
||||
|
||||
logger.info("Watching Chain-138 for transactions...");
|
||||
|
||||
// Subscribe to new blocks
|
||||
provider.on("block", async (blockNumber) => {
|
||||
try {
|
||||
const block = await provider.getBlock(blockNumber, true);
|
||||
if (!block || !block.transactions) return;
|
||||
|
||||
// Process transactions in block
|
||||
for (const tx of block.transactions) {
|
||||
if (typeof tx === "string") continue; // Skip if just hash
|
||||
|
||||
// Check if transaction matches criteria (you can add filters here)
|
||||
const txReceipt = await provider.getTransactionReceipt(tx.hash);
|
||||
if (!txReceipt || txReceipt.status !== 1) continue; // Skip failed txs
|
||||
|
||||
// Insert into outbox (idempotent)
|
||||
await pool.query(
|
||||
`INSERT INTO outbox (tx_hash, from_address, to_address, value, block_number, confirmed_at)
|
||||
VALUES ($1, $2, $3, $4, $5, NOW())
|
||||
ON CONFLICT (tx_hash) DO NOTHING`,
|
||||
[
|
||||
tx.hash,
|
||||
tx.from || ethers.ZeroAddress,
|
||||
tx.to || ethers.ZeroAddress,
|
||||
tx.value?.toString() || "0",
|
||||
blockNumber,
|
||||
]
|
||||
);
|
||||
|
||||
logger.debug(`Observed transaction: ${tx.hash}`);
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error("Error processing block:", error);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Batch dispatcher
|
||||
async function dispatchBatches() {
|
||||
if (!CONFIG.CCIP_REPORTER_ADDRESS || !CONFIG.RELAYER_PRIVATE_KEY) {
|
||||
throw new Error("CCIP Reporter address or relayer key not configured");
|
||||
}
|
||||
|
||||
const provider = new ethers.JsonRpcProvider(CONFIG.CHAIN138_RPC_HTTP);
|
||||
const wallet = new ethers.Wallet(CONFIG.RELAYER_PRIVATE_KEY, provider);
|
||||
const reporter = new ethers.Contract(
|
||||
CONFIG.CCIP_REPORTER_ADDRESS,
|
||||
[
|
||||
"function reportBatch(bytes32,bytes32[],address[],address[],uint256[],bytes) payable",
|
||||
"function estimateFee(bytes32[],address[],address[],uint256[],bytes) view returns(uint256)",
|
||||
],
|
||||
wallet
|
||||
);
|
||||
|
||||
setInterval(async () => {
|
||||
try {
|
||||
// Get unrelayed, confirmed transactions
|
||||
const result = await pool.query(
|
||||
`SELECT * FROM outbox
|
||||
WHERE relayed = FALSE
|
||||
AND confirmed_at IS NOT NULL
|
||||
ORDER BY block_number ASC
|
||||
LIMIT $1`,
|
||||
[CONFIG.BATCH_SIZE]
|
||||
);
|
||||
|
||||
if (result.rows.length === 0) return;
|
||||
|
||||
// Prepare batch data
|
||||
const txHashes: string[] = [];
|
||||
const froms: string[] = [];
|
||||
const tos: string[] = [];
|
||||
const values: string[] = [];
|
||||
|
||||
for (const row of result.rows) {
|
||||
txHashes.push(row.tx_hash);
|
||||
froms.push(row.from_address);
|
||||
tos.push(row.to_address);
|
||||
values.push(row.value);
|
||||
}
|
||||
|
||||
// Generate batch ID
|
||||
const batchId = ethers.keccak256(
|
||||
ethers.AbiCoder.defaultAbiCoder().encode(
|
||||
["bytes32[]", "uint256"],
|
||||
[txHashes, Date.now()]
|
||||
)
|
||||
);
|
||||
|
||||
// Sign batch (if signer key configured)
|
||||
let signature = "0x";
|
||||
if (CONFIG.BATCH_SIGNER_PRIVATE_KEY) {
|
||||
const signerWallet = new ethers.Wallet(CONFIG.BATCH_SIGNER_PRIVATE_KEY);
|
||||
const digest = ethers.keccak256(
|
||||
ethers.AbiCoder.defaultAbiCoder().encode(
|
||||
["bytes32", "bytes32", "bytes32", "bytes32", "bytes32"],
|
||||
[
|
||||
batchId,
|
||||
ethers.keccak256(ethers.AbiCoder.defaultAbiCoder().encode(["bytes32[]"], [txHashes])),
|
||||
ethers.keccak256(ethers.AbiCoder.defaultAbiCoder().encode(["address[]"], [froms])),
|
||||
ethers.keccak256(ethers.AbiCoder.defaultAbiCoder().encode(["address[]"], [tos])),
|
||||
ethers.keccak256(ethers.AbiCoder.defaultAbiCoder().encode(["uint256[]"], [values])),
|
||||
]
|
||||
)
|
||||
);
|
||||
const messageHash = ethers.hashMessage(ethers.getBytes(digest));
|
||||
signature = await signerWallet.signMessage(ethers.getBytes(messageHash));
|
||||
}
|
||||
|
||||
// Estimate fee
|
||||
const fee = await reporter.estimateFee(
|
||||
txHashes,
|
||||
froms,
|
||||
tos,
|
||||
values,
|
||||
signature
|
||||
);
|
||||
|
||||
// Check balance
|
||||
const balance = await provider.getBalance(wallet.address);
|
||||
if (balance < fee) {
|
||||
logger.warn(`Insufficient balance. Required: ${ethers.formatEther(fee)} ETH, Have: ${ethers.formatEther(balance)} ETH`);
|
||||
return;
|
||||
}
|
||||
|
||||
// Send batch
|
||||
logger.info(`Relaying batch ${batchId} with ${txHashes.length} transactions`);
|
||||
const tx = await reporter.reportBatch(
|
||||
batchId,
|
||||
txHashes,
|
||||
froms,
|
||||
tos,
|
||||
values,
|
||||
signature,
|
||||
{ value: fee }
|
||||
);
|
||||
|
||||
const receipt = await tx.wait();
|
||||
logger.info(`Batch relayed. TX: ${receipt.hash}`);
|
||||
|
||||
// Update outbox
|
||||
await pool.query(
|
||||
`UPDATE outbox
|
||||
SET relayed = TRUE, relayed_txhash = $1, batch_id = $2
|
||||
WHERE tx_hash = ANY($3::varchar[])`,
|
||||
[receipt.hash, batchId, txHashes]
|
||||
);
|
||||
|
||||
logger.info(`Batch ${batchId} successfully relayed`);
|
||||
} catch (error) {
|
||||
logger.error("Error dispatching batch:", error);
|
||||
// Update attempts
|
||||
// (Implementation for retry logic)
|
||||
}
|
||||
}, CONFIG.BATCH_INTERVAL_MS);
|
||||
}
|
||||
|
||||
// Main function
|
||||
async function main() {
|
||||
logger.info("Starting CCIP Watcher and Relayer...");
|
||||
|
||||
try {
|
||||
await initDatabase();
|
||||
await watchChain138();
|
||||
await dispatchBatches();
|
||||
|
||||
logger.info("CCIP Watcher and Relayer started successfully");
|
||||
} catch (error) {
|
||||
logger.error("Failed to start:", error);
|
||||
process.exit(1);
|
||||
}
|
||||
}
|
||||
|
||||
// Handle graceful shutdown
|
||||
process.on("SIGINT", async () => {
|
||||
logger.info("Shutting down...");
|
||||
await pool.end();
|
||||
process.exit(0);
|
||||
});
|
||||
|
||||
if (require.main === module) {
|
||||
main();
|
||||
}
|
||||
|
||||
export { main, watchChain138, dispatchBatches };
|
||||
|
||||
Reference in New Issue
Block a user