Append-only conversation item log with live-tail-no-replay streaming.
Wraps a Raxol.Agent.Conversation.Store (durability) with an in-process
subscriber fan-out (liveness). The stream itself keeps NO replay buffer -- all
history lives in the store. This is the omnigent model: durability in the store,
the stream is ephemeral pub/sub.
Exact snapshot/live-tail partition
subscribe/3 returns a snapshot AND registers the caller as a subscriber in a
single serialized GenServer.call. Because append/3 is also a serialized
call, the two never interleave: a subscribe processed before an append excludes
those items from its snapshot but receives them on the live tail; a subscribe
processed after an append includes them in the snapshot and is NOT re-sent them.
So a reconnecting client sees every item exactly once with no gap and no dupe --
belt-and-suspenders deduping by Item.id is then trivial.
Live items arrive as {:conversation_item, conversation_id, item} messages.
Subscribers are monitored; their subscriptions are dropped automatically when
they exit.
Usage
{:ok, log} = Log.start_link(name: MyLog)
{:ok, %{snapshot: history, last_seq: cursor}} = Log.subscribe(log, "conv-1")
{:ok, [item]} = Log.append(log, "conv-1", %{type: :message, data: %{content: "hi"}})
# the subscriber's mailbox now holds {:conversation_item, "conv-1", item}Reconnect by passing the last seq seen: Log.subscribe(log, conv, after: cursor).
Summary
Functions
Append one item (a map) or a list of items; broadcasts to subscribers.
Returns a specification to start this module under a supervisor.
Conversation metadata (%{id, last_seq, item_count}) or {:error, :not_found}.
Page through stored items (passthrough to the store, no subscription).
List known conversation ids.
Subscribe the caller to a conversation; returns its snapshot + last seq.
Stop receiving live items for a conversation.
Types
@type conversation_id() :: binary()
@type server() :: GenServer.server()
Functions
@spec append( server(), conversation_id(), Raxol.Agent.Conversation.Store.new_item() | [Raxol.Agent.Conversation.Store.new_item()] ) :: {:ok, [Raxol.Agent.Conversation.Item.t()]}
Append one item (a map) or a list of items; broadcasts to subscribers.
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec get_conversation(server(), conversation_id()) :: {:ok, map()} | {:error, :not_found}
Conversation metadata (%{id, last_seq, item_count}) or {:error, :not_found}.
@spec items(server(), conversation_id(), keyword()) :: {:ok, [Raxol.Agent.Conversation.Item.t()]}
Page through stored items (passthrough to the store, no subscription).
@spec list_conversations(server()) :: {:ok, [conversation_id()]}
List known conversation ids.
@spec start_link(keyword()) :: GenServer.on_start()
@spec subscribe(server(), conversation_id(), keyword()) :: {:ok, %{snapshot: [Raxol.Agent.Conversation.Item.t()], last_seq: integer() | nil}}
Subscribe the caller to a conversation; returns its snapshot + last seq.
Options: :after -- only snapshot items after this seq (for reconnect).
@spec unsubscribe(server(), conversation_id()) :: :ok
Stop receiving live items for a conversation.