Elasticsearch

.coderabbit.yaml

@@ -64,6 +64,54 @@ reviews:
         - Ensure domain-layer code invoked by the job (outside the DynamicListAttr exception) remains free of persistence calls.
         - Batch flush operations where practical.
 
+    - path: "src/Domain/Search/**"
+      instructions: &elasticsearch_consistency |
+        You are reviewing code for the MySQL/Elasticsearch dual-write system (write-through DB +
+        Elasticsearch via `SearchIndexDoctrineListener`, with `*ConfigurableReader`/`*HybridReader`
+        classes able to read from either backend, and a retention/purge command that deletes DB rows
+        once confirmed in Elasticsearch). Flag anything that can let the two stores drift apart:
+
+        - ❌ Bulk/raw deletes or updates on a `SearchIndexableInterface` entity's table that bypass
+          Doctrine's entity lifecycle - DQL `->delete()`/`->update()` queries, or DBAL
+          `executeStatement()`/raw SQL - never trigger `SearchIndexDoctrineListener::preRemove/postRemove`
+          (those only fire for `EntityManager::remove()`/`persist()` + `flush()`). Any such bulk
+          operation silently skips the Elasticsearch side. This is legitimate ONLY when it is an
+          explicit, reviewed "delete from MySQL but deliberately keep the Elasticsearch document"
+          retention/purge path (and even then, it must verify the row exists in Elasticsearch via a
+          real ES query *before* deleting - never delete on the assumption that dual-write "should
+          have" succeeded).
+        - ❌ A `SearchIndexableInterface` entity with a foreign key that has a DB-level
+          `onDelete: 'CASCADE'` (`#[ORM\JoinColumn(..., onDelete: 'CASCADE')]`) but no matching
+          Doctrine-level `cascade: ['remove']` on the owning/parent side's association. Without the
+          Doctrine-level cascade, deleting the parent lets the database silently drop child rows that
+          Doctrine never loads/removes individually, so the listener never fires for them and their
+          Elasticsearch documents are orphaned.
+        - ❌ Calls to `ElasticsearchClientInterface::index()`/`delete()`, or construction of
+          `IndexDocumentMessage`, using a hardcoded, reused, or non-monotonic revision instead of the
+          established wall-clock-microseconds pattern (see `SearchIndexDoctrineListener::nextRevision()`).
+          External versioning is what stops a delayed/retried write or delete from clobbering newer
+          state - a wrong revision can let a stale message win.
+        - ❌ New/changed `*ConfigurableReader` implementations that don't correctly gate on
+          `elasticsearch.enabled` and fall back to the Doctrine repository, or that silently mix data
+          from both backends for the same logical query/response without that being the explicit intent.
+        - ⚠️ Any consumer that needs a *complete*/unbounded history or aggregate (not a bounded recent
+          window) reading directly from the Doctrine repository for an entity that has a retention/purge
+          policy configured elsewhere - it will silently see only the retained window once older rows
+          are purged. Point out that it should read through the ES-backed reader/interface instead.
+        - ⚠️ Missing or incomplete tests around the above: a repository method that deletes/updates rows
+          for a `SearchIndexableInterface` entity should have a test proving it does (or intentionally
+          does not) go through the dual-write path, and any new cascade relationship should have a test
+          that actually removes the parent and asserts the child is gone too.
+
+    - path: "src/**/*Elasticsearch*.php"
+      instructions: *elasticsearch_consistency
+
+    - path: "src/**/Service/Search/**"
+      instructions: *elasticsearch_consistency
+
+    - path: "src/Core/Doctrine/SearchIndexDoctrineListener.php"
+      instructions: *elasticsearch_consistency
+
   auto_review:
     enabled: true
     base_branches:

.env.dist

@@ -52,6 +52,20 @@ BOUNCE_IMAP_PURGE_UNPROCESSED=0
 
 # Messenger configuration for asynchronous processing
 MESSENGER_TRANSPORT_DSN=doctrine://default?auto_setup=true
+SEARCH_TRANSPORT_DSN=doctrine://default?queue_name=search_index
+
+# Elasticsearch configuration
+# Set to false to fall back to reading/writing these tables straight from MySQL - no Elasticsearch
+# cluster required. See docs/ElasticsearchSearch.md.
+ELASTICSEARCH_ENABLED=false
+ELASTICSEARCH_HOSTS=http://127.0.0.1:9200
+ELASTICSEARCH_USERNAME=
+ELASTICSEARCH_PASSWORD=
+ELASTICSEARCH_CONNECT_TIMEOUT=2
+ELASTICSEARCH_REQUEST_TIMEOUT=5
+# ISO-8601 duration (e.g. P1M) for how long SubscriberHistory rows are kept in MySQL after being
+# confirmed in Elasticsearch. Empty disables purging. See docs/ElasticsearchSearch.md.
+ELASTICSEARCH_PURGE_SUBSCRIBER_HISTORY_RETENTION=
 
 # A secret key that's used to generate certain security-related tokens
 PHPLIST_SECRET=%s

.env.test

@@ -0,0 +1,3 @@
+PHPLIST_DATABASE_DRIVER=pdo_sqlite
+PHPLIST_DATABASE_PATH=:memory:
+SEARCH_TRANSPORT_DSN=sync://

.github/workflows/ci.yml

@@ -19,6 +19,17 @@ jobs:
         ports:
           - 3306/tcp
         options: --health-cmd="mysqladmin ping" --health-interval=10s --health-timeout=5s --health-retries=3
+      elasticsearch:
+        image: docker.elastic.co/elasticsearch/elasticsearch:8.15.0
+        env:
+          discovery.type: single-node
+          xpack.security.enabled: false
+          ES_JAVA_OPTS: -Xms256m -Xmx256m
+        ports:
+          - 9200/tcp
+        options: >-
+          --health-cmd="curl -sf http://localhost:9200/_cluster/health || exit 1"
+          --health-interval=10s --health-timeout=5s --health-retries=5
     strategy:
       fail-fast: false
       matrix:
@@ -68,13 +79,18 @@ jobs:
           php bin/console doctrine:schema:validate --skip-sync
       - name: Run units tests with phpunit
         run: vendor/bin/phpunit tests/Unit/ --testdox
+      - name: Initialize Elasticsearch indices
+        run: |
+          export ELASTICSEARCH_HOSTS=http://127.0.0.1:${{ job.services.elasticsearch.ports['9200'] }}
+          php bin/console phplist:search:init-indices
       - name: Run integration tests with phpunit
         run: |
           export PHPLIST_DATABASE_NAME=${{ env.DB_DATABASE }}
           export PHPLIST_DATABASE_USER=${{ env.DB_USERNAME }}
           export PHPLIST_DATABASE_PASSWORD=${{ env.DB_PASSWORD }}
           export PHPLIST_DATABASE_PORT=${{ job.services.mysql.ports['3306'] }}
           export PHPLIST_DATABASE_HOST=127.0.0.1
+          export ELASTICSEARCH_HOSTS=http://127.0.0.1:${{ job.services.elasticsearch.ports['9200'] }}
           vendor/bin/phpunit tests/Integration/
       - name: Running the system tests
         run: vendor/bin/phpunit tests/System/ --testdox;

README.md

@@ -57,6 +57,7 @@ this code.
 * [Mailer transports](docs/MailerTransports.md) - configuring Gmail, Amazon SES, Mailchimp, and SendGrid
 * [Asynchronous email sending](docs/AsyncEmailSending.md) - queuing email delivery with Symfony Messenger
 * [Graylog integration](docs/Graylog.md) - centralized log management
+* [Elasticsearch-backed search for big tables](docs/ElasticsearchSearch.md) - dual-write to the database and Elasticsearch, reading from Elasticsearch only
 * [Generating class API docs](PHPDOC.md) - regenerating the phpDocumentor output
 
 ## Running the web server
@@ -201,6 +202,9 @@ To extract translation strings from the source into an XLIFF catalog:
 
 ```bash
 php bin/console translation:extract --force en --format=xlf
+php bin/console messenger:setup-transports
+php bin/console messenger:consume async --limit=1
+php bin/console phplist:search:init-indices
 ```
 
 ## Copyright

composer.json

@@ -89,7 +89,8 @@
         "phpdocumentor/reflection-docblock": "^5.2",
         "guzzlehttp/guzzle": "^7.4.5",
         "symfony/dotenv": "^6.4",
-        "symfony/doctrine-messenger": "^6.4"
+        "symfony/doctrine-messenger": "^6.4",
+        "elasticsearch/elasticsearch": "^8.9"
     },
     "require-dev": {
         "phpunit/phpunit": "^9.5",

config/packages/messenger.yaml

@@ -14,13 +14,30 @@ framework:
                     check_delayed_interval: 60000
                 retry_strategy:
                     max_retries: 3
-                    # milliseconds delay
+                    # millisecond delay
                     delay: 1000
                     multiplier: 2
                     max_delay: 0
 
             failed: 'doctrine://default?queue_name=failed'
 
+            # Dedicated queue (same Doctrine connection/table, distinct queue_name) for Elasticsearch
+            # dual-write. auto_setup is deliberately disabled: the messenger_messages table is already
+            # created by the async_email transport, and no DDL must ever run while a Doctrine ORM
+            # transaction is open (see SearchIndexDoctrineListener, which dispatches from postFlush).
+            # On a fresh install that enables search before ever sending email, run
+            # `bin/console messenger:setup-transports` once.
+            # Configurable so tests can swap in 'sync://' (see .env.test) and index synchronously.
+            async_search:
+                dsn: '%env(SEARCH_TRANSPORT_DSN)%'
+                options:
+                    auto_setup: false
+                retry_strategy:
+                    max_retries: 5
+                    delay: 1000
+                    multiplier: 2
+                    max_delay: 30000
+
         routing:
             # Route your messages to the transports
             'PhpList\Core\Domain\Messaging\Message\AsyncEmailMessage': async_email
@@ -30,4 +47,5 @@ framework:
             'PhpList\Core\Domain\Messaging\Message\CampaignProcessor\CampaignProcessorMessage': async_email
             'PhpList\Core\Domain\Messaging\Message\CampaignProcessor\SyncCampaignProcessorMessage': sync
             'PhpList\Core\Domain\Subscription\Message\DynamicTableMessage': sync
+            'PhpList\Core\Domain\Search\Message\IndexDocumentMessage': async_search
 

config/parameters.yml

@@ -50,6 +50,17 @@ parameters:
 
     # Messenger configuration for asynchronous processing
     app.messenger_transport_dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
+    app.search_transport_dsn: '%env(SEARCH_TRANSPORT_DSN)%'
+
+    # Elasticsearch configuration
+    elasticsearch.enabled: '%env(bool:ELASTICSEARCH_ENABLED)%'
+    elasticsearch.hosts: '%env(csv:ELASTICSEARCH_HOSTS)%'
+    elasticsearch.username: '%env(ELASTICSEARCH_USERNAME)%'
+    elasticsearch.password: '%env(ELASTICSEARCH_PASSWORD)%'
+    elasticsearch.index_prefix: '%env(DATABASE_PREFIX)%'
+    elasticsearch.connect_timeout: '%env(int:ELASTICSEARCH_CONNECT_TIMEOUT)%'
+    elasticsearch.request_timeout: '%env(int:ELASTICSEARCH_REQUEST_TIMEOUT)%'
+    elasticsearch.purge.subscriber_history_retention: '%env(ELASTICSEARCH_PURGE_SUBSCRIBER_HISTORY_RETENTION)%'
 
     # A secret key that's used to generate certain security-related tokens
     secret: '%env(PHPLIST_SECRET)%'

config/services.yml

@@ -55,6 +55,10 @@ services:
     arguments:
       $tablePrefix: '%database_prefix%'
 
+  PhpList\Core\Core\Doctrine\SearchIndexDoctrineListener:
+    arguments:
+      $enabled: '%elasticsearch.enabled%'
+
   HTMLPurifier_Config:
     class: HTMLPurifier_Config
     factory: [ 'HTMLPurifier_Config', 'createDefault' ]

config/services/commands.yml

@@ -12,6 +12,10 @@ services:
     resource: '../../src/Domain/Identity/Command'
     tags: ['console.command']
 
+  PhpList\Core\Domain\Search\Command\:
+    resource: '../../src/Domain/Search/Command'
+    tags: ['console.command']
+
   PhpList\Core\Domain\Messaging\Command\ProcessBouncesCommand:
     arguments:
       $protocolProcessors: !tagged_iterator 'phplist.bounce_protocol_processor'

config/services/elasticsearch.yml

@@ -0,0 +1,69 @@
+services:
+  _defaults:
+    autowire: true
+    autoconfigure: true
+    public: false
+
+  _instanceof:
+    PhpList\Core\Domain\Search\Model\Interfaces\SearchIndexDefinitionInterface:
+      tags: ['phplist.search_index_definition']
+    PhpList\Core\Domain\Search\Model\Interfaces\SearchReindexProviderInterface:
+      tags: ['phplist.search_reindex_provider']
+    PhpList\Core\Domain\Search\Model\Interfaces\SearchPurgeProviderInterface:
+      tags: ['phplist.search_purge_provider']
+
+  Elastic\Elasticsearch\Client:
+    factory: ['PhpList\Core\Domain\Search\Client\ElasticsearchClientFactory', 'create']
+    arguments:
+      $hosts: '%elasticsearch.hosts%'
+      $username: '%elasticsearch.username%'
+      $password: '%elasticsearch.password%'
+      $connectTimeout: '%elasticsearch.connect_timeout%'
+      $requestTimeout: '%elasticsearch.request_timeout%'
+
+  PhpList\Core\Domain\Search\Client\ElasticsearchClientInterface:
+    alias: PhpList\Core\Domain\Search\Client\ElasticsearchClientAdapter
+
+  PhpList\Core\Domain\Search\Client\ElasticsearchClientAdapter: ~
+
+  PhpList\Core\Domain\Search\Service\ElasticsearchIndexerInterface:
+    alias: PhpList\Core\Domain\Search\Service\ElasticsearchIndexer
+
+  PhpList\Core\Domain\Search\Service\ElasticsearchIndexer:
+    arguments:
+      $indexPrefix: '%elasticsearch.index_prefix%'
+
+  PhpList\Core\Domain\Search\Registry\SearchIndexDefinitionRegistry:
+    arguments:
+      $definitions: !tagged_iterator 'phplist.search_index_definition'
+
+  PhpList\Core\Domain\Search\Registry\SearchReindexProviderRegistry:
+    arguments:
+      $providers: !tagged_iterator 'phplist.search_reindex_provider'
+
+  PhpList\Core\Domain\Search\Registry\SearchPurgeProviderRegistry:
+    arguments:
+      $providers: !tagged_iterator 'phplist.search_purge_provider'
+
+  PhpList\Core\Domain\Subscription\Repository\SubscriberHistoryElasticsearchReader:
+    arguments:
+      $indexPrefix: '%elasticsearch.index_prefix%'
+
+  PhpList\Core\Domain\Subscription\Service\Search\:
+    resource: '../../src/Domain/Subscription/Service/Search'
+
+  PhpList\Core\Domain\Subscription\Service\Search\SubscriberHistoryPurgeProvider:
+    arguments:
+      $indexPrefix: '%elasticsearch.index_prefix%'
+      $retentionPeriod: '%elasticsearch.purge.subscriber_history_retention%'
+
+  PhpList\Core\Domain\Messaging\Repository\UserMessageBounceElasticsearchReader:
+    arguments:
+      $indexPrefix: '%elasticsearch.index_prefix%'
+
+  PhpList\Core\Domain\Messaging\Repository\UserMessageBounceElasticsearchHybridReader:
+    arguments:
+      $indexPrefix: '%elasticsearch.index_prefix%'
+
+  PhpList\Core\Domain\Messaging\Service\Search\:
+    resource: '../../src/Domain/Messaging/Service/Search'

config/services/messenger.yml

@@ -11,6 +11,12 @@ services:
         resource: '../../src/Domain/Subscription/MessageHandler'
         tags: [ 'messenger.message_handler' ]
 
+    # Register Search message handlers (e.g., IndexDocumentMessageHandler)
+    PhpList\Core\Domain\Search\MessageHandler\:
+        autowire: true
+        resource: '../../src/Domain/Search/MessageHandler'
+        tags: [ 'messenger.message_handler' ]
+
     PhpList\Core\Domain\Messaging\MessageHandler\CampaignProcessor\CampaignProcessorMessageHandler:
         autowire: true
         autoconfigure: true

config/services/repositories.yml

@@ -83,6 +83,16 @@ services:
     parent: PhpList\Core\Domain\Common\Repository\AbstractRepository
     arguments:
       - PhpList\Core\Domain\Subscription\Model\SubscriberHistory
+
+  # Reads for SubscriberHistoryManager/SubscriberManager go through this configurable reader, which
+  # picks Elasticsearch or the database based on elasticsearch.enabled (ELASTICSEARCH_ENABLED) - see
+  # SubscriberHistoryConfigurableReader and docs/ElasticsearchSearch.md.
+  PhpList\Core\Domain\Subscription\Repository\SubscriberHistoryConfigurableReader:
+    autowire: true
+    arguments:
+      $elasticsearchEnabled: '%elasticsearch.enabled%'
+  PhpList\Core\Domain\Subscription\Repository\Interfaces\SubscriberHistoryReaderInterface:
+    alias: PhpList\Core\Domain\Subscription\Repository\SubscriberHistoryConfigurableReader
   PhpList\Core\Domain\Subscription\Repository\UserBlacklistRepository:
     parent: PhpList\Core\Domain\Common\Repository\AbstractRepository
     arguments:
@@ -117,6 +127,25 @@ services:
     parent: PhpList\Core\Domain\Common\Repository\AbstractRepository
     arguments:
       - PhpList\Core\Domain\Messaging\Model\UserMessageBounce
+
+  # Reads for UserMessageBounceReaderInterface consumers go through this configurable reader, which
+  # picks Elasticsearch or the database based on elasticsearch.enabled (ELASTICSEARCH_ENABLED) - see
+  # UserMessageBounceConfigurableReader and docs/ElasticsearchSearch.md.
+  PhpList\Core\Domain\Messaging\Repository\UserMessageBounceConfigurableReader:
+    autowire: true
+    arguments:
+      $elasticsearchEnabled: '%elasticsearch.enabled%'
+  PhpList\Core\Domain\Messaging\Repository\Interfaces\UserMessageBounceReaderInterface:
+    alias: PhpList\Core\Domain\Messaging\Repository\UserMessageBounceConfigurableReader
+
+  # Same configurable swap for the reporting queries (getListBounceTotals/getCampaignBounceTotals)
+  # that join bounce data against Subscriber/Message.
+  PhpList\Core\Domain\Messaging\Repository\UserMessageBounceReportConfigurableReader:
+    autowire: true
+    arguments:
+      $elasticsearchEnabled: '%elasticsearch.enabled%'
+  PhpList\Core\Domain\Messaging\Repository\Interfaces\UserMessageBounceReportReaderInterface:
+    alias: PhpList\Core\Domain\Messaging\Repository\UserMessageBounceReportConfigurableReader
   PhpList\Core\Domain\Messaging\Repository\UserMessageForwardRepository:
     parent: PhpList\Core\Domain\Common\Repository\AbstractRepository
     arguments: