//! Liebhaber, Ehezufriedenheit, Ansehen, Monatskosten (Handoff: docs/FALUKANT_DAEMON_HANDOFF.md).
//! Benötigt `migrations/001_falukant_family_lovers.sql` (ggf. `002` bei Altbestand), `003` für Ehe-Buffs,
//! optional `004` + Backend-Stammdaten für Dienerschaft (`falukant_servants`),
//! `005` Ehe öffentliche Stabilität + Hausfrieden (`household_tension_score`),
//! `006` Liebschafts-Unterhalt 12× pro Spieltag (alle 2 h), 1 Spieltag = 1 Spieljahr.
//! Schuldturm/Verzug/Pfändung: `falukant_debtors::run_daily` (siehe `docs/FALUKANT_DEBTORS_DAEMON.md`).
//!
//! WebSocket: `falukantUpdateFamily` (reason) + `falukantUpdateStatus` für betroffene Nutzer.

use std::collections::{HashMap, HashSet};
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::{Duration, Instant};

use rand::distributions::{Distribution, Uniform};
use rand::rngs::StdRng;
use rand::SeedableRng;

use super::base::{BaseWorker, Worker, WorkerState};
use super::sql::{
    QUERY_COUNT_LOVER_CHILDREN_FOR_USER, QUERY_FAMILY_SCHEMA_READY,
    QUERY_GET_ACTIVE_LOVER_ROWS_FOR_DAILY, QUERY_GET_ACTIVE_LOVER_ROWS_FOR_MONTHLY,
    QUERY_GET_ACTIVE_LOVER_ROWS_FOR_INSTALLMENT, QUERY_GET_LOVER_PREGNANCY_CANDIDATES,
    QUERY_GET_MARRIAGE_ROWS, QUERY_GET_USER_HOUSE_ROW_BY_USER, QUERY_INSERT_CHILD,
    QUERY_INSERT_CHILD_RELATION_LOVER, QUERY_LOVER_BIRTH_PENALTY_MARRIAGE,
    QUERY_LOVER_BIRTH_PENALTY_REPUTATION, QUERY_LOVER_INSTALLMENT_SCHEMA_READY,
    QUERY_MARK_LOVER_DAILY_DONE, QUERY_MARK_LOVER_INSTALLMENT_AT, QUERY_MARK_LOVER_MONTHLY_DONE,
    QUERY_MARRIAGE_SUBTRACT_SATISFACTION, QUERY_RESET_LOVER_UNDERPAY_COUNTERS,
    QUERY_UPDATE_CHARACTER_REPUTATION, QUERY_UPDATE_LOVER_UNDERPAY_STATE,
    QUERY_UPDATE_LOVER_VISIBILITY_DISCRETION, QUERY_UPDATE_MARRIAGE_STATE_AND_BUFFS,
    QUERY_UPDATE_USER_HOUSE_TENSION_BY_USER,
};
use crate::db::{ConnectionPool, DbError};
use crate::message_broker::MessageBroker;

const DAILY_INTERVAL: Duration = Duration::from_secs(24 * 3600);
const MONTHLY_INTERVAL: Duration = Duration::from_secs(30 * 24 * 3600);
/// 12 Monatsticke pro Spieltag (24 h = 1 Spieljahr); 2 h = 1 Spielmonat (Liebschaft + Dienerschaft).
const GAME_MONTH_SLICE_INTERVAL: Duration = Duration::from_secs(2 * 3600);

pub struct FalukantFamilyWorker {
    base: BaseWorker,
    rng: StdRng,
    dist: Uniform<f64>,
    last_daily: Option<Instant>,
    last_monthly: Option<Instant>,
    /// Gemeinsamer Tick für Liebschafts-Raten + Dienerschaft (Monatstick ≈ 2 h).
    last_game_month_slice: Option<Instant>,
    schema_ready: bool,
    /// Migration `004_falukant_servants_daemon.sql` (Dienerschaft-Ticks).
    servants_schema_ready: bool,
    /// Migration `006_falukant_lover_installments.sql`.
    lover_installment_schema_ready: bool,
}

impl FalukantFamilyWorker {
    pub fn new(pool: ConnectionPool, broker: MessageBroker) -> Self {
        Self {
            base: BaseWorker::new("FalukantFamilyWorker", pool, broker),
            rng: StdRng::from_entropy(),
            dist: Uniform::from(0.0..1.0),
            last_daily: None,
            last_monthly: None,
            last_game_month_slice: None,
            schema_ready: false,
            servants_schema_ready: false,
            lover_installment_schema_ready: false,
        }
    }

    fn run_iteration(&mut self, state: &WorkerState) {
        self.base.set_current_step("FalukantFamilyWorker iteration");
        let now = Instant::now();

        if !self.schema_ready {
            if let Ok(true) = self.check_schema() {
                self.schema_ready = true;
            }
        }

        if Self::should_run(self.last_daily, now, DAILY_INTERVAL) {
            if let Err(e) = super::falukant_debtors::run_daily(&self.base, &self.base.broker) {
                eprintln!("[FalukantFamilyWorker] falukant_debtors::run_daily: {e}");
            }
            if let Err(e) = super::falukant_certificate::run_daily(&self.base, &self.base.broker) {
                eprintln!("[FalukantFamilyWorker] falukant_certificate::run_daily: {e}");
            }
            if self.schema_ready {
                if let Err(e) = self.process_daily() {
                    eprintln!("[FalukantFamilyWorker] process_daily: {e}");
                }
            }
            self.last_daily = Some(now);
        }

        if Self::should_run(self.last_monthly, now, MONTHLY_INTERVAL) {
            if let Err(e) = self.process_monthly() {
                eprintln!("[FalukantFamilyWorker] process_monthly: {e}");
            }
            self.last_monthly = Some(now);
        }

        if self.servants_schema_ready || self.lover_installment_schema_ready {
            if Self::should_run(self.last_game_month_slice, now, GAME_MONTH_SLICE_INTERVAL) {
                if self.servants_schema_ready {
                    if let Err(e) =
                        super::falukant_servants::run_monthly(&self.base, &self.base.broker)
                    {
                        eprintln!("[FalukantFamilyWorker] falukant_servants::run_monthly: {e}");
                    }
                }
                if self.lover_installment_schema_ready {
                    if let Err(e) = self.process_lover_installments() {
                        eprintln!("[FalukantFamilyWorker] process_lover_installments: {e}");
                    }
                }
                self.last_game_month_slice = Some(now);
            }
        }

        std::thread::sleep(Duration::from_secs(1));
        if !state.running_worker.load(Ordering::Relaxed) {
            // stopping
        }
    }

    fn should_run(last: Option<Instant>, now: Instant, interval: Duration) -> bool {
        match last {
            None => true,
            Some(t) => now.saturating_duration_since(t) >= interval,
        }
    }

    fn check_schema(&mut self) -> Result<bool, DbError> {
        let mut conn = self
            .base
            .pool
            .get()
            .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?;
        conn.prepare("family_schema", QUERY_FAMILY_SCHEMA_READY)?;
        let rows = conn.execute("family_schema", &[])?;
        let family_ok = rows
            .first()
            .and_then(|r| r.get("ready"))
            .map(|v| v == "true" || v == "t")
            .unwrap_or(false);
        self.servants_schema_ready =
            super::falukant_servants::servants_schema_ready(&self.base.pool).unwrap_or(false);
        conn.prepare("lover_inst_schema", QUERY_LOVER_INSTALLMENT_SCHEMA_READY)?;
        let inst_rows = conn.execute("lover_inst_schema", &[])?;
        self.lover_installment_schema_ready = inst_rows
            .first()
            .and_then(|r| r.get("ready"))
            .map(|v| v == "true" || v == "t")
            .unwrap_or(false);
        Ok(family_ok)
    }

    fn process_daily(&mut self) -> Result<(), DbError> {
        if self.servants_schema_ready {
            super::falukant_servants::run_daily(&self.base, &self.base.broker)?;
        }

        let mut conn = self
            .base
            .pool
            .get()
            .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?;

        conn.prepare("get_lovers", QUERY_GET_ACTIVE_LOVER_ROWS_FOR_DAILY)?;
        let lover_rows = conn.execute("get_lovers", &[])?;

        conn.prepare("get_marriages", QUERY_GET_MARRIAGE_ROWS)?;
        let marriage_rows = conn.execute("get_marriages", &[])?;

        let mut marriages: Vec<MarriageData> = marriage_rows
            .into_iter()
            .filter_map(|r| {
                Some(MarriageData {
                    id: parse_i32(&r, "marriage_id", -1),
                    m1: parse_i32(&r, "m1", -1),
                    m2: parse_i32(&r, "m2", -1),
                    satisfaction: parse_i32(&r, "marriage_satisfaction", 55),
                    public_stability: parse_i32(&r, "marriage_public_stability", 55),
                    drift_high: parse_i32(&r, "marriage_drift_high", 0),
                    drift_low: parse_i32(&r, "marriage_drift_low", 0),
                    title1_tr: r.get("title1_tr").cloned().unwrap_or_default(),
                    title2_tr: r.get("title2_tr").cloned().unwrap_or_default(),
                    user1_id: parse_opt_i32(&r, "user1_id"),
                    user2_id: parse_opt_i32(&r, "user2_id"),
                    gift_days: parse_i32(&r, "marriage_gift_buff_days_remaining", 0),
                    feast_pending: parse_i32(&r, "marriage_pending_feast_bonus", 0),
                    house_supply: parse_i32(&r, "marriage_house_supply", 50),
                    no_lover_counter: parse_i32(&r, "marriage_no_lover_bonus_counter", 0),
                    household_order_1: parse_i32(&r, "household_order_1", 55),
                    household_order_2: parse_i32(&r, "household_order_2", 55),
                    servant_quality_1: parse_i32(&r, "servant_quality_1", 50),
                    servant_quality_2: parse_i32(&r, "servant_quality_2", 50),
                })
            })
            .filter(|m| m.id > 0)
            .collect();

        let mut lovers: Vec<LoverData> = lover_rows
            .into_iter()
            .filter_map(|r| {
                Some(LoverData {
                    rel_id: parse_i32(&r, "rel_id", -1),
                    c1: parse_i32(&r, "c1", -1),
                    c2: parse_i32(&r, "c2", -1),
                    lover_role: r.get("lover_role").cloned().unwrap_or_default(),
                    affection: parse_i32(&r, "affection", 50),
                    visibility: parse_i32(&r, "visibility", 0),
                    discretion: parse_i32(&r, "discretion", 50),
                    maintenance_level: parse_i32(&r, "maintenance_level", 50),
                    status_fit: parse_i32(&r, "status_fit", 0),
                    scandal_extra: parse_i32(&r, "scandal_extra_daily_pct", 0),
                    months_underfunded: parse_i32(&r, "months_underfunded", 0),
                    acknowledged: parse_bool_row(&r, "acknowledged"),
                    title1_tr: r.get("title1_tr").cloned().unwrap_or_default(),
                    title2_tr: r.get("title2_tr").cloned().unwrap_or_default(),
                    user1_id: parse_opt_i32(&r, "user1_id"),
                    user2_id: parse_opt_i32(&r, "user2_id"),
                    min_age_years: parse_i32(&r, "min_age_years", 99),
                    servant_disc_u1: parse_i32(&r, "servant_disc_u1", 0),
                    servant_disc_u2: parse_i32(&r, "servant_disc_u2", 0),
                })
            })
            .filter(|l| l.rel_id > 0)
            .collect();

        let mut char_user: HashMap<i32, Option<i32>> = HashMap::new();
        for l in &lovers {
            char_user.insert(l.c1, l.user1_id);
            char_user.insert(l.c2, l.user2_id);
        }

        let mut char_lover_count: std::collections::HashMap<i32, usize> =
            std::collections::HashMap::new();
        let mut char_visible_lover_rel_count: std::collections::HashMap<i32, usize> =
            std::collections::HashMap::new();
        for l in &lovers {
            *char_lover_count.entry(l.c1).or_insert(0) += 1;
            *char_lover_count.entry(l.c2).or_insert(0) += 1;
            if l.visibility >= 60 {
                *char_visible_lover_rel_count.entry(l.c1).or_insert(0) += 1;
                *char_visible_lover_rel_count.entry(l.c2).or_insert(0) += 1;
            }
        }

        conn.prepare("upd_vd", QUERY_UPDATE_LOVER_VISIBILITY_DISCRETION)?;
        for l in &mut lovers {
            let ehe_conflict = marriages.iter().any(|m| {
                (m.m1 == l.c1 || m.m2 == l.c1 || m.m1 == l.c2 || m.m2 == l.c2) && m.satisfaction < 40
            });

            let mut v_net = 0i32;
            if l.maintenance_level < 35 {
                v_net += 1;
            }
            if l.affection < 30 {
                v_net += 1;
            }
            if l.status_fit == -2 {
                v_net += 2;
            }
            if ehe_conflict {
                v_net += 1;
            }
            if l.discretion >= 60 {
                v_net -= 1;
            }
            if l.maintenance_level >= 70 {
                v_net -= 1;
            }
            let comb = (l.servant_disc_u1 + l.servant_disc_u2) / 2;
            v_net += comb / 4;
            let new_vis = clamp_i32(l.visibility + v_net, 0, 100);

            let mut d_net = 0i32;
            if l.maintenance_level >= 70 {
                d_net += 1;
            }
            if l.maintenance_level < 35 {
                d_net -= 1;
            }
            if l.visibility > 60 {
                d_net -= 1;
            }
            let new_disc = clamp_i32(l.discretion + d_net, 0, 100);

            conn.execute(
                "upd_vd",
                &[&new_vis, &new_disc, &l.rel_id],
            )?;
            l.visibility = new_vis;
            l.discretion = new_disc;
        }

        conn.prepare("mark_daily", QUERY_MARK_LOVER_DAILY_DONE)?;
        for l in &lovers {
            conn.execute("mark_daily", &[&l.rel_id])?;
        }

        let mut marriage_socket_users: HashSet<i32> = HashSet::new();

        conn.prepare("upd_marriage_full", QUERY_UPDATE_MARRIAGE_STATE_AND_BUFFS)?;

        for m in marriages.iter_mut() {
            let touching: Vec<&LoverData> = lovers
                .iter()
                .filter(|l| {
                    let (a, b) = (l.c1, l.c2);
                    ((a == m.m1 && b != m.m2) || (b == m.m1 && a != m.m2))
                        || ((a == m.m2 && b != m.m1) || (b == m.m2 && a != m.m1))
                })
                .collect();

            let touching_empty = touching.is_empty();
            let sat0 = m.satisfaction;
            let pst0 = m.public_stability;
            let dh0 = m.drift_high;
            let dl0 = m.drift_low;
            let g0 = m.gift_days;
            let f0 = m.feast_pending;
            let n0 = m.no_lover_counter;

            let mut sat = m.satisfaction;
            let mut dh = m.drift_high;
            let mut dl = m.drift_low;
            let mut gift_days = m.gift_days;
            let mut feast = m.feast_pending;
            let mut nl_counter = m.no_lover_counter;
            let house_supply = m.house_supply;

            if touching_empty {
                if sat > 55 {
                    dh += 1;
                    if dh >= 3 {
                        sat = (sat - 1).max(0);
                        dh = 0;
                    }
                } else if sat < 55 {
                    dl += 1;
                    if dl >= 5 {
                        sat = (sat + 1).min(100);
                        dl = 0;
                    }
                }
            } else {
                let mg = marriage_rank_group(&m.title1_tr, &m.title2_tr);
                let mistress_n = touching
                    .iter()
                    .filter(|l| l.lover_role == "mistress_or_favorite")
                    .count();
                let total_lovers_here = touching.len();
                let mut delta = 0i32;
                for l in &touching {
                    delta += lover_marriage_daily_delta(
                        mg,
                        l,
                        m.satisfaction,
                        mistress_n,
                        total_lovers_here,
                    );
                }
                if touching.iter().any(|l| l.visibility >= 60) {
                    delta -= 2;
                }
                if total_lovers_here >= 2 {
                    delta -= 2;
                }
                if touching
                    .iter()
                    .any(|l| l.lover_role == "mistress_or_favorite" && l.maintenance_level < 35)
                {
                    delta -= 1;
                }
                if touching.iter().any(|l| l.min_age_years <= 15) {
                    delta -= 1;
                }
                sat = clamp_i32(sat + delta, 0, 100);
            }

            Self::apply_marriage_positive_buffs(
                &mut sat,
                &mut gift_days,
                &mut feast,
                &mut nl_counter,
                touching_empty,
                house_supply,
            );

            m.satisfaction = sat;
            m.drift_high = dh;
            m.drift_low = dl;
            m.gift_days = gift_days;
            m.feast_pending = feast;
            m.no_lover_counter = nl_counter;

            let mut pst = m.public_stability;
            if !touching_empty {
                let max_vis = touching.iter().map(|l| l.visibility).max().unwrap_or(0);
                pst -= (max_vis / 25).min(4);
                if touching.iter().any(|l| l.visibility >= 60) {
                    pst -= 2;
                }
                if touching.iter().any(|l| l.months_underfunded >= 1) {
                    pst -= 1;
                }
                if touching.iter().any(|l| l.months_underfunded >= 3) {
                    pst -= 1;
                }
                if touching.iter().any(|l| l.acknowledged) {
                    pst -= 2;
                }
                let g1 = title_group(&m.title1_tr);
                let g2 = title_group(&m.title2_tr);
                if (g1 as i32 - g2 as i32).abs() >= 2 {
                    pst -= 1;
                }
                if touching.iter().any(|l| l.min_age_years <= 15) {
                    pst -= 2;
                }
                let ho_avg = (m.household_order_1 + m.household_order_2) / 2;
                if ho_avg < 35 {
                    pst -= 2;
                } else if ho_avg < 50 {
                    pst -= 1;
                }
                let sq_avg = (m.servant_quality_1 + m.servant_quality_2) / 2;
                if sq_avg < 40 {
                    pst -= 1;
                }
                if sat < 40 {
                    pst -= 1;
                }
            } else if pst < 55 {
                pst += 1;
            } else if pst > 55 {
                pst -= 1;
            }
            pst = clamp_i32(pst, 0, 100);
            m.public_stability = pst;

            conn.execute(
                "upd_marriage_full",
                &[&sat, &dh, &dl, &gift_days, &feast, &nl_counter, &pst, &m.id],
            )?;

            if sat != sat0
                || pst != pst0
                || dh != dh0
                || dl != dl0
                || gift_days != g0
                || feast != f0
                || nl_counter != n0
            {
                push_user_id(&mut marriage_socket_users, m.user1_id);
                push_user_id(&mut marriage_socket_users, m.user2_id);
            }
        }

        // Hausfrieden: Spannungswert 0..100 persistieren (Handoff Ehe & Hausfrieden)
        let mut tension_users: HashSet<i32> = HashSet::new();
        for l in &lovers {
            push_user_id(&mut tension_users, l.user1_id);
            push_user_id(&mut tension_users, l.user2_id);
        }
        for m in &marriages {
            push_user_id(&mut tension_users, m.user1_id);
            push_user_id(&mut tension_users, m.user2_id);
        }

        let mut tension_agg: HashMap<i32, UserTensionAgg> = HashMap::new();
        for uid in &tension_users {
            tension_agg.insert(*uid, UserTensionAgg::default());
        }
        for l in &lovers {
            for uid in [l.user1_id, l.user2_id].into_iter().flatten() {
                if let Some(a) = tension_agg.get_mut(&uid) {
                    a.max_lover_visibility = a.max_lover_visibility.max(l.visibility);
                    if l.acknowledged {
                        a.n_acknowledged += 1;
                    }
                    a.max_months_underfunded = a.max_months_underfunded.max(l.months_underfunded);
                }
            }
        }
        for m in &marriages {
            for uid in [m.user1_id, m.user2_id].into_iter().flatten() {
                if let Some(a) = tension_agg.get_mut(&uid) {
                    a.min_marriage_sat = a.min_marriage_sat.min(m.satisfaction);
                }
            }
        }

        conn.prepare("upd_tension", QUERY_UPDATE_USER_HOUSE_TENSION_BY_USER)?;
        conn.prepare("child_cnt_lover", QUERY_COUNT_LOVER_CHILDREN_FOR_USER)?;
        conn.prepare("uh_row_u", QUERY_GET_USER_HOUSE_ROW_BY_USER)?;

        let mut tension_socket_users: HashSet<i32> = HashSet::new();
        for (&uid, agg) in &tension_agg {
            let child_rows = conn.execute("child_cnt_lover", &[&uid])?;
            let lover_children = child_rows
                .first()
                .map(|r| parse_i32(r, "cnt", 0))
                .unwrap_or(0);

            let house_rows = conn.execute("uh_row_u", &[&uid])?;
            let Some(hr) = house_rows.first() else {
                continue;
            };
            let ho = parse_i32(hr, "household_order", 55);
            let sq = parse_i32(hr, "servant_quality", 50);
            let pay = hr
                .get("servant_pay_level")
                .cloned()
                .unwrap_or_else(|| "normal".into());
            let prev_ts = parse_i32(hr, "prev_tension_score", 0);

            let score = compute_household_tension_score(agg, lover_children, ho, sq, pay.as_str());
            if score != prev_ts {
                conn.execute("upd_tension", &[&score, &uid])?;
                tension_socket_users.insert(uid);
            }
        }

        conn.prepare("upd_rep", QUERY_UPDATE_CHARACTER_REPUTATION)?;
        for l in &lovers {
            let g = pair_rank_group(&l.title1_tr, &l.title2_tr);
            let scandal = l.visibility > 70;
            let order_ok = l.maintenance_level >= 65
                && l.discretion >= 60
                && l.visibility <= 35
                && mistress_count_for_pair(&lovers, l.c1, l.c2) <= 1;
            let rm = rank_rep_modifier(g, scandal, order_ok);
            let vf = 0.4 + (l.visibility as f64 / 100.0) * 1.6;
            let base = match l.lover_role.as_str() {
                "secret_affair" => -0.2,
                "lover" => -0.4,
                "mistress_or_favorite" => -0.6,
                _ => -0.4,
            };
            let mut delta = base * vf * rm;

            if l.lover_role == "mistress_or_favorite" && order_ok {
                if g == 2 {
                    delta = 0.1;
                } else if g == 3 {
                    delta = 0.2;
                }
            }

            delta = delta.clamp(-3.0, 1.0);

            let age_rep = age_reputation_delta(l.min_age_years);
            let vis_young = visibility_young_penalty(l.min_age_years, l.visibility);
            let final_delta = delta + age_rep + vis_young;

            for cid in [l.c1, l.c2] {
                let cur = fetch_reputation(&mut conn, cid)?;
                let new_rep = (cur + final_delta).clamp(0.0, 100.0);
                let s = format!("{:.2}", new_rep);
                conn.execute("upd_rep", &[&s, &cid])?;
            }

            let married_c1 = marriages
                .iter()
                .any(|m| m.m1 == l.c1 || m.m2 == l.c1);
            let married_c2 = marriages
                .iter()
                .any(|m| m.m1 == l.c2 || m.m2 == l.c2);
            let mut p = 1.0
                + (l.visibility as f64) / 25.0
                + if married_c1 || married_c2 { 2.0 } else { 0.0 }
                + if l.status_fit == -2 { 2.0 } else { 0.0 }
                + if l.maintenance_level < 25 { 3.0 } else { 0.0 }
                + if char_lover_count.get(&l.c1).copied().unwrap_or(0) >= 2
                    || char_lover_count.get(&l.c2).copied().unwrap_or(0) >= 2
                {
                    3.0
                } else {
                    0.0
                }
                + l.scandal_extra as f64
                + scandal_age_extra_pct(l.min_age_years);
            if l.discretion >= 75 {
                p -= 2.0;
            }
            if g == 3
                && l.lover_role == "mistress_or_favorite"
                && l.maintenance_level >= 65
                && l.visibility <= 40
            {
                p -= 2.0;
            }
            p = p.clamp(0.0, 25.0);
            if self.dist.sample(&mut self.rng) * 100.0 < p {
                let _ = self.base.broker.publish(format!(
                    r#"{{"event":"falukant_family_scandal_hint","relationship_id":{}}}"#,
                    l.rel_id
                ));
                if let Some(uid) = l.user1_id.filter(|x| *x > 0) {
                    self.publish_falukant_update_family_and_status(uid, "scandal");
                }
                if let Some(uid) = l.user2_id.filter(|x| *x > 0) {
                    self.publish_falukant_update_family_and_status(uid, "scandal");
                }
            }
        }

        // Zwei sichtbare Liebschaften (visibility ≥ 60) pro Person: zusätzlich −4 Ansehen.
        for (&cid, &n_vis) in &char_visible_lover_rel_count {
            if n_vis < 2 {
                continue;
            }
            let cur = fetch_reputation(&mut conn, cid)?;
            let new_rep = (cur - 4.0).clamp(0.0, 100.0);
            let s = format!("{:.2}", new_rep);
            conn.execute("upd_rep", &[&s, &cid])?;
        }

        // Seltene Einmal-Malus (Gerücht / Tadel) für beteiligte Charaktere.
        let mut unique_cids: HashSet<i32> = HashSet::new();
        for l in &lovers {
            unique_cids.insert(l.c1);
            unique_cids.insert(l.c2);
        }
        let mut rng_malus_cids: Vec<i32> = Vec::new();
        for cid in unique_cids.iter().copied() {
            let r = self.dist.sample(&mut self.rng) * 100.0;
            let malus = if r < 1.0 {
                -5.0
            } else if r < 3.0 {
                -3.0
            } else {
                0.0
            };
            if malus < 0.0 {
                let cur = fetch_reputation(&mut conn, cid)?;
                let new_rep = (cur + malus).clamp(0.0, 100.0);
                let s = format!("{:.2}", new_rep);
                conn.execute("upd_rep", &[&s, &cid])?;
                rng_malus_cids.push(cid);
            }
        }

        let mut notify: HashSet<i32> = HashSet::new();
        for l in &lovers {
            push_user_id(&mut notify, l.user1_id);
            push_user_id(&mut notify, l.user2_id);
        }
        for (&cid, &n_vis) in &char_visible_lover_rel_count {
            if n_vis >= 2 {
                push_user_id(&mut notify, char_user.get(&cid).copied().flatten());
            }
        }
        for cid in &rng_malus_cids {
            push_user_id(&mut notify, char_user.get(cid).copied().flatten());
        }
        notify.extend(marriage_socket_users);
        notify.extend(tension_socket_users);
        self.publish_falukant_update_family_batch(&notify, "daily");

        drop(conn);
        // Liebschafts-Geburt: früher nur alle ~30 Tage in process_monthly — zu selten für kurze Testphasen.
        if let Err(e) = self.process_lover_births() {
            eprintln!("[FalukantFamilyWorker] process_lover_births: {e}");
        }

        Ok(())
    }

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

        conn.prepare("get_lovers_m", QUERY_GET_ACTIVE_LOVER_ROWS_FOR_MONTHLY)?;
        let lover_rows = conn.execute("get_lovers_m", &[])?;
        conn.prepare("mark_monthly", QUERY_MARK_LOVER_MONTHLY_DONE)?;

        let mut monthly_notify: HashSet<i32> = HashSet::new();

        for r in lover_rows {
            let rel_id = parse_i32(&r, "rel_id", -1);
            if rel_id < 0 {
                continue;
            }
            let u1 = parse_opt_i32(&r, "user1_id");
            let u2 = parse_opt_i32(&r, "user2_id");
            conn.execute("mark_monthly", &[&rel_id])?;
            push_user_id(&mut monthly_notify, u1);
            push_user_id(&mut monthly_notify, u2);
        }

        self.publish_falukant_update_family_batch(&monthly_notify, "monthly");

        drop(conn);
        self.process_lover_births()?;
        Ok(())
    }

    /// Ein Zwölftel des bisherigen Monatsunterhalts, alle 2 h (12 Spielmonate pro Spieljahr).
    fn process_lover_installments(&mut self) -> Result<(), DbError> {
        let mut conn = self
            .base
            .pool
            .get()
            .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?;

        conn.prepare("get_lovers_i", QUERY_GET_ACTIVE_LOVER_ROWS_FOR_INSTALLMENT)?;
        let lover_rows = conn.execute("get_lovers_i", &[])?;

        conn.prepare("upd_under", QUERY_UPDATE_LOVER_UNDERPAY_STATE)?;
        conn.prepare("reset_under", QUERY_RESET_LOVER_UNDERPAY_COUNTERS)?;
        conn.prepare("mar_sub", QUERY_MARRIAGE_SUBTRACT_SATISFACTION)?;
        conn.prepare("mark_inst", QUERY_MARK_LOVER_INSTALLMENT_AT)?;

        let mut notify: HashSet<i32> = HashSet::new();

        for r in lover_rows {
            let rel_id = parse_i32(&r, "rel_id", -1);
            if rel_id < 0 {
                continue;
            }
            let lover_role = r.get("lover_role").cloned().unwrap_or_default();
            let maintenance_level = parse_i32(&r, "maintenance_level", 50);
            let status_fit = parse_i32(&r, "status_fit", 0);
            let mut base = parse_i32(&r, "monthly_base_cost", 0);
            if base <= 0 {
                base = match lover_role.as_str() {
                    "secret_affair" => 10,
                    "lover" => 30,
                    "mistress_or_favorite" => 80,
                    _ => 30,
                };
            }
            let title1 = r.get("title1_tr").cloned().unwrap_or_default();
            let title2 = r.get("title2_tr").cloned().unwrap_or_default();
            let g = pair_rank_group(&title1, &title2);
            let rank_m = rank_cost_multiplier(g);
            let maint_f = 0.6 + (maintenance_level as f64 / 100.0) * 1.2;
            let sf_m = match status_fit {
                -2 => 1.35,
                -1 => 1.15,
                _ => 1.0,
            };
            let cost = ((base as f64) * rank_m * maint_f * sf_m).round() as i32;
            let installment = ((cost as f64) / 12.0 * 100.0).round() / 100.0;

            let u1 = parse_opt_i32(&r, "user1_id");
            let u2 = parse_opt_i32(&r, "user2_id");
            let payer = u1.or(u2).filter(|x| *x > 0);

            let affection = parse_i32(&r, "affection", 50);
            let visibility = parse_i32(&r, "visibility", 0);
            let discretion = parse_i32(&r, "discretion", 50);
            let mut consec = parse_i32(&r, "months_underfunded", 0);
            let mut scandal_extra = parse_i32(&r, "scandal_extra_daily_pct", 0);

            if let Some(uid) = payer {
                if installment <= 0.0 {
                    conn.execute("mark_inst", &[&rel_id])?;
                } else {
                    let money = self.get_user_money(uid)?;
                    if money >= installment {
                        self.base.change_falukant_user_money(
                            uid,
                            -installment,
                            "lover maintenance",
                        )?;
                        conn.execute("reset_under", &[&rel_id])?;
                        conn.execute("mark_inst", &[&rel_id])?;
                    } else {
                        let new_aff = clamp_i32(affection - 8, 0, 100);
                        let new_disc = clamp_i32(discretion - 6, 0, 100);
                        let new_vis = clamp_i32(visibility + 8, 0, 100);
                        consec += 1;
                        if consec >= 2 {
                            scandal_extra = (scandal_extra + 2).min(100);
                        }
                        conn.execute(
                            "upd_under",
                            &[
                                &new_aff,
                                &new_disc,
                                &new_vis,
                                &consec,
                                &scandal_extra,
                                &rel_id,
                            ],
                        )?;

                        for cid in [parse_i32(&r, "c1", 0), parse_i32(&r, "c2", 0)] {
                            if cid <= 0 {
                                continue;
                            }
                            if let Ok(Some(mid)) = marriage_id_for_character(&mut conn, cid) {
                                conn.execute("mar_sub", &[&mid, &4])?;
                            }
                        }

                        if visibility >= 40 {
                            for cid in [parse_i32(&r, "c1", 0), parse_i32(&r, "c2", 0)] {
                                if cid > 0 {
                                    let cur = fetch_reputation(&mut conn, cid)?;
                                    let s = format!("{:.2}", (cur - 1.0).max(0.0));
                                    conn.prepare("upd_rep", QUERY_UPDATE_CHARACTER_REPUTATION)?;
                                    conn.execute("upd_rep", &[&s, &cid])?;
                                }
                            }
                        }
                        conn.execute("mark_inst", &[&rel_id])?;
                    }
                }
            } else {
                conn.execute("mark_inst", &[&rel_id])?;
            }

            push_user_id(&mut notify, u1);
            push_user_id(&mut notify, u2);
        }

        self.publish_falukant_update_family_batch(&notify, "lover_installment");
        Ok(())
    }

    fn get_user_money(&mut self, user_id: i32) -> Result<f64, DbError> {
        use super::sql::{QUERY_GET_CURRENT_MONEY};
        let mut conn = self
            .base
            .pool
            .get()
            .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?;
        conn.prepare("gcm", QUERY_GET_CURRENT_MONEY)?;
        let rows = conn.execute("gcm", &[&user_id])?;
        Ok(rows
            .first()
            .and_then(|r| r.get("sum"))
            .and_then(|v| v.parse::<f64>().ok())
            .unwrap_or(0.0))
    }

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

        conn.prepare("lover_preg", QUERY_GET_LOVER_PREGNANCY_CANDIDATES)?;
        let rows = conn.execute("lover_preg", &[])?;

        conn.prepare("insert_child", QUERY_INSERT_CHILD)?;
        conn.prepare("insert_child_rel_lover", QUERY_INSERT_CHILD_RELATION_LOVER)?;
        conn.prepare("pen_mar", QUERY_LOVER_BIRTH_PENALTY_MARRIAGE)?;
        conn.prepare("pen_rep", QUERY_LOVER_BIRTH_PENALTY_REPUTATION)?;

        for row in rows {
            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 {
                continue;
            }
            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
                .first()
                .and_then(|r| r.get("child_cid"))
                .and_then(|v| v.parse::<i32>().ok())
                .unwrap_or(-1);
            if child_cid < 0 {
                continue;
            }

            conn.execute(
                "insert_child_rel_lover",
                &[&father_cid, &mother_cid, &child_cid],
            )?;

            conn.execute("pen_mar", &[&father_cid])?;
            conn.execute("pen_mar", &[&mother_cid])?;
            conn.execute("pen_rep", &[&father_cid])?;
            conn.execute("pen_rep", &[&mother_cid])?;

            if let Some(uid) = father_uid {
                self.publish_children(uid);
            }
            if let Some(uid) = mother_uid {
                self.publish_children(uid);
            }
        }

        Ok(())
    }

    fn publish_children(&self, user_id: i32) {
        let children_update =
            format!(r#"{{"event":"children_update","user_id":{}}}"#, user_id);
        self.base.broker.publish(children_update);
        self.publish_falukant_update_family_and_status(user_id, "lover_birth");
    }

    /// `falukantUpdateFamily` (mit `reason`) + `falukantUpdateStatus` — für UI-Refresh.
    fn publish_falukant_update_family_and_status(&self, user_id: i32, reason: &str) {
        let family = format!(
            r#"{{"event":"falukantUpdateFamily","user_id":{},"reason":"{}"}}"#,
            user_id, reason
        );
        self.base.broker.publish(family);
        let status = format!(r#"{{"event":"falukantUpdateStatus","user_id":{}}}"#, user_id);
        self.base.broker.publish(status);
    }

    fn publish_falukant_update_family_batch(&self, user_ids: &HashSet<i32>, reason: &str) {
        for &uid in user_ids {
            self.publish_falukant_update_family_and_status(uid, reason);
        }
    }

    /// Fest (einmal), Geschenk (−1 Tag, +1/Tag), Haus hoch & keine Liebschaft: alle 4 Tage +1.
    fn apply_marriage_positive_buffs(
        sat: &mut i32,
        gift_days: &mut i32,
        feast: &mut i32,
        nl_counter: &mut i32,
        touching_empty: bool,
        house_supply: i32,
    ) {
        if *feast > 0 {
            let bonus = (*feast).min(20).min(5);
            *sat = (*sat + bonus).min(100);
            *feast = 0;
        }
        if *gift_days > 0 {
            *sat = (*sat + 1).min(100);
            *gift_days -= 1;
        }
        if touching_empty && house_supply >= 65 {
            *nl_counter += 1;
            if *nl_counter >= 4 {
                *sat = (*sat + 1).min(100);
                *nl_counter = 0;
            }
        } else {
            *nl_counter = 0;
        }
    }
}

struct MarriageData {
    id: i32,
    m1: i32,
    m2: i32,
    satisfaction: i32,
    /// Öffentliche Ehe-Stabilität (Daemon-Drift).
    public_stability: i32,
    drift_high: i32,
    drift_low: i32,
    title1_tr: String,
    title2_tr: String,
    user1_id: Option<i32>,
    user2_id: Option<i32>,
    /// Tage mit +1 Zufriedenheit (Geschenk), runtergezählt.
    gift_days: i32,
    /// Einmal-Bonus (z. B. Fest), wird beim nächsten Daily verbraucht.
    feast_pending: i32,
    /// Hausversorgung 0–100; ab ~65: Bonus alle 4 Tage ohne Liebschaft.
    house_supply: i32,
    /// Tageszähler für Haus-Bonus (0..4).
    no_lover_counter: i32,
    household_order_1: i32,
    household_order_2: i32,
    servant_quality_1: i32,
    servant_quality_2: i32,
}

struct LoverData {
    rel_id: i32,
    c1: i32,
    c2: i32,
    lover_role: String,
    affection: i32,
    visibility: i32,
    discretion: i32,
    maintenance_level: i32,
    status_fit: i32,
    scandal_extra: i32,
    months_underfunded: i32,
    acknowledged: bool,
    title1_tr: String,
    title2_tr: String,
    user1_id: Option<i32>,
    user2_id: Option<i32>,
    /// Jüngeres Alter beider Partner (Jahre, ganzzahlig); für Altersmalus / Skandal / Ehe.
    min_age_years: i32,
    /// Dienerschaft: Diskretionsmodifikator je Haushalt (user_house), MAX pro User.
    servant_disc_u1: i32,
    servant_disc_u2: i32,
}

/// Aggregation für `household_tension_score` (0..100) pro Falukant-User.
#[derive(Clone)]
struct UserTensionAgg {
    max_lover_visibility: i32,
    n_acknowledged: i32,
    max_months_underfunded: i32,
    /// Mindest-Ehezufriedenheit aller Ehen dieses Users (100 = keine belastende Ehe).
    min_marriage_sat: i32,
}

impl Default for UserTensionAgg {
    fn default() -> Self {
        Self {
            max_lover_visibility: 0,
            n_acknowledged: 0,
            max_months_underfunded: 0,
            min_marriage_sat: 100,
        }
    }
}

fn push_user_id(set: &mut HashSet<i32>, uid: Option<i32>) {
    if let Some(id) = uid.filter(|x| *x > 0) {
        set.insert(id);
    }
}

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())
}

fn parse_bool_row(row: &crate::db::Row, key: &str) -> bool {
    row.get(key)
        .map(|v| {
            matches!(
                v.as_str(),
                "t" | "true" | "True" | "1" | "TRUE"
            )
        })
        .unwrap_or(false)
}

fn clamp_i32(v: i32, lo: i32, hi: i32) -> i32 {
    v.max(lo).min(hi)
}

/// Hausfrieden 0..100 (persistiert als `user_house.household_tension_score`).
fn compute_household_tension_score(
    agg: &UserTensionAgg,
    lover_children: i32,
    household_order: i32,
    servant_quality: i32,
    servant_pay: &str,
) -> i32 {
    let mut s: f64 = 0.0;
    s += agg.max_lover_visibility as f64 * 0.35;
    s += agg.n_acknowledged as f64 * 12.0;
    s += agg.max_months_underfunded as f64 * 8.0;
    s += lover_children as f64 * 10.0;
    s += (100 - household_order.clamp(0, 100)) as f64 * 0.22;
    if servant_pay == "low" {
        s += 6.0;
    }
    s += (50 - servant_quality.min(50)).max(0) as f64 * 0.35;
    if agg.min_marriage_sat < 100 {
        if agg.min_marriage_sat < 40 {
            s += 18.0;
        } else if agg.min_marriage_sat < 55 {
            s += 8.0;
        }
    }
    s = s.clamp(0.0, 100.0);
    s.round() as i32
}

fn title_group(tr: &str) -> u8 {
    match tr {
        "noncivil" | "civil" | "sir" => 0,
        "townlord" | "by" | "landlord" => 1,
        "knight" | "baron" | "count" | "palsgrave" | "margrave" | "landgrave" => 2,
        "ruler" | "elector"
        | "imperial-prince"
        | "duke"
        | "grand-duke"
        | "prince-regent"
        | "king" => 3,
        _ => 0,
    }
}

fn marriage_rank_group(t1: &str, t2: &str) -> u8 {
    title_group(t1).max(title_group(t2))
}

fn pair_rank_group(t1: &str, t2: &str) -> u8 {
    title_group(t1).max(title_group(t2))
}

fn rank_cost_multiplier(g: u8) -> f64 {
    match g {
        0 => 1.0,
        1 => 1.6,
        2 => 2.6,
        3 => 4.0,
        _ => 1.0,
    }
}

/// Gruppe 3: 0,7 nur bei „geordneter“ Liebschaft; sonst 1,0; Skandal 1,5.
fn rank_rep_modifier(g: u8, scandal: bool, order_ok: bool) -> f64 {
    if g == 3 && scandal {
        return 1.5;
    }
    match g {
        0 => 1.8,
        1 => 1.3,
        2 => 1.0,
        3 => if order_ok { 0.7 } else { 1.0 },
        _ => 1.0,
    }
}

/// Spec 5a: Altersmalus (jüngeres Alter der Beteiligten), zusätzlich zur Basis-Reputation.
fn age_reputation_delta(min_age_years: i32) -> f64 {
    if min_age_years <= 13 {
        -1.5
    } else if min_age_years <= 15 {
        -0.8
    } else if min_age_years <= 17 {
        -0.3
    } else {
        0.0
    }
}

/// Spec 5a: minAge <= 15 und hohe Sichtbarkeit.
fn visibility_young_penalty(min_age_years: i32, visibility: i32) -> f64 {
    if min_age_years <= 15 && visibility >= 50 {
        -0.5
    } else {
        0.0
    }
}

/// Spec Skandalrisiko: stufenweise Zusatz (exklusiv).
fn scandal_age_extra_pct(min_age_years: i32) -> f64 {
    if min_age_years <= 13 {
        6.0
    } else if min_age_years <= 15 {
        3.0
    } else if min_age_years <= 17 {
        1.0
    } else {
        0.0
    }
}

fn lover_marriage_daily_delta(
    mg: u8,
    l: &LoverData,
    m_sat: i32,
    mistress_n: usize,
    total_lovers: usize,
) -> i32 {
    let role = l.lover_role.as_str();
    match mg {
        0 => match role {
            "secret_affair" => -1,
            "lover" => -2,
            "mistress_or_favorite" => -3,
            _ => -2,
        },
        1 => match role {
            "secret_affair" => -1,
            "lover" => -1,
            "mistress_or_favorite" => -2,
            _ => -1,
        },
        2 => match role {
            "secret_affair" => -1,
            "lover" => -1,
            "mistress_or_favorite" => -1,
            _ => -1,
        },
        3 => {
            if role == "mistress_or_favorite" {
                if l.visibility <= 35
                    && l.maintenance_level >= 65
                    && m_sat >= 45
                    && mistress_n <= 1
                    && total_lovers <= 1
                {
                    return 1;
                }
                if l.visibility > 50 || l.maintenance_level < 40 {
                    return -1;
                }
                return 0;
            }
            if role == "secret_affair" {
                return 0;
            }
            if role == "lover" {
                return 0;
            }
            0
        }
        _ => -1,
    }
}

fn mistress_count_for_pair(lovers: &[LoverData], c1: i32, c2: i32) -> usize {
    lovers
        .iter()
        .filter(|l| {
            (l.c1 == c1 || l.c2 == c1 || l.c1 == c2 || l.c2 == c2)
                && l.lover_role == "mistress_or_favorite"
        })
        .count()
}

fn fetch_reputation(conn: &mut crate::db::DbConnection, cid: i32) -> Result<f64, DbError> {
    conn.prepare("fr", "SELECT COALESCE(reputation, 50)::float8 AS r FROM falukant_data.character WHERE id = $1")?;
    let rows = conn.execute("fr", &[&cid])?;
    Ok(rows
        .first()
        .and_then(|r| r.get("r"))
        .and_then(|v| v.parse::<f64>().ok())
        .unwrap_or(50.0))
}

fn marriage_id_for_character(
    conn: &mut crate::db::DbConnection,
    cid: i32,
) -> Result<Option<i32>, DbError> {
    conn.prepare(
        "mid",
        r#"SELECT r.id FROM falukant_data.relationship r
        JOIN falukant_type.relationship rt ON rt.id = r.relationship_type_id
        AND rt.tr IN ('married', 'engaged', 'wooing')
        WHERE r.character1_id = $1 OR r.character2_id = $1 LIMIT 1"#,
    )?;
    let rows = conn.execute("mid", &[&cid])?;
    Ok(rows
        .first()
        .and_then(|r| r.get("id"))
        .and_then(|v| v.parse::<i32>().ok()))
}

impl Worker for FalukantFamilyWorker {
    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 = FalukantFamilyWorker::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();
    }
}