Merge pull request #3351 from akvo/issue/3236-purge-year-old-messages

Issue/3236 purge year old messages

GAE/src/com/gallatinsystems/framework/dao/BaseDAO.java

@@ -49,6 +49,8 @@
 import com.gallatinsystems.user.domain.User;
 import com.gallatinsystems.user.domain.UserAuthorization;
 import com.google.appengine.api.datastore.Cursor;
+import com.google.appengine.api.datastore.DatastoreService;
+import com.google.appengine.api.datastore.DatastoreServiceFactory;
 import com.google.appengine.api.datastore.Key;
 import com.google.appengine.api.datastore.KeyFactory;
 
@@ -635,6 +637,25 @@ public <E extends BaseDomain> void delete(Collection<E> obj) {
         pm.deletePersistentAll(obj);
     }
 
+    /**
+     * deletes an object from the db, by key
+     *
+     * @param <E>
+     * @param obj
+     */
+    public <E extends BaseDomain> void deleteByKey(Key k) {
+        DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
+        datastore.delete(k);
+    }
+
+    /**
+     * deletes a list of objects, by key, in a single datastore interaction
+     */
+    public <E extends BaseDomain> void deleteByKeys(Collection<Key> kList) {
+        DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
+        datastore.delete(kList);
+    }
+
     /**
      * utility method to form a hash map of query parameters using an equality operator
      *

GAE/src/com/gallatinsystems/messaging/dao/MessageDao.java → GAE/src/org/akvo/flow/dao/MessageDao.java

@@ -1,5 +1,5 @@
 /*
- *  Copyright (C) 2010-2012 Stichting Akvo (Akvo Foundation)
+ *  Copyright (C) 2010-2012, 2019 Stichting Akvo (Akvo Foundation)
  *
  *  This file is part of Akvo FLOW.
  *
@@ -14,22 +14,31 @@
  *  The full license text can also be seen at <http://www.gnu.org/licenses/agpl.html>.
  */
 
-package com.gallatinsystems.messaging.dao;
+package org.akvo.flow.dao;
 
+import java.util.ArrayList;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 import javax.jdo.PersistenceManager;
 
+import org.akvo.flow.domain.Message;
+
 import com.gallatinsystems.framework.dao.BaseDAO;
 import com.gallatinsystems.framework.servlet.PersistenceFilter;
-import com.gallatinsystems.messaging.domain.Message;
+import com.google.appengine.api.datastore.DatastoreService;
+import com.google.appengine.api.datastore.DatastoreServiceFactory;
+import com.google.appengine.api.datastore.Entity;
+import com.google.appengine.api.datastore.Key;
+import com.google.appengine.api.datastore.PreparedQuery;
+import com.google.appengine.api.datastore.Query;
+import com.google.appengine.api.datastore.Query.FilterOperator;
 
 /**
  * Data access object for manipulating Message objects
- * 
- * @author Christopher Fagiani
+ *
  */
 public class MessageDao extends BaseDAO<Message> {
 
@@ -39,7 +48,7 @@ public MessageDao() {
 
     /**
      * lists all messages
-     * 
+     *
      * @param about - optional subject
      * @param id - optional ID
      * @param cursor - cursor string
@@ -68,4 +77,24 @@ public List<Message> listBySubject(String about, Long id, String cursor) {
         return results;
     }
 
+    /**
+     * lists keys of messages older than a specific date
+     */
+    public List<Key> listKeysCreatedBefore(Date beforeDate) {
+        DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
+        // The Query interface assembles a query
+        com.google.appengine.api.datastore.Query q =
+                new com.google.appengine.api.datastore.Query("Message");
+        q.setKeysOnly();
+        q.setFilter(new Query.FilterPredicate(
+                "createdDateTime", FilterOperator.LESS_THAN_OR_EQUAL, beforeDate));
+        PreparedQuery pq = datastore.prepare(q);
+        List<Key> result = new ArrayList<>();
+        for (Entity e: pq.asIterable()) {
+            result.add(e.getKey());
+        }
+        return result;
+    }
+
+
 }

GAE/src/com/gallatinsystems/messaging/domain/Message.java → GAE/src/org/akvo/flow/domain/Message.java

@@ -14,7 +14,7 @@
  *  The full license text can also be seen at <http://www.gnu.org/licenses/agpl.html>.
  */
 
-package com.gallatinsystems.messaging.domain;
+package org.akvo.flow.domain;
 
 import javax.jdo.annotations.PersistenceCapable;
 

GAE/src/org/waterforpeople/mapping/app/web/rest/MessageRestService.java → GAE/src/org/akvo/flow/rest/MessageRestService.java

@@ -14,13 +14,16 @@
  *  The full license text can also be seen at <http://www.gnu.org/licenses/agpl.html>.
  */
 
-package org.waterforpeople.mapping.app.web.rest;
+package org.akvo.flow.rest;
 
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import org.akvo.flow.dao.MessageDao;
+import org.akvo.flow.domain.Message;
+import org.akvo.flow.rest.dto.MessageDto;
 import org.springframework.stereotype.Controller;
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -30,10 +33,6 @@
 import org.waterforpeople.mapping.app.util.DtoMarshaller;
 import org.waterforpeople.mapping.app.web.rest.dto.RestStatusDto;
 
-import com.gallatinsystems.messaging.app.gwt.client.MessageDto;
-import com.gallatinsystems.messaging.dao.MessageDao;
-import com.gallatinsystems.messaging.domain.Message;
-
 @Controller
 @RequestMapping("/messages")
 public class MessageRestService {

GAE/src/com/gallatinsystems/messaging/app/gwt/client/MessageDto.java → GAE/src/org/akvo/flow/rest/dto/MessageDto.java

@@ -14,7 +14,7 @@
  *  The full license text can also be seen at <http://www.gnu.org/licenses/agpl.html>.
  */
 
-package com.gallatinsystems.messaging.app.gwt.client;
+package org.akvo.flow.rest.dto;
 
 import java.util.Date;
 

GAE/src/org/waterforpeople/mapping/app/web/CronCommanderServlet.java

@@ -32,6 +32,7 @@
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 
+import org.akvo.flow.dao.MessageDao;
 import org.akvo.flow.dao.ReportDao;
 import org.akvo.flow.dao.SurveyAssignmentDao;
 import org.akvo.flow.domain.persistent.Report;
@@ -94,6 +95,8 @@ public void doGet(HttpServletRequest req, HttpServletResponse resp)
             extractImageFileGeotags();
         } else if ("purgeReportRecords".equals(action)) {
             purgeReportRecords();
+        } else if ("purgeOldMessages".equals(action)) {
+            purgeOldMessages();
         }
     }
 
@@ -241,18 +244,18 @@ private void extractImageFileGeotags() {
         deadline.add(Calendar.MONTH, ONE_MONTH_AGO);
         log.info("Starting scan for image answers, newer than: " + deadline.getTime());
         QuestionAnswerStoreDao qaDao = new QuestionAnswerStoreDao();
-        String cursor = null;
-        List<QuestionAnswerStore> dfjqList;
+        String cursor = "";
         int json = 0;
         int nonjson = 0;
         Media media;
 
         do {
-            dfjqList = qaDao.listByTypeAndDate("IMAGE", null, deadline.getTime(), cursor, 1000);
-            if (dfjqList.size() == 0) break; //no more answers
+            List<QuestionAnswerStore> qaList = qaDao.listByTypeAndDate("IMAGE", null, deadline.getTime(), cursor, 1000);
+            if (qaList == null || qaList.size() == 0) break; //no more answers
+            cursor = QuestionAnswerStoreDao.getCursor(qaList);
 
             //loop over this batch
-            for (QuestionAnswerStore item : dfjqList) {
+            for (QuestionAnswerStore item : qaList) {
                 boolean forceSave = false;
                 String v = item.getValue();
                 log.fine(String.format(" Old IMAGE value '%s'", v));
@@ -397,4 +400,18 @@ InputStream fetchImageFileFromS3(String filename) {
         }
     }
 
+    /**
+     * scans for and deletes Message entries that are more than one year old
+     */
+    private void purgeOldMessages() {
+        Calendar deadline = Calendar.getInstance();
+        deadline.add(Calendar.YEAR, ONE_YEAR_AGO);
+        log.info("Starting scan for Message entries older than: " + deadline.getTime());
+        MessageDao messageDao = new MessageDao();
+        List<Key> purgable = messageDao.listKeysCreatedBefore(deadline.getTime());
+        log.fine("Deleting " + purgable.size() + " old Message entries");
+        messageDao.deleteByKeys(purgable);
+    }
+
+
 }

GAE/src/org/waterforpeople/mapping/app/web/DataProcessorRestServlet.java

@@ -39,7 +39,9 @@
 import net.sf.jsr107cache.CacheFactory;
 import net.sf.jsr107cache.CacheManager;
 
+import org.akvo.flow.dao.MessageDao;
 import org.akvo.flow.domain.DataUtils;
+import org.akvo.flow.domain.Message;
 import org.waterforpeople.mapping.analytics.dao.SurveyQuestionSummaryDao;
 import org.waterforpeople.mapping.analytics.domain.SurveyQuestionSummary;
 import org.waterforpeople.mapping.app.web.dto.DataProcessorRequest;
@@ -54,8 +56,6 @@
 import com.gallatinsystems.framework.rest.RestRequest;
 import com.gallatinsystems.framework.rest.RestResponse;
 import com.gallatinsystems.framework.servlet.PersistenceFilter;
-import com.gallatinsystems.messaging.dao.MessageDao;
-import com.gallatinsystems.messaging.domain.Message;
 import com.gallatinsystems.operations.dao.ProcessingStatusDao;
 import com.gallatinsystems.operations.domain.ProcessingStatus;
 import com.gallatinsystems.survey.dao.CascadeNodeDao;

GAE/src/org/waterforpeople/mapping/app/web/ProcessorServlet.java

@@ -20,15 +20,16 @@
 import com.gallatinsystems.device.dao.DeviceFileJobQueueDAO;
 import com.gallatinsystems.device.domain.Device;
 import com.gallatinsystems.device.domain.DeviceFileJobQueue;
-import com.gallatinsystems.messaging.dao.MessageDao;
-import com.gallatinsystems.messaging.domain.Message;
 import com.gallatinsystems.survey.dao.CascadeResourceDao;
 import com.gallatinsystems.survey.dao.SurveyDAO;
 import com.gallatinsystems.survey.domain.CascadeResource;
 import com.gallatinsystems.survey.domain.CascadeResource.Status;
 import com.google.appengine.api.taskqueue.Queue;
 import com.google.appengine.api.taskqueue.QueueFactory;
 import com.google.appengine.api.taskqueue.TaskOptions;
+
+import org.akvo.flow.dao.MessageDao;
+import org.akvo.flow.domain.Message;
 import org.apache.commons.lang.StringUtils;
 import org.waterforpeople.mapping.app.web.dto.TaskRequest;
 

GAE/src/org/waterforpeople/mapping/app/web/RawDataRestServlet.java

@@ -27,6 +27,8 @@
 
 import javax.servlet.http.HttpServletRequest;
 
+import org.akvo.flow.dao.MessageDao;
+import org.akvo.flow.domain.Message;
 import org.waterforpeople.mapping.app.gwt.client.survey.QuestionDto.QuestionType;
 import org.waterforpeople.mapping.app.web.dto.DataProcessorRequest;
 import org.waterforpeople.mapping.app.web.dto.RawDataImportRequest;
@@ -39,8 +41,6 @@
 import com.gallatinsystems.framework.rest.AbstractRestApiServlet;
 import com.gallatinsystems.framework.rest.RestRequest;
 import com.gallatinsystems.framework.rest.RestResponse;
-import com.gallatinsystems.messaging.dao.MessageDao;
-import com.gallatinsystems.messaging.domain.Message;
 import com.gallatinsystems.survey.dao.QuestionDao;
 import com.gallatinsystems.survey.dao.SurveyDAO;
 import com.gallatinsystems.survey.dao.SurveyGroupDAO;

GAE/src/org/waterforpeople/mapping/app/web/SurveyAssemblyServlet.java

@@ -22,8 +22,6 @@
 import com.gallatinsystems.framework.rest.AbstractRestApiServlet;
 import com.gallatinsystems.framework.rest.RestRequest;
 import com.gallatinsystems.framework.rest.RestResponse;
-import com.gallatinsystems.messaging.dao.MessageDao;
-import com.gallatinsystems.messaging.domain.Message;
 import com.gallatinsystems.operations.dao.ProcessingStatusDao;
 import com.gallatinsystems.operations.domain.ProcessingStatus;
 import com.gallatinsystems.survey.dao.*;
@@ -34,6 +32,8 @@
 import com.google.appengine.api.taskqueue.TaskOptions;
 import com.google.appengine.api.utils.SystemProperty;
 
+import org.akvo.flow.dao.MessageDao;
+import org.akvo.flow.domain.Message;
 import org.akvo.flow.xml.PublishedForm;
 import org.akvo.flow.xml.XmlForm;
 import org.apache.log4j.Logger;

GAE/src/org/waterforpeople/mapping/app/web/TaskServlet.java

@@ -39,6 +39,8 @@
 
 import javax.servlet.http.HttpServletRequest;
 
+import org.akvo.flow.dao.MessageDao;
+import org.akvo.flow.domain.Message;
 import org.apache.commons.io.IOUtils;
 import org.waterforpeople.mapping.app.web.dto.TaskRequest;
 import org.waterforpeople.mapping.dao.DeviceFilesDao;
@@ -57,8 +59,6 @@
 import com.gallatinsystems.framework.rest.AbstractRestApiServlet;
 import com.gallatinsystems.framework.rest.RestRequest;
 import com.gallatinsystems.framework.rest.RestResponse;
-import com.gallatinsystems.messaging.dao.MessageDao;
-import com.gallatinsystems.messaging.domain.Message;
 import com.gallatinsystems.survey.dao.SurveyDAO;
 import com.gallatinsystems.survey.dao.SurveyUtils;
 import com.gallatinsystems.survey.domain.Survey;

GAE/war/WEB-INF/cron.xml

@@ -36,4 +36,9 @@
         <description>Extract geotags from legacy image answers</description>
         <schedule>every 24 hours</schedule>
     </cron>
+    <cron>
+        <url>/cron/croncommander?action=purgeOldMessages</url>
+        <description>Purge old messages</description>
+        <schedule>every monday at 00:15</schedule>
+    </cron>
 </cronentries>

GAE/war/WEB-INF/rest-servlet.xml

@@ -30,7 +30,7 @@
     <bean name="/data_point_assignments" class="org.akvo.flow.rest.DataPointAssignmentRestService" lazy-init="true" />
     <bean name="/device_groups" class="org.waterforpeople.mapping.app.web.rest.DeviceGroupRestService" lazy-init="true" />
     <bean name="/devices" class="org.waterforpeople.mapping.app.web.rest.DeviceRestService" lazy-init="true" />
-    <bean name="/messages" class="org.waterforpeople.mapping.app.web.rest.MessageRestService" lazy-init="true" />
+    <bean name="/messages" class="org.akvo.flow.rest.MessageRestService" lazy-init="true" />
     <bean name="/notification_subscriptions" class="org.waterforpeople.mapping.app.web.rest.NotificationSubscriptionRestService" lazy-init="true" />
     <bean name="/placemarks" class="org.waterforpeople.mapping.app.web.rest.PlacemarkRestService" lazy-init="true" />
     <bean name="/question_answers" class="org.waterforpeople.mapping.app.web.rest.QuestionAnswerRestService" lazy-init="true" />