share gossip

This commit is contained in:
callebtc
2026-01-04 22:17:45 +07:00
parent 5983d700cd
commit 27c6dc5ed9
9 changed files with 147 additions and 17 deletions
@@ -168,6 +168,10 @@ class BluetoothMeshService(private val context: Context) : MeshService, Transpor
// Direct injection prevents routing loops (bridge handles source check)
meshCore.sendFromBridge(packet)
}
override fun sendToPeer(peerID: String, packet: BitchatPacket) {
connectionManager.sendPacketToPeer(peerID, packet)
}
/**
* Start periodic debug logging every 10 seconds
@@ -56,6 +56,7 @@ class MeshCore(
val gossipSyncManager: GossipSyncManager =
sharedGossipManager ?: GossipSyncManager(myPeerID = myPeerID, scope = scope, configProvider = gossipConfigProvider)
private val ownsGossipManager: Boolean = sharedGossipManager == null
var delegate: MeshDelegate? = null
@@ -74,6 +75,7 @@ class MeshCore(
override fun sendPacketToPeer(peerID: String, packet: BitchatPacket) {
transport.sendPacketToPeer(peerID, packet)
TransportBridgeService.sendToPeer(transport.id, peerID, packet)
}
override fun signPacketForBroadcast(packet: BitchatPacket): BitchatPacket {
@@ -87,7 +89,9 @@ class MeshCore(
if (isActive) return
isActive = true
startPeriodicBroadcastAnnounce()
gossipSyncManager.start()
if (ownsGossipManager) {
gossipSyncManager.start()
}
}
fun stopCore() {
@@ -95,7 +99,9 @@ class MeshCore(
isActive = false
announceJob?.cancel()
announceJob = null
gossipSyncManager.stop()
if (ownsGossipManager) {
gossipSyncManager.stop()
}
}
fun shutdown() {
@@ -91,7 +91,8 @@ class NoiseEncryptionService(private val context: Context) {
}
// Initialize session manager
sessionManager = NoiseSessionManager(staticIdentityPrivateKey, staticIdentityPublicKey)
val localPeerID = calculateFingerprint(staticIdentityPublicKey).take(16)
sessionManager = NoiseSessionManager(staticIdentityPrivateKey, staticIdentityPublicKey, localPeerID)
// Set up session callbacks
sessionManager.onSessionEstablished = { peerID, remoteStaticKey ->
@@ -153,6 +153,9 @@ class NoiseSession(
// Session state
private var state: NoiseSessionState = NoiseSessionState.Uninitialized
private val creationTime = System.currentTimeMillis()
private var handshakeStartMs: Long? = null
private var lastHandshakeActivityMs: Long? = null
private var handshakeMessage1: ByteArray? = null
// Session counters
private var currentPattern = 0;
@@ -195,6 +198,15 @@ class NoiseSession(
fun isEstablished(): Boolean = state is NoiseSessionState.Established
fun isHandshaking(): Boolean = state is NoiseSessionState.Handshaking
fun getCreationTime(): Long = creationTime
fun isInitiatorRole(): Boolean = isInitiator
fun getHandshakeStartMs(): Long? = handshakeStartMs
fun getLastHandshakeActivityMs(): Long? = lastHandshakeActivityMs
internal fun getHandshakeMessage1(): ByteArray? = handshakeMessage1?.clone()
internal fun setLastHandshakeActivityForTest(timestampMs: Long) {
lastHandshakeActivityMs = timestampMs
}
init {
try {
@@ -317,12 +329,17 @@ class NoiseSession(
// Initialize handshake as initiator
initializeNoiseHandshake(HandshakeState.INITIATOR)
state = NoiseSessionState.Handshaking
if (handshakeStartMs == null) {
handshakeStartMs = System.currentTimeMillis()
}
lastHandshakeActivityMs = System.currentTimeMillis()
val messageBuffer = ByteArray(XX_MESSAGE_1_SIZE)
val handshakeStateLocal = handshakeState ?: throw IllegalStateException("Handshake state is null")
val messageLength = handshakeStateLocal.writeMessage(messageBuffer, 0, null, 0, 0)
currentPattern++
val firstMessage = messageBuffer.copyOf(messageLength)
handshakeMessage1 = firstMessage
// Validate message size matches XX pattern expectations
if (firstMessage.size != XX_MESSAGE_1_SIZE) {
@@ -351,12 +368,16 @@ class NoiseSession(
if (state == NoiseSessionState.Uninitialized && !isInitiator) {
initializeNoiseHandshake(HandshakeState.RESPONDER)
state = NoiseSessionState.Handshaking
if (handshakeStartMs == null) {
handshakeStartMs = System.currentTimeMillis()
}
Log.d(TAG, "Initialized as RESPONDER for XX handshake with $peerID")
}
if (state != NoiseSessionState.Handshaking) {
throw IllegalStateException("Invalid state for handshake: $state")
}
lastHandshakeActivityMs = System.currentTimeMillis()
val handshakeStateLocal = handshakeState ?: throw IllegalStateException("Handshake state is null")
@@ -8,11 +8,14 @@ import java.util.concurrent.ConcurrentHashMap
*/
class NoiseSessionManager(
private val localStaticPrivateKey: ByteArray,
private val localStaticPublicKey: ByteArray
private val localStaticPublicKey: ByteArray,
private val localPeerID: String
) {
companion object {
private const val TAG = "NoiseSessionManager"
private const val HANDSHAKE_TIMEOUT_MS = 20_000L
private const val HANDSHAKE_MESSAGE_1_SIZE = 32
}
private val sessions = ConcurrentHashMap<String, NoiseSession>()
@@ -51,11 +54,30 @@ class NoiseSessionManager(
/**
* SIMPLIFIED: Initiate handshake - no tie breaker, just start
*/
fun initiateHandshake(peerID: String): ByteArray {
fun initiateHandshake(peerID: String): ByteArray? {
Log.d(TAG, "initiateHandshake($peerID)")
// Remove any existing session first
removeSession(peerID)
val now = System.currentTimeMillis()
val existing = getSession(peerID)
if (existing != null) {
when {
existing.isEstablished() -> {
Log.d(TAG, "Handshake already established with $peerID, skipping initiate")
return null
}
existing.isHandshaking() -> {
if (!isHandshakeStale(existing, now)) {
Log.d(TAG, "Handshake already in progress with $peerID, not restarting")
return null
}
Log.d(TAG, "Handshake with $peerID is stale; restarting")
removeSession(peerID)
}
else -> {
removeSession(peerID)
}
}
}
// Create new session as initiator
val session = NoiseSession(
@@ -85,6 +107,23 @@ class NoiseSessionManager(
try {
var session = getSession(peerID)
// Collision handling: both sides initiated and we received message 1
if (session != null &&
session.isHandshaking() &&
session.isInitiatorRole() &&
message.size == HANDSHAKE_MESSAGE_1_SIZE
) {
val shouldYield = localPeerID > peerID
if (shouldYield) {
Log.d(TAG, "Handshake collision with $peerID; yielding to responder role")
removeSession(peerID)
session = null
} else {
Log.d(TAG, "Handshake collision with $peerID; keeping initiator role")
return null
}
}
// If no session exists, create one as responder
if (session == null) {
@@ -119,6 +158,12 @@ class NoiseSessionManager(
throw e
}
}
private fun isHandshakeStale(session: NoiseSession, nowMs: Long): Boolean {
val lastActivity = session.getLastHandshakeActivityMs() ?: session.getHandshakeStartMs()
if (lastActivity == null) return false
return (nowMs - lastActivity) > HANDSHAKE_TIMEOUT_MS
}
/**
* SIMPLIFIED: Encrypt data
@@ -2,6 +2,7 @@ package com.bitchat.android.service
import android.util.Log
import com.bitchat.android.model.RoutedPacket
import com.bitchat.android.protocol.BitchatPacket
import java.util.concurrent.ConcurrentHashMap
/**
@@ -23,6 +24,11 @@ object TransportBridgeService {
* Send a packet out via this transport.
*/
fun send(packet: RoutedPacket)
/**
* Send a packet to a specific peer via this transport (optional).
*/
fun sendToPeer(peerID: String, packet: BitchatPacket) { }
}
private val transports = ConcurrentHashMap<String, TransportLayer>()
@@ -66,4 +72,20 @@ object TransportBridgeService {
}
}
}
/**
* Send a packet to a specific peer across all other transports.
*/
fun sendToPeer(sourceId: String, peerID: String, packet: BitchatPacket) {
val targets = transports.filterKeys { it != sourceId }
if (targets.isEmpty()) return
targets.forEach { (id, layer) ->
try {
layer.sendToPeer(peerID, packet)
} catch (e: Exception) {
Log.e(TAG, "Failed to bridge unicast packet to $id: ${e.message}")
}
}
}
}
@@ -308,15 +308,34 @@ class MeshDelegateHandler(
if (shouldSendReadReceipt) {
android.util.Log.d("MeshDelegateHandler", "Sending reactive read receipt for focused chat with $senderPeerID (message=${message.id})")
val nickname = state.getNicknameValue() ?: "unknown"
// Send directly for this message to avoid relying on unread queues
getMeshService().sendReadReceipt(message.id, senderPeerID!!, nickname)
// Ensure unread badge is cleared for this peer immediately
try {
val current = state.getUnreadPrivateMessagesValue().toMutableSet()
if (current.remove(senderPeerID)) {
state.setUnreadPrivateMessages(current)
val mesh = getMeshService()
val sent = try {
val hasMesh = mesh.getPeerInfo(senderPeerID!!)?.isConnected == true && mesh.hasEstablishedSession(senderPeerID)
if (hasMesh) {
mesh.sendReadReceipt(message.id, senderPeerID, nickname)
true
} else {
val aware = try { com.bitchat.android.wifiaware.WifiAwareController.getService() } catch (_: Exception) { null }
val hasAware = try { aware?.getPeerInfo(senderPeerID)?.isConnected == true && aware.hasEstablishedSession(senderPeerID) } catch (_: Exception) { false }
if (hasAware) {
aware?.sendReadReceipt(message.id, senderPeerID, nickname)
true
} else {
false
}
}
} catch (_: Exception) { }
} catch (_: Exception) {
false
}
if (sent) {
// Ensure unread badge is cleared for this peer immediately
try {
val current = state.getUnreadPrivateMessagesValue().toMutableSet()
if (current.remove(senderPeerID)) {
state.setUnreadPrivateMessages(current)
}
} catch (_: Exception) { }
}
} else {
android.util.Log.d("MeshDelegateHandler", "Skipping read receipt - chat not focused (background: $isAppInBackground, current peer: $currentPrivateChatPeer, sender: $senderPeerID)")
}
@@ -333,13 +333,21 @@ class PrivateChatManager(
}
val myNickname = state.getNicknameValue() ?: "unknown"
val aware = try { com.bitchat.android.wifiaware.WifiAwareController.getService() } catch (_: Exception) { null }
val hasMesh = try { meshService.getPeerInfo(peerID)?.isConnected == true && meshService.hasEstablishedSession(peerID) } catch (_: Exception) { false }
val hasAware = try { aware?.getPeerInfo(peerID)?.isConnected == true && aware.hasEstablishedSession(peerID) } catch (_: Exception) { false }
var sentCount = 0
messages.forEach { msg ->
// Only for incoming messages from this peer
if (msg.senderPeerID == peerID) {
try {
meshService.sendReadReceipt(msg.id, peerID, myNickname)
sentCount += 1
if (hasMesh) {
meshService.sendReadReceipt(msg.id, peerID, myNickname)
sentCount += 1
} else if (hasAware) {
aware?.sendReadReceipt(msg.id, peerID, myNickname)
sentCount += 1
}
} catch (e: Exception) {
Log.w(TAG, "Failed to send read receipt for message ${msg.id}: ${e.message}")
}
@@ -144,6 +144,10 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor
meshCore.sendFromBridge(packet)
}
override fun sendToPeer(peerID: String, packet: BitchatPacket) {
sendPacketToPeer(peerID, packet)
}
/**
* Broadcasts routed packet to currently connected peers.
*/