feat: real-time live activity feed for platform-wide registrations & claims

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(

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);

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);
+  });
+});

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
    */

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;
+  }
+}