path: root/src/importer.rs

use diesel::PgConnection;
use std::sync::mpsc::channel;
use std::sync::mpsc::Receiver;
use std::sync::mpsc::Sender;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::RwLock;
use threadpool::ThreadPool;
use chrono::Utc;
use timer::Timer;
use timer::Guard;
use std::time::Instant;
use std::time::Duration;
use std::thread;
use serde::Deserialize;
use serde::Serialize;

use crate::error::Error;
use crate::db;
use crate::models;
use crate::strava;
use crate::strava::StravaApi;
use crate::Params;

pub const WORKERS: usize = 10;
pub const EMPTY_PARAMS: &[(&str, &str)] = &[];

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum Command {
    ImportStravaUser { username: String },
}

macro_rules! clone {
    ( [ $( $i:ident ),* ] $e:expr ) => {
        {
            $(let $i = $i.clone();)*
                $e
        }
    }
}

pub struct ImporterSharedData<StravaApi: strava::StravaApi + 'static> {
    strava: RwLock<StravaApi>,
    pool: Mutex<ThreadPool>,
    conn: Mutex<PgConnection>,
    running: Mutex<bool>,
}

pub struct Importer<StravaApi: strava::StravaApi + 'static> {
    shared: Arc<ImporterSharedData<StravaApi>>,
}

fn run_periodically<S: strava::StravaApi>(
    shared: Arc<ImporterSharedData<S>>,
    period: Duration) {
    let sleep_time = Duration::from_millis(1000);
    let mut now = Instant::now();
    loop {
        while now.elapsed() < period {
            if !*shared.running.lock().unwrap() {
                return;
            }
            thread::sleep(sleep_time);
        }
        now = Instant::now();

        info!("run_periodically: wakeup");
        handle_tasks(shared.clone())
    }
}


fn handle_one_task<S: strava::StravaApi>(
    shared: Arc<ImporterSharedData<S>>) -> Result<models::Task, Error> {
    let task = {
        let conn = shared.conn.lock().unwrap();
        let now = Utc::now();
        let eta = now + chrono::Duration::seconds(5);

        db::take_task(&conn,
                      models::TaskState::NEW,
                      now,
                      eta)?
    };

    let command = serde_json::from_value(task.payload.clone())?;

    match command {
        Command::ImportStravaUser{ username } => {
            import_strava_user(shared, username.as_str())?
        },
    }

    Ok(task)
}

fn handle_tasks<S: strava::StravaApi>(
    shared: Arc<ImporterSharedData<S>>) {
    let mut done = false;
    while !done {
        match handle_one_task(shared.clone()) {
            Err(Error::NotFound) => {
                info!("No more tasks");
                done = true;
            },
            Err(e) => {
                error!("Error handling task: {}", e);
            }
            Ok(t) => {
                info!("Successfully handled task: {:?}", t);
            }
        };
    }
}

impl<StravaApi: strava::StravaApi> Importer<StravaApi> {
    pub fn new(conn: PgConnection, strava: StravaApi) -> Importer<StravaApi> {
        let shared = Arc::new(ImporterSharedData {
            pool: Mutex::new(ThreadPool::with_name("importer".to_string(), WORKERS)),
            conn: Mutex::new(conn),
            strava: RwLock::new(strava),
            running: Mutex::new(false),
        });
        Importer { shared: shared }
    }

    pub fn run(&self) {
        info!("run()");
        let pool = self.shared.pool.lock().unwrap();
        let mut running = self.shared.running.lock().unwrap();
        if !*running {
            *running = true;
            pool.execute({
                let shared = self.shared.clone();
                move || run_periodically(shared, Duration::from_secs(10))
            });
        }
    }

    pub fn join(&self) {
        self.shared.pool.lock().expect("FIX").join()
    }
}

fn import_strava_user<S: strava::StravaApi>(
    shared: Arc<ImporterSharedData<S>>,
    username: &str) -> Result<(), Error> {
    let strava = shared.strava.read().unwrap();
    let user = db::get_user(&shared.conn.lock().unwrap(), username)?;

    let token = {
        let conn = shared.conn.lock().unwrap();
        get_or_refresh_token(&*strava, &conn, &user)?
    };

    let per_page = 30;
    for page in 1.. {
        let params = [
            ("page", &format!("{}", page)[..]),
            ("per_page", &format!("{}", per_page)[..])
        ];

        let result = strava
            .get("/athlete/activities", &token.access_token, &params[..])?;

        let result = result.as_array().ok_or(
            Error::UnexpectedJson(result.clone()))?;

        for activity in result {
            info!("activity id: {} start: {}", activity["id"], activity["start_date"]);
        }

        if result.len() < per_page {
            break;
        }
        thread::sleep(Duration::from_secs(1));
    };

    Err(Error::InternalError)
}

fn get_or_refresh_token<Strava: strava::StravaApi>(strava: &Strava, conn: &PgConnection, user: &models::User) -> Result<models::StravaToken, Error> {
    let mut token = db::get_strava_token(&conn, &user).expect("FIX");

    if token.expires_at < Utc::now() {
        info!("refresh expired token: {:?}", token.expires_at);
        let new_token = strava.refresh_token(&From::from(&token))?;
        new_token.update_model(&mut token);
    }

    Ok(token)
}

// fn handle_command(state: Importer, command: Command) {
//     info!("handle_command {:?}", command);
//     match command {
//         Command::ImportStravaUser(user) => import_strava_user(state, user),
//         Command::Quit => (),
//     }
// }

// fn receive_commands(state: Importer) {
//     info!("receive_commands");
//     match (|| -> Result<(), Box<dyn std::error::Error>> {
//         let rx = state.rx.lock()?;
//         let mut command = rx.recv()?;
//         loop {
//             info!("got command: {:?}", command);
//             let state0 = state.clone();
//             state.pool.execute(move || handle_command(state0, command));
//             command = rx.recv()?;
//         }
//     })() {
//         Ok(()) => (),
//         Err(e) => {
//             error!("receive_commands: {:?}", e);
//             ()
//         }
//     }
// }

// pub fn run(pool: ThreadPool, conn: PgConnection, params: &Params) -> Sender<Command> {
//     let (tx, rx0) = channel();
//     let importer = Arc::new(Importer {
//         pool: Mutex::new(pool.clone()),
//         conn: Mutex::new(conn),
//         strava: RwLock::new(strava::StravaImpl::new(
//         params.strava_client_id.clone(), params.strava_client_secret.clone())),
//         rx: Mutex::new(rx0),
//     });
//     // pool.execute(move || receive_commands(state));
//     pool.execute(clone! { [importer] move || importer.run() });
//     tx
// }