From 21b40358c23cd14a7e843330cc3e00507763c2ec Mon Sep 17 00:00:00 2001 From: digua Date: Mon, 20 Jul 2026 23:58:51 +0800 Subject: [PATCH] fix(web): keep member filters out of proximity queries Preserve time bounds for group proximity analysis without pre-filtering the message stream to a single sender. --- .../session/SessionInsights.test.ts | 10 ++++ .../components/session/SessionInsights.vue | 3 + .../insights/GroupRelationshipInsights.vue | 4 +- .../group-relationship-filter.test.ts | 17 ++++++ .../insights/group-relationship-filter.ts | 11 ++++ src/services/data/browser.test.ts | 56 +++++++++++++++++++ src/services/data/browser.ts | 52 ++++++++++------- src/services/utils/http.ts | 28 +++++++--- 8 files changed, 152 insertions(+), 29 deletions(-) create mode 100644 apps/web-wasm/src/components/session/SessionInsights.test.ts create mode 100644 apps/web-wasm/src/components/session/insights/group-relationship-filter.test.ts create mode 100644 apps/web-wasm/src/components/session/insights/group-relationship-filter.ts diff --git a/apps/web-wasm/src/components/session/SessionInsights.test.ts b/apps/web-wasm/src/components/session/SessionInsights.test.ts new file mode 100644 index 00000000..0370e772 --- /dev/null +++ b/apps/web-wasm/src/components/session/SessionInsights.test.ts @@ -0,0 +1,10 @@ +import assert from 'node:assert/strict' +import { readFileSync } from 'node:fs' +import test from 'node:test' + +const source = readFileSync(new URL('./SessionInsights.vue', import.meta.url), 'utf8') + +test('keeps member filter invalidation scoped to the active insight subtree', () => { + assert.equal(source.includes('abortAnalyticsRequests'), false) + assert.equal(source.match(/:key="selectedMemberId \?\? 'all'"/g)?.length, 3) +}) diff --git a/apps/web-wasm/src/components/session/SessionInsights.vue b/apps/web-wasm/src/components/session/SessionInsights.vue index 7ca90d61..276946a2 100644 --- a/apps/web-wasm/src/components/session/SessionInsights.vue +++ b/apps/web-wasm/src/components/session/SessionInsights.vue @@ -74,12 +74,14 @@ const viewTimeFilter = computed(() => ({ ...props.timeFilter, memberId: selected /> ({ ...props.timeFilter, memberId: selected /> diff --git a/apps/web-wasm/src/components/session/insights/GroupRelationshipInsights.vue b/apps/web-wasm/src/components/session/insights/GroupRelationshipInsights.vue index 2f8430b2..fd771950 100644 --- a/apps/web-wasm/src/components/session/insights/GroupRelationshipInsights.vue +++ b/apps/web-wasm/src/components/session/insights/GroupRelationshipInsights.vue @@ -6,6 +6,7 @@ import ClusterView from '@/components/analysis/relationships/ClusterView.vue' import InteractionView from '@/components/analysis/relationships/InteractionView.vue' import MentionRankingView from '@/components/analysis/relationships/MentionRankingView.vue' import type { TimeFilter } from '@openchatlab/shared-types' +import { getProximityTimeFilter } from './group-relationship-filter' const props = defineProps<{ sessionId: string @@ -14,6 +15,7 @@ const props = defineProps<{ const { t } = useI18n() const activeSubTab = ref('mention-graph') +const proximityTimeFilter = computed(() => getProximityTimeFilter(props.timeFilter)) const subTabs = computed(() => [ { id: 'mention-graph', @@ -52,7 +54,7 @@ const subTabs = computed(() => [ diff --git a/apps/web-wasm/src/components/session/insights/group-relationship-filter.test.ts b/apps/web-wasm/src/components/session/insights/group-relationship-filter.test.ts new file mode 100644 index 00000000..eb99fefe --- /dev/null +++ b/apps/web-wasm/src/components/session/insights/group-relationship-filter.test.ts @@ -0,0 +1,17 @@ +import assert from 'node:assert/strict' +import { describe, it } from 'node:test' +import { getProximityTimeFilter } from './group-relationship-filter' + +describe('getProximityTimeFilter', () => { + it('keeps time bounds but removes the member filter', () => { + assert.deepEqual(getProximityTimeFilter({ startTs: 10, endTs: 20, memberId: 3 }), { + startTs: 10, + endTs: 20, + }) + }) + + it('returns no filter when only a member is selected', () => { + assert.equal(getProximityTimeFilter({ memberId: 3 }), undefined) + assert.equal(getProximityTimeFilter(), undefined) + }) +}) diff --git a/apps/web-wasm/src/components/session/insights/group-relationship-filter.ts b/apps/web-wasm/src/components/session/insights/group-relationship-filter.ts new file mode 100644 index 00000000..70ccb998 --- /dev/null +++ b/apps/web-wasm/src/components/session/insights/group-relationship-filter.ts @@ -0,0 +1,11 @@ +import type { TimeFilter } from '@openchatlab/shared-types' + +export function getProximityTimeFilter(filter?: TimeFilter): TimeFilter | undefined { + if (!filter) return undefined + + const timeFilter: TimeFilter = {} + if (filter.startTs !== undefined) timeFilter.startTs = filter.startTs + if (filter.endTs !== undefined) timeFilter.endTs = filter.endTs + + return timeFilter.startTs === undefined && timeFilter.endTs === undefined ? undefined : timeFilter +} diff --git a/src/services/data/browser.test.ts b/src/services/data/browser.test.ts index 42ff77ed..0c8ad391 100644 --- a/src/services/data/browser.test.ts +++ b/src/services/data/browser.test.ts @@ -6,8 +6,22 @@ import type { WebRuntimeTaskResult, WebRuntimeTaskType, } from '@openchatlab/web-runtime' +import { abortAnalyticsRequests } from '../utils/http' import { createBrowserDataAdapter } from './browser' +interface Deferred { + promise: Promise + resolve: (value: T) => void +} + +function createDeferred(): Deferred { + let resolve!: (value: T) => void + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise + }) + return { promise, resolve } +} + describe('BrowserDataAdapter', () => { it('maps catalog sessions and forwards session mutations through RPC', async () => { const requests: Array<{ type: WebRuntimeTaskType; payload: unknown }> = [] @@ -232,6 +246,48 @@ describe('BrowserDataAdapter', () => { ) }) + it('ignores stale Worker analysis results after the analytics epoch changes', async () => { + const loads = [createDeferred(), createDeferred()] + const requestOptions: Array = [] + let loadIndex = 0 + const rpc = { + request( + type: T, + _payload: WebRuntimeTaskPayload, + options?: RpcRequestOptions + ): Promise> { + assert.equal(type, 'analysis.hourly') + requestOptions.push(options) + return loads[loadIndex++]!.promise as Promise> + }, + dispose: () => undefined, + } + const adapter = createBrowserDataAdapter(rpc) + + let staleSettled = false + void adapter.getHourlyActivity('session-one', { startTs: 1 }).then( + () => { + staleSettled = true + }, + () => { + staleSettled = true + } + ) + + abortAnalyticsRequests() + const latest = adapter.getHourlyActivity('session-one', { startTs: 2 }) + + assert.equal(requestOptions[0]?.signal?.aborted, true) + assert.equal(requestOptions[1]?.signal?.aborted, false) + + loads[0]!.resolve([{ hour: 1, messageCount: 1 }]) + await new Promise((resolve) => setImmediate(resolve)) + assert.equal(staleSettled, false) + + loads[1]!.resolve([{ hour: 2, messageCount: 2 }]) + assert.deepEqual(await latest, [{ hour: 2, messageCount: 2 }]) + }) + it('rejects unsupported DataAdapter capabilities instead of returning fake data', async () => { const rpc = { request: () => Promise.reject(new Error('not used')), diff --git a/src/services/data/browser.ts b/src/services/data/browser.ts index af38a39c..d5457ba2 100644 --- a/src/services/data/browser.ts +++ b/src/services/data/browser.ts @@ -1,4 +1,10 @@ -import { sessionDatabaseFilename, type BrowserSessionCatalogItem } from '@openchatlab/web-runtime' +import { + sessionDatabaseFilename, + type BrowserSessionCatalogItem, + type WebRuntimeTaskPayload, + type WebRuntimeTaskResult, + type WebRuntimeTaskType, +} from '@openchatlab/web-runtime' import type { AnalysisSession, MessageType } from '@/types/base' import type { ClusterGraphData, @@ -16,6 +22,7 @@ import type { LanguagePreferenceResult } from '@/types/quotes/languagePreference import type { MemberMonthlyTrend, WordFrequencyParams, WordFrequencyResult } from '@openchatlab/core' import type { TimeFilter } from '@openchatlab/shared-types' import type { BrowserRuntimeRpcPort } from '../browser-runtime/types' +import { withAnalyticsRequestEpoch } from '../utils/http' import type { DataAdapter } from './types' type BrowserSessionDataAdapter = Pick< @@ -50,6 +57,13 @@ type BrowserSessionDataAdapter = Pick< export class BrowserDataAdapter implements BrowserSessionDataAdapter { constructor(private readonly rpc: BrowserRuntimeRpcPort) {} + private requestAnalysis( + type: T, + payload: WebRuntimeTaskPayload + ): Promise> { + return withAnalyticsRequestEpoch((signal) => this.rpc.request(type, payload, { signal })) + } + async getSessions(): Promise { return (await this.rpc.request('session.list', undefined)).map(mapSession) } @@ -68,15 +82,15 @@ export class BrowserDataAdapter implements BrowserSessionDataAdapter { } getHourlyActivity(sessionId: string, filter?: TimeFilter): Promise { - return this.rpc.request('analysis.hourly', { sessionId, filter }) + return this.requestAnalysis('analysis.hourly', { sessionId, filter }) } getDailyActivity(sessionId: string, filter?: TimeFilter): Promise { - return this.rpc.request('analysis.daily', { sessionId, filter }) + return this.requestAnalysis('analysis.daily', { sessionId, filter }) } getWeekdayActivity(sessionId: string, filter?: TimeFilter): Promise { - return this.rpc.request('analysis.weekday', { sessionId, filter }) + return this.requestAnalysis('analysis.weekday', { sessionId, filter }) } getTimeRange(sessionId: string): Promise<{ start: number; end: number } | null> { @@ -88,45 +102,45 @@ export class BrowserDataAdapter implements BrowserSessionDataAdapter { } getMemberActivity(sessionId: string, filter?: TimeFilter): Promise { - return this.rpc.request('analysis.members', { sessionId, filter }) + return this.requestAnalysis('analysis.members', { sessionId, filter }) } async getMessageTypeDistribution( sessionId: string, filter?: TimeFilter ): Promise> { - return (await this.rpc.request('analysis.messageTypes', { sessionId, filter })).map((item) => ({ + return (await this.requestAnalysis('analysis.messageTypes', { sessionId, filter })).map((item) => ({ type: item.type as MessageType, count: item.count, })) } getMessageLengthDistribution(sessionId: string, filter?: TimeFilter) { - return this.rpc.request('analysis.messageLengths', { sessionId, filter }) + return this.requestAnalysis('analysis.messageLengths', { sessionId, filter }) } getTextStats(sessionId: string, filter?: TimeFilter) { - return this.rpc.request('analysis.textStats', { sessionId, filter }) + return this.requestAnalysis('analysis.textStats', { sessionId, filter }) } getLongMessageCount(sessionId: string, filter?: TimeFilter, minLength?: number) { - return this.rpc.request('analysis.longMessages', { sessionId, filter, minLength }) + return this.requestAnalysis('analysis.longMessages', { sessionId, filter, minLength }) } getTextLengthPercentiles(sessionId: string, filter?: TimeFilter) { - return this.rpc.request('analysis.textPercentiles', { sessionId, filter }) + return this.requestAnalysis('analysis.textPercentiles', { sessionId, filter }) } getMonthlyActivity(sessionId: string, filter?: TimeFilter): Promise { - return this.rpc.request('analysis.monthly', { sessionId, filter }) + return this.requestAnalysis('analysis.monthly', { sessionId, filter }) } getYearlyActivity(sessionId: string, filter?: TimeFilter): Promise> { - return this.rpc.request('analysis.yearly', { sessionId, filter }) + return this.requestAnalysis('analysis.yearly', { sessionId, filter }) } getMemberMonthlyTrend(sessionId: string, filter?: TimeFilter): Promise { - return this.rpc.request('analysis.memberMonthlyTrend', { sessionId, filter }) + return this.requestAnalysis('analysis.memberMonthlyTrend', { sessionId, filter }) } getMembers(sessionId: string): Promise { @@ -134,15 +148,15 @@ export class BrowserDataAdapter implements BrowserSessionDataAdapter { } getMentionAnalysis(sessionId: string, filter?: TimeFilter): Promise { - return this.rpc.request('analysis.mentions', { sessionId, filter }) + return this.requestAnalysis('analysis.mentions', { sessionId, filter }) } getMentionGraph(sessionId: string, filter?: TimeFilter) { - return this.rpc.request('analysis.mentionGraph', { sessionId, filter }) + return this.requestAnalysis('analysis.mentionGraph', { sessionId, filter }) } getClusterGraph(sessionId: string, filter?: TimeFilter, options?: ClusterGraphOptions): Promise { - return this.rpc.request('analysis.clusterGraph', { sessionId, filter, options }) + return this.requestAnalysis('analysis.clusterGraph', { sessionId, filter, options }) } getRelationshipStats( @@ -150,7 +164,7 @@ export class BrowserDataAdapter implements BrowserSessionDataAdapter { filter?: TimeFilter, options?: { perseveranceThreshold?: number } ): Promise { - return this.rpc.request('analysis.relationship', { sessionId, filter, options }) + return this.requestAnalysis('analysis.relationship', { sessionId, filter, options }) } async getLanguagePreferenceAnalysis( @@ -158,7 +172,7 @@ export class BrowserDataAdapter implements BrowserSessionDataAdapter { locale: string, filter?: TimeFilter ): Promise { - return (await this.rpc.request('analysis.languagePreference', { + return (await this.requestAnalysis('analysis.languagePreference', { sessionId, locale, filter, @@ -166,7 +180,7 @@ export class BrowserDataAdapter implements BrowserSessionDataAdapter { } getWordFrequency(sessionId: string, params: Omit): Promise { - return this.rpc.request('analysis.wordFrequency', { sessionId, params }) + return this.requestAnalysis('analysis.wordFrequency', { sessionId, params }) } } diff --git a/src/services/utils/http.ts b/src/services/utils/http.ts index 8f064fb1..cd84264f 100644 --- a/src/services/utils/http.ts +++ b/src/services/utils/http.ts @@ -120,7 +120,7 @@ export function abortAnalyticsRequests(): void { } function isAbortError(err: unknown): boolean { - return err instanceof DOMException && err.name === 'AbortError' + return err instanceof Error && err.name === 'AbortError' } // 已作废的请求永不 settle：避免旧数据覆盖新数据，也避免在调用方触发无意义的错误处理与空态闪烁。 @@ -128,20 +128,30 @@ function neverSettle(): Promise { return new Promise(() => {}) } +/** + * Bind an analysis request to the current epoch. + * + * Besides aborting queued work, the identity check also discards a result when + * the underlying runtime cannot interrupt a task that has already started. + */ +export function withAnalyticsRequestEpoch(request: (signal: AbortSignal) => Promise): Promise { + const controller = _analyticsController + return request(controller.signal) + .then((result) => (controller === _analyticsController ? result : neverSettle())) + .catch((err) => { + if (controller !== _analyticsController || isAbortError(err)) return neverSettle() + throw err + }) +} + /** 绑定当前 epoch 的 GET：被 abortAnalyticsRequests() 取消时静默作废。 */ export function analyticsGet(path: string): Promise { - return get(path, _analyticsController.signal).catch((err) => { - if (isAbortError(err)) return neverSettle() - throw err - }) + return withAnalyticsRequestEpoch((signal) => get(path, signal)) } /** 绑定当前 epoch 的 POST：被 abortAnalyticsRequests() 取消时静默作废。 */ export function analyticsPost(path: string, body?: unknown): Promise { - return post(path, body, _analyticsController.signal).catch((err) => { - if (isAbortError(err)) return neverSettle() - throw err - }) + return withAnalyticsRequestEpoch((signal) => post(path, body, signal)) } export async function del(path: string): Promise {