Refactor Falukant certificate processing logic: Introduced a new SQL query to check for the existence of the completion_count column in the production log, allowing for conditional selection of certificate input rows. Added a legacy query for environments lacking this column, enhancing compatibility and robustness in certificate processing.

src/worker/falukant_certificate.rs

@@ -9,6 +9,7 @@ use crate::message_broker::MessageBroker;
 use super::base::BaseWorker;
 use crate::worker::sql::{
     QUERY_GET_FALUKANT_USER_CERT_AND_EVENT, QUERY_GET_PRODUCTION_CERTIFICATE_INPUT_ROWS,
+    QUERY_GET_PRODUCTION_CERTIFICATE_INPUT_ROWS_LEGACY, QUERY_PRODUCTION_COMPLETION_COUNT_READY,
     QUERY_UPDATE_FALUKANT_USER_CERTIFICATE,
 };
 
@@ -22,7 +23,20 @@ pub fn run_daily(base: &BaseWorker, broker: &MessageBroker) -> Result<(), DbErro
         .get()
         .map_err(|e| DbError::new(format!("DB-Verbindung fehlgeschlagen: {e}")))?;
 
-    conn.prepare("cert_rows", QUERY_GET_PRODUCTION_CERTIFICATE_INPUT_ROWS)?;
+    conn.prepare("cert_rows_completion_ready", QUERY_PRODUCTION_COMPLETION_COUNT_READY)?;
+    let ready_rows = conn.execute("cert_rows_completion_ready", &[])?;
+    let completion_count_ready = ready_rows
+        .first()
+        .and_then(|r| r.get("ready"))
+        .map(|v| v == "t" || v == "true")
+        .unwrap_or(false);
+
+    let cert_rows_query = if completion_count_ready {
+        QUERY_GET_PRODUCTION_CERTIFICATE_INPUT_ROWS
+    } else {
+        QUERY_GET_PRODUCTION_CERTIFICATE_INPUT_ROWS_LEGACY
+    };
+    conn.prepare("cert_rows", cert_rows_query)?;
     conn.prepare("cert_upd", QUERY_UPDATE_FALUKANT_USER_CERTIFICATE)?;
     let rows = conn.execute("cert_rows", &[])?;
 

src/worker/sql.rs

@@ -4015,6 +4015,75 @@ pub const QUERY_GET_PRODUCTION_CERTIFICATE_INPUT_ROWS: &str = r#"
   ORDER BY fu.id, c.id DESC;
 "#;
 
+/// Fallback für Bestands-DBs ohne `falukant_log.production.completion_count`:
+/// nutzt `COUNT(*)` statt `SUM(completion_count)`.
+pub const QUERY_GET_PRODUCTION_CERTIFICATE_INPUT_ROWS_LEGACY: &str = r#"
+    SELECT DISTINCT ON (fu.id)
+           fu.id AS falukant_user_id,
+           COALESCE(fu.user_id, fu.id)::int AS app_user_id,
+           COALESCE(fu.certificate, 1)::int AS certificate,
+           COALESCE(fu.money, 0)::float8 AS money,
+           c.id AS character_id,
+           COALESCE(c.reputation, 50)::float8 AS reputation,
+           COALESCE(t.level, 0)::int AS title_level,
+           COALESCE((
+               SELECT AVG(k.knowledge)::float8
+                 FROM falukant_data.knowledge k
+                WHERE k.character_id = c.id
+           ), 0.0) AS avg_knowledge,
+           COALESCE((
+               SELECT COUNT(*)::bigint
+                 FROM falukant_log.production pl
+                WHERE (pl.producer_id = fu.id OR pl.producer_id = c.id)
+                  AND (
+                      fu.certificate_productions_count_since IS NULL
+                      OR COALESCE(
+                          pl.production_timestamp,
+                          pl.production_date::timestamp
+                      ) >= fu.certificate_productions_count_since
+                  )
+           ), 0) AS completed_production_count,
+           GREATEST(
+               COALESCE(c.highest_church_hierarchy_ever, 0)::int,
+               COALESCE((
+                   SELECT MAX(cot.hierarchy_level)::int
+                     FROM falukant_data.church_office co
+                     JOIN falukant_type.church_office_type cot ON cot.id = co.office_type_id
+                    WHERE co.character_id = c.id
+               ), 0)
+           ) AS max_church_hierarchy,
+           COALESCE((
+               SELECT STRING_AGG(DISTINCT pot.name, '|')
+                 FROM falukant_data.political_office po
+                 JOIN falukant_type.political_office_type pot ON pot.id = po.office_type_id
+                WHERE po.character_id = c.id
+                  AND (po.created_at + (pot.term_length * INTERVAL '1 day')) > NOW()
+           ), '') AS political_office_names,
+           COALESCE((
+               SELECT h.position::int
+                 FROM falukant_data.user_house uh
+                 JOIN falukant_type.house h ON h.id = uh.house_type_id
+                WHERE uh.user_id = fu.id
+                ORDER BY uh.id
+                LIMIT 1
+           ), 0) AS house_position
+      FROM falukant_data.falukant_user fu
+      JOIN falukant_data.character c ON c.user_id = fu.id AND c.health > 0
+ LEFT JOIN falukant_type.title t ON t.id = c.title_of_nobility
+  ORDER BY fu.id, c.id DESC;
+"#;
+
+/// Prüft, ob `falukant_log.production.completion_count` existiert.
+pub const QUERY_PRODUCTION_COMPLETION_COUNT_READY: &str = r#"
+    SELECT EXISTS (
+        SELECT 1
+          FROM information_schema.columns
+         WHERE table_schema = 'falukant_log'
+           AND table_name = 'production'
+           AND column_name = 'completion_count'
+    ) AS ready;
+"#;
+
 /// Setzt bei jeder Stufenänderung `certificate_productions_count_since` (Mindest-Produktionen / PP neu ab Aufstieg).
 pub const QUERY_UPDATE_FALUKANT_USER_CERTIFICATE: &str = r#"
     UPDATE falukant_data.falukant_user