diff --git a/README.md b/README.md index ef9fe28..197e72e 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,8 @@ flowfi/ │ ├── stream_contract/ # Core streaming logic ├── frontend/ # Next.js + Tailwind CSS frontend ├── docs/ # Documentation -│ └── ARCHITECTURE.md # Architecture overview +│ ├── ARCHITECTURE.md # Architecture overview +│ └── DEVELOPMENT.md # Local development guide ``` ## Architecture @@ -40,6 +41,8 @@ For full local setup and contributor onboarding, see the [Development Guide](doc ## Getting Started +For full step-by-step instructions, see our [Development Guide](docs/DEVELOPMENT.md). + ### Prerequisites - Node.js & npm @@ -94,11 +97,12 @@ npm install npm run dev ``` -### Smart Contracts +### Deployment + +To build, optimize, and deploy the smart contract to a Stellar network, you can use the automated deployment script: ```bash -cd contracts -cargo build --target wasm32-unknown-unknown --release +./scripts/deploy.sh --network testnet ``` ## Deployment diff --git a/backend/src/__tests__/integration/streams.test.ts b/backend/src/__tests__/integration/streams.test.ts new file mode 100644 index 0000000..8d3c061 --- /dev/null +++ b/backend/src/__tests__/integration/streams.test.ts @@ -0,0 +1,221 @@ +import 'dotenv/config'; +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import request from 'supertest'; +import { nativeToScVal, xdr, StrKey, Keypair } from '@stellar/stellar-sdk'; +import app from '../../app.js'; +import { prisma } from '../../lib/prisma.js'; +import { sorobanEventWorker } from '../../workers/soroban-event-worker.js'; +import { sseService } from '../../services/sse.service.js'; + +describe('Stream Lifecycle Integration Tests', () => { + const senderPair = Keypair.random(); + const recipientPair = Keypair.random(); + const sender = senderPair.publicKey(); + const recipient = recipientPair.publicKey(); + const tokenAddress = StrKey.encodeContract(Buffer.alloc(32)); + const streamId = 999; + let sseEvents: any[] = []; + + beforeAll(async () => { + // Set up a test SSE client + const mockRes = { + write: (chunk: string) => { + const lines = chunk.split('\n'); + let eventName = ''; + let data: any = null; + for (const line of lines) { + if (line.startsWith('event: ')) eventName = line.slice(7).trim(); + if (line.startsWith('data: ')) { + try { + data = JSON.parse(line.slice(6).trim()); + } catch (e) {} + } + } + if (eventName && data) { + sseEvents.push({ event: eventName, data }); + } + }, + on: () => {}, + } as any; + + sseService.addClient('test-integration-client', mockRes, ['*']); + + // Clean up DB before test + await prisma.streamEvent.deleteMany({ where: { streamId } }).catch(() => {}); + await prisma.stream.deleteMany({ where: { streamId } }).catch(() => {}); + }); + + afterAll(async () => { + await prisma.streamEvent.deleteMany({ where: { streamId } }).catch(() => {}); + await prisma.stream.deleteMany({ where: { streamId } }).catch(() => {}); + }); + + it('Indexer processes stream_created event -> stream appears in GET /v1/streams/{id}', async () => { + sseEvents = []; + + const event = { + id: 'created-event-1', + txHash: 'hash-created', + ledger: 100, + inSuccessfulContractCall: true, + topic: [ + xdr.ScVal.scvSymbol('stream_created'), + nativeToScVal(BigInt(streamId), { type: 'u64' }), + ], + value: xdr.ScVal.scvMap([ + new xdr.ScMapEntry({ + key: xdr.ScVal.scvSymbol('sender'), + val: nativeToScVal(sender, { type: 'address' }), + }), + new xdr.ScMapEntry({ + key: xdr.ScVal.scvSymbol('recipient'), + val: nativeToScVal(recipient, { type: 'address' }), + }), + new xdr.ScMapEntry({ + key: xdr.ScVal.scvSymbol('token_address'), + val: nativeToScVal(tokenAddress, { type: 'address' }), + }), + new xdr.ScMapEntry({ + key: xdr.ScVal.scvSymbol('rate_per_second'), + val: nativeToScVal(BigInt(10), { type: 'i128' }), + }), + new xdr.ScMapEntry({ + key: xdr.ScVal.scvSymbol('deposited_amount'), + val: nativeToScVal(BigInt(1000), { type: 'i128' }), + }), + new xdr.ScMapEntry({ + key: xdr.ScVal.scvSymbol('start_time'), + val: nativeToScVal(BigInt(Math.floor(Date.now() / 1000)), { type: 'u64' }), + }), + ]), + } as any; + + await sorobanEventWorker.processEvent(event); + + // Verify stream appears in GET API + const res = await request(app).get(`/v1/streams/${streamId}`); + expect(res.status).toBe(200); + expect(res.body.streamId).toBe(streamId); + expect(res.body.depositedAmount).toBe('1000'); + + // Verify SSE + expect(sseEvents.some(e => e.event === 'stream.created')).toBe(true); + }); + + it('Indexer processes stream_topped_up -> depositedAmount updated in DB', async () => { + sseEvents = []; + + const event = { + id: 'topped-up-event-1', + txHash: 'hash-topped-up', + ledger: 101, + inSuccessfulContractCall: true, + topic: [ + xdr.ScVal.scvSymbol('stream_topped_up'), + nativeToScVal(BigInt(streamId), { type: 'u64' }), + ], + value: xdr.ScVal.scvMap([ + new xdr.ScMapEntry({ + key: xdr.ScVal.scvSymbol('amount'), + val: nativeToScVal(BigInt(500), { type: 'i128' }), + }), + new xdr.ScMapEntry({ + key: xdr.ScVal.scvSymbol('new_deposited_amount'), + val: nativeToScVal(BigInt(1500), { type: 'i128' }), + }), + ]), + } as any; + + await sorobanEventWorker.processEvent(event); + + const dbStream = await prisma.stream.findUnique({ where: { streamId } }); + expect(dbStream?.depositedAmount).toBe('1500'); + + expect(sseEvents.some(e => e.event === 'stream.topped_up')).toBe(true); + }); + + it('Indexer processes stream_paused -> isPaused = true', async () => { + sseEvents = []; + + const event = { + id: 'paused-event-1', + txHash: 'hash-paused', + ledger: 102, + inSuccessfulContractCall: true, + topic: [ + xdr.ScVal.scvSymbol('stream_paused'), + nativeToScVal(BigInt(streamId), { type: 'u64' }), + ], + value: xdr.ScVal.scvMap([]), + } as any; + + await sorobanEventWorker.processEvent(event); + + const dbStream = await prisma.stream.findUnique({ where: { streamId } }); + expect(dbStream?.isPaused).toBe(true); + + expect(sseEvents.some(e => e.event === 'stream.paused')).toBe(true); + }); + + it('Indexer processes stream_resumed -> isPaused = false', async () => { + sseEvents = []; + + const event = { + id: 'resumed-event-1', + txHash: 'hash-resumed', + ledger: 103, + inSuccessfulContractCall: true, + topic: [ + xdr.ScVal.scvSymbol('stream_resumed'), + nativeToScVal(BigInt(streamId), { type: 'u64' }), + ], + value: xdr.ScVal.scvMap([]), + } as any; + + await sorobanEventWorker.processEvent(event); + + const dbStream = await prisma.stream.findUnique({ where: { streamId } }); + expect(dbStream?.isPaused).toBe(false); + + expect(sseEvents.some(e => e.event === 'stream.resumed')).toBe(true); + }); + + it('Indexer processes stream_cancelled -> stream isActive = false', async () => { + sseEvents = []; + + const event = { + id: 'cancelled-event-1', + txHash: 'hash-cancelled', + ledger: 104, + inSuccessfulContractCall: true, + topic: [ + xdr.ScVal.scvSymbol('stream_cancelled'), + nativeToScVal(BigInt(streamId), { type: 'u64' }), + ], + value: xdr.ScVal.scvMap([ + new xdr.ScMapEntry({ + key: xdr.ScVal.scvSymbol('amount_withdrawn'), + val: nativeToScVal(BigInt(100), { type: 'i128' }), + }), + new xdr.ScMapEntry({ + key: xdr.ScVal.scvSymbol('refunded_amount'), + val: nativeToScVal(BigInt(1400), { type: 'i128' }), + }), + ]), + } as any; + + await sorobanEventWorker.processEvent(event); + + const dbStream = await prisma.stream.findUnique({ where: { streamId } }); + expect(dbStream?.isActive).toBe(false); + + expect(sseEvents.some(e => e.event === 'stream.cancelled')).toBe(true); + }); + + it('GET /v1/streams/{id}/events returns all lifecycle events', async () => { + const res = await request(app).get(`/v1/streams/${streamId}/events`); + expect(res.status).toBe(200); + expect(Array.isArray(res.body.data)).toBe(true); + expect(res.body.data.length).toBeGreaterThan(0); + }); +}); diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index 29da161..cd36d5e 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/backend/src/workers/soroban-event-worker.ts @@ -233,7 +233,7 @@ export class SorobanEventWorker { * Dispatch a single contract event to the appropriate handler based on the * first topic symbol. */ - private async processEvent( + public async processEvent( event: rpc.Api.EventResponse, ): Promise { if (!event.topic || event.topic.length < 2) return; @@ -254,6 +254,12 @@ export class SorobanEventWorker { case 'tokens_withdrawn': await this.handleTokensWithdrawn(event, topic1); break; + case 'stream_paused': + await this.handleStreamPaused(event, topic1); + break; + case 'stream_resumed': + await this.handleStreamResumed(event, topic1); + break; case 'stream_cancelled': await this.handleStreamCancelled(event, topic1); break; @@ -426,6 +432,88 @@ export class SorobanEventWorker { }); } + private async handleStreamPaused( + event: rpc.Api.EventResponse, + streamIdTopic: xdr.ScVal, + ): Promise { + const streamId = Number(decodeU64(streamIdTopic)); + const timestamp = Math.floor(Date.now() / 1000); + + await prisma.$transaction(async (tx: any) => { + await tx.stream.update({ + where: { streamId }, + data: { + isPaused: true, + lastPausedAt: timestamp, + }, + }); + + await tx.streamEvent.create({ + data: { + streamId, + eventType: 'PAUSED', + transactionHash: event.txHash, + ledgerSequence: event.ledger, + timestamp, + }, + }); + }); + + sseService.broadcastToStream(String(streamId), 'stream.paused', { + streamId, + isPaused: true, + transactionHash: event.txHash, + ledger: event.ledger, + timestamp, + }); + } + + private async handleStreamResumed( + event: rpc.Api.EventResponse, + streamIdTopic: xdr.ScVal, + ): Promise { + const streamId = Number(decodeU64(streamIdTopic)); + const timestamp = Math.floor(Date.now() / 1000); + + await prisma.$transaction(async (tx: any) => { + const stream = await tx.stream.findUniqueOrThrow({ + where: { streamId }, + select: { totalPausedSeconds: true, lastPausedAt: true }, + }); + + const lastPausedAt = stream.lastPausedAt ?? timestamp; + const pausedDuration = Math.max(0, timestamp - lastPausedAt); + const nextTotalPausedSeconds = stream.totalPausedSeconds + pausedDuration; + + await tx.stream.update({ + where: { streamId }, + data: { + isPaused: false, + totalPausedSeconds: nextTotalPausedSeconds, + lastPausedAt: null, + }, + }); + + await tx.streamEvent.create({ + data: { + streamId, + eventType: 'RESUMED', + transactionHash: event.txHash, + ledgerSequence: event.ledger, + timestamp, + }, + }); + }); + + sseService.broadcastToStream(String(streamId), 'stream.resumed', { + streamId, + isPaused: false, + transactionHash: event.txHash, + ledger: event.ledger, + timestamp, + }); + } + private async handleTokensWithdrawn( event: rpc.Api.EventResponse, streamIdTopic: xdr.ScVal, diff --git a/backend/vitest.config.ts b/backend/vitest.config.ts index 13b4bb7..151f773 100644 --- a/backend/vitest.config.ts +++ b/backend/vitest.config.ts @@ -5,7 +5,7 @@ export default defineConfig({ environment: 'node', globals: true, setupFiles: [], - include: ['tests/**/*.{test,spec}.ts'], + include: ['tests/**/*.{test,spec}.ts', 'src/__tests__/**/*.{test,spec}.ts'], coverage: { reporter: ['text', 'json', 'html'], }, diff --git a/docker-compose.yml b/docker-compose.yml index c05b2f7..2eb1e67 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -7,7 +7,7 @@ services: POSTGRES_PASSWORD: flowfi_dev_password POSTGRES_DB: flowfi ports: - - "5432:5432" + - "5433:5432" volumes: - postgres_data:/var/lib/postgresql/data healthcheck: diff --git a/frontend/src/components/dashboard/dashboard-view.tsx b/frontend/src/components/dashboard/dashboard-view.tsx index 59568a2..ed17d4f 100644 --- a/frontend/src/components/dashboard/dashboard-view.tsx +++ b/frontend/src/components/dashboard/dashboard-view.tsx @@ -3,6 +3,7 @@ import React from "react"; import Link from "next/link"; import toast from "react-hot-toast"; +import { StreamListSkeleton } from "../ui/Skeleton"; /** * components/dashboard/dashboard-view.tsx @@ -740,6 +741,7 @@ export function DashboardView({ session, onDisconnect }: DashboardViewProps) { // ── Overview ────────────────────────────────────────────────────────── if (activeTab === "overview") { + return (
{renderStats(snapshot)} diff --git a/frontend/src/components/stream-creation/StreamCreationWizard.tsx b/frontend/src/components/stream-creation/StreamCreationWizard.tsx index 18de64c..fe1c55d 100644 --- a/frontend/src/components/stream-creation/StreamCreationWizard.tsx +++ b/frontend/src/components/stream-creation/StreamCreationWizard.tsx @@ -483,18 +483,8 @@ export const StreamCreationWizard: React.FC = ({ ) : ( - )}
diff --git a/frontend/src/components/stream-creation/TopUpModal.tsx b/frontend/src/components/stream-creation/TopUpModal.tsx index ad6a0b4..094a965 100644 --- a/frontend/src/components/stream-creation/TopUpModal.tsx +++ b/frontend/src/components/stream-creation/TopUpModal.tsx @@ -165,18 +165,8 @@ export const TopUpModal: React.FC = ({ - diff --git a/frontend/src/components/ui/Button.tsx b/frontend/src/components/ui/Button.tsx index 59925b4..80abe34 100644 --- a/frontend/src/components/ui/Button.tsx +++ b/frontend/src/components/ui/Button.tsx @@ -4,6 +4,7 @@ interface ButtonProps extends React.ButtonHTMLAttributes { variant?: 'primary' | 'secondary' | 'outline' | 'ghost'; size?: 'sm' | 'md' | 'lg'; glow?: boolean; + loading?: boolean; } export const Button: React.FC = ({ @@ -12,6 +13,8 @@ export const Button: React.FC = ({ variant = 'primary', size = 'md', glow = false, + loading = false, + disabled, ...props }) => { const baseStyles = 'inline-flex items-center justify-center rounded-full font-semibold transition-all active:scale-95 disabled:opacity-50 disabled:pointer-events-none'; @@ -34,8 +37,31 @@ export const Button: React.FC = ({ return ( ); diff --git a/frontend/src/components/ui/Skeleton.tsx b/frontend/src/components/ui/Skeleton.tsx new file mode 100644 index 0000000..3377e22 --- /dev/null +++ b/frontend/src/components/ui/Skeleton.tsx @@ -0,0 +1,24 @@ +import React from "react"; + +export function Skeleton({ className = "" }: { className?: string }) { + return ( +
+ ); +} + +export function StreamListSkeleton() { + return ( +
+
+ +
+
+ + + +
+
+ ); +} diff --git a/package.json b/package.json index 8b4907c..e49a4a7 100644 --- a/package.json +++ b/package.json @@ -22,6 +22,7 @@ ] }, "dependencies": { + "@rollup/rollup-linux-x64-gnu": "4.60.2", "react-hot-toast": "^2.6.0" }, "optionalDependencies": { @@ -30,4 +31,4 @@ "lightningcss-darwin-arm64": "^1.31.1", "lightningcss-linux-x64-gnu": "^1.31.1" } -} \ No newline at end of file +} diff --git a/scripts/deploy.sh b/scripts/deploy.sh new file mode 100755 index 0000000..894a108 --- /dev/null +++ b/scripts/deploy.sh @@ -0,0 +1,73 @@ +#!/bin/bash +set -e + +# Default to testnet +NETWORK="testnet" +ADMIN_ALIAS="admin" +TREASURY_ALIAS="treasury" + +# Parse arguments +while [[ "$#" -gt 0 ]]; do + case $1 in + --network) NETWORK="$2"; shift ;; + --admin) ADMIN_ALIAS="$2"; shift ;; + --treasury) TREASURY_ALIAS="$2"; shift ;; + *) echo "Unknown parameter passed: $1"; exit 1 ;; + esac + shift +done + +echo "Deploying to $NETWORK..." + +# Check if admin alias exists +if ! stellar keys ls | grep -q "$ADMIN_ALIAS"; then + echo "Generating admin key ($ADMIN_ALIAS)..." + stellar keys generate "$ADMIN_ALIAS" --network "$NETWORK" +fi + +# Check if treasury alias exists +if ! stellar keys ls | grep -q "$TREASURY_ALIAS"; then + echo "Generating treasury key ($TREASURY_ALIAS)..." + stellar keys generate "$TREASURY_ALIAS" --network "$NETWORK" +fi + +ADMIN_ADDRESS=$(stellar keys address "$ADMIN_ALIAS") +TREASURY_ADDRESS=$(stellar keys address "$TREASURY_ALIAS") + +cd contracts + +# Build the contract +echo "Building contract..." +cargo build --target wasm32-unknown-unknown --release + +# Optimize the contract +echo "Optimizing contract..." +stellar contract optimize --wasm target/wasm32-unknown-unknown/release/stream_contract.wasm + +# Deploy the contract +echo "Deploying contract..." +CONTRACT_ID=$(stellar contract deploy \ + --wasm target/wasm32-unknown-unknown/release/stream_contract.optimized.wasm \ + --network "$NETWORK" \ + --source "$ADMIN_ALIAS") + +echo "Contract deployed with ID: $CONTRACT_ID" + +# Initialize the contract +echo "Initializing contract..." +stellar contract invoke \ + --id "$CONTRACT_ID" \ + --network "$NETWORK" \ + --source "$ADMIN_ALIAS" \ + -- \ + initialize \ + --admin "$ADMIN_ADDRESS" \ + --treasury "$TREASURY_ADDRESS" \ + --fee_rate_bps 100 + +cd .. + +# Save to deployment-info.json +echo "{\"network\": \"$NETWORK\", \"contract_id\": \"$CONTRACT_ID\"}" > deployment-info.json + +echo "Deployment complete! Info saved to deployment-info.json"