asyavee/medusa-store
1
0 Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

chore(): Emit events in batch and index process event ids in batch (#12097)

Browse Source 
**What**
First iteration to prevent events from overwhelming the systems.
- Group emitted event ids when possible instead of creating a message per id which leads to reduced amount of events to process massively in cases of import for example
- Update the index engine to process event data in batches of 100
- Update event handling by the index engine to be able to upsert by batch as well
- Fix index engine build config for intermediate listeners inferrence
This commit is contained in:
Adrien de Peretti authored and GitHub committed 2025-04-08 16:57:08 +00:00
1 parent b05807bfc1
commit 74381addc3
21 files changed
+548 -463
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/modules/index/integration-tests/__tests__/data-synchronizer.spec.ts
+8 -7
View File
@@ -19,7 +19,7 @@ import { asValue } from "awilix"
import * as path from "path"
import { setTimeout } from "timers/promises"
import { EventBusServiceMock } from "../__fixtures__"
import { dbName } from "../__fixtures__/medusa-config"
import config, { dbName } from "../__fixtures__/medusa-config"
const eventBusMock = new EventBusServiceMock()
const queryMock = {
@@ -147,14 +147,15 @@ describe("DataSynchronizer", () => {
// Mock query response for products
queryMock.graph.mockImplementation(async (config) => {
if (Array.isArray(config.filters.id)) {
let data: any[] = []
if (config.filters.id.includes(testProductId)) {
return {
data: [mockData[0]],
}
data.push(mockData[0])
} else if (config.filters.id.includes(testProductId2)) {
return {
data: [mockData[1]],
}
data.push(mockData[1])
}
return {
data,
}
}
packages/modules/index/integration-tests/__tests__/index-engine-module.spec.ts
+6 -6
View File
@@ -141,6 +141,7 @@ const beforeAll_ = async () => {
;(index as any).eventBusModuleService_ = eventBusMock
await globalApp.onApplicationStart()
await setTimeout(3000)
;(index as any).storageProvider_.query_ = queryMock
return globalApp
@@ -196,6 +197,8 @@ describe("IndexModuleService", function () {
await dbUtils.shutdown(dbName)
})
afterEach(afterEach_)
describe("on created or attached events", function () {
let manager
@@ -222,7 +225,7 @@ describe("IndexModuleService", function () {
},
},
{
name: "PriceSet.created",
name: "pricing.price-set.created",
data: {
id: priceSetId,
},
@@ -247,7 +250,6 @@ describe("IndexModuleService", function () {
]
beforeEach(async () => {
await setTimeout(1000)
await beforeEach_(eventDataToEmit)
manager = (medusaApp.sharedContainer!.resolve(Modules.INDEX) as any)
@@ -392,7 +394,7 @@ describe("IndexModuleService", function () {
},
},
{
name: "PriceSet.created",
name: "pricing.price-set.created",
data: {
id: priceSetId,
},
@@ -689,7 +691,7 @@ describe("IndexModuleService", function () {
},
},
{
name: "PriceSet.created",
name: "pricing.price-set.created",
data: {
id: priceSetId,
},
@@ -761,8 +763,6 @@ describe("IndexModuleService", function () {
await eventBusMock.emit(deleteEventDataToEmit)
})
afterEach(afterEach_)
it("should consume all deleted events and delete the index entries", async () => {
const indexEntries = await manager.find(toMikroORMEntity(IndexData), {})
const indexRelationEntries = await manager.find(
packages/modules/index/src/services/data-synchronizer.ts
+4 -2
View File
@@ -290,7 +290,7 @@ export class DataSynchronizer {
filters["updated_at"] = { $gt: pagination.updated_at }
}
const { data } = await this.#query.graph({
const queryResult = await this.#query.graph({
entity: alias,
fields: [entityPrimaryKey],
filters,
@@ -302,10 +302,12 @@ export class DataSynchronizer {
},
})
if (!data.length) {
if (!queryResult?.data?.length) {
break
}
const data = queryResult.data
const envelop: Event = {
data,
name: !isLink
packages/modules/index/src/services/postgres-provider.ts
+168 -157
View File
@@ -163,9 +163,11 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
}
}
protected static parseMessageData<T>(message?: Event): {
protected static parseMessageData<TData extends { id: string | string[] }>(
message?: Event
): {
action: string
data: { id: string }[]
data: TData[]
ids: string[]
} | void {
const isExpectedFormat =
@@ -177,7 +179,7 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
const result: {
action: string
data: { id: string }[]
data: TData[]
ids: string[]
} = {
action: "",
@@ -186,60 +188,76 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
}
result.action = message!.metadata!.action as string
result.data = message!.data as { id: string }[]
result.data = message!.data as TData[]
result.data = Array.isArray(result.data) ? result.data : [result.data]
result.ids = result.data.map((d) => d.id)
result.ids = result.data.flatMap((d) =>
Array.isArray(d.id) ? d.id : [d.id]
)
return result
}
consumeEvent(
schemaEntityObjectRepresentation: IndexTypes.SchemaObjectEntityRepresentation
): Subscriber<{ id: string }> {
return async (data: Event) => {
): Subscriber<{ id: string | string[] }> {
return async (event: Event) => {
await this.#isReady_
const data_: { id: string }[] = Array.isArray(data.data)
? data.data
: [data.data]
let ids: string[] = data_.map((d) => d.id)
let action = data.name.split(".").pop() || ""
const data_: { id: string }[] = Array.isArray(event.data)
? event.data
: [event.data]
let ids: string[] = data_.flatMap((d) =>
Array.isArray(d.id) ? d.id : [d.id]
)
let action = event.name.split(".").pop() || ""
const parsedMessage = PostgresProvider.parseMessageData(data)
const parsedMessage = PostgresProvider.parseMessageData(event)
if (parsedMessage) {
action = parsedMessage.action
ids = parsedMessage.ids
}
const { fields, alias } = schemaEntityObjectRepresentation
const graphConfig: Parameters<QueryGraphFunction>[0] = {
entity: alias,
filters: {
id: ids,
},
fields: [...new Set(["id", ...fields])],
}
if (action === CommonEvents.DELETED || action === CommonEvents.DETACHED) {
graphConfig.withDeleted = true
}
const { data: entityData } = await this.query_.graph(graphConfig)
const argument = {
entity: schemaEntityObjectRepresentation.entity,
data: entityData,
schemaEntityObjectRepresentation,
}
const targetMethod = this.eventActionToMethodMap_[action]
if (!targetMethod) {
return
}
await this[targetMethod](argument)
const { fields, alias } = schemaEntityObjectRepresentation
let withDeleted: boolean | undefined
if (action === CommonEvents.DELETED || action === CommonEvents.DETACHED) {
withDeleted = true
}
// Process ids in batches of 100
const batchSize = 100
const idsBatches: string[][] = []
for (let i = 0; i < ids.length; i += batchSize) {
idsBatches.push(ids.slice(i, i + batchSize))
}
for (const idsBatch of idsBatches) {
const graphConfig: Parameters<QueryGraphFunction>[0] = {
entity: alias,
filters: {
id: idsBatch,
},
fields: [...new Set(["id", ...fields])],
withDeleted,
}
const { data: entityData } = await this.query_.graph(graphConfig)
const argument = {
entity: schemaEntityObjectRepresentation.entity,
data: entityData,
schemaEntityObjectRepresentation,
}
await this[targetMethod](argument)
}
}
}
@@ -374,7 +392,9 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
@MedusaContext() sharedContext: Context<SqlEntityManager> = {}
) {
const { transactionManager: em } = sharedContext
const indexRepository = em!.getRepository(toMikroORMEntity(IndexData))
const indexRepository = em!.getRepository(
toMikroORMEntity(IndexData)
) as EntityRepository<any>
const indexRelationRepository: EntityRepository<any> = em!.getRepository(
toMikroORMEntity(IndexRelation)
)
@@ -385,33 +405,32 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
parentsProperties,
} = PostgresProvider.parseData(data, schemaEntityObjectRepresentation)
/**
* Clean the entity data to only keep the properties that are defined in the schema
*/
const cleanedData = data_.map((entityData) => {
return entityProperties.reduce((acc, property) => {
acc[property] = entityData[property]
return acc
}, {}) as TData
})
/**
* Loop through the data and create index entries for each entity as well as the
* index relation entries if the entity has parents
*/
for (const entityData of data_) {
/**
* Clean the entity data to only keep the properties that are defined in the schema
*/
const entitiesToUpsert: Set<string> = new Set()
const relationsToUpsert: Set<string> = new Set()
const cleanedEntityData = entityProperties.reduce((acc, property) => {
acc[property] = entityData[property]
return acc
}, {}) as TData
await indexRepository.upsert(
{
id: cleanedEntityData.id,
cleanedData.forEach((entityData, index) => {
entitiesToUpsert.add(
JSON.stringify({
id: entityData.id,
name: entity,
data: cleanedEntityData,
data: entityData,
staled_at: null,
},
{
onConflictAction: "merge",
onConflictFields: ["id", "name"],
onConflictMergeFields: ["data", "staled_at"],
}
})
)
/**
@@ -422,7 +441,7 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
parentsProperties
)) {
const parentAlias = parentProperties[0].split(".")[0]
const parentData = entityData[parentAlias] as TData[]
const parentData = data_[index][parentAlias] as TData[]
if (!parentData) {
continue
@@ -433,43 +452,46 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
: [parentData]
for (const parentData_ of parentDataCollection) {
await indexRepository.upsert(
{
id: (parentData_ as any).id,
name: parentEntity,
data: parentData_,
staled_at: null,
},
{
onConflictAction: "merge",
onConflictFields: ["id", "name"],
onConflictMergeFields: ["staled_at"],
}
)
await indexRelationRepository.upsert(
{
parent_id: (parentData_ as any).id,
relationsToUpsert.add(
JSON.stringify({
parent_id: parentData_.id,
parent_name: parentEntity,
child_id: cleanedEntityData.id,
child_id: entityData.id,
child_name: entity,
pivot: `${parentEntity}-${entity}`,
staled_at: null,
},
{
onConflictAction: "merge",
onConflictFields: [
"pivot",
"parent_id",
"child_id",
"parent_name",
"child_name",
],
onConflictMergeFields: ["staled_at"],
}
})
)
}
}
})
if (entitiesToUpsert.size) {
await indexRepository.upsertMany(
Array.from(entitiesToUpsert).map((entity) => JSON.parse(entity)),
{
onConflictAction: "merge",
onConflictFields: ["id", "name"],
onConflictMergeFields: ["data", "staled_at"],
}
)
}
if (relationsToUpsert.size) {
await indexRelationRepository.upsertMany(
Array.from(relationsToUpsert).map((relation) => JSON.parse(relation)),
{
onConflictAction: "merge",
onConflictFields: [
"pivot",
"parent_id",
"child_id",
"parent_name",
"child_name",
],
onConflictMergeFields: ["staled_at"],
}
)
}
}
@@ -495,7 +517,9 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
@MedusaContext() sharedContext: Context<SqlEntityManager> = {}
) {
const { transactionManager: em } = sharedContext
const indexRepository = em!.getRepository(toMikroORMEntity(IndexData))
const indexRepository = em!.getRepository(
toMikroORMEntity(IndexData)
) as EntityRepository<any>
const { data: data_, entityProperties } = PostgresProvider.parseData(
data,
@@ -503,24 +527,22 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
)
await indexRepository.upsertMany(
data_.map(
(entityData) => {
return {
id: entityData.id,
name: entity,
data: entityProperties.reduce((acc, property) => {
acc[property] = entityData[property]
return acc
}, {}),
staled_at: null,
}
},
{
onConflictAction: "merge",
onConflictFields: ["id", "name"],
onConflictMergeFields: ["data", "staled_at"],
data_.map((entityData) => {
return {
id: entityData.id,
name: entity,
data: entityProperties.reduce((acc, property) => {
acc[property] = entityData[property]
return acc
}, {}),
staled_at: null,
}
)
}),
{
onConflictAction: "merge",
onConflictFields: ["id", "name"],
onConflictMergeFields: ["data", "staled_at"],
}
)
}
@@ -601,7 +623,7 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
const indexRepository = em!.getRepository(toMikroORMEntity(IndexData))
const indexRelationRepository = em!.getRepository(
toMikroORMEntity(IndexRelation)
)
) as EntityRepository<any>
const { data: data_, entityProperties } = PostgresProvider.parseData(
data,
@@ -652,77 +674,66 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
)
}
for (const entityData of data_) {
/**
* Clean the link entity data to only keep the properties that are defined in the schema
*/
/**
* Clean the link entity data to only keep the properties that are defined in the schema
*/
const cleanedEntityData = entityProperties.reduce((acc, property) => {
const cleanedData = data_.map((entityData) => {
return entityProperties.reduce((acc, property) => {
acc[property] = entityData[property]
return acc
}, {}) as TData
})
await indexRepository.upsert(
{
id: cleanedEntityData.id,
name: entity,
data: cleanedEntityData,
staled_at: null,
},
{
onConflictAction: "merge",
onConflictFields: ["id", "name"],
onConflictMergeFields: ["data", "staled_at"],
}
)
/**
* Create the index relation entries for the parent entity and the child entity
*/
await indexRelationRepository.upsert(
let relationsToUpsert: any[] = []
const entitiesToUpdate = cleanedData.map((entityData) => {
relationsToUpsert.push(
{
parent_id: entityData[parentPropertyId] as string,
parent_name: parentEntityName,
child_id: cleanedEntityData.id,
child_id: entityData.id,
child_name: entity,
pivot: `${parentEntityName}-${entity}`,
staled_at: null,
},
{
onConflictAction: "merge",
onConflictFields: [
"pivot",
"parent_id",
"child_id",
"parent_name",
"child_name",
],
onConflictMergeFields: ["staled_at"],
}
)
await indexRelationRepository.upsert(
{
parent_id: cleanedEntityData.id,
parent_id: entityData.id,
parent_name: entity,
child_id: entityData[childPropertyId] as string,
child_name: childEntityName,
pivot: `${entity}-${childEntityName}`,
staled_at: null,
},
{
onConflictAction: "merge",
onConflictFields: [
"pivot",
"parent_id",
"child_id",
"parent_name",
"child_name",
],
onConflictMergeFields: ["staled_at"],
}
)
return {
id: entityData.id,
name: entity,
data: entityData,
staled_at: null,
}
})
if (entitiesToUpdate.length) {
await indexRepository.upsertMany(entitiesToUpdate, {
onConflictAction: "merge",
onConflictFields: ["id", "name"],
onConflictMergeFields: ["data", "staled_at"],
})
}
if (relationsToUpsert.length) {
await indexRelationRepository.upsertMany(relationsToUpsert, {
onConflictAction: "merge",
onConflictFields: [
"pivot",
"parent_id",
"child_id",
"parent_name",
"child_name",
],
onConflictMergeFields: ["staled_at"],
})
}
}
packages/modules/index/src/utils/build-config.ts
+21 -3
View File
@@ -5,7 +5,11 @@ import {
ModuleJoinerConfig,
ModuleJoinerRelationship,
} from "@medusajs/framework/types"
import { CommonEvents, GraphQLUtils } from "@medusajs/framework/utils"
import {
buildModuleResourceEventName,
CommonEvents,
GraphQLUtils,
} from "@medusajs/framework/utils"
import { schemaObjectRepresentationPropertiesToOmit } from "@types"
export const CustomDirectives = {
@@ -561,9 +565,23 @@ function processEntity(
intermediateEntityObjectRepresentationRef.alias =
intermediateEntityAlias
intermediateEntityObjectRepresentationRef.listeners = [
intermediateEntityName + "." + CommonEvents.CREATED,
intermediateEntityName + "." + CommonEvents.UPDATED,
buildModuleResourceEventName({
action: CommonEvents.CREATED,
objectName: intermediateEntityName,
prefix: intermediateEntityModule.serviceName,
}),
buildModuleResourceEventName({
action: CommonEvents.UPDATED,
objectName: intermediateEntityName,
prefix: intermediateEntityModule.serviceName,
}),
buildModuleResourceEventName({
action: CommonEvents.DELETED,
objectName: intermediateEntityName,
prefix: intermediateEntityModule.serviceName,
}),
]
intermediateEntityObjectRepresentationRef.moduleConfig =
intermediateEntityModule