feat: align investment client with hourly sync
This commit is contained in:
@@ -1,24 +1,8 @@
|
||||
import {
|
||||
focusManager,
|
||||
onlineManager,
|
||||
type QueryKey,
|
||||
useQueryClient,
|
||||
} from "@tanstack/react-query";
|
||||
import type { QueryKey } from "@tanstack/react-query";
|
||||
|
||||
import { useEffect, useRef } from "react";
|
||||
|
||||
import {
|
||||
ApiErrorCode,
|
||||
apiErrorCode,
|
||||
apiQueryClient,
|
||||
invalidateResourceAccess,
|
||||
type InvestmentResourceMeta,
|
||||
isApiErrorCode,
|
||||
type paths,
|
||||
useAuthenticatedQueries,
|
||||
} from "@/shared/api";
|
||||
|
||||
type InvestmentScreen
|
||||
= paths["/api/v1/tbank-sync"]["post"]["requestBody"]["content"]["application/json"];
|
||||
import type { InvestmentResourceMeta } from "@/shared/api";
|
||||
|
||||
type ScreenResource = {
|
||||
isError: boolean;
|
||||
@@ -29,241 +13,43 @@ type ScreenResource = {
|
||||
refetch: () => Promise<unknown>;
|
||||
};
|
||||
|
||||
const STATUS_INTERVAL_MS = 5000;
|
||||
const FIRST_SOURCE_RECOVERY_DELAY_MS = 5000;
|
||||
const SECOND_SOURCE_RECOVERY_DELAY_MS = 15_000;
|
||||
type ScreenScope = {
|
||||
[key: string]: unknown;
|
||||
screen: string;
|
||||
};
|
||||
|
||||
function requiresSourceRecovery(meta?: InvestmentResourceMeta): boolean {
|
||||
if (!meta || meta.unavailableReason) return false;
|
||||
const READ_INTERVAL_MS = 1_800_000;
|
||||
|
||||
return ["cancelled", "failed"].includes(meta.sync.state)
|
||||
|| meta.availability === "missing"
|
||||
|| meta.freshness === "stale"
|
||||
|| (meta.dependencies ?? []).some(dependency =>
|
||||
requiresSourceRecovery(dependency),
|
||||
);
|
||||
}
|
||||
|
||||
function sourceRecoveryDelay(attempt: number): number | undefined {
|
||||
switch (attempt) {
|
||||
case 0: {
|
||||
return FIRST_SOURCE_RECOVERY_DELAY_MS;
|
||||
}
|
||||
|
||||
case 1: {
|
||||
return SECOND_SOURCE_RECOVERY_DELAY_MS;
|
||||
}
|
||||
|
||||
default: {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function useScreenSync(scope: InvestmentScreen, resources: ScreenResource[]) {
|
||||
const mutation = apiQueryClient.useMutation("post", "/api/v1/tbank-sync", {
|
||||
retry: false,
|
||||
});
|
||||
|
||||
const client = useQueryClient();
|
||||
const deniedRef = useRef(new Set<unknown>());
|
||||
const scopeKey = JSON.stringify(scope);
|
||||
|
||||
const requestInFlightRef = useRef<null | string>(null);
|
||||
const lifecycleRecoveryRef = useRef(new Set<string>());
|
||||
const sourceRecoveryRef = useRef(new Map<string, number>());
|
||||
const sourceTimersRef = useRef(new Set<ReturnType<typeof setTimeout>>());
|
||||
|
||||
const isCurrentMutation
|
||||
= JSON.stringify(mutation.variables?.body) === scopeKey;
|
||||
|
||||
const observedRef = useRef("");
|
||||
|
||||
const ids = [
|
||||
...new Set([
|
||||
...(isCurrentMutation && mutation.data ? [mutation.data.id] : []),
|
||||
...resources
|
||||
.flatMap(({ meta }) =>
|
||||
meta ? [meta, ...(meta.dependencies ?? [])] : [],
|
||||
)
|
||||
.flatMap(meta =>
|
||||
meta.sync.id
|
||||
&& meta.sync.state === "running"
|
||||
? [meta.sync.id]
|
||||
: [],
|
||||
),
|
||||
]),
|
||||
];
|
||||
|
||||
const statuses = useAuthenticatedQueries(
|
||||
ids.map(id => ({
|
||||
...apiQueryClient.queryOptions("get", "/api/v1/tbank-sync/{id}", {
|
||||
params: { path: { id } },
|
||||
}),
|
||||
refetchInterval: (query: {
|
||||
state: { data?: { state: string }; error: unknown };
|
||||
}) =>
|
||||
!(
|
||||
query.state.error
|
||||
|| (query.state.data && query.state.data.state !== "running")
|
||||
) && STATUS_INTERVAL_MS,
|
||||
refetchOnMount: false,
|
||||
refetchOnWindowFocus: false,
|
||||
retry: false,
|
||||
})),
|
||||
);
|
||||
function useScreenSync(_scope: ScreenScope, resources: ScreenResource[]) {
|
||||
const resourcesRef = useRef(resources);
|
||||
useEffect(() => {
|
||||
resourcesRef.current = resources;
|
||||
}, [resources]);
|
||||
|
||||
useEffect(() => {
|
||||
for (const status of statuses) {
|
||||
if (!status.error || deniedRef.current.has(status.error)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
deniedRef.current.add(status.error);
|
||||
if (!isApiErrorCode(apiErrorCode(status.error), ApiErrorCode.NotFound))
|
||||
invalidateResourceAccess(client, "brokerage", status.error);
|
||||
}
|
||||
|
||||
if (statuses.every(status => !status.data)) return;
|
||||
|
||||
const observation = JSON.stringify([
|
||||
scopeKey,
|
||||
statuses.map(status => [
|
||||
status.data?.id,
|
||||
status.data?.state,
|
||||
status.dataUpdatedAt,
|
||||
]),
|
||||
resources.map(resource => resource.queryKey),
|
||||
]);
|
||||
|
||||
if (observedRef.current === observation) return;
|
||||
observedRef.current = observation;
|
||||
|
||||
const isCompleted = statuses.every(
|
||||
status => status.data && status.data.state !== "running",
|
||||
);
|
||||
|
||||
for (const resource of resources)
|
||||
if (isCompleted || !resource.isFetching)
|
||||
void resource.observe().catch(() => {});
|
||||
}, [client, statuses, resources, scopeKey]);
|
||||
|
||||
useEffect(() => {
|
||||
const timers = sourceTimersRef.current;
|
||||
const attempts = sourceRecoveryRef.current;
|
||||
|
||||
return () => {
|
||||
for (const timer of timers) clearTimeout(timer);
|
||||
timers.clear();
|
||||
attempts.clear();
|
||||
};
|
||||
}, [scopeKey]);
|
||||
|
||||
useEffect(() => {
|
||||
for (const resource of resources) {
|
||||
if (
|
||||
!requiresSourceRecovery(resource.meta)
|
||||
|| resource.isFetching
|
||||
) continue;
|
||||
|
||||
const key = JSON.stringify([scopeKey, resource.queryKey]);
|
||||
const attempt = sourceRecoveryRef.current.get(key) ?? 0;
|
||||
const delay = sourceRecoveryDelay(attempt);
|
||||
if (delay === undefined) continue;
|
||||
|
||||
sourceRecoveryRef.current.set(key, attempt + 1);
|
||||
|
||||
const timer = setTimeout(() => {
|
||||
sourceTimersRef.current.delete(timer);
|
||||
const timer = setInterval(() => {
|
||||
for (const resource of resourcesRef.current)
|
||||
void resource.refetch().catch(() => {});
|
||||
}, delay);
|
||||
}, READ_INTERVAL_MS);
|
||||
|
||||
sourceTimersRef.current.add(timer);
|
||||
}
|
||||
}, [resources, scopeKey]);
|
||||
return () => clearInterval(timer);
|
||||
}, []);
|
||||
|
||||
useEffect(() => {
|
||||
const recover = () => {
|
||||
for (const resource of resources) {
|
||||
if (
|
||||
!requiresSourceRecovery(resource.meta)
|
||||
|| resource.isFetching
|
||||
) continue;
|
||||
|
||||
const key = JSON.stringify([scopeKey, resource.queryKey]);
|
||||
if (lifecycleRecoveryRef.current.has(key)) continue;
|
||||
|
||||
sourceRecoveryRef.current.delete(key);
|
||||
lifecycleRecoveryRef.current.add(key);
|
||||
void resource.refetch()
|
||||
.catch(() => {})
|
||||
.finally(() => lifecycleRecoveryRef.current.delete(key));
|
||||
}
|
||||
};
|
||||
|
||||
const unsubscribeFocus = focusManager.subscribe((focused) => {
|
||||
if (focused) recover();
|
||||
});
|
||||
|
||||
const unsubscribeOnline = onlineManager.subscribe((online) => {
|
||||
if (online) recover();
|
||||
});
|
||||
|
||||
return () => {
|
||||
unsubscribeFocus();
|
||||
unsubscribeOnline();
|
||||
};
|
||||
}, [resources, scopeKey]);
|
||||
|
||||
const isRefreshing
|
||||
= (isCurrentMutation && mutation.isPending)
|
||||
|| statuses.some(
|
||||
status => !status.isError && status.data?.state === "running",
|
||||
)
|
||||
|| resources.some(({ meta }) => {
|
||||
const status = statuses[ids.indexOf(meta?.sync.id ?? "")];
|
||||
|
||||
return (
|
||||
!status?.isError
|
||||
&& (!status?.data || status.data.state === "running")
|
||||
&& meta?.sync.state === "running"
|
||||
);
|
||||
});
|
||||
const isRefreshing = resources.some(resource => resource.isFetching);
|
||||
|
||||
const refresh = async () => {
|
||||
if (isRefreshing || requestInFlightRef.current === scopeKey) return;
|
||||
requestInFlightRef.current = scopeKey;
|
||||
|
||||
try {
|
||||
await mutation.mutateAsync({ body: scope });
|
||||
}
|
||||
catch {
|
||||
/*
|
||||
The mutation retains the transport error for presentation.
|
||||
*/
|
||||
}
|
||||
finally {
|
||||
if (requestInFlightRef.current === scopeKey)
|
||||
requestInFlightRef.current
|
||||
|
||||
= null;
|
||||
}
|
||||
if (isRefreshing) return;
|
||||
await Promise.allSettled(resources.map(resource => resource.refetch()));
|
||||
};
|
||||
|
||||
// Retry failed HTTP reads without asking the source to import again.
|
||||
// A source failure (or explicit refresh with no read error) still needs a sync.
|
||||
const retry = async () => {
|
||||
const failed = resources.filter(
|
||||
resource => resource.isError,
|
||||
);
|
||||
|
||||
const failedStatuses = statuses.filter(status => status.isError);
|
||||
if (failed.length > 0 || failedStatuses.length > 0) {
|
||||
await Promise.allSettled([
|
||||
...failed.map(resource => resource.observe()),
|
||||
...failedStatuses.map(status => status.refetch()),
|
||||
]);
|
||||
}
|
||||
if (failed.length > 0)
|
||||
await Promise.allSettled(failed.map(resource => resource.observe()));
|
||||
else await refresh();
|
||||
};
|
||||
|
||||
@@ -271,13 +57,9 @@ function useScreenSync(scope: InvestmentScreen, resources: ScreenResource[]) {
|
||||
refresh,
|
||||
refreshing: isRefreshing,
|
||||
retry,
|
||||
sourceError:
|
||||
statuses.some(status => status.data?.state === "failed")
|
||||
|| resources.some(({ meta }) => meta?.sync.state === "failed"),
|
||||
transportError:
|
||||
(isCurrentMutation && mutation.isError)
|
||||
|| statuses.some(status => status.isError),
|
||||
sourceError: resources.some(({ meta }) => meta?.sync.state === "failed"),
|
||||
transportError: resources.some(resource => resource.isError),
|
||||
};
|
||||
}
|
||||
|
||||
export { type InvestmentScreen, useScreenSync };
|
||||
export { useScreenSync };
|
||||
|
||||
Reference in New Issue
Block a user