From bf4383e61673977a3ddac00c342e890361e0cd36 Mon Sep 17 00:00:00 2001 From: Wilson Rivera Date: Fri, 29 May 2026 16:40:25 -0400 Subject: [PATCH 1/7] 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 2/7] 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 3/7] 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 4/7] 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 5/7] 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 6/7] 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 7/7] 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({