feat(database): unify device database across transports (#6096)

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

core/common/src/commonMain/kotlin/org/meshtastic/core/common/database/DatabaseManager.kt

@@ -32,6 +32,14 @@ interface DatabaseManager {
     /** Switches the active database to the one associated with the given [address]. */
     suspend fun switchActiveDatabase(address: String?)
 
+    /**
+     * Associates the currently-active connection with the physical node [nodeNum] learned from the radio handshake, so
+     * the same node reached over a different transport (BLE / TCP / USB) shares one database. The first transport to
+     * report a node claims the active DB for it; a later transport folds its own data in and switches to the claimed
+     * DB. Idempotent once a node/transport pair has been unified.
+     */
+    suspend fun associateNode(nodeNum: Int)
+
     /** Returns true if a database exists for the given device address. */
     fun hasDatabaseFor(address: String?): Boolean
 }

core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseConstants.kt

@@ -32,6 +32,11 @@ object DatabaseConstants {
 
     const val LEGACY_DB_CLEANED_KEY: String = "legacy_db_cleaned"
 
+    // Cross-transport unification: map a node / transport address to the DB file that node's data lives in.
+    // Keys are dynamic (suffixed with the node num / normalized address), mirroring the `db_last_used:` pattern.
+    const val NODE_DB_FOR_PREFIX: String = "node_db_for:"
+    const val ADDR_DB_FOR_PREFIX: String = "addr_db_for:"
+
     // Display/truncation and hash sizing for DB names
     const val DB_NAME_HASH_LEN: Int = 10
     const val DB_NAME_SEPARATOR_LEN: Int = 1

core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt

@@ -22,6 +22,7 @@ import androidx.datastore.preferences.core.booleanPreferencesKey
 import androidx.datastore.preferences.core.edit
 import androidx.datastore.preferences.core.intPreferencesKey
 import androidx.datastore.preferences.core.longPreferencesKey
+import androidx.datastore.preferences.core.stringPreferencesKey
 import co.touchlab.kermit.Logger
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
@@ -44,6 +45,7 @@ import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
 import org.koin.core.annotation.Named
 import org.koin.core.annotation.Single
+import org.meshtastic.core.common.util.normalizeAddress
 import org.meshtastic.core.common.util.nowMillis
 import org.meshtastic.core.di.CoroutineDispatchers
 import kotlin.concurrent.Volatile
@@ -67,6 +69,11 @@ open class DatabaseManager(
 
     private fun lastUsedKey(dbName: String) = longPreferencesKey("db_last_used:$dbName")
 
+    private fun addrDbKey(address: String?) =
+        stringPreferencesKey("${DatabaseConstants.ADDR_DB_FOR_PREFIX}${normalizeAddress(address)}")
+
+    private fun nodeDbKey(nodeNum: Int) = stringPreferencesKey("${DatabaseConstants.NODE_DB_FOR_PREFIX}$nodeNum")
+
     private var backfillJob: Job? = null
 
     @Volatile private var hasDelayedFirstDeviceBackfill = false
@@ -83,9 +90,7 @@ open class DatabaseManager(
         managerScope.launch {
             datastore.edit { it[cacheLimitKey] = clamped }
             // Enforce asynchronously with current active DB protected
-            val active =
-                _currentDb.value?.let { buildDbName(_currentAddress.value) } ?: DatabaseConstants.DEFAULT_DB_NAME
-            enforceCacheLimit(activeDbName = active)
+            enforceCacheLimit(activeDbName = currentDbName)
         }
     }
 
@@ -105,6 +110,13 @@ open class DatabaseManager(
     private val _currentAddress = MutableStateFlow<String?>(null)
     val currentAddress: StateFlow<String?> = _currentAddress
 
+    /**
+     * Name of the currently active database. Tracked explicitly rather than recomputed from the address, because
+     * cross-transport aliasing ([associateNode]) decouples the two: a secondary transport's address maps to the DB
+     * claimed by the first transport, which `buildDbName(address)` would never produce. Written under [mutex].
+     */
+    @Volatile private var currentDbName: String = DatabaseConstants.DEFAULT_DB_NAME
+
     /** Initialize the active database for [address]. */
     suspend fun init(address: String?) {
         switchActiveDatabase(address)
@@ -118,12 +130,24 @@ open class DatabaseManager(
     private fun getOrOpenDatabase(dbName: String): MeshtasticDatabase =
         dbCache.getOrPut(dbName) { getDatabaseBuilder(dbName).build() }
 
+    /**
+     * Resolves the DB name to use for [address], honoring a cross-transport alias when one exists. A secondary
+     * transport (e.g. TCP) that has been unified with a node points at the DB the first transport (e.g. BLE) claimed;
+     * without an alias this falls back to the address-hashed name — today's default — for a first-time or primary
+     * connection. See [associateNode].
+     */
+    private suspend fun resolveDbName(address: String?): String {
+        val fallback = buildDbName(address)
+        if (fallback == DatabaseConstants.DEFAULT_DB_NAME) return fallback
+        return datastore.data.first()[addrDbKey(address)] ?: fallback
+    }
+
     /** Switch active database to the one associated with [address]. Serialized via mutex. */
     override suspend fun switchActiveDatabase(address: String?) = mutex.withLock {
-        val dbName = buildDbName(address)
+        val dbName = resolveDbName(address)
 
         // Remember the previously active DB name (any) so we can record its last-used time as well.
-        val previousDbName = _currentDb.value?.let { buildDbName(_currentAddress.value) }
+        val previousDbName = if (_currentDb.value != null) currentDbName else null
 
         // Fast path: no-op if already on this address
         if (_currentAddress.value == address && _currentDb.value != null) {
@@ -140,6 +164,7 @@ open class DatabaseManager(
         // collectors, causing "Connection pool is closed" crashes.
         _currentDb.value = db
         _currentAddress.value = address
+        currentDbName = dbName
         markLastUsed(dbName)
         // Also mark the previous DB as used "just now" so LRU has an accurate, recent timestamp
         previousDbName?.let { markLastUsed(it) }
@@ -167,6 +192,78 @@ open class DatabaseManager(
         Logger.i { "Switched active DB to ${anonymizeDbName(dbName)} for address ${anonymizeAddress(address)}" }
     }
 
+    @Suppress("TooGenericExceptionCaught")
+    override suspend fun associateNode(nodeNum: Int) {
+        mutex.withLock {
+            val sourceName = currentDbName
+            // Never claim or merge into the sentinel "no device" DB.
+            if (sourceName == DatabaseConstants.DEFAULT_DB_NAME) return@withLock
+
+            val claimed = datastore.data.first()[nodeDbKey(nodeNum)]
+            when {
+                claimed == null -> {
+                    // First transport to learn this node: its current DB becomes the node's canonical DB.
+                    // No address alias is needed — a primary connection already resolves to this DB via buildDbName.
+                    datastore.edit { it[nodeDbKey(nodeNum)] = sourceName }
+                    Logger.i { "Claimed ${anonymizeDbName(sourceName)} as canonical DB for node $nodeNum" }
+                }
+
+                claimed == sourceName -> Unit
+
+                // Already unified — nothing to do.
+
+                else -> {
+                    // Secondary transport reached an already-known node: fold this DB into the canonical one,
+                    // switch the active DB to it, alias this address to it, and retire the now-merged source.
+                    val source = _currentDb.value ?: return@withLock
+                    val dest = withContext(dispatchers.io) { getOrOpenDatabase(claimed) }
+
+                    // Redirect live writers to the canonical DB BEFORE merging. Otherwise a concurrent withDb
+                    // write (connect triggers a full NodeDB re-dump) could land in `source` after the merge has
+                    // already snapshotted it, then be lost when `source` is retired — and withDb's closed-pool
+                    // retry can't recover it because the write itself succeeded. New writers now capture `dest`;
+                    // any still holding `source` are covered by that retry once retireDatabase closes it.
+                    _currentDb.value = dest
+                    currentDbName = claimed
+                    try {
+                        withContext(dispatchers.io) { DatabaseMerger.merge(source, dest) }
+                    } catch (e: CancellationException) {
+                        throw e
+                    } catch (e: Exception) {
+                        // merge() is atomic, so on failure `dest` is unchanged. Roll the active DB back to
+                        // `source` so the address still resolves consistently and the merge retries next connect.
+                        _currentDb.value = source
+                        currentDbName = sourceName
+                        Logger.w(e) {
+                            "Merge into ${anonymizeDbName(claimed)} failed; kept ${anonymizeDbName(sourceName)} active"
+                        }
+                        return@withLock
+                    }
+
+                    markLastUsed(claimed)
+                    datastore.edit { it[addrDbKey(_currentAddress.value)] = claimed }
+                    Logger.i {
+                        "Unified ${anonymizeDbName(sourceName)} into ${anonymizeDbName(claimed)} for node $nodeNum"
+                    }
+
+                    // Retire the merged source off the critical path; its data now lives in the canonical DB.
+                    managerScope.launch(dispatchers.io) { retireDatabase(sourceName) }
+                }
+            }
+        }
+    }
+
+    /** Closes, deletes, and forgets a database whose contents have been merged into another. */
+    private suspend fun retireDatabase(dbName: String) = mutex.withLock {
+        runCatching {
+            closeCachedDatabase(dbName)
+            deleteDatabase(dbName)
+            datastore.edit { it.remove(lastUsedKey(dbName)) }
+        }
+            .onSuccess { Logger.i { "Retired merged DB ${anonymizeDbName(dbName)}" } }
+            .onFailure { Logger.w(it) { "Failed to retire merged database ${anonymizeDbName(dbName)}" } }
+    }
+
     /**
      * Closes and removes a cached database by name. Safe to call even if the database was already closed or not in the
      * cache. Does NOT delete the underlying file — the database can be re-opened on next access.
@@ -188,7 +285,7 @@ open class DatabaseManager(
     @Suppress("TooGenericExceptionCaught")
     override suspend fun <T> withDb(block: suspend (MeshtasticDatabase) -> T): T? = withContext(limitedIo) {
         val db = _currentDb.value ?: return@withContext null
-        val active = buildDbName(_currentAddress.value)
+        val active = currentDbName
         markLastUsed(active)
         try {
             block(db)

core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseMerger.kt

@@ -0,0 +1,148 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.database
+
+import androidx.room3.immediateTransaction
+import androidx.room3.useWriterConnection
+import co.touchlab.kermit.Logger
+
+/**
+ * Folds the contents of one device database ([source]) into another ([dest]) when both turn out to belong to the same
+ * physical node reached over different transports (BLE / TCP / USB). Called once, by [DatabaseManager.associateNode],
+ * the first time a secondary transport learns a `myNodeNum` that another transport already claimed.
+ *
+ * Every per-device table is unified so switching transport is seamless: messages (+FTS), reactions, contact mute/read
+ * settings, nodes and their notes, per-node metadata, the audit log (each session's received-packet history — the
+ * source of the telemetry timelines, position history, and traceroute results the UI reconstructs), traceroute
+ * positions, and discovery sessions. Where the same row exists on both sides the destination is preferred (it will be
+ * refreshed by the same radio's re-dump on connect); source-only rows and strictly newer history are brought over.
+ * Packets/reactions in [source] and [dest] already share the same `myNodeNum` (same node), so no key remapping is
+ * needed there; autoincrement-keyed rows (packets, discovery) are re-inserted with fresh ids to avoid collisions.
+ */
+object DatabaseMerger {
+
+    suspend fun merge(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+        // All destination writes run in a single transaction so a crash or exception mid-merge rolls
+        // back cleanly instead of leaving `dest` half-merged. This also makes a retried merge safe: if
+        // associateNode re-runs (e.g. a crash before the address alias is persisted), the rolled-back
+        // partial writes never happened, so the fresh-id packet/discovery re-inserts can't duplicate.
+        // Reads from `source` use its own connection pool and don't participate in this transaction.
+        var packets = 0
+        dest.useWriterConnection { transactor ->
+            transactor.immediateTransaction {
+                packets = mergePackets(source, dest)
+                mergeReactions(source, dest)
+                mergeContactSettings(source, dest)
+                mergeNodes(source, dest)
+                mergeMetadata(source, dest)
+                // Logs must precede traceroute positions: the latter has a CASCADE foreign key onto log.uuid.
+                mergeLogs(source, dest)
+                mergeTraceroutePositions(source, dest)
+                mergeDiscovery(source, dest)
+            }
+        }
+        Logger.i { "Merged $packets packets across transports into unified node DB" }
+    }
+
+    /**
+     * Message history is the primary payload. `uuid` is per-DB autoincrement, so re-insert with uuid = 0 to mint a
+     * fresh id and avoid collisions; rebuild the FTS index once at the end so copied messages are searchable.
+     */
+    private suspend fun mergePackets(source: MeshtasticDatabase, dest: MeshtasticDatabase): Int {
+        val packets = source.packetDao().getAllPacketsSnapshot()
+        packets.forEach { dest.packetDao().insertPacketForMerge(it.copy(uuid = 0L)) }
+        if (packets.isNotEmpty()) dest.packetDao().rebuildFtsIndex()
+        return packets.size
+    }
+
+    /** Composite PK (myNodeNum, reply_id, user_id, emoji) makes IGNORE a natural dedupe. */
+    private suspend fun mergeReactions(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+        dest.packetDao().insertReactionsIgnore(source.packetDao().getAllReactionsSnapshot())
+    }
+
+    /** Keep the destination's settings on conflict (IGNORE); only bring over contacts it has never seen. */
+    private suspend fun mergeContactSettings(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+        dest.packetDao().insertContactSettingsIgnore(source.packetDao().getAllContactSettingsSnapshot())
+    }
+
+    /**
+     * Bring over every node the destination has never seen (so its last-known identity/telemetry/position shows
+     * immediately, including offline nodes the radio may not re-report). For overlapping nodes, keep the destination's
+     * row — the same radio re-dumps it on connect — but fill in a user note where the destination has none, never
+     * clobbering one the user already wrote on the canonical DB.
+     */
+    private suspend fun mergeNodes(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+        val destNums = dest.nodeInfoDao().getAllNodesSnapshot().mapTo(mutableSetOf()) { it.num }
+        source.nodeInfoDao().getAllNodesSnapshot().forEach { node ->
+            when {
+                node.num !in destNums -> dest.nodeInfoDao().upsert(node)
+
+                node.notes.isNotBlank() -> {
+                    val existing = dest.nodeInfoDao().getNodeByNum(node.num)
+                    if (existing != null && existing.node.notes.isBlank()) {
+                        dest.nodeInfoDao().setNodeNotes(node.num, node.notes)
+                    }
+                }
+            }
+        }
+    }
+
+    /** Per-node DeviceMetadata: bring source rows the dest lacks or that are strictly newer (by timestamp). */
+    private suspend fun mergeMetadata(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+        val destByNum = dest.nodeInfoDao().getAllMetadataSnapshot().associateBy { it.num }
+        source.nodeInfoDao().getAllMetadataSnapshot().forEach { meta ->
+            val existing = destByNum[meta.num]
+            if (existing == null || meta.timestamp > existing.timestamp) dest.nodeInfoDao().upsert(meta)
+        }
+    }
+
+    /**
+     * The audit log holds each transport session's received-packet history — the source of the telemetry timelines,
+     * position history, and node-info history the UI reconstructs per node/port. This is the only place that time
+     * series diverges between transports (the current node snapshot is re-dumped identically by the same radio), so
+     * merging it is what actually unifies "metrics/telemetry/positions". uuid is a unique string, so IGNORE is safe.
+     */
+    private suspend fun mergeLogs(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+        dest.meshLogDao().insertIgnore(source.meshLogDao().getAllLogsSnapshot())
+    }
+
+    /**
+     * Traceroute-derived positions, keyed to their log entry (CASCADE FK) — merge after [mergeLogs] so the parent log
+     * rows already exist in [dest].
+     */
+    private suspend fun mergeTraceroutePositions(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+        dest.tracerouteNodePositionDao().insertAll(source.tracerouteNodePositionDao().getAllSnapshot())
+    }
+
+    /**
+     * Discovery sessions are distinct events per transport session, so append them all (no dedupe). The three tables
+     * form an autoincrement-keyed FK chain (session → preset result → discovered node), so re-insert each level with a
+     * fresh id and rewire the child's foreign key to the parent's newly-assigned id.
+     */
+    private suspend fun mergeDiscovery(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+        val src = source.discoveryDao()
+        val dst = dest.discoveryDao()
+        src.getAllSessionsSnapshot().forEach { session ->
+            val newSessionId = dst.insertSession(session.copy(id = 0))
+            src.getPresetResults(session.id).forEach { preset ->
+                val newPresetId = dst.insertPresetResult(preset.copy(id = 0, sessionId = newSessionId))
+                val nodes = src.getDiscoveredNodes(preset.id).map { it.copy(id = 0, presetResultId = newPresetId) }
+                if (nodes.isNotEmpty()) dst.insertDiscoveredNodes(nodes)
+            }
+        }
+    }
+}

core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/DiscoveryDao.kt

@@ -39,6 +39,10 @@ interface DiscoveryDao {
     @Query("SELECT * FROM discovery_session ORDER BY timestamp DESC")
     fun getAllSessions(): Flow<List<DiscoverySessionEntity>>
 
+    /** Snapshot used by DatabaseMerger to append a transport's discovery sessions to the unified DB. */
+    @Query("SELECT * FROM discovery_session")
+    suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity>
+
     @Query("SELECT * FROM discovery_session WHERE id = :sessionId")
     suspend fun getSession(sessionId: Long): DiscoverySessionEntity?
 

core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/MeshLogDao.kt

@@ -18,6 +18,7 @@ package org.meshtastic.core.database.dao
 
 import androidx.room3.Dao
 import androidx.room3.Insert
+import androidx.room3.OnConflictStrategy
 import androidx.room3.Query
 import kotlinx.coroutines.flow.Flow
 import org.meshtastic.core.database.entity.MeshLog
@@ -47,6 +48,16 @@ interface MeshLogDao {
 
     @Insert suspend fun insert(log: MeshLog)
 
+    /**
+     * Snapshot + IGNORE-insert used by DatabaseMerger to carry telemetry/position/traceroute history across transports.
+     * uuid is a unique string per row, so IGNORE is just belt-and-suspenders against a rare clash.
+     */
+    @Query("SELECT * FROM log")
+    suspend fun getAllLogsSnapshot(): List<MeshLog>
+
+    @Insert(onConflict = OnConflictStrategy.IGNORE)
+    suspend fun insertIgnore(logs: List<MeshLog>)
+
     @Query("DELETE FROM log")
     suspend fun deleteAll()
 

core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/NodeInfoDao.kt

@@ -283,6 +283,10 @@ interface NodeInfoDao {
     @Query("DELETE FROM metadata WHERE num=:num")
     suspend fun deleteMetadata(num: Int)
 
+    /** Snapshot used by DatabaseMerger to carry per-node DeviceMetadata across transports (newest timestamp wins). */
+    @Query("SELECT * FROM metadata")
+    suspend fun getAllMetadataSnapshot(): List<MetadataEntity>
+
     @Query("SELECT * FROM nodes WHERE num=:num")
     @Transaction
     suspend fun getNodeByNum(num: Int): NodeWithRelations?

core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt

@@ -489,6 +489,26 @@ interface PacketDao {
     @Query("DELETE FROM contact_settings")
     suspend fun deleteAllContactSettings()
 
+    // region ── Cross-transport merge ──
+    // Snapshots + inserts used by DatabaseMerger to fold one transport's DB into another for the same node.
+
+    @Query("SELECT * FROM packet")
+    suspend fun getAllPacketsSnapshot(): List<Packet>
+
+    /** Insert a packet copied from another DB. Pass uuid = 0 so a fresh auto-generated id is assigned. */
+    @Insert suspend fun insertPacketForMerge(packet: Packet)
+
+    @Query("SELECT * FROM reactions")
+    suspend fun getAllReactionsSnapshot(): List<ReactionEntity>
+
+    @Insert(onConflict = OnConflictStrategy.IGNORE)
+    suspend fun insertReactionsIgnore(reactions: List<ReactionEntity>)
+
+    @Query("SELECT * FROM contact_settings")
+    suspend fun getAllContactSettingsSnapshot(): List<ContactSettings>
+
+    // endregion
+
     /**
      * One-time migration: Remap all message DataPacket.channel indices to new mapping using PSK after a channel
      * reorder. For each Packet (with port_num = 1), finds the old PSK then sets the channel index to the matching

core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/TracerouteNodePositionDao.kt

@@ -31,5 +31,8 @@ interface TracerouteNodePositionDao {
     @Query("DELETE FROM traceroute_node_position WHERE log_uuid = :logUuid")
     suspend fun deleteByLogUuid(logUuid: String)
 
+    @Query("SELECT * FROM traceroute_node_position")
+    suspend fun getAllSnapshot(): List<TracerouteNodePositionEntity>
+
     @Upsert suspend fun insertAll(entities: List<TracerouteNodePositionEntity>)
 }

core/database/src/commonTest/kotlin/org/meshtastic/core/database/DatabaseMergerTest.kt

@@ -0,0 +1,223 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.database
+
+import kotlinx.coroutines.test.runTest
+import okio.ByteString.Companion.toByteString
+import org.meshtastic.core.database.entity.ContactSettings
+import org.meshtastic.core.database.entity.DiscoveredNodeEntity
+import org.meshtastic.core.database.entity.DiscoveryPresetResultEntity
+import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.MeshLog
+import org.meshtastic.core.database.entity.MetadataEntity
+import org.meshtastic.core.database.entity.MyNodeEntity
+import org.meshtastic.core.database.entity.NodeEntity
+import org.meshtastic.core.database.entity.Packet
+import org.meshtastic.core.database.entity.ReactionEntity
+import org.meshtastic.core.database.entity.TracerouteNodePositionEntity
+import org.meshtastic.core.model.DataPacket
+import org.meshtastic.core.model.NodeAddress
+import org.meshtastic.proto.DeviceMetadata
+import org.meshtastic.proto.HardwareModel
+import org.meshtastic.proto.PortNum
+import org.meshtastic.proto.Position
+import org.meshtastic.proto.User
+import kotlin.test.AfterTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
+
+/**
+ * Verifies [DatabaseMerger] folds a secondary transport's DB into the canonical one for the same node: message history
+ * combines with fresh non-colliding ids and stays searchable, node notes are preserved without clobbering, and
+ * reactions dedupe on their composite key. This is the data-integrity path for cross-transport DB unification.
+ */
+class DatabaseMergerTest {
+
+    // Same physical node reached over two transports → same myNodeNum in both DBs.
+    private val nodeNum = 0x1234
+
+    private val source: MeshtasticDatabase = getInMemoryDatabaseBuilder().build()
+    private val dest: MeshtasticDatabase = getInMemoryDatabaseBuilder().build()
+
+    @AfterTest
+    fun tearDown() {
+        source.close()
+        dest.close()
+    }
+
+    private fun myNode() = MyNodeEntity(
+        myNodeNum = nodeNum,
+        model = "TBEAM",
+        firmwareVersion = "2.5.0",
+        couldUpdate = false,
+        shouldUpdate = false,
+        currentPacketId = 1L,
+        messageTimeoutMsec = 300_000,
+        minAppVersion = 1,
+        maxChannels = 8,
+        hasWifi = false,
+    )
+
+    private fun textPacket(contact: String, text: String, time: Long) = Packet(
+        uuid = 0L, // auto-generated
+        myNodeNum = nodeNum,
+        port_num = PortNum.TEXT_MESSAGE_APP.value,
+        contact_key = contact,
+        received_time = time,
+        read = true,
+        data =
+        DataPacket(
+            to = NodeAddress.ID_BROADCAST,
+            bytes = text.encodeToByteArray().toByteString(),
+            dataType = PortNum.TEXT_MESSAGE_APP.value,
+        ),
+        messageText = text, // FTS indexes this column
+    )
+
+    private fun node(num: Int, notes: String = "") = NodeEntity(
+        num = num,
+        user = User(id = "!$num", long_name = "Node $num", hw_model = HardwareModel.TBEAM),
+        notes = notes,
+    )
+
+    private fun logEntry(uuid: String, portNum: Int, time: Long) = MeshLog(
+        uuid = uuid,
+        message_type = "Packet",
+        received_date = time,
+        raw_message = "",
+        fromNum = 30,
+        portNum = portNum,
+    )
+
+    @Test
+    fun mergeFoldsSecondaryTransportIntoCanonical() = runTest {
+        // Canonical (dest): one message, a noted node (#10), an un-noted node (#20), a contact setting, a reaction.
+        dest.nodeInfoDao().setMyNodeInfo(myNode())
+        dest.packetDao().insert(textPacket("0!broadcast", "destmsg", time = 100))
+        dest.nodeInfoDao().upsert(node(10, notes = "keep me"))
+        dest.nodeInfoDao().upsert(node(20)) // no notes yet
+        dest.packetDao().upsertContactSettings(listOf(ContactSettings(contact_key = "0!broadcast", muteUntil = 999)))
+        dest
+            .packetDao()
+            .insert(ReactionEntity(myNodeNum = nodeNum, replyId = 1, userId = "!u", emoji = "👍", timestamp = 1))
+        dest.meshLogDao().insert(logEntry("destlog", PortNum.TELEMETRY_APP.value, time = 50))
+        dest
+            .nodeInfoDao()
+            .upsert(MetadataEntity(num = 10, proto = DeviceMetadata(), timestamp = 1000)) // newer than src
+
+        // Secondary (source): two messages, notes for #20 (dest blank → fill) and a source-only node (#30 → insert),
+        // the same reaction (dedupe) plus a new one, and a conflicting contact setting (dest's must win).
+        source.nodeInfoDao().setMyNodeInfo(myNode())
+        source.packetDao().insert(textPacket("0!broadcast", "srcmsgone", time = 200))
+        source.packetDao().insert(textPacket("0!broadcast", "srcmsgtwo", time = 300))
+        source.nodeInfoDao().upsert(node(20, notes = "src note"))
+        source.nodeInfoDao().upsert(node(30, notes = "src-only note"))
+        source.packetDao().upsertContactSettings(listOf(ContactSettings(contact_key = "0!broadcast", muteUntil = 1)))
+        source
+            .packetDao()
+            .insert(ReactionEntity(myNodeNum = nodeNum, replyId = 1, userId = "!u", emoji = "👍", timestamp = 1))
+        source
+            .packetDao()
+            .insert(ReactionEntity(myNodeNum = nodeNum, replyId = 1, userId = "!v", emoji = "❤️", timestamp = 2))
+        // Telemetry + position history (audit log) plus a traceroute position referencing a log entry.
+        source.meshLogDao().insert(logEntry("srctelemetry", PortNum.TELEMETRY_APP.value, time = 200))
+        source.meshLogDao().insert(logEntry("srcposition", PortNum.POSITION_APP.value, time = 210))
+        source
+            .tracerouteNodePositionDao()
+            .insertAll(
+                listOf(
+                    TracerouteNodePositionEntity(
+                        logUuid = "srcposition",
+                        requestId = 7,
+                        nodeNum = 30,
+                        position = Position(),
+                    ),
+                ),
+            )
+        source.nodeInfoDao().upsert(node(40)) // source-only node, no notes → still brought over
+        source
+            .nodeInfoDao()
+            .upsert(MetadataEntity(num = 10, proto = DeviceMetadata(), timestamp = 500)) // older → dest wins
+        source
+            .nodeInfoDao()
+            .upsert(MetadataEntity(num = 30, proto = DeviceMetadata(), timestamp = 500)) // dest lacks → added
+        val srcSession =
+            source
+                .discoveryDao()
+                .insertSession(
+                    DiscoverySessionEntity(timestamp = 1, presetsScanned = "LongFast", homePreset = "LongFast"),
+                )
+        val srcPreset =
+            source
+                .discoveryDao()
+                .insertPresetResult(DiscoveryPresetResultEntity(sessionId = srcSession, presetName = "LongFast"))
+        source
+            .discoveryDao()
+            .insertDiscoveredNodes(listOf(DiscoveredNodeEntity(presetResultId = srcPreset, nodeNum = 30L)))
+
+        DatabaseMerger.merge(source, dest)
+
+        // Packets: 1 + 2, with fresh distinct non-zero uuids (no collision with dest's existing uuid).
+        val packets = dest.packetDao().getAllPacketsSnapshot()
+        assertEquals(3, packets.size, "all messages combined")
+        assertEquals(3, packets.map { it.uuid }.toSet().size, "uuids are unique")
+        assertTrue(packets.none { it.uuid == 0L }, "every merged packet got a real id")
+
+        // Merged messages remain searchable via the rebuilt FTS index.
+        assertEquals(1, dest.packetDao().searchMessages("srcmsgone").size, "copied message is searchable")
+
+        // Node notes: dest's existing note kept, blank note filled from source, source-only node brought over.
+        assertEquals("keep me", dest.nodeInfoDao().getNodeByNum(10)?.node?.notes)
+        assertEquals("src note", dest.nodeInfoDao().getNodeByNum(20)?.node?.notes)
+        assertNotNull(dest.nodeInfoDao().getNodeByNum(30), "source-only noted node inserted")
+
+        // Reactions dedupe on composite PK: R1 (already in dest) + R2 (new) = 2.
+        assertEquals(2, dest.packetDao().getAllReactionsSnapshot().size)
+
+        // Contact settings: destination's value wins on conflict (IGNORE), never overwritten.
+        assertEquals(
+            999,
+            dest.packetDao().getAllContactSettingsSnapshot().first { it.contact_key == "0!broadcast" }.muteUntil,
+        )
+
+        // Telemetry/position history lives in the audit log: both transports' entries combine, and traceroute
+        // positions come along with their (now-present) parent log rows.
+        assertEquals(3, dest.meshLogDao().getAllLogsSnapshot().size, "log history from both transports combined")
+        assertEquals(1, dest.tracerouteNodePositionDao().getAllSnapshot().size, "traceroute positions merged")
+
+        // Source-only nodes are brought over even without notes.
+        assertNotNull(dest.nodeInfoDao().getNodeByNum(40), "source-only node brought over")
+
+        // Metadata: newest timestamp wins on overlap, source-only rows added.
+        val metadata = dest.nodeInfoDao().getAllMetadataSnapshot().associateBy { it.num }
+        assertEquals(1000, metadata[10]?.timestamp, "dest's newer metadata kept")
+        assertNotNull(metadata[30], "source-only metadata inserted")
+
+        // Discovery: the session → preset → node chain is appended with rewired foreign keys.
+        val sessions = dest.discoveryDao().getAllSessionsSnapshot()
+        assertEquals(1, sessions.size, "discovery session appended")
+        val presets = dest.discoveryDao().getPresetResults(sessions.first().id)
+        assertEquals(1, presets.size, "preset appended under rewired session id")
+        assertEquals(
+            1,
+            dest.discoveryDao().getDiscoveredNodes(presets.first().id).size,
+            "node appended under rewired preset id",
+        )
+    }
+}

core/service/src/commonMain/kotlin/org/meshtastic/core/service/RadioControllerImpl.kt

@@ -19,6 +19,9 @@ package org.meshtastic.core.service
 import co.touchlab.kermit.Logger
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.launch
 import org.meshtastic.core.common.database.DatabaseManager
 import org.meshtastic.core.model.ConnectionState
 import org.meshtastic.core.repository.AdminController
@@ -87,6 +90,18 @@ class RadioControllerImpl(
     NodeController by NodeControllerImpl(commandSender, nodeManager, packetRepository, scope),
     QueryController by QueryControllerImpl(commandSender, nodeManager, uiPrefs) {
 
+    init {
+        // Unify per-node databases across transports. When the handshake reports our node number, tell the
+        // DatabaseManager to claim (or merge into) that node's canonical DB, so the same node reached over BLE, TCP,
+        // or USB shares one stored history. Keyed on (address, node) so a second transport for the same node re-fires
+        // even though the node number itself is unchanged.
+        scope.launch {
+            combine(meshPrefs.deviceAddress, nodeManager.myNodeNum) { address, nodeNum -> address to nodeNum }
+                .distinctUntilChanged()
+                .collect { (_, nodeNum) -> nodeNum?.let { databaseManager.associateNode(it) } }
+        }
+    }
+
     // ── Connection State ────────────────────────────────────────────────────
 
     override val connectionState: StateFlow<ConnectionState>

core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeDatabaseManager.kt

@@ -29,12 +29,14 @@ class FakeDatabaseManager :
     override val cacheLimit: StateFlow<Int> = _cacheLimit
 
     var lastSwitchedAddress: String? = null
+    var lastAssociatedNode: Int? = null
     val existingDatabases = mutableSetOf<String>()
 
     init {
         registerResetAction {
             _cacheLimit.value = DEFAULT_CACHE_LIMIT
             lastSwitchedAddress = null
+            lastAssociatedNode = null
             existingDatabases.clear()
         }
     }
@@ -49,6 +51,10 @@ class FakeDatabaseManager :
         lastSwitchedAddress = address
     }
 
+    override suspend fun associateNode(nodeNum: Int) {
+        lastAssociatedNode = nodeNum
+    }
+
     override fun hasDatabaseFor(address: String?): Boolean = address != null && existingDatabases.contains(address)
 
     companion object {

feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHistoryBehaviorTest.kt

@@ -185,6 +185,8 @@ private class HistoryTestDao : DiscoveryDao {
 
     override fun getAllSessions(): Flow<List<DiscoverySessionEntity>> = sessionsFlow
 
+    override suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity> = sessions.values.toList()
+
     override suspend fun getSession(sessionId: Long) = sessions[sessionId]
 
     override fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?> = MutableStateFlow(sessions[sessionId])

feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryMapFilterTest.kt

@@ -171,6 +171,8 @@ private class MapTestDao : DiscoveryDao {
     override fun getAllSessions(): Flow<List<DiscoverySessionEntity>> =
         flowOf(sessions.values.sortedByDescending { it.timestamp })
 
+    override suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity> = sessions.values.toList()
+
     override suspend fun getSession(sessionId: Long) = sessions[sessionId]
 
     override fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?> = MutableStateFlow(sessions[sessionId])

feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryPacketCollectionTest.kt

@@ -352,6 +352,8 @@ private class InMemoryDiscoveryDao : DiscoveryDao {
     override fun getAllSessions(): Flow<List<DiscoverySessionEntity>> =
         flowOf(sessions.values.sortedByDescending { it.timestamp })
 
+    override suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity> = sessions.values.toList()
+
     override suspend fun getSession(sessionId: Long): DiscoverySessionEntity? = sessions[sessionId]
 
     override fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?> = MutableStateFlow(sessions[sessionId])

feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngineTest.kt

@@ -89,6 +89,8 @@ private class FakeDiscoveryDao : DiscoveryDao {
     override fun getAllSessions(): Flow<List<DiscoverySessionEntity>> =
         flowOf(sessions.values.sortedByDescending { it.timestamp })
 
+    override suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity> = sessions.values.toList()
+
     override suspend fun getSession(sessionId: Long): DiscoverySessionEntity? = sessions[sessionId]
 
     override fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?> = MutableStateFlow(sessions[sessionId])