Skip to content

Commit 59dc1e2

Browse files
ChidubemkingsleyChidubemkingsley
authored
fix: improve stream lifecycle mutation UX (#1004)
1 parent a9d49a2 commit 59dc1e2

8 files changed

Lines changed: 153 additions & 26 deletions

File tree

‎packages/react/src/__tests__/useStreamLifecycleActions.test.tsx‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ import { render, screen, waitFor } from '@testing-library/react';
33
import {
44
StreamFiProvider,
55
useCancelStream,
6+
useCreateBatchStreams,
67
usePauseStream,
78
useResumeStream,
89
useTopUpStream,
@@ -14,8 +15,13 @@ const mockCancel = vi.fn();
1415
const mockPause = vi.fn();
1516
const mockResume = vi.fn();
1617
const mockTopUp = vi.fn();
18+
const mockCreateBatchStreams = vi.fn();
1719

1820
vi.mock('@conduit-protocol/sdk', () => ({
21+
SUPPORTED_NETWORKS: ['testnet'],
22+
UnsupportedChainError: class UnsupportedChainError extends Error {},
23+
clearServerCache: vi.fn(),
24+
clearTokenDecimalsCache: vi.fn(),
1925
ConduitClient: vi.fn(function () {
2026
return {
2127
streams: {
@@ -24,6 +30,7 @@ vi.mock('@conduit-protocol/sdk', () => ({
2430
pause: mockPause,
2531
resume: mockResume,
2632
topUp: mockTopUp,
33+
createBatchStreams: mockCreateBatchStreams,
2734
},
2835
};
2936
}),
@@ -59,6 +66,36 @@ describe('stream lifecycle hooks', () => {
5966
mockPause.mockReset().mockResolvedValue('pause-tx');
6067
mockResume.mockReset().mockResolvedValue('resume-tx');
6168
mockTopUp.mockReset().mockResolvedValue('top-up-tx');
69+
mockCreateBatchStreams.mockReset().mockResolvedValue([
70+
{ index: 0, success: true, streamId: 1n, txHash: 'batch-tx-1' },
71+
{ index: 1, success: true, streamId: 2n, txHash: 'batch-tx-2' },
72+
]);
73+
});
74+
75+
it('reports batch creation progress through the hook callback', async () => {
76+
const onProgress = vi.fn();
77+
const TestComponent = () => {
78+
const batch = useCreateBatchStreams();
79+
return (
80+
<button
81+
data-testid="batch-create"
82+
onClick={() => batch.createBatchStreams([{ recipient: 'GRECIPIENT', token: 'native', depositAmount: '10' }], onProgress)}
83+
>
84+
create
85+
</button>
86+
);
87+
};
88+
89+
render(
90+
<StreamFiProvider config={{ network: 'testnet' as const }}>
91+
<TestComponent />
92+
</StreamFiProvider>,
93+
);
94+
95+
screen.getByTestId('batch-create').click();
96+
97+
await waitFor(() => expect(mockCreateBatchStreams).toHaveBeenCalledTimes(1));
98+
expect(onProgress).toHaveBeenCalledWith(1, 1);
6299
});
63100

64101
it('wraps withdraw, cancel, pause, resume, and top up stream actions', async () => {

‎packages/react/src/hooks/useCancelStream.ts‎

Lines changed: 25 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import { useCallback, useMemo, useState } from 'react';
22
import { useStreamMutation, type StreamMutationId, type UseStreamMutationState } from './useStreamMutation.js';
3-
import { useStream } from './useStream.js';
3+
import { useStream, type UseStreamResult } from './useStream.js';
44
import { useStreamFiClient } from '../context/useStreamFiClient.js';
55

66
export interface CancelSettlement {
@@ -14,7 +14,7 @@ export interface CancelSettlement {
1414
cancellationFeeAmount?: bigint;
1515
}
1616

17-
export type CancelStreamFn = () => Promise<void>;
17+
export type CancelStreamFn = () => Promise<string>;
1818

1919
export interface UseCancelStreamResult extends UseStreamMutationState {
2020
/** Opens the in-app confirmation flow; does not contact the wallet. */
@@ -53,7 +53,7 @@ function settlementFor(stream: {
5353

5454
export function useCancelStream(streamId: StreamMutationId): UseCancelStreamResult {
5555
const { client } = useStreamFiClient();
56-
const { stream } = useStream(streamId);
56+
const { stream, setStream } = useStream(streamId) as UseStreamResult;
5757
const mutation = useStreamMutation(streamId, (client, id) => client.streams.cancel(id));
5858
const [isConfirming, setIsConfirming] = useState(false);
5959
const settlement = useMemo(() => stream ? settlementFor(stream) : null, [stream]);
@@ -63,17 +63,35 @@ export function useCancelStream(streamId: StreamMutationId): UseCancelStreamResu
6363
// mutations. A real ConduitClient always exposes streams.get, so browser
6464
// callers follow the confirmation path above.
6565
if (typeof client?.streams?.get !== 'function') {
66-
await mutation.mutate();
66+
const previous = stream;
67+
if (stream) {
68+
setStream((current) => current ? { ...current, cancelled: true, paused: false, pausedAt: 0 } : current);
69+
}
70+
try {
71+
await mutation.mutate();
72+
} catch (error) {
73+
if (previous) setStream(previous);
74+
throw error;
75+
}
6776
return;
6877
}
6978
setIsConfirming(true);
70-
}, [client, mutation.mutate]);
79+
}, [client, mutation, setStream, stream]);
7180
const dismissCancel = useCallback(() => setIsConfirming(false), []);
7281
const confirmCancel = useCallback(async () => {
7382
if (!settlement) throw new Error('Stream settlement is not available yet.');
83+
const previous = stream;
84+
if (stream) {
85+
setStream((current) => current ? { ...current, cancelled: true, paused: false, pausedAt: 0 } : current);
86+
}
7487
setIsConfirming(false);
75-
return mutation.mutate();
76-
}, [mutation.mutate, settlement]);
88+
try {
89+
return await mutation.mutate();
90+
} catch (error) {
91+
if (previous) setStream(previous);
92+
throw error;
93+
}
94+
}, [mutation, setStream, settlement, stream]);
7795
const reset = useCallback(() => {
7896
setIsConfirming(false);
7997
mutation.reset();

‎packages/react/src/hooks/useCreateBatchStreams.ts‎

Lines changed: 22 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,12 @@ export interface UseCreateBatchStreamsState {
88
results: BatchCreateStreamResult[];
99
}
1010

11-
export type CreateBatchStreamsFn = (configs: StreamConfig[]) => Promise<BatchCreateStreamResult[]>;
11+
export type OnBatchProgress = (completed: number, total: number) => void;
12+
13+
export type CreateBatchStreamsFn = (
14+
configs: StreamConfig[],
15+
onProgress?: OnBatchProgress,
16+
) => Promise<BatchCreateStreamResult[]>;
1217

1318
export interface UseCreateBatchStreamsResult extends UseCreateBatchStreamsState {
1419
createBatchStreams: CreateBatchStreamsFn;
@@ -28,17 +33,31 @@ export function useCreateBatchStreams(): UseCreateBatchStreamsResult {
2833
}, []);
2934

3035
const createBatchStreams = useCallback(
31-
async (configs: StreamConfig[]): Promise<BatchCreateStreamResult[]> => {
36+
async (configs: StreamConfig[], onProgress?: OnBatchProgress): Promise<BatchCreateStreamResult[]> => {
3237
if (!isReady || !client) {
3338
const err = new Error('ConduitClient is not connected. Wrap your component with <StreamFiProvider>.');
3439
setState({ loading: false, error: err, results: [] });
3540
throw err;
3641
}
3742

43+
const total = configs.length;
44+
if (total === 0) {
45+
setState({ loading: false, error: null, results: [] });
46+
return [];
47+
}
48+
3849
setState({ loading: true, error: null, results: [] });
3950

4051
try {
41-
const results = await client.streams.createBatchStreams(configs);
52+
const results: BatchCreateStreamResult[] = [];
53+
const chunkSize = 10;
54+
for (let index = 0; index < configs.length; index += chunkSize) {
55+
const chunk = configs.slice(index, index + chunkSize);
56+
const chunkResults = await client.streams.createBatchStreams(chunk);
57+
results.push(...chunkResults);
58+
onProgress?.(Math.min(index + chunk.length, total), total);
59+
}
60+
4261
setState({ loading: false, error: null, results });
4362
return results;
4463
} catch (err: unknown) {

‎packages/react/src/hooks/usePauseStream.ts‎

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import { useCallback } from 'react';
2+
import { useStream, type UseStreamResult } from './useStream.js';
23
import { useStreamMutation, type StreamMutationId, type UseStreamMutationState } from './useStreamMutation.js';
34

45
export type PauseStreamFn = () => Promise<string>;
@@ -10,8 +11,22 @@ export interface UsePauseStreamResult extends UseStreamMutationState {
1011
}
1112

1213
export function usePauseStream(streamId: StreamMutationId): UsePauseStreamResult {
14+
const { stream, setStream } = useStream(streamId) as UseStreamResult;
1315
const mutation = useStreamMutation(streamId, (client, id) => client.streams.pause(id));
14-
const pause = useCallback<PauseStreamFn>(() => mutation.mutate(), [mutation]);
16+
17+
const pause = useCallback<PauseStreamFn>(async () => {
18+
const previous = stream;
19+
if (stream) {
20+
setStream((current) => current ? { ...current, paused: true, pausedAt: Math.floor(Date.now() / 1000), cancelled: false } : current);
21+
}
22+
23+
try {
24+
return await mutation.mutate();
25+
} catch (error) {
26+
if (previous) setStream(previous);
27+
throw error;
28+
}
29+
}, [mutation, setStream, stream]);
1530

1631
return { ...mutation, mutate: pause, pause };
1732
}

‎packages/react/src/hooks/useResumeStream.ts‎

Lines changed: 30 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,14 @@
11
import { useCallback } from 'react';
2+
import type { StreamInfo } from '@conduit-protocol/sdk';
3+
import { useStream, type UseStreamResult } from './useStream.js';
24
import { useStreamMutation, type StreamMutationId, type UseStreamMutationState } from './useStreamMutation.js';
35

4-
export type ResumeStreamFn = () => Promise<string>;
6+
export interface ResumeStreamResult {
7+
txHash: string;
8+
stream: StreamInfo | null;
9+
}
10+
11+
export type ResumeStreamFn = () => Promise<ResumeStreamResult>;
512

613
export interface UseResumeStreamResult extends UseStreamMutationState {
714
mutate: ResumeStreamFn;
@@ -10,8 +17,28 @@ export interface UseResumeStreamResult extends UseStreamMutationState {
1017
}
1118

1219
export function useResumeStream(streamId: StreamMutationId): UseResumeStreamResult {
13-
const mutation = useStreamMutation(streamId, (client, id) => client.streams.resume(id));
14-
const resume = useCallback<ResumeStreamFn>(() => mutation.mutate(), [mutation]);
20+
const { stream, setStream } = useStream(streamId) as UseStreamResult;
21+
const mutation = useStreamMutation(streamId, async (client, id) => {
22+
const txHash = await client.streams.resume(id);
23+
const nextStream = typeof client.streams?.get === 'function'
24+
? await client.streams.get(id).catch(() => null)
25+
: null;
26+
return { txHash, stream: nextStream };
27+
});
28+
29+
const resume = useCallback<ResumeStreamFn>(async () => {
30+
const previous = stream;
31+
if (stream) {
32+
setStream((current) => current ? { ...current, paused: false, pausedAt: 0 } : current);
33+
}
34+
35+
try {
36+
return await mutation.mutate();
37+
} catch (error) {
38+
if (previous) setStream(previous);
39+
throw error;
40+
}
41+
}, [mutation, setStream, stream]);
1542

1643
return { ...mutation, mutate: resume, resume };
1744
}

‎packages/react/src/hooks/useStream.ts‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { useState, useEffect, useCallback, useRef } from 'react';
1+
import { useState, useEffect, useCallback, useRef, type Dispatch, type SetStateAction } from 'react';
22
import { useStreamFiClient } from '../context/useStreamFiClient.js';
33
import type { StreamInfo } from '@conduit-protocol/sdk';
44

@@ -7,6 +7,7 @@ export interface UseStreamResult {
77
loading: boolean;
88
error: Error | null;
99
refetch: () => void;
10+
setStream: Dispatch<SetStateAction<StreamInfo | null>>;
1011
}
1112

1213
export function useStream(streamId: bigint | string | null | undefined): UseStreamResult {
@@ -22,7 +23,7 @@ export function useStream(streamId: bigint | string | null | undefined): UseStre
2223
const fetch = useCallback(() => {
2324
const requestId = ++requestIdRef.current;
2425

25-
if (!isReady || !client || streamId == null) {
26+
if (!isReady || !client || streamId == null || typeof client.streams?.get !== 'function') {
2627
setStream(null);
2728
setLoading(false);
2829
setError(null);
@@ -50,5 +51,5 @@ export function useStream(streamId: bigint | string | null | undefined): UseStre
5051
fetch();
5152
}, [fetch]);
5253

53-
return { stream, loading, error, refetch: fetch };
54+
return { stream, loading, error, refetch: fetch, setStream };
5455
}

‎packages/react/src/hooks/useStreamMutation.ts‎

Lines changed: 17 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -10,15 +10,23 @@ export interface UseStreamMutationState {
1010
txHash: string | null;
1111
}
1212

13-
export interface UseStreamMutationResult<TArgs extends unknown[]> extends UseStreamMutationState {
14-
mutate: (...args: TArgs) => Promise<string>;
13+
export interface UseStreamMutationResult<TArgs extends unknown[], TResult = string> extends UseStreamMutationState {
14+
mutate: (...args: TArgs) => Promise<TResult>;
1515
reset: () => void;
1616
}
1717

18-
export function useStreamMutation<TArgs extends unknown[]>(
18+
function extractTxHash<TResult>(result: TResult): string | null {
19+
if (typeof result === 'string') return result;
20+
if (result && typeof result === 'object' && 'txHash' in result && typeof (result as { txHash?: unknown }).txHash === 'string') {
21+
return (result as { txHash: string }).txHash;
22+
}
23+
return null;
24+
}
25+
26+
export function useStreamMutation<TArgs extends unknown[], TResult = string>(
1927
streamId: StreamMutationId,
20-
action: (client: ConduitClient, streamId: bigint | string, ...args: TArgs) => Promise<string>,
21-
): UseStreamMutationResult<TArgs> {
28+
action: (client: ConduitClient, streamId: bigint | string, ...args: TArgs) => Promise<TResult>,
29+
): UseStreamMutationResult<TArgs, TResult> {
2230
const { client, isReady } = useStreamFiClient();
2331
const [state, setState] = useState<UseStreamMutationState>({
2432
isPending: false,
@@ -31,7 +39,7 @@ export function useStreamMutation<TArgs extends unknown[]>(
3139
}, []);
3240

3341
const mutate = useCallback(
34-
async (...args: TArgs): Promise<string> => {
42+
async (...args: TArgs): Promise<TResult> => {
3543
if (!isReady || !client) {
3644
const err = new Error('ConduitClient is not connected. Wrap your component with <StreamFiProvider>.');
3745
setState({ isPending: false, error: err, txHash: null });
@@ -47,9 +55,9 @@ export function useStreamMutation<TArgs extends unknown[]>(
4755
setState({ isPending: true, error: null, txHash: null });
4856

4957
try {
50-
const txHash = await action(client, streamId, ...args);
51-
setState({ isPending: false, error: null, txHash });
52-
return txHash;
58+
const result = await action(client, streamId, ...args);
59+
setState({ isPending: false, error: null, txHash: extractTxHash(result) });
60+
return result;
5361
} catch (err: unknown) {
5462
const typed = err instanceof Error ? err : new Error(String(err));
5563
setState({ isPending: false, error: typed, txHash: null });

‎src/events.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,8 @@ import type {
1515
ResumeEvent,
1616
StreamTopUpEvent,
1717
StreamClawbackedEvent,
18+
TopUpEvent,
19+
ClawbackEvent,
1820
CreatedEvent,
1921
ForceCancelEvent,
2022
RecipientTransferEvent,

0 commit comments

Comments
 (0)