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}; use crate::worker::sql::{ QUERY_GET_NEW_HOUSE_DATA, QUERY_ADD_NEW_BUYABLE_HOUSE, QUERY_UPDATE_BUYABLE_HOUSE_STATE, QUERY_UPDATE_USER_HOUSE_STATE, QUERY_DELETE_COLLAPSED_BUYABLE_HOUSES, QUERY_DELETE_COLLAPSED_USER_HOUSES, }; use crate::worker::{insert_notification_conn, publish_update_status}; pub struct HouseWorker { base: BaseWorker, } 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}"); } if let Err(err) = Self::cleanup_collapsed_houses(&pool, &broker) { eprintln!("[HouseWorker] Fehler in cleanup_collapsed_houses: {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}"); } // Nach der täglichen Verschlechterung direkt zusammengefallene Häuser aufräumen. if let Err(err) = Self::cleanup_collapsed_houses(&pool, &broker) { eprintln!("[HouseWorker] Fehler in cleanup_collapsed_houses: {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(()) } fn cleanup_collapsed_houses(pool: &ConnectionPool, broker: &MessageBroker) -> Result<(), DbError> { let mut conn = pool .get() .map_err(|e| DbError::new(format!("[HouseWorker] DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare( "delete_collapsed_buyable_houses", QUERY_DELETE_COLLAPSED_BUYABLE_HOUSES, )?; conn.prepare( "delete_collapsed_user_houses", QUERY_DELETE_COLLAPSED_USER_HOUSES, )?; // 1) Kaufbare Häuser (ohne Besitzer) löschen let deleted_buyable = conn.execute("delete_collapsed_buyable_houses", &[])?; if !deleted_buyable.is_empty() { eprintln!( "[HouseWorker] {} kaufbare(s) Haus/Häuser zusammengefallen und entfernt", deleted_buyable.len() ); } // 2) Häuser im Besitz löschen + Owner informieren let deleted_user_houses = conn.execute("delete_collapsed_user_houses", &[])?; if deleted_user_houses.is_empty() { return Ok(()); } for row in deleted_user_houses { let house_id = row .get("house_id") .and_then(|v| v.parse::().ok()) .unwrap_or(-1); let user_id = row .get("user_id") .and_then(|v| v.parse::().ok()) .unwrap_or(-1); if house_id < 0 || user_id < 0 { continue; } // Notification + Frontend-Update let tr = format!(r#"{{"tr":"house.collapsed","house_id":{}}}"#, house_id); let _ = insert_notification_conn(&mut conn, user_id, &tr, None); publish_update_status(broker, user_id); } 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(); } }