path: root/sql_db.cc

#include <functional>
#include <iomanip>
#include <iostream>
#include <sstream>
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <openssl/rand.h>
#include "misc.h"
#include "sql_db.h"

using namespace std;

const char dbvers[] = "muchsync 0";

const char muchsync_schema[] = R"(
-- General table
CREATE TABLE configuration (
  key TEXT PRIMARY KEY NOT NULL,
  value TEXT);
CREATE TABLE sync_vector (
  replica INTEGER PRIMARY KEY,
  version INTEGER);

-- Shadow copy of the Xapian database to detect changes
CREATE TABLE xapian_dirs (
  dir_path TEXT UNIQUE NOT NULL,
  dir_docid INTEGER PRIMARY KEY,
  dir_mtime INTEGER);
CREATE TABLE tags (
  tag TEXT NOT NULL,
  docid INTEGER NOT NULL,
  UNIQUE (docid, tag),
  UNIQUE (tag, docid));
CREATE TABLE message_ids (
  message_id TEXT UNIQUE NOT NULL,
  docid INTEGER PRIMARY KEY,
  replica INTEGER,
  version INTEGER);
CREATE INDEX message_ids_writestamp ON message_ids (replica, version);
CREATE TABLE xapian_files (
  dir_docid INTEGER NOT NULL,
  name TEXT NOT NULL,
  docid INTEGER,
  mtime REAL,
  inode INTEGER,
  hash_id INGEGER,
  PRIMARY KEY (dir_docid, name));
CREATE INDEX xapian_files_hash_id ON xapian_files (hash_id, dir_docid);
CREATE TABLE maildir_hashes (
  hash_id INTEGER PRIMARY KEY,
  hash TEXT UNIQUE NOT NULL,
  size INTEGER,
  message_id TEXT,
  replica INTEGER,
  version INTEGER);
CREATE INDEX maildir_hashes_message_id ON maildir_hashes (message_id);
CREATE INDEX maildir_hashes_writestamp ON maildir_hashes (replica, version);
CREATE TABLE xapian_nlinks (
  hash_id INTEGER NOT NULL,
  dir_docid INTEGER NOT NULL,
  link_count INTEGER,
  PRIMARY KEY (hash_id, dir_docid));
)";

i64
create_random_id()
{
  i64 id = 0;
  if (RAND_pseudo_bytes ((unsigned char *) &id, sizeof (id)) == -1 || id == 0) {
    cerr << "RAND_pseudo_bytes failed\n";
    return -1;
  }
  id &= ~(i64 (1) << 63);
  return id;
}

static sqlite3 *
dbcreate (const char *path)
{
  i64 self = create_random_id();
  if (self <= 0)
    return nullptr;

  sqlite3 *db = nullptr;
  int err = sqlite3_open_v2 (path, &db,
			     SQLITE_OPEN_READWRITE|SQLITE_OPEN_CREATE, nullptr);
  if (err) {
    cerr << path << ": " << sqlite3_errstr (err) << '\n';
    return nullptr;
  }
  sqlexec(db, "PRAGMA locking_mode=EXCLUSIVE;");

  try {
    sqlexec (db, "BEGIN;");
    sqlexec (db, muchsync_schema);
    setconfig (db, "dbvers", dbvers);
    setconfig (db, "self", self);
    sqlexec (db, "INSERT INTO sync_vector (replica, version)"
	     " VALUES (%lld, 1);", self);
    sqlexec (db, "COMMIT;");
  } catch (sqlerr_t exc) {
    sqlite3_close_v2 (db);
    cerr << exc.what () << '\n';
    return nullptr;
  }
  return db;
}

sqlite3 *
dbopen (const char *path, bool exclusive)
{
  sqlite3 *db = nullptr;
  if (access (path, 0) && errno == ENOENT)
    db = dbcreate (path);
  else {
    sqlite3_open_v2 (path, &db, SQLITE_OPEN_READWRITE, nullptr);
    if (exclusive)
      sqlexec(db, "PRAGMA locking_mode=EXCLUSIVE;");
  }
  if (!db)
    return nullptr;

  sqlexec (db, "PRAGMA secure_delete = 0;");

  try {
    if (getconfig<string> (db, "dbvers") != dbvers) {
      cerr << path << ": invalid database version\n";
      sqlite3_close_v2 (db);
      return nullptr;
    }
    getconfig<i64> (db, "self");
  }
  catch (sqldone_t) {
    cerr << path << ": invalid configuration\n";
    sqlite3_close_v2 (db);
    return nullptr;
  }
  catch (sqlerr_t &e) {
    cerr << path << ": " << e.what() << '\n';
    sqlite3_close_v2 (db);
    return nullptr;
  }

  return db;
}


istream &
read_writestamp (istream &in, writestamp &ws)
{
  input_match (in, 'R');
  in >> ws.first;
  input_match (in, '=');
  in >> ws.second;
  return in;
}

istream &
read_sync_vector (istream &in, versvector &vv)
{
  input_match (in, '<');
  vv.clear();
  for (;;) {
    char c;
    if ((in >> c) && c == '>')
      return in;
    in.unget();
    writestamp ws;
    if (!read_writestamp (in, ws))
      break;
    vv.insert (ws);
    if (!(in >> c) || c == '>')
      break;
    if (c != ',') {
      in.setstate (ios_base::failbit);
      break;
    }
  }
  return in;
}

string
show_sync_vector (const versvector &vv)
{
  ostringstream sb;
  sb << '<';
  bool first = true;
  for (auto ws : vv) {
    if (first)
      first = false;
    else
      sb << ",";
    sb << 'R' << ws.first << '=' << ws.second;
  }
  sb << '>';
  return sb.str();
}

string
permissive_percent_encode (const string &raw)
{
  ostringstream outbuf;
  outbuf.fill('0');
  outbuf.setf(ios::hex, ios::basefield);
  for (char c : raw)
    if (c <= ' ' || c >= '\177' || c == '%' || c == '(' || c == ')')
      outbuf << '%' << setw(2) << int (uint8_t(c));
    else
      outbuf << c;
  return outbuf.str();
}

template<typename C, typename F> inline void
intercalate (const C &c, F &&each, function<void()> between)
{
  auto i = c.begin(), end = c.end();
  if (i != end) {
    each (i);
    while (++i != end) {
      between();
      each(i);
    }
  }
}

ostream &
operator<< (ostream &os, const hash_info &hi)
{
  os << "L " << hi.hash << ' ' << hi.size << ' '
     << permissive_percent_encode(hi.message_id)
     << " R" << hi.hash_stamp.first << '=' << hi.hash_stamp.second
     << " (";
  intercalate (hi.dirs,
	       [&](decltype(hi.dirs.begin()) i) {
		 os << i->second << '*' << permissive_percent_encode(i->first);
	       },
	       [&]() {os << ' ';});
  os << ')';
  return os;
}

istream &
operator>> (istream &is, hash_info &hi)
{
  string hash, msgid;
  size_t size;
  writestamp stamp;
  decltype(hi.dirs) d;