Merge branch 'user-stream' into 'develop'

Streaming: support user stream

See merge request soapbox-pub/ditto!23

src/app.ts

@@ -53,8 +53,6 @@ interface AppEnv extends HonoEnv {
     pubkey?: string;
     /** Hex secret key for the current user. Optional, but easiest way to use legacy Mastodon apps. */
     seckey?: string;
-    /** UUID from the access token. Used for WebSocket event signing. */
-    session?: string;
     /** NIP-98 signed event proving the pubkey is owned by the user. */
     proof?: Event<27235>;
   };

src/controllers/api/oauth.ts

@@ -1,4 +1,4 @@
-import { lodash, nip19, uuid62, z } from '@/deps.ts';
+import { lodash, nip19, z } from '@/deps.ts';
 import { AppController } from '@/app.ts';
 import { nostrNow } from '@/utils.ts';
 import { parseBody } from '@/utils/web.ts';
@@ -137,19 +137,12 @@ const oauthAuthorizeController: AppController = async (c) => {
   // Parsed FormData values.
   const { pubkey, nip19: nip19id, redirect_uri: redirectUri } = result.data;
 
-  /**
-   * Normally the auth token is just an npub, which is public information.
-   * The sessionId helps us know that Request "B" and Request "A" came from the same person.
-   * Useful for sending websocket events to the correct client.
-   */
-  const sessionId: string = uuid62.v4();
-
   if (pubkey) {
     const encoded = nip19.npubEncode(pubkey!);
-    const url = addCodeToRedirectUri(redirectUri, `${encoded}_${sessionId}`);
+    const url = addCodeToRedirectUri(redirectUri, encoded);
     return c.redirect(url);
   } else if (nip19id) {
-    const url = addCodeToRedirectUri(redirectUri, `${nip19id}_${sessionId}`);
+    const url = addCodeToRedirectUri(redirectUri, nip19id);
     return c.redirect(url);
   }
 

src/controllers/api/streaming.ts

@@ -1,21 +1,32 @@
-import { AppController } from '@/app.ts';
+import { type AppController } from '@/app.ts';
 import { z } from '@/deps.ts';
 import { type DittoFilter } from '@/filter.ts';
-import { TOKEN_REGEX } from '@/middleware/auth19.ts';
+import { getFeedPubkeys } from '@/queries.ts';
 import { Sub } from '@/subs.ts';
 import { toStatus } from '@/transformers/nostr-to-mastoapi.ts';
+import { bech32ToPubkey } from '@/utils.ts';
 
 /**
  * Streaming timelines/categories.
  * https://docs.joinmastodon.org/methods/streaming/#streams
  */
 const streamSchema = z.enum([
-  'nostr',
   'public',
+  'public:media',
   'public:local',
+  'public:local:media',
+  'public:remote',
+  'public:remote:media',
+  'hashtag',
+  'hashtag:local',
   'user',
+  'user:notification',
+  'list',
+  'direct',
 ]);
 
+type Stream = z.infer<typeof streamSchema>;
+
 const streamingController: AppController = (c) => {
   const upgrade = c.req.headers.get('upgrade');
   const token = c.req.headers.get('sec-websocket-protocol');
@@ -29,8 +40,8 @@ const streamingController: AppController = (c) => {
     return c.json({ error: 'Missing access token' }, 401);
   }
 
-  const match = token.match(new RegExp(`^${TOKEN_REGEX.source}$`));
+  const pubkey = token ? bech32ToPubkey(token) : undefined;
-  if (!match) {
+  if (!pubkey) {
     return c.json({ error: 'Invalid access token' }, 401);
   }
 
@@ -48,7 +59,7 @@ const streamingController: AppController = (c) => {
 
   socket.onopen = async () => {
     if (!stream) return;
-    const filter = topicToFilter(stream);
+    const filter = await topicToFilter(stream, pubkey, c.req.query());
 
     if (filter) {
       for await (const event of Sub.sub(socket, '1', [filter])) {
@@ -67,12 +78,27 @@ const streamingController: AppController = (c) => {
   return response;
 };
 
-function topicToFilter(topic: string): DittoFilter<1> | undefined {
+async function topicToFilter(
+  topic: Stream,
+  pubkey: string,
+  query: Record<string, string>,
+): Promise<DittoFilter<1> | undefined> {
   switch (topic) {
     case 'public':
       return { kinds: [1] };
     case 'public:local':
       return { kinds: [1], local: true };
+    case 'hashtag':
+      if (query.tag) return { kinds: [1], '#t': [query.tag] };
+      break;
+    case 'hashtag:local':
+      if (query.tag) return { kinds: [1], '#t': [query.tag], local: true };
+      break;
+    case 'user':
+      // HACK: this puts the user's entire contacts list into RAM,
+      // and then calls `matchFilters` over it. Refreshing the page
+      // is required after following a new user.
+      return { kinds: [1], authors: await getFeedPubkeys(pubkey) };
   }
 }
 

src/deps.ts

@@ -37,7 +37,6 @@ import 'npm:linkify-plugin-hashtag@^4.1.1';
 export { default as mime } from 'npm:mime@^3.0.0';
 export { unfurl } from 'npm:unfurl.js@^6.3.2';
 export { default as TTLCache } from 'npm:@isaacs/ttlcache@^1.4.1';
-export { default as uuid62 } from 'npm:uuid62@^1.0.2';
 // @deno-types="npm:@types/sanitize-html@2.9.0"
 export { default as sanitizeHtml } from 'npm:sanitize-html@^2.11.0';
 export { default as ISO6391 } from 'npm:iso-639-1@2.1.15';

src/middleware/auth19.ts

@@ -1,10 +1,8 @@
 import { type AppMiddleware } from '@/app.ts';
 import { getPublicKey, HTTPException, nip19 } from '@/deps.ts';
 
-/** The token includes a Bech32 Nostr ID (npub, nsec, etc) and an optional session ID. */
-const TOKEN_REGEX = new RegExp(`(${nip19.BECH32_REGEX.source})(?:_(\\w+))?`);
 /** We only accept "Bearer" type. */
-const BEARER_REGEX = new RegExp(`^Bearer (${TOKEN_REGEX.source})$`);
+const BEARER_REGEX = new RegExp(`^Bearer (${nip19.BECH32_REGEX.source})$`);
 
 /** NIP-19 auth middleware. */
 const auth19: AppMiddleware = async (c, next) => {
@@ -12,8 +10,7 @@ const auth19: AppMiddleware = async (c, next) => {
   const match = authHeader?.match(BEARER_REGEX);
 
   if (match) {
-    const [_, _token, bech32, session] = match;
+    const [_, bech32] = match;
-    c.set('session', session);
 
     try {
       const decoded = nip19.decode(bech32!);
@@ -47,4 +44,4 @@ const requireAuth: AppMiddleware = async (c, next) => {
   await next();
 };
 
-export { auth19, requireAuth, TOKEN_REGEX };
+export { auth19, requireAuth };

src/queries.ts

@@ -37,16 +37,25 @@ const getFollows = async (pubkey: string, timeout = 1000): Promise<Event<3> | un
   return event;
 };
 
-/** Get events from people the user follows. */
+/** Get pubkeys the user follows. */
-async function getFeed(pubkey: string, params: PaginationParams): Promise<Event<1>[]> {
+async function getFollowedPubkeys(pubkey: string): Promise<string[]> {
-  const event3 = await getFollows(pubkey);
+  const event = await getFollows(pubkey);
-  if (!event3) return [];
+  if (!event) return [];
 
-  const authors = event3.tags
+  return event.tags
     .filter((tag) => tag[0] === 'p')
     .map((tag) => tag[1]);
+}
 
-  authors.push(event3.pubkey); // see own events in feed
+/** Get pubkeys the user follows, including the user's own pubkey. */
+async function getFeedPubkeys(pubkey: string): Promise<string[]> {
+  const authors = await getFollowedPubkeys(pubkey);
+  return [...authors, pubkey];
+}
+
+/** Get events from people the user follows. */
+async function getFeed(pubkey: string, params: PaginationParams): Promise<Event<1>[]> {
+  const authors = await getFeedPubkeys(pubkey);
 
   const filter: Filter<1> = {
     authors,
@@ -103,6 +112,7 @@ export {
   getDescendants,
   getEvent,
   getFeed,
+  getFeedPubkeys,
   getFollows,
   getPublicFeed,
   isLocallyFollowed,

src/utils.ts

@@ -102,6 +102,7 @@ function isFollowing(source: Event<3>, targetPubkey: string): boolean {
 }
 
 export {
+  bech32ToPubkey,
   eventAge,
   eventDateComparator,
   findTag,