diff --git a/core/common/src/commonMain/kotlin/org/meshtastic/core/common/database/DatabaseManager.kt b/core/common/src/commonMain/kotlin/org/meshtastic/core/common/database/DatabaseManager.kt index 9e84b74ee..adf0b93dc 100644 --- a/core/common/src/commonMain/kotlin/org/meshtastic/core/common/database/DatabaseManager.kt +++ b/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 } diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseConstants.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseConstants.kt index d72187b9b..589bceb77 100644 --- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseConstants.kt +++ b/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 diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt index 3511c4e59..bbfe75cbe 100644 --- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt +++ b/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(null) val currentAddress: StateFlow = _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 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) diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseMerger.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseMerger.kt new file mode 100644 index 000000000..3dfda6384 --- /dev/null +++ b/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 . + */ +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) + } + } + } +} diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/DiscoveryDao.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/DiscoveryDao.kt index 7e4ebdf5e..4a115fbea 100644 --- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/DiscoveryDao.kt +++ b/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> + /** Snapshot used by DatabaseMerger to append a transport's discovery sessions to the unified DB. */ + @Query("SELECT * FROM discovery_session") + suspend fun getAllSessionsSnapshot(): List + @Query("SELECT * FROM discovery_session WHERE id = :sessionId") suspend fun getSession(sessionId: Long): DiscoverySessionEntity? diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/MeshLogDao.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/MeshLogDao.kt index 3edabfc15..9369bbf0e 100644 --- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/MeshLogDao.kt +++ b/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 + + @Insert(onConflict = OnConflictStrategy.IGNORE) + suspend fun insertIgnore(logs: List) + @Query("DELETE FROM log") suspend fun deleteAll() diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/NodeInfoDao.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/NodeInfoDao.kt index 2966e4e49..39686488e 100644 --- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/NodeInfoDao.kt +++ b/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 + @Query("SELECT * FROM nodes WHERE num=:num") @Transaction suspend fun getNodeByNum(num: Int): NodeWithRelations? diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt index a7083e37d..801f20b86 100644 --- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt +++ b/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 + + /** 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 + + @Insert(onConflict = OnConflictStrategy.IGNORE) + suspend fun insertReactionsIgnore(reactions: List) + + @Query("SELECT * FROM contact_settings") + suspend fun getAllContactSettingsSnapshot(): List + + // 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 diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/TracerouteNodePositionDao.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/TracerouteNodePositionDao.kt index ffd0ecc2b..85988f781 100644 --- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/TracerouteNodePositionDao.kt +++ b/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 + @Upsert suspend fun insertAll(entities: List) } diff --git a/core/database/src/commonTest/kotlin/org/meshtastic/core/database/DatabaseMergerTest.kt b/core/database/src/commonTest/kotlin/org/meshtastic/core/database/DatabaseMergerTest.kt new file mode 100644 index 000000000..e91ac184f --- /dev/null +++ b/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 . + */ +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", + ) + } +} diff --git a/core/service/src/commonMain/kotlin/org/meshtastic/core/service/RadioControllerImpl.kt b/core/service/src/commonMain/kotlin/org/meshtastic/core/service/RadioControllerImpl.kt index 0daded9f3..d52e93c09 100644 --- a/core/service/src/commonMain/kotlin/org/meshtastic/core/service/RadioControllerImpl.kt +++ b/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 diff --git a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeDatabaseManager.kt b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeDatabaseManager.kt index 5693f112f..5929cebef 100644 --- a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeDatabaseManager.kt +++ b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeDatabaseManager.kt @@ -29,12 +29,14 @@ class FakeDatabaseManager : override val cacheLimit: StateFlow = _cacheLimit var lastSwitchedAddress: String? = null + var lastAssociatedNode: Int? = null val existingDatabases = mutableSetOf() 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 { diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHistoryBehaviorTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHistoryBehaviorTest.kt index d4558cd06..0ba758465 100644 --- a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHistoryBehaviorTest.kt +++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHistoryBehaviorTest.kt @@ -185,6 +185,8 @@ private class HistoryTestDao : DiscoveryDao { override fun getAllSessions(): Flow> = sessionsFlow + override suspend fun getAllSessionsSnapshot(): List = sessions.values.toList() + override suspend fun getSession(sessionId: Long) = sessions[sessionId] override fun getSessionFlow(sessionId: Long): Flow = MutableStateFlow(sessions[sessionId]) diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryMapFilterTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryMapFilterTest.kt index 6a62fced8..b334fda62 100644 --- a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryMapFilterTest.kt +++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryMapFilterTest.kt @@ -171,6 +171,8 @@ private class MapTestDao : DiscoveryDao { override fun getAllSessions(): Flow> = flowOf(sessions.values.sortedByDescending { it.timestamp }) + override suspend fun getAllSessionsSnapshot(): List = sessions.values.toList() + override suspend fun getSession(sessionId: Long) = sessions[sessionId] override fun getSessionFlow(sessionId: Long): Flow = MutableStateFlow(sessions[sessionId]) diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryPacketCollectionTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryPacketCollectionTest.kt index b89915674..d5688b2d5 100644 --- a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryPacketCollectionTest.kt +++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryPacketCollectionTest.kt @@ -352,6 +352,8 @@ private class InMemoryDiscoveryDao : DiscoveryDao { override fun getAllSessions(): Flow> = flowOf(sessions.values.sortedByDescending { it.timestamp }) + override suspend fun getAllSessionsSnapshot(): List = sessions.values.toList() + override suspend fun getSession(sessionId: Long): DiscoverySessionEntity? = sessions[sessionId] override fun getSessionFlow(sessionId: Long): Flow = MutableStateFlow(sessions[sessionId]) diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngineTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngineTest.kt index ae5dc1a80..05ac0a433 100644 --- a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngineTest.kt +++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngineTest.kt @@ -89,6 +89,8 @@ private class FakeDiscoveryDao : DiscoveryDao { override fun getAllSessions(): Flow> = flowOf(sessions.values.sortedByDescending { it.timestamp }) + override suspend fun getAllSessionsSnapshot(): List = sessions.values.toList() + override suspend fun getSession(sessionId: Long): DiscoverySessionEntity? = sessions[sessionId] override fun getSessionFlow(sessionId: Long): Flow = MutableStateFlow(sessions[sessionId])