path: root/zenserver/zenserver.cpp

// Copyright Epic Games, Inc. All Rights Reserved.

#include <zencore/filesystem.h>
#include <zencore/fmtutils.h>
#include <zencore/httpserver.h>
#include <zencore/iobuffer.h>
#include <zencore/refcount.h>
#include <zencore/scopeguard.h>
#include <zencore/string.h>
#include <zencore/thread.h>
#include <zencore/timer.h>
#include <zencore/windows.h>
#include <zenserverprocess.h>
#include <zenstore/cas.h>
#include <zenstore/cidstore.h>

#include <fmt/format.h>
#include <mimalloc-new-delete.h>
#include <mimalloc.h>
#include <spdlog/spdlog.h>
#include <asio.hpp>
#include <exception>
#include <list>
#include <lua.hpp>
#include <optional>
#include <regex>
#include <unordered_map>

//////////////////////////////////////////////////////////////////////////
// We don't have any doctest code in this file but this is needed to bring
// in some shared code into the executable

#define DOCTEST_CONFIG_IMPLEMENT
#include <doctest/doctest.h>
#undef DOCTEST_CONFIG_IMPLEMENT

//////////////////////////////////////////////////////////////////////////

#include "casstore.h"
#include "config.h"
#include "diag/logging.h"

#define SENTRY_BUILD_STATIC 1
#include <sentry.h>
#pragma comment(lib, "sentry.lib")
#pragma comment(lib, "dbghelp.lib")
#pragma comment(lib, "winhttp.lib")
#pragma comment(lib, "version.lib")

//////////////////////////////////////////////////////////////////////////
// Services
//

#include "admin/admin.h"
#include "cache/kvcache.h"
#include "cache/structuredcache.h"
#include "cache/structuredcachestore.h"
#include "compute/apply.h"
#include "diag/diagsvcs.h"
#include "experimental/usnjournal.h"
#include "projectstore.h"
#include "testing/launch.h"
#include "upstream/jupiter.h"
#include "upstream/upstreamcache.h"
#include "upstream/zen.h"
#include "zenstore/gc.h"
#include "zenstore/scrub.h"

#define ZEN_APP_NAME "Zen store"

class ZenServer
{
public:
	void Initialize(ZenServiceConfig& ServiceConfig, int BasePort, int ParentPid)
	{
		using namespace fmt::literals;
		spdlog::info(ZEN_APP_NAME " initializing");

		m_DebugOptionForcedCrash = ServiceConfig.ShouldCrash;

		if (ParentPid)
		{
			m_Process.Initialize(ParentPid);

			if (!m_Process.IsValid())
			{
				spdlog::warn("Unable to initialize process handle for specified parent pid #{}", ParentPid);
			}
			else
			{
				spdlog::info("Using parent pid #{} to control process lifetime", ParentPid);
			}
		}

		// Initialize/check mutex based on base port

		std::string MutexName = "zen_{}"_format(BasePort);

		if (zen::NamedMutex::Exists(MutexName) || (m_ServerMutex.Create(MutexName) == false))
		{
			throw std::exception("Failed to create mutex '{}' - is another instance already running?"_format(MutexName).c_str());
		}

		// Ok so now we're configured, let's kick things off

		spdlog::info("initializing storage");

		zen::CasStoreConfiguration Config;
		Config.RootDirectory = m_DataRoot / "cas";

		m_CasStore->Initialize(Config);

		m_CidStore = std::make_unique<zen::CidStore>(*m_CasStore, m_DataRoot / "cid");

		spdlog::info("instantiating project service");

		m_ProjectStore = new zen::ProjectStore(*m_CasStore, m_DataRoot / "projects");
		m_HttpProjectService.reset(new zen::HttpProjectService{*m_CasStore, m_ProjectStore});
		m_LocalProjectService = zen::LocalProjectService::New(*m_CasStore, m_ProjectStore);

		spdlog::info("instantiating compute services");

		std::filesystem::path SandboxDir = m_DataRoot / "exec" / "sandbox";
		zen::CreateDirectories(SandboxDir);
		m_HttpLaunchService = std::make_unique<zen::HttpLaunchService>(*m_CasStore, SandboxDir);

		std::filesystem::path ApplySandboxDir = m_DataRoot / "exec" / "apply";
		zen::CreateDirectories(ApplySandboxDir);
		m_HttpFunctionService = std::make_unique<zen::HttpFunctionService>(*m_CasStore, *m_CidStore, ApplySandboxDir);

		if (ServiceConfig.LegacyCacheEnabled)
		{
			spdlog::info("instantiating legacy cache service");
			m_CacheService.reset(new zen::HttpKvCacheService());
		}
		else
		{
			spdlog::info("NOT instantiating legacy cache service");
		}

		if (ServiceConfig.StructuredCacheEnabled)
		{
			spdlog::info("instantiating structured cache service");
			m_CacheStore = std::make_unique<ZenCacheStore>(*m_CasStore, m_DataRoot / "cache");

			std::unique_ptr<zen::UpstreamCache> UpstreamCache;
			if (ServiceConfig.UpstreamCacheEnabled)
			{
				using namespace std::literals;

				zen::UpstreamCacheOptions UpstreamOptions;
				UpstreamOptions.ThreadCount				  = 4;
				UpstreamOptions.JupiterEnabled			  = true;
				UpstreamOptions.JupiterEndpoint			  = "https://jupiter.devtools-dev.epicgames.com"sv;
				UpstreamOptions.JupiterDdcNamespace		  = "ue4.ddc"sv;
				UpstreamOptions.JupiterBlobStoreNamespace = "test.ddc"sv;
				UpstreamOptions.JupiterOAuthProvider	  = "https://epicgames.okta.com/oauth2/auso645ojjWVdRI3d0x7/v1/token"sv;
				UpstreamOptions.JupiterOAuthClientId	  = "0oao91lrhqPiAlaGD0x7"sv;
				UpstreamOptions.JupiterOAuthSecret		  = "-GBWjjenhCgOwhxL5yBKNJECVIoDPH0MK4RDuN7d"sv;

				UpstreamCache = zen::MakeUpstreamCache(UpstreamOptions, *m_CacheStore, *m_CidStore);
			}

			m_StructuredCacheService.reset(
				new zen::HttpStructuredCacheService(*m_CacheStore, *m_CasStore, *m_CidStore, std::move(UpstreamCache)));
		}
		else
		{
			spdlog::info("NOT instantiating structured cache service");
		}

		if (ServiceConfig.MeshEnabled)
		{
			StartMesh(BasePort);
		}

		m_Http.Initialize(BasePort);
		m_Http.AddEndpoint(m_HealthService);
		m_Http.AddEndpoint(m_TestService);
		m_Http.AddEndpoint(m_AdminService);

		if (m_HttpProjectService)
		{
			m_Http.AddEndpoint(*m_HttpProjectService);
		}

		m_Http.AddEndpoint(m_CasService);

		if (m_CacheService)
		{
			spdlog::info("instantiating legacy cache service");
			m_Http.AddEndpoint(*m_CacheService);
		}

		if (m_StructuredCacheService)
		{
			m_Http.AddEndpoint(*m_StructuredCacheService);
		}

		if (m_HttpLaunchService)
		{
			m_Http.AddEndpoint(*m_HttpLaunchService);
		}

		if (m_HttpFunctionService)
		{
			m_Http.AddEndpoint(*m_HttpFunctionService);
		}
	}

	void StartMesh(int BasePort)
	{
		spdlog::info("initializing mesh discovery");
		m_ZenMesh.Start(uint16_t(BasePort));
	}

	void Run()
	{
		if (m_Process.IsValid())
		{
			EnqueueTimer();
		}

		if (!m_TestMode)
		{
			spdlog::info("__________                _________ __                        ");
			spdlog::info("\\____    /____   ____    /   _____//  |_  ___________   ____  ");
			spdlog::info("  /     // __ \\ /    \\   \\_____  \\\\   __\\/  _ \\_  __ \\_/ __ \\ ");
			spdlog::info(" /     /\\  ___/|   |  \\  /        \\|  | (  <_> )  | \\/\\  ___/ ");
			spdlog::info("/_______ \\___  >___|  / /_______  /|__|  \\____/|__|    \\___  >");
			spdlog::info("        \\/   \\/     \\/          \\/                         \\/ ");
		}

		spdlog::info(ZEN_APP_NAME " now running");

		sentry_clear_modulecache();

		if (m_DebugOptionForcedCrash)
		{
			__debugbreak();
		}

		m_Http.Run(m_TestMode);

		spdlog::info(ZEN_APP_NAME " exiting");

		m_IoContext.stop();

		Flush();
	}

	void RequestExit(int ExitCode)
	{
		RequestApplicationExit(ExitCode);
		m_Http.RequestExit();
	}

	void Cleanup() { spdlog::info(ZEN_APP_NAME " cleaning up"); }

	void SetTestMode(bool State) { m_TestMode = State; }
	void SetDataRoot(std::filesystem::path Root) { m_DataRoot = Root; }

	void EnsureIoRunner()
	{
		if (!m_IoRunner.joinable())
		{
			m_IoRunner = std::move(std::jthread{[this] { m_IoContext.run(); }});
		}
	}

	void EnqueueTimer()
	{
		m_PidCheckTimer.expires_after(std::chrono::seconds(1));
		m_PidCheckTimer.async_wait([this](const asio::error_code&) { CheckOwnerPid(); });

		EnsureIoRunner();
	}

	void CheckOwnerPid()
	{
		if (m_Process.IsRunning())
		{
			EnqueueTimer();
		}
		else
		{
			spdlog::info(ZEN_APP_NAME " exiting since parent process id {} is gone", m_Process.Pid());

			RequestExit(0);
		}
	}

	void Flush()
	{
		if (m_CasStore)
			m_CasStore->Flush();

		if (m_CidStore)
			m_CidStore->Flush();

		if (m_CacheService)
			m_CacheService->Flush();

		if (m_StructuredCacheService)
			m_StructuredCacheService->Flush();

		if (m_ProjectStore)
			m_ProjectStore->Flush();
	}

private:
	bool				  m_TestMode = false;
	std::filesystem::path m_DataRoot;
	std::jthread		  m_IoRunner;
	asio::io_context	  m_IoContext;
	asio::steady_timer	  m_PidCheckTimer{m_IoContext};
	zen::ProcessHandle	  m_Process;
	zen::NamedMutex		  m_ServerMutex;

	zen::HttpServer									 m_Http;
	std::unique_ptr<zen::CasStore>					 m_CasStore{zen::CreateCasStore()};
	std::unique_ptr<zen::CidStore>					 m_CidStore;
	std::unique_ptr<ZenCacheStore>					 m_CacheStore;
	zen::CasGc										 m_Gc{*m_CasStore};
	zen::CasScrubber								 m_Scrubber{*m_CasStore};
	HttpTestService									 m_TestService;
	zen::HttpCasService								 m_CasService{*m_CasStore};
	std::unique_ptr<zen::HttpKvCacheService>		 m_CacheService;
	zen::RefPtr<zen::ProjectStore>					 m_ProjectStore;
	zen::Ref<zen::LocalProjectService>				 m_LocalProjectService;
	std::unique_ptr<zen::HttpLaunchService>			 m_HttpLaunchService;
	std::unique_ptr<zen::HttpProjectService>		 m_HttpProjectService;
	std::unique_ptr<zen::HttpStructuredCacheService> m_StructuredCacheService;
	HttpAdminService								 m_AdminService;
	HttpHealthService								 m_HealthService;
	zen::Mesh										 m_ZenMesh{m_IoContext};
	std::unique_ptr<zen::HttpFunctionService>		 m_HttpFunctionService;

	bool m_DebugOptionForcedCrash = false;
};

int
main(int argc, char* argv[])
{
	mi_version();

	ZenServerOptions GlobalOptions;
	ZenServiceConfig ServiceConfig;
	ParseGlobalCliOptions(argc, argv, GlobalOptions, ServiceConfig);
	InitializeLogging(GlobalOptions);

	// Initialize sentry.io client

	sentry_options_t* SentryOptions = sentry_options_new();
	sentry_options_set_dsn(SentryOptions, "https://[email protected]/5919284");
	sentry_init(SentryOptions);

	auto _ = zen::MakeGuard([&] { sentry_close(); });

	// Prototype config system, let's see how this pans out

	ParseServiceConfig(GlobalOptions.DataDir, /* out */ ServiceConfig);

	spdlog::info("zen cache server starting on port {}", GlobalOptions.BasePort);

	try
	{
		ZenServerState ServerState;
		ServerState.Initialize();
		ServerState.Sweep();

		if (ZenServerState::ZenServerEntry* Entry = ServerState.Lookup(GlobalOptions.BasePort))
		{
			// Instance already running for this port? Should double check pid

			spdlog::warn("Looks like there is already a process listening to this port (pid: {})", Entry->Pid);
		}
		else
		{
			ServerState.Register(GlobalOptions.BasePort);
		}

		std::unique_ptr<std::thread>	 ShutdownThread;
		std::unique_ptr<zen::NamedEvent> ShutdownEvent;

		zen::ExtendableStringBuilder<64> ShutdownEventName;
		ShutdownEventName << "Zen_" << GlobalOptions.BasePort << "_Shutdown";
		ShutdownEvent.reset(new zen::NamedEvent{ShutdownEventName});

		ZenServer Server;
		Server.SetDataRoot(GlobalOptions.DataDir);
		Server.SetTestMode(GlobalOptions.IsTest);
		Server.Initialize(ServiceConfig, GlobalOptions.BasePort, GlobalOptions.OwnerPid);

		// Monitor shutdown signals

		ShutdownThread.reset(new std::thread{[&] {
			spdlog::info("shutdown monitor thread waiting for shutdown signal '{}'", ShutdownEventName);
			ShutdownEvent->Wait();
			spdlog::info("shutdown signal received");
			Server.RequestExit(0);
		}});

		// If we have a parent process, establish the mechanisms we need
		// to be able to communicate with the parent

		if (!GlobalOptions.ChildId.empty())
		{
			zen::NamedEvent ParentEvent{GlobalOptions.ChildId};
			ParentEvent.Set();
		}

		Server.Run();
		Server.Cleanup();

		ShutdownEvent->Set();
		ShutdownThread->join();
	}
	catch (std::exception& e)
	{
		SPDLOG_CRITICAL("Caught exception in main: {}", e.what());
	}

	ShutdownLogging();

	return 0;
}