375 lines
10 KiB
JavaScript
375 lines
10 KiB
JavaScript
import b4a from 'b4a'
|
|
import { randomBytes } from 'bare-crypto'
|
|
import {
|
|
setupBareOsChatChannel,
|
|
BARE_OS_CHAT_WIRE_SCHEMA_VERSION,
|
|
BARE_OS_CHAT_EVT_TEXT,
|
|
PROTOCOL_CHAT_CHANNEL_NAME,
|
|
bareOsProtMuxChatChannelEnabled
|
|
} from 'bare-os-protocol'
|
|
|
|
/** @typedef {{ chan: import('protomux').Channel, mux: import('protomux').Protomux, socket: any, id: string | null, chatChan?: import('protomux').Channel | null }} SwarmPeer */
|
|
|
|
/**
|
|
* Stock default: swarm chat is **on** unless disabled via env (see `bare-os-protocol` `bareOsProtMuxChatChannelEnabled`).
|
|
* @param {Record<string, string | undefined>} [env]
|
|
*/
|
|
export function bareOsChatMuxEnabled(env = globalThis.process?.env) {
|
|
return bareOsProtMuxChatChannelEnabled(env || {})
|
|
}
|
|
|
|
/**
|
|
* Swarm Protomux chat transport on `disk` (receive + relay + broadcast). Stock-on when mux env allows.
|
|
* Independent of the **`bare-os-chat`** initd unit (that unit is only started after identity unlock).
|
|
*
|
|
* @param {import('./swarm-disk.js').SwarmDisk} disk
|
|
* @param {Record<string, string | undefined>} [env]
|
|
*/
|
|
export function ensureDiskBareOsChatTransport(disk, env = {}) {
|
|
if (!disk || !bareOsChatMuxEnabled(globalThis.process?.env)) return
|
|
const merged = {
|
|
.../** @type {Record<string, string | undefined>} */ (
|
|
globalThis.process?.env || {}
|
|
),
|
|
...env
|
|
}
|
|
disk.bareOsChatService = createBareOsChatService({ env: merged })
|
|
}
|
|
|
|
/**
|
|
* @param {Record<string, string | undefined>} env
|
|
*/
|
|
function chatGossipTtlDefault(env) {
|
|
const raw = String(env.BARE_OS_CHAT_GOSSIP_TTL ?? '').trim()
|
|
const n = raw ? Number.parseInt(raw, 10) : NaN
|
|
if (Number.isFinite(n) && n >= 0 && n <= 32) return n
|
|
return 4
|
|
}
|
|
|
|
/**
|
|
* @param {Record<string, string | undefined>} env
|
|
*/
|
|
function chatHistoryMax(env) {
|
|
const raw = String(env.BARE_OS_CHAT_HISTORY_MAX ?? '').trim()
|
|
const n = raw ? Number.parseInt(raw, 10) : NaN
|
|
if (Number.isFinite(n) && n >= 16 && n <= 10000) return n
|
|
return 512
|
|
}
|
|
|
|
/**
|
|
* @param {Record<string, string | undefined>} env
|
|
*/
|
|
function chatMaxBodyBytes(env) {
|
|
const raw = String(env.BARE_OS_CHAT_MAX_BODY_BYTES ?? '').trim()
|
|
const n = raw ? Number.parseInt(raw, 10) : NaN
|
|
if (Number.isFinite(n) && n >= 256 && n <= 65536) return n
|
|
return 4096
|
|
}
|
|
|
|
/**
|
|
* Global swarm chat (host-side): fan-out, gossip, dedupe, metrics.
|
|
*/
|
|
export function createBareOsChatService(opts = {}) {
|
|
const env = opts.env || globalThis.process?.env || {}
|
|
const gossipTtlMax = chatGossipTtlDefault(
|
|
/** @type {Record<string, string | undefined>} */ (env)
|
|
)
|
|
const historyMax = chatHistoryMax(
|
|
/** @type {Record<string, string | undefined>} */ (env)
|
|
)
|
|
const maxBodyBytes = chatMaxBodyBytes(
|
|
/** @type {Record<string, string | undefined>} */ (env)
|
|
)
|
|
|
|
/** @type {Set<(ev: Record<string, unknown>) => void>} */
|
|
const subscribers = new Set()
|
|
/** @type {Map<string, number>} evt dedupe -> expireAtMs */
|
|
const seenEvt = new Map()
|
|
/** @type {Array<Record<string, unknown>>} */
|
|
const history = []
|
|
/** @type {{ rxEvent: number, txEvent: number, droppedRate: number, droppedVerify: number }} */
|
|
const metrics = {
|
|
rxEvent: 0,
|
|
txEvent: 0,
|
|
droppedRate: 0,
|
|
droppedVerify: 0
|
|
}
|
|
/** @type {Map<string, { tokens: number, resetAt: number }>} */
|
|
const ratePeer = new Map()
|
|
const RATE_WINDOW_MS = 2000
|
|
const RATE_MAX_MSG = 24
|
|
|
|
function trimDedupe() {
|
|
const now = Date.now()
|
|
for (const [k, exp] of seenEvt) {
|
|
if (exp < now) seenEvt.delete(k)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @param {string} peerKey
|
|
*/
|
|
function allowRate(peerKey) {
|
|
const now = Date.now()
|
|
let r = ratePeer.get(peerKey)
|
|
if (!r || r.resetAt < now) {
|
|
r = { tokens: RATE_MAX_MSG, resetAt: now + RATE_WINDOW_MS }
|
|
ratePeer.set(peerKey, r)
|
|
}
|
|
if (r.tokens <= 0) {
|
|
metrics.droppedRate++
|
|
return false
|
|
}
|
|
r.tokens--
|
|
return true
|
|
}
|
|
|
|
function pushHistory(rec) {
|
|
history.push(rec)
|
|
while (history.length > historyMax) history.shift()
|
|
}
|
|
|
|
/**
|
|
* @param {Uint8Array} pk
|
|
* @param {Uint8Array} evtId
|
|
*/
|
|
function dedupeKey(pk, evtId) {
|
|
return `${b4a.toString(pk, 'hex')}:${b4a.toString(evtId, 'hex')}`
|
|
}
|
|
|
|
/**
|
|
* @param {import('./swarm-disk.js').SwarmDisk} disk
|
|
* @param {SwarmPeer} fromPeer
|
|
* @param {Record<string, unknown>} evt
|
|
*/
|
|
function relayEvent(disk, fromPeer, evt) {
|
|
const ttl =
|
|
typeof evt.ttl === 'number' ? evt.ttl : gossipTtlMax
|
|
if (ttl <= 0) return
|
|
const next = { ...evt, ttl: ttl - 1 }
|
|
for (const p of disk.peers) {
|
|
if (p === fromPeer) continue
|
|
const ch = p.chatChan
|
|
if (!ch || !ch.messages || !ch.messages[4]) continue
|
|
try {
|
|
ch.messages[4].send(next)
|
|
metrics.txEvent++
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @param {import('./swarm-disk.js').SwarmDisk} disk
|
|
* @param {SwarmPeer} fromPeer
|
|
* @param {Record<string, unknown>} evt
|
|
*/
|
|
function ingestEvent(disk, fromPeer, evt) {
|
|
const senderPk = /** @type {Uint8Array | undefined} */ (evt.senderPk)
|
|
const evtId = /** @type {Uint8Array | undefined} */ (evt.evtId)
|
|
if (!senderPk || senderPk.byteLength !== 32 || !evtId || evtId.byteLength !== 16) {
|
|
return
|
|
}
|
|
const body = String(evt.body ?? '')
|
|
if (body.byteLength > maxBodyBytes) return
|
|
|
|
const sock = fromPeer.socket
|
|
const ttlMaxGossip = Math.max(1, gossipTtlMax)
|
|
if (sock && sock.remotePublicKey && senderPk && senderPk.byteLength === 32) {
|
|
if (!b4a.equals(sock.remotePublicKey, senderPk)) {
|
|
const ttlNum = typeof evt.ttl === 'number' ? evt.ttl : -1
|
|
// Direct frames: Noise remote PK must equal claimed senderPk.
|
|
// Relayed hops: ttl is decremented before forward; mux peer ≠ origin — allow ttl < configured max TTL.
|
|
if (!(ttlNum >= 0 && ttlNum < ttlMaxGossip)) {
|
|
metrics.droppedVerify++
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
const pk = fromPeer.id || b4a.toString(senderPk, 'hex')
|
|
if (!allowRate(pk)) return
|
|
|
|
trimDedupe()
|
|
const dk = dedupeKey(senderPk, evtId)
|
|
const now = Date.now()
|
|
if (seenEvt.has(dk)) return
|
|
seenEvt.set(dk, now + 120_000)
|
|
|
|
metrics.rxEvent++
|
|
const rec = {
|
|
...evt,
|
|
body,
|
|
receivedAtMs: now,
|
|
fromPeerKey: pk
|
|
}
|
|
pushHistory(rec)
|
|
for (const fn of subscribers) {
|
|
try {
|
|
fn(rec)
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
}
|
|
|
|
relayEvent(disk, fromPeer, evt)
|
|
}
|
|
|
|
return {
|
|
PROTOCOL_CHAT_CHANNEL_NAME,
|
|
metrics,
|
|
history() {
|
|
return [...history]
|
|
},
|
|
presence() {
|
|
/** @type {Record<string, { displayName: string, roomId: string }>} */
|
|
const out = {}
|
|
return out
|
|
},
|
|
rooms() {
|
|
return ['general']
|
|
},
|
|
subscribe(fn) {
|
|
subscribers.add(fn)
|
|
return () => subscribers.delete(fn)
|
|
},
|
|
/**
|
|
* @param {import('./swarm-disk.js').SwarmDisk} disk
|
|
* @param {import('protomux').Protomux} mux
|
|
* @param {any} socket
|
|
* @param {SwarmPeer} peer
|
|
*/
|
|
pairOnMux(disk, mux, socket, peer) {
|
|
setupBareOsChatChannel(mux, {
|
|
onHello(_m, chan) {
|
|
try {
|
|
chan.messages[1].send({
|
|
chatSchemaVersion: 1,
|
|
maxPayloadBytes: 16384
|
|
})
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
},
|
|
onHelloAck(_m, chan) {
|
|
try {
|
|
const id16 = randomBytes(16)
|
|
let pk = b4a.alloc(32)
|
|
if (socket?.publicKey && socket.publicKey.byteLength === 32) {
|
|
pk = b4a.from(socket.publicKey)
|
|
} else if (
|
|
disk.localNoiseWirePk &&
|
|
disk.localNoiseWirePk.byteLength === 32
|
|
) {
|
|
pk = disk.localNoiseWirePk
|
|
}
|
|
chan.messages[2].send({
|
|
roomId: 'general',
|
|
displayName: String(env.USER || env.LOGNAME || 'peer'),
|
|
senderPk: pk,
|
|
clientInstanceId: id16
|
|
})
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
},
|
|
onJoin(_m, _chan) {},
|
|
onLeave(_m, _chan) {},
|
|
onEvent(m, _chan) {
|
|
try {
|
|
disk.protomuxChatChannelRxTotal =
|
|
(disk.protomuxChatChannelRxTotal || 0) + 1
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
ingestEvent(disk, peer, m)
|
|
},
|
|
onControl(_m, _chan) {},
|
|
onChannelOpened(chan) {
|
|
peer.chatChan = chan
|
|
try {
|
|
chan.messages[0].send({
|
|
chatSchemaVersion: 1,
|
|
swarmChatCapability: true
|
|
})
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
mux.stream?.once?.('close', () => {
|
|
peer.chatChan = null
|
|
})
|
|
}
|
|
})
|
|
},
|
|
/**
|
|
* @param {import('./swarm-disk.js').SwarmDisk} disk
|
|
* @param {string} text
|
|
* @param {{ displayName?: string, senderPk?: Uint8Array }} [meta]
|
|
*/
|
|
broadcastLocal(disk, text, meta = {}) {
|
|
const evtId = randomBytes(16)
|
|
let pk =
|
|
meta.senderPk && meta.senderPk.byteLength === 32 ? meta.senderPk : null
|
|
if (!pk) {
|
|
for (const p of disk.peers) {
|
|
const sk = p.socket?.publicKey
|
|
if (sk && sk.byteLength === 32) {
|
|
pk = b4a.from(sk)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
if (!pk && disk.localNoiseWirePk && disk.localNoiseWirePk.byteLength === 32) {
|
|
pk = disk.localNoiseWirePk
|
|
}
|
|
if (!pk || pk.byteLength !== 32) {
|
|
pk = b4a.alloc(32)
|
|
}
|
|
const evt = {
|
|
schemaVersion: BARE_OS_CHAT_WIRE_SCHEMA_VERSION,
|
|
evtKind: BARE_OS_CHAT_EVT_TEXT,
|
|
roomId: 'general',
|
|
evtId,
|
|
tsMs: Date.now(),
|
|
senderPk: pk,
|
|
displayName: meta.displayName || String(env.USER || 'local'),
|
|
body: text,
|
|
ttl: gossipTtlMax,
|
|
sigDetached: null
|
|
}
|
|
pushHistory({ ...evt, local: true, receivedAtMs: Date.now() })
|
|
for (const fn of subscribers) {
|
|
try {
|
|
fn({ ...evt, local: true })
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
}
|
|
for (const p of disk.peers) {
|
|
const ch = p.chatChan
|
|
if (!ch || !ch.messages || !ch.messages[4]) continue
|
|
void ch.fullyOpened().then((opened) => {
|
|
if (!opened) return
|
|
if (p.chatChan !== ch) return
|
|
try {
|
|
ch.messages[4].send(evt)
|
|
metrics.txEvent++
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
})
|
|
}
|
|
},
|
|
snapshotMetrics() {
|
|
return {
|
|
...metrics,
|
|
gossipTtlDefault: gossipTtlMax,
|
|
historyMax,
|
|
maxBodyBytes,
|
|
protocol: PROTOCOL_CHAT_CHANNEL_NAME
|
|
}
|
|
}
|
|
}
|
|
}
|