record_watch/src/main.rs

use sqlx::sqlite::SqliteConnectOptions;
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().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();
    for dir in directories {
        for entry in WalkDir::new(dir) {
            let entry = entry.unwrap();
            if entry.file_type().is_file() {
                files.push(entry);
            }
        }
    }

    files.sort_by_key(|file| file.path().to_owned());

    let db_path = dirs::home_dir()
        .map(|pb| pb.join(".record_watch.sqlite"))
        .unwrap();
    let url = format!("sqlite://{}", db_path.as_path().display());

    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?;

    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
    sleep(Duration::from_secs(1)).await;

    let mut mpvctl = MpvCtl::connect("/tmp/recordwatchsocket").await?;
    let mut end_file_listener = mpvctl.listen(&["end-file"])?;
    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(*)")?;

        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;
    }

    dbconn.close().await?;
    Ok(())
}

async fn launch_mpv() -> Result<(), Box<dyn std::error::Error>> {
    let mut cmd = Command::new("mpv");
    cmd.arg("--input-ipc-server=/tmp/recordwatchsocket")
        .arg("--idle");
    cmd.kill_on_drop(true);
    let mut child = cmd.spawn().expect("failed to spawn command");
    // Launch the child in the runtime so it can run without blocking this thread
    tokio::spawn(async move {
        let status = child.wait().await.expect("mpv encountered an error");
        println!("mpv status was: {}", status);
    });
    Ok(())
}