interactive_session.ex

defmodule InteractiveSession do
use GenServer

@ansi_csi_move ~r/\x1b\[[0-9;?]*[GHfABCDEJKsu]/
@ansi_csi_other ~r/\x1b\[[0-9;?]*[A-Za-z]/
@ansi_osc ~r/\x1b\][^\x07]*\x07/
@ansi_single ~r/\x1b[=>]/
@ansi_charset ~r/\x1b[()*+][A-Za-z0-9]/
@max_buffer_lines 500
@reap_grace_ms 300_000


def start_link(opts) do
GenServer.start_link(__MODULE__, opts, name: :interactive_session)
end

def init(_opts) do
settings_path = Path.join(__DIR__, "settings.txt")
content = read_file(settings_path)
settings = case Jason.decode(content) do
	{:ok, decoded} -> decoded
	{:error, _} -> %{}
end

homepath = Map.get(settings, "homepath", "/app/")
sessions_dir = homepath <> "work/interactive/sessions/"
commands_dir = homepath <> "work/interactive/commands/"
responses_dir = homepath <> "work/interactive/responses/"

File.mkdir_p(sessions_dir)
File.mkdir_p(commands_dir)
File.mkdir_p(responses_dir)

{:ok, watcher_pid} = FileSystem.start_link(dirs: [commands_dir])
FileSystem.subscribe(watcher_pid)

state = %{
	homepath: homepath,
	sessions_dir: sessions_dir,
	commands_dir: commands_dir,
	responses_dir: responses_dir,
	watcher_pid: watcher_pid,
	sessions: %{}
}

{:ok, state, {:continue, :initial_check}}
end


def handle_continue(:initial_check, state) do
{:noreply, process_pending_commands(state)}
end


# ===== FILE WATCHER =====

def handle_info({:file_event, _watcher, {path, _events}}, state) do
if String.starts_with?(Path.basename(path), "cmd_") and String.ends_with?(path, ".json") do
	{:noreply, process_command_file(state, path)}
else
	{:noreply, state}
end
end

def handle_info({:file_event, _watcher, :stop}, state) do
{:noreply, state}
end


# ===== ERLEXEC OUTPUT =====

def handle_info({:stdout, os_pid, data}, state) do
case find_session_by_os_pid(state.sessions, os_pid) do
	nil -> {:noreply, state}
	{id, session} ->
		cleaned = strip_ansi(data)
		new_buffer = append_buffer(session.buffer, cleaned)
		append_to_file(session.log_path, cleaned)
		append_to_file(session.raw_path, data)
		updated = %{session | buffer: new_buffer, unread: append_unread(session.unread, cleaned)}
		{:noreply, update_session(state, id, updated)}
end
end

def handle_info({:stderr, os_pid, data}, state) do
case find_session_by_os_pid(state.sessions, os_pid) do
	nil -> {:noreply, state}
	{id, session} ->
		cleaned = strip_ansi(data)
		new_buffer = append_buffer(session.buffer, cleaned)
		append_to_file(session.log_path, cleaned)
		append_to_file(session.raw_path, data)
		updated = %{session | buffer: new_buffer, unread: append_unread(session.unread, cleaned)}
		{:noreply, update_session(state, id, updated)}
end
end


# ===== ERLEXEC PROCESS EXIT =====

def handle_info({:DOWN, _os_pid, :process, pid, status}, state) do
case find_session_by_pid(state.sessions, pid) do
	nil -> {:noreply, state}
	{id, session} ->
		exit_data = %{
			"code" => exit_code_from_status(status),
			"ended" => timestamp_now()
		}
		exit_path = Path.join(session.session_dir, "exit")
		File.write(exit_path, Jason.encode!(exit_data))
		updated = %{session | running: false, buffer: []}
		# reap the dead session after a grace period so `read` still works
		# briefly, but exited sessions and their buffers do not pile up forever
		Process.send_after(self(), {:reap_session, id}, @reap_grace_ms)
		{:noreply, update_session(state, id, updated)}
end
end


# Force-kill handler for sessions that don't respond to SIGTERM
def handle_info({:force_kill, id}, state) do
case Map.get(state.sessions, id) do
	nil -> {:noreply, state}
	session ->
		if session.running do
			:exec.kill(session.os_pid, 9)
		end
		{:noreply, state}
end
end

def handle_info({:retry_command, path}, state) do
if File.exists?(path) do
	{:noreply, process_command_file(state, path, 1)}
else
	{:noreply, state}
end
end

def handle_info({:reap_session, id}, state) do
case Map.get(state.sessions, id) do
	nil -> {:noreply, state}
	session ->
		if session.running do
			{:noreply, state}
		else
			{:noreply, %{state | sessions: Map.delete(state.sessions, id)}}
		end
end
end

def handle_info(_msg, state) do
{:noreply, state}
end


# ===== CAST/CALL INTERFACE =====

def handle_cast({:core_message, _from, _data}, state) do
{:noreply, state}
end

def handle_cast({:master_message, _data}, state) do
{:noreply, state}
end


# ===== COMMAND PROCESSING =====

defp process_pending_commands(state) do
case File.ls(state.commands_dir) do
	{:ok, files} ->
		files
		|> Enum.filter(&(String.starts_with?(&1, "cmd_") and String.ends_with?(&1, ".json")))
		|> Enum.sort()
		|> Enum.reduce(state, fn file, acc ->
			process_command_file(acc, Path.join(acc.commands_dir, file))
		end)
	{:error, _} -> state
end
end

defp process_command_file(state, path), do: process_command_file(state, path, 0)

# Decode BEFORE deleting so a partial read (event fired before the write
# flushed) is not lost; on a decode failure retry once after a short delay
# rather than dropping the command.
defp process_command_file(state, path, attempt) do
case File.read(path) do
	{:ok, content} ->
		case Jason.decode(String.trim(content)) do
			{:ok, data} ->
				File.rm(path)
				{response, new_state} = handle_action(state, data)
				req_id = Map.get(data, "req_id", "unknown")
				resp_path = Path.join(state.responses_dir, "resp_" <> req_id <> ".json")
				File.write(resp_path, Jason.encode!(response))
				new_state
			{:error, _} ->
				if attempt < 1 do
					Process.send_after(self(), {:retry_command, path}, 200)
					state
				else
					File.rm(path)
					state
				end
		end
	{:error, _} -> state
end
end


# ===== ACTIONS =====

defp handle_action(state, %{"action" => "start"} = data) do
command = Map.get(data, "command", "bash")
id = :crypto.strong_rand_bytes(4) |> Base.encode16(case: :lower)
session_dir = Path.join(state.sessions_dir, id)
File.mkdir_p(session_dir)

log_path = Path.join(session_dir, "log")
raw_path = Path.join(session_dir, "raw")
File.write(log_path, "")
File.write(raw_path, "")

meta = %{
	"id" => id,
	"command" => command,
	"started" => timestamp_now(),
	"user" => System.get_env("USER") || "unknown"
}
File.write(Path.join(session_dir, "meta.json"), Jason.encode!(meta))

case :exec.run(String.to_charlist(command), [:stdin, :stdout, :stderr, :monitor, :pty, {:env, [{~c"TERM", ~c"xterm-256color"}]}]) do
	{:ok, pid, os_pid} ->
		session = %{
			id: id,
			command: command,
			pid: pid,
			os_pid: os_pid,
			started: meta["started"],
			session_dir: session_dir,
			log_path: log_path,
			raw_path: raw_path,
			buffer: [],
			unread: [],
			running: true
		}
		new_state = update_session(state, id, session)
		{%{"ok" => true, "id" => id, "pid" => os_pid}, new_state}

	{:error, reason} ->
		exit_data = %{"code" => nil, "ended" => timestamp_now(), "note" => "spawn_failed: #{inspect(reason)}"}
		File.write(Path.join(session_dir, "exit"), Jason.encode!(exit_data))
		{%{"error" => "spawn_failed", "detail" => inspect(reason)}, state}
end
end

defp handle_action(state, %{"action" => "send", "session" => id} = data) do
input = Map.get(data, "input", "")
case Map.get(state.sessions, id) do
	nil ->
		{%{"error" => "session not found: " <> id}, state}
	session ->
		if not session.running do
			{%{"error" => "session already exited: " <> id}, state}
		else
			bytes = resolve_input(input)
			:exec.send(session.pid, bytes)
			{%{"ok" => true, "id" => id, "sent" => input}, state}
		end
end
end

defp handle_action(state, %{"action" => "read", "session" => id}) do
case Map.get(state.sessions, id) do
	nil ->
		{%{"error" => "session not found: " <> id}, state}
	session ->
		new_text = Enum.join(session.unread, "")
		updated = %{session | unread: []}
		new_state = update_session(state, id, updated)
		{%{"ok" => true, "id" => id, "output" => new_text, "running" => session.running}, new_state}
end
end

defp handle_action(state, %{"action" => "stop", "session" => id}) do
case Map.get(state.sessions, id) do
	nil ->
		{%{"error" => "session not found: " <> id}, state}
	session ->
		if not session.running do
			{%{"ok" => true, "id" => id, "note" => "already exited"}, state}
		else
			:exec.kill(session.os_pid, 15)
			Process.send_after(self(), {:force_kill, id}, 5000)
			{%{"ok" => true, "id" => id, "method" => "sigterm"}, state}
		end
end
end

defp handle_action(state, %{"action" => "list"}) do
sessions = Enum.map(state.sessions, fn {id, s} ->
	%{"id" => id, "command" => s.command, "started" => s.started, "running" => s.running}
end)
{sessions, state}
end

defp handle_action(state, _data) do
{%{"error" => "unknown action"}, state}
end


# ===== INPUT RESOLUTION =====

@token_map %{
"__ENTER__" => "\r",
"__TAB__" => "\t",
"__ESC__" => "\x1b",
"__BACKSPACE__" => "\x7f",
"__UP__" => "\x1b[A",
"__DOWN__" => "\x1b[B",
"__RIGHT__" => "\x1b[C",
"__LEFT__" => "\x1b[D",
"__CTRL_C__" => "\x03",
"__CTRL_D__" => "\x04",
"__CTRL_Z__" => "\x1a"
}

defp resolve_input(input) do
tokens = String.split(input)
all_tokens = Enum.all?(tokens, fn t -> Regex.match?(~r/^__[A-Z_]+__$/, t) end) and length(tokens) > 0
if all_tokens do
	tokens
	|> Enum.map(fn t -> Map.get(@token_map, t, "") end)
	|> Enum.join("")
else
	input <> "\r"
end
end


# ===== ANSI STRIPPING =====

defp strip_ansi(data) when is_binary(data) do
data
|> then(&Regex.replace(@ansi_osc, &1, ""))
|> then(&Regex.replace(@ansi_csi_move, &1, " "))
|> then(&Regex.replace(@ansi_csi_other, &1, ""))
|> then(&Regex.replace(@ansi_charset, &1, ""))
|> then(&Regex.replace(@ansi_single, &1, ""))
end

defp strip_ansi(data) when is_list(data) do
strip_ansi(IO.iodata_to_binary(data))
end


# ===== BUFFER MANAGEMENT =====

defp append_unread(unread, cleaned) do
appended = unread ++ [cleaned]
if length(appended) > @max_buffer_lines do
	["[... earlier output truncated ...]\n" | Enum.take(appended, -(@max_buffer_lines - 1))]
else
	appended
end
end

defp append_buffer(buffer, text) do
new = buffer ++ [text]
if length(new) > @max_buffer_lines do
	Enum.drop(new, length(new) - @max_buffer_lines)
else
	new
end
end


# ===== HELPERS =====

defp find_session_by_os_pid(sessions, os_pid) do
Enum.find(sessions, fn {_id, s} -> s.os_pid == os_pid end)
end

defp find_session_by_pid(sessions, pid) do
Enum.find(sessions, fn {_id, s} -> s.pid == pid end)
end

defp update_session(state, id, value) do
%{state | sessions: Map.put(state.sessions, id, value)}
end

defp exit_code_from_status(:normal), do: 0
defp exit_code_from_status({:exit_status, code}), do: code
defp exit_code_from_status(_), do: nil

defp timestamp_now do
Calendar.strftime(DateTime.utc_now(), "%Y-%m-%d %H:%M:%S")
end

defp append_to_file(path, data) when is_binary(data) do
File.write(path, data, [:append])
end

defp append_to_file(path, data) when is_list(data) do
File.write(path, IO.iodata_to_binary(data), [:append])
end

defp read_file(path) do
case File.read(path) do
	{:ok, content} -> content
	{:error, _} -> ""
end
end

end