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

#include "sessions_cmd.h"

#include "../browser_launcher.h"
#include "../zenserviceclient.h"

#include <zencore/compactbinary.h>
#include <zencore/compactbinaryutil.h>
#include <zencore/fmtutils.h>
#include <zencore/logbase.h>
#include <zencore/logging.h>
#include <zencore/logging/helpers.h>
#include <zencore/string.h>
#include <zencore/thread.h>
#include <zencore/uid.h>
#include <zenhttp/httpclient.h>
#include <zenhttp/httpwsclient.h>
#include <zenutil/consoletui.h>

#include <algorithm>
#include <atomic>
#include <chrono>
#include <csignal>
#include <vector>

namespace zen {

using namespace std::literals;

namespace {

	// Render a TimeSpan as a compact human-readable duration: "12d 3h", "2h 14m",
	// "45m 02s", or "12.345s". Picks the two most-significant non-zero units.
	std::string FormatDuration(const TimeSpan& Span)
	{
		const int Days	  = Span.GetDays();
		const int Hours	  = Span.GetHours();
		const int Minutes = Span.GetMinutes();
		const int Seconds = Span.GetSeconds();

		ExtendableStringBuilder<32> Out;
		if (Days > 0)
		{
			fmt::format_to(StringBuilderAppender(Out), "{}d {}h", Days, Hours);
		}
		else if (Hours > 0)
		{
			fmt::format_to(StringBuilderAppender(Out), "{}h {:02}m", Hours, Minutes);
		}
		else if (Minutes > 0)
		{
			fmt::format_to(StringBuilderAppender(Out), "{}m {:02}s", Minutes, Seconds);
		}
		else
		{
			fmt::format_to(StringBuilderAppender(Out), "{}.{:03}s", Seconds, Span.GetFractionMilli());
		}
		return Out.ToString();
	}

	// Short ISO-like timestamp without fractional seconds — matches what 'ps'-style
	// listings expect when scanning by eye.
	std::string FormatTimestamp(const DateTime& When) { return When.ToString("%Y-%m-%dT%H:%M:%S"); }

	std::string_view OrDash(std::string_view Value) { return Value.empty() ? "-"sv : Value; }

	// File-static abort flag for `sessions tail --follow`. Mirrors the pattern in
	// projectstore_cmd.cpp: a signal handler can only touch atomic state, and we
	// don't want to share a flag with the global SignalCounter / other tail-style
	// commands. RAII'd by ScopedSignalHandler so the previous handler is
	// restored when Run() returns.
	std::atomic<bool> g_TailAbortFlag{false};

	void TailSignalHandler(int SigNum)
	{
		if (SigNum == SIGINT)
		{
			g_TailAbortFlag.store(true, std::memory_order_relaxed);
		}
#if ZEN_PLATFORM_WINDOWS
		if (SigNum == SIGBREAK)
		{
			g_TailAbortFlag.store(true, std::memory_order_relaxed);
		}
#endif
	}

	// Cached once on first use so we don't re-probe the console / re-toggle VT
	// mode for every line. Mirrors the static caching in TuiIsStdoutTty.
	bool ColorEnabledForTail()
	{
		static const bool Cached = []() {
			if (!TuiIsStdoutTty())
			{
				return false;
			}
			// On Windows this flips ENABLE_VIRTUAL_TERMINAL_PROCESSING so ANSI
			// sequences render. On POSIX it's a no-op and just returns true.
			return logging::EnableVTMode();
		}();
		return Cached;
	}

	// Render a single log entry in the same shape as zen::logging::FullFormatter:
	//   [YY-MM-DD HH:MM:SS.mmm] [<lvl-color>lvl</reset>] [<bold>logger</reset>] message
	// Continuation lines (embedded \n) are indented to the visible width of the
	// prefix so banners and stack traces align under the message column.
	//
	// Wire schema (see WriteLogEntry in httpsessions.cpp): timestamp (DateTime),
	// level (string, optional), logger (string, optional), message (string).
	void PrintLogEntry(CbObjectView Entry)
	{
		const DateTime		   Ts	   = Entry["timestamp"sv].AsDateTime();
		const std::string_view Level   = Entry["level"sv].AsString();
		const std::string_view Logger  = Entry["logger"sv].AsString();
		const std::string_view Message = Entry["message"sv].AsString();

		const bool UseColor = ColorEnabledForTail();

		// Resolve the level once. Wire form is the long lowercase name
		// ("info"/"warning"/"error"/...); we render the short form to mirror
		// FullFormatter, and use the enum to pick the ANSI color.
		logging::LogLevel Lvl = logging::Off;
		if (!Level.empty())
		{
			Lvl = logging::ParseLogLevelString(Level);
		}
		const bool			   HasLevel = (Lvl < logging::LogLevelCount && Lvl != logging::Off);
		const std::string_view LvlShort = HasLevel ? logging::ShortToString(Lvl) : "---"sv;

		ExtendableStringBuilder<256> Prefix;
		size_t						 VisibleWidth = 0;

		// Timestamp — full date so a tail that crosses midnight stays unambiguous.
		fmt::format_to(StringBuilderAppender(Prefix),
					   "[{:02}-{:02}-{:02} {:02}:{:02}:{:02}.{:03}] ",
					   Ts.GetYear() % 100,
					   Ts.GetMonth(),
					   Ts.GetDay(),
					   Ts.GetHour(),
					   Ts.GetMinute(),
					   Ts.GetSecond(),
					   Ts.GetMillisecond());
		VisibleWidth = Prefix.Size();

		// [LVL] — short form. Color goes inside the brackets (matches FullFormatter
		// so the brackets themselves stay grep-friendly even on a colorized line).
		Prefix.Append('[');
		if (UseColor && HasLevel)
		{
			Prefix.Append(logging::helpers::AnsiColorForLevel(Lvl));
		}
		Prefix.Append(LvlShort);
		if (UseColor && HasLevel)
		{
			Prefix.Append(logging::helpers::kAnsiReset);
		}
		Prefix.Append("] ");
		VisibleWidth += 1 + LvlShort.size() + 2;

		// [logger] — bold. Omitted when logger name is empty (mirrors FullFormatter).
		if (!Logger.empty())
		{
			if (UseColor)
			{
				Prefix.Append("\033[1m"sv);
			}
			Prefix.Append('[');
			Prefix.Append(Logger);
			Prefix.Append(']');
			if (UseColor)
			{
				Prefix.Append(logging::helpers::kAnsiReset);
			}
			Prefix.Append(' ');
			VisibleWidth += 1 + Logger.size() + 1 + 1;
		}

		auto NextLine = [](std::string_view Text, size_t Pos) -> std::pair<std::string_view, size_t> {
			const size_t	 End = Text.find('\n', Pos);
			std::string_view Line;
			if (End == std::string_view::npos)
			{
				Line = Text.substr(Pos);
				Pos	 = Text.size();
			}
			else
			{
				Line = Text.substr(Pos, End - Pos);
				Pos	 = End + 1;
			}
			// Strip a trailing CR so CRLF-terminated lines don't render the carriage
			// return (which on terminals would slam the cursor back to column 0).
			if (!Line.empty() && Line.back() == '\r')
			{
				Line.remove_suffix(1);
			}
			return {Line, Pos};
		};

		size_t Pos	 = 0;
		bool   First = true;
		while (true)
		{
			auto [Line, NextPos] = NextLine(Message, Pos);

			if (First)
			{
				ZEN_CONSOLE("{}{}", Prefix.ToView(), Line);
				First = false;
			}
			else
			{
				// Continuation: pad with the same VISIBLE width as the prefix so
				// the message column lines up regardless of the embedded ANSI
				// bytes. Skip a trailing empty line so a message ending in '\n'
				// doesn't add a stray blank row.
				if (NextPos >= Message.size() && Line.empty())
				{
					break;
				}
				ZEN_CONSOLE("{:<{}}{}", "", VisibleWidth, Line);
			}

			if (NextPos >= Message.size())
			{
				break;
			}
			Pos = NextPos;
		}
	}

	// WebSocket subscriber for `tail --follow`. The server pushes binary CB
	// frames shaped exactly like the HTTP cursor response (see
	// HttpSessionsService::BroadcastLogAppended), so we parse and hand each
	// entry to PrintLogEntry. Used when /sessions/ws is reachable; if the
	// upgrade fails (older server, blocked port, unix socket transport) the
	// caller falls back to the HTTP polling loop.
	class TailWsSubscriber : public IWsClientHandler
	{
	public:
		enum class State : uint8_t
		{
			Connecting,
			Open,
			Closed
		};

		TailWsSubscriber(std::string SessionIdHex, uint64_t StartCursor) : m_SessionIdHex(std::move(SessionIdHex)), m_Cursor(StartCursor) {}

		// Set after construction so OnWsOpen can SendText through the owning
		// client. The lifetime contract: the HttpWsClient is declared after
		// this subscriber in the caller's scope, so it tears down first and no
		// callbacks fire on a dangling m_Client.
		void  SetClient(HttpWsClient* Client) { m_Client = Client; }
		State GetState() const { return m_State.load(std::memory_order_acquire); }

		void OnWsOpen() override
		{
			// Subscribe to log push for our session, starting after the cursor
			// we already drained via the initial HTTP fetch. Wire format
			// matches the dashboard (see frontend/html/pages/sessions.js).
			ExtendableStringBuilder<128> Frame;
			fmt::format_to(StringBuilderAppender(Frame), R"({{"type":"sub_log","session":"{}","cursor":{}}})", m_SessionIdHex, m_Cursor);
			if (m_Client)
			{
				m_Client->SendText(Frame.ToView());
			}
			m_State.store(State::Open, std::memory_order_release);
		}

		void OnWsMessage(const WebSocketMessage& Msg) override
		{
			// The text-frame path on this socket is the periodic session-list
			// snapshot — irrelevant for tail. We only care about the binary
			// log-delta frames.
			if (Msg.Opcode != WebSocketOpcode::kBinary)
			{
				return;
			}
			IoBuffer		Payload = Msg.Payload;
			CbValidateError Err		= CbValidateError::None;
			CbObject		Frame	= ValidateAndReadCompactBinaryObject(std::move(Payload), Err);
			if (Err != CbValidateError::None)
			{
				return;
			}
			m_Cursor = Frame["cursor"sv].AsUInt64(m_Cursor);
			for (CbFieldView Entry : Frame["entries"sv].AsArrayView())
			{
				PrintLogEntry(Entry.AsObjectView());
			}
		}

		void OnWsClose(uint16_t /*Code*/, std::string_view /*Reason*/) override { m_State.store(State::Closed, std::memory_order_release); }

	private:
		HttpWsClient*	   m_Client = nullptr;
		std::string		   m_SessionIdHex;
		uint64_t		   m_Cursor = 0;
		std::atomic<State> m_State{State::Connecting};
	};

}  // namespace

////////////////////////////////////////////////////////////////////////////////
// SessionsSubCmdBase

SessionsSubCmdBase::SessionsSubCmdBase(std::string_view Name, std::string_view Description) : ZenSubCmdBase(Name, Description)
{
	m_SubOptions.add_option("", "u", "hosturl", ZenCmdBase::kHostUrlHelp, cxxopts::value(m_HostName)->default_value(""), "<hosturl>");
}

////////////////////////////////////////////////////////////////////////////////
// SessionsLsSubCmd

SessionsLsSubCmd::SessionsLsSubCmd() : SessionsSubCmdBase("ls", "List sessions known to the zenserver")
{
	m_SubOptions.add_option("",
							"s",
							"status",
							"Filter by status: 'active', 'ended', or 'all' (default)",
							cxxopts::value(m_StatusFilter)->default_value("all"),
							"<status>");
}

void
SessionsLsSubCmd::Run(const ZenCliOptions& /*GlobalOptions*/)
{
	if (m_StatusFilter != "active" && m_StatusFilter != "ended" && m_StatusFilter != "all")
	{
		throw OptionParseException("--status must be 'active', 'ended', or 'all'", m_SubOptions.help());
	}

	ZenServiceClient Service({.HostSpec = m_HostName, .CommandName = "sessions.ls"});
	HttpClient&		 Http = Service.Http();

	const std::string Path = fmt::format("/sessions/?status={}", m_StatusFilter);

	HttpClient::Response Result = Http.Get(Path, HttpClient::Accept(ZenContentType::kCbObject));
	if (!Result)
	{
		Result.ThrowError("failed to list sessions"sv);
	}

	CbObjectView Response = Result.AsObject();
	const Oid	 SelfId	  = Response["self_id"sv].AsObjectId();
	CbArrayView	 Sessions = Response["sessions"sv].AsArrayView();

	const DateTime Now = DateTime::Now();

	// Buffer the entries so we can present them in a stable order: active
	// sessions first, then ended; within each group sorted by creation time
	// (newest first — recent sessions are usually what the operator cares
	// about). The CbObjectView keeps a view into the response buffer, which
	// lives as long as `Result`.
	struct Row
	{
		CbObjectView Obj;
		DateTime	 CreatedAt{0};
		bool		 IsEnded = false;
	};
	std::vector<Row> Rows;
	Rows.reserve(Sessions.Num());
	for (CbFieldView Entry : Sessions)
	{
		CbObjectView Obj = Entry.AsObjectView();
		Rows.push_back({.Obj = Obj, .CreatedAt = Obj["created_at"sv].AsDateTime(), .IsEnded = Obj["ended_at"sv].HasValue()});
	}
	std::sort(Rows.begin(), Rows.end(), [](const Row& A, const Row& B) {
		if (A.IsEnded != B.IsEnded)
		{
			return !A.IsEnded;	// active rows first
		}
		return A.CreatedAt > B.CreatedAt;  // newest first within each group
	});

	// ID column is 25 chars wide: 24 chars of Oid plus a one-char self-marker
	// slot ('*' for the self session, otherwise a space). Keeping the slot in
	// every row lines up the trailing columns whether or not the marker is
	// present.
	ZEN_CONSOLE("{:<25}  {:<6}  {:<24}  {:>6}  {:<19}  {:>10}  {:>5}",
				"ID",
				"STATUS",
				"APPNAME/MODE",
				"PID",
				"CREATED",
				"DURATION",
				"LOGS");

	for (const Row& R : Rows)
	{
		const Oid  Id	  = R.Obj["id"sv].AsObjectId();
		const bool IsSelf = (SelfId != Oid::Zero && Id == SelfId);

		ExtendableStringBuilder<32> IdField;
		Id.ToString(IdField);
		IdField.Append(IsSelf ? '*' : ' ');

		// Combine appname and mode into a single column (e.g. "zen/sessions.tail")
		// to save horizontal real estate. Mode is omitted when empty so a session
		// with only an appname renders as "zenserver" rather than "zenserver/".
		const std::string_view		Appname = R.Obj["appname"sv].AsString();
		const std::string_view		Mode	= R.Obj["mode"sv].AsString();
		ExtendableStringBuilder<48> AppMode;
		if (Appname.empty() && Mode.empty())
		{
			AppMode.Append('-');
		}
		else
		{
			AppMode.Append(OrDash(Appname));
			if (!Mode.empty())
			{
				AppMode.Append('/');
				AppMode.Append(Mode);
			}
		}

		const DateTime EndAt	= R.IsEnded ? R.Obj["ended_at"sv].AsDateTime() : Now;
		const TimeSpan Duration = EndAt - R.CreatedAt;

		ZEN_CONSOLE("{:<25}  {:<6}  {:<24}  {:>6}  {:<19}  {:>10}  {:>5}",
					IdField.ToView(),
					R.IsEnded ? "ended"sv : "active"sv,
					AppMode.ToView(),
					R.Obj["pid"sv].AsUInt32(0),
					FormatTimestamp(R.CreatedAt),
					FormatDuration(Duration),
					R.Obj["log_count"sv].AsUInt64(0));
	}

	if (Rows.empty())
	{
		ZEN_CONSOLE("(no sessions)");
	}
	else if (SelfId != Oid::Zero)
	{
		ZEN_CONSOLE("");
		ZEN_CONSOLE("(* = the zenserver's own session)");
	}
}

////////////////////////////////////////////////////////////////////////////////
// SessionsStatusSubCmd

SessionsStatusSubCmd::SessionsStatusSubCmd() : SessionsSubCmdBase("status", "Show top-level status of the sessions service")
{
}

void
SessionsStatusSubCmd::Run(const ZenCliOptions& /*GlobalOptions*/)
{
	ZenServiceClient Service({.HostSpec = m_HostName, .CommandName = "sessions.status"});
	HttpClient&		 Http = Service.Http();

	// Two cheap reads: the stats handler exposes the operational counters, and the
	// session list lets us split the count into active/ended without per-state
	// endpoints (the service itself doesn't expose those separately).
	HttpClient::Response StatsResult = Http.Get("/stats/sessions", HttpClient::Accept(ZenContentType::kCbObject));
	if (!StatsResult)
	{
		StatsResult.ThrowError("failed to get sessions stats"sv);
	}

	HttpClient::Response ListResult = Http.Get("/sessions/?status=all", HttpClient::Accept(ZenContentType::kCbObject));
	if (!ListResult)
	{
		ListResult.ThrowError("failed to list sessions"sv);
	}

	CbObjectView Stats		   = StatsResult.AsObject();
	CbObjectView Sessions	   = Stats["sessions"sv].AsObjectView();
	CbFieldView	 RequestsField = Stats["requests"sv];

	CbObjectView ListObj	 = ListResult.AsObject();
	const Oid	 SelfId		 = ListObj["self_id"sv].AsObjectId();
	CbArrayView	 SessionsArr = ListObj["sessions"sv].AsArrayView();

	uint64_t ActiveCount = 0;
	uint64_t EndedCount	 = 0;
	for (CbFieldView Entry : SessionsArr)
	{
		if (Entry.AsObjectView()["ended_at"sv].HasValue())
		{
			++EndedCount;
		}
		else
		{
			++ActiveCount;
		}
	}

	ZEN_CONSOLE("Sessions service");
	ZEN_CONSOLE("  Self id:          {}", SelfId == Oid::Zero ? "(none)"s : SelfId.ToString());
	ZEN_CONSOLE("  Active sessions:  {}", ActiveCount);
	ZEN_CONSOLE("  Ended sessions:   {}", EndedCount);
	ZEN_CONSOLE("");
	ZEN_CONSOLE("Counters");
	ZEN_CONSOLE("  Reads:            {}", Sessions["readcount"sv].AsUInt64(0));
	ZEN_CONSOLE("  Writes:           {}", Sessions["writecount"sv].AsUInt64(0));
	ZEN_CONSOLE("  Deletes:          {}", Sessions["deletecount"sv].AsUInt64(0));
	ZEN_CONSOLE("  Lists:            {}", Sessions["listcount"sv].AsUInt64(0));
	ZEN_CONSOLE("  Requests:         {}", Sessions["requestcount"sv].AsUInt64(0));
	ZEN_CONSOLE("  Bad requests:     {}", Sessions["badrequestcount"sv].AsUInt64(0));

	// Operation timing snapshot is shaped by EmitSnapshot — surface the headline
	// fields if present without binding too tightly to its layout.
	if (RequestsField.HasValue())
	{
		CbObjectView   Requests = RequestsField.AsObjectView();
		const uint64_t Total	= Requests["count"sv].AsUInt64(0);
		if (Total > 0)
		{
			ZEN_CONSOLE("");
			ZEN_CONSOLE("Request timing");
			ZEN_CONSOLE("  Total:            {}", Total);
			CbFieldView P50 = Requests["latency_p50_us"sv];
			if (P50.HasValue())
			{
				ZEN_CONSOLE("  Latency p50:      {} us", P50.AsUInt64(0));
			}
			CbFieldView P99 = Requests["latency_p99_us"sv];
			if (P99.HasValue())
			{
				ZEN_CONSOLE("  Latency p99:      {} us", P99.AsUInt64(0));
			}
		}
	}
}

////////////////////////////////////////////////////////////////////////////////
// SessionsTailSubCmd

SessionsTailSubCmd::SessionsTailSubCmd()
: SessionsSubCmdBase("tail", "Tail log output from a session (defaults to the zenserver's own session)")
{
	m_SubOptions.add_option("",
							"",
							"session",
							"Session id (24-hex). Omit to tail the zenserver's own session.",
							cxxopts::value(m_SessionId)->default_value(""),
							"<id>");
	m_SubOptions.add_option("",
							"n",
							"lines",
							"Number of recent lines to show before tailing. 0 = all buffered.",
							cxxopts::value(m_Lines)->default_value("50"),
							"<n>");
	m_SubOptions.add_option("",
							"f",
							"follow",
							"After printing initial lines, keep streaming new entries until interrupted (Ctrl-C).",
							cxxopts::value(m_Follow),
							"");
	m_SubOptions.add_option("",
							"",
							"interval-ms",
							"Poll interval in milliseconds for follow mode.",
							cxxopts::value(m_PollIntervalMs)->default_value("500"),
							"<ms>");
	m_SubOptions.parse_positional({"session"});
}

void
SessionsTailSubCmd::Run(const ZenCliOptions& /*GlobalOptions*/)
{
	ZenServiceClient Service({.HostSpec = m_HostName, .CommandName = "sessions.tail"});
	HttpClient&		 Http = Service.Http();

	// Resolve the target session id. An explicit argument wins; otherwise we
	// fetch the server's self_id from the session list (which is also what the
	// `*` marker in `sessions ls` points at).
	std::string TargetId = m_SessionId;
	if (TargetId.empty())
	{
		HttpClient::Response ListResult = Http.Get("/sessions/list", HttpClient::Accept(ZenContentType::kCbObject));
		if (!ListResult)
		{
			ListResult.ThrowError("failed to resolve self session id"sv);
		}
		const Oid SelfId = ListResult.AsObject()["self_id"sv].AsObjectId();
		if (SelfId == Oid::Zero)
		{
			throw std::runtime_error("zenserver did not register a self-session; pass an explicit session id");
		}
		TargetId = SelfId.ToString();
	}

	const std::string LogPath = fmt::format("/sessions/{}/log", TargetId);

	// Initial fetch: cursor=0 returns every entry currently buffered in the
	// session, plus the cursor we should hand back to keep a follow stream
	// gap-free. We then locally trim to the last `--lines` entries before
	// printing — no protocol affordance for "last N" so the trim happens here.
	HttpClient::Response InitialResult = Http.Get(fmt::format("{}?cursor=0", LogPath), HttpClient::Accept(ZenContentType::kCbObject));
	if (!InitialResult)
	{
		if (InitialResult.StatusCode == HttpResponseCode::NotFound)
		{
			throw std::runtime_error(fmt::format("session '{}' not found", TargetId));
		}
		InitialResult.ThrowError("failed to fetch session log"sv);
	}

	CbObjectView   Initial = InitialResult.AsObject();
	uint64_t	   Cursor  = Initial["cursor"sv].AsUInt64(0);
	CbArrayView	   Entries = Initial["entries"sv].AsArrayView();
	const uint64_t Total   = Entries.Num();
	const uint64_t SkipCnt = (m_Lines > 0 && Total > m_Lines) ? Total - m_Lines : 0;

	uint64_t Index = 0;
	for (CbFieldView Entry : Entries)
	{
		if (Index >= SkipCnt)
		{
			PrintLogEntry(Entry.AsObjectView());
		}
		++Index;
	}

	if (!m_Follow)
	{
		return;
	}

	// Follow mode. Install a Ctrl-C handler that flips the abort flag rather
	// than killing the process — RAII restores the previous handler on exit.
	g_TailAbortFlag.store(false, std::memory_order_relaxed);
	ScopedSignalHandler SigInt(SIGINT, TailSignalHandler);
#if ZEN_PLATFORM_WINDOWS
	ScopedSignalHandler SigBreak(SIGBREAK, TailSignalHandler);
#endif

	// Prefer WebSocket push (zero idle traffic, no poll lag). Fall back to the
	// polling loop below if /sessions/ws can't be reached — older servers,
	// blocked port, unix socket transport, etc.
	auto TryWebSocketTail = [&]() -> bool {
		if (Service.IsUnixSocket())
		{
			// HttpWsClient supports unix sockets via UnixSocketPath, but
			// resolving and rewriting the URL is more plumbing than this
			// fallback warrants — polling works fine over unix.
			return false;
		}

		const std::chrono::milliseconds WsConnectTimeout{3000};

		std::string			 WsUrl = HttpToWsUrl(Service.HostSpec(), "/sessions/ws");
		TailWsSubscriber	 Subscriber(TargetId, Cursor);
		HttpWsClientSettings Settings;
		Settings.LogCategory	= "sessions_tail_ws";
		Settings.ConnectTimeout = WsConnectTimeout;

		HttpWsClient Client(WsUrl, Subscriber, Settings);
		Subscriber.SetClient(&Client);
		Client.Connect();

		// Wait for the upgrade to complete (or fail). Honor abort during
		// connect so a tail-then-immediate-Ctrl-C doesn't sit in here for the
		// full timeout. We add a small slack on top of ConnectTimeout so we
		// give the client's own timer a chance to flip state to Closed before
		// we declare WS unavailable.
		const auto Deadline = std::chrono::steady_clock::now() + WsConnectTimeout + std::chrono::seconds(1);
		while (Subscriber.GetState() == TailWsSubscriber::State::Connecting)
		{
			if (g_TailAbortFlag.load(std::memory_order_relaxed))
			{
				return true;  // user aborted; Client dtor closes the socket cleanly
			}
			if (std::chrono::steady_clock::now() >= Deadline)
			{
				return false;
			}
			zen::Sleep(50);
		}

		if (Subscriber.GetState() != TailWsSubscriber::State::Open)
		{
			return false;
		}

		// WS-driven follow. OnWsMessage runs on the client's IO thread and
		// prints incoming entries directly; the main thread just parks here
		// until the user aborts or the server closes the socket.
		while (Subscriber.GetState() == TailWsSubscriber::State::Open && !g_TailAbortFlag.load(std::memory_order_relaxed))
		{
			zen::Sleep(100);
		}

		if (Subscriber.GetState() == TailWsSubscriber::State::Closed && !g_TailAbortFlag.load(std::memory_order_relaxed))
		{
			ZEN_CONSOLE("(server gone)");
		}
		return true;
	};

	if (TryWebSocketTail())
	{
		return;
	}

	ZEN_DEBUG("WebSocket tail unavailable; falling back to polling at {} ms intervals", m_PollIntervalMs);

	constexpr uint32_t SleepStepMs = 50;
	while (!g_TailAbortFlag.load(std::memory_order_relaxed))
	{
		// Sleep in small chunks so a Ctrl-C reacts within ~50 ms even when the
		// user picks a long --interval-ms.
		uint32_t SleptMs = 0;
		while (SleptMs < m_PollIntervalMs && !g_TailAbortFlag.load(std::memory_order_relaxed))
		{
			zen::Sleep(SleepStepMs);
			SleptMs += SleepStepMs;
		}
		if (g_TailAbortFlag.load(std::memory_order_relaxed))
		{
			break;
		}

		HttpClient::Response Poll = Http.Get(fmt::format("{}?cursor={}", LogPath, Cursor), HttpClient::Accept(ZenContentType::kCbObject));
		if (!Poll)
		{
			// Server went away mid-tail (connection refused / timeout / DNS).
			// Treat as a clean end-of-stream rather than an error so an operator
			// that just stopped the server with `zen down` doesn't see a stack
			// trace from their tail.
			if (Poll.Error.has_value() && Poll.Error->IsConnectionError())
			{
				ZEN_CONSOLE("(server gone)");
				break;
			}
			if (Poll.StatusCode == HttpResponseCode::NotFound)
			{
				ZEN_CONSOLE("(session ended)");
				break;
			}
			Poll.ThrowError("session log poll failed"sv);
		}

		CbObjectView   P		 = Poll.AsObject();
		const uint64_t NewCursor = P["cursor"sv].AsUInt64(Cursor);
		if (NewCursor == Cursor)
		{
			continue;
		}
		Cursor = NewCursor;
		for (CbFieldView Entry : P["entries"sv].AsArrayView())
		{
			PrintLogEntry(Entry.AsObjectView());
		}
	}
}

////////////////////////////////////////////////////////////////////////////////
// SessionsUiSubCmd

SessionsUiSubCmd::SessionsUiSubCmd() : SessionsSubCmdBase("ui", "Open the sessions tab of the zenserver dashboard in a web browser")
{
}

void
SessionsUiSubCmd::Run(const ZenCliOptions& /*GlobalOptions*/)
{
	// Resolve through ZenServiceClient so we use the same host-discovery logic
	// as the rest of the sessions subcommands (and so this invocation shows up
	// as its own session in 'sessions ls', which is a feature, not a bug).
	ZenServiceClient Service({.HostSpec = m_HostName, .CommandName = "sessions.ui"});

	if (Service.IsUnixSocket())
	{
		throw std::runtime_error("Cannot open browser for a Unix domain socket connection");
	}

	// Page route in the dashboard SPA — see add_service_nav() in
	// frontend/html/pages/page.js.
	ExtendableStringBuilder<256> Url;
	Url << Service.HostSpec() << "/dashboard/?page=sessions";

	LaunchBrowser(Url.ToView());
}

////////////////////////////////////////////////////////////////////////////////
// SessionsCommand

SessionsCommand::SessionsCommand()
{
	m_Options.add_options()("h,help", "Print help");

	AddSubCommand(m_LsSubCmd);
	AddSubCommand(m_StatusSubCmd);
	AddSubCommand(m_TailSubCmd);
	AddSubCommand(m_UiSubCmd);
}

SessionsCommand::~SessionsCommand() = default;

}  // namespace zen