jarvis/dd0c
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

cargo fmt: format all Rust source files
All checks were successful
CI — P1 Route (Rust) / test (push) Successful in 6m35s
Details

Browse Source
This commit is contained in:
Max Mayfield
2026-03-01 17:53:28 +00:00
parent 00db59ff83
commit a8a8c53917
12 changed files with 305 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
products/01-llm-cost-router/src/worker/anomaly.rs
View File

@@ -4,7 +4,7 @@ use uuid::Uuid;
async fn get_org_owner_email(pool: &PgPool, org_id: Uuid) -> Result<String, anyhow::Error> {
let row = sqlx::query_as::<_, (String,)>(
"SELECT email FROM users WHERE org_id = $1 AND role = 'owner' LIMIT 1"
"SELECT email FROM users WHERE org_id = $1 AND role = 'owner' LIMIT 1",
)
.bind(org_id)
.fetch_one(pool)
@@ -18,7 +18,7 @@ async fn get_org_owner_email(pool: &PgPool, org_id: Uuid) -> Result<String, anyh
pub async fn check_anomalies(ts_pool: &PgPool, pg_pool: &PgPool) -> anyhow::Result<()> {
// Get orgs with recent activity
let orgs = sqlx::query_as::<_, (Uuid,)>(
"SELECT DISTINCT org_id FROM request_events WHERE time >= now() - interval '1 hour'"
"SELECT DISTINCT org_id FROM request_events WHERE time >= now() - interval '1 hour'",
)
.fetch_all(ts_pool)
.await?;
@@ -35,7 +35,7 @@ async fn check_org_anomaly(ts_pool: &PgPool, pg_pool: &PgPool, org_id: Uuid) ->
let current = sqlx::query_as::<_, (f64,)>(
"SELECT COALESCE(SUM(cost_actual), 0)::float8
FROM request_events
WHERE org_id = $1 AND time >= now() - interval '1 hour'"
WHERE org_id = $1 AND time >= now() - interval '1 hour'",
)
.bind(org_id)
.fetch_one(ts_pool)
@@ -88,11 +88,7 @@ async fn check_org_anomaly(ts_pool: &PgPool, pg_pool: &PgPool, org_id: Uuid) ->
]
});
let client = reqwest::Client::new();
if let Err(e) = client.post(&slack_url)
.json(&payload)
.send()
.await
{
if let Err(e) = client.post(&slack_url).json(&payload).send().await {
warn!(error = %e, "Failed to send Slack anomaly alert");
}
}
@@ -114,7 +110,8 @@ async fn check_org_anomaly(ts_pool: &PgPool, pg_pool: &PgPool, org_id: Uuid) ->
)
});
let client = reqwest::Client::new();
if let Err(e) = client.post("https://api.resend.com/emails")
if let Err(e) = client
.post("https://api.resend.com/emails")
.bearer_auth(&resend_key)
.json(&email_body)
.send()

145
products/01-llm-cost-router/src/worker/digest.rs
View File

@@ -6,19 +6,17 @@ use uuid::Uuid;
/// Calculate next Monday 9 AM UTC from a given time
pub fn next_monday_9am(from: DateTime<Utc>) -> DateTime<Utc> {
let days_until_monday = (7 - from.weekday().num_days_from_monday()) % 7;
let days_until_monday = if days_until_monday == 0 && from.time() >= NaiveTime::from_hms_opt(9, 0, 0).unwrap() {
7 // Already past Monday 9 AM, go to next week
} else if days_until_monday == 0 {
0 // It's Monday but before 9 AM
} else {
days_until_monday
};
let days_until_monday =
if days_until_monday == 0 && from.time() >= NaiveTime::from_hms_opt(9, 0, 0).unwrap() {
7 // Already past Monday 9 AM, go to next week
} else if days_until_monday == 0 {
0 // It's Monday but before 9 AM
} else {
days_until_monday
};
let target_date = from.date_naive() + chrono::Duration::days(days_until_monday as i64);
target_date
.and_hms_opt(9, 0, 0)
.unwrap()
.and_utc()
target_date.and_hms_opt(9, 0, 0).unwrap().and_utc()
}
#[derive(Debug)]
@@ -58,7 +56,7 @@ pub async fn generate_all_digests(ts_pool: &PgPool, pg_pool: &PgPool) -> anyhow:
WHERE o.id IN (
SELECT DISTINCT org_id FROM request_events
WHERE time >= now() - interval '7 days'
)"
)",
)
.fetch_all(pg_pool)
.await?;
@@ -70,13 +68,27 @@ pub async fn generate_all_digests(ts_pool: &PgPool, pg_pool: &PgPool) -> anyhow:
Ok(digest) => {
// Send weekly digest via Resend
if let Ok(resend_key) = std::env::var("RESEND_API_KEY") {
let models_html: String = digest.top_models.iter().map(|m| {
format!("<tr><td>{}</td><td>{}</td><td>${:.4}</td></tr>", m.model, m.request_count, m.cost)
}).collect();
let models_html: String = digest
.top_models
.iter()
.map(|m| {
format!(
"<tr><td>{}</td><td>{}</td><td>${:.4}</td></tr>",
m.model, m.request_count, m.cost
)
})
.collect();
let savings_html: String = digest.top_savings.iter().map(|s| {
format!("<tr><td>{} → {}</td><td>{}</td><td>${:.4}</td></tr>", s.original_model, s.routed_model, s.requests_routed, s.cost_saved)
}).collect();
let savings_html: String = digest
.top_savings
.iter()
.map(|s| {
format!(
"<tr><td>{} → {}</td><td>{}</td><td>${:.4}</td></tr>",
s.original_model, s.routed_model, s.requests_routed, s.cost_saved
)
})
.collect();
let html = format!(
"<h2>Weekly Cost Digest: {}</h2>\
@@ -93,10 +105,14 @@ pub async fn generate_all_digests(ts_pool: &PgPool, pg_pool: &PgPool) -> anyhow:
<table style='border-collapse:collapse;width:100%'>\
<tr><th>Route</th><th>Requests</th><th>Saved</th></tr>{}</table>\
<p><a href='https://route.dd0c.dev/dashboard'>View Dashboard →</a></p>",
digest.org_name, digest.total_requests,
digest.total_cost_original, digest.total_cost_actual,
digest.total_cost_saved, digest.savings_pct,
models_html, savings_html
digest.org_name,
digest.total_requests,
digest.total_cost_original,
digest.total_cost_actual,
digest.total_cost_saved,
digest.savings_pct,
models_html,
savings_html
);
let email_body = serde_json::json!({
@@ -107,7 +123,8 @@ pub async fn generate_all_digests(ts_pool: &PgPool, pg_pool: &PgPool) -> anyhow:
});
let client = reqwest::Client::new();
match client.post("https://api.resend.com/emails")
match client
.post("https://api.resend.com/emails")
.bearer_auth(&resend_key)
.json(&email_body)
.send()
@@ -134,7 +151,11 @@ pub async fn generate_all_digests(ts_pool: &PgPool, pg_pool: &PgPool) -> anyhow:
Ok(())
}
async fn generate_digest(ts_pool: &PgPool, org_id: Uuid, org_name: &str) -> anyhow::Result<DigestData> {
async fn generate_digest(
ts_pool: &PgPool,
org_id: Uuid,
org_name: &str,
) -> anyhow::Result<DigestData> {
// Summary stats
let summary = sqlx::query_as::<_, (i64, f64, f64, f64)>(
"SELECT COUNT(*),
@@ -142,7 +163,7 @@ async fn generate_digest(ts_pool: &PgPool, org_id: Uuid, org_name: &str) -> anyh
COALESCE(SUM(cost_actual), 0)::float8,
COALESCE(SUM(cost_saved), 0)::float8
FROM request_events
WHERE org_id = $1 AND time >= now() - interval '7 days'"
WHERE org_id = $1 AND time >= now() - interval '7 days'",
)
.bind(org_id)
.fetch_one(ts_pool)
@@ -155,7 +176,7 @@ async fn generate_digest(ts_pool: &PgPool, org_id: Uuid, org_name: &str) -> anyh
WHERE org_id = $1 AND time >= now() - interval '7 days'
GROUP BY original_model
ORDER BY SUM(cost_actual) DESC
LIMIT 5"
LIMIT 5",
)
.bind(org_id)
.fetch_all(ts_pool)
@@ -168,13 +189,17 @@ async fn generate_digest(ts_pool: &PgPool, org_id: Uuid, org_name: &str) -> anyh
WHERE org_id = $1 AND time >= now() - interval '7 days' AND strategy != 'passthrough'
GROUP BY original_model, routed_model
ORDER BY SUM(cost_saved) DESC
LIMIT 5"
LIMIT 5",
)
.bind(org_id)
.fetch_all(ts_pool)
.await?;
let savings_pct = if summary.1 > 0.0 { (summary.3 / summary.1) * 100.0 } else { 0.0 };
let savings_pct = if summary.1 > 0.0 {
(summary.3 / summary.1) * 100.0
} else {
0.0
};
Ok(DigestData {
_org_id: org_id,
@@ -184,17 +209,23 @@ async fn generate_digest(ts_pool: &PgPool, org_id: Uuid, org_name: &str) -> anyh
total_cost_actual: summary.2,
total_cost_saved: summary.3,
savings_pct,
top_models: top_models.iter().map(|r| ModelUsage {
model: r.0.clone(),
request_count: r.1,
cost: r.2,
}).collect(),
top_savings: top_savings.iter().map(|r| RoutingSaving {
original_model: r.0.clone(),
routed_model: r.1.clone(),
requests_routed: r.2,
cost_saved: r.3,
}).collect(),
top_models: top_models
.iter()
.map(|r| ModelUsage {
model: r.0.clone(),
request_count: r.1,
cost: r.2,
})
.collect(),
top_savings: top_savings
.iter()
.map(|r| RoutingSaving {
original_model: r.0.clone(),
routed_model: r.1.clone(),
requests_routed: r.2,
cost_saved: r.3,
})
.collect(),
})
}
@@ -205,27 +236,45 @@ mod tests {
#[test]
fn next_monday_from_wednesday() {
let wed = chrono::NaiveDate::from_ymd_opt(2026, 3, 4).unwrap() // Wednesday
.and_hms_opt(14, 0, 0).unwrap().and_utc();
let wed = chrono::NaiveDate::from_ymd_opt(2026, 3, 4)
.unwrap() // Wednesday
.and_hms_opt(14, 0, 0)
.unwrap()
.and_utc();
let next = next_monday_9am(wed);
assert_eq!(next.weekday(), Weekday::Mon);
assert_eq!(next.date_naive(), chrono::NaiveDate::from_ymd_opt(2026, 3, 9).unwrap());
assert_eq!(
next.date_naive(),
chrono::NaiveDate::from_ymd_opt(2026, 3, 9).unwrap()
);
assert_eq!(next.time(), NaiveTime::from_hms_opt(9, 0, 0).unwrap());
}
#[test]
fn next_monday_from_monday_before_9am() {
let mon = chrono::NaiveDate::from_ymd_opt(2026, 3, 2).unwrap() // Monday
.and_hms_opt(8, 0, 0).unwrap().and_utc();
let mon = chrono::NaiveDate::from_ymd_opt(2026, 3, 2)
.unwrap() // Monday
.and_hms_opt(8, 0, 0)
.unwrap()
.and_utc();
let next = next_monday_9am(mon);
assert_eq!(next.date_naive(), chrono::NaiveDate::from_ymd_opt(2026, 3, 2).unwrap());
assert_eq!(
next.date_naive(),
chrono::NaiveDate::from_ymd_opt(2026, 3, 2).unwrap()
);
}
#[test]
fn next_monday_from_monday_after_9am() {
let mon = chrono::NaiveDate::from_ymd_opt(2026, 3, 2).unwrap()
.and_hms_opt(10, 0, 0).unwrap().and_utc();
let mon = chrono::NaiveDate::from_ymd_opt(2026, 3, 2)
.unwrap()
.and_hms_opt(10, 0, 0)
.unwrap()
.and_utc();
let next = next_monday_9am(mon);
assert_eq!(next.date_naive(), chrono::NaiveDate::from_ymd_opt(2026, 3, 9).unwrap());
assert_eq!(
next.date_naive(),
chrono::NaiveDate::from_ymd_opt(2026, 3, 9).unwrap()
);
}
}

10
products/01-llm-cost-router/src/worker/main.rs
View File

@@ -1,11 +1,11 @@
use std::sync::Arc;
use tracing::{info, error};
use chrono::Utc;
use std::sync::Arc;
use tracing::{error, info};
use dd0c_route::AppConfig;
mod digest;
mod anomaly;
mod digest;
/// Refresh model pricing from known provider pricing pages.
/// Falls back to hardcoded defaults if fetch fails.
@@ -83,7 +83,9 @@ async fn main() -> anyhow::Result<()> {
loop {
let now = Utc::now();
let next_monday = digest::next_monday_9am(now);
let sleep_duration = (next_monday - now).to_std().unwrap_or(std::time::Duration::from_secs(3600));
let sleep_duration = (next_monday - now)
.to_std()
.unwrap_or(std::time::Duration::from_secs(3600));
tokio::time::sleep(sleep_duration).await;
info!("Generating weekly digests");