[fix](trx-rs): use per-message rig_id for map marker tagging

The map module was tagging all decode markers (APRS, AIS, VDES,
FT8/FT4/FT2/WSPR locators) with the global rig picker's active rig
instead of the actual source rig. This made the map's own rig filter
dropdown ineffective in multi-rig setups.

- Add rig_id field to all decode message structs (AisMessage,
  VdesMessage, AprsPacket, CwEvent, Ft8Message, WsprMessage)
- Set rig_id on messages in audio_client before broadcasting, using
  the actual rig connection identifier
- Update history collector to prefer message rig_id over the global
  active rig fallback
- Pass rig_id through plugin normalize functions (AIS, APRS, VDES,
  HF-APRS) so it reaches the map add functions
- Update all map marker functions (aprsMapAddStation, aisMapAddVessel,
  vdesMapAddPoint, mapAddLocator) to use the message's rig_id with
  fallback to the global picker for backward compatibility

https://claude.ai/code/session_015gC7axHk2jmp7HbFPdbivN
Signed-off-by: Claude <noreply@anthropic.com>

src/trx-client/src/audio_client.rs

@@ -548,7 +548,8 @@ async fn handle_single_rig_connection(
                                 break;
                             }
                             let json = &decompressed[pos..pos + len];
-                            if let Ok(msg) = serde_json::from_slice::<DecodedMessage>(json) {
+                            if let Ok(mut msg) = serde_json::from_slice::<DecodedMessage>(json) {
+                                msg.set_rig_id(rig_id_for_rx.clone());
                                 if let Some(ref sink) = replay_history_sink {
                                     sink(msg);
                                 }
@@ -569,7 +570,8 @@ async fn handle_single_rig_connection(
                     | AUDIO_MSG_WSPR_DECODE,
                     payload,
                 )) => {
-                    if let Ok(msg) = serde_json::from_slice::<DecodedMessage>(&payload) {
+                    if let Ok(mut msg) = serde_json::from_slice::<DecodedMessage>(&payload) {
+                        msg.set_rig_id(rig_id_for_rx.clone());
                         let _ = decode_tx_clone.send(msg);
                     }
                 }

src/trx-client/trx-frontend/trx-frontend-http/assets/web/app.js

@@ -6275,6 +6275,7 @@ function _aprsAddMarkerToMap(call, entry) {
 window.aprsMapAddStation = function(call, lat, lon, info, symbolTable, symbolCode, pkt) {
   const nextPoint = [lat, lon];
   const tsMs = Number.isFinite(pkt?._tsMs) ? Number(pkt._tsMs) : Date.now();
+  const msgRigId = pkt?.rig_id || lastActiveRigId;
   const existing = stationMarkers.get(call);
   if (existing) {
     existing.pkt = pkt;
@@ -6283,9 +6284,9 @@ window.aprsMapAddStation = function(call, lat, lon, info, symbolTable, symbolCod
     existing.info = info;
     existing.symbolTable = symbolTable;
     existing.symbolCode = symbolCode;
-    if (lastActiveRigId) {
+    if (msgRigId) {
       if (!existing.rigIds) existing.rigIds = new Set();
-      existing.rigIds.add(lastActiveRigId);
+      existing.rigIds.add(msgRigId);
     }
     if (!Array.isArray(existing.trackHistory)) existing.trackHistory = [];
     const prevPoint = existing.trackHistory[existing.trackHistory.length - 1];
@@ -6312,7 +6313,7 @@ window.aprsMapAddStation = function(call, lat, lon, info, symbolTable, symbolCod
       info,
       symbolTable,
       symbolCode,
-      rigIds: new Set(lastActiveRigId ? [lastActiveRigId] : []),
+      rigIds: new Set(msgRigId ? [msgRigId] : []),
     };
     stationMarkers.set(call, entry);
     pruneAprsEntry(call, entry, mapHistoryCutoffMs());
@@ -6407,12 +6408,13 @@ window.aisMapAddVessel = function(msg) {
   const popupHtml = buildAisPopupHtml(msg);
   const nextPoint = [msg.lat, msg.lon];
   const tsMs = Number.isFinite(msg?._tsMs) ? Number(msg._tsMs) : Date.now();
+  const msgRigId = msg?.rig_id || lastActiveRigId;
   const existing = aisMarkers.get(key);
   if (existing) {
     existing.msg = msg;
-    if (lastActiveRigId) {
+    if (msgRigId) {
       if (!existing.rigIds) existing.rigIds = new Set();
-      existing.rigIds.add(lastActiveRigId);
+      existing.rigIds.add(msgRigId);
     }
     if (!Array.isArray(existing.trackHistory)) existing.trackHistory = [];
     const prevPoint = existing.trackHistory[existing.trackHistory.length - 1];
@@ -6433,7 +6435,7 @@ window.aisMapAddVessel = function(msg) {
     trackHistory: [{ lat: msg.lat, lon: msg.lon, tsMs }],
     trackPoints: [nextPoint],
     msg,
-    rigIds: new Set(lastActiveRigId ? [lastActiveRigId] : []),
+    rigIds: new Set(msgRigId ? [msgRigId] : []),
   });
   pruneAisEntry(key, aisMarkers.get(key), mapHistoryCutoffMs());
   if (aisMarkers.get(key)?.visibleInHistoryWindow) ensureAisMarker(key, aisMarkers.get(key));
@@ -6447,13 +6449,14 @@ window.vdesMapAddPoint = function(msg) {
   const popupHtml = buildVdesPopupHtml(msg);
   const visible = Number.isFinite(Number(msg?._tsMs))
     && Number(msg._tsMs) >= mapHistoryCutoffMs();
+  const msgRigId = msg?.rig_id || lastActiveRigId;
   const existing = vdesMarkers.get(key);
   if (existing) {
     existing.msg = msg;
     existing.visibleInHistoryWindow = visible;
-    if (lastActiveRigId) {
+    if (msgRigId) {
       if (!existing.rigIds) existing.rigIds = new Set();
-      existing.rigIds.add(lastActiveRigId);
+      existing.rigIds.add(msgRigId);
     }
     if (!visible) {
       if (!decodeHistoryMapRenderingDeferred()) {
@@ -6479,7 +6482,7 @@ window.vdesMapAddPoint = function(msg) {
     marker: null,
     msg,
     visibleInHistoryWindow: visible,
-    rigIds: new Set(lastActiveRigId ? [lastActiveRigId] : []),
+    rigIds: new Set(msgRigId ? [msgRigId] : []),
   };
   vdesMarkers.set(key, entry);
   if (!visible) return;
@@ -7169,6 +7172,7 @@ window.syncBookmarkMapLocators = function(bookmarks) {
 window.mapAddLocator = function(message, grids, type = "ft8", station = null, details = null) {
   if (!Array.isArray(grids) || grids.length === 0) return;
   const markerType = type === "wspr" ? "wspr" : (type === "ft4" ? "ft4" : (type === "ft2" ? "ft2" : "ft8"));
+  const msgRigId = details?.rig_id || lastActiveRigId;
   const unique = [...new Set(grids.map((g) => String(g).toUpperCase()))];
   const stationId = station && String(station).trim() ? String(station).trim().toUpperCase() : "";
   const locatorDetails = new Map();
@@ -7199,7 +7203,7 @@ window.mapAddLocator = function(message, grids, type = "ft8", station = null, de
       dt_s: Number.isFinite(details?.dt_s) ? Number(details.dt_s) : null,
       freq_hz: Number.isFinite(details?.freq_hz) ? Number(details.freq_hz) : null,
       message: String(details?.message || message || "").trim() || null,
-      remote: lastActiveRigId || null,
+      remote: msgRigId || null,
     };
     const detailKey = detailStationId || `${targetId || "decode"}:${detailEntry.message || "decode"}:${detailEntry.ts_ms || Date.now()}`;
     const key = `${markerType}:${grid}`;
@@ -7213,9 +7217,9 @@ window.mapAddLocator = function(message, grids, type = "ft8", station = null, de
       }
       existing.allStationDetails.set(detailKey, { ...detailEntry });
       existing.sourceType = markerType;
-      if (lastActiveRigId) {
+      if (msgRigId) {
         if (!existing.rigIds) existing.rigIds = new Set();
-        existing.rigIds.add(lastActiveRigId);
+        existing.rigIds.add(msgRigId);
       }
       pruneLocatorEntry(key, existing, mapHistoryCutoffMs());
       if (existing.marker) sendLocatorOverlayToBack(existing.marker);
@@ -7233,7 +7237,7 @@ window.mapAddLocator = function(message, grids, type = "ft8", station = null, de
       allStationDetails,
       sourceType: markerType,
       bandMeta: new Map(),
-      rigIds: new Set(lastActiveRigId ? [lastActiveRigId] : []),
+      rigIds: new Set(msgRigId ? [msgRigId] : []),
     };
     locatorMarkers.set(key, entry);
     pruneLocatorEntry(key, entry, mapHistoryCutoffMs());

src/trx-client/trx-frontend/trx-frontend-http/assets/web/plugins/ais.js

@@ -326,6 +326,7 @@ function addAisMessage(msg) {
 
 function normalizeServerAisMessage(msg) {
   return {
+    rig_id: msg.rig_id || null,
     channel: msg.channel,
     message_type: msg.message_type,
     mmsi: msg.mmsi,

src/trx-client/trx-frontend/trx-frontend-http/assets/web/plugins/aprs.js

@@ -393,6 +393,7 @@ function addAprsPacket(pkt) {
 
 function normalizeServerAprsPacket(pkt) {
   return {
+    rig_id: pkt.rig_id || null,
     receiver: window.getDecodeRigMeta ? window.getDecodeRigMeta() : null,
     srcCall: pkt.src_call,
     destCall: pkt.dest_call,

src/trx-client/trx-frontend/trx-frontend-http/assets/web/plugins/hf-aprs.js

@@ -335,6 +335,7 @@ function addHfAprsPacket(pkt) {
 
 function normalizeServerHfAprsPacket(pkt) {
   return {
+    rig_id: pkt.rig_id || null,
     receiver: window.getDecodeRigMeta ? window.getDecodeRigMeta() : null,
     srcCall: pkt.src_call,
     destCall: pkt.dest_call,

src/trx-client/trx-frontend/trx-frontend-http/assets/web/plugins/vdes.js

@@ -256,6 +256,7 @@ function addVdesMessage(msg) {
 
 function normalizeServerVdesMessage(msg) {
   return {
+    rig_id: msg.rig_id || null,
     message_type: msg.message_type,
     bit_len: msg.bit_len,
     raw_bytes: msg.raw_bytes,

src/trx-client/trx-frontend/trx-frontend-http/src/audio.rs

@@ -121,11 +121,12 @@ fn record_ais(context: &FrontendRuntimeContext, mut msg: AisMessage) {
     if msg.ts_ms.is_none() {
         msg.ts_ms = Some(current_timestamp_ms());
     }
+    let rig_id = msg.rig_id.clone().or_else(|| active_rig_id(context));
     let mut history = context
         .ais_history
         .lock()
         .expect("ais history mutex poisoned");
-    history.push_back((Instant::now(), active_rig_id(context), msg));
+    history.push_back((Instant::now(), rig_id, msg));
     prune_ais_history(context, &mut history);
 }
 
@@ -133,11 +134,12 @@ fn record_vdes(context: &FrontendRuntimeContext, mut msg: VdesMessage) {
     if msg.ts_ms.is_none() {
         msg.ts_ms = Some(current_timestamp_ms());
     }
+    let rig_id = msg.rig_id.clone().or_else(|| active_rig_id(context));
     let mut history = context
         .vdes_history
         .lock()
         .expect("vdes history mutex poisoned");
-    history.push_back((Instant::now(), active_rig_id(context), msg));
+    history.push_back((Instant::now(), rig_id, msg));
     prune_vdes_history(context, &mut history);
 }
 
@@ -210,11 +212,12 @@ fn record_aprs(context: &FrontendRuntimeContext, mut pkt: AprsPacket) {
     if pkt.ts_ms.is_none() {
         pkt.ts_ms = Some(current_timestamp_ms());
     }
+    let rig_id = pkt.rig_id.clone().or_else(|| active_rig_id(context));
     let mut history = context
         .aprs_history
         .lock()
         .expect("aprs history mutex poisoned");
-    history.push_back((Instant::now(), active_rig_id(context), pkt));
+    history.push_back((Instant::now(), rig_id, pkt));
     prune_aprs_history(context, &mut history);
 }
 
@@ -222,56 +225,62 @@ fn record_hf_aprs(context: &FrontendRuntimeContext, mut pkt: AprsPacket) {
     if pkt.ts_ms.is_none() {
         pkt.ts_ms = Some(current_timestamp_ms());
     }
+    let rig_id = pkt.rig_id.clone().or_else(|| active_rig_id(context));
     let mut history = context
         .hf_aprs_history
         .lock()
         .expect("hf_aprs history mutex poisoned");
-    history.push_back((Instant::now(), active_rig_id(context), pkt));
+    history.push_back((Instant::now(), rig_id, pkt));
     prune_hf_aprs_history(context, &mut history);
 }
 
 fn record_cw(context: &FrontendRuntimeContext, event: CwEvent) {
+    let rig_id = event.rig_id.clone().or_else(|| active_rig_id(context));
     let mut history = context
         .cw_history
         .lock()
         .expect("cw history mutex poisoned");
-    history.push_back((Instant::now(), active_rig_id(context), event));
+    history.push_back((Instant::now(), rig_id, event));
     prune_cw_history(context, &mut history);
 }
 
 fn record_ft8(context: &FrontendRuntimeContext, msg: Ft8Message) {
+    let rig_id = msg.rig_id.clone().or_else(|| active_rig_id(context));
     let mut history = context
         .ft8_history
         .lock()
         .expect("ft8 history mutex poisoned");
-    history.push_back((Instant::now(), active_rig_id(context), msg));
+    history.push_back((Instant::now(), rig_id, msg));
     prune_ft8_history(context, &mut history);
 }
 
 fn record_ft4(context: &FrontendRuntimeContext, msg: Ft8Message) {
+    let rig_id = msg.rig_id.clone().or_else(|| active_rig_id(context));
     let mut history = context
         .ft4_history
         .lock()
         .expect("ft4 history mutex poisoned");
-    history.push_back((Instant::now(), active_rig_id(context), msg));
+    history.push_back((Instant::now(), rig_id, msg));
     prune_ft4_history(context, &mut history);
 }
 
 fn record_ft2(context: &FrontendRuntimeContext, msg: Ft8Message) {
+    let rig_id = msg.rig_id.clone().or_else(|| active_rig_id(context));
     let mut history = context
         .ft2_history
         .lock()
         .expect("ft2 history mutex poisoned");
-    history.push_back((Instant::now(), active_rig_id(context), msg));
+    history.push_back((Instant::now(), rig_id, msg));
     prune_ft2_history(context, &mut history);
 }
 
 fn record_wspr(context: &FrontendRuntimeContext, msg: WsprMessage) {
+    let rig_id = msg.rig_id.clone().or_else(|| active_rig_id(context));
     let mut history = context
         .wspr_history
         .lock()
         .expect("wspr history mutex poisoned");
-    history.push_back((Instant::now(), active_rig_id(context), msg));
+    history.push_back((Instant::now(), rig_id, msg));
     prune_wspr_history(context, &mut history);
 }