fix: keep a table of latest stacks node event timestamps (hirosystems#2266)

Author: rafaelcr

migrations/1745877592337_event-observer-timestamps.js

@@ -0,0 +1,20 @@
+/* eslint-disable camelcase */
+
+exports.shorthands = undefined;
+
+exports.up = pgm => {
+  pgm.createTable('event_observer_timestamps', {
+    event_path: {
+      type: 'string',
+      primaryKey: true,
+    },
+    id: {
+      type: 'bigint',
+      notNull: true,
+    },
+    receive_timestamp: {
+      type: 'timestamptz',
+      notNull: true,
+    },
+  });
+};

src/datastore/pg-store.ts

@@ -4684,9 +4684,7 @@ export class PgStore extends BasePgStore {
   /// by event type.
   async getLastStacksNodeEventTimestamps() {
     return await this.sql<{ event_path: string; receive_timestamp: Date }[]>`
-      SELECT DISTINCT ON (event_path) event_path, receive_timestamp
-      FROM event_observer_requests
-      ORDER BY event_path, receive_timestamp DESC
+      SELECT event_path, receive_timestamp FROM event_observer_timestamps
     `;
   }
 }

src/datastore/pg-write-store.ts

@@ -178,10 +178,20 @@ export class PgWriteStore extends PgStore {
           receive_timestamp: string;
           event_path: string;
         }[]
-      >`INSERT INTO event_observer_requests(
-          event_path, payload
-        ) values(${eventPath}, ${payload})
-        RETURNING id, receive_timestamp::text, event_path
+      >`WITH inserted AS (
+          INSERT INTO event_observer_requests(
+            event_path, payload
+          ) values(${eventPath}, ${payload})
+          RETURNING id, receive_timestamp, event_path
+        ),
+        latest AS (
+          INSERT INTO event_observer_timestamps (id, receive_timestamp, event_path)
+          (SELECT id, receive_timestamp, event_path FROM inserted)
+          ON CONFLICT (event_path) DO UPDATE SET
+            id = EXCLUDED.id,
+            receive_timestamp = EXCLUDED.receive_timestamp
+        )
+        SELECT id, receive_timestamp::text, event_path FROM inserted
       `;
       if (insertResult.length !== 1) {
         throw new Error(