defmodule DB.Dataset do
  @moduledoc """
  Dataset schema

  There's a trigger on update on postgres to update the search vector.
  There are also trigger on update on aom and region that will force an update on this model
  so the search vector is up-to-date.
  """
  alias DB.{AOM, Commune, LogsImport, NotificationSubscription, Region, Repo, Resource}
  alias Phoenix.HTML.Link
  import Ecto.{Changeset, Query}
  use Gettext, backend: TransportWeb.Gettext
  require Logger
  use Ecto.Schema
  use TypedEctoSchema

  @type conversion_details :: %{
          url: binary(),
          filesize: pos_integer(),
          resource_history_last_up_to_date_at: DateTime.t(),
          format: binary(),
          stable_url: binary()
        }

  @licences_ouvertes ["fr-lo", "lov2"]
  @licence_mobilités_tag "licence-mobilités"
  @hidden_dataset_custom_tag_value "masqué"
  @experimental_tag "experimental"

  typed_schema "dataset" do
    field(:datagouv_id, :string)
    field(:custom_title, :string)
    # `created_at` comes from data.gouv.fr.
    # To know when the dataset has been added on the NAP, use `inserted_at`
    field(:created_at, :utc_datetime_usec)
    field(:description, :string)
    field(:frequency, :string)
    field(:last_update, :utc_datetime_usec)
    field(:licence, :string)
    field(:logo, :string)
    field(:full_logo, :string)
    field(:slug, :string)
    field(:tags, {:array, :string})
    field(:datagouv_title, :string)
    field(:type, :string)
    field(:organization, :string)
    field(:organization_type, :string)
    field(:has_realtime, :boolean)
    field(:is_active, :boolean)
    field(:is_hidden, :boolean)
    field(:population, :integer)
    field(:nb_reuses, :integer)
    field(:latest_data_gouv_comment_timestamp, :utc_datetime)
    field(:archived_at, :utc_datetime_usec)
    field(:custom_tags, {:array, :string}, default: [])
    # URLs for custom logos.
    # Currently we host custom logos in Cellar buckets.
    # See config: `:transport, :logos_bucket_url`
    field(:custom_logo, :string)
    field(:custom_full_logo, :string)
    field(:custom_logo_changed_at, :utc_datetime_usec)

    timestamps(type: :utc_datetime_usec)

    # When the dataset is linked to some cities
    many_to_many(:communes, Commune, join_through: "dataset_communes", on_replace: :delete)

    many_to_many(:legal_owners_aom, AOM,
      join_through: "dataset_aom_legal_owner",
      on_replace: :delete
    )

    many_to_many(:legal_owners_region, Region,
      join_through: "dataset_region_legal_owner",
      on_replace: :delete
    )

    # This links the dataset to one or more `DB.AdministrativeDivision`. The x/y data
    # contained within the dataset's resources should more or less be consistent with
    # the polygons defining the territories of associated administrative divisions.
    #
    # Not to be confused with the "legal owners" (or even with the "publishers")
    # of the data.
    many_to_many(:declarative_spatial_areas, DB.AdministrativeDivision,
      join_through: "dataset_declarative_spatial_area",
      on_replace: :delete
    )

    field(:legal_owner_company_siren, :string)

    has_many(:resources, Resource, on_replace: :delete, on_delete: :delete_all)
    has_many(:logs_import, LogsImport, on_replace: :delete, on_delete: :delete_all)
    has_many(:notification_subscriptions, NotificationSubscription, on_delete: :delete_all)
    has_many(:reuser_improved_data, DB.ReuserImprovedData, on_delete: :delete_all)

    # Deprecation notice: datasets won't be linked to region and aom like that in the future
    # ⬇️⬇️⬇️

    # A Dataset can be linked to *either*:
    # - a Region (and there is a special Region 'national' that represents the national datasets);
    # - an AOM;
    # - or a list of cities.
    belongs_to(:region, Region)
    belongs_to(:aom, AOM)
    belongs_to(:organization_object, DB.Organization, foreign_key: :organization_id, type: :string, on_replace: :nilify)

    # we ask in the backoffice for a name to display
    # (used in the long title of a dataset and to find the associated datasets)
    field(:associated_territory_name, :string)

    field(:search_payload, :map)
    many_to_many(:followers, DB.Contact, join_through: "dataset_followers", on_replace: :delete)
  end

  def base_query do
    from(d in DB.Dataset, as: :dataset, where: d.is_active and not d.is_hidden)
  end

  def all_datasets, do: from(d in DB.Dataset, as: :dataset)
  def archived, do: base_query() |> where([dataset: d], not is_nil(d.archived_at))
  def inactive, do: from(d in DB.Dataset, as: :dataset, where: not d.is_active)
  def hidden, do: from(d in DB.Dataset, as: :dataset, where: d.is_active and d.is_hidden)
  def include_hidden_datasets(%Ecto.Query{} = query), do: or_where(query, [dataset: d], d.is_hidden)
  def base_with_hidden_datasets, do: base_query() |> include_hidden_datasets()

  @spec archived?(__MODULE__.t()) :: boolean()
  def archived?(%__MODULE__{archived_at: nil}), do: false
  def archived?(%__MODULE__{archived_at: %DateTime{}}), do: true

  @spec active?(__MODULE__.t()) :: boolean()
  def active?(%__MODULE__{is_active: is_active}), do: is_active

  @doc """
  Creates a query with the following inner joins:
  datasets <> Resource <> ResourceHistory <> MultiValidation <> ResourceMetadata
  """
  def join_from_dataset_to_metadata(query, validator_name) do
    query
    |> DB.Resource.join_dataset_with_resource()
    |> DB.ResourceHistory.join_resource_with_latest_resource_history()
    |> DB.MultiValidation.join_resource_history_with_latest_validation(validator_name)
    |> DB.ResourceMetadata.join_validation_with_metadata()
  end

  @doc """
  Returns a list of resources, with their last resource_history preloaded
  """
  def last_resource_history(dataset_id) do
    DB.Dataset.all_datasets()
    |> where([dataset: d], d.id == ^dataset_id)
    |> join(:left, [dataset: d], r in DB.Resource, on: d.id == r.dataset_id, as: :resource)
    |> join(:left, [resource: r], rh in DB.ResourceHistory,
      on: rh.resource_id == r.id,
      as: :resource_history
    )
    |> distinct([resource: r], r.id)
    |> order_by([resource: r, resource_history: rh],
      asc: r.id,
      desc: rh.inserted_at
    )
    |> preload([resource: r, resource_history: rh], resources: {r, resource_history: rh})
    |> DB.Repo.one!()
    |> Map.get(:resources, [])
  end

  @spec type_to_str_map() :: %{binary() => binary()}
  def type_to_str_map,
    do: %{
      "public-transit" => dgettext("db-dataset", "Public transit"),
      "carpooling-areas" => dgettext("db-dataset", "Carpooling areas"),
      "carpooling-lines" => dgettext("db-dataset", "Carpooling lines"),
      "carpooling-offers" => dgettext("db-dataset", "Carpooling offers"),
      "charging-stations" => dgettext("db-dataset", "Charging & refuelling stations"),
      "vehicles-sharing" => dgettext("db-dataset", "Vehicles sharing"),
      "road-data" => dgettext("db-dataset", "Road data"),
      "informations" => dgettext("db-dataset", "Other informations"),
      "bike-data" => dgettext("db-dataset", "Bike data"),
      "pedestrian-path" => dgettext("db-dataset", "Pedestrian path")
    }

  @spec type_to_str(binary()) :: binary()
  def type_to_str(type), do: type_to_str_map()[type]

  @spec types() :: [binary()]
  def types, do: Map.keys(type_to_str_map())

  @spec no_validations_query() :: Ecto.Query.t()
  defp no_validations_query do
    from(r in Resource,
      select: %Resource{
        format: r.format,
        title: r.title,
        url: r.url,
        id: r.id,
        dataset_id: r.dataset_id,
        datagouv_id: r.datagouv_id,
        last_update: r.last_update,
        latest_url: r.latest_url,
        is_community_resource: r.is_community_resource,
        is_available: r.is_available,
        description: r.description,
        community_resource_publisher: r.community_resource_publisher,
        original_resource_url: r.original_resource_url,
        schema_name: r.schema_name,
        schema_version: r.schema_version,
        type: r.type,
        display_position: r.display_position
      }
    )
  end

  @spec preload_without_validations() :: Ecto.Query.t()
  defp preload_without_validations do
    s = no_validations_query()
    __MODULE__ |> preload(resources: ^s)
  end

  @spec preload_without_validations(Ecto.Query.t()) :: Ecto.Query.t()
  defp preload_without_validations(query) do
    s = no_validations_query()
    query |> preload(resources: ^s)
  end

  @spec filter_by_fulltext(Ecto.Query.t(), map()) :: Ecto.Query.t()
  defp filter_by_fulltext(query, %{"q" => ""}), do: query

  defp filter_by_fulltext(query, %{"q" => q}) do
    where(
      query,
      [d],
      fragment("search_vector @@ plainto_tsquery('custom_french', ?) or unaccent(datagouv_title) = unaccent(?)", ^q, ^q)
    )
  end

  defp filter_by_fulltext(query, _), do: query

  @spec filter_by_region(Ecto.Query.t(), map()) :: Ecto.Query.t()
  defp filter_by_region(query, %{"region" => region}) do
    query
    |> join(:inner, [dataset: d], r in DB.Region, on: r.insee == ^region, as: :region)
    |> join(:inner, [dataset: d, region: r], d_geo in DB.DatasetGeographicView,
      on: d.id == d_geo.dataset_id and d_geo.region_id == r.id
    )
  end

  defp filter_by_region(query, _), do: query

  @spec filter_by_departement(Ecto.Query.t(), map()) :: Ecto.Query.t()
  defp filter_by_departement(query, %{"departement" => insee}) do
    query
    |> where(
      [dataset: d],
      fragment(
        """
        (
          ? IN (
              select dataset_id
              from (
                -- departement
                select distinct d.id dataset_id, 1 as filter
                from dataset d
                join administrative_division ad on ad.type = 'departement' and ad.insee = ?
                join dataset_declarative_spatial_area ddsa on ddsa.administrative_division_id = ad.id and d.id = ddsa.dataset_id
                union
                -- commune
                select distinct d.id dataset_id, 2 as filter
                from dataset d
                join departement de on de.insee = ?
                join commune c on c.departement_insee = de.insee
                join administrative_division ad on ad.type = 'commune' and c.insee = ad.insee
                join dataset_declarative_spatial_area ddsa on ddsa.administrative_division_id = ad.id and d.id = ddsa.dataset_id
                union
                -- epci
                select distinct d.id dataset_id, 3 as filter
                from dataset d
                join departement de on de.insee = ?
                join commune c on c.departement_insee = de.insee
                join administrative_division ad on ad.type = 'epci' and c.epci_insee = ad.insee
                join dataset_declarative_spatial_area ddsa on ddsa.administrative_division_id = ad.id and d.id = ddsa.dataset_id
                union
                -- region
                select distinct d.id dataset_id, 4 as filter
                from dataset d
                join departement de on de.insee = ?
                join commune c on c.departement_insee = de.insee
                join region r on r.id = c.region_id
                join administrative_division ad on ad.type = 'region' and r.insee = ad.insee
                join dataset_declarative_spatial_area ddsa on ddsa.administrative_division_id = ad.id and d.id = ddsa.dataset_id
              ) t
              order by filter
        ))
        """,
        d.id,
        ^insee,
        ^insee,
        ^insee,
        ^insee
      )
    )
  end

  defp filter_by_departement(query, _), do: query

  @spec filter_by_category(Ecto.Query.t(), map()) :: Ecto.Query.t()
  defp filter_by_category(query, %{"filter" => filter_key}) do
    case filter_key do
      "has_realtime" -> where(query, [d], d.has_realtime == true)
      "urban_public_transport" -> where(query, [d], not is_nil(d.aom_id) and d.type == "public-transit")
      _ -> query
    end
  end

  defp filter_by_category(query, _), do: query

  @spec filter_by_custom_tag(Ecto.Query.t(), binary() | map()) :: Ecto.Query.t()
  def filter_by_custom_tag(%Ecto.Query{} = query, custom_tag) when is_binary(custom_tag) do
    where(query, [dataset: d], ^custom_tag in d.custom_tags)
  end

  def filter_by_custom_tag(%Ecto.Query{} = query, %{"custom_tag" => custom_tag}),
    do: filter_by_custom_tag(query, custom_tag)

  def filter_by_custom_tag(%Ecto.Query{} = query, _), do: query

  @spec filter_by_feature(Ecto.Query.t(), map()) :: Ecto.Query.t()
  defp filter_by_feature(query, %{"features" => [feature]})
       when feature in ["service_alerts", "trip_updates", "vehicle_positions"] do
    recent_limit = Transport.Jobs.GTFSRTMetadataJob.datetime_limit()

    query
    |> join(:inner, [dataset: d], r in DB.Resource, on: r.dataset_id == d.id, as: :resource)
    |> join(
      :inner,
      [resource: r],
      rm in DB.ResourceMetadata,
      on:
        rm.resource_id == r.id and
          fragment(
            "? = ANY (?)",
            ^feature,
            rm.features
          ) and
          rm.inserted_at > ^recent_limit
    )
  end

  defp filter_by_feature(query, %{"features" => feature}) do
    query
    |> join(:inner, [dataset: d], r in assoc(d, :resources), as: :resource_for_features)
    |> where([resource_for_features: r], fragment("?->'gtfs_features' @> ?", r.counter_cache, ^feature))
  end

  defp filter_by_feature(query, _), do: query

  @spec filter_by_mode(Ecto.Query.t(), map()) :: Ecto.Query.t()
  defp filter_by_mode(query, %{"modes" => modes}) when is_list(modes) do
    query
    |> join(:inner, [dataset: d], r in assoc(d, :resources), as: :resource_for_mode)
    |> where([resource_for_mode: r], fragment("?->'gtfs_modes' @> ?", r.counter_cache, ^modes))
  end

  defp filter_by_mode(query, _), do: query

  @spec filter_by_resource_format(Ecto.Query.t(), map()) :: Ecto.Query.t()
  defp filter_by_resource_format(query, %{"format" => format}) do
    query
    |> join(:inner, [dataset: d], r in assoc(d, :resources), as: :resources)
    |> where([resources: r], r.format == ^format)
  end

  defp filter_by_resource_format(query, _), do: query

  @spec filter_by_type(Ecto.Query.t(), map()) :: Ecto.Query.t()
  defp filter_by_type(query, %{"type" => type}), do: where(query, [d], d.type == ^type)
  defp filter_by_type(query, _), do: query

  @spec filter_by_epci(Ecto.Query.t(), map()) :: Ecto.Query.t()
  defp filter_by_epci(query, %{"epci" => epci}) do
    query
    |> where(
      [dataset: d],
      fragment(
        """
        (
          ? IN (
              select dataset_id
              from (
                -- region
                select distinct d.id dataset_id, 1 as filter
                from dataset d
                join epci e on e.insee = ?
                join commune c on c.epci_insee = e.insee
                join region r on r.id = c.region_id
                join administrative_division ad on ad.type = 'region' and r.insee = ad.insee
                join dataset_declarative_spatial_area ddsa on ddsa.administrative_division_id = ad.id and d.id = ddsa.dataset_id
                union
                -- departement
                select distinct d.id dataset_id, 2 as filter
                from dataset d
                join epci e on e.insee = ?
                join commune c on c.epci_insee = e.insee
                join administrative_division ad on ad.type = 'departement' and c.departement_insee = ad.insee
                join dataset_declarative_spatial_area ddsa on ddsa.administrative_division_id = ad.id and d.id = ddsa.dataset_id
                union
                -- epci
                select distinct d.id dataset_id, 3 as filter
                from dataset d
                join epci e on e.insee = ?
                join administrative_division ad on ad.type = 'epci' and e.insee = ad.insee
                join dataset_declarative_spatial_area ddsa on ddsa.administrative_division_id = ad.id and d.id = ddsa.dataset_id
                union
                -- commune
                select distinct d.id dataset_id, 4 as filter
                from dataset d
                join epci e on e.insee = ?
                join commune c on c.epci_insee = e.insee
                join administrative_division ad on ad.type = 'commune' and c.insee = ad.insee
                join dataset_declarative_spatial_area ddsa on ddsa.administrative_division_id = ad.id and d.id = ddsa.dataset_id
              ) t
              order by filter
            )
        )
        """,
        d.id,
        ^epci,
        ^epci,
        ^epci,
        ^epci
      )
    )
  end

  defp filter_by_epci(query, _), do: query

  @spec filter_by_commune(Ecto.Query.t(), map()) :: Ecto.Query.t()
  defp filter_by_commune(query, %{"commune" => commune}) do
    query
    |> where(
      [dataset: d],
      fragment(
        """
        (
          ? IN (
              select dataset_id
              from (
                -- region
                select distinct d.id dataset_id, 1 as filter
                from dataset d
                join commune c on c.insee = ?
                join region r on r.id = c.region_id
                join administrative_division ad on ad.type = 'region' and r.insee = ad.insee
                join dataset_declarative_spatial_area ddsa on ddsa.administrative_division_id = ad.id and d.id = ddsa.dataset_id
                union
                -- departement
                select distinct d.id dataset_id, 2 as filter
                from dataset d
                join commune c on c.insee = ?
                join administrative_division ad on ad.type = 'departement' and c.departement_insee = ad.insee
                join dataset_declarative_spatial_area ddsa on ddsa.administrative_division_id = ad.id and d.id = ddsa.dataset_id
                union
                -- epci
                select distinct d.id dataset_id, 3 as filter
                from dataset d
                join commune c on c.insee = ?
                join administrative_division ad on ad.type = 'epci' and c.epci_insee = ad.insee
                join dataset_declarative_spatial_area ddsa on ddsa.administrative_division_id = ad.id and d.id = ddsa.dataset_id
                union
                -- commune
                select distinct d.id dataset_id, 4 as filter
                from dataset d
                join commune c on c.insee = ?
                join administrative_division ad on ad.type = 'commune' and c.insee = ad.insee
                join dataset_declarative_spatial_area ddsa on ddsa.administrative_division_id = ad.id and d.id = ddsa.dataset_id
              ) t
              order by filter
            )
        )
        """,
        d.id,
        ^commune,
        ^commune,
        ^commune,
        ^commune
      )
    )
  end

  defp filter_by_commune(query, _), do: query

  @spec filter_by_licence(Ecto.Query.t(), map()) :: Ecto.Query.t()
  defp filter_by_licence(query, %{"licence" => "licence-ouverte"}),
    do: where(query, [d], d.licence in @licences_ouvertes)

  defp filter_by_licence(query, %{"licence" => licence}), do: where(query, [d], d.licence == ^licence)
  defp filter_by_licence(query, _), do: query

  @spec filter_by_organization(Ecto.Query.t(), map()) :: Ecto.Query.t()
  defp filter_by_organization(query, %{"organization_id" => organization_id}) do
    where(query, [d], d.organization_id == ^organization_id)
  end

  defp filter_by_organization(query, _), do: query

  @spec list_datasets(map()) :: Ecto.Query.t()
  def list_datasets(%{} = params) do
    params
    |> list_datasets_no_order()
    |> order_datasets(params)
  end

  @spec list_datasets_no_order(map()) :: Ecto.Query.t()
  def list_datasets_no_order(%{} = params) do
    q =
      base_query()
      |> distinct([dataset: d], d.id)
      |> filter_by_region(params)
      |> filter_by_departement(params)
      |> filter_by_epci(params)
      |> filter_by_commune(params)
      |> filter_by_feature(params)
      |> filter_by_mode(params)
      |> filter_by_category(params)
      |> filter_by_type(params)
      |> filter_by_licence(params)
      |> filter_by_custom_tag(params)
      |> filter_by_organization(params)
      |> filter_by_resource_format(params)
      |> filter_by_fulltext(params)
      |> select([dataset: d], d.id)

    base_query()
    |> where([dataset: d], d.id in subquery(q))
    |> preload_without_validations()
  end

  @spec order_datasets(Ecto.Query.t(), map()) :: Ecto.Query.t()
  def order_datasets(datasets, %{"order_by" => "alpha"}), do: order_by(datasets, asc: :custom_title)
  def order_datasets(datasets, %{"order_by" => "most_recent"}), do: order_by(datasets, desc_nulls_last: :inserted_at)

  def order_datasets(datasets, %{"q" => q}),
    do:
      order_by(datasets,
        desc: fragment("ts_rank_cd(search_vector, plainto_tsquery('custom_french', ?), 32) DESC, population", ^q),
        asc: :custom_title
      )

  def order_datasets(datasets, _params) do
    pan_publisher = Application.fetch_env!(:transport, :datagouvfr_transport_publisher_id)

    order_by(datasets,
      desc:
        fragment(
          "case when organization_id = ? and custom_title ilike 'base nationale%' then 1 else 0 end",
          ^pan_publisher
        ),
      # Gotcha, population can be null for datasets covering France/Europe
      # https://github.com/etalab/transport-site/issues/3848
      desc: fragment("coalesce(population, 100000000)"),
      asc: :custom_title
    )
  end

  @spec changeset(map()) :: {:error, binary()} | {:ok, Ecto.Changeset.t()}
  def changeset(params = %{}) when is_map_key(params, "datagouv_id") or is_map_key(params, "dataset_id") do
    dataset =
      __MODULE__
      |> preload([:legal_owners_aom, :legal_owners_region])
      |> get_dataset(params)
      |> case do
        nil -> %__MODULE__{}
        dataset -> dataset
      end

    apply_changeset(dataset, params)
  end

  def changeset(_) do
    {:error, "datagouv_id or dataset_id are required"}
  end

  # this case is used to update a dataset slug from backoffice without changing the dataset_id
  # useful when the dataset has been deleted / recreated on data.gouv.fr but we want to keep the resources history
  def get_dataset(query, %{"dataset_id" => dataset_id}) do
    query |> Repo.get(dataset_id)
  end

  def get_dataset(query, %{"datagouv_id" => datagouv_id}) when is_binary(datagouv_id) do
    query |> Repo.get_by(datagouv_id: datagouv_id)
  end

  defp apply_changeset(%__MODULE__{} = dataset, params) do
    territory_name = Map.get(params, "associated_territory_name") || dataset.associated_territory_name

    dataset =
      dataset
      |> Repo.preload([
        :legal_owners_aom,
        :legal_owners_region,
        :declarative_spatial_areas
      ])

    legal_owners_aom = get_legal_owners_aom(dataset, params)
    legal_owners_region = get_legal_owners_region(dataset, params)
    declarative_spatial_areas = get_administrative_divisions(dataset, params)

    dataset
    |> Repo.preload([
      :resources,
      :communes,
      :region,
      :organization_object
    ])
    |> cast(params, [
      :datagouv_id,
      :custom_title,
      :created_at,
      :description,
      :frequency,
      :organization_type,
      :last_update,
      :licence,
      :logo,
      :full_logo,
      :slug,
      :tags,
      :datagouv_title,
      :type,
      :region_id,
      :nb_reuses,
      :is_active,
      :associated_territory_name,
      :latest_data_gouv_comment_timestamp,
      :archived_at,
      :custom_tags,
      :legal_owner_company_siren,
      :custom_logo,
      :custom_full_logo,
      :custom_logo_changed_at,
      :organization_id
    ])
    |> update_change(:custom_title, &String.trim/1)
    |> cast_aom(params)
    |> cast_datagouv_zone(params, territory_name)
    |> cast_nation_dataset(params)
    |> cast_assoc(:resources)
    |> validate_siren()
    |> validate_territory_mutual_exclusion()
    |> maybe_overwrite_licence()
    |> has_real_time()
    |> set_is_hidden()
    |> set_population(declarative_spatial_areas)
    |> validate_organization_type()
    |> add_organization(params)
    |> maybe_set_custom_logo_changed_at()
    |> put_assoc(:legal_owners_aom, legal_owners_aom)
    |> put_assoc(:legal_owners_region, legal_owners_region)
    |> put_assoc(:declarative_spatial_areas, declarative_spatial_areas)
    |> validate_required([
      :datagouv_id,
      :custom_title,
      :licence,
      :slug,
      :datagouv_title,
      :type,
      :organization_id
    ])
    |> case do
      %{valid?: false, changes: changes} = changeset when changes == %{} ->
        {:ok, %{changeset | action: :ignore}}

      %{valid?: true} = changeset ->
        {:ok, changeset}

      %{valid?: false} = errors ->
        Logger.warning("error while importing dataset: #{format_error(errors)}")
        {:error, format_error(errors)}
    end
  end

  defp add_organization(%Ecto.Changeset{} = changeset, %{"organization" => %{"id" => id, "name" => name} = org}) do
    changeset
    |> put_assoc(
      :organization_object,
      DB.Organization.changeset(DB.Repo.get(DB.Organization, id) || %DB.Organization{}, org)
    )
    |> put_change(:organization, name)
    # `put_assoc` already sets `organization_id` but we set it again
    # to pass `validate_required([:organization_id])`
    |> put_change(:organization_id, id)
  end

  defp add_organization(%Ecto.Changeset{} = changeset, _), do: changeset

  defp maybe_set_custom_logo_changed_at(%Ecto.Changeset{} = changeset) do
    if changed?(changeset, :custom_logo) do
      put_change(changeset, :custom_logo_changed_at, DateTime.utc_now())
    else
      changeset
    end
  end

  defp get_legal_owners_aom(dataset, params) do
    case params["legal_owners_aom"] do
      nil ->
        if Ecto.assoc_loaded?(dataset.legal_owners_aom) do
          # get existing aom legal owners
          dataset.legal_owners_aom
        else
          # new dataset
          []
        end

      # aom legal owners are updated from the params
      legal_owners_aom_id ->
        Repo.all(from(aom in AOM, where: aom.id in ^legal_owners_aom_id))
    end
  end

  defp get_legal_owners_region(dataset, params) do
    case params["legal_owners_region"] do
      nil ->
        if Ecto.assoc_loaded?(dataset.legal_owners_region) do
          dataset.legal_owners_region
        else
          []
        end

      legal_owners_region_id ->
        Repo.all(from(region in Region, where: region.id in ^legal_owners_region_id))
    end
  end

  defp get_administrative_divisions(dataset, params) do
    case params["declarative_spatial_areas"] do
      nil ->
        if Ecto.assoc_loaded?(dataset.declarative_spatial_areas) do
          dataset.declarative_spatial_areas
        else
          []
        end

      ids ->
        Repo.all(from(ad in DB.AdministrativeDivision, where: ad.id in ^ids))
    end
  end

  @spec format_error(any()) :: binary()
  defp format_error(changeset), do: "#{inspect(Ecto.Changeset.traverse_errors(changeset, fn {msg, _opts} -> msg end))}"

  @spec link_to_datagouv(DB.Dataset.t()) :: any()
  def link_to_datagouv(%__MODULE__{} = dataset) do
    Link.link(
      dgettext("db-dataset", "See on data.gouv.fr"),
      to: datagouv_url(dataset),
      role: "link",
      target: "_blank"
    )
  end

  @spec datagouv_url(DB.Dataset.t()) :: binary()
  def datagouv_url(%__MODULE__{slug: slug}) do
    Path.join([Application.fetch_env!(:transport, :datagouvfr_site), "datasets", slug])
  end

  @spec count_by_mode(binary()) :: number()
  def count_by_mode(mode) do
    count_by_mode_query(mode)
    |> DB.Repo.aggregate(:count, :id)
  end

  @spec count_coach() :: number()
  def count_coach do
    count_by_mode_query("bus")
    # 14 is the national "region". It means that it is not bound to a region or local territory
    |> where([dataset: d], d.region_id == 14)
    |> DB.Repo.aggregate(:count, :id)
  end

  defp count_by_mode_query(mode) do
    base_query()
    |> join(:inner, [dataset: d], r in assoc(d, :resources), as: :resource)
    |> where([resource: r], fragment("?->'gtfs_modes' @> ?", r.counter_cache, ^mode))
  end

  @spec count_by_type(binary()) :: any()
  def count_by_type(type) do
    base_query()
    |> where([d], d.type == ^type)
    |> Repo.aggregate(:count, :id)
  end

  @spec count_by_type :: map
  def count_by_type, do: for(type <- __MODULE__.types(), into: %{}, do: {type, count_by_type(type)})

  @spec count_public_transport_has_realtime :: number()
  def count_public_transport_has_realtime do
    base_query()
    |> where([d], d.has_realtime and d.type == "public-transit")
    |> Repo.aggregate(:count, :id)
  end

  @spec count_by_custom_tag(binary()) :: non_neg_integer()
  def count_by_custom_tag(custom_tag) do
    base_query() |> filter_by_custom_tag(custom_tag) |> Repo.aggregate(:count, :id)
  end

  @spec get_by_slug(binary) :: {:ok, __MODULE__.t()} | {:error, binary()}
  def get_by_slug(slug) do
    preload_without_validations()
    |> where(slug: ^slug)
    |> preload([
      :aom,
      :communes,
      :region,
      :legal_owners_aom,
      :legal_owners_region,
      :declarative_spatial_areas,
      resources: [:resources_related, :dataset]
    ])
    |> Repo.one()
    |> case do
      nil -> {:error, "Dataset with slug #{slug} not found"}
      dataset -> {:ok, dataset}
    end
  end

  @spec get_other_datasets(__MODULE__.t()) :: [__MODULE__.t()]
  def get_other_datasets(%__MODULE__{declarative_spatial_areas: []}), do: []

  def get_other_datasets(%__MODULE__{id: id, declarative_spatial_areas: declarative_spatial_areas}) do
    %DB.AdministrativeDivision{id: target_id} = declarative_spatial_areas |> DB.AdministrativeDivision.sorted() |> hd()

    __MODULE__.base_query()
    |> join(:inner, [dataset: d], r in assoc(d, :declarative_spatial_areas), as: :administrative_divison)
    |> where([dataset: d], d.id != ^id)
    |> where([administrative_divison: ad], ad.id == ^target_id)
    |> DB.Repo.all()
  end

  @spec get_covered_area(__MODULE__.t()) :: {:ok, binary()} | {:error, binary()}
  def get_covered_area(%__MODULE__{declarative_spatial_areas: declarative_spatial_areas}) do
    {:ok, declarative_spatial_areas |> DB.AdministrativeDivision.names()}
  end

  @spec get_covered_area_or_nil(__MODULE__.t()) :: binary() | nil
  def get_covered_area_or_nil(%__MODULE__{} = d) do
    case get_covered_area(d) do
      {:ok, t} -> t
      _ -> nil
    end
  end

  @spec official_resources(__MODULE__.t()) :: list(Resource.t())
  def official_resources(%__MODULE__{resources: resources}),
    do: resources |> Stream.reject(&DB.Resource.community_resource?/1) |> Enum.to_list()

  def official_resources(%__MODULE__{}), do: []

  @spec community_resources(__MODULE__.t()) :: list(Resource.t())
  def community_resources(%__MODULE__{resources: resources}),
    do: resources |> Stream.filter(&DB.Resource.community_resource?/1) |> Enum.to_list()

  def community_resources(%__MODULE__{}), do: []

  @spec formats(__MODULE__.t()) :: [binary]
  def formats(%__MODULE__{} = dataset) do
    dataset
    |> official_resources()
    |> Enum.map(& &1.format)
    |> Enum.reject(&is_nil/1)
    |> Enum.sort()
    |> Enum.dedup()
  end

  def formats(_), do: []

  @spec validate(binary | integer | __MODULE__.t()) :: {:error, String.t()} | {:ok, nil}
  @spec validate(binary | integer | __MODULE__.t(), force_validation: boolean()) :: {:error, String.t()} | {:ok, nil}
  def validate(d), do: validate(d, force_validation: false)

  def validate(%__MODULE__{id: id}, opt), do: validate(id, opt)

  def validate(id, opt) when is_binary(id), do: id |> String.to_integer() |> validate(opt)

  def validate(id, opt) when is_integer(id) do
    force_validation = Keyword.get(opt, :force_validation, false)
    dataset = __MODULE__ |> Repo.get!(id) |> Repo.preload(:resources)

    {real_time_resources, static_resources} =
      dataset
      |> official_resources()
      |> Enum.split_with(&Resource.real_time?/1)

    # unique period is set to nil, to force the resource history job to be executed
    static_resources
    |> Enum.map(
      &Transport.Jobs.ResourceHistoryJob.historize_and_validate_job(%{resource_id: &1.id},
        history_options: [unique: nil],
        validation_custom_args: %{"force_validation" => force_validation}
      )
    )
    |> Oban.insert_all()

    # Oban.insert_all does not enforce `unique` params
    # https://hexdocs.pm/oban/Oban.html#insert_all/3
    # This is something we rely on to force the job execution
    real_time_resources
    |> Enum.map(&Transport.Jobs.ResourceValidationJob.new(%{"resource_id" => &1.id}))
    |> Oban.insert_all()

    {:ok, nil}
  end

  @doc """
  Find datasets present on the NAP for which the user is a member of the organization.
  """
  @spec datasets_for_user(Plug.Conn.t() | OAuth2.AccessToken.t()) :: [__MODULE__.t()] | {:error, OAuth2.Error.t()}
  def datasets_for_user(conn_or_token) do
    case Datagouvfr.Client.User.Wrapper.impl().me(conn_or_token) do
      {:ok, %{"organizations" => organizations}} ->
        organization_ids = Enum.map(organizations, fn %{"id" => id} -> id end)

        __MODULE__.base_query()
        |> preload(:resources)
        |> where([dataset: d], d.organization_id in ^organization_ids)
        |> Repo.all()

      error ->
        error
    end
  end

  @spec get_resources_related_files(any()) :: %{integer() => %{optional(atom()) => conversion_details() | nil}}
  def get_resources_related_files(%__MODULE__{resources: resources} = dataset) when is_list(resources) do
    target_formats = target_conversion_formats(dataset)
    # The filler's purpose is to make sure we have a {conversion_format, nil} value
    # for every resource, even if we don't have a conversion
    filler = Enum.into(available_conversion_formats(), %{}, &{&1, nil})
    resource_ids = Enum.map(resources, & &1.id)

    results =
      DB.Resource.base_query()
      |> where([resource: r], r.id in ^resource_ids)
      |> DB.ResourceHistory.join_resource_with_latest_resource_history()
      |> DB.DataConversion.join_resource_history_with_data_conversion(target_formats)
      |> select(
        [resource: r, resource_history: rh, data_conversion: dc],
        {r.id,
         {dc.convert_to,
          %{
            url: fragment("? ->> 'permanent_url'", dc.payload),
            filesize: fragment("(? ->> 'filesize')::int", dc.payload),
            # Using `fragment` to avoid `convert_to` being cast to atoms
            format: fragment("?", dc.convert_to),
            resource_history_last_up_to_date_at: rh.last_up_to_date_at
          }}}
      )
      |> Repo.all()
      # transform from
      # [{id1, {:GeoJSON, %{infos}}}, {id1, {:NeTEx, %{infos}}}, {id2, {:NeTEx, %{infos}}}]
      # to
      # %{id1 => %{:GeoJSON: %{infos}, :NeTEx: %{infos}}, id2 => %{:GeoJSON: nil, :NeTEx: %{infos}}}
      |> Enum.group_by(fn {id, _} -> id end, fn {resource_id, {format, data}} ->
        stable_url = TransportWeb.Router.Helpers.conversion_url(TransportWeb.Endpoint, :get, resource_id, format)
        {format, Map.put(data, :stable_url, stable_url)}
      end)
      |> Enum.into(%{}, fn {id, l} -> {id, Map.merge(filler, Enum.into(l, %{}))} end)

    empty_results = Enum.into(resource_ids, %{}, fn id -> {id, filler} end)

    Map.merge(empty_results, results)
  end

  def get_resources_related_files(_), do: %{}

  @doc """
  The list of conversion formats we are interested in for a dataset.
  iex> target_conversion_formats(%DB.Dataset{resources: [%DB.Resource{format: "gtfs"}, %DB.Resource{format: "NeTEx"}]})
  [:GeoJSON]
  """
  @spec target_conversion_formats(DB.Dataset.t()) :: [atom()]
  def target_conversion_formats(%__MODULE__{}) do
    available_conversion_formats()
  end

  defp available_conversion_formats, do: Ecto.Enum.values(DB.DataConversion, :convert_to)

  defp validate_siren(%Ecto.Changeset{} = changeset) do
    case get_change(changeset, :legal_owner_company_siren) do
      nil ->
        changeset

      siren ->
        if Transport.Companies.is_valid_siren?(siren) do
          changeset
        else
          add_error(changeset, :legal_owner_company_siren, "SIREN is not valid")
        end
    end
  end

  @spec validate_territory_mutual_exclusion(Ecto.Changeset.t()) :: Ecto.Changeset.t()
  defp validate_territory_mutual_exclusion(changeset) do
    has_cities =
      changeset
      |> get_field(:communes)
      |> length
      |> Kernel.min(1)

    other_fields =
      [:region_id, :aom_id]
      |> Enum.map(fn f -> get_field(changeset, f) end)
      |> Enum.count(fn f -> f not in ["", nil] end)

    fields = other_fields + has_cities

    case fields do
      1 ->
        changeset

      _ ->
        add_error(
          changeset,
          :region,
          dgettext("db-dataset", "You need to fill either aom, region or use datagouv's zone")
        )
    end
  end

  @spec validate_organization_type(Ecto.Changeset.t()) :: Ecto.Changeset.t()
  defp validate_organization_type(changeset) do
    changeset
    |> get_field(:organization_type)
    # allow a nil value for the moment
    |> Kernel.in(TransportWeb.EditDatasetLive.organization_types() ++ [nil])
    |> case do
      true -> changeset
      false -> changeset |> add_error(:organization_type, dgettext("db-dataset", "Organization type is invalid"))
    end
  end

  @spec cast_aom(Ecto.Changeset.t(), map()) :: Ecto.Changeset.t()
  defp cast_aom(changeset, %{"insee" => insee}) when insee in [nil, ""], do: change(changeset, aom_id: nil)

  defp cast_aom(changeset, %{"insee" => insee}) do
    Commune
    |> preload([:aom_res])
    |> Repo.get_by(insee: insee)
    |> case do
      nil ->
        add_error(changeset, :aom_id, dgettext("db-dataset", "Unable to find INSEE code '%{insee}'", insee: insee))

      commune ->
        case commune.aom_res do
          nil ->
            add_error(
              changeset,
              :aom_id,
              dgettext("db-dataset", "INSEE code '%{insee}' not associated with an AOM", insee: insee)
            )

          aom_res ->
            change(changeset, aom_id: aom_res.id)
        end
    end
  end

  defp cast_aom(changeset, _), do: changeset

  @spec cast_nation_dataset(Ecto.Changeset.t(), map()) :: Ecto.Changeset.t()
  defp cast_nation_dataset(changeset, %{"national_dataset" => "true"}) do
    if is_nil(get_field(changeset, :region_id)) do
      national =
        Region
        |> where([r], r.nom == "National")
        |> Repo.one!()

      put_change(changeset, :region_id, national.id)
    else
      add_error(changeset, :region, dgettext("db-dataset", "A dataset cannot be national and regional"))
    end
  end

  defp cast_nation_dataset(changeset, _), do: changeset

  @spec cast_datagouv_zone(Ecto.Changeset.t(), map(), binary()) :: Ecto.Changeset.t()
  defp cast_datagouv_zone(changeset, _, nil) do
    changeset
    |> put_assoc(:communes, [])
  end

  defp cast_datagouv_zone(changeset, _, "") do
    changeset
    |> put_assoc(:communes, [])
  end

  # We’ll only cast datagouv zone if there is something written in the associated territory name in the backoffice
  defp cast_datagouv_zone(changeset, %{"zones" => zones_insee}, _associated_territory_name) do
    communes =
      Commune
      |> where([c], c.insee in ^zones_insee)
      |> Repo.all()

    changeset
    |> put_assoc(:communes, communes)
  end

  defp maybe_overwrite_licence(%Ecto.Changeset{} = changeset) do
    custom_tags = get_field(changeset, :custom_tags) || []

    if @licence_mobilités_tag in custom_tags do
      changeset |> change(licence: "mobility-licence")
    else
      changeset
    end
  end

  defp has_real_time(changeset) do
    has_realtime = changeset |> get_field(:resources) |> Enum.any?(&Resource.real_time?/1)
    changeset |> change(has_realtime: has_realtime)
  end

  defp set_is_hidden(%Ecto.Changeset{} = changeset) do
    is_hidden =
      has_custom_tag?(%__MODULE__{custom_tags: get_field(changeset, :custom_tags)}, @hidden_dataset_custom_tag_value)

    change(changeset, is_hidden: is_hidden)
  end

  @spec set_population(Ecto.Changeset.t(), [DB.AdministrativeDivision.t()]) :: Ecto.Changeset.t()
  defp set_population(%Ecto.Changeset{} = changeset, administrative_divisions) do
    population = Enum.sum_by(administrative_divisions, & &1.population)
    change(changeset, population: population)
  end

  @spec resources_content_updated_at(__MODULE__.t()) :: map()
  def resources_content_updated_at(%__MODULE__{id: dataset_id}) do
    DB.Resource
    |> join(:left, [r], rh in DB.ResourceHistory, on: rh.resource_id == r.id)
    |> where([r], r.dataset_id == ^dataset_id)
    |> group_by([r, rh], [r.id, rh.resource_id])
    |> select([r, rh], {r.id, count(rh.id), max(fragment("payload ->>'download_datetime'"))})
    |> DB.Repo.all()
    |> Enum.map(fn {id, count, updated_at} ->
      case count do
        n when n in [0, 1] ->
          {id, nil}

        _ ->
          {:ok, datetime_updated_at, 0} = updated_at |> DateTime.from_iso8601()
          {id, datetime_updated_at}
      end
    end)
    |> Enum.into(%{})
  end

  @doc """
  Should this dataset not be historicized?

  iex> should_skip_history?(%DB.Dataset{type: "public-transit"})
  false
  iex> should_skip_history?(%DB.Dataset{type: "public-transit", custom_tags: ["skip_history", "foo"]})
  true
  """
  def should_skip_history?(%__MODULE__{type: type} = dataset) do
    type in ["vehicles-sharing"] or has_custom_tag?(dataset, "skip_history")
  end

  def has_licence_ouverte?(%__MODULE__{licence: licence}), do: licence in @licences_ouvertes

  @doc """
  iex> climate_resilience_bill?(%DB.Dataset{custom_tags: ["licence-osm"]})
  false
  iex> climate_resilience_bill?(%DB.Dataset{custom_tags: ["loi-climat-resilience", "foo"]})
  true
  """
  def climate_resilience_bill?(%__MODULE__{} = dataset), do: has_custom_tag?(dataset, "loi-climat-resilience")

  @doc """
  iex> has_custom_tag?(%DB.Dataset{custom_tags: ["foo"]}, "foo")
  true
  iex> has_custom_tag?(%DB.Dataset{custom_tags: ["foo"]}, "bar")
  false
  iex> has_custom_tag?(%DB.Dataset{custom_tags: nil}, "bar")
  false
  """
  def has_custom_tag?(%__MODULE__{custom_tags: custom_tags}, tag_name), do: tag_name in (custom_tags || [])

  @doc """
  iex> logo(%DB.Dataset{logo: "https://example.com/logo.png", custom_logo: nil})
  "https://example.com/logo.png"
  iex> logo(%DB.Dataset{logo: "https://example.com/logo.png", custom_logo: "https://example.com/custom.png"})
  "https://example.com/custom.png"
  """
  @spec logo(__MODULE__.t()) :: binary()
  def logo(%__MODULE__{logo: logo, custom_logo: custom_logo}), do: custom_logo || logo

  @doc """
  iex> full_logo(%DB.Dataset{full_logo: "https://example.com/logo.png", custom_full_logo: nil})
  "https://example.com/logo.png"
  iex> full_logo(%DB.Dataset{full_logo: "https://example.com/logo.png", custom_full_logo: "https://example.com/custom.png"})
  "https://example.com/custom.png"
  """
  @spec full_logo(__MODULE__.t()) :: binary()
  def full_logo(%__MODULE__{full_logo: full_logo, custom_full_logo: custom_full_logo}),
    do: custom_full_logo || full_logo

  @doc """
  iex> experimental?(%DB.Dataset{custom_tags: ["experimental", "foo"]})
  true
  iex> experimental?(%DB.Dataset{custom_tags: ["foo"]})
  false
  """
  def experimental?(%__MODULE__{} = dataset), do: has_custom_tag?(dataset, @experimental_tag)

  def reject_experimental_datasets(queryable) do
    queryable
    |> where([d], @experimental_tag not in d.tags)
  end
end
