From 8d1b6133d8803bab9010e73a9497498d9c27a5f2 Mon Sep 17 00:00:00 2001 From: Kuchizu Date: Tue, 10 Nov 2026 23:25:08 +0300 Subject: [PATCH] Ensure project event collections have indexes before grouper writes --- workers/grouper/src/index.ts | 53 +++++++ workers/grouper/src/metrics/grouperMetrics.ts | 1 + workers/grouper/src/utils/eventIndexes.ts | 130 ++++++++++++++++++ workers/grouper/tests/eventIndexes.test.ts | 61 ++++++++ 4 files changed, 245 insertions(+) create mode 100644 workers/grouper/src/utils/eventIndexes.ts create mode 100644 workers/grouper/tests/eventIndexes.test.ts diff --git a/workers/grouper/src/index.ts b/workers/grouper/src/index.ts index 60e9cb25..c9c0b0ea 100644 --- a/workers/grouper/src/index.ts +++ b/workers/grouper/src/index.ts @@ -20,6 +20,7 @@ import DataFilter from './data-filter'; import RedisHelper from './redisHelper'; import { computeDelta } from './utils/repetitionDiff'; import { bucketTimestampMs } from './utils/bucketTimestamp'; +import { ensureEventIndexes } from './utils/eventIndexes'; import GrouperMetrics from './metrics/grouperMetrics'; import GrouperMemoryMonitor from './metrics/memoryMonitor'; import SlowHandleDiagnostics, { SlowHandleSession } from './metrics/slowHandleDiagnostics'; @@ -43,6 +44,11 @@ const MEMOIZATION_TTL = 60_000; */ const CACHE_CLEANUP_INTERVAL_SECONDS = 30; +/** + * How long a project's event collections are considered checked + */ +const EVENT_INDEXES_CHECK_TTL = TimeMs.HOUR; + /** * Error code of MongoDB key duplication error */ @@ -107,6 +113,11 @@ export default class GrouperWorker extends Worker { */ private handledTasksCount = 0; + /** + * Running or finished event collections checks by project id + */ + private eventIndexesChecks: Map }> = new Map(); + /** * Start consuming messages */ @@ -202,6 +213,8 @@ export default class GrouperWorker extends Worker { */ this.dataFilter.processEvent(task.payload); + await session.measureStep('ensureEventIndexes', () => this.ensureEventIndexes(task.projectId)); + let uniqueEventHash = await session.measureStep('hash', () => this.getUniqueEventHash(task)); let existedEvent: GroupedEventDBScheme; let repetitionId = null; @@ -498,6 +511,46 @@ export default class GrouperWorker extends Worker { } } + /** + * Makes sure project's event collections have indexes before the first write, once per TTL. + * Failures are logged and do not stop event processing. + * + * @param projectId - project to check + */ + private ensureEventIndexes(projectId: string): Promise { + if (!projectId || !mongodb.ObjectID.isValid(projectId)) { + return Promise.resolve(); + } + + const cached = this.eventIndexesChecks.get(projectId); + + if (cached && cached.expiresAt > Date.now()) { + return cached.done; + } + + const done = ensureEventIndexes(this.eventsDb.getConnection(), projectId) + .then(({ created, missing }) => { + if (created.length > 0) { + this.logger.info(`[ensureEventIndexes] project=${projectId} created collections: ${created.join(', ')}`); + } + + if (missing.length > 0) { + this.logger.warn(`[ensureEventIndexes] project=${projectId} missing indexes: ${missing.join(', ')}`); + } + }) + .catch((error) => { + this.eventIndexesChecks.delete(projectId); + this.logger.error(`[ensureEventIndexes] project=${projectId} failed: ${error.message}`); + }); + + this.eventIndexesChecks.set(projectId, { + expiresAt: Date.now() + EVENT_INDEXES_CHECK_TTL, + done, + }); + + return done; + } + /** * Get unique hash based on event type and title * diff --git a/workers/grouper/src/metrics/grouperMetrics.ts b/workers/grouper/src/metrics/grouperMetrics.ts index 02a36306..add82a8c 100644 --- a/workers/grouper/src/metrics/grouperMetrics.ts +++ b/workers/grouper/src/metrics/grouperMetrics.ts @@ -19,6 +19,7 @@ export type GrouperStep = 'computeDelta' | 'decodeEvent' | 'enqueueNotifier' | + 'ensureEventIndexes' | 'findSimilarEvent' | 'getEvent' | 'hash' | diff --git a/workers/grouper/src/utils/eventIndexes.ts b/workers/grouper/src/utils/eventIndexes.ts new file mode 100644 index 00000000..54956c89 --- /dev/null +++ b/workers/grouper/src/utils/eventIndexes.ts @@ -0,0 +1,130 @@ +import type { Db, IndexSpecification } from 'mongodb'; + +const NAMESPACE_NOT_FOUND = 26; + +/** + * Must match the indexes hawk.api.nodejs creates for a new project + */ +export const EVENT_COLLECTIONS_INDEXES: Record = { + events: [ + { + key: { 'payload.release': 1 }, + name: 'payloadRelease', + sparse: true, + }, + { + key: { groupHash: 1 }, + name: 'groupHashUnique', + unique: true, + }, + { + key: { timestamp: 1 }, + name: 'timestamp', + sparse: true, + }, + ], + repetitions: [ + { + key: { groupHash: 'hashed' }, + name: 'groupHash_hashed', + }, + { + key: { + groupHash: 1, + release: 1, + }, + name: 'groupHashAndRelease', + }, + { + key: { 'payload.user.id': 1 }, + name: 'userId', + sparse: true, + }, + { + key: { timestamp: 1 }, + name: 'timestamp', + sparse: true, + }, + ], + dailyEvents: [ + { + key: { groupingTimestamp: 1 }, + name: 'groupingTimestamp', + }, + { + key: { + groupingTimestamp: 1, + groupHash: 1, + }, + name: 'groupingTimestampAndGroupHash', + }, + { + key: { + groupingTimestamp: -1, + lastRepetitionTime: -1, + _id: -1, + }, + name: 'groupingTimestampAndLastRepetitionTimeAndId', + }, + { + key: { groupHash: 1 }, + name: 'groupHash', + }, + ], +}; + +export interface EventIndexesReport { + /** + * Collections that did not exist and were created with indexes + */ + created: string[]; + + /** + * Expected indexes absent in already existing collections + */ + missing: string[]; +} + +/** + * Creates missing event collections of a project with their indexes. + * Existing collections are only compared with the expected set: building an index on live data is not a hot path job. + * + * @param db - events database + * @param projectId - project to check + */ +export async function ensureEventIndexes(db: Db, projectId: string): Promise { + const report: EventIndexesReport = { + created: [], + missing: [], + }; + + for (const [prefix, specs] of Object.entries(EVENT_COLLECTIONS_INDEXES)) { + const collection = db.collection(`${prefix}:${projectId}`); + const existing: IndexSpecification[] = await collection.indexes().catch((error) => { + if (error.code !== NAMESPACE_NOT_FOUND) { + throw error; + } + + return []; + }); + + if (existing.length === 0) { + await collection.createIndexes(specs); + report.created.push(prefix); + + continue; + } + + for (const spec of specs) { + const found = existing.some((index) => { + return JSON.stringify(index.key) === JSON.stringify(spec.key) && !!index.unique === !!spec.unique; + }); + + if (!found) { + report.missing.push(`${prefix}.${spec.name}`); + } + } + } + + return report; +} diff --git a/workers/grouper/tests/eventIndexes.test.ts b/workers/grouper/tests/eventIndexes.test.ts new file mode 100644 index 00000000..a746bd17 --- /dev/null +++ b/workers/grouper/tests/eventIndexes.test.ts @@ -0,0 +1,61 @@ +import '../../../env-test'; +import { MongoClient, ObjectID } from 'mongodb'; +import type { Db } from 'mongodb'; +import { ensureEventIndexes, EVENT_COLLECTIONS_INDEXES } from '../src/utils/eventIndexes'; + +describe('ensureEventIndexes', () => { + let connection: MongoClient; + let db: Db; + + beforeAll(async () => { + connection = await MongoClient.connect(process.env.MONGO_EVENTS_DATABASE_URI, { + useNewUrlParser: true, + useUnifiedTopology: true, + }); + db = connection.db(); + }); + + afterAll(async () => { + await connection.close(); + }); + + it('creates missing collections with all indexes and reports nothing on the next run', async () => { + const projectId = new ObjectID().toString(); + + expect(await ensureEventIndexes(db, projectId)).toEqual({ + created: ['events', 'repetitions', 'dailyEvents'], + missing: [], + }); + + for (const [prefix, specs] of Object.entries(EVENT_COLLECTIONS_INDEXES)) { + const names = (await db.collection(`${prefix}:${projectId}`).indexes()).map(index => index.name); + + expect(names.sort()).toEqual([...specs.map(spec => spec.name), '_id_'].sort()); + } + + const events = db.collection(`events:${projectId}`); + + await events.insertOne({ groupHash: 'hash' }); + await expect(events.insertOne({ groupHash: 'hash' })).rejects.toMatchObject({ code: 11000 }); + + expect(await ensureEventIndexes(db, projectId)).toEqual({ + created: [], + missing: [], + }); + }); + + it('reports indexes missing in an existing collection without building them', async () => { + const projectId = new ObjectID().toString(); + const events = db.collection(`events:${projectId}`); + + await events.createIndex({ groupHash: 1 }); + + const report = await ensureEventIndexes(db, projectId); + + expect(report).toEqual({ + created: ['repetitions', 'dailyEvents'], + missing: ['events.payloadRelease', 'events.groupHashUnique', 'events.timestamp'], + }); + expect(await events.indexes()).toHaveLength(2); + }); +});