diff --git a/lib/commands/metadata.js b/lib/commands/metadata.js index 8eeca31ce..a468f6f6e 100644 --- a/lib/commands/metadata.js +++ b/lib/commands/metadata.js @@ -13,7 +13,8 @@ Data.getMetadataRaw = function (Env, channel /* channelName */, _cb) { const cb = Util.once(Util.mkAsync(_cb)); if (!Core.isValidId(channel)) { return void cb('INVALID_CHAN'); } if (channel.length !== HK.STANDARD_CHANNEL_LENGTH && - channel.length !== HK.ADMIN_CHANNEL_LENGTH) { return cb("INVALID_CHAN_LENGTH"); } + channel.length !== HK.ADMIN_CHANNEL_LENGTH && + channel.length !== HK.BLOB_ID_LENGTH) { return cb("INVALID_CHAN_LENGTH"); } // return synthetic metadata for admin broadcast channels as a safety net // in case anybody manages to write metadata @@ -79,6 +80,7 @@ Data.setMetadata = function (Env, safeKey, data, cb, Server) { var channel = data.channel; var command = data.command; + // XXX BLOBMD allow blobs if (!channel || !Core.isValidId(channel)) { return void cb ('INVALID_CHAN'); } if (!command || typeof (command) !== 'string') { return void cb('INVALID_COMMAND'); } if (Meta.commands.indexOf(command) === -1) { return void cb('UNSUPPORTED_COMMAND'); } @@ -140,6 +142,7 @@ Data.setMetadata = function (Env, safeKey, data, cb, Server) { cb(void 0, metadata); return void next(); } + // XXX BLOBMD use correct store for blobs Env.msgStore.writeMetadata(channel, JSON.stringify(line), function (e) { if (e) { cb(e); @@ -155,6 +158,7 @@ Data.setMetadata = function (Env, safeKey, data, cb, Server) { // update the cached metadata metadata_cache[channel] = metadata; + Env.checkCache(channel); // XXX ??? // it's easy to check if the channel is restricted const isRestricted = metadata.restricted; diff --git a/lib/hk-util.js b/lib/hk-util.js index e9d177968..3355f1c09 100644 --- a/lib/hk-util.js +++ b/lib/hk-util.js @@ -42,6 +42,8 @@ const ADMIN_CHANNEL_LENGTH = HK.ADMIN_CHANNEL_LENGTH = 33; // with a 34 character id const EPHEMERAL_CHANNEL_LENGTH = HK.EPHEMERAL_CHANNEL_LENGTH = 34; +const BLOB_ID_LENGTH = HK.BLOB_ID_LENGTH = 48; + // Temporary channels are archived X ms after everyone has left them const TEMPORARY_CHANNEL_LIFETIME = 30 * 1000; diff --git a/lib/storage/blob.js b/lib/storage/blob.js index aeb9c1f70..8c6b9b2be 100644 --- a/lib/storage/blob.js +++ b/lib/storage/blob.js @@ -10,6 +10,10 @@ var BlobStore = module.exports; var nThen = require("nthen"); var Semaphore = require("saferphore"); var Util = require("../common-util"); +var Meta = require("../metadata"); +const PERMISSIVE = 511; + +const readFileBin = require("../stream-file").readFileBin; const BLOB_LENGTH = 48; @@ -31,7 +35,7 @@ var prependArchive = function (Env, path) { return Path.join(Env.archivePath, 'blob', relativePathToBlob); }; -// /blob//// +// /blob// var makeBlobPath = function (Env, blobId) { return Path.join(Env.blobPath, blobId.slice(0, 2), blobId); }; @@ -40,6 +44,11 @@ var makeActivityPath = function (Env, blobId) { return makeBlobPath(Env, blobId) + '.activity'; }; +// /blob//.metadata.ndjson +var mkMetadataPath = function (Env, blobId) { + return Path.join(Env.blobPath, blobId.slice(0, 2), blobId) + '.metadata.ndjson'; +}; + // /blobstate// var makeStagePath = function (Env, safeKey) { return Path.join(Env.blobStagingPath, safeKey.slice(0, 2), safeKey); @@ -190,6 +199,71 @@ var getActivity = function (Env, blobId, cb) { }); }; +// destroyStream && createIdleStreamCollector +// copied from lib/storage/file.js +// see comments there +const STREAM_CLOSE_TIMEOUT = 120000; +const STREAM_DESTROY_TIMEOUT = 30000; +const destroyStream = function (stream) { + if (!stream) { return; } + try { + stream.close(); + if (stream.closed && stream.fd === null) { return; } + } catch (err) { + console.error(err); + } + setTimeout(function () { + try { stream.destroy(); } catch (err) { console.error(err); } + }, STREAM_DESTROY_TIMEOUT); +}; +const createIdleStreamCollector = function (stream) { + var collector = Util.once(Util.mkAsync(Util.bake(destroyStream, [stream]))); + collector.keepAlive = Util.throttle(collector, STREAM_CLOSE_TIMEOUT); + collector.keepAlive(); + return collector; +}; + +// writeMetadata appends to the dedicated log of metadata amendments +var writeMetadata = function (env, channelId, data, cb) { + var path = mkMetadataPath(env, channelId); + + Fse.mkdirp(Path.dirname(path), PERMISSIVE, function (err) { + if (err && err.code !== 'EEXIST') { return void cb(err); } + Fs.appendFile(path, data + '\n', cb); + }); +}; +var archiveMetadata = (Env, blobId, cb) => { + var path = mkMetadataPath(Env, blobId); + var archivePath = prependArchive(Env, path); + // XXX eviction clean lone md files + // if we fail to delete the metadata file, it can still be removed later by the eviction script + Fse.move(path, archivePath, { overwrite: true }, cb); +}; +var readBlobMetadata = function (env, blobId, handler, _cb) { + var metadataPath = mkMetadataPath(env, blobId); + var stream = Fs.createReadStream(metadataPath, {start: 0}); + + const collector = createIdleStreamCollector(stream); + var cb = Util.both(_cb, collector); + + readFileBin(stream, function (msgObj, readMore) { + collector.keepAlive(); + var line = msgObj.buff.toString('utf8'); + try { + var parsed = JSON.parse(line); + handler(null, parsed); + } catch (err) { + handler(err, line); + } + readMore(); + }, function (err) { + // ENOENT => there is no metadata log + if (!err || err.code === 'ENOENT') { return void cb(); } + // otherwise stream errors? + cb(err); + }); +}; + /********** METHODS **************/ var upload = function (Env, safeKey, content, cb) { @@ -313,6 +387,9 @@ var tryId = function (path, cb) { }; // owned_upload_complete +let unescapeKeyCharacters = function (key) { + return key.replace(/\-/g, '/'); +}; var owned_upload_complete = function (Env, safeKey, id, cb) { closeBlobstage(Env, safeKey); if (!isValidId(id)) { @@ -325,12 +402,10 @@ var owned_upload_complete = function (Env, safeKey, id, cb) { } var finalPath = makeBlobPath(Env, id); + let unsafeKey = unescapeKeyCharacters(safeKey); + //var finalOwnPath = makeProofPath(Env, safeKey, id); - var finalOwnPath = makeProofPath(Env, safeKey, id); - - // the user wants to move it into blob and create a empty file with the same id - // in their own space: - // /blob/safeKeyPrefix/safeKey/blobPrefix/blobID + // the user wants to move it into blob and create a metadata log with an owner nThen(function (w) { // make the requisite directory structure using Mkdirp @@ -340,12 +415,6 @@ var owned_upload_complete = function (Env, safeKey, id, cb) { return void cb(e.code); } })); - Fse.mkdirp(Path.dirname(finalOwnPath), w(function (e /*, path */) { - if (e) { // does not throw error if the directory already existed - w.abort(); - return void cb(e.code); - } - })); }).nThen(function (w) { // make sure the id does not collide with another tryId(finalPath, w(function (e) { @@ -355,8 +424,11 @@ var owned_upload_complete = function (Env, safeKey, id, cb) { } })); }).nThen(function (w) { - // Create the empty file proving ownership - Fs.writeFile(finalOwnPath, '', w(function (e) { + // Write the metadata + let md = JSON.stringify({ + owners: [unsafeKey] + }); + writeMetadata(Env, id, md, w((e) => { if (e) { w.abort(); return void cb(e.code); @@ -411,6 +483,7 @@ var archiveBlob = function (Env, blobId, reason, cb) { var blobPath = makeBlobPath(Env, blobId); var archivePath = prependArchive(Env, blobPath); Fse.move(blobPath, archivePath, { overwrite: true }, cb); + archiveMetadata(Env, blobId, () => {}); archiveActivity(Env, blobId, () => {}); addPlaceholder(Env, blobId, reason, () => {}); }; @@ -674,6 +747,15 @@ BlobStore.create = function (config, _cb) { isOwnedBy(Env, safeKey, blobId, cb); }, + readMetadata: (blobId, handler, cb) => { + if (!isValidId(blobId)) { return void cb("INVALID_ID"); } + readBlobMetadata(Env, blobId, handler, cb); + }, + writeMetadata: (blobId, data, cb) => { + if (!isValidId(blobId)) { return void cb("INVALID_ID"); } + writeMetadata(Env, blobId, data, cb); + }, + remove: { blob: function (blobId, _cb) { var cb = Util.once(Util.mkAsync(_cb)); diff --git a/lib/workers/db-worker.js b/lib/workers/db-worker.js index 64376236f..ddd49ccc0 100644 --- a/lib/workers/db-worker.js +++ b/lib/workers/db-worker.js @@ -338,7 +338,13 @@ const computeMetadata = function (data, cb) { const ref = {}; const lineHandler = Meta.createLineHandler(ref, Env.Log.error); monitoringIncrement('computeMetadata'); - return void store.readChannelMetadata(data.channel, lineHandler, function (err) { + + let f = store.readChannelMetadata; + if (data.channel.length === HK.BLOB_ID_LENGTH) { + f = blobStore.readMetadata; + } + + return void f(data.channel, lineHandler, function (err) { if (err) { // stream errors? return void cb(err); @@ -558,6 +564,7 @@ const removeOwnedBlob = function (data, cb) { if (typeof(data.safeKey) !== 'string') { return void cb("INVALID_KEY"); } const blobId = data.blobId; const safeKey = Util.escapeKeyCharacters(data.safeKey); + const unsafeKey = Util.unescapeKeyCharacters(data.safeKey); const reason = data.reason || 'ARCHIVE_OWNED'; @@ -569,6 +576,18 @@ const removeOwnedBlob = function (data, cb) { return void cb("INSUFFICIENT_PERMISSIONS"); } })); + computeMetadata({channel: blobId}, w((err, meta) => { + if (err || !meta) { + w.abort(); + return void cb("INSUFFICIENT_PERMISSIONS"); + } + let owners = meta.owners; + if (!owners || !owners.includes(unsafeKey)) { + w.abort(); + return void cb("INSUFFICIENT_PERMISSIONS"); + } + // Owned, continue + })); }).nThen(function (w) { // remove the blob blobStore.archive.blob(blobId, reason, w(function (err) { @@ -581,20 +600,8 @@ const removeOwnedBlob = function (data, cb) { w.abort(); return void cb(err); } - })); - }).nThen(function () { - // archive the proof - blobStore.archive.proof(safeKey, blobId, function (err) { - Env.Log.info("ARCHIVAL_PROOF_REMOVAL_BY_OWNER_RPC", { - safeKey: safeKey, - blobId: blobId, - status: err? String(err): 'SUCCESS', - }); - if (err) { - return void cb("E_PROOF_REMOVAL"); - } cb(void 0, 'OK'); - }); + })); }); };