Add pg migration to track approved changes; ui; resolvers and storage

Author: jdolle

packages/migrations/src/actions/2026.03.28T00-00-00.proposal-change-tracking.ts

@@ -0,0 +1,43 @@
+import { type MigrationExecutor } from '../pg-migrator';
+
+export default {
+  name: '2026.03.28T00-00-00.proposal-change-tracking.ts',
+  run: ({ psql }) => psql`
+    CREATE TABLE IF NOT EXISTS "proposal_approved_changes" (
+      "id" UUID PRIMARY KEY DEFAULT uuid_generate_v4 (),
+
+      -- hash of the metadata. Because this is a hash, it has a slim
+      -- chance of conflicting. This is why we need the "change" to perform a more
+      -- exact match using the "@graphql-inspector/compare-changes" package
+      "hash" TEXT NOT NULL,
+
+      -- this is the exact change
+      "change" JSONB NOT NULL,
+
+      "proposal_id" UUID NOT NULL
+          REFERENCES "schema_proposals"("id"),
+
+      -- This is the version in which the approved change has been implemented.
+      -- It is for linking the change to the schema history
+      "schema_version_id" UUID
+          REFERENCES "schema_versions"("id"),
+
+      "service" TEXT,
+      "target_id" UUID NOT NULL
+          REFERENCES "targets"("id")
+          ON DELETE CASCADE
+    );
+
+    -- partial index used on publish to look up if a change is approved and not implemented
+    CREATE INDEX "target_change_not_implemented" ON "proposal_approved_changes" ("target_id", "hash")
+      WHERE "schema_version_id" IS NULL;
+
+    -- look up on a proposal, which changes have been implemented
+    -- and where they were implemented
+    CREATE INDEX "proposal_change_implemented" ON "proposal_approved_changes" ("target_id", "proposal_id")
+      WHERE "schema_version_id" IS NOT NULL;
+
+    -- look up on the schema history page, which proposal the change belonged to
+    CREATE INDEX "schema_implements_changes" ON "proposal_approved_changes" ("target_id", "schema_version_id");
+  `,
+} satisfies MigrationExecutor;

packages/migrations/src/run-pg-migrations.ts

@@ -184,5 +184,6 @@ export const runPGMigrations = async (args: { slonik: PostgresDatabasePool; runT
       await import('./actions/2026.02.24T00-00-00.proposal-composition'),
       await import('./actions/2026.02.25T00-00-00.oidc-integration-domains'),
       await import('./actions/2026.03.25T00-00-00.access-token-expiration'),
+      await import('./actions/2026.03.28T00-00-00.proposal-change-tracking'),
     ],
   });

packages/services/api/package.json

@@ -23,6 +23,7 @@
     "@date-fns/utc": "2.1.1",
     "@graphql-hive/core": "workspace:*",
     "@graphql-hive/signal": "1.0.0",
+    "@graphql-inspector/compare-changes": "0.1.0-alpha-20260328000712-2cf98744a012fa66614588ed89ba99719b0f461b",
     "@graphql-inspector/core": "7.1.2",
     "@graphql-tools/merge": "9.1.1",
     "@hive/cdn-script": "workspace:*",

packages/services/api/src/modules/proposals/module.graphql.mappers.ts

@@ -0,0 +1,7 @@
+export type SchemaProposalChangeDetailsMapper =
+  | {
+      schemaProposal: { id: string };
+      implementedBy?: { id: string } | null;
+    }
+  | null
+  | undefined;

packages/services/api/src/modules/proposals/module.graphql.ts

@@ -331,12 +331,6 @@ export default gql`
     """
     serviceName: String!
 
-    # @todo
-    # """
-    # The specific version of the proposal that this review is for.
-    # """
-    # schemaProposalVersion: SchemaProposalVersion
-
     """
     If null then this review is just a comment. Otherwise, the reviewer changed the state of the
     proposal as part of their review. E.g. The reviewer can approve a version with a comment.
@@ -375,6 +369,26 @@ export default gql`
 
   extend type SchemaChange {
     meta: SchemaChangeMeta
+
+    """
+    If this change was proposed by a schema proposal, then this returns
+    details about that proposal and where the change was implemented.
+    """
+    schemaProposalChangeDetails: SchemaProposalChangeDetails
+  }
+
+  type SchemaProposalChangeDetails {
+    """
+    If this schema change is associated with a proposal, this returns that proposal
+    This is useful for matching changes from a schema check or schema history/version to a proposal.
+    """
+    schemaProposal: SchemaProposal!
+
+    """
+    If this change is for a schema proposal, this is the schema version
+    (viewable in the History) where the change was implemented.
+    """
+    implementedBy: SchemaVersion
   }
 
   union SchemaChangeMeta =

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

@@ -1,22 +1,38 @@
 /**
  * This wraps the higher level logic with schema proposals.
  */
+import { SchemaProposalChangeDetailsMapper } from '../module.graphql.mappers';
+import DataLoader from 'dataloader';
 import { Inject, Injectable, Scope } from 'graphql-modules';
-import { TargetReferenceInput } from 'packages/libraries/core/src/client/__generated__/types';
+import type { TargetReferenceInput } from 'packages/libraries/core/src/client/__generated__/types';
+import { isChangeEqual } from '@graphql-inspector/compare-changes';
+import { Change } from '@graphql-inspector/core';
+import { traceFn } from '@hive/service-common';
+import { SchemaChangeType } from '@hive/storage';
 import type { SchemaProposalStage } from '../../../__generated__/types';
 import { HiveError } from '../../../shared/errors';
+import { cache } from '../../../shared/helpers';
 import { Session } from '../../auth/lib/authz';
 import { IdTranslator } from '../../shared/providers/id-translator';
 import { Logger } from '../../shared/providers/logger';
 import { PUB_SUB_CONFIG, type HivePubSub } from '../../shared/providers/pub-sub';
 import { Storage } from '../../shared/providers/storage';
 import { SchemaProposalStorage } from './schema-proposal-storage';
 
+type ChangeSelector = {
+  targetId: string;
+  change: SchemaChangeType;
+};
+
 @Injectable({
   scope: Scope.Operation,
 })
 export class SchemaProposalManager {
   private logger: Logger;
+  private approvedChangeProposalLoader: DataLoader<
+    ChangeSelector,
+    SchemaProposalChangeDetailsMapper
+  >;
 
   constructor(
     logger: Logger,
@@ -27,6 +43,69 @@ export class SchemaProposalManager {
     @Inject(PUB_SUB_CONFIG) private pubSub: HivePubSub,
   ) {
     this.logger = logger.child({ source: 'SchemaProposalsManager' });
+
+    const cacheKeyFn = function (selector: ChangeSelector) {
+      return `${selector.targetId}/${selector.change.id}`;
+    };
+    this.approvedChangeProposalLoader = new DataLoader(
+      async selectors => {
+        // map the list of keys in the order in which they are passed in so it can
+        // be mapped back to.
+        const keys = selectors.map(cacheKeyFn);
+
+        // combine selectors by targetId
+        const groupedSelectors = selectors.reduce((groups, selector) => {
+          const group = groups.get(selector.targetId);
+          if (!group) {
+            groups.set(selector.targetId, [selector.change]);
+          } else {
+            group.push(selector.change);
+          }
+          return groups;
+        }, new Map<string, SchemaChangeType[]>());
+
+        // fetch the groups
+        const approvedProposedChangesByCacheId = await Promise.all(
+          groupedSelectors.entries().map(async ([targetId, changes]) => {
+            const result = await this.proposalStorage.getEquivalentUnimplementedApprovedChanges({
+              changes: changes as unknown as Change<any>[],
+              targetId,
+            });
+            const entries = result
+              .map((apc, i) => {
+                if (apc) {
+                  return [cacheKeyFn({ change: changes[i], targetId }), apc] as const;
+                }
+              })
+              .filter(e => e !== undefined);
+            return Object.fromEntries(entries);
+          }),
+        );
+
+        // map back to the original selectors
+        return keys.map((cacheKey): SchemaProposalChangeDetailsMapper => {
+          for (const targetChangesByCacheId of approvedProposedChangesByCacheId) {
+            const approvedRecord = targetChangesByCacheId[cacheKey];
+            if (approvedRecord) {
+              return {
+                schemaProposal: {
+                  id: approvedRecord.proposalId,
+                },
+                implementedBy: approvedRecord.schemaVersionId
+                  ? {
+                      id: approvedRecord.schemaVersionId,
+                    }
+                  : null,
+              };
+            }
+          }
+          return null;
+        });
+      },
+      {
+        cacheKeyFn,
+      },
+    );
   }
 
   async subscribeToSchemaProposalCompositions(args: { proposalId: string }) {
@@ -116,6 +195,7 @@ export class SchemaProposalManager {
     };
   }
 
+  @cache<{ id: string }>(({ id }) => id)
   async getProposal(args: { id: string }) {
     const proposal = await this.proposalStorage.getProposal(args);
 
@@ -255,4 +335,90 @@ export class SchemaProposalManager {
 
     throw new HiveError('Not implemented');
   }
+
+  @traceFn('SchemaProposalManager.getChangeDetailsForCheck', {
+    initAttributes: input => ({
+      'hive.schema_proposal.id': input.schemaProposalId || '',
+      'hive.target.id': input.targetId,
+    }),
+    resultAttributes: result => ({
+      'hive.proposals.details.length': result?.length || 0,
+    }),
+  })
+  @cache<{
+    schemaProposalId: string;
+    targetId: string;
+  }>(({ schemaProposalId, targetId }) => `${targetId}/${schemaProposalId}`)
+  async _getImplementedVersionsByProposalId(args: { schemaProposalId: string; targetId: string }) {
+    return this.proposalStorage.getImplementedVersionsByProposalId({
+      schemaProposalId: args.schemaProposalId,
+      targetId: args.targetId,
+    });
+  }
+
+  /**
+   * If dealing with a non-schema proposal related check or schema version,
+   * then this function can search to find if any of the check's or historic changes
+   * are matches for approved schema proposal changes.
+   *
+   * Note that if you need the ChangeDetails object for a schema proposal's changes, then use
+   * `getProposalChangeDetails`.
+   **/
+  async getMatchingApprovedProposalChangeDetails({
+    targetId,
+    change,
+  }: {
+    targetId: string;
+    change: SchemaChangeType;
+  }) {
+    return this.approvedChangeProposalLoader.load({ targetId, change });
+  }
+
+  /**
+   * Schema proposals use checks under the hood which makes this a bit confusing.
+   * This function is for finding `proposalChangeDetails` for a schema proposal's
+   * associated check's changes -- effectively the proposed changes.
+   */
+  async getProposalChangeDetails({
+    schemaProposalId,
+    targetId,
+    change,
+  }: {
+    targetId: string;
+
+    // The ID of the schema proposal that is proposing this change
+    schemaProposalId: string;
+
+    // The change record. Used to compare against the returned implemented change
+    // to verify that they are equal.
+    change: SchemaChangeType;
+  }) {
+    // `_getImplementedVersionsByProposalId` is cached so it's safe to call multiple times.
+    // This is intended to be used by the SchemaChange resolver and would be called on every
+    // SchemaChange.
+    const implementedChanges = await this._getImplementedVersionsByProposalId({
+      schemaProposalId,
+      targetId,
+    });
+
+    // Verify which versions implement the associated changes using an exact compare
+    const implementation = implementedChanges.find(
+      implementedChange =>
+        isChangeEqual(
+          implementedChange.change as unknown as Change<any>,
+          change as unknown as Change<any>,
+        ) && implementedChange.schemaVersionId,
+    );
+    if (implementation) {
+      return {
+        schemaProposal: { id: schemaProposalId },
+        implementedBy: implementation?.schemaVersionId
+          ? {
+              id: implementation.schemaVersionId,
+            }
+          : null,
+      };
+    }
+    return null;
+  }
 }