Add per channel and per user max server_rev tracking.

Author: rtibbles

contentcuration/contentcuration/frontend/shared/data/constants.js

@@ -62,3 +62,5 @@ export const CURRENT_USER = 'CURRENT_USER';
 export const ACTIVE_CHANNELS = 'ACTIVE_CHANNELS';
 
 export const CHANNEL_SYNC_KEEP_ALIVE_INTERVAL = 300 * 1000;
+
+export const MAX_REV_KEY = 'max_rev';

contentcuration/contentcuration/frontend/shared/data/resources.js

@@ -26,6 +26,7 @@ import {
   CURRENT_USER,
   ACTIVE_CHANNELS,
   CHANNEL_SYNC_KEEP_ALIVE_INTERVAL,
+  MAX_REV_KEY,
 } from './constants';
 import applyChanges, { applyMods, collectChanges } from './applyRemoteChanges';
 import mergeAllChanges from './mergeChanges';
@@ -889,10 +890,14 @@ export const Session = new IndexedDBResource({
       ? this.updateSession({ [`${ACTIVE_CHANNELS}.${channelId}`]: Date.now() })
       : null;
   },
-  setChannelScope() {
+  async setChannelScope() {
     const channelId = (window.CHANNEL_EDIT_GLOBAL || {}).channel_id || null;
     if (channelId) {
       channelScope.id = channelId;
+      const channelRev = (window.CHANNEL_EDIT_GLOBAL || {}).channel_rev || 0;
+      await this.updateSession({
+        [`${MAX_REV_KEY}.${window.CHANNEL_EDIT_GLOBAL.channel_id}`]: channelRev,
+      });
       this.channelSyncKeepAlive(channelId);
       setInterval(() => this.channelSyncKeepAlive(channelId), CHANNEL_SYNC_KEEP_ALIVE_INTERVAL);
       window.addEventListener('focus', () => this.channelSyncKeepAlive(channelId));

contentcuration/contentcuration/frontend/shared/data/serverSync.js

@@ -1,6 +1,8 @@
 import debounce from 'lodash/debounce';
 import get from 'lodash/get';
 import pick from 'lodash/pick';
+import orderBy from 'lodash/orderBy';
+import uniq from 'lodash/uniq';
 import applyChanges from './applyRemoteChanges';
 import { hasActiveLocks, cleanupLocks } from './changes';
 import {
@@ -10,6 +12,7 @@ import {
   IGNORED_SOURCE,
   CHANNEL_SYNC_KEEP_ALIVE_INTERVAL,
   ACTIVE_CHANNELS,
+  MAX_REV_KEY,
 } from './constants';
 import db from './db';
 import mergeAllChanges from './mergeChanges';
@@ -22,6 +25,10 @@ import urls from 'shared/urls';
 // change being registered, sync changes!
 const SYNC_IF_NO_CHANGES_FOR = 2;
 
+// When this many seconds pass, repoll the backend to
+// check for any updates to active channels, or the user and sync any current changes
+const SYNC_POLL_INTERVAL = 5;
+
 // Flag to check if a sync is currently active.
 let syncActive = false;
 
@@ -142,15 +149,32 @@ function handleSuccesses(response) {
   const successes = get(response, ['data', 'successes'], []);
   if (successes.length) {
     return db[CHANGES_TABLE].where('server_rev')
-      .anyOf(successes)
+      .anyOf(successes.map(c => c.server_rev))
       .delete();
   }
   return Promise.resolve();
 }
 
-function handleMaxRev(response) {
-  const max_rev = response.data.max_rev;
-  return Session.updateSession({ max_rev });
+function handleMaxRevs(response, userId) {
+  const allChanges = orderBy(
+    get(response, ['data', 'changes'], [])
+      .concat(get(response, ['data', 'errors'], []))
+      .concat(get(response, ['data', 'successes'], [])),
+    'server_rev',
+    'desc'
+  );
+  const channelIds = uniq(allChanges.map(c => c.channel_id)).filter(Boolean);
+  const maxRevs = {};
+  for (let channelId of channelIds) {
+    maxRevs[`${MAX_REV_KEY}.${channelId}`] = allChanges.find(
+      c => c.channel_id === channelId
+    ).server_rev;
+  }
+  const lastUserChange = allChanges.find(c => c.user_id === userId);
+  if (lastUserChange) {
+    maxRevs.user_rev = lastUserChange.server_rev;
+  }
+  return Session.updateSession(maxRevs);
 }
 
 async function syncChanges() {
@@ -169,26 +193,29 @@ async function syncChanges() {
   // might have come in during processing - leave them for the next cycle.
   // This is the primary key of the change objects, so the collection is ordered by this
   // by default - if we just grab the last object, we can get the key from there.
-  const [lastChange, earliestServerChange, user] = await Promise.all([
+  const [lastChange, user] = await Promise.all([
     db[CHANGES_TABLE].orderBy('rev').last(),
-    db[CHANGES_TABLE].orderBy('server_rev').first(),
     Session.getSession(),
   ]);
   if (!user) {
     // If not logged in, nothing to do.
     return;
   }
+
   const now = Date.now();
-  const currentUser = await Session.getSession();
-  const channel_ids = Object.entries(currentUser[ACTIVE_CHANNELS])
+  const channel_ids = Object.entries(user[ACTIVE_CHANNELS])
     .filter(([id, time]) => id && time > now - CHANNEL_SYNC_KEEP_ALIVE_INTERVAL)
     .map(([id]) => id);
+  const channel_revs = {};
+  console.log(user);
+  for (let channelId of channel_ids) {
+    channel_revs[channelId] = get(user, [MAX_REV_KEY, channelId], 0);
+  }
+
   const requestPayload = {
     changes: [],
-    channel_ids,
-    // Last rev to send to the server is either the earliest change we are still seeking
-    // confirmation on, or the current max_rev that we have synced to the frontend.
-    last_rev: (earliestServerChange && earliestServerChange.server_rev) || user.max_rev,
+    channel_revs,
+    user_rev: user.user_rev || 0,
   };
 
   if (lastChange) {
@@ -216,7 +243,7 @@ async function syncChanges() {
     //   "allowed": [],
     //   "changes": [],
     //   "errors": [],
-    //   "successess": [],
+    //   "successes": [],
     // }
     const response = await client.post(urls['sync'](), requestPayload);
     try {
@@ -226,7 +253,7 @@ async function syncChanges() {
         handleReturnedChanges(response),
         handleErrors(response),
         handleSuccesses(response),
-        handleMaxRev(response),
+        handleMaxRevs(response, user.id),
       ]);
     } catch (err) {
       console.error('There was an error updating change status', err); // eslint-disable-line no-console
@@ -279,15 +306,20 @@ async function handleChanges(changes) {
   }
 }
 
+let intervalTimer;
+
 export function startSyncing() {
   cleanupLocks();
   // Initiate a sync immediately in case any data
   // is left over in the database.
   debouncedSyncChanges();
+  // Start the sync interval
+  intervalTimer = setInterval(debouncedSyncChanges, SYNC_POLL_INTERVAL * 1000);
   db.on('changes', handleChanges);
 }
 
 export function stopSyncing() {
+  clearInterval(intervalTimer);
   debouncedSyncChanges.cancel();
   // Dexie's slightly counterintuitive method for unsubscribing from events
   db.on('changes').unsubscribe(handleChanges);

contentcuration/contentcuration/views/base.py

@@ -77,7 +77,7 @@ def current_user_for_context(user):
 
     user_data = {field: getattr(user, field) for field in user_fields}
 
-    user_data["max_rev"] = Change.objects.filter(applied=True).values_list("server_rev", flat=True).order_by("-server_rev").first() or 0
+    user_data["user_rev"] = Change.objects.filter(applied=True, user=user).values_list("server_rev", flat=True).order_by("-server_rev").first() or 0
 
     return json_for_parse_from_data(user_data)
 
@@ -218,6 +218,7 @@ def accounts(request):
 @permission_classes((IsAuthenticated,))
 def channel(request, channel_id):
     channel_error = ""
+    channel_rev = 0
 
     # Check if channel exists
     try:
@@ -231,13 +232,15 @@ def channel(request, channel_id):
         # an option to restore the channel in the Administration page
         if channel.deleted:
             channel_error = 'CHANNEL_EDIT_ERROR_CHANNEL_DELETED'
+        else:
+            channel_rev = Change.objects.filter(applied=True, channel=channel).values_list("server_rev", flat=True).order_by("-server_rev").first() or 0
 
     return render(
         request,
         "channel_edit.html",
         {
             CHANNEL_EDIT_GLOBAL: json_for_parse_from_data(
-                {"channel_id": channel_id, "channel_error": channel_error}
+                {"channel_id": channel_id, "channel_error": channel_error, "channel_rev": channel_rev}
             ),
             CURRENT_USER: current_user_for_context(request.user),
             PREFERENCES: json_for_parse_from_data(request.user.content_defaults),

contentcuration/contentcuration/viewsets/sync/endpoint.py

@@ -16,7 +16,7 @@
 
 def change_model_values_to_change_dict(c):
     datum = c["kwargs"]
-    datum.update({"server_rev": c["server_rev"], "table": c["table"], "type": c["change_type"]})
+    datum.update({"server_rev": c["server_rev"], "table": c["table"], "type": c["change_type"], "channel_id": c["channel_id"], "user_id": c["user_id"]})
     return datum
 
 
@@ -51,30 +51,33 @@ def handle_changes(self, request):
             if user_only_changes:
                 get_or_create_async_task("apply_user_changes", request.user, user_id=request.user.id)
             for channel_id in allowed_ids:
-                print(channel_id)
-                task_info = get_or_create_async_task("apply_channel_changes", request.user, channel_id=channel_id)
-                print(task_info.status)
+                get_or_create_async_task("apply_channel_changes", request.user, channel_id=channel_id)
             allowed_changes = [{"rev": c.client_rev, "server_rev": c.server_rev} for c in change_models]
 
             return {"disallowed": disallowed_changes, "allowed": allowed_changes}
         return {}
 
     def return_changes(self, request):
-        channel_ids = request.data.get("channel_ids", [])
-        last_rev = request.data.get("last_rev") or 0
+        channel_revs = request.data.get("channel_revs", {})
+        user_rev = request.data.get("user_rev") or 0
         session_key = request.session.session_key
-        if channel_ids:
-            channel_ids = Channel.filter_view_queryset(Channel.objects.all(), request.user).filter(id__in=channel_ids).values_list("id", flat=True)
+        if channel_revs:
+            # Filter to only the channels that the user has permissions to view.
+            channel_ids = Channel.filter_view_queryset(Channel.objects.all(), request.user).filter(id__in=channel_revs.keys()).values_list("id", flat=True)
+            channel_revs = {channel_id: channel_revs[channel_id] for channel_id in channel_ids}
 
-        change_filter = (Q(session_id=session_key) & (Q(errored=True) | Q(applied=True))) | Q(user=request.user, applied=True)
+        # Create a filter that returns all applied changes, and any errored changes made by this session
+        relevant_to_session_filter = (Q(applied=True) | Q(errored=True, session_id=session_key))
 
-        if channel_ids:
-            change_filter |= Q(channel_id__in=channel_ids, applied=True)
+        change_filter = (Q(user=request.user, server_rev__gt=user_rev) & relevant_to_session_filter)
+
+        for channel_id, rev in channel_revs.items():
+            change_filter |= (Q(channel_id=channel_id, server_rev__gt=rev) & relevant_to_session_filter)
 
         changes_to_return = list(
             Change.objects.filter(
-                server_rev__gt=last_rev
-            ).filter(change_filter).values("server_rev", "session_id", "applied", "errored", "table", "change_type", "kwargs")
+                change_filter
+            ).values("server_rev", "session_id", "channel_id", "user_id", "applied", "errored", "table", "change_type", "kwargs")
         )
 
         if not changes_to_return:
@@ -87,7 +90,7 @@ def return_changes(self, request):
         for c in changes_to_return:
             if c["applied"]:
                 if c["session_id"] == session_key:
-                    successes.append(c["server_rev"])
+                    successes.append(change_model_values_to_change_dict(c))
                 else:
                     changes.append(change_model_values_to_change_dict(c))
             if c["errored"] and c["session_id"] == session_key:
@@ -108,6 +111,4 @@ def post(self, request):
 
         response_payload.update(self.return_changes(request))
 
-        response_payload["max_rev"] = Change.objects.filter(applied=True).values_list("server_rev", flat=True).order_by("-server_rev").first() or 0
-
         return Response(response_payload)