#include "world.hpp"
#include "idalloc.hpp"
#include "animqueue.hpp"
#include "gui.hpp"
#include "traceback.hpp"
#include "pprint.hpp"
#include <iostream>

void World::store_global_pointer(lua_State *L, World *v) {
    lua_pushstring(L, "world");
    lua_pushlightuserdata(L, v);
    lua_rawset(L, LUA_REGISTRYINDEX);
}

World *World::fetch_global_pointer(lua_State *L) {
    lua_pushstring(L, "world");
    lua_rawget(L, LUA_REGISTRYINDEX);
    World *result = (World *)lua_touserdata(L, -1);
    if (result == nullptr) {
        luaL_error(L, "No world pointer stored.");
    }
    lua_pop(L, 1);
    return result;
}

World::~World() {
}

World::World(util::WorldType wt) {
    // Master world model by default.
    world_type_ = wt;

    // Initialize the ID allocator in master mode.
    if (wt == util::WORLD_TYPE_MASTER || wt == util::WORLD_TYPE_STANDALONE) {
        id_global_pool_.init_master();
    } else {
        id_global_pool_.init_synch();
    }

    // Prepare to manipulate the lua state.
    LuaVar world, globtab;
    LuaStack LS(state(), world, globtab);

    // Put the world pointer into the lua registry.
    World::store_global_pointer(state(), this);

    // Clear the global GUI pointer.
    Gui::store_global_pointer(state(), nullptr);

    // Clear the lthread state.
    clear_lthread_state();

    // Set the tabletype of the registry.
    LS.settabletype(LuaRegistry, LUA_TT_REGISTRY);

    // Set the tabletype of the global environment.
    LS.getglobaltable(globtab);
    LS.settabletype(globtab, LUA_TT_GLOBALENV);

    // Create the tangibles table in the registry.
    LS.rawset(LuaRegistry, "tangibles", LuaNewTable);

    // Create the globaldb and oncedb in the registry.
    if (util::world_type_authoritative(wt)) {
        LS.rawset(LuaRegistry, "globaldb", LuaNewTable);
        LS.rawset(LuaRegistry, "oncedb", LuaNewTable);
    }

    // Initialize the SourceDB.  At this stage, the sourcedb is
    // empty, so it's just populating the lua builtins.
    source_db_.init(state());
    eng::string srcerrs = source_db_.rebuild();

    // Clear the clock.
    clock_ = 0;

    // There shouldn't be any lua errors in the sourceDB at this
    // point, since there's no lua code in the sourceDB.
    assert(srcerrs == "");

    LS.result();
    assert (stack_is_clear());
}

Tangible::Tangible(World *w, int64_t id) : world_(w), anim_queue_(w->world_type_), id_player_pool_(&w->id_global_pool_) {
    plane_item_.set_id(id);
    plane_item_.track(&w->plane_map_);
}

void Tangible::update_plane_item() {
    const AnimStep &aqback = anim_queue_.back();
    plane_item_.set_pos(aqback.plane(), aqback.xyz().x, aqback.xyz().y, aqback.xyz().z);
}

void Tangible::serialize(StreamBuffer *sb) {
    anim_queue_.serialize(sb);
    id_player_pool_.serialize(sb);
    print_buffer_.serialize(sb);
}

void Tangible::deserialize(StreamBuffer *sb) {
    anim_queue_.deserialize(sb);
    id_player_pool_.deserialize(sb);
    print_buffer_.deserialize(sb);
    update_plane_item();
}

Tangible *World::tangible_get(int64_t id) {
    auto iter = tangibles_.find(id);
    if (iter == tangibles_.end()) {
        return nullptr;
    } else {
        return iter->second.get();
    }
}

const Tangible *World::tangible_get(int64_t id) const {
    auto iter = tangibles_.find(id);
    if (iter == tangibles_.end()) {
        return nullptr;
    } else {
        return iter->second.get();
    }
}

World::TanVector World::tangible_get_all(const IdVector &ids) const {
    TanVector result(ids.size());
    for (int i = 0; i < int(ids.size()); i++) {
        result[i] = tangible_get(ids[i]);
    }
    return result;
}

Tangible *World::tangible_get(const LuaStack &LS, LuaSlot tab) {
    int64_t id = LS.tanid(tab);
    if (id == 0) {
        luaL_error(LS.state(), "parameter is not a tangible");
    }
    Tangible *result = tangible_get(id);
    if (result == nullptr) {
        luaL_error(LS.state(), "parameter is not a tangible");
    }
    return result;
}

Tangible *World::tangible_make(lua_State *L, int64_t id, const eng::string &plane, bool pushdb) {
    // Get a state if we don't already have one.
    if (L == nullptr) {
        L = state();
        assert(!pushdb);
    }
    assert(id != 0);

    LuaVar tangibles, metatab;
    LuaRet database;
    LuaStack LS(L, tangibles, database, metatab);

    // Create the C++ part of the structure.
    UniqueTangible &t = tangibles_[id];
    assert (t == nullptr);
    t.reset(new Tangible(this, id));

    // Set up initial animation state.
    t->anim_queue_.clear(plane);
    t->update_plane_item();
    
    // Create the tangible's database and metatable.
    LS.set(database, LuaNewTable);
    LS.set(metatab, LuaNewTable);
    LS.setmetatable(database, metatab);

    // Mark the tangible using the tabletype field.
    LS.settabletype(database, LUA_TT_TANGIBLE);
    LS.settabletype(metatab, LUA_TT_TANGIBLEMETA);

    // Store the database into the tangibles table.
    LS.rawget(tangibles, LuaRegistry, "tangibles");
    LS.rawset(tangibles, id, database);

    // Populate the database and metatable with initial stuff.
    LS.rawset(database, "inventory", LuaNewTable);
    LS.rawset(metatab, "id", id);
    LS.rawset(metatab, "threads", LuaNewTable);
    // LS.rawset(metatab, "__metatable", LuaNil);

    LS.result();
    if (!pushdb) lua_pop(L, 1);
    return t.get();
}

void World::tangible_delete(int64_t id) {
    lua_State *L = state();
    LuaVar tangibles, database;
    LuaStack LS(L, tangibles, database);

    // Fetch the C++ side of the tangible.
    auto iter = tangibles_.find(id);
    if (iter == tangibles_.end()) {
        LS.result();
        return; // Nothing to delete.
    }

    // Fetch the lua side of the tangible.
    LS.rawget(tangibles, LuaRegistry, "tangibles");
    LS.rawget(database, tangibles, id);
    assert(LS.istable(database));
    
    // Clear out the database.
    LS.cleartable(database, true);
    LS.settabletype(database, LUA_TT_DEADTANGIBLE);

    // Remove the lua portion from the tangibles table.
    LS.rawset(tangibles, id, LuaNil);

    // Remove the C++ portion from the tangibles table.
    tangibles_.erase(iter);
    LS.result();
}

util::IdVector World::get_near_unsorted(int64_t player_id, float radius, bool exclude_nowhere, bool omit_player) const {
    const Tangible *player = tangible_get(player_id);
    if (player == nullptr) {
        return IdVector();
    }

    // Find out where's the center of the world.
    const AnimStep &aqback = player->anim_queue_.back();
    return plane_map_.scan_radius_unsorted(aqback.plane(), aqback.xyz().x, aqback.xyz().y, radius, exclude_nowhere, player_id, omit_player);
}

util::IdVector World::get_near(int64_t player_id, float radius, bool exclude_nowhere, bool omit_player) const {
    const Tangible *player = tangible_get(player_id);
    if (player == nullptr) {
        return IdVector();
    }

    // Find out where's the center of the world.
    const AnimStep &aqback = player->anim_queue_.back();
    return plane_map_.scan_radius(aqback.plane(), aqback.xyz().x, aqback.xyz().y, radius, exclude_nowhere, player_id, omit_player);
}

World::Redirects World::fetch_redirects() {
    World::Redirects result = std::move(redirects_);
    redirects_.clear();
    return result;
}

int64_t World::create_login_actor() {
    assert(stack_is_clear());
    int64_t id = id_global_pool_.get_one();
    Tangible *tan = tangible_make(state(), id, "nowhere", true);
    LuaArg database;
    LuaVar classtab, mt;
    LuaStack LS(state(), database, classtab, mt);
    LS.makeclass(classtab, "login");
    LS.getmetatable(mt, database);
    LS.rawset(mt, "__index", classtab);
    LS.result();
    tan->configure_id_pool_for_actor();
    tan->print_buffer_.clear();
    assert(stack_is_clear());
    return tan->id();
}

eng::string World::probe_lua(int64_t actor_id, const eng::string &lua) {
    assert(stack_is_clear());
    lua_State *L = state();

    Tangible *actor = tangible_get(actor_id);
    if (actor == nullptr) {
        return "";
    }

    LuaVar closure;
    LuaStack LS(L, closure);

    // create the compiled closure.
    int status = luaL_loadbuffer(L, lua.c_str(), lua.size(), "=probe");
    lua_replace(L, closure.index());
    if (status != LUA_OK) {
        // The closure is actually an error message. Do nothing.
        // This should normally not happen: LuaConsole should filter
        // out syntax errors.
        LS.result();
        return "";
    }

    // Call the closure.
    int top = lua_gettop(L);
    lua_pushvalue(L, closure.index());
    open_lthread_state(actor_id, actor_id, false, true);
    status = traceback_pcall(L, 0, LUA_MULTRET);
    
    // If there's an error message, print it.
    // Otherwise, pretty-print the results.
    std::ostream *ostream = lthread_print_stream();
    if (status == LUA_OK) {
        for (int i = top + 1; i <= lua_gettop(L); i++) {
            LuaSpecial root(i);
            pprint(LS, root, true, ostream);
            (*ostream) << std::endl;
        }
    } else {
        const char *msg = lua_tostring(L, -1);
        if (msg == NULL) {
            msg = "(error object is not a string)";
        }
        (*ostream) << msg << std::endl;
    }

    // Collect the lthread_prints (and also make sure they
    // don't go into the printbuffer).
    eng::string result = lthread_prints_->str();
    lthread_prints_.reset();

    close_lthread_state();

    // And we're done.
    LS.result();
    assert(stack_is_clear());
    return result;
}

void World::update_gui(int64_t actor_id, int64_t place_id, Gui *gui) {
    assert(stack_is_clear());
    gui->clear(place_id);
    lua_State *L = state();

    LuaVar actor, place, ugui, func, tangibles, mt, index;
    LuaStack LS(L, actor, place, ugui, func, tangibles, mt, index);

    // Get the actor and place.
    LS.rawget(tangibles, LuaRegistry, "tangibles");
    LS.rawget(actor, tangibles, actor_id);
    LS.rawget(place, tangibles, place_id);
    if (!LS.istable(actor) || !LS.istable(place)) {
        LS.result();
        return;
    }

    // Get the interface closure.
    LS.getmetatable(mt, place);
    if (!LS.istable(mt)) {
        LS.result();
        return;
    }
    LS.rawget(index, mt, "__index");
    if (!LS.istable(index)) {
        LS.result();
        return;
    }
    LS.rawget(func, index, "interface");
    if (!LS.isfunction(func)) {
        LS.result();
        return;
    }

    // Call the interface function.
    lua_pushvalue(L, func.index());
    lua_pushvalue(L, actor.index());
    lua_pushvalue(L, place.index());
    Gui::store_global_pointer(L, gui);
    open_lthread_state(actor_id, place_id, false, false);
    int status = traceback_pcall(L, 2, 0);
    close_lthread_state();
    Gui::store_global_pointer(L, nullptr);
    if (status != 0) {
        gui->clear(0);
        std::cerr << lua_tostring(L, -1);
        LS.result();
        return;
    }

    // And we're done.
    LS.result();
    assert(stack_is_clear());
}

void World::update_source(const util::LuaSourcePtr &source) {
    if (source != nullptr) {
        update_source(*source);
    }
}

void World::update_source(const util::LuaSourceVec &source) {
    assert(stack_is_clear());
    source_db_.update(source);
    assert(stack_is_clear());
    eng::string errs = source_db_.rebuild();
    // I don't have a good place to send the error messages right
    // now.  The engine needs a catch-all place to send errors that
    // occur at unexpected times.
    std::cerr << errs;
    assert(stack_is_clear());
}

void World::run_unittests() {
    assert(stack_is_clear());
    source_db_.run_unittests();
    assert(stack_is_clear());
}

void World::invoke(const Invocation &inv) {
    switch (inv.kind()) {
    case Invocation::KIND_PLAN:
        invoke_plan(inv.actor(), inv.place(), inv.action(), inv.data());
        break;
    case Invocation::KIND_LUA:
        invoke_lua(inv.actor(), inv.place(), inv.action(), inv.data());
        break;
    case Invocation::KIND_FLUSH_PRINTS:
        invoke_flush_prints(inv.actor(), inv.place(), inv.action(), inv.data());
        break;
    case Invocation::KIND_TICK:
        invoke_tick(inv.actor(), inv.place(), inv.action(), inv.data());
        break;
    case Invocation::KIND_LUA_SOURCE:
        invoke_lua_source(inv.actor(), inv.place(), inv.action(), inv.data());
        break;
    default:
        // Do nothing.  Standard behavior for any invalid command is to
        // simply do nothing at all.  Perhaps eventually we may add a flag
        // to the world model to indicate that we've detected an invalid
        // command, to allow us to close the connection to a client that
        // is misbehaving.
        break;
    }
}

void World::invoke_flush_prints(int64_t actor_id, int64_t place_id, const eng::string &action, const InvocationData &data) {
    assert(stack_is_clear());
    // Check argument sanity.
    if (actor_id != place_id) {
        return;
    }
    int64_t line = util::str_to_int64(action, -1);
    if ((line < 0)||(line > INT_MAX)) {
        return;
    }
    Tangible *tactor = tangible_get(actor_id);
    if (tactor == nullptr) {
        return;
    }
    tactor->print_buffer_.discard_upto(line);
    assert(stack_is_clear());
}



void World::invoke_lua(int64_t actor_id, int64_t place_id, const eng::string &action, const InvocationData &data) {
    assert(stack_is_clear());

    // Make sure that actor and place exist.
    Tangible *tactor = tangible_get(actor_id);
    Tangible *tplace = tangible_get(place_id);
    if ((tactor == nullptr) || (tplace == nullptr)) {
        return;
    }

    // Get a thread ID for the new thread.
    int64_t tid = tplace->id_player_pool_.get_one();

    // Set up for lua manipulation.
    lua_State *L = state();
    LuaVar func, tangibles, place, mt, thread, thinfo, threads;
    LuaStack LS(L, func, tangibles, place, mt, thread, thinfo, threads);

    // create the compiled closure.
    int status = luaL_loadbuffer(L, action.c_str(), action.size(), "=invoke");
    lua_replace(L, func.index());
    if (status != LUA_OK) {
        // The closure is actually an error message. Do nothing.
        // This should normally not happen: LuaConsole should filter
        // out syntax errors.
        LS.result();
        return;
    }

    // Get the place.
    LS.rawget(tangibles, LuaRegistry, "tangibles");
    LS.rawget(place, tangibles, place_id);
    if (!LS.istable(place)) {
        LS.result();
        return;
    }

    // Get the place's metatable.
    LS.getmetatable(mt, place);
    if (!LS.istable(mt)) {
        LS.result();
        return;
    }

    // Create a new thread, set up function and parameters.
    lua_State *CO = LS.newthread(thread);
    lua_pushvalue(L, func.index());
    lua_xmove(L, CO, 1);

    // Create the thread info table.
    LS.newtable(thinfo);
    LS.rawset(thinfo, "thread", thread);
    LS.rawset(thinfo, "actorid", actor_id);
    LS.rawset(thinfo, "nargs", 0);
    LS.rawset(thinfo, "useppool", true);
    LS.rawset(thinfo, "print", true);
    
    // Store the thread into place's thread table.
    LS.rawget(threads, mt, "threads");
    if (!LS.istable(threads)) {
        LS.result();
        return;
    }
    LS.rawset(threads, tid, thinfo);
    LS.result();

    thread_sched_.add(0, tid, place_id);
    run_scheduled_threads();
    assert(stack_is_clear());
}

void World::invoke_plan(int64_t actor_id, int64_t place_id, const eng::string &action, const InvocationData &data) {
    assert(stack_is_clear());

    // Validate that the action is legal.
    Gui validation_gui;
    update_gui(actor_id, place_id, &validation_gui);
    if (!validation_gui.has_action(action)) {
        return;
    }

    // Make sure the action starts with "cb_"
    if (!util::has_prefix(action, "cb_")) {
        return;
    }

    // Get the actor and place.  Make sure both exist.
    Tangible *tactor = tangible_get(actor_id);
    Tangible *tplace = tangible_get(place_id);
    if ((tactor == nullptr) || (tplace == nullptr)) {
        return;
    }

    // Get an ID for the thread.  We always use the player
    // pool in this case.
    int64_t tid = tactor->id_player_pool_.get_one();

    // Set up for Lua manipulation.
    lua_State *L = state();
    LuaVar actor, place, func, tangibles, mt, index, thread, threads, thinfo, message, invdata;
    LuaStack LS(L, actor, place, func, tangibles, mt, index, thread, threads, thinfo, message, invdata);

    // Get the actor and place.
    LS.rawget(tangibles, LuaRegistry, "tangibles");
    LS.rawget(actor, tangibles, actor_id);
    LS.rawget(place, tangibles, place_id);
    if (!LS.istable(actor) || !LS.istable(place)) {
        LS.result();
        return;
    }

    // Get the action closure.
    LS.getmetatable(mt, place);
    if (!LS.istable(mt)) {
        LS.result();
        return;
    }
    LS.rawget(index, mt, "__index");
    if (!LS.istable(index)) {
        LS.result();
        return;
    }
    LS.rawget(func, index, action);
    if (!LS.isfunction(func)) {
        LS.result();
        return;
    }

    // Convert the InvocationData into a lua table.
    LS.newtable(invdata);
    for (const auto &p : data) {
        LS.rawset(invdata, p.first, p.second);
    }

    // Create a new thread, set up function and parameters.
    lua_State *CO = LS.newthread(thread);
    lua_pushvalue(L, func.index());
    lua_pushvalue(L, actor.index());
    lua_pushvalue(L, place.index());
    lua_pushvalue(L, invdata.index());
    lua_xmove(L, CO, 4);

    // Create the thread info table.
    LS.newtable(thinfo);
    LS.rawset(thinfo, "thread", thread);
    LS.rawset(thinfo, "actorid", actor_id);
    LS.rawset(thinfo, "nargs", 3); // actor, place, invdata
    LS.rawset(thinfo, "useppool", true);
    LS.rawset(thinfo, "print", false);
    
    // Store the thread into place's thread table.
    LS.rawget(threads, mt, "threads");
    if (!LS.istable(threads)) {
        LS.result();
        return;
    }
    LS.rawset(threads, tid, thinfo);
    LS.result();
    
    // Push the thread's ID into the runnable thread queue,
    // then run the thread queue.
    thread_sched_.add(0, tid, place_id);
    run_scheduled_threads();
    assert(stack_is_clear());
}

void World::invoke_tick(int64_t actor_id, int64_t place_id, const eng::string &action, const InvocationData &data) {
    if (!util::world_type_authoritative(world_type_)) {
        return;
    }
    clock_ += 1;
    run_scheduled_threads();
}

void World::invoke_lua_source(int64_t actor_id, int64_t place_id, const eng::string &action, const InvocationData &data) {
    if (!util::world_type_authoritative(world_type_)) {
        return;
    }
    // We need some kind of authentication here.
    try {
        StreamBuffer sb(action);
        util::LuaSourceVec sv;
        SourceDB::deserialize_source(&sv, &sb);
        update_source(sv);
    } catch (const StreamException &ex) {
        return;
    }
}

void World::run_scheduled_threads() {
    assert(stack_is_clear());
    lua_State *L = state();
    LuaVar tangibles, place, mt, threads, thinfo, actorid, nargs, useppool, thread, print;
    LuaStack LS(L, tangibles, place, mt, threads, thinfo, actorid, nargs, useppool, thread, print);

    LS.rawget(tangibles, LuaRegistry, "tangibles");
    while (thread_sched_.ready(clock_)) {
        SchedEntry sched = thread_sched_.pop();
        LS.rawget(place, tangibles, sched.place_id());
        if (!LS.istable(place)) {
            continue;
        }
        LS.getmetatable(mt, place);
        if (!LS.istable(mt)) {
            continue;
        }
        LS.rawget(threads, mt, "threads");
        if (!LS.istable(threads)) {
            continue;
        }
        LS.rawget(thinfo, threads, sched.thread_id());
        if (!LS.istable(thinfo)) {
            continue;
        }
        LS.rawget(actorid, thinfo, "actorid");
        if (!LS.isnumber(actorid)) {
            continue;
        }
        LS.rawget(nargs, thinfo, "nargs");
        if (!LS.isnumber(nargs)) {
            continue;
        }
        LS.rawget(useppool, thinfo, "useppool");
        if (!LS.isboolean(useppool)) {
            continue;
        }
        LS.rawget(thread, thinfo, "thread");
        if (!LS.isthread(thread)) {
            continue;
        }

        // Resume the coroutine.
        lua_State *CO = LS.ckthread(thread);
        open_lthread_state(LS.ckinteger(actorid), sched.place_id(), LS.ckboolean(useppool), true);
        int status = lua_resume(CO, nullptr, LS.ckint(nargs));
        std::ostream *ostream = lthread_print_stream();

        // Three possible outcomes: finished, yielded, or errored.
        if (!util::world_type_authoritative(world_type_)) {
            LS.rawset(threads, sched.thread_id(), LuaNil);
        } else if (status == LUA_YIELD) {
            // If there's nothing on the stack, infer that tangible.nopredict yielded.
            if (lua_gettop(CO) == 0) {
                LS.rawset(threads, sched.thread_id(), LuaNil);
            }
            // If there's a single number on the stack, infer that 'wait' yielded.
            else if ((lua_gettop(CO) == 1) && (lua_isnumber(CO, 1))) {
                lua_Number delay = lua_tonumber(CO, 1);
                lua_settop(CO, 0);
                LS.rawset(thinfo, "nargs", 0);
                LS.rawset(thinfo, "useppool", false);
                thread_sched_.add(clock_ + int64_t(delay), sched.thread_id(), sched.place_id());
            }
            // In any other case, generate an error and kill the coroutine.
            else {
                std::cerr << "Thread yielded incorrectly.  Killing it." << std::endl;
                LS.rawset(threads, sched.thread_id(), LuaNil);
            }
        } else if (status == LUA_OK) {
            // Successfully ran to completion.  Print any return values.
            // Remove from thread table.
            LS.rawget(print, thinfo, "print");
            LuaStack LSCO(CO);
            if (LS.ckboolean(print)) {
                for (int i = 1; i <= lua_gettop(CO); i++) {
                    pprint(LSCO, LuaSpecial(i), true, ostream);
                    (*ostream) << std::endl;
                }
            }
            LS.rawset(threads, sched.thread_id(), LuaNil);
        } else {
            // Generated an error.  Add a traceback, print, and kill the coroutine.
            // Currently, the error is sent to the actor.  That seems... not right in the long run.
            traceback_coroutine(CO);
            (*ostream) << lua_tostring(CO, -1);
            LS.rawset(threads, sched.thread_id(), LuaNil);
        }
        close_lthread_state();
    }

    LS.result();
    assert(stack_is_clear());
}

int64_t World::alloc_id_predictable() {
    if (!lthread_use_ppool_) {
        return id_global_pool_.get_one();
    }
    Tangible *t = tangible_get(lthread_actor_id_);
    if (t == nullptr) {
        return id_global_pool_.get_one();
    }
    return t->id_player_pool_.get_one();
}

const PrintBuffer *World::get_printbuffer(int64_t actor_id) {
    Tangible *actor = tangible_get(actor_id);
    if (actor != nullptr) {
        return &actor->print_buffer_;
    }
    return nullptr;
}

void World::clear_lthread_state() {
    lthread_prints_.reset();
    lthread_actor_id_ = 0;
    lthread_place_id_ = 0;
    lthread_use_ppool_ = false;
}

void World::open_lthread_state(int64_t actor, int64_t place, bool ppool, bool prints) {
    lthread_actor_id_ = actor;
    lthread_place_id_ = place;
    lthread_use_ppool_ = ppool;
    if (prints) {
        lthread_prints_.reset(new eng::ostringstream);
    } else {
        lthread_prints_.reset();
    }
}

void World::close_lthread_state() {
    // Copy prints from lthread_prints_ stringstream into
    // the appropriate actor's PrintBuffer.  If for some reason
    // there isn't an actor, or if the actor doesn't have a PrintBuffer,
    // send the output to std::cerr.
    if (lthread_prints_ != nullptr) {
        const eng::string &output = lthread_prints_->str();
        Tangible *actor = tangible_get(lthread_actor_id_);
        if (actor != nullptr) {
            bool auth = util::world_type_authoritative(world_type_);
            actor->print_buffer_.add_string(output, auth);
        }
    }
    // Now clean up everything.
    clear_lthread_state();
}

std::ostream *World::lthread_print_stream() const {
    if (lthread_prints_ != nullptr) {
        return lthread_prints_.get();
    } else {
        return &std::cerr;
    }
}

void World::serialize(StreamBuffer *sb) {
    assert(stack_is_clear());
    assert(redirects_.empty());
    // int64_t wc0 = sb->total_writes();
    lua_snap_.serialize(sb);
    id_global_pool_.serialize(sb);
    sb->write_int64(clock_);
    thread_sched_.serialize(sb);
    sb->write_uint32(tangibles_.size());
    for (const auto &p : tangibles_) {
        sb->write_int64(p.first);
        p.second->serialize(sb);
    }
    // int64_t wc1 = sb->total_writes();
    // std::cerr << "World serialized to " << wc1-wc0 << " bytes." << std::endl;
    assert(stack_is_clear());
}

void World::deserialize(StreamBuffer *sb) {
    assert(stack_is_clear());
    redirects_.clear();
    lua_snap_.deserialize(sb);
    id_global_pool_.deserialize(sb);
    clock_ = sb->read_int64();
    thread_sched_.deserialize(sb);
    // Mark all tangibles for deletion by setting ID to zero.
    for (const auto &p : tangibles_) {
        p.second->plane_item_.set_id(0);
    }
    // Deserialize tangibles.
    size_t ntan = sb->read_uint32();
    for (size_t i = 0; i < ntan; i++) {
        int64_t id = sb->read_int64();
        UniqueTangible &t = tangibles_[id];
        if (t == nullptr) {
            t.reset(new Tangible(this, id));
        } else {
            t->plane_item_.set_id(id);
        }
        t->deserialize(sb);
    }
    // Delete tangibles that didn't get deserialized.
    for (auto iter = tangibles_.begin(); iter != tangibles_.end(); ) {
        if (iter->second->plane_item_.id() == 0) {
            tangibles_.erase(iter++);
        } else {
            ++iter;
        }
    }
    assert(stack_is_clear());
}

void World::snapshot() {
    assert(snapshot_.empty());
    serialize(&snapshot_);
    assert(!snapshot_.empty());
}

void World::rollback() {
    assert(!snapshot_.empty());
    deserialize(&snapshot_);
    assert(snapshot_.empty());
}