add verified parser: first checks to parse the file without emiting envents; if is ok, then parses emiting

lib/feed_reader.ex

@@ -24,7 +24,8 @@ defmodule FeedReader do
     max_compressed_bytes: 52_428_800,
     max_decoded_bytes: 52_428_800,
     max_temp_bytes: 52_428_800,
-    max_expansion_ratio: 100
+    max_expansion_ratio: 100,
+    max_parse_passes: 5
   ]
 
   @parsing_options_doc """
@@ -49,6 +50,8 @@ defmodule FeedReader do
   - `:max_queue_bytes` — non-negative maximum serialized size of one pending
     event; defaults to `:max_item_bytes`. Zero is valid; invalid values use
     that default.
+  - Verified APIs also accept `:max_parse_passes` (default
+    `#{@option_defaults[:max_parse_passes]}`), counting validation and delivery.
   """
 
   @remote_options_doc """
@@ -131,6 +134,22 @@ defmodule FeedReader do
     FeedReader.Parser.stream_file(path, opts)
   end
 
+  @doc """
+  Validates the complete caller-owned local feed before returning an event session.
+
+  The source file is never removed. `:max_parse_passes` bounds validation and
+  event delivery passes.
+
+  ## Options
+
+  #{String.trim(@parsing_options_doc)}
+  """
+  @spec stream_file_verified(Path.t(), keyword()) ::
+          {:ok, FeedReader.Session.t()} | {:error, term()}
+  def stream_file_verified(path, opts \\ []) do
+    FeedReader.VerifiedParser.stream_file(path, opts)
+  end
+
   @doc """
   Downloads and checksums a feed before returning a lazy event session.
 
@@ -157,6 +176,43 @@ defmodule FeedReader do
     end
   end
 
+  @doc """
+  Downloads and validates a feed before exposing its remote event session.
+
+  Matching checksums and conditional HTTP 304 responses return `:not_changed`
+  and `:not_modified` without a session. Call `close/1` if enumeration may stop
+  early.
+
+  ## Options
+
+  #{String.trim(@remote_options_doc)}
+
+  #{String.trim(@parsing_options_doc)}
+  """
+  @spec stream_verified(String.t(), keyword()) ::
+          {:ok, FeedReader.Session.t() | :not_changed | :not_modified} | {:error, term()}
+  def stream_verified(url, opts \\ []) do
+    case FeedReader.Downloader.download(url, opts) do
+      {:ok, %{path: path, fetch: fetch} = download} ->
+        parser_opts =
+          opts
+          |> Keyword.put_new(:feed_url, fetch.final_url)
+          |> Keyword.put(:owned_source_path, path)
+
+        case FeedReader.VerifiedParser.stream_file(path, parser_opts) do
+          {:ok, %FeedReader.Session{} = session} ->
+            {:ok, %FeedReader.Session{session | fetch: fetch}}
+
+          {:error, reason} ->
+            FeedReader.Downloader.close(download)
+            {:error, reason}
+        end
+
+      result ->
+        result
+    end
+  end
+
   defp create_session(%{path: path, fetch: fetch} = download, opts) do
     parser_opts = Keyword.put_new(opts, :feed_url, fetch.final_url)
 
@@ -181,7 +237,7 @@ defmodule FeedReader do
     end
   end
 
-  @doc "Closes a remote session; safe after completion or repeated calls."
+  @doc "Closes an owned remote session; safe after completion or repeated calls."
   @spec close(FeedReader.Session.t()) :: :ok
   def close(%FeedReader.Session{resource: %{path: _path} = resource}) do
     FeedReader.Downloader.close(resource)

lib/feed_reader/verified_parser.ex

@@ -0,0 +1,57 @@
+defmodule FeedReader.VerifiedParser do
+  @moduledoc "Validates a complete XML pass before returning a feed event session."
+
+  alias FeedReader.{Parser, Session}
+
+  @default_max_passes 5
+
+  @doc "Validates a local or library-owned file before exposing its events."
+  @spec stream_file(Path.t(), keyword()) :: {:ok, Session.t()} | {:error, term()}
+  def stream_file(path, opts) do
+    max_passes = positive_option(opts, :max_parse_passes, @default_max_passes)
+
+    with :ok <- validate(path, opts),
+         :ok <- require_delivery_pass(max_passes),
+         {:ok, %Session{} = session} <- Parser.stream_file(path, opts) do
+      resource = owned_resource(opts)
+
+      events =
+        Stream.transform(
+          session.events,
+          fn -> :ok end,
+          fn event, state -> {[event], state} end,
+          fn _state ->
+            if resource, do: FeedReader.close(%Session{resource: resource}), else: :ok
+          end
+        )
+
+      {:ok, %Session{session | events: events, resource: resource}}
+    end
+  end
+
+  defp validate(path, opts) do
+    with {:ok, %{events: events}} <- Parser.stream_file(path, opts) do
+      case Enum.find(events, &match?({:error, _}, &1)) do
+        {:error, reason} -> {:error, reason}
+        nil -> :ok
+      end
+    end
+  end
+
+  defp require_delivery_pass(limit) when limit >= 2, do: :ok
+  defp require_delivery_pass(limit), do: {:error, {:resource_limit, :max_parse_passes, limit}}
+
+  defp owned_resource(opts) do
+    case Keyword.get(opts, :owned_source_path) do
+      path when is_binary(path) -> %{path: path}
+      _ -> nil
+    end
+  end
+
+  defp positive_option(opts, key, default) do
+    case Keyword.get(opts, key, default) do
+      value when is_integer(value) and value > 0 -> value
+      _ -> default
+    end
+  end
+end