// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.

#include "attribute_populator.h"
#include <vespa/searchcore/proton/common/eventlogger.h>
#include <vespa/vespalib/util/idestructorcallback.h>
#include <vespa/searchlib/common/flush_token.h>
#include <vespa/vespalib/util/destructor_callbacks.h>
#include <vespa/vespalib/util/gate.h>
#include <vespa/searchlib/attribute/attributevector.h>
#include <cassert>

#include <vespa/log/log.h>
LOG_SETUP(".proton.attribute.attribute_populator");

using vespalib::IDestructorCallback;

namespace proton {

namespace {

class PopulateDoneContext : public IDestructorCallback
{
    std::shared_ptr<document::Document> _doc;
public:
    PopulateDoneContext(std::shared_ptr<document::Document> doc) noexcept
        : _doc(std::move(doc))
    {
    }
    ~PopulateDoneContext() override = default;
};

}

search::SerialNum
AttributePopulator::nextSerialNum()
{
    assert(_currSerialNum <= _configSerialNum);
    return _currSerialNum++;
}

std::vector<vespalib::string>
AttributePopulator::getNames() const
{
    std::vector<search::AttributeGuard> attrs;
    _writer.getAttributeManager()->getAttributeList(attrs);
    std::vector<vespalib::string> names;
    names.reserve(attrs.size());
    for (const search::AttributeGuard &attr : attrs) {
        names.push_back(_subDbName + ".attribute." + attr->getName());
    }
    return names;
}

AttributePopulator::AttributePopulator(const proton::IAttributeManager::SP &mgr,
                                       search::SerialNum initSerialNum,
                                       const vespalib::string &subDbName,
                                       search::SerialNum configSerialNum)
    : _writer(mgr),
      _initSerialNum(initSerialNum),
      _currSerialNum(initSerialNum),
      _configSerialNum(configSerialNum),
      _subDbName(subDbName)
{
    if (LOG_WOULD_LOG(event)) {
        EventLogger::populateAttributeStart(getNames());
    }
}

AttributePopulator::~AttributePopulator()
{
    if (LOG_WOULD_LOG(event)) {
        EventLogger::populateAttributeComplete(getNames(),_currSerialNum - _initSerialNum);
    }
}

void
AttributePopulator::handleExisting(uint32_t lid, const std::shared_ptr<document::Document> &doc)
{
    search::SerialNum serialNum(nextSerialNum());
    _writer.put(serialNum, *doc, lid, std::make_shared<PopulateDoneContext>(doc));
    vespalib::Gate gate;
    _writer.forceCommit(serialNum, std::make_shared<vespalib::GateCallback>(gate));
    gate.await();
}

void
AttributePopulator::done()
{
    auto mgr = _writer.getAttributeManager();
    auto flushTargets = mgr->getFlushTargets();
    for (const auto &flushTarget : flushTargets) {
        assert(flushTarget->getFlushedSerialNum() < _configSerialNum);
        auto task = flushTarget->initFlush(_configSerialNum, std::make_shared<search::FlushToken>());
        // shrink target only return task if able to shrink.
        if (task) {
            task->run();
        }
        assert(flushTarget->getFlushedSerialNum() == _configSerialNum);
    }
}

} // namespace proton