yourpart-daemon/src/worker/user_character.rs

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};

/// 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<f64>,
    last_hourly_run: Option<Instant>,
    last_pregnancy_run: Option<Instant>,
    last_mood_run: Option<Instant>,
}

// SQL-Queries (1:1 aus der C++-Implementierung übernommen, gruppiert nach Themen)
const QUERY_GET_USERS_TO_UPDATE: &str = r#"
    SELECT id, CURRENT_DATE - birthdate::date AS age, health
    FROM falukant_data."character"
    WHERE user_id IS NOT NULL;
"#;

const QUERY_UPDATE_CHARACTERS_HEALTH: &str = r#"
    UPDATE falukant_data."character"
       SET health = $1
     WHERE id = $2;
"#;

// Mood-Update mit zufälliger Auswahl pro Charakter:
// Jeder lebende Charakter hat pro Aufruf (ca. 1x pro Minute)
// eine kleine Chance auf Mood-Wechsel. Die Bedingung `random() < 1.0 / 50.0`
// ergibt im Erwartungswert ca. alle 50 Minuten einen Wechsel, verteilt
// individuell und zufällig.
const QUERY_UPDATE_MOOD: &str = r#"
    UPDATE falukant_data."character" AS c
       SET mood_id = falukant_data.get_random_mood_id()
     WHERE c.health > 0
       AND random() < (1.0 / 50.0);
"#;

const QUERY_UPDATE_GET_ITEMS_TO_UPDATE: &str = r#"
    SELECT id, product_id, producer_id, quantity
      FROM falukant_log.production p
     WHERE p.production_timestamp::date < current_date;
"#;

const QUERY_UPDATE_GET_CHARACTER_IDS: &str = r#"
    SELECT fu.id AS user_id,
           c.id  AS character_id,
           c2.id AS director_id
      FROM falukant_data.falukant_user fu
      JOIN falukant_data.character c
        ON c.user_id = fu.id
 LEFT JOIN falukant_data.director d
        ON d.employer_user_id = fu.id
 LEFT JOIN falukant_data.character c2
        ON c2.id = d.director_character_id
     WHERE fu.id = $1;
"#;

const QUERY_UPDATE_KNOWLEDGE: &str = r#"
    UPDATE falukant_data.knowledge
       SET knowledge = LEAST(knowledge + $3, 100)
     WHERE character_id = $1
       AND product_id   = $2;
"#;

const QUERY_DELETE_LOG_ENTRY: &str = r#"
    DELETE FROM falukant_log.production
     WHERE id = $1;
"#;

// Kredit- und Vermögens-Queries
const QUERY_GET_OPEN_CREDITS: &str = r#"
    SELECT
        c.id                 AS credit_id,
        c.amount,
        c.remaining_amount,
        c.interest_rate,
        fu.id                AS user_id,
        fu.money,
        c2.id                AS character_id,
        dp.created_at        AS debitor_prism_start,
        dp.created_at::date < current_date AS prism_started_previously
    FROM falukant_data.credit c
    JOIN falukant_data.falukant_user fu
      ON fu.id = c.id
    JOIN falukant_data.character c2
      ON c2.user_id = c.falukant_user_id
    LEFT JOIN falukant_data.debtors_prism dp
      ON dp.character_id = c2.id
   WHERE c.remaining_amount > 0
     AND c.updated_at::date < current_date;
"#;

const QUERY_UPDATE_CREDIT: &str = r#"
    UPDATE falukant_data.credit c
       SET remaining_amount = $1
     WHERE falukant_user_id = $2;
"#;

const QUERY_CLEANUP_CREDITS: &str = r#"
    DELETE FROM falukant_data.credit
     WHERE remaining_amount >= 0.01;
"#;

const QUERY_ADD_CHARACTER_TO_DEBTORS_PRISM: &str = r#"
    INSERT INTO falukant_data.debtors_prism (character_id)
    VALUES ($1);
"#;

const QUERY_GET_CURRENT_MONEY: &str = r#"
    SELECT COALESCE(money, 0) AS sum
      FROM falukant_data.falukant_user
     WHERE user_id = $1;
"#;

const QUERY_HOUSE_VALUE: &str = r#"
    SELECT COALESCE(SUM(h.cost), 0) AS sum
      FROM falukant_data.user_house AS uh
      JOIN falukant_type.house      AS h
        ON uh.house_type_id = h.id
     WHERE uh.user_id = $1;
"#;

const QUERY_SETTLEMENT_VALUE: &str = r#"
    SELECT COALESCE(SUM(b.base_cost), 0) AS sum
      FROM falukant_data.branch AS br
      JOIN falukant_type.branch AS b
        ON br.branch_type_id = b.id
     WHERE br.falukant_user_id = $1;
"#;

const QUERY_INVENTORY_VALUE: &str = r#"
    SELECT COALESCE(SUM(i.quantity * p.sell_cost), 0) AS sum
      FROM falukant_data.inventory AS i
      JOIN falukant_type.product   AS p
        ON i.product_id = p.id
      JOIN falukant_data.branch    AS br
        ON i.stock_id = br.id
     WHERE br.falukant_user_id = $1;
"#;

const QUERY_CREDIT_DEBT: &str = r#"
    SELECT COALESCE(SUM(remaining_amount), 0) AS sum
      FROM falukant_data.credit
     WHERE falukant_user_id = $1;
"#;

const QUERY_COUNT_CHILDREN: &str = r#"
    SELECT COUNT(*) AS cnt
      FROM falukant_data.child_relation
     WHERE father_character_id = $1
        OR mother_character_id = $1;
"#;

// Vererbungs-Queries
const QUERY_GET_HEIR: &str = r#"
    SELECT child_character_id
      FROM falukant_data.child_relation
     WHERE father_character_id = $1
        OR mother_character_id = $1
  ORDER BY (is_heir IS TRUE) DESC,
           updated_at         DESC
     LIMIT 1;
"#;

const QUERY_RANDOM_HEIR: &str = r#"
    WITH chosen AS (
        SELECT
          cr.id               AS relation_id,
          cr.child_character_id
        FROM
          falukant_data.child_relation AS cr
        JOIN
          falukant_data.character      AS ch
          ON ch.id = cr.child_character_id
        WHERE
          (cr.father_character_id = $1 OR cr.mother_character_id = $1)
          AND ch.region_id = (
              SELECT region_id
                FROM falukant_data.character
               WHERE id = $1
          )
          AND ch.birthdate >= NOW() - INTERVAL '10 days'
          AND ch.title_of_nobility = (
              SELECT id
                FROM falukant_type.title
               WHERE label_tr = 'noncivil'
          )
        ORDER BY RANDOM()
        LIMIT 1
    )
    UPDATE
      falukant_data.child_relation AS cr2
    SET
      is_heir    = TRUE,
      updated_at = NOW()
    FROM
      chosen
    WHERE
      cr2.id = chosen.relation_id
    RETURNING
      chosen.child_character_id;
"#;

const QUERY_SET_CHARACTER_USER: &str = r#"
    UPDATE falukant_data.character
       SET user_id   = $1,
           updated_at = NOW()
     WHERE id = $2;
"#;

const QUERY_UPDATE_USER_MONEY: &str = r#"
    UPDATE falukant_data.falukant_user
       SET money     = $1,
           updated_at = NOW()
     WHERE user_id = $2;
"#;

const QUERY_GET_FALUKANT_USER_ID: &str = r#"
    SELECT user_id
      FROM falukant_data.character
     WHERE id = $1
     LIMIT 1;
"#;

// Schwangerschafts-Queries
const QUERY_AUTOBATISM: &str = r#"
    UPDATE falukant_data.child_relation
       SET name_set = TRUE
     WHERE id IN (
         SELECT cr.id
           FROM falukant_data.child_relation cr
           JOIN falukant_data.character c
             ON c.id = cr.child_character_id
          WHERE cr.name_set = FALSE
            AND c.birthdate < current_date - INTERVAL '5 days'
     );
"#;

const QUERY_GET_PREGNANCY_CANDIDATES: &str = r#"
    SELECT
      r.character1_id AS father_cid,
      r.character2_id AS mother_cid,
      c1.title_of_nobility,
      c1.last_name,
      c1.region_id,
      fu1.id AS father_uid,
      fu2.id AS mother_uid,
      ((CURRENT_DATE - c1.birthdate::date)
       + (CURRENT_DATE - c2.birthdate::date)) / 2 AS avg_age_days,
      100.0 /
        (1 + EXP(
             0.0647 * (
               ((CURRENT_DATE - c1.birthdate::date)
                + (CURRENT_DATE - c2.birthdate::date)) / 2
             ) - 0.0591
           )) AS prob_pct
    FROM falukant_data.relationship r
    JOIN falukant_type.relationship r2
      ON r2.id = r.relationship_type_id
     AND r2.tr = 'married'
    JOIN falukant_data.character c1
      ON c1.id = r.character1_id
    JOIN falukant_data.character c2
      ON c2.id = r.character2_id
    LEFT JOIN falukant_data.falukant_user fu1
      ON fu1.id = c1.user_id
    LEFT JOIN falukant_data.falukant_user fu2
      ON fu2.id = c2.user_id
   WHERE random() * 100 < (
      100.0 /
        (1 + EXP(
             0.11166347 * (
               ((CURRENT_DATE - c1.birthdate::date)
                + (CURRENT_DATE - c2.birthdate::date)) / 2
             ) - 2.638267
           ))
   );
"#;

const QUERY_INSERT_CHILD: &str = r#"
    INSERT INTO falukant_data.character (
      user_id,
      region_id,
      first_name,
      last_name,
      birthdate,
      gender,
      title_of_nobility,
      mood_id,
      created_at,
      updated_at
    ) VALUES (
      NULL,
      $1::int,
      (
        SELECT id
          FROM falukant_predefine.firstname
         WHERE gender = $2
         ORDER BY RANDOM()
         LIMIT 1
      ),
      $3::int,
      NOW(),
      $2::varchar,
      $4::int,
      (
        SELECT id
          FROM falukant_type.mood
         ORDER BY RANDOM()
         LIMIT 1
      ),
      NOW(),
      NOW()
    )
    RETURNING id AS child_cid;
"#;

const QUERY_INSERT_CHILD_RELATION: &str = r#"
    INSERT INTO falukant_data.child_relation (
      father_character_id,
      mother_character_id,
      child_character_id,
      name_set,
      created_at,
      updated_at
    )
    VALUES (
      $1::int,
      $2::int,
      $3::int,
      FALSE,
      NOW(), NOW()
    );
"#;

// Aufräum-Queries beim Tod eines Characters
const QUERY_DELETE_DIRECTOR: &str = r#"
    DELETE FROM falukant_data.director
     WHERE director_character_id = $1;
"#;

const QUERY_DELETE_RELATIONSHIP: &str = r#"
    DELETE FROM falukant_data.relationship
     WHERE character1_id = $1
        OR character2_id = $1;
"#;

const QUERY_DELETE_CHILD_RELATION: &str = r#"
    DELETE FROM falukant_data.child_relation
     WHERE child_character_id  = $1
        OR father_character_id = $1
        OR mother_character_id = $1;
"#;

const QUERY_DELETE_KNOWLEDGE: &str = r#"
    DELETE FROM falukant_data.knowledge
     WHERE character_id = $1;
"#;

const QUERY_DELETE_DEBTORS_PRISM: &str = r#"
    DELETE FROM falukant_data.debtors_prism
     WHERE character_id = $1;
"#;

/// Löscht alle Ämter eines Charakters und stellt anschließend sicher, dass
/// für die betroffenen Amtstyp/Region-Kombinationen nicht mehr Ämter
/// besetzt sind als durch `seats_per_region` vorgesehen.
///
/// Die überzähligen Ämter werden deterministisch nach `created_at DESC`
/// gekappt, d. h. neuere Amtsinhaber bleiben bevorzugt im Amt.
const QUERY_DELETE_POLITICAL_OFFICE: &str = r#"
    WITH removed AS (
        DELETE FROM falukant_data.political_office
         WHERE character_id = $1
        RETURNING office_type_id, region_id
    ),
    affected AS (
        SELECT DISTINCT office_type_id, region_id
          FROM removed
    ),
    seats AS (
        SELECT
            pot.id               AS office_type_id,
            rt.id                AS region_id,
            pot.seats_per_region AS seats_total
        FROM falukant_type.political_office_type AS pot
        JOIN falukant_type.region AS rt
          ON pot.region_type = rt.label_tr
        JOIN affected AS a
          ON a.office_type_id = pot.id
         AND a.region_id      = rt.id
    ),
    ranked AS (
        SELECT
            po.id,
            po.office_type_id,
            po.region_id,
            s.seats_total,
            ROW_NUMBER() OVER (
                PARTITION BY po.office_type_id, po.region_id
                ORDER BY po.created_at DESC
            ) AS rn
        FROM falukant_data.political_office AS po
        JOIN seats AS s
          ON s.office_type_id = po.office_type_id
         AND s.region_id      = po.region_id
    ),
    to_delete AS (
        SELECT id
          FROM ranked
         WHERE rn > seats_total
    )
    DELETE FROM falukant_data.political_office
     WHERE id IN (SELECT id FROM to_delete);
"#;

const QUERY_DELETE_ELECTION_CANDIDATE: &str = r#"
    DELETE FROM falukant_data.election_candidate
     WHERE character_id = $1;
"#;

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) {
            return;
        }
    }

    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<Character> = 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<Rows, DbError> {
        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<Character> {
        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::<i32>().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::<i32>().ok())
        {
            Some(id) => id,
            None => return Ok(()),
        };

        let product_id = match update_item.get("product_id").and_then(|v| v.parse::<i32>().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::<i32>().ok())
            {
                Some(id) => id,
                None => continue,
            };

            let director_id = character_row
                .get("director_id")
                .and_then(|v| v.parse::<i32>().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::<i32>().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<Credit> {
        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", &[&region_id, &gender, &last_name, &title_of_nobility])?;
        let child_cid = inserted
            .get(0)
            .and_then(|r| r.get("child_cid"))
            .and_then(|v| v.parse::<i32>().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<i32, DbError> {
        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
            .get(0)
            .and_then(|r| r.get("user_id"))
            .and_then(|v| v.parse::<i32>().ok())
            .unwrap_or(-1))
    }

    fn get_heir_from_children(&mut self, deceased_character_id: i32) -> Result<i32, DbError> {
        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
            .get(0)
            .and_then(|r| r.get("child_character_id"))
            .and_then(|v| v.parse::<i32>().ok())
            .unwrap_or(-1))
    }

    fn get_random_heir(&mut self, deceased_character_id: i32) -> Result<i32, DbError> {
        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
            .get(0)
            .and_then(|r| r.get("child_character_id"))
            .and_then(|v| v.parse::<i32>().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<f64, DbError> {
        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
            .get(0)
            .and_then(|r| r.get("sum"))
            .and_then(|v| v.parse::<f64>().ok())
            .unwrap_or(0.0))
    }

    fn get_house_value(&mut self, falukant_user_id: i32) -> Result<f64, DbError> {
        let mut conn = self
            .base
            .pool
            .get()
            .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?;

        conn.prepare("house_value", QUERY_HOUSE_VALUE)?;
        let rows = conn.execute("house_value", &[&falukant_user_id])?;

        Ok(rows
            .get(0)
            .and_then(|r| r.get("sum"))
            .and_then(|v| v.parse::<f64>().ok())
            .unwrap_or(0.0))
    }

    fn get_settlement_value(&mut self, falukant_user_id: i32) -> Result<f64, DbError> {
        let mut conn = self
            .base
            .pool
            .get()
            .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?;

        conn.prepare("settlement_value", QUERY_SETTLEMENT_VALUE)?;
        let rows = conn.execute("settlement_value", &[&falukant_user_id])?;

        Ok(rows
            .get(0)
            .and_then(|r| r.get("sum"))
            .and_then(|v| v.parse::<f64>().ok())
            .unwrap_or(0.0))
    }

    fn get_inventory_value(&mut self, falukant_user_id: i32) -> Result<f64, DbError> {
        let mut conn = self
            .base
            .pool
            .get()
            .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?;

        conn.prepare("inventory_value", QUERY_INVENTORY_VALUE)?;
        let rows = conn.execute("inventory_value", &[&falukant_user_id])?;

        Ok(rows
            .get(0)
            .and_then(|r| r.get("sum"))
            .and_then(|v| v.parse::<f64>().ok())
            .unwrap_or(0.0))
    }

    fn get_credit_debt(&mut self, falukant_user_id: i32) -> Result<f64, DbError> {
        let mut conn = self
            .base
            .pool
            .get()
            .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?;

        conn.prepare("credit_debt", QUERY_CREDIT_DEBT)?;
        let rows = conn.execute("credit_debt", &[&falukant_user_id])?;

        Ok(rows
            .get(0)
            .and_then(|r| r.get("sum"))
            .and_then(|v| v.parse::<f64>().ok())
            .unwrap_or(0.0))
    }

    fn get_child_count(&mut self, deceased_user_id: i32) -> Result<i32, DbError> {
        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
            .get(0)
            .and_then(|r| r.get("cnt"))
            .and_then(|v| v.parse::<i32>().ok())
            .unwrap_or(0))
    }

    fn calculate_new_money(
        &mut self,
        falukant_user_id: i32,
        has_heir: bool,
    ) -> Result<f64, DbError> {
        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::<i32>().ok())
        .unwrap_or(default)
}

fn parse_opt_i32(row: &crate::db::Row, key: &str) -> Option<i32> {
    row.get(key).and_then(|v| v.parse::<i32>().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<WorkerState>| {
                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();
    }
}