fix(media): a hub admin's workspace rename reaches every member

service/private/media.js

@@ -312,7 +312,7 @@
     const proc = `${des_db}.mfs_access_node`;
     for (let r of toArray(recipients)) {
       if (!oldItems[r.uid]) {
         oldItems[r.uid] = await this.db.await_proc(proc, r.uid, nid);
       }
     }
     let nodes = {};
@@ -323,7 +323,7 @@
       if (nodes[s.uid]) {
         r = nodes[s.uid];
       } else {
         r = await this.yp.await_proc(access, s.uid, nid);
       }
       if (!r || !r.actual_db) continue;
       if (r.filetype == Attr.hub && r.actual_hub_id) {
@@ -409,9 +409,9 @@
         case "show":
           let nid = node.nid;
           let access = `${node.des_db}.mfs_access_node`;
           let sockets = await this.yp.await_proc("entity_sockets", {
             db_name: node.des_db,
           });
           nodes = {};
           let counts = {};
           for (let s of toArray(sockets)) {
@@ -420,7 +420,7 @@
             if (nodes[s.uid]) {
               r = nodes[s.uid];
             } else {
               r = await this.yp.await_proc(access, s.uid, nid);
               result.push(r);
             }
             if (!r || !r.actual_db) continue;
@@ -441,7 +441,7 @@
               counts[s.uid] = c;
             }
             nodes[s.uid] = r;
             await RedisStore.sendData(this.payload(r), s);
           }
           break;
 
@@ -469,7 +469,7 @@
         dest.hub_id = destinationHubId;
         if (!workspaceMoveHubNames.has(destinationHubId)) {
           try {
             const hub = toArray(await this.yp.await_proc('get_hub', destinationHubId))[0] || {};
             let profile = {};
             try {
               profile = isString(hub.profile) ? JSON.parse(hub.profile) : (hub.profile || {});
@@ -653,13 +653,13 @@
         continue;
       }
       granted.push(node);
       let recipients = await this.yp.await_proc("entity_sockets", {
         hub_id: n.hub_id
       });
       let proc = `${n.db_name}.mfs_access_node`;
       for (let r of toArray(recipients)) {
         if (this.heap.oldItems[r.uid]) continue;
         node = await this.db.await_proc(proc, r.uid, n.id);
         if (node && node.privilege) {
           this.heap.oldItems[r.uid] = node;
         }
@@ -852,16 +852,16 @@
       const nid = node && (node.nid || node.id);
       const hubId = node && (node.actual_hub_id || node.hub_id);
       if (!nid || !hubId) continue;
       const storage = firstRow(await this.yp.await_proc("file_move_entity_storage", hubId));
       const dbName = storage && storage.db_name;
       if (!dbName) continue;
       try {
         // Handles a file and a folder alike: given a file it returns at most
         // that file's own thread, given a folder it returns every thread
         // beneath it at any depth.
         const threads = toArray(await this.yp.await_proc(
           `${dbName}.channel_file_thread_list_in_subtree`, nid
         ));
         for (const row of threads) {
           if (!row || !row.file_nid || !row.file_thread_id) continue;
           const key = String(row.file_nid);
@@ -877,7 +877,7 @@
           // operation touching this thread mid-move. If it fails the thread is
           // simply left alone — the file still moves, and a thread left marked
           // active over a departed file is recoverable.
           const reservation = await this._reserveDirectFileThreadTrash(target);
           if (reservation && !reservation.failed
             && Number(reservation.reserved) === 1) {
             target.reserved = true;
@@ -908,7 +908,7 @@
     for (const [key, target] of snapshots) {
       if (!target || !target.reserved) continue;
       if (movedFileNids && movedFileNids.has(key)) continue;
       await this._releaseDirectFileThreadTrash(target);
       target.reserved = false;
     }
   }
@@ -1077,17 +1077,17 @@
       for (const row of orphaned) {
         // Each event goes to the workspace that still holds that thread, never
         // to the one where the file was deleted.
         const recipients = await this.yp.await_proc("entity_sockets", row.current_hub_id);
         await RedisStore.sendData(this.payload({
           lineage_id: row.lineage_id,
           access_revision: row.access_revision,
           actor: this._fileMoveActor(),
           reason: "orphaned",
           state: "orphaned",
           hub_id: row.current_hub_id,
           file_nid: row.current_file_nid,
           file_thread_id: row.current_thread_id,
         }, { service: "channel.file_thread_access_changed" }), recipients);
       }
       return orphaned;
     } catch (error) {
@@ -1120,24 +1120,24 @@
         // Every thread anywhere that is waiting on this file. A file discussed
         // in more than one workspace has one sleeping thread per workspace, and
         // a single move can touch several of them at once.
         const waiting = toArray(await this.yp.await_proc(
           "file_thread_lineage_resolve_holder", sourceHubId, pair.source_file_nid
         )).filter((row) => row && row.lineage_id);
 
         for (const lineage of waiting) {
           if (String(lineage.current_hub_id) === String(destinationHubId)) {
             // This thread's own workspace is where the file is going: its wait
             // is over.
             await this._markFileThreadMovedBack(
               lineage, pair.destination_file_nid, sourceDbName
             );
           } else {
             // Still away from home. Follow the file so the thread's info card
             // keeps naming the workspace that actually holds it.
             const tracked = firstRow(await this.yp.await_proc(
               "file_thread_lineage_track_holder",
               lineage.lineage_id, destinationHubId, pair.destination_file_nid
             ));
             if (!tracked || tracked.failed || Number(tracked.tracked) !== 1) {
               this.warn("File thread holder tracking failed", {
                 lineage_id: lineage.lineage_id,
@@ -1268,15 +1268,15 @@
     }
 
     for (const source of values(sourceUpdates)) {
       const recipients = await this.yp.await_proc("entity_sockets", source.hub_id);
       await RedisStore.sendData(
         this.payload(source, { keys: [Attr.nid, Attr.hub_id], service: "media.remove" }),
         recipients
       );
       await RedisStore.sendData(
         this.payload({}, { service: "notification.resync" }),
         recipients
       );
     }
 
     // The thread half of the move. Threads are never carried across databases:
@@ -1299,12 +1299,12 @@
       const storage = firstRow(await this.yp.await_proc(
         "file_move_entity_storage", source.hub_id
       ));
       await this._applyFileThreadMove(
         source.hub_id,
         storage && storage.db_name,
         destinationHubId,
         threadSnapshots
       );
     }
 
     // A selection can be partly refused (permissions, circular reference), so a
@@ -1512,13 +1512,13 @@
     if (departed.length) {
       const recipients = await this.yp.await_proc("entity_sockets", sourceHubId);
       for (const node of departed) {
         await RedisStore.sendData(
           this.payload(
             { nid: node.nid, hub_id: sourceHubId },
             { keys: [Attr.nid, Attr.hub_id], service: "media.remove" }
           ),
           recipients
         );
       }
       await RedisStore.sendData(
         this.payload({}, { service: "notification.resync" }),
@@ -1895,7 +1895,7 @@
     let node;
     for (let uid of users) {
       let user = await this.yp.await_proc("get_visitor", uid);
       let privilege = await this.db.await_func(`user_permission`, uid, "*");
       user.privilege = privilege;
       if (!(privilege & Permission.OWNER) && privilege & Permission.READ) {
         let profile = JSON.parse(user.profile);
@@ -1918,23 +1918,23 @@
           stringify(md.fullname),
           1
         );
         await this.db.await_proc(
           "mfs_set_attr",
           node.id,
           "metadata",
           md
         );
         for (let id of users) {
           if (id == user.id) {
             await this.db.await_proc(
               "permission_grant",
               node.id,
               user.id,
               0,
               Privilege.WRITE,
               "system",
               `Writable by ${profile.email}`
             );
           } else {
             await this.db.await_proc(
               "permission_grant",
@@ -2059,7 +2059,7 @@
     let denied = [];
     for (let node of src) {
       var proc = `${node.db_name}.mfs_access_node`;
       source_node = await this.yp.await_proc(proc, uid, node.id);
       if (source_node.permission & node.privilege) {
         this.heap.srcgrantlst.push({
           nid: source_node.nid,
@@ -2177,7 +2177,7 @@
             mfs_root: row.des_mfs_root,
           };
           proc = `${row.dest_db_name}.mfs_access_node`;
           r = await this.yp.await_proc(proc, this.uid, dest.nid);
           r.privilege = r.permission;
           show_node.push(r);
           copy_node(src, dest, 1);
@@ -2185,9 +2185,9 @@
         case "show":
         case "showone":
           if (!row.dest_db_name) {
             let entity = await this.yp.await_proc('get_entity', this.input.get('recipient_id'))
             proc = `${entity.db_name}.mfs_access_node`;
             r = await this.db.await_proc(proc, this.uid, row.nid);
             r.privilege = r.permission;
             if (r.filetype == Attr.hub) {
               r.hub_id = row.nid; // hub_id is inconsistent after trash
@@ -2196,7 +2196,7 @@
             continue;
           }
           proc = `${row.dest_db_name}.mfs_access_node`;
           r = await this.yp.await_proc(proc, this.uid, row.nid);
           r.privilege = r.permission;
 
           if (r.filetype == Attr.hub) {
@@ -2231,20 +2231,20 @@
     let sockets = [];
 
     for (var m of show_node) {
       const restoredThread = await this._directFileThreadSnapshot(
         m.hub_id,
         m.nid || m.id,
         m.db_name || m.actual_db
       );
       await this._transitionDirectFileThreadAccess(restoredThread, "active", "direct_restore");
       let changelog = await this.changelog_write({ src: m, event: "media.new" });
       let dest = await this.yp.await_proc("entity_sockets", m.hub_id);
       sockets = sockets.concat(dest);
       m.args = { ...m.args, changelog };
       await RedisStore.sendData(
         this.payload(m, { service: "media.restore_into" }),
         dest
       );
     }
     await RedisStore.sendData(
       this.payload({ rebuild: 1 }, { service: "notification.resync" }),
@@ -2349,12 +2349,12 @@
     for (var node of data) {
       db_name = node.db_name;
       if (!db_name || db_name == null) {
         files = await this.db.await_proc("mfs_purge", node.id);
         continue;
       }
       switch (node.category) {
         case Attr.folder:
           files = await this.yp.await_proc(db_name + ".mfs_purge", node.id);
           files = toArray(files);
           for (let f of files) {
             res.push(f);
@@ -2372,7 +2372,7 @@
       }
     }
     for (var entity of entities) {
       await this.yp.await_proc("entity_delete", entity);
     }
     return data.concat(res);
   }
@@ -2397,7 +2397,7 @@
       if (!nid) continue;
       if (entry.category === FOLDER || entry.category === HUB) continue;
       try {
         await this._markFileThreadOrphaned(entry.hub_id || hubId, nid);
       } catch (error) {
         this.warn("Unable to orphan thread for purged file", { nid, error });
       }
@@ -2644,11 +2644,11 @@
     const directAccessKeys = new Set();
     for (const g of granted) {
       try {
         const hub_db = await this.yp.await_func('get_db_name', g.hub_id);
         if (!hub_db) continue;
         const attr = await this.yp.await_proc(`${hub_db}.mfs_node_attr`, g.nid);
         if (!attr) continue;
         const directTarget = await this._directFileThreadSnapshot(g.hub_id, g.nid, hub_db);
         const directKey = directTarget && `${directTarget.hub_id}:${directTarget.file_nid}`;
         if (directTarget && !directAccessKeys.has(directKey)) {
           directAccessKeys.add(directKey);
@@ -2699,11 +2699,11 @@
     let changelog = await this.changelog_write({ src: data, event: service });
     if (isEmpty(data)) {
       for (let h of granted) {
         recipients = await this.yp.await_proc("entity_sockets", h.hub_id);
         await RedisStore.sendData(
           this.payload({ ...h, changelog }, { keys, service }),
           recipients
         );
         await RedisStore.sendData(
           this.payload({}, { service: "notification.resync" }),
           recipients
@@ -2725,14 +2725,14 @@
     this.output.add_data({ changelog });
 
     for (const t of auditTargets) {
       await writeAudit(this, {
         db: t.hub_db,
         uid: this.uid,
         action: 'deleted',
         category: 'media',
         entity_id: t.nid,
         log: `${t.filetype === 'hub' ? 'Workspace' : t.filetype === 'folder' ? 'Folder' : 'File'} '${t.filename}' moved to trash`,
       });
     }
 
     await this._evaluateOverLimitAfterResolve();
@@ -2865,7 +2865,9 @@
     let newItems = {};
     let recipients;
 
-    /**  Renaming hubname must not change other's name*/
+    /** A hub row lives on the caller's desk: start with the caller's own
+     *  sockets. A hub admin's rename widens this to every member in
+     *  renameHubForMembers; anyone else renames only their own label. */
     if (node[FILETYPE] == Attr.hub) {
       recipients = await this.yp.await_proc(
         "entity_sockets",
@@ -2893,11 +2895,11 @@
       default:
         for (let r of toArray(recipients)) {
           if (!oldItems[r.uid]) {
             oldItems[r.uid] = await this.db.await_proc(
               "mfs_access_node",
               r.uid,
               nid
             );
           }
         }
         res = await this.db.await_proc("mfs_rename", nid, filename);
@@ -2917,12 +2919,22 @@
           await this.changelog_write({ src: old, dest: attr });
         }
     }
+    if (node[FILETYPE] == Attr.hub) {
+      // The caller's own rename has already landed; a failure here must not
+      // turn it into an error, so it degrades to the personal rename.
+      try {
+        const members = await this.renameHubForMembers(nid, filename, oldItems, newItems);
+        if (members) recipients = members;
+      } catch (e) {
+        this.warn("media.rename: could not rename the workspace for its members", e);
+      }
+    }
     for (let r of toArray(recipients)) {
       let dest;
       if (newItems[r.uid] && newItems[r.uid].filename) {
         dest = newItems[r.uid]
       } else {
         dest = await this.db.await_proc("mfs_access_node", r.uid, nid);
         newItems[r.uid] = dest;
       }
       let model = {
@@ -2934,7 +2946,7 @@
           changelog: this.__changelog
         }
       };
       await RedisStore.sendData(this.payload(model), r);
     }
     let model;
     if (newItems[this.uid]) {
@@ -2961,6 +2973,52 @@
     this.output.data(model);
   }
 
+  /**
+   * A hub admin's rename reaches EVERY member of the workspace.
+   *
+   * Each member's desk holds its own row for the workspace and the desk lists
+   * that row's name, so renaming only the caller's row (above) left everyone
+   * else on the old name. yp.hub_rename_for_members writes yp.hub.name and
+   * every member's row; each member's node is then read back from THEIR own
+   * db, because this.db is the caller's desk and answers the caller's name.
+   *
+   * Anyone below admin keeps the personal rename: only their own label moves.
+   *
+   * @returns the hub's member sockets to push to, or null for a personal rename
+   */
+  async renameHubForMembers(hub_id, filename, oldItems, newItems) {
+    const hub = firstRow(
+      await this.yp.await_query("SELECT db_name FROM entity WHERE id=?", hub_id)
+    );
+    if (!hub || !hub.db_name) return null;
+    const hubDb = hub.db_name;
+    const privilege = await this.yp.await_func(
+      `${hubDb}.user_permission`, this.uid, "*"
+    );
+    if (!(Number(privilege) & Permission.ADMIN)) return null;
+
+    const members = toArray(
+      await this.yp.await_proc("hub_rename_for_members", hub_id, filename)
+    );
+    for (const m of members) {
+      if (!m || !m.uid || !m.db_name) continue;
+      // The caller's row was renamed and read back by the main path.
+      if (m.uid == this.uid && newItems[m.uid]) continue;
+      const dest = await this.yp.await_proc(
+        `${m.db_name}.mfs_access_node`, m.uid, hub_id
+      );
+      if (!dest || !dest.filename) continue;
+      dest.hub_id = dest.actual_hub_id;
+      dest.privilege = dest.permission;
+      dest.home_id = dest.actual_home_id;
+      newItems[m.uid] = dest;
+      if (!oldItems[m.uid]) {
+        oldItems[m.uid] = { ...dest, filename: m.old_filename, fname: m.old_filename };
+      }
+    }
+    return this.yp.await_proc("entity_sockets", hub_id);
+  }
+
   /**
    * Not used
    */

test/hub-rename-members.test.js

@@ -0,0 +1,157 @@
+// media.rename on a WORKSPACE reaches every member when a hub admin does it.
+//
+// Each member's desk holds its own row for a workspace (media, category hub)
+// and the desk lists that row's name. rename() only ever ran mfs_rename on the
+// CALLER's desk and pushed to the CALLER's sockets, so every other member kept
+// the old name — and the dest it would have pushed them was read from the
+// caller's db anyway.
+//
+// A hub admin's rename now also calls yp.hub_rename_for_members, reads each
+// member's node back from THEIR db, and pushes to the hub's sockets. Anyone
+// below admin keeps the personal rename.
+//
+// Run: node --test test/hub-rename-members.test.js
+const assert = require("node:assert/strict");
+const test = require("node:test");
+
+global.myDrumee = { arch: "pod", useEmail: 0 };
+global.verbosity = 0;
+global.debug = {};
+
+const { RedisStore, Attr } = require("@drumee/server-essentials");
+const MediaPrivate = require("../service/private/media");
+
+const sent = [];
+RedisStore.sendData = async (payload, dest) => {
+  sent.push({ payload, dest });
+};
+
+const HUB_ID = "hub-1";
+const CALLER = { uid: "u1", db: "db_u1" };
+const MEMBER = { uid: "u2", db: "db_u2" };
+
+function node(uid, filename) {
+  return {
+    nid: HUB_ID,
+    id: HUB_ID,
+    filename,
+    filetype: Attr.hub,
+    actual_hub_id: HUB_ID,
+    actual_home_id: "home-1",
+    permission: 63,
+    owner: uid,
+  };
+}
+
+function fakeService({ privilege }) {
+  const calls = [];
+  // Per-desk names, as the two databases would hold them.
+  const names = { [CALLER.uid]: "Old", [MEMBER.uid]: "Old" };
+  const svc = Object.create(MediaPrivate.prototype);
+  Object.assign(svc, {
+    calls,
+    uid: CALLER.uid,
+    __changelog: {},
+    randomString: () => "tag",
+    source_granted: () => ({ node: { ...node(CALLER.uid, "Old"), hub_id: CALLER.uid } }),
+    input: { need: () => "New Name", get: () => undefined, use: () => undefined },
+    exception: { user: (code) => calls.push(["exception", code]) },
+    hub: { get: (k) => (k === Attr.id ? CALLER.uid : k === Attr.db_name ? CALLER.db : undefined) },
+    changelog_write: async () => {},
+    payload: (model) => ({ model }),
+    output: { data: (d) => (svc.result = d) },
+    warn: () => {},
+    db: {
+      // The CALLER's desk: whatever uid it is asked about, it answers its own row.
+      await_proc: async (name, ...args) => {
+        calls.push(["db", name, ...args]);
+        if (name === "mfs_rename") names[CALLER.uid] = args[1];
+        if (name === "mfs_access_node") return node(args[0], names[CALLER.uid]);
+        return {};
+      },
+    },
+    yp: {
+      await_query: async (sql, id) => {
+        calls.push(["yp.query", id]);
+        return [{ db_name: "hub_db" }];
+      },
+      await_func: async (name, ...args) => {
+        calls.push(["yp.func", name, ...args]);
+        return privilege;
+      },
+      await_proc: async (name, ...args) => {
+        calls.push(["yp", name, ...args]);
+        if (name === "entity_sockets") {
+          return args[0] === HUB_ID
+            ? [{ socket_id: "s1", uid: CALLER.uid }, { socket_id: "s2", uid: MEMBER.uid }]
+            : [{ socket_id: "s1", uid: CALLER.uid }];
+        }
+        if (name === "hub_rename_for_members") {
+          names[MEMBER.uid] = args[1];
+          return [
+            { uid: CALLER.uid, db_name: CALLER.db, old_filename: "New Name", filename: "New Name", changed: 0 },
+            { uid: MEMBER.uid, db_name: MEMBER.db, old_filename: "Old", filename: "New Name", changed: 1 },
+          ];
+        }
+        if (name === `${MEMBER.db}.mfs_access_node`) return node(args[0], names[MEMBER.uid]);
+        return {};
+      },
+    },
+  });
+  return svc;
+}
+
+test("hub admin: every member's row is renamed and every member is told", async () => {
+  sent.length = 0;
+  const svc = fakeService({ privilege: 63 });
+  await svc.rename();
+
+  assert.ok(svc.calls.some((c) => c[0] === "yp.func" && c[1] === "hub_db.user_permission" && c[2] === CALLER.uid));
+  assert.ok(svc.calls.some((c) => c[1] === "hub_rename_for_members" && c[2] === HUB_ID && c[3] === "New Name"));
+
+  const toMember = sent.find((s) => s.dest.socket_id === "s2");
+  assert.ok(toMember, "the other member gets a push");
+  assert.equal(toMember.payload.model.args.dest.filename, "New Name");
+  assert.equal(toMember.payload.model.args.src.filename, "Old");
+  assert.equal(toMember.payload.model.filetype, Attr.hub);
+  // Read from the MEMBER's db, never the caller's.
+  assert.ok(svc.calls.some((c) => c[1] === `${MEMBER.db}.mfs_access_node` && c[2] === MEMBER.uid));
+
+  const toCaller = sent.find((s) => s.dest.socket_id === "s1");
+  assert.equal(toCaller.payload.model.args.dest.filename, "New Name");
+  assert.equal(svc.result.args.dest.filename, "New Name");
+});
+
+test("member below admin: personal rename only, nobody else touched", async () => {
+  sent.length = 0;
+  const svc = fakeService({ privilege: 3 });
+  await svc.rename();
+
+  assert.ok(!svc.calls.some((c) => c[1] === "hub_rename_for_members"));
+  assert.deepEqual(sent.map((s) => s.dest.socket_id), ["s1"]);
+  assert.equal(svc.result.args.dest.filename, "New Name");
+});
+
+test("not a hub: the shared path never runs", async () => {
+  sent.length = 0;
+  const svc = fakeService({ privilege: 63 });
+  const granted = svc.source_granted;
+  svc.source_granted = () => ({ node: { ...granted().node, filetype: Attr.folder } });
+  await svc.rename();
+
+  assert.ok(!svc.calls.some((c) => c[0] === "yp.func" || c[1] === "hub_rename_for_members"));
+});
+
+test("shared path throws: the caller's rename still answers, personal push only", async () => {
+  sent.length = 0;
+  const svc = fakeService({ privilege: 63 });
+  const proc = svc.yp.await_proc;
+  svc.yp.await_proc = async (name, ...args) => {
+    if (name === "hub_rename_for_members") throw new Error("boom");
+    return proc(name, ...args);
+  };
+  await svc.rename();
+
+  assert.equal(svc.result.args.dest.filename, "New Name");
+  assert.deepEqual(sent.map((s) => s.dest.socket_id), ["s1"]);
+});