clickhouse: back map, node search, and chat with live materialized views

Replace the per-request argMax/GROUP-BY views with insert-triggered
(incremental) materialized views so map node positions, node search, and
public-channel chat read pre-aggregated state instead of re-scanning all of
meshcore_packets on every query.

- 005: meshcore_adverts_latest_state (AggregatingMergeTree of argMaxState/
  min/maxState) + incremental MV + backfill; meshcore_adverts_latest becomes a
  -Merge view with the identical column contract. Node search reads it directly;
  map (unified_latest_nodeinfo) is unchanged.
- 006: meshcore_public_channel_messages_raw, a decoded payload_type=5 MergeTree
  keyed (channel_hash, ingest_timestamp); chat dedups by message_id at read time
  over a timestamp-bounded scan. Streaming/pagination push channel+cursor onto
  the primary key.
- Neighbor-edge MVs stay hourly REFRESH (they read the preserved view).

Verified against full prod data (14.5M rows): exact parity (0 mismatches) and
5-9x faster reads with no regressions.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

ingest/migrations/005_live_adverts_latest_mv.sql

@@ -0,0 +1,214 @@
+-- +goose Up
+-- Convert meshcore_adverts_latest from a plain argMax/GROUP-BY-public_key VIEW (which
+-- re-aggregates the entire payload_type=4 history on every read) into a LIVE, insert-triggered
+-- materialized view backed by an AggregatingMergeTree state table.
+--
+-- An incremental MV only sees the newly-inserted block, so it cannot do a global GROUP BY
+-- across history. The standard pattern (used here) is: an AggregatingMergeTree target table
+-- holding per-public_key partial aggregate states (argMaxState/minState/maxState), an
+-- incremental MV that emits those states per inserted block, and a thin read-side VIEW that
+-- collapses them with -Merge ... GROUP BY public_key. The public name `meshcore_adverts_latest`
+-- stays a VIEW exposing the identical column contract, so unified_latest_nodeinfo,
+-- api/stats/repeater-prefixes, and the hourly REFRESH MVs (meshcore_all_neighbor_edges,
+-- meshcore_node_direct_neighbors, meshcore_regions) keep working unchanged.
+--
+-- Type-pinning notes (argMaxState value type must EXACTLY equal the SELECT expression type):
+--   * bool flags (is_*/has_*) -> toUInt8(...) over UInt8 state columns
+--   * broker/topic are LowCardinality(String) -> toString(...) over plain String state columns
+--     (the merge view re-exposes plain String; all consumers accept String)
+--   * latitude_i/longitude_i are Nullable(Int32); latitude/longitude are Nullable(Float64)
+
+-- +goose StatementBegin
+CREATE TABLE IF NOT EXISTS meshcore_adverts_latest_state
+(
+    public_key      String,
+    first_heard     AggregateFunction(min, DateTime64(3)),
+    last_seen       AggregateFunction(max, DateTime64(3)),
+    broker          AggregateFunction(argMax, String,            DateTime64(3)),
+    topic           AggregateFunction(argMax, String,            DateTime64(3)),
+    region          AggregateFunction(argMax, String,            DateTime64(3)),
+    origin          AggregateFunction(argMax, String,            DateTime64(3)),
+    mesh_timestamp  AggregateFunction(argMax, DateTime64(6),     DateTime64(3)),
+    packet          AggregateFunction(argMax, String,            DateTime64(3)),
+    path_len        AggregateFunction(argMax, UInt8,             DateTime64(3)),
+    path            AggregateFunction(argMax, String,            DateTime64(3)),
+    adv_timestamp   AggregateFunction(argMax, UInt32,            DateTime64(3)),
+    signature       AggregateFunction(argMax, String,            DateTime64(3)),
+    appdata_flags   AggregateFunction(argMax, UInt8,             DateTime64(3)),
+    is_chat_node    AggregateFunction(argMax, UInt8,             DateTime64(3)),
+    is_repeater     AggregateFunction(argMax, UInt8,             DateTime64(3)),
+    is_room_server  AggregateFunction(argMax, UInt8,             DateTime64(3)),
+    has_location    AggregateFunction(argMax, UInt8,             DateTime64(3)),
+    has_feature1    AggregateFunction(argMax, UInt8,             DateTime64(3)),
+    has_feature2    AggregateFunction(argMax, UInt8,             DateTime64(3)),
+    has_name        AggregateFunction(argMax, UInt8,             DateTime64(3)),
+    latitude_i      AggregateFunction(argMax, Nullable(Int32),   DateTime64(3)),
+    longitude_i     AggregateFunction(argMax, Nullable(Int32),   DateTime64(3)),
+    latitude        AggregateFunction(argMax, Nullable(Float64), DateTime64(3)),
+    longitude       AggregateFunction(argMax, Nullable(Float64), DateTime64(3)),
+    node_name       AggregateFunction(argMax, String,            DateTime64(3)),
+    node_hash       AggregateFunction(argMax, String,            DateTime64(3)),
+    packet_hash     AggregateFunction(argMax, String,            DateTime64(3))
+)
+ENGINE = AggregatingMergeTree
+ORDER BY public_key;
+-- +goose StatementEnd
+
+-- Incremental MV: fires on every insert into meshcore_packets. Reads the base table directly
+-- (an MV cannot trigger off the meshcore_adverts VIEW) and inlines the payload decode.
+-- +goose StatementBegin
+CREATE MATERIALIZED VIEW IF NOT EXISTS meshcore_adverts_latest_mv
+TO meshcore_adverts_latest_state
+AS
+SELECT
+    hex(substring(payload, 1, 32)) AS public_key,
+    minState(ingest_timestamp) AS first_heard,
+    maxState(ingest_timestamp) AS last_seen,
+    argMaxState(toString(broker), ingest_timestamp) AS broker,
+    argMaxState(toString(topic), ingest_timestamp) AS topic,
+    argMaxState(multiIf(lower(meshcore_packets.topic) IN ('meshcore','meshcore/salish'), 'SEA', match(splitByChar('/', lower(meshcore_packets.topic))[2], '^[a-z]{3}$'), upper(splitByChar('/', lower(meshcore_packets.topic))[2]), ''), ingest_timestamp) AS region,
+    argMaxState(origin, ingest_timestamp) AS origin,
+    argMaxState(mesh_timestamp, ingest_timestamp) AS mesh_timestamp,
+    argMaxState(packet, ingest_timestamp) AS packet,
+    argMaxState(path_len, ingest_timestamp) AS path_len,
+    argMaxState(path, ingest_timestamp) AS path,
+    argMaxState(reinterpretAsUInt32(substring(payload, 33, 4)), ingest_timestamp) AS adv_timestamp,
+    argMaxState(hex(substring(payload, 37, 64)), ingest_timestamp) AS signature,
+    argMaxState(reinterpretAsUInt8(substring(payload, 101, 1)), ingest_timestamp) AS appdata_flags,
+    argMaxState(toUInt8(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x01) = 0x01), ingest_timestamp) AS is_chat_node,
+    argMaxState(toUInt8(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x02) = 0x02), ingest_timestamp) AS is_repeater,
+    argMaxState(toUInt8(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x03) = 0x03), ingest_timestamp) AS is_room_server,
+    argMaxState(toUInt8(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10), ingest_timestamp) AS has_location,
+    argMaxState(toUInt8(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x20) = 0x20), ingest_timestamp) AS has_feature1,
+    argMaxState(toUInt8(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x40) = 0x40), ingest_timestamp) AS has_feature2,
+    argMaxState(toUInt8(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x80) = 0x80), ingest_timestamp) AS has_name,
+    argMaxState(CASE WHEN bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10 THEN reinterpretAsInt32(substring(payload, 102, 4)) ELSE NULL END, ingest_timestamp) AS latitude_i,
+    argMaxState(CASE WHEN bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10 THEN reinterpretAsInt32(substring(payload, 106, 4)) ELSE NULL END, ingest_timestamp) AS longitude_i,
+    argMaxState((CASE WHEN bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10 THEN reinterpretAsInt32(substring(payload, 102, 4)) ELSE NULL END) * 1e-6, ingest_timestamp) AS latitude,
+    argMaxState((CASE WHEN bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10 THEN reinterpretAsInt32(substring(payload, 106, 4)) ELSE NULL END) * 1e-6, ingest_timestamp) AS longitude,
+    argMaxState(substring(payload, 102 + multiIf(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10, 8, 0)), ingest_timestamp) AS node_name,
+    argMaxState(hex(substring(payload, 1, 1)), ingest_timestamp) AS node_hash,
+    argMaxState(packet_hash, ingest_timestamp) AS packet_hash
+FROM meshcore_packets
+WHERE payload_type = 4
+GROUP BY public_key;
+-- +goose StatementEnd
+
+-- One-time backfill of existing history. Created AFTER the MV so no insert is lost; the small
+-- create->backfill overlap can double-count rows, which is harmless for min/max/argMax
+-- (duplicate (value, ingest_timestamp) pairs collapse on merge).
+-- +goose StatementBegin
+INSERT INTO meshcore_adverts_latest_state
+SELECT
+    hex(substring(payload, 1, 32)) AS public_key,
+    minState(ingest_timestamp),
+    maxState(ingest_timestamp),
+    argMaxState(toString(broker), ingest_timestamp),
+    argMaxState(toString(topic), ingest_timestamp),
+    argMaxState(multiIf(lower(meshcore_packets.topic) IN ('meshcore','meshcore/salish'), 'SEA', match(splitByChar('/', lower(meshcore_packets.topic))[2], '^[a-z]{3}$'), upper(splitByChar('/', lower(meshcore_packets.topic))[2]), ''), ingest_timestamp),
+    argMaxState(origin, ingest_timestamp),
+    argMaxState(mesh_timestamp, ingest_timestamp),
+    argMaxState(packet, ingest_timestamp),
+    argMaxState(path_len, ingest_timestamp),
+    argMaxState(path, ingest_timestamp),
+    argMaxState(reinterpretAsUInt32(substring(payload, 33, 4)), ingest_timestamp),
+    argMaxState(hex(substring(payload, 37, 64)), ingest_timestamp),
+    argMaxState(reinterpretAsUInt8(substring(payload, 101, 1)), ingest_timestamp),
+    argMaxState(toUInt8(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x01) = 0x01), ingest_timestamp),
+    argMaxState(toUInt8(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x02) = 0x02), ingest_timestamp),
+    argMaxState(toUInt8(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x03) = 0x03), ingest_timestamp),
+    argMaxState(toUInt8(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10), ingest_timestamp),
+    argMaxState(toUInt8(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x20) = 0x20), ingest_timestamp),
+    argMaxState(toUInt8(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x40) = 0x40), ingest_timestamp),
+    argMaxState(toUInt8(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x80) = 0x80), ingest_timestamp),
+    argMaxState(CASE WHEN bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10 THEN reinterpretAsInt32(substring(payload, 102, 4)) ELSE NULL END, ingest_timestamp),
+    argMaxState(CASE WHEN bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10 THEN reinterpretAsInt32(substring(payload, 106, 4)) ELSE NULL END, ingest_timestamp),
+    argMaxState((CASE WHEN bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10 THEN reinterpretAsInt32(substring(payload, 102, 4)) ELSE NULL END) * 1e-6, ingest_timestamp),
+    argMaxState((CASE WHEN bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10 THEN reinterpretAsInt32(substring(payload, 106, 4)) ELSE NULL END) * 1e-6, ingest_timestamp),
+    argMaxState(substring(payload, 102 + multiIf(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10, 8, 0)), ingest_timestamp),
+    argMaxState(hex(substring(payload, 1, 1)), ingest_timestamp),
+    argMaxState(packet_hash, ingest_timestamp)
+FROM meshcore_packets
+WHERE payload_type = 4
+GROUP BY public_key;
+-- +goose StatementEnd
+
+-- Replace the public view in place: same column set/names/order as before, now collapsing the
+-- partial states with -Merge. Stays a VIEW, so downstream consumers are untouched.
+-- +goose StatementBegin
+CREATE OR REPLACE VIEW meshcore_adverts_latest AS
+SELECT
+    public_key,
+    minMerge(first_heard) AS first_heard,
+    maxMerge(last_seen) AS last_seen,
+    argMaxMerge(broker) AS broker,
+    argMaxMerge(topic) AS topic,
+    argMaxMerge(region) AS region,
+    argMaxMerge(origin) AS origin,
+    argMaxMerge(mesh_timestamp) AS mesh_timestamp,
+    argMaxMerge(packet) AS packet,
+    argMaxMerge(path_len) AS path_len,
+    argMaxMerge(path) AS path,
+    argMaxMerge(adv_timestamp) AS adv_timestamp,
+    argMaxMerge(signature) AS signature,
+    argMaxMerge(appdata_flags) AS appdata_flags,
+    argMaxMerge(is_chat_node) AS is_chat_node,
+    argMaxMerge(is_repeater) AS is_repeater,
+    argMaxMerge(is_room_server) AS is_room_server,
+    argMaxMerge(has_location) AS has_location,
+    argMaxMerge(has_feature1) AS has_feature1,
+    argMaxMerge(has_feature2) AS has_feature2,
+    argMaxMerge(has_name) AS has_name,
+    argMaxMerge(latitude_i) AS latitude_i,
+    argMaxMerge(longitude_i) AS longitude_i,
+    argMaxMerge(latitude) AS latitude,
+    argMaxMerge(longitude) AS longitude,
+    argMaxMerge(node_name) AS node_name,
+    argMaxMerge(node_hash) AS node_hash,
+    argMaxMerge(packet_hash) AS packet_hash
+FROM meshcore_adverts_latest_state
+GROUP BY public_key
+ORDER BY last_seen DESC;
+-- +goose StatementEnd
+
+
+-- +goose Down
+-- Restore the plain argMax/GROUP-BY VIEW (verbatim from migration 004) before dropping the
+-- state objects so the public name is never broken.
+-- +goose StatementBegin
+CREATE OR REPLACE VIEW meshcore_adverts_latest AS
+SELECT
+    public_key,
+    min(ingest_timestamp) AS first_heard,
+    max(ingest_timestamp) AS last_seen,
+    argMax(broker, ingest_timestamp) AS broker,
+    argMax(topic, ingest_timestamp) AS topic,
+    argMax(region, ingest_timestamp) AS region,
+    argMax(origin, ingest_timestamp) AS origin,
+    argMax(mesh_timestamp, ingest_timestamp) AS mesh_timestamp,
+    argMax(packet, ingest_timestamp) AS packet,
+    argMax(path_len, ingest_timestamp) AS path_len,
+    argMax(path, ingest_timestamp) AS path,
+    argMax(adv_timestamp, ingest_timestamp) AS adv_timestamp,
+    argMax(signature, ingest_timestamp) AS signature,
+    argMax(appdata_flags, ingest_timestamp) AS appdata_flags,
+    argMax(is_chat_node, ingest_timestamp) AS is_chat_node,
+    argMax(is_repeater, ingest_timestamp) AS is_repeater,
+    argMax(is_room_server, ingest_timestamp) AS is_room_server,
+    argMax(has_location, ingest_timestamp) AS has_location,
+    argMax(has_feature1, ingest_timestamp) AS has_feature1,
+    argMax(has_feature2, ingest_timestamp) AS has_feature2,
+    argMax(has_name, ingest_timestamp) AS has_name,
+    argMax(latitude_i, ingest_timestamp) AS latitude_i,
+    argMax(longitude_i, ingest_timestamp) AS longitude_i,
+    argMax(latitude, ingest_timestamp) AS latitude,
+    argMax(longitude, ingest_timestamp) AS longitude,
+    argMax(node_name, ingest_timestamp) AS node_name,
+    argMax(node_hash, ingest_timestamp) AS node_hash,
+    argMax(packet_hash, ingest_timestamp) AS packet_hash
+FROM meshcore_adverts
+GROUP BY public_key
+ORDER BY last_seen DESC;
+-- +goose StatementEnd
+DROP VIEW IF EXISTS meshcore_adverts_latest_mv;
+DROP TABLE IF EXISTS meshcore_adverts_latest_state;

ingest/migrations/006_live_public_channel_messages_mv.sql

@@ -0,0 +1,130 @@
+-- +goose Up
+-- Convert meshcore_public_channel_messages from a plain GROUP-BY-payload VIEW into a LIVE,
+-- insert-triggered materialized view.
+--
+-- Public channel messages are ALWAYS queried by timestamp (streaming poll + pagination cursor on
+-- ingest_timestamp, usually scoped to a channel). An AggregatingMergeTree keyed by message identity
+-- would make those queries un-indexed (the timestamp would be a merged aggregate, not a sort key).
+-- So instead the MV is a DECODED, payload_type=5-only MergeTree ordered by (channel_hash,
+-- ingest_timestamp) with one row per gateway-copy. Cross-gateway dedup (collapsing the same
+-- encrypted message heard via multiple gateways) is done at READ time with GROUP BY message_id over
+-- a timestamp-bounded scan -- exactly what the app does today, but against a smaller, pre-filtered,
+-- pre-decoded table instead of all of meshcore_packets. This keeps per-channel timestamp queries
+-- index-accelerated, preserves correct dedup, and is a true speedup.
+--
+-- This is a row-per-packet transform (no GROUP BY), so it is a pure live incremental MV and there
+-- are no aggregate-state type-pinning concerns.
+
+-- +goose StatementBegin
+CREATE TABLE IF NOT EXISTS meshcore_public_channel_messages_raw
+(
+    ingest_timestamp  DateTime64(3),
+    mesh_timestamp    DateTime64(6),
+    channel_hash      String,
+    mac               String,
+    encrypted_message String,
+    message_id        String,
+    origin            String,
+    origin_pubkey     String,
+    path              String,
+    broker            String,
+    topic             String,
+    region            String
+)
+ENGINE = MergeTree
+ORDER BY (channel_hash, ingest_timestamp)
+PARTITION BY toYYYYMM(ingest_timestamp);
+-- +goose StatementEnd
+
+-- Incremental MV: decode each payload_type=5 packet into a row. Created before the backfill so no
+-- insert is lost.
+-- +goose StatementBegin
+CREATE MATERIALIZED VIEW IF NOT EXISTS meshcore_public_channel_messages_mv
+TO meshcore_public_channel_messages_raw
+AS
+SELECT
+    ingest_timestamp,
+    mesh_timestamp,
+    hex(substring(payload, 1, 1)) AS channel_hash,
+    hex(substring(payload, 2, 2)) AS mac,
+    substring(payload, 4) AS encrypted_message,
+    packet_hash AS message_id,
+    origin,
+    hex(origin_pubkey) AS origin_pubkey,
+    path,
+    toString(broker) AS broker,
+    toString(topic) AS topic,
+    multiIf(lower(meshcore_packets.topic) IN ('meshcore','meshcore/salish'), 'SEA', match(splitByChar('/', lower(meshcore_packets.topic))[2], '^[a-z]{3}$'), upper(splitByChar('/', lower(meshcore_packets.topic))[2]), '') AS region
+FROM meshcore_packets
+WHERE payload_type = 5;
+-- +goose StatementEnd
+
+-- One-time backfill of existing history (expressions identical to the MV).
+-- +goose StatementBegin
+INSERT INTO meshcore_public_channel_messages_raw
+SELECT
+    ingest_timestamp,
+    mesh_timestamp,
+    hex(substring(payload, 1, 1)) AS channel_hash,
+    hex(substring(payload, 2, 2)) AS mac,
+    substring(payload, 4) AS encrypted_message,
+    packet_hash AS message_id,
+    origin,
+    hex(origin_pubkey) AS origin_pubkey,
+    path,
+    toString(broker) AS broker,
+    toString(topic) AS topic,
+    multiIf(lower(meshcore_packets.topic) IN ('meshcore','meshcore/salish'), 'SEA', match(splitByChar('/', lower(meshcore_packets.topic))[2], '^[a-z]{3}$'), upper(splitByChar('/', lower(meshcore_packets.topic))[2]), '') AS region
+FROM meshcore_packets
+WHERE payload_type = 5;
+-- +goose StatementEnd
+
+-- Replace the public view: dedup by message_id (1:1 with the distinct payload). Exposes exactly the
+-- consumed read contract; the deprecated array columns and `regions` are dropped (verified unused).
+-- The hot streaming/pagination paths use a pushed-down subquery (publicChannelMessagesSubquery) that
+-- filters channel_hash/ingest_timestamp on the (channel_hash, ingest_timestamp) primary key before
+-- grouping; this plain view is the fallback contract and serves api/stats/popular-channels.
+-- +goose StatementBegin
+CREATE OR REPLACE VIEW meshcore_public_channel_messages AS
+SELECT
+    any(channel_hash) AS channel_hash,
+    max(ingest_timestamp) AS ingest_timestamp,
+    min(mesh_timestamp) AS mesh_timestamp,
+    any(mac) AS mac,
+    any(encrypted_message) AS encrypted_message,
+    count() AS message_count,
+    groupArray((origin, origin_pubkey, path, broker, topic)) AS origin_path_info,
+    message_id
+FROM meshcore_public_channel_messages_raw
+GROUP BY message_id
+ORDER BY ingest_timestamp DESC;
+-- +goose StatementEnd
+
+
+-- +goose Down
+-- Restore the plain GROUP-BY-payload VIEW (verbatim from migration 004) before dropping the MV table.
+-- +goose StatementBegin
+CREATE OR REPLACE VIEW meshcore_public_channel_messages AS
+SELECT
+    max(ingest_timestamp) AS ingest_timestamp,
+    min(mesh_timestamp) AS mesh_timestamp,
+    groupArray(origin) AS origins,
+    any(packet) AS packet,
+    any(path_len) AS path_len,
+    hex(substring(payload, 1, 1)) AS channel_hash,
+    hex(substring(payload, 2, 2)) AS mac,
+    substring(payload, 4) AS encrypted_message,
+    count() AS message_count,
+    groupArray((origin, hex(path))) AS origin_path_array, --deprecated
+    groupArray((origin, hex(origin_pubkey), hex(path))) AS origin_key_path_array, --deprecated
+    groupArray((broker, topic)) AS topic_broker_array, --deprecated
+    groupArray((origin, hex(origin_pubkey), hex(path), broker, topic)) AS origin_path_info,
+    arrayDistinct(arrayFilter(r -> r != '', groupArray(multiIf(lower(topic) IN ('meshcore','meshcore/salish'), 'SEA', match(splitByChar('/', lower(topic))[2], '^[a-z]{3}$'), upper(splitByChar('/', lower(topic))[2]), '')))) AS regions,
+    any(packet_hash) AS message_id
+FROM meshcore_packets
+WHERE payload_type = 5
+GROUP BY payload
+ORDER BY ingest_timestamp DESC;
+-- +goose StatementEnd
+DROP VIEW IF EXISTS meshcore_public_channel_messages_mv;
+DROP TABLE IF EXISTS meshcore_public_channel_messages_raw;

meshexplorer/src/lib/clickhouse/actions.ts

@@ -67,20 +67,20 @@ export async function getLatestChatMessages({ limit = 20, before, after, channel
     const outerWhere: string[] = [];
     const params: Record<string, any> = { limit };
 
-    // ingest_timestamp must be table-qualified: unqualified, the analyzer binds
-    // it to the `max(ingest_timestamp) AS ingest_timestamp` output alias and
-    // rejects the WHERE (ILLEGAL_AGGREGATION).
+    // Inner-scan columns must be table-qualified: unqualified, the analyzer binds
+    // them to the aggregate output aliases (max(ingest_timestamp) etc.) and rejects
+    // the WHERE (ILLEGAL_AGGREGATION). Qualified, they push onto the decoded table's
+    // (channel_hash, ingest_timestamp) primary key before the GROUP BY message_id.
     if (before) {
-      innerWhere.push('meshcore_packets.ingest_timestamp < {before:DateTime64}');
+      innerWhere.push('meshcore_public_channel_messages_raw.ingest_timestamp < {before:DateTime64}');
       params.before = before;
     }
     if (after) {
-      innerWhere.push('meshcore_packets.ingest_timestamp > {after:DateTime64}');
+      innerWhere.push('meshcore_public_channel_messages_raw.ingest_timestamp > {after:DateTime64}');
       params.after = after;
     }
     if (channelId) {
-      // channel_hash == hex(substring(payload, 1, 1)); push to the raw scan.
-      innerWhere.push('hex(substring(payload, 1, 1)) = {channelId:String}');
+      innerWhere.push('meshcore_public_channel_messages_raw.channel_hash = {channelId:String}');
       params.channelId = channelId;
     }
 
@@ -524,8 +524,12 @@ export async function searchMeshcoreNodes(searchParams: SearchQuery | SearchQuer
       
       const whereClause = where.length > 0 ? `WHERE ${where.join(' AND ')}` : '';
       
+      // Reads the live, state-backed meshcore_adverts_latest view (incremental MV over
+      // meshcore_adverts_latest_state) instead of re-aggregating meshcore_adverts on every
+      // request. All filter columns (public_key, node_name, last_seen, region, is_repeater)
+      // exist on the view, so the WHERE pushes straight onto it.
       const queryPart = `
-        SELECT 
+        SELECT
           public_key,
           node_name,
           latitude,
@@ -540,27 +544,9 @@ export async function searchMeshcoreNodes(searchParams: SearchQuery | SearchQuer
           broker,
           topic,
           ${index} as query_index
-        FROM (
-          SELECT 
-            public_key,
-            argMax(node_name, ingest_timestamp) as node_name,
-            argMax(latitude, ingest_timestamp) as latitude,
-            argMax(longitude, ingest_timestamp) as longitude,
-            argMax(has_location, ingest_timestamp) as has_location,
-            argMax(is_repeater, ingest_timestamp) as is_repeater,
-            argMax(is_chat_node, ingest_timestamp) as is_chat_node,
-            argMax(is_room_server, ingest_timestamp) as is_room_server,
-            argMax(has_name, ingest_timestamp) as has_name,
-            min(ingest_timestamp) as first_heard,
-            max(ingest_timestamp) as last_seen,
-            argMax(broker, ingest_timestamp) as broker,
-            argMax(topic, ingest_timestamp) as topic,
-            argMax(region, ingest_timestamp) as region
-          FROM meshcore_adverts
-          GROUP BY public_key
-        ) 
-        ${whereClause} 
-        ORDER BY last_seen DESC 
+        FROM meshcore_adverts_latest
+        ${whereClause}
+        ORDER BY last_seen DESC
         LIMIT {limit_${index}:UInt32}
       `;
       

meshexplorer/src/lib/clickhouse/chatMessages.ts

@@ -1,40 +1,39 @@
 /**
  * Builds the public-channel-messages aggregation as an inline subquery.
  *
- * This mirrors the `meshcore_public_channel_messages` view (group meshcore
- * packets by payload to dedup the same message seen via multiple gateways), but
- * lets callers push filters into the inner `meshcore_packets` scan instead of
- * filtering the fully-aggregated view output.
+ * Reads the live, pre-decoded `meshcore_public_channel_messages_raw` materialized
+ * view (one row per gateway-copy of a payload_type=5 packet) and dedups the same
+ * message seen via multiple gateways with `GROUP BY message_id`. Callers push
+ * filters into the inner scan instead of filtering the fully-aggregated output.
  *
- * Why this matters: in the view, `ingest_timestamp` is `max(ingest_timestamp)`
- * and `channel_hash` is derived from the grouped `payload`. A WHERE on those
- * columns can't be pushed below the GROUP BY, so ClickHouse must aggregate the
- * entire payload_type=5 history (millions of rows) on every query before the
- * filter applies — the timestamp primary key never gets used. Pushing the time
- * and channel filters into `innerConditions` lets partition + primary-key
- * pruning (ORDER BY starts with ingest_timestamp) kick in, turning a ~1 GiB
- * full scan into a few-millisecond ranged read.
+ * Why this matters: in the output, `ingest_timestamp` is `max(ingest_timestamp)`
+ * and `channel_hash`/`mac`/`encrypted_message` are `any(...)` over the message_id
+ * group. A WHERE on those columns can't be pushed below the GROUP BY. Pushing the
+ * time and channel filters into `innerConditions` lets partition + primary-key
+ * pruning kick in — the table's ORDER BY is `(channel_hash, ingest_timestamp)`, so
+ * a per-channel timestamp range is a few-millisecond ranged read instead of a full
+ * scan + merge.
  *
- * @param innerConditions Extra predicates applied to the meshcore_packets scan,
- *   before grouping. `payload_type = 5` is always included. Reference
- *   `ingest_timestamp` as `meshcore_packets.ingest_timestamp` — unqualified it
- *   binds to the `max(ingest_timestamp)` output alias and the query is rejected
- *   with ILLEGAL_AGGREGATION.
+ * @param innerConditions Extra predicates applied to the
+ *   meshcore_public_channel_messages_raw scan, before grouping. Reference columns
+ *   table-qualified (e.g. `meshcore_public_channel_messages_raw.ingest_timestamp`,
+ *   `meshcore_public_channel_messages_raw.channel_hash`) — unqualified they bind to
+ *   the aggregate output aliases and the query is rejected with ILLEGAL_AGGREGATION.
  */
 export function publicChannelMessagesSubquery(innerConditions: string[] = []): string {
-  const where = ["payload_type = 5", ...innerConditions].join(" AND ");
+  const where = innerConditions.length > 0 ? `WHERE ${innerConditions.join(" AND ")}` : "";
   return `(
     SELECT
+      any(channel_hash) AS channel_hash,
       max(ingest_timestamp) AS ingest_timestamp,
       min(mesh_timestamp) AS mesh_timestamp,
-      hex(substring(payload, 1, 1)) AS channel_hash,
-      hex(substring(payload, 2, 2)) AS mac,
-      substring(payload, 4) AS encrypted_message,
+      any(mac) AS mac,
+      any(encrypted_message) AS encrypted_message,
       count() AS message_count,
-      groupArray((origin, hex(origin_pubkey), hex(path), broker, topic)) AS origin_path_info,
-      any(packet_hash) AS message_id
-    FROM meshcore_packets
-    WHERE ${where}
-    GROUP BY payload
+      groupArray((origin, origin_pubkey, path, broker, topic)) AS origin_path_info,
+      message_id
+    FROM meshcore_public_channel_messages_raw
+    ${where}
+    GROUP BY message_id
   )`;
 }

meshexplorer/src/lib/clickhouse/streaming.ts

@@ -244,25 +244,26 @@ export function createChatMessagesStreamerConfig(
   channelId?: string,
   region?: string
 ): StreamingConfig {
-  let additionalWhereClause = '';
-  
+  // Push the poll cursor — and the channel filter when scoped to one channel — into the inner
+  // meshcore_public_channel_messages_raw scan so the (channel_hash, ingest_timestamp) primary key
+  // limits it to the last few seconds of packets, instead of re-grouping the whole payload_type=5
+  // history every 250ms. Inner columns are table-qualified so they bind to the scan, not the
+  // aggregate output aliases.
+  const innerConditions = ['meshcore_public_channel_messages_raw.ingest_timestamp > {lastTimestamp:DateTime64}'];
   if (channelId) {
-    additionalWhereClause = `channel_hash = {channelId:String}`;
+    innerConditions.push('meshcore_public_channel_messages_raw.channel_hash = {channelId:String}');
   }
 
+  // Region filtering keys off origin_path_info, which only exists post-group, so it stays as an
+  // outer predicate spliced in by the streamer.
+  let additionalWhereClause = '';
   if (region) {
-    // Add region filtering for chat messages using origin_path_info
     const regionClause = generateRegionArrayConditionForStreaming(region);
     if (regionClause) {
-      additionalWhereClause += (additionalWhereClause ? ' AND ' : '') + regionClause;
+      additionalWhereClause = regionClause;
     }
   }
 
-  // Push the poll cursor into the inner meshcore_packets scan so partition /
-  // primary-key pruning limits it to the last few seconds of packets, instead
-  // of re-aggregating the entire payload_type=5 history every 250ms. The outer
-  // WHERE keeps the same predicate so the streamer can still splice in
-  // channel_hash / region filters (origin_path_info only exists post-group).
   return {
     queryTemplate: `
       SELECT
@@ -274,7 +275,7 @@ export function createChatMessagesStreamerConfig(
         message_count,
         origin_path_info,
         message_id
-      FROM ${publicChannelMessagesSubquery(['meshcore_packets.ingest_timestamp > {lastTimestamp:DateTime64}'])}
+      FROM ${publicChannelMessagesSubquery(innerConditions)}
       WHERE ingest_timestamp > {lastTimestamp:DateTime64}
       ORDER BY ingest_timestamp DESC
       LIMIT {maxRows:UInt32}