From 2fa34c0459e77ea21eb5a8c552aa91353ab25f66 Mon Sep 17 00:00:00 2001 From: Roman Mitasov Date: Wed, 24 Jun 2026 11:50:23 +0300 Subject: [PATCH 1/4] feat(collab): surface Zulip chat in the Collaboration bullet end-to-end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Zulip chat data reached `silver.class_collab_chat_activity` but stopped there: nothing carried it through gold → metric → API → UI. Three gaps, all fixed, plus an e2e test that drives the whole path. 1. Silver class. `class_collab_chat_activity` only declared a `-- depends_on` on the m365/slack chat models, so a `tag:zulip-proxy+` run never rebuilt it. Added the zulip ref. (`union_by_tag` already includes any tagged model whose table exists; this just wires the build-order edge so prod's connector-scoped run refreshes the class.) 2. Gold view. `insight.collab_bullet_rows` only branched on `insight_m365` / `insight_slack`. Added Branch 4b for `data_source = 'insight_zulip_proxy'`, emitting the single counter `zulip_messages_sent` (mirrors the m365 Teams branch). 3. Metric query_ref + catalog. The IC (…0012) / Team (…0005) collab bullet `query_ref`s enumerate every FE-visible `metric_key` via `sumIf` + `ARRAY JOIN`. New append-only migration `m20260624_000001_collab_zulip_chat` re-sets both with the zulip key (based on the latest `m20260604_000002` distribution shape); `m20260624_000002_seed_zulip_collab_catalog` adds the catalog row (`collab_bullet_rows.zulip_messages_sent`, "Zulip Messages", source_tags ["zulip-proxy"]). Both registered in migration/mod.rs. E2E: new `specs/collab_zulip_chat.test.yaml` (+ schemas/templates and zulip bronze placeholders) seeds alice/bob/carol chat (40/20/10, one Engineering team) with a duplicate `uniq` row and asserts the IC bullet returns `zulip_messages_sent` value=40, median=20, range[10,40] (dedup holds). Adding the key raises the collab bullet from 20 to 21 FE keys, so `collab_emails_sent.test.yaml`'s `size(items)` is bumped 20 -> 21. Full e2e suite: 33 passed. Scope: the secondary collab `query_ref`s — member heatmap (…0041) and department distribution (…0045) — carry their own copies of the key list and are intentionally left for a follow-up; the IC/Team bullet that backs the person-profile Collaboration card (and this test) is fully wired. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Roman Mitasov --- .../m20260624_000001_collab_zulip_chat.rs | 345 ++++++++++++++++++ ...260624_000002_seed_zulip_collab_catalog.rs | 164 +++++++++ .../analytics-api/src/migration/mod.rs | 4 + .../scripts/create-bronze-placeholders.sh | 49 +++ .../20260518000000_collab-bullet-rewrite.sql | 21 +- .../class_collab_chat_activity.sql | 1 + .../e2e/specs/collab_emails_sent.test.yaml | 2 +- .../e2e/specs/collab_zulip_chat.test.yaml | 80 ++++ .../schemas/bronze_zulip_proxy.messages.yaml | 20 + .../schemas/bronze_zulip_proxy.users.yaml | 23 ++ .../tests/e2e/specs/templates/zulip.yaml | 78 ++++ 11 files changed, 785 insertions(+), 2 deletions(-) create mode 100644 src/backend/services/analytics-api/src/migration/m20260624_000001_collab_zulip_chat.rs create mode 100644 src/backend/services/analytics-api/src/migration/m20260624_000002_seed_zulip_collab_catalog.rs create mode 100644 src/ingestion/tests/e2e/specs/collab_zulip_chat.test.yaml create mode 100644 src/ingestion/tests/e2e/specs/schemas/bronze_zulip_proxy.messages.yaml create mode 100644 src/ingestion/tests/e2e/specs/schemas/bronze_zulip_proxy.users.yaml create mode 100644 src/ingestion/tests/e2e/specs/templates/zulip.yaml diff --git a/src/backend/services/analytics-api/src/migration/m20260624_000001_collab_zulip_chat.rs b/src/backend/services/analytics-api/src/migration/m20260624_000001_collab_zulip_chat.rs new file mode 100644 index 000000000..c0459d2ab --- /dev/null +++ b/src/backend/services/analytics-api/src/migration/m20260624_000001_collab_zulip_chat.rs @@ -0,0 +1,345 @@ +//! Add the Zulip chat counter (`zulip_messages_sent`) to the Collaboration +//! bullet `query_ref`s (team `…0005`, IC `…0012`). +//! +//! Pairs with the ingestion-side gold view branch added to +//! `insight.collab_bullet_rows` (`20260518000000_collab-bullet-rewrite.sql`, +//! Branch 4b) which emits `zulip_messages_sent` for +//! `data_source = 'insight_zulip_proxy'` rows of +//! `silver.class_collab_chat_activity`, and the catalog row added in +//! `m20260624_000002_seed_zulip_collab_catalog`. +//! +//! Why a new migration rather than editing `m20260604_000002`: SeaORM +//! migrations are append-only (already applied in prod). This re-sets the two +//! collab bullet `query_ref`s to the `m20260604_000002` distribution shape +//! (value + median/range + p25/p75/n) with one extra raw counter +//! (`zulip_messages_sent`) wired through `wide_aggregate_pp` (`sumIf`) and the +//! `ARRAY JOIN` unpivot. All other keys are copied verbatim. `down()` restores +//! the `m20260604_000002` shape (without the Zulip key). +//! +//! Scope note: the per-person member-values (`…0041`) and per-department +//! distribution (`…0045`) collab `query_ref`s carry their own copies of the +//! same key list; if the Zulip card must also appear in the team heatmap / +//! department-distribution widgets, those need the same one-line additions in +//! follow-up migrations. The IC/Team bullet (this migration) is what backs the +//! person-profile Collaboration section card. + +use sea_orm_migration::prelude::*; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +const TEAM_BULLET_COLLAB_ID: &str = "00000000000000000001000000000005"; +const IC_BULLET_COLLAB_ID: &str = "00000000000000000001000000000012"; + +/// Inner wide-aggregate block — `m20260604_000002` verbatim PLUS the +/// `zulip_messages_sent` raw counter. One row per `person_id`, every +/// FE-visible `metric_key` in its own `_v` column. +fn wide_aggregate_pp() -> &'static str { + "SELECT person_id, any(org_unit_id) AS org_unit_id, \ + sumIf(metric_value, metric_key = 'm365_emails_sent') AS m365_emails_sent_v, \ + sumIf(metric_value, metric_key = 'm365_emails_received') AS m365_emails_received_v, \ + sumIf(metric_value, metric_key = 'm365_emails_read') AS m365_emails_read_v, \ + sumIf(metric_value, metric_key = 'meeting_hours') AS meeting_hours_v, \ + sumIf(metric_value, metric_key = 'meetings_count') AS meetings_count_v, \ + sumIf(metric_value, metric_key = 'teams_meeting_hours') AS teams_meeting_hours_v, \ + sumIf(metric_value, metric_key = 'zoom_meeting_hours') AS zoom_meeting_hours_v, \ + sumIf(metric_value, metric_key = 'teams_meetings') AS teams_meetings_v, \ + sumIf(metric_value, metric_key = 'zoom_meetings') AS zoom_meetings_v, \ + sumIf(metric_value, metric_key = 'meeting_free') AS meeting_free_v, \ + sumIf(metric_value, metric_key = 'm365_teams_chats') AS m365_teams_chats_v, \ + sumIf(metric_value, metric_key = 'slack_messages_sent') AS slack_messages_sent_v, \ + sumIf(metric_value, metric_key = 'slack_channel_posts') AS slack_channel_posts_v, \ + sumIf(metric_value, metric_key = 'slack_active_days') AS slack_active_days_v, \ + sumIf(metric_value, metric_key = 'zulip_messages_sent') AS zulip_messages_sent_v, \ + sumIf(metric_value, metric_key = 'm365_files_shared_internal') AS m365_files_shared_internal_v, \ + sumIf(metric_value, metric_key = 'm365_files_shared_external') AS m365_files_shared_external_v, \ + sumIf(metric_value, metric_key = 'm365_files_engaged') AS m365_files_engaged_v, \ + sumIf(metric_value, metric_key = 'm365_active_days') AS m365_active_days_v, \ + if(sumIf(metric_value, metric_key = 'slack_active_days') > 0, \ + round(sumIf(metric_value, metric_key = 'slack_messages_sent') \ + / sumIf(metric_value, metric_key = 'slack_active_days'), 1), \ + CAST(NULL AS Nullable(Float64))) AS slack_msgs_per_active_day_v, \ + if(sumIf(metric_value, metric_key = 'slack_messages_sent') > 0, \ + round(toFloat64(100) \ + * greatest(toFloat64(0), \ + sumIf(metric_value, metric_key = 'slack_messages_sent') \ + - sumIf(metric_value, metric_key = 'slack_channel_posts')) \ + / sumIf(metric_value, metric_key = 'slack_messages_sent'), 1), \ + CAST(NULL AS Nullable(Float64))) AS slack_dm_ratio_v \ + FROM insight.collab_bullet_rows \ + GROUP BY person_id" +} + +/// `ARRAY JOIN` unpivot — `m20260604_000002` verbatim PLUS the +/// `zulip_messages_sent` tuple. +fn array_join_kv() -> &'static str { + "ARRAY JOIN [ \ + ('m365_emails_sent', m365_emails_sent_v), \ + ('m365_emails_received', m365_emails_received_v), \ + ('m365_emails_read', m365_emails_read_v), \ + ('meeting_hours', meeting_hours_v), \ + ('meetings_count', meetings_count_v), \ + ('teams_meeting_hours', teams_meeting_hours_v), \ + ('zoom_meeting_hours', zoom_meeting_hours_v), \ + ('teams_meetings', teams_meetings_v), \ + ('zoom_meetings', zoom_meetings_v), \ + ('meeting_free', meeting_free_v), \ + ('m365_teams_chats', m365_teams_chats_v), \ + ('slack_messages_sent', slack_messages_sent_v), \ + ('slack_channel_posts', slack_channel_posts_v), \ + ('slack_active_days', slack_active_days_v), \ + ('zulip_messages_sent', zulip_messages_sent_v), \ + ('m365_files_shared_internal', m365_files_shared_internal_v), \ + ('m365_files_shared_external', m365_files_shared_external_v), \ + ('m365_files_engaged', m365_files_engaged_v), \ + ('m365_active_days', m365_active_days_v), \ + ('slack_msgs_per_active_day', slack_msgs_per_active_day_v), \ + ('slack_dm_ratio', slack_dm_ratio_v) \ + ] AS kv" +} + +/// Build a collab bullet `query_ref` (distribution shape: value + cohort +/// median/range + p25/p75/n) over a per-person/per-`metric_key` rollup. +/// `cohort_extra_group` is the cohort-side `GROUP BY` tail (`""` for the +/// company-wide team bullet, `", org_unit_id"` + join key for the IC bullet). +fn bullet_query(team_scoped: bool) -> String { + let pp = wide_aggregate_pp(); + let kv = array_join_kv(); + if team_scoped { + // Team bullet …0005: cohort = each member's own department, blended + // headcount-weighted via avg(c.team_*). Verbatim m20260604_000002. + format!( + "SELECT p.metric_key AS metric_key, \ + avg(p.v_period) AS value, \ + avg(c.team_median) AS median, \ + avg(c.team_min) AS range_min, \ + avg(c.team_max) AS range_max, \ + avg(c.team_p25) AS p25, \ + avg(c.team_p75) AS p75, \ + toFloat64(count(p.v_period)) AS n \ + FROM ( \ + SELECT person_id, org_unit_id, \ + kv.1 AS metric_key, kv.2 AS v_period \ + FROM ({pp}) pp \ + {kv} \ + ) p \ + LEFT JOIN ( \ + SELECT metric_key, org_unit_id, \ + quantileExact(0.5)(v_period) AS team_median, \ + min(v_period) AS team_min, \ + max(v_period) AS team_max, \ + quantileExact(0.25)(v_period) AS team_p25, \ + quantileExact(0.75)(v_period) AS team_p75, \ + count(v_period) AS team_n \ + FROM ( \ + SELECT person_id, org_unit_id, \ + kv.1 AS metric_key, kv.2 AS v_period \ + FROM ({pp}) ppc \ + {kv} \ + ) inner_c \ + GROUP BY metric_key, org_unit_id \ + ) c ON c.metric_key = p.metric_key AND c.org_unit_id = p.org_unit_id \ + GROUP BY p.metric_key" + ) + } else { + // IC bullet …0012: cohort = the person's own department. Verbatim + // m20260604_000002. + format!( + "SELECT p.metric_key AS metric_key, \ + avg(p.v_period) AS value, \ + any(c.team_median) AS median, \ + any(c.team_min) AS range_min, \ + any(c.team_max) AS range_max, \ + any(c.team_p25) AS p25, \ + any(c.team_p75) AS p75, \ + any(c.team_n) AS n \ + FROM ( \ + SELECT person_id, org_unit_id, \ + kv.1 AS metric_key, kv.2 AS v_period \ + FROM ({pp}) pp \ + {kv} \ + ) p \ + LEFT JOIN ( \ + SELECT metric_key, org_unit_id, \ + quantileExact(0.5)(v_period) AS team_median, \ + min(v_period) AS team_min, \ + max(v_period) AS team_max, \ + quantileExact(0.25)(v_period) AS team_p25, \ + quantileExact(0.75)(v_period) AS team_p75, \ + count(v_period) AS team_n \ + FROM ( \ + SELECT person_id, org_unit_id, \ + kv.1 AS metric_key, kv.2 AS v_period \ + FROM ({pp}) ppc \ + {kv} \ + ) inner_c \ + GROUP BY metric_key, org_unit_id \ + ) c ON c.metric_key = p.metric_key AND c.org_unit_id = p.org_unit_id \ + GROUP BY p.metric_key" + ) + } +} + +/// Predecessor (`m20260604_000002`) `query_ref`s — same as `bullet_query` +/// but WITHOUT the `zulip_messages_sent` key. Used by `down()`. +fn old_wide_aggregate_pp() -> &'static str { + "SELECT person_id, any(org_unit_id) AS org_unit_id, \ + sumIf(metric_value, metric_key = 'm365_emails_sent') AS m365_emails_sent_v, \ + sumIf(metric_value, metric_key = 'm365_emails_received') AS m365_emails_received_v, \ + sumIf(metric_value, metric_key = 'm365_emails_read') AS m365_emails_read_v, \ + sumIf(metric_value, metric_key = 'meeting_hours') AS meeting_hours_v, \ + sumIf(metric_value, metric_key = 'meetings_count') AS meetings_count_v, \ + sumIf(metric_value, metric_key = 'teams_meeting_hours') AS teams_meeting_hours_v, \ + sumIf(metric_value, metric_key = 'zoom_meeting_hours') AS zoom_meeting_hours_v, \ + sumIf(metric_value, metric_key = 'teams_meetings') AS teams_meetings_v, \ + sumIf(metric_value, metric_key = 'zoom_meetings') AS zoom_meetings_v, \ + sumIf(metric_value, metric_key = 'meeting_free') AS meeting_free_v, \ + sumIf(metric_value, metric_key = 'm365_teams_chats') AS m365_teams_chats_v, \ + sumIf(metric_value, metric_key = 'slack_messages_sent') AS slack_messages_sent_v, \ + sumIf(metric_value, metric_key = 'slack_channel_posts') AS slack_channel_posts_v, \ + sumIf(metric_value, metric_key = 'slack_active_days') AS slack_active_days_v, \ + sumIf(metric_value, metric_key = 'm365_files_shared_internal') AS m365_files_shared_internal_v, \ + sumIf(metric_value, metric_key = 'm365_files_shared_external') AS m365_files_shared_external_v, \ + sumIf(metric_value, metric_key = 'm365_files_engaged') AS m365_files_engaged_v, \ + sumIf(metric_value, metric_key = 'm365_active_days') AS m365_active_days_v, \ + if(sumIf(metric_value, metric_key = 'slack_active_days') > 0, \ + round(sumIf(metric_value, metric_key = 'slack_messages_sent') \ + / sumIf(metric_value, metric_key = 'slack_active_days'), 1), \ + CAST(NULL AS Nullable(Float64))) AS slack_msgs_per_active_day_v, \ + if(sumIf(metric_value, metric_key = 'slack_messages_sent') > 0, \ + round(toFloat64(100) \ + * greatest(toFloat64(0), \ + sumIf(metric_value, metric_key = 'slack_messages_sent') \ + - sumIf(metric_value, metric_key = 'slack_channel_posts')) \ + / sumIf(metric_value, metric_key = 'slack_messages_sent'), 1), \ + CAST(NULL AS Nullable(Float64))) AS slack_dm_ratio_v \ + FROM insight.collab_bullet_rows \ + GROUP BY person_id" +} + +fn old_array_join_kv() -> &'static str { + "ARRAY JOIN [ \ + ('m365_emails_sent', m365_emails_sent_v), \ + ('m365_emails_received', m365_emails_received_v), \ + ('m365_emails_read', m365_emails_read_v), \ + ('meeting_hours', meeting_hours_v), \ + ('meetings_count', meetings_count_v), \ + ('teams_meeting_hours', teams_meeting_hours_v), \ + ('zoom_meeting_hours', zoom_meeting_hours_v), \ + ('teams_meetings', teams_meetings_v), \ + ('zoom_meetings', zoom_meetings_v), \ + ('meeting_free', meeting_free_v), \ + ('m365_teams_chats', m365_teams_chats_v), \ + ('slack_messages_sent', slack_messages_sent_v), \ + ('slack_channel_posts', slack_channel_posts_v), \ + ('slack_active_days', slack_active_days_v), \ + ('m365_files_shared_internal', m365_files_shared_internal_v), \ + ('m365_files_shared_external', m365_files_shared_external_v), \ + ('m365_files_engaged', m365_files_engaged_v), \ + ('m365_active_days', m365_active_days_v), \ + ('slack_msgs_per_active_day', slack_msgs_per_active_day_v), \ + ('slack_dm_ratio', slack_dm_ratio_v) \ + ] AS kv" +} + +fn old_bullet_query(team_scoped: bool) -> String { + let pp = old_wide_aggregate_pp(); + let kv = old_array_join_kv(); + let agg = if team_scoped { + "avg(c.team_median) AS median, avg(c.team_min) AS range_min, \ + avg(c.team_max) AS range_max, avg(c.team_p25) AS p25, \ + avg(c.team_p75) AS p75, toFloat64(count(p.v_period)) AS n" + } else { + "any(c.team_median) AS median, any(c.team_min) AS range_min, \ + any(c.team_max) AS range_max, any(c.team_p25) AS p25, \ + any(c.team_p75) AS p75, any(c.team_n) AS n" + }; + format!( + "SELECT p.metric_key AS metric_key, avg(p.v_period) AS value, {agg} \ + FROM ( \ + SELECT person_id, org_unit_id, kv.1 AS metric_key, kv.2 AS v_period \ + FROM ({pp}) pp {kv} \ + ) p \ + LEFT JOIN ( \ + SELECT metric_key, org_unit_id, \ + quantileExact(0.5)(v_period) AS team_median, \ + min(v_period) AS team_min, max(v_period) AS team_max, \ + quantileExact(0.25)(v_period) AS team_p25, \ + quantileExact(0.75)(v_period) AS team_p75, \ + count(v_period) AS team_n \ + FROM ( \ + SELECT person_id, org_unit_id, kv.1 AS metric_key, kv.2 AS v_period \ + FROM ({pp}) ppc {kv} \ + ) inner_c \ + GROUP BY metric_key, org_unit_id \ + ) c ON c.metric_key = p.metric_key AND c.org_unit_id = p.org_unit_id \ + GROUP BY p.metric_key" + ) +} + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + let db = manager.get_connection(); + for (hex_id, query) in [ + (TEAM_BULLET_COLLAB_ID, bullet_query(true)), + (IC_BULLET_COLLAB_ID, bullet_query(false)), + ] { + db.execute_unprepared(&format!( + "UPDATE metrics SET query_ref = '{qr}' WHERE id = UNHEX('{hex_id}')", + qr = query.replace('\'', "''"), + )) + .await?; + } + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + let db = manager.get_connection(); + for (hex_id, query) in [ + (TEAM_BULLET_COLLAB_ID, old_bullet_query(true)), + (IC_BULLET_COLLAB_ID, old_bullet_query(false)), + ] { + db.execute_unprepared(&format!( + "UPDATE metrics SET query_ref = '{qr}' WHERE id = UNHEX('{hex_id}')", + qr = query.replace('\'', "''"), + )) + .await?; + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn zulip_key_wired_into_pp_and_kv() { + let pp = wide_aggregate_pp(); + assert!( + pp.contains("metric_key = 'zulip_messages_sent') AS zulip_messages_sent_v"), + "wide_aggregate_pp must sumIf the zulip key" + ); + let kv = array_join_kv(); + assert!( + kv.contains("('zulip_messages_sent', zulip_messages_sent_v)"), + "array_join_kv must unpivot the zulip key" + ); + } + + #[test] + fn down_shape_omits_zulip() { + assert!(!old_wide_aggregate_pp().contains("zulip_messages_sent")); + assert!(!old_array_join_kv().contains("zulip_messages_sent")); + } + + #[test] + fn ic_and_team_carry_distribution_columns() { + for q in [bullet_query(true), bullet_query(false)] { + for col in ["p25", "p75", " n ", "median", "range_min", "range_max"] { + assert!(q.contains(col), "bullet query missing {col}"); + } + } + } +} diff --git a/src/backend/services/analytics-api/src/migration/m20260624_000002_seed_zulip_collab_catalog.rs b/src/backend/services/analytics-api/src/migration/m20260624_000002_seed_zulip_collab_catalog.rs new file mode 100644 index 000000000..add30d89d --- /dev/null +++ b/src/backend/services/analytics-api/src/migration/m20260624_000002_seed_zulip_collab_catalog.rs @@ -0,0 +1,164 @@ +//! Seed the `metric_catalog` + product-default `metric_threshold` row for the +//! Zulip chat counter `collab_bullet_rows.zulip_messages_sent` introduced in +//! `m20260624_000001_collab_zulip_chat` (paired gold branch in +//! `20260518000000_collab-bullet-rewrite.sql`, Branch 4b). +//! +//! One `metric_key`, routed through the existing `collab_bullet_rows` section, +//! `source_tags = ["zulip-proxy"]`. Mirrors `m20260620_000002_seed_wiki_catalog` +//! (additive catalog seed). The threshold is a product-default placeholder; +//! tune per tenant via the admin CRUD API. + +use sea_orm::{ConnectionTrait, Statement, Value}; +use sea_orm_migration::prelude::*; +use uuid::Uuid; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +struct SeedRow { + metric_key: &'static str, + label: &'static str, + sublabel: Option<&'static str>, + description: Option<&'static str>, + unit: Option<&'static str>, + format: Option<&'static str>, + higher_is_better: bool, + is_member_scale: bool, + source_tags: &'static [&'static str], + good: f64, + warn: f64, +} + +fn source_tags_json(tags: &[&'static str]) -> String { + let mut out = String::from("["); + for (i, tag) in tags.iter().enumerate() { + if i > 0 { + out.push(','); + } + out.push('"'); + for c in tag.chars() { + match c { + '"' => out.push_str("\\\""), + '\\' => out.push_str("\\\\"), + _ => out.push(c), + } + } + out.push('"'); + } + out.push(']'); + out +} + +const SEEDS: &[SeedRow] = &[SeedRow { + metric_key: "collab_bullet_rows.zulip_messages_sent", + label: "Zulip Messages", + sublabel: Some("Zulip \u{b7} chat messages sent \u{b7} period total"), + description: Some("Chat messages sent by the person in the period (Zulip)."), + unit: Some("messages"), + format: None, + higher_is_better: true, + is_member_scale: false, + source_tags: &["zulip-proxy"], + good: 50.0, + warn: 20.0, +}]; + +const INSERT_CATALOG_SQL: &str = "\ + INSERT INTO metric_catalog \ + (id, tenant_id, metric_key, label, sublabel, description, unit, format, \ + higher_is_better, is_member_scale, source_tags, is_enabled) \ + VALUES (?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, TRUE) \ + ON DUPLICATE KEY UPDATE \ + label = VALUES(label), \ + sublabel = VALUES(sublabel), \ + description = VALUES(description), \ + unit = VALUES(unit), \ + format = VALUES(format), \ + higher_is_better = VALUES(higher_is_better), \ + is_member_scale = VALUES(is_member_scale), \ + source_tags = VALUES(source_tags), \ + is_enabled = VALUES(is_enabled)"; + +const INSERT_THRESHOLD_SQL: &str = "\ + INSERT INTO metric_threshold \ + (id, tenant_id, metric_key, scope, role_slug, team_id, good, warn, is_locked) \ + VALUES (?, NULL, ?, 'product-default', '', '', ?, ?, FALSE) \ + ON DUPLICATE KEY UPDATE \ + good = VALUES(good), \ + warn = VALUES(warn)"; + +fn nullable_str_value(v: Option<&str>) -> Value { + match v { + Some(s) => Value::from(s), + None => Value::String(None), + } +} + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + let conn = manager.get_connection(); + let backend = manager.get_database_backend(); + + for row in SEEDS { + let catalog_id = Uuid::now_v7(); + let threshold_id = Uuid::now_v7(); + let source_tags_json_str = source_tags_json(row.source_tags); + + conn.execute(Statement::from_sql_and_values( + backend, + INSERT_CATALOG_SQL, + [ + Value::Bytes(Some(Box::new(catalog_id.as_bytes().to_vec()))), + Value::from(row.metric_key), + Value::from(row.label), + nullable_str_value(row.sublabel), + nullable_str_value(row.description), + nullable_str_value(row.unit), + nullable_str_value(row.format), + Value::from(row.higher_is_better), + Value::from(row.is_member_scale), + Value::from(source_tags_json_str.as_str()), + ], + )) + .await?; + + conn.execute(Statement::from_sql_and_values( + backend, + INSERT_THRESHOLD_SQL, + [ + Value::Bytes(Some(Box::new(threshold_id.as_bytes().to_vec()))), + Value::from(row.metric_key), + Value::from(row.good), + Value::from(row.warn), + ], + )) + .await?; + } + + tracing::info!(seeded = SEEDS.len(), "zulip collab metric_catalog seed applied"); + Ok(()) + } + + async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> { + Err(DbErr::Custom( + "m20260624_000002_seed_zulip_collab_catalog is irreversible: \ + delete the collab_bullet_rows.zulip_messages_sent catalog row manually if needed." + .to_string(), + )) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn one_seed_routed_to_collab_bullet_rows() { + assert_eq!(SEEDS.len(), 1); + assert!(SEEDS[0] + .metric_key + .starts_with("collab_bullet_rows.")); + assert_eq!(SEEDS[0].source_tags, &["zulip-proxy"]); + } +} diff --git a/src/backend/services/analytics-api/src/migration/mod.rs b/src/backend/services/analytics-api/src/migration/mod.rs index 8f5e0ab05..435ae7bb3 100644 --- a/src/backend/services/analytics-api/src/migration/mod.rs +++ b/src/backend/services/analytics-api/src/migration/mod.rs @@ -40,6 +40,8 @@ mod m20260618_000001_ai_claude_team_overage_metric; mod m20260618_000002_seed_claude_team_overage_catalog; mod m20260620_000001_seed_wiki_metrics; mod m20260620_000002_seed_wiki_catalog; +mod m20260624_000001_collab_zulip_chat; +mod m20260624_000002_seed_zulip_collab_catalog; #[cfg(test)] mod live_tests; @@ -92,6 +94,8 @@ impl MigratorTrait for Migrator { Box::new(m20260618_000002_seed_claude_team_overage_catalog::Migration), Box::new(m20260620_000001_seed_wiki_metrics::Migration), Box::new(m20260620_000002_seed_wiki_catalog::Migration), + Box::new(m20260624_000001_collab_zulip_chat::Migration), + Box::new(m20260624_000002_seed_zulip_collab_catalog::Migration), ] } } diff --git a/src/ingestion/scripts/create-bronze-placeholders.sh b/src/ingestion/scripts/create-bronze-placeholders.sh index 172e06953..f0153592a 100644 --- a/src/ingestion/scripts/create-bronze-placeholders.sh +++ b/src/ingestion/scripts/create-bronze-placeholders.sh @@ -72,6 +72,7 @@ CREATE DATABASE IF NOT EXISTS bronze_cursor; CREATE DATABASE IF NOT EXISTS bronze_slack; CREATE DATABASE IF NOT EXISTS bronze_bamboohr; CREATE DATABASE IF NOT EXISTS bronze_bitbucket_cloud; +CREATE DATABASE IF NOT EXISTS bronze_zulip_proxy; SQL # --------------------------------------------------------------------------- @@ -986,4 +987,52 @@ CREATE TABLE IF NOT EXISTS bronze_slack.users_details ( SQL fi +# bronze_zulip_proxy.messages — per-(sender, bucket) aggregated chat counts +# from the Zulip proxy. zulip_proxy__collab_chat_activity dedups by `uniq`, +# joins users on sender_id = id, and sums `count` per (sender email, date). +# The real Airbyte connector overwrites this on first sync (full schema in +# src/ingestion/connectors/collaboration/zulip-proxy/connector.yaml). +if ! ch_table_exists bronze_zulip_proxy messages; then + echo " Creating placeholder: bronze_zulip_proxy.messages" + run_ch <<'SQL' +CREATE TABLE IF NOT EXISTS bronze_zulip_proxy.messages ( + uniq String, + sender_id Nullable(Int64), + count Nullable(Int64), + created_at String, + tenant_id Nullable(String), + source_id Nullable(String), + unique_key String, + _airbyte_raw_id String DEFAULT toString(generateUUIDv4()), + _airbyte_extracted_at DateTime64(3) DEFAULT now64(3), + _airbyte_meta String DEFAULT '{}', + _airbyte_generation_id UInt32 DEFAULT 0 +) ENGINE = ReplacingMergeTree(_airbyte_extracted_at) ORDER BY unique_key; +SQL +fi + +# bronze_zulip_proxy.users — Zulip user directory (full-refresh each sync). +# Joined by id = messages.sender_id to attach the sender email. +if ! ch_table_exists bronze_zulip_proxy users; then + echo " Creating placeholder: bronze_zulip_proxy.users" + run_ch <<'SQL' +CREATE TABLE IF NOT EXISTS bronze_zulip_proxy.users ( + id Nullable(Int64), + uuid Nullable(String), + email Nullable(String), + full_name Nullable(String), + role Nullable(Int64), + is_active Nullable(Bool), + recipient_id Nullable(Int64), + tenant_id Nullable(String), + source_id Nullable(String), + unique_key String, + _airbyte_raw_id String DEFAULT toString(generateUUIDv4()), + _airbyte_extracted_at DateTime64(3) DEFAULT now64(3), + _airbyte_meta String DEFAULT '{}', + _airbyte_generation_id UInt32 DEFAULT 0 +) ENGINE = ReplacingMergeTree(_airbyte_extracted_at) ORDER BY unique_key; +SQL +fi + echo "=== Placeholders: done ===" diff --git a/src/ingestion/scripts/migrations/20260518000000_collab-bullet-rewrite.sql b/src/ingestion/scripts/migrations/20260518000000_collab-bullet-rewrite.sql index d29cab4c9..be26cf281 100644 --- a/src/ingestion/scripts/migrations/20260518000000_collab-bullet-rewrite.sql +++ b/src/ingestion/scripts/migrations/20260518000000_collab-bullet-rewrite.sql @@ -48,10 +48,11 @@ -- ARRAY JOIN (sumIf per source) -- 3. `class_collab_chat_activity` M365 → 1 key -- 4. `class_collab_chat_activity` Slack → 3 keys via ARRAY JOIN +-- 4b. `class_collab_chat_activity` Zulip → 1 key (zulip_messages_sent) -- 5. `class_collab_document_activity` → 3 keys via ARRAY JOIN -- 6. cross-class CTE → 1 key (m365_active_days) -- --- 18 distinct metric_keys after rewrite (down from 20). The two +-- 19 distinct metric_keys after the Zulip add (was 18). The two -- composite-ratio keys (`slack_msgs_per_active_day`, `slack_dm_ratio`) -- visible on the FE live ONLY in the `query_ref` projection — they are -- not emitted by the view. @@ -194,6 +195,24 @@ WHERE s.data_source = 'insight_slack' UNION ALL +-- ─── Branch 4b: class_collab_chat_activity — Zulip ─────────────────── +-- Single key (zulip_messages_sent), no ARRAY JOIN needed. Mirrors the +-- m365 Teams branch: one silver row per (person, date) where +-- data_source = 'insight_zulip_proxy' → period-summed chat messages. +SELECT + lower(z.email) AS person_id, + p.org_unit_id AS org_unit_id, + z.date AS metric_date, + 'zulip_messages_sent' AS metric_key, + toFloat64(ifNull(z.total_chat_messages, 0)) AS metric_value +FROM silver.class_collab_chat_activity AS z +LEFT JOIN insight.people AS p ON lower(z.email) = p.person_id +WHERE z.data_source = 'insight_zulip_proxy' + AND z.email IS NOT NULL + AND z.email != '' + +UNION ALL + -- ─── Branch 5: class_collab_document_activity (M365 OneDrive/SharePoint) ─ -- ARRAY JOIN emits 3 raw counters. SELECT diff --git a/src/ingestion/silver/collaboration/class_collab_chat_activity.sql b/src/ingestion/silver/collaboration/class_collab_chat_activity.sql index 0256c63aa..7595b9f25 100644 --- a/src/ingestion/silver/collaboration/class_collab_chat_activity.sql +++ b/src/ingestion/silver/collaboration/class_collab_chat_activity.sql @@ -1,5 +1,6 @@ -- depends_on: {{ ref('m365__collab_chat_activity') }} -- depends_on: {{ ref('slack__collab_chat_activity') }} +-- depends_on: {{ ref('zulip_proxy__collab_chat_activity') }} {{ config( materialized='incremental', unique_key='unique_key', diff --git a/src/ingestion/tests/e2e/specs/collab_emails_sent.test.yaml b/src/ingestion/tests/e2e/specs/collab_emails_sent.test.yaml index 2ad35227d..ba8168895 100644 --- a/src/ingestion/tests/e2e/specs/collab_emails_sent.test.yaml +++ b/src/ingestion/tests/e2e/specs/collab_emails_sent.test.yaml @@ -79,7 +79,7 @@ cases: find: { metric_key: m365_emails_sent } equal: { value: 40, median: 20, range_min: 10, range_max: 40 } - in: collaboration - assert: "size(items) == 20" + assert: "size(items) == 21" # 20 base collab keys + zulip_messages_sent (Branch 4b) - in: collaboration find: { metric_key: slack_dm_ratio } equal: { value: null } diff --git a/src/ingestion/tests/e2e/specs/collab_zulip_chat.test.yaml b/src/ingestion/tests/e2e/specs/collab_zulip_chat.test.yaml new file mode 100644 index 000000000..0b50a0b67 --- /dev/null +++ b/src/ingestion/tests/e2e/specs/collab_zulip_chat.test.yaml @@ -0,0 +1,80 @@ +spec_version: 1 +description: >- + Zulip chat activity flows bronze → silver class_collab_chat_activity (via + union_by_tag) → gold insight.collab_bullet_rows (Branch 4b, + data_source='insight_zulip_proxy') → IC Bullet Collaboration metric_key + `zulip_messages_sent`. Mirrors collab_emails_sent: alice's own value vs her + team's median/range, with a bronze duplicate that must dedup. + +# bronze — seed fixture. Key = TABLE NAME. People (org_unit) come from +# bamboohr; Zulip users supply the emails that join to them; Zulip messages +# carry the per-sender counts. alice 40, bob 20, carol 10 (all Engineering → +# one team) → value(alice)=40, team median=20, range [10,40]. alice has a +# duplicate `uniq` row (Airbyte re-emit) that must dedup → 40, not 80. +bronze: + bronze_bamboohr.employees: + - $ref: templates/people.yaml#/templates/alice + - $ref: templates/people.yaml#/templates/bob + - $ref: templates/people.yaml#/templates/carol + + bronze_zulip_proxy.users: + - $ref: templates/zulip.yaml#/templates/alice_user + - $ref: templates/zulip.yaml#/templates/bob_user + - $ref: templates/zulip.yaml#/templates/carol_user + + bronze_zulip_proxy.messages: + - $ref: templates/zulip.yaml#/templates/alice_message + uniq: z-alice-20260105 + unique_key: zulip-msg-alice-20260105 + count: 40 + + - $ref: templates/zulip.yaml#/templates/alice_message # duplicate uniq → must dedup, NOT sum to 80 + uniq: z-alice-20260105 + unique_key: zulip-msg-alice-20260105 + count: 40 + + - $ref: templates/zulip.yaml#/templates/bob_message + uniq: z-bob-20260105 + unique_key: zulip-msg-bob-20260105 + count: 20 + + - $ref: templates/zulip.yaml#/templates/carol_message + uniq: z-carol-20260105 + unique_key: zulip-msg-carol-20260105 + count: 10 + +cases: + # IC Bullet Collaboration (…0012): value = the requested person's own value + # (from $filter); median/range = over their team (org_unit). All three are in + # Engineering → one team. zulip_messages_sent over the team = [alice 40, + # bob 20, carol 10]: + # value (alice) = 40 (the requested person) + # median = quantileExact(0.5)[10,20,40] = 20 (team median, NOT alice's) + # range (team) = [10, 40] + # alice is duplicated in bronze (same `uniq`) — her sum must stay 40 (dedup); + # otherwise the value, the median and the range would all drift. + - name: zulip_messages_sent — IC bullet (alice value vs team median, dedup) + request: + url: /v1/metrics/queries + method: POST + body: + queries: + - id: collaboration + metric_id: 00000000-0000-0000-0001-000000000012 + $top: 50 + $filter: "person_id eq 'alice@example.com' and metric_date ge '2026-01-01' and metric_date le '2026-01-31'" + $orderby: metric_key + expect: + - assert: "status == 200" + - in: collaboration + assert: "result.status == 'ok'" + - in: collaboration + find: { metric_key: zulip_messages_sent } + equal: { value: 40, median: 20, range_min: 10, range_max: 40 } + # The Zulip add raises the collaboration bullet from 20 to 21 FE-visible keys. + - in: collaboration + assert: "size(items) == 21" + # No Slack data seeded → the Slack-only DM ratio is null for this person. + - in: collaboration + find: { metric_key: slack_dm_ratio } + equal: { value: null } diff --git a/src/ingestion/tests/e2e/specs/schemas/bronze_zulip_proxy.messages.yaml b/src/ingestion/tests/e2e/specs/schemas/bronze_zulip_proxy.messages.yaml new file mode 100644 index 000000000..bd2a24fd2 --- /dev/null +++ b/src/ingestion/tests/e2e/specs/schemas/bronze_zulip_proxy.messages.yaml @@ -0,0 +1,20 @@ +# Reusable JSON schema for the table (all real placeholder columns). Resolved +# by table name. Mirrors bronze_zulip_proxy.messages in +# src/ingestion/scripts/create-bronze-placeholders.sh. +schemas: + bronze_zulip_proxy.messages: + $schema: http://json-schema.org/draft-07/schema# + type: object + additionalProperties: false + properties: + _airbyte_raw_id: { type: string } + _airbyte_extracted_at: { type: string, format: date-time } + _airbyte_meta: { type: string } + _airbyte_generation_id: { type: integer } + uniq: { type: string } + sender_id: { type: [number, "null"] } + count: { type: [number, "null"] } + created_at: { type: string } + tenant_id: { type: [string, "null"] } + source_id: { type: [string, "null"] } + unique_key: { type: string } diff --git a/src/ingestion/tests/e2e/specs/schemas/bronze_zulip_proxy.users.yaml b/src/ingestion/tests/e2e/specs/schemas/bronze_zulip_proxy.users.yaml new file mode 100644 index 000000000..eb3995024 --- /dev/null +++ b/src/ingestion/tests/e2e/specs/schemas/bronze_zulip_proxy.users.yaml @@ -0,0 +1,23 @@ +# Reusable JSON schema for the table (all real placeholder columns). Resolved +# by table name. Mirrors bronze_zulip_proxy.users in +# src/ingestion/scripts/create-bronze-placeholders.sh. +schemas: + bronze_zulip_proxy.users: + $schema: http://json-schema.org/draft-07/schema# + type: object + additionalProperties: false + properties: + _airbyte_raw_id: { type: string } + _airbyte_extracted_at: { type: string, format: date-time } + _airbyte_meta: { type: string } + _airbyte_generation_id: { type: integer } + id: { type: [number, "null"] } + uuid: { type: [string, "null"] } + email: { type: [string, "null"] } + full_name: { type: [string, "null"] } + role: { type: [number, "null"] } + is_active: { type: [boolean, "null"] } + recipient_id: { type: [number, "null"] } + tenant_id: { type: [string, "null"] } + source_id: { type: [string, "null"] } + unique_key: { type: string } diff --git a/src/ingestion/tests/e2e/specs/templates/zulip.yaml b/src/ingestion/tests/e2e/specs/templates/zulip.yaml new file mode 100644 index 000000000..1f68872af --- /dev/null +++ b/src/ingestion/tests/e2e/specs/templates/zulip.yaml @@ -0,0 +1,78 @@ +# Reusable Zulip records (bronze_zulip_proxy.users + .messages). +# A record is a plain field map; an optional `$ref` inherits from another record +# and sibling keys override it (closest wins). Nested `$ref: "#/..."` stays local +# to this file. +# +# Each base record lists EVERY schema column so a merged bronze row is complete; +# unused fields default to null. The 4 `_airbyte_*` CDK columns are included +# (the chat model dedups by `uniq` ORDER BY `_airbyte_extracted_at`). +# +# Identity: messages join users on `sender_id = id` (+ tenant_id/source_id), so +# the user emails here must match the bamboohr people emails (templates/people.yaml) +# for the collab bullet's `lower(email) = insight.people.person_id` join. +templates: + # ── users ──────────────────────────────────────────────────────────── + zulip_user: + _airbyte_raw_id: "00000000-0000-0000-0000-000000000000" + _airbyte_extracted_at: "2026-01-05T00:00:00" + _airbyte_meta: "{}" + _airbyte_generation_id: 0 + tenant_id: "00000000-0000-0000-0000-000000000000" + source_id: "zulip-proxy-test" + id: null + uuid: null + email: null + full_name: null + role: 400 + is_active: true + recipient_id: null + unique_key: null + + alice_user: + $ref: "#/templates/zulip_user" + id: 1 + email: alice@example.com + full_name: Alice Alpha + unique_key: zulip-user-1 + + bob_user: + $ref: "#/templates/zulip_user" + id: 2 + email: bob@example.com + full_name: Bob Beta + unique_key: zulip-user-2 + + carol_user: + $ref: "#/templates/zulip_user" + id: 3 + email: carol@example.com + full_name: Carol Gamma + unique_key: zulip-user-3 + + # ── messages ───────────────────────────────────────────────────────── + # The proxy ships one row per (sender, bucket) carrying a `count`. The chat + # model dedups by `uniq` then sums `count` per (sender email, date). + zulip_message: + _airbyte_raw_id: "00000000-0000-0000-0000-000000000000" + _airbyte_extracted_at: "2026-01-05T00:00:00" + _airbyte_meta: "{}" + _airbyte_generation_id: 0 + tenant_id: "00000000-0000-0000-0000-000000000000" + source_id: "zulip-proxy-test" + uniq: null + sender_id: null + count: null + created_at: "2026-01-05T10:00:00" + unique_key: null + + alice_message: + $ref: "#/templates/zulip_message" + sender_id: 1 + + bob_message: + $ref: "#/templates/zulip_message" + sender_id: 2 + + carol_message: + $ref: "#/templates/zulip_message" + sender_id: 3 From 0aa8047ba5bc60627c583f66d52cfe47384381cf Mon Sep 17 00:00:00 2001 From: Roman Mitasov Date: Wed, 24 Jun 2026 12:34:15 +0300 Subject: [PATCH 2/4] =?UTF-8?q?docs(connector-skill):=20document=20the=20s?= =?UTF-8?q?ilver=E2=86=92gold=E2=86=92query=5Fref=E2=86=92catalog=20surfac?= =?UTF-8?q?ing=20chain?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Captures what this PR's zulip work had to discover by hand so the next connector reaches the UI first try: - /connector create.md §3.6c: a connector that only reaches silver does NOT appear in the UI — union_by_tag folds it into silver.class_ and nothing else is automatic. Documents the 4 layers to wire (silver depends_on edge, gold
_bullet_rows branch on data_source='insight_', the bullet query_ref sumIf+ARRAY JOIN — noting a section has several copies: IC/Team/member …0041/dept …0045 — and the metric_catalog row), with file locations and PR #1466 as the worked example. - /connector validate.md: new "Dashboard metric surfacing" checklist mirroring the chain. - /metric-e2e-test SKILL.md: rig gotchas that cost real time — adding a bronze table for a not-yet-seeded connector (placeholder heredoc + schema yaml, else the seeder fails on system.columns); stale cargo-target volume serving a binary without new SeaORM migrations (confirm via seaql_migrations, fix with down -v); 1045 access-denied from stale .env; querying the still-up CH/MariaDB after a run; and that adding a metric_key to a shared section shifts other tests' size(items). Co-Authored-By: Claude Opus 4.8 Signed-off-by: Roman Mitasov --- .claude/skills/metric-e2e-test/SKILL.md | 38 +++++++++++ .../skills/connector/workflows/create.md | 63 +++++++++++++++++++ .../skills/connector/workflows/validate.md | 15 +++++ 3 files changed, 116 insertions(+) diff --git a/.claude/skills/metric-e2e-test/SKILL.md b/.claude/skills/metric-e2e-test/SKILL.md index e52c8cf0f..a24668e8b 100644 --- a/.claude/skills/metric-e2e-test/SKILL.md +++ b/.claude/skills/metric-e2e-test/SKILL.md @@ -192,3 +192,41 @@ ls specs/*.test.yaml # list existing tests ``` `` is the file stem (e.g. `collab_emails_sent` for `specs/collab_emails_sent.test.yaml`). Warm re-runs are fine — the session resets the multi-reader collab silver/staging tables at start (conftest). `./e2e.sh down` is only the e2e compose teardown (it is not a deploy), for when you want a fully clean ClickHouse. + +## New bronze table for a not-yet-seeded connector + +The seeder INSERTs into a table that MUST already exist (it reads +`system.columns` and fails otherwise — it does NOT create from the schema YAML). +Bronze tables come from `src/ingestion/scripts/create-bronze-placeholders.sh` +(the rig parses the `run_ch <<'SQL' … SQL` heredocs out of it). So to seed a +connector that isn't there yet: + +1. Add `CREATE DATABASE IF NOT EXISTS bronze_;` to the database heredoc. +2. Add a `CREATE TABLE IF NOT EXISTS bronze_. (…)` block (inside a + `run_ch <<'SQL' … SQL` heredoc) with the columns your dbt model reads + the 4 + `_airbyte_*` CDK columns. Real Airbyte overwrites it on first sync. +3. Add a matching `schemas/bronze_..yaml` (every column; + `additionalProperties: false`) and a base template covering all of them. + +## Gotchas (rig operations + cross-test impact) + +- **Stale binary / your migration didn't run.** `./e2e.sh` runs analytics-api + from the `cargo-target` Docker VOLUME. A volume left over from a prior session + does NOT always rebuild on new Rust files (cargo mtime quirk over the bind + mount), so the binary silently lacks your new SeaORM migrations — symptoms: + `query_ref`/catalog changes don't take effect, a `find` matches 0 rows, a + `size(items)` is off by your new key. Confirm by querying `seaql_migrations` + (below); fix with a full volume wipe + cold rebuild: + `INSIGHT_REPO_ROOT="$(cd ../../../.. && pwd)" docker compose -f compose/docker-compose.yml -f compose/docker-compose.runner.yml down -v`, + then re-run (cold build ~7 min). +- **`1045 Access denied for user 'insight'` at API startup.** Stale + `compose/.env` creds vs a persisted MariaDB volume. Same `down -v` fixes it. +- **Inspect the live DB after a run.** CH + MariaDB stay UP after `./e2e.sh test` + (only the runner is `--rm`). Query directly: + `docker exec insight-e2e-mariadb mariadb -uroot -p"$(grep ^MARIADB_ROOT_PASSWORD compose/.env|cut -d= -f2)" analytics -e "SELECT version FROM seaql_migrations"` + and `docker exec insight-e2e-clickhouse clickhouse-client -q "SELECT … FROM silver.class_"`. +- **Cross-test impact.** Adding a `metric_key` to a shared bullet section raises + that section's `size(items)` for EVERY test that queries it — bump the sibling + tests' count assertions in the same change (e.g. the Zulip add moved the + Collaboration bullet 20 → 21, so `collab_emails_sent.test.yaml` needed the bump + too). diff --git a/cypilot/.core/skills/connector/workflows/create.md b/cypilot/.core/skills/connector/workflows/create.md index 0ecd6566c..4a4a57399 100644 --- a/cypilot/.core/skills/connector/workflows/create.md +++ b/cypilot/.core/skills/connector/workflows/create.md @@ -433,6 +433,69 @@ Rules: no emails), skip this section and document in the README how identities resolve instead (cross-connector JOIN, Silver Step 2 direct mapping). +#### 3.6c Surfacing the connector in a dashboard metric (silver class → gold → query_ref → catalog) + +> **A connector that only reaches silver does NOT appear in the UI.** Tagging a +> model `silver:class_` makes `union_by_tag` fold it into the shared +> `silver.class_` table — and that is ALL that is automatic. The gold views, +> the metric `query_ref`s, and the metric catalog each enumerate their inputs +> EXPLICITLY; a new source contributes nothing past silver until you wire four +> more layers by hand. Skip this whole section if the connector is bronze-only +> or its silver class is not consumed by any gold metric (then say so in the +> README). Reference implementation: PR constructorfabric/insight#1466 (added +> Zulip chat to the Collaboration bullet) — open it as a worked example. + +The data path for a per-person dashboard metric: + +``` +bronze_. + → __.sql (silver, tag silver:class_) ← §3.5 + → silver.class_ (union_by_tag — AUTOMATIC) + → insight.
_bullet_rows (gold VIEW, CH migration) ← layer A + → metrics.query_ref (…IC/Team/member/dept) (SeaORM) ← layer B + → metric_catalog row (label/threshold, SeaORM) ← layer C + → analytics-api → person-profile
card (UI) +``` + +What to add (use an EXISTING section like `collab` / `git` / `ai` / `support` +when the metric fits one; only create a new `
_bullet_rows` view + +metric ids for a genuinely new section): + +- **Layer 0 — silver class build edge.** Add + `-- depends_on: {{ ref('__') }}` to + `src/ingestion/silver//class_.sql`. `union_by_tag` already includes + your model at SQL level once its table exists; this edge makes a + `tag:+` prod run rebuild the class (and the e2e rig's `derive_selectors` + build it from the `silver:` tag). +- **Layer A — gold view branch.** In the CH migration that defines + `insight.
_bullet_rows` + (`src/ingestion/scripts/migrations/*-bullet-rewrite.sql`; idempotent + `DROP+CREATE`, no tracking table → edit the canonical definition in place), + add a branch `... FROM silver.class_ WHERE data_source = 'insight_'` + emitting your `metric_key`(s). The `data_source` literal is exactly what your + silver model SELECTs (`'insight_'`). Join `insight.people` on + `lower(email) = p.person_id` for `org_unit_id`. +- **Layer B — query_ref(s).** The bullet `query_ref`s materialize EVERY + FE-visible `metric_key` as `sumIf(metric_value, metric_key='') AS _v` + then unpivot via `ARRAY JOIN [('', _v), …]`. Add your key to BOTH lists. + SeaORM migrations are append-only — write a NEW + `m__
_.rs` that `UPDATE metrics SET query_ref=… WHERE + id = UNHEX('')`, basing the SQL on the LATEST migration that set that + query_ref (grep the metric id; a section has SEVERAL copies — IC bullet, + Team bullet, member-values `…0041`, dept-distribution `…0045` — update each + surface you need). Register the migration in + `src/backend/services/analytics-api/src/migration/mod.rs`. +- **Layer C — catalog row.** New append-only migration + `m__seed__
_catalog.rs` inserting a `metric_catalog` + row (`
_bullet_rows.`, label, sublabel, unit, + `source_tags: [""]`) + a product-default `metric_threshold`. Model it on + `m20260620_000002_seed_wiki_catalog.rs`. Register in `mod.rs`. + +Then prove the whole chain with a `/metric-e2e-test` fixture (see that skill) — +seed bronze, query the bullet metric, assert your `metric_key`'s value. Adding a +key to a shared section raises its `size(items)`, so bump any sibling test that +asserts the old count. + ### For CDK (`CONNECTOR_TYPE=cdk`): Create Python scaffold: diff --git a/cypilot/.core/skills/connector/workflows/validate.md b/cypilot/.core/skills/connector/workflows/validate.md index 8dde54e4e..46c063744 100644 --- a/cypilot/.core/skills/connector/workflows/validate.md +++ b/cypilot/.core/skills/connector/workflows/validate.md @@ -224,6 +224,21 @@ Exit 0 = PASS for the targeted connector(s); exit 2 = at least one FAIL. Rule ID | `BP-8` | every `promote_bronze_to_rmt` call passes `order_by` | | `BP-9` | every other model that reads bronze depends on `__bronze_promoted` | +### Dashboard metric surfacing (only if the connector should appear in the UI) + +A connector whose silver class feeds a dashboard metric does NOT surface in the +UI from the silver model alone — the gold view, the metric `query_ref`(s), and +the catalog each enumerate inputs explicitly (see `connector-create.md` §3.6c). +If the connector is expected to show a per-person card, verify the full chain; +if it is bronze-only or its class has no gold consumer, confirm the README says +so and skip this section. + +- [ ] Silver class `class_.sql` carries `-- depends_on: {{ ref('__') }}` for the connector. +- [ ] The gold `insight.
_bullet_rows` view has a branch `FROM silver.class_ WHERE data_source = 'insight_'` emitting the connector's `metric_key`(s). (`data_source` literal == what the silver model SELECTs.) +- [ ] EVERY bullet `query_ref` that should show the key was re-set in a NEW append-only SeaORM migration — the key appears in both the `sumIf(... metric_key='') AS _v` list AND the `ARRAY JOIN [('', _v), …]` unpivot. A section typically has several copies (IC `…0012`-style, Team, member-values `…0041`, dept-dist `…0045`); each must be updated for the surface it backs. Base the new SQL on the LATEST migration that set that id (grep the metric hex id), and register the migration in `migration/mod.rs`. +- [ ] A `metric_catalog` row + product-default `metric_threshold` for `
_bullet_rows.` exists (new append-only migration, `source_tags: [""]`, registered in `mod.rs`). +- [ ] An e2e fixture (`/metric-e2e-test`) seeds bronze and asserts the metric_key's value end-to-end; any sibling test asserting the section's `size(items)` was bumped for the new key. + ### Credentials Template - [ ] `credentials.yaml.example` lists all required fields - [ ] `insight_source_id` is included From df6dd942e6a3844bd82573bc064ac71516c339a1 Mon Sep 17 00:00:00 2001 From: Roman Mitasov Date: Wed, 24 Jun 2026 12:55:22 +0300 Subject: [PATCH 3/4] fix(e2e): force analytics-api recompile so new migrations are never missed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `build()` builds analytics-api into the `cargo-target` Docker volume. On Docker Desktop (macOS) the mtimes cargo reads through the bind mount do not reliably advance when sources change on the host, so cargo relinked a stale cached object and the spawned binary silently lacked newly-added SeaORM migrations. Symptom: a freshly-added metric/query_ref/catalog migration never ran (absent from `seaql_migrations`), so `./e2e.sh test` failed with NO_ZULIP / `size(items)` off-by-one and only a full `down -v` cold rebuild (itself flaky — the volume is often "in use") worked around it. Fix: touch the analytics-api crate sources before `cargo build`, forcing a recompile of that crate every run (~1-2 min; it is a leaf bin, so its deps stay cached — not a cold build). `./e2e.sh test` now picks up new migrations with no `down -v` ritual. Confirmed: full suite 33 passed on a warm volume. Also updates the /metric-e2e-test skill note (the stale-binary trap is now auto-handled; `down -v` is only for a DB/disk reset). Co-Authored-By: Claude Opus 4.8 Signed-off-by: Roman Mitasov --- .claude/skills/metric-e2e-test/SKILL.md | 20 ++++++++++--------- .../tests/e2e/e2e_lib/analytics_api.py | 19 ++++++++++++++++++ 2 files changed, 30 insertions(+), 9 deletions(-) diff --git a/.claude/skills/metric-e2e-test/SKILL.md b/.claude/skills/metric-e2e-test/SKILL.md index a24668e8b..77eb40e31 100644 --- a/.claude/skills/metric-e2e-test/SKILL.md +++ b/.claude/skills/metric-e2e-test/SKILL.md @@ -210,15 +210,17 @@ connector that isn't there yet: ## Gotchas (rig operations + cross-test impact) -- **Stale binary / your migration didn't run.** `./e2e.sh` runs analytics-api - from the `cargo-target` Docker VOLUME. A volume left over from a prior session - does NOT always rebuild on new Rust files (cargo mtime quirk over the bind - mount), so the binary silently lacks your new SeaORM migrations — symptoms: - `query_ref`/catalog changes don't take effect, a `find` matches 0 rows, a - `size(items)` is off by your new key. Confirm by querying `seaql_migrations` - (below); fix with a full volume wipe + cold rebuild: - `INSIGHT_REPO_ROOT="$(cd ../../../.. && pwd)" docker compose -f compose/docker-compose.yml -f compose/docker-compose.runner.yml down -v`, - then re-run (cold build ~7 min). +- **Stale binary / your migration didn't run.** Historically the biggest trap: + `./e2e.sh` builds analytics-api into the `cargo-target` Docker volume, and on + Docker Desktop (macOS) the mtimes cargo reads through the bind mount don't + reliably advance, so cargo relinked a stale object and the binary silently + lacked new SeaORM migrations (symptoms: `query_ref`/catalog changes have no + effect, a `find` matches 0 rows, `size(items)` off by your new key). FIXED in + `e2e_lib/analytics_api.py::build` — it now `touch`es the analytics-api crate + sources before `cargo build`, forcing a recompile every run (~1-2 min, only + that crate). So a plain `./e2e.sh test` picks up new migrations now; you should + NOT need `down -v` for this. If you still suspect a stale binary, confirm by + querying `seaql_migrations` (below) — your migration version must be present. - **`1045 Access denied for user 'insight'` at API startup.** Stale `compose/.env` creds vs a persisted MariaDB volume. Same `down -v` fixes it. - **Inspect the live DB after a run.** CH + MariaDB stay UP after `./e2e.sh test` diff --git a/src/ingestion/tests/e2e/e2e_lib/analytics_api.py b/src/ingestion/tests/e2e/e2e_lib/analytics_api.py index 33a0c4004..a1a6ff014 100644 --- a/src/ingestion/tests/e2e/e2e_lib/analytics_api.py +++ b/src/ingestion/tests/e2e/e2e_lib/analytics_api.py @@ -179,6 +179,25 @@ def build(cfg: SessionConfig) -> Path: f"rust-version ≥ {required[0]}.{required[1]}. " f"Run `rustup update stable` and retry." ) + # Force cargo to recompile the analytics-api crate from the CURRENT source. + # + # The repo is bind-mounted into the runner; on Docker Desktop (macOS) the + # mtimes cargo reads through that mount do not reliably advance when files + # are edited on the host, so cargo's fingerprint check misses new/changed + # sources (most painfully: a new SeaORM migration) and relinks a stale + # cached object instead of recompiling. The binary then silently lacks the + # migration — tests fail with NO_ZULIP / size off-by-one and no `down -v` + # short of a full cold rebuild fixes it. Bumping the mtimes here (a real + # write the container's FS layer registers) makes cargo recompile the crate + # every run. Only the analytics-api crate is affected (it is a leaf bin — + # nothing depends on it); its dependencies stay cached, so the cost is one + # crate recompile (~1-2 min), not a cold build. + crate_src = cfg.repo_root / "src/backend/services/analytics-api/src" + touched = 0 + for rs in crate_src.rglob("*.rs"): + rs.touch() + touched += 1 + LOG.info("touched %d analytics-api source files to force a fresh compile", touched) LOG.info("cargo build --release -p analytics-api (cargo=%s, version=%s)", cargo, version) try: result = subprocess.run( From 0a58c4f88cf15c53cea5d254efa11c9a8d9914e1 Mon Sep 17 00:00:00 2001 From: Roman Mitasov Date: Wed, 24 Jun 2026 13:01:49 +0300 Subject: [PATCH 4/4] style(analytics-api): rustfmt the zulip catalog migration cargo fmt --check flagged line-wrapping in m20260624_000002_seed_zulip_collab_catalog (tracing::info! args + assert! chain). No logic change. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Roman Mitasov --- .../m20260624_000002_seed_zulip_collab_catalog.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/backend/services/analytics-api/src/migration/m20260624_000002_seed_zulip_collab_catalog.rs b/src/backend/services/analytics-api/src/migration/m20260624_000002_seed_zulip_collab_catalog.rs index add30d89d..2b71dc193 100644 --- a/src/backend/services/analytics-api/src/migration/m20260624_000002_seed_zulip_collab_catalog.rs +++ b/src/backend/services/analytics-api/src/migration/m20260624_000002_seed_zulip_collab_catalog.rs @@ -136,7 +136,10 @@ impl MigrationTrait for Migration { .await?; } - tracing::info!(seeded = SEEDS.len(), "zulip collab metric_catalog seed applied"); + tracing::info!( + seeded = SEEDS.len(), + "zulip collab metric_catalog seed applied" + ); Ok(()) } @@ -156,9 +159,7 @@ mod tests { #[test] fn one_seed_routed_to_collab_bullet_rows() { assert_eq!(SEEDS.len(), 1); - assert!(SEEDS[0] - .metric_key - .starts_with("collab_bullet_rows.")); + assert!(SEEDS[0].metric_key.starts_with("collab_bullet_rows.")); assert_eq!(SEEDS[0].source_tags, &["zulip-proxy"]); } }