/* 
 * Copyright (C) 2005  Network Applied Communication Laboratory Co., Ltd.
 *
 * This file is part of Rast.
 * See the file COPYING for redistribution information.
 *
 */

#include <apr_pools.h>
#include <apr_strings.h>
#include <xmlrpc.h>
#include <xmlrpc_client.h>

#include <rast/error.h>
#include <rast/local_db.h>
#include <rast/db.h>

#include "rast/xmlrpc_client.h"
#include "rast/util.h"

typedef struct {
    rast_document_t base;
    rast_buf_ring_head_t *buf_ring;
    int nbytes;
    apr_hash_t *property_values;
} rast_xmlrpc_client_document_t;

static int initialized = 0;

static rast_error_type_t
str_to_rast_error_type_t(const char *str)
{
    if (strcmp("rast", str) == 0) {
        return RAST_ERROR_TYPE_RAST;
    }
    if (strcmp("apr", str) == 0) {
        return RAST_ERROR_TYPE_APR;
    }
    if (strcmp("bdb", str) == 0) {
        return RAST_ERROR_TYPE_BDB;
    }
    if (strcmp("xmlrpc", str) == 0) {
        return RAST_ERROR_TYPE_XMLRPC;
    }
    return -1;
}

static rast_type_e
str_to_rast_type_e(const char *str)
{
    if (strcmp("string", str) == 0) {
        return RAST_TYPE_STRING;
    }
    if (strcmp("date", str) == 0) {
        return RAST_TYPE_DATE;
    }
    if (strcmp("datetime", str) == 0) {
        return RAST_TYPE_DATETIME;
    }
    if (strcmp("uint", str) == 0) {
        return RAST_TYPE_UINT;
    }
    return -1;
}

static const char *
rast_sort_order_e_to_str(rast_sort_order_e sort_order)
{
    switch (sort_order) {
    case RAST_SORT_ORDER_ASCENDING:
        return "ascending";
    case RAST_SORT_ORDER_DESCENDING:
        return "descending";
    default:
        return "default";
    }
}

static const char *
rast_score_method_e_to_str(rast_score_method_e score_method)
{
    if (score_method == RAST_SCORE_METHOD_TFIDF) {
        return "tfidf";
    }
    return "none";
}

static const char *
rast_sort_method_e_to_str(rast_sort_method_e sort_method)
{
    if (sort_method == RAST_SORT_METHOD_PROPERTY) {
        return "property";
    }
    return "score";
}

static rast_error_t *
xmlrpc_client_db_open(rast_db_t **base,
                      const char *name, int flags,
                      rast_db_open_option_t *options, apr_pool_t *pool)
{
    const char *xmlrpc_http_scheme = "xmlrpc.http://";

    if (strncmp(name, xmlrpc_http_scheme, strlen(xmlrpc_http_scheme)) != 0) {
        return rast_error(RAST_ERROR_UNSUPPORTED_SCHEME,
                          "%s is not URL", name);
    }

    return rast_xmlrpc_client_open(base, name, flags, options, pool);
}

rast_error_t *
rast_xmlrpc_client_initialize()
{
    initialized++;
    if (initialized > 1) {
        return RAST_OK;
    }
    xmlrpc_client_init(XMLRPC_CLIENT_NO_FLAGS, "rast-xmlrpc-client", VERSION);
    rast_initialize();
    rast_add_open_function(xmlrpc_client_db_open);
    return RAST_OK;
}

void
rast_xmlrpc_client_finalize()
{
    initialized--;
    if (initialized > 0) {
        return;
    }
    rast_finalize();
    xmlrpc_client_cleanup();
}

static rast_error_t *
rast_xmlrpc_error(int code, const char *fmt, ...)
{
    rast_error_t *error;
    va_list ap;

    va_start(ap, fmt);
    error = rast_error_vcreate(RAST_ERROR_TYPE_XMLRPC, code, fmt, ap);
    va_end(ap);

    return error;
}

static rast_error_t *
xmlrpc_rast_error_to_rast_error(xmlrpc_env *env, xmlrpc_value *value)
{
    if (xmlrpc_struct_has_key(env, value, "error")) {
        xmlrpc_int32 code;
        const char *type, *message;

        xmlrpc_parse_value(env, value, "{s:{s:s,s:i,s:s,*},*}",
                           "error", "type", &type,
                           "code", &code, "message", &message);
        if (env->fault_occurred) {
            return xmlrpc_error_to_rast_error(env, "");
        }

        return rast_error_create(str_to_rast_error_type_t(type),
                                 code, "%s", message);
    }

    return RAST_OK;
}

rast_error_t *
set_authenticate_if_needed(rast_xmlrpc_client_t *db, const char *name)
{
    apr_pool_t *pool;
    char *p, *q, *username, *password;
    xmlrpc_env *env = &db->env;

    p = strstr(name, "://");
    p += 3;
    q = strstr(p, "@");

    if (q == NULL) {
        return RAST_OK;
    }

    apr_pool_create(&pool, db->base.pool);
    username = apr_pstrndup(pool, p, q - p);
    q = strstr(username, ":");
    if (q == NULL) {
        apr_pool_destroy(pool);
        return rast_error(RAST_ERROR_UNSUPPORTED_SCHEME,
                          "%s includes invalid userinfo", name);
    }
    *q = '\0';
    password = q + 1;

    xmlrpc_server_info_set_basic_auth(env, db->server, username, password);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }
    apr_pool_destroy(pool);
    return xmlrpc_error_to_rast_error(&db->env, "");
}

rast_error_t *
rast_xmlrpc_client_open(rast_db_t **base,
                        const char *name, int flags,
                        rast_db_open_option_t *options, apr_pool_t *pool)
{
    const char *http_scheme = "http://";
    const char *xmlrpc_prefix = "xmlrpc.";
    const char *url;
    xmlrpc_env *env;
    const static rast_db_t default_base = {
        NULL,
        NULL,
        rast_xmlrpc_client_close,
        rast_xmlrpc_client_register,
        rast_xmlrpc_client_create_document,
        rast_xmlrpc_client_search,
        rast_xmlrpc_client_delete,
        rast_xmlrpc_client_update,
        rast_xmlrpc_client_get_text,
        NULL,
        rast_xmlrpc_client_encoding,
        rast_xmlrpc_client_properties,
        NULL,
    };
    rast_xmlrpc_client_t *db;

    if (strncmp(name, xmlrpc_prefix, strlen(xmlrpc_prefix)) == 0) {
        url = name + strlen(xmlrpc_prefix);
    }
    else {
        url = name;
    }
    if (strncmp(url, http_scheme, strlen(http_scheme)) != 0) {
        return rast_error(RAST_ERROR_UNSUPPORTED_SCHEME,
                          "%s is not URL", name);
    }

    db = (rast_xmlrpc_client_t *)
        apr_palloc(pool, sizeof(rast_xmlrpc_client_t));
    *base = (rast_db_t *) db;
    db->base = default_base;
    db->url = url;
    db->properties = NULL;
    db->base.pool = pool;
    xmlrpc_env_init(&db->env);
    env = &db->env;

    db->server = xmlrpc_server_info_new(env, (char *) url);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }
    return set_authenticate_if_needed(db, url);
}

rast_error_t *
rast_xmlrpc_client_close(rast_db_t *base)
{
    rast_xmlrpc_client_t *db = (rast_xmlrpc_client_t *) base;

    xmlrpc_server_info_free(db->server);
    xmlrpc_env_clean(&db->env);
    return RAST_OK;
}

static rast_error_t *
get_register_properties(rast_xmlrpc_client_t *db,
                        rast_value_t *properties,
                        xmlrpc_value **rpc_properties)
{
    xmlrpc_env *env = &db->env;
    int num_properties, i;

    *rpc_properties = xmlrpc_build_value(env, "()");
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }
    (void) rast_xmlrpc_client_properties(&db->base, &num_properties);
    for (i = 0; i < db->num_properties; i++) {
        xmlrpc_value *rpc_property;

        switch (db->properties[i].type) {
        case RAST_TYPE_STRING:
            rpc_property =
                xmlrpc_build_value(env, "s",
                                   rast_value_string(&properties[i]));
            break;
        case RAST_TYPE_DATE:
            rpc_property =
                xmlrpc_build_value(env, "s", rast_value_date(&properties[i]));
            break;
        case RAST_TYPE_DATETIME:
            rpc_property =
                xmlrpc_build_value(env, "s",
                                   rast_value_datetime(&properties[i]));
            break;
        case RAST_TYPE_UINT:
            rpc_property =
                xmlrpc_build_value(env, "i", rast_value_uint(&properties[i]));
            break;
        default:
            return rast_error_create(RAST_ERROR_TYPE_RAST,
                                     RAST_ERROR_INVALID_ARGUMENT, "");
        }
        if (env->fault_occurred) {
            xmlrpc_DECREF(rpc_property);
            xmlrpc_DECREF(*rpc_properties);
            return xmlrpc_error_to_rast_error(env, "");
        }
        xmlrpc_array_append_item(env, *rpc_properties, rpc_property);
        xmlrpc_DECREF(rpc_property);
        if (env->fault_occurred) {
            xmlrpc_DECREF(*rpc_properties);
            return xmlrpc_error_to_rast_error(env, "");
        }
    }
    return RAST_OK;
}

static rast_error_t *
get_doc_id(xmlrpc_env *env, xmlrpc_value *rpc_value, rast_doc_id_t *result)
{
    xmlrpc_int32 rpc_doc_id;

    if (result == NULL) {
        return RAST_OK;
    }

    xmlrpc_parse_value(env, rpc_value, "{s:i,*}", "doc_id", &rpc_doc_id);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }

    *result = (rast_doc_id_t) rpc_doc_id;
    return RAST_OK;
}

rast_error_t *
rast_xmlrpc_client_register(rast_db_t *base,
                            const char *text, rast_size_t nbytes,
                            rast_value_t *properties,
                            rast_doc_id_t *new_doc_id)
{
    rast_xmlrpc_client_t *db = (rast_xmlrpc_client_t *) base;
    xmlrpc_env *env = &db->env;
    xmlrpc_value *rpc_properties, *rpc_result;
    rast_error_t *error;

    error = get_register_properties(db, properties, &rpc_properties);
    if (error != RAST_OK) {
        return error;
    }
    rpc_result = xmlrpc_client_call_server(env, db->server,
                                           "rast.register",
                                           "({s:s#,s:V})",
                                           "text", text, nbytes,
                                           "properties", rpc_properties);
    xmlrpc_DECREF(rpc_properties);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }

    error = get_doc_id(env, rpc_result, new_doc_id);
    xmlrpc_DECREF(rpc_result);
    return error;
}

static rast_error_t *xmlrpc_client_document_add_text(rast_document_t *doc,
                                                     const char *text,
                                                     int nbytes);
static rast_error_t *xmlrpc_client_document_commit(rast_document_t *doc);
static rast_error_t *xmlrpc_client_document_abort(rast_document_t *doc);
static rast_error_t *xmlrpc_client_document_set_property(rast_document_t *doc,
                                                         const char *name,
                                                         const rast_value_t *value);

rast_error_t *
rast_xmlrpc_client_create_document(rast_db_t *base, rast_document_t **result)
{
    const static rast_document_t default_base = {
        NULL,
        NULL,
        xmlrpc_client_document_add_text,
        xmlrpc_client_document_set_property,
        xmlrpc_client_document_commit,
        xmlrpc_client_document_abort,
        NULL,
    };
    rast_xmlrpc_client_t *db = (rast_xmlrpc_client_t *) base;
    apr_pool_t *pool;
    rast_xmlrpc_client_document_t *doc;

    apr_pool_create(&pool, db->base.pool);
    doc = (rast_xmlrpc_client_document_t *)
        apr_palloc(pool, sizeof(rast_xmlrpc_client_document_t));
    doc->base = default_base;
    doc->base.pool = pool;
    doc->base.db = base;
    doc->buf_ring = (rast_buf_ring_head_t *)
        apr_palloc(pool, sizeof(rast_buf_ring_head_t));
    APR_RING_INIT(doc->buf_ring, rast_buf_ring_entry_t, link);
    doc->nbytes = 0;
    doc->property_values = apr_hash_make(doc->base.pool);
    *result = (rast_document_t *) doc;

    return RAST_OK;
}

static rast_error_t *
get_result_terms(apr_pool_t *pool, xmlrpc_env *env, xmlrpc_value *rpc_terms,
                 rast_result_t *result)
{
    int num_terms, i;

    num_terms = xmlrpc_array_size(env, rpc_terms);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }
    result->terms = (rast_result_term_t *)
        apr_palloc(pool, sizeof(rast_result_item_t) * num_terms);
    for (i = 0; i < num_terms; i++) {
        xmlrpc_value *item = xmlrpc_array_get_item(env, rpc_terms, i);

        if (env->fault_occurred) {
            return xmlrpc_error_to_rast_error(env, "");
        }
        xmlrpc_int32 doc_count;
        char *term;

        xmlrpc_parse_value(env, item, "{s:s,s:i,*}",
                           "term", &term, "doc_count", &doc_count);
        if (env->fault_occurred) {
            return xmlrpc_error_to_rast_error(env, "");
        }

        result->terms[i].term = apr_pstrdup(pool, term);
        result->terms[i].doc_count = (int) doc_count;
    }

    result->num_terms = num_terms;
    return RAST_OK;
}

static rast_error_t *
create_result_properties(xmlrpc_env *env, xmlrpc_value *rpc_item,
                         rast_value_t **result, apr_pool_t *pool)
{
    xmlrpc_value *rpc_properties;
    int i, num_properties;
    rast_value_t *properties;

    if (!xmlrpc_struct_has_key(env, rpc_item, "properties")) {
        *result = NULL;
        return RAST_OK;
    }

    xmlrpc_parse_value(env, rpc_item, "{s:V,*}",
                       "properties", &rpc_properties);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }
    num_properties = xmlrpc_array_size(env, rpc_properties);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }
    properties = apr_palloc(pool, sizeof(rast_value_t) * num_properties);
    *result = properties;
    for (i = 0; i < num_properties; i++) {
        xmlrpc_value *rpc_property;
        const char *str_type, *str_property;
        xmlrpc_int32 int_property;
        rast_type_e type;

        rpc_property = xmlrpc_array_get_item(env, rpc_properties, i);
        if (env->fault_occurred) {
            return xmlrpc_error_to_rast_error(env, "");
        }
        xmlrpc_parse_value(env, rpc_property, "{s:s,*}", "type", &str_type);
        type = str_to_rast_type_e(str_type);
        switch (type) {
        case RAST_TYPE_STRING:
            xmlrpc_parse_value(env, rpc_property, "{s:s,*}",
                               "value", &str_property);
            rast_value_set_string(&properties[i],
                                  apr_pstrdup(pool, str_property));
            break;
        case RAST_TYPE_DATE:
            xmlrpc_parse_value(env, rpc_property, "{s:s,*}",
                               "value", &str_property);
            rast_value_set_date(&properties[i],
                                apr_pstrdup(pool, str_property));
            break;
        case RAST_TYPE_DATETIME:
            xmlrpc_parse_value(env, rpc_property, "{s:s,*}",
                               "value", &str_property);
            rast_value_set_datetime(&properties[i],
                                    apr_pstrdup(pool, str_property));
            break;
        case RAST_TYPE_UINT:
            xmlrpc_parse_value(env, rpc_property, "{s:i,*}",
                               "value", &int_property);
            rast_value_set_uint(&properties[i], (int) int_property);
            break;
        default:
            return rast_xmlrpc_error(RAST_ERROR_GENERAL,
                                     "invalid value type: %d", type);
        }
        if (env->fault_occurred) {
            return xmlrpc_error_to_rast_error(env, "");
        }

        rast_value_set_type(&properties[i], type);
    }

    return RAST_OK;
}

static rast_error_t *
get_result_items(apr_pool_t *pool, xmlrpc_env *env, xmlrpc_value *rpc_items,
                 rast_result_t *result)
{
    int num_items, i;
    rast_error_t *error;

    num_items = xmlrpc_array_size(env, rpc_items);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }
    result->items = (rast_result_item_t **)
        apr_palloc(pool, sizeof(rast_result_item_t *) * num_items);
    for (i = 0; i < num_items; i++) {
        xmlrpc_value *item;
        xmlrpc_int32 doc_id, db_index, score, summary_nbytes;
        char *summary;

        item = xmlrpc_array_get_item(env, rpc_items, i);
        if (env->fault_occurred) {
            return xmlrpc_error_to_rast_error(env, "");
        }
        xmlrpc_parse_value(env, item, "{s:i,s:i,s:i,*}",
                           "doc_id", &doc_id, "db_index", &db_index,
                           "score", &score);
        if (env->fault_occurred) {
            return xmlrpc_error_to_rast_error(env, "");
        }
        if (xmlrpc_struct_has_key(env, item, "summary")) {
            char *s;
            xmlrpc_parse_value(env, item, "{s:s#,*}",
                               "summary", &s, &summary_nbytes);
            summary = apr_pmemdup(pool, s, summary_nbytes);
        }
        else {
            summary = NULL;
            summary_nbytes = 0;
        }
        if (env->fault_occurred) {
            return xmlrpc_error_to_rast_error(env, "");
        }

        result->items[i] = (rast_result_item_t *)
            apr_palloc(pool, sizeof(rast_result_item_t));
        result->items[i]->doc_id = doc_id;
        result->items[i]->db_index = db_index;
        result->items[i]->score = score;
        result->items[i]->summary = summary;
        result->items[i]->summary_nbytes = summary_nbytes;
        error = create_result_properties(env, item,
                                         &result->items[i]->properties, pool);
        if (error != RAST_OK) {
            return error;
        }
    }

    result->num_items = num_items;
    return RAST_OK;
}

static rast_error_t *
set_terms_option(xmlrpc_env *env, rast_search_option_t *options,
                 xmlrpc_value *rpc_options)
{
    xmlrpc_value *rpc_option_terms;
    int i;

    if (options->num_terms <= 0) {
        return RAST_OK;
    }

    rpc_option_terms = xmlrpc_build_value(env, "()");
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }
    for (i = 0; i < options->num_terms; i++) {
        xmlrpc_value *term_count;

        term_count = xmlrpc_build_value(env, "i",
                                        (xmlrpc_int32) options->terms[i]);
        if (env->fault_occurred) {
            return xmlrpc_error_to_rast_error(env, "");
        }
        xmlrpc_array_append_item(env, rpc_option_terms, term_count);
        if (env->fault_occurred) {
            return xmlrpc_error_to_rast_error(env, "");
        }
        xmlrpc_DECREF(term_count);
    }
    xmlrpc_struct_set_value(env, rpc_options, "terms", rpc_option_terms);
    xmlrpc_DECREF(rpc_option_terms);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }

    return RAST_OK;
}

static rast_error_t *
set_sort_property_option(xmlrpc_env *env, rast_search_option_t *options,
                         xmlrpc_value *rpc_options)
{
    xmlrpc_value *sort_property;

    if (options->sort_property == NULL) {
        return RAST_OK;
    }

    sort_property = xmlrpc_build_value(env, "s", options->sort_property);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }

    xmlrpc_struct_set_value(env, rpc_options, "sort_property", sort_property);
    xmlrpc_DECREF(sort_property);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }

    return RAST_OK;
}

static rast_error_t *
set_properties_option(xmlrpc_env *env, rast_search_option_t *options,
                      xmlrpc_value *rpc_options)
{
    xmlrpc_value *properties;
    int i;

    if (options->num_properties <= 0) {
        return RAST_OK;
    }

    properties = xmlrpc_build_value(env, "()");
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }

    for (i = 0; i < options->num_properties; i++) {
        xmlrpc_value *property =
            xmlrpc_build_value(env, "s", options->properties[i]);

        if (env->fault_occurred) {
            return xmlrpc_error_to_rast_error(env, "");
        }
        xmlrpc_array_append_item(env, properties, property);
        xmlrpc_DECREF(property);
        if (env->fault_occurred) {
            return xmlrpc_error_to_rast_error(env, "");
        }
    }
    xmlrpc_struct_set_value(env, rpc_options, "properties", properties);
    xmlrpc_DECREF(properties);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }

    return RAST_OK;
}

static rast_error_t *
create_search_options(xmlrpc_env *env, rast_search_option_t *options,
                      xmlrpc_value **result)
{
    xmlrpc_value *search_options;
    const char *sort_order, *score_method, *sort_method;
    rast_error_t *error;

    sort_order = rast_sort_order_e_to_str(options->sort_order);
    score_method = rast_score_method_e_to_str(options->score_method);
    sort_method = rast_sort_method_e_to_str(options->sort_method);
    search_options = xmlrpc_build_value(env,
                                        "{s:b,s:i,s:s,s:s,s:s,s:i,s:i,s:i}",
                                        "need_summary", options->need_summary,
                                        "summary_nchars",
                                        options->summary_nchars,
                                        "sort_order", sort_order,
                                        "score_method", score_method,
                                        "sort_method", sort_method,
                                        "start_no", options->start_no,
                                        "num_items", options->num_items,
                                        "all_num_docs", options->all_num_docs);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }

    error = set_terms_option(env, options, search_options);
    if (error != RAST_OK) {
        xmlrpc_DECREF(search_options);
        return error;
    }

    error = set_sort_property_option(env, options, search_options);
    if (error != RAST_OK) {
        xmlrpc_DECREF(search_options);
        return error;
    }

    error = set_properties_option(env, options, search_options);
    if (error != RAST_OK) {
        xmlrpc_DECREF(search_options);
        return error;
    }

    *result = search_options;
    return RAST_OK;
}

static rast_error_t *
create_result(xmlrpc_env *env, xmlrpc_value *rpc_result,
              rast_result_t **result, apr_pool_t *pool)
{
    xmlrpc_value *rpc_terms, *rpc_items;
    xmlrpc_int32 num_indices, num_docs, hit_count;
    rast_error_t *error;

    *result = (rast_result_t *) apr_palloc(pool, sizeof(rast_result_t));
    xmlrpc_parse_value(env, rpc_result, "{s:i,s:i,s:i,s:V,s:V,*}",
                       "num_indices", &num_indices,
                       "num_docs", &num_docs,
                       "hit_count", &hit_count,
                       "terms", &rpc_terms, "items", &rpc_items);
    (*result)->num_indices = (int) num_indices;
    (*result)->num_docs = (int) num_docs;
    (*result)->hit_count = (int) hit_count;
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }

    error = get_result_terms(pool, env, rpc_terms, *result);
    if (error != RAST_OK) {
        return error;
    }
    error = get_result_items(pool, env, rpc_items, *result);
    if (error != RAST_OK) {
        return error;
    }

    return RAST_OK;
}

rast_error_t *
rast_xmlrpc_client_search(rast_db_t *base, const char *query,
                          rast_search_option_t *options,
                          rast_result_t **result, apr_pool_t *pool)
{
    rast_xmlrpc_client_t *db = (rast_xmlrpc_client_t *) base;
    xmlrpc_value *rpc_result, *rpc_options;
    rast_error_t *error;
    xmlrpc_env *env = &db->env;

    error = create_search_options(env, options, &rpc_options);
    if (error != RAST_OK) {
        return error;
    }

    rpc_result = xmlrpc_client_call_server(env, db->server,
                                           "rast.search",
                                           "({s:s,s:V})", "query", query,
                                           "options", rpc_options);
    xmlrpc_DECREF(rpc_options);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }

    error = xmlrpc_rast_error_to_rast_error(env, rpc_result);
    if (error != RAST_OK) {
        xmlrpc_DECREF(rpc_result);
        return error;
    }

    error = create_result(env, rpc_result, result, pool);
    xmlrpc_DECREF(rpc_result);
    return error;
}

rast_error_t *
rast_xmlrpc_client_delete(rast_db_t *base, rast_doc_id_t doc_id)
{
    rast_xmlrpc_client_t *db = (rast_xmlrpc_client_t *) base;
    xmlrpc_value *rpc_result;
    xmlrpc_env *env = &db->env;
    rast_error_t *error;

    rpc_result = xmlrpc_client_call_server(env, db->server,
                                           "rast.delete", "({s:i})",
                                           "doc_id", (xmlrpc_int32) doc_id);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }

    error = xmlrpc_rast_error_to_rast_error(env, rpc_result);
    xmlrpc_DECREF(rpc_result);
    return error;
}

rast_error_t *
rast_xmlrpc_client_update(rast_db_t *base, rast_doc_id_t doc_id,
                          const char *text, rast_size_t nbytes,
                          rast_value_t *properties, rast_doc_id_t *new_doc_id)
{
    rast_xmlrpc_client_t *db = (rast_xmlrpc_client_t *) base;
    xmlrpc_value *rpc_properties, *rpc_result;
    rast_error_t *error;
    xmlrpc_env *env = &db->env;

    error = get_register_properties(db, properties, &rpc_properties);
    if (error != RAST_OK) {
        return error;
    }

    rpc_result = xmlrpc_client_call_server(env, db->server, "rast.update",
                                           "({s:i,s:s#,s:V})",
                                           "doc_id", (xmlrpc_int32) doc_id,
                                           "text", text, nbytes,
                                           "properties", rpc_properties);
    xmlrpc_DECREF(rpc_properties);
    if (env->fault_occurred) {
        return xmlrpc_error_to_rast_error(env, "");
    }

    error = xmlrpc_rast_error_to_rast_error(env, rpc_result);
    if (error != RAST_OK) {
        xmlrpc_DECREF(rpc_result);
        return error;
    }

    error = get_doc_id(env, rpc_result, new_doc_id);
    xmlrpc_DECREF(rpc_result);
    return error;
}

rast_error_t *
rast_xmlrpc_client_get_text(rast_db_t *base, rast_doc_id_t doc_id,
                            char **s, rast_size_t *nbytes, apr_pool_t *pool)
{
    rast_xmlrpc_client_t *db = (rast_xmlrpc_client_t *) base;
    xmlrpc_value *rpc_result;
    xmlrpc_env *env = &db->env;
    xmlrpc_int32 rpc_nbytes;
    char *rpc_text;
    rast_error_t *error;

    rpc_result = xmlrpc_client_call_server(env, db->server,
                                           "rast.get_text", "({s:i})",
                                           "doc_id", (xmlrpc_int32) doc_id);
    if (env->fault_occurred) {
        xmlrpc_DECREF(rpc_result);
        return xmlrpc_error_to_rast_error(env, "");
    }

    error = xmlrpc_rast_error_to_rast_error(env, rpc_result);
    if (error != RAST_OK) {
        xmlrpc_DECREF(rpc_result);
        return error;
    }

    xmlrpc_parse_value(env, rpc_result, "{s:s#,*}",
                       "text", &rpc_text, &rpc_nbytes);
    if (env->fault_occurred) {
        xmlrpc_DECREF(rpc_result);
        return xmlrpc_error_to_rast_error(env, "");
    }
    *s = apr_pmemdup(pool, rpc_text, rpc_nbytes);
    *nbytes = (rast_size_t) rpc_nbytes;

    xmlrpc_DECREF(rpc_result);
    return RAST_OK;
}

const char *
rast_xmlrpc_client_encoding(rast_db_t *base)
{
    rast_xmlrpc_client_t *db = (rast_xmlrpc_client_t *) base;
    xmlrpc_env *env = &db->env;
    xmlrpc_value *rpc_encoding;
    const char *encoding, *result;

    rpc_encoding = xmlrpc_client_call_server(env, db->server,
                                               "rast.encoding", "()");
    if (env->fault_occurred) {
        xmlrpc_DECREF(rpc_encoding);
        /* return xmlrpc_error_to_rast_error(env); */
        return NULL;
    }

    xmlrpc_parse_value(env, rpc_encoding, "{s:s,*}",
                       "encoding", &encoding);
    if (env->fault_occurred) {
        xmlrpc_DECREF(rpc_encoding);
        /* return xmlrpc_error_to_rast_error(env); */
        return NULL;
    }

    result = apr_pstrdup(db->base.pool, encoding);
    xmlrpc_DECREF(rpc_encoding);
    return result;
}

const rast_property_t *
rast_xmlrpc_client_properties(rast_db_t *base, int *num_properties)
{
    rast_xmlrpc_client_t *db = (rast_xmlrpc_client_t *) base;
    xmlrpc_env *env = &db->env;

    if (db->properties == NULL) {
        int i;
        xmlrpc_value *rpc_properties;
        rast_property_t *properties;

        rpc_properties = xmlrpc_client_call_server(env, db->server,
                                                   "rast.properties", "()");
        if (env->fault_occurred) {
            xmlrpc_DECREF(rpc_properties);
            /* return xmlrpc_error_to_rast_error(env); */
            return NULL;
        }

        db->num_properties = xmlrpc_array_size(env, rpc_properties);
        if (env->fault_occurred) {
            xmlrpc_DECREF(rpc_properties);
            /* return xmlrpc_error_to_rast_error(env); */
            return NULL;
        }
        properties = (rast_property_t *)
            apr_palloc(db->base.pool,
                       sizeof(rast_property_t) * db->num_properties);
        for (i = 0; i < db->num_properties; i++) {
            xmlrpc_value *rpc_property;
            xmlrpc_int32 flags;
            const char *name, *type;

            rpc_property = xmlrpc_array_get_item(env, rpc_properties, i);
            if (env->fault_occurred) {
                xmlrpc_DECREF(rpc_properties);
                /* return xmlrpc_error_to_rast_error(env); */
                return NULL;
            }
            xmlrpc_parse_value(env, rpc_property, "{s:s,s:s,s:i,*}",
                               "name", &name, "type", &type, "flags", &flags);
            if (env->fault_occurred) {
                xmlrpc_DECREF(rpc_properties);
                /* return xmlrpc_error_to_rast_error(env); */
                return NULL;
            }
            properties[i].name = apr_pstrdup(db->base.pool, name);
            properties[i].type = str_to_rast_type_e(type);
            properties[i].flags = (int) flags;
        }

        xmlrpc_DECREF(rpc_properties);
        db->properties = properties;
    }

    *num_properties = db->num_properties;
    return db->properties;
}

static rast_error_t *
xmlrpc_client_document_add_text(rast_document_t *base, const char *text,
                                int nbytes)
{
    rast_xmlrpc_client_document_t *doc;
    rast_buf_ring_entry_t *entry;

    doc = (rast_xmlrpc_client_document_t *) base;
    entry = (rast_buf_ring_entry_t *)
        apr_palloc(doc->base.pool, sizeof(rast_buf_ring_entry_t));
    entry->buf = apr_pmemdup(doc->base.pool, text, nbytes);
    entry->nbytes = nbytes;
    APR_RING_INSERT_TAIL(doc->buf_ring, entry, rast_buf_ring_entry_t, link);

    doc->nbytes += nbytes;

    return RAST_OK;
}

static rast_error_t *
xmlrpc_client_document_set_property(rast_document_t *base, const char *name,
                                    const rast_value_t *value)
{
    rast_xmlrpc_client_document_t *doc =
        (rast_xmlrpc_client_document_t *) base;
    rast_value_t *property_value;

    property_value = (rast_value_t *) apr_palloc(doc->base.pool,
                                                 sizeof(rast_value_t));
    rast_value_set_type(property_value, value->type);
    switch (value->type) {
    case RAST_TYPE_STRING:
        rast_value_set_string(property_value,
                              apr_pstrdup(doc->base.pool,
                                          rast_value_string(value)));
        break;
    case RAST_TYPE_DATE:
        rast_value_set_date(property_value,
                            apr_pstrdup(doc->base.pool, rast_value_date(value)));
        break;
    case RAST_TYPE_DATETIME:
        rast_value_set_datetime(property_value,
                                apr_pstrdup(doc->base.pool,
                                            rast_value_datetime(value)));
        break;
    case RAST_TYPE_UINT:
        rast_value_set_uint(property_value, rast_value_uint(value));
        break;
    }
    rast_value_set_type(property_value, value->type);
    apr_hash_set(doc->property_values,
                 apr_pstrdup(doc->base.pool, name), strlen(name), property_value);

    return RAST_OK;
}

static rast_error_t *
xmlrpc_client_document_commit(rast_document_t *base)
{
    rast_xmlrpc_client_document_t *doc =
        (rast_xmlrpc_client_document_t *) base;
    char *text, *p;
    rast_buf_ring_entry_t *entry;
    rast_value_t *property_values;
    const rast_property_t *properties;
    int num_properties;
    rast_error_t *error;

    text = apr_palloc(doc->base.pool, doc->nbytes);
    p = text;
    for (entry = APR_RING_FIRST(doc->buf_ring);
         entry != APR_RING_SENTINEL(doc->buf_ring, rast_buf_ring_entry_t,
                                    link);
         entry = APR_RING_NEXT(entry, link)) {
        memcpy(p, entry->buf, entry->nbytes);
        p += entry->nbytes;
    }

    properties = rast_db_properties(base->db, &num_properties);
    error = rast_apr_hash_to_rast_value_array(properties,
                                              num_properties,
                                              doc->property_values,
                                              &property_values,
                                              doc->base.pool);
    if (error != RAST_OK) {
        return error;
    }

    error = rast_db_register(base->db, text, doc->nbytes, property_values,
                             NULL);
    if (error != RAST_OK) {
        return error;
    }

    return xmlrpc_client_document_abort(base);
}

static rast_error_t *
xmlrpc_client_document_abort(rast_document_t *doc)
{
    if (doc != NULL) {
        apr_pool_destroy(doc->pool);
    }
    return RAST_OK;
}