Merge branch 'main' into issue/mitgliederverwaltung-420

Integrate current main (CSV import, GDPR join-form description, dependency and
tooling bumps) into the bulk-actions-dropdown feature. Gettext catalogs were
reconciled with mix gettext.extract --merge; the CHANGELOG Unreleased entries
of both sides were combined.

lib/mv/membership/custom_field_value_formatter.ex

@@ -4,13 +4,13 @@ defmodule Mv.Membership.CustomFieldValueFormatter do
 
   Same logic as the member overview Formatter but without Gettext or web helpers,
   so it can be used from the Membership context. For boolean: "Yes"/"No";
-  for date: European format (dd.mm.yyyy).
+  for date: ISO-8601 (YYYY-MM-DD) so exported values can be re-imported.
   """
   @doc """
   Formats a custom field value for plain text (e.g. CSV).
 
   Handles nil, Ash.Union, JSONB map, and direct values. Uses custom_field.value_type
-  for typing. Boolean -> "Yes"/"No", Date -> dd.mm.yyyy.
+  for typing. Boolean -> "Yes"/"No", Date -> ISO-8601 (YYYY-MM-DD).
   """
   def format_custom_field_value(nil, _custom_field), do: ""
 
@@ -18,6 +18,10 @@ defmodule Mv.Membership.CustomFieldValueFormatter do
     format_value_by_type(value, type, custom_field)
   end
 
+  def format_custom_field_value(%Date{} = value, custom_field) do
+    format_value_by_type(value, :date, custom_field)
+  end
+
   def format_custom_field_value(value, custom_field) when is_map(value) do
     type = Map.get(value, "type") || Map.get(value, "_union_type")
     val = Map.get(value, "value") || Map.get(value, "_union_value")
@@ -41,12 +45,12 @@ defmodule Mv.Membership.CustomFieldValueFormatter do
   defp format_value_by_type(value, :boolean, _), do: to_string(value)
 
   defp format_value_by_type(%Date{} = date, :date, _) do
-    Calendar.strftime(date, "%d.%m.%Y")
+    Date.to_iso8601(date)
   end
 
   defp format_value_by_type(value, :date, _) when is_binary(value) do
     case Date.from_iso8601(value) do
-      {:ok, date} -> Calendar.strftime(date, "%d.%m.%Y")
+      {:ok, date} -> Date.to_iso8601(date)
       _ -> value
     end
   end

lib/mv/membership/import/column_resolver.ex

@@ -0,0 +1,258 @@
+defmodule Mv.Membership.Import.ColumnResolver do
+  @moduledoc """
+  Read-only resolution of CSV import columns against the database.
+
+  Given the `HeaderMapper.build_maps/2` result, the raw numbered rows, and an
+  actor, `resolve/3` determines:
+
+  - which group names in the groups column already exist (`groups_found`) and
+    which would have to be created (`groups_to_create`);
+  - a small set of preview rows for the mapping preview UI.
+
+  No database writes happen here; the resolver only reads. Group creation and
+  member-group assignment happen during processing via `create_or_find_group/3`.
+
+  This module has no Phoenix or web dependencies.
+  """
+
+  require Logger
+
+  alias Mv.Membership.Import.HeaderMapper
+
+  @preview_row_limit 3
+
+  @type numbered_row :: {pos_integer(), [String.t()]}
+
+  @type resolution :: %{
+          groups_found: [%{id: String.t(), name: String.t()}],
+          groups_to_create: [String.t()],
+          fee_type_map: %{String.t() => String.t()},
+          fee_type_warnings: [String.t()],
+          has_empty_fee_type_cells?: boolean(),
+          preview_rows: [[String.t()]]
+        }
+
+  @doc """
+  Resolves the group and fee-type columns of an import against the database and
+  extracts preview rows.
+
+  Returns a map with `:groups_found`, `:groups_to_create`, `:fee_type_map`,
+  `:fee_type_warnings`, `:has_empty_fee_type_cells?`, and `:preview_rows`.
+  """
+  @spec resolve(map(), [numbered_row()], term()) :: resolution()
+  def resolve(header_maps, rows, actor) do
+    %{
+      groups_found: groups_found,
+      groups_to_create: groups_to_create
+    } = resolve_groups(header_maps, rows, actor)
+
+    %{
+      fee_type_map: fee_type_map,
+      fee_type_warnings: fee_type_warnings,
+      has_empty_fee_type_cells?: has_empty_fee_type_cells?
+    } = resolve_fee_types(header_maps, rows, actor)
+
+    %{
+      groups_found: groups_found,
+      groups_to_create: groups_to_create,
+      fee_type_map: fee_type_map,
+      fee_type_warnings: fee_type_warnings,
+      has_empty_fee_type_cells?: has_empty_fee_type_cells?,
+      preview_rows: preview_rows(rows)
+    }
+  end
+
+  defp resolve_groups(%{groups_column_index: nil}, _rows, _actor) do
+    %{groups_found: [], groups_to_create: []}
+  end
+
+  defp resolve_groups(%{groups_column_index: index}, rows, actor) do
+    existing_groups = list_groups(actor)
+    lookup = build_group_lookup(existing_groups)
+
+    names = unique_group_names(rows, index)
+
+    {found, to_create} =
+      Enum.reduce(names, {[], []}, fn name, {found, to_create} ->
+        case Map.get(lookup, normalize_name(name)) do
+          nil -> {found, [name | to_create]}
+          group -> {[%{id: group.id, name: group.name} | found], to_create}
+        end
+      end)
+
+    %{groups_found: Enum.reverse(found), groups_to_create: Enum.reverse(to_create)}
+  end
+
+  defp resolve_fee_types(%{fee_type_column_index: nil}, _rows, _actor) do
+    %{fee_type_map: %{}, fee_type_warnings: [], has_empty_fee_type_cells?: false}
+  end
+
+  defp resolve_fee_types(%{fee_type_column_index: index}, rows, actor) do
+    lookup = build_fee_type_lookup(actor)
+
+    cells = Enum.map(rows, fn {_line, values} -> Enum.at(values, index) end)
+
+    has_empty? = Enum.any?(cells, &blank?/1)
+
+    {fee_type_map, warnings} =
+      cells
+      |> Enum.reject(&blank?/1)
+      |> Enum.uniq_by(&normalize_fee_type_name/1)
+      |> Enum.reduce({%{}, []}, fn name, {map, warnings} ->
+        case Map.get(lookup, normalize_fee_type_name(name)) do
+          nil -> {map, [String.trim(name) | warnings]}
+          id -> {Map.put(map, normalize_fee_type_name(name), id), warnings}
+        end
+      end)
+
+    %{
+      fee_type_map: fee_type_map,
+      fee_type_warnings: Enum.reverse(warnings),
+      has_empty_fee_type_cells?: has_empty?
+    }
+  end
+
+  @doc """
+  Normalizes a fee-type name using the same rules as CSV header normalization
+  (trim, lowercase, transliterate, drop hyphens and whitespace).
+  """
+  @spec normalize_fee_type_name(String.t() | nil) :: String.t()
+  def normalize_fee_type_name(name) when is_binary(name), do: HeaderMapper.normalize_header(name)
+  def normalize_fee_type_name(_), do: ""
+
+  defp build_fee_type_lookup(actor) do
+    actor
+    |> list_fee_types()
+    |> Enum.reduce(%{}, fn fee_type, acc ->
+      normalized = normalize_fee_type_name(fee_type.name)
+
+      if Map.has_key?(acc, normalized) do
+        Logger.warning(
+          "Multiple membership fee types normalize to #{inspect(normalized)}; using the first match for CSV import."
+        )
+
+        acc
+      else
+        Map.put(acc, normalized, fee_type.id)
+      end
+    end)
+  end
+
+  defp list_fee_types(actor) do
+    Mv.MembershipFees.list_membership_fee_types!(actor: actor)
+  end
+
+  defp blank?(nil), do: true
+  defp blank?(value) when is_binary(value), do: String.trim(value) == ""
+  defp blank?(_), do: false
+
+  @doc """
+  Finds an existing group by name (case-insensitive) or creates it.
+
+  Looks first in the pre-fetched `groups` list, then in the database (to catch
+  groups created earlier in the same import), and only creates a new group when
+  none is found. This keeps group resolution idempotent across re-imports.
+  """
+  @spec create_or_find_group(String.t(), [Mv.Membership.Group.t()], term()) ::
+          {:ok, Mv.Membership.Group.t()} | {:error, term()}
+  def create_or_find_group(name, groups, actor) when is_binary(name) do
+    trimmed = String.trim(name)
+    normalized = normalize_name(trimmed)
+
+    case find_group_in_list(groups, normalized) do
+      nil -> find_or_create_group(trimmed, normalized, actor)
+      group -> {:ok, group}
+    end
+  end
+
+  defp find_group_in_list(groups, normalized) do
+    Enum.find(groups, fn group -> normalize_name(group.name) == normalized end)
+  end
+
+  defp find_or_create_group(trimmed, normalized, actor) do
+    case fetch_group_by_normalized_name(normalized, actor) do
+      nil -> create_group(trimmed, normalized, actor)
+      group -> {:ok, group}
+    end
+  end
+
+  # Normalizes the Ash code-interface return to a two-shape result.
+  #
+  # On a create failure the group may have been created concurrently by another
+  # import session between our read and our write (the DB unique index is the
+  # final arbiter, and the name validation is fail-open). Re-fetch by normalized
+  # name and link to the existing group rather than failing the row.
+  defp create_group(name, normalized, actor) do
+    case Mv.Membership.create_group(%{name: name}, actor: actor) do
+      {:ok, %Mv.Membership.Group{} = group} ->
+        {:ok, group}
+
+      {:error, reason} ->
+        case fetch_group_by_normalized_name(normalized, actor) do
+          nil -> {:error, reason}
+          group -> {:ok, group}
+        end
+    end
+  end
+
+  # Fetches a single group by case-insensitive name using a name-filtered query
+  # rather than reading the whole groups table. `normalized` is the trimmed,
+  # lower-cased name; the DB comparison uses LOWER(name) consistent with the
+  # Group resource's case-insensitive uniqueness constraint.
+  defp fetch_group_by_normalized_name(normalized, actor) do
+    require Ash.Query
+
+    Mv.Membership.Group
+    |> Ash.Query.filter(fragment("LOWER(?) = ?", name, ^normalized))
+    |> Ash.read(actor: actor, domain: Mv.Membership)
+    |> case do
+      {:ok, [group | _]} -> group
+      _ -> nil
+    end
+  end
+
+  @doc """
+  Splits a raw groups-cell value into trimmed, non-empty group names.
+  """
+  @spec split_group_names(String.t() | nil) :: [String.t()]
+  def split_group_names(nil), do: []
+
+  def split_group_names(cell) when is_binary(cell) do
+    cell
+    |> String.split(",")
+    |> Enum.map(&String.trim/1)
+    |> Enum.reject(&(&1 == ""))
+  end
+
+  defp unique_group_names(rows, index) do
+    rows
+    |> Enum.flat_map(fn {_line, values} ->
+      values
+      |> Enum.at(index)
+      |> split_group_names()
+    end)
+    |> Enum.uniq_by(&normalize_name/1)
+  end
+
+  defp preview_rows(rows) do
+    rows
+    |> Enum.take(@preview_row_limit)
+    |> Enum.map(fn {_line, values} -> values end)
+  end
+
+  defp list_groups(actor) do
+    Mv.Membership.list_groups!(actor: actor)
+  end
+
+  defp build_group_lookup(groups) do
+    Enum.reduce(groups, %{}, fn group, acc ->
+      Map.put(acc, normalize_name(group.name), group)
+    end)
+  end
+
+  # Case-insensitive comparison consistent with the Group resource's
+  # case-insensitive name uniqueness.
+  defp normalize_name(name) when is_binary(name) do
+    name |> String.trim() |> String.downcase()
+  end
+end

lib/mv/membership/import/header_mapper.ex

@@ -29,12 +29,21 @@ defmodule Mv.Membership.Import.HeaderMapper do
 
   Supports English and German header variants (e.g. "Email" / "E-Mail", "Join Date" / "Beitrittsdatum").
 
+  ## Special columns
+
+  - **groups** – Many-to-many relationship (through member_groups). Recognized via the
+    `groups_column_index` key (headers `Groups`, `Gruppen`, `Gruppe`). Comma-separated
+    names are resolved during processing; missing groups are auto-created.
+  - **membership_fee_type** – Recognized via the `fee_type_column_index` key (headers
+    `Fee Type`, `fee_type`, `membership_fee_type`, `Beitragsart`). Names are matched to
+    existing fee types; unknown names fall back to the default fee type.
+
   ## Fields not supported for import
 
   - **membership_fee_status** – Computed (calculation from membership fee cycles). Not stored;
-    cannot be set via CSV. Export can include it.
-  - **groups** – Many-to-many relationship (through member_groups). Import would require
-    resolving group names/slugs to IDs and creating associations; not in current import scope.
+    cannot be set via CSV. Export can include it. Fee-status header variants
+    (`Membership Fee Status`, `Bezahlstatus`, `Mitgliedsbeitragsstatus`) are explicitly
+    placed in the `ignored` list and never mapped.
 
   ## Custom Field Detection
 
@@ -47,10 +56,10 @@ defmodule Mv.Membership.Import.HeaderMapper do
       "e-mail"
 
       iex> HeaderMapper.build_maps(["Email", "First Name"], [])
-      {:ok, %{member: %{email: 0, first_name: 1}, custom: %{}, unknown: []}}
+      {:ok, %{member: %{email: 0, first_name: 1}, custom: %{}, unknown: [], ignored: [], groups_column_index: nil, fee_type_column_index: nil}}
 
       iex> HeaderMapper.build_maps(["Email", "CustomField"], [%{id: "cf1", name: "CustomField"}])
-      {:ok, %{member: %{email: 0}, custom: %{"cf1" => 1}, unknown: []}}
+      {:ok, %{member: %{email: 0}, custom: %{"cf1" => 1}, unknown: [], ignored: [], groups_column_index: nil, fee_type_column_index: nil}}
   """
 
   @type column_map :: %{atom() => non_neg_integer()}
@@ -60,6 +69,33 @@ defmodule Mv.Membership.Import.HeaderMapper do
   # Required member fields
   @required_member_fields [:email]
 
+  # Fee-status header variants that must never be imported (computed/read-only field).
+  # Stored already-normalized; checked before member, custom, groups, and fee-type mapping.
+  # Maintain this list when new locale translations for fee-status are added.
+  @ignored_normalized [
+    "membershipfeestatus",
+    "mitgliedsbeitragsstatus",
+    "bezahlstatus",
+    # DE export label for membership_fee_start_date — system-managed, not importable
+    "startdatummitgliedsbeitrag"
+  ]
+
+  # Normalized header variants for the groups column. The column is resolved to
+  # group associations during import; it is never a member or custom field.
+  @groups_column_normalized [
+    "groups",
+    "gruppen",
+    "gruppe"
+  ]
+
+  # Normalized header variants for the membership fee-type column. The column is
+  # resolved to a MembershipFeeType during import; it is never a member or custom field.
+  @fee_type_column_normalized [
+    "membershipfeetype",
+    "feetype",
+    "beitragsart"
+  ]
+
   # Canonical member fields with their raw variants
   # These will be normalized at runtime when building the lookup map
   @member_field_variants_raw %{
@@ -239,30 +275,79 @@ defmodule Mv.Membership.Import.HeaderMapper do
 
   ## Returns
 
-  - `{:ok, %{member: column_map, custom: custom_field_map, unknown: unknown_headers}}` on success
+  - `{:ok, %{member: column_map, custom: custom_field_map, unknown: unknown_headers,
+    ignored: [non_neg_integer], groups_column_index: non_neg_integer | nil,
+    fee_type_column_index: non_neg_integer | nil}}` on success
   - `{:error, reason}` on error (missing required field, duplicate headers)
 
+  The `ignored` list holds the indices of fee-status columns (computed/read-only),
+  which are never mapped to member or custom fields.
+
   ## Examples
 
       iex> build_maps(["Email", "First Name"], [])
-      {:ok, %{member: %{email: 0, first_name: 1}, custom: %{}, unknown: []}}
+      {:ok, %{member: %{email: 0, first_name: 1}, custom: %{}, unknown: [], ignored: [], groups_column_index: nil, fee_type_column_index: nil}}
 
       iex> build_maps(["Email", "CustomField"], [%{id: "cf1", name: "CustomField"}])
-      {:ok, %{member: %{email: 0}, custom: %{"cf1" => 1}, unknown: []}}
+      {:ok, %{member: %{email: 0}, custom: %{"cf1" => 1}, unknown: [], ignored: [], groups_column_index: nil, fee_type_column_index: nil}}
 
   """
   @spec build_maps([String.t()], [map()]) ::
-          {:ok, %{member: column_map(), custom: custom_field_map(), unknown: unknown_headers()}}
+          {:ok,
+           %{
+             member: column_map(),
+             custom: custom_field_map(),
+             unknown: unknown_headers(),
+             ignored: [non_neg_integer()],
+             groups_column_index: non_neg_integer() | nil,
+             fee_type_column_index: non_neg_integer() | nil
+           }}
           | {:error, String.t()}
   def build_maps(headers, custom_fields) when is_list(headers) and is_list(custom_fields) do
-    with {:ok, member_map, unknown_after_member} <- build_member_map(headers),
+    ignored = ignored_indices(headers)
+    groups_column_index = first_matching_index(headers, @groups_column_normalized)
+    fee_type_column_index = first_matching_index(headers, @fee_type_column_normalized)
+
+    reserved =
+      [groups_column_index, fee_type_column_index | ignored]
+      |> Enum.reject(&is_nil/1)
+      |> MapSet.new()
+
+    with {:ok, member_map, unknown_after_member} <- build_member_map(headers, reserved),
          {:ok, custom_map, unknown_after_custom} <-
            build_custom_field_map(headers, unknown_after_member, custom_fields, member_map) do
       unknown = Enum.map(unknown_after_custom, &Enum.at(headers, &1))
-      {:ok, %{member: member_map, custom: custom_map, unknown: unknown}}
+
+      {:ok,
+       %{
+         member: member_map,
+         custom: custom_map,
+         unknown: unknown,
+         ignored: ignored,
+         groups_column_index: groups_column_index,
+         fee_type_column_index: fee_type_column_index
+       }}
     end
   end
 
+  # Returns the index of the first header whose normalized form is in `variants`,
+  # or nil if none match.
+  defp first_matching_index(headers, variants) do
+    headers
+    |> Enum.with_index()
+    |> Enum.find_value(fn {header, index} ->
+      if normalize_header(header) in variants, do: index
+    end)
+  end
+
+  # Returns the column indices whose normalized header is in the fee-status ignore list.
+  defp ignored_indices(headers) do
+    headers
+    |> Enum.with_index()
+    |> Enum.filter(fn {header, _index} -> normalize_header(header) in @ignored_normalized end)
+    |> Enum.map(fn {_header, index} -> index end)
+  end
+
   # --- Private Functions ---
 
   # Transliterates German umlauts and special characters
@@ -304,13 +389,14 @@ defmodule Mv.Membership.Import.HeaderMapper do
     |> String.replace(" ", "")
   end
 
-  # Builds member field column map
-  defp build_member_map(headers) do
+  # Builds member field column map, skipping reserved (e.g. ignored) indices.
+  defp build_member_map(headers, reserved) do
     result =
       headers
       |> Enum.with_index()
       |> Enum.reduce_while({%{}, []}, fn {header, index}, {acc_map, acc_unknown} ->
-        normalized = normalize_header(header)
+        normalized =
+          if MapSet.member?(reserved, index), do: "", else: normalize_header(header)
 
         case process_member_header(header, index, normalized, acc_map, %{}) do
           {:error, reason} ->

lib/mv/membership/import/import_runner.ex

@@ -80,7 +80,7 @@ defmodule Mv.Membership.Import.ImportRunner do
     all_errors = progress.errors ++ chunk_result.errors
     new_errors = Enum.take(all_errors, max_errors)
     errors_truncated? = length(all_errors) > max_errors
-    new_warnings = progress.warnings ++ Map.get(chunk_result, :warnings, [])
+    new_warnings = Enum.uniq(progress.warnings ++ Map.get(chunk_result, :warnings, []))
 
     chunks_processed = current_chunk_idx + 1
     new_status = if chunks_processed >= progress.total_chunks, do: :done, else: :running
@@ -97,6 +97,20 @@ defmodule Mv.Membership.Import.ImportRunner do
     }
   end
 
+  @doc """
+  Carries the in-memory group snapshot grown by a chunk back into `import_state`
+  so the next chunk reuses groups created earlier instead of re-reading the
+  Group table. When the chunk result omits `groups_found`, the state is returned
+  unchanged.
+  """
+  @spec carry_groups_forward(map(), map()) :: map()
+  def carry_groups_forward(import_state, chunk_result) do
+    case Map.fetch(chunk_result, :groups_found) do
+      {:ok, groups_found} -> Map.put(import_state, :groups_found, groups_found)
+      :error -> import_state
+    end
+  end
+
   @doc """
   Returns the next action after processing a chunk: send the next chunk index or done.
   """

lib/mv/membership/import/member_csv.ex

@@ -6,7 +6,7 @@ defmodule Mv.Membership.Import.MemberCSV do
 
   This module provides the core API for CSV member import functionality:
   - `prepare/2` - Parses and validates CSV content, returns import state
-  - `process_chunk/3` - Processes a chunk of rows and creates members
+  - `process_chunk/4` - Processes a chunk of rows and creates members
 
   ## Error Handling
 
@@ -22,13 +22,24 @@ defmodule Mv.Membership.Import.MemberCSV do
   - `column_map` - Map of canonical field names to column indices
   - `custom_field_map` - Map of custom field names to column indices
   - `warnings` - List of warning messages (e.g., unknown custom field columns)
+  - `headers` - The raw CSV header row
+  - `ignored` - Header names of ignored (fee-status) columns
+  - `groups_column_index` / `fee_type_column_index` - Indices for resolved columns (or nil)
+  - `groups_found` / `groups_to_create` - Existing and to-be-created groups from the preview
+  - `fee_type_map` - Normalized fee-type name to id, for matched fee types
+  - `fee_type_warnings` - Unmatched fee-type names surfaced in the preview
+  - `has_empty_fee_type_cells?` - Whether any fee-type cell is blank (default applies)
+  - `preview_rows` - Up to 3 sample data rows for the mapping preview
 
   ## Chunk Results
 
-  The `chunk_result` returned by `process_chunk/3` contains:
+  The `chunk_result` returned by `process_chunk/4` contains:
   - `inserted` - Number of successfully created members
   - `failed` - Number of failed member creations
   - `errors` - List of `%MemberCSV.Error{}` structs (capped at 50 per import)
+  - `groups_found` - The in-memory group snapshot grown while processing this
+    chunk; thread it into the next chunk's `:groups_found` opt so groups created
+    in an earlier chunk are reused without re-reading the Group table
 
   ## Examples
 
@@ -37,7 +48,9 @@ defmodule Mv.Membership.Import.MemberCSV do
 
       # Process first chunk
       chunk = Enum.at(import_state.chunks, 0)
-      {:ok, result} = MemberCSV.process_chunk(chunk, import_state.column_map)
+
+      {:ok, result} =
+        MemberCSV.process_chunk(chunk, import_state.column_map, import_state.custom_field_map, [])
   """
 
   defmodule Error do
@@ -66,16 +79,29 @@ defmodule Mv.Membership.Import.MemberCSV do
           custom_field_lookup: %{
             String.t() => %{id: String.t(), value_type: atom(), name: String.t()}
           },
-          warnings: list(String.t())
+          warnings: list(String.t()),
+          headers: list(String.t()),
+          ignored: list(String.t()),
+          groups_column_index: non_neg_integer() | nil,
+          fee_type_column_index: non_neg_integer() | nil,
+          groups_found: list(%{id: String.t(), name: String.t()}),
+          groups_to_create: list(String.t()),
+          fee_type_map: %{String.t() => String.t()},
+          fee_type_warnings: list(String.t()),
+          has_empty_fee_type_cells?: boolean(),
+          preview_rows: list(list(String.t()))
         }
 
   @type chunk_result :: %{
           inserted: non_neg_integer(),
           failed: non_neg_integer(),
           errors: list(Error.t()),
-          errors_truncated?: boolean()
+          errors_truncated?: boolean(),
+          warnings: list(String.t()),
+          groups_found: list(Mv.Membership.Group.t() | %{id: String.t(), name: String.t()})
         }
 
+  alias Mv.Membership.Import.ColumnResolver
   alias Mv.Membership.Import.CsvParser
   alias Mv.Membership.Import.HeaderMapper
 
@@ -139,13 +165,27 @@ defmodule Mv.Membership.Import.MemberCSV do
       # Build custom field lookup for efficient value processing
       custom_field_lookup = build_custom_field_lookup(custom_fields)
 
+      # Resolve DB-backed columns (groups, fee types) read-only for the preview.
+      resolution = ColumnResolver.resolve(maps, rows, actor)
+      ignored_headers = Enum.map(maps.ignored, &Enum.at(headers, &1))
+
       {:ok,
        %{
          chunks: chunks,
          column_map: maps.member,
          custom_field_map: maps.custom,
          custom_field_lookup: custom_field_lookup,
-         warnings: warnings
+         warnings: warnings,
+         headers: headers,
+         ignored: ignored_headers,
+         groups_column_index: maps.groups_column_index,
+         fee_type_column_index: maps.fee_type_column_index,
+         groups_found: resolution.groups_found,
+         groups_to_create: resolution.groups_to_create,
+         fee_type_map: resolution.fee_type_map,
+         fee_type_warnings: resolution.fee_type_warnings,
+         has_empty_fee_type_cells?: resolution.has_empty_fee_type_cells?,
+         preview_rows: resolution.preview_rows
        }}
     end
   end
@@ -180,7 +220,7 @@ defmodule Mv.Membership.Import.MemberCSV do
       end)
 
     case HeaderMapper.build_maps(headers, custom_field_maps) do
-      {:ok, %{member: member_map, custom: custom_map, unknown: unknown}} ->
+      {:ok, %{unknown: unknown} = maps} ->
         # Build warnings for unknown custom field columns
         warnings =
           unknown
@@ -197,7 +237,7 @@ defmodule Mv.Membership.Import.MemberCSV do
             )
           end)
 
-        {:ok, %{member: member_map, custom: custom_map}, warnings}
+        {:ok, maps, warnings}
 
       {:error, reason} ->
         {:error, reason}
@@ -250,9 +290,20 @@ defmodule Mv.Membership.Import.MemberCSV do
         Map.put(acc, custom_field_id, value)
       end)
 
-    %{member: member_map, custom: custom_map}
+    %{
+      member: member_map,
+      custom: custom_map,
+      fee_type: cell_at(row_tuple, tuple_size, maps.fee_type_column_index),
+      groups: cell_at(row_tuple, tuple_size, maps.groups_column_index)
+    }
   end
 
+  # Returns the raw cell at the given index, or nil if the column is absent.
+  defp cell_at(_row_tuple, _size, nil), do: nil
+
+  defp cell_at(row_tuple, size, index) when index < size, do: elem(row_tuple, index)
+  defp cell_at(_row_tuple, _size, _index), do: ""
+
   @doc """
   Processes a chunk of CSV rows and creates members.
 
@@ -268,12 +319,18 @@ defmodule Mv.Membership.Import.MemberCSV do
   - `chunk_rows_with_lines` - List of tuples `{csv_line_number, row_map}` where:
     - `csv_line_number` - Physical line number in CSV (1-based)
     - `row_map` - Map with `:member` and `:custom` keys containing field values
-  - `column_map` - Map of canonical field names (atoms) to column indices (for reference)
-  - `custom_field_map` - Map of custom field IDs (strings) to column indices (for reference)
+  - `column_map` - Unused; kept for backward-compatible call sites. Field values are
+    read from each row's pre-built `:member`/`:custom` maps, not from this argument.
+  - `custom_field_map` - Unused; kept for backward-compatible call sites (see above).
   - `opts` - Optional keyword list for processing options:
     - `:custom_field_lookup` - Map of custom field IDs to metadata (default: `%{}`)
     - `:existing_error_count` - Number of errors already collected in previous chunks (default: `0`)
     - `:max_errors` - Maximum number of errors to collect per import overall (default: `50`)
+    - `:actor` - Actor used for all writes (default: the system actor)
+    - `:fee_type_map` - Map of normalized fee-type name to fee-type id, used to resolve
+      each row's fee-type cell (default: `%{}`)
+    - `:groups_found` - List of pre-fetched `Group` structs seeding in-memory group
+      resolution; the snapshot grows as groups are auto-created (default: `[]`)
 
   ## Error Capping
 
@@ -312,27 +369,49 @@ defmodule Mv.Membership.Import.MemberCSV do
     existing_error_count = Keyword.get(opts, :existing_error_count, 0)
     max_errors = Keyword.get(opts, :max_errors, @default_max_errors)
     actor = Keyword.get(opts, :actor, SystemActor.get_system_actor())
+    fee_type_map = Keyword.get(opts, :fee_type_map, %{})
+    groups_found = Keyword.get(opts, :groups_found, [])
 
-    {inserted, failed, errors, _collected_error_count, truncated?} =
-      Enum.reduce(chunk_rows_with_lines, {0, 0, [], 0, false}, fn {line_number, row_map},
-                                                                  {acc_inserted, acc_failed,
-                                                                   acc_errors, acc_error_count,
-                                                                   acc_truncated?} ->
+    base_row_opts = %{
+      custom_field_lookup: custom_field_lookup,
+      fee_type_map: fee_type_map,
+      actor: actor
+    }
+
+    {inserted, failed, errors, _collected_error_count, truncated?, warnings, groups_acc} =
+      Enum.reduce(chunk_rows_with_lines, {0, 0, [], 0, false, [], groups_found}, fn {line_number,
+                                                                                     row_map},
+                                                                                    {acc_inserted,
+                                                                                     acc_failed,
+                                                                                     acc_errors,
+                                                                                     acc_error_count,
+                                                                                     acc_truncated?,
+                                                                                     acc_warnings,
+                                                                                     acc_groups} ->
         current_error_count = existing_error_count + acc_error_count
+        row_opts = Map.put(base_row_opts, :groups_found, acc_groups)
 
-        case process_row(row_map, line_number, custom_field_lookup, actor) do
-          {:ok, _member} ->
-            update_inserted(
-              {acc_inserted, acc_failed, acc_errors, acc_error_count, acc_truncated?}
-            )
+        case process_row(row_map, line_number, row_opts) do
+          {:ok, _member, row_warnings, new_groups} ->
+            {new_inserted, new_failed, new_errors, new_error_count, new_truncated?} =
+              update_inserted(
+                {acc_inserted, acc_failed, acc_errors, acc_error_count, acc_truncated?}
+              )
 
-          {:error, error} ->
-            handle_row_error(
-              {acc_inserted, acc_failed, acc_errors, acc_error_count, acc_truncated?},
-              error,
-              current_error_count,
-              max_errors
-            )
+            {new_inserted, new_failed, new_errors, new_error_count, new_truncated?,
+             acc_warnings ++ row_warnings, new_groups}
+
+          {:error, error, new_groups} ->
+            {new_inserted, new_failed, new_errors, new_error_count, new_truncated?} =
+              handle_row_error(
+                {acc_inserted, acc_failed, acc_errors, acc_error_count, acc_truncated?},
+                error,
+                current_error_count,
+                max_errors
+              )
+
+            {new_inserted, new_failed, new_errors, new_error_count, new_truncated?, acc_warnings,
+             new_groups}
         end
       end)
 
@@ -341,7 +420,9 @@ defmodule Mv.Membership.Import.MemberCSV do
        inserted: inserted,
        failed: failed,
        errors: Enum.reverse(errors),
-       errors_truncated?: truncated?
+       errors_truncated?: truncated?,
+       warnings: warnings,
+       groups_found: groups_acc
      }}
   end
 
@@ -505,18 +586,27 @@ defmodule Mv.Membership.Import.MemberCSV do
 
   defp gettext_error_message(_), do: gettext("Email is invalid.")
 
-  # Processes a single row and creates member with custom field values
+  # Processes a single row and creates member with custom field values.
+  # On success returns {:ok, member, warnings, groups}; warnings carry non-fatal
+  # notices such as an unresolved fee-type name. The returned groups list is the
+  # accumulated in-memory group snapshot (seeded from the chunk, grown with any
+  # group created while linking this row) so later rows reuse it instead of
+  # re-reading the whole Group table per row.
   defp process_row(
          row_map,
          line_number,
-         custom_field_lookup,
-         actor
+         %{
+           custom_field_lookup: custom_field_lookup,
+           fee_type_map: fee_type_map,
+           groups_found: groups_found,
+           actor: actor
+         } = _row_opts
        ) do
     # Validate row before database insertion
     case validate_row(row_map, line_number, []) do
       {:error, error} ->
         # Return validation error immediately, no DB insert attempted
-        {:error, error}
+        {:error, error, groups_found}
 
       {:ok, %{member: trimmed_member_attrs, custom: custom_attrs}} ->
         # Prepare custom field values for Ash
@@ -524,20 +614,119 @@ defmodule Mv.Membership.Import.MemberCSV do
           {:error, validation_errors} ->
             # Custom field validation errors - return first error
             first_error = List.first(validation_errors)
-            {:error, %Error{csv_line_number: line_number, field: nil, message: first_error}}
+
+            {:error, %Error{csv_line_number: line_number, field: nil, message: first_error},
+             groups_found}
 
           {:ok, custom_field_values} ->
-            create_member_with_custom_fields(
-              trimmed_member_attrs,
+            {fee_attrs, warnings} =
+              resolve_fee_type_attrs(Map.get(row_map, :fee_type), fee_type_map)
+
+            create_member_and_assign_groups(
+              Map.merge(trimmed_member_attrs, fee_attrs),
               custom_field_values,
+              Map.get(row_map, :groups),
+              groups_found,
               line_number,
-              actor
+              actor,
+              warnings
             )
         end
     end
   rescue
     e ->
-      {:error, %Error{csv_line_number: line_number, field: nil, message: Exception.message(e)}}
+      {:error, %Error{csv_line_number: line_number, field: nil, message: Exception.message(e)},
+       groups_found}
+  end
+
+  # Creates the member, then assigns groups as a post-creation step. A group
+  # assignment failure fails the row (the member was already created, but the
+  # row is reported as failed so the operator can act on it).
+  defp create_member_and_assign_groups(
+         member_attrs,
+         custom_field_values,
+         groups_cell,
+         groups_found,
+         line_number,
+         actor,
+         warnings
+       ) do
+    case create_member_with_custom_fields(
+           member_attrs,
+           custom_field_values,
+           line_number,
+           actor,
+           warnings
+         ) do
+      {:ok, member, member_warnings} ->
+        assign_groups(member, groups_cell, groups_found, line_number, actor, member_warnings)
+
+      {:error, error} ->
+        {:error, error, groups_found}
+    end
+  end
+
+  # Assigns the member to all groups listed in the cell, creating missing groups.
+  # Returns the (possibly grown) group snapshot so the caller can reuse it.
+  defp assign_groups(member, groups_cell, groups_found, line_number, actor, warnings) do
+    names = ColumnResolver.split_group_names(groups_cell)
+
+    Enum.reduce_while(names, {:ok, member, warnings, groups_found}, fn name,
+                                                                       {:ok, _m, _w, acc_groups} ->
+      case link_member_to_group(member, name, acc_groups, actor) do
+        {:ok, group} ->
+          {:cont, {:ok, member, warnings, add_group(acc_groups, group)}}
+
+        {:error, reason} ->
+          {:halt,
+           {:error,
+            %Error{
+              csv_line_number: line_number,
+              field: nil,
+              message: gettext("Group assignment failed: %{reason}", reason: inspect(reason))
+            }, acc_groups}}
+      end
+    end)
+  end
+
+  defp add_group(groups, group) do
+    if Enum.any?(groups, &(&1.id == group.id)), do: groups, else: [group | groups]
+  end
+
+  defp link_member_to_group(member, name, groups_found, actor) do
+    with {:ok, group} <- ColumnResolver.create_or_find_group(name, groups_found, actor),
+         {:ok, _member_group} <-
+           Mv.Membership.create_member_group(
+             %{member_id: member.id, group_id: group.id},
+             actor: actor
+           ) do
+      {:ok, group}
+    end
+  end
+
+  # Resolves the fee-type cell into member attrs plus optional warnings.
+  # Empty cell -> default fee type (SetDefaultMembershipFeeType), no warning.
+  # Matched name -> membership_fee_type_id attr.
+  # Unmatched name -> no attr (default applies), warning naming the value.
+  defp resolve_fee_type_attrs(nil, _fee_type_map), do: {%{}, []}
+
+  defp resolve_fee_type_attrs(cell, fee_type_map) when is_binary(cell) do
+    trimmed = String.trim(cell)
+
+    if trimmed == "" do
+      {%{}, []}
+    else
+      case Map.get(fee_type_map, ColumnResolver.normalize_fee_type_name(trimmed)) do
+        nil ->
+          {%{},
+           [
+             gettext("Fee type '%{name}' not found; using the default fee type.", name: trimmed)
+           ]}
+
+        fee_type_id ->
+          {%{membership_fee_type_id: fee_type_id}, []}
+      end
+    end
   end
 
   # Creates a member with custom field values, handling errors appropriately
@@ -545,7 +734,8 @@ defmodule Mv.Membership.Import.MemberCSV do
          trimmed_member_attrs,
          custom_field_values,
          line_number,
-         actor
+         actor,
+         warnings
        ) do
     # Convert empty strings to nil for date fields so Ash accepts them
     member_attrs = sanitize_date_fields(trimmed_member_attrs)
@@ -565,7 +755,7 @@ defmodule Mv.Membership.Import.MemberCSV do
 
     case Mv.Membership.create_member(final_attrs, actor: actor) do
       {:ok, member} ->
-        {:ok, member}
+        {:ok, member, warnings}
 
       {:error, %Ash.Error.Invalid{} = error} ->
         # Extract email from final_attrs for better error messages