diff --git a/controlplane/src/core/repositories/FeatureFlagRepository.ts b/controlplane/src/core/repositories/FeatureFlagRepository.ts index 0d64c93875..6984daca6e 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, @@ -908,8 +910,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 +930,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 +951,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 +973,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[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..5fc5378479 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,84 @@ 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 +1422,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 4a9601132c..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; - - if (!subgraph) { - return; - } + const collected = await this.writeSchemaAndCollectAffected(this.db, data); + const { subgraph, affectedFederatedGraphById, affectedFeatureFlagIds } = collected; + subgraphChanged = collected.subgraphChanged; + labelChanged = collected.labelChanged; - // 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, @@ -1103,6 +1111,37 @@ export class SubgraphRepository { : Promise.resolve([]); } + public async getSubgraphNameByIds(subgraphIds: string[]): Promise> { + const results: Record = {}; + 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[subgraph.id] = subgraph.name; + } + + if (chunkOfIds.length < 100) { + break; + } + } + + return results; + } + private async getSubgraphsMatching({ conditions, published, @@ -1137,7 +1176,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 @@ -1182,14 +1221,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 ?? '', };