Delete notifications from index after 90 days, fixes alephdata#1746.

Makefile

@@ -24,7 +24,7 @@ api: services
 	$(COMPOSE) up --abort-on-container-exit api
 
 web: services
-	$(COMPOSE) up --abort-on-container-exit api ui
+	$(COMPOSE) up api ui
 
 worker: services
 	$(COMPOSE) run --rm app aleph worker

aleph/index/notifications.py

@@ -53,7 +53,7 @@ def index_notification(event, actor_id, params, channels, sync=False):
     return index_safe(index, id_, data, sync=sync)
 
 
-def delete_notifications(channel, sync=False):
-    """Delete entities from a collection."""
-    query = {"bool": {"filter": [{"term": {"channels": channel}}]}}
+def delete_notifications(filter_, sync=False):
+    """Delete notifications from a specific channel."""
+    query = {"bool": {"filter": [filter_]}}
     query_delete(notifications_index(), query, sync=sync)

aleph/logic/notifications.py

@@ -11,13 +11,8 @@
 from aleph.model import Event, Events
 from aleph.logic.mail import email_role
 from aleph.logic.html import html_link
-from aleph.logic.util import (
-    collection_url,
-    entity_url,
-    ui_url,
-    entityset_url,
-    archive_url,
-)
+from aleph.logic.util import collection_url, entity_url
+from aleph.logic.util import entityset_url, archive_url, ui_url
 from aleph.index.notifications import index_notification, delete_notifications
 from aleph.index.notifications import notifications_index
 from aleph.index.util import unpack_result
@@ -44,9 +39,18 @@ def publish(event, actor_id=None, params=None, channels=None):
     index_notification(event, actor_id, params, channels)
 
 
+def delete_old_notifications(sync=False):
+    """Delete out-dated notifications from the index."""
+    cutoff = datetime.utcnow() - settings.NOTIFICATIONS_DELETE
+    filter_ = {"range": {"created_at": {"lt": cutoff}}}
+    log.debug("Deleting old notifications before: %r", cutoff)
+    delete_notifications(filter_, sync=sync)
+
+
 def flush_notifications(obj, clazz=None, sync=False):
     """Delete all notifications in a given channel."""
-    delete_notifications(channel_tag(obj, clazz=clazz), sync=sync)
+    filter_ = {"term": {"channels": channel_tag(obj, clazz=clazz)}}
+    delete_notifications(filter_, sync=sync)
 
 
 def get_role_channels(role):

aleph/settings.py

@@ -95,6 +95,9 @@
 ROLE_INACTIVE = env.to_int("ALEPH_ROLE_INACTIVE", 6 * 30)
 ROLE_INACTIVE = timedelta(days=ROLE_INACTIVE)
 
+# Delete notifications after N days.
+NOTIFICATIONS_DELETE = env.to_int("ALEPH_NOTIFICATIONS_DELETE", 3 * 30)
+NOTIFICATIONS_DELETE = timedelta(days=NOTIFICATIONS_DELETE)
 
 ###############################################################################
 # Content processing options

aleph/worker.py

@@ -22,7 +22,7 @@
 from aleph.logic.alerts import check_alerts
 from aleph.logic.collections import reingest_collection, reindex_collection
 from aleph.logic.collections import compute_collections, refresh_collection
-from aleph.logic.notifications import generate_digest
+from aleph.logic.notifications import generate_digest, delete_old_notifications
 from aleph.logic.roles import update_roles
 from aleph.logic.export import delete_expired_exports, export_entities
 from aleph.logic.processing import index_many
@@ -90,6 +90,7 @@ def periodic(self):
                 check_alerts()
                 generate_digest()
                 delete_expired_exports()
+                delete_old_notifications()
 
     def dispatch_task(self, task):
         collection = get_dataset_collection_id(task.job.dataset.name)