Adapters

Storage

Wiregrid.Storage is an append-oriented stream interface. The behaviour requires bootstrap/1, append/6, get/3, page/4, delete/3, prune/4, and health/1. Streams are validated topics. Ids are binaries bounded by max_storage_id_bytes (256). A page is ordered by (inserted_at_ms, id) so two rows with the same millisecond are neither skipped nor repeated. The cursor is an opaque binary from Wiregrid.Storage.Cursor.

The application chooses what is persisted. publish(..., persist: true) appends after validation and authorization and before local fanout. Wiregrid.Chat.say/4 sets that flag. Wiregrid.publish/4 leaves it to the caller.

:ok = Wiregrid.storage_bootstrap(:chat)
:ok = Wiregrid.append_event(:chat, {:channel, "general"}, id, event, %{})
{:ok, row} = Wiregrid.get_event(:chat, {:channel, "general"}, id)
{:ok, rows, cursor} = Wiregrid.page_events(:chat, {:channel, "general"}, nil, 100)
{:ok, rows, cursor} = Wiregrid.read_history(:chat, session_id, {:channel, "general"}, nil, 50)
:ok = Wiregrid.delete_event(:chat, {:channel, "general"}, id)
{:ok, deleted} = Wiregrid.prune_events(:chat, {:channel, "general"}, before_ms, 1_000)

read_history/5 checks the :read authorizer action for that session, then pages. page_events/4 is the in-process call for code that has already decided the caller may read. Default limits are 100 and 50 respectively. prune_events/4 defaults the delete cap to 1,000.

Adapter config is {module, keyword} or a module atom, which means {module, []}. The module must export the behaviour callbacks or startup fails with {:invalid_module, :storage, module}. Connections and pools belong to the host application. Wiregrid does not open them and does not store credentials.

Memory

The default is {Wiregrid.Storage.Memory, []}. Rows live in a bounded ETS table, idempotent on (stream, id). The cap is max_memory_events: 250,000 on :small, 1,000,000 on :balanced, 5,000,000 on :large. Memory storage is the right default for tests and for data you can rebuild. It is empty after the table owner restarts.

Postgres

storage: {Wiregrid.Storage.Postgres, conn: MyApp.Repo}

Pass a Postgrex-compatible connection or pool as conn:. bootstrap/1 runs the idempotent statements from priv/migrations/postgres/001_events.sql inside one transaction, under a transaction-scoped advisory lock. The table is wiregrid_events (stream, id, event, meta, inserted_at_ms) with primary key (stream, inserted_at_ms, id) and a unique (stream, id). Event and meta columns are bytea produced by Wiregrid.SafeTerm, capped at 16,777,216 bytes inside the adapter.

DATABASE_URL=... ./scripts/db-init.sh postgres

Integration tests accept an Ecto URL such as ecto://wiregrid:wiregrid@127.0.0.1:5432/wiregrid and pass Postgrex hostname, port, username, password, and database. Pool size, TLS, failover, backups, and retention policy stay with the operator. prune/4 deletes a bounded number of rows older than a timestamp.

Redis is a cache

cache: {Wiregrid.Cache.Redis, conn: MyApp.Redix, prefix: "myapp:wg:"}
:ok = Wiregrid.cache_put(:chat, "roster", term, 60_000)
{:ok, term} = Wiregrid.cache_get(:chat, "roster")
:ok = Wiregrid.cache_delete(:chat, "roster")
{:ok, n} = Wiregrid.cache_incr(:chat, "views", 1, 60_000)

Redis does not store chat history. The cache behaviour is get/2, put/4, delete/2, incr/4, and health/1. The default in-process cache is {Wiregrid.Cache.Memory, []}, capped by max_cache_entries.

The Redis prefix defaults to "wg1:". It must be 1 to 128 bytes and must not contain CR, LF, or NUL. Wiregrid appends i: plus a 22-character URL-safe SHA-256 of the instance name, then v: for values and c: for counters. A put deletes the counter key in the same Lua script. An incr refuses a key that already holds a normal value. TTL is milliseconds; :infinity omits expiry. Values are SafeTerm blobs, decoded with a 16,777,216 byte cap. The public cache API does not forward arbitrary Redis commands. Key and value ceilings on the instance are max_cache_key_bytes (1,024) and max_cache_value_bytes (1,048,576).

Scylla

storage: {Wiregrid.Storage.Scylla,
  conn: MyApp.Xandra,
  keyspace: "wiregrid",
  replication_factor: 3,
  bucket_ms: 86_400_000
}

Long streams are split into time buckets so one partition does not grow for the life of the channel. The default bucket is one day (86,400,000 ms). bucket_ms must be a positive integer no larger than 31 days. Changing it for existing data is a migration.

bootstrap/1 runs three statements. The keyspace is created with NetworkTopologyStrategy and this tablet clause, which keeps a single-node or mixed-version cluster on the classic storage path:

CREATE KEYSPACE IF NOT EXISTS wiregrid
  WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor': 3}
  AND tablets = {'enabled': false}

Replication factor defaults to 3 and must be an integer from 1 to 16. Use 1 only for a disposable one-node dev cluster. The keyspace name is validated as an identifier of 1 to 48 characters because CQL cannot bind identifiers. The default name is wiregrid.

Three tables follow:

wiregrid_events — primary key ((stream, bucket), inserted_at_ms, id), clustering ascending.
wiregrid_event_lookup — primary key ((stream, id)), holding bucket, timestamp, event, and meta.
wiregrid_stream_buckets — primary key ((stream), bucket), the directory page/4 walks.

The first write of an id is an LWT: INSERT INTO wiregrid_event_lookup (...) VALUES (...) IF NOT EXISTS. That row is the canonical copy. append/6 then reads the lookup back and repairs the time-bucket row and the bucket directory from it, so a retry cannot invent a second timestamp. get/3 uses the lookup. If the lookup exists and the bucket row does not, the read repairs the secondary row and still returns the canonical event.

Xandra returns a Xandra.Page struct whose :content field is raw column values. Rows become string-key maps only when the page is enumerated. The adapter enumerates struct pages that carry :columns and :content, and it still accepts a plain %{content: rows} map used by tests. Queries are prepared. Bucket directory pages are 64 buckets wide. Event and meta blobs are capped at 16,777,216 bytes.

./scripts/db-init.sh scylla

SCYLLA_KEYSPACE and SCYLLA_REPLICATION_FACTOR can render a validated copy without editing the migration. The integration test is tagged :scylla_integration only, and the default test helper excludes that tag. Point WIREGRID_SCYLLA_NODE at host:port (CI uses 127.0.0.1:9042) and run mix test --include scylla_integration test/integration/scylla_test.exs with the compose file compose.test.yml.

Isolation

Wiregrid.start_instance(:chat,
  adapter_mode: :isolated,
  adapter_timeout_ms: 2_000,
  max_adapter_pending: 512,
  storage: {MyApp.Storage, []}
)

The default adapter_mode is :inline, with adapter_timeout_ms 5,000. Isolated mode runs adapter calls as supervised tasks. When max_adapter_pending is full the call returns {:error, :adapter_overloaded}. Timeouts and crashes are counted. Profile defaults for that pending cap are 128, 1,024, and 4,096.