core/post_auction/post_auction_loop.h

/* post_auction_loop.h                                             -*- C++ -*-
   Jeremy Barnes, 30 May 2012
   Router post-auction loop.
*/

#ifndef __router__post_auction_loop_h__
#define __router__post_auction_loop_h__

#include "soa/service/service_base.h"
#include "soa/service/pending_list.h"
#include <unordered_map>
#include "soa/service/message_loop.h"
#include "soa/service/typed_message_channel.h"
#include "rtbkit/common/auction.h"
#include "rtbkit/common/auction_events.h"
#include "soa/service/zmq_endpoint.h"
#include "soa/service/zmq_message_router.h"
#include "soa/service/zmq_named_pub_sub.h"
#include "rtbkit/core/agent_configuration/agent_configuration_listener.h"
#include "rtbkit/core/banker/banker.h"
#include "rtbkit/core/monitor/monitor_provider.h"

namespace RTBKIT {

/*****************************************************************************/
/* SUBMISSION INFO                                                           */
/*****************************************************************************/

struct SubmissionInfo {
    SubmissionInfo()
        : fromOldRouter(false)
    {
    }

    std::shared_ptr<BidRequest> bidRequest;
    std::string bidRequestStr;
    std::string bidRequestStrFormat;
    JsonHolder augmentations;
    Auction::Response  bid;               
    bool fromOldRouter;                   

    std::vector<std::shared_ptr<PostAuctionEvent> > earlyWinEvents;
    std::vector<std::shared_ptr<PostAuctionEvent> > earlyCampaignEvents;

    std::string serializeToString() const;
    void reconstituteFromString(const std::string & str);
};


/*****************************************************************************/
/* FINISHED INFO                                                             */
/*****************************************************************************/

struct FinishedInfo {
    FinishedInfo()
        : fromOldRouter(false)
    {
    }

    Date auctionTime;            
    Id auctionId;       
    Id adSpotId;          
    int spotIndex;
    std::shared_ptr<BidRequest> bidRequest;  
    std::string bidRequestStr;
    std::string bidRequestStrFormat;
    JsonHolder augmentations;
    std::set<Id> uids;                

    SegmentList visitChannels;

    void addUids(const UserIds & toAdd)
    {
        for (auto it = toAdd.begin(), end = toAdd.end();  it != end;  ++it) {
            auto jt = uids.find(it->second);
            if (jt != uids.end())
                return;
            uids.insert(it->second);
        }
    }

    Date bidTime;                
    Auction::Response bid;       
    Json::Value bidToJson() const;

    bool hasWin() const { return winTime != Date(); }
    void setWin(Date winTime, BidStatus status, Amount winPrice,
                const std::string & winMeta)
    {
        if (hasWin())
            throw ML::Exception("already has win");
        this->winTime = winTime;
        this->reportedStatus = status;
        this->winPrice = winPrice;
        this->winMeta = winMeta;
    }

    Date winTime;                
    BidStatus reportedStatus;    
    Amount winPrice;             
    std::string winMeta;         
    Json::Value winToJson() const;

    CampaignEvents campaignEvents;

    struct Visit {
        Date visitTime;           
        SegmentList channels;     
        std::string meta;         

        Json::Value toJson() const;
        void serialize(ML::DB::Store_Writer & store) const;
        void reconstitute(ML::DB::Store_Reader & store);
    };

    std::vector<Visit> visits;

    void addVisit(Date visitTime,
                  const std::string & visitMeta,
                  const SegmentList & channels);

    Json::Value visitsToJson() const;

    Json::Value toJson() const;

    bool fromOldRouter;

    std::string serializeToString() const;
    void reconstituteFromString(const std::string & str);
};


/*****************************************************************************/
/* POST AUCTION LOOP                                                         */
/*****************************************************************************/

struct PostAuctionLoop : public ServiceBase, public MonitorProvider
{

    PostAuctionLoop(ServiceBase & parent,
                    const std::string & serviceName);
    PostAuctionLoop(std::shared_ptr<ServiceProxies> proxies,
                    const std::string & serviceName);

    ~PostAuctionLoop()
    {
        shutdown();
    }

    std::shared_ptr<Banker> getBanker() const
    {
        return banker;
    }

    void setBanker(const std::shared_ptr<Banker> & newBanker)
    {
        banker = newBanker;
    }

    std::shared_ptr<Banker> banker;

    /* ROUTERSHARED */
    uint64_t numWins;
    uint64_t numLosses;
    uint64_t numCampaignEvents;

    /* /ROUTERSHARED */

    /* ROUTERBASE */
    /************************************************************************/
    /* EXCEPTIONS                                                           */
    /************************************************************************/

    void throwException(const std::string & key, const std::string & fmt,
                        ...) __attribute__((__noreturn__));

    /************************************************************************/
    /* LOGGING                                                              */
    /************************************************************************/
    
    ZmqNamedPublisher logger;

    template<typename... Args>
    void logMessage(const std::string & channel, Args... args)
    {
        using namespace std;
        //cerr << "********* logging message to " << channel << endl;
        logger.publish(channel, Date::now().print(5), args...);
    }

    template<typename... Args>
    void logPAError(const std::string & function,
                    const std::string & exception,
                    Args... args)
    {
        logger.publish("PAERROR", Date::now().print(5),
                       function, exception, args...);
        recordHit("error.%s", function);
    }

    /* /ROUTERBASE */

    void bindTcp();

    void init();

    void start(std::function<void ()> onStop = std::function<void ()>());

    void shutdown();

    size_t numAwaitingWinLoss() const
    {
        return submitted.size();
    }

    size_t numFinishedAuctionsTracked() const
    {
        return finished.size();
    }

    void initStatePersistence(const std::string & path);

    virtual Json::Value getServiceStatus() const;

    void injectSubmittedAuction(const Id & auctionId,
                                const Id & adSpotId,
                                std::shared_ptr<BidRequest> bidRequest,
                                const std::string & bidRequestStr,
                                const std::string & bidRequestStrFormat,
                                const JsonHolder & augmentations,
                                const Auction::Response & bidResponse,
                                Date lossTimeout);

    void injectWin(const Id & auctionId,
                   const Id & adspot,
                   Amount winPrice,
                   Date timestamp,
                   const JsonHolder & winMeta,
                   const UserIds & ids,
                   const AccountKey & account,
                   Date bidTimestamp);

    void injectLoss(const Id & auctionId,
                    const Id & adspot,
                    Date timestamp,
                    const JsonHolder & lossMeta,
                    const AccountKey & account,
                    Date bidTimestamp);

    void injectCampaignEvent(const std::string & label,
                             const Id & auctionId,
                             const Id & adSpotId,
                             Date timestamp,
                             const JsonHolder & eventMeta,
                             const UserIds & ids);

    void notifyFinishedSpot(const Id & auctionId, const Id & adSpotId);

    /* Post service health status to Monitor */
    MonitorProviderClient monitorProviderClient;

    /* MonitorProvider interface */
    std::string getProviderName() const;
    Json::Value getProviderIndicators() const;

    Date lastWinLoss;
    Date lastCampaignEvent;

private:
    void initConnections();

    void doAuction(const SubmittedAuctionEvent & event);

    void doEvent(const std::shared_ptr<PostAuctionEvent> & event);

    void doAuctionMessage(const std::vector<std::string> & message);

    void doWinMessage(const std::vector<std::string> & message);

    void doLossMessage(const std::vector<std::string> & message);

    void doCampaignEventMessage(const std::vector<std::string> & message);

    void checkExpiredAuctions();

    void doWinLoss(const std::shared_ptr<PostAuctionEvent> & event,
                   bool isReplay);

    void doSubmitted(const std::shared_ptr<PostAuctionEvent> & event);

    void doCampaignEvent(const std::shared_ptr<PostAuctionEvent> & event);

    bool routePostAuctionEvent(const std::string & label,
                               const FinishedInfo & finished,
                               const SegmentList & channels,
                               bool filterChannels);

    void doBidResult(const Id & auctionId,
                     const Id & adSpotId,
                     const SubmissionInfo & submission,
                     Amount price,
                     Date timestamp,
                     BidStatus status,
                     const std::string & confidence,
                     const std::string & winLossMeta,
                     const UserIds & uids);

    typedef PendingList<std::pair<Id, Id>,
                        SubmissionInfo> Submitted;
    Submitted submitted;

    typedef PendingList<std::pair<Id, Id>, FinishedInfo> Finished;
    Finished finished;

    MessageLoop loop;

    TypedMessageSink<SubmittedAuctionEvent> auctions;

    TypedMessageSink<std::shared_ptr<PostAuctionEvent> > events;

    ZmqNamedEndpoint endpoint;

    ZmqMessageRouter router;

    ZmqNamedClientBus toAgents;

    template<typename... Args>
    void sendAgentMessage(const std::string & agent,
                          const std::string & messageType,
                          const Date & date,
                          Args... args)
    {
        toAgents.sendMessage(agent, messageType, date,
                             std::forward<Args>(args)...);
    }

    static std::string makeBidId(Id auctionId, Id spotId, const std::string & agent);

    AgentConfigurationListener configListener;
};


} // namespace RTBKIT


#endif /* __router__post_auction_loop_h__ */