Fix not being able to log in for chrissakes

packages/ditto/pipeline.ts

@@ -77,21 +77,24 @@ async function handleEvent(event: DittoEvent, opts: PipelineOpts): Promise<void>
   // NIP-46 events get special treatment.
   // They are exempt from policies and other side-effects, and should be streamed out immediately.
   // If streaming fails, an error should be returned.
-  if (event.kind !== 24133) {
+  if (event.kind === 24133) {
-    // Ensure the event doesn't violate the policy.
+    const store = await Storages.db();
-    if (event.pubkey !== Conf.pubkey) {
+    await store.event(event, { signal: opts.signal });
-      await policyFilter(event, opts.signal);
+  }
-    }
 
-    // Prepare the event for additional checks.
+  // Ensure the event doesn't violate the policy.
-    // FIXME: This is kind of hacky. Should be reorganized to fetch only what's needed for each stage.
+  if (event.pubkey !== Conf.pubkey) {
-    await hydrateEvent(event, opts.signal);
+    await policyFilter(event, opts.signal);
+  }
 
-    // Ensure that the author is not banned.
+  // Prepare the event for additional checks.
-    const n = getTagSet(event.user?.tags ?? [], 'n');
+  // FIXME: This is kind of hacky. Should be reorganized to fetch only what's needed for each stage.
-    if (n.has('disabled')) {
+  await hydrateEvent(event, opts.signal);
-      throw new RelayError('blocked', 'author is blocked');
+
-    }
+  // Ensure that the author is not banned.
+  const n = getTagSet(event.user?.tags ?? [], 'n');
+  if (n.has('disabled')) {
+    throw new RelayError('blocked', 'author is blocked');
   }
 
   const kysely = await Storages.kysely();

packages/ditto/storages/DittoPgStore.ts

@@ -126,10 +126,6 @@ export class DittoPgStore extends NPostgres {
   override async event(event: NostrEvent, opts: { signal?: AbortSignal; timeout?: number } = {}): Promise<void> {
     event = purifyEvent(event);
 
-    if (this.opts.notify) {
-      this.encounters.set(event.id, true);
-    }
-
     logi({ level: 'debug', ns: 'ditto.event', source: 'db', id: event.id, kind: event.kind });
     dbEventsCounter.inc({ kind: event.kind });
 
@@ -137,6 +133,10 @@ export class DittoPgStore extends NPostgres {
       return await this.fulfill(event);
     }
 
+    if (this.opts.notify) {
+      this.encounters.set(event.id, true);
+    }
+
     if (await this.isDeletedAdmin(event)) {
       throw new RelayError('blocked', 'event deleted by admin');
     }
@@ -590,8 +590,28 @@ export class DittoPgStore extends NPostgres {
     return filters;
   }
 
-  // deno-lint-ignore no-explicit-any
+  /** Execute the callback in a new transaction, unless the Kysely instance is already a transaction. */
-  override async transaction(callback: (store: NPostgres, kysely: Kysely<any>) => Promise<void>): Promise<void> {
+  private static override async trx<T = unknown>(
-    return super.transaction((store, kysely) => callback(store, kysely as unknown as Kysely<DittoTables>));
+    db: Kysely<DittoTables>,
+    callback: (trx: Kysely<DittoTables>) => Promise<T>,
+  ): Promise<T> {
+    if (db.isTransaction) {
+      return await callback(db);
+    } else {
+      return await db.transaction().execute((trx) => callback(trx));
+    }
+  }
+
+  /** Execute NPostgres functions in a transaction. */
+  // @ts-ignore gg
+  override async transaction(
+    callback: (store: DittoPgStore, kysely: Kysely<DittoTables>) => Promise<void>,
+  ): Promise<void> {
+    const { db } = this.opts;
+
+    await DittoPgStore.trx(db.kysely, async (trx) => {
+      const store = new DittoPgStore({ ...this.opts, db: { ...db, kysely: trx }, notify: false });
+      await callback(store, trx);
+    });
   }
 }