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",