Merge remote-tracking branch 'origin/main' into ws-limit

.gitlab-ci.yml

@@ -1,4 +1,4 @@
-image: denoland/deno:1.44.2
+image: denoland/deno:1.45.0
 
 default:
   interruptible: true

.tool-versions

@@ -1 +1 @@
-deno 1.44.2
+deno 1.45.0

deno.json

@@ -6,6 +6,7 @@
     "dev": "deno run -A --watch src/server.ts",
     "hook": "deno run --allow-read --allow-run --allow-write https://deno.land/x/deno_hooks@0.1.1/mod.ts",
     "db:migrate": "deno run -A scripts/db-migrate.ts",
+    "nostr:pull": "deno run -A scripts/nostr-pull.ts",
     "debug": "deno run -A --inspect src/server.ts",
     "test": "DATABASE_URL=\"sqlite://:memory:\" deno test -A --junit-path=./deno-test.xml",
     "check": "deno check src/server.ts",
@@ -26,7 +27,7 @@
     "@hono/hono": "jsr:@hono/hono@^4.4.6",
     "@isaacs/ttlcache": "npm:@isaacs/ttlcache@^1.4.1",
     "@noble/secp256k1": "npm:@noble/secp256k1@^2.0.0",
-    "@nostrify/nostrify": "jsr:@nostrify/nostrify@^0.23.3",
+    "@nostrify/nostrify": "jsr:@nostrify/nostrify@^0.25.0",
     "@scure/base": "npm:@scure/base@^1.1.6",
     "@sentry/deno": "https://deno.land/x/sentry@7.112.2/index.mjs",
     "@soapbox/kysely-deno-sqlite": "jsr:@soapbox/kysely-deno-sqlite@^2.1.0",
@@ -50,7 +51,7 @@
     "iso-639-1": "npm:iso-639-1@2.1.15",
     "isomorphic-dompurify": "npm:isomorphic-dompurify@^2.11.0",
     "kysely": "npm:kysely@^0.27.3",
-    "kysely_deno_postgres": "https://gitlab.com/soapbox-pub/kysely-deno-postgres/-/raw/f2948b86190a10faa293588775e162b3a8b52e70/mod.ts",
+    "kysely_deno_postgres": "https://gitlab.com/soapbox-pub/kysely-deno-postgres/-/raw/c6869b9e12d74af78a846ad503d84493f5db9df4/mod.ts",
     "light-bolt11-decoder": "npm:light-bolt11-decoder",
     "linkify-plugin-hashtag": "npm:linkify-plugin-hashtag@^4.1.1",
     "linkify-string": "npm:linkify-string@^4.1.1",
@@ -59,6 +60,8 @@
     "nostr-relaypool": "npm:nostr-relaypool2@0.6.34",
     "nostr-tools": "npm:nostr-tools@2.5.1",
     "nostr-wasm": "npm:nostr-wasm@^0.1.0",
+    "postgres": "https://deno.land/x/postgres@v0.19.0/mod.ts",
+    "prom-client": "npm:prom-client@^15.1.2",
     "question-deno": "https://raw.githubusercontent.com/ocpu/question-deno/10022b8e52555335aa510adb08b0a300df3cf904/mod.ts",
     "tldts": "npm:tldts@^6.0.14",
     "tseep": "npm:tseep@^1.2.1",

deno.lock

@@ -7,11 +7,12 @@
       "jsr:@denosaurs/plug@1": "jsr:@denosaurs/plug@1.0.6",
       "jsr:@gleasonator/policy": "jsr:@gleasonator/policy@0.2.0",
       "jsr:@gleasonator/policy@0.2.0": "jsr:@gleasonator/policy@0.2.0",
+      "jsr:@gleasonator/policy@0.4.0": "jsr:@gleasonator/policy@0.4.0",
       "jsr:@hono/hono@^4.4.6": "jsr:@hono/hono@4.4.6",
       "jsr:@nostrify/nostrify@^0.22.1": "jsr:@nostrify/nostrify@0.22.5",
       "jsr:@nostrify/nostrify@^0.22.4": "jsr:@nostrify/nostrify@0.22.4",
       "jsr:@nostrify/nostrify@^0.22.5": "jsr:@nostrify/nostrify@0.22.5",
-      "jsr:@nostrify/nostrify@^0.23.3": "jsr:@nostrify/nostrify@0.23.3",
+      "jsr:@nostrify/nostrify@^0.25.0": "jsr:@nostrify/nostrify@0.25.0",
       "jsr:@soapbox/kysely-deno-sqlite@^2.1.0": "jsr:@soapbox/kysely-deno-sqlite@2.2.0",
       "jsr:@soapbox/stickynotes@^0.4.0": "jsr:@soapbox/stickynotes@0.4.0",
       "jsr:@std/assert@^0.217.0": "jsr:@std/assert@0.217.0",
@@ -61,6 +62,7 @@
       "npm:nostr-tools@^2.5.0": "npm:nostr-tools@2.5.1",
       "npm:nostr-tools@^2.7.0": "npm:nostr-tools@2.7.0",
       "npm:nostr-wasm@^0.1.0": "npm:nostr-wasm@0.1.0",
+      "npm:prom-client@^15.1.2": "npm:prom-client@15.1.2",
       "npm:tldts@^6.0.14": "npm:tldts@6.1.18",
       "npm:type-fest@^4.3.0": "npm:type-fest@4.18.2",
       "npm:unfurl.js@^6.4.0": "npm:unfurl.js@6.4.0",
@@ -96,6 +98,12 @@
           "jsr:@nostrify/nostrify@^0.22.1"
         ]
       },
+      "@gleasonator/policy@0.4.0": {
+        "integrity": "59c2f3ab1dc663e99a3e10b7eb69bf9fe581ce5d428fe56653e38f7f961da5ea",
+        "dependencies": [
+          "jsr:@nostrify/nostrify@^0.22.1"
+        ]
+      },
       "@hono/hono@4.4.6": {
         "integrity": "aa557ca9930787ee86b9ca1730691f1ce1c379174c2cb244d5934db2b6314453"
       },
@@ -128,8 +136,8 @@
           "npm:zod@^3.23.8"
         ]
       },
-      "@nostrify/nostrify@0.23.3": {
+      "@nostrify/nostrify@0.25.0": {
-        "integrity": "868b10dd094801e28f4982ef9815f0d43f2a807b6f8ad291c78ecb3eb291605a",
+        "integrity": "98f26f44e95ac87fc91b3f3809d38432e1a7f6aebf10380b2554b6f9526313c6",
         "dependencies": [
           "jsr:@std/encoding@^0.224.1",
           "npm:@scure/base@^1.1.6",
@@ -275,6 +283,10 @@
         "integrity": "sha512-V1JJ1WTRUqHHrOSh597hURcMqVKVGL/ea3kv0gSnEdsEZ0/+VyPghM1lMNGc00z7CIQorSvbKpuJkxvuHbvdbg==",
         "dependencies": {}
       },
+      "@opentelemetry/api@1.9.0": {
+        "integrity": "sha512-3giAOQvZiH5F9bMlMiv8+GSPMeqg0dbaeo58/0SlA9sxSqZhnUtxzX9/2FzyhS9sWQf5S0GJE0AKBrFqjpeYcg==",
+        "dependencies": {}
+      },
       "@scure/base@1.1.1": {
         "integrity": "sha512-ZxOhsSyxYwLJj3pLZCefNitxsj093tb2vq90mp2txoYeBqbcjDjqFhyM8eUjq/uFm6zJ+mUuqxlS2FkuSY1MTA==",
         "dependencies": {}
@@ -351,6 +363,10 @@
         "integrity": "sha512-Oei9OH4tRh0YqU3GxhX79dM/mwVgvbZJaSNaRk+bshkj0S5cfHcgYakreBjrHwatXKbz+IoIdYLxrKim2MjW0Q==",
         "dependencies": {}
       },
+      "bintrees@1.0.2": {
+        "integrity": "sha512-VOMgTMwjAaUG580SXn3LacVgjurrbMme7ZZNYGSSV7mmtY6QQRh0Eg3pwIcntQ77DErK1L0NxkbetjcoXzVwKw==",
+        "dependencies": {}
+      },
       "braces@3.0.2": {
         "integrity": "sha512-b8um+L1RzM3WDSzvhm6gIz1yfTbBt6YTlcEKAvsmqCZZFw46z626lVj9j1yEPW33H5H+lBQpZMP1k8l+78Ha0A==",
         "dependencies": {
@@ -852,6 +868,13 @@
         "integrity": "sha512-eG2dWTVw5bzqGRztnHExczNxt5VGsE6OwTeCG3fdUf9KBsZzO3R5OIIIzWR+iZA0NtZ+RDVdaoE2dK1cn6jH4g==",
         "dependencies": {}
       },
+      "prom-client@15.1.2": {
+        "integrity": "sha512-on3h1iXb04QFLLThrmVYg1SChBQ9N1c+nKAjebBjokBqipddH3uxmOUcEkTnzmJ8Jh/5TSUnUqS40i2QB2dJHQ==",
+        "dependencies": {
+          "@opentelemetry/api": "@opentelemetry/api@1.9.0",
+          "tdigest": "tdigest@0.1.2"
+        }
+      },
       "psl@1.9.0": {
         "integrity": "sha512-E/ZsdU4HLs/68gYzgGTkMicWTLPdAftJLfJFlLUAAKZGkStNU72sZjT66SnMDVOfOWY/YAoiD7Jxa9iHvngcag==",
         "dependencies": {}
@@ -955,6 +978,12 @@
         "integrity": "sha512-9QNk5KwDF+Bvz+PyObkmSYjI5ksVUYtjW7AU22r2NKcfLJcXp96hkDWU3+XndOsUb+AQ9QhfzfCT2O+CNWT5Tw==",
         "dependencies": {}
       },
+      "tdigest@0.1.2": {
+        "integrity": "sha512-+G0LLgjjo9BZX2MfdvPfH+MKLCrxlXSYec5DaPYP1fe6Iyhf0/fSmJ0bFiZ1F8BT6cGXl2LpltQptzjXKWEkKA==",
+        "dependencies": {
+          "bintrees": "bintrees@1.0.2"
+        }
+      },
       "tldts-core@6.1.18": {
         "integrity": "sha512-e4wx32F/7dMBSZyKAx825Yte3U0PQtZZ0bkWxYQiwLteRVnQ5zM40fEbi0IyNtwQssgJAk3GCr7Q+w39hX0VKA==",
         "dependencies": {}
@@ -1362,6 +1391,14 @@
     "https://esm.sh/v135/kysely@0.17.1/denonext/dist/esm/index-nodeless.js": "6f73bbf2d73bc7e96cdabf941c4ae8c12f58fd7b441031edec44c029aed9532b",
     "https://gitlab.com/soapbox-pub/deno-safe-fetch/-/raw/v1.0.0/load.ts": "3f74ab08cf97d4a3e6994cb79422e9b0069495e017416858121d5ff8ae04ac2a",
     "https://gitlab.com/soapbox-pub/deno-safe-fetch/-/raw/v1.0.0/mod.ts": "5f505cd265aefbcb687cde6f98c79344d3292ee1dd978e85e5ffa84a617c6682",
+    "https://gitlab.com/soapbox-pub/kysely-deno-postgres/-/raw/b4725e74ad6ca359ba0e370b55dbb8bb845a8a83/deps.ts": "b3dbecae69c30a5f161323b8c8ebd91d9af1eceb98fafab3091c7281a4b64fed",
+    "https://gitlab.com/soapbox-pub/kysely-deno-postgres/-/raw/b4725e74ad6ca359ba0e370b55dbb8bb845a8a83/mod.ts": "662438fd3909984bb8cbaf3fd44d2121e949d11301baf21d6c3f057ccf9887de",
+    "https://gitlab.com/soapbox-pub/kysely-deno-postgres/-/raw/b4725e74ad6ca359ba0e370b55dbb8bb845a8a83/src/PostgreSQLDriver.ts": "ea5a523bceeed420858b744beeb95d48976cb2b0d3f519a68b65a8229036cf6a",
+    "https://gitlab.com/soapbox-pub/kysely-deno-postgres/-/raw/b4725e74ad6ca359ba0e370b55dbb8bb845a8a83/src/PostgreSQLDriverDatabaseConnection.ts": "11e2fc10a3abb3d0729613c4b7cdb9cb73b597fd77353311bb6707c73a635fc5",
+    "https://gitlab.com/soapbox-pub/kysely-deno-postgres/-/raw/c6869b9e12d74af78a846ad503d84493f5db9df4/deps.ts": "b3dbecae69c30a5f161323b8c8ebd91d9af1eceb98fafab3091c7281a4b64fed",
+    "https://gitlab.com/soapbox-pub/kysely-deno-postgres/-/raw/c6869b9e12d74af78a846ad503d84493f5db9df4/mod.ts": "662438fd3909984bb8cbaf3fd44d2121e949d11301baf21d6c3f057ccf9887de",
+    "https://gitlab.com/soapbox-pub/kysely-deno-postgres/-/raw/c6869b9e12d74af78a846ad503d84493f5db9df4/src/PostgreSQLDriver.ts": "0f5d1bc2b24d4e0052e38ee289fb2f5e8e1470544f61aa2afe65e1059bf35dfb",
+    "https://gitlab.com/soapbox-pub/kysely-deno-postgres/-/raw/c6869b9e12d74af78a846ad503d84493f5db9df4/src/PostgreSQLDriverDatabaseConnection.ts": "e5d4e0fc9737c3ec253e679a51f5b43d2bb9a3386c147b7b1d14f4f5a5f734f1",
     "https://gitlab.com/soapbox-pub/kysely-deno-postgres/-/raw/f2948b86190a10faa293588775e162b3a8b52e70/deps.ts": "b3dbecae69c30a5f161323b8c8ebd91d9af1eceb98fafab3091c7281a4b64fed",
     "https://gitlab.com/soapbox-pub/kysely-deno-postgres/-/raw/f2948b86190a10faa293588775e162b3a8b52e70/mod.ts": "662438fd3909984bb8cbaf3fd44d2121e949d11301baf21d6c3f057ccf9887de",
     "https://gitlab.com/soapbox-pub/kysely-deno-postgres/-/raw/f2948b86190a10faa293588775e162b3a8b52e70/src/PostgreSQLDriver.ts": "ac1a39e86fd676973bce215e19db1f26b82408b8f2bb09a3601802974ea7cec6",
@@ -1387,7 +1424,7 @@
       "jsr:@bradenmacdonald/s3-lite-client@^0.7.4",
       "jsr:@db/sqlite@^0.11.1",
       "jsr:@hono/hono@^4.4.6",
-      "jsr:@nostrify/nostrify@^0.23.3",
+      "jsr:@nostrify/nostrify@^0.25.0",
       "jsr:@soapbox/kysely-deno-sqlite@^2.1.0",
       "jsr:@soapbox/stickynotes@^0.4.0",
       "jsr:@std/assert@^0.225.1",
@@ -1418,6 +1455,7 @@
       "npm:nostr-relaypool2@0.6.34",
       "npm:nostr-tools@2.5.1",
       "npm:nostr-wasm@^0.1.0",
+      "npm:prom-client@^15.1.2",
       "npm:tldts@^6.0.14",
       "npm:tseep@^1.2.1",
       "npm:type-fest@^4.3.0",

installation/ditto.conf

@@ -50,6 +50,12 @@ server {
     root /opt/ditto/public;
   }
 
+  location /metrics {
+    allow 127.0.0.1;
+    deny all;
+    proxy_pass http://ditto;
+  }
+
   location ~ ^/(instance|sw\.js$|sw\.js\.map$) {
     root /opt/ditto/public;
     try_files $uri =404;

scripts/nostr-pull.ts

@@ -0,0 +1,157 @@
+/**
+ * Script to import a user/list of users into Ditto given their npub/pubkey
+ * by looking them up on a list of relays.
+ */
+
+import { NostrEvent, NRelay1, NSchema } from '@nostrify/nostrify';
+import { nip19 } from 'nostr-tools';
+
+import { DittoDB } from '@/db/DittoDB.ts';
+import { EventsDB } from '@/storages/EventsDB.ts';
+
+const kysely = await DittoDB.getInstance();
+const eventsDB = new EventsDB(kysely);
+
+interface ImportEventsOpts {
+  profilesOnly: boolean;
+}
+
+type DoEvent = (event: NostrEvent) => void | Promise<void>;
+const importUsers = async (
+  authors: string[],
+  relays: string[],
+  opts?: Partial<ImportEventsOpts>,
+  doEvent: DoEvent = async (event: NostrEvent) => await eventsDB.event(event),
+) => {
+  // Kind 0s + follow lists.
+  const profiles: Record<string, Record<number, NostrEvent>> = {};
+  // Kind 1s.
+  const notes = new Set<string>();
+  const { profilesOnly = false } = opts || {};
+
+  await Promise.all(relays.map(async (relay) => {
+    if (!relay.startsWith('wss://')) console.error(`Invalid relay url ${relay}`);
+    const conn = new NRelay1(relay);
+    const matched = await conn.query([{ kinds: [0, 3], authors, limit: 1000 }]);
+
+    if (!profilesOnly) {
+      matched.push(
+        ...await conn.query(
+          authors.map((author) => ({ kinds: [1], authors: [author], limit: 200 })),
+        ),
+      );
+    }
+
+    await conn.close();
+    await Promise.all(
+      matched.map(async (event) => {
+        const { kind, pubkey } = event;
+        if (kind === 1 && !notes.has(event.id)) {
+          // add the event to eventsDB only if it has not been found already.
+          notes.add(event.id);
+          await doEvent(event);
+          return;
+        }
+
+        profiles[pubkey] ??= {};
+        const existing = profiles[pubkey][kind];
+        if (existing?.created_at > event.created_at) return;
+        else profiles[pubkey][kind] = event;
+      }),
+    );
+  }));
+
+  for (const user in profiles) {
+    const profile = profiles[user];
+    for (const kind in profile) {
+      await doEvent(profile[kind]);
+    }
+
+    let name = user;
+    // kind 0, not first idx
+    const event = profile[0];
+    if (event) {
+      // if event exists, print name
+      const parsed = JSON.parse(event.content);
+      name = parsed.nip05 || parsed.name || name;
+    }
+    if (NSchema.id().safeParse(name).success) {
+      // if no kind 0 found and this is a pubkey, encode as npub
+      name = nip19.npubEncode(name);
+    }
+    console.info(`Imported user ${name}${profilesOnly ? "'s profile" : ''}.`);
+  }
+};
+
+if (import.meta.main) {
+  if (!Deno.args.length) {
+    showHelp();
+    Deno.exit(1);
+  }
+  const pubkeys: string[] = [];
+  const relays: string[] = [];
+
+  const opts: Partial<ImportEventsOpts> = {};
+
+  let optionsEnd = false;
+  let relaySectionBegun = false;
+  for (const arg of Deno.args) {
+    if (arg.startsWith('-')) {
+      if (optionsEnd) {
+        console.error('Option encountered after end of options section.');
+        showUsage();
+        Deno.exit(1);
+      }
+      switch (arg) {
+        case '-p':
+        case '--profile-only':
+          console.info('Only importing profiles.');
+          opts.profilesOnly = true;
+          break;
+      }
+    } else if (arg.startsWith('npub1')) {
+      optionsEnd = true;
+
+      if (relaySectionBegun) {
+        console.error('npub specified in relay section');
+        Deno.exit(1);
+      }
+      const decoded = nip19.decode(arg as `npub1${string}`).data;
+      if (!NSchema.id().safeParse(decoded).success) {
+        console.error(`invalid pubkey ${arg}, skipping...`);
+        continue;
+      }
+      pubkeys.push(decoded);
+    } else if (NSchema.id().safeParse(arg).success) {
+      pubkeys.push(arg);
+    } else {
+      relaySectionBegun = true;
+      if (!arg.startsWith('wss://')) {
+        console.error(`invalid relay url ${arg}, skipping...`);
+      }
+      relays.push(arg);
+    }
+  }
+
+  await importUsers(pubkeys, relays, opts);
+  Deno.exit(0);
+}
+
+function showHelp() {
+  console.info('ditto - db:import');
+  console.info("Import users' posts and kind 0s from a given set of relays.\n");
+  showUsage();
+  console.info(`
+OPTIONS:
+
+-p, --profile-only
+  Only import profiles and not posts. Default: off.
+`);
+}
+
+function showUsage() {
+  console.info(
+    'Usage: deno task db:import [options] npub1xxxxxx[ npub1yyyyyyy]...' +
+      ' wss://first.relay[ second.relay]...',
+  );
+}

src/app.ts

@@ -108,11 +108,14 @@ import {
   trendingStatusesController,
   trendingTagsController,
 } from '@/controllers/api/trends.ts';
+import { errorHandler } from '@/controllers/error.ts';
+import { metricsController } from '@/controllers/metrics.ts';
 import { indexController } from '@/controllers/site.ts';
 import { nodeInfoController, nodeInfoSchemaController } from '@/controllers/well-known/nodeinfo.ts';
 import { nostrController } from '@/controllers/well-known/nostr.ts';
 import { auth98Middleware, requireProof, requireRole } from '@/middleware/auth98Middleware.ts';
 import { cspMiddleware } from '@/middleware/cspMiddleware.ts';
+import { metricsMiddleware } from '@/middleware/metricsMiddleware.ts';
 import { rateLimitMiddleware } from '@/middleware/rateLimitMiddleware.ts';
 import { requireSigner } from '@/middleware/requireSigner.ts';
 import { signerMiddleware } from '@/middleware/signerMiddleware.ts';
@@ -149,14 +152,14 @@ if (Conf.cronEnabled) {
 
 app.use('*', rateLimitMiddleware(300, Time.minutes(5)));
 
-app.use('/api/*', logger(debug));
+app.use('/api/*', metricsMiddleware, logger(debug));
-app.use('/.well-known/*', logger(debug));
+app.use('/.well-known/*', metricsMiddleware, logger(debug));
-app.use('/users/*', logger(debug));
+app.use('/users/*', metricsMiddleware, logger(debug));
-app.use('/nodeinfo/*', logger(debug));
+app.use('/nodeinfo/*', metricsMiddleware, logger(debug));
-app.use('/oauth/*', logger(debug));
+app.use('/oauth/*', metricsMiddleware, logger(debug));
 
-app.get('/api/v1/streaming', streamingController);
+app.get('/api/v1/streaming', metricsMiddleware, streamingController);
-app.get('/relay', relayController);
+app.get('/relay', metricsMiddleware, relayController);
 
 app.use(
   '*',
@@ -168,6 +171,8 @@ app.use(
   storeMiddleware,
 );
 
+app.get('/metrics', metricsController);
+
 app.get('/.well-known/nodeinfo', nodeInfoController);
 app.get('/.well-known/nostr.json', nostrController);
 
@@ -335,6 +340,8 @@ app.get('/', frontendController, indexController);
 // Fallback
 app.get('*', publicFiles, staticFiles, frontendController);
 
+app.onError(errorHandler);
+
 export default app;
 
 export type { AppContext, AppController, AppMiddleware };

src/config.ts

@@ -63,6 +63,14 @@ class Conf {
   static get localDomain(): string {
     return Deno.env.get('LOCAL_DOMAIN') || `http://localhost:${Conf.port}`;
   }
+  /** Link to an external nostr viewer. */
+  static get externalDomain(): string {
+    return Deno.env.get('NOSTR_EXTERNAL') || 'https://njump.me';
+  }
+  /** Get a link to a nip19-encoded entity in the configured external viewer. */
+  static external(path: string) {
+    return new URL(path, Conf.externalDomain).toString();
+  }
   /**
    * Heroku-style database URL. This is used in production to connect to the
    * database.

src/controllers/api/accounts.ts

@@ -209,7 +209,9 @@ const accountStatusesController: AppController = async (c) => {
     filter['#t'] = [tagged];
   }
 
-  const events = await store.query([filter], { signal })
+  const opts = { signal, limit, timeout: 10_000 };
+
+  const events = await store.query([filter], opts)
     .then((events) => hydrateEvents({ events, store, signal }))
     .then((events) => {
       if (exclude_replies) {

src/controllers/api/notifications.ts

@@ -78,7 +78,7 @@ async function renderNotifications(
   const store = c.get('store');
   const pubkey = await c.get('signer')?.getPublicKey()!;
   const { signal } = c.req.raw;
-  const opts = { signal, limit: params.limit };
+  const opts = { signal, limit: params.limit, timeout: 15_000 };
 
   const events = await store
     .query(filters, opts)

src/controllers/api/statuses.ts

@@ -8,14 +8,21 @@ import { z } from 'zod';
 import { type AppController } from '@/app.ts';
 import { Conf } from '@/config.ts';
 import { DittoDB } from '@/db/DittoDB.ts';
-import { getAmount } from '@/utils/bolt11.ts';
 import { getAncestors, getAuthor, getDescendants, getEvent } from '@/queries.ts';
 import { getUnattachedMediaByIds } from '@/db/unattached-media.ts';
 import { renderEventAccounts } from '@/views.ts';
 import { renderReblog, renderStatus } from '@/views/mastodon/statuses.ts';
 import { Storages } from '@/storages.ts';
 import { hydrateEvents, purifyEvent } from '@/storages/hydrate.ts';
-import { createEvent, paginated, paginationSchema, parseBody, updateListEvent } from '@/utils/api.ts';
+import {
+  createEvent,
+  listPaginationSchema,
+  paginated,
+  paginatedList,
+  paginationSchema,
+  parseBody,
+  updateListEvent,
+} from '@/utils/api.ts';
 import { getInvoice, getLnurl } from '@/utils/lnurl.ts';
 import { lookupPubkey } from '@/utils/lookup.ts';
 import { addTag, deleteTag } from '@/utils/tags.ts';
@@ -90,8 +97,8 @@ const createStatusController: AppController = async (c) => {
 
     const root = ancestor.tags.find((tag) => tag[0] === 'e' && tag[3] === 'root')?.[1] ?? ancestor.id;
 
-    tags.push(['e', root, 'root']);
+    tags.push(['e', root, Conf.relay, 'root']);
-    tags.push(['e', data.in_reply_to_id, 'reply']);
+    tags.push(['e', data.in_reply_to_id, Conf.relay, 'reply']);
   }
 
   if (data.quote_id) {
@@ -195,7 +202,7 @@ const deleteStatusController: AppController = async (c) => {
     if (event.pubkey === pubkey) {
       await createEvent({
         kind: 5,
-        tags: [['e', id]],
+        tags: [['e', id, Conf.relay]],
       }, c);
 
       const author = await getAuthor(event.pubkey);
@@ -253,8 +260,8 @@ const favouriteController: AppController = async (c) => {
       kind: 7,
       content: '+',
       tags: [
-        ['e', target.id],
+        ['e', target.id, Conf.relay],
-        ['p', target.pubkey],
+        ['p', target.pubkey, Conf.relay],
       ],
     }, c);
 
@@ -295,7 +302,10 @@ const reblogStatusController: AppController = async (c) => {
 
   const reblogEvent = await createEvent({
     kind: 6,
-    tags: [['e', event.id], ['p', event.pubkey]],
+    tags: [
+      ['e', event.id, Conf.relay],
+      ['p', event.pubkey, Conf.relay],
+    ],
   }, c);
 
   await hydrateEvents({
@@ -330,7 +340,7 @@ const unreblogStatusController: AppController = async (c) => {
 
   await createEvent({
     kind: 5,
-    tags: [['e', repostEvent.id]],
+    tags: [['e', repostEvent.id, Conf.relay]],
   }, c);
 
   return c.json(await renderStatus(event, { viewerPubkey: pubkey }));
@@ -382,7 +392,7 @@ const bookmarkController: AppController = async (c) => {
   if (event) {
     await updateListEvent(
       { kinds: [10003], authors: [pubkey], limit: 1 },
-      (tags) => addTag(tags, ['e', eventId]),
+      (tags) => addTag(tags, ['e', eventId, Conf.relay]),
       c,
     );
 
@@ -409,7 +419,7 @@ const unbookmarkController: AppController = async (c) => {
   if (event) {
     await updateListEvent(
       { kinds: [10003], authors: [pubkey], limit: 1 },
-      (tags) => deleteTag(tags, ['e', eventId]),
+      (tags) => deleteTag(tags, ['e', eventId, Conf.relay]),
       c,
     );
 
@@ -436,7 +446,7 @@ const pinController: AppController = async (c) => {
   if (event) {
     await updateListEvent(
       { kinds: [10001], authors: [pubkey], limit: 1 },
-      (tags) => addTag(tags, ['e', eventId]),
+      (tags) => addTag(tags, ['e', eventId, Conf.relay]),
       c,
     );
 
@@ -465,7 +475,7 @@ const unpinController: AppController = async (c) => {
   if (event) {
     await updateListEvent(
       { kinds: [10001], authors: [pubkey], limit: 1 },
-      (tags) => deleteTag(tags, ['e', eventId]),
+      (tags) => deleteTag(tags, ['e', eventId, Conf.relay]),
       c,
     );
 
@@ -509,7 +519,7 @@ const zapController: AppController = async (c) => {
     lnurl = getLnurl(meta);
     if (target && lnurl) {
       tags.push(
-        ['e', target.id],
+        ['e', target.id, Conf.relay],
         ['p', target.pubkey],
         ['amount', amount.toString()],
         ['relays', Conf.relay],
@@ -545,33 +555,26 @@ const zapController: AppController = async (c) => {
 
 const zappedByController: AppController = async (c) => {
   const id = c.req.param('id');
+  const params = listPaginationSchema.parse(c.req.query());
   const store = await Storages.db();
-  const amountSchema = z.coerce.number().int().nonnegative().catch(0);
+  const db = await DittoDB.getInstance();
 
-  const events = (await store.query([{ kinds: [9735], '#e': [id], limit: 100 }])).map((event) => {
+  const zaps = await db.selectFrom('event_zaps')
-    const zapRequestString = event.tags.find(([name]) => name === 'description')?.[1];
+    .selectAll()
-    if (!zapRequestString) return;
+    .where('target_event_id', '=', id)
-    try {
+    .orderBy('amount_millisats', 'desc')
-      const zapRequest = n.json().pipe(n.event()).parse(zapRequestString);
+    .limit(params.limit)
-      const amount = zapRequest?.tags.find(([name]: any) => name === 'amount')?.[1];
+    .offset(params.offset).execute();
-      if (!amount) {
-        const amount = getAmount(event?.tags.find(([name]) => name === 'bolt11')?.[1]);
-        if (!amount) return;
-        zapRequest.tags.push(['amount', amount]);
-      }
-      return zapRequest;
-    } catch {
-      return;
-    }
-  }).filter(Boolean) as DittoEvent[];
 
-  await hydrateEvents({ events, store });
+  const authors = await store.query([{ kinds: [0], authors: zaps.map((zap) => zap.sender_pubkey) }]);
 
   const results = (await Promise.all(
-    events.map(async (event) => {
+    zaps.map(async (zap) => {
-      const amount = amountSchema.parse(event.tags.find(([name]) => name === 'amount')?.[1]);
+      const amount = zap.amount_millisats;
-      const comment = event?.content ?? '';
+      const comment = zap.comment;
-      const account = event?.author ? await renderAccount(event.author) : await accountFromPubkey(event.pubkey);
+
+      const sender = authors.find((author) => author.pubkey === zap.sender_pubkey);
+      const account = sender ? await renderAccount(sender) : await accountFromPubkey(zap.sender_pubkey);
 
       return {
         comment,
@@ -581,7 +584,7 @@ const zappedByController: AppController = async (c) => {
     }),
   )).filter(Boolean);
 
-  return c.json(results);
+  return paginatedList(c, params, results);
 };
 
 export {

src/controllers/api/streaming.ts

@@ -5,6 +5,7 @@ import { z } from 'zod';
 import { type AppController } from '@/app.ts';
 import { Conf } from '@/config.ts';
 import { DittoDB } from '@/db/DittoDB.ts';
+import { streamingConnectionsGauge } from '@/metrics.ts';
 import { MuteListPolicy } from '@/policies/MuteListPolicy.ts';
 import { getFeedPubkeys } from '@/queries.ts';
 import { hydrateEvents } from '@/storages/hydrate.ts';
@@ -111,6 +112,8 @@ const streamingController: AppController = async (c) => {
   }
 
   socket.onopen = async () => {
+    streamingConnectionsGauge.inc();
+
     if (!stream) return;
     const topicFilter = await topicToFilter(stream, c.req.query(), pubkey);
 
@@ -151,6 +154,7 @@ const streamingController: AppController = async (c) => {
   };
 
   socket.onclose = () => {
+    streamingConnectionsGauge.dec();
     controller.abort();
   };
 

src/controllers/api/timelines.ts

@@ -60,9 +60,10 @@ const suggestedTimelineController: AppController = async (c) => {
 async function renderStatuses(c: AppContext, filters: NostrFilter[]) {
   const { signal } = c.req.raw;
   const store = c.get('store');
+  const opts = { signal, timeout: 10_000 };
 
   const events = await store
-    .query(filters, { signal })
+    .query(filters, opts)
     .then((events) => hydrateEvents({ events, store, signal }));
 
   if (!events.length) {

src/controllers/error.ts

@@ -0,0 +1,16 @@
+import { ErrorHandler } from '@hono/hono';
+import { HTTPException } from '@hono/hono/http-exception';
+
+export const errorHandler: ErrorHandler = (err, c) => {
+  if (err instanceof HTTPException) {
+    return c.json({ error: err.message }, err.status);
+  }
+
+  console.error(err);
+
+  if (err.message === 'canceling statement due to statement timeout') {
+    return c.json({ error: 'The server was unable to respond in a timely manner' }, 500);
+  }
+
+  return c.json({ error: 'Something went wrong' }, 500);
+};

src/controllers/metrics.ts

@@ -0,0 +1,20 @@
+import { register } from 'prom-client';
+
+import { AppController } from '@/app.ts';
+import { DittoDB } from '@/db/DittoDB.ts';
+import { dbAvailableConnectionsGauge, dbPoolSizeGauge } from '@/metrics.ts';
+
+/** Prometheus/OpenMetrics controller. */
+export const metricsController: AppController = async (c) => {
+  // Update some metrics at request time.
+  dbPoolSizeGauge.set(DittoDB.poolSize);
+  dbAvailableConnectionsGauge.set(DittoDB.availableConnections);
+
+  const metrics = await register.metrics();
+
+  const headers: HeadersInit = {
+    'Content-Type': register.contentType,
+  };
+
+  return c.text(metrics, 200, headers);
+};

src/controllers/nostr/relay.ts

@@ -10,6 +10,7 @@ import {
 
 import { AppController } from '@/app.ts';
 import { relayInfoController } from '@/controllers/nostr/relay-info.ts';
+import { relayConnectionsGauge, relayEventCounter, relayMessageCounter } from '@/metrics.ts';
 import * as pipeline from '@/pipeline.ts';
 import { RelayError } from '@/RelayError.ts';
 import { Storages } from '@/storages.ts';
@@ -21,16 +22,24 @@ const FILTER_LIMIT = 100;
 function connectStream(socket: WebSocket) {
   const controllers = new Map<string, AbortController>();
 
+  socket.onopen = () => {
+    relayConnectionsGauge.inc();
+  };
+
   socket.onmessage = (e) => {
     const result = n.json().pipe(n.clientMsg()).safeParse(e.data);
     if (result.success) {
+      relayMessageCounter.inc({ verb: result.data[0] });
       handleMsg(result.data);
     } else {
+      relayMessageCounter.inc();
       send(['NOTICE', 'Invalid message.']);
     }
   };
 
   socket.onclose = () => {
+    relayConnectionsGauge.dec();
+
     for (const controller of controllers.values()) {
       controller.abort();
     }
@@ -64,11 +73,15 @@ function connectStream(socket: WebSocket) {
     const pubsub = await Storages.pubsub();
 
     try {
-      for (const event of await store.query(filters, { limit: FILTER_LIMIT })) {
+      for (const event of await store.query(filters, { limit: FILTER_LIMIT, timeout: 1000 })) {
         send(['EVENT', subId, event]);
       }
     } catch (e) {
-      send(['CLOSED', subId, e.message]);
+      if (e instanceof RelayError) {
+        send(['CLOSED', subId, e.message]);
+      } else {
+        send(['CLOSED', subId, 'error: something went wrong']);
+      }
       controllers.delete(subId);
       return;
     }
@@ -88,6 +101,7 @@ function connectStream(socket: WebSocket) {
 
   /** Handle EVENT. Store the event. */
   async function handleEvent([_, event]: NostrClientEVENT): Promise<void> {
+    relayEventCounter.inc({ kind: event.kind.toString() });
     try {
       // This will store it (if eligible) and run other side-effects.
       await pipeline.handleEvent(event, AbortSignal.timeout(1000));
@@ -114,7 +128,7 @@ function connectStream(socket: WebSocket) {
   /** Handle COUNT. Return the number of events matching the filters. */
   async function handleCount([_, subId, ...filters]: NostrClientCOUNT): Promise<void> {
     const store = await Storages.db();
-    const { count } = await store.count(filters);
+    const { count } = await store.count(filters, { timeout: 100 });
     send(['COUNT', subId, { count, approximate: false }]);
   }
 

src/db/DittoDB.ts

@@ -37,6 +37,20 @@ export class DittoDB {
     return kysely;
   }
 
+  static get poolSize(): number {
+    if (Conf.db.dialect === 'postgres') {
+      return DittoPostgres.getPool().size;
+    }
+    return 1;
+  }
+
+  static get availableConnections(): number {
+    if (Conf.db.dialect === 'postgres') {
+      return DittoPostgres.getPool().available;
+    }
+    return 1;
+  }
+
   /** Migrate the database to the latest version. */
   static async migrate(kysely: Kysely<DittoTables>) {
     const migrator = new Migrator({

src/db/DittoTables.ts

@@ -7,6 +7,7 @@ export interface DittoTables {
   author_stats: AuthorStatsRow;
   event_stats: EventStatsRow;
   pubkey_domains: PubkeyDomainRow;
+  event_zaps: EventZapRow;
 }
 
 interface AuthorStatsRow {
@@ -69,3 +70,11 @@ interface PubkeyDomainRow {
   domain: string;
   last_updated_at: number;
 }
+
+interface EventZapRow {
+  receipt_id: string;
+  target_event_id: string;
+  sender_pubkey: string;
+  amount_millisats: number;
+  comment: string;
+}

src/db/KyselyLogger.ts

@@ -1,5 +1,6 @@
 import { Stickynotes } from '@soapbox/stickynotes';
 import { Logger } from 'kysely';
+import { dbQueryTimeHistogram } from '@/metrics.ts';
 
 /** Log the SQL for queries. */
 export const KyselyLogger: Logger = (event) => {
@@ -9,6 +10,8 @@ export const KyselyLogger: Logger = (event) => {
     const { query, queryDurationMillis } = event;
     const { sql, parameters } = query;
 
+    dbQueryTimeHistogram.observe(queryDurationMillis);
+
     console.debug(
       sql,
       JSON.stringify(parameters),

src/db/adapters/DittoPostgres.ts

@@ -1,5 +1,6 @@
 import { Kysely, PostgresAdapter, PostgresIntrospector, PostgresQueryCompiler } from 'kysely';
 import { PostgreSQLDriver } from 'kysely_deno_postgres';
+import { Pool } from 'postgres';
 
 import { Conf } from '@/config.ts';
 import { DittoTables } from '@/db/DittoTables.ts';
@@ -7,6 +8,14 @@ import { KyselyLogger } from '@/db/KyselyLogger.ts';
 
 export class DittoPostgres {
   static db: Kysely<DittoTables> | undefined;
+  static pool: Pool | undefined;
+
+  static getPool(): Pool {
+    if (!this.pool) {
+      this.pool = new Pool(Conf.databaseUrl, Conf.pg.poolSize, true);
+    }
+    return this.pool;
+  }
 
   // deno-lint-ignore require-await
   static async getInstance(): Promise<Kysely<DittoTables>> {
@@ -17,10 +26,7 @@ export class DittoPostgres {
             return new PostgresAdapter();
           },
           createDriver() {
-            return new PostgreSQLDriver(
+            return new PostgreSQLDriver(DittoPostgres.getPool());
-              { connectionString: Conf.databaseUrl },
-              Conf.pg.poolSize,
-            );
           },
           createIntrospector(db: Kysely<unknown>) {
             return new PostgresIntrospector(db);

src/db/migrations/027_add_zap_events.ts

@@ -0,0 +1,32 @@
+import { Kysely } from 'kysely';
+
+export async function up(db: Kysely<any>): Promise<void> {
+  await db.schema
+    .createTable('event_zaps')
+    .addColumn('receipt_id', 'text', (col) => col.primaryKey())
+    .addColumn('target_event_id', 'text', (col) => col.notNull())
+    .addColumn('sender_pubkey', 'text', (col) => col.notNull())
+    .addColumn('amount_millisats', 'integer', (col) => col.notNull())
+    .addColumn('comment', 'text', (col) => col.notNull())
+    .execute();
+
+  await db.schema
+    .createIndex('idx_event_zaps_amount_millisats')
+    .on('event_zaps')
+    .column('amount_millisats')
+    .ifNotExists()
+    .execute();
+
+  await db.schema
+    .createIndex('idx_event_zaps_target_event_id')
+    .on('event_zaps')
+    .column('target_event_id')
+    .ifNotExists()
+    .execute();
+}
+
+export async function down(db: Kysely<any>): Promise<void> {
+  await db.schema.dropIndex('idx_event_zaps_amount_millisats').ifExists().execute();
+  await db.schema.dropIndex('idx_event_zaps_target_event_id').ifExists().execute();
+  await db.schema.dropTable('event_zaps').execute();
+}

src/entities/MastodonAccount.ts

@@ -40,6 +40,7 @@ export interface MastodonAccount {
   username: string;
   ditto: {
     accepts_zaps: boolean;
+    external_url: string;
   };
   pleroma: {
     deactivated: boolean;

src/entities/MastodonStatus.ts

@@ -39,4 +39,7 @@ export interface MastodonStatus {
     expires_at?: string;
     quotes_count: number;
   };
+  ditto: {
+    external_url: string;
+  };
 }

src/firehose.ts

@@ -1,9 +1,10 @@
 import { Stickynotes } from '@soapbox/stickynotes';
 
+import { firehoseEventCounter } from '@/metrics.ts';
 import { Storages } from '@/storages.ts';
 import { nostrNow } from '@/utils.ts';
 
-import * as pipeline from './pipeline.ts';
+import * as pipeline from '@/pipeline.ts';
 
 const console = new Stickynotes('ditto:firehose');
 
@@ -12,13 +13,14 @@ const console = new Stickynotes('ditto:firehose');
  * side-effects based on them, such as trending hashtag tracking
  * and storing events for notifications and the home feed.
  */
-export async function startFirehose() {
+export async function startFirehose(): Promise<void> {
   const store = await Storages.client();
 
   for await (const msg of store.req([{ kinds: [0, 1, 3, 5, 6, 7, 9735, 10002], limit: 0, since: nostrNow() }])) {
     if (msg[0] === 'EVENT') {
       const event = msg[2];
       console.debug(`NostrEvent<${event.kind}> ${event.id}`);
+      firehoseEventCounter.inc({ kind: event.kind });
 
       pipeline
         .handleEvent(event, AbortSignal.timeout(5000))

src/metrics.ts

@@ -0,0 +1,80 @@
+import { Counter, Gauge, Histogram } from 'prom-client';
+
+export const httpRequestCounter = new Counter({
+  name: 'http_requests_total',
+  help: 'Total number of HTTP requests',
+  labelNames: ['method'],
+});
+
+export const httpResponseCounter = new Counter({
+  name: 'http_responses_total',
+  help: 'Total number of HTTP responses',
+  labelNames: ['status', 'path'],
+});
+
+export const streamingConnectionsGauge = new Gauge({
+  name: 'streaming_connections',
+  help: 'Number of active connections to the streaming API',
+});
+
+export const fetchCounter = new Counter({
+  name: 'fetch_total',
+  help: 'Total number of fetch requests',
+  labelNames: ['method'],
+});
+
+export const firehoseEventCounter = new Counter({
+  name: 'firehose_events_total',
+  help: 'Total number of Nostr events processed by the firehose',
+  labelNames: ['kind'],
+});
+
+export const pipelineEventCounter = new Counter({
+  name: 'pipeline_events_total',
+  help: 'Total number of Nostr events processed by the pipeline',
+  labelNames: ['kind'],
+});
+
+export const relayEventCounter = new Counter({
+  name: 'relay_events_total',
+  help: 'Total number of EVENT messages processed by the relay',
+  labelNames: ['kind'],
+});
+
+export const relayMessageCounter = new Counter({
+  name: 'relay_messages_total',
+  help: 'Total number of Nostr messages processed by the relay',
+  labelNames: ['verb'],
+});
+
+export const relayConnectionsGauge = new Gauge({
+  name: 'relay_connections',
+  help: 'Number of active connections to the relay',
+});
+
+export const dbQueryCounter = new Counter({
+  name: 'db_query_total',
+  help: 'Total number of database queries',
+  labelNames: ['kind'],
+});
+
+export const dbEventCounter = new Counter({
+  name: 'db_events_total',
+  help: 'Total number of database inserts',
+  labelNames: ['kind'],
+});
+
+export const dbPoolSizeGauge = new Gauge({
+  name: 'db_pool_size',
+  help: 'Number of connections in the database pool',
+});
+
+export const dbAvailableConnectionsGauge = new Gauge({
+  name: 'db_available_connections',
+  help: 'Number of available connections in the database pool',
+});
+
+export const dbQueryTimeHistogram = new Histogram({
+  name: 'db_query_duration_ms',
+  help: 'Duration of database queries',
+});

src/middleware/metricsMiddleware.ts

@@ -0,0 +1,14 @@
+import { MiddlewareHandler } from '@hono/hono';
+
+import { httpRequestCounter, httpResponseCounter } from '@/metrics.ts';
+
+export const metricsMiddleware: MiddlewareHandler = async (c, next) => {
+  const { method } = c.req;
+  httpRequestCounter.inc({ method });
+
+  await next();
+
+  const { status } = c.res;
+  const path = c.req.matchedRoutes.find((r) => r.method !== 'ALL')?.path ?? c.req.routePath;
+  httpResponseCounter.inc({ status, path });
+};

src/pipeline.test.ts

@@ -0,0 +1,125 @@
+import { assertEquals } from '@std/assert';
+import { generateSecretKey } from 'nostr-tools';
+
+import { genEvent, getTestDB } from '@/test.ts';
+import { handleZaps } from '@/pipeline.ts';
+
+Deno.test('store one zap receipt in nostr_events; convert it into event_zaps table format and store it', async () => {
+  await using db = await getTestDB();
+  const kysely = db.kysely;
+
+  const sk = generateSecretKey();
+
+  const event = genEvent({
+    'id': '67b48a14fb66c60c8f9070bdeb37afdfcc3d08ad01989460448e4081eddda446',
+    'pubkey': '9630f464cca6a5147aa8a35f0bcdd3ce485324e732fd39e09233b1d848238f31',
+    'created_at': 1674164545,
+    'kind': 9735,
+    'tags': [
+      ['p', '32e1827635450ebb3c5a7d12c1f8e7b2b514439ac10a67eef3d9fd9c5c68e245'],
+      ['P', '97c70a44366a6535c145b333f973ea86dfdc2d7a99da618c40c64705ad98e322'],
+      ['e', '3624762a1274dd9636e0c552b53086d70bc88c165bc4dc0f9e836a1eaf86c3b8'],
+      [
+        'bolt11',
+        'lnbc10u1p3unwfusp5t9r3yymhpfqculx78u027lxspgxcr2n2987mx2j55nnfs95nxnzqpp5jmrh92pfld78spqs78v9euf2385t83uvpwk9ldrlvf6ch7tpascqhp5zvkrmemgth3tufcvflmzjzfvjt023nazlhljz2n9hattj4f8jq8qxqyjw5qcqpjrzjqtc4fc44feggv7065fqe5m4ytjarg3repr5j9el35xhmtfexc42yczarjuqqfzqqqqqqqqlgqqqqqqgq9q9qxpqysgq079nkq507a5tw7xgttmj4u990j7wfggtrasah5gd4ywfr2pjcn29383tphp4t48gquelz9z78p4cq7ml3nrrphw5w6eckhjwmhezhnqpy6gyf0',
+      ],
+      [
+        'description',
+        '{"pubkey":"97c70a44366a6535c145b333f973ea86dfdc2d7a99da618c40c64705ad98e322","content":"","id":"d9cc14d50fcb8c27539aacf776882942c1a11ea4472f8cdec1dea82fab66279d","created_at":1674164539,"sig":"77127f636577e9029276be060332ea565deaf89ff215a494ccff16ae3f757065e2bc59b2e8c113dd407917a010b3abd36c8d7ad84c0e3ab7dab3a0b0caa9835d","kind":9734,"tags":[["e","3624762a1274dd9636e0c552b53086d70bc88c165bc4dc0f9e836a1eaf86c3b8"],["p","32e1827635450ebb3c5a7d12c1f8e7b2b514439ac10a67eef3d9fd9c5c68e245"],["relays","wss://relay.damus.io","wss://nostr-relay.wlvs.space","wss://nostr.fmt.wiz.biz","wss://relay.nostr.bg","wss://nostr.oxtr.dev","wss://nostr.v0l.io","wss://brb.io","wss://nostr.bitcoiner.social","ws://monad.jb55.com:8080","wss://relay.snort.social"]]}',
+      ],
+      ['preimage', '5d006d2cf1e73c7148e7519a4c68adc81642ce0e25a432b2434c99f97344c15f'],
+    ],
+    'content': '',
+  }, sk);
+
+  await db.store.event(event);
+
+  await handleZaps(kysely, event);
+  await handleZaps(kysely, event);
+
+  const zapReceipts = await kysely.selectFrom('nostr_events').selectAll().execute();
+  const customEventZaps = await kysely.selectFrom('event_zaps').selectAll().execute();
+
+  assertEquals(zapReceipts.length, 1); // basic check
+  assertEquals(customEventZaps.length, 1); // basic check
+
+  const expected = {
+    receipt_id: event.id,
+    target_event_id: '3624762a1274dd9636e0c552b53086d70bc88c165bc4dc0f9e836a1eaf86c3b8',
+    sender_pubkey: '97c70a44366a6535c145b333f973ea86dfdc2d7a99da618c40c64705ad98e322',
+    amount_millisats: 1000000,
+    comment: '',
+  };
+
+  assertEquals(customEventZaps[0], expected);
+});
+
+// The function tests below only handle the edge cases and don't assert anything
+// If no error happens = ok
+
+Deno.test('zap receipt does not have a "description" tag', async () => {
+  await using db = await getTestDB();
+  const kysely = db.kysely;
+
+  const sk = generateSecretKey();
+
+  const event = genEvent({ kind: 9735 }, sk);
+
+  await handleZaps(kysely, event);
+
+  // no error happened = ok
+});
+
+Deno.test('zap receipt does not have a zap request stringified value in the "description" tag', async () => {
+  await using db = await getTestDB();
+  const kysely = db.kysely;
+
+  const sk = generateSecretKey();
+
+  const event = genEvent({ kind: 9735, tags: [['description', 'yolo']] }, sk);
+
+  await handleZaps(kysely, event);
+
+  // no error happened = ok
+});
+
+Deno.test('zap receipt does not have a "bolt11" tag', async () => {
+  await using db = await getTestDB();
+  const kysely = db.kysely;
+
+  const sk = generateSecretKey();
+
+  const event = genEvent({
+    kind: 9735,
+    tags: [[
+      'description',
+      '{"pubkey":"97c70a44366a6535c145b333f973ea86dfdc2d7a99da618c40c64705ad98e322","content":"","id":"d9cc14d50fcb8c27539aacf776882942c1a11ea4472f8cdec1dea82fab66279d","created_at":1674164539,"sig":"77127f636577e9029276be060332ea565deaf89ff215a494ccff16ae3f757065e2bc59b2e8c113dd407917a010b3abd36c8d7ad84c0e3ab7dab3a0b0caa9835d","kind":9734,"tags":[["e","3624762a1274dd9636e0c552b53086d70bc88c165bc4dc0f9e836a1eaf86c3b8"],["p","32e1827635450ebb3c5a7d12c1f8e7b2b514439ac10a67eef3d9fd9c5c68e245"],["relays","wss://relay.damus.io","wss://nostr-relay.wlvs.space","wss://nostr.fmt.wiz.biz","wss://relay.nostr.bg","wss://nostr.oxtr.dev","wss://nostr.v0l.io","wss://brb.io","wss://nostr.bitcoiner.social","ws://monad.jb55.com:8080","wss://relay.snort.social"]]}',
+    ]],
+  }, sk);
+
+  await handleZaps(kysely, event);
+
+  // no error happened = ok
+});
+
+Deno.test('zap request inside zap receipt does not have an "e" tag', async () => {
+  await using db = await getTestDB();
+  const kysely = db.kysely;
+
+  const sk = generateSecretKey();
+
+  const event = genEvent({
+    kind: 9735,
+    tags: [[
+      'bolt11',
+      'lnbc10u1p3unwfusp5t9r3yymhpfqculx78u027lxspgxcr2n2987mx2j55nnfs95nxnzqpp5jmrh92pfld78spqs78v9euf2385t83uvpwk9ldrlvf6ch7tpascqhp5zvkrmemgth3tufcvflmzjzfvjt023nazlhljz2n9hattj4f8jq8qxqyjw5qcqpjrzjqtc4fc44feggv7065fqe5m4ytjarg3repr5j9el35xhmtfexc42yczarjuqqfzqqqqqqqqlgqqqqqqgq9q9qxpqysgq079nkq507a5tw7xgttmj4u990j7wfggtrasah5gd4ywfr2pjcn29383tphp4t48gquelz9z78p4cq7ml3nrrphw5w6eckhjwmhezhnqpy6gyf0',
+    ], [
+      'description',
+      '{"pubkey":"97c70a44366a6535c145b333f973ea86dfdc2d7a99da618c40c64705ad98e322","content":"","id":"d9cc14d50fcb8c27539aacf776882942c1a11ea4472f8cdec1dea82fab66279d","created_at":1674164539,"sig":"77127f636577e9029276be060332ea565deaf89ff215a494ccff16ae3f757065e2bc59b2e8c113dd407917a010b3abd36c8d7ad84c0e3ab7dab3a0b0caa9835d","kind":9734,"tags":[["p","32e1827635450ebb3c5a7d12c1f8e7b2b514439ac10a67eef3d9fd9c5c68e245"],["relays","wss://relay.damus.io","wss://nostr-relay.wlvs.space","wss://nostr.fmt.wiz.biz","wss://relay.nostr.bg","wss://nostr.oxtr.dev","wss://nostr.v0l.io","wss://brb.io","wss://nostr.bitcoiner.social","ws://monad.jb55.com:8080","wss://relay.snort.social"]]}',
+    ]],
+  }, sk);
+
+  await handleZaps(kysely, event);
+
+  // no error happened = ok
+});

src/pipeline.ts

@@ -1,12 +1,14 @@
 import { NKinds, NostrEvent, NSchema as n } from '@nostrify/nostrify';
 import Debug from '@soapbox/stickynotes/debug';
-import { sql } from 'kysely';
+import { Kysely, sql } from 'kysely';
 import { LRUCache } from 'lru-cache';
+import { z } from 'zod';
 
 import { Conf } from '@/config.ts';
 import { DittoDB } from '@/db/DittoDB.ts';
 import { deleteAttachedMedia } from '@/db/unattached-media.ts';
 import { DittoEvent } from '@/interfaces/DittoEvent.ts';
+import { pipelineEventCounter } from '@/metrics.ts';
 import { RelayError } from '@/RelayError.ts';
 import { AdminSigner } from '@/signers/AdminSigner.ts';
 import { hydrateEvents } from '@/storages/hydrate.ts';
@@ -17,6 +19,8 @@ import { verifyEventWorker } from '@/workers/verify.ts';
 import { nip05Cache } from '@/utils/nip05.ts';
 import { updateStats } from '@/utils/stats.ts';
 import { getTagSet } from '@/utils/tags.ts';
+import { DittoTables } from '@/db/DittoTables.ts';
+import { getAmount } from '@/utils/bolt11.ts';
 
 const debug = Debug('ditto:pipeline');
 
@@ -36,6 +40,7 @@ async function handleEvent(event: DittoEvent, signal: AbortSignal): Promise<void
   if (encounterEvent(event)) return;
   if (await existsInDB(event)) return;
   debug(`NostrEvent<${event.kind}> ${event.id}`);
+  pipelineEventCounter.inc({ kind: event.kind });
 
   if (event.kind !== 24133) {
     await policyFilter(event);
@@ -49,8 +54,11 @@ async function handleEvent(event: DittoEvent, signal: AbortSignal): Promise<void
     throw new RelayError('blocked', 'user is disabled');
   }
 
+  const kysely = await DittoDB.getInstance();
+
   await Promise.all([
     storeEvent(event, signal),
+    handleZaps(kysely, event),
     parseMetadata(event, signal),
     generateSetEvents(event),
     processMedia(event),
@@ -108,7 +116,7 @@ async function hydrateEvent(event: DittoEvent, signal: AbortSignal): Promise<voi
 }
 
 /** Maybe store the event, if eligible. */
-async function storeEvent(event: DittoEvent, signal?: AbortSignal): Promise<void> {
+async function storeEvent(event: DittoEvent, signal?: AbortSignal): Promise<undefined> {
   if (NKinds.ephemeral(event.kind)) return;
   const store = await Storages.db();
   const kysely = await DittoDB.getInstance();
@@ -218,4 +226,33 @@ async function generateSetEvents(event: NostrEvent): Promise<void> {
   }
 }
 
-export { handleEvent };
+/** Stores the event in the 'event_zaps' table */
+async function handleZaps(kysely: Kysely<DittoTables>, event: NostrEvent) {
+  if (event.kind !== 9735) return;
+
+  const zapRequestString = event?.tags?.find(([name]) => name === 'description')?.[1];
+  if (!zapRequestString) return;
+  const zapRequest = n.json().pipe(n.event()).optional().catch(undefined).parse(zapRequestString);
+  if (!zapRequest) return;
+
+  const amountSchema = z.coerce.number().int().nonnegative().catch(0);
+  const amount_millisats = amountSchema.parse(getAmount(event?.tags.find(([name]) => name === 'bolt11')?.[1]));
+  if (!amount_millisats || amount_millisats < 1) return;
+
+  const zappedEventId = zapRequest.tags.find(([name]) => name === 'e')?.[1];
+  if (!zappedEventId) return;
+
+  try {
+    await kysely.insertInto('event_zaps').values({
+      receipt_id: event.id,
+      target_event_id: zappedEventId,
+      sender_pubkey: zapRequest.pubkey,
+      amount_millisats,
+      comment: zapRequest.content,
+    }).execute();
+  } catch {
+    // receipt_id is unique, do nothing
+  }
+}
+
+export { handleEvent, handleZaps };

src/signers/ConnectSigner.ts

@@ -1,4 +1,5 @@
 // deno-lint-ignore-file require-await
+import { HTTPException } from '@hono/hono/http-exception';
 import { NConnectSigner, NostrEvent, NostrSigner } from '@nostrify/nostrify';
 
 import { Storages } from '@/storages.ts';
@@ -27,30 +28,78 @@ export class ConnectSigner implements NostrSigner {
 
   async signEvent(event: Omit<NostrEvent, 'id' | 'pubkey' | 'sig'>): Promise<NostrEvent> {
     const signer = await this.signer;
-    return signer.signEvent(event);
+    try {
+      return await signer.signEvent(event);
+    } catch (e) {
+      if (e.name === 'AbortError') {
+        throw new HTTPException(408, { message: 'The event was not signed quickly enough' });
+      } else {
+        throw e;
+      }
+    }
   }
 
   readonly nip04 = {
     encrypt: async (pubkey: string, plaintext: string): Promise<string> => {
       const signer = await this.signer;
-      return signer.nip04.encrypt(pubkey, plaintext);
+      try {
+        return await signer.nip04.encrypt(pubkey, plaintext);
+      } catch (e) {
+        if (e.name === 'AbortError') {
+          throw new HTTPException(408, {
+            message: 'Text was not encrypted quickly enough',
+          });
+        } else {
+          throw e;
+        }
+      }
     },
 
     decrypt: async (pubkey: string, ciphertext: string): Promise<string> => {
       const signer = await this.signer;
-      return signer.nip04.decrypt(pubkey, ciphertext);
+      try {
+        return await signer.nip04.decrypt(pubkey, ciphertext);
+      } catch (e) {
+        if (e.name === 'AbortError') {
+          throw new HTTPException(408, {
+            message: 'Text was not decrypted quickly enough',
+          });
+        } else {
+          throw e;
+        }
+      }
     },
   };
 
   readonly nip44 = {
     encrypt: async (pubkey: string, plaintext: string): Promise<string> => {
       const signer = await this.signer;
-      return signer.nip44.encrypt(pubkey, plaintext);
+      try {
+        return await signer.nip44.encrypt(pubkey, plaintext);
+      } catch (e) {
+        if (e.name === 'AbortError') {
+          throw new HTTPException(408, {
+            message: 'Text was not encrypted quickly enough',
+          });
+        } else {
+          throw e;
+        }
+      }
     },
 
     decrypt: async (pubkey: string, ciphertext: string): Promise<string> => {
       const signer = await this.signer;
-      return signer.nip44.decrypt(pubkey, ciphertext);
+      try {
+        return await signer.nip44.decrypt(pubkey, ciphertext);
+      } catch (e) {
+        if (e.name === 'AbortError') {
+          throw new HTTPException(408, {
+            message: 'Text was not decrypted quickly enough',
+          });
+        } else {
+          throw e;
+        }
+      }
     },
   };
 

src/signers/ReadOnlySigner.ts

@@ -7,7 +7,7 @@ export class ReadOnlySigner implements NostrSigner {
 
   async signEvent(): Promise<NostrEvent> {
     throw new HTTPException(401, {
-      message: 'Log out and back in',
+      message: 'Log in with Nostr Connect to sign events',
     });
   }
 

src/storages/EventsDB.test.ts

@@ -191,3 +191,33 @@ Deno.test('inserting replaceable events', async () => {
   await eventsDB.event(newerEvent);
   assertEquals(await eventsDB.query([{ kinds: [0] }]), [newerEvent]);
 });
+
+Deno.test("throws a RelayError when querying an event with a large 'since'", async () => {
+  const { eventsDB } = await createDB();
+
+  await assertRejects(
+    () => eventsDB.query([{ since: 33333333333333 }]),
+    RelayError,
+    'since filter too far into the future',
+  );
+});
+
+Deno.test("throws a RelayError when querying an event with a large 'until'", async () => {
+  const { eventsDB } = await createDB();
+
+  await assertRejects(
+    () => eventsDB.query([{ until: 66666666666666 }]),
+    RelayError,
+    'until filter too far into the future',
+  );
+});
+
+Deno.test("throws a RelayError when querying an event with a large 'kind'", async () => {
+  const { eventsDB } = await createDB();
+
+  await assertRejects(
+    () => eventsDB.query([{ kinds: [99999999999999] }]),
+    RelayError,
+    'kind filter too far into the future',
+  );
+});

src/storages/EventsDB.ts

@@ -7,6 +7,7 @@ import { nip27 } from 'nostr-tools';
 
 import { Conf } from '@/config.ts';
 import { DittoTables } from '@/db/DittoTables.ts';
+import { dbEventCounter, dbQueryCounter } from '@/metrics.ts';
 import { RelayError } from '@/RelayError.ts';
 import { purifyEvent } from '@/storages/hydrate.ts';
 import { isNostrId, isURL } from '@/utils.ts';
@@ -44,15 +45,17 @@ class EventsDB implements NStore {
   constructor(private kysely: Kysely<DittoTables>) {
     this.store = new NDatabase(kysely, {
       fts: Conf.db.dialect,
+      timeoutStrategy: Conf.db.dialect === 'postgres' ? 'setStatementTimeout' : undefined,
       indexTags: EventsDB.indexTags,
       searchText: EventsDB.searchText,
     });
   }
 
   /** Insert an event (and its tags) into the database. */
-  async event(event: NostrEvent, _opts?: { signal?: AbortSignal }): Promise<void> {
+  async event(event: NostrEvent, opts: { signal?: AbortSignal; timeout?: number } = {}): Promise<void> {
     event = purifyEvent(event);
     this.console.debug('EVENT', JSON.stringify(event));
+    dbEventCounter.inc({ kind: event.kind });
 
     if (await this.isDeletedAdmin(event)) {
       throw new RelayError('blocked', 'event deleted by admin');
@@ -61,7 +64,7 @@ class EventsDB implements NStore {
     await this.deleteEventsAdmin(event);
 
     try {
-      await this.store.event(event);
+      await this.store.event(event, { ...opts, timeout: opts.timeout ?? 1000 });
     } catch (e) {
       if (e.message === 'Cannot add a deleted event') {
         throw new RelayError('blocked', 'event deleted by user');
@@ -135,19 +138,23 @@ class EventsDB implements NStore {
   }
 
   /** Get events for filters from the database. */
-  async query(filters: NostrFilter[], opts: { signal?: AbortSignal; limit?: number } = {}): Promise<NostrEvent[]> {
+  async query(
+    filters: NostrFilter[],
+    opts: { signal?: AbortSignal; timeout?: number; limit?: number } = {},
+  ): Promise<NostrEvent[]> {
     filters = await this.expandFilters(filters);
+    dbQueryCounter.inc();
 
     for (const filter of filters) {
       if (filter.since && filter.since >= 2_147_483_647) {
-        throw new Error('since filter too far into the future');
+        throw new RelayError('invalid', 'since filter too far into the future');
       }
       if (filter.until && filter.until >= 2_147_483_647) {
-        throw new Error('until filter too far into the future');
+        throw new RelayError('invalid', 'until filter too far into the future');
       }
       for (const kind of filter.kinds ?? []) {
         if (kind >= 2_147_483_647) {
-          throw new Error('kind filter too far into the future');
+          throw new RelayError('invalid', 'kind filter too far into the future');
         }
       }
     }
@@ -157,28 +164,28 @@ class EventsDB implements NStore {
 
     this.console.debug('REQ', JSON.stringify(filters));
 
-    return this.store.query(filters, opts);
+    return this.store.query(filters, { ...opts, timeout: opts.timeout ?? 1000 });
   }
 
   /** Delete events based on filters from the database. */
-  async remove(filters: NostrFilter[], _opts?: { signal?: AbortSignal }): Promise<void> {
+  async remove(filters: NostrFilter[], opts: { signal?: AbortSignal; timeout?: number } = {}): Promise<void> {
     if (!filters.length) return Promise.resolve();
     this.console.debug('DELETE', JSON.stringify(filters));
 
-    return this.store.remove(filters);
+    return this.store.remove(filters, { ...opts, timeout: opts.timeout ?? 3000 });
   }
 
   /** Get number of events that would be returned by filters. */
   async count(
     filters: NostrFilter[],
-    opts: { signal?: AbortSignal } = {},
+    opts: { signal?: AbortSignal; timeout?: number } = {},
   ): Promise<{ count: number; approximate: boolean }> {
     if (opts.signal?.aborted) return Promise.reject(abortError());
     if (!filters.length) return Promise.resolve({ count: 0, approximate: false });
 
     this.console.debug('COUNT', JSON.stringify(filters));
 
-    return this.store.count(filters);
+    return this.store.count(filters, { ...opts, timeout: opts.timeout ?? 500 });
   }
 
   /** Return only the tags that should be indexed. */
@@ -243,6 +250,8 @@ class EventsDB implements NStore {
 
   /** Converts filters to more performant, simpler filters that are better for SQLite. */
   async expandFilters(filters: NostrFilter[]): Promise<NostrFilter[]> {
+    filters = structuredClone(filters);
+
     for (const filter of filters) {
       if (filter.search) {
         const tokens = NIP50.parseInput(filter.search);
@@ -272,6 +281,12 @@ class EventsDB implements NStore {
 
         filter.search = tokens.filter((t) => typeof t === 'string').join(' ');
       }
+
+      if (filter.kinds) {
+        // Ephemeral events are not stored, so don't bother querying for them.
+        // If this results in an empty kinds array, NDatabase will remove the filter before querying and return no results.
+        filter.kinds = filter.kinds.filter((kind) => !NKinds.ephemeral(kind));
+      }
     }
 
     return filters;

src/views/mastodon/accounts.ts

@@ -82,6 +82,7 @@ async function renderAccount(
     username: parsed05?.nickname || npub.substring(0, 8),
     ditto: {
       accepts_zaps: Boolean(getLnurl({ lud06, lud16 })),
+      external_url: Conf.external(npub),
     },
     pleroma: {
       deactivated: names.has('disabled'),

src/views/mastodon/statuses.ts

@@ -124,6 +124,9 @@ async function renderStatus(event: DittoEvent, opts: RenderStatusOpts): Promise<
     uri: Conf.local(`/${note}`),
     url: Conf.local(`/${note}`),
     zapped: Boolean(zapEvent),
+    ditto: {
+      external_url: Conf.external(note),
+    },
     pleroma: {
       emoji_reactions: reactions,
       expires_at: !isNaN(expiresAt.getTime()) ? expiresAt.toISOString() : undefined,

src/workers/fetch.ts

@@ -1,8 +1,9 @@
 import * as Comlink from 'comlink';
 
+import { FetchWorker } from './fetch.worker.ts';
 import './handlers/abortsignal.ts';
 
-import type { FetchWorker } from './fetch.worker.ts';
+import { fetchCounter } from '@/metrics.ts';
 
 const worker = new Worker(new URL('./fetch.worker.ts', import.meta.url), { type: 'module' });
 const client = Comlink.wrap<typeof FetchWorker>(worker);
@@ -24,6 +25,7 @@ const fetchWorker: typeof fetch = async (...args) => {
   await ready;
   const [url, init] = serializeFetchArgs(args);
   const { body, signal, ...rest } = init;
+  fetchCounter.inc({ method: init.method });
   const result = await client.fetch(url, { ...rest, body: await prepareBodyForWorker(body) }, signal);
   return new Response(...result);
 };

src/workers/fetch.worker.ts

@@ -2,6 +2,7 @@ import Debug from '@soapbox/stickynotes/debug';
 import * as Comlink from 'comlink';
 
 import './handlers/abortsignal.ts';
+import '@/sentry.ts';
 
 const debug = Debug('ditto:fetch.worker');
 

src/workers/verify.worker.ts

@@ -3,6 +3,7 @@ import * as Comlink from 'comlink';
 import { VerifiedEvent, verifyEvent } from 'nostr-tools';
 
 import '@/nostr-wasm.ts';
+import '@/sentry.ts';
 
 export const VerifyWorker = {
   verifyEvent(event: NostrEvent): event is VerifiedEvent {