imp(workers): move timestamp out of event payload (#425)

* chore(): update types deps

* imp(): update worker task types

* imp(): update worker handlers

* feat(): migration for moving timestamp out of payload

* fix(): tests fixed

* imp(): naming, worker tasks (default, sentry, js, grouper)

* imp(): imp worker tasks

* imp(sentry): types cast

* fix types

* chore(): clean up

* fix(): tests fixed

* fix(): tests minor fixes

* deps: update types dependency

* fix(): tests minor fixes

* fix(): types fix

* chore(): get rid of EventDataAccepted

* chore: lint remove redundant imports

* fix(): sentry tests fixed

* chore: fix lint

* imp(): change migration to use bulkWrite instead of cursor

* imp(): js worker event type improved

* chore(): fix lint

* imp(): add convertor script

* chore: lint fix

* imp: convertor

* chore(): converter script improvements

---------

Co-authored-by: Peter Savchenko <specc.dev@gmail.com>

Author: e11sy

convertors/move-timestamp-out-of-payload.js

@@ -0,0 +1,231 @@
+require('dotenv').config();
+const { MongoClient } = require('mongodb');
+
+/**
+ * Limit for one time documents selection, used to reduce the load on the database
+ */
+const documentsSelectionLimit = 10000;
+
+/**
+ * @param db - mongo db instance
+ * @param collectionName - name of the collection to be updated
+ */
+async function movePayloadTimestampToEventLevel(db, collectionName) {
+  const collection = db.collection(collectionName);
+
+  const docsToUpdate = collection.find(
+    { timestamp: { $exists: false } },
+    { projection: { _id: 1, 'payload.timestamp': 1 } }
+  ).limit(documentsSelectionLimit);
+
+  const batchedOps = [];
+
+  let currentCount = 0;
+
+  for await (const doc of docsToUpdate) {
+    process.stdout.write(`\r${currentCount} documents added to batch`);
+
+    if (!doc.payload.timestamp) {
+      continue;
+    }
+
+    batchedOps.push({
+      updateOne: {
+        filter: { _id: doc._id },
+        update: {
+          $set: { timestamp: Number(doc.payload.timestamp)},
+          $unset: {'payload.timestamp': ''},
+        }
+      }
+    })
+
+    currentCount++;
+  }
+
+  if (currentCount > 0) {
+    await collection.bulkWrite(batchedOps);
+  }
+
+  return currentCount
+}
+/**
+ * @param db - mongo db instance
+ * @param repetitionCollectionName - repetitions collection to be updated
+ * @param projectId - project id of current repetitions collection
+ */
+async function backfillTimestampsFromEvents(db, repetitionCollectionName, projectId) {
+  const repetitions = db.collection(repetitionCollectionName);
+  const events = db.collection(`events:${projectId}`);
+
+  let bulkOps = [];
+  let repetitionCount = 1;
+
+  const repetitionsList = await repetitions.find(
+    {
+      timestamp: { $exists: false },
+    },
+    { projection: { _id: 1, groupHash: 1 } }
+  ).limit(documentsSelectionLimit).toArray();
+
+  const groupHashList = [];
+
+  for (const repetition of repetitionsList) {
+    process.stdout.write(`\r[${repetitionCount} repetition] update with timestamp now have [${bulkOps.length + 1}] ops in bulkOps`);
+    groupHashList.push(repetition.groupHash);
+    repetitionCount++;
+  }
+
+  const relatedEvents = await events.find(
+    { groupHash: { $in: groupHashList } },
+    { projection: { timestamp: 1, groupHash: 1 } }
+  ).toArray();
+
+  const relatedEventsMap = new Map()
+
+  relatedEvents.forEach(e => {
+    relatedEventsMap.set(e.groupHash, e);
+  })
+
+  for (const repetition of repetitionsList) {
+    const relatedEvent = relatedEventsMap.get(repetition.groupHash);
+
+    if (!relatedEvent) {
+      bulkOps.push({
+        deleteOne: {
+          filter: { _id: repetition._id }
+        }
+      })
+    } else if (relatedEvent?.timestamp !== null) {
+      bulkOps.push({
+        updateOne: {
+          filter: { _id: repetition._id },
+          update: { $set: { timestamp: Number(relatedEvent.timestamp) } },
+        },
+      });
+    }
+  }
+
+  let processed = 0;
+
+  if (bulkOps.length > 0) {
+    const result = await repetitions.bulkWrite(bulkOps);
+    const updated = result.modifiedCount;
+    const deleted = result.deletedCount;
+    processed = bulkOps.length;
+    console.log(`  updates (${processed} processed, ${updated} updated, ${deleted} deleted)`);
+
+    if (updated + deleted === 0) {
+      repetitionCollectionsToCheck.filter(collection => collection !== repetition)
+    }
+  }
+
+  return processed;
+}
+
+/**
+ * Method that runs convertor script
+ */
+async function run() {
+  const fullUri = 'mongodb://hawk_new:evieg9bauK0ahs2youhoh7aer7kohT@rc1d-2jltinutse1eadfs.mdb.yandexcloud.net:27018/hawk_events?authSource=admin&replicaSet=rs01&tls=true&tlsInsecure=true';
+
+  // Parse the Mongo URL manually
+  const mongoUrl = new URL(fullUri);
+  const databaseName = 'hawk_events';
+
+  // Extract query parameters
+  const queryParams = Object.fromEntries(mongoUrl.searchParams.entries());
+
+  // Compose connection options manually
+  const options = {
+    useNewUrlParser: true,
+    useUnifiedTopology: true,
+    authSource: queryParams.authSource || 'admin',
+    replicaSet: queryParams.replicaSet || undefined,
+    tls: queryParams.tls === 'true',
+    tlsInsecure: queryParams.tlsInsecure === 'true',
+    // connectTimeoutMS: 3600000,
+    // socketTimeoutMS: 3600000,
+  };
+
+  // Remove query string from URI
+  mongoUrl.search = '';
+  const cleanUri = mongoUrl.toString();
+
+  console.log('Connecting to:', cleanUri);
+  console.log('With options:', options);
+
+  const client = new MongoClient(cleanUri, options);
+
+  await client.connect();
+  const db = client.db(databaseName);
+
+  console.log(`Connected to database: ${databaseName}`);
+
+  const collections = await db.listCollections({}, {
+    authorizedCollections: true,
+    nameOnly: true,
+  }).toArray();
+
+  let eventCollectionsToCheck = collections.filter(col => /^events:/.test(col.name)).map(col => col.name);
+  let repetitionCollectionsToCheck = collections.filter(col => /^repetitions:/.test(col.name)).map(col => col.name);
+
+  console.log(`Found ${eventCollectionsToCheck.length} event collections.`);
+  console.log(`Found ${repetitionCollectionsToCheck.length} repetition collections.`);
+
+  // Convert events
+  let i = 1;
+  let documentsUpdatedCount = 1
+
+  while (documentsUpdatedCount != 0) {
+    documentsUpdatedCount = 0;
+    i = 1;
+    const collectionsToUpdateCount = eventCollectionsToCheck.length;
+  
+    for (const collectionName of eventCollectionsToCheck) {
+      console.log(`[${i}/${collectionsToUpdateCount}] Processing ${collectionName}`);
+      const updated = await movePayloadTimestampToEventLevel(db, collectionName);
+
+      if (updated === 0) {
+        eventCollectionsToCheck = eventCollectionsToCheck.filter(collection => collection !== collectionName);
+      }
+
+      documentsUpdatedCount += updated
+      i++;
+    }
+  } 
+
+  // Convert repetitions + backfill from events
+  documentsUpdatedCount = 1;
+
+  while (documentsUpdatedCount != 0) {
+    documentsUpdatedCount = 0;
+    i = 1;
+    const collectionsToUpdateCount = repetitionCollectionsToCheck.length;
+
+    for (const collectionName of repetitionCollectionsToCheck) {
+      console.log(`[${i}/${collectionsToUpdateCount}] Processing ${collectionName}`);
+      const projectId = collectionName.split(':')[1];
+
+      let updated = 0;
+
+      updated += await movePayloadTimestampToEventLevel(db, collectionName);
+      updated += await backfillTimestampsFromEvents(db, collectionName, projectId);
+
+      if (updated === 0) {
+        repetitionCollectionsToCheck = repetitionCollectionsToCheck.filter(collection => collection !== collectionName);
+      }
+
+      documentsUpdatedCount += updated;
+      i++;
+    }
+
+    console.log(`Conversion iteration complete. ${documentsUpdatedCount} documents updated`);
+  }
+
+  await client.close();
+}
+
+run().catch(err => {
+  console.error('❌ Script failed:', err);
+  process.exit(1);
+});

lib/event-worker.ts

@@ -1,7 +1,7 @@
 import { Worker } from './worker';
 import * as WorkerNames from './workerNames';
 import { GroupWorkerTask } from 'hawk-worker-grouper/types/group-worker-task';
-import { EventWorkerTask } from './types/event-worker-task';
+import { CatcherMessageType, CatcherMessagePayload, CatcherMessageAccepted, ErrorsCatcherType } from '@hawk.so/types';
 
 /**
  * Defines a Worker that handles events from Catcher.
@@ -12,30 +12,31 @@ export abstract class EventWorker extends Worker {
    * Worker type (will pull tasks from Registry queue with the same name)
    * 'errors/nodejs' for example
    */
-  public type = '';
+  public type: ErrorsCatcherType;
 
   /**
    * Message handle function
    *
-   * @param {EventWorkerTask} event - event to handle
+   * @param {CatcherMessageAccepted<CatcherMessageType>} task - worker task to handle
    */
-  public async handle(event: EventWorkerTask): Promise<void> {
-    this.validate(event);
+  public async handle(task: CatcherMessageAccepted<CatcherMessageType>): Promise<void> {
+    this.validate(task);
 
     await this.addTask(WorkerNames.GROUPER, {
-      projectId: event.projectId,
-      catcherType: this.type,
-      event: event.payload,
-    } as GroupWorkerTask);
+      projectId: task.projectId,
+      catcherType: this.type as CatcherMessageType,
+      payload: task.payload as CatcherMessagePayload<ErrorsCatcherType>,
+      timestamp: task.timestamp,
+    } as GroupWorkerTask<ErrorsCatcherType>);
   }
 
   /**
    * Validate passed event data
    *
-   * @param {EventWorkerTask} event - event to be validated
+   * @param {CatcherMessageAccepted<CatcherMessageType>} task - task to be validated
    */
-  protected validate(event: EventWorkerTask): void {
-    if (!event.projectId || !event.payload) {
+  protected validate(task: CatcherMessageAccepted<CatcherMessageType>): void {
+    if (!task.projectId || !task.payload || !task.timestamp) {
       throw new Error('Bad data was given');
     }
   }

lib/types/event-worker-task.d.ts

@@ -4,4 +4,4 @@
  * When you create new type of worker, describe its task's structure with interface inherited from WorkerTask
  */
 export interface WorkerTask {
-}
+}

lib/types/worker-task.d.ts

@@ -50,6 +50,14 @@ export function encodeUnsafeFields(event: GroupedEventDBScheme | RepetitionDBSch
      * Repetition includes delta field, grouped event includes payload
      */
     if ('delta' in event) {
+      /**
+       * We need to check if delta field exists but with undefined value
+       * It would mean that repetition payload is same with original event paylaod
+       */
+      if (event.delta === undefined) {
+        return;
+      }
+
       fieldValue = event.delta[field];
     } else {
       fieldValue = event.payload[field];

lib/utils/unsafeFields.ts

@@ -2,7 +2,7 @@
  * Class for critical errors
  * have to stop process
  */
-import { EventAddons, EventContext, EventDataAccepted } from '@hawk.so/types';
+import { EventAddons, EventContext, EventData } from '@hawk.so/types';
 
 /**
  * Error class with additional error context for debugging
@@ -71,8 +71,8 @@ export class DiffCalculationError extends NonCriticalError {
    */
   constructor(
     msg: string | Error,
-    originalEvent: EventDataAccepted<EventAddons>,
-    eventToCompare: EventDataAccepted<EventAddons>
+    originalEvent: EventData<EventAddons>,
+    eventToCompare: EventData<EventAddons>
   ) {
     super(msg);
     this.context = {

lib/workerErrors.ts

@@ -14,6 +14,7 @@
         "migration-add": "migrate-mongo create",
         "migrate:up": "migrate-mongo up",
         "migrate:down": "migrate-mongo down",
+        "convert": "node ./convertors/move-timestamp-out-of-payload.js",
         "lint": "eslint -c ./.eslintrc.js --ext .ts,.js --fix .",
         "test": "jest --coverage --runInBand",
         "test:archiver": "jest workers/archiver",
@@ -48,7 +49,7 @@
     },
     "dependencies": {
         "@hawk.so/nodejs": "^3.1.1",
-        "@hawk.so/types": "^0.1.29",
+        "@hawk.so/types": "^0.1.32",
         "@types/amqplib": "^0.8.2",
         "@types/jest": "^29.2.3",
         "@types/mongodb": "^3.5.15",

package.json

@@ -168,7 +168,7 @@ export default class ArchiverWorker extends Worker {
     const repetitionsBulk = repetitionsCollection.initializeUnorderedBulkOp();
 
     repetitionsBulk.find({
-      'payload.timestamp': {
+      timestamp: {
         $lt: maxOldTimestamp,
       },
     }).delete();