diff --git a/Makefile b/Makefile
index adff3746..baf412dd 100644
--- a/Makefile
+++ b/Makefile
@@ -59,3 +59,25 @@ test: test_with_es ## run tests and generate coverage report
 install-local: ## installs your local edx-search into the LMS and CMS python virtualenvs
 	docker exec -t edx.devstack.lms bash -c '. /edx/app/edxapp/venvs/edxapp/bin/activate && cd /edx/app/edxapp/edx-platform && pip uninstall -y edx-search && pip install -e /edx/src/edx-search && pip freeze | grep edx-search'
 	docker exec -t edx.devstack.cms bash -c '. /edx/app/edxapp/venvs/edxapp/bin/activate && cd /edx/app/edxapp/edx-platform && pip uninstall -y edx-search && pip install -e /edx/src/edx-search && pip freeze | grep edx-search'
+
+test-meili: meili-up
+	@echo "Running Meilisearch tests..."
+	@MEILISEARCH_MASTER_KEY=test_master_key \
+	MEILISEARCH_URL=http://localhost:7700 \
+	pytest -v \
+	    search/tests/test_meilisearch.py \
+	    search/tests/test_course_discovery.py -k Meilisearch \
+	    search/tests/test_course_discovery_views.py -k Meilisearch || true
+	@$(MAKE) meili-down
+
+meili-up:
+	@echo "Starting Meilisearch..."
+	@docker compose up -d test_meilisearch
+	@echo "Waiting for Meilisearch to be healthy..."
+	@timeout 15 bash -c \
+    	'until curl -sf http://localhost:7700/health > /dev/null; do echo "Waiting..."; sleep 1; done'
+
+
+meili-down:
+	@echo "Shutting down Meilisearch..."
+	@docker compose down -v test_meilisearch
diff --git a/docker-compose.yml b/docker-compose.yml
index 2be4003b..ff4cd4b4 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -19,6 +19,22 @@ services:
     ports:
       - "9200:9200"
 
+  test_meilisearch:
+    image: getmeili/meilisearch:v1.7
+    container_name: test_meilisearch
+    ports:
+      - "7700:7700"
+    environment:
+      MEILISEARCH_MASTER_KEY: test_master_key
+      MEILISEARCH_URL: http://localhost:7700
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:7700/health"]
+      interval: 2s
+      timeout: 1s
+      retries: 10
+
+version: '3.8'
+
 volumes:
   data01:
     driver: local
diff --git a/edxsearch/settings.py b/edxsearch/settings.py
index e200eaff..d414fcbc 100644
--- a/edxsearch/settings.py
+++ b/edxsearch/settings.py
@@ -26,7 +26,7 @@
 # This is just a container for running tests
 DEBUG = True
 
-ALLOWED_HOSTS = []
+ALLOWED_HOSTS = ['testserver', 'localhost', '127.0.0.1']
 
 TEMPLATES = [
     {
@@ -128,3 +128,6 @@
         }
     }
 }
+
+MEILISEARCH_API_KEY = os.environ.get("MEILISEARCH_MASTER_KEY", "test_master_key")
+MEILISEARCH_URL = os.environ.get("MEILISEARCH_URL", "http://meilisearch")
diff --git a/search/api.py b/search/api.py
index 43d2ed18..233f8414 100644
--- a/search/api.py
+++ b/search/api.py
@@ -122,7 +122,7 @@ def emit_api_timing_event(search_term, course_id, filter_generation_timer, proce
     })
 
 
-def course_discovery_search(search_term=None, size=20, from_=0, field_dictionary=None):
+def course_discovery_search(search_term=None, size=20, from_=0, field_dictionary=None, is_multivalue=False):
     """
     Course Discovery activities against the search engine index of course details
     """
@@ -155,6 +155,7 @@ def course_discovery_search(search_term=None, size=20, from_=0, field_dictionary
         filter_dictionary={"enrollment_end": DateRange(datetime.utcnow(), None)},
         exclude_dictionary=exclude_dictionary,
         aggregation_terms=course_discovery_aggregations(),
+        is_multivalue=is_multivalue,
     )
 
     return results
diff --git a/search/elastic.py b/search/elastic.py
index 6a942ce5..e707f01c 100644
--- a/search/elastic.py
+++ b/search/elastic.py
@@ -22,7 +22,7 @@
 RESERVED_CHARACTERS = "+=><!(){}[]^~*:\\/&|?"
 
 
-def _translate_hits(es_response):
+def _translate_hits(es_response, aggregation_terms, is_multivalue=False):
     """
     Provide result set in our desired format from elasticsearch results.
 
@@ -98,7 +98,7 @@ def translate_result(result):
 
     def translate_agg_bucket(bucket, agg_result):
         """
-        Any conversion from ES aggregations result into our search engine syntax
+        Convert ES aggregation result following our search engine syntax.
 
         agg_result argument needs for getting total number of
         documents per bucket.
@@ -108,19 +108,29 @@ def translate_agg_bucket(bucket, agg_result):
         :return: dict
         """
         agg_item = agg_result[bucket]
-        terms = {
-            bucket["key"]: bucket["doc_count"]
-            for bucket in agg_item["buckets"]
-        }
-        total_docs = (
-            agg_result[_get_total_doc_key(bucket)]["value"]
-            + agg_item["sum_other_doc_count"]
-            + agg_item["doc_count_error_upper_bound"]
-        )
+
+        if is_multivalue:
+            values_agg = agg_item["values"]
+            terms = {
+                bucket["key"]: bucket["doc_count"]
+                for bucket in values_agg["buckets"]
+            }
+            total_docs = sum(terms.values())
+        else:
+            terms = {
+                bucket["key"]: bucket["doc_count"]
+                for bucket in agg_item["buckets"]
+            }
+            total_docs = (
+                    agg_result[_get_total_doc_key(bucket)]["value"]
+                    + agg_item["sum_other_doc_count"]
+                    + agg_item["doc_count_error_upper_bound"]
+            )
+
         return {
             "terms": terms,
             "total": total_docs,
-            "other": agg_item["sum_other_doc_count"],
+            "other": agg_item.get("sum_other_doc_count", 0),
         }
 
     results = list(map(translate_result, es_response["hits"]["hits"]))
@@ -131,11 +141,19 @@ def translate_agg_bucket(bucket, agg_result):
         "results": results,
     }
     if "aggregations" in es_response:
-        response["aggs"] = {
-            bucket: translate_agg_bucket(bucket, es_response["aggregations"])
-            for bucket in es_response["aggregations"]
-            if "total_" not in bucket
-        }
+        if is_multivalue:
+            global_aggs = es_response["aggregations"].get("global_aggs", {})
+            response["aggs"] = {
+                facet: translate_agg_bucket(facet, global_aggs)
+                for facet in global_aggs
+                if facet in aggregation_terms
+            }
+        else:
+            response["aggs"] = {
+                bucket: translate_agg_bucket(bucket, es_response["aggregations"])
+                for bucket in es_response["aggregations"]
+                if "total_" not in bucket
+            }
 
     return response
 
@@ -240,6 +258,47 @@ def _process_aggregation_terms(aggregation_terms):
     return elastic_aggs
 
 
+def _process_multivalue_aggregations(aggregation_terms, field_dictionary):
+    """
+    Calculate facet distributions for multi-value faceted search.
+    """
+    aggs = {}
+    for facet_field, options in aggregation_terms.items():
+        filters_excluding_facet = {
+            field: value for field, value in field_dictionary.items()
+            if field != facet_field
+        }
+        filter_clauses = [
+            _get_filter_field(field, value)
+            for field, value in filters_excluding_facet.items()
+            if value
+        ]
+        facet_filter = {
+            "bool": {
+                "must": filter_clauses
+            }
+        } if filter_clauses else {"match_all": {}}
+
+        aggs[facet_field] = {
+            "filter": facet_filter,
+            "aggs": {
+                "values": {
+                    "terms": {
+                        "field": facet_field,
+                        **options
+                    }
+                }
+            }
+        }
+
+    return {
+        "global_aggs": {
+            "global": {},
+            "aggs": aggs
+        }
+    }
+
+
 class ElasticSearchEngine(SearchEngine):
     """
     ElasticSearch implementation of SearchEngine abstraction
@@ -652,8 +711,13 @@ def search(self,
                 }
 
         body = {"query": query}
+
+        is_multivalue = kwargs.pop("is_multivalue", False)
         if aggregation_terms:
-            body["aggs"] = _process_aggregation_terms(aggregation_terms)
+            if is_multivalue:
+                body["aggs"] = _process_multivalue_aggregations(aggregation_terms, field_dictionary)
+            else:
+                body["aggs"] = _process_aggregation_terms(aggregation_terms)
 
         if log_search_params:
             log.info(f"full elastic search body {body}")
@@ -664,4 +728,4 @@ def search(self,
             log.exception("error while searching index - %r", ex)
             raise
 
-        return _translate_hits(es_response)
+        return _translate_hits(es_response, aggregation_terms, is_multivalue)
diff --git a/search/meilisearch.py b/search/meilisearch.py
index 9ee75015..58ebe566 100644
--- a/search/meilisearch.py
+++ b/search/meilisearch.py
@@ -70,6 +70,7 @@
 from django.conf import settings
 from django.utils import timezone
 
+from search.api import course_discovery_filter_fields
 from search.search_engine_base import SearchEngine
 from search.utils import ValueRange
 
@@ -168,6 +169,7 @@ def search(
         """
         See meilisearch docs: https://www.meilisearch.com/docs/reference/api/search
         """
+        is_multivalue = kwargs.pop("is_multivalue", False)
         opt_params = get_search_params(
             field_dictionary=field_dictionary,
             filter_dictionary=filter_dictionary,
@@ -178,8 +180,36 @@ def search(
         if log_search_params:
             logger.info("Search query: opt_params=%s", opt_params)
         meilisearch_results = self.meilisearch_index.search(query_string, opt_params)
-        processed_results = process_results(meilisearch_results, self.index_name)
-        return processed_results
+
+        if is_multivalue:
+            self._expand_facet_distibutions(field_dictionary, query_string, opt_params, meilisearch_results)
+
+        return process_results(meilisearch_results, self.index_name)
+
+    def _expand_facet_distibutions(self, field_dictionary, query_string, opt_params, meilisearch_results):
+        """
+        For each selected facet, get all its available options within the selected filters.
+        """
+        for facet in field_dictionary.keys():
+            expanded_facet_distribution = self._get_expanded_distribution(
+                query_string,
+                facet,
+                opt_params.get("filter", []),
+            )
+            meilisearch_results.setdefault("facetDistribution", {})[facet] = expanded_facet_distribution
+
+    def _get_expanded_distribution(self, query, facet_to_exclude, filter_rules):
+        """
+        Run a secondary query excluding one facet to get its full distribution.
+        Only return distribution data, without any actual results.
+        """
+        secondary_opt_params = {
+            'facets': [facet_to_exclude],
+            'filter': [rule for rule in filter_rules if not rule.startswith(f"{facet_to_exclude} = ")],
+            'limit': 0,
+        }
+        result = self.meilisearch_index.search(query, secondary_opt_params)
+        return result.get("facetDistribution", {}).get(facet_to_exclude, {})
 
     def remove(self, doc_ids, **kwargs):
         """
@@ -407,16 +437,20 @@ def get_filter_rules(
     Convert inclusion/exclusion rules.
     """
     rules = []
-    for key, value in rule_dict.items():
-        if isinstance(value, list):
-            for v in value:
-                rules.append(
-                    get_filter_rule(key, v, exclude=exclude, optional=optional)
-                )
+    filter_fields = course_discovery_filter_fields()
+    for rule_name, rule_value in rule_dict.items():
+        if isinstance(rule_value, list):
+            if rule_name in filter_fields:
+                rules.append(" OR ".join(f'{rule_name} = "{nested_value}"' for nested_value in rule_value))
+            else:
+                rules += [
+                    get_filter_rule(
+                        rule_name, nested_value, exclude=exclude, optional=optional
+                    ) for nested_value in rule_value
+                ]
         else:
-            rules.append(
-                get_filter_rule(key, value, exclude=exclude, optional=optional)
-            )
+            rules.append(get_filter_rule(rule_name, rule_value, exclude=exclude, optional=optional))
+
     return rules
 
 
diff --git a/search/tests/test_course_discovery.py b/search/tests/test_course_discovery.py
index 0c823f50..c57c3179 100644
--- a/search/tests/test_course_discovery.py
+++ b/search/tests/test_course_discovery.py
@@ -4,8 +4,10 @@
 """ Tests for search functionalty """
 
 import copy
+import time
 from datetime import datetime
 import ddt
+import meilisearch
 
 from django.core.cache import cache
 from django.test import TestCase
@@ -16,6 +18,7 @@
 from search.elastic import ElasticSearchEngine
 from search.tests.utils import SearcherMixin, TEST_INDEX_NAME
 from .mock_search_engine import MockSearchEngine
+from search.meilisearch import get_meilisearch_client, create_indexes
 
 
 class DemoCourse:
@@ -363,6 +366,80 @@ def test_course_matching(self, term, result_count):
         results = course_discovery_search(term)
         self.assertEqual(results["total"], result_count)
 
+    def test_aggregating_with_single_values_in_two_facets(self):
+        DemoCourse.get_and_index(self.searcher, {
+            "language": "en",
+            "org": "EDX",
+            "modes": "audit",
+        })
+
+        DemoCourse.get_and_index(self.searcher, {
+            "language": "en",
+            "org": "ORG2",
+            "modes": "honor",
+        })
+
+        results = course_discovery_search(
+            search_term="",
+            field_dictionary={"language": "en", "org": "EDX"}
+        )
+
+        self.assertIn("audit", results["aggs"]["modes"]["terms"])
+        self.assertNotIn("honor", results["aggs"]["modes"]["terms"])
+        self.assertEqual(results["aggs"]["language"]["terms"], {"en": 1})
+        self.assertEqual(results["aggs"]["org"]["terms"], {"EDX": 1})
+
+    def test_aggregating_with_multi_value_facet(self):
+        DemoCourse.get_and_index(self.searcher, {
+            "org": "EDX",
+            "language": "en",
+            "modes": "audit",
+        })
+
+        DemoCourse.get_and_index(self.searcher, {
+            "org": "EDX",
+            "language": "fr",
+            "modes": "honor",
+        })
+
+        DemoCourse.get_and_index(self.searcher, {
+            "org": "ORG2",
+            "language": "uk",
+            "modes": "verified",
+        })
+
+        results = course_discovery_search(
+            search_term="",
+            field_dictionary={"language": ["en", "fr"]},
+            is_multivalue=True
+        )
+
+        aggregations = results["aggs"]
+        self.assertIn("en", aggregations["language"]["terms"])
+        self.assertIn("fr", aggregations["language"]["terms"])
+        self.assertIn("uk", aggregations["language"]["terms"])
+        self.assertDictEqual(aggregations["language"]["terms"], {"en": 1, "fr": 1, "uk": 1})
+
+        self.assertNotIn("verified", aggregations["modes"]["terms"])
+        self.assertDictEqual(aggregations["modes"]["terms"], {"audit": 1, "honor": 1})
+
+        self.assertNotIn("ORG2", aggregations["org"]["terms"])
+        self.assertDictEqual(aggregations["org"]["terms"], {"EDX": 2})
+
+    def test_aggregating_facet_narrowed_if_single_value_search(self):
+        DemoCourse.get_and_index(self.searcher, {"language": "en", "modes": "audit"})
+
+        DemoCourse.get_and_index(self.searcher, {"language": "en", "modes": "honor"})
+
+        results = course_discovery_search(
+            search_term="",
+            field_dictionary={"modes": ["honor"]},
+            is_multivalue=False
+        )
+
+        self.assertNotIn("audit", results["aggs"]["modes"]["terms"])
+        self.assertDictEqual(results["aggs"]["modes"]["terms"], {'honor': 1})
+
 
 @override_settings(SEARCH_ENGINE=None)
 class TestNone(TestCase):
@@ -372,3 +449,288 @@ def test_perform_search(self):
         """ search opertaion should yeild an exception with no search engine """
         with self.assertRaises(NoSearchEngineError):
             course_discovery_search("abc test")
+
+
+@override_settings(SEARCH_ENGINE="search.meilisearch.MeilisearchEngine")
+@override_settings(COURSEWARE_INFO_INDEX_NAME=TEST_INDEX_NAME)
+class TestMeilisearchCourseDiscoverySearch(TestCase, SearcherMixin):
+    """
+    Integration tests using real Meilisearch engine.
+    """
+
+    def setUp(self):
+        super().setUp()
+        create_indexes({TEST_INDEX_NAME: [
+            "language",
+            "modes",
+            "org",
+            "catalog_visibility",
+            "enrollment_start",
+            "enrollment_end",
+        ]})
+        self.wait_for_meilisearch_indexing()
+
+    def tearDown(self):
+        client = get_meilisearch_client()
+        try:
+            client.index(TEST_INDEX_NAME).delete()
+        except meilisearch.errors.MeilisearchApiError:
+            pass
+        super().tearDown()
+
+    @staticmethod
+    def wait_for_meilisearch_indexing():
+        from search.meilisearch import get_meilisearch_client
+        client = get_meilisearch_client()
+        task = client.index(TEST_INDEX_NAME).get_tasks().results[-1]
+        if not task:
+            return
+        client.wait_for_task(task.uid)
+        time.sleep(0.2)
+
+    def test_course_matching_empty_index(self):
+        """ Check for empty result count before indexing """
+        results = course_discovery_search("defensive")
+        self.assertEqual(results["total"], 0)
+
+    def test_course_matching(self):
+        """ Make sure that matches within content can be located and processed """
+        DemoCourse.get_and_index(self.searcher, {
+            "content": {
+                "short_description": "This is a defensive move",
+                "overview": "Defensive teams often win"
+            }
+        })
+        self.wait_for_meilisearch_indexing()
+
+        DemoCourse.get_and_index(self.searcher, {
+            "content": {
+                "short_description": "This is an offensive move",
+                "overview": "Offensive teams often win"
+            }
+        })
+        self.wait_for_meilisearch_indexing()
+
+        DemoCourse.get_and_index(self.searcher, {
+            "content": {
+                "short_description": "This is a hyphenated move",
+                "overview": "Highly-offensive teams often win"
+            }
+        })
+        self.wait_for_meilisearch_indexing()
+
+        results = course_discovery_search()
+        self.assertEqual(results["total"], 3)
+
+    @override_settings(COURSE_DISCOVERY_AGGREGATIONS={"subject": {}, "lang": {}})
+    def test_aggregating_override(self):
+        """
+        Test that aggregation under consideration can be specified
+        with custom setting
+        """
+        create_indexes({TEST_INDEX_NAME: [
+            "lang",
+            "subject",
+        ]})
+
+        DemoCourse.get_and_index(self.searcher, {"subject": "Mathematics", "lang": ["en", "fr"]})
+        DemoCourse.get_and_index(self.searcher, {"subject": "Mathematics", "lang": ["en"]})
+        DemoCourse.get_and_index(self.searcher, {"subject": "History", "lang": ["en"]})
+        DemoCourse.get_and_index(self.searcher, {"subject": "History", "lang": ["fr"]})
+        DemoCourse.get_and_index(self.searcher, {"lang": ["de"]})
+        self.wait_for_meilisearch_indexing()
+
+        results = course_discovery_search()
+        self.assertEqual(results["total"], 5)
+        self.assertIn("aggs", results)
+
+        self.assertNotIn("org", results["aggs"])
+        self.assertNotIn("modes", results["aggs"])
+
+        self.assertIn("subject", results["aggs"])
+        self.assertEqual(results["aggs"]["subject"]["total"], 4)
+        self.assertEqual(results["aggs"]["subject"]["terms"]["Mathematics"], 2)
+        self.assertEqual(results["aggs"]["subject"]["terms"]["History"], 2)
+
+        self.assertIn("lang", results["aggs"])
+        self.assertEqual(results["aggs"]["lang"]["total"], 6)
+        self.assertEqual(results["aggs"]["lang"]["terms"]["en"], 3)
+        self.assertEqual(results["aggs"]["lang"]["terms"]["fr"], 2)
+        self.assertEqual(results["aggs"]["lang"]["terms"]["de"], 1)
+
+    def test_course_list(self):
+        """ No arguments to course_discovery_search should show all available courses"""
+        results = course_discovery_search()
+        self.assertEqual(results["total"], 0)
+
+        DemoCourse.get_and_index(self.searcher)
+        self.wait_for_meilisearch_indexing()
+        results = course_discovery_search()
+        self.assertEqual(results["total"], 1)
+
+    def test_discovery_field_matching(self):
+        """ Test that field specifications only show those results with the desired field values """
+        DemoCourse.get_and_index(self.searcher, {"org": "OrgA"})
+        DemoCourse.get_and_index(self.searcher, {"org": "OrgB"})
+        self.wait_for_meilisearch_indexing()
+
+        results = course_discovery_search()
+        self.assertEqual(results["total"], 2)
+
+        results = course_discovery_search(field_dictionary={"org": "OrgA"})
+        self.assertEqual(results["total"], 1)
+        self.assertEqual(results["results"][0]["data"]["org"], "OrgA")
+
+        results = course_discovery_search(field_dictionary={"org": "OrgB"})
+        self.assertEqual(results["total"], 1)
+        self.assertEqual(results["results"][0]["data"]["org"], "OrgB")
+
+    def test_multivalue_field_matching(self):
+        """
+        Test that field specifications only show those results with the desired
+        field values - even when there is an array of possible values
+        """
+        DemoCourse.get_and_index(self.searcher, {"modes": ["honor", "verified"]})
+        DemoCourse.get_and_index(self.searcher, {"modes": ["honor"]})
+        self.wait_for_meilisearch_indexing()
+
+        results = course_discovery_search()
+        self.assertEqual(results["total"], 2)
+
+        results = course_discovery_search(field_dictionary={"modes": "honor"})
+        self.assertEqual(results["total"], 2)
+
+        results = course_discovery_search(field_dictionary={"modes": "verified"})
+        self.assertEqual(results["total"], 1)
+
+    def test_enroll_date(self):
+        """
+        Test that we don't show any courses that have no published enrollment date, or an enrollment date in the future
+        """
+        # demo_course_1 should be found cos it has a date that is valid
+        DemoCourse.get_and_index(self.searcher, {"enrollment_start": datetime(2014, 1, 1)})
+
+        # demo_course_2 should not be found because it has enrollment_start date set explicitly to None
+        DemoCourse.get_and_index(self.searcher, {"enrollment_start": None})
+
+        # demo_course_3 should not be found because it has enrollment_start date in the future
+        DemoCourse.get_and_index(self.searcher, {"enrollment_start": datetime(2114, 1, 1)})
+
+        # demo_course_4 should not be found because it has no enrollment_start specification
+        DemoCourse.get_and_index(self.searcher, {}, ["enrollment_start"])
+
+        self.wait_for_meilisearch_indexing()
+
+        results = course_discovery_search()
+        self.assertEqual(results["total"], 1)
+
+        additional_course = DemoCourse.get()
+        DemoCourse.index(self.searcher, [additional_course])
+        self.wait_for_meilisearch_indexing()
+
+        results = course_discovery_search()
+        self.assertEqual(results["total"], 2)
+
+        # Mark the course as having ended enrollment
+        additional_course["enrollment_end"] = datetime(2015, 1, 1)
+        DemoCourse.index(self.searcher, [additional_course])
+        self.wait_for_meilisearch_indexing()
+
+        results = course_discovery_search()
+        self.assertEqual(results["total"], 1)
+
+    def test_aggregating(self):
+        DemoCourse.get_and_index(self.searcher, {"language": "en", "org": "EDX"})
+        self.wait_for_meilisearch_indexing()
+
+        DemoCourse.get_and_index(self.searcher, {"language": "fr", "org": "ORG2"})
+        self.wait_for_meilisearch_indexing()
+
+        results = course_discovery_search(
+            search_term="",
+            field_dictionary={"language": ["en"]},
+            is_multivalue=True
+        )
+        self.assertDictEqual(results["aggs"]["language"]["terms"], {"en": 1, "fr": 1})
+        self.assertDictEqual(results["aggs"]["org"]["terms"], {"EDX": 1})
+
+    def test_aggregating_with_single_values_in_two_facets(self):
+        DemoCourse.get_and_index(self.searcher, {
+            "language": "en",
+            "org": "EDX",
+            "modes": "audit",
+        })
+        self.wait_for_meilisearch_indexing()
+
+        DemoCourse.get_and_index(self.searcher, {
+            "language": "en",
+            "org": "ORG2",
+            "modes": "honor",
+        })
+        self.wait_for_meilisearch_indexing()
+
+        results = course_discovery_search(
+            search_term="",
+            field_dictionary={"language": "en", "org": "EDX"}
+        )
+
+        self.assertIn("audit", results["aggs"]["modes"]["terms"])
+        self.assertNotIn("honor", results["aggs"]["modes"]["terms"])
+        self.assertEqual(results["aggs"]["language"]["terms"], {"en": 1})
+        self.assertEqual(results["aggs"]["org"]["terms"], {"EDX": 1})
+
+    def test_aggregating_with_multi_value_facet(self):
+        DemoCourse.get_and_index(self.searcher, {
+            "org": "EDX",
+            "language": "en",
+            "modes": "audit",
+        })
+        self.wait_for_meilisearch_indexing()
+
+        DemoCourse.get_and_index(self.searcher, {
+            "org": "EDX",
+            "language": "fr",
+            "modes": "honor",
+        })
+        self.wait_for_meilisearch_indexing()
+
+        DemoCourse.get_and_index(self.searcher, {
+            "org": "ORG2",
+            "language": "uk",
+            "modes": "verified",
+        })
+        self.wait_for_meilisearch_indexing()
+
+        results = course_discovery_search(
+            search_term="",
+            field_dictionary={"language": ["en", "fr"]},
+            is_multivalue=True
+        )
+
+        aggregations = results["aggs"]
+        self.assertIn("en", aggregations["language"]["terms"])
+        self.assertIn("fr", aggregations["language"]["terms"])
+        self.assertIn("uk", aggregations["language"]["terms"])
+        self.assertDictEqual(aggregations["language"]["terms"], {"en": 1, "fr": 1, "uk": 1})
+
+        self.assertNotIn("verified", aggregations["modes"]["terms"])
+        self.assertDictEqual(aggregations["modes"]["terms"], {"audit": 1, "honor": 1})
+
+        self.assertNotIn("ORG2", aggregations["org"]["terms"])
+        self.assertDictEqual(aggregations["org"]["terms"], {"EDX": 2})
+
+    def test_aggregating_facet_narrowed_if_single_value_search(self):
+        DemoCourse.get_and_index(self.searcher, {"language": "en", "modes": "audit"})
+        self.wait_for_meilisearch_indexing()
+
+        DemoCourse.get_and_index(self.searcher, {"language": "en", "modes": "honor"})
+        self.wait_for_meilisearch_indexing()
+
+        results = course_discovery_search(
+            search_term="",
+            field_dictionary={"modes": ["honor"]},
+            is_multivalue=False
+        )
+
+        self.assertNotIn("audit", results["aggs"]["modes"]["terms"])
+        self.assertDictEqual(results["aggs"]["modes"]["terms"], {'honor': 1})
diff --git a/search/tests/test_course_discovery_views.py b/search/tests/test_course_discovery_views.py
index f01a442a..18e76c62 100644
--- a/search/tests/test_course_discovery_views.py
+++ b/search/tests/test_course_discovery_views.py
@@ -1,11 +1,18 @@
 """ High-level view tests"""
+import time
 
+from django.core.cache import cache
+from django.test import TestCase
 from django.test.utils import override_settings
+from django.urls import reverse
+from elasticsearch.client import Elasticsearch
 
 from search.tests.tests import TEST_INDEX_NAME
-from search.tests.utils import post_discovery_request
+from search.tests.utils import post_discovery_request, SearcherMixin
 from .test_views import MockSearchUrlTest
 from .test_course_discovery import DemoCourse
+from search.meilisearch import create_indexes, get_meilisearch_client
+from search.search_engine_base import SearchEngine
 
 
 @override_settings(ELASTIC_FIELD_MAPPINGS={
@@ -138,3 +145,515 @@ def test_bad_engine(self):
         code, results = post_discovery_request({"search_string": "sun"})
         self.assertGreater(code, 499)
         self.assertEqual(results["error"], 'An error occurred when searching for "sun"')
+
+
+@override_settings(
+    SEARCH_ENGINE="search.meilisearch.MeilisearchEngine",
+    COURSEWARE_CONTENT_INDEX_NAME=TEST_INDEX_NAME,
+    COURSEWARE_INFO_INDEX_NAME=TEST_INDEX_NAME,
+)
+class TestMeilisearchSingleValueDiscoveryUrl(TestCase, SearcherMixin):
+    """
+    Integration tests for Meilisearch + /course_discovery/ endpoint
+    """
+    meilisearch_client = get_meilisearch_client()
+
+    def setUp(self):
+        super().setUp()
+        try:
+            self.meilisearch_client.get_index(TEST_INDEX_NAME).delete()
+        except Exception:
+            pass
+        create_indexes({TEST_INDEX_NAME: [
+            "language",
+            "modes",
+            "org",
+            "catalog_visibility",
+            "enrollment_start",
+            "enrollment_end",
+        ]})
+        self.wait_for_meilisearch_indexing()
+
+        DemoCourse.reset_count()
+        DemoCourse.get_and_index(
+            self.searcher, {"org": "OrgA", "content": {"short_description": "Find this one with the right parameter"}}
+        )
+        DemoCourse.get_and_index(
+            self.searcher, {"org": "OrgB", "content": {"short_description": "Find this one with another parameter"}}
+        )
+        DemoCourse.get_and_index(
+            self.searcher, {"content": {"short_description": "Find this one somehow"}}
+        )
+        self.wait_for_meilisearch_indexing()
+
+    def wait_for_meilisearch_indexing(self):
+        task = self.meilisearch_client.index(TEST_INDEX_NAME).get_tasks().results[-1]
+        if not task:
+            return
+        self.meilisearch_client.wait_for_task(task.uid)
+        time.sleep(0.2)
+
+    def test_search_string(self):
+        code, results = post_discovery_request({})
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 3)
+
+        code, results = post_discovery_request({"search_string": "right"})
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 1)
+
+        code, results = post_discovery_request({"search_string": "parameter"})
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 2)
+
+    def test_org_filter(self):
+        code, results = post_discovery_request({"org": "OrgA"})
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 1)
+        self.assertEqual(results["results"][0]["data"]["org"], "OrgA")
+
+        code, results = post_discovery_request({"org": "OrgB"})
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 1)
+        self.assertEqual(results["results"][0]["data"]["org"], "OrgB")
+
+    def test_search_with_pagination(self):
+        code, results = post_discovery_request({"page_size": 2})
+        self.assertEqual(code, 200)
+        self.assertEqual(len(results["results"]), 2)
+
+        code, results = post_discovery_request({"page_size": 2, "page_index": 1})
+        self.assertEqual(code, 200)
+        self.assertEqual(len(results["results"]), 1)
+
+    def test_bad_search_string(self):
+        code, results = post_discovery_request({"search_string": "doesnotexist123"})
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 0)
+
+    def test_aggregations_basic(self):
+        code, results = post_discovery_request({})
+        self.assertEqual(code, 200)
+        aggs = results.get("aggs", {})
+        self.assertIn("org", aggs)
+        self.assertEqual(aggs["org"]["terms"].get("OrgA", 0), 1)
+        self.assertEqual(aggs["org"]["terms"].get("OrgB", 0), 1)
+
+    def test_aggregations_filtered_down(self):
+        code, results = post_discovery_request({"org": "OrgA"})
+        self.assertEqual(code, 200)
+        aggs = results.get("aggs", {})
+        self.assertIn("org", aggs)
+        self.assertEqual(aggs["org"]["terms"].get("OrgA", 0), 1)
+        self.assertNotIn("OrgB", aggs["org"]["terms"])
+
+    def test_aggregations_empty_search(self):
+        code, results = post_discovery_request({"org": "DoesNotExist"})
+        self.assertEqual(code, 200)
+        aggs = results.get("aggs", {})
+        self.assertIn("org", aggs)
+        self.assertEqual(aggs["org"]["terms"], {})
+
+
+@override_settings(
+    SEARCH_ENGINE="search.meilisearch.MeilisearchEngine",
+    COURSEWARE_CONTENT_INDEX_NAME=TEST_INDEX_NAME,
+    COURSEWARE_INFO_INDEX_NAME=TEST_INDEX_NAME,
+)
+class TestMeilisearchMultiValueDiscoveryUrl(TestCase, SearcherMixin):
+    """
+    Integration tests for Meilisearch + /course_discovery_multivalue/ endpoint
+    """
+    meilisearch_client = get_meilisearch_client()
+    multivalue_search_url = reverse("course_discovery_multivalue")
+
+    def setUp(self):
+        super().setUp()
+        try:
+            self.meilisearch_client.get_index(TEST_INDEX_NAME).delete()
+        except Exception:
+            pass
+        create_indexes({TEST_INDEX_NAME: [
+            "language",
+            "modes",
+            "org",
+            "catalog_visibility",
+            "enrollment_start",
+            "enrollment_end",
+        ]})
+        self.wait_for_meilisearch_indexing()
+
+        DemoCourse.reset_count()
+        DemoCourse.get_and_index(
+            self.searcher, {
+                "org": "OrgA",
+                "language": "en",
+                "content": {"short_description": "Find this one with the right parameter"}
+            }
+        )
+        DemoCourse.get_and_index(
+            self.searcher, {
+                "org": "OrgB",
+                "language": "fr",
+                "content": {"short_description": "Find this one with another parameter"}
+            }
+        )
+        DemoCourse.get_and_index(
+            self.searcher, {
+                "org": "OrgC",
+                "language": "en",
+                "content": {"short_description": "Find this one somehow"}
+            }
+        )
+        self.wait_for_meilisearch_indexing()
+
+    def wait_for_meilisearch_indexing(self):
+        task = self.meilisearch_client.index(TEST_INDEX_NAME).get_tasks().results[-1]
+        if not task:
+            return
+        self.meilisearch_client.wait_for_task(task.uid)
+        time.sleep(0.2)
+
+    def test_search_string(self):
+        code, results = post_discovery_request({}, address=self.multivalue_search_url)
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 3)
+
+        code, results = post_discovery_request({"search_string": "right"}, address=self.multivalue_search_url)
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 1)
+
+        code, results = post_discovery_request({"search_string": "parameter"}, address=self.multivalue_search_url)
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 2)
+
+    def test_org_filter(self):
+        code, results = post_discovery_request({"org": "OrgA"})
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 1)
+        self.assertEqual(results["results"][0]["data"]["org"], "OrgA")
+
+        code, results = post_discovery_request({"org": "OrgB"}, address=self.multivalue_search_url)
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 1)
+        self.assertEqual(results["results"][0]["data"]["org"], "OrgB")
+
+    def test_search_with_pagination(self):
+        code, results = post_discovery_request({"page_size": 2}, address=self.multivalue_search_url)
+        self.assertEqual(code, 200)
+        self.assertEqual(len(results["results"]), 2)
+
+        code, results = post_discovery_request({"page_size": 2, "page_index": 1})
+        self.assertEqual(code, 200)
+        self.assertEqual(len(results["results"]), 1)
+
+    def test_bad_search_string(self):
+        code, results = post_discovery_request(
+            {"search_string": "doesnotexist123"}, address=self.multivalue_search_url
+        )
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 0)
+
+    def test_no_filters_returns_all_aggregations(self):
+        code, results = post_discovery_request({}, address=self.multivalue_search_url)
+        self.assertEqual(code, 200)
+        aggs = results.get("aggs", {})
+        self.assertIn("org", aggs)
+        self.assertIn("language", aggs)
+        self.assertEqual(aggs["org"]["terms"]["OrgA"], 1)
+        self.assertEqual(aggs["org"]["terms"]["OrgB"], 1)
+        self.assertEqual(aggs["org"]["terms"]["OrgC"], 1)
+        self.assertEqual(aggs["language"]["terms"]["en"], 2)
+        self.assertEqual(aggs["language"]["terms"]["fr"], 1)
+
+    def test_single_value_filter_keeps_full_facet(self):
+        code, results = post_discovery_request(
+            {"language": ["en"]}, address=self.multivalue_search_url
+        )
+        self.assertEqual(code, 200)
+        aggs = results.get("aggs", {})
+        self.assertIn("language", aggs)
+        # This is the key difference with multi-facet logic:
+        # all language options should be returned, even though "en" is selected
+        self.assertIn("en", aggs["language"]["terms"])
+        self.assertIn("fr", aggs["language"]["terms"])
+        self.assertEqual(results["total"], 2)
+
+    def test_multi_value_filter_keeps_full_facet(self):
+        code, results = post_discovery_request(
+            {"language": ["en", "fr"]}, address=self.multivalue_search_url
+        )
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 3)
+
+        aggs = results.get("aggs", {})
+        self.assertIn("language", aggs)
+        self.assertIn("en", aggs["language"]["terms"])
+        self.assertIn("fr", aggs["language"]["terms"])
+        self.assertEqual(aggs["language"]["terms"]["en"], 2)
+        self.assertEqual(aggs["language"]["terms"]["fr"], 1)
+
+    def test_combined_facet_filter_aggregated_correctly(self):
+        code, results = post_discovery_request(
+            {"language": ["en"], "org": ["OrgA", "OrgC"]},
+            address=self.multivalue_search_url
+        )
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 2)
+
+        aggs = results.get("aggs", {})
+        self.assertIn("org", aggs)
+        self.assertIn("OrgA", aggs["org"]["terms"])
+        self.assertIn("OrgC", aggs["org"]["terms"])
+
+
+@override_settings(
+    SEARCH_ENGINE="search.tests.utils.ForceRefreshElasticSearchEngine",
+    COURSEWARE_CONTENT_INDEX_NAME=TEST_INDEX_NAME,
+    COURSEWARE_INFO_INDEX_NAME=TEST_INDEX_NAME,
+)
+class TestElasticsearchSingleValueDiscoveryUrl(TestCase, SearcherMixin):
+    """
+    Integration tests for Elasticsearch + /course_discovery/ endpoint
+    """
+
+    def setUp(self):
+        super().setUp()
+        _elasticsearch = Elasticsearch()
+        _elasticsearch.indices.delete(index=TEST_INDEX_NAME, ignore=[400, 404])
+        cache.clear()
+        config_body = {}
+        _elasticsearch.indices.create(index=TEST_INDEX_NAME, ignore=400, body=config_body)
+        DemoCourse.reset_count()
+        self._searcher = None
+
+        DemoCourse.get_and_index(self.searcher, {
+            "org": "OrgA", "content": {"short_description": "Find this one with the right parameter"}
+        })
+        DemoCourse.get_and_index(self.searcher, {
+            "org": "OrgB", "content": {"short_description": "Find this one with another parameter"}
+        })
+        DemoCourse.get_and_index(self.searcher, {
+            "content": {"short_description": "Find this one somehow"}
+        })
+
+    def tearDown(self):
+        _elasticsearch = Elasticsearch()
+        _elasticsearch.indices.delete(index=TEST_INDEX_NAME, ignore=[400, 404])
+        self._searcher = None
+        super().tearDown()
+
+        DemoCourse.reset_count()
+
+    def test_search_string(self):
+        """Tests that keyword search returns correct number of matching documents."""
+        code, results = post_discovery_request({})
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 3)
+
+        code, results = post_discovery_request({"search_string": "right"})
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 1)
+
+        code, results = post_discovery_request({"search_string": "parameter"})
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 2)
+
+    def test_org_filter(self):
+        """Tests filtering results by the 'org' facet."""
+        code, results = post_discovery_request({"org": "OrgA"})
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 1)
+        self.assertEqual(results["results"][0]["data"]["org"], "OrgA")
+
+        code, results = post_discovery_request({"org": "OrgB"})
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 1)
+        self.assertEqual(results["results"][0]["data"]["org"], "OrgB")
+
+    def test_search_with_pagination(self):
+        """Tests that pagination limits and offsets results correctly."""
+        code, results = post_discovery_request({"page_size": 2})
+        self.assertEqual(code, 200)
+        self.assertEqual(len(results["results"]), 2)
+
+        code, results = post_discovery_request({"page_size": 2, "page_index": 1})
+        self.assertEqual(code, 200)
+        self.assertEqual(len(results["results"]), 1)
+
+    def test_bad_search_string(self):
+        """Tests that non-matching search terms return no results."""
+        code, results = post_discovery_request({"search_string": "doesnotexist123"})
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 0)
+
+    def test_aggregations_basic(self):
+        """Tests that facet aggregations include all indexed orgs."""
+        code, results = post_discovery_request({})
+        self.assertEqual(code, 200)
+        aggs = results.get("aggs", {})
+        self.assertIn("org", aggs)
+        self.assertEqual(aggs["org"]["terms"].get("OrgA", 0), 1)
+        self.assertEqual(aggs["org"]["terms"].get("OrgB", 0), 1)
+
+    def test_aggregations_filtered_down(self):
+        """Tests that aggregations reflect active filters correctly."""
+        code, results = post_discovery_request({"org": "OrgA"})
+        self.assertEqual(code, 200)
+        aggs = results.get("aggs", {})
+        self.assertIn("org", aggs)
+        self.assertEqual(aggs["org"]["terms"].get("OrgA", 0), 1)
+        self.assertNotIn("OrgB", aggs["org"]["terms"])
+
+    def test_aggregations_empty_search(self):
+        """Tests that aggregations are returned even if there are no matches."""
+        code, results = post_discovery_request({"org": "DoesNotExist"})
+        self.assertEqual(code, 200)
+        aggs = results.get("aggs", {})
+        self.assertIn("org", aggs)
+        self.assertEqual(aggs["org"]["terms"], {})
+
+
+@override_settings(
+    SEARCH_ENGINE="search.tests.utils.ForceRefreshElasticSearchEngine",
+    COURSEWARE_CONTENT_INDEX_NAME=TEST_INDEX_NAME,
+    COURSEWARE_INFO_INDEX_NAME=TEST_INDEX_NAME,
+)
+class TestElasticsearchMultiValueDiscoveryUrl(TestCase, SearcherMixin):
+    """
+    Integration tests for Elasticsearch + /course_discovery_multivalue/ endpoint
+    """
+
+    multivalue_search_url = reverse("course_discovery_multivalue")
+
+    def setUp(self):
+        super().setUp()
+        _elasticsearch = Elasticsearch()
+        _elasticsearch.indices.delete(index=TEST_INDEX_NAME, ignore=[400, 404])
+        cache.clear()
+        config_body = {}
+        _elasticsearch.indices.create(index=TEST_INDEX_NAME, ignore=400, body=config_body)
+        DemoCourse.reset_count()
+        self._searcher = None
+
+        DemoCourse.get_and_index(
+            self.searcher, {
+                "org": "OrgA",
+                "language": "en",
+                "content": {"short_description": "Find this one with the right parameter"}
+            }
+        )
+        DemoCourse.get_and_index(
+            self.searcher, {
+                "org": "OrgB",
+                "language": "fr",
+                "content": {"short_description": "Find this one with another parameter"}
+            }
+        )
+        DemoCourse.get_and_index(
+            self.searcher, {
+                "org": "OrgC",
+                "language": "en",
+                "content": {"short_description": "Find this one somehow"}
+            }
+        )
+
+    def test_search_string(self):
+        """Tests that keyword search returns correct number of matching documents."""
+        code, results = post_discovery_request({}, address=self.multivalue_search_url)
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 3)
+
+        code, results = post_discovery_request({"search_string": "right"}, address=self.multivalue_search_url)
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 1)
+
+        code, results = post_discovery_request({"search_string": "parameter"}, address=self.multivalue_search_url)
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 2)
+
+    def test_org_filter(self):
+        """Tests filtering results by the 'org' facet."""
+        code, results = post_discovery_request({"org": "OrgA"})
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 1)
+        self.assertEqual(results["results"][0]["data"]["org"], "OrgA")
+
+        code, results = post_discovery_request({"org": "OrgB"}, address=self.multivalue_search_url)
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 1)
+        self.assertEqual(results["results"][0]["data"]["org"], "OrgB")
+
+    def test_search_with_pagination(self):
+        """Tests that pagination limits and offsets results correctly."""
+        code, results = post_discovery_request({"page_size": 2}, address=self.multivalue_search_url)
+        self.assertEqual(code, 200)
+        self.assertEqual(len(results["results"]), 2)
+
+        code, results = post_discovery_request({"page_size": 2, "page_index": 1})
+        self.assertEqual(code, 200)
+        self.assertEqual(len(results["results"]), 1)
+
+    def test_bad_search_string(self):
+        """Tests that non-matching search terms return no results."""
+        code, results = post_discovery_request(
+            {"search_string": "doesnotexist123"}, address=self.multivalue_search_url
+        )
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 0)
+
+    def test_no_filters_returns_all_aggregations(self):
+        """Tests that full facet counts are returned when no filters are applied."""
+        code, results = post_discovery_request({}, address=self.multivalue_search_url)
+        self.assertEqual(code, 200)
+        aggs = results.get("aggs", {})
+        self.assertIn("org", aggs)
+        self.assertIn("language", aggs)
+        self.assertEqual(aggs["org"]["terms"]["OrgA"], 1)
+        self.assertEqual(aggs["org"]["terms"]["OrgB"], 1)
+        self.assertEqual(aggs["org"]["terms"]["OrgC"], 1)
+        self.assertEqual(aggs["language"]["terms"]["en"], 2)
+        self.assertEqual(aggs["language"]["terms"]["fr"], 1)
+
+    def test_single_value_filter_keeps_full_facet(self):
+        """Tests that single-value filters preserve all facet options in aggregations."""
+        code, results = post_discovery_request(
+            {"language": ["en"]}, address=self.multivalue_search_url
+        )
+        self.assertEqual(code, 200)
+        aggs = results.get("aggs", {})
+        self.assertIn("language", aggs)
+        # This is the key difference with multi-facet logic:
+        # all language options should be returned, even though "en" is selected
+        self.assertIn("en", aggs["language"]["terms"])
+        self.assertIn("fr", aggs["language"]["terms"])
+        self.assertEqual(results["total"], 2)
+
+    def test_multi_value_filter_keeps_full_facet(self):
+        """Tests that multi-value filters preserve all facet options in aggregations."""
+        code, results = post_discovery_request(
+            {"language": ["en", "fr"]}, address=self.multivalue_search_url
+        )
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 3)
+
+        aggs = results.get("aggs", {})
+        self.assertIn("language", aggs)
+        self.assertIn("en", aggs["language"]["terms"])
+        self.assertIn("fr", aggs["language"]["terms"])
+        self.assertEqual(aggs["language"]["terms"]["en"], 2)
+        self.assertEqual(aggs["language"]["terms"]["fr"], 1)
+
+    def test_combined_facet_filter_aggregated_correctly(self):
+        """Tests that combining multiple facet filters returns correct aggregations."""
+        code, results = post_discovery_request(
+            {"language": ["en"], "org": ["OrgA", "OrgC"]},
+            address=self.multivalue_search_url
+        )
+        self.assertEqual(code, 200)
+        self.assertEqual(results["total"], 2)
+
+        aggs = results.get("aggs", {})
+        self.assertIn("org", aggs)
+        self.assertIn("OrgA", aggs["org"]["terms"])
+        self.assertIn("OrgC", aggs["org"]["terms"])
diff --git a/search/tests/test_engines.py b/search/tests/test_engines.py
index 7888041c..7c98d4a1 100644
--- a/search/tests/test_engines.py
+++ b/search/tests/test_engines.py
@@ -6,14 +6,15 @@
 import json
 import os
 from datetime import datetime
-from unittest.mock import patch
+from unittest.mock import patch, MagicMock
 
 from django.test import TestCase
 from django.test.utils import override_settings
 from elasticsearch import exceptions
+from elasticsearch.client import Elasticsearch
 from elasticsearch.helpers import BulkIndexError
 from search.api import NoSearchEngineError, perform_search
-from search.elastic import RESERVED_CHARACTERS
+from search.elastic import RESERVED_CHARACTERS, ElasticSearchEngine
 from search.tests.mock_search_engine import (MockSearchEngine,
                                              json_date_to_datetime)
 from search.tests.tests import MockSearchTests
@@ -250,3 +251,265 @@ def test_config(self):
         elasticsearch = self.searcher._es  # pylint: disable=protected-access
         hosts = elasticsearch.transport.hosts
         self.assertEqual(hosts, [{'host': '127.0.0.1'}, {'host': 'localhost'}])
+
+
+class ElasticSearchUnitTests(TestCase):
+    """
+    ElasticSearch tests.
+    """
+
+    @patch("search.elastic.Elasticsearch")
+    def test_multivalue_aggregations_translated_correctly(self, mock_elasticsearch_class):
+        """Tests that multivalue facet aggregations return full facet buckets despite filtering."""
+        mock_es = MagicMock()
+        mock_elasticsearch_class.return_value = mock_es
+
+        mock_es.search.return_value = {
+            "hits": {
+                "total": {"value": 2},
+                "max_score": 1.0,
+                "hits": [
+                    {
+                        "_source": {"org": "OrgA", "language": "en"},
+                        "_score": 1.0
+                    },
+                    {
+                        "_source": {"org": "OrgC", "language": "en"},
+                        "_score": 0.8
+                    }
+                ]
+            },
+            "aggregations": {
+                "global_aggs": {
+                    "language": {
+                        "doc_count": 3,
+                        "values": {
+                            "buckets": [
+                                {"key": "en", "doc_count": 2},
+                                {"key": "fr", "doc_count": 1}
+                            ]
+                        }
+                    },
+                    "org": {
+                        "doc_count": 2,
+                        "values": {
+                            "buckets": [
+                                {"key": "OrgA", "doc_count": 1},
+                                {"key": "OrgC", "doc_count": 1}
+                            ]
+                        }
+                    }
+                }
+            },
+            "took": 2,
+        }
+
+        engine = ElasticSearchEngine(index=TEST_INDEX_NAME)
+
+        result = engine.search(
+            field_dictionary={"language": ["en"]},
+            aggregation_terms={
+                "language": {},
+                "org": {},
+            },
+            is_multivalue=True
+        )
+
+        self.assertEqual(result["total"], 2)
+        self.assertEqual(result["aggs"]["language"]["terms"]["en"], 2)
+        self.assertEqual(result["aggs"]["language"]["terms"]["fr"], 1)
+        self.assertEqual(set(result["aggs"]["org"]["terms"].keys()), {"OrgA", "OrgC"})
+
+        mock_es.search.assert_called_once()
+
+    @patch("search.elastic.Elasticsearch")
+    def test_multivalue_with_empty_filters_uses_match_all(self, mock_elasticsearch_class):
+        """Tests that multivalue aggregation works when no filters are applied."""
+        mock_es = MagicMock()
+        mock_elasticsearch_class.return_value = mock_es
+
+        mock_es.search.return_value = {
+            "hits": {
+                "total": {"value": 3},
+                "max_score": 0.0,
+                "hits": []
+            },
+            "aggregations": {
+                "global_aggs": {
+                    "language": {
+                        "doc_count": 3,
+                        "values": {
+                            "buckets": [
+                                {"key": "en", "doc_count": 2},
+                                {"key": "fr", "doc_count": 1}
+                            ]
+                        }
+                    }
+                }
+            },
+            "took": 2,
+        }
+
+        engine = ElasticSearchEngine(index=TEST_INDEX_NAME)
+
+        result = engine.search(
+            aggregation_terms={"language": {}},
+            field_dictionary={},
+            is_multivalue=True
+        )
+
+        self.assertEqual(result["total"], 3)
+        self.assertEqual(result["aggs"]["language"]["terms"]["en"], 2)
+        self.assertEqual(result["aggs"]["language"]["terms"]["fr"], 1)
+
+    @patch("search.elastic.Elasticsearch")
+    def test_regular_aggregations_do_not_use_global_aggs(self, mock_elasticsearch_class):
+        """Tests that single-value aggregation does not include global_aggs wrapper."""
+        mock_es = MagicMock()
+        mock_elasticsearch_class.return_value = mock_es
+        mock_es.search.return_value = {
+            "hits": {
+                "total": {"value": 1},
+                "max_score": 1.0,
+                "hits": [{
+                    "_source": {"org": "OrgX", "language": "en"},
+                    "_score": 1.0
+                }]
+            },
+            "aggregations": {
+                "language": {
+                    "buckets": [
+                        {"key": "en", "doc_count": 1}
+                    ],
+                    "doc_count_error_upper_bound": 0,
+                    "sum_other_doc_count": 0
+                },
+                "total_language_docs": {"value": 1.0},
+                "total_modes_docs": {"value": 1.0},
+                "total_org_docs": {"value": 1.0}
+            },
+            "took": 2,
+        }
+
+        engine = ElasticSearchEngine(index=TEST_INDEX_NAME)
+
+        result = engine.search(
+            field_dictionary={"language": ["en"]},
+            aggregation_terms={"language": {}},
+            is_multivalue=False
+        )
+
+        self.assertEqual(result["total"], 1)
+        self.assertEqual(result["aggs"]["language"]["terms"]["en"], 1)
+
+        call_args = mock_es.search.call_args[1]
+        search_body = call_args["body"]
+        self.assertIn("aggs", search_body)
+        self.assertIn("language", search_body["aggs"])
+        self.assertNotIn("global_aggs", search_body["aggs"])
+
+    @patch("search.elastic._process_multivalue_aggregations")
+    @patch("search.elastic._process_aggregation_terms")
+    @patch("search.elastic.Elasticsearch")
+    def test_single_value_calls_process_aggregation_terms(
+            self, mock_elasticsearch_class, mock_process_single, mock_process_multi
+    ):
+        """Tests that single-value aggregation calls the standard aggregation processor."""
+        mock_es = MagicMock()
+        mock_elasticsearch_class.return_value = mock_es
+        mock_es.search.return_value = {
+            "hits": {
+                "total": {"value": 1},
+                "max_score": 1.0,
+                "hits": [{
+                    "_source": {"org": "OrgX", "language": "en"},
+                    "_score": 1.0
+                }]
+            },
+            "aggregations": {
+                "language": {
+                    "buckets": [
+                        {"key": "en", "doc_count": 1}
+                    ],
+                    "doc_count_error_upper_bound": 0,
+                    "sum_other_doc_count": 0
+                },
+                "total_language_docs": {"value": 1.0},
+                "total_modes_docs": {"value": 1.0},
+                "total_org_docs": {"value": 1.0}
+            },
+            "took": 2,
+        }
+
+        mock_process_single.return_value = {
+            "language": {"terms": {"fields": "language"}}
+        }
+
+        engine = ElasticSearchEngine(index=TEST_INDEX_NAME)
+
+        aggregation_terms = {"language": {}}
+
+        engine.search(
+            field_dictionary={"language": ["en"]},
+            aggregation_terms=aggregation_terms,
+            is_multivalue=False
+        )
+
+        mock_process_single.assert_called_once_with(aggregation_terms)
+        mock_process_multi.assert_not_called()
+
+    @patch("search.elastic._process_multivalue_aggregations")
+    @patch("search.elastic._process_aggregation_terms")
+    @patch("search.elastic.Elasticsearch")
+    def test_multivalue_calls_process_multivalue_aggregations(
+            self, mock_elasticsearch_class, mock_process_single, mock_process_multi
+    ):
+        """Tests that multivalue aggregation calls the multivalue aggregation processor."""
+        mock_es = MagicMock()
+        mock_elasticsearch_class.return_value = mock_es
+        mock_es.search.return_value = {
+            "hits": {
+                "total": {"value": 1},
+                "max_score": 1.0,
+                "hits": [{
+                    "_source": {"org": "OrgX", "language": "en"},
+                    "_score": 1.0
+                }]
+            },
+            "aggregations": {
+                "language": {
+                    "buckets": [
+                        {"key": "en", "doc_count": 1}
+                    ],
+                    "doc_count_error_upper_bound": 0,
+                    "sum_other_doc_count": 0
+                },
+                "total_language_docs": {"value": 1.0},
+                "total_modes_docs": {"value": 1.0},
+                "total_org_docs": {"value": 1.0}
+            },
+            "took": 2,
+        }
+
+        mock_process_multi.return_value = {
+            "global_aggs": {
+                "language": {"filter": {}, "aggs": {"values": {"terms": {"field": "language"}}}}
+            }
+        }
+
+        engine = ElasticSearchEngine(index=TEST_INDEX_NAME)
+
+        field_dictionary = {"language": ["en"]}
+        aggregation_terms = {"language": {}}
+
+        engine.search(
+            field_dictionary=field_dictionary,
+            aggregation_terms=aggregation_terms,
+            is_multivalue=True
+        )
+
+        mock_process_single.assert_not_called()
+        mock_process_multi.assert_called_once_with(
+            aggregation_terms,
+            field_dictionary
+        )
diff --git a/search/tests/test_meilisearch.py b/search/tests/test_meilisearch.py
index 844854f5..ed5afa9a 100644
--- a/search/tests/test_meilisearch.py
+++ b/search/tests/test_meilisearch.py
@@ -3,7 +3,7 @@
 """
 
 from datetime import datetime
-from unittest.mock import Mock, patch
+from unittest.mock import Mock, patch, MagicMock
 
 import django.test
 from django.utils import timezone
@@ -313,7 +313,7 @@ def test_engine_search(self):
             }
         )
 
-        results = engine.search(
+        result = engine.search(
             query_string="abc",
             field_dictionary={
                 "course": "course-v1:testorg+test1+alpha",
@@ -325,7 +325,7 @@ def test_engine_search(self):
             log_search_params=True,
         )
 
-        engine.meilisearch_index.search.assert_called_with(
+        engine.meilisearch_index.search.assert_called_once_with(
             "abc",
             {
                 "showRankingScore": True,
@@ -338,10 +338,12 @@ def test_engine_search(self):
                 ],
             },
         )
-        assert results == {
-            "aggs": {},
-            "max_score": 0.865,
-            "results": [
+        self.assertEqual(result["max_score"], 0.865)
+        self.assertEqual(result["took"], 0)
+        self.assertEqual(result["total"], 1)
+        self.assertListEqual(
+            result["results"],
+            [
                 {
                     "_id": "course-v1:OpenedX+DemoX+DemoCourse",
                     "_index": "my_index",
@@ -351,10 +353,8 @@ def test_engine_search(self):
                         "pk": "f381d4f1914235c9532576c0861d09b484ade634",
                     },
                 },
-            ],
-            "took": 0,
-            "total": 1,
-        }
+            ]
+        )
 
     def test_engine_remove(self):
         engine = search.meilisearch.MeilisearchEngine(index="my_index")
@@ -366,6 +366,148 @@ def test_engine_remove(self):
         engine.remove(doc_ids=[doc_id])
         engine.meilisearch_index.delete_documents.assert_called_with([doc_pk])
 
+    def test_multivalue_search_uses_or_to_join_rules_within_facet(self):
+        filter_dict = {
+            "language": ["en", "fr"]
+        }
+        rules = search.meilisearch.get_filter_rules(filter_dict)
+
+        self.assertListEqual(rules, ['language = "en" OR language = "fr"'])
+
+    def test_multivalue_search_expands_selected_facet_without_filtering(self):
+        multivalue_distribution = {'en': 1, 'fr': 2}
+
+        engine = search.meilisearch.MeilisearchEngine(index="test_index")
+        engine.meilisearch_index.search = Mock(
+            return_value={
+                'hits': [],
+                'query': '',
+                'processingTimeMs': 0,
+                'limit': 0,
+                'offset': 0,
+                'estimatedTotalHits': 4,
+                'facetDistribution':
+                    {'language': multivalue_distribution},
+                'facetStats': {}
+            }
+        )
+
+        original_filter = [
+            'language = "en" OR language = "fr"',
+            'modes = "audit" OR modes = "honor"',
+            'org = "EDX"',
+        ]
+        selected_facet = 'language'
+        actual_distribution = engine._get_expanded_distribution(
+            '', selected_facet, original_filter
+        )
+        self.assertDictEqual(actual_distribution, multivalue_distribution)
+        (query, opt_params), _ = engine.meilisearch_index.search.call_args
+        self.assertIn(selected_facet, opt_params['facets'])
+        self.assertFalse(any(rule.startswith(f'{selected_facet} = ') for rule in opt_params['filter']))
+
+    def test_multivalue_search_merges_expanded_facet_distributions(self):
+        engine = search.meilisearch.MeilisearchEngine(index='test_index')
+        engine.meilisearch_index.search = Mock(side_effect=[
+            {
+                "hits": [],
+                "query": "",
+                "processingTimeMs": 5,
+                "limit": 20,
+                "offset": 0,
+                "estimatedTotalHits": 0,
+                "facetDistribution": {
+                    "language": {"en": 2},  # Narrowed distribution after selecting a facet value
+                    "org": {"EDX": 2}
+                },
+            },
+            {
+                "hits": [],
+                "facetDistribution": {
+                    "language": {"en": 2, "fr": 1}  # Expanded distribution for multivalue search
+                }
+            }
+        ])
+
+        results = engine.search(
+            query_string='',
+            field_dictionary={'language': ['en']},
+            is_multivalue=True,
+        )
+        aggregations = results["aggs"]
+        self.assertIn("language", aggregations)
+        self.assertIn("org", aggregations)
+        self.assertDictEqual(
+            aggregations["language"]["terms"],
+            {"en": 2, "fr": 1}
+        )
+        self.assertDictEqual(aggregations["org"]["terms"], {"EDX": 2})
+
+    def test_single_value_search_narrows_selected_facet(self):
+        engine = search.meilisearch.MeilisearchEngine(index='test_index')
+        engine.meilisearch_index.search = Mock(side_effect=[
+            {
+                "hits": [],
+                "query": "",
+                "processingTimeMs": 5,
+                "limit": 20,
+                "offset": 0,
+                "estimatedTotalHits": 0,
+                "facetDistribution": {
+                    "language": {"en": 2},
+                    "org": {"EDX": 2}
+                },
+            },
+            {
+                "hits": [],
+                "facetDistribution": {
+                    "language": {"en": 2, "fr": 1}
+                }
+            }
+        ])
+
+        results = engine.search(
+            query_string='',
+            field_dictionary={'language': ['en']},
+            is_multivalue=False,
+        )
+        aggregations = results["aggs"]
+        self.assertIn("language", aggregations)
+        self.assertIn("org", aggregations)
+        self.assertDictEqual(
+            aggregations["language"]["terms"],
+            {"en": 2}
+        )
+        self.assertDictEqual(aggregations["org"]["terms"], {"EDX": 2})
+
+    def test_facet_expansion_not_triggered_if_not_multivalue(self):
+        engine = search.meilisearch.MeilisearchEngine(index="test_index")
+        engine._expand_facet_distibutions = MagicMock()
+        engine.meilisearch_index.search = Mock(
+            return_value={
+                "hits": [],
+                "facetDistribution": {},
+                "estimatedTotalHits": 0,
+                "processingTimeMs": 1,
+            }
+        )
+        engine.search(field_dictionary={"language": "en"}, is_multivalue=False)
+        engine._expand_facet_distibutions.assert_not_called()
+
+    def test_facet_expansion_is_triggered_if_multivalue(self):
+        engine = search.meilisearch.MeilisearchEngine(index="test_index")
+        engine._expand_facet_distibutions = MagicMock()
+        engine.meilisearch_index.search = Mock(
+            return_value={
+                "hits": [],
+                "facetDistribution": {},
+                "estimatedTotalHits": 0,
+                "processingTimeMs": 1,
+            }
+        )
+        engine.search(query_string="demo", field_dictionary={"language": ["en"]}, is_multivalue=True)
+        engine._expand_facet_distibutions.assert_called_once()
+
 
 class UtilitiesTests(django.test.TestCase):
     """
diff --git a/search/tests/utils.py b/search/tests/utils.py
index 5d38d616..8ebd8038 100644
--- a/search/tests/utils.py
+++ b/search/tests/utils.py
@@ -21,9 +21,8 @@ def post_request(body, course_id=None):
     return getattr(response, "status_code", 500), json.loads(getattr(response, "content", None).decode('utf-8'))
 
 
-def post_discovery_request(body):
+def post_discovery_request(body, address='/course_discovery/'):
     """ Helper method to post the request and process the response """
-    address = '/course_discovery/'
     response = Client().post(address, body)
 
     return getattr(response, "status_code", 500), json.loads(getattr(response, "content", None).decode('utf-8'))
diff --git a/search/urls.py b/search/urls.py
index 10439d54..7c2a7e32 100644
--- a/search/urls.py
+++ b/search/urls.py
@@ -12,4 +12,5 @@
     path('', views.do_search, name='do_search'),
     re_path(r'^{}$'.format(COURSE_ID_PATTERN), views.do_search, name='do_search'),
     path('course_discovery/', views.course_discovery, name='course_discovery'),
+    path('v1/course_discovery/', views.course_discovery_multivalue, name='course_discovery_multivalue'),
 ]
diff --git a/search/views.py b/search/views.py
index e9e0bcc4..de78b0c8 100644
--- a/search/views.py
+++ b/search/views.py
@@ -34,10 +34,11 @@ def _process_pagination_values(request):
     return size, from_, page
 
 
-def _process_field_values(request):
+def _process_field_values(request, is_multivalue=False):
     """ Create separate dictionary of supported filter values provided """
+    get_value = request.POST.getlist if is_multivalue else request.POST.get
     return {
-        field_key: request.POST[field_key]
+        field_key: get_value(field_key)
         for field_key in request.POST
         if field_key in course_discovery_filter_fields()
     }
@@ -138,11 +139,24 @@ def do_search(request, course_id=None):
 
 @require_POST
 def course_discovery(request):
+    """ Legacy single-value search endpoint """
+    return _course_discovery(request, is_multivalue=False)
+
+
+@require_POST
+def course_discovery_multivalue(request):
+    """ Main endpoint for multi-value faceted search """
+    return _course_discovery(request, is_multivalue=True)
+
+
+@require_POST
+def _course_discovery(request, is_multivalue=False):
     """
     Search for courses
 
     Args:
         request (required) - django request object
+        is_multivalue (optional) - boolean indicating whether to use multi-value faceted search
 
     Returns:
         http json response with the following fields
@@ -169,7 +183,7 @@ def course_discovery(request):
 
     try:
         size, from_, page = _process_pagination_values(request)
-        field_dictionary = _process_field_values(request)
+        field_dictionary = _process_field_values(request, is_multivalue=is_multivalue)
 
         # Analytics - log search request
         track.emit(
@@ -186,6 +200,7 @@ def course_discovery(request):
             size=size,
             from_=from_,
             field_dictionary=field_dictionary,
+            is_multivalue=is_multivalue,
         )
 
         # Analytics - log search results before sending to browser