diff --git a/lib/commands/channel.js b/lib/commands/channel.js index 659d7986b..26dde5103 100644 --- a/lib/commands/channel.js +++ b/lib/commands/channel.js @@ -8,6 +8,7 @@ const Util = require("../common-util"); const nThen = require("nthen"); const Core = require("./core"); const Metadata = require("./metadata"); +const Linked = require("./linked"); const HK = require("../hk-util"); const Nacl = require("tweetnacl/nacl-fast"); @@ -200,7 +201,11 @@ Channel.trimHistory = function (Env, safeKey, data, cb) { } // else fall through to the next block })); + }).nThen(function (w) { + // Archive old checkpoints + Linked.trimHistory(Env, { channel: channelId }, w()); }).nThen(function () { + // Trim chainpad doc: Env.msgStore.trimChannel(channelId, hash, function (err) { Env.Log.info('HK_TRIM_HISTORY', { unsafeKey: unsafeKey, diff --git a/lib/commands/linked.js b/lib/commands/linked.js index 44c32cf22..4b139d118 100644 --- a/lib/commands/linked.js +++ b/lib/commands/linked.js @@ -21,12 +21,14 @@ const getMetadata = (Env, channel, _cb) => { MetaRPC.getMetadataRaw(Env, channel, (err, metadata) => { if (err) { return void cb(err); } - if (metadata?.channel !== channel) { + if (metadata?.channel !== channel && channel.length !== HK.BLOB_ID_LENGTH) { return cb(); } // cache it - Env.metadata_cache[channel] = metadata; + if (channel.length !== HK.BLOB_ID_LENGTH) { + Env.metadata_cache[channel] = metadata; + } cb(undefined, metadata); }); }; @@ -65,6 +67,23 @@ Linked.listLinkedDocuments = (Env, channel, cb) => { }); }; +Linked.listOldCheckpoints = (Env, channel, cb) => { + const list = new Set(); + Linked.getLinkedDocuments(Env, { channel }, (err, json) => { + if (err) { return void cb(err); } + + const cps = json.checkpoints || []; + cps.pop(); // preserve last cp + + cps.forEach(obj => { + if (obj?.rtChannel) { list.add(obj.rtChannel); } + if (obj?.blob) { list.add(obj.blob); } + }); + + cb(void 0, Array.from(list)); + }); +}; + const checkContent = (content, user) => { const { type, data } = content; @@ -286,6 +305,8 @@ Linked.getHistorySize = (Env, data, _cb) => { let linked; let channelTotalSize = 0; let size = 0; + let start = 0; + let hash; nThen(waitFor => { Linked.getLinkedDocuments(Env, data, waitFor((err, json) => { @@ -311,9 +332,19 @@ Linked.getHistorySize = (Env, data, _cb) => { waitFor.abort(); return void cb(err); } + start = offset; const chanSize = channelTotalSize - offset; size += chanSize; })); + }).nThen(waitFor => { + // Get oldest hash of non-history data + Env.store.readMessagesBin(channel, start, (msgObj, readMore, abort) => { + const parsed = Util.tryParse(msgObj.buff.toString('utf8')); + if (!parsed) { return void readMore(); } + hash = HK.getHash(parsed[4]); + abort(); + }, waitFor()); + }).nThen(waitFor => { // Get last checkpoint size (blob + rtChannel) // Note: blob may be falsy if no checkpoint @@ -340,6 +371,46 @@ Linked.getHistorySize = (Env, data, _cb) => { size += _size; }), true); }).nThen(() => { - cb(void 0, size); + cb(void 0, { + size, hash + }); }); }; + +Linked.trimHistory = (Env, data, cb) => { + const channel = data.channel; + let linked; + // if we reach this step, it means this user is an owner of "channel" + // so we can also delete any document linked to "channel" (from metadata) + nThen(waitFor => { + // List all but the current checkpoints + Linked.listOldCheckpoints(Env, channel, waitFor((err, channels) => { + if (err) { + waitFor.abort(); + return void cb(err); + } + linked = channels || []; + })); + }).nThen(() => { + let n = nThen; + linked.forEach(chan => { + n = n(w => { + // If channel is "linked", we can archive all but last cp + getMetadata(Env, chan, w((err, md) => { + if (md?.linked !== channel) { return; } + // This is an old checkpoint linked to our document, + // we can archive it + const reason = "TRIM_HISTORY"; + if (chan.length === HK.BLOB_ID_LENGTH) { + return Env.blobStore.archive.blob(chan, reason, w()); + } + Env.store.archiveChannel(chan, reason, w()); + })); + }).nThen; + }); + n(() => { + cb(); + }); + }); + +}; diff --git a/lib/commands/upload.js b/lib/commands/upload.js index d1c4963a9..b21443ecb 100644 --- a/lib/commands/upload.js +++ b/lib/commands/upload.js @@ -69,7 +69,9 @@ Upload.status = function (Env, safeKey, data, _cb) { // FIXME FILES var user = Core.getSession(Env.Sessions, safeKey); user.pendingUploadSize = filesize; user.currentUploadSize = 0; + user.linked = data.linked; +console.error('UPLOAD STATUS', user.linked); cb(void 0, false); }); }); @@ -88,7 +90,9 @@ var completeUpload = function (owned) { Env.blobStore.closeBlobstage(safeKey); var user = Core.getSession(Env.Sessions, safeKey); var size = user.pendingUploadSize; - Env.completeUpload(safeKey, arg, Boolean(owned), size, cb); + var linked = user.linked; + console.error('COMPLETE USER', user.linked); + Env.completeUpload(safeKey, arg, Boolean(owned), size, linked, cb); }; }; diff --git a/lib/storage/blob.js b/lib/storage/blob.js index 3fda70217..fd25d638d 100644 --- a/lib/storage/blob.js +++ b/lib/storage/blob.js @@ -378,7 +378,7 @@ var upload_cancel = function (Env, safeKey, fileSize, cb) { }; // upload_complete -var upload_complete = function (Env, safeKey, id, cb) { +var upload_complete = function (Env, safeKey, id, cb, linked) { closeBlobstage(Env, safeKey); var oldPath = makeStagePath(Env, safeKey); @@ -405,6 +405,12 @@ var upload_complete = function (Env, safeKey, id, cb) { } cb(void 0, id); })); + }).nThen(function (w) { + if (!linked) { return; } + // Write the metadata + let meta = { linked }; + let md = JSON.stringify(meta); + writeMetadata(Env, id, md, w()); }).nThen(function () { // finally, move the old file to the new path // FIXME we could just move and handle the EEXISTS instead of the above block @@ -438,7 +444,7 @@ var tryId = function (path, cb) { let unescapeKeyCharacters = function (key) { return key.replace(/\-/g, '/'); }; -var owned_upload_complete = function (Env, safeKey, id, cb) { +var owned_upload_complete = function (Env, safeKey, id, cb, linked) { closeBlobstage(Env, safeKey); if (!isValidId(id)) { return void cb('EINVAL_ID'); @@ -472,9 +478,9 @@ var owned_upload_complete = function (Env, safeKey, id, cb) { })); }).nThen(function (w) { // Write the metadata - let md = JSON.stringify({ - owners: [unsafeKey] - }); + let meta = { owners: [unsafeKey] }; + if (linked) { meta.linked = linked; } + let md = JSON.stringify(meta); writeMetadata(Env, id, md, w((e) => { if (e) { w.abort(); @@ -883,17 +889,17 @@ BlobStore.create = function (config, _cb) { closeBlobstage: function (safeKey) { closeBlobstage(Env, safeKey); }, - complete: function (safeKey, id, _cb) { + complete: function (safeKey, id, _cb, linked) { var cb = Util.once(Util.mkAsync(_cb)); if (!isValidSafeKey(safeKey)) { return void cb('INVALID_SAFEKEY'); } if (!isValidId(id)) { return void cb("INVALID_ID"); } - upload_complete(Env, safeKey, id, cb); + upload_complete(Env, safeKey, id, cb, linked); }, - completeOwned: function (safeKey, id, _cb) { + completeOwned: function (safeKey, id, _cb, linked) { var cb = Util.once(Util.mkAsync(_cb)); if (!isValidSafeKey(safeKey)) { return void cb('INVALID_SAFEKEY'); } if (!isValidId(id)) { return void cb("INVALID_ID"); } - owned_upload_complete(Env, safeKey, id, cb); + owned_upload_complete(Env, safeKey, id, cb, linked); }, size: function (id, _cb) { var cb = Util.once(Util.mkAsync(_cb)); diff --git a/lib/workers/db-worker.js b/lib/workers/db-worker.js index e9aa15e20..626080ec4 100644 --- a/lib/workers/db-worker.js +++ b/lib/workers/db-worker.js @@ -702,7 +702,7 @@ const completeUpload = function (data, cb) { Env.blobStore[method](safeKey, arg, function (err, id) { reportStatus(Env, label, safeKey, err, id, size); cb(err, id); - }); + }, data.linked); }; const getPinActivity = function (data, cb) { diff --git a/lib/workers/index.js b/lib/workers/index.js index 02b49da19..03de96a1e 100644 --- a/lib/workers/index.js +++ b/lib/workers/index.js @@ -584,9 +584,10 @@ Workers.initialize = function (Env, config, _cb) { }, cb); }; - Env.completeUpload = function (safeKey, arg, owned, size, cb) { + Env.completeUpload = function (safeKey, arg, owned, size, linked, cb) { sendCommand({ command: "COMPLETE_UPLOAD", + linked, owned: owned, // Boolean safeKey: safeKey, // String (public key) arg: arg, // String (file id) diff --git a/src/worker/async-store.js b/src/worker/async-store.js index 7f2c78519..ed2b980d7 100644 --- a/src/worker/async-store.js +++ b/src/worker/async-store.js @@ -393,7 +393,8 @@ const factory = (Sortify, UserObject, ProxyManager, if (!s.rpc) { return void cb({error: 'RPC_NOT_READY'}); } s.rpc.uploadStatus({ id: data.id, - size: data.size + size: data.size, + linked: data.linked }, function (err, res) { if (err) { return void cb({error:err}); } cb(res); diff --git a/src/worker/modules/history.js b/src/worker/modules/history.js index 3ccdaa1a4..6effb4d7a 100644 --- a/src/worker/modules/history.js +++ b/src/worker/modules/history.js @@ -139,7 +139,9 @@ const factory = (Util, Hash, UserObject, nThen) => { waitFor.abort(); return void cb(obj); } - dataSize = obj[0] || 0; + let value = obj[0]; + dataSize = value?.size || 0; + hash = value?.hash; })); } else { Store.getHistory(null, { @@ -239,6 +241,7 @@ const factory = (Util, Hash, UserObject, nThen) => { }); }; + // XXX commands.TRIM_HISTORY = function (ctx, data, cId, cb) { if (!ctx.store.loggedIn || !ctx.store.rpc) { return void cb({ error: 'INSUFFICIENT_PERMISSIONS' }); } var channels = data.channels; diff --git a/www/common/cryptpad-common.js b/www/common/cryptpad-common.js index 0611534fa..f77443069 100644 --- a/www/common/cryptpad-common.js +++ b/www/common/cryptpad-common.js @@ -635,8 +635,8 @@ define([ }); }; - common.uploadStatus = function (teamId, id, size, cb) { - postMessage("UPLOAD_STATUS", {teamId, id, size}, function (obj) { + common.uploadStatus = function (data, cb) { + postMessage("UPLOAD_STATUS", data, function (obj) { if (obj && obj.error) { return void cb(obj.error); } cb(null, obj); }); diff --git a/www/common/inner/properties.js b/www/common/inner/properties.js index 83dcc86d2..efd6fa7ce 100644 --- a/www/common/inner/properties.js +++ b/www/common/inner/properties.js @@ -19,6 +19,7 @@ define([ var getPadProperties = function (Env, data, opts, _cb) { var cb = Util.once(Util.mkAsync(_cb)); var common = Env.common; + var sframeChan = common.getSframeChannel(); opts = opts || {}; var $d = $('