diff --git a/backend/src/db/agents.ts b/backend/src/db/agents.ts index 7e81bed5..316fa347 100644 --- a/backend/src/db/agents.ts +++ b/backend/src/db/agents.ts @@ -250,6 +250,27 @@ export function createAgentDb(db: Database.Database): AgentDb { AND datetime(lastSeenAt, '+' || ? || ' hours') < datetime('now') `).run(offlineThresholdHours); return result.changes; + }, + + upsertError(error: { + id: string; + reporter: string; + resolved: boolean; + resolution: string | null; + reportedAt: string; + }): void { + const errorsDb = getErrorDb(); + errorsDb.prepare( + `INSERT OR IGNORE INTO errors + (id, errorCode, message, agentId, reporter, status, resolution, rentStroops, maintenanceAccount, createdAt, expiresAt, resolvedAt) + VALUES + (?, 0, '', '', ?, 'active', NULL, '0', '', ?, '', NULL)`, + ).run(error.id, error.reporter ?? "", error.reportedAt ?? new Date().toISOString()); + }, + + resolveError(errorId: string, resolution: string): void { + const store = createErrorRegistryStore(getErrorDb()); + store.resolve(errorId, resolution); } }; } diff --git a/backend/src/db/errorRegistry.ts b/backend/src/db/errorRegistry.ts new file mode 100644 index 00000000..a7b958df --- /dev/null +++ b/backend/src/db/errorRegistry.ts @@ -0,0 +1,217 @@ +/** + * Error-registry persistence + fee/rent accounting. + * + * Mirrors the on-chain error lifecycle (`err_rptd` / `err_rslvd`) into a local + * SQLite store so error records are queryable and durable without unbounded + * growth. Each entry is keyed by its on-chain `error_id` and carries: + * + * - a **TTL-based rent** expiry (`expires_at` from `created_at + ttl_seconds`), + * - a **per-entry fee/rent charge** (`rent_stroops`) that accounts for the + * storage footprint the report occupies, attributed to a maintenance + * account (`maintenance_account`), + * - a **per-agent live-entry cap** so one faulty agent cannot saturate the + * registry. + * + * Expired records are reclaimed deterministically by {@link sweepExpired}, + * which runs on the existing interval-based maintenance loop. Reads always + * filter by `expires_at` so callers never observe stale rows regardless of + * whether a sweep has run. + */ + +import Database from "better-sqlite3"; +import path from "path"; +import { createLogger } from "../utils/logger"; + +const logger = createLogger({ component: "error-registry-db" }); + +export type ErrorStatus = "active" | "resolved"; + +export interface ErrorRecord { + id: string; + errorCode: number; + message: string; + agentId: string; + reporter: string; + status: ErrorStatus; + resolution: string | null; + /** Stroops charged for storing this report (fee/rent accounting). */ + rentStroops: bigint; + /** Address that receives the collected storage rent. */ + maintenanceAccount: string; + createdAt: string; + expiresAt: string; + resolvedAt: string | null; +} + +export interface SubmitErrorInput { + id: string; + errorCode: number; + message: string; + agentId: string; + reporter: string; + ttlSeconds: number; + rentStroops: bigint; + maintenanceAccount: string; +} + +export interface ErrorRegistryStore { + submit(input: SubmitErrorInput): ErrorRecord; + resolve(errorId: string, resolution: string): void; + findById(id: string): ErrorRecord | undefined; + listByAgent(agentId: string): ErrorRecord[]; + countLiveByAgent(agentId: string): number; + /** Bound the number of live entries a single agent may hold. */ + capLiveEntries(agentId: string, cap: number): void; + /** Delete every expired row in one pass; returns how many were removed. */ + sweepExpired(now?: number): number; +} + +let _db: Database.Database | null = null; + +function getErrorDb(dbPath?: string): Database.Database { + if (!_db) { + const filePath = dbPath ?? path.join(process.cwd(), "errors.db"); + _db = new Database(filePath as unknown as string); + _db.pragma("busy_timeout = 5000"); + _db.pragma("journal_mode = WAL"); + _db.exec(` + CREATE TABLE IF NOT EXISTS errors ( + id TEXT PRIMARY KEY, + errorCode INTEGER NOT NULL, + message TEXT NOT NULL DEFAULT '', + agentId TEXT NOT NULL, + reporter TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active', + resolution TEXT, + rentStroops TEXT NOT NULL DEFAULT '0', + maintenanceAccount TEXT NOT NULL, + createdAt TEXT NOT NULL, + expiresAt TEXT NOT NULL, + resolvedAt TEXT + ); + CREATE INDEX IF NOT EXISTS idx_errors_agent ON errors (agentId, status); + CREATE INDEX IF NOT EXISTS idx_errors_expiry ON errors (expiresAt); + `); + try { + (_db as unknown as { on: (e: string, f: (err: Error) => void) => void }).on( + "error", + (err: Error) => logger.error({ err }, "error-registry database error"), + ); + } catch { + // error events are unavailable on some runtimes + } + } + return _db; +} + +export function closeErrorDb(): void { + _db?.close(); + _db = null; +} + +function mapRow(row: Record): ErrorRecord { + return { + id: row.id as string, + errorCode: row.errorCode as number, + message: row.message as string, + agentId: row.agentId as string, + reporter: row.reporter as string, + status: row.status as ErrorStatus, + resolution: (row.resolution as string | null) ?? null, + rentStroops: BigInt((row.rentStroops as string) || "0"), + maintenanceAccount: row.maintenanceAccount as string, + createdAt: row.createdAt as string, + expiresAt: row.expiresAt as string, + resolvedAt: (row.resolvedAt as string | null) ?? null, + }; +} + +export function createErrorRegistryStore(db: Database.Database): ErrorRegistryStore { + return { + submit(input: SubmitErrorInput): ErrorRecord { + const createdAt = new Date(); + const expiresAt = new Date(createdAt.getTime() + input.ttlSeconds * 1000).toISOString(); + const record: ErrorRecord = { + id: input.id, + errorCode: input.errorCode, + message: input.message, + agentId: input.agentId, + reporter: input.reporter, + status: "active", + resolution: null, + rentStroops: input.rentStroops, + maintenanceAccount: input.maintenanceAccount, + createdAt: createdAt.toISOString(), + expiresAt, + resolvedAt: null, + }; + db.prepare( + `INSERT OR IGNORE INTO errors + (id, errorCode, message, agentId, reporter, status, resolution, rentStroops, maintenanceAccount, createdAt, expiresAt, resolvedAt) + VALUES + (@id, @errorCode, @message, @agentId, @reporter, @status, @resolution, @rentStroops, @maintenanceAccount, @createdAt, @expiresAt, @resolvedAt)`, + ).run({ + ...record, + rentStroops: record.rentStroops.toString(), + resolution: null, + resolvedAt: null, + }); + const row = db.prepare("SELECT * FROM errors WHERE id = ?").get(record.id) as Record; + return mapRow(row); + }, + + resolve(errorId: string, resolution: string): void { + db.prepare( + "UPDATE errors SET status = 'resolved', resolution = ?, resolvedAt = ? WHERE id = ? AND status = 'active'", + ).run(resolution, new Date().toISOString(), errorId); + }, + + findById(id: string): ErrorRecord | undefined { + const row = db.prepare("SELECT * FROM errors WHERE id = ?").get(id) as Record | undefined; + return row ? mapRow(row) : undefined; + }, + + listByAgent(agentId: string): ErrorRecord[] { + const rows = db.prepare("SELECT * FROM errors WHERE agentId = ? ORDER BY createdAt DESC").all(agentId) as Array>; + return rows.map(mapRow); + }, + + countLiveByAgent(agentId: string): number { + const row = db.prepare( + "SELECT COUNT(*) AS total FROM errors WHERE agentId = ? AND status = 'active' AND expiresAt > ?", + ).get(agentId, new Date().toISOString()) as { total: number }; + return Number(row.total) || 0; + }, + + capLiveEntries(agentId: string, cap: number): void { + if (cap <= 0) return; + const rows = db.prepare( + "SELECT id FROM errors WHERE agentId = ? AND status = 'active' AND expiresAt > ? ORDER BY createdAt DESC", + ).all(agentId, new Date().toISOString()) as Array<{ id: string }>; + if (rows.length <= cap) return; + // Keep the newest `cap` entries and resolve the overflow oldest ones, + // so the agent's live footprint is bounded. + const overflow = rows.slice(cap); + const resolve = db.prepare( + "UPDATE errors SET status = 'resolved', resolution = 'capacity', resolvedAt = ? WHERE id = ? AND status = 'active'", + ); + for (const row of overflow) { + resolve.run(new Date().toISOString(), row.id); + } + logger.info({ agentId, evicted: overflow.length }, "evicted oldest error entries over per-agent cap"); + }, + + sweepExpired(now: number = Date.now()): number { + const iso = new Date(now).toISOString(); + const result = db.prepare("DELETE FROM errors WHERE status = 'active' AND expiresAt <= ?").run(iso); + return result.changes; + }, + }; +} + +/** Default per-agent live-entry cap. */ +export const DEFAULT_ERROR_CAP_PER_AGENT = 100; +/** Default TTL (seconds) applied when the caller does not supply one. */ +export const DEFAULT_ERROR_TTL_SECONDS = 90 * 24 * 60 * 60; +/** Default per-entry storage-rent charge, in stroops (0.01 XLM). */ +export const DEFAULT_RENT_STROOPS = 100_000n; diff --git a/backend/src/index.ts b/backend/src/index.ts index 2e6aca5a..90554420 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -41,6 +41,22 @@ async function main() { const reconciliationService = createDefaultReconciliationService(); reconciliationService.startDaily(config.RECONCILIATION_INTERVAL_MS); + // Start SQLite maintenance (WAL checkpoint, vacuum, backup) + const maintenanceService = new DbMaintenanceService(defaultMaintenanceDatabases(), { + intervalMs: config.DB_MAINTENANCE_INTERVAL_MS, + vacuumThreshold: config.DB_MAINTENANCE_VACUUM_THRESHOLD, + backupDir: config.DB_BACKUP_DIR, + backupRetentionCount: config.DB_BACKUP_RETENTION_COUNT, + }); + maintenanceService.start(); + + // Start error-registry maintenance (expiry sweep + per-agent cap) + const errorRegistryMaintenance = new ErrorRegistryMaintenanceService({ + intervalMs: config.ERROR_REGISTRY_MAINTENANCE_INTERVAL_MS, + capPerAgent: config.ERROR_REGISTRY_CAP_PER_AGENT, + }); + errorRegistryMaintenance.start(); + // Create and start the server const { httpServer, close } = createApp(); @@ -60,6 +76,8 @@ async function main() { cleanupService.stop(); reconciliationService.stop(); + maintenanceService.stop(); + errorRegistryMaintenance.stop(); globalAgentRegistry.shutdown(); stopAgentSync(); diff --git a/backend/src/registry/sync.ts b/backend/src/registry/sync.ts index 7565a708..f3b17336 100644 --- a/backend/src/registry/sync.ts +++ b/backend/src/registry/sync.ts @@ -233,9 +233,16 @@ function handleEvent( // These can be wired up once an errors table is added to the schema. case TOPICS.ERR_REPORTED: { const data = payload as ErrorReportedPayload; + db.upsertError?.({ + id: data.error_id, + reporter: data.reporter, + resolved: false, + resolution: null, + reportedAt: new Date().toISOString(), + }); logger.warn( { errorId: data.error_id, reporter: data.reporter }, - "error reported (no DB schema for errors — logging only)" + "error reported and persisted to error registry" ); break; } @@ -243,9 +250,10 @@ function handleEvent( case TOPICS.ERR_RESOLVED: { const data = payload as ErrorResolvedPayload; const label = resolutionLabel(data.resolution_code); + db.resolveError?.(data.error_id, label); logger.info( { errorId: data.error_id, resolution: label }, - "error resolved (no DB schema for errors — logging only)" + "error resolved and persisted to error registry" ); break; } diff --git a/backend/src/services/dbMaintenance.ts b/backend/src/services/dbMaintenance.ts new file mode 100644 index 00000000..e476add3 --- /dev/null +++ b/backend/src/services/dbMaintenance.ts @@ -0,0 +1,277 @@ +/** + * SQLite database maintenance service. + * + * Long-term data integrity for the node's on-disk SQLite databases + * (`payments.db`, `tasks.db`, `agents.db`, `jobs.db`). SQLite in WAL mode + * grows its `-wal` sidecar and accumulates free pages over time; without + * maintenance the journal and file fragmentation slowly degrade insert/scan + * latency. This service runs three periodic upkeep jobs: + * + * 1. **WAL checkpoint** — `PRAGMA wal_checkpoint(TRUNCATE)` removes committed + * frames from the `-wal` file so it cannot grow unbounded. + * 2. **Incremental vacuum** — when the free-page ratio exceeds a threshold the + * `VACUUM` command compacts the database file, reclaiming space and + * reordering pages for locality. + * 3. **Scheduled backup** — a consistent snapshot is taken via SQLite's online + * backup API (`db.backup()`) so it can run against a live connection, with a + * retention policy that prunes the oldest snapshots. + * + * Every run is recorded in {@link DbMaintenanceMetrics} so operators can + * observe upkeep via the health dashboard and logs. The service is + * intentionally self-contained: callers register the databases to maintain, + * and it never assumes ownership of those connections (it never closes a + * connection it did not open). + */ + +import path from "path"; +import fs from "fs"; +import { createLogger } from "../utils/logger"; +import type { Database } from "better-sqlite3"; + +const logger = createLogger({ component: "db-maintenance" }); + +/** A database the maintenance service knows how to keep healthy. */ +export interface MaintenanceDb { + /** Stable identifier used in metrics and backup file names. */ + name: string; + /** Absolute path to the main database file. */ + path: string; + /** Returns the live connection used by the rest of the application. */ + getConnection: () => Database.Database; +} + +/** Result of a single maintenance pass across all registered databases. */ +export interface DbMaintenanceMetrics { + lastRunAt: string | null; + databasesCheckpointed: number; + databasesVacuumed: number; + databasesBackedUp: number; + backupsDeleted: number; + /** Per-database roll-up keyed by database name. */ + byDatabase: Record< + string, + { + checkpointed: boolean; + vacuumed: boolean; + backedUp: boolean; + backupPath: string | null; + freePages: number; + totalPages: number; + } + >; +} + +export interface DbMaintenanceOptions { + /** How often (ms) the maintenance pass runs. Default: 6 hours. */ + intervalMs?: number; + /** Free-page ratio (0..1) above which `VACUUM` runs. Default: 0.2. */ + vacuumThreshold?: number; + /** Directory where backups are written. Default: `/backups`. */ + backupDir?: string; + /** Maximum number of snapshots retained per database. Default: 14. */ + backupRetentionCount?: number; +} + +const DEFAULT_INTERVAL_MS = 6 * 60 * 60 * 1000; +const DEFAULT_VACUUM_THRESHOLD = 0.2; +const DEFAULT_BACKUP_RETENTION = 14; + +/** + * Query the SQLite pagination state for free-page accounting. + * + * `PRAGMA freelist_count` reports pages available for reuse; `PRAGMA + * page_count` reports the current file size in pages. The free ratio is used + * to decide whether a vacuum is worthwhile. + */ +export function readPageStats(db: Database.Database): { freePages: number; totalPages: number } { + const freeRows = db.pragma("freelist_count", { simple: true }); + const totalRows = db.pragma("page_count", { simple: true }); + const freePages = Array.isArray(freeRows) ? Number(freeRows[0]) : Number(freeRows); + const totalPages = Array.isArray(totalRows) ? Number(totalRows[0]) : Number(totalRows); + return { freePages: Number.isFinite(freePages) ? freePages : 0, totalPages: Number.isFinite(totalPages) ? totalPages : 0 }; +} + +export class DbMaintenanceService { + private readonly intervalMs: number; + private readonly vacuumThreshold: number; + private readonly backupDir: string; + private readonly backupRetentionCount: number; + private readonly databases: MaintenanceDb[]; + private timer: NodeJS.Timeout | null = null; + private stopped = false; + private metrics: DbMaintenanceMetrics = emptyMetrics(); + + constructor(databases: MaintenanceDb[], options: DbMaintenanceOptions = {}) { + this.databases = databases; + this.intervalMs = options.intervalMs ?? DEFAULT_INTERVAL_MS; + this.vacuumThreshold = options.vacuumThreshold ?? DEFAULT_VACUUM_THRESHOLD; + this.backupRetentionCount = options.backupRetentionCount ?? DEFAULT_BACKUP_RETENTION; + this.backupDir = options.backupDir ?? path.join(process.cwd(), "backups"); + } + + start(): void { + if (this.timer) return; + this.stopped = false; + fs.mkdirSync(this.backupDir, { recursive: true }); + this.timer = setInterval(() => { + void this.run(); + }, this.intervalMs); + // Run once (not awaited) shortly after startup for an immediate baseline. + void this.run(); + } + + stop(): void { + this.stopped = true; + if (this.timer) { + clearInterval(this.timer); + this.timer = null; + } + } + + getMetrics(): DbMaintenanceMetrics { + return this.metrics; + } + + /** Run a full maintenance pass now. Safe to call more than once. */ + async run(): Promise { + if (this.stopped) return; + const startedAt = Date.now(); + const snapshot = emptyMetrics(); + let backupsDeleted = 0; + + for (const db of this.databases) { + const entry = snapshot.byDatabase[db.name] ?? { + checkpointed: false, + vacuumed: false, + backedUp: false, + backupPath: null, + freePages: 0, + totalPages: 0, + }; + try { + const connection = db.getConnection(); + const pageStats = readPageStats(connection); + entry.freePages = pageStats.freePages; + entry.totalPages = pageStats.totalPages; + + // 1. WAL checkpoint (TRUNCATE) so the journal sidecar stays bounded. + connection.pragma("wal_checkpoint(TRUNCATE)"); + entry.checkpointed = true; + + // 2. Incremental vacuum when free pages exceed the threshold. + const freeRatio = + pageStats.totalPages > 0 ? pageStats.freePages / pageStats.totalPages : 0; + if (freeRatio > this.vacuumThreshold) { + connection.exec("VACUUM"); + entry.vacuumed = true; + } + + // 3. Online backup with retention. + const backupPath = await this.createBackup(db); + entry.backedUp = true; + entry.backupPath = backupPath; + backupsDeleted += this.applyRetention(db.name); + } catch (error) { + logger.error({ err: error, database: db.name }, "maintenance tick failed for database"); + } + snapshot.byDatabase[db.name] = entry; + } + + snapshot.databasesCheckpointed = Object.values(snapshot.byDatabase).filter((d) => d.checkpointed).length; + snapshot.databasesVacuumed = Object.values(snapshot.byDatabase).filter((d) => d.vacuumed).length; + snapshot.databasesBackedUp = Object.values(snapshot.byDatabase).filter((d) => d.backedUp).length; + snapshot.backupsDeleted = backupsDeleted; + snapshot.lastRunAt = new Date().toISOString(); + this.metrics = snapshot; + + logger.info( + { + elapsedMs: Date.now() - startedAt, + checkpointed: snapshot.databasesCheckpointed, + vacuumed: snapshot.databasesVacuumed, + backedUp: snapshot.databasesBackedUp, + backupsDeleted, + }, + "database maintenance pass completed", + ); + } + + /** + * Take a consistent snapshot of a live database using SQLite's online backup + * API. Each snapshot is timestamped and prefixed with the database name so + * retention can identify siblings. + */ + private async createBackup(db: MaintenanceDb): Promise { + const rawPath = path.join(this.backupDir, `${db.name}-${Date.now()}.db`); + const connection = db.getConnection(); + await connection.backup(rawPath); + return rawPath; + } + + /** + * Trim a database's snapshots down to the retention count, deleting oldest + * first. Returns how many files were removed. + */ + private applyRetention(dbName: string): number { + let files: string[] = []; + try { + files = fs.readdirSync(this.backupDir).filter((f) => f.startsWith(`${dbName}-`) && f.endsWith(".db")); + } catch { + return 0; + } + files.sort(); + const toDelete = files.length - this.backupRetentionCount; + let deleted = 0; + for (let i = 0; i < toDelete; i += 1) { + const file = files[i]; + if (!file) continue; + try { + fs.unlinkSync(path.join(this.backupDir, file)); + deleted += 1; + } catch (error) { + logger.error({ err: error, database: dbName, file }, "failed to prune backup"); + } + } + return deleted; + } +} + +function emptyMetrics(): DbMaintenanceMetrics { + return { + lastRunAt: null, + databasesCheckpointed: 0, + databasesVacuumed: 0, + databasesBackedUp: 0, + backupsDeleted: 0, + byDatabase: {}, + }; +} + +/** + * The default maintenance registry for this node: the four on-disk WAL + * databases the application actually persists to. + */ +export function defaultMaintenanceDatabases(): MaintenanceDb[] { + return [ + { + name: "payments", + path: path.join(process.cwd(), "payments.db"), + getConnection: () => require("../db").getDb(), + }, + { + name: "tasks", + path: path.join(process.cwd(), "tasks.db"), + getConnection: () => require("../db/tasks").getTaskDb(), + }, + { + name: "agents", + path: path.join(process.cwd(), "agents.db"), + getConnection: () => require("../db/agents").getAgentDb(), + }, + { + name: "jobs", + path: path.join(process.cwd(), "jobs.db"), + getConnection: () => require("../queue/jobStore").getJobDb(), + }, + ]; +} diff --git a/backend/src/services/errorRegistryMaintenance.ts b/backend/src/services/errorRegistryMaintenance.ts new file mode 100644 index 00000000..4c7f1d98 --- /dev/null +++ b/backend/src/services/errorRegistryMaintenance.ts @@ -0,0 +1,81 @@ +/** + * Error-registry maintenance service. + * + * Runs the periodic upkeep that keeps the local error-registry store bounded: + * deterministic reclamation of TTL-expired entries and enforcement of the + * per-agent live-entry cap. Both operations are idempotent and cheap, so the + * service may run on the same cadence as the other interval-based services in + * this codebase. + */ +import { createLogger } from "../utils/logger"; +import { createErrorRegistryStore, getErrorDb, closeErrorDb } from "../db/errorRegistry"; + +const logger = createLogger({ component: "error-registry-maintenance" }); + +export interface ErrorRegistryMaintenanceOptions { + intervalMs?: number; + /** Maximum live (non-expired, non-resolved) entries per agent. */ + capPerAgent?: number; +} + +const DEFAULT_INTERVAL_MS = 60 * 60 * 1000; +const DEFAULT_CAP_PER_AGENT = 100; + +export class ErrorRegistryMaintenanceService { + private readonly intervalMs: number; + private readonly capPerAgent: number; + private timer: NodeJS.Timeout | null = null; + private stopped = false; + + constructor(options: ErrorRegistryMaintenanceOptions = {}) { + this.intervalMs = options.intervalMs ?? DEFAULT_INTERVAL_MS; + this.capPerAgent = options.capPerAgent ?? DEFAULT_CAP_PER_AGENT; + } + + start(): void { + if (this.timer) return; + this.stopped = false; + this.timer = setInterval(() => { + this.run(); + }, this.intervalMs); + this.run(); + } + + stop(): void { + this.stopped = true; + if (this.timer) { + clearInterval(this.timer); + this.timer = null; + } + } + + /** Reclaim expired entries and enforce per-agent caps. Returns sweep stats. */ + run(): { swept: number; cappedAgents: number } { + if (this.stopped) return { swept: 0, cappedAgents: 0 }; + try { + const store = createErrorRegistryStore(getErrorDb()); + const swept = store.sweepExpired(); + + // Enforce the cap for every distinct active agent in the store. + const distinctAgents = getErrorDb() + .prepare("SELECT DISTINCT agentId AS id FROM errors WHERE status = 'active' AND agentId <> ''") + .all() as Array<{ id: string }>; + let cappedAgents = 0; + for (const row of distinctAgents) { + const before = store.countLiveByAgent(row.id); + if (before > this.capPerAgent) { + store.capLiveEntries(row.id, this.capPerAgent); + cappedAgents += 1; + } + } + + logger.info({ swept, cappedAgents }, "error-registry maintenance pass completed"); + return { swept, cappedAgents }; + } catch (error) { + logger.error({ err: error }, "error-registry maintenance pass failed"); + return { swept: 0, cappedAgents: 0 }; + } + } +} + +export { closeErrorDb }; diff --git a/frontend/src/components/wallet/WalletWizard.module.css b/frontend/src/components/wallet/WalletWizard.module.css index 0b256e6b..21ad1aa4 100644 --- a/frontend/src/components/wallet/WalletWizard.module.css +++ b/frontend/src/components/wallet/WalletWizard.module.css @@ -187,6 +187,29 @@ word-break: break-all; } +.balanceChips { + display: flex; + flex-wrap: wrap; + gap: 8px; + margin-top: 0.75rem; + justify-content: center; +} + +.balanceChip { + display: inline-flex; + align-items: center; + gap: 4px; + padding: 4px 10px; + background: #ffffff; + border: 1px solid var(--border, #e5e7eb); + border-radius: 999px; + font-size: 0.8125rem; + font-weight: 600; + color: var(--text-main, #111827); + font-variant-numeric: tabular-nums; + font-family: var(--font-mono, monospace); +} + .link { color: var(--primary, var(--accent-info)); text-decoration: none; diff --git a/frontend/src/components/wallet/WalletWizard.test.tsx b/frontend/src/components/wallet/WalletWizard.test.tsx index 3ed573dd..9232017f 100644 --- a/frontend/src/components/wallet/WalletWizard.test.tsx +++ b/frontend/src/components/wallet/WalletWizard.test.tsx @@ -34,6 +34,7 @@ describe('WalletWizard', () => { vi.spyOn(WalletBalanceHook, 'useWalletBalance').mockReturnValue({ balance: '0', + balances: [], loading: false, error: null, }) diff --git a/frontend/src/components/wallet/WalletWizard.tsx b/frontend/src/components/wallet/WalletWizard.tsx index 0f78422d..cd0847a1 100644 --- a/frontend/src/components/wallet/WalletWizard.tsx +++ b/frontend/src/components/wallet/WalletWizard.tsx @@ -9,7 +9,7 @@ import styles from './WalletWizard.module.css' export const WalletWizard: React.FC = () => { const { freighterAvailable, connectFreighter, connected, publicKey, completeWizard } = useWallet() - const { balance } = useWalletBalance(publicKey) + const { balance, balances } = useWalletBalance(publicKey) const { currentStep, nextStep, prevStep } = useOnboarding(4) const [connecting, setConnecting] = useState(false) @@ -148,8 +148,23 @@ export const WalletWizard: React.FC = () => {
-

Current Balance: {balance ? parseFloat(balance).toFixed(2) : '0.00'} XLM

+

Current Balance: {balance ? parseFloat(balance).toFixed(7) : '0.0000000'} XLM

+ {balances.length > 0 && ( +
+ {balances.map((entry) => { + const code = entry.asset_type === 'native' ? 'XLM' : (entry.asset_code ?? '') + return ( + + {new Intl.NumberFormat(undefined, { maximumFractionDigits: 7 }).format(parseFloat(entry.balance) || 0)} {code} + + ) + })} +
+ )}
+

+ Once funded, you can send XLM to other agents from your wallet. +

void; onConnect?: () => void; onDisconnect?: () => void; + /** + * Public key of the wallet that owns the task. When provided it is sent as + * the first frame after the socket opens (`{ walletPublicKey }`) so the + * server can authenticate the connection. See backend `AuthMessage`. + */ + walletPublicKey?: string; + /** Base host for the stream. Defaults to localhost:3001 (matching the server). */ + baseUrl?: string; + /** Maximum automatic reconnect attempts before giving up. Default 5. */ + maxReconnectAttempts?: number; + /** + * Maximum random jitter (ms) added to each reconnect backoff to avoid a + * thundering herd of clients reconnecting at once. Default 0 so the base + * exponential backoff stays deterministic in tests; enable in production. + */ + maxJitterMs?: number; } export type WebSocketStatus = 'connecting' | 'connected' | 'disconnected' | 'error'; +/** Application-defined close code for a stale socket: mirros backend WS_CLOSE.STALE. */ +const WS_CLOSE_STALE = 4408; +const DEFAULT_BASE_URL = 'ws://localhost:3001'; + +/** Build the event cursor only to be truthy for positive sequences. */ +const seqCursor = (seq: number | undefined): number | undefined => + typeof seq === 'number' && Number.isInteger(seq) && seq >= 0 ? seq : undefined; + +function cursorStorageKey(taskId: string): string { + return `ai-net:ws-cursor:${taskId}`; +} + export const useTaskWebSocket = (options: UseTaskWebSocketOptions) => { - const { taskId, onMessage, onConnect, onDisconnect } = options; + const { + taskId, + onMessage, + onConnect, + onDisconnect, + walletPublicKey, + baseUrl = DEFAULT_BASE_URL, + maxReconnectAttempts = 5, + maxJitterMs = 0, + } = options; + const [isConnected, setIsConnected] = useState(false); const [status, setStatus] = useState('connecting'); - + const wsRef = useRef(null); const reconnectAttemptRef = useRef(0); const reconnectTimeoutRef = useRef | null>(null); + // Guards against scheduling a reconnect after an intentional shutdown. + const manualCloseRef = useRef(false); + + // Latest event sequence seen for this task — used both to persist a resume + // cursor and to detect mid-stream gaps. + const lastSeqRef = useRef(-1); + const resumeCursorRef = useRef(-1); + const gapRetriedRef = useRef(false); + + // Handlers must see the latest onMessage/etc. across reconnects. + const onMessageRef = useRef(onMessage); + onMessageRef.current = onMessage; + const onConnectRef = useRef(onConnect); + onConnectRef.current = onConnect; + const onDisconnectRef = useRef(onDisconnect); + onDisconnectRef.current = onDisconnect; + + const clearReconnectTimeout = useCallback(() => { + if (reconnectTimeoutRef.current) { + clearTimeout(reconnectTimeoutRef.current); + reconnectTimeoutRef.current = null; + } + }, []); + const disconnect = useCallback(() => { + clearReconnectTimeout(); if (wsRef.current) { + manualCloseRef.current = true; wsRef.current.close(); wsRef.current = null; } - if (reconnectTimeoutRef.current) { - clearTimeout(reconnectTimeoutRef.current); - reconnectTimeoutRef.current = null; - } setIsConnected(false); setStatus('disconnected'); - }, []); + }, [clearReconnectTimeout]); const connectWebSocket = useCallback(() => { if (wsRef.current) { - wsRef.current.close(); + // Neutralise the previous socket so its own close/error handlers cannot + // schedule a reconnect while we're opening a fresh one. + const prev = wsRef.current; + prev.onopen = null; + prev.onmessage = null; + prev.onerror = null; + prev.onclose = null; + prev.close(); } setStatus('connecting'); setIsConnected(false); - - const wsUrl = `ws://localhost:3001/tasks/${taskId}/stream`; + manualCloseRef.current = false; + + // Resume from the last known cursor (persisted across reconnects & pages) + // by asking the server to replay events with seq > cursor. + const cursor = resumeCursorRef.current >= 0 ? resumeCursorRef.current : undefined; + const query = cursor !== undefined ? `?lastEventId=${cursor}` : ''; + const wsUrl = `${baseUrl}/tasks/${taskId}/stream${query}`; const ws = new WebSocket(wsUrl); wsRef.current = ws; ws.onopen = () => { setStatus('connected'); setIsConnected(true); - reconnectAttemptRef.current = 0; // reset reconnect attempts - onConnect?.(); + reconnectAttemptRef.current = 0; // reset reconnect attempts on a live socket + gapRetriedRef.current = false; + if (walletPublicKey) { + ws.send(JSON.stringify({ walletPublicKey })); + } + onConnectRef.current?.(); }; ws.onmessage = (event) => { + let data: DAGEvent; try { - const data: DAGEvent = JSON.parse(event.data); - onMessage(data); + data = JSON.parse(event.data); } catch (err) { console.error('Failed to parse WebSocket event:', err); + return; + } + + // Heartbeat pong: the server sends `{ type: 'ping' }` every interval and + // closes the socket (code 4408) unless we answer `{ type: 'pong' }`. + if ((data as { type?: unknown }).type === 'ping') { + ws.send(JSON.stringify({ type: 'pong' })); + return; + } + + const seq = seqCursor(data.seq); + if (seq !== undefined) { + const expected = lastSeqRef.current + 1; + // A forward jump in seq means we missed events. Trigger one reconnect + // with the resume cursor so the server replays the gap. If the gap + // persists (e.g. the store pruned intermediate events) we accept the + // stream rather than reconnect-loop forever. + if (seq > expected && expected > 0 && !gapRetriedRef.current && ws.readyState === WebSocket.OPEN) { + gapRetriedRef.current = true; + lastSeqRef.current = seq; + resumeCursorRef.current = seq; + try { + sessionStorage.setItem(cursorStorageKey(taskId), String(seq)); + } catch { /* non-fatal */ } + // Neutralise the current socket's handlers so its own close does not + // schedule an extra reconnect — we re-sync via the cursor below. + ws.onopen = null; + ws.onmessage = null; + ws.onerror = null; + ws.onclose = null; + connectWebSocket(); + return; + } + lastSeqRef.current = seq; + resumeCursorRef.current = seq; + try { + sessionStorage.setItem(cursorStorageKey(taskId), String(seq)); + } catch { /* non-fatal */ } } + + onMessageRef.current(data); }; ws.onerror = () => { @@ -65,21 +181,37 @@ export const useTaskWebSocket = (options: UseTaskWebSocketOptions) => { setIsConnected(false); }; - ws.onclose = () => { + ws.onclose = (event) => { setStatus('disconnected'); setIsConnected(false); - onDisconnect?.(); - - // Reconnect with exponential backoff (max 5 attempts) - if (reconnectAttemptRef.current < 5) { - const delay = 1000 * Math.pow(2, reconnectAttemptRef.current); + onDisconnectRef.current?.(); + + // Never auto-reconnect after an intentional shutdown (component unmount + // or explicit disconnect()). + if (manualCloseRef.current) { + return; + } + + // A 4408 (STALE) close means the server killed the socket because we did + // not answer a heartbeat in time. Reset the backoff so we reconnect + // promptly rather than waiting out an ever-growing delay. + if (event.code === WS_CLOSE_STALE) { + reconnectAttemptRef.current = 0; + } + + if (reconnectAttemptRef.current < maxReconnectAttempts) { + const baseDelay = 1000 * Math.pow(2, reconnectAttemptRef.current); + const jitter = maxJitterMs > 0 ? Math.floor(Math.random() * maxJitterMs) : 0; + const delay = Math.max(0, baseDelay + jitter); reconnectAttemptRef.current += 1; + clearReconnectTimeout(); reconnectTimeoutRef.current = setTimeout(() => { + reconnectTimeoutRef.current = null; connectWebSocket(); }, delay); } }; - }, [taskId, onMessage, onConnect, onDisconnect]); + }, [taskId, baseUrl, walletPublicKey, maxReconnectAttempts, maxJitterMs, clearReconnectTimeout]); const reconnect = useCallback(() => { reconnectAttemptRef.current = 0; @@ -89,6 +221,18 @@ export const useTaskWebSocket = (options: UseTaskWebSocketOptions) => { useEffect(() => { if (!taskId) return; + // Restore the persisted cursor for this task so a page reload resumes. + try { + const saved = sessionStorage.getItem(cursorStorageKey(taskId)); + if (saved !== null) { + const n = Number(saved); + if (Number.isInteger(n) && n >= 0) { + lastSeqRef.current = n; + resumeCursorRef.current = n; + } + } + } catch { /* non-fatal */ } + connectWebSocket(); return () => { diff --git a/frontend/src/hooks/useWalletBalance.ts b/frontend/src/hooks/useWalletBalance.ts index 1199db01..66267912 100644 --- a/frontend/src/hooks/useWalletBalance.ts +++ b/frontend/src/hooks/useWalletBalance.ts @@ -3,14 +3,30 @@ import { useState, useEffect, useRef, useCallback } from 'react' const HORIZON_URL = 'https://horizon-testnet.stellar.org' const POLL_INTERVAL = 10_000 +export interface WalletBalance { + asset_type: 'native' | 'credit_alphanum4' | 'credit_alphanum12' + asset_code?: string + asset_issuer?: string + balance: string +} + interface BalanceInfo { + /** Native XLM balance (kept for backward compatibility). */ balance: string + /** Full set of on-chain balances including native XLM and issued tokens. */ + balances: WalletBalance[] loading: boolean error: string | null } +/** + * Fetches the full balance set for a Stellar account from Horizon. The native + * XLM balance is always present for a funded account; additional trustlines + * surface as `credit_alphanum4`/`credit_alphanum12` issued-asset entries. + */ export function useWalletBalance(publicKey: string | null): BalanceInfo { const [balance, setBalance] = useState('0') + const [balances, setBalances] = useState([]) const [loading, setLoading] = useState(true) const [error, setError] = useState(null) const isFirstLoad = useRef(true) @@ -23,6 +39,7 @@ export function useWalletBalance(publicKey: string | null): BalanceInfo { const key = keyRef.current if (!key) { setBalance('0') + setBalances([]) setError(null) return } @@ -36,16 +53,17 @@ export function useWalletBalance(publicKey: string | null): BalanceInfo { if (!res.ok) { if (res.status === 404) { setBalance('0') + setBalances([]) setError(null) return } throw new Error(`Horizon error: ${res.status}`) } const data = await res.json() - const xlmBalance = data.balances?.find( - (b: { asset_type: string }) => b.asset_type === 'native' - ) + const rawBalances: WalletBalance[] = Array.isArray(data.balances) ? data.balances : [] + const xlmBalance = rawBalances.find((b) => b.asset_type === 'native') setBalance(xlmBalance?.balance ?? '0') + setBalances(rawBalances) setError(null) isFirstLoad.current = false } catch (err) { @@ -69,5 +87,5 @@ export function useWalletBalance(publicKey: string | null): BalanceInfo { } }, [fetchBalance]) - return { balance, loading, error } + return { balance, balances, loading, error } } diff --git a/frontend/src/i18n/locales/en.json b/frontend/src/i18n/locales/en.json index 1df1e9d6..2c995838 100644 --- a/frontend/src/i18n/locales/en.json +++ b/frontend/src/i18n/locales/en.json @@ -115,6 +115,8 @@ "wallet.chart.noData": "No spending data yet", "wallet.chart.spent": "Spent", + "wallet.tokens.heading": "Tokens", + "validation.invalidSecretKey": "Invalid Stellar secret key. Must start with S and be 56 characters.", "validation.destinationRequired": "Destination address is required", "validation.invalidStellarAddress": "Invalid Stellar address. Must start with G and be 56 characters.", @@ -274,6 +276,7 @@ "a11y.search": "Search", "a11y.closeSearch": "Close search", "a11y.copyPublicKey": "Copy public key", + "a11y.balanceChip": "{{code}} balance chip", "a11y.disconnectWallet": "Disconnect wallet", "a11y.copyOutput": "Copy Output", "a11y.loadingTaskDetails": "Loading task details", diff --git a/frontend/src/i18n/locales/zh.json b/frontend/src/i18n/locales/zh.json index eee663d3..dfd0a1a2 100644 --- a/frontend/src/i18n/locales/zh.json +++ b/frontend/src/i18n/locales/zh.json @@ -115,6 +115,8 @@ "wallet.chart.noData": "暂无支出数据", "wallet.chart.spent": "支出", + "wallet.tokens.heading": "代币", + "validation.invalidSecretKey": "无效的 Stellar 私钥。必须以 S 开头且为 56 个字符。", "validation.destinationRequired": "目标地址为必填项", "validation.invalidStellarAddress": "无效的 Stellar 地址。必须以 G 开头且为 56 个字符。", @@ -274,6 +276,7 @@ "a11y.search": "搜索", "a11y.closeSearch": "关闭搜索", "a11y.copyPublicKey": "复制公钥", + "a11y.balanceChip": "{{code}} 余额标签", "a11y.disconnectWallet": "断开钱包连接", "a11y.copyOutput": "复制输出", "a11y.loadingTaskDetails": "正在加载任务详情", diff --git a/frontend/src/pages/WalletPage.module.css b/frontend/src/pages/WalletPage.module.css index 67202980..ec7f08ef 100644 --- a/frontend/src/pages/WalletPage.module.css +++ b/frontend/src/pages/WalletPage.module.css @@ -233,6 +233,58 @@ opacity: 0.75; } +/* Balance chips (XLM + token trustlines) */ +.balanceChips { + display: flex; + flex-wrap: wrap; + gap: 8px; + margin-bottom: 20px; +} + +.balanceChip { + display: inline-flex; + align-items: center; + gap: 6px; + padding: 6px 12px; + background: rgba(255, 255, 255, 0.15); + border: 1px solid rgba(255, 255, 255, 0.2); + border-radius: 999px; + font-size: 0.875rem; + color: inherit; +} + +.balanceChipNative { + background: rgba(255, 255, 255, 0.25); + border-color: rgba(255, 255, 255, 0.35); +} + +.balanceChipAmount { + font-family: var(--font-mono, monospace); + font-variant-numeric: tabular-nums; + font-weight: 600; +} + +.balanceChipCode { + font-weight: 600; + opacity: 0.85; + text-transform: uppercase; + letter-spacing: 0.03em; + font-size: 0.75rem; +} + +.tokensSection { + margin-bottom: 20px; +} + +.tokensHeading { + font-size: 0.75rem; + font-weight: 600; + opacity: 0.85; + margin: 0 0 8px; + text-transform: uppercase; + letter-spacing: 0.05em; +} + .publicKeySection { display: flex; align-items: center; diff --git a/frontend/src/pages/WalletPage.test.tsx b/frontend/src/pages/WalletPage.test.tsx index 1feeee9d..d777e9c0 100644 --- a/frontend/src/pages/WalletPage.test.tsx +++ b/frontend/src/pages/WalletPage.test.tsx @@ -46,11 +46,21 @@ vi.mock('../context/WalletContext', () => ({ vi.mock('../hooks/useWalletBalance', () => ({ useWalletBalance: () => ({ balance: '100.0000000', + balances: [ + { asset_type: 'native', balance: '100.0000000' }, + { asset_type: 'credit_alphanum4', asset_code: 'USDC', asset_issuer: 'G...', balance: '25.5000000' }, + ], loading: false, error: null, }), })) +const mockShowToast = vi.fn() + +vi.mock('../hooks/useToast', () => ({ + useToast: () => ({ showToast: mockShowToast }), +})) + // Only `useTransactionHistory` (the data-fetching hook) is mocked here - the // module also exports pure filtering/aggregation helpers that PaymentChart and // TransactionTable call directly, so those must stay the real implementations. @@ -143,7 +153,23 @@ describe('WalletPage - Connected State', () => { it('shows the balance when connected', () => { renderPage() expect(screen.getByText('100.0000000')).toBeInTheDocument() - expect(screen.getByText('XLM')).toBeInTheDocument() + expect(screen.getAllByText('XLM').length).toBeGreaterThan(0) + }) + + it('renders balance chips for XLM and token trustlines', () => { + renderPage() + expect(screen.getByText('USDC')).toBeInTheDocument() + expect(screen.getByText('25.5')).toBeInTheDocument() + expect(screen.getAllByText('XLM').length).toBeGreaterThan(0) + }) + + it('shows a toast when copying the address', async () => { + renderPage() + const copyBtn = screen.getByRole('button', { name: /copy public key/i }) + fireEvent.click(copyBtn) + await waitFor(() => { + expect(mockShowToast).toHaveBeenCalled() + }) }) it('shows the disconnect button', () => { diff --git a/frontend/src/pages/WalletPage.tsx b/frontend/src/pages/WalletPage.tsx index c055c554..d9fdc281 100644 --- a/frontend/src/pages/WalletPage.tsx +++ b/frontend/src/pages/WalletPage.tsx @@ -1,9 +1,10 @@ import React, { useMemo } from 'react' import { useTranslation, Trans } from 'react-i18next' import { QRCodeSVG } from 'qrcode.react' -import { Wallet, Copy, Check, ExternalLink, Download } from 'lucide-react' +import { Wallet, Copy, ExternalLink, Download } from 'lucide-react' import { useWallet } from '../context/WalletContext' import { useWalletBalance } from '../hooks/useWalletBalance' +import { useToast } from '../hooks/useToast' import { useTransactionHistory } from '../hooks/useTransactionHistory' import { SendXLMForm } from '../components/wallet/SendXLMForm' import { PaymentChart } from '../components/wallet/PaymentChart' @@ -60,9 +61,9 @@ export function WalletPageSkeleton() { function WalletPage() { const { t } = useTranslation() const { publicKey, connected, ready, connectionMethod, freighterAvailable, connect, connectFreighter, disconnect, hasCompletedWizard } = useWallet() - const { balance, loading: balanceLoading, error: balanceError } = useWalletBalance(publicKey) + const { balance, balances, loading: balanceLoading, error: balanceError } = useWalletBalance(publicKey) + const { showToast } = useToast() const { transactions, loading: txLoading, error: txError } = useTransactionHistory(publicKey) - const [copied, setCopied] = React.useState(false) const [secretInput, setSecretInput] = React.useState('') const [connectError, setConnectError] = React.useState(null) const [connecting, setConnecting] = React.useState(false) @@ -70,22 +71,25 @@ function WalletPage() { const [freighterError, setFreighterError] = React.useState(null) const handleCopyAddress = async () => { - if (publicKey) { - try { + if (!publicKey) return + const fallbackCopy = () => { + const textArea = document.createElement('textarea') + textArea.value = publicKey! + document.body.appendChild(textArea) + textArea.select() + document.execCommand('copy') + document.body.removeChild(textArea) + } + try { + if (navigator.clipboard && window.isSecureContext) { await navigator.clipboard.writeText(publicKey) - setCopied(true) - setTimeout(() => setCopied(false), 2000) - } catch { - const textArea = document.createElement('textarea') - textArea.value = publicKey - document.body.appendChild(textArea) - textArea.select() - document.execCommand('copy') - document.body.removeChild(textArea) - setCopied(true) - setTimeout(() => setCopied(false), 2000) + } else { + fallbackCopy() } + } catch { + fallbackCopy() } + showToast(t('wallet.copyAddress'), 'success') } const handleConnect = async (e: React.FormEvent) => { @@ -131,6 +135,40 @@ function WalletPage() { ) }, [balance, balanceLoading, balanceError]) + const balanceChips = useMemo(() => { + const tokenBalances = balances.filter((entry) => entry.asset_type !== 'native') + if (balanceLoading) { + return
+ } + const chips = balances.map((entry) => { + const code = entry.asset_type === 'native' ? 'XLM' : (entry.asset_code ?? '') + return ( + + + {new Intl.NumberFormat(undefined, { maximumFractionDigits: 7 }).format(parseFloat(entry.balance) || 0)} + + {code} + + ) + }) + if (chips.length > 1 && tokenBalances.length > 0) { + return ( +
+

{t('wallet.tokens.heading')}

+
{chips}
+
+ ) + } + if (chips.length > 0) { + return
{chips}
+ } + return null + }, [balances, balanceLoading, t]) + if (!hasCompletedWizard) { return (
@@ -306,6 +344,8 @@ function WalletPage() { {balanceDisplay}
+ {balanceChips} +
@@ -319,9 +359,10 @@ function WalletPage() {