diff --git a/apps/web/hooks/processing-poll-cycle.test.ts b/apps/web/hooks/processing-poll-cycle.test.ts new file mode 100644 index 00000000..d1ad4309 --- /dev/null +++ b/apps/web/hooks/processing-poll-cycle.test.ts @@ -0,0 +1,129 @@ +import { describe, expect, it } from "bun:test" +import { + decideProcessingPoll, + MAX_PROCESSING_POLLS, + type ProcessingPollCycle, +} from "./processing-poll-cycle" + +function activeSnapshot( + documentIds: readonly string[], + dataUpdateCount: number, + queryHash = "processing-scope-a", +) { + return { + documentIds, + totalCount: documentIds.length, + dataUpdateCount, + queryHash, + } +} + +describe("processing document poll cycles", () => { + it("stops the same processing job at the 60-update cap", () => { + let cycle: ProcessingPollCycle | null = null + + for ( + let updateCount = 1; + updateCount < MAX_PROCESSING_POLLS; + updateCount++ + ) { + const decision = decideProcessingPoll( + cycle, + activeSnapshot( + updateCount % 2 === 0 + ? ["document-a", "document-b"] + : ["document-b", "document-a"], + updateCount, + ), + ) + cycle = decision.cycle + expect(decision.shouldPoll).toBe(true) + } + + const capped = decideProcessingPoll( + cycle, + activeSnapshot(["document-a", "document-b"], MAX_PROCESSING_POLLS), + ) + expect(capped.shouldPoll).toBe(false) + }) + + it("restarts after an idle response even when the absolute count exceeds 60", () => { + let decision = decideProcessingPoll(null, activeSnapshot(["document-a"], 1)) + decision = decideProcessingPoll( + decision.cycle, + activeSnapshot(["document-a"], MAX_PROCESSING_POLLS), + ) + expect(decision.shouldPoll).toBe(false) + + decision = decideProcessingPoll(decision.cycle, { + documentIds: [], + totalCount: 0, + dataUpdateCount: MAX_PROCESSING_POLLS + 1, + queryHash: "processing-scope-a", + }) + expect(decision.shouldPoll).toBe(false) + + decision = decideProcessingPoll( + decision.cycle, + activeSnapshot(["document-a"], MAX_PROCESSING_POLLS + 2), + ) + expect(decision.shouldPoll).toBe(true) + }) + + it("starts a fresh budget when the query scope changes", () => { + let decision = decideProcessingPoll( + null, + activeSnapshot(["shared-document-id"], 1, "processing-scope-a"), + ) + decision = decideProcessingPoll( + decision.cycle, + activeSnapshot( + ["shared-document-id"], + MAX_PROCESSING_POLLS, + "processing-scope-a", + ), + ) + expect(decision.shouldPoll).toBe(false) + + decision = decideProcessingPoll( + decision.cycle, + activeSnapshot( + ["shared-document-id"], + MAX_PROCESSING_POLLS, + "processing-scope-b", + ), + ) + expect(decision.shouldPoll).toBe(true) + }) + + it("starts a fresh budget when the processing document IDs change", () => { + let decision = decideProcessingPoll(null, activeSnapshot(["document-a"], 1)) + decision = decideProcessingPoll( + decision.cycle, + activeSnapshot(["document-a"], MAX_PROCESSING_POLLS), + ) + expect(decision.shouldPoll).toBe(false) + + decision = decideProcessingPoll( + decision.cycle, + activeSnapshot(["document-b"], MAX_PROCESSING_POLLS + 1), + ) + expect(decision.shouldPoll).toBe(true) + }) + + it("does not spend the budget when the callback re-evaluates unchanged data", () => { + let decision = decideProcessingPoll(null, activeSnapshot(["document-a"], 1)) + + for ( + let evaluation = 0; + evaluation < MAX_PROCESSING_POLLS * 2; + evaluation++ + ) { + decision = decideProcessingPoll( + decision.cycle, + activeSnapshot(["document-a"], 1), + ) + expect(decision.shouldPoll).toBe(true) + } + }) +}) diff --git a/apps/web/hooks/processing-poll-cycle.ts b/apps/web/hooks/processing-poll-cycle.ts new file mode 100644 index 00000000..b343126a --- /dev/null +++ b/apps/web/hooks/processing-poll-cycle.ts @@ -0,0 +1,61 @@ +export const MAX_PROCESSING_POLLS = 60 + +export type ProcessingPollCycle = { + documentIds: readonly string[] + firstDataUpdateCount: number + queryHash: string +} + +export type ProcessingPollSnapshot = { + documentIds: readonly string[] + totalCount: number + dataUpdateCount: number + queryHash: string +} + +export type ProcessingPollDecision = { + cycle: ProcessingPollCycle | null + shouldPoll: boolean +} + +function normalizedDocumentIds(documentIds: readonly string[]) { + return [...new Set(documentIds)].sort() +} + +function hasSameDocumentIds(left: readonly string[], right: readonly string[]) { + return ( + left.length === right.length && + left.every((documentId, index) => documentId === right[index]) + ) +} + +export function decideProcessingPoll( + cycle: ProcessingPollCycle | null, + snapshot: ProcessingPollSnapshot, +): ProcessingPollDecision { + if (snapshot.totalCount === 0) { + return { cycle: null, shouldPoll: false } + } + + const documentIds = normalizedDocumentIds(snapshot.documentIds) + const startsNewCycle = + cycle === null || + cycle.queryHash !== snapshot.queryHash || + !hasSameDocumentIds(cycle.documentIds, documentIds) || + snapshot.dataUpdateCount < cycle.firstDataUpdateCount + const nextCycle = startsNewCycle + ? { + documentIds, + firstDataUpdateCount: snapshot.dataUpdateCount, + queryHash: snapshot.queryHash, + } + : cycle + // React Query's count spans the cache lifetime, so make it cycle-relative. + const updatesInCycle = + snapshot.dataUpdateCount - nextCycle.firstDataUpdateCount + 1 + + return { + cycle: nextCycle, + shouldPoll: updatesInCycle < MAX_PROCESSING_POLLS, + } +} diff --git a/apps/web/hooks/use-processing-documents.ts b/apps/web/hooks/use-processing-documents.ts index 99e5ca56..0329266f 100644 --- a/apps/web/hooks/use-processing-documents.ts +++ b/apps/web/hooks/use-processing-documents.ts @@ -5,15 +5,29 @@ import { useQuery, useQueryClient } from "@tanstack/react-query" import { $fetch } from "@lib/api" import { useProject } from "@/stores" import { useAuth } from "@lib/auth-context" +import { + decideProcessingPoll, + type ProcessingPollCycle, +} from "./processing-poll-cycle" -const MAX_POLLS = 60 const POLL_INTERVAL_MS = 5_000 +type ProcessingDocument = { + id?: string | null + status?: string | null +} + +type ProcessingDocumentsData = { + documents?: ProcessingDocument[] + totalCount?: number +} + export function useProcessingDocuments() { const { user } = useAuth() const { effectiveContainerTags } = useProject() const queryClient = useQueryClient() const prevIdsRef = useRef>(new Set()) + const pollCycleRef = useRef(null) const { data } = useQuery({ queryKey: ["processing-documents", effectiveContainerTags], @@ -27,12 +41,17 @@ export function useProcessingDocuments() { }, enabled: !!user, refetchInterval: (query) => { - const count = - (query.state.data as { totalCount?: number } | undefined)?.totalCount ?? - 0 - const polls = query.state.dataUpdateCount - if (count === 0 || polls >= MAX_POLLS) return false - return POLL_INTERVAL_MS + const queryData = query.state.data as ProcessingDocumentsData | undefined + const decision = decideProcessingPoll(pollCycleRef.current, { + documentIds: (queryData?.documents ?? []).flatMap((document) => + document.id ? [document.id] : [], + ), + totalCount: queryData?.totalCount ?? 0, + dataUpdateCount: query.state.dataUpdateCount, + queryHash: query.queryHash, + }) + pollCycleRef.current = decision.cycle + return decision.shouldPoll ? POLL_INTERVAL_MS : false }, staleTime: 0, }) @@ -41,14 +60,7 @@ export function useProcessingDocuments() { // Query's structural sharing) so `processingMap` only changes identity // when the poll payload actually changes — the effect below depends on it. const docs = useMemo( - () => - ( - data as - | { - documents?: Array<{ id?: string | null; status?: string | null }> - } - | undefined - )?.documents ?? [], + () => (data as ProcessingDocumentsData | undefined)?.documents ?? [], [data], )