diff --git a/controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts b/controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts index 42c3e3229d..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 @@ -115,15 +116,11 @@ export function publishFederatedSubgraphs( // Resolve every requested subgraph; all of them must already exist. 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 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.`, @@ -137,14 +134,22 @@ export function publishFederatedSubgraphs( continue; } - resolved.push({ subgraph, schema: entry.schema }); + const schema = requestedEntries + .find((re) => re.name.toLowerCase() === subgraph.name.toLowerCase())! + .schema.trimEnd(); + + resolved.push({ subgraph, schema }); } - if (notFound.length > 0) { + 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 (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: [], @@ -166,29 +171,27 @@ export function publishFederatedSubgraphs( }; } - // The user must be authorized to publish each of the subgraphs. - for (const { subgraph } of resolved) { - await opts.authorizer.authorize({ - db: opts.db, - graph: { - targetId: subgraph.targetId, - targetType: 'subgraph', - }, - headers: ctx.requestHeader, - authContext, - }); - } + withSpan('RBACEvaluator.hasSubGraphWriteAccess', () => { + for (const { subgraph } of resolved) { + if (!authContext.rbac.hasSubGraphWriteAccess(subgraph)) { + throw new UnauthorizedError(); + } + } + }); // Validate every schema as a subgraph SDL before writing anything. 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 { @@ -223,6 +226,7 @@ export function publishFederatedSubgraphs( updatedBy: authContext.userId, namespaceId: namespace.id, isV2Graph, + subgraph, }); } 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 5fc5378479..d3945cf86f 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,18 @@ 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 { CompositionSubgraphRecord } from '../composition/composer.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'; @@ -710,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, @@ -733,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 this.db.insert(federatedGraphsToFeatureFlagSchemaVersions).values({ - composedSchemaVersionId: schemaVersionId, - federatedGraphId: federatedGraph.id, - baseCompositionSchemaVersionId: federatedGraph.composedSchemaVersionId!, - featureFlagId, + // 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)); + } + + // 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 8f7ceb4a8e..b7a0ee50cb 100644 --- a/controlplane/src/core/repositories/GraphCompositionRepository.ts +++ b/controlplane/src/core/repositories/GraphCompositionRepository.ts @@ -2,14 +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, - targets, - 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'; @@ -47,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 07790a2b87..9277480fc8 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; }; /** @@ -257,65 +259,67 @@ export class SubgraphRepository { subgraphChanged: boolean; } > { - const fedGraphRepo = new FederatedGraphRepository(this.logger, this.db, this.organizationId); 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; - const collected = await this.writeSchemaAndCollectAffected(this.db, data); - const { subgraph, affectedFederatedGraphById, affectedFeatureFlagIds } = collected; - subgraphChanged = collected.subgraphChanged; - labelChanged = collected.labelChanged; + await this.db.transaction(async (tx) => { + const fedGraphRepo = new FederatedGraphRepository(this.logger, tx, this.organizationId); - if (!subgraph) { - return { - compositionErrors, - compositionWarnings, - updatedFederatedGraphs, - deploymentErrors, - subgraphChanged: subgraphChanged || labelChanged || data.unsetLabels, - }; - } + // 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; + labelChanged = collected.labelChanged; - // 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, - }; - } + if (!subgraph) { + 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, - }); + // 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, + }; + } - deploymentErrors.push(...result.deploymentErrors); - compositionErrors.push(...result.compositionErrors); - compositionWarnings.push(...result.compositionWarnings); + updatedFederatedGraphs.push(...affectedFederatedGraphById.values()); + const result = await compositionService.recomposeAndDeployAffected({ + actorId: data.updatedBy, + affectedFederatedGraphs: [...affectedFederatedGraphById.values()], + affectedFeatureFlags, + isFeatureSubgraph: subgraph.isFeatureSubgraph, + }); - // 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; + 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; + } } - } + }); return { compositionErrors, @@ -365,6 +369,7 @@ export class SubgraphRepository { private async writeSchemaAndCollectAffected( tx: PostgresJsDatabase, data: UpdateSubgraphSchemaData, + splitConfigFeature?: Feature, ): Promise<{ subgraph: SubgraphDTO | undefined; affectedFederatedGraphById: Map; @@ -391,18 +396,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, @@ -552,7 +557,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 @@ -660,14 +666,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 @@ -1099,6 +1114,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(targets.organizationId, this.organizationId), + eq(targets.namespaceId, namespaceId), + eq(targets.type, 'subgraph'), + inArray(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), @@ -1127,7 +1161,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) { @@ -1218,52 +1258,61 @@ export class SubgraphRepository { .execute(); // Transform the selected subgraphs into SubgraphDTO objects - return subgraphs.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); }