LemonChannels.Adapters.Email.ThreadStore (lemon_channels v0.1.0)

View Source

Per-thread state for the email adapter: which thread a message id belongs to, and what a reply on that thread needs in order to thread correctly.

Why this exists

Two independent reasons, either of which would justify it on its own.

Stateless resolution cannot stitch a chain it meets out of order. LemonChannels.Adapters.Email.thread_id/1 picks the oldest ancestor named in the message itself, which is correct for a well-behaved client. It is wrong the moment the first message the adapter sees is the middle of a chain — message C referencing B and A arrives first and seeds thread "A"; if a client then sends B naming only A, both land together only because A happens to be named in both. A client that trims References to just its parent breaks that, and the conversation splits in two. Recording message id → thread id as messages arrive closes the gap: any later message naming any known ancestor joins the existing thread.

Outbound has nowhere else to get its headers. LemonChannels.OutboundPayload carries the recipient, a thread id and the id being replied to, but no Subject and no References chain — nor should it, since those are email's business and not the platform's. Without persisted state a reply would have to invent a subject and send an empty References, which is the difference between a reply that lands inside the recipient's existing conversation and one that starts a new one.

Storage

Two LemonCore.Store tables, carried over unchanged from LemonGateway.Transports.Email so an existing deployment's threads survive the cutover:

  • :email_message_threadsmessage_id => %{"thread_id" => …}
  • :email_thread_statethread_id => %{"references" => […], "subject" => …}

Every operation degrades to a no-op when no store is running: reads answer nil and writes answer {:error, :store_unavailable} without raising, so an adapter in a runtime with no store behaves exactly like the stateless one rather than failing to accept mail.

Summary

Functions

Records an inbound message against thread_id and returns the thread's state.

Records a message this adapter sent, so the reply to it resolves to the same thread and the chain keeps growing.

The thread id for a parsed message.

The stored state for a thread, string-keyed, or %{} when there is none.

Functions

record_inbound(parsed, thread_id)

@spec record_inbound(map(), binary()) :: map()

Records an inbound message against thread_id and returns the thread's state.

The returned map is authoritative for the send that follows even when the store is unavailable, since it is computed before being written.

record_outbound(thread_id, message_id, references)

@spec record_outbound(binary() | nil, binary(), [binary()]) :: :ok

Records a message this adapter sent, so the reply to it resolves to the same thread and the chain keeps growing.

resolve(parsed, fallback)

@spec resolve(map(), binary()) :: binary()

The thread id for a parsed message.

Returns the thread of the first known ancestor — nearest first, since the immediate parent is the strongest evidence — and otherwise fallback, which is the caller's stateless computation.

state(thread_id)

@spec state(binary() | nil) :: map()

The stored state for a thread, string-keyed, or %{} when there is none.

Normalizes atom-keyed values too: the tables outlive process restarts and backend swaps, and older rows were written with atom keys.