fix: event_observer_requests json writes (#1334)

* fix: event_observer_requests json format

* chore: remove tsv from another branch

Author: rafaelcr

src/datastore/pg-write-store.ts

@@ -74,7 +74,7 @@ import {
 } from './helpers';
 import { PgNotifier } from './pg-notifier';
 import { PgStore } from './pg-store';
-import { connectPostgres, PgServer, PgSqlClient } from './connection';
+import { connectPostgres, PgJsonb, PgServer, PgSqlClient } from './connection';
 import { runMigrations } from './migrations';
 import { getPgClientConfig } from './connection-legacy';
 import { isProcessableTokenMetadata } from '../token-metadata/helpers';
@@ -162,7 +162,7 @@ export class PgWriteStore extends PgStore {
     };
   }
 
-  async storeRawEventRequest(eventPath: string, payload: string): Promise<void> {
+  async storeRawEventRequest(eventPath: string, payload: PgJsonb): Promise<void> {
     // To avoid depending on the DB more than once and to allow the query transaction to settle,
     // we'll take the complete insert result and move that to the output TSV file instead of taking
     // only the `id` and performing a `COPY` of that row later.

src/event-stream/event-server.ts

@@ -68,7 +68,7 @@ import {
 
 async function handleRawEventRequest(
   eventPath: string,
-  payload: string,
+  payload: any,
   db: PgWriteStore
 ): Promise<void> {
   await db.storeRawEventRequest(eventPath, payload);
@@ -583,7 +583,7 @@ async function handleNewAttachmentMessage(msg: CoreNodeAttachmentMessage[], db:
 }
 
 interface EventMessageHandler {
-  handleRawEventRequest(eventPath: string, payload: string, db: PgWriteStore): Promise<void> | void;
+  handleRawEventRequest(eventPath: string, payload: any, db: PgWriteStore): Promise<void> | void;
   handleBlockMessage(
     chainId: ChainID,
     msg: CoreNodeBlockMessage,
@@ -607,7 +607,7 @@ function createMessageProcessorQueue(): EventMessageHandler {
   // Create a promise queue so that only one message is handled at a time.
   const processorQueue = new PQueue({ concurrency: 1 });
   const handler: EventMessageHandler = {
-    handleRawEventRequest: (eventPath: string, payload: string, db: PgWriteStore) => {
+    handleRawEventRequest: (eventPath: string, payload: any, db: PgWriteStore) => {
       return processorQueue
         .add(() => handleRawEventRequest(eventPath, payload, db))
         .catch(e => {
@@ -733,7 +733,7 @@ export async function startEventServer(opts: {
     asyncHandler(async (req, res, next) => {
       const eventPath = req.path;
       let payload = JSON.stringify(req.body);
-      await messageHandler.handleRawEventRequest(eventPath, payload, db);
+      await messageHandler.handleRawEventRequest(eventPath, req.body, db);
       if (logger.isDebugEnabled()) {
         // Skip logging massive event payloads, this _should_ only exclude the genesis block payload which is ~80 MB.
         if (payload.length > 10_000_000) {