Batched updates

Cargo.toml

@@ -23,6 +23,7 @@ hex = "0.4.2"
 serde-big-array = "0.3.0"
 rayon = "1.1"
 rusqlite = {version = "0.23.1", features = ["bundled"]}
+timer = "0.2.0"
 
 [dependencies.reqwest]
 default-features = false # do not include the default features, and optionally

src/composition_root.rs

@@ -1,7 +1,7 @@
 use crate::networking::{TcnApi, TcnApiImpl};
 use crate::reports_updater::{
-    ObservedTcnProcessor, ObservedTcnProcessorImpl, ReportsUpdater, TcnDao, TcnDaoImpl, TcnMatcher,
-    TcnMatcherRayon,
+    ExposureGrouper, ObservedTcnProcessor, ObservedTcnProcessorImpl, ReportsUpdater,
+    TcnBatchesManager, TcnDao, TcnDaoImpl, TcnMatcher, TcnMatcherRayon,
 };
 use crate::{
     errors::ServicesError,
@@ -152,6 +152,7 @@ fn create_comp_root(
     };
 
     let tcn_dao = Arc::new(TcnDaoImpl::new(database.clone()));
+    let exposure_grouper = ExposureGrouper { threshold: 3600 };
 
     CompositionRoot {
         api,
@@ -161,16 +162,18 @@ fn create_comp_root(
             tcn_matcher: TcnMatcherRayon {},
             api,
             memo_mapper,
+            exposure_grouper: exposure_grouper.clone(),
         },
         symptom_inputs_processor: SymptomInputsProcessorImpl {
             inputs_manager: SymptomInputsManagerImpl {
                 inputs: Arc::new(RwLock::new(SymptomInputs::default())),
                 inputs_submitter: symptom_inputs_submitter,
             },
         },
-        observed_tcn_processor: ObservedTcnProcessorImpl {
-            tcn_dao: tcn_dao.clone(),
-        },
+        observed_tcn_processor: ObservedTcnProcessorImpl::new(TcnBatchesManager::new(
+            tcn_dao.clone(),
+            exposure_grouper.clone(),
+        )),
         tcn_keys: tcn_keys.clone(),
     }
 }

src/errors.rs

@@ -1,4 +1,5 @@
 use crate::networking::NetworkingError;
+use rusqlite::Error::QueryReturnedNoRows;
 use std::{error, fmt, io::Error as StdError, io::ErrorKind};
 use tcn::Error as TcnError;
 pub type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
@@ -8,6 +9,7 @@ pub enum ServicesError {
     Networking(NetworkingError),
     Error(Error),
     FFIParameters(String),
+    NotFound,
     General(String),
 }
 
@@ -79,10 +81,13 @@ impl From<std::str::Utf8Error> for ServicesError {
 
 impl From<rusqlite::Error> for ServicesError {
     fn from(error: rusqlite::Error) -> Self {
-        ServicesError::Error(Box::new(StdError::new(
-            ErrorKind::Other,
-            format!("{}", error),
-        )))
+        match error {
+            QueryReturnedNoRows => ServicesError::NotFound,
+            _ => ServicesError::Error(Box::new(StdError::new(
+                ErrorKind::Other,
+                format!("{}", error),
+            ))),
+        }
     }
 }
 

src/preferences.rs

@@ -1,6 +1,8 @@
-use crate::{byte_vec_to_32_byte_array, expect_log, reports_interval::ReportsInterval};
+use crate::{
+    byte_vec_to_32_byte_array, errors::ServicesError, expect_log, reports_interval::ReportsInterval,
+};
 use log::*;
-use rusqlite::{params, Connection, Row, ToSql};
+use rusqlite::{params, Connection, Row, ToSql, Transaction};
 use serde::{Deserialize, Serialize};
 use std::fmt;
 use std::{
@@ -152,6 +154,22 @@ impl Database {
         conn.query_row(sql, params, f)
     }
 
+    pub fn transaction<F>(&self, f: F) -> Result<(), ServicesError>
+    where
+        F: FnOnce(&Transaction) -> Result<(), ServicesError>,
+    {
+        let conn_res = self.conn.lock();
+        let mut conn = expect_log!(conn_res, "Couldn't lock connection");
+
+        let t = conn.transaction()?;
+        if f(&t).is_ok() {
+            t.commit()
+        } else {
+            t.rollback()
+        }
+        .map_err(ServicesError::from)
+    }
+
     pub fn new(conn: Connection) -> Database {
         Database {
             conn: Mutex::new(conn),