| … | … | @@ -0,0 +1,267 @@ |
| 1 | import Pipe from 'bare-pipe' |
| 2 | import FramedStream from 'framed-stream' |
| 3 | import b4a from 'b4a' |
| 4 | import fs from 'fs' |
| 5 | |
| 6 | export class UdsTransport { |
| 7 | constructor (router, opts = {}) { |
| 8 | this.router = router |
| 9 | this.socketPath = opts.socketPath || '/tmp/keet-daemon.sock' |
| 10 | this.server = null |
| 11 | this._clients = new Set() |
| 12 | } |
| 13 | |
| 14 | start () { |
| 15 | return new Promise((resolve, reject) => { |
| 16 | // Remove stale socket file |
| 17 | try { fs.unlinkSync(this.socketPath) } catch {} |
| 18 | |
| 19 | this.server = Pipe.createServer((rawPipe) => { |
| 20 | this._onConnection(rawPipe) |
| 21 | }) |
| 22 | |
| 23 | this.server.on('error', (err) => { |
| 24 | console.error('[keet-daemon] UDS server error:', err.message) |
| 25 | reject(err) |
| 26 | }) |
| 27 | |
| 28 | this.server.listen(this.socketPath, () => { |
| 29 | console.log(`[keet-daemon] UDS listening on ${this.socketPath}`) |
| 30 | resolve() |
| 31 | }) |
| 32 | }) |
| 33 | } |
| 34 | |
| 35 | stop () { |
| 36 | for (const client of this._clients) { |
| 37 | client.destroy() |
| 38 | } |
| 39 | this._clients.clear() |
| 40 | if (this.server) this.server.close() |
| 41 | try { fs.unlinkSync(this.socketPath) } catch {} |
| 42 | } |
| 43 | |
| 44 | _onConnection (rawPipe) { |
| 45 | const framed = new FramedStream(rawPipe) |
| 46 | const client = { |
| 47 | framed, |
| 48 | rawPipe, |
| 49 | subscriptions: new Map(), // roomId -> { stream, highSeq, initialized } |
| 50 | destroy () { |
| 51 | // Clean up all subscriptions |
| 52 | for (const [, sub] of this.subscriptions) { |
| 53 | if (sub.stream && typeof sub.stream.destroy === 'function') { |
| 54 | sub.stream.destroy() |
| 55 | } |
| 56 | } |
| 57 | this.subscriptions.clear() |
| 58 | framed.destroy() |
| 59 | } |
| 60 | } |
| 61 | |
| 62 | this._clients.add(client) |
| 63 | |
| 64 | framed.on('data', (buf) => { |
| 65 | this._onFrame(client, buf) |
| 66 | }) |
| 67 | |
| 68 | framed.on('error', (err) => { |
| 69 | if (err.message !== 'Stream destroyed') { |
| 70 | console.error('[keet-daemon] UDS client error:', err.message) |
| 71 | } |
| 72 | }) |
| 73 | |
| 74 | framed.on('close', () => { |
| 75 | client.destroy() |
| 76 | this._clients.delete(client) |
| 77 | }) |
| 78 | } |
| 79 | |
| 80 | async _onFrame (client, buf) { |
| 81 | let msg |
| 82 | try { |
| 83 | msg = JSON.parse(b4a.toString(buf, 'utf8')) |
| 84 | } catch { |
| 85 | this._sendFrame(client, { id: null, ok: false, error: 'Invalid JSON' }) |
| 86 | return |
| 87 | } |
| 88 | |
| 89 | const { id, method, params } = msg |
| 90 | |
| 91 | if (!method) { |
| 92 | this._sendFrame(client, { id, ok: false, error: 'Missing "method"' }) |
| 93 | return |
| 94 | } |
| 95 | |
| 96 | // Handle subscribe/unsubscribe at transport level |
| 97 | if (method === 'subscribe') { |
| 98 | await this._handleSubscribe(client, id, params || {}) |
| 99 | return |
| 100 | } |
| 101 | |
| 102 | if (method === 'unsubscribe') { |
| 103 | this._handleUnsubscribe(client, id, params || {}) |
| 104 | return |
| 105 | } |
| 106 | |
| 107 | try { |
| 108 | const result = await this.router.handle(method, params || {}) |
| 109 | this._sendFrame(client, { id, ...result }) |
| 110 | } catch (err) { |
| 111 | this._sendFrame(client, { id, ok: false, error: err.message || String(err) }) |
| 112 | } |
| 113 | } |
| 114 | |
| 115 | async _handleSubscribe (client, id, params) { |
| 116 | const { roomId } = params |
| 117 | |
| 118 | if (!roomId) { |
| 119 | this._sendFrame(client, { id, ok: false, error: 'Missing roomId' }) |
| 120 | return |
| 121 | } |
| 122 | |
| 123 | // If already subscribed, unsubscribe first |
| 124 | if (client.subscriptions.has(roomId)) { |
| 125 | const old = client.subscriptions.get(roomId) |
| 126 | if (old.stream && typeof old.stream.destroy === 'function') { |
| 127 | old.stream.destroy() |
| 128 | } |
| 129 | client.subscriptions.delete(roomId) |
| 130 | } |
| 131 | |
| 132 | // Resolve member names |
| 133 | const nameMap = await this.router._getMemberNames(roomId) |
| 134 | |
| 135 | // Get subscription stream |
| 136 | const stream = this.router.subscribe(roomId) |
| 137 | |
| 138 | const sub = { stream, highSeq: -1, initialized: false } |
| 139 | client.subscriptions.set(roomId, sub) |
| 140 | |
| 141 | // Confirm subscription |
| 142 | this._sendFrame(client, { id, ok: true, data: { subscribed: roomId } }) |
| 143 | |
| 144 | if (!stream) return |
| 145 | |
| 146 | if (typeof stream[Symbol.asyncIterator] === 'function') { |
| 147 | ;(async () => { |
| 148 | try { |
| 149 | for await (const snapshot of stream) { |
| 150 | if (!snapshot) continue |
| 151 | const messages = Array.isArray(snapshot) ? snapshot : [snapshot] |
| 152 | |
| 153 | if (!sub.initialized) { |
| 154 | // First snapshot = current state — record high water mark |
| 155 | for (const event of messages) { |
| 156 | const seq = event.seq ?? event.clock ?? -1 |
| 157 | if (seq > sub.highSeq) sub.highSeq = seq |
| 158 | } |
| 159 | sub.initialized = true |
| 160 | continue |
| 161 | } |
| 162 | |
| 163 | // Emit only newer messages |
| 164 | for (const event of messages) { |
| 165 | const seq = event.seq ?? event.clock ?? -1 |
| 166 | if (seq <= sub.highSeq) continue |
| 167 | sub.highSeq = seq |
| 168 | const msg = normalizeEvent(event, nameMap) |
| 169 | if (msg) { |
| 170 | this._sendFrame(client, { event: 'message', roomId, data: msg }) |
| 171 | } |
| 172 | } |
| 173 | } |
| 174 | } catch (err) { |
| 175 | if (err.message !== 'Stream destroyed') { |
| 176 | this._sendFrame(client, { event: 'error', roomId, data: { error: err.message } }) |
| 177 | } |
| 178 | } |
| 179 | })() |
| 180 | } else if (stream && typeof stream.on === 'function') { |
| 181 | stream.on('data', (snapshot) => { |
| 182 | if (!snapshot) return |
| 183 | const messages = Array.isArray(snapshot) ? snapshot : [snapshot] |
| 184 | if (!sub.initialized) { |
| 185 | for (const event of messages) { |
| 186 | const seq = event.seq ?? event.clock ?? -1 |
| 187 | if (seq > sub.highSeq) sub.highSeq = seq |
| 188 | } |
| 189 | sub.initialized = true |
| 190 | return |
| 191 | } |
| 192 | for (const event of messages) { |
| 193 | const seq = event.seq ?? event.clock ?? -1 |
| 194 | if (seq <= sub.highSeq) continue |
| 195 | sub.highSeq = seq |
| 196 | const msg = normalizeEvent(event, nameMap) |
| 197 | if (msg) { |
| 198 | this._sendFrame(client, { event: 'message', roomId, data: msg }) |
| 199 | } |
| 200 | } |
| 201 | }) |
| 202 | stream.on('error', (err) => { |
| 203 | this._sendFrame(client, { event: 'error', roomId, data: { error: err.message } }) |
| 204 | }) |
| 205 | stream.on('close', () => { |
| 206 | client.subscriptions.delete(roomId) |
| 207 | }) |
| 208 | } |
| 209 | } |
| 210 | |
| 211 | _handleUnsubscribe (client, id, params) { |
| 212 | const { roomId } = params |
| 213 | |
| 214 | if (!roomId) { |
| 215 | this._sendFrame(client, { id, ok: false, error: 'Missing roomId' }) |
| 216 | return |
| 217 | } |
| 218 | |
| 219 | const sub = client.subscriptions.get(roomId) |
| 220 | if (sub) { |
| 221 | if (sub.stream && typeof sub.stream.destroy === 'function') { |
| 222 | sub.stream.destroy() |
| 223 | } |
| 224 | client.subscriptions.delete(roomId) |
| 225 | } |
| 226 | |
| 227 | this._sendFrame(client, { id, ok: true, data: { unsubscribed: roomId } }) |
| 228 | } |
| 229 | |
| 230 | _sendFrame (client, obj) { |
| 231 | try { |
| 232 | const json = JSON.stringify(obj) |
| 233 | client.framed.write(b4a.from(json, 'utf8')) |
| 234 | } catch {} |
| 235 | } |
| 236 | } |
| 237 | |
| 238 | function normalizeEvent (event, nameMap) { |
| 239 | if (!event || typeof event !== 'object') return null |
| 240 | |
| 241 | const text = event.chat?.text || event.text || '' |
| 242 | const generic = event.message?.generic || null |
| 243 | |
| 244 | let system = null |
| 245 | if (generic && typeof generic === 'object') { |
| 246 | for (const v of Object.values(generic)) { |
| 247 | if (v !== null && v !== false) { system = generic; break } |
| 248 | } |
| 249 | } else if (generic) { |
| 250 | system = generic |
| 251 | } |
| 252 | |
| 253 | if (!text && !system) return null |
| 254 | |
| 255 | const memberId = event.memberId || 'unknown' |
| 256 | |
| 257 | return { |
| 258 | memberId, |
| 259 | name: (nameMap && nameMap[memberId]) || memberId.slice(0, 8), |
| 260 | text, |
| 261 | timestamp: event.timestamp || Date.now(), |
| 262 | deleted: event.deleted || false, |
| 263 | system, |
| 264 | seq: event.seq ?? event.clock ?? null, |
| 265 | id: event.id || null |
| 266 | } |
| 267 | } |