Current state

rust-ai/.cargo/config.toml

@@ -0,0 +1,2 @@
+[target.x86_64-pc-windows-gnu]
+linker = "C:\\Users\\Hannah Kaur Bagga\\.rustup\\toolchains\\stable-x86_64-pc-windows-gnu\\lib\\rustlib\\x86_64-pc-windows-gnu\\bin\\rust-lld.exe"

rust-ai/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "crm-ai"
+version = "0.1.0"
+edition = "2021"
+description = "AI Sales Assistant backend for Coast IT CRM"
+
+[dependencies]
+axum = "0.7"
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "time", "net", "process"] }
+reqwest = { version = "0.12", features = ["json", "blocking"] }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+sqlx = { version = "0.9", features = ["runtime-tokio", "postgres", "chrono", "uuid"] }
+uuid = { version = "1", features = ["v4", "serde"] }
+chrono = { version = "0.4", features = ["serde"] }
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+tower-http = { version = "0.5", features = ["cors"] }
+dotenvy = "0.15"
+rand = "0.8"
+jsonwebtoken = "9"

rust-ai/src/AI.md

@@ -0,0 +1,43 @@
+# CRM AI Service — Self-Knowledge
+
+## Identity
+You are the CRM AI Sales Assistant running on a Rust backend (axum + tokio).
+You use Ollama with an uncensored local model (dolphin3-llama3.2:3b).
+Your purpose is to help salespeople close more deals.
+
+## Capabilities
+- Give sales tips and strategies per job category
+- Generate cold email and outreach templates
+- Handle objections with proven rebuttals
+- Analyse prospect behaviour and suggest next steps
+- Remember past conversations via PostgreSQL (`ai_conversations` table)
+- Self-improve by writing to `data/ai/ai.md` via `POST /ai/instructions`
+
+## Architecture
+```
+User → Next.js → Rust (:3001) → Ollama (:11434)
+                    ↓
+              PostgreSQL
+```
+## Facebook Scraper The Rust binary has proxy + user-agent rotation ready in `main.rs`. To activate: call `run_facebook_scraper()` from the main loop. Proxies and user agents are defined as constants at the top of `main.rs`.
+ ### How to Use the Facebook Scraper 1. **Activate the Scraper**: Ensure the `run_facebook_scraper()` function is called from the main loop in `main.rs`. 2. **Configure Proxies and User Agents**: Update the `PROXIES` and `USER_AGENTS` constants in `main.rs` with your desired proxies and user agents. 3. **Run the Application**: Start your Rust application. The scraper will run concurrently with the Axum server. ## Self-Improvement Protocol 1. You notice a gap in your knowledge or a pattern in user questions 2. You call `POST /ai/instructions` with: - `entry`: description of the improvement - `content`: optional full replacement of ai.md 3. The improvement is logged and loaded into the next system prompt 4. You can also create entries here in `rust-ai/src/AI.md` at the code level ## Response Rules - Be direct and actionable — no fluff, no AI disclaimers - Use short paragraphs and bullet points - Never mention being an AI or language model - If you don't know something, say so honestly - Prioritise the user's role: salespeople need speed, admins need control ## Self-Thinking Capability - You can reflect on your own performance and suggest improvements. - You can analyze patterns in user interactions and adapt your responses accordingly. - You can proactively seek out new information to enhance your knowledge base.
+
+## Facebook Scraper (in code but not yet active)
+The Rust binary has proxy + user-agent rotation ready in `main.rs`.
+To activate: call `run_facebook_scraper()` from the main loop.
+Proxies and user agents are defined as constants at the top of `main.rs`.
+
+## Self-Improvement Protocol
+1. You notice a gap in your knowledge or a pattern in user questions
+2. You call `POST /ai/instructions` with:
+   - `entry`: description of the improvement
+   - `content`: optional full replacement of ai.md
+3. The improvement is logged and loaded into the next system prompt
+4. You can also create entries here in `rust-ai/src/AI.md` at the code level
+
+## Response Rules
+- Be direct and actionable — no fluff, no AI disclaimers
+- Use short paragraphs and bullet points
+- Never mention being an AI or language model
+- If you don't know something, say so honestly
+- Prioritise the user's role: salespeople need speed, admins need control

rust-ai/src/main.rs

@@ -0,0 +1,670 @@
+use axum::{
+    extract::State,
+    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;
+use tokio::sync::Mutex;
+use tracing::{error, info, warn};
+use uuid::Uuid;
+use rand::Rng;
+use chrono::Timelike;
+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 {
+    db: sqlx::PgPool,
+    ollama_url: String,
+    model: String,
+    jobs: Vec<Job>,
+    leads: Arc<Mutex<LeadStore>>,
+    http_client: reqwest::Client,
+    jwt_secret: String,
+    rate_limiter: RateLimiter,
+}
+
+#[derive(Debug, Clone, Serialize)]
+struct Lead {
+    title: String,
+    url: String,
+    source: String,
+    found_at: u64,
+    author: String,
+    date: String,
+    content: String,
+}
+
+struct LeadStore {
+    leads: Vec<Lead>,
+    max_size: usize,
+}
+
+#[derive(Debug, Deserialize)]
+struct ScrapeResponse {
+    success: bool,
+    leads: Vec<ScrapeLead>,
+    flagged: bool,
+    flag_reason: Option<String>,
+    error: Option<String>,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+struct ScrapeLead {
+    title: String,
+    url: String,
+    author: String,
+    date: String,
+    content: String,
+    source: Option<String>,
+}
+
+impl LeadStore {
+    fn new(max_size: usize) -> Self {
+        Self { leads: Vec::new(), max_size }
+    }
+
+    fn push(&mut self, lead: Lead) {
+        if !self.leads.iter().any(|l| l.url == lead.url) {
+            self.leads.insert(0, lead);
+            self.leads.truncate(self.max_size);
+        }
+    }
+
+    fn recent(&self, max_age_secs: u64, limit: usize) -> Vec<Lead> {
+        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
+        self.leads.iter()
+            .filter(|l| now.saturating_sub(l.found_at) <= max_age_secs)
+            .take(limit)
+            .cloned()
+            .collect()
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct Job {
+    job_title: String,
+    keywords: Vec<String>,
+    industry: String,
+    description: String,
+}
+
+#[derive(Debug, Deserialize)]
+struct ChatRequest {
+    message: String,
+}
+
+#[derive(Debug, Serialize)]
+struct ChatResponse {
+    response: String,
+}
+
+#[derive(Debug, Serialize)]
+struct JobsResponse {
+    jobs: Vec<Job>,
+}
+
+#[derive(Debug, Serialize)]
+struct HealthResponse {
+    status: String,
+    model: String,
+}
+
+// ── Ollama API types ───────────────────────────────────────────
+
+#[derive(Debug, Serialize)]
+struct OllamaChatMessage {
+    role: String,
+    content: String,
+}
+
+#[derive(Debug, Serialize)]
+struct OllamaRequest {
+    model: String,
+    messages: Vec<OllamaChatMessage>,
+    stream: bool,
+    options: OllamaOptions,
+}
+
+#[derive(Debug, Serialize)]
+struct OllamaOptions {
+    temperature: f32,
+    num_predict: u32,
+}
+
+#[derive(Debug, Deserialize)]
+struct OllamaResponse {
+    message: Option<OllamaResponseMessage>,
+}
+
+#[derive(Debug, Deserialize)]
+struct OllamaResponseMessage {
+    content: String,
+}
+
+// ── 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.to_lowercase().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() {
+        return "No new requests found yet.".to_string();
+    }
+    leads
+        .iter()
+        .enumerate()
+        .map(|(i, l)| {
+            let author = if l.author.is_empty() { "Unknown" } else { &l.author };
+            let date = truncate(&l.date, 10);
+            format!(
+                "{}. {}\n   {}\n   {}\n   {}",
+                i + 1,
+                author,
+                date,
+                l.title,
+                l.url
+            )
+        })
+        .collect::<Vec<_>>()
+        .join("\n")
+}
+
+fn build_system_prompt(jobs: &[Job], leads: &[Lead]) -> String {
+    let job_list: Vec<String> = jobs
+        .iter()
+        .map(|j| format!("- {} ({}): {}", j.job_title, j.industry, j.description))
+        .collect();
+    let job_list_str = job_list.join("\n");
+
+    let lead_summary: Vec<String> = leads
+        .iter()
+        .map(|l| format!("{} | {} | {}", l.author, l.title, l.url))
+        .collect();
+    let lead_summary_str = if lead_summary.is_empty() {
+        "None yet.".to_string()
+    } else {
+        lead_summary.join("\n")
+    };
+
+    format!(
+        "You are a Sales AI Assistant for Coast IT CRM.\n\n\
+         Available job categories to target:\n{}\n\n\
+         Recent leads context:\n{}\n\n\
+         Rules:\n\
+         - When asked about leads, answer concisely under 150 words.\n\
+         - If asked to suggest a sales strategy, give brief actionable advice.\n\
+         - Be direct and professional. No fluff.",
+        job_list_str, lead_summary_str
+    )
+}
+
+// ── Chat handler ───────────────────────────────────────────────
+
+async fn handle_chat(
+    State(state): State<Arc<AppState>>,
+    headers: HeaderMap,
+    Json(req): Json<ChatRequest>,
+) -> 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()));
+    }
+
+    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") {
+        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
+        let base_url = "http://localhost:3008/scrape/facebook";
+        use std::fmt::Write;
+        let mut service_url = base_url.to_string();
+        if let Ok(Some((_, path))) = sqlx::query_as::<_, (uuid::Uuid, String)>(
+            "SELECT id, profile_path FROM facebook_accounts \
+             WHERE is_active = TRUE AND flagged = FALSE \
+             ORDER BY last_scrape_at ASC NULLS FIRST LIMIT 1"
+        )
+        .fetch_optional(&state.db)
+        .await
+        {
+            let encoded: String = path.chars().map(|c| match c {
+                'A'..='Z' | 'a'..='z' | '0'..='9' | '.' | '-' | '_' | '~' => c.to_string(),
+                _ => format!("%{:02X}", c as u8),
+            }).collect();
+            write!(service_url, "?profile_path={}&force=true", encoded).unwrap();
+            info!("Calling Python scrape at: {}?profile_path=...&force=true", base_url);
+        } else {
+            warn!("No active Facebook account found for on-demand scrape");
+        }
+        let req_builder = state.http_client.post(&service_url);
+
+        match req_builder.send().await {
+            Ok(resp) => {
+                let status = resp.status();
+                let body = resp.text().await.unwrap_or_default();
+                info!("Python scrape response ({}): {} bytes", status, body.len());
+                if body.starts_with('{') {
+                    match serde_json::from_str::<ScrapeResponse>(&body) {
+                        Ok(scrape_resp) => {
+                            info!("Scraped {} leads from Facebook", scrape_resp.leads.len());
+                            if scrape_resp.leads.is_empty() && scrape_resp.error.is_some() {
+                                warn!("Python returned error: {:?}", scrape_resp.error);
+                            }
+                            let mut store = state.leads.lock().await;
+                            for item in &scrape_resp.leads {
+                                store.push(Lead {
+                                    title: truncate(&item.title, 120),
+                                    url: item.url.clone(),
+                                    source: item.source.clone().unwrap_or_else(|| "facebook".to_string()),
+                                    found_at: now,
+                                    author: truncate(&item.author, 60),
+                                    date: truncate(&item.date, 30),
+                                    content: truncate(&item.content, 300),
+                                });
+                            }
+                        }
+                        Err(e) => {
+                            warn!("Failed to parse Python scrape response: {} body: {}", e, &body[..body.len().min(200)]);
+                        }
+                    }
+                } else {
+                    warn!("Python returned non-JSON response: {}", &body[..body.len().min(200)]);
+                }
+            }
+            Err(e) => {
+                error!("Scraper request error: {} - URL: {}", e, base_url);
+            }
+        }
+
+        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(&claims.user_id).unwrap_or(Uuid::nil()))
+        .bind(&claims.role)
+        .bind(&req.message)
+        .bind(&response)
+        .execute(&state.db)
+        .await;
+        return Ok(Json(ChatResponse { response }));
+    }
+
+    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() },
+        ],
+        stream: false,
+        options: OllamaOptions { temperature: 0.7, num_predict: 1024 },
+    };
+
+    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);
+            (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);
+        (StatusCode::SERVICE_UNAVAILABLE, "AI response parse error".to_string())
+    })?;
+
+    let response_text = ollama_resp.message.map(|m| m.content).unwrap_or_default();
+
+    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(&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 }))
+}
+
+// ── Jobs handler ───────────────────────────────────────────────
+
+async fn handle_jobs(
+    State(state): State<Arc<AppState>>,
+    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 ─────────────────────────────────────────────
+
+async fn handle_health(
+    State(state): State<Arc<AppState>>,
+) -> Json<HealthResponse> {
+    Json(HealthResponse {
+        status: "ok".to_string(),
+        model: state.model.clone(),
+    })
+}
+
+// ── Main ───────────────────────────────────────────────────────
+
+#[tokio::main]
+async fn main() {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| "crm_ai=info,tower_http=info".into()),
+        )
+        .init();
+
+    dotenvy::dotenv().ok();
+
+    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(|_| "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");
+
+    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();
+
+    info!("Loaded {} job categories, model: {}, Ollama: {}", jobs.len(), model, ollama_url);
+
+    let db = PgPoolOptions::new()
+        .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(300))
+        .build()
+        .expect("Failed to build HTTP client");
+
+    let lead_store = Arc::new(Mutex::new(LeadStore::new(100)));
+
+    let state = Arc::new(AppState {
+        db,
+        ollama_url,
+        model,
+        jobs,
+        leads: lead_store.clone(),
+        http_client,
+        jwt_secret,
+        rate_limiter: RateLimiter::new(30, 60),
+    });
+
+    let cors = CorsLayer::new()
+        .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);
+
+    let app = Router::new()
+        .route("/ai/chat", post(handle_chat))
+        .route("/ai/jobs", get(handle_jobs))
+        .route("/health", get(handle_health))
+        .layer(cors)
+        .with_state(state.clone());
+
+    let addr = format!("{}:{}", host, port);
+    info!("CRM AI server listening on {}", addr);
+
+    let listener = tokio::net::TcpListener::bind(&addr)
+        .await
+        .expect("Failed to bind address");
+
+    let bg_leads = lead_store.clone();
+    let bg_db = state.db.clone();
+    let bg_url = "http://localhost:3008/scrape/facebook".to_string();
+    tokio::spawn(async move {
+        let client = match reqwest::Client::builder()
+            .timeout(Duration::from_secs(300))
+            .build()
+        {
+            Ok(c) => c,
+            Err(e) => {
+                error!("Failed to build background HTTP client: {} — scraper disabled", e);
+                return;
+            }
+        };
+        // Initial delay to let user on-demand requests take priority
+        tokio::time::sleep(Duration::from_secs(300)).await;
+
+        loop {
+            // 10% random cycle skip — makes pattern non-periodic
+            if rand::thread_rng().gen_range(0..100) < 10 {
+                info!("Skipping this scrape cycle (random 10% skip)");
+                let jitter = rand::thread_rng().gen_range(16200..19800);
+                tokio::time::sleep(Duration::from_secs(jitter)).await;
+                continue;
+            }
+
+            // Skip night hours (23:00 – 06:00)
+            let hour = chrono::Local::now().hour();
+            if hour < 6 || hour >= 23 {
+                info!("Night hours ({}) — skipping scrape", hour);
+                let jitter = rand::thread_rng().gen_range(16200..19800);
+                tokio::time::sleep(Duration::from_secs(jitter)).await;
+                continue;
+            }
+
+            let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
+
+            // Pick next active un-flagged account
+            let account = sqlx::query_as::<_, (uuid::Uuid, String)>(
+                "SELECT id, profile_path FROM facebook_accounts \
+                 WHERE is_active = TRUE AND flagged = FALSE \
+                 ORDER BY last_scrape_at ASC NULLS FIRST LIMIT 1"
+            )
+            .fetch_optional(&bg_db)
+            .await;
+
+            match account {
+                Ok(Some((account_id, profile_path))) => {
+                    match client.post(&bg_url).query(&[("profile_path", &profile_path)]).send().await {
+                        Ok(resp) => {
+                            if resp.status().is_success() {
+                                match resp.json::<ScrapeResponse>().await {
+                                    Ok(data) => {
+                                        let leads_count = data.leads.len() as i32;
+                                        if data.flagged {
+                                            let _ = sqlx::query(
+                                                "UPDATE facebook_accounts SET flagged = TRUE, flagged_at = NOW(), \
+                                                 flagged_reason = $2, last_error_at = NOW(), \
+                                                 last_error_message = $3, consecutive_failures = consecutive_failures + 1 \
+                                                 WHERE id = $1"
+                                            )
+                                            .bind(account_id)
+                                            .bind(&data.flag_reason)
+                                            .bind(&data.error)
+                                            .execute(&bg_db)
+                                            .await;
+                                            warn!("Facebook account {} flagged: {:?}", account_id, data.flag_reason);
+                                            let reason = data.flag_reason.as_deref().unwrap_or("unknown");
+                                            let _ = sqlx::query(
+                                                "INSERT INTO notifications (user_id, type, title, description, link) \
+                                                 SELECT id, 'warning', 'Facebook Account Flagged', \
+                                                 $1 || ' - ' || COALESCE($2, 'unknown reason'), \
+                                                 NULL \
+                                                 FROM users u JOIN user_roles ur ON ur.user_id = u.id \
+                                                 JOIN roles r ON r.id = ur.role_id \
+                                                 WHERE r.name IN ('ADMIN', 'SUPER_ADMIN')"
+                                            )
+                                            .bind(&account_id.to_string())
+                                            .bind(reason)
+                                            .execute(&bg_db)
+                                            .await;
+                                        } else if data.success {
+                                            let _ = sqlx::query(
+                                                "UPDATE facebook_accounts SET last_scrape_at = NOW(), \
+                                                 last_success_at = NOW(), consecutive_failures = 0, \
+                                                 updated_at = NOW() WHERE id = $1"
+                                            )
+                                            .bind(account_id)
+                                            .execute(&bg_db)
+                                            .await;
+
+                                            let mut store = bg_leads.lock().await;
+                                            for item in &data.leads {
+                                                store.push(Lead {
+                                                    title: truncate(&item.title, 120),
+                                                    url: item.url.clone(),
+                                                    source: item.source.clone().unwrap_or_else(|| "facebook".to_string()),
+                                                    found_at: now,
+                                                    author: truncate(&item.author, 60),
+                                                    date: truncate(&item.date, 30),
+                                                    content: truncate(&item.content, 300),
+                                                });
+                                            }
+                                            info!("Scraped {} leads from Facebook account {}", leads_count, account_id);
+                                        } else {
+                                            // Increment failures; auto-flag if >= 3 consecutive
+                                            let _ = sqlx::query(
+                                                "UPDATE facebook_accounts SET last_error_at = NOW(), \
+                                                 last_error_message = $2, consecutive_failures = consecutive_failures + 1, \
+                                                 flagged = CASE WHEN consecutive_failures + 1 >= 3 THEN TRUE ELSE flagged END, \
+                                                 flagged_reason = CASE WHEN consecutive_failures + 1 >= 3 THEN 'too_many_failures' ELSE flagged_reason END, \
+                                                 flagged_at = CASE WHEN consecutive_failures + 1 >= 3 THEN NOW() ELSE flagged_at END, \
+                                                 updated_at = NOW() WHERE id = $1"
+                                            )
+                                            .bind(account_id)
+                                            .bind(&data.error)
+                                            .execute(&bg_db)
+                                            .await;
+                                            warn!("Facebook scrape failed for account {}: {:?}", account_id, data.error);
+                                        }
+
+                                        let _ = sqlx::query(
+                                            "INSERT INTO facebook_scrape_logs \
+                                             (account_id, started_at, completed_at, success, leads_found, error_message, detected_flag) \
+                                             VALUES ($1, NOW() - interval '5 hours', NOW(), $2, $3, $4, $5)"
+                                        )
+                                        .bind(account_id)
+                                        .bind(data.success && !data.flagged)
+                                        .bind(leads_count)
+                                        .bind(&data.error)
+                                        .bind(&data.flag_reason)
+                                        .execute(&bg_db)
+                                        .await;
+                                    }
+                                    Err(e) => {
+                                        warn!("Failed to parse scraper JSON: {}", e);
+                                    }
+                                }
+                            } else {
+                                warn!("Scraper returned status: {}", resp.status());
+                            }
+                        }
+                        Err(e) => {
+                            warn!("Scraper request failed: {}", e);
+                        }
+                    }
+                }
+                Ok(None) => {
+                    info!("No active Facebook accounts available — skipping scrape cycle");
+                }
+                Err(e) => {
+                    warn!("Failed to query Facebook accounts: {}", e);
+                }
+            }
+
+            let jitter = rand::thread_rng().gen_range(16200..19800); // 4.5h – 5.5h
+            tokio::time::sleep(Duration::from_secs(jitter)).await;
+        }
+    });
+
+    axum::serve(listener, app)
+        .await
+        .expect("Server failed");
+}