defmodule Symbiont.Telepathy.SMTP do @moduledoc """ Muse Inbox SMTP receiver — gen_smtp_server_session behaviour. Listens on port 25 (configurable via `:smtp_port`), accepts mail addressed to `muse@hydrascale.net` or `muse@cortex.hydrascale.net`, and hands off each accepted message to `Symbiont.Telepathy.Pipeline` for async processing. ## Data flow SMTP client → handle_RCPT/2 — validate recipient → handle_DATA/4 — parse MIME via :mimemail, log to inbox.jsonl, spawn Pipeline task via Task.Supervisor → "250 OK" — response returned to sender immediately Processing (dispatch to Claude, send reply, mark read) happens asynchronously under `Symbiont.TaskSupervisor` so the SMTP response is never delayed by upstream latency. See `Symbiont.Telepathy.Pipeline` for the pipeline logic. ## Supervision `Symbiont.Telepathy.Supervisor` starts this module as a `:supervisor`-type child (gen_smtp_server is itself a supervisor internally). ## Inbox log Every accepted message is appended to `/data/muse/inbox.jsonl` in the same schema used by the Python `smtp_handler.py`, ensuring backward compatibility with any tooling that reads that file. The `processed` flag is flipped to `true` by `Pipeline` after successful dispatch. """ @behaviour :gen_smtp_server_session require Logger @accepted_recipients ["muse@hydrascale.net", "muse@cortex.hydrascale.net"] @inbox_log "/data/muse/inbox.jsonl" @domain "cortex.hydrascale.net" defstruct from: nil, rcpt_to: [] # --------------------------------------------------------------------------- # Supervision interface # --------------------------------------------------------------------------- @doc """ Returns a child spec suitable for `Supervisor.start_link/2`. Passes `:smtp_port` through to `start_link/1`. """ def child_spec(opts) do %{ id: __MODULE__, start: {__MODULE__, :start_link, [opts]}, type: :supervisor, restart: :permanent } end @doc """ Starts the gen_smtp_server listener process. `opts` must include `:smtp_port` (integer). Falls back to 25 if absent. """ def start_link(opts) do port = Keyword.get(opts, :smtp_port, 25) Logger.info("Telepathy SMTP: starting listener on port #{port}") :gen_smtp_server.start(__MODULE__, [ [port: port, domain: @domain, address: {0, 0, 0, 0}] ]) end # --------------------------------------------------------------------------- # gen_smtp_server_session callbacks # --------------------------------------------------------------------------- @impl :gen_smtp_server_session def init(_hostname, _session_count, _peer_info, _options) do banner = "#{@domain} ESMTP Muse Inbox ready" {:ok, banner, %__MODULE__{}} end @impl :gen_smtp_server_session def handle_HELO(_hostname, state) do {:ok, state} end @impl :gen_smtp_server_session def handle_EHLO(_hostname, extensions, state) do # Return the default extension list unchanged. SIZE and 8BITMIME are # included by gen_smtp by default; we don't advertise AUTH. {:ok, extensions, state} end @impl :gen_smtp_server_session def handle_MAIL(from, state) do {:ok, %{state | from: from}} end @impl :gen_smtp_server_session def handle_MAIL_extension(_extension, state) do {:ok, state} end @impl :gen_smtp_server_session def handle_RCPT(to, state) do addr = to |> to_string() |> String.trim() |> String.trim_leading("<") |> String.trim_trailing(">") |> String.downcase() if addr in @accepted_recipients do {:ok, %{state | rcpt_to: [to | state.rcpt_to]}} else Logger.info("Telepathy SMTP: rejected recipient #{addr}") {:error, "550 Recipient #{addr} not accepted", state} end end @impl :gen_smtp_server_session def handle_RCPT_extension(_extension, state) do {:ok, state} end @doc """ Core DATA handler. Called once the entire message has been received. 1. Parses the raw SMTP payload via `:mimemail.decode/1`. 2. Appends a raw entry (processed: false) to `inbox.jsonl`. 3. Fires a `Symbiont.Telepathy.Pipeline.process/4` task under `Symbiont.TaskSupervisor` — returns 250 immediately without waiting. """ @impl :gen_smtp_server_session def handle_DATA(from, to, data, state) do case parse_email(data) do {:ok, parsed} -> entry = %{ "timestamp" => DateTime.utc_now() |> DateTime.to_iso8601(), "from" => parsed.from, "to" => Enum.join(to, ", "), "subject" => parsed.subject, "message_id" => parsed.message_id, "body" => String.slice(parsed.body, 0, 5000), "processed" => false } append_inbox_log(entry) Task.Supervisor.start_child(Symbiont.TaskSupervisor, fn -> Symbiont.Telepathy.Pipeline.process( parsed.from, parsed.subject, parsed.body, parsed.message_id ) end) ref = :crypto.strong_rand_bytes(4) |> Base.encode16(case: :lower) Logger.info("Telepathy SMTP: accepted #{ref} from #{from} (#{parsed.subject})") {:ok, ref, state} {:error, reason} -> Logger.error("Telepathy SMTP: parse error from #{from}: #{inspect(reason)}") {:error, "451 Error processing message", state} end end @impl :gen_smtp_server_session def handle_RSET(state) do {:ok, %{state | from: nil, rcpt_to: []}} end @impl :gen_smtp_server_session def handle_VRFY(_address, state) do {:error, "252 Cannot VRFY user, but will accept message and attempt delivery", state} end @impl :gen_smtp_server_session def handle_other(_verb, _args, state) do {"500 Command unrecognized", state} end @impl :gen_smtp_server_session def handle_AUTH(_type, _username, _password, state) do {:error, state} end @impl :gen_smtp_server_session def handle_STARTTLS(state) do {:ok, state} end @impl :gen_smtp_server_session def code_change(_old_vsn, state, _extra) do {:ok, state} end @impl :gen_smtp_server_session def terminate(reason, state) do if reason not in [:normal, :shutdown] do Logger.debug("Telepathy SMTP: session terminated (#{inspect(reason)})") end {:ok, state} end # --------------------------------------------------------------------------- # Private: MIME parsing via :mimemail # --------------------------------------------------------------------------- defp parse_email(data) do try do {type, subtype, headers, _params, body} = :mimemail.decode(data) from = find_header(headers, "From") || "unknown" subject = find_header(headers, "Subject") || "(no subject)" message_id = find_header(headers, "Message-ID") || "" text_body = extract_body(type, subtype, headers, body) {:ok, %{from: from, subject: subject, message_id: message_id, body: text_body}} rescue e -> {:error, Exception.message(e)} catch kind, value -> {:error, {kind, value}} end end # Plain text — use directly. defp extract_body("text", "plain", _headers, body) when is_binary(body) do String.trim(body) end # HTML with no plain-text alternative — return a placeholder matching the # Python smtp_handler.py behaviour. defp extract_body("text", "html", headers, _body) do from = find_header(headers, "From") || "sender" "[HTML content from #{from}]" end # Multipart — walk parts, prefer text/plain, fall back to HTML placeholder. defp extract_body("multipart", _subtype, _headers, parts) when is_list(parts) do plain = find_part(parts, "text", "plain") html = find_part(parts, "text", "html") cond do plain != nil -> {_, _, _, _, body} = plain String.trim(body) html != nil -> {_, _, part_headers, _, _} = html from = find_header(part_headers, "From") || "sender" "[HTML content from #{from}]" true -> "" end end defp extract_body(_type, _subtype, _headers, body) when is_binary(body) do String.trim(body) end defp extract_body(_type, _subtype, _headers, _body), do: "" defp find_part(parts, type, subtype) do Enum.find(parts, fn {^type, ^subtype, _, _, _} -> true _ -> false end) end defp find_header(headers, name) do name_lower = String.downcase(name) Enum.find_value(headers, nil, fn {n, v} -> n_str = to_string(n) if String.downcase(n_str) == name_lower, do: to_string(v), else: nil _ -> nil end) end # --------------------------------------------------------------------------- # Private: inbox.jsonl logging # --------------------------------------------------------------------------- defp append_inbox_log(entry) do File.mkdir_p!(Path.dirname(@inbox_log)) File.write!(@inbox_log, Jason.encode!(entry) <> "\n", [:append]) rescue e -> Logger.warning("Telepathy SMTP: inbox log write failed: #{Exception.message(e)}") end end