Add support for a very basic randomize.

Cargo.toml

@@ -3,7 +3,9 @@ name = "record_watch"
 version = "0.1.0"
 edition = "2021"
 
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+[[bin]]
+name = "rw"
+path = "src/main.rs"
 
 [dependencies]
 walkdir = "2.3.2"
@@ -13,3 +15,5 @@ dirs = "4.0.0"
 serde_json = "1.0.79"
 serde = { version = "1.0.136", features= ["derive"] }
 bytes = "1.1.0"
+rand = "0.8.5"
+

migrations/00001_create_watched_table.sql

@@ -1,11 +1,11 @@
 CREATE TABLE profile (
-       id INTEGER PRIMARY KEY AUTOINCREMENT,
+       id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        UNIQUE(name)
 );
 
 CREATE TABLE watched (
-       id INTEGER PRIMARY KEY AUTOINCREMENT,
+       id INTEGER PRIMARY KEY,
        profile INTEGER NOT NULL,
        path TEXT NOT NULL,
        watched_at DATE NOT NULL,

src/main.rs

@@ -1,22 +1,30 @@
 use sqlx::sqlite::SqliteConnectOptions;
-use sqlx::sqlite::SqlitePool;
-use std::env;
-use std::path::Path;
-use std::process::Stdio;
-use std::str::FromStr;
-use tokio::io::{AsyncBufReadExt, BufReader};
+use sqlx::Connection;
+use sqlx::Row;
+use std::{env, str::FromStr};
 use tokio::process::Command;
 use walkdir::DirEntry;
 use walkdir::WalkDir;
 mod mpvctl;
 use crate::mpvctl::MpvCtl;
+use rand::seq::SliceRandom;
+use rand::thread_rng;
 use tokio::time::{sleep, Duration};
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let mut args = env::args();
+    let mut args = env::args().peekable();
     let _program = args.next().expect("argv[0] should be this program?");
+    let randomize: bool = match args.peek() {
+        Some(flag) if flag == "-r" => {
+            // Consume the flag
+            args.next().expect("We already know this arg exists");
+            true
+        }
+        _ => false,
+    };
     let profile = args.next().expect("Must provide a profile");
+
     let directories: Vec<String> = args.collect();
 
     let mut files: Vec<DirEntry> = Vec::new();
@@ -36,14 +44,24 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .unwrap();
     let url = format!("sqlite://{}", db_path.as_path().display());
 
-    let pool =
-        SqlitePool::connect_with(SqliteConnectOptions::from_str(&url)?.create_if_missing(true))
-            .await?;
-    sqlx::migrate!("./migrations").run(&pool).await?;
+    let mut dbconn = sqlx::SqliteConnection::connect_with(
+        &SqliteConnectOptions::from_str(&url)?
+            .create_if_missing(true)
+            // Use the normal journal mode to avoid those pesky shm and wal files
+            .journal_mode(sqlx::sqlite::SqliteJournalMode::Delete),
+    )
+    .await?;
+    sqlx::migrate!("./migrations").run(&mut dbconn).await?;
 
-    for f in files {
-        println!("Launching {}", f.path().display());
+    let profiles_created: u64 = sqlx::query(r#"INSERT OR IGNORE INTO profile (name) VALUES (?)"#)
+        .bind(&profile)
+        .execute(&mut dbconn)
+        .await?
+        .rows_affected();
+    if profiles_created != 0 {
+        println!("Created a new profile");
     }
+
     launch_mpv().await?;
 
     // TODO: Figure out a better way to wait for the socket to exist and be connectable
@@ -51,20 +69,46 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let mut mpvctl = MpvCtl::connect("/tmp/recordwatchsocket").await?;
     let mut end_file_listener = mpvctl.listen(&["end-file"])?;
-    let client_name = mpvctl.get_client_name().await?;
-    let _ = mpvctl
-        .play_video("/home/talexander/Downloads/test.mp4")
-        .await?;
+    if randomize {
+        files.shuffle(&mut thread_rng());
+    }
+    for f in files {
+        let canonicalized_path = f.path().canonicalize()?;
+        let already_watched_count: i64 = sqlx::query(r#"SELECT count(*) FROM watched WHERE path = ? AND profile = (SELECT id FROM PROFILE WHERE name = ?)"#)
+            .bind(canonicalized_path.as_os_str().to_str())
+            .bind(&profile)
+            .fetch_one(&mut dbconn).await?.try_get("count(*)")?;
 
-    while let Some(evt) = end_file_listener.recv().await {
-        println!("end file event {}", evt);
+        println!("Already watched count: {}", already_watched_count);
+        if already_watched_count > 0 {
+            println!("Skipping: Already watched {}", canonicalized_path.display());
+            continue;
+        }
+
+        println!("Launching {}", canonicalized_path.display());
+        mpvctl.play_video(&canonicalized_path).await?;
+        if let Some(evt) = end_file_listener.recv().await {
+            println!("end file event {}", evt);
+            if let serde_json::Value::Object(obj) = evt {
+                let reason = obj.get("reason");
+                if let Some(serde_json::Value::String(reason_body)) = reason {
+                    if reason_body == "eof" {
+                        // Watched the video until the end
+                        let insert: u64 = sqlx::query(r#"INSERT INTO watched (profile, path, watched_at) SELECT (SELECT id FROM PROFILE WHERE name = ?) profile, ? path, DATETIME('now') watched_at"#)
+                            .bind(&profile)
+                            .bind(canonicalized_path.as_os_str().to_str())
+                            .execute(&mut dbconn).await?.rows_affected();
+                        println!("Insert: {:?}", insert);
+                        assert!(insert == 1);
+                        continue;
+                    }
+                }
+            }
+        }
+        break;
     }
 
-    sleep(Duration::from_secs(50)).await;
-    let client_name = mpvctl.get_client_name().await?;
-
-    println!("done {}", client_name);
-
+    dbconn.close().await?;
     Ok(())
 }
 

src/mpvctl/command.rs

@@ -1,5 +1,5 @@
 use serde::ser::SerializeMap;
-use serde::{Deserialize, Serialize, Serializer};
+use serde::{Serialize, Serializer};
 
 pub struct Command {
     command: Vec<String>,

src/mpvctl/mpv_framed.rs

@@ -1,5 +1,5 @@
 use bytes::{Buf, BytesMut};
-use std::io::{self, Cursor};
+use std::io::Cursor;
 use tokio::{io::AsyncReadExt, net::unix::OwnedReadHalf};
 
 pub struct MpvFramed {

src/mpvctl/mpvctl.rs

@@ -1,14 +1,10 @@
 use super::command::Command;
 use super::mpv_framed::MpvFramed;
-use bytes::BytesMut;
-use std::borrow::BorrowMut;
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
-use std::io::Read;
-use std::os::unix::prelude::OsStrExt;
+use std::path::Path;
 use std::sync::{Arc, Mutex};
-use std::{collections::BTreeMap, path::Path};
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::io::AsyncWriteExt;
 use tokio::net::unix::OwnedWriteHalf;
 use tokio::net::UnixStream;
 use tokio::sync::mpsc;
@@ -88,75 +84,70 @@ impl MpvCtl {
         db: &CommandDb,
         event_listeners: &EventDb,
     ) -> Result<(), Box<dyn std::error::Error>> {
-        match framed_read.read_frame().await {
-            Ok(Some(frame)) => {
-                println!("Read {}", frame);
-                // check if its an event
-                let is_event = match &frame {
-                    serde_json::Value::Object(obj) => obj.contains_key("event"),
-                    _ => false,
+        let current_frame = framed_read.read_frame().await.map_err(|e| e.to_string())?;
+        if let Some(frame) = current_frame {
+            println!("Read {}", frame);
+            // check if its an event
+            let is_event = match &frame {
+                serde_json::Value::Object(obj) => obj.contains_key("event"),
+                _ => false,
+            };
+            if is_event {
+                let obj = match &frame {
+                    serde_json::Value::Object(obj) => obj,
+                    _ => {
+                        return Err("Got back a json value that wasn't an object.".into());
+                    }
                 };
-                if is_event {
-                    let obj = match &frame {
-                        serde_json::Value::Object(obj) => obj,
-                        _ => {
-                            return Err("Got back a json value that wasn't an object.".into());
-                        }
-                    };
-                    match obj.get("event") {
-                        Some(orig @ serde_json::Value::String(s)) => {
-                            println!("Notifying listeners for event {}", s);
-                            let listeners = {
-                                let db_handle = event_listeners.lock().unwrap();
-                                match (*db_handle).get(s) {
-                                    None => Vec::new(),
-                                    Some(listeners) => listeners.to_owned(),
-                                }
-                            };
-                            for listener in listeners {
-                                listener.send(frame.clone()).await;
+                match obj.get("event") {
+                    Some(serde_json::Value::String(s)) => {
+                        println!("Notifying listeners for event {}", s);
+                        let listeners = {
+                            let db_handle = event_listeners.lock().unwrap();
+                            match (*db_handle).get(s) {
+                                None => Vec::new(),
+                                Some(listeners) => listeners.to_owned(),
                             }
-                        }
-                        _ => return Err("Event with no string value".into()),
-                    };
-                    return Ok(());
-                }
-                // get the request id and push the result into the channel
-                let reqid = {
-                    let obj = match &frame {
-                        serde_json::Value::Object(obj) => obj,
-                        _ => {
-                            return Err("Got back a json value that wasn't an object.".into());
-                        }
-                    };
-                    match obj.get("request_id") {
-                        Some(serde_json::Value::Number(reqid)) if reqid.is_u64() => {
-                            reqid.as_u64().unwrap()
-                        }
-                        _ => {
-                            return Err("Unrecognized request_id".into());
+                        };
+                        for listener in listeners {
+                            listener.send(frame.clone()).await?;
                         }
                     }
+                    _ => return Err("Event with no string value".into()),
                 };
-
-                {
-                    let mut db_handle = db.lock().unwrap();
-                    if let Entry::Occupied(o) = (*db_handle).entry(reqid) {
-                        o.remove()
-                            .send(frame)
-                            .map_err(|e| "Failed to send frame".to_string())?;
-                    } else {
-                        return Err("No entry found for request id".into());
-                    }
-                }
-            }
-            Ok(None) => {
                 return Ok(());
             }
-            Err(e) => {
-                return Err("Ran into a problem".into());
+            // get the request id and push the result into the channel
+            let reqid = {
+                let obj = match &frame {
+                    serde_json::Value::Object(obj) => obj,
+                    _ => {
+                        return Err("Got back a json value that wasn't an object.".into());
+                    }
+                };
+                match obj.get("request_id") {
+                    Some(serde_json::Value::Number(reqid)) if reqid.is_u64() => {
+                        reqid.as_u64().unwrap()
+                    }
+                    _ => {
+                        return Err("Unrecognized request_id".into());
+                    }
+                }
+            };
+
+            {
+                let mut db_handle = db.lock().unwrap();
+                if let Entry::Occupied(o) = (*db_handle).entry(reqid) {
+                    o.remove()
+                        .send(frame)
+                        .map_err(|e| "Failed to send frame".to_string())?;
+                } else {
+                    return Err("No entry found for request id".into());
+                }
             }
-        }
+        } else if let None = current_frame {
+            return Ok(());
+        };
         Ok(())
     }