poc: implement new FrameworkRepository

api/lib/infrastructure/repositories/framework-repository.js

@@ -1,28 +1,33 @@
-import { DomainTransaction } from '../../../src/shared/domain/DomainTransaction.js';
 import { NotFoundError } from '../../../src/shared/domain/errors.js';
-import { Framework } from '../../../src/shared/domain/models/Framework.js';
+import { Framework } from '../../../src/shared/domain/models/index.js';
+import { LearningContentRepository } from '../../../src/shared/infrastructure/repositories/learning-content-repository.js';
 
 const TABLE_NAME = 'learningcontent.frameworks';
 
 export async function list() {
-  const knex = DomainTransaction.getConnection();
-  const frameworkDtos = await knex.select('*').from(TABLE_NAME).orderBy('name');
-  return frameworkDtos.map(toDomain);
+  const cacheKey = 'list';
+  const listCallback = (knex) => knex.orderBy('name');
+  return getInstance().find(cacheKey, listCallback);
 }
 
 export async function getByName(name) {
-  const knex = DomainTransaction.getConnection();
-  const frameworkDto = await knex.select('*').from(TABLE_NAME).where('name', name).first();
-  if (!frameworkDto) {
+  const cacheKey = `getByName(${name})`;
+  const findByNameCallback = (knex) => knex.where('name', name);
+  const [framework] = await getInstance().find(cacheKey, findByNameCallback);
+  if (!framework) {
     throw new NotFoundError(`Framework not found for name ${name}`);
   }
-  return toDomain(frameworkDto);
+  return framework;
 }
 
-export async function findByRecordIds(frameworkIds) {
-  const knex = DomainTransaction.getConnection();
-  const frameworkDtos = await knex.select('*').from(TABLE_NAME).whereIn('id', frameworkIds).orderBy('name');
-  return frameworkDtos.map(toDomain);
+export async function findByRecordIds(ids) {
+  const cacheKey = `findByRecordIds(${ids.sort()})`;
+  const findByIdsCallback = (knex) => knex.whereIn('id', ids).orderBy('name');
+  return getInstance().find(cacheKey, findByIdsCallback);
+}
+
+export function clear() {
+  return getInstance().clear();
 }
 
 function toDomain(frameworkData) {
@@ -32,3 +37,12 @@ function toDomain(frameworkData) {
     areas: [],
   });
 }
+
+let instance;
+
+function getInstance() {
+  if (!instance) {
+    instance = new LearningContentRepository({ tableName: TABLE_NAME, toDomain });
+  }
+  return instance;
+}

api/package.json

@@ -21,6 +21,8 @@
     "@aws-sdk/lib-storage": "^3.121.0",
     "@aws-sdk/s3-request-presigner": "^3.145.0",
     "@getbrevo/brevo": "^2.1.1",
+    "@graphql-yoga/redis-event-target": "^3.0.1",
+    "@graphql-yoga/subscription": "^5.0.1",
     "@hapi/accept": "^6.0.0",
     "@hapi/boom": "^10.0.1",
     "@hapi/hapi": "^21.0.0",
@@ -35,6 +37,7 @@
     "axios": "^1.0.0",
     "bcrypt": "^5.0.1",
     "cron-parser": "^4.9.0",
+    "dataloader": "^2.2.2",
     "dayjs": "^1.11.5",
     "debug": "^4.3.4",
     "dotenv": "^16.0.1",

api/src/shared/domain/models/Framework.js

@@ -4,6 +4,14 @@ class Framework {
     this.name = name;
     this.areas = areas;
   }
+
+  clone() {
+    return new Framework({
+      id: this.id,
+      name: this.name,
+      areas: this.areas,
+    });
+  }
 }
 
 export { Framework };

api/src/shared/infrastructure/caches/learning-content-pubsub.js

@@ -0,0 +1,45 @@
+import { createRedisEventTarget } from '@graphql-yoga/redis-event-target';
+import { createPubSub } from '@graphql-yoga/subscription';
+import { Redis } from 'ioredis';
+
+import { config } from '../../config.js';
+
+/**
+ * @typedef {import('@graphql-yoga/subscription').PubSub<{
+ *   [key: string]: [message: object]
+ * }>} LearningContentPubSub
+ */
+
+/** @type {LearningContentPubSub} */
+let pubSub;
+
+/** @type {import('ioredis').Redis} */
+let publishClient;
+/** @type {import('ioredis').Redis} */
+let subscribeClient;
+
+export function getPubSub() {
+  if (pubSub) return pubSub;
+
+  if (!config.caching.redisUrl) {
+    pubSub = createPubSub();
+    return pubSub;
+  }
+
+  publishClient = new Redis(config.caching.redisUrl);
+  subscribeClient = new Redis(config.caching.redisUrl);
+
+  pubSub = createPubSub({
+    eventTarget: createRedisEventTarget({
+      publishClient,
+      subscribeClient,
+    }),
+  });
+
+  return pubSub;
+}
+
+export async function quit() {
+  await publishClient?.quit();
+  await subscribeClient?.quit();
+}

api/src/shared/infrastructure/repositories/learning-content-repository.js

@@ -0,0 +1,114 @@
+import Dataloader from 'dataloader';
+
+import { knex } from '../../../../db/knex-database-connection.js';
+import * as learningContentPubSub from '../caches/learning-content-pubsub.js';
+
+export class LearningContentRepository {
+  #tableName;
+  #toDomain;
+  #idType;
+  #dataloader;
+  #findCache;
+  #findCacheMiss;
+
+  constructor({ tableName, toDomain, idType = 'text', pubSub = learningContentPubSub.getPubSub() }) {
+    this.#tableName = tableName;
+    this.#toDomain = toDomain;
+    this.#idType = idType;
+
+    this.#dataloader = new Dataloader((ids) => this.#batchLoad(ids), {
+      cacheMap: new LearningContentCache({
+        name: `${tableName}:entities`,
+        pubSub,
+      }),
+    });
+
+    this.#findCache = new LearningContentCache({
+      name: `${tableName}:results`,
+      pubSub,
+    });
+
+    this.#findCacheMiss = new Map();
+  }
+
+  async find(cacheKey, callback) {
+    const dtos = await this.#findDtos(callback, cacheKey);
+    return dtos.map(this.#toDomain);
+  }
+
+  #findDtos(callback, cacheKey) {
+    let dtos = this.#findCache.get(cacheKey);
+    if (dtos) return dtos;
+
+    dtos = this.#findCacheMiss.get(cacheKey);
+    if (dtos) return dtos;
+
+    dtos = this.#loadDtos(callback, cacheKey).finally(() => {
+      this.#findCacheMiss.delete(cacheKey);
+    });
+    this.#findCacheMiss.set(cacheKey, dtos);
+
+    return dtos;
+  }
+
+  async #loadDtos(callback, cacheKey) {
+    const ids = await callback(knex.pluck('id').from(this.#tableName));
+    const dtos = await this.#dataloader.loadMany(ids);
+    this.#findCache.set(cacheKey, dtos);
+    return dtos;
+  }
+
+  async #batchLoad(ids) {
+    return knex
+      .select(`${this.#tableName}.*`)
+      .from(knex.raw(`unnest(?::${this.#idType}[]) with ordinality as ids(id, idx)`, [ids])) // eslint-disable-line knex/avoid-injections
+      .leftJoin(this.#tableName, `${this.#tableName}.id`, 'ids.id')
+      .orderBy('ids.idx');
+  }
+
+  clear() {
+    this.#dataloader.clearAll();
+    this.#findCache.clear();
+  }
+}
+
+class LearningContentCache {
+  #map = new Map();
+  #pubSub;
+  #name;
+
+  /**
+   * @param {{
+   *   pubSub: import('../caches/learning-content-pubsub.js').LearningContentPubSub
+   *   name: string
+   * }} config
+   * @returns
+   */
+  constructor({ pubSub, name }) {
+    this.#pubSub = pubSub;
+    this.#name = name;
+
+    (async () => {
+      for await (const message of pubSub.subscribe(name)) {
+        if (message.type === 'clear') this.#map.clear();
+        if (message.type === 'delete') this.#map.delete(this.message.key);
+      }
+    })();
+  }
+
+  get(key) {
+    return this.#map.get(key);
+  }
+
+  set(key, value) {
+    return this.#map.set(key, value);
+  }
+
+  delete(key) {
+    return this.#pubSub.publish(this.#name, { type: 'delete', key });
+  }
+
+  clear() {
+    return this.#pubSub.publish(this.#name, { type: 'clear' });
+  }
+}