ExWapp.Store.MessageTable (ExWapp v0.1.2)

Indexed message storage used by the built-in store adapters.

Chat metadata and message payloads deliberately live in different tables. A chat can therefore be listed or updated without copying its whole history onto the caller process heap. Message streams walk the ordered ETS table one record at a time and only materialize a list when a caller explicitly asks for a paginated list or a complete snapshot.

Custom ExWapp.Store adapters do not need to use this module. They can implement the message callbacks directly with a database cursor.

Summary

Types

order()

t()

Functions

clear(message_store)

Removes all indexed messages.

delete(message_store, jid, message_ids)

Deletes selected messages after an application has durably persisted them.

delete_chat(message_store, jid)

Deletes all messages for one chat.

extract(data)

Separates message payloads from persisted chat metadata.

find(message_store, jid, message_id)

Finds a message by chat JID and message ID.

find_by_id(message_store, message_id)

Finds the first message with an ID across all chats.

list(message_store, jid, opts \\ [])

Materializes a requested message page.

new(initial_messages \\ %{}, opts \\ [])

Creates an empty message table and imports optional initial messages.

snapshot(message_store)

Returns a complete, newest-first snapshot grouped by chat JID.

stream(message_store, jid, opts \\ [])

Returns a lazy stream over messages for one chat.

update_status(message_store, jid, message_id, status)

Updates the status of one stored message.

update_status_by_id(message_store, message_id, status)

Updates every stored message matching an ID.

upsert(message_store, jid, messages)

Inserts or replaces messages for a chat by message ID.

Types

order()

@type order() :: :newest_first | :oldest_first

t()

@type t() :: %ExWapp.Store.MessageTable{
  count_table: :ets.tid(),
  index_table: :ets.tid(),
  max_per_chat: pos_integer() | :all,
  table: :ets.tid()
}

Functions

clear(message_store)

@spec clear(t()) :: :ok

Removes all indexed messages.

delete(message_store, jid, message_ids)

@spec delete(t(), String.t(), [String.t()]) :: :ok

Deletes selected messages after an application has durably persisted them.

delete_chat(message_store, jid)

@spec delete_chat(t(), String.t()) :: :ok

Deletes all messages for one chat.

extract(data)

@spec extract(map()) :: {map(), map()}

Separates message payloads from persisted chat metadata.

It understands both the current top-level :chat_messages snapshot and the legacy %ExWapp.Chat{messages: [...]} representation. The returned data no longer contains either representation.

find(message_store, jid, message_id)

@spec find(t(), String.t(), String.t()) :: map() | nil

Finds a message by chat JID and message ID.

find_by_id(message_store, message_id)

@spec find_by_id(t(), String.t()) :: {String.t(), map()} | nil

Finds the first message with an ID across all chats.

list(message_store, jid, opts \\ [])

@spec list(t(), String.t(), keyword()) :: [map()]

Materializes a requested message page.

new(initial_messages \\ %{}, opts \\ [])

@spec new(
  map(),
  keyword()
) :: t()

Creates an empty message table and imports optional initial messages.

Initial data is a map keyed by chat JID. Duplicate message IDs are replaced by the last value imported.

:max_per_chat bounds how much history a chat keeps, oldest dropped first; :all, the default, keeps everything.

Bounding it is worth considering for any host that does not read old messages back. The built-in file adapter re-encodes its whole snapshot on every write, so unbounded history makes each arriving message cost more than the last one did — the price of storing a conversation grows with the conversation. What the window has to stay above is the library's own reads: finding a payload to resend after a rejection, recognising a redelivery, matching a media retry. Those all look at recent traffic, so a few hundred is generous; anything older is being kept for the host's benefit, not the protocol's.

snapshot(message_store)

@spec snapshot(t()) :: %{optional(String.t()) => [map()]}

Returns a complete, newest-first snapshot grouped by chat JID.

This intentionally materializes all messages and is reserved for explicit backup/export operations.

stream(message_store, jid, opts \\ [])

@spec stream(t(), String.t(), keyword()) :: Enumerable.t()

Returns a lazy stream over messages for one chat.

The table is not read until the stream is enumerated. Supported options are :order, :offset, and :limit. The default stream is newest first and unbounded.

update_status(message_store, jid, message_id, status)

@spec update_status(t(), String.t(), String.t(), atom()) :: :ok

Updates the status of one stored message.

update_status_by_id(message_store, message_id, status)

@spec update_status_by_id(t(), String.t(), atom()) :: :ok

Updates every stored message matching an ID.

upsert(message_store, jid, messages)

@spec upsert(t(), String.t(), [map()]) :: :ok

Inserts or replaces messages for a chat by message ID.