Merge branch 'kysely' into 'develop'

Rework the data layer

See merge request soapbox-pub/ditto!5

.gitlab-ci.yml

@@ -1,4 +1,4 @@
-image: denoland/deno:1.36.0
+image: denoland/deno:1.36.1
 
 default:
   interruptible: true

.tool-versions

@@ -1 +1 @@
-deno 1.36.0
+deno 1.36.1

deno.json

@@ -2,12 +2,15 @@
   "$schema": "https://deno.land/x/deno@v1.32.3/cli/schemas/config-file.v1.json",
   "lock": false,
   "tasks": {
-    "start": "deno run --allow-read --allow-write=data --allow-env --allow-net --unstable --watch src/server.ts",
+    "start": "deno run --allow-read --allow-write=data --allow-env --allow-net --unstable src/server.ts",
-    "test": "deno test --allow-read --allow-write=data --allow-env --unstable src",
+    "dev": "deno run --allow-read --allow-write=data --allow-env --allow-net --unstable --watch src/server.ts",
+    "debug": "deno run --allow-read --allow-write=data --allow-env --allow-net --unstable --inspect src/server.ts",
+    "test": "DB_PATH=\":memory:\" deno test --allow-read --allow-write=data --allow-env --unstable src",
     "check": "deno check --unstable src/server.ts"
   },
   "imports": {
-    "@/": "./src/"
+    "@/": "./src/",
+    "~/": "./"
   },
   "lint": {
     "include": ["src/"],

fixtures/events/55920b75.json

@@ -0,0 +1,15 @@
+{
+  "kind": 1,
+  "content": "I'm vegan btw",
+  "tags": [
+    [
+      "proxy",
+      "https://gleasonator.com/objects/8f6fac53-4f66-4c6e-ac7d-92e5e78c3e79",
+      "activitypub"
+    ]
+  ],
+  "pubkey": "79c2cae114ea28a981e7559b4fe7854a473521a8d22a66bbab9fa248eb820ff6",
+  "created_at": 1691091365,
+  "id": "55920b758b9c7b17854b6e3d44e6a02a83d1cb49e1227e75a30426dea94d4cb2",
+  "sig": "a72f12c08f18e85d98fb92ae89e2fe63e48b8864c5e10fbdd5335f3c9f936397a6b0a7350efe251f8168b1601d7012d4a6d0ee6eec958067cf22a14f5a5ea579"
+}

src/config.ts

@@ -45,6 +45,9 @@ const Conf = {
   get localDomain() {
     return Deno.env.get('LOCAL_DOMAIN') || 'http://localhost:8000';
   },
+  get dbPath() {
+    return Deno.env.get('DB_PATH') || 'data/db.sqlite3';
+  },
   get postCharLimit() {
     return Number(Deno.env.get('POST_CHAR_LIMIT') || 5000);
   },

src/controllers/activitypub/actor.ts

@@ -1,5 +1,5 @@
 import { getAuthor } from '@/client.ts';
-import { db } from '@/db.ts';
+import { findUser } from '@/db/users.ts';
 import { toActor } from '@/transformers/nostr-to-activitypub.ts';
 import { activityJson } from '@/utils.ts';
 
@@ -7,7 +7,9 @@ import type { AppContext, AppController } from '@/app.ts';
 
 const actorController: AppController = async (c) => {
   const username = c.req.param('username');
-  const user = await db.users.findFirst({ where: { username } });
+
+  const user = await findUser({ username });
+  if (!user) return notFound(c);
 
   const event = await getAuthor(user.pubkey);
   if (!event) return notFound(c);

src/controllers/api/statuses.ts

@@ -95,8 +95,8 @@ const contextController: AppController = async (c) => {
     const descendantEvents = await getDescendants(event.id);
 
     return c.json({
-      ancestors: (await Promise.all((ancestorEvents).map(toStatus))).filter(Boolean),
+      ancestors: (await Promise.all(ancestorEvents.map(toStatus))).filter(Boolean),
-      descendants: (await Promise.all((descendantEvents).map(toStatus))).filter(Boolean),
+      descendants: (await Promise.all(descendantEvents.map(toStatus))).filter(Boolean),
     });
   }
 

src/controllers/well-known/nostr.ts

@@ -1,5 +1,5 @@
 import { Conf } from '@/config.ts';
-import { db } from '@/db.ts';
+import { findUser } from '@/db/users.ts';
 import { z } from '@/deps.ts';
 
 import type { AppController } from '@/app.ts';
@@ -11,24 +11,19 @@ const nameSchema = z.string().min(1).regex(/^\w+$/);
  * https://github.com/nostr-protocol/nips/blob/master/05.md
  */
 const nostrController: AppController = async (c) => {
-  try {
+  const name = nameSchema.safeParse(c.req.query('name'));
-    const name = nameSchema.parse(c.req.query('name'));
+  const user = name.success ? await findUser({ username: name.data }) : null;
-    const user = await db.users.findFirst({ where: { username: name } });
-    const relay = Conf.relay;
 
-    return c.json({
+  if (!user) return c.json({ names: {}, relays: {} });
-      names: {
+
-        [user.username]: user.pubkey,
+  return c.json({
-      },
+    names: {
-      relays: relay
+      [user.username]: user.pubkey,
-        ? {
+    },
-          [user.pubkey]: [relay],
+    relays: {
-        }
+      [user.pubkey]: [Conf.relay],
-        : {},
+    },
-    });
+  });
-  } catch (_e) {
-    return c.json({ names: {}, relays: {} });
-  }
 };
 
 export { nostrController };

src/controllers/well-known/webfinger.ts

@@ -1,9 +1,9 @@
 import { Conf } from '@/config.ts';
-import { db } from '@/db.ts';
 import { nip19, z } from '@/deps.ts';
 
 import type { AppContext, AppController } from '@/app.ts';
 import type { Webfinger } from '@/schemas/webfinger.ts';
+import { findUser } from '@/db/users.ts';
 
 const webfingerQuerySchema = z.object({
   resource: z.string().url(),
@@ -37,25 +37,26 @@ const acctSchema = z.custom<URL>((value) => value instanceof URL)
   });
 
 async function handleAcct(c: AppContext, resource: URL): Promise<Response> {
-  try {
+  const result = acctSchema.safeParse(resource);
-    const [username, host] = acctSchema.parse(resource);
+  if (!result.success) {
-    const user = await db.users.findFirst({ where: { username } });
+    return c.json({ error: 'Invalid acct URI', schema: result.error }, 400);
-
-    const json = renderWebfinger({
-      pubkey: user.pubkey,
-      username: user.username,
-      subject: `acct:${username}@${host}`,
-    });
-
-    c.header('content-type', 'application/jrd+json');
-    return c.body(JSON.stringify(json));
-  } catch (e) {
-    if (e instanceof z.ZodError) {
-      return c.json({ error: 'Invalid acct URI', schema: e }, 400);
-    } else {
-      return c.json({ error: 'Not found' }, 404);
-    }
   }
+
+  const [username, host] = result.data;
+  const user = await findUser({ username });
+
+  if (!user) {
+    return c.json({ error: 'Not found' }, 404);
+  }
+
+  const json = renderWebfinger({
+    pubkey: user.pubkey,
+    username: user.username,
+    subject: `acct:${username}@${host}`,
+  });
+
+  c.header('content-type', 'application/jrd+json');
+  return c.body(JSON.stringify(json));
 }
 
 interface RenderWebfingerOpts {

src/db.ts

@@ -1,18 +1,56 @@
-import { createPentagon, z } from '@/deps.ts';
+import fs from 'node:fs/promises';
-import { hexIdSchema } from '@/schema.ts';
+import path from 'node:path';
 
-const kv = await Deno.openKv();
+import { DenoSqliteDialect, FileMigrationProvider, Kysely, Migrator, Sqlite } from '@/deps.ts';
+import { Conf } from '@/config.ts';
 
-const userSchema = z.object({
+interface DittoDB {
-  pubkey: hexIdSchema.describe('primary'),
+  events: EventRow;
-  username: z.string().regex(/^\w{1,30}$/).describe('unique'),
+  tags: TagRow;
-  createdAt: z.date(),
+  users: UserRow;
+}
+
+interface EventRow {
+  id: string;
+  kind: number;
+  pubkey: string;
+  content: string;
+  created_at: number;
+  tags: string;
+  sig: string;
+}
+
+interface TagRow {
+  tag: string;
+  value_1: string | null;
+  value_2: string | null;
+  value_3: string | null;
+  event_id: string;
+}
+
+interface UserRow {
+  pubkey: string;
+  username: string;
+  inserted_at: Date;
+}
+
+const db = new Kysely<DittoDB>({
+  dialect: new DenoSqliteDialect({
+    database: new Sqlite(Conf.dbPath),
+  }),
 });
 
-const db = createPentagon(kv, {
+const migrator = new Migrator({
-  users: {
+  db,
-    schema: userSchema,
+  provider: new FileMigrationProvider({
-  },
+    fs,
+    path,
+    migrationFolder: new URL(import.meta.resolve('./db/migrations')).pathname,
+  }),
 });
 
-export { db };
+console.log('Running migrations...');
+const results = await migrator.migrateToLatest();
+console.log('Migrations finished:', results);
+
+export { db, type DittoDB, type EventRow, type TagRow, type UserRow };

src/db/events.test.ts

@@ -0,0 +1,17 @@
+import event55920b75 from '~/fixtures/events/55920b75.json' assert { type: 'json' };
+import { assertEquals } from '@/deps-test.ts';
+
+import { getFilter, insertEvent } from './events.ts';
+
+Deno.test('insert and filter events', async () => {
+  await insertEvent(event55920b75);
+
+  assertEquals(await getFilter({ kinds: [1] }), [event55920b75]);
+  assertEquals(await getFilter({ kinds: [3] }), []);
+  assertEquals(await getFilter({ since: 1691091000 }), [event55920b75]);
+  assertEquals(await getFilter({ until: 1691091000 }), []);
+  assertEquals(
+    await getFilter({ '#proxy': ['https://gleasonator.com/objects/8f6fac53-4f66-4c6e-ac7d-92e5e78c3e79'] }),
+    [event55920b75],
+  );
+});

src/db/events.ts

@@ -0,0 +1,121 @@
+import { type Filter, type Insertable } from '@/deps.ts';
+import { type SignedEvent } from '@/event.ts';
+
+import { db, type TagRow } from '@/db.ts';
+
+type TagCondition = ({ event, count }: { event: SignedEvent; count: number }) => boolean;
+
+/** Conditions for when to index certain tags. */
+const tagConditions: Record<string, TagCondition> = {
+  'd': ({ event, count }) => 30000 <= event.kind && event.kind < 40000 && count === 0,
+  'e': ({ count }) => count < 15,
+  'p': ({ event, count }) => event.kind === 3 || count < 15,
+  'proxy': ({ count }) => count === 0,
+  'q': ({ event, count }) => event.kind === 1 && count === 0,
+  't': ({ count }) => count < 5,
+};
+
+function insertEvent(event: SignedEvent): Promise<void> {
+  return db.transaction().execute(async (trx) => {
+    await trx.insertInto('events')
+      .values({
+        ...event,
+        tags: JSON.stringify(event.tags),
+      })
+      .executeTakeFirst();
+
+    const tagCounts: Record<string, number> = {};
+    const tags = event.tags.reduce<Insertable<TagRow>[]>((results, tag) => {
+      const tagName = tag[0];
+      tagCounts[tagName] = (tagCounts[tagName] || 0) + 1;
+
+      if (tagConditions[tagName]?.({ event, count: tagCounts[tagName] - 1 })) {
+        results.push({
+          event_id: event.id,
+          tag: tagName,
+          value_1: tag[1] || null,
+          value_2: tag[2] || null,
+          value_3: tag[3] || null,
+        });
+      }
+
+      return results;
+    }, []);
+
+    await Promise.all(tags.map((tag) => {
+      return trx.insertInto('tags')
+        .values(tag)
+        .execute();
+    }));
+  });
+}
+
+function getFilterQuery(filter: Filter) {
+  let query = db
+    .selectFrom('events')
+    .select(['id', 'kind', 'pubkey', 'content', 'tags', 'created_at', 'sig'])
+    .orderBy('created_at', 'desc');
+
+  for (const key of Object.keys(filter)) {
+    switch (key as keyof Filter) {
+      case 'ids':
+        query = query.where('id', 'in', filter.ids!);
+        break;
+      case 'kinds':
+        query = query.where('kind', 'in', filter.kinds!);
+        break;
+      case 'authors':
+        query = query.where('pubkey', 'in', filter.authors!);
+        break;
+      case 'since':
+        query = query.where('created_at', '>=', filter.since!);
+        break;
+      case 'until':
+        query = query.where('created_at', '<=', filter.until!);
+        break;
+      case 'limit':
+        query = query.limit(filter.limit!);
+        break;
+    }
+
+    if (key.startsWith('#')) {
+      const tag = key.replace(/^#/, '');
+      const value = filter[key as `#${string}`] as string[];
+      return query
+        .leftJoin('tags', 'tags.event_id', 'events.id')
+        .where('tags.tag', '=', tag)
+        .where('tags.value_1', 'in', value) as typeof query;
+    }
+  }
+
+  return query;
+}
+
+async function getFilters<K extends number>(filters: [Filter<K>]): Promise<SignedEvent<K>[]>;
+async function getFilters(filters: Filter[]): Promise<SignedEvent[]>;
+async function getFilters(filters: Filter[]) {
+  const queries = filters
+    .map(getFilterQuery)
+    .map((query) => query.execute());
+
+  const events = (await Promise.all(queries)).flat();
+
+  return events.map((event) => (
+    { ...event, tags: JSON.parse(event.tags) }
+  ));
+}
+
+function getFilter<K extends number = number>(filter: Filter<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 })
+    .innerJoin('users', 'users.pubkey', 'events.pubkey')
+    .executeTakeFirst();
+
+  return !!event;
+}
+
+export { getFilter, getFilters, insertEvent, isLocallyFollowed };

src/db/migrations/000_create_events.ts

@@ -0,0 +1,66 @@
+import { Kysely, sql } from '@/deps.ts';
+
+export async function up(db: Kysely<any>): Promise<void> {
+  await db.schema
+    .createTable('events')
+    .addColumn('id', 'text', (col) => col.primaryKey())
+    .addColumn('kind', 'integer', (col) => col.notNull())
+    .addColumn('pubkey', 'text', (col) => col.notNull())
+    .addColumn('content', 'text', (col) => col.notNull())
+    .addColumn('created_at', 'integer', (col) => col.notNull())
+    .addColumn('tags', 'text', (col) => col.notNull())
+    .addColumn('sig', 'text', (col) => col.notNull())
+    .execute();
+
+  await db.schema
+    .createTable('tags')
+    .addColumn('tag', 'text', (col) => col.notNull())
+    .addColumn('value_1', 'text')
+    .addColumn('value_2', 'text')
+    .addColumn('value_3', 'text')
+    .addColumn('event_id', 'text', (col) => col.notNull())
+    .execute();
+
+  await db.schema
+    .createTable('users')
+    .addColumn('pubkey', 'text', (col) => col.primaryKey())
+    .addColumn('username', 'text', (col) => col.notNull().unique())
+    .addColumn('inserted_at', 'datetime', (col) => col.notNull().defaultTo(sql`CURRENT_TIMESTAMP`))
+    .execute();
+
+  await db.schema
+    .createIndex('idx_events_kind')
+    .on('events')
+    .column('kind')
+    .execute();
+
+  await db.schema
+    .createIndex('idx_events_pubkey')
+    .on('events')
+    .column('pubkey')
+    .execute();
+
+  await db.schema
+    .createIndex('idx_tags_tag')
+    .on('tags')
+    .column('tag')
+    .execute();
+
+  await db.schema
+    .createIndex('idx_tags_value_1')
+    .on('tags')
+    .column('value_1')
+    .execute();
+
+  await db.schema
+    .createIndex('idx_tags_event_id')
+    .on('tags')
+    .column('event_id')
+    .execute();
+}
+
+export async function down(db: Kysely<any>): Promise<void> {
+  await db.schema.dropTable('events').execute();
+  await db.schema.dropTable('tags').execute();
+  await db.schema.dropTable('users').execute();
+}

src/db/users.ts

@@ -0,0 +1,27 @@
+import { type Insertable } from '@/deps.ts';
+
+import { db, type UserRow } from '../db.ts';
+
+/** Adds a user to the database. */
+function insertUser(user: Insertable<UserRow>) {
+  return db.insertInto('users').values(user).execute();
+}
+
+/**
+ * Finds a single user based on one or more properties.
+ *
+ * ```ts
+ * await findUser({ username: 'alex' });
+ * ```
+ */
+function findUser(user: Partial<Insertable<UserRow>>) {
+  let query = db.selectFrom('users').selectAll();
+
+  for (const [key, value] of Object.entries(user)) {
+    query = query.where(key as keyof UserRow, '=', value);
+  }
+
+  return query.executeTakeFirst();
+}
+
+export { findUser, insertUser };

src/deps-test.ts

@@ -1 +1 @@
-export { assert, assertEquals, assertThrows } from 'https://deno.land/std@0.177.0/testing/asserts.ts';
+export { assert, assertEquals, assertThrows } from 'https://deno.land/std@0.198.0/assert/mod.ts';

src/deps.ts

@@ -49,4 +49,13 @@ export { generateSeededRsa } from 'https://gitlab.com/soapbox-pub/seeded-rsa/-/r
 export * as secp from 'npm:@noble/secp256k1@^2.0.0';
 export { LRUCache } from 'npm:lru-cache@^10.0.0';
 export { DB as Sqlite } from 'https://deno.land/x/sqlite@v3.7.3/mod.ts';
-export * as dotenv from 'https://deno.land/std@0.197.0/dotenv/mod.ts';
+export * as dotenv from 'https://deno.land/std@0.198.0/dotenv/mod.ts';
+export {
+  FileMigrationProvider,
+  type Insertable,
+  Kysely,
+  Migrator,
+  type NullableInsertKeys,
+  sql,
+} from 'npm:kysely@^0.25.0';
+export { DenoSqliteDialect } from 'https://gitlab.com/soapbox-pub/kysely-deno-sqlite/-/raw/76748303a45fac64a889cd2b9265c6c9b8ef2e8b/mod.ts';

src/loopback.ts

@@ -1,9 +1,11 @@
 import { Conf } from '@/config.ts';
+import { insertEvent, isLocallyFollowed } from '@/db/events.ts';
+import { findUser } from '@/db/users.ts';
 import { RelayPool } from '@/deps.ts';
 import { trends } from '@/trends.ts';
 import { nostrDate, nostrNow } from '@/utils.ts';
 
-import type { Event } from '@/event.ts';
+import type { SignedEvent } from '@/event.ts';
 
 const relay = new RelayPool([Conf.relay]);
 
@@ -19,13 +21,18 @@ relay.subscribe(
 );
 
 /** Handle events through the loopback pipeline. */
-function handleEvent(event: Event): void {
+async function handleEvent(event: SignedEvent): Promise<void> {
   console.info('loopback event:', event.id);
+
   trackHashtags(event);
+
+  if (await findUser({ pubkey: event.pubkey }) || await isLocallyFollowed(event.pubkey)) {
+    insertEvent(event).catch(console.warn);
+  }
 }
 
 /** Track whenever a hashtag is used, for processing trending tags. */
-function trackHashtags(event: Event): void {
+function trackHashtags(event: SignedEvent): void {
   const date = nostrDate(event.created_at);
 
   const tags = event.tags