Enhance SQL query in Worker to include character reputation: Add COALESCE for reputation in the vote count query and adjust the GROUP BY and ORDER BY clauses to incorporate reputation, improving data accuracy in election results.

Reviewed-on: #1

renovate.json

@@ -0,0 +1,6 @@
+{
+  "$schema": "https://docs.renovatebot.com/renovate-schema.json",
+  "extends": [
+    "config:recommended"
+  ]
+}

src/worker/director.rs

@@ -808,18 +808,29 @@ impl DirectorWorker {
 
     // compute piece price and full sell price
     let piece_price = Self::compute_piece_sell_price(item);
-    let sell_price = piece_price * item.quantity as f64;
-
+    
         let one_piece_cost = Self::resolve_one_piece_cost(conn, item.product_id, item.sell_cost)?;
         let cumulative_tax_percent = Self::get_cumulative_tax_percent(conn, item.branch_id, item.user_id)?;
 
-        let revenue_cents = (sell_price * 100.0).round() as i64;
-        let cost_cents = (one_piece_cost * item.quantity as f64 * 100.0).round() as i64;
-        let profit_cents = (revenue_cents - cost_cents).max(0);
-        let tax_cents = ((profit_cents as f64) * cumulative_tax_percent / 100.0).round() as i64;
+        // Preis-Inflation: Der Preis wird basierend auf der Steuer inflatiert,
+        // damit der Netto-Betrag für den Verkäufer gleich bleibt (wie im Frontend)
+        let inflation_factor = if cumulative_tax_percent >= 100.0 {
+            1.0
+        } else {
+            1.0 / (1.0 - cumulative_tax_percent / 100.0)
+        };
+        let adjusted_price_per_unit = (piece_price * inflation_factor * 100.0).round() / 100.0;
+        let revenue = adjusted_price_per_unit * item.quantity as f64;
+        
+        // Steuer wird auf Revenue berechnet (nicht auf Profit), wie im Frontend
+        let revenue_cents = (revenue * 100.0).round() as i64;
+        let tax_cents = ((revenue * cumulative_tax_percent / 100.0) * 100.0).round() as i64;
         let payout_cents = revenue_cents - tax_cents;
+        
+        let cost = one_piece_cost * item.quantity as f64;
+        let profit_cents = revenue_cents - (cost * 100.0).round() as i64;
 
-        eprintln!("[DirectorWorker] sell: revenue={:.2}, cost={:.2}, profit_cents={}, tax%={:.2}, tax_cents={}, payout_cents={}", sell_price, one_piece_cost * item.quantity as f64, profit_cents, cumulative_tax_percent, tax_cents, payout_cents);
+        eprintln!("[DirectorWorker] sell: revenue={:.2}, cost={:.2}, profit_cents={}, tax%={:.2}, tax_cents={}, payout_cents={}", revenue, cost, profit_cents, cumulative_tax_percent, tax_cents, payout_cents);
 
         if tax_cents > 0 {
             let tax_amount = (tax_cents as f64) / 100.0;
@@ -837,7 +848,7 @@ impl DirectorWorker {
         eprintln!(
             "[DirectorWorker] sell: user_id={}, revenue={:.2}, tax={:.2}, payout={:.2}, product_id={}",
             item.user_id,
-            sell_price,
+            revenue,
             (tax_cents as f64) / 100.0,
             payout_amount,
             item.product_id

src/worker/events.rs

@@ -432,9 +432,15 @@ impl EventsWorker {
         }
     }
 
-    // Globaler Skalierungsfaktor für Ereignisfrequenz (1.0 = unverändert).
-    // Setze auf 0.05, um Ereignisse auf 1/20 der ursprünglichen Häufigkeit zu reduzieren.
-    const EVENT_RATE_SCALE: f64 = 0.05;
+    // Globaler Skalierungsfaktor für Ereignisfrequenz.
+    // Default: 1.0 (unverändert). Optional per ENV `EVENT_RATE_SCALE` konfigurierbar.
+    fn event_rate_scale() -> f64 {
+        std::env::var("EVENT_RATE_SCALE")
+            .ok()
+            .and_then(|v| v.parse::<f64>().ok())
+            .filter(|v| v.is_finite() && *v >= 0.0)
+            .unwrap_or(1.0)
+    }
 
     fn check_and_trigger_events_inner(
         pool: &ConnectionPool,
@@ -443,11 +449,12 @@ impl EventsWorker {
         rng: &mut impl Rng,
         events: &[RandomEvent],
     ) -> Result<(), DbError> {
+        let rate_scale = Self::event_rate_scale();
         // Prüfe jedes mögliche Ereignis
         for event in events {
             // Zufällige Prüfung basierend auf Wahrscheinlichkeit
         let roll = rng.gen_range(0.0..=1.0);
-    let effective_prob = event.probability_per_minute * Self::EVENT_RATE_SCALE;
+    let effective_prob = event.probability_per_minute * rate_scale;
         if roll < effective_prob {
                 eprintln!(
             "[EventsWorker] Ereignis '{}' wurde ausgelöst (Wahrscheinlichkeit: {:.4}% -> skaliert {:.4}%)",

src/worker/sql.rs

@@ -30,11 +30,27 @@ SELECT DISTINCT b.falukant_user_id AS user_id FROM falukant_data.branch b WHERE
 
 pub const QUERY_UPDATE_WEATHER: &str = r#"
 WITH all_regions AS (
-    SELECT DISTINCT r.id AS region_id FROM falukant_data.region r JOIN falukant_type.region tr ON r.region_type_id = tr.id WHERE tr.label_tr = 'city'
+    SELECT DISTINCT r.id AS region_id
+    FROM falukant_data.region r
+    JOIN falukant_type.region tr ON r.region_type_id = tr.id
+    WHERE tr.label_tr = 'city'
+),
+random_weather AS (
+    SELECT 
+        ar.region_id,
+        (
+            SELECT wt.id
+            FROM falukant_type.weather wt
+            ORDER BY RANDOM()
+            LIMIT 1
+        ) AS weather_type_id
+    FROM all_regions ar
 )
 INSERT INTO falukant_data.weather (region_id, weather_type_id)
-SELECT ar.region_id, (SELECT wt.id FROM falukant_type.weather wt ORDER BY random() + ar.region_id * 0 LIMIT 1) FROM all_regions ar
-ON CONFLICT (region_id) DO UPDATE SET weather_type_id = EXCLUDED.weather_type_id;
+SELECT rw.region_id, rw.weather_type_id
+FROM random_weather rw
+ON CONFLICT (region_id)
+DO UPDATE SET weather_type_id = EXCLUDED.weather_type_id;
 "#;
 
 pub const QUERY_INSERT_NOTIFICATION: &str = r#"
@@ -44,7 +60,7 @@ VALUES ($1, $2, FALSE, NOW(), NOW());
 
 // Product pricing
 pub const QUERY_GET_PRODUCT_COST: &str = r#"
-SELECT original_sell_cost, sell_cost FROM falukant_type.product WHERE id = $1;
+SELECT sell_cost, sell_cost AS original_sell_cost FROM falukant_type.product WHERE id = $1;
 "#;
 
 pub const QUERY_GET_DIRECTORS: &str = r#"
@@ -190,8 +206,12 @@ DELETE FROM falukant_data.inventory WHERE id = $1;
 "#;
 
 pub const QUERY_ADD_SELL_LOG: &str = r#"
-INSERT INTO falukant_log.sell (region_id, product_id, quantity, seller_id) VALUES ($1, $2, $3, $4)
-ON CONFLICT (region_id, product_id, seller_id) DO UPDATE SET quantity = falukant_log.sell.quantity + EXCLUDED.quantity;
+INSERT INTO falukant_log.sell (region_id, product_id, quantity, seller_id, sell_timestamp) 
+VALUES ($1, $2, $3, $4, NOW())
+ON CONFLICT (seller_id, product_id, region_id) 
+DO UPDATE SET 
+    quantity = falukant_log.sell.quantity + EXCLUDED.quantity, 
+    sell_timestamp = COALESCE(EXCLUDED.sell_timestamp, NOW());
 "#;
 pub const QUERY_GET_ARRIVED_TRANSPORTS: &str = r#"
 SELECT
@@ -203,15 +223,37 @@ SELECT
     t.target_region_id,
     b_target.id AS target_branch_id,
     b_source.id AS source_branch_id,
-    rd.distance AS distance,
+    COALESCE(rd.distance, 0.0) AS distance,
     v.falukant_user_id AS user_id
 FROM falukant_data.transport AS t
 JOIN falukant_data.vehicle AS v ON v.id = t.vehicle_id
 JOIN falukant_type.vehicle AS vt ON vt.id = v.vehicle_type_id
-JOIN falukant_data.region_distance AS rd ON ((rd.source_region_id = t.source_region_id AND rd.target_region_id = t.target_region_id) OR (rd.source_region_id = t.target_region_id AND rd.target_region_id = t.source_region_id)) AND (rd.transport_mode = vt.transport_mode OR rd.transport_mode IS NULL)
+LEFT JOIN LATERAL (
+    SELECT rd.distance
+      FROM falukant_data.region_distance AS rd
+     WHERE (
+            (rd.source_region_id = t.source_region_id AND rd.target_region_id = t.target_region_id)
+         OR (rd.source_region_id = t.target_region_id AND rd.target_region_id = t.source_region_id)
+     )
+       AND (
+            rd.transport_mode = vt.transport_mode
+         OR rd.transport_mode IS NULL
+         OR vt.transport_mode IS NULL
+       )
+     ORDER BY
+         (rd.transport_mode = vt.transport_mode) DESC,
+         (rd.transport_mode IS NULL) DESC
+     LIMIT 1
+) AS rd ON TRUE
 LEFT JOIN falukant_data.branch AS b_target ON b_target.region_id = t.target_region_id AND b_target.falukant_user_id = v.falukant_user_id
 LEFT JOIN falukant_data.branch AS b_source ON b_source.region_id = t.source_region_id AND b_source.falukant_user_id = v.falukant_user_id
-WHERE vt.speed > 0 AND t.created_at + (rd.distance / vt.speed::double precision) * INTERVAL '1 minute' <= NOW();
+WHERE (
+    -- Transport ist angekommen basierend auf Distanz und Geschwindigkeit (wenn speed > 0)
+    (vt.speed > 0 AND rd.distance IS NOT NULL AND t.created_at + (rd.distance / vt.speed::double precision) * INTERVAL '1 minute' <= NOW())
+    OR
+    -- Fallback: Transport ist länger als 24 Stunden unterwegs (falls keine Distanz gefunden wurde oder speed = 0/NULL)
+    (t.created_at + INTERVAL '24 hours' <= NOW())
+  );
 "#;
 
 pub const QUERY_GET_AVAILABLE_STOCKS: &str = r#"
@@ -305,7 +347,13 @@ SELECT v.id AS vehicle_id, vt.capacity AS capacity
 FROM falukant_data.vehicle v
 JOIN falukant_type.vehicle vt ON vt.id = v.vehicle_type_id
 JOIN falukant_data.region_distance rd ON ((rd.source_region_id = v.region_id AND rd.target_region_id = $3) OR (rd.source_region_id = $3 AND rd.target_region_id = v.region_id)) AND (rd.transport_mode = vt.transport_mode OR rd.transport_mode IS NULL)
-WHERE v.falukant_user_id = $1 AND v.region_id = $2;
+WHERE v.falukant_user_id = $1
+  AND v.region_id = $2
+  AND v.id NOT IN (
+      SELECT DISTINCT t.vehicle_id
+        FROM falukant_data.transport t
+       WHERE t.vehicle_id IS NOT NULL
+  );
 "#;
 
 pub const QUERY_INSERT_TRANSPORT: &str = r#"
@@ -782,7 +830,8 @@ pub const QUERY_PROCESS_EXPIRED_AND_FILL: &str = r#"
                         dt.office_type_id,
                         dt.region_id,
                         c.character_id,
-                        COUNT(v.id) AS vote_count
+                        COUNT(v.id) AS vote_count,
+                        COALESCE(ch.reputation, 0) AS reputation
                 FROM distinct_types AS dt
                 JOIN falukant_data.election AS e
                     ON e.office_type_id = dt.office_type_id
@@ -791,8 +840,10 @@ pub const QUERY_PROCESS_EXPIRED_AND_FILL: &str = r#"
                 JOIN falukant_data.candidate AS c
                     ON c.election_id = e.id
                  AND c.id           = v.candidate_id
+                JOIN falukant_data."character" AS ch
+                    ON ch.id = c.character_id
                 WHERE e.date >= (NOW() - INTERVAL '30 days')
-                GROUP BY dt.office_type_id, dt.region_id, c.character_id
+                GROUP BY dt.office_type_id, dt.region_id, c.character_id, ch.reputation
         ),
         ranked_winners AS (
                 SELECT
@@ -801,7 +852,7 @@ pub const QUERY_PROCESS_EXPIRED_AND_FILL: &str = r#"
                         vpc.character_id,
                         ROW_NUMBER() OVER (
                                 PARTITION BY vpc.office_type_id, vpc.region_id
-                                ORDER BY vpc.vote_count DESC
+                                ORDER BY vpc.vote_count DESC, vpc.reputation DESC, vpc.character_id ASC
                         ) AS rn
                 FROM votes_per_candidate AS vpc
         ),
@@ -1440,10 +1491,11 @@ pub const QUERY_ADD_OVERPRODUCTION_NOTIFICATION: &str = r#"
     INSERT INTO falukant_log.notification (
         user_id,
         tr,
+        character_id,
         shown,
         created_at,
         updated_at
-    ) VALUES ($1, $2, FALSE, NOW(), NOW());
+    ) VALUES ($1, $2, NULL, FALSE, NOW(), NOW());
 "#;
 
 // Aliases for personal variants (keeps original prepared statement names used in events.worker)

src/worker/transport.rs

@@ -2,8 +2,8 @@ use crate::db::{ConnectionPool, DbError};
 use crate::message_broker::MessageBroker;
 use std::cmp::min;
 use std::sync::atomic::Ordering;
-use std::sync::Arc;
-use std::time::Duration;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
 
 use super::base::{BaseWorker, Worker, WorkerState};
 use crate::worker::sql::{
@@ -51,8 +51,8 @@ impl TransportWorker {
                 eprintln!("[TransportWorker] Fehler in process_arrived_transports: {err}");
             }
 
-            // Einmal pro Sekunde prüfen
-            for _ in 0..1 {
+            // Minütlich prüfen (nicht sekündlich pollen)
+            for _ in 0..60 {
                 if !state.running_worker.load(Ordering::Relaxed) {
                     break;
                 }
@@ -67,11 +67,18 @@ impl TransportWorker {
     ) -> Result<(), DbError> {
         let transports = Self::load_arrived_transports(pool)?;
 
+        if !transports.is_empty() {
+            eprintln!(
+                "[TransportWorker] {} angekommene Transport(e) gefunden",
+                transports.len()
+            );
+        }
+
         for t in transports {
             if let Err(err) = Self::handle_arrived_transport(pool, broker, &t) {
                 eprintln!(
-                    "[TransportWorker] Fehler beim Verarbeiten von Transport {}: {err}",
-                    t.id
+                    "[TransportWorker] Fehler beim Verarbeiten von Transport {} (vehicle_id={}, product_id={:?}, size={}): {}",
+                    t.id, t.vehicle_id, t.product_id, t.size, err
                 );
             }
         }
@@ -87,6 +94,19 @@ impl TransportWorker {
         conn.prepare("get_arrived_transports", QUERY_GET_ARRIVED_TRANSPORTS)?;
         let rows = conn.execute("get_arrived_transports", &[])?;
 
+        if rows.is_empty() {
+            // Nur alle 60 Sekunden loggen, um Log-Flut zu vermeiden
+            static LAST_LOG: Mutex<Option<Instant>> = Mutex::new(None);
+            let mut last_log = LAST_LOG.lock().unwrap();
+            let should_log = last_log
+                .map(|t| t.elapsed().as_secs() >= 60)
+                .unwrap_or(true);
+            if should_log {
+                eprintln!("[TransportWorker] Keine angekommenen Transporte gefunden");
+                *last_log = Some(Instant::now());
+            }
+        }
+
         let mut result = Vec::with_capacity(rows.len());
         for row in rows {
             let id = parse_i32(&row, "id", -1);
@@ -223,6 +243,13 @@ impl TransportWorker {
                     );
                     broker.publish(target_inventory_message);
                 }
+            } else {
+                // Nichts konnte eingelagert werden - Transport bleibt unverändert
+                // Logge dies, damit wir sehen, dass der Transport wartet
+                eprintln!(
+                    "[TransportWorker] Transport {} wartet: Kein Lagerplatz verfügbar (branch_id={}, product_id={}, size={})",
+                    t.id, t.target_branch_id, product_id, t.size
+                );
             }
 
             // Keine Notification für wartende Transporte, um Notification-System zu entlasten.

src/worker/weather.rs

@@ -5,39 +5,12 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use super::base::{BaseWorker, Worker, WorkerState};
+use crate::worker::sql::QUERY_UPDATE_WEATHER;
 
 pub struct WeatherWorker {
     base: BaseWorker,
 }
 
-// Query zum Aktualisieren des Wetters für alle Regionen
-// Wählt für jede Region ein zufälliges Wetter aus allen verfügbaren Wettertypen aus
-// Wichtig: Jede Region bekommt ein individuelles, zufälliges Wetter
-const QUERY_UPDATE_WEATHER: &str = r#"
-    WITH all_regions AS (
-        SELECT DISTINCT r.id AS region_id
-        FROM falukant_data.region r
-        JOIN falukant_type.region tr ON r.region_type_id = tr.id
-        WHERE tr.label_tr = 'city'
-    ),
-    random_weather AS (
-        SELECT 
-            ar.region_id,
-            (
-                SELECT wt.id
-                FROM falukant_type.weather wt
-                ORDER BY RANDOM()
-                LIMIT 1
-            ) AS weather_type_id
-        FROM all_regions ar
-    )
-    INSERT INTO falukant_data.weather (region_id, weather_type_id)
-    SELECT rw.region_id, rw.weather_type_id
-    FROM random_weather rw
-    ON CONFLICT (region_id)
-    DO UPDATE SET weather_type_id = EXCLUDED.weather_type_id;
-"#;
-
 impl WeatherWorker {
     pub fn new(pool: ConnectionPool, broker: MessageBroker) -> Self {
         Self {