feat(api): dual-write to unified events collection for visit/mark/assignee

- Add src/events/unified/dualWrite.js (visitEventUnified, toggleEventMarkUnified, updateAssigneeUnified)
- Add hawk_dual_write_failures_total metric
- eventsFactory: call dual-write after visitEvent, toggleEventMark, updateAssignee
- Update .env.sample with USE_UNIFIED_EVENTS_COLLECTIONS
- Update docs/METRICS.md

Made-with: Cursor

Author: pavelzotikov

.env.sample

@@ -32,6 +32,9 @@ INVITE_LINK_HASH_SALT=secret_hash_salt
 # Option to enable playground (if true playground will be available at /graphql route)
 PLAYGROUND_ENABLE=false
 
+# If true, dual-write to unified events/repetitions collections (events, repetitions with projectId)
+USE_UNIFIED_EVENTS_COLLECTIONS=false
+
 # AMQP URL
 AMQP_URL=amqp://guest:guest@rabbitmq
 

docs/METRICS.md

@@ -135,6 +135,15 @@ Labels:
 
 **Purpose**: Track transient or persistent database errors.
 
+#### hawk_dual_write_failures_total (Counter)
+
+Counter of dual-write failures to unified events/repetitions collections.
+
+Labels:
+- `type` - `events` | `repetitions`
+
+**Purpose**: Track failures when writing to unified collections (when `USE_UNIFIED_EVENTS_COLLECTIONS=true`). Used for monitoring dual-write reliability during migration.
+
 ## Testing
 
 ### Manual Testing

src/events/unified/dualWrite.js

@@ -0,0 +1,106 @@
+/**
+ * Dual-write to unified events collection (API layer)
+ *
+ * When USE_UNIFIED_EVENTS_COLLECTIONS=true, duplicates updates to the unified
+ * events collection. Errors are logged, not thrown.
+ *
+ * @see docs/mongodb-unified-collections/
+ */
+
+const { ObjectId } = require('mongodb');
+const { incrementDualWriteFailure } = require('../../metrics/dualWrite');
+
+const EVENTS_COLLECTION = 'events';
+
+function isUnifiedEnabled() {
+  return process.env.USE_UNIFIED_EVENTS_COLLECTIONS === 'true';
+}
+
+function getEventsDb() {
+  const mongo = require('../../mongo');
+  return mongo.databases?.events;
+}
+
+/**
+ * Add user to visitedBy in unified events collection
+ *
+ * @param {string} projectId - project ObjectId
+ * @param {string} eventId - event ObjectId
+ * @param {string} userId - user ObjectId
+ */
+async function visitEventUnified(projectId, eventId, userId) {
+  if (!isUnifiedEnabled()) return;
+
+  try {
+    const db = getEventsDb();
+    if (!db) return;
+
+    await db.collection(EVENTS_COLLECTION).updateOne(
+      { projectId: new ObjectId(projectId), _id: new ObjectId(eventId) },
+      { $addToSet: { visitedBy: new ObjectId(userId) } }
+    );
+  } catch (err) {
+    incrementDualWriteFailure('events');
+    console.error('[dualWrite] visitEventUnified failed', { projectId, eventId, err });
+  }
+}
+
+/**
+ * Toggle event mark (resolved, ignored, starred) in unified events collection
+ *
+ * @param {string} projectId - project ObjectId
+ * @param {string} eventId - event ObjectId
+ * @param {string} mark - mark name (e.g. 'resolved', 'ignored', 'starred')
+ * @param {boolean} isUnset - if true, remove mark; if false, set mark with timestamp
+ */
+async function toggleEventMarkUnified(projectId, eventId, mark, isUnset) {
+  if (!isUnifiedEnabled()) return;
+
+  try {
+    const db = getEventsDb();
+    if (!db) return;
+
+    const markKey = `marks.${mark}`;
+    const update = isUnset
+      ? { $unset: { [markKey]: '' } }
+      : { $set: { [markKey]: Math.floor(Date.now() / 1000) } };
+
+    await db.collection(EVENTS_COLLECTION).updateOne(
+      { projectId: new ObjectId(projectId), _id: new ObjectId(eventId) },
+      update
+    );
+  } catch (err) {
+    incrementDualWriteFailure('events');
+    console.error('[dualWrite] toggleEventMarkUnified failed', { projectId, eventId, mark, err });
+  }
+}
+
+/**
+ * Update assignee in unified events collection
+ *
+ * @param {string} projectId - project ObjectId
+ * @param {string} eventId - event ObjectId
+ * @param {string} assignee - assignee user id or '' to unassign
+ */
+async function updateAssigneeUnified(projectId, eventId, assignee) {
+  if (!isUnifiedEnabled()) return;
+
+  try {
+    const db = getEventsDb();
+    if (!db) return;
+
+    await db.collection(EVENTS_COLLECTION).updateOne(
+      { projectId: new ObjectId(projectId), _id: new ObjectId(eventId) },
+      { $set: { assignee } }
+    );
+  } catch (err) {
+    incrementDualWriteFailure('events');
+    console.error('[dualWrite] updateAssigneeUnified failed', { projectId, eventId, err });
+  }
+}
+
+module.exports = {
+  visitEventUnified,
+  toggleEventMarkUnified,
+  updateAssigneeUnified,
+};

src/metrics/dualWrite.ts

@@ -0,0 +1,21 @@
+/**
+ * Prometheus metrics for dual-write to unified collections
+ *
+ * @see docs/mongodb-unified-collections/
+ */
+
+import promClient from 'prom-client';
+
+export const dualWriteFailuresTotal = new promClient.Counter({
+  name: 'hawk_dual_write_failures_total',
+  help: 'Counter of dual-write failures to unified collections by type',
+  labelNames: ['type'],
+});
+
+/**
+ * Increment dual-write failure counter
+ * @param type - 'events' | 'repetitions'
+ */
+export function incrementDualWriteFailure(type: 'events' | 'repetitions'): void {
+  dualWriteFailuresTotal.labels(type).inc();
+}

src/metrics/index.ts

@@ -2,6 +2,7 @@ import client from 'prom-client';
 import express from 'express';
 import { gqlOperationDuration, gqlOperationErrors, gqlResolverDuration } from './graphql';
 import { mongoCommandDuration, mongoCommandErrors } from './mongodb';
+import { dualWriteFailuresTotal } from './dualWrite';
 
 /**
  * Create a Registry to register the metrics
@@ -49,6 +50,11 @@ register.registerMetric(gqlResolverDuration);
 register.registerMetric(mongoCommandDuration);
 register.registerMetric(mongoCommandErrors);
 
+/**
+ * Register dual-write metrics
+ */
+register.registerMetric(dualWriteFailuresTotal);
+
 /**
  * Express middleware to track HTTP metrics
  * @param req - Express request object

src/models/eventsFactory.js

@@ -6,6 +6,11 @@ import ChartDataService from '../services/chartDataService';
 
 const Factory = require('./modelFactory');
 const mongo = require('../mongo');
+const {
+  visitEventUnified,
+  toggleEventMarkUnified,
+  updateAssigneeUnified,
+} = require('../events/unified/dualWrite');
 const Event = require('../models/event');
 const { ObjectId } = require('mongodb');
 const { composeEventPayloadByRepetition } = require('../utils/merge');
@@ -836,6 +841,8 @@ class EventsFactory extends Factory {
       throw new Error(`Event not found for eventId: ${eventId}`);
     }
 
+    visitEventUnified(this.projectId.toString(), eventId, userId).catch(() => {});
+
     return result;
   }
 
@@ -861,8 +868,9 @@ class EventsFactory extends Factory {
     const markKey = `marks.${mark}`;
 
     let update;
+    const isUnset = !!(event.marks && event.marks[mark]);
 
-    if (event.marks && event.marks[mark]) {
+    if (isUnset) {
       update = {
         $unset: { [markKey]: '' },
       };
@@ -872,7 +880,11 @@ class EventsFactory extends Factory {
       };
     }
 
-    return collection.updateOne(query, update);
+    const result = await collection.updateOne(query, update);
+
+    toggleEventMarkUnified(this.projectId.toString(), eventId, mark, isUnset).catch(() => {});
+
+    return result;
   }
 
   /**
@@ -920,6 +932,8 @@ class EventsFactory extends Factory {
       throw new Error(`Event not found for eventId: ${eventId}`);
     }
 
+    updateAssigneeUnified(this.projectId.toString(), eventId, assignee).catch(() => {});
+
     return result;
   }
 

src/types/env.d.ts

@@ -101,5 +101,11 @@ declare namespace NodeJS {
      * @example "my-github-app"
      */
     GITHUB_APP_SLUG?: string;
+
+    /**
+     * If true, dual-write to unified events/repetitions collections (events, repetitions with projectId)
+     * @default 'false'
+     */
+    USE_UNIFIED_EVENTS_COLLECTIONS?: string;
   }
 }