Add proposal change approval tracking job

Author: jdolle

packages/services/api/src/modules/proposals/providers/schema-proposal-manager.ts

@@ -409,16 +409,13 @@ export class SchemaProposalManager {
           change as unknown as Change<any>,
         ) && implementedChange.schemaVersionId,
     );
-    if (implementation) {
-      return {
-        schemaProposal: { id: schemaProposalId },
-        implementedBy: implementation?.schemaVersionId
-          ? {
-              id: implementation.schemaVersionId,
-            }
-          : null,
-      };
-    }
-    return null;
+    return {
+      schemaProposal: { id: schemaProposalId },
+      implementedBy: implementation?.schemaVersionId
+        ? {
+            id: implementation.schemaVersionId,
+          }
+        : null,
+    };
   }
 }

packages/services/api/src/modules/proposals/providers/schema-proposal-storage.ts

@@ -12,6 +12,7 @@ import {
   SchemaChangeModel,
 } from '@hive/storage';
 import { TaskScheduler } from '@hive/workflows/kit';
+import { SchemaProposalApprovalTask } from '@hive/workflows/tasks/schema-proposal-approval';
 import { SchemaProposalCompositionTask } from '@hive/workflows/tasks/schema-proposal-composition';
 import { SchemaProposalStage } from '../../../__generated__/types';
 import { Logger } from '../../shared/providers/logger';
@@ -61,6 +62,10 @@ export class SchemaProposalStorage {
     await this.taskScheduler.scheduleTask(SchemaProposalCompositionTask, input);
   }
 
+  async runBackgroundApproval(input: { proposalId: string; targetId: string }) {
+    await this.taskScheduler.scheduleTask(SchemaProposalApprovalTask, input);
+  }
+
   private async assertSchemaProposalsEnabled(args: {
     organizationId: string;
     targetId: string;
@@ -140,9 +145,15 @@ export class SchemaProposalStorage {
           )
           RETURNING ${schemaProposalReviewFields}
         `);
+
       return SchemaProposalReviewModel.parse(row);
     });
 
+    // @todo rollback if this fails
+    if (args.stage === 'APPROVED') {
+      this.runBackgroundApproval({ proposalId: args.id, targetId: args.targetId });
+    }
+
     return {
       type: 'ok' as const,
       review,
@@ -384,37 +395,6 @@ export class SchemaProposalStorage {
     };
   }
 
-  async _approveChanges(
-    conn: CommonQueryMethods,
-    records: {
-      change: Change;
-      proposalId: string;
-      service?: string;
-      targetId: string;
-    }[],
-  ) {
-    const values = records.map(
-      r => psql`(
-      ${generateChangeHash(r.change)}
-      ,${JSON.stringify(r.change)}
-      ,${r.proposalId}
-      ,${r.service ?? null}
-      ,${r.targetId}
-    )`,
-    );
-
-    await conn.query(psql`
-      INSERT INTO "proposal_approved_changes" (
-         hash
-        ,change
-        ,proposal_id
-        ,service
-        ,target_id
-      )
-      VALUES ${psql.join(values, psql.fragment`,`)}
-    `);
-  }
-
   /**
    * Updates an approved change to set the schema version where it has been implemented.
    */

packages/services/api/src/modules/proposals/resolvers/SchemaProposalChangeDetails.ts

@@ -37,6 +37,7 @@ export const SchemaProposalChangeDetails: SchemaProposalChangeDetailsResolvers =
   },
   schemaProposal: async (parent, _arg, { injector }) => {
     if (!parent) {
+      // this should never happen
       throw new Error('Uh oh');
     }
     const schemaProposalId = parent.schemaProposal.id;
@@ -46,7 +47,7 @@ export const SchemaProposalChangeDetails: SchemaProposalChangeDetailsResolvers =
       .getProposal({ id: schemaProposalId });
 
     if (!proposal) {
-      throw new Error('Something went wrong!');
+      throw new Error('Proposal not found');
     }
     return proposal;
   },

packages/services/api/src/modules/schema/index.ts

@@ -1,4 +1,5 @@
 import { createModule } from 'graphql-modules';
+import { SchemaProposalManager } from '../proposals/providers/schema-proposal-manager';
 import { SchemaProposalStorage } from '../proposals/providers/schema-proposal-storage';
 import { BreakingSchemaChangeUsageHelper } from './providers/breaking-schema-changes-helper';
 import { Contracts } from './providers/contracts';
@@ -34,5 +35,6 @@ export const schemaModule = createModule({
     CompositionOrchestrator,
     ...models,
     SchemaProposalStorage,
+    SchemaProposalManager,
   ],
 });

packages/services/workflows/package.json

@@ -10,6 +10,7 @@
   },
   "devDependencies": {
     "@graphql-hive/logger": "1.0.9",
+    "@graphql-inspector/compare-changes": "0.1.0-alpha-20260328000712-2cf98744a012fa66614588ed89ba99719b0f461b",
     "@graphql-inspector/core": "7.1.2",
     "@graphql-inspector/patch": "0.1.3",
     "@graphql-yoga/redis-event-target": "3.0.3",

packages/services/workflows/src/index.ts

@@ -43,6 +43,7 @@ const modules = await Promise.all([
   import('./tasks/usage-rate-limit-exceeded.js'),
   import('./tasks/usage-rate-limit-warning.js'),
   import('./tasks/schema-proposal-composition.js'),
+  import('./tasks/schema-proposal-approval.js'),
 ]);
 
 const crontab = `

packages/services/workflows/src/lib/schema/provider.ts

@@ -1,11 +1,12 @@
 import { DocumentNode, GraphQLError, parse, print, SourceLocation } from 'graphql';
 import { z } from 'zod';
 import type { Logger } from '@graphql-hive/logger';
+import { generateChangeHash } from '@graphql-inspector/compare-changes';
 import type { Change } from '@graphql-inspector/core';
 import { errors, patch } from '@graphql-inspector/patch';
 import type { Project, SchemaObject } from '@hive/api';
 import type { ComposeAndValidateResult } from '@hive/api/shared/entities';
-import { PostgresDatabasePool, psql } from '@hive/postgres';
+import { CommonQueryMethods, PostgresDatabasePool, psql } from '@hive/postgres';
 import type { ContractsInputType, SchemaBuilderApi } from '@hive/schema';
 import { decodeCreatedAtAndUUIDIdBasedCursor, HiveSchemaChangeModel } from '@hive/storage';
 import { createTRPCProxyClient, httpLink } from '@trpc/client';
@@ -32,6 +33,8 @@ const SchemaProposalChangesModel = z.object({
   createdAt: z.string(),
 });
 
+type SchemaProposalChangesModelType = z.TypeOf<typeof SchemaProposalChangesModel>;
+
 function createExternalConfig(config: Project['externalComposition']) {
   // : ExternalCompositionConfig {
   if (config && config.enabled) {
@@ -270,12 +273,25 @@ export function schemaProvider(providerConfig: SchemaProviderConfig) {
       return baseSchema;
     },
 
-    async proposedSchemas(args: {
-      targetId: string;
-      proposalId: string;
-      cursor?: string | null;
-      pool: PostgresDatabasePool;
-    }) {
+    /**
+     * Loops through all schema checks that are part of the schema proposal. This uses pagination
+     * to avoid loading too much at once
+     *
+     * This is hard capped at 2_000 subgraphs for safety.
+     **/
+    async forEachProposalCheck(
+      args: {
+        targetId: string;
+        proposalId: string;
+        cursor?: string | null;
+        pool: PostgresDatabasePool;
+      },
+      callback: (
+        change: Omit<SchemaProposalChangesModelType, 'schemaProposalChanges'> & {
+          schemaProposalChanges: Change[];
+        },
+      ) => void | Promise<void>,
+    ) {
       const now = new Date().toISOString();
       let cursor: {
         createdAt: string;
@@ -288,11 +304,6 @@ export function schemaProvider(providerConfig: SchemaProviderConfig) {
 
       // fetch all latest schemas. Support up to 2_000 subgraphs.
       const maxLoops = 100;
-      const services = await this.latestComposableSchemas({
-        targetId: args.targetId,
-        pool: args.pool,
-      });
-
       let nextCursor = cursor;
       // collect changes in paginated requests to avoid stalling the db
       let i = 0;
@@ -350,48 +361,96 @@ export function schemaProvider(providerConfig: SchemaProviderConfig) {
           LIMIT 20
         `);
 
-        const changes = result.map(row => {
-          const value = SchemaProposalChangesModel.parse(row);
+        const checks = result.map(row => {
+          const check = SchemaProposalChangesModel.parse(row);
           return {
-            ...value,
-            schemaProposalChanges: value.schemaProposalChanges.map(c => {
-              const change: Change<any> = {
-                ...c,
+            ...check,
+            schemaProposalChanges: check.schemaProposalChanges.map((c): Change<any> => {
+              return {
+                message: c.message,
+                meta: c.meta,
+                type: c.type,
                 path: c.path ?? undefined,
                 criticality: {
                   level: c.criticality,
                 },
-              };
-              return change;
+              } satisfies Change<any>;
             }),
           };
         });
 
-        if (changes.length === 20) {
+        for (const check of checks) {
+          await callback(check);
+        }
+
+        if (checks.length === 20) {
           nextCursor = {
             // Keep the created at because we want the same set of checks when joining on the "latest".
-            createdAt: nextCursor?.createdAt ?? changes[0]?.createdAt ?? now,
-            id: changes[changes.length - 1].id,
+            createdAt: nextCursor?.createdAt ?? checks[0]?.createdAt ?? now,
+            id: checks[checks.length - 1].id,
           };
         } else {
           nextCursor = null;
         }
+      } while (nextCursor && ++i < maxLoops);
+    },
 
-        for (const change of changes) {
-          const service = services.find(s => change.serviceName === s.serviceName);
-          if (service) {
-            const ast = parse(service.sdl, { noLocation: true });
-            service.sdl = print(
-              patch(ast, change.schemaProposalChanges, { onError: errors.looseErrorHandler }),
-            );
-            if (change.serviceUrl) {
-              service.serviceUrl = change.serviceUrl;
-            }
+    async proposedSchemas(args: {
+      targetId: string;
+      proposalId: string;
+      cursor?: string | null;
+      pool: PostgresDatabasePool;
+    }) {
+      const services = await this.latestComposableSchemas({
+        targetId: args.targetId,
+        pool: args.pool,
+      });
+
+      await this.forEachProposalCheck(args, change => {
+        const service = services.find(s => change.serviceName === s.serviceName);
+        if (service) {
+          const ast = parse(service.sdl, { noLocation: true });
+          service.sdl = print(
+            patch(ast, change.schemaProposalChanges, { onError: errors.looseErrorHandler }),
+          );
+          if (change.serviceUrl) {
+            service.serviceUrl = change.serviceUrl;
           }
         }
-      } while (nextCursor && ++i < maxLoops);
+      });
 
       return services;
     },
+
+    async approveChanges(
+      conn: CommonQueryMethods,
+      records: {
+        change: Change;
+        proposalId: string;
+        service?: string;
+        targetId: string;
+      }[],
+    ) {
+      const values = records.map(
+        r => psql`(
+          ${generateChangeHash(r.change)}
+          ,${JSON.stringify(r.change)}
+          ,${r.proposalId}
+          ,${r.service ?? null}
+          ,${r.targetId}
+        )`,
+      );
+
+      await conn.query(psql`
+          INSERT INTO "proposal_approved_changes" (
+             hash
+            ,change
+            ,proposal_id
+            ,service
+            ,target_id
+          )
+          VALUES ${psql.join(values, psql.fragment`,`)}
+        `);
+    },
   };
 }

packages/services/workflows/src/tasks/schema-proposal-approval.ts

@@ -0,0 +1,55 @@
+/**
+ * When a schema proposal is approved, then all changes for all checks associated
+ * that proposal must be added to the approved changes list. This is to allow
+ * future checks/publishes/etc to validate their change against this approved list.
+ *
+ * This job listens for a schema proposal approval, loops over all the latest
+ * checks for that proposal, and inserts those change records into the approved list table.
+ */
+
+import { z } from 'zod';
+import { defineTask, implementTask } from '../kit.js';
+
+export const SchemaProposalApprovalTask = defineTask({
+  name: 'schemaProposalApproval',
+  schema: z.object({
+    proposalId: z.string(),
+    targetId: z.string(),
+  }),
+});
+
+export const task = implementTask(SchemaProposalApprovalTask, async args => {
+  const pool = args.context.pg;
+  try {
+    // collect all changes
+    const changes: Parameters<typeof args.context.schema.approveChanges>[1] = [];
+    await args.context.schema.forEachProposalCheck(
+      {
+        pool,
+        proposalId: args.input.proposalId,
+        targetId: args.input.targetId,
+      },
+      async check => {
+        const checkChanges = check.schemaProposalChanges.map(change => ({
+          change,
+          proposalId: args.input.proposalId,
+          service: check.serviceName ?? undefined,
+          targetId: args.input.targetId,
+        }));
+        changes.push(...checkChanges);
+      },
+    );
+
+    args.logger.info(
+      'Approving changes (count=%d, proposal=%s)',
+      changes.length,
+      args.input.proposalId,
+    );
+    // Approve all changes in a single call. If this becomes a bottleneck, then
+    // implement snapshots to allow inserting in chunks
+    await args.context.schema.approveChanges(pool, changes);
+  } catch (e: unknown) {
+    args.logger.error('Proposal approval failed from %s', String(e));
+    throw e;
+  }
+});