feat: add message management and delivery states

This commit is contained in:
2026-07-16 17:23:24 +08:00
parent 07c7260a3c
commit 565d044f18
17 changed files with 881 additions and 97 deletions
+158 -37
View File
@@ -9,8 +9,8 @@ import { getSettings, updateSettings, ensureDownloadDir } from './settings'
import {
db, upsertDevice, listDevices, listMessages, insertMessage, getMessage,
updateMessageStatus, updateMessageBody, deleteMessage, conversationKey, incrUnread, clearUnread,
getUnread, totalUnread, listPendingMessages, listAllPending,
setDeviceIgnored, listIgnoredDevices, deleteConversation
getUnread, totalUnread, listPendingMessages, listAllPending, listSentFileUpdates,
setDeviceIgnored, listIgnoredDevices, getDevice, deleteConversation
} from './db'
import { Discovery, listNetInterfaces, type NetInterface } from './discovery'
import { ChatServer } from './chat-server'
@@ -19,7 +19,7 @@ import { FileServer, uploadFile, uploadBuffer } from './file-server'
import { broadcastToRenderer, focusMainWindow } from './window'
import { rebuildMenu } from './tray'
import { notify, setBadgeCount } from './notify'
import type { DeviceInfo, MessageEnvelope, MessageBody, ImageMessage, FileMessage, TextMessage, SystemMessage } from './protocol'
import type { DeviceInfo, MessageEnvelope, MessageBody, ImageMessage, FileMessage, TextMessage, SystemMessage, ReplyRef } from './protocol'
interface NetCtx {
self: DeviceInfo
@@ -32,7 +32,8 @@ interface NetCtx {
let ctx: NetCtx | null = null
// 接收端: 收到的 WS metadata 帧但还没拿到文件本体, 启动 N 秒期待 timer
// 任一进度事件或 received 完成事件取消; 到了 -> mark failed 并删消息, 避免气泡永远 "receiving"
// 任一进度事件或 received 完成事件取消; 到了 -> 只通知 sender, 不自动删消息
// 气泡留在 receiving 状态, 让用户右键手动删
const PENDING_FILE_TIMEOUT_MS = 60_000
const pendingMetaTimers = new Map<string, { timer: NodeJS.Timeout; messageId: string; fromDeviceId: string }>()
@@ -40,10 +41,8 @@ function armPendingMetaTimer(fileId: string, messageId: string, fromDeviceId: st
disarmPendingMetaTimer(fileId)
const timer = setTimeout(() => {
pendingMetaTimers.delete(fileId)
console.warn(`[ipc] file ${fileId} timeout: metadata arrived but no upload in ${PENDING_FILE_TIMEOUT_MS / 1000}s, marking failed`)
try { deleteMessage(messageId) } catch {}
broadcastToRenderer('message:recalled', { messageId, conversationKey: '' })
// 也告诉 sender: 这文件永远来不了 (如果是 sender 在线, 它能 mark failed)
console.warn(`[ipc] file ${fileId.slice(0, 8)} timeout: metadata arrived but no upload in ${PENDING_FILE_TIMEOUT_MS / 1000}s; bubble stays for user to delete`)
// 通知 sender 标记失败 (sender 在线时能把 status='sent' 升级到 'failed')
ctx?.chatClient.send({ type: 'fileFailed', fileId, messageId, reason: 'timeout' } as any, fromDeviceId)
}, PENDING_FILE_TIMEOUT_MS)
pendingMetaTimers.set(fileId, { timer, messageId, fromDeviceId })
@@ -61,6 +60,8 @@ export function bindNetworkContext(c: NetCtx) {
ctx = c
// 网络事件 -> 渲染进程
c.discovery.on('found', (d) => {
// 用户已删除的设备 (ignored=1) — 静默忽略, 不进 sidebar, 不重连
if (getDevice(d.deviceId)?.ignored) return
upsertDevice({
device_id: d.deviceId, name: d.name, hostname: d.hostname, platform: d.platform,
app_version: d.appVersion, last_ip: d.address, last_seen: Date.now(), ignored: 0
@@ -69,6 +70,7 @@ export function bindNetworkContext(c: NetCtx) {
c.chatClient.connectTo(d)
})
c.discovery.on('updated', (d: DeviceInfo) => {
if (getDevice(d.deviceId)?.ignored) return
upsertDevice({
device_id: d.deviceId, name: d.name, hostname: d.hostname, platform: d.platform,
app_version: d.appVersion, last_ip: d.address, last_seen: Date.now(), ignored: 0
@@ -77,6 +79,7 @@ export function bindNetworkContext(c: NetCtx) {
c.chatClient.connectTo(d)
})
c.discovery.on('lost', (id) => {
if (getDevice(id)?.ignored) return
broadcastToRenderer('device:lost', { deviceId: id })
})
@@ -172,6 +175,17 @@ export function bindNetworkContext(c: NetCtx) {
console.warn(`[chat-client] ws error to ${peer.name} (${peer.address}:${peer.chatPort}): ${err.message}`)
})
// 我们这端 outgoing WS 接通 / 断开 — 都要广播给 renderer, 让 UI 能区分
// "discovery 在线" 和 "WS 在线"; 同时 open 时主动 flush 一次待发消息 (双端 WS 都建好后必走 chat-server.on('connect') 那条路,
// 但如果对方先来认我们这边, 我们 outgoing 还没通, flush 在自己这条线路上反而能立刻推)
c.chatClient.on('open', (peer) => {
broadcastToRenderer('device:wsState', { deviceId: peer.deviceId, wsOpen: true })
flushPendingFor(peer.deviceId)
})
c.chatClient.on('close', (peer) => {
broadcastToRenderer('device:wsState', { deviceId: peer.deviceId, wsOpen: false })
})
c.chatServer.on('disconnect', (id) => {
broadcastToRenderer('device:offline', { deviceId: id })
})
@@ -189,6 +203,37 @@ export function bindNetworkContext(c: NetCtx) {
c.fileServer.on('received', (info: { fileId: string; name: string; size: number; mime: string; savedPath: string; hash: string; fromAddr: string }) => {
disarmPendingMetaTimer(info.fileId)
})
// 对端发起撤回 — 本地默默删 (含磁盘文件), 不弹框不提示
c.chatServer.on('recall', (from, messageId) => {
const row = getMessage(messageId)
if (!row) return
let savedPath: string | undefined
try { savedPath = (JSON.parse(row.body_json) as any)?.savedPath } catch {}
deleteMessage(messageId)
if (savedPath) {
fsp.unlink(savedPath).catch((e) => {
if (e?.code !== 'ENOENT') console.warn(`[recall] unlink ${savedPath} failed: ${e.message}`)
})
}
broadcastToRenderer('message:recalled', { messageId, conversationKey: row.conversation_key })
})
// 已读回执: 对端说"我看到 upTo 了", 把我方所有 from=self, to=peer, ts<=read.ts, status='delivered' 的消息升 'read'
c.chatServer.on('read', (from, upToMessageId) => {
if (!ctx) return
const row = getMessage(upToMessageId)
if (!row) return
const upToTs = row.ts
// 只升级 delivered -> read, sent 留给 ack 处理
const stmt = db.prepare(`UPDATE messages SET status = 'read' WHERE from_id = ? AND to_id = ? AND status = 'delivered' AND ts <= ?`)
stmt.run(ctx.self.deviceId, from.deviceId, upToTs)
// 把这次升级告诉 renderer (逐条 statusChanged, 复用现有渲染层逻辑)
const updated = db.prepare(`SELECT message_id FROM messages WHERE from_id = ? AND to_id = ? AND ts <= ? AND status = 'read'`).all(ctx.self.deviceId, from.deviceId, upToTs) as Array<{ message_id: string }>
for (const r of updated) {
broadcastToRenderer('message:statusChanged', { messageId: r.message_id, status: 'read' })
}
})
}
function notifyIfNeeded(from: DeviceInfo, env: MessageEnvelope) {
@@ -229,8 +274,10 @@ async function deliverAndStore(env: MessageEnvelope): Promise<{ ok: boolean; rea
// 2. 通过 ws 发送
const sent = ctx.chatClient.send({ type: 'message', payload: env, receivedAt: Date.now() }, env.toDeviceId)
if (!sent) {
// 对方不在线, 保持 pending, 等 flush
return { ok: false, reason: 'peer offline', status: 'pending' }
// WS 没开 — 消息已在 DB 里 status='pending', flushPendingFor 在对端 outgoing WS 接通时会自动重发.
// 不当作 "失败" 报给前端, 避免误导 (因为对方 UDP 可能还在, 只是 WS 抖了).
// 副作用: 渲染端的 "待对方上线" 气泡/重试按钮照常显示; 不再弹 "发送失败" toast.
return { ok: true, status: 'pending' }
}
// 3. 标记 sent (本地认为已送出; 真实送达需要 ack 协议, 见 deliverAndStoreAck)
updateMessageStatus(env.messageId, 'sent')
@@ -267,24 +314,23 @@ async function flushPendingFor(peerId: string) {
}
// 2) 上传完但 WS update 帧没送达的消息 (savedPath 已填, status 还在 'sent')
// 实际是 WS metadata 已发, 接收方在等 savedPath; 重连后重发完整 envelope 让他出 "打开/位置"
for (const row of rows) {
// 注意: rows 上面已经被 status='pending' 过滤, 这里必须重新查 status='sent' + savedPath 已填的文件
const updateRows = listSentFileUpdates(ctx.self.deviceId, peerId)
for (const row of updateRows) {
const body = JSON.parse(row.body_json)
const fileBody = body as any
if ((row.type === 'file' || row.type === 'image') && fileBody.savedPath) {
const env: MessageEnvelope = {
messageId: row.message_id,
type: row.type as any,
fromDeviceId: row.from_id,
toDeviceId: row.to_id,
ts: row.ts,
body,
}
const sentUpdate = ctx.chatClient.send({ type: 'message', payload: env, receivedAt: Date.now() }, peerId)
if (sentUpdate) {
// 触发 renderer 同步一下 body (保险 — 让对端的 uploaded bubble 也再被 forced update)
broadcastToRenderer('message:statusChanged', { messageId: row.message_id, status: 'sent' })
count++
}
const env: MessageEnvelope = {
messageId: row.message_id,
type: row.type as any,
fromDeviceId: row.from_id,
toDeviceId: row.to_id,
ts: row.ts,
body,
}
const sentUpdate = ctx.chatClient.send({ type: 'message', payload: env, receivedAt: Date.now() }, peerId)
if (sentUpdate) {
// 触发 renderer 同步一下 body (保险 — 让对端的 uploaded bubble 也再被 forced update)
broadcastToRenderer('message:statusChanged', { messageId: row.message_id, status: 'sent' })
count++
}
}
if (count > 0) console.log(`[ipc] flushed ${count} pending/update messages to ${peerId.slice(0, 8)}`)
@@ -326,6 +372,56 @@ export function registerIpc() {
return true
})
// 删除设备: 标记 ignored=1, 停掉后台重连, 从 sidebar 消失
// UDP 后续再发现也不会重新弹 (discovery 上游已过滤 ignored)
ipcMain.handle('device:delete', (_e, deviceId: string) => {
if (!ctx) return { ok: false, reason: 'no ctx' }
if (deviceId === ctx.self.deviceId) return { ok: false, reason: '不能删除本机' }
setDeviceIgnored(deviceId, true)
ctx.chatClient.remove(deviceId) // 停掉 backoff 重连
ctx.discovery.forget(deviceId) // 从 in-memory peers Map 移除
broadcastToRenderer('device:lost', { deviceId }) // 通知 renderer 把它从 device store 摘掉
// 清掉对应未读
clearUnread(deviceId)
return { ok: true }
})
// 删除单条消息 (本地) — 不广播, 不撤回对端
// opts.deleteFile: 同时删掉磁盘上的文件 (按 msg.body.savedPath)
// 返回 { ok, deletedFile? } 让 UI 在弹框里展示真实路径
ipcMain.handle('message:delete', async (_e, messageId: string, opts?: { deleteFile?: boolean }) => {
const row = getMessage(messageId)
if (!row) return { ok: false, reason: '消息不存在' }
let deletedFile: string | undefined
if (opts?.deleteFile) {
const body = (() => { try { return JSON.parse(row.body_json) } catch { return null } })()
const filePath: string | undefined = body?.savedPath
if (filePath) {
try {
await fsp.unlink(filePath)
deletedFile = filePath
} catch (e: any) {
// 文件不在磁盘上 (用户手动删了/换盘) 不视为失败 — 把消息也删了
if (e?.code !== 'ENOENT') return { ok: false, reason: `删除文件失败: ${e?.message || e}` }
}
}
}
deleteMessage(messageId)
broadcastToRenderer('message:recalled', { messageId, conversationKey: row.conversation_key })
return { ok: true, deletedFile }
})
// 已读回执: 本机在 active 对话里看到了新消息, 告诉对端把 upTo (含) 之前的消息标 'read'
ipcMain.handle('message:markRead', (_e, args: { peerId: string; upToMessageId: string }) => {
if (!ctx) return { ok: false, reason: 'no ctx' }
if (!args?.peerId || !args?.upToMessageId) return { ok: false, reason: '参数缺失' }
const sent = ctx.chatClient.send(
{ type: 'read', upTo: args.upToMessageId, ts: Date.now() },
args.peerId
)
return { ok: sent, reason: sent ? undefined : '对方离线' }
})
// ---- 消息 ----
ipcMain.handle('message:list', (_e, peerId: string) => {
if (!ctx) return []
@@ -333,7 +429,7 @@ export function registerIpc() {
return rows.map(rowToEnvelope)
})
ipcMain.handle('message:sendText', async (_e, args: { toDeviceId: string; content: string }) => {
ipcMain.handle('message:sendText', async (_e, args: { toDeviceId: string; content: string; replyTo?: ReplyRef }) => {
if (!ctx) return { ok: false, reason: 'no ctx' }
const env: MessageEnvelope = {
messageId: randomUUID(),
@@ -345,6 +441,7 @@ export function registerIpc() {
type: 'text',
messageId: '', ts: 0, fromDeviceId: ctx.self.deviceId, toDeviceId: args.toDeviceId,
content: args.content,
...(args.replyTo ? { replyTo: args.replyTo } : {}),
} as TextMessage,
}
;(env.body as TextMessage).messageId = env.messageId
@@ -365,6 +462,7 @@ export function registerIpc() {
mime?: string
asImage?: boolean
onProgress?: boolean
replyTo?: ReplyRef
}) => {
if (!ctx) return { ok: false, reason: 'no ctx' }
const peer = ctx.discovery.getPeers().find(p => p.deviceId === args.toDeviceId)
@@ -387,6 +485,7 @@ export function registerIpc() {
savedPath: null as string | null,
hash: null as string | null,
localPath: args.localPath, // retry 时复用
...(args.replyTo ? { replyTo: args.replyTo } : {}),
}
const env: MessageEnvelope = {
messageId,
@@ -453,8 +552,12 @@ export function registerIpc() {
}
broadcastToRenderer('message:local', { ...filledEnv, status: 'sent' })
// 再发一次 WS, 这次带 savedPath (对端气泡更新, 出现"打开")
if (ctx!.chatClient.isOpen(args.toDeviceId)) {
ctx!.chatClient.send({ type: 'message', payload: filledEnv, receivedAt: Date.now() }, args.toDeviceId)
// 此刻 WS 不开 -> DB 里 status='sent'+savedPath, 留给 flushPendingFor 在对端重连后补发
const sentUpdate = ctx!.chatClient.isOpen(args.toDeviceId)
? ctx!.chatClient.send({ type: 'message', payload: filledEnv, receivedAt: Date.now() }, args.toDeviceId)
: false
if (!sentUpdate) {
console.warn(`[sendFile] ws closed, update frame will retry on reconnect (mid=${messageId.slice(0, 8)})`)
}
// 不用再 broadcast statusChanged: 上一帧的 status='sent' 已经是终态之一, 等 ack 升级 delivered
void fileId // suppress unused warning
@@ -473,6 +576,7 @@ export function registerIpc() {
dataBase64: string
name: string
mime: string
replyTo?: ReplyRef
}) => {
if (!ctx) return { ok: false, reason: 'no ctx' }
const peer = ctx.discovery.getPeers().find(p => p.deviceId === args.toDeviceId)
@@ -490,6 +594,7 @@ export function registerIpc() {
fileId, name: args.name, size: buf.length, mime: args.mime,
thumbDataUrl: buf.length < 200_000 ? `data:${args.mime};base64,${args.dataBase64}` : undefined,
dataBase64: args.dataBase64,
...(args.replyTo ? { replyTo: args.replyTo } : {}),
}
const env: MessageEnvelope = {
messageId, type: 'image',
@@ -540,8 +645,12 @@ export function registerIpc() {
body: { ...baseBody, savedPath: uploadResult.savedPath as any, hash: uploadResult.hash as any },
}
broadcastToRenderer('message:local', { ...filledEnv, status: 'sent' })
if (ctx!.chatClient.isOpen(args.toDeviceId)) {
ctx!.chatClient.send({ type: 'message', payload: filledEnv, receivedAt: Date.now() }, args.toDeviceId)
// 此刻 WS 不开 -> DB 里 status='sent'+savedPath, 留给 flushPendingFor 在对端重连后补发
const sentUpdate = ctx!.chatClient.isOpen(args.toDeviceId)
? ctx!.chatClient.send({ type: 'message', payload: filledEnv, receivedAt: Date.now() }, args.toDeviceId)
: false
if (!sentUpdate) {
console.warn(`[sendBuffer] ws closed, update frame will retry on reconnect (mid=${messageId.slice(0, 8)})`)
}
void fileId
})().catch((e: any) => {
@@ -556,11 +665,17 @@ export function registerIpc() {
ipcMain.handle('message:recall', async (_e, messageId: string) => {
if (!ctx) return { ok: false }
const row = getMessage(messageId)
if (!row) return { ok: false, reason: 'no message' }
if (row.from_id !== ctx.self.deviceId) return { ok: false, reason: 'not your message' }
if (!row) return { ok: false, reason: '消息不存在' }
if (row.from_id !== ctx.self.deviceId) return { ok: false, reason: '只能撤回自己发的消息' }
// 微信式 2 分钟时限
const RECALL_LIMIT_MS = 2 * 60_000
if (Date.now() - row.ts > RECALL_LIMIT_MS) {
const ageMin = Math.round((Date.now() - row.ts) / 60_000)
return { ok: false, reason: `超过 2 分钟无法撤回 (已 ${ageMin} 分钟)` }
}
// 删除本地
deleteMessage(messageId)
// 通知对端
// 通知对端 (不传 savedPath 让对端从 DB 自行取 — 这里的 row 还在; 但保险起见让对端查自己的 DB)
ctx.chatClient.send({ type: 'recall', messageId, ts: Date.now() }, row.to_id)
broadcastToRenderer('message:recalled', { messageId, conversationKey: row.conversation_key })
return { ok: true }
@@ -599,7 +714,7 @@ export function registerIpc() {
const peer = ctx.discovery.getPeers().find(p => p.deviceId === row.to_id)
if (!peer) {
setStatus('pending')
return { ok: false, reason: 'peer offline', status: 'pending' }
return { ok: false, reason: '对方设备不在线', status: 'pending', wsDown: true }
}
if (!ctx.chatClient.isOpen(row.to_id)) {
setStatus('pending')
@@ -701,7 +816,13 @@ export function registerIpc() {
ensureDownloadDir()
if (patch.deviceName) {
ctx!.self.name = patch.deviceName
// 通知前端并重启 broadcast 频次
}
if ('autoStart' in patch && app.isPackaged) {
try {
app.setLoginItemSettings({ openAtLogin: !!patch.autoStart, path: process.execPath })
} catch (e: any) {
console.warn('[settings:set] setLoginItemSettings failed:', e?.message)
}
}
rebuildMenu()
broadcastToRenderer('settings:changed', ns)