normalise reaping query

synapse/storage/monthly_active_users.py

@@ -7,6 +7,7 @@ class MonthlyActiveUsersStore(SQLBaseStore):
     def __init__(self, hs):
         super(MonthlyActiveUsersStore, self).__init__(None, hs)
         self._clock = hs.get_clock()
+        self.max_mau_value = hs.config.max_mau_value
 
     def reap_monthly_active_users(self):
         """
@@ -19,8 +20,42 @@ class MonthlyActiveUsersStore(SQLBaseStore):
             thirty_days_ago = (
                 int(self._clock.time_msec()) - (1000 * 60 * 60 * 24 * 30)
             )
-            sql = "DELETE FROM monthly_active_users WHERE timestamp < ?"
+
-            txn.execute(sql, (thirty_days_ago,))
+            # Query deletes the union of users that have either:
+            #     * not visited in the last 30 days
+            #     * exceeded the total max_mau_value threshold. Where there is
+            #       an excess, more recent users are favoured - this is to cover
+            #       the case where the limit has been step change reduced.
+            #
+            sql = """
+            DELETE FROM monthly_active_users
+            WHERE user_id
+            IN (
+                SELECT * FROM (
+                    SELECT monthly_active_users.user_id
+                    FROM monthly_active_users
+                    LEFT JOIN (
+                        SELECT user_id, max(last_seen) AS last_seen
+                        FROM user_ips
+                        GROUP BY user_id
+                    ) AS uip ON uip.user_id=monthly_active_users.user_id
+                    ORDER BY uip.last_seen desc LIMIT -1 OFFSET ?
+                )
+                UNION
+                SELECT * FROM (
+                    SELECT monthly_active_users.user_id
+                    FROM monthly_active_users
+                    LEFT JOIN (
+                        SELECT user_id, max(last_seen) AS last_seen
+                        FROM user_ips
+                        GROUP BY user_id
+                    ) AS uip  ON uip.user_id=monthly_active_users.user_id
+                    WHERE uip.last_seen < ?
+                )
+            )
+            """
+
+            txn.execute(sql, (self.max_mau_value, thirty_days_ago,))
 
         return self.runInteraction("reap_monthly_active_users", _reap_users)
 
@@ -45,7 +80,7 @@ class MonthlyActiveUsersStore(SQLBaseStore):
                 user_id (str): user to add/update
         """
         return self._simple_insert(
-            desc="upsert_monthly_active_user",
+            desc="insert_monthly_active_user",
             table="monthly_active_users",
             values={
                 "user_id": user_id,