CRM_ENVIROMENT/Newbie_CRM
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Files Changed

Browse Source 
Security Fixes
File	Change
.env.local	Placeholder JWT secret (rotate to random value: openssl rand -base64 64)
src/middleware.ts	Removed /api/auth/me bypass; API routes return JSON 401 (not 302 redirect)
src/lib/auth.ts	sameSite: "lax" → "strict" (create + destroy cookie); SVG name chars are pre-computed (no injection)
src/lib/avatar.ts	Added sanitizeName() — strips non-alphanumeric/safe chars, max 50 chars
src/app/api/users/route.ts	POST restricted to super_admin; validates role against whitelist ["sales","admin","super_admin","dev"]; validates active defaults to true
src/app/api/users/[id]/route.ts	DELETE restricted to super_admin; blocks self-deletion
src/app/api/leads/[id]/route.ts	GET: non-admin users filtered by assigned_to; PATCH: ownership check + score validated 0-100 range
src/app/api/leads/route.ts	DELETE: ownership + admin check; GET: ownership filter for non-admin users
src/app/api/conversations/[id]/messages/route.ts	GET/POST: verify user is a conversation_participant before access
rust-ai/src/main.rs	CORS: AllowOrigin::list(["localhost:3006", "127.0.0.1:3006"]) (was Any)
Performance Fixes
File	Change
src/app/api/leads/route.ts	Added limit (default 50), offset query params; ownership filter in SQL
src/app/api/users/route.ts	Added limit (default 50), offset query params
src/app/api/conversations/route.ts	Added LIMIT 50
src/app/api/leads/[id]/notes/route.ts	Added limit (default 50), offset query params
src/app/api/conversations/[id]/messages/route.ts	Added limit (default 100), offset, before query params
src/app/(dashboard)/chats/page.tsx	Pruned to 5 cached conversations; useMemo wraps messages/conversation/filteredConversations; otherParticipant moved outside component; added 10MB file size limit; recordingChunksRef cleared on discard
src/app/(dashboard)/leads/[id]/page.tsx	Optimistic status update rolls back on fetch failure
database/migrations/003_chat.sql	Added composite index idx_messages_conversation_created on (conversation_id, created_at DESC)
Code Quality Fixes
File	Change
src/lib/ai.ts	Removed dead getInstructions()/updateInstructions() (called nonexistent /ai/instructions)
src/lib/auth.ts	bcrypt.hash(password, 12) — kept (acceptable, OWASP-recommended)
src/app/login/page.tsx	"Forgot password?" button now shows alert to contact admin (was no-op)
src/components/users/user-form-dialog.tsx	Password min 6 → 8 chars
26 silent catch {} blocks across 16 files (see below)	Added console.warn("...") context messages
This commit is contained in:
Ace
2026-06-23 09:45:30 +02:00
parent d891197d8b
commit 1adc4806fa
51 changed files with 1145 additions and 1052 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rust-ai/src/instructions.rs
-85
View File
@@ -1,85 +0,0 @@
use std::fs;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{error, info, warn};
/// Manages the ai.md self-improvement file.
/// Loaded on startup and periodically refreshed.
pub struct InstructionsManager {
path: PathBuf,
current: RwLock<String>,
}
impl InstructionsManager {
pub fn new(path: PathBuf) -> Self {
let initial = fs::read_to_string(&path).unwrap_or_default();
if initial.is_empty() {
warn!("ai.md is empty or missing at {:?}", path);
} else {
info!("Loaded ai.md ({} bytes)", initial.len());
}
Self {
path,
current: RwLock::new(initial),
}
}
/// Read the current instructions
pub async fn get(&self) -> String {
self.current.read().await.clone()
}
/// Append a new entry to the Improvement Log and optionally update the instructions.
/// Returns the updated full content.
pub async fn update(&self, entry: &str, new_content: Option<&str>) -> Result<String, String> {
if let Some(content) = new_content {
// Full replacement
fs::write(&self.path, content).map_err(|e| format!("Write failed: {}", e))?;
let mut current = self.current.write().await;
*current = content.to_string();
info!("ai.md fully replaced ({} bytes)", content.len());
Ok(content.to_string())
} else {
// Append to Improvement Log only
let mut current = self.current.write().await;
let log_entry = format!("\n- {} — {}", chrono::Utc::now().format("%Y-%m-%d %H:%M"), entry);
// Find the Improvement Log section and append
if let Some(pos) = current.rfind("\n## Improvement Log") {
// Find the next section after Improvement Log, or end of file
let after_log = &current[pos..];
if let Some(section_start) = after_log[1..].find("\n## ") {
let insert_at = pos + 1 + section_start;
current.insert_str(insert_at, &format!("{}\n", log_entry));
} else {
current.push_str(&format!("{}\n", log_entry));
}
} else {
current.push_str(&format!("\n## Improvement Log\n{}", log_entry));
}
fs::write(&self.path, &*current).map_err(|e| format!("Write failed: {}", e))?;
info!("ai.md improvement log appended: {}", entry);
Ok(current.clone())
}
}
/// Reload from disk
pub async fn reload(&self) {
match fs::read_to_string(&self.path) {
Ok(content) => {
let len = content.len();
let mut current = self.current.write().await;
*current = content;
info!("ai.md reloaded from disk ({} bytes)", len);
}
Err(e) => error!("Failed to reload ai.md: {}", e),
}
}
}
/// Wrapper for thread-safe sharing
pub type SharedInstructions = Arc<InstructionsManager>;
pub fn create_shared(path: PathBuf) -> SharedInstructions {
Arc::new(InstructionsManager::new(path))
}
rust-ai/src/main.rs
+186 -142
View File
@@ -1,20 +1,69 @@
use axum::{
extract::State,
http::Method,
http::{HeaderMap, Method, StatusCode},
routing::{get, post},
Json, Router,
};
use tower_http::cors::{CorsLayer, AllowOrigin, Any};
use jsonwebtoken::{decode, DecodingKey, Validation, Algorithm};
use serde::{Deserialize, Serialize};
use sqlx::postgres::PgPoolOptions;
use std::collections::HashMap;
use std::fs;
use std::sync::{Arc, Mutex};
use tower_http::cors::{Any, CorsLayer};
use tracing::{error, info};
use std::sync::Arc;
use tokio::sync::Mutex;
use tracing::{error, info, warn};
use uuid::Uuid;
use rand::Rng;
use std::time::Duration;
use std::time::{SystemTime, UNIX_EPOCH};
// ── JWT Claims ────────────────────────────────────────────────
#[derive(Debug, Deserialize)]
struct Claims {
#[serde(rename = "userId")]
user_id: String,
role: String,
}
fn verify_jwt(token: &str, secret: &str) -> Option<Claims> {
let key = DecodingKey::from_secret(secret.as_bytes());
let validation = Validation::new(Algorithm::HS256);
decode::<Claims>(token, &key, &validation).ok().map(|d| d.claims)
}
// ── Rate limiter ──────────────────────────────────────────────
struct RateLimiter {
buckets: Mutex<HashMap<String, Vec<u64>>>,
max_requests: usize,
window_secs: u64,
}
impl RateLimiter {
fn new(max_requests: usize, window_secs: u64) -> Self {
Self {
buckets: Mutex::new(HashMap::new()),
max_requests,
window_secs,
}
}
async fn check(&self, key: &str) -> bool {
let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
let mut buckets = self.buckets.lock().await;
let timestamps = buckets.entry(key.to_string()).or_default();
timestamps.retain(|t| now - *t <= self.window_secs);
if timestamps.len() >= self.max_requests {
false
} else {
timestamps.push(now);
true
}
}
}
// ── Shared state ───────────────────────────────────────────────
struct AppState {
@@ -23,6 +72,9 @@ struct AppState {
model: String,
jobs: Vec<Job>,
leads: Arc<Mutex<LeadStore>>,
http_client: reqwest::Client,
jwt_secret: String,
rate_limiter: RateLimiter,
}
#[derive(Debug, Clone, Serialize)]
@@ -47,7 +99,6 @@ impl LeadStore {
}
fn push(&mut self, lead: Lead) {
// Deduplicate by URL
if !self.leads.iter().any(|l| l.url == lead.url) {
self.leads.insert(0, lead);
self.leads.truncate(self.max_size);
@@ -55,9 +106,9 @@ impl LeadStore {
}
fn recent(&self, max_age_secs: u64, limit: usize) -> Vec<Lead> {
let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
self.leads.iter()
.filter(|l| now - l.found_at <= max_age_secs)
.filter(|l| now.saturating_sub(l.found_at) <= max_age_secs)
.take(limit)
.cloned()
.collect()
@@ -75,8 +126,6 @@ struct Job {
#[derive(Debug, Deserialize)]
struct ChatRequest {
message: String,
user_id: String,
user_role: String,
}
#[derive(Debug, Serialize)]
@@ -127,7 +176,23 @@ struct OllamaResponseMessage {
content: String,
}
// ── System prompt builder ─────────────────────────────────────
// ── Helpers ────────────────────────────────────────────────────
fn truncate(s: &str, max: usize) -> String {
s.chars().take(max).collect()
}
fn extract_claims(headers: &HeaderMap, state: &AppState) -> Result<Claims, (StatusCode, String)> {
let auth_header = headers.get("Authorization").and_then(|v| v.to_str().ok()).unwrap_or("");
let token = auth_header.strip_prefix("Bearer ").unwrap_or("");
let claims = verify_jwt(token, &state.jwt_secret).ok_or_else(|| {
(StatusCode::UNAUTHORIZED, "Unauthorized".to_string())
})?;
match claims.role.as_str() {
"sales" | "admin" | "super_admin" => Ok(claims),
_ => Err((StatusCode::FORBIDDEN, "Forbidden".to_string())),
}
}
fn format_leads_output(leads: &[Lead]) -> String {
if leads.is_empty() {
@@ -138,7 +203,7 @@ fn format_leads_output(leads: &[Lead]) -> String {
.enumerate()
.map(|(i, l)| {
let author = if l.author.is_empty() { "Unknown" } else { &l.author };
let date = if l.date.is_empty() { "Unknown" } else { &l.date[..l.date.len().min(10)] };
let date = truncate(&l.date, 10);
format!(
"{}. {}\n {}\n {}\n {}",
i + 1,
@@ -161,9 +226,7 @@ fn build_system_prompt(jobs: &[Job], leads: &[Lead]) -> String {
let lead_summary: Vec<String> = leads
.iter()
.map(|l| {
format!("{} | {} | {}", l.author, l.title, l.url)
})
.map(|l| format!("{} | {} | {}", l.author, l.title, l.url))
.collect();
let lead_summary_str = if lead_summary.is_empty() {
"None yet.".to_string()
@@ -187,62 +250,56 @@ fn build_system_prompt(jobs: &[Job], leads: &[Lead]) -> String {
async fn handle_chat(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
Json(req): Json<ChatRequest>,
) -> Result<Json<ChatResponse>, (axum::http::StatusCode, String)> {
// Validate role
match req.user_role.as_str() {
"sales" | "admin" | "super_admin" => {}
_ => return Err((axum::http::StatusCode::FORBIDDEN, "Forbidden".to_string())),
) -> Result<Json<ChatResponse>, (StatusCode, String)> {
let claims = extract_claims(&headers, &state)?;
if !state.rate_limiter.check(&claims.user_id).await {
return Err((StatusCode::TOO_MANY_REQUESTS, "Rate limit exceeded".to_string()));
}
// Intercept lead listing requests — scrape on demand then return
let msg_lower = req.message.to_lowercase();
let msg_words: Vec<&str> = msg_lower.split_whitespace().collect();
let has_listing = msg_words.iter().any(|w| ["listings", "listing", "leads", "links", "lists"].contains(w));
let has_show = msg_lower.contains("show me") || msg_lower.contains("give me") || msg_lower.contains("pull");
let has_job = msg_words.contains(&"jobs") || msg_words.contains(&"job");
if has_listing || (has_show && has_job) || (has_show && msg_lower.contains("links")) || msg_lower.contains("recent leads") {
// Scrape fresh leads now — call both services
let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
let client = reqwest::Client::new();
let services = [
"http://localhost:3008/scrape/all".to_string(),
];
for service_url in &services {
if let Ok(resp) = client
.post(service_url)
.timeout(Duration::from_secs(120))
.send()
.await
{
if let Ok(leads_data) = resp.json::<Vec<serde_json::Value>>().await {
info!("Scraped {} leads from {}", leads_data.len(), service_url);
let mut store = state.leads.lock().unwrap();
for item in &leads_data {
store.push(Lead {
title: item["title"].as_str().unwrap_or("").chars().take(120).collect(),
url: item["url"].as_str().unwrap_or("").to_string(),
source: item["source"].as_str().unwrap_or("unknown").to_string(),
found_at: now,
author: item["author"].as_str().unwrap_or("").chars().take(60).collect(),
date: item["date"].as_str().unwrap_or("").chars().take(30).collect(),
content: item["content"].as_str().unwrap_or("").chars().take(300).collect(),
});
}
let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
let service_url = "http://localhost:3008/scrape/all".to_string();
if let Ok(resp) = state.http_client
.post(&service_url)
.timeout(Duration::from_secs(120))
.send()
.await
{
if let Ok(leads_data) = resp.json::<Vec<serde_json::Value>>().await {
info!("Scraped {} leads from {}", leads_data.len(), service_url);
let mut store = state.leads.lock().await;
for item in &leads_data {
store.push(Lead {
title: truncate(item["title"].as_str().unwrap_or(""), 120),
url: item["url"].as_str().unwrap_or("").to_string(),
source: item["source"].as_str().unwrap_or("unknown").to_string(),
found_at: now,
author: truncate(item["author"].as_str().unwrap_or(""), 60),
date: truncate(item["date"].as_str().unwrap_or(""), 30),
content: truncate(item["content"].as_str().unwrap_or(""), 300),
});
}
} else {
error!("Scraper service unreachable: {}", service_url);
}
} else {
error!("Scraper service unreachable: {}", service_url);
}
let recent_leads = state.leads.lock().unwrap().recent(604800, 20);
let recent_leads = state.leads.lock().await.recent(604800, 20);
let response = format_leads_output(&recent_leads);
let _ = sqlx::query(
"INSERT INTO ai_conversations (id, user_id, role, message, response) VALUES ($1, $2, $3, $4, $5)",
)
.bind(Uuid::new_v4())
.bind(Uuid::parse_str(&req.user_id).unwrap_or(Uuid::nil()))
.bind(&req.user_role)
.bind(Uuid::parse_str(&claims.user_id).unwrap_or(Uuid::nil()))
.bind(&claims.role)
.bind(&req.message)
.bind(&response)
.execute(&state.db)
@@ -250,84 +307,62 @@ async fn handle_chat(
return Ok(Json(ChatResponse { response }));
}
let recent_leads = state.leads.lock().unwrap().recent(604800, 20);
let recent_leads = state.leads.lock().await.recent(604800, 20);
let system_prompt = build_system_prompt(&state.jobs, &recent_leads);
let message_text = req.message.clone();
let ollama_req = OllamaRequest {
model: state.model.clone(),
messages: vec![
OllamaChatMessage {
role: "system".to_string(),
content: system_prompt,
},
OllamaChatMessage {
role: "user".to_string(),
content: req.message.clone(),
},
OllamaChatMessage { role: "system".to_string(), content: system_prompt },
OllamaChatMessage { role: "user".to_string(), content: req.message.clone() },
],
stream: false,
options: OllamaOptions {
temperature: 0.7,
num_predict: 1024,
},
options: OllamaOptions { temperature: 0.7, num_predict: 1024 },
};
let client = reqwest::Client::new();
let resp = client
let resp = state.http_client
.post(format!("{}/api/chat", state.ollama_url))
.json(&ollama_req)
.send()
.await
.map_err(|e| {
error!("Ollama request failed: {}", e);
(
axum::http::StatusCode::SERVICE_UNAVAILABLE,
"AI service unavailable".to_string(),
)
(StatusCode::SERVICE_UNAVAILABLE, "AI service unavailable".to_string())
})?;
let ollama_resp: OllamaResponse = resp.json().await.map_err(|e| {
error!("Failed to parse Ollama response: {}", e);
(
axum::http::StatusCode::SERVICE_UNAVAILABLE,
"AI response parse error".to_string(),
)
(StatusCode::SERVICE_UNAVAILABLE, "AI response parse error".to_string())
})?;
let response_text = ollama_resp
.message
.map(|m| m.content)
.unwrap_or_default();
let response_text = ollama_resp.message.map(|m| m.content).unwrap_or_default();
// Store in database
let user_id = Uuid::parse_str(&req.user_id).unwrap_or(Uuid::nil());
let _ = sqlx::query(
"INSERT INTO ai_conversations (id, user_id, role, message, response) VALUES ($1, $2, $3, $4, $5)",
)
.bind(Uuid::new_v4())
.bind(user_id)
.bind(&req.user_role)
.bind(Uuid::parse_str(&claims.user_id).unwrap_or(Uuid::nil()))
.bind(&claims.role)
.bind(&message_text)
.bind(&response_text)
.execute(&state.db)
.await;
Ok(Json(ChatResponse {
response: response_text,
}))
Ok(Json(ChatResponse { response: response_text }))
}
// ── Jobs handler ───────────────────────────────────────────────
async fn handle_jobs(
State(state): State<Arc<AppState>>,
) -> Json<JobsResponse> {
Json(JobsResponse {
jobs: state.jobs.clone(),
})
headers: HeaderMap,
) -> Result<Json<JobsResponse>, (StatusCode, String)> {
let _claims = extract_claims(&headers, &state)?;
if !state.rate_limiter.check(&_claims.user_id).await {
return Err((StatusCode::TOO_MANY_REQUESTS, "Rate limit exceeded".to_string()));
}
Ok(Json(JobsResponse { jobs: state.jobs.clone() }))
}
// ── Health handler ─────────────────────────────────────────────
@@ -354,42 +389,32 @@ async fn main() {
dotenvy::dotenv().ok();
let database_url =
std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
let ollama_url =
std::env::var("OLLAMA_BASE_URL").unwrap_or_else(|_| "http://localhost:11434".to_string());
let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
let jwt_secret = std::env::var("JWT_SECRET").expect("JWT_SECRET must be set");
let ollama_url = std::env::var("OLLAMA_BASE_URL").unwrap_or_else(|_| "http://localhost:11434".to_string());
let model = std::env::var("AI_MODEL").unwrap_or_else(|_| "dolphin-phi".to_string());
let host = std::env::var("AI_HOST").unwrap_or_else(|_| "0.0.0.0".to_string());
let port: u16 = std::env::var("AI_PORT")
.unwrap_or_else(|_| "3001".to_string())
.parse()
.expect("AI_PORT must be a number");
let host = std::env::var("AI_HOST").unwrap_or_else(|_| "127.0.0.1".to_string());
let port: u16 = std::env::var("AI_PORT").unwrap_or_else(|_| "3001".to_string()).parse().expect("AI_PORT must be a number");
// Load jobs
let jobs_path = std::env::var("JOBS_PATH").unwrap_or_else(|_| "data/ai/jobs.jsonl".to_string());
let jobs_content = fs::read_to_string(&jobs_path).unwrap_or_default();
let jobs: Vec<Job> = jobs_content
.lines()
.filter(|l| !l.trim().is_empty())
.filter_map(|l| serde_json::from_str(l).ok())
.collect();
let jobs: Vec<Job> = jobs_content.lines().filter(|l| !l.trim().is_empty()).filter_map(|l| serde_json::from_str(l).ok()).collect();
info!(
"Loaded {} job categories, model: {}, Ollama: {}",
jobs.len(),
model,
ollama_url
);
info!("Loaded {} job categories, model: {}, Ollama: {}", jobs.len(), model, ollama_url);
// Connect to database
let db = PgPoolOptions::new()
.max_connections(5)
.max_connections(20)
.connect(&database_url)
.await
.expect("Failed to connect to database");
info!("Connected to PostgreSQL");
let http_client = reqwest::Client::builder()
.timeout(Duration::from_secs(120))
.build()
.expect("Failed to build HTTP client");
let lead_store = Arc::new(Mutex::new(LeadStore::new(100)));
let state = Arc::new(AppState {
@@ -398,11 +423,16 @@ async fn main() {
model,
jobs,
leads: lead_store.clone(),
http_client,
jwt_secret,
rate_limiter: RateLimiter::new(30, 60),
});
// CORS layer
let cors = CorsLayer::new()
.allow_origin(Any)
.allow_origin(AllowOrigin::list([
"http://localhost:3006".parse().unwrap(),
"http://127.0.0.1:3006".parse().unwrap(),
]))
.allow_methods([Method::GET, Method::POST])
.allow_headers(Any);
@@ -420,43 +450,57 @@ async fn main() {
.await
.expect("Failed to bind address");
// Start background scrapers
let bg_leads = lead_store.clone();
tokio::task::spawn_blocking(move || {
let bg_client = reqwest::blocking::Client::builder()
let bg_url = "http://localhost:3008/scrape/all".to_string();
tokio::spawn(async move {
let client = match reqwest::Client::builder()
.timeout(Duration::from_secs(120))
.build().ok();
.build()
{
Ok(c) => c,
Err(e) => {
error!("Failed to build background HTTP client: {} — scraper disabled", e);
return;
}
};
loop {
if let Some(ref c) = bg_client {
let urls = vec![
"http://localhost:3008/scrape/all".to_string(),
];
let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
for url in &urls {
if let Ok(resp) = c.post(url).send() {
if let Ok(data) = resp.json::<Vec<serde_json::Value>>() {
let mut store = bg_leads.lock().unwrap();
for item in &data {
store.push(Lead {
title: item["title"].as_str().unwrap_or("").chars().take(120).collect(),
url: item["url"].as_str().unwrap_or("").to_string(),
source: item["source"].as_str().unwrap_or("unknown").to_string(),
found_at: now,
author: item["author"].as_str().unwrap_or("").chars().take(60).collect(),
date: item["date"].as_str().unwrap_or("").chars().take(30).collect(),
content: item["content"].as_str().unwrap_or("").chars().take(300).collect(),
});
let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
match client.post(&bg_url).send().await {
Ok(resp) => {
if resp.status().is_success() {
match resp.json::<Vec<serde_json::Value>>().await {
Ok(data) => {
let mut store = bg_leads.lock().await;
for item in &data {
store.push(Lead {
title: truncate(item["title"].as_str().unwrap_or(""), 120),
url: item["url"].as_str().unwrap_or("").to_string(),
source: item["source"].as_str().unwrap_or("unknown").to_string(),
found_at: now,
author: truncate(item["author"].as_str().unwrap_or(""), 60),
date: truncate(item["date"].as_str().unwrap_or(""), 30),
content: truncate(item["content"].as_str().unwrap_or(""), 300),
});
}
}
Err(e) => {
warn!("Failed to parse scraper JSON: {}", e);
}
}
} else {
warn!("Scraper returned status: {}", resp.status());
}
}
Err(e) => {
warn!("Scraper request failed: {}", e);
}
}
let delay = rand::thread_rng().gen_range(120..300);
std::thread::sleep(Duration::from_secs(delay));
tokio::time::sleep(Duration::from_secs(delay)).await;
}
});
axum::serve(listener, app)
.await
.expect("Server failed");
}
}