diff --git a/apps/cli/tsup.config.ts b/apps/cli/tsup.config.ts index 501de58b..e104a9a9 100644 --- a/apps/cli/tsup.config.ts +++ b/apps/cli/tsup.config.ts @@ -7,6 +7,7 @@ export default defineConfig({ 'semantic-index-worker': '../../packages/node-runtime/src/semantic-index/worker-thread-entry.ts', 'contacts-worker': '../../packages/node-runtime/src/services/contacts/worker-entry.ts', 'people-relationships-worker': '../../packages/node-runtime/src/services/people/relationships/worker-entry.ts', + 'global-insight-worker': '../../packages/node-runtime/src/services/global-insight/worker-entry.ts', }, format: ['esm'], outDir: 'dist', diff --git a/apps/desktop/electron.vite.config.ts b/apps/desktop/electron.vite.config.ts index b151b716..73d9307b 100644 --- a/apps/desktop/electron.vite.config.ts +++ b/apps/desktop/electron.vite.config.ts @@ -34,6 +34,10 @@ export default defineConfig({ rootDir, 'packages/node-runtime/src/services/people/relationships/worker-entry.ts' ), + 'global-insight-worker': resolve( + rootDir, + 'packages/node-runtime/src/services/global-insight/worker-entry.ts' + ), }, }, }, diff --git a/apps/desktop/main/database/core.test.ts b/apps/desktop/main/database/core.test.ts index 7914d4a8..506b0971 100644 --- a/apps/desktop/main/database/core.test.ts +++ b/apps/desktop/main/database/core.test.ts @@ -44,6 +44,9 @@ test('openDatabase keeps readonly validation read-only for DELETE journal databa contactsService: { getContactsFactsCacheDir: () => path.join(root, 'cache', 'contacts'), }, + globalInsightService: { + getGlobalInsightFactsCacheDir: () => path.join(root, 'cache', 'global-insight'), + }, peopleRelationshipsService: { getPeopleRelationshipsFactsCacheDir: () => path.join(root, 'cache', 'people-relationships'), }, diff --git a/apps/desktop/main/database/core.ts b/apps/desktop/main/database/core.ts index de310f79..a120a6ff 100644 --- a/apps/desktop/main/database/core.ts +++ b/apps/desktop/main/database/core.ts @@ -16,6 +16,7 @@ import { BetterSqliteAdapter, writeParseResultToDb, contactsService, + globalInsightService, peopleRelationshipsService, } from '@openchatlab/node-runtime' import type { RuntimeIdentity } from '@openchatlab/node-runtime/src/data-dir-compat' @@ -171,6 +172,10 @@ export function deleteSession(sessionId: string): boolean { deleteSessionCache(sessionId, cacheDir) deleteSessionCache(sessionId, path.join(cacheDir, 'query')) deleteSessionCache(sessionId, contactsService.getContactsFactsCacheDir(getPathProvider().getUserDataDir())) + deleteSessionCache( + sessionId, + globalInsightService.getGlobalInsightFactsCacheDir(getPathProvider().getUserDataDir()) + ) deleteSessionCache( sessionId, peopleRelationshipsService.getPeopleRelationshipsFactsCacheDir(getPathProvider().getUserDataDir()) diff --git a/packages/node-runtime/src/index.ts b/packages/node-runtime/src/index.ts index b3340388..b1397c49 100644 --- a/packages/node-runtime/src/index.ts +++ b/packages/node-runtime/src/index.ts @@ -376,6 +376,7 @@ export * as exportService from './services/export-service' export * as ownerProfileService from './services/owner-profile-service' export * as contactsService from './services/contacts' export * as peopleRelationshipsService from './services/people/relationships' +export * as globalInsightService from './services/global-insight' // Semantic index (Phase 1 vector search) — independent of legacy ai/rag export * as semanticIndex from './semantic-index' export { @@ -407,9 +408,11 @@ export type { export { CONTACTS_ALGORITHM_VERSION, PEOPLE_RELATIONSHIPS_ALGORITHM_VERSION, + ANNUAL_SUMMARY_ALGORITHM_VERSION, MergeSessionCache, createContactsService, createPeopleRelationshipsService, + createGlobalInsightService, createDatabaseManagerAdapter, pushImport, } from './services' @@ -429,6 +432,10 @@ export type { PeopleRelationshipsService, PeopleRelationshipsServiceDeps, PeopleRelationshipsServiceOptions, + AnnualSummaryComputeRunner, + GlobalInsightService, + GlobalInsightServiceDeps, + GlobalInsightServiceOptions, PushImportPayload, PushImportResult, PushImportOutcome, diff --git a/packages/node-runtime/src/services/global-insight/index.ts b/packages/node-runtime/src/services/global-insight/index.ts new file mode 100644 index 00000000..323b24f4 --- /dev/null +++ b/packages/node-runtime/src/services/global-insight/index.ts @@ -0,0 +1,5 @@ +export { createGlobalInsightService } from './service' +export type { GlobalInsightService, GlobalInsightServiceDeps, GlobalInsightServiceOptions } from './service' +export { ANNUAL_SUMMARY_ALGORITHM_VERSION } from './types' +export type { AnnualSummaryComputeRunner, AnnualSummaryRunnerOptions } from './worker-runner' +export { getGlobalInsightDir, getGlobalInsightFactsCacheDir } from './paths' diff --git a/packages/node-runtime/src/services/global-insight/service.test.ts b/packages/node-runtime/src/services/global-insight/service.test.ts new file mode 100644 index 00000000..1b12ee84 --- /dev/null +++ b/packages/node-runtime/src/services/global-insight/service.test.ts @@ -0,0 +1,166 @@ +import assert from 'node:assert/strict' +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import test from 'node:test' +import type { AnnualSummaryRange } from '@openchatlab/shared-types' +import type { SessionRuntimeAdapter } from '../adapters' +import { createGlobalInsightService } from './service' +import type { AnnualSummarySnapshot } from './types' +import type { AnnualSummaryComputeRunner } from './worker-runner' + +function makeSnapshot(range: AnnualSummaryRange, signature: string, count = 3): AnnualSummarySnapshot { + return { + algorithmVersion: 'annual-summary-v1', + signature, + computedAt: 100, + range, + availableDataYears: [2026], + latestDataYear: 2026, + metrics: { + sentMessageCount: count, + activeDayCount: 1, + directContactCount: 1, + averageMessagesPerDay: count, + averageDirectContactsPerDay: 1, + }, + monthlyActivity: [], + dailyActivity: [], + messageTypes: [], + textLength: { textMessageCount: 0, median: null, p90: null, buckets: [] }, + coverage: { + totalSessions: 1, + analyzedSessions: 1, + missingOwnerSessions: 0, + unresolvedOwnerSessions: 0, + failedSessions: 0, + }, + workerStats: { durationMs: 1, totalSessions: 1, processedSessions: 1, cacheHits: 0, cacheMisses: 1 }, + } +} + +function createEnv(runner: AnnualSummaryComputeRunner) { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'chatlab-global-insight-service-')) + const dbPath = path.join(dir, 'chat.db') + fs.writeFileSync(dbPath, 'db') + const adapter = { + listSessionIds: () => ['chat-1'], + getDbPath: () => dbPath, + } as unknown as SessionRuntimeAdapter + const service = createGlobalInsightService({ adapter, userDataDir: dir, runner, now: () => Date.UTC(2026, 0, 2) }) + return { dir, dbPath, adapter, service } +} + +async function flushTasks(): Promise { + await new Promise((resolve) => setTimeout(resolve, 0)) +} + +test('starts one background task on a missing snapshot and then serves it fresh', async (t) => { + let calls = 0 + const env = createEnv(async ({ signature, range }) => { + calls++ + return makeSnapshot(range, signature) + }) + t.after(async () => { + await env.service.close() + fs.rmSync(env.dir, { recursive: true, force: true }) + }) + + const first = env.service.getAnnualSummary({ mode: 'year', year: 2026 }) + const duplicate = env.service.getAnnualSummary({ mode: 'year', year: 2026 }) + assert.equal(first.cache.status, 'missing') + assert.equal(first.metrics, null) + assert.equal(duplicate.task.status, 'running') + assert.equal(calls, 1) + + await flushTasks() + const fresh = env.service.getAnnualSummary({ mode: 'year', year: 2026 }) + assert.equal(fresh.cache.status, 'fresh') + assert.equal(fresh.metrics?.sentMessageCount, 3) +}) + +test('serves stale data while recomputing after the DB signature changes', async (t) => { + let resolveRunner: ((snapshot: AnnualSummarySnapshot) => void) | undefined + const runner: AnnualSummaryComputeRunner = ({ signature, range }) => + new Promise((resolve) => { + resolveRunner = () => resolve(makeSnapshot(range, signature, 9)) + }) + const env = createEnv(runner) + t.after(async () => { + await env.service.close() + fs.rmSync(env.dir, { recursive: true, force: true }) + }) + const range = env.service.normalizeRange({ mode: 'year', year: 2026 }) + env.service.replaceSnapshotForTests(makeSnapshot(range, 'old-signature')) + + const stale = env.service.getAnnualSummary({ mode: 'year', year: 2026, acceptStale: true }) + assert.equal(stale.cache.status, 'stale') + assert.equal(stale.metrics?.sentMessageCount, 3) + assert.equal(stale.task.status, 'running') + + resolveRunner?.(makeSnapshot(range, stale.cache.signature ?? '', 9)) +}) + +test('preserves failed state until explicit recompute', async (t) => { + let calls = 0 + const env = createEnv(async ({ signature, range }) => { + calls++ + if (calls === 1) throw new Error('worker failed') + return makeSnapshot(range, signature) + }) + t.after(async () => { + await env.service.close() + fs.rmSync(env.dir, { recursive: true, force: true }) + }) + + env.service.getAnnualSummary({ mode: 'year', year: 2026 }) + await flushTasks() + assert.equal(env.service.getAnnualSummary({ mode: 'year', year: 2026 }).task.status, 'failed') + assert.equal(calls, 1) + + env.service.startRecompute({ mode: 'year', year: 2026 }) + await flushTasks() + assert.equal(calls, 2) + assert.equal(env.service.getAnnualSummary({ mode: 'year', year: 2026 }).cache.status, 'fresh') +}) + +test('discards a worker result when the DB signature changes during compute', async (t) => { + let finish: (() => void) | undefined + const env = createEnv( + ({ signature, range }) => + new Promise((resolve) => { + finish = () => resolve(makeSnapshot(range, signature)) + }) + ) + t.after(async () => { + await env.service.close() + fs.rmSync(env.dir, { recursive: true, force: true }) + }) + env.service.getAnnualSummary({ mode: 'year', year: 2026 }) + fs.appendFileSync(env.dbPath, 'changed') + finish?.() + await flushTasks() + + const response = env.service.getAnnualSummary({ mode: 'year', year: 2026 }) + assert.equal(response.cache.status, 'missing') + assert.equal(response.task.status, 'running') +}) + +test('close aborts an in-flight task', async (t) => { + let aborted = false + const env = createEnv( + ({ signal }) => + new Promise((_resolve, reject) => { + signal.addEventListener('abort', () => { + aborted = true + reject(new Error('aborted')) + }) + }) + ) + t.after(() => fs.rmSync(env.dir, { recursive: true, force: true })) + env.service.getAnnualSummary({ mode: 'year', year: 2026 }) + + await env.service.close() + + assert.equal(aborted, true) +}) diff --git a/packages/node-runtime/src/services/global-insight/service.ts b/packages/node-runtime/src/services/global-insight/service.ts new file mode 100644 index 00000000..039653f2 --- /dev/null +++ b/packages/node-runtime/src/services/global-insight/service.ts @@ -0,0 +1,262 @@ +import type { PathProvider } from '@openchatlab/core' +import type { AnnualSummaryRange, AnnualSummaryResponse, AnnualSummaryTaskState } from '@openchatlab/shared-types' +import type { RuntimeIdentity } from '../../data-dir-compat' +import { appLogger } from '../../logging/app-logger' +import type { SessionRuntimeAdapter } from '../adapters' +import { getGlobalInsightDir } from './paths' +import { buildAnnualSummarySignature } from './signature' +import { + cleanupAnnualSummarySnapshotTempFiles, + readAnnualSummarySnapshot, + writeAnnualSummarySnapshot, +} from './snapshot' +import { normalizeAnnualSummaryRange, toAnnualSummaryRangeKey, type AnnualSummaryRangeInput } from './time-range' +import { createAnnualSummaryWorkerRunner, type AnnualSummaryComputeRunner } from './worker-runner' +import type { AnnualSummarySnapshot } from './types' + +export interface GlobalInsightServiceOptions extends AnnualSummaryRangeInput { + acceptStale?: boolean + forceRecompute?: boolean +} + +export interface GlobalInsightServiceDeps { + adapter: SessionRuntimeAdapter + pathProvider?: PathProvider + userDataDir?: string + runtimeIdentity?: RuntimeIdentity + nativeBinding?: string + workerEntryUrl?: string | URL + runner?: AnnualSummaryComputeRunner + now?: () => number +} + +export interface GlobalInsightService { + getAnnualSummary(options?: GlobalInsightServiceOptions): AnnualSummaryResponse + startRecompute(options?: GlobalInsightServiceOptions): AnnualSummaryResponse + invalidateCache(): void + close(): Promise + normalizeRange(input?: AnnualSummaryRangeInput): AnnualSummaryRange + replaceSnapshotForTests(snapshot: AnnualSummarySnapshot): void +} + +interface InFlightTask { + id: string + signature: string + promise: Promise + abortController: AbortController +} + +export function createGlobalInsightService(deps: GlobalInsightServiceDeps): GlobalInsightService { + return new DefaultGlobalInsightService(deps) +} + +class DefaultGlobalInsightService implements GlobalInsightService { + private readonly snapshots = new Map() + private readonly snapshotDir: string + private readonly runner: AnnualSummaryComputeRunner + private inFlight: InFlightTask | null = null + private task: AnnualSummaryTaskState = createIdleTaskState() + + constructor(private readonly deps: GlobalInsightServiceDeps) { + this.snapshotDir = deps.userDataDir + ? getGlobalInsightDir(deps.userDataDir) + : getGlobalInsightDir(requirePathProvider(deps).getUserDataDir()) + cleanupAnnualSummarySnapshotTempFiles(this.snapshotDir) + this.runner = + deps.runner ?? + createAnnualSummaryWorkerRunner({ + pathProvider: requirePathProvider(deps), + runtimeIdentity: deps.runtimeIdentity, + nativeBinding: deps.nativeBinding, + workerEntryUrl: deps.workerEntryUrl, + }) + } + + getAnnualSummary(options: GlobalInsightServiceOptions = {}): AnnualSummaryResponse { + const range = this.normalizeRange(options) + const signature = buildAnnualSummarySignature(this.deps.adapter, range) + const status = this.getCacheStatus(signature, range) + if (this.shouldStartTask(options, status)) this.ensureTaskStarted(signature, range) + return this.toResponse(signature, range, options.acceptStale === true) + } + + startRecompute(options: GlobalInsightServiceOptions = {}): AnnualSummaryResponse { + const range = this.normalizeRange(options) + const signature = buildAnnualSummarySignature(this.deps.adapter, range) + this.ensureTaskStarted(signature, range) + return this.toResponse(signature, range, true) + } + + invalidateCache(): void { + this.snapshots.clear() + } + + normalizeRange(input: AnnualSummaryRangeInput = {}): AnnualSummaryRange { + return normalizeAnnualSummaryRange(input, new Date(this.now())) + } + + replaceSnapshotForTests(snapshot: AnnualSummarySnapshot): void { + this.snapshots.set(toAnnualSummaryRangeKey(snapshot.range), snapshot) + } + + async close(): Promise { + const inFlight = this.inFlight + if (!inFlight) return + this.inFlight = null + inFlight.abortController.abort() + this.task = { + ...this.task, + status: 'failed', + finishedAt: this.now(), + lastError: 'annual summary task aborted', + } + } + + private shouldStartTask( + options: GlobalInsightServiceOptions, + status: AnnualSummaryResponse['cache']['status'] + ): boolean { + if (options.forceRecompute) return true + if (status === 'fresh') return false + return this.task.status !== 'failed' + } + + private ensureTaskStarted(signature: string, range: AnnualSummaryRange): void { + if (this.inFlight) return + const id = `annual_summary_${this.now()}_${Math.random().toString(36).slice(2)}` + this.task = { + id, + status: 'running', + startedAt: this.now(), + finishedAt: null, + processedSessions: 0, + totalSessions: this.deps.adapter.listSessionIds().length, + } + const abortController = new AbortController() + const promise = this.runner({ + signature, + range, + signal: abortController.signal, + onProgress: (progress) => { + if (this.task.id !== id || this.task.status !== 'running') return + this.task = { ...this.task, ...progress } + }, + }) + this.inFlight = { id, signature, promise, abortController } + promise + .then((snapshot) => this.handleTaskSuccess(id, signature, snapshot)) + .catch((error) => this.handleTaskFailure(id, error)) + } + + private handleTaskSuccess(id: string, inputSignature: string, snapshot: AnnualSummarySnapshot): void { + if (this.inFlight?.id !== id) return + this.inFlight = null + const latestSignature = buildAnnualSummarySignature(this.deps.adapter, snapshot.range) + const finishedAt = this.now() + if (inputSignature !== latestSignature || snapshot.signature !== latestSignature) { + this.task = { ...this.task, status: 'superseded', finishedAt, currentSessionId: undefined } + appLogger.info('global-insight', 'annual summary worker result discarded because signature changed') + return + } + try { + writeAnnualSummarySnapshot(this.snapshotDir, snapshot) + this.snapshots.set(toAnnualSummaryRangeKey(snapshot.range), snapshot) + this.task = { + ...this.task, + status: 'succeeded', + finishedAt, + processedSessions: snapshot.workerStats.processedSessions, + totalSessions: snapshot.workerStats.totalSessions, + currentSessionId: undefined, + } + appLogger.info('global-insight', 'annual summary snapshot persisted', { + analyzedSessions: snapshot.coverage.analyzedSessions, + durationMs: snapshot.workerStats.durationMs, + }) + } catch (error) { + this.handleTaskFailure(id, error) + } + } + + private handleTaskFailure(id: string, error: unknown): void { + if (this.inFlight?.id !== id) return + this.inFlight = null + this.task = { + ...this.task, + status: 'failed', + finishedAt: this.now(), + currentSessionId: undefined, + lastError: error instanceof Error ? error.message : String(error), + } + appLogger.error('global-insight', 'annual summary worker failed', error) + } + + private toResponse(signature: string, range: AnnualSummaryRange, acceptStale: boolean): AnnualSummaryResponse { + const snapshot = this.getSnapshot(range) + const status = this.getCacheStatus(signature, range) + const includeSnapshot = status === 'fresh' || (status === 'stale' && acceptStale) + return { + range, + availableDataYears: includeSnapshot ? (snapshot?.availableDataYears ?? []) : [], + latestDataYear: includeSnapshot ? (snapshot?.latestDataYear ?? null) : null, + metrics: includeSnapshot ? (snapshot?.metrics ?? null) : null, + monthlyActivity: includeSnapshot ? (snapshot?.monthlyActivity ?? []) : [], + dailyActivity: includeSnapshot ? (snapshot?.dailyActivity ?? []) : [], + messageTypes: includeSnapshot ? (snapshot?.messageTypes ?? []) : [], + textLength: includeSnapshot ? (snapshot?.textLength ?? null) : null, + coverage: includeSnapshot ? (snapshot?.coverage ?? emptyCoverage()) : emptyCoverage(), + cache: { + status, + computedAt: snapshot?.computedAt ?? null, + signature: snapshot?.signature, + staleReason: status === 'stale' ? 'signature_changed' : undefined, + }, + task: this.task, + } + } + + private getCacheStatus(signature: string, range: AnnualSummaryRange): AnnualSummaryResponse['cache']['status'] { + const snapshot = this.getSnapshot(range) + if (!snapshot) return 'missing' + return snapshot.signature === signature ? 'fresh' : 'stale' + } + + private getSnapshot(range: AnnualSummaryRange): AnnualSummarySnapshot | null { + const key = toAnnualSummaryRangeKey(range) + if (!this.snapshots.has(key)) { + this.snapshots.set(key, readAnnualSummarySnapshot(this.snapshotDir, range, { now: this.deps.now })) + } + return this.snapshots.get(key) ?? null + } + + private now(): number { + return this.deps.now?.() ?? Date.now() + } +} + +function requirePathProvider(deps: GlobalInsightServiceDeps): PathProvider { + if (!deps.pathProvider) + throw new Error('GlobalInsightService requires pathProvider unless userDataDir and runner are provided') + return deps.pathProvider +} + +function createIdleTaskState(): AnnualSummaryTaskState { + return { + id: null, + status: 'idle', + startedAt: null, + finishedAt: null, + processedSessions: 0, + totalSessions: 0, + } +} + +function emptyCoverage(): AnnualSummaryResponse['coverage'] { + return { + totalSessions: 0, + analyzedSessions: 0, + missingOwnerSessions: 0, + unresolvedOwnerSessions: 0, + failedSessions: 0, + } +} diff --git a/packages/node-runtime/src/services/global-insight/worker-entry.ts b/packages/node-runtime/src/services/global-insight/worker-entry.ts new file mode 100644 index 00000000..1325788d --- /dev/null +++ b/packages/node-runtime/src/services/global-insight/worker-entry.ts @@ -0,0 +1,43 @@ +import { parentPort, workerData } from 'node:worker_threads' +import type { AnnualSummaryRange } from '@openchatlab/shared-types' +import { DatabaseManager } from '../../database-manager' +import type { RuntimeIdentity } from '../../data-dir-compat' +import { initAppLogger } from '../../logging/app-logger' +import { StaticPathProvider, type StaticPathProviderSnapshot } from '../../semantic-index/static-path-provider' +import { createDatabaseManagerAdapter } from '../adapters' +import { computeAnnualSummarySnapshot, type ComputeAnnualSummarySnapshotOptions } from './compute' +import { getGlobalInsightFactsCacheDir } from './paths' + +interface StartupOptions { + paths: StaticPathProviderSnapshot + runtimeIdentity?: RuntimeIdentity + nativeBinding?: string + signature: string + range: AnnualSummaryRange +} + +async function main(): Promise { + if (!parentPort) throw new Error('annual summary worker requires parentPort') + const options = workerData as StartupOptions + initAppLogger(options.paths.logsDir) + const pathProvider = new StaticPathProvider(options.paths) + const dbManager = new DatabaseManager(pathProvider, { + nativeBinding: options.nativeBinding, + runtime: options.runtimeIdentity, + allowMissingRuntimeForTests: !options.runtimeIdentity, + }) + const adapter = createDatabaseManagerAdapter(dbManager) + const computeOptions: ComputeAnnualSummarySnapshotOptions = { + adapter, + signature: options.signature, + range: options.range, + factsCacheDir: getGlobalInsightFactsCacheDir(pathProvider.getUserDataDir()), + onProgress: (progress) => parentPort?.postMessage({ type: 'progress', progress }), + } + const snapshot = computeAnnualSummarySnapshot(computeOptions) + parentPort.postMessage({ type: 'success', snapshot }) +} + +main().catch((error) => { + parentPort?.postMessage({ type: 'error', error: error instanceof Error ? error.message : String(error) }) +}) diff --git a/packages/node-runtime/src/services/global-insight/worker-runner.test.ts b/packages/node-runtime/src/services/global-insight/worker-runner.test.ts new file mode 100644 index 00000000..4d191061 --- /dev/null +++ b/packages/node-runtime/src/services/global-insight/worker-runner.test.ts @@ -0,0 +1,26 @@ +import assert from 'node:assert/strict' +import test from 'node:test' +import { resolveDefaultAnnualSummaryWorkerEntryUrl } from './worker-runner' + +test('resolves TypeScript development worker beside the runner', () => { + assert.equal( + resolveDefaultAnnualSummaryWorkerEntryUrl( + 'file:///repo/packages/node-runtime/src/services/global-insight/worker-runner.ts' + ).href, + 'file:///repo/packages/node-runtime/src/services/global-insight/worker-entry.ts' + ) +}) + +test('resolves bundled CLI worker beside the mjs bundle', () => { + assert.equal( + resolveDefaultAnnualSummaryWorkerEntryUrl('file:///app/cli/dist/global-insight-worker.mjs').href, + 'file:///app/cli/dist/global-insight-worker.mjs' + ) +}) + +test('falls back to the desktop bundled worker when no sibling entry exists', () => { + assert.equal( + resolveDefaultAnnualSummaryWorkerEntryUrl('file:///app/dist/main/index.js', () => false).href, + 'file:///app/dist/main/global-insight-worker.js' + ) +}) diff --git a/packages/node-runtime/src/services/global-insight/worker-runner.ts b/packages/node-runtime/src/services/global-insight/worker-runner.ts new file mode 100644 index 00000000..9ad1f196 --- /dev/null +++ b/packages/node-runtime/src/services/global-insight/worker-runner.ts @@ -0,0 +1,119 @@ +import { existsSync } from 'node:fs' +import { Worker, type WorkerOptions } from 'node:worker_threads' +import type { PathProvider } from '@openchatlab/core' +import type { AnnualSummaryRange } from '@openchatlab/shared-types' +import type { RuntimeIdentity } from '../../data-dir-compat' +import { snapshotPathProvider } from '../../semantic-index/static-path-provider' +import type { AnnualSummaryComputeProgress, AnnualSummarySnapshot } from './types' + +type ModuleWorkerOptions = WorkerOptions & { type: 'module' } +type EntryExists = (url: URL) => boolean + +export interface AnnualSummaryRunnerOptions { + signature: string + range: AnnualSummaryRange + onProgress: (progress: AnnualSummaryComputeProgress) => void + signal: AbortSignal +} + +export type AnnualSummaryComputeRunner = (options: AnnualSummaryRunnerOptions) => Promise + +export interface AnnualSummaryWorkerRunnerOptions { + pathProvider: PathProvider + runtimeIdentity?: RuntimeIdentity + nativeBinding?: string + workerEntryUrl?: string | URL +} + +interface WorkerMessage { + type: 'progress' | 'success' | 'error' + progress?: AnnualSummaryComputeProgress + snapshot?: AnnualSummarySnapshot + error?: string +} + +export function resolveDefaultAnnualSummaryWorkerEntryUrl( + currentModuleUrl: string | URL = import.meta.url, + entryExists: EntryExists = (url) => existsSync(url) +): URL { + const moduleUrl = typeof currentModuleUrl === 'string' ? currentModuleUrl : currentModuleUrl.href + if (moduleUrl.endsWith('.ts')) return new URL('./worker-entry.ts', moduleUrl) + if (moduleUrl.endsWith('.mjs')) return new URL('./global-insight-worker.mjs', moduleUrl) + const sibling = new URL('./worker-entry.js', moduleUrl) + return entryExists(sibling) ? sibling : new URL('./global-insight-worker.js', moduleUrl) +} + +export function createAnnualSummaryWorkerRunner(options: AnnualSummaryWorkerRunnerOptions): AnnualSummaryComputeRunner { + return ({ signature, range, signal, onProgress }) => + new Promise((resolve, reject) => { + if (signal.aborted) { + reject(new Error('annual summary worker aborted')) + return + } + const worker = createWorker( + { + paths: snapshotPathProvider(options.pathProvider), + runtimeIdentity: options.runtimeIdentity, + nativeBinding: options.nativeBinding, + signature, + range, + }, + options.workerEntryUrl + ) + let settled = false + const abort = () => { + if (settled) return + settled = true + void worker.terminate() + reject(new Error('annual summary worker aborted')) + } + signal.addEventListener('abort', abort, { once: true }) + worker.on('message', (message: WorkerMessage) => { + if (message.type === 'progress' && message.progress) { + onProgress(message.progress) + return + } + if (message.type === 'success' && message.snapshot) { + settled = true + signal.removeEventListener('abort', abort) + resolve(message.snapshot) + void worker.terminate() + return + } + if (message.type === 'error') { + settled = true + signal.removeEventListener('abort', abort) + reject(new Error(message.error ?? 'annual summary worker failed')) + void worker.terminate() + } + }) + worker.on('error', (error) => { + if (settled) return + settled = true + signal.removeEventListener('abort', abort) + reject(error) + }) + worker.on('exit', (code) => { + if (settled || code === 0) return + settled = true + signal.removeEventListener('abort', abort) + reject(new Error(`annual summary worker exited with code ${code}`)) + }) + }) +} + +function createWorker(workerData: unknown, entryUrlInput?: string | URL): Worker { + const entryUrl = entryUrlInput + ? typeof entryUrlInput === 'string' + ? new URL(entryUrlInput) + : entryUrlInput + : resolveDefaultAnnualSummaryWorkerEntryUrl() + if (!entryUrl.href.endsWith('.ts')) return new Worker(entryUrl, { workerData }) + const bootstrap = ` + import { register } from 'tsx/esm/api'; + register(); + await import(${JSON.stringify(entryUrl.href)}); + ` + const workerOptions: ModuleWorkerOptions = { eval: true, type: 'module', workerData, execArgv: [] } + return new Worker(bootstrap, workerOptions) +} diff --git a/packages/node-runtime/src/services/index.ts b/packages/node-runtime/src/services/index.ts index 37186a96..5067e47a 100644 --- a/packages/node-runtime/src/services/index.ts +++ b/packages/node-runtime/src/services/index.ts @@ -68,6 +68,15 @@ export type { PeopleRelationshipsServiceOptions, } from './people/relationships' +// Global insight service +export { ANNUAL_SUMMARY_ALGORITHM_VERSION, createGlobalInsightService } from './global-insight' +export type { + AnnualSummaryComputeRunner, + GlobalInsightService, + GlobalInsightServiceDeps, + GlobalInsightServiceOptions, +} from './global-insight' + // Merge cache export { MergeSessionCache } from './merge-cache'