Packages
electric
0.7.4
1.7.8
1.7.7
1.7.6
1.7.5
1.7.4
1.7.3
1.7.2
1.7.1
1.7.0
1.6.10
1.6.9
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.16
1.4.16-beta-1
1.4.15
1.4.14
1.4.13
1.4.12
1.4.11
1.4.10
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.4
1.3.3
1.3.2
1.2.4
1.2.3
1.2.2
1.2.1
1.2.0
1.1.14
1.1.13
1.1.12
1.1.11
1.1.10
1.1.9
1.1.8
1.1.7
1.1.6
retired
1.1.5
retired
1.1.4
retired
1.1.3
retired
1.1.2
1.1.1
1.1.0
1.0.24
1.0.23
1.0.22
1.0.21
1.0.20
1.0.19
1.0.18
1.0.17
1.0.15
1.0.13
1.0.12
1.0.11
1.0.10
1.0.9
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
1.0.0-beta.23
1.0.0-beta.22
1.0.0-beta.20
1.0.0-beta.19
1.0.0-beta.18
1.0.0-beta.17
1.0.0-beta.16
1.0.0-beta.15
1.0.0-beta.14
1.0.0-beta.13
1.0.0-beta.12
1.0.0-beta.11
1.0.0-beta.10
1.0.0-beta.9
1.0.0-beta.8
1.0.0-beta.7
1.0.0-beta.6
1.0.0-beta.5
1.0.0-beta.4
1.0.0-beta.3
1.0.0-beta.2
1.0.0-beta.1
0.9.5
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.1
0.8.0
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.3
0.6.2
0.6.1
0.5.2
0.4.4
Postgres sync engine. Sync little subsets of your Postgres data into local apps and services.
Current section
Files
Jump to
Current section
Files
lib/electric/log_items.ex
defmodule Electric.LogItems do
alias Electric.Replication.Changes
alias Electric.Replication.LogOffset
alias Electric.Shapes.Shape
@moduledoc """
Defines the structure and how to create the items in the log that the electric client reads.
The log_item() data structure is a map for ease of consumption in the Elixir code,
however when JSON encoded (not done in this module) it's the format that the electric
client accepts.
"""
@type log_item :: %{
key: String.t(),
value: map(),
headers: map(),
offset: LogOffset.t()
}
@spec from_change(
Changes.data_change(),
txid :: non_neg_integer() | nil,
pk_cols :: [String.t()]
) :: [log_item(), ...]
def from_change(%Changes.NewRecord{} = change, txid, _) do
[
%{
key: change.key,
value: change.record,
headers: %{operation: :insert, txid: txid, relation: Tuple.to_list(change.relation)},
offset: change.log_offset
}
]
end
def from_change(%Changes.DeletedRecord{} = change, txid, pk_cols) do
[
%{
key: change.key,
value: take_pks_or_all(change.old_record, pk_cols),
headers: %{operation: :delete, txid: txid, relation: Tuple.to_list(change.relation)},
offset: change.log_offset
}
]
end
# `old_key` is nil when it's unchanged. This is not possible when there is no PK defined.
def from_change(%Changes.UpdatedRecord{old_key: nil} = change, txid, pk_cols) do
[
%{
key: change.key,
value: Map.take(change.record, Enum.concat(pk_cols, change.changed_columns)),
headers: %{operation: :update, txid: txid, relation: Tuple.to_list(change.relation)},
offset: change.log_offset
}
]
end
def from_change(%Changes.UpdatedRecord{} = change, txid, pk_cols) do
[
%{
key: change.old_key,
value: take_pks_or_all(change.old_record, pk_cols),
headers: %{
operation: :delete,
txid: txid,
relation: Tuple.to_list(change.relation),
key_change_to: change.key
},
offset: change.log_offset
},
%{
key: change.key,
value: change.record,
headers: %{
operation: :insert,
txid: txid,
relation: Tuple.to_list(change.relation),
key_change_from: change.old_key
},
offset: LogOffset.increment(change.log_offset)
}
]
end
defp take_pks_or_all(record, []), do: record
defp take_pks_or_all(record, pks), do: Map.take(record, pks)
@spec from_snapshot_row_stream(
row_stream :: Enumerable.t(list()),
offset :: LogOffset.t(),
shape :: Shape.t(),
query_info :: Postgrex.Query.t()
) :: Enumerable.t(log_item())
def from_snapshot_row_stream(row_stream, offset, shape, query_info) do
Stream.map(row_stream, &from_snapshot_row(&1, offset, shape, query_info))
end
defp from_snapshot_row(row, offset, shape, query_info) do
value = value(row, query_info)
key = Changes.build_key(shape.root_table, value, Shape.pk(shape))
%{
key: key,
value: value,
headers: %{operation: :insert, relation: shape.root_table |> Tuple.to_list()},
offset: offset
}
end
# We're assuming the the postgres query casts every column to string
defp value(row, %Postgrex.Query{columns: columns}) do
[columns, row]
|> Enum.zip()
|> Map.new()
end
end