Enhance pricing logic in DirectorWorker and implement hourly price recalculation: Added a new worth_percent field to the InventoryItem struct and updated SQL queries to incorporate this value in price calculations. Refactored price computation logic to use a base price derived from worth_percent. Introduced a new hourly price recalculation mechanism that adjusts prices based on sales data from the last hour, ensuring dynamic pricing adjustments. Enhanced logging for better monitoring of price updates.

src/worker/value_recalculation.rs

@@ -1,5 +1,6 @@
 use crate::db::{ConnectionPool, DbError, Row};
 use crate::message_broker::MessageBroker;
+use std::collections::HashSet;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
@@ -75,6 +76,53 @@ const QUERY_GET_SELL_REGIONS: &str = r#"
      GROUP BY region_id;
 "#;
 
+// Stündliche Preisneuberechnung basierend auf Verkäufen der letzten Stunde
+// Für jedes Produkt in jeder Stadt: Wenn überdurchschnittlich viel verkauft wurde,
+// sinkt der Preis um 10% (begrenzt auf 0-100)
+const QUERY_HOURLY_PRICE_RECALCULATION: &str = r#"
+    WITH city_sales AS (
+        SELECT
+            s.region_id,
+            s.product_id,
+            SUM(s.quantity) AS total_sold
+        FROM falukant_log.sell s
+        WHERE s.sell_timestamp >= NOW() - INTERVAL '1 hour'
+        GROUP BY s.region_id, s.product_id
+    ),
+    avg_sales AS (
+        SELECT
+            product_id,
+            AVG(total_sold) AS avg_sold
+        FROM city_sales
+        GROUP BY product_id
+    ),
+    price_updates AS (
+        SELECT
+            cs.region_id,
+            cs.product_id,
+            cs.total_sold,
+            COALESCE(av.avg_sold, 0) AS avg_sold,
+            tpw.worth_percent AS current_price
+        FROM city_sales cs
+        JOIN avg_sales av ON av.product_id = cs.product_id
+        JOIN falukant_data.town_product_worth tpw
+            ON tpw.region_id = cs.region_id
+            AND tpw.product_id = cs.product_id
+        WHERE cs.total_sold > COALESCE(av.avg_sold, 0)
+    )
+    UPDATE falukant_data.town_product_worth tpw
+    SET worth_percent = GREATEST(
+        0,
+        LEAST(
+            100,
+            pu.current_price * 0.9  -- 10% Preisreduktion
+        )
+    )
+    FROM price_updates pu
+    WHERE tpw.region_id = pu.region_id
+      AND tpw.product_id = pu.product_id;
+"#;
+
 // Ehen / Beziehungen
 const QUERY_SET_MARRIAGES_BY_PARTY: &str = r#"
     WITH updated_relations AS (
@@ -168,6 +216,7 @@ impl ValueRecalculationWorker {
         // statt exakte Uhrzeiten nachzubilden – Verhalten ist funktional ähnlich.
         let mut last_product = None;
         let mut last_sell_price = None;
+        let mut last_hourly_price_recalc = None;
 
         loop {
             if !state.running_worker.load(Ordering::Relaxed) {
@@ -192,6 +241,14 @@ impl ValueRecalculationWorker {
                 last_sell_price = Some(now);
             }
 
+            // Stündliche Preisneuberechnung basierend auf Verkäufen der letzten Stunde
+            if should_run_interval(last_hourly_price_recalc, now, Duration::from_secs(3600)) {
+                if let Err(err) = Self::calculate_hourly_price_recalculation_inner(&pool, &broker) {
+                    eprintln!("[ValueRecalculationWorker] Fehler in calculateHourlyPriceRecalculation: {err}");
+                }
+                last_hourly_price_recalc = Some(now);
+            }
+
             // Ehen & Studium bei jedem Durchlauf
             if let Err(err) = Self::calculate_marriages_inner(&pool, &broker) {
                 eprintln!("[ValueRecalculationWorker] Fehler in calculateMarriages: {err}");
@@ -275,6 +332,51 @@ impl ValueRecalculationWorker {
         Ok(())
     }
 
+    fn calculate_hourly_price_recalculation_inner(
+        pool: &ConnectionPool,
+        broker: &MessageBroker,
+    ) -> Result<(), DbError> {
+        let mut conn = pool
+            .get()
+            .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?;
+
+        conn.prepare("hourly_price_recalculation", QUERY_HOURLY_PRICE_RECALCULATION)?;
+        let updated_rows = conn.execute("hourly_price_recalculation", &[])?;
+
+        // Sammle alle betroffenen Regionen für Event-Benachrichtigungen
+        let mut affected_regions = std::collections::HashSet::new();
+        
+        // Da die Query bereits die Updates durchführt, müssen wir die betroffenen Regionen
+        // separat abfragen. Alternativ können wir auch einfach alle Regionen benachrichtigen,
+        // die in der letzten Stunde Verkäufe hatten.
+        conn.prepare("get_sell_regions_hourly", r#"
+            SELECT DISTINCT region_id
+            FROM falukant_log.sell
+            WHERE sell_timestamp >= NOW() - INTERVAL '1 hour'
+        "#)?;
+        let regions = conn.execute("get_sell_regions_hourly", &[])?;
+
+        for row in regions {
+            if let Some(region_id) = row.get("region_id").and_then(|v| v.parse::<i32>().ok()) {
+                affected_regions.insert(region_id);
+            }
+        }
+
+        // Benachrichtige alle betroffenen Regionen über Preisänderungen
+        for region_id in affected_regions {
+            let message =
+                format!(r#"{{"event":"price_update","region_id":{}}}"#, region_id);
+            broker.publish(message);
+        }
+
+        eprintln!(
+            "[ValueRecalculationWorker] Stündliche Preisneuberechnung abgeschlossen. {} Regionen aktualisiert.",
+            affected_regions.len()
+        );
+
+        Ok(())
+    }
+
     fn calculate_marriages_inner(
         pool: &ConnectionPool,
         broker: &MessageBroker,