From 9c8faeba09bd426519339435035ac11704b794dd Mon Sep 17 00:00:00 2001 From: digua Date: Fri, 26 Jun 2026 18:27:39 +0800 Subject: [PATCH] feat(node-runtime): add contacts aggregation service --- packages/node-runtime/src/index.ts | 17 +- .../src/services/contacts-overrides.test.ts | 103 ++++ .../src/services/contacts-overrides.ts | 158 +++++++ .../src/services/contacts-service.test.ts | 365 ++++++++++++++ .../src/services/contacts-service.ts | 445 ++++++++++++++++++ packages/node-runtime/src/services/index.ts | 8 + 6 files changed, 1095 insertions(+), 1 deletion(-) create mode 100644 packages/node-runtime/src/services/contacts-overrides.test.ts create mode 100644 packages/node-runtime/src/services/contacts-overrides.ts create mode 100644 packages/node-runtime/src/services/contacts-service.test.ts create mode 100644 packages/node-runtime/src/services/contacts-service.ts diff --git a/packages/node-runtime/src/index.ts b/packages/node-runtime/src/index.ts index 8d517d98..599c94d5 100644 --- a/packages/node-runtime/src/index.ts +++ b/packages/node-runtime/src/index.ts @@ -364,6 +364,8 @@ export * as sessionIndexService from './services/session-index-service' export * as summaryService from './services/summary-service' export * as exportService from './services/export-service' export * as ownerProfileService from './services/owner-profile-service' +export * as contactsService from './services/contacts-service' +export * as contactsOverrides from './services/contacts-overrides' // Semantic index (Phase 1 vector search) — independent of legacy ai/rag export * as semanticIndex from './semantic-index' export { @@ -391,7 +393,15 @@ export type { SemanticIndexMode, } from './semantic-index' -export { createDatabaseManagerAdapter, MergeSessionCache, pushImport } from './services' +export { + CONTACTS_ALGORITHM_VERSION, + ContactsOverridesManager, + MergeSessionCache, + buildContactOverrideKey, + createContactsService, + createDatabaseManagerAdapter, + pushImport, +} from './services' export type { SessionRuntimeAdapter, AnalysisSessionDTO, @@ -402,6 +412,11 @@ export type { ApplyOwnerProfileReason, ApplyOwnerProfileResult, SetOwnerAndApplyProfileResult, + BuildContactOverrideKeyInput, + ContactsOverrides, + ContactsService, + ContactsServiceDeps, + ContactsServiceOptions, PushImportPayload, PushImportResult, PushImportOutcome, diff --git a/packages/node-runtime/src/services/contacts-overrides.test.ts b/packages/node-runtime/src/services/contacts-overrides.test.ts new file mode 100644 index 00000000..cfa3526f --- /dev/null +++ b/packages/node-runtime/src/services/contacts-overrides.test.ts @@ -0,0 +1,103 @@ +/** + * Tests for contacts override persistence. + * + * Run: pnpm test -- packages/node-runtime/src/services/contacts-overrides.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 { initAppLogger } from '../logging/app-logger' +import { ContactsOverridesManager, buildContactOverrideKey } from './contacts-overrides' + +function makeTempDir(): string { + const baseDir = fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir() + return fs.mkdtempSync(path.join(baseDir, 'chatlab-contacts-overrides-')) +} + +function readJson(filePath: string): unknown { + return JSON.parse(fs.readFileSync(filePath, 'utf-8')) +} + +test('loads empty overrides when contacts-overrides.json is missing', (t) => { + const dir = makeTempDir() + t.after(() => fs.rmSync(dir, { recursive: true, force: true })) + + const manager = new ContactsOverridesManager(dir) + + assert.equal(manager.getFilePath(), path.join(dir, 'contacts-overrides.json')) + assert.deepEqual(manager.load(), {}) + assert.equal(manager.getSignaturePart(), '-') +}) + +test('saveOverride writes valid JSON and deleteOverride removes entries', (t) => { + const dir = makeTempDir() + t.after(() => fs.rmSync(dir, { recursive: true, force: true })) + const manager = new ContactsOverridesManager(dir) + + const saved = manager.saveOverride('weixin:wxid_a', { lockedTier: 'core' }) + assert.equal(saved.lockedTier, 'core') + assert.equal(typeof saved.updatedAt, 'number') + + assert.deepEqual(readJson(manager.getFilePath()), { + 'weixin:wxid_a': saved, + }) + + manager.deleteOverride('weixin:wxid_a') + assert.deepEqual(readJson(manager.getFilePath()), {}) +}) + +test('buildContactOverrideKey uses stable ids and session scope for name-match platforms', () => { + assert.equal(buildContactOverrideKey({ platform: 'weixin', platformId: 'wxid_a' }), 'weixin:wxid_a') + assert.equal( + buildContactOverrideKey({ platform: 'whatsapp', platformId: 'Alice', sessionId: 'session-1', matchMode: 'name' }), + 'whatsapp:session-1:Alice' + ) + assert.equal( + buildContactOverrideKey({ platform: 'line', platformId: 'Bob', sessionId: 'line-session' }), + 'line:line-session:Bob' + ) + assert.throws(() => buildContactOverrideKey({ platform: 'instagram', platformId: 'Carol' }), /sessionId is required/) +}) + +test('corrupt override file is backed up and loads empty overrides', (t) => { + const dir = makeTempDir() + t.after(() => fs.rmSync(dir, { recursive: true, force: true })) + initAppLogger(path.join(dir, 'logs')) + const filePath = path.join(dir, 'contacts-overrides.json') + fs.writeFileSync(filePath, '{ broken', 'utf-8') + + const manager = new ContactsOverridesManager(dir) + assert.deepEqual(manager.load(), {}) + + const backupsDir = path.join(dir, 'backups') + const backups = fs.readdirSync(backupsDir).filter((name) => name.startsWith('contacts-overrides-corrupt-')) + assert.equal(backups.length, 1) + assert.equal(fs.readFileSync(path.join(backupsDir, backups[0]), 'utf-8'), '{ broken') + + const log = fs.readFileSync(path.join(dir, 'logs', 'app.log'), 'utf-8') + assert.match(log, /contacts overrides file is corrupt/) +}) + +test('atomic write leaves complete JSON and no temp file behind', (t) => { + const dir = makeTempDir() + t.after(() => fs.rmSync(dir, { recursive: true, force: true })) + const manager = new ContactsOverridesManager(dir) + + manager.saveOverride('qq:10086', { lockedTier: 'core' }) + manager.saveOverride('qq:10010', { lockedTier: 'friend' }) + + const raw = fs.readFileSync(manager.getFilePath(), 'utf-8') + assert.doesNotThrow(() => JSON.parse(raw)) + assert.deepEqual(Object.keys(readJson(manager.getFilePath()) as Record).sort(), [ + 'qq:10010', + 'qq:10086', + ]) + assert.deepEqual( + fs.readdirSync(dir).filter((name) => name.includes('contacts-overrides') && name.endsWith('.tmp')), + [] + ) + assert.notEqual(manager.getSignaturePart(), '-') +}) diff --git a/packages/node-runtime/src/services/contacts-overrides.ts b/packages/node-runtime/src/services/contacts-overrides.ts new file mode 100644 index 00000000..746d2cea --- /dev/null +++ b/packages/node-runtime/src/services/contacts-overrides.ts @@ -0,0 +1,158 @@ +import fs from 'node:fs' +import path from 'node:path' +import { isNameMatchPlatform } from '@openchatlab/core' +import type { + ChatPlatform, + ContactOverride, + ContactOverridePatch, + ContactTier, + OwnerMatchMode, +} from '@openchatlab/shared-types' +import { appLogger } from '../logging/app-logger' + +export type ContactsOverrides = Record + +export interface BuildContactOverrideKeyInput { + platform: ChatPlatform + platformId: string + sessionId?: string + matchMode?: OwnerMatchMode +} + +const OVERRIDES_FILE = 'contacts-overrides.json' +const VALID_CONTACT_TIERS: ReadonlySet = new Set([ + 'core', + 'friend', + 'acquaintance', + 'high_interaction', + 'medium_interaction', + 'low_interaction', +]) + +export function buildContactOverrideKey(input: BuildContactOverrideKeyInput): string { + const platform = input.platform.trim() + const platformId = input.platformId.trim() + if (!platform) throw new Error('platform is required') + if (!platformId) throw new Error('platformId is required') + + const sessionScoped = input.matchMode === 'name' || isNameMatchPlatform(platform) + if (!sessionScoped) return `${platform}:${platformId}` + + const sessionId = input.sessionId?.trim() + if (!sessionId) throw new Error('sessionId is required for session-scoped contact override keys') + return `${platform}:${sessionId}:${platformId}` +} + +export class ContactsOverridesManager { + private readonly filePath: string + + constructor(systemDir: string) { + this.filePath = path.join(systemDir, OVERRIDES_FILE) + } + + getFilePath(): string { + return this.filePath + } + + load(): ContactsOverrides { + if (!fs.existsSync(this.filePath)) return {} + + let raw = '' + try { + raw = fs.readFileSync(this.filePath, 'utf-8') + return normalizeOverrides(JSON.parse(raw)) + } catch (error) { + this.backupCorruptFile(raw) + appLogger.warn('contacts', 'contacts overrides file is corrupt', error) + return {} + } + } + + saveOverride(key: string, patch: ContactOverridePatch): ContactOverride { + const overrides = this.load() + const current = overrides[key] ?? {} + const next = normalizeOverride({ + ...current, + ...patch, + updatedAt: Date.now(), + }) + + overrides[key] = next + this.writeOverrides(overrides) + return next + } + + deleteOverride(key: string): void { + const overrides = this.load() + delete overrides[key] + this.writeOverrides(overrides) + } + + getSignaturePart(): string { + try { + const stat = fs.statSync(this.filePath) + return `${Math.floor(stat.mtimeMs)}:${stat.size}` + } catch { + return '-' + } + } + + private writeOverrides(overrides: ContactsOverrides): void { + const dir = path.dirname(this.filePath) + if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true }) + + const tempPath = `${this.filePath}.${process.pid}.${Date.now()}.tmp` + try { + fs.writeFileSync(tempPath, `${JSON.stringify(overrides, null, 2)}\n`, 'utf-8') + fs.renameSync(tempPath, this.filePath) + } catch (error) { + try { + fs.rmSync(tempPath, { force: true }) + } catch { + // ignore cleanup failures; the original write error is the actionable one + } + throw error + } + } + + private backupCorruptFile(raw: string): void { + try { + const backupDir = path.join(path.dirname(this.filePath), 'backups') + if (!fs.existsSync(backupDir)) fs.mkdirSync(backupDir, { recursive: true }) + const timestamp = new Date().toISOString().replace(/[:.]/g, '-') + const backupPath = path.join(backupDir, `contacts-overrides-corrupt-${timestamp}.json`) + fs.writeFileSync(backupPath, raw, 'utf-8') + } catch (error) { + appLogger.warn('contacts', 'failed to back up corrupt contacts overrides file', error) + } + } +} + +function normalizeOverrides(value: unknown): ContactsOverrides { + if (!value || typeof value !== 'object' || Array.isArray(value)) return {} + + const result: ContactsOverrides = {} + for (const [key, override] of Object.entries(value as Record)) { + if (!key.trim()) continue + if (!override || typeof override !== 'object' || Array.isArray(override)) continue + result[key] = normalizeOverride(override as ContactOverride) + } + return result +} + +function normalizeOverride(override: ContactOverride): ContactOverride { + const normalized: ContactOverride = {} + + if (override.lockedTier === null || isValidContactTier(override.lockedTier)) { + normalized.lockedTier = override.lockedTier + } + if (typeof override.updatedAt === 'number' && Number.isFinite(override.updatedAt)) { + normalized.updatedAt = override.updatedAt + } + + return normalized +} + +function isValidContactTier(value: unknown): value is ContactTier { + return typeof value === 'string' && VALID_CONTACT_TIERS.has(value as ContactTier) +} diff --git a/packages/node-runtime/src/services/contacts-service.test.ts b/packages/node-runtime/src/services/contacts-service.test.ts new file mode 100644 index 00000000..a76d9e68 --- /dev/null +++ b/packages/node-runtime/src/services/contacts-service.test.ts @@ -0,0 +1,365 @@ +/** + * Integration tests for the cross-session contacts service. + * + * Run: pnpm test -- packages/node-runtime/src/services/contacts-service.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 { CHAT_DB_SCHEMA } from '@openchatlab/core' +import type { DatabaseAdapter } from '@openchatlab/core' +import { openBetterSqliteDatabase } from '../better-sqlite3-adapter' +import type { SessionRuntimeAdapter } from './adapters' +import { createContactsService } from './contacts-service' + +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-service-')) +} + +interface SeedMember { + id: number + platformId: string + accountName?: string + groupNickname?: string + avatar?: string | null +} + +interface SeedMessage { + id: number + senderId: number + ts: number + content?: string + platformMessageId?: string | null + replyToMessageId?: string | null +} + +interface SeedSession { + id: string + platform: string + type: 'private' | 'group' + ownerId?: string | null + members: SeedMember[] + messages?: SeedMessage[] +} + +class TestEnv { + readonly dir = makeTempDir() + readonly adapter: SessionRuntimeAdapter + private dbPaths = new Map() + private openDbs: DatabaseAdapter[] = [] + + constructor() { + const open = (sessionId: string, readonly: boolean): DatabaseAdapter | null => { + const dbPath = this.dbPaths.get(sessionId) + if (!dbPath) return null + const db = openBetterSqliteDatabase(dbPath, { readonly, nativeBinding }) + this.openDbs.push(db) + return db + } + + this.adapter = { + listSessionIds: () => [...this.dbPaths.keys()], + openReadonly: (id) => open(id, true), + openWritable: (id) => open(id, false), + closeSession: () => {}, + getDbPath: (id) => this.dbPaths.get(id) ?? '', + deleteSessionFile: () => false, + ensureReadonly: (id) => { + const db = open(id, true) + if (!db) throw Object.assign(new Error(`Session not found: ${id}`), { statusCode: 404 }) + return db + }, + ensureWritable: (id) => { + const db = open(id, false) + if (!db) throw Object.assign(new Error(`Session not found: ${id}`), { statusCode: 404 }) + return db + }, + } + } + + seed(session: SeedSession): void { + const dbPath = path.join(this.dir, `${session.id}.db`) + const db = openBetterSqliteDatabase(dbPath, { nativeBinding }) + db.exec(CHAT_DB_SCHEMA) + db.prepare(`INSERT INTO meta (name, platform, type, imported_at, owner_id) VALUES (?, ?, ?, ?, ?)`).run( + session.id, + session.platform, + session.type, + 1780000000, + session.ownerId ?? null + ) + for (const member of session.members) { + db.prepare( + `INSERT INTO member (id, platform_id, account_name, group_nickname, avatar) VALUES (?, ?, ?, ?, ?)` + ).run( + member.id, + member.platformId, + member.accountName ?? member.platformId, + member.groupNickname ?? null, + member.avatar ?? null + ) + } + for (const message of session.messages ?? []) { + db.prepare( + `INSERT INTO message + (id, sender_id, ts, type, content, platform_message_id, reply_to_message_id) + VALUES (?, ?, ?, 0, ?, ?, ?)` + ).run( + message.id, + message.senderId, + message.ts, + message.content ?? `message ${message.id}`, + message.platformMessageId ?? `m-${message.id}`, + message.replyToMessageId ?? null + ) + } + db.close() + this.dbPaths.set(session.id, dbPath) + } + + dbPath(sessionId: string): string { + const dbPath = this.dbPaths.get(sessionId) + assert.ok(dbPath) + return dbPath + } + + cleanup(): void { + for (const db of this.openDbs) { + try { + db.close() + } catch { + // already closed + } + } + fs.rmSync(this.dir, { recursive: true, force: true }) + } +} + +function privateMessages(count: number, startId: number, startTs: number): SeedMessage[] { + return Array.from({ length: count }, (_, index) => ({ + id: startId + index, + senderId: index % 2 === 0 ? 1 : 2, + ts: startTs + index, + })) +} + +test('aggregates stable-id contacts across private and group sessions', (t) => { + const env = new TestEnv() + t.after(() => env.cleanup()) + + env.seed({ + id: 'private-a', + platform: 'weixin', + type: 'private', + ownerId: 'owner', + members: [ + { id: 1, platformId: 'owner', accountName: 'Me' }, + { id: 2, platformId: 'alice', accountName: 'Alice', avatar: 'alice.png' }, + ], + messages: privateMessages(60, 1, 1704103200), + }) + env.seed({ + id: 'private-b', + platform: 'weixin', + type: 'private', + ownerId: 'owner', + members: [ + { id: 1, platformId: 'owner', accountName: 'Me' }, + { id: 2, platformId: 'alice', accountName: 'Alice B' }, + ], + messages: privateMessages(5, 1, 1706781600), + }) + env.seed({ + id: 'group-a', + platform: 'weixin', + type: 'group', + ownerId: 'owner', + members: [ + { id: 1, platformId: 'owner', accountName: 'Me' }, + { id: 2, platformId: 'alice', accountName: 'Alice' }, + { id: 3, platformId: 'bob', accountName: 'Bob' }, + ], + messages: [ + { id: 1, senderId: 1, ts: 1704103200, platformMessageId: 'owner-1' }, + { id: 2, senderId: 2, ts: 1704103201, platformMessageId: 'alice-1', replyToMessageId: 'owner-1' }, + { id: 3, senderId: 3, ts: 1704103800, platformMessageId: 'bob-1' }, + ], + }) + + const result = createContactsService({ adapter: env.adapter, systemDir: env.dir }).getContacts() + const byKey = new Map(result.contacts.map((contact) => [contact.key, contact])) + const alice = byKey.get('weixin:alice') + const bob = byKey.get('weixin:bob') + + assert.ok(alice) + assert.equal(alice.isFriend, true) + assert.equal(alice.pool, 'friend') + assert.equal(alice.scoreBreakdown.privateMessageCount, 65) + assert.equal(alice.scoreBreakdown.activePrivateMonths, 2) + assert.equal(alice.scoreBreakdown.commonGroupCount, 1) + assert.equal(alice.avatar, 'alice.png') + assert.deepEqual(alice.sourceSessions.map((source) => source.id).sort(), ['group-a', 'private-a', 'private-b']) + + assert.ok(bob) + assert.equal(bob.isFriend, false) + assert.equal(bob.pool, 'non_friend') + assert.equal(bob.scoreBreakdown.commonGroupCount, 1) + assert.equal(bob.sourceSessions.length, 1) + + 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) => { + const env = new TestEnv() + t.after(() => env.cleanup()) + + env.seed({ + id: 'missing-owner', + platform: 'weixin', + type: 'private', + ownerId: null, + members: [{ id: 1, platformId: 'alice' }], + }) + env.seed({ + id: 'unresolved-owner', + platform: 'weixin', + type: 'private', + ownerId: 'owner', + members: [{ id: 1, platformId: 'alice' }], + }) + env.seed({ + id: 'ambiguous-private', + platform: 'weixin', + type: 'private', + ownerId: 'owner', + members: [ + { id: 1, platformId: 'owner' }, + { id: 2, platformId: 'alice' }, + { id: 3, platformId: 'bob' }, + ], + }) + + const result = createContactsService({ adapter: env.adapter, systemDir: env.dir }).getContacts() + + assert.equal(result.contacts.length, 0) + assert.equal(result.diagnostics.privateSessionCount, 3) + assert.equal(result.diagnostics.skippedMissingOwnerSessions, 1) + assert.equal(result.diagnostics.skippedUnresolvedOwnerSessions, 1) + assert.equal(result.diagnostics.skippedAmbiguousPrivateSessions, 1) +}) + +test('keeps name-match platform contacts session-scoped and applies manual overrides only to existing contacts', (t) => { + const env = new TestEnv() + t.after(() => env.cleanup()) + + for (const id of ['whatsapp-a', 'whatsapp-b']) { + env.seed({ + id, + platform: 'whatsapp', + type: 'private', + ownerId: 'Me', + members: [ + { id: 1, platformId: 'Me' }, + { id: 2, platformId: 'Alice' }, + ], + messages: privateMessages(10, 1, 1704103200), + }) + } + + const service = createContactsService({ adapter: env.adapter, systemDir: env.dir }) + service.setContactOverride('whatsapp:whatsapp-a:Alice', { lockedTier: 'core' }) + service.setContactOverride('whatsapp:ghost:Alice', { lockedTier: 'core' }) + const result = service.getContacts() + const keys = result.contacts.map((contact) => contact.key).sort() + + assert.deepEqual(keys, ['whatsapp:whatsapp-a:Alice', 'whatsapp:whatsapp-b:Alice']) + const locked = result.contacts.find((contact) => contact.key === 'whatsapp:whatsapp-a:Alice') + const unlocked = result.contacts.find((contact) => contact.key === 'whatsapp:whatsapp-b:Alice') + assert.ok(locked) + assert.equal(locked.tier, 'core') + assert.equal(locked.lockedTier, 'core') + assert.ok(unlocked) + assert.equal(unlocked.lockedTier, null) +}) + +test('returns stale cached contacts when signature changes and acceptStale is true', (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 service = createContactsService({ adapter: env.adapter, systemDir: env.dir, now: () => now }) + const first = service.getContacts() + assert.equal(first.cache.status, 'fresh') + assert.equal(first.cache.computedAt, 1000) + + 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 fresh = service.getContacts() + assert.equal(fresh.cache.status, 'fresh') + assert.equal(fresh.cache.computedAt, 2000) +}) + +test('recomputes contacts after owner_id changes in a session database', (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 service = createContactsService({ adapter: env.adapter, systemDir: env.dir, now: () => now }) + const first = service.getContacts() + assert.deepEqual( + first.contacts.map((contact) => contact.key), + ['weixin:alice'] + ) + + now = 2000 + env.adapter.ensureWritable('private-a').prepare('UPDATE meta SET owner_id = ?').run('alice') + + 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'] + ) +}) diff --git a/packages/node-runtime/src/services/contacts-service.ts b/packages/node-runtime/src/services/contacts-service.ts new file mode 100644 index 00000000..e4dc0819 --- /dev/null +++ b/packages/node-runtime/src/services/contacts-service.ts @@ -0,0 +1,445 @@ +import { ChatType } from '@openchatlab/shared-types' +import type { + ChatPlatform, + ContactItem, + ContactOverridePatch, + ContactsCacheState, + ContactsDiagnostics, + ContactsResponse, + ContactSourceSession, + ContactTier, +} from '@openchatlab/shared-types' +import { + MIN_PRIVATE_SESSIONS_FOR_CONTACTS, + applyContactOverride, + assignFriendTiers, + assignNonFriendTiers, + computeFriendScores, + computeNonFriendScores, + getGroupContactFacts, + getPrivateContactFacts, + getSessionMeta, + isChatSessionDb, + isNameMatchPlatform, + resolveOwnerMember, +} from '@openchatlab/core' +import type { ContactMemberRef, SessionMeta } from '@openchatlab/core' +import { getDbFileVersion } from '../cache/analytics-cache' +import { appLogger } from '../logging/app-logger' +import type { SessionRuntimeAdapter } from './adapters' +import { ContactsOverridesManager, buildContactOverrideKey } from './contacts-overrides' + +export const CONTACTS_ALGORITHM_VERSION = 'contacts-v1' + +export interface ContactsServiceOptions { + forceRecompute?: boolean + acceptStale?: boolean +} + +export interface ContactsServiceDeps { + adapter: SessionRuntimeAdapter + systemDir: string + now?: () => number +} + +export interface ContactsService { + getContacts(options?: ContactsServiceOptions): ContactsResponse + setContactOverride(key: string, patch: ContactOverridePatch): void + deleteContactOverride(key: string): void + invalidateContactsCache(): void +} + +interface CachedContacts { + contacts: ContactItem[] + diagnostics: ContactsDiagnostics + algorithmVersion: 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 +} + +export function createContactsService(deps: ContactsServiceDeps): ContactsService { + return new DefaultContactsService(deps) +} + +class DefaultContactsService implements ContactsService { + private readonly overrides: ContactsOverridesManager + private cache: CachedContacts | null = null + + constructor(private readonly deps: ContactsServiceDeps) { + this.overrides = new ContactsOverridesManager(deps.systemDir) + } + + 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 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') + } + + setContactOverride(key: string, patch: ContactOverridePatch): void { + this.overrides.saveOverride(key, patch) + this.invalidateContactsCache() + appLogger.info('contacts', 'contact override saved', { key }) + } + + deleteContactOverride(key: string): void { + this.overrides.deleteOverride(key) + this.invalidateContactsCache() + appLogger.info('contacts', 'contact override deleted', { key }) + } + + invalidateContactsCache(): void { + this.cache = null + } + + private computeContacts(): BuildContactsResult { + const diagnostics = createEmptyDiagnostics() + const accumulators = new Map() + + 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 + + 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) + } + } + + diagnostics.contactsEnabled = diagnostics.privateSessionCount > MIN_PRIVATE_SESSIONS_FOR_CONTACTS + const contacts = this.buildContactItems([...accumulators.values()], diagnostics) + return { contacts, diagnostics } + } + + 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++ + 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, + }) + } + } + + private buildContactItems(accumulators: ContactAccumulator[], diagnostics: ContactsDiagnostics): ContactItem[] { + const overrides = this.overrides.load() + 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 friendTierInputs = friendInputs.map((input) => ({ + input, + score: friendScores.get(input)?.score ?? 0, + privateMessageCount: input.privateMessageCount, + })) + const friendTiers = assignFriendTiers(friendTierInputs) + + const nonFriendScores = computeNonFriendScores(nonFriendInputs) + const nonFriendTierInputs = nonFriendInputs.map((input) => ({ + input, + score: nonFriendScores.get(input)?.score ?? 0, + coOccurrenceCount: input.coOccurrenceCount, + replyInteractionCount: input.replyInteractionCount, + })) + const nonFriendTiers = assignNonFriendTiers(nonFriendTierInputs) + + const contacts: ContactItem[] = [] + + // 评分和分层按好友/非好友分池计算；最终再应用手动锁定，保证用户选择拥有最高优先级。 + for (const tierInput of friendTierInputs) { + const { input } = tierInput + const score = friendScores.get(input) ?? { score: 0, scoreBreakdown: {} } + const algorithmTier = friendTiers.tiers.get(tierInput) ?? 'acquaintance' + contacts.push(this.toContactItem(input.acc, 'friend', algorithmTier, score, overrides[input.acc.key])) + } + + for (const tierInput of nonFriendTierInputs) { + const { input } = tierInput + const score = nonFriendScores.get(input) ?? { score: 0, scoreBreakdown: {} } + const algorithmTier = nonFriendTiers.tiers.get(tierInput) ?? 'low_interaction' + const item = this.toContactItem(input.acc, 'non_friend', algorithmTier, score, overrides[input.acc.key]) + if (item.tier === 'low_interaction') diagnostics.hiddenLowInteractionNonFriends++ + contacts.push(item) + } + + return contacts.sort((a, b) => b.score - a.score || a.displayName.localeCompare(b.displayName)) + } + + private toContactItem( + acc: ContactAccumulator, + pool: 'friend' | 'non_friend', + algorithmTier: ContactTier, + scoring: { score: number; scoreBreakdown: ContactItem['scoreBreakdown'] }, + override: Parameters[1] + ): ContactItem { + const applied = applyContactOverride(algorithmTier, override) + 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, + tier: applied.tier, + algorithmTier: applied.algorithmTier, + lockedTier: applied.lockedTier, + 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}`, `overrides:${this.overrides.getSignaturePart()}`] + 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, + cache: { + status, + computedAt: cached.computedAt, + signature: cached.signature, + staleReason, + }, + } + } + + private now(): number { + return this.deps.now?.() ?? Date.now() + } +} + +function createEmptyDiagnostics(): ContactsDiagnostics { + return { + privateSessionCount: 0, + contactsEnabled: false, + skippedMissingOwnerSessions: 0, + skippedUnresolvedOwnerSessions: 0, + skippedAmbiguousPrivateSessions: 0, + skippedInvalidPlatformIdMembers: 0, + skippedFailedSessions: 0, + hiddenLowInteractionNonFriends: 0, + warnings: [], + } +} + +function getOrCreateAccumulator( + accumulators: Map, + sessionId: string, + meta: SessionMeta, + contact: ContactMemberRef +): ContactAccumulator { + const sessionScoped = isNameMatchPlatform(meta.platform) + const key = buildContactOverrideKey({ + platform: meta.platform, + platformId: contact.platformId, + sessionId, + matchMode: sessionScoped ? 'name' : 'platform_id', + }) + 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 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/index.ts b/packages/node-runtime/src/services/index.ts index 71cbf65c..c4f2ae6b 100644 --- a/packages/node-runtime/src/services/index.ts +++ b/packages/node-runtime/src/services/index.ts @@ -55,6 +55,14 @@ export type { LlmConfig, SummaryServiceDeps } from './summary-service' // Export service export { exportMarkdown } from './export-service' +// Contacts overrides +export { ContactsOverridesManager, buildContactOverrideKey } from './contacts-overrides' +export type { BuildContactOverrideKeyInput, ContactsOverrides } from './contacts-overrides' + +// Contacts service +export { CONTACTS_ALGORITHM_VERSION, createContactsService } from './contacts-service' +export type { ContactsService, ContactsServiceDeps, ContactsServiceOptions } from './contacts-service' + // Merge cache export { MergeSessionCache } from './merge-cache'