v1.4.0: Auto-compact streaming, Bedrock summarization support, subagent orchestration, and Galaxy rebrand continuation
Major features: - Auto-compact: triggers conversation summarization when context window >= 85%, compacts Bedrock message history to a summary pair, and tracks live context tokens - Bedrock summarization: plumbs `is_summarization` flag through translator/client/response pipeline, handles SummarizeConversation input type, and marks `summarized` in metadata - Session restore: rebuilds bedrock_message_history from persisted task messages via newly-public `convert_proto_message`, preventing empty history on reconnect - Subagent orchestration: adds SubagentQuestion/Answer/CompletionSummary event types, parent-child question routing with depth limits, retry counting, and drain methods - Summarization UI: inline SummarizationView in AI blocks with progress/finished states Refactors: - Rename WarpTheme → GalaxyTheme across ~100 files (rebrand continuation) - Rename warp_home_config_dir → galaxy_home_config_dir and related path functions - Predefined rules: replace "System Defined Rule #N" with descriptive names (e.g. "Correctness Over Speed", "Never Guess") and add lookup helpers - Usage view: replace cumulative input/output token display with live context tokens, cache hit rate calculation, and separate cache read/write stats - Telemetry: remove verbose doc comments, simplify trait definitions - Facts view: simplify delete permission check (always allow local deletion) - Remove warp_managed_paths_watcher.rs (dead code) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ryan Ward
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
eaa2ddc75e
commit
6f54e2cb30
@@ -4475,8 +4475,27 @@ impl UpdateManager {
|
||||
initiated_by: InitiatedBy,
|
||||
ctx: &mut ModelContext<Self>,
|
||||
) {
|
||||
// If the object isn't known to the server yet, we can't delete it.
|
||||
// If the object isn't known to the server yet, delete it locally.
|
||||
let Some(server_id) = id.server_id() else {
|
||||
let uid = id.uid();
|
||||
let sync_ids_and_types = CloudModel::handle(ctx).update(ctx, |cloud_model, ctx| {
|
||||
let (sync_ids_and_types, _) = cloud_model.delete_objects_by_id(vec![uid], ctx);
|
||||
sync_ids_and_types
|
||||
});
|
||||
if !sync_ids_and_types.is_empty() {
|
||||
self.save_to_db([ModelEvent::DeleteObjects {
|
||||
ids: sync_ids_and_types,
|
||||
}]);
|
||||
}
|
||||
ctx.emit(UpdateManagerEvent::ObjectOperationComplete {
|
||||
result: ObjectOperationResult {
|
||||
success_type: OperationSuccessType::Success,
|
||||
operation: ObjectOperation::Delete { initiated_by },
|
||||
client_id: None,
|
||||
server_id: None,
|
||||
num_objects: Some(1),
|
||||
},
|
||||
});
|
||||
return;
|
||||
};
|
||||
|
||||
|
||||
@@ -13,5 +13,3 @@ pub mod sync_queue;
|
||||
pub mod telemetry;
|
||||
pub(crate) mod telemetry_ext;
|
||||
pub mod voice_transcriber;
|
||||
|
||||
pub use galaxy_core::operating_system_info::OperatingSystemInfo;
|
||||
|
||||
@@ -1,228 +1,24 @@
|
||||
use std::sync::Arc;
|
||||
use std::{fs::remove_file, time::Duration};
|
||||
|
||||
use anyhow::Context;
|
||||
use chrono::{LocalResult, TimeZone, Utc};
|
||||
use galaxy_core::execution_mode::AppExecutionMode;
|
||||
use galaxy_core::{report_error, report_if_error};
|
||||
use galaxyui::r#async::{FutureExt as _, Timer};
|
||||
use galaxyui::{App, Entity, ModelContext, SingletonEntity};
|
||||
use galaxyui::{Entity, ModelContext, SingletonEntity};
|
||||
|
||||
use super::{rudder_event_file_path, RUDDER_TELEMETRY_EVENTS_FILE_NAME};
|
||||
use crate::auth::AuthStateProvider;
|
||||
use crate::channel::ChannelState;
|
||||
use crate::features::FeatureFlag;
|
||||
use crate::{
|
||||
server::server_api::ServerApi,
|
||||
settings::{PrivacySettings, PrivacySettingsChangedEvent},
|
||||
};
|
||||
use crate::server::server_api::ServerApi;
|
||||
|
||||
use super::clear_event_queue;
|
||||
|
||||
// How often we send Active Usage signals.
|
||||
const ACTIVE_USAGE_DURATION: Duration = Duration::from_secs(60);
|
||||
|
||||
/// Duration to wait before flushing the event queue to Rudderstack.
|
||||
const TELEMETRY_FLUSH_DURATION: Duration = Duration::from_secs(30);
|
||||
|
||||
/// Max telemetry events to write to disk. This is bounded to limit the size of the file as well
|
||||
/// as latency of writing the file.
|
||||
const MAX_TELEMETRY_EVENTS_TO_STORE: usize = 20;
|
||||
|
||||
/// Maximum time to wait for the telemetry flush network request during shutdown.
|
||||
/// If the network is unavailable or slow, we don't want the CLI process to hang indefinitely.
|
||||
const TELEMETRY_SHUTDOWN_FLUSH_TIMEOUT: Duration = Duration::from_secs(5);
|
||||
|
||||
/// App singleton responsible for scheduling periodic background tasks for sending batches of
|
||||
/// telemetry events to Rudderstack. This model respects the user's telemetry enablement setting.
|
||||
/// No-op telemetry collector. Telemetry has been removed from Galaxy.
|
||||
pub struct TelemetryCollector {
|
||||
server_api: Arc<ServerApi>,
|
||||
_server_api: Arc<ServerApi>,
|
||||
}
|
||||
|
||||
impl TelemetryCollector {
|
||||
pub fn new(server_api: Arc<ServerApi>) -> Self {
|
||||
Self { server_api }
|
||||
}
|
||||
|
||||
pub fn initialize_telemetry_collection(&self, ctx: &mut ModelContext<TelemetryCollector>) {
|
||||
// Start a background thread to periodically flush events from the telemetry event queue.
|
||||
if ChannelState::is_release_bundle() || FeatureFlag::WithSandboxTelemetry.is_enabled() {
|
||||
// Flush the events to Rudderstack that were persisted into a file the last time the app was
|
||||
// quit.
|
||||
self.flush_persisted_events_from_disk(ctx);
|
||||
}
|
||||
|
||||
// Send Active App Usage signals
|
||||
if FeatureFlag::RecordAppActiveEvents.is_enabled()
|
||||
&& (ChannelState::is_release_bundle() || FeatureFlag::WithSandboxTelemetry.is_enabled())
|
||||
{
|
||||
self.schedule_send_active_usage_event(ctx);
|
||||
}
|
||||
|
||||
// Start a background thread to periodically flush events from the telemetry event queue.
|
||||
if ChannelState::is_release_bundle()
|
||||
|| FeatureFlag::WithSandboxTelemetry.is_enabled()
|
||||
|| FeatureFlag::SendTelemetryToFile.is_enabled()
|
||||
{
|
||||
self.schedule_event_queue_flush(ctx);
|
||||
}
|
||||
|
||||
// Clear queued telemetry events when telemetry is enabled or disabled. If telemetry is
|
||||
// enabled, we will start sending Rudderstack requests when the event queue is periodically
|
||||
// flushed. The initial request should not contain any events recorded when the user was
|
||||
// previously opted-out of telemetry. In the case where the user turns the telemetry from
|
||||
// on to off, we should not send another request with any telemetry, even if the event was
|
||||
// initially recorded prior to the user turning telemetry off.`
|
||||
ctx.subscribe_to_model(&PrivacySettings::handle(ctx), |_me, event, _ctx| {
|
||||
if let PrivacySettingsChangedEvent::UpdateIsTelemetryEnabled { .. } = event {
|
||||
clear_event_queue();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Writes all queued but unsent telemetry telemetry events to disk so that they may be sent
|
||||
/// on the next app startup.
|
||||
pub fn write_telemetry_events_to_disk(&self, ctx: &mut ModelContext<TelemetryCollector>) {
|
||||
match self.server_api.persist_telemetry_events(
|
||||
MAX_TELEMETRY_EVENTS_TO_STORE,
|
||||
PrivacySettings::as_ref(ctx).get_snapshot(ctx),
|
||||
) {
|
||||
Ok(()) => {
|
||||
log::info!("Successfully wrote telemetry events to disk")
|
||||
}
|
||||
Err(e) => {
|
||||
log::error!("Failed to write telemetry events to disk {e:#}");
|
||||
}
|
||||
Self {
|
||||
_server_api: server_api,
|
||||
}
|
||||
}
|
||||
|
||||
/// Flushes telemetry events when the app is shutting down.
|
||||
///
|
||||
/// Depending on the app's execution mode, this will either:
|
||||
/// * Write events to disk, for sending on the next app startup
|
||||
/// * Synchronously send events to rudderstack
|
||||
pub fn flush_telemetry_events_for_shutdown(&self, ctx: &mut ModelContext<TelemetryCollector>) {
|
||||
let execution_mode = AppExecutionMode::as_ref(ctx);
|
||||
pub fn initialize_telemetry_collection(&self, _ctx: &mut ModelContext<TelemetryCollector>) {}
|
||||
|
||||
if execution_mode.send_telemetry_at_shutdown() {
|
||||
let privacy_settings_snapshot = PrivacySettings::as_ref(ctx).get_snapshot(ctx);
|
||||
let server_api = self.server_api.clone();
|
||||
match galaxyui::r#async::block_on(async move {
|
||||
server_api
|
||||
.flush_telemetry_events(privacy_settings_snapshot)
|
||||
.with_timeout(TELEMETRY_SHUTDOWN_FLUSH_TIMEOUT)
|
||||
.await
|
||||
}) {
|
||||
Ok(Ok(count)) => {
|
||||
if count > 0 {
|
||||
log::info!("Successfully flushed telemetry events before shutdown");
|
||||
}
|
||||
}
|
||||
Ok(Err(e)) => {
|
||||
report_error!(e.context("Error flushing telemetry events before shutdown"));
|
||||
}
|
||||
Err(_) => {
|
||||
log::warn!(
|
||||
"Telemetry flush timed out after {}s during shutdown, skipping",
|
||||
TELEMETRY_SHUTDOWN_FLUSH_TIMEOUT.as_secs()
|
||||
);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
self.write_telemetry_events_to_disk(ctx);
|
||||
}
|
||||
}
|
||||
|
||||
/// Sends rudderstack requests containing events persisted to disk (if telemetry is enabled).
|
||||
/// Events may be written to disk at the end of a session prior to app termination; this
|
||||
/// function should be called on startup to track events that were recorded at the end of the
|
||||
/// last session and were not flushed.
|
||||
fn flush_persisted_events_from_disk(&self, ctx: &mut ModelContext<TelemetryCollector>) {
|
||||
let privacy_settings_snapshot = PrivacySettings::as_ref(ctx).get_snapshot(ctx);
|
||||
let server_api = self.server_api.clone();
|
||||
let _ = ctx.spawn(
|
||||
async move {
|
||||
let new_path = rudder_event_file_path();
|
||||
let old_path =
|
||||
galaxy_core::paths::state_dir().join(RUDDER_TELEMETRY_EVENTS_FILE_NAME);
|
||||
|
||||
// Try flushing from both new and legacy locations.
|
||||
for path in [new_path, old_path] {
|
||||
report_if_error!(server_api
|
||||
.flush_persisted_events_to_rudder(&path, privacy_settings_snapshot)
|
||||
.await
|
||||
.context("Failed to flush rudder events from disk"));
|
||||
// Remove the file regardless of outcome of flushing the events to avoid the
|
||||
// case where we accidentally try to re-flush the events on the next app startup.
|
||||
if let Err(e) = remove_file(&path) {
|
||||
if e.kind() != std::io::ErrorKind::NotFound {
|
||||
galaxy_core::report_error!(
|
||||
anyhow::anyhow!(e).context("Failed to remove persisted event file")
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
|_, _, _| (),
|
||||
);
|
||||
}
|
||||
|
||||
/// Schedules a background task to send an active usage event in a rudderstack request if
|
||||
/// telemetry is enabled. The scheduled task once again schedules itself after
|
||||
/// `ACTIVE_USAGE_DURATION`.
|
||||
fn schedule_send_active_usage_event(&self, ctx: &mut ModelContext<TelemetryCollector>) {
|
||||
let auth_state = AuthStateProvider::as_ref(ctx).get().clone();
|
||||
let is_telemetry_enabled = PrivacySettings::as_ref(ctx).is_telemetry_enabled;
|
||||
let _ = ctx.spawn(
|
||||
async move {
|
||||
// Record app active if there was any activity now or right after the previous check
|
||||
let last_active_timestamp = App::last_active_timestamp();
|
||||
if is_telemetry_enabled
|
||||
&& last_active_timestamp + ACTIVE_USAGE_DURATION.as_secs() as i64
|
||||
> Utc::now().timestamp()
|
||||
{
|
||||
if let LocalResult::Single(timestamp) =
|
||||
Utc.timestamp_opt(last_active_timestamp, 0)
|
||||
{
|
||||
galaxyui::telemetry::record_app_active_event(
|
||||
auth_state.user_id().map(|uid| uid.as_string()),
|
||||
auth_state.anonymous_id(),
|
||||
timestamp,
|
||||
);
|
||||
}
|
||||
}
|
||||
Timer::after(ACTIVE_USAGE_DURATION).await;
|
||||
},
|
||||
|me, _, ctx| me.schedule_send_active_usage_event(ctx),
|
||||
);
|
||||
}
|
||||
|
||||
/// Flushes events from the in-memory event queue and schedules a background task to send
|
||||
/// them in rudderstack request if telemetry is enabled. The scheduled task once again schedules
|
||||
/// itself after `TELEMETRY_FLUSH_DURATION`.
|
||||
fn schedule_event_queue_flush(&self, ctx: &mut ModelContext<TelemetryCollector>) {
|
||||
let server_api = self.server_api.clone();
|
||||
let privacy_settings_snapshot = PrivacySettings::as_ref(ctx).get_snapshot(ctx);
|
||||
let _ = ctx.spawn(
|
||||
async move {
|
||||
match server_api
|
||||
.flush_telemetry_events(privacy_settings_snapshot)
|
||||
.await
|
||||
{
|
||||
Ok(count) => {
|
||||
if count > 0 {
|
||||
log::debug!("Flushed telemetry events.");
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
log::info!("Failed to flush events from Telemetry queue: {e}");
|
||||
}
|
||||
}
|
||||
Timer::after(TELEMETRY_FLUSH_DURATION).await;
|
||||
},
|
||||
|me, _, ctx| me.schedule_event_queue_flush(ctx),
|
||||
);
|
||||
}
|
||||
pub fn flush_telemetry_events_for_shutdown(&self, _ctx: &mut ModelContext<Self>) {}
|
||||
}
|
||||
|
||||
impl Entity for TelemetryCollector {
|
||||
|
||||
@@ -1,33 +1,5 @@
|
||||
//! Module that builds a static context to attach to each of our events that are sent to Rudderstack.
|
||||
//! This is needed so we know the backing operating system and version of each telemetry event.
|
||||
|
||||
use super::rudder_message::Message as RudderMessage;
|
||||
use crate::server::OperatingSystemInfo;
|
||||
|
||||
use serde::Serialize;
|
||||
use serde_json::{json, Value};
|
||||
|
||||
use std::sync::OnceLock;
|
||||
|
||||
#[cfg(target_family = "wasm")]
|
||||
use galaxyui::platform::wasm;
|
||||
|
||||
static TELEMETRY_CONTEXT: OnceLock<TelemetryContext> = OnceLock::new();
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct TelemetryContextInfo {
|
||||
/// Info about the operating system of the client.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
os: Option<&'static OperatingSystemInfo>,
|
||||
/// The user agent provided by the browser, if running on Web. If not on
|
||||
/// Web, this is always `None`.
|
||||
#[serde(rename = "userAgent", skip_serializing_if = "Option::is_none")]
|
||||
user_agent: Option<String>,
|
||||
}
|
||||
|
||||
/// Newtype representing a [`Value`] with a serialized version of the context that we send to
|
||||
/// Rudderstack.
|
||||
/// See https://www.rudderstack.com/docs/event-spec/standard-events/common-fields/#contextual-fields.
|
||||
pub struct TelemetryContext(Value);
|
||||
|
||||
impl TelemetryContext {
|
||||
@@ -36,65 +8,8 @@ impl TelemetryContext {
|
||||
}
|
||||
}
|
||||
|
||||
impl TelemetryContext {
|
||||
fn new() -> Self {
|
||||
let context = TelemetryContextInfo {
|
||||
os: OperatingSystemInfo::get().ok(),
|
||||
user_agent: user_agent(),
|
||||
};
|
||||
|
||||
match serde_json::to_value(context) {
|
||||
Ok(value) => Self(value),
|
||||
Err(e) => {
|
||||
log::error!("Failed to serialize telemetry context info to JSON value: {e:?}");
|
||||
Self(json!({}))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Extension trait used to attach a telemetry context.
|
||||
pub(super) trait AttachContext {
|
||||
/// Attaches a context to the given object.
|
||||
fn attach_context(&mut self);
|
||||
}
|
||||
|
||||
impl AttachContext for RudderMessage {
|
||||
/// Attaches the context to the [`RudderMessage`]. Note this is currently last write wins; if a
|
||||
/// message already has a `context` set it will be overridden.
|
||||
// TODO(alokedesai): Merge the incoming context with the static `TelemetryContext`, if set.
|
||||
fn attach_context(&mut self) {
|
||||
let context = telemetry_context().as_value();
|
||||
match self {
|
||||
RudderMessage::Identify(identify) => {
|
||||
identify.context = Some(context);
|
||||
}
|
||||
RudderMessage::Track(track) => track.context = Some(context),
|
||||
RudderMessage::Page(page) => page.context = Some(context),
|
||||
RudderMessage::Screen(screen) => screen.context = Some(context),
|
||||
RudderMessage::Group(group) => group.context = Some(context),
|
||||
RudderMessage::Alias(alias) => alias.context = Some(context),
|
||||
RudderMessage::Batch(batch) => batch.context = Some(context),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the user agent provided by the browser, if on Web. If not on Web,
|
||||
/// or if the user agent was not able to be read, returns None.
|
||||
fn user_agent() -> Option<String> {
|
||||
cfg_if::cfg_if! {
|
||||
if #[cfg(target_family = "wasm")] {
|
||||
wasm::user_agent()
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the telemetry context
|
||||
/// that should be attached to all telemetry events associated to this client.
|
||||
///
|
||||
/// [Rudderstack](https://www.rudderstack.com/docs/event-spec/standard-events/common-fields/#contextual-fields)
|
||||
pub fn telemetry_context() -> &'static TelemetryContext {
|
||||
TELEMETRY_CONTEXT.get_or_init(TelemetryContext::new)
|
||||
use std::sync::OnceLock;
|
||||
static TELEMETRY_CONTEXT: OnceLock<TelemetryContext> = OnceLock::new();
|
||||
TELEMETRY_CONTEXT.get_or_init(|| TelemetryContext(json!({})))
|
||||
}
|
||||
|
||||
@@ -1,93 +1,27 @@
|
||||
/// Sends a telemetry event to Rudderstack immediately instead of adding it to the event queue that is
|
||||
/// periodically flushed. This is useful under certain conditions where we want to ensure an event
|
||||
/// is immediately sent to Rudderstack even if the user quits before the queue is flushed.
|
||||
/// No-op: telemetry has been removed from Galaxy.
|
||||
#[macro_export]
|
||||
macro_rules! send_telemetry_sync_from_ctx {
|
||||
($event:expr, $ctx:expr) => {
|
||||
#[allow(unused_imports)]
|
||||
use galaxy_core::telemetry::TelemetryEvent as _;
|
||||
let event = $event;
|
||||
if event.enablement_state().is_enabled() {
|
||||
let server_api =
|
||||
<$crate::server::server_api::ServerApiProvider as galaxyui::SingletonEntity>::handle(
|
||||
$ctx,
|
||||
)
|
||||
.as_ref($ctx)
|
||||
.get();
|
||||
let privacy_settings_snapshot =
|
||||
<$crate::settings::PrivacySettings as galaxyui::SingletonEntity>::handle($ctx)
|
||||
.as_ref($ctx)
|
||||
.get_snapshot($ctx);
|
||||
let _ = $ctx.spawn(
|
||||
async move {
|
||||
if let Err(error) = server_api
|
||||
.send_telemetry_event(event, privacy_settings_snapshot)
|
||||
.await
|
||||
{
|
||||
log::warn!("Error occurred with sending telemetry event: {}", error);
|
||||
}
|
||||
},
|
||||
|_, _, _| {},
|
||||
);
|
||||
}
|
||||
let _ = &$event;
|
||||
let _ = &$ctx;
|
||||
};
|
||||
}
|
||||
|
||||
/// Sends a telemetry event to Rudderstack immediately. This is the same as [`send_telemetry_sync_from_ctx`],
|
||||
/// but can be used when the caller only has access to an [`App`] and not a
|
||||
/// `ViewContext`.
|
||||
/// No-op: telemetry has been removed from Galaxy.
|
||||
#[macro_export]
|
||||
macro_rules! send_telemetry_sync_from_app_ctx {
|
||||
($event:expr, $app_ctx:expr) => {
|
||||
#[allow(unused_imports)]
|
||||
use galaxy_core::telemetry::TelemetryEvent as _;
|
||||
if $event.enablement_state().is_enabled() {
|
||||
let server_api =
|
||||
<$crate::server::server_api::ServerApiProvider as galaxyui::SingletonEntity>::handle(
|
||||
$app_ctx,
|
||||
)
|
||||
.as_ref($app_ctx)
|
||||
.get();
|
||||
let privacy_settings_snapshot =
|
||||
<$crate::settings::PrivacySettings as galaxyui::SingletonEntity>::handle($app_ctx)
|
||||
.as_ref($app_ctx)
|
||||
.get_snapshot($app_ctx);
|
||||
$app_ctx
|
||||
.background_executor()
|
||||
.spawn(async move {
|
||||
if let Err(error) = server_api
|
||||
.send_telemetry_event($event, privacy_settings_snapshot)
|
||||
.await
|
||||
{
|
||||
log::warn!("Error occurred with sending telemetry event: {error}");
|
||||
}
|
||||
})
|
||||
.detach();
|
||||
}
|
||||
let _ = &$event;
|
||||
let _ = &$app_ctx;
|
||||
};
|
||||
}
|
||||
|
||||
/// Sends a telemetry `track` event Rudderstack asynchronously. This is the same as the
|
||||
/// [`send_telemetry_from_ctx`], except can be called any time you have an Arc<Background>.
|
||||
/// This should only be called when invoking one of the other macros isn't possible; for example,
|
||||
/// when you are already on a background thread and thus can't access any app context.
|
||||
/// No-op: telemetry has been removed from Galaxy.
|
||||
#[macro_export]
|
||||
macro_rules! send_telemetry_on_executor {
|
||||
($auth_state: expr, $event:expr, $executor:expr) => {
|
||||
#[allow(unused_imports)]
|
||||
use galaxy_core::telemetry::TelemetryEvent as _;
|
||||
let event = $event;
|
||||
if event.enablement_state().is_enabled() {
|
||||
let user_id = $auth_state.user_id().map(|uid| uid.as_string());
|
||||
let anonymous_id = $auth_state.anonymous_id();
|
||||
galaxyui::record_telemetry_on_executor!(
|
||||
user_id,
|
||||
anonymous_id,
|
||||
event.name().into(),
|
||||
event.payload(),
|
||||
event.contains_ugc(),
|
||||
$executor
|
||||
);
|
||||
}
|
||||
($auth_state:expr, $event:expr, $executor:expr) => {
|
||||
let _ = &$auth_state;
|
||||
let _ = &$event;
|
||||
let _ = &$executor;
|
||||
};
|
||||
}
|
||||
|
||||
+27
-373
@@ -1,53 +1,19 @@
|
||||
mod collector;
|
||||
mod context;
|
||||
pub mod context_provider;
|
||||
mod events;
|
||||
pub mod events;
|
||||
mod macros;
|
||||
pub mod rudder_message;
|
||||
pub mod secret_redaction;
|
||||
|
||||
use chrono::Utc;
|
||||
pub use collector::*;
|
||||
pub use context::telemetry_context;
|
||||
pub use events::*;
|
||||
|
||||
use crate::auth::UserUid;
|
||||
use crate::features::FeatureFlag;
|
||||
use crate::server::telemetry::context::AttachContext;
|
||||
use crate::server::telemetry_ext::TelemetryExt;
|
||||
use crate::settings::PrivacySettingsSnapshot;
|
||||
use crate::ChannelState;
|
||||
use anyhow::Result;
|
||||
use futures::FutureExt;
|
||||
use galaxy_core::channel::RudderStackDestination;
|
||||
use galaxyui::telemetry::Event;
|
||||
use rudder_message::{
|
||||
Batch as RudderBatch, BatchMessage as RudderBatchMessageWithMetadata,
|
||||
BatchMessageItem as RudderBatchMessage, Message as RudderMessage,
|
||||
};
|
||||
use std::fs::File;
|
||||
#[cfg(not(target_family = "wasm"))]
|
||||
use std::fs::OpenOptions;
|
||||
use std::future::Future;
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
/// Filename for file where telemetry events are written on app quit.
|
||||
const RUDDER_TELEMETRY_EVENTS_FILE_NAME: &str = "rudder_telemetry_events.json";
|
||||
|
||||
/// Filepath where the Rudder events should be written on app quit.
|
||||
fn rudder_event_file_path() -> PathBuf {
|
||||
galaxy_core::paths::secure_state_dir()
|
||||
.unwrap_or_else(galaxy_core::paths::state_dir)
|
||||
.join(RUDDER_TELEMETRY_EVENTS_FILE_NAME)
|
||||
}
|
||||
|
||||
/// Removes all telemetry events from the app telemetry event queue.
|
||||
pub fn clear_event_queue() {
|
||||
let _ = galaxyui::telemetry::flush_events();
|
||||
}
|
||||
|
||||
/// No-op stub. Telemetry has been removed from Galaxy.
|
||||
pub struct TelemetryApi {
|
||||
pub(super) client: http_client::Client,
|
||||
pub client: http_client::Client,
|
||||
}
|
||||
|
||||
impl Default for TelemetryApi {
|
||||
@@ -58,353 +24,41 @@ impl Default for TelemetryApi {
|
||||
|
||||
impl TelemetryApi {
|
||||
pub fn new() -> Self {
|
||||
cfg_if::cfg_if! {
|
||||
if #[cfg(test)] {
|
||||
let client = http_client::Client::new_for_test();
|
||||
} else if #[cfg(target_family = "wasm")] {
|
||||
let client = http_client::Client::default();
|
||||
} else {
|
||||
use std::time::Duration;
|
||||
|
||||
let client = http_client::Client::from_client_builder(
|
||||
// We use our own http client directly instead of the Rudderstack SDK's because using
|
||||
// our own client gives us the ability to have universal hooks for pre/post
|
||||
// request/response logic.
|
||||
reqwest::Client::builder()
|
||||
// Don't allow insecure connections; they will be rejected by
|
||||
// the server with a 403 Forbidden.
|
||||
.https_only(true)
|
||||
// Keep idle connections in the pool for up to 55s. AWS
|
||||
// Application Load Balancers will drop idle connections after
|
||||
// 60s and the default pool idle timeout is 90s; a pool idle
|
||||
// timeout longer than the server timeout can lead to errors
|
||||
// upon trying to use an idle connection.
|
||||
.pool_idle_timeout(Duration::from_secs(55))
|
||||
.connect_timeout(Duration::from_secs(10)),
|
||||
).expect("Client should be constructed since we use a compatibility layer to use reqwest::Client");
|
||||
}
|
||||
Self {
|
||||
client: http_client::Client::default(),
|
||||
}
|
||||
|
||||
Self { client }
|
||||
}
|
||||
|
||||
// Batches up telemetry events from the global queue and sends a Message to the Rudderstack API.
|
||||
// Returns the number of events that were flushed.
|
||||
pub async fn flush_events(&self, settings_snapshot: PrivacySettingsSnapshot) -> Result<usize> {
|
||||
let events = galaxyui::telemetry::flush_events();
|
||||
let event_count = events.len();
|
||||
|
||||
#[cfg(not(target_family = "wasm"))]
|
||||
if FeatureFlag::SendTelemetryToFile.is_enabled() {
|
||||
self.persist_events_to_telemetry_log_file(events.clone())?;
|
||||
}
|
||||
|
||||
if ChannelState::is_release_bundle() || FeatureFlag::WithSandboxTelemetry.is_enabled() {
|
||||
self.send_batch_messages_to_rudder(
|
||||
events
|
||||
.into_iter()
|
||||
.map(Event::to_rudder_batch_message)
|
||||
.collect(),
|
||||
settings_snapshot,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
|
||||
Ok(event_count)
|
||||
}
|
||||
|
||||
/// Flushes events directly to Rudder that were previously written into a file at `path`
|
||||
/// (likely via a call to `write_events_to_disk`).
|
||||
pub async fn flush_persisted_events_to_rudder(
|
||||
&self,
|
||||
path: &Path,
|
||||
settings_snapshot: PrivacySettingsSnapshot,
|
||||
) -> Result<()> {
|
||||
if path.exists() {
|
||||
let file = File::open(path)?;
|
||||
let events: Vec<RudderBatchMessage> = serde_json::from_reader(file)?;
|
||||
if !events.is_empty() {
|
||||
let rudder_batch_messages = events
|
||||
.into_iter()
|
||||
.map(|message| RudderBatchMessageWithMetadata {
|
||||
message,
|
||||
// We don't persist any events that contain sensitive user data.
|
||||
contains_ugc: false,
|
||||
})
|
||||
.collect();
|
||||
self.send_batch_messages_to_rudder(rudder_batch_messages, settings_snapshot)
|
||||
.await?;
|
||||
log::info!("Successfully flushed events to rudder from disk");
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Writes the last `max_event_count` events into disk. This is useful for persisting events
|
||||
/// where we can't make a network call to Rudder (such as when the app quits). To flush these
|
||||
/// events to Rudder, call `flush_events_to_rudder_from_disk`.
|
||||
pub fn flush_and_persist_events(
|
||||
&self,
|
||||
max_event_count: usize,
|
||||
settings_snapshot: PrivacySettingsSnapshot,
|
||||
) -> Result<()> {
|
||||
self.flush_and_persist_events_at_path(
|
||||
max_event_count,
|
||||
settings_snapshot,
|
||||
rudder_event_file_path(),
|
||||
)
|
||||
}
|
||||
|
||||
fn flush_and_persist_events_at_path(
|
||||
&self,
|
||||
max_event_count: usize,
|
||||
settings_snapshot: PrivacySettingsSnapshot,
|
||||
path: impl AsRef<Path>,
|
||||
) -> Result<()> {
|
||||
if settings_snapshot.should_disable_telemetry() {
|
||||
log::info!("Not writing queued events to disk because telemetry is disabled.");
|
||||
return Result::Ok(());
|
||||
}
|
||||
log::info!("Writing queued events to disk because telemetry is enabled.");
|
||||
|
||||
let file = File::create(path)?;
|
||||
|
||||
let events = galaxyui::telemetry::flush_events();
|
||||
if events.len() > max_event_count {
|
||||
log::error!("More telemetry events in queue than the limit to persist")
|
||||
}
|
||||
|
||||
self.persist_events_at_path(&file, max_event_count, events)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn persist_events_at_path(
|
||||
&self,
|
||||
file: &File,
|
||||
max_event_count: usize,
|
||||
events: Vec<Event>,
|
||||
) -> Result<()> {
|
||||
let rudder_events_to_persist: Vec<_> = events
|
||||
.into_iter()
|
||||
.rev()
|
||||
.take(max_event_count)
|
||||
.map(TelemetryExt::to_rudder_batch_message)
|
||||
.filter_map(|message| (!message.contains_ugc).then_some(message.message))
|
||||
.collect();
|
||||
serde_json::to_writer(file, &rudder_events_to_persist)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(not(target_family = "wasm"))]
|
||||
fn persist_events_to_telemetry_log_file(&self, events: Vec<Event>) -> Result<()> {
|
||||
let log_directory = galaxy_logging::log_directory()?;
|
||||
let telemetry_file_path = log_directory.join(&*ChannelState::telemetry_file_name());
|
||||
|
||||
let file = OpenOptions::new()
|
||||
.create(true)
|
||||
.append(true)
|
||||
.open(&telemetry_file_path)?;
|
||||
|
||||
self.persist_events_at_path(&file, events.len(), events)
|
||||
}
|
||||
|
||||
/// Sends a `TelemetryEvent` to the Rudderstack API.
|
||||
pub async fn send_telemetry_event(
|
||||
&self,
|
||||
user_id: Option<UserUid>,
|
||||
anonymous_id: String,
|
||||
event: impl galaxy_core::telemetry::TelemetryEvent,
|
||||
settings_snapshot: PrivacySettingsSnapshot,
|
||||
) -> Result<()> {
|
||||
let event = galaxyui::telemetry::create_event(
|
||||
user_id.map(|uid| uid.as_string()),
|
||||
anonymous_id,
|
||||
event.name().into(),
|
||||
event.payload(),
|
||||
event.contains_ugc(),
|
||||
galaxyui::time::get_current_time(),
|
||||
);
|
||||
|
||||
self.send_telemetry_event_internal(event, settings_snapshot)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Internal implementation for sending telemetry events. This reduces code size, since
|
||||
// we:
|
||||
// 1. Return a boxed future, so calling `async` functions don't need to inline this one.
|
||||
// 2. Don't have to monomorphize for each telemetry event implementation.
|
||||
fn send_telemetry_event_internal(
|
||||
&self,
|
||||
event: Event,
|
||||
settings_snapshot: PrivacySettingsSnapshot,
|
||||
) -> impl Future<Output = Result<()>> + '_ {
|
||||
let work = async move {
|
||||
if settings_snapshot.should_disable_telemetry() {
|
||||
log::info!("Not sending telemetry event because telemetry is disabled.");
|
||||
return Result::Ok(());
|
||||
}
|
||||
|
||||
#[cfg(not(target_family = "wasm"))]
|
||||
if FeatureFlag::SendTelemetryToFile.is_enabled() {
|
||||
self.persist_events_to_telemetry_log_file(vec![event.clone()])?;
|
||||
}
|
||||
|
||||
if !(ChannelState::is_release_bundle()
|
||||
|| FeatureFlag::WithSandboxTelemetry.is_enabled())
|
||||
{
|
||||
return Result::Ok(());
|
||||
}
|
||||
|
||||
let rudder_batch = vec![event.to_rudder_batch_message()];
|
||||
|
||||
let result = self
|
||||
.send_batch_messages_to_rudder(rudder_batch, settings_snapshot)
|
||||
.await;
|
||||
|
||||
// This is only conditionally compiled because `is_connect` is not
|
||||
// available on wasm. If additional checks are made against the
|
||||
// `reqwest::Error`, this condition should be performed specifically
|
||||
// against `is_connect` and not the whole loop.
|
||||
#[cfg(not(target_family = "wasm"))]
|
||||
if let Err(error) = &result {
|
||||
for cause in error.chain() {
|
||||
if let Some(err) = cause.downcast_ref::<reqwest::Error>() {
|
||||
if err.is_connect() {
|
||||
log::warn!("Failed to send telemetry event: {error}");
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result
|
||||
};
|
||||
|
||||
// On WASM, the work future is non-Send, because the HTTP request future contains a reference to a JS
|
||||
// value (which is fine, since our WASM executor is single-threaded). On all other platforms, we must
|
||||
// return a Send future in order to use the background executor.
|
||||
cfg_if::cfg_if! {
|
||||
if #[cfg(target_family = "wasm")] {
|
||||
work.boxed_local()
|
||||
} else {
|
||||
work.boxed()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Send a batch of RudderStack messages to their HTTP API.
|
||||
/// Note that the rudderanalytics SDK provides a client, but we don't
|
||||
/// use it for a few reasons:
|
||||
/// 1. It only supports a blocking HTTP client instead of an async one
|
||||
/// 2. We want to use our own HTTP client which has before/after request logging hooks
|
||||
#[cfg_attr(target_family = "wasm", allow(clippy::question_mark))]
|
||||
async fn send_batch_messages_to_rudder(
|
||||
&self,
|
||||
messages: Vec<RudderBatchMessageWithMetadata>,
|
||||
settings_snapshot: PrivacySettingsSnapshot,
|
||||
) -> Result<()> {
|
||||
if messages.is_empty() {
|
||||
log::debug!("Dropping empty RudderStack telemetry batch");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if settings_snapshot.should_disable_telemetry() {
|
||||
log::info!("Not sending batched messages because telemetry is disabled.");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
log::info!("Start to send telemetry events to RudderStack");
|
||||
|
||||
let (mut messages_with_ugc, messages_without_ugc): (Vec<_>, Vec<_>) = messages
|
||||
.into_iter()
|
||||
.partition(|message| message.contains_ugc);
|
||||
|
||||
// If we shouldn't collect UGC telemetry, forceably clear any messages with UGC before trying to send.
|
||||
if !settings_snapshot.should_collect_ai_ugc_telemetry() {
|
||||
messages_with_ugc.clear();
|
||||
}
|
||||
|
||||
for (messages, rudder_stack_destination) in [
|
||||
(
|
||||
messages_with_ugc,
|
||||
ChannelState::rudderstack_ugc_destination(),
|
||||
),
|
||||
(
|
||||
messages_without_ugc,
|
||||
ChannelState::rudderstack_non_ugc_destination(),
|
||||
),
|
||||
] {
|
||||
if messages.is_empty() {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Note that timestamp and context are already included in the individual RudderBatchMessages
|
||||
// and these are the most important ones,
|
||||
// but we also add them to the RudderMessage::Batch wrapper.
|
||||
let rudder_message = RudderMessage::Batch(RudderBatch {
|
||||
batch: messages
|
||||
.into_iter()
|
||||
.map(|message| message.message)
|
||||
.collect(),
|
||||
original_timestamp: Some(Utc::now()),
|
||||
..Default::default()
|
||||
});
|
||||
if let Err(e) = self
|
||||
.send_rudder_request(rudder_message, rudder_stack_destination)
|
||||
.await
|
||||
{
|
||||
// Don't treat a connection issue as an error as these are outside of our control.
|
||||
//
|
||||
// This is only conditionally compiled because `is_connect` is not
|
||||
// available on wasm. If additional checks are made against the
|
||||
// `reqwest::Error`, this condition should be performed specifically
|
||||
// against `is_connect` and not the whole loop.
|
||||
#[cfg(not(target_family = "wasm"))]
|
||||
for cause in e.chain() {
|
||||
if let Some(err) = cause.downcast_ref::<reqwest::Error>() {
|
||||
if err.is_connect() {
|
||||
log::warn!("Failed to send event to RudderStack: {e}");
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
_user_id: Option<crate::auth::UserUid>,
|
||||
_anonymous_id: String,
|
||||
_event: impl galaxy_core::telemetry::TelemetryEvent,
|
||||
_settings_snapshot: crate::settings::PrivacySettingsSnapshot,
|
||||
) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Sends a POST request to the RudderStack HTTP API.
|
||||
async fn send_rudder_request(
|
||||
pub async fn flush_events(
|
||||
&self,
|
||||
mut msg: RudderMessage,
|
||||
rudder_stack_destination: RudderStackDestination,
|
||||
) -> Result<()> {
|
||||
msg.attach_context();
|
||||
_settings_snapshot: crate::settings::PrivacySettingsSnapshot,
|
||||
) -> anyhow::Result<usize> {
|
||||
Ok(0)
|
||||
}
|
||||
|
||||
let path = match msg {
|
||||
RudderMessage::Identify(_) => "/v1/identify",
|
||||
RudderMessage::Track(_) => "/v1/track",
|
||||
RudderMessage::Page(_) => "/v1/page",
|
||||
RudderMessage::Screen(_) => "/v1/screen",
|
||||
RudderMessage::Group(_) => "/v1/group",
|
||||
RudderMessage::Alias(_) => "/v1/alias",
|
||||
RudderMessage::Batch(_) => "/v1/batch",
|
||||
};
|
||||
|
||||
self.client
|
||||
.post(&format!("{}{}", rudder_stack_destination.root_url, path))
|
||||
.basic_auth(rudder_stack_destination.write_key, Some(""))
|
||||
.json(&msg)
|
||||
.send()
|
||||
.await?
|
||||
.error_for_status()?;
|
||||
pub async fn flush_persisted_events_to_rudder(
|
||||
&self,
|
||||
_path: &std::path::Path,
|
||||
_settings_snapshot: crate::settings::PrivacySettingsSnapshot,
|
||||
) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn flush_and_persist_events(
|
||||
&self,
|
||||
_max_event_count: usize,
|
||||
_settings_snapshot: crate::settings::PrivacySettingsSnapshot,
|
||||
) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "mod_tests.rs"]
|
||||
mod tests;
|
||||
|
||||
@@ -1,249 +1 @@
|
||||
//! Module that contains RudderStack API message types.
|
||||
//! This is directly copied from the RudderStack Rust SDK: https://github.com/rudderlabs/rudder-sdk-rust/blob/master/src/message.rs
|
||||
//! We do not use the SDK directly because it unconditionally uses a blocking HTTP client, which we don't want for a few reasons:
|
||||
//! 1. The blocking HTTP client is not allowed when compiling for WASM, so the crate itself cannot be compiled for WASM
|
||||
//! 2. An async HTTP client is more efficient
|
||||
//! 3. We want to use our own HTTP client which has before/after request logging hooks
|
||||
//! We can consider using the SDK if it adds support for an async HTTP client, tracked by this issue: https://github.com/rudderlabs/rudder-sdk-rust/issues/23
|
||||
use chrono::{DateTime, Utc};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::Value;
|
||||
|
||||
use crate::auth::UserUid;
|
||||
|
||||
/// An enum containing all values which may be sent to RudderStack's API.
|
||||
#[derive(PartialEq, Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(untagged)]
|
||||
pub enum Message {
|
||||
Identify(Identify),
|
||||
Track(Track),
|
||||
Page(Page),
|
||||
Screen(Screen),
|
||||
Group(Group),
|
||||
Alias(Alias),
|
||||
Batch(Batch),
|
||||
}
|
||||
|
||||
/// An identify event.
|
||||
#[derive(PartialEq, Debug, Clone, Serialize, Deserialize, Default)]
|
||||
pub struct Identify {
|
||||
/// The user id associated with this message.
|
||||
#[serde(rename = "userId", skip_serializing_if = "Option::is_none")]
|
||||
pub user_id: Option<UserUid>,
|
||||
|
||||
/// The anonymous user id associated with this message.
|
||||
#[serde(rename = "anonymousId", skip_serializing_if = "Option::is_none")]
|
||||
pub anonymous_id: Option<String>,
|
||||
|
||||
/// The traits to assign to the user.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub traits: Option<Value>,
|
||||
|
||||
/// The timestamp associated with this message.
|
||||
#[serde(rename = "originalTimestamp", skip_serializing_if = "Option::is_none")]
|
||||
pub original_timestamp: Option<DateTime<Utc>>,
|
||||
|
||||
/// Context associated with this message.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub context: Option<Value>,
|
||||
|
||||
/// Integrations to route this message to.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub integrations: Option<Value>,
|
||||
}
|
||||
|
||||
/// A track event.
|
||||
#[derive(PartialEq, Debug, Clone, Serialize, Deserialize, Default)]
|
||||
pub struct Track {
|
||||
/// The user id associated with this message.
|
||||
#[serde(rename = "userId", skip_serializing_if = "Option::is_none")]
|
||||
pub user_id: Option<UserUid>,
|
||||
|
||||
/// The anonymous user id associated with this message.
|
||||
#[serde(rename = "anonymousId", skip_serializing_if = "Option::is_none")]
|
||||
pub anonymous_id: Option<String>,
|
||||
|
||||
/// The name of the event being tracked.
|
||||
pub event: String,
|
||||
|
||||
/// The properties associated with the event.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub properties: Option<Value>,
|
||||
|
||||
/// The timestamp associated with this message.
|
||||
#[serde(rename = "originalTimestamp", skip_serializing_if = "Option::is_none")]
|
||||
pub original_timestamp: Option<DateTime<Utc>>,
|
||||
|
||||
/// Context associated with this message.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub context: Option<Value>,
|
||||
|
||||
/// Integrations to route this message to.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub integrations: Option<Value>,
|
||||
}
|
||||
|
||||
/// A page event.
|
||||
#[derive(PartialEq, Debug, Clone, Serialize, Deserialize, Default)]
|
||||
pub struct Page {
|
||||
/// The user id associated with this message.
|
||||
#[serde(rename = "userId", skip_serializing_if = "Option::is_none")]
|
||||
pub user_id: Option<UserUid>,
|
||||
|
||||
/// The anonymous user id associated with this message.
|
||||
#[serde(rename = "anonymousId", skip_serializing_if = "Option::is_none")]
|
||||
pub anonymous_id: Option<String>,
|
||||
|
||||
/// The name of the page being tracked.
|
||||
pub name: String,
|
||||
|
||||
/// The properties associated with the event.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub properties: Option<Value>,
|
||||
|
||||
/// The timestamp associated with this message.
|
||||
#[serde(rename = "originalTimestamp", skip_serializing_if = "Option::is_none")]
|
||||
pub original_timestamp: Option<DateTime<Utc>>,
|
||||
|
||||
/// Context associated with this message.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub context: Option<Value>,
|
||||
|
||||
/// Integrations to route this message to.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub integrations: Option<Value>,
|
||||
}
|
||||
|
||||
/// A screen event.
|
||||
#[derive(PartialEq, Debug, Clone, Serialize, Deserialize, Default)]
|
||||
pub struct Screen {
|
||||
/// The user id associated with this message.
|
||||
#[serde(rename = "userId", skip_serializing_if = "Option::is_none")]
|
||||
pub user_id: Option<UserUid>,
|
||||
|
||||
/// The anonymous user id associated with this message.
|
||||
#[serde(rename = "anonymousId", skip_serializing_if = "Option::is_none")]
|
||||
pub anonymous_id: Option<String>,
|
||||
|
||||
/// The name of the screen being tracked.
|
||||
pub name: String,
|
||||
|
||||
/// The properties associated with the event.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub properties: Option<Value>,
|
||||
|
||||
/// The timestamp associated with this message.
|
||||
#[serde(rename = "originalTimestamp", skip_serializing_if = "Option::is_none")]
|
||||
pub original_timestamp: Option<DateTime<Utc>>,
|
||||
|
||||
/// Context associated with this message.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub context: Option<Value>,
|
||||
|
||||
/// Integrations to route this message to.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub integrations: Option<Value>,
|
||||
}
|
||||
|
||||
/// A group event.
|
||||
#[derive(PartialEq, Debug, Clone, Serialize, Deserialize, Default)]
|
||||
pub struct Group {
|
||||
/// The user id associated with this message.
|
||||
#[serde(rename = "userId", skip_serializing_if = "Option::is_none")]
|
||||
pub user_id: Option<UserUid>,
|
||||
|
||||
/// The anonymous user id associated with this message.
|
||||
#[serde(rename = "anonymousId", skip_serializing_if = "Option::is_none")]
|
||||
pub anonymous_id: Option<String>,
|
||||
|
||||
/// The group the user is being associated with.
|
||||
#[serde(rename = "groupId")]
|
||||
pub group_id: String,
|
||||
|
||||
/// The traits to assign to the group.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub traits: Option<Value>,
|
||||
|
||||
/// The timestamp associated with this message.
|
||||
#[serde(rename = "originalTimestamp", skip_serializing_if = "Option::is_none")]
|
||||
pub original_timestamp: Option<DateTime<Utc>>,
|
||||
|
||||
/// Context associated with this message.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub context: Option<Value>,
|
||||
|
||||
/// Integrations to route this message to.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub integrations: Option<Value>,
|
||||
}
|
||||
|
||||
/// An alias event.
|
||||
#[derive(PartialEq, Debug, Clone, Serialize, Deserialize, Default)]
|
||||
pub struct Alias {
|
||||
/// The user id associated with this message.
|
||||
#[serde(rename = "userId")]
|
||||
pub user_id: UserUid,
|
||||
|
||||
/// The user's previous ID.
|
||||
#[serde(rename = "previousId")]
|
||||
pub previous_id: String,
|
||||
|
||||
/// The traits to assign to the alias.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub traits: Option<Value>,
|
||||
|
||||
/// The timestamp associated with this message.
|
||||
#[serde(rename = "originalTimestamp", skip_serializing_if = "Option::is_none")]
|
||||
pub original_timestamp: Option<DateTime<Utc>>,
|
||||
|
||||
/// Context associated with this message.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub context: Option<Value>,
|
||||
|
||||
/// Integrations to route this message to.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub integrations: Option<Value>,
|
||||
}
|
||||
|
||||
/// A batch of events.
|
||||
#[derive(PartialEq, Debug, Clone, Serialize, Deserialize, Default)]
|
||||
pub struct Batch {
|
||||
/// The batch of messages to send.
|
||||
pub batch: Vec<BatchMessageItem>,
|
||||
|
||||
/// Context associated with this message.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub context: Option<Value>,
|
||||
|
||||
/// Integrations to route this message to.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub integrations: Option<Value>,
|
||||
|
||||
/// The timestamp associated with this message.
|
||||
#[serde(rename = "originalTimestamp", skip_serializing_if = "Option::is_none")]
|
||||
pub original_timestamp: Option<DateTime<Utc>>,
|
||||
}
|
||||
|
||||
/// An enum containing all messages which may be placed inside a batch.
|
||||
#[derive(PartialEq, Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(tag = "type")]
|
||||
pub enum BatchMessageItem {
|
||||
#[serde(rename = "identify")]
|
||||
Identify(Identify),
|
||||
#[serde(rename = "track")]
|
||||
Track(Track),
|
||||
#[serde(rename = "page")]
|
||||
Page(Page),
|
||||
#[serde(rename = "screen")]
|
||||
Screen(Screen),
|
||||
#[serde(rename = "group")]
|
||||
Group(Group),
|
||||
#[serde(rename = "alias")]
|
||||
Alias(Alias),
|
||||
}
|
||||
|
||||
/// Metadata about a batch sent to Rudderstack and whether it contains user generated content.
|
||||
pub struct BatchMessage {
|
||||
pub message: BatchMessageItem,
|
||||
pub contains_ugc: bool,
|
||||
}
|
||||
|
||||
@@ -1,125 +1,7 @@
|
||||
//! Best-effort secret redaction for telemetry payloads.
|
||||
//!
|
||||
//! Unlike the AI-side secret redaction in `app/src/ai/blocklist/block/secret_redaction.rs`,
|
||||
//! which is gated on the user's secret-redaction (a.k.a. "safe mode") setting and is used
|
||||
//! for visual obfuscation in the terminal, the redaction in this module is unconditional:
|
||||
//! we always do a redaction pass on telemetry payloads that may contain user-generated
|
||||
//! content, regardless of the user's safe-mode setting. The two settings are deliberately
|
||||
//! decoupled — visual obfuscation is a UX preference, while telemetry-side redaction is a
|
||||
//! defence-in-depth measure for data leaving the device.
|
||||
//!
|
||||
//! The regex used for redaction always includes the default patterns defined in
|
||||
//! `crate::terminal::model::secrets::regexes::DEFAULT_REGEXES_WITH_NAMES`. Any custom
|
||||
//! patterns the user has configured (or that their organization has configured via
|
||||
//! enterprise secret redaction) are layered on top of those defaults.
|
||||
//!
|
||||
//! This module is intentionally lightweight: it does byte-range matching only and does
|
||||
//! not track `SecretLevel`s or character ranges, since the telemetry path doesn't need
|
||||
//! either.
|
||||
use crate::terminal::model::secrets::regexes::DEFAULT_REGEXES_WITH_NAMES;
|
||||
use lazy_static::lazy_static;
|
||||
use parking_lot::RwLock;
|
||||
use regex_automata::meta::Regex;
|
||||
use serde_json::Value;
|
||||
use std::collections::HashSet;
|
||||
use std::ops::Range;
|
||||
const REDACTION_REPLACEMENT_CHARACTER: &str = "*";
|
||||
lazy_static! {
|
||||
/// Regex used to redact secrets from telemetry payloads. Initialized with the
|
||||
/// default patterns so that redaction works even before the user's privacy
|
||||
/// settings are loaded (and even for users who have never configured any
|
||||
/// custom patterns).
|
||||
static ref TELEMETRY_SECRETS_REGEX: RwLock<Regex> = RwLock::new(build_default_regex());
|
||||
use regex::Regex;
|
||||
|
||||
pub fn update_telemetry_secrets_regex<'a>(
|
||||
_user_secrets: impl Iterator<Item = &'a Regex>,
|
||||
_enterprise_secrets: impl Iterator<Item = &'a Regex>,
|
||||
) {
|
||||
}
|
||||
/// Builds a regex containing only the default patterns. Used to seed the static
|
||||
/// regex before the privacy settings are loaded.
|
||||
fn build_default_regex() -> Regex {
|
||||
let patterns: Vec<&str> = DEFAULT_REGEXES_WITH_NAMES
|
||||
.iter()
|
||||
.map(|d| d.pattern)
|
||||
.collect();
|
||||
Regex::new_many(&patterns).expect("default secret patterns should compile")
|
||||
}
|
||||
/// Rebuilds [`TELEMETRY_SECRETS_REGEX`] from the user's and enterprise's secret
|
||||
/// regex lists, layered on top of the default patterns. The default patterns are
|
||||
/// always included, so redaction works even when the user has not configured any
|
||||
/// custom patterns.
|
||||
pub fn update_telemetry_secrets_regex<'a, U, E>(user_secrets: U, enterprise_secrets: E)
|
||||
where
|
||||
U: IntoIterator<Item = &'a regex::Regex>,
|
||||
E: IntoIterator<Item = &'a regex::Regex>,
|
||||
{
|
||||
let patterns = compose_patterns(
|
||||
user_secrets.into_iter().map(regex::Regex::as_str),
|
||||
enterprise_secrets.into_iter().map(regex::Regex::as_str),
|
||||
);
|
||||
match Regex::new_many(&patterns) {
|
||||
Ok(regex) => *TELEMETRY_SECRETS_REGEX.write() = regex,
|
||||
Err(err) => log::error!("Failed to build telemetry secrets regex: {err:?}"),
|
||||
}
|
||||
}
|
||||
/// Composes the full list of patterns to compile into the telemetry regex,
|
||||
/// ordered enterprise → user → defaults, with later occurrences of an already-
|
||||
/// seen pattern string deduped out.
|
||||
fn compose_patterns<'a>(
|
||||
user: impl Iterator<Item = &'a str>,
|
||||
enterprise: impl Iterator<Item = &'a str>,
|
||||
) -> Vec<&'a str> {
|
||||
let mut seen: HashSet<&str> = HashSet::new();
|
||||
let mut patterns: Vec<&str> = Vec::new();
|
||||
let all = enterprise
|
||||
.chain(user)
|
||||
.chain(DEFAULT_REGEXES_WITH_NAMES.iter().map(|d| d.pattern));
|
||||
for pattern in all {
|
||||
if seen.insert(pattern) {
|
||||
patterns.push(pattern);
|
||||
}
|
||||
}
|
||||
patterns
|
||||
}
|
||||
/// Replaces every detected secret in `input` with a run of asterisks of the same
|
||||
/// byte length. Overlapping matches (which can occur when multiple patterns match
|
||||
/// the same region) are merged before replacement, so each character is replaced
|
||||
/// at most once.
|
||||
pub fn redact_secrets_in_string(input: &mut String) {
|
||||
let ranges: Vec<Range<usize>> = {
|
||||
let regex = TELEMETRY_SECRETS_REGEX.read();
|
||||
regex.find_iter(input.as_str()).map(|m| m.range()).collect()
|
||||
};
|
||||
replace_byte_ranges_with_asterisks(input, ranges);
|
||||
}
|
||||
/// Replaces each byte range in `input` with a run of asterisks of the same byte
|
||||
/// length. Handles overlapping ranges by merging them first, and replaces from
|
||||
/// the end of the string so earlier byte indices stay valid as we mutate.
|
||||
fn replace_byte_ranges_with_asterisks(input: &mut String, mut ranges: Vec<Range<usize>>) {
|
||||
if ranges.is_empty() {
|
||||
return;
|
||||
}
|
||||
// Sort and merge overlapping ranges so we don't double-replace.
|
||||
ranges.sort_by_key(|r| r.start);
|
||||
let mut merged: Vec<Range<usize>> = Vec::with_capacity(ranges.len());
|
||||
for range in ranges {
|
||||
match merged.last_mut() {
|
||||
Some(last) if range.start <= last.end => last.end = last.end.max(range.end),
|
||||
_ => merged.push(range),
|
||||
}
|
||||
}
|
||||
// Replace from the end of the string so earlier byte indices stay valid.
|
||||
for range in merged.into_iter().rev() {
|
||||
let len = range.end - range.start;
|
||||
input.replace_range(range, &REDACTION_REPLACEMENT_CHARACTER.repeat(len));
|
||||
}
|
||||
}
|
||||
/// Walks a [`Value`] and runs [`redact_secrets_in_string`] on every string within
|
||||
/// it. Non-string scalars (numbers, booleans, nulls) are left untouched.
|
||||
pub fn redact_secrets_in_value(value: &mut Value) {
|
||||
match value {
|
||||
Value::String(s) => redact_secrets_in_string(s),
|
||||
Value::Array(arr) => arr.iter_mut().for_each(redact_secrets_in_value),
|
||||
Value::Object(obj) => obj.values_mut().for_each(redact_secrets_in_value),
|
||||
Value::Null | Value::Bool(_) | Value::Number(_) => {}
|
||||
}
|
||||
}
|
||||
#[cfg(test)]
|
||||
#[path = "secret_redaction_tests.rs"]
|
||||
mod tests;
|
||||
|
||||
@@ -1,129 +1 @@
|
||||
use super::telemetry::rudder_message::{
|
||||
BatchMessage as RudderBatchMessage, BatchMessageItem as RudderBatchMessageItem,
|
||||
Identify as RudderIdentify, Track as RudderTrack,
|
||||
};
|
||||
use super::telemetry::secret_redaction::redact_secrets_in_value;
|
||||
use crate::auth::UserUid;
|
||||
use chrono::{DateTime, Utc};
|
||||
use galaxy_core::{
|
||||
channel::{Channel, ChannelState},
|
||||
execution_mode,
|
||||
};
|
||||
use galaxyui::telemetry::EventPayload;
|
||||
use serde_json::{json, Value};
|
||||
|
||||
use super::telemetry::telemetry_context;
|
||||
|
||||
pub trait TelemetryExt {
|
||||
fn to_rudder_batch_message(self) -> RudderBatchMessage;
|
||||
}
|
||||
|
||||
impl TelemetryExt for galaxyui::telemetry::Event {
|
||||
fn to_rudder_batch_message(self) -> RudderBatchMessage {
|
||||
let message = match self.payload {
|
||||
EventPayload::IdentifyUser {
|
||||
user_id,
|
||||
anonymous_id,
|
||||
} => RudderBatchMessageItem::Identify(RudderIdentify {
|
||||
user_id: Some(UserUid::new(user_id.as_str())),
|
||||
anonymous_id: Some(anonymous_id),
|
||||
original_timestamp: Some(self.timestamp),
|
||||
integrations: Some(json!({
|
||||
"Amplitude": {
|
||||
"session_id": self.session_created_at.timestamp(),
|
||||
}
|
||||
})),
|
||||
context: Some(telemetry_context().as_value()),
|
||||
..Default::default()
|
||||
}),
|
||||
EventPayload::AppActive {
|
||||
user_id,
|
||||
anonymous_id,
|
||||
} => form_rudder_track_message(
|
||||
user_id.map(|uid| UserUid::new(uid.as_str())),
|
||||
anonymous_id,
|
||||
"Active App Usage".to_string(),
|
||||
None,
|
||||
self.timestamp,
|
||||
self.session_created_at,
|
||||
),
|
||||
EventPayload::NamedEvent {
|
||||
user_id,
|
||||
anonymous_id,
|
||||
name,
|
||||
mut value,
|
||||
} => {
|
||||
// For events that may contain user-generated content, run a
|
||||
// best-effort secret-redaction pass on the payload before
|
||||
// sending. This is independent of the user's safe-mode setting:
|
||||
// visual obfuscation is a UX preference, while telemetry-side
|
||||
// redaction is a defence-in-depth measure for data leaving the
|
||||
// device. See `secret_redaction.rs` for details.
|
||||
if self.contains_ugc {
|
||||
if let Some(value) = value.as_mut() {
|
||||
redact_secrets_in_value(value);
|
||||
}
|
||||
}
|
||||
form_rudder_track_message(
|
||||
user_id.map(|uid| UserUid::new(uid.as_str())),
|
||||
anonymous_id,
|
||||
name.to_string(),
|
||||
value,
|
||||
self.timestamp,
|
||||
self.session_created_at,
|
||||
)
|
||||
}
|
||||
};
|
||||
|
||||
RudderBatchMessage {
|
||||
message,
|
||||
contains_ugc: self.contains_ugc,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn form_rudder_track_message(
|
||||
user_id: Option<UserUid>,
|
||||
anonymous_id: String,
|
||||
name: String,
|
||||
payload: Option<Value>,
|
||||
timestamp: DateTime<Utc>,
|
||||
session_created_at: DateTime<Utc>,
|
||||
) -> RudderBatchMessageItem {
|
||||
RudderBatchMessageItem::Track(RudderTrack {
|
||||
user_id,
|
||||
anonymous_id: Some(anonymous_id),
|
||||
event: name,
|
||||
properties: Some(json!({
|
||||
"release_mode": release_mode(ChannelState::channel()),
|
||||
"tag": ChannelState::app_version().unwrap_or("<no tag>"),
|
||||
"client_id": execution_mode::current_client_id(),
|
||||
"payload": payload
|
||||
})),
|
||||
original_timestamp: Some(timestamp),
|
||||
integrations: Some(json!({
|
||||
"Amplitude": {
|
||||
"session_id": session_created_at.timestamp(),
|
||||
}
|
||||
})),
|
||||
context: Some(telemetry_context().as_value()),
|
||||
})
|
||||
}
|
||||
|
||||
fn release_mode(channel: Channel) -> &'static str {
|
||||
match channel {
|
||||
Channel::Stable => "stable_release",
|
||||
Channel::Preview => "preview_release",
|
||||
Channel::Local => "local",
|
||||
Channel::Integration => "integration_test",
|
||||
Channel::Dev => "dev_release",
|
||||
// We don't ever expect to send telemetry for the OSS build, but
|
||||
// until we have some time to clean things up here, we'll set a valid
|
||||
// value that we never intend to receive.
|
||||
Channel::Oss => "oss_release",
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "telemetry_ext_tests.rs"]
|
||||
mod tests;
|
||||
|
||||
Reference in New Issue
Block a user