feat: show persisted investment data and sync state

Consume the provided resource-envelope API snapshot. Preserve financial snapshots during
refresh, observe synchronization without importing, and fence late responses across
account and connection changes.

src/shared/api/authenticated-query.test.tsx

@@ -2,6 +2,7 @@ import { renderHook, waitFor } from "@testing-library/react";
 import { HttpResponse, http } from "msw";
 import { describe, expect, it, vi } from "vitest";
 import { createQueryWrapper } from "@/test/query";
+import { resource } from "@/test/resource";
 import { server } from "@/test/server";
 import {
 	useAuthenticatedQueries,
@@ -17,14 +18,16 @@ describe("useAuthenticatedQuery", () => {
 			server.use(
 				http.get("*/api/v1/brokerage-accounts", () => {
 					requests();
-					return HttpResponse.json([
-						{
-							id: "account-1",
-							name: "Основной",
-							type: "BROKERAGE",
-							openedOn: "2024-08-13",
-						},
-					]);
+					return HttpResponse.json(
+						resource([
+							{
+								id: "account-1",
+								name: "Основной",
+								type: "BROKERAGE",
+								openedOn: "2024-08-13",
+							},
+						]),
+					);
 				}),
 			);
 
@@ -53,7 +56,7 @@ describe("useAuthenticatedQuery", () => {
 
 			await waitFor(() => expect(result.current.isSuccess).toBe(true));
 			expect(requests).toHaveBeenCalledOnce();
-			expect(result.current.data).toEqual([
+			expect(result.current.data?.data).toEqual([
 				{
 					id: "account-1",
 					name: "Основной",

src/shared/api/client.test.ts

@@ -1,5 +1,6 @@
 import { HttpResponse, http } from "msw";
 import { describe, expect, it, vi } from "vitest";
+import { resource } from "@/test/resource";
 import { server } from "@/test/server";
 import { apiClient, configureAuthClient } from "./client";
 
@@ -16,14 +17,14 @@ describe("apiClient authentication", () => {
 				authorization.push(request.headers.get("Authorization") ?? "");
 				return authorization.length === 1
 					? HttpResponse.json({ code: "UNAUTHORIZED" }, { status: 401 })
-					: HttpResponse.json([]);
+					: HttpResponse.json(resource([]));
 			}),
 		);
 		configureAuthClient({ accessToken: () => token, refresh, clear: vi.fn() });
 
 		const { data } = await apiClient.GET("/api/v1/brokerage-accounts");
 
-		expect(data).toEqual([]);
+		expect(data?.data).toEqual([]);
 		expect(refresh).toHaveBeenCalledOnce();
 		expect(authorization).toEqual(["Bearer expired", "Bearer fresh"]);
 	});
@@ -76,7 +77,7 @@ describe("apiClient authentication", () => {
 				if (++calls === 1)
 					return HttpResponse.json({ code: "UNAUTHORIZED" }, { status: 401 });
 				await pending;
-				return HttpResponse.json([{ id: "previous-user-account" }]);
+				return HttpResponse.json(resource([{ id: "previous-user-account" }]));
 			}),
 		);
 		configureAuthClient({

src/shared/api/client.ts

@@ -14,6 +14,13 @@ type AuthHandlers = {
 	clear: () => void;
 };
 
+const inFlight = new Set<{ pathname: string; invalidated: boolean }>();
+const latestReads = new Map<string, { sequence: number; pending: number }>();
+function invalidateApiRequests(matches: (pathname: string) => boolean) {
+	for (const request of inFlight)
+		if (matches(request.pathname)) request.invalidated = true;
+}
+
 let auth: AuthHandlers | null = null;
 let hasAuthenticatedSession = false;
 const authSessionListeners = new Set<() => void>();
@@ -25,6 +32,7 @@ function setAuthenticatedSession(next: boolean) {
 }
 
 function configureAuthClient(handlers: AuthHandlers | null) {
+	if (!handlers) invalidateApiRequests(() => true);
 	auth = handlers;
 	setAuthenticatedSession(Boolean(auth?.accessToken()));
 }
@@ -39,12 +47,47 @@ function subscribeAuthenticatedSession(listener: () => void) {
 }
 
 async function authenticatedFetch(request: Request): Promise<Response> {
+	const url = new URL(request.url);
+	url.searchParams.delete("sync");
+	url.searchParams.sort();
+	const key = `${request.method} ${url}`;
+	const ticket = { pathname: url.pathname, invalidated: false };
+	const read =
+		request.method === "GET"
+			? (latestReads.get(key) ?? { sequence: 0, pending: 0 })
+			: null;
+	if (read) {
+		read.sequence += 1;
+		read.pending += 1;
+		latestReads.set(key, read);
+	}
+	const sequence = read?.sequence;
+	inFlight.add(ticket);
+	try {
+		return await fetchCurrent(
+			request,
+			() => !ticket.invalidated && (!read || sequence === read.sequence),
+		);
+	} finally {
+		inFlight.delete(ticket);
+		if (read && --read.pending === 0) latestReads.delete(key);
+	}
+}
+
+async function fetchCurrent(
+	request: Request,
+	isLatest: () => boolean,
+): Promise<Response> {
 	const isAuthRequest = new URL(request.url).pathname.startsWith(
 		"/api/v1/auth/",
 	);
 	const session = auth;
 	const generation = session?.generation?.();
-	const isCurrent = () => generation === auth?.generation?.();
+	const isCurrent = () =>
+		(session?.generation
+			? generation === session.generation() &&
+				generation === auth?.generation?.()
+			: session === auth) && isLatest();
 	const token = session?.accessToken();
 	const withToken = (source: Request, value: string) => {
 		const headers = new Headers(source.headers);
@@ -55,6 +98,7 @@ async function authenticatedFetch(request: Request): Promise<Response> {
 		token && !isAuthRequest ? withToken(request, token) : request;
 	const retryRequest = authorized.clone();
 	const response = await fetch(authorized);
+	await response.clone().arrayBuffer();
 	if (!isAuthRequest && !isCurrent())
 		throw new DOMException("Session changed", "AbortError");
 	if (
@@ -78,6 +122,7 @@ async function authenticatedFetch(request: Request): Promise<Response> {
 	const refreshedToken = auth.accessToken();
 	if (!refreshedToken) return response;
 	const retried = await fetch(withToken(retryRequest, refreshedToken));
+	await retried.clone().arrayBuffer();
 	if (!isCurrent()) throw new DOMException("Session changed", "AbortError");
 	return retried;
 }
@@ -95,5 +140,6 @@ export {
 	apiQueryClient,
 	configureAuthClient,
 	getAuthenticatedSessionSnapshot,
+	invalidateApiRequests,
 	subscribeAuthenticatedSession,
 };

src/shared/api/index.ts

@@ -2,6 +2,17 @@ export {
 	useAuthenticatedQueries,
 	useAuthenticatedQuery,
 } from "./authenticated-query";
-export { apiClient, apiQueryClient, configureAuthClient } from "./client";
+export {
+	apiClient,
+	apiQueryClient,
+	configureAuthClient,
+	invalidateApiRequests,
+} from "./client";
 export type { paths } from "./generated/schema";
 export { apiErrorCode } from "./problem-details";
+
+export {
+	invalidateResourceAccess,
+	unwrapResourceResult,
+	useResourceQuery,
+} from "./resource-query";

src/shared/api/resource-query.ts

@@ -0,0 +1,150 @@
+import type { QueryClient } from "@tanstack/react-query";
+import {
+	type QueryFunctionContext,
+	type QueryKey,
+	type UseQueryOptions,
+	type UseQueryResult,
+	useQueryClient,
+} from "@tanstack/react-query";
+import { useEffect } from "react";
+import { useAuthenticatedQuery } from "./authenticated-query";
+import type { components } from "./generated/schema";
+import { apiErrorCode } from "./problem-details";
+
+const accessEpochs = new WeakMap<QueryClient, Map<string, number>>();
+
+type Resource = {
+	data: unknown;
+	meta: components["schemas"]["InvestmentResourceMetaDto"];
+};
+function unwrapResourceResult<T extends Resource, E>(
+	result: UseQueryResult<T, E>,
+) {
+	const missing = result.isSuccess && result.data.data === null;
+	return {
+		...result,
+		transport: {
+			status: result.status,
+			fetchStatus: result.fetchStatus,
+			error: result.error,
+		},
+		data: result.data?.data ?? undefined,
+		meta: result.data?.meta,
+		isPending: result.isPending || missing,
+		isSuccess: result.isSuccess && !missing,
+		status: missing ? "pending" : result.status,
+	} as UseQueryResult<NonNullable<T["data"]>, E> & {
+		meta?: T["meta"];
+		transport: {
+			status: UseQueryResult<T, E>["status"];
+			fetchStatus: UseQueryResult<T, E>["fetchStatus"];
+			error: E | null;
+		};
+	};
+}
+
+function invalidateResourceAccess(
+	client: QueryClient,
+	group: string,
+	error: unknown,
+) {
+	if (
+		![
+			"NOT_FOUND",
+			"FORBIDDEN",
+			"TBANK_NOT_CONNECTED",
+			"INVALID_TBANK_TOKEN",
+		].includes(apiErrorCode(error) ?? "")
+	)
+		return;
+	const epochs = accessEpochs.get(client) ?? new Map<string, number>();
+	accessEpochs.set(client, epochs);
+	epochs.set(group, (epochs.get(group) ?? 0) + 1);
+	const affected = {
+		predicate: (cached: { meta?: Record<string, unknown> }) =>
+			cached.meta?.accessGroup === group,
+	};
+	void client.cancelQueries(affected);
+	for (const cached of client.getQueryCache().findAll(affected))
+		cached.setState({
+			data: undefined,
+			dataUpdatedAt: 0,
+			status: "error",
+			error: error as Error,
+			fetchStatus: "idle",
+		});
+}
+
+// Observation changes transport mode, never the generated cache identity.
+function useResourceQuery<T extends Resource, E, K extends QueryKey>(
+	options: UseQueryOptions<T, E, T, K>,
+) {
+	const client = useQueryClient();
+	const group =
+		typeof options.meta?.accessGroup === "string"
+			? options.meta.accessGroup
+			: undefined;
+	let epochs = accessEpochs.get(client);
+	if (!epochs) {
+		epochs = new Map();
+		accessEpochs.set(client, epochs);
+	}
+	const groupEpochs = epochs;
+	const queryFn = options.queryFn;
+	const guarded =
+		typeof queryFn === "function"
+			? async (context: QueryFunctionContext<K>) => {
+					const epoch = group ? (groupEpochs.get(group) ?? 0) : 0;
+					const result = await queryFn(context);
+					if (group && epoch !== (groupEpochs.get(group) ?? 0))
+						throw new DOMException("Access changed", "AbortError");
+					return result;
+				}
+			: queryFn;
+	const query = useAuthenticatedQuery({ ...options, queryFn: guarded });
+	useEffect(() => {
+		if (group) invalidateResourceAccess(client, group, query.error);
+	}, [client, group, query.error]);
+	const observe = async (expectedStateVersion?: string) => {
+		const queryFn = guarded;
+		if (typeof queryFn !== "function") return;
+		await client.cancelQueries({ queryKey: options.queryKey, exact: true });
+		return client.fetchQuery({
+			...options,
+			staleTime: 0,
+			retry: false,
+			queryFn: async (context: QueryFunctionContext<K>) => {
+				const [method, path, input] = context.queryKey;
+				const init = input as
+					| { params?: { query?: Record<string, unknown> } }
+					| undefined;
+				const queryKey = [
+					method,
+					path,
+					{
+						...init,
+						params: {
+							...init?.params,
+							query: { ...init?.params?.query, sync: "observe" },
+						},
+					},
+				] as unknown as K;
+				const response = await queryFn({ ...context, queryKey });
+				if (
+					expectedStateVersion &&
+					response.meta.stateVersion !== expectedStateVersion
+				) {
+					// One more observational read; never turn a stale notification into an import.
+					const current = await queryFn({ ...context, queryKey });
+					if (current.meta.stateVersion !== expectedStateVersion)
+						throw new DOMException("Resource changed", "AbortError");
+					return current;
+				}
+				return response;
+			},
+		});
+	};
+	return { ...unwrapResourceResult(query), observe };
+}
+
+export { invalidateResourceAccess, unwrapResourceResult, useResourceQuery };