web: fix meshcore message duplication with 120s dupe protection (#758)

* web: fix meshcore message duplication with 120s dupe protection

* web: fix meshcore message duplication with 120s dupe protection

* web: address review comments

* web: address review comments

web/spec/data_processing_spec.rb

@@ -804,4 +804,226 @@ RSpec.describe PotatoMesh::App::DataProcessing do
       db&.close
     end
   end
+
+  # ---------------------------------------------------------------------------
+  # insert_message — meshcore content dedup (issue #756).
+  # ---------------------------------------------------------------------------
+  describe "#insert_message — meshcore content dedup" do
+    include_context "with isolated db"
+
+    let(:now) { Time.now.to_i }
+
+    # Shared builder for a minimal ``insert_message`` harness parameterised
+    # by the protocol it advertises for every POST.  Keeping this in one
+    # place (rather than duplicating per-describe) matches CLAUDE.md's
+    # modularity guidance and makes it trivial to add a third protocol.
+    def self.build_protocol_harness(protocol_name)
+      Class.new do
+        include PotatoMesh::App::DataProcessing
+        include PotatoMesh::App::Helpers
+
+        define_method(:resolve_protocol) do |_db, _ingestor, cache: nil|
+          protocol_name
+        end
+
+        def debug_log(message, **); end
+
+        def warn_log(message, **); end
+
+        def with_busy_retry
+          yield
+        end
+
+        def update_prometheus_metrics(*); end
+
+        def prom_report_ids
+          []
+        end
+
+        def private_mode?
+          false
+        end
+
+        def normalize_node_id(_db, node_ref)
+          parts = canonical_node_parts(node_ref)
+          parts ? parts[0] : nil
+        end
+
+        def touch_node_last_seen(*); end
+
+        def ensure_unknown_node(*); end
+      end.new
+    end
+
+    let(:meshcore_harness) { self.class.build_protocol_harness("meshcore") }
+    let(:meshtastic_harness) { self.class.build_protocol_harness("meshtastic") }
+
+    # rx_time sits in the past so we can shift later copies forward (up to
+    # ``now``) without tripping the ``rx_time > now`` clamp in
+    # ``insert_message``.
+    let(:base_rx_time) { now - 1_000 }
+    let(:dedup_window) { PotatoMesh::App::DataProcessing::MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS }
+
+    let(:base_message) do
+      {
+        "rx_time" => base_rx_time,
+        "from_id" => "!aabbccdd",
+        "to_id" => "^all",
+        "channel" => 5,
+        "text" => "hello from alice",
+        "portnum" => "TEXT_MESSAGE_APP",
+        "ingestor" => "!ingest01",
+      }
+    end
+
+    def message_count(db)
+      db.get_first_value("SELECT COUNT(*) FROM messages").to_i
+    end
+
+    it "skips a second meshcore message with identical content within the dedup window" do
+      db = open_db
+      meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_001))
+      meshcore_harness.insert_message(
+        db,
+        base_message.merge("id" => 1_000_002, "rx_time" => base_rx_time + (dedup_window - 1)),
+      )
+      expect(message_count(db)).to eq(1)
+      expect(db.get_first_value("SELECT id FROM messages").to_i).to eq(1_000_001)
+    ensure
+      db&.close
+    end
+
+    it "treats the dedup window as inclusive on the upper boundary" do
+      # Pins the ``BETWEEN`` inclusivity: a row exactly ``dedup_window`` seconds
+      # later still collapses.  One-second-past-the-window inserts below prove
+      # the other side of the boundary.
+      db = open_db
+      meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_021))
+      meshcore_harness.insert_message(
+        db,
+        base_message.merge("id" => 1_000_022, "rx_time" => base_rx_time + dedup_window),
+      )
+      expect(message_count(db)).to eq(1)
+      meshcore_harness.insert_message(
+        db,
+        base_message.merge("id" => 1_000_023, "rx_time" => base_rx_time + dedup_window + 1),
+      )
+      expect(message_count(db)).to eq(2)
+    ensure
+      db&.close
+    end
+
+    it "inserts both copies when rx_time delta exceeds the dedup window" do
+      db = open_db
+      meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_003))
+      meshcore_harness.insert_message(
+        db,
+        base_message.merge("id" => 1_000_004, "rx_time" => base_rx_time + (dedup_window * 3)),
+      )
+      expect(message_count(db)).to eq(2)
+    ensure
+      db&.close
+    end
+
+    it "does not collapse two meshcore messages on different channels" do
+      db = open_db
+      meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_005, "channel" => 5))
+      meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_006, "channel" => 6))
+      expect(message_count(db)).to eq(2)
+    ensure
+      db&.close
+    end
+
+    it "does not collapse two meshcore messages with different text" do
+      db = open_db
+      meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_007, "text" => "first"))
+      meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_008, "text" => "second"))
+      expect(message_count(db)).to eq(2)
+    ensure
+      db&.close
+    end
+
+    it "does not collapse two meshcore DMs to different recipients sharing text" do
+      db = open_db
+      meshcore_harness.insert_message(
+        db,
+        base_message.merge("id" => 1_000_009, "to_id" => "!bbbbbbbb"),
+      )
+      meshcore_harness.insert_message(
+        db,
+        base_message.merge("id" => 1_000_010, "to_id" => "!cccccccc", "rx_time" => base_rx_time + 5),
+      )
+      expect(message_count(db)).to eq(2)
+    ensure
+      db&.close
+    end
+
+    it "does not collapse when the incoming message has no text" do
+      db = open_db
+      meshcore_harness.insert_message(
+        db,
+        base_message.merge("id" => 1_000_011, "text" => "blob"),
+      )
+      # Second payload has no text — the content-dedup branch must not fire,
+      # so this falls through to the normal id-PK path and inserts.
+      meshcore_harness.insert_message(
+        db,
+        base_message.merge("id" => 1_000_012, "text" => nil, "rx_time" => base_rx_time + 5),
+      )
+      expect(message_count(db)).to eq(2)
+    ensure
+      db&.close
+    end
+
+    it "leaves meshtastic traffic untouched" do
+      db = open_db
+      # Two meshtastic packets with the same logical content but distinct
+      # firmware-assigned packet ids must both land — the new guard is
+      # scoped to meshcore by design.
+      meshtastic_harness.insert_message(db, base_message.merge("id" => 1_000_013))
+      meshtastic_harness.insert_message(
+        db,
+        base_message.merge("id" => 1_000_014, "rx_time" => base_rx_time + 5),
+      )
+      expect(message_count(db)).to eq(2)
+    ensure
+      db&.close
+    end
+
+    it "never issues the content-dedup SELECT for non-meshcore traffic" do
+      # Pins the performance contract: meshtastic traffic must skip the
+      # partial-index lookup entirely so any future regression that makes
+      # the pre-check unconditional surfaces as a failing test.
+      db = open_db
+      content_select_pattern = /SELECT\s+id\s+FROM\s+messages\s+WHERE\s+protocol\s*=\s*'meshcore'/im
+      captured_sql = []
+      wrapped = db.method(:get_first_value)
+      allow(db).to receive(:get_first_value) do |sql, *rest|
+        captured_sql << sql
+        wrapped.call(sql, *rest)
+      end
+      meshtastic_harness.insert_message(db, base_message.merge("id" => 1_000_020))
+      expect(captured_sql.any? { |s| s =~ content_select_pattern }).to be(false)
+    ensure
+      db&.close
+    end
+
+    it "still merges on the id-PK path when sender_timestamps collide on the wire" do
+      db = open_db
+      # Same id, same content — the existing update-on-match code path should
+      # patch the stored row rather than insert a duplicate.  This proves the
+      # new dedup guard does not short-circuit the id-match merge behaviour.
+      meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_015, "ingestor" => nil))
+      meshcore_harness.insert_message(
+        db,
+        base_message.merge("id" => 1_000_015, "ingestor" => "!ingest99"),
+      )
+      expect(message_count(db)).to eq(1)
+      expect(
+        db.get_first_value("SELECT ingestor FROM messages WHERE id = 1000015"),
+      ).to eq("!ingest99")
+    ensure
+      db&.close
+    end
+  end
 end