Current section

Files

Jump to
phoenix_sync lib phoenix sync sandbox stack.ex
Raw

lib/phoenix/sync/sandbox/stack.ex

if Phoenix.Sync.sandbox_enabled?() do
defmodule Phoenix.Sync.Sandbox.Stack do
@moduledoc false
use Supervisor, restart: :transient
alias Phoenix.Sync.Sandbox
def child_spec(opts) do
{:ok, stack_id} = Keyword.fetch(opts, :stack_id)
{:ok, repo} = Keyword.fetch(opts, :repo)
{:ok, owner} = Keyword.fetch(opts, :owner)
%{
id: {__MODULE__, stack_id},
start: {__MODULE__, :start_link, [stack_id, repo, owner]},
type: :supervisor,
restart: :transient
}
end
def name(stack_id) do
Phoenix.Sync.Sandbox.name({__MODULE__, stack_id})
end
def start_link(stack_id, repo, owner) do
Supervisor.start_link(__MODULE__, {stack_id, repo, owner}, name: name(stack_id))
end
alias Electric.Shapes.Querying
alias Electric.ShapeCache.Storage
def snapshot_query(
parent,
shape_handle,
shape,
db_pool,
storage,
stack_id,
chunk_bytes_threshold
) do
Postgrex.transaction(
db_pool,
fn conn ->
GenServer.cast(parent, {:pg_snapshot_known, shape_handle, {1000, 1100, []}})
# Enforce display settings *before* querying initial data to maintain consistent
# formatting between snapshot and live log entries.
Enum.each(Electric.Postgres.display_settings(), &Postgrex.query!(conn, &1, []))
stream =
Querying.stream_initial_data(conn, stack_id, shape, chunk_bytes_threshold)
|> Stream.transform(
fn -> false end,
fn item, acc ->
if not acc, do: GenServer.cast(parent, {:snapshot_started, shape_handle})
{[item], true}
end,
fn acc ->
if not acc, do: GenServer.cast(parent, {:snapshot_started, shape_handle})
acc
end
)
# could pass the shape and then make_new_snapshot! can pass it to row_to_snapshot_item
# that way it has the relation, but it is still missing the pk_cols
Storage.make_new_snapshot!(stream, storage)
end,
timeout: :infinity
)
end
def config(stack_id, repo, owner \\ nil) do
publication_manager_spec =
{Sandbox.PublicationManager, stack_id: stack_id, owner: owner, repo: repo}
# persistent_kv = Electric.PersistentKV.Memory.new!()
inspector = {Sandbox.Inspector, stack_id}
%{pid: pool} = Ecto.Adapter.lookup_meta(repo.get_dynamic_repo())
registry = :"#{__MODULE__}.Registry-#{stack_id}"
[
purge_all_shapes?: false,
stack_id: stack_id,
storage: {
Electric.ShapeCache.InMemoryStorage,
%{stack_id: stack_id, table_base_name: :"#{stack_id}"}
},
inspector: inspector,
publication_manager: publication_manager_spec,
chunk_bytes_threshold: 10_485_760,
db_pool: pool,
create_snapshot_fn: &snapshot_query/7,
log_producer: Electric.Replication.ShapeLogCollector.name(stack_id),
consumer_supervisor: Electric.Shapes.DynamicConsumerSupervisor.name(stack_id),
registry: registry,
max_shapes: nil
]
end
def init({stack_id, repo, owner}) do
# shape_cache_spec = {Electric.ShapeCache, shape_cache_opts}
config = config(stack_id, repo, owner)
shape_cache_spec = {Electric.ShapeCache, config}
persistent_kv = Electric.PersistentKV.Memory.new!()
shape_status_spec =
{Electric.ShapeCache.ShapeStatus,
%Electric.ShapeCache.ShapeStatus{
shape_meta_table: Electric.ShapeCache.get_shape_meta_table(stack_id: stack_id)
}}
children = [
{Registry, keys: :duplicate, name: config[:registry]},
{Electric.ProcessRegistry, stack_id: stack_id},
{Electric.StatusMonitor, stack_id},
{Electric.Shapes.Monitor,
stack_id: stack_id,
storage: config[:storage],
shape_status: shape_status_spec,
publication_manager: config[:publication_manager]},
# TODO: start an electric stack, decoupled from the db connection
# with in memory storage, a mock publication_manager and inspector
Supervisor.child_spec(
{
Electric.Replication.Supervisor,
stack_id: stack_id,
shape_cache: shape_cache_spec,
publication_manager: config[:publication_manager],
log_collector: {
Electric.Replication.ShapeLogCollector,
stack_id: stack_id, inspector: config[:inspector], persistent_kv: persistent_kv
},
schema_reconciler: {Phoenix.Sync.Sandbox.SchemaReconciler, stack_id},
stack_events_registry: config[:registry]
},
restart: :temporary
),
{Sandbox.Inspector, stack_id: stack_id, repo: repo},
{Sandbox.Producer, stack_id: stack_id},
{DynamicSupervisor,
name: Phoenix.Sync.Sandbox.Fetch.name(stack_id), strategy: :one_for_one}
]
Supervisor.init(children, strategy: :one_for_one)
end
end
end