implement schema garbage collector endpoint

Author: StarpTech

README.md

@@ -39,7 +39,7 @@ POST - `/schema/push` Creates a new graph and schema for a service.
 <summary>Example Request</summary>
 <p>
 
-```json
+```jsonc
 {
   "type_defs": "type Query { hello: String }",
   "version": "1",
@@ -57,10 +57,10 @@ POST - `/schema/compose` Returns the last registered schema definition of all se
 <summary>Example Request</summary>
 <p>
 
-```json
+```jsonc
 {
   "graph_name": "my_graph",
-  "services": [{ "name": "foo", "version": "1" }]
+  "services": [{ "name": "foo", "version": "1" }] // if versions can't be found it fails
 }
 ```
 
@@ -73,7 +73,7 @@ PUT - `/schema/deactivate` Deactivates a schema by id. The schema will no longer
 <summary>Example Request</summary>
 <p>
 
-```json
+```jsonc
 {
   "graph_name": "my_graph",
   "schemaId": "916348424"
@@ -83,6 +83,21 @@ PUT - `/schema/deactivate` Deactivates a schema by id. The schema will no longer
 </p>
 </details>
 
+POST - `/schema/garbage_collect` Removes all schemas except the most recent N of every service. Returns the removed schemas. This could be called by a [trigger](https://developers.cloudflare.com/workers/platform/cron-triggers).
+
+<details>
+<summary>Example Request</summary>
+<p>
+
+```jsonc
+{
+  "num_schemas_keep": 10 // minimum is 10
+}
+```
+
+</p>
+</details>
+
 ### Validation
 
 POST - `/schema/diff` Returns the schema report of all services and the provided new schema.
@@ -147,9 +162,9 @@ DELETE - `/persisted_query` Deletes persisted query from KV Storage.
 <summary>Example Request</summary>
 <p>
 
-```json
+```jsonc
 {
-  "key": "apq:foo"
+  "key": "foo"
 }
 ```
 
@@ -185,6 +200,7 @@ npm run dev
 ### Benchmark
 
 Run a benchmark with:
+
 ```
 docker run -e SECRET=<basic_auth_secret> -e URL=<worker_url> -i loadimpact/k6 run - < benchmark/composed-schema.js
 ```

src/index.ts

@@ -12,6 +12,7 @@ import { deletePersistedQuery } from './routes/delete-persisted-query'
 import { healthcheck } from './routes/healthcheck'
 import { deactivateSchema } from './routes/deactivate-schema'
 import { getGraphs } from './routes/get-graphs'
+import { garbageCollectSchemas } from './routes/garbage-collect'
 
 const API = new Router()
 
@@ -27,6 +28,11 @@ API.add(
   '/schema/compose',
   compose(basicAuth, getComposedSchemaByVersions),
 )
+API.add(
+  'POST',
+  '/schema/garbage_collect',
+  compose(basicAuth, garbageCollectSchemas),
+)
 
 // Tooling
 API.add('POST', '/schema/validate', compose(basicAuth, getSchemaValidation))

src/repositories/Graph.ts

@@ -11,10 +11,6 @@ export interface Graph {
   created_at: number
 }
 
-export interface GraphIndex {
-  name: string
-}
-
 export type NewGraph = Omit<Graph, 'created_at' | 'updated_at' | 'uid'>
 
 export const key_owner = () => `graphs`
@@ -25,19 +21,19 @@ export function find(name: string) {
   return DB.read<Graph>(GRAPHS, key, 'json')
 }
 
-export async function list(): Promise<GraphIndex[]> {
+export async function list(): Promise<Graph['name'][]> {
   const key = key_owner()
-  return (await DB.read<GraphIndex[]>(GRAPHS, key, 'json')) || []
+  return (await DB.read<Graph['name'][]>(GRAPHS, key, 'json')) || []
 }
 
-export function syncIndex(versions: GraphIndex[]) {
+export function syncIndex(versions: string[]) {
   const key = key_owner()
   return DB.write(GRAPHS, key, versions)
 }
 
 export function remove(name: string) {
   const key = key_item(name)
-  return DB.read<Graph>(GRAPHS, key, 'json')
+  return DB.remove(GRAPHS, key)
 }
 
 export function save(item: Graph) {
@@ -57,9 +53,7 @@ export async function insert(graph: NewGraph) {
     return false
   }
 
-  let allGraphs = (await list()).concat({
-    name: values.name,
-  })
+  let allGraphs = (await list()).concat(values.name)
 
   if (!(await syncIndex(allGraphs))) {
     return false

src/repositories/Schema.ts

@@ -10,7 +10,7 @@ declare const SCHEMAS: KV.Namespace
 export interface Schema {
   uid: string
   graph_name: string
-  service_id: string
+  service_name: string
   is_active: boolean
   hash: string
   type_defs: string
@@ -20,7 +20,7 @@ export interface Schema {
 
 export interface SchemaIndex {
   uid: string
-  service_id: string
+  service_name: string
   graph_name: string
   hash: string
 }
@@ -57,7 +57,7 @@ export function syncIndex(graph_name: string, versions: SchemaIndex[]) {
 
 export function remove(graph_name: string, uid: string) {
   const key = key_item(graph_name, uid)
-  return DB.read<Schema>(SCHEMAS, key, 'json')
+  return DB.remove(SCHEMAS, key)
 }
 
 export function save(item: Schema) {
@@ -70,7 +70,7 @@ export async function insert(schema: NewSchema) {
     uid: ulid(),
     graph_name: schema.graph_name,
     hash: fnv1a(schema.type_defs).toString(),
-    service_id: schema.service_id,
+    service_name: schema.service_name,
     is_active: schema.is_active,
     type_defs: schema.type_defs,
     created_at: Date.now(),
@@ -83,7 +83,7 @@ export async function insert(schema: NewSchema) {
 
   let allSchemas = (await list(schema.graph_name)).concat({
     uid: values.uid,
-    service_id: values.service_id,
+    service_name: values.service_name,
     graph_name: values.graph_name,
     hash: values.hash,
   })

src/repositories/SchemaVersion.ts

@@ -52,6 +52,15 @@ export function find(graphName: string, serviceName: string, version: string) {
   return DB.read<SchemaVersion>(VERSIONS, key, 'json')
 }
 
+export function remove(
+  graphName: string,
+  serviceName: string,
+  version: string,
+) {
+  const key = key_item(graphName, serviceName, version)
+  return DB.remove(VERSIONS, key)
+}
+
 export function save(
   graphName: string,
   serviceName: string,

src/routes/add-persisted-query.test.ts

@@ -1,9 +1,9 @@
 import test from 'ava'
 import { addPersistedQuery } from './add-persisted-query'
-import { NewNamespace, Request, Response } from '../test-utils'
+import { NewKVNamespace, Request, Response } from '../test-utils'
 
 test.serial('Should store PQ from KV', async (t) => {
-  const store = NewNamespace({
+  const store = NewKVNamespace({
     name: 'PERSISTED_QUERIES',
   })
 
@@ -19,7 +19,7 @@ test.serial('Should store PQ from KV', async (t) => {
 test.serial(
   'Should return validation error because no query was provided',
   async (t) => {
-    const store = NewNamespace({
+    const store = NewKVNamespace({
       name: 'PERSISTED_QUERIES',
     })
 
@@ -36,7 +36,7 @@ test.serial(
   },
 )
 test.serial('Should accept ttl values', async (t) => {
-  const store = NewNamespace({
+  const store = NewKVNamespace({
     name: 'PERSISTED_QUERIES',
   })
 

src/routes/deactivate-schema.test.ts

@@ -1,12 +1,12 @@
 import test from 'ava'
-import { createEmptyNamespaces, Request, Response } from '../test-utils'
+import { createEmptyKVNamespaces, Request, Response } from '../test-utils'
 import { SchemaResponseModel, SuccessResponse } from '../types'
 import { deactivateSchema } from './deactivate-schema'
 import { getComposedSchema } from './get-composed-schema'
 import { registerSchema } from './register-schema'
 
 test.serial('Should deactivate schema', async (t) => {
-  createEmptyNamespaces(['GRAPHS', 'SERVICES', 'SCHEMAS', 'VERSIONS'])
+  createEmptyKVNamespaces(['GRAPHS', 'SERVICES', 'SCHEMAS', 'VERSIONS'])
 
   let req = Request('POST', '', {
     type_defs: 'type Query { hello: String }',

src/routes/delete-persisted-query.test.ts

@@ -1,11 +1,11 @@
 import test from 'ava'
-import { NewNamespace, Request, Response } from '../test-utils'
+import { NewKVNamespace, Request, Response } from '../test-utils'
 import { addPersistedQuery } from './add-persisted-query'
 import { getPersistedQuery } from './get-persisted-query'
 import { deletePersistedQuery } from './delete-persisted-query'
 
 test.serial('Should delete PQ from KV', async (t) => {
-  NewNamespace({
+  NewKVNamespace({
     name: 'PERSISTED_QUERIES',
   })
 
@@ -35,7 +35,7 @@ test.serial('Should delete PQ from KV', async (t) => {
 })
 
 test.serial('Should return 400 when key was not provided', async (t) => {
-  NewNamespace({
+  NewKVNamespace({
     name: 'PERSISTED_QUERIES',
   })
 

src/routes/garbage-collect.test.ts

@@ -0,0 +1,79 @@
+import test from 'ava'
+import { createEmptyKVNamespaces, Request, Response } from '../test-utils'
+import {
+  GarbageCollectResponseModel,
+  SchemaResponseModel,
+  SuccessResponse,
+} from '../types'
+import { garbageCollectSchemas } from './garbage-collect'
+import { getComposedSchema } from './get-composed-schema'
+import { registerSchema } from './register-schema'
+
+test.serial(
+  'Should keep the most recent 10 schemas of every servcie in the graph',
+  async (t) => {
+    createEmptyKVNamespaces(['GRAPHS', 'SERVICES', 'SCHEMAS', 'VERSIONS'])
+
+    for (let i = 0; i < 15; i++) {
+      let req = Request('POST', '', {
+        type_defs: `type Query { hello${i}: String }`,
+        version: i.toString(),
+        service_name: `foo`,
+        graph_name: 'my_graph',
+      })
+      let res = Response()
+      await registerSchema(req, res)
+      t.is(res.statusCode, 200)
+
+      req = Request('POST', '', {
+        type_defs: `type Query { world${i}: String }`,
+        version: i.toString(),
+        service_name: `bar`,
+        graph_name: 'my_graph',
+      })
+      res = Response()
+      await registerSchema(req, res)
+      t.is(res.statusCode, 200)
+    }
+
+    let req = Request('POST', '', {
+      num_schemas_keep: 10,
+    })
+    let res = Response()
+    await garbageCollectSchemas(req, res)
+
+    t.is(res.statusCode, 200)
+
+    const result = (res.body as any) as SuccessResponse<
+      GarbageCollectResponseModel[]
+    >
+
+    t.is(result.success, true)
+    t.is(result.data.length, 10) // removed 5 schemas per service
+
+    t.truthy(result.data[0].schemaId)
+    t.is(result.data[0].service_name, 'foo')
+    t.is(result.data[0].graph_name, 'my_graph')
+
+    t.truthy(result.data[6].schemaId)
+    t.is(result.data[6].service_name, 'bar')
+    t.is(result.data[6].graph_name, 'my_graph')
+
+    req = Request('GET', 'graph_name=my_graph')
+    res = Response()
+    await getComposedSchema(req, res)
+
+    t.is(res.statusCode, 200)
+
+    const composed = (res.body as any) as SuccessResponse<SchemaResponseModel[]>
+
+    t.is(composed.success, true)
+    t.is(composed.data.length, 2)
+
+    t.is(composed.data[0].version, '14')
+    t.is(composed.data[0].service_name, 'foo')
+
+    t.is(composed.data[1].version, '14')
+    t.is(composed.data[1].service_name, 'bar')
+  },
+)