| … | … | @@ -120,6 +120,9 @@ export class HttpTransport { |
| 120 | 120 | Connection: 'keep-alive' |
| 121 | 121 | }) |
| 122 | 122 | |
| 123 | // Resolve member names for display |
| 124 | const nameMap = await this.router._getMemberNames(roomId) |
| 125 | |
| 123 | 126 | // Send recent messages first |
| 124 | 127 | const recent = await this.router.handle('read', { roomId, limit: 10 }) |
| 125 | 128 | if (recent.ok && recent.data) { |
| … | … | @@ -130,26 +133,68 @@ export class HttpTransport { |
| 130 | 133 | res.write('event: ready\ndata: {}\n\n') |
| 131 | 134 | |
| 132 | 135 | // Subscribe to live messages |
| 136 | // subscribeChatMessages yields arrays (full snapshot) on each change |
| 133 | 137 | const stream = this.router.subscribe(roomId) |
| 134 | 138 | |
| 139 | // Track highest seq — skip first snapshot (it's the current state), |
| 140 | // only emit messages from subsequent snapshots that are newer |
| 141 | let highSeq = -1 |
| 142 | let initialized = false |
| 143 | |
| 135 | 144 | if (stream && typeof stream[Symbol.asyncIterator] === 'function') { |
| 136 | 145 | ;(async () => { |
| 137 | 146 | try { |
| 138 | | for await (const event of stream) { |
| 139 | | if (!event) continue |
| 140 | | const msg = normalizeEvent(event) |
| 141 | | if (msg) res.write(`event: message\ndata: ${JSON.stringify(msg)}\n\n`) |
| 147 | for await (const snapshot of stream) { |
| 148 | if (!snapshot) continue |
| 149 | |
| 150 | const messages = Array.isArray(snapshot) ? snapshot : [snapshot] |
| 151 | |
| 152 | if (!initialized) { |
| 153 | // First snapshot = current state — just record the high water mark |
| 154 | for (const event of messages) { |
| 155 | const seq = event.seq ?? event.clock ?? -1 |
| 156 | if (seq > highSeq) highSeq = seq |
| 157 | } |
| 158 | initialized = true |
| 159 | continue |
| 160 | } |
| 161 | |
| 162 | // Subsequent snapshots — emit only messages newer than highSeq |
| 163 | for (const event of messages) { |
| 164 | const seq = event.seq ?? event.clock ?? -1 |
| 165 | if (seq <= highSeq) continue |
| 166 | |
| 167 | highSeq = seq |
| 168 | const msg = normalizeEvent(event, nameMap) |
| 169 | if (msg) res.write(`event: message\ndata: ${JSON.stringify(msg)}\n\n`) |
| 170 | } |
| 142 | 171 | } |
| 143 | 172 | } catch (err) { |
| 144 | | res.write(`event: error\ndata: ${JSON.stringify({ error: err.message })}\n\n`) |
| 173 | if (err.message !== 'Stream was destroyed') { |
| 174 | res.write(`event: error\ndata: ${JSON.stringify({ error: err.message })}\n\n`) |
| 175 | } |
| 145 | 176 | } |
| 146 | 177 | res.end() |
| 147 | 178 | })() |
| 148 | 179 | } else if (stream && typeof stream.on === 'function') { |
| 149 | | stream.on('data', event => { |
| 150 | | if (!event) return |
| 151 | | const msg = normalizeEvent(event) |
| 152 | | if (msg) res.write(`event: message\ndata: ${JSON.stringify(msg)}\n\n`) |
| 180 | stream.on('data', snapshot => { |
| 181 | if (!snapshot) return |
| 182 | const messages = Array.isArray(snapshot) ? snapshot : [snapshot] |
| 183 | if (!initialized) { |
| 184 | for (const event of messages) { |
| 185 | const seq = event.seq ?? event.clock ?? -1 |
| 186 | if (seq > highSeq) highSeq = seq |
| 187 | } |
| 188 | initialized = true |
| 189 | return |
| 190 | } |
| 191 | for (const event of messages) { |
| 192 | const seq = event.seq ?? event.clock ?? -1 |
| 193 | if (seq <= highSeq) continue |
| 194 | highSeq = seq |
| 195 | const msg = normalizeEvent(event, nameMap) |
| 196 | if (msg) res.write(`event: message\ndata: ${JSON.stringify(msg)}\n\n`) |
| 197 | } |
| 153 | 198 | }) |
| 154 | 199 | stream.on('error', err => { |
| 155 | 200 | res.write(`event: error\ndata: ${JSON.stringify({ error: err.message })}\n\n`) |
| … | … | @@ -173,16 +218,35 @@ export class HttpTransport { |
| 173 | 218 | } |
| 174 | 219 | } |
| 175 | 220 | |
| 176 | | function normalizeEvent (event) { |
| 221 | function normalizeEvent (event, nameMap) { |
| 222 | if (!event || typeof event !== 'object') return null |
| 223 | |
| 177 | 224 | const text = event.chat?.text || event.text || '' |
| 178 | | const system = event.message?.generic || null |
| 225 | const generic = event.message?.generic || null |
| 226 | |
| 227 | // Normalize system events |
| 228 | let system = null |
| 229 | if (generic && typeof generic === 'object') { |
| 230 | for (const v of Object.values(generic)) { |
| 231 | if (v !== null && v !== false) { system = generic; break } |
| 232 | } |
| 233 | } else if (generic) { |
| 234 | system = generic |
| 235 | } |
| 236 | |
| 179 | 237 | if (!text && !system) return null |
| 238 | |
| 239 | const memberId = event.memberId || 'unknown' |
| 240 | |
| 180 | 241 | return { |
| 181 | | memberId: event.memberId || 'unknown', |
| 242 | memberId, |
| 243 | name: (nameMap && nameMap[memberId]) || memberId.slice(0, 8), |
| 182 | 244 | text, |
| 183 | 245 | timestamp: event.timestamp || Date.now(), |
| 184 | 246 | deleted: event.deleted || false, |
| 185 | | system |
| 247 | system, |
| 248 | seq: event.seq ?? event.clock ?? null, |
| 249 | id: event.id || null |
| 186 | 250 | } |
| 187 | 251 | } |
| 188 | 252 | |