Syntropy.Persistence (syntropy v0.3.0)

Copy Markdown

Optional Postgres-backed durability for Syntropy replay-critical read models.

Summary

Types

cursor()
event()
page()
recommendation()
run_envelope()
snapshot()
task_result()
webhook()
write_result()

Functions

clear_all!()
counter_from_ids(records, key, prefix)
deactivate_api_token(token_id)
decode_cursor(value)

Decodes an opaque cursor back into {timestamp, id}.

delete_knowledge_items(agent_id)
delete_webhook(webhook_id)
enabled?()
encode_cursor(arg)

Encodes a {timestamp, id} pagination cursor as an opaque URL-safe string.

find_active_api_token(token_hash)
get_recommendation(recommendation_id)
get_run_envelope(run_id)
get_setting(key)
get_snapshot(snapshot_id)
hydrate_limit()
insert_api_token(attrs)
list_api_tokens()
list_recommendations(opts \\ [])

Lists structural recommendations newest-first with cursor pagination.

list_run_envelopes(opts \\ [])

Lists run envelopes newest-first with cursor pagination.

list_snapshots(opts \\ [])

Lists lattice snapshots newest-first with cursor pagination.

load_knowledge_items()
load_recent_events(limit)
load_recent_recommendations(limit)
load_recent_run_envelopes(limit)
load_recent_snapshots(limit)
load_recent_tasks(limit)
load_webhooks()
observe_write(error, operation)
persist_event(event)
persist_events(events)
persist_knowledge_items(agent_id, items)
persist_recommendation(recommendation)
persist_run_envelope(run_envelope)
persist_snapshot(snapshot)
persist_task_run(task_result)
persist_usage_record(attrs)
purge_replay_data(cutoff)

Deletes replay data older than cutoff (or all replay data with :all).

put_setting(key, value)
recent_usage_records(limit)
require_write!(result, operation)
tolerate_write(result, operation)
upsert_webhook(webhook)
usage_by_model_since(cutoff)
usage_daily_since(cutoff)
usage_totals_since(cutoff)

Types

cursor()

@type cursor() :: {DateTime.t(), String.t()}

event()

@type event() :: Syntropy.EventRecorder.event()

page()

@type page() :: %{
  entries: [map()],
  has_more: boolean(),
  next_cursor: String.t() | nil
}

recommendation()

@type recommendation() :: Syntropy.RecommendationStore.recommendation()

run_envelope()

@type run_envelope() :: map()

snapshot()

@type snapshot() :: Syntropy.HistoryStore.snapshot()

task_result()

@type task_result() :: Syntropy.TaskScheduler.task_result()

webhook()

@type webhook() :: Syntropy.Webhooks.Registry.webhook()

write_result()

@type write_result() :: :ok | {:error, term()}

Functions

clear_all!()

@spec clear_all!() :: :ok

counter_from_ids(records, key, prefix)

@spec counter_from_ids([map()], atom(), String.t()) :: non_neg_integer()

deactivate_api_token(token_id)

@spec deactivate_api_token(String.t()) ::
  :ok | {:error, :not_found} | {:error, term()}

decode_cursor(value)

@spec decode_cursor(String.t()) :: {:ok, cursor()} | :error

Decodes an opaque cursor back into {timestamp, id}.

delete_knowledge_items(agent_id)

@spec delete_knowledge_items(String.t()) :: :ok | {:error, term()}

delete_webhook(webhook_id)

@spec delete_webhook(String.t()) :: :ok | {:error, term()}

enabled?()

@spec enabled?() :: boolean()

encode_cursor(arg)

@spec encode_cursor({DateTime.t(), term()}) :: String.t()

Encodes a {timestamp, id} pagination cursor as an opaque URL-safe string.

find_active_api_token(token_hash)

@spec find_active_api_token(String.t()) :: Syntropy.Persistence.ApiToken.t() | nil

get_recommendation(recommendation_id)

@spec get_recommendation(String.t()) :: recommendation() | nil

get_run_envelope(run_id)

@spec get_run_envelope(String.t()) :: run_envelope() | nil

get_setting(key)

@spec get_setting(String.t()) :: map() | nil

get_snapshot(snapshot_id)

@spec get_snapshot(String.t()) :: snapshot() | nil

hydrate_limit()

@spec hydrate_limit() :: pos_integer()

insert_api_token(attrs)

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

list_api_tokens()

@spec list_api_tokens() :: [Syntropy.Persistence.ApiToken.t()]

list_recommendations(opts \\ [])

@spec list_recommendations(keyword()) :: page()

Lists structural recommendations newest-first with cursor pagination.

Options: :limit, :cursor, :status, :kind, :q (matches the recommendation id or candidate agent ids).

list_run_envelopes(opts \\ [])

@spec list_run_envelopes(keyword()) :: page()

Lists run envelopes newest-first with cursor pagination.

Options: :limit, :cursor ({timestamp, id}), :status, :kind, :q (matches the run id or prompt).

list_snapshots(opts \\ [])

@spec list_snapshots(keyword()) :: page()

Lists lattice snapshots newest-first with cursor pagination.

Options: :limit, :cursor, :kind (trigger kind), :q (matches the snapshot or task id).

load_knowledge_items()

@spec load_knowledge_items() :: %{required(String.t()) => [String.t()]}

load_recent_events(limit)

@spec load_recent_events(non_neg_integer()) :: [event()]

load_recent_recommendations(limit)

@spec load_recent_recommendations(non_neg_integer()) :: [recommendation()]

load_recent_run_envelopes(limit)

@spec load_recent_run_envelopes(non_neg_integer()) :: [run_envelope()]

load_recent_snapshots(limit)

@spec load_recent_snapshots(non_neg_integer()) :: [snapshot()]

load_recent_tasks(limit)

@spec load_recent_tasks(non_neg_integer()) :: [task_result()]

load_webhooks()

@spec load_webhooks() :: [webhook()]

observe_write(error, operation)

@spec observe_write(write_result(), String.t()) :: write_result()

persist_event(event)

@spec persist_event(event()) :: :ok | {:error, term()}

persist_events(events)

@spec persist_events([event()]) :: :ok | {:error, term()}

persist_knowledge_items(agent_id, items)

@spec persist_knowledge_items(String.t(), [String.t()]) :: :ok | {:error, term()}

persist_recommendation(recommendation)

@spec persist_recommendation(recommendation()) :: :ok | {:error, term()}

persist_run_envelope(run_envelope)

@spec persist_run_envelope(run_envelope()) :: :ok | {:error, term()}

persist_snapshot(snapshot)

@spec persist_snapshot(snapshot()) :: :ok | {:error, term()}

persist_task_run(task_result)

@spec persist_task_run(task_result()) :: :ok | {:error, term()}

persist_usage_record(attrs)

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

purge_replay_data(cutoff)

@spec purge_replay_data(DateTime.t() | :all) ::
  {:ok, %{required(String.t()) => non_neg_integer()}} | {:error, term()}

Deletes replay data older than cutoff (or all replay data with :all).

Returns per-table deletion counts. Failures are observed through PersistenceHealth and returned; callers decide whether to tolerate them.

put_setting(key, value)

@spec put_setting(String.t(), map()) :: :ok | {:error, term()}

recent_usage_records(limit)

@spec recent_usage_records(non_neg_integer()) :: [
  Syntropy.Persistence.UsageRecord.t()
]

require_write!(result, operation)

@spec require_write!(write_result(), String.t()) :: :ok

tolerate_write(result, operation)

@spec tolerate_write(write_result(), String.t()) :: :ok

upsert_webhook(webhook)

@spec upsert_webhook(webhook()) :: :ok | {:error, term()}

usage_by_model_since(cutoff)

@spec usage_by_model_since(DateTime.t()) :: [map()]

usage_daily_since(cutoff)

@spec usage_daily_since(DateTime.t()) :: [map()]

usage_totals_since(cutoff)

@spec usage_totals_since(DateTime.t()) :: map() | nil