refactor(database): migrate to new modular database layer and unify data access
- Remove legacy database.ts, LocalDataSource.ts, and MessageRepository.ts - Create new src/database/ module with messageRepository, userCacheRepository, conversationRepository, and groupCacheRepository - Update all consumers to import from @/database instead of services/database - Add web platform blur handling for modal components to fix focus issues - Flatten SystemMessageItem and NotificationsScreen styles for consistent design - Add draggable slider in ChatSettingsScreen and dynamic font size support - Introduce 9 new chat color themes - Add profile screens for about, terms, and privacy policy with navigation routes - Add policy links to login and registration screens - Fix post share URL format from /posts/ to /post/
This commit is contained in:
lafay
@@ -1,767 +0,0 @@
|
||||
/**
|
||||
* 本地数据库服务
|
||||
* 使用 SQLite 存储聊天记录
|
||||
* 每个用户使用独立的数据库文件，避免数据串用
|
||||
*
|
||||
* Web/OPFS 关键设计：
|
||||
* - expo-sqlite 在 Web 上使用 OPFS Worker，文件句柄严格唯一
|
||||
* - 数据库打开后永不关闭（直到用户切换或登出）
|
||||
* - 使用全局单例 + 互斥锁，避免并发打开
|
||||
* - React Strict Mode 双次 useEffect 会被幂等处理
|
||||
*/
|
||||
|
||||
import * as SQLite from 'expo-sqlite';
|
||||
import type {
|
||||
UserDTO,
|
||||
GroupResponse,
|
||||
GroupMemberResponse,
|
||||
ConversationResponse,
|
||||
ConversationDetailResponse,
|
||||
} from '../types/dto';
|
||||
|
||||
let db: SQLite.SQLiteDatabase | null = null;
|
||||
let currentDbUserId: string | null = null;
|
||||
let writeQueue: Promise<void> = Promise.resolve();
|
||||
|
||||
let initPromise: Promise<void> | null = null;
|
||||
let initTargetUserId: string | null = null;
|
||||
let isInitialized = false;
|
||||
|
||||
let dbMutex: Promise<void> = Promise.resolve();
|
||||
|
||||
const ensureInitialized = async (): Promise<void> => {
|
||||
if (isInitialized && db) return;
|
||||
if (initPromise) {
|
||||
await initPromise;
|
||||
return;
|
||||
}
|
||||
throw new Error('数据库未初始化');
|
||||
};
|
||||
|
||||
async function withMutex<T>(fn: () => Promise<T>): Promise<T> {
|
||||
const prev = dbMutex;
|
||||
let release: () => void = () => {};
|
||||
dbMutex = new Promise<void>((r) => { release = r; });
|
||||
try {
|
||||
await prev;
|
||||
return await fn();
|
||||
} finally {
|
||||
release();
|
||||
}
|
||||
}
|
||||
|
||||
async function openDatabaseWithRetry(dbName: string, maxRetries: number = 3): Promise<SQLite.SQLiteDatabase> {
|
||||
const isWeb = typeof window !== 'undefined';
|
||||
|
||||
if (isWeb) {
|
||||
try {
|
||||
const tempDb = await SQLite.openDatabaseAsync(dbName);
|
||||
await tempDb.closeAsync();
|
||||
console.log('[Database] Web: 预打开成功，已释放可能残留的句柄');
|
||||
} catch (e) {
|
||||
console.warn('[Database] Web: 预打开失败:', String(e));
|
||||
}
|
||||
await new Promise(r => setTimeout(r, 50));
|
||||
}
|
||||
|
||||
for (let attempt = 0; attempt <= maxRetries; attempt++) {
|
||||
try {
|
||||
const database = await SQLite.openDatabaseAsync(dbName);
|
||||
return database;
|
||||
} catch (error) {
|
||||
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')) {
|
||||
try {
|
||||
await SQLite.deleteDatabaseAsync(dbName);
|
||||
console.warn('[Database] 已删除损坏的数据库文件');
|
||||
} catch {}
|
||||
}
|
||||
} else {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
throw new Error('打开数据库失败');
|
||||
}
|
||||
|
||||
async function createTables(database: SQLite.SQLiteDatabase): Promise<void> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
// 拆分为单独的 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
|
||||
)`);
|
||||
|
||||
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
|
||||
)`);
|
||||
|
||||
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<void> {
|
||||
try {
|
||||
const tableInfo = await database.getAllAsync<any>(`PRAGMA table_info(messages)`);
|
||||
const columns = tableInfo.map(col => col.name);
|
||||
|
||||
if (!columns.includes('seq')) {
|
||||
await database.execAsync(`ALTER TABLE messages ADD COLUMN seq INTEGER DEFAULT 0`);
|
||||
}
|
||||
if (!columns.includes('status')) {
|
||||
await database.execAsync(`ALTER TABLE messages ADD COLUMN status TEXT DEFAULT 'normal'`);
|
||||
}
|
||||
if (!columns.includes('segments')) {
|
||||
await database.execAsync(`ALTER TABLE messages ADD COLUMN segments TEXT`);
|
||||
}
|
||||
|
||||
const convTableInfo = await database.getAllAsync<any>(`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('[Database] 迁移失败:', error);
|
||||
}
|
||||
}
|
||||
|
||||
export const initDatabase = async (userId?: string): Promise<void> => {
|
||||
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;
|
||||
}
|
||||
|
||||
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) {
|
||||
console.log(`[Database ${mutexId}] 关闭旧数据库连接...`);
|
||||
try {
|
||||
await db.closeAsync();
|
||||
console.log(`[Database ${mutexId}] 旧连接已关闭`);
|
||||
} catch (e) {
|
||||
console.warn(`[Database ${mutexId}] 关闭旧连接失败:`, e);
|
||||
}
|
||||
db = null;
|
||||
currentDbUserId = null;
|
||||
isInitialized = false;
|
||||
|
||||
if (isWeb) {
|
||||
await new Promise(r => setTimeout(r, 100));
|
||||
}
|
||||
}
|
||||
|
||||
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<void> => {
|
||||
if (initPromise) {
|
||||
await initPromise.catch(() => {});
|
||||
}
|
||||
|
||||
await withMutex(async () => {
|
||||
if (!db) return;
|
||||
|
||||
try {
|
||||
await db.closeAsync();
|
||||
console.log('[Database] 数据库已关闭');
|
||||
} catch (e) {
|
||||
console.warn('[Database] 关闭失败:', e);
|
||||
}
|
||||
db = null;
|
||||
currentDbUserId = null;
|
||||
isInitialized = false;
|
||||
});
|
||||
};
|
||||
|
||||
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<SQLite.SQLiteDatabase> => {
|
||||
if (isInitialized && db) return db;
|
||||
|
||||
if (initPromise) {
|
||||
const timeoutPromise = new Promise<null>((_, reject) =>
|
||||
setTimeout(() => reject(new Error('等待数据库初始化超时')), timeout)
|
||||
);
|
||||
|
||||
await Promise.race([initPromise, timeoutPromise]);
|
||||
|
||||
if (db) return db;
|
||||
}
|
||||
|
||||
throw new Error('数据库未初始化，请先调用 initDatabase(userId)');
|
||||
};
|
||||
|
||||
const enqueueWrite = async <T>(op: (db: SQLite.SQLiteDatabase) => Promise<T>): Promise<T> => {
|
||||
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 <T>(op: (db: SQLite.SQLiteDatabase) => Promise<T>): Promise<T> => {
|
||||
return withMutex(async () => {
|
||||
const database = await getDb();
|
||||
return op(database);
|
||||
});
|
||||
};
|
||||
|
||||
export interface CachedMessage {
|
||||
id: string;
|
||||
conversationId: string;
|
||||
senderId: string;
|
||||
content?: string;
|
||||
type?: string;
|
||||
isRead: boolean;
|
||||
createdAt: string;
|
||||
seq: number;
|
||||
status: string;
|
||||
segments?: any;
|
||||
}
|
||||
|
||||
export const saveMessage = async (message: CachedMessage): Promise<void> => {
|
||||
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]
|
||||
);
|
||||
});
|
||||
};
|
||||
|
||||
export const saveMessagesBatch = async (messages: CachedMessage[]): Promise<void> => {
|
||||
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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
[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]
|
||||
);
|
||||
}
|
||||
});
|
||||
};
|
||||
|
||||
export const getMessagesByConversation = async (conversationId: string, limit: number = 20): Promise<CachedMessage[]> => {
|
||||
return withDbRead(async (database) => {
|
||||
const rows = await database.getAllAsync<any>(
|
||||
`SELECT * FROM messages WHERE conversationId = ? ORDER BY seq DESC LIMIT ?`,
|
||||
[conversationId, limit]
|
||||
);
|
||||
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<number> => {
|
||||
return withDbRead(async (database) => {
|
||||
const r = await database.getFirstAsync<{ maxSeq: number }>(
|
||||
`SELECT MAX(seq) as maxSeq FROM messages WHERE conversationId = ?`, [conversationId]
|
||||
);
|
||||
return r?.maxSeq || 0;
|
||||
});
|
||||
};
|
||||
|
||||
export const getMinSeq = async (conversationId: string): Promise<number> => {
|
||||
return withDbRead(async (database) => {
|
||||
const r = await database.getFirstAsync<{ minSeq: number }>(
|
||||
`SELECT MIN(seq) as minSeq FROM messages WHERE conversationId = ?`, [conversationId]
|
||||
);
|
||||
return r?.minSeq || 0;
|
||||
});
|
||||
};
|
||||
|
||||
export const getMessagesBeforeSeq = async (conversationId: string, beforeSeq: number, limit: number = 20): Promise<CachedMessage[]> => {
|
||||
return withDbRead(async (database) => {
|
||||
const rows = await database.getAllAsync<any>(
|
||||
`SELECT * FROM messages WHERE conversationId = ? AND seq < ? ORDER BY seq DESC LIMIT ?`,
|
||||
[conversationId, beforeSeq, limit]
|
||||
);
|
||||
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<number> => {
|
||||
return withDbRead(async (database) => {
|
||||
const r = await database.getFirstAsync<{ count: number }>(
|
||||
`SELECT COUNT(*) as count FROM messages WHERE conversationId = ? AND seq < ?`, [conversationId, beforeSeq]
|
||||
);
|
||||
return r?.count || 0;
|
||||
});
|
||||
};
|
||||
|
||||
export const getMessageCount = async (conversationId: string): Promise<number> => {
|
||||
return withDbRead(async (database) => {
|
||||
const r = await database.getFirstAsync<{ count: number }>(
|
||||
`SELECT COUNT(*) as count FROM messages WHERE conversationId = ?`, [conversationId]
|
||||
);
|
||||
return r?.count || 0;
|
||||
});
|
||||
};
|
||||
|
||||
export const markMessageAsRead = async (messageId: string): Promise<void> => {
|
||||
await enqueueWrite(async (db) => {
|
||||
await db.runAsync(`UPDATE messages SET isRead = 1 WHERE id = ?`, [messageId]);
|
||||
});
|
||||
};
|
||||
|
||||
export const markConversationAsRead = async (conversationId: string): Promise<void> => {
|
||||
await enqueueWrite(async (db) => {
|
||||
await db.runAsync(`UPDATE messages SET isRead = 1 WHERE conversationId = ?`, [conversationId]);
|
||||
});
|
||||
};
|
||||
|
||||
export const deleteMessage = async (messageId: string): Promise<void> => {
|
||||
await enqueueWrite(async (db) => {
|
||||
await db.runAsync(`DELETE FROM messages WHERE id = ?`, [messageId]);
|
||||
});
|
||||
};
|
||||
|
||||
export const updateMessageStatus = async (messageId: string, status: string, clearContent: boolean = false): Promise<void> => {
|
||||
await enqueueWrite(async (db) => {
|
||||
if (clearContent) {
|
||||
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]);
|
||||
}
|
||||
});
|
||||
};
|
||||
|
||||
export const clearConversationMessages = async (conversationId: string): Promise<void> => {
|
||||
await enqueueWrite(async (db) => {
|
||||
await db.runAsync(`DELETE FROM messages WHERE conversationId = ?`, [conversationId]);
|
||||
});
|
||||
};
|
||||
|
||||
export const saveConversation = async (conv: {
|
||||
id: string; participantId: string; lastMessageId?: string; lastSeq?: number; unreadCount?: number;
|
||||
createdAt: string; updatedAt: string;
|
||||
}): Promise<void> => {
|
||||
await enqueueWrite(async (db) => {
|
||||
await db.runAsync(
|
||||
`INSERT OR REPLACE INTO conversations (id, participantId, lastMessageId, lastSeq, unreadCount, createdAt, updatedAt)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)`,
|
||||
[conv.id, conv.participantId, conv.lastMessageId || null, conv.lastSeq || 0, conv.unreadCount || 0, conv.createdAt, conv.updatedAt]
|
||||
);
|
||||
});
|
||||
};
|
||||
|
||||
export const getAllConversations = async (): Promise<any[]> => {
|
||||
return withDbRead(async (db) => db.getAllAsync<any>(`SELECT * FROM conversations ORDER BY updatedAt DESC`));
|
||||
};
|
||||
|
||||
export const updateConversationUnreadCount = async (id: string, count: number): Promise<void> => {
|
||||
await enqueueWrite(async (db) => {
|
||||
await db.runAsync(`UPDATE conversations SET unreadCount = ? WHERE id = ?`, [count, id]);
|
||||
});
|
||||
};
|
||||
|
||||
export const updateConversationLastMessage = async (id: string, lastMessageId: string, lastSeq: number, updatedAt: string): Promise<void> => {
|
||||
await enqueueWrite(async (db) => {
|
||||
await db.runAsync(`UPDATE conversations SET lastMessageId = ?, lastSeq = ?, updatedAt = ? WHERE id = ?`,
|
||||
[lastMessageId, lastSeq, updatedAt, id]);
|
||||
});
|
||||
};
|
||||
|
||||
export const deleteConversation = async (conversationId: string): Promise<void> => {
|
||||
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<CachedMessage[]> => {
|
||||
return withDbRead(async (db) => {
|
||||
const rows = await db.getAllAsync<any>(
|
||||
`SELECT * FROM messages WHERE content LIKE ? ORDER BY createdAt DESC LIMIT 50`, [`%${keyword}%`]
|
||||
);
|
||||
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 (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<void> => {
|
||||
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`);
|
||||
});
|
||||
};
|
||||
|
||||
const safeParseJson = <T>(value: string): T | null => {
|
||||
try { return JSON.parse(value) as T; } catch { return null; }
|
||||
};
|
||||
|
||||
export const saveUserCache = async (user: UserDTO): Promise<void> => {
|
||||
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<void> => {
|
||||
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<UserDTO | null> => {
|
||||
return withDbRead(async (db) => {
|
||||
const r = await db.getFirstAsync<{ data: string }>(`SELECT data FROM users WHERE id = ?`, [String(userId)]);
|
||||
return r?.data ? safeParseJson<UserDTO>(r.data) : null;
|
||||
});
|
||||
};
|
||||
|
||||
export const getLatestUserCache = async (): Promise<UserDTO | null> => {
|
||||
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<UserDTO>(r.data) : null;
|
||||
});
|
||||
};
|
||||
|
||||
export const saveCurrentUserCache = async (user: UserDTO): Promise<void> => {
|
||||
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<UserDTO | null> => {
|
||||
return withDbRead(async (db) => {
|
||||
const r = await db.getFirstAsync<{ data: string }>(`SELECT data FROM current_user_cache WHERE id = 'me'`);
|
||||
return r?.data ? safeParseJson<UserDTO>(r.data) : null;
|
||||
});
|
||||
};
|
||||
|
||||
export const clearCurrentUserCache = async (): Promise<void> => {
|
||||
await enqueueWrite(async (db) => {
|
||||
await db.runAsync(`DELETE FROM current_user_cache WHERE id = 'me'`);
|
||||
});
|
||||
};
|
||||
|
||||
export const saveGroupCache = async (group: GroupResponse): Promise<void> => {
|
||||
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<void> => {
|
||||
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<GroupResponse | null> => {
|
||||
return withDbRead(async (db) => {
|
||||
const r = await db.getFirstAsync<{ data: string }>(`SELECT data FROM groups WHERE id = ?`, [String(groupId)]);
|
||||
return r?.data ? safeParseJson<GroupResponse>(r.data) : null;
|
||||
});
|
||||
};
|
||||
|
||||
export const getAllGroupsCache = async (): Promise<GroupResponse[]> => {
|
||||
return withDbRead(async (db) => {
|
||||
const rows = await db.getAllAsync<{ data: string }>(`SELECT data FROM groups ORDER BY updatedAt DESC`);
|
||||
return rows.map(r => safeParseJson<GroupResponse>(r.data)).filter((g): g is GroupResponse => Boolean(g));
|
||||
});
|
||||
};
|
||||
|
||||
export const saveGroupMembersCache = async (groupId: string, members: GroupMemberResponse[]): Promise<void> => {
|
||||
await enqueueWrite(async (db) => {
|
||||
const now = new Date().toISOString();
|
||||
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<GroupMemberResponse[]> => {
|
||||
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(r => safeParseJson<GroupMemberResponse>(r.data)).filter((m): m is GroupMemberResponse => Boolean(m));
|
||||
});
|
||||
};
|
||||
|
||||
type ConversationCacheData = ConversationResponse | ConversationDetailResponse;
|
||||
|
||||
export const saveConversationCache = async (conv: ConversationCacheData): Promise<void> => {
|
||||
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 (convs: ConversationResponse[]): Promise<void> => {
|
||||
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 (id: string): Promise<ConversationCacheData | null> => {
|
||||
return withDbRead(async (db) => {
|
||||
const r = await db.getFirstAsync<{ data: string }>(`SELECT data FROM conversation_cache WHERE id = ?`, [String(id)]);
|
||||
return r?.data ? safeParseJson<ConversationCacheData>(r.data) : null;
|
||||
});
|
||||
};
|
||||
|
||||
export const getConversationListCache = async (): Promise<ConversationResponse[]> => {
|
||||
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(r => safeParseJson<ConversationResponse>(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;
|
||||
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;
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
export const updateConversationListCacheEntry = async (id: string, updates: Partial<ConversationResponse>): Promise<void> => {
|
||||
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<ConversationResponse>(r.data);
|
||||
if (!conv) return;
|
||||
const updated = { ...conv, ...updates };
|
||||
const updatedAt = updates.last_message_at || updated.last_message_at || new Date().toISOString();
|
||||
await db.runAsync(`UPDATE conversation_list_cache SET data = ?, updatedAt = ? WHERE id = ?`,
|
||||
[JSON.stringify(updated), updatedAt, String(id)]);
|
||||
});
|
||||
};
|
||||
|
||||
export const updateConversationCacheUnreadCount = async (id: string, count: number, myLastReadSeq?: number): Promise<void> => {
|
||||
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<ConversationResponse>(listRow.data);
|
||||
if (conv) {
|
||||
conv.unread_count = count;
|
||||
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)]);
|
||||
}
|
||||
}
|
||||
|
||||
const cacheRow = await db.getFirstAsync<{ data: string }>(`SELECT data FROM conversation_cache WHERE id = ?`, [String(id)]);
|
||||
if (cacheRow?.data) {
|
||||
const conv = safeParseJson<Record<string, unknown>>(cacheRow.data);
|
||||
if (conv) {
|
||||
conv.unread_count = count;
|
||||
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)]);
|
||||
}
|
||||
}
|
||||
});
|
||||
};
|
||||
|
||||
export const saveConversationsWithRelatedCache = async (
|
||||
convs: ConversationResponse[], users?: UserDTO[], groups?: GroupResponse[]
|
||||
): Promise<void> => {
|
||||
if (!convs?.length) return;
|
||||
const now = new Date().toISOString();
|
||||
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]);
|
||||
}
|
||||
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]);
|
||||
}
|
||||
}
|
||||
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]);
|
||||
}
|
||||
}
|
||||
});
|
||||
};
|
||||
@@ -22,14 +22,7 @@ import {
|
||||
CursorPaginationRequest,
|
||||
CursorPaginationResponse,
|
||||
} from '../types/dto';
|
||||
import {
|
||||
getConversationCache,
|
||||
getConversationListCache,
|
||||
saveConversationCache,
|
||||
saveConversationsWithRelatedCache,
|
||||
updateConversationCacheUnreadCount,
|
||||
saveUsersCache,
|
||||
} from './database';
|
||||
import { conversationRepository, userCacheRepository } from '@/database';
|
||||
|
||||
/** 远端会话列表分页（offset / cursor）统一结果：拉取成功后均已执行本地缓存同步 */
|
||||
export interface RemoteConversationListPageResult {
|
||||
@@ -55,7 +48,7 @@ class MessageService {
|
||||
const groups = list
|
||||
.map(conv => conv.group)
|
||||
.filter((group): group is NonNullable<typeof group> => Boolean(group));
|
||||
await saveConversationsWithRelatedCache(list, users, groups);
|
||||
await conversationRepository.saveWithRelated(list, users, groups);
|
||||
}
|
||||
|
||||
/** offset 会话列表单页：请求 + 落库 */
|
||||
@@ -152,12 +145,12 @@ class MessageService {
|
||||
const response = await api.get<ConversationDetailResponse>(
|
||||
`/conversations/${encodeURIComponent(id)}`
|
||||
);
|
||||
await saveConversationCache(response.data);
|
||||
await conversationRepository.saveCache(response.data);
|
||||
if (response.data.participants?.length) {
|
||||
await saveUsersCache(response.data.participants);
|
||||
await userCacheRepository.saveBatch(response.data.participants);
|
||||
}
|
||||
if (response.data.last_message?.sender) {
|
||||
await saveUsersCache([response.data.last_message.sender]);
|
||||
await userCacheRepository.save(response.data.last_message.sender);
|
||||
}
|
||||
return response.data;
|
||||
}
|
||||
@@ -195,7 +188,7 @@ class MessageService {
|
||||
): Promise<ConversationListResponse> {
|
||||
// 如果强制刷新,直接从服务器获取,不使用缓存
|
||||
if (!forceRefresh) {
|
||||
const cachedList = await getConversationListCache();
|
||||
const cachedList = await conversationRepository.getListCache();
|
||||
if (cachedList.length > 0) {
|
||||
// 后台刷新数据,但不阻塞当前返回
|
||||
this.refreshConversationsFromServer(page, pageSize).catch(error => {
|
||||
@@ -235,9 +228,9 @@ class MessageService {
|
||||
const response = await api.post<ConversationResponse>('/conversations', {
|
||||
user_id: userId,
|
||||
});
|
||||
await saveConversationCache(response.data);
|
||||
await conversationRepository.saveCache(response.data);
|
||||
if (response.data.participants?.length) {
|
||||
await saveUsersCache(response.data.participants);
|
||||
await userCacheRepository.saveBatch(response.data.participants);
|
||||
}
|
||||
return response.data;
|
||||
}
|
||||
@@ -247,7 +240,7 @@ class MessageService {
|
||||
* GET /api/v1/conversations/:id
|
||||
*/
|
||||
async getConversationById(id: string): Promise<ConversationDetailResponse> {
|
||||
const cached = await getConversationCache(id);
|
||||
const cached = await conversationRepository.getCache(id);
|
||||
if (cached) {
|
||||
this.refreshConversationDetailFromServer(id).catch(error => {
|
||||
console.error('后台刷新会话详情失败:', error);
|
||||
@@ -564,7 +557,7 @@ class MessageService {
|
||||
last_read_seq: Number(seq),
|
||||
});
|
||||
// 立即清零本地缓存中的 unread_count，并写入已读游标，避免冷启动仍显示旧红点
|
||||
updateConversationCacheUnreadCount(conversationId, 0, Number(seq)).catch(error => {
|
||||
conversationRepository.updateCacheUnreadCount(conversationId, 0, Number(seq)).catch(error => {
|
||||
console.error('更新本地会话未读数失败:', error);
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user