Merge remote-tracking branch 'origin/main' into nip05-db

deno.json

@@ -46,7 +46,6 @@
     "@gfx/canvas-wasm": "jsr:@gfx/canvas-wasm@^0.4.2",
     "@hono/hono": "jsr:@hono/hono@^4.4.6",
     "@isaacs/ttlcache": "npm:@isaacs/ttlcache@^1.4.1",
-    "@lambdalisue/async": "jsr:@lambdalisue/async@^2.1.1",
     "@negrel/webpush": "jsr:@negrel/webpush@^0.3.0",
     "@noble/secp256k1": "npm:@noble/secp256k1@^2.0.0",
     "@nostrify/db": "jsr:@nostrify/db@^0.38.0",

deno.lock

@@ -3,6 +3,7 @@
   "specifiers": {
     "jsr:@b-fuze/deno-dom@~0.1.47": "0.1.48",
     "jsr:@bradenmacdonald/s3-lite-client@~0.7.4": "0.7.6",
+    "jsr:@core/asyncutil@^1.2.0": "1.2.0",
     "jsr:@denosaurs/plug@1.0.3": "1.0.3",
     "jsr:@esroyo/scoped-performance@^3.1.0": "3.1.0",
     "jsr:@gfx/canvas-wasm@~0.4.2": "0.4.2",
@@ -28,7 +29,6 @@
     "jsr:@gleasonator/policy@0.9.3": "0.9.3",
     "jsr:@gleasonator/policy@0.9.4": "0.9.4",
     "jsr:@hono/hono@^4.4.6": "4.6.15",
-    "jsr:@lambdalisue/async@^2.1.1": "2.1.1",
     "jsr:@negrel/http-ece@0.6.0": "0.6.0",
     "jsr:@negrel/webpush@0.3": "0.3.0",
     "jsr:@nostrify/db@0.38": "0.38.0",
@@ -153,6 +153,9 @@
         "jsr:@std/io@0.224"
       ]
     },
+    "@core/asyncutil@1.2.0": {
+      "integrity": "9967f15190c60df032c13f72ce5ac73d185c34f31c53dc918d8800025854c118"
+    },
     "@denosaurs/plug@1.0.3": {
       "integrity": "b010544e386bea0ff3a1d05e0c88f704ea28cbd4d753439c2f1ee021a85d4640",
       "dependencies": [
@@ -337,9 +340,6 @@
     "@hono/hono@4.6.15": {
       "integrity": "935b3b12e98e4b22bcd1aa4dbe6587321e431c79829eba61f535b4ede39fd8b1"
     },
-    "@lambdalisue/async@2.1.1": {
-      "integrity": "1fc9bc6f4ed50215cd2f7217842b18cea80f81c25744f88f8c5eb4be5a1c9ab4"
-    },
     "@negrel/http-ece@0.6.0": {
       "integrity": "7afdd81b86ea5b21a9677b323c01c3338705e11cc2bfed250870f5349d8f86f7",
       "dependencies": [
@@ -2371,7 +2371,6 @@
       "jsr:@esroyo/scoped-performance@^3.1.0",
       "jsr:@gfx/canvas-wasm@~0.4.2",
       "jsr:@hono/hono@^4.4.6",
-      "jsr:@lambdalisue/async@^2.1.1",
       "jsr:@negrel/webpush@0.3",
       "jsr:@nostrify/db@0.38",
       "jsr:@nostrify/nostrify@~0.38.1",

scripts/db-import.ts

@@ -1,4 +1,4 @@
-import { Semaphore } from '@lambdalisue/async';
+import { Semaphore } from '@core/asyncutil';
 import { NostrEvent } from '@nostrify/nostrify';
 import { JsonParseStream } from '@std/json/json-parse-stream';
 import { TextLineStream } from '@std/streams/text-line-stream';

scripts/db-populate-nip05.ts

@@ -1,4 +1,4 @@
-import { Semaphore } from '@lambdalisue/async';
+import { Semaphore } from '@core/asyncutil';
 
 import { updateAuthorData } from '@/pipeline.ts';
 import { Storages } from '@/storages.ts';

src/firehose.ts

@@ -1,4 +1,4 @@
-import { Semaphore } from '@lambdalisue/async';
+import { Semaphore } from '@core/asyncutil';
 import { logi } from '@soapbox/logi';
 
 import { Conf } from '@/config.ts';

src/middleware/translatorMiddleware.ts

@@ -1,6 +1,7 @@
+import { safeFetch } from '@soapbox/safe-fetch';
+
 import { AppMiddleware } from '@/app.ts';
 import { Conf } from '@/config.ts';
-import { fetchWorker } from '@/workers/fetch.ts';
 import { DeepLTranslator } from '@/translators/DeepLTranslator.ts';
 import { LibreTranslateTranslator } from '@/translators/LibreTranslateTranslator.ts';
 
@@ -10,7 +11,7 @@ export const translatorMiddleware: AppMiddleware = async (c, next) => {
     case 'deepl': {
       const { deeplApiKey: apiKey, deeplBaseUrl: baseUrl } = Conf;
       if (apiKey) {
-        c.set('translator', new DeepLTranslator({ baseUrl, apiKey, fetch: fetchWorker }));
+        c.set('translator', new DeepLTranslator({ baseUrl, apiKey, fetch: safeFetch }));
       }
       break;
     }
@@ -18,7 +19,7 @@ export const translatorMiddleware: AppMiddleware = async (c, next) => {
     case 'libretranslate': {
       const { libretranslateApiKey: apiKey, libretranslateBaseUrl: baseUrl } = Conf;
       if (apiKey) {
-        c.set('translator', new LibreTranslateTranslator({ baseUrl, apiKey, fetch: fetchWorker }));
+        c.set('translator', new LibreTranslateTranslator({ baseUrl, apiKey, fetch: safeFetch }));
       }
       break;
     }

src/middleware/uploaderMiddleware.ts

@@ -1,11 +1,11 @@
 import { BlossomUploader, NostrBuildUploader } from '@nostrify/nostrify/uploaders';
+import { safeFetch } from '@soapbox/safe-fetch';
 
 import { AppMiddleware } from '@/app.ts';
 import { Conf } from '@/config.ts';
 import { DenoUploader } from '@/uploaders/DenoUploader.ts';
 import { IPFSUploader } from '@/uploaders/IPFSUploader.ts';
 import { S3Uploader } from '@/uploaders/S3Uploader.ts';
-import { fetchWorker } from '@/workers/fetch.ts';
 
 /** Set an uploader for the user. */
 export const uploaderMiddleware: AppMiddleware = async (c, next) => {
@@ -29,17 +29,17 @@ export const uploaderMiddleware: AppMiddleware = async (c, next) => {
       );
       break;
     case 'ipfs':
-      c.set('uploader', new IPFSUploader({ baseUrl: Conf.mediaDomain, apiUrl: Conf.ipfs.apiUrl, fetch: fetchWorker }));
+      c.set('uploader', new IPFSUploader({ baseUrl: Conf.mediaDomain, apiUrl: Conf.ipfs.apiUrl, fetch: safeFetch }));
       break;
     case 'local':
       c.set('uploader', new DenoUploader({ baseUrl: Conf.mediaDomain, dir: Conf.uploadsDir }));
       break;
     case 'nostrbuild':
-      c.set('uploader', new NostrBuildUploader({ endpoint: Conf.nostrbuildEndpoint, signer, fetch: fetchWorker }));
+      c.set('uploader', new NostrBuildUploader({ endpoint: Conf.nostrbuildEndpoint, signer, fetch: safeFetch }));
       break;
     case 'blossom':
       if (signer) {
-        c.set('uploader', new BlossomUploader({ servers: Conf.blossomServers, signer, fetch: fetchWorker }));
+        c.set('uploader', new BlossomUploader({ servers: Conf.blossomServers, signer, fetch: safeFetch }));
       }
       break;
   }

src/notify.ts

@@ -1,4 +1,4 @@
-import { Semaphore } from '@lambdalisue/async';
+import { Semaphore } from '@core/asyncutil';
 
 import { pipelineEncounters } from '@/caches/pipelineEncounters.ts';
 import { Conf } from '@/config.ts';

src/utils/favicon.ts

@@ -1,5 +1,6 @@
 import { DOMParser } from '@b-fuze/deno-dom';
 import { logi } from '@soapbox/logi';
+import { safeFetch } from '@soapbox/safe-fetch';
 import { Kysely } from 'kysely';
 import tldts from 'tldts';
 
@@ -9,7 +10,6 @@ import { cachedFaviconsSizeGauge } from '@/metrics.ts';
 import { Storages } from '@/storages.ts';
 import { nostrNow } from '@/utils.ts';
 import { SimpleLRU } from '@/utils/SimpleLRU.ts';
-import { fetchWorker } from '@/workers/fetch.ts';
 
 export const faviconCache = new SimpleLRU<string, URL>(
   async (domain, { signal }) => {
@@ -58,7 +58,7 @@ async function fetchFavicon(domain: string, signal?: AbortSignal): Promise<URL>
   }
 
   const rootUrl = new URL('/', `https://${domain}/`);
-  const response = await fetchWorker(rootUrl, { signal });
+  const response = await safeFetch(rootUrl, { signal });
   const html = await response.text();
 
   const doc = new DOMParser().parseFromString(html, 'text/html');
@@ -88,7 +88,7 @@ async function fetchFavicon(domain: string, signal?: AbortSignal): Promise<URL>
 
   // Fallback to checking `/favicon.ico` of the domain.
   const url = new URL('/favicon.ico', `https://${domain}/`);
-  const fallback = await fetchWorker(url, { method: 'HEAD', signal });
+  const fallback = await safeFetch(url, { method: 'HEAD', signal });
   const contentType = fallback.headers.get('content-type');
 
   if (fallback.ok && contentType === 'image/vnd.microsoft.icon') {

src/utils/lnurl.ts

@@ -1,19 +1,19 @@
 import { NostrEvent } from '@nostrify/nostrify';
 import { LNURL, LNURLDetails } from '@nostrify/nostrify/ln';
 import { logi } from '@soapbox/logi';
+import { safeFetch } from '@soapbox/safe-fetch';
 import { JsonValue } from '@std/json';
 
 import { cachedLnurlsSizeGauge } from '@/metrics.ts';
 import { SimpleLRU } from '@/utils/SimpleLRU.ts';
 import { errorJson } from '@/utils/log.ts';
 import { Time } from '@/utils/time.ts';
-import { fetchWorker } from '@/workers/fetch.ts';
 
 const lnurlCache = new SimpleLRU<string, LNURLDetails>(
   async (lnurl, { signal }) => {
     logi({ level: 'info', ns: 'ditto.lnurl', lnurl, state: 'started' });
     try {
-      const details = await LNURL.lookup(lnurl, { fetch: fetchWorker, signal });
+      const details = await LNURL.lookup(lnurl, { fetch: safeFetch, signal });
       logi({ level: 'info', ns: 'ditto.lnurl', lnurl, state: 'found', details: details as unknown as JsonValue });
       return details;
     } catch (e) {
@@ -62,7 +62,7 @@ async function getInvoice(params: CallbackParams, signal?: AbortSignal): Promise
   const { pr } = await LNURL.callback(
     details.callback,
     params,
-    { fetch: fetchWorker, signal },
+    { fetch: safeFetch, signal },
   );
 
   return pr;

src/utils/nip05.ts

@@ -1,6 +1,7 @@
 import { nip19 } from 'nostr-tools';
 import { NIP05, NStore } from '@nostrify/nostrify';
 import { logi } from '@soapbox/logi';
+import { safeFetch } from '@soapbox/safe-fetch';
 import tldts from 'tldts';
 
 import { Conf } from '@/config.ts';
@@ -8,7 +9,6 @@ import { cachedNip05sSizeGauge } from '@/metrics.ts';
 import { Storages } from '@/storages.ts';
 import { errorJson } from '@/utils/log.ts';
 import { SimpleLRU } from '@/utils/SimpleLRU.ts';
-import { fetchWorker } from '@/workers/fetch.ts';
 
 export const nip05Cache = new SimpleLRU<string, nip19.ProfilePointer>(
   async (nip05, { signal }) => {
@@ -43,7 +43,7 @@ async function getNip05(
         throw new Error(`Not found: ${nip05}`);
       }
     } else {
-      const pointer = await NIP05.lookup(nip05, { fetch: fetchWorker, signal });
+      const pointer = await NIP05.lookup(nip05, { fetch: safeFetch, signal });
       logi({ level: 'info', ns: 'ditto.nip05', nip05, state: 'found', source: 'fetch', pubkey: pointer.pubkey });
       return pointer;
     }

src/utils/unfurl.ts

@@ -1,5 +1,6 @@
 import TTLCache from '@isaacs/ttlcache';
 import { logi } from '@soapbox/logi';
+import { safeFetch } from '@soapbox/safe-fetch';
 import DOMPurify from 'isomorphic-dompurify';
 import { unfurl } from 'unfurl.js';
 
@@ -7,13 +8,12 @@ import { Conf } from '@/config.ts';
 import { PreviewCard } from '@/entities/PreviewCard.ts';
 import { cachedLinkPreviewSizeGauge } from '@/metrics.ts';
 import { errorJson } from '@/utils/log.ts';
-import { fetchWorker } from '@/workers/fetch.ts';
 
 async function unfurlCard(url: string, signal: AbortSignal): Promise<PreviewCard | null> {
   try {
     const result = await unfurl(url, {
       fetch: (url) =>
-        fetchWorker(url, {
+        safeFetch(url, {
           headers: {
             'Accept': 'text/html, application/xhtml+xml',
             'User-Agent': Conf.fetchUserAgent,

src/views/mastodon/accounts.ts

@@ -69,7 +69,7 @@ function renderAccount(event: Omit<DittoEvent, 'id' | 'sig'>, opts: ToAccountOpt
       streakEnd = null;
     } else {
       const delta = streakEnd - streakStart;
-      streakDays = Math.max(Math.ceil(delta / streakWindow), 1);
+      streakDays = Math.max(Math.ceil(delta / 86400), 1);
     }
   }
 

src/workers/fetch.test.ts

@@ -1,29 +0,0 @@
-import { assertEquals, assertRejects } from '@std/assert';
-
-import { fetchWorker } from '@/workers/fetch.ts';
-
-Deno.test({
-  name: 'fetchWorker',
-  async fn() {
-    const response = await fetchWorker('https://httpbingo.org/get');
-    const json = await response.json();
-    assertEquals(json.headers.Host, ['httpbingo.org']);
-  },
-  sanitizeResources: false,
-});
-
-Deno.test({
-  name: 'fetchWorker with AbortSignal',
-  async fn() {
-    const controller = new AbortController();
-    const signal = controller.signal;
-
-    setTimeout(() => controller.abort(), 100);
-    assertRejects(() => fetchWorker('https://httpbingo.org/delay/10', { signal }));
-
-    await new Promise<void>((resolve) => {
-      signal.addEventListener('abort', () => resolve(), { once: true });
-    });
-  },
-  sanitizeResources: false,
-});

src/workers/fetch.ts

@@ -1,90 +0,0 @@
-import * as Comlink from 'comlink';
-
-import { FetchWorker } from './fetch.worker.ts';
-import './handlers/abortsignal.ts';
-
-import { fetchResponsesCounter } from '@/metrics.ts';
-
-const worker = new Worker(new URL('./fetch.worker.ts', import.meta.url), { type: 'module', name: 'fetchWorker' });
-const client = Comlink.wrap<typeof FetchWorker>(worker);
-
-// Wait for the worker to be ready before we start using it.
-const ready = new Promise<void>((resolve) => {
-  const handleEvent = () => {
-    self.removeEventListener('message', handleEvent);
-    resolve();
-  };
-  worker.addEventListener('message', handleEvent);
-});
-
-/**
- * Fetch implementation with a Web Worker.
- * Calling this performs the fetch in a separate CPU thread so it doesn't block the main thread.
- */
-const fetchWorker: typeof fetch = async (...args) => {
-  await ready;
-
-  const [url, init] = serializeFetchArgs(args);
-  const { body, signal, ...rest } = init;
-
-  if (signal?.aborted) {
-    throw new DOMException('The signal has been aborted', 'AbortError');
-  }
-
-  const result = await client.fetch(url, { ...rest, body: await prepareBodyForWorker(body) }, signal);
-  const response = new Response(...result);
-
-  const { method } = init;
-  const { status } = response;
-  fetchResponsesCounter.inc({ method, status });
-
-  return response;
-};
-
-/** Take arguments to `fetch`, and turn them into something we can send over Comlink. */
-function serializeFetchArgs(args: Parameters<typeof fetch>): [string, RequestInit] {
-  const request = normalizeRequest(args);
-  const init = requestToInit(request);
-  return [request.url, init];
-}
-
-/** Get a `Request` object from arguments to `fetch`. */
-function normalizeRequest(args: Parameters<typeof fetch>): Request {
-  return new Request(...args);
-}
-
-/** Get the body as a type we can transfer over Web Workers. */
-async function prepareBodyForWorker(
-  body: BodyInit | undefined | null,
-): Promise<ArrayBuffer | Blob | string | undefined | null> {
-  if (!body || typeof body === 'string' || body instanceof ArrayBuffer || body instanceof Blob) {
-    return body;
-  } else {
-    const response = new Response(body);
-    return await response.arrayBuffer();
-  }
-}
-
-/**
- * Convert a `Request` object into its serialized `RequestInit` format.
- * `RequestInit` is a subset of `Request`, just lacking helper methods like `json()`,
- * making it easier to serialize (exceptions: `body` and `signal`).
- */
-function requestToInit(request: Request): RequestInit {
-  return {
-    method: request.method,
-    headers: [...request.headers.entries()],
-    body: request.body,
-    referrer: request.referrer,
-    referrerPolicy: request.referrerPolicy,
-    mode: request.mode,
-    credentials: request.credentials,
-    cache: request.cache,
-    redirect: request.redirect,
-    integrity: request.integrity,
-    keepalive: request.keepalive,
-    signal: request.signal,
-  };
-}
-
-export { fetchWorker };

src/workers/fetch.worker.ts

@@ -1,46 +0,0 @@
-/// <reference lib="webworker" />
-
-import { safeFetch } from '@soapbox/safe-fetch';
-import { logi } from '@soapbox/logi';
-import * as Comlink from 'comlink';
-
-import '@/workers/handlers/abortsignal.ts';
-import '@/sentry.ts';
-
-export const FetchWorker = {
-  async fetch(
-    url: string,
-    init: Omit<RequestInit, 'signal'>,
-    signal: AbortSignal | null | undefined,
-  ): Promise<[BodyInit, ResponseInit]> {
-    if (signal?.aborted) {
-      throw new DOMException('The signal has been aborted', 'AbortError');
-    }
-
-    logi({ level: 'debug', ns: 'ditto.fetch', state: 'started', method: init.method ?? 'GET', url });
-
-    const response = await safeFetch(url, { ...init, signal });
-
-    logi({
-      level: 'debug',
-      ns: 'ditto.fetch',
-      state: 'finished',
-      method: init.method ?? 'GET',
-      url,
-      status: response.status,
-    });
-
-    return [
-      await response.arrayBuffer(),
-      {
-        status: response.status,
-        statusText: response.statusText,
-        headers: [...response.headers.entries()],
-      },
-    ];
-  },
-};
-
-Comlink.expose(FetchWorker);
-
-self.postMessage('ready');

src/workers/handlers/abortsignal.ts

@@ -1,46 +0,0 @@
-import * as Comlink from 'comlink';
-
-const signalFinalizers = new FinalizationRegistry((port: MessagePort) => {
-  port.postMessage(null);
-  port.close();
-});
-
-Comlink.transferHandlers.set('abortsignal', {
-  canHandle(value) {
-    return value instanceof AbortSignal || value?.constructor?.name === 'AbortSignal';
-  },
-  serialize(signal) {
-    if (signal.aborted) {
-      return [{ aborted: true }];
-    }
-
-    const { port1, port2 } = new MessageChannel();
-    signal.addEventListener(
-      'abort',
-      () => port1.postMessage({ reason: signal.reason }),
-      { once: true },
-    );
-
-    signalFinalizers?.register(signal, port1);
-
-    return [{ aborted: false, port: port2 }, [port2]];
-  },
-  deserialize({ aborted, port }) {
-    if (aborted || !port) {
-      return AbortSignal.abort();
-    }
-
-    const ctrl = new AbortController();
-
-    port.addEventListener('message', (ev) => {
-      if (ev.data && 'reason' in ev.data) {
-        ctrl.abort(ev.data.reason);
-      }
-      port.close();
-    }, { once: true });
-
-    port.start();
-
-    return ctrl.signal;
-  },
-} as Comlink.TransferHandler<AbortSignal, { aborted: boolean; port?: MessagePort }>);

src/workers/policy.ts

@@ -5,8 +5,6 @@ import * as Comlink from 'comlink';
 import { Conf } from '@/config.ts';
 import type { CustomPolicy } from '@/workers/policy.worker.ts';
 
-import '@/workers/handlers/abortsignal.ts';
-
 class PolicyWorker implements NPolicy {
   private worker: Comlink.Remote<CustomPolicy>;
   private ready: Promise<void>;

src/workers/policy.worker.ts

@@ -6,8 +6,6 @@ import * as Comlink from 'comlink';
 import { DittoDB } from '@/db/DittoDB.ts';
 import { EventsDB } from '@/storages/EventsDB.ts';
 
-import '@/workers/handlers/abortsignal.ts';
-
 // @ts-ignore Don't try to access the env from this worker.
 Deno.env = new Map<string, string>();