stamp(ext/workers): sync apis with rust land

Author: nyannyacha

ext/workers/user_workers.js

@@ -1,4 +1,5 @@
 import { primordials, core } from "ext:core/mod.js";
+import { SymbolDispose } from "ext:deno_web/00_infra.js";
 import { readableStreamForRid, writableStreamForRid } from "ext:deno_web/06_streams.js";
 import { getSupabaseTag } from "ext:sb_core_main_js/js/http.js";
 
@@ -9,6 +10,8 @@ const { TypeError } = primordials;
 const {
 	op_user_worker_fetch_send,
 	op_user_worker_create,
+	op_user_user_worker_wait_token_cancelled,
+	op_user_worker_is_active,
 } = ops;
 
 const NO_SUPABASE_TAG_WARN_MSG = `Unable to find the supabase tag from the request instance.\n\
@@ -24,8 +27,34 @@ function redirectStatus(status) {
 }
 
 class UserWorker {
-	constructor(key) {
-		this.key = key;
+	/** @type {string} */
+	#key = "";
+
+	/** @type {number | null} */
+	#rid = null;
+
+	/** @type {boolean} */
+	#disposed = false;
+
+	/** 
+	 * @param {string} key
+	 * @param {number} rid
+	 */
+	constructor(key, rid) {
+		this.#key = key;
+		this.#rid = rid;
+
+		// deno-lint-ignore no-this-alias
+		const self = this;
+
+		setTimeout(async () => {
+			try {
+				await op_user_user_worker_wait_token_cancelled(rid);
+				self.dispose();
+			} catch {
+				// TODO(Nyannyacha): Link it with the tracing for telemetry.
+			}
+		});
 	}
 
 	async fetch(request, options = {}) {
@@ -62,7 +91,7 @@ class UserWorker {
 		}
 
 		const responsePromise = op_user_worker_fetch_send(
-			this.key,
+			this.#key,
 			requestRid,
 			requestBodyRid,
 			tag.streamRid,
@@ -75,6 +104,7 @@ class UserWorker {
 		]);
 
 		if (requestBodyPromiseResult.status === "rejected") {
+			// TODO(Nyannyacha): Link it with the tracing for telemetry.
 			// console.warn(requestBodyPromiseResult.reason);
 		}
 
@@ -114,6 +144,26 @@ class UserWorker {
 		});
 	}
 
+	/** @returns {boolean} */
+	get active() {
+		if (this.#disposed) {
+			return false;
+		}
+
+		return op_user_worker_is_active(this.#rid);
+	}
+
+	dispose() {
+		if (!this.#disposed) {
+			core.tryClose(this.#rid);
+			this.#disposed = true;
+		}
+	}
+
+	[SymbolDispose]() {
+		this.dispose();
+	}
+
 	static async create(opts) {
 		const readyOptions = {
 			noModuleCache: false,
@@ -136,9 +186,9 @@ class UserWorker {
 			throw new TypeError("service path must be defined");
 		}
 
-		const key = await op_user_worker_create(readyOptions);
+		const [key, rid] = await op_user_worker_create(readyOptions);
 
-		return new UserWorker(key);
+		return new UserWorker(key, rid);
 	}
 }