/**
 *    Copyright (C) 2014 MongoDB Inc.
 *
 *    This program is free software: you can redistribute it and/or  modify
 *    it under the terms of the GNU Affero General Public License, version 3,
 *    as published by the Free Software Foundation.
 *
 *    This program 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 Affero General Public License for more details.
 *
 *    You should have received a copy of the GNU Affero General Public License
 *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 *    As a special exception, the copyright holders give permission to link the
 *    code of portions of this program with the OpenSSL library under certain
 *    conditions as described in each individual source file and distribute
 *    linked combinations including the program with the OpenSSL library. You
 *    must comply with the GNU Affero General Public License in all respects for
 *    all of the code used other than as permitted herein. If you modify file(s)
 *    with this exception, you may extend this exception to your version of the
 *    file(s), but you are not obligated to do so. If you do not wish to do so,
 *    delete this exception statement from your version. If you delete this
 *    exception statement from all source files in the program, then also delete
 *    it in the license file.
 */

#include "mongo/platform/basic.h"

#include "mongo/db/db_raii.h"

#include "mongo/db/catalog/database_holder.h"
#include "mongo/db/client.h"
#include "mongo/db/curop.h"
#include "mongo/db/stats/top.h"
#include "mongo/s/d_state.h"

namespace mongo {

AutoGetDb::AutoGetDb(OperationContext* txn, StringData ns, LockMode mode)
    : _dbLock(txn->lockState(), ns, mode), _db(dbHolder().get(txn, ns)) {}

AutoGetOrCreateDb::AutoGetOrCreateDb(OperationContext* txn, StringData ns, LockMode mode)
    : _transaction(txn, MODE_IX),
      _dbLock(txn->lockState(), ns, mode),
      _db(dbHolder().get(txn, ns)) {
    invariant(mode == MODE_IX || mode == MODE_X);
    _justCreated = false;
    // If the database didn't exist, relock in MODE_X
    if (_db == NULL) {
        if (mode != MODE_X) {
            _dbLock.relockWithMode(MODE_X);
        }
        _db = dbHolder().openDb(txn, ns);
        _justCreated = true;
    }
}

AutoGetCollectionForRead::AutoGetCollectionForRead(OperationContext* txn, const std::string& ns)
    : _txn(txn),
      _transaction(txn, MODE_IS),
      _db(_txn, nsToDatabaseSubstring(ns), MODE_IS),
      _collLock(_txn->lockState(), ns, MODE_IS),
      _coll(NULL) {
    _init(ns, nsToCollectionSubstring(ns));
}

AutoGetCollectionForRead::AutoGetCollectionForRead(OperationContext* txn,
                                                   const NamespaceString& nss)
    : _txn(txn),
      _transaction(txn, MODE_IS),
      _db(_txn, nss.db(), MODE_IS),
      _collLock(_txn->lockState(), nss.toString(), MODE_IS),
      _coll(NULL) {
    _init(nss.toString(), nss.coll());
}

void AutoGetCollectionForRead::_init(const std::string& ns, StringData coll) {
    massert(28535, "need a non-empty collection name", !coll.empty());

    // We have both the DB and collection locked, which the prerequisite to do a stable shard
    // version check.
    ensureShardVersionOKOrThrow(_txn->getClient(), ns);

    auto curOp = CurOp::get(_txn);
    stdx::lock_guard<Client> lk(*_txn->getClient());
    // TODO: OldClientContext legacy, needs to be removed
    curOp->ensureStarted();
    curOp->setNS_inlock(ns);

    // At this point, we are locked in shared mode for the database by the DB lock in the
    // constructor, so it is safe to load the DB pointer.
    if (_db.getDb()) {
        // TODO: OldClientContext legacy, needs to be removed
        curOp->enter_inlock(ns.c_str(), _db.getDb()->getProfilingLevel());

        _coll = _db.getDb()->getCollection(ns);
    }
}

AutoGetCollectionForRead::~AutoGetCollectionForRead() {
    // Report time spent in read lock
    auto currentOp = CurOp::get(_txn);
    Top::get(_txn->getClient()->getServiceContext())
        .record(currentOp->getNS(),
                currentOp->getOp(),
                -1,  // "read locked"
                _timer.micros(),
                currentOp->isCommand());
}


OldClientContext::OldClientContext(OperationContext* txn, const std::string& ns, Database* db)
    : _justCreated(false), _doVersion(true), _ns(ns), _db(db), _txn(txn) {}

OldClientContext::OldClientContext(OperationContext* txn,
                                   const std::string& ns,
                                   Database* db,
                                   bool justCreated)
    : _justCreated(justCreated), _doVersion(true), _ns(ns), _db(db), _txn(txn) {
    _finishInit();
}

OldClientContext::OldClientContext(OperationContext* txn,
                                   const std::string& ns,
                                   bool doVersion)
    : _justCreated(false),  // set for real in finishInit
      _doVersion(doVersion),
      _ns(ns),
      _db(NULL),
      _txn(txn) {
    _finishInit();
}

void OldClientContext::_finishInit() {
    _db = dbHolder().get(_txn, _ns);
    if (_db) {
        _justCreated = false;
    } else {
        invariant(_txn->lockState()->isDbLockedForMode(nsToDatabaseSubstring(_ns), MODE_X));
        _db = dbHolder().openDb(_txn, _ns, &_justCreated);
        invariant(_db);
    }

    if (_doVersion) {
        _checkNotStale();
    }

    stdx::lock_guard<Client> lk(*_txn->getClient());
    CurOp::get(_txn)->enter_inlock(_ns.c_str(), _db->getProfilingLevel());
}

void OldClientContext::_checkNotStale() const {
    switch (CurOp::get(_txn)->getOp()) {
        case dbGetMore:  // getMore is special and should be handled elsewhere.
        case dbUpdate:   // update & delete check shard version in instance.cpp, so don't check
        case dbDelete:   // here as well.
            break;
        default:
            ensureShardVersionOKOrThrow(_txn->getClient(), _ns);
    }
}

OldClientContext::~OldClientContext() {
    // Lock must still be held
    invariant(_txn->lockState()->isLocked());

    auto currentOp = CurOp::get(_txn);
    Top::get(_txn->getClient()->getServiceContext())
        .record(currentOp->getNS(),
                currentOp->getOp(),
                _txn->lockState()->isWriteLocked() ? 1 : -1,
                _timer.micros(),
                currentOp->isCommand());
}


OldClientWriteContext::OldClientWriteContext(OperationContext* opCtx, const std::string& ns)
    : _txn(opCtx),
      _nss(ns),
      _autodb(opCtx, _nss.db(), MODE_IX),
      _collk(opCtx->lockState(), ns, MODE_IX),
      _c(opCtx, ns, _autodb.getDb(), _autodb.justCreated()) {
    _collection = _c.db()->getCollection(ns);
    if (!_collection && !_autodb.justCreated()) {
        // relock database in MODE_X to allow collection creation
        _collk.relockAsDatabaseExclusive(_autodb.lock());
        Database* db = dbHolder().get(_txn, ns);
        invariant(db == _c.db());
    }
}

}  // namespace mongo