feat: initial Telepathy Elixir port — GenServer + JMAP fetch loop
Adds the Telepathy email communication layer as an OTP subtree: - Telepathy.JMAP: GenServer that discovers Fastmail JMAP session, periodically polls inbox for new emails, and provides send_email/3. Uses :httpc (no external HTTP dep). Deduplicates via seen_ids MapSet seeded from existing MessageStore entries. - Telepathy.MessageStore: JSONL-backed message persistence, compatible with the existing Python messages.jsonl format. Supports store, unread, mark_read, list, count operations. - Telepathy.Supervisor: rest_for_one supervisor (MessageStore then JMAP). - Conditionally enabled via config (:telepathy, :enabled). Disabled in test env. Can be disabled at runtime via TELEPATHY_ENABLED=false. Validated against live Fastmail: session discovery, inbox fetch (20 emails), message storage all working. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.6
parent
244feb4d7e
commit
fe5896be2d
@ -8,6 +8,10 @@ config :symbiont,
|
||||
default_tier: :haiku,
|
||||
claude_cli: "claude"
|
||||
|
||||
config :symbiont, :telepathy,
|
||||
enabled: true,
|
||||
poll_interval_ms: 60_000
|
||||
|
||||
config :logger,
|
||||
level: :info
|
||||
|
||||
|
||||
@ -7,3 +7,7 @@ end
|
||||
if data_dir = System.get_env("SYMBIONT_DATA_DIR") do
|
||||
config :symbiont, data_dir: data_dir
|
||||
end
|
||||
|
||||
if System.get_env("TELEPATHY_ENABLED") == "false" do
|
||||
config :symbiont, :telepathy, enabled: false
|
||||
end
|
||||
|
||||
@ -6,5 +6,8 @@ config :symbiont,
|
||||
heartbeat_interval_ms: :timer.hours(24),
|
||||
claude_cli: "echo"
|
||||
|
||||
config :symbiont, :telepathy,
|
||||
enabled: false
|
||||
|
||||
config :logger,
|
||||
level: :warning
|
||||
|
||||
@ -9,6 +9,9 @@ defmodule Symbiont.Application do
|
||||
├── Symbiont.Queue — persistent task queue
|
||||
├── Symbiont.Engram — cross-session memory (SQLite)
|
||||
├── Symbiont.Heartbeat — periodic health checks + queue processing
|
||||
├── Symbiont.Telepathy.Supervisor — email communication layer (if enabled)
|
||||
│ ├── Symbiont.Telepathy.MessageStore — JSONL-backed message persistence
|
||||
│ └── Symbiont.Telepathy.JMAP — Fastmail JMAP client + polling
|
||||
└── Bandit (Symbiont.API) — HTTP API
|
||||
|
||||
In test mode, the supervisor starts empty — tests manage their own processes.
|
||||
@ -59,14 +62,25 @@ defmodule Symbiont.Application do
|
||||
shutdown: @bandit_child_shutdown_ms
|
||||
)
|
||||
|
||||
children = [
|
||||
{Task.Supervisor, name: Symbiont.TaskSupervisor},
|
||||
{Symbiont.Ledger, data_dir: data_dir},
|
||||
{Symbiont.Queue, data_dir: data_dir},
|
||||
{Symbiont.Engram, db_path: engram_db},
|
||||
{Symbiont.Heartbeat, []},
|
||||
bandit_spec
|
||||
]
|
||||
telepathy_cfg = Application.get_env(:symbiont, :telepathy, [])
|
||||
telepathy_enabled = Keyword.get(telepathy_cfg, :enabled, true)
|
||||
poll_ms = Keyword.get(telepathy_cfg, :poll_interval_ms, 60_000)
|
||||
|
||||
telepathy_children =
|
||||
if telepathy_enabled do
|
||||
[{Symbiont.Telepathy.Supervisor, poll_interval_ms: poll_ms}]
|
||||
else
|
||||
[]
|
||||
end
|
||||
|
||||
children =
|
||||
[
|
||||
{Task.Supervisor, name: Symbiont.TaskSupervisor},
|
||||
{Symbiont.Ledger, data_dir: data_dir},
|
||||
{Symbiont.Queue, data_dir: data_dir},
|
||||
{Symbiont.Engram, db_path: engram_db},
|
||||
{Symbiont.Heartbeat, []}
|
||||
] ++ telepathy_children ++ [bandit_spec]
|
||||
|
||||
opts = [strategy: :rest_for_one, name: Symbiont.Supervisor]
|
||||
Supervisor.start_link(children, opts)
|
||||
|
||||
484
lib/symbiont/telepathy/jmap.ex
Normal file
484
lib/symbiont/telepathy/jmap.ex
Normal file
@ -0,0 +1,484 @@
|
||||
defmodule Symbiont.Telepathy.JMAP do
|
||||
@moduledoc """
|
||||
JMAP client GenServer for Fastmail (RFC 8620/8621).
|
||||
|
||||
Discovers the JMAP session, then periodically polls for new inbound emails.
|
||||
New emails are stored in the MessageStore. Also provides send_email/3 for
|
||||
outbound mail.
|
||||
|
||||
Uses :httpc from :inets (no external HTTP dependency).
|
||||
"""
|
||||
use GenServer
|
||||
require Logger
|
||||
|
||||
@jmap_session_url "https://api.fastmail.com/jmap/session"
|
||||
@core_cap "urn:ietf:params:jmap:core"
|
||||
@mail_cap "urn:ietf:params:jmap:mail"
|
||||
@submit_cap "urn:ietf:params:jmap:submission"
|
||||
|
||||
@default_poll_interval_ms 60_000
|
||||
@config_path "/data/telepathy/config.json"
|
||||
|
||||
defstruct [
|
||||
:token,
|
||||
:api_url,
|
||||
:account_id,
|
||||
:identity_id,
|
||||
:from_email,
|
||||
:drafts_id,
|
||||
:inbox_id,
|
||||
:poll_interval_ms,
|
||||
seen_ids: MapSet.new()
|
||||
]
|
||||
|
||||
# -- Client API --
|
||||
|
||||
def start_link(opts) do
|
||||
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
|
||||
end
|
||||
|
||||
def send_email(to, subject, body) do
|
||||
GenServer.call(__MODULE__, {:send_email, to, subject, body}, 30_000)
|
||||
end
|
||||
|
||||
def session_info do
|
||||
GenServer.call(__MODULE__, :session_info)
|
||||
end
|
||||
|
||||
def fetch_now do
|
||||
send(__MODULE__, :poll)
|
||||
:ok
|
||||
end
|
||||
|
||||
# -- Server --
|
||||
|
||||
@impl true
|
||||
def init(opts) do
|
||||
poll_ms = Keyword.get(opts, :poll_interval_ms, @default_poll_interval_ms)
|
||||
token = load_token()
|
||||
|
||||
state = %__MODULE__{
|
||||
token: token,
|
||||
poll_interval_ms: poll_ms
|
||||
}
|
||||
|
||||
send(self(), :discover_session)
|
||||
{:ok, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info(:discover_session, state) do
|
||||
case discover_session(state.token) do
|
||||
{:ok, session} ->
|
||||
Logger.info("Telepathy JMAP session discovered: account=#{session.account_id}")
|
||||
|
||||
new_state = %{
|
||||
state
|
||||
| api_url: session.api_url,
|
||||
account_id: session.account_id,
|
||||
identity_id: session.identity_id,
|
||||
from_email: session.from_email,
|
||||
drafts_id: session.drafts_id,
|
||||
inbox_id: session.inbox_id
|
||||
}
|
||||
|
||||
seen = seed_seen_ids()
|
||||
schedule_poll(new_state.poll_interval_ms)
|
||||
{:noreply, %{new_state | seen_ids: seen}}
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.error("Telepathy JMAP session discovery failed: #{inspect(reason)}")
|
||||
Process.send_after(self(), :discover_session, 10_000)
|
||||
{:noreply, state}
|
||||
end
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info(:poll, %{api_url: nil} = state) do
|
||||
schedule_poll(state.poll_interval_ms)
|
||||
{:noreply, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info(:poll, state) do
|
||||
new_state =
|
||||
case fetch_inbox_emails(state) do
|
||||
{:ok, emails} ->
|
||||
{new_count, updated_seen} = store_new_emails(emails, state.seen_ids)
|
||||
|
||||
if new_count > 0,
|
||||
do: Logger.info("Telepathy: fetched #{new_count} new email(s)")
|
||||
|
||||
%{state | seen_ids: updated_seen}
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.warning("Telepathy JMAP poll failed: #{inspect(reason)}")
|
||||
state
|
||||
end
|
||||
|
||||
schedule_poll(new_state.poll_interval_ms)
|
||||
{:noreply, new_state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_call({:send_email, to, subject, body}, _from, state) do
|
||||
result = do_send_email(state, to, subject, body)
|
||||
{:reply, result, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_call(:session_info, _from, state) do
|
||||
info = %{
|
||||
api_url: state.api_url,
|
||||
account_id: state.account_id,
|
||||
identity_id: state.identity_id,
|
||||
from_email: state.from_email,
|
||||
inbox_id: state.inbox_id,
|
||||
seen_count: MapSet.size(state.seen_ids)
|
||||
}
|
||||
|
||||
{:reply, info, state}
|
||||
end
|
||||
|
||||
# -- Private: Session Discovery --
|
||||
|
||||
defp discover_session(token) do
|
||||
with {:ok, session_resp} <- jmap_get(@jmap_session_url, token),
|
||||
api_url <- session_resp["apiUrl"],
|
||||
account_id <- get_in(session_resp, ["primaryAccounts", @mail_cap]),
|
||||
{:ok, details} <- fetch_identity_and_mailboxes(api_url, account_id, token) do
|
||||
{:ok, Map.merge(%{api_url: api_url, account_id: account_id}, details)}
|
||||
end
|
||||
end
|
||||
|
||||
defp fetch_identity_and_mailboxes(api_url, account_id, token) do
|
||||
method_calls = [
|
||||
["Identity/get", %{"accountId" => account_id, "ids" => nil}, "id0"],
|
||||
["Mailbox/get", %{"accountId" => account_id, "ids" => nil}, "mb0"]
|
||||
]
|
||||
|
||||
case jmap_call(api_url, token, method_calls) do
|
||||
{:ok, responses} ->
|
||||
identity = extract_identity(responses)
|
||||
{drafts_id, inbox_id} = extract_mailbox_ids(responses)
|
||||
|
||||
{:ok,
|
||||
%{
|
||||
identity_id: identity[:id],
|
||||
from_email: identity[:email] || "mdwyer@michaelmdwyer.com",
|
||||
drafts_id: drafts_id,
|
||||
inbox_id: inbox_id
|
||||
}}
|
||||
|
||||
error ->
|
||||
error
|
||||
end
|
||||
end
|
||||
|
||||
defp extract_identity(responses) do
|
||||
target = "mdwyer@michaelmdwyer.com"
|
||||
|
||||
Enum.find_value(responses, %{id: nil, email: nil}, fn
|
||||
["Identity/get", data, _] ->
|
||||
identities = data["list"] || []
|
||||
|
||||
match =
|
||||
Enum.find(identities, fn i ->
|
||||
String.downcase(i["email"] || "") == target
|
||||
end) || List.first(identities)
|
||||
|
||||
if match, do: %{id: match["id"], email: match["email"]}, else: %{id: nil, email: nil}
|
||||
|
||||
_ ->
|
||||
nil
|
||||
end)
|
||||
end
|
||||
|
||||
defp extract_mailbox_ids(responses) do
|
||||
Enum.find_value(responses, {nil, nil}, fn
|
||||
["Mailbox/get", data, _] ->
|
||||
mailboxes = data["list"] || []
|
||||
drafts = Enum.find(mailboxes, &(&1["role"] == "drafts"))
|
||||
inbox = Enum.find(mailboxes, &(&1["role"] == "inbox"))
|
||||
{drafts && drafts["id"], inbox && inbox["id"]}
|
||||
|
||||
_ ->
|
||||
nil
|
||||
end)
|
||||
end
|
||||
|
||||
# -- Private: Fetch Loop --
|
||||
|
||||
defp fetch_inbox_emails(state) do
|
||||
method_calls = [
|
||||
[
|
||||
"Email/query",
|
||||
%{
|
||||
"accountId" => state.account_id,
|
||||
"filter" => %{"inMailbox" => state.inbox_id},
|
||||
"sort" => [%{"property" => "receivedAt", "isAscending" => false}],
|
||||
"limit" => 20
|
||||
},
|
||||
"q0"
|
||||
],
|
||||
[
|
||||
"Email/get",
|
||||
%{
|
||||
"accountId" => state.account_id,
|
||||
"#ids" => %{
|
||||
"resultOf" => "q0",
|
||||
"name" => "Email/query",
|
||||
"path" => "/ids"
|
||||
},
|
||||
"properties" => [
|
||||
"id",
|
||||
"from",
|
||||
"to",
|
||||
"subject",
|
||||
"receivedAt",
|
||||
"textBody",
|
||||
"bodyValues",
|
||||
"preview"
|
||||
],
|
||||
"fetchTextBodyValues" => true
|
||||
},
|
||||
"g0"
|
||||
]
|
||||
]
|
||||
|
||||
case jmap_call(state.api_url, state.token, method_calls) do
|
||||
{:ok, responses} ->
|
||||
emails =
|
||||
Enum.find_value(responses, [], fn
|
||||
["Email/get", data, _] -> data["list"] || []
|
||||
_ -> nil
|
||||
end) || []
|
||||
|
||||
{:ok, emails}
|
||||
|
||||
error ->
|
||||
error
|
||||
end
|
||||
end
|
||||
|
||||
defp store_new_emails(emails, seen_ids) do
|
||||
new_emails = Enum.reject(emails, &MapSet.member?(seen_ids, &1["id"]))
|
||||
|
||||
Enum.each(new_emails, fn email ->
|
||||
body = extract_body(email)
|
||||
from = extract_from(email)
|
||||
subject = email["subject"] || "(no subject)"
|
||||
|
||||
msg = %{
|
||||
"source" => "email-in",
|
||||
"subject" => subject,
|
||||
"content" => "[From: #{from}]\n#{body}",
|
||||
"jmap_id" => email["id"],
|
||||
"received_at" => email["receivedAt"]
|
||||
}
|
||||
|
||||
Symbiont.Telepathy.MessageStore.store(msg)
|
||||
end)
|
||||
|
||||
new_seen =
|
||||
new_emails
|
||||
|> Enum.map(& &1["id"])
|
||||
|> Enum.reduce(seen_ids, &MapSet.put(&2, &1))
|
||||
|
||||
{length(new_emails), new_seen}
|
||||
end
|
||||
|
||||
defp extract_body(email) do
|
||||
body_values = email["bodyValues"] || %{}
|
||||
text_body = email["textBody"] || []
|
||||
|
||||
case text_body do
|
||||
[%{"partId" => part_id} | _] ->
|
||||
get_in(body_values, [part_id, "value"]) || email["preview"] || ""
|
||||
|
||||
_ ->
|
||||
email["preview"] || ""
|
||||
end
|
||||
end
|
||||
|
||||
defp extract_from(email) do
|
||||
case email["from"] do
|
||||
[%{"name" => name, "email" => addr} | _] when is_binary(name) and name != "" ->
|
||||
"#{name} <#{addr}>"
|
||||
|
||||
[%{"email" => addr} | _] ->
|
||||
addr
|
||||
|
||||
_ ->
|
||||
"unknown"
|
||||
end
|
||||
end
|
||||
|
||||
# -- Private: Send Email --
|
||||
|
||||
defp do_send_email(state, to, subject, body) do
|
||||
method_calls = [
|
||||
[
|
||||
"Email/set",
|
||||
%{
|
||||
"accountId" => state.account_id,
|
||||
"create" => %{
|
||||
"e1" => %{
|
||||
"mailboxIds" => %{state.drafts_id => true},
|
||||
"from" => [%{"name" => "Muse", "email" => state.from_email}],
|
||||
"to" => [%{"email" => to}],
|
||||
"subject" => subject,
|
||||
"replyTo" => [%{"name" => "Muse", "email" => "muse@hydrascale.net"}],
|
||||
"keywords" => %{"$draft" => true},
|
||||
"bodyStructure" => %{"type" => "text/plain", "partId" => "body"},
|
||||
"bodyValues" => %{"body" => %{"value" => body, "charset" => "utf-8"}}
|
||||
}
|
||||
}
|
||||
},
|
||||
"create-email"
|
||||
],
|
||||
[
|
||||
"EmailSubmission/set",
|
||||
%{
|
||||
"accountId" => state.account_id,
|
||||
"create" => %{
|
||||
"s1" => %{
|
||||
"identityId" => state.identity_id,
|
||||
"emailId" => "#e1"
|
||||
}
|
||||
}
|
||||
},
|
||||
"send"
|
||||
]
|
||||
]
|
||||
|
||||
case jmap_call(state.api_url, state.token, method_calls) do
|
||||
{:ok, responses} ->
|
||||
parse_send_response(responses)
|
||||
|
||||
{:error, reason} ->
|
||||
{:error, reason}
|
||||
end
|
||||
end
|
||||
|
||||
defp parse_send_response(responses) do
|
||||
Enum.reduce(responses, %{email_id: nil, sent: false, errors: []}, fn
|
||||
["Email/set", data, _], acc ->
|
||||
case get_in(data, ["created", "e1", "id"]) do
|
||||
nil ->
|
||||
error = get_in(data, ["notCreated", "e1"])
|
||||
%{acc | errors: ["Email create: #{inspect(error)}" | acc.errors]}
|
||||
|
||||
id ->
|
||||
%{acc | email_id: id}
|
||||
end
|
||||
|
||||
["EmailSubmission/set", data, _], acc ->
|
||||
if get_in(data, ["created", "s1"]) do
|
||||
%{acc | sent: true}
|
||||
else
|
||||
error = get_in(data, ["notCreated", "s1"])
|
||||
%{acc | errors: ["Submission: #{inspect(error)}" | acc.errors]}
|
||||
end
|
||||
|
||||
_, acc ->
|
||||
acc
|
||||
end)
|
||||
|> then(fn
|
||||
%{sent: true, email_id: id} -> {:ok, id}
|
||||
%{errors: errors} -> {:error, Enum.join(errors, "; ")}
|
||||
end)
|
||||
end
|
||||
|
||||
# -- Private: HTTP Helpers (using :httpc) --
|
||||
|
||||
defp jmap_get(url, token) do
|
||||
headers = [{~c"Authorization", ~c"Bearer #{token}"}]
|
||||
|
||||
case :httpc.request(:get, {String.to_charlist(url), headers}, [{:timeout, 15_000}], []) do
|
||||
{:ok, {{_, 200, _}, _, body}} ->
|
||||
Jason.decode(IO.iodata_to_binary(body))
|
||||
|
||||
{:ok, {{_, status, _}, _, body}} ->
|
||||
{:error, {:http, status, IO.iodata_to_binary(body)}}
|
||||
|
||||
{:error, reason} ->
|
||||
{:error, reason}
|
||||
end
|
||||
end
|
||||
|
||||
defp jmap_call(api_url, token, method_calls) do
|
||||
payload =
|
||||
Jason.encode!(%{
|
||||
"using" => [@core_cap, @mail_cap, @submit_cap],
|
||||
"methodCalls" => method_calls
|
||||
})
|
||||
|
||||
headers = [
|
||||
{~c"Authorization", ~c"Bearer #{token}"},
|
||||
{~c"Content-Type", ~c"application/json"}
|
||||
]
|
||||
|
||||
case :httpc.request(
|
||||
:post,
|
||||
{String.to_charlist(api_url), headers, ~c"application/json", payload},
|
||||
[{:timeout, 30_000}],
|
||||
[]
|
||||
) do
|
||||
{:ok, {{_, 200, _}, _, body}} ->
|
||||
case Jason.decode(IO.iodata_to_binary(body)) do
|
||||
{:ok, %{"methodResponses" => responses}} -> {:ok, responses}
|
||||
{:ok, other} -> {:error, {:unexpected_response, other}}
|
||||
{:error, _} = err -> err
|
||||
end
|
||||
|
||||
{:ok, {{_, status, _}, _, body}} ->
|
||||
{:error, {:http, status, IO.iodata_to_binary(body)}}
|
||||
|
||||
{:error, reason} ->
|
||||
{:error, reason}
|
||||
end
|
||||
end
|
||||
|
||||
defp schedule_poll(interval_ms) do
|
||||
Process.send_after(self(), :poll, interval_ms)
|
||||
end
|
||||
|
||||
defp load_token do
|
||||
case System.get_env("FASTMAIL_API_TOKEN") do
|
||||
nil -> load_token_from_config()
|
||||
"" -> load_token_from_config()
|
||||
token -> token
|
||||
end
|
||||
end
|
||||
|
||||
defp load_token_from_config do
|
||||
case File.read(@config_path) do
|
||||
{:ok, content} ->
|
||||
case Jason.decode(content) do
|
||||
{:ok, %{"fastmail_token" => token}}
|
||||
when is_binary(token) and token != "" and token != "PLACEHOLDER" ->
|
||||
token
|
||||
|
||||
_ ->
|
||||
raise "No valid fastmail_token in #{@config_path}"
|
||||
end
|
||||
|
||||
{:error, _} ->
|
||||
raise "Cannot read #{@config_path} — set FASTMAIL_API_TOKEN env var"
|
||||
end
|
||||
end
|
||||
|
||||
defp seed_seen_ids do
|
||||
case Symbiont.Telepathy.MessageStore.list(limit: 10_000) do
|
||||
messages when is_list(messages) ->
|
||||
messages
|
||||
|> Enum.map(& &1["jmap_id"])
|
||||
|> Enum.reject(&is_nil/1)
|
||||
|> MapSet.new()
|
||||
|
||||
_ ->
|
||||
MapSet.new()
|
||||
end
|
||||
end
|
||||
end
|
||||
126
lib/symbiont/telepathy/message_store.ex
Normal file
126
lib/symbiont/telepathy/message_store.ex
Normal file
@ -0,0 +1,126 @@
|
||||
defmodule Symbiont.Telepathy.MessageStore do
|
||||
@moduledoc """
|
||||
Persistent message store backed by JSONL, compatible with the Python
|
||||
Telepathy messages.jsonl format.
|
||||
|
||||
Message schema:
|
||||
%{"id" => hex12, "timestamp" => iso8601, "source" => string,
|
||||
"subject" => string, "content" => string, "read" => boolean}
|
||||
"""
|
||||
use GenServer
|
||||
require Logger
|
||||
|
||||
defstruct [:path, messages: []]
|
||||
|
||||
def start_link(opts) do
|
||||
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
|
||||
end
|
||||
|
||||
def store(message) do
|
||||
GenServer.call(__MODULE__, {:store, message})
|
||||
end
|
||||
|
||||
def unread do
|
||||
GenServer.call(__MODULE__, :unread)
|
||||
end
|
||||
|
||||
def mark_read(msg_id) do
|
||||
GenServer.call(__MODULE__, {:mark_read, msg_id})
|
||||
end
|
||||
|
||||
def list(opts \\ []) do
|
||||
GenServer.call(__MODULE__, {:list, opts})
|
||||
end
|
||||
|
||||
def count do
|
||||
GenServer.call(__MODULE__, :count)
|
||||
end
|
||||
|
||||
# -- Server --
|
||||
|
||||
@impl true
|
||||
def init(opts) do
|
||||
path = Keyword.get(opts, :path, "/data/telepathy/messages.jsonl")
|
||||
File.mkdir_p!(Path.dirname(path))
|
||||
unless File.exists?(path), do: File.write!(path, "")
|
||||
|
||||
messages = load(path)
|
||||
Logger.info("Telepathy MessageStore loaded: #{length(messages)} messages")
|
||||
{:ok, %__MODULE__{path: path, messages: messages}}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_call({:store, msg}, _from, state) do
|
||||
entry =
|
||||
Map.merge(
|
||||
%{
|
||||
"id" => generate_id(),
|
||||
"timestamp" => DateTime.utc_now() |> DateTime.to_iso8601(),
|
||||
"read" => false
|
||||
},
|
||||
msg
|
||||
)
|
||||
|
||||
new_messages = state.messages ++ [entry]
|
||||
append!(state.path, entry)
|
||||
Logger.info("Telepathy message stored: #{entry["id"]} from #{entry["source"]}")
|
||||
{:reply, {:ok, entry["id"]}, %{state | messages: new_messages}}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_call(:unread, _from, state) do
|
||||
unread = Enum.filter(state.messages, &(not &1["read"]))
|
||||
{:reply, unread, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_call({:mark_read, msg_id}, _from, state) do
|
||||
new_messages =
|
||||
Enum.map(state.messages, fn msg ->
|
||||
if msg["id"] == msg_id, do: Map.put(msg, "read", true), else: msg
|
||||
end)
|
||||
|
||||
persist!(state.path, new_messages)
|
||||
{:reply, :ok, %{state | messages: new_messages}}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_call({:list, opts}, _from, state) do
|
||||
limit = Keyword.get(opts, :limit, 50)
|
||||
msgs = Enum.take(state.messages, -limit)
|
||||
{:reply, msgs, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_call(:count, _from, state) do
|
||||
{:reply, length(state.messages), state}
|
||||
end
|
||||
|
||||
# -- Private --
|
||||
|
||||
defp load(path) do
|
||||
path
|
||||
|> File.stream!()
|
||||
|> Stream.reject(&(&1 in ["", "\n"]))
|
||||
|> Enum.map(fn line ->
|
||||
case Jason.decode(String.trim(line)) do
|
||||
{:ok, msg} -> msg
|
||||
{:error, _} -> nil
|
||||
end
|
||||
end)
|
||||
|> Enum.reject(&is_nil/1)
|
||||
end
|
||||
|
||||
defp append!(path, entry) do
|
||||
File.write!(path, Jason.encode!(entry) <> "\n", [:append])
|
||||
end
|
||||
|
||||
defp persist!(path, messages) do
|
||||
content = Enum.map_join(messages, fn m -> Jason.encode!(m) <> "\n" end)
|
||||
File.write!(path, content)
|
||||
end
|
||||
|
||||
defp generate_id do
|
||||
:crypto.strong_rand_bytes(6) |> Base.encode16(case: :lower)
|
||||
end
|
||||
end
|
||||
27
lib/symbiont/telepathy/supervisor.ex
Normal file
27
lib/symbiont/telepathy/supervisor.ex
Normal file
@ -0,0 +1,27 @@
|
||||
defmodule Symbiont.Telepathy.Supervisor do
|
||||
@moduledoc """
|
||||
Supervision subtree for Telepathy (email communication layer).
|
||||
|
||||
Children:
|
||||
- MessageStore — JSONL-backed message persistence
|
||||
- JMAP — Fastmail JMAP client with periodic inbox polling
|
||||
"""
|
||||
use Supervisor
|
||||
|
||||
def start_link(opts) do
|
||||
Supervisor.start_link(__MODULE__, opts, name: __MODULE__)
|
||||
end
|
||||
|
||||
@impl true
|
||||
def init(opts) do
|
||||
messages_path = Keyword.get(opts, :messages_path, "/data/telepathy/messages.jsonl")
|
||||
poll_interval_ms = Keyword.get(opts, :poll_interval_ms, 60_000)
|
||||
|
||||
children = [
|
||||
{Symbiont.Telepathy.MessageStore, path: messages_path},
|
||||
{Symbiont.Telepathy.JMAP, poll_interval_ms: poll_interval_ms}
|
||||
]
|
||||
|
||||
Supervisor.init(children, strategy: :rest_for_one)
|
||||
end
|
||||
end
|
||||
Loading…
Reference in New Issue
Block a user