Merge pull request #834 from xunchahaha/dev

Dev

electron/services/httpService.ts

@@ -76,6 +76,12 @@ interface ApiExportedMedia {
   relativePath: string
 }
 
+interface MessagePushReplayEvent {
+  id: number
+  body: string
+  createdAt: number
+}
+
 // ChatLab 消息类型映射
 const ChatLabType = {
   TEXT: 0,
@@ -107,8 +113,12 @@ class HttpService {
   private running: boolean = false
   private connections: Set<import('net').Socket> = new Set()
   private messagePushClients: Set<http.ServerResponse> = new Set()
+  private messagePushReplayBuffer: MessagePushReplayEvent[] = []
   private messagePushHeartbeatTimer: ReturnType<typeof setInterval> | null = null
   private connectionMutex: boolean = false
+  private messagePushEventId = 0
+  private readonly messagePushReplayLimit = 1000
+  private readonly messagePushReplayTtlMs = 10 * 60 * 1000
 
   constructor() {
     this.configService = ConfigService.getInstance()
@@ -178,6 +188,7 @@ class HttpService {
           } catch {}
         }
         this.messagePushClients.clear()
+        this.messagePushReplayBuffer = []
         if (this.messagePushHeartbeatTimer) {
           clearInterval(this.messagePushHeartbeatTimer)
           this.messagePushHeartbeatTimer = null
@@ -232,9 +243,56 @@ class HttpService {
     return `http://${this.host}:${this.port}/api/v1/push/messages`
   }
 
+  private nextMessagePushEventId(): number {
+    this.messagePushEventId += 1
+    if (!Number.isSafeInteger(this.messagePushEventId) || this.messagePushEventId <= 0) {
+      this.messagePushEventId = 1
+    }
+    return this.messagePushEventId
+  }
+
+  private rememberMessagePushEvent(id: number, body: string): void {
+    this.pruneMessagePushReplayBuffer()
+    this.messagePushReplayBuffer.push({ id, body, createdAt: Date.now() })
+    if (this.messagePushReplayBuffer.length > this.messagePushReplayLimit) {
+      this.messagePushReplayBuffer.splice(0, this.messagePushReplayBuffer.length - this.messagePushReplayLimit)
+    }
+  }
+
+  private pruneMessagePushReplayBuffer(): void {
+    const cutoff = Date.now() - this.messagePushReplayTtlMs
+    while (this.messagePushReplayBuffer.length > 0 && this.messagePushReplayBuffer[0].createdAt < cutoff) {
+      this.messagePushReplayBuffer.shift()
+    }
+  }
+
+  private parseMessagePushLastEventId(req: http.IncomingMessage, url?: URL): number {
+    const queryValue = url?.searchParams.get('lastEventId') || url?.searchParams.get('last_event_id') || ''
+    const headerValue = Array.isArray(req.headers['last-event-id'])
+      ? req.headers['last-event-id'][0]
+      : req.headers['last-event-id']
+    const parsed = Number.parseInt(String(queryValue || headerValue || '0').trim(), 10)
+    return Number.isFinite(parsed) && parsed > 0 ? parsed : 0
+  }
+
+  private replayMessagePushEvents(res: http.ServerResponse, lastEventId: number): void {
+    this.pruneMessagePushReplayBuffer()
+    const events = lastEventId > 0
+      ? this.messagePushReplayBuffer.filter((event) => event.id > lastEventId)
+      : this.messagePushReplayBuffer
+
+    for (const event of events) {
+      if (res.writableEnded || res.destroyed) return
+      res.write(event.body)
+    }
+  }
+
   broadcastMessagePush(payload: Record<string, unknown>): void {
-    if (!this.running || this.messagePushClients.size === 0) return
+    if (!this.running) return
-    const eventBody = `event: message.new\ndata: ${JSON.stringify(payload)}\n\n`
+    const eventId = this.nextMessagePushEventId()
+    const eventBody = `id: ${eventId}\nevent: message.new\ndata: ${JSON.stringify(payload)}\n\n`
+    this.rememberMessagePushEvent(eventId, eventBody)
+    if (this.messagePushClients.size === 0) return
 
     for (const client of Array.from(this.messagePushClients)) {
       try {
@@ -365,7 +423,7 @@ class HttpService {
             if (pathname === '/health' || pathname === '/api/v1/health') {
                 this.sendJson(res, { status: 'ok' })
             } else if (pathname === '/api/v1/push/messages') {
-                this.handleMessagePushStream(req, res)
+                this.handleMessagePushStream(req, res, url)
             } else if (pathname === '/api/v1/messages') {
                 await this.handleMessages(url, res)
             } else if (pathname === '/api/v1/sessions') {
@@ -440,7 +498,7 @@ class HttpService {
     }, 25000)
   }
 
-  private handleMessagePushStream(req: http.IncomingMessage, res: http.ServerResponse): void {
+  private handleMessagePushStream(req: http.IncomingMessage, res: http.ServerResponse, url: URL): void {
     if (this.configService.get('messagePushEnabled') !== true) {
       this.sendError(res, 403, 'Message push is disabled')
       return
@@ -453,9 +511,10 @@ class HttpService {
       'X-Accel-Buffering': 'no'
     })
     res.flushHeaders?.()
-    res.write(`event: ready\ndata: ${JSON.stringify({ success: true, stream: this.getMessagePushStreamUrl() })}\n\n`)
 
     this.messagePushClients.add(res)
+    res.write(`event: ready\ndata: ${JSON.stringify({ success: true, stream: this.getMessagePushStreamUrl() })}\n\n`)
+    this.replayMessagePushEvents(res, this.parseMessagePushLastEventId(req, url))
 
     const cleanup = () => {
       this.messagePushClients.delete(res)

electron/services/messagePushService.ts

@@ -12,6 +12,15 @@ interface SessionBaseline {
   unreadCount: number
 }
 
+interface PushSessionResult {
+  fetched: boolean
+  maxFetchedTimestamp: number
+  incomingCandidateCount: number
+  observedIncomingCount: number
+  expectedIncomingCount: number
+  retry: boolean
+}
+
 interface MessagePushPayload {
   event: 'message.new'
   sessionId: string
@@ -37,10 +46,13 @@ class MessagePushService {
   private readonly configService: ConfigService
   private readonly sessionBaseline = new Map<string, SessionBaseline>()
   private readonly recentMessageKeys = new Map<string, number>()
+  private readonly seenMessageKeys = new Map<string, number>()
+  private readonly seenPrimedSessions = new Set<string>()
   private readonly groupNicknameCache = new Map<string, { nicknames: Record<string, string>; updatedAt: number }>()
   private readonly pushAvatarCacheDir: string
   private readonly pushAvatarDataCache = new Map<string, string>()
   private readonly debounceMs = 350
+  private readonly lookbackSeconds = 2
   private readonly recentMessageTtlMs = 10 * 60 * 1000
   private readonly groupNicknameCacheTtlMs = 5 * 60 * 1000
   private debounceTimer: ReturnType<typeof setTimeout> | null = null
@@ -111,6 +123,8 @@ class MessagePushService {
   private resetRuntimeState(): void {
     this.sessionBaseline.clear()
     this.recentMessageKeys.clear()
+    this.seenMessageKeys.clear()
+    this.seenPrimedSessions.clear()
     this.groupNicknameCache.clear()
     this.baselineReady = false
     this.messageTableScanRequested = false
@@ -190,7 +204,6 @@ class MessagePushService {
       }
 
       const previousBaseline = new Map(this.sessionBaseline)
-      this.setBaseline(sessions)
 
       const candidates = sessions.filter((session) => {
         const previous = previousBaseline.get(session.username)
@@ -199,11 +212,24 @@ class MessagePushService {
         }
         return scanMessageBackedSessions && this.shouldScanMessageBackedSession(previous, session)
       })
+      const candidateIds = new Set<string>()
       for (const session of candidates) {
-        await this.pushSessionMessages(
+        const sessionId = String(session.username || '').trim()
+        if (sessionId) candidateIds.add(sessionId)
+        const result = await this.pushSessionMessages(
           session,
           previousBaseline.get(session.username) || this.sessionBaseline.get(session.username)
         )
+        this.updateInspectedBaseline(session, previousBaseline.get(session.username), result)
+        if (result.retry) {
+          this.rerunRequested = true
+        }
+      }
+
+      for (const session of sessions) {
+        const sessionId = String(session.username || '').trim()
+        if (!sessionId || candidateIds.has(sessionId)) continue
+        this.updateObservedBaseline(session, previousBaseline.get(sessionId))
       }
     } finally {
       this.processing = false
@@ -235,6 +261,40 @@ class MessagePushService {
     }
   }
 
+  private updateObservedBaseline(session: ChatSession, previous: SessionBaseline | undefined): void {
+    const username = String(session.username || '').trim()
+    if (!username) return
+
+    const sessionTimestamp = Number(session.lastTimestamp || 0)
+    const previousTimestamp = Number(previous?.lastTimestamp || 0)
+    this.sessionBaseline.set(username, {
+      lastTimestamp: Math.max(sessionTimestamp, previousTimestamp),
+      unreadCount: Number(session.unreadCount ?? previous?.unreadCount ?? 0)
+    })
+  }
+
+  private updateInspectedBaseline(
+    session: ChatSession,
+    previous: SessionBaseline | undefined,
+    result: PushSessionResult
+  ): void {
+    const username = String(session.username || '').trim()
+    if (!username) return
+
+    const previousTimestamp = Number(previous?.lastTimestamp || 0)
+    const current = this.sessionBaseline.get(username) || previous || { lastTimestamp: 0, unreadCount: 0 }
+    const nextTimestamp = result.retry
+      ? previousTimestamp
+      : Math.max(previousTimestamp, current.lastTimestamp, result.maxFetchedTimestamp)
+
+    this.sessionBaseline.set(username, {
+      lastTimestamp: nextTimestamp,
+      unreadCount: result.retry
+        ? Number(previous?.unreadCount || 0)
+        : Number(session.unreadCount || 0)
+    })
+  }
+
   private shouldInspectSession(previous: SessionBaseline | undefined, session: ChatSession): boolean {
     const sessionId = String(session.username || '').trim()
     if (!sessionId || sessionId.toLowerCase().includes('placeholder_foldgroup')) {
@@ -275,26 +335,84 @@ class MessagePushService {
     return Boolean(previous) || Number(session.lastTimestamp || 0) > 0
   }
 
-  private async pushSessionMessages(session: ChatSession, previous: SessionBaseline | undefined): Promise<void> {
+  private async pushSessionMessages(session: ChatSession, previous: SessionBaseline | undefined): Promise<PushSessionResult> {
-    const since = Math.max(0, Number(previous?.lastTimestamp || 0))
+    const previousTimestamp = Math.max(0, Number(previous?.lastTimestamp || 0))
+    const previousUnreadCount = Math.max(0, Number(previous?.unreadCount || 0))
+    const currentUnreadCount = Math.max(0, Number(session.unreadCount || 0))
+    const expectedIncomingCount = previous
+      ? Math.max(0, currentUnreadCount - previousUnreadCount)
+      : 0
+    const since = previous
+      ? Math.max(0, previousTimestamp - this.lookbackSeconds)
+      : 0
     const newMessagesResult = await chatService.getNewMessages(session.username, since, 1000)
     if (!newMessagesResult.success || !newMessagesResult.messages || newMessagesResult.messages.length === 0) {
-      return
+      return {
+        fetched: false,
+        maxFetchedTimestamp: previousTimestamp,
+        incomingCandidateCount: 0,
+        observedIncomingCount: 0,
+        expectedIncomingCount,
+        retry: expectedIncomingCount > 0
+      }
     }
 
+    const sessionId = String(session.username || '').trim()
+    const maxFetchedTimestamp = newMessagesResult.messages.reduce((max, message) => {
+      const createTime = Number(message.createTime || 0)
+      return Number.isFinite(createTime) && createTime > max ? createTime : max
+    }, previousTimestamp)
+    const seenPrimed = sessionId ? this.seenPrimedSessions.has(sessionId) : false
+    const sameTimestampIncoming: Message[] = []
+    const candidateMessages: Message[] = []
+    let observedIncomingCount = 0
+
     for (const message of newMessagesResult.messages) {
       const messageKey = String(message.messageKey || '').trim()
       if (!messageKey) continue
+      const createTime = Number(message.createTime || 0)
+      const seen = this.isSeenMessage(messageKey)
+      const recent = this.isRecentMessage(messageKey)
+
+      if (message.isSend !== 1) {
+        if (!previous || createTime > previousTimestamp || (seenPrimed && createTime === previousTimestamp)) {
+          observedIncomingCount += 1
+        }
+      }
+
+      if (previous && !seenPrimed && createTime < previousTimestamp) {
+        this.rememberSeenMessageKey(messageKey)
+        continue
+      }
+
+      if (seen || recent) {
+        continue
+      }
       if (message.isSend === 1) continue
-
+      if (previous && !seenPrimed && createTime === previousTimestamp) {
-      if (previous && Number(message.createTime || 0) <= Number(previous.lastTimestamp || 0)) {
+        sameTimestampIncoming.push(message)
         continue
       }
 
-      if (this.isRecentMessage(messageKey)) {
+      candidateMessages.push(message)
-        continue
+    }
-      }
 
+    const futureIncomingCount = candidateMessages.filter((message) => {
+      const createTime = Number(message.createTime || 0)
+      return !previous || createTime > previousTimestamp || seenPrimed
+    }).length
+    const sameTimestampAllowance = previous && !seenPrimed
+      ? Math.max(0, expectedIncomingCount - futureIncomingCount)
+      : 0
+    const selectedSameTimestamp = sameTimestampAllowance > 0
+      ? sameTimestampIncoming.slice(-sameTimestampAllowance)
+      : []
+    const messagesToPush = [...selectedSameTimestamp, ...candidateMessages]
+    const incomingCandidateCount = messagesToPush.length
+
+    for (const message of messagesToPush) {
+      const messageKey = String(message.messageKey || '').trim()
+      if (!messageKey) continue
       const payload = await this.buildPayload(session, message)
       if (!payload) continue
       if (!this.shouldPushPayload(payload)) continue
@@ -303,6 +421,21 @@ class MessagePushService {
       this.rememberMessageKey(messageKey)
       this.bumpSessionBaseline(session.username, message)
     }
+
+    for (const message of newMessagesResult.messages) {
+      const messageKey = String(message.messageKey || '').trim()
+      if (messageKey) this.rememberSeenMessageKey(messageKey)
+    }
+    if (sessionId) this.seenPrimedSessions.add(sessionId)
+
+    return {
+      fetched: true,
+      maxFetchedTimestamp,
+      incomingCandidateCount,
+      observedIncomingCount,
+      expectedIncomingCount,
+      retry: expectedIncomingCount > 0 && observedIncomingCount < expectedIncomingCount
+    }
   }
 
   private async buildPayload(session: ChatSession, message: Message): Promise<MessagePushPayload | null> {
@@ -558,6 +691,17 @@ class MessagePushService {
     this.pruneRecentMessageKeys()
   }
 
+  private isSeenMessage(messageKey: string): boolean {
+    this.pruneSeenMessageKeys()
+    const timestamp = this.seenMessageKeys.get(messageKey)
+    return typeof timestamp === 'number' && Date.now() - timestamp < this.recentMessageTtlMs
+  }
+
+  private rememberSeenMessageKey(messageKey: string): void {
+    this.seenMessageKeys.set(messageKey, Date.now())
+    this.pruneSeenMessageKeys()
+  }
+
   private pruneRecentMessageKeys(): void {
     const now = Date.now()
     for (const [key, timestamp] of this.recentMessageKeys.entries()) {
@@ -567,6 +711,15 @@ class MessagePushService {
     }
   }
 
+  private pruneSeenMessageKeys(): void {
+    const now = Date.now()
+    for (const [key, timestamp] of this.seenMessageKeys.entries()) {
+      if (now - timestamp > this.recentMessageTtlMs) {
+        this.seenMessageKeys.delete(key)
+      }
+    }
+  }
+
 }
 
 export const messagePushService = new MessagePushService()

src/pages/ExportPage.scss

@@ -1915,14 +1915,14 @@
   --contacts-select-col-width: 34px;
   --contacts-avatar-col-width: 44px;
   --contacts-inline-padding: 12px;
-  --contacts-column-gap: 12px;
+  --contacts-column-gap: 10px;
-  --contacts-name-text-width: 10em;
+  --contacts-name-text-width: 9.5em;
   --contacts-main-col-width: calc(var(--contacts-avatar-col-width) + var(--contacts-column-gap) + var(--contacts-name-text-width));
   --contacts-left-sticky-width: calc(var(--contacts-select-col-width) + var(--contacts-main-col-width) + var(--contacts-column-gap));
-  --contacts-message-col-width: 120px;
+  --contacts-message-col-width: 104px;
-  --contacts-media-col-width: 72px;
+  --contacts-media-col-width: 66px;
   --contacts-action-col-width: 140px;
-  --contacts-actions-sticky-width: 240px;
+  --contacts-actions-sticky-width: 180px;
   --contacts-table-min-width: 1240px;
   overflow: hidden;
   border: none;
@@ -2197,22 +2197,8 @@
     gap: 10px;
     flex-wrap: nowrap;
     flex-shrink: 0;
-    position: sticky;
-    right: 0;
-    z-index: 13;
     background: var(--contacts-header-bg);
     white-space: nowrap;
-
-    &::before {
-      content: '';
-      position: absolute;
-      top: 0;
-      bottom: 0;
-      left: -8px;
-      width: 8px;
-      pointer-events: none;
-      background: linear-gradient(to right, transparent, var(--contacts-header-bg));
-    }
   }
 
   .contacts-list {
@@ -2396,7 +2382,7 @@
     display: flex;
     align-items: center;
     gap: var(--contacts-column-gap);
-    padding: 12px 6px 12px var(--contacts-inline-padding);
+    padding: 12px var(--contacts-inline-padding);
     min-width: max(100%, var(--contacts-table-min-width));
     height: 72px;
     box-sizing: border-box;
@@ -2801,22 +2787,8 @@
   width: var(--contacts-actions-sticky-width);
   min-width: var(--contacts-actions-sticky-width);
   flex-shrink: 0;
-  position: sticky;
-  right: 0;
-  z-index: 10;
   background: var(--contacts-row-bg);
 
-  &::before {
-    content: '';
-    position: absolute;
-    top: 0;
-    bottom: 0;
-    left: -9px;
-    width: 9px;
-    pointer-events: none;
-    background: linear-gradient(to right, transparent, var(--contacts-row-bg));
-  }
-
   .row-action-main {
     display: inline-flex;
     align-items: flex-start;