path: root/server/src/pracrodaopgsql.cc

/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim: set et sw=2 ts=2: */
/***************************************************************************
 *            pracrodaopgsql.cc
 *
 *  Wed Feb 11 11:18:26 CET 2009
 *  Copyright 2009 Bent Bisballe Nyeng
 *  deva@aasimon.org
 ****************************************************************************/

/*
 *  This file is part of Pracro.
 *
 *  Pracro is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 2 of the License, or
 *  (at your option) any later version.
 *
 *  Pracro is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with Pracro; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
 */
#include "pracrodaopgsql.h"

/*
 * Updating the old tables;
 *
 * ALTER TABLE transactions ADD COLUMN uid bigint;
 * CREATE SEQUENCE 'trseq';
 * SELECT setval('trseq', (SELECT MAX(oid) FROM transactions));
 * UPDATE transactions SET uid = oid;
 * INSERT INTO fieldnames (name, description, timestamp) 
 *   VALUES ('journal.resume', 'Journal resume text', 
 *   (SELECT EXTRACT(EPOCH FROM now())::integer));
 */
#include <config.h>

#ifndef WITHOUT_DB

#include <stdlib.h>

#include "debug.h"

PracroDAOPgsql::PracroDAOPgsql(std::string _host, std::string _port,
                               std::string _user, std::string _passwd,
                               std::string _dbname)
  : PracroDAO(_host, _port, _user, _passwd, _dbname)
{
  conn = NULL;
  std::string cs;
  if(host.size()) cs += " host=" + host;
  if(port.size()) cs += " port=" + port;
  if(user.size()) cs += " user=" + user;
  if(passwd.size()) cs += " password=" + passwd;
  cs += " dbname=" + (dbname.size() ? dbname : "pracro");
  try {
    conn = new pqxx::connection(cs);
    /*
    std::string ts;
    try {
      ts = "BEGIN;";
      DEBUG(sql, "Query: %s\n", ts.c_str());
      pqxx::result R = W->exec(ts);
    } catch(...) {
    }
    */
  } catch(std::exception &e) {
    ERR_LOG(db, "Postgresql init failed: %s\n", e.what());
    conn = NULL;
  }

  DEBUG(db, "Pgsql connection %p (%s)\n", conn, cs.c_str());
}

PracroDAOPgsql::~PracroDAOPgsql()
{
  if(conn) {
    //    if(W) delete W;
    delete conn;
  }
}

std::string PracroDAOPgsql::newSessionId()
{
  if(!conn) {
    ERR(db, "No pgsql connection\n");
    return "";
  }

  pqxx::work W(*conn);
  pqxx::result R = W.exec("SELECT nextval('sessionseq');");
  pqxx::result::const_iterator ri = R.begin();
  if(ri != R.end()) {
    DEBUG(db, "New session id: %s\n", (*ri)[0].c_str());
    return (*ri)[0].c_str();
  }
  ERR(db, "No pgsql connection\n");
  return "";
}

void PracroDAOPgsql::commitTransaction(std::string sessionid,
                                       Transaction &transaction,
                                       Commit &commit,
                                       Macro &_macro,
                                       time_t now)
{
  DEBUG(db, "commitTransaction (%s, %s, %s, <%u fields>, %ld)\n",
        transaction.user.c_str(), transaction.cpr.c_str(),
        _macro.attributes["name"].c_str(),
        commit.fields.size(), now);
  
  if(!conn) {
    ERR(db, "No pgsql connection\n");
    return;
  }

  if(commit.fields.size() == 0) return;

  pqxx::work W(*conn);

  std::string version = _macro.attributes["version"];
  std::string macro = _macro.attributes["name"];
  std::stringstream timestamp; timestamp << now;

  std::string ts;

  ts = "SELECT status FROM commits WHERE uid='"+sessionid+"';";
  pqxx::result R = W.exec(ts);
  if(!R.size()) {
    ts = "INSERT INTO commits (patientid, template, version,"
      " \"timestamp\", uid, status) VALUES ("
      " '" + W.esc(transaction.cpr) + "', "
      " '" + W.esc(commit.templ) + "', "
      " '" + "1.0" + "', "
      " '" + W.esc(timestamp.str()) + "', "
      " '" + W.esc(sessionid) + "', "
      " 'active' "
      ");"
      ;
    DEBUG(sql, "Query: %s\n", ts.c_str());
    pqxx::result R = W.exec(ts);
  } else {

    pqxx::result::const_iterator ri = R.begin();
    if(ri != R.end()) {
      std::string status = (*ri)[0].c_str();
      if(status == "committed") {
        ERR_LOG(db, "Attempt to add to committed session %s blocked!\n",
                sessionid.c_str());
        return;
      }
    }

    ts = "UPDATE commits SET status='active' WHERE uid="+sessionid+";";
    DEBUG(sql, "Query: %s\n", ts.c_str());
    /*pqxx::result R = */W.exec(ts);
  }

  try {
    ts = "INSERT INTO transactions (uid, macro, version,"
      " \"timestamp\", \"user\", cid) VALUES ("
      " nextval('trseq'), "
      " '" + W.esc(macro) + "', "
      " '" + W.esc(version) + "', "
      " '" + W.esc(timestamp.str()) + "', "
      " '" + W.esc(transaction.user) + "', "
      " '" + W.esc(sessionid) + "' "
      ");"
      ;
    DEBUG(sql, "Query: %s\n", ts.c_str());
    pqxx::result R = W.exec(ts);

    if(commit.fields.size() > 0) {
      // field table lookup
      ts = "SELECT DISTINCT name FROM fieldnames WHERE name IN ( ";
      std::map< std::string, std::string >::iterator i = commit.fields.begin();
      ts += "'" + W.esc(i->first) + "'";
      i++;
      while(i != commit.fields.end()) {
        ts += ", '" + W.esc(i->first) + "'";
        i++;
      }
      ts += ");";
      DEBUG(sql, "Query: %s\n", ts.c_str());
      R = W.exec(ts);

      DEBUG(db, "input fields: %d, output fields: %lu\n",
            commit.fields.size(), R.size());

      // Store known fields
      pqxx::result::const_iterator ri = R.begin();
      if(ri != R.end()) {
        std::string name = (*ri)[0].c_str();
        DEBUG(db, "Storing: %s with value %s\n",
              name.c_str(), commit.fields[name].c_str());
        ts = "INSERT INTO fields (transaction, name, value) "
          "VALUES ( currval('trseq'), '" + W.esc(name) + "', '" +
          W.esc(commit.fields[name]) + "')";
        ri++;
        while(ri != R.end()) {
          name = (*ri)[0].c_str();

          DEBUG(db, "Storing: %s with value %s\n",
                name.c_str(), commit.fields[name].c_str());

          ts += ", (currval('trseq'), '" + W.esc(name) + "', '" +
            W.esc(commit.fields[name]) + "')";
          ri++;
        }
        ts += ";";
        DEBUG(sql, "Query: %s\n", ts.c_str());
        W.exec(ts);
      }
    }

    W.commit();

  } catch(std::exception &e) {
    ERR_LOG(db, "Query failed: %s: %s\n", e.what(), ts.c_str());
  }

}


/*
 * The following select finds the newest timestamps for each fieldname
 * belonging to the designated patient and from a set of fieldnames in the
 * inner query. The outer query then finds the corresponding field values.
 * Note: there is no protection agains duplicate fields that could
 * theoretically occur within the same transaction.
 *
 * SELECT ff.name, ff.value, tt.timestamp FROM (
 *    SELECT f.name, max(t.timestamp) AS ts
 *      FROM fields f, transactions t
 *      WHERE t.uid = f.transaction
 *        AND t.patientid = '1505050505'
 *        AND t.timestamp >= 0
 *        AND f.name IN ('current_eye_disease')
 *      GROUP BY f.name) xx,
 *  transactions tt, fields ff
 *  WHERE xx.ts = tt.timestamp
 *    AND xx.name = ff.name
 *    AND tt.uid = ff.transaction
 *    AND tt.patientid = '1505050505'
 */
Values PracroDAOPgsql::getLatestValues(std::string sessionid,
                                       std::string patientid,
                                       Macro *macro,
                                       Fieldnames &fieldnames,
                                       time_t oldest)
{

  bool uncom = false; // get results that are not yet committed?

  DEBUG(db, "(%s, %s, <%u fieldnames>, %ld)\n",
        patientid.c_str(),
        macro ? macro->attributes["name"].c_str() : "(null)",
        fieldnames.size(), oldest);
  if(!conn) DEBUG(db, "No pgsql connection\n");
  Values values;

  std::string query;
  std::stringstream soldest; soldest << oldest;
  try {
    {
      pqxx::work W(*conn);
      query = "UPDATE commits SET status='active' WHERE status='idle'"
        " AND uid="+sessionid+";";
      DEBUG(sql, "Query: %s\n", query.c_str());
      /*pqxx::result R = */W.exec(query);
      W.commit();
    }

    pqxx::work W(*conn);
    std::string namecond;

    if(fieldnames.size() > 0) {
      std::vector< std::string >::iterator i = fieldnames.begin();
      namecond += " AND f.name IN ('" + W.esc(*i) + "'";
      i++;
      while(i != fieldnames.end()) {
        namecond += ", '" + W.esc(*i) + "'";
        i++;
      }
      namecond += ')';
    }
    query = "SELECT ff.name, ff.value, tt.timestamp FROM "
    // Begin inner query
      " (SELECT f.name, MAX(t.timestamp) AS ts "
      "   FROM commits c, fields f, transactions t "
      "   WHERE ";
    if(!uncom) query += "(c.status='committed' OR c.uid="+sessionid+") AND ";
    query += "c.uid = t.cid AND t.uid = f.transaction"
      " AND t.timestamp >= " + soldest.str() +
      " AND c.patientid = '" + W.esc(patientid) + "' "
      + namecond;
    if(macro) {
      query += " AND t.macro = '" + macro->attributes["name"] + "'";
      if(macro->attributes["version"].size() > 0)
        query += " AND t.version = '" + macro->attributes["version"] + "'";
    }
    query += " GROUP BY f.name) xx, "
    // End inner query
      " transactions tt, fields ff, commits cc "
      " WHERE ";
    if(!uncom) query += "(cc.status='committed' OR cc.uid="+sessionid+") AND ";
    query += " xx.ts = tt.timestamp "
      "   AND xx.name = ff.name "
      "   AND tt.uid = ff.transaction "
      "   AND tt.cid = cc.uid "
      "   AND cc.patientid = '" + W.esc(patientid) + "' "
      ;
    if(macro) {
      query += " AND tt.macro = '" + macro->attributes["name"] + "'";
      if(macro->attributes["version"].size() > 0)
        query += " AND tt.version = '" + macro->attributes["version"] + "'";
    }

    DEBUG(sql, "Query: %s\n", query.c_str());
    pqxx::result R = W.exec(query);
    pqxx::result::const_iterator ri = R.begin();
    while(ri != R.end()) {
      Value v;
      v.value = (*ri)[1].c_str();
      v.timestamp = atol((*ri)[2].c_str());
      values[(*ri)[0].c_str()] = v;
      ri++;
    }
  } catch (std::exception &e) {
    ERR_LOG(db, "Query failed: %s: %s\n", e.what(), query.c_str());
  }

  return values;
}


unsigned PracroDAOPgsql::nrOfCommits(std::string sessionid,
                                     std::string patientid,
                                     std::string macroname,
                                     time_t oldest)
{

  bool uncom = false; // get results that are not yet committed?

  std::string query;
  std::stringstream soldest; soldest << oldest;
  try {
    pqxx::work W(*conn);
    query = "SELECT count(*) FROM commits c, transactions f"
      " WHERE c.patientid = '" + W.esc(patientid) + "' AND c.uid = f.cid";
    if(!uncom) query += " AND (c.status='committed' OR c.uid="+sessionid+")";
    query += " AND f.macro = '" + W.esc(macroname) + "' "
      " AND f.timestamp >= " + soldest.str()
      ;
    DEBUG(sql, "Query: %s\n", query.c_str());
    pqxx::result R = W.exec(query);
    if(R.size() != 1) {
      ERR_LOG(db, "No result set; expected one row with one column\n");
      return 0;
    }
    unsigned n = (unsigned)atol((*R.begin())[0].c_str());
    DEBUG(db, "Found %u commits for %s(%s) from %ld\n",
                 n, patientid.c_str(), macroname.c_str(), oldest);
    return n;
  } catch (std::exception &e) {
    ERR_LOG(db, "Query failed: %s: %s\n", e.what(), query.c_str());
  }

  return 0;
}

void PracroDAOPgsql::addFieldname(std::string name, std::string description)
{
  std::stringstream timestamp; timestamp << time(NULL);
  std::string ts;
  try {
    pqxx::work W(*conn);
    ts = "INSERT INTO fieldnames (name, description, \"timestamp\") VALUES ("
      " '" + W.esc(name) + "', "
      " '" + W.esc(description) + "', "
      " '" + W.esc(timestamp.str()) + "' "
      ")"
      ;
    DEBUG(sql, "Query: %s\n", ts.c_str());
    pqxx::result R = W.exec(ts);
    W.commit();
  } catch (std::exception &e) {
    ERR_LOG(db, "Query failed: %s: %s\n", e.what(), ts.c_str());
  }
}

void PracroDAOPgsql::delFieldname(std::string name)
{
  std::string ts;
  try {
    pqxx::work W(*conn);
    ts = "DELETE FROM fieldnames WHERE name="
      "'" + W.esc(name) + "' ";
    DEBUG(sql, "Query: %s\n", ts.c_str());
    pqxx::result R = W.exec(ts);
    W.commit();
  } catch (std::exception &e) {
    ERR_LOG(db, "Query failed: %s: %s\n", e.what(), ts.c_str());
  }
}

std::vector<Fieldname> PracroDAOPgsql::getFieldnames()
{
  if(!conn) DEBUG(db, "No pgsql connection\n");
  std::vector<Fieldname> fieldnames;

  std::string query;
  try {
    pqxx::work W(*conn);
    query = "SELECT * FROM fieldnames";
    DEBUG(sql, "Query: %s\n", query.c_str());
    pqxx::result R = W.exec(query);
    pqxx::result::const_iterator ri = R.begin();
    while(ri != R.end()) {
      Fieldname f;
      f.name = (*ri)[0].c_str();
      f.description = (*ri)[1].c_str();
      f.timestamp = atol((*ri)[2].c_str());
      fieldnames.push_back(f);
      ri++;
    }
  } catch (std::exception &e) {
    ERR_LOG(db, "Query failed: %s: %s\n", e.what(), query.c_str());
  }

  return fieldnames;
}

void PracroDAOPgsql::commit(std::string sessionid)
{
  std::string ts;
  try {
    pqxx::work W(*conn);
    ts = "UPDATE commits SET status='committed' WHERE uid="+sessionid+";";
    /*pqxx::result R = */W.exec(ts);
    
    W.commit();
  } catch (std::exception &e) {
    ERR_LOG(db, "Commit failed: %s: %s\n", e.what(), ts.c_str());
  }
}

void PracroDAOPgsql::nocommit(std::string sessionid)
{
  std::string ts;
  try {
    pqxx::work W(*conn);
    ts = "UPDATE commits SET status='idle' WHERE uid="+sessionid+";";
    /*pqxx::result R = */W.exec(ts);
    
    W.commit();
  } catch (std::exception &e) {
    ERR_LOG(db, "NoCommit failed: %s: %s\n", e.what(), ts.c_str());
  }
}

void PracroDAOPgsql::discard(std::string sessionid)
{
  std::string ts;
  try {
    pqxx::work W(*conn);
    ts = "DELETE FROM commits WHERE uid="+sessionid+";";
    /*pqxx::result R = */W.exec(ts);
    W.commit();
  } catch (std::exception &e) {
    ERR_LOG(db, "Abort (rollback) failed: %s: %s\n", e.what(), ts.c_str());
  }
}

#endif/*WITHOUT_DB*/

#ifdef TEST_PRACRODAOPGSQL

#include "configuration.h"
#include "exception.h"

int main()
{
#ifndef WITHOUT_DB
  try {
    PracroDAOPgsql db(Conf::database_addr, "", Conf::database_user,
                      Conf::database_passwd, "");
  } catch(Exception &e) {
    printf("ERROR: %s\n", e.what());
    return 1;
  }
#endif/*WITHOUT_DB*/
  return 0;
}
#endif/*TEST_PRACRODAOPGSQL*/