errors

package.json

@@ -15,21 +15,23 @@
   },
   "dependencies": {
     "@hazae41/berith": "^1.1.40",
-    "@hazae41/binary": "^1.2.31",
-    "@hazae41/bytes": "^1.1.14",
-    "@hazae41/cursor": "^1.1.14",
-    "@hazae41/echalote": "^0.2.39",
+    "@hazae41/binary": "^1.2.33",
+    "@hazae41/bytes": "^1.1.19",
+    "@hazae41/cascade": "^1.1.23",
+    "@hazae41/cleaner": "^1.0.0",
+    "@hazae41/cursor": "^1.1.18",
+    "@hazae41/echalote": "^0.2.45",
     "@hazae41/ed25519": "^1.0.0",
-    "@hazae41/fleche": "^1.1.29",
+    "@hazae41/fleche": "^1.1.37",
     "@hazae41/morax": "^1.0.20",
-    "@hazae41/mutex": "^1.2.5",
-    "@hazae41/option": "^1.0.9",
-    "@hazae41/piscine": "^1.0.7",
-    "@hazae41/plume": "^2.0.6",
-    "@hazae41/result": "^1.0.34",
+    "@hazae41/mutex": "^1.2.7",
+    "@hazae41/option": "^1.0.11",
+    "@hazae41/piscine": "^1.0.28",
+    "@hazae41/plume": "^2.0.9",
+    "@hazae41/result": "^1.0.42",
     "@hazae41/sha1": "^1.0.0",
     "@hazae41/x25519": "^1.0.0",
-    "@hazae41/xswr": "^1.5.19",
+    "@hazae41/xswr": "^1.5.22",
     "@heroicons/react": "^2.0.18",
     "@noble/curves": "^1.0.0",
     "@noble/hashes": "^1.3.0",
@@ -53,7 +55,7 @@
     "postcss": "^8.4.24",
     "rimraf": "^5.0.1",
     "tailwindcss": "^3.3.2",
-    "typescript": "5.0.4"
+    "typescript": "5.1.3"
   },
   "watch": {
     "build": {

pages/_app.tsx

@@ -5,7 +5,7 @@ import { Overlay } from "@/mods/overlay/overlay"
 import { GlobalStorageProvider } from "@/mods/storage/global/context"
 import { UserStorageProvider } from "@/mods/storage/user/context"
 import { CircuitsProvider } from "@/mods/tor/circuits/context"
-import { TorProvider } from "@/mods/tor/context"
+import { TorPoolProvider } from "@/mods/tor/context"
 import { SessionsProvider } from "@/mods/tor/sessions/context"
 import '@/styles/globals.css'
 import { CoreProvider } from "@hazae41/xswr"
@@ -42,13 +42,13 @@ export default function App({ Component, pageProps }: AppProps) {
         <GlobalStorageProvider>
           <UserProvider>
             <UserStorageProvider>
-              <TorProvider>
+              <TorPoolProvider>
                 <CircuitsProvider>
                   <SessionsProvider>
                     <Component {...pageProps} />
                   </SessionsProvider>
                 </CircuitsProvider>
-              </TorProvider>
+              </TorPoolProvider>
             </UserStorageProvider>
           </UserProvider>
         </GlobalStorageProvider>

src/libs/bitcoin/address.ts

@@ -6,7 +6,7 @@ import { Sha256 } from "../hashes/sha256"
 
 export namespace Address {
 
-  export async function tryFrom(maybeCompressedPublicKey: Uint8Array) {
+  export async function tryFrom(maybeCompressedPublicKey: Uint8Array): Promise<Result<string, unknown>> {
     return await Result.unthrow(async t => {
       const sha256 = await Sha256.digest(maybeCompressedPublicKey)
       const ripemd160 = Ripemd160.digest(sha256)

src/libs/objects/objects.ts

@@ -10,6 +10,10 @@ export namespace Objects {
     return Object.entries(object) as [K, T][]
   }
 
+  export function values<K extends keyof any, T>(object: Record<K, T>) {
+    return Object.values(object) as T[]
+  }
+
   export async function mapValues<K extends keyof any, I, O>(object: Record<K, I>, mapper: (v: I) => Promiseable<O>) {
     return fromEntries(await Promise.all(entries(object).map(async ([k, v]) => [k, await mapper(v)])))
   }

src/libs/pools/pools.ts

@@ -1,27 +1,12 @@
-import { Mutex } from "@hazae41/mutex"
-import { Pool } from "@hazae41/piscine"
+import { Pool, PoolEntry } from "@hazae41/piscine"
 import { Result } from "@hazae41/result"
 import { useEffect } from "react"
 
-export namespace Pools {
-
-  export async function takeLocked<T>(pool: Pool<T>) {
-    const element = await pool.cryptoRandom()
-    pool.delete(element)
-    return element
-  }
-
-  export async function take<T>(pool: Mutex<Pool<T>>) {
-    return await pool.lock(takeLocked)
-  }
-
-}
-
-export function usePoolChange<T>(pool: Pool<T> | undefined, callback: (pool: Pool<T>) => Result<void, unknown>) {
+export function usePoolChange<T, E>(pool: Pool<T, E> | undefined, callback: (pool: Pool<T, E>, entry: PoolEntry<T, E>) => Result<void, unknown>) {
   useEffect(() => {
     if (!pool) return
 
-    const onCreatedOrDeleted = () => callback(pool)
+    const onCreatedOrDeleted = (entry: PoolEntry<T, E>) => callback(pool, entry)
 
     pool.events.on("created", onCreatedOrDeleted, { passive: true })
     pool.events.on("deleted", onCreatedOrDeleted, { passive: true })

src/libs/rpc/rpc.ts

@@ -1,5 +1,6 @@
 import { Future } from "@hazae41/future"
-import { Sockets } from "../sockets/sockets"
+import { AbortError, CloseError, ErrorError } from "@hazae41/plume"
+import { Err, Ok, Result } from "@hazae41/result"
 import { RpcRequest, RpcRequestInit } from "./request"
 import { Response } from "./response"
 
@@ -31,8 +32,8 @@ export class Client {
     return fetch<T>(input, { ...rest, ...request })
   }
 
-  async fetchWithSocket<T>(socket: WebSocket, request: RpcRequestInit, signal?: AbortSignal) {
-    return await fetchWithSocket<T>(socket, this.new(request), signal)
+  async tryFetchWithSocket<T>(socket: WebSocket, request: RpcRequestInit, signal: AbortSignal) {
+    return await tryFetchWithSocket<T>(socket, this.new(request), signal)
   }
 
 }
@@ -63,19 +64,47 @@ export async function fetch<T>(input: RequestInfo | URL, init: RequestInit & Rpc
   return response
 }
 
-export async function fetchWithSocket<T>(socket: WebSocket, request: RpcRequest, signal?: AbortSignal) {
+export async function tryFetchWithSocket<T>(socket: WebSocket, request: RpcRequest, signal: AbortSignal) {
   socket.send(JSON.stringify(request))
 
-  const future = new Future<Response<T>>()
+  const future = new Future<Result<Response<T>, CloseError | ErrorError | AbortError>>()
 
-  const onEvent = async (event: Event) => {
+  const onMessage = async (event: Event) => {
     const msgEvent = event as MessageEvent<string>
     const response = Response.from<T>(JSON.parse(msgEvent.data))
 
-    if (response.id !== request.id) return
+    if (response.id !== request.id)
+      return
+    future.resolve(new Ok(response))
+  }
+
+  const onError = (e: unknown) => {
+    const result = new Err(ErrorError.from(e))
+    future.resolve(result)
+  }
 
-    future.resolve(response)
+  const onClose = (e: unknown) => {
+    const result = new Err(CloseError.from(e))
+    future.resolve(result)
   }
 
-  return await Sockets.waitMap(socket, "message", { future, onEvent, signal })
+  const onAbort = () => {
+    socket.close()
+    const result = new Err(AbortError.from(signal.reason))
+    future.resolve(result)
+  }
+
+  try {
+    socket.addEventListener("message", onMessage, { passive: true })
+    socket.addEventListener("close", onClose, { passive: true })
+    socket.addEventListener("error", onError, { passive: true })
+    signal.addEventListener("abort", onAbort, { passive: true })
+
+    return await future.promise
+  } finally {
+    socket.removeEventListener("message", onMessage)
+    socket.removeEventListener("close", onClose)
+    socket.removeEventListener("error", onError)
+    signal.removeEventListener("abort", onAbort)
+  }
 }

src/libs/signals/signals.ts

@@ -0,0 +1,26 @@
+export namespace AbortSignals {
+
+  export function timeout(delay: number, parent?: AbortSignal) {
+    return merge(AbortSignal.timeout(delay), parent)
+  }
+
+  export function merge(a: AbortSignal, b?: AbortSignal) {
+    if (b === undefined)
+      return a
+
+    const c = new AbortController()
+
+    const onAbort = (reason?: unknown) => {
+      c.abort(reason)
+
+      a.removeEventListener("abort", onAbort)
+      b.removeEventListener("abort", onAbort)
+    }
+
+    a.addEventListener("abort", onAbort, { passive: true })
+    b.addEventListener("abort", onAbort, { passive: true })
+
+    return c.signal
+  }
+
+}

src/libs/sockets/sockets.ts

@@ -1,34 +1,45 @@
 import { Future } from "@hazae41/future";
+import { AbortError, CloseError, ErrorError } from "@hazae41/plume";
+import { Err, Ok, Result } from "@hazae41/result";
 
 export namespace Sockets {
 
-  export interface WaitParams<T> {
-    future: Future<T>,
-    onEvent: (event: Event) => void,
-    signal?: AbortSignal
-  }
+  export async function tryWaitOpen(socket: WebSocket, signal: AbortSignal) {
+    const future = new Future<Result<void, CloseError | ErrorError | AbortError>>()
 
-  export async function wait(socket: WebSocket, type: string, signal?: AbortSignal) {
-    const future = new Future<Event>()
-    const onEvent = (e: Event) => future.resolve(e)
-    return await waitMap(socket, type, { future, onEvent, signal })
-  }
+    const onOpen = () => {
+      const result = Ok.void()
+      future.resolve(result)
+    }
+
+    const onError = (e: unknown) => {
+      const result = new Err(ErrorError.from(e))
+      future.resolve(result)
+    }
+
+    const onClose = (e: unknown) => {
+      const result = new Err(CloseError.from(e))
+      future.resolve(result)
+    }
 
-  export async function waitMap<T>(socket: WebSocket, type: string, params: WaitParams<T>) {
-    const { future, onEvent, signal } = params
+    const onAbort = () => {
+      socket.close()
+      const result = new Err(AbortError.from(signal.reason))
+      future.resolve(result)
+    }
 
     try {
-      signal?.addEventListener("abort", future.reject)
-      socket.addEventListener("error", future.reject)
-      socket.addEventListener("close", future.reject)
-      socket.addEventListener(type, onEvent)
+      socket.addEventListener("open", onOpen, { passive: true })
+      socket.addEventListener("close", onClose, { passive: true })
+      socket.addEventListener("error", onError, { passive: true })
+      signal.addEventListener("abort", onAbort, { passive: true })
 
       return await future.promise
     } finally {
-      signal?.removeEventListener("abort", future.reject)
-      socket.removeEventListener("error", future.reject)
-      socket.removeEventListener("close", future.reject)
-      socket.removeEventListener(type, onEvent)
+      socket.removeEventListener("open", onOpen,)
+      socket.removeEventListener("close", onClose)
+      socket.removeEventListener("error", onError)
+      signal.removeEventListener("abort", onAbort)
     }
   }
 

src/libs/tor/circuits/circuits.ts

@@ -0,0 +1,31 @@
+import { Circuit, TorClientDuplex, createPooledCircuit } from "@hazae41/echalote"
+import { Mutex } from "@hazae41/mutex"
+import { Pool, PoolParams } from "@hazae41/piscine"
+import { Ok, Result } from "@hazae41/result"
+
+export function createCircuitPool<TorPoolError>(tors: Mutex<Pool<TorClientDuplex, TorPoolError>>, params: PoolParams = {}) {
+  const pool = new Mutex(new Pool<Circuit, Error | TorPoolError>(async (params) => {
+    return await Result.unthrow(async t => {
+      const { index, signal } = params
+
+      const tor = await tors.inner.tryGet(index % tors.inner.capacity).then(r => r.throw(t))
+      const circuit = await tor.tryCreateAndExtendLoop(signal).then(r => r.throw(t))
+
+      return new Ok(createPooledCircuit(circuit, params))
+    })
+  }, params))
+
+  pool.inner.signal.addEventListener("abort", async (reason) => {
+    tors.inner.abort(reason)
+
+    return Ok.void()
+  }, { passive: true, once: true })
+
+  tors.inner.signal.addEventListener("abort", async (reason) => {
+    pool.inner.abort(reason)
+
+    return Ok.void()
+  }, { passive: true, once: true })
+
+  return pool
+}