From 847476951a776ed517e7735f26d986ad03e47bff Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Wed, 26 May 2021 22:57:28 +0200 Subject: [PATCH 1/5] Reorient `ActionManager` to group by teamId for practicality --- src/utils/db/db.ts | 4 +- src/utils/utils.ts | 31 ++++++++ src/worker/ingestion/action-manager.ts | 46 +++++++----- src/worker/ingestion/process-event.ts | 1 - src/worker/tasks.ts | 10 +-- tests/postgres/worker.test.ts | 8 +- tests/utils.test.ts | 74 +++++++++++++++++++ tests/worker/ingestion/action-manager.test.ts | 66 ++++++++++++++--- 8 files changed, 198 insertions(+), 42 deletions(-) diff --git a/src/utils/db/db.ts b/src/utils/db/db.ts index 1cf47cc3..fe3a505f 100644 --- a/src/utils/db/db.ts +++ b/src/utils/db/db.ts @@ -725,7 +725,7 @@ export class DB { // Action & ActionStep - public async fetchAllActionsMap(): Promise> { + public async fetchAllActions(): Promise { const rawActions: RawAction[] = ( await this.postgresQuery(`SELECT * FROM posthog_action WHERE deleted = FALSE`, undefined, 'fetchActions') ).rows @@ -747,7 +747,7 @@ export class DB { actionsMap[actionStep.action_id].steps.push(actionStep) } } - return actionsMap + return Object.values(actionsMap) } public async fetchAction(id: Action['id']): Promise { diff --git a/src/utils/utils.ts b/src/utils/utils.ts index 536b0689..4ef95fe1 100644 --- a/src/utils/utils.ts +++ b/src/utils/utils.ts @@ -566,3 +566,34 @@ export function filterIncrementProperties(incrementProperties: unknown): Record< return filteredIncrementProperties } + +export function groupBy, K extends keyof T>( + objects: T[], + key: K, + flat?: false +): Record +export function groupBy, K extends keyof T>( + objects: T[], + key: K, + flat: true +): Record +export function groupBy, K extends keyof T>( + objects: T[], + key: K, + flat = false +): Record { + return flat + ? objects.reduce((grouping, currentItem) => { + if (currentItem[key] in grouping) { + throw new Error( + `Key "${key}" has more than one matching value, which is not allowed in flat groupBy!` + ) + } + grouping[currentItem[key]] = currentItem + return grouping + }, {} as Record) + : objects.reduce((grouping, currentItem) => { + ;(grouping[currentItem[key]] = grouping[currentItem[key]] || []).push(currentItem) + return grouping + }, {} as Record) +} diff --git a/src/worker/ingestion/action-manager.ts b/src/worker/ingestion/action-manager.ts index 33659926..cd0d38b9 100644 --- a/src/worker/ingestion/action-manager.ts +++ b/src/worker/ingestion/action-manager.ts @@ -1,8 +1,9 @@ -import { Action, PluginsServerConfig } from '../../types' +import { Action, Team } from '../../types' import { DB } from '../../utils/db/db' import { status } from '../../utils/status' +import { groupBy } from '../../utils/utils' -type ActionCache = Record +type ActionCache = Record> export class ActionManager { private ready: boolean @@ -20,48 +21,57 @@ export class ActionManager { this.ready = true } - public getAction(id: Action['id']): Action | undefined { + public getTeamActions(teamId: Team['id']): Action[] { if (!this.ready) { throw new Error('ActionManager is not ready! Run actionManager.prepare() before this') } - return this.actionCache[id] + return Object.values(this.actionCache[teamId] ?? {}) } public async reloadAllActions(): Promise { - this.actionCache = await this.db.fetchAllActionsMap() + this.actionCache = Object.fromEntries( + Object.entries(groupBy(await this.db.fetchAllActions(), 'team_id')).map(([teamId, actions]) => [ + teamId, + groupBy(actions, 'id', true), + ]) + ) status.info('🍿', 'Fetched all actions from DB anew') } - public async reloadAction(id: Action['id']): Promise { - const refetchedAction = await this.db.fetchAction(id) + public async reloadAction(teamId: Team['id'], actionId: Action['id']): Promise { + const refetchedAction = await this.db.fetchAction(actionId) + const wasCachedAlready = teamId in this.actionCache && actionId in this.actionCache[teamId] if (refetchedAction) { status.info( '🍿', - id in this.actionCache ? `Refetched action ID ${id} from DB` : `Fetched new action ID ${id} from DB` + wasCachedAlready + ? `Refetched action ID ${actionId} from DB` + : `Fetched new action ID ${actionId} from DB` ) - this.actionCache[id] = refetchedAction - } else if (id in this.actionCache) { + this.actionCache[teamId][actionId] = refetchedAction + } else if (wasCachedAlready) { status.info( '🍿', - `Tried to fetch action ID ${id} from DB, but it wasn't found in DB, so deleted from cache instead` + `Tried to fetch action ID ${actionId} from DB, but it wasn't found in DB, so deleted from cache instead` ) - delete this.actionCache[id] + delete this.actionCache[teamId][actionId] } else { status.info( '🍿', - `Tried to fetch action ID ${id} from DB, but it wasn't found in DB or cache, so did nothing instead` + `Tried to fetch action ID ${actionId} from DB, but it wasn't found in DB or cache, so did nothing instead` ) } } - public dropAction(id: Action['id']): void { - if (id in this.actionCache) { - status.info('🍿', `Deleted action ID ${id} from cache`) - delete this.actionCache[id] + public dropAction(teamId: Team['id'], actionId: Action['id']): void { + const wasCachedAlready = teamId in this.actionCache && actionId in this.actionCache[teamId] + if (wasCachedAlready) { + status.info('🍿', `Deleted action ID ${actionId} from cache`) + delete this.actionCache[teamId][actionId] } else { status.info( '🍿', - `Tried to delete action ID ${id} from cache, but it wasn't found in cache, so did nothing instead` + `Tried to delete action ID ${actionId} from cache, but it wasn't found in cache, so did nothing instead` ) } } diff --git a/src/worker/ingestion/process-event.ts b/src/worker/ingestion/process-event.ts index 24722edd..4ce55bcb 100644 --- a/src/worker/ingestion/process-event.ts +++ b/src/worker/ingestion/process-event.ts @@ -24,7 +24,6 @@ import { Client } from '../../utils/celery/client' import { DB } from '../../utils/db/db' import { KafkaProducerWrapper } from '../../utils/db/kafka-producer-wrapper' import { elementsToString, personInitialAndUTMProperties, sanitizeEventName, timeoutGuard } from '../../utils/db/utils' -import { PubSub } from '../../utils/pubsub' import { status } from '../../utils/status' import { castTimestampOrNow, filterIncrementProperties, UUID, UUIDT } from '../../utils/utils' import { ActionManager } from './action-manager' diff --git a/src/worker/tasks.ts b/src/worker/tasks.ts index 59f9ba78..823bccfe 100644 --- a/src/worker/tasks.ts +++ b/src/worker/tasks.ts @@ -1,6 +1,6 @@ import { PluginEvent } from '@posthog/plugin-scaffold/src/types' -import { Action, EnqueuedJob, Hub, PluginTaskType } from '../types' +import { Action, EnqueuedJob, Hub, PluginTaskType, Team } from '../types' import { ingestEvent } from './ingestion/ingest-event' import { runOnEvent, runOnSnapshot, runPluginTask, runProcessEvent, runProcessEventBatch } from './plugins/run' import { loadSchedule, setupPlugins } from './plugins/setup' @@ -48,11 +48,11 @@ export const workerTasks: Record = { reloadAllActions: async (hub) => { return await hub.eventsProcessor.actionManager.reloadAllActions() }, - reloadAction: async (hub, args: { actionId: Action['id'] }) => { - return await hub.eventsProcessor.actionManager.reloadAction(args.actionId) + reloadAction: async (hub, args: { teamId: Team['id']; actionId: Action['id'] }) => { + return await hub.eventsProcessor.actionManager.reloadAction(args.teamId, args.actionId) }, - dropAction: (hub, args: { actionId: Action['id'] }) => { - return hub.eventsProcessor.actionManager.dropAction(args.actionId) + dropAction: (hub, args: { teamId: Team['id']; actionId: Action['id'] }) => { + return hub.eventsProcessor.actionManager.dropAction(args.teamId, args.actionId) }, teardownPlugins: async (hub) => { await teardownPlugins(hub) diff --git a/tests/postgres/worker.test.ts b/tests/postgres/worker.test.ts index 2b612a25..51534e87 100644 --- a/tests/postgres/worker.test.ts +++ b/tests/postgres/worker.test.ts @@ -320,15 +320,15 @@ describe('createTaskRunner()', () => { }) it('handles `reloadAction` task', async () => { - await taskRunner({ task: 'reloadAction', args: { actionId: 777 } }) + await taskRunner({ task: 'reloadAction', args: { teamId: 2, actionId: 777 } }) - expect(hub.eventsProcessor.actionManager.reloadAction).toHaveBeenCalledWith(777) + expect(hub.eventsProcessor.actionManager.reloadAction).toHaveBeenCalledWith(2, 777) }) it('handles `dropAction` task', async () => { - await taskRunner({ task: 'dropAction', args: { actionId: 777 } }) + await taskRunner({ task: 'dropAction', args: { teamId: 2, actionId: 777 } }) - expect(hub.eventsProcessor.actionManager.dropAction).toHaveBeenCalledWith(777) + expect(hub.eventsProcessor.actionManager.dropAction).toHaveBeenCalledWith(2, 777) }) it('handles `teardownPlugin` task', async () => { diff --git a/tests/utils.test.ts b/tests/utils.test.ts index 6e853bdc..7904d13c 100644 --- a/tests/utils.test.ts +++ b/tests/utils.test.ts @@ -8,6 +8,7 @@ import { getFileFromArchive, getFileFromTGZ, getFileFromZip, + groupBy, sanitizeSqlIdentifier, setLogLevel, UUID, @@ -320,3 +321,76 @@ describe('escapeClickHouseString', () => { expect(sanitizedString).toStrictEqual("insert\\'escape \\\\") }) }) + +describe('groupBy', () => { + it('groups simple objects', () => { + const objects = [ + { i: 2, foo: 'x' }, + { i: 2, foo: 'y' }, + { i: 4, foo: 'x' }, + { i: 7, foo: 'z' }, + ] + + const groupingByI = groupBy(objects, 'i') + expect(groupingByI).toEqual({ + 2: [ + { i: 2, foo: 'x' }, + { i: 2, foo: 'y' }, + ], + 4: [{ i: 4, foo: 'x' }], + 7: [{ i: 7, foo: 'z' }], + }) + + const groupingByFoo = groupBy(objects, 'foo') + expect(groupingByFoo).toEqual({ + x: [ + { i: 2, foo: 'x' }, + { i: 4, foo: 'x' }, + ], + y: [{ i: 2, foo: 'y' }], + z: [{ i: 7, foo: 'z' }], + }) + }) + + it('handles undefineds', () => { + const objects = [{ i: 2, foo: 'x' }, { i: 2, foo: 'y' }, { i: 4, foo: 'x' }, { foo: 'z' }] + + const groupingByI = groupBy(objects, 'i') + expect(groupingByI).toEqual({ + 2: [ + { i: 2, foo: 'x' }, + { i: 2, foo: 'y' }, + ], + 4: [{ i: 4, foo: 'x' }], + undefined: [{ foo: 'z' }], + }) + }) + + it('works in flat mode', () => { + const objects = [ + { i: 2, foo: 'x' }, + { i: 4, foo: 'x' }, + { i: 7, foo: 'z' }, + ] + + const groupingByI = groupBy(objects, 'i', true) + expect(groupingByI).toEqual({ + 2: { i: 2, foo: 'x' }, + 4: { i: 4, foo: 'x' }, + 7: { i: 7, foo: 'z' }, + }) + }) + + it("doesn't work in flat mode if multiple values match a single key", () => { + const objects = [ + { i: 2, foo: 'x' }, + { i: 2, foo: 'y' }, + { i: 4, foo: 'x' }, + { i: 7, foo: 'z' }, + ] + + expect(() => groupBy(objects, 'i', true)).toThrowError( + 'Key "i" has more than one matching value, which is not allowed in flat groupBy!' + ) + }) +}) diff --git a/tests/worker/ingestion/action-manager.test.ts b/tests/worker/ingestion/action-manager.test.ts index 42b7f866..91544e22 100644 --- a/tests/worker/ingestion/action-manager.test.ts +++ b/tests/worker/ingestion/action-manager.test.ts @@ -1,4 +1,4 @@ -import { Hub, PropertyOperator, RawAction } from '../../../src/types' +import { Hub, PropertyOperator } from '../../../src/types' import { createHub } from '../../../src/utils/db/hub' import { ActionManager } from '../../../src/worker/ingestion/action-manager' import { resetTestDatabase } from '../../helpers/sql' @@ -14,17 +14,20 @@ describe('ActionManager', () => { actionManager = new ActionManager(hub.db) await actionManager.prepare() }) + afterEach(async () => { await closeServer() }) - it('returns the correct action', async () => { - const ACTION_ID = 69 - const ACTION_STEP_ID = 913 + const TEAM_ID = 2 + const ACTION_ID = 69 + const ACTION_STEP_ID = 913 - const action = actionManager.getAction(ACTION_ID) + it('returns the correct actions generally', async () => { + const action = actionManager.getTeamActions(TEAM_ID) - expect(action).toMatchObject({ + expect(action.length).toEqual(1) + expect(action[0]).toMatchObject({ id: ACTION_ID, name: 'Test Action', deleted: false, @@ -55,11 +58,12 @@ describe('ActionManager', () => { ) // This is normally dispatched by Django and broadcasted by Piscina - await actionManager.reloadAction(ACTION_ID) + await actionManager.reloadAction(TEAM_ID, ACTION_ID) - const reloadedAction = actionManager.getAction(ACTION_ID) + const reloadedAction = actionManager.getTeamActions(TEAM_ID) - expect(reloadedAction).toMatchObject({ + expect(reloadedAction.length).toEqual(1) + expect(reloadedAction[0]).toMatchObject({ id: ACTION_ID, name: 'Test Action', deleted: false, @@ -84,10 +88,48 @@ describe('ActionManager', () => { }) // This is normally dispatched by Django and broadcasted by Piscina - actionManager.dropAction(ACTION_ID) + actionManager.dropAction(TEAM_ID, ACTION_ID) + + const droppedAction = actionManager.getTeamActions(TEAM_ID) + + expect(droppedAction).toEqual([]) + }) + + it('returns the correct actions when deleted = TRUE', async () => { + const action = actionManager.getTeamActions(TEAM_ID) + + expect(action.length).toEqual(1) + expect(action[0]).toMatchObject({ + id: ACTION_ID, + name: 'Test Action', + deleted: false, + post_to_slack: false, + slack_message_format: '', + is_calculating: false, + steps: [ + { + id: ACTION_STEP_ID, + action_id: ACTION_ID, + tag_name: null, + text: null, + href: null, + selector: null, + url: null, + url_matching: null, + name: null, + event: null, + properties: [{ type: 'event', operator: PropertyOperator.Exact, key: 'foo', value: ['bar'] }], + }, + ], + }) + + await hub.db.postgresQuery(`UPDATE posthog_action SET deleted = TRUE WHERE id = $1`, [ACTION_ID], 'testKey') + + // This is normally dispatched by Django and broadcasted by Piscina + await actionManager.reloadAction(TEAM_ID, ACTION_ID) - const droppedAction = actionManager.getAction(ACTION_ID) + const droppedAction = actionManager.getTeamActions(TEAM_ID) - expect(droppedAction).toBeUndefined() + expect(droppedAction).toEqual([]) }) }) From dfda86bac93d63a9fd0c639164dbca478588fc12 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Wed, 26 May 2021 23:05:03 +0200 Subject: [PATCH 2/5] Make `getTeamActions()` return type more versatile --- src/worker/ingestion/action-manager.ts | 7 ++++--- tests/worker/ingestion/action-manager.test.ts | 16 ++++++++-------- 2 files changed, 12 insertions(+), 11 deletions(-) diff --git a/src/worker/ingestion/action-manager.ts b/src/worker/ingestion/action-manager.ts index cd0d38b9..5e2aa97f 100644 --- a/src/worker/ingestion/action-manager.ts +++ b/src/worker/ingestion/action-manager.ts @@ -3,7 +3,8 @@ import { DB } from '../../utils/db/db' import { status } from '../../utils/status' import { groupBy } from '../../utils/utils' -type ActionCache = Record> +export type ActionMap = Record +type ActionCache = Record export class ActionManager { private ready: boolean @@ -21,11 +22,11 @@ export class ActionManager { this.ready = true } - public getTeamActions(teamId: Team['id']): Action[] { + public getTeamActions(teamId: Team['id']): ActionMap | null { if (!this.ready) { throw new Error('ActionManager is not ready! Run actionManager.prepare() before this') } - return Object.values(this.actionCache[teamId] ?? {}) + return this.actionCache[teamId] || null } public async reloadAllActions(): Promise { diff --git a/tests/worker/ingestion/action-manager.test.ts b/tests/worker/ingestion/action-manager.test.ts index 91544e22..360fc400 100644 --- a/tests/worker/ingestion/action-manager.test.ts +++ b/tests/worker/ingestion/action-manager.test.ts @@ -26,8 +26,8 @@ describe('ActionManager', () => { it('returns the correct actions generally', async () => { const action = actionManager.getTeamActions(TEAM_ID) - expect(action.length).toEqual(1) - expect(action[0]).toMatchObject({ + expect(Object.values(action!).length).toEqual(1) + expect(action![ACTION_ID]).toMatchObject({ id: ACTION_ID, name: 'Test Action', deleted: false, @@ -62,8 +62,8 @@ describe('ActionManager', () => { const reloadedAction = actionManager.getTeamActions(TEAM_ID) - expect(reloadedAction.length).toEqual(1) - expect(reloadedAction[0]).toMatchObject({ + expect(Object.values(action!).length).toEqual(1) + expect(reloadedAction![ACTION_ID]).toMatchObject({ id: ACTION_ID, name: 'Test Action', deleted: false, @@ -92,14 +92,14 @@ describe('ActionManager', () => { const droppedAction = actionManager.getTeamActions(TEAM_ID) - expect(droppedAction).toEqual([]) + expect(Object.values(droppedAction!).length).toEqual(0) }) it('returns the correct actions when deleted = TRUE', async () => { const action = actionManager.getTeamActions(TEAM_ID) - expect(action.length).toEqual(1) - expect(action[0]).toMatchObject({ + expect(Object.values(action!).length).toEqual(1) + expect(action![ACTION_ID]).toMatchObject({ id: ACTION_ID, name: 'Test Action', deleted: false, @@ -130,6 +130,6 @@ describe('ActionManager', () => { const droppedAction = actionManager.getTeamActions(TEAM_ID) - expect(droppedAction).toEqual([]) + expect(Object.values(droppedAction!).length).toEqual(0) }) }) From 11457d044536dcbb8af98f398cb857ab89a6f2b3 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Thu, 27 May 2021 11:00:07 +0200 Subject: [PATCH 3/5] Fix PubSub's lack of teamId --- src/main/pluginsServer.ts | 4 ++-- src/worker/ingestion/action-manager.ts | 12 ++++++------ 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/src/main/pluginsServer.ts b/src/main/pluginsServer.ts index 3fd90256..69dc7ba4 100644 --- a/src/main/pluginsServer.ts +++ b/src/main/pluginsServer.ts @@ -151,9 +151,9 @@ export async function startPluginsServer( await scheduleControl?.reloadSchedule() }, 'reload-action': async (message) => - await piscina?.broadcastTask({ task: 'reloadAction', args: { actionId: parseInt(message) } }), + await piscina?.broadcastTask({ task: 'reloadAction', args: JSON.parse(message) }), 'drop-action': async (message) => - await piscina?.broadcastTask({ task: 'dropAction', args: { actionId: parseInt(message) } }), + await piscina?.broadcastTask({ task: 'dropAction', args: JSON.parse(message) }), }) await pubSub.start() diff --git a/src/worker/ingestion/action-manager.ts b/src/worker/ingestion/action-manager.ts index 5e2aa97f..531154b2 100644 --- a/src/worker/ingestion/action-manager.ts +++ b/src/worker/ingestion/action-manager.ts @@ -46,20 +46,20 @@ export class ActionManager { status.info( '🍿', wasCachedAlready - ? `Refetched action ID ${actionId} from DB` - : `Fetched new action ID ${actionId} from DB` + ? `Refetched action ID ${actionId} (team ID ${teamId}) from DB` + : `Fetched new action ID ${actionId} (team ID ${teamId}) from DB` ) this.actionCache[teamId][actionId] = refetchedAction } else if (wasCachedAlready) { status.info( '🍿', - `Tried to fetch action ID ${actionId} from DB, but it wasn't found in DB, so deleted from cache instead` + `Tried to fetch action ID ${actionId} (team ID ${teamId}) from DB, but it wasn't found in DB, so deleted from cache instead` ) delete this.actionCache[teamId][actionId] } else { status.info( '🍿', - `Tried to fetch action ID ${actionId} from DB, but it wasn't found in DB or cache, so did nothing instead` + `Tried to fetch action ID ${actionId} (team ID ${teamId}) from DB, but it wasn't found in DB or cache, so did nothing instead` ) } } @@ -67,12 +67,12 @@ export class ActionManager { public dropAction(teamId: Team['id'], actionId: Action['id']): void { const wasCachedAlready = teamId in this.actionCache && actionId in this.actionCache[teamId] if (wasCachedAlready) { - status.info('🍿', `Deleted action ID ${actionId} from cache`) + status.info('🍿', `Deleted action ID ${actionId} (team ID ${teamId}) from cache`) delete this.actionCache[teamId][actionId] } else { status.info( '🍿', - `Tried to delete action ID ${actionId} from cache, but it wasn't found in cache, so did nothing instead` + `Tried to delete action ID ${actionId} (team ID ${teamId}) from cache, but it wasn't found in cache, so did nothing instead` ) } } From 6f91d60b44486e24eae6731fbf6831a6e2066252 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Thu, 27 May 2021 11:39:21 +0200 Subject: [PATCH 4/5] Add DB.fetchAllActionsGroupedByTeam --- src/utils/db/db.ts | 11 +++++++++++ src/worker/ingestion/action-manager.ts | 8 +------- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/src/utils/db/db.ts b/src/utils/db/db.ts index fe3a505f..dc1b4d76 100644 --- a/src/utils/db/db.ts +++ b/src/utils/db/db.ts @@ -32,6 +32,7 @@ import { RawOrganization, RawPerson, SessionRecordingEvent, + Team, TimestampFormat, } from '../../types' import { instrumentQuery } from '../metrics' @@ -39,6 +40,7 @@ import { castTimestampOrNow, clickHouseTimestampToISO, escapeClickHouseString, + groupBy, sanitizeSqlIdentifier, tryTwice, UUID, @@ -750,6 +752,15 @@ export class DB { return Object.values(actionsMap) } + public async fetchAllActionsGroupedByTeam(): Promise>> { + return Object.fromEntries( + Object.entries(groupBy(await this.fetchAllActions(), 'team_id')).map(([teamId, actions]) => [ + teamId, + groupBy(actions, 'id', true), + ]) + ) + } + public async fetchAction(id: Action['id']): Promise { const rawActions: RawAction[] = ( await this.postgresQuery( diff --git a/src/worker/ingestion/action-manager.ts b/src/worker/ingestion/action-manager.ts index 531154b2..f11820ca 100644 --- a/src/worker/ingestion/action-manager.ts +++ b/src/worker/ingestion/action-manager.ts @@ -1,7 +1,6 @@ import { Action, Team } from '../../types' import { DB } from '../../utils/db/db' import { status } from '../../utils/status' -import { groupBy } from '../../utils/utils' export type ActionMap = Record type ActionCache = Record @@ -30,12 +29,7 @@ export class ActionManager { } public async reloadAllActions(): Promise { - this.actionCache = Object.fromEntries( - Object.entries(groupBy(await this.db.fetchAllActions(), 'team_id')).map(([teamId, actions]) => [ - teamId, - groupBy(actions, 'id', true), - ]) - ) + this.actionCache = await this.db.fetchAllActionsGroupedByTeam() status.info('🍿', 'Fetched all actions from DB anew') } From afb45fcc01aaece279f2dcb99e0b01e81a99d880 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Thu, 27 May 2021 12:30:47 +0200 Subject: [PATCH 5/5] Address feedback --- src/utils/db/db.ts | 32 +++++++++------------- tests/shared/db.test.ts | 59 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 19 deletions(-) create mode 100644 tests/shared/db.test.ts diff --git a/src/utils/db/db.ts b/src/utils/db/db.ts index dc1b4d76..c5177768 100644 --- a/src/utils/db/db.ts +++ b/src/utils/db/db.ts @@ -727,38 +727,32 @@ export class DB { // Action & ActionStep - public async fetchAllActions(): Promise { + public async fetchAllActionsGroupedByTeam(): Promise>> { const rawActions: RawAction[] = ( await this.postgresQuery(`SELECT * FROM posthog_action WHERE deleted = FALSE`, undefined, 'fetchActions') ).rows - const actionSteps: ActionStep[] = ( + const actionSteps: (ActionStep & { team_id: Team['id'] })[] = ( await this.postgresQuery( - `SELECT posthog_actionstep.*, posthog_action.deleted FROM posthog_actionstep - JOIN posthog_action ON (posthog_action.id = posthog_actionstep.action_id) - WHERE posthog_action.deleted = FALSE`, + `SELECT posthog_actionstep.*, posthog_action.team_id + FROM posthog_actionstep JOIN posthog_action ON (posthog_action.id = posthog_actionstep.action_id) + WHERE posthog_action.deleted = FALSE`, undefined, 'fetchActionSteps' ) ).rows - const actionsMap: Record = {} + const actions: Record> = {} for (const rawAction of rawActions) { - actionsMap[rawAction.id] = { ...rawAction, steps: [] } + if (!actions[rawAction.team_id]) { + actions[rawAction.team_id] = {} + } + actions[rawAction.team_id][rawAction.id] = { ...rawAction, steps: [] } } for (const actionStep of actionSteps) { - if (actionStep.action_id in actionsMap) { - actionsMap[actionStep.action_id].steps.push(actionStep) + if (actions[actionStep.team_id]?.[actionStep.action_id]) { + actions[actionStep.team_id][actionStep.action_id].steps.push(actionStep) } } - return Object.values(actionsMap) - } - - public async fetchAllActionsGroupedByTeam(): Promise>> { - return Object.fromEntries( - Object.entries(groupBy(await this.fetchAllActions(), 'team_id')).map(([teamId, actions]) => [ - teamId, - groupBy(actions, 'id', true), - ]) - ) + return actions } public async fetchAction(id: Action['id']): Promise { diff --git a/tests/shared/db.test.ts b/tests/shared/db.test.ts new file mode 100644 index 00000000..55534113 --- /dev/null +++ b/tests/shared/db.test.ts @@ -0,0 +1,59 @@ +import { Hub, PropertyOperator } from '../../src/types' +import { DB } from '../../src/utils/db/db' +import { createHub } from '../../src/utils/db/hub' +import { ActionManager } from '../../src/worker/ingestion/action-manager' +import { resetTestDatabase } from '../helpers/sql' + +describe('DB', () => { + let hub: Hub + let closeServer: () => Promise + let db: DB + + beforeEach(async () => { + ;[hub, closeServer] = await createHub() + await resetTestDatabase() + db = hub.db + }) + + afterEach(async () => { + await closeServer() + }) + + const TEAM_ID = 2 + const ACTION_ID = 69 + const ACTION_STEP_ID = 913 + + test('fetchAllActionsGroupedByTeam', async () => { + const action = await db.fetchAllActionsGroupedByTeam() + + expect(action).toMatchObject({ + [TEAM_ID]: { + [ACTION_ID]: { + id: ACTION_ID, + name: 'Test Action', + deleted: false, + post_to_slack: false, + slack_message_format: '', + is_calculating: false, + steps: [ + { + id: ACTION_STEP_ID, + action_id: ACTION_ID, + tag_name: null, + text: null, + href: null, + selector: null, + url: null, + url_matching: null, + name: null, + event: null, + properties: [ + { type: 'event', operator: PropertyOperator.Exact, key: 'foo', value: ['bar'] }, + ], + }, + ], + }, + }, + }) + }) +})