Enhance sales and salary processing in DirectorWorker: Introduced a flag for conditional WebSocket message publishing in sell_single_inventory_item, allowing for more flexible event notifications. Refactored salary processing to aggregate total income by employer and ensure unique status updates, improving financial accuracy and reducing redundant notifications.

src/worker/director.rs

@@ -1,5 +1,5 @@
 use crate::db::{DbConnection, DbError, Row};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use crate::message_broker::MessageBroker;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
@@ -577,9 +577,12 @@ impl DirectorWorker {
         // statt einer Zeile pro Inventarzeile).
         let mut total_payout: f64 = 0.0;
         let mut total_tax: f64 = 0.0;
+        let mut sold_any: bool = false;
         for item in items.drain(..) {
             if item.quantity > 0 {
-                let (payout, tax) = self.sell_single_inventory_item(&mut conn, &item)?;
+                sold_any = true;
+                let (payout, tax) =
+                    self.sell_single_inventory_item(&mut conn, &item, false)?;
                 total_payout += payout;
                 total_tax += tax;
             } else {
@@ -601,6 +604,13 @@ impl DirectorWorker {
         if total_payout != 0.0 {
             let _ = self.base.change_falukant_user_money(falukant_user_id, total_payout, "sell products");
         }
+        if sold_any {
+            let message = format!(
+                r#"{{"event":"selled_items","branch_id":{}}}"#,
+                director.branch_id
+            );
+            self.base.broker.publish(message);
+        }
 
         Ok(())
     }
@@ -706,11 +716,13 @@ impl DirectorWorker {
     }
 
     /// Verkauft eine Inventarzeile; Geld fließt erst nach Aggregation in [`start_sellings`].
+    /// `publish_selled_ws`: bei gebündeltem Lauf `false`, dann ein `selled_items` am Ende von [`start_sellings`].
     /// Rückgabe: `(Auszahlung an Spieler, Steuer ans Finanzamt)` in gleicher Währung wie die DB.
     fn sell_single_inventory_item(
         &mut self,
         conn: &mut DbConnection,
         item: &InventoryItem,
+        publish_selled_ws: bool,
     ) -> Result<(f64, f64), DbError> {
         if item.quantity <= 0 {
             conn.execute("remove_inventory", &[&item.id])?;
@@ -765,11 +777,13 @@ impl DirectorWorker {
 
         conn.execute("remove_inventory", &[&item.id])?;
 
-        let message = format!(
+        if publish_selled_ws {
-            r#"{{"event":"selled_items","branch_id":{}}}"#,
+            let message = format!(
-            item.branch_id
+                r#"{{"event":"selled_items","branch_id":{}}}"#,
-        );
+                item.branch_id
-        self.base.broker.publish(message);
+            );
+            self.base.broker.publish(message);
+        }
 
         Ok((payout_amount, tax_amount))
     }
@@ -1176,19 +1190,37 @@ impl DirectorWorker {
         let salaries: Vec<SalaryItem> =
             rows.into_iter().filter_map(Self::map_row_to_salary_item).collect();
 
-        for item in salaries {
+        // Ein money_history-Eintrag pro Arbeitgeber, nicht pro Director-Zeile (mehrere Branches).
+        let mut salary_total_by_employer: HashMap<i32, f64> = HashMap::new();
+        for item in &salaries {
+            *salary_total_by_employer
+                .entry(item.employer_user_id)
+                .or_insert(0.0) += item.income as f64;
+        }
+
+        for (employer_user_id, total_income) in salary_total_by_employer {
+            if total_income <= 0.0 {
+                continue;
+            }
+            let rounded = (total_income * 100.0).round() / 100.0;
             if let Err(_err) = self.base.change_falukant_user_money(
-                item.employer_user_id,
+                employer_user_id,
-                -(item.income as f64),
+                -rounded,
                 "director payed out",
             ) {
             }
+        }
 
+        let mut status_sent: HashSet<i32> = HashSet::new();
+        for item in salaries {
             conn.execute("set_salary_payed", &[&item.id])?;
-
+            if status_sent.insert(item.employer_user_id) {
-            let message =
+                let message = format!(
-                format!(r#"{{"event":"falukantUpdateStatus","user_id":{}}}"#, item.employer_user_id);
+                    r#"{{"event":"falukantUpdateStatus","user_id":{}}}"#,
-            self.base.broker.publish(message);
+                    item.employer_user_id
+                );
+                self.base.broker.publish(message);
+            }
         }
 
         Ok(())
@@ -1215,18 +1247,22 @@ impl DirectorWorker {
         conn.prepare("update_satisfaction", QUERY_UPDATE_SATISFACTION)?;
         let rows = conn.execute("update_satisfaction", &[])?;
 
+        let mut notify_users: HashSet<i32> = HashSet::new();
         for row in rows {
             if let Some(employer_id) = row
                 .get("employer_user_id")
                 .and_then(|v| v.parse::<i32>().ok())
             {
-                let message = format!(
+                notify_users.insert(employer_id);
-                    r#"{{"event":"directorchanged","user_id":{}}}"#,
-                    employer_id
-                );
-                self.base.broker.publish(message);
             }
         }
+        for employer_id in notify_users {
+            let message = format!(
+                r#"{{"event":"directorchanged","user_id":{}}}"#,
+                employer_id
+            );
+            self.base.broker.publish(message);
+        }
 
         Ok(())
     }