Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update #51

Merged
merged 11 commits into from
Mar 13, 2019
Prev Previous commit
Next Next commit
merges with insert
Danwhy committed Feb 28, 2019
commit 19f05296d262d3b666d10972b2d58ce7a58d14cf
85 changes: 84 additions & 1 deletion lib/alog.ex
Original file line number Diff line number Diff line change
@@ -14,7 +14,16 @@ defmodule Alog do
def supports_ddl_transaction?, do: true

@impl true
defdelegate storage_down(opts), to: Ecto.Adapters.Postgres
defdelegate storage_up(opts), to: EAP

@impl true
defdelegate storage_down(opts), to: EAP

@impl true
defdelegate structure_dump(default, config), to: EAP

@impl true
defdelegate structure_load(default, config), to: EAP

@impl true
def update(adapter_meta, %{source: source, prefix: prefix}, fields, params, returning, opts) do
@@ -59,4 +68,78 @@ defmodule Alog do
@impl true
def dumpers(:binary_id, type), do: [:binary, type]
def dumpers(_primitive, type), do: [type]

# overrides insert/6 defined in Ecto.Adapters.SQL
@impl true
def insert(
adapter_meta,
%{source: source, prefix: prefix},
params,
on_conflict,
returning,
opts
) do
# converts params from a keyword list to a map
params_map = Enum.into(params, %{})

# removes inserted_at and updated_at from map (will not error if keys are not in map)
map_for_cid = Map.drop(params_map, [:inserted_at, :updated_at])

# creates a cid from the map witout the inserted_at and updated_at_values
cid = Cid.cid(map_for_cid)

# creates a unique entry_id for the data based on the CID generated
entry_id = create_entry_id(source, adapter_meta, cid, 2)

# updates params to ensure that timestamps, cid, and entry_id are all added.
# then converts the map back into a list for use in existing functionality (original format)
params =
map_for_cid
|> add_timestamps()
# <==== Should this be Map.put(:id, cid)??????????
|> Map.put(:cid, cid)
|> Map.put(:entry_id, entry_id)
|> Enum.into([])

{kind, conflict_params, _} = on_conflict
{fields, values} = :lists.unzip(params)
sql = @conn.insert(prefix, source, fields, [fields], on_conflict, returning)

Ecto.Adapters.SQL.struct(
adapter_meta,
@conn,
sql,
:insert,
source,
[],
values ++ conflict_params,
kind,
returning,
opts
)
end

# I think that this step need to also make sure that the data is not an exact copy.
# if the full cid already exists then this is duplicate data.
# Should we insert duplicate data.
# i was thinking maybe if it was existing data but not the most recent data we should re-insert the data
# e.g. if the comment was hi, edited to hey, and then changed back to hi.
defp create_entry_id(source, adapter_meta, cid, n) do
entry_id = String.slice(cid, 0..n)
entry_id_query = "SELECT * FROM #{source} where entry_id='#{entry_id}'"
{:ok, results} = Ecto.Adapters.SQL.query(adapter_meta, entry_id_query, [])

if results.num_rows == 0 do
entry_id
else
create_entry_id(source, adapter_meta, cid, n + 1)
end
end

defp add_timestamps(params) do
params
|> Enum.into(%{})
|> Map.put_new(:inserted_at, NaiveDateTime.utc_now())
|> Map.put_new(:updated_at, NaiveDateTime.utc_now())
end
end
49 changes: 34 additions & 15 deletions lib/alog/connection.ex
Original file line number Diff line number Diff line change
@@ -5,21 +5,40 @@ defmodule Alog.Connection do
@default_port 5432

@impl true
defdelegate child_spec(opts), to: Ecto.Adapters.Postgres.Connection
def child_spec(opts) do
opts
|> Keyword.put_new(:port, @default_port)
|> Postgrex.child_spec()
end

@impl true
defdelegate prepare_execute(conn, name, statement, params, opts), to: EAPC

@impl true
defdelegate execute(conn, query, params, opts), to: EAPC

@impl true
defdelegate query(conn, statement, params, opts), to: EAPC

@impl true
defdelegate stream(conn, statement, params, opts), to: EAPC

@impl true
defdelegate to_constraints(error_struct), to: EAPC

@impl true
defdelegate ddl_logs(result), to: Ecto.Adapters.Postgres.Connection
defdelegate ddl_logs(result), to: EAPC

@impl true
defdelegate prepare_execute(connection, name, statement, params, options),
to: Ecto.Adapters.Postgres.Connection
to: EAPC

@impl true
defdelegate query(connection, statement, params, options), to: Ecto.Adapters.Postgres.Connection
defdelegate query(connection, statement, params, options), to: EAPC

@impl true
defdelegate stream(connection, statement, params, options),
to: Ecto.Adapters.Postgres.Connection
to: EAPC

@impl true
def execute_ddl({c, %Ecto.Migration.Table{} = table, columns} = command)
@@ -38,10 +57,10 @@ defmodule Alog.Connection do
raise ArgumentError, "you cannot add a primary key"
else
:schema_migrations ->
Ecto.Adapters.Postgres.Connection.execute_ddl({c, table, columns})
EAPC.execute_ddl({c, table, columns})

_ ->
Ecto.Adapters.Postgres.Connection.execute_ddl({c, table, update_columns(columns)})
EAPC.execute_ddl({c, table, update_columns(columns)})
end
end

@@ -60,7 +79,7 @@ defmodule Alog.Connection do
nil
end
) do
Ecto.Adapters.Postgres.Connection.execute_ddl(command)
EAPC.execute_ddl(command)
end
end

@@ -69,7 +88,7 @@ defmodule Alog.Connection do
raise ArgumentError, "you cannot create a unique index"
end

defdelegate execute_ddl(command), to: Ecto.Adapters.Postgres.Connection
defdelegate execute_ddl(command), to: EAPC

# Add required columns if they are missing
defp update_columns(columns) do
@@ -91,20 +110,20 @@ defmodule Alog.Connection do
# Temporary delegate functions to make tests work

@impl true
defdelegate all(a), to: Ecto.Adapters.Postgres.Connection
defdelegate all(query), to: EAPC

@impl true
defdelegate insert(a, b, c, d, e, f), to: Ecto.Adapters.Postgres.Connection
defdelegate insert(prefix, table, header, rows, on_conflict, returning), to: EAPC

@impl true
defdelegate execute(a, b, c, d), to: Ecto.Adapters.Postgres.Connection
defdelegate delete_all(query), to: EAPC

@impl true
defdelegate delete_all(a), to: Ecto.Adapters.Postgres.Connection
defdelegate update(prefix, table, fields, filters, returning), to: EAPC

@impl true
defdelegate to_constraints(a), to: Ecto.Adapters.Postgres.Connection
defdelegate delete(prefix, table, filters, returning), to: EAPC

@impl true
defdelegate update(a, b, c, d, e), to: Ecto.Adapters.Postgres.Connection
defdelegate update_all(query, prefix \\ nil), to: EAPC
end
You are viewing a condensed version of this merge commit. You can view the full changes here.