From 2fe52cb97219277325f6642cd926f116cd094636 Mon Sep 17 00:00:00 2001 From: ProgrammerIn-wonderland <3838shah@gmail.com> Date: Tue, 29 Jul 2025 12:08:34 -0700 Subject: [PATCH] Filesystem backed web workers --- .../src/om/entitystorage/SubdomainES.js | 19 ++- .../src/services/EntityStoreService.js | 1 + .../src/services/worker/WorkerService.js | 129 ++++++++++++++++-- .../worker/workerUtils/cloudflareDeploy.js | 14 +- .../services/worker/workerUtils/nameUtils.js | 9 +- src/dev-center/index.html | 2 +- src/puter-js/src/modules/Workers.js | 11 +- 7 files changed, 151 insertions(+), 34 deletions(-) diff --git a/src/backend/src/om/entitystorage/SubdomainES.js b/src/backend/src/om/entitystorage/SubdomainES.js index 4a063f582..45cc9473f 100644 --- a/src/backend/src/om/entitystorage/SubdomainES.js +++ b/src/backend/src/om/entitystorage/SubdomainES.js @@ -24,6 +24,8 @@ const { Context } = require("../../util/context"); const { Eq } = require("../query/query"); const { BaseES } = require("./BaseES"); +const PERM_READ_ALL_SUBDOMAINS = 'read-all-subdomains'; + class SubdomainES extends BaseES { static METHODS = { async _on_context_provided () { @@ -52,12 +54,17 @@ class SubdomainES extends BaseES { // Note: we don't need to worry about read; // non-owner users don't have permission to list // but they still have permission to read. - options.predicate = options.predicate.and( - new Eq({ - key: 'owner', - value: user.id, - }), - ); + const svc_permission = this.context.get('services').get('permission'); + const has_permission_to_read_all = await svc_permission.check(Context.get("actor"), PERM_READ_ALL_SUBDOMAINS); + + if (!has_permission_to_read_all) { + options.predicate = options.predicate.and( + new Eq({ + key: 'owner', + value: user.id, + }), + ); + } return await this.upstream.select(options); }, diff --git a/src/backend/src/services/EntityStoreService.js b/src/backend/src/services/EntityStoreService.js index 2bae2f92b..8927a9d54 100644 --- a/src/backend/src/services/EntityStoreService.js +++ b/src/backend/src/services/EntityStoreService.js @@ -122,6 +122,7 @@ class EntityStoreService extends BaseService { if ( id && ! uid ) { const entity = await this.fetch_based_on_complex_id_(id); if ( ! entity ) { + console.log("API Error ID ",id) throw APIError.create('entity_not_found', null, { identifier: id }); diff --git a/src/backend/src/services/worker/WorkerService.js b/src/backend/src/services/worker/WorkerService.js index f1987fb6b..bf552c437 100644 --- a/src/backend/src/services/worker/WorkerService.js +++ b/src/backend/src/services/worker/WorkerService.js @@ -27,9 +27,12 @@ const { getUserInfo } = require("./workerUtils/puterUtils"); const { LLRead } = require("../../filesystem/ll_operations/ll_read"); const { Context } = require("../../util/context"); const { NodePathSelector } = require("../../filesystem/node/selectors"); -const { calculateWorkerName } = require("./workerUtils/nameUtils"); +const { calculateWorkerNameNew } = require("./workerUtils/nameUtils"); const { Entity } = require("../../om/entitystorage/Entity"); const { SKIP_ES_VALIDATION } = require("../../om/entitystorage/consts"); +const { Eq } = require("../../om/query/query"); +const { get_app } = require("../../helpers"); +const { UsernameNotifSelector } = require("../NotificationService"); async function readPuterFile(actor, filePath) { try { @@ -68,28 +71,127 @@ try { } const PREAMBLE_LENGTH = preamble.split("\n").length - 1 class WorkerService extends BaseService { - ['__on_install.routes'](_, { app }) { + _init() { setCloudflareKeys(this.config); + + // Services used const svc_event = this.services.get('event'); - svc_event.on('fs.write.*', (_key, data, meta) => { + const svc_su = this.services.get("su"); + const es_subdomain = this.services.get('es:subdomain'); + const svc_auth = this.services.get("auth"); + const svc_notification = this.services.get('notification'); + + svc_event.on('fs.written.file', async (_key, data, meta) => { + // Code should only run on the same server as the write if (meta.from_outside) return; + + // Check if the file that was written correlates to a worker + const result = await svc_su.sudo(async ()=> { + return await es_subdomain.select({ predicate: new Eq({ key: "root_dir", value: data.node }) }); + }); + if (!result) + return; + + // Person who just wrote file (not necessarily file owner) + const actor = Context.get("actor"); + + // Worker data + const fileData = (await readPuterFile(Context.get("actor"), data.node.path)).toString(); + const workerName = (await result[0].get("subdomain")).split(".").pop(); + + // Get appropriate deploy time auth token to give to the worker + let authToken; + const appOwner = await result[0].get("app_owner"); + if (appOwner) { // If the deployer is an app... + const appID = await appOwner.get("uid"); + console.log("appID", appID) + authToken = await svc_su.sudo(await data.node.get("owner"), async () => { + return await svc_auth.get_user_app_token(appID); + }) + } else { // If the deployer is not attached to any application + authToken = (await svc_auth.create_session_token((await data.node.get("owner")).type.user)).token + } + + + svc_notification.notify( + UsernameNotifSelector(actor.type.user.username), + { + source: 'worker', + title: `Deploying CF worker ${workerName}`, + template: 'user-requesting-share', + fields: { + username: actor.type.user.username, + }, + } + ); + try { + // Create the worker + const cfData = await createWorker((await data.node.get("owner")).type.user, authToken, workerName, preamble + fileData, PREAMBLE_LENGTH); + + // Send user the appropriate notification + if (cfData.success) { + svc_notification.notify( + UsernameNotifSelector(actor.type.user.username), + { + source: 'worker', + title: `Succesfully deployed ${cfData.url}`, + template: 'user-requesting-share', + fields: { + username: actor.type.user.username, + }, + } + ); + } else { + svc_notification.notify( + UsernameNotifSelector(actor.type.user.username), + { + source: 'worker', + title: `Failed to deploy ${workerName}! ${cfData.errors}`, + template: 'user-requesting-share', + fields: { + username: actor.type.user.username, + }, + } + ); + } + + + } catch (e) { + svc_notification.notify( + UsernameNotifSelector(actor.type.user.username), + { + source: 'worker', + title: `Failed to deploy ${workerName}!!\n ${e}`, + template: 'user-requesting-share', + fields: { + username: actor.type.user.username, + }, + } + ); + } }); } static IMPLEMENTS = { ['workers']: { + /** + * + * @param {{filePath: string, workerName: string, authorization: string}} param0 + * @returns {any} + */ async create({ filePath, workerName, authorization }) { try { + workerName = workerName.toLocaleLowerCase(); // just incase + if(!(/^[a-zA-Z0-9_]+$/.test(workerName))) return; + const userData = await getUserInfo(authorization, this.global_config.api_base_url); const actor = Context.get("actor"); const es_subdomain = this.services.get('es:subdomain'); - console.log(actor) const fileData = (await readPuterFile(actor, filePath)).toString(); - const cfData = await createWorker(userData, authorization, workerName, preamble + fileData, PREAMBLE_LENGTH); + const cfData = await createWorker(userData, authorization, calculateWorkerNameNew(userData.uuid, workerName), preamble + fileData, PREAMBLE_LENGTH); await Context.sub({ [SKIP_ES_VALIDATION]: true }).arun(async () => { const entity = await Entity.create({ om: es_subdomain.om }, { - subdomain: "workers.puter." + calculateWorkerName(userData.username, workerName), - owner: userData.id, + subdomain: "workers.puter." + calculateWorkerNameNew(userData.uuid, workerName), root_dir: filePath }); await es_subdomain.upsert(entity); @@ -103,16 +205,25 @@ class WorkerService extends BaseService { }, async destroy({ workerName, authorization }) { try { + workerName = workerName.toLocaleLowerCase(); // just incase + const svc_su = this.services.get("su"); const userData = await getUserInfo(authorization, this.global_config.api_base_url); const cfData = await deleteWorker(userData, authorization, workerName); const es_subdomain = this.services.get('es:subdomain'); - const crudqSubdomain = es_subdomain.as('crud-q'); - await crudq_subdomain.delete({ id: { subdomain: "workers.puter." + calculateWorkerName(userData.username, workerName) } }) + console.log("workers.puter." + calculateWorkerNameNew(userData.uuid, workerName)) + const result = await svc_su.sudo(async () => { + const row = (await es_subdomain.select({ predicate: new Eq({ key: "subdomain", value: "workers.puter." + calculateWorkerNameNew(userData.uuid, workerName) }) })); + return row; + }) + console.log("search result: ", result) + + await es_subdomain.delete(await result[0].get("uid")); return cfData; } catch (e) { + console.error(e); return { success: false, e } } }, diff --git a/src/backend/src/services/worker/workerUtils/cloudflareDeploy.js b/src/backend/src/services/worker/workerUtils/cloudflareDeploy.js index a712c9788..ba785a4b6 100644 --- a/src/backend/src/services/worker/workerUtils/cloudflareDeploy.js +++ b/src/backend/src/services/worker/workerUtils/cloudflareDeploy.js @@ -1,5 +1,5 @@ const fs = require('fs') -const { calculateWorkerName } = require("./nameUtils.js"); +const { calculateWorkerNameNew } = require("./nameUtils.js"); let config = {}; // Constants const CF_BASE_URL = "https://api.cloudflare.com/" @@ -16,9 +16,9 @@ function cfFetch(url, method = "GET", body, givenHeaders) { return fetch(url, { headers, method, body }) } async function getWorker(userData, authorization, workerId) { - await cfFetch(`${WORKERS_BASE_URL}/scripts/${calculateWorkerName(userData.username, workerId)}`, "GET"); + await cfFetch(`${WORKERS_BASE_URL}/scripts/${calculateWorkerNameNew(userData.uuid, workerId)}`, "GET"); } -async function createWorker(userData, authorization, workerId, body, PREAMBLE_LENGTH) { +async function createWorker(userData, authorization, workerName, body, PREAMBLE_LENGTH) { const formData = new FormData(); const workerMetaData = { @@ -42,10 +42,10 @@ async function createWorker(userData, authorization, workerId, body, PREAMBLE_LE } formData.append("metadata", JSON.stringify(workerMetaData)); formData.append("swCode", body); - const cfReturnCodes = await (await cfFetch(`${WORKERS_BASE_URL}/scripts/${calculateWorkerName(userData.username, workerId)}/`, "PUT", formData)).json(); + const cfReturnCodes = await (await cfFetch(`${WORKERS_BASE_URL}/scripts/${workerName}/`, "PUT", formData)).json(); if (cfReturnCodes.success) { - return JSON.stringify({ success: true, errors: [], url: `${calculateWorkerName(userData.username, workerId)}.puter.work` }); + return { success: true, errors: [], url: `${workerName}.puter.work` }; } else { const parsedErrors = []; for (const error of cfReturnCodes.errors) { @@ -71,7 +71,7 @@ async function createWorker(userData, authorization, workerId, body, PREAMBLE_LE parsedErrors.push(finalMessage) } - return JSON.stringify({ success: false, errors: parsedErrors, url: null, body }); + return { success: false, errors: parsedErrors, url: null, body }; } } function setPreambleLength(length) { @@ -87,7 +87,7 @@ function setCloudflareKeys(givenConfig) { } async function deleteWorker(userData, authorization, workerId) { - return await (await cfFetch(`${WORKERS_BASE_URL}/scripts/${calculateWorkerName(userData.username, workerId)}/`, "DELETE")).json(); + return await (await cfFetch(`${WORKERS_BASE_URL}/scripts/${calculateWorkerNameNew(userData.uuid, workerId)}/`, "DELETE")).json(); } diff --git a/src/backend/src/services/worker/workerUtils/nameUtils.js b/src/backend/src/services/worker/workerUtils/nameUtils.js index c0db12d23..46f883cc0 100644 --- a/src/backend/src/services/worker/workerUtils/nameUtils.js +++ b/src/backend/src/services/worker/workerUtils/nameUtils.js @@ -4,11 +4,12 @@ const crypto = require("node:crypto"); function sha1(input) { return crypto.createHash('sha1').update(input, 'utf8').digest().toString("hex").slice(0, 7) } -function calculateWorkerName(username, workerId) { - return `${username}-${sha1(workerId).slice(0, 7)}` -} +function calculateWorkerNameNew(uuid, workerId) { + + return `${workerId}-${uuid.replaceAll("-", "")}` +} module.exports = { sha1, - calculateWorkerName + calculateWorkerNameNew } \ No newline at end of file diff --git a/src/dev-center/index.html b/src/dev-center/index.html index df732a9ba..02fb8d873 100644 --- a/src/dev-center/index.html +++ b/src/dev-center/index.html @@ -241,7 +241,7 @@ - + diff --git a/src/puter-js/src/modules/Workers.js b/src/puter-js/src/modules/Workers.js index c797b60bd..2c59f7378 100644 --- a/src/puter-js/src/modules/Workers.js +++ b/src/puter-js/src/modules/Workers.js @@ -5,13 +5,14 @@ export class WorkersHandler { } async create(workerName, filePath) { + workerName = workerName.toLocaleLowerCase(); // just incase let currentWorkers = await puter.kv.get("user-workers"); if (!currentWorkers) { currentWorkers = {}; } const driverCall = await puter.drivers.call("workers", "worker-service", "create", { authorization: puter.authToken, filePath, workerName }); - const driverResult = JSON.parse(driverCall.result); + const driverResult = driverCall.result; if (!driverCall.success || !driverResult.success) { throw new Error(driverResult?.errors || "Driver failed to execute, do you have the necessary permissions?"); } @@ -21,17 +22,12 @@ export class WorkersHandler { return driverResult; } - // This is temporary until FS stuff is hooked properly - async update(workerName) { - let filePath = (await puter.kv.get("user-workers"))[workerName]["filePath"]; - return this.create(workerName, filePath); - } - async list() { return await puter.kv.get("user-workers"); } async get(workerName) { + workerName = workerName.toLocaleLowerCase(); // just incase try { return (await puter.kv.get("user-workers"))[workerName].url; } catch (e) { @@ -40,6 +36,7 @@ export class WorkersHandler { } async delete(workerName) { + workerName = workerName.toLocaleLowerCase(); // just incase const driverCall = await puter.drivers.call("workers", "worker-service", "destroy", { authorization: puter.authToken, workerName }); if (!driverCall.success || !driverCall.result.result) {