From 775deeb9c484cd8b46e47e7ef335e737c605d362 Mon Sep 17 00:00:00 2001 From: lafay <2021211506@stu.hit.edu.cn> Date: Sat, 4 Apr 2026 00:02:22 +0800 Subject: [PATCH] refactor(database): unify database connections and add retry logic Refactor LocalDataSource to reuse the shared database connection from database.ts, avoiding OPFS file handle conflicts. Add retry logic with exponential backoff for database open operations, particularly important for Web/OPFS environments. Also add dynamic theme support to chat input and message bubble styles, using theme colors instead of hardcoded values for better light/dark mode support. BREAKING CHANGE: Database initialization now requires userId to be passed explicitly for user-specific databases --- src/data/datasources/LocalDataSource.ts | 201 +-- .../components/ChatScreen/ChatInput.tsx | 85 +- .../message/components/ChatScreen/styles.ts | 18 +- .../message/components/EmbeddedChat.tsx | 2 +- src/services/database.ts | 1212 +++++++---------- src/stores/authStore.ts | 4 +- tsconfig.json | 3 +- 7 files changed, 573 insertions(+), 952 deletions(-) diff --git a/src/data/datasources/LocalDataSource.ts b/src/data/datasources/LocalDataSource.ts index a0b5e1e..ef919bf 100644 --- a/src/data/datasources/LocalDataSource.ts +++ b/src/data/datasources/LocalDataSource.ts @@ -1,41 +1,37 @@ /** * 本地数据库数据源实现 - * 封装所有 SQLite 操作，提供统一的错误处理和事务支持 + * 复用 database.ts 的共享连接，避免 OPFS 句柄冲突 */ import * as SQLite from 'expo-sqlite'; import { ILocalDataSource, DataSourceError } from './interfaces'; +import { + initDatabase, + getCurrentDbUserId, + isDbInitialized, + getDbInstance, +} from '../../services/database'; -// 数据库实例管理 -let dbInstance: SQLite.SQLiteDatabase | null = null; -let currentDbName: string | null = null; let writeQueue: Promise = Promise.resolve(); export interface LocalDataSourceConfig { - dbName?: string; userId?: string; } export class LocalDataSource implements ILocalDataSource { - private db: SQLite.SQLiteDatabase | null = null; - private dbName: string; + private userId: string | null = null; private initialized = false; private initializingPromise: Promise | null = null; constructor(config: LocalDataSourceConfig = {}) { - // 如果提供了userId，使用用户专属数据库 - this.dbName = config.dbName || (config.userId ? `carrot_bbs_${config.userId}.db` : 'carrot_bbs.db'); + this.userId = config.userId || null; } - /** - * 初始化数据库连接 - */ async initialize(): Promise { - if (this.initialized && this.db) { + if (this.initialized && isDbInitialized()) { return; } - // 防止并发初始化导致连接状态竞争 if (this.initializingPromise) { await this.initializingPromise; return; @@ -51,131 +47,39 @@ export class LocalDataSource implements ILocalDataSource { private async doInitialize(): Promise { try { - // 使用全局实例管理 - if (dbInstance && currentDbName === this.dbName) { - this.db = dbInstance; - this.initialized = true; - return; - } - - // 关闭旧连接 - if (dbInstance) { - try { - await dbInstance.closeAsync(); - } catch (e) { - console.warn('关闭旧数据库连接失败:', e); + // 如果 database.ts 已经初始化，复用它的连接 + if (isDbInitialized()) { + const currentUserId = getCurrentDbUserId(); + if (this.userId && this.userId !== currentUserId) { + await initDatabase(this.userId); + } + } else if (this.userId) { + await initDatabase(this.userId); + } else { + const currentUserId = getCurrentDbUserId(); + if (!currentUserId) { + console.warn('[LocalDataSource] 没有用户登录，跳过初始化'); + return; } } - // 创建新连接 - this.db = await SQLite.openDatabaseAsync(this.dbName); - dbInstance = this.db; - currentDbName = this.dbName; + const db = getDbInstance(); + if (db) { + await db.execAsync(` + CREATE TABLE IF NOT EXISTS posts_cache ( + id TEXT PRIMARY KEY NOT NULL, + data TEXT NOT NULL, + updatedAt TEXT NOT NULL + ) + `); + } + this.initialized = true; - - // 初始化数据库表结构 - await this.createTables(); } catch (error) { this.handleError(error, 'INITIALIZE'); } } - /** - * 创建数据库表结构 - */ - private async createTables(): Promise { - if (!this.db) return; - - const tables = [ - // 消息表 - `CREATE TABLE IF NOT EXISTS messages ( - id TEXT PRIMARY KEY NOT NULL, - conversationId TEXT NOT NULL, - senderId TEXT NOT NULL, - content TEXT, - type TEXT DEFAULT 'text', - isRead INTEGER DEFAULT 0, - createdAt TEXT NOT NULL, - seq INTEGER DEFAULT 0, - status TEXT DEFAULT 'normal', - segments TEXT - )`, - // 会话表 - `CREATE TABLE IF NOT EXISTS conversations ( - id TEXT PRIMARY KEY NOT NULL, - participantId TEXT NOT NULL, - lastMessageId TEXT, - lastSeq INTEGER DEFAULT 0, - unreadCount INTEGER DEFAULT 0, - createdAt TEXT NOT NULL, - updatedAt TEXT NOT NULL - )`, - // 会话缓存表 - `CREATE TABLE IF NOT EXISTS conversation_cache ( - id TEXT PRIMARY KEY NOT NULL, - data TEXT NOT NULL, - updatedAt TEXT NOT NULL - )`, - // 会话列表缓存表 - `CREATE TABLE IF NOT EXISTS conversation_list_cache ( - id TEXT PRIMARY KEY NOT NULL, - data TEXT NOT NULL, - updatedAt TEXT NOT NULL - )`, - // 用户缓存表 - `CREATE TABLE IF NOT EXISTS users ( - id TEXT PRIMARY KEY NOT NULL, - data TEXT NOT NULL, - updatedAt TEXT NOT NULL - )`, - // 当前登录用户缓存 - `CREATE TABLE IF NOT EXISTS current_user_cache ( - id TEXT PRIMARY KEY NOT NULL, - data TEXT NOT NULL, - updatedAt TEXT NOT NULL - )`, - // 群组缓存表 - `CREATE TABLE IF NOT EXISTS groups ( - id TEXT PRIMARY KEY NOT NULL, - data TEXT NOT NULL, - updatedAt TEXT NOT NULL - )`, - // 群成员缓存表 - `CREATE TABLE IF NOT EXISTS group_members ( - groupId TEXT NOT NULL, - userId TEXT NOT NULL, - data TEXT NOT NULL, - updatedAt TEXT NOT NULL, - PRIMARY KEY (groupId, userId) - )`, - // 帖子缓存表 - `CREATE TABLE IF NOT EXISTS posts_cache ( - id TEXT PRIMARY KEY NOT NULL, - data TEXT NOT NULL, - updatedAt TEXT NOT NULL - )`, - ]; - - for (const sql of tables) { - await this.db.execAsync(sql); - } - - // 创建索引 - const indexes = [ - `CREATE INDEX IF NOT EXISTS idx_messages_conversationId ON messages(conversationId)`, - `CREATE INDEX IF NOT EXISTS idx_messages_seq ON messages(seq)`, - `CREATE INDEX IF NOT EXISTS idx_messages_conversation_seq ON messages(conversationId, seq)`, - `CREATE INDEX IF NOT EXISTS idx_group_members_groupId ON group_members(groupId)`, - ]; - - for (const sql of indexes) { - await this.db.execAsync(sql); - } - } - - /** - * 处理错误 - */ private handleError(error: unknown, operation: string): never { const message = error instanceof Error ? error.message : 'Unknown error'; throw new DataSourceError( @@ -186,24 +90,21 @@ export class LocalDataSource implements ILocalDataSource { ); } - /** - * 检查数据库连接 - */ private ensureDb(): SQLite.SQLiteDatabase { - if (!this.db) { + const db = getDbInstance(); + if (!db) { throw new DataSourceError( 'Database not initialized', 'DB_NOT_INITIALIZED', 'LocalDataSource' ); } - return this.db; + return db; } - // ==================== ILocalDataSource 实现 ==================== - async query(sql: string, params?: any[]): Promise { try { + await this.initialize(); const db = this.ensureDb(); if (params && params.length > 0) { return await db.getAllAsync(sql, params as any); @@ -216,6 +117,7 @@ export class LocalDataSource implements ILocalDataSource { async getFirst(sql: string, params?: any[]): Promise { try { + await this.initialize(); const db = this.ensureDb(); if (params && params.length > 0) { return await db.getFirstAsync(sql, params as any); @@ -228,6 +130,7 @@ export class LocalDataSource implements ILocalDataSource { async execute(sql: string, params?: any[]): Promise { try { + await this.initialize(); const db = this.ensureDb(); await db.execAsync(sql); } catch (error) { @@ -237,6 +140,7 @@ export class LocalDataSource implements ILocalDataSource { async run(sql: string, params?: any[]): Promise<{ lastInsertRowId: number; changes: number }> { try { + await this.initialize(); const db = this.ensureDb(); if (params && params.length > 0) { return await db.runAsync(sql, params as any); @@ -244,9 +148,7 @@ export class LocalDataSource implements ILocalDataSource { return await db.runAsync(sql); } catch (error) { if (this.isRecoverableError(error)) { - this.db = null; this.initialized = false; - dbInstance = null; await this.initialize(); const db = this.ensureDb(); if (params && params.length > 0) { @@ -260,6 +162,7 @@ export class LocalDataSource implements ILocalDataSource { async transaction(operations: () => Promise): Promise { try { + await this.initialize(); const db = this.ensureDb(); await db.withTransactionAsync(async () => { await operations(); @@ -279,21 +182,15 @@ export class LocalDataSource implements ILocalDataSource { } } - // ==================== 队列写入支持 ==================== - - /** - * 使用队列执行写入操作（避免并发写入冲突） - */ async enqueueWrite(operation: () => Promise): Promise { + await this.initialize(); const wrappedOperation = async () => { try { return await operation(); } catch (error) { - // 尝试重连后重试 if (this.isRecoverableError(error)) { - console.error('数据库写入异常，尝试重连后重试:', error); - this.db = null; - dbInstance = null; + console.error('[LocalDataSource] 数据库写入异常，尝试重连后重试:', error); + this.initialized = false; await this.initialize(); return operation(); } @@ -306,18 +203,16 @@ export class LocalDataSource implements ILocalDataSource { return queued; } - /** - * 判断错误是否可恢复 - */ private isRecoverableError(error: unknown): boolean { const message = String(error); return ( message.includes('NativeDatabase.prepareAsync') || message.includes('NullPointerException') || - message.includes('database is locked') + message.includes('database is locked') || + message.includes('finalizing statement') || + message.includes('cannot create file') ); } } -// 导出单例实例 export const localDataSource = new LocalDataSource(); diff --git a/src/screens/message/components/ChatScreen/ChatInput.tsx b/src/screens/message/components/ChatScreen/ChatInput.tsx index 608d605..0e19907 100644 --- a/src/screens/message/components/ChatScreen/ChatInput.tsx +++ b/src/screens/message/components/ChatScreen/ChatInput.tsx @@ -51,8 +51,8 @@ export const ChatInput: React.FC StyleSheet.create({ + stripScroll: { + maxHeight: 88, + marginBottom: 6, + paddingHorizontal: 4, + }, + stripContent: { + flexDirection: 'row', + alignItems: 'center', + paddingVertical: 4, + }, + thumbWrap: { + position: 'relative', + marginRight: 8, + }, + thumb: { + width: 72, + height: 72, + borderRadius: 10, + backgroundColor: colors.chat.surfaceMuted || '#E8EAED', + }, + removeBtn: { + position: 'absolute', + top: -6, + right: -6, + backgroundColor: colors.chat.card || '#FFF', + borderRadius: 12, + }, + }), [colors]); + const canSend = (!!inputText.trim() || pendingAttachments.length > 0) && !isComposerBusy && @@ -162,11 +193,11 @@ export const ChatInput: React.FC - + @@ -212,11 +243,11 @@ export const ChatInput: React.FC - + )} @@ -224,34 +255,4 @@ export const ChatInput: React.FC = Promise.resolve(); -// 当前数据库对应的用户ID -let currentDbUserId: string | null = null; +let initPromise: Promise | null = null; +let initTargetUserId: string | null = null; +let isInitialized = false; -/** - * 串行化 open/close，避免并发 initDatabase（如 React Strict Mode 双次 effect、登录与冷启动校验交错） - * 导致后完成的 open 覆盖先完成的连接，进而在 Android 上出现 NativeDatabase.execAsync NPE。 - */ -let dbOpenCloseMutex: Promise = Promise.resolve(); +let dbMutex: Promise = Promise.resolve(); -async function withDbOpenCloseLock(fn: () => Promise): Promise { - const previous = dbOpenCloseMutex; - let release!: () => void; - dbOpenCloseMutex = new Promise((resolve) => { - release = resolve; - }); +const ensureInitialized = async (): Promise => { + if (isInitialized && db) return; + if (initPromise) { + await initPromise; + return; + } + throw new Error('数据库未初始化'); +}; + +async function withMutex(fn: () => Promise): Promise { + const prev = dbMutex; + let release: () => void = () => {}; + dbMutex = new Promise((r) => { release = r; }); try { - await previous; + await prev; return await fn(); } finally { release(); } } -// 数据库版本，用于迁移 -const DB_VERSION = 2; - -/** 关闭连接（须在已持有 dbOpenCloseMutex 时调用，或作为 withDbOpenCloseLock 内的实现） */ -const closeDatabaseUnlocked = async (): Promise => { - if (db) { +async function openDatabaseWithRetry(dbName: string, maxRetries: number = 3): Promise { + const isWeb = typeof window !== 'undefined'; + + for (let attempt = 0; attempt <= maxRetries; attempt++) { try { - await db.closeAsync(); + const database = await SQLite.openDatabaseAsync(dbName); + return database; } catch (error) { - console.error('[Database] 关闭数据库失败:', error); - } - db = null; - currentDbUserId = null; - } -}; - -async function initDatabaseUnlocked(userId?: string): Promise { - // 如果指定了用户ID，使用用户专属数据库 - // 否则使用默认数据库（兼容旧版本） - const dbName = userId ? `carrot_bbs_${userId}.db` : 'carrot_bbs.db'; - - // 如果存在旧的数据库连接且用户ID变化，需要关闭旧连接 - if (db && currentDbUserId !== userId) { - await closeDatabaseUnlocked(); - } - - // 如果已经打开了正确的数据库，直接返回 - if (db && currentDbUserId === userId) { - return; - } - - let opened: SQLite.SQLiteDatabase | null = null; - try { - opened = await SQLite.openDatabaseAsync(dbName); - db = opened; - currentDbUserId = userId || null; - - // 创建消息表（包含新字段 seq, status, segments） - await db.execAsync(` - CREATE TABLE IF NOT EXISTS messages ( - id TEXT PRIMARY KEY NOT NULL, - conversationId TEXT NOT NULL, - senderId TEXT NOT NULL, - content TEXT, - type TEXT DEFAULT 'text', - isRead INTEGER DEFAULT 0, - createdAt TEXT NOT NULL, - seq INTEGER DEFAULT 0, - status TEXT DEFAULT 'normal', - segments TEXT - ); - `); - - // 创建会话表 - await db.execAsync(` - CREATE TABLE IF NOT EXISTS conversations ( - id TEXT PRIMARY KEY NOT NULL, - participantId TEXT NOT NULL, - lastMessageId TEXT, - lastSeq INTEGER DEFAULT 0, - unreadCount INTEGER DEFAULT 0, - createdAt TEXT NOT NULL, - updatedAt TEXT NOT NULL - ); - `); - - // 会话缓存表（完整 JSON，兼容私聊/群聊不同结构） - await db.execAsync(` - CREATE TABLE IF NOT EXISTS conversation_cache ( - id TEXT PRIMARY KEY NOT NULL, - data TEXT NOT NULL, - updatedAt TEXT NOT NULL - ); - `); - - // 会话列表缓存表（仅用于 ChatList，避免被详情缓存污染） - await db.execAsync(` - CREATE TABLE IF NOT EXISTS conversation_list_cache ( - id TEXT PRIMARY KEY NOT NULL, - data TEXT NOT NULL, - updatedAt TEXT NOT NULL - ); - `); - - // 用户缓存表 - await db.execAsync(` - CREATE TABLE IF NOT EXISTS users ( - id TEXT PRIMARY KEY NOT NULL, - data TEXT NOT NULL, - updatedAt TEXT NOT NULL - ); - `); - - // 当前登录用户缓存（仅保存 me） - await db.execAsync(` - CREATE TABLE IF NOT EXISTS current_user_cache ( - id TEXT PRIMARY KEY NOT NULL, - data TEXT NOT NULL, - updatedAt TEXT NOT NULL - ); - `); - - // 群组缓存表 - await db.execAsync(` - CREATE TABLE IF NOT EXISTS groups ( - id TEXT PRIMARY KEY NOT NULL, - data TEXT NOT NULL, - updatedAt TEXT NOT NULL - ); - `); - - // 群成员缓存表 - await db.execAsync(` - CREATE TABLE IF NOT EXISTS group_members ( - groupId TEXT NOT NULL, - userId TEXT NOT NULL, - data TEXT NOT NULL, - updatedAt TEXT NOT NULL, - PRIMARY KEY (groupId, userId) - ); - `); - - // 创建索引 - await db.execAsync(` - CREATE INDEX IF NOT EXISTS idx_messages_conversationId ON messages(conversationId); - `); - await db.execAsync(` - CREATE INDEX IF NOT EXISTS idx_group_members_groupId ON group_members(groupId); - `); - - // 迁移：检查并添加新列（兼容旧版本数据库） - await migrateDatabase(); - - // 迁移后再创建依赖新列的索引 - await db.execAsync(` - CREATE INDEX IF NOT EXISTS idx_messages_seq ON messages(seq); - `); - await db.execAsync(` - CREATE INDEX IF NOT EXISTS idx_messages_conversation_seq ON messages(conversationId, seq); - `); - } catch (error) { - console.error('数据库初始化失败:', error); - try { - if (opened && opened === db) { - await opened.closeAsync(); + const msg = String(error); + console.error(`[Database] 打开失败 (尝试 ${attempt + 1}/${maxRetries + 1}):`, msg); + + if (attempt < maxRetries) { + const delay = isWeb ? 500 * (attempt + 1) : 100; + console.warn(`[Database] ${delay}ms 后重试...`); + await new Promise(r => setTimeout(r, delay)); + + if (msg.includes('Invalid VFS state') || msg.includes('cannot create file')) { + try { + await SQLite.deleteDatabaseAsync(dbName); + console.warn('[Database] 已删除损坏的数据库文件'); + } catch {} + } + } else { + throw error; } - } catch { - /* ignore */ } - db = null; - currentDbUserId = null; - throw error; } + throw new Error('打开数据库失败'); } -/** - * 初始化用户数据库 - * @param userId 用户ID，用于创建用户专属数据库文件 - */ -export const initDatabase = async (userId?: string): Promise => { - return withDbOpenCloseLock(() => initDatabaseUnlocked(userId)); -}; +async function createTables(database: SQLite.SQLiteDatabase): Promise { + if (typeof window !== 'undefined') { + try { + // 关键：使用 EXCLUSIVE 锁定模式，避免 OPFS 多句柄冲突 + // MEMORY journal 避免 journal 文件创建 + // MEMORY temp_store 避免 temp 文件创建 + await database.execAsync(`PRAGMA locking_mode = EXCLUSIVE;`); + await database.execAsync(`PRAGMA journal_mode = MEMORY;`); + await database.execAsync(`PRAGMA temp_store = MEMORY;`); + console.log('[Database] Web: 已设置 EXCLUSIVE locking + MEMORY journal/temp 模式'); + } catch (e) { + console.warn('[Database] 设置 PRAGMA 失败:', e); + } + } -/** - * 关闭数据库连接 - */ -export const closeDatabase = async (): Promise => { - return withDbOpenCloseLock(() => closeDatabaseUnlocked()); -}; + // 拆分为单独的 execAsync 调用，避免 OPFS 句柄冲突 + await database.execAsync(`CREATE TABLE IF NOT EXISTS messages ( + id TEXT PRIMARY KEY NOT NULL, + conversationId TEXT NOT NULL, + senderId TEXT NOT NULL, + content TEXT, + type TEXT DEFAULT 'text', + isRead INTEGER DEFAULT 0, + createdAt TEXT NOT NULL, + seq INTEGER DEFAULT 0, + status TEXT DEFAULT 'normal', + segments TEXT + )`); -/** - * 获取当前数据库用户ID - */ -export const getCurrentDbUserId = (): string | null => { - return currentDbUserId; -}; + await database.execAsync(`CREATE TABLE IF NOT EXISTS conversations ( + id TEXT PRIMARY KEY NOT NULL, + participantId TEXT NOT NULL, + lastMessageId TEXT, + lastSeq INTEGER DEFAULT 0, + unreadCount INTEGER DEFAULT 0, + createdAt TEXT NOT NULL, + updatedAt TEXT NOT NULL + )`); -// 数据库迁移 -const migrateDatabase = async (): Promise => { - const database = await getDb(); + await database.execAsync(`CREATE TABLE IF NOT EXISTS conversation_cache ( + id TEXT PRIMARY KEY NOT NULL, + data TEXT NOT NULL, + updatedAt TEXT NOT NULL + )`); + + await database.execAsync(`CREATE TABLE IF NOT EXISTS conversation_list_cache ( + id TEXT PRIMARY KEY NOT NULL, + data TEXT NOT NULL, + updatedAt TEXT NOT NULL + )`); + + await database.execAsync(`CREATE TABLE IF NOT EXISTS users ( + id TEXT PRIMARY KEY NOT NULL, + data TEXT NOT NULL, + updatedAt TEXT NOT NULL + )`); + + await database.execAsync(`CREATE TABLE IF NOT EXISTS current_user_cache ( + id TEXT PRIMARY KEY NOT NULL, + data TEXT NOT NULL, + updatedAt TEXT NOT NULL + )`); + + await database.execAsync(`CREATE TABLE IF NOT EXISTS groups ( + id TEXT PRIMARY KEY NOT NULL, + data TEXT NOT NULL, + updatedAt TEXT NOT NULL + )`); + + await database.execAsync(`CREATE TABLE IF NOT EXISTS group_members ( + groupId TEXT NOT NULL, + userId TEXT NOT NULL, + data TEXT NOT NULL, + updatedAt TEXT NOT NULL, + PRIMARY KEY (groupId, userId) + )`); + + await database.execAsync(`CREATE INDEX IF NOT EXISTS idx_messages_conversationId ON messages(conversationId)`); + await database.execAsync(`CREATE INDEX IF NOT EXISTS idx_group_members_groupId ON group_members(groupId)`); + await migrateDatabase(database); + + await database.execAsync(`CREATE INDEX IF NOT EXISTS idx_messages_seq ON messages(seq)`); + await database.execAsync(`CREATE INDEX IF NOT EXISTS idx_messages_conversation_seq ON messages(conversationId, seq)`); +} + +async function migrateDatabase(database: SQLite.SQLiteDatabase): Promise { try { - // 检查 messages 表是否有 seq 列 - const tableInfo = await database.getAllAsync( - `PRAGMA table_info(messages)` - ); - + const tableInfo = await database.getAllAsync(`PRAGMA table_info(messages)`); const columns = tableInfo.map(col => col.name); - // 添加 seq 列 if (!columns.includes('seq')) { await database.execAsync(`ALTER TABLE messages ADD COLUMN seq INTEGER DEFAULT 0`); } - - // 添加 status 列 if (!columns.includes('status')) { await database.execAsync(`ALTER TABLE messages ADD COLUMN status TEXT DEFAULT 'normal'`); } - - // 添加 segments 列（用于存储消息的 segments JSON） if (!columns.includes('segments')) { await database.execAsync(`ALTER TABLE messages ADD COLUMN segments TEXT`); } - // 检查 conversations 表是否有 lastSeq 列 - const convTableInfo = await database.getAllAsync( - `PRAGMA table_info(conversations)` - ); + const convTableInfo = await database.getAllAsync(`PRAGMA table_info(conversations)`); const convColumns = convTableInfo.map(col => col.name); - if (!convColumns.includes('lastSeq')) { await database.execAsync(`ALTER TABLE conversations ADD COLUMN lastSeq INTEGER DEFAULT 0`); } } catch (error) { - console.error('数据库迁移失败:', error); + console.error('[Database] 迁移失败:', error); } -}; +} -// 获取数据库实例 -const getDb = async (): Promise => { - if (!db) { - throw new Error('数据库未初始化，请先调用 initDatabase(userId)'); +export const initDatabase = async (userId?: string): Promise => { + const targetUserId = userId || null; + const callId = Date.now(); + + console.log(`[Database ${callId}] initDatabase 开始, userId=${targetUserId}, 当前db=${!!db}, 当前userId=${currentDbUserId}`); + + if (db && currentDbUserId === targetUserId) { + console.log(`[Database ${callId}] 已是目标数据库，直接返回`); + isInitialized = true; + return; } - return db; -}; - -const isRecoverableDbError = (error: unknown): boolean => { - const message = String(error); - return ( - message.includes('NativeDatabase.prepareAsync') || - message.includes('NullPointerException') || - message.includes('Cannot use shared object that was already released') || - message.includes('NativeStatement') - ); -}; - -/** 在已持有 dbOpenCloseMutex 时重建连接（禁止在持有 SQLiteDatabase 期间 await 会抢锁的 recoverDbConnection） */ -const reconnectDatabaseUnlocked = async (): Promise => { - await closeDatabaseUnlocked(); - await initDatabaseUnlocked(currentDbUserId || undefined); -}; - -/** - * 读路径与 init/close/recover 共用同一把锁，避免： - * 线程 A 持有 database 引用执行 prepareAsync 时，线程 B 的 recover 关闭连接导致 NPE。 - */ -const withDbRead = async (operation: (database: SQLite.SQLiteDatabase) => Promise): Promise => { - return withDbOpenCloseLock(async () => { - try { - const database = await getDb(); - return await operation(database); - } catch (error) { - if (!isRecoverableDbError(error)) { - throw error; - } - console.error('数据库读取异常，尝试重连后重试:', error); - await reconnectDatabaseUnlocked(); - const database = await getDb(); - return await operation(database); + + if (initPromise && initTargetUserId === targetUserId) { + console.log(`[Database ${callId}] 等待进行中的初始化`); + await initPromise; + console.log(`[Database ${callId}] 初始化完成`); + return; + } + + if (initPromise) { + console.log(`[Database ${callId}] 等待其他初始化完成`); + await initPromise.catch(() => {}); + } + + if (db && currentDbUserId === targetUserId) { + console.log(`[Database ${callId}] 其他初始化已完成目标，直接返回`); + isInitialized = true; + return; + } + + initTargetUserId = targetUserId; + console.log(`[Database ${callId}] 开始实际初始化`); + + initPromise = withMutex(async () => { + const mutexId = callId; + console.log(`[Database ${mutexId}] 获得互斥锁`); + + if (db && currentDbUserId === targetUserId) { + console.log(`[Database ${mutexId}] 互斥锁内检查：已是目标数据库`); + isInitialized = true; + return; } + + const isWeb = typeof window !== 'undefined'; + + if (db) { + if (isWeb) { + console.log(`[Database ${mutexId}] Web：切换用户，直接置空旧连接`); + db = null; + currentDbUserId = null; + isInitialized = false; + } else { + try { + await db.closeAsync(); + } catch (e) { + console.warn(`[Database ${mutexId}] 关闭旧连接失败:`, e); + } + db = null; + currentDbUserId = null; + isInitialized = false; + } + } + + const dbName = targetUserId ? `carrot_bbs_${targetUserId}.db` : 'carrot_bbs.db'; + console.log(`[Database ${mutexId}] 打开数据库: ${dbName}`); + + const database = await openDatabaseWithRetry(dbName); + console.log(`[Database ${mutexId}] 数据库已打开，创建表`); + + await createTables(database); + console.log(`[Database ${mutexId}] 表创建完成`); + + db = database; + currentDbUserId = targetUserId; + isInitialized = true; + console.log(`[Database ${mutexId}] 初始化完成`); + }).finally(() => { + console.log(`[Database ${callId}] 清理初始化状态`); + initPromise = null; + initTargetUserId = null; + }); + + await initPromise; + console.log(`[Database ${callId}] initDatabase 返回`); +}; + +export const closeDatabase = async (): Promise => { + if (initPromise) { + await initPromise.catch(() => {}); + } + + await withMutex(async () => { + if (!db) return; + + const isWeb = typeof window !== 'undefined'; + if (isWeb) { + console.log('[Database] Web 环境：跳过关闭数据库，避免 OPFS 句柄冲突'); + db = null; + currentDbUserId = null; + isInitialized = false; + return; + } + + try { + await db.closeAsync(); + } catch (e) { + console.warn('[Database] 关闭失败:', e); + } + db = null; + currentDbUserId = null; + isInitialized = false; }); }; -const enqueueWrite = async (operation: () => Promise): Promise => { - const wrappedOperation = async () => { - return withDbOpenCloseLock(async () => { - try { - return await operation(); - } catch (error) { - if (!isRecoverableDbError(error)) { - throw error; - } - console.error('数据库写入异常，尝试重连后重试:', error); - await reconnectDatabaseUnlocked(); - return await operation(); - } - }); - }; - const queued = writeQueue.then(wrappedOperation, wrappedOperation); - writeQueue = queued.then(() => undefined, () => undefined); - return queued; +export const getCurrentDbUserId = (): string | null => currentDbUserId; + +export const isDbInitialized = (): boolean => isInitialized && db !== null; + +export const getDbInstance = (): SQLite.SQLiteDatabase | null => db; + +const getDb = async (timeout: number = 10000): Promise => { + if (isInitialized && db) return db; + + if (initPromise) { + const timeoutPromise = new Promise((_, reject) => + setTimeout(() => reject(new Error('等待数据库初始化超时')), timeout) + ); + + await Promise.race([initPromise, timeoutPromise]); + + if (db) return db; + } + + throw new Error('数据库未初始化，请先调用 initDatabase(userId)'); }; -// ==================== 消息类型定义 ==================== +const enqueueWrite = async (op: (db: SQLite.SQLiteDatabase) => Promise): Promise => { + let result: T; + const wrapped = async () => { + result = await withMutex(async () => { + const database = await getDb(); + return op(database); + }); + }; + const queued = writeQueue.then(wrapped, wrapped); + writeQueue = queued.then(() => {}, () => {}); + await queued; + return result!; +}; + +const withDbRead = async (op: (db: SQLite.SQLiteDatabase) => Promise): Promise => { + return withMutex(async () => { + const database = await getDb(); + return op(database); + }); +}; export interface CachedMessage { id: string; @@ -338,575 +361,340 @@ export interface CachedMessage { segments?: any; } -// ==================== 消息操作 ==================== - -// 保存单条消息 export const saveMessage = async (message: CachedMessage): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); + await enqueueWrite(async (database) => { await database.runAsync( `INSERT OR REPLACE INTO messages (id, conversationId, senderId, content, type, isRead, createdAt, seq, status, segments) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, - [ - message.id, - message.conversationId, - message.senderId, - message.content || '', - message.type || 'text', - message.isRead ? 1 : 0, - message.createdAt, - message.seq || 0, - message.status || 'normal', - message.segments ? JSON.stringify(message.segments) : null, - ] + [message.id, message.conversationId, message.senderId, message.content || '', message.type || 'text', + message.isRead ? 1 : 0, message.createdAt, message.seq || 0, message.status || 'normal', + message.segments ? JSON.stringify(message.segments) : null] ); }); }; -// 批量保存消息 export const saveMessagesBatch = async (messages: CachedMessage[]): Promise => { - if (!messages || messages.length === 0) return; - - await enqueueWrite(async () => { - const database = await getDb(); - for (const message of messages) { + if (!messages?.length) return; + await enqueueWrite(async (database) => { + for (const msg of messages) { await database.runAsync( `INSERT OR REPLACE INTO messages (id, conversationId, senderId, content, type, isRead, createdAt, seq, status, segments) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, - [ - message.id, - message.conversationId, - message.senderId, - message.content || '', - message.type || 'text', - message.isRead ? 1 : 0, - message.createdAt, - message.seq || 0, - message.status || 'normal', - message.segments ? JSON.stringify(message.segments) : null, - ] + [msg.id, msg.conversationId, msg.senderId, msg.content || '', msg.type || 'text', + msg.isRead ? 1 : 0, msg.createdAt, msg.seq || 0, msg.status || 'normal', + msg.segments ? JSON.stringify(msg.segments) : null] ); } }); }; -// 获取会话的消息（默认只加载最近的20条，避免卡顿） -export const getMessagesByConversation = async ( - conversationId: string, - limit: number = 20 -): Promise => { +export const getMessagesByConversation = async (conversationId: string, limit: number = 20): Promise => { return withDbRead(async (database) => { - const result = await database.getAllAsync( - `SELECT * FROM messages - WHERE conversationId = ? - ORDER BY seq DESC - LIMIT ?`, + const rows = await database.getAllAsync( + `SELECT * FROM messages WHERE conversationId = ? ORDER BY seq DESC LIMIT ?`, [conversationId, limit] ); - - // 反转以保持时间正序（最新的在底部） - return result.reverse().map(msg => ({ - ...msg, - isRead: msg.isRead === 1, - segments: msg.segments ? JSON.parse(msg.segments) : undefined, + return rows.reverse().map(r => ({ + ...r, isRead: r.isRead === 1, segments: r.segments ? JSON.parse(r.segments) : undefined })); }); }; -// 获取会话的最大消息序号 export const getMaxSeq = async (conversationId: string): Promise => { return withDbRead(async (database) => { - const result = await database.getFirstAsync<{ maxSeq: number }>( - `SELECT MAX(seq) as maxSeq FROM messages WHERE conversationId = ?`, - [conversationId] + const r = await database.getFirstAsync<{ maxSeq: number }>( + `SELECT MAX(seq) as maxSeq FROM messages WHERE conversationId = ?`, [conversationId] ); - - return result?.maxSeq || 0; + return r?.maxSeq || 0; }); }; -// 获取会话的最小消息序号 export const getMinSeq = async (conversationId: string): Promise => { return withDbRead(async (database) => { - const result = await database.getFirstAsync<{ minSeq: number }>( - `SELECT MIN(seq) as minSeq FROM messages WHERE conversationId = ?`, - [conversationId] + const r = await database.getFirstAsync<{ minSeq: number }>( + `SELECT MIN(seq) as minSeq FROM messages WHERE conversationId = ?`, [conversationId] ); - - return result?.minSeq || 0; + return r?.minSeq || 0; }); }; -// 获取指定 seq 之前的消息（用于加载历史） export const getMessagesBeforeSeq = async (conversationId: string, beforeSeq: number, limit: number = 20): Promise => { return withDbRead(async (database) => { - const result = await database.getAllAsync( - `SELECT * FROM messages - WHERE conversationId = ? AND seq < ? - ORDER BY seq DESC - LIMIT ?`, + const rows = await database.getAllAsync( + `SELECT * FROM messages WHERE conversationId = ? AND seq < ? ORDER BY seq DESC LIMIT ?`, [conversationId, beforeSeq, limit] ); - - return result.map(msg => ({ - ...msg, - isRead: msg.isRead === 1, - segments: msg.segments ? JSON.parse(msg.segments) : undefined, - })).reverse(); // 反转以保持时间正序 + return rows.map(r => ({ ...r, isRead: r.isRead === 1, segments: r.segments ? JSON.parse(r.segments) : undefined })).reverse(); }); }; -// 获取本地消息数量（用于判断是否还有更多历史） export const getLocalMessageCountBeforeSeq = async (conversationId: string, beforeSeq: number): Promise => { return withDbRead(async (database) => { - const result = await database.getFirstAsync<{ count: number }>( - `SELECT COUNT(*) as count FROM messages WHERE conversationId = ? AND seq < ?`, - [conversationId, beforeSeq] + const r = await database.getFirstAsync<{ count: number }>( + `SELECT COUNT(*) as count FROM messages WHERE conversationId = ? AND seq < ?`, [conversationId, beforeSeq] ); - return result?.count || 0; + return r?.count || 0; }); }; -// 获取消息数量 export const getMessageCount = async (conversationId: string): Promise => { return withDbRead(async (database) => { - const result = await database.getFirstAsync<{ count: number }>( - `SELECT COUNT(*) as count FROM messages WHERE conversationId = ?`, - [conversationId] + const r = await database.getFirstAsync<{ count: number }>( + `SELECT COUNT(*) as count FROM messages WHERE conversationId = ?`, [conversationId] ); - return result?.count || 0; + return r?.count || 0; }); }; -// 标记消息为已读 export const markMessageAsRead = async (messageId: string): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - await database.runAsync( - `UPDATE messages SET isRead = 1 WHERE id = ?`, - [messageId] - ); + await enqueueWrite(async (db) => { + await db.runAsync(`UPDATE messages SET isRead = 1 WHERE id = ?`, [messageId]); }); }; -// 标记会话的所有消息为已读 export const markConversationAsRead = async (conversationId: string): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - await database.runAsync( - `UPDATE messages SET isRead = 1 WHERE conversationId = ?`, - [conversationId] - ); + await enqueueWrite(async (db) => { + await db.runAsync(`UPDATE messages SET isRead = 1 WHERE conversationId = ?`, [conversationId]); }); }; -// 删除单条消息 export const deleteMessage = async (messageId: string): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - await database.runAsync( - `DELETE FROM messages WHERE id = ?`, - [messageId] - ); + await enqueueWrite(async (db) => { + await db.runAsync(`DELETE FROM messages WHERE id = ?`, [messageId]); }); }; -// 更新消息状态（如撤回） -// clearContent=true 时，会同时清空本地存储的消息内容与 segments，仅保留状态占位 -export const updateMessageStatus = async ( - messageId: string, - status: string, - clearContent: boolean = false -): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); +export const updateMessageStatus = async (messageId: string, status: string, clearContent: boolean = false): Promise => { + await enqueueWrite(async (db) => { if (clearContent) { - await database.runAsync( - `UPDATE messages SET status = ?, content = '', segments = '[]' WHERE id = ?`, - [status, messageId] - ); - return; + await db.runAsync(`UPDATE messages SET status = ?, content = '', segments = '[]' WHERE id = ?`, [status, messageId]); + } else { + await db.runAsync(`UPDATE messages SET status = ? WHERE id = ?`, [status, messageId]); } - await database.runAsync( - `UPDATE messages SET status = ? WHERE id = ?`, - [status, messageId] - ); }); }; -// 清空会话的所有消息 export const clearConversationMessages = async (conversationId: string): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - await database.runAsync( - `DELETE FROM messages WHERE conversationId = ?`, - [conversationId] - ); + await enqueueWrite(async (db) => { + await db.runAsync(`DELETE FROM messages WHERE conversationId = ?`, [conversationId]); }); }; -// ==================== 会话操作 ==================== - -// 保存或更新会话 -export const saveConversation = async (conversation: { - id: string; - participantId: string; - lastMessageId?: string; - lastSeq?: number; - unreadCount?: number; - createdAt: string; - updatedAt: string; +export const saveConversation = async (conv: { + id: string; participantId: string; lastMessageId?: string; lastSeq?: number; unreadCount?: number; + createdAt: string; updatedAt: string; }): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - await database.runAsync( + await enqueueWrite(async (db) => { + await db.runAsync( `INSERT OR REPLACE INTO conversations (id, participantId, lastMessageId, lastSeq, unreadCount, createdAt, updatedAt) VALUES (?, ?, ?, ?, ?, ?, ?)`, - [ - conversation.id, - conversation.participantId, - conversation.lastMessageId || null, - conversation.lastSeq || 0, - conversation.unreadCount || 0, - conversation.createdAt, - conversation.updatedAt, - ] + [conv.id, conv.participantId, conv.lastMessageId || null, conv.lastSeq || 0, conv.unreadCount || 0, conv.createdAt, conv.updatedAt] ); }); }; -// 获取所有会话 export const getAllConversations = async (): Promise => { - return withDbRead(async (database) => { - return database.getAllAsync(`SELECT * FROM conversations ORDER BY updatedAt DESC`); + return withDbRead(async (db) => db.getAllAsync(`SELECT * FROM conversations ORDER BY updatedAt DESC`)); +}; + +export const updateConversationUnreadCount = async (id: string, count: number): Promise => { + await enqueueWrite(async (db) => { + await db.runAsync(`UPDATE conversations SET unreadCount = ? WHERE id = ?`, [count, id]); }); }; -// 更新会话未读数 -export const updateConversationUnreadCount = async (conversationId: string, count: number): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - await database.runAsync( - `UPDATE conversations SET unreadCount = ? WHERE id = ?`, - [count, conversationId] - ); +export const updateConversationLastMessage = async (id: string, lastMessageId: string, lastSeq: number, updatedAt: string): Promise => { + await enqueueWrite(async (db) => { + await db.runAsync(`UPDATE conversations SET lastMessageId = ?, lastSeq = ?, updatedAt = ? WHERE id = ?`, + [lastMessageId, lastSeq, updatedAt, id]); }); }; -// 更新会话最后消息 -export const updateConversationLastMessage = async ( - conversationId: string, - lastMessageId: string, - lastSeq: number, - updatedAt: string -): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - await database.runAsync( - `UPDATE conversations SET lastMessageId = ?, lastSeq = ?, updatedAt = ? WHERE id = ?`, - [lastMessageId, lastSeq, updatedAt, conversationId] - ); - }); -}; - -// 删除会话（包含所有消息） export const deleteConversation = async (conversationId: string): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - await database.runAsync( - `DELETE FROM messages WHERE conversationId = ?`, - [conversationId] - ); - await database.runAsync( - `DELETE FROM conversations WHERE id = ?`, - [conversationId] - ); + await enqueueWrite(async (db) => { + await db.runAsync(`DELETE FROM messages WHERE conversationId = ?`, [conversationId]); + await db.runAsync(`DELETE FROM conversations WHERE id = ?`, [conversationId]); }); }; -// ==================== 搜索 ==================== - -// 搜索消息 export const searchMessages = async (keyword: string): Promise => { - return withDbRead(async (database) => { - const result = await database.getAllAsync( - `SELECT * FROM messages - WHERE content LIKE ? - ORDER BY createdAt DESC - LIMIT 50`, - [`%${keyword}%`] + return withDbRead(async (db) => { + const rows = await db.getAllAsync( + `SELECT * FROM messages WHERE content LIKE ? ORDER BY createdAt DESC LIMIT 50`, [`%${keyword}%`] ); - return result.map(msg => ({ - ...msg, - isRead: msg.isRead === 1, - segments: msg.segments ? JSON.parse(msg.segments) : undefined, - })); + return rows.map(r => ({ ...r, isRead: r.isRead === 1, segments: r.segments ? JSON.parse(r.segments) : undefined })); }); }; -// ==================== 工具函数 ==================== - -// 获取数据库统计信息 -export const getDatabaseStats = async (): Promise<{ - totalMessages: number; - totalConversations: number; -}> => { - return withDbRead(async (database) => { - const msgCount = await database.getFirstAsync<{ count: number }>( - `SELECT COUNT(*) as count FROM messages` - ); - const convCount = await database.getFirstAsync<{ count: number }>( - `SELECT COUNT(*) as count FROM conversations` - ); - return { - totalMessages: msgCount?.count || 0, - totalConversations: convCount?.count || 0, - }; +export const getDatabaseStats = async (): Promise<{ totalMessages: number; totalConversations: number }> => { + return withDbRead(async (db) => { + const msgCount = await db.getFirstAsync<{ count: number }>(`SELECT COUNT(*) as count FROM messages`); + const convCount = await db.getFirstAsync<{ count: number }>(`SELECT COUNT(*) as count FROM conversations`); + return { totalMessages: msgCount?.count || 0, totalConversations: convCount?.count || 0 }; }); }; -// 清空所有数据（用于调试或用户退出登录） export const clearAllData = async (): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - await database.execAsync(`DELETE FROM messages`); - await database.execAsync(`DELETE FROM conversations`); - await database.execAsync(`DELETE FROM conversation_cache`); - await database.execAsync(`DELETE FROM conversation_list_cache`); - await database.execAsync(`DELETE FROM users`); - await database.execAsync(`DELETE FROM current_user_cache`); - await database.execAsync(`DELETE FROM groups`); - await database.execAsync(`DELETE FROM group_members`); + await enqueueWrite(async (db) => { + await db.execAsync(`DELETE FROM messages`); + await db.execAsync(`DELETE FROM conversations`); + await db.execAsync(`DELETE FROM conversation_cache`); + await db.execAsync(`DELETE FROM conversation_list_cache`); + await db.execAsync(`DELETE FROM users`); + await db.execAsync(`DELETE FROM current_user_cache`); + await db.execAsync(`DELETE FROM groups`); + await db.execAsync(`DELETE FROM group_members`); }); }; -// ==================== 通用 JSON 缓存辅助 ==================== - const safeParseJson = (value: string): T | null => { - try { - return JSON.parse(value) as T; - } catch { - return null; - } + try { return JSON.parse(value) as T; } catch { return null; } }; -// ==================== 用户缓存 ==================== - export const saveUserCache = async (user: UserDTO): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - await database.runAsync( - `INSERT OR REPLACE INTO users (id, data, updatedAt) VALUES (?, ?, ?)`, - [String(user.id), JSON.stringify(user), new Date().toISOString()] - ); + await enqueueWrite(async (db) => { + await db.runAsync(`INSERT OR REPLACE INTO users (id, data, updatedAt) VALUES (?, ?, ?)`, + [String(user.id), JSON.stringify(user), new Date().toISOString()]); }); }; export const saveUsersCache = async (users: UserDTO[]): Promise => { - if (!users || users.length === 0) return; - await enqueueWrite(async () => { - const database = await getDb(); - for (const user of users) { - await database.runAsync( - `INSERT OR REPLACE INTO users (id, data, updatedAt) VALUES (?, ?, ?)`, - [String(user.id), JSON.stringify(user), new Date().toISOString()] - ); + if (!users?.length) return; + await enqueueWrite(async (db) => { + const now = new Date().toISOString(); + for (const u of users) { + await db.runAsync(`INSERT OR REPLACE INTO users (id, data, updatedAt) VALUES (?, ?, ?)`, + [String(u.id), JSON.stringify(u), now]); } }); }; export const getUserCache = async (userId: string): Promise => { - return withDbRead(async (database) => { - const row = await database.getFirstAsync<{ data: string }>( - `SELECT data FROM users WHERE id = ?`, - [String(userId)] - ); - if (!row?.data) return null; - return safeParseJson(row.data); + return withDbRead(async (db) => { + const r = await db.getFirstAsync<{ data: string }>(`SELECT data FROM users WHERE id = ?`, [String(userId)]); + return r?.data ? safeParseJson(r.data) : null; }); }; export const getLatestUserCache = async (): Promise => { - return withDbRead(async (database) => { - const row = await database.getFirstAsync<{ data: string }>( - `SELECT data FROM users ORDER BY updatedAt DESC LIMIT 1` - ); - if (!row?.data) return null; - return safeParseJson(row.data); + return withDbRead(async (db) => { + const r = await db.getFirstAsync<{ data: string }>(`SELECT data FROM users ORDER BY updatedAt DESC LIMIT 1`); + return r?.data ? safeParseJson(r.data) : null; }); }; export const saveCurrentUserCache = async (user: UserDTO): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - await database.runAsync( - `INSERT OR REPLACE INTO current_user_cache (id, data, updatedAt) VALUES (?, ?, ?)`, - ['me', JSON.stringify(user), new Date().toISOString()] - ); + await enqueueWrite(async (db) => { + await db.runAsync(`INSERT OR REPLACE INTO current_user_cache (id, data, updatedAt) VALUES (?, ?, ?)`, + ['me', JSON.stringify(user), new Date().toISOString()]); }); }; export const getCurrentUserCache = async (): Promise => { - return withDbRead(async (database) => { - const row = await database.getFirstAsync<{ data: string }>( - `SELECT data FROM current_user_cache WHERE id = 'me'` - ); - if (!row?.data) return null; - return safeParseJson(row.data); + return withDbRead(async (db) => { + const r = await db.getFirstAsync<{ data: string }>(`SELECT data FROM current_user_cache WHERE id = 'me'`); + return r?.data ? safeParseJson(r.data) : null; }); }; export const clearCurrentUserCache = async (): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - await database.runAsync(`DELETE FROM current_user_cache WHERE id = 'me'`); + await enqueueWrite(async (db) => { + await db.runAsync(`DELETE FROM current_user_cache WHERE id = 'me'`); }); }; -// ==================== 群组缓存 ==================== - export const saveGroupCache = async (group: GroupResponse): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - await database.runAsync( - `INSERT OR REPLACE INTO groups (id, data, updatedAt) VALUES (?, ?, ?)`, - [String(group.id), JSON.stringify(group), new Date().toISOString()] - ); + await enqueueWrite(async (db) => { + await db.runAsync(`INSERT OR REPLACE INTO groups (id, data, updatedAt) VALUES (?, ?, ?)`, + [String(group.id), JSON.stringify(group), new Date().toISOString()]); }); }; export const saveGroupsCache = async (groups: GroupResponse[]): Promise => { - if (!groups || groups.length === 0) return; - await enqueueWrite(async () => { - const database = await getDb(); - for (const group of groups) { - await database.runAsync( - `INSERT OR REPLACE INTO groups (id, data, updatedAt) VALUES (?, ?, ?)`, - [String(group.id), JSON.stringify(group), new Date().toISOString()] - ); + if (!groups?.length) return; + await enqueueWrite(async (db) => { + const now = new Date().toISOString(); + for (const g of groups) { + await db.runAsync(`INSERT OR REPLACE INTO groups (id, data, updatedAt) VALUES (?, ?, ?)`, + [String(g.id), JSON.stringify(g), now]); } }); }; export const getGroupCache = async (groupId: string): Promise => { - return withDbRead(async (database) => { - const row = await database.getFirstAsync<{ data: string }>( - `SELECT data FROM groups WHERE id = ?`, - [String(groupId)] - ); - if (!row?.data) return null; - return safeParseJson(row.data); + return withDbRead(async (db) => { + const r = await db.getFirstAsync<{ data: string }>(`SELECT data FROM groups WHERE id = ?`, [String(groupId)]); + return r?.data ? safeParseJson(r.data) : null; }); }; export const getAllGroupsCache = async (): Promise => { - return withDbRead(async (database) => { - const rows = await database.getAllAsync<{ data: string }>( - `SELECT data FROM groups ORDER BY updatedAt DESC` - ); - return rows - .map(row => safeParseJson(row.data)) - .filter((item): item is GroupResponse => Boolean(item)); + return withDbRead(async (db) => { + const rows = await db.getAllAsync<{ data: string }>(`SELECT data FROM groups ORDER BY updatedAt DESC`); + return rows.map(r => safeParseJson(r.data)).filter((g): g is GroupResponse => Boolean(g)); }); }; -// ==================== 群成员缓存 ==================== - -export const saveGroupMembersCache = async ( - groupId: string, - members: GroupMemberResponse[] -): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); +export const saveGroupMembersCache = async (groupId: string, members: GroupMemberResponse[]): Promise => { + await enqueueWrite(async (db) => { const now = new Date().toISOString(); - await database.runAsync(`DELETE FROM group_members WHERE groupId = ?`, [String(groupId)]); - for (const member of members) { - await database.runAsync( - `INSERT OR REPLACE INTO group_members (groupId, userId, data, updatedAt) VALUES (?, ?, ?, ?)`, - [String(groupId), String(member.user_id), JSON.stringify(member), now] - ); + await db.runAsync(`DELETE FROM group_members WHERE groupId = ?`, [String(groupId)]); + for (const m of members) { + await db.runAsync(`INSERT OR REPLACE INTO group_members (groupId, userId, data, updatedAt) VALUES (?, ?, ?, ?)`, + [String(groupId), String(m.user_id), JSON.stringify(m), now]); } }); }; export const getGroupMembersCache = async (groupId: string): Promise => { - return withDbRead(async (database) => { - const rows = await database.getAllAsync<{ data: string }>( - `SELECT data FROM group_members WHERE groupId = ? ORDER BY updatedAt DESC`, - [String(groupId)] + return withDbRead(async (db) => { + const rows = await db.getAllAsync<{ data: string }>( + `SELECT data FROM group_members WHERE groupId = ? ORDER BY updatedAt DESC`, [String(groupId)] ); - return rows - .map(row => safeParseJson(row.data)) - .filter((item): item is GroupMemberResponse => Boolean(item)); + return rows.map(r => safeParseJson(r.data)).filter((m): m is GroupMemberResponse => Boolean(m)); }); }; -// ==================== 会话缓存 ==================== - type ConversationCacheData = ConversationResponse | ConversationDetailResponse; -export const saveConversationCache = async (conversation: ConversationCacheData): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - const cacheUpdatedAt = ('updated_at' in conversation && conversation.updated_at) - ? String(conversation.updated_at) - : new Date().toISOString(); - await database.runAsync( - `INSERT OR REPLACE INTO conversation_cache (id, data, updatedAt) VALUES (?, ?, ?)`, - [String(conversation.id), JSON.stringify(conversation), cacheUpdatedAt] - ); +export const saveConversationCache = async (conv: ConversationCacheData): Promise => { + await enqueueWrite(async (db) => { + const updatedAt = ('updated_at' in conv && conv.updated_at) ? String(conv.updated_at) : new Date().toISOString(); + await db.runAsync(`INSERT OR REPLACE INTO conversation_cache (id, data, updatedAt) VALUES (?, ?, ?)`, + [String(conv.id), JSON.stringify(conv), updatedAt]); }); }; -export const saveConversationsCache = async (conversations: ConversationResponse[]): Promise => { - if (!conversations || conversations.length === 0) return; - await enqueueWrite(async () => { - const database = await getDb(); - for (const conversation of conversations) { - await database.runAsync( - `INSERT OR REPLACE INTO conversation_list_cache (id, data, updatedAt) VALUES (?, ?, ?)`, - [ - String(conversation.id), - JSON.stringify(conversation), - conversation.updated_at || new Date().toISOString(), - ] - ); +export const saveConversationsCache = async (convs: ConversationResponse[]): Promise => { + if (!convs?.length) return; + await enqueueWrite(async (db) => { + for (const c of convs) { + await db.runAsync(`INSERT OR REPLACE INTO conversation_list_cache (id, data, updatedAt) VALUES (?, ?, ?)`, + [String(c.id), JSON.stringify(c), c.updated_at || new Date().toISOString()]); } }); }; -export const getConversationCache = async (conversationId: string): Promise => { - return withDbRead(async (database) => { - const row = await database.getFirstAsync<{ data: string }>( - `SELECT data FROM conversation_cache WHERE id = ?`, - [String(conversationId)] - ); - if (!row?.data) return null; - return safeParseJson(row.data); +export const getConversationCache = async (id: string): Promise => { + return withDbRead(async (db) => { + const r = await db.getFirstAsync<{ data: string }>(`SELECT data FROM conversation_cache WHERE id = ?`, [String(id)]); + return r?.data ? safeParseJson(r.data) : null; }); }; export const getConversationListCache = async (): Promise => { - return withDbRead(async (database) => { - let rows = await database.getAllAsync<{ data: string }>( - `SELECT data FROM conversation_list_cache ORDER BY updatedAt DESC` - ); - // 兼容旧版本：首次升级时尝试从旧表读取一次 - if (rows.length === 0) { - rows = await database.getAllAsync<{ data: string }>( - `SELECT data FROM conversation_cache ORDER BY updatedAt DESC` - ); + return withDbRead(async (db) => { + let rows = await db.getAllAsync<{ data: string }>(`SELECT data FROM conversation_list_cache ORDER BY updatedAt DESC`); + if (!rows.length) { + rows = await db.getAllAsync<{ data: string }>(`SELECT data FROM conversation_cache ORDER BY updatedAt DESC`); } - const list = rows - .map(row => safeParseJson(row.data)) - .filter((item): item is ConversationResponse => Boolean(item)) - .filter(item => typeof item.id !== 'undefined' && typeof item.type !== 'undefined'); + const list = rows.map(r => safeParseJson(r.data)) + .filter((c): c is ConversationResponse => Boolean(c)) + .filter(c => typeof c.id !== 'undefined' && typeof c.type !== 'undefined'); return list.sort((a, b) => { const aPinned = a.is_pinned ? 1 : 0; const bPinned = b.is_pinned ? 1 : 0; - if (aPinned !== bPinned) { - return bPinned - aPinned; - } + if (aPinned !== bPinned) return bPinned - aPinned; const aTime = new Date(a.last_message_at || a.updated_at || 0).getTime(); const bTime = new Date(b.last_message_at || b.updated_at || 0).getTime(); return bTime - aTime; @@ -914,130 +702,64 @@ export const getConversationListCache = async (): Promise -): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - const row = await database.getFirstAsync<{ data: string }>( - `SELECT data FROM conversation_list_cache WHERE id = ?`, - [String(conversationId)] - ); - if (!row?.data) return; - const conv = safeParseJson(row.data); +export const updateConversationListCacheEntry = async (id: string, updates: Partial): Promise => { + await enqueueWrite(async (db) => { + const r = await db.getFirstAsync<{ data: string }>(`SELECT data FROM conversation_list_cache WHERE id = ?`, [String(id)]); + if (!r?.data) return; + const conv = safeParseJson(r.data); if (!conv) return; const updated = { ...conv, ...updates }; const updatedAt = updates.last_message_at || updated.last_message_at || new Date().toISOString(); - await database.runAsync( - `UPDATE conversation_list_cache SET data = ?, updatedAt = ? WHERE id = ?`, - [JSON.stringify(updated), updatedAt, String(conversationId)] - ); + await db.runAsync(`UPDATE conversation_list_cache SET data = ?, updatedAt = ? WHERE id = ?`, + [JSON.stringify(updated), updatedAt, String(id)]); }); }; -/** - * 更新会话 SWR 缓存中的未读数（标记已读后立即同步清零，避免下次从缓存加载时仍显示旧红点） - * @param myLastReadSeq 若传入，同时写入 my_last_read_seq，便于冷启动用 last_seq 与已读游标校正未读 - */ -export const updateConversationCacheUnreadCount = async ( - conversationId: string, - count: number, - myLastReadSeq?: number -): Promise => { - await enqueueWrite(async () => { - const database = await getDb(); - - // 更新 conversation_list_cache - const listRow = await database.getFirstAsync<{ data: string }>( - `SELECT data FROM conversation_list_cache WHERE id = ?`, - [String(conversationId)] - ); +export const updateConversationCacheUnreadCount = async (id: string, count: number, myLastReadSeq?: number): Promise => { + await enqueueWrite(async (db) => { + const listRow = await db.getFirstAsync<{ data: string }>(`SELECT data FROM conversation_list_cache WHERE id = ?`, [String(id)]); if (listRow?.data) { const conv = safeParseJson(listRow.data); if (conv) { conv.unread_count = count; - if (typeof myLastReadSeq === 'number' && !Number.isNaN(myLastReadSeq)) { - conv.my_last_read_seq = myLastReadSeq; - } - await database.runAsync( - `UPDATE conversation_list_cache SET data = ? WHERE id = ?`, - [JSON.stringify(conv), String(conversationId)] - ); + if (typeof myLastReadSeq === 'number' && !Number.isNaN(myLastReadSeq)) conv.my_last_read_seq = myLastReadSeq; + await db.runAsync(`UPDATE conversation_list_cache SET data = ? WHERE id = ?`, [JSON.stringify(conv), String(id)]); } } - - // 更新 conversation_cache - const cacheRow = await database.getFirstAsync<{ data: string }>( - `SELECT data FROM conversation_cache WHERE id = ?`, - [String(conversationId)] - ); + + const cacheRow = await db.getFirstAsync<{ data: string }>(`SELECT data FROM conversation_cache WHERE id = ?`, [String(id)]); if (cacheRow?.data) { const conv = safeParseJson>(cacheRow.data); if (conv) { conv.unread_count = count; - if (typeof myLastReadSeq === 'number' && !Number.isNaN(myLastReadSeq)) { - conv.my_last_read_seq = myLastReadSeq; - } - await database.runAsync( - `UPDATE conversation_cache SET data = ? WHERE id = ?`, - [JSON.stringify(conv), String(conversationId)] - ); + if (typeof myLastReadSeq === 'number' && !Number.isNaN(myLastReadSeq)) conv.my_last_read_seq = myLastReadSeq; + await db.runAsync(`UPDATE conversation_cache SET data = ? WHERE id = ?`, [JSON.stringify(conv), String(id)]); } } }); }; -/** - * 批量保存会话列表及其相关数据（用户、群组） - * 在单个队列操作中完成，避免多次并发数据库操作导致的 statement 生命周期问题 - */ export const saveConversationsWithRelatedCache = async ( - conversations: ConversationResponse[], - users?: UserDTO[], - groups?: GroupResponse[] + convs: ConversationResponse[], users?: UserDTO[], groups?: GroupResponse[] ): Promise => { - if (!conversations || conversations.length === 0) return; - + if (!convs?.length) return; const now = new Date().toISOString(); - - await enqueueWrite(async () => { - const database = await getDb(); - - // 1. 保存会话列表 - for (const conversation of conversations) { - await database.runAsync( - `INSERT OR REPLACE INTO conversation_list_cache (id, data, updatedAt) VALUES (?, ?, ?)`, - [ - String(conversation.id), - JSON.stringify(conversation), - conversation.updated_at || now, - ] - ); + await enqueueWrite(async (db) => { + for (const c of convs) { + await db.runAsync(`INSERT OR REPLACE INTO conversation_list_cache (id, data, updatedAt) VALUES (?, ?, ?)`, + [String(c.id), JSON.stringify(c), c.updated_at || now]); } - - // 2. 保存用户 - if (users && users.length > 0) { - for (const user of users) { - await database.runAsync( - `INSERT OR REPLACE INTO users (id, data, updatedAt) VALUES (?, ?, ?)`, - [String(user.id), JSON.stringify(user), now] - ); + if (users?.length) { + for (const u of users) { + await db.runAsync(`INSERT OR REPLACE INTO users (id, data, updatedAt) VALUES (?, ?, ?)`, + [String(u.id), JSON.stringify(u), now]); } } - - // 3. 保存群组 - if (groups && groups.length > 0) { - for (const group of groups) { - await database.runAsync( - `INSERT OR REPLACE INTO groups (id, data, updatedAt) VALUES (?, ?, ?)`, - [String(group.id), JSON.stringify(group), now] - ); + if (groups?.length) { + for (const g of groups) { + await db.runAsync(`INSERT OR REPLACE INTO groups (id, data, updatedAt) VALUES (?, ?, ?)`, + [String(g.id), JSON.stringify(g), now]); } } }); -}; +}; \ No newline at end of file diff --git a/src/stores/authStore.ts b/src/stores/authStore.ts index a6398a6..687aceb 100644 --- a/src/stores/authStore.ts +++ b/src/stores/authStore.ts @@ -258,11 +258,11 @@ export const useAuthStore = create((set) => ({ const savedUserId = await loadUserId(); if (savedUserId) { - // 2. 用已知的 userId 提前初始化 DB + // 2. 用已知的 userId 提前初始化 DB（数据库模块内部已处理重试） try { await initDatabase(savedUserId); } catch (dbErr) { - console.warn('[AuthStore] DB 预初始化失败:', dbErr); + console.warn('[AuthStore] DB 预初始化失败，将尝试在 API 成功后重新初始化:', dbErr); } } diff --git a/tsconfig.json b/tsconfig.json index 79991c0..f17d546 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -2,9 +2,8 @@ "extends": "expo/tsconfig.base", "compilerOptions": { "strict": true, - "baseUrl": ".", "paths": { - "@/*": ["src/*"] + "@/*": ["./src/*"] }, "skipLibCheck": true, "moduleResolution": "bundler",