Files
frontend/src/database/repositories/ConversationRepository.ts

197 lines
8.4 KiB
TypeScript
Raw Normal View History

import { localDataSource } from '../LocalDataSource';
import type { ILocalDataSource } from '@/data/datasources/interfaces';
import type { ConversationResponse, ConversationDetailResponse } from '@/types/dto';
type ConversationCacheData = ConversationResponse | ConversationDetailResponse;
const safeParseJson = <T>(value: string): T | null => {
try { return JSON.parse(value) as T; } catch { return null; }
};
export interface IConversationRepository {
save(conv: { id: string; participantId: string; lastMessageId?: string; lastSeq?: number; unreadCount?: number; createdAt: string; updatedAt: string }): Promise<void>;
getAll(): Promise<any[]>;
updateUnreadCount(id: string, count: number): Promise<void>;
updateLastMessage(id: string, lastMessageId: string, lastSeq: number, updatedAt: string): Promise<void>;
delete(conversationId: string): Promise<void>;
getCache(id: string): Promise<ConversationCacheData | null>;
getListCache(): Promise<ConversationResponse[]>;
saveCache(conv: ConversationCacheData): Promise<void>;
saveListCache(convs: ConversationResponse[]): Promise<void>;
updateListEntry(id: string, updates: Partial<ConversationResponse>): Promise<void>;
updateCacheUnreadCount(id: string, count: number, myLastReadSeq?: number): Promise<void>;
saveWithRelated(convs: ConversationResponse[], users?: any[], groups?: any[]): Promise<void>;
}
export class ConversationRepository implements IConversationRepository {
constructor(private dataSource: ILocalDataSource = localDataSource) {}
async save(conv: { id: string; participantId: string; lastMessageId?: string; lastSeq?: number; unreadCount?: number; createdAt: string; updatedAt: string }): Promise<void> {
await this.dataSource.enqueueWrite(async () => {
await this.dataSource.run(
`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]
);
});
}
async getAll(): Promise<any[]> {
return this.dataSource.query<any>(`SELECT * FROM conversations ORDER BY updatedAt DESC`);
}
async updateUnreadCount(id: string, count: number): Promise<void> {
await this.dataSource.enqueueWrite(async () => {
await this.dataSource.run(`UPDATE conversations SET unreadCount = ? WHERE id = ?`, [count, id]);
});
}
async updateLastMessage(id: string, lastMessageId: string, lastSeq: number, updatedAt: string): Promise<void> {
await this.dataSource.enqueueWrite(async () => {
await this.dataSource.run(
`UPDATE conversations SET lastMessageId = ?, lastSeq = ?, updatedAt = ? WHERE id = ?`,
[lastMessageId, lastSeq, updatedAt, id]
);
});
}
async delete(conversationId: string): Promise<void> {
await this.dataSource.enqueueWrite(async () => {
await this.dataSource.run(`DELETE FROM messages WHERE conversationId = ?`, [conversationId]);
await this.dataSource.run(`DELETE FROM conversations WHERE id = ?`, [conversationId]);
});
}
async getCache(id: string): Promise<ConversationCacheData | null> {
const r = await this.dataSource.getFirst<{ data: string }>(
`SELECT data FROM conversation_cache WHERE id = ?`, [String(id)]
);
return r?.data ? safeParseJson<ConversationCacheData>(r.data) : null;
}
async getListCache(): Promise<ConversationResponse[]> {
let rows = await this.dataSource.query<{ data: string }>(
`SELECT data FROM conversation_list_cache ORDER BY updatedAt DESC`
);
if (!rows.length) {
rows = await this.dataSource.query<{ 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;
});
}
async saveCache(conv: ConversationCacheData): Promise<void> {
await this.dataSource.enqueueWrite(async () => {
const updatedAt = ('updated_at' in conv && conv.updated_at) ? String(conv.updated_at) : new Date().toISOString();
await this.dataSource.run(
`INSERT OR REPLACE INTO conversation_cache (id, data, updatedAt) VALUES (?, ?, ?)`,
[String(conv.id), JSON.stringify(conv), updatedAt]
);
});
}
async saveListCache(convs: ConversationResponse[]): Promise<void> {
if (!convs?.length) return;
await this.dataSource.enqueueWrite(async () => {
for (const c of convs) {
await this.dataSource.run(
`INSERT OR REPLACE INTO conversation_list_cache (id, data, updatedAt) VALUES (?, ?, ?)`,
[String(c.id), JSON.stringify(c), c.updated_at || new Date().toISOString()]
);
}
});
}
async updateListEntry(id: string, updates: Partial<ConversationResponse>): Promise<void> {
await this.dataSource.enqueueWrite(async () => {
const r = await this.dataSource.getFirst<{ 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 this.dataSource.run(
`UPDATE conversation_list_cache SET data = ?, updatedAt = ? WHERE id = ?`,
[JSON.stringify(updated), updatedAt, String(id)]
);
});
}
async updateCacheUnreadCount(id: string, count: number, myLastReadSeq?: number): Promise<void> {
await this.dataSource.enqueueWrite(async () => {
const listRow = await this.dataSource.getFirst<{ 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 this.dataSource.run(
`UPDATE conversation_list_cache SET data = ? WHERE id = ?`,
[JSON.stringify(conv), String(id)]
);
}
}
const cacheRow = await this.dataSource.getFirst<{ 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 this.dataSource.run(
`UPDATE conversation_cache SET data = ? WHERE id = ?`,
[JSON.stringify(conv), String(id)]
);
}
}
});
}
async saveWithRelated(convs: ConversationResponse[], users?: any[], groups?: any[]): Promise<void> {
if (!convs?.length) return;
const now = new Date().toISOString();
await this.dataSource.enqueueWrite(async () => {
for (const c of convs) {
await this.dataSource.run(
`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 this.dataSource.run(
`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 this.dataSource.run(
`INSERT OR REPLACE INTO groups (id, data, updatedAt) VALUES (?, ?, ?)`,
[String(g.id), JSON.stringify(g), now]
);
}
}
});
}
}
export const conversationRepository = new ConversationRepository();