From bf4383e61673977a3ddac00c342e890361e0cd36 Mon Sep 17 00:00:00 2001 From: Wilson Rivera Date: Fri, 29 May 2026 16:40:25 -0400 Subject: [PATCH 01/12] feat: add span for `composeGraphsInWorker` --- .../core/composition/composeGraphs.pool.ts | 20 +++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/controlplane/src/core/composition/composeGraphs.pool.ts b/controlplane/src/core/composition/composeGraphs.pool.ts index 5d3ada1075..6b4db74baa 100644 --- a/controlplane/src/core/composition/composeGraphs.pool.ts +++ b/controlplane/src/core/composition/composeGraphs.pool.ts @@ -12,6 +12,7 @@ import { availableParallelism } from 'node:os'; import { Warning } from '@wundergraph/composition'; import { RouterConfig } from '@wundergraph/cosmo-connect/dist/node/v1/node_pb'; import WorkerPool from 'tinypool'; +import * as Sentry from '@sentry/node'; import { FederatedGraphDTO } from '../../types/index.js'; import { validateRouterCompatibilityVersion } from './composition.js'; import { ComposedFederatedGraph, CompositionSubgraphRecord } from './composer.js'; @@ -111,12 +112,27 @@ export function deserializeRouterExecutionConfig(routerExecutionConfigJson?: Ret return RouterConfig.fromJson(routerExecutionConfigJson); } -export function composeGraphsInWorker(task: Omit) { +export function composeGraphsInWorker( + task: Omit, +): Promise { const fullTask: ComposeGraphsTaskInput = { ...task, routerCompatibilityVersion: validateRouterCompatibilityVersion(task.federatedGraph.routerCompatibilityVersion), }; - return getComposeGraphsPool().run(fullTask) as Promise; + + return Sentry.startSpan( + { + name: 'composeGraphsInWorker', + attributes: { + federatedGraphId: task.federatedGraph.id, + federatedGraphName: task.federatedGraph.name, + subgraphsCount: task.federatedGraph.subgraphsCount, + organizationId: task.federatedGraph.organizationId, + namespaceId: task.federatedGraph.namespaceId, + }, + }, + () => getComposeGraphsPool().run(fullTask) as Promise, + ); } export function configureComposeGraphsPool(options: ConfigureComposeGraphsPoolOptions) { From d1e184e82db04d607c2122322e22543a8c4f8872 Mon Sep 17 00:00:00 2001 From: Wilson Rivera Date: Fri, 29 May 2026 18:12:37 -0400 Subject: [PATCH 02/12] feat: improve queries and remove double transactions --- .../repositories/FeatureFlagRepository.ts | 180 ++++--- .../repositories/FederatedGraphRepository.ts | 509 +++--------------- .../GraphCompositionRepository.ts | 224 ++++---- .../core/repositories/SubgraphRepository.ts | 37 +- 4 files changed, 311 insertions(+), 639 deletions(-) diff --git a/controlplane/src/core/repositories/FeatureFlagRepository.ts b/controlplane/src/core/repositories/FeatureFlagRepository.ts index c8a797a428..9d9cd3d0e9 100644 --- a/controlplane/src/core/repositories/FeatureFlagRepository.ts +++ b/controlplane/src/core/repositories/FeatureFlagRepository.ts @@ -19,6 +19,8 @@ import { subgraphsToFederatedGraph, targets, users, + protobufSchemaVersions, + pluginImageVersions, } from '../../db/schema.js'; import { FeatureFlagCompositionDTO, @@ -112,7 +114,7 @@ export class FeatureFlagRepository { }); } - public updateFeatureFlag({ + public async updateFeatureFlag({ featureFlag, labels, featureSubgraphIds, @@ -124,33 +126,31 @@ export class FeatureFlagRepository { unsetLabels: boolean; }) { const uniqueLabels = normalizeLabels(labels); - return this.db.transaction(async (tx) => { - if (labels.length > 0 || unsetLabels) { - const newLabels = unsetLabels ? [] : uniqueLabels; - await tx - .update(featureFlags) - .set({ - labels: newLabels.map((ul) => joinLabel(ul)), - }) - .where(and(eq(featureFlags.id, featureFlag.id), eq(featureFlags.organizationId, this.organizationId))) - .execute(); - } + if (labels.length > 0 || unsetLabels) { + const newLabels = unsetLabels ? [] : uniqueLabels; + await this.db + .update(featureFlags) + .set({ + labels: newLabels.map((ul) => joinLabel(ul)), + }) + .where(and(eq(featureFlags.id, featureFlag.id), eq(featureFlags.organizationId, this.organizationId))) + .execute(); + } - if (featureSubgraphIds.length > 0) { - // delete all the feature subgraphs of the feature flag - await tx - .delete(featureFlagToFeatureSubgraphs) - .where(eq(featureFlagToFeatureSubgraphs.featureFlagId, featureFlag.id)) - .execute(); + if (featureSubgraphIds.length > 0) { + // delete all the feature subgraphs of the feature flag + await this.db + .delete(featureFlagToFeatureSubgraphs) + .where(eq(featureFlagToFeatureSubgraphs.featureFlagId, featureFlag.id)) + .execute(); - await tx.insert(featureFlagToFeatureSubgraphs).values( - featureSubgraphIds.map((featureSubgraphId) => ({ - featureFlagId: featureFlag.id, - featureSubgraphId, - })), - ); - } - }); + await this.db.insert(featureFlagToFeatureSubgraphs).values( + featureSubgraphIds.map((featureSubgraphId) => ({ + featureFlagId: featureFlag.id, + featureSubgraphId, + })), + ); + } } public async enableFeatureFlag({ @@ -908,8 +908,7 @@ export class FeatureFlagRepository { featureFlagId: string; namespaceId: string; }): Promise { - const subgraphRepo = new SubgraphRepository(this.logger, this.db, this.organizationId); - const fgs = await this.db + const featureSubgraphs = await this.db .select({ name: targets.name, labels: targets.labels, @@ -929,6 +928,18 @@ export class FeatureFlagRepository { isEventDrivenGraph: subgraphs.isEventDrivenGraph, isFeatureSubgraph: subgraphs.isFeatureSubgraph, type: subgraphs.type, + // Schema Version + svId: schemaVersion.id, + svLastUpdated: schemaVersion.createdAt, + svSchemaSDL: schemaVersion.schemaSDL, + svIsV2Graph: schemaVersion.isV2Graph, + // Proto + protoSchema: protobufSchemaVersions.protoSchema, + protoMappings: protobufSchemaVersions.protoMappings, + protoLock: protobufSchemaVersions.protoLock, + // Plugin Data + pluginDataPlatforms: pluginImageVersions.platform, + pluginDataVersion: pluginImageVersions.version, }) .from(featureFlagToFeatureSubgraphs) .innerJoin( @@ -938,6 +949,18 @@ export class FeatureFlagRepository { .innerJoin(subgraphs, eq(subgraphs.id, featureSubgraphsToBaseSubgraphs.featureSubgraphId)) .innerJoin(targets, eq(subgraphs.targetId, targets.id)) .innerJoin(namespaces, eq(namespaces.id, targets.namespaceId)) + .leftJoin(schemaVersion, eq(schemaVersion.id, subgraphs.schemaVersionId)) + .leftJoin( + protobufSchemaVersions, + and( + inArray(schema.subgraphs.type, ['grpc_plugin', 'grpc_service']), + eq(protobufSchemaVersions.schemaVersionId, subgraphs.schemaVersionId), + ), + ) + .leftJoin( + pluginImageVersions, + and(eq(subgraphs.type, 'grpc_plugin'), eq(pluginImageVersions.schemaVersionId, subgraphs.schemaVersionId)), + ) .where( and( eq(featureFlagToFeatureSubgraphs.featureFlagId, featureFlagId), @@ -948,77 +971,64 @@ export class FeatureFlagRepository { ) .execute(); - const featureGraphsByFlag = []; + if (featureSubgraphs.length === 0) { + return []; + } + + const subgraphRepo = new SubgraphRepository(this.logger, this.db, this.organizationId); + const baseSubgraphNameById = await subgraphRepo.getSubgraphNameByIds( + featureSubgraphs.map((fg) => fg.baseSubgraphId), + ); - for (const fg of fgs) { - let lastUpdatedAt = ''; - let schemaSDL = ''; - let schemaVersionId = ''; - let isV2Graph: boolean | undefined; + const featureGraphsByFlag: FeatureSubgraphDTO[] = []; + for (const graph of featureSubgraphs) { let proto: ProtoSubgraph | undefined; + if (graph.schemaVersionId !== null && (graph.type === 'grpc_plugin' || graph.type === 'grpc_service')) { + if (!graph.protoSchema) { + this.logger.warn( + `Missing protobuf schema for ${graph.type} subgraph with schemaVersionId: ${graph.schemaVersionId}`, + ); + } - if (fg.schemaVersionId !== null) { - const sv = await this.db.query.schemaVersion.findFirst({ - where: eq(schemaVersion.id, fg.schemaVersionId), - }); - lastUpdatedAt = sv?.createdAt?.toISOString() ?? ''; - schemaSDL = sv?.schemaSDL ?? ''; - schemaVersionId = sv?.id ?? ''; - isV2Graph = sv?.isV2Graph || undefined; - if (fg.type === 'grpc_plugin' || fg.type === 'grpc_service') { - const protobufSchemaVersion = await this.db.query.protobufSchemaVersions.findFirst({ - where: eq(schema.protobufSchemaVersions.schemaVersionId, fg.schemaVersionId), - }); - - if (!protobufSchemaVersion) { + proto = { + schema: graph.protoSchema ?? '', + mappings: graph.protoMappings ?? '', + lock: graph.protoLock ?? '', + }; + + if (graph.type === 'grpc_plugin') { + if (!graph.pluginDataVersion) { this.logger.warn( - `Missing protobuf schema for ${fg.type} subgraph with schemaVersionId: ${fg.schemaVersionId}`, + `Missing plugin image version for ${graph.type} subgraph with schemaVersionId: ${graph.schemaVersionId}`, ); } - proto = { - schema: protobufSchemaVersion?.protoSchema ?? '', - mappings: protobufSchemaVersion?.protoMappings ?? '', - lock: protobufSchemaVersion?.protoLock ?? '', + proto.pluginData = { + platforms: graph.pluginDataPlatforms ?? [], + version: graph.pluginDataVersion ?? 'v1', }; - - if (fg.type === 'grpc_plugin') { - const pluginImageVersion = await this.db.query.pluginImageVersions.findFirst({ - where: eq(schema.pluginImageVersions.schemaVersionId, fg.schemaVersionId), - }); - - if (!pluginImageVersion) { - this.logger.warn( - `Missing plugin image version for ${fg.type} subgraph with schemaVersionId: ${fg.schemaVersionId}`, - ); - } - - proto.pluginData = { - platforms: pluginImageVersion?.platform ?? [], - version: pluginImageVersion?.version ?? 'v1', - }; - } } } - const baseSubgraph = await subgraphRepo.byId(fg.baseSubgraphId); - if (!baseSubgraph) { + const baseSubgraphName = baseSubgraphNameById.get(graph.baseSubgraphId); + if (!baseSubgraphName) { continue; } + featureGraphsByFlag.push({ - ...fg, - readme: fg.readme || undefined, - subscriptionUrl: fg.subscriptionUrl ?? '', - subscriptionProtocol: fg.subscriptionProtocol ?? 'ws', - websocketSubprotocol: fg.websocketSubprotocol || undefined, - creatorUserId: fg.createdBy || undefined, - labels: fg.labels?.map?.((l) => splitLabel(l)) ?? [], - namespace: fg.namespaceName, - schemaVersionId, - schemaSDL, - lastUpdatedAt, - baseSubgraphName: baseSubgraph.name, - isV2Graph, + ...graph, + readme: graph.readme || undefined, + subscriptionUrl: graph.subscriptionUrl ?? '', + subscriptionProtocol: graph.subscriptionProtocol ?? 'ws', + websocketSubprotocol: graph.websocketSubprotocol || undefined, + creatorUserId: graph.createdBy || undefined, + labels: graph.labels?.map?.((l) => splitLabel(l)) ?? [], + namespace: graph.namespaceName, + schemaVersionId: graph.svId ?? '', + schemaSDL: graph.svSchemaSDL ?? '', + lastUpdatedAt: graph.svLastUpdated?.toISOString() ?? '', + baseSubgraphName, + isV2Graph: graph.svIsV2Graph ?? undefined, proto, }); } diff --git a/controlplane/src/core/repositories/FederatedGraphRepository.ts b/controlplane/src/core/repositories/FederatedGraphRepository.ts index f82901a91c..3fdd8b4e50 100644 --- a/controlplane/src/core/repositories/FederatedGraphRepository.ts +++ b/controlplane/src/core/repositories/FederatedGraphRepository.ts @@ -710,7 +710,7 @@ export class FederatedGraphRepository { * the schema version is not composable the errors are stored in the compositionErrors * but the composedSchemaVersionId is not updated. */ - public addSchemaVersion({ + public async addSchemaVersion({ targetId, composedSDL, clientSchema, @@ -733,87 +733,86 @@ export class FederatedGraphRepository { isFeatureFlagComposition: boolean; featureFlagId: string; }) { - return this.db.transaction(async (tx) => { - const fedGraphRepo = new FederatedGraphRepository(this.logger, tx, this.organizationId); - const compositionRepo = new GraphCompositionRepository(this.logger, tx); - const fedGraph = await fedGraphRepo.byTargetId(targetId); - if (fedGraph === undefined) { - return undefined; - } + const compositionRepo = new GraphCompositionRepository(this.logger, this.db); + const [federatedGraph] = await this.db + .select({ + targetId: targets.id, + id: federatedGraphs.id, + composedSchemaVersionId: federatedGraphs.composedSchemaVersionId, + routerCompatibilityVersion: federatedGraphs.routerCompatibilityVersion, + }) + .from(targets) + .innerJoin(federatedGraphs, eq(federatedGraphs.targetId, targetId)) + .where(and( + eq(targets.type, 'federated'), + eq(targets.organizationId, this.organizationId), + eq(targets.id, targetId), + )) + .execute(); - let compositionErrorString = ''; - let compositionWarningString = ''; + if (federatedGraph === undefined) { + return undefined; + } - if (compositionErrors && compositionErrors.length > 0) { - compositionErrorString = compositionErrors.map((e) => e.toString()).join('\n'); - } + let compositionErrorString = ''; + let compositionWarningString = ''; - if (compositionWarnings && compositionWarnings.length > 0) { - compositionWarningString = compositionWarnings.map((w) => w.toString()).join('\n'); - } + if (compositionErrors && compositionErrors.length > 0) { + compositionErrorString = compositionErrors.map((e) => e.toString()).join('\n'); + } - const insertedVersion = await tx - .insert(schemaVersion) - .values({ - id: schemaVersionId, - organizationId: this.organizationId, - targetId: fedGraph.targetId, - schemaSDL: composedSDL, - clientSchema, - }) - .returning({ - insertedId: schemaVersion.id, - }); + if (compositionWarnings && compositionWarnings.length > 0) { + compositionWarningString = compositionWarnings.map((w) => w.toString()).join('\n'); + } - // Always update the federated schema after composing, even if the schema is not composable. - // That allows us to display the latest schema version in the UI. The router will only fetch - // the latest composable schema version. - if (isFeatureFlagComposition) { - await tx.insert(federatedGraphsToFeatureFlagSchemaVersions).values({ - composedSchemaVersionId: schemaVersionId, - federatedGraphId: fedGraph.id, - baseCompositionSchemaVersionId: fedGraph.composedSchemaVersionId || '', - featureFlagId, - }); - } else { - await tx - .update(federatedGraphs) - .set({ - composedSchemaVersionId: insertedVersion[0].insertedId, - }) - .where(eq(federatedGraphs.id, fedGraph.id)); - } + const insertedVersion = await this.db + .insert(schemaVersion) + .values({ + id: schemaVersionId, + organizationId: this.organizationId, + targetId: federatedGraph.targetId, + schemaSDL: composedSDL, + clientSchema, + }) + .returning({ + insertedId: schemaVersion.id, + }); - // adding the composition entry and the relation between fedGraph schema version and subgraph schema version - await compositionRepo.addComposition({ - fedGraphTargetId: fedGraph.targetId, - fedGraphSchemaVersionId: insertedVersion[0].insertedId, - composedSubgraphs, - compositionErrorString, - compositionWarningString, - composedById, - isFeatureFlagComposition, - routerCompatibilityVersion: fedGraph.routerCompatibilityVersion, + // Always update the federated schema after composing, even if the schema is not composable. + // That allows us to display the latest schema version in the UI. The router will only fetch + // the latest composable schema version. + if (isFeatureFlagComposition) { + await this.db.insert(federatedGraphsToFeatureFlagSchemaVersions).values({ + composedSchemaVersionId: schemaVersionId, + federatedGraphId: federatedGraph.id, + baseCompositionSchemaVersionId: federatedGraph.composedSchemaVersionId || '', + featureFlagId, }); + } else { + await this.db + .update(federatedGraphs) + .set({ + composedSchemaVersionId: insertedVersion[0].insertedId, + }) + .where(eq(federatedGraphs.id, federatedGraph.id)); + } - return { - id: fedGraph.id, - targetId: fedGraph.targetId, - supportsFederation: fedGraph.supportsFederation, - name: fedGraph.name, - labelMatchers: fedGraph.labelMatchers, - compositionErrors: compositionErrorString, - isComposable: fedGraph.isComposable, - lastUpdatedAt: fedGraph.lastUpdatedAt, - routingUrl: fedGraph.routingUrl, - subgraphsCount: fedGraph.subgraphsCount, - composedSchemaVersionId: insertedVersion[0].insertedId, - namespace: fedGraph.namespace, - namespaceId: fedGraph.namespaceId, - routerCompatibilityVersion: fedGraph.routerCompatibilityVersion, - organizationId: fedGraph.organizationId, - }; + // adding the composition entry and the relation between fedGraph schema version and subgraph schema version + await compositionRepo.addComposition({ + fedGraphTargetId: federatedGraph.targetId, + fedGraphSchemaVersionId: insertedVersion[0].insertedId, + composedSubgraphs, + compositionErrorString, + compositionWarningString, + composedById, + isFeatureFlagComposition, + routerCompatibilityVersion: federatedGraph.routerCompatibilityVersion, }); + + return { + composedSchemaVersionId: insertedVersion[0].insertedId, + routerCompatibilityVersion: federatedGraph.routerCompatibilityVersion, + }; } public async isLatestValidSchemaVersion(targetId: string, schemaVersionId: string): Promise { @@ -1425,370 +1424,6 @@ export class FederatedGraphRepository { }); } - /** - * This method recomposes and deploys federated graphs and their respective contract graphs. - */ - public composeAndDeployGraphs({ - actorId, - admissionConfig, - compositionOptions, - chClient, - blobStorage, - federatedGraphs, - webhookProxyUrl, - }: { - actorId: string; - admissionConfig: { - webhookJWTSecret: string; - cdnBaseUrl: string; - }; - blobStorage: BlobStorage; - chClient: ClickHouseClient; - federatedGraphs: FederatedGraphDTO[]; - compositionOptions?: CompositionOptions; - webhookProxyUrl?: string; - }) { - return this.db.transaction(async (tx) => { - const subgraphRepo = new SubgraphRepository(this.logger, tx, this.organizationId); - const fedGraphRepo = new FederatedGraphRepository(this.logger, tx, this.organizationId); - const contractRepo = new ContractRepository(this.logger, tx, this.organizationId); - const featureFlagRepo = new FeatureFlagRepository(this.logger, tx, this.organizationId); - const graphCompositionRepo = new GraphCompositionRepository(this.logger, tx); - const composer = new Composer( - this.logger, - this.db, - fedGraphRepo, - subgraphRepo, - contractRepo, - graphCompositionRepo, - chClient, - webhookProxyUrl, - ); - - const allDeploymentErrors: PlainMessage[] = []; - const allCompositionErrors: PlainMessage[] = []; - const allCompositionWarnings: PlainMessage[] = []; - - parentLoop: for (const federatedGraph of federatedGraphs) { - // Get published subgraphs for recomposition of the federated graph - const subgraphs = await subgraphRepo.listByFederatedGraph({ - federatedGraphTargetId: federatedGraph.targetId, - published: true, - }); - - const contracts = await contractRepo.bySourceFederatedGraphId(federatedGraph.id); - const tagOptionsByContractName = contracts.map((contract) => ({ - contractName: contract.downstreamFederatedGraph.target.name, - excludeTags: contract.excludeTags, - includeTags: contract.includeTags, - })); - - const baseCompositionSubgraphs = subgraphs.map((s) => ({ - name: s.name, - url: s.routingUrl, - definitions: parse(s.schemaSDL), - })); - - // Collects the base graph and applicable feature flag related graphs - const allSubgraphsToCompose: SubgraphsToCompose[] = await featureFlagRepo.getSubgraphsToCompose({ - baseSubgraphs: subgraphs, - baseCompositionSubgraphs, - fedGraphLabelMatchers: federatedGraph.labelMatchers, - }); - - const { results } = await composeGraphsInWorker({ - federatedGraph, - subgraphsToCompose: allSubgraphsToCompose.map((subgraphsToCompose) => ({ - subgraphs: subgraphsToCompose.subgraphs, - isFeatureFlagComposition: subgraphsToCompose.isFeatureFlagComposition, - featureFlagName: subgraphsToCompose.featureFlagName, - featureFlagId: subgraphsToCompose.featureFlagId, - })), - tagOptionsByContractName, - compositionOptions, - }); - - /* baseCompositionData contains the router execution config and the schema version ID for the source graph - * base composition (not a contract or feature flag composition) - * */ - const baseCompositionData: BaseCompositionData = { - featureFlagRouterExecutionConfigByFeatureFlagName: new Map(), - }; - - /* Map of the contract base composition schema version ID, router execution config, - * and any feature flag schema version IDs by contract ID */ - const contractBaseCompositionDataByContractId = new Map(); - - for (const compositionResult of results) { - if (!compositionResult.base.success) { - // Collect all composition errors - allCompositionErrors.push( - ...compositionResult.base.errors.map((message) => ({ - federatedGraphName: federatedGraph.name, - namespace: federatedGraph.namespace, - message, - featureFlag: compositionResult.featureFlagName || '', - })), - ); - } - - // Collect all composition warnings - allCompositionWarnings.push( - ...compositionResult.base.warnings.map((warning) => ({ - federatedGraphName: federatedGraph.name, - namespace: federatedGraph.namespace, - message: warning.message, - featureFlag: compositionResult.featureFlagName || '', - })), - ); - - if ( - !compositionResult.isFeatureFlagComposition && - !compositionResult.base.success && - !federatedGraph.contract - ) { - allCompositionErrors.push(unsuccessfulBaseCompositionError(federatedGraph.name, federatedGraph.namespace)); - } - - const federatedSchemaVersionId = randomUUID(); - const baseComposedGraph = deserializeComposedGraphArtifact(federatedGraph, compositionResult.base); - let routerExecutionConfig; - if (compositionResult.base.success) { - if (!compositionResult.base.routerExecutionConfigJson) { - throw new Error( - `Successful composition for federated graph "${federatedGraph.name}" does not contain a router execution config.`, - ); - } - - routerExecutionConfig = deserializeRouterExecutionConfig(compositionResult.base.routerExecutionConfigJson); - } - - if (routerExecutionConfig) { - routerExecutionConfig.version = federatedSchemaVersionId; - } - - const baseComposition = await composer.saveComposition({ - composedGraph: baseComposedGraph, - composedById: actorId, - isFeatureFlagComposition: compositionResult.isFeatureFlagComposition, - federatedSchemaVersionId, - routerExecutionConfig, - featureFlagId: compositionResult.featureFlagId, - }); - - if (!compositionResult.base.success || !baseComposition.schemaVersionId) { - /* If the base composition failed to compose or deploy, return to the parent loop, because - * contracts are not composed if the base composition fails. - */ - if (!compositionResult.isFeatureFlagComposition) { - continue parentLoop; - } - // Record the feature flag composition to upload (if there are no errors) - } else if (compositionResult.isFeatureFlagComposition) { - if (!routerExecutionConfig) { - throw new Error( - `Successful feature flag composition for federated graph "${federatedGraph.name}" does not contain a router execution config.`, - ); - } - baseCompositionData.featureFlagRouterExecutionConfigByFeatureFlagName.set( - compositionResult.featureFlagName, - routerConfigToFeatureFlagExecutionConfig(routerExecutionConfig), - ); - // Otherwise, this is the base composition, so store the schema version id - } else { - if (!routerExecutionConfig) { - throw new Error( - `Successful composition for federated graph "${federatedGraph.name}" does not contain a router execution config.`, - ); - } - baseCompositionData.schemaVersionId = baseComposition.schemaVersionId; - baseCompositionData.routerExecutionConfig = routerExecutionConfig; - } - - // If there are no contracts, there is nothing further to do - if (compositionResult.contracts.length === 0) { - continue; - } - - for (const { contractName, artifact } of compositionResult.contracts) { - const contractGraph = await fedGraphRepo.byName(contractName, federatedGraph.namespace); - if (!contractGraph) { - throw new Error(`The contract graph "${contractName}" was not found.`); - } - if (!artifact.success) { - allCompositionErrors.push( - ...artifact.errors.map((message) => ({ - federatedGraphName: contractGraph.name, - namespace: contractGraph.namespace, - message, - featureFlag: compositionResult.featureFlagName, - })), - ); - } - - allCompositionWarnings.push( - ...artifact.warnings.map((warning) => ({ - federatedGraphName: contractGraph.name, - namespace: contractGraph.namespace, - message: warning.message, - featureFlag: compositionResult.featureFlagName, - })), - ); - - const contractSchemaVersionId = randomUUID(); - const contractComposedGraph = deserializeComposedGraphArtifact(contractGraph, artifact); - let contractRouterExecutionConfig; - if (artifact.success) { - if (!artifact.routerExecutionConfigJson) { - throw new Error( - `Successful contract composition for federated graph "${contractGraph.name}" does not contain a router execution config.`, - ); - } - contractRouterExecutionConfig = deserializeRouterExecutionConfig(artifact.routerExecutionConfigJson); - if (!contractRouterExecutionConfig) { - throw new Error( - `Successful contract composition for federated graph "${contractGraph.name}" did not produce a router execution config.`, - ); - } - contractRouterExecutionConfig.version = contractSchemaVersionId; - } - - const contractComposition = await composer.saveComposition({ - composedGraph: contractComposedGraph, - composedById: actorId, - isFeatureFlagComposition: compositionResult.isFeatureFlagComposition, - federatedSchemaVersionId: contractSchemaVersionId, - routerExecutionConfig: contractRouterExecutionConfig, - featureFlagId: compositionResult.featureFlagId, - }); - - if (!artifact.success || !contractComposition.schemaVersionId) { - continue; - } - if (!contractRouterExecutionConfig) { - throw new Error( - `Successful contract composition for federated graph "${contractGraph.name}" did not produce a router execution config.`, - ); - } - - /* If the base composition for which this contract has been made is NOT a feature flag composition, - * it must be the contract base composition, which must always be uploaded. - * The base composition is always the first item in the subgraphsToCompose array. - * */ - if (!compositionResult.isFeatureFlagComposition) { - contractBaseCompositionDataByContractId.set(contractGraph.id, { - schemaVersionId: contractComposition.schemaVersionId, - routerExecutionConfig: contractRouterExecutionConfig, - featureFlagRouterExecutionConfigByFeatureFlagName: new Map(), - }); - continue; - } - - /* If the contract has a feature flag, get the current array feature flag versions (or set a new one), - * and then push the current schema version to the array - * */ - const existingContractBaseCompositionData = contractBaseCompositionDataByContractId.get(contractGraph.id); - /* If the existingContractSchemaVersions is undefined, it means the contract base composition failed. - * In this case, simply continue, because when iterating a feature flag for the source graph composition, - * there may not be any errors for the feature flag. - * */ - if (!existingContractBaseCompositionData) { - continue; - } - existingContractBaseCompositionData.featureFlagRouterExecutionConfigByFeatureFlagName.set( - compositionResult.featureFlagName, - routerConfigToFeatureFlagExecutionConfig(contractRouterExecutionConfig), - ); - } - } - - const federatedGraphDTO = await this.byId(federatedGraph.id); - if (!federatedGraphDTO) { - throw new Error(`Fatal: The federated graph "${federatedGraph.name}" was not found.`); - } - if (!baseCompositionData.routerExecutionConfig) { - throw new Error( - `Fatal: The latest router execution config for federated graph "${federatedGraph.name}" was not generated.`, - ); - } - if (!baseCompositionData.schemaVersionId) { - throw new Error( - `Fatal: The latest base composition for federated graph "${federatedGraph.name}" was not found.`, - ); - } - - const { errors: uploadErrors } = await composer.composeAndUploadRouterConfig({ - federatedGraphId: federatedGraphDTO.id, - featureFlagRouterExecutionConfigByFeatureFlagName: - baseCompositionData.featureFlagRouterExecutionConfigByFeatureFlagName, - blobStorage, - organizationId: this.organizationId, - admissionConfig: { - cdnBaseUrl: admissionConfig.cdnBaseUrl, - jwtSecret: admissionConfig.webhookJWTSecret, - }, - baseCompositionRouterExecutionConfig: baseCompositionData.routerExecutionConfig, - baseCompositionSchemaVersionId: baseCompositionData.schemaVersionId, - federatedGraphAdmissionWebhookURL: federatedGraphDTO.admissionWebhookURL, - federatedGraphAdmissionWebhookSecret: federatedGraphDTO.admissionWebhookSecret, - actorId, - }); - - allDeploymentErrors.push( - ...uploadErrors - .filter((e) => e instanceof AdmissionError || e instanceof RouterConfigUploadError) - .map((e) => ({ - federatedGraphName: federatedGraph.name, - namespace: federatedGraph.namespace, - message: e.message ?? '', - })), - ); - - for (const [ - contractId, - { featureFlagRouterExecutionConfigByFeatureFlagName, schemaVersionId, routerExecutionConfig }, - ] of contractBaseCompositionDataByContractId) { - const contractDTO = await this.byId(contractId); - if (!contractDTO) { - throw new Error(`Unexpected: Contract graph with id "${contractId}" not found after latest composition`); - } - - const { errors: uploadErrors } = await composer.composeAndUploadRouterConfig({ - admissionConfig: { - cdnBaseUrl: admissionConfig.cdnBaseUrl, - jwtSecret: admissionConfig.webhookJWTSecret, - }, - baseCompositionRouterExecutionConfig: routerExecutionConfig, - baseCompositionSchemaVersionId: schemaVersionId, - blobStorage, - featureFlagRouterExecutionConfigByFeatureFlagName, - federatedGraphId: contractDTO.id, - organizationId: this.organizationId, - federatedGraphAdmissionWebhookURL: contractDTO.admissionWebhookURL, - federatedGraphAdmissionWebhookSecret: contractDTO.admissionWebhookSecret, - actorId, - }); - - allDeploymentErrors.push( - ...uploadErrors - .filter((e) => e instanceof AdmissionError || e instanceof RouterConfigUploadError) - .map((e) => ({ - federatedGraphName: federatedGraph.name, - namespace: federatedGraph.namespace, - message: e.message ?? '', - })), - ); - } - } - - return { - compositionErrors: allCompositionErrors, - deploymentErrors: allDeploymentErrors, - compositionWarnings: allCompositionWarnings, - }; - }); - } - public updateRouterCompatibilityVersion(id: string, version: string) { return this.db .update(federatedGraphs) diff --git a/controlplane/src/core/repositories/GraphCompositionRepository.ts b/controlplane/src/core/repositories/GraphCompositionRepository.ts index d82dbcc84d..8f7ceb4a8e 100644 --- a/controlplane/src/core/repositories/GraphCompositionRepository.ts +++ b/controlplane/src/core/repositories/GraphCompositionRepository.ts @@ -47,127 +47,123 @@ export class GraphCompositionRepository { isFeatureFlagComposition: boolean; routerCompatibilityVersion: string; }) { - await this.db.transaction(async (tx) => { - const actor = await tx.query.users.findFirst({ - where: eq(users.id, composedById), - }); - if (!actor) { - throw new Error(`Could not find actor ${composedById}`); - } + const actor = await this.db.query.users.findFirst({ + where: eq(users.id, composedById), + }); + if (!actor) { + throw new Error(`Could not find actor ${composedById}`); + } - const subgraphSchemaVersionIds = composedSubgraphs.map((subgraph) => subgraph.schemaVersionId); + const subgraphSchemaVersionIds = composedSubgraphs.map((subgraph) => subgraph.schemaVersionId); + const previousComposition = ( + await this.db + .select({ + id: graphCompositions.id, + }) + .from(graphCompositions) + .innerJoin(schemaVersion, eq(schemaVersion.id, graphCompositions.schemaVersionId)) + .where(eq(schemaVersion.targetId, fedGraphTargetId)) + .orderBy(desc(graphCompositions.createdAt)) + .limit(1) + .execute() + )[0]; + + const insertedComposition = await this.db + .insert(graphCompositions) + .values({ + schemaVersionId: fedGraphSchemaVersionId, + compositionErrors: compositionErrorString, + compositionWarnings: compositionWarningString, + isComposable: compositionErrorString === '', + routerConfigSignature, + createdById: composedById, + createdByEmail: actor.email, + deploymentError: deploymentErrorString, + admissionError: admissionErrorString, + isFeatureFlagComposition, + routerCompatibilityVersion, + }) + .returning() + .execute(); - const previousComposition = ( - await tx + if (subgraphSchemaVersionIds.length > 0) { + const prevCompositionSubgraphs: { + id: string; + name: string; + schemaVersionId: string; + targetId: string; + isFeatureSubgraph: boolean; + }[] = []; + if (previousComposition) { + const prevSubgraphs = await this.db .select({ - id: graphCompositions.id, + id: graphCompositionSubgraphs.subgraphId, + name: graphCompositionSubgraphs.subgraphName, + schemaVersionId: graphCompositionSubgraphs.schemaVersionId, + targetId: graphCompositionSubgraphs.subgraphTargetId, + isFeatureSubgraph: graphCompositionSubgraphs.isFeatureSubgraph, }) - .from(graphCompositions) - .innerJoin(schemaVersion, eq(schemaVersion.id, graphCompositions.schemaVersionId)) - .where(eq(schemaVersion.targetId, fedGraphTargetId)) - .orderBy(desc(graphCompositions.createdAt)) - .limit(1) - .execute() - )[0]; - - const insertedComposition = await tx - .insert(graphCompositions) - .values({ - schemaVersionId: fedGraphSchemaVersionId, - compositionErrors: compositionErrorString, - compositionWarnings: compositionWarningString, - isComposable: compositionErrorString === '', - routerConfigSignature, - createdById: composedById, - createdByEmail: actor.email, - deploymentError: deploymentErrorString, - admissionError: admissionErrorString, - isFeatureFlagComposition, - routerCompatibilityVersion, - }) - .returning() - .execute(); - - if (subgraphSchemaVersionIds.length > 0) { - const prevCompositionSubgraphs: { - id: string; - name: string; - schemaVersionId: string; - targetId: string; - isFeatureSubgraph: boolean; - }[] = []; - if (previousComposition) { - const prevSubgraphs = await tx - .select({ - id: graphCompositionSubgraphs.subgraphId, - name: graphCompositionSubgraphs.subgraphName, - schemaVersionId: graphCompositionSubgraphs.schemaVersionId, - targetId: graphCompositionSubgraphs.subgraphTargetId, - isFeatureSubgraph: graphCompositionSubgraphs.isFeatureSubgraph, - }) - .from(graphCompositionSubgraphs) - .where( - and( - eq(graphCompositionSubgraphs.graphCompositionId, previousComposition.id), - not(eq(graphCompositionSubgraphs.changeType, 'removed')), - ), - ) - .execute(); - prevCompositionSubgraphs.push(...prevSubgraphs); - } - - const addedSubgraphs = composedSubgraphs.filter( - (subgraph) => !prevCompositionSubgraphs.some((prevSubgraph) => prevSubgraph.id === subgraph.id), - ); - const removedSubgraphs = prevCompositionSubgraphs.filter( - (subgraph) => !composedSubgraphs.some((prevSubgraph) => prevSubgraph.id === subgraph.id), - ); + .from(graphCompositionSubgraphs) + .where( + and( + eq(graphCompositionSubgraphs.graphCompositionId, previousComposition.id), + not(eq(graphCompositionSubgraphs.changeType, 'removed')), + ), + ) + .execute(); + prevCompositionSubgraphs.push(...prevSubgraphs); + } - const updatedSubgraphs = composedSubgraphs.filter((subgraph) => { - const prevSubgraph = prevCompositionSubgraphs.find((prevSubgraph) => prevSubgraph.id === subgraph.id); - return ( - prevSubgraph && - prevSubgraph.schemaVersionId !== subgraphSchemaVersionIds[composedSubgraphs.indexOf(subgraph)] - ); - }); - - const unchangedSubgraphs = composedSubgraphs.filter((subgraph) => - prevCompositionSubgraphs.some( - (prevSubgraph) => - prevSubgraph.id === subgraph.id && - prevSubgraph.schemaVersionId === subgraphSchemaVersionIds[composedSubgraphs.indexOf(subgraph)], - ), + const addedSubgraphs = composedSubgraphs.filter( + (subgraph) => !prevCompositionSubgraphs.some((prevSubgraph) => prevSubgraph.id === subgraph.id), + ); + const removedSubgraphs = prevCompositionSubgraphs.filter( + (subgraph) => !composedSubgraphs.some((prevSubgraph) => prevSubgraph.id === subgraph.id), + ); + + const updatedSubgraphs = composedSubgraphs.filter((subgraph) => { + const prevSubgraph = prevCompositionSubgraphs.find((prevSubgraph) => prevSubgraph.id === subgraph.id); + return ( + prevSubgraph && prevSubgraph.schemaVersionId !== subgraphSchemaVersionIds[composedSubgraphs.indexOf(subgraph)] ); + }); - const insertValues: (typeof graphCompositionSubgraphs.$inferInsert)[] = [ - ...addedSubgraphs, - ...updatedSubgraphs, - ...removedSubgraphs, - ...unchangedSubgraphs, - ].map((subgraph) => ({ - graphCompositionId: insertedComposition[0].id, - subgraphId: subgraph.id, - subgraphTargetId: subgraph.targetId, - subgraphName: subgraph.name, - schemaVersionId: subgraph.schemaVersionId, - isFeatureSubgraph: subgraph.isFeatureSubgraph, - changeType: (() => { - if (addedSubgraphs.some((s) => s.id === subgraph.id)) { - return 'added'; - } - if (removedSubgraphs.some((s) => s.id === subgraph.id)) { - return 'removed'; - } - if (updatedSubgraphs.some((s) => s.id === subgraph.id)) { - return 'updated'; - } - return 'unchanged'; - })(), - })); - - await tx.insert(graphCompositionSubgraphs).values(insertValues).execute(); - } - }); + const unchangedSubgraphs = composedSubgraphs.filter((subgraph) => + prevCompositionSubgraphs.some( + (prevSubgraph) => + prevSubgraph.id === subgraph.id && + prevSubgraph.schemaVersionId === subgraphSchemaVersionIds[composedSubgraphs.indexOf(subgraph)], + ), + ); + + const insertValues: (typeof graphCompositionSubgraphs.$inferInsert)[] = [ + ...addedSubgraphs, + ...updatedSubgraphs, + ...removedSubgraphs, + ...unchangedSubgraphs, + ].map((subgraph) => ({ + graphCompositionId: insertedComposition[0].id, + subgraphId: subgraph.id, + subgraphTargetId: subgraph.targetId, + subgraphName: subgraph.name, + schemaVersionId: subgraph.schemaVersionId, + isFeatureSubgraph: subgraph.isFeatureSubgraph, + changeType: (() => { + if (addedSubgraphs.some((s) => s.id === subgraph.id)) { + return 'added'; + } + if (removedSubgraphs.some((s) => s.id === subgraph.id)) { + return 'removed'; + } + if (updatedSubgraphs.some((s) => s.id === subgraph.id)) { + return 'updated'; + } + return 'unchanged'; + })(), + })); + + await this.db.insert(graphCompositionSubgraphs).values(insertValues).execute(); + } } public updateComposition({ diff --git a/controlplane/src/core/repositories/SubgraphRepository.ts b/controlplane/src/core/repositories/SubgraphRepository.ts index ce3d506255..afafdaf3ef 100644 --- a/controlplane/src/core/repositories/SubgraphRepository.ts +++ b/controlplane/src/core/repositories/SubgraphRepository.ts @@ -1016,6 +1016,37 @@ export class SubgraphRepository { : Promise.resolve([]); } + public async getSubgraphNameByIds(subgraphIds: string[]): Promise> { + const results = new Map(); + if (subgraphIds.length === 0) { + return results; + } + + const pendingIds = [...new Set(subgraphIds)]; + while (pendingIds.length > 0) { + const chunkOfIds = pendingIds.splice(0, 100); + const chunkOfSubgraphNames = await this.db + .select({ + id: subgraphs.id, + name: targets.name, + }) + .from(targets) + .innerJoin(subgraphs, eq(subgraphs.targetId, targets.id)) + .where(and(eq(targets.type, 'subgraph'), inArray(subgraphs.id, chunkOfIds))) + .execute(); + + for (const subgraph of chunkOfSubgraphNames) { + results.set(subgraph.id, subgraph.name); + } + + if (chunkOfIds.length < 100) { + break; + } + } + + return results; + } + private async getSubgraphsMatching({ conditions, published, @@ -1050,7 +1081,7 @@ export class SubgraphRepository { svSchemaSDL: schema.schemaVersion.schemaSDL, svIsV2Graph: schema.schemaVersion.isV2Graph, // Proto - protoSchemaVersion: schema.protobufSchemaVersions.protoSchema, + protoSchema: schema.protobufSchemaVersions.protoSchema, protoMappings: schema.protobufSchemaVersions.protoMappings, protoLock: schema.protobufSchemaVersions.protoLock, // Plugin Data @@ -1095,14 +1126,14 @@ export class SubgraphRepository { return subgraphs.map((sg) => { let proto: ProtoSubgraph | undefined; if (sg.type === 'grpc_plugin' || sg.type === 'grpc_service') { - if (!sg.protoSchemaVersion) { + if (!sg.protoSchema) { this.logger.warn( `Missing protobuf schema for ${sg.type} subgraph with schemaVersionId: ${sg.schemaVersionId}`, ); } proto = { - schema: sg.protoSchemaVersion ?? '', + schema: sg.protoSchema ?? '', mappings: sg.protoMappings ?? '', lock: sg.protoLock ?? '', }; From dc8ba9d20cd6de3ce0b515a9af769541166bbaa1 Mon Sep 17 00:00:00 2001 From: Wilson Rivera Date: Fri, 29 May 2026 18:20:24 -0400 Subject: [PATCH 03/12] chore: linting --- .../src/core/repositories/FederatedGraphRepository.ts | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/controlplane/src/core/repositories/FederatedGraphRepository.ts b/controlplane/src/core/repositories/FederatedGraphRepository.ts index 3fdd8b4e50..6842fbe184 100644 --- a/controlplane/src/core/repositories/FederatedGraphRepository.ts +++ b/controlplane/src/core/repositories/FederatedGraphRepository.ts @@ -743,11 +743,9 @@ export class FederatedGraphRepository { }) .from(targets) .innerJoin(federatedGraphs, eq(federatedGraphs.targetId, targetId)) - .where(and( - eq(targets.type, 'federated'), - eq(targets.organizationId, this.organizationId), - eq(targets.id, targetId), - )) + .where( + and(eq(targets.type, 'federated'), eq(targets.organizationId, this.organizationId), eq(targets.id, targetId)), + ) .execute(); if (federatedGraph === undefined) { From e66af9500ff64b7ff737cbbbcae325f6b028db7c Mon Sep 17 00:00:00 2001 From: Wilson Rivera Date: Fri, 29 May 2026 20:11:48 -0400 Subject: [PATCH 04/12] chore: remove fallback empty string --- controlplane/src/core/repositories/FederatedGraphRepository.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/controlplane/src/core/repositories/FederatedGraphRepository.ts b/controlplane/src/core/repositories/FederatedGraphRepository.ts index 6842fbe184..5fc5378479 100644 --- a/controlplane/src/core/repositories/FederatedGraphRepository.ts +++ b/controlplane/src/core/repositories/FederatedGraphRepository.ts @@ -783,7 +783,7 @@ export class FederatedGraphRepository { await this.db.insert(federatedGraphsToFeatureFlagSchemaVersions).values({ composedSchemaVersionId: schemaVersionId, federatedGraphId: federatedGraph.id, - baseCompositionSchemaVersionId: federatedGraph.composedSchemaVersionId || '', + baseCompositionSchemaVersionId: federatedGraph.composedSchemaVersionId!, featureFlagId, }); } else { From 081a8e1bad3b919d00f52872e03b710df944e46e Mon Sep 17 00:00:00 2001 From: Wilson Rivera Date: Fri, 29 May 2026 20:16:07 -0400 Subject: [PATCH 05/12] chore: use simple object instead of a Map --- controlplane/src/core/repositories/FeatureFlagRepository.ts | 2 +- controlplane/src/core/repositories/SubgraphRepository.ts | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/controlplane/src/core/repositories/FeatureFlagRepository.ts b/controlplane/src/core/repositories/FeatureFlagRepository.ts index 9d9cd3d0e9..3c63812a0e 100644 --- a/controlplane/src/core/repositories/FeatureFlagRepository.ts +++ b/controlplane/src/core/repositories/FeatureFlagRepository.ts @@ -1010,7 +1010,7 @@ export class FeatureFlagRepository { } } - const baseSubgraphName = baseSubgraphNameById.get(graph.baseSubgraphId); + const baseSubgraphName = baseSubgraphNameById[graph.baseSubgraphId]; if (!baseSubgraphName) { continue; } diff --git a/controlplane/src/core/repositories/SubgraphRepository.ts b/controlplane/src/core/repositories/SubgraphRepository.ts index afafdaf3ef..cd13523c7e 100644 --- a/controlplane/src/core/repositories/SubgraphRepository.ts +++ b/controlplane/src/core/repositories/SubgraphRepository.ts @@ -1016,8 +1016,8 @@ export class SubgraphRepository { : Promise.resolve([]); } - public async getSubgraphNameByIds(subgraphIds: string[]): Promise> { - const results = new Map(); + public async getSubgraphNameByIds(subgraphIds: string[]): Promise> { + const results: Record = {}; if (subgraphIds.length === 0) { return results; } @@ -1036,7 +1036,7 @@ export class SubgraphRepository { .execute(); for (const subgraph of chunkOfSubgraphNames) { - results.set(subgraph.id, subgraph.name); + results[subgraph.id] = subgraph.name; } if (chunkOfIds.length < 100) { From ed501e4ecb29b12c130b4cdf036172cdfedc81b4 Mon Sep 17 00:00:00 2001 From: Wilson Rivera Date: Mon, 8 Jun 2026 14:43:58 -0400 Subject: [PATCH 06/12] chore: remove nested transaction --- .../core/repositories/SubgraphRepository.ts | 80 ++++++++++--------- 1 file changed, 44 insertions(+), 36 deletions(-) diff --git a/controlplane/src/core/repositories/SubgraphRepository.ts b/controlplane/src/core/repositories/SubgraphRepository.ts index 62d624aa4c..07790a2b87 100644 --- a/controlplane/src/core/repositories/SubgraphRepository.ts +++ b/controlplane/src/core/repositories/SubgraphRepository.ts @@ -257,6 +257,7 @@ export class SubgraphRepository { subgraphChanged: boolean; } > { + const fedGraphRepo = new FederatedGraphRepository(this.logger, this.db, this.organizationId); const deploymentErrors: PlainMessage[] = []; const compositionErrors: PlainMessage[] = []; const compositionWarnings: PlainMessage[] = []; @@ -266,48 +267,55 @@ export class SubgraphRepository { let subgraphChanged = false; let labelChanged = false; - await this.db.transaction(async (tx) => { - const fedGraphRepo = new FederatedGraphRepository(this.logger, tx, this.organizationId); - - const collected = await this.writeSchemaAndCollectAffected(tx, data); - const { subgraph, affectedFederatedGraphById, affectedFeatureFlagIds } = collected; - subgraphChanged = collected.subgraphChanged; - labelChanged = collected.labelChanged; + const collected = await this.writeSchemaAndCollectAffected(this.db, data); + const { subgraph, affectedFederatedGraphById, affectedFeatureFlagIds } = collected; + subgraphChanged = collected.subgraphChanged; + labelChanged = collected.labelChanged; - if (!subgraph) { - return; - } - - // Resolve the affected feature flag DTOs. - const affectedFeatureFlags = await this.resolveFeatureFlags(tx, data.namespaceId, affectedFeatureFlagIds); + if (!subgraph) { + return { + compositionErrors, + compositionWarnings, + updatedFederatedGraphs, + deploymentErrors, + subgraphChanged: subgraphChanged || labelChanged || data.unsetLabels, + }; + } - if (affectedFederatedGraphById.size === 0 && affectedFeatureFlags.length === 0) { - return; - } + // Resolve the affected feature flag DTOs. + const affectedFeatureFlags = await this.resolveFeatureFlags(this.db, data.namespaceId, affectedFeatureFlagIds); + if (affectedFederatedGraphById.size === 0 && affectedFeatureFlags.length === 0) { + return { + compositionErrors, + compositionWarnings, + updatedFederatedGraphs, + deploymentErrors, + subgraphChanged: subgraphChanged || labelChanged || data.unsetLabels, + }; + } - updatedFederatedGraphs.push(...affectedFederatedGraphById.values()); - const result = await compositionService.recomposeAndDeployAffected({ - actorId: data.updatedBy, - affectedFederatedGraphs: [...affectedFederatedGraphById.values()], - affectedFeatureFlags, - isFeatureSubgraph: subgraph.isFeatureSubgraph, - }); + updatedFederatedGraphs.push(...affectedFederatedGraphById.values()); + const result = await compositionService.recomposeAndDeployAffected({ + actorId: data.updatedBy, + affectedFederatedGraphs: [...affectedFederatedGraphById.values()], + affectedFeatureFlags, + isFeatureSubgraph: subgraph.isFeatureSubgraph, + }); - deploymentErrors.push(...result.deploymentErrors); - compositionErrors.push(...result.compositionErrors); - compositionWarnings.push(...result.compositionWarnings); + deploymentErrors.push(...result.deploymentErrors); + compositionErrors.push(...result.compositionErrors); + compositionWarnings.push(...result.compositionWarnings); - // Re-fetch the federated graphs to get the updated composedSchemaVersionId - const refreshedGraphs = await Promise.all( - [...affectedFederatedGraphById.keys()].map((id) => fedGraphRepo.byId(id)), - ); - for (let i = 0; i < updatedFederatedGraphs.length; i++) { - const refreshedGraph = refreshedGraphs[i]; - if (refreshedGraph) { - updatedFederatedGraphs[i] = refreshedGraph; - } + // Re-fetch the federated graphs to get the updated composedSchemaVersionId + const refreshedGraphs = await Promise.all( + [...affectedFederatedGraphById.keys()].map((id) => fedGraphRepo.byId(id)), + ); + for (let i = 0; i < updatedFederatedGraphs.length; i++) { + const refreshedGraph = refreshedGraphs[i]; + if (refreshedGraph) { + updatedFederatedGraphs[i] = refreshedGraph; } - }); + } return { compositionErrors, From a43866148095d85841b6caaaf869e2b2f5f5768f Mon Sep 17 00:00:00 2001 From: Wilson Rivera Date: Wed, 10 Jun 2026 10:27:12 -0400 Subject: [PATCH 07/12] chore: restore nested transaction --- .../repositories/FeatureFlagRepository.ts | 50 ++++++++++--------- 1 file changed, 26 insertions(+), 24 deletions(-) diff --git a/controlplane/src/core/repositories/FeatureFlagRepository.ts b/controlplane/src/core/repositories/FeatureFlagRepository.ts index 192691b7c5..6984daca6e 100644 --- a/controlplane/src/core/repositories/FeatureFlagRepository.ts +++ b/controlplane/src/core/repositories/FeatureFlagRepository.ts @@ -114,7 +114,7 @@ export class FeatureFlagRepository { }); } - public async updateFeatureFlag({ + public updateFeatureFlag({ featureFlag, labels, featureSubgraphIds, @@ -126,31 +126,33 @@ export class FeatureFlagRepository { unsetLabels: boolean; }) { const uniqueLabels = normalizeLabels(labels); - if (labels.length > 0 || unsetLabels) { - const newLabels = unsetLabels ? [] : uniqueLabels; - await this.db - .update(featureFlags) - .set({ - labels: newLabels.map((ul) => joinLabel(ul)), - }) - .where(and(eq(featureFlags.id, featureFlag.id), eq(featureFlags.organizationId, this.organizationId))) - .execute(); - } + return this.db.transaction(async (tx) => { + if (labels.length > 0 || unsetLabels) { + const newLabels = unsetLabels ? [] : uniqueLabels; + await tx + .update(featureFlags) + .set({ + labels: newLabels.map((ul) => joinLabel(ul)), + }) + .where(and(eq(featureFlags.id, featureFlag.id), eq(featureFlags.organizationId, this.organizationId))) + .execute(); + } - if (featureSubgraphIds.length > 0) { - // delete all the feature subgraphs of the feature flag - await this.db - .delete(featureFlagToFeatureSubgraphs) - .where(eq(featureFlagToFeatureSubgraphs.featureFlagId, featureFlag.id)) - .execute(); + if (featureSubgraphIds.length > 0) { + // delete all the feature subgraphs of the feature flag + await tx + .delete(featureFlagToFeatureSubgraphs) + .where(eq(featureFlagToFeatureSubgraphs.featureFlagId, featureFlag.id)) + .execute(); - await this.db.insert(featureFlagToFeatureSubgraphs).values( - featureSubgraphIds.map((featureSubgraphId) => ({ - featureFlagId: featureFlag.id, - featureSubgraphId, - })), - ); - } + await tx.insert(featureFlagToFeatureSubgraphs).values( + featureSubgraphIds.map((featureSubgraphId) => ({ + featureFlagId: featureFlag.id, + featureSubgraphId, + })), + ); + } + }); } public async enableFeatureFlag({ From 88e4442788fec7828f7cd32524a353b9bc3c841f Mon Sep 17 00:00:00 2001 From: Wilson Rivera Date: Wed, 10 Jun 2026 10:42:33 -0400 Subject: [PATCH 08/12] feat: improve performance for subgraph batch publishing --- .../subgraph/publishFederatedSubgraphs.ts | 63 ++++++++++++------- .../repositories/FederatedGraphRepository.ts | 27 +------- .../GraphCompositionRepository.ts | 5 +- .../core/repositories/SubgraphRepository.ts | 27 ++++++-- 4 files changed, 66 insertions(+), 56 deletions(-) diff --git a/controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts b/controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts index 42c3e3229d..ada2fea581 100644 --- a/controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts +++ b/controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts @@ -114,15 +114,17 @@ export function publishFederatedSubgraphs( } // Resolve every requested subgraph; all of them must already exist. + let now = performance.now(); const resolved: { subgraph: SubgraphDTO; schema: string }[] = []; const notFound: string[] = []; const typeErrors: string[] = []; - for (const entry of requestedEntries) { - const subgraph = await subgraphRepo.byName(entry.name, req.namespace); - if (!subgraph) { - notFound.push(entry.name); - continue; - } + // for (const entry of requestedEntries) { + for (const subgraph of await subgraphRepo.getSubgraphsByNames(requestedEntries.map((e) => e.name), namespace.id)) { + // const subgraph = await subgraphRepo.byName(entry.name, req.namespace); + // if (!subgraph) { + // notFound.push(entry.name); + // continue; + // } if (subgraph.type === 'grpc_plugin') { typeErrors.push( @@ -137,9 +139,13 @@ export function publishFederatedSubgraphs( continue; } - resolved.push({ subgraph, schema: entry.schema }); + const schema = requestedEntries.find((re) => re.name === subgraph.name)!.schema; + // const schema = entry.schema; + resolved.push({ subgraph, schema }); } + console.log('load subgraphs: ' + (performance.now() - now)); + if (notFound.length > 0) { return { response: { @@ -166,29 +172,30 @@ export function publishFederatedSubgraphs( }; } - // The user must be authorized to publish each of the subgraphs. + now = performance.now(); for (const { subgraph } of resolved) { - await opts.authorizer.authorize({ - db: opts.db, - graph: { - targetId: subgraph.targetId, - targetType: 'subgraph', - }, - headers: ctx.requestHeader, - authContext, - }); + if (!authContext.rbac.hasSubGraphWriteAccess(subgraph)) { + throw new UnauthorizedError(); + } } + console.log('authorization: ' + (performance.now() - now)); + // Validate every schema as a subgraph SDL before writing anything. + now = performance.now(); + const schemaErrors: string[] = []; const items: (UpdateSubgraphSchemaData & { name: string })[] = []; - for (const { subgraph, schema } of resolved) { - const federatedGraphs = await fedGraphRepo.bySubgraphLabels({ - labels: subgraph.labels, - namespaceId: namespace.id, - }); - const routerCompatibilityVersion = getFederatedGraphRouterCompatibilityVersion(federatedGraphs); + /** + * @TODO: + * + * As of 2026-06-10 we only support v1, so instead of loading the federated graphs just to get that value we are + * going to pass no federated graphs to this method which will return the latest supported version. In the future, + * when we support different versions, we need to revisit this. + */ + const routerCompatibilityVersion = getFederatedGraphRouterCompatibilityVersion([]); + for (const { subgraph, schema } of resolved) { let isEventDrivenGraph = false; let isV2Graph: boolean | undefined; try { @@ -226,6 +233,7 @@ export function publishFederatedSubgraphs( }); } + console.log('load federated graphs: ' + (performance.now() - now)); if (schemaErrors.length > 0) { return { response: { @@ -241,12 +249,17 @@ export function publishFederatedSubgraphs( // Phase 1: persist all schema versions and collect the deduplicated union of affected graphs/flags in a single // short transaction. + now = performance.now(); const { affectedFederatedGraphs, affectedFeatureFlags, changedSubgraphNames } = await subgraphRepo.batchWriteAndCollect(items); + console.log('batch write and collect: ' + (performance.now() - now)); + console.log('affected federated graphs: ' + affectedFederatedGraphs.length); + console.log('affected feature flags: ' + affectedFeatureFlags.length); // Phase 2: compose and deploy the affected graphs OUTSIDE the transaction. Composition is long-running (worker // composition, blob uploads, admission webhooks); holding a DB transaction open across it — for the whole batch — // would tie up a connection and risk timeouts and lock contention. + now = performance.now(); const compositionService = new CompositionService( opts.db, authContext.organizationId, @@ -266,6 +279,8 @@ export function publishFederatedSubgraphs( isFeatureSubgraph: false, }); + console.log('composition: ' + (performance.now() - now)); + // Re-fetch the affected federated graphs to pick up the updated composedSchemaVersionId for the webhook payloads. const updatedFederatedGraphs = ( await Promise.all(affectedFederatedGraphs.map((graph) => fedGraphRepo.byId(graph.id))) @@ -299,6 +314,7 @@ export function publishFederatedSubgraphs( } // Audit log per subgraph that actually changed. + now = performance.now(); const changedSet = new Set(changedSubgraphNames); for (const { subgraph } of resolved) { if (!changedSet.has(subgraph.name)) { @@ -319,6 +335,7 @@ export function publishFederatedSubgraphs( targetNamespaceDisplayName: subgraph.namespace, }); } + console.log('audit logs: ' + (performance.now() - now)); const boundedLimit = req.limit === undefined ? maxRowLimitForChecks : clamp(req.limit, 1, maxRowLimitForChecks); diff --git a/controlplane/src/core/repositories/FederatedGraphRepository.ts b/controlplane/src/core/repositories/FederatedGraphRepository.ts index f82901a91c..966f54c6d5 100644 --- a/controlplane/src/core/repositories/FederatedGraphRepository.ts +++ b/controlplane/src/core/repositories/FederatedGraphRepository.ts @@ -1,12 +1,5 @@ /* eslint-disable no-labels */ -import { KeyObject, randomUUID } from 'node:crypto'; -import { PlainMessage } from '@bufbuild/protobuf'; -import { FeatureFlagRouterExecutionConfig } from '@wundergraph/cosmo-connect/dist/node/v1/node_pb'; -import { - CompositionError, - CompositionWarning, - DeploymentError, -} from '@wundergraph/cosmo-connect/dist/platform/v1/platform_pb'; +import { KeyObject } from 'node:crypto'; import { joinLabel, normalizeURL } from '@wundergraph/cosmo-shared'; import { and, @@ -28,10 +21,9 @@ import { } from 'drizzle-orm'; import { PostgresJsDatabase } from 'drizzle-orm/postgres-js'; import { FastifyBaseLogger } from 'fastify'; -import { parse } from 'graphql'; import { generateKeyPair, importPKCS8, SignJWT } from 'jose'; import { uid } from 'uid/secure'; -import { CompositionOptions, Warning } from '@wundergraph/composition'; +import type { Warning } from '@wundergraph/composition'; import * as schema from '../../db/schema.js'; import { federatedGraphs, @@ -55,35 +47,20 @@ import { RouterRequestKeysDTO, ComposeAndDeployResult, } from '../../types/index.js'; -import { BlobStorage } from '../blobstorage/index.js'; import { - BaseCompositionData, CompositionSubgraphRecord, - Composer, - ContractBaseCompositionData, - routerConfigToFeatureFlagExecutionConfig, - RouterConfigUploadError, } from '../composition/composer.js'; -import { - composeGraphsInWorker, - deserializeComposedGraphArtifact, - deserializeRouterExecutionConfig, -} from '../composition/composeGraphs.pool.js'; import { SchemaDiff } from '../composition/schemaCheck.js'; -import { AdmissionError } from '../services/AdmissionWebhookController.js'; import { applyIdpNamespaceGate, checkIfLabelMatchersChanged, normalizeLabelMatchers, normalizeLabels, } from '../util.js'; -import { unsuccessfulBaseCompositionError } from '../errors/errors.js'; -import { ClickHouseClient } from '../clickhouse/index.js'; import { RBACEvaluator } from '../services/RBACEvaluator.js'; import { traced } from '../tracing.js'; import type { CompositionService } from '../services/CompositionService.js'; import { ContractRepository } from './ContractRepository.js'; -import { FeatureFlagRepository, SubgraphsToCompose } from './FeatureFlagRepository.js'; import { GraphCompositionRepository } from './GraphCompositionRepository.js'; import { SubgraphRepository } from './SubgraphRepository.js'; import { TargetRepository } from './TargetRepository.js'; diff --git a/controlplane/src/core/repositories/GraphCompositionRepository.ts b/controlplane/src/core/repositories/GraphCompositionRepository.ts index d82dbcc84d..c24038c8ec 100644 --- a/controlplane/src/core/repositories/GraphCompositionRepository.ts +++ b/controlplane/src/core/repositories/GraphCompositionRepository.ts @@ -7,7 +7,6 @@ import { graphCompositionSubgraphs, schemaVersion, subgraphs, - targets, users, } from '../../db/schema.js'; import { DateRange, GraphCompositionDTO } from '../../types/index.js'; @@ -56,7 +55,6 @@ export class GraphCompositionRepository { } const subgraphSchemaVersionIds = composedSubgraphs.map((subgraph) => subgraph.schemaVersionId); - const previousComposition = ( await tx .select({ @@ -126,8 +124,7 @@ export class GraphCompositionRepository { const updatedSubgraphs = composedSubgraphs.filter((subgraph) => { const prevSubgraph = prevCompositionSubgraphs.find((prevSubgraph) => prevSubgraph.id === subgraph.id); return ( - prevSubgraph && - prevSubgraph.schemaVersionId !== subgraphSchemaVersionIds[composedSubgraphs.indexOf(subgraph)] + prevSubgraph && prevSubgraph.schemaVersionId !== subgraphSchemaVersionIds[composedSubgraphs.indexOf(subgraph)] ); }); diff --git a/controlplane/src/core/repositories/SubgraphRepository.ts b/controlplane/src/core/repositories/SubgraphRepository.ts index 4a9601132c..ed80f92860 100644 --- a/controlplane/src/core/repositories/SubgraphRepository.ts +++ b/controlplane/src/core/repositories/SubgraphRepository.ts @@ -260,8 +260,6 @@ export class SubgraphRepository { const deploymentErrors: PlainMessage[] = []; const compositionErrors: PlainMessage[] = []; const compositionWarnings: PlainMessage[] = []; - - // The collection of federated graphs that will be potentially re-composed const updatedFederatedGraphs: FederatedGraphDTO[] = []; let subgraphChanged = false; let labelChanged = false; @@ -269,6 +267,7 @@ export class SubgraphRepository { await this.db.transaction(async (tx) => { const fedGraphRepo = new FederatedGraphRepository(this.logger, tx, this.organizationId); + // The collection of federated graphs that will be potentially re-composed const collected = await this.writeSchemaAndCollectAffected(tx, data); const { subgraph, affectedFederatedGraphById, affectedFeatureFlagIds } = collected; subgraphChanged = collected.subgraphChanged; @@ -280,7 +279,6 @@ export class SubgraphRepository { // Resolve the affected feature flag DTOs. const affectedFeatureFlags = await this.resolveFeatureFlags(tx, data.namespaceId, affectedFeatureFlagIds); - if (affectedFederatedGraphById.size === 0 && affectedFeatureFlags.length === 0) { return; } @@ -1091,6 +1089,25 @@ export class SubgraphRepository { }); } + public async getSubgraphsByNames(names: string[], namespaceId: string): Promise { + const uniqueNames = [...new Set(names)]; + + const subgraphs: SubgraphDTO[] = []; + while (uniqueNames.length > 0) { + const chunkOfNames = uniqueNames.splice(0, 100); + const conditions: (SQL | undefined)[] = [ + eq(schema.targets.organizationId, this.organizationId), + eq(schema.targets.namespaceId, namespaceId), + eq(schema.targets.type, 'subgraph'), + inArray(schema.targets.name, chunkOfNames), + ]; + + subgraphs.push(...(await this.getSubgraphsMatching({ conditions }))); + } + + return subgraphs; + } + public getSubgraphsByTargetIds(ids: string[], rbac?: RBACEvaluator): Promise { const conditions: (SQL | undefined)[] = [ eq(schema.targets.organizationId, this.organizationId), @@ -1179,7 +1196,9 @@ export class SubgraphRepository { .execute(); // Transform the selected subgraphs into SubgraphDTO objects - return subgraphs.map((sg) => { + return subgraphs + .filter((sg, index, self) => self.findIndex((x) => x.targetId === sg.targetId) === index) + .map((sg) => { let proto: ProtoSubgraph | undefined; if (sg.type === 'grpc_plugin' || sg.type === 'grpc_service') { if (!sg.protoSchemaVersion) { From a930bc15feddc7861052d05514bbb98522c6b84a Mon Sep 17 00:00:00 2001 From: Wilson Rivera Date: Wed, 10 Jun 2026 14:22:00 -0400 Subject: [PATCH 09/12] chore: cleanup --- .../subgraph/publishFederatedSubgraphs.ts | 55 ++--- .../core/composition/composeGraphs.worker.ts | 2 +- .../repositories/FederatedGraphRepository.ts | 4 +- .../GraphCompositionRepository.ts | 14 +- .../core/repositories/SubgraphRepository.ts | 203 ++++++++++-------- controlplane/src/core/tracing.ts | 4 +- 6 files changed, 141 insertions(+), 141 deletions(-) diff --git a/controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts b/controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts index ada2fea581..225c92d349 100644 --- a/controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts +++ b/controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts @@ -24,6 +24,7 @@ import { } from '../../util.js'; import { OrganizationWebhookService } from '../../webhooks/OrganizationWebhookService.js'; import { CompositionService } from '../../services/CompositionService.js'; +import { withSpan } from '../../tracing.js'; /** * PublishFederatedSubgraphs publishes the schemas of multiple existing subgraphs (and feature subgraphs) in a single @@ -114,18 +115,12 @@ export function publishFederatedSubgraphs( } // Resolve every requested subgraph; all of them must already exist. - let now = performance.now(); const resolved: { subgraph: SubgraphDTO; schema: string }[] = []; - const notFound: string[] = []; const typeErrors: string[] = []; - // for (const entry of requestedEntries) { - for (const subgraph of await subgraphRepo.getSubgraphsByNames(requestedEntries.map((e) => e.name), namespace.id)) { - // const subgraph = await subgraphRepo.byName(entry.name, req.namespace); - // if (!subgraph) { - // notFound.push(entry.name); - // continue; - // } - + for (const subgraph of await subgraphRepo.getSubgraphsByNames( + requestedEntries.map((e) => e.name), + namespace.id, + )) { if (subgraph.type === 'grpc_plugin') { typeErrors.push( `Subgraph "${subgraph.name}" is a plugin. Please use the 'wgc router plugin publish' command to publish it.`, @@ -139,18 +134,22 @@ export function publishFederatedSubgraphs( continue; } - const schema = requestedEntries.find((re) => re.name === subgraph.name)!.schema; - // const schema = entry.schema; + const schema = requestedEntries + .find((re) => re.name.toLowerCase() === subgraph.name.toLowerCase())! + .schema.trimEnd(); + resolved.push({ subgraph, schema }); } - console.log('load subgraphs: ' + (performance.now() - now)); + const resolvedSubgraphNames = new Set(resolved.map((re) => re.subgraph.name)); + const requestedSubgraphNames = new Set(requestedEntries.map((re) => re.name)); + const notFoundSubgraphNames = [...requestedSubgraphNames.difference(resolvedSubgraphNames)]; - if (notFound.length > 0) { + if (notFoundSubgraphNames.length > 0) { return { response: { code: EnumStatusCode.ERR_NOT_FOUND, - details: `The following subgraphs do not exist in the namespace "${req.namespace}": ${notFound.join(', ')}`, + details: `The following subgraphs do not exist in the namespace "${req.namespace}": ${notFoundSubgraphNames.join(', ')}`, }, compositionErrors: [], deploymentErrors: [], @@ -172,18 +171,15 @@ export function publishFederatedSubgraphs( }; } - now = performance.now(); - for (const { subgraph } of resolved) { - if (!authContext.rbac.hasSubGraphWriteAccess(subgraph)) { - throw new UnauthorizedError(); + withSpan('RBACEvaluator.hasSubGraphWriteAccess', () => { + for (const { subgraph } of resolved) { + if (!authContext.rbac.hasSubGraphWriteAccess(subgraph)) { + throw new UnauthorizedError(); + } } - } - - console.log('authorization: ' + (performance.now() - now)); + }); // Validate every schema as a subgraph SDL before writing anything. - now = performance.now(); - const schemaErrors: string[] = []; const items: (UpdateSubgraphSchemaData & { name: string })[] = []; @@ -230,10 +226,10 @@ export function publishFederatedSubgraphs( updatedBy: authContext.userId, namespaceId: namespace.id, isV2Graph, + subgraph, }); } - console.log('load federated graphs: ' + (performance.now() - now)); if (schemaErrors.length > 0) { return { response: { @@ -249,17 +245,12 @@ export function publishFederatedSubgraphs( // Phase 1: persist all schema versions and collect the deduplicated union of affected graphs/flags in a single // short transaction. - now = performance.now(); const { affectedFederatedGraphs, affectedFeatureFlags, changedSubgraphNames } = await subgraphRepo.batchWriteAndCollect(items); - console.log('batch write and collect: ' + (performance.now() - now)); - console.log('affected federated graphs: ' + affectedFederatedGraphs.length); - console.log('affected feature flags: ' + affectedFeatureFlags.length); // Phase 2: compose and deploy the affected graphs OUTSIDE the transaction. Composition is long-running (worker // composition, blob uploads, admission webhooks); holding a DB transaction open across it — for the whole batch — // would tie up a connection and risk timeouts and lock contention. - now = performance.now(); const compositionService = new CompositionService( opts.db, authContext.organizationId, @@ -279,8 +270,6 @@ export function publishFederatedSubgraphs( isFeatureSubgraph: false, }); - console.log('composition: ' + (performance.now() - now)); - // Re-fetch the affected federated graphs to pick up the updated composedSchemaVersionId for the webhook payloads. const updatedFederatedGraphs = ( await Promise.all(affectedFederatedGraphs.map((graph) => fedGraphRepo.byId(graph.id))) @@ -314,7 +303,6 @@ export function publishFederatedSubgraphs( } // Audit log per subgraph that actually changed. - now = performance.now(); const changedSet = new Set(changedSubgraphNames); for (const { subgraph } of resolved) { if (!changedSet.has(subgraph.name)) { @@ -335,7 +323,6 @@ export function publishFederatedSubgraphs( targetNamespaceDisplayName: subgraph.namespace, }); } - console.log('audit logs: ' + (performance.now() - now)); const boundedLimit = req.limit === undefined ? maxRowLimitForChecks : clamp(req.limit, 1, maxRowLimitForChecks); diff --git a/controlplane/src/core/composition/composeGraphs.worker.ts b/controlplane/src/core/composition/composeGraphs.worker.ts index 98de684225..b97b8f262c 100644 --- a/controlplane/src/core/composition/composeGraphs.worker.ts +++ b/controlplane/src/core/composition/composeGraphs.worker.ts @@ -359,6 +359,6 @@ export default async function composeGraphsInWorkerActual( try { return composeGraphsInWorker(task); } finally { - await Sentry.flush(); + await Sentry.flush(2000); } } diff --git a/controlplane/src/core/repositories/FederatedGraphRepository.ts b/controlplane/src/core/repositories/FederatedGraphRepository.ts index 35ce8e5d3b..81a527efa1 100644 --- a/controlplane/src/core/repositories/FederatedGraphRepository.ts +++ b/controlplane/src/core/repositories/FederatedGraphRepository.ts @@ -47,9 +47,7 @@ import { RouterRequestKeysDTO, ComposeAndDeployResult, } from '../../types/index.js'; -import { - CompositionSubgraphRecord, -} from '../composition/composer.js'; +import { CompositionSubgraphRecord } from '../composition/composer.js'; import { SchemaDiff } from '../composition/schemaCheck.js'; import { applyIdpNamespaceGate, diff --git a/controlplane/src/core/repositories/GraphCompositionRepository.ts b/controlplane/src/core/repositories/GraphCompositionRepository.ts index e236b9f5da..809e6b95cd 100644 --- a/controlplane/src/core/repositories/GraphCompositionRepository.ts +++ b/controlplane/src/core/repositories/GraphCompositionRepository.ts @@ -2,13 +2,7 @@ import { PostgresJsDatabase } from 'drizzle-orm/postgres-js'; import { SQL, and, count, desc, eq, gt, lt, not } from 'drizzle-orm'; import { FastifyBaseLogger } from 'fastify'; import * as schema from '../../db/schema.js'; -import { - graphCompositions, - graphCompositionSubgraphs, - schemaVersion, - subgraphs, - users, -} from '../../db/schema.js'; +import { graphCompositions, graphCompositionSubgraphs, schemaVersion, subgraphs, users } from '../../db/schema.js'; import { DateRange, GraphCompositionDTO } from '../../types/index.js'; import { CompositionSubgraphRecord } from '../composition/composer.js'; import { traced } from '../tracing.js'; @@ -53,9 +47,9 @@ export class GraphCompositionRepository { throw new Error(`Could not find actor ${composedById}`); } - const subgraphSchemaVersionIds = composedSubgraphs.map((subgraph) => subgraph.schemaVersionId); - const previousComposition = ( - await this.db + const subgraphSchemaVersionIds = composedSubgraphs.map((subgraph) => subgraph.schemaVersionId); + const previousComposition = ( + await this.db .select({ id: graphCompositions.id, }) diff --git a/controlplane/src/core/repositories/SubgraphRepository.ts b/controlplane/src/core/repositories/SubgraphRepository.ts index 69e3617b65..587d4ed809 100644 --- a/controlplane/src/core/repositories/SubgraphRepository.ts +++ b/controlplane/src/core/repositories/SubgraphRepository.ts @@ -49,6 +49,7 @@ import { SubgraphListFilterOptions, SubgraphMemberDTO, ComposeAndDeployResult, + Feature, } from '../../types/index.js'; import { BlobStorage } from '../blobstorage/index.js'; import { ClickHouseClient } from '../clickhouse/index.js'; @@ -99,6 +100,7 @@ export type UpdateSubgraphSchemaData = { isV2Graph?: boolean; readme?: string; proto?: ProtoSubgraph; + subgraph?: SubgraphDTO; }; /** @@ -274,50 +276,50 @@ export class SubgraphRepository { subgraphChanged = collected.subgraphChanged; labelChanged = collected.labelChanged; - if (!subgraph) { - return { - compositionErrors, - compositionWarnings, - updatedFederatedGraphs, - deploymentErrors, - subgraphChanged: subgraphChanged || labelChanged || data.unsetLabels, - }; - } + if (!subgraph) { + return { + compositionErrors, + compositionWarnings, + updatedFederatedGraphs, + deploymentErrors, + subgraphChanged: subgraphChanged || labelChanged || data.unsetLabels, + }; + } - // Resolve the affected feature flag DTOs. - const affectedFeatureFlags = await this.resolveFeatureFlags(this.db, data.namespaceId, affectedFeatureFlagIds); + // Resolve the affected feature flag DTOs. + const affectedFeatureFlags = await this.resolveFeatureFlags(this.db, data.namespaceId, affectedFeatureFlagIds); if (affectedFederatedGraphById.size === 0 && affectedFeatureFlags.length === 0) { - return{ - compositionErrors, - compositionWarnings, - updatedFederatedGraphs, - deploymentErrors, - subgraphChanged: subgraphChanged || labelChanged || data.unsetLabels, - }; - } + return { + compositionErrors, + compositionWarnings, + updatedFederatedGraphs, + deploymentErrors, + subgraphChanged: subgraphChanged || labelChanged || data.unsetLabels, + }; + } - updatedFederatedGraphs.push(...affectedFederatedGraphById.values()); - const result = await compositionService.recomposeAndDeployAffected({ - actorId: data.updatedBy, - affectedFederatedGraphs: [...affectedFederatedGraphById.values()], - affectedFeatureFlags, - isFeatureSubgraph: subgraph.isFeatureSubgraph, - }); + updatedFederatedGraphs.push(...affectedFederatedGraphById.values()); + const result = await compositionService.recomposeAndDeployAffected({ + actorId: data.updatedBy, + affectedFederatedGraphs: [...affectedFederatedGraphById.values()], + affectedFeatureFlags, + isFeatureSubgraph: subgraph.isFeatureSubgraph, + }); - deploymentErrors.push(...result.deploymentErrors); - compositionErrors.push(...result.compositionErrors); - compositionWarnings.push(...result.compositionWarnings); + deploymentErrors.push(...result.deploymentErrors); + compositionErrors.push(...result.compositionErrors); + compositionWarnings.push(...result.compositionWarnings); - // Re-fetch the federated graphs to get the updated composedSchemaVersionId - const refreshedGraphs = await Promise.all( - [...affectedFederatedGraphById.keys()].map((id) => fedGraphRepo.byId(id)), - ); - for (let i = 0; i < updatedFederatedGraphs.length; i++) { - const refreshedGraph = refreshedGraphs[i]; - if (refreshedGraph) { - updatedFederatedGraphs[i] = refreshedGraph; + // Re-fetch the federated graphs to get the updated composedSchemaVersionId + const refreshedGraphs = await Promise.all( + [...affectedFederatedGraphById.keys()].map((id) => fedGraphRepo.byId(id)), + ); + for (let i = 0; i < updatedFederatedGraphs.length; i++) { + const refreshedGraph = refreshedGraphs[i]; + if (refreshedGraph) { + updatedFederatedGraphs[i] = refreshedGraph; + } } - } }); return { @@ -347,6 +349,7 @@ export class SubgraphRepository { const featureFlag = await featureFlagRepo.getFeatureFlagById({ namespaceId, featureFlagId, + includeSubgraphs: false, }); if (!featureFlag) { @@ -368,6 +371,7 @@ export class SubgraphRepository { private async writeSchemaAndCollectAffected( tx: PostgresJsDatabase, data: UpdateSubgraphSchemaData, + splitConfigFeature?: Feature, ): Promise<{ subgraph: SubgraphDTO | undefined; affectedFederatedGraphById: Map; @@ -394,18 +398,18 @@ export class SubgraphRepository { const featureFlagRepo = new FeatureFlagRepository(this.logger, tx, this.organizationId); const orgRepo = new OrganizationRepository(this.logger, tx); - const splitConfigFeature = await orgRepo.getFeature({ + splitConfigFeature ??= await orgRepo.getFeature({ organizationId: this.organizationId, featureId: 'split-config-loading', }); - const subgraph = await subgraphRepo.byTargetId(data.targetId); + const subgraph = data.subgraph ?? (await subgraphRepo.byTargetId(data.targetId)); if (!subgraph) { return { subgraph: undefined, affectedFederatedGraphById, affectedFeatureFlagIds, subgraphChanged, labelChanged }; } // TODO: avoid downloading the schema use hash instead - if (data.schemaSDL && (subgraph.type === 'grpc_plugin' || data.schemaSDL !== subgraph.schemaSDL)) { + if (data.schemaSDL && (subgraph.type === 'grpc_plugin' || data.schemaSDL !== subgraph.schemaSDL.trimEnd())) { subgraphChanged = true; const updatedSubgraph = await subgraphRepo.addSchemaVersion({ targetId: subgraph.targetId, @@ -555,7 +559,8 @@ export class SubgraphRepository { .from(featureSubgraphsToBaseSubgraphs) .innerJoin(subgraphs, eq(subgraphs.id, featureSubgraphsToBaseSubgraphs.baseSubgraphId)) .innerJoin(targets, eq(targets.id, subgraphs.targetId)) - .where(eq(featureSubgraphsToBaseSubgraphs.featureSubgraphId, subgraph.id)); + .where(eq(featureSubgraphsToBaseSubgraphs.featureSubgraphId, subgraph.id)) + .execute(); if (baseSubgraph.length > 0) { // Retrieve the federated graphs that match the labels for the base graph of the feature graph @@ -663,14 +668,23 @@ export class SubgraphRepository { const namespaceId = items[0].namespaceId; + const orgRepo = new OrganizationRepository(this.logger, this.db, this.organizationId); + const splitConfigFeature = await orgRepo.getFeature({ + organizationId: this.organizationId, + featureId: 'split-config-loading', + }); + await this.db.transaction(async (tx) => { // Write every schema version and collect the affected graphs/flags. NO composition happens here. - for (const item of items) { - const { subgraph, affectedFederatedGraphById, affectedFeatureFlagIds, subgraphChanged, labelChanged } = - await this.writeSchemaAndCollectAffected(tx, item); + const results = await Promise.all( + items.map((item) => this.writeSchemaAndCollectAffected(tx, item, splitConfigFeature)), + ); + + for (const [index, result] of results.entries()) { + const { subgraph, affectedFederatedGraphById, affectedFeatureFlagIds, subgraphChanged, labelChanged } = result; if (subgraph && (subgraphChanged || labelChanged)) { - changedSubgraphNames.push(item.name); + changedSubgraphNames.push(items[index].name); } // Merge via union only — never delete, so one subgraph's old/new label reconciliation cannot drop a @@ -1240,54 +1254,61 @@ export class SubgraphRepository { .execute(); // Transform the selected subgraphs into SubgraphDTO objects - return subgraphs - .filter((sg, index, self) => self.findIndex((x) => x.targetId === sg.targetId) === index) - .map((sg) => { - let proto: ProtoSubgraph | undefined; - if (sg.type === 'grpc_plugin' || sg.type === 'grpc_service') { - if (!sg.protoSchema) { - this.logger.warn( - `Missing protobuf schema for ${sg.type} subgraph with schemaVersionId: ${sg.schemaVersionId}`, - ); - } + return ( + subgraphs + /** + * Because a subgraph can be part of multiple federated graphs in the same namespace, we need to filter out + * duplicates. This have not been an issue so far because the method was called for a specific federated graph + * or with specific target ids. + */ + .filter((sg, index, self) => self.findIndex((x) => x.targetId === sg.targetId) === index) + .map((sg) => { + let proto: ProtoSubgraph | undefined; + if (sg.type === 'grpc_plugin' || sg.type === 'grpc_service') { + if (!sg.protoSchema) { + this.logger.warn( + `Missing protobuf schema for ${sg.type} subgraph with schemaVersionId: ${sg.schemaVersionId}`, + ); + } - proto = { - schema: sg.protoSchema ?? '', - mappings: sg.protoMappings ?? '', - lock: sg.protoLock ?? '', - }; + proto = { + schema: sg.protoSchema ?? '', + mappings: sg.protoMappings ?? '', + lock: sg.protoLock ?? '', + }; - if (sg.type === 'grpc_plugin') { - proto.pluginData = { - platforms: sg.pluginDataPlatforms ?? [], - version: sg.pluginDataVersion ?? 'v1', - }; - } - } + if (sg.type === 'grpc_plugin') { + proto.pluginData = { + platforms: sg.pluginDataPlatforms ?? [], + version: sg.pluginDataVersion ?? 'v1', + }; + } + } - return { - id: sg.id, - targetId: sg.targetId, - routingUrl: sg.routingUrl, - readme: sg.readme || undefined, - subscriptionUrl: sg.subscriptionUrl || '', - subscriptionProtocol: sg.subscriptionProtocol ?? 'ws', - websocketSubprotocol: sg.websocketSubprotocol || undefined, - name: sg.name, - schemaSDL: sg.svSchemaSDL ?? '', - schemaVersionId: sg.schemaVersionId || '', - lastUpdatedAt: sg.svLastUpdated?.toISOString() ?? '', - labels: sg.labels?.map?.((l) => splitLabel(l)) ?? [], - creatorUserId: sg.createdBy || undefined, - namespace: sg.namespaceName, - namespaceId: sg.namespaceId, - isEventDrivenGraph: sg.isEventDrivenGraph, - isV2Graph: sg.svIsV2Graph || undefined, - isFeatureSubgraph: sg.isFeatureSubgraph, - type: sg.type, - proto, - }; - }); + return { + id: sg.id, + targetId: sg.targetId, + routingUrl: sg.routingUrl, + readme: sg.readme || undefined, + subscriptionUrl: sg.subscriptionUrl || '', + subscriptionProtocol: sg.subscriptionProtocol ?? 'ws', + websocketSubprotocol: sg.websocketSubprotocol || undefined, + name: sg.name, + schemaSDL: sg.svSchemaSDL ?? '', + schemaVersionId: sg.schemaVersionId || '', + lastUpdatedAt: sg.svLastUpdated?.toISOString() ?? '', + labels: sg.labels?.map?.((l) => splitLabel(l)) ?? [], + creatorUserId: sg.createdBy || undefined, + namespace: sg.namespaceName, + namespaceId: sg.namespaceId, + isEventDrivenGraph: sg.isEventDrivenGraph, + isV2Graph: sg.svIsV2Graph || undefined, + isFeatureSubgraph: sg.isFeatureSubgraph, + type: sg.type, + proto, + }; + }) + ); } private async getSubgraph(conditions: SQL[]): Promise { diff --git a/controlplane/src/core/tracing.ts b/controlplane/src/core/tracing.ts index b0979d28c4..ff222040b1 100644 --- a/controlplane/src/core/tracing.ts +++ b/controlplane/src/core/tracing.ts @@ -39,6 +39,6 @@ export function traced any>(target: T): T { * Wraps a function call with a Sentry span. * Use for ad-hoc tracing of service calls, auth, etc. */ -export function withSpan(name: string, fn: () => Promise | T): Promise { - return Sentry.startSpan({ name }, () => fn()) as Promise; +export function withSpan(name: string, fn: () => Promise | T): Promise | T { + return Sentry.startSpan({ name }, fn); } From cdb4fc5d35d4ccb2ec29bfd9ac73f06dbb0b4aef Mon Sep 17 00:00:00 2001 From: Wilson Rivera Date: Wed, 10 Jun 2026 14:37:19 -0400 Subject: [PATCH 10/12] chore: add transaction to `SubgraphRepository.update` --- .../core/repositories/SubgraphRepository.ts | 43 +++++++++---------- 1 file changed, 21 insertions(+), 22 deletions(-) diff --git a/controlplane/src/core/repositories/SubgraphRepository.ts b/controlplane/src/core/repositories/SubgraphRepository.ts index da09da8ea6..31952469f9 100644 --- a/controlplane/src/core/repositories/SubgraphRepository.ts +++ b/controlplane/src/core/repositories/SubgraphRepository.ts @@ -259,7 +259,6 @@ export class SubgraphRepository { subgraphChanged: boolean; } > { - const fedGraphRepo = new FederatedGraphRepository(this.logger, this.db, this.organizationId); const deploymentErrors: PlainMessage[] = []; const compositionErrors: PlainMessage[] = []; const compositionWarnings: PlainMessage[] = []; @@ -286,8 +285,8 @@ export class SubgraphRepository { }; } - // Resolve the affected feature flag DTOs. - const affectedFeatureFlags = await this.resolveFeatureFlags(this.db, data.namespaceId, affectedFeatureFlagIds); + // Resolve the affected feature flag DTOs. + const affectedFeatureFlags = await this.resolveFeatureFlags(this.db, data.namespaceId, affectedFeatureFlagIds); if (affectedFederatedGraphById.size === 0 && affectedFeatureFlags.length === 0) { return { compositionErrors, @@ -298,28 +297,28 @@ export class SubgraphRepository { }; } - updatedFederatedGraphs.push(...affectedFederatedGraphById.values()); - const result = await compositionService.recomposeAndDeployAffected({ - actorId: data.updatedBy, - affectedFederatedGraphs: [...affectedFederatedGraphById.values()], - affectedFeatureFlags, - isFeatureSubgraph: subgraph.isFeatureSubgraph, - }); + updatedFederatedGraphs.push(...affectedFederatedGraphById.values()); + const result = await compositionService.recomposeAndDeployAffected({ + actorId: data.updatedBy, + affectedFederatedGraphs: [...affectedFederatedGraphById.values()], + affectedFeatureFlags, + isFeatureSubgraph: subgraph.isFeatureSubgraph, + }); - deploymentErrors.push(...result.deploymentErrors); - compositionErrors.push(...result.compositionErrors); - compositionWarnings.push(...result.compositionWarnings); + deploymentErrors.push(...result.deploymentErrors); + compositionErrors.push(...result.compositionErrors); + compositionWarnings.push(...result.compositionWarnings); - // Re-fetch the federated graphs to get the updated composedSchemaVersionId - const refreshedGraphs = await Promise.all( - [...affectedFederatedGraphById.keys()].map((id) => fedGraphRepo.byId(id)), - ); - for (let i = 0; i < updatedFederatedGraphs.length; i++) { - const refreshedGraph = refreshedGraphs[i]; - if (refreshedGraph) { - updatedFederatedGraphs[i] = refreshedGraph; + // Re-fetch the federated graphs to get the updated composedSchemaVersionId + const refreshedGraphs = await Promise.all( + [...affectedFederatedGraphById.keys()].map((id) => fedGraphRepo.byId(id)), + ); + for (let i = 0; i < updatedFederatedGraphs.length; i++) { + const refreshedGraph = refreshedGraphs[i]; + if (refreshedGraph) { + updatedFederatedGraphs[i] = refreshedGraph; + } } - } }); return { From 5011ba73383a96a206fd6a593513059873bf12f2 Mon Sep 17 00:00:00 2001 From: Wilson Rivera Date: Wed, 10 Jun 2026 14:49:19 -0400 Subject: [PATCH 11/12] chore: restore removed transactions --- .../repositories/FederatedGraphRepository.ts | 142 +++++------ .../GraphCompositionRepository.ts | 225 +++++++++--------- .../core/repositories/SubgraphRepository.ts | 16 +- 3 files changed, 197 insertions(+), 186 deletions(-) diff --git a/controlplane/src/core/repositories/FederatedGraphRepository.ts b/controlplane/src/core/repositories/FederatedGraphRepository.ts index 81a527efa1..d3945cf86f 100644 --- a/controlplane/src/core/repositories/FederatedGraphRepository.ts +++ b/controlplane/src/core/repositories/FederatedGraphRepository.ts @@ -685,7 +685,7 @@ export class FederatedGraphRepository { * the schema version is not composable the errors are stored in the compositionErrors * but the composedSchemaVersionId is not updated. */ - public async addSchemaVersion({ + public addSchemaVersion({ targetId, composedSDL, clientSchema, @@ -708,84 +708,86 @@ export class FederatedGraphRepository { isFeatureFlagComposition: boolean; featureFlagId: string; }) { - const compositionRepo = new GraphCompositionRepository(this.logger, this.db); - const [federatedGraph] = await this.db - .select({ - targetId: targets.id, - id: federatedGraphs.id, - composedSchemaVersionId: federatedGraphs.composedSchemaVersionId, - routerCompatibilityVersion: federatedGraphs.routerCompatibilityVersion, - }) - .from(targets) - .innerJoin(federatedGraphs, eq(federatedGraphs.targetId, targetId)) - .where( - and(eq(targets.type, 'federated'), eq(targets.organizationId, this.organizationId), eq(targets.id, targetId)), - ) - .execute(); + return this.db.transaction(async (tx) => { + const compositionRepo = new GraphCompositionRepository(this.logger, tx); + const [federatedGraph] = await tx + .select({ + targetId: targets.id, + id: federatedGraphs.id, + composedSchemaVersionId: federatedGraphs.composedSchemaVersionId, + routerCompatibilityVersion: federatedGraphs.routerCompatibilityVersion, + }) + .from(targets) + .innerJoin(federatedGraphs, eq(federatedGraphs.targetId, targetId)) + .where( + and(eq(targets.type, 'federated'), eq(targets.organizationId, this.organizationId), eq(targets.id, targetId)), + ) + .execute(); - if (federatedGraph === undefined) { - return undefined; - } + if (federatedGraph === undefined) { + return undefined; + } - let compositionErrorString = ''; - let compositionWarningString = ''; + let compositionErrorString = ''; + let compositionWarningString = ''; - if (compositionErrors && compositionErrors.length > 0) { - compositionErrorString = compositionErrors.map((e) => e.toString()).join('\n'); - } + if (compositionErrors && compositionErrors.length > 0) { + compositionErrorString = compositionErrors.map((e) => e.toString()).join('\n'); + } - if (compositionWarnings && compositionWarnings.length > 0) { - compositionWarningString = compositionWarnings.map((w) => w.toString()).join('\n'); - } + if (compositionWarnings && compositionWarnings.length > 0) { + compositionWarningString = compositionWarnings.map((w) => w.toString()).join('\n'); + } - const insertedVersion = await this.db - .insert(schemaVersion) - .values({ - id: schemaVersionId, - organizationId: this.organizationId, - targetId: federatedGraph.targetId, - schemaSDL: composedSDL, - clientSchema, - }) - .returning({ - insertedId: schemaVersion.id, - }); + const insertedVersion = await tx + .insert(schemaVersion) + .values({ + id: schemaVersionId, + organizationId: this.organizationId, + targetId: federatedGraph.targetId, + schemaSDL: composedSDL, + clientSchema, + }) + .returning({ + insertedId: schemaVersion.id, + }); + + // Always update the federated schema after composing, even if the schema is not composable. + // That allows us to display the latest schema version in the UI. The router will only fetch + // the latest composable schema version. + if (isFeatureFlagComposition) { + await tx.insert(federatedGraphsToFeatureFlagSchemaVersions).values({ + composedSchemaVersionId: schemaVersionId, + federatedGraphId: federatedGraph.id, + baseCompositionSchemaVersionId: federatedGraph.composedSchemaVersionId!, + featureFlagId, + }); + } else { + await tx + .update(federatedGraphs) + .set({ + composedSchemaVersionId: insertedVersion[0].insertedId, + }) + .where(eq(federatedGraphs.id, federatedGraph.id)); + } - // Always update the federated schema after composing, even if the schema is not composable. - // That allows us to display the latest schema version in the UI. The router will only fetch - // the latest composable schema version. - if (isFeatureFlagComposition) { - await this.db.insert(federatedGraphsToFeatureFlagSchemaVersions).values({ - composedSchemaVersionId: schemaVersionId, - federatedGraphId: federatedGraph.id, - baseCompositionSchemaVersionId: federatedGraph.composedSchemaVersionId!, - featureFlagId, + // adding the composition entry and the relation between fedGraph schema version and subgraph schema version + await compositionRepo.addComposition({ + fedGraphTargetId: federatedGraph.targetId, + fedGraphSchemaVersionId: insertedVersion[0].insertedId, + composedSubgraphs, + compositionErrorString, + compositionWarningString, + composedById, + isFeatureFlagComposition, + routerCompatibilityVersion: federatedGraph.routerCompatibilityVersion, }); - } else { - await this.db - .update(federatedGraphs) - .set({ - composedSchemaVersionId: insertedVersion[0].insertedId, - }) - .where(eq(federatedGraphs.id, federatedGraph.id)); - } - // adding the composition entry and the relation between fedGraph schema version and subgraph schema version - await compositionRepo.addComposition({ - fedGraphTargetId: federatedGraph.targetId, - fedGraphSchemaVersionId: insertedVersion[0].insertedId, - composedSubgraphs, - compositionErrorString, - compositionWarningString, - composedById, - isFeatureFlagComposition, - routerCompatibilityVersion: federatedGraph.routerCompatibilityVersion, + return { + composedSchemaVersionId: insertedVersion[0].insertedId, + routerCompatibilityVersion: federatedGraph.routerCompatibilityVersion, + }; }); - - return { - composedSchemaVersionId: insertedVersion[0].insertedId, - routerCompatibilityVersion: federatedGraph.routerCompatibilityVersion, - }; } public async isLatestValidSchemaVersion(targetId: string, schemaVersionId: string): Promise { diff --git a/controlplane/src/core/repositories/GraphCompositionRepository.ts b/controlplane/src/core/repositories/GraphCompositionRepository.ts index 809e6b95cd..b7a0ee50cb 100644 --- a/controlplane/src/core/repositories/GraphCompositionRepository.ts +++ b/controlplane/src/core/repositories/GraphCompositionRepository.ts @@ -40,123 +40,126 @@ export class GraphCompositionRepository { isFeatureFlagComposition: boolean; routerCompatibilityVersion: string; }) { - const actor = await this.db.query.users.findFirst({ - where: eq(users.id, composedById), - }); - if (!actor) { - throw new Error(`Could not find actor ${composedById}`); - } - - const subgraphSchemaVersionIds = composedSubgraphs.map((subgraph) => subgraph.schemaVersionId); - const previousComposition = ( - await this.db - .select({ - id: graphCompositions.id, - }) - .from(graphCompositions) - .innerJoin(schemaVersion, eq(schemaVersion.id, graphCompositions.schemaVersionId)) - .where(eq(schemaVersion.targetId, fedGraphTargetId)) - .orderBy(desc(graphCompositions.createdAt)) - .limit(1) - .execute() - )[0]; - - const insertedComposition = await this.db - .insert(graphCompositions) - .values({ - schemaVersionId: fedGraphSchemaVersionId, - compositionErrors: compositionErrorString, - compositionWarnings: compositionWarningString, - isComposable: compositionErrorString === '', - routerConfigSignature, - createdById: composedById, - createdByEmail: actor.email, - deploymentError: deploymentErrorString, - admissionError: admissionErrorString, - isFeatureFlagComposition, - routerCompatibilityVersion, - }) - .returning() - .execute(); + await this.db.transaction(async (tx) => { + const actor = await tx.query.users.findFirst({ + where: eq(users.id, composedById), + }); + if (!actor) { + throw new Error(`Could not find actor ${composedById}`); + } - if (subgraphSchemaVersionIds.length > 0) { - const prevCompositionSubgraphs: { - id: string; - name: string; - schemaVersionId: string; - targetId: string; - isFeatureSubgraph: boolean; - }[] = []; - if (previousComposition) { - const prevSubgraphs = await this.db + const subgraphSchemaVersionIds = composedSubgraphs.map((subgraph) => subgraph.schemaVersionId); + const previousComposition = ( + await tx .select({ - id: graphCompositionSubgraphs.subgraphId, - name: graphCompositionSubgraphs.subgraphName, - schemaVersionId: graphCompositionSubgraphs.schemaVersionId, - targetId: graphCompositionSubgraphs.subgraphTargetId, - isFeatureSubgraph: graphCompositionSubgraphs.isFeatureSubgraph, + id: graphCompositions.id, }) - .from(graphCompositionSubgraphs) - .where( - and( - eq(graphCompositionSubgraphs.graphCompositionId, previousComposition.id), - not(eq(graphCompositionSubgraphs.changeType, 'removed')), - ), - ) - .execute(); - prevCompositionSubgraphs.push(...prevSubgraphs); - } + .from(graphCompositions) + .innerJoin(schemaVersion, eq(schemaVersion.id, graphCompositions.schemaVersionId)) + .where(eq(schemaVersion.targetId, fedGraphTargetId)) + .orderBy(desc(graphCompositions.createdAt)) + .limit(1) + .execute() + )[0]; + + const insertedComposition = await tx + .insert(graphCompositions) + .values({ + schemaVersionId: fedGraphSchemaVersionId, + compositionErrors: compositionErrorString, + compositionWarnings: compositionWarningString, + isComposable: compositionErrorString === '', + routerConfigSignature, + createdById: composedById, + createdByEmail: actor.email, + deploymentError: deploymentErrorString, + admissionError: admissionErrorString, + isFeatureFlagComposition, + routerCompatibilityVersion, + }) + .returning() + .execute(); + + if (subgraphSchemaVersionIds.length > 0) { + const prevCompositionSubgraphs: { + id: string; + name: string; + schemaVersionId: string; + targetId: string; + isFeatureSubgraph: boolean; + }[] = []; + if (previousComposition) { + const prevSubgraphs = await tx + .select({ + id: graphCompositionSubgraphs.subgraphId, + name: graphCompositionSubgraphs.subgraphName, + schemaVersionId: graphCompositionSubgraphs.schemaVersionId, + targetId: graphCompositionSubgraphs.subgraphTargetId, + isFeatureSubgraph: graphCompositionSubgraphs.isFeatureSubgraph, + }) + .from(graphCompositionSubgraphs) + .where( + and( + eq(graphCompositionSubgraphs.graphCompositionId, previousComposition.id), + not(eq(graphCompositionSubgraphs.changeType, 'removed')), + ), + ) + .execute(); + prevCompositionSubgraphs.push(...prevSubgraphs); + } + + const addedSubgraphs = composedSubgraphs.filter( + (subgraph) => !prevCompositionSubgraphs.some((prevSubgraph) => prevSubgraph.id === subgraph.id), + ); + const removedSubgraphs = prevCompositionSubgraphs.filter( + (subgraph) => !composedSubgraphs.some((prevSubgraph) => prevSubgraph.id === subgraph.id), + ); - const addedSubgraphs = composedSubgraphs.filter( - (subgraph) => !prevCompositionSubgraphs.some((prevSubgraph) => prevSubgraph.id === subgraph.id), - ); - const removedSubgraphs = prevCompositionSubgraphs.filter( - (subgraph) => !composedSubgraphs.some((prevSubgraph) => prevSubgraph.id === subgraph.id), - ); - - const updatedSubgraphs = composedSubgraphs.filter((subgraph) => { - const prevSubgraph = prevCompositionSubgraphs.find((prevSubgraph) => prevSubgraph.id === subgraph.id); - return ( - prevSubgraph && prevSubgraph.schemaVersionId !== subgraphSchemaVersionIds[composedSubgraphs.indexOf(subgraph)] + const updatedSubgraphs = composedSubgraphs.filter((subgraph) => { + const prevSubgraph = prevCompositionSubgraphs.find((prevSubgraph) => prevSubgraph.id === subgraph.id); + return ( + prevSubgraph && + prevSubgraph.schemaVersionId !== subgraphSchemaVersionIds[composedSubgraphs.indexOf(subgraph)] + ); + }); + + const unchangedSubgraphs = composedSubgraphs.filter((subgraph) => + prevCompositionSubgraphs.some( + (prevSubgraph) => + prevSubgraph.id === subgraph.id && + prevSubgraph.schemaVersionId === subgraphSchemaVersionIds[composedSubgraphs.indexOf(subgraph)], + ), ); - }); - const unchangedSubgraphs = composedSubgraphs.filter((subgraph) => - prevCompositionSubgraphs.some( - (prevSubgraph) => - prevSubgraph.id === subgraph.id && - prevSubgraph.schemaVersionId === subgraphSchemaVersionIds[composedSubgraphs.indexOf(subgraph)], - ), - ); - - const insertValues: (typeof graphCompositionSubgraphs.$inferInsert)[] = [ - ...addedSubgraphs, - ...updatedSubgraphs, - ...removedSubgraphs, - ...unchangedSubgraphs, - ].map((subgraph) => ({ - graphCompositionId: insertedComposition[0].id, - subgraphId: subgraph.id, - subgraphTargetId: subgraph.targetId, - subgraphName: subgraph.name, - schemaVersionId: subgraph.schemaVersionId, - isFeatureSubgraph: subgraph.isFeatureSubgraph, - changeType: (() => { - if (addedSubgraphs.some((s) => s.id === subgraph.id)) { - return 'added'; - } - if (removedSubgraphs.some((s) => s.id === subgraph.id)) { - return 'removed'; - } - if (updatedSubgraphs.some((s) => s.id === subgraph.id)) { - return 'updated'; - } - return 'unchanged'; - })(), - })); - - await this.db.insert(graphCompositionSubgraphs).values(insertValues).execute(); - } + const insertValues: (typeof graphCompositionSubgraphs.$inferInsert)[] = [ + ...addedSubgraphs, + ...updatedSubgraphs, + ...removedSubgraphs, + ...unchangedSubgraphs, + ].map((subgraph) => ({ + graphCompositionId: insertedComposition[0].id, + subgraphId: subgraph.id, + subgraphTargetId: subgraph.targetId, + subgraphName: subgraph.name, + schemaVersionId: subgraph.schemaVersionId, + isFeatureSubgraph: subgraph.isFeatureSubgraph, + changeType: (() => { + if (addedSubgraphs.some((s) => s.id === subgraph.id)) { + return 'added'; + } + if (removedSubgraphs.some((s) => s.id === subgraph.id)) { + return 'removed'; + } + if (updatedSubgraphs.some((s) => s.id === subgraph.id)) { + return 'updated'; + } + return 'unchanged'; + })(), + })); + + await tx.insert(graphCompositionSubgraphs).values(insertValues).execute(); + } + }); } public updateComposition({ diff --git a/controlplane/src/core/repositories/SubgraphRepository.ts b/controlplane/src/core/repositories/SubgraphRepository.ts index 31952469f9..9f09ca8a5b 100644 --- a/controlplane/src/core/repositories/SubgraphRepository.ts +++ b/controlplane/src/core/repositories/SubgraphRepository.ts @@ -1122,10 +1122,10 @@ export class SubgraphRepository { while (uniqueNames.length > 0) { const chunkOfNames = uniqueNames.splice(0, 100); const conditions: (SQL | undefined)[] = [ - eq(schema.targets.organizationId, this.organizationId), - eq(schema.targets.namespaceId, namespaceId), - eq(schema.targets.type, 'subgraph'), - inArray(schema.targets.name, chunkOfNames), + eq(targets.organizationId, this.organizationId), + eq(targets.namespaceId, namespaceId), + eq(targets.type, 'subgraph'), + inArray(targets.name, chunkOfNames), ]; subgraphs.push(...(await this.getSubgraphsMatching({ conditions }))); @@ -1162,7 +1162,13 @@ export class SubgraphRepository { }) .from(targets) .innerJoin(subgraphs, eq(subgraphs.targetId, targets.id)) - .where(and(eq(targets.type, 'subgraph'), inArray(subgraphs.id, chunkOfIds))) + .where( + and( + eq(targets.organizationId, this.organizationId), + eq(targets.type, 'subgraph'), + inArray(subgraphs.id, chunkOfIds), + ), + ) .execute(); for (const subgraph of chunkOfSubgraphNames) { From 3f661f3309be7bf1ffc8aa670d3aef77dcd097e8 Mon Sep 17 00:00:00 2001 From: Wilson Rivera Date: Wed, 10 Jun 2026 16:18:57 -0400 Subject: [PATCH 12/12] chore: fix tests --- controlplane/src/core/repositories/SubgraphRepository.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/controlplane/src/core/repositories/SubgraphRepository.ts b/controlplane/src/core/repositories/SubgraphRepository.ts index 9f09ca8a5b..9277480fc8 100644 --- a/controlplane/src/core/repositories/SubgraphRepository.ts +++ b/controlplane/src/core/repositories/SubgraphRepository.ts @@ -348,7 +348,6 @@ export class SubgraphRepository { const featureFlag = await featureFlagRepo.getFeatureFlagById({ namespaceId, featureFlagId, - includeSubgraphs: false, }); if (!featureFlag) {