Merge branch 'fix-follow-people' into 'main'

fix: follow people

Closes #258

See merge request soapbox-pub/ditto!572

src/db/migrations/041_pg_notify_id_only.ts

@@ -0,0 +1,27 @@
+import { Kysely, sql } from 'kysely';
+
+export async function up(db: Kysely<any>): Promise<void> {
+  await sql`DROP TRIGGER IF EXISTS nostr_event_trigger ON nostr_events`.execute(db);
+
+  await sql`
+    CREATE OR REPLACE FUNCTION notify_nostr_event()
+    RETURNS TRIGGER AS $$
+    BEGIN
+        PERFORM pg_notify('nostr_event', NEW.id::text);
+
+        RETURN NEW;
+    END;
+    $$ LANGUAGE plpgsql;
+  `.execute(db);
+
+  await sql`
+    CREATE TRIGGER nostr_event_trigger
+    AFTER INSERT OR UPDATE ON nostr_events
+    FOR EACH ROW EXECUTE FUNCTION notify_nostr_event()
+  `.execute(db);
+}
+
+export async function down(db: Kysely<any>): Promise<void> {
+  await sql`DROP TRIGGER nostr_event_trigger ON nostr_events`.execute(db);
+  await sql`DROP FUNCTION notify_nostr_event()`.execute(db);
+}

src/notify.ts

@@ -1,5 +1,6 @@
 import { Semaphore } from '@lambdalisue/async';
 
+import { Conf } from '@/config.ts';
 import * as pipeline from '@/pipeline.ts';
 import { Storages } from '@/storages.ts';
 
@@ -7,12 +8,18 @@ const sem = new Semaphore(1);
 
 export async function startNotify(): Promise<void> {
   const { listen } = await Storages.database();
+  const store = await Storages.db();
 
   listen('nostr_event', (payload) => {
     sem.lock(async () => {
       try {
-        const event = JSON.parse(payload);
-        await pipeline.handleEvent(event, AbortSignal.timeout(5000));
+        const id = payload;
+        const timeout = Conf.db.timeouts.default;
+
+        const [event] = await store.query([{ ids: [id], limit: 1 }], { signal: AbortSignal.timeout(timeout) });
+        if (event) {
+          await pipeline.handleEvent(event, AbortSignal.timeout(timeout));
+        }
       } catch (e) {
         console.warn(e);
       }