text: use std::ffi::{CStr, CString};
  use std::fmt::{self, Debug};
  use std::os::raw::{c_int, c_void};
  use std::ptr;
  use std::thread::panicking;
  
  use libsqlite3_sys as ffi;
  
  use base::crypto::{Crypto, Key};
  use base::vio;
  use error::{Error, Result};
  use trans::Eid;
  use volume::address::Span;
  use volume::storage::Storable;
  use volume::BLK_SIZE;
  
  // check result code returned by sqlite
  fn check_result(result: c_int) -> Result<()> {
      if result != ffi::SQLITE_OK {
          let err = ffi::Error::new(result);
          return Err(Error::from(err));
      }
      Ok(())
  }
  
  // reset and clean up statement
  fn reset_stmt(stmt: *mut ffi::sqlite3_stmt) -> Result<()> {
      let result = unsafe { ffi::sqlite3_reset(stmt) };
      check_result(result)?;
      let result = unsafe { ffi::sqlite3_clear_bindings(stmt) };
      check_result(result)?;
      Ok(())
  }
  
  // bind integer parameter
  fn bind_int(
      stmt: *mut ffi::sqlite3_stmt,
      col_idx: c_int,
      n: usize,
  ) -> Result<()> {
      let result = unsafe { ffi::sqlite3_bind_int(stmt, col_idx, n as c_int) };
      check_result(result)
  }
  
  // bind EID parameter
  fn bind_id(
      stmt: *mut ffi::sqlite3_stmt,
      col_idx: c_int,
      id_str: &CStr,
  ) -> Result<()> {
      let result = unsafe {
          ffi::sqlite3_bind_text(
              stmt,
              col_idx,
              id_str.as_ptr(),
              -1,
              ffi::SQLITE_STATIC(),
          )
      };
      check_result(result)
  }
  
  // bind blob parameter
  fn bind_blob(
      stmt: *mut ffi::sqlite3_stmt,
      col_idx: c_int,
      data: &[u8],
  ) -> Result<()> {
      let result = unsafe {
          ffi::sqlite3_bind_blob(
              stmt,
              col_idx,
              data.as_ptr() as *const c_void,
              data.len() as c_int,
              ffi::SQLITE_STATIC(),
          )
      };
      check_result(result)
  }
  
  // run DML statement, such as INSERT and DELETE
  fn run_dml(stmt: *mut ffi::sqlite3_stmt) -> Result<()> {
      let result = unsafe { ffi::sqlite3_step(stmt) };
      match result {
          ffi::SQLITE_DONE => Ok(()),
          _ => Err(Error::from(ffi::Error::new(result))),
      }
  }
  
  // run SELECT statement on a blob column
  fn run_select_blob(stmt: *mut ffi::sqlite3_stmt) -> Result<Vec<u8>> {
      let result = unsafe { ffi::sqlite3_step(stmt) };
      match result {
          ffi::SQLITE_ROW => {
              //  get data and data size
              let (data, data_len) = unsafe {
                  (
                      ffi::sqlite3_column_blob(stmt, 0),
                      ffi::sqlite3_column_bytes(stmt, 0) as usize,
                  )
              };
  
              // copy data to vec and return it
              let mut ret = vec![0u8; data_len];
              unsafe {
                  ptr::copy_nonoverlapping(
                      data,
                      (&mut ret).as_mut_ptr() as *mut c_void,
                      data_len,
                  );
              }
              Ok(ret)
          }
          ffi::SQLITE_DONE => Err(Error::NotFound),
          _ => Err(Error::from(ffi::Error::new(result))),
      }
  }
  
  /// Sqlite Storage
  pub struct SqliteStorage {
      is_attached: bool,  // attached to sqlite db
      file_path: CString, // database file path
      db: *mut ffi::sqlite3,
      stmts: Vec<*mut ffi::sqlite3_stmt>,
  }
  
  impl SqliteStorage {
      // table name constants
      const TBL_REPO_LOCK: &'static str = "repo_lock";
      const TBL_SUPER_BLOCK: &'static str = "super_block";
      const TBL_WALS: &'static str = "wals";
      const TBL_ADDRESSES: &'static str = "addresses";
      const TBL_BLOCKS: &'static str = "blocks";
  
      pub fn new(file_path: &str) -> Self {
          SqliteStorage {
              is_attached: false,
              file_path: CString::new(file_path).unwrap(),
              db: ptr::null_mut(),
              stmts: Vec::with_capacity(14),
          }
      }
  
      // prepare one sql statement
      fn prepare_sql(&mut self, sql: String) -> Result<()> {
          let mut stmt = ptr::null_mut();
          let sql = CString::new(sql).unwrap();
          let result = unsafe {
              ffi::sqlite3_prepare_v2(
                  self.db,
                  sql.as_ptr(),
                  -1,
                  &mut stmt,
                  ptr::null_mut(),
              )
          };
          check_result(result)?;
          self.stmts.push(stmt);
          Ok(())
      }
  
      // prepare and cache all sql statements
      fn prepare_stmts(&mut self) -> Result<()> {
          // check if all statements are prepared
          if self.stmts.len() == 14 {
              return Ok(());
          }
  
          self.stmts.clear();
  
          // repo lock sql
          self.prepare_sql(format!(
              "
              SELECT lock FROM {} WHERE lock = 1
          ",
              Self::TBL_REPO_LOCK
          ))?;
          self.prepare_sql(format!(
              "
              INSERT OR REPLACE INTO {}(lock) VALUES (1)
          ",
              Self::TBL_REPO_LOCK
          ))?;
          self.prepare_sql(format!(
              "
              DELETE FROM {} WHERE lock = 1
          ",
              Self::TBL_REPO_LOCK
          ))?;
  
          // super block sql
          self.prepare_sql(format!(
              "
              SELECT data FROM {} WHERE suffix = ?
          ",
              Self::TBL_SUPER_BLOCK
          ))?;
          self.prepare_sql(format!(
              "
              INSERT OR REPLACE INTO {}(suffix, data) VALUES (?, ?)
          ",
              Self::TBL_SUPER_BLOCK
          ))?;
  
          // wal sql
          self.prepare_sql(format!(
              "
              SELECT data FROM {} WHERE id = ?
          ",
              Self::TBL_WALS
          ))?;
          self.prepare_sql(format!(
              "
              INSERT OR REPLACE INTO {}(id, data) VALUES (?, ?)
          ",
              Self::TBL_WALS
          ))?;
          self.prepare_sql(format!(
              "
              DELETE FROM {} WHERE id = ?
          ",
              Self::TBL_WALS
          ))?;
  
          // addresses sql
          self.prepare_sql(format!(
              "
              SELECT data FROM {} WHERE id = ?
          ",
              Self::TBL_ADDRESSES
          ))?;
          self.prepare_sql(format!(
              "
              INSERT OR REPLACE INTO {}(id, data) VALUES (?, ?)
          ",
              Self::TBL_ADDRESSES
          ))?;
          self.prepare_sql(format!(
              "
              DELETE FROM {} WHERE id = ?
          ",
              Self::TBL_ADDRESSES
          ))?;
  
          // blocks sql
          self.prepare_sql(format!(
              "
              SELECT data FROM {} WHERE blk_idx = ?
          ",
              Self::TBL_BLOCKS
          ))?;
          self.prepare_sql(format!(
              "
              INSERT INTO {}(blk_idx, data) VALUES (?, ?)
          ",
              Self::TBL_BLOCKS
          ))?;
          self.prepare_sql(format!(
              "
              DELETE FROM {} WHERE blk_idx = ?
          ",
              Self::TBL_BLOCKS
          ))?;
  
          Ok(())
      }
  
      fn lock_repo(&mut self, force: bool) -> Result<()> {
          let stmt = self.stmts[0];
          reset_stmt(stmt)?;
          let result = unsafe { ffi::sqlite3_step(stmt) };
          match result {
              ffi::SQLITE_ROW => {
                  // repo is locked
                  if force {
                      warn!("Repo was locked, forced to open");
                      self.is_attached = true;
                      Ok(())
                  } else {
                      Err(Error::RepoOpened)
                  }
              }
              ffi::SQLITE_DONE => {
                  // repo is not locked yet, lock it now
                  let stmt = self.stmts[1];
                  reset_stmt(stmt)?;
                  run_dml(stmt)?;
                  self.is_attached = true;
                  Ok(())
              }
              _ => Err(Error::from(ffi::Error::new(result))),
          }
      }
  }
  
  impl Storable for SqliteStorage {
      fn exists(&self) -> Result<bool> {
          let mut db: *mut ffi::sqlite3 = ptr::null_mut();
          let result = unsafe {
              ffi::sqlite3_open_v2(
                  self.file_path.as_ptr(),
                  &mut db,
                  ffi::SQLITE_OPEN_READONLY,
                  ptr::null(),
              )
          };
          if !db.is_null() {
              unsafe { ffi::sqlite3_close(db) };
          }
          Ok(result == ffi::SQLITE_OK)
      }
  
      fn connect(&mut self, _force: bool) -> Result<()> {
          let result = unsafe {
              ffi::sqlite3_open_v2(
                  self.file_path.as_ptr(),
                  &mut self.db,
                  ffi::SQLITE_OPEN_READWRITE
                      | ffi::SQLITE_OPEN_CREATE
                      | ffi::SQLITE_OPEN_FULLMUTEX,
                  ptr::null(),
              )
          };
          if result != ffi::SQLITE_OK {
              let err = ffi::Error::new(result);
              if !self.db.is_null() {
                  unsafe { ffi::sqlite3_close(self.db) };
                  self.db = ptr::null_mut();
              }
              return Err(Error::from(err));
          }
  
          Ok(())
      }
  
      fn init(&mut self, _crypto: Crypto, _key: Key) -> Result<()> {
          // create tables
          let sql = format!(
              "
              CREATE TABLE {} (
                  lock        INTEGER
              );
              CREATE TABLE {} (
                  suffix      INTEGER PRIMARY KEY,
                  data        BLOB
              );
              CREATE TABLE {} (
                  id          TEXT PRIMARY KEY,
                  data        BLOB
              );
              CREATE TABLE {} (
                  id          TEXT PRIMARY KEY,
                  data        BLOB
              );
              CREATE TABLE {} (
                  blk_idx     INTEGER PRIMARY KEY,
                  data        BLOB
              );
          ",
              Self::TBL_REPO_LOCK,
              Self::TBL_SUPER_BLOCK,
              Self::TBL_WALS,
              Self::TBL_ADDRESSES,
              Self::TBL_BLOCKS
          );
          let sql = CString::new(sql).unwrap();
          let result = unsafe {
              ffi::sqlite3_exec(
                  self.db,
                  sql.as_ptr(),
                  None,
                  ptr::null_mut(),
                  ptr::null_mut(),
              )
          };
          check_result(result)?;
  
          self.prepare_stmts()?;
          self.lock_repo(false)
      }
  
      #[inline]
      fn open(&mut self, _crypto: Crypto, _key: Key, force: bool) -> Result<()> {
          self.prepare_stmts()?;
          self.lock_repo(force)
      }
  
      fn get_super_block(&mut self, suffix: u64) -> Result<Vec<u8>> {
          // prepare statements
          self.prepare_stmts()?;
  
          let stmt = self.stmts[3];
          reset_stmt(stmt)?;
  
          // bind parameters and run sql
          bind_int(stmt, 1, suffix as usize)?;
          run_select_blob(stmt)
      }
  
      fn put_super_block(&mut self, super_blk: &[u8], suffix: u64) -> Result<()> {
          let stmt = self.stmts[4];
          reset_stmt(stmt)?;
  
          // bind parameters and run sql
          bind_int(stmt, 1, suffix as usize)?;
          bind_blob(stmt, 2, super_blk)?;
          run_dml(stmt)
      }
  
      fn get_wal(&mut self, id: &Eid) -> Result<Vec<u8>> {
          let stmt = self.stmts[5];
          reset_stmt(stmt)?;
  
          // bind parameters and run sql
          let id_str = CString::new(id.to_string()).unwrap();
          bind_id(stmt, 1, &id_str)?;
          run_select_blob(stmt)
      }
  
      fn put_wal(&mut self, id: &Eid, wal: &[u8]) -> Result<()> {
          let stmt = self.stmts[6];
          reset_stmt(stmt)?;
  
          // bind parameters and run sql
          let id_str = CString::new(id.to_string()).unwrap();
          bind_id(stmt, 1, &id_str)?;
          bind_blob(stmt, 2, wal)?;
          run_dml(stmt)
      }
  
      fn del_wal(&mut self, id: &Eid) -> Result<()> {
          let stmt = self.stmts[7];
          reset_stmt(stmt)?;
  
          // bind parameters and run sql
          let id_str = CString::new(id.to_string()).unwrap();
          bind_id(stmt, 1, &id_str)?;
          run_dml(stmt)
      }
  
      fn get_address(&mut self, id: &Eid) -> Result<Vec<u8>> {
          let stmt = self.stmts[8];
          reset_stmt(stmt)?;
  
          // bind parameters and run sql
          let id_str = CString::new(id.to_string()).unwrap();
          bind_id(stmt, 1, &id_str)?;
          run_select_blob(stmt)
      }
  
      fn put_address(&mut self, id: &Eid, addr: &[u8]) -> Result<()> {
          let stmt = self.stmts[9];
          reset_stmt(stmt)?;
  
          // bind parameters and run sql
          let id_str = CString::new(id.to_string()).unwrap();
          bind_id(stmt, 1, &id_str)?;
          bind_blob(stmt, 2, addr)?;
          run_dml(stmt)
      }
  
      fn del_address(&mut self, id: &Eid) -> Result<()> {
          let stmt = self.stmts[10];
          reset_stmt(stmt)?;
  
          // bind parameters and run sql
          let id_str = CString::new(id.to_string()).unwrap();
          bind_id(stmt, 1, &id_str)?;
          run_dml(stmt)
      }
  
      fn get_blocks(&mut self, dst: &mut [u8], span: Span) -> Result<()> {
          let stmt = self.stmts[11];
  
          let mut read = 0;
          for blk_idx in span {
              // reset statement and binding
              reset_stmt(stmt)?;
  
              // bind parameters and run sql
              bind_int(stmt, 1, blk_idx)?;
              let blk = run_select_blob(stmt)?;
              assert_eq!(blk.len(), BLK_SIZE);
              dst[read..read + BLK_SIZE].copy_from_slice(&blk);
              read += BLK_SIZE;
          }
  
          Ok(())
      }
  
      fn put_blocks(&mut self, span: Span, mut blks: &[u8]) -> Result<()> {
          let stmt = self.stmts[12];
  
          for blk_idx in span {
              // reset statement and binding
              reset_stmt(stmt)?;
  
              // bind parameters and run sql
              bind_int(stmt, 1, blk_idx)?;
              bind_blob(stmt, 2, &blks[..BLK_SIZE])?;
              run_dml(stmt)?;
  
              blks = &blks[BLK_SIZE..];
          }
  
          Ok(())
      }
  
      fn del_blocks(&mut self, span: Span) -> Result<()> {
          let stmt = self.stmts[13];
  
          for blk_idx in span {
              // reset statement and binding
              reset_stmt(stmt)?;
  
              // bind parameters and run sql
              bind_int(stmt, 1, blk_idx)?;
              run_dml(stmt)?;
          }
  
          Ok(())
      }
  
      #[inline]
      fn flush(&mut self) -> Result<()> {
          Ok(())
      }
  
      #[inline]
      fn destroy(&mut self) -> Result<()> {
          self.connect(false)?;
          if self.prepare_stmts().is_ok() {
              let stmt = self.stmts[0];
              reset_stmt(stmt)?;
              if let ffi::SQLITE_ROW = unsafe { ffi::sqlite3_step(stmt) } {
                  // repo is locked
                  warn!("Destroy an opened repo");
              }
          }
          vio::remove_file(self.file_path.to_str().unwrap())?;
          Ok(())
      }
  }
  
  impl Drop for SqliteStorage {
      fn drop(&mut self) {
          // release repo lock and ignore the result
          if self.is_attached {
              let stmt = self.stmts[2];
              let _ = reset_stmt(stmt).and_then(|_| unsafe {
                  ffi::sqlite3_step(stmt);
                  Ok(())
              });
              self.is_attached = false;
          }
  
          // release statements
          unsafe {
              for stmt in self.stmts.iter() {
                  ffi::sqlite3_finalize(*stmt);
              }
          }
  
          // close db connection
          let result = unsafe { ffi::sqlite3_close(self.db) };
          if result != ffi::SQLITE_OK {
              if panicking() {
                  eprintln!("Error while closing SQLite connection: {}", result);
              } else {
                  panic!("Error while closing SQLite connection: {}", result);
              }
          }
      }
  }
  
  impl Debug for SqliteStorage {
      fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
          f.debug_struct("SqliteStorage")
              .field("file_path", &self.file_path)
              .finish()
      }
  }
  
  unsafe impl Send for SqliteStorage {}
  unsafe impl Sync for SqliteStorage {}
  
  #[cfg(test)]
  mod tests {
      extern crate tempdir;
  
      use self::tempdir::TempDir;
  
      use super::*;
  
      use base::init_env;
  
      #[test]
      fn sqlite_storage() {
          init_env();
          let tmpdir = TempDir::new("zbox_test").expect("Create temp dir failed");
          let dir = tmpdir.path().join("storage.db");
          let mut ss = SqliteStorage::new(dir.to_str().unwrap());
  
          ss.connect(false).unwrap();
          ss.init(Crypto::default(), Key::new_empty()).unwrap();
  
          let id = Eid::new();
          let buf = vec![1, 2, 3];
          let blks = vec![42u8; BLK_SIZE * 3];
          let mut dst = vec![0u8; BLK_SIZE * 3];
  
          // super block
          ss.put_super_block(&buf, 0).unwrap();
          let s = ss.get_super_block(0).unwrap();
          assert_eq!(&s[..], &buf[..]);
  
          // wal
          ss.put_wal(&id, &buf).unwrap();
          let s = ss.get_wal(&id).unwrap();
          assert_eq!(&s[..], &buf[..]);
          ss.del_wal(&id).unwrap();
          assert_eq!(ss.get_wal(&id).unwrap_err(), Error::NotFound);
  
          // address
          ss.put_address(&id, &buf).unwrap();
          let s = ss.get_address(&id).unwrap();
          assert_eq!(&s[..], &buf[..]);
          ss.del_address(&id).unwrap();
          assert_eq!(ss.get_address(&id).unwrap_err(), Error::NotFound);
  
          // block
          let span = Span::new(0, 3);
          ss.put_blocks(span, &blks).unwrap();
          ss.get_blocks(&mut dst, span).unwrap();
          assert_eq!(&dst[..], &blks[..]);
          ss.del_blocks(Span::new(1, 2)).unwrap();
          assert_eq!(ss.get_blocks(&mut dst, span).unwrap_err(), Error::NotFound);
          assert_eq!(
              ss.get_blocks(&mut dst[..BLK_SIZE], Span::new(1, 1))
                  .unwrap_err(),
              Error::NotFound
          );
          assert_eq!(
              ss.get_blocks(&mut dst[..BLK_SIZE], Span::new(2, 1))
                  .unwrap_err(),
              Error::NotFound
          );
  
          // re-open
          drop(ss);
          let mut ss = SqliteStorage::new(dir.to_str().unwrap());
          ss.connect(false).unwrap();
          ss.open(Crypto::default(), Key::new_empty(), false).unwrap();
  
          ss.get_blocks(&mut dst[..BLK_SIZE], Span::new(0, 1))
              .unwrap();
          assert_eq!(&dst[..BLK_SIZE], &blks[..BLK_SIZE]);
          assert_eq!(
              ss.get_blocks(&mut dst[..BLK_SIZE], Span::new(1, 1))
                  .unwrap_err(),
              Error::NotFound
          );
          assert_eq!(
              ss.get_blocks(&mut dst[..BLK_SIZE], Span::new(2, 1))
                  .unwrap_err(),
              Error::NotFound
          );
      }
  }
  