fix(interchain-indexer): Fix zero-count expansion for filtered message paths (#1663)

* Tune message-paths logic

* Fix codereview issues

Author: EvgenKor

interchain-indexer/.memory-bank/research/stats-subsystem.md

@@ -413,7 +413,10 @@ Zero-chain visibility is service-wide and configurable:
   - message-path endpoints drive the query from `chains` (excluding the
     selected chain) and left-join `stats_messages` / aggregated
     `stats_messages_days`
-  - explicit `counterparty_chain_ids` still suppresses synthesized zero rows
+  - with explicit `counterparty_chain_ids`, message-path endpoints still drive
+    from `chains`, restrict rows to the requested counterparties, exclude the
+    selected chain itself, and return zero-valued rows for requested
+    counterparties that exist in `chains` but have no aggregate row
 - when `false`: both families return only rows with positive aggregated stats
   - `/stats/chains` filters
     `COALESCE(sc.unique_transfer_users_count, 0) > 0 OR COALESCE(sc.unique_message_users_count, 0) > 0)`

interchain-indexer/README.md

@@ -104,7 +104,7 @@ Defines which bridges (cross-chain mechanisms) to index. Each entry is one bridg
 | `INTERCHAIN_INDEXER__BUFFER_SETTINGS__MAINTENANCE_INTERVAL`             |                          |                                                                                                                                  | `500`         |
 | `INTERCHAIN_INDEXER__STATS__BACKFILL_ON_START`                          |                          | Recalculate the statistics tables for messages and transfers (`stats_messages`, `stats_asset*`) on service startup. This is needed only after the first application of the `m20260312_175120_add_stats_tables` migration, and only if there are existing DB records before it. This option should normally be disabled after the migration to reduce service startup time. | `false`       |
 | `INTERCHAIN_INDEXER__STATS__CHAINS_RECALCULATION_PERIOD_SECS`           |                          | Interval in seconds between full recomputations of per-chain distinct user counters in `stats_chains` (from `crosschain_messages` / `crosschain_transfers`, any status). Only chains with at least one counted user address keep a row; stale rows are deleted. Set to `0` to disable the background task. | `3600`        |
-| `INTERCHAIN_INDEXER__STATS__INCLUDE_ZERO_CHAINS`                        |                          | When `true`, stats endpoints (`/api/v1/stats/chains` and `/api/v1/stats/chain/{chain_id}/messages-paths/*`) include known chains from `chains` even when the aggregated stats row is missing or has a zero value. Disable it to return only chains with positive aggregated stats. | `true`        |
+| `INTERCHAIN_INDEXER__STATS__INCLUDE_ZERO_CHAINS`                        |                          | When `true`, stats endpoints (`/api/v1/stats/chains` and `/api/v1/stats/chain/{chain_id}/messages-paths/*`) include known chains from `chains` even when the aggregated stats row is missing or has a zero value. For message paths with `counterparty_chain_ids`, zero rows are still returned for the explicitly requested counterparties that exist in `chains`, and no other counterparties are added. Disable it to return only chains with positive aggregated stats. | `true`        |
 
 [anchor]: <> (anchors.envs.end.service)
 

interchain-indexer/interchain-indexer-logic/src/database.rs

@@ -173,25 +173,29 @@ enum MessagePathDirection {
     Incoming,
 }
 
-fn should_expand_message_paths(
-    include_zero_chains: bool,
-    counterparty_chain_ids: Option<&[i64]>,
-) -> bool {
-    include_zero_chains
-        && counterparty_chain_ids
-            .map(|chain_ids| chain_ids.is_empty())
-            .unwrap_or(true)
-}
-
 fn build_all_time_message_paths_query(
     chain_id: i64,
     direction: MessagePathDirection,
     counterparty_chain_ids: Option<&[i64]>,
     include_zero_chains: bool,
 ) -> (String, Vec<Value>) {
-    if should_expand_message_paths(include_zero_chains, counterparty_chain_ids) {
+    if include_zero_chains {
+        let mut values = vec![Value::BigInt(Some(chain_id))];
+        let mut where_parts = vec![
+            "c.id <> $1".to_string(),
+            "EXISTS (SELECT 1 FROM chains WHERE id = $1)".to_string(),
+        ];
+
+        if let Some(ids) = counterparty_chain_ids.filter(|chain_ids| !chain_ids.is_empty()) {
+            let placeholders: Vec<String> = (0..ids.len()).map(|i| format!("${}", i + 2)).collect();
+            where_parts.push(format!("c.id IN ({})", placeholders.join(", ")));
+            for &id in ids {
+                values.push(Value::BigInt(Some(id)));
+            }
+        }
+
         let sql = match direction {
-            MessagePathDirection::Outgoing => {
+            MessagePathDirection::Outgoing => format!(
                 r#"
 SELECT $1::bigint AS src_chain_id,
        c.id AS dst_chain_id,
@@ -200,12 +204,12 @@ FROM chains c
 LEFT JOIN stats_messages sm
     ON sm.src_chain_id = $1
    AND sm.dst_chain_id = c.id
-WHERE c.id <> $1
-  AND EXISTS (SELECT 1 FROM chains WHERE id = $1)
+WHERE {}
 ORDER BY messages_count DESC, src_chain_id ASC, dst_chain_id ASC
-"#
-            }
-            MessagePathDirection::Incoming => {
+"#,
+                where_parts.join("\n  AND ")
+            ),
+            MessagePathDirection::Incoming => format!(
                 r#"
 SELECT c.id AS src_chain_id,
        $1::bigint AS dst_chain_id,
@@ -214,14 +218,14 @@ FROM chains c
 LEFT JOIN stats_messages sm
     ON sm.src_chain_id = c.id
    AND sm.dst_chain_id = $1
-WHERE c.id <> $1
-  AND EXISTS (SELECT 1 FROM chains WHERE id = $1)
+WHERE {}
 ORDER BY messages_count DESC, src_chain_id ASC, dst_chain_id ASC
-"#
-            }
+"#,
+                where_parts.join("\n  AND ")
+            ),
         };
 
-        return (sql.to_string(), vec![Value::BigInt(Some(chain_id))]);
+        return (sql, values);
     }
 
     let filter_column = match direction {
@@ -265,7 +269,7 @@ fn build_bounded_message_paths_query(
     counterparty_chain_ids: Option<&[i64]>,
     include_zero_chains: bool,
 ) -> (String, Vec<Value>) {
-    if should_expand_message_paths(include_zero_chains, counterparty_chain_ids) {
+    if include_zero_chains {
         let mut aggregate_where_parts = vec![match direction {
             MessagePathDirection::Outgoing => "src_chain_id = $1".to_string(),
             MessagePathDirection::Incoming => "dst_chain_id = $1".to_string(),
@@ -282,6 +286,28 @@ fn build_bounded_message_paths_query(
         if let Some(to_date) = to_date {
             aggregate_where_parts.push(format!("date < ${placeholder}"));
             values.push(Value::ChronoDate(Some(Box::new(to_date))));
+            placeholder += 1;
+        }
+
+        let mut where_parts = vec![
+            "c.id <> $1".to_string(),
+            "EXISTS (SELECT 1 FROM chains WHERE id = $1)".to_string(),
+        ];
+
+        if let Some(ids) = counterparty_chain_ids.filter(|chain_ids| !chain_ids.is_empty()) {
+            let placeholders: Vec<String> = (0..ids.len())
+                .map(|i| format!("${}", placeholder + i))
+                .collect();
+            let placeholders = placeholders.join(", ");
+            let counterparty_column = match direction {
+                MessagePathDirection::Outgoing => "dst_chain_id",
+                MessagePathDirection::Incoming => "src_chain_id",
+            };
+            aggregate_where_parts.push(format!("{counterparty_column} IN ({placeholders})"));
+            where_parts.push(format!("c.id IN ({placeholders})"));
+            for &id in ids {
+                values.push(Value::BigInt(Some(id)));
+            }
         }
 
         let sql = match direction {
@@ -298,11 +324,11 @@ LEFT JOIN (
     WHERE {}
     GROUP BY dst_chain_id
 ) sm ON sm.dst_chain_id = c.id
-WHERE c.id <> $1
-  AND EXISTS (SELECT 1 FROM chains WHERE id = $1)
+WHERE {}
 ORDER BY messages_count DESC, src_chain_id ASC, dst_chain_id ASC
 "#,
-                aggregate_where_parts.join(" AND ")
+                aggregate_where_parts.join(" AND "),
+                where_parts.join("\n  AND ")
             ),
             MessagePathDirection::Incoming => format!(
                 r#"
@@ -317,11 +343,11 @@ LEFT JOIN (
     WHERE {}
     GROUP BY src_chain_id
 ) sm ON sm.src_chain_id = c.id
-WHERE c.id <> $1
-  AND EXISTS (SELECT 1 FROM chains WHERE id = $1)
+WHERE {}
 ORDER BY messages_count DESC, src_chain_id ASC, dst_chain_id ASC
 "#,
-                aggregate_where_parts.join(" AND ")
+                aggregate_where_parts.join(" AND "),
+                where_parts.join("\n  AND ")
             ),
         };
 
@@ -6160,8 +6186,8 @@ mod tests {
 
     #[tokio::test]
     #[ignore = "needs database to run"]
-    async fn message_paths_include_zero_counterparty_does_not_synthesize_missing_rows() {
-        let _db = init_db("message_paths_include_zero_counterparty_missing").await;
+    async fn message_paths_include_zero_counterparty_expands_requested_known_rows_only() {
+        let _db = init_db("message_paths_include_zero_counterparty_expand").await;
         let interchain_db = InterchainDatabase::new(_db.client());
         interchain_db
             .upsert_chains(vec![
@@ -6180,19 +6206,43 @@ mod tests {
                     name: Set("C".into()),
                     ..Default::default()
                 },
+                chains::ActiveModel {
+                    id: Set(4),
+                    name: Set("D".into()),
+                    ..Default::default()
+                },
             ])
             .await
             .unwrap();
         interchain_db
             .create_or_update_stats_messages(1, 2, 5)
             .await
             .unwrap();
+        interchain_db
+            .create_or_update_stats_messages(1, 4, 7)
+            .await
+            .unwrap();
 
         let rows = interchain_db
-            .get_outgoing_message_paths(1, None, None, Some(&[3]), true)
+            .get_outgoing_message_paths(1, None, None, Some(&[1, 3, 4, 999]), true)
             .await
             .unwrap();
-        assert!(rows.is_empty());
+        assert_eq!(
+            rows,
+            vec![
+                MessagePathStatsRow {
+                    src_chain_id: 1,
+                    dst_chain_id: 4,
+                    messages_count: 7
+                },
+                MessagePathStatsRow {
+                    src_chain_id: 1,
+                    dst_chain_id: 3,
+                    messages_count: 0
+                },
+            ]
+        );
+        assert!(rows.iter().all(|row| row.dst_chain_id != 1));
     }
 
     #[tokio::test]
@@ -6245,8 +6295,8 @@ mod tests {
 
     #[tokio::test]
     #[ignore = "needs database to run"]
-    async fn message_paths_bounded_counterparty_filter_applies_in_sql() {
-        let _db = init_db("message_paths_bounded_counterparty").await;
+    async fn message_paths_include_zero_incoming_counterparty_expands_requested_known_rows_only() {
+        let _db = init_db("message_paths_include_zero_incoming_counterparty_expand").await;
         let interchain_db = InterchainDatabase::new(_db.client());
         interchain_db
             .upsert_chains(vec![
@@ -6265,13 +6315,79 @@ mod tests {
                     name: Set("C".into()),
                     ..Default::default()
                 },
+                chains::ActiveModel {
+                    id: Set(4),
+                    name: Set("D".into()),
+                    ..Default::default()
+                },
+            ])
+            .await
+            .unwrap();
+        interchain_db
+            .create_or_update_stats_messages(1, 3, 4)
+            .await
+            .unwrap();
+        interchain_db
+            .create_or_update_stats_messages(4, 3, 8)
+            .await
+            .unwrap();
+
+        let rows = interchain_db
+            .get_incoming_message_paths(3, None, None, Some(&[2, 3, 4, 999]), true)
+            .await
+            .unwrap();
+        assert_eq!(
+            rows,
+            vec![
+                MessagePathStatsRow {
+                    src_chain_id: 4,
+                    dst_chain_id: 3,
+                    messages_count: 8
+                },
+                MessagePathStatsRow {
+                    src_chain_id: 2,
+                    dst_chain_id: 3,
+                    messages_count: 0
+                },
+            ]
+        );
+        assert!(rows.iter().all(|row| row.src_chain_id != 3));
+    }
+
+    #[tokio::test]
+    #[ignore = "needs database to run"]
+    async fn message_paths_include_zero_bounded_counterparty_expands_requested_known_rows_only() {
+        let _db = init_db("message_paths_include_zero_bounded_counterparty").await;
+        let interchain_db = InterchainDatabase::new(_db.client());
+        interchain_db
+            .upsert_chains(vec![
+                chains::ActiveModel {
+                    id: Set(1),
+                    name: Set("A".into()),
+                    ..Default::default()
+                },
+                chains::ActiveModel {
+                    id: Set(2),
+                    name: Set("B".into()),
+                    ..Default::default()
+                },
+                chains::ActiveModel {
+                    id: Set(3),
+                    name: Set("C".into()),
+                    ..Default::default()
+                },
+                chains::ActiveModel {
+                    id: Set(4),
+                    name: Set("D".into()),
+                    ..Default::default()
+                },
             ])
             .await
             .unwrap();
 
         for (date, src, dst, count) in [
             (NaiveDate::from_ymd_opt(2026, 3, 8).unwrap(), 1, 2, 2),
-            (NaiveDate::from_ymd_opt(2026, 3, 8).unwrap(), 1, 3, 7),
+            (NaiveDate::from_ymd_opt(2026, 3, 8).unwrap(), 1, 4, 7),
         ] {
             stats_messages_days::Entity::insert(stats_messages_days::ActiveModel {
                 date: Set(date),
@@ -6290,19 +6406,27 @@ mod tests {
                 1,
                 Some(NaiveDate::from_ymd_opt(2026, 3, 8).unwrap()),
                 Some(NaiveDate::from_ymd_opt(2026, 3, 9).unwrap()),
-                Some(&[2]),
+                Some(&[1, 2, 3, 999]),
                 true,
             )
             .await
             .unwrap();
         assert_eq!(
             rows,
-            vec![MessagePathStatsRow {
-                src_chain_id: 1,
-                dst_chain_id: 2,
-                messages_count: 2
-            }]
+            vec![
+                MessagePathStatsRow {
+                    src_chain_id: 1,
+                    dst_chain_id: 2,
+                    messages_count: 2
+                },
+                MessagePathStatsRow {
+                    src_chain_id: 1,
+                    dst_chain_id: 3,
+                    messages_count: 0
+                },
+            ]
         );
+        assert!(rows.iter().all(|row| row.dst_chain_id != 1));
     }
 
     #[tokio::test]