feat: On-Chain-Scanner für Watchlist-Transfers (eth_getLogs, RPC-Fallback, Chunking)

src/server/signals/onchain.test.ts

@@ -0,0 +1,45 @@
+import { describe, expect, test } from 'bun:test';
+import { buildLogFilter, decodeTransferLogs, TRANSFER_TOPIC } from './onchain';
+
+const WALLET = '0x5be9a4959308a0d0c7bc0870e319314d8d957dbb';
+
+describe('buildLogFilter', () => {
+  test('filtert auf Token-Whitelist + Transfer-Topic + Wallet als Empfänger', () => {
+    const f = buildLogFilter(100, 200);
+    expect(f.fromBlock).toBe('0x64');
+    expect(f.toBlock).toBe('0xc8');
+    expect(f.address).toContain('0x514910771af9ca656af840dff83e8264ecf986ca'); // LINK
+    expect(f.topics[0]).toBe(TRANSFER_TOPIC);
+    expect(f.topics[1]).toBeNull(); // from: beliebig
+    expect(f.topics[2]).toContain('0x000000000000000000000000' + WALLET.slice(2));
+  });
+});
+
+describe('decodeTransferLogs', () => {
+  test('dekodiert Token, Menge (decimals-skaliert), Tx-Hash, Block', () => {
+    const log = {
+      address: '0x514910771af9ca656af840dff83e8264ecf986ca', // LINK, 18 decimals
+      topics: [
+        TRANSFER_TOPIC,
+        '0x000000000000000000000000aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa',
+        '0x000000000000000000000000' + WALLET.slice(2),
+      ],
+      data: '0x' + (5n * 10n ** 18n).toString(16).padStart(64, '0'), // 5 LINK
+      transactionHash: '0xabc',
+      blockNumber: '0x64',
+    };
+    const out = decodeTransferLogs([log]);
+    expect(out).toHaveLength(1);
+    expect(out[0]).toEqual({ symbol: 'LINK', instrument: 'LINK_USDT', amount: 5, txHash: '0xabc', blockNumber: 100 });
+  });
+  test('ignoriert Logs unbekannter Token-Contracts', () => {
+    const log = {
+      address: '0x000000000000000000000000000000000000dead',
+      topics: [TRANSFER_TOPIC, '0x0', '0x0'],
+      data: '0x1',
+      transactionHash: '0xdef',
+      blockNumber: '0x65',
+    };
+    expect(decodeTransferLogs([log])).toHaveLength(0);
+  });
+});

src/server/signals/onchain.ts

@@ -0,0 +1,86 @@
+import type { Pair } from '../types';
+import { RPC_URLS, TOKEN_BY_CONTRACT, TRACKED_TOKENS, WATCHED_WALLETS } from './watchlist';
+
+/** keccak256("Transfer(address,address,uint256)") */
+export const TRANSFER_TOPIC = '0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef';
+
+export interface OnchainTransfer {
+  symbol: string;
+  instrument: Pair | null;
+  amount: number; // decimals-skaliert
+  txHash: string;
+  blockNumber: number;
+}
+
+export function buildLogFilter(fromBlock: number, toBlock: number) {
+  return {
+    fromBlock: '0x' + fromBlock.toString(16),
+    toBlock: '0x' + toBlock.toString(16),
+    address: TRACKED_TOKENS.map((t) => t.contract),
+    topics: [
+      TRANSFER_TOPIC,
+      null, // from: beliebig
+      WATCHED_WALLETS.map((w) => '0x000000000000000000000000' + w.address.slice(2)),
+    ] as (string | string[] | null)[],
+  };
+}
+
+export function decodeTransferLogs(logs: any[]): OnchainTransfer[] {
+  const out: OnchainTransfer[] = [];
+  for (const log of logs) {
+    const token = TOKEN_BY_CONTRACT.get(String(log.address).toLowerCase());
+    if (!token) continue;
+    const raw = BigInt(log.data === '0x' ? '0x0' : log.data);
+    out.push({
+      symbol: token.symbol,
+      instrument: token.instrument,
+      amount: Number(raw) / 10 ** token.decimals,
+      txHash: log.transactionHash,
+      blockNumber: Number(BigInt(log.blockNumber)),
+    });
+  }
+  return out;
+}
+
+/** JSON-RPC mit URL-Fallback (erst alle URLs einmal, dann Fehler). */
+export async function rpc(method: string, params: unknown[]): Promise<any> {
+  let lastErr: unknown;
+  for (const url of RPC_URLS) {
+    try {
+      const res = await fetch(url, {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json' },
+        body: JSON.stringify({ jsonrpc: '2.0', id: 1, method, params }),
+        signal: AbortSignal.timeout(15_000),
+      });
+      if (!res.ok) throw new Error(`HTTP ${res.status} (${url})`);
+      const json = await res.json();
+      if (json.error) throw new Error(`RPC ${method}: ${json.error.message} (${url})`);
+      return json.result;
+    } catch (err) {
+      lastErr = err;
+    }
+  }
+  throw lastErr;
+}
+
+export async function getBlockNumber(): Promise<number> {
+  return Number(BigInt(await rpc('eth_blockNumber', [])));
+}
+
+/** Block-Timestamp in Unix ms. */
+export async function getBlockTs(blockNumber: number): Promise<number> {
+  const block = await rpc('eth_getBlockByNumber', ['0x' + blockNumber.toString(16), false]);
+  return Number(BigInt(block.timestamp)) * 1000;
+}
+
+/** Transfers in Watchlist-Wallets im Blockbereich [fromBlock, toBlock] (inkl.), gechunkt à maxChunk. */
+export async function fetchTransfers(fromBlock: number, toBlock: number, maxChunk = 5000): Promise<OnchainTransfer[]> {
+  const out: OnchainTransfer[] = [];
+  for (let from = fromBlock; from <= toBlock; from += maxChunk) {
+    const to = Math.min(from + maxChunk - 1, toBlock);
+    const logs = await rpc('eth_getLogs', [buildLogFilter(from, to)]);
+    out.push(...decodeTransferLogs(logs));
+  }
+  return out;
+}