Packages
ash_authentication
5.0.0-rc.1
5.0.0-rc.12
5.0.0-rc.11
5.0.0-rc.10
5.0.0-rc.9
5.0.0-rc.8
5.0.0-rc.7
5.0.0-rc.6
5.0.0-rc.5
5.0.0-rc.4
5.0.0-rc.3
5.0.0-rc.2
5.0.0-rc.1
5.0.0-rc.0
4.14.1
4.14.0
4.13.7
4.13.6
4.13.5
4.13.4
4.13.3
4.13.2
retired
4.13.1
retired
4.13.0
4.12.0
4.11.0
4.10.0
4.9.9
4.9.8
4.9.7
4.9.6
4.9.5
4.9.4
4.9.3
4.9.2
4.9.1
4.9.0
4.8.7
4.8.6
4.8.5
4.8.3
4.8.2
4.8.1
4.8.0
4.7.6
4.7.5
4.7.4
4.7.3
4.7.2
4.7.1
4.7.0
4.6.4
4.6.3
4.6.2
4.6.1
4.6.0
4.5.6
4.5.5
4.5.4
4.5.3
4.5.2
4.5.1
4.5.0
4.4.9
4.4.8
4.4.7
4.4.6
4.4.5
4.4.4
4.4.3
4.4.2
4.4.1
4.4.0
4.3.12
4.3.11
4.3.10
4.3.9
4.3.8
4.3.7
4.3.6
4.3.5
4.3.4
4.3.3
4.3.2
4.3.1
4.3.0
4.2.7
4.2.6
4.2.5
4.2.4
4.2.3
4.2.2
4.2.1
4.2.0
4.1.0
4.0.4
4.0.3
4.0.2
4.0.1
4.0.0
4.0.0-rc.6
4.0.0-rc.5
4.0.0-rc.3
4.0.0-rc.2
4.0.0-rc.1
4.0.0-rc.0
3.12.4
3.12.3
3.12.2
3.12.1
3.12.0
3.11.16
3.11.15
3.11.14
3.11.13
3.11.12
3.11.11
3.11.10
3.11.9
3.11.8
3.11.7
3.11.6
3.11.5
3.11.4
3.11.3
3.11.2
3.11.1
3.11.0
3.10.8
3.10.7
3.10.6
3.10.5
3.10.4
3.10.3
3.10.2
3.10.1
3.10.0
3.9.6
3.9.5
3.9.4
3.9.3
3.9.2
3.9.1
3.9.0
3.8.0
3.7.9
3.7.8
3.7.6
3.7.5
3.7.4
3.7.3
3.7.2
3.7.1
3.7.0
3.6.1
3.6.0
3.5.3
3.5.2
3.5.1
3.5.0
3.3.1
3.3.0
3.2.2
3.2.1
3.2.0
3.1.0
3.0.3
Authentication extension for the Ash Framework.
Security advisory:
This version has known vulnerabilities.
View advisories
Current section
Files
Jump to
Current section
Files
lib/ash_authentication/audit_log_resource/batcher.ex
# SPDX-FileCopyrightText: 2022 Alembic Pty Ltd
#
# SPDX-License-Identifier: MIT
defmodule AshAuthentication.AuditLogResource.Batcher do
@moduledoc """
A `GenServer` which batches up writes to the audit log to reduce write pressure in busy environments.
Scans all audit log resources based on their configured write batching options.
```elixir
defmodule MyApp.Accounts.AuditLog do
use Ash.Resource,
extensions: [AshAuthentication.AuditLogResource],
domain: MyApp.Accounts
audit_log do
write_batching do
enabled? true
timeout :timer.seconds(10)
max_size 100
end
end
end
```
This GenServer is started by the `AshAuthentication.Supervisor` which should be added to your app's supervision tree.
"""
use GenServer
alias AshAuthentication.AuditLogResource
alias AshAuthentication.AuditLogResource.Info
require Logger
@doc """
Queues an event for writing.
"""
def enqueue(changeset), do: GenServer.cast(__MODULE__, {:enqueue, changeset})
@doc """
Flushes all queued events to the database immediately.
Useful for testing to ensure all audit log entries are written before assertions.
"""
def flush, do: GenServer.call(__MODULE__, :flush)
@doc false
@spec start_link(any) :: GenServer.on_start()
def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)
@doc false
@impl true
@spec init(keyword) :: {:ok, map}
def init(opts) do
otp_app = Keyword.fetch!(opts, :otp_app)
audit_configs =
otp_app
|> Spark.sparks(Ash.Resource)
|> Stream.filter(&(AuditLogResource in Spark.extensions(&1)))
|> Enum.reduce(%{}, fn resource, configs ->
Map.put(configs, resource, %{
timeout: Info.audit_log_write_batching_timeout!(resource),
max_size: Info.audit_log_write_batching_max_size!(resource),
action: Info.audit_log_write_action_name!(resource),
domain: Info.audit_log_domain!(resource),
queue: [],
queue_size: 0,
flush_timer: nil
})
end)
if map_size(audit_configs) == 0 do
:ignore
else
{:ok, audit_configs}
end
end
@doc false
@impl true
def handle_info({:write_batch, resource}, state) do
state =
state
|> Map.replace_lazy(resource, &write_batch(resource, %{&1 | flush_timer: nil}))
{:noreply, state}
end
@doc false
@impl true
def handle_cast({:enqueue, changeset}, state) do
state =
state
|> enqueue(changeset)
|> maybe_write_batches()
{:noreply, state}
end
@doc false
@impl true
def handle_call(:flush, _from, state) do
state =
state
|> Enum.reduce(%{}, fn {resource, config}, configs ->
config =
if config.queue_size > 0 do
if is_reference(config.flush_timer) do
Process.cancel_timer(config.flush_timer)
end
write_batch(resource, %{config | flush_timer: nil})
else
config
end
Map.put(configs, resource, config)
end)
{:reply, :ok, state}
end
@doc false
@impl true
def terminate(_reason, state) do
state
|> Enum.filter(fn {_, %{queue_size: qs}} -> qs > 0 end)
|> Enum.each(fn {resource, config} ->
write_batch(resource, config)
end)
end
defp enqueue(state, changeset) do
Map.update!(state, changeset.resource, fn
config when config.queue_size == 0 and config.queue == [] and config.timeout >= 0 ->
timer = Process.send_after(self(), {:write_batch, changeset.resource}, config.timeout)
%{
config
| queue: [changeset],
queue_size: 1,
flush_timer: timer
}
config ->
%{config | queue: [changeset | config.queue], queue_size: config.queue_size + 1}
end)
end
defp maybe_write_batches(state) do
state
|> Enum.reduce(%{}, fn
{resource, config}, configs when config.queue_size >= config.max_size ->
config =
if is_reference(config.flush_timer) do
Process.cancel_timer(config.flush_timer)
write_batch(resource, %{config | flush_timer: nil})
else
write_batch(resource, config)
end
Map.put(configs, resource, config)
{resource, config}, configs ->
Map.put(configs, resource, config)
end)
end
defp write_batch(resource, config) do
bulk_result =
config.queue
|> Stream.map(& &1.attributes)
|> Ash.bulk_create(resource, config.action,
domain: config.domain,
return_errors?: true,
assume_casted?: true,
context: %{private: %{ash_authentication?: true}}
)
if Enum.any?(bulk_result.errors) do
Logger.error(fn ->
"""
Errors occurred while writing audit logs to `#{inspect(resource)}`:
#{inspect(bulk_result.errors)}
"""
end)
end
%{config | queue: [], queue_size: 0}
end
end