opencadc_vospace_client

defmodule OpencadcVospaceClient do
  @moduledoc """
  VOSpace client to upload a stream of data.  Useful for transferring from one location to a VOSpace.
  """

  @vos_namespace "http://www.ivoa.net/xml/VOSpace/v2.0"

  
  @doc """
  Upload a file to a VOSpace

  `destination_uri`:    VOS URI to put to.

  `cookie_credential`:   Cookie value for auth.
  """
  def upload(destination_uri, cookie_credential) do
    (OpencadcRegistryClient.lookupServiceURL(
       "ivo://cadc.nrc.ca/vospace",
       "ivo://ivoa.net/std/VOSpace/v2.0#nodes",
       "vs:ParamHTTP"
     ) <> URI.parse(destination_uri).path)
    |> HTTPoison.put!(
      to_xml(destination_uri),
      [{"Content-Type", "text/xml"}, {"Cookie", "CADC_SSO=\"#{cookie_credential}\""}],
      [{:follow_redirect, true}, hackney: [:force_redirect, true]]
    )
    |> write(destination_uri, cookie_credential)
  end

  defp write(%HTTPoison.Response{status_code: 200}, destination_uri, cookie_credential) do
    with transferServiceURL =
           OpencadcRegistryClient.lookupServiceURL(
             "ivo://cadc.nrc.ca/vospace",
             "ivo://ivoa.net/std/VOSpace/v2.0#transfers",
             "vs:ParamHTTP"
           ),
         vospaceSyncURL =
           OpencadcRegistryClient.lookupServiceURL(
             "ivo://cadc.nrc.ca/vospace",
             "ivo://ivoa.net/std/VOSpace#sync-2.1",
             "vs:ParamHTTP"
           ),
         xmlBody = to_xml(destination_uri, transferServiceURL) do
      HTTPoison.post!(
        vospaceSyncURL,
        xmlBody,
        [
          {"Content-Type", "text/xml"},
          {"Cookie", "CADC_SSO=\"#{cookie_credential}\""}
        ]
      )
      |> upload_to(cookie_credential)
    end
  end

  defp write(%HTTPoison.Response{status_code: 401}, destination_uri, _) do
    raise ArgumentError, message: "Write permission denied on " <> destination_uri
  end

  defp write(%HTTPoison.Response{status_code: 404}, destination_uri, _) do
    raise ArgumentError,
      message:
        "No such target #{destination_uri}!  This service will create the path eventually..."
  end

  defp write(%HTTPoison.Response{status_code: 409}, destination_uri, _) do
    raise ArgumentError,
      message: "The path #{destination_uri} already exists.  Please select another destination."
  end

  defp upload_to(%HTTPoison.Response{status_code: 303} = response, cookie_credential) do
    response.headers
    |> get_header("Location")
    |> HTTPoison.get!([{"Cookie", "CADC_SSO=\"#{cookie_credential}\""}])
    |> upload_to(cookie_credential)
  end

  defp upload_to(%HTTPoison.Response{status_code: 200} = response, cookie_credential) do
    with transfer_url = Parser.all(
      Parser.from_string(response.body),
      "//vos:transfer/vos:protocol[@uri=\"ivo://ivoa.net/vospace/core#httpput\"]/vos:endpoint"
    )
    |> extract_transfer_endpoint_url do
      # Dip into Hackney to allow streaming the body asynchronously.
      :hackney.request("put", transfer_url, [{<<"Cookie">>, <<"CADC_SSO=\"#{cookie_credential}\"">>}], :stream, [])
    end
  end

  defp upload_to(response, _) do
    raise ArgumentError, message: response.body
  end

  defp extract_transfer_endpoint_url([]) do
    raise ArgumentError, message: "No service endpoints found to transfer to."
  end

  defp extract_transfer_endpoint_url(endpoints) do
    endpoints |> Enum.at(0) |> Parser.text()
  end

  defp to_xml(destination_uri) do
    "<vos:node uri=\"" <>
      destination_uri <>
      "\" busy=\"false\" xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\" xsi:type=\"vos:DataNode\" xmlns:vos=\"#{
        @vos_namespace
      }\"><vos:properties/></vos:node>"
  end

  defp to_xml(destination_uri, transferURL) do
    "<vos:transfer xmlns:vos=\"#{@vos_namespace}\"><vos:target>#{destination_uri}</vos:target><vos:direction>pushToVoSpace</vos:direction><vos:view uri=\"ivo://ivoa.net/vospace/core#defaultview\"/><vos:protocol uri=\"ivo://ivoa.net/vospace/core#httpput\"><vos:endpoint>#{
      transferURL
    }</vos:endpoint></vos:protocol><vos:keepBytes>true</vos:keepBytes></vos:transfer>"
  end

  defp get_header(headers, key) do
    headers
    |> Enum.filter(fn {k, _} -> k == key end)
    |> hd
    |> elem(1)
  end
end