Merge branch 'gossip-routing-2'

This commit is contained in:
callebtc
2026-01-09 23:56:33 +07:00
17 changed files with 1180 additions and 103 deletions
@@ -109,11 +109,19 @@ class BluetoothConnectionManager(
}
// Connection caps: enforce on change
connectionScope.launch {
dbg.maxConnectionsOverall.collect {
dbg.maxConnectionsOverall.collect { maxOverall ->
if (!isActive) return@collect
// 1. Enforce client limits (handled by tracker)
connectionTracker.enforceConnectionLimits()
// Also enforce server side best-effort
serverManager.enforceServerLimit(dbg.maxServerConnections.value)
// 2. Enforce overall limit on server connections if needed
// (Tracker knows about all connections but can't disconnect servers directly)
val maxServer = dbg.maxServerConnections.value
val excessServers = connectionTracker.getExcessServerConnections(maxServer, maxOverall)
excessServers.forEach { device ->
Log.d(TAG, "Disconnecting server ${device.address} due to overall cap")
serverManager.disconnectDevice(device)
}
}
}
connectionScope.launch {
@@ -123,9 +131,18 @@ class BluetoothConnectionManager(
}
}
connectionScope.launch {
dbg.maxServerConnections.collect {
dbg.maxServerConnections.collect { maxServer ->
if (!isActive) return@collect
serverManager.enforceServerLimit(dbg.maxServerConnections.value)
// Enforce server specific limit
serverManager.enforceServerLimit(maxServer)
// Also check if this change puts us over the overall limit
val maxOverall = dbg.maxConnectionsOverall.value
val excessServers = connectionTracker.getExcessServerConnections(maxServer, maxOverall)
excessServers.forEach { device ->
Log.d(TAG, "Disconnecting server ${device.address} due to overall cap")
serverManager.disconnectDevice(device)
}
}
}
} catch (_: Exception) { }
@@ -261,6 +278,16 @@ class BluetoothConnectionManager(
)
}
fun sendToPeer(peerID: String, routed: RoutedPacket): Boolean {
if (!isActive) return false
return packetBroadcaster.sendToPeer(
peerID,
routed,
serverManager.getGattServer(),
serverManager.getCharacteristic()
)
}
fun cancelTransfer(transferId: String): Boolean {
return packetBroadcaster.cancelTransfer(transferId)
}
@@ -229,11 +229,17 @@ class BluetoothConnectionTracker(
*/
fun getConnectedDeviceCount(): Int = connectedDevices.size
/**
* Check if connection limit is reached
*/
/**
* Check if connection limit is reached
*/
fun isConnectionLimitReached(): Boolean {
return connectedDevices.size >= powerManager.getMaxConnections()
// Respect debug override if set
val dbg = try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance() } catch (_: Exception) { null }
val maxConnections = dbg?.maxConnectionsOverall?.value ?: powerManager.getMaxConnections()
return connectedDevices.size >= maxConnections
}
/**
@@ -244,10 +250,9 @@ class BluetoothConnectionTracker(
val dbg = try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance() } catch (_: Exception) { null }
val maxOverall = dbg?.maxConnectionsOverall?.value ?: powerManager.getMaxConnections()
val maxClient = dbg?.maxClientConnections?.value ?: maxOverall
val maxServer = dbg?.maxServerConnections?.value ?: maxOverall
// Note: maxServer is handled by GattServerManager, but we need to respect overall limit here too
val clients = connectedDevices.values.filter { it.isClient }
val servers = connectedDevices.values.filter { !it.isClient }
// Enforce client cap first (we can actively disconnect)
if (clients.size > maxClient) {
@@ -259,22 +264,56 @@ class BluetoothConnectionTracker(
}
}
// Note: server cap enforced in GattServerManager (we don't have server handle here)
// Enforce overall cap by disconnecting oldest client connections
// Re-check overall cap after client cleanup
if (connectedDevices.size > maxOverall) {
Log.i(TAG, "Enforcing overall cap: ${connectedDevices.size} > $maxOverall")
val excess = connectedDevices.size - maxOverall
val toDisconnect = connectedDevices.values
.filter { it.isClient } // only clients from here
// Prefer disconnecting clients first to satisfy overall cap
val clientsToDisconnect = connectedDevices.values
.filter { it.isClient }
.sortedBy { it.connectedAt }
.take(excess)
toDisconnect.forEach { dc ->
clientsToDisconnect.forEach { dc ->
Log.d(TAG, "Disconnecting client ${dc.device.address} due to overall cap")
dc.gatt?.disconnect()
}
}
}
/**
* Get excess server connections that should be disconnected to satisfy limits.
* This allows the Manager to coordinate server disconnects since Tracker doesn't control the server.
*/
fun getExcessServerConnections(maxServer: Int, maxOverall: Int): List<BluetoothDevice> {
val servers = connectedDevices.values.filter { !it.isClient }
val excessList = mutableListOf<BluetoothDevice>()
// 1. Check server specific limit
if (servers.size > maxServer) {
val excessCount = servers.size - maxServer
val toRemove = servers.sortedBy { it.connectedAt }.take(excessCount)
excessList.addAll(toRemove.map { it.device })
}
// 2. Check overall limit (considering we might have already removed some above)
// We need to count how many connections we will have after the above removals
val currentTotal = connectedDevices.size
val plannedRemovals = excessList.size
val projectedTotal = currentTotal - plannedRemovals
if (projectedTotal > maxOverall) {
val furtherExcess = projectedTotal - maxOverall
// We can only remove servers here. Clients are handled in enforceConnectionLimits.
// Filter out devices we already planned to remove
val remainingServers = servers.filter { s -> excessList.none { it.address == s.device.address } }
val toRemove = remainingServers.sortedBy { it.connectedAt }.take(furtherExcess)
excessList.addAll(toRemove.map { it.device })
}
return excessList
}
/**
* Clean up a specific device connection
@@ -49,15 +49,28 @@ class BluetoothGattServerManager(
fun enforceServerLimit(maxServer: Int) {
if (maxServer <= 0) return
try {
val subs = connectionTracker.getSubscribedDevices()
if (subs.size > maxServer) {
val excess = subs.size - maxServer
subs.take(excess).forEach { d ->
try { gattServer?.cancelConnection(d) } catch (_: Exception) { }
// Use connection tracker to get actual connected server devices
val servers = connectionTracker.getConnectedDevices().values.filter { !it.isClient }
if (servers.size > maxServer) {
val excess = servers.size - maxServer
// Disconnect oldest
servers.sortedBy { it.connectedAt }.take(excess).forEach { d ->
try { gattServer?.cancelConnection(d.device) } catch (_: Exception) { }
}
}
} catch (_: Exception) { }
}
/**
* Disconnect a specific device (used by ConnectionManager to enforce overall limits)
*/
fun disconnectDevice(device: BluetoothDevice) {
try {
gattServer?.cancelConnection(device)
} catch (e: Exception) {
Log.w(TAG, "Error disconnecting device ${device.address}: ${e.message}")
}
}
/**
* Start GATT server
@@ -122,9 +135,10 @@ class BluetoothGattServerManager(
// Try to cancel any active connections explicitly before closing
try {
val devices = connectionTracker.getSubscribedDevices()
devices.forEach { d ->
try { gattServer?.cancelConnection(d) } catch (_: Exception) { }
// Disconnect ALL server connections
val servers = connectionTracker.getConnectedDevices().values.filter { !it.isClient }
servers.forEach { d ->
try { gattServer?.cancelConnection(d.device) } catch (_: Exception) { }
}
} catch (_: Exception) { }
@@ -526,6 +526,10 @@ class BluetoothMeshService(private val context: Context) {
connectionManager.broadcastPacket(routed)
}
override fun sendToPeer(peerID: String, routed: RoutedPacket): Boolean {
return connectionManager.sendToPeer(peerID, routed)
}
override fun handleRequestSync(routed: RoutedPacket) {
// Decode request and respond with missing packets
val fromPeer = routed.peerID ?: return
@@ -1020,11 +1024,25 @@ class BluetoothMeshService(private val context: Context) {
// Create iOS-compatible IdentityAnnouncement with TLV encoding
val announcement = IdentityAnnouncement(nickname, staticKey, signingKey)
val tlvPayload = announcement.encode()
var tlvPayload = announcement.encode()
if (tlvPayload == null) {
Log.e(TAG, "Failed to encode announcement as TLV")
return@launch
}
// Append gossip TLV containing up to 10 direct neighbors (compact IDs)
try {
val directPeers = getDirectPeerIDsForGossip()
if (directPeers.isNotEmpty()) {
val gossip = com.bitchat.android.services.meshgraph.GossipTLV.encodeNeighbors(directPeers)
tlvPayload = tlvPayload + gossip
}
// Always update our own node in the mesh graph with the neighbor list we used
try {
com.bitchat.android.services.meshgraph.MeshGraphService.getInstance()
.updateFromAnnouncement(myPeerID, nickname, directPeers, System.currentTimeMillis().toULong())
} catch (_: Exception) { }
} catch (_: Exception) { }
val announcePacket = BitchatPacket(
type = MessageType.ANNOUNCE.value,
@@ -1069,11 +1087,25 @@ class BluetoothMeshService(private val context: Context) {
// Create iOS-compatible IdentityAnnouncement with TLV encoding
val announcement = IdentityAnnouncement(nickname, staticKey, signingKey)
val tlvPayload = announcement.encode()
var tlvPayload = announcement.encode()
if (tlvPayload == null) {
Log.e(TAG, "Failed to encode peer announcement as TLV")
return
}
// Append gossip TLV containing up to 10 direct neighbors (compact IDs)
try {
val directPeers = getDirectPeerIDsForGossip()
if (directPeers.isNotEmpty()) {
val gossip = com.bitchat.android.services.meshgraph.GossipTLV.encodeNeighbors(directPeers)
tlvPayload = tlvPayload + gossip
}
// Always update our own node in the mesh graph with the neighbor list we used
try {
com.bitchat.android.services.meshgraph.MeshGraphService.getInstance()
.updateFromAnnouncement(myPeerID, nickname, directPeers, System.currentTimeMillis().toULong())
} catch (_: Exception) { }
} catch (_: Exception) { }
val packet = BitchatPacket(
type = MessageType.ANNOUNCE.value,
@@ -1095,6 +1127,20 @@ class BluetoothMeshService(private val context: Context) {
try { gossipSyncManager.onPublicPacketSeen(signedPacket) } catch (_: Exception) { }
}
/**
* Collect up to 10 direct neighbors for gossip TLV.
*/
private fun getDirectPeerIDsForGossip(): List<String> {
return try {
// Prefer verified peers that are currently marked as direct
val verified = peerManager.getVerifiedPeers()
val direct = verified.filter { it.value.isDirectConnection }.keys.toList()
direct.take(10)
} catch (_: Exception) {
emptyList()
}
}
/**
* Send leave announcement
*/
@@ -1277,21 +1323,38 @@ class BluetoothMeshService(private val context: Context) {
*/
private fun signPacketBeforeBroadcast(packet: BitchatPacket): BitchatPacket {
return try {
// Optionally compute and attach a source route for addressed packets
val withRoute = try {
val rec = packet.recipientID
if (rec != null && !rec.contentEquals(SpecialRecipients.BROADCAST)) {
val dest = rec.joinToString("") { b -> "%02x".format(b) }
val path = com.bitchat.android.services.meshgraph.RoutePlanner.shortestPath(myPeerID, dest)
if (path != null && path.size >= 3) {
// Exclude first (sender) and last (recipient); only intermediates
val intermediates = path.subList(1, path.size - 1)
val hopsBytes = intermediates.map { hexStringToByteArray(it) }
Log.d(TAG, "✅ Signed packet type ${packet.type} (route ${hopsBytes.size} hops: $intermediates)")
// Attach route and upgrade to v2 (required for HAS_ROUTE flag)
packet.copy(route = hopsBytes, version = 2u)
} else packet.copy(route = null)
} else packet
} catch (_: Exception) { packet }
// Get the canonical packet data for signing (without signature)
val packetDataForSigning = packet.toBinaryDataForSigning()
val packetDataForSigning = withRoute.toBinaryDataForSigning()
if (packetDataForSigning == null) {
Log.w(TAG, "Failed to encode packet type ${packet.type} for signing, sending unsigned")
return packet
return withRoute
}
// Sign the packet data using our signing key
val signature = encryptionService.signData(packetDataForSigning)
if (signature != null) {
Log.d(TAG, "✅ Signed packet type ${packet.type} (signature ${signature.size} bytes)")
packet.copy(signature = signature)
withRoute.copy(signature = signature)
} else {
Log.w(TAG, "Failed to sign packet type ${packet.type}, sending unsigned")
packet
withRoute
}
} catch (e: Exception) {
Log.w(TAG, "Error signing packet type ${packet.type}: ${e.message}, sending unsigned")
@@ -283,6 +283,46 @@ class BluetoothPacketBroadcaster(
}
}
}
/**
* Targeted send to a specific peer (by peerID) if directly connected.
* Returns true if sent to at least one matching connection.
*/
fun sendToPeer(
targetPeerID: String,
routed: RoutedPacket,
gattServer: BluetoothGattServer?,
characteristic: BluetoothGattCharacteristic?
): Boolean {
val packet = routed.packet
val data = packet.toBinaryData() ?: return false
val typeName = MessageType.fromValue(packet.type)?.name ?: packet.type.toString()
val senderPeerID = routed.peerID ?: packet.senderID.toHexString()
val incomingAddr = routed.relayAddress
val incomingPeer = incomingAddr?.let { connectionTracker.addressPeerMap[it] }
val senderNick = senderPeerID.let { pid -> nicknameResolver?.invoke(pid) }
// Try server-side connections first
val targetDevice = connectionTracker.getSubscribedDevices()
.firstOrNull { connectionTracker.addressPeerMap[it.address] == targetPeerID }
if (targetDevice != null) {
if (notifyDevice(targetDevice, data, gattServer, characteristic)) {
logPacketRelay(typeName, senderPeerID, senderNick, incomingPeer, incomingAddr, targetPeerID, targetDevice.address, packet.ttl)
return true
}
}
// Try client-side connections next
val targetConn = connectionTracker.getConnectedDevices().values
.firstOrNull { connectionTracker.addressPeerMap[it.device.address] == targetPeerID }
if (targetConn != null) {
if (writeToDeviceConn(targetConn, data)) {
logPacketRelay(typeName, senderPeerID, senderNick, incomingPeer, incomingAddr, targetPeerID, targetConn.device.address, packet.ttl)
return true
}
}
return false
}
/**
* Internal broadcast implementation - runs in serialized actor context
@@ -338,7 +378,7 @@ class BluetoothPacketBroadcaster(
val subscribedDevices = connectionTracker.getSubscribedDevices()
val connectedDevices = connectionTracker.getConnectedDevices()
Log.i(TAG, "Broadcasting packet type ${packet.type} to ${subscribedDevices.size} server + ${connectedDevices.size} client connections")
Log.i(TAG, "Broadcasting packet v${packet.version} type ${packet.type} to ${subscribedDevices.size} server + ${connectedDevices.size} client connections")
val senderID = String(packet.senderID).replace("\u0000", "")
@@ -77,8 +77,31 @@ class FragmentManager {
val fragmentID = FragmentPayload.generateFragmentID()
// iOS: stride(from: 0, to: fullData.count, by: maxFragmentSize)
val fragmentChunks = stride(0, fullData.size, MAX_FRAGMENT_SIZE) { offset ->
val endOffset = minOf(offset + MAX_FRAGMENT_SIZE, fullData.size)
// Calculate dynamic fragment size to fit in MTU (512)
// Packet = Header + Sender + Recipient + Route + FragmentHeader + Payload + PaddingBuffer
val hasRoute = packet.route != null
val version = if (hasRoute) 2 else 1
val headerSize = if (version == 2) 15 else 13
val senderSize = 8
val recipientSize = if (packet.recipientID != null) 8 else 0
// Route: 1 byte count + 8 bytes per hop
val routeSize = if (hasRoute) (1 + (packet.route?.size ?: 0) * 8) else 0
val fragmentHeaderSize = 13 // FragmentPayload header
val paddingBuffer = 16 // MessagePadding.optimalBlockSize adds 16 bytes overhead
// 512 - Overhead
val packetOverhead = headerSize + senderSize + recipientSize + routeSize + fragmentHeaderSize + paddingBuffer
val maxDataSize = (512 - packetOverhead).coerceAtMost(MAX_FRAGMENT_SIZE)
if (maxDataSize <= 0) {
Log.e(TAG, "❌ Calculated maxDataSize is non-positive ($maxDataSize). Route too large?")
return emptyList()
}
Log.d(TAG, "📏 Dynamic fragment size: $maxDataSize (MAX: $MAX_FRAGMENT_SIZE, Overhead: $packetOverhead)")
val fragmentChunks = stride(0, fullData.size, maxDataSize) { offset ->
val endOffset = minOf(offset + maxDataSize, fullData.size)
fullData.sliceArray(offset..<endOffset)
}
@@ -98,13 +121,16 @@ class FragmentManager {
)
// iOS: MessageType.fragment.rawValue (single fragment type)
// Fix: Fragments must inherit source route and use v2 if routed
val fragmentPacket = BitchatPacket(
version = if (packet.route != null) 2u else 1u,
type = MessageType.FRAGMENT.value,
ttl = packet.ttl,
senderID = packet.senderID,
recipientID = packet.recipientID,
timestamp = packet.timestamp,
payload = fragmentPayload.encode(),
route = packet.route,
signature = null // iOS: signature: nil
)
@@ -286,6 +286,13 @@ class MessageHandler(private val myPeerID: String, private val appContext: andro
previousPeerID = null
)
// Update mesh graph from gossip neighbors (only if TLV present)
try {
val neighborsOrNull = com.bitchat.android.services.meshgraph.GossipTLV.decodeNeighborsFromAnnouncementPayload(packet.payload)
com.bitchat.android.services.meshgraph.MeshGraphService.getInstance()
.updateFromAnnouncement(peerID, nickname, neighborsOrNull, packet.timestamp)
} catch (_: Exception) { }
Log.d(TAG, "✅ Processed verified TLV announce: stored identity for $peerID")
return isFirstAnnounce
}
@@ -112,6 +112,9 @@ class PacketProcessor(private val myPeerID: String) {
override fun broadcastPacket(routed: RoutedPacket) {
delegate?.relayPacket(routed)
}
override fun sendToPeer(peerID: String, routed: RoutedPacket): Boolean {
return delegate?.sendToPeer(peerID, routed) ?: false
}
}
}
@@ -323,4 +326,5 @@ interface PacketProcessorDelegate {
fun sendAnnouncementToPeer(peerID: String)
fun sendCachedMessages(peerID: String)
fun relayPacket(routed: RoutedPacket)
fun sendToPeer(peerID: String, routed: RoutedPacket): Boolean
}
@@ -65,9 +65,40 @@ class PacketRelayManager(private val myPeerID: String) {
val relayPacket = packet.copy(ttl = (packet.ttl - 1u).toUByte())
Log.d(TAG, "Decremented TTL from ${packet.ttl} to ${relayPacket.ttl}")
// Source-based routing: if route is set and includes us, try targeted next-hop forwarding
val route = relayPacket.route
if (!route.isNullOrEmpty()) {
// Check for duplicate hops to prevent routing loops
if (route.map { it.toHexString() }.toSet().size < route.size) {
Log.w(TAG, "Packet with duplicate hops dropped")
return
}
val myIdBytes = hexStringToPeerBytes(myPeerID)
val index = route.indexOfFirst { it.contentEquals(myIdBytes) }
if (index >= 0) {
val nextHopIdHex: String? = run {
val nextIndex = index + 1
if (nextIndex < route.size) {
route[nextIndex].toHexString()
} else {
// We are the last intermediate; try final recipient as next hop
relayPacket.recipientID?.toHexString()
}
}
if (nextHopIdHex != null) {
val success = try { delegate?.sendToPeer(nextHopIdHex, RoutedPacket(relayPacket, peerID, routed.relayAddress)) } catch (_: Exception) { false } ?: false
if (success) {
Log.i(TAG, "📦 Source-route relay: ${myPeerID.take(8)} -> ${nextHopIdHex.take(8)} (type ${'$'}{packet.type}, TTL ${'$'}{relayPacket.ttl})")
return
} else {
Log.w(TAG, "Source-route next hop ${nextHopIdHex.take(8)} not directly connected; falling back to broadcast")
}
}
}
}
// Apply relay logic based on packet type and debug switch
val shouldRelay = isRelayEnabled() && shouldRelayPacket(relayPacket, peerID)
if (shouldRelay) {
relayPacket(RoutedPacket(relayPacket, peerID, routed.relayAddress))
} else {
@@ -170,4 +201,17 @@ interface PacketRelayManagerDelegate {
// Packet operations
fun broadcastPacket(routed: RoutedPacket)
fun sendToPeer(peerID: String, routed: RoutedPacket): Boolean
}
private fun hexStringToPeerBytes(hex: String): ByteArray {
val result = ByteArray(8)
var idx = 0
var out = 0
while (idx + 1 < hex.length && out < 8) {
val b = hex.substring(idx, idx + 2).toIntOrNull(16)?.toByte() ?: 0
result[out++] = b
idx += 2
}
return result
}
@@ -59,7 +59,8 @@ data class BitchatPacket(
val timestamp: ULong,
val payload: ByteArray,
var signature: ByteArray? = null, // Changed from val to var for packet signing
var ttl: UByte
var ttl: UByte,
var route: List<ByteArray>? = null // Optional source route: ordered list of peerIDs (8 bytes each), not including sender and final recipient
) : Parcelable {
constructor(
@@ -97,6 +98,7 @@ data class BitchatPacket(
timestamp = timestamp,
payload = payload,
signature = null, // Remove signature for signing
route = route,
ttl = com.bitchat.android.util.AppConstants.SYNC_TTL_HOPS // Use fixed TTL=0 for signing to ensure relay compatibility
)
return BinaryProtocol.encode(unsignedPacket)
@@ -149,6 +151,11 @@ data class BitchatPacket(
if (!signature.contentEquals(other.signature)) return false
} else if (other.signature != null) return false
if (ttl != other.ttl) return false
if (route != null || other.route != null) {
val a = route?.map { it.toList() } ?: emptyList()
val b = other.route?.map { it.toList() } ?: emptyList()
if (a != b) return false
}
return true
}
@@ -162,6 +169,7 @@ data class BitchatPacket(
result = 31 * result + payload.contentHashCode()
result = 31 * result + (signature?.contentHashCode() ?: 0)
result = 31 * result + ttl.hashCode()
result = 31 * result + (route?.fold(1) { acc, bytes -> 31 * acc + bytes.contentHashCode() } ?: 0)
return result
}
}
@@ -180,6 +188,7 @@ object BinaryProtocol {
const val HAS_RECIPIENT: UByte = 0x01u
const val HAS_SIGNATURE: UByte = 0x02u
const val IS_COMPRESSED: UByte = 0x04u
const val HAS_ROUTE: UByte = 0x08u
}
private fun getHeaderSize(version: UByte): Int {
@@ -193,12 +202,12 @@ object BinaryProtocol {
try {
// Try to compress payload if beneficial
var payload = packet.payload
var originalPayloadSize: UShort? = null
var originalPayloadSize: Int? = null
var isCompressed = false
if (CompressionUtil.shouldCompress(payload)) {
CompressionUtil.compress(payload)?.let { compressedPayload ->
originalPayloadSize = payload.size.toUShort()
originalPayloadSize = payload.size
payload = compressedPayload
isCompressed = true
}
@@ -208,8 +217,12 @@ object BinaryProtocol {
val headerSize = getHeaderSize(packet.version)
val recipientBytes = if (packet.recipientID != null) RECIPIENT_ID_SIZE else 0
val signatureBytes = if (packet.signature != null) SIGNATURE_SIZE else 0
val payloadBytes = payload.size + if (isCompressed) 2 else 0
val capacity = headerSize + SENDER_ID_SIZE + recipientBytes + payloadBytes + signatureBytes + 16 // small slack
val sizeFieldBytes = if (isCompressed) (if (packet.version >= 2u.toUByte()) 4 else 2) else 0
val payloadBytes = payload.size + sizeFieldBytes
val routeBytes = if (!packet.route.isNullOrEmpty() && packet.version >= 2u.toUByte()) {
1 + (packet.route!!.size.coerceAtMost(255) * SENDER_ID_SIZE)
} else 0
val capacity = headerSize + SENDER_ID_SIZE + recipientBytes + payloadBytes + signatureBytes + routeBytes + 16 // small slack
val buffer = ByteBuffer.allocate(capacity.coerceAtLeast(512)).apply { order(ByteOrder.BIG_ENDIAN) }
// Header
@@ -231,10 +244,14 @@ object BinaryProtocol {
if (isCompressed) {
flags = flags or Flags.IS_COMPRESSED
}
// HAS_ROUTE is only supported for v2+ packets
if (!packet.route.isNullOrEmpty() && packet.version >= 2u.toUByte()) {
flags = flags or Flags.HAS_ROUTE
}
buffer.put(flags.toByte())
// Payload length (2 or 4 bytes, big-endian) - includes original size if compressed
val payloadDataSize = payload.size + if (isCompressed) 2 else 0
val payloadDataSize = payload.size + sizeFieldBytes
if (packet.version >= 2u.toUByte()) {
buffer.putInt(payloadDataSize) // 4 bytes for v2+
} else {
@@ -256,12 +273,26 @@ object BinaryProtocol {
buffer.put(ByteArray(RECIPIENT_ID_SIZE - recipientBytes.size))
}
}
// Route (optional, v2+ only): 1 byte count + N*8 bytes
if (packet.version >= 2u.toUByte()) {
packet.route?.let { routeList ->
val cleaned = routeList.map { bytes -> bytes.take(SENDER_ID_SIZE).toByteArray().let { if (it.size < SENDER_ID_SIZE) it + ByteArray(SENDER_ID_SIZE - it.size) else it } }
val count = cleaned.size.coerceAtMost(255)
buffer.put(count.toByte())
cleaned.take(count).forEach { hop -> buffer.put(hop) }
}
}
// Payload (with original size prepended if compressed)
if (isCompressed) {
val originalSize = originalPayloadSize
if (originalSize != null) {
buffer.putShort(originalSize.toShort())
if (packet.version >= 2u.toUByte()) {
buffer.putInt(originalSize.toInt())
} else {
buffer.putShort(originalSize.toShort())
}
}
}
buffer.put(payload)
@@ -324,6 +355,8 @@ object BinaryProtocol {
val hasRecipient = (flags and Flags.HAS_RECIPIENT) != 0u.toUByte()
val hasSignature = (flags and Flags.HAS_SIGNATURE) != 0u.toUByte()
val isCompressed = (flags and Flags.IS_COMPRESSED) != 0u.toUByte()
// HAS_ROUTE is only valid for v2+ packets; ignore the flag for v1
val hasRoute = (version >= 2u.toUByte()) && (flags and Flags.HAS_ROUTE) != 0u.toUByte()
// Payload length - version-dependent (2 or 4 bytes)
val payloadLength = if (version >= 2u.toUByte()) {
@@ -335,6 +368,22 @@ object BinaryProtocol {
// Calculate expected total size
var expectedSize = headerSize + SENDER_ID_SIZE + payloadLength.toInt()
if (hasRecipient) expectedSize += RECIPIENT_ID_SIZE
var routeCount = 0
if (hasRoute) {
// Peek count (1 byte) without consuming buffer for now
// The buffer is currently positioned at the start of SenderID (after fixed header)
// We must skip SenderID and RecipientID (if present) to find the route count
val currentPos = buffer.position()
var routeOffset = currentPos + SENDER_ID_SIZE
if (hasRecipient) {
routeOffset += RECIPIENT_ID_SIZE
}
if (raw.size >= routeOffset + 1) {
routeCount = raw[routeOffset].toUByte().toInt()
}
expectedSize += 1 + (routeCount * SENDER_ID_SIZE)
}
if (hasSignature) expectedSize += SIGNATURE_SIZE
if (raw.size < expectedSize) return null
@@ -350,15 +399,42 @@ object BinaryProtocol {
recipientBytes
} else null
// Route (optional)
val route: List<ByteArray>? = if (hasRoute) {
val count = buffer.get().toUByte().toInt()
val hops = mutableListOf<ByteArray>()
repeat(count) {
val hop = ByteArray(SENDER_ID_SIZE)
buffer.get(hop)
hops.add(hop)
}
hops
} else null
// Payload
val payload = if (isCompressed) {
// First 2 bytes are original size
if (payloadLength.toInt() < 2) return null
val originalSize = buffer.getShort().toInt()
val lengthFieldBytes = if (version >= 2u.toUByte()) 4 else 2
if (payloadLength.toInt() < lengthFieldBytes) return null
val originalSize = if (version >= 2u.toUByte()) {
buffer.getInt()
} else {
buffer.getShort().toUShort().toInt()
}
// Compressed payload
val compressedPayload = ByteArray(payloadLength.toInt() - 2)
val compressedSize = payloadLength.toInt() - lengthFieldBytes
val compressedPayload = ByteArray(compressedSize)
buffer.get(compressedPayload)
// Security check: Compression bomb protection
if (compressedSize > 0) {
val ratio = originalSize.toDouble() / compressedSize.toDouble()
if (ratio > 50_000.0) {
Log.w("BinaryProtocol", "🚫 Suspicious compression ratio: ${ratio}:1")
return null
}
}
// Decompress
CompressionUtil.decompress(compressedPayload, originalSize) ?: return null
@@ -383,7 +459,8 @@ object BinaryProtocol {
timestamp = timestamp,
payload = payload,
signature = signature,
ttl = ttl
ttl = ttl,
route = route
)
} catch (e: Exception) {
@@ -0,0 +1,77 @@
package com.bitchat.android.services.meshgraph
import android.util.Log
/**
* Gossip TLV helpers for embedding direct neighbor peer IDs in ANNOUNCE payloads.
* Uses compact TLV: [type=0x04][len=1 byte][value=N*8 bytes of peerIDs]
*/
object GossipTLV {
// TLV type for a compact list of direct neighbor peerIDs (each 8 bytes)
const val DIRECT_NEIGHBORS_TYPE: UByte = 0x04u
/**
* Encode up to 10 unique peerIDs (hex string up to 16 chars) as TLV value.
*/
fun encodeNeighbors(peerIDs: List<String>): ByteArray {
val unique = peerIDs.distinct().take(10)
val valueBytes = unique.flatMap { id -> hexStringPeerIdTo8Bytes(id).toList() }.toByteArray()
if (valueBytes.size > 255) {
// Safety check, though 10*8 = 80 bytes, so well under 255
Log.w("GossipTLV", "Neighbors value exceeds 255, truncating")
}
return byteArrayOf(DIRECT_NEIGHBORS_TYPE.toByte(), valueBytes.size.toByte()) + valueBytes
}
/**
* Scan a TLV-encoded announce payload and extract neighbor peerIDs.
* Returns null if the TLV is not present at all; returns an empty list if present with length 0.
*/
fun decodeNeighborsFromAnnouncementPayload(payload: ByteArray): List<String>? {
val result = mutableListOf<String>()
var offset = 0
while (offset + 2 <= payload.size) {
val type = payload[offset].toUByte()
val len = payload[offset + 1].toUByte().toInt()
offset += 2
if (offset + len > payload.size) break
val value = payload.sliceArray(offset until offset + len)
offset += len
if (type == DIRECT_NEIGHBORS_TYPE) {
// Value is N*8 bytes of peer IDs
var pos = 0
while (pos + 8 <= value.size) {
val idBytes = value.sliceArray(pos until pos + 8)
result.add(bytesToPeerIdHex(idBytes))
pos += 8
}
return result // present (possibly empty)
}
}
// Not present
return null
}
private fun hexStringPeerIdTo8Bytes(hexString: String): ByteArray {
val clean = hexString.lowercase().take(16)
val result = ByteArray(8) { 0 }
var idx = 0
var out = 0
while (idx + 1 < clean.length && out < 8) {
val byteStr = clean.substring(idx, idx + 2)
val b = byteStr.toIntOrNull(16)?.toByte() ?: 0
result[out++] = b
idx += 2
}
return result
}
private fun bytesToPeerIdHex(bytes: ByteArray): String {
val sb = StringBuilder()
for (b in bytes.take(8)) {
sb.append(String.format("%02x", b))
}
return sb.toString()
}
}
@@ -0,0 +1,111 @@
package com.bitchat.android.services.meshgraph
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import java.util.concurrent.ConcurrentHashMap
/**
* Maintains an internal graph of the mesh based on gossip.
* Nodes are peers (peerID), edges are direct connections.
*/
class MeshGraphService private constructor() {
data class GraphNode(val peerID: String, val nickname: String?)
data class GraphEdge(val a: String, val b: String, val isConfirmed: Boolean, val confirmedBy: String? = null)
data class GraphSnapshot(val nodes: List<GraphNode>, val edges: List<GraphEdge>)
// Map peerID -> nickname (may be null if unknown)
private val nicknames = ConcurrentHashMap<String, String?>()
// Announcements: peerID -> set of neighbor peerIDs that *this* peer claims to see
private val announcements = ConcurrentHashMap<String, Set<String>>()
// Latest announcement timestamp per peer (ULong from packet)
private val lastUpdate = ConcurrentHashMap<String, ULong>()
private val _graphState = MutableStateFlow(GraphSnapshot(emptyList(), emptyList()))
val graphState: StateFlow<GraphSnapshot> = _graphState.asStateFlow()
/**
* Update graph from a verified announcement.
* Replaces previous neighbors for origin if this is newer (by timestamp).
*/
fun updateFromAnnouncement(originPeerID: String, originNickname: String?, neighborsOrNull: List<String>?, timestamp: ULong) {
synchronized(this) {
// Always update nickname if provided
if (originNickname != null) nicknames[originPeerID] = originNickname
// If no neighbors TLV present, do not modify edges or timestamps
if (neighborsOrNull == null) {
publishSnapshot()
return
}
// Newer-only replacement per origin (based on TLV-bearing announcements only)
val prevTs = lastUpdate[originPeerID]
if (prevTs != null && prevTs >= timestamp) {
// Older or equal TLV-bearing update: ignore
return
}
lastUpdate[originPeerID] = timestamp
// Update what originPeerID announces
// Filter out self-loops just in case
val newSet = neighborsOrNull.distinct().take(10).filter { it != originPeerID }.toSet()
announcements[originPeerID] = newSet
publishSnapshot()
}
}
fun updateNickname(peerID: String, nickname: String?) {
if (nickname == null) return
nicknames[peerID] = nickname
publishSnapshot()
}
private fun publishSnapshot() {
// Collect all known nodes from nicknames and announcements
val allNodes = mutableSetOf<String>()
allNodes.addAll(nicknames.keys)
announcements.forEach { (origin, neighbors) ->
allNodes.add(origin)
allNodes.addAll(neighbors)
}
val nodeList = allNodes.map { GraphNode(it, nicknames[it]) }.sortedBy { it.peerID }
val edges = mutableListOf<GraphEdge>()
val processedPairs = mutableSetOf<Pair<String, String>>()
// We only care about connections that exist in at least one direction.
// So iterating through all entries in `announcements` covers every declared edge.
announcements.forEach { (source, targets) ->
targets.forEach { target ->
val pair = if (source <= target) source to target else target to source
if (processedPairs.add(pair)) {
// This is a new pair we haven't evaluated yet
val (a, b) = pair
val aAnnouncesB = announcements[a]?.contains(b) == true
val bAnnouncesA = announcements[b]?.contains(a) == true
if (aAnnouncesB && bAnnouncesA) {
edges.add(GraphEdge(a, b, isConfirmed = true))
} else if (aAnnouncesB) {
edges.add(GraphEdge(a, b, isConfirmed = false, confirmedBy = a))
} else if (bAnnouncesA) {
edges.add(GraphEdge(a, b, isConfirmed = false, confirmedBy = b))
}
}
}
}
val sortedEdges = edges.sortedWith(compareBy({ it.a }, { it.b }))
_graphState.value = GraphSnapshot(nodeList, sortedEdges)
}
companion object {
@Volatile private var INSTANCE: MeshGraphService? = null
fun getInstance(): MeshGraphService = INSTANCE ?: synchronized(this) {
INSTANCE ?: MeshGraphService().also { INSTANCE = it }
}
}
}
@@ -0,0 +1,67 @@
package com.bitchat.android.services.meshgraph
import android.util.Log
import java.util.PriorityQueue
/**
* Computes shortest paths on the current mesh graph snapshot using Dijkstra.
* Assumes unit edge weights.
*/
object RoutePlanner {
private const val TAG = "RoutePlanner"
/**
* Return full path [src, ..., dst] if reachable, else null.
*/
fun shortestPath(src: String, dst: String): List<String>? {
if (src == dst) return listOf(src)
val snapshot = MeshGraphService.getInstance().graphState.value
val neighbors = mutableMapOf<String, MutableSet<String>>()
// Only consider confirmed edges for routing
snapshot.edges.filter { it.isConfirmed }.forEach { e ->
neighbors.getOrPut(e.a) { mutableSetOf() }.add(e.b)
neighbors.getOrPut(e.b) { mutableSetOf() }.add(e.a)
}
// Ensure nodes known even if isolated
snapshot.nodes.forEach { n -> neighbors.putIfAbsent(n.peerID, mutableSetOf()) }
if (!neighbors.containsKey(src) || !neighbors.containsKey(dst)) return null
val dist = mutableMapOf<String, Int>()
val prev = mutableMapOf<String, String?>()
val pq = PriorityQueue<Pair<String, Int>>(compareBy { it.second })
neighbors.keys.forEach { v ->
dist[v] = if (v == src) 0 else Int.MAX_VALUE
prev[v] = null
}
pq.add(src to 0)
while (pq.isNotEmpty()) {
val (u, d) = pq.poll()
if (d > (dist[u] ?: Int.MAX_VALUE)) continue
if (u == dst) break
neighbors[u]?.forEach { v ->
val alt = d + 1
if (alt < (dist[v] ?: Int.MAX_VALUE)) {
dist[v] = alt
prev[v] = u
pq.add(v to alt)
}
}
}
if ((dist[dst] ?: Int.MAX_VALUE) == Int.MAX_VALUE) return null
val path = mutableListOf<String>()
var cur: String? = dst
while (cur != null) {
path.add(cur)
cur = prev[cur]
}
path.reverse()
Log.d(TAG, "Computed path $path")
return path
}
}
@@ -27,13 +27,120 @@ import androidx.compose.ui.unit.dp
import androidx.compose.ui.unit.sp
import androidx.compose.ui.draw.rotate
import com.bitchat.android.mesh.BluetoothMeshService
import com.bitchat.android.services.meshgraph.MeshGraphService
import kotlinx.coroutines.launch
import androidx.compose.ui.graphics.toArgb
import androidx.compose.ui.graphics.drawscope.drawIntoCanvas
import androidx.compose.ui.graphics.nativeCanvas
import androidx.compose.ui.res.stringResource
import com.bitchat.android.R
import androidx.compose.ui.platform.LocalContext
import com.bitchat.android.service.MeshServicePreferences
import com.bitchat.android.service.MeshForegroundService
@Composable
fun MeshTopologySection() {
val colorScheme = MaterialTheme.colorScheme
val graphService = remember { MeshGraphService.getInstance() }
val snapshot by graphService.graphState.collectAsState()
Surface(shape = RoundedCornerShape(12.dp), color = colorScheme.surfaceVariant.copy(alpha = 0.2f)) {
Column(Modifier.padding(16.dp), verticalArrangement = Arrangement.spacedBy(8.dp)) {
Row(verticalAlignment = Alignment.CenterVertically, horizontalArrangement = Arrangement.spacedBy(8.dp)) {
Icon(Icons.Filled.SettingsEthernet, contentDescription = null, tint = Color(0xFF8E8E93))
Text("mesh topology", fontFamily = FontFamily.Monospace, fontSize = 14.sp, fontWeight = FontWeight.Medium)
}
val nodes = snapshot.nodes
val edges = snapshot.edges
val empty = nodes.isEmpty()
if (empty) {
Text("no gossip yet", fontFamily = FontFamily.Monospace, fontSize = 11.sp, color = colorScheme.onSurface.copy(alpha = 0.6f))
} else {
androidx.compose.foundation.Canvas(Modifier.fillMaxWidth().height(220.dp).background(colorScheme.surface.copy(alpha = 0.4f))) {
val w = size.width
val h = size.height
val cx = w / 2f
val cy = h / 2f
val radius = (minOf(w, h) * 0.36f)
val n = nodes.size
if (n == 1) {
// Single node centered
drawCircle(color = Color(0xFF00C851), radius = 12f, center = androidx.compose.ui.geometry.Offset(cx, cy))
} else {
// Circular layout
val positions = nodes.mapIndexed { i, node ->
val angle = (2 * Math.PI * i.toDouble()) / n
val x = cx + (radius * Math.cos(angle)).toFloat()
val y = cy + (radius * Math.sin(angle)).toFloat()
node.peerID to androidx.compose.ui.geometry.Offset(x, y)
}.toMap()
// Draw edges
edges.forEach { e ->
val p1 = positions[e.a]
val p2 = positions[e.b]
if (p1 != null && p2 != null) {
if (e.isConfirmed) {
drawLine(color = Color(0xFF4A90E2), start = p1, end = p2, strokeWidth = 2f)
} else {
// Unconfirmed: draw "solid" from declarer, "dashed" from other
val start = if (e.confirmedBy == e.a) p1 else p2
val end = if (e.confirmedBy == e.a) p2 else p1
val midX = (start.x + end.x) / 2
val midY = (start.y + end.y) / 2
val mid = androidx.compose.ui.geometry.Offset(midX, midY)
// Solid half
drawLine(color = Color(0xFF4A90E2), start = start, end = mid, strokeWidth = 2f)
// Dotted half
drawLine(
color = Color(0xFF4A90E2),
start = mid,
end = end,
strokeWidth = 2f,
pathEffect = androidx.compose.ui.graphics.PathEffect.dashPathEffect(floatArrayOf(5f, 5f), 0f)
)
}
}
}
// Draw nodes
nodes.forEach { node ->
val pos = positions[node.peerID] ?: androidx.compose.ui.geometry.Offset(cx, cy)
drawCircle(color = Color(0xFF00C851), radius = 10f, center = pos)
}
// Draw labels near nodes (nickname or short ID)
val labelColor = colorScheme.onSurface.toArgb()
val textSizePx = 10.sp.toPx()
drawIntoCanvas { canvas ->
val paint = android.graphics.Paint().apply {
isAntiAlias = true
color = labelColor
textSize = textSizePx
}
nodes.forEach { node ->
val pos = positions[node.peerID] ?: androidx.compose.ui.geometry.Offset(cx, cy)
val label = (node.nickname?.takeIf { it.isNotBlank() } ?: node.peerID.take(8))
canvas.nativeCanvas.drawText(label, pos.x + 12f, pos.y - 12f, paint)
}
}
}
}
// Label list for clarity under the canvas
LazyColumn(modifier = Modifier.fillMaxWidth().heightIn(max = 140.dp)) {
items(nodes) { node ->
val label = "${node.peerID.take(8)}${node.nickname ?: "unknown"}"
Text(label, fontFamily = FontFamily.Monospace, fontSize = 11.sp, color = colorScheme.onSurface.copy(alpha = 0.85f))
}
}
}
}
}
}
private enum class GraphMode { OVERALL, PER_DEVICE, PER_PEER }
@OptIn(ExperimentalMaterial3Api::class, ExperimentalLayoutApi::class)
@@ -144,6 +251,11 @@ fun DebugSettingsSheet(
}
}
// Mesh topology visualization (moved below verbose logging)
item {
MeshTopologySection()
}
// GATT controls
item {
Surface(shape = RoundedCornerShape(12.dp), color = colorScheme.surfaceVariant.copy(alpha = 0.2f)) {
@@ -353,7 +465,7 @@ fun DebugSettingsSheet(
kotlinx.coroutines.delay(1000)
}
}
// Helper functions moved to top-level composable below to avoid scope issues
// Render two blocks: Incoming and Outgoing
@@ -493,9 +605,6 @@ fun DebugSettingsSheet(
}
}
// Connected devices
item {
Surface(shape = RoundedCornerShape(12.dp), color = colorScheme.surfaceVariant.copy(alpha = 0.2f)) {