fix(loader): dfe-clickhouse 0.1.11, risk scoring and discovery from config

Cargo.toml

@@ -82,7 +82,7 @@ maxminddb = { version = ">=0.32.0, <0.33", features = ["mmap"] }
 clickhouse = { version = ">=0.15.3, <0.16", features = ["lz4", "time", "rustls-tls", "rustls-tls-native-roots", "inserter"] }
 # `ext` reads the server's exception code out of an HTTP rejection; `tls`
 # (implies `tcp`) is the native-protocol client, rustls only.
-dfe-clickhouse = { version = ">=0.1.10, <0.2", default-features = false, features = ["ext", "tls", "dynamic"] }
+dfe-clickhouse = { version = ">=0.1.11, <0.2", default-features = false, features = ["ext", "tls", "dynamic"] }
 time = ">=0.3.55, <0.4"
 
 # The HTTP client handed to clickhouse::Client::with_http_client: connect timeout,

benches/enrichment.rs

@@ -6,9 +6,10 @@
 //! Measures CPU cost per operation for `GeoIP`, Reputation, and Risk scoring.
 
 use criterion::{Criterion, Throughput, criterion_group, criterion_main};
+use dfe_loader::config::RiskScoringConfig;
 use dfe_loader::enrich::geoip::{GeoIpEnricher, GeoIpResult};
 use dfe_loader::enrich::reputation::{ReputationEnricher, ThreatSource, ThreatType};
-use dfe_loader::enrich::risk::{RiskInput, RiskPreset, RiskScorer};
+use dfe_loader::enrich::risk::{RiskInput, RiskScorer};
 use std::hint::black_box;
 use std::net::IpAddr;
 
@@ -160,7 +161,7 @@ fn bench_reputation_prefix_match(c: &mut Criterion) {
 }
 
 fn bench_risk_minimal(c: &mut Criterion) {
-    let scorer = RiskScorer::new();
+    let scorer = RiskScorer::default();
 
     c.bench_function("risk/minimal_input", |b| {
         b.iter(|| {
@@ -171,7 +172,7 @@ fn bench_risk_minimal(c: &mut Criterion) {
 }
 
 fn bench_risk_full_input(c: &mut Criterion) {
-    let scorer = RiskScorer::new();
+    let scorer = RiskScorer::default();
 
     c.bench_function("risk/full_input", |b| {
         b.iter(|| {
@@ -190,7 +191,10 @@ fn bench_risk_full_input(c: &mut Criterion) {
 }
 
 fn bench_risk_with_preset(c: &mut Criterion) {
-    let scorer = RiskScorer::from_preset(RiskPreset::HighSecurity);
+    let scorer = RiskScorer::new(&RiskScoringConfig {
+        preset: "high_security".to_string(),
+        ..RiskScoringConfig::default()
+    });
 
     c.bench_function("risk/high_security_preset", |b| {
         b.iter(|| {
@@ -205,7 +209,7 @@ fn bench_risk_with_preset(c: &mut Criterion) {
 }
 
 fn bench_risk_from_enrichment(c: &mut Criterion) {
-    let scorer = RiskScorer::new();
+    let scorer = RiskScorer::default();
 
     // Simulate results from GeoIP and Reputation
     let geo_result = GeoIpResult {
@@ -235,7 +239,7 @@ fn bench_risk_from_enrichment(c: &mut Criterion) {
 fn bench_full_enrichment_pipeline(c: &mut Criterion) {
     let geoip = GeoIpEnricher::new();
     let reputation = ReputationEnricher::new();
-    let risk = RiskScorer::new();
+    let risk = RiskScorer::default();
 
     // Add some blocklist entries
     reputation.add_ip(

config.example.yaml

@@ -159,6 +159,8 @@ routing:
 
   # Topic suffixes stripped when deriving _source from Kafka topic name
   # Example: topic "auth_land" with suffix "_land" → _source = "auth"
+  # With kafka.topics and kafka.topic_regex unset, auto-discovery subscribes to
+  # the topics ending in one of these.
   topic_suffixes:
     - _land
     - _load
@@ -304,10 +306,44 @@ metadata:
   raw_source_fields: []
   raw_output: "_raw"
 
-# Type coercion settings
-coercion:
-  unknown_type_fallback: String
-  array_to_json: true
+# IP enrichment. Risk scoring weighs the GeoIP and reputation results, so it
+# needs at least one of them on.
+enrichment:
+  ip_fields: [src_ip, client_ip, ip, source_ip]
+  risk_scoring:
+    enabled: false
+    # One of the presets below. A name not listed is refused at startup.
+    preset: global
+    # Shipped: global, us_enterprise, eu_enterprise, apac_enterprise,
+    # high_security. An entry here replaces the shipped one of that name, or
+    # adds a preset, and the rest stay as shipped.
+    # presets:
+    #   global:
+    #     high: [KP, IR, SY]
+    #     medium: [RU, CN, VE, CU]
+    weights: {geo: 15, reputation: 25, privacy: 25, threat: 35}
+    # Score each signal gives its component, 0-100. Reputation and threat
+    # signals add up, capped at 100.
+    scores:
+      unknown_country: 35
+      high_risk_country: 90
+      medium_risk_country: 60
+      other_country: 15
+      vpn: 50
+      proxy: 60
+      tor: 80
+      datacenter: 40
+      relay: 45
+      one_anonymiser: 40
+      two_anonymisers: 70
+      three_anonymisers: 95
+      botnet: 95
+      malicious: 90
+      spam: 70
+      scanner: 60
+      high_abuse_score: 80   # abuse score that adds the high_abuse_score factor
+    # Lowest composite score of each level. Below low is minimal.
+    levels: {low: 20, medium: 40, high: 60, critical: 80}
 
 # Field mapping (rename/copy source fields to destination schema names)
 # Maps incoming field names to standardised schemas (ECS, CIM, etc.)