Add support for external storages on lock, unlock, update, get, and delete

.github/workflows/pr-build.yaml

@@ -133,6 +133,11 @@ jobs:
           reporter: github-pr-review
           level: info
           locale: "US"
+          exclude: |
+            ./hub-js/package-lock.json
+            ./hub-js/package.json
+            ./go.mod
+            ./go.sum
       - name: Check links in *.md files
         if: always() # validate also *.md even if errors found in mdx files.
         uses: gaurav-nelson/github-action-markdown-link-check@v1

hub-js/graphql/local/schema.graphql

@@ -105,7 +105,33 @@ type TypeInstanceResourceVersionSpec {
   value: Any!
     @cypher(
       statement: """
-      RETURN apoc.convert.fromJsonMap(this.value)
+      MATCH (this)<-[:SPECIFIED_BY]-(rev:TypeInstanceResourceVersion)<-[:CONTAINS]-(ti:TypeInstance)
+      MATCH (this)-[:WITH_BACKEND]->(backendCtx)
+      MATCH (ti)-[:STORED_IN]->(backendRef)
+      WITH *
+      CALL apoc.when(
+          backendRef.abstract,
+          '
+              WITH {
+                abstract: backendRef.abstract,
+                builtinValue: apoc.convert.fromJsonMap(spec.value)
+              } AS value
+              RETURN value
+          ',
+          '
+              WITH {
+                abstract: backendRef.abstract,
+                fetchInput: {
+                   typeInstance: { resourceVersion: rev.resourceVersion, id: ti.id },
+                   backend: { context: backendCtx.context, id: backendRef.id}
+                }
+              } AS value
+              RETURN value
+          ',
+          {spec: this, rev: rev, ti: ti, backendRef: backendRef, backendCtx: backendCtx}
+      ) YIELD value as out
+
+      RETURN out.value
       """
     )
 
@@ -293,6 +319,15 @@ input UpdateTypeInstanceInput {
   The value property is optional. If not provided, previous value is used.
   """
   value: Any
+
+  """
+  The backend property is optional. If not provided, previous value is used.
+  """
+  backend: UpdateTypeInstanceBackendInput
+}
+
+input UpdateTypeInstanceBackendInput {
+  context: Any
 }
 
 input UpdateTypeInstancesInput {
@@ -440,26 +475,45 @@ type Mutation {
       CREATE (tir)-[:SPECIFIED_BY]->(spec)
 
       WITH ti, tir, spec, latestRevision, item
-      CALL apoc.do.when(
-          item.typeInstance.value IS NOT NULL,
+      MATCH (ti)-[:STORED_IN]->(storageRef:TypeInstanceBackendReference)
+
+      WITH ti, tir, spec, latestRevision, item, storageRef
+      CALL apoc.do.case([
+          storageRef.abstract AND item.typeInstance.value IS NOT NULL, // built-in: store new value
         '
           SET spec.value = apoc.convert.toJson(item.typeInstance.value) RETURN spec
         ',
+          storageRef.abstract AND item.typeInstance.value IS NULL, // built-in: no value, so use old one
         '
           MATCH (latestRevision)-[:SPECIFIED_BY]->(latestSpec: TypeInstanceResourceVersionSpec)
           SET spec.value = latestSpec.value RETURN spec
+        '
+        ],
+        '
+          RETURN spec // external storage, do nothing
         ',
-        {spec:spec, latestRevision: latestRevision, item: item}) YIELD value
+      {spec:spec, latestRevision: latestRevision, item: item}) YIELD value
+
+      // Handle the `backend.context`
+      WITH ti, tir, spec, latestRevision, item
+      CALL apoc.do.when(
+          item.typeInstance.backend IS NOT NULL,
+        '
+          CREATE (specBackend: TypeInstanceResourceVersionSpecBackend {context: apoc.convert.toJson(item.typeInstance.backend.context)})
+          RETURN specBackend
+        ',
+        '
+          CREATE (specBackend: TypeInstanceResourceVersionSpecBackend {context: apoc.convert.toJson(item.typeInstance.backend.context)})
+          RETURN specBackend
+        ',
+      {spec:spec, latestRevision: latestRevision, item: item}) YIELD value as backendRef
+      WITH ti, tir, spec, latestRevision, item, backendRef.specBackend as specBackend
+      CREATE (spec)-[:WITH_BACKEND]->(specBackend)
 
       // Handle the `metadata.attributes` property
       CREATE (metadata: TypeInstanceResourceVersionMetadata)
       CREATE (tir)-[:DESCRIBED_BY]->(metadata)
 
-      // TODO: Temporary don't allow backend update, will be fixed in follow-up PR
-      WITH *
-      MATCH (latestRevision)-[:SPECIFIED_BY]->(latestSpec: TypeInstanceResourceVersionSpec)-[:WITH_BACKEND]->(specBackend: TypeInstanceResourceVersionSpecBackend)
-      CREATE (spec)-[:WITH_BACKEND]->(specBackend)
-
       WITH ti, tir, latestRevision, metadata, item
       CALL apoc.do.when(
         item.typeInstance.attributes IS NOT NULL,

hub-js/package.json

@@ -21,10 +21,13 @@
   "dependencies": {
     "@godaddy/terminus": "^4.8.0",
     "@grpc/grpc-js": "^1.3.1",
+    "@types/lodash": "^4.14.179",
     "apollo-server-express": "^3.6.2",
+    "async-mutex": "^0.3.2",
     "express": "^4.17.0",
     "graphql": "^15.4.0",
     "graphql-tools": "^8.1.0",
+    "lodash": "^4.17.21",
     "long": "^5.2.0",
     "neo4j-driver": "^4.3.0",
     "neo4j-graphql-js": "~2.19.4",
@@ -44,9 +47,9 @@
     "@types/express-serve-static-core": "^4.17.19",
     "@types/node": "^16.4.13",
     "@types/ws": "^7.4.7",
-    "eslint": "^8.10.0",
     "@typescript-eslint/eslint-plugin": "^5.13.0",
     "@typescript-eslint/parser": "^5.13.0",
+    "eslint": "^8.10.0",
     "husky": "^4.0.0",
     "lint-staged": "^10.5.4",
     "npm-force-resolutions": "^0.0.10",

hub-js/proto/storage_backend.proto

@@ -22,6 +22,7 @@ message OnUpdateRequest {
   uint32 new_resource_version = 2;
   bytes new_value = 3;
   optional bytes context = 4;
+  optional string owner_id = 5;
 }
 
 message OnUpdateResponse {
@@ -31,6 +32,7 @@ message OnUpdateResponse {
 message OnDeleteRequest {
   string type_instance_id = 1;
   optional bytes context = 2;
+  optional string owner_id = 3;
 }
 
 message OnDeleteResponse {}

hub-js/src/generated/grpc/storage_backend.ts

@@ -24,6 +24,7 @@ export interface OnUpdateRequest {
   newResourceVersion: number;
   newValue: Uint8Array;
   context?: Uint8Array | undefined;
+  ownerId?: string | undefined;
 }
 
 export interface OnUpdateResponse {
@@ -33,6 +34,7 @@ export interface OnUpdateResponse {
 export interface OnDeleteRequest {
   typeInstanceId: string;
   context?: Uint8Array | undefined;
+  ownerId?: string | undefined;
 }
 
 export interface OnDeleteResponse {}
@@ -293,6 +295,7 @@ function createBaseOnUpdateRequest(): OnUpdateRequest {
     newResourceVersion: 0,
     newValue: new Uint8Array(),
     context: undefined,
+    ownerId: undefined,
   };
 }
 
@@ -313,6 +316,9 @@ export const OnUpdateRequest = {
     if (message.context !== undefined) {
       writer.uint32(34).bytes(message.context);
     }
+    if (message.ownerId !== undefined) {
+      writer.uint32(42).string(message.ownerId);
+    }
     return writer;
   },
 
@@ -335,6 +341,9 @@ export const OnUpdateRequest = {
         case 4:
           message.context = reader.bytes();
           break;
+        case 5:
+          message.ownerId = reader.string();
+          break;
         default:
           reader.skipType(tag & 7);
           break;
@@ -357,6 +366,7 @@ export const OnUpdateRequest = {
       context: isSet(object.context)
         ? bytesFromBase64(object.context)
         : undefined,
+      ownerId: isSet(object.ownerId) ? String(object.ownerId) : undefined,
     };
   },
 
@@ -375,6 +385,7 @@ export const OnUpdateRequest = {
         message.context !== undefined
           ? base64FromBytes(message.context)
           : undefined);
+    message.ownerId !== undefined && (obj.ownerId = message.ownerId);
     return obj;
   },
 
@@ -384,6 +395,7 @@ export const OnUpdateRequest = {
     message.newResourceVersion = object.newResourceVersion ?? 0;
     message.newValue = object.newValue ?? new Uint8Array();
     message.context = object.context ?? undefined;
+    message.ownerId = object.ownerId ?? undefined;
     return message;
   },
 };
@@ -447,7 +459,7 @@ export const OnUpdateResponse = {
 };
 
 function createBaseOnDeleteRequest(): OnDeleteRequest {
-  return { typeInstanceId: "", context: undefined };
+  return { typeInstanceId: "", context: undefined, ownerId: undefined };
 }
 
 export const OnDeleteRequest = {
@@ -461,6 +473,9 @@ export const OnDeleteRequest = {
     if (message.context !== undefined) {
       writer.uint32(18).bytes(message.context);
     }
+    if (message.ownerId !== undefined) {
+      writer.uint32(26).string(message.ownerId);
+    }
     return writer;
   },
 
@@ -477,6 +492,9 @@ export const OnDeleteRequest = {
         case 2:
           message.context = reader.bytes();
           break;
+        case 3:
+          message.ownerId = reader.string();
+          break;
         default:
           reader.skipType(tag & 7);
           break;
@@ -493,6 +511,7 @@ export const OnDeleteRequest = {
       context: isSet(object.context)
         ? bytesFromBase64(object.context)
         : undefined,
+      ownerId: isSet(object.ownerId) ? String(object.ownerId) : undefined,
     };
   },
 
@@ -505,13 +524,15 @@ export const OnDeleteRequest = {
         message.context !== undefined
           ? base64FromBytes(message.context)
           : undefined);
+    message.ownerId !== undefined && (obj.ownerId = message.ownerId);
     return obj;
   },
 
   fromPartial(object: DeepPartial<OnDeleteRequest>): OnDeleteRequest {
     const message = createBaseOnDeleteRequest();
     message.typeInstanceId = object.typeInstanceId ?? "";
     message.context = object.context ?? undefined;
+    message.ownerId = object.ownerId ?? undefined;
     return message;
   },
 };

hub-js/src/index.ts

@@ -12,8 +12,9 @@ import { GraphQLSchema } from "graphql";
 import { assertSchemaOnDatabase, getSchemaForMode, HubMode } from "./schema";
 import { config } from "./config";
 import { logger } from "./logger";
-import { ensureCoreStorageTypeInstance } from "./local/mutation/register-built-in-storage";
+import { ensureCoreStorageTypeInstance } from "./local/resolver/mutation/register-built-in-storage";
 import DelegatedStorageService from "./local/storage/service";
+import UpdateArgsContainer from "./local/storage/update-args-container";
 
 async function main() {
   logger.info("Using Neo4j database", { endpoint: config.neo4j.endpoint });
@@ -68,7 +69,13 @@ async function setupHttpServer(
   const delegatedStorage = new DelegatedStorageService(driver);
   const apolloServer = new ApolloServer({
     schema,
-    context: { driver, delegatedStorage },
+    context: () => {
+      return {
+        driver,
+        delegatedStorage,
+        updateArgs: new UpdateArgsContainer(),
+      };
+    },
   });
   await apolloServer.start();
   apolloServer.applyMiddleware({ app });