Function: Keep the Last N User Items

// To run configure the settings for this Function, keepLastN, as follows:
//
// Version 7.1+
//   "Function Scope"
//     *.* (or try bulk.data if non-privileged)
// Version 7.0+
//   "Listen to Location"
//     bulk.data.source
//   "Eventing Storage"
//     rr100.eventing.metadata
//   Binding(s)
//    1. "binding type", "alias name...", "bucket.scope.collection", "Access"
//       "bucket alias", "src_col",       "bulk.data.source",        "read and write"
//
// Version 6.X
//   "Source Bucket"
//     source
//   "MetaData Bucket"
//     metadata
//   Binding(s)
//    1. "binding type", "alias name...", "bucket", "Access"
//       "bucket alias", "src_col",       "source", "read and write"

/*
 * Process all mutations, however updateNotifyArrayInKV(...) will only
 * data with KEYS like nu:#:#
 */
 function OnUpdate(doc, meta) {
    const MAX_RETRY = 16;
    const MAX_ARRAY = 3;
    const DEBUG = false;

    updateNotifyArrayInKV(doc, meta, MAX_RETRY, MAX_ARRAY, DEBUG);
}

/*
 * reads form KV and manipulates the result to only keep 'MAX_ARRAY' items
 */
function addToNtfyArray(src_col, user_id, insert_json, MAX_ARRAY, DEBUG) {
    var ntfy_id = insert_json.nid;
    var random_csum;
    var user_doc = src_col["user_plus_ntfys:" + user_id];
    if (!user_doc) {
        // generate unique random #
        random_csum = Math.random();
        user_doc = { "type": "user_plus_ntfys", "id": user_id, "notifications" : [], "random": random_csum };
        user_doc.notifications.push(insert_json);
    } else {
        if (user_doc.notifications[0].nid >= ntfy_id && user_doc.notifications.length === MAX_ARRAY) {
            // do nothing this is older data, we assume that nid always increases
            return null;
        } else {
            // find insert position
            for(var i=0; i<=user_doc.notifications.length + 1 ; i++) {
                if (i < user_doc.notifications.length && user_doc.notifications[i].nid === ntfy_id) {
                    // do nothing this is duplicate data we already have it, assume no updates to notifys
                    return null;
                }
                if (i == user_doc.notifications.length || user_doc.notifications[i].nid > ntfy_id) {
                    // add to array middle or end
                    user_doc.notifications.splice(i, 0, insert_json);
                    random_csum = Math.random();
                    // update unique random #
                    user_doc.random = random_csum;
                    break;
                }
            }
        }
        while (user_doc.notifications.length > MAX_ARRAY) {
            // ensure proper size
            user_doc.notifications.shift();
        }
    }
    if (DEBUG) log("user_plus_ntfys:" + user_id,user_doc);
    return user_doc;
}

/*
 * creates, gets, and updates (via replace) the KV tracking array document
 */
function updateNotifyArrayInKV(doc, meta, MAX_RETRY, MAX_ARRAY, DEBUG) {
    // will process ALL data like nu:#:#
    var parts = meta.id.split(':');
    if (!parts || parts.length != 3 || parts[0] != "nu") return;
    var ntfy_id = parseInt(parts[1]);
    var user_id = parseInt(parts[2]);
    //log("Doc created/updated " +  meta.id + " ntfy_id " + ntfy_id  + " user_id " + user_id);

    var insert_json = {"nid": ntfy_id, doc};
    // this is a can be improved in version 6.6.1 as CAS operations are supported.
    // we utilize a 'random' tag in the document itself to emulate a CAS operation.
    for (var tries=0; tries < 16; tries++) {
        var user_doc = addToNtfyArray(src_col, user_id, insert_json, MAX_ARRAY, DEBUG);
        if (user_doc == null) {
            // do nothing
            return;
        }
        var random_csum = user_doc.random;
        // this is a read write alias to the functions source bucket, write then reread
        src_col["user_plus_ntfys:" + user_id] = user_doc;
        user_doc = src_col["user_plus_ntfys:" + user_id];
        // we use the random values to perform a CAS, we give up after 16 attempts
        if (random_csum !== user_doc.random) {
            // failure need to retry
            tries++;
            if (DEBUG) log("CAS like operation failed: tries "+tries+
                " random_csum:",random_csum,"!==", "user_doc.random",user_doc.random);
        } else {
            // success could even delete the input notification doc here
            if (DEBUG) log("CAS like operation success: tries "+tries+
                " random_csum:",random_csum,"!==", "user_doc.random",user_doc.random);
            return;
        }
    }
    log ("FAILED to insert id: " + meta.id, doc)
}

  UPSERT INTO `bulk`.`data`.`source` (KEY,VALUE)
  VALUES ( "nu:1:1",  {"somekey":"someValue"} ),
  VALUES ( "nu:2:2",  {"somekey":"someValue"} ),
  VALUES ( "nu:3:1",  {"somekey":"someValue"} ),
  VALUES ( "nu:4:1",  {"somekey":"someValue"} ),
  VALUES ( "nu:5:1",  {"somekey":"someValue"} ),
  VALUES ( "nu:6:2",  {"somekey":"someValue"} ),
  VALUES ( "nu:7:2",  {"somekey":"someValue"} ),
  VALUES ( "nu:8:1",  {"somekey":"someValue"} ),
  VALUES ( "nu:9:2",  {"somekey":"someValue"} ),
  VALUES ( "nu:10:2", {"somekey":"someValue"} );

NEW/OUTPUT: KEY user_plus_ntfys:1

Note, we add/create the property _random_ in the tracking doc as we use _random_ is used to emulate CAS.

{
  "type": "user_plus_ntfys",
  "id": 1,
  "notifications": [{
    "nid": 4,
    "doc": {
      "somekey": "someValue"
    }
  }, {
    "nid": 5,
    "doc": {
      "somekey": "someValue"
    }
  }, {
    "nid": 8,
    "doc": {
      "somekey": "someValue"
    }
  }],
  "random": 0.9071605464143964
}

NEW/OUTPUT: KEY user_plus_ntfys:2

{
  "type": "user_plus_ntfys",
  "id": 2,
  "notifications": [{
    "nid": 7,
    "doc": {
      "somekey": "someValue"
    }
  }, {
    "nid": 9,
    "doc": {
      "somekey": "someValue"
    }
  }, {
    "nid": 10,
    "doc": {
      "somekey": "someValue"
    }
  }],
  "random": 0.5637501636850883
}