feature:Status filter, shared token, live notification, contract depl…

README.md

@@ -101,6 +101,96 @@ cd contracts
 cargo build --target wasm32-unknown-unknown --release
 ```
 
+## Deployment
+
+### Contract Deployment
+
+The FlowFi smart contracts can be deployed to both testnet and mainnet using the automated deployment script.
+
+#### Prerequisites
+
+- Stellar CLI installed and configured
+- Sufficient XLM in the deployment account for network fees
+- Required environment variables set
+
+#### Environment Variables
+
+Before deploying, set the following environment variables:
+
+```bash
+export STELLAR_SECRET_KEY="your_secret_key_here"
+export ADMIN_ADDRESS="your_admin_address_here"
+export TREASURY_ADDRESS="your_treasury_address_here"
+export FEE_RATE_BPS="25"  # 0.25% fee rate
+```
+
+#### Deploy to Testnet
+
+```bash
+npx tsx scripts/deploy.ts --network testnet
+```
+
+#### Deploy to Mainnet
+
+```bash
+npx tsx scripts/deploy.ts --network mainnet
+```
+
+#### Deployment Process
+
+The deployment script automates the following steps:
+
+1. **Build WASM**: Compiles the Rust contract to WebAssembly
+2. **Optimize WASM**: Optimizes the WASM for deployment size
+3. **Deploy Contract**: Deploys the contract to the specified network
+4. **Initialize Contract**: Sets up admin, treasury, and fee rate parameters
+5. **Save Deployment Info**: Stores contract details in `deployment-info.json`
+
+#### Deployment Information
+
+After successful deployment, contract details are saved to `deployment-info.json`:
+
+```json
+{
+  "testnet": {
+    "network": "testnet",
+    "contractId": "CD...ID",
+    "deployedAt": "2024-01-01T00:00:00.000Z",
+    "adminAddress": "G...ADMIN",
+    "treasuryAddress": "G...TREASURY",
+    "feeRateBps": 25,
+    "transactionHash": "TX...HASH"
+  },
+  "mainnet": {
+    "network": "mainnet",
+    "contractId": "CD...ID",
+    "deployedAt": "2024-01-01T00:00:00.000Z",
+    "adminAddress": "G...ADMIN",
+    "treasuryAddress": "G...TREASURY",
+    "feeRateBps": 25,
+    "transactionHash": "TX...HASH"
+  },
+  "lastUpdated": "2024-01-01T00:00:00.000Z"
+}
+```
+
+#### Manual Deployment
+
+If you prefer to deploy manually, you can use the Stellar CLI directly:
+
+```bash
+# Build and optimize
+cd contracts
+cargo build --target wasm32-unknown-unknown --release
+stellar contract optimize --wasm target/wasm32-unknown-unknown/release/stream_contract.wasm
+
+# Deploy
+stellar contract deploy --wasm target/wasm32-unknown-unknown/release/stream_contract.optimized.wasm --source YOUR_SECRET_KEY --network https://soroban-testnet.stellar.org
+
+# Initialize
+stellar contract invoke --id CONTRACT_ID --source YOUR_SECRET_KEY --network https://soroban-testnet.stellar.org initialize --admin ADMIN_ADDRESS --treasury TREASURY_ADDRESS --fee_rate_bps 25
+```
+
 ## API Documentation
 
 The FlowFi backend API uses URL-based versioning. All endpoints are prefixed with a version (e.g., `/v1/streams`).

backend/src/controllers/stream.controller.ts

@@ -1,4 +1,4 @@
 import type { Request, Response } from 'express';
 import { prisma } from '../lib/prisma.js';
 import logger from '../logger.js';
 import { claimableAmountService } from '../services/claimable.service.js';
@@ -79,26 +79,95 @@
 };
 
 /**
- * List streams by sender or recipient
+ * List streams by sender, recipient, status, token with sorting and pagination
  */
 export const listStreams = async (req: Request, res: Response) => {
   try {
-    const { sender, recipient } = req.query;
+    const { 
+      sender, 
+      recipient, 
+      status, 
+      token,
+      sort = 'createdAt',
+      order = 'desc',
+      limit = '20',
+      offset = '0'
+    } = req.query;
 
     const where: any = {};
     if (typeof sender === 'string') where.sender = sender;
     if (typeof recipient === 'string') where.recipient = recipient;
+    if (typeof token === 'string') where.tokenAddress = token;
 
-    const streams = await prisma.stream.findMany({
-      where,
-      orderBy: { createdAt: 'desc' },
-      include: {
-        senderUser: true,
-        recipientUser: true
+    // Handle status filtering
+    if (typeof status === 'string') {
+      const validStatuses = ['active', 'cancelled', 'completed', 'paused'];
+      if (!validStatuses.includes(status)) {
+        return res.status(400).json({ 
+          error: 'Invalid status parameter',
+          message: `status must be one of: ${validStatuses.join(', ')}`
+        });
       }
-    });
 
-    return res.status(200).json(streams);
+      // Map status to database conditions
+      switch (status) {
+        case 'active':
+          where.isActive = true;
+          break;
+        case 'cancelled':
+          where.isActive = false;
+          // Additional check for cancelled events could be added here
+          break;
+        case 'completed':
+          where.isActive = false;
+          // Additional check for completed events could be added here
+          break;
+        case 'paused':
+          where.isActive = false;
+          // Additional check for paused events could be added here
+          break;
+      }
+    }
+
+    // Validate and parse pagination parameters
+    const parsedLimit = Math.min(
+      typeof limit === 'string' ? (Number.parseInt(limit, 10) || 20) : 20,
+      100
+    );
+    const parsedOffset = typeof offset === 'string' ? (Number.parseInt(offset, 10) || 0) : 0;
+
+    // Validate sort field
+    const validSortFields = ['createdAt', 'startTime', 'lastUpdateTime', 'depositedAmount'];
+    const sortField = validSortFields.includes(typeof sort === 'string' ? sort : 'createdAt') 
+      ? (sort as 'createdAt' | 'startTime' | 'lastUpdateTime' | 'depositedAmount') 
+      : 'createdAt';
+
+    // Validate order
+    const sortOrder = order === 'asc' ? 'asc' : 'desc';
+
+    const [streams, total] = await Promise.all([
+      prisma.stream.findMany({
+        where,
+        orderBy: { [sortField]: sortOrder },
+        take: parsedLimit,
+        skip: parsedOffset,
+        include: {
+          senderUser: true,
+          recipientUser: true
+        }
+      }),
+      prisma.stream.count({ where })
+    ]);
+
+    const hasMore = parsedOffset + streams.length < total;
+
+    return res.status(200).json({ 
+      data: streams, 
+      total, 
+      hasMore,
+      limit: parsedLimit,
+      offset: parsedOffset
+    });
   } catch (error) {
     logger.error('Error listing streams:', error);
     return res.status(500).json({ error: 'Internal server error' });
@@ -302,7 +371,7 @@
  */
 export const getUserStreamSummary = async (req: Request, res: Response) => {
   try {
-    const address = (req.params.address ?? '').trim();
+    const address = Array.isArray(req.params.address) ? req.params.address[0] : (req.params.address ?? '').trim();
     if (!address) {
       return res.status(400).json({ error: 'Address is required' });
     }
@@ -339,10 +408,10 @@
     ]);
 
     const totalStreamsCreated = outgoingStreams.length;
-    const totalStreamedOut = sumStringI128(outgoingStreams.map((stream) => stream.withdrawnAmount));
-    const totalStreamedIn = sumStringI128(incomingStreams.map((stream) => stream.withdrawnAmount));
-    const activeOutgoingCount = outgoingStreams.filter((stream) => stream.isActive).length;
-    const activeIncomingCount = incomingStreams.filter((stream) => stream.isActive).length;
+    const totalStreamedOut = sumStringI128(outgoingStreams.map((stream: any) => stream.withdrawnAmount));
+    const totalStreamedIn = sumStringI128(incomingStreams.map((stream: any) => stream.withdrawnAmount));
+    const activeOutgoingCount = outgoingStreams.filter((stream: any) => stream.isActive).length;
+    const activeIncomingCount = incomingStreams.filter((stream: any) => stream.isActive).length;
 
     const calculatedAt = Math.floor(nowMs / 1000);
     let claimableTotal = 0n;

backend/src/lib/redis.ts

@@ -1,27 +1,26 @@
-import type { Redis } from 'ioredis';
-import RedisClass from 'ioredis';
+const Redis = require('ioredis');
 import logger from '../logger.js';
 
 const REDIS_URL = process.env.REDIS_URL;
 
-let _publisher: Redis | null = null;
-let _subscriber: Redis | null = null;
+let _publisher: typeof Redis | null = null;
+let _subscriber: typeof Redis | null = null;
 let _available = false;
 
-export function getPublisher(): Redis | null {
+export function getPublisher(): typeof Redis | null {
   return _publisher;
 }
 
-export function getSubscriber(): Redis | null {
+export function getSubscriber(): typeof Redis | null {
   return _subscriber;
 }
 
 export function isRedisAvailable(): boolean {
   return _available;
 }
 
-function makeClient(url: string): Redis {
-  return new RedisClass(url, {
+function makeClient(url: string): typeof Redis {
+  return new Redis(url, {
     maxRetriesPerRequest: 3,
     retryStrategy: (times: number) =>
       times > 3 ? null : Math.min(times * 200, 2000),

backend/src/services/sse.service.ts

@@ -183,4 +183,5 @@ class SSEService {
   }
 }
 
+export { SSEService };
 export const sseService = new SSEService();

backend/src/workers/soroban-event-worker.ts

@@ -133,7 +133,6 @@ export class SorobanEventWorker {
       logger.error('[SorobanWorker] Manual poll error:', err);
     }
   }
-  }
 
   // ─── Internal ──────────────────────────────────────────────────────────────
 

backend/tests/sse.service.test.ts

@@ -1,6 +1,6 @@
 import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
 import { EventEmitter } from 'node:events';
-import { SSEService } from '../src/services/sse.service.js';
+import { SSEService, sseService } from '../src/services/sse.service.js';
 
 function createMockResponse() {
   const emitter = new EventEmitter();

backend/tests/stream.test.ts

@@ -111,7 +111,7 @@
       .get('/v1/streams')
       .set('Accept', 'application/json');
 
     expect(response.status).toBe(200);
     expect(Array.isArray(response.body)).toBe(true);
   });
 });
@@ -122,7 +122,7 @@
   });
 
   it('returns all-zero summary for addresses with no streams', async () => {
-    prisma.stream.findMany
+    vi.mocked(prisma.stream.findMany)
       .mockResolvedValueOnce([])
       .mockResolvedValueOnce([]);
 
@@ -142,29 +142,69 @@
   });
 
   it('returns accurate outgoing/incoming aggregates and claimable sum', async () => {
-    prisma.stream.findMany
+    vi.mocked(prisma.stream.findMany)
       .mockResolvedValueOnce([
-        { withdrawnAmount: '30', isActive: true },
-        { withdrawnAmount: '20', isActive: false },
+        { 
+          id: '1', 
+          createdAt: new Date(), 
+          updatedAt: new Date(),
+          streamId: 1,
+          sender: 'GSENDER',
+          recipient: 'GRECIPIENT',
+          tokenAddress: 'TOKEN',
+          ratePerSecond: '10',
+          depositedAmount: '100',
+          withdrawnAmount: '30', 
+          startTime: 1000,
+          lastUpdateTime: 2000,
+          isActive: true 
+        },
+        { 
+          id: '2', 
+          createdAt: new Date(), 
+          updatedAt: new Date(),
+          streamId: 2,
+          sender: 'GSENDER2',
+          recipient: 'GRECIPIENT2',
+          tokenAddress: 'TOKEN2',
+          ratePerSecond: '20',
+          depositedAmount: '200',
+          withdrawnAmount: '20', 
+          startTime: 1000,
+          lastUpdateTime: 2000,
+          isActive: false 
+        },
       ])
       .mockResolvedValueOnce([
         {
+          id: '3',
+          createdAt: new Date(),
+          updatedAt: new Date(),
           streamId: 11,
+          sender: 'GSENDER3',
+          recipient: 'GRECIPIENT3',
+          tokenAddress: 'TOKEN3',
           ratePerSecond: '10',
           depositedAmount: '1000',
           withdrawnAmount: '100',
+          startTime: 1000,
           lastUpdateTime: 0,
           isActive: true,
-          updatedAt: new Date(),
         },
         {
+          id: '4',
+          createdAt: new Date(),
+          updatedAt: new Date(),
           streamId: 12,
-          ratePerSecond: '1',
-          depositedAmount: '200',
-          withdrawnAmount: '50',
+          sender: 'GSENDER4',
+          recipient: 'GRECIPIENT4',
+          tokenAddress: 'TOKEN4',
+          ratePerSecond: '5',
+          depositedAmount: '500',
+          withdrawnAmount: '0',
+          startTime: 1000,
           lastUpdateTime: 0,
           isActive: false,
-          updatedAt: new Date(),
         },
       ]);
 
@@ -172,7 +212,7 @@
     const response = await request(app).get(`/v1/users/${address}/summary`);
 
     expect(response.status).toBe(200);
     expect(response.body).toMatchObject({
       address,
       totalStreamsCreated: 2,
       totalStreamedOut: '50',
@@ -184,8 +224,22 @@
   });
 
   it('caches summary results for repeated requests within TTL', async () => {
-    prisma.stream.findMany
-      .mockResolvedValueOnce([{ withdrawnAmount: '1', isActive: true }])
+    vi.mocked(prisma.stream.findMany)
+      .mockResolvedValueOnce([{ 
+        id: '5',
+        createdAt: new Date(),
+        updatedAt: new Date(),
+        streamId: 13,
+        sender: 'GSENDER5',
+        recipient: 'GRECIPIENT5',
+        tokenAddress: 'TOKEN5',
+        ratePerSecond: '1',
+        depositedAmount: '100',
+        withdrawnAmount: '1',
+        startTime: 1000,
+        lastUpdateTime: 2000,
+        isActive: true 
+      }])
       .mockResolvedValueOnce([]);
 
     const address = 'GCACHE000000000000000000000000000000000000000000000000000000';