kjannette/koin_ping_0.2.0
1
0
Fork 0
Code Issues Pull Requests Actions 26 Packages Projects Releases Wiki Activity

move over

Browse Source
This commit is contained in:
KS Jannette
2026-02-27 16:41:18 -05:00
parent 5f81a4b1cc
commit 35d1c13d0a
54 changed files with 312 additions and 5626 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
backend/.gitignore vendored
View File

@@ -1,7 +0,0 @@
node_modules/
.env
*.log
.DS_Store
dist/
build/

51
backend/ENV_TEMPLATE.md
View File

@@ -1,51 +0,0 @@
# Environment Variables Template
Copy this to `.env` in the backend directory and fill in your values.
```bash
# Server Configuration
PORT=3001
NODE_ENV=development
# Database Configuration
DATABASE_URL=postgresql://user:password@localhost:5432/koin_ping
# Ethereum JSON-RPC Endpoint
# Examples:
# - Infura: https://mainnet.infura.io/v3/YOUR-PROJECT-ID
# - Alchemy: https://eth-mainnet.g.alchemy.com/v2/YOUR-API-KEY
# - Local node: http://localhost:8545
ETH_RPC_URL=https://mainnet.infura.io/v3/YOUR-PROJECT-ID
# Polling interval in milliseconds
# Default: 60000 (1 minute)
POLL_INTERVAL_MS=60000
# Firebase Configuration (for authentication)
# Get this from Firebase Console > Project Settings > Project ID
FIREBASE_PROJECT_ID=koin-ping
```
## Quick Setup
```bash
cd backend
cp ENV_TEMPLATE.md .env
# Edit .env with your actual values
```
## Firebase Setup
For Firebase Admin SDK to work, you need to set up Application Default Credentials:
**Option 1: Use Firebase Project ID (easiest for development)**
- Just set `FIREBASE_PROJECT_ID` in .env
- Firebase Admin will use Application Default Credentials
**Option 2: Use Service Account Key (production)**
1. Go to Firebase Console > Project Settings > Service Accounts
2. Click "Generate new private key"
3. Download the JSON file
4. Either:
- Set `GOOGLE_APPLICATION_CREDENTIALS=/path/to/serviceAccountKey.json` in .env
- Or keep it in backend/ and add to .gitignore

56
backend/README.md
View File

@@ -1,56 +0,0 @@
# Koin Ping Backend
On-chain monitoring and alerting system backend.
## Environment Variables
Create a `.env` file in the backend directory with the following variables:
```env
PORT=3001
NODE_ENV=development
# PostgreSQL Database Configuration
DB_HOST=localhost
DB_PORT=5432
DB_USER=your_db_user
DB_PASSWORD=your_db_password
DB_NAME=koin_ping
# Blockchain RPC Endpoints (examples)
ETH_RPC_URL=https://mainnet.infura.io/v3/YOUR_INFURA_KEY
POLYGON_RPC_URL=https://polygon-rpc.com
# Polling Configuration
POLL_INTERVAL_MS=10000
```
## Getting Started
1. Install dependencies:
```bash
npm install
```
2. Set up your `.env` file with the required environment variables
3. Run the development server:
```bash
npm run dev
```
4. Run in production:
```bash
npm start
```
## Project Structure
- `api/` - API endpoints and server configuration
- `poller/` - Blockchain polling logic
- `alerts/` - Alert evaluation and management
- `notifications/` - Notification delivery system
- `domain/` - Domain models and business logic
- `infra/` - Infrastructure (database, external services)
- `shared/` - Shared utilities and helpers

20
backend/api/routes/addresses.js
View File

@@ -1,20 +0,0 @@
import express from 'express';
import { authenticate } from '../../middleware/authenticate.js';
import * as AddressController from '../../controllers/AddressController.js';
const router = express.Router();
// Apply authentication to all routes
router.use(authenticate);
// POST /addresses - Create Blockchain Address
router.post('/', AddressController.create);
// GET /addresses - List Addresses
router.get('/', AddressController.list);
// DELETE /addresses/:addressId - Delete Address
router.delete('/:addressId', AddressController.remove);
export default router;

12
backend/api/routes/alertEvents.js
View File

@@ -1,12 +0,0 @@
import express from 'express';
import { authenticate } from '../../middleware/authenticate.js';
import * as AlertEventController from '../../controllers/AlertEventController.js';
const router = express.Router();
router.use(authenticate);
router.get('/', AlertEventController.list);
export default router;

18
backend/api/routes/alerts.js
View File

@@ -1,18 +0,0 @@
import express from 'express';
import { authenticate } from '../../middleware/authenticate.js';
import * as AlertRuleController from '../../controllers/AlertRuleController.js';
const router = express.Router();
router.use(authenticate);
router.post('/:addressId/alerts', AlertRuleController.create);
router.get('/:addressId/alerts', AlertRuleController.listByAddress);
router.patch('/:alertId', AlertRuleController.updateStatus);
router.delete('/:alertId', AlertRuleController.remove);
export default router;

16
backend/api/routes/notificationConfig.js
View File

@@ -1,16 +0,0 @@
import express from 'express';
import { authenticate } from '../../middleware/authenticate.js';
import * as NotificationConfigController from '../../controllers/NotificationConfigController.js';
const router = express.Router();
router.use(authenticate);
router.get('/', NotificationConfigController.getConfig);
router.put('/', NotificationConfigController.updateConfig);
router.delete('/', NotificationConfigController.deleteConfig);
export default router;

17
backend/api/routes/status.js
View File

@@ -1,17 +0,0 @@
import express from 'express';
const router = express.Router();
// GET /status - System Status (mock)
router.get('/', (req, res) => {
// TEMP - Mock response until poller service is complete
res.json({
latestBlock: 0,
lag: 0,
status: 'healthy',
timestamp: new Date().toISOString()
});
});
export default router;

47
backend/api/server.js
View File

@@ -1,47 +0,0 @@
import express from 'express';
import cors from 'cors';
import dotenv from 'dotenv';
import addressesRouter from './routes/addresses.js';
import alertsRouter from './routes/alerts.js';
import alertEventsRouter from './routes/alertEvents.js';
import statusRouter from './routes/status.js';
import notificationConfigRouter from './routes/notificationConfig.js';
dotenv.config();
const app = express();
const PORT = process.env.PORT || 3001;
app.use(cors());
app.use(express.json());
// Health check
app.get('/api/health', (req, res) => {
res.json({
status: 'ok',
timestamp: new Date().toISOString(),
service: 'koin-ping-backend'
});
});
app.use('/api/addresses', addressesRouter);
app.use('/api/addresses', alertsRouter); // Alert routes include addressId in path
app.use('/api/alerts', alertsRouter); // Direct alert operations (PATCH, DELETE)
app.use('/api/alert-events', alertEventsRouter);
app.use('/api/status', statusRouter);
app.use('/api/notification-config', notificationConfigRouter);
app.use((err, req, res, next) => {
console.error(err.stack);
res.status(500).json({
error: 'Something went wrong!',
message: process.env.NODE_ENV === 'development' ? err.message : undefined
});
});
app.listen(PORT, () => {
console.log(`Server running on port ${PORT}`);
console.log(`Environment: ${process.env.NODE_ENV || 'development'}`);
});

82
backend/controllers/AddressController.js
View File

@@ -1,82 +0,0 @@
import * as AddressModel from '../models/AddressModel.js';
/**
* Create a new address
*/
export async function create(req, res) {
const { address, label } = req.body;
const user_id = req.user_id; // From authenticate middleware
console.log(`User ${user_id} creating address: ${address}`);
if (!address) {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'Address is required'
});
}
// Validation: ETH address format
if (!/^0x[a-fA-F0-9]{40}$/.test(address)) {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'Invalid Ethereum address format'
});
}
try {
const newAddress = await AddressModel.create(user_id, address, label || null);
console.log(`Address created with ID: ${newAddress.id}`);
return res.status(201).json(newAddress);
} catch (error) {
// Handle duplicate address (unique constraint violation)
if (error.code === '23505') {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'You are already tracking this address'
});
}
throw error;
}
}
export async function list(req, res) {
const user_id = req.user_id; // From authenticate middleware
console.log(`User ${user_id} listing addresses`);
const addresses = await AddressModel.listByUser(user_id);
console.log(`Found ${addresses.length} addresses for user`);
return res.json(addresses);
}
export async function remove(req, res) {
const addressId = parseInt(req.params.addressId);
const user_id = req.user_id; // From authenticate middleware
console.log(`User ${user_id} deleting address ID: ${addressId}`);
if (isNaN(addressId)) {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'Invalid address ID'
});
}
const deleted = await AddressModel.remove(addressId, user_id);
if (!deleted) {
console.log(`Address ${addressId} not found or not owned by user`);
return res.status(404).json({
error: 'NOT_FOUND',
message: 'Address not found'
});
}
console.log(`Address ${addressId} deleted`);
return res.status(204).send();
}

54
backend/controllers/AlertEventController.js
View File

@@ -1,54 +0,0 @@
import * as AlertEventModel from '../models/AlertEventModel.js';
// Mock data for when DB is empty (MVP scaffolding)
const MOCK_EVENTS = [
{
id: 1,
alert_rule_id: 1,
message: 'Incoming transaction detected: 5.5 ETH received',
address_label: 'Treasury Wallet',
timestamp: new Date(Date.now() - 2 * 60 * 60 * 1000).toISOString() // 2 hours ago
},
{
id: 2,
alert_rule_id: 2,
message: 'Balance dropped below threshold: Current balance 8.2 ETH',
address_label: 'Treasury Wallet',
timestamp: new Date(Date.now() - 5 * 60 * 60 * 1000).toISOString() // 5 hours ago
},
{
id: 3,
alert_rule_id: 3,
message: 'Outgoing transaction detected: 2.0 ETH sent',
address_label: 'Cold Storage',
timestamp: new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString() // 1 day ago
}
];
export async function list(req, res) {
const limit = parseInt(req.query.limit) || 20;
const user_id = req.user_id; // From authenticate middleware
console.log(`User ${user_id} listing alert events (limit: ${limit})`);
if (limit < 1 || limit > 100) {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'Limit must be between 1 and 100'
});
}
const events = await AlertEventModel.listRecentByUser(user_id, limit);
console.log(`Found ${events.length} alert events for user`);
// Return mock data if DB is empty (MVP scaffolding)
if (events.length === 0) {
return res.json(MOCK_EVENTS.slice(0, limit));
}
return res.json(events);
}

185
backend/controllers/AlertRuleController.js
View File

@@ -1,185 +0,0 @@
import * as AlertRuleModel from '../models/AlertRuleModel.js';
import * as AddressModel from '../models/AddressModel.js';
const VALID_ALERT_TYPES = ['incoming_tx', 'outgoing_tx', 'large_transfer', 'balance_below'];
const THRESHOLD_REQUIRED_TYPES = ['large_transfer', 'balance_below'];
/**
* Create a new alert rule
*/
export async function create(req, res) {
const addressId = parseInt(req.params.addressId);
const { type, threshold } = req.body;
const user_id = req.user_id; // From authenticate middleware
console.log(`User ${user_id} creating alert for address ID: ${addressId}`);
if (isNaN(addressId)) {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'Invalid address ID'
});
}
if (!type) {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'Alert type is required'
});
}
// Validation: type must be valid
if (!VALID_ALERT_TYPES.includes(type)) {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: `Invalid alert type. Must be one of: ${VALID_ALERT_TYPES.join(', ')}`
});
}
if (THRESHOLD_REQUIRED_TYPES.includes(type)) {
if (!threshold || threshold === '') {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: `Alert type '${type}' requires a threshold value`
});
}
const thresholdNum = parseFloat(threshold);
if (isNaN(thresholdNum) || thresholdNum <= 0) {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'Threshold must be a positive number'
});
}
}
try {
// Verify address exists AND belongs to this user
const address = await AddressModel.findById(addressId, user_id);
if (!address) {
console.log(`Address ${addressId} not found or not owned by user`);
return res.status(404).json({
error: 'NOT_FOUND',
message: 'Address not found'
});
}
const newAlert = await AlertRuleModel.create(
addressId,
type,
threshold || null
);
console.log(`Alert rule created with ID: ${newAlert.id}`);
return res.status(201).json(newAlert);
} catch (error) {
console.error(error);
return res.status(500).json({
error: 'INTERNAL_ERROR',
message: 'Failed to create alert rule'
});
}
}
export async function listByAddress(req, res) {
const addressId = parseInt(req.params.addressId);
const user_id = req.user_id; // From authenticate middleware
console.log(`User ${user_id} listing alerts for address ID: ${addressId}`);
if (isNaN(addressId)) {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'Invalid address ID'
});
}
const address = await AddressModel.findById(addressId, user_id);
if (!address) {
console.log(`Address ${addressId} not found or not owned by user`);
return res.status(404).json({
error: 'NOT_FOUND',
message: 'Address not found'
});
}
const alerts = await AlertRuleModel.listByAddress(addressId);
console.log(`Found ${alerts.length} alert rules`);
return res.json(alerts);
}
export async function updateStatus(req, res) {
const alertId = parseInt(req.params.alertId);
const { enabled } = req.body;
const user_id = req.user_id;
console.log(`User ${user_id} updating alert ID: ${alertId}`);
// Validation: alertId must be valid
if (isNaN(alertId)) {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'Invalid alert ID'
});
}
// Validation: enabled must be boolean
if (typeof enabled !== 'boolean') {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'enabled must be a boolean value'
});
}
// Verify user owns this alert (through address ownership)
const alert = await AlertRuleModel.findById(alertId, user_id);
if (!alert) {
console.log(`Alert ${alertId} not found or not owned by user`);
return res.status(404).json({
error: 'NOT_FOUND',
message: 'Alert rule not found'
});
}
const updatedAlert = await AlertRuleModel.updateEnabled(alertId, enabled);
console.log(`Alert ${alertId} updated: enabled=${enabled}`);
return res.json(updatedAlert);
}
export async function remove(req, res) {
const alertId = parseInt(req.params.alertId);
const user_id = req.user_id;
console.log(`User ${user_id} deleting alert ID: ${alertId}`);
if (isNaN(alertId)) {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'Invalid alert ID'
});
}
const alert = await AlertRuleModel.findById(alertId, user_id);
if (!alert) {
console.log(`Alert ${alertId} not found or not owned by user`);
return res.status(404).json({
error: 'NOT_FOUND',
message: 'Alert rule not found'
});
}
const deleted = await AlertRuleModel.remove(alertId);
console.log(`Alert ${alertId} deleted`);
return res.status(204).send();
}

105
backend/controllers/NotificationConfigController.js
View File

@@ -1,105 +0,0 @@
import * as NotificationConfigModel from '../models/NotificationConfigModel.js';
/**
* Get notification config for the authenticated user
*/
export async function getConfig(req, res) {
const user_id = req.user_id; // From authenticate middleware
console.log(`User ${user_id} getting notification config`);
const config = await NotificationConfigModel.getConfig(user_id);
if (!config) {
// Return empty config if not set
return res.json({
user_id,
discord_webhook_url: null,
telegram_chat_id: null,
email: null,
notification_enabled: true
});
}
console.log(`Config found`);
return res.json(config);
}
/**
* Update notification config for the authenticated user
*/
export async function updateConfig(req, res) {
const user_id = req.user_id; // From authenticate middleware
const { discord_webhook_url, telegram_chat_id, email, notification_enabled } = req.body;
console.log(`User ${user_id} updating notification config`);
// Validation: at least one field should be provided
if (
discord_webhook_url === undefined &&
telegram_chat_id === undefined &&
email === undefined &&
notification_enabled === undefined
) {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'At least one configuration field must be provided'
});
}
// Validation: Discord webhook URL format (basic check)
if (discord_webhook_url && !discord_webhook_url.startsWith('https://discord.com/api/webhooks/')) {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'Invalid Discord webhook URL format'
});
}
// Validation: Email format (basic check)
if (email && !/^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(email)) {
return res.status(400).json({
error: 'VALIDATION_ERROR',
message: 'Invalid email address format'
});
}
try {
const updatedConfig = await NotificationConfigModel.upsertConfig(user_id, {
discord_webhook_url,
telegram_chat_id,
email,
notification_enabled
});
console.log(`Notification config updated`);
return res.json(updatedConfig);
} catch (error) {
console.error('Error updating notification config:', error);
return res.status(500).json({
error: 'INTERNAL_ERROR',
message: 'Failed to update notification configuration'
});
}
}
/**
* Delete notification config for the authenticated user
*/
export async function deleteConfig(req, res) {
const user_id = req.user_id; // From authenticate middleware
console.log(`User ${user_id} deleting notification config`);
const deleted = await NotificationConfigModel.remove(user_id);
if (!deleted) {
return res.status(404).json({
error: 'NOT_FOUND',
message: 'No notification configuration found'
});
}
console.log(`Notification config deleted`);
return res.status(204).send();
}

19
backend/domain/NormalizedTx.js
View File

@@ -1,19 +0,0 @@
/**
* NormalizedTx - Standardized transaction representation
*
* Defines the minimal, normalized transaction data relevant
* for tracking blockchain activity.
*/
/**
* @typedef {Object} NormalizedTx
* @property {string} hash - Transaction hash (0x prefixed)
* @property {string} from - Sender address (lowercase, 0x prefixed)
* @property {string|null} to - Recipient address (lowercase, 0x prefixed, null for contract creation)
* @property {string} value - Value transferred in Wei (as string to preserve precision)
* @property {number} blockNumber - Block number where transaction was included
* @property {number} blockTimestamp - Unix timestamp of the block
*/
export default {};

17
backend/domain/ObservedTx.js
View File

@@ -1,17 +0,0 @@
/**
* @typedef {import('./NormalizedTx.js').NormalizedTx} NormalizedTx
*/
/**
* @typedef {Object} ObservedTxMetadata
* @property {number} addressId - ID of the tracked address (from addresses table)
* @property {'incoming'|'outgoing'} direction - Direction relative to tracked address
*/
/**
* @typedef {NormalizedTx & ObservedTxMetadata} ObservedTx
*/
export default {};

20
backend/firebase/admin.js
View File

@@ -1,20 +0,0 @@
import admin from 'firebase-admin';
import dotenv from 'dotenv';
dotenv.config();
if (!admin.apps.length) {
admin.initializeApp({
projectId: process.env.FIREBASE_PROJECT_ID,
// For local development, you can use application default credentials
// or provide a service account key file
});
}
export const auth = admin.auth();
export default admin;

94
backend/infra/README.md
View File

@@ -1,94 +0,0 @@
# Database Setup
## Prerequisites
- PostgreSQL installed locally
- Database created (e.g., `koin_ping_dev`)
## Setup Steps
### 1. Create Database
```bash
# Using psql
createdb koin_ping_dev
# Or via psql command line
psql -U postgres
CREATE DATABASE koin_ping_dev;
\q
```
### 2. Set Environment Variables
Create a `.env` file in the `backend/` directory:
```env
DATABASE_URL=postgresql://your_user:your_password@localhost:5432/koin_ping_dev
```
Or if using individual variables (for `infra/database.js`):
```env
DB_HOST=localhost
DB_PORT=5432
DB_USER=your_user
DB_PASSWORD=your_password
DB_NAME=koin_ping_dev
```
### 3. Run Schema
```bash
# From the backend
psql -d koin_ping_dev -f db/schema.sql
# Or the full connection string
psql postgresql://your_user:your_password@localhost:5432/koin_ping_dev -f db/schema.sql
```
### 4. Verify Tables
```bash
psql -d koin_ping_dev
# In psql:
\dt # List tables
\d addresses # Describe addresses table
\d alert_rules # Describe alert_rules table
\d alert_events # Describe alert_events table
```
## Tables
### addresses
- `id` - Primary key
- `address` - Ethereum address (unique)
- `label` - Optional display name
- `created_at` - Timestamp
### alert_rules
- `id` - Primary key
- `address_id` - Foreign key to addresses
- `type` - Alert type: `incoming_tx`, `outgoing_tx`, `large_transfer`, `balance_below`
- `threshold` - ETH amount (nullable)
- `enabled` - Active status
- `created_at` - Timestamp
### alert_events
- `id` - Primary key
- `alert_rule_id` - Foreign key to alert_rules
- `message` - Alert description
- `address_label` - Denormalized label for display
- `timestamp` - When alert fired
## Reset Database
To start fresh:
```bash
psql -d koin_ping_dev -f db/schema.sql
```
The schema includes `DROP TABLE IF EXISTS` statements for clean reruns.

21
backend/infra/add_notification_config.sql
View File

@@ -1,21 +0,0 @@
-- Run: psql -d koin_ping_dev -f add_notification_config.sql
-- Create user notification configuration tableStores notification preferences and webhook URLs for each user
CREATE TABLE user_notification_configs (
user_id VARCHAR(128) PRIMARY KEY,
discord_webhook_url TEXT, -- Discord webhook URL (nullable)
telegram_chat_id VARCHAR(128), -- Telegram chat ID (nullable)
telegram_bot_token VARCHAR(255), -- Telegram bot token (nullable, future use)
email VARCHAR(255), -- Email for notifications (nullable)
notification_enabled BOOLEAN DEFAULT TRUE, -- Master on/off switch
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
-- Create index for active notifications lookup
CREATE INDEX idx_notification_configs_enabled ON user_notification_configs(notification_enabled);
-- Note: No foreign key to a users table (doesn't exist yet)
-- user_id is the Firebase UID, validated by backend middleware

16
backend/infra/add_user_id_migration.sql
View File

@@ -1,16 +0,0 @@
ALTER TABLE addresses
ADD COLUMN user_id VARCHAR(128);
UPDATE addresses
SET user_id = 'legacy_user'
WHERE user_id IS NULL;
ALTER TABLE addresses
ALTER COLUMN user_id SET NOT NULL;
CREATE INDEX idx_addresses_user_id ON addresses(user_id);
ALTER TABLE addresses DROP CONSTRAINT IF EXISTS addresses_address_key;
ALTER TABLE addresses ADD CONSTRAINT addresses_user_address_unique UNIQUE (user_id, address);

32
backend/infra/database.js
View File

@@ -1,32 +0,0 @@
import pg from "pg";
import dotenv from "dotenv";
dotenv.config();
const { Pool } = pg;
const pool = new Pool(
process.env.DATABASE_URL
? { connectionString: process.env.DATABASE_URL }
: {
host: process.env.DB_HOST || "localhost",
port: process.env.DB_PORT || 5432,
user: process.env.DB_USER,
password: process.env.DB_PASSWORD,
database: process.env.DB_NAME,
max: 20,
idleTimeoutMillis: 30000,
connectionTimeoutMillis: 2000,
}
);
pool.on("connect", () => {
console.log("Connected to PostgreSQL database");
});
pool.on("error", (err) => {
console.error("Unexpected error on idle client", err);
process.exit(1);
});
export default pool;

71
backend/infra/schema.sql
View File

@@ -1,71 +0,0 @@
DROP TABLE IF EXISTS alert_events CASCADE;
DROP TABLE IF EXISTS alert_rules CASCADE;
DROP TABLE IF EXISTS address_checkpoints CASCADE;
DROP TABLE IF EXISTS user_notification_configs CASCADE;
DROP TABLE IF EXISTS addresses CASCADE;
CREATE TABLE addresses (
id SERIAL PRIMARY KEY,
user_id VARCHAR(128) NOT NULL, -- Firebase user ID (multi-user support)
address VARCHAR(42) NOT NULL, -- Ethereum address format (0x + 40 hex chars)
label VARCHAR(255), -- Optional human-readable label
created_at TIMESTAMP DEFAULT NOW(),
-- Unique users can track the same address independently
UNIQUE(user_id, address)
);
CREATE TABLE alert_rules (
id SERIAL PRIMARY KEY,
address_id INTEGER NOT NULL REFERENCES addresses(id) ON DELETE CASCADE,
type VARCHAR(50) NOT NULL, -- 'incoming_tx', 'outgoing_tx', 'large_transfer', 'balance_below'
threshold DECIMAL(20, 6), -- ETH amount threshold (nullable for tx types that don't need it)
enabled BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT NOW(),
CONSTRAINT valid_alert_type CHECK (
type IN ('incoming_tx', 'outgoing_tx', 'large_transfer', 'balance_below')
),
CONSTRAINT positive_threshold CHECK (
threshold IS NULL OR threshold > 0
)
);
CREATE TABLE alert_events (
id SERIAL PRIMARY KEY,
alert_rule_id INTEGER NOT NULL REFERENCES alert_rules(id) ON DELETE CASCADE,
message TEXT NOT NULL, -- Human-readable alert message
address_label VARCHAR(255), -- Denormalized for display (avoids joins)
tx_hash VARCHAR(66), -- Transaction hash that triggered alert (nullable for balance_below)
timestamp TIMESTAMP DEFAULT NOW()
);
CREATE TABLE address_checkpoints (
address_id INTEGER PRIMARY KEY REFERENCES addresses(id) ON DELETE CASCADE,
last_checked_block INTEGER NOT NULL, -- Last block number that was checked for this address
last_checked_at TIMESTAMP DEFAULT NOW()
);
CREATE TABLE user_notification_configs (
user_id VARCHAR(128) PRIMARY KEY,
discord_webhook_url TEXT, -- Discord webhook URL (nullable)
telegram_chat_id VARCHAR(128), -- Telegram chat ID (nullable)
telegram_bot_token VARCHAR(255), -- Telegram bot token (nullable, future use)
email VARCHAR(255), -- Email for notifications (nullable)
notification_enabled BOOLEAN DEFAULT TRUE, -- Master on/off switch
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
-- Create indexes for common queries
CREATE INDEX idx_addresses_user_id ON addresses(user_id);
CREATE INDEX idx_alert_rules_address_id ON alert_rules(address_id);
CREATE INDEX idx_alert_rules_enabled ON alert_rules(enabled);
CREATE INDEX idx_alert_events_timestamp ON alert_events(timestamp DESC);
CREATE INDEX idx_alert_events_alert_rule_id ON alert_events(alert_rule_id);
CREATE INDEX idx_address_checkpoints_last_checked_at ON address_checkpoints(last_checked_at);
CREATE INDEX idx_notification_configs_enabled ON user_notification_configs(notification_enabled);

68
backend/middleware/authenticate.js
View File

@@ -1,68 +0,0 @@
import { auth } from '../firebase/admin.js';
export async function authenticate(req, res, next) {
try {
// Get Authorization header
const authHeader = req.headers.authorization;
if (!authHeader || !authHeader.startsWith('Bearer ')) {
console.log('No Authorization header or invalid format');
return res.status(401).json({
error: 'UNAUTHORIZED',
message: 'No authentication token provided'
});
}
const token = authHeader.split('Bearer ')[1];
if (!token) {
console.log('Empty token');
return res.status(401).json({
error: 'UNAUTHORIZED',
message: 'Invalid token format'
});
}
console.log('Verifying Firebase token...');
const decodedToken = await auth.verifyIdToken(token);
const user_id = decodedToken.uid;
// SMOKE TEST: Console log the user_id
console.log('Token verified!');
console.log(' User ID:', user_id);
console.log(' Email:', decodedToken.email);
console.log(' Token issued at:', new Date(decodedToken.iat * 1000).toISOString());
req.user_id = user_id;
req.user_email = decodedToken.email;
next();
} catch (error) {
console.error('Token verification failed:', error.message);
// Handle specific Firebase errors
if (error.code === 'auth/id-token-expired') {
return res.status(401).json({
error: 'TOKEN_EXPIRED',
message: 'Authentication token has expired'
});
}
if (error.code === 'auth/argument-error') {
return res.status(401).json({
error: 'INVALID_TOKEN',
message: 'Invalid authentication token format'
});
}
return res.status(401).json({
error: 'UNAUTHORIZED',
message: 'Failed to verify authentication token'
});
}
}

76
backend/models/AddressCheckpointModel.js
View File

@@ -1,76 +0,0 @@
import pool from '../infra/database.js';
/**
* Get the last checked block number for an address
* @param {number} addressId - Address ID from the addresses table
* @returns {Promise<number|null>} Last checked block number, or null if never checked
*/
export async function getLastCheckedBlock(addressId) {
const result = await pool.query(
`SELECT last_checked_block
FROM address_checkpoints
WHERE address_id = $1`,
[addressId]
);
return result.rows[0]?.last_checked_block || null;
}
/**
* Update the last checked block number for an address
* Creates a new checkpoint record if one doesn't exist
* @param {number} addressId - Address ID from the addresses table
* @param {number} blockNumber - Block number that was just checked
* @returns {Promise<Object>} Updated checkpoint record
*/
export async function updateLastCheckedBlock(addressId, blockNumber) {
const result = await pool.query(
`INSERT INTO address_checkpoints (address_id, last_checked_block, last_checked_at)
VALUES ($1, $2, NOW())
ON CONFLICT (address_id)
DO UPDATE SET
last_checked_block = $2,
last_checked_at = NOW()
RETURNING address_id, last_checked_block, last_checked_at`,
[addressId, blockNumber]
);
return result.rows[0];
}
/**
* Get all address checkpoints (useful for debugging/monitoring)
* @returns {Promise<Array>} Array of checkpoint objects
*/
export async function listAll() {
const result = await pool.query(
`SELECT
ac.address_id,
a.address,
a.label,
ac.last_checked_block,
ac.last_checked_at
FROM address_checkpoints ac
JOIN addresses a ON a.id = ac.address_id
ORDER BY ac.last_checked_at DESC`
);
return result.rows;
}
/**
* Delete checkpoint for an address (useful when address is removed)
* Note: This should happen automatically via CASCADE when address is deleted
* @param {number} addressId - Address ID
* @returns {Promise<boolean>} True if deleted, false if not found
*/
export async function remove(addressId) {
const result = await pool.query(
`DELETE FROM address_checkpoints WHERE address_id = $1`,
[addressId]
);
return result.rowCount > 0;
}

84
backend/models/AddressModel.js
View File

@@ -1,84 +0,0 @@
import pool from '../infra/database.js';
/**
* Create a new address
* @param {string} user_id - Firebase user ID
* @param {string} address - Ethereum address
* @param {string|null} label - Optional label
* @returns {Promise<Object>} Created address object
*/
export async function create(user_id, address, label) {
const result = await pool.query(
`INSERT INTO addresses (user_id, address, label)
VALUES ($1, $2, $3)
RETURNING id, user_id, address, label, created_at`,
[user_id, address, label]
);
return result.rows[0];
}
/**
* List all addresses for a specific user
* @param {string} user_id - Firebase user ID
* @returns {Promise<Array>} Array of address objects
*/
export async function listByUser(user_id) {
const result = await pool.query(
`SELECT id, address, label, created_at
FROM addresses
WHERE user_id = $1
ORDER BY created_at DESC`,
[user_id]
);
return result.rows;
}
/**
* List all addresses (for system use only, e.g., poller)
* @returns {Promise<Array>} Array of address objects
*/
export async function list() {
const result = await pool.query(
`SELECT id, user_id, address, label, created_at
FROM addresses
ORDER BY created_at DESC`
);
return result.rows;
}
/**
* Find address by ID (for specific user)
* @param {number} id - Address ID
* @param {string} user_id - Firebase user ID (for ownership verification)
* @returns {Promise<Object|null>} Address object or null if not found or not owned by user
*/
export async function findById(id, user_id = null) {
const query = user_id
? `SELECT id, user_id, address, label, created_at
FROM addresses
WHERE id = $1 AND user_id = $2`
: `SELECT id, user_id, address, label, created_at
FROM addresses
WHERE id = $1`;
const params = user_id ? [id, user_id] : [id];
const result = await pool.query(query, params);
return result.rows[0] || null;
}
/**
* Remove address by ID (for specific user)
* @param {number} id - Address ID
* @param {string} user_id - Firebase user ID (for ownership verification)
* @returns {Promise<boolean>} True if deleted, false if not found or not owned by user
*/
export async function remove(id, user_id) {
const result = await pool.query(
`DELETE FROM addresses
WHERE id = $1 AND user_id = $2`,
[id, user_id]
);
return result.rowCount > 0;
}

56
backend/models/AlertEventModel.js
View File

@@ -1,56 +0,0 @@
import pool from '../infra/database.js';
/**
* List recent alert events for a specific user
* @param {string} user_id - Firebase user ID
* @param {number} limit - Maximum number of events to return
* @returns {Promise<Array>} Array of alert event objects
*/
export async function listRecentByUser(user_id, limit = 20) {
const result = await pool.query(
`SELECT ae.id, ae.alert_rule_id, ae.message, ae.address_label, ae.tx_hash, ae.timestamp
FROM alert_events ae
JOIN alert_rules ar ON ar.id = ae.alert_rule_id
JOIN addresses a ON a.id = ar.address_id
WHERE a.user_id = $1
ORDER BY ae.timestamp DESC
LIMIT $2`,
[user_id, limit]
);
return result.rows;
}
/**
* List recent alert events
* @param {number} limit - Maximum number of event
* @returns {Promise<Array>} Array of alert event objects
*/
export async function listRecent(limit = 20) {
const result = await pool.query(
`SELECT id, alert_rule_id, message, address_label, tx_hash, timestamp
FROM alert_events
ORDER BY timestamp DESC
LIMIT $1`,
[limit]
);
return result.rows;
}
/**
* Create a new alert event
* @param {number} alertRuleId - Alert rule ID
* @param {string} message - Alert message
* @param {string|null} addressLabel - Address label (denormalized)
* @param {string|null} txHash - Transaction hash (optional, null for non-tx alerts)
* @returns {Promise<Object>} Created alert event object
*/
export async function create(alertRuleId, message, addressLabel, txHash = null) {
const result = await pool.query(
`INSERT INTO alert_events (alert_rule_id, message, address_label, tx_hash)
VALUES ($1, $2, $3, $4)
RETURNING id, alert_rule_id, message, address_label, tx_hash, timestamp`,
[alertRuleId, message, addressLabel, txHash]
);
return result.rows[0];
}

92
backend/models/AlertRuleModel.js
View File

@@ -1,92 +0,0 @@
import pool from '../infra/database.js';
/**
* Create new alertrule
* @param {number} addressId - Address ID
* @param {string} type - Alert type
* @param {string|null} threshold - Threshold value (nullable)
* @returns {Promise<Object>} Created alert rule object
*/
export async function create(addressId, type, threshold) {
const result = await pool.query(
`INSERT INTO alert_rules (address_id, type, threshold, enabled)
VALUES ($1, $2, $3, TRUE)
RETURNING id, address_id, type, threshold, enabled, created_at`,
[addressId, type, threshold]
);
return result.rows[0];
}
/**
* List alert rules by address ID
* @param {number} addressId - Address ID
* @returns {Promise<Array>} Array of alert rule objects
*/
export async function listByAddress(addressId) {
const result = await pool.query(
`SELECT id, address_id, type, threshold, enabled, created_at
FROM alert_rules
WHERE address_id = $1
ORDER BY created_at DESC`,
[addressId]
);
return result.rows;
}
/**
* Find alert rule by ID
* @param {number} id - Alert rule ID
* @param {string} user_id - Optional user ID for ownership verification
* @returns {Promise<Object|null>} Alert rule object or null if not found/not owned
*/
export async function findById(id, user_id = null) {
let query, params;
if (user_id) {
// Verify ownership
query = `SELECT ar.id, ar.address_id, ar.type, ar.threshold, ar.enabled, ar.created_at
FROM alert_rules ar
JOIN addresses a ON a.id = ar.address_id
WHERE ar.id = $1 AND a.user_id = $2`;
params = [id, user_id];
} else {
query = `SELECT id, address_id, type, threshold, enabled, created_at
FROM alert_rules
WHERE id = $1`;
params = [id];
}
const result = await pool.query(query, params);
return result.rows[0] || null;
}
/**
* Update alert rule enabled status
* @param {number} id - Alert rule ID
* @param {boolean} enabled - Enabled status
* @returns {Promise<Object|null>} Updated alert rule or null if not found
*/
export async function updateEnabled(id, enabled) {
const result = await pool.query(
`UPDATE alert_rules
SET enabled = $2
WHERE id = $1
RETURNING id, address_id, type, threshold, enabled, created_at`,
[id, enabled]
);
return result.rows[0] || null;
}
/**
* Remove alert rule by ID
* @param {number} id - Alert rule ID
* @returns {Promise<boolean>} True if deleted, false if not found
*/
export async function remove(id) {
const result = await pool.query(
`DELETE FROM alert_rules WHERE id = $1`,
[id]
);
return result.rowCount > 0;
}

86
backend/models/NotificationConfigModel.js
View File

@@ -1,86 +0,0 @@
import pool from '../infra/database.js';
/**
* Get notification config for a user
* @param {string} user_id - Firebase user ID
* @returns {Promise<Object|null>} Notification config or null if not set
*/
export async function getConfig(user_id) {
const result = await pool.query(
`SELECT user_id, discord_webhook_url, telegram_chat_id, telegram_bot_token,
email, notification_enabled, created_at, updated_at
FROM user_notification_configs
WHERE user_id = $1`,
[user_id]
);
return result.rows[0] || null;
}
/**
* Create or update notification config for a user
* @param {string} user_id - Firebase user ID
* @param {Object} config - Notification configuration
* @param {string} [config.discord_webhook_url] - Discord webhook URL
* @param {string} [config.telegram_chat_id] - Telegram chat ID
* @param {string} [config.email] - Email address
* @param {boolean} [config.notification_enabled] - Enable/disable notifications
* @returns {Promise<Object>} Updated config
*/
export async function upsertConfig(user_id, config) {
const {
discord_webhook_url,
telegram_chat_id,
telegram_bot_token,
email,
notification_enabled = true
} = config;
const result = await pool.query(
`INSERT INTO user_notification_configs
(user_id, discord_webhook_url, telegram_chat_id, telegram_bot_token, email, notification_enabled, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, NOW())
ON CONFLICT (user_id)
DO UPDATE SET
discord_webhook_url = COALESCE($2, user_notification_configs.discord_webhook_url),
telegram_chat_id = COALESCE($3, user_notification_configs.telegram_chat_id),
telegram_bot_token = COALESCE($4, user_notification_configs.telegram_bot_token),
email = COALESCE($5, user_notification_configs.email),
notification_enabled = $6,
updated_at = NOW()
RETURNING user_id, discord_webhook_url, telegram_chat_id, telegram_bot_token,
email, notification_enabled, created_at, updated_at`,
[user_id, discord_webhook_url, telegram_chat_id, telegram_bot_token, email, notification_enabled]
);
return result.rows[0];
}
/**
* Delete notification config for a user
* @param {string} user_id - Firebase user ID
* @returns {Promise<boolean>} True if deleted, false if not found
*/
export async function remove(user_id) {
const result = await pool.query(
`DELETE FROM user_notification_configs WHERE user_id = $1`,
[user_id]
);
return result.rowCount > 0;
}
/**
* Get all users with notifications enabled (for system use)
* @returns {Promise<Array>} Array of user configs
*/
export async function listEnabled() {
const result = await pool.query(
`SELECT user_id, discord_webhook_url, telegram_chat_id, email
FROM user_notification_configs
WHERE notification_enabled = TRUE`
);
return result.rows;
}

114
backend/notifications/discordNotifier.js
View File

@@ -1,114 +0,0 @@
/**
* https://discord.com/developers/docs/resources/webhook#execute-webhook
*/
/**
* Send an alert notification to webhook
* @param {string} webhookUrl - Discord webhook URL
* @param {string} message - Alert message
* @param {Object} metadata - Additional alert metadata
* @param {string} metadata.txHash - Transaction hash (nullable)
* @param {string} metadata.addressLabel - Address label
* @param {string} metadata.alertType - Alert type
* @param {string} metadata.address - Blockchain address
* @returns {Promise<boolean>} True if sent successfully, false otherwise
*/
export async function sendDiscordNotification(webhookUrl, message, metadata) {
const { txHash, addressLabel, alertType, address } = metadata;
try {
const payload = {
content: null,
embeds: [
{
title: 'Koin Ping Alert',
description: message,
color: getColorForAlertType(alertType),
fields: [
{
name: 'Address',
value: addressLabel || 'Unknown',
inline: true,
},
{
name: 'Blockchain Address',
value: `\`${address}\``,
inline: false,
},
],
timestamp: new Date().toISOString(),
footer: {
text: 'Koin Ping',
},
},
],
};
if (txHash) {
payload.embeds[0].fields.push({
name: 'Transaction',
value: `[View on Etherscan](https://etherscan.io/tx/${txHash})`,
inline: false,
});
}
const response = await fetch(webhookUrl, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(payload),
});
if (!response.ok) {
const errorText = await response.text();
throw new Error(
`Discord webhook failed: HTTP ${response.status} - ${errorText}`
);
}
return true;
} catch (error) {
console.error('Failed to send Discord notification:', error.message);
return false;
}
}
/**
* Get Discord embed color based on alert type
* @param {string} alertType - Alert type
* @returns {number}
*/
function getColorForAlertType(alertType) {
const colors = {
incoming_tx: 0x00ff00, // Green - incoming money
outgoing_tx: 0xff9900, // Orange - outgoing money
large_transfer: 0xff0000, // Red - large movement
balance_below: 0xff0000, // Red - low balance warning
};
return colors[alertType] || 0x0099ff;
}
/**
* Sends a test message to verify webhook works
* @param {string} webhookUrl
* @returns {Promise<boolean>}
*/
export async function testDiscordWebhook(webhookUrl) {
try {
const payload = {
content: 'Koin Ping test notification - Your Discord webhook is configured correctly!',
};
const response = await fetch(webhookUrl, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(payload),
});
return response.ok;
} catch (error) {
console.error('Discord webhook test failed:', error.message);
return false;
}
}

3225
backend/package-lock.json generated
View File

File diff suppressed because it is too large Load Diff

28
backend/package.json
View File

@@ -1,28 +0,0 @@
{
"name": "backend",
"version": "1.0.0",
"description": "",
"main": "index.js",
"scripts": {
"start": "node api/server.js",
"dev": "nodemon api/server.js",
"poller": "node poller/observerPoller.js",
"poller:dev": "nodemon poller/observerPoller.js",
"dev:all": "concurrently \"npm run dev\" \"npm run poller:dev\" --names \"API,POLLER\" --prefix-colors \"cyan,magenta\""
},
"keywords": [],
"author": "",
"license": "ISC",
"type": "module",
"dependencies": {
"cors": "^2.8.5",
"dotenv": "^17.2.3",
"express": "^5.2.1",
"firebase-admin": "^13.6.0",
"pg": "^8.16.3"
},
"devDependencies": {
"concurrently": "^9.2.1",
"nodemon": "^3.1.11"
}
}

125
backend/poller/observerPoller.js
View File

@@ -1,125 +0,0 @@
/**
* pure scheduling - no business logic, no DB, no blockchain details.
*/
import dotenv from 'dotenv';
import { ObserverService } from '../services/observerService.js';
import { EvaluatorService } from '../services/evaluatorService.js';
import { JsonRpcEthereum } from '../protocols/ethereum/JsonRpcEthereum.js';
dotenv.config();
const ETH_RPC_URL = process.env.ETH_RPC_URL;
const POLL_INTERVAL_MS = parseInt(process.env.POLL_INTERVAL_MS || '60000', 10);
if (!ETH_RPC_URL) {
console.error('ERROR: ETH_RPC_URL environment variable is required');
console.error('Please set it in your .env file');
process.exit(1);
}
if (isNaN(POLL_INTERVAL_MS) || POLL_INTERVAL_MS < 1000) {
console.error('ERROR: POLL_INTERVAL_MS must be a number >= 1000');
process.exit(1);
}
const eth = new JsonRpcEthereum(ETH_RPC_URL);
const observer = new ObserverService(eth);
const evaluator = new EvaluatorService(eth);
let isRunning = false;
/**
* Sleep for specified milliseconds
* @param {number} ms - Milliseconds to sleep
* @returns {Promise<void>}
*/
function sleep(ms) {
return new Promise((resolve) => setTimeout(resolve, ms));
}
/**
* Run observation cycle with error handling
*
* @returns {Promise<void>}
*/
async function runCycle() {
try {
const startTime = Date.now();
console.log(`[${new Date().toISOString()}] Starting observation cycle...`);
const observations = await observer.runOnce();
const alertsFired = await evaluator.evaluate(observations);
const duration = Date.now() - startTime;
console.log(
`[${new Date().toISOString()}] Cycle complete: ` +
`${observations.length} observations, ${alertsFired} alerts fired in ${duration}ms`
);
} catch (error) {
console.error(`[${new Date().toISOString()}] Observation cycle failed:`, error.message);
}
}
async function startPolling() {
if (isRunning) {
console.warn('Poller is already running');
return;
}
isRunning = true;
console.log('='.repeat(60));
console.log('Koin Ping Observer Poller Starting');
console.log('='.repeat(60));
console.log(`RPC URL: ${ETH_RPC_URL}`);
console.log(`Poll Interval: ${POLL_INTERVAL_MS}ms (${POLL_INTERVAL_MS / 1000}s)`);
console.log('='.repeat(60));
await runCycle();
while (isRunning) {
await sleep(POLL_INTERVAL_MS);
if (isRunning) {
await runCycle();
}
}
console.log('Poller stopped');
}
function stopPolling() {
if (!isRunning) {
return;
}
console.log('\n' + '='.repeat(60));
console.log('Shutting down poller gracefully...');
console.log('='.repeat(60));
isRunning = false;
}
// -------------------------
// Graceful Shutdown Handlers
// -------------------------
process.on('SIGINT', () => {
stopPolling();
setTimeout(() => process.exit(0), 2000);
});
process.on('SIGTERM', () => {
stopPolling();
setTimeout(() => process.exit(0), 2000);
});
if (import.meta.url === `file://${process.argv[1]}`) {
startPolling().catch((error) => {
console.error('Fatal error in poller:', error);
process.exit(1);
});
}
export { startPolling, stopPolling };

61
backend/protocols/ethereum/EthereumObserver.js
View File

@@ -1,61 +0,0 @@
/**
* @typedef {import('../../domain/NormalizedTx.js').NormalizedTx} NormalizedTx
*/
/**
* @typedef {Object} EthereumObserver
* @property {() => Promise<number>} getLatestBlockNumber - Get the most recent block number on the chain
* @property {(blockNumber: number) => Promise<NormalizedTx[]>} getBlockTransactions - Get all transactions in a specific block
* @property {(address: string) => Promise<string>} getBalance - Get the current balance of an address (in Wei as string)
*/
/**
* Validates that an object implements the EthereumObserver interface
* @param {any} obj - Object to validate
* @returns {boolean} True if object implements the interface
*/
export function isEthereumObserver(obj) {
if (!obj || typeof obj !== 'object') {
return false;
}
return (
typeof obj.getLatestBlockNumber === 'function' &&
typeof obj.getBlockTransactions === 'function' &&
typeof obj.getBalance === 'function'
);
}
/**
* Asserts that an object implements the EthereumObserver interface
* Throws an error if validation fails
* @param {any} obj - Object to validate
* @param {string} [name='object'] - Name for error messages
* @throws {Error} If object doesn't implement the interface
*/
export function assertEthereumObserver(obj, name = 'object') {
if (!isEthereumObserver(obj)) {
throw new Error(
`${name} must implement EthereumObserver interface ` +
`(getLatestBlockNumber, getBlockTransactions, getBalance)`
);
}
}
/**
* Example usage
*
* import { assertEthereumObserver } from './protocols/ethereum/EthereumObserver.js';
*
* export function createObserverService(ethereumObserver) {
* assertEthereumObserver(ethereumObserver, 'ethereumObserver');
* // ... use ethereumObserver safely
* }
*/
export default {
isEthereumObserver,
assertEthereumObserver,
};

159
backend/protocols/ethereum/JsonRpcEthereum.js
View File

@@ -1,159 +0,0 @@
import { assertEthereumObserver } from './EthereumObserver.js';
/**
* @typedef {import('../../domain/NormalizedTx.js').NormalizedTx} NormalizedTx
*/
const RPC_TIMEOUT_MS = 30000; // 30 seconds
/**
* JsonRpcEthereum - Concrete implementation of EthereumObserver using JSON-RPC
* @implements {EthereumObserver}
*/
export class JsonRpcEthereum {
/**
* @param {string} rpcUrl - Ethereum JSON-RPC endpoint URL
*/
constructor(rpcUrl) {
if (!rpcUrl || typeof rpcUrl !== 'string') {
throw new Error('JsonRpcEthereum requires a valid RPC URL');
}
this.rpcUrl = rpcUrl;
}
/**
* Make a JSON-RPC call to the Ethereum node
* @private
* @param {string} method - RPC method name (e.g., 'eth_blockNumber')
* @param {any[]} params - Method parameters
* @returns {Promise<any>} RPC result
* @throws {Error} If RPC call fails or returns an error
*/
async _callRpc(method, params = []) {
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), RPC_TIMEOUT_MS);
try {
const response = await fetch(this.rpcUrl, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
jsonrpc: '2.0',
id: Date.now(),
method,
params,
}),
signal: controller.signal,
});
clearTimeout(timeoutId);
if (!response.ok) {
throw new Error(
`HTTP ${response.status}: ${response.statusText} for ${method}`
);
}
const payload = await response.json();
if (payload.error) {
throw new Error(
`RPC Error [${method}]: ${payload.error.message} (code: ${payload.error.code})`
);
}
return payload.result;
} catch (error) {
clearTimeout(timeoutId);
if (error.name === 'AbortError') {
throw new Error(`RPC timeout after ${RPC_TIMEOUT_MS}ms for ${method}`);
}
throw new Error(`RPC call failed [${method}]: ${error.message}`);
}
}
/**
* Get the latest block number on the chain
* @returns {Promise<number>} Current block number
*/
async getLatestBlockNumber() {
const hexBlock = await this._callRpc('eth_blockNumber');
return parseInt(hexBlock, 16);
}
/**
* Get all transactions in a specific block
* @param {number} blockNumber - Block number to fetch
* @returns {Promise<NormalizedTx[]>} Array of normalized transactions
*/
async getBlockTransactions(blockNumber) {
const hexBlockNumber = '0x' + blockNumber.toString(16);
const block = await this._callRpc('eth_getBlockByNumber', [
hexBlockNumber,
true, // true = return full transaction objects, not just hashes
]);
if (!block || !block.transactions) {
return [];
}
const blockTimestamp = parseInt(block.timestamp, 16);
return block.transactions.map((tx) => ({
hash: tx.hash,
from: tx.from?.toLowerCase() || '',
to: tx.to?.toLowerCase() || null, // null for contract creation
value: this._hexToDecimalString(tx.value),
blockNumber: parseInt(tx.blockNumber, 16),
blockTimestamp: blockTimestamp,
}));
}
/**
* Get the current balance of an address
* @param {string} address - Ethereum address
* @returns {Promise<string>} Balance in Wei as decimal string
*/
async getBalance(address) {
const balanceHex = await this._callRpc('eth_getBalance', [
address,
'latest',
]);
return this._hexToDecimalString(balanceHex);
}
/**
* Convert hex string to decimal string (for large numbers)
* @private
* @param {string} hexValue - Hex value (with or without 0x prefix)
* @returns {string}
*/
_hexToDecimalString(hexValue) {
if (!hexValue || hexValue === '0x' || hexValue === '0x0') {
return '0';
}
try {
return BigInt(hexValue).toString();
} catch (error) {
throw new Error(`Invalid hex value: ${hexValue}`);
}
}
}
/**
* @param {string} rpcUrl - Ethereum JSON-RPC endpoint URL
* @returns {JsonRpcEthereum}
*/
export function createJsonRpcEthereum(rpcUrl) {
const instance = new JsonRpcEthereum(rpcUrl);
// Validate that it implements the interface
assertEthereumObserver(instance, 'JsonRpcEthereum');
return instance;
}

258
backend/services/evaluatorService.js
View File

@@ -1,258 +0,0 @@
import { assertEthereumObserver } from '../protocols/ethereum/EthereumObserver.js';
import * as AlertRuleModel from '../models/AlertRuleModel.js';
import * as AlertEventModel from '../models/AlertEventModel.js';
import * as AddressModel from '../models/AddressModel.js';
import * as NotificationConfigModel from '../models/NotificationConfigModel.js';
import { ethToWei, weiGreaterThanOrEqual, weiLessThan, formatWeiAsEth } from '../shared/weiConverter.js';
import { sendDiscordNotification } from '../notifications/discordNotifier.js';
/**
* @typedef {import('../domain/ObservedTx.js').ObservedTx} ObservedTx
* @typedef {import('../protocols/ethereum/EthereumObserver.js').EthereumObserver} EthereumObserver
*/
export class EvaluatorService {
/**
* @param {EthereumObserver} ethObserver - Implementation of EthereumObserver interface
*/
constructor(ethObserver) {
assertEthereumObserver(ethObserver, 'ethObserver');
this.eth = ethObserver;
}
/**
* Evaluate all observations against their associated alert rules
* @param {ObservedTx[]} observations
* @returns {Promise<number>}
*/
async evaluate(observations) {
let alertsFired = 0;
for (const obs of observations) {
try {
const fired = await this._evaluateObservation(obs);
alertsFired += fired;
} catch (error) {
console.error(
`Error evaluating observation for address ID ${obs.addressId}:`,
error.message
);
// Continue processing other observations - tbd
}
}
return alertsFired;
}
/**
* Evaluate a single observation against all rules
* @private
* @param {ObservedTx} obs
* @returns {Promise<number>}
*/
async _evaluateObservation(obs) {
const rules = await AlertRuleModel.listByAddress(obs.addressId);
let alertsFired = 0;
for (const rule of rules) {
// Skip disabled
if (!rule.enabled) {
continue;
}
const matches = await this._ruleMatches(rule, obs);
if (matches) {
await this._fireAlert(rule, obs);
alertsFired++;
}
}
return alertsFired;
}
/**
* Fire an alert by creating an alert event
* @private
* @param {Object} rule - Alert rule that matched
* @param {ObservedTx} obs - Observed transaction that triggered it
* @returns {Promise<void>}
*/
async _fireAlert(rule, obs) {
const address = await AddressModel.findById(obs.addressId);
const addressLabel = address?.label || address?.address || 'Unknown';
const message = this._buildMessage(rule, obs);
const txHash = obs.hash;
await AlertEventModel.create(rule.id, message, addressLabel, txHash);
console.log(
`[ALERT FIRED] Rule ${rule.id} (${rule.type}) - ${message} - TX: ${txHash}`
);
// Send Discord notification if configured
try {
const user_id = address.user_id;
const notificationConfig = await NotificationConfigModel.getConfig(user_id);
if (
notificationConfig?.discord_webhook_url &&
notificationConfig.notification_enabled
) {
const sent = await sendDiscordNotification(
notificationConfig.discord_webhook_url,
message,
{
txHash,
addressLabel,
alertType: rule.type,
address: address.address,
}
);
if (sent) {
console.log(`Discord notification sent to user ${user_id}`);
} else {
console.warn(`Discord notification failed for user ${user_id}`);
}
}
} catch (notificationError) {
// Don't fail the alert if notification fails
console.error(
`Failed to send Discord notification:`,
notificationError.message
);
}
}
/**
* Check if a rule matches an observation
* @private
* @param {Object} rule - Alert rule to evaluate
* @param {ObservedTx} obs - Observed transaction
* @returns {Promise<boolean>} True if rule conditions are met
*/
async _ruleMatches(rule, obs) {
switch (rule.type) {
case 'incoming_tx':
return this._matchesIncomingTx(obs);
case 'outgoing_tx':
return this._matchesOutgoingTx(obs);
case 'large_transfer':
return this._matchesLargeTransfer(rule, obs);
case 'balance_below':
return await this._matchesBalanceBelow(rule, obs);
default:
console.warn(`Unknown rule type: ${rule.type}`);
return false;
}
}
/**
* Match incoming transaction rule
* @private
*/
_matchesIncomingTx(obs) {
return obs.direction === 'incoming';
}
/**
* Match outgoing transaction rule
* @private
*/
_matchesOutgoingTx(obs) {
return obs.direction === 'outgoing';
}
/**
* Match large transfer rule
* @private
*
* IMPORTANT: Threshold is stored as ETH in DB, but we compare in Wei
* for precision. Both incoming and outgoing transfers are checked.
*/
_matchesLargeTransfer(rule, obs) {
if (!rule.threshold) {
return false;
}
const thresholdWei = ethToWei(Number(rule.threshold));
const transferValueWei = obs.value;
return weiGreaterThanOrEqual(transferValueWei, thresholdWei);
}
/**
* balance below threshold rule
* @private
*
* OPTIMIZATION: Only check balance after outgoing transactions
* (incoming transactions increase balance, so can't trigger this alert)
*
* IMPORTANT: Threshold stored as ETH in DB, but compared in Wei.
*/
async _matchesBalanceBelow(rule, obs) {
if (!rule.threshold) {
return false;
}
if (obs.direction !== 'outgoing') {
return false;
}
const address = await AddressModel.findById(obs.addressId);
if (!address) {
console.warn(`Address ID ${obs.addressId} not found`);
return false;
}
const balanceWei = await this.eth.getBalance(address.address);
// Convert ETH threshold from DB to Wei for comparison
const thresholdWei = ethToWei(Number(rule.threshold));
return weiLessThan(balanceWei, thresholdWei);
}
/**
* Build a human-readable alert message
* @private
* @param {Object} rule - Alert rule that matched
* @param {ObservedTx} obs - Observed transaction
* @returns {string} Human-readable message
*/
_buildMessage(rule, obs) {
switch (rule.type) {
case 'incoming_tx':
return `Incoming transaction: ${formatWeiAsEth(obs.value)} received`;
case 'outgoing_tx':
return `Outgoing transaction: ${formatWeiAsEth(obs.value)} sent`;
case 'large_transfer':
return `Large transfer detected: ${formatWeiAsEth(obs.value)} (threshold: ${rule.threshold} ETH)`;
case 'balance_below':
return `Balance dropped below threshold of ${rule.threshold} ETH`;
default:
return 'Alert triggered';
}
}
}
/**
* Factory function to create an EvaluatorService instance
* @param {EthereumObserver} ethObserver - Implementation of EthereumObserver interface
* @returns {EvaluatorService}
*/
export function createEvaluatorService(ethObserver) {
return new EvaluatorService(ethObserver);
}

158
backend/services/observerService.js
View File

@@ -1,158 +0,0 @@
import { assertEthereumObserver } from '../protocols/ethereum/EthereumObserver.js';
import * as AddressModel from '../models/AddressModel.js';
import * as CheckpointModel from '../models/AddressCheckpointModel.js';
/**
* @typedef {import('../domain/NormalizedTx.js').NormalizedTx} NormalizedTx
* @typedef {import('../domain/ObservedTx.js').ObservedTx} ObservedTx
* @typedef {import('../protocols/ethereum/EthereumObserver.js').EthereumObserver} EthereumObserver
*/
// Safety limit: maximum blocks to process per address per run
// Prevents overwhelming RPC endpoints and ensures bounded execution time
const MAX_BLOCKS_PER_RUN = 100;
export class ObserverService {
/**
* @param {EthereumObserver} ethObserver - Implementation of EthereumObserver interface
*/
constructor(ethObserver) {
assertEthereumObserver(ethObserver, 'ethObserver');
this.eth = ethObserver;
}
/**
* Run one observation cycle across all tracked addresses
*
*
* @returns {Promise<ObservedTx[]>} All observations from this cycle
*/
async runOnce() {
const addresses = await AddressModel.list();
if (addresses.length === 0) {
return [];
}
const latestBlock = await this.eth.getLatestBlockNumber();
const observations = [];
for (const address of addresses) {
try {
const addressObservations = await this._observeAddress(address, latestBlock);
observations.push(...addressObservations);
} catch (error) {
console.error(`Error observing address ${address.address}:`, error);
}
}
return observations;
}
/**
* Observe a single address for activity
* @private
* @param {Object} address - Address record from database
* @param {number} latestBlock - Current latest block number
* @returns {Promise<ObservedTx[]>} Observations for this address
*/
async _observeAddress(address, latestBlock) {
const lastChecked = await CheckpointModel.getLastCheckedBlock(address.id);
const startBlock = this._getStartBlock(lastChecked, latestBlock);
const endBlock = this._getEndBlock(startBlock, latestBlock);
// No new blocks to check
if (startBlock > endBlock) {
return [];
}
const observations = [];
for (let blockNumber = startBlock; blockNumber <= endBlock; blockNumber++) {
const blockTxs = await this.eth.getBlockTransactions(blockNumber);
const relevantTxs = this._filterRelevantTransactions(blockTxs, address.address);
for (const tx of relevantTxs) {
observations.push(this._createObservedTx(tx, address));
}
}
await CheckpointModel.updateLastCheckedBlock(address.id, endBlock);
return observations;
}
/**
* Determine starting block for observation
* @private
* @param {number|null} lastChecked - Last checked block, or null if never checked
* @param {number} latestBlock - Current latest block
* @returns {number} Block number to start from
*/
_getStartBlock(lastChecked, latestBlock) {
if (lastChecked === null) {
return latestBlock;
}
return lastChecked + 1;
}
/**
* Determine ending block for observation (respects MAX_BLOCKS_PER_RUN)
* @private
* @param {number} startBlock
* @param {number} latestBlock
* @returns {number}
*/
_getEndBlock(startBlock, latestBlock) {
return Math.min(startBlock + MAX_BLOCKS_PER_RUN - 1, latestBlock);
}
/**
* Filter transactions to only those involving the tracked address
* @private
* @param {NormalizedTx[]} transactions
* @param {string} trackedAddress
* @returns {NormalizedTx[]}
*/
_filterRelevantTransactions(transactions, trackedAddress) {
const addressLower = trackedAddress.toLowerCase();
return transactions.filter((tx) => {
const fromMatch = tx.from?.toLowerCase() === addressLower;
const toMatch = tx.to?.toLowerCase() === addressLower;
return fromMatch || toMatch;
});
}
/**
* Create an ObservedTx with direction metadata
* @private
* @param {NormalizedTx} tx - Normalized transaction
* @param {Object} address - Address record from database
* @returns {ObservedTx} Transaction with observation metadata
*/
_createObservedTx(tx, address) {
const addressLower = address.address.toLowerCase();
const direction = tx.to?.toLowerCase() === addressLower ? 'incoming' : 'outgoing';
return {
...tx,
addressId: address.id,
direction,
};
}
}
/**
* Factory to create an ObserverService instance
* @param {EthereumObserver} ethObserver - Implementation of EthereumObserver interface
* @returns {ObserverService}
*/
export function createObserverService(ethObserver) {
return new ObserverService(ethObserver);
}

95
backend/shared/weiConverter.js
View File

@@ -1,95 +0,0 @@
const WEI_PER_ETH = 1000000000000000000n; // 10^18 Wei = 1 ETH
/**
* Convert Wei (as string) to ETH (as number)
* Use for display purposes only, not for calculations
* @param {string} weiString - Wei value as string
* @returns {number} ETH value as number
*/
export function weiToEth(weiString) {
if (!weiString || weiString === '0') {
return 0;
}
try {
const wei = BigInt(weiString);
// Convert to number (safe for display values)
return Number(wei) / Number(WEI_PER_ETH);
} catch (error) {
throw new Error(`Invalid Wei value: ${weiString}`);
}
}
/**
* Convert ETH (as number) to Wei (as string)
* Use when converting DB thresholds for comparison
* @param {number} eth - ETH value as number
* @returns {string} Wei value as string
*/
export function ethToWei(eth) {
if (typeof eth !== 'number' || isNaN(eth)) {
throw new Error(`Invalid ETH value: ${eth}`);
}
if (eth === 0) {
return '0';
}
// Convert to string with fixed precision to avoid floating point issues
const ethString = eth.toFixed(18); // Max precision
const [whole, decimal = ''] = ethString.split('.');
const wholePart = BigInt(whole) * WEI_PER_ETH;
const decimalPart = BigInt(decimal.padEnd(18, '0'));
return (wholePart + decimalPart).toString();
}
/**
* Compare two Wei values
* @param {string} weiA
* @param {string} weiB
* @returns {number}
*/
export function compareWei(weiA, weiB) {
const a = BigInt(weiA);
const b = BigInt(weiB);
if (a < b) return -1;
if (a > b) return 1;
return 0;
}
/**
* Check if Wei value A is greater than or equal to Wei value B
* @param {string} weiA - First Wei value as string
* @param {string} weiB - Second Wei value as string
* @returns {boolean}
*/
export function weiGreaterThanOrEqual(weiA, weiB) {
return BigInt(weiA) >= BigInt(weiB);
}
/**
* Check if Wei value A is less than Wei value B
* @param {string} weiA
* @param {string} weiB
* @returns {boolean}
*/
export function weiLessThan(weiA, weiB) {
return BigInt(weiA) < BigInt(weiB);
}
/**
* @param {string} weiString
* @param {number} decimals - Number of decimal places (default: 4)
* @returns {string}
*/
export function formatWeiAsEth(weiString, decimals = 4) {
const eth = weiToEth(weiString);
return `${eth.toFixed(decimals)} ETH`;
}