use crate::db::{ConnectionPool, DbError, Rows}; use crate::message_broker::MessageBroker; use rand::distributions::{Distribution, Uniform}; use rand::rngs::StdRng; use rand::SeedableRng; 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_USERS_TO_UPDATE, QUERY_UPDATE_CHARACTERS_HEALTH, QUERY_UPDATE_MOOD, QUERY_UPDATE_GET_ITEMS_TO_UPDATE, QUERY_UPDATE_GET_CHARACTER_IDS, QUERY_UPDATE_KNOWLEDGE, QUERY_DELETE_LOG_ENTRY, QUERY_GET_OPEN_CREDITS, QUERY_UPDATE_CREDIT, QUERY_CLEANUP_CREDITS, QUERY_ADD_CHARACTER_TO_DEBTORS_PRISM, QUERY_GET_CURRENT_MONEY, QUERY_GET_HOUSE_VALUE, QUERY_GET_SETTLEMENT_VALUE, QUERY_GET_INVENTORY_VALUE, QUERY_GET_CREDIT_DEBT, QUERY_COUNT_CHILDREN, QUERY_GET_HEIR, QUERY_RANDOM_HEIR, QUERY_SET_CHARACTER_USER, QUERY_UPDATE_USER_MONEY, QUERY_GET_FALUKANT_USER_ID, QUERY_AUTOBATISM, QUERY_GET_PREGNANCY_CANDIDATES, QUERY_INSERT_CHILD, QUERY_INSERT_CHILD_RELATION, QUERY_DELETE_DIRECTOR, QUERY_DELETE_RELATIONSHIP, QUERY_DELETE_CHILD_RELATION, QUERY_DELETE_KNOWLEDGE, QUERY_DELETE_DEBTORS_PRISM, QUERY_DELETE_POLITICAL_OFFICE, QUERY_DELETE_ELECTION_CANDIDATE, }; /// Vereinfachtes Abbild eines Characters aus `QUERY_GET_USERS_TO_UPDATE`. #[derive(Debug, Clone)] struct Character { id: i32, age: i32, health: i32, } pub struct UserCharacterWorker { base: BaseWorker, rng: StdRng, dist: Uniform, last_hourly_run: Option, last_pregnancy_run: Option, last_mood_run: Option, } // SQL moved to `src/worker/sql.rs` impl UserCharacterWorker { pub fn new(pool: ConnectionPool, broker: MessageBroker) -> Self { let base = BaseWorker::new("UserCharacterWorker", pool, broker); let rng = StdRng::from_entropy(); let dist = Uniform::from(0.0..1.0); Self { base, rng, dist, last_hourly_run: None, last_pregnancy_run: None, last_mood_run: None, } } fn run_iteration(&mut self, state: &WorkerState) { self.base.set_current_step("UserCharacterWorker iteration"); self.maybe_run_hourly_tasks(); self.maybe_run_mood_updates(); self.maybe_run_daily_pregnancies(); // Entspricht in etwa der 1-Sekunden-Schleife im C++-Code std::thread::sleep(Duration::from_secs(1)); if let Err(err) = self.recalculate_knowledge() { eprintln!("[UserCharacterWorker] Fehler in recalculateKnowledge: {err}"); } if !state.running_worker.load(Ordering::Relaxed) { // worker stopping } } fn maybe_run_hourly_tasks(&mut self) { let now = Instant::now(); let should_run = match self.last_hourly_run { None => true, Some(last) => now.saturating_duration_since(last) >= Duration::from_secs(3600), }; if !should_run { return; } if let Err(err) = self.run_hourly_tasks() { eprintln!("[UserCharacterWorker] Fehler in stündlichen Tasks: {err}"); } self.last_hourly_run = Some(now); } fn run_hourly_tasks(&mut self) -> Result<(), DbError> { self.process_character_events()?; self.handle_credits()?; Ok(()) } fn maybe_run_daily_pregnancies(&mut self) { let now = Instant::now(); let should_run = match self.last_pregnancy_run { None => true, Some(last) => now.saturating_duration_since(last) >= Duration::from_secs(24 * 3600), }; if !should_run { return; } if let Err(err) = self.process_pregnancies() { eprintln!("[UserCharacterWorker] Fehler in processPregnancies: {err}"); } self.last_pregnancy_run = Some(now); } fn process_character_events(&mut self) -> Result<(), DbError> { self.base.set_current_step("Get character data"); let rows = self.load_characters_to_update()?; let mut characters: Vec = rows .into_iter() .filter_map(Self::map_row_to_character) .collect(); for character in &mut characters { self.update_character_health(character)?; } Ok(()) } fn load_characters_to_update(&mut self) -> Result { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("get_users_to_update", QUERY_GET_USERS_TO_UPDATE)?; conn.execute("get_users_to_update", &[]) } fn map_row_to_character(row: crate::db::Row) -> Option { Some(Character { id: row.get("id")?.parse().ok()?, age: row.get("age")?.parse().ok()?, health: row.get("health")?.parse().ok()?, }) } fn update_character_health(&mut self, character: &mut Character) -> Result<(), DbError> { let health_change = self.calculate_health_change(character.age); if health_change == 0 { return Ok(()); } character.health = std::cmp::max(0, character.health + health_change); if character.health == 0 { self.handle_character_death(character.id)?; return Ok(()); } let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare( "update_characters_health", QUERY_UPDATE_CHARACTERS_HEALTH, )?; conn.execute( "update_characters_health", &[&character.health, &character.id], )?; Ok(()) } fn calculate_health_change(&mut self, age: i32) -> i32 { if age < 30 { return 0; } if age >= 45 { let probability = (0.1 + (age - 45) as f64 * 0.02).min(1.0); if self.dist.sample(&mut self.rng) < probability { let damage_dist = Uniform::from(1..=10); return -damage_dist.sample(&mut self.rng); } return 0; } let probability = (age - 30) as f64 / 30.0; if self.dist.sample(&mut self.rng) < probability { -1 } else { 0 } } fn maybe_run_mood_updates(&mut self) { let now = Instant::now(); let should_run = match self.last_mood_run { None => true, Some(last) => now.saturating_duration_since(last) >= Duration::from_secs(60), }; if !should_run { return; } if let Err(err) = self.update_characters_mood_randomized() { eprintln!("[UserCharacterWorker] Fehler in updateCharactersMood: {err}"); } self.last_mood_run = Some(now); } /// Setzt die Stimmung einzelner lebender Charaktere zufällig neu. /// Jeder Charakter hat pro Minute eine kleine Chance auf einen Wechsel, /// so dass sich über die Zeit ein individueller, zufälliger Rhythmus entsteht. fn update_characters_mood_randomized(&mut self) -> Result<(), DbError> { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("update_mood", QUERY_UPDATE_MOOD)?; conn.execute("update_mood", &[])?; Ok(()) } fn recalculate_knowledge(&mut self) -> Result<(), DbError> { self.base.set_current_step("recalculate knowledge"); let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare( "get_items_to_update", QUERY_UPDATE_GET_ITEMS_TO_UPDATE, )?; let update_rows = conn.execute("get_items_to_update", &[])?; for update_item in update_rows { let quantity: i32 = match update_item.get("quantity").and_then(|v| v.parse().ok()) { Some(q) => q, None => continue, }; if quantity < 10 { self.delete_production_log_entry(&mut conn, &update_item)?; continue; } self.update_knowledge_for_production(&mut conn, &update_item)?; self.delete_production_log_entry(&mut conn, &update_item)?; if let Some(producer_id) = update_item .get("producer_id") .and_then(|v| v.parse::().ok()) { self.send_knowledge_update(producer_id); } } Ok(()) } fn update_knowledge_for_production( &mut self, conn: &mut crate::db::DbConnection, update_item: &crate::db::Row, ) -> Result<(), DbError> { let producer_id = match update_item.get("producer_id").and_then(|v| v.parse::().ok()) { Some(id) => id, None => return Ok(()), }; let product_id = match update_item.get("product_id").and_then(|v| v.parse::().ok()) { Some(id) => id, None => return Ok(()), }; conn.prepare( "get_character_ids", QUERY_UPDATE_GET_CHARACTER_IDS, )?; let characters_data = conn.execute("get_character_ids", &[&producer_id])?; conn.prepare("update_knowledge", QUERY_UPDATE_KNOWLEDGE)?; for character_row in characters_data { let character_id = match character_row .get("character_id") .and_then(|v| v.parse::().ok()) { Some(id) => id, None => continue, }; let director_id = character_row .get("director_id") .and_then(|v| v.parse::().ok()) .unwrap_or(0); if director_id == 0 { conn.execute( "update_knowledge", &[&character_id, &product_id, &2_i32], )?; } else { conn.execute( "update_knowledge", &[&character_id, &product_id, &1_i32], )?; conn.execute( "update_knowledge", &[&director_id, &product_id, &1_i32], )?; } } Ok(()) } fn delete_production_log_entry( &mut self, conn: &mut crate::db::DbConnection, update_item: &crate::db::Row, ) -> Result<(), DbError> { let id = match update_item.get("id").and_then(|v| v.parse::().ok()) { Some(id) => id, None => return Ok(()), }; conn.prepare("delete_log_entry", QUERY_DELETE_LOG_ENTRY)?; conn.execute("delete_log_entry", &[&id])?; Ok(()) } fn send_knowledge_update(&self, producer_id: i32) { let message = format!(r#"{{"event":"knowledge_update","user_id":{}}}"#, producer_id); self.base.broker.publish(message); } // Kredit-Logik (portiert aus handleCredits) fn handle_credits(&mut self) -> Result<(), DbError> { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("get_open_credits", QUERY_GET_OPEN_CREDITS)?; conn.prepare("update_credit", QUERY_UPDATE_CREDIT)?; conn.prepare("cleanup_credits", QUERY_CLEANUP_CREDITS)?; conn.prepare( "add_character_to_debtors_prism", QUERY_ADD_CHARACTER_TO_DEBTORS_PRISM, )?; let credits_rows = conn.execute("get_open_credits", &[])?; for row in credits_rows { if let Some(credit) = Self::map_row_to_credit(&row) { self.process_single_credit(&mut conn, &credit)?; } } conn.execute("cleanup_credits", &[])?; Ok(()) } fn map_row_to_credit(row: &crate::db::Row) -> Option { Some(Credit { amount: row.get("amount")?.parse().ok()?, remaining_amount: row.get("remaining_amount")?.parse().ok()?, interest_rate: row.get("interest_rate")?.parse().ok()?, user_id: row.get("user_id")?.parse().ok()?, money: row.get("money")?.parse().ok()?, character_id: row.get("character_id")?.parse().ok()?, prism_started_previously: row .get("prism_started_previously") .map(|v| v == "t" || v == "true") .unwrap_or(false), }) } fn process_single_credit( &mut self, conn: &mut crate::db::DbConnection, credit: &Credit, ) -> Result<(), DbError> { let Credit { amount, mut remaining_amount, interest_rate, user_id, money, character_id, prism_started_previously, .. } = *credit; let pay_rate = amount / 10.0 + amount * interest_rate as f64 / 100.0; remaining_amount -= pay_rate; // Kann der User zahlen? if pay_rate <= money - (pay_rate * 3.0) { if let Err(err) = self .base .change_falukant_user_money(user_id, -pay_rate, "credit pay rate") { eprintln!( "[UserCharacterWorker] Fehler bei change_falukant_user_money (credit pay rate): {err}" ); } } else if prism_started_previously { if let Err(err) = self .base .change_falukant_user_money(user_id, pay_rate, "debitor_prism") { eprintln!( "[UserCharacterWorker] Fehler bei change_falukant_user_money (debitor_prism): {err}" ); } } else { conn.execute("add_character_to_debtors_prism", &[&character_id])?; } conn.execute("update_credit", &[&remaining_amount, &user_id])?; Ok(()) } // Schwangerschafts-Logik (portiert aus processPregnancies) fn process_pregnancies(&mut self) -> Result<(), DbError> { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("autobatism", QUERY_AUTOBATISM)?; conn.execute("autobatism", &[])?; conn.prepare("get_pregnancy_candidates", QUERY_GET_PREGNANCY_CANDIDATES)?; let rows = conn.execute("get_pregnancy_candidates", &[])?; conn.prepare("insert_child", QUERY_INSERT_CHILD)?; conn.prepare("insert_child_relation", QUERY_INSERT_CHILD_RELATION)?; for row in rows { self.process_single_pregnancy_candidate(&mut conn, &row)?; } Ok(()) } fn process_single_pregnancy_candidate( &mut self, conn: &mut crate::db::DbConnection, row: &crate::db::Row, ) -> Result<(), DbError> { let father_cid = parse_i32(row, "father_cid", -1); let mother_cid = parse_i32(row, "mother_cid", -1); if father_cid < 0 || mother_cid < 0 { return Ok(()); } let title_of_nobility = parse_i32(row, "title_of_nobility", 0); let last_name = parse_i32(row, "last_name", 0); let region_id = parse_i32(row, "region_id", 0); let father_uid = parse_opt_i32(row, "father_uid"); let mother_uid = parse_opt_i32(row, "mother_uid"); let gender = if self.dist.sample(&mut self.rng) < 0.5 { "male" } else { "female" }; let inserted = conn.execute("insert_child", &[®ion_id, &gender, &last_name, &title_of_nobility])?; let child_cid = inserted .first() .and_then(|r| r.get("child_cid")) .and_then(|v| v.parse::().ok()) .unwrap_or(-1); if child_cid < 0 { return Ok(()); } conn.execute( "insert_child_relation", &[&father_cid, &mother_cid, &child_cid], )?; if let Some(f_uid) = father_uid { self.send_children_update_and_status(f_uid); } if let Some(m_uid) = mother_uid { self.send_children_update_and_status(m_uid); } Ok(()) } fn send_children_update_and_status(&self, user_id: i32) { let children_update = format!(r#"{{"event":"children_update","user_id":{}}}"#, user_id); self.base.broker.publish(children_update); let update_status = format!(r#"{{"event":"falukantUpdateStatus","user_id":{}}}"#, user_id); self.base.broker.publish(update_status); } // Todes- und Erb-Logik fn handle_character_death(&mut self, character_id: i32) -> Result<(), DbError> { self.set_heir(character_id)?; let death_event = format!( r#"{{"event":"CharacterDeath","character_id":{}}}"#, character_id ); self.base.broker.publish(death_event); let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("delete_director", QUERY_DELETE_DIRECTOR)?; conn.prepare("delete_relationship", QUERY_DELETE_RELATIONSHIP)?; conn.prepare("delete_child_relation", QUERY_DELETE_CHILD_RELATION)?; conn.prepare("delete_knowledge", QUERY_DELETE_KNOWLEDGE)?; conn.prepare("delete_debtors_prism", QUERY_DELETE_DEBTORS_PRISM)?; conn.prepare("delete_political_office", QUERY_DELETE_POLITICAL_OFFICE)?; conn.prepare("delete_election_candidate", QUERY_DELETE_ELECTION_CANDIDATE)?; conn.execute("delete_director", &[&character_id])?; conn.execute("delete_relationship", &[&character_id])?; conn.execute("delete_child_relation", &[&character_id])?; conn.execute("delete_knowledge", &[&character_id])?; conn.execute("delete_debtors_prism", &[&character_id])?; conn.execute("delete_political_office", &[&character_id])?; conn.execute("delete_election_candidate", &[&character_id])?; // Character selbst löschen conn.prepare( "delete_character", r#"DELETE FROM falukant_data.character WHERE id = $1"#, )?; conn.execute("delete_character", &[&character_id])?; Ok(()) } fn set_heir(&mut self, character_id: i32) -> Result<(), DbError> { let falukant_user_id = self.get_falukant_user_id(character_id)?; if falukant_user_id < 0 { return Ok(()); } let mut heir_id = self.get_heir_from_children(character_id)?; let mut new_money = self.calculate_new_money(falukant_user_id, heir_id > 0)?; if heir_id < 1 { heir_id = self.get_random_heir(character_id)?; new_money = self.calculate_new_money(falukant_user_id, heir_id > 0)?; } if heir_id > 0 { self.set_new_character(falukant_user_id, heir_id)?; } self.set_new_money(falukant_user_id, new_money)?; Ok(()) } fn get_falukant_user_id(&mut self, character_id: i32) -> Result { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("get_falukant_user_id", QUERY_GET_FALUKANT_USER_ID)?; let rows = conn.execute("get_falukant_user_id", &[&character_id])?; Ok(rows .first() .and_then(|r| r.get("user_id")) .and_then(|v| v.parse::().ok()) .unwrap_or(-1)) } fn get_heir_from_children(&mut self, deceased_character_id: i32) -> Result { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("get_heir", QUERY_GET_HEIR)?; let rows = conn.execute("get_heir", &[&deceased_character_id])?; Ok(rows .first() .and_then(|r| r.get("child_character_id")) .and_then(|v| v.parse::().ok()) .unwrap_or(-1)) } fn get_random_heir(&mut self, deceased_character_id: i32) -> Result { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("random_heir", QUERY_RANDOM_HEIR)?; let rows = conn.execute("random_heir", &[&deceased_character_id])?; Ok(rows .first() .and_then(|r| r.get("child_character_id")) .and_then(|v| v.parse::().ok()) .unwrap_or(-1)) } fn set_new_character( &mut self, falukant_user_id: i32, heir_character_id: i32, ) -> Result<(), DbError> { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("set_character_user", QUERY_SET_CHARACTER_USER)?; conn.execute( "set_character_user", &[&falukant_user_id, &heir_character_id], )?; Ok(()) } fn set_new_money(&mut self, falukant_user_id: i32, new_amount: f64) -> Result<(), DbError> { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("update_user_money", QUERY_UPDATE_USER_MONEY)?; conn.execute("update_user_money", &[&new_amount, &falukant_user_id])?; Ok(()) } fn get_current_money(&mut self, falukant_user_id: i32) -> Result { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("get_current_money", QUERY_GET_CURRENT_MONEY)?; let rows = conn.execute("get_current_money", &[&falukant_user_id])?; Ok(rows .first() .and_then(|r| r.get("sum")) .and_then(|v| v.parse::().ok()) .unwrap_or(0.0)) } fn get_house_value(&mut self, falukant_user_id: i32) -> Result { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("house_value", QUERY_GET_HOUSE_VALUE)?; let rows = conn.execute("house_value", &[&falukant_user_id])?; Ok(rows .first() .and_then(|r| r.get("sum")) .and_then(|v| v.parse::().ok()) .unwrap_or(0.0)) } fn get_settlement_value(&mut self, falukant_user_id: i32) -> Result { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("settlement_value", QUERY_GET_SETTLEMENT_VALUE)?; let rows = conn.execute("settlement_value", &[&falukant_user_id])?; Ok(rows .first() .and_then(|r| r.get("sum")) .and_then(|v| v.parse::().ok()) .unwrap_or(0.0)) } fn get_inventory_value(&mut self, falukant_user_id: i32) -> Result { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("inventory_value", QUERY_GET_INVENTORY_VALUE)?; let rows = conn.execute("inventory_value", &[&falukant_user_id])?; Ok(rows .first() .and_then(|r| r.get("sum")) .and_then(|v| v.parse::().ok()) .unwrap_or(0.0)) } fn get_credit_debt(&mut self, falukant_user_id: i32) -> Result { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("credit_debt", QUERY_GET_CREDIT_DEBT)?; let rows = conn.execute("credit_debt", &[&falukant_user_id])?; Ok(rows .first() .and_then(|r| r.get("sum")) .and_then(|v| v.parse::().ok()) .unwrap_or(0.0)) } fn get_child_count(&mut self, deceased_user_id: i32) -> Result { let mut conn = self .base .pool .get() .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?; conn.prepare("count_children", QUERY_COUNT_CHILDREN)?; let rows = conn.execute("count_children", &[&deceased_user_id])?; Ok(rows .first() .and_then(|r| r.get("cnt")) .and_then(|v| v.parse::().ok()) .unwrap_or(0)) } fn calculate_new_money( &mut self, falukant_user_id: i32, has_heir: bool, ) -> Result { if !has_heir { return Ok(800.0); } let cash = self.get_current_money(falukant_user_id)?; let houses = self.get_house_value(falukant_user_id)?; let settlements = self.get_settlement_value(falukant_user_id)?; let inventory = self.get_inventory_value(falukant_user_id)?; let debt = self.get_credit_debt(falukant_user_id)?; let total_assets = cash + houses + settlements + inventory - debt; let child_count = self.get_child_count(falukant_user_id)?; let single = child_count <= 1; let heir_share = if single { total_assets } else { total_assets * 0.8 }; let net = heir_share - (houses + settlements + inventory + debt); if net <= 1000.0 { Ok(1000.0) } else { Ok(net) } } } /// Kleine Hilfsfunktionen für robustes Parsen aus `Row`. fn parse_i32(row: &crate::db::Row, key: &str, default: i32) -> i32 { row.get(key) .and_then(|v| v.parse::().ok()) .unwrap_or(default) } fn parse_opt_i32(row: &crate::db::Row, key: &str) -> Option { row.get(key).and_then(|v| v.parse::().ok()) } #[derive(Debug, Clone)] struct Credit { amount: f64, remaining_amount: f64, interest_rate: i32, user_id: i32, money: f64, character_id: i32, prism_started_previously: bool, } impl Worker for UserCharacterWorker { 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| { let mut worker = UserCharacterWorker::new(pool.clone(), broker.clone()); while state.running_worker.load(Ordering::Relaxed) { worker.run_iteration(&state); } }); } fn stop_worker_thread(&mut self) { self.base.stop_worker(); } fn enable_watchdog(&mut self) { self.base.start_watchdog(); } }