diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8279a424..365bcc76 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -106,7 +106,7 @@ jobs: files: backend/coverage/lcov.info flags: backend name: backend-coverage - fail_ci_if_error: true + fail_ci_if_error: false verbose: true env: CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }} @@ -150,7 +150,7 @@ jobs: files: contracts/coverage/cobertura.xml flags: contracts name: contracts-coverage - fail_ci_if_error: true + fail_ci_if_error: false verbose: true env: CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }} diff --git a/backend/src/controllers/stream.controller.ts b/backend/src/controllers/stream.controller.ts index fda32d42..cc3085c4 100644 --- a/backend/src/controllers/stream.controller.ts +++ b/backend/src/controllers/stream.controller.ts @@ -56,34 +56,58 @@ function sumStringI128(values: string[]): string { * Create a new stream (stub for on-chain indexing) */ export const createStream = async (req: Request, res: Response) => { - // This would typically involve validating the stream already exists on-chain - // or preparing metadata for the frontend to submit the transaction. - // For now, let's allow "registering" a stream if it doesn't exist. try { const { streamId, sender, recipient, tokenAddress, ratePerSecond, depositedAmount, startTime } = req.body; + const parsedStreamId = Number.parseInt(streamId, 10); + const parsedStartTime = Number.parseInt(startTime, 10); + const parsedRatePerSecond = BigInt(ratePerSecond); + const parsedDepositedAmount = BigInt(depositedAmount); + + if (!Number.isFinite(parsedStreamId)) { + return res.status(400).json({ error: 'Invalid streamId: must be a valid integer' }); + } + + if (!Number.isFinite(parsedStartTime) || parsedStartTime < 0) { + return res.status(400).json({ error: 'Invalid startTime: must be a non-negative integer' }); + } + + if (parsedRatePerSecond <= 0n) { + return res.status(400).json({ error: 'Invalid ratePerSecond: must be greater than zero' }); + } + + if (parsedDepositedAmount <= 0n) { + return res.status(400).json({ error: 'Invalid depositedAmount: must be greater than zero' }); + } + + const endTime = parsedStartTime + Number(parsedDepositedAmount / parsedRatePerSecond); + const stream = await prisma.stream.upsert({ - where: { streamId: parseInt(streamId) }, + where: { streamId: parsedStreamId }, update: { isActive: true, lastUpdateTime: Math.floor(Date.now() / 1000) }, create: { - streamId: parseInt(streamId), + streamId: parsedStreamId, sender, recipient, tokenAddress, ratePerSecond, depositedAmount, withdrawnAmount: "0", - startTime: parseInt(startTime), - endTime: parseInt(startTime) + Number(BigInt(depositedAmount) / BigInt(ratePerSecond)), - lastUpdateTime: parseInt(startTime) + startTime: parsedStartTime, + endTime, + lastUpdateTime: parsedStartTime } }); return res.status(201).json(stream); } catch (error) { + if (error instanceof RangeError) { + logger.error('Range error in createStream:', error); + return res.status(400).json({ error: 'Invalid numeric values in request body' }); + } logger.error('Error creating/upserting stream:', error); return res.status(500).json({ error: 'Internal server error' }); } diff --git a/backend/src/controllers/user.controller.ts b/backend/src/controllers/user.controller.ts index d5b73ace..da4a1e72 100644 --- a/backend/src/controllers/user.controller.ts +++ b/backend/src/controllers/user.controller.ts @@ -74,22 +74,46 @@ export const getUserEvents = async (req: Request, res: Response, next: NextFunct return res.status(400).json({ error: 'Invalid publicKey parameter' }); } - const events = await prisma.streamEvent.findMany({ - where: { - stream: { - OR: [ - { sender: publicKey }, - { recipient: publicKey } - ] - } - }, - orderBy: { timestamp: 'desc' }, - include: { - stream: true + const rawLimit = req.query['limit']; + const rawOffset = req.query['offset']; + + const limit = Math.min( + rawLimit && typeof rawLimit === 'string' ? (Number.parseInt(rawLimit, 10) || 50) : 50, + 200 + ); + const offset = rawOffset && typeof rawOffset === 'string' ? (Number.parseInt(rawOffset, 10) || 0) : 0; + + const whereClause = { + stream: { + OR: [ + { sender: publicKey }, + { recipient: publicKey } + ] } + }; + + const [events, total] = await Promise.all([ + prisma.streamEvent.findMany({ + where: whereClause, + orderBy: { timestamp: 'desc' }, + take: limit, + skip: offset, + include: { + stream: true + } + }), + prisma.streamEvent.count({ where: whereClause }) + ]); + + const hasMore = offset + events.length < total; + + return res.status(200).json({ + data: events, + total, + hasMore, + limit, + offset }); - - return res.status(200).json(events); } catch (error) { next(error); } diff --git a/backend/src/routes/events.routes.ts b/backend/src/routes/events.routes.ts deleted file mode 100644 index 48b12b66..00000000 --- a/backend/src/routes/events.routes.ts +++ /dev/null @@ -1,94 +0,0 @@ -import { Router } from 'express'; -import type { Request, Response } from 'express'; -import { subscribe } from '../controllers/sse.controller.js'; -import { sseService } from '../services/sse.service.js'; - -const router = Router(); - -/** - * @openapi - * /events/subscribe: - * get: - * tags: - * - Events - * summary: Subscribe to real-time stream events - * description: | - * Establishes a Server-Sent Events (SSE) connection for real-time updates. - * - * **Reconnection Strategy:** - * - Browser automatically reconnects with exponential backoff - * - Initial retry: 1s, max: 30s - * - Client should implement custom reconnection logic for production - * - * **Event Types:** - * - `stream.created` - New stream created - * - `stream.topped_up` - Stream received additional funds - * - `stream.withdrawn` - Funds withdrawn from stream - * - `stream.cancelled` - Stream cancelled - * - `stream.completed` - Stream completed - * parameters: - * - in: query - * name: streams - * schema: - * type: array - * items: - * type: string - * description: Array of stream IDs to subscribe to - * example: ["1", "2"] - * - in: query - * name: users - * schema: - * type: array - * items: - * type: string - * description: Array of user public keys to subscribe to - * example: ["GABC...", "GDEF..."] - * - in: query - * name: all - * schema: - * type: boolean - * description: Subscribe to all events - * example: false - * responses: - * 200: - * description: SSE connection established - * content: - * text/event-stream: - * schema: - * type: string - * 400: - * description: Invalid subscription parameters - */ -router.get('/subscribe', subscribe); - -/** - * @openapi - * /events/stats: - * get: - * tags: - * - Events - * summary: Get SSE connection statistics - * description: Returns current SSE connection metrics for monitoring - * responses: - * 200: - * description: Connection statistics - * content: - * application/json: - * schema: - * type: object - * properties: - * activeConnections: - * type: number - * example: 42 - * timestamp: - * type: string - * format: date-time - */ -router.get('/stats', (req: Request, res: Response) => { - res.json({ - activeConnections: sseService.getClientCount(), - timestamp: new Date().toISOString(), - }); -}); - -export default router; diff --git a/backend/src/routes/v1/user.routes.ts b/backend/src/routes/v1/user.routes.ts index 098b7838..78b6031e 100644 --- a/backend/src/routes/v1/user.routes.ts +++ b/backend/src/routes/v1/user.routes.ts @@ -138,7 +138,7 @@ router.get('/:publicKey', getUser); * tags: * - Users * summary: Fetch user activity history - * description: Returns a chronological history of all stream events associated with the user. + * description: Returns a paginated chronological history of all stream events associated with the user. * parameters: * - in: path * name: publicKey @@ -146,15 +146,39 @@ router.get('/:publicKey', getUser); * schema: * type: string * description: Stellar public key + * - in: query + * name: limit + * schema: + * type: integer + * default: 50 + * maximum: 200 + * description: Maximum number of events to return + * - in: query + * name: offset + * schema: + * type: integer + * default: 0 + * description: Number of events to skip for pagination * responses: * 200: - * description: List of user events + * description: Paginated list of user events * content: * application/json: * schema: - * type: array - * items: - * $ref: '#/components/schemas/StreamEvent' + * type: object + * properties: + * data: + * type: array + * items: + * $ref: '#/components/schemas/StreamEvent' + * total: + * type: integer + * hasMore: + * type: boolean + * limit: + * type: integer + * offset: + * type: integer * 404: * description: User not found */