/*
 *
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 */

#include <stdlib.h>
#include <string>
#include <windows.h>
#include <qpid/broker/RecoverableQueue.h>
#include <qpid/log/Statement.h>
#include <qpid/store/MessageStorePlugin.h>
#include <qpid/store/StorageProvider.h>
#include "AmqpTransaction.h"
#include "BlobAdapter.h"
#include "BlobRecordset.h"
#include "BindingRecordset.h"
#include "MessageMapRecordset.h"
#include "MessageRecordset.h"
#include "TplRecordset.h"
#include "DatabaseConnection.h"
#include "Exception.h"
#include "State.h"
#include "VariantHelper.h"

// Bring in ADO 2.8 (yes, I know it says "15", but that's it...)
#import "C:\Program Files\Common Files\System\ado\msado15.dll" \
        no_namespace rename("EOF", "EndOfFile")
#include <comdef.h>
namespace {
inline void TESTHR(HRESULT x) {if FAILED(x) _com_issue_error(x);};

// Table names
const std::string TblBinding("tblBinding");
const std::string TblConfig("tblConfig");
const std::string TblExchange("tblExchange");
const std::string TblMessage("tblMessage");
const std::string TblMessageMap("tblMessageMap");
const std::string TblQueue("tblQueue");
const std::string TblTpl("tblTPL");
}

namespace qpid {
namespace store {
namespace ms_sql {

/**
 * @class MSSqlProvider
 *
 * Implements a qpid::store::StorageProvider that uses Microsoft SQL Server as
 * the backend data store for Qpid.
 */
class MSSqlProvider : public qpid::store::StorageProvider
{
protected:
    void finalizeMe();

    void dump();

public:
    MSSqlProvider();
    ~MSSqlProvider();

    virtual qpid::Options* getOptions() { return &options; }

    virtual void earlyInitialize (Plugin::Target& target);
    virtual void initialize(Plugin::Target& target);

    /**
     * Receive notification that this provider is the one that will actively
     * handle provider storage for the target. If the provider is to be used,
     * this method will be called after earlyInitialize() and before any
     * recovery operations (recovery, in turn, precedes call to initialize()).
     */
    virtual void activate(MessageStorePlugin &store);

    /**
     * @name Methods inherited from qpid::broker::MessageStore
     */

    /**
     * Record the existence of a durable queue
     */
    virtual void create(PersistableQueue& queue,
                        const qpid::framing::FieldTable& args);
    /**
     * Destroy a durable queue
     */
    virtual void destroy(PersistableQueue& queue);

    /**
     * Record the existence of a durable exchange
     */
    virtual void create(const PersistableExchange& exchange,
                        const qpid::framing::FieldTable& args);
    /**
     * Destroy a durable exchange
     */
    virtual void destroy(const PersistableExchange& exchange);

    /**
     * Record a binding
     */
    virtual void bind(const PersistableExchange& exchange,
                      const PersistableQueue& queue,
                      const std::string& key,
                      const qpid::framing::FieldTable& args);

    /**
     * Forget a binding
     */
    virtual void unbind(const PersistableExchange& exchange,
                        const PersistableQueue& queue,
                        const std::string& key,
                        const qpid::framing::FieldTable& args);

    /**
     * Record generic durable configuration
     */
    virtual void create(const PersistableConfig& config);

    /**
     * Destroy generic durable configuration
     */
    virtual void destroy(const PersistableConfig& config);

    /**
     * Stores a messages before it has been enqueued
     * (enqueueing automatically stores the message so this is
     * only required if storage is required prior to that
     * point). If the message has not yet been stored it will
     * store the headers as well as any content passed in. A
     * persistence id will be set on the message which can be
     * used to load the content or to append to it.
     */
    virtual void stage(const boost::intrusive_ptr<PersistableMessage>& msg);

    /**
     * Destroys a previously staged message. This only needs
     * to be called if the message is never enqueued. (Once
     * enqueued, deletion will be automatic when the message
     * is dequeued from all queues it was enqueued onto).
     */
    virtual void destroy(PersistableMessage& msg);

    /**
     * Appends content to a previously staged message
     */
    virtual void appendContent(const boost::intrusive_ptr<const PersistableMessage>& msg,
                               const std::string& data);

    /**
     * Loads (a section) of content data for the specified
     * message (previously stored through a call to stage or
     * enqueue) into data. The offset refers to the content
     * only (i.e. an offset of 0 implies that the start of the
     * content should be loaded, not the headers or related
     * meta-data).
     */
    virtual void loadContent(const qpid::broker::PersistableQueue& queue,
                             const boost::intrusive_ptr<const PersistableMessage>& msg,
                             std::string& data,
                             uint64_t offset,
                             uint32_t length);

    /**
     * Enqueues a message, storing the message if it has not
     * been previously stored and recording that the given
     * message is on the given queue.
     *
     * Note: that this is async so the return of the function does
     * not mean the opperation is complete.
     *
     * @param msg the message to enqueue
     * @param queue the name of the queue onto which it is to be enqueued
     * @param xid (a pointer to) an identifier of the
     * distributed transaction in which the operation takes
     * place or null for 'local' transactions
     */
    virtual void enqueue(qpid::broker::TransactionContext* ctxt,
                         const boost::intrusive_ptr<PersistableMessage>& msg,
                         const PersistableQueue& queue);

    /**
     * Dequeues a message, recording that the given message is
     * no longer on the given queue and deleting the message
     * if it is no longer on any other queue.
     *
     * Note: that this is async so the return of the function does
     * not mean the opperation is complete.
     *
     * @param msg the message to dequeue
     * @param queue the name of the queue from which it is to be dequeued
     * @param xid (a pointer to) an identifier of the
     * distributed transaction in which the operation takes
     * place or null for 'local' transactions
     */
    virtual void dequeue(qpid::broker::TransactionContext* ctxt,
                         const boost::intrusive_ptr<PersistableMessage>& msg,
                         const PersistableQueue& queue);

    /**
     * Flushes all async messages to disk for the specified queue
     *
     * Note: this is a no-op for this provider.
     *
     * @param queue the name of the queue from which it is to be dequeued
     */
    virtual void flush(const PersistableQueue& queue) {};

    /**
     * Returns the number of outstanding AIO's for a given queue
     *
     * If 0, than all the enqueue / dequeues have been stored
     * to disk
     *
     * @param queue the name of the queue to check for outstanding AIO
     */
    virtual uint32_t outstandingQueueAIO(const PersistableQueue& queue)
        {return 0;}
    //@}

    /**
     * @name Methods inherited from qpid::broker::TransactionalStore
     */
    //@{
    virtual std::auto_ptr<qpid::broker::TransactionContext> begin();
    virtual std::auto_ptr<qpid::broker::TPCTransactionContext> begin(const std::string& xid);
    virtual void prepare(qpid::broker::TPCTransactionContext& txn);
    virtual void commit(qpid::broker::TransactionContext& txn);
    virtual void abort(qpid::broker::TransactionContext& txn);
    virtual void collectPreparedXids(std::set<std::string>& xids);
    //@}

    virtual void recoverConfigs(qpid::broker::RecoveryManager& recoverer);
    virtual void recoverExchanges(qpid::broker::RecoveryManager& recoverer,
                                  ExchangeMap& exchangeMap);
    virtual void recoverQueues(qpid::broker::RecoveryManager& recoverer,
                               QueueMap& queueMap);
    virtual void recoverBindings(qpid::broker::RecoveryManager& recoverer,
                                 const ExchangeMap& exchangeMap,
                                 const QueueMap& queueMap);
    virtual void recoverMessages(qpid::broker::RecoveryManager& recoverer,
                                 MessageMap& messageMap,
                                 MessageQueueMap& messageQueueMap);
    virtual void recoverTransactions(qpid::broker::RecoveryManager& recoverer,
                                     PreparedTransactionMap& dtxMap);

private:
    struct ProviderOptions : public qpid::Options
    {
        std::string connectString;
        std::string catalogName;

        ProviderOptions(const std::string &name)
            : qpid::Options(name),
              catalogName("QpidStore")
        {
            const enum { NAMELEN = MAX_COMPUTERNAME_LENGTH + 1 };
            TCHAR myName[NAMELEN];
            DWORD myNameLen = NAMELEN;
            GetComputerName(myName, &myNameLen);
            connectString = "Data Source=";
            connectString += myName;
            connectString += "\\SQLEXPRESS;Integrated Security=SSPI";
            addOptions()
                ("connect",
                 qpid::optValue(connectString, "STRING"),
                 "Connection string for the database to use. Will prepend "
                 "Provider=SQLOLEDB;")
                ("catalog",
                 qpid::optValue(catalogName, "DB NAME"),
                 "Catalog (database) name")
                ;
        }
    };
    ProviderOptions options;

    // Each thread has a separate connection to the database and also needs
    // to manage its COM initialize/finalize individually. This is done by
    // keeping a thread-specific State.
    boost::thread_specific_ptr<State> dbState;

    State *initState();
    DatabaseConnection *initConnection(void);
    void createDb(DatabaseConnection *db, const std::string &name);
};

static MSSqlProvider static_instance_registers_plugin;

void
MSSqlProvider::finalizeMe()
{
    dbState.reset();
}

MSSqlProvider::MSSqlProvider()
    : options("MS SQL Provider options")
{
}

MSSqlProvider::~MSSqlProvider()
{
}

void
MSSqlProvider::earlyInitialize(Plugin::Target &target)
{
    MessageStorePlugin *store = dynamic_cast<MessageStorePlugin *>(&target);
    if (store) {
        // If the database init fails, report it and don't register; give
        // the rest of the broker a chance to run.
        //
        // Don't try to initConnection() since that will fail if the
        // database doesn't exist. Instead, try to open a connection without
        // a database name, then search for the database. There's still a
        // chance this provider won't be selected for the store too, so be
        // be sure to close the database connection before return to avoid
        // leaving a connection up that will not be used.
        try {
            initState();     // This initializes COM
            std::auto_ptr<DatabaseConnection> db(new DatabaseConnection());
            db->open(options.connectString, "");
            _ConnectionPtr conn(*db);
            _RecordsetPtr pCatalogs = NULL;
            VariantHelper<std::string> catalogName(options.catalogName);
            pCatalogs = conn->OpenSchema(adSchemaCatalogs, catalogName);
            if (pCatalogs->EndOfFile) {
                // Database doesn't exist; create it
                QPID_LOG(notice,
                         "MSSQL: Creating database " + options.catalogName);
                createDb(db.get(), options.catalogName);
            }
            else {
                QPID_LOG(notice,
                         "MSSQL: Database located: " + options.catalogName);
            }
            if (pCatalogs) {
                if (pCatalogs->State == adStateOpen)
                    pCatalogs->Close();
                pCatalogs = 0;
            }
            db->close();
            store->providerAvailable("MSSQL", this);
        }
        catch (qpid::Exception &e) {
            QPID_LOG(error, e.what());
            return;
        }
        store->addFinalizer(boost::bind(&MSSqlProvider::finalizeMe, this));
    }
}

void
MSSqlProvider::initialize(Plugin::Target& target)
{
}

void
MSSqlProvider::activate(MessageStorePlugin &store)
{
  QPID_LOG(info, "MS SQL Provider is up");
}

void
MSSqlProvider::create(PersistableQueue& queue,
                      const qpid::framing::FieldTable& /*args needed for jrnl*/)
{
    DatabaseConnection *db = initConnection();
    BlobRecordset rsQueues;
    try {
        db->beginTransaction();
        rsQueues.open(db, TblQueue);
        rsQueues.add(queue);
        db->commitTransaction();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        db->rollbackTransaction();
        throw ADOException("Error creating queue " + queue.getName(), e, errs);
    }
}

/**
 * Destroy a durable queue
 */
void
MSSqlProvider::destroy(PersistableQueue& queue)
{
    DatabaseConnection *db = initConnection();
    BlobRecordset rsQueues;
    BindingRecordset rsBindings;
    MessageRecordset rsMessages;
    MessageMapRecordset rsMessageMaps;
    try {
        db->beginTransaction();
        rsQueues.open(db, TblQueue);
        rsBindings.open(db, TblBinding);
        rsMessages.open(db, TblMessage);
        rsMessageMaps.open(db, TblMessageMap);
        // Remove bindings first; the queue IDs can't be ripped out from
        // under the references in the bindings table. Then remove the
        // message->queue entries for the queue, also because the queue can't
        // be deleted while there are references to it. If there are messages
        // orphaned by removing the queue references, they're deleted by
        // a trigger on the tblMessageMap table. Lastly, the queue record
        // can be removed.
        rsBindings.removeForQueue(queue.getPersistenceId());
        rsMessageMaps.removeForQueue(queue.getPersistenceId());
        rsQueues.remove(queue);
        db->commitTransaction();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        db->rollbackTransaction();
        throw ADOException("Error deleting queue " + queue.getName(), e, errs);
    }
}

/**
 * Record the existence of a durable exchange
 */
void
MSSqlProvider::create(const PersistableExchange& exchange,
                      const qpid::framing::FieldTable& args)
{
    DatabaseConnection *db = initConnection();
    BlobRecordset rsExchanges;
    try {
        db->beginTransaction();
        rsExchanges.open(db, TblExchange);
        rsExchanges.add(exchange);
        db->commitTransaction();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        db->rollbackTransaction();
        throw ADOException("Error creating exchange " + exchange.getName(),
                           e,
                           errs);
    }
}

/**
 * Destroy a durable exchange
 */
void
MSSqlProvider::destroy(const PersistableExchange& exchange)
{
    DatabaseConnection *db = initConnection();
    BlobRecordset rsExchanges;
    BindingRecordset rsBindings;
    try {
        db->beginTransaction();
        rsExchanges.open(db, TblExchange);
        rsBindings.open(db, TblBinding);
        // Remove bindings first; the exchange IDs can't be ripped out from
        // under the references in the bindings table.
        rsBindings.removeForExchange(exchange.getPersistenceId());
        rsExchanges.remove(exchange);
        db->commitTransaction();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        db->rollbackTransaction();
        throw ADOException("Error deleting exchange " + exchange.getName(),
                           e,
                           errs);
    }
}

/**
 * Record a binding
 */
void
MSSqlProvider::bind(const PersistableExchange& exchange,
                    const PersistableQueue& queue,
                    const std::string& key,
                    const qpid::framing::FieldTable& args)
{
    DatabaseConnection *db = initConnection();
    BindingRecordset rsBindings;
    try {
        db->beginTransaction();
        rsBindings.open(db, TblBinding);
        rsBindings.add(exchange.getPersistenceId(),
                       queue.getPersistenceId(),
                       key,
                       args);
        db->commitTransaction();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        db->rollbackTransaction();
        throw ADOException("Error binding exchange " + exchange.getName() +
                           " to queue " + queue.getName(),
                           e,
                           errs);
    }
}

/**
 * Forget a binding
 */
void
MSSqlProvider::unbind(const PersistableExchange& exchange,
                      const PersistableQueue& queue,
                      const std::string& key,
                      const qpid::framing::FieldTable& args)
{
    DatabaseConnection *db = initConnection();
    BindingRecordset rsBindings;
    try {
        db->beginTransaction();
        rsBindings.open(db, TblBinding);
        rsBindings.remove(exchange.getPersistenceId(),
                          queue.getPersistenceId(),
                          key,
                          args);
        db->commitTransaction();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        db->rollbackTransaction();
        throw ADOException("Error unbinding exchange " + exchange.getName() +
                           " from queue " + queue.getName(),
                           e,
                           errs);
    }
}

/**
 * Record generic durable configuration
 */
void
MSSqlProvider::create(const PersistableConfig& config)
{
    DatabaseConnection *db = initConnection();
    BlobRecordset rsConfigs;
    try {
        db->beginTransaction();
        rsConfigs.open(db, TblConfig);
        rsConfigs.add(config);
        db->commitTransaction();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        db->rollbackTransaction();
        throw ADOException("Error creating config " + config.getName(), e, errs);
    }
}

/**
 * Destroy generic durable configuration
 */
void
MSSqlProvider::destroy(const PersistableConfig& config)
{
    DatabaseConnection *db = initConnection();
    BlobRecordset rsConfigs;
    try {
        db->beginTransaction();
        rsConfigs.open(db, TblConfig);
        rsConfigs.remove(config);
        db->commitTransaction();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        db->rollbackTransaction();
        throw ADOException("Error deleting config " + config.getName(), e, errs);
    }
}

/**
 * Stores a messages before it has been enqueued
 * (enqueueing automatically stores the message so this is
 * only required if storage is required prior to that
 * point). If the message has not yet been stored it will
 * store the headers as well as any content passed in. A
 * persistence id will be set on the message which can be
 * used to load the content or to append to it.
 */
void
MSSqlProvider::stage(const boost::intrusive_ptr<PersistableMessage>& msg)
{
    DatabaseConnection *db = initConnection();
    MessageRecordset rsMessages;
    try {
        db->beginTransaction();
        rsMessages.open(db, TblMessage);
        rsMessages.add(msg);
        db->commitTransaction();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        db->rollbackTransaction();
        throw ADOException("Error staging message", e, errs);
    }  
}

/**
 * Destroys a previously staged message. This only needs
 * to be called if the message is never enqueued. (Once
 * enqueued, deletion will be automatic when the message
 * is dequeued from all queues it was enqueued onto).
 */
void
MSSqlProvider::destroy(PersistableMessage& msg)
{
    DatabaseConnection *db = initConnection();
    BlobRecordset rsMessages;
    try {
        db->beginTransaction();
        rsMessages.open(db, TblMessage);
        rsMessages.remove(msg);
        db->commitTransaction();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        db->rollbackTransaction();
        throw ADOException("Error deleting message", e, errs);
    }
}

/**
 * Appends content to a previously staged message
 */
void
MSSqlProvider::appendContent(const boost::intrusive_ptr<const PersistableMessage>& msg,
                             const std::string& data)
{
    DatabaseConnection *db = initConnection();
    MessageRecordset rsMessages;
    try {
        db->beginTransaction();
        rsMessages.open(db, TblMessage);
        rsMessages.append(msg, data);
        db->commitTransaction();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        db->rollbackTransaction();
        throw ADOException("Error appending to message", e, errs);
    }  
}

/**
 * Loads (a section) of content data for the specified
 * message (previously stored through a call to stage or
 * enqueue) into data. The offset refers to the content
 * only (i.e. an offset of 0 implies that the start of the
 * content should be loaded, not the headers or related
 * meta-data).
 */
void
MSSqlProvider::loadContent(const qpid::broker::PersistableQueue& /*queue*/,
                           const boost::intrusive_ptr<const PersistableMessage>& msg,
                           std::string& data,
                           uint64_t offset,
                           uint32_t length)
{
    // SQL store keeps all messages in one table, so we don't need the
    // queue reference.
    DatabaseConnection *db = initConnection();
    MessageRecordset rsMessages;
    try {
        rsMessages.open(db, TblMessage);
        rsMessages.loadContent(msg, data, offset, length);
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        throw ADOException("Error loading message content", e, errs);
    }  
}

/**
 * Enqueues a message, storing the message if it has not
 * been previously stored and recording that the given
 * message is on the given queue.
 *
 * @param ctxt The transaction context under which this enqueue happens.
 * @param msg The message to enqueue
 * @param queue the name of the queue onto which it is to be enqueued
 */
void
MSSqlProvider::enqueue(qpid::broker::TransactionContext* ctxt,
                       const boost::intrusive_ptr<PersistableMessage>& msg,
                       const PersistableQueue& queue)
{
    // If this enqueue is in the context of a transaction, use the specified
    // transaction to nest a new transaction for this operation. However, if
    // this is not in the context of a transaction, then just use the thread's
    // DatabaseConnection with a ADO transaction.
    DatabaseConnection *db = 0;
    std::string xid;
    AmqpTransaction *atxn = dynamic_cast<AmqpTransaction*> (ctxt);
    if (atxn == 0) {
        db = initConnection();
        db->beginTransaction();
    }
    else {
        (void)initState();     // Ensure this thread is initialized
        // It's a transactional enqueue; if it's TPC, grab the xid.
        AmqpTPCTransaction *tpcTxn = dynamic_cast<AmqpTPCTransaction*> (ctxt);
        if (tpcTxn)
            xid = tpcTxn->getXid();
        db = atxn->dbConn();
        try {
            atxn->sqlBegin();
        }
        catch(_com_error &e) {
            throw ADOException("Error queuing message", e, db->getErrors());
        }
    }

    MessageRecordset rsMessages;
    MessageMapRecordset rsMap;
    try {
        if (msg->getPersistenceId() == 0) {    // Message itself not yet saved
            rsMessages.open(db, TblMessage);
            rsMessages.add(msg);
        }
        rsMap.open(db, TblMessageMap);
        rsMap.add(msg->getPersistenceId(), queue.getPersistenceId(), xid);
        if (atxn)
            atxn->sqlCommit();
        else
            db->commitTransaction();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        if (atxn)
            atxn->sqlAbort();
        else
            db->rollbackTransaction();
        throw ADOException("Error queuing message", e, errs);
    }
    msg->enqueueComplete();
}

/**
 * Dequeues a message, recording that the given message is
 * no longer on the given queue and deleting the message
 * if it is no longer on any other queue.
 *
 * @param ctxt The transaction context under which this dequeue happens.
 * @param msg The message to dequeue
 * @param queue The queue from which it is to be dequeued
 */
void
MSSqlProvider::dequeue(qpid::broker::TransactionContext* ctxt,
                       const boost::intrusive_ptr<PersistableMessage>& msg,
                       const PersistableQueue& queue)
{
    // If this dequeue is in the context of a transaction, use the specified
    // transaction to nest a new transaction for this operation. However, if
    // this is not in the context of a transaction, then just use the thread's
    // DatabaseConnection with a ADO transaction.
    DatabaseConnection *db = 0;
    std::string xid;
    AmqpTransaction *atxn = dynamic_cast<AmqpTransaction*> (ctxt);
    if (atxn == 0) {
        db = initConnection();
        db->beginTransaction();
    }
    else {
        (void)initState();     // Ensure this thread is initialized
        // It's a transactional dequeue; if it's TPC, grab the xid.
        AmqpTPCTransaction *tpcTxn = dynamic_cast<AmqpTPCTransaction*> (ctxt);
        if (tpcTxn)
            xid = tpcTxn->getXid();
        db = atxn->dbConn();
        try {
            atxn->sqlBegin();
        }
        catch(_com_error &e) {
            throw ADOException("Error queuing message", e, db->getErrors());
        }
    }

    MessageMapRecordset rsMap;
    try {
        rsMap.open(db, TblMessageMap);
        // TPC dequeues are just marked pending and will actually be removed
        // when the transaction commits; Single-phase dequeues are removed
        // now, relying on the SQL transaction to put it back if the
        // transaction doesn't commit.
        if (!xid.empty()) {
            rsMap.pendingRemove(msg->getPersistenceId(),
                                queue.getPersistenceId(),
                                xid);
        }
        else {
            rsMap.remove(msg->getPersistenceId(),
                         queue.getPersistenceId());
        }
        if (atxn)
            atxn->sqlCommit();
        else
            db->commitTransaction();
    }
    catch(ms_sql::Exception&) {
        if (atxn)
            atxn->sqlAbort();
        else
            db->rollbackTransaction();
        throw;
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        if (atxn)
            atxn->sqlAbort();
        else
            db->rollbackTransaction();
        throw ADOException("Error dequeuing message", e, errs);
    }  
    msg->dequeueComplete();
}

std::auto_ptr<qpid::broker::TransactionContext>
MSSqlProvider::begin()
{
    (void)initState();     // Ensure this thread is initialized

    // Transactions are associated with the Connection, so this transaction
    // context needs its own connection. At the time of writing, single-phase
    // transactions are dealt with completely on one thread, so we really
    // could just use the thread-specific DatabaseConnection for this.
    // However, that would introduce an ugly, hidden coupling, so play
    // it safe and handle this just like a TPC transaction, which actually
    // can be prepared and committed/aborted from different threads,
    // making it a bad idea to try using the thread-local DatabaseConnection.
    boost::shared_ptr<DatabaseConnection> db(new DatabaseConnection);
    db->open(options.connectString, options.catalogName);
    std::auto_ptr<AmqpTransaction> tx(new AmqpTransaction(db));
    tx->sqlBegin();
    std::auto_ptr<qpid::broker::TransactionContext> tc(tx);
    return tc;
}

std::auto_ptr<qpid::broker::TPCTransactionContext>
MSSqlProvider::begin(const std::string& xid)
{
    (void)initState();     // Ensure this thread is initialized
    boost::shared_ptr<DatabaseConnection> db(new DatabaseConnection);
    db->open(options.connectString, options.catalogName);
    std::auto_ptr<AmqpTPCTransaction> tx(new AmqpTPCTransaction(db, xid));
    tx->sqlBegin();

    TplRecordset rsTpl;
    try {
        tx->sqlBegin();
        rsTpl.open(db.get(), TblTpl);
        rsTpl.add(xid);
        tx->sqlCommit();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        tx->sqlAbort();
        throw ADOException("Error adding TPL record", e, errs);
    }  

    std::auto_ptr<qpid::broker::TPCTransactionContext> tc(tx);
    return tc;
}

void
MSSqlProvider::prepare(qpid::broker::TPCTransactionContext& txn)
{
    // Commit all the marked-up enqueue/dequeue ops and the TPL record.
    // On commit/rollback the TPL will be removed and the TPL markups
    // on the message map will be cleaned up as well.
    (void)initState();     // Ensure this thread is initialized
    AmqpTPCTransaction *atxn = dynamic_cast<AmqpTPCTransaction*> (&txn);
    if (atxn == 0)
        throw qpid::broker::InvalidTransactionContextException();
    try {
        atxn->sqlCommit();
    }
    catch(_com_error &e) {
        throw ADOException("Error preparing", e, atxn->dbConn()->getErrors());
    }  
    atxn->setPrepared();
}

void
MSSqlProvider::commit(qpid::broker::TransactionContext& txn)
{
    (void)initState();     // Ensure this thread is initialized
    /*
     * One-phase transactions simply commit the outer SQL transaction
     * that was begun on begin(). Two-phase transactions are different -
     * the SQL transaction started on begin() was committed on prepare()
     * so all the SQL records reflecting the enqueue/dequeue actions for
     * the transaction are recorded but with xid markups on them to reflect
     * that they are prepared but not committed. Now go back and remove
     * the markups, deleting those marked for removal.
     */
    AmqpTPCTransaction *p2txn = dynamic_cast<AmqpTPCTransaction*> (&txn);
    if (p2txn == 0) {
        AmqpTransaction *p1txn = dynamic_cast<AmqpTransaction*> (&txn);
        if (p1txn == 0)
            throw qpid::broker::InvalidTransactionContextException();
        p1txn->sqlCommit();
        return;
    }

    DatabaseConnection *db(p2txn->dbConn());
    TplRecordset rsTpl;
    MessageMapRecordset rsMessageMap;
    try {
        db->beginTransaction();
        rsTpl.open(db, TblTpl);
        rsMessageMap.open(db, TblMessageMap);
        rsMessageMap.commitPrepared(p2txn->getXid());
        rsTpl.remove(p2txn->getXid());
        db->commitTransaction();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        db->rollbackTransaction();
        throw ADOException("Error committing transaction", e, errs);
    }  
}

void
MSSqlProvider::abort(qpid::broker::TransactionContext& txn)
{
    (void)initState();     // Ensure this thread is initialized
    /*
     * One-phase and non-prepared two-phase transactions simply abort
     * the outer SQL transaction that was begun on begin(). However, prepared
     * two-phase transactions are different - the SQL transaction started
     * on begin() was committed on prepare() so all the SQL records
     * reflecting the enqueue/dequeue actions for the transaction are
     * recorded but with xid markups on them to reflect that they are
     * prepared but not committed. Now go back and remove the markups,
     * deleting those marked for addition.
     */
    AmqpTPCTransaction *p2txn = dynamic_cast<AmqpTPCTransaction*> (&txn);
    if (p2txn == 0 || !p2txn->isPrepared()) {
        AmqpTransaction *p1txn = dynamic_cast<AmqpTransaction*> (&txn);
        if (p1txn == 0)
            throw qpid::broker::InvalidTransactionContextException();
        p1txn->sqlAbort();
        return;
    }

    DatabaseConnection *db(p2txn->dbConn());
    TplRecordset rsTpl;
    MessageMapRecordset rsMessageMap;
    try {
        db->beginTransaction();
        rsTpl.open(db, TblTpl);
        rsMessageMap.open(db, TblMessageMap);
        rsMessageMap.abortPrepared(p2txn->getXid());
        rsTpl.remove(p2txn->getXid());
        db->commitTransaction();
    }
    catch(_com_error &e) {
        std::string errs = db->getErrors();
        db->rollbackTransaction();
        throw ADOException("Error committing transaction", e, errs);
    }  


    (void)initState();     // Ensure this thread is initialized
    AmqpTransaction *atxn = dynamic_cast<AmqpTransaction*> (&txn);
    if (atxn == 0)
        throw qpid::broker::InvalidTransactionContextException();
    atxn->sqlAbort();
}

void
MSSqlProvider::collectPreparedXids(std::set<std::string>& xids)
{
    DatabaseConnection *db = initConnection();
    try {
        TplRecordset rsTpl;
        rsTpl.open(db, TblTpl);
        rsTpl.recover(xids);
    }
    catch(_com_error &e) {
        throw ADOException("Error reading TPL", e, db->getErrors());
    }  
}

// @TODO Much of this recovery code is way too similar... refactor to
// a recover template method on BlobRecordset.

void
MSSqlProvider::recoverConfigs(qpid::broker::RecoveryManager& recoverer)
{
    DatabaseConnection *db = 0;
    try {
        db = initConnection();
        BlobRecordset rsConfigs;
        rsConfigs.open(db, TblConfig);
        _RecordsetPtr p = (_RecordsetPtr)rsConfigs;
        if (p->BOF && p->EndOfFile)
            return;   // Nothing to do
        p->MoveFirst();
        while (!p->EndOfFile) {
            uint64_t id = p->Fields->Item["persistenceId"]->Value;
            long blobSize = p->Fields->Item["fieldTableBlob"]->ActualSize;
            BlobAdapter blob(blobSize);
            blob = p->Fields->Item["fieldTableBlob"]->GetChunk(blobSize);
            // Recreate the Config instance and reset its ID.
            broker::RecoverableConfig::shared_ptr config =
                recoverer.recoverConfig(blob);
            config->setPersistenceId(id);
            p->MoveNext();
        }
    }
    catch(_com_error &e) {
        throw ADOException("Error recovering configs",
                           e,
                           db ? db->getErrors() : "");
    }  
}

void
MSSqlProvider::recoverExchanges(qpid::broker::RecoveryManager& recoverer,
                                ExchangeMap& exchangeMap)
{
    DatabaseConnection *db = 0;
    try {
        db = initConnection();
        BlobRecordset rsExchanges;
        rsExchanges.open(db, TblExchange);
        _RecordsetPtr p = (_RecordsetPtr)rsExchanges;
        if (p->BOF && p->EndOfFile)
            return;   // Nothing to do
        p->MoveFirst();
        while (!p->EndOfFile) {
            uint64_t id = p->Fields->Item["persistenceId"]->Value;
            long blobSize = p->Fields->Item["fieldTableBlob"]->ActualSize;
            BlobAdapter blob(blobSize);
            blob = p->Fields->Item["fieldTableBlob"]->GetChunk(blobSize);
            // Recreate the Exchange instance, reset its ID, and remember the
            // ones restored for matching up when recovering bindings.
            broker::RecoverableExchange::shared_ptr exchange =
                recoverer.recoverExchange(blob);
            exchange->setPersistenceId(id);
            exchangeMap[id] = exchange;
            p->MoveNext();
        }
    }
    catch(_com_error &e) {
        throw ADOException("Error recovering exchanges",
                           e,
                           db ? db->getErrors() : "");
    }  
}

void
MSSqlProvider::recoverQueues(qpid::broker::RecoveryManager& recoverer,
                             QueueMap& queueMap)
{
    DatabaseConnection *db = 0;
    try {
        db = initConnection();
        BlobRecordset rsQueues;
        rsQueues.open(db, TblQueue);
        _RecordsetPtr p = (_RecordsetPtr)rsQueues;
        if (p->BOF && p->EndOfFile)
            return;   // Nothing to do
        p->MoveFirst();
        while (!p->EndOfFile) {
            uint64_t id = p->Fields->Item["persistenceId"]->Value;
            long blobSize = p->Fields->Item["fieldTableBlob"]->ActualSize;
            BlobAdapter blob(blobSize);
            blob = p->Fields->Item["fieldTableBlob"]->GetChunk(blobSize);
            // Recreate the Queue instance and reset its ID.
            broker::RecoverableQueue::shared_ptr queue =
                recoverer.recoverQueue(blob);
            queue->setPersistenceId(id);
            queueMap[id] = queue;
            p->MoveNext();
        }
    }
    catch(_com_error &e) {
        throw ADOException("Error recovering queues",
                           e,
                           db ? db->getErrors() : "");
    }  
}

void
MSSqlProvider::recoverBindings(qpid::broker::RecoveryManager& recoverer,
                               const ExchangeMap& exchangeMap,
                               const QueueMap& queueMap)
{
    DatabaseConnection *db = 0;
    try {
        db = initConnection();
        BindingRecordset rsBindings;
        rsBindings.open(db, TblBinding);
        rsBindings.recover(recoverer, exchangeMap, queueMap);
    }
    catch(_com_error &e) {
        throw ADOException("Error recovering bindings",
                           e,
                           db ? db->getErrors() : "");
    }  
}

void
MSSqlProvider::recoverMessages(qpid::broker::RecoveryManager& recoverer,
                               MessageMap& messageMap,
                               MessageQueueMap& messageQueueMap)
{
    DatabaseConnection *db = 0;
    try {
        db = initConnection();
        MessageRecordset rsMessages;
        rsMessages.open(db, TblMessage);
        rsMessages.recover(recoverer, messageMap);

        MessageMapRecordset rsMessageMaps;
        rsMessageMaps.open(db, TblMessageMap);
        rsMessageMaps.recover(messageQueueMap);
    }
    catch(_com_error &e) {
        throw ADOException("Error recovering messages",
                           e,
                           db ? db->getErrors() : "");
    }  
}

void
MSSqlProvider::recoverTransactions(qpid::broker::RecoveryManager& recoverer,
                                   PreparedTransactionMap& dtxMap)
{
    DatabaseConnection *db = initConnection();
    std::set<std::string> xids;
    try {
        TplRecordset rsTpl;
        rsTpl.open(db, TblTpl);
        rsTpl.recover(xids);
    }
    catch(_com_error &e) {
        throw ADOException("Error recovering TPL records", e, db->getErrors());
    }  

    try {
        // Rebuild the needed RecoverableTransactions.
        for (std::set<std::string>::const_iterator iXid = xids.begin();
             iXid != xids.end();
             ++iXid) {
            boost::shared_ptr<DatabaseConnection> dbX(new DatabaseConnection);
            dbX->open(options.connectString, options.catalogName);
            std::auto_ptr<AmqpTPCTransaction> tx(new AmqpTPCTransaction(dbX,
                                                                        *iXid));
            tx->setPrepared();
            std::auto_ptr<qpid::broker::TPCTransactionContext> tc(tx);
            dtxMap[*iXid] = recoverer.recoverTransaction(*iXid, tc);
        }
    }
    catch(_com_error &e) {
        throw ADOException("Error recreating dtx connection", e);
    }  
}

////////////// Internal Methods

State *
MSSqlProvider::initState()
{
    State *state = dbState.get();   // See if thread has initialized
    if (!state) {
        state = new State;
        dbState.reset(state);
    }
    return state;
}
  
DatabaseConnection *
MSSqlProvider::initConnection(void)
{
    State *state = initState();
    if (state->dbConn != 0)
        return state->dbConn;    // And the DatabaseConnection is set up too
    std::auto_ptr<DatabaseConnection> db(new DatabaseConnection);
    db->open(options.connectString, options.catalogName);
    state->dbConn = db.release();
    return state->dbConn;
}

void
MSSqlProvider::createDb(DatabaseConnection *db, const std::string &name)
{
    const std::string dbCmd = "CREATE DATABASE " + name;
    const std::string useCmd = "USE " + name;
    const std::string tableCmd = "CREATE TABLE ";
    const std::string colSpecs =
        " (persistenceId bigint PRIMARY KEY NOT NULL IDENTITY(1,1),"
        "  fieldTableBlob varbinary(MAX) NOT NULL)";
    const std::string bindingSpecs =
        " (exchangeId bigint REFERENCES tblExchange(persistenceId) NOT NULL,"
        "  queueId bigint REFERENCES tblQueue(persistenceId) NOT NULL,"
        "  routingKey varchar(255),"
        "  fieldTableBlob varbinary(MAX))";
    const std::string messageMapSpecs =
        " (messageId bigint REFERENCES tblMessage(persistenceId) NOT NULL,"
        "  queueId bigint REFERENCES tblQueue(persistenceId) NOT NULL,"
        "  prepareStatus tinyint CHECK (prepareStatus IS NULL OR "
        "    prepareStatus IN (1, 2)),"
        "  xid varbinary(512) REFERENCES tblTPL(xid)"
        "  CONSTRAINT CK_NoDups UNIQUE NONCLUSTERED (messageId, queueId) )";
    const std::string tplSpecs = " (xid varbinary(512) PRIMARY KEY NOT NULL)";
    // SET NOCOUNT ON added to prevent extra result sets from
    // interfering with SELECT statements. (Added by SQL Management)
    const std::string removeUnrefMsgsTrigger =
        "CREATE TRIGGER dbo.RemoveUnreferencedMessages "
        "ON  tblMessageMap AFTER DELETE AS BEGIN "
        "SET NOCOUNT ON; "
        "DELETE FROM tblMessage "
        "WHERE tblMessage.persistenceId IN "
        "  (SELECT messageId FROM deleted) AND"
        "  NOT EXISTS(SELECT * FROM tblMessageMap"
        "             WHERE tblMessageMap.messageId IN"
        "               (SELECT messageId FROM deleted)) "
        "END";

    _variant_t unused;
    _bstr_t dbStr = dbCmd.c_str();
    _ConnectionPtr conn(*db);
    try {
        conn->Execute(dbStr, &unused, adExecuteNoRecords);
        _bstr_t useStr = useCmd.c_str();
        conn->Execute(useStr, &unused, adExecuteNoRecords);
        std::string makeTable = tableCmd + TblQueue + colSpecs;
        _bstr_t makeTableStr = makeTable.c_str();
        conn->Execute(makeTableStr, &unused, adExecuteNoRecords);
        makeTable = tableCmd + TblExchange + colSpecs;
        makeTableStr = makeTable.c_str();
        conn->Execute(makeTableStr, &unused, adExecuteNoRecords);
        makeTable = tableCmd + TblConfig + colSpecs;
        makeTableStr = makeTable.c_str();
        conn->Execute(makeTableStr, &unused, adExecuteNoRecords);
        makeTable = tableCmd + TblMessage + colSpecs;
        makeTableStr = makeTable.c_str();
        conn->Execute(makeTableStr, &unused, adExecuteNoRecords);
        makeTable = tableCmd + TblBinding + bindingSpecs;
        makeTableStr = makeTable.c_str();
        conn->Execute(makeTableStr, &unused, adExecuteNoRecords);
        makeTable = tableCmd + TblTpl + tplSpecs;
        makeTableStr = makeTable.c_str();
        conn->Execute(makeTableStr, &unused, adExecuteNoRecords);
        makeTable = tableCmd + TblMessageMap + messageMapSpecs;
        makeTableStr = makeTable.c_str();
        conn->Execute(makeTableStr, &unused, adExecuteNoRecords);
        _bstr_t addTriggerStr = removeUnrefMsgsTrigger.c_str();
        conn->Execute(addTriggerStr, &unused, adExecuteNoRecords);
    }
    catch(_com_error &e) {
        throw ADOException("MSSQL can't create " + name, e, db->getErrors());
    }
}

void
MSSqlProvider::dump()
{
  // dump all db records to qpid_log
  QPID_LOG(notice, "DB Dump: (not dumping anything)");
  //  rsQueues.dump();
}


}}} // namespace qpid::store::ms_sql