Merge branch 'ndatabase' into 'main'

EventsDB: make it a simple wrapper around NDatabase

See merge request soapbox-pub/ditto!258

src/controllers/api/admin.ts

@@ -2,11 +2,12 @@ import { z } from 'zod';
 
 import { type AppController } from '@/app.ts';
 import { Conf } from '@/config.ts';
+import { DittoEvent } from '@/interfaces/DittoEvent.ts';
 import { booleanParamSchema } from '@/schema.ts';
 import { Storages } from '@/storages.ts';
-import { renderAdminAccount } from '@/views/mastodon/admin-accounts.ts';
-import { paginated, paginationSchema, parseBody, updateListAdminEvent } from '@/utils/api.ts';
 import { addTag } from '@/tags.ts';
+import { paginated, paginationSchema, parseBody, updateListAdminEvent } from '@/utils/api.ts';
+import { renderAdminAccount } from '@/views/mastodon/admin-accounts.ts';
 
 const adminAccountQuerySchema = z.object({
   local: booleanParamSchema.optional(),
@@ -49,7 +50,7 @@ const adminAccountsController: AppController = async (c) => {
 
   for (const event of events) {
     const d = event.tags.find(([name]) => name === 'd')?.[1];
-    event.d_author = authors.find((author) => author.pubkey === d);
+    (event as DittoEvent).d_author = authors.find((author) => author.pubkey === d);
   }
 
   const accounts = await Promise.all(

src/db/DittoTables.ts

@@ -30,7 +30,6 @@ interface EventRow {
   created_at: number;
   tags: string;
   sig: string;
-  deleted_at: number | null;
 }
 
 interface EventFTSRow {

src/db/migrations/020_drop_deleted_at.ts

@@ -0,0 +1,10 @@
+import { Kysely } from 'kysely';
+
+export async function up(db: Kysely<any>): Promise<void> {
+  await db.deleteFrom('nostr_events').where('deleted_at', 'is not', null).execute();
+  await db.schema.alterTable('nostr_events').dropColumn('deleted_at').execute();
+}
+
+export async function down(db: Kysely<any>): Promise<void> {
+  await db.schema.alterTable('nostr_events').addColumn('deleted_at', 'integer').execute();
+}

src/kinds.ts

@@ -1,46 +0,0 @@
-/** Events are **regular**, which means they're all expected to be stored by relays. */
-function isRegularKind(kind: number) {
-  return (1000 <= kind && kind < 10000) || [1, 2, 4, 5, 6, 7, 8, 16, 40, 41, 42, 43, 44].includes(kind);
-}
-
-/** Events are **replaceable**, which means that, for each combination of `pubkey` and `kind`, only the latest event is expected to (SHOULD) be stored by relays, older versions are expected to be discarded. */
-function isReplaceableKind(kind: number) {
-  return (10000 <= kind && kind < 20000) || [0, 3].includes(kind);
-}
-
-/** Events are **ephemeral**, which means they are not expected to be stored by relays. */
-function isEphemeralKind(kind: number) {
-  return 20000 <= kind && kind < 30000;
-}
-
-/** Events are **parameterized replaceable**, which means that, for each combination of `pubkey`, `kind` and the `d` tag, only the latest event is expected to be stored by relays, older versions are expected to be discarded. */
-function isParameterizedReplaceableKind(kind: number) {
-  return 30000 <= kind && kind < 40000;
-}
-
-/** These events are only valid if published by the server keypair. */
-function isDittoInternalKind(kind: number) {
-  return kind === 30361;
-}
-
-/** Classification of the event kind. */
-type KindClassification = 'regular' | 'replaceable' | 'ephemeral' | 'parameterized' | 'unknown';
-
-/** Determine the classification of this kind of event if known, or `unknown`. */
-function classifyKind(kind: number): KindClassification {
-  if (isRegularKind(kind)) return 'regular';
-  if (isReplaceableKind(kind)) return 'replaceable';
-  if (isEphemeralKind(kind)) return 'ephemeral';
-  if (isParameterizedReplaceableKind(kind)) return 'parameterized';
-  return 'unknown';
-}
-
-export {
-  classifyKind,
-  isDittoInternalKind,
-  isEphemeralKind,
-  isParameterizedReplaceableKind,
-  isRegularKind,
-  isReplaceableKind,
-  type KindClassification,
-};

src/pipeline.ts

@@ -1,4 +1,4 @@
-import { NostrEvent, NPolicy, NSchema as n } from '@nostrify/nostrify';
+import { NKinds, NostrEvent, NPolicy, NSchema as n } from '@nostrify/nostrify';
 import { LNURL } from '@nostrify/nostrify/ln';
 import { PipePolicy } from '@nostrify/nostrify/policies';
 import Debug from '@soapbox/stickynotes/debug';
@@ -8,7 +8,6 @@ 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 { isEphemeralKind } from '@/kinds.ts';
 import { DVM } from '@/pipeline/DVM.ts';
 import { RelayError } from '@/RelayError.ts';
 import { updateStats } from '@/stats.ts';
@@ -103,7 +102,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> {
-  if (isEphemeralKind(event.kind)) return;
+  if (NKinds.ephemeral(event.kind)) return;
   const store = await Storages.db();
 
   const [deletion] = await store.query(

src/storages/events-db.ts

@@ -1,31 +1,35 @@
-import { NIP50, NostrEvent, NostrFilter, NSchema as n, NStore } from '@nostrify/nostrify';
+// deno-lint-ignore-file require-await
-import Debug from '@soapbox/stickynotes/debug';
+
-import { Kysely, type SelectQueryBuilder } from 'kysely';
+import { NDatabase, NIP50, NKinds, NostrEvent, NostrFilter, NSchema as n, NStore } from '@nostrify/nostrify';
-import { sortEvents } from 'nostr-tools';
+import { Stickynotes } from '@soapbox/stickynotes';
+import { Kysely } from 'kysely';
 
 import { Conf } from '@/config.ts';
 import { DittoTables } from '@/db/DittoTables.ts';
 import { normalizeFilters } from '@/filter.ts';
-import { DittoEvent } from '@/interfaces/DittoEvent.ts';
-import { isDittoInternalKind, isParameterizedReplaceableKind, isReplaceableKind } from '@/kinds.ts';
 import { purifyEvent } from '@/storages/hydrate.ts';
 import { isNostrId, isURL } from '@/utils.ts';
 import { abortError } from '@/utils/abort.ts';
 
 /** Function to decide whether or not to index a tag. */
 type TagCondition = ({ event, count, value }: {
-  event: DittoEvent;
+  event: NostrEvent;
   count: number;
   value: string;
 }) => boolean;
 
+/** SQLite database storage adapter for Nostr events. */
+class EventsDB implements NStore {
+  private store: NDatabase;
+  private console = new Stickynotes('ditto:db:events');
+
   /** Conditions for when to index certain tags. */
-const tagConditions: Record<string, TagCondition> = {
+  static tagConditions: Record<string, TagCondition> = {
-  'd': ({ event, count }) => count === 0 && isParameterizedReplaceableKind(event.kind),
+    'd': ({ event, count }) => count === 0 && NKinds.parameterizedReplaceable(event.kind),
-  'e': ({ event, count, value }) => ((event.user && event.kind === 10003) || count < 15) && isNostrId(value),
+    'e': ({ event, count, value }) => ((event.kind === 10003) || count < 15) && isNostrId(value),
     'L': ({ event, count }) => event.kind === 1985 || count === 0,
     'l': ({ event, count }) => event.kind === 1985 || count === 0,
-  'media': ({ event, count, value }) => (event.user || count < 4) && isURL(value),
+    'media': ({ count, value }) => (count < 4) && isURL(value),
     'P': ({ count, value }) => count === 0 && isNostrId(value),
     'p': ({ event, count, value }) => (count < 15 || event.kind === 3) && isNostrId(value),
     'proxy': ({ count, value }) => count === 0 && isURL(value),
@@ -35,202 +39,110 @@ const tagConditions: Record<string, TagCondition> = {
     'role': ({ event, count }) => event.kind === 30361 && count === 0,
   };
 
-type EventQuery = SelectQueryBuilder<DittoTables, 'nostr_events', {
+  constructor(private kysely: Kysely<DittoTables>) {
-  id: string;
+    this.store = new NDatabase(kysely, {
-  tags: string;
+      fts5: Conf.databaseUrl.protocol === 'sqlite:',
-  kind: number;
+      indexTags: EventsDB.indexTags,
-  pubkey: string;
+      searchText: EventsDB.searchText,
-  content: string;
+    });
-  created_at: number;
-  sig: string;
-  stats_replies_count?: number;
-  stats_reposts_count?: number;
-  stats_reactions_count?: number;
-  author_id?: string;
-  author_tags?: string;
-  author_kind?: number;
-  author_pubkey?: string;
-  author_content?: string;
-  author_created_at?: number;
-  author_sig?: string;
-  author_stats_followers_count?: number;
-  author_stats_following_count?: number;
-  author_stats_notes_count?: number;
-}>;
-
-/** SQLite database storage adapter for Nostr events. */
-class EventsDB implements NStore {
-  #db: Kysely<DittoTables>;
-  #debug = Debug('ditto:db:events');
-  private protocol = Conf.databaseUrl.protocol;
-
-  constructor(db: Kysely<DittoTables>) {
-    this.#db = db;
   }
 
   /** Insert an event (and its tags) into the database. */
   async event(event: NostrEvent, _opts?: { signal?: AbortSignal }): Promise<void> {
     event = purifyEvent(event);
-    this.#debug('EVENT', JSON.stringify(event));
+    this.console.debug('EVENT', JSON.stringify(event));
-
+    return this.store.event(event);
-    if (isDittoInternalKind(event.kind) && event.pubkey !== Conf.pubkey) {
-      throw new Error('Internal events can only be stored by the server keypair');
   }
 
-    return await this.#db.transaction().execute(async (trx) => {
+  /** Get events for filters from the database. */
-      /** Insert the event into the database. */
+  async query(filters: NostrFilter[], opts: { signal?: AbortSignal; limit?: number } = {}): Promise<NostrEvent[]> {
-      async function addEvent() {
+    filters = await this.expandFilters(filters);
-        await trx.insertInto('nostr_events')
+
-          .values({ ...event, tags: JSON.stringify(event.tags) })
+    if (opts.signal?.aborted) return Promise.resolve([]);
-          .execute();
+    if (!filters.length) return Promise.resolve([]);
+
+    this.console.debug('REQ', JSON.stringify(filters));
+
+    return this.store.query(filters, opts);
   }
 
-      const protocol = this.protocol;
+  /** Delete events based on filters from the database. */
-      /** Add search data to the FTS table. */
+  async remove(filters: NostrFilter[], _opts?: { signal?: AbortSignal }): Promise<void> {
-      async function indexSearch() {
+    if (!filters.length) return Promise.resolve();
-        if (protocol !== 'sqlite:') return;
+    this.console.debug('DELETE', JSON.stringify(filters));
-        const searchContent = buildSearchContent(event);
+
-        if (!searchContent) return;
+    return this.store.remove(filters);
-        await trx.insertInto('nostr_fts5')
-          .values({ event_id: event.id, content: searchContent.substring(0, 1000) })
-          .execute();
   }
 
-      /** Index event tags depending on the conditions defined above. */
+  /** Get number of events that would be returned by filters. */
-      async function indexTags() {
+  async count(
-        const tags = filterIndexableTags(event);
+    filters: NostrFilter[],
-        const rows = tags.map(([name, value]) => ({ event_id: event.id, name, value }));
+    opts: { signal?: AbortSignal } = {},
+  ): Promise<{ count: number; approximate: boolean }> {
+    if (opts.signal?.aborted) return Promise.reject(abortError());
+    if (!filters.length) return Promise.resolve({ count: 0, approximate: false });
 
-        if (!tags.length) return;
+    this.console.debug('COUNT', JSON.stringify(filters));
-        await trx.insertInto('nostr_tags')
+
-          .values(rows)
+    return this.store.count(filters);
-          .execute();
   }
 
-      if (isReplaceableKind(event.kind)) {
+  /** Return only the tags that should be indexed. */
-        const prevEvents = await this.getFilterQuery(trx, { kinds: [event.kind], authors: [event.pubkey] }).execute();
+  static indexTags(event: NostrEvent): string[][] {
-        for (const prevEvent of prevEvents) {
+    const tagCounts: Record<string, number> = {};
-          if (prevEvent.created_at >= event.created_at) {
+
-            throw new Error('Cannot replace an event with an older event');
+    function getCount(name: string) {
-          }
+      return tagCounts[name] || 0;
-        }
-        await this.deleteEventsTrx(trx, [{ kinds: [event.kind], authors: [event.pubkey] }]);
     }
 
-      if (isParameterizedReplaceableKind(event.kind)) {
+    function incrementCount(name: string) {
-        const d = event.tags.find(([tag]) => tag === 'd')?.[1];
+      tagCounts[name] = getCount(name) + 1;
-        if (d) {
-          const prevEvents = await this.getFilterQuery(trx, { kinds: [event.kind], authors: [event.pubkey], '#d': [d] })
-            .execute();
-          for (const prevEvent of prevEvents) {
-            if (prevEvent.created_at >= event.created_at) {
-              throw new Error('Cannot replace an event with an older event');
-            }
-          }
-          await this.deleteEventsTrx(trx, [{ kinds: [event.kind], authors: [event.pubkey], '#d': [d] }]);
-        }
     }
 
-      // Run the queries.
+    function checkCondition(name: string, value: string, condition: TagCondition) {
-      await Promise.all([
+      return condition({
-        addEvent(),
+        event,
-        indexTags(),
+        count: getCount(name),
-        indexSearch(),
+        value,
-      ]);
-    }).catch((error) => {
-      // Don't throw for duplicate events.
-      if (error.message.includes('UNIQUE constraint failed')) {
-        return;
-      } else {
-        throw error;
-      }
       });
     }
 
-  /** Build the query for a filter. */
+    return event.tags.reduce<string[][]>((results, tag) => {
-  getFilterQuery(db: Kysely<DittoTables>, filter: NostrFilter): EventQuery {
+      const [name, value] = tag;
-    let query = db
+      const condition = EventsDB.tagConditions[name] as TagCondition | undefined;
-      .selectFrom('nostr_events')
-      .select([
-        'nostr_events.id',
-        'nostr_events.kind',
-        'nostr_events.pubkey',
-        'nostr_events.content',
-        'nostr_events.tags',
-        'nostr_events.created_at',
-        'nostr_events.sig',
-      ])
-      .where('nostr_events.deleted_at', 'is', null);
 
-    /** Whether we are querying for replaceable events by author. */
+      if (value && condition && value.length < 200 && checkCondition(name, value, condition)) {
-    const isAddrQuery = filter.authors &&
+        results.push(tag);
-      filter.kinds &&
-      filter.kinds.every((kind) => isReplaceableKind(kind) || isParameterizedReplaceableKind(kind));
-
-    // Avoid ORDER BY when querying for replaceable events by author.
-    if (!isAddrQuery) {
-      query = query.orderBy('nostr_events.created_at', 'desc');
       }
 
-    for (const [key, value] of Object.entries(filter)) {
+      incrementCount(name);
-      if (value === undefined) continue;
+      return results;
+    }, []);
+  }
 
-      switch (key as keyof NostrFilter) {
+  /** Build a search index from the event. */
-        case 'ids':
+  static searchText(event: NostrEvent): string {
-          query = query.where('nostr_events.id', 'in', filter.ids!);
+    switch (event.kind) {
-          break;
+      case 0:
-        case 'kinds':
+        return EventsDB.buildUserSearchContent(event);
-          query = query.where('nostr_events.kind', 'in', filter.kinds!);
+      case 1:
-          break;
+        return event.content;
-        case 'authors':
+      case 30009:
-          query = query.where('nostr_events.pubkey', 'in', filter.authors!);
+        return EventsDB.buildTagsSearchContent(event.tags.filter(([t]) => t !== 'alt'));
-          break;
+      default:
-        case 'since':
+        return '';
-          query = query.where('nostr_events.created_at', '>=', filter.since!);
-          break;
-        case 'until':
-          query = query.where('nostr_events.created_at', '<=', filter.until!);
-          break;
-        case 'limit':
-          query = query.limit(filter.limit!);
-          break;
     }
   }
 
-    const joinedQuery = query.leftJoin('nostr_tags', 'nostr_tags.event_id', 'nostr_events.id');
+  /** Build search content for a user. */
-
+  static buildUserSearchContent(event: NostrEvent): string {
-    for (const [key, value] of Object.entries(filter)) {
+    const { name, nip05, about } = n.json().pipe(n.metadata()).catch({}).parse(event.content);
-      if (key.startsWith('#') && Array.isArray(value)) {
+    return [name, nip05, about].filter(Boolean).join('\n');
-        const name = key.replace(/^#/, '');
-        query = joinedQuery
-          .where('nostr_tags.name', '=', name)
-          .where('nostr_tags.value', 'in', value);
-      }
   }
 
-    if (filter.search && this.protocol === 'sqlite:') {
+  /** Build search content from tag values. */
-      query = query
+  static buildTagsSearchContent(tags: string[][]): string {
-        .innerJoin('nostr_fts5', 'nostr_fts5.event_id', 'nostr_events.id')
+    return tags.map(([_tag, value]) => value).join('\n');
-        .where('nostr_fts5.content', 'match', JSON.stringify(filter.search));
-    }
-
-    return query;
-  }
-
-  /** Combine filter queries into a single union query. */
-  getEventsQuery(filters: NostrFilter[]) {
-    return filters
-      .map((filter) => this.#db.selectFrom(() => this.getFilterQuery(this.#db, filter).as('events')).selectAll())
-      .reduce((result, query) => result.unionAll(query));
-  }
-
-  /** Query to get user events, joined by tags. */
-  usersQuery() {
-    return this.getFilterQuery(this.#db, { kinds: [30361], authors: [Conf.pubkey] })
-      .leftJoin('nostr_tags', 'nostr_tags.event_id', 'nostr_events.id')
-      .where('nostr_tags.name', '=', 'd')
-      .select('nostr_tags.value as d_tag')
-      .as('users');
   }
 
   /** Converts filters to more performant, simpler filters that are better for SQLite. */
@@ -244,7 +156,7 @@ class EventsDB implements NStore {
         ) as { key: 'domain'; value: string } | undefined)?.value;
 
         if (domain) {
-          const query = this.#db
+          const query = this.kysely
             .selectFrom('pubkey_domains')
             .select('pubkey')
             .where('domain', '=', domain);
@@ -268,166 +180,6 @@ class EventsDB implements NStore {
 
     return normalizeFilters(filters); // Improves performance of `{ kinds: [0], authors: ['...'] }` queries.
   }
-
-  /** Get events for filters from the database. */
-  async query(filters: NostrFilter[], opts: { signal?: AbortSignal; limit?: number } = {}): Promise<DittoEvent[]> {
-    filters = await this.expandFilters(filters);
-
-    if (opts.signal?.aborted) return Promise.resolve([]);
-    if (!filters.length) return Promise.resolve([]);
-
-    this.#debug('REQ', JSON.stringify(filters));
-    let query = this.getEventsQuery(filters);
-
-    if (typeof opts.limit === 'number') {
-      query = query.limit(opts.limit);
-    }
-
-    const events = (await query.execute()).map((row) => {
-      const event: DittoEvent = {
-        id: row.id,
-        kind: row.kind,
-        pubkey: row.pubkey,
-        content: row.content,
-        created_at: row.created_at,
-        tags: JSON.parse(row.tags),
-        sig: row.sig,
-      };
-
-      if (row.author_id) {
-        event.author = {
-          id: row.author_id,
-          kind: row.author_kind! as 0,
-          pubkey: row.author_pubkey!,
-          content: row.author_content!,
-          created_at: row.author_created_at!,
-          tags: JSON.parse(row.author_tags!),
-          sig: row.author_sig!,
-        };
-      }
-
-      if (typeof row.author_stats_followers_count === 'number') {
-        event.author_stats = {
-          followers_count: row.author_stats_followers_count,
-          following_count: row.author_stats_following_count!,
-          notes_count: row.author_stats_notes_count!,
-        };
-      }
-
-      if (typeof row.stats_replies_count === 'number') {
-        event.event_stats = {
-          replies_count: row.stats_replies_count,
-          reposts_count: row.stats_reposts_count!,
-          reactions_count: row.stats_reactions_count!,
-        };
-      }
-
-      return event;
-    });
-
-    return sortEvents(events);
-  }
-
-  /** Delete events from each table. Should be run in a transaction! */
-  async deleteEventsTrx(db: Kysely<DittoTables>, filters: NostrFilter[]) {
-    if (!filters.length) return Promise.resolve();
-    this.#debug('DELETE', JSON.stringify(filters));
-
-    const query = this.getEventsQuery(filters).clearSelect().select('id');
-
-    return await db.updateTable('nostr_events')
-      .where('id', 'in', () => query)
-      .set({ deleted_at: Math.floor(Date.now() / 1000) })
-      .execute();
-  }
-
-  /** Delete events based on filters from the database. */
-  async remove(filters: NostrFilter[], _opts?: { signal?: AbortSignal }): Promise<void> {
-    if (!filters.length) return Promise.resolve();
-    this.#debug('DELETE', JSON.stringify(filters));
-
-    await this.#db.transaction().execute((trx) => this.deleteEventsTrx(trx, filters));
-  }
-
-  /** Get number of events that would be returned by filters. */
-  async count(
-    filters: NostrFilter[],
-    opts: { signal?: AbortSignal } = {},
-  ): Promise<{ count: number; approximate: boolean }> {
-    if (opts.signal?.aborted) return Promise.reject(abortError());
-    if (!filters.length) return Promise.resolve({ count: 0, approximate: false });
-
-    this.#debug('COUNT', JSON.stringify(filters));
-    const query = this.getEventsQuery(filters);
-
-    const [{ count }] = await query
-      .clearSelect()
-      .select((eb) => eb.fn.count('id').as('count'))
-      .execute();
-
-    return {
-      count: Number(count),
-      approximate: false,
-    };
-  }
-}
-
-/** Return only the tags that should be indexed. */
-function filterIndexableTags(event: DittoEvent): string[][] {
-  const tagCounts: Record<string, number> = {};
-
-  function getCount(name: string) {
-    return tagCounts[name] || 0;
-  }
-
-  function incrementCount(name: string) {
-    tagCounts[name] = getCount(name) + 1;
-  }
-
-  function checkCondition(name: string, value: string, condition: TagCondition) {
-    return condition({
-      event,
-      count: getCount(name),
-      value,
-    });
-  }
-
-  return event.tags.reduce<string[][]>((results, tag) => {
-    const [name, value] = tag;
-    const condition = tagConditions[name] as TagCondition | undefined;
-
-    if (value && condition && value.length < 200 && checkCondition(name, value, condition)) {
-      results.push(tag);
-    }
-
-    incrementCount(name);
-    return results;
-  }, []);
-}
-
-/** Build a search index from the event. */
-function buildSearchContent(event: NostrEvent): string {
-  switch (event.kind) {
-    case 0:
-      return buildUserSearchContent(event);
-    case 1:
-      return event.content;
-    case 30009:
-      return buildTagsSearchContent(event.tags.filter(([t]) => t !== 'alt'));
-    default:
-      return '';
-  }
-}
-
-/** Build search content for a user. */
-function buildUserSearchContent(event: NostrEvent): string {
-  const { name, nip05, about } = n.json().pipe(n.metadata()).catch({}).parse(event.content);
-  return [name, nip05, about].filter(Boolean).join('\n');
-}
-
-/** Build search content from tag values. */
-function buildTagsSearchContent(tags: string[][]): string {
-  return tags.map(([_tag, value]) => value).join('\n');
 }
 
 export { EventsDB };