Merge branch 'relay' into 'develop'

Add a Nostr relay

See merge request soapbox-pub/ditto!6

src/app.ts

@@ -29,6 +29,7 @@ import {
 } from './controllers/api/statuses.ts';
 import { streamingController } from './controllers/api/streaming.ts';
 import { trendingTagsController } from './controllers/api/trends.ts';
+import { relayController } from './controllers/nostr/relay.ts';
 import { indexController } from './controllers/site.ts';
 import { hostMetaController } from './controllers/well-known/host-meta.ts';
 import { nodeInfoController, nodeInfoSchemaController } from './controllers/well-known/nodeinfo.ts';
@@ -60,6 +61,7 @@ app.use('*', logger());
 
 app.get('/api/v1/streaming', streamingController);
 app.get('/api/v1/streaming/', streamingController);
+app.get('/relay', relayController);
 
 app.use('*', cors({ origin: '*', exposeHeaders: ['link'] }), auth19, auth98());
 

src/client.ts

@@ -1,4 +1,4 @@
-import { Author, findReplyTag, matchFilter, RelayPool, TTLCache } from '@/deps.ts';
+import { Author, type Filter, findReplyTag, matchFilter, RelayPool, TTLCache } from '@/deps.ts';
 import { type Event, type SignedEvent } from '@/event.ts';
 
 import { Conf } from './config.ts';
@@ -29,17 +29,6 @@ function getPool(): Pool {
   return pool;
 }
 
-type Filter<K extends number = number> = {
-  ids?: string[];
-  kinds?: K[];
-  authors?: string[];
-  since?: number;
-  until?: number;
-  limit?: number;
-  search?: string;
-  [key: `#${string}`]: string[];
-};
-
 interface GetFilterOpts {
   timeout?: number;
 }

src/controllers/api/accounts.ts

@@ -1,7 +1,7 @@
 import { type AppController } from '@/app.ts';
 import { type Filter, findReplyTag, z } from '@/deps.ts';
 import { getAuthor, getFilter, getFollows, publish } from '@/client.ts';
-import { parseMetaContent } from '@/schema.ts';
+import { jsonMetaContentSchema } from '@/schemas/nostr.ts';
 import { signEvent } from '@/sign.ts';
 import { toAccount, toStatus } from '@/transformers/nostr-to-mastoapi.ts';
 import { buildLinkHeader, eventDateComparator, lookupAccount, nostrNow, paginationSchema, parseBody } from '@/utils.ts';
@@ -154,7 +154,7 @@ const updateCredentialsController: AppController = async (c) => {
     return c.json({ error: 'Could not find user.' }, 404);
   }
 
-  const meta = parseMetaContent(author);
+  const meta = jsonMetaContentSchema.parse(author.content);
   meta.name = result.data.display_name ?? meta.name;
   meta.about = result.data.note ?? meta.about;
 

src/controllers/nostr/relay.ts

@@ -0,0 +1,99 @@
+import { getFilters, insertEvent } from '@/db/events.ts';
+import { findUser } from '@/db/users.ts';
+import { jsonSchema } from '@/schema.ts';
+import {
+  type ClientCLOSE,
+  type ClientEVENT,
+  type ClientMsg,
+  clientMsgSchema,
+  type ClientREQ,
+} from '@/schemas/nostr.ts';
+
+import type { AppController } from '@/app.ts';
+import type { Filter } from '@/deps.ts';
+import type { SignedEvent } from '@/event.ts';
+
+/** Limit of events returned per-filter. */
+const FILTER_LIMIT = 100;
+
+type RelayMsg =
+  | ['EVENT', string, SignedEvent]
+  | ['NOTICE', string]
+  | ['EOSE', string]
+  | ['OK', string, boolean, string];
+
+function connectStream(socket: WebSocket) {
+  socket.onmessage = (e) => {
+    const result = jsonSchema.pipe(clientMsgSchema).safeParse(e.data);
+    if (result.success) {
+      handleMsg(result.data);
+    } else {
+      send(['NOTICE', 'Invalid message.']);
+    }
+  };
+
+  function handleMsg(msg: ClientMsg) {
+    switch (msg[0]) {
+      case 'REQ':
+        handleReq(msg);
+        return;
+      case 'EVENT':
+        handleEvent(msg);
+        return;
+      case 'CLOSE':
+        handleClose(msg);
+        return;
+    }
+  }
+
+  async function handleReq([_, sub, ...filters]: ClientREQ) {
+    for (const event of await getFilters(prepareFilters(filters))) {
+      send(['EVENT', sub, event]);
+    }
+    send(['EOSE', sub]);
+  }
+
+  async function handleEvent([_, event]: ClientEVENT) {
+    if (await findUser({ pubkey: event.pubkey })) {
+      insertEvent(event);
+      send(['OK', event.id, true, '']);
+    } else {
+      send(['OK', event.id, false, 'blocked: only registered users can post']);
+    }
+  }
+
+  function handleClose([_, _sub]: ClientCLOSE) {
+    // TODO: ???
+    return;
+  }
+
+  function send(msg: RelayMsg) {
+    return socket.send(JSON.stringify(msg));
+  }
+}
+
+/** Enforce the filters with certain criteria. */
+function prepareFilters(filters: ClientREQ[2][]): Filter[] {
+  return filters.map((filter) => ({
+    ...filter,
+    // Limit the number of events returned per-filter.
+    limit: Math.min(filter.limit || FILTER_LIMIT, FILTER_LIMIT),
+    // Return only local events unless the query is already narrow.
+    local: !filter.ids?.length && !filter.authors?.length,
+  }));
+}
+
+const relayController: AppController = (c) => {
+  const upgrade = c.req.headers.get('upgrade');
+
+  if (upgrade?.toLowerCase() !== 'websocket') {
+    return c.text('Please use a Nostr client to connect.', 400);
+  }
+
+  const { socket, response } = Deno.upgradeWebSocket(c.req.raw);
+  connectStream(socket);
+
+  return response;
+};
+
+export { relayController };

src/db/events.ts

@@ -15,6 +15,7 @@ const tagConditions: Record<string, TagCondition> = {
   't': ({ count }) => count < 5,
 };
 
+/** Insert an event (and its tags) into the database. */
 function insertEvent(event: SignedEvent): Promise<void> {
   return db.transaction().execute(async (trx) => {
     await trx.insertInto('events')
@@ -50,28 +51,42 @@ function insertEvent(event: SignedEvent): Promise<void> {
   });
 }
 
-function getFilterQuery(filter: Filter) {
+/** Custom filter interface that extends Nostr filters with extra options for Ditto. */
+interface DittoFilter<K extends number = number> extends Filter<K> {
+  local?: boolean;
+}
+
+/** Build the query for a filter. */
+function getFilterQuery(filter: DittoFilter) {
   let query = db
     .selectFrom('events')
-    .select(['id', 'kind', 'pubkey', 'content', 'tags', 'created_at', 'sig'])
+    .select([
-    .orderBy('created_at', 'desc');
+      'events.id',
+      'events.kind',
+      'events.pubkey',
+      'events.content',
+      'events.tags',
+      'events.created_at',
+      'events.sig',
+    ])
+    .orderBy('events.created_at', 'desc');
 
   for (const key of Object.keys(filter)) {
-    switch (key as keyof Filter) {
+    switch (key as keyof DittoFilter) {
       case 'ids':
-        query = query.where('id', 'in', filter.ids!);
+        query = query.where('events.id', 'in', filter.ids!);
         break;
       case 'kinds':
-        query = query.where('kind', 'in', filter.kinds!);
+        query = query.where('events.kind', 'in', filter.kinds!);
         break;
       case 'authors':
-        query = query.where('pubkey', 'in', filter.authors!);
+        query = query.where('events.pubkey', 'in', filter.authors!);
         break;
       case 'since':
-        query = query.where('created_at', '>=', filter.since!);
+        query = query.where('events.created_at', '>=', filter.since!);
         break;
       case 'until':
-        query = query.where('created_at', '<=', filter.until!);
+        query = query.where('events.created_at', '<=', filter.until!);
         break;
       case 'limit':
         query = query.limit(filter.limit!);
@@ -81,19 +96,24 @@ function getFilterQuery(filter: Filter) {
     if (key.startsWith('#')) {
       const tag = key.replace(/^#/, '');
       const value = filter[key as `#${string}`] as string[];
-      return query
+      query = query
         .leftJoin('tags', 'tags.event_id', 'events.id')
         .where('tags.tag', '=', tag)
         .where('tags.value_1', 'in', value) as typeof query;
     }
   }
 
+  if (filter.local) {
+    query = query.innerJoin('users', 'users.pubkey', 'events.pubkey');
+  }
+
   return query;
 }
 
-async function getFilters<K extends number>(filters: [Filter<K>]): Promise<SignedEvent<K>[]>;
+/** Get events for filters from the database. */
-async function getFilters(filters: Filter[]): Promise<SignedEvent[]>;
+async function getFilters<K extends number>(filters: [DittoFilter<K>]): Promise<SignedEvent<K>[]>;
-async function getFilters(filters: Filter[]) {
+async function getFilters(filters: DittoFilter[]): Promise<SignedEvent[]>;
+async function getFilters(filters: DittoFilter[]) {
   const queries = filters
     .map(getFilterQuery)
     .map((query) => query.execute());
@@ -105,17 +125,21 @@ async function getFilters(filters: Filter[]) {
   ));
 }
 
-function getFilter<K extends number = number>(filter: Filter<K>): Promise<SignedEvent<K>[]> {
+/** Get events for a filter from the database. */
+function getFilter<K extends number = number>(filter: DittoFilter<K>): Promise<SignedEvent<K>[]> {
   return getFilters<K>([filter]);
 }
 
 /** Returns whether the pubkey is followed by a local user. */
 async function isLocallyFollowed(pubkey: string): Promise<boolean> {
-  const event = await getFilterQuery({ kinds: [3], '#p': [pubkey], limit: 1 })
+  return Boolean(
-    .innerJoin('users', 'users.pubkey', 'events.pubkey')
+    await getFilterQuery({
-    .executeTakeFirst();
+      kinds: [3],
-
+      '#p': [pubkey],
-  return !!event;
+      limit: 1,
+      local: true,
+    }).executeTakeFirst(),
+  );
 }
 
 export { getFilter, getFilters, insertEvent, isLocallyFollowed };

src/middleware/auth98.ts

@@ -2,7 +2,8 @@ import { type AppMiddleware } from '@/app.ts';
 import { Conf } from '@/config.ts';
 import { HTTPException } from '@/deps.ts';
 import { type Event } from '@/event.ts';
-import { decode64Schema, jsonSchema, signedEventSchema } from '@/schema.ts';
+import { decode64Schema, jsonSchema } from '@/schema.ts';
+import { signedEventSchema } from '@/schemas/nostr.ts';
 import { eventAge, findTag, sha256, Time } from '@/utils.ts';
 
 const decodeEventSchema = decode64Schema.pipe(jsonSchema).pipe(signedEventSchema);

src/schema.ts

@@ -1,8 +1,4 @@
-import { verifySignature, z } from '@/deps.ts';
+import { z } from '@/deps.ts';
-
-import type { Event } from './event.ts';
-
-const optionalString = z.string().optional().catch(undefined);
 
 /** Validates individual items in an array, dropping any that aren't valid. */
 function filteredArray<T extends z.ZodTypeAny>(schema: T) {
@@ -24,31 +20,6 @@ const jsonSchema = z.string().transform((value, ctx) => {
   }
 });
 
-const metaContentSchema = z.object({
-  name: optionalString,
-  about: optionalString,
-  picture: optionalString,
-  banner: optionalString,
-  nip05: optionalString,
-  lud16: optionalString,
-});
-
-/** Author metadata from Event<0>. */
-type MetaContent = z.infer<typeof metaContentSchema>;
-
-/**
- * Get (and validate) data from a kind 0 event.
- * https://github.com/nostr-protocol/nips/blob/master/01.md
- */
-function parseMetaContent(event: Event<0>): MetaContent {
-  try {
-    const json = JSON.parse(event.content);
-    return metaContentSchema.passthrough().parse(json);
-  } catch (_e) {
-    return {};
-  }
-}
-
 /** Alias for `safeParse`, but instead of returning a success object it returns the value (or undefined on fail). */
 function parseValue<T>(schema: z.ZodType<T>, value: unknown): T | undefined {
   const result = schema.safeParse(value);
@@ -67,20 +38,6 @@ const relaySchema = z.custom<URL>((relay) => {
   }
 });
 
-const hexIdSchema = z.string().regex(/^[0-9a-f]{64}$/);
-
-const eventSchema = z.object({
-  id: hexIdSchema,
-  kind: z.number(),
-  tags: z.array(z.array(z.string())),
-  content: z.string(),
-  created_at: z.number(),
-  pubkey: hexIdSchema,
-  sig: z.string(),
-});
-
-const signedEventSchema = eventSchema.refine(verifySignature);
-
 const emojiTagSchema = z.tuple([z.literal('emoji'), z.string(), z.string().url()]);
 
 /** https://developer.mozilla.org/en-US/docs/Glossary/Base64#the_unicode_problem */
@@ -97,17 +54,4 @@ const decode64Schema = z.string().transform((value, ctx) => {
 
 const hashtagSchema = z.string().regex(/^\w{1,30}$/);
 
-export {
+export { decode64Schema, emojiTagSchema, filteredArray, hashtagSchema, jsonSchema, parseRelay, relaySchema };
-  decode64Schema,
-  emojiTagSchema,
-  filteredArray,
-  hashtagSchema,
-  hexIdSchema,
-  jsonSchema,
-  type MetaContent,
-  metaContentSchema,
-  parseMetaContent,
-  parseRelay,
-  relaySchema,
-  signedEventSchema,
-};

src/schemas/nostr.ts

@@ -0,0 +1,81 @@
+import { verifySignature, z } from '@/deps.ts';
+
+import { jsonSchema } from '../schema.ts';
+
+/** Schema to validate Nostr hex IDs such as event IDs and pubkeys. */
+const nostrIdSchema = z.string().regex(/^[0-9a-f]{64}$/);
+
+/** Nostr event schema. */
+const eventSchema = z.object({
+  id: nostrIdSchema,
+  kind: z.number(),
+  tags: z.array(z.array(z.string())),
+  content: z.string(),
+  created_at: z.number(),
+  pubkey: nostrIdSchema,
+  sig: z.string(),
+});
+
+/** Nostr event schema that also verifies the event's signature. */
+const signedEventSchema = eventSchema.refine(verifySignature);
+
+/** Nostr relay filter schema. */
+const filterSchema = z.object({
+  kinds: z.number().int().positive().array().optional(),
+  ids: nostrIdSchema.array().optional(),
+  authors: nostrIdSchema.array().optional(),
+  since: z.number().int().positive().optional(),
+  until: z.number().int().positive().optional(),
+  limit: z.number().int().positive().optional(),
+}).passthrough().and(
+  z.record(
+    z.custom<`#${string}`>((val) => typeof val === 'string' && val.startsWith('#')),
+    z.string().array(),
+  ).catch({}),
+);
+
+const clientReqSchema = z.tuple([z.literal('REQ'), z.string().min(1)]).rest(filterSchema);
+const clientEventSchema = z.tuple([z.literal('EVENT'), signedEventSchema]);
+const clientCloseSchema = z.tuple([z.literal('CLOSE'), z.string().min(1)]);
+
+/** Client message to a Nostr relay. */
+const clientMsgSchema = z.union([
+  clientReqSchema,
+  clientEventSchema,
+  clientCloseSchema,
+]);
+
+/** REQ message from client to relay. */
+type ClientREQ = z.infer<typeof clientReqSchema>;
+/** EVENT message from client to relay. */
+type ClientEVENT = z.infer<typeof clientEventSchema>;
+/** CLOSE message from client to relay. */
+type ClientCLOSE = z.infer<typeof clientCloseSchema>;
+/** Client message to a Nostr relay. */
+type ClientMsg = z.infer<typeof clientMsgSchema>;
+
+/** Kind 0 content schema. */
+const metaContentSchema = z.object({
+  name: z.string().optional().catch(undefined),
+  about: z.string().optional().catch(undefined),
+  picture: z.string().optional().catch(undefined),
+  banner: z.string().optional().catch(undefined),
+  nip05: z.string().optional().catch(undefined),
+  lud16: z.string().optional().catch(undefined),
+}).partial().passthrough();
+
+/** Parses kind 0 content from a JSON string. */
+const jsonMetaContentSchema = jsonSchema.pipe(metaContentSchema).catch({});
+
+export {
+  type ClientCLOSE,
+  type ClientEVENT,
+  type ClientMsg,
+  clientMsgSchema,
+  type ClientREQ,
+  filterSchema,
+  jsonMetaContentSchema,
+  metaContentSchema,
+  nostrIdSchema,
+  signedEventSchema,
+};

src/sign.ts

@@ -1,6 +1,6 @@
 import { type AppContext } from '@/app.ts';
 import { getEventHash, getPublicKey, getSignature, HTTPException, z } from '@/deps.ts';
-import { signedEventSchema } from '@/schema.ts';
+import { signedEventSchema } from '@/schemas/nostr.ts';
 import { ws } from '@/stream.ts';
 
 import type { Event, EventTemplate, SignedEvent } from '@/event.ts';

src/transformers/nostr-to-activitypub.ts

@@ -1,5 +1,5 @@
 import { Conf } from '@/config.ts';
-import { parseMetaContent } from '@/schema.ts';
+import { jsonMetaContentSchema } from '@/schemas/nostr.ts';
 import { getPublicKeyPem } from '@/utils/rsa.ts';
 
 import type { Event } from '@/event.ts';
@@ -7,7 +7,7 @@ import type { Actor } from '@/schemas/activitypub.ts';
 
 /** Nostr metadata event to ActivityPub actor. */
 async function toActor(event: Event<0>, username: string): Promise<Actor | undefined> {
-  const content = parseMetaContent(event);
+  const content = jsonMetaContentSchema.parse(event.content);
 
   return {
     type: 'Person',

src/transformers/nostr-to-mastoapi.ts

@@ -6,7 +6,8 @@ import { findReplyTag, lodash, nip19, sanitizeHtml, TTLCache, unfurl, z } from '
 import { type Event } from '@/event.ts';
 import { verifyNip05Cached } from '@/nip05.ts';
 import { getMediaLinks, type MediaLink, parseNoteContent } from '@/note.ts';
-import { emojiTagSchema, filteredArray, type MetaContent, parseMetaContent } from '@/schema.ts';
+import { emojiTagSchema, filteredArray } from '@/schema.ts';
+import { jsonMetaContentSchema } from '@/schemas/nostr.ts';
 import { type Nip05, nostrDate, parseNip05, Time } from '@/utils.ts';
 
 const DEFAULT_AVATAR = 'https://gleasonator.com/images/avi.png';
@@ -20,7 +21,7 @@ async function toAccount(event: Event<0>, opts: ToAccountOpts = {}) {
   const { withSource = false } = opts;
 
   const { pubkey } = event;
-  const { name, nip05, picture, banner, about }: MetaContent = parseMetaContent(event);
+  const { name, nip05, picture, banner, about } = jsonMetaContentSchema.parse(event.content);
   const npub = nip19.npubEncode(pubkey);
 
   let parsed05: Nip05 | undefined;

src/trends.ts

@@ -1,5 +1,6 @@
 import { Sqlite } from '@/deps.ts';
-import { hashtagSchema, hexIdSchema } from '@/schema.ts';
+import { hashtagSchema } from '@/schema.ts';
+import { nostrIdSchema } from '@/schemas/nostr.ts';
 import { Time } from '@/utils.ts';
 import { generateDateRange } from '@/utils/time.ts';
 
@@ -100,7 +101,7 @@ class TrendsDB {
   }
 
   addTagUsages(pubkey: string, hashtags: string[], date = new Date()): void {
-    const pubkey8 = hexIdSchema.parse(pubkey).substring(0, 8);
+    const pubkey8 = nostrIdSchema.parse(pubkey).substring(0, 8);
     const tags = hashtagSchema.array().min(1).parse(hashtags);
 
     this.#db.query(