LemonCore.Store (lemon_core v0.1.0)

View Source

Persistent key-value store with pluggable backends.

Configuration

Configure the backend in your application config:

config :lemon_core, LemonCore.Store,
  backend: LemonCore.Store.SqliteBackend,
  backend_opts: [path: "/var/lib/lemon/store"]

Defaults to LemonCore.Store.EtsBackend (in-memory, ephemeral).

Instances

A store is named; the default name is LemonCore.Store and every public function defaults to it, so single-store applications never pass a server. Several stores can run in one node:

{LemonCore.Store, name: :scratch_store, backend: LemonCore.Store.EtsBackend}

LemonCore.Store.put(:scratch_store, :notes, "k", "v")

Configuration comes from start_link/1 opts first (:backend, :backend_opts, :chat_state_ttl_ms, :cached_tables, :finalize_run_hooks, :run_history_provider), falling back to Application.get_env(:lemon_core, name) — which for the default name is the historical config :lemon_core, LemonCore.Store block. Each instance gets its own LemonCore.Store.ReadCache tables.

Collaborators

This module is a storage primitive and deliberately knows nothing about run history, memory, chat platforms, or any other domain. Collaborators attach to it instead:

See LemonCore.Store.Hooks for hook shapes and failure isolation.

Read cache coherence

LemonCore.Store.ReadCache mirrors hot domains into public ETS so reads skip this GenServer. The store process is the cache's only writer, and it writes only after the backend confirms, so the cache can never advertise a value the backend rejected. Two consequences worth knowing:

  • put_chat_state/3 and put_progress_mapping/4 are synchronous. They were casts with a caller-side cache write, which made a failed backend write look like a success and let two writers on one key land out of order.
  • append_run_event/3 and finalize_run/3 stay asynchronous, because runs stream events far more often than anything reads them back. Their cache entry lands with the backend write, so a read taken before the cast drains can lag by a message. ping/1 is the barrier when that matters.

Summary

Types

A running store: its registered name (the usual case) or its pid.

Functions

Append a canonical introspection event.

Append an event to a run's record.

Returns a specification to start this module under a supervisor.

Delete a key from a named table.

Delete an agent policy by agent_id.

Delete a channel policy by channel_id.

Delete global runtime policy overrides.

Delete a session policy by session key.

Record a run's summary and fire the store's finalize-run hooks.

Get a value from a named table.

Get an agent policy by agent_id.

Get a channel policy by channel_id.

Get a specific run by ID.

Get run history for a session key, ordered by most recent first.

Get run history for a session key from a specific store instance.

Get global runtime policy overrides.

Get a session policy by session key.

List all key-value pairs in a named table.

List all channel policies.

List introspection events.

List introspection events from a specific store instance.

List runtime policy entries.

List all session policies.

Performs a non-mutating store/backend liveness check.

Put a value into a named table.

Put a value into a named table only if the key does not already exist.

Put global runtime policy overrides.

Mirror table into this store's read cache from now on.

Register a hook invoked after each run is finalized with a session key.

Stop mirroring table; takes effect on the store's next start.

Remove a previously registered finalize-run hook.

Types

server()

@type server() :: atom() | pid()

A running store: its registered name (the usual case) or its pid.

Read-cache fast paths are only available when addressing a store by name.

Functions

append_introspection_event(server \\ __MODULE__, event)

@spec append_introspection_event(server(), map()) :: :ok | {:error, term()}

Append a canonical introspection event.

append_run_event(server \\ __MODULE__, run_id, event)

@spec append_run_event(server(), term(), term()) :: :ok

Append an event to a run's record.

Asynchronous, because runs stream events far more often than anything reads them back. The store updates the cache after the backend write, so a read taken before the cast drains can lag by a message; ping/1 is the barrier when a caller needs the write to have landed.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

delete(server \\ __MODULE__, table, key)

@spec delete(server(), table :: atom(), key :: term()) :: :ok | {:error, term()}

Delete a key from a named table.

delete_agent_policy(server \\ __MODULE__, agent_id)

@spec delete_agent_policy(server(), agent_id :: term()) :: :ok

Delete an agent policy by agent_id.

delete_channel_policy(server \\ __MODULE__, channel_id)

@spec delete_channel_policy(server(), channel_id :: term()) :: :ok

Delete a channel policy by channel_id.

delete_chat_state(server \\ __MODULE__, scope)

@spec delete_chat_state(server(), term()) :: :ok | {:error, term()}

delete_progress_mapping(server \\ __MODULE__, scope, progress_msg_id)

@spec delete_progress_mapping(server(), term(), integer()) :: :ok | {:error, term()}

delete_runtime_policy(server \\ __MODULE__)

@spec delete_runtime_policy(server()) :: :ok

Delete global runtime policy overrides.

delete_session_policy(server \\ __MODULE__, session_key)

@spec delete_session_policy(server(), session_key :: term()) :: :ok

Delete a session policy by session key.

finalize_run(server \\ __MODULE__, run_id, summary)

@spec finalize_run(server(), term(), map()) :: :ok

Record a run's summary and fire the store's finalize-run hooks.

When the summary carries a non-empty :session_key, every hook registered for this store (see register_finalize_run_hook/2) is invoked with:

%{
  store: atom(),        # the store's name
  run_id: term(),
  record: map(),        # %{events: [...], summary: ..., started_at: ...}
  summary: map(),
  session_key: String.t(),
  started_at: integer()
}

Hooks run in the store process, in registration order, and failures are isolated. Runs without a session key do not fire hooks.

get(server \\ __MODULE__, table, key)

@spec get(server(), table :: atom(), key :: term()) :: term() | nil

Get a value from a named table.

Returns nil if the key doesn't exist.

get_agent_policy(server \\ __MODULE__, agent_id)

@spec get_agent_policy(server(), agent_id :: term()) :: map() | nil

Get an agent policy by agent_id.

get_channel_policy(server \\ __MODULE__, channel_id)

@spec get_channel_policy(server(), channel_id :: term()) :: map() | nil

Get a channel policy by channel_id.

get_chat_state(server \\ __MODULE__, scope)

@spec get_chat_state(server(), term()) :: LemonCore.ChatState.t() | map() | nil

get_run(server \\ __MODULE__, run_id)

@spec get_run(server(), term()) :: map() | nil

Get a specific run by ID.

get_run_by_progress(server \\ __MODULE__, scope, progress_msg_id)

@spec get_run_by_progress(server(), term(), integer()) :: term() | nil

get_run_history(session_key, opts \\ [])

@spec get_run_history(
  binary(),
  keyword()
) :: [{term(), map()}]

Get run history for a session key, ordered by most recent first.

Options

  • :limit - Maximum number of runs to return (default: 10)

Returns a list of {run_id, %{events: [...], summary: %{...}, session_key: key, started_at: ts}}.

Session keys are binaries, and the guard says so: it is what stops get_run_history(:my_store, opts) — which reads as "from this instance" but means "for this session on the default store" — from silently returning the wrong answer. Address an instance with get_run_history/3.

get_run_history(server, session_key, opts)

@spec get_run_history(server(), term(), keyword()) :: [{term(), map()}]

Get run history for a session key from a specific store instance.

get_runtime_policy(server \\ __MODULE__)

@spec get_runtime_policy(server()) :: map() | nil

Get global runtime policy overrides.

get_session_policy(server \\ __MODULE__, session_key)

@spec get_session_policy(server(), session_key :: term()) :: map() | nil

Get a session policy by session key.

list(server \\ __MODULE__, table)

@spec list(server(), table :: atom()) :: [{term(), term()}]

List all key-value pairs in a named table.

list_agent_policies(server \\ __MODULE__)

@spec list_agent_policies(server()) :: [{term(), map()}]

List all agent policies.

list_channel_policies(server \\ __MODULE__)

@spec list_channel_policies(server()) :: [{term(), map()}]

List all channel policies.

list_introspection_events(opts \\ [])

@spec list_introspection_events(keyword()) :: [map()]

List introspection events.

Options

  • :run_id - Filter by run id
  • :session_key - Filter by session key
  • :agent_id - Filter by agent id
  • :event_type - Filter by event type
  • :since_ms - Include events at or after this timestamp
  • :until_ms - Include events at or before this timestamp
  • :limit - Maximum number of events to return (default: 100)

list_introspection_events(server, opts)

@spec list_introspection_events(
  server(),
  keyword()
) :: [map()]

List introspection events from a specific store instance.

list_runtime_policies(server \\ __MODULE__)

@spec list_runtime_policies(server()) :: [{term(), map()}]

List runtime policy entries.

list_session_policies(server \\ __MODULE__)

@spec list_session_policies(server()) :: [{term(), map()}]

List all session policies.

ping(server \\ __MODULE__)

@spec ping(server()) :: :ok | {:error, term()}

Performs a non-mutating store/backend liveness check.

put(server \\ __MODULE__, table, key, value)

@spec put(server(), table :: atom(), key :: term(), value :: term()) ::
  :ok | {:error, term()}

Put a value into a named table.

This is a generic API for use by other apps (e.g., lemon_core, lemon_automation).

put_agent_policy(server \\ __MODULE__, agent_id, policy)

@spec put_agent_policy(server(), agent_id :: term(), policy :: map()) :: :ok

Put an agent policy.

put_channel_policy(server \\ __MODULE__, channel_id, policy)

@spec put_channel_policy(server(), channel_id :: term(), policy :: map()) :: :ok

Put a channel policy.

put_chat_state(server \\ __MODULE__, scope, state)

@spec put_chat_state(server(), term(), LemonCore.ChatState.t() | map()) ::
  :ok | {:error, term()}

put_new(server \\ __MODULE__, table, key, value)

@spec put_new(server(), table :: atom(), key :: term(), value :: term()) ::
  :ok | {:error, term()}

Put a value into a named table only if the key does not already exist.

put_progress_mapping(server \\ __MODULE__, scope, progress_msg_id, run_id)

@spec put_progress_mapping(server(), term(), integer(), term()) ::
  :ok | {:error, term()}

put_runtime_policy(server \\ __MODULE__, policy)

@spec put_runtime_policy(server(), policy :: map()) :: :ok

Put global runtime policy overrides.

put_session_policy(server \\ __MODULE__, session_key, policy)

@spec put_session_policy(server(), session_key :: term(), policy :: map()) :: :ok

Put a session policy.

register_cached_table(server \\ __MODULE__, table)

@spec register_cached_table(atom(), atom()) :: :ok

Mirror table into this store's read cache from now on.

Registration is remembered in :persistent_term, so it survives a store restart and may be made before the store boots. Collaborators that own a generic table call this at startup rather than the store hard-coding it:

LemonCore.Store.register_cached_table(:my_hot_index)

register_finalize_run_hook(server \\ __MODULE__, hook)

@spec register_finalize_run_hook(atom(), LemonCore.Store.Hooks.hook()) :: :ok

Register a hook invoked after each run is finalized with a session key.

See finalize_run/3 for the payload and LemonCore.Store.Hooks for the accepted hook shapes. Registration survives a store restart.

start_link(opts \\ [])

unregister_cached_table(server \\ __MODULE__, table)

@spec unregister_cached_table(atom(), atom()) :: :ok

Stop mirroring table; takes effect on the store's next start.

unregister_finalize_run_hook(server \\ __MODULE__, hook)

@spec unregister_finalize_run_hook(atom(), LemonCore.Store.Hooks.hook()) :: :ok

Remove a previously registered finalize-run hook.