Merge branch 'stats-inflation' into 'main'

Fix stat inflation

See merge request soapbox-pub/ditto!487

src/pipeline.ts

@@ -5,6 +5,7 @@ import { LRUCache } from 'lru-cache';
 import { z } from 'zod';
 
 import { Conf } from '@/config.ts';
+import { DittoTables } from '@/db/DittoTables.ts';
 import { DittoEvent } from '@/interfaces/DittoEvent.ts';
 import { pipelineEventsCounter, policyEventsCounter } from '@/metrics.ts';
 import { RelayError } from '@/RelayError.ts';
@@ -14,11 +15,11 @@ import { Storages } from '@/storages.ts';
 import { eventAge, parseNip05, Time } from '@/utils.ts';
 import { policyWorker } from '@/workers/policy.ts';
 import { verifyEventWorker } from '@/workers/verify.ts';
+import { getAmount } from '@/utils/bolt11.ts';
 import { nip05Cache } from '@/utils/nip05.ts';
+import { purifyEvent } from '@/utils/purify.ts';
 import { updateStats } from '@/utils/stats.ts';
 import { getTagSet } from '@/utils/tags.ts';
-import { getAmount } from '@/utils/bolt11.ts';
-import { DittoTables } from '@/db/DittoTables.ts';
 
 const debug = Debug('ditto:pipeline');
 
@@ -55,7 +56,7 @@ async function handleEvent(event: DittoEvent, signal: AbortSignal): Promise<void
   const kysely = await Storages.kysely();
 
   await Promise.all([
-    storeEvent(event, signal),
+    storeEvent(purifyEvent(event), signal),
     handleZaps(kysely, event),
     parseMetadata(event, signal),
     generateSetEvents(event),
@@ -117,10 +118,11 @@ async function hydrateEvent(event: DittoEvent, signal: AbortSignal): Promise<voi
 async function storeEvent(event: DittoEvent, signal?: AbortSignal): Promise<undefined> {
   if (NKinds.ephemeral(event.kind)) return;
   const store = await Storages.db();
-  const kysely = await Storages.kysely();
 
-  await updateStats({ event, store, kysely }).catch(debug);
-  await store.event(event, { signal });
+  await store.transaction(async (store, kysely) => {
+    await updateStats({ event, store, kysely });
+    await store.event(event, { signal });
+  });
 }
 
 /** Parse kind 0 metadata and track indexes in the database. */

src/storages.ts

@@ -12,7 +12,7 @@ import { seedZapSplits } from '@/utils/zap-split.ts';
 
 export class Storages {
   private static _db: Promise<EventsDB> | undefined;
-  private static _database: DittoDatabase | undefined;
+  private static _database: Promise<DittoDatabase> | undefined;
   private static _admin: Promise<AdminStore> | undefined;
   private static _client: Promise<NPool> | undefined;
   private static _pubsub: Promise<InternalRelay> | undefined;
@@ -20,8 +20,11 @@ export class Storages {
 
   public static async database(): Promise<DittoDatabase> {
     if (!this._database) {
-      this._database = DittoDB.create(Conf.databaseUrl, { poolSize: Conf.pg.poolSize });
-      await DittoDB.migrate(this._database.kysely);
+      this._database = (async () => {
+        const db = DittoDB.create(Conf.databaseUrl, { poolSize: Conf.pg.poolSize });
+        await DittoDB.migrate(db.kysely);
+        return db;
+      })();
     }
     return this._database;
   }
@@ -35,7 +38,7 @@ export class Storages {
   public static async db(): Promise<EventsDB> {
     if (!this._db) {
       this._db = (async () => {
-        const { kysely } = await this.database();
+        const kysely = await this.kysely();
         const store = new EventsDB({ kysely, pubkey: Conf.pubkey, timeout: Conf.db.timeouts.default });
         await seedZapSplits(store);
         return store;

src/storages/EventsDB.ts

@@ -316,6 +316,10 @@ class EventsDB implements NStore {
 
     return filters;
   }
+
+  async transaction(callback: (store: NPostgres, kysely: Kysely<DittoTables>) => Promise<void>): Promise<void> {
+    return this.store.transaction((store, kysely) => callback(store, kysely as unknown as Kysely<DittoTables>));
+  }
 }
 
 export { EventsDB };