exsql

defmodule ExSQL.Log do
  @moduledoc """
  A logical redo log with an async writer, the BEAM-native answer to SQLite's
  WAL + checkpoint.

  Each committed transaction's effects are appended to a per-database log file;
  on open the base SQLite file is read and the log replayed on top; periodically
  the log is folded back into a fresh base file (a checkpoint). Write cost per
  commit is O(change), not O(total DB size) — unlike the whole-file rewrite of
  `:file` mode.

  One writer process per base path (registered in `ExSQL.LogRegistry`, started
  under `ExSQL.LogSupervisor`) owns the log file. A connection `cast`s redo
  records to it and returns immediately; the writer batches appends and fsyncs
  on a timer / on `flush/1` / checkpoint / shutdown.

  ## Record format

  One record per committed transaction: `term_to_binary([{sql, params}, …])`,
  framed as `<<size::32, crc32::32, payload::binary>>`. A torn or
  crc-mismatched tail stops replay, so an interrupted append is dropped whole.

  Connections append whole transactions (the driver buffers statements between
  BEGIN and COMMIT), so records from a pool of connections never interleave
  mid-transaction and replay stays in commit order. Each processed append bumps
  a per-path commit version in `ExSQL.Registry` (under `version_key/1`), which
  lets other connections on the same path detect that their snapshot is stale
  and reload via `fetch/1`.

  ## Crash-safe checkpoint

  Folding the log into the base must survive a crash at any step. Done in the
  serialized writer (no appends interleave):

    1. fsync the active log
    2. `db = read(base) + replay(log)`
    3. write `base.new` (temp); fsync
    4. rename `log` → `log.archived`; open a fresh empty `log`
    5. commit point: rename `base.new` → `base`
    6. delete `log.archived`

  Recovery decides from which files exist (see `recover/1`): `base.new` present
  ⇒ checkpoint didn't commit (discard it); `base.new` absent + `log.archived`
  present ⇒ it committed (base already folded — drop the archive). Never
  double-applies.

  ## Determinism (v1, statement log)

  Replay re-runs the logged SQL, so SQL-level volatile functions (`random()`,
  `datetime('now')`) replay differently. rowid/AUTOINCREMENT is deterministic
  under ordered replay. The Ecto path passes resolved values as params, so this
  does not affect it.
  """

  use GenServer

  alias ExSQL.{Database, Executor, FileFormat}

  @flush_ms 50

  # -- client -------------------------------------------------------------------

  @doc "Ensures the writer for `base_path` is running and returns the recovered database."
  @spec open(Path.t(), keyword()) :: {:ok, Database.t()} | {:error, term()}
  def open(base_path, opts \\ []) do
    with {:ok, _pid} <- ensure_started(base_path, opts) do
      {:ok, GenServer.call(via(base_path), :load_db, :infinity)}
    end
  end

  @doc """
  Appends one committed transaction's statements (a list of `{sql, params}`)
  and returns the bumped commit version (or `:no_registry`).

  The append itself is a cast; the version is bumped here, after the cast, so
  by the time another connection can observe the new version its `fetch/1`
  call is guaranteed to queue behind this append in the writer's mailbox.
  """
  @spec append(Path.t(), [{String.t(), list()}]) :: non_neg_integer() | :no_registry | :ok
  def append(_base_path, []), do: :ok

  def append(base_path, records) do
    :ok = GenServer.cast(via(base_path), {:append, records})
    ExSQL.Registry.bump(version_key(base_path))
  end

  @doc """
  Returns the current database (base + replayed log) and its commit version.

  The call is serialized behind pending appends in the writer's mailbox, so the
  returned database reflects every append whose version bump was visible when
  the caller decided to fetch.
  """
  @spec fetch(Path.t()) :: {Database.t(), non_neg_integer() | :no_registry}
  def fetch(base_path), do: GenServer.call(via(base_path), :fetch, :infinity)

  @doc "The `ExSQL.Registry` key holding the commit version for `base_path`."
  @spec version_key(Path.t()) :: term()
  def version_key(base_path), do: {:log, base_path}

  @doc "Forces pending appends to disk (fsync). Returns after the write is durable."
  @spec flush(Path.t()) :: :ok
  def flush(base_path), do: GenServer.call(via(base_path), :flush, :infinity)

  @doc "Folds the log into a fresh base file and truncates the log."
  @spec checkpoint(Path.t()) :: :ok
  def checkpoint(base_path), do: GenServer.call(via(base_path), :checkpoint, :infinity)

  @doc "Stops the writer for `base_path` (flushing first)."
  @spec stop(Path.t()) :: :ok
  def stop(base_path) do
    case Registry.lookup(ExSQL.LogRegistry, base_path) do
      [{pid, _}] -> GenServer.stop(pid)
      [] -> :ok
    end
  end

  @doc false
  def ensure_started(base_path, opts \\ []) do
    case DynamicSupervisor.start_child(ExSQL.LogSupervisor, {__MODULE__, {base_path, opts}}) do
      {:ok, pid} -> {:ok, pid}
      {:error, {:already_started, pid}} -> {:ok, pid}
      other -> other
    end
  end

  @doc false
  def start_link({base_path, opts}) do
    GenServer.start_link(__MODULE__, {base_path, opts}, name: via(base_path))
  end

  def child_spec({base_path, opts}) do
    %{
      id: {__MODULE__, base_path},
      start: {__MODULE__, :start_link, [{base_path, opts}]},
      restart: :transient
    }
  end

  defp via(base_path), do: {:via, Registry, {ExSQL.LogRegistry, base_path}}

  # -- server -------------------------------------------------------------------

  @impl true
  def init({base_path, opts}) do
    paths = paths(base_path)
    File.mkdir_p!(Path.dirname(base_path))
    recover(paths)

    io = open_log(paths.log)

    state = %{
      paths: paths,
      io: io,
      sync: Keyword.get(opts, :sync, false),
      flush_ms: Keyword.get(opts, :flush_ms, @flush_ms),
      unsynced: false
    }

    schedule_flush(state)
    {:ok, state}
  end

  @impl true
  def handle_call(:load_db, _from, state) do
    {:reply, load_from_files(state.paths), state}
  end

  def handle_call(:flush, _from, state), do: {:reply, :ok, do_sync(state)}

  def handle_call(:checkpoint, _from, state), do: {:reply, :ok, do_checkpoint(state)}

  def handle_call(:fetch, _from, state) do
    version = ExSQL.Registry.current_version(version_key(state.paths.base))
    {:reply, {load_from_files(state.paths), version}, state}
  end

  @impl true
  def handle_cast({:append, records}, state) do
    :ok = :file.write(state.io, encode(records))
    state = %{state | unsynced: true}
    {:noreply, if(state.sync, do: do_sync(state), else: state)}
  end

  @impl true
  def handle_info(:flush_tick, state) do
    state = if state.unsynced, do: do_sync(state), else: state
    schedule_flush(state)
    {:noreply, state}
  end

  @impl true
  def terminate(_reason, state) do
    _ = do_sync(state)
    _ = :file.close(state.io)
    :ok
  end

  # -- file handling ------------------------------------------------------------

  defp paths(base_path) do
    %{
      base: base_path,
      base_new: base_path <> ".new",
      log: base_path <> ".log",
      archived: base_path <> ".log.archived"
    }
  end

  defp open_log(path), do: File.open!(path, [:append, :binary, :raw])

  defp schedule_flush(%{flush_ms: ms}), do: Process.send_after(self(), :flush_tick, ms)

  defp do_sync(%{unsynced: false} = state), do: state

  defp do_sync(state) do
    :ok = :file.sync(state.io)
    %{state | unsynced: false}
  end

  # Reduce any post-crash file state back to the canonical {base, log} layout.
  defp recover(p) do
    cond do
      File.exists?(p.base_new) ->
        # Checkpoint did not reach its commit point — discard the partial base.
        # The archived log (if any) was NOT folded, so fold base + archive + log
        # into a fresh base now.
        File.rm!(p.base_new)
        canonicalize(p)

      File.exists?(p.archived) ->
        # Checkpoint committed (base already includes the archive) but cleanup
        # was interrupted — just drop the archive.
        File.rm!(p.archived)

      true ->
        :ok
    end
  end

  defp canonicalize(p) do
    db =
      read_base(p.base) |> maybe_replay(p.archived) |> maybe_replay(p.log) |> rollback_if_open()

    write_base(db, p)
    if File.exists?(p.archived), do: File.rm!(p.archived)
    if File.exists?(p.log), do: File.rm!(p.log)
  end

  defp do_checkpoint(state) do
    p = state.paths
    :ok = :file.sync(state.io)
    db = load_from_files(p)
    write_base_new(db, p)

    :ok = :file.close(state.io)
    :ok = File.rename(p.log, p.archived)
    io = open_log(p.log)

    :ok = File.rename(p.base_new, p.base)
    if File.exists?(p.archived), do: File.rm!(p.archived)

    %{state | io: io, unsynced: false}
  end

  # base + replay(active log). After recover/1 there is never an archive here.
  defp load_from_files(p), do: read_base(p.base) |> maybe_replay(p.log) |> rollback_if_open()

  # Each statement (including BEGIN/COMMIT/SAVEPOINT) is logged verbatim and
  # replayed in order, so transaction semantics reproduce themselves. A log that
  # ends mid-transaction (a crash before COMMIT) leaves the engine with an open
  # transaction; discard it, mirroring SQLite rolling back an incomplete txn on
  # recovery.
  defp rollback_if_open(%Database{txn_stack: []} = db), do: db

  defp rollback_if_open(db) do
    case Executor.run(db, "ROLLBACK") do
      {:ok, _results, db} -> db
      {:error, _error, db} -> db
    end
  end

  defp read_base(base) do
    if File.exists?(base) do
      case FileFormat.read(base) do
        {:ok, db} -> db
        {:error, message} -> raise ExSQL.Error, message: "log: unreadable base file: #{message}"
      end
    else
      Database.new()
    end
  end

  defp maybe_replay(db, log_path) do
    case File.read(log_path) do
      {:ok, bin} -> bin |> decode_frames([]) |> Enum.reduce(db, &apply_record/2)
      _ -> db
    end
  end

  defp apply_record(records, db) do
    Enum.reduce(records, db, fn {sql, params}, db ->
      case Executor.run(db, sql, params) do
        {:ok, _results, db} -> db
        {:error, _error, db} -> db
      end
    end)
  end

  # base.new := db, fsync, then rename over base.
  defp write_base(db, p) do
    write_base_new(db, p)
    :ok = File.rename(p.base_new, p.base)
  end

  defp write_base_new(db, p) do
    case FileFormat.write(db, p.base_new, journal_mode: :memory) do
      {:ok, _path} -> fsync_file(p.base_new)
      {:error, message} -> raise ExSQL.Error, message: "log: cannot write base: #{message}"
    end
  end

  defp fsync_file(path) do
    {:ok, fd} = :file.open(path, [:read, :write, :raw, :binary])
    :ok = :file.sync(fd)
    :ok = :file.close(fd)
  end

  # -- framing ------------------------------------------------------------------

  defp encode(records) do
    payload = :erlang.term_to_binary(records)
    <<byte_size(payload)::32, :erlang.crc32(payload)::32, payload::binary>>
  end

  defp decode_frames(<<size::32, crc::32, rest::binary>>, acc) when byte_size(rest) >= size do
    <<payload::binary-size(^size), tail::binary>> = rest

    if :erlang.crc32(payload) == crc do
      decode_frames(tail, [:erlang.binary_to_term(payload, [:safe]) | acc])
    else
      Enum.reverse(acc)
    end
  end

  defp decode_frames(_partial, acc), do: Enum.reverse(acc)
end