use crate::db::{ConnectionPool, DbError}; use crate::message_broker::MessageBroker; use std::sync::atomic::Ordering; use std::sync::Arc; use std::time::{Duration, Instant}; use super::base::{BaseWorker, Worker, WorkerState}; pub struct HouseWorker { base: BaseWorker, } // SQL-Queries analog zu `houseworker.h` const QUERY_GET_NEW_HOUSE_DATA: &str = r#" SELECT h.id AS house_id FROM falukant_type.house AS h WHERE random() < 0.0001 AND label_tr <> 'under_bridge'; "#; const QUERY_ADD_NEW_BUYABLE_HOUSE: &str = r#" INSERT INTO falukant_data.buyable_house (house_type_id) VALUES ($1); "#; const QUERY_UPDATE_BUYABLE_HOUSE_STATE: &str = r#" UPDATE falukant_data.buyable_house SET roof_condition = ROUND(roof_condition - random() * (3 + 0 * id)), floor_condition = ROUND(floor_condition - random() * (3 + 0 * id)), wall_condition = ROUND(wall_condition - random() * (3 + 0 * id)), window_condition = ROUND(window_condition - random() * (3 + 0 * id)); "#; const QUERY_UPDATE_USER_HOUSE_STATE: &str = r#" UPDATE falukant_data.user_house SET roof_condition = ROUND(roof_condition - random() * (3 + 0 * id)), floor_condition = ROUND(floor_condition - random() * (3 + 0 * id)), wall_condition = ROUND(wall_condition - random() * (3 + 0 * id)), window_condition = ROUND(window_condition - random() * (3 + 0 * id)) WHERE house_type_id NOT IN ( SELECT id FROM falukant_type.house h WHERE h.label_tr = 'under_bridge' ); "#; impl HouseWorker { pub fn new(pool: ConnectionPool, broker: MessageBroker) -> Self { Self { base: BaseWorker::new("HouseWorker", pool, broker), } } fn run_loop(pool: ConnectionPool, _broker: MessageBroker, state: Arc) { let mut last_hourly_run: Option = None; let mut last_daily_run: Option = None; while state.running_worker.load(Ordering::Relaxed) { let now = Instant::now(); // Stündliche Aufgaben: neue Häuser erzeugen let should_run_hourly = match last_hourly_run { None => true, Some(last) => now.saturating_duration_since(last) >= Duration::from_secs(3600), }; if should_run_hourly { if let Err(err) = Self::perform_task_inner(&pool) { eprintln!("[HouseWorker] Fehler in performTask: {err}"); } last_hourly_run = Some(now); } // Tägliche Aufgaben: Hauszustände verschlechtern let should_run_daily = match last_daily_run { None => true, Some(last) => now.saturating_duration_since(last) >= Duration::from_secs(24 * 3600), }; if should_run_daily { if let Err(err) = Self::perform_house_state_change_inner(&pool) { eprintln!("[HouseWorker] Fehler in performHouseStateChange: {err}"); } last_daily_run = Some(now); } std::thread::sleep(Duration::from_secs(1)); } } fn perform_task_inner(pool: &ConnectionPool) -> Result<(), DbError> { let mut conn = pool .get() .map_err(|e| DbError::new(format!("[HouseWorker] DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("get_new_house_data", QUERY_GET_NEW_HOUSE_DATA)?; let rows = conn.execute("get_new_house_data", &[])?; conn.prepare("add_new_buyable_house", QUERY_ADD_NEW_BUYABLE_HOUSE)?; for row in rows { if let Some(house_id) = row .get("house_id") .and_then(|v| v.parse::().ok()) { conn.execute("add_new_buyable_house", &[&house_id])?; } } Ok(()) } fn perform_house_state_change_inner(pool: &ConnectionPool) -> Result<(), DbError> { let mut conn = pool .get() .map_err(|e| DbError::new(format!("[HouseWorker] DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare( "update_buyable_house_state", QUERY_UPDATE_BUYABLE_HOUSE_STATE, )?; conn.prepare( "update_user_house_state", QUERY_UPDATE_USER_HOUSE_STATE, )?; conn.execute("update_buyable_house_state", &[])?; conn.execute("update_user_house_state", &[])?; Ok(()) } } impl Worker for HouseWorker { fn start_worker_thread(&mut self) { let pool = self.base.pool.clone(); let broker = self.base.broker.clone(); self.base .start_worker_with_loop(move |state: Arc| { Self::run_loop(pool.clone(), broker.clone(), state); }); } fn stop_worker_thread(&mut self) { self.base.stop_worker(); } fn enable_watchdog(&mut self) { self.base.start_watchdog(); } }