Merge pull request #1090 from neo4j/feature/subscriptions-meta

Create subscriptions metadata

.eslintrc.js

@@ -19,7 +19,7 @@ const baseExtends = [
 
 const baseRules = {
     "no-underscore-dangle": ["warn", { allow: ["_on", "__resolveType", "__typename"], allowAfterThis: true }], // TODO Refactor instances of _varName to remove dangling underscore, and delete this line (also fixes @typescript-eslint/naming-convention)
-    "no-param-reassign": "warn", // Dangerous to have this off (out-of-scope side effects are bad), but there are some valid reasons for violations (Array.reduce(), setting GraphQL context, etc.), so use comments to disable ESLint for these
+    "no-param-reassign": ["warn", { props: false }], // Dangerous to have this off (out-of-scope side effects are bad), but there are some valid reasons for violations (Array.reduce(), setting GraphQL context, etc.), so use comments to disable ESLint for these
     "max-classes-per-file": "off", // Stylistic decision - we can judge whether there are too many classes in one file during code review
     "eslint-comments/no-unused-disable": "error", // Turn on optional rule to report eslint-disable comments having no effect
     "class-methods-use-this": "off",

packages/graphql/src/constants.ts

@@ -102,3 +102,5 @@ export enum RelationshipQueryDirectionOption {
     DIRECTED_ONLY = "DIRECTED_ONLY",
     UNDIRECTED_ONLY = "UNDIRECTED_ONLY",
 }
+
+export const META_CYPHER_VARIABLE = "meta";

packages/graphql/src/schema/resolvers/create.ts

@@ -23,6 +23,8 @@ import { translateCreate } from "../../translate";
 import { Node } from "../../classes";
 import { Context } from "../../types";
 import getNeo4jResolveTree from "../../utils/get-neo4j-resolve-tree";
+import { EventMeta, RawEventMeta } from "../../subscriptions/event-meta";
+import { serializeNeo4jValue } from "../../utils/neo4j-serializers";
 
 export default function createResolver({ node }: { node: Node }) {
     async function resolve(_root: any, args: any, _context: unknown, info: GraphQLResolveInfo) {
@@ -41,6 +43,17 @@ export default function createResolver({ node }: { node: Node }) {
             (selection) => selection.kind === "Field" && selection.name.value === node.plural
         ) as FieldNode;
         const nodeKey = nodeProjection?.alias ? nodeProjection.alias.value : nodeProjection?.name?.value;
+
+        const subscriptionsPlugin = context.plugins?.subscriptions;
+        if (subscriptionsPlugin) {
+            const metaData: RawEventMeta[] = executeResult.records[0]?.meta || [];
+            for (const meta of metaData) {
+                const serializedMeta = serializeEventMeta(meta);
+                // eslint-disable-next-line @typescript-eslint/no-floating-promises
+                subscriptionsPlugin.publish(serializedMeta);
+            }
+        }
+
         return {
             info: {
                 bookmark: executeResult.bookmark,
@@ -56,3 +69,26 @@ export default function createResolver({ node }: { node: Node }) {
         args: { input: `[${node.name}CreateInput!]!` },
     };
 }
+
+function serializeProperties(properties: Record<string, any> | undefined): Record<string, any> | undefined {
+    if (!properties) {
+        return undefined;
+    }
+
+    return Object.entries(properties).reduce((serializedProps, [k, v]) => {
+        serializedProps[k] = serializeNeo4jValue(v);
+        return serializedProps;
+    }, {} as Record<string, any>);
+}
+
+function serializeEventMeta(event: RawEventMeta): EventMeta {
+    return {
+        id: serializeNeo4jValue(event.id),
+        timestamp: serializeNeo4jValue(event.timestamp),
+        event: event.event,
+        properties: {
+            old: serializeProperties(event.properties.old),
+            new: serializeProperties(event.properties.new),
+        },
+    } as EventMeta;
+}

packages/graphql/src/schema/resolvers/wrapper.ts

@@ -74,6 +74,7 @@ export const wrapResolver =
         context.relationships = relationships;
         context.schema = schema;
         context.plugins = plugins;
+        context.subscriptionsEnabled = Boolean(context.plugins?.subscriptions);
 
         if (!context.jwt) {
             if (context.plugins?.auth) {

packages/graphql/src/subscriptions/event-meta.ts

@@ -17,25 +17,38 @@
  * limitations under the License.
  */
 
-export type EventMeta =
+import * as neo4j from "neo4j-driver";
+
+export type RawEventMeta = {
+    event: "create" | "update" | "delete";
+    properties: {
+        old: Record<string, any>;
+        new: Record<string, any>;
+    };
+    id: neo4j.Integer | string | number;
+    timestamp: neo4j.Integer | string | number;
+};
+
+export type EventMeta = (
     | {
           event: "create";
-          id: string;
-          oldProps: undefined;
-          newProps: Record<string, any>;
-          timestamp: number;
+          properties: {
+              old: undefined;
+              new: Record<string, any>;
+          };
       }
     | {
           event: "update";
-          id: string;
-          oldProps: Record<string, any>;
-          newProps: Record<string, any>;
-          timestamp: number;
+          properties: {
+              old: Record<string, any>;
+              new: Record<string, any>;
+          };
       }
     | {
           event: "delete";
-          id: string;
-          oldProps: Record<string, any>;
-          newProps: undefined;
-          timestamp: number;
-      };
+          properties: {
+              old: Record<string, any>;
+              new: undefined;
+          };
+      }
+) & { id: number; timestamp: number };

packages/graphql/src/translate/connect-or-create/create-connect-or-create-and-params.ts

@@ -44,13 +44,15 @@ export function createConnectOrCreateAndParams({
     relationField,
     refNode,
     context,
+    withVars,
 }: {
     input: CreateOrConnectInput[] | CreateOrConnectInput;
     varName: string;
     parentVar: string;
     relationField: RelationField;
     refNode: Node;
     context: Context;
+    withVars: string[];
 }): CypherStatement {
     const statements = asArray(input).map((inputItem, index): CypherStatement => {
         const subqueryBaseName = `${varName}${index}`;
@@ -64,7 +66,7 @@ export function createConnectOrCreateAndParams({
         });
     });
     const [statement, params] = joinStatements(statements);
-    return [wrapInCall(statement, parentVar), params];
+    return [wrapInCall(statement, withVars), params];
 }
 
 function createConnectOrCreatePartialStatement({

packages/graphql/src/translate/connection/create-connection-and-params.ts

@@ -34,7 +34,7 @@ import { createOffsetLimitStr } from "../../schema/pagination";
 import filterInterfaceNodes from "../../utils/filter-interface-nodes";
 import { getRelationshipDirection } from "../cypher-builder/get-relationship-direction";
 import { CypherStatement } from "../types";
-import { isString } from "../../utils/utils";
+import { asArray, isString, removeDuplicates } from "../../utils/utils";
 import { generateMissingOrAliasedFields } from "../utils/resolveTree";
 
 function createConnectionAndParams({
@@ -43,17 +43,20 @@ function createConnectionAndParams({
     context,
     nodeVariable,
     parameterPrefix,
+    withVars,
 }: {
     resolveTree: ResolveTree;
     field: ConnectionField;
     context: Context;
     nodeVariable: string;
     parameterPrefix?: string;
+    withVars?: string[];
 }): CypherStatement {
     let globalParams = {};
-    let nestedConnectionFieldParams;
+    let nestedConnectionFieldParams: any;
 
-    let subquery = ["CALL {", `WITH ${nodeVariable}`];
+    const withVarsAndNodeName = removeDuplicates([...asArray(withVars), nodeVariable]);
+    let subquery = ["CALL {", `WITH ${withVarsAndNodeName.join(", ")}`];
 
     const sortInput = (resolveTree.args.sort ?? []) as ConnectionSortArg[];
     // Fields of {edge, node} to sort on. A simple resolve tree will be added if not in selection set
@@ -181,6 +184,7 @@ function createConnectionAndParams({
                                 parameterPrefix: `${parameterPrefix ? `${parameterPrefix}.` : `${nodeVariable}_`}${
                                     resolveTree.alias
                                 }.edges.node`,
+                                withVars: withVarsAndNodeName,
                             });
                             nestedSubqueries.push(nestedConnection[0]);
 

packages/graphql/src/translate/create-create-and-params.ts

@@ -26,6 +26,8 @@ import createSetRelationshipPropertiesAndParams from "./create-set-relationship-
 import mapToDbProperty from "../utils/map-to-db-property";
 import { createConnectOrCreateAndParams } from "./connect-or-create/create-connect-or-create-and-params";
 import createRelationshipValidationStr from "./create-relationship-validation-string";
+import { createEventMeta } from "./subscriptions/create-event-meta";
+import { filterMetaVariable } from "./subscriptions/filter-meta-variable";
 
 interface Res {
     creates: string[];
@@ -45,6 +47,7 @@ function createCreateAndParams({
     withVars,
     insideDoWhen,
     includeRelationshipValidation,
+    topLevelNodeVariable,
 }: {
     input: any;
     varName: string;
@@ -53,6 +56,7 @@ function createCreateAndParams({
     withVars: string[];
     insideDoWhen?: boolean;
     includeRelationshipValidation?: boolean;
+    topLevelNodeVariable?: string;
 }): [string, any] {
     function reducer(res: Res, [key, value]: [string, any]): Res {
         const varNameKey = `${varName}_${key}`;
@@ -87,7 +91,9 @@ function createCreateAndParams({
                             return;
                         }
 
-                        res.creates.push(`\nWITH ${withVars.join(", ")}`);
+                        if (!context.subscriptionsEnabled) {
+                            res.creates.push(`\nWITH ${withVars.join(", ")}`);
+                        }
 
                         const baseName = `${varNameKey}${relationField.union ? "_" : ""}${unionTypeName}${index}`;
                         const nodeName = `${baseName}_node`;
@@ -100,6 +106,7 @@ function createCreateAndParams({
                             varName: nodeName,
                             withVars: [...withVars, nodeName],
                             includeRelationshipValidation: false,
+                            topLevelNodeVariable,
                         });
                         res.creates.push(recurse[0]);
                         res.params = { ...res.params, ...recurse[1] };
@@ -161,6 +168,7 @@ function createCreateAndParams({
                         relationField,
                         refNode,
                         context,
+                        withVars,
                     });
                     res.creates.push(connectOrCreateQuery);
                     res.params = { ...res.params, ...connectOrCreateParams };
@@ -245,6 +253,12 @@ function createCreateAndParams({
         params: {},
     });
 
+    if (context.subscriptionsEnabled) {
+        const eventWithMetaStr = createEventMeta({ event: "create", nodeVariable: varName });
+        const withStrs = [eventWithMetaStr];
+        creates.push(`WITH ${withStrs.join(", ")}, ${filterMetaVariable(withVars).join(", ")}`);
+    }
+
     const forbiddenString = insideDoWhen ? `\\"${AUTH_FORBIDDEN_ERROR}\\"` : `"${AUTH_FORBIDDEN_ERROR}"`;
 
     if (node.auth) {
@@ -261,6 +275,7 @@ function createCreateAndParams({
             params = { ...params, ...bindAndParams[1] };
         }
     }
+
     if (meta?.authStrs.length) {
         creates.push(`WITH ${withVars.join(", ")}`);
         creates.push(`CALL apoc.util.validate(NOT(${meta.authStrs.join(" AND ")}), ${forbiddenString}, [0])`);

packages/graphql/src/translate/create-interface-projection-and-params.ts

@@ -18,7 +18,7 @@
  */
 
 import { ResolveTree } from "graphql-parse-resolve-info";
-import { Node } from "../classes";
+import { asArray, removeDuplicates } from "../utils/utils";
 import { AUTH_FORBIDDEN_ERROR } from "../constants";
 import { ConnectionField, Context, InterfaceWhereArg, RelationField } from "../types";
 import filterInterfaceNodes from "../utils/filter-interface-nodes";
@@ -32,20 +32,20 @@ function createInterfaceProjectionAndParams({
     resolveTree,
     field,
     context,
-    node,
     nodeVariable,
     parameterPrefix,
+    withVars,
 }: {
     resolveTree: ResolveTree;
     field: RelationField;
     context: Context;
-    node: Node;
     nodeVariable: string;
     parameterPrefix?: string;
+    withVars?: string[];
 }): { cypher: string; params: Record<string, any> } {
     let globalParams = {};
     let params: { args?: any } = {};
-
+    const fullWithVars = removeDuplicates([...asArray(withVars), nodeVariable]);
     const relTypeStr = `[:${field.type}]`;
 
     const { inStr, outStr } = getRelationshipDirection(field, resolveTree.args);
@@ -61,7 +61,7 @@ function createInterfaceProjectionAndParams({
     const subqueries = referenceNodes.map((refNode) => {
         const param = `${nodeVariable}_${refNode.name}`;
         const subquery = [
-            `WITH ${nodeVariable}`,
+            `WITH ${fullWithVars.join(", ")}`,
             `MATCH (${nodeVariable})${inStr}${relTypeStr}${outStr}(${param}:${refNode.name})`,
         ];
 
@@ -192,7 +192,6 @@ function createInterfaceProjectionAndParams({
                     resolveTree: interfaceResolveTree,
                     field: relationshipField,
                     context,
-                    node: refNode,
                     nodeVariable: param,
                 });
                 subquery.push(interfaceProjection.cypher);
@@ -208,10 +207,10 @@ function createInterfaceProjectionAndParams({
 
         return subquery.join("\n");
     });
-    const interfaceProjection = [`WITH ${nodeVariable}`, "CALL {", subqueries.join("\nUNION\n"), "}"];
+    const interfaceProjection = [`WITH ${fullWithVars.join(", ")}`, "CALL {", subqueries.join("\nUNION\n"), "}"];
 
     if (field.typeMeta.array) {
-        interfaceProjection.push(`WITH ${nodeVariable}, collect(${field.fieldName}) AS ${field.fieldName}`);
+        interfaceProjection.push(`WITH ${fullWithVars.join(", ")}, collect(${field.fieldName}) AS ${field.fieldName}`);
     }
 
     if (Object.keys(whereArgs).length) {

packages/graphql/src/translate/create-update-and-params.ts

@@ -326,6 +326,7 @@ function createUpdateAndParams({
                             relationField,
                             refNode,
                             context,
+                            withVars,
                         });
                         subquery.push(connectOrCreateQuery);
                         res.params = { ...res.params, ...connectOrCreateParams };

packages/graphql/src/translate/subscriptions/create-event-meta.test.ts

@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { createEventMeta } from "./create-event-meta";
+
+describe("createEventMeta", () => {
+    test("create", () => {
+        expect(createEventMeta({ event: "create", nodeVariable: "this0" })).toBe(
+            `meta + { event: "create", id: id(this0), properties: { old: null, new: this0 { .* } }, timestamp: timestamp() } AS meta`
+        );
+    });
+
+    test("update", () => {
+        expect(createEventMeta({ event: "update", nodeVariable: "this" })).toBe(
+            `meta + { event: "update", id: id(this), properties: { old: this { .* }, new: this { .* } }, timestamp: timestamp() } AS meta`
+        );
+    });
+
+    test("delete", () => {
+        expect(createEventMeta({ event: "delete", nodeVariable: "this" })).toBe(
+            `meta + { event: "delete", id: id(this), properties: { old: this { .* }, new: null }, timestamp: timestamp() } AS meta`
+        );
+    });
+});