diff --git a/apps/cli/src/http/index.ts b/apps/cli/src/http/index.ts index 2f9d5b28..248a2a01 100644 --- a/apps/cli/src/http/index.ts +++ b/apps/cli/src/http/index.ts @@ -206,6 +206,7 @@ export async function startHttpServer(options?: HttpServerOptions): Promise<{ registerWebRoutes(server, dbManager, { pathProvider, nativeBinding, + runtimeIdentity: runtime, semanticIndexService, aiContext: { aiDataDir, diff --git a/apps/cli/src/http/routes/web/index.ts b/apps/cli/src/http/routes/web/index.ts index 1cc92a90..f44beebd 100644 --- a/apps/cli/src/http/routes/web/index.ts +++ b/apps/cli/src/http/routes/web/index.ts @@ -30,6 +30,7 @@ import type { CustomProviderStore, CustomModelStore, PendingDataDirMigration, + RuntimeIdentity, } from '@openchatlab/node-runtime' import { createDatabaseManagerAdapter, @@ -64,6 +65,7 @@ export function registerWebRoutes( options?: { pathProvider?: PathProvider nativeBinding?: string + runtimeIdentity?: RuntimeIdentity aiContext?: AiContextOptions /** 由 server 入口注入的共享语义索引运行时；传入时由调用方管理生命周期 */ semanticIndexService?: SemanticIndexRuntime @@ -114,6 +116,7 @@ export function registerWebRoutes( dbManager, sessionAdapter: adapter, pathProvider: resolvedPathProvider, + runtimeIdentity: options?.runtimeIdentity, getVersion, nativeBinding: options?.nativeBinding, semanticIndexService, diff --git a/apps/cli/tsup.config.ts b/apps/cli/tsup.config.ts index 6c737ab9..433672be 100644 --- a/apps/cli/tsup.config.ts +++ b/apps/cli/tsup.config.ts @@ -5,6 +5,7 @@ export default defineConfig({ cli: 'src/cli.ts', index: 'src/index.ts', 'semantic-index-worker': '../../packages/node-runtime/src/semantic-index/worker-thread-entry.ts', + 'contacts-worker': '../../packages/node-runtime/src/services/contacts-worker-entry.ts', }, format: ['esm'], outDir: 'dist', diff --git a/apps/desktop/electron.vite.config.ts b/apps/desktop/electron.vite.config.ts index 0fb2bceb..610e73f6 100644 --- a/apps/desktop/electron.vite.config.ts +++ b/apps/desktop/electron.vite.config.ts @@ -29,6 +29,7 @@ export default defineConfig({ index: resolve(__dirname, 'main/index.ts'), 'worker/dbWorker': resolve(__dirname, 'main/worker/dbWorker.ts'), 'semantic-index-worker': resolve(rootDir, 'packages/node-runtime/src/semantic-index/worker-thread-entry.ts'), + 'contacts-worker': resolve(rootDir, 'packages/node-runtime/src/services/contacts-worker-entry.ts'), }, }, }, diff --git a/apps/desktop/main/internal-api.ts b/apps/desktop/main/internal-api.ts index b6392374..b39a76ac 100644 --- a/apps/desktop/main/internal-api.ts +++ b/apps/desktop/main/internal-api.ts @@ -161,6 +161,7 @@ export async function startInternalServer(pathProvider: PathProvider): Promise getDesktopAppVersion(app.getVersion()), mergeSessionCache: newMergeCache, streamImport: electronStreamImport, diff --git a/packages/http-routes/src/context.ts b/packages/http-routes/src/context.ts index e0a89220..7cb521d8 100644 --- a/packages/http-routes/src/context.ts +++ b/packages/http-routes/src/context.ts @@ -25,7 +25,9 @@ import type { MergeSessionCache, AgentStreamChunk, SemanticIndexRuntime, + ContactsService, } from '@openchatlab/node-runtime' +import type { RuntimeIdentity } from '@openchatlab/node-runtime' export interface HttpRouteContext { dbManager: DatabaseManager @@ -33,9 +35,11 @@ export interface HttpRouteContext { pathProvider: PathProvider getVersion: () => string + runtimeIdentity?: RuntimeIdentity /** Native binding path for better-sqlite3 (CLI needs it, Electron does not) */ nativeBinding?: string + contactsService?: ContactsService preferencesManager?: PreferencesManager diff --git a/packages/http-routes/src/routes/web/contacts.test.ts b/packages/http-routes/src/routes/web/contacts.test.ts index 4c5c2eff..3c1ee860 100644 --- a/packages/http-routes/src/routes/web/contacts.test.ts +++ b/packages/http-routes/src/routes/web/contacts.test.ts @@ -5,143 +5,126 @@ */ 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 Fastify from 'fastify' -import { CHAT_DB_SCHEMA } from '@openchatlab/core' import type { PathProvider } from '@openchatlab/core' import type { ContactsResponse } from '@openchatlab/shared-types' -import { openBetterSqliteDatabase } from '@openchatlab/node-runtime' -import type { DatabaseManager, SessionRuntimeAdapter } from '@openchatlab/node-runtime' +import type { ContactsService, DatabaseManager, SessionRuntimeAdapter } from '@openchatlab/node-runtime' import type { HttpRouteContext } from '../../context' import { registerContactsRoutes } from './contacts' -const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') - -function makeTempDir(): string { - const baseDir = fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir() - return fs.mkdtempSync(path.join(baseDir, 'chatlab-contacts-routes-')) -} - -class TestEnv { - readonly dir = makeTempDir() - private readonly dbPath = path.join(this.dir, 'private-a.db') - - seed(): void { - const db = openBetterSqliteDatabase(this.dbPath, { nativeBinding }) - db.exec(CHAT_DB_SCHEMA) - db.prepare(`INSERT INTO meta (name, platform, type, imported_at, owner_id) VALUES (?, ?, ?, ?, ?)`).run( - 'private-a', - 'weixin', - 'private', - 1780000000, - 'owner' - ) - db.prepare(`INSERT INTO member (id, platform_id, account_name) VALUES (?, ?, ?), (?, ?, ?)`).run( - 1, - 'owner', - 'Me', - 2, - 'alice', - 'Alice' - ) - for (let i = 0; i < 60; i++) { - db.prepare( - `INSERT INTO message (id, sender_id, ts, type, content, platform_message_id) VALUES (?, ?, ?, 0, ?, ?)` - ).run(i + 1, i % 2 === 0 ? 1 : 2, 1704103200 + i, `message ${i + 1}`, `m-${i + 1}`) - } - db.close() - } - - context(): HttpRouteContext { - const pathProvider: PathProvider = { - getSystemDir: () => this.dir, - getUserDataDir: () => this.dir, - getDatabaseDir: () => this.dir, - getVectorDir: () => path.join(this.dir, 'vector'), - getAiDataDir: () => path.join(this.dir, 'ai'), - getSettingsDir: () => path.join(this.dir, 'settings'), - getCacheDir: () => path.join(this.dir, 'cache'), - getTempDir: () => path.join(this.dir, 'temp'), - getLogsDir: () => path.join(this.dir, 'logs'), - getDownloadsDir: () => path.join(this.dir, 'downloads'), - } - const open = (readonly: boolean) => openBetterSqliteDatabase(this.dbPath, { readonly, nativeBinding }) - const sessionAdapter: SessionRuntimeAdapter = { - listSessionIds: () => ['private-a'], - openReadonly: () => open(true), - openWritable: () => open(false), - closeSession: () => {}, - getDbPath: () => this.dbPath, - deleteSessionFile: () => false, - ensureReadonly: () => open(true), - ensureWritable: () => open(false), - } - - return { - sessionAdapter, - pathProvider, - dbManager: {} as DatabaseManager, - getVersion: () => 'test', - } as HttpRouteContext - } - - touchDb(): void { - fs.utimesSync(this.dbPath, new Date(), new Date(Date.now() + 5000)) - } - - cleanup(): void { - fs.rmSync(this.dir, { recursive: true, force: true }) +function emptyContactsResponse(status: ContactsResponse['cache']['status'] = 'missing'): ContactsResponse { + return { + contacts: [], + diagnostics: { + privateSessionCount: 0, + contactsEnabled: false, + skippedMissingOwnerSessions: 0, + skippedUnresolvedOwnerSessions: 0, + skippedAmbiguousPrivateSessions: 0, + skippedInvalidPlatformIdMembers: 0, + skippedFailedSessions: 0, + hiddenLowSignalNonFriends: 0, + warnings: [], + }, + cache: { + status, + computedAt: null, + }, + algorithmVersion: 'contacts-v1', + task: { + id: 'task-1', + status: 'running', + startedAt: 1000, + finishedAt: null, + processedSessions: 0, + totalSessions: 1, + }, } } -test('contacts routes return contacts and recompute results', async (t) => { - const env = new TestEnv() - env.seed() - t.after(() => env.cleanup()) +class FakeContactsService implements ContactsService { + getCalls: Array<{ acceptStale?: boolean }> = [] + recomputeCalls = 0 + + getContacts(options?: { acceptStale?: boolean }): ContactsResponse { + this.getCalls.push({ acceptStale: options?.acceptStale }) + return emptyContactsResponse('missing') + } + + startRecompute(): ContactsResponse { + this.recomputeCalls++ + return emptyContactsResponse('stale') + } + + invalidateContactsCache(): void { + throw new Error('not used in route contract tests') + } +} + +function createMockContext(contactsService: ContactsService): HttpRouteContext { + const pathProvider: PathProvider = { + getSystemDir: () => path.join('/tmp', 'chatlab-contacts-route-test'), + getUserDataDir: () => path.join('/tmp', 'chatlab-contacts-route-test', 'data'), + getDatabaseDir: () => path.join('/tmp', 'chatlab-contacts-route-test', 'data', 'databases'), + getVectorDir: () => path.join('/tmp', 'chatlab-contacts-route-test', 'vector'), + getAiDataDir: () => path.join('/tmp', 'chatlab-contacts-route-test', 'ai'), + getSettingsDir: () => path.join('/tmp', 'chatlab-contacts-route-test', 'settings'), + getCacheDir: () => path.join('/tmp', 'chatlab-contacts-route-test', 'cache'), + getTempDir: () => path.join('/tmp', 'chatlab-contacts-route-test', 'temp'), + getLogsDir: () => path.join('/tmp', 'chatlab-contacts-route-test', 'logs'), + getDownloadsDir: () => path.join('/tmp', 'chatlab-contacts-route-test', 'downloads'), + } + const sessionAdapter = { + listSessionIds: () => [], + } as unknown as SessionRuntimeAdapter + + return { + sessionAdapter, + pathProvider, + contactsService, + dbManager: {} as DatabaseManager, + getVersion: () => 'test', + } as HttpRouteContext +} + +test('GET /_web/contacts returns contacts response with task state', async (t) => { + const service = new FakeContactsService() const app = Fastify() t.after(async () => app.close()) - registerContactsRoutes(app, env.context()) + registerContactsRoutes(app, createMockContext(service)) await app.ready() - const first = await app.inject({ method: 'GET', url: '/_web/contacts' }) - assert.equal(first.statusCode, 200) - const firstBody = first.json() - assert.equal(firstBody.contacts[0].key, 'weixin:alice') - assert.equal(firstBody.cache.status, 'fresh') + const response = await app.inject({ method: 'GET', url: '/_web/contacts?acceptStale=1' }) - const recomputed = await app.inject({ method: 'POST', url: '/_web/contacts/recompute' }) - assert.equal(recomputed.statusCode, 200) - assert.equal(recomputed.json().cache.status, 'fresh') + assert.equal(response.statusCode, 200) + const body = response.json() + assert.equal(body.cache.status, 'missing') + assert.equal(body.task?.status, 'running') + assert.deepEqual(service.getCalls, [{ acceptStale: true }]) }) -test('GET /_web/contacts supports acceptStale query', async (t) => { - const env = new TestEnv() - env.seed() - t.after(() => env.cleanup()) +test('POST /_web/contacts/recompute starts or reuses background recompute without waiting for completion', async (t) => { + const service = new FakeContactsService() const app = Fastify() t.after(async () => app.close()) - registerContactsRoutes(app, env.context()) + registerContactsRoutes(app, createMockContext(service)) await app.ready() - const first = await app.inject({ method: 'GET', url: '/_web/contacts' }) - assert.equal(first.statusCode, 200) + const response = await app.inject({ method: 'POST', url: '/_web/contacts/recompute' }) - env.touchDb() - const stale = await app.inject({ method: 'GET', url: '/_web/contacts?acceptStale=1' }) - assert.equal(stale.statusCode, 200) - assert.equal(stale.json().cache.status, 'stale') + assert.equal(response.statusCode, 200) + const body = response.json() + assert.equal(body.cache.status, 'stale') + assert.equal(body.task?.status, 'running') + assert.equal(service.recomputeCalls, 1) }) test('override routes are not registered', async (t) => { - const env = new TestEnv() - env.seed() - t.after(() => env.cleanup()) const app = Fastify() t.after(async () => app.close()) - registerContactsRoutes(app, env.context()) + registerContactsRoutes(app, createMockContext(new FakeContactsService())) await app.ready() const patched = await app.inject({ diff --git a/packages/http-routes/src/routes/web/contacts.ts b/packages/http-routes/src/routes/web/contacts.ts index c81c6471..1444866f 100644 --- a/packages/http-routes/src/routes/web/contacts.ts +++ b/packages/http-routes/src/routes/web/contacts.ts @@ -5,16 +5,21 @@ import type { HttpRouteContext } from '../../context' type ContactsQuery = { acceptStale?: string } export function registerContactsRoutes(server: FastifyInstance, ctx: HttpRouteContext): void { - const service = createContactsService({ - adapter: ctx.sessionAdapter, - }) + const service = + ctx.contactsService ?? + createContactsService({ + adapter: ctx.sessionAdapter, + pathProvider: ctx.pathProvider, + runtimeIdentity: ctx.runtimeIdentity, + nativeBinding: ctx.nativeBinding, + }) server.get<{ Querystring: ContactsQuery }>('/_web/contacts', async (request) => { return service.getContacts({ acceptStale: isTruthy(request.query.acceptStale) }) }) server.post('/_web/contacts/recompute', async () => { - return service.getContacts({ forceRecompute: true }) + return service.startRecompute() }) } diff --git a/packages/node-runtime/src/services/contacts-compute.ts b/packages/node-runtime/src/services/contacts-compute.ts new file mode 100644 index 00000000..11cf559a --- /dev/null +++ b/packages/node-runtime/src/services/contacts-compute.ts @@ -0,0 +1,369 @@ +import { ChatType } from '@openchatlab/shared-types' +import type { ChatPlatform, ContactItem, ContactsDiagnostics, ContactSourceSession } from '@openchatlab/shared-types' +import { + MIN_PRIVATE_SESSIONS_FOR_CONTACTS, + computeFriendScores, + computeNonFriendScores, + getGroupContactFacts, + getPrivateContactFacts, + getSessionMeta, + isChatSessionDb, + isLowSignalNonFriend, + isNameMatchPlatform, + resolveOwnerMember, +} from '@openchatlab/core' +import type { ContactMemberRef, SessionMeta } from '@openchatlab/core' +import { appLogger } from '../logging/app-logger' +import type { SessionRuntimeAdapter } from './adapters' + +export const CONTACTS_ALGORITHM_VERSION = 'contacts-v1' + +export interface ContactsWorkerStats { + durationMs: number + totalSessions: number + processedSessions: number + skippedFailedSessions: number +} + +export interface ContactsSnapshot { + contacts: ContactItem[] + diagnostics: ContactsDiagnostics + algorithmVersion: string + signature: string + computedAt: number + workerStats: ContactsWorkerStats +} + +export interface ContactsComputeProgress { + processedSessions: number + totalSessions: number + currentSessionId?: string +} + +export interface ComputeContactsSnapshotOptions { + adapter: SessionRuntimeAdapter + signature: string + now?: () => number + onProgress?: (progress: ContactsComputeProgress) => void +} + +interface ContactAccumulator { + key: string + platform: ChatPlatform + platformId: string + sessionScoped: boolean + sessionId?: string + displayName: string + aliases: Set + avatar: string | null + isFriend: boolean + privateMessageCount: number + activePrivateMonths: Set + commonGroupSessionIds: Set + coOccurrenceCount: number + coOccurrenceRawScore: number + replyInteractionCount: number + repliesFromOwnerToContact: number + repliesFromContactToOwner: number + sourceSessions: ContactSourceSession[] + lastInteractionTs: number | null +} + +interface BuildContactsResult { + contacts: ContactItem[] + diagnostics: ContactsDiagnostics +} + +export function computeContactsSnapshot(options: ComputeContactsSnapshotOptions): ContactsSnapshot { + const startedAt = options.now?.() ?? Date.now() + const sessionIds = options.adapter.listSessionIds() + const result = computeContacts({ + adapter: options.adapter, + sessionIds, + onProgress: options.onProgress, + }) + const finishedAt = options.now?.() ?? Date.now() + return { + ...result, + algorithmVersion: CONTACTS_ALGORITHM_VERSION, + signature: options.signature, + computedAt: finishedAt, + workerStats: { + durationMs: Math.max(0, finishedAt - startedAt), + totalSessions: sessionIds.length, + processedSessions: sessionIds.length, + skippedFailedSessions: result.diagnostics.skippedFailedSessions, + }, + } +} + +function computeContacts(options: { + adapter: SessionRuntimeAdapter + sessionIds: string[] + onProgress?: (progress: ContactsComputeProgress) => void +}): BuildContactsResult { + const diagnostics = createEmptyDiagnostics() + const accumulators = new Map() + let processedSessions = 0 + + for (const sessionId of options.sessionIds) { + options.onProgress?.({ processedSessions, totalSessions: options.sessionIds.length, currentSessionId: sessionId }) + try { + const db = options.adapter.openReadonly(sessionId) + if (!db || !isChatSessionDb(db)) continue + const meta = getSessionMeta(db) + if (!meta) continue + if (meta.type === ChatType.PRIVATE) diagnostics.privateSessionCount++ + if (meta.type !== ChatType.PRIVATE && meta.type !== ChatType.GROUP) continue + + if (!meta.ownerId?.trim()) { + diagnostics.skippedMissingOwnerSessions++ + continue + } + + const owner = resolveOwnerMember(db) + if (!owner) { + diagnostics.skippedUnresolvedOwnerSessions++ + continue + } + + if (meta.type === ChatType.PRIVATE) { + collectPrivateSession(accumulators, diagnostics, sessionId, meta, owner.id, db) + } else { + collectGroupSession(accumulators, sessionId, meta, owner.id, db) + } + } catch (error) { + diagnostics.skippedFailedSessions++ + appLogger.error('contacts', `failed to process contact session: ${sessionId}`, error) + } finally { + processedSessions++ + options.onProgress?.({ processedSessions, totalSessions: options.sessionIds.length, currentSessionId: sessionId }) + } + } + + diagnostics.contactsEnabled = diagnostics.privateSessionCount > MIN_PRIVATE_SESSIONS_FOR_CONTACTS + const contacts = buildContactItems([...accumulators.values()], diagnostics) + return { contacts, diagnostics } +} + +function collectPrivateSession( + accumulators: Map, + diagnostics: ContactsDiagnostics, + sessionId: string, + meta: SessionMeta, + ownerMemberId: number, + db: Parameters[0] +): void { + const facts = getPrivateContactFacts(db, ownerMemberId) + if (facts.type === 'missing') return + if (facts.type === 'ambiguous') { + diagnostics.skippedAmbiguousPrivateSessions++ + return + } + + const acc = getOrCreateAccumulator(accumulators, sessionId, meta, facts.contact) + acc.isFriend = true + acc.privateMessageCount += facts.privateMessageCount + for (const month of facts.activeMonths) acc.activePrivateMonths.add(month) + updateLastInteraction(acc, facts.lastMessageTs) + acc.sourceSessions.push({ + id: sessionId, + name: meta.name, + platform: meta.platform, + type: ChatType.PRIVATE, + messageCount: facts.privateMessageCount, + privateMessageCount: facts.privateMessageCount, + lastMessageTs: facts.lastMessageTs, + }) +} + +function collectGroupSession( + accumulators: Map, + sessionId: string, + meta: SessionMeta, + ownerMemberId: number, + db: Parameters[0] +): void { + for (const facts of getGroupContactFacts(db, ownerMemberId)) { + const acc = getOrCreateAccumulator(accumulators, sessionId, meta, facts.contact) + acc.commonGroupSessionIds.add(sessionId) + acc.coOccurrenceCount += facts.coOccurrenceCount + acc.coOccurrenceRawScore += facts.coOccurrenceRawScore + acc.replyInteractionCount += facts.replyInteractionCount + acc.repliesFromOwnerToContact += facts.repliesFromOwnerToContact + acc.repliesFromContactToOwner += facts.repliesFromContactToOwner + updateLastInteraction(acc, facts.lastInteractionTs) + acc.sourceSessions.push({ + id: sessionId, + name: meta.name, + platform: meta.platform, + type: ChatType.GROUP, + messageCount: facts.messageCount, + coOccurrenceCount: facts.coOccurrenceCount, + coOccurrenceRawScore: facts.coOccurrenceRawScore, + replyInteractionCount: facts.replyInteractionCount, + repliesFromOwnerToContact: facts.repliesFromOwnerToContact, + repliesFromContactToOwner: facts.repliesFromContactToOwner, + lastInteractionTs: facts.lastInteractionTs, + }) + } +} + +function buildContactItems(accumulators: ContactAccumulator[], diagnostics: ContactsDiagnostics): ContactItem[] { + const friendInputs = accumulators + .filter((acc) => acc.isFriend) + .map((acc) => ({ + acc, + privateMessageCount: acc.privateMessageCount, + activeMonths: [...acc.activePrivateMonths], + commonGroupCount: acc.commonGroupSessionIds.size, + })) + const nonFriendInputs = accumulators + .filter((acc) => !acc.isFriend) + .map((acc) => ({ + acc, + coOccurrenceRawScore: acc.coOccurrenceRawScore, + commonGroupCount: acc.commonGroupSessionIds.size, + replyInteractionCount: acc.replyInteractionCount, + coOccurrenceCount: acc.coOccurrenceCount, + })) + + const friendScores = computeFriendScores(friendInputs) + const nonFriendScores = computeNonFriendScores(nonFriendInputs) + const contacts: ContactItem[] = [] + + for (const input of friendInputs) { + const score = friendScores.get(input) ?? { score: 0, scoreBreakdown: {} } + contacts.push(toContactItem(input.acc, 'friend', false, score)) + } + + for (const input of nonFriendInputs) { + const score = nonFriendScores.get(input) ?? { score: 0, scoreBreakdown: {} } + const isLowSignal = isLowSignalNonFriend(input) + const item = toContactItem(input.acc, 'non_friend', isLowSignal, score) + if (item.isLowSignal) diagnostics.hiddenLowSignalNonFriends++ + contacts.push(item) + } + + return contacts.sort((a, b) => b.score - a.score || a.displayName.localeCompare(b.displayName)) +} + +function toContactItem( + acc: ContactAccumulator, + pool: 'friend' | 'non_friend', + isLowSignal: boolean, + scoring: { score: number; scoreBreakdown: ContactItem['scoreBreakdown'] } +): ContactItem { + const aliases = [...acc.aliases].filter((alias) => alias !== acc.displayName) + const searchText = [acc.displayName, acc.platformId, ...aliases].join(' ').toLowerCase() + + return { + key: acc.key, + platform: acc.platform, + platformId: acc.platformId, + sessionScoped: acc.sessionScoped, + sessionId: acc.sessionId, + displayName: acc.displayName, + aliases, + avatar: acc.avatar, + isFriend: acc.isFriend, + pool, + isLowSignal, + score: scoring.score, + scoreBreakdown: { + ...scoring.scoreBreakdown, + privateMessageCount: acc.privateMessageCount || scoring.scoreBreakdown.privateMessageCount, + activePrivateMonths: acc.activePrivateMonths.size || scoring.scoreBreakdown.activePrivateMonths, + commonGroupCount: acc.commonGroupSessionIds.size, + coOccurrenceCount: acc.coOccurrenceCount, + coOccurrenceRawScore: acc.coOccurrenceRawScore, + replyInteractionCount: acc.replyInteractionCount, + repliesFromOwnerToContact: acc.repliesFromOwnerToContact, + repliesFromContactToOwner: acc.repliesFromContactToOwner, + }, + sourceSessions: acc.sourceSessions, + searchText, + lastInteractionTs: acc.lastInteractionTs, + } +} + +export function createEmptyContactsDiagnostics(): ContactsDiagnostics { + return createEmptyDiagnostics() +} + +function createEmptyDiagnostics(): ContactsDiagnostics { + return { + privateSessionCount: 0, + contactsEnabled: false, + skippedMissingOwnerSessions: 0, + skippedUnresolvedOwnerSessions: 0, + skippedAmbiguousPrivateSessions: 0, + skippedInvalidPlatformIdMembers: 0, + skippedFailedSessions: 0, + hiddenLowSignalNonFriends: 0, + warnings: [], + } +} + +function getOrCreateAccumulator( + accumulators: Map, + sessionId: string, + meta: SessionMeta, + contact: ContactMemberRef +): ContactAccumulator { + const sessionScoped = isNameMatchPlatform(meta.platform) + const key = buildContactKey(meta.platform, contact.platformId, sessionScoped ? sessionId : undefined) + const existing = accumulators.get(key) + if (existing) { + mergeContactIdentity(existing, contact) + return existing + } + + const created: ContactAccumulator = { + key, + platform: meta.platform, + platformId: contact.platformId, + sessionScoped, + sessionId: sessionScoped ? sessionId : undefined, + displayName: contact.name || contact.platformId, + aliases: new Set([contact.platformId, contact.name].filter(Boolean)), + avatar: contact.avatar, + isFriend: false, + privateMessageCount: 0, + activePrivateMonths: new Set(), + commonGroupSessionIds: new Set(), + coOccurrenceCount: 0, + coOccurrenceRawScore: 0, + replyInteractionCount: 0, + repliesFromOwnerToContact: 0, + repliesFromContactToOwner: 0, + sourceSessions: [], + lastInteractionTs: null, + } + accumulators.set(key, created) + return created +} + +function buildContactKey(platform: ChatPlatform, platformId: string, sessionId?: string): string { + const normalizedPlatform = platform.trim() + const normalizedPlatformId = platformId.trim() + if (!normalizedPlatform) throw new Error('platform is required') + if (!normalizedPlatformId) throw new Error('platformId is required') + return sessionId?.trim() + ? `${normalizedPlatform}:${sessionId.trim()}:${normalizedPlatformId}` + : `${normalizedPlatform}:${normalizedPlatformId}` +} + +function mergeContactIdentity(acc: ContactAccumulator, contact: ContactMemberRef): void { + if (contact.name) acc.aliases.add(contact.name) + acc.aliases.add(contact.platformId) + if ((!acc.displayName || acc.displayName === acc.platformId) && contact.name) { + acc.displayName = contact.name + } + if (!acc.avatar && contact.avatar) acc.avatar = contact.avatar +} + +function updateLastInteraction(acc: ContactAccumulator, ts: number | null): void { + if (ts === null) return + acc.lastInteractionTs = Math.max(acc.lastInteractionTs ?? 0, ts) +} diff --git a/packages/node-runtime/src/services/contacts-service.test.ts b/packages/node-runtime/src/services/contacts-service.test.ts index 9382a1ab..f49d8872 100644 --- a/packages/node-runtime/src/services/contacts-service.test.ts +++ b/packages/node-runtime/src/services/contacts-service.test.ts @@ -11,9 +11,13 @@ import path from 'node:path' import test from 'node:test' import { CHAT_DB_SCHEMA } from '@openchatlab/core' import type { DatabaseAdapter } from '@openchatlab/core' +import { ChatType } from '@openchatlab/shared-types' +import type { ContactsResponse } from '@openchatlab/shared-types' import { openBetterSqliteDatabase } from '../better-sqlite3-adapter' import type { SessionRuntimeAdapter } from './adapters' +import { CONTACTS_ALGORITHM_VERSION, computeContactsSnapshot, type ContactsSnapshot } from './contacts-compute' import { createContactsService } from './contacts-service' +import type { PathProvider } from '@openchatlab/core' const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') @@ -129,6 +133,21 @@ class TestEnv { return dbPath } + pathProvider(): PathProvider { + return { + getSystemDir: () => this.dir, + getUserDataDir: () => this.dir, + getDatabaseDir: () => this.dir, + getVectorDir: () => path.join(this.dir, 'vector'), + getAiDataDir: () => path.join(this.dir, 'ai'), + getSettingsDir: () => path.join(this.dir, 'settings'), + getCacheDir: () => path.join(this.dir, 'cache'), + getTempDir: () => path.join(this.dir, 'temp'), + getLogsDir: () => path.join(this.dir, 'logs'), + getDownloadsDir: () => path.join(this.dir, 'downloads'), + } + } + cleanup(): void { for (const db of this.openDbs) { try { @@ -192,7 +211,7 @@ test('aggregates stable-id contacts across private and group sessions', (t) => { ], }) - const result = createContactsService({ adapter: env.adapter }).getContacts() + const result = computeContactsSnapshot({ adapter: env.adapter, signature: 'sig-1' }) const byKey = new Map(result.contacts.map((contact) => [contact.key, contact])) const alice = byKey.get('weixin:alice') const bob = byKey.get('weixin:bob') @@ -214,7 +233,6 @@ test('aggregates stable-id contacts across private and group sessions', (t) => { assert.equal(result.diagnostics.privateSessionCount, 2) assert.equal(result.diagnostics.contactsEnabled, false) - assert.equal(result.cache.status, 'fresh') }) test('records diagnostics for missing owner, unresolved owner, and ambiguous private sessions', (t) => { @@ -247,7 +265,7 @@ test('records diagnostics for missing owner, unresolved owner, and ambiguous pri ], }) - const result = createContactsService({ adapter: env.adapter }).getContacts() + const result = computeContactsSnapshot({ adapter: env.adapter, signature: 'sig-1' }) assert.equal(result.contacts.length, 0) assert.equal(result.diagnostics.privateSessionCount, 3) @@ -274,8 +292,7 @@ test('keeps name-match platform contacts session-scoped', (t) => { }) } - const service = createContactsService({ adapter: env.adapter }) - const result = service.getContacts() + const result = computeContactsSnapshot({ adapter: env.adapter, signature: 'sig-1' }) const keys = result.contacts.map((contact) => contact.key).sort() assert.deepEqual(keys, ['whatsapp:whatsapp-a:Alice', 'whatsapp:whatsapp-b:Alice']) @@ -312,8 +329,7 @@ test('sorts contacts by score and marks low-signal non-friends', (t) => { ], }) - const service = createContactsService({ adapter: env.adapter }) - const result = service.getContacts() + const result = computeContactsSnapshot({ adapter: env.adapter, signature: 'sig-1' }) assert.deepEqual( result.contacts.map((contact) => contact.key), @@ -323,7 +339,77 @@ test('sorts contacts by score and marks low-signal non-friends', (t) => { assert.equal(result.diagnostics.hiddenLowSignalNonFriends, 1) }) -test('returns stale cached contacts when signature changes and acceptStale is true', (t) => { +function deferred() { + let resolve!: (value: T) => void + let reject!: (error: Error) => void + const promise = new Promise((res, rej) => { + resolve = res + reject = rej + }) + return { promise, resolve, reject } +} + +async function waitForTaskSettled(service: { getContacts: (options?: { acceptStale?: boolean }) => ContactsResponse }) { + for (let i = 0; i < 100; i++) { + const response = service.getContacts({ acceptStale: true }) + if (response.task?.status !== 'running') return response + await new Promise((resolve) => setTimeout(resolve, 20)) + } + return service.getContacts({ acceptStale: true }) +} + +function makeRuntimeSnapshot(signature: string, computedAt: number): ContactsSnapshot { + return { + contacts: [ + { + key: 'weixin:alice', + platform: 'weixin', + platformId: 'alice', + sessionScoped: false, + displayName: 'Alice', + aliases: [], + avatar: null, + isFriend: true, + pool: 'friend', + isLowSignal: false, + score: 1, + scoreBreakdown: {}, + sourceSessions: [ + { + id: 'private-a', + name: 'private-a', + platform: 'weixin', + type: ChatType.PRIVATE, + }, + ], + searchText: 'alice', + lastInteractionTs: null, + }, + ], + diagnostics: { + privateSessionCount: 1, + contactsEnabled: false, + skippedMissingOwnerSessions: 0, + skippedUnresolvedOwnerSessions: 0, + skippedAmbiguousPrivateSessions: 0, + skippedInvalidPlatformIdMembers: 0, + skippedFailedSessions: 0, + hiddenLowSignalNonFriends: 0, + warnings: [], + }, + algorithmVersion: CONTACTS_ALGORITHM_VERSION, + signature, + computedAt, + workerStats: { + durationMs: 10, + totalSessions: 1, + processedSessions: 1, + skippedFailedSessions: 0, + }, + } +} + +test('returns missing snapshot and starts a background contacts task without synchronous compute', async (t) => { const env = new TestEnv() t.after(() => env.cleanup()) let now = 1000 @@ -340,31 +426,84 @@ test('returns stale cached contacts when signature changes and acceptStale is tr messages: privateMessages(5, 1, 1704103200), }) - const service = createContactsService({ adapter: env.adapter, now: () => now }) - const first = service.getContacts() - assert.equal(first.cache.status, 'fresh') - assert.equal(first.cache.computedAt, 1000) + const pending = deferred() + let runCalls = 0 + let runnerSignature = '' + const service = createContactsService({ + adapter: env.adapter, + systemDir: env.dir, + now: () => now, + runner: ({ signature }) => { + runCalls++ + runnerSignature = signature + return pending.promise + }, + }) + + const first = service.getContacts({ acceptStale: true }) + + assert.equal(runCalls, 1) + assert.equal(first.cache.status, 'missing') + assert.equal(first.contacts.length, 0) + assert.equal(first.task?.status, 'running') + + now = 2000 + pending.resolve(makeRuntimeSnapshot(runnerSignature, now)) + const finished = await waitForTaskSettled(service) + + assert.equal(finished.cache.status, 'fresh') + assert.equal(finished.cache.computedAt, 2000) + assert.equal(finished.contacts[0].key, 'weixin:alice') + assert.equal(finished.task?.status, 'succeeded') +}) + +test('returns stale snapshot and reuses one in-flight task after signature changes', async (t) => { + const env = new TestEnv() + t.after(() => env.cleanup()) + let now = 1000 + + env.seed({ + id: 'private-a', + platform: 'weixin', + type: 'private', + ownerId: 'owner', + members: [ + { id: 1, platformId: 'owner' }, + { id: 2, platformId: 'alice' }, + ], + messages: privateMessages(5, 1, 1704103200), + }) + + const firstSnapshot = computeContactsSnapshot({ adapter: env.adapter, signature: 'old-signature', now: () => now }) + const pending = deferred() + let runCalls = 0 + const service = createContactsService({ + adapter: env.adapter, + systemDir: env.dir, + now: () => now, + runner: () => { + runCalls++ + return pending.promise + }, + }) + service.replaceSnapshotForTests!(firstSnapshot) now = 2000 fs.utimesSync(env.dbPath('private-a'), new Date(), new Date(Date.now() + 5000)) const stale = service.getContacts({ acceptStale: true }) - assert.equal(stale.cache.status, 'stale') - assert.equal(stale.cache.computedAt, 1000) - assert.deepEqual( - stale.contacts.map((contact) => contact.key), - first.contacts.map((contact) => contact.key) - ) + const recompute = service.startRecompute() - const fresh = service.getContacts() - assert.equal(fresh.cache.status, 'fresh') - assert.equal(fresh.cache.computedAt, 2000) + assert.equal(runCalls, 1) + assert.equal(stale.cache.status, 'stale') + assert.equal(stale.contacts[0].key, 'weixin:alice') + assert.equal(stale.task?.status, 'running') + assert.equal(recompute.task?.status, 'running') }) -test('recomputes contacts after owner_id changes in a session database', (t) => { +test('temporary contacts worker computes and persists a fresh snapshot', async (t) => { const env = new TestEnv() t.after(() => env.cleanup()) - let now = 1000 env.seed({ id: 'private-a', @@ -378,21 +517,19 @@ test('recomputes contacts after owner_id changes in a session database', (t) => messages: privateMessages(5, 1, 1704103200), }) - const service = createContactsService({ adapter: env.adapter, now: () => now }) - const first = service.getContacts() - assert.deepEqual( - first.contacts.map((contact) => contact.key), - ['weixin:alice'] - ) + const service = createContactsService({ + adapter: env.adapter, + pathProvider: env.pathProvider(), + nativeBinding, + }) - now = 2000 - env.adapter.ensureWritable('private-a').prepare('UPDATE meta SET owner_id = ?').run('alice') + const first = service.getContacts({ acceptStale: true }) + assert.equal(first.cache.status, 'missing') + assert.equal(first.task?.status, 'running') - const second = service.getContacts() - assert.equal(second.cache.status, 'fresh') - assert.equal(second.cache.computedAt, 2000) - assert.deepEqual( - second.contacts.map((contact) => contact.key), - ['weixin:owner'] - ) + const finished = await waitForTaskSettled(service) + assert.equal(finished.cache.status, 'fresh') + assert.equal(finished.contacts[0].key, 'weixin:alice') + assert.equal(finished.task?.status, 'succeeded') + assert.ok(fs.existsSync(path.join(env.dir, 'contacts-snapshot.json'))) }) diff --git a/packages/node-runtime/src/services/contacts-service.ts b/packages/node-runtime/src/services/contacts-service.ts index c9788e95..d126f520 100644 --- a/packages/node-runtime/src/services/contacts-service.ts +++ b/packages/node-runtime/src/services/contacts-service.ts @@ -1,79 +1,52 @@ -import { ChatType } from '@openchatlab/shared-types' -import type { - ChatPlatform, - ContactItem, - ContactsCacheState, - ContactsDiagnostics, - ContactsResponse, - ContactSourceSession, -} from '@openchatlab/shared-types' -import { - MIN_PRIVATE_SESSIONS_FOR_CONTACTS, - computeFriendScores, - computeNonFriendScores, - getGroupContactFacts, - getPrivateContactFacts, - getSessionMeta, - isChatSessionDb, - isLowSignalNonFriend, - isNameMatchPlatform, - resolveOwnerMember, -} from '@openchatlab/core' -import type { ContactMemberRef, SessionMeta } from '@openchatlab/core' -import { getDbFileVersion } from '../cache/analytics-cache' +import type { PathProvider } from '@openchatlab/core' +import type { ContactsCacheState, ContactsResponse, ContactsTaskState } from '@openchatlab/shared-types' +import type { RuntimeIdentity } from '../data-dir-compat' import { appLogger } from '../logging/app-logger' import type { SessionRuntimeAdapter } from './adapters' - -export const CONTACTS_ALGORITHM_VERSION = 'contacts-v1' +import { + CONTACTS_ALGORITHM_VERSION, + createEmptyContactsDiagnostics, + type ContactsComputeProgress, + type ContactsSnapshot, +} from './contacts-compute' +import { buildContactsSignature } from './contacts-signature' +import { cleanupContactsSnapshotTempFiles, readContactsSnapshot, writeContactsSnapshot } from './contacts-snapshot' +import { createContactsWorkerRunner } from './contacts-worker-runner' export interface ContactsServiceOptions { forceRecompute?: boolean acceptStale?: boolean } +export interface ContactsRunnerOptions { + signature: string + onProgress: (progress: ContactsComputeProgress) => void +} + +export type ContactsComputeRunner = (options: ContactsRunnerOptions) => Promise + export interface ContactsServiceDeps { adapter: SessionRuntimeAdapter + systemDir?: string + pathProvider?: PathProvider + runtimeIdentity?: RuntimeIdentity + nativeBinding?: string + workerEntryUrl?: string | URL + runner?: ContactsComputeRunner now?: () => number } export interface ContactsService { getContacts(options?: ContactsServiceOptions): ContactsResponse + startRecompute(): ContactsResponse invalidateContactsCache(): void + replaceSnapshotForTests?(snapshot: ContactsSnapshot): void } -interface CachedContacts { - contacts: ContactItem[] - diagnostics: ContactsDiagnostics - algorithmVersion: string +interface InFlightTask { + id: string signature: string - computedAt: number -} - -interface ContactAccumulator { - key: string - platform: ChatPlatform - platformId: string - sessionScoped: boolean - sessionId?: string - displayName: string - aliases: Set - avatar: string | null - isFriend: boolean - privateMessageCount: number - activePrivateMonths: Set - commonGroupSessionIds: Set - coOccurrenceCount: number - coOccurrenceRawScore: number - replyInteractionCount: number - repliesFromOwnerToContact: number - repliesFromContactToOwner: number - sourceSessions: ContactSourceSession[] - lastInteractionTs: number | null -} - -interface BuildContactsResult { - contacts: ContactItem[] - diagnostics: ContactsDiagnostics + promise: Promise } export function createContactsService(deps: ContactsServiceDeps): ContactsService { @@ -81,244 +54,150 @@ export function createContactsService(deps: ContactsServiceDeps): ContactsServic } class DefaultContactsService implements ContactsService { - private cache: CachedContacts | null = null + private snapshot: ContactsSnapshot | null + private inFlight: InFlightTask | null = null + private task: ContactsTaskState = createIdleTaskState() + private readonly systemDir: string + private readonly runner: ContactsComputeRunner - constructor(private readonly deps: ContactsServiceDeps) {} + constructor(private readonly deps: ContactsServiceDeps) { + this.systemDir = deps.systemDir ?? deps.pathProvider?.getSystemDir() ?? '' + if (!this.systemDir) throw new Error('contacts service requires systemDir or pathProvider') + cleanupContactsSnapshotTempFiles(this.systemDir) + this.snapshot = readContactsSnapshot(this.systemDir, { now: deps.now }) + this.runner = + deps.runner ?? + createContactsWorkerRunner({ + pathProvider: requirePathProvider(deps), + runtimeIdentity: deps.runtimeIdentity, + nativeBinding: deps.nativeBinding, + workerEntryUrl: deps.workerEntryUrl, + }) + } getContacts(options: ContactsServiceOptions = {}): ContactsResponse { - const signature = this.buildSignature() - if (!options.forceRecompute && this.cache) { - if (this.cache.signature === signature) return this.toResponse(this.cache, 'fresh') - if (options.acceptStale) return this.toResponse(this.cache, 'stale', 'signature_changed') - } + const signature = buildContactsSignature(this.deps.adapter) + const cacheStatus = this.getCacheStatus(signature) + if (options.forceRecompute || cacheStatus !== 'fresh') this.ensureTaskStarted(signature) + return this.toResponse(signature) + } - const computedAt = this.now() - const result = this.computeContacts() - this.cache = { - ...result, - algorithmVersion: CONTACTS_ALGORITHM_VERSION, - signature, - computedAt, - } - appLogger.info('contacts', 'contacts recomputed', { - contactCount: result.contacts.length, - privateSessionCount: result.diagnostics.privateSessionCount, - skippedFailedSessions: result.diagnostics.skippedFailedSessions, - }) - return this.toResponse(this.cache, 'fresh') + startRecompute(): ContactsResponse { + const signature = buildContactsSignature(this.deps.adapter) + this.ensureTaskStarted(signature) + return this.toResponse(signature) } invalidateContactsCache(): void { - this.cache = null + this.snapshot = null } - private computeContacts(): BuildContactsResult { - const diagnostics = createEmptyDiagnostics() - const accumulators = new Map() + replaceSnapshotForTests(snapshot: ContactsSnapshot): void { + this.snapshot = snapshot + } - for (const sessionId of this.deps.adapter.listSessionIds()) { - try { - const db = this.deps.adapter.openReadonly(sessionId) - if (!db || !isChatSessionDb(db)) continue - const meta = getSessionMeta(db) - if (!meta) continue - if (meta.type === ChatType.PRIVATE) diagnostics.privateSessionCount++ - if (meta.type !== ChatType.PRIVATE && meta.type !== ChatType.GROUP) continue + private ensureTaskStarted(signature: string): void { + if (this.inFlight) return - if (!meta.ownerId?.trim()) { - diagnostics.skippedMissingOwnerSessions++ - continue - } - - const owner = resolveOwnerMember(db) - if (!owner) { - diagnostics.skippedUnresolvedOwnerSessions++ - continue - } - - if (meta.type === ChatType.PRIVATE) { - this.collectPrivateSession(accumulators, diagnostics, sessionId, meta, owner.id, db) - } else { - this.collectGroupSession(accumulators, sessionId, meta, owner.id, db) - } - } catch (error) { - diagnostics.skippedFailedSessions++ - appLogger.error('contacts', `failed to process contact session: ${sessionId}`, error) - } + const taskId = `contacts_${this.now()}_${Math.random().toString(36).slice(2)}` + this.task = { + id: taskId, + status: 'running', + startedAt: this.now(), + finishedAt: null, + processedSessions: 0, + totalSessions: this.deps.adapter.listSessionIds().length, } - diagnostics.contactsEnabled = diagnostics.privateSessionCount > MIN_PRIVATE_SESSIONS_FOR_CONTACTS - const contacts = this.buildContactItems([...accumulators.values()], diagnostics) - return { contacts, diagnostics } + const promise = this.runner({ + signature, + onProgress: (progress) => { + if (this.task.id !== taskId || this.task.status !== 'running') return + this.task = { + ...this.task, + processedSessions: progress.processedSessions, + totalSessions: progress.totalSessions, + currentSessionId: progress.currentSessionId, + } + }, + }) + this.inFlight = { id: taskId, signature, promise } + + promise + .then((snapshot) => this.handleTaskSuccess(taskId, signature, snapshot)) + .catch((error) => this.handleTaskFailure(taskId, error)) } - private collectPrivateSession( - accumulators: Map, - diagnostics: ContactsDiagnostics, - sessionId: string, - meta: SessionMeta, - ownerMemberId: number, - db: Parameters[0] - ): void { - const facts = getPrivateContactFacts(db, ownerMemberId) - if (facts.type === 'missing') return - if (facts.type === 'ambiguous') { - diagnostics.skippedAmbiguousPrivateSessions++ + private handleTaskSuccess(taskId: string, inputSignature: string, snapshot: ContactsSnapshot): void { + if (this.inFlight?.id !== taskId) return + this.inFlight = null + const latestSignature = buildContactsSignature(this.deps.adapter) + const finishedAt = this.now() + + if (inputSignature !== latestSignature || snapshot.signature !== latestSignature) { + this.task = { + ...this.task, + status: 'superseded', + finishedAt, + } + appLogger.info('contacts', 'contacts worker result discarded because signature changed', { + inputSignature, + latestSignature, + }) return } - const acc = getOrCreateAccumulator(accumulators, sessionId, meta, facts.contact) - acc.isFriend = true - acc.privateMessageCount += facts.privateMessageCount - for (const month of facts.activeMonths) acc.activePrivateMonths.add(month) - updateLastInteraction(acc, facts.lastMessageTs) - acc.sourceSessions.push({ - id: sessionId, - name: meta.name, - platform: meta.platform, - type: ChatType.PRIVATE, - messageCount: facts.privateMessageCount, - privateMessageCount: facts.privateMessageCount, - lastMessageTs: facts.lastMessageTs, - }) - } - - private collectGroupSession( - accumulators: Map, - sessionId: string, - meta: SessionMeta, - ownerMemberId: number, - db: Parameters[0] - ): void { - for (const facts of getGroupContactFacts(db, ownerMemberId)) { - const acc = getOrCreateAccumulator(accumulators, sessionId, meta, facts.contact) - acc.commonGroupSessionIds.add(sessionId) - acc.coOccurrenceCount += facts.coOccurrenceCount - acc.coOccurrenceRawScore += facts.coOccurrenceRawScore - acc.replyInteractionCount += facts.replyInteractionCount - acc.repliesFromOwnerToContact += facts.repliesFromOwnerToContact - acc.repliesFromContactToOwner += facts.repliesFromContactToOwner - updateLastInteraction(acc, facts.lastInteractionTs) - acc.sourceSessions.push({ - id: sessionId, - name: meta.name, - platform: meta.platform, - type: ChatType.GROUP, - messageCount: facts.messageCount, - coOccurrenceCount: facts.coOccurrenceCount, - coOccurrenceRawScore: facts.coOccurrenceRawScore, - replyInteractionCount: facts.replyInteractionCount, - repliesFromOwnerToContact: facts.repliesFromOwnerToContact, - repliesFromContactToOwner: facts.repliesFromContactToOwner, - lastInteractionTs: facts.lastInteractionTs, + try { + writeContactsSnapshot(this.systemDir, snapshot) + this.snapshot = snapshot + this.task = { + ...this.task, + status: 'succeeded', + finishedAt, + processedSessions: snapshot.workerStats.processedSessions, + totalSessions: snapshot.workerStats.totalSessions, + currentSessionId: undefined, + } + appLogger.info('contacts', 'contacts worker snapshot persisted', { + contactCount: snapshot.contacts.length, + durationMs: snapshot.workerStats.durationMs, }) + } catch (error) { + this.handleTaskFailure(taskId, error) } } - private buildContactItems(accumulators: ContactAccumulator[], diagnostics: ContactsDiagnostics): ContactItem[] { - const friendInputs = accumulators - .filter((acc) => acc.isFriend) - .map((acc) => ({ - acc, - privateMessageCount: acc.privateMessageCount, - activeMonths: [...acc.activePrivateMonths], - commonGroupCount: acc.commonGroupSessionIds.size, - })) - const nonFriendInputs = accumulators - .filter((acc) => !acc.isFriend) - .map((acc) => ({ - acc, - coOccurrenceRawScore: acc.coOccurrenceRawScore, - commonGroupCount: acc.commonGroupSessionIds.size, - replyInteractionCount: acc.replyInteractionCount, - coOccurrenceCount: acc.coOccurrenceCount, - })) - - const friendScores = computeFriendScores(friendInputs) - const nonFriendScores = computeNonFriendScores(nonFriendInputs) - - const contacts: ContactItem[] = [] - - // 评分仍按好友/群聊联系人分池计算；分池后的 score 只用于排序，不再生成关系定性标签。 - for (const input of friendInputs) { - const score = friendScores.get(input) ?? { score: 0, scoreBreakdown: {} } - contacts.push(this.toContactItem(input.acc, 'friend', false, score)) + private handleTaskFailure(taskId: string, error: unknown): void { + if (this.inFlight?.id === taskId) this.inFlight = null + const message = error instanceof Error ? error.message : String(error) + this.task = { + ...this.task, + status: 'failed', + finishedAt: this.now(), + lastError: message, } - - for (const input of nonFriendInputs) { - const score = nonFriendScores.get(input) ?? { score: 0, scoreBreakdown: {} } - const isLowSignal = isLowSignalNonFriend(input) - const item = this.toContactItem(input.acc, 'non_friend', isLowSignal, score) - if (item.isLowSignal) diagnostics.hiddenLowSignalNonFriends++ - contacts.push(item) - } - - return contacts.sort((a, b) => b.score - a.score || a.displayName.localeCompare(b.displayName)) + appLogger.error('contacts', 'contacts worker failed', error) } - private toContactItem( - acc: ContactAccumulator, - pool: 'friend' | 'non_friend', - isLowSignal: boolean, - scoring: { score: number; scoreBreakdown: ContactItem['scoreBreakdown'] } - ): ContactItem { - const aliases = [...acc.aliases].filter((alias) => alias !== acc.displayName) - const searchText = [acc.displayName, acc.platformId, ...aliases].join(' ').toLowerCase() + private getCacheStatus(signature: string): ContactsCacheState['status'] { + if (!this.snapshot) return 'missing' + return this.snapshot.signature === signature ? 'fresh' : 'stale' + } + private toResponse(signature: string): ContactsResponse { + const snapshot = this.snapshot + const status = this.getCacheStatus(signature) return { - key: acc.key, - platform: acc.platform, - platformId: acc.platformId, - sessionScoped: acc.sessionScoped, - sessionId: acc.sessionId, - displayName: acc.displayName, - aliases, - avatar: acc.avatar, - isFriend: acc.isFriend, - pool, - isLowSignal, - score: scoring.score, - scoreBreakdown: { - ...scoring.scoreBreakdown, - privateMessageCount: acc.privateMessageCount || scoring.scoreBreakdown.privateMessageCount, - activePrivateMonths: acc.activePrivateMonths.size || scoring.scoreBreakdown.activePrivateMonths, - commonGroupCount: acc.commonGroupSessionIds.size, - coOccurrenceCount: acc.coOccurrenceCount, - coOccurrenceRawScore: acc.coOccurrenceRawScore, - replyInteractionCount: acc.replyInteractionCount, - repliesFromOwnerToContact: acc.repliesFromOwnerToContact, - repliesFromContactToOwner: acc.repliesFromContactToOwner, - }, - sourceSessions: acc.sourceSessions, - searchText, - lastInteractionTs: acc.lastInteractionTs, - } - } - - private buildSignature(): string { - const parts = [`algorithm:${CONTACTS_ALGORITHM_VERSION}`] - for (const sessionId of [...this.deps.adapter.listSessionIds()].sort()) { - const dbPath = this.deps.adapter.getDbPath(sessionId) - parts.push(`${sessionId}:${getDbFileVersion(dbPath)}`) - } - return parts.join('|') - } - - private toResponse( - cached: CachedContacts, - status: ContactsCacheState['status'], - staleReason?: string - ): ContactsResponse { - return { - contacts: cached.contacts, - diagnostics: cached.diagnostics, - algorithmVersion: cached.algorithmVersion, + contacts: snapshot?.contacts ?? [], + diagnostics: snapshot?.diagnostics ?? createEmptyContactsDiagnostics(), + algorithmVersion: snapshot?.algorithmVersion ?? CONTACTS_ALGORITHM_VERSION, cache: { status, - computedAt: cached.computedAt, - signature: cached.signature, - staleReason, + computedAt: snapshot?.computedAt ?? null, + signature: snapshot?.signature, + staleReason: status === 'stale' ? 'signature_changed' : undefined, }, + task: this.task, } } @@ -327,79 +206,20 @@ class DefaultContactsService implements ContactsService { } } -function createEmptyDiagnostics(): ContactsDiagnostics { +function createIdleTaskState(): ContactsTaskState { return { - privateSessionCount: 0, - contactsEnabled: false, - skippedMissingOwnerSessions: 0, - skippedUnresolvedOwnerSessions: 0, - skippedAmbiguousPrivateSessions: 0, - skippedInvalidPlatformIdMembers: 0, - skippedFailedSessions: 0, - hiddenLowSignalNonFriends: 0, - warnings: [], + id: null, + status: 'idle', + startedAt: null, + finishedAt: null, + processedSessions: 0, + totalSessions: 0, } } -function getOrCreateAccumulator( - accumulators: Map, - sessionId: string, - meta: SessionMeta, - contact: ContactMemberRef -): ContactAccumulator { - const sessionScoped = isNameMatchPlatform(meta.platform) - const key = buildContactKey(meta.platform, contact.platformId, sessionScoped ? sessionId : undefined) - const existing = accumulators.get(key) - if (existing) { - mergeContactIdentity(existing, contact) - return existing +function requirePathProvider(deps: ContactsServiceDeps): PathProvider { + if (!deps.pathProvider) { + throw new Error('contacts worker runner requires pathProvider') } - - const created: ContactAccumulator = { - key, - platform: meta.platform, - platformId: contact.platformId, - sessionScoped, - sessionId: sessionScoped ? sessionId : undefined, - displayName: contact.name || contact.platformId, - aliases: new Set([contact.platformId, contact.name].filter(Boolean)), - avatar: contact.avatar, - isFriend: false, - privateMessageCount: 0, - activePrivateMonths: new Set(), - commonGroupSessionIds: new Set(), - coOccurrenceCount: 0, - coOccurrenceRawScore: 0, - replyInteractionCount: 0, - repliesFromOwnerToContact: 0, - repliesFromContactToOwner: 0, - sourceSessions: [], - lastInteractionTs: null, - } - accumulators.set(key, created) - return created -} - -function buildContactKey(platform: ChatPlatform, platformId: string, sessionId?: string): string { - const normalizedPlatform = platform.trim() - const normalizedPlatformId = platformId.trim() - if (!normalizedPlatform) throw new Error('platform is required') - if (!normalizedPlatformId) throw new Error('platformId is required') - return sessionId?.trim() - ? `${normalizedPlatform}:${sessionId.trim()}:${normalizedPlatformId}` - : `${normalizedPlatform}:${normalizedPlatformId}` -} - -function mergeContactIdentity(acc: ContactAccumulator, contact: ContactMemberRef): void { - if (contact.name) acc.aliases.add(contact.name) - acc.aliases.add(contact.platformId) - if ((!acc.displayName || acc.displayName === acc.platformId) && contact.name) { - acc.displayName = contact.name - } - if (!acc.avatar && contact.avatar) acc.avatar = contact.avatar -} - -function updateLastInteraction(acc: ContactAccumulator, ts: number | null): void { - if (ts === null) return - acc.lastInteractionTs = Math.max(acc.lastInteractionTs ?? 0, ts) + return deps.pathProvider } diff --git a/packages/node-runtime/src/services/contacts-signature.test.ts b/packages/node-runtime/src/services/contacts-signature.test.ts new file mode 100644 index 00000000..ba7eedc0 --- /dev/null +++ b/packages/node-runtime/src/services/contacts-signature.test.ts @@ -0,0 +1,75 @@ +/** + * Run: pnpm test -- packages/node-runtime/src/services/contacts-signature.test.ts + */ + +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 { SessionRuntimeAdapter } from './adapters' +import { CONTACTS_ALGORITHM_VERSION } from './contacts-compute' +import { buildContactsSignature } from './contacts-signature' + +function makeTempDir(): string { + const baseDir = fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir() + return fs.mkdtempSync(path.join(baseDir, 'chatlab-contacts-signature-')) +} + +function createAdapter(pathsById: Map): SessionRuntimeAdapter { + return { + listSessionIds: () => [...pathsById.keys()], + openReadonly: () => { + throw new Error('signature must not open databases') + }, + openWritable: () => { + throw new Error('signature must not open databases') + }, + closeSession: () => {}, + getDbPath: (sessionId) => pathsById.get(sessionId) ?? '', + deleteSessionFile: () => false, + ensureReadonly: () => { + throw new Error('signature must not open databases') + }, + ensureWritable: () => { + throw new Error('signature must not open databases') + }, + } +} + +test('contacts signature is stable for sorted session ids and db file versions', (t) => { + const dir = makeTempDir() + t.after(() => fs.rmSync(dir, { recursive: true, force: true })) + const first = path.join(dir, 'b.db') + const second = path.join(dir, 'a.db') + fs.writeFileSync(first, 'first') + fs.writeFileSync(second, 'second') + const adapter = createAdapter( + new Map([ + ['session-b', first], + ['session-a', second], + ]) + ) + + const signature = buildContactsSignature(adapter) + + assert.match(signature, new RegExp(`algorithm:${CONTACTS_ALGORITHM_VERSION}`)) + assert.ok(signature.indexOf('session-a:') < signature.indexOf('session-b:')) + assert.match(signature, /session-a:[^|]+/) + assert.match(signature, /session-b:[^|]+/) +}) + +test('contacts signature changes when wal file version changes', (t) => { + const dir = makeTempDir() + t.after(() => fs.rmSync(dir, { recursive: true, force: true })) + const dbPath = path.join(dir, 'session.db') + fs.writeFileSync(dbPath, 'db') + const adapter = createAdapter(new Map([['session', dbPath]])) + + const before = buildContactsSignature(adapter) + fs.writeFileSync(`${dbPath}-wal`, 'wal') + const after = buildContactsSignature(adapter) + + assert.notEqual(after, before) + assert.match(after, /session:[^|]+\|[^|]+/) +}) diff --git a/packages/node-runtime/src/services/contacts-signature.ts b/packages/node-runtime/src/services/contacts-signature.ts new file mode 100644 index 00000000..04ecbb4a --- /dev/null +++ b/packages/node-runtime/src/services/contacts-signature.ts @@ -0,0 +1,12 @@ +import { getDbFileVersion } from '../cache/analytics-cache' +import type { SessionRuntimeAdapter } from './adapters' +import { CONTACTS_ALGORITHM_VERSION } from './contacts-compute' + +export function buildContactsSignature(adapter: SessionRuntimeAdapter): string { + const parts = [`algorithm:${CONTACTS_ALGORITHM_VERSION}`] + for (const sessionId of [...adapter.listSessionIds()].sort()) { + const dbPath = adapter.getDbPath(sessionId) + parts.push(`${sessionId}:${getDbFileVersion(dbPath)}`) + } + return parts.join('|') +} diff --git a/packages/node-runtime/src/services/contacts-snapshot.test.ts b/packages/node-runtime/src/services/contacts-snapshot.test.ts new file mode 100644 index 00000000..b73d2d5c --- /dev/null +++ b/packages/node-runtime/src/services/contacts-snapshot.test.ts @@ -0,0 +1,88 @@ +/** + * Run: pnpm test -- packages/node-runtime/src/services/contacts-snapshot.test.ts + */ + +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 { ContactsDiagnostics } from '@openchatlab/shared-types' +import { CONTACTS_ALGORITHM_VERSION, type ContactsSnapshot } from './contacts-compute' +import { + cleanupContactsSnapshotTempFiles, + getContactsSnapshotPath, + readContactsSnapshot, + writeContactsSnapshot, +} from './contacts-snapshot' + +function makeTempDir(): string { + const baseDir = fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir() + return fs.mkdtempSync(path.join(baseDir, 'chatlab-contacts-snapshot-')) +} + +function emptyDiagnostics(): ContactsDiagnostics { + return { + privateSessionCount: 0, + contactsEnabled: false, + skippedMissingOwnerSessions: 0, + skippedUnresolvedOwnerSessions: 0, + skippedAmbiguousPrivateSessions: 0, + skippedInvalidPlatformIdMembers: 0, + skippedFailedSessions: 0, + hiddenLowSignalNonFriends: 0, + warnings: [], + } +} + +function makeSnapshot(signature = 'sig-1'): ContactsSnapshot { + return { + contacts: [], + diagnostics: emptyDiagnostics(), + algorithmVersion: CONTACTS_ALGORITHM_VERSION, + signature, + computedAt: 1234, + workerStats: { + durationMs: 10, + totalSessions: 0, + processedSessions: 0, + skippedFailedSessions: 0, + }, + } +} + +test('contacts snapshot reads missing file as null and writes atomically readable json', (t) => { + const dir = makeTempDir() + t.after(() => fs.rmSync(dir, { recursive: true, force: true })) + + assert.equal(readContactsSnapshot(dir), null) + + writeContactsSnapshot(dir, makeSnapshot()) + + assert.deepEqual(readContactsSnapshot(dir), makeSnapshot()) + assert.ok(fs.existsSync(getContactsSnapshotPath(dir))) + assert.equal(fs.readdirSync(dir).filter((name) => name.startsWith('contacts-snapshot.tmp-')).length, 0) +}) + +test('contacts snapshot backs up corrupt json and returns null', (t) => { + const dir = makeTempDir() + t.after(() => fs.rmSync(dir, { recursive: true, force: true })) + fs.writeFileSync(getContactsSnapshotPath(dir), '{ broken') + + assert.equal(readContactsSnapshot(dir, { now: () => 5678 }), null) + + assert.equal(fs.existsSync(getContactsSnapshotPath(dir)), false) + assert.ok(fs.existsSync(path.join(dir, 'contacts-snapshot.corrupt-5678.json'))) +}) + +test('contacts snapshot cleanup removes stale temp files only', (t) => { + const dir = makeTempDir() + t.after(() => fs.rmSync(dir, { recursive: true, force: true })) + fs.writeFileSync(path.join(dir, 'contacts-snapshot.tmp-old'), 'tmp') + fs.writeFileSync(path.join(dir, 'contacts-snapshot.json'), '{}') + + cleanupContactsSnapshotTempFiles(dir) + + assert.equal(fs.existsSync(path.join(dir, 'contacts-snapshot.tmp-old')), false) + assert.equal(fs.existsSync(path.join(dir, 'contacts-snapshot.json')), true) +}) diff --git a/packages/node-runtime/src/services/contacts-snapshot.ts b/packages/node-runtime/src/services/contacts-snapshot.ts new file mode 100644 index 00000000..91baa1f8 --- /dev/null +++ b/packages/node-runtime/src/services/contacts-snapshot.ts @@ -0,0 +1,56 @@ +import fs from 'node:fs' +import path from 'node:path' +import { appLogger } from '../logging/app-logger' +import type { ContactsSnapshot } from './contacts-compute' + +const CONTACTS_SNAPSHOT_FILE = 'contacts-snapshot.json' +const CONTACTS_SNAPSHOT_TMP_PREFIX = 'contacts-snapshot.tmp-' + +export interface ReadContactsSnapshotOptions { + now?: () => number +} + +export function getContactsSnapshotPath(systemDir: string): string { + return path.join(systemDir, CONTACTS_SNAPSHOT_FILE) +} + +export function readContactsSnapshot( + systemDir: string, + options: ReadContactsSnapshotOptions = {} +): ContactsSnapshot | null { + const snapshotPath = getContactsSnapshotPath(systemDir) + if (!fs.existsSync(snapshotPath)) return null + + try { + return JSON.parse(fs.readFileSync(snapshotPath, 'utf-8')) as ContactsSnapshot + } catch (error) { + const ts = options.now?.() ?? Date.now() + const backupPath = path.join(systemDir, `contacts-snapshot.corrupt-${ts}.json`) + try { + fs.renameSync(snapshotPath, backupPath) + } catch (renameError) { + appLogger.warn('contacts', 'failed to backup corrupt contacts snapshot', renameError) + } + appLogger.warn('contacts', 'contacts snapshot is corrupt', error) + return null + } +} + +export function writeContactsSnapshot(systemDir: string, snapshot: ContactsSnapshot): void { + if (!fs.existsSync(systemDir)) fs.mkdirSync(systemDir, { recursive: true }) + const tmpPath = path.join(systemDir, `${CONTACTS_SNAPSHOT_TMP_PREFIX}${process.pid}-${Date.now()}`) + fs.writeFileSync(tmpPath, JSON.stringify(snapshot, null, 2), 'utf-8') + fs.renameSync(tmpPath, getContactsSnapshotPath(systemDir)) +} + +export function cleanupContactsSnapshotTempFiles(systemDir: string): void { + if (!fs.existsSync(systemDir)) return + for (const name of fs.readdirSync(systemDir)) { + if (!name.startsWith(CONTACTS_SNAPSHOT_TMP_PREFIX)) continue + try { + fs.rmSync(path.join(systemDir, name), { force: true }) + } catch (error) { + appLogger.warn('contacts', 'failed to remove contacts snapshot temp file', error) + } + } +} diff --git a/packages/node-runtime/src/services/contacts-worker-entry.ts b/packages/node-runtime/src/services/contacts-worker-entry.ts new file mode 100644 index 00000000..dc7d0188 --- /dev/null +++ b/packages/node-runtime/src/services/contacts-worker-entry.ts @@ -0,0 +1,37 @@ +import { parentPort, workerData } from 'node:worker_threads' +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 { computeContactsSnapshot, type ContactsComputeProgress } from './contacts-compute' + +interface ContactsWorkerStartupOptions { + paths: StaticPathProviderSnapshot + runtimeIdentity?: RuntimeIdentity + nativeBinding?: string + signature: string +} + +async function main(): Promise { + if (!parentPort) throw new Error('contacts worker requires parentPort') + const options = workerData as ContactsWorkerStartupOptions + 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 onProgress = (progress: ContactsComputeProgress) => parentPort?.postMessage({ type: 'progress', progress }) + const snapshot = computeContactsSnapshot({ adapter, signature: options.signature, onProgress }) + 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/contacts-worker-runner.ts b/packages/node-runtime/src/services/contacts-worker-runner.ts new file mode 100644 index 00000000..d3e7fd86 --- /dev/null +++ b/packages/node-runtime/src/services/contacts-worker-runner.ts @@ -0,0 +1,96 @@ +import { Worker } from 'node:worker_threads' +import type { WorkerOptions } from 'node:worker_threads' +import type { PathProvider } from '@openchatlab/core' +import type { RuntimeIdentity } from '../data-dir-compat' +import { snapshotPathProvider } from '../semantic-index/static-path-provider' +import type { ContactsComputeProgress, ContactsSnapshot } from './contacts-compute' +import type { ContactsComputeRunner } from './contacts-service' + +export interface ContactsWorkerRunnerOptions { + pathProvider: PathProvider + runtimeIdentity?: RuntimeIdentity + nativeBinding?: string + workerEntryUrl?: string | URL +} + +interface ContactsWorkerMessage { + type: 'progress' | 'success' | 'error' + progress?: ContactsComputeProgress + snapshot?: ContactsSnapshot + error?: string +} + +type ModuleWorkerOptions = WorkerOptions & { type: 'module' } + +function defaultWorkerEntryUrl(): URL { + return import.meta.url.endsWith('.ts') + ? new URL('./contacts-worker-entry.ts', import.meta.url) + : new URL('./contacts-worker-entry.js', import.meta.url) +} + +function normalizeWorkerEntryUrl(entryUrl?: string | URL): URL { + if (!entryUrl) return defaultWorkerEntryUrl() + return typeof entryUrl === 'string' ? new URL(entryUrl) : entryUrl +} + +function createWorker(workerData: unknown, entryUrlInput?: string | URL): Worker { + const entryUrl = normalizeWorkerEntryUrl(entryUrlInput) + 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 options: ModuleWorkerOptions = { + eval: true, + type: 'module', + workerData, + execArgv: [], + } + return new Worker(bootstrap, options) +} + +export function createContactsWorkerRunner(options: ContactsWorkerRunnerOptions): ContactsComputeRunner { + return ({ signature, onProgress }) => + new Promise((resolve, reject) => { + const worker = createWorker( + { + paths: snapshotPathProvider(options.pathProvider), + runtimeIdentity: options.runtimeIdentity, + nativeBinding: options.nativeBinding, + signature, + }, + options.workerEntryUrl + ) + let settled = false + + worker.on('message', (message: ContactsWorkerMessage) => { + if (message.type === 'progress' && message.progress) { + onProgress(message.progress) + return + } + if (message.type === 'success' && message.snapshot) { + settled = true + resolve(message.snapshot) + void worker.terminate() + return + } + if (message.type === 'error') { + settled = true + reject(new Error(message.error ?? 'contacts worker failed')) + void worker.terminate() + } + }) + worker.on('error', (error) => { + if (settled) return + settled = true + reject(error) + }) + worker.on('exit', (code) => { + if (settled || code === 0) return + settled = true + reject(new Error(`contacts worker exited with code ${code}`)) + }) + }) +} diff --git a/packages/node-runtime/src/services/index.ts b/packages/node-runtime/src/services/index.ts index ca789f74..95c67003 100644 --- a/packages/node-runtime/src/services/index.ts +++ b/packages/node-runtime/src/services/index.ts @@ -56,8 +56,14 @@ export type { LlmConfig, SummaryServiceDeps } from './summary-service' export { exportMarkdown } from './export-service' // Contacts service -export { CONTACTS_ALGORITHM_VERSION, createContactsService } from './contacts-service' -export type { ContactsService, ContactsServiceDeps, ContactsServiceOptions } from './contacts-service' +export { CONTACTS_ALGORITHM_VERSION } from './contacts-compute' +export { createContactsService } from './contacts-service' +export type { + ContactsComputeRunner, + ContactsService, + ContactsServiceDeps, + ContactsServiceOptions, +} from './contacts-service' // Merge cache export { MergeSessionCache } from './merge-cache' diff --git a/packages/shared-types/index.ts b/packages/shared-types/index.ts index bc38e609..0691cfff 100644 --- a/packages/shared-types/index.ts +++ b/packages/shared-types/index.ts @@ -232,6 +232,8 @@ export type ContactPool = 'friend' | 'non_friend' export type ContactsCacheStatus = 'fresh' | 'stale' | 'missing' +export type ContactsTaskStatus = 'idle' | 'running' | 'succeeded' | 'failed' | 'superseded' + export interface ContactScoreBreakdown { privateMessageScore?: number privateRegularityScore?: number @@ -302,11 +304,23 @@ export interface ContactsCacheState { staleReason?: string } +export interface ContactsTaskState { + id: string | null + status: ContactsTaskStatus + startedAt: number | null + finishedAt: number | null + processedSessions: number + totalSessions: number + currentSessionId?: string + lastError?: string +} + export interface ContactsResponse { contacts: ContactItem[] diagnostics: ContactsDiagnostics cache: ContactsCacheState algorithmVersion: string + task?: ContactsTaskState } export interface Preferences { diff --git a/src/i18n/locales/en-US/contacts.json b/src/i18n/locales/en-US/contacts.json index f009a1fc..1aed4c3c 100644 --- a/src/i18n/locales/en-US/contacts.json +++ b/src/i18n/locales/en-US/contacts.json @@ -67,12 +67,19 @@ }, "stale": { "title": "Contacts may be stale", - "message": "Chat data changed. Recompute contacts now?", + "message": "Chat data changed. Contacts are updating in the background.", "inline": "Contacts may be stale.", "later": "Later" }, + "task": { + "running": "Computing contacts: {current} / {total} sessions", + "updating": "Contacts may be stale. Updating in the background.", + "failed": "Contact computation failed", + "retry": "Retry" + }, "toast": { "recomputed": "Contacts recomputed", + "recomputeStarted": "Started computing contacts in the background", "recomputeFailed": "Recompute failed" } } diff --git a/src/i18n/locales/ja-JP/contacts.json b/src/i18n/locales/ja-JP/contacts.json index 196eeab3..56212d15 100644 --- a/src/i18n/locales/ja-JP/contacts.json +++ b/src/i18n/locales/ja-JP/contacts.json @@ -67,12 +67,19 @@ }, "stale": { "title": "連絡先データが古い可能性があります", - "message": "チャットデータが変更されました。今すぐ再計算しますか？", + "message": "チャットデータが変更されました。バックグラウンドで連絡先を更新しています。", "inline": "連絡先データが古い可能性があります。", "later": "後で" }, + "task": { + "running": "連絡先を計算中: {current} / {total} セッション", + "updating": "連絡先データが古い可能性があります。バックグラウンドで更新しています。", + "failed": "連絡先の計算に失敗しました", + "retry": "再試行" + }, "toast": { "recomputed": "連絡先を再計算しました", + "recomputeStarted": "バックグラウンドで連絡先の計算を開始しました", "recomputeFailed": "再計算に失敗しました" } } diff --git a/src/i18n/locales/zh-CN/contacts.json b/src/i18n/locales/zh-CN/contacts.json index 6f213c12..5ff2bdfc 100644 --- a/src/i18n/locales/zh-CN/contacts.json +++ b/src/i18n/locales/zh-CN/contacts.json @@ -67,12 +67,19 @@ }, "stale": { "title": "联系人数据可能已过期", - "message": "检测到聊天库发生变化，是否立即重新计算联系人？", + "message": "检测到聊天库发生变化，正在后台更新联系人。", "inline": "联系人数据可能已过期。", "later": "稍后" }, + "task": { + "running": "正在计算联系人：{current} / {total} 个会话", + "updating": "联系人数据可能已过期，正在后台更新。", + "failed": "联系人计算失败", + "retry": "重试" + }, "toast": { "recomputed": "联系人已重新计算", + "recomputeStarted": "已开始后台计算联系人", "recomputeFailed": "重新计算失败" } } diff --git a/src/i18n/locales/zh-TW/contacts.json b/src/i18n/locales/zh-TW/contacts.json index bd8f0dfb..b3997f7f 100644 --- a/src/i18n/locales/zh-TW/contacts.json +++ b/src/i18n/locales/zh-TW/contacts.json @@ -67,12 +67,19 @@ }, "stale": { "title": "聯絡人資料可能已過期", - "message": "偵測到聊天庫發生變化，是否立即重新計算聯絡人？", + "message": "偵測到聊天庫發生變化，正在背景更新聯絡人。", "inline": "聯絡人資料可能已過期。", "later": "稍後" }, + "task": { + "running": "正在計算聯絡人：{current} / {total} 個會話", + "updating": "聯絡人資料可能已過期，正在背景更新。", + "failed": "聯絡人計算失敗", + "retry": "重試" + }, "toast": { "recomputed": "聯絡人已重新計算", + "recomputeStarted": "已開始在背景計算聯絡人", "recomputeFailed": "重新計算失敗" } } diff --git a/src/pages/contacts/index.vue b/src/pages/contacts/index.vue index 28de71c8..d348f483 100644 --- a/src/pages/contacts/index.vue +++ b/src/pages/contacts/index.vue @@ -1,5 +1,5 @@