Skip to content

Commit 9c63f81

Browse files
ogazboizogazboiz
authored
Merge pull request #435 from davedumto/feat/stream-pause-resume-and-topup-modal
feat: top-up modal, pause/resume tracker, activity timeline, admin metrics
2 parents a761f43 + 3d38457 commit 9c63f81

8 files changed

Lines changed: 752 additions & 118 deletions

File tree

‎backend/src/routes/v1/admin.routes.ts‎

Lines changed: 122 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -28,84 +28,133 @@ router.use(requireAdmin);
2828
* 200:
2929
* description: Protocol health metrics
3030
*/
31-
router.get('/metrics', async (_req: Request, res: Response) => {
32-
try {
33-
const since24h = new Date(Date.now() - 24 * 60 * 60 * 1000);
34-
35-
const [activeCount, totalCount, cancelledCount, completedCount, eventsLast24h, indexerState, feeEvents, feesLast24h] =
36-
await Promise.all([
37-
prisma.stream.count({ where: { isActive: true } }),
38-
prisma.stream.count(),
39-
prisma.stream.count({
40-
where: { isActive: false, events: { some: { eventType: 'CANCELLED' } } },
41-
}),
42-
prisma.stream.count({
43-
where: { isActive: false, events: { some: { eventType: 'COMPLETED' } } },
44-
}),
45-
prisma.streamEvent.count({ where: { createdAt: { gte: since24h } } }),
46-
prisma.indexerState.findUnique({ where: { id: 'singleton' } }),
47-
prisma.streamEvent.findMany({
48-
where: { eventType: 'FEE_COLLECTED' },
49-
select: { amount: true, metadata: true },
50-
}),
51-
prisma.streamEvent.findMany({
52-
where: { eventType: 'FEE_COLLECTED', createdAt: { gte: since24h } },
53-
select: { amount: true, metadata: true },
54-
}),
55-
]);
56-
57-
// Aggregate fees by token
58-
const totalFeesCollectedByToken: Record<string, string> = {};
59-
const feesLast24hByToken: Record<string, string> = {};
60-
61-
for (const event of feeEvents) {
62-
const metadata = event.metadata ? JSON.parse(event.metadata) : {};
63-
const token = metadata.token || 'unknown';
64-
const amount = BigInt(event.amount || '0');
65-
totalFeesCollectedByToken[token] = (
66-
BigInt(totalFeesCollectedByToken[token] || '0') + amount
67-
).toString();
68-
}
31+
const ADMIN_METRICS_CACHE_KEY = 'admin:metrics';
32+
const ADMIN_METRICS_CACHE_TTL_SECONDS = 60;
6933

70-
for (const event of feesLast24h) {
71-
const metadata = event.metadata ? JSON.parse(event.metadata) : {};
72-
const token = metadata.token || 'unknown';
73-
const amount = BigInt(event.amount || '0');
74-
feesLast24hByToken[token] = (
75-
BigInt(feesLast24hByToken[token] || '0') + amount
76-
).toString();
77-
}
34+
async function buildAdminMetrics() {
35+
const since24h = new Date(Date.now() - 24 * 60 * 60 * 1000);
36+
37+
const [
38+
activeCount,
39+
pausedCount,
40+
totalCount,
41+
cancelledCount,
42+
completedCount,
43+
eventsLast24h,
44+
indexerState,
45+
feeEvents,
46+
feesLast24h,
47+
withdrawnSums,
48+
] = await Promise.all([
49+
prisma.stream.count({ where: { isActive: true } }),
50+
prisma.stream.count({ where: { isPaused: true } }),
51+
prisma.stream.count(),
52+
prisma.stream.count({
53+
where: { isActive: false, events: { some: { eventType: 'CANCELLED' } } },
54+
}),
55+
prisma.stream.count({
56+
where: { isActive: false, events: { some: { eventType: 'COMPLETED' } } },
57+
}),
58+
prisma.streamEvent.count({ where: { createdAt: { gte: since24h } } }),
59+
prisma.indexerState.findUnique({ where: { id: 'singleton' } }),
60+
prisma.streamEvent.findMany({
61+
where: { eventType: 'FEE_COLLECTED' },
62+
select: { amount: true, metadata: true },
63+
}),
64+
prisma.streamEvent.findMany({
65+
where: { eventType: 'FEE_COLLECTED', createdAt: { gte: since24h } },
66+
select: { amount: true, metadata: true },
67+
}),
68+
prisma.stream.findMany({ select: { withdrawnAmount: true } }),
69+
]);
70+
71+
// Aggregate fees by token
72+
const totalFeesCollectedByToken: Record<string, string> = {};
73+
const feesLast24hByToken: Record<string, string> = {};
74+
75+
for (const event of feeEvents) {
76+
const metadata = event.metadata ? JSON.parse(event.metadata) : {};
77+
const token = metadata.token || 'unknown';
78+
const amount = BigInt(event.amount || '0');
79+
totalFeesCollectedByToken[token] = (
80+
BigInt(totalFeesCollectedByToken[token] || '0') + amount
81+
).toString();
82+
}
83+
84+
for (const event of feesLast24h) {
85+
const metadata = event.metadata ? JSON.parse(event.metadata) : {};
86+
const token = metadata.token || 'unknown';
87+
const amount = BigInt(event.amount || '0');
88+
feesLast24hByToken[token] = (
89+
BigInt(feesLast24hByToken[token] || '0') + amount
90+
).toString();
91+
}
7892

79-
const nowSec = Math.floor(Date.now() / 1000);
80-
const lagSeconds = indexerState
81-
? nowSec - Math.floor(indexerState.updatedAt.getTime() / 1000)
82-
: null;
93+
// Sum total volume streamed (sum of withdrawn amounts) as BigInt to preserve i128 precision.
94+
let totalVolumeStreamed = BigInt(0);
95+
for (const row of withdrawnSums) {
96+
totalVolumeStreamed += BigInt(row.withdrawnAmount || '0');
97+
}
98+
99+
const nowSec = Math.floor(Date.now() / 1000);
100+
const lagSeconds = indexerState
101+
? nowSec - Math.floor(indexerState.updatedAt.getTime() / 1000)
102+
: null;
103+
104+
return {
105+
// Snake_case summary requested by issue #426. Exposed at the top level so
106+
// operators (and future dashboards) can read aggregate counts without
107+
// walking the nested protocol-health tree below.
108+
total_streams: totalCount,
109+
active_streams: activeCount,
110+
paused_streams: pausedCount,
111+
completed_streams: completedCount,
112+
cancelled_streams: cancelledCount,
113+
total_volume_streamed: totalVolumeStreamed.toString(),
83114

84-
res.json({
85-
streams: {
115+
streams: {
116+
active: activeCount,
117+
paused: pausedCount,
118+
total: totalCount,
119+
byStatus: {
86120
active: activeCount,
87-
total: totalCount,
88-
byStatus: {
89-
active: activeCount,
90-
cancelled: cancelledCount,
91-
completed: completedCount,
92-
},
93-
},
94-
events: { last24h: eventsLast24h },
95-
fees: {
96-
totalFeesCollectedByToken,
97-
feesLast24h: feesLast24hByToken,
98-
},
99-
sse: { activeConnections: sseService.getClientCount() },
100-
cache: cache.getStats(), // Added my cache metrics
101-
indexer: {
102-
lastLedger: indexerState?.lastLedger ?? 0,
103-
lagSeconds,
104-
lastUpdated: indexerState?.updatedAt ?? null,
121+
paused: pausedCount,
122+
cancelled: cancelledCount,
123+
completed: completedCount,
105124
},
106-
uptime: process.uptime(),
107-
timestamp: new Date().toISOString(),
108-
});
125+
},
126+
events: { last24h: eventsLast24h },
127+
fees: {
128+
totalFeesCollectedByToken,
129+
feesLast24h: feesLast24hByToken,
130+
},
131+
sse: { activeConnections: sseService.getClientCount() },
132+
cache: cache.getStats(),
133+
indexer: {
134+
lastLedger: indexerState?.lastLedger ?? 0,
135+
lagSeconds,
136+
lastUpdated: indexerState?.updatedAt ?? null,
137+
},
138+
uptime: process.uptime(),
139+
timestamp: new Date().toISOString(),
140+
};
141+
}
142+
143+
router.get('/metrics', async (_req: Request, res: Response) => {
144+
try {
145+
const cached = cache.get<Awaited<ReturnType<typeof buildAdminMetrics>>>(
146+
ADMIN_METRICS_CACHE_KEY,
147+
);
148+
if (cached) {
149+
res.set('X-Cache', 'HIT');
150+
res.json(cached);
151+
return;
152+
}
153+
154+
const payload = await buildAdminMetrics();
155+
cache.set(ADMIN_METRICS_CACHE_KEY, payload, ADMIN_METRICS_CACHE_TTL_SECONDS);
156+
res.set('X-Cache', 'MISS');
157+
res.json(payload);
109158
} catch (err) {
110159
logger.error('Error fetching admin metrics:', err);
111160
res.status(500).json({ error: 'Internal server error' });

‎backend/src/routes/v1/events.routes.ts‎

Lines changed: 125 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,135 @@
11
import { Router } from 'express';
2-
import type { Request, Response } from 'express';
2+
import type { Request, Response, NextFunction } from 'express';
33
import { subscribe } from '../../controllers/sse.controller.js';
44
import { sseService } from '../../services/sse.service.js';
55
import { requireAuth } from '../../middleware/auth.js';
6+
import { prisma } from '../../lib/prisma.js';
7+
import logger from '../../logger.js';
68

79
const router = Router();
810

11+
const EVENT_TYPES = new Set([
12+
'CREATED',
13+
'TOPPED_UP',
14+
'WITHDRAWN',
15+
'CANCELLED',
16+
'COMPLETED',
17+
'PAUSED',
18+
'RESUMED',
19+
'FEE_COLLECTED',
20+
]);
21+
22+
const MAX_EVENT_LIMIT = 200;
23+
const DEFAULT_EVENT_LIMIT = 50;
24+
25+
/**
26+
* @openapi
27+
* /v1/events:
28+
* get:
29+
* tags: [Events]
30+
* summary: List stream events for a wallet (paginated, filterable)
31+
* description: |
32+
* Returns a reverse-chronological list of stream events where the wallet
33+
* was either the sender or recipient. Supports event-type filtering and
34+
* limit/offset pagination — used by the frontend activity timeline.
35+
* parameters:
36+
* - in: query
37+
* name: address
38+
* required: true
39+
* schema: { type: string }
40+
* description: Stellar public key (G...)
41+
* - in: query
42+
* name: type
43+
* required: false
44+
* schema: { type: string }
45+
* description: |
46+
* Comma-separated list of event types to include. Allowed values:
47+
* CREATED, TOPPED_UP, WITHDRAWN, CANCELLED, COMPLETED, PAUSED,
48+
* RESUMED, FEE_COLLECTED.
49+
* - in: query
50+
* name: limit
51+
* required: false
52+
* schema: { type: integer, default: 50, maximum: 200 }
53+
* - in: query
54+
* name: offset
55+
* required: false
56+
* schema: { type: integer, default: 0 }
57+
* - in: query
58+
* name: page
59+
* required: false
60+
* schema: { type: integer, default: 1 }
61+
* description: Optional 1-based page index. Ignored when offset is set.
62+
* responses:
63+
* 200:
64+
* description: Paginated event list
65+
*/
66+
router.get('/', async (req: Request, res: Response, next: NextFunction) => {
67+
try {
68+
const address = typeof req.query.address === 'string' ? req.query.address.trim() : '';
69+
if (!address) {
70+
res.status(400).json({ error: 'address query parameter is required' });
71+
return;
72+
}
73+
74+
const rawType = typeof req.query.type === 'string' ? req.query.type : '';
75+
const requested = rawType
76+
.split(',')
77+
.map((t) => t.trim().toUpperCase())
78+
.filter(Boolean);
79+
const types = requested.filter((t) => EVENT_TYPES.has(t));
80+
if (requested.length > 0 && types.length === 0) {
81+
res.status(400).json({ error: 'No valid event types in `type` filter' });
82+
return;
83+
}
84+
85+
const parsedLimit = Number.parseInt(String(req.query.limit ?? ''), 10);
86+
const limit = Number.isFinite(parsedLimit) && parsedLimit > 0
87+
? Math.min(parsedLimit, MAX_EVENT_LIMIT)
88+
: DEFAULT_EVENT_LIMIT;
89+
90+
const hasOffset = req.query.offset !== undefined;
91+
const parsedOffset = Number.parseInt(String(req.query.offset ?? ''), 10);
92+
const parsedPage = Number.parseInt(String(req.query.page ?? ''), 10);
93+
const page = Number.isFinite(parsedPage) && parsedPage > 0 ? parsedPage : 1;
94+
const offset = hasOffset && Number.isFinite(parsedOffset) && parsedOffset >= 0
95+
? parsedOffset
96+
: (page - 1) * limit;
97+
98+
const where: {
99+
stream: { OR: Array<{ sender: string } | { recipient: string }> };
100+
eventType?: { in: string[] };
101+
} = {
102+
stream: {
103+
OR: [{ sender: address }, { recipient: address }],
104+
},
105+
};
106+
if (types.length > 0) {
107+
where.eventType = { in: types };
108+
}
109+
110+
const [events, total] = await Promise.all([
111+
prisma.streamEvent.findMany({
112+
where,
113+
orderBy: { timestamp: 'desc' },
114+
skip: offset,
115+
take: limit,
116+
}),
117+
prisma.streamEvent.count({ where }),
118+
]);
119+
120+
res.json({
121+
events,
122+
total,
123+
limit,
124+
offset,
125+
hasMore: offset + events.length < total,
126+
});
127+
} catch (err) {
128+
logger.error('GET /v1/events failed:', err);
129+
next(err);
130+
}
131+
});
132+
9133
/**
10134
* @openapi
11135
* /v1/events/subscribe:

0 commit comments

Comments
 (0)