From f344c98368c643b19b8dbcaceea37c529b88d62c Mon Sep 17 00:00:00 2001
From: Dominik Werder <dominik.werder@gmail.com>
Date: Fri, 3 Jun 2022 16:57:16 +0200
Subject: [PATCH] Distribute values

---
 netfetch/src/ca.rs       | 2 ++
 netfetch/src/ca/store.rs | 4 +++-
 netfetch/src/series.rs   | 4 ++--
 netfetch/src/store.rs    | 1 +
 4 files changed, 8 insertions(+), 3 deletions(-)

diff --git a/netfetch/src/ca.rs b/netfetch/src/ca.rs
index 440b13f..075c5d0 100644
--- a/netfetch/src/ca.rs
+++ b/netfetch/src/ca.rs
@@ -154,6 +154,7 @@ pub async fn ca_search(opts: ListenFromFileOpts) -> Result<(), Error> {
     let d = Database {
         name: "daqbuffer".into(),
         host: "sf-nube-11".into(),
+        port: 5432,
         user: "daqbuffer".into(),
         pass: opts.pg_pass.clone(),
     };
@@ -317,6 +318,7 @@ pub async fn ca_connect(opts: ListenFromFileOpts) -> Result<(), Error> {
     let d = Database {
         name: "daqbuffer".into(),
         host: "sf-nube-11".into(),
+        port: 5432,
         user: "daqbuffer".into(),
         pass: opts.pg_pass.clone(),
     };
diff --git a/netfetch/src/ca/store.rs b/netfetch/src/ca/store.rs
index 4ddc44c..bb9a3f0 100644
--- a/netfetch/src/ca/store.rs
+++ b/netfetch/src/ca/store.rs
@@ -64,7 +64,9 @@ impl DataStore {
             .map_err(|e| Error::with_msg_no_trace(format!("{e:?}")))?;
         let qu_insert_ts_msp = Arc::new(q);
         let q = scy
-            .prepare("insert into series_by_ts_msp (ts_msp, shape_kind, scalar_type, series) values (?, ?, ?, ?)")
+            .prepare(
+                "insert into series_by_ts_msp (part, ts_msp, shape_kind, scalar_type, series) values (?, ?, ?, ?, ?)",
+            )
             .await
             .map_err(|e| Error::with_msg_no_trace(format!("{e:?}")))?;
         let qu_insert_series_by_ts_msp = Arc::new(q);
diff --git a/netfetch/src/series.rs b/netfetch/src/series.rs
index 73f91a4..adb466c 100644
--- a/netfetch/src/series.rs
+++ b/netfetch/src/series.rs
@@ -53,8 +53,8 @@ pub async fn get_series_id(pg_client: &PgClient, cd: &ChannelDescDecoded) -> Res
             series &= 0x7fffffffffffffff;
         }
         for _ in 0..2000 {
-            if series > i64::MAX as u64 {
-                series = 0;
+            if series < 1 || series > i64::MAX as u64 {
+                series = 1;
             }
             let res = pg_client
                 .execute(
diff --git a/netfetch/src/store.rs b/netfetch/src/store.rs
index dc1357e..ff7073a 100644
--- a/netfetch/src/store.rs
+++ b/netfetch/src/store.rs
@@ -196,6 +196,7 @@ pub async fn insert_item(item: InsertItem, data_store: &DataStore, stats: &CaCon
     }
     if let Some(ts_msp_grid) = item.ts_msp_grid {
         let params = (
+            (item.series as i32) & 0xff,
             ts_msp_grid as i32,
             if item.shape.to_scylla_vec().is_empty() { 0 } else { 1 } as i32,
             item.scalar_type.to_scylla_i32(),