/*
 *
 * 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 "qpid/framing/FieldTable.h"
#include "qpid/types/Variant.h"
#include "qpid/log/Statement.h"
#include "qpid/broker/Queue.h"
#include "qpid/broker/MessageGroupManager.h"

using namespace qpid::broker;

namespace {
    const std::string GROUP_QUERY_KEY("qpid.message_group_queue");
    const std::string GROUP_HEADER_KEY("group_header_key");
    const std::string GROUP_STATE_KEY("group_state");
    const std::string GROUP_ID_KEY("group_id");
    const std::string GROUP_MSG_COUNT("msg_count");
    const std::string GROUP_TIMESTAMP("timestamp");
    const std::string GROUP_CONSUMER("consumer");
}


const std::string MessageGroupManager::qpidMessageGroupKey("qpid.group_header_key");
const std::string MessageGroupManager::qpidSharedGroup("qpid.shared_msg_group");
const std::string MessageGroupManager::qpidMessageGroupTimestamp("qpid.group_timestamp");


void MessageGroupManager::unFree( const GroupState& state )
{
    GroupFifo::iterator pos = freeGroups.find( state.members.front() );
    assert( pos != freeGroups.end() && pos->second == &state );
    freeGroups.erase( pos );
}

void MessageGroupManager::own( GroupState& state, const std::string& owner )
{
    state.owner = owner;
    unFree( state );
}

void MessageGroupManager::disown( GroupState& state )
{
    state.owner.clear();
    assert(state.members.size());
    assert(freeGroups.find(state.members.front()) == freeGroups.end());
    freeGroups[state.members.front()] = &state;
}

MessageGroupManager::GroupState& MessageGroupManager::findGroup( const QueuedMessage& qm )
{
    uint32_t thisMsg = qm.position.getValue();
    if (cachedGroup && lastMsg == thisMsg) {
        hits++;
        return *cachedGroup;
    }

    std::string group = defaultGroupId;
    const qpid::framing::FieldTable* headers = qm.payload->getApplicationHeaders();
    if (headers) {
        qpid::framing::FieldTable::ValuePtr id = headers->get( groupIdHeader );
        if (id && id->convertsTo<std::string>()) {
            std::string tmp = id->get<std::string>();
            if (!tmp.empty())   // empty group is reserved
                group = tmp;
        }
    }

    if (cachedGroup && group == lastGroup) {
        hits++;
        lastMsg = thisMsg;
        return *cachedGroup;
    }

    misses++;

    GroupState& found = messageGroups[group];
    if (found.group.empty())
        found.group = group;    // new group, assign name
    lastMsg = thisMsg;
    lastGroup = group;
    cachedGroup = &found;
    return found;
}


void MessageGroupManager::enqueued( const QueuedMessage& qm )
{
    // @todo KAG optimization - store reference to group state in QueuedMessage
    // issue: const-ness??
    GroupState& state = findGroup(qm);
    state.members.push_back(qm.position);
    uint32_t total = state.members.size();
    QPID_LOG( trace, "group queue " << qName <<
              ": added message to group id=" << state.group << " total=" << total );
    if (total == 1) {
        // newly created group, no owner
        assert(freeGroups.find(qm.position) == freeGroups.end());
        freeGroups[qm.position] = &state;
    }
}


void MessageGroupManager::acquired( const QueuedMessage& qm )
{
    // @todo KAG  avoid lookup: retrieve direct reference to group state from QueuedMessage
    // issue: const-ness??
    GroupState& state = findGroup(qm);
    assert(state.members.size());   // there are msgs present
    state.acquired += 1;
    QPID_LOG( trace, "group queue " << qName <<
              ": acquired message in group id=" << state.group << " acquired=" << state.acquired );
}


void MessageGroupManager::requeued( const QueuedMessage& qm )
{
    // @todo KAG  avoid lookup: retrieve direct reference to group state from QueuedMessage
    // issue: const-ness??
    GroupState& state = findGroup(qm);
    assert( state.acquired != 0 );
    state.acquired -= 1;
    if (state.acquired == 0 && state.owned()) {
        QPID_LOG( trace, "group queue " << qName <<
                  ": consumer name=" << state.owner << " released group id=" << state.group);
        disown(state);
    }
    QPID_LOG( trace, "group queue " << qName <<
              ": requeued message to group id=" << state.group << " acquired=" << state.acquired );
}


void MessageGroupManager::dequeued( const QueuedMessage& qm )
{
    // @todo KAG  avoid lookup: retrieve direct reference to group state from QueuedMessage
    // issue: const-ness??
    GroupState& state = findGroup(qm);
    assert( state.members.size() != 0 );
    assert( state.acquired != 0 );
    state.acquired -= 1;

    // likely to be at or near begin() if dequeued in order
    bool reFreeNeeded = false;
    if (state.members.front() == qm.position) {
        if (!state.owned()) {
            // will be on the freeGroups list if mgmt is dequeueing rather than a consumer!
            // if on freelist, it is indexed by first member, which is about to be removed!
            unFree(state);
            reFreeNeeded = true;
        }
        state.members.pop_front();
    } else {
        GroupState::PositionFifo::iterator pos = state.members.begin() + 1;
        GroupState::PositionFifo::iterator end = state.members.end();
        while (pos != end) {
            if (*pos == qm.position) {
                state.members.erase(pos);
                break;
            }
            ++pos;
        }
    }

    uint32_t total = state.members.size();
    QPID_LOG( trace, "group queue " << qName <<
              ": dequeued message from group id=" << state.group << " total=" << total );

    if (total == 0) {
        QPID_LOG( trace, "group queue " << qName << ": deleting group id=" << state.group);
        if (cachedGroup == &state) {
            cachedGroup = 0;
        }
        std::string key(state.group);
        messageGroups.erase( key );
    } else if (state.acquired == 0 && state.owned()) {
        QPID_LOG( trace, "group queue " << qName <<
                  ": consumer name=" << state.owner << " released group id=" << state.group);
        disown(state);
    } else if (reFreeNeeded) {
        disown(state);
    }
}

MessageGroupManager::~MessageGroupManager()
{
    QPID_LOG( debug, "group queue " << qName << " cache results: hits=" << hits << " misses=" << misses );
}
bool MessageGroupManager::nextConsumableMessage( Consumer::shared_ptr& c, QueuedMessage& next )
{
    if (!messages.size())
        return false;

    next.position = c->getPosition();
    if (!freeGroups.empty()) {
        const framing::SequenceNumber& nextFree = freeGroups.begin()->first;
        if (nextFree <= next.position) {  // take oldest free
            next.position = nextFree;
            --next.position;
        }
    }

    while (messages.browse( next.position, next, true )) {
        GroupState& group = findGroup(next);
        if (!group.owned()) {
            //TODO: make acquire more efficient when we already have the message in question
            if (group.members.front() == next.position && messages.acquire(next.position, next)) {    // only take from head!
                return true;
            }
            QPID_LOG(debug, "Skipping " << next.position << " since group " << group.group
                     << "'s head message still pending. pos=" << group.members.front());
        } else if (group.owner == c->getName() && messages.acquire(next.position, next)) {
            return true;
        }
    }
    return false;
}


bool MessageGroupManager::allocate(const std::string& consumer, const QueuedMessage& qm)
{
    // @todo KAG avoid lookup: retrieve direct reference to group state from QueuedMessage
    GroupState& state = findGroup(qm);

    if (!state.owned()) {
        own( state, consumer );
        QPID_LOG( trace, "group queue " << qName <<
                  ": consumer name=" << consumer << " has acquired group id=" << state.group);
        return true;
    }
    return state.owner == consumer;
}

bool MessageGroupManager::nextBrowsableMessage( Consumer::shared_ptr& c, QueuedMessage& next )
{
    // browse: allow access to any available msg, regardless of group ownership (?ok?)
    return messages.browse(c->getPosition(), next, false);
}

void MessageGroupManager::query(qpid::types::Variant::Map& status) const
{
    /** Add a description of the current state of the message groups for this queue.
        FORMAT:
        { "qpid.message_group_queue":
            { "group_header_key" : "<KEY>",
              "group_state" :
                   [ { "group_id"  : "<name>",
                       "msg_count" : <int>,
                       "timestamp" : <absTime>,
                       "consumer"  : <consumer name> },
                     {...} // one for each known group
                   ]
            }
        }
    **/

    assert(status.find(GROUP_QUERY_KEY) == status.end());
    qpid::types::Variant::Map state;
    qpid::types::Variant::List groups;

    state[GROUP_HEADER_KEY] = groupIdHeader;
    for (GroupMap::const_iterator g = messageGroups.begin();
         g != messageGroups.end(); ++g) {
        qpid::types::Variant::Map info;
        info[GROUP_ID_KEY] = g->first;
        info[GROUP_MSG_COUNT] = (uint64_t) g->second.members.size();
        // set the timestamp to the arrival timestamp of the oldest (HEAD) message, if present
        info[GROUP_TIMESTAMP] = 0;
        if (g->second.members.size() != 0) {
            QueuedMessage qm;
            if (messages.find(g->second.members.front(), qm) &&
                qm.payload &&
                qm.payload->hasProperties<framing::DeliveryProperties>()) {
                info[GROUP_TIMESTAMP] = qm.payload->getProperties<framing::DeliveryProperties>()->getTimestamp();
            }
        }
        info[GROUP_CONSUMER] = g->second.owner;
        groups.push_back(info);
    }
    state[GROUP_STATE_KEY] = groups;
    status[GROUP_QUERY_KEY] = state;
}


boost::shared_ptr<MessageGroupManager> MessageGroupManager::create( const std::string& qName,
                                                                    Messages& messages,
                                                                    const qpid::framing::FieldTable& settings )
{
    boost::shared_ptr<MessageGroupManager> empty;

    if (settings.isSet(qpidMessageGroupKey)) {

        // @todo: remove once "sticky" consumers are supported - see QPID-3347
        if (!settings.isSet(qpidSharedGroup)) {
            QPID_LOG( error, "Only shared groups are supported in this version of the broker. Use '--shared-groups' in qpid-config." );
            return empty;
        }

        std::string headerKey = settings.getAsString(qpidMessageGroupKey);
        if (headerKey.empty()) {
            QPID_LOG( error, "A Message Group header key must be configured, queue=" << qName);
            return empty;
        }
        unsigned int timestamp = settings.getAsInt(qpidMessageGroupTimestamp);

        boost::shared_ptr<MessageGroupManager> manager( new MessageGroupManager( headerKey, qName, messages, timestamp ) );

        QPID_LOG( debug, "Configured Queue '" << qName <<
                  "' for message grouping using header key '" << headerKey << "'" <<
                  " (timestamp=" << timestamp << ")");
        return manager;
    }
    return empty;
}

std::string MessageGroupManager::defaultGroupId;
void MessageGroupManager::setDefaults(const std::string& groupId)   // static
{
    defaultGroupId = groupId;
}

/** Cluster replication:

   state map format:

   { "group-state": [ {"name": <group-name>,
                       "owner": <consumer-name>-or-empty,
                       "acquired-ct": <acquired count>,
                       "positions": [Seqnumbers, ... ]},
                      {...}
                    ]
   }
*/

namespace {
    const std::string GROUP_NAME("name");
    const std::string GROUP_OWNER("owner");
    const std::string GROUP_ACQUIRED_CT("acquired-ct");
    const std::string GROUP_POSITIONS("positions");
    const std::string GROUP_STATE("group-state");
}


/** Runs on UPDATER to snapshot current state */
void MessageGroupManager::getState(qpid::framing::FieldTable& state ) const
{
    using namespace qpid::framing;
    state.clear();
    framing::Array groupState(TYPE_CODE_MAP);
    for (GroupMap::const_iterator g = messageGroups.begin();
         g != messageGroups.end(); ++g) {

        framing::FieldTable group;
        group.setString(GROUP_NAME, g->first);
        group.setString(GROUP_OWNER, g->second.owner);
        group.setInt(GROUP_ACQUIRED_CT, g->second.acquired);
        framing::Array positions(TYPE_CODE_UINT32);
        for (GroupState::PositionFifo::const_iterator p = g->second.members.begin();
             p != g->second.members.end(); ++p)
            positions.push_back(framing::Array::ValuePtr(new IntegerValue( *p )));
        group.setArray(GROUP_POSITIONS, positions);
        groupState.push_back(framing::Array::ValuePtr(new FieldTableValue(group)));
    }
    state.setArray(GROUP_STATE, groupState);

    QPID_LOG(debug, "Queue \"" << qName << "\": replicating message group state, key=" << groupIdHeader);
}


/** called on UPDATEE to set state from snapshot */
void MessageGroupManager::setState(const qpid::framing::FieldTable& state)
{
    using namespace qpid::framing;
    messageGroups.clear();
    freeGroups.clear();
    cachedGroup = 0;

    framing::Array groupState(TYPE_CODE_MAP);

    bool ok = state.getArray(GROUP_STATE, groupState);
    if (!ok) {
        QPID_LOG(error, "Unable to find message group state information for queue \"" <<
                 qName << "\": cluster inconsistency error!");
        return;
    }

    for (framing::Array::const_iterator g = groupState.begin();
         g != groupState.end(); ++g) {
        framing::FieldTable group;
        ok = framing::getEncodedValue<FieldTable>(*g, group);
        if (!ok) {
            QPID_LOG(error, "Invalid message group state information for queue \"" <<
                     qName << "\": table encoding error!");
            return;
        }
        MessageGroupManager::GroupState state;
        if (!group.isSet(GROUP_NAME) || !group.isSet(GROUP_OWNER) || !group.isSet(GROUP_ACQUIRED_CT)) {
            QPID_LOG(error, "Invalid message group state information for queue \"" <<
                     qName << "\": fields missing error!");
            return;
        }
        state.group = group.getAsString(GROUP_NAME);
        state.owner = group.getAsString(GROUP_OWNER);
        state.acquired = group.getAsInt(GROUP_ACQUIRED_CT);
        framing::Array positions(TYPE_CODE_UINT32);
        ok = group.getArray(GROUP_POSITIONS, positions);
        if (!ok) {
            QPID_LOG(error, "Invalid message group state information for queue \"" <<
                     qName << "\": position encoding error!");
            return;
        }

        for (Array::const_iterator p = positions.begin(); p != positions.end(); ++p)
            state.members.push_back((*p)->getIntegerValue<uint32_t, 4>());
        messageGroups[state.group] = state;
        if (!state.owned()) {
            assert(state.members.size());
            freeGroups[state.members.front()] = &messageGroups[state.group];
        }
    }

    QPID_LOG(debug, "Queue \"" << qName << "\": message group state replicated, key =" << groupIdHeader)
}