WIP: add different implementation sketches

.gitignore

@@ -4,5 +4,6 @@
 .envrc
 .swc
 node_modules
+ca-cert*
 
 dist

README.md

@@ -19,6 +19,16 @@ it is done with running the SQL command below, and restating the database server
 ALTER SYSTEM SET wal_level = logical;
 ```
 
+When registering a new region on a DigitalOcean postgres server, the default user doesn't have the required roles to set up a subscription.
+On DO we can use [aiven-extras](https://github.com/aiven/aiven-extras) to create subs without root access.
+The current branch is utilizing just that. But it needs a setup step executed on each database, that is registered as a region
+
+Run this in a `psql` shell
+
+```sql
+CREATE EXTENSION aiven_extras;
+```
+
 Note: Postgres subscriptions (which we use) in the same db server don't work that easily; easiest way to get things going is to set up multiple db servers locally.
 
 ## Project description
@@ -52,11 +62,10 @@ Organizations may be granted access to any given region. That action creates a n
 ## Steps to flex this POC
 
 Using the exposed graphql explorer, you can go ahead and
-- create a user 
+
+- create a user
 - create an organisation
 - add the user to the organisation
 - create regions & associate them with an organisation
 - create a resource in the default organisation, or for a specific organisation & region
 - etc.
-
-  

knexfile.ts

@@ -1,8 +1,26 @@
-export default {
+import { Knex } from 'knex'
+import fs from 'fs'
+import path from 'path'
+
+console.log(`foobar ${process.env.POSTGRES_CA_CERT_PATH}`)
+
+const config: Knex.Config = {
   client: 'pg',
-  connection: process.env.POSTGRES_URL,
+  connection: {
+    connectionString: process.env.POSTGRES_URL,
+    ssl: process.env.POSTGRES_CA_CERT_PATH
+      ? {
+          ca: fs.readFileSync(
+            path.resolve(__dirname, process.env.POSTGRES_CA_CERT_PATH)
+          ),
+          rejectUnauthorized: true
+        }
+      : undefined
+  },
   migrations: {
     directory: 'src/migrations',
     extension: 'ts'
   }
 }
+
+export default config

package.json

@@ -32,6 +32,7 @@
   "dependencies": {
     "@apollo/server": "^4.10.0",
     "crypto-random-string": "^3.0.0",
+    "dataloader": "^2.2.2",
     "dotenv": "^16.4.1",
     "graphql": "^16.8.1",
     "graphql-scalars": "^1.22.4",

parse_cert.py

@@ -0,0 +1,8 @@
+from pathlib import Path
+import json
+
+cert = Path("./ca-cert").read_text()
+
+cert_json = json.dumps({"cert": cert})
+
+Path("./ca-cert.json").write_text(cert_json)

pnpm-lock.yaml

@@ -11,6 +11,9 @@ dependencies:
   crypto-random-string:
     specifier: ^3.0.0
     version: 3.3.1
+  dataloader:
+    specifier: ^2.2.2
+    version: 2.2.2
   dotenv:
     specifier: ^16.4.1
     version: 16.4.1
@@ -1599,6 +1602,10 @@ packages:
       type-fest: 0.8.1
     dev: false
 
+  /dataloader@2.2.2:
+    resolution: {integrity: sha512-8YnDaaf7N3k/q5HnTJVuzSyLETjoZjVmHc4AeKAzOvKHEFQKcn64OKBfzHYtE9zGjctNM7V9I0MfnUVLpi7M5g==}
+    dev: false
+
   /date-fns@2.30.0:
     resolution: {integrity: sha512-fnULvOpxnC5/Vg3NCiWelDsLiUc9bRwAPs/+LfTLNvetFCtCTN+yQz15C/fs4AwX1R9K5GLtLfn8QW+dWisaAw==}
     engines: {node: '>=0.11'}

src/app.ts

@@ -1,31 +1,31 @@
-import { ApolloServer } from "@apollo/server";
-import { resolvers } from "./resolvers";
-import { startStandaloneServer } from "@apollo/server/standalone";
-import { readFileSync } from "fs";
-import { typeDefs as scalarTypeDefs } from "graphql-scalars";
-import { migrateAll } from "./services/databaseManagement";
+import { ApolloServer } from '@apollo/server'
+import { resolvers } from './resolvers'
+import { startStandaloneServer } from '@apollo/server/standalone'
+import { readFileSync } from 'fs'
+import { typeDefs as scalarTypeDefs } from 'graphql-scalars'
+import { migrateAll } from './services/databaseManagement'
 
-const typeDefs = readFileSync("src/schema.graphql", { encoding: "utf-8" });
+const typeDefs = readFileSync('src/schema.graphql', { encoding: 'utf-8' })
 
 // The ApolloServer constructor requires two parameters: your schema
 // definition and your set of resolvers.
 const server = new ApolloServer({
   typeDefs: [typeDefs, ...scalarTypeDefs],
-  resolvers,
-});
+  resolvers
+})
 
 const startServer = async (): Promise<void> => {
   const { url } = await startStandaloneServer(server, {
-    listen: { port: 4000 },
-  });
+    listen: { port: 4000 }
+  })
 
-  await migrateAll();
+  await migrateAll()
 
-  console.log(`🚀 Server ready at: ${url}`);
-};
+  console.log(`🚀 Server ready at: ${url}`)
+}
 
 startServer()
   .then()
   .catch((err: Error) =>
-    console.log(`🔥 failed to start server ${err.message}`),
-  );
+    console.log(`🔥 failed to start server ${err.message}`)
+  )

src/config.ts

@@ -2,8 +2,7 @@ import 'dotenv/config'
 import { parseEnv } from 'znv'
 import { z } from 'zod'
 
-export const { POSTGRES_URL } = parseEnv(process.env, {
-  POSTGRES_URL: z.string().min(1)
+export const { POSTGRES_URL, POSTGRES_CA_CERT_PATH } = parseEnv(process.env, {
+  POSTGRES_URL: z.string().min(1),
+  POSTGRES_CA_CERT_PATH: z.string().min(1).nullish()
 })
-
-console.log([POSTGRES_URL].join(', '))

src/migrations/20240205145527_organizations.ts

@@ -1,14 +1,14 @@
-import type { Knex } from "knex";
+import type { Knex } from 'knex'
 
-const tableName = "organizations";
+const tableName = 'organizations'
 
-export async function up(knex: Knex): Promise<void> {
+export async function up (knex: Knex): Promise<void> {
   return await knex.schema.createTable(tableName, (table) => {
-    table.text("id").primary();
-    table.text("name");
-  });
+    table.text('id').primary()
+    table.text('name')
+  })
 }
 
-export async function down(knex: Knex): Promise<void> {
-  return await knex.schema.dropTable(tableName);
+export async function down (knex: Knex): Promise<void> {
+  return await knex.schema.dropTable(tableName)
 }

src/migrations/20240205150932_regions.ts

@@ -1,49 +1,49 @@
-import type { Knex } from "knex";
+import type { Knex } from 'knex'
 
-const regionsTableName = "regions";
+const regionsTableName = 'regions'
 
-export async function up(knex: Knex): Promise<void> {
+export async function up (knex: Knex): Promise<void> {
   await knex.schema.createTable(regionsTableName, (table) => {
-    table.text("id").primary();
-    table.text("connectionString");
-  });
-  await knex.schema.createTable("organizations_regions", (table) => {
+    table.text('id').primary()
+    table.text('connectionString')
+  })
+  await knex.schema.createTable('organizations_regions', (table) => {
     table
-      .text("organizationId")
-      .references("id")
-      .inTable("organizations")
+      .text('organizationId')
+      .references('id')
+      .inTable('organizations')
       .notNullable()
-      .onDelete("cascade");
+      .onDelete('cascade')
     table
-      .text("regionId")
-      .references("id")
-      .inTable("regions")
+      .text('regionId')
+      .references('id')
+      .inTable('regions')
       .notNullable()
-      .onDelete("cascade");
-  });
-  await knex.schema.createTable("resource_organization_region", (table) => {
+      .onDelete('cascade')
+  })
+  await knex.schema.createTable('resource_organization_region', (table) => {
     table
-      .text("resourceId")
-      .references("id")
-      .inTable("resources")
+      .text('resourceId')
+      .references('id')
+      .inTable('resources')
       .notNullable()
-      .onDelete("cascade");
+      .onDelete('cascade')
     table
-      .text("organizationId")
-      .references("id")
-      .inTable("organizations")
+      .text('organizationId')
+      .references('id')
+      .inTable('organizations')
       .notNullable()
-      .onDelete("cascade");
+      .onDelete('cascade')
     table
-      .text("regionId")
-      .references("id")
-      .inTable("regions")
+      .text('regionId')
+      .references('id')
+      .inTable('regions')
       .notNullable()
-      .onDelete("cascade");
-  });
+      .onDelete('cascade')
+  })
 }
 
-export async function down(knex: Knex): Promise<void> {
-  await knex.schema.dropTable(regionsTableName);
-  await knex.schema.dropTable("organizations_regions");
+export async function down (knex: Knex): Promise<void> {
+  await knex.schema.dropTable(regionsTableName)
+  await knex.schema.dropTable('organizations_regions')
 }

src/migrations/20240206210140_region_name.ts

@@ -1,15 +1,15 @@
-import type { Knex } from "knex";
+import type { Knex } from 'knex'
 
-const regionsTableName = "regions";
+const regionsTableName = 'regions'
 
-export async function up(knex: Knex): Promise<void> {
+export async function up (knex: Knex): Promise<void> {
   await knex.schema.alterTable(regionsTableName, (table) => {
-    table.text("name").notNullable().defaultTo("region");
-  });
+    table.text('name').notNullable().defaultTo('region')
+  })
 }
 
-export async function down(knex: Knex): Promise<void> {
+export async function down (knex: Knex): Promise<void> {
   await knex.schema.alterTable(regionsTableName, (table) => {
-    table.dropColumn("name");
-  });
+    table.dropColumn('name')
+  })
 }

src/migrations/20240206213031_region_maintenance_db.ts

@@ -1,15 +1,15 @@
-import type { Knex } from "knex";
+import type { Knex } from 'knex'
 
-const regionsTableName = "regions";
+const regionsTableName = 'regions'
 
-export async function up(knex: Knex): Promise<void> {
+export async function up (knex: Knex): Promise<void> {
   await knex.schema.alterTable(regionsTableName, (table) => {
-    table.text("maintenanceDb").notNullable().defaultTo("region");
-  });
+    table.text('maintenanceDb').notNullable().defaultTo('region')
+  })
 }
 
-export async function down(knex: Knex): Promise<void> {
+export async function down (knex: Knex): Promise<void> {
   await knex.schema.alterTable(regionsTableName, (table) => {
-    table.dropColumn("maintenanceDb");
-  });
+    table.dropColumn('maintenanceDb')
+  })
 }

src/migrations/20240207203256_organization_acl.ts

@@ -1,22 +1,22 @@
-import type { Knex } from "knex";
+import type { Knex } from 'knex'
 
-const tableName = "organization_acl";
+const tableName = 'organization_acl'
 
-export async function up(knex: Knex): Promise<void> {
+export async function up (knex: Knex): Promise<void> {
   return await knex.schema.createTable(tableName, (table) => {
     table
-      .string("userId")
-      .references("id")
-      .inTable("users")
-      .onDelete("cascade");
+      .string('userId')
+      .references('id')
+      .inTable('users')
+      .onDelete('cascade')
     table
-      .string("organizationId")
-      .references("id")
-      .inTable("organizations")
-      .onDelete("cascade");
-  });
+      .string('organizationId')
+      .references('id')
+      .inTable('organizations')
+      .onDelete('cascade')
+  })
 }
 
-export async function down(knex: Knex): Promise<void> {
-  return await knex.schema.dropTable(tableName);
+export async function down (knex: Knex): Promise<void> {
+  return await knex.schema.dropTable(tableName)
 }

src/migrations/20240207212311_organization_resource_acl.ts

@@ -1,22 +1,22 @@
-import type { Knex } from "knex";
+import type { Knex } from 'knex'
 
-const tableName = "organization_resource_acl";
+const tableName = 'organization_resource_acl'
 
-export async function up(knex: Knex): Promise<void> {
+export async function up (knex: Knex): Promise<void> {
   return await knex.schema.createTable(tableName, (table) => {
     table
-      .string("resourceId")
-      .references("id")
-      .inTable("resources")
-      .onDelete("cascade");
+      .string('resourceId')
+      .references('id')
+      .inTable('resources')
+      .onDelete('cascade')
     table
-      .string("organizationId")
-      .references("id")
-      .inTable("organizations")
-      .onDelete("cascade");
-  });
+      .string('organizationId')
+      .references('id')
+      .inTable('organizations')
+      .onDelete('cascade')
+  })
 }
 
-export async function down(knex: Knex): Promise<void> {
-  return await knex.schema.dropTable(tableName);
+export async function down (knex: Knex): Promise<void> {
+  return await knex.schema.dropTable(tableName)
 }

src/migrations/20240207214054_resource_region.ts

@@ -1,28 +1,28 @@
-import type { Knex } from "knex";
+import type { Knex } from 'knex'
 
-const tableName = "resource_region_organization";
+const tableName = 'resource_region_organization'
 
-export async function up(knex: Knex): Promise<void> {
+export async function up (knex: Knex): Promise<void> {
   return await knex.schema.createTable(tableName, (table) => {
     table
-      .string("resourceId")
-      .references("id")
-      .inTable("resources")
-      .onDelete("cascade")
-      .primary();
+      .string('resourceId')
+      .references('id')
+      .inTable('resources')
+      .onDelete('cascade')
+      .primary()
     table
-      .string("regionId")
-      .references("id")
-      .inTable("regions")
-      .onDelete("cascade");
+      .string('regionId')
+      .references('id')
+      .inTable('regions')
+      .onDelete('cascade')
     table
-      .string("organizationId")
-      .references("id")
-      .inTable("organizations")
-      .onDelete("cascade");
-  });
+      .string('organizationId')
+      .references('id')
+      .inTable('organizations')
+      .onDelete('cascade')
+  })
 }
 
-export async function down(knex: Knex): Promise<void> {
-  return await knex.schema.dropTable(tableName);
+export async function down (knex: Knex): Promise<void> {
+  return await knex.schema.dropTable(tableName)
 }

src/migrations/20240213175105_region_pruning.ts

@@ -0,0 +1,15 @@
+import type { Knex } from 'knex'
+
+const regionsTableName = 'regions'
+
+export async function up (knex: Knex): Promise<void> {
+  await knex.schema.alterTable(regionsTableName, (table) => {
+    table.dropColumn('maintenanceDb')
+  })
+}
+
+export async function down (knex: Knex): Promise<void> {
+  await knex.schema.alterTable(regionsTableName, (table) => {
+    table.text('maintenanceDb').notNullable().defaultTo('region')
+  })
+}

src/migrations/20240213214212_split_resource_region_org.ts

@@ -0,0 +1,56 @@
+import type { Knex } from 'knex'
+
+const tableName = 'resource_region_organization'
+
+export async function up (knex: Knex): Promise<void> {
+  await knex.schema.dropTable(tableName)
+  await knex.schema.createTable('resource_region', (table) => {
+    table
+      .string('resourceId')
+      .references('id')
+      .inTable('resources')
+      .onDelete('cascade')
+      .primary()
+    table
+      .string('regionId')
+      .references('id')
+      .inTable('regions')
+      .onDelete('cascade')
+  })
+  await knex.schema.createTable('resource_organization', (table) => {
+    table
+      .string('resourceId')
+      .references('id')
+      .inTable('resources')
+      .onDelete('cascade')
+      .primary()
+    table
+      .string('organizationId')
+      .references('id')
+      .inTable('organizations')
+      .onDelete('cascade')
+  })
+}
+
+export async function down (knex: Knex): Promise<void> {
+  await knex.schema.dropTable('resource_organization')
+  await knex.schema.dropTable('resource_region')
+  await knex.schema.createTable(tableName, (table) => {
+    table
+      .string('resourceId')
+      .references('id')
+      .inTable('resources')
+      .onDelete('cascade')
+      .primary()
+    table
+      .string('regionId')
+      .references('id')
+      .inTable('regions')
+      .onDelete('cascade')
+    table
+      .string('organizationId')
+      .references('id')
+      .inTable('organizations')
+      .onDelete('cascade')
+  })
+}

src/migrations/20240214151340_region_ssl_cert.ts

@@ -0,0 +1,13 @@
+import type { Knex } from 'knex'
+
+export async function up (knex: Knex): Promise<void> {
+  await knex.schema.alterTable('regions', (table) => {
+    table.text('sslCaCert').nullable()
+  })
+}
+
+export async function down (knex: Knex): Promise<void> {
+  await knex.schema.alterTable('regions', (table) => {
+    table.dropColumn('sslCaCert')
+  })
+}

src/migrations/20240220152312_region_name_unique.ts

@@ -0,0 +1,13 @@
+import type { Knex } from 'knex'
+
+export async function up (knex: Knex): Promise<void> {
+  await knex.schema.alterTable('regions', (table) => {
+    table.unique('name')
+  })
+}
+
+export async function down (knex: Knex): Promise<void> {
+  await knex.schema.alterTable('regions', (table) => {
+    table.dropUnique(['name'])
+  })
+}

src/repositories.ts

@@ -1,5 +1,4 @@
-import { Knex } from "knex";
-import { knex } from "./db";
+import { Knex } from 'knex'
 import {
   UserRecord,
   Resource,
@@ -10,191 +9,239 @@ import {
   Organization,
   OrganizationAcl,
   OrganizationResourceAcl,
-  ResourceRegion,
-  ResourceRegionOrg,
-} from "./types";
+  ResourceRegion
+} from './types'
 
-const Users = () => knex<UserRecord>("users");
-const Resources = () => knex<Resource>("resources");
-const ResourceAclRepo = () => knex<ResourceAcl>("resource_acl");
+export class RegionRepo {
+  db: Knex
 
-export const queryUser = async (userId: string): Promise<UserRecord | null> => {
-  return (await Users().where("id", "=", userId).first()) ?? null;
-};
-
-export const getUsersFrom = (db: Knex) => async (): Promise<UserRecord[]> => {
-  return await db<UserRecord>("users").select();
-};
-
-export const saveUserTo =
-  (db: Knex) =>
-  async (user: UserRecord): Promise<void> => {
-    await db<UserRecord>("users").insert(user);
-  };
-
-export const saveResourceTo =
-  (db: Knex) =>
-  async (resource: Resource): Promise<void> => {
-    await db<Resource>("resources").insert(resource);
-  };
-
-export const queryResourceFrom =
-  (db: Knex) =>
-  async (resourceId: string): Promise<Resource | null> => {
-    return (
-      (await db<Resource>("resources").where({ id: resourceId }).first()) ??
-      null
-    );
-  };
-
-export const queryResourceAclFrom =
-  (db: Knex) =>
-  async ({ resourceId, userId }: ResourceAcl): Promise<ResourceAcl | null> => {
-    return (
-      (await db<ResourceAcl>("resource_acl")
-        .where({ userId, resourceId })
-        .first()) ?? null
-    );
-  };
-
-export const saveResourceAclTo =
-  (db: Knex) =>
-  async (resourceAcl: ResourceAcl): Promise<void> => {
-    await db<ResourceAcl>("resource_acl").insert(resourceAcl);
-  };
-
-export const countResources = async (userId: string): Promise<number> => {
-  const [rawCount] = await ResourceAclRepo().count().where({ userId });
-  return parseInt(rawCount.count as string);
-};
-
-export const queryResources = async ({
-  userId,
-  limit,
-  cursor,
-}: {
-  userId: string;
-  limit: number;
-  cursor: string | null;
-}) => {
-  const query = Resources()
-    .join("resource_acl", "resources.id", "resource_acl.resourceId")
-    .where({ userId });
-  if (cursor) {
-    query.andWhere("createdAt", "<", cursor);
+  constructor (db: Knex) {
+    this.db = db
   }
-  return await query.limit(limit);
-};
 
-export const countCommentsIn =
-  (db: Knex) =>
-  async (resourceId: string): Promise<number> => {
-    const [rawCount] = await db<Comment>("comments")
+  async saveResource (resource: Resource): Promise<void> {
+    await this.db<Resource>('resources').insert(resource)
+  }
+
+  async findResource (resourceId: string): Promise<Resource | null> {
+    return (
+      (await this.db<Resource>('resources')
+        .where({ id: resourceId })
+        .first()) ?? null
+    )
+  }
+
+  async saveComment (comment: Comment): Promise<void> {
+    await this.db<Comment>('comments').insert(comment)
+  }
+
+  async countComments (resourceId: string): Promise<number> {
+    const [rawCount] = await this.db<Comment>('comments')
       .count()
-      .where({ resourceId });
-    return parseInt(rawCount.count as string);
-  };
+      .where({ resourceId })
+    return parseInt(rawCount.count as string)
+  }
 
-export const queryCommentsFrom =
-  (db: Knex) =>
-  async ({
+  async queryComments ({
     resourceId,
     limit,
-    cursor,
+    cursor
   }: {
-    resourceId: string;
-    limit: number;
-    cursor: string | null;
-  }): Promise<Comment[]> => {
-    const query = db<Comment>("comments").where({ resourceId });
+    resourceId: string
+    limit: number
+    cursor: string | null
+  }): Promise<Comment[]> {
+    const query = this.db<Comment>('comments').where({ resourceId })
     if (cursor) {
-      query.andWhere("createdAt", "<", cursor);
+      query.andWhere('createdAt', '<', cursor)
     }
-    return await query.limit(limit);
-  };
+    return await query.limit(limit)
+  }
+}
 
-export const saveCommentTo =
-  (db: Knex) =>
-  async (comment: Comment): Promise<void> => {
-    await db<Comment>("comments").insert(comment);
-  };
-
-export const getRegionsFrom = (db: Knex) => async (): Promise<Array<Region>> =>
-  await db<Region>("regions").select();
-
-export const getRegionFrom =
-  (db: Knex) =>
-  async (id: string): Promise<Region | null> =>
-    (await db<Region>("regions").where({ id }).first()) ?? null;
-
-export const getOrganizationRegionsFrom =
-  (db: Knex) => async (): Promise<Array<OrganizationsRegions>> =>
-    await db<OrganizationsRegions>("organizations_regions").select();
-
-export const queryOrganizationRegionsFrom =
-  (db: Knex) =>
-  async ({
-    regionId,
-    organizationId,
-  }: OrganizationsRegions): Promise<OrganizationsRegions | null> =>
-    (await db<OrganizationsRegions>("organizations_regions")
-      .where({ regionId, organizationId })
-      .first()) ?? null;
-
-export const saveRegionTo = (db: Knex) => async (region: Region) =>
-  await db<Region>("regions").insert(region);
-
-export const saveOrganizationTo =
-  (db: Knex) => async (organization: Organization) =>
-    await db<Organization>("organizations").insert(organization);
-
-export const getOrganizationFrom =
-  (db: Knex) =>
-  async (id: string): Promise<Organization | null> => {
+export class MainRepo extends RegionRepo {
+  async findUser (userId: string): Promise<UserRecord | null> {
     return (
-      (await db<Organization>("organizations").where({ id }).first()) ?? null
-    );
-  };
+      (await this.db<UserRecord>('users').where('id', '=', userId).first()) ??
+      null
+    )
+  }
 
-export const getOrganizationsFrom =
-  (db: Knex) => async (): Promise<Organization[]> =>
-    await db<Organization>("organizations").select();
+  async queryUsers (): Promise<UserRecord[]> {
+    return await this.db<UserRecord>('users').select()
+  }
 
-export const saveOrganizationsRegionsTo =
-  (db: Knex) =>
-  async (or: OrganizationsRegions): Promise<void> =>
-    await db<OrganizationsRegions>("organizations_regions").insert(or);
+  async saveUser (user: UserRecord): Promise<void> {
+    await this.db<UserRecord>('users').insert(user)
+  }
 
-export const saveOrganizationAclTo =
-  (db: Knex) =>
-  async (orgAcl: OrganizationAcl): Promise<void> => {
-    await db<OrganizationsRegions>("organization_acl").insert(orgAcl);
-  };
+  async getUsersResourceAcl ({
+    resourceId,
+    userId
+  }: ResourceAcl): Promise<ResourceAcl | null> {
+    return (
+      (await this.db<ResourceAcl>('resource_acl')
+        .where({ userId, resourceId })
+        .first()) ?? null
+    )
+  }
 
-export const queryOrganizationAclFrom =
-  (db: Knex) =>
-  async ({
+  async saveResourceAcl (resourceAcl: ResourceAcl): Promise<void> {
+    await this.db<ResourceAcl>('resource_acl').insert(resourceAcl)
+  }
+
+  async countUsersResources (userId: string): Promise<number> {
+    const [rawCount] = await this.db<ResourceAcl>('resource_acl')
+      .count()
+      .where({ userId })
+    return parseInt(rawCount.count as string)
+  }
+
+  async findUsersResource ({
+    resourceId,
+    userId
+  }: ResourceAcl): Promise<ResourceAcl | null> {
+    return (
+      (await this.db<ResourceAcl>('resource_acl')
+        .where({ userId, resourceId })
+        .first()) ?? null
+    )
+  }
+
+  async queryResources ({
     userId,
-    organizationId,
-  }: OrganizationAcl): Promise<OrganizationAcl | null> =>
-    (await db<OrganizationAcl>("organization_acl")
-      .where({ userId, organizationId })
-      .first()) ?? null;
+    limit,
+    cursor
+  }: {
+    userId: string
+    limit: number
+    cursor: string | null
+  }): Promise<Resource[]> {
+    let query = this.db<Resource & ResourceAcl>('resources')
+      .join('resource_acl', 'resources.id', 'resource_acl.resourceId')
+      .where({ userId })
+    if (cursor) {
+      query = query.andWhere('createdAt', '<', cursor)
+    }
+    const items = await query.orderBy('createdAt', 'desc').limit(limit)
+    return items
+  }
 
-export const saveOrganizationResourceAclTo =
-  (db: Knex) =>
-  async (item: OrganizationResourceAcl): Promise<void> => {
-    await db<OrganizationResourceAcl>("organization_resource_acl").insert(item);
-  };
-
-export const saveResourceRegionOrganizationTo =
-  (db: Knex) => async (item: ResourceRegionOrg) => {
-    await db<ResourceRegionOrg>("resource_region_organization").insert(item);
-  };
-
-export const queryResourceRegionOrganizationFrom =
-  (db: Knex) =>
-  async (resourceId: string): Promise<ResourceRegion | null> =>
-    (await db<ResourceRegionOrg>("resource_region_organization")
+  async countResourceComments (resourceId: string): Promise<number> {
+    const [rawCount] = await this.db<Comment>('comments')
+      .count()
       .where({ resourceId })
-      .first()) ?? null;
+    return parseInt(rawCount.count as string)
+  }
+
+  async queryComments ({
+    resourceId,
+    limit,
+    cursor
+  }: {
+    resourceId: string
+    limit: number
+    cursor: string | null
+  }): Promise<Comment[]> {
+    let query = this.db<Comment>('comments').where({ resourceId })
+    if (cursor) {
+      query = query.andWhere('createdAt', '<', cursor)
+    }
+    return await query.orderBy('createdAt', 'desc').limit(limit)
+  }
+
+  async queryRegions (
+    params:
+    | {
+      connectionString?: string | undefined
+    }
+    | undefined = undefined
+  ): Promise<Region[]> {
+    const query = this.db<Region>('regions')
+    if ((params != null) && params.connectionString) query.where(params)
+    return await query.select()
+  }
+
+  async findRegion (id: string): Promise<Region | null> {
+    return (await this.db<Region>('regions').where({ id }).first()) ?? null
+  }
+
+  async queryOrganizationsRegions (): Promise<OrganizationsRegions[]> {
+    return await this.db<OrganizationsRegions>('organizations_regions').select()
+  }
+
+  async findOrganizationRegion ({
+    regionId,
+    organizationId
+  }: OrganizationsRegions): Promise<OrganizationsRegions | null> {
+    return (
+      (await this.db<OrganizationsRegions>('organizations_regions')
+        .where({ regionId, organizationId })
+        .first()) ?? null
+    )
+  }
+
+  async saveRegion (region: Region): Promise<void> {
+    await this.db<Region>('regions').insert(region)
+  }
+
+  async saveOrganization (organization: Organization) {
+    await this.db<Organization>('organizations').insert(organization)
+  }
+
+  async findOrganization (id: string): Promise<Organization | null> {
+    return (
+      (await this.db<Organization>('organizations').where({ id }).first()) ??
+      null
+    )
+  }
+
+  async queryOrganizations (): Promise<Organization[]> {
+    return await this.db<Organization>('organizations').select()
+  }
+
+  async saveOrganizationRegion (or: OrganizationsRegions): Promise<void> {
+    return await this.db<OrganizationsRegions>('organizations_regions').insert(
+      or
+    )
+  }
+
+  async saveOrganizationAcl (orgAcl: OrganizationAcl): Promise<void> {
+    await this.db<OrganizationsRegions>('organization_acl').insert(orgAcl)
+  }
+
+  async findOrganizationAcl ({
+    userId,
+    organizationId
+  }: OrganizationAcl): Promise<OrganizationAcl | null> {
+    return (
+      (await this.db<OrganizationAcl>('organization_acl')
+        .where({ userId, organizationId })
+        .first()) ?? null
+    )
+  }
+
+  async saveOrganizationResourceAcl (
+    item: OrganizationResourceAcl
+  ): Promise<void> {
+    await this.db<OrganizationResourceAcl>('organization_resource_acl').insert(
+      item
+    )
+  }
+
+  async findResourceRegion ({
+    resourceId
+  }: {
+    resourceId: string
+  }): Promise<ResourceRegion | null> {
+    return (
+      (await this.db<ResourceRegion>('resource_region')
+        .where({ resourceId })
+        .first()) ?? null
+    )
+  }
+
+  async saveResourceRegion (item: ResourceRegion): Promise<void> {
+    await this.db<ResourceRegion>('resource_region').insert(item)
+  }
+}

src/resolvers.ts

@@ -1,25 +1,7 @@
-import {
-  getOrganizationsFrom,
-  getRegionsFrom,
-  queryOrganizationAclFrom,
-  queryOrganizationRegionsFrom,
-  queryResourceAclFrom,
-  saveOrganizationResourceAclTo,
-  saveOrganizationAclTo,
-  saveResourceAclTo,
-  saveResourceTo,
-  saveResourceRegionOrganizationTo,
-  saveCommentTo,
-  queryResourceFrom,
-  queryUser,
-  countCommentsIn,
-  queryCommentsFrom,
-  getUsersFrom,
-  saveUserTo,
-} from "./repositories";
-import { getComments } from "./services/comments";
-import { createResource, getResources } from "./services/resources";
-import { GraphQLError } from "graphql";
+import { RegionRepo, MainRepo } from './repositories'
+import { getComments } from './services/comments'
+import { createResource, getResources } from './services/resources'
+import { GraphQLError } from 'graphql'
 import {
   Resource,
   UserRecord,
@@ -29,168 +11,167 @@ import {
   OrganizationsRegions,
   OrganizationAcl,
   CommentCreateArgs,
-  UserCreateArgs,
-} from "./types";
+  UserCreateArgs
+} from './types'
 import {
-  bindRegionToOrganization,
   createOrganization,
-  getDbClient,
-  getMainDbClient,
-  getResourceDatabaseConnection,
   registerRegion,
-} from "./services/databaseManagement";
-import { authorizeUserOrgRegion } from "./services/authz";
-import cryptoRandomString from "crypto-random-string";
+  getMainRepo,
+  getRegionRepo,
+  getResourceRepo
+} from './services/databaseManagement'
+import { authorizeUserOrgRegion } from './services/authz'
+import cryptoRandomString from 'crypto-random-string'
 
 // Resolvers define how to fetch the types defined in your schema.
 // This resolver retrieves books from the "books" array above.
 export const resolvers = {
   Query: {
-    async users() {
-      return await getUsersFrom(getMainDbClient())();
+    async users () {
+      return await getMainRepo().queryUsers()
     },
-    async user(_: unknown, args: { id: string }) {
-      return await queryUser(args.id);
+    async user (_: unknown, args: { id: string }) {
+      return await getMainRepo().findUser(args.id)
     },
-    async resource(
+    async resource (
       _: unknown,
-      args: { id: string; userId: string },
+      args: { id: string, userId: string }
     ): Promise<Resource> {
-      const mainDb = getMainDbClient();
-      const maybeAcl = await queryResourceAclFrom(mainDb)({
+      const mainRepo = getMainRepo()
+      const maybeAcl = await mainRepo.getUsersResourceAcl({
         userId: args.userId,
-        resourceId: args.id,
-      });
+        resourceId: args.id
+      })
       if (maybeAcl == null) {
         throw new GraphQLError(
           "The user doesn't have access to the given resource",
           {
             extensions: {
-              code: "FORBIDDEN",
-            },
-          },
-        );
+              code: 'FORBIDDEN'
+            }
+          }
+        )
       }
-      const db = await getResourceDatabaseConnection(args.id);
-      const maybeResource = await queryResourceFrom(db)(args.id);
+      const resourceRepo = await getResourceRepo(args.id)
+      const maybeResource = await resourceRepo.findResource(args.id)
       if (maybeResource == null) {
-        throw new GraphQLError("Resource not found", {
-          extensions: { code: "RESOURCE_NOT_FOUND" },
-        });
+        throw new GraphQLError('Resource not found', {
+          extensions: { code: 'RESOURCE_NOT_FOUND' }
+        })
       }
-      return maybeResource;
+      return maybeResource
     },
-    async organizations() {
-      return await getOrganizationsFrom(getMainDbClient())();
-    },
-    async regions() {
-      return await getRegionsFrom(getMainDbClient())();
+    async organizations () {
+      return await getMainRepo().queryOrganizations()
     },
+    async regions () {
+      return await getMainRepo().queryRegions()
+    }
   },
   User: {
-    async resources(parent: UserRecord, args: PaginationArgs) {
-      return await getResources({ userId: parent.id, ...args });
-    },
+    async resources (parent: UserRecord, args: PaginationArgs) {
+      const mainRepo = getMainRepo()
+      return await getResources(
+        mainRepo.countUsersResources.bind(mainRepo),
+        mainRepo.queryResources.bind(mainRepo)
+      )({ userId: parent.id, ...args })
+    }
   },
   Resource: {
-    async comments(
+    async comments (
       parent: Resource,
-      { limit, cursor }: PaginationArgs,
+      { limit, cursor }: PaginationArgs
     ): Promise<CommentCollection> {
-      const db = await getResourceDatabaseConnection(parent.id);
+      const resourceRepo = await getResourceRepo(parent.id)
       return await getComments(
-        countCommentsIn(db),
-        queryCommentsFrom(db),
+        resourceRepo.countComments.bind(resourceRepo),
+        resourceRepo.queryComments.bind(resourceRepo)
       )({
         resourceId: parent.id,
         limit,
-        cursor,
-      });
-    },
+        cursor
+      })
+    }
   },
   Mutation: {
-    async createUser(
+    async createUser (
       _: unknown,
-      { input: { name } }: { input: UserCreateArgs },
+      { input: { name } }: { input: UserCreateArgs }
     ) {
-      const id = cryptoRandomString({ length: 10 });
-      await saveUserTo(getMainDbClient())({ id, name });
-      return id;
+      const id = cryptoRandomString({ length: 10 })
+      await getMainRepo().saveUser({ id, name })
+      return id
     },
-    async registerRegion(
+    async registerRegion (
       _: unknown,
       args: {
-        name: string;
-        connectionString: string;
-        maintenanceDb: string;
-      },
+        name: string
+        connectionString: string
+        sslCaCert: string | null
+      }
     ) {
-      return await registerRegion(args);
+      return await registerRegion(args)
     },
-    async createOrganization(_: unknown, args: { name: string }) {
-      return await createOrganization(args.name);
+    async createOrganization (_: unknown, args: { name: string }) {
+      return await createOrganization(args.name)
     },
-    async addRegionToOrganization(_: unknown, args: OrganizationsRegions) {
-      await bindRegionToOrganization(args);
+    async addRegionToOrganization (_: unknown, args: OrganizationsRegions) {
+      await getMainRepo().saveOrganizationRegion(args)
     },
-    async addUserToOrganization(
+    async addUserToOrganization (
       _: unknown,
-      { input: args }: { input: OrganizationAcl },
+      { input: args }: { input: OrganizationAcl }
     ) {
-      await saveOrganizationAclTo(getMainDbClient())(args);
+      await getMainRepo().saveOrganizationAcl(args)
     },
-    async createResource(
+    async createResource (
       _: unknown,
-      { input: args }: { input: ResourceCreateArgs },
+      { input: args }: { input: ResourceCreateArgs }
     ) {
-      const mainDb = getMainDbClient();
+      const mainRepo = getMainRepo()
       await authorizeUserOrgRegion(
-        queryOrganizationAclFrom(mainDb),
-        queryOrganizationRegionsFrom(mainDb),
-      )(args);
+        mainRepo.findOrganizationAcl.bind(mainRepo),
+        mainRepo.findOrganizationRegion.bind(mainRepo)
+      )(args)
 
-      const db =
-        args.regionId && args.organizationId
-          ? await getDbClient({
-              regionId: args.regionId,
-              organizationId: args.organizationId,
-            })
-          : mainDb;
+      const repo = args.regionId
+        ? await getRegionRepo({ regionId: args.regionId })
+        : mainRepo
 
       const resourceId = await createResource(
-        saveResourceTo(db),
-        saveResourceAclTo(mainDb),
-      )(args);
+        repo.saveResource.bind(repo),
+        mainRepo.saveResourceAcl.bind(mainRepo)
+      )(args)
 
       if (args.organizationId) {
-        await saveOrganizationResourceAclTo(mainDb)({
+        await mainRepo.saveOrganizationResourceAcl({
           organizationId: args.organizationId,
-          resourceId,
-        });
-        await saveResourceRegionOrganizationTo(mainDb)({
-          resourceId,
-          organizationId: args.organizationId,
-          // i know its not null here, the authz function ensures it
-          regionId: args.regionId!,
-        });
+          resourceId
+        })
+        if (args.regionId) {
+          await mainRepo.saveResourceRegion({
+            resourceId,
+            // i know its not null here, the authz function ensures it
+            regionId: args.regionId
+          })
+        }
       }
-      return resourceId;
+      return resourceId
     },
-    async addComment(
+    async addComment (
       _: unknown,
-      { input: args }: { input: CommentCreateArgs },
+      { input: args }: { input: CommentCreateArgs }
     ) {
-      const mainDb = getMainDbClient();
-      const resourceAcl = await queryResourceAclFrom(mainDb)(args);
-      if (!resourceAcl)
-        throw new Error("The user doesn't have access to the given resource");
-      //2. get resource db client
-      const db = await getResourceDatabaseConnection(args.resourceId);
-      //3. save comment to db
-      const id = cryptoRandomString({ length: 10 });
-      const createdAt = new Date();
-      await saveCommentTo(db)({ id, createdAt, ...args });
-      return id;
-    },
-  },
-};
+      const mainRepo = getMainRepo()
+      const resourceAcl = await mainRepo.getUsersResourceAcl(args)
+      if (resourceAcl == null) { throw new Error("The user doesn't have access to the given resource") }
+      // 2. get resource db client
+      const resourceRepo = await getResourceRepo(args.resourceId)
+      // 3. save comment to db
+      const id = cryptoRandomString({ length: 10 })
+      const createdAt = new Date()
+      await resourceRepo.saveComment({ id, createdAt, ...args })
+      return id
+    }
+  }
+}

src/schema.graphql

@@ -38,7 +38,6 @@ type Organization {
 type Region {
   id: String!
   name: String!
-  maintenanceDb: String!
 }
 
 type Query {
@@ -78,7 +77,7 @@ type Mutation {
   registerRegion(
     name: String!
     connectionString: String!
-    maintenanceDb: String!
+    sslCaCert: String
   ): String!
   createOrganization(name: String!): String!
   addRegionToOrganization(organizationId: String!, regionId: String!): Boolean

src/services/authz.ts

@@ -1,26 +1,23 @@
 import {
   OrganizationAcl,
   OrganizationsRegions,
-  UserOrgRegionArgs,
-} from "../types";
+  UserOrgRegionArgs
+} from '../types'
 
 export const authorizeUserOrgRegion =
   (
     orgAclGetter: (params: OrganizationAcl) => Promise<OrganizationAcl | null>,
     orgRegionGetter: (
       params: OrganizationsRegions,
-    ) => Promise<OrganizationsRegions | null>,
+    ) => Promise<OrganizationsRegions | null>
   ) =>
-  async ({ userId, regionId, organizationId }: UserOrgRegionArgs) => {
-    if (!organizationId && regionId)
-      throw new Error("public org doesn't support regions");
-    if (organizationId) {
-      if (!regionId) throw new Error("organizations can only write to regions");
-      const orgAcl = await orgAclGetter({ organizationId, userId });
-      if (!orgAcl)
-        throw new Error("user doesn't have access to this organization");
-      const orgRegion = await orgRegionGetter({ organizationId, regionId });
-      if (!orgRegion)
-        throw new Error("organization doesnt have access to this region");
+    async ({ userId, regionId, organizationId }: UserOrgRegionArgs) => {
+      if (!organizationId && regionId) { throw new Error("public org doesn't support regions") }
+      if (organizationId) {
+        if (!regionId) throw new Error('organizations can only write to regions')
+        const orgAcl = await orgAclGetter({ organizationId, userId })
+        if (orgAcl == null) { throw new Error("user doesn't have access to this organization") }
+        const orgRegion = await orgRegionGetter({ organizationId, regionId })
+        if (orgRegion == null) { throw new Error('organization doesnt have access to this region') }
+      }
     }
-  };

src/services/comments.ts

@@ -1,25 +1,25 @@
-import { CommentCollection, PaginationArgs, Comment } from "../types";
+import { CommentCollection, PaginationArgs, Comment } from '../types'
 
 interface GetCommentsArgs extends PaginationArgs {
-  resourceId: string;
+  resourceId: string
 }
 
 export const getComments =
   (
     countComments: (resourceId: string) => Promise<number>,
-    queryComments: (params: GetCommentsArgs) => Promise<Comment[]>,
+    queryComments: (params: GetCommentsArgs) => Promise<Comment[]>
   ) =>
-  async (params: GetCommentsArgs): Promise<CommentCollection> => {
+    async (params: GetCommentsArgs): Promise<CommentCollection> => {
     // yes, i should be doing base64 de and encoding with the cursor...
-    const totalCount = await countComments(params.resourceId);
-    const items = await queryComments(params);
-    let cursor = null;
-    if (items.length > 0) {
-      cursor = items.slice(-1)[0].createdAt.toISOString();
+      const totalCount = await countComments(params.resourceId)
+      const items = await queryComments(params)
+      let cursor = null
+      if (items.length > 0) {
+        cursor = items.slice(-1)[0].createdAt.toISOString()
+      }
+      return {
+        totalCount,
+        items,
+        cursor
+      }
     }
-    return {
-      totalCount,
-      items,
-      cursor,
-    };
-  };

src/services/databaseManagement.ts

@@ -1,243 +1,212 @@
-import { POSTGRES_URL } from "../config";
-import {
-  getOrganizationFrom,
-  getOrganizationRegionsFrom,
-  getRegionFrom,
-  queryResourceRegionOrganizationFrom,
-  saveOrganizationTo,
-  saveOrganizationsRegionsTo,
-  saveRegionTo,
-} from "../repositories";
-import { OrganizationsRegions, Region } from "../types";
-import knex, { Knex } from "knex";
-import cryptoRandomString from "crypto-random-string";
+import { POSTGRES_URL } from '../config'
+import { RegionRepo, MainRepo } from '../repositories'
+import knex, { Knex } from 'knex'
+import cryptoRandomString from 'crypto-random-string'
 
-const migrateToLatest = async (client: Knex): Promise<void> => {
+const migrateToLatest = async (db: Knex): Promise<void> => {
   const plannedMigrations: Array<{ file: string }> = (
-    await client.migrate.list()
-  )[1];
+    await db.migrate.list()
+  )[1]
   if (plannedMigrations.length > 0) {
     console.log(
       `🕰️  planning migrations: ${plannedMigrations
         .map((m) => m.file)
-        .join(",")}`,
-    );
+        .join(',')}`
+    )
   } else {
-    console.log("no migrations are planned");
+    console.log('no migrations are planned')
   }
   // TODO: make sure if a migration fails, all migrations are rolled back
-  await client.migrate.latest();
-};
-
-export const migrateAll = async (): Promise<void> => {
-  await migrateToLatest(mainClient);
-  const databaseSchemas = await getAllDatabaseSchemaConnections();
-
-  await Promise.all([
-    ...databaseSchemas.map(async (sc) => await migrateToLatest(sc)),
-  ]);
-  // 1. get all regions from main DB
-  // 2. construct region specific knex clients and cache them by
-  // 3. structure the cache so that it accomodates client creation by resource id
-  // 4. get all organization regions from main DB
-  // 5. for in all regions for all organizations, run the migration
-  // 6. do not forget the migration for the main DB
-  //
-};
-
-const createDatabaseConfig = (connectionString: string): Knex.Config => {
-  return {
-    client: "pg",
-    connection: {
-      connectionString,
-    },
-    // connection: connectionString,
-    migrations: {
-      directory: "src/migrations",
-      extension: "ts",
-    },
-  };
-};
-
-const mainClient = knex(createDatabaseConfig(POSTGRES_URL));
-
-const _connectionStore: Map<string, Knex> = new Map();
-
-interface RegionWithMaybeOrganization {
-  regionId: string;
-  organizationId?: string | undefined;
+  await db.migrate.latest()
 }
 
-const _createConnectionKey = ({
-  organizationId,
-  regionId,
-}: RegionWithMaybeOrganization): string => {
-  return organizationId ? `${organizationId}@${regionId}` : regionId;
-};
+export const migrateAll = async (): Promise<void> => {
+  await migrateToLatest(mainRepo.db)
+  const repos = await getAllRepositories()
 
-export const getDbClient = async ({
-  regionId,
-  organizationId,
-}: RegionWithMaybeOrganization): Promise<Knex> => {
-  const connectionKey = _createConnectionKey({ organizationId, regionId });
-  const maybeClient = _connectionStore.get(connectionKey);
-  if (maybeClient) return maybeClient;
-  const maybeRegion = await mainClient<Region>("regions")
-    .select()
-    .where({ id: regionId })
-    .first();
-  if (!maybeRegion) throw Error(`region ${regionId} not found`);
-  const connectionString = organizationId
-    ? `${maybeRegion.connectionString}/${organizationId}`
-    : `${maybeRegion.connectionString}/${maybeRegion.maintenanceDb}`;
-  const client = knex(createDatabaseConfig(connectionString));
-  _connectionStore.set(connectionKey, client);
-  return client;
-};
+  await Promise.all([
+    ...repos.map(async (repo) => await migrateToLatest(repo.db))
+  ])
+}
 
-export const getMainDbClient = (): Knex => mainClient;
+const createDatabaseConfig = (
+  connectionString: string,
+  sslCaCert: string | null
+): Knex.Config => {
+  const config: Knex.Config = {
+    client: 'pg',
+    connection: {
+      connectionString,
+      ssl: sslCaCert
+        ? {
+            ca: sslCaCert,
+            rejectUnauthorized: true
+          }
+        : undefined
+    },
+    migrations: {
+      directory: 'src/migrations',
+      extension: 'ts'
+    }
+  }
+  return config
+}
+
+const mainRepo = new MainRepo(knex(createDatabaseConfig(POSTGRES_URL, null)))
+
+const _repoStore: Map<string, RegionRepo> = new Map()
+export const getRegionRepo = async ({
+  regionId
+}: {
+  regionId: string | undefined
+}): Promise<RegionRepo> => {
+  if (!regionId) return mainRepo
+  const maybeRepo = _repoStore.get(regionId)
+  if (maybeRepo != null) return maybeRepo
+  const maybeRegion = await mainRepo.findRegion(regionId)
+  if (maybeRegion == null) throw Error(`region ${regionId} not found`)
+  const repo = new RegionRepo(
+    knex(
+      createDatabaseConfig(maybeRegion.connectionString, maybeRegion.sslCaCert)
+    )
+  )
+  _repoStore.set(regionId, repo)
+  return repo
+}
+
+export const getMainRepo = (): MainRepo => mainRepo
 
 export const registerRegion = async ({
   name,
   connectionString,
-  maintenanceDb,
+  sslCaCert
 }: {
-  name: string;
-  connectionString: string;
-  maintenanceDb: string;
+  name: string
+  connectionString: string
+  sslCaCert: string | null
 }): Promise<string> => {
-  // TODO: validate the connectionString, so that the knex client can connect to it
-  const id = cryptoRandomString({ length: 10 });
-  await saveRegionTo(mainClient)({
+  const regions = await mainRepo.queryRegions({ connectionString })
+  if (regions.length > 0) throw new Error('This region is already registered')
+  const id = cryptoRandomString({ length: 10 })
+  const repo = new RegionRepo(
+    knex(createDatabaseConfig(connectionString, sslCaCert))
+  )
+  await migrateToLatest(repo.db)
+  _repoStore.set(id, repo)
+
+  const sslmode = sslCaCert ? 'require' : 'disable'
+  await setUpUserReplication({
+    from: mainRepo.db,
+    to: repo.db,
+    regionName: name,
+    sslmode
+  })
+  await setUpResourceReplication({
+    from: repo.db,
+    to: mainRepo.db,
+    regionName: name,
+    sslmode
+  })
+
+  await mainRepo.saveRegion({
     id,
     name,
     connectionString,
-    maintenanceDb,
-  });
-  return id;
-};
+    sslCaCert
+  })
+  return id
+}
 
 export const createOrganization = async (name: string): Promise<string> => {
-  const id = cryptoRandomString({ length: 10 });
-  await saveOrganizationTo(mainClient)({ id, name });
-  return id;
-};
+  const id = cryptoRandomString({ length: 10 })
+  await mainRepo.saveOrganization({ id, name })
+  return id
+}
 
-const createDb = async (client: Knex, name: string): Promise<void> => {
-  try {
-    await client.raw(`create database "${name}"`);
-  } catch (err) {
-    if (!(err instanceof Error)) throw err;
-    if (!err.message.includes("already exists")) throw err;
-  }
-};
+interface ReplicationArgs {
+  from: Knex
+  to: Knex
+  sslmode: string
+  regionName: string
+}
 
 const setUpUserReplication = async ({
   from,
   to,
-}: {
-  from: Knex;
-  to: Knex;
-}): Promise<void> => {
+  sslmode,
+  regionName
+}: ReplicationArgs): Promise<void> => {
   // TODO: ensure its created...
-  const connectionString: string =
-    from.client.config.connection.connectionString;
   try {
-    await from.raw("CREATE PUBLICATION userspub FOR TABLE users;");
+    await from.raw('CREATE PUBLICATION userspub FOR TABLE users;')
   } catch (err) {
-    if (!(err instanceof Error)) throw err;
-    if (!err.message.includes("already exists")) throw err;
+    if (!(err instanceof Error)) throw err
+    if (!err.message.includes('already exists')) throw err
   }
+
+  const fromUrl = new URL(from.client.config.connection.connectionString)
+  const fromDbName = fromUrl.pathname.replace('/', '')
+  const subName = `userssub_${regionName}`
+  const rawSqeel = `SELECT * FROM aiven_extras.pg_create_subscription(
+    '${subName}',
+    'dbname=${fromDbName} host=${fromUrl.hostname} port=${fromUrl.port} sslmode=${sslmode} user=${fromUrl.username} password=${fromUrl.password}',
+    'userspub', 
+    '${subName}',
+    TRUE,
+    TRUE
+  );`
   try {
-    const toUrl = new URL(to.client.config.connection.connectionString);
-    await to.raw(
-      `CREATE SUBSCRIPTION userssub_${toUrl.pathname.replace("/", "")} CONNECTION '${connectionString}' PUBLICATION userspub;`,
-    );
+    await to.raw(rawSqeel)
   } catch (err) {
-    if (!(err instanceof Error)) throw err;
-    if (!err.message.includes("already exists")) throw err;
+    if (!(err instanceof Error)) throw err
+    if (!err.message.includes('already exists')) throw err
   }
-};
+}
 
 const setUpResourceReplication = async ({
   from,
-  fromRegionName,
   to,
-}: {
-  from: Knex;
-  fromRegionName: string;
-  to: Knex;
-}): Promise<void> => {
+  regionName,
+  sslmode
+}: ReplicationArgs): Promise<void> => {
   // TODO: ensure its created...
-  const connectionString: string =
-    from.client.config.connection.connectionString;
-  const connUrl = new URL(connectionString);
   try {
-    await from.raw("CREATE PUBLICATION resourcepub FOR TABLE resources;");
+    await from.raw('CREATE PUBLICATION resourcepub FOR TABLE resources;')
   } catch (err) {
-    if (!(err instanceof Error)) throw err;
-    if (!err.message.includes("already exists")) throw err;
+    if (!(err instanceof Error)) throw err
+    if (!err.message.includes('already exists')) throw err
   }
+
+  const fromUrl = new URL(from.client.config.connection.connectionString)
+  const fromDbName = fromUrl.pathname.replace('/', '')
+  const subName = `resourcesub_${regionName}`
+  const rawSqeel = `SELECT * FROM aiven_extras.pg_create_subscription(
+    '${subName}',
+    'dbname=${fromDbName} host=${fromUrl.hostname} port=${fromUrl.port} sslmode=${sslmode} user=${fromUrl.username} password=${fromUrl.password}',
+    'resourcepub', 
+    '${subName}',
+    TRUE,
+    TRUE
+  );`
   try {
-    await to.raw(
-      `CREATE SUBSCRIPTION "resroucesub_${fromRegionName.replace(
-        " ",
-        "",
-      )}_${connUrl.pathname.replace(
-        "/",
-        "",
-      )}" CONNECTION '${connectionString}' PUBLICATION resourcepub;`,
-    );
+    await to.raw(rawSqeel)
   } catch (err) {
-    if (!(err instanceof Error)) throw err;
-    if (!err.message.includes("already exists")) throw err;
+    if (!(err instanceof Error)) throw err
+    if (!err.message.includes('already exists')) throw err
   }
-};
+}
 
-export const bindRegionToOrganization = async ({
-  regionId,
-  organizationId,
-}: OrganizationsRegions): Promise<void> => {
-  const region = await getRegionFrom(mainClient)(regionId);
-  if (!region) throw Error(`region ${regionId} not found`);
-  const organization = await getOrganizationFrom(mainClient)(organizationId);
-  if (!organization) throw Error(`organization ${organizationId} not found`);
+export const getAllRepositories = async (): Promise<RegionRepo[]> => {
+  const regions = await mainRepo.queryRegions({})
+  const regionRepos = await Promise.all(
+    regions.map(async (region) => await getRegionRepo({ regionId: region.id }))
+  )
+  return [mainRepo, ...regionRepos]
+}
 
-  const regionClient = await getDbClient({ regionId });
-
-  await createDb(regionClient, organizationId);
-
-  const client = await getDbClient({ organizationId, regionId });
-  const connectionKey = _createConnectionKey({ organizationId, regionId });
-
-  await migrateToLatest(client);
-
-  await setUpUserReplication({ from: mainClient, to: client });
-  await setUpResourceReplication({
-    from: client,
-    fromRegionName: region.name,
-    to: mainClient,
-  });
-
-  _connectionStore.set(connectionKey, client);
-  await saveOrganizationsRegionsTo(mainClient)({ organizationId, regionId });
-};
-
-export const getAllDatabaseSchemaConnections = async (): Promise<Knex[]> => {
-  const organizationRegions = await getOrganizationRegionsFrom(mainClient)();
-  const clients = await Promise.all(
-    organizationRegions.map(async (or) => {
-      const client = await getDbClient(or);
-      return client;
-    }),
-  );
-  return [mainClient, ...clients];
-};
-
-export const getResourceDatabaseConnection = async (
-  resourceId: string,
-): Promise<Knex> => {
-  const resourceRegionOrg =
-    await queryResourceRegionOrganizationFrom(mainClient)(resourceId);
-  return resourceRegionOrg ? await getDbClient(resourceRegionOrg) : mainClient;
-};
+export const getResourceRepo = async (
+  resourceId: string
+): Promise<RegionRepo> => {
+  const resourceRegion = await mainRepo.findResourceRegion({ resourceId })
+  return (resourceRegion != null) ? await getRegionRepo(resourceRegion) : getMainRepo()
+}

src/services/resources.ts

@@ -1,45 +1,47 @@
-import cryptoRandomString from "crypto-random-string";
-import { countResources, queryResources } from "../repositories";
+import cryptoRandomString from 'crypto-random-string'
 import {
   Resource,
   PaginationArgs,
   ResourceCollection,
   ResourceCreateArgs,
-  ResourceAcl,
-} from "../types";
+  ResourceAcl
+} from '../types'
 
 interface GetResourcesArgs extends PaginationArgs {
-  userId: string;
+  userId: string
 }
-export const getResources = async (
-  params: GetResourcesArgs,
-): Promise<ResourceCollection> => {
-  const totalCount = await countResources(params.userId);
-  const items = await queryResources(params);
-  let cursor = null;
-  if (items.length > 0) {
-    cursor = items.slice(-1)[0].createdAt.toISOString();
-  }
-  return {
-    totalCount,
-    items,
-    cursor,
-  };
-};
+export const getResources =
+  (
+    countResources: (userId: string) => Promise<number>,
+    queryResources: (params: GetResourcesArgs) => Promise<Resource[]>
+  ) =>
+    async (params: GetResourcesArgs): Promise<ResourceCollection> => {
+      const totalCount = await countResources(params.userId)
+      const items = await queryResources(params)
+      let cursor = null
+      if (items.length > 0) {
+        cursor = items.slice(-1)[0].createdAt.toISOString()
+      }
+      return {
+        totalCount,
+        items,
+        cursor
+      }
+    }
 
 export const createResource =
   (
     resourceSaver: (resource: Resource) => Promise<void>,
-    resourceAclSaver: (resourceAcl: ResourceAcl) => Promise<void>,
+    resourceAclSaver: (resourceAcl: ResourceAcl) => Promise<void>
   ) =>
-  async ({ userId, name }: ResourceCreateArgs): Promise<string> => {
-    //1. if no org, create project in main region, validate that, regionId is null
-    //2. if org, validate if user has access to the org
-    //3. if org and region, validate if org has access to region
-    //4. create resource
-    const id = cryptoRandomString({ length: 10 });
-    const resource = { id, name, createdAt: new Date() };
-    await resourceSaver(resource);
-    await resourceAclSaver({ resourceId: id, userId });
-    return id;
-  };
+    async ({ userId, name }: ResourceCreateArgs): Promise<string> => {
+    // 1. if no org, create project in main region, validate that, regionId is null
+    // 2. if org, validate if user has access to the org
+    // 3. if org and region, validate if org has access to region
+    // 4. create resource
+      const id = cryptoRandomString({ length: 10 })
+      const resource = { id, name, createdAt: new Date() }
+      await resourceSaver(resource)
+      await resourceAclSaver({ resourceId: id, userId })
+      return id
+    }

src/sketch/dir1/take1.ts

@@ -0,0 +1,33 @@
+import { knex } from "../../db";
+import { Knex } from "knex";
+
+type Thing = {
+  id: string;
+  name: string;
+};
+
+// talk to the DB
+const repo =
+  (db: Knex<Thing>) =>
+  async (id: string): Promise<Thing | null> => {
+    return (await db.where({ id }).first()) ?? null;
+  };
+
+// business / domain logic
+const service =
+  (thingGetter: (id: string) => Promise<Thing | null>) =>
+  async (id: string): Promise<Thing | null> => {
+    return thingGetter(id);
+  };
+
+const getThingClient = (id: string | undefined): Knex => {
+  if (!id) return knex;
+  return knex;
+};
+
+// graphql entry
+export const resolver = async (args: { id: string }): Promise<Thing> => {
+  const thing = await service(repo(getThingClient(args.id)))(args.id);
+  if (!thing) throw new Error("not found");
+  return thing;
+};

src/sketch/dir2/take2.ts

@@ -0,0 +1,33 @@
+import { Knex } from "knex";
+import { knex } from "../../db";
+type Thing = {
+  id: string;
+  name: string;
+};
+
+type ServedThing = {
+  foo: number;
+} & Thing;
+
+// talk to the DB
+const repo =
+  ({ db }: { db: Knex<Thing> }) =>
+  async (id: string): Promise<Thing | null> => {
+    return (await db().where({ id }).first()) ?? null;
+  };
+
+// business / domain logic
+const service =
+  ({ thingGetter }: { thingGetter: (id: string) => Promise<Thing | null> }) =>
+  async (id: string): Promise<ServedThing | null> => {
+    const thing = await thingGetter(id);
+    const foo = 123;
+    return thing ? { ...thing, foo } : null;
+  };
+
+// graphql entry
+export const resolver = async (id: string): Promise<ServedThing> => {
+  const thing = await service({ thingGetter: repo({ db: knex }) })(id);
+  if (!thing) throw new Error("not found");
+  return thing;
+};

src/sketch/dir3/domain.ts

@@ -0,0 +1,8 @@
+export type Thing = {
+  id: string;
+};
+
+export type ThingRepo = {
+  findThing: (id: string) => Promise<Thing | null>;
+  queryThing: () => Promise<Thing[]>;
+};

src/sketch/dir3/repo.ts

@@ -0,0 +1,12 @@
+import { Knex } from "knex";
+import { Thing } from "./domain";
+
+const findThing =
+  ({ db }: { db: Knex }) =>
+  async (id: string): Promise<Thing | null> => {
+    return null;
+  };
+
+export const thingRepo = ({ db }: { db: Knex }) => ({
+  findThing: findThing({ db }),
+});

src/sketch/dir3/resolver.ts

@@ -0,0 +1,11 @@
+import { ServedThing, service } from "./services/service";
+import { thingRepo } from "./repo";
+import { knex } from "../../db";
+
+export const resolver = async (id: string): Promise<ServedThing> => {
+  const thing = await service({
+    thingRepo: thingRepo({ db: knex }),
+  })(id);
+  if (!thing) throw new Error("not found");
+  return thing;
+};

src/sketch/dir3/services/service.ts

@@ -0,0 +1,19 @@
+import { Thing, type ThingRepo } from "../domain";
+
+export type ServedThing = {
+  foo: number;
+} & Thing;
+
+export const service =
+  ({ thingRepo }: { thingRepo: Pick<ThingRepo, "findThing"> }) =>
+  async (id: string): Promise<ServedThing | null> => {
+    const thing = await thingRepo.findThing(id);
+    const foo = 123;
+    return thing ? { ...thing, foo } : null;
+  };
+
+export const service2 = ({
+  thingRepo,
+}: {
+  thingRepo: Pick<ThingRepo, "queryThing">;
+}) => {};

src/sketch/existing/existing.ts

@@ -0,0 +1,24 @@
+import { knex } from "../../db";
+type Thing = {
+  id: string;
+  name: string;
+};
+
+const Things = () => knex<Thing>("things");
+
+// talk to the DB
+const repo = async (id: string): Promise<Thing | null> => {
+  return (await Things().where({ id }).first()) ?? null;
+};
+
+// business / domain logic
+const service = async (id: string): Promise<Thing | null> => {
+  return repo(id);
+};
+
+// graphql entry
+export const resolver = async (id: string): Promise<Thing> => {
+  const thing = await service(id);
+  if (!thing) throw new Error("not found");
+  return thing;
+};

src/types.ts

@@ -1,98 +1,99 @@
 export interface CommentCreateArgs {
-  userId: string;
-  content: string;
-  resourceId: string;
+  userId: string
+  content: string
+  resourceId: string
 }
 
 export interface Comment extends CommentCreateArgs {
-  id: string;
-  createdAt: Date;
+  id: string
+  createdAt: Date
 }
 
 export interface PaginationArgs {
-  limit: number;
-  cursor: string | null;
+  limit: number
+  cursor: string | null
 }
 
 interface Collection<T> {
-  totalCount: number;
-  cursor: string | null;
-  items: T[];
+  totalCount: number
+  cursor: string | null
+  items: T[]
 }
 
 export interface CommentCollection extends Collection<Comment> {}
 
 export interface UserOrgRegionArgs {
-  userId: string;
-  organizationId: string | null;
-  regionId: string | null;
+  userId: string
+  organizationId: string | null
+  regionId: string | null
 }
 
 export interface ResourceCreateArgs extends UserOrgRegionArgs {
-  name: string;
+  name: string
 }
 
 export interface Resource {
-  id: string;
-  name: string;
-  createdAt: Date;
+  id: string
+  name: string
+  createdAt: Date
 }
 
 export interface ResourceCollection extends Collection<Resource> {}
 
 export interface UserCreateArgs {
-  name: string;
+  name: string
 }
 
 export interface UserRecord extends UserCreateArgs {
-  id: string;
+  id: string
 }
 
 export interface User extends UserRecord {
   resources: {
-    cursor: string | null;
-    totalCount: number;
-    items: Resource[];
-  };
+    cursor: string | null
+    totalCount: number
+    items: Resource[]
+  }
 }
 
 export interface ResourceAcl {
-  userId: string;
-  resourceId: string;
+  userId: string
+  resourceId: string
 }
 
 export interface Region {
-  id: string;
-  name: string;
-  connectionString: string;
-  maintenanceDb: string;
+  id: string
+  name: string
+  connectionString: string
+  sslCaCert: string | null
 }
 
 export interface Organization {
-  id: string;
-  name: string;
+  id: string
+  name: string
 }
 
 export interface OrganizationAcl {
-  userId: string;
-  organizationId: string;
+  userId: string
+  organizationId: string
 }
 
 export interface OrganizationsRegions {
-  organizationId: string;
-  regionId: string;
+  organizationId: string
+  regionId: string
 }
 
 export interface OrganizationResourceAcl {
-  organizationId: string;
-  resourceId: string;
+  organizationId: string
+  resourceId: string
 }
 
 export interface ResourceRegion {
-  resourceId: string;
-  regionId: string;
+  resourceId: string
+  regionId: string
 }
 
-export interface ResourceRegionOrg extends ResourceRegion {
-  organizationId: string;
+export interface ResourceOrganization {
+  resourceId: string
+  organizationId: string
 }

tests/integration/regions.spec.ts

@@ -1,29 +1,29 @@
-import { expect, beforeAll, describe, it } from "vitest";
+import { expect, beforeAll, describe, it } from 'vitest'
 import {
   getOrganizationRegionsFrom,
-  getRegionsFrom,
-} from "../../src/repositories";
+  getRegionsFrom
+} from '../../src/repositories'
 import {
   getMainDbClient,
-  migrateAll,
-} from "../../src/services/databaseManagement";
-import { Knex } from "knex";
+  migrateAll
+} from '../../src/services/databaseManagement'
+import { Knex } from 'knex'
 
-describe("regions", () => {
-  let dbClient: Knex;
+describe('regions', () => {
+  let dbClient: Knex
 
   beforeAll(async () => {
-    dbClient = await getMainDbClient();
-  });
-  it("gets all regions", async () => {
-    const regions = await getRegionsFrom(dbClient)();
-    expect(regions.length).toBeGreaterThan(0);
-  });
-  it("gets organizations regions", async () => {
-    const orgRegions = await getOrganizationRegionsFrom(dbClient)();
-    expect(orgRegions.length).toBeGreaterThan(0);
-  });
-  it("migrates all", async () => {
-    await migrateAll();
-  });
-});
+    dbClient = await getMainDbClient()
+  })
+  it('gets all regions', async () => {
+    const regions = await getRegionsFrom(dbClient)()
+    expect(regions.length).toBeGreaterThan(0)
+  })
+  it('gets organizations regions', async () => {
+    const orgRegions = await getOrganizationRegionsFrom(dbClient)()
+    expect(orgRegions.length).toBeGreaterThan(0)
+  })
+  it('migrates all', async () => {
+    await migrateAll()
+  })
+})

vitest.config.ts

@@ -1,7 +1,7 @@
-import { defineConfig } from "vitest/config";
+import { defineConfig } from 'vitest/config'
 
 export default defineConfig({
   test: {
-    dir: "tests",
-  },
-});
+    dir: 'tests'
+  }
+})