[tpop3d-discuss]Early proof of concept for Pgsql mailbox driver

[Chris Travers]

Disclaimer:  I am not even sure if this will compile yet.  It is merely 
offered for feedback purposes.  My C skills are still under active 
development, as well, and it may be some time before this is 
production-ready.  That and my time is currently limited.

This also assumes that you move a few of the auth_pgsql utility functions into 
another module called util_pgsql.

Should be pretty straight-forward to follow.  If anyone has any feedback, I 
would love to hear it.

/*
 *   mailpgsql.c"
 *   Mailboxes from a PostgreSQL
 *
 *   Copyright (c) 2004 Chris Travers
 *   All rights reserved except as under the GNU General Public License (GPL)
 *
 *   address@hidden
 *
 *  Notes:  Currently, rather than implement the complex interface found in 
 *  auth_pgsql.c, I have opted to use a more flexible, if obtuse, structure 
and
 *  require that the connection string be specified in the tpop3d.conf file.
 *  The auth_pgsql interface may be supported at a later time.  Also, I will
 *  likely patch auth_pgsql to allow it to use the pg_cnx_string configuration
 *  parameter as well, but this has not happened yet.
 *  
 */

#ifdef HAVE_CONFIG_H
#include "configuration.h"
#endif /* HAVE_CONFIG_H */

#ifdef MBOX_PGSQL
#include <stdio.h>
#include <errno.h>
#include <syslog.h>
#include <unistd.h>
#include <utime.h>
#include <time.h>
#include <stdlib.h>

#include "mailbox.h"
#include "connection.h"
#include "config.h"
#include "util.h"
#include "vector.h"
#include "libpq-fe.h"

/* Internal Status Codes:
 *
 * Note that all internal functions returning an int and wrapping database ops
 * return one of these unless they are returning data.
 */

#define MB_OK   0
#define MBERR_BAD_CNX   1
#define MBERR_DB_OP_FAILED      2
#define MBERR_NO_MBOX   4


/* Definitions for Postgresql data type lengths in text */
#define PG_LEN_BIGINT   18

/* Setting up the query templates for a dedicated schema.  Should be easy to 
 * implement using views for any database with sufficient information.
 */

char *mailpgsql_mailbox_query_template = 
        "SELECT message_id, message_header, message_body, "
        "char_length(message_header || '\n' || message_body) as message_length "
        "FROM tpop3d.messages "
        "WHERE deliver_to = $1";

char *mailpgsql_mailbox_create_temp = 
        "CREATE TEMPORARY TABLE tpop3_maildrop "
        "(message_id BIGINT, "
        "message_headers TEXT, "
        "message_body TEXT, "
        "message_length INT) "
        "ON COMMIT DROP ";

/* pgsql_dbop_test PGRESULT
 * 
 * tests to see if the operation was successful.  Returns MB_OK (0) or
 * MBERR_DB_OP_FAILED */
function pgsql_dbop_test (PGresult *test_result){

        int retval;

        if ((PGRES_TUPLES_OK == PQresultStatus(test_result)) || 
                        (PGRES_COMMAND_OK == PQresultStatus(test_result))){
                retval = 0;
        } else {
                retval = MBERR_DB_OP_FAILED;
        }

        return retval;
}

/* pgsql_dbop_test_only PGRESULT
 *
 * Tests to see if the operation committed successfully.  If so, it frees the
 * result and returns 0.  This is needed because of the large number of 
 * database operations which do not return anything. If the op was not
 * successful, it still frees the result but returns MBERR_DB_OP_FAILED. */

int pgsql_dbop_test_only (PGresult *test_result){
        int retval;
        retval = pgsql_dbop_test(test_result);
        PQClear(test_result);

        return retval;
}
        
/* mailpgsql_lock PGCONNECTION
 *
 * Locks are implemented in mailpgsql using temporary tables.  This allows for
 * exclusive, nonblocking access to the maildrop.  In other words, exclusive
 * mailboxes are materialized on demand and do not block other attempts to
 * access the same content.  This is used to simulate REPEATABLE READ 
 * transaction isolation level (not yet implemented in PostgreSQL as of 7.4).
 *
 * Still trying to decide if this is the right way to go. */

static int mailpgsql_lock(mailbox *M){
        char *param_values[1];
        PGresult *pg_result;
        int retval = MB_OK;

        if (CONNECTION_BAD == PQstatus(M->pg_cnx)){
                return MBERR_BAD_CNX;
        }

        param_values[0] = M->name;

        if (retval = pgsql_dbop_test_only(PGExec(M->pg_cnx, "BEGIN"i))){
                goto end_func;
        }
        
        if (retval = pgsql_dbop_test_only(
                        PGExec(M->pg_cnx, mailpgsql_mailbox_create_temp)
        )){
                goto end_func;
        }
        if (retval = pgsql_dbop_test_only(PGExecParams(M->pg_cnx, 
                sprintf("INSERT INTO tpop3_maildrop %s",  
                        mailpgsql_mailbox_query_template),
                1,
                NULL,
                param_values,
                NULL,
                NULL,
                1
        ))){
                goto end_func;
        }

end_func:
        return retval;
}


/* The following 2 functions are different ways of undoing a lock.  At the
 * moment, only mailpgsql_commit will likely be used, but there may be cases
 * when an error occurs and if so, we should rollback the transaction.
 */

/* mailpgsql_commit PGCONNECTION
 * Ends a transaction block by issuing a COMMIT statement. */

static int mailpgsql_commit(PGconn *cnx){
        return pgsql_dbop_test_only(PGexec(cnx, "COMMIT"));
}

/* mailpgsql_rollback PGCONNECTION
 * Ends a transaction block by issuing a ROLLBACK statement. */

static int mailpgsql_rollback(PGconn *cnx){
        return pgsql_dbop_test_only(PGexec(cnx, "ROLLBACK"));
}

/* mailpgsql_make_indexpoint
 * makes in index point for a mailpgsql mailbox. */

static void mailpgsql_make_indexpoint(
                struct indexpoint *indexp,
                const char *messgid,
                off_t size,
                time_t mtime){
        memset(indexp, 0, sizeof(struct indexpoint));

        indexp->filename = strdup(messgid);
        if (!m->filename){
                return;
        }
        indexp->offset = 0;
        indexp->length = 0;
        indexp->deleted = 0;
        indexp->msgsize = size;
        indexp->mtime = mtime;

        md5_digest(messgid, strcspn(messgid, ":"), indexp->hash); 
}


/* mailpgsql_build_index MAILBOX, TIME
 *
 * Builds the index for the mailgpsql mailbox */

static int mailpgsql_build_index (mailbox M, time_t mtime){
        int retval = MB_OK;
        unsigned long mr_length;
        char *messgid;
        PGresult *mb_row;

        messgid = xmalloc(PG_LEN_BIGINT + 2); /* TO BE SAFE */
        if (!M){
                log_print(LOG_ERR, "mailpgsql build index failed: No MBOX: %m");
                return MBERR_NO_MBOX;
        }
        retval = pgsql_dbop_test_only(PGExec(
                "DECLARE mb_index CURSOR FOR SELECT * FROM tpop3_maildrop"
        ));
        if (retval != MB_OK){
                return retval;
        }
        while(1){
                mb_row = PGexec("FETCH mb_index");
                if (0 == PQntuples(mb_row)){
                        break;
                }
                mr_length = atoi(PQgetvalue(mbrow, 0, 3));

                sprintf(messgid, "%s", PQgetvalue(mb_row, 0, 0));
                struct indexpoint ipt;
                mailpgsql_make_indexpoint(&ipt, mr_length, NULL);
                mailbox_add_indexpoint(M, &ipt);
                M->totalsize += mr_length; 
        }
        xfree messgid;
}

/* mailpgsql_new EMAIL_ADDRESS 
 * Constructor for mailbox object.
 *
 * For mailpgsql, the mailboxes actually consist of temporary tables which 
 * are cleared when the database transaction is closed.  This is how 
exclusive,
 * non-blocking locks are achieved.  However, deleting the emails is done 
 * against the master relation.  This allows duplicate deletions to occur 
 * without error. */
mailbox mailpgsql_new_from_addr(const char *email_addr){
        mailbox M, FailM = NULL;
        char *s, cnx_string = NULL;

        if (s = config_get_string("pg-cnx-string")){
                cnx_string = s;
        } 
        
        alloc_struct(_mailbox, M);

        M->pg_cnx = PQconnectdb(cnx_string);
        M->delete = mailpgsql_delete;
        M->apply_changes = mailpgsql_apply_changes;
        M->sendmessage = mailpgsql_sendmessage;

        /* Allocate space for the index. */
        M->index = (struct indexpoint*)xcalloc(32, sizeof(struct indexpoint));
        M->size = 32;

        if (mailpgsql_lock(M) != MB_OK){ // Try to get non-blocking exclusive
                                        // lock.
                goto fail;
        }
        return M;

fail:
        if (M){
                mailpgsql_rollback(M->pg_cnx);
                if (M->name) xfree(M->name);
                if (M->index) xfree(M->index);
                xfree(M);
        }
        return failM;
}

/* mailpgsql_delete MAILBOX
 * Inserts runs a transaction rollback into the current transaction in order 
to 
 * terminate any current transaction code.  This is will result in the 
 * database-side mailbox being dropped, and then it can safely deallocate the
 * client-side object.  Note that this means that it is the responsibility of
 * the commit_changes function to commit the transaction and begin a new dummy
 * transaction. */
void mailpgsql_delete (mailbox M){
        mailpgsql_rollback(M->pg_cnx);
        mailbox_delete(M);
}

/* mailpgsql_sendmessage MAILBOX CONNECTION MSGNUM LINES
 * writes an +OK or -ERR message followed by the headers and a specified 
number
 * of LINES of the body.
 *
 * This is a complex problem, as we do NOT assume that the messages use the 
 * proper \r\n EOL handling (PostgreSQL uses \n only).  We assume that \n may
 * be used by itself, and we discount the possibility that the EOL could be 
\r.
 * We also properly escape the beginning . in any lines.  Finally, we attempt
 * to handle all this inside the query using regular expressions and replace()
 * calls, which results in a particularly complex SQL query. Returns 0 or -1. 
*/
int mailpgsql_sendmessage(mailbox M, connection c, 
                        const int msgnum, const int lines){
        struct indexpoint ipoint;
        int status;

        char *getmsgquery;
        char *repeat_clause;
        char *msg_text;
        PGresult *query_result;

        /* Template for query handles EOL errors (except for EOL being "\r" as
         * MacOS) and escaping lines which begin with . as per rfc1939.
         * Finally, we leave room for the ability to limit the resulting msg
         * body length. */
        char *getmsg_query_template = 
                "SELECT replace(replace(replace(message_header, "
                                        "'\r\n', '\n'), "
                                "'\n.', '\n..'), "
                        "'\n', '\r\n') || \r\n || "
                        "replace(replace(%sreplace(message_body, "
                                                "'\r\n', '\n') "
                                        "%s, "
                                "'\n.', '\n..'), "
                        "'\n', '\r\n') || %s AS message "
                "FROM tpop3_maildrop LIMIT 1 OFFSET %d";
        /* fill in the offset */
        sprintf (getmsg_query_template, getmsg_query_template, msgnum - 1);

        /* Try first limiting the lines returned. */
        sprintf(repeat_clause, "FROM repeat('[^\n]*\n', %d))", lines);
        sprintf(getmsgquery, getmsg_query_template, 
                        "substring(", repeat_clause, ".");

        query_result = PGexec(getmsgquery);

        msg_text = PQgetvalue(query_result, 0, 0);

        if(msg_text != NULL || *msg_text != ""){
                /* We didn't get anything.  Try without the line limit */
                sprintf(getmsgquery, getmsg_query_template, "", "", "\r\n.");
                query_result = PGexec(getmsgquery);
                msg_text = PQgetvalue(query_result, 0, 0);
        }
        /* If we still don't have a message, we have a problem */
        if(msg_text != NULL || *msg_text != ""){
                connection_sendresponse(c, 0, 
                        "Sorry, the operation failed. Please try again later.");
                return -1;
        }

        /* Now we just need to send the message to the client. */
        connection_sendline(c, msg_text);
        return 0;
}
        

#endif /* MBOX_PGSQL */