Merge pull request #257 from ActivityWatch/dev/sync-stuff

another sync PR

Author: ErikBjare

.gitignore

@@ -1,6 +1,11 @@
 target
-**/*.rs.bk
-NDK
 .cargo/config
+NDK
+
+**/*.rs.bk
 *.zip
 *.profraw
+
+*.sqlite*
+*.db
+*.db-journal

aw-client-rust/src/lib.rs

@@ -12,18 +12,26 @@ use serde_json::Map;
 
 pub use aw_models::{Bucket, BucketMetadata, Event};
 
-#[derive(Debug)]
 pub struct AwClient {
     client: reqwest::blocking::Client,
     pub baseurl: String,
     pub name: String,
     pub hostname: String,
 }
 
+impl std::fmt::Debug for AwClient {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        write!(f, "AwClient(baseurl={:?})", self.baseurl)
+    }
+}
+
 impl AwClient {
     pub fn new(ip: &str, port: &str, name: &str) -> AwClient {
         let baseurl = format!("http://{}:{}", ip, port);
-        let client = reqwest::blocking::Client::new();
+        let client = reqwest::blocking::Client::builder()
+            .timeout(std::time::Duration::from_secs(120))
+            .build()
+            .unwrap();
         let hostname = gethostname::gethostname().into_string().unwrap();
         AwClient {
             client,
@@ -44,9 +52,18 @@ impl AwClient {
         self.client.get(&url).send()?.json()
     }
 
-    pub fn create_bucket(&self, bucketname: &str, buckettype: &str) -> Result<(), reqwest::Error> {
-        let url = format!("{}/api/0/buckets/{}", self.baseurl, bucketname);
-        let data = Bucket {
+    pub fn create_bucket(&self, bucket: &Bucket) -> Result<(), reqwest::Error> {
+        let url = format!("{}/api/0/buckets/{}", self.baseurl, bucket.id);
+        self.client.post(&url).json(bucket).send()?;
+        Ok(())
+    }
+
+    pub fn create_bucket_simple(
+        &self,
+        bucketname: &str,
+        buckettype: &str,
+    ) -> Result<(), reqwest::Error> {
+        let bucket = Bucket {
             bid: None,
             id: bucketname.to_string(),
             client: self.name.clone(),
@@ -58,8 +75,7 @@ impl AwClient {
             created: None,
             last_updated: None,
         };
-        self.client.post(&url).json(&data).send()?;
-        Ok(())
+        self.create_bucket(&bucket)
     }
 
     pub fn delete_bucket(&self, bucketname: &str) -> Result<(), reqwest::Error> {
@@ -103,6 +119,16 @@ impl AwClient {
         Ok(())
     }
 
+    pub fn insert_events(
+        &self,
+        bucketname: &str,
+        events: Vec<Event>,
+    ) -> Result<(), reqwest::Error> {
+        let url = format!("{}/api/0/buckets/{}/events", self.baseurl, bucketname);
+        self.client.post(&url).json(&events).send()?;
+        Ok(())
+    }
+
     pub fn heartbeat(
         &self,
         bucketname: &str,

aw-client-rust/tests/test.rs

@@ -76,7 +76,9 @@ mod test {
 
         let bucketname = format!("aw-client-rust-test_{}", client.hostname);
         let buckettype = "test-type";
-        client.create_bucket(&bucketname, &buckettype).unwrap();
+        client
+            .create_bucket_simple(&bucketname, &buckettype)
+            .unwrap();
 
         let bucket = client.get_bucket(&bucketname).unwrap();
         assert!(bucket.id == bucketname);

aw-datastore/src/lib.rs

@@ -22,6 +22,7 @@ mod worker;
 pub use self::datastore::DatastoreInstance;
 pub use self::worker::Datastore;
 
+#[derive(Debug, Clone)]
 pub enum DatastoreMethod {
     Memory(),
     File(String),

aw-datastore/src/worker.rs

@@ -53,6 +53,8 @@ pub enum Response {
     Count(i64),
     KeyValue(KeyValue),
     StringVec(Vec<String>),
+    // Used to indicate that no response should occur at all (not even an empty one)
+    NoResponse(),
 }
 
 #[allow(clippy::large_enum_variant)]
@@ -78,6 +80,7 @@ pub enum Command {
     GetKeyValue(String),
     GetKeysStarting(String),
     DeleteKeyValue(String),
+    Close(),
 }
 
 fn _unwrap_response(
@@ -118,7 +121,7 @@ impl DatastoreWorker {
 
     fn work_loop(&mut self, method: DatastoreMethod) {
         // Open SQLite connection
-        let mut conn = match method {
+        let mut conn = match &method {
             DatastoreMethod::Memory() => {
                 Connection::open_in_memory().expect("Failed to create in-memory datastore")
             }
@@ -150,12 +153,20 @@ impl DatastoreWorker {
         // Start handling and respond to requests
         loop {
             let last_commit_time: DateTime<Utc> = Utc::now();
-            let mut transaction = conn
-                .transaction_with_behavior(TransactionBehavior::Immediate)
-                .unwrap();
+            let mut tx: Transaction =
+                match conn.transaction_with_behavior(TransactionBehavior::Immediate) {
+                    Ok(tx) => tx,
+                    Err(err) => {
+                        error!("Unable to start transaction! {:?}", err);
+                        // Wait 1s before retrying
+                        std::thread::sleep(std::time::Duration::from_millis(1000));
+                        continue;
+                    }
+                };
+            tx.set_drop_behavior(DropBehavior::Commit);
+
             self.uncommited_events = 0;
             self.commit = false;
-            transaction.set_drop_behavior(DropBehavior::Commit);
             loop {
                 let (request, response_sender) = match self.responder.poll() {
                     Ok((req, res_sender)) => (req, res_sender),
@@ -166,19 +177,28 @@ impl DatastoreWorker {
                         break;
                     }
                 };
-                let response = self.handle_request(request, &mut ds, &transaction);
-                response_sender.respond(response);
+                let response = self.handle_request(request, &mut ds, &tx);
+                match response {
+                    // The NoResponse is used by commands like close(), which should
+                    // not be responded to, as the requester might have disappeared.
+                    Ok(Response::NoResponse()) => (),
+                    _ => response_sender.respond(response),
+                }
                 let now: DateTime<Utc> = Utc::now();
                 let commit_interval_passed: bool = (now - last_commit_time) > Duration::seconds(15);
-                if self.commit || commit_interval_passed || self.uncommited_events > 100 {
+                if self.commit
+                    || commit_interval_passed
+                    || self.uncommited_events > 100
+                    || self.quit
+                {
                     break;
                 };
             }
             debug!(
                 "Commiting DB! Force commit {}, {} uncommited events",
                 self.commit, self.uncommited_events
             );
-            match transaction.commit() {
+            match tx.commit() {
                 Ok(_) => (),
                 Err(err) => panic!("Failed to commit datastore transaction! {}", err),
             }
@@ -193,32 +213,30 @@ impl DatastoreWorker {
         &mut self,
         request: Command,
         ds: &mut DatastoreInstance,
-        transaction: &Transaction,
+        tx: &Transaction,
     ) -> Result<Response, DatastoreError> {
         match request {
-            Command::CreateBucket(bucket) => match ds.create_bucket(&transaction, bucket) {
+            Command::CreateBucket(bucket) => match ds.create_bucket(tx, bucket) {
                 Ok(_) => {
                     self.commit = true;
                     Ok(Response::Empty())
                 }
                 Err(e) => Err(e),
             },
-            Command::DeleteBucket(bucketname) => {
-                match ds.delete_bucket(&transaction, &bucketname) {
-                    Ok(_) => {
-                        self.commit = true;
-                        Ok(Response::Empty())
-                    }
-                    Err(e) => Err(e),
+            Command::DeleteBucket(bucketname) => match ds.delete_bucket(tx, &bucketname) {
+                Ok(_) => {
+                    self.commit = true;
+                    Ok(Response::Empty())
                 }
-            }
+                Err(e) => Err(e),
+            },
             Command::GetBucket(bucketname) => match ds.get_bucket(&bucketname) {
                 Ok(b) => Ok(Response::Bucket(b)),
                 Err(e) => Err(e),
             },
             Command::GetBuckets() => Ok(Response::BucketMap(ds.get_buckets())),
             Command::InsertEvents(bucketname, events) => {
-                match ds.insert_events(&transaction, &bucketname, events) {
+                match ds.insert_events(tx, &bucketname, events) {
                     Ok(events) => {
                         self.uncommited_events += events.len();
                         self.last_heartbeat.insert(bucketname.to_string(), None); // invalidate last_heartbeat cache
@@ -228,13 +246,7 @@ impl DatastoreWorker {
                 }
             }
             Command::Heartbeat(bucketname, event, pulsetime) => {
-                match ds.heartbeat(
-                    &transaction,
-                    &bucketname,
-                    event,
-                    pulsetime,
-                    &mut self.last_heartbeat,
-                ) {
+                match ds.heartbeat(tx, &bucketname, event, pulsetime, &mut self.last_heartbeat) {
                     Ok(e) => {
                         self.uncommited_events += 1;
                         Ok(Response::Event(e))
@@ -243,31 +255,25 @@ impl DatastoreWorker {
                 }
             }
             Command::GetEvent(bucketname, event_id) => {
-                match ds.get_event(&transaction, &bucketname, event_id) {
+                match ds.get_event(tx, &bucketname, event_id) {
                     Ok(el) => Ok(Response::Event(el)),
                     Err(e) => Err(e),
                 }
             }
             Command::GetEvents(bucketname, starttime_opt, endtime_opt, limit_opt) => {
-                match ds.get_events(
-                    &transaction,
-                    &bucketname,
-                    starttime_opt,
-                    endtime_opt,
-                    limit_opt,
-                ) {
+                match ds.get_events(tx, &bucketname, starttime_opt, endtime_opt, limit_opt) {
                     Ok(el) => Ok(Response::EventList(el)),
                     Err(e) => Err(e),
                 }
             }
             Command::GetEventCount(bucketname, starttime_opt, endtime_opt) => {
-                match ds.get_event_count(&transaction, &bucketname, starttime_opt, endtime_opt) {
+                match ds.get_event_count(tx, &bucketname, starttime_opt, endtime_opt) {
                     Ok(n) => Ok(Response::Count(n)),
                     Err(e) => Err(e),
                 }
             }
             Command::DeleteEventsById(bucketname, event_ids) => {
-                match ds.delete_events_by_id(&transaction, &bucketname, event_ids) {
+                match ds.delete_events_by_id(tx, &bucketname, event_ids) {
                     Ok(()) => Ok(Response::Empty()),
                     Err(e) => Err(e),
                 }
@@ -276,26 +282,26 @@ impl DatastoreWorker {
                 self.commit = true;
                 Ok(Response::Empty())
             }
-            Command::InsertKeyValue(key, data) => {
-                match ds.insert_key_value(&transaction, &key, &data) {
-                    Ok(()) => Ok(Response::Empty()),
-                    Err(e) => Err(e),
-                }
-            }
-            Command::GetKeyValue(key) => match ds.get_key_value(&transaction, &key) {
+            Command::InsertKeyValue(key, data) => match ds.insert_key_value(tx, &key, &data) {
+                Ok(()) => Ok(Response::Empty()),
+                Err(e) => Err(e),
+            },
+            Command::GetKeyValue(key) => match ds.get_key_value(tx, &key) {
                 Ok(result) => Ok(Response::KeyValue(result)),
                 Err(e) => Err(e),
             },
-            Command::GetKeysStarting(pattern) => {
-                match ds.get_keys_starting(&transaction, &pattern) {
-                    Ok(result) => Ok(Response::StringVec(result)),
-                    Err(e) => Err(e),
-                }
-            }
-            Command::DeleteKeyValue(key) => match ds.delete_key_value(&transaction, &key) {
+            Command::GetKeysStarting(pattern) => match ds.get_keys_starting(tx, &pattern) {
+                Ok(result) => Ok(Response::StringVec(result)),
+                Err(e) => Err(e),
+            },
+            Command::DeleteKeyValue(key) => match ds.delete_key_value(tx, &key) {
                 Ok(()) => Ok(Response::Empty()),
                 Err(e) => Err(e),
             },
+            Command::Close() => {
+                self.quit = true;
+                Ok(Response::NoResponse())
+            }
         }
     }
 }
@@ -516,4 +522,10 @@ impl Datastore {
             Err(e) => Err(e),
         }
     }
+
+    // TODO: Should this block until worker has stopped?
+    pub fn close(&self) {
+        info!("Sending close request to database");
+        self.requester.request(Command::Close()).unwrap();
+    }
 }

aw-server/src/main.rs

@@ -27,6 +27,7 @@ struct Opts {
     #[clap(long)]
     port: Option<String>,
     /// Path to database override
+    /// Also implies --no-legacy-import if no db found
     #[clap(long)]
     dbpath: Option<String>,
     /// Path to webui override
@@ -69,7 +70,7 @@ async fn main() -> Result<(), rocket::Error> {
     }
 
     // Set db path if overridden
-    let db_path: String = if let Some(dbpath) = opts.dbpath {
+    let db_path: String = if let Some(dbpath) = opts.dbpath.clone() {
         dbpath
     } else {
         dirs::db_path(testing)
@@ -86,7 +87,11 @@ async fn main() -> Result<(), rocket::Error> {
     };
     info!("Using aw-webui assets at path {:?}", asset_path);
 
-    let legacy_import = !opts.no_legacy_import;
+    // Only use legacy import if opts.dbpath is not set
+    let legacy_import = !opts.no_legacy_import && opts.dbpath.is_none();
+    if opts.dbpath.is_some() {
+        info!("Since custom dbpath is set, --no-legacy-import is implied");
+    }
 
     let device_id: String = if let Some(id) = opts.device_id {
         id

aw-sync/Cargo.toml

@@ -9,7 +9,7 @@ name = "aw_sync"
 path = "src/lib.rs"
 
 [[bin]]
-name = "aw-sync-rust"
+name = "aw-sync"
 path = "src/main.rs"
 
 [dependencies]