zias/stats
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

wip

Browse Source
This commit is contained in:
Carl-Gerhard Lindesvärd
2025-11-12 11:13:34 +01:00
parent 2286c331d3
commit 3d1cdadb6f
1 changed files with 3 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
apps/worker/src/jobs/events.incoming-event.ts
View File

@@ -18,8 +18,6 @@ import {
} from '@openpanel/db'; } from '@openpanel/db';
import type { ILogger } from '@openpanel/logger'; import type { ILogger } from '@openpanel/logger';
import type { EventsQueuePayloadIncomingEvent } from '@openpanel/queue'; import type { EventsQueuePayloadIncomingEvent } from '@openpanel/queue';
import { getLock, getRedisCache } from '@openpanel/redis';
import { DelayedError, type Job } from 'bullmq';
import * as R from 'ramda'; import * as R from 'ramda';
import { v4 as uuid } from 'uuid'; import { v4 as uuid } from 'uuid';
@@ -33,10 +31,9 @@ const merge = <A, B>(a: Partial<A>, b: Partial<B>): A & B =>
async function createEventAndNotify( async function createEventAndNotify(
payload: IServiceCreateEventPayload, payload: IServiceCreateEventPayload,
jobData: Job<EventsQueuePayloadIncomingEvent>['data']['payload'],
logger: ILogger, logger: ILogger,
) { ) {
logger.info('Creating event', { event: payload, jobData }); logger.info('Creating event', { event: payload });
const [event] = await Promise.all([ const [event] = await Promise.all([
createEvent(payload), createEvent(payload),
checkNotificationRulesForEvent(payload).catch(() => {}), checkNotificationRulesForEvent(payload).catch(() => {}),
@@ -94,7 +91,6 @@ export async function incomingEvent(
projectId, projectId,
properties: R.omit(GLOBAL_PROPERTIES, { properties: R.omit(GLOBAL_PROPERTIES, {
...properties, ...properties,
__user_agent: userAgent,
__hash: hash, __hash: hash,
__query: query, __query: query,
__reqId: reqId, __reqId: reqId,
@@ -154,7 +150,7 @@ export async function incomingEvent(
origin: screenView?.origin ?? baseEvent.origin, origin: screenView?.origin ?? baseEvent.origin,
}; };
return createEventAndNotify(payload as IServiceEvent, jobPayload, logger); return createEventAndNotify(payload as IServiceEvent, logger);
} }
const sessionEnd = await getSessionEnd({ const sessionEnd = await getSessionEnd({
@@ -190,7 +186,7 @@ export async function incomingEvent(
}); });
} }
const event = await createEventAndNotify(payload, jobPayload, logger); const event = await createEventAndNotify(payload, logger);
if (!sessionEnd) { if (!sessionEnd) {
logger.info('Creating session end job', { event: payload }); logger.info('Creating session end job', { event: payload });