diff --git a/backend/src/index.js b/backend/src/index.js index 9b6ea2b6..6371d94c 100644 --- a/backend/src/index.js +++ b/backend/src/index.js @@ -578,6 +578,9 @@ export async function createApp(options = {}) { const eventIndexer = createEventIndexer({ db: dal.db, + // Live activity feed (#1201): resolved lazily because the WebSocket + // server is attached after the HTTP server starts listening. + onActivity: (activity) => getWebSocketServer()?.publishActivity(activity), rpcPool, logger: log, referralBonus: normalizePositiveInteger( diff --git a/backend/src/jobs/eventIndexer.js b/backend/src/jobs/eventIndexer.js index 4fd128f2..bd6307fb 100644 --- a/backend/src/jobs/eventIndexer.js +++ b/backend/src/jobs/eventIndexer.js @@ -126,6 +126,10 @@ export function createEventIndexer({ referralBonus = 0, confirmationDepth = 0, notificationService, + // Called after a claim/register event is projected, for the platform-wide + // live activity feed (#1201). Errors are swallowed so the feed can never + // stall indexing. + onActivity, } = {}) { const sql = createSqlAdapter(db); const depth = Math.max(0, Math.floor(Number(confirmationDepth)) || 0); @@ -147,13 +151,19 @@ export function createEventIndexer({ const handlers = { credit: (event, db) => handleCreditEvent(event, db, notificationService), - claim: (event, db) => handleClaimEvent(event, db, notificationService), + claim: async (event, db) => { + await handleClaimEvent(event, db, notificationService); + emitActivity(onActivity, 'claim', event, logger); + }, snapshot: handleSnapshotEvent, vcredit: (event, db) => handleVestedCreditEvent(event, db, notificationService), vclaim: (event, db) => handleVestedClaimEvent(event, db, notificationService), referred: (event, database) => handleReferredEvent(event, database, referralBonus), refbonus: handleRefBonusEvent, - register: handleRegisterEvent, + register: async (event, db) => { + await handleRegisterEvent(event, db); + emitActivity(onActivity, 'registration', event, logger); + }, deregister: handleDeregisterEvent, }; @@ -716,6 +726,29 @@ function applyBalanceDelta(db, user, delta) { ); } +/** + * Forward a projected event to the live activity feed (#1201). + * claim topics: [name, user, campaignId?], data = amount + * register topics: [name, user, campaignId] + */ +export function emitActivity(onActivity, kind, event, logger = console) { + if (typeof onActivity !== 'function') return; + const wallet = event.topic?.[1]; + if (!wallet) return; + try { + onActivity({ + kind, + wallet: String(wallet), + campaignId: event.topic?.[2] != null ? String(event.topic[2]) : null, + amount: kind === 'claim' ? String(event.data ?? 0) : null, + ledger: event.ledger ?? null, + txHash: event.txHash ?? null, + }); + } catch (err) { + logger.warn?.({ err, kind }, 'live activity publish failed'); + } +} + async function handleCreditEvent(event, db, notificationService) { const user = event.topic?.[1]; const amount = BigInt(event.data ?? 0); diff --git a/backend/src/websocket/activity-feed.test.js b/backend/src/websocket/activity-feed.test.js new file mode 100644 index 00000000..22d3cd5c --- /dev/null +++ b/backend/src/websocket/activity-feed.test.js @@ -0,0 +1,112 @@ +// Live activity feed (#1201): indexer -> WebSocket `activity` room. +import { describe, it, before, after } from 'node:test'; +import assert from 'node:assert'; +import { createServer } from 'node:http'; +import WebSocket from 'ws'; +import WebSocketServer, { ACTIVITY_ROOM, maskWallet } from './server.js'; +import { emitActivity } from '../jobs/eventIndexer.js'; + +const WALLET = 'GABCDEFGHIJKLMNOPQRSTUVWXYZ234567ABCDEFGHIJKLMNOPQRSTUVW'; + +function nextMessage(ws, type) { + return new Promise((resolve) => { + const onMessage = (raw) => { + const data = JSON.parse(raw.toString()); + if (data.type === type) { + ws.off('message', onMessage); + resolve(data); + } + }; + ws.on('message', onMessage); + }); +} + +describe('maskWallet', () => { + it('masks long addresses and leaves short values alone', () => { + assert.strictEqual(maskWallet(WALLET), 'GABC…TUVW'); + assert.strictEqual(maskWallet('short'), 'short'); + assert.strictEqual(maskWallet(null), ''); + }); +}); + +describe('emitActivity', () => { + it('maps claim events to activity payloads', () => { + const seen = []; + emitActivity((a) => seen.push(a), 'claim', { + topic: ['claim', WALLET, 'camp-1'], + data: 50000000n, + ledger: 42, + txHash: 'abc', + }); + assert.deepStrictEqual(seen, [ + { kind: 'claim', wallet: WALLET, campaignId: 'camp-1', amount: '50000000', ledger: 42, txHash: 'abc' }, + ]); + }); + + it('omits amount for registrations and skips events without a wallet', () => { + const seen = []; + emitActivity((a) => seen.push(a), 'registration', { topic: ['register', WALLET, 7] }); + emitActivity((a) => seen.push(a), 'registration', { topic: ['register'] }); + assert.strictEqual(seen.length, 1); + assert.strictEqual(seen[0].amount, null); + assert.strictEqual(seen[0].campaignId, '7'); + }); + + it('is a no-op without a callback and never throws from the callback', () => { + emitActivity(undefined, 'claim', { topic: ['claim', WALLET] }); + emitActivity( + () => { + throw new Error('boom'); + }, + 'claim', + { topic: ['claim', WALLET] }, + { warn() {} }, + ); + }); +}); + +describe('WebSocketServer.publishActivity', () => { + let httpServer; + let wsServer; + let wsUrl; + + before(async () => { + httpServer = createServer(); + await new Promise((resolve) => httpServer.listen(0, resolve)); + wsUrl = `ws://localhost:${httpServer.address().port}/ws`; + wsServer = new WebSocketServer(httpServer, { path: '/ws' }); + }); + + after(() => { + wsServer.close(); + httpServer.close(); + }); + + it('broadcasts masked activity to `activity` subscribers', async () => { + const ws = new WebSocket(wsUrl); + await nextMessage(ws, 'connected'); + ws.send(JSON.stringify({ type: 'subscribe', channel: ACTIVITY_ROOM })); + await nextMessage(ws, 'subscribed'); + + const received = nextMessage(ws, 'activity'); + const sent = wsServer.publishActivity({ + kind: 'claim', + wallet: WALLET, + campaignId: 'camp-1', + amount: '10', + txHash: 'tx1', + }); + const msg = await received; + assert.strictEqual(sent, 1); + assert.strictEqual(msg.kind, 'claim'); + assert.strictEqual(msg.wallet, 'GABC…TUVW'); + assert.strictEqual(msg.id, 'claim:tx1'); + assert.ok(!JSON.stringify(msg).includes(WALLET), 'full wallet must not be broadcast'); + ws.close(); + }); + + it('ignores unknown kinds', () => { + assert.strictEqual(wsServer.publishActivity({ kind: 'transfer', wallet: WALLET }), 0); + assert.strictEqual(wsServer.publishActivity(null), 0); + }); +}); diff --git a/backend/src/websocket/server.js b/backend/src/websocket/server.js index b793b832..dbb7d21b 100644 --- a/backend/src/websocket/server.js +++ b/backend/src/websocket/server.js @@ -6,6 +6,14 @@ import { log as logger } from '../middleware/logger.js'; * WebSocket server for real-time updates * Handles: campaign updates, participant changes, reward notifications */ +export const ACTIVITY_ROOM = 'activity'; + +/** GABCD…WXYZ — keeps the feed social without broadcasting full addresses. */ +export function maskWallet(wallet) { + const value = String(wallet ?? ''); + return value.length > 10 ? `${value.slice(0, 4)}…${value.slice(-4)}` : value; +} + class WebSocketServer extends EventEmitter { constructor(httpServer, options = {}) { super(); @@ -288,6 +296,32 @@ class WebSocketServer extends EventEmitter { this.broadcastToRooms([walletRoomId, campaignRoomId], message); } + /** + * Publish a platform-wide activity event (registration or reward claim) to + * every client subscribed to the `activity` channel (#1201). Wallets are + * masked before leaving the server. + * @param {{kind: 'registration'|'claim', wallet: string, campaignId?: string|null, + * amount?: string|null, ledger?: number|null, txHash?: string|null}} activity + * @returns {number} number of clients the event was sent to + */ + publishActivity(activity) { + if (!activity || !['registration', 'claim'].includes(activity.kind)) return 0; + const message = { + type: 'activity', + // Never embed the full wallet here -- the id is sent to every client. + id: activity.txHash + ? `${activity.kind}:${activity.txHash}` + : `${activity.kind}:${Date.now()}:${Math.random().toString(36).slice(2, 8)}`, + kind: activity.kind, + wallet: maskWallet(activity.wallet), + campaignId: activity.campaignId ?? null, + amount: activity.amount ?? null, + ledger: activity.ledger ?? null, + timestamp: new Date().toISOString(), + }; + return this.broadcast(ACTIVITY_ROOM, message); + } + /** * Notify about reward claimed */ diff --git a/frontend/src/components/LiveActivityFeed.css b/frontend/src/components/LiveActivityFeed.css new file mode 100644 index 00000000..5f00b052 --- /dev/null +++ b/frontend/src/components/LiveActivityFeed.css @@ -0,0 +1,112 @@ +/* Live activity feed (#1201) */ +.live-activity { + border: 1px solid var(--border, rgba(127, 127, 127, 0.25)); + border-radius: 16px; + padding: 1rem 1.25rem; + background: var(--surface, transparent); +} + +.live-activity__header { + display: flex; + align-items: center; + justify-content: space-between; + gap: 0.75rem; + margin-bottom: 0.75rem; +} + +.live-activity__title { + margin: 0; + font-size: 1.05rem; +} + +.live-activity__status { + display: inline-flex; + align-items: center; + gap: 0.4rem; + font-size: 0.8rem; + font-weight: 600; +} + +.live-activity__status::before { + content: ''; + width: 8px; + height: 8px; + border-radius: 50%; + background: currentColor; +} + +.live-activity__status.is-live { + color: var(--success, #16a34a); +} + +.live-activity__status.is-live::before { + animation: live-activity-pulse 1.6s ease-in-out infinite; +} + +.live-activity__status.is-offline { + color: var(--muted, #6b7280); +} + +.live-activity__empty { + margin: 0; + color: var(--muted, #6b7280); + font-size: 0.9rem; +} + +.live-activity__list { + list-style: none; + margin: 0; + padding: 0; + max-height: 18rem; + overflow-y: auto; +} + +.live-activity__item { + display: grid; + grid-template-columns: auto 1fr auto; + align-items: center; + gap: 0.6rem; + padding: 0.5rem 0; + border-top: 1px solid var(--border, rgba(127, 127, 127, 0.15)); + font-size: 0.9rem; + animation: live-activity-in 0.35s ease-out; +} + +.live-activity__item:first-child { + border-top: 0; +} + +.live-activity__text { + min-width: 0; + overflow-wrap: anywhere; +} + +.live-activity__time { + color: var(--muted, #6b7280); + font-size: 0.75rem; + white-space: nowrap; +} + +@keyframes live-activity-in { + from { + opacity: 0; + transform: translateY(-6px); + } + to { + opacity: 1; + transform: none; + } +} + +@keyframes live-activity-pulse { + 50% { + opacity: 0.35; + } +} + +@media (prefers-reduced-motion: reduce) { + .live-activity__item, + .live-activity__status.is-live::before { + animation: none; + } +} diff --git a/frontend/src/components/LiveActivityFeed.jsx b/frontend/src/components/LiveActivityFeed.jsx new file mode 100644 index 00000000..77859df7 --- /dev/null +++ b/frontend/src/components/LiveActivityFeed.jsx @@ -0,0 +1,94 @@ +import { useEffect, useState } from 'react'; +import { getWebSocketClient } from '../lib/websocket'; +import { + ACTIVITY_CHANNEL, + addActivity, + describeActivity, + formatRelativeTime, + normalizeActivity, +} from '../lib/activityFeed'; +import './LiveActivityFeed.css'; + +/** + * LiveActivityFeed — platform-wide ticker of recent registrations and reward + * claims, streamed over the `activity` WebSocket channel (#1201). + * + * @param {object} props + * @param {object} [props.client] WebSocket client (defaults to the app singleton; injectable for tests) + * @param {number} [props.maxItems] Items kept in the feed (default 20) + */ +export default function LiveActivityFeed({ client, maxItems }) { + const [items, setItems] = useState([]); + const [connected, setConnected] = useState(false); + const [now, setNow] = useState(() => Date.now()); + + useEffect(() => { + let ws; + try { + ws = client || getWebSocketClient(); + } catch { + return undefined; + } + const offs = [ + ws.on('connected', () => setConnected(true)), + ws.on('disconnected', () => setConnected(false)), + ws.on('activity', (message) => { + const item = normalizeActivity(message); + if (item) setItems((prev) => addActivity(prev, item, maxItems)); + }), + ]; + if (ws.isConnected) setConnected(true); + else ws.connect?.(); + ws.subscribe(ACTIVITY_CHANNEL); + + return () => { + offs.forEach((off) => off()); + ws.unsubscribe?.(ACTIVITY_CHANNEL); + }; + }, [client, maxItems]); + + // Keep relative timestamps fresh without re-rendering on every frame. + useEffect(() => { + const timer = setInterval(() => setNow(Date.now()), 15_000); + return () => clearInterval(timer); + }, []); + + return ( +
+
+

+ Live activity +

+ + {connected ? 'Live' : 'Connecting…'} + +
+ + {items.length === 0 ? ( +

+ Waiting for registrations and reward claims across Trivela… +

+ ) : ( + + )} +
+ ); +} diff --git a/frontend/src/lib/activityFeed.js b/frontend/src/lib/activityFeed.js new file mode 100644 index 00000000..93713c6c --- /dev/null +++ b/frontend/src/lib/activityFeed.js @@ -0,0 +1,58 @@ +// Live activity feed state helpers (#1201). Pure so they can be unit tested +// without a WebSocket. + +export const ACTIVITY_CHANNEL = 'activity'; +export const MAX_FEED_ITEMS = 20; +const KINDS = new Set(['registration', 'claim']); +const STROOPS_PER_UNIT = 10_000_000; + +/** + * Validate an incoming `activity` WebSocket message. Returns a normalized item + * or null for anything malformed, so a bad frame can never break the widget. + */ +export function normalizeActivity(message) { + if (!message || message.type !== 'activity' || !KINDS.has(message.kind)) return null; + if (typeof message.id !== 'string' || !message.id) return null; + if (typeof message.wallet !== 'string' || !message.wallet) return null; + const time = Date.parse(message.timestamp); + return { + id: message.id, + kind: message.kind, + wallet: message.wallet, + campaignId: message.campaignId != null ? String(message.campaignId) : null, + amount: message.amount != null ? String(message.amount) : null, + timestamp: Number.isNaN(time) ? Date.now() : time, + }; +} + +/** Newest first, de-duplicated by id, capped at `max` items. */ +export function addActivity(items, item, max = MAX_FEED_ITEMS) { + if (!item || items.some((existing) => existing.id === item.id)) return items; + return [item, ...items].slice(0, max); +} + +/** Contract amounts are i128 stroops (7 decimals); show up to 2 decimals. */ +export function formatActivityAmount(amount) { + if (amount == null || amount === '') return null; + const value = Number(amount) / STROOPS_PER_UNIT; + if (!Number.isFinite(value)) return null; + return value.toLocaleString(undefined, { maximumFractionDigits: 2 }); +} + +export function describeActivity(item) { + const campaign = item.campaignId ? ` campaign #${item.campaignId}` : ' a campaign'; + if (item.kind === 'claim') { + const amount = formatActivityAmount(item.amount); + return `${item.wallet} claimed ${amount ? `${amount} points` : 'rewards'}${item.campaignId ? ` from${campaign}` : ''}`; + } + return `${item.wallet} joined${campaign}`; +} + +export function formatRelativeTime(timestamp, now = Date.now()) { + const seconds = Math.max(0, Math.round((now - timestamp) / 1000)); + if (seconds < 5) return 'just now'; + if (seconds < 60) return `${seconds}s ago`; + const minutes = Math.floor(seconds / 60); + if (minutes < 60) return `${minutes}m ago`; + return `${Math.floor(minutes / 60)}h ago`; +}