import Pipe from 'bare-pipe'
import tcp from 'bare-tcp'
import FramedStream from 'framed-stream'
import b4a from 'b4a'
import fs from 'fs'
export class UdsTransport {
constructor (router, opts = {}) {
this.router = router
this.socketPath = opts.socketPath || '/tmp/keet-daemon.sock'
// LOOPBACK TCP, for Windows: Godot's Windows sockets cannot open a Unix
// socket (NetSocketWinSock::open rejects Family::UNIX). { host, port, token }.
// 127.0.0.1 is reachable by EVERY local process, which the 0600 socket file is
// not -- so the first frame of every connection must be { auth: token } or the
// connection is closed unanswered. The token is per launch, handed to the
// clients this process spawns.
this.tcp = opts.tcp || null
this.port = null
this.server = null
this._clients = new Set()
}
start () {
if (this.tcp) return this._startTcp()
return new Promise((resolve, reject) => {
// Remove stale socket file
try { fs.unlinkSync(this.socketPath) } catch {}
this.server = Pipe.createServer((rawPipe) => {
this._onConnection(rawPipe)
})
this.server.on('error', (err) => {
console.error('[keet-daemon] UDS server error:', err.message)
reject(err)
})
this.server.listen(this.socketPath, () => {
// Owner-only permissions (security audit D-F6)
try { fs.chmodSync(this.socketPath, 0o600) } catch {}
console.log(`[keet-daemon] UDS listening on ${this.socketPath} (mode 0600)`)
resolve()
})
})
}
_startTcp () {
const { host = '127.0.0.1', port = 0, token } = this.tcp
if (typeof token !== 'string' || token.length < 32) {
return Promise.reject(new Error('tcp transport needs a token of at least 32 characters'))
}
return new Promise((resolve, reject) => {
this.server = tcp.createServer((socket) => {
this._onConnection(socket, { needsAuth: true })
})
this.server.on('error', (err) => {
console.error('[keet-daemon] TCP server error:', err.message)
reject(err)
})
this.server.listen(port, host, () => {
this.port = this.server.address().port
console.log(`[keet-daemon] TCP listening on ${host}:${this.port} (token-authenticated)`)
resolve()
})
})
}
stop () {
for (const client of this._clients) {
client.destroy()
}
this._clients.clear()
if (this.server) this.server.close()
if (!this.tcp) {
try { fs.unlinkSync(this.socketPath) } catch {}
}
}
broadcastEvent (eventName, data) {
for (const client of this._clients) {
if (!client.authed) continue // an unauthenticated TCP client hears nothing
this._sendFrame(client, { event: eventName, data })
}
}
destroyAllSubscriptions () {
for (const client of this._clients) {
for (const [, sub] of client.subscriptions) {
if (sub.stream && typeof sub.stream.destroy === 'function') {
sub.stream.destroy()
}
}
client.subscriptions.clear()
}
}
_onConnection (rawPipe, opts = {}) {
const framed = new FramedStream(rawPipe)
const client = {
framed,
rawPipe,
authed: !opts.needsAuth,
subscriptions: new Map(), // roomId -> { stream, highSeq, initialized }
destroy () {
// Clean up all subscriptions
for (const [, sub] of this.subscriptions) {
if (sub.stream && typeof sub.stream.destroy === 'function') {
sub.stream.destroy()
}
}
this.subscriptions.clear()
framed.destroy()
}
}
this._clients.add(client)
framed.on('data', (buf) => {
this._onFrame(client, buf)
})
framed.on('error', (err) => {
if (err.message !== 'Stream destroyed') {
console.error('[keet-daemon] UDS client error:', err.message)
}
})
framed.on('close', () => {
client.destroy()
this._clients.delete(client)
})
}
async _onFrame (client, buf) {
if (!client.authed) {
// Nothing is answered before auth -- not even a parse error -- and one
// wrong frame ends the connection.
let auth = null
try { auth = JSON.parse(b4a.toString(buf, 'utf8')).auth } catch {}
if (typeof auth === 'string' && tokensEqual(auth, this.tcp.token)) {
client.authed = true
} else {
console.error('[keet-daemon] TCP client failed auth -- closing')
client.destroy()
}
return
}
let msg
try {
msg = JSON.parse(b4a.toString(buf, 'utf8'))
} catch {
this._sendFrame(client, { id: null, ok: false, error: 'Invalid JSON' })
return
}
const { id, method, params } = msg
if (!method) {
this._sendFrame(client, { id, ok: false, error: 'Missing "method"' })
return
}
// Handle subscribe/unsubscribe at transport level
if (method === 'subscribe') {
await this._handleSubscribe(client, id, params || {})
return
}
if (method === 'unsubscribe') {
this._handleUnsubscribe(client, id, params || {})
return
}
if (method === 'file.subscribe') {
await this._handleFileSubscribe(client, id, params || {})
return
}
if (method === 'file.unsubscribe') {
this._handleFileUnsubscribe(client, id, params || {})
return
}
try {
const result = await this.router.handle(method, params || {})
this._sendFrame(client, { id, ...result })
} catch (err) {
this._sendFrame(client, { id, ok: false, error: err.message || String(err) })
}
}
async _handleSubscribe (client, id, params) {
const { roomId } = params
if (!roomId) {
this._sendFrame(client, { id, ok: false, error: 'Missing roomId' })
return
}
// If already subscribed, unsubscribe first
if (client.subscriptions.has(roomId)) {
const old = client.subscriptions.get(roomId)
if (old.stream && typeof old.stream.destroy === 'function') {
old.stream.destroy()
}
client.subscriptions.delete(roomId)
}
// Resolve member names
const nameMap = await this.router._getMemberNames(roomId)
// Get subscription stream
const stream = this.router.subscribe(roomId)
const sub = { stream, highSeq: -1, initialized: false }
client.subscriptions.set(roomId, sub)
// Confirm subscription
this._sendFrame(client, { id, ok: true, data: { subscribed: roomId } })
if (!stream) return
if (typeof stream[Symbol.asyncIterator] === 'function') {
;(async () => {
try {
for await (const snapshot of stream) {
if (!snapshot) continue
const messages = Array.isArray(snapshot) ? snapshot : [snapshot]
if (!sub.initialized) {
// First snapshot = current state — record high water mark
for (const event of messages) {
const seq = event.seq ?? event.clock ?? -1
if (seq > sub.highSeq) sub.highSeq = seq
}
sub.initialized = true
continue
}
// Emit only newer messages
for (const event of messages) {
const seq = event.seq ?? event.clock ?? -1
if (seq <= sub.highSeq) continue
sub.highSeq = seq
const msg = normalizeEvent(event, nameMap)
if (msg) {
this._sendFrame(client, { event: 'message', roomId, data: msg })
}
}
}
} catch (err) {
if (err.message !== 'Stream destroyed') {
this._sendFrame(client, { event: 'error', roomId, data: { error: err.message } })
}
}
})()
} else if (stream && typeof stream.on === 'function') {
stream.on('data', (snapshot) => {
if (!snapshot) return
const messages = Array.isArray(snapshot) ? snapshot : [snapshot]
if (!sub.initialized) {
for (const event of messages) {
const seq = event.seq ?? event.clock ?? -1
if (seq > sub.highSeq) sub.highSeq = seq
}
sub.initialized = true
return
}
for (const event of messages) {
const seq = event.seq ?? event.clock ?? -1
if (seq <= sub.highSeq) continue
sub.highSeq = seq
const msg = normalizeEvent(event, nameMap)
if (msg) {
this._sendFrame(client, { event: 'message', roomId, data: msg })
}
}
})
stream.on('error', (err) => {
this._sendFrame(client, { event: 'error', roomId, data: { error: err.message } })
})
stream.on('close', () => {
client.subscriptions.delete(roomId)
})
}
}
async _handleFileSubscribe (client, id, params) {
const { roomId } = params
if (!roomId) {
this._sendFrame(client, { id, ok: false, error: 'Missing roomId' })
return
}
const subKey = 'files:' + roomId
// Tear down existing file subscription for this room
if (client.subscriptions.has(subKey)) {
const old = client.subscriptions.get(subKey)
if (old.stream && typeof old.stream.destroy === 'function') old.stream.destroy()
client.subscriptions.delete(subKey)
}
// Try subscribeFiles if available on keet-core
let stream = null
try {
if (typeof this.router.backend.api.core.subscribeFiles === 'function') {
stream = this.router.backend.api.core.subscribeFiles(roomId)
}
} catch {}
if (!stream) {
this._sendFrame(client, { id, ok: true, data: { subscribed: subKey, note: 'no subscribeFiles on core' } })
return
}
const sub = { stream, highSeq: -1, initialized: false }
client.subscriptions.set(subKey, sub)
this._sendFrame(client, { id, ok: true, data: { subscribed: subKey } })
if (typeof stream[Symbol.asyncIterator] === 'function') {
;(async () => {
try {
for await (const snapshot of stream) {
if (!snapshot) continue
const events = Array.isArray(snapshot) ? snapshot : [snapshot]
for (const ev of events) {
this._sendFrame(client, { event: 'file_update', roomId, data: ev })
}
}
} catch (err) {
if (err.message !== 'Stream destroyed') {
this._sendFrame(client, { event: 'error', roomId, data: { error: err.message } })
}
}
})()
} else if (stream && typeof stream.on === 'function') {
stream.on('data', (snapshot) => {
const events = Array.isArray(snapshot) ? snapshot : [snapshot]
for (const ev of events) {
this._sendFrame(client, { event: 'file_update', roomId, data: ev })
}
})
stream.on('error', (err) => {
this._sendFrame(client, { event: 'error', roomId, data: { error: err.message } })
})
stream.on('close', () => {
client.subscriptions.delete(subKey)
})
}
}
_handleFileUnsubscribe (client, id, params) {
const { roomId } = params
if (!roomId) {
this._sendFrame(client, { id, ok: false, error: 'Missing roomId' })
return
}
const subKey = 'files:' + roomId
const sub = client.subscriptions.get(subKey)
if (sub) {
if (sub.stream && typeof sub.stream.destroy === 'function') sub.stream.destroy()
client.subscriptions.delete(subKey)
}
this._sendFrame(client, { id, ok: true, data: { unsubscribed: subKey } })
}
_handleUnsubscribe (client, id, params) {
const { roomId } = params
if (!roomId) {
this._sendFrame(client, { id, ok: false, error: 'Missing roomId' })
return
}
const sub = client.subscriptions.get(roomId)
if (sub) {
if (sub.stream && typeof sub.stream.destroy === 'function') {
sub.stream.destroy()
}
client.subscriptions.delete(roomId)
}
this._sendFrame(client, { id, ok: true, data: { unsubscribed: roomId } })
}
_sendFrame (client, obj) {
try {
const json = JSON.stringify(obj)
client.framed.write(b4a.from(json, 'utf8'))
} catch {}
}
}
// Constant-time for equal lengths, so a wrong token cannot be found a byte at a time.
function tokensEqual (a, b) {
if (a.length !== b.length) return false
let diff = 0
for (let i = 0; i < a.length; i++) diff |= a.charCodeAt(i) ^ b.charCodeAt(i)
return diff === 0
}
function normalizeEvent (event, nameMap) {
if (!event || typeof event !== 'object') return null
const text = event.chat?.text || event.text || ''
const generic = event.message?.generic || null
let system = null
if (generic && typeof generic === 'object') {
for (const v of Object.values(generic)) {
if (v !== null && v !== false) { system = generic; break }
}
} else if (generic) {
system = generic
}
const files = extractFiles(event)
if (!text && !system && !files) return null
const memberId = event.memberId || 'unknown'
return {
memberId,
name: (nameMap && nameMap[memberId]) || memberId.slice(0, 8),
text,
timestamp: event.timestamp || Date.now(),
deleted: event.deleted || false,
system,
files,
seq: event.seq ?? event.clock ?? null,
id: event.id || null
}
}
function extractFiles (msg) {
let files = msg.message?.files
if (Array.isArray(files) && files.length > 0) {
return files.map(f => normalizeFileEntry(f))
}
if (msg.file && typeof msg.file === 'object' && msg.file.metadata) {
return [normalizeFileEntry(msg.file)]
}
return null
}
function normalizeFileEntry (f) {
const meta = f.metadata || {}
const pointer = f.pointer || {}
const dims = meta.dimensions || null
const blob = pointer.externalBlob?.blob || pointer.blob || null
// Use filename as unique ID — externalBlob.id is shared across all files in a room
const fallbackId = meta.name || pointer.externalBlob?.id || null
const entry = f.entry || {}
const entryPreview = entry.preview || {}
const smallLink = entryPreview.small?.httpLink || null
const mediumLink = entryPreview.medium?.httpLink || null
const largeLink = entryPreview.large?.httpLink || null
const fullLink = entry.httpLink || null
return {
id: fallbackId,
name: meta.name || null,
type: meta.mimetype || null,
size: blob?.byteLength || 0,
dimensions: dims ? { width: dims.width, height: dims.height } : null,
httpLink: fullLink,
preview: (smallLink || mediumLink || largeLink) ? {
small: smallLink,
medium: mediumLink,
large: largeLink
} : null
}
}