yindo/posthog
1
0
Fork 0
mirror of https://github.com/BillyOutlast/posthog.git synced 2026-02-04 03:01:23 +01:00
Code Issues Packages Projects Releases Wiki Activity

feat(cli): improve error handling / logging (#40995)

Browse Source 
Signed-off-by: Hugues Pouillot <hpouillot@gmail.com>
This commit is contained in:
Hugues Pouillot
2025-11-06 13:50:29 +01:00
committed by GitHub
parent ce1b794687
commit 1ad23f48fe
21 changed files with 414 additions and 302 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
cli/CHANGELOG.md
View File

@@ -1,5 +1,10 @@
# posthog-cli
# 0.5.9
- Improve error handling from api
- Reduce logs for sourcemap processing
# 0.5.8
- Adding experimental support for proguard mappings

2
cli/Cargo.lock generated
View File

@@ -1521,7 +1521,7 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
[[package]]
name = "posthog-cli"
version = "0.5.8"
version = "0.5.9"
dependencies = [
"anyhow",
"chrono",

2
cli/Cargo.toml
View File

@@ -1,6 +1,6 @@
[package]
name = "posthog-cli"
version = "0.5.8"
version = "0.5.9"
authors = [
"David <david@posthog.com>",
"Olly <oliver@posthog.com>",

211
cli/src/api/client.rs Normal file
View File

@@ -0,0 +1,211 @@
use std::fmt::Display;
use anyhow::{Context, Result};
use reqwest::{
blocking::{Client, RequestBuilder, Response},
header::{HeaderMap, HeaderValue},
Method, Url,
};
use serde::{Deserialize, Serialize};
use thiserror::Error;
use tracing::debug;
use crate::invocation_context::InvocationConfig;
#[derive(Clone)]
pub struct PHClient {
config: InvocationConfig,
base_url: Url,
client: Client,
}
#[derive(Error, Debug)]
pub enum ClientError {
RequestError(reqwest::Error),
// All invalid status codes
ApiError(u16, Box<Url>, String),
InvalidUrl(String, String),
}
impl Display for ClientError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
ClientError::RequestError(err) => write!(f, "Request error: {err}"),
ClientError::InvalidUrl(base_url, path) => {
write!(f, "Failed to build URL: {base_url} {path}")
}
ClientError::ApiError(status, url, body) => {
// We only parse api error on display to catch all errors even when the body is not JSON
match serde_json::from_str::<ApiErrorResponse>(body) {
Ok(api_error) => {
write!(f, "API error: {api_error}")
}
Err(_) => write!(
f,
"API error: status='{status}' url='{url}' message='{body}'",
),
}
}
}
}
}
impl From<reqwest::Error> for ClientError {
fn from(error: reqwest::Error) -> Self {
ClientError::RequestError(error)
}
}
#[derive(Serialize, Deserialize, Debug)]
pub struct ApiErrorResponse {
r#type: String,
code: String,
detail: String,
attr: Option<String>,
}
impl Display for ApiErrorResponse {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"error='{}' code='{}' details='{}'",
self.r#type, self.code, self.detail
)?;
if let Some(attr) = &self.attr {
write!(f, ", attributes='{attr}'")?;
}
Ok(())
}
}
pub trait SendRequestFn: FnOnce(RequestBuilder) -> RequestBuilder {}
impl PHClient {
pub fn from_config(config: InvocationConfig) -> anyhow::Result<Self> {
let base_url = Self::build_base_url(&config)?;
let client = Self::build_client(&config)?;
Ok(Self {
config,
base_url,
client,
})
}
pub fn get(&self, path: &str) -> Result<RequestBuilder, ClientError> {
self.create_request(Method::GET, path)
}
pub fn post(&self, path: &str) -> Result<RequestBuilder, ClientError> {
self.create_request(Method::POST, path)
}
pub fn put(&self, path: &str) -> Result<RequestBuilder, ClientError> {
self.create_request(Method::PUT, path)
}
pub fn delete(&self, path: &str) -> Result<RequestBuilder, ClientError> {
self.create_request(Method::DELETE, path)
}
pub fn patch(&self, path: &str) -> Result<RequestBuilder, ClientError> {
self.create_request(Method::PATCH, path)
}
pub fn send_get<F: FnOnce(RequestBuilder) -> RequestBuilder>(
&self,
path: &str,
builder: F,
) -> Result<Response, ClientError> {
self.send_request(Method::GET, path, builder)
}
pub fn send_post<F: FnOnce(RequestBuilder) -> RequestBuilder>(
&self,
path: &str,
builder: F,
) -> Result<Response, ClientError> {
self.send_request(Method::POST, path, builder)
}
pub fn send_delete<F: FnOnce(RequestBuilder) -> RequestBuilder>(
&self,
path: &str,
builder: F,
) -> Result<Response, ClientError> {
self.send_request(Method::DELETE, path, builder)
}
pub fn send_put<F: FnOnce(RequestBuilder) -> RequestBuilder>(
&self,
path: &str,
builder: F,
) -> Result<Response, ClientError> {
self.send_request(Method::PUT, path, builder)
}
pub fn send_request<F: FnOnce(RequestBuilder) -> RequestBuilder>(
&self,
method: Method,
path: &str,
builder: F,
) -> Result<Response, ClientError> {
let request = builder(self.create_request(method, path)?);
match request.send() {
Ok(response) => {
if response.status().is_success() {
Ok(response)
} else {
let status = response.status().as_u16();
let box_url = Box::new(response.url().clone());
let body = response.text()?;
Err(ClientError::ApiError(status, box_url, body))
}
}
Err(err) => Err(ClientError::from(err)),
}
}
pub fn get_env_id(&self) -> &String {
&self.config.env_id
}
fn create_request(&self, method: Method, path: &str) -> Result<RequestBuilder, ClientError> {
let url = self.build_url(path)?;
let headers = self.build_headers();
debug!("building request for {method} {url}");
Ok(self
.client
.request(method, url)
.bearer_auth(&self.config.api_key)
.headers(headers))
}
fn build_client(config: &InvocationConfig) -> anyhow::Result<Client> {
let client = Client::builder()
.danger_accept_invalid_certs(config.skip_ssl)
.build()?;
Ok(client)
}
fn build_base_url(config: &InvocationConfig) -> anyhow::Result<Url> {
let base_url = Url::parse(&format!(
"{}/api/environments/{}/",
config.host, config.env_id
))
.context("Invalid base URL")?;
Ok(base_url)
}
fn build_headers(&self) -> HeaderMap {
let mut headers = HeaderMap::new();
headers.insert("Content-Type", HeaderValue::from_static("application/json"));
headers.insert("User-Agent", HeaderValue::from_static("posthog-cli"));
headers
}
fn build_url(&self, path: &str) -> Result<Url, ClientError> {
self.base_url
.join(path)
.map_err(|_| ClientError::InvalidUrl(self.base_url.clone().into(), path.to_string()))
}
}

1
cli/src/api/mod.rs
View File

@@ -1,2 +1,3 @@
pub mod client;
pub mod releases;
pub mod symbol_sets;

72
cli/src/api/releases.rs
View File

@@ -1,12 +1,13 @@
use std::collections::HashMap;
use anyhow::Result;
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tracing::{info, warn};
use uuid::Uuid;
use crate::{
api::client::ClientError,
invocation_context::context,
utils::{files::content_hash, git::GitInfo},
};
@@ -37,35 +38,23 @@ struct CreateReleaseRequest {
}
impl Release {
pub fn lookup(project: &str, version: &str) -> Result<Option<Self>> {
pub fn lookup(project: &str, version: &str) -> Result<Option<Self>, ClientError> {
let hash_id = content_hash([project, version]);
let context = context();
let token = &context.token;
let client = &context().client;
let url = format!(
"{}/api/environments/{}/error_tracking/releases/hash/{hash_id}",
token.get_host(),
token.env_id,
);
let path = format!("error_tracking/releases/hash/{hash_id}");
let response = client.send_get(&path, |req| req);
let response = context
.client
.get(&url)
.header("Authorization", format!("Bearer {}", token.token))
.header("Content-Type", "application/json")
.send()?;
if response.status().as_u16() == 404 {
warn!("Release {} of project {} not found", version, project);
return Ok(None);
}
if response.status().is_success() {
info!("Found release {} of project {}", version, project);
Ok(Some(response.json()?))
if let Err(err) = response {
if let ClientError::ApiError(404, _, _) = err {
warn!("release {} of project {} not found", version, project);
return Ok(None);
}
warn!("failed to get release from hash: {}", err);
Err(err)
} else {
response.error_for_status()?;
Ok(None) // unreachable
info!("found release {} of project {}", version, project);
Ok(Some(response.unwrap().json()?))
}
}
}
@@ -157,7 +146,6 @@ impl ReleaseBuilder {
let hash_id = content_hash([project.as_bytes(), version.as_bytes()]);
let token = &context().token;
let request = CreateReleaseRequest {
metadata,
hash_id,
@@ -165,31 +153,17 @@ impl ReleaseBuilder {
project,
};
let url = format!(
"{}/api/environments/{}/error_tracking/releases",
token.get_host(),
token.env_id
);
let client = &context().client;
let response = client
.post(&url)
.header("Authorization", format!("Bearer {}", token.token))
.header("Content-Type", "application/json")
.json(&request)
.send()?;
.send_post("error_tracking/releases", |req| req.json(&request))
.context("Failed to create release")?;
if response.status().is_success() {
let response = response.json::<Release>()?;
info!(
"Release {} of {} created successfully! {}",
request.version, request.project, response.id
);
Ok(response)
} else {
let e = response.text()?;
Err(anyhow::anyhow!("Failed to create release: {e}"))
}
let response = response.json::<Release>()?;
info!(
"Release {} of {} created successfully! {}",
request.version, request.project, response.id
);
Ok(response)
}
}

43
cli/src/api/symbol_sets.rs
View File

@@ -97,15 +97,7 @@ pub fn upload(input_sets: &[SymbolSetUpload], batch_size: usize) -> Result<()> {
}
fn start_upload(symbol_sets: &[&SymbolSetUpload]) -> Result<BulkUploadStartResponse> {
let base_url = format!(
"{}/api/environments/{}/error_tracking/symbol_sets",
context().token.get_host(),
context().token.env_id
);
let client = &context().client;
let auth_token = &context().token.token;
let start_upload_url: String = format!("{}{}", base_url, "/bulk_start_upload");
let request = BulkUploadStartRequest {
symbol_sets: symbol_sets
@@ -115,19 +107,16 @@ fn start_upload(symbol_sets: &[&SymbolSetUpload]) -> Result<BulkUploadStartRespo
};
let res = client
.post(&start_upload_url)
.header("Authorization", format!("Bearer {auth_token}"))
.json(&request)
.send()
.context(format!("While starting upload to {start_upload_url}"))?;
let res = raise_for_err(res)?;
.send_post("error_tracking/symbol_sets/bulk_start_upload", |req| {
req.json(&request)
})
.context("Failed to start upload")?;
Ok(res.json()?)
}
fn upload_to_s3(presigned_url: PresignedUrl, data: &[u8]) -> Result<()> {
let client = &context().client;
let client = &context().build_http_client()?;
let mut last_err = None;
let mut delay = std::time::Duration::from_millis(500);
for attempt in 1..=3 {
@@ -161,26 +150,14 @@ fn upload_to_s3(presigned_url: PresignedUrl, data: &[u8]) -> Result<()> {
}
fn finish_upload(content_hashes: HashMap<String, String>) -> Result<()> {
let base_url = format!(
"{}/api/environments/{}/error_tracking/symbol_sets",
context().token.get_host(),
context().token.env_id
);
let client = &context().client;
let auth_token = &context().token.token;
let finish_upload_url: String = format!("{}/{}", base_url, "bulk_finish_upload");
let request = BulkUploadFinishRequest { content_hashes };
let res = client
.post(finish_upload_url)
.header("Authorization", format!("Bearer {auth_token}"))
.header("Content-Type", "application/json")
.json(&request)
.send()
.context(format!("While finishing upload to {base_url}"))?;
raise_for_err(res)?;
client
.send_post("error_tracking/symbol_sets/bulk_finish_upload", |req| {
req.json(&request)
})
.context("Failed to finish upload")?;
Ok(())
}

14
cli/src/experimental/query/command.rs
View File

@@ -43,9 +43,6 @@ pub enum QueryCommand {
}
pub fn query_command(query: &QueryCommand) -> Result<(), Error> {
let creds = context().token.clone();
let host = creds.get_host();
match query {
QueryCommand::Editor {
no_print,
@@ -54,13 +51,12 @@ pub fn query_command(query: &QueryCommand) -> Result<(), Error> {
} => {
// Given this is an interactive command, we're happy enough to not join the capture handle
context().capture_command_invoked("query_editor");
let res = start_query_editor(&host, creds.clone(), *debug)?;
let res = start_query_editor(*debug)?;
if !no_print {
println!("Final query: {res}");
}
if *execute {
let query_endpoint = format!("{}/api/environments/{}/query", host, creds.env_id);
let res = run_query(&query_endpoint, &creds.token, &res)??;
let res = run_query(&res)??;
for result in res.results {
println!("{}", serde_json::to_string(&result)?);
}
@@ -69,8 +65,7 @@ pub fn query_command(query: &QueryCommand) -> Result<(), Error> {
QueryCommand::Run { query, debug } => {
// Given this is an interactive command, we're happy enough to not join the capture handle
context().capture_command_invoked("query_run");
let query_endpoint = format!("{}/api/environments/{}/query", host, creds.env_id);
let res = run_query(&query_endpoint, &creds.token, query)??;
let res = run_query(query)??;
if *debug {
println!("{}", serde_json::to_string_pretty(&res)?);
} else {
@@ -81,8 +76,7 @@ pub fn query_command(query: &QueryCommand) -> Result<(), Error> {
}
QueryCommand::Check { query, raw } => {
context().capture_command_invoked("query_check");
let query_endpoint = format!("{}/api/environments/{}/query", host, creds.env_id);
let res = check_query(&query_endpoint, &creds.token, query)?;
let res = check_query(query)?;
if *raw {
println!("{}", serde_json::to_string_pretty(&res)?);
} else {

23
cli/src/experimental/query/mod.rs
View File

@@ -4,6 +4,8 @@ use anyhow::Error;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::invocation_context::context;
pub mod command;
// TODO - we could formalise a lot of this and move it into posthog-rs, tbh
@@ -157,9 +159,8 @@ where
}
}
pub fn run_query(endpoint: &str, token: &str, to_run: &str) -> Result<HogQLQueryResult, Error> {
let client = reqwest::blocking::Client::new();
pub fn run_query(to_run: &str) -> Result<HogQLQueryResult, Error> {
let client = &context().client;
let request = QueryRequest {
query: Query::HogQLQuery {
query: to_run.to_string(),
@@ -167,11 +168,7 @@ pub fn run_query(endpoint: &str, token: &str, to_run: &str) -> Result<HogQLQuery
refresh: Some(QueryRefresh::Blocking),
};
let response = client
.post(endpoint)
.json(&request)
.bearer_auth(token)
.send()?;
let response = client.post("query")?.json(&request).send()?;
let code = response.status();
let body = response.text()?;
@@ -188,8 +185,8 @@ pub fn run_query(endpoint: &str, token: &str, to_run: &str) -> Result<HogQLQuery
Ok(Ok(response))
}
pub fn check_query(endpoint: &str, token: &str, to_run: &str) -> Result<MetadataResponse, Error> {
let client = reqwest::blocking::Client::new();
pub fn check_query(to_run: &str) -> Result<MetadataResponse, Error> {
let client = &context().client;
let query = MetadataQuery {
language: MetadataLanguage::HogQL,
@@ -204,11 +201,7 @@ pub fn check_query(endpoint: &str, token: &str, to_run: &str) -> Result<Metadata
refresh: None,
};
let response = client
.post(endpoint)
.json(&request)
.bearer_auth(token)
.send()?;
let response = client.post("query")?.json(&request).send()?;
let code = response.status();
let body = response.text()?;

33
cli/src/experimental/schema.rs
View File

@@ -6,6 +6,7 @@ use std::fs;
use std::path::Path;
use tracing::info;
use crate::api::client::PHClient;
use crate::invocation_context::context;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
@@ -130,15 +131,14 @@ pub fn pull(_host: Option<String>, output_override: Option<String>) -> Result<()
language.display_name()
);
// Load credentials
let token = context().token.clone();
let host = token.get_host();
// Get PH client
let client = &context().client;
// Determine output path
let output_path = determine_output_path(language, output_override)?;
// Fetch definitions from the server
let response = fetch_definitions(&host, &token.env_id, &token.token, language)?;
let response = fetch_definitions(client, language)?;
info!(
"✓ Fetched {} definitions for {} events",
@@ -263,14 +263,14 @@ pub fn status() -> Result<()> {
println!("\nPostHog Schema Sync Status\n");
println!("Authentication:");
let token = context().token.clone();
let config = context().config.clone();
println!(" ✓ Authenticated");
println!(" Host: {}", token.get_host());
println!(" Project ID: {}", token.env_id);
println!(" Host: {}", config.host);
println!(" Project ID: {}", config.env_id);
let masked_token = format!(
"{}****{}",
&token.token[..4],
&token.token[token.token.len() - 4..]
&config.api_key[..4],
&config.api_key[config.api_key.len() - 4..]
);
println!(" Token: {masked_token}");
@@ -311,21 +311,14 @@ pub fn status() -> Result<()> {
Ok(())
}
fn fetch_definitions(
host: &str,
env_id: &str,
token: &str,
language: Language,
) -> Result<DefinitionsResponse> {
fn fetch_definitions(client: &PHClient, language: Language) -> Result<DefinitionsResponse> {
let url = format!(
"{}/api/projects/{}/event_definitions/{}/",
host,
env_id,
"/api/projects/{}/event_definitions/{}/",
client.get_env_id(),
language.as_str()
);
let client = &context().client;
let response = client.get(&url).bearer_auth(token).send().context(format!(
let response = client.get(&url)?.send().context(format!(
"Failed to fetch {} definitions",
language.display_name()
))?;

35
cli/src/experimental/tasks/list.rs
View File

@@ -1,14 +1,14 @@
use anyhow::{Context, Result};
use reqwest::blocking::Client;
use std::collections::VecDeque;
use crate::{
api::client::PHClient,
experimental::tasks::{
utils::{fetch_stages, fetch_tasks, fetch_workflows},
Task,
},
invocation_context::context,
utils::{auth::Token, raise_for_err},
utils::raise_for_err,
};
use super::{TaskListResponse, TaskWorkflow, WorkflowStage};
@@ -16,15 +16,13 @@ use super::{TaskListResponse, TaskWorkflow, WorkflowStage};
const BUFFER_SIZE: usize = 50;
pub struct TaskIterator {
client: Client,
token: Token,
client: PHClient,
buffer: VecDeque<Task>,
next_url: Option<String>,
}
impl TaskIterator {
pub fn new(client: Client, host: String, token: Token, offset: Option<usize>) -> Result<Self> {
let initial_url = format!("{}/api/environments/{}/tasks/", host, token.env_id);
pub fn new(client: PHClient, offset: Option<usize>) -> Result<Self> {
let mut params = vec![];
params.push(("limit", BUFFER_SIZE.to_string()));
if let Some(offset) = offset {
@@ -32,13 +30,8 @@ impl TaskIterator {
}
let response = client
.get(&initial_url)
.query(&params)
.header("Authorization", format!("Bearer {}", token.token))
.send()
.context("Failed to send request")?;
let response = raise_for_err(response)?;
.send_get("tasks", |req| req.query(&params))
.context("Failed to get tasks")?;
let task_response: TaskListResponse = response
.json()
@@ -49,7 +42,6 @@ impl TaskIterator {
Ok(Self {
client,
token,
buffer,
next_url: task_response.next,
})
@@ -64,8 +56,7 @@ impl TaskIterator {
let response = self
.client
.get(&url)
.header("Authorization", format!("Bearer {}", self.token.token))
.get(&url)?
.send()
.context("Failed to send request")?;
@@ -164,19 +155,15 @@ pub fn print_task(task: &Task, workflows: &[TaskWorkflow], stages: &[WorkflowSta
}
pub fn list_tasks(limit: Option<&usize>, offset: Option<&usize>) -> Result<()> {
let token = context().token.clone();
let host = token.get_host();
let client = context().client.clone();
let client = &context().client;
let workflows: Result<Vec<TaskWorkflow>> =
fetch_workflows(client.clone(), host.clone(), token.clone())?.collect();
let workflows: Result<Vec<TaskWorkflow>> = fetch_workflows(client.clone())?.collect();
let workflows = workflows?;
let stages: Result<Vec<WorkflowStage>> =
fetch_stages(client.clone(), host.clone(), token.clone())?.collect();
let stages: Result<Vec<WorkflowStage>> = fetch_stages(client.clone())?.collect();
let stages = stages?;
let tasks = fetch_tasks(client, host, token, offset.cloned())?;
let tasks = fetch_tasks(client.clone(), offset.cloned())?;
let task_iter: Box<dyn Iterator<Item = Result<Task>>> = if let Some(limit) = limit {
Box::new(tasks.take(*limit))

12
cli/src/experimental/tasks/progress.rs
View File

@@ -55,21 +55,13 @@ pub fn show_progress(task_id: Option<&Uuid>) -> Result<()> {
}
fn fetch_progress(task_id: &Uuid) -> Result<TaskProgressResponse> {
let token = context().token.clone();
let host = token.get_host();
let client = context().client.clone();
let url = format!(
"{}/api/environments/{}/tasks/{}/progress/",
host, token.env_id, task_id
);
let path = format!("tasks/{task_id}/progress/");
let response = client
.get(&url)
.header("Authorization", format!("Bearer {}", token.token))
.get(&path)?
.send()
.context("Failed to send request")?;
let response = raise_for_err(response)?;
let progress: TaskProgressResponse = response

62
cli/src/experimental/tasks/update_stage.rs
View File

@@ -1,12 +1,12 @@
use anyhow::{Context, Result};
use inquire::Select;
use reqwest::blocking::Client;
use serde::Serialize;
use uuid::Uuid;
use super::{Task, TaskWorkflow, WorkflowStage};
use crate::{
experimental::tasks::utils::select_task, invocation_context::context, utils::raise_for_err,
api::client::PHClient, experimental::tasks::utils::select_task, invocation_context::context,
utils::raise_for_err,
};
struct StageChoice(WorkflowStage);
@@ -23,12 +23,10 @@ struct UpdateStageRequest {
}
pub fn update_stage(task_id: Option<&Uuid>) -> Result<()> {
let token = context().token.clone();
let host = token.get_host();
let client = context().client.clone();
let task = match task_id {
Some(id) => fetch_task(&client, &host, &token, id)?,
Some(id) => fetch_task(&client, id)?,
None => select_task("Select a task to update stage:")?,
};
@@ -41,7 +39,7 @@ pub fn update_stage(task_id: Option<&Uuid>) -> Result<()> {
}
};
let workflow = fetch_workflow(&client, &host, &token, &workflow_id)?;
let workflow = fetch_workflow(&client, &workflow_id)?;
if workflow.stages.is_empty() {
anyhow::bail!("The workflow '{}' has no stages defined.", workflow.name);
@@ -75,7 +73,7 @@ pub fn update_stage(task_id: Option<&Uuid>) -> Result<()> {
.prompt()
.context("Failed to get stage selection")?;
update_task_stage(&client, &host, &token, &task.id, &selected_stage.0.id)?;
update_task_stage(&client, &task.id, &selected_stage.0.id)?;
println!(
"\n✓ Successfully updated task stage to: {} ({})",
@@ -85,20 +83,9 @@ pub fn update_stage(task_id: Option<&Uuid>) -> Result<()> {
Ok(())
}
fn fetch_task(
client: &Client,
host: &str,
token: &crate::utils::auth::Token,
task_id: &Uuid,
) -> Result<Task> {
let url = format!(
"{}/api/environments/{}/tasks/{}/",
host, token.env_id, task_id
);
fn fetch_task(client: &PHClient, task_id: &Uuid) -> Result<Task> {
let response = client
.get(&url)
.header("Authorization", format!("Bearer {}", token.token))
.get(&format!("tasks/{task_id}/"))?
.send()
.context("Failed to send request")?;
@@ -109,23 +96,11 @@ fn fetch_task(
Ok(task)
}
fn fetch_workflow(
client: &Client,
host: &str,
token: &crate::utils::auth::Token,
workflow_id: &Uuid,
) -> Result<TaskWorkflow> {
let url = format!(
"{}/api/environments/{}/task_workflows/{}/",
host, token.env_id, workflow_id
);
fn fetch_workflow(client: &PHClient, workflow_id: &Uuid) -> Result<TaskWorkflow> {
let response = client
.get(&url)
.header("Authorization", format!("Bearer {}", token.token))
.get(&format!("task_workflows/{workflow_id}/"))?
.send()
.context("Failed to send request")?;
let response = raise_for_err(response)?;
let workflow: TaskWorkflow = response
@@ -135,29 +110,16 @@ fn fetch_workflow(
Ok(workflow)
}
fn update_task_stage(
client: &Client,
host: &str,
token: &crate::utils::auth::Token,
task_id: &Uuid,
stage_id: &Uuid,
) -> Result<()> {
let url = format!(
"{}/api/environments/{}/tasks/{}/update_stage/",
host, token.env_id, task_id
);
fn update_task_stage(client: &PHClient, task_id: &Uuid, stage_id: &Uuid) -> Result<()> {
let request_body = UpdateStageRequest {
current_stage: *stage_id,
};
let response = client
.patch(&url)
.header("Authorization", format!("Bearer {}", token.token))
.header("Content-Type", "application/json")
.patch(&format!("tasks/{task_id}/update_stage/"))?
.json(&request_body)
.send()
.context("Failed to send request")?;
.context("Failed to update task stage")?;
raise_for_err(response)?;

62
cli/src/experimental/tasks/utils.rs
View File

@@ -1,11 +1,10 @@
use anyhow::{Context, Result};
use inquire::Select;
use reqwest::blocking::Client;
use std::collections::VecDeque;
use super::{Task, TaskWorkflow, WorkflowStage};
use crate::{
experimental::tasks::list::TaskIterator, invocation_context::context, utils::auth::Token,
api::client::PHClient, experimental::tasks::list::TaskIterator, invocation_context::context,
};
const PAGE_SIZE: usize = 10;
@@ -29,11 +28,9 @@ impl std::fmt::Display for SelectionChoice {
}
pub fn select_task(prompt: &str) -> Result<Task> {
let token = context().token.clone();
let host = token.get_host();
let client = context().client.clone();
let mut task_iter = fetch_tasks(client, host, token, None)?;
let mut task_iter = fetch_tasks(client, None)?;
loop {
let mut choices = Vec::new();
@@ -67,13 +64,8 @@ pub fn select_task(prompt: &str) -> Result<Task> {
}
}
pub fn fetch_tasks(
client: Client,
host: String,
token: Token,
offset: Option<usize>,
) -> Result<TaskIterator> {
TaskIterator::new(client, host, token, offset)
pub fn fetch_tasks(client: PHClient, offset: Option<usize>) -> Result<TaskIterator> {
TaskIterator::new(client, offset)
}
#[derive(serde::Deserialize)]
@@ -83,30 +75,22 @@ struct WorkflowListResponse {
}
pub struct WorkflowIterator {
client: Client,
token: Token,
client: PHClient,
buffer: VecDeque<TaskWorkflow>,
next_url: Option<String>,
}
impl WorkflowIterator {
fn new(client: Client, host: String, token: Token) -> Result<Self> {
let initial_url = format!(
"{}/api/environments/{}/task_workflows/?limit={}",
host, token.env_id, BUFFER_SIZE
);
fn new(client: PHClient) -> Result<Self> {
let response = client
.get(&initial_url)
.header("Authorization", format!("Bearer {}", token.token))
.get(&format!("task_workflows/?limit={BUFFER_SIZE}"))?
.send()
.context("Failed to send request")?;
if !response.status().is_success() {
// Return empty iterator on error, don't fail
return Ok(Self {
client,
token,
client: client.clone(),
buffer: VecDeque::new(),
next_url: None,
});
@@ -121,7 +105,6 @@ impl WorkflowIterator {
Ok(Self {
client,
token,
buffer,
next_url: workflow_response.next,
})
@@ -136,8 +119,7 @@ impl WorkflowIterator {
let response = self
.client
.get(&url)
.header("Authorization", format!("Bearer {}", self.token.token))
.get(&url)?
.send()
.context("Failed to send request")?;
@@ -175,8 +157,8 @@ impl Iterator for WorkflowIterator {
}
}
pub fn fetch_workflows(client: Client, host: String, token: Token) -> Result<WorkflowIterator> {
WorkflowIterator::new(client, host, token)
pub fn fetch_workflows(client: PHClient) -> Result<WorkflowIterator> {
WorkflowIterator::new(client)
}
#[derive(serde::Deserialize)]
@@ -186,29 +168,21 @@ struct StageListResponse {
}
pub struct StageIterator {
client: Client,
token: Token,
client: PHClient,
buffer: VecDeque<WorkflowStage>,
next_url: Option<String>,
}
impl StageIterator {
fn new(client: Client, host: String, token: Token) -> Result<Self> {
let initial_url = format!(
"{}/api/environments/{}/workflow_stages/?limit={}",
host, token.env_id, BUFFER_SIZE
);
fn new(client: PHClient) -> Result<Self> {
let response = client
.get(&initial_url)
.header("Authorization", format!("Bearer {}", token.token))
.get(&format!("workflow_stages/?limit={BUFFER_SIZE}"))?
.send()
.context("Failed to send request")?;
if !response.status().is_success() {
return Ok(Self {
client,
token,
buffer: VecDeque::new(),
next_url: None,
});
@@ -223,7 +197,6 @@ impl StageIterator {
Ok(Self {
client,
token,
buffer,
next_url: stage_response.next,
})
@@ -238,8 +211,7 @@ impl StageIterator {
let response = self
.client
.get(&url)
.header("Authorization", format!("Bearer {}", self.token.token))
.get(&url)?
.send()
.context("Failed to send request")?;
@@ -277,6 +249,6 @@ impl Iterator for StageIterator {
}
}
pub fn fetch_stages(client: Client, host: String, token: Token) -> Result<StageIterator> {
StageIterator::new(client, host, token)
pub fn fetch_stages(client: PHClient) -> Result<StageIterator> {
StageIterator::new(client)
}

16
cli/src/experimental/tui/query.rs
View File

@@ -17,12 +17,10 @@ use crate::{
experimental::query::{
run_query, HogQLQueryErrorResponse, HogQLQueryResponse, HogQLQueryResult,
},
utils::{auth::Token, homedir::posthog_home_dir},
utils::homedir::posthog_home_dir,
};
pub struct QueryTui {
host: String,
creds: Token,
current_result: Option<HogQLQueryResult>,
lower_panel_state: Option<LowerPanelState>,
bg_query_handle: Option<JoinHandle<Result<HogQLQueryResult, Error>>>,
@@ -50,12 +48,10 @@ struct PersistedEditorState {
}
impl QueryTui {
pub fn new(creds: Token, host: String, debug: bool) -> Self {
pub fn new(debug: bool) -> Self {
Self {
current_result: None,
lower_panel_state: None,
creds,
host,
focus: Focus::Editor,
debug,
bg_query_handle: None,
@@ -296,9 +292,7 @@ impl QueryTui {
fn spawn_bg_query(&mut self, lines: Vec<String>) {
let query = lines.join("\n");
let query_endpoint = format!("{}/api/environments/{}/query", self.host, self.creds.env_id);
let m_token = self.creds.token.clone();
let handle = std::thread::spawn(move || run_query(&query_endpoint, &m_token, &query));
let handle = std::thread::spawn(move || run_query(&query));
// We drop any previously running thread handle here, but don't kill the thread... this is fine,
// I think. The alternative is to switch to tokio and get true task cancellation, but :shrug:,
@@ -307,10 +301,10 @@ impl QueryTui {
}
}
pub fn start_query_editor(host: &str, token: Token, debug: bool) -> Result<String, Error> {
pub fn start_query_editor(debug: bool) -> Result<String, Error> {
let terminal = ratatui::init();
let mut app = QueryTui::new(token, host.to_string(), debug);
let mut app = QueryTui::new(debug);
let res = app.enter_draw_loop(terminal);
ratatui::restore();
res

74
cli/src/invocation_context.rs
View File

@@ -1,4 +1,8 @@
use anyhow::Result;
use inquire::{
validator::{ErrorMessage, Validation},
CustomUserError,
};
use posthog_rs::Event;
use reqwest::blocking::Client;
use std::{
@@ -7,14 +11,17 @@ use std::{
};
use tracing::{debug, info, warn};
use crate::utils::auth::{get_token, Token};
use crate::{
api::client::PHClient,
utils::auth::{env_id_validator, get_token, host_validator, token_validator},
};
// I've decided in my infinite wisdom that global state is fine, actually.
pub static INVOCATION_CONTEXT: OnceLock<InvocationContext> = OnceLock::new();
pub struct InvocationContext {
pub token: Token,
pub client: Client,
pub config: InvocationConfig,
pub client: PHClient,
handles: Mutex<Vec<JoinHandle<()>>>,
}
@@ -24,17 +31,19 @@ pub fn context() -> &'static InvocationContext {
}
pub fn init_context(host: Option<String>, skip_ssl: bool) -> Result<()> {
let mut token = get_token()?;
if let Some(host) = host {
// If the user passed a host, respect it
token.host = Some(host);
}
let token = get_token()?;
let config = InvocationConfig {
api_key: token.token.clone(),
host: host.unwrap_or(token.host.unwrap_or("https://us.i.posthog.com".into())),
env_id: token.env_id.clone(),
skip_ssl,
};
let client = reqwest::blocking::Client::builder()
.danger_accept_invalid_certs(skip_ssl)
.build()?;
config.validate()?;
INVOCATION_CONTEXT.get_or_init(|| InvocationContext::new(token, client));
let client: PHClient = PHClient::from_config(config.clone())?;
INVOCATION_CONTEXT.get_or_init(|| InvocationContext::new(config, client));
// This is pulled at compile time, not runtime - we set it at build.
if let Some(token) = option_env!("POSTHOG_API_TOKEN") {
@@ -51,17 +60,52 @@ pub fn init_context(host: Option<String>, skip_ssl: bool) -> Result<()> {
Ok(())
}
#[derive(Clone)]
pub struct InvocationConfig {
pub api_key: String,
pub host: String,
pub env_id: String,
pub skip_ssl: bool,
}
impl InvocationConfig {
pub fn validate(&self) -> Result<()> {
fn handle_validation(
validation: Result<Validation, CustomUserError>,
context: &str,
) -> Result<()> {
let validation = validation.map_err(|err| anyhow::anyhow!("{context}: {err}"))?;
if let Validation::Invalid(ErrorMessage::Custom(msg)) = validation {
anyhow::bail!("{context}: {msg:?}");
}
Ok(())
}
handle_validation(token_validator(&self.api_key), "Invalid Personal API key")?;
handle_validation(host_validator(&self.host), "Invalid Host")?;
handle_validation(env_id_validator(&self.env_id), "Invalid Environment ID")?;
Ok(())
}
}
impl InvocationContext {
pub fn new(token: Token, client: Client) -> Self {
pub fn new(config: InvocationConfig, client: PHClient) -> Self {
Self {
token,
config,
client,
handles: Default::default(),
}
}
pub fn build_http_client(&self) -> Result<Client> {
let client = Client::builder()
.danger_accept_invalid_certs(self.config.skip_ssl)
.build()?;
Ok(client)
}
pub fn capture_command_invoked(&self, command: &str) {
let env_id = &self.token.env_id;
let env_id = self.client.get_env_id();
let event_name = "posthog cli command run".to_string();
let mut event = Event::new_anon(event_name);

10
cli/src/login.rs
View File

@@ -7,7 +7,10 @@ use tracing::info;
use crate::{
invocation_context::{context, init_context},
utils::auth::{host_validator, token_validator, CredentialProvider, HomeDirProvider, Token},
utils::auth::{
env_id_validator, host_validator, token_validator, CredentialProvider, HomeDirProvider,
Token,
},
};
#[derive(Debug, Deserialize)]
@@ -267,8 +270,9 @@ fn manual_login() -> Result<(), Error> {
.with_validator(host_validator)
.prompt()?;
let env_id =
Text::new("Enter your project ID (the number in your PostHog homepage URL)").prompt()?;
let env_id = Text::new("Enter your project ID (the number in your PostHog homepage URL)")
.with_validator(env_id_validator)
.prompt()?;
let token = Text::new(
"Enter your personal API token",

2
cli/src/sourcemaps/content.rs
View File

@@ -5,7 +5,6 @@ use serde::{Deserialize, Serialize};
use serde_json::Value;
use sourcemap::SourceMap;
use std::{collections::BTreeMap, path::PathBuf};
use tracing::info;
use crate::{
api::symbol_sets::SymbolSetUpload,
@@ -196,7 +195,6 @@ impl MinifiedSourceFile {
for path in possible_paths.into_iter() {
if path.exists() {
info!("Found sourcemap at path: {}", path.display());
return Ok(Some(path));
}
}

7
cli/src/sourcemaps/inject.rs
View File

@@ -55,12 +55,11 @@ pub fn inject_impl(args: &InjectArgs, matcher: impl Fn(&DirEntry) -> bool) -> Re
)
})?;
info!("Processing directory: {}", directory.display());
info!("injecting directory: {}", directory.display());
let mut pairs = read_pairs(&directory, ignore, matcher, public_path_prefix)?;
if pairs.is_empty() {
bail!("No source files found");
bail!("no source files found");
}
info!("Found {} pairs", pairs.len());
let created_release_id = get_release_for_pairs(&directory, project, version, &pairs)?
.as_ref()
@@ -72,7 +71,7 @@ pub fn inject_impl(args: &InjectArgs, matcher: impl Fn(&DirEntry) -> bool) -> Re
for pair in &pairs {
pair.save()?;
}
info!("Finished processing directory");
info!("injecting done");
Ok(())
}

14
cli/src/sourcemaps/source_pairs.rs
View File

@@ -110,29 +110,25 @@ pub fn read_pairs(
let entry_path = entry_path?;
if set.is_match(&entry_path) {
info!(
"Skipping because it matches an ignored glob: {}",
entry_path.display()
);
info!("skip [ignored]: {}", entry_path.display());
continue;
}
info!("Processing file: {}", entry_path.display());
let source = MinifiedSourceFile::load(&entry_path)?;
let sourcemap_path = source.get_sourcemap_path(prefix)?;
let Some(path) = sourcemap_path else {
warn!(
"No sourcemap file found for file {}, skipping",
entry_path.display()
);
warn!("skip [no sourcemap]: {}", entry_path.display());
continue;
};
info!("new pair: {}", entry_path.display());
let sourcemap = SourceMapFile::load(&path).context(format!("reading {path:?}"))?;
pairs.push(SourcePair { source, sourcemap });
}
info!("found {} pairs", pairs.len());
Ok(pairs)
}

16
cli/src/utils/auth.rs
View File

@@ -105,6 +105,22 @@ pub fn token_validator(token: &str) -> Result<Validation, CustomUserError> {
Ok(Validation::Valid)
}
pub fn env_id_validator(env_id: &str) -> Result<Validation, CustomUserError> {
// Must be a number
if env_id.is_empty() {
return Ok(Validation::Invalid("Environment ID cannot be empty".into()));
}
// Must be a number
if env_id.parse::<u32>().is_err() {
return Ok(Validation::Invalid(
"Environment ID must be a number".into(),
));
}
Ok(Validation::Valid)
}
pub fn get_token() -> Result<Token, Error> {
let env = EnvVarProvider;
let env_err = match env.get_credentials() {