record_watch/src/main.rs

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 tokio::process::Command;
use walkdir::DirEntry;
use walkdir::WalkDir;
mod mpvctl;
use crate::mpvctl::MpvCtl;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut args = env::args();
    let _program = args.next().expect("argv[0] should be this program?");
    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 pool =
        SqlitePool::connect_with(SqliteConnectOptions::from_str(&url)?.create_if_missing(true))
            .await?;
    sqlx::migrate!("./migrations").run(&pool).await?;

    let profile_result = sqlx::query(r#"INSERT OR IGNORE INTO profile (name) VALUES (?)"#)
        .bind(&profile)
        .execute(&pool)
        .await?;
    let count = profile_result.rows_affected();
    if count != 0 {
        println!("Profile: {:?}", count);
    }
    println!("Profile: {:?}", profile_result);

    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"])?;
    let client_name = mpvctl.get_client_name().await?;
    for f in files {
        let result = sqlx::query(r#"SELECT 1 FROM watched WHERE path = ? AND profile = (SELECT id FROM PROFILE WHERE name = ?)"#)
            .bind(f.path().as_os_str().to_str())
            .bind(&profile)
            .execute(&pool).await?;
        println!("Result: {:?}", result);

        println!("Launching {}", f.path().display());
        mpvctl.play_video(f.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 = sqlx::query(r#"INTO INTO watched (profile, path, watched_at)WHERE path = ? AND profile = (SELECT id FROM PROFILE WHERE name = ?)"#)
                        //     .bind(f.path().as_os_str().to_str())
                        //     .bind(&profile)
                        //     .execute(&pool).await?;
                        // println!("Insert: {:?}", insert);

                        continue;
                    }
                }
            }
        }
        break;
    }

    println!("done {}", client_name);

    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(())
}