/**
 *    Copyright (C) 2008 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.
 */

#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kDefault

#include "mongo/platform/basic.h"

#include "mongo/db/introspect.h"

#include "mongo/bson/util/builder.h"
#include "mongo/db/auth/authorization_manager.h"
#include "mongo/db/auth/authorization_session.h"
#include "mongo/db/auth/user_set.h"
#include "mongo/db/curop.h"
#include "mongo/db/jsobj.h"
#include "mongo/db/catalog/collection.h"
#include "mongo/util/log.h"

namespace mongo {

    using boost::scoped_ptr;
    using std::endl;
    using std::string;

namespace {

    void _appendUserInfo(const CurOp& c,
                         BSONObjBuilder& builder,
                         AuthorizationSession* authSession) {
        UserNameIterator nameIter = authSession->getAuthenticatedUserNames();

        UserName bestUser;
        if (nameIter.more())
            bestUser = *nameIter;

        std::string opdb( nsToDatabase( c.getNS() ) );

        BSONArrayBuilder allUsers(builder.subarrayStart("allUsers"));
        for ( ; nameIter.more(); nameIter.next()) {
            BSONObjBuilder nextUser(allUsers.subobjStart());
            nextUser.append(AuthorizationManager::USER_NAME_FIELD_NAME, nameIter->getUser());
            nextUser.append(AuthorizationManager::USER_DB_FIELD_NAME, nameIter->getDB());
            nextUser.doneFast();

            if (nameIter->getDB() == opdb) {
                bestUser = *nameIter;
            }
        }
        allUsers.doneFast();

        builder.append("user", bestUser.getUser().empty() ? "" : bestUser.getFullName());

    }

} // namespace


    void profile(OperationContext* txn, int op) {
        // Initialize with 1kb at start in order to avoid realloc later
        BufBuilder profileBufBuilder(1024);

        BSONObjBuilder b(profileBufBuilder);

        {
            Locker::LockerInfo lockerInfo;
            txn->lockState()->getLockerInfo(&lockerInfo);
            txn->getCurOp()->debug().append(*txn->getCurOp(), lockerInfo.stats, b);
        }

        b.appendDate("ts", jsTime());
        b.append("client", txn->getClient()->clientAddress());

        AuthorizationSession * authSession = txn->getClient()->getAuthorizationSession();
        _appendUserInfo(*txn->getCurOp(), b, authSession);

        const BSONObj p = b.done();

        const bool wasLocked = txn->lockState()->isLocked();

        const string dbName(nsToDatabase(txn->getCurOp()->getNS()));

        try {
            bool acquireDbXLock = false;
            while (true) {
                ScopedTransaction scopedXact(txn, MODE_IX);

                boost::scoped_ptr<AutoGetDb> autoGetDb;
                if (acquireDbXLock) {
                    autoGetDb.reset(new AutoGetDb(txn, dbName, MODE_X));
                    if (autoGetDb->getDb()) {
                        createProfileCollection(txn, autoGetDb->getDb());
                    }
                }
                else {
                    autoGetDb.reset(new AutoGetDb(txn, dbName, MODE_IX));
                }

                Database* const db = autoGetDb->getDb();
                if (!db) {
                    // Database disappeared
                    log() << "note: not profiling because db went away for "
                          << txn->getCurOp()->getNS();
                    break;
                }

                Lock::CollectionLock collLock(txn->lockState(), db->getProfilingNS(), MODE_IX);

                Collection* const coll = db->getCollection(db->getProfilingNS());
                if (coll) {
                    WriteUnitOfWork wuow(txn);
                    coll->insertDocument(txn, p, false);
                    wuow.commit();

                    break;
                }
                else if (!acquireDbXLock &&
                            (!wasLocked || txn->lockState()->isDbLockedForMode(dbName, MODE_X))) {
                    // Try to create the collection only if we are not under lock, in order to
                    // avoid deadlocks due to lock conversion. This would only be hit if someone
                    // deletes the profiler collection after setting profile level.
                    acquireDbXLock = true;
                }
                else {
                    // Cannot write the profile information
                    break;
                }
            }
        }
        catch (const AssertionException& assertionEx) {
            warning() << "Caught Assertion while trying to profile "
                      << opToString(op)
                      << " against " << txn->getCurOp()->getNS()
                      << ": " << assertionEx.toString() << endl;
        }
    }


    Status createProfileCollection(OperationContext* txn, Database *db) {
        invariant(txn->lockState()->isDbLockedForMode(db->name(), MODE_X));

        const std::string dbProfilingNS(db->getProfilingNS());

        Collection* const collection = db->getCollection(dbProfilingNS);
        if (collection) {
            if (!collection->isCapped()) {
                return Status(ErrorCodes::NamespaceExists,
                              str::stream() << dbProfilingNS << " exists but isn't capped");
            }

            return Status::OK();
        }

        // system.profile namespace doesn't exist; create it
        log() << "Creating profile collection: " << dbProfilingNS << endl;

        CollectionOptions collectionOptions;
        collectionOptions.capped = true;
        collectionOptions.cappedSize = 1024 * 1024;

        WriteUnitOfWork wunit(txn);
        invariant(db->createCollection(txn, dbProfilingNS, collectionOptions));
        wunit.commit();

        return Status::OK();
    }

} // namespace mongo