[PoC]: Database level Data Integration

postgres/Dockerfile

@@ -0,0 +1,3 @@
+FROM postgres:16-bullseye
+
+RUN apt-get update && apt-get -y install postgresql-16-http

postgres/lib/PostgresService.js

@@ -622,6 +622,7 @@ GROUP BY k
         DROP USER IF EXISTS "${creds.user}";
         CREATE GROUP "${creds.usergroup}";
         CREATE USER "${creds.user}" WITH CREATEROLE IN GROUP "${creds.usergroup}" PASSWORD '${creds.user}';
+        ALTER USER "${creds.user}" SUPERUSER;
         GRANT "${creds.usergroup}" TO "${creds.user}" WITH ADMIN OPTION;
       `)
       await this.exec(`CREATE DATABASE "${creds.database}" OWNER="${creds.user}" TEMPLATE=template0`)

postgres/pg-stack.yml

@@ -3,13 +3,13 @@ version: '3.1'
 
 services:
   db:
-    image: postgres:16-alpine
+    build: .
     restart: always
     environment:
       POSTGRES_PASSWORD: postgres
     ports:
       - '5432:5432'
-    command: ['postgres', '-c', 'log_statement=all']
+    command: postgres
   ### use at will at dev time - save mem on ci time
   # adminer:
   #   image: adminer

postgres/test/integration.cds

@@ -0,0 +1,6 @@
+using {CatalogService} from '../../test/bookshop/apis/CatalogService';
+
+service integration {
+  entity Genres as projection on CatalogService.Genres;
+  entity Books  as projection on CatalogService.ListOfBooks;
+}

postgres/test/integration.test.js

@@ -0,0 +1,86 @@
+const cds = require('../../test/cds.js')
+
+describe('Data Integration', () => {
+  before(() => {
+    cds.env.features.ieee754compatible = true
+  })
+
+  const _deploy = cds.deploy
+  cds.deploy = async function () {
+    const sys = await cds.connect.to('sys', {
+      ...cds.requires.db,
+      credentials: {
+        ...cds.db.options.credentials,
+        user: `${cds.db.options.credentials.database}_USER_MANAGER`,
+        password: `${cds.db.options.credentials.database}_USER_MANAGER`,
+      },
+    })
+    await sys.run(`CREATE EXTENSION IF NOT EXISTS http SCHEMA public`)
+
+    const db = await cds.connect.to('db')
+
+    const convertInput = cds.db.class.CQN2SQL._convertInput ?? new cds.db.class.CQN2SQL().class._convertInput
+
+    for (const name in cds.model.definitions) {
+      const entity = cds.model.definitions[name]
+      if (!entity.__REMOTE__) continue
+
+      const columns = []
+      function add(element, name = element.name) {
+        if (element.isAssociation) {
+          for (const key of element.keys || []) {
+            if (key.ref.length > 1) { cds.error`Association with deep foreign key currently not supported!!!` }
+            if (!entity.elements[`${name}_${key.ref[0]}`]) add(element._target.elements[key.ref[0]], `${name}_${key.ref[0]}`)
+          }
+        } else {
+          const converter = element[convertInput] || (a => a)
+          columns.push(`${converter(`value->>'${name}'`)} as ${name}`)
+        }
+      }
+
+      for (const name in entity.elements) add(entity.elements[name])
+
+      const from = `jsonb_array_elements((SELECT (content::jsonb->>'value')::jsonb FROM public.http_get('http://host.docker.internal:4004/browse/${name.split('.').at(-1)}?src=postgres') LIMIT 1))`
+
+      await db.run(`CREATE VIEW ${entity} AS SELECT ${columns} FROM ${from}`)
+    }
+
+    return _deploy.apply(this, arguments)
+  }
+
+  cds.on('loaded', (csn) => {
+    const remotes = []
+    for (const name in csn.definitions) {
+      const service = csn.definitions[name]
+      if (service.kind !== 'service') continue
+      if (service['@data.product']) remotes.push(name)
+    }
+
+    for (const name in csn.definitions) {
+      const entity = csn.definitions[name]
+      if (entity.kind !== 'entity') continue
+      const service = remotes.find(srv => name.startsWith(srv))
+      if (!service) continue
+      entity.__REMOTE__ = true
+      entity['@cds.persistence.exists'] = true
+    }
+  })
+
+  const { expect, GET } = cds.test(__dirname, 'integration.cds')
+
+  test('debug', async () => {
+    const [db, org] = await Promise.all([
+      GET`/odata/v4/integration/Books`,
+      GET`http://localhost:4004/browse/ListOfBooks?src=test`,
+    ])
+    expect(db.data.value).deep.eq(org.data.value)
+  })
+
+  test('expand', async () => {
+    const [db, org] = await Promise.all([
+      GET`/odata/v4/integration/Books?$expand=genre`,
+      GET`http://localhost:4004/browse/ListOfBooks?$expand=genre&src=test`,
+    ])
+    expect(db.data.value).deep.eq(org.data.value)
+  })
+})

sqlite/lib/SQLiteService.js

@@ -1,3 +1,5 @@
+const child_process = require('node:child_process')
+
 const { SQLService } = require('@cap-js/db-service')
 const cds = require('@sap/cds')
 const sqlite = require('better-sqlite3')
@@ -41,6 +43,13 @@
         dbc.function('hour', deterministic, d => d === null ? null : toDate(d, true).getUTCHours())
         dbc.function('minute', deterministic, d => d === null ? null : toDate(d, true).getUTCMinutes())
         dbc.function('second', deterministic, d => d === null ? null : toDate(d, true).getUTCSeconds())
+
+        let http_get = url => child_process.execSync(`node -e "fetch('${url}').then(r => r.body.pipeTo(require('node:stream').Writable.toWeb(process.stdout)))"`)
+        try {
+          child_process.execSync(`curl --help`)
+          http_get = url => child_process.execSync(`curl "${url}"`, { stdio: ['ignore', 'pipe', 'ignore'] })
+        } catch { }
+        dbc.function('http_get', deterministic, http_get)
         if (!dbc.memory) dbc.pragma('journal_mode = WAL')
         return dbc
       },

sqlite/test/integration.cds

@@ -0,0 +1,6 @@
+using {CatalogService} from '../../test/bookshop/apis/CatalogService';
+
+service integration {
+  entity Genres as projection on CatalogService.Genres;
+  entity Books  as projection on CatalogService.ListOfBooks;
+}

sqlite/test/integration.test.js

@@ -0,0 +1,76 @@
+const cds = require('../../test/cds.js')
+
+describe('Data Integration', () => {
+  before(() => {
+    cds.env.features.ieee754compatible = true
+  })
+
+  const _deploy = cds.deploy
+  cds.deploy = async function () {
+    const db = await cds.connect.to('db')
+
+    const convertInput = cds.db.class.CQN2SQL._convertInput ?? new cds.db.class.CQN2SQL().class._convertInput
+
+    for (const name in cds.model.definitions) {
+      const entity = cds.model.definitions[name]
+      if (!entity.__REMOTE__) continue
+
+      const columns = []
+      function add(element, name = element.name) {
+        if (element.isAssociation) {
+          for (const key of element.keys || []) {
+            if (key.ref.length > 1) { cds.error`Association with deep foreign key currently not supported!!!` }
+            if (!entity.elements[`${name}_${key.ref[0]}`]) add(element._target.elements[key.ref[0]], `${name}_${key.ref[0]}`)
+          }
+        } else {
+          const converter = element[convertInput] || (a => a)
+          columns.push(`${converter(`value->>'$.${JSON.stringify(name)}'`)} as ${name}`)
+        }
+      }
+
+      for (const name in entity.elements) add(entity.elements[name])
+
+      const from = `json_each(http_get('http://localhost:4004/browse/${name.split('.').at(-1)}?src=sqlite')->>'$.value')`
+
+      await db.run(`CREATE VIEW ${entity} AS SELECT ${columns} FROM ${from}`)
+    }
+
+    return _deploy.apply(this, arguments)
+  }
+
+  cds.on('loaded', (csn) => {
+    const remotes = []
+    for (const name in csn.definitions) {
+      const service = csn.definitions[name]
+      if (service.kind !== 'service') continue
+      if (service['@data.product']) remotes.push(name)
+    }
+
+    for (const name in csn.definitions) {
+      const entity = csn.definitions[name]
+      if (entity.kind !== 'entity') continue
+      const service = remotes.find(srv => name.startsWith(srv))
+      if (!service) continue
+      entity.__REMOTE__ = true
+      entity['@cds.persistence.exists'] = true
+    }
+  })
+
+  const { expect, GET } = cds.test(__dirname, 'integration.cds')
+
+  test('simple', async () => {
+    const [db, org] = await Promise.all([
+      GET`/odata/v4/integration/Books`,
+      GET`http://localhost:4004/browse/ListOfBooks?src=test`,
+    ])
+    expect(db.data.value).deep.eq(org.data.value)
+  })
+
+  test('expand', async () => {
+    const [db, org] = await Promise.all([
+      GET`/odata/v4/integration/Books?$expand=genre`,
+      GET`http://localhost:4004/browse/ListOfBooks?$expand=genre&src=test`,
+    ])
+    expect(db.data.value).deep.eq(org.data.value)
+  })
+})

test/bookshop/apis/CatalogService/cds-plugin.js

@@ -0,0 +1 @@
+// just a tag file for plug & play

test/bookshop/apis/CatalogService/index.cds

@@ -0,0 +1,3 @@
+// This file acts as a central facade to exported service definitions.
+// You can modify it to tweak things, without your changes being overridden.
+using from './services';

test/bookshop/apis/CatalogService/package.json

@@ -0,0 +1,9 @@
+{
+  "name": "@capire/bookshop-CatalogService",
+  "version": "1.0.0",
+  "cds": {
+    "requires": {
+      "CatalogService": true
+    }
+  }
+}