From c689ca1b3d178947e5afa58e880862e25f28941a Mon Sep 17 00:00:00 2001 From: Alex Vanderpot Date: Mon, 15 Jun 2026 00:13:00 -0400 Subject: [PATCH] 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) --- .../migrations/005_live_adverts_latest_mv.sql | 214 ++++++++++++++++++ .../006_live_public_channel_messages_mv.sql | 130 +++++++++++ meshexplorer/src/lib/clickhouse/actions.ts | 44 ++-- .../src/lib/clickhouse/chatMessages.ts | 51 ++--- meshexplorer/src/lib/clickhouse/streaming.ts | 23 +- 5 files changed, 396 insertions(+), 66 deletions(-) create mode 100644 ingest/migrations/005_live_adverts_latest_mv.sql create mode 100644 ingest/migrations/006_live_public_channel_messages_mv.sql diff --git a/ingest/migrations/005_live_adverts_latest_mv.sql b/ingest/migrations/005_live_adverts_latest_mv.sql new file mode 100644 index 0000000..ad7bbbd --- /dev/null +++ b/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; diff --git a/ingest/migrations/006_live_public_channel_messages_mv.sql b/ingest/migrations/006_live_public_channel_messages_mv.sql new file mode 100644 index 0000000..8e1340f --- /dev/null +++ b/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; diff --git a/meshexplorer/src/lib/clickhouse/actions.ts b/meshexplorer/src/lib/clickhouse/actions.ts index 893de49..4b77f99 100644 --- a/meshexplorer/src/lib/clickhouse/actions.ts +++ b/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 = { 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} `; diff --git a/meshexplorer/src/lib/clickhouse/chatMessages.ts b/meshexplorer/src/lib/clickhouse/chatMessages.ts index 0b4b5bb..387152d 100644 --- a/meshexplorer/src/lib/clickhouse/chatMessages.ts +++ b/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 )`; } diff --git a/meshexplorer/src/lib/clickhouse/streaming.ts b/meshexplorer/src/lib/clickhouse/streaming.ts index db1d340..b2d0026 100644 --- a/meshexplorer/src/lib/clickhouse/streaming.ts +++ b/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}