diff --git a/.env.example b/.env.example index 467f47f..dd0b25f 100644 --- a/.env.example +++ b/.env.example @@ -14,6 +14,7 @@ CHIRPSTACK_HOST=localhost CONTRACT_LABEL=M3ters MAINNET_RPC=https://ethereum-sepolia-rpc.publicnode.com #alt https://eth-sepolia.g.alchemy.com/v2/ ETHEREUM_PRIVATE_KEY="..." +ROLLUP_CONTRACT_ADDRESS="0x9a9ED9b9A9BEC10d1B5E41b1873E78d19501D5Fa" # Streamr Configuration STREAMR_STREAM_ID="0x567853282663b601bfdb9203819b1fbb3fe18926/m3tering/test" diff --git a/Dockerfile b/Dockerfile index 6a78d91..9a99712 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,8 @@ FROM node:20-alpine +# Use Node.js headers bundled with the base image +ENV npm_config_nodedir=/usr/local + # Create working directory WORKDIR /opt/app diff --git a/README.md b/README.md index d07dd31..1782790 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,7 @@ A modular, extensible service console for providers on the M3tering protocol. Fe CONTRACT_LABEL=M3ters MAINNET_RPC=https://ethereum-sepolia-rpc.publicnode.com #alt https://eth-sepolia.g.alchemy.com/v2/ ETHEREUM_PRIVATE_KEY="..." + ROLLUP_CONTRACT_ADDRESS="0x9a9ED9b9A9BEC10d1B5E41b1873E78d19501D5Fa" # Streamr Configuration STREAMR_STREAM_ID="0x567853282663b601bfdb9203819b1fbb3fe18926/m3tering/test" diff --git a/compose.dev.yaml b/compose.dev.yaml new file mode 100644 index 0000000..f526379 --- /dev/null +++ b/compose.dev.yaml @@ -0,0 +1,18 @@ +services: + console: + build: + context: . + dockerfile: Dockerfile + image: m3tering-console:dev + network_mode: host + restart: unless-stopped + env_file: + - .env + volumes: + - ./.data-dev:/opt/app/.data + - console_dev_src_modules:/opt/app/src/lib + - console_dev_dist_modules:/opt/app/dist/lib + +volumes: + console_dev_src_modules: + console_dev_dist_modules: diff --git a/docker-compose.yaml b/compose.yaml similarity index 100% rename from docker-compose.yaml rename to compose.yaml diff --git a/package-lock.json b/package-lock.json index c156c50..bfa77aa 100644 --- a/package-lock.json +++ b/package-lock.json @@ -2170,21 +2170,6 @@ "xstream": "^11.14.0" } }, - "node_modules/@cosmjs/socket/node_modules/utf-8-validate": { - "version": "5.0.10", - "resolved": "https://registry.npmjs.org/utf-8-validate/-/utf-8-validate-5.0.10.tgz", - "integrity": "sha512-Z6czzLq4u8fPOyx7TU6X3dvUZVvoJmxSQ+IcrlmagKhilxlhZgxPK6C5Jqbkw1IDUmFTM+cz9QDnnLTwDz/2gQ==", - "hasInstallScript": true, - "license": "MIT", - "optional": true, - "peer": true, - "dependencies": { - "node-gyp-build": "^4.3.0" - }, - "engines": { - "node": ">=6.14.2" - } - }, "node_modules/@cosmjs/socket/node_modules/ws": { "version": "7.5.10", "resolved": "https://registry.npmjs.org/ws/-/ws-7.5.10.tgz", @@ -3104,21 +3089,6 @@ "ws": "7.4.6" } }, - "node_modules/@ethersproject/providers/node_modules/utf-8-validate": { - "version": "5.0.10", - "resolved": "https://registry.npmjs.org/utf-8-validate/-/utf-8-validate-5.0.10.tgz", - "integrity": "sha512-Z6czzLq4u8fPOyx7TU6X3dvUZVvoJmxSQ+IcrlmagKhilxlhZgxPK6C5Jqbkw1IDUmFTM+cz9QDnnLTwDz/2gQ==", - "hasInstallScript": true, - "license": "MIT", - "optional": true, - "peer": true, - "dependencies": { - "node-gyp-build": "^4.3.0" - }, - "engines": { - "node": ">=6.14.2" - } - }, "node_modules/@ethersproject/providers/node_modules/ws": { "version": "7.4.6", "resolved": "https://registry.npmjs.org/ws/-/ws-7.4.6.tgz", @@ -9971,21 +9941,6 @@ "ws": "^7.5.1" } }, - "node_modules/@walletconnect/jsonrpc-ws-connection/node_modules/utf-8-validate": { - "version": "5.0.10", - "resolved": "https://registry.npmjs.org/utf-8-validate/-/utf-8-validate-5.0.10.tgz", - "integrity": "sha512-Z6czzLq4u8fPOyx7TU6X3dvUZVvoJmxSQ+IcrlmagKhilxlhZgxPK6C5Jqbkw1IDUmFTM+cz9QDnnLTwDz/2gQ==", - "hasInstallScript": true, - "license": "MIT", - "optional": true, - "peer": true, - "dependencies": { - "node-gyp-build": "^4.3.0" - }, - "engines": { - "node": ">=6.14.2" - } - }, "node_modules/@walletconnect/jsonrpc-ws-connection/node_modules/ws": { "version": "7.5.10", "resolved": "https://registry.npmjs.org/ws/-/ws-7.5.10.tgz", @@ -14608,21 +14563,6 @@ "integrity": "sha512-GpVkmM8vF2vQUkj2LvZmD35JxeJOLCwJ9cUkugyk2nuhbv3+mJvpLYYt+0+USMxE+oj+ey/lJEnhZw75x/OMcQ==", "license": "MIT" }, - "node_modules/jayson/node_modules/utf-8-validate": { - "version": "5.0.10", - "resolved": "https://registry.npmjs.org/utf-8-validate/-/utf-8-validate-5.0.10.tgz", - "integrity": "sha512-Z6czzLq4u8fPOyx7TU6X3dvUZVvoJmxSQ+IcrlmagKhilxlhZgxPK6C5Jqbkw1IDUmFTM+cz9QDnnLTwDz/2gQ==", - "hasInstallScript": true, - "license": "MIT", - "optional": true, - "peer": true, - "dependencies": { - "node-gyp-build": "^4.3.0" - }, - "engines": { - "node": ">=6.14.2" - } - }, "node_modules/jayson/node_modules/uuid": { "version": "8.3.2", "resolved": "https://registry.npmjs.org/uuid/-/uuid-8.3.2.tgz", diff --git a/src/index.ts b/src/index.ts index 017c9c2..d870e54 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,102 +1,57 @@ -import "dotenv/config"; -import { handleUplinks } from "./services/mqtt"; -import { Request, Response } from "express"; -import { app, contractsReady } from "./services/context"; -import { - loadExtensionsFromConfig, - loadUIExtensionsFromConfig, - getUIComponents, - invokeUIAction, - runHook, -} from "./lib/utils"; -import setupDatabase, { getAllMeterRecords, deleteMeterByPublicKey } from "./store/sqlite"; - -// Async initialization function -async function initializeApp() { +import 'dotenv/config'; +import { handleUplinks } from './services/mqtt'; +import { contractsReady } from './services/contracts'; +import { startServer } from './services/server'; +import './services/ssh'; // attaches ssh and websocket to server +import { loadExtensionsFromConfig, loadUIExtensionsFromConfig, runHook } from './lib/utils'; +import setupDatabase from './store/sqlite'; +import { createLogger } from './utils/logger'; + +const logger = createLogger('[index]'); + +(async function () { try { - console.log("[info] Starting application initialization..."); + logger.info('Starting application initialization(v1.0.0)...'); // Load extensions from config await loadExtensionsFromConfig(); - console.log("[info] Extensions loaded successfully"); + logger.info('Extensions loaded successfully'); // Load UI extensions await loadUIExtensionsFromConfig(); - console.log("[info] UI extensions loaded successfully"); + logger.info('UI extensions loaded successfully'); - await contractsReady; - console.log("[info] resolved contract addresses"); + await contractsReady(); - runHook("onBeforeInit"); + runHook('onBeforeInit'); // Initialize database tables and jobs setupDatabase(); - console.log("[info] Database setup completed"); + logger.info('Database setup completed'); - runHook("onDatabaseSetup"); + runHook('onDatabaseSetup'); try { - console.log("[info] Initializing MQTT handling..."); + logger.info('Initializing MQTT handling...'); // Start MQTT handling await handleUplinks(); - console.log("[info] MQTT handling initialized successfully"); + logger.info('MQTT handling initialized successfully'); } catch (mqttError) { - console.error("[error] MQTT initialization failed:", mqttError); + logger.error('[error] MQTT initialization failed:', mqttError); throw mqttError; } - console.log("[info] Application initialization completed successfully"); + logger.info('Application initialization completed successfully'); + + startServer(); - runHook("onAfterInit"); + runHook('onAfterInit'); } catch (error) { - console.error("[fatal] Failed to initialize application:", error); - runHook("onInitError", error); + logger.error('[fatal] Failed to initialize application:', error); + runHook('onInitError', error); process.exit(1); } -} - -// Start initialization -initializeApp(); - -app.get("/", async (req: Request, res: Response) => { - const m3ters = getAllMeterRecords(); - - // Get UI components from loaded UI extensions - const { icons, windows } = await getUIComponents(); - - res.render("index", { m3ters, icons, windows }); - console.log("[server]: Server handled GET request at `/`"); -}); - -// API endpoint to invoke UI actions -app.post("/api/actions/:moduleId/:actionId", async (req: Request, res: Response) => { - const { moduleId, actionId } = req.params; - console.log(`[server]: Invoking action '${actionId}' from module '${moduleId}'`); - - const result = await invokeUIAction(moduleId, actionId); - - if (result.success) { - res.status(200).json(result); - } else { - res.status(400).json(result); - } -}); - -app.delete("/delete-meter", async (req: Request, res: Response) => { - let publicKey = decodeURIComponent(req.query?.publicKey?.toString() as string); - if (publicKey) { - const deleted = deleteMeterByPublicKey(publicKey.toString()); - if (!deleted) { - res.status(404).send({ message: "meter not found" }); - return; - } - } else { - res.status(400).send({ message: "public key not specified" }); - return; - } - console.log("[server]: Server handled DELETE request at `/delete-meter`"); - res.status(200).send({ message: "deleted meter" }); -}); +})(); diff --git a/src/lib/core/arweave/index.ts b/src/lib/core/arweave/index.ts index d0e076a..49d6db5 100644 --- a/src/lib/core/arweave/index.ts +++ b/src/lib/core/arweave/index.ts @@ -2,6 +2,9 @@ import { ArweaveSigner, TurboFactory } from "@ardrive/turbo-sdk"; import { Readable } from "stream"; import Arweave from "arweave"; import type { DecodedPayload, Hooks } from "../../../types"; +import { createLogger } from "../../../utils/logger"; + +const logger = createLogger("[arweave]"); export default class implements Hooks { async onTransactionDistribution(m3terId: number, decoded: DecodedPayload) { @@ -45,20 +48,20 @@ export default class implements Hooks { }, events: { onUploadProgress: (progress) => { - console.log("[arweave] Upload progress:", progress); + logger.info("Upload progress:", progress); }, onError: (error) => { - console.error("[arweave] Upload error:", error); + logger.error("Upload error:", error); }, onSuccess(event) { - console.log("[arweave] Upload successful! Transaction ID:", event); + logger.info("Upload successful! Transaction ID:", event); }, onUploadSuccess(event) { - console.log("[arweave] Upload completed! Transaction ID:", event); + logger.info("Upload completed! Transaction ID:", event); }, }, }); - console.log(`[arweave] Uploaded transaction ${decoded.nonce} for M3ter ID ${m3terId} to Arweave.`); + logger.info(`Uploaded transaction ${decoded.nonce} for M3ter ID ${m3terId} to Arweave.`); } } diff --git a/src/lib/core/prover/index.ts b/src/lib/core/prover/index.ts index d0dbd67..61bc7e6 100644 --- a/src/lib/core/prover/index.ts +++ b/src/lib/core/prover/index.ts @@ -1,28 +1,30 @@ -import cron from "node-cron"; -import { pruneAndSyncOnchain } from "../../sync"; -import { buildBatchPayload, retry } from "../../utils"; -import { getAllMeterRecords, getAllTransactionRecords, getTransactionByNonce } from "../../../store/sqlite"; -import type { BatchTransactionPayload, DecodedPayload, Hooks, TransactionRecord } from "../../../types"; +import cron from 'node-cron'; +import { pruneAndSyncOnchain } from '../../sync'; +import { buildBatchPayload, retry } from '../../utils'; +import { getAllMeterRecords, getAllTransactionRecords, getTransactionByNonce } from '../../../store/sqlite'; +import type { BatchTransactionPayload, DecodedPayload, Hooks, TransactionRecord } from '../../../types'; +import { createLogger } from '../../../utils/logger'; const PREFERRED_PROVER_NODE = process.env.PREFERRED_PROVER_NODE; +const logger = createLogger('[prover-job]'); export default class implements Hooks { - private proverSchedule: string = process.env.PROVER_CRONSCHEDULE || "0 0 * * *"; // default to every 24 hours + private proverSchedule: string = process.env.PROVER_CRONSCHEDULE || '0 0 * * *'; // default to every 24 hours async onAfterInit() { if (!PREFERRED_PROVER_NODE) { - console.warn("No PREFERRED_PROVER_NODE configured. Prover hooks will not be registered."); + logger.warn('No PREFERRED_PROVER_NODE configured. Prover hooks will not be registered.'); return; } - console.log("Registering prover cron job..."); - console.log("Schedule: ", this.proverSchedule, " Prover URL: ", PREFERRED_PROVER_NODE); + logger.info('Registering prover cron job...'); + logger.info('Schedule: ', this.proverSchedule, ' Prover URL: ', PREFERRED_PROVER_NODE); // Schedule a cron job to publish pending transactions cron.schedule( this.proverSchedule, async () => { - console.log( + logger.info( `Prover cron job started: Pruning meters and sending pending transactions to the prover at ${PREFERRED_PROVER_NODE}...`, ); const m3ters = getAllMeterRecords(); @@ -30,11 +32,11 @@ export default class implements Hooks { try { pruneAndSyncOnchain(m3ter.publicKey); } catch (error) { - console.error(`Error pruning and syncing meter ${m3ter.publicKey}:`, error); + logger.error(`Error pruning and syncing meter ${m3ter.publicKey}:`, error); } } - console.log("sending pending transactions to prover..."); + logger.info('sending pending transactions to prover...'); const pendingTransactions = getAllTransactionRecords(); const proverUrl = await this.getProverURL(); @@ -43,17 +45,17 @@ export default class implements Hooks { try { const response = await this.sendPendingTransactionsToProver(proverUrl!, pendingTransactions); - console.info("done sending all pending transactions to prover"); - console.info(`Prover response (text): ${await response?.text()}`); + logger.info('done sending all pending transactions to prover'); + logger.info(`Prover response (text): ${await response?.text()}`); } catch (error) { - console.error(`Error sending pending transactions to prover: ${error}`); + logger.error(`Error sending pending transactions to prover: ${error}`); } } }, - { name: "prover-publish-pending-transactions", noOverlap: true }, + { name: 'prover-publish-pending-transactions', noOverlap: true }, ); - console.log("Prover cron job registered."); + logger.info('Prover cron job registered.'); return; } @@ -63,15 +65,15 @@ export default class implements Hooks { const proverURL = await this.getProverURL(); if (!proverURL) { - console.warn("No prover URL configured. Skipping sending transactions to prover."); + logger.warn('No prover URL configured. Skipping sending transactions to prover.'); return; } - console.info(`Sending new transaction to prover: ${proverURL}`); + logger.info(`Sending new transaction to prover: ${proverURL}`); const latestTransaction = getTransactionByNonce(currentTransaction.nonce, tokenId); if (!latestTransaction) { - console.warn( + logger.warn( `No transactions found for meter ${tokenId} with nonce ${currentTransaction.nonce}. Skipping sending to prover.`, ); return; @@ -79,10 +81,10 @@ export default class implements Hooks { const response = await this.sendPendingTransactionsToProver(proverURL!, [latestTransaction]); - console.info("done sending to prover"); - console.info(`Prover response (text): ${await response?.text()}`); + logger.info('done sending to prover'); + logger.info(`Prover response (text): ${await response?.text()}`); } catch (error) { - console.error(`Error sending pending transactions to prover: ${error}`); + logger.error(`Error sending pending transactions to prover: ${error}`); } } @@ -92,9 +94,9 @@ export default class implements Hooks { ): Promise { try { const response = await fetch(`${proverURL}/batch-payloads`, { - method: "POST", + method: 'POST', headers: { - "Content-Type": "application/json", + 'Content-Type': 'application/json', }, body: JSON.stringify(transactionData), }); @@ -105,7 +107,7 @@ export default class implements Hooks { return response; } catch (err: any) { - console.error("Failed to send transactions to prover:", err.message); + logger.error('Failed to send transactions to prover:', err.message); throw err; } } @@ -115,7 +117,7 @@ export default class implements Hooks { } async sendPendingTransactionsToProver(proverURL: string, pendingTransactions: TransactionRecord[]) { - console.log("[info] Sending", pendingTransactions.length, "transactions to prover at", proverURL); + logger.info('[info] Sending', pendingTransactions.length, 'transactions to prover at', proverURL); const requestPayload = buildBatchPayload(pendingTransactions); diff --git a/src/lib/core/streamr/index.ts b/src/lib/core/streamr/index.ts index 966447d..0a4812d 100644 --- a/src/lib/core/streamr/index.ts +++ b/src/lib/core/streamr/index.ts @@ -4,6 +4,9 @@ import { getAllMeterRecords, getAllTransactionRecords } from "../../../store/sql import { buildBatchPayload, retry } from "../../utils"; import type { Hooks, TransactionRecord } from "../../../types"; import { pruneAndSyncOnchain } from "../../sync"; +import { createLogger } from "../../../utils/logger"; + +const logger = createLogger("[streamr]"); const { ETHEREUM_PRIVATE_KEY } = process.env; @@ -16,7 +19,7 @@ export default class implements Hooks { private cronSchedule: string = process.env.STREAMR_CRONSCHEDULE || "0 * * * *"; async onAfterInit() { - console.log( + logger.info( "Registering Streamr cron job... Schedule: ", this.cronSchedule, " Stream IDs: ", @@ -27,23 +30,23 @@ export default class implements Hooks { cron.schedule( this.cronSchedule, async () => { - console.log("Streamr cron job started: Pruning meters and publishing pending transactions..."); + logger.info("Streamr cron job started: Pruning meters and publishing pending transactions..."); const m3ters = getAllMeterRecords(); for (const m3ter of m3ters) { try { pruneAndSyncOnchain(m3ter.publicKey); } catch (error) { - console.error(`Error pruning and syncing meter ${m3ter.publicKey}:`, error); + logger.error(`Error pruning and syncing meter ${m3ter.publicKey}:`, error); } } - console.log("Publishing pending transactions to Streamr..."); + logger.info("Publishing pending transactions to Streamr..."); const pendingTransactions = await this.getPendingTransactions(); if (pendingTransactions.length > 0) { await Promise.all( this.streamIds.map(async (STREAMR_STREAM_ID) => { - console.log(`Publishing to Streamr stream: ${STREAMR_STREAM_ID}`); + logger.info(`Publishing to Streamr stream: ${STREAMR_STREAM_ID}`); await retry( () => this.publishPendingTransactionsToStreamr(STREAMR_STREAM_ID, pendingTransactions), 3, @@ -56,7 +59,7 @@ export default class implements Hooks { { name: "streamr-publish-pending-transactions", noOverlap: true }, ); - console.log("Streamr cron job registered."); + logger.info("Streamr cron job registered."); return; } @@ -72,26 +75,26 @@ export default class implements Hooks { }); try { - console.log(`[streamr] Connecting to ${STREAMR_STREAM_ID}...`); + logger.info(`Connecting to ${STREAMR_STREAM_ID}...`); const stream = await retry(() => streamrClient.getStream(STREAMR_STREAM_ID!), 3, 2000); await new Promise((resolve) => setTimeout(resolve, 2000)); // wait for 2 seconds to ensure connection is established - console.log(`[streamr] Connected. Publishing ${pendingTransactions.length} transactions...`); + logger.info(`Connected. Publishing ${pendingTransactions.length} transactions...`); const batchPayload = buildBatchPayload(pendingTransactions); await stream.publish(batchPayload); - console.log(`[streamr] Published ${pendingTransactions.length} transactions to stream ${STREAMR_STREAM_ID}`); + logger.info(`Published ${pendingTransactions.length} transactions to stream ${STREAMR_STREAM_ID}`); await new Promise((resolve) => setTimeout(resolve, 100000)); // wait for 100 seconds to ensure message is sent } catch (error) { - console.error(`[streamr] Error publishing to Streamr:`, error); + logger.error(`Error publishing to Streamr:`, error); throw error; } finally { // destroy the client to free resources try { await streamrClient.destroy(); } catch (destroyError) { - console.error(`[streamr] Error destroying client:`, destroyError); + logger.error(`Error destroying client:`, destroyError); } } } diff --git a/src/lib/core/streamr/ui.ts b/src/lib/core/streamr/ui.ts index 2f1a0bd..2f584e8 100644 --- a/src/lib/core/streamr/ui.ts +++ b/src/lib/core/streamr/ui.ts @@ -2,6 +2,9 @@ import { getAllTransactionRecords } from "../../../store/sqlite"; import { buildBatchPayload, retry } from "../../utils"; import type { UIHooks, UIAppIcon, UIAppWindow, UIAction, TransactionRecord } from "../../../types"; import { StreamrClient } from "@streamr/sdk"; +import { createLogger } from "../../../utils/logger"; + +const logger = createLogger("[streamr-ui]"); const { ETHEREUM_PRIVATE_KEY } = process.env; @@ -97,7 +100,7 @@ export default class implements UIHooks { try { await Promise.all( this.streamIds.map(async (streamId) => { - console.log(`[streamr-ui] Publishing to stream: ${streamId}`); + logger.info(`Publishing to stream: ${streamId}`); await retry(() => this.publishToStreamr(streamId, pendingTransactions), 3, 2000); }), ); @@ -148,15 +151,15 @@ export default class implements UIHooks { await new Promise((resolve) => setTimeout(resolve, 2000)); // wait for 2 seconds to ensure connection is established await stream.publish(batchPayload); await new Promise((resolve) => setTimeout(resolve, 100000)); // wait for 100 seconds to ensure message is sent - console.log(`[streamr-ui] Published ${pendingTransactions.length} transactions to stream ${streamId}`); + logger.info(`Published ${pendingTransactions.length} transactions to stream ${streamId}`); } catch (error) { - console.error(`[streamr-ui] Error publishing to Streamr:`, error); + logger.error(`Error publishing to Streamr:`, error); throw error; } finally { try { await streamrClient.destroy(); } catch (destroyError) { - console.error(`[streamr-ui] Error destroying client:`, destroyError); + logger.error(`Error destroying client:`, destroyError); } } } diff --git a/src/lib/sync.ts b/src/lib/sync.ts index 07d7036..38e40ac 100644 --- a/src/lib/sync.ts +++ b/src/lib/sync.ts @@ -5,17 +5,19 @@ import { pruneTransactionsAfter, pruneTransactionsBefore, updateMeterNonce, -} from "../store/sqlite"; +} from '../store/sqlite'; import { provider, - rollup as rollupContract, - contractsReady, ccipRevenueReader as ccipRevenueReaderContract, priceContext as priceContextContract, -} from "../services/context"; -import { ZeroAddress } from "ethers"; -import { retry } from "./utils"; -import type { VerifierInfo } from "../types"; +} from '../services/contracts'; +import { ZeroAddress } from 'ethers'; +import { retry } from './utils'; +import type { VerifierInfo } from '../types'; +import { meterLatestNonceContractCall } from '../utils/contracts'; +import { createLogger } from '../utils/logger'; + +const logger = createLogger('[sync]'); // Cache for verifiers - populated once on startup let verifiersCache: VerifierInfo[] | null = null; @@ -28,11 +30,11 @@ let isCacheInitialized = false; */ export async function initializeVerifiersCache(): Promise { try { - console.log("[info] Initializing verifiers cache..."); + logger.info('Initializing verifiers cache...'); // Get the number of verifiers const verifierCount = Number(await retry(() => ccipRevenueReaderContract.verifierCount())); - console.log(`[info] Found ${verifierCount} verifiers to cache`); + logger.info(`Found ${verifierCount} verifiers to cache`); const verifiers: VerifierInfo[] = []; @@ -42,7 +44,7 @@ export async function initializeVerifiersCache(): Promise { // Get verifier info (ensName, targetContractAddress) const [ensName, targetAddress] = await retry(() => ccipRevenueReaderContract.verifiers(i)); - console.log(`[info] Fetching verifier ${i}: ENS: ${ensName}, target: ${targetAddress}`); + logger.info(`Fetching verifier ${i}: ENS: ${ensName}, target: ${targetAddress}`); // Resolve ENS name to get the verifier address const verifierAddress = await retry(() => provider.resolveName(ensName)); @@ -51,7 +53,7 @@ export async function initializeVerifiersCache(): Promise { throw new Error(`Failed to resolve ENS name: ${ensName}`); } - console.log(`[info] Resolved ${ensName} to verifier address: ${verifierAddress}`); + logger.info(`Resolved ${ensName} to verifier address: ${verifierAddress}`); verifiers.push({ ensName, @@ -59,7 +61,7 @@ export async function initializeVerifiersCache(): Promise { verifierAddress, }); } catch (error) { - console.error(`[error] Failed to initialize verifier ${i}:`, error); + logger.error(`Failed to initialize verifier ${i}:`, error); throw error; // Fail fast as requested } } @@ -68,9 +70,9 @@ export async function initializeVerifiersCache(): Promise { verifiersCache = verifiers; isCacheInitialized = true; - console.log(`[info] Successfully cached ${verifiers.length} verifiers`); + logger.info(`Successfully cached ${verifiers.length} verifiers`); } catch (error) { - console.error("[error] Failed to initialize verifiers cache:", error); + logger.error('Failed to initialize verifiers cache:', error); isCacheInitialized = false; verifiersCache = null; throw error; @@ -102,17 +104,15 @@ export function getCachedVerifiersCount(): number { } export async function pruneAndSyncOnchain(meterIdentifier: number | string): Promise { - await contractsReady; - const meter = - typeof meterIdentifier === "number" ? getMeterByTokenId(meterIdentifier) : getMeterByPublicKey(meterIdentifier); + typeof meterIdentifier === 'number' ? getMeterByTokenId(meterIdentifier) : getMeterByPublicKey(meterIdentifier); if (!meter) { throw new Error(`Meter with identifier ${meterIdentifier} not found`); } // Check the latest nonce on the blockchain - const onchainNonce = Number(await rollupContract.nonce(meter.tokenId)); + const onchainNonce = await meterLatestNonceContractCall(meter.tokenId); const latestNonce = meter.latestNonce; if (onchainNonce > latestNonce) { @@ -127,26 +127,22 @@ export async function pruneAndSyncOnchain(meterIdentifier: number | string): Pro } export async function getLatestTransactionNonce(meterIdentifier: number): Promise { - await contractsReady; - // get latest nonce from chain - let latestNonce = Number(await rollupContract.nonce(meterIdentifier)); + let latestNonce = await meterLatestNonceContractCall(meterIdentifier); // check local state for the highest nonce we have - while (true) { - const existingTransaction = getTransactionByNonce(latestNonce + 1, meterIdentifier); - if (existingTransaction) { - latestNonce += 1; - } else { - pruneTransactionsAfter(latestNonce, meterIdentifier); - break; - } + let localLatestNonce = (getMeterByTokenId(meterIdentifier))?.latestNonce || 0; + + if (localLatestNonce > latestNonce) { + latestNonce = localLatestNonce; } + pruneTransactionsAfter(latestNonce, meterIdentifier); // remove zombie transaction records (just in case) + return latestNonce; } -// get revenue across suppored chains +// get revenue across supported chains export async function getCrossChainRevenue(tokenId: number): Promise { try { // Use cached verifiers instead of fetching them each time @@ -157,8 +153,8 @@ export async function getCrossChainRevenue(tokenId: number): Promise { // Iterate through all cached verifiers and get revenue from each chain for (const verifier of verifiers) { try { - console.log( - `[info] Getting revenue from ENS: ${verifier.ensName}, target: ${verifier.targetAddress}, verifier: ${verifier.verifierAddress}`, + logger.info( + `Getting revenue from ENS: ${verifier.ensName}, target: ${verifier.targetAddress}, verifier: ${verifier.verifierAddress}`, ); // Get revenue from this specific chain using CCIP read @@ -170,18 +166,18 @@ export async function getCrossChainRevenue(tokenId: number): Promise { ); const revenueAmount = Number(revenue); - console.log(`[info] Revenue from ${verifier.ensName} (${verifier.verifierAddress}): ${revenueAmount}`); + logger.info(`Revenue from ${verifier.ensName} (${verifier.verifierAddress}): ${revenueAmount}`); totalRevenue += revenueAmount; } catch (error) { - console.error(`[error] Failed to get revenue from verifier ${verifier.ensName}:`, error); + logger.error(`Failed to get revenue from verifier ${verifier.ensName}:`, error); // Continue with other verifiers even if one fails } } - console.log(`[info] Total cross-chain revenue for token ${tokenId}: ${totalRevenue}`); + logger.info(`Total cross-chain revenue for token ${tokenId}: ${totalRevenue}`); return totalRevenue; } catch (error) { - console.error(`[error] Failed to get cross-chain revenue for token ${tokenId}:`, error); + logger.error(`Failed to get cross-chain revenue for token ${tokenId}:`, error); throw error; } } @@ -190,17 +186,17 @@ export async function getCrossChainRevenue(tokenId: number): Promise { export async function getOwedFromPriceContext(tokenId: number): Promise { try { return await retry(async () => { - console.log(`[info] Getting owed amount for token ${tokenId} from price context`); + logger.info(`Getting owed amount for token ${tokenId} from price context`); // Call the price context to get the amount the user owes with CCIP read enabled const owedAmount = await priceContextContract.owed(tokenId); const owed = Number(owedAmount); - console.log(`[info] Owed amount for token ${tokenId}: ${owed}`); + logger.info(`Owed amount for token ${tokenId}: ${owed}`); return owed; }); } catch (error) { - console.error(`[error] Failed to get owed amount for token ${tokenId}:`, error); + logger.error(`Failed to get owed amount for token ${tokenId}:`, error); throw error; } } diff --git a/src/lib/utils.ts b/src/lib/utils.ts index 4707db8..eae218c 100644 --- a/src/lib/utils.ts +++ b/src/lib/utils.ts @@ -10,6 +10,9 @@ import type { UIAppWindow, UIAction, } from "../types"; +import { createLogger } from "../utils/logger"; + +const logger = createLogger("[utils]"); const extensions: Hooks[] = []; const uiExtensions: Map = new Map(); @@ -17,11 +20,11 @@ const uiExtensions: Map = new Map(); const { BACKEND_MODULES, UI_MODULES } = process.env; if (!BACKEND_MODULES) { - console.warn("[config] No BACKEND_MODULES specified"); + logger.warn("No BACKEND_MODULES specified"); } if (!UI_MODULES) { - console.warn("[config] No UI_MODULES specified"); + logger.warn("No UI_MODULES specified"); } export const defaultConfigurations: AppConfig = { @@ -75,7 +78,7 @@ export async function loadUIExtensionsFromConfig(): Promise const config = defaultConfigurations as AppConfig & { uiModules?: Record }; if (!config.uiModules) { - console.warn("[ui] No UI modules configured"); + logger.warn("No UI modules configured"); return uiExtensions; } @@ -85,9 +88,9 @@ export async function loadUIExtensionsFromConfig(): Promise const mod = await import(resolved); const instance = new mod.default(); uiExtensions.set(moduleId, instance); - console.log(`[ui] Loaded UI module: ${moduleId}`); + logger.info(`Loaded UI module: ${moduleId}`); } catch (error) { - console.error(`[ui] Failed to load UI module ${moduleId}:`, error); + logger.error(`Failed to load UI module ${moduleId}:`, error); } } @@ -121,7 +124,7 @@ export async function getUIComponents(): Promise<{ actions.set(moduleId, moduleActions); } } catch (error) { - console.error(`[ui] Error getting components from ${moduleId}:`, error); + logger.error(`Error getting components from ${moduleId}:`, error); } } @@ -191,7 +194,7 @@ export async function retry(fn: () => Promise, maxRetries: number = 5, bas } const delay = baseDelay * Math.pow(2, attempt); - console.log(`Attempt ${attempt + 1} failed, retrying in ${delay}ms...`, error); + logger.warn(`Attempt ${attempt + 1} failed, retrying in ${delay}ms...`, error); await new Promise((resolve) => setTimeout(resolve, delay)); } @@ -227,7 +230,7 @@ export function verifyPayloadSignature(transaction: Buffer, rawPubKey: Buffer): return ok; } catch (error) { - console.error("Error verifying signature:", error); + logger.error("Error verifying signature:", error); return false; } } diff --git a/src/services/context.ts b/src/services/context.ts deleted file mode 100644 index bddc3d2..0000000 --- a/src/services/context.ts +++ /dev/null @@ -1,189 +0,0 @@ -import { create } from "express-handlebars"; -import express, { Express } from "express"; -import { JsonRpcProvider, Contract } from "ethers"; -import WebSocket from "ws"; -import { ConnectConfig, Client as SSHClient } from "ssh2"; -import http from "http"; -import { retry } from "../lib/utils"; - -// HBS CONFIG -const hbs = create({ - defaultLayout: "main", - extname: "hbs", - helpers: { - encodeURIComponent: function (value: string) { - return encodeURIComponent(value); - }, - }, -}); - -// EXPRESS APP CONFIG -export const app: Express = express(); -const server = http.createServer(app); -const wss = new WebSocket.Server({ server }); - -app.engine("hbs", hbs.engine); -app.set("view engine", "hbs"); -app.set("views", "./src/views"); -app.use(express.json()); -app.use(express.static("./src/public")); -app.use(express.urlencoded({ extended: true })); - -const port = process.env.PORT || 3000; - -wss.on("connection", (ws: WebSocket, request: http.IncomingMessage) => { - console.log("[ws]: New WebSocket connection established"); - - console.log("url", request.url); - - const url = new URL(request.url!, "http://localhost"); - const cols = url.searchParams.get("cols") ?? "80"; - const rows = url.searchParams.get("rows") ?? "24"; - const username = url.searchParams.get("username") ?? "ubuntu"; - const password = url.searchParams.get("password") ?? "Mauchly92618"; - - const ssh = new SSHClient(); - const sshConfig: ConnectConfig = { - host: "127.0.0.1", - port: 22, - username, - password, - }; - - ssh - .on("ready", () => { - console.log("[ws/ssh]: SSH connection established"); - ssh.shell({ term: "xterm", cols: parseInt(cols), rows: parseInt(rows) }, (err, stream) => { - if (err) { - ws.send(`[ws/ssh]: SSH shell error: ${err.message}`); - ws.close(); - ssh.end(); - return; - } - - // SSH -> WebSocket - stream.on("data", (data: Buffer) => { - if (ws.readyState === WebSocket.OPEN) { - ws.send(data); - } - }); - - stream.on("close", () => { - console.log("[ws/ssh]: SSH stream closed"); - ws.close(); - ssh.end(); - }); - - stream.stderr.on("data", (data: Buffer) => { - if (ws.readyState === WebSocket.OPEN) { - ws.send(`[ws/ssh]: SSH stderr: ${data.toString()}`); - } - }); - - // WebSocket -> SSH - console.log("[ws/ssh]: Setting up WebSocket message handler"); - ws.on("message", (msg) => { - // support JSON control messages for resize - try { - const parsed = JSON.parse(msg.toString()); - if (parsed.type === "resize") { - const { cols, rows } = parsed; - stream.setWindow(rows, cols, cols * 8, rows * 16); // width/height px optional - return; - } - } catch (e) { - /* not JSON - treat as raw data */ - } - - if (stream.writable) stream.write(msg); - }); - - ws.on("close", () => { - console.log("[ws]: WebSocket closed, closing SSH stream"); - stream.end(); - ssh.end(); - }); - - ws.on("error", () => { - console.log("[ws]: WebSocket error, closing SSH stream"); - stream.end(); - ssh.end(); - }); - }); - }) - .on("error", (err) => { - console.error("[ws/ssh]: SSH connection error:", err); - ws.send(`[ws/ssh]: SSH connection error: ${err.message}`); - ws.close(); - ssh.end(); - }) - .connect(sshConfig); -}); - -server.listen(port, () => { - console.log(`[server]: Server is running at port ${port}`); -}); - -console.log("[info] Starting application initialization..."); - -// ETHERS JS CONTRACT CONFIG -const rpcUrl = process.env.MAINNET_RPC; -console.log("[info] using RPC", rpcUrl) -export const provider = new JsonRpcProvider(rpcUrl); - -const m3terRef = process.env.M3TER_CONTRACT_ADDRESS || "heads.m3ter.eth"; -const rollupRef = process.env.ROLLUP_CONTRACT_ADDRESS || "rollup.m3ter.eth"; - -const m3terAbi = ["function publicKey(uint256) view returns (bytes32)", "function tokenID(bytes32) view returns (uint256)"]; -const rollupAbi = ["function nonce(uint256) external view returns (bytes6)"]; - -function isHexAddress(value: string): boolean { - return /^0x[a-fA-F0-9]{40}$/.test(value); -} - -async function resolveContractAddress(contractName: string, ref: string): Promise { - if (isHexAddress(ref)) { - console.log(`[ethers] ${contractName}: using address ${ref}`); - return ref; - } - - console.log(`[ethers] ${contractName}: resolving ENS name ${ref}...`); - - const resolved = await retry(() => provider.resolveName(ref), 3, 1500); - if (!resolved) { - throw new Error(`[ethers] ${contractName}: failed to resolve ENS ${ref}`); - } - - console.log(`[ethers] ${contractName}: resolved ENS ${ref} -> ${resolved}`); - return resolved; -} - -export let m3ter!: Contract; -export let rollup!: Contract; - -export const contractsReady = (async () => { - const [m3terAddress, rollupAddress] = await Promise.all([ - resolveContractAddress("m3ter", m3terRef), - resolveContractAddress("rollup", rollupRef), - ]); - - m3ter = new Contract(m3terAddress, m3terAbi, provider); - rollup = new Contract(rollupAddress, rollupAbi, provider); -})(); - -export const ccipRevenueReader = new Contract( - process.env.CCIP_REVENUE_READER_ADDRESS || "0xD648cdF47e9534B2FCfb18C1E94CA9AAff07BA0E", - [ - "function read(uint256 tokenId, address target, address verifier) public view returns (uint256)", - "function readCallback(bytes[] memory data, bytes memory) external pure returns (uint256)", - "function verifierCount() external view returns (uint256)", - "function verifiers(uint256) external view returns (string, address)", - ], - provider, -); - -export const priceContext = new Contract( - process.env.PRICE_CONTEXT_ADDRESS || "0xc6D5Ff8E80F4Ee511Db4bCf6a0BcEbF9f41aAA32", - ["function owed(uint256 tokenId) public view returns (uint256)"], - provider, -); diff --git a/src/services/contracts.ts b/src/services/contracts.ts new file mode 100644 index 0000000..d5d71d4 --- /dev/null +++ b/src/services/contracts.ts @@ -0,0 +1,62 @@ +import { JsonRpcProvider, Contract, isAddress } from 'ethers'; +import { retry } from '../lib/utils'; +import { createLogger } from '../utils/logger'; + +const logger = createLogger('[Context]'); + +const rpcUrl = process.env.MAINNET_RPC; + +logger.info('using RPC', rpcUrl); + +export const provider = new JsonRpcProvider(rpcUrl); + +export let rollup!: Contract; + +const rollupRef = process.env.ROLLUP_CONTRACT_ADDRESS || 'rollup.m3ter.eth'; + +const rollupAbi = [ + 'function nonce(uint64) external view returns (uint24)', + 'function m3terId(bytes32 key) external view returns (uint64)', + 'function publicKey(uint256) view returns (bytes32)', +]; + +export const ccipRevenueReader = new Contract( + process.env.CCIP_REVENUE_READER_ADDRESS || '0xD648cdF47e9534B2FCfb18C1E94CA9AAff07BA0E', + [ + 'function read(uint256 tokenId, address target, address verifier) public view returns (uint256)', + 'function readCallback(bytes[] memory data, bytes memory) external pure returns (uint256)', + 'function verifierCount() external view returns (uint256)', + 'function verifiers(uint256) external view returns (string, address)', + ], + provider, +); + +export const priceContext = new Contract( + process.env.PRICE_CONTEXT_ADDRESS || '0xc6D5Ff8E80F4Ee511Db4bCf6a0BcEbF9f41aAA32', + ['function owed(uint256 tokenId) public view returns (uint256)'], + provider, +); + +async function resolveContractAddress(contractName: string, ref: string): Promise { + if (isAddress(ref)) { + logger.info(`[ethers] ${contractName}: using address ${ref}`); + return ref; + } + + logger.info(`[ethers] ${contractName}: resolving ENS name ${ref}...`); + + const resolved = await retry(() => provider.resolveName(ref), 3, 1500); + if (!resolved) { + throw new Error(`[ethers] ${contractName}: failed to resolve ENS ${ref}`); + } + + logger.info(`[ethers] ${contractName}: resolved ENS ${ref} -> ${resolved}`); + return resolved; +} + +export const contractsReady = async () => { + if (rollup) return; + const rollupAddress = await resolveContractAddress('rollup', rollupRef); + + rollup = new Contract(rollupAddress, rollupAbi, provider); +}; diff --git a/src/services/grpc.ts b/src/services/grpc.ts index 6fe95dd..ba07918 100644 --- a/src/services/grpc.ts +++ b/src/services/grpc.ts @@ -1,6 +1,9 @@ import { credentials, Metadata } from "@grpc/grpc-js"; import { DeviceServiceClient } from "@chirpstack/chirpstack-api/api/device_grpc_pb"; import { DeviceQueueItem, EnqueueDeviceQueueItemRequest } from "@chirpstack/chirpstack-api/api/device_pb"; +import { createLogger } from "../utils/logger"; + +const logger = createLogger("[grpc]"); const deviceService = new DeviceServiceClient(`${process.env.CHIRPSTACK_HOST}:8080`, credentials.createInsecure()); @@ -18,7 +21,7 @@ export function enqueue(devEui: string, data: number[]) { enqueueReq.setQueueItem(item); deviceService.enqueue(enqueueReq, metadata, (err: any, resp: any) => { - if (err !== null) return console.log(err); - console.log(`${data} has been enqueued with id: ` + resp.getId()); + if (err !== null) return logger.error("Failed to enqueue device data:", err); + logger.info(`${data} has been enqueued with id: ` + resp.getId()); }); } diff --git a/src/services/mqtt.ts b/src/services/mqtt.ts index 14e586d..9f2b357 100644 --- a/src/services/mqtt.ts +++ b/src/services/mqtt.ts @@ -1,35 +1,27 @@ -import { connect } from "mqtt"; -import { enqueue } from "./grpc"; -import { encode } from "../lib/encode"; -import { app, m3ter as m3terContract, contractsReady } from "./context"; +import { connect } from 'mqtt'; +import { enqueue } from './grpc'; +import { encode } from '../lib/encode'; import { - deleteMeterByPublicKey, getAllTransactionRecords, getMeterByDevEui, getMeterByPublicKey, - getMeterByTokenId, insertTransaction, saveMeter, updateMeterDevEui, updateMeterNonce, -} from "../store/sqlite"; -import type { State, TransactionRecord } from "../types"; -import { decodePayload } from "../lib/decode"; -import { runHook, verifyPayloadSignature } from "../lib/utils"; -import { - getLatestTransactionNonce, - pruneAndSyncOnchain, - getCrossChainRevenue, - getOwedFromPriceContext, -} from "../lib/sync"; -import { createMeterLogger } from "../utils/logger"; - +} from '../store/sqlite'; +import type { State, TransactionRecord } from '../types'; +import { decodePayload } from '../lib/decode'; +import { runHook, verifyPayloadSignature } from '../lib/utils'; +import { getLatestTransactionNonce, getCrossChainRevenue, getOwedFromPriceContext } from '../lib/sync'; +import { createMeterLogger, createLogger } from '../utils/logger'; +import { meterIdFromPublicKeyContractCall } from '../utils/contracts'; + +const logger = createLogger('[mqtt]'); const CHIRPSTACK_HOST = process.env.CHIRPSTACK_HOST; const deviceLocks = new Map(); // Lock per devEUI to prevent concurrent message processing export async function handleUplinks(): Promise { - await contractsReady; - return new Promise(function (resolve, reject) { const client = connect({ host: CHIRPSTACK_HOST, @@ -39,30 +31,30 @@ export async function handleUplinks(): Promise { reconnectPeriod: 1000, }); - client.on("reconnect", () => { - runHook("onMqttReconnect", client); + client.on('reconnect', () => { + runHook('onMqttReconnect', client); }); - client.on("message", async (_, blob) => { + client.on('message', async (_, blob) => { return await handleMessage(blob); }); - client.on("connect", () => { + client.on('connect', () => { const topic = `application/${process.env.APPLICATION_ID}/device/+/event/up`; client.subscribe(topic, () => { - console.log(`\nConnected & Subscribed to CHIRPSTACK_HOST: ${CHIRPSTACK_HOST}\n`); + logger.info(`\nConnected & Subscribed to CHIRPSTACK_HOST: ${CHIRPSTACK_HOST}\n`); - runHook("onMqttSubscribed", client, topic); + runHook('onMqttSubscribed', client, topic); resolve(true); }); - runHook("onMqttConnect", client); + runHook('onMqttConnect', client); }); - client.on("error", (err) => { - console.error("Connection error: ", err); + client.on('error', (err) => { + logger.error('Connection error: ', err); - runHook("onMqttError", err, client); + runHook('onMqttError', err, client); client.end(); reject(err); @@ -71,24 +63,24 @@ export async function handleUplinks(): Promise { } export async function handleMessage(blob: Buffer) { - runHook("onMessageReceived", blob); + runHook('onMessageReceived', blob); const message = JSON.parse(blob.toString()); - const devEui = message["deviceInfo"]["devEui"] || null; + const devEui = message['deviceInfo']['devEui'] || null; // Create a logger with devEui context const logger = createMeterLogger({ devEui }); try { if (!devEui) { - console.log("[warn] Message dropped - no devEui found in message"); + logger.warn('Message dropped - no devEui found in message'); return; } // Check if this specific device is already being processed if (deviceLocks.get(devEui)) { logger.warn(`Message dropped - device is already being processed`); - runHook("onMessageDropped", "locked", devEui); + runHook('onMessageDropped', 'locked', devEui); return; } @@ -101,7 +93,7 @@ export async function handleMessage(blob: Buffer) { // encoded transaction in standard format (payload is hex string) // format: nonce | energy | signature | voltage | device_id | longitude | latitude - const transactionHex = Buffer.from(message["data"], "base64"); + const transactionHex = Buffer.from(message['data'], 'base64'); const decoded = decodePayload(transactionHex); let publicKey = decoded.extensions.deviceId; let payloadHadPublicKey = !!publicKey; @@ -113,19 +105,19 @@ export async function handleMessage(blob: Buffer) { const meterByDevEui = getMeterByDevEui(devEui); if (!meterByDevEui) { - throw new Error("Device EUI not associated with any meter: " + devEui); + throw new Error('Device EUI not associated with any meter: ' + devEui); } - publicKey = meterByDevEui.publicKey.replace("0x", ""); + publicKey = meterByDevEui.publicKey.replace('0x', ''); } // verify transaction signature - const isValid = verifyPayloadSignature(transactionHex, Buffer.from(publicKey!, "hex")); + const isValid = verifyPayloadSignature(transactionHex, Buffer.from(publicKey!, 'hex')); if (!isValid) { - throw new Error("Invalid transaction signature for meter with public key: " + publicKey); + throw new Error('Invalid transaction signature for meter with public key: ' + publicKey); } - logger.info("Verified signature"); + logger.info('Verified signature'); if (payloadHadPublicKey) { logger.info(`Payload contained public key: ${publicKey}`); @@ -133,7 +125,7 @@ export async function handleMessage(blob: Buffer) { const existingMeter = getMeterByPublicKey(`0x${publicKey}`); if (!existingMeter) { - const tokenId = Number(await m3terContract.tokenID(`0x${publicKey}`)); + const tokenId = await meterIdFromPublicKeyContractCall(`0x${publicKey}`); const latestNonce = await getLatestTransactionNonce(tokenId); @@ -142,27 +134,20 @@ export async function handleMessage(blob: Buffer) { // save new meter with devEui const newMeter = { publicKey: `0x${publicKey}`, - devEui: message["deviceInfo"]["devEui"], + devEui: message['deviceInfo']['devEui'], tokenId, latestNonce, }; - const existingMeter = getMeterByTokenId(tokenId); - - // in-case of the public key being updated - if (existingMeter && existingMeter.publicKey !== `0x${publicKey}`) { - deleteMeterByPublicKey(`0x${publicKey}`); - } - saveMeter(newMeter); logger.info(`Saved new meter: ${JSON.stringify(newMeter)}`); - runHook("onMeterCreated", newMeter); + runHook('onMeterCreated', newMeter); } else { // update existing meter with devEui if not already set - if (!existingMeter.devEui || existingMeter.devEui !== message["deviceInfo"]["devEui"]) { - logger.info(`Updating meter with DevEui: ${message["deviceInfo"]["devEui"]}`); - updateMeterDevEui(`0x${publicKey}`, message["deviceInfo"]["devEui"]); + if (!existingMeter.devEui || existingMeter.devEui !== message['deviceInfo']['devEui']) { + logger.info(`Updating meter with DevEui: ${message['deviceInfo']['devEui']}`); + updateMeterDevEui(`0x${publicKey}`, message['deviceInfo']['devEui']); } // fetch and update latest nonce from chain @@ -177,14 +162,14 @@ export async function handleMessage(blob: Buffer) { let m3ter = getMeterByPublicKey(`0x${publicKey}`) ?? null; if (!m3ter) { - throw new Error("Meter not found for public key: " + publicKey); + throw new Error('Meter not found for public key: ' + publicKey); } logger.info(`Found meter: ${JSON.stringify(m3ter)}`); // If both latest nonce and received nonce are 0, enqueue 0 immediately if (m3ter.latestNonce === 0 && decoded.nonce === 0) { - logger.info("Both latest nonce and received nonce are 0, enqueuing 0 immediately"); + logger.info('Both latest nonce and received nonce are 0, enqueuing 0 immediately'); try { is_on = true; // Always on @@ -199,7 +184,7 @@ export async function handleMessage(blob: Buffer) { logger.info(`Enqueuing state: ${JSON.stringify(state)}`); enqueue( - message["deviceInfo"]["devEui"], + message['deviceInfo']['devEui'], encode(state as State, decoded.extensions.latitude ?? 0, decoded.extensions.longitude ?? 0), ); @@ -223,7 +208,7 @@ export async function handleMessage(blob: Buffer) { nonce: decoded.nonce, identifier: m3ter.tokenId, receivedAt: Date.now(), - raw: transactionHex.toString("hex"), + raw: transactionHex.toString('hex'), } as TransactionRecord; insertTransaction(transactionRecord); updateMeterNonce(`0x${publicKey}`, expectedNonce); @@ -231,37 +216,37 @@ export async function handleMessage(blob: Buffer) { logger.info(`Updated meter nonce to: ${expectedNonce}`); const pendingTransactions = getAllTransactionRecords(); - console.log(`${pendingTransactions.length} Pending transactions: ${JSON.stringify(pendingTransactions)}`); - await runHook("onTransactionDistribution", m3ter.tokenId, decoded, pendingTransactions); + logger.info(`${pendingTransactions.length} Pending transactions: ${JSON.stringify(pendingTransactions)}`); + await runHook('onTransactionDistribution', m3ter.tokenId, decoded, pendingTransactions); } try { - is_on = await runHook("isOnStateCompute", m3ter.tokenId); + is_on = await runHook('isOnStateCompute', m3ter.tokenId); } catch (error) { - runHook("onIsOnStateComputeError", m3ter.tokenId, error); + runHook('onIsOnStateComputeError', m3ter.tokenId, error); logger.error(`Error in isOnStateCompute hook: ${error}`); } - runHook("onIsOnStateComputed", m3ter.tokenId, is_on); + runHook('onIsOnStateComputed', m3ter.tokenId, is_on); const state = decoded.nonce === expectedNonce ? { is_on } : { nonce: m3ter.latestNonce, is_on }; logger.info(`Enqueuing state: ${JSON.stringify(state)}`); enqueue( - message["deviceInfo"]["devEui"], + message['deviceInfo']['devEui'], encode(state as State, decoded.extensions.latitude ?? 0, decoded.extensions.longitude ?? 0), ); - runHook("onStateEnqueued", state, decoded.extensions.latitude ?? 0, decoded.extensions.longitude ?? 0); + runHook('onStateEnqueued', state, decoded.extensions.latitude ?? 0, decoded.extensions.longitude ?? 0); } catch (error) { logger.error(`Error handling MQTT message: ${error}`); - runHook("onMessageError", error); + runHook('onMessageError', error); } finally { // Release lock for this specific device if (devEui) { deviceLocks.delete(devEui); - runHook("onDeviceUnlocked", devEui); + runHook('onDeviceUnlocked', devEui); } - runHook("onMessageProcessingComplete"); + runHook('onMessageProcessingComplete'); } } diff --git a/src/services/server.ts b/src/services/server.ts new file mode 100644 index 0000000..a78cb22 --- /dev/null +++ b/src/services/server.ts @@ -0,0 +1,77 @@ +import http from 'http'; +import { create } from 'express-handlebars'; +import express, { Express, Request, Response } from 'express'; +import { createLogger } from '../utils/logger'; +import { getAllMeterRecords, deleteMeterByPublicKey } from '../store/sqlite'; +import { getUIComponents, invokeUIAction } from '../lib/utils'; + +const logger = createLogger('[server]'); +export const app: Express = express(); +const port = process.env.PORT || 3000; + +const hbs = create({ + defaultLayout: 'main', + extname: 'hbs', + helpers: { + encodeURIComponent: function (value: string) { + return encodeURIComponent(value); + }, + }, +}); + +app.engine('hbs', hbs.engine); +app.set('view engine', 'hbs'); +app.set('views', './src/views'); +app.use(express.json()); +app.use(express.static('./src/public')); +app.use(express.urlencoded({ extended: true })); + +app.get('/', async (req: Request, res: Response) => { + const m3ters = getAllMeterRecords(); + + // Get UI components from loaded UI extensions + const { icons, windows } = await getUIComponents(); + + logger.info('Rendering index page with m3ters, icons, and windows'); + + res.render('index', { m3ters, icons, windows }); + logger.info('Server handled GET request at `/`'); +}); + +// API endpoint to invoke UI actions +app.post('/api/actions/:moduleId/:actionId', async (req: Request, res: Response) => { + const { moduleId, actionId } = req.params; + logger.info(`Invoking action '${actionId}' from module '${moduleId}'`); + + const result = await invokeUIAction(moduleId, actionId); + + if (result.success) { + res.status(200).json(result); + } else { + res.status(400).json(result); + } +}); + +app.delete('/delete-meter', async (req: Request, res: Response) => { + let publicKey = decodeURIComponent(req.query?.publicKey?.toString() as string); + if (publicKey) { + const deleted = deleteMeterByPublicKey(publicKey.toString()); + if (!deleted) { + res.status(404).send({ message: 'meter not found' }); + return; + } + } else { + res.status(400).send({ message: 'public key not specified' }); + return; + } + logger.info('Server handled DELETE request at `/delete-meter`'); + res.status(200).send({ message: 'deleted meter' }); +}); + +export const server = http.createServer(app); + +export function startServer() { + server.listen(port, () => { + logger.info(`Server is running at port ${port}`); + }); +} diff --git a/src/services/ssh.ts b/src/services/ssh.ts new file mode 100644 index 0000000..235369b --- /dev/null +++ b/src/services/ssh.ts @@ -0,0 +1,95 @@ +import http from 'http'; +import WebSocket from 'ws'; +import { ConnectConfig, Client as SSHClient } from 'ssh2'; +import { server } from './server'; +import { createLogger } from '../utils/logger'; + +const logger = createLogger('[ssh]'); + +const wss = new WebSocket.Server({ server }); + +wss.on('connection', (ws: WebSocket, request: http.IncomingMessage) => { + logger.info('[ws]: New WebSocket connection established', { url: request.url }); + + const url = new URL(request.url!, 'http://localhost'); + const cols = url.searchParams.get('cols') ?? '80'; + const rows = url.searchParams.get('rows') ?? '24'; + const username = url.searchParams.get('username') ?? 'ubuntu'; + const password = url.searchParams.get('password') ?? 'Mauchly92618'; + + const ssh = new SSHClient(); + const sshConfig: ConnectConfig = { + host: '127.0.0.1', + port: 22, + username, + password, + }; + + ssh + .on('ready', () => { + logger.info('[ws/ssh]: SSH connection established'); + ssh.shell({ term: 'xterm', cols: parseInt(cols), rows: parseInt(rows) }, (err, stream) => { + if (err) { + ws.send(`[ws/ssh]: SSH shell error: ${err.message}`); + ws.close(); + ssh.end(); + return; + } + + // SSH -> WebSocket + stream.on('data', (data: Buffer) => { + if (ws.readyState === WebSocket.OPEN) { + ws.send(data); + } + }); + + stream.on('close', () => { + logger.info('[ws/ssh]: SSH stream closed'); + ws.close(); + ssh.end(); + }); + + stream.stderr.on('data', (data: Buffer) => { + if (ws.readyState === WebSocket.OPEN) { + ws.send(`[ws/ssh]: SSH stderr: ${data.toString()}`); + } + }); + + logger.info('[ws/ssh]: Setting up WebSocket message handler'); + ws.on('message', (msg) => { + // support JSON control messages for resize + try { + const parsed = JSON.parse(msg.toString()); + if (parsed.type === 'resize') { + const { cols, rows } = parsed; + stream.setWindow(rows, cols, cols * 8, rows * 16); // width/height px optional + return; + } + } catch (e) { + /* not JSON - treat as raw data */ + } + + if (stream.writable) stream.write(msg); + }); + + ws.on('close', () => { + logger.info('[ws]: WebSocket closed, closing SSH stream'); + stream.end(); + ssh.end(); + }); + + ws.on('error', () => { + logger.error('[ws]: WebSocket error, closing SSH stream'); + stream.end(); + ssh.end(); + }); + }); + }) + .on('error', (err) => { + logger.error('[ws/ssh]: SSH connection error:', err); + ws.send(`[ws/ssh]: SSH connection error: ${err.message}`); + ws.close(); + ssh.end(); + }) + .connect(sshConfig); +}); diff --git a/src/store/sqlite.ts b/src/store/sqlite.ts index 61763ac..4804fb1 100644 --- a/src/store/sqlite.ts +++ b/src/store/sqlite.ts @@ -2,6 +2,9 @@ import fs from "fs"; import Database from "better-sqlite3"; import type { Database as DatabaseType, Statement as DatabaseStatementType } from "better-sqlite3"; import { MeterRecord, TransactionRecord } from "../types"; +import { createLogger } from "../utils/logger"; + +const logger = createLogger("[sqlite]"); // meter queries let db: DatabaseType; @@ -37,7 +40,7 @@ export function deleteDatabase(databaseName = "m3tering.db") { db.close(); fs.unlinkSync(`./.data/${databaseName}`); } catch (err: any) { - console.error("Failed to delete database:", err); + logger.error("Failed to delete database:", err); } } @@ -132,7 +135,7 @@ export function saveMeter(meterData: MeterRecord): void { devEui: meterData.devEui ?? null, }); } catch (err: any) { - console.error("Failed to save meter:", err); + logger.error("Failed to save meter:", err); throw err; } } @@ -142,7 +145,7 @@ export function getMeterByPublicKey(publicKey: string): MeterRecord | null { const result = getMeterByPublicKeyQuery.get(publicKey) as MeterRecord | undefined; return result || null; } catch (err: any) { - console.error("Failed to get meter:", err); + logger.error("Failed to get meter:", err); return null; } } @@ -152,7 +155,7 @@ export function getMeterByDevEui(devEui: string): MeterRecord | null { const result = getMeterByDevEuiQuery.get(devEui) as MeterRecord | undefined; return result || null; } catch (err: any) { - console.error("Failed to get meter by DevEui:", err); + logger.error("Failed to get meter by DevEui:", err); return null; } } @@ -162,7 +165,7 @@ export function getMeterByTokenId(tokenId: number): MeterRecord | null { const result = getMeterByTokenIdQuery.get(tokenId) as MeterRecord | undefined; return result || null; } catch (err: any) { - console.error("Failed to get meter:", err); + logger.error("Failed to get meter:", err); return null; } } @@ -172,7 +175,7 @@ export function getAllMeterRecords(): MeterRecord[] { const results = getAllMetersQuery.all() as MeterRecord[]; return results; } catch (err: any) { - console.error("Failed to get all meters:", err); + logger.error("Failed to get all meters:", err); return []; } } @@ -182,11 +185,11 @@ export function deleteMeterByPublicKey(publicKey: string): boolean { const result = deleteMeterByPublicKeyQuery.run(publicKey); const deleted = result.changes > 0; if (!deleted) { - console.log("Meter not found for deletion:", { publicKey }); + logger.info("Meter not found for deletion:", { publicKey }); } return deleted; } catch (err: any) { - console.error("Failed to delete meter:", err); + logger.error("Failed to delete meter:", err); return false; } } @@ -196,11 +199,11 @@ export function updateMeterNonce(publicKey: string, nonce: number): boolean { const result = updateMeterNonceQuery.run(nonce, publicKey); const updated = result.changes > 0; if (!updated) { - console.log("Meter not found for nonce update:", { publicKey }); + logger.info("Meter not found for nonce update:", { publicKey }); } return updated; } catch (err: any) { - console.error("Failed to update meter nonce:", err); + logger.error("Failed to update meter nonce:", err); return false; } } @@ -210,11 +213,11 @@ export function updateMeterDevEui(publicKey: string, devEui: string): boolean { const result = updateMeterDevEuiQuery.run(devEui, publicKey); const updated = result.changes > 0; if (!updated) { - console.log("Meter not found for DevEui update:", { publicKey }); + logger.info("Meter not found for DevEui update:", { publicKey }); } return updated; } catch (err: any) { - console.error("Failed to update meter DevEui:", err); + logger.error("Failed to update meter DevEui:", err); return false; } } @@ -226,7 +229,7 @@ export function getTransactionByNonce(nonce: number, identifier: number): Transa | undefined; return result || null; } catch (err: any) { - console.error("Failed to get transaction by nonce:", err); + logger.error("Failed to get transaction by nonce:", err); return null; } } @@ -245,7 +248,7 @@ export function insertTransaction(transactionData: TransactionRecord): void { createTransactionQuery.run(transactionData); } catch (err: any) { - console.error("Failed to insert transaction:", err); + logger.error("Failed to insert transaction:", err); throw err; } } @@ -255,7 +258,7 @@ export function getAllTransactionRecords(): TransactionRecord[] { const results = db.prepare(`SELECT * FROM transactions`).all() as TransactionRecord[]; return results; } catch (err: any) { - console.error("Failed to get all transactions:", err); + logger.error("Failed to get all transactions:", err); return []; } } @@ -265,11 +268,11 @@ export function pruneTransactionsBefore(nonce: number, meterNumber: number) { const result = db .prepare(`DELETE FROM transactions WHERE identifier = ? AND nonce < ?`) .run(meterNumber, nonce); - console.log( + logger.info( `Pruned ${result.changes} transactions for meter ${meterNumber} with nonce < ${nonce}` ); } catch (err: any) { - console.error("Failed to prune transactions:", err); + logger.error("Failed to prune transactions:", err); } } @@ -278,10 +281,10 @@ export function pruneTransactionsAfter(nonce: number, meterNumber: number) { const result = db .prepare(`DELETE FROM transactions WHERE identifier = ? AND nonce > ?`) .run(meterNumber, nonce); - console.log( + logger.info( `Pruned ${result.changes} transactions for meter ${meterNumber} with nonce > ${nonce}` ); } catch (err: any) { - console.error("Failed to prune transactions:", err); + logger.error("Failed to prune transactions:", err); } } diff --git a/src/utils/contracts.ts b/src/utils/contracts.ts new file mode 100644 index 0000000..95a5a42 --- /dev/null +++ b/src/utils/contracts.ts @@ -0,0 +1,11 @@ +import { contractsReady, rollup as rollupContract } from '../services/contracts'; + +export async function meterIdFromPublicKeyContractCall(publicKey: string) { + await contractsReady(); + return Number(await rollupContract.m3terId(publicKey)); +} + +export async function meterLatestNonceContractCall(meterId: number) { + await contractsReady(); + return Number(await rollupContract.nonce(meterId)); +} diff --git a/src/utils/logger.ts b/src/utils/logger.ts index 90b0598..cd3f7be 100644 --- a/src/utils/logger.ts +++ b/src/utils/logger.ts @@ -1,22 +1,25 @@ -// Custom logger for meter-specific logging -export interface MeterLogger { - info: (message: string) => void; - warn: (message: string) => void; - error: (message: string) => void; - debug: (message: string) => void; +export interface Logger { + info: (message: string, ...args: any[]) => void; + warn: (message: string, ...args: any[]) => void; + error: (message: string, ...args: any[]) => void; + debug: (message: string, ...args: any[]) => void; } export interface MeterContext { devEui?: string; } -export function createMeterLogger(context: MeterContext): MeterLogger { - const prefix = context.devEui ? `[Device-${context.devEui}]` : "[Unknown-Meter]"; - +export function createLogger(prefix: string): Logger { return { - info: (message: string) => console.log(`${prefix} [info] ${message}`), - warn: (message: string) => console.log(`${prefix} [warn] ${message}`), - error: (message: string) => console.error(`${prefix} [error] ${message}`), - debug: (message: string) => console.log(`${prefix} [debug] ${message}`), + info: (message: string, ...args: any[]) => console.log(`[info] ${prefix} ${message}`, ...args), + warn: (message: string, ...args: any[]) => console.log(`[warn] ${prefix} ${message}`, ...args), + error: (message: string, ...args: any[]) => console.error(`[error] ${prefix} ${message}`, ...args), + debug: (message: string, ...args: any[]) => console.log(`[debug] ${prefix} ${message}`, ...args), }; } + +export function createMeterLogger(context: MeterContext): Logger { + const prefix = context.devEui ? `[Device-${context.devEui}]` : '[Unknown-Meter]'; + + return createLogger(prefix); +} diff --git a/tests/logic/sync.test.ts b/tests/logic/sync.test.ts index 78b6706..2e9d28c 100644 --- a/tests/logic/sync.test.ts +++ b/tests/logic/sync.test.ts @@ -6,8 +6,8 @@ import { getCrossChainRevenue, } from "../../src/lib/sync"; -// Mock the context module -jest.mock("../../src/services/context", () => ({ +// Mock the contracts module +jest.mock("../../src/services/contracts", () => ({ provider: { resolveName: jest.fn(), }, @@ -30,7 +30,7 @@ describe("Verifiers Cache", () => { }); it("should initialize cache successfully", async () => { - const { provider, ccipRevenueReader } = require("../../src/services/context"); + const { provider, ccipRevenueReader } = require("../../src/services/contracts"); // Mock successful responses ccipRevenueReader.verifierCount.mockResolvedValue(2n); @@ -55,7 +55,7 @@ describe("Verifiers Cache", () => { }); it("should throw error if ENS resolution fails (returns null)", async () => { - const { provider, ccipRevenueReader } = require("../../src/services/context"); + const { provider, ccipRevenueReader } = require("../../src/services/contracts"); // Mock responses with ENS resolution failure ccipRevenueReader.verifierCount.mockResolvedValue(1n); @@ -69,7 +69,7 @@ describe("Verifiers Cache", () => { }); it("should throw error if ENS resolution fails (returns zero address)", async () => { - const { provider, ccipRevenueReader } = require("../../src/services/context"); + const { provider, ccipRevenueReader } = require("../../src/services/contracts"); // Mock responses with ENS resolution failure ccipRevenueReader.verifierCount.mockResolvedValue(1n); @@ -83,7 +83,7 @@ describe("Verifiers Cache", () => { }); it("should use cached verifiers for getCrossChainRevenue", async () => { - const { provider, ccipRevenueReader } = require("../../src/services/context"); + const { provider, ccipRevenueReader } = require("../../src/services/contracts"); // Mock successful initialization ccipRevenueReader.verifierCount.mockResolvedValue(1n);