diff --git a/lib/mv/membership/import/header_mapper.ex b/lib/mv/membership/import/header_mapper.ex
index 3047944..eb8bb04 100644
--- a/lib/mv/membership/import/header_mapper.ex
+++ b/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
 
diff --git a/lib/mv/membership/import/member_csv.ex b/lib/mv/membership/import/member_csv.ex
index dda1d04..d0ab74b 100644
--- a/lib/mv/membership/import/member_csv.ex
+++ b/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,10 +22,18 @@ 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)
@@ -37,7 +45,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 +76,28 @@ 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())
         }
 
+  alias Mv.Membership.Import.ColumnResolver
   alias Mv.Membership.Import.CsvParser
   alias Mv.Membership.Import.HeaderMapper
 
@@ -139,13 +161,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 +216,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 +233,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 +286,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 +315,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 +365,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 +416,8 @@ defmodule Mv.Membership.Import.MemberCSV do
        inserted: inserted,
        failed: failed,
        errors: Enum.reverse(errors),
-       errors_truncated?: truncated?
+       errors_truncated?: truncated?,
+       warnings: warnings
      }}
   end
 
@@ -505,18 +581,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 +609,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 +729,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 +750,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
diff --git a/priv/gettext/de/LC_MESSAGES/default.po b/priv/gettext/de/LC_MESSAGES/default.po
index 81d91f7..9fa6cd4 100644
--- a/priv/gettext/de/LC_MESSAGES/default.po
+++ b/priv/gettext/de/LC_MESSAGES/default.po
@@ -3968,7 +3968,12 @@ msgstr "Zeitraum"
 msgid "To"
 msgstr "Bis"
 
-#~ #: lib/mv_web/live/group_live/show.ex
-#~ #, elixir-autogen, elixir-format
-#~ msgid "No members selected."
-#~ msgstr "Keine Mitglieder ausgewählt."
+#: lib/mv/membership/import/member_csv.ex
+#, elixir-autogen, elixir-format
+msgid "Fee type '%{name}' not found; using the default fee type."
+msgstr "Beitragsart '%{name}' nicht gefunden; Standard-Beitragsart wird verwendet."
+
+#: lib/mv/membership/import/member_csv.ex
+#, elixir-autogen, elixir-format
+msgid "Group assignment failed: %{reason}"
+msgstr "Gruppenzuordnung fehlgeschlagen: %{reason}"
diff --git a/priv/gettext/default.pot b/priv/gettext/default.pot
index 5e9abca..c961420 100644
--- a/priv/gettext/default.pot
+++ b/priv/gettext/default.pot
@@ -3967,3 +3967,13 @@ msgstr ""
 #, elixir-autogen, elixir-format
 msgid "To"
 msgstr ""
+
+#: lib/mv/membership/import/member_csv.ex
+#, elixir-autogen, elixir-format
+msgid "Fee type '%{name}' not found; using the default fee type."
+msgstr ""
+
+#: lib/mv/membership/import/member_csv.ex
+#, elixir-autogen, elixir-format
+msgid "Group assignment failed: %{reason}"
+msgstr ""
diff --git a/priv/gettext/en/LC_MESSAGES/default.po b/priv/gettext/en/LC_MESSAGES/default.po
index 1ae6a49..58aeead 100644
--- a/priv/gettext/en/LC_MESSAGES/default.po
+++ b/priv/gettext/en/LC_MESSAGES/default.po
@@ -3968,7 +3968,12 @@ msgstr ""
 msgid "To"
 msgstr ""
 
-#~ #: lib/mv_web/live/group_live/show.ex
-#~ #, elixir-autogen, elixir-format, fuzzy
-#~ msgid "No members selected."
-#~ msgstr ""
+#: lib/mv/membership/import/member_csv.ex
+#, elixir-autogen, elixir-format
+msgid "Fee type '%{name}' not found; using the default fee type."
+msgstr ""
+
+#: lib/mv/membership/import/member_csv.ex
+#, elixir-autogen, elixir-format
+msgid "Group assignment failed: %{reason}"
+msgstr ""
diff --git a/test/mv/membership/import/member_csv_test.exs b/test/mv/membership/import/member_csv_test.exs
index b4a099a..0701a92 100644
--- a/test/mv/membership/import/member_csv_test.exs
+++ b/test/mv/membership/import/member_csv_test.exs
@@ -1,5 +1,6 @@
 defmodule Mv.Membership.Import.MemberCSVTest do
   use Mv.DataCase, async: true
+  use ExUnitProperties
 
   alias Mv.Membership.Import.MemberCSV
 
@@ -899,4 +900,255 @@ defmodule Mv.Membership.Import.MemberCSVTest do
       assert import_state.chunks != []
     end
   end
+
+  describe "prepare/2 column resolution integration" do
+    setup do
+      %{actor: Mv.Helpers.SystemActor.get_system_actor()}
+    end
+
+    test "exposes resolver output keys in import_state", %{actor: actor} do
+      csv_content = "email\njohn@example.com"
+
+      assert {:ok, import_state} = MemberCSV.prepare(csv_content, actor: actor)
+
+      assert Map.has_key?(import_state, :ignored)
+      assert Map.has_key?(import_state, :groups_to_create)
+      assert Map.has_key?(import_state, :fee_type_map)
+      assert Map.has_key?(import_state, :fee_type_warnings)
+      assert Map.has_key?(import_state, :has_empty_fee_type_cells?)
+      assert Map.has_key?(import_state, :preview_rows)
+    end
+
+    test "fee-status column is reported as ignored, not as a custom field", %{actor: actor} do
+      {:ok, _custom_field} =
+        Mv.Membership.CustomField
+        |> Ash.Changeset.for_create(:create, %{name: "Bezahlstatus", value_type: :string})
+        |> Ash.create(actor: actor)
+
+      csv_content = "email;Bezahlstatus\njohn@example.com;paid"
+
+      assert {:ok, import_state} = MemberCSV.prepare(csv_content, actor: actor)
+
+      assert import_state.ignored == ["Bezahlstatus"]
+      assert import_state.custom_field_map == %{}
+    end
+
+    test "preview rows are limited to 3", %{actor: actor} do
+      csv_content = "email\na@example.com\nb@example.com\nc@example.com\nd@example.com"
+
+      assert {:ok, import_state} = MemberCSV.prepare(csv_content, actor: actor)
+
+      assert length(import_state.preview_rows) == 3
+    end
+  end
+
+  describe "process_chunk/4 fee-type assignment" do
+    setup do
+      actor = Mv.Helpers.SystemActor.get_system_actor()
+
+      {:ok, fee_type} =
+        Mv.MembershipFees.create_membership_fee_type(
+          %{name: "Premium", amount: Decimal.new("25.00"), interval: :yearly},
+          actor: actor
+        )
+
+      %{actor: actor, fee_type: fee_type}
+    end
+
+    test "sets membership_fee_type_id when fee-type cell matches a known type", %{
+      actor: actor,
+      fee_type: fee_type
+    } do
+      chunk = [
+        {2, %{member: %{email: "fee-known@example.com"}, custom: %{}, fee_type: "Premium"}}
+      ]
+
+      opts = [
+        actor: actor,
+        fee_type_map: %{"premium" => fee_type.id}
+      ]
+
+      assert {:ok, result} = MemberCSV.process_chunk(chunk, %{email: 0}, %{}, opts)
+      assert result.inserted == 1
+
+      member =
+        Mv.Membership.list_members!(actor: actor)
+        |> Enum.find(&(&1.email == "fee-known@example.com"))
+
+      assert member.membership_fee_type_id == fee_type.id
+    end
+
+    test "adds a warning when the fee-type name is unknown", %{actor: actor} do
+      chunk = [
+        {2, %{member: %{email: "fee-unknown@example.com"}, custom: %{}, fee_type: "Ghost Type"}}
+      ]
+
+      opts = [actor: actor, fee_type_map: %{}]
+
+      assert {:ok, result} = MemberCSV.process_chunk(chunk, %{email: 0}, %{}, opts)
+      assert result.inserted == 1
+      assert Enum.any?(result.warnings, &(&1 =~ "Ghost Type"))
+    end
+
+    test "uses the default fee type when the fee-type cell is empty", %{
+      actor: actor,
+      fee_type: fee_type
+    } do
+      {:ok, settings} = Mv.Membership.get_settings()
+
+      {:ok, _settings} =
+        Mv.Membership.update_settings(
+          settings,
+          %{default_membership_fee_type_id: fee_type.id},
+          actor: actor
+        )
+
+      chunk = [{2, %{member: %{email: "fee-empty@example.com"}, custom: %{}, fee_type: ""}}]
+
+      opts = [actor: actor, fee_type_map: %{}]
+
+      assert {:ok, result} = MemberCSV.process_chunk(chunk, %{email: 0}, %{}, opts)
+      assert result.inserted == 1
+
+      member =
+        Mv.Membership.list_members!(actor: actor)
+        |> Enum.find(&(&1.email == "fee-empty@example.com"))
+
+      # Default fee type assigned via SetDefaultMembershipFeeType.
+      assert member.membership_fee_type_id == fee_type.id
+    end
+  end
+
+  describe "process_chunk/4 group assignment" do
+    setup do
+      %{actor: Mv.Helpers.SystemActor.get_system_actor()}
+    end
+
+    defp group_names_for(email, actor) do
+      member =
+        Mv.Membership.list_members!(actor: actor)
+        |> Enum.find(&(&1.email == email))
+
+      member = Ash.load!(member, :groups, actor: actor)
+      member.groups |> Enum.map(& &1.name) |> Enum.sort()
+    end
+
+    test "assigns member to an existing group", %{actor: actor} do
+      existing = Mv.Fixtures.group_fixture(%{name: "Orchester"})
+
+      chunk = [
+        {2, %{member: %{email: "g-existing@example.com"}, custom: %{}, groups: "Orchester"}}
+      ]
+
+      opts = [actor: actor, groups_found: [existing]]
+
+      assert {:ok, result} = MemberCSV.process_chunk(chunk, %{email: 0}, %{}, opts)
+      assert result.inserted == 1
+
+      assert group_names_for("g-existing@example.com", actor) == ["Orchester"]
+
+      # No new group was created.
+      orchester = Enum.filter(Mv.Membership.list_groups!(actor: actor), &(&1.name == "Orchester"))
+      assert length(orchester) == 1
+    end
+
+    test "auto-creates an unknown group and assigns the member", %{actor: actor} do
+      chunk = [
+        {2, %{member: %{email: "g-new@example.com"}, custom: %{}, groups: "Frische Gruppe"}}
+      ]
+
+      opts = [actor: actor, groups_found: []]
+
+      assert {:ok, result} = MemberCSV.process_chunk(chunk, %{email: 0}, %{}, opts)
+      assert result.inserted == 1
+
+      assert group_names_for("g-new@example.com", actor) == ["Frische Gruppe"]
+      assert Enum.any?(Mv.Membership.list_groups!(actor: actor), &(&1.name == "Frische Gruppe"))
+    end
+
+    test "handles multiple comma-separated groups", %{actor: actor} do
+      chunk = [
+        {2, %{member: %{email: "g-multi@example.com"}, custom: %{}, groups: "Orchester, Chor"}}
+      ]
+
+      opts = [actor: actor, groups_found: []]
+
+      assert {:ok, result} = MemberCSV.process_chunk(chunk, %{email: 0}, %{}, opts)
+      assert result.inserted == 1
+
+      assert group_names_for("g-multi@example.com", actor) == ["Chor", "Orchester"]
+    end
+
+    test "does not re-read the group table once per row for a repeated novel name",
+         %{actor: actor} do
+      rows =
+        for i <- 1..10 do
+          {i + 1,
+           %{member: %{email: "g-nplus1-#{i}@example.com"}, custom: %{}, groups: "Shared Group"}}
+        end
+
+      group_read_count = Agent.start_link(fn -> 0 end) |> elem(1)
+      test_pid = self()
+
+      # process_chunk runs synchronously in this test process, so the telemetry
+      # handler (invoked in the query-executing process) sees self() == test_pid.
+      # Filtering on the pid keeps concurrent tests' group queries out of the count.
+      handler = fn _event, _measurements, metadata, _config ->
+        if self() == test_pid and metadata[:source] == "groups" and
+             is_binary(metadata[:query]) and String.starts_with?(metadata.query, "SELECT") do
+          Agent.update(group_read_count, &(&1 + 1))
+        end
+      end
+
+      handler_id = "test-group-read-counter-#{System.unique_integer([:positive])}"
+      :telemetry.attach(handler_id, [:mv, :repo, :query], handler, nil)
+
+      assert {:ok, %{inserted: 10}} =
+               MemberCSV.process_chunk(rows, %{email: 0}, %{}, actor: actor, groups_found: [])
+
+      reads = Agent.get(group_read_count, & &1)
+      :telemetry.detach(handler_id)
+
+      # The novel group is created on the first row and reused in memory for the
+      # remaining nine. Without accumulation each row triggers a fresh full-table
+      # read, scaling linearly with the row count.
+      assert reads <= 3,
+             "Expected the group table read at most a few times, got #{reads} reads for 10 rows (N+1)."
+    end
+
+    test "empty groups cell leaves the member without group assignment", %{actor: actor} do
+      chunk = [{2, %{member: %{email: "g-empty@example.com"}, custom: %{}, groups: "  "}}]
+      opts = [actor: actor, groups_found: []]
+
+      assert {:ok, result} = MemberCSV.process_chunk(chunk, %{email: 0}, %{}, opts)
+      assert result.inserted == 1
+      assert result.errors == []
+
+      assert group_names_for("g-empty@example.com", actor) == []
+    end
+
+    property "re-importing the same groups does not create duplicates", %{actor: actor} do
+      check all(
+              name <- StreamData.string(:alphanumeric, min_length: 1, max_length: 15),
+              max_runs: 15
+            ) do
+        group_name = "dup-" <> name
+        email1 = "dup-#{System.unique_integer([:positive])}@example.com"
+        email2 = "dup-#{System.unique_integer([:positive])}@example.com"
+        opts = [actor: actor, groups_found: []]
+
+        chunk1 = [{2, %{member: %{email: email1}, custom: %{}, groups: group_name}}]
+        chunk2 = [{2, %{member: %{email: email2}, custom: %{}, groups: group_name}}]
+
+        assert {:ok, %{inserted: 1}} = MemberCSV.process_chunk(chunk1, %{email: 0}, %{}, opts)
+        assert {:ok, %{inserted: 1}} = MemberCSV.process_chunk(chunk2, %{email: 0}, %{}, opts)
+
+        matching =
+          Mv.Membership.list_groups!(actor: actor)
+          |> Enum.filter(&(String.downcase(&1.name) == String.downcase(group_name)))
+
+        assert length(matching) == 1
+      end
+    end
+  end
 end