import Database from 'better-sqlite3' import { paths } from './paths' import type { MessageBody, MessageType } from './protocol' import { existsSync, mkdirSync } from 'node:fs' import { dirname } from 'node:path' if (!existsSync(dirname(paths.db))) mkdirSync(dirname(paths.db), { recursive: true }) export const db: Database.Database = new Database(paths.db) db.pragma('journal_mode = WAL') db.pragma('foreign_keys = ON') // 设备: 持久化最近一次见到的设备信息, 便于离线后回看 db.exec(` CREATE TABLE IF NOT EXISTS devices ( device_id TEXT PRIMARY KEY, name TEXT NOT NULL, hostname TEXT, platform TEXT, app_version TEXT, last_ip TEXT, last_seen INTEGER NOT NULL, ignored INTEGER NOT NULL DEFAULT 0, drive_token TEXT ); CREATE TABLE IF NOT EXISTS messages ( message_id TEXT PRIMARY KEY, type TEXT NOT NULL, from_id TEXT NOT NULL, to_id TEXT NOT NULL, ts INTEGER NOT NULL, body_json TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'delivered', -- 引用方向, 用于查询某个会话 (1对1) conversation_key TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_messages_conv_ts ON messages (conversation_key, ts); CREATE TABLE IF NOT EXISTS unread ( device_id TEXT PRIMARY KEY, count INTEGER NOT NULL DEFAULT 0 ); -- 远程操作审计 (terminal / forward / serial) CREATE TABLE IF NOT EXISTS audit_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts INTEGER NOT NULL, action TEXT NOT NULL, source_device TEXT, target_device TEXT, session_id TEXT, payload_json TEXT, result TEXT NOT NULL, bytes_in INTEGER NOT NULL DEFAULT 0, bytes_out INTEGER NOT NULL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_audit_ts ON audit_log(ts DESC); CREATE INDEX IF NOT EXISTS idx_audit_session ON audit_log(session_id); `) // 迁移: 给老库补 ignored 列 (用 try/catch 包住, 已存在会抛) try { db.exec(`ALTER TABLE devices ADD COLUMN ignored INTEGER NOT NULL DEFAULT 0`) } catch {} // 迁移: 给老库补 drive_token 列 (网盘访问令牌, 老版本没有) try { db.exec(`ALTER TABLE devices ADD COLUMN drive_token TEXT`) } catch {} // conversation_key: 任意两个 deviceId 排序后 join, 1对1 export function conversationKey(a: string, b: string): string { return [a, b].sort().join('::') } // 设备 export interface DeviceRow { device_id: string name: string hostname: string | null platform: string | null app_version: string | null last_ip: string | null last_seen: number ignored: number // 0/1 drive_token: string | null // 网盘访问令牌 (对端用它访问我的网盘) } const upsertDeviceStmt = db.prepare(` INSERT INTO devices (device_id, name, hostname, platform, app_version, last_ip, last_seen, drive_token) VALUES (@device_id, @name, @hostname, @platform, @app_version, @last_ip, @last_seen, @drive_token) ON CONFLICT(device_id) DO UPDATE SET name = excluded.name, hostname = excluded.hostname, platform = excluded.platform, app_version = excluded.app_version, last_ip = excluded.last_ip, last_seen = excluded.last_seen, drive_token = excluded.drive_token `) export function upsertDevice(d: DeviceRow) { upsertDeviceStmt.run(d) } export function listDevices(includeIgnored = false): DeviceRow[] { const sql = includeIgnored ? `SELECT * FROM devices ORDER BY last_seen DESC` : `SELECT * FROM devices WHERE ignored = 0 ORDER BY last_seen DESC` return db.prepare(sql).all() as DeviceRow[] } export function getDevice(id: string): DeviceRow | undefined { return db.prepare(`SELECT * FROM devices WHERE device_id = ?`).get(id) as DeviceRow | undefined } export function setDeviceIgnored(deviceId: string, ignored: boolean) { db.prepare(`UPDATE devices SET ignored = ? WHERE device_id = ?`).run(ignored ? 1 : 0, deviceId) } export function listIgnoredDevices(): DeviceRow[] { return db.prepare(`SELECT * FROM devices WHERE ignored = 1 ORDER BY last_seen DESC`).all() as DeviceRow[] } // 删除设备本地记录 (DB 行)。下次 UDP announce 会作为新设备重新入 export function deleteDevice(deviceId: string): boolean { const r = db.prepare(`DELETE FROM devices WHERE device_id = ?`).run(deviceId) db.prepare(`DELETE FROM unread WHERE device_id = ?`).run(deviceId) return r.changes > 0 } export function deleteConversation(selfId: string, peerId: string): number { const key = conversationKey(selfId, peerId) const r = db.prepare(`DELETE FROM messages WHERE conversation_key = ?`).run(key) db.prepare(`DELETE FROM unread WHERE device_id = ?`).run(peerId) return r.changes } // 消息 export interface MessageRow { message_id: string type: MessageType from_id: string to_id: string ts: number body_json: string status: string conversation_key: string } const insertMsgStmt = db.prepare(` INSERT OR REPLACE INTO messages (message_id, type, from_id, to_id, ts, body_json, status, conversation_key) VALUES (@message_id, @type, @from_id, @to_id, @ts, @body_json, @status, @conversation_key) `) export function insertMessage(args: { messageId: string type: MessageType fromId: string toId: string ts: number body: MessageBody status?: string }) { const row: MessageRow = { message_id: args.messageId, type: args.type, from_id: args.fromId, to_id: args.toId, ts: args.ts, body_json: JSON.stringify(args.body), status: args.status ?? 'delivered', conversation_key: conversationKey(args.fromId, args.toId), } insertMsgStmt.run(row) return row } export function getMessage(id: string): MessageRow | undefined { return db.prepare(`SELECT * FROM messages WHERE message_id = ?`).get(id) as MessageRow | undefined } export function listMessages(peerId: string, selfId: string, limit = 200, before?: number): MessageRow[] { const key = conversationKey(selfId, peerId) if (before) { return db.prepare( `SELECT * FROM messages WHERE conversation_key = ? AND ts < ? ORDER BY ts DESC LIMIT ?` ).all(key, before, limit).reverse() as MessageRow[] } return db.prepare( `SELECT * FROM messages WHERE conversation_key = ? ORDER BY ts DESC LIMIT ?` ).all(key, limit).reverse() as MessageRow[] } // 待发送 (离线时缓存, 上线后由 flushPendingFor 触发重发) export function listPendingMessages(peerId: string, selfId: string): MessageRow[] { const key = conversationKey(selfId, peerId) return db.prepare( `SELECT * FROM messages WHERE conversation_key = ? AND status = 'pending' ORDER BY ts ASC` ).all(key) as MessageRow[] } // 已上传完但 WS update 帧 (带 savedPath) 没送达: metadata 帧 status='sent', // body.savedPath 已填, 重发完整 envelope 让收件方出 "打开/位置" export function listSentFileUpdates(selfId: string, peerId: string): MessageRow[] { return db.prepare( `SELECT * FROM messages WHERE from_id = ? AND to_id = ? AND status = 'sent' AND type IN ('file', 'image') AND json_extract(body_json, '$.savedPath') IS NOT NULL ORDER BY ts ASC` ).all(selfId, peerId) as MessageRow[] } export function listAllPending(selfId: string): MessageRow[] { return db.prepare( `SELECT * FROM messages WHERE status = 'pending' AND from_id = ? ORDER BY ts ASC` ).all(selfId) as MessageRow[] } export function updateMessageStatus(messageId: string, status: string) { db.prepare(`UPDATE messages SET status = ? WHERE message_id = ?`).run(status, messageId) } // 上传完成后回写 savedPath / hash 到已存在的消息体 (进度条场景: 先显示气泡, 后台上传完再填充) export function updateMessageBody(messageId: string, patch: Record) { const row = getMessage(messageId) if (!row) return const body = JSON.parse(row.body_json) const next = { ...body, ...patch } db.prepare(`UPDATE messages SET body_json = ? WHERE message_id = ?`).run(JSON.stringify(next), messageId) } export function deleteMessage(messageId: string) { db.prepare(`DELETE FROM messages WHERE message_id = ?`).run(messageId) } // 未读 export function incrUnread(deviceId: string) { db.prepare(` INSERT INTO unread (device_id, count) VALUES (?, 1) ON CONFLICT(device_id) DO UPDATE SET count = count + 1 `).run(deviceId) } export function clearUnread(deviceId: string) { db.prepare(`DELETE FROM unread WHERE device_id = ?`).run(deviceId) } export function getUnread(): Record { const rows = db.prepare(`SELECT device_id, count FROM unread`).all() as { device_id: string; count: number }[] const out: Record = {} for (const r of rows) out[r.device_id] = r.count return out } export function totalUnread(): number { const r = db.prepare(`SELECT COALESCE(SUM(count), 0) AS s FROM unread`).get() as { s: number } return r.s } // 审计日志 (远程操作: terminal / forward / serial) export interface AuditRow { id: number ts: number action: string source_device: string | null target_device: string | null session_id: string | null payload_json: string | null result: string bytes_in: number bytes_out: number } const insertAuditStmt = db.prepare(` INSERT INTO audit_log (ts, action, source_device, target_device, session_id, payload_json, result, bytes_in, bytes_out) VALUES (@ts, @action, @source_device, @target_device, @session_id, @payload_json, @result, @bytes_in, @bytes_out) `) export function recordAudit(args: { action: string source?: string | null target?: string | null sessionId?: string | null payload?: any result: 'ok' | 'denied' | 'error' | 'closed' bytesIn?: number bytesOut?: number }): void { insertAuditStmt.run({ ts: Date.now(), action: args.action, source_device: args.source ?? null, target_device: args.target ?? null, session_id: args.sessionId ?? null, payload_json: args.payload !== undefined ? JSON.stringify(args.payload) : null, result: args.result, bytes_in: args.bytesIn ?? 0, bytes_out: args.bytesOut ?? 0, }) } export function listAudit(opts: { limit?: number; since?: number; sessionId?: string } = {}): AuditRow[] { const limit = Math.min(Math.max(opts.limit ?? 200, 1), 2000) if (opts.sessionId) { return db.prepare(`SELECT * FROM audit_log WHERE session_id = ? ORDER BY ts DESC LIMIT ?`).all(opts.sessionId, limit) as AuditRow[] } if (opts.since) { return db.prepare(`SELECT * FROM audit_log WHERE ts >= ? ORDER BY ts DESC LIMIT ?`).all(opts.since, limit) as AuditRow[] } return db.prepare(`SELECT * FROM audit_log ORDER BY ts DESC LIMIT ?`).all(limit) as AuditRow[] } export function pruneAudit(olderThanMs: number): number { const r = db.prepare(`DELETE FROM audit_log WHERE ts < ?`).run(Date.now() - olderThanMs) return r.changes } // 远程操作白名单 (按 deviceId -> { terminal, forward, serial }) // 写入 settings.json (electron-store) 而不是 DB, 便于跨设备查看与回滚 export interface RemoteAllowRow { device_id: string terminal: number // 0/1 forward: number serial: number }