Create database entries on file upload

@@ -20,3 +20,15 @@ pub struct MarkdownDiffJson {
     pub id: i32,
     pub date_created: String,
 }
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct ImageJson {
+    pub upload_uuid  :Option<Vec<u8>>,
+    pub uuid         :Option<Vec<u8>>,
+    pub size         :i32,
+    pub dim_x        :Option<i32>,
+    pub dim_y        :Option<i32>,
+    pub mime_type    :String,
+    pub date_created :String,
+    pub date_updated :String
+}

+-- This file should undo anything in `up.sql`
+DROP TABLE "images";

+CREATE TABLE "images" (
+	id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
+	-- identical uuid means identical file.
+	upload_uuid BLOB(16) UNIQUE,
+	uuid BLOB(16) UNIQUE,
+	size INTEGER NOT NULL,
+	dim_x INTEGER,
+	dim_y INTEGER,
+	mime_type VARCHAR(256) NOT NULL,
+	date_created TEXT NOT NULL,
+	date_updated TEXT NOT NULL
+);

@@ -7,7 +7,7 @@ use r2d2;
 type ParentError = Option<Pin<Box<dyn std::error::Error>>>;
 
 #[derive(Debug)]
-pub(crate) struct Error {
+pub struct Error {
     source: ParentError,
     message: String,
 }
@@ -77,3 +77,11 @@ impl From<ParsePatchError> for Error {
         }
     }
 }
+
+impl From<uuid::Error> for Error {
+    fn from(source: uuid::Error) -> Self {
+        Self { source: Some(Box::pin(source))
+             , message: String::from("UUID error")
+        }
+    }
+}

@@ -6,11 +6,15 @@ mod models;
 mod routes;
 mod schema;
 mod uuid;
+mod upload;
 
+use async_std::channel::Receiver;
+use models::image::Image;
 use routes::markdown::*;
 use routes::other::*;
 use routes::user::*;
 use routes::upload::*;
+use crate::upload::get_sample;
 use crate::uuid::Uuid;
 
 use actix_web::{guard, web, App, HttpResponse, HttpServer};
@@ -18,10 +22,10 @@ use async_std::{ channel::{ Sender, bounded }
                , fs::File };
 use diesel::r2d2::{self, ConnectionManager};
 use diesel::SqliteConnection;
-use futures::{ AsyncReadExt, AsyncSeekExt, FutureExt, StreamExt, select
+use futures::{ FutureExt, StreamExt, select
              , stream::FuturesUnordered };
 use listenfd::ListenFd;
-use std::io::SeekFrom;
+use std::convert::TryFrom;
 use std::sync::Arc;
 
 pub(crate) type Pool = r2d2::Pool<ConnectionManager<SqliteConnection>>;
@@ -29,34 +33,7 @@ pub(crate) type Pool = r2d2::Pool<ConnectionManager<SqliteConnection>>;
 #[derive(Clone)]
 pub struct AppData {
     pub database_pool: Arc<Pool>,
-    pub tx_upload_worker: Sender<String>,
-}
-
-async fn read_at( f   :&mut File
-                , pos :SeekFrom
-                , buf :&mut [u8]) -> std::io::Result<()> {
-    f.seek(pos).await?;
-    f.read_exact(buf).await
-}
-
-async fn get_sample( f   :&mut File
-                   , buf :&mut [u8]) -> std::io::Result<()> {
-    let file_len = f.metadata().await?.len();
-    let chunk_size = buf.len() / 3;
-
-    read_at(f, SeekFrom::Start(0), &mut buf[0..chunk_size]).await?;
-    if file_len >= 2 * chunk_size as u64 {
-        read_at( f
-               , SeekFrom::End(-(chunk_size as i64))
-               , &mut buf[2*chunk_size..]).await?;
-    }
-    if file_len >= 3 * chunk_size as u64 {
-        read_at( f
-               , SeekFrom::Start((file_len-chunk_size as u64) / 2)
-               , &mut buf[chunk_size..2*chunk_size]).await?;
-    }
-
-    Ok(())
+    pub tx_upload_worker: Sender<Image>,
 }
 
 #[actix_rt::main]
@@ -65,7 +42,8 @@ async fn main() -> std::io::Result<()> {
 
     dotenv::dotenv().ok();
 
-    let (tx_upload_worker, rx_upload_worker) = bounded(32);
+    let (tx_upload_worker, rx_upload_worker)
+        : (Sender<Image>, Receiver<Image>) = bounded(32);
 
     let _upload_worker = actix_rt::spawn(async move {
         let mut workers = FuturesUnordered::new();
@@ -75,12 +53,14 @@ async fn main() -> std::io::Result<()> {
                 filename = rx_upload_worker.recv().fuse() => {
                     match filename {
                         Err(_) => break,
-                        Ok(filename) => workers.push(async move {
-                            let mut f = File::open(&filename).await.unwrap();
+                        Ok(upload) => workers.push(async move {
+                            let upload_uuid = Uuid::try_from(upload.upload_uuid.unwrap()).unwrap();
+                            let upload_filename = format!("/tmp/upload_{}", upload_uuid);
+                            let mut f = File::open(&upload_filename).await.unwrap();
                             let mut buf = vec!['.' as u8; 3 * 4096];
                             get_sample(&mut f, buf.as_mut()).await.unwrap();
                             println!( "[UPLOAD WORKER] filename: {}"
-                                    , filename );
+                                    , upload_filename );
                             println!( "[UPLOAD WORKER] uuid: {}"
                                     , Uuid::get( "some.unique.namespace"
                                                , buf.as_mut() ) );

+use std::sync::Arc;
+
+use crate::error::*;
+use crate::{schema::*, Pool};
+use diesel::{Connection, insert_into};
+use diesel::prelude::*;
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Serialize, Deserialize, Queryable, Identifiable)]
+pub struct Image {
+    pub id           :i32,
+    pub upload_uuid  :Option<Vec<u8>>,
+    pub uuid         :Option<Vec<u8>>,
+    pub size         :i32,
+    pub dim_x        :Option<i32>,
+    pub dim_y        :Option<i32>,
+    pub mime_type    :String,
+    pub date_created :String,
+    pub date_updated :String
+}
+
+#[derive(Debug, Insertable)]
+#[table_name = "images"]
+pub struct ImageNew<'a> {
+    pub upload_uuid  :Option<&'a [u8]>,
+    pub size         :i32,
+    pub mime_type    :&'a str,
+    pub date_created :&'a str,
+    pub date_updated :&'a str
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, AsChangeset)]
+#[table_name = "images"]
+pub struct Upload {
+    pub upload_uuid  :Option<Vec<u8>>,
+    pub size         :i32,
+    pub mime_type    :String,
+}
+
+pub(crate) fn upload( pool: Arc<Pool>
+                    , item: Upload ) -> Result<Image> {
+    use crate::schema::images::dsl::*;
+    let db_connection = pool.get()?;
+
+    let now = chrono::Local::now().naive_local();
+    let new_image = ImageNew {
+        upload_uuid  : item.upload_uuid.as_deref(),
+        size         : item.size,
+        mime_type    : &item.mime_type,
+        date_created : &format!("{}", now),
+        date_updated : &format!("{}", now)
+    };
+
+    Ok(db_connection.transaction(|| {
+        insert_into(images) . values(&new_image)
+                            . execute(&db_connection)?;
+        images . order(id.desc())
+               . first::<Image>(&db_connection)
+    })?)
+}

-pub(crate) mod user;
+pub(crate) mod image;
 pub(crate) mod markdown;
+pub(crate) mod user;

@@ -4,25 +4,47 @@ use futures::stream::StreamExt;
 use async_std::{fs::OpenOptions, io::WriteExt};
 use uuid::Uuid;
 
-use crate::AppData;
+use crate::{AppData, models::image::{Upload, self}};
 
-pub async fn upload( app_data: web::Data<AppData>
-                   , mut body: web::Payload) -> Result<HttpResponse, Error>
+pub async fn upload( app_data :web::Data<AppData>
+                   , mut body :web::Payload
+                   , request  :web::HttpRequest ) -> Result<HttpResponse, Error>
 {
+    let pool = app_data.database_pool.clone();
     let worker = app_data.tx_upload_worker.clone();
 
-    let upload_filename = format!("/tmp/upload_{}", Uuid::new_v4());
+    let random_uuid = Uuid::new_v4();
+    let upload_uuid = Some(random_uuid.as_bytes().to_vec());
+    let size = request.headers().get("content-length")
+             . and_then(|h| Some(h.to_str().unwrap().parse::<i32>()))
+             . unwrap().unwrap();
+    let mime_type = String::from( request.headers().get("content-type")
+                                . and_then(|h| Some(h.to_str().unwrap()))
+                                . unwrap() );
+
+    let upload_filename = format!("/tmp/upload_{}", random_uuid);
     let mut output = OpenOptions::new();
-    output . create(true)
-           . write(true);
-    let mut output = output.open(&upload_filename).await?;
+    let mut output = output
+                   . create(true)
+                   . write(true)
+                   . open(&upload_filename).await?;
 
     while let Some(item) = body.next().await {
         output.write_all(&item?).await?;
     }
 
-    // TODO handle this as error response...
-    worker.send(upload_filename).await.unwrap();
+    let upload = Upload {
+        upload_uuid,
+        size,
+        mime_type
+    };
 
-    Ok(HttpResponse::Ok().finish())
+    Ok( match web::block(move || image::upload(pool, upload)).await {
+        Ok(image) => {
+            // TODO handle this as error response...
+            worker.send(image.clone()).await.unwrap();
+            HttpResponse::Ok().json(image)
+        },
+        Err(_) => HttpResponse::InternalServerError().finish()
+    } )
 }

 table! {
+    images (id) {
+        id -> Integer,
+        upload_uuid -> Nullable<Binary>,
+        uuid -> Nullable<Binary>,
+        size -> Integer,
+        dim_x -> Nullable<Integer>,
+        dim_y -> Nullable<Integer>,
+        mime_type -> Text,
+        date_created -> Text,
+        date_updated -> Text,
+    }
+}
+
+table! {
     markdown_diffs (markdown_id, diff_id) {
         markdown_id -> Integer,
         diff_id -> Integer,
@@ -28,6 +42,7 @@ table! {
 }
 
 allow_tables_to_appear_in_same_query!(
+    images,
     markdown_diffs,
     markdowns,
     users,

+use std::io::SeekFrom;
+use async_std::fs::File;
+use futures::{AsyncSeekExt, AsyncReadExt};
+
+
+async fn read_at( f   :&mut File
+                    , pos :SeekFrom
+                    , buf :&mut [u8]) -> std::io::Result<()> {
+    f.seek(pos).await?;
+    f.read_exact(buf).await
+}
+
+pub async fn get_sample( f   :&mut File
+                       , buf :&mut [u8]) -> std::io::Result<()> {
+    let file_len = f.metadata().await?.len();
+    let chunk_size = buf.len() / 3;
+
+    read_at(f, SeekFrom::Start(0), &mut buf[0..chunk_size]).await?;
+    if file_len >= 2 * chunk_size as u64 {
+        read_at( f
+               , SeekFrom::End(-(chunk_size as i64))
+               , &mut buf[2*chunk_size..]).await?;
+    }
+    if file_len >= 3 * chunk_size as u64 {
+        read_at( f
+               , SeekFrom::Start((file_len-chunk_size as u64) / 2)
+               , &mut buf[chunk_size..2*chunk_size]).await?;
+    }
+
+    Ok(())
+}

-use std::fmt::Display;
+use std::{fmt::Display, convert::TryFrom};
+
+use crate::error::Error;
 
 #[derive(Clone,Copy,Debug)]
 pub struct Uuid(pub uuid::Uuid);
@@ -20,3 +22,11 @@ impl Uuid {
         Self(uuid::Uuid::new_v5(&ns!(ns), buf))
     }
 }
+
+impl TryFrom<Vec<u8>> for Uuid {
+    type Error = Error;
+
+    fn try_from(value: Vec<u8>) -> Result<Self, Self::Error> {
+        Ok(Self(uuid::Uuid::from_slice(value.as_slice())?))
+    }
+}