homeio

1"use client";
2
3import type {
4  StoreOperation,
5  StoreOperationEvent,
6} from "@/lib/shared/contracts/apps";
7import { queryKeys } from "@/lib/shared/query-keys";
8import { useQuery, useQueryClient } from "@tanstack/react-query";
9import { useEffect, useMemo, useState } from "react";
10
11type StoreOperationResponse = {
12  data: StoreOperation;
13};
14
15type StoreOperationSubscriptionOptions = {
16  onEvent?: (event: StoreOperationEvent) => void;
17  onError?: (error: unknown) => void;
18};
19
20function isTerminalStatus(status: string) {
21  return status === "success" || status === "error";
22}
23
24function eventToOperation(
25  event: StoreOperationEvent,
26  previous: StoreOperation | null | undefined,
27): StoreOperation {
28  return {
29    id: event.operationId,
30    appId: event.appId,
31    action: event.action,
32    status: event.status,
33    progressPercent: event.progressPercent,
34    currentStep: event.step,
35    errorMessage:
36      event.status === "error"
37        ? (event.message ?? previous?.errorMessage ?? "Operation failed")
38        : null,
39    startedAt: previous?.startedAt ?? event.timestamp,
40    finishedAt:
41      event.status === "success" || event.status === "error"
42        ? event.timestamp
43        : (previous?.finishedAt ?? null),
44    updatedAt: event.timestamp,
45  };
46}
47
48export async function fetchStoreOperationSnapshot(operationId: string) {
49  const endpoint = `/api/v1/store/operations/${encodeURIComponent(operationId)}`;
50
51  const response = await fetch(endpoint, {
52    cache: "no-store",
53  });
54
55  if (response.status === 404) {
56    return null;
57  }
58
59  if (!response.ok) {
60    throw new Error(`Failed to fetch store operation (${response.status})`);
61  }
62
63  const json = (await response.json()) as StoreOperationResponse;
64  return json.data;
65}
66
67export function subscribeToStoreOperationEvents(
68  operationId: string,
69  options?: StoreOperationSubscriptionOptions,
70) {
71  const endpoint = `/api/v1/store/operations/${encodeURIComponent(operationId)}/stream`;
72  const source = new EventSource(endpoint);
73
74  const eventTypes: StoreOperationEvent["type"][] = [
75    "operation.started",
76    "operation.step",
77    "operation.pull.progress",
78    "operation.completed",
79    "operation.failed",
80  ];
81
82  const listeners: Array<{
83    type: string;
84    listener: EventListener;
85  }> = [];
86
87  for (const type of eventTypes) {
88    const listener: EventListener = (rawEvent) => {
89      const event = rawEvent as MessageEvent<string>;
90      try {
91        const payload = JSON.parse(event.data) as StoreOperationEvent;
92        options?.onEvent?.(payload);
93      } catch (error) {
94        options?.onError?.(error);
95      }
96    };
97
98    source.addEventListener(type, listener);
99    listeners.push({ type, listener });
100  }
101
102  source.addEventListener("error", (error) => {
103    options?.onError?.(error);
104  });
105
106  return () => {
107    listeners.forEach(({ type, listener }) => {
108      source.removeEventListener(type, listener);
109    });
110    source.close();
111  };
112}
113
114export function useStoreOperation(operationId: string | null) {
115  const queryClient = useQueryClient();
116  const [latestEvent, setLatestEvent] = useState<StoreOperationEvent | null>(
117    null,
118  );
119
120  const snapshotQuery = useQuery({
121    queryKey: operationId
122      ? queryKeys.storeOperation(operationId)
123      : ["store", "operation", "none"],
124    queryFn: () => fetchStoreOperationSnapshot(operationId as string),
125    enabled: Boolean(operationId),
126    staleTime: 5_000,
127    refetchInterval: (query) => {
128      const status = query.state.data?.status;
129      return status && isTerminalStatus(status) ? false : 4_000;
130    },
131    refetchOnWindowFocus: false,
132  });
133
134  useEffect(() => {
135    if (!operationId) {
136      setLatestEvent(null);
137      return;
138    }
139
140    const unsubscribe = subscribeToStoreOperationEvents(operationId, {
141      onEvent: (event) => {
142        setLatestEvent(event);
143
144        queryClient.setQueryData<StoreOperation | null>(
145          queryKeys.storeOperation(operationId),
146          (previous) => eventToOperation(event, previous),
147        );
148
149        if (isTerminalStatus(event.status)) {
150          void Promise.all([
151            queryClient.invalidateQueries({ queryKey: queryKeys.storeCatalog }),
152            queryClient.invalidateQueries({
153              queryKey: queryKeys.installedApps,
154            }),
155            queryClient.invalidateQueries({
156              queryKey: queryKeys.storeApp(event.appId),
157            }),
158            queryClient.invalidateQueries({
159              queryKey: queryKeys.storeOperation(operationId),
160            }),
161          ]);
162        }
163      },
164      onError: (_error) => {
165        // SSE stream errors are transient (reconnects, auth expiry).
166        // The operation state from the last successful event is preserved in the query cache.
167      },
168    });
169
170    return () => {
171      unsubscribe();
172    };
173  }, [operationId, queryClient]);
174
175  const operation = useMemo(() => {
176    if (!snapshotQuery.data) {
177      return latestEvent ? eventToOperation(latestEvent, null) : null;
178    }
179
180    if (!latestEvent) return snapshotQuery.data;
181    return eventToOperation(latestEvent, snapshotQuery.data);
182  }, [latestEvent, snapshotQuery.data]);
183
184  return {
185    operation,
186    latestEvent,
187    isLoading: snapshotQuery.isLoading,
188    isError: snapshotQuery.isError,
189    error: snapshotQuery.error,
190  };
191}
192