XWIKI-23420: Notifications that aren't pre-filtered in time aren't sent by schedulers

xwiki-platform-core/xwiki-platform-eventstream/xwiki-platform-eventstream-api/src/main/java/org/xwiki/eventstream/Event.java

@@ -204,6 +204,13 @@ public interface Event
      */
     String FIELD_REMOTE_OBSERVATION_ID = "observationInstanceId";
 
+    /**
+     * Field storing the prefiltering date information.
+     *
+     * @since 17.10.0
+     */
+    String FIELD_PREFILTERING_DATE = "preFilteringDate";
+
     /** The importance of an event. */
     enum Importance
     {

xwiki-platform-core/xwiki-platform-eventstream/xwiki-platform-eventstream-stores/xwiki-platform-eventstream-store-solr/src/main/java/org/xwiki/eventstream/store/solr/internal/EventsSolrCoreInitializer.java

@@ -19,11 +19,13 @@
  */
 package org.xwiki.eventstream.store.solr.internal;
 
+import javax.inject.Inject;
 import javax.inject.Named;
 import javax.inject.Singleton;
 
 import org.xwiki.component.annotation.Component;
 import org.xwiki.eventstream.Event;
+import org.xwiki.eventstream.store.solr.internal.migration.SolrDocumentMigration171000000;
 import org.xwiki.search.solr.AbstractSolrCoreInitializer;
 import org.xwiki.search.solr.SolrException;
 
@@ -79,6 +81,9 @@ public class EventsSolrCoreInitializer extends AbstractSolrCoreInitializer
      */
     public static final String FIELD_DOCUMENT_INDEX = "document_index";
 
+    @Inject
+    private SolrDocumentMigration171000000 migration171000000;
+
     @Override
     protected long getVersion()
     {
@@ -129,6 +134,10 @@ protected void migrateSchema(long cversion) throws SolrException
         if (cversion < SCHEMA_VERSION_14_7) {
             setStringField(Event.FIELD_REMOTE_OBSERVATION_ID, false, false);
         }
+        if (cversion < SCHEMA_VERSION_17_10) {
+            setPDateField(Event.FIELD_PREFILTERING_DATE, false, false);
+            this.migration171000000.migrateAllDocuments(this.core);
+        }
     }
 
     @Override

xwiki-platform-core/xwiki-platform-eventstream/xwiki-platform-eventstream-stores/xwiki-platform-eventstream-store-solr/src/main/java/org/xwiki/eventstream/store/solr/internal/SolrEventStore.java

@@ -245,6 +245,7 @@ protected Event syncPrefilterEvent(Event event) throws EventStreamException
         this.utils.set(EventsSolrCoreInitializer.SOLR_FIELD_ID, event.getId(), document);
 
         this.utils.setAtomic(SolrUtils.ATOMIC_UPDATE_MODIFIER_SET, Event.FIELD_PREFILTERED, true, document);
+        this.utils.set(Event.FIELD_PREFILTERING_DATE, new Date(), document);
 
         try {
             this.client.add(document);

xwiki-platform-core/xwiki-platform-eventstream/xwiki-platform-eventstream-stores/xwiki-platform-eventstream-store-solr/src/main/java/org/xwiki/eventstream/store/solr/internal/migration/SolrDocumentMigration171000000.java

@@ -0,0 +1,114 @@
+/*
+ * See the NOTICE file distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.xwiki.eventstream.store.solr.internal.migration;
+
+import java.io.IOException;
+import java.util.Date;
+
+import javax.inject.Inject;
+
+import org.apache.solr.client.solrj.SolrQuery;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.response.QueryResponse;
+import org.apache.solr.common.SolrDocument;
+import org.apache.solr.common.SolrDocumentList;
+import org.apache.solr.common.SolrInputDocument;
+import org.slf4j.Logger;
+import org.xwiki.component.annotation.Component;
+import org.xwiki.eventstream.Event;
+import org.xwiki.search.solr.SolrException;
+import org.xwiki.search.solr.SolrUtils;
+import org.xwiki.search.solr.XWikiSolrCore;
+
+import jakarta.inject.Singleton;
+
+/**
+ * Migration of documents after introducing {@link Event#FIELD_PREFILTERING_DATE}.
+ *
+ * @version $Id$
+ * @since 17.10.0
+ */
+@Component(roles = SolrDocumentMigration171000000.class)
+@Singleton
+public class SolrDocumentMigration171000000
+{
+    /**
+     * Number of documents to consider at once for migration.
+     */
+    private static final int BATCH_MIGRATION_SIZE = 100;
+
+    @Inject
+    private SolrUtils solrUtils;
+
+    @Inject
+    private Logger logger;
+
+    /**
+     * Perform migration of documents to fill the field {@link Event#FIELD_PREFILTERING_DATE}.
+     *
+     * @param core the core for which to perform the migration.
+     * @throws SolrException in case of problem when performing the migration.
+     */
+    public void migrateAllDocuments(XWikiSolrCore core) throws SolrException
+    {
+        SolrDocumentList documentList;
+        int startIndex = 0;
+        int totalMigrated = 0;
+        long totalNumber = 0;
+        do {
+            SolrQuery solrQuery = new SolrQuery("*:*")
+                .setStart(startIndex)
+                .setRows(BATCH_MIGRATION_SIZE)
+                .setSort(Event.FIELD_DATE, SolrQuery.ORDER.desc);
+            try {
+                QueryResponse queryResponse = core.getClient().query(solrQuery);
+                documentList = queryResponse.getResults();
+                totalNumber = queryResponse.getResults().getNumFound();
+                if (!documentList.isEmpty()) {
+                    for (SolrDocument solrDocument : documentList) {
+                        String id = this.solrUtils.getId(solrDocument);
+                        Date date = this.solrUtils.get(Event.FIELD_DATE, solrDocument);
+
+                        SolrInputDocument solrInputDocument = new SolrInputDocument();
+                        this.solrUtils.setId(id, solrInputDocument);
+                        this.solrUtils.set(Event.FIELD_PREFILTERING_DATE, date, solrInputDocument);
+                        core.getClient().add(solrInputDocument);
+                    }
+                    startIndex += BATCH_MIGRATION_SIZE;
+                    totalMigrated += documentList.size();
+                    logger.info("[{}] Solr events information migrated on [{}].", totalMigrated, totalNumber);
+                }
+            } catch (SolrServerException | IOException e) {
+                throw new SolrException("Error when performing 171000000 Solr events documents migration", e);
+            }
+        } while (!documentList.isEmpty() && totalMigrated < totalNumber);
+
+        if (totalMigrated > 0) {
+            // We commit when all documents are migrated.
+            try {
+                core.getClient().commit();
+            } catch (SolrServerException | IOException e) {
+                throw new SolrException("Error when committing after performing 171000000 Solr "
+                    + "events documents migration.",
+                    e);
+            }
+        }
+    }
+}

xwiki-platform-core/xwiki-platform-eventstream/xwiki-platform-eventstream-stores/xwiki-platform-eventstream-store-solr/src/main/resources/META-INF/components.txt

@@ -1,3 +1,4 @@
+org.xwiki.eventstream.store.solr.internal.migration.SolrDocumentMigration171000000
 org.xwiki.eventstream.store.solr.internal.EventsSolrCoreInitializer
 org.xwiki.eventstream.store.solr.internal.SolrEventStore
 org.xwiki.eventstream.store.solr.internal.WikiDeletedListener

xwiki-platform-core/xwiki-platform-eventstream/xwiki-platform-eventstream-stores/xwiki-platform-eventstream-store-solr/src/test/java/org/xwiki/eventstream/store/solr/internal/EventStoreTest.java

@@ -55,6 +55,7 @@
 import org.xwiki.eventstream.internal.DefaultEventStatus;
 import org.xwiki.eventstream.query.SimpleEventQuery;
 import org.xwiki.eventstream.query.SortableEventQuery.SortClause.Order;
+import org.xwiki.eventstream.store.solr.internal.migration.SolrDocumentMigration171000000;
 import org.xwiki.model.internal.reference.converter.EntityReferenceConverter;
 import org.xwiki.model.internal.reference.converter.WikiReferenceConverter;
 import org.xwiki.model.reference.DocumentReference;
@@ -165,6 +166,9 @@ public class EventStoreTest
     @MockComponent
     private WikiDescriptorManager wikis;
 
+    @MockComponent
+    private SolrDocumentMigration171000000 migration171000000;
+
     @InjectComponentManager
     private MockitoComponentManager componentManager;
 

xwiki-platform-core/xwiki-platform-notifications/xwiki-platform-notifications-filters/xwiki-platform-notifications-filters-api/src/main/java/org/xwiki/notifications/filters/expression/EventProperty.java

@@ -19,6 +19,8 @@
  */
 package org.xwiki.notifications.filters.expression;
 
+import org.xwiki.stability.Unstable;
+
 /**
  * The several properties you can have in an {@link org.xwiki.eventstream.Event}.
  *
@@ -96,5 +98,12 @@ public enum EventProperty
      * 
      * @since 14.7RC1
      */
-    REMOTE_OBSERVATION_ID
+    REMOTE_OBSERVATION_ID,
+    /**
+     * The date of the prefiltering of the event.
+     *
+     * @since 17.10.0
+     */
+    @Unstable
+    PREFILTERING_DATE
 }

xwiki-platform-core/xwiki-platform-notifications/xwiki-platform-notifications-notifiers/xwiki-platform-notifications-notifiers-default/src/main/java/org/xwiki/notifications/notifiers/internal/email/DefaultPeriodicMimeMessageIterator.java

@@ -83,7 +83,7 @@ protected List<CompositeEvent> retrieveCompositeEventList(DocumentReference user
         notificationParameters.user = user;
         notificationParameters.format = NotificationFormat.EMAIL;
         notificationParameters.expectedCount = Integer.MAX_VALUE / 4;
-        notificationParameters.fromDate = this.lastTrigger;
+        notificationParameters.fromPrefilteringDate = this.lastTrigger;
         notificationParameters.endDateIncluded = false;
         notificationParametersFactory.useUserPreferences(notificationParameters);
         UserReference userReference = this.userReferenceResolver.resolve(user);

xwiki-platform-core/xwiki-platform-notifications/xwiki-platform-notifications-notifiers/xwiki-platform-notifications-notifiers-default/src/test/java/org/xwiki/notifications/notifiers/internal/email/DefaultPeriodicMimeMessageIteratorTest.java

@@ -189,7 +189,7 @@ void test() throws Exception
         notificationParameters.user = userA;
         notificationParameters.format = NotificationFormat.EMAIL;
         notificationParameters.expectedCount = Integer.MAX_VALUE / 4;
-        notificationParameters.fromDate = new Date(0L);
+        notificationParameters.fromPrefilteringDate = new Date(0L);
         notificationParameters.endDateIncluded = false;
 
         when(this.notificationManager.getRawEvents(notificationParameters))
@@ -199,7 +199,7 @@ void test() throws Exception
         notificationParameters2.user = userC;
         notificationParameters2.format = NotificationFormat.EMAIL;
         notificationParameters2.expectedCount = Integer.MAX_VALUE / 4;
-        notificationParameters2.fromDate = new Date(0L);
+        notificationParameters2.fromPrefilteringDate = new Date(0L);
         notificationParameters2.endDateIncluded = false;
 
         when(this.notificationManager.getRawEvents(notificationParameters2))

xwiki-platform-core/xwiki-platform-notifications/xwiki-platform-notifications-sources/src/main/java/org/xwiki/notifications/sources/NotificationParameters.java

@@ -30,6 +30,7 @@
 import org.xwiki.notifications.filters.NotificationFilter;
 import org.xwiki.notifications.filters.NotificationFilterPreference;
 import org.xwiki.notifications.preferences.NotificationPreference;
+import org.xwiki.stability.Unstable;
 import org.xwiki.text.XWikiToStringBuilder;
 
 /**
@@ -105,6 +106,14 @@ public class NotificationParameters
      */
     public String groupingEventTarget = "alert";
 
+    /**
+     * Don't get notifications filtered before that date.
+     *
+     * @since 17.10.0
+     */
+    @Unstable
+    public Date fromPrefilteringDate;
+
     @Override
     public boolean equals(Object o)
     {

xwiki-platform-core/xwiki-platform-notifications/xwiki-platform-notifications-sources/src/main/java/org/xwiki/notifications/sources/internal/ExpressionNodeToEventQueryConverter.java

@@ -87,6 +87,7 @@ public class ExpressionNodeToEventQueryConverter
         PROPERTY_MAPPING.put(EventProperty.TITLE, Event.FIELD_TITLE);
         PROPERTY_MAPPING.put(EventProperty.BODY, Event.FIELD_BODY);
         PROPERTY_MAPPING.put(EventProperty.DOCUMENT_VERSION, Event.FIELD_DOCUMENTVERSION);
+        PROPERTY_MAPPING.put(EventProperty.PREFILTERING_DATE, Event.FIELD_PREFILTERING_DATE);
     }
 
     @Inject

xwiki-platform-core/xwiki-platform-notifications/xwiki-platform-notifications-sources/src/main/java/org/xwiki/notifications/sources/internal/QueryExpressionGenerator.java

@@ -111,6 +111,11 @@ public ExpressionNode generateQueryExpression(NotificationParameters parameters)
                 new GreaterThanNode(new PropertyValueNode(EventProperty.DATE), new DateValueNode(parameters.fromDate));
         }
 
+        if (parameters.fromPrefilteringDate != null) {
+            topNode = new GreaterThanNode(new PropertyValueNode(EventProperty.PREFILTERING_DATE),
+                new DateValueNode(parameters.fromPrefilteringDate));
+        }
+
         // Condition 2: handle other preferences
         AbstractOperatorNode preferencesNode = handleEventPreferences(parameters);
 

xwiki-platform-core/xwiki-platform-search/xwiki-platform-search-solr/xwiki-platform-search-solr-api/src/main/java/org/xwiki/search/solr/AbstractSolrCoreInitializer.java

@@ -144,6 +144,13 @@ public abstract class AbstractSolrCoreInitializer implements SolrCoreInitializer
      */
     public static final long SCHEMA_VERSION_16_7 = 160700000;
 
+    /**
+     * The base schema version for XWiki 17.10.0.
+     *
+     * @since 17.10.0
+     */
+    public static final long SCHEMA_VERSION_17_10 = 171000000;
+
     /**
      * The base schema version.
      */