// projection.cpp

/*    Copyright 2009 10gen 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/projection.h"

#include "mongo/db/matcher/matcher.h"
#include "mongo/util/log.h"
#include "mongo/util/mongoutils/str.h"

namespace mongo {

    MONGO_LOG_DEFAULT_COMPONENT_FILE(::mongo::logger::LogComponent::kQuery);

    void Projection::init(const BSONObj& o,
                          const MatchExpressionParser::WhereCallback& whereCallback) {
        massert( 10371 , "can only add to Projection once", _source.isEmpty());
        _source = o;

        BSONObjIterator i( o );
        int true_false = -1;
        while ( i.more() ) {
            BSONElement e = i.next();

            if ( ! e.isNumber() )
                _hasNonSimple = true;

            if (e.type() == Object) {
                BSONObj obj = e.embeddedObject();
                BSONElement e2 = obj.firstElement();
                if ( mongoutils::str::equals( e2.fieldName(), "$slice" ) ) {
                    if (e2.isNumber()) {
                        int i = e2.numberInt();
                        if (i < 0)
                            add(e.fieldName(), i, -i); // limit is now positive
                        else
                            add(e.fieldName(), 0, i);

                    }
                    else if (e2.type() == Array) {
                        BSONObj arr = e2.embeddedObject();
                        uassert(13099, "$slice array wrong size", arr.nFields() == 2 );

                        BSONObjIterator it(arr);
                        int skip = it.next().numberInt();
                        int limit = it.next().numberInt();
                        uassert(13100, "$slice limit must be positive", limit > 0 );
                        add(e.fieldName(), skip, limit);

                    }
                    else {
                        uassert(13098, "$slice only supports numbers and [skip, limit] arrays", false);
                    }
                }
                else if ( mongoutils::str::equals( e2.fieldName(), "$elemMatch" ) ) {
                    // validate $elemMatch arguments and dependencies
                    uassert( 16342, "elemMatch: invalid argument.  object required.",
                             e2.type() == Object );
                    uassert( 16343, "Cannot specify positional operator and $elemMatch"
                                    " (currently unsupported).",
                             _arrayOpType != ARRAY_OP_POSITIONAL );
                    uassert( 16344, "Cannot use $elemMatch projection on a nested field"
                                    " (currently unsupported).",
                             ! mongoutils::str::contains( e.fieldName(), '.' ) );
                    _arrayOpType = ARRAY_OP_ELEM_MATCH;

                    // initialize new Matcher object(s)

                    _matchers[mongoutils::str::before(e.fieldName(), '.').c_str()]
                        = boost::make_shared<Matcher>(e.wrap(), whereCallback);
                    add( e.fieldName(), true );
                }
                else {
                    uasserted(13097, string("Unsupported projection option: ") +
                                     obj.firstElementFieldName() );
                }

            }
            else if (!strcmp(e.fieldName(), "_id") && !e.trueValue()) {
                _includeID = false;
            }
            else {
                add( e.fieldName(), e.trueValue() );

                // validate input
                if (true_false == -1) {
                    true_false = e.trueValue();
                    _include = !e.trueValue();
                }
                else {
                    uassert( 10053 , "You cannot currently mix including and excluding fields. "
                                     "Contact us if this is an issue." ,
                             (bool)true_false == e.trueValue() );
                }
            }
            if ( mongoutils::str::contains( e.fieldName(), ".$" ) ) {
                // positional op found; verify dependencies
                uassert( 16345, "Cannot exclude array elements with the positional operator"
                                " (currently unsupported).", e.trueValue() );
                uassert( 16346, "Cannot specify more than one positional array element per query"
                                " (currently unsupported).", _arrayOpType != ARRAY_OP_POSITIONAL );
                uassert( 16347, "Cannot specify positional operator and $elemMatch"
                                " (currently unsupported).", _arrayOpType != ARRAY_OP_ELEM_MATCH );
                _arrayOpType = ARRAY_OP_POSITIONAL;
            }
        }
    }

    void Projection::add(const string& field, bool include) {
        if (field.empty()) { // this is the field the user referred to
            _include = include;
        }
        else {
            _include = !include;

            const size_t dot = field.find('.');
            const string subfield = field.substr(0,dot);
            const string rest = (dot == string::npos ? "" : field.substr(dot+1,string::npos));

            boost::shared_ptr<Projection>& fm = _fields[subfield.c_str()];
            if (!fm)
                fm = boost::make_shared<Projection>();

            fm->add(rest, include);
        }
    }

    void Projection::add(const string& field, int skip, int limit) {
        _special = true; // can't include or exclude whole object

        if (field.empty()) { // this is the field the user referred to
            _skip = skip;
            _limit = limit;
        }
        else {
            const size_t dot = field.find('.');
            const string subfield = field.substr(0,dot);
            const string rest = (dot == string::npos ? "" : field.substr(dot+1,string::npos));

            boost::shared_ptr<Projection>& fm = _fields[subfield.c_str()];
            if (!fm)
                fm = boost::make_shared<Projection>();

            fm->add(rest, skip, limit);
        }
    }

    void Projection::transform( const BSONObj& in , BSONObjBuilder& b, const MatchDetails* details ) const {
        const ArrayOpType& arrayOpType = getArrayOpType();

        BSONObjIterator i(in);
        while ( i.more() ) {
            BSONElement e = i.next();
            if ( mongoutils::str::equals( "_id" , e.fieldName() ) ) {
                if ( _includeID )
                    b.append( e );
            }
            else {
                Matchers::const_iterator matcher = _matchers.find( e.fieldName() );
                if ( matcher == _matchers.end() ) {
                    // no array projection matchers for this field
                    append( b, e, details, arrayOpType );
                } else {
                    // field has array projection with $elemMatch specified.
                    massert( 16348, "matchers are only supported for $elemMatch", 
                             arrayOpType == ARRAY_OP_ELEM_MATCH );
                    MatchDetails arrayDetails;
                    arrayDetails.requestElemMatchKey();
                    if ( matcher->second->matches( in, &arrayDetails ) ) {
                        LOG(4) << "Matched array on field: " << matcher->first  << endl
                               << " from array: " << in.getField( matcher->first ) << endl
                               << " in object: " << in << endl
                               << " at position: " << arrayDetails.elemMatchKey() << endl;
                        FieldMap::const_iterator field = _fields.find( e.fieldName()  );
                        massert( 16349, "$elemMatch specified, but projection field not found.",
                            field != _fields.end() );
                        BSONArrayBuilder a;
                        BSONObjBuilder o;
                        massert( 16350, "$elemMatch called on document element with eoo",
                                 ! in.getField( e.fieldName() ).eoo() );
                        massert( 16351, "$elemMatch called on array element with eoo",
                                 ! in.getField( e.fieldName() ).Obj().getField(
                                        arrayDetails.elemMatchKey() ).eoo() );
                        a.append( in.getField( e.fieldName() ).Obj()
                                    .getField( arrayDetails.elemMatchKey() ) );
                        o.appendArray( matcher->first, a.arr() );
                        append( b, o.done().firstElement(), details, arrayOpType );
                    }
                }
            }
        }
    }

    BSONObj Projection::transform( const BSONObj& in, const MatchDetails* details ) const {
        BSONObjBuilder b;
        transform( in , b, details );
        return b.obj();
    }


    //b will be the value part of an array-typed BSONElement
    void Projection::appendArray( BSONObjBuilder& b , const BSONObj& a , bool nested) const {
        int skip  = nested ?  0 : _skip;
        int limit = nested ? -1 : _limit;

        if (skip < 0) {
            skip = max(0, skip + a.nFields());
        }

        int i=0;
        BSONObjIterator it(a);
        while (it.more()) {
            BSONElement e = it.next();

            if (skip) {
                skip--;
                continue;
            }

            if (limit != -1 && (limit-- == 0)) {
                break;
            }

            switch(e.type()) {
            case Array: {
                BSONObjBuilder subb;
                appendArray(subb , e.embeddedObject(), true);
                b.appendArray(b.numStr(i++), subb.obj());
                break;
            }
            case Object: {
                BSONObjBuilder subb;
                BSONObjIterator jt(e.embeddedObject());
                while (jt.more()) {
                    append(subb , jt.next());
                }
                b.append(b.numStr(i++), subb.obj());
                break;
            }
            default:
                if (_include)
                    b.appendAs(e, b.numStr(i++));
            }
        }
    }

    void Projection::append( BSONObjBuilder& b , const BSONElement& e, const MatchDetails* details,
                             const ArrayOpType arrayOpType ) const {

        FieldMap::const_iterator field = _fields.find( e.fieldName() );
        if (field == _fields.end()) {
            if (_include)
                b.append(e);
        }
        else {
            Projection& subfm = *field->second;
            if ( ( subfm._fields.empty() && !subfm._special ) ||
                 !(e.type()==Object || e.type()==Array) ) {
                // field map empty, or element is not an array/object
                if (subfm._include)
                    b.append(e);
            }
            else if (e.type() == Object) {
                BSONObjBuilder subb;
                BSONObjIterator it(e.embeddedObject());
                while (it.more()) {
                    subfm.append(subb, it.next(), details, arrayOpType);
                }
                b.append(e.fieldName(), subb.obj());
            }
            else { //Array
                BSONObjBuilder matchedBuilder;
                if ( details && arrayOpType == ARRAY_OP_POSITIONAL ) {
                    // $ positional operator specified

                    LOG(4) << "projection: checking if element " << e << " matched spec: "
                           << getSpec() << " match details: " << *details << endl;
                    uassert( 16352, mongoutils::str::stream() << "positional operator ("
                                        << e.fieldName()
                                        << ".$) requires corresponding field in query specifier",
                                   details && details->hasElemMatchKey() );

                    uassert( 16353, "positional operator element mismatch",
                             ! e.embeddedObject()[details->elemMatchKey()].eoo() );

                    // append as the first and only element in the projected array
                    matchedBuilder.appendAs( e.embeddedObject()[details->elemMatchKey()], "0" );
                }
                else {
                    // append exact array; no subarray matcher specified
                    subfm.appendArray( matchedBuilder, e.embeddedObject() );
                }
                b.appendArray( e.fieldName(), matchedBuilder.obj() );
            }
        }
    }

    Projection::ArrayOpType Projection::getArrayOpType( ) const {
        return _arrayOpType;
    }

    void Projection::validateQuery( const BSONObj query ) const {
        // this function only validates positional operator ($) projections
        if ( getArrayOpType() != ARRAY_OP_POSITIONAL )
            return;

        BSONObjIterator querySpecIter( query );
        while ( querySpecIter.more() ) {
            // for each query element

            BSONElement queryElement = querySpecIter.next();
            if ( mongoutils::str::equals( queryElement.fieldName(), "$and" ) )
                // don't check $and to avoid deep comparison of the arguments.
                // TODO: can be replaced with Matcher::FieldSink when complete (SERVER-4644)
                return;

            BSONObjIterator projectionSpecIter( getSpec() );
            while ( projectionSpecIter.more() ) {
                // for each projection element

                BSONElement projectionElement = projectionSpecIter.next();
                if ( mongoutils::str::contains( projectionElement.fieldName(), ".$" ) &&
                        mongoutils::str::before( queryElement.fieldName(), '.' ) ==
                        mongoutils::str::before( projectionElement.fieldName(), "." ) ) {

                    // found query spec that matches positional array projection spec
                    LOG(4) << "Query specifies field named for positional operator: "
                           << queryElement.fieldName() << endl;
                    return;
                }
            }
        }

        uasserted( 16354, "Positional operator does not match the query specifier." );
    }

    Projection::KeyOnly* Projection::checkKey( const BSONObj& keyPattern ) const {
        if ( _include ) {
            // if we default to including then we can't
            // use an index because we don't know what we're missing
            return 0;
        }

        if ( _hasNonSimple )
            return 0;

        if ( _includeID && keyPattern["_id"].eoo() )
            return 0;

        // at this point we know its all { x : 1 } style

        auto_ptr<KeyOnly> p( new KeyOnly() );

        int got = 0;
        BSONObjIterator i( keyPattern );
        while ( i.more() ) {
            BSONElement k = i.next();

            if ( _source[k.fieldName()].type() ) {

                if ( strchr( k.fieldName() , '.' ) ) {
                    // TODO we currently don't support dotted fields
                    //      SERVER-2104
                    return 0;
                }

                if ( ! _includeID && mongoutils::str::equals( k.fieldName() , "_id" ) ) {
                    p->addNo();
                }
                else {
                    p->addYes( k.fieldName() );
                    got++;
                }
            }
            else if ( mongoutils::str::equals( "_id" , k.fieldName() ) && _includeID ) {
                p->addYes( "_id" );
            }
            else {
                p->addNo();
            }

        }

        int need = _source.nFields();
        if ( ! _includeID )
            need--;

        if ( got == need )
            return p.release();

        return 0;
    }

    BSONObj Projection::KeyOnly::hydrate( const BSONObj& key ) const {
        verify( _include.size() == _names.size() );

        BSONObjBuilder b( key.objsize() + _stringSize + 16 );

        BSONObjIterator i(key);
        unsigned n=0;
        while ( i.more() ) {
            verify( n < _include.size() );
            BSONElement e = i.next();
            if ( _include[n] ) {
                b.appendAs( e , _names[n] );
            }
            n++;
        }

        return b.obj();
    }
}