defmodule TTYCast do
@moduledoc """
Seekable, compressed terminal recordings for BEAM applications.
TTYCast records terminal output, input metadata, resize events, and custom
semantic events into a self-contained `.ttycast` file. Recordings are split
into independently compressed chunks with Ghostty terminal keyframes so readers
can seek by timestamp without replaying the whole file.
## Writing
TTYCast.write("demo.ttycast", [width: 120, height: 40], fn writer ->
TTYCast.Writer.write(writer, "hello\\r\\n")
end)
## Reading
cast = TTYCast.open!("demo.ttycast")
TTYCast.info(cast)
TTYCast.snapshot!(cast, time_ms: 1_000)
TTYCast.stream(cast) |> Enum.to_list()
## Recording commands
TTYCast.record(["sh", "-lc", "echo hello"], path: "demo.ttycast")
See `FORMAT.md` for the binary container layout.
"""
alias TTYCast.Cast
alias TTYCast.Writer
@type event :: Writer.event()
@doc "Starts a recording writer."
@spec start_writer(keyword()) :: GenServer.on_start()
defdelegate start_writer(opts), to: Writer, as: :start_link
@doc "Returns a collectable sink that writes iodata into the recording."
@spec into(pid()) :: TTYCast.Sink.t()
def into(writer), do: %TTYCast.Sink{writer: writer}
@doc "Opens a writer, calls `fun`, and closes the writer afterwards."
@spec write(Path.t(), keyword(), (pid() -> term())) :: {:ok, term()} | {:error, term()}
def write(path, opts \\ [], fun) when is_function(fun, 1) do
with {:ok, writer} <- Writer.start_link(Keyword.put(opts, :path, path)) do
try do
result = fun.(writer)
:ok = Writer.close(writer)
{:ok, result}
rescue
error in [RuntimeError, ArgumentError, MatchError, FunctionClauseError, CaseClauseError] ->
Writer.close(writer)
reraise error, __STACKTRACE__
end
end
end
@doc "Records a command under a real pseudo-terminal."
@spec record([String.t()] | {String.t(), [String.t()]}, keyword()) ::
{:ok, TTYCast.Recorder.result()} | {:error, term()}
def record({cmd, args}, opts) when is_binary(cmd) and is_list(args),
do: record([cmd | args], opts)
def record(command, opts), do: TTYCast.Recorder.record(command, opts)
@spec record(String.t(), [String.t()], keyword()) ::
{:ok, TTYCast.Recorder.result()} | {:error, term()}
def record(cmd, args, opts) when is_binary(cmd) and is_list(args),
do: record([cmd | args], opts)
@doc "Records a command interactively, forwarding the current terminal to the child PTY."
@spec record_interactive([String.t()], keyword()) :: {:ok, map()} | {:error, term()}
defdelegate record_interactive(command, opts), to: TTYCast.Interactive, as: :record
@doc "Opens a recording without decoding chunks eagerly."
@spec open(Path.t() | Cast.t()) :: {:ok, Cast.t()} | {:error, term()}
def open(%Cast{} = cast), do: {:ok, cast}
def open(path) when is_binary(path), do: TTYCast.Reader.open(path)
@doc "Opens a recording or raises."
@spec open!(Path.t() | Cast.t()) :: Cast.t()
def open!(%Cast{} = cast), do: cast
def open!(path) do
case open(path) do
{:ok, cast} -> cast
{:error, reason} -> raise "cannot open ttycast #{inspect(path)}: #{inspect(reason)}"
end
end
@doc "Returns a compact summary map for a recording."
@spec info(Path.t() | Cast.t()) :: map()
def info(path_or_cast) do
cast = open!(path_or_cast)
chunks = Map.get(cast.index, :chunks, [])
%{
path: cast.path,
version: cast.header.version,
codec: cast.header.codec,
width: cast.header.width,
height: cast.header.height,
cwd: cast.header.cwd,
started_at_unix_ms: cast.header.started_at_unix_ms,
duration_ms: div(Map.get(cast.index, :duration_us, 0), 1_000),
chunks: length(chunks),
events: Enum.reduce(chunks, 0, &(&1.event_count + &2)),
input_policy: cast.header.input_policy,
metadata: cast.header.metadata
}
end
@doc "Returns a Ghostty-rendered terminal snapshot at a timestamp."
@spec snapshot(Path.t() | Cast.t(), keyword()) :: {:ok, binary() | map()} | {:error, term()}
def snapshot(path_or_cast, opts \\ []) do
path_or_cast
|> open!()
|> TTYCast.Replay.snapshot(opts)
end
@doc "Returns a terminal snapshot or raises."
@spec snapshot!(Path.t() | Cast.t(), keyword()) :: binary() | map()
def snapshot!(path_or_cast, opts \\ []) do
case snapshot(path_or_cast, opts) do
{:ok, snapshot} -> snapshot
{:error, reason} -> raise "cannot snapshot ttycast: #{inspect(reason)}"
end
end
@doc "Streams decoded events lazily from chunks matching the optional time range."
@spec stream(Path.t() | Cast.t(), keyword()) :: Enumerable.t()
def stream(path_or_cast, opts \\ []) do
path_or_cast
|> open!()
|> TTYCast.Events.stream(opts)
end
@doc "Returns decoded events as a list. Prefer `stream/2` for large recordings."
@spec events(Path.t() | Cast.t(), keyword()) :: [event()]
def events(path_or_cast, opts \\ []) do
path_or_cast
|> open!()
|> TTYCast.Events.list(opts)
end
@doc "Finds times where plain terminal snapshots contain a string or regex match."
@spec find(Path.t() | Cast.t(), binary() | Regex.t(), keyword()) :: [
%{time_ms: non_neg_integer(), match: binary()}
]
def find(path_or_cast, pattern, opts \\ []) do
path_or_cast
|> open!()
|> TTYCast.Search.find(pattern, opts)
end
@doc "Rebuilds the trailer/footer index from intact chunks."
@spec reindex(Path.t()) :: {:ok, map()} | {:error, term()}
defdelegate reindex(path), to: TTYCast.Reindex, as: :run
@doc "Exports a recording to a supported format."
@spec export(Path.t() | Cast.t(), :asciinema, Path.t()) :: :ok | {:error, term()}
def export(path_or_cast, :asciinema, output_path),
do: export_asciinema(path_or_cast, output_path)
@doc "Imports a recording from a supported format."
@spec import(Path.t(), :asciinema, Path.t(), keyword()) :: {:ok, Cast.t()} | {:error, term()}
def import(input_path, :asciinema, output_path, opts \\ []),
do: import_asciinema(input_path, output_path, opts)
@doc "Imports asciinema v2 JSON lines into a ttycast file."
@spec import_asciinema(Path.t(), Path.t(), keyword()) :: {:ok, Cast.t()} | {:error, term()}
def import_asciinema(input_path, output_path, opts \\ []),
do: TTYCast.Asciinema.import(input_path, output_path, opts)
@doc "Exports terminal input/output streams as asciinema v2 JSON lines."
@spec export_asciinema(Path.t() | Cast.t(), Path.t()) :: :ok | {:error, term()}
def export_asciinema(path_or_cast, output_path) do
path_or_cast
|> open!()
|> TTYCast.Asciinema.export(output_path)
end
@doc "Reads and decodes one compressed chunk by index entry."
@spec read_chunk(Path.t() | Cast.t(), map()) :: {:ok, map()} | {:error, term()}
def read_chunk(path_or_cast, chunk_index) do
path_or_cast
|> open!()
|> TTYCast.Reader.read_chunk(chunk_index)
end
@doc "Reads one compressed chunk or raises."
@spec read_chunk!(Path.t() | Cast.t(), map()) :: map()
def read_chunk!(path_or_cast, chunk_index) do
case read_chunk(path_or_cast, chunk_index) do
{:ok, chunk} ->
chunk
{:error, reason} ->
raise "cannot read ttycast chunk #{inspect(chunk_index.seq)}: #{inspect(reason)}"
end
end
end