/**
 *    Copyright (C) 2018-present MongoDB, Inc.
 *
 *    This program is free software: you can redistribute it and/or modify
 *    it under the terms of the Server Side Public License, version 1,
 *    as published by MongoDB, Inc.
 *
 *    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
 *    Server Side Public License for more details.
 *
 *    You should have received a copy of the Server Side Public License
 *    along with this program. If not, see
 *    <http://www.mongodb.com/licensing/server-side-public-license>.
 *
 *    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 Server Side 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.
 */

#pragma once

#include "mongo/base/disallow_copying.h"
#include "mongo/db/logical_session_id.h"
#include "mongo/db/session.h"
#include "mongo/db/session_killer.h"
#include "mongo/db/sessions_collection.h"
#include "mongo/stdx/condition_variable.h"
#include "mongo/stdx/mutex.h"
#include "mongo/stdx/unordered_map.h"

namespace mongo {

class OperationContext;
class ScopedSession;
class ScopedCheckedOutSession;
class ServiceContext;

/**
 * Keeps track of the transaction runtime state for every active session on this instance.
 */
class SessionCatalog {
    MONGO_DISALLOW_COPYING(SessionCatalog);

    friend class ScopedSession;
    friend class ScopedCheckedOutSession;

public:
    SessionCatalog() = default;
    ~SessionCatalog();

    /**
     * Retrieves the session transaction table associated with the service or operation context.
     */
    static SessionCatalog* get(OperationContext* opCtx);
    static SessionCatalog* get(ServiceContext* service);

    /**
     * Fetches the UUID of the transaction table, or an empty optional if the collection does not
     * exist or has no UUID. Acquires a lock on the collection. Required for rollback via refetch.
     */
    static boost::optional<UUID> getTransactionTableUUID(OperationContext* opCtx);

    /**
     * Locates session entries from the in-memory catalog and in 'config.transactions' which have
     * not been referenced before 'possiblyExpired' and deletes them.
     *
     * Returns the number of sessions, which were reaped from the persisted store on disk.
     */
    static int reapSessionsOlderThan(OperationContext* opCtx,
                                     SessionsCollection& sessionsCollection,
                                     Date_t possiblyExpired);

    /**
     * Resets the transaction table to an uninitialized state.
     * Meant only for testing.
     */
    void reset_forTest();

    /**
     * Invoked when the node enters the primary state. Ensures that the transactions collection is
     * created. Throws on severe exceptions due to which it is not safe to continue the step-up
     * process.
     */
    void onStepUp(OperationContext* opCtx);

    /**
     * Potentially blocking call, which uses the session information stored in the specified
     * operation context and either creates a new session runtime state (if one doesn't exist) or
     * "checks-out" the existing one (if it is not currently in use).
     *
     * Checking out a session puts it in the 'in use' state and all subsequent calls to checkout
     * will block until it is put back in the 'available' state when the returned object goes out of
     * scope.
     *
     * Throws exception on errors.
     */
    ScopedCheckedOutSession checkOutSession(OperationContext* opCtx);

    /**
     * Returns a reference to the specified cached session regardless of whether it is checked-out
     * or not. The returned session is not returned checked-out and is allowed to be checked-out
     * concurrently.
     *
     * The intended usage for this method is to allow migrations to run in parallel with writes for
     * the same session without blocking it. Because of this, it may not be used from operations
     * which run on a session.
     */
    ScopedSession getOrCreateSession(OperationContext* opCtx, const LogicalSessionId& lsid);

    /**
     * Callback to be invoked when it is suspected that the on-disk session contents might not be in
     * sync with what is in the sessions cache.
     *
     * If no specific document is available, the method will invalidate all sessions. Otherwise if
     * one is avaiable (which is the case for insert/update/delete), it must contain _id field with
     * a valid session entry, in which case only that particular session will be invalidated. If the
     * _id field is missing or doesn't contain a valid serialization of logical session, the method
     * will throw. This prevents invalid entries from making it in the collection.
     */
    void invalidateSessions(OperationContext* opCtx, boost::optional<BSONObj> singleSessionDoc);

    /**
     * Iterates through the SessionCatalog and applies 'workerFn' to each Session. This locks the
     * SessionCatalog.
     * TODO SERVER-33850: Take Matcher out of the SessionKiller namespace.
     */
    void scanSessions(OperationContext* opCtx,
                      const SessionKiller::Matcher& matcher,
                      stdx::function<void(OperationContext*, Session*)> workerFn);

    /**
     * Returns the total number of entries currently cached on the session catalog.
     */
    size_t size() const;

private:
    struct SessionRuntimeInfo {
        SessionRuntimeInfo(LogicalSessionId lsid) : txnState(std::move(lsid)) {}

        // Current check-out state of the session. If set to false, the session can be checked out.
        // If set to true, the session is in use by another operation and the caller must wait to
        // check it out.
        bool checkedOut{false};

        // Keeps the last time this session was checked-out
        Date_t lastCheckout{Date_t::now()};

        // Signaled when the state becomes available. Uses the transaction table's mutex to protect
        // the state transitions.
        stdx::condition_variable availableCondVar;

        // Must only be accessed when the state is kInUse and only by the operation context, which
        // currently has it checked out
        Session txnState;
    };

    using SessionRuntimeInfoMap = LogicalSessionIdMap<std::shared_ptr<SessionRuntimeInfo>>;

    /**
     * May release and re-acquire it zero or more times before returning. The returned
     * 'SessionRuntimeInfo' is guaranteed to be linked on the catalog's _txnTable as long as the
     * lock is held.
     */
    std::shared_ptr<SessionRuntimeInfo> _getOrCreateSessionRuntimeInfo(
        WithLock, OperationContext* opCtx, const LogicalSessionId& lsid);

    /**
     * Makes a session, previously checked out through 'checkoutSession', available again.
     */
    void _releaseSession(const LogicalSessionId& lsid);

    void _invalidateSession(WithLock, SessionRuntimeInfoMap::iterator it);

    /**
     * Snapshots the set of in-memory sessions currently on the catalog, checks whether they are
     * still in use though the passed 'sessionsCollection' and if any have expired, calls
     * 'invalidateSession' on them.
     */
    void _reapInMemorySessionsOlderThan(OperationContext* opCtx,
                                        SessionsCollection& sessionsCollection,
                                        Date_t possiblyExpired);

    // Protects the state below
    mutable stdx::mutex _mutex;

    SessionRuntimeInfoMap _txnTable;
};

/**
 * Scoped object representing a reference to a session.
 */
class ScopedSession {
public:
    explicit ScopedSession(std::shared_ptr<SessionCatalog::SessionRuntimeInfo> sri)
        : _sri(std::move(sri)) {
        invariant(_sri);
    }

    Session* get() const {
        return &_sri->txnState;
    }

    Session* operator->() const {
        return get();
    }

    Session& operator*() const {
        return *get();
    }

    operator bool() const {
        return !!_sri;
    }

private:
    std::shared_ptr<SessionCatalog::SessionRuntimeInfo> _sri;
};

/**
 * Scoped object representing a checked-out session. See comments for the 'checkoutSession' method
 * for more information on its behaviour.
 */
class ScopedCheckedOutSession {
    MONGO_DISALLOW_COPYING(ScopedCheckedOutSession);

    friend ScopedCheckedOutSession SessionCatalog::checkOutSession(OperationContext*);

public:
    ScopedCheckedOutSession(ScopedCheckedOutSession&&) = default;

    ~ScopedCheckedOutSession() {
        if (_scopedSession) {
            SessionCatalog::get(_opCtx)->_releaseSession(_scopedSession->getSessionId());
        }
    }

    Session* get() const {
        return _scopedSession.get();
    }

    Session* operator->() const {
        return get();
    }

    Session& operator*() const {
        return *get();
    }

    operator bool() const {
        return _scopedSession;
    }

private:
    ScopedCheckedOutSession(OperationContext* opCtx, ScopedSession scopedSession)
        : _opCtx(opCtx), _scopedSession(std::move(scopedSession)) {}

    OperationContext* const _opCtx;

    ScopedSession _scopedSession;
};

/**
 * Scoped object, which checks out the session specified in the passed operation context and stores
 * it for later access by the command. The session is installed at construction time and is removed
 * at destruction.
 */
class OperationContextSession {
    MONGO_DISALLOW_COPYING(OperationContextSession);

public:
    OperationContextSession(OperationContext* opCtx, bool checkOutSession);

    ~OperationContextSession();

    /**
     * Returns the session checked out in the constructor.
     */
    static Session* get(OperationContext* opCtx);

private:
    OperationContext* const _opCtx;
};

}  // namespace mongo