initial commit

app/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "hesinde-sync"
+version = "0.1.0"
+edition = "2021"
+license =  { workspace = true }
+authors =  { workspace = true }
+repository =  { workspace = true }
+
+[dependencies]
+entity = { path = "../entity" }
+migration = { path = "../migration" }
+anyhow = "1.0.86"
+clap = { version = "4.5.8", features = ["env", "derive"] }
+poem = "3.0.1"
+poem-openapi = { version = "5.0.2", features = ["swagger-ui"] }
+sea-orm = { workspace = true, features = ["with-time", "sqlx-sqlite", "sqlx-postgres", "sqlx-mysql", "runtime-tokio-rustls", "macros" ] }
+serde = { workspace = true, features = ["derive"] }
+thiserror = "1.0.61"
+time = { workspace = true, features = ["macros", "serde", "formatting", "parsing" ] }
+tokio = { version = "1.38.0", features = ["macros", "rt-multi-thread"] }
+tracing = "0.1.40"
+tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
+uuid = { version = "1.10.0", features = ["v4", "fast-rng"] }

app/src/api.rs

@@ -0,0 +1,137 @@
+use poem::{
+    error::ResponseError,
+    http::StatusCode,
+    web::{Data, Json},
+    Error, Result,
+};
+use poem_openapi::{
+    param::Path,
+    payload::{self, Json as JsonPayload},
+    Object, OpenApi,
+};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+use crate::{
+    app_state::AppState,
+    auth_middleware::{authorize, User},
+    db::DataStoreError,
+    error_response,
+};
+
+#[derive(Object, Deserialize)]
+struct RegisterRequest {
+    username: String,
+    password: String,
+}
+
+#[derive(Object, Serialize)]
+struct UserCreated {
+    username: String,
+}
+
+#[derive(Debug, Clone, Object, Deserialize)]
+pub struct DocumentUpdate {
+    pub device: String,
+    pub device_id: String,
+    pub document: String,
+    pub percentage: f32,
+    pub progress: String,
+}
+
+#[derive(Debug, Clone, Object, Deserialize)]
+pub struct DocumentProgress {
+    pub device: String,
+    pub device_id: String,
+    pub percentage: f32,
+    pub progress: String,
+    pub timestamp: i64,
+}
+
+impl ResponseError for DataStoreError {
+    fn status(&self) -> StatusCode {
+        StatusCode::INTERNAL_SERVER_ERROR
+    }
+
+    fn as_response(&self) -> poem::Response
+    where
+        Self: std::error::Error + Send + Sync + 'static,
+    {
+        error_response::create_and_log(self, || match self {
+            DataStoreError::DatabaseError(_) => (
+                "An internal server error ocurred".to_string(),
+                StatusCode::INTERNAL_SERVER_ERROR,
+            ),
+        })
+    }
+}
+
+pub struct Api;
+
+#[OpenApi]
+impl Api {
+    #[oai(path = "/users/create", method = "post")]
+    async fn register(
+        &self,
+        req: Json<RegisterRequest>,
+        state: Data<&Arc<AppState>>,
+    ) -> Result<payload::Response<payload::Json<UserCreated>>> {
+        let db = &state.0.db;
+        if db.get_user(&req.username).await?.is_some() {
+            Err(Error::from_status(StatusCode::CONFLICT))
+        } else {
+            db.add_user(&req.username, &req.password).await?;
+            Ok(payload::Response::new(payload::Json(UserCreated {
+                username: req.username.clone(),
+            }))
+            .status(StatusCode::CREATED))
+        }
+    }
+
+    #[oai(path = "/users/auth", method = "get", transform = "authorize")]
+    async fn login(&self) -> Result<payload::Response<()>> {
+        Ok(payload::Response::new(()).status(StatusCode::OK))
+    }
+
+    #[oai(path = "/syncs/progress", method = "put", transform = "authorize")]
+    async fn push_progress(
+        &self,
+        state: Data<&Arc<AppState>>,
+        user: Data<&User>,
+        doc: JsonPayload<DocumentUpdate>,
+    ) -> Result<payload::Response<()>> {
+        let db = &state.db;
+
+        let doc_update = (user.name.clone(), doc.0).into();
+        db.update_position(&doc_update).await?;
+
+        Ok(payload::Response::new(()).status(StatusCode::OK))
+    }
+
+    #[oai(
+        path = "/syncs/progress/:doc_id",
+        method = "get",
+        transform = "authorize"
+    )]
+    async fn pull_progress(
+        &self,
+        state: Data<&Arc<AppState>>,
+        user: Data<&User>,
+        doc_id: Path<String>,
+    ) -> Result<payload::Response<JsonPayload<DocumentProgress>>> {
+        let db = &state.db;
+
+        if let Some(doc) = db.get_position(&user.name, &doc_id).await? {
+            Ok(payload::Response::new(JsonPayload(DocumentProgress {
+                device: doc.device,
+                device_id: doc.device_id,
+                percentage: doc.percentage,
+                progress: doc.progress,
+                timestamp: doc.timestamp.assume_utc().unix_timestamp(),
+            }))
+            .status(StatusCode::OK))
+        } else {
+            Err(Error::from_status(StatusCode::NOT_FOUND))
+        }
+    }
+}

app/src/app_state.rs

@@ -0,0 +1,6 @@
+use crate::{cli::Config, db::Db};
+
+pub struct AppState {
+    pub config: Config,
+    pub db: Db,
+}

app/src/auth_middleware.rs

@@ -0,0 +1,108 @@
+use poem::{
+    error::ResponseError, http::StatusCode, Endpoint, EndpointExt, Error, Middleware, Request,
+    Result,
+};
+use std::sync::Arc;
+use thiserror::Error as ThisError;
+
+use crate::{app_state::AppState, error_response};
+
+pub fn authorize(ep: impl Endpoint) -> impl Endpoint {
+    ep.with(AuthMiddleware {})
+}
+
+#[derive(ThisError, Debug)]
+pub enum AuthError {
+    #[error("failed to extract request data")]
+    DataExtractError,
+}
+
+#[derive(Debug, Clone)]
+pub struct User {
+    pub name: String,
+}
+
+impl ResponseError for AuthError {
+    fn status(&self) -> StatusCode {
+        StatusCode::INTERNAL_SERVER_ERROR
+    }
+
+    fn as_response(&self) -> poem::Response
+    where
+        Self: std::error::Error + Send + Sync + 'static,
+    {
+        error_response::create_and_log(self, || match self {
+            AuthError::DataExtractError => (
+                "An internal server error ocurred".to_string(),
+                StatusCode::INTERNAL_SERVER_ERROR,
+            ),
+        })
+    }
+}
+
+struct AuthMiddleware;
+
+impl<E: Endpoint> Middleware<E> for AuthMiddleware {
+    type Output = AuthMiddlewareImpl<E>;
+
+    fn transform(&self, ep: E) -> Self::Output {
+        AuthMiddlewareImpl { ep }
+    }
+}
+
+/// The new endpoint type generated by the TokenMiddleware.
+struct AuthMiddlewareImpl<E> {
+    ep: E,
+}
+
+const HEADER_USER: &str = "x-auth-user";
+const HEADER_KEY: &str = "x-auth-key";
+
+fn get_header(
+    header: &str,
+    req: &Request,
+    error_msg: &str,
+    error_status: StatusCode,
+) -> Result<String> {
+    Ok(req
+        .header(header)
+        .ok_or(Error::from_string(error_msg, error_status))?
+        .to_string())
+}
+
+impl<E: Endpoint> Endpoint for AuthMiddlewareImpl<E> {
+    type Output = E::Output;
+
+    async fn call(&self, mut req: Request) -> Result<Self::Output> {
+        let username = get_header(
+            HEADER_USER,
+            &req,
+            "No user specified",
+            StatusCode::UNAUTHORIZED,
+        )?;
+
+        let key = get_header(
+            HEADER_KEY,
+            &req,
+            "No key specified",
+            StatusCode::UNAUTHORIZED,
+        )?;
+
+        let state = req
+            .data::<Arc<AppState>>()
+            .ok_or(AuthError::DataExtractError)?;
+
+        let db = &state.db;
+        if let Some(user) = db.get_user(&username).await? {
+            if user.key == key {
+                req.extensions_mut().insert(User { name: username });
+                return self.ep.call(req).await;
+            }
+        }
+
+        Err(Error::from_string(
+            "Unauthorized user",
+            StatusCode::FORBIDDEN,
+        ))
+    }
+}

app/src/cli.rs

@@ -0,0 +1,16 @@
+//! Cli interface.
+
+use clap::Parser;
+
+/// Implementation of a koreader-sync server.
+#[derive(Parser, Debug, Clone)]
+#[command(version, about, long_about = None)]
+pub struct Config {
+    /// Address to listen on
+    #[arg(short, long, env, default_value = "localhost:3030")]
+    pub address: String,
+
+    /// From which file to read the database connection string ("-" for stdin)
+    #[arg(short, long, env, default_value = "-")]
+    pub db_connection: String,
+}

app/src/db.rs

@@ -0,0 +1,101 @@
+use ::entity::{document, user};
+use migration::{Migrator, MigratorTrait};
+use sea_orm::{
+    ActiveModelTrait, ColumnTrait, Database, DatabaseConnection, DbErr, EntityTrait, QueryFilter,
+    Set,
+};
+use thiserror::Error;
+use time::{OffsetDateTime, PrimitiveDateTime};
+use tracing::debug;
+
+use crate::api::DocumentUpdate;
+
+#[derive(Error, Debug)]
+pub enum DataStoreError {
+    #[error("database error")]
+    DatabaseError(#[from] DbErr),
+}
+
+#[derive(Debug)]
+pub struct DocumentInsert {
+    id: String,
+    user: String,
+    device: String,
+    device_id: String,
+    percentage: f32,
+    progress: String,
+}
+
+impl From<(String, DocumentUpdate)> for DocumentInsert {
+    fn from(value: (String, DocumentUpdate)) -> Self {
+        Self {
+            id: value.1.document,
+            user: value.0,
+            device: value.1.device,
+            device_id: value.1.device_id,
+            percentage: value.1.percentage,
+            progress: value.1.progress,
+        }
+    }
+}
+
+pub struct Db {
+    connection: DatabaseConnection,
+}
+
+impl Db {
+    pub async fn add_user(&self, id: &str, key: &str) -> Result<user::Model, DataStoreError> {
+        let user = user::ActiveModel {
+            id: Set(id.to_owned()),
+            key: Set(key.to_owned()),
+        };
+        debug!("saving user {user:?}");
+        Ok(user.insert(&self.connection).await?)
+    }
+
+    pub async fn get_user(&self, id: &str) -> Result<Option<user::Model>, DataStoreError> {
+        Ok(user::Entity::find_by_id(id).one(&self.connection).await?)
+    }
+
+    pub async fn get_position(
+        &self,
+        user_id: &str,
+        doc_id: &str,
+    ) -> Result<Option<document::Model>, DataStoreError> {
+        Ok(document::Entity::find()
+            .filter(document::Column::Id.eq(doc_id))
+            .filter(document::Column::User.eq(user_id))
+            .one(&self.connection)
+            .await?)
+    }
+
+    pub async fn update_position(&self, doc: &DocumentInsert) -> Result<(), DataStoreError> {
+        let now = OffsetDateTime::now_utc();
+        let now = PrimitiveDateTime::new(now.date(), now.time());
+
+        let old_doc = self.get_position(&doc.user, &doc.id).await?;
+        let new_doc = document::ActiveModel {
+            id: Set(doc.id.clone()),
+            user: Set(doc.user.clone()),
+            device: Set(doc.device.clone()),
+            device_id: Set(doc.device_id.clone()),
+            percentage: Set(doc.percentage),
+            progress: Set(doc.progress.clone()),
+            timestamp: Set(now),
+        };
+
+        match old_doc {
+            Some(_) => new_doc.update(&self.connection).await?,
+            None => new_doc.insert(&self.connection).await?,
+        };
+
+        Ok(())
+    }
+}
+
+pub async fn connect(connection_string: &str) -> Result<Db, DataStoreError> {
+    let connection: DatabaseConnection = Database::connect(connection_string).await?;
+    Migrator::up(&connection, None).await?;
+
+    Ok(Db { connection })
+}

app/src/error_response.rs

@@ -0,0 +1,38 @@
+use poem::{http::StatusCode, Body, Response};
+use poem_openapi::Object;
+use serde::Serialize;
+use std::error::Error;
+use tracing::error;
+use uuid::Uuid;
+
+#[derive(Object, Serialize)]
+struct ErrorResponse {
+    id: String,
+    message: String,
+}
+
+pub fn create_and_log<F>(error: impl Error, cb: F) -> Response
+where
+    F: Fn() -> (String, StatusCode),
+{
+    let id = Uuid::new_v4().to_string();
+    error!("{id}: {error:?}");
+
+    let (message, status) = cb();
+    let response_json = ErrorResponse {
+        id: id.clone(),
+        message: message.clone(),
+    };
+    let response_body = Body::from_json(response_json).unwrap_or(Body::from_string(
+        format!(
+            "{{\"id\": \"{id}\", \"message\": \"{}\" }}",
+            message.clone()
+        )
+        .to_owned(),
+    ));
+
+    Response::builder()
+        .status(status)
+        .header("Content-Type", "application/json")
+        .body(response_body)
+}

app/src/lib.rs

@@ -0,0 +1,46 @@
+use anyhow::Result;
+use api::Api;
+use app_state::AppState;
+use cli::Config;
+use poem::{
+    http::{uri::Scheme, Uri},
+    listener::TcpListener,
+    middleware::Tracing,
+    EndpointExt, Route,
+};
+use poem_openapi::OpenApiService;
+use std::sync::Arc;
+
+pub mod api;
+pub mod app_state;
+pub mod auth_middleware;
+pub mod cli;
+pub mod db;
+pub mod error_response;
+
+pub async fn run(args: &Config, db_url: &str) -> Result<()> {
+    let db = db::connect(db_url).await?;
+    let app_state = Arc::new(AppState {
+        config: args.clone(),
+        db,
+    });
+
+    const API_PATH: &str = "/api";
+    let api_uri = Uri::builder()
+        .scheme(Scheme::HTTP)
+        .authority(args.address.clone())
+        .path_and_query(API_PATH)
+        .build()?;
+
+    let api_service = OpenApiService::new(Api, "Hesinde Sync", "1.0").server(api_uri.to_string());
+    let ui = api_service.swagger_ui();
+    let app = Route::new()
+        .nest(API_PATH, api_service)
+        .nest("/", ui)
+        .data(app_state)
+        .with(Tracing);
+
+    Ok(poem::Server::new(TcpListener::bind(&args.address))
+        .run(app)
+        .await?)
+}

app/src/main.rs

@@ -0,0 +1,41 @@
+use std::{
+    fs::File,
+    io::{self, BufRead, Read},
+};
+
+use anyhow::Result;
+use clap::Parser;
+use hesinde_sync::cli::Config;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    if std::env::var_os("RUST_LOG").is_none() {
+        std::env::set_var("RUST_LOG", "info");
+    }
+    tracing_subscriber::fmt::init();
+
+    let args = Config::parse();
+    let db_url = read_db_url(&args.db_connection)?;
+
+    hesinde_sync::run(&args, &db_url).await
+}
+
+/// Read db url from file or stdin.
+fn read_db_url(db_arg: &str) -> Result<String> {
+    if db_arg == "-" {
+        let stdin = io::stdin();
+        let mut buffer = String::new();
+
+        stdin.lock().read_to_string(&mut buffer)?;
+        let db_url = buffer.trim();
+        Ok(db_url.to_string())
+    } else {
+        let file = File::open(db_arg)?;
+        let mut reader = io::BufReader::new(file);
+        let mut first_line = String::new();
+        reader.read_line(&mut first_line)?;
+
+        let first_line = first_line.trim();
+        Ok(first_line.to_string())
+    }
+}