diff --git a/common/config/rush/pnpm-lock.yaml b/common/config/rush/pnpm-lock.yaml index 654735b716..4d7dca213d 100644 --- a/common/config/rush/pnpm-lock.yaml +++ b/common/config/rush/pnpm-lock.yaml @@ -874,6 +874,9 @@ importers: '@rush-temp/postgres': specifier: file:./projects/postgres.tgz version: file:projects/postgres.tgz(@babel/core@7.23.9)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.23.9))(esbuild@0.24.2)(ts-node@10.9.2(@types/node@22.15.29)(typescript@5.8.3)) + '@rush-temp/postgres-base': + specifier: file:./projects/postgres-base.tgz + version: file:projects/postgres-base.tgz(@babel/core@7.23.9)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.23.9))(esbuild@0.24.2)(ts-node@10.9.2(@types/node@22.15.29)(typescript@5.8.3)) '@rush-temp/preference': specifier: file:./projects/preference.tgz version: file:projects/preference.tgz(@babel/core@7.23.9)(@jest/types@29.6.3)(@types/node@22.15.29)(babel-jest@29.7.0(@babel/core@7.23.9))(esbuild@0.24.2)(ts-node@10.9.2(@types/node@22.15.29)(typescript@5.8.3)) @@ -4008,7 +4011,7 @@ packages: version: 0.0.0 '@rush-temp/card@file:projects/card.tgz': - resolution: {integrity: sha512-j4+YvBnVqr2kQ4tx5ziMEgy+Fq5EpeBSfZtb6OtdJJgChiR69o1EYWOY55wY4olegUtd/48HMR7d14w+8VKliQ==, tarball: file:projects/card.tgz} + resolution: {integrity: sha512-WEYVW+LenbXcGUuwBLhqqEqwUUOX1Gh0f49ZxXBitExw4VYt3d0tLchcqutFg6KKkmTvWf9kvCCsT73DhSEsCQ==, tarball: file:projects/card.tgz} version: 0.0.0 '@rush-temp/chat-assets@file:projects/chat-assets.tgz': @@ -4927,6 +4930,10 @@ packages: resolution: {integrity: sha512-fwOcCP6U6p4QldSJaRplYRM68QhIyaToQJ6qzNwv6BW+psQkCv2BhamTUjqfCnsS4H14wQ5Z8LwVr3D4bn5iOw==, tarball: file:projects/pod-workspace.tgz} version: 0.0.0 + '@rush-temp/postgres-base@file:projects/postgres-base.tgz': + resolution: {integrity: sha512-kyX1UpbrAJtR8AvnYMxcsx8Dt9Yk78mCoe3CB8iqHsqfch4qfOvH1Z0kxvzjkFVFvKgRGUqQUfuN3gVeEjuF0A==, tarball: file:projects/postgres-base.tgz} + version: 0.0.0 + '@rush-temp/postgres@file:projects/postgres.tgz': resolution: {integrity: sha512-qYzUM38pKeU84/AeMFejxqKsJOOcNB+oZUJ+/rfT4tLi8Ccr4/uKIF3ZfhSwKt2d3ztHwEZj3jdWNuTzEgWyHg==, tarball: file:projects/postgres.tgz} version: 0.0.0 @@ -23224,6 +23231,32 @@ snapshots: - socks - supports-color + '@rush-temp/postgres-base@file:projects/postgres-base.tgz(@babel/core@7.23.9)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.23.9))(esbuild@0.24.2)(ts-node@10.9.2(@types/node@22.15.29)(typescript@5.8.3))': + dependencies: + '@types/jest': 29.5.12 + '@types/node': 22.15.29 + '@typescript-eslint/eslint-plugin': 6.21.0(@typescript-eslint/parser@6.21.0(eslint@8.56.0)(typescript@5.8.3))(eslint@8.56.0)(typescript@5.8.3) + '@typescript-eslint/parser': 6.21.0(eslint@8.56.0)(typescript@5.8.3) + eslint: 8.56.0 + eslint-config-standard-with-typescript: 40.0.0(@typescript-eslint/eslint-plugin@6.21.0(@typescript-eslint/parser@6.21.0(eslint@8.56.0)(typescript@5.8.3))(eslint@8.56.0)(typescript@5.8.3))(eslint-plugin-import@2.29.1(eslint@8.56.0))(eslint-plugin-n@15.7.0(eslint@8.56.0))(eslint-plugin-promise@6.1.1(eslint@8.56.0))(eslint@8.56.0)(typescript@5.8.3) + eslint-plugin-import: 2.29.1(eslint@8.56.0) + eslint-plugin-n: 15.7.0(eslint@8.56.0) + eslint-plugin-promise: 6.1.1(eslint@8.56.0) + jest: 29.7.0(@types/node@22.15.29)(ts-node@10.9.2(@types/node@22.15.29)(typescript@5.8.3)) + postgres: 3.4.7 + prettier: 3.2.5 + ts-jest: 29.1.2(@babel/core@7.23.9)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.23.9))(esbuild@0.24.2)(jest@29.7.0(@types/node@22.15.29)(ts-node@10.9.2(@types/node@22.15.29)(typescript@5.8.3)))(typescript@5.8.3) + typescript: 5.8.3 + transitivePeerDependencies: + - '@babel/core' + - '@jest/types' + - babel-jest + - babel-plugin-macros + - esbuild + - node-notifier + - supports-color + - ts-node + '@rush-temp/postgres@file:projects/postgres.tgz(@babel/core@7.23.9)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.23.9))(esbuild@0.24.2)(ts-node@10.9.2(@types/node@22.15.29)(typescript@5.8.3))': dependencies: '@types/jest': 29.5.12 diff --git a/dev/tool/src/db.ts b/dev/tool/src/db.ts index 6469bd8194..66edc4c541 100644 --- a/dev/tool/src/db.ts +++ b/dev/tool/src/db.ts @@ -37,7 +37,6 @@ import { } from '@hcengineering/postgres' import { type DBDoc } from '@hcengineering/postgres/types/utils' import { getTransactorEndpoint } from '@hcengineering/server-client' -import { sharedPipelineContextVars } from '@hcengineering/server-pipeline' import { generateToken } from '@hcengineering/server-token' import { connect } from '@hcengineering/server-tool' import { type MongoClient } from 'mongodb' @@ -57,7 +56,7 @@ export async function moveFromMongoToPG ( } const client = getMongoClient(mongoUrl) const mongo = await client.getClient() - const pg = getDBClient(sharedPipelineContextVars, dbUrl) + const pg = getDBClient(dbUrl) const pgClient = await pg.getClient() for (let index = 0; index < workspaces.length; index++) { @@ -181,7 +180,7 @@ export async function moveWorkspaceFromMongoToPG ( } const client = getMongoClient(mongoUrl) const mongo = await client.getClient() - const pg = getDBClient(sharedPipelineContextVars, dbUrl) + const pg = getDBClient(dbUrl) const pgClient = await pg.getClient() await moveWorkspace(accountDb, mongo, pgClient, ws, region, include, force) @@ -530,7 +529,7 @@ export async function migrateCreatedModifiedBy ( progressMade = false connectsCount++ - pg = getDBClient(sharedPipelineContextVars, dbUrl) + pg = getDBClient(dbUrl) pgClient = await pg.getClient() // Expect temp table with mapping to be created manually @@ -907,7 +906,7 @@ export async function migrateMergedAccounts ( throw new Error('Only CockroachDB is supported') } - const pg = getDBClient(sharedPipelineContextVars, dbUrl) + const pg = getDBClient(dbUrl) const pgClient = await pg.getClient() const token = getToolToken() @@ -1020,7 +1019,7 @@ export async function filterMergedAccountsInMembers ( throw new Error('Only CockroachDB is supported') } - const pg = getDBClient(sharedPipelineContextVars, dbUrl) + const pg = getDBClient(dbUrl) const pgClient = await pg.getClient() try { @@ -1083,7 +1082,7 @@ export async function ensureGlobalPersonsForLocalAccounts ( throw new Error('Only CockroachDB is supported') } - const pg = getDBClient(sharedPipelineContextVars, dbUrl) + const pg = getDBClient(dbUrl) const pgClient = await pg.getClient() const token = getToolToken() diff --git a/dev/tool/src/index.ts b/dev/tool/src/index.ts index 068a992ed1..774b60f63b 100644 --- a/dev/tool/src/index.ts +++ b/dev/tool/src/index.ts @@ -44,8 +44,7 @@ import { registerServerPlugins, registerStringLoaders, registerTxAdapterFactory, - setAdapterSecurity, - sharedPipelineContextVars + setAdapterSecurity } from '@hcengineering/server-pipeline' import serverToken, { generateToken } from '@hcengineering/server-token' import { createWorkspace, upgradeWorkspace } from '@hcengineering/workspace-service' @@ -134,10 +133,10 @@ const colorConstants = { // Register close on process exit. process.on('exit', () => { - shutdownPostgres(sharedPipelineContextVars).catch((err) => { + shutdownPostgres().catch((err) => { console.error(err) }) - shutdownMongo(sharedPipelineContextVars).catch((err) => { + shutdownMongo().catch((err) => { console.error(err) }) }) @@ -663,7 +662,6 @@ export function devTool ( // }, // cmd.region, // 5000, // 5 gigabytes per blob - // sharedPipelineContextVars, // async (storage, workspaceStorage) => { // if (cmd.remove) { // await updateArchiveInfo(toolCtx, db, ws.workspace, true) @@ -783,7 +781,6 @@ export function devTool ( // }, // cmd.region, // 100, - // sharedPipelineContextVars // ) // ) { // processed++ diff --git a/dev/tool/src/markup.ts b/dev/tool/src/markup.ts index 0a4a89119c..f032f4b2df 100644 --- a/dev/tool/src/markup.ts +++ b/dev/tool/src/markup.ts @@ -55,11 +55,7 @@ import { getDBClient } from '@hcengineering/postgres' import { withRetry } from '@hcengineering/retry' import { getAccountClient } from '@hcengineering/server-client' import { type PipelineFactory, type StorageAdapter, createDummyStorageAdapter } from '@hcengineering/server-core' -import { - createBackupPipeline, - createEmptyBroadcastOps, - sharedPipelineContextVars -} from '@hcengineering/server-pipeline' +import { createBackupPipeline, createEmptyBroadcastOps } from '@hcengineering/server-pipeline' import { generateToken } from '@hcengineering/server-token' import { isEmptyMarkup } from '@hcengineering/text-core' @@ -396,7 +392,7 @@ export async function restoreMarkupRefs ( usePassedCtx: true }) - const pg = getDBClient(sharedPipelineContextVars, dbUrl) + const pg = getDBClient(dbUrl) const pgClient = await pg.getClient() try { diff --git a/pods/backup/src/index.ts b/pods/backup/src/index.ts index 4b8f0cca7c..a77804956a 100644 --- a/pods/backup/src/index.ts +++ b/pods/backup/src/index.ts @@ -24,8 +24,7 @@ import { registerAdapterFactory, registerDestroyFactory, registerTxAdapterFactory, - setAdapterSecurity, - sharedPipelineContextVars + setAdapterSecurity } from '@hcengineering/server-pipeline' import { join } from 'path' @@ -47,10 +46,10 @@ const model = JSON.parse(readFileSync(process.env.MODEL_JSON ?? 'model.json').to // Register close on process exit. process.on('exit', () => { - shutdownPostgres(sharedPipelineContextVars).catch((err) => { + shutdownPostgres().catch((err) => { console.error(err) }) - shutdownMongo(sharedPipelineContextVars).catch((err) => { + shutdownMongo().catch((err) => { console.error(err) }) }) @@ -103,6 +102,5 @@ startBackup( externalStorage, disableTriggers: true }) - }, - sharedPipelineContextVars + } ) diff --git a/pods/fulltext/src/server.ts b/pods/fulltext/src/server.ts index 9049bc5753..0e355e32d2 100644 --- a/pods/fulltext/src/server.ts +++ b/pods/fulltext/src/server.ts @@ -33,8 +33,7 @@ import { registerServerPlugins, registerStringLoaders, registerTxAdapterFactory, - setAdapterSecurity, - sharedPipelineContextVars + setAdapterSecurity } from '@hcengineering/server-pipeline' import serverToken, { decodeToken } from '@hcengineering/server-token' import cors from '@koa/cors' @@ -71,10 +70,10 @@ interface Reindex { } // Register close on process exit. process.on('exit', () => { - shutdownPostgres(sharedPipelineContextVars).catch((err) => { + shutdownPostgres().catch((err) => { console.error(err) }) - shutdownMongo(sharedPipelineContextVars).catch((err) => { + shutdownMongo().catch((err) => { console.error(err) }) }) diff --git a/pods/server/src/server.ts b/pods/server/src/server.ts index 8b752cd324..a1290e4e79 100644 --- a/pods/server/src/server.ts +++ b/pods/server/src/server.ts @@ -34,8 +34,7 @@ import { registerServerPlugins, registerStringLoaders, registerTxAdapterFactory, - setAdapterSecurity, - sharedPipelineContextVars + setAdapterSecurity } from '@hcengineering/server-pipeline' import { @@ -59,10 +58,10 @@ registerStringLoaders() // Register close on process exit. process.on('exit', () => { - shutdownPostgres(sharedPipelineContextVars).catch((err) => { + shutdownPostgres().catch((err) => { console.error(err) }) - shutdownMongo(sharedPipelineContextVars).catch((err) => { + shutdownMongo().catch((err) => { console.error(err) }) }) diff --git a/rush.json b/rush.json index 63b19fea66..3470459e60 100644 --- a/rush.json +++ b/rush.json @@ -912,6 +912,11 @@ "projectFolder": "server/postgres", "shouldPublish": false }, + { + "packageName": "@hcengineering/postgres-base", + "projectFolder": "server/postgres-base", + "shouldPublish": false + }, { "packageName": "@hcengineering/elastic", "projectFolder": "server/elastic", diff --git a/server/account/src/utils.ts b/server/account/src/utils.ts index 128659feaa..92b6d79ddc 100644 --- a/server/account/src/utils.ts +++ b/server/account/src/utils.ts @@ -40,7 +40,6 @@ import { pbkdf2Sync, randomBytes } from 'crypto' import otpGenerator from 'otp-generator' import { Analytics } from '@hcengineering/analytics' -import { sharedPipelineContextVars } from '@hcengineering/server-pipeline' import { decodeTokenVerbose, generateToken, TokenError } from '@hcengineering/server-token' import { MongoAccountDB } from './collections/mongo' import { PostgresAccountDB } from './collections/postgres/postgres' @@ -93,7 +92,7 @@ export async function getAccountDB ( application_name: appName } }) - const client = getDBClient(sharedPipelineContextVars, uri) + const client = getDBClient(uri) const pgClient = await client.getClient() const pgAccount = new PostgresAccountDB(pgClient, dbNs ?? 'global_account') diff --git a/server/backup-service/src/index.ts b/server/backup-service/src/index.ts index e3237492dc..b4c12d69e4 100644 --- a/server/backup-service/src/index.ts +++ b/server/backup-service/src/index.ts @@ -36,8 +36,7 @@ export function startBackup ( workspace: WorkspaceIds, branding: Branding | null, externalStorage: StorageAdapter - ) => DbConfiguration, - contextVars: Record + ) => DbConfiguration ): void { const config = _config() setMetadata(serverToken.metadata.Secret, config.Secret) @@ -66,8 +65,7 @@ export function startBackup ( (ctx, workspace, branding, externalStorage) => { return getConfig(ctx, mainDbUrl, workspace, branding, externalStorage) }, - config.Region, - contextVars + config.Region ) process.on('SIGINT', shutdown) @@ -129,7 +127,6 @@ export async function backupWorkspace ( region, downloadLimit, [], - contextVars, fullCheck ) if (result && onFinish !== undefined) { diff --git a/server/backup/src/service.ts b/server/backup/src/service.ts index e60ec92050..a43f2342e0 100644 --- a/server/backup/src/service.ts +++ b/server/backup/src/service.ts @@ -79,7 +79,6 @@ class BackupWorker { externalStorage: StorageAdapter ) => DbConfiguration, readonly region: string, - readonly contextVars: Record, readonly skipDomains: string[] = [], readonly fullCheck: boolean = false ) { @@ -324,7 +323,6 @@ class BackupWorker { const modelDb = new ModelDb(hierarchy) const txAdapter = await adapterConf.factory( ctx, - this.contextVars, hierarchy, adapterConf.url, wsIds, @@ -332,7 +330,7 @@ class BackupWorker { this.workspaceStorageAdapter ) try { - await txAdapter.init?.(ctx, this.contextVars) + await txAdapter.init?.(ctx, {}) return ( await txAdapter.rawFindAll( @@ -418,18 +416,9 @@ export function backupService ( externalStorage: StorageAdapter ) => DbConfiguration, region: string, - contextVars: Record, recheck?: boolean ): () => void { - const backupWorker = new BackupWorker( - storage, - config, - pipelineFactory, - workspaceStorageAdapter, - getConfig, - region, - contextVars - ) + const backupWorker = new BackupWorker(storage, config, pipelineFactory, workspaceStorageAdapter, getConfig, region) const shutdown = (): void => { void backupWorker.close() @@ -455,7 +444,6 @@ export async function doBackupWorkspace ( region: string, downloadLimit: number, skipDomains: string[], - contextVars: Record, fullCheck: boolean = false, notify?: (progress: number) => Promise ): Promise { @@ -466,7 +454,6 @@ export async function doBackupWorkspace ( workspaceStorageAdapter, getConfig, region, - contextVars, skipDomains, fullCheck ) diff --git a/server/core/src/adapter.ts b/server/core/src/adapter.ts index 5be268f620..89776f06fb 100644 --- a/server/core/src/adapter.ts +++ b/server/core/src/adapter.ts @@ -108,12 +108,7 @@ export interface TxAdapter extends DbAdapter { * @public */ export interface WorkspaceDestroyAdapter { - deleteWorkspace: ( - ctx: MeasureContext, - contextVars: Record, - workspace: WorkspaceUuid, - dataId?: string - ) => Promise + deleteWorkspace: (ctx: MeasureContext, workspace: WorkspaceUuid, dataId?: string) => Promise } /** @@ -121,7 +116,6 @@ export interface WorkspaceDestroyAdapter { */ export type DbAdapterFactory = ( ctx: MeasureContext, - contextVars: Record, hierarchy: Hierarchy, url: string, workspaceId: WorkspaceIds, diff --git a/server/core/src/benchmark/index.ts b/server/core/src/benchmark/index.ts index 44f95a68aa..6468b9a80a 100644 --- a/server/core/src/benchmark/index.ts +++ b/server/core/src/benchmark/index.ts @@ -125,7 +125,6 @@ class BenchmarkDbAdapter extends DummyDbAdapter { */ export async function createBenchmarkAdapter ( ctx: MeasureContext, - contextVars: Record, hierarchy: Hierarchy, url: string, workspaceId: WorkspaceIds, diff --git a/server/core/src/mem.ts b/server/core/src/mem.ts index 3a3d683a43..a8a7542979 100644 --- a/server/core/src/mem.ts +++ b/server/core/src/mem.ts @@ -181,7 +181,6 @@ class InMemoryAdapter extends DummyDbAdapter implements DbAdapter { */ export async function createInMemoryAdapter ( ctx: MeasureContext, - contextVars: Record, hierarchy: Hierarchy, url: string, workspaceId: WorkspaceIds diff --git a/server/core/src/nullAdapter.ts b/server/core/src/nullAdapter.ts index b239cc8e9f..9c0d3adb7e 100644 --- a/server/core/src/nullAdapter.ts +++ b/server/core/src/nullAdapter.ts @@ -22,7 +22,6 @@ import { DummyDbAdapter } from './mem' */ export async function createNullAdapter ( ctx: MeasureContext, - contextVars: Record, hierarchy: Hierarchy, url: string, workspaceId: WorkspaceIds, diff --git a/server/middleware/src/dbAdapter.ts b/server/middleware/src/dbAdapter.ts index efd0b4213d..4cd7825e5c 100644 --- a/server/middleware/src/dbAdapter.ts +++ b/server/middleware/src/dbAdapter.ts @@ -55,7 +55,6 @@ export class DBAdapterMiddleware extends BaseMiddleware implements Middleware { key, await adapterConf.factory( ctx, - this.context.contextVars, this.context.hierarchy, adapterConf.url, this.context.workspace, diff --git a/server/mongo/src/__tests__/storage.test.ts b/server/mongo/src/__tests__/storage.test.ts index 0464657fd0..9bfb163073 100644 --- a/server/mongo/src/__tests__/storage.test.ts +++ b/server/mongo/src/__tests__/storage.test.ts @@ -80,7 +80,6 @@ describe('mongo operations', () => { const mctx = new MeasureMetricsContext('', {}) const txStorage = await createMongoTxAdapter( new MeasureMetricsContext('', {}), - {}, hierarchy, mongodbUri, { @@ -92,7 +91,6 @@ describe('mongo operations', () => { serverStorage = await createMongoAdapter( new MeasureMetricsContext('', {}), - {}, hierarchy, mongodbUri, { diff --git a/server/mongo/src/index.ts b/server/mongo/src/index.ts index f9c5913220..8f15486013 100644 --- a/server/mongo/src/index.ts +++ b/server/mongo/src/index.ts @@ -22,7 +22,7 @@ export * from './utils' export function createMongoDestroyAdapter (url: string): WorkspaceDestroyAdapter { return { - deleteWorkspace: async (ctx, contextVars, workspace, dataId): Promise => { + deleteWorkspace: async (ctx, workspace, dataId): Promise => { const client = getMongoClient(url) try { await ctx.with('delete-workspace', {}, async () => { diff --git a/server/mongo/src/storage.ts b/server/mongo/src/storage.ts index 012a2a3ff3..c9269773d9 100644 --- a/server/mongo/src/storage.ts +++ b/server/mongo/src/storage.ts @@ -1767,7 +1767,6 @@ function translateLikeQuery (pattern: string): { $regex: string, $options: strin */ export async function createMongoAdapter ( ctx: MeasureContext, - contextVars: Record, hierarchy: Hierarchy, url: string, workspaceId: WorkspaceIds, @@ -1786,7 +1785,6 @@ export async function createMongoAdapter ( */ export async function createMongoTxAdapter ( ctx: MeasureContext, - contextVars: Record, hierarchy: Hierarchy, url: string, workspaceId: WorkspaceIds, diff --git a/server/mongo/src/utils.ts b/server/mongo/src/utils.ts index ce7692afeb..f1fc4105c4 100644 --- a/server/mongo/src/utils.ts +++ b/server/mongo/src/utils.ts @@ -25,7 +25,7 @@ const clientRefs = new Map() /** * @public */ -export async function shutdownMongo (contextVars: Record = {}): Promise { +export async function shutdownMongo (): Promise { for (const it of Array.from(clientRefs.values())) { console.error((it as any).stack) } diff --git a/server/postgres-base/.eslintrc.js b/server/postgres-base/.eslintrc.js new file mode 100644 index 0000000000..ce90fb9646 --- /dev/null +++ b/server/postgres-base/.eslintrc.js @@ -0,0 +1,7 @@ +module.exports = { + extends: ['./node_modules/@hcengineering/platform-rig/profiles/node/eslint.config.json'], + parserOptions: { + tsconfigRootDir: __dirname, + project: './tsconfig.json' + } +} diff --git a/server/postgres-base/.npmignore b/server/postgres-base/.npmignore new file mode 100644 index 0000000000..e3ec093c38 --- /dev/null +++ b/server/postgres-base/.npmignore @@ -0,0 +1,4 @@ +* +!/lib/** +!CHANGELOG.md +/lib/**/__tests__/ diff --git a/server/postgres-base/config/rig.json b/server/postgres-base/config/rig.json new file mode 100644 index 0000000000..78cc5a1733 --- /dev/null +++ b/server/postgres-base/config/rig.json @@ -0,0 +1,5 @@ +{ + "$schema": "https://developer.microsoft.com/json-schemas/rig-package/rig.schema.json", + "rigPackageName": "@hcengineering/platform-rig", + "rigProfile": "node" +} diff --git a/server/postgres-base/jest.config.js b/server/postgres-base/jest.config.js new file mode 100644 index 0000000000..2cfd408b67 --- /dev/null +++ b/server/postgres-base/jest.config.js @@ -0,0 +1,7 @@ +module.exports = { + preset: 'ts-jest', + testEnvironment: 'node', + testMatch: ['**/?(*.)+(spec|test).[jt]s?(x)'], + roots: ["./src"], + coverageReporters: ["text-summary", "html"] +} diff --git a/server/postgres-base/package.json b/server/postgres-base/package.json new file mode 100644 index 0000000000..c150c3d35b --- /dev/null +++ b/server/postgres-base/package.json @@ -0,0 +1,39 @@ +{ + "name": "@hcengineering/postgres-base", + "version": "0.6.0", + "main": "lib/index.js", + "svelte": "src/index.ts", + "types": "types/index.d.ts", + "author": "Copyright © Hardcore Engineering Inc.", + "template": "@hcengineering/node-package", + "license": "EPL-2.0", + "scripts": { + "build": "compile", + "build:watch": "compile", + "test": "jest --passWithNoTests --silent --forceExit", + "format": "format src", + "_phase:build": "compile transpile src", + "_phase:test": "jest --passWithNoTests --silent --forceExit", + "_phase:format": "format src", + "_phase:validate": "compile validate" + }, + "devDependencies": { + "@hcengineering/platform-rig": "^0.6.0", + "@typescript-eslint/eslint-plugin": "^6.11.0", + "eslint-plugin-import": "^2.26.0", + "eslint-plugin-promise": "^6.1.1", + "eslint-plugin-n": "^15.4.0", + "eslint": "^8.54.0", + "@typescript-eslint/parser": "^6.11.0", + "eslint-config-standard-with-typescript": "^40.0.0", + "prettier": "^3.1.0", + "typescript": "^5.8.3", + "jest": "^29.7.0", + "ts-jest": "^29.1.1", + "@types/jest": "^29.5.5", + "@types/node": "^22.15.29" + }, + "dependencies": { + "postgres": "^3.4.7" + } +} diff --git a/server/postgres-base/src/index.ts b/server/postgres-base/src/index.ts new file mode 100644 index 0000000000..9c60dbb356 --- /dev/null +++ b/server/postgres-base/src/index.ts @@ -0,0 +1,434 @@ +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +import postgres, { type Options, type ParameterOrJSON } from 'postgres' + +const clientRefs = new Map() + +let clId = 0 + +export type DBResult = any[] & { count: number } +export interface DBClient { + execute: (query: string, parameters?: ParameterOrJSON[] | undefined) => Promise + + release: () => void + + reserve: () => Promise + + raw: () => postgres.Sql +} + +export function createDBClient (client: postgres.Sql, release: () => void = () => {}): DBClient { + return { + execute: (query, parameters) => + client.unsafe(query, doFetchTypes ? parameters : convertArrayParams(parameters), getPrepare()), + release, + reserve: async () => { + const reserved = await client.reserve() + return createDBClient(reserved, () => { + reserved.release() + }) + }, + raw: () => client + } +} + +export function convertArrayParams (params?: unknown[]): any[] | undefined { + if (params === undefined) return undefined + + return params.map((param) => { + if (!Array.isArray(param)) return param + + if (param.length === 0) return '{}' + + const sanitized = param.map((item) => { + if (item === null || item === undefined) return 'NULL' + + if (typeof item === 'number' || typeof item === 'boolean') { + return String(item) + } + + if (typeof item === 'string') { + const escaped = item.replace(/\\/g, '\\\\').replace(/"/g, '\\"') + return `"${escaped}"` + } + + const json = JSON.stringify(item) + const escapedJson = json.replace(/\\/g, '\\\\').replace(/"/g, '\\"') + return `"${escapedJson}"` + }) + + return `{${sanitized.join(',')}}` + }) +} + +export async function retryTxn ( + pool: postgres.Sql, + operation: (client: postgres.TransactionSql) => Promise +): Promise { + await pool.begin(async (client) => { + const result = await operation(client) + return result + }) +} + +/** + * @public + */ +export async function shutdownPostgres (): Promise { + for (const c of connections.values()) { + c.close(true) + } + connections.clear() +} + +export interface PostgresClientReference { + getClient: () => Promise + + mgr: ConnectionMgr + close: () => void + url: () => string +} + +class PostgresClientReferenceImpl { + count: number + client: postgres.Sql + + mgr: ConnectionMgr + + constructor ( + readonly connectionString: string, + client: postgres.Sql, + readonly onclose: () => void + ) { + this.count = 0 + this.client = client + this.mgr = new ConnectionMgr(createDBClient(this.client)) + } + + url (): string { + return this.connectionString + } + + getClient (): postgres.Sql { + return this.client + } + + close (force: boolean = false): void { + this.count-- + if (this.count === 0 || force) { + if (force) { + this.count = 0 + } + void (async () => { + this.onclose() + const cl = this.client + await cl.end({ timeout: 1 }) + })() + } + } + + addRef (): void { + this.count++ + } +} +export class ClientRef implements PostgresClientReference { + id = ++clId + constructor ( + readonly client: PostgresClientReferenceImpl, + readonly mgr: ConnectionMgr + ) { + clientRefs.set(this.id, this) + } + + url (): string { + return this.client.url() + } + + closed = false + async getClient (): Promise { + if (!this.closed) { + return this.client.getClient() + } else { + throw Error('DB client is already closed') + } + } + + close (): void { + // Do not allow double close of mongo connection client + if (!this.closed) { + clientRefs.delete(this.id) + this.closed = true + this.client.close() + } + } +} + +export let dbExtraOptions: Partial> = {} +export function setDBExtraOptions (options: Partial>): void { + dbExtraOptions = options +} + +export function getPrepare (): { prepare: boolean } { + return { prepare: dbExtraOptions.prepare ?? false } +} + +export const doFetchTypes = true + +const connections = new Map() + +/** + * Initialize a connection to DB + * @public + */ +export function getDBClient ( + connectionString: string, + database?: string, + serviceName: string = 'transactor' +): PostgresClientReference { + const extraOptions = JSON.parse(process.env.POSTGRES_OPTIONS ?? '{}') + const key = `${connectionString}${extraOptions}` + + let existing = connections.get(key) + + if (existing === undefined) { + const sql = postgres(connectionString, { + connection: { + application_name: serviceName + }, + database, + max: 10, + min: 2, + connect_timeout: 30, + idle_timeout: 0, + transform: { + undefined: null + }, + debug: false, + notice: false, + onnotice (notice) {}, + onparameter (key, value) {}, + ...dbExtraOptions, + ...extraOptions, + fetch_types: doFetchTypes + }) + + existing = new PostgresClientReferenceImpl(connectionString, sql, () => { + connections.delete(key) + }) + connections.set(key, existing) + } + // Add reference and return once closable + existing.addRef() + return new ClientRef(existing, existing.mgr) +} + +class ConnectionInfo { + // It should preserve at least one available connection in pool, other connection should be closed + available: DBClient[] = [] + + released: boolean = false + + constructor ( + readonly connectionId: string, + protected readonly client: DBClient, + readonly managed: boolean + ) {} + + async withReserve (action: (reservedClient: DBClient) => Promise, forced: boolean = false): Promise { + let reserved: DBClient | undefined + + // Check if we have at least one available connection and reserve one more if required. + if (this.available.length === 0) { + if (this.managed || forced) { + reserved = await this.client.reserve() + } + } else { + reserved = this.available.shift() as DBClient + } + + try { + // Use reserved or pool + return await action(reserved ?? this.client) + } catch (err: any) { + console.error(err) + throw err + } finally { + if (this.released) { + try { + reserved?.release() + } catch (err: any) { + console.error('failed to release', err) + } + } else if (reserved !== undefined) { + if (this.available.length > 0) { + reserved?.release() + } else { + this.available.push(reserved) + } + } + } + } + + release (): void { + for (const c of [...this.available]) { + c.release() + } + this.available = [] + } +} + +export class ConnectionMgr { + private readonly connections = new Map() + constructor (protected readonly client: DBClient) {} + + async write (id: string | undefined, fn: (client: DBClient) => Promise): Promise { + const backoffInterval = 25 // millis + const maxTries = 5 + let tries = 0 + + const realId = id ?? `${++clId}` + + const connection = this.getConnection(realId, false) + + try { + while (true) { + const retry: boolean | Error = await connection.withReserve(async (client) => { + tries++ + try { + await client.execute('BEGIN;') + await fn(client) + await client.execute('COMMIT;') + return true + } catch (err: any) { + await client.execute('ROLLBACK;') + console.error({ message: 'failed to process tx', error: err.message, cause: err }) + + if (!this.isRetryableError(err) || tries === maxTries) { + return err + } else { + console.log('Transaction failed. Retrying.') + console.log(err.message) + return false + } + } + }, true) + if (retry === true) { + break + } + if (retry instanceof Error) { + // Pass it to exit + throw retry + } + // Retry for a timeout + await new Promise((resolve) => setTimeout(resolve, backoffInterval)) + } + } finally { + if (!connection.managed) { + // We need to relase in case it temporaty connection was used + connection.release() + } + } + } + + async retry (id: string | undefined, fn: (client: DBClient) => Promise): Promise { + const backoffInterval = 25 // millis + const maxTries = 5 + let tries = 0 + + const realId = id ?? `${++clId}` + // Will reuse reserved if had and use new one if not + const connection = this.getConnection(realId, false) + + try { + while (true) { + const retry: false | { result: any } | Error = await connection.withReserve(async (client) => { + tries++ + try { + return { result: await fn(client) } + } catch (err: any) { + console.error({ message: 'failed to process sql', error: err.message, cause: err }) + if (!this.isRetryableError(err) || tries === maxTries) { + return err + } else { + console.log('Read Transaction failed. Retrying.') + console.log(err.message) + return false + } + } + }) + if (retry instanceof Error) { + // Pass it to exit + throw retry + } + if (retry === false) { + // Retry for a timeout + await new Promise((resolve) => setTimeout(resolve, backoffInterval)) + continue + } + return retry.result + } + } finally { + if (!connection.managed) { + // We need to relase in case it temporaty connection was used + connection.release() + } + } + } + + release (id: string): void { + const conn = this.connections.get(id) + if (conn !== undefined) { + conn.released = true + this.connections.delete(id) // We need to delete first + conn.release() + } + } + + close (): void { + const cnts = this.connections + for (const [k, conn] of Array.from(cnts.entries())) { + cnts.delete(k) + try { + conn.release() + } catch (err: any) { + console.error('failed to release connection') + } + } + } + + getConnection (id: string, managed: boolean = true): ConnectionInfo { + let conn = this.connections.get(id) + if (conn === undefined) { + conn = new ConnectionInfo(id, this.client, managed) + } + if (managed) { + this.connections.set(id, conn) + } + return conn + } + + private isRetryableError (err: any): boolean { + const msg: string = err?.message ?? '' + + return ( + err.code === '40001' || // Retry transaction + err.code === '55P03' || // Lock not available + err.code === 'CONNECTION_CLOSED' || // This error is thrown if the connection was closed without an error. + err.code === 'CONNECTION_DESTROYED' || // This error is thrown for any queries that were pending when the timeout to sql.end({ timeout: X }) was reached. If the DB client is being closed completely retry will result in CONNECTION_ENDED which is not retried so should be fine. + msg.includes('RETRY_SERIALIZABLE') + ) + } +} diff --git a/server/postgres-base/tsconfig.json b/server/postgres-base/tsconfig.json new file mode 100644 index 0000000000..c6a877cf6c --- /dev/null +++ b/server/postgres-base/tsconfig.json @@ -0,0 +1,12 @@ +{ + "extends": "./node_modules/@hcengineering/platform-rig/profiles/node/tsconfig.json", + + "compilerOptions": { + "rootDir": "./src", + "outDir": "./lib", + "declarationDir": "./types", + "tsBuildInfoFile": ".build/build.tsbuildinfo" + }, + "include": ["src/**/*"], + "exclude": ["node_modules", "lib", "dist", "types", "bundle"] +} \ No newline at end of file diff --git a/server/postgres/package.json b/server/postgres/package.json index 42f1c7ec62..d09b44fb3f 100644 --- a/server/postgres/package.json +++ b/server/postgres/package.json @@ -37,6 +37,7 @@ "postgres": "^3.4.7", "@hcengineering/core": "^0.6.32", "@hcengineering/platform": "^0.6.11", - "@hcengineering/server-core": "^0.6.1" + "@hcengineering/server-core": "^0.6.1", + "@hcengineering/postgres-base": "^0.6.0" } } diff --git a/server/postgres/src/__tests__/conversion.spec.ts b/server/postgres/src/__tests__/conversion.spec.ts index b8ccaf07bf..47421a8199 100644 --- a/server/postgres/src/__tests__/conversion.spec.ts +++ b/server/postgres/src/__tests__/conversion.spec.ts @@ -14,6 +14,7 @@ import { PostgresAdapter } from '../storage' import { convertArrayParams, decodeArray, filterProjection } from '../utils' import { genMinModel, test, type ComplexClass } from './minmodel' import { createDummyClient, type TypedQuery } from './utils' +import { ConnectionMgr } from '@hcengineering/postgres-base' describe('array conversion', () => { it('should handle undefined parameters', () => { @@ -149,6 +150,7 @@ function createTestContext (): { adapter: PostgresAdapter, ctx: MeasureMetricsCo modelDb.addTxes(ctx, minModel, true) const adapter = new PostgresAdapter( c, + new ConnectionMgr(c), { url: () => 'test', close: () => {} diff --git a/server/postgres/src/__tests__/storage.test.ts b/server/postgres/src/__tests__/storage.test.ts index 88ab3b367d..4efad9985e 100644 --- a/server/postgres/src/__tests__/storage.test.ts +++ b/server/postgres/src/__tests__/storage.test.ts @@ -25,8 +25,13 @@ import core, { type WorkspaceUuid } from '@hcengineering/core' import { type DbAdapter, wrapAdapterToClient } from '@hcengineering/server-core' -import { createPostgresAdapter, createPostgresTxAdapter } from '..' -import { getDBClient, type PostgresClientReference, shutdownPostgres } from '../utils' +import { + createPostgresAdapter, + createPostgresTxAdapter, + getDBClient, + shutdownPostgres, + type PostgresClientReference +} from '..' import { genMinModel } from './minmodel' import { createTaskModel, type Task, type TaskComment, taskPlugin } from './tasks' @@ -40,7 +45,7 @@ describe('postgres operations', () => { const baseDbUri: string = process.env.DB_URL ?? 'postgresql://root@localhost:26257/defaultdb?sslmode=disable' let dbUuid = crypto.randomUUID() as WorkspaceUuid let dbUri: string = baseDbUri.replace('defaultdb', dbUuid) - const clientRef: PostgresClientReference = getDBClient(contextVars, baseDbUri) + const clientRef: PostgresClientReference = getDBClient(baseDbUri) let hierarchy: Hierarchy let model: ModelDb let client: Client @@ -49,7 +54,7 @@ describe('postgres operations', () => { afterAll(async () => { clientRef.close() - await shutdownPostgres(contextVars) + await shutdownPostgres() }) beforeEach(async () => { @@ -90,7 +95,6 @@ describe('postgres operations', () => { const mctx = new MeasureMetricsContext('', {}) const txStorage = await createPostgresTxAdapter( mctx, - contextVars, hierarchy, dbUri, { @@ -110,7 +114,6 @@ describe('postgres operations', () => { const ctx = new MeasureMetricsContext('client', {}) const serverStorage = await createPostgresAdapter( ctx, - contextVars, hierarchy, dbUri, { diff --git a/server/postgres/src/__tests__/utils.ts b/server/postgres/src/__tests__/utils.ts index 96b2da94ee..0266183437 100644 --- a/server/postgres/src/__tests__/utils.ts +++ b/server/postgres/src/__tests__/utils.ts @@ -1,4 +1,4 @@ -import type { DBClient } from '../client' +import type { DBClient } from '@hcengineering/postgres-base' export interface TypedQuery { query: string diff --git a/server/postgres/src/client.ts b/server/postgres/src/client.ts deleted file mode 100644 index 91fdc4fddc..0000000000 --- a/server/postgres/src/client.ts +++ /dev/null @@ -1,29 +0,0 @@ -import type postgres from 'postgres' -import type { ParameterOrJSON } from 'postgres' -import { convertArrayParams, doFetchTypes, getPrepare } from './utils' - -export type DBResult = any[] & { count: number } -export interface DBClient { - execute: (query: string, parameters?: ParameterOrJSON[] | undefined) => Promise - - release: () => void - - reserve: () => Promise - - raw: () => postgres.Sql -} - -export function createDBClient (client: postgres.Sql, release: () => void = () => {}): DBClient { - return { - execute: (query, parameters) => - client.unsafe(query, doFetchTypes ? parameters : convertArrayParams(parameters), getPrepare()), - release, - reserve: async () => { - const reserved = await client.reserve() - return createDBClient(reserved, () => { - reserved.release() - }) - }, - raw: () => client - } -} diff --git a/server/postgres/src/index.ts b/server/postgres/src/index.ts index 21b1b3c114..9e6646e1da 100644 --- a/server/postgres/src/index.ts +++ b/server/postgres/src/index.ts @@ -13,19 +13,20 @@ // limitations under the License. // +import { getDBClient, retryTxn } from '@hcengineering/postgres-base' import type { WorkspaceDestroyAdapter } from '@hcengineering/server-core' import { domainSchemas } from './schemas' -import { getDBClient, retryTxn } from './utils' -export { createDBClient } from './client' export { getDocFieldsByDomains, translateDomain } from './schemas' export * from './storage' -export { convertDoc, createTables, getDBClient, retryTxn, setDBExtraOptions, shutdownPostgres } from './utils' +export { convertDoc, createTables } from './utils' + +export * from '@hcengineering/postgres-base' export function createPostgreeDestroyAdapter (url: string): WorkspaceDestroyAdapter { return { - deleteWorkspace: async (ctx, contextVars, workspaceUuid): Promise => { - const client = getDBClient(contextVars, url) + deleteWorkspace: async (ctx, workspaceUuid): Promise => { + const client = getDBClient(url) try { if (workspaceUuid == null) { throw new Error('Workspace uuid is not defined') diff --git a/server/postgres/src/storage.ts b/server/postgres/src/storage.ts index 55f70f3ed0..e9178aa65c 100644 --- a/server/postgres/src/storage.ts +++ b/server/postgres/src/storage.ts @@ -30,7 +30,6 @@ import core, { DOMAIN_TX, type FindOptions, type FindResult, - generateId, groupByArray, type Hierarchy, isOperator, @@ -62,6 +61,13 @@ import core, { type WorkspaceIds, type WorkspaceUuid } from '@hcengineering/core' +import { + type ConnectionMgr, + createDBClient, + type DBClient, + doFetchTypes, + getDBClient +} from '@hcengineering/postgres-base' import { calcHashHash, type DbAdapter, @@ -72,7 +78,6 @@ import { type TxAdapter } from '@hcengineering/server-core' import type postgres from 'postgres' -import { createDBClient, type DBClient } from './client' import { getDocFieldsByDomains, getSchema, @@ -88,10 +93,8 @@ import { createTables, DBCollectionHelper, type DBDoc, - doFetchTypes, escape, filterProjection, - getDBClient, inferType, isDataField, isOwner, @@ -128,212 +131,6 @@ async function * createCursorGenerator ( } } -class ConnectionInfo { - // It should preserve at least one available connection in pool, other connection should be closed - available: DBClient[] = [] - - released: boolean = false - - constructor ( - readonly mgrId: string, - readonly connectionId: string, - protected readonly client: DBClient, - readonly managed: boolean - ) {} - - async withReserve (action: (reservedClient: DBClient) => Promise, forced: boolean = false): Promise { - let reserved: DBClient | undefined - - // Check if we have at least one available connection and reserve one more if required. - if (this.available.length === 0) { - if (this.managed || forced) { - reserved = await this.client.reserve() - } - } else { - reserved = this.available.shift() as DBClient - } - - try { - // Use reserved or pool - return await action(reserved ?? this.client) - } catch (err: any) { - console.error(err) - throw err - } finally { - if (this.released) { - try { - reserved?.release() - } catch (err: any) { - console.error('failed to release', err) - } - } else if (reserved !== undefined) { - if (this.available.length > 0) { - reserved?.release() - } else { - this.available.push(reserved) - } - } - } - } - - release (): void { - for (const c of [...this.available]) { - c.release() - } - this.available = [] - } -} - -class ConnectionMgr { - constructor ( - protected readonly client: DBClient, - protected readonly connections: () => Map, - readonly mgrId: string - ) {} - - async write (id: string | undefined, fn: (client: DBClient) => Promise): Promise { - const backoffInterval = 25 // millis - const maxTries = 5 - let tries = 0 - - const realId = id ?? generateId() - - const connection = this.getConnection(realId, false) - - try { - while (true) { - const retry: boolean | Error = await connection.withReserve(async (client) => { - tries++ - try { - await client.execute('BEGIN;') - await fn(client) - await client.execute('COMMIT;') - return true - } catch (err: any) { - await client.execute('ROLLBACK;') - console.error({ message: 'failed to process tx', error: err.message, cause: err }) - - if (!this.isRetryableError(err) || tries === maxTries) { - return err - } else { - console.log('Transaction failed. Retrying.') - console.log(err.message) - return false - } - } - }, true) - if (retry === true) { - break - } - if (retry instanceof Error) { - // Pass it to exit - throw retry - } - // Retry for a timeout - await new Promise((resolve) => setTimeout(resolve, backoffInterval)) - } - } finally { - if (!connection.managed) { - // We need to relase in case it temporaty connection was used - connection.release() - } - } - } - - async retry (id: string | undefined, fn: (client: DBClient) => Promise): Promise { - const backoffInterval = 25 // millis - const maxTries = 5 - let tries = 0 - - const realId = id ?? generateId() - // Will reuse reserved if had and use new one if not - const connection = this.getConnection(realId, false) - - try { - while (true) { - const retry: false | { result: any } | Error = await connection.withReserve(async (client) => { - tries++ - try { - return { result: await fn(client) } - } catch (err: any) { - console.error({ message: 'failed to process sql', error: err.message, cause: err }) - if (!this.isRetryableError(err) || tries === maxTries) { - return err - } else { - console.log('Read Transaction failed. Retrying.') - console.log(err.message) - return false - } - } - }) - if (retry instanceof Error) { - // Pass it to exit - throw retry - } - if (retry === false) { - // Retry for a timeout - await new Promise((resolve) => setTimeout(resolve, backoffInterval)) - continue - } - return retry.result - } - } finally { - if (!connection.managed) { - // We need to relase in case it temporaty connection was used - connection.release() - } - } - } - - release (id: string): void { - const conn = this.connections().get(id) - if (conn !== undefined) { - conn.released = true - this.connections().delete(id) // We need to delete first - conn.release() - } else { - console.log('wrne') - } - } - - close (): void { - const cnts = this.connections() - for (const [k, conn] of Array.from(cnts.entries()).filter( - ([, it]: [string, ConnectionInfo]) => it.mgrId === this.mgrId - )) { - cnts.delete(k) - try { - conn.release() - } catch (err: any) { - console.error('failed to release connection') - } - } - } - - getConnection (id: string, managed: boolean = true): ConnectionInfo { - let conn = this.connections().get(id) - if (conn === undefined) { - conn = new ConnectionInfo(this.mgrId, id, this.client, managed) - } - if (managed) { - this.connections().set(id, conn) - } - return conn - } - - private isRetryableError (err: any): boolean { - const msg: string = err?.message ?? '' - - return ( - err.code === '40001' || // Retry transaction - err.code === '55P03' || // Lock not available - err.code === 'CONNECTION_CLOSED' || // This error is thrown if the connection was closed without an error. - err.code === 'CONNECTION_DESTROYED' || // This error is thrown for any queries that were pending when the timeout to sql.end({ timeout: X }) was reached. If the DB client is being closed completely retry will result in CONNECTION_ENDED which is not retried so should be fine. - msg.includes('RETRY_SERIALIZABLE') - ) - } -} - class ValuesVariables { index: number = 1 values: any[] = [] @@ -409,12 +206,10 @@ abstract class PostgresAdapterBase implements DbAdapter { protected readonly _helper: DBCollectionHelper protected readonly tableFields = new Map() - protected connections = new Map() - - mgr: ConnectionMgr - constructor ( protected readonly client: DBClient, + + protected readonly mgr: ConnectionMgr, protected readonly refClient: { url: () => string close: () => void @@ -425,15 +220,12 @@ abstract class PostgresAdapterBase implements DbAdapter { readonly mgrId: string ) { this._helper = new DBCollectionHelper(this.client, this.workspaceId) - this.mgr = new ConnectionMgr(client, () => this.connections, mgrId) } reserveContext (id: string): () => void { - const conn = this.mgr.getConnection(id, true) + this.mgr.getConnection(id, true) return () => { - conn.released = true - conn.release() - this.connections.delete(id) // We need to delete first + this.mgr.release(id) // We need to release first } } @@ -1863,8 +1655,6 @@ export class PostgresAdapter extends PostgresAdapterBase { domains?: string[], excludeDomains?: string[] ): Promise { - this.connections = contextVars.cntInfoPG ?? new Map() - contextVars.cntInfoPG = this.connections let resultDomains = [...(domains ?? this.hierarchy.domains()), 'kanban'] if (excludeDomains !== undefined) { resultDomains = resultDomains.filter((it) => !excludeDomains.includes(it)) @@ -2186,9 +1976,6 @@ class PostgresTxAdapter extends PostgresAdapterBase implements TxAdapter { domains?: string[], excludeDomains?: string[] ): Promise { - this.connections = contextVars.cntInfoPG ?? new Map() - contextVars.cntInfoPG = this.connections - const resultDomains = domains ?? [DOMAIN_TX, DOMAIN_MODEL_TX] await initRateLimit.exec(async () => { const url = this.refClient.url() @@ -2263,31 +2050,45 @@ function prepareJsonValue (tkey: string, valType: string): { tlkey: string, arro */ export async function createPostgresAdapter ( ctx: MeasureContext, - contextVars: Record, hierarchy: Hierarchy, url: string, wsIds: WorkspaceIds, modelDb: ModelDb ): Promise { - const client = getDBClient(contextVars, url) + const client = getDBClient(url) const connection = await client.getClient() - return new PostgresAdapter(createDBClient(connection), client, wsIds.uuid, hierarchy, modelDb, 'default-' + wsIds.url) + return new PostgresAdapter( + createDBClient(connection), + client.mgr, + client, + wsIds.uuid, + hierarchy, + modelDb, + 'default-' + wsIds.url + ) } /** * @public */ export async function createPostgresTxAdapter ( ctx: MeasureContext, - contextVars: Record, hierarchy: Hierarchy, url: string, wsIds: WorkspaceIds, modelDb: ModelDb ): Promise { - const client = getDBClient(contextVars, url) + const client = getDBClient(url) const connection = await client.getClient() - return new PostgresTxAdapter(createDBClient(connection), client, wsIds.uuid, hierarchy, modelDb, 'tx' + wsIds.url) + return new PostgresTxAdapter( + createDBClient(connection), + client.mgr, + client, + wsIds.uuid, + hierarchy, + modelDb, + 'tx' + wsIds.url + ) } function isPersonAccount (tx: Tx): boolean { diff --git a/server/postgres/src/utils.ts b/server/postgres/src/utils.ts index 7e59c7b730..3712b721ee 100644 --- a/server/postgres/src/utils.ts +++ b/server/postgres/src/utils.ts @@ -21,7 +21,6 @@ import core, { type DocumentUpdate, type Domain, type FieldIndexConfig, - generateId, type MeasureContext, type MixinUpdate, platformNow, @@ -31,10 +30,9 @@ import core, { systemAccountUuid, type WorkspaceUuid } from '@hcengineering/core' -import { PlatformError, unknownStatus } from '@hcengineering/platform' import { type DomainHelperOperations } from '@hcengineering/server-core' -import postgres, { type Options, type ParameterOrJSON } from 'postgres' -import type { DBClient } from './client' +import type postgres from 'postgres' +import { type ParameterOrJSON } from 'postgres' import { addSchema, type DataType, @@ -46,22 +44,12 @@ import { type SchemaAndFields, translateDomain } from './schemas' +import { retryTxn, type DBClient } from '@hcengineering/postgres-base' -const clientRefs = new Map() const loadedDomains = new Set() let loadedTables = new Set() -export async function retryTxn ( - pool: postgres.Sql, - operation: (client: postgres.TransactionSql) => Promise -): Promise { - await pool.begin(async (client) => { - const result = await operation(client) - return result - }) -} - export const NumericTypes = [ core.class.TypeNumber, core.class.TypeTimestamp, @@ -190,158 +178,6 @@ async function createTable (client: postgres.Sql, domain: string): Promise } } -/** - * @public - */ -export async function shutdownPostgres (contextVars: Record): Promise { - const connections: Map | undefined = - contextVars.pgConnections ?? new Map() - if (connections === undefined) { - return - } - for (const c of connections.values()) { - c.close(true) - } - connections.clear() -} - -export interface PostgresClientReference { - getClient: () => Promise - close: () => void - - url: () => string -} - -class PostgresClientReferenceImpl { - count: number - client: postgres.Sql | Promise - - constructor ( - readonly connectionString: string, - client: postgres.Sql | Promise, - readonly onclose: () => void - ) { - this.count = 0 - this.client = client - } - - url (): string { - return this.connectionString - } - - async getClient (): Promise { - if (this.client instanceof Promise) { - this.client = await this.client - } - return this.client - } - - close (force: boolean = false): void { - this.count-- - if (this.count === 0 || force) { - if (force) { - this.count = 0 - } - void (async () => { - this.onclose() - const cl = await this.client - await cl.end({ timeout: 1 }) - })() - } - } - - addRef (): void { - this.count++ - } -} -export class ClientRef implements PostgresClientReference { - id = generateId() - constructor (readonly client: PostgresClientReferenceImpl) { - clientRefs.set(this.id, this) - } - - url (): string { - return this.client.url() - } - - closed = false - async getClient (): Promise { - if (!this.closed) { - return await this.client.getClient() - } else { - throw new PlatformError(unknownStatus('DB client is already closed')) - } - } - - close (): void { - // Do not allow double close of mongo connection client - if (!this.closed) { - clientRefs.delete(this.id) - this.closed = true - this.client.close() - } - } -} - -export let dbExtraOptions: Partial> = {} -export function setDBExtraOptions (options: Partial>): void { - dbExtraOptions = options -} - -export function getPrepare (): { prepare: boolean } { - return { prepare: dbExtraOptions.prepare ?? false } -} - -export const doFetchTypes = true - -/** - * Initialize a workspace connection to DB - * @public - */ -export function getDBClient ( - contextVars: Record, - connectionString: string, - database?: string -): PostgresClientReference { - const extraOptions = JSON.parse(process.env.POSTGRES_OPTIONS ?? '{}') - const key = `${connectionString}${extraOptions}` - const connections = contextVars.pgConnections ?? new Map() - contextVars.pgConnections = connections - - let existing = connections.get(key) - - if (existing === undefined) { - const sql = postgres(connectionString, { - connection: { - application_name: 'transactor' - }, - database, - max: 10, - min: 2, - connect_timeout: 30, - idle_timeout: 0, - transform: { - undefined: null - }, - debug: false, - notice: false, - onnotice (notice) {}, - onparameter (key, value) {}, - ...dbExtraOptions, - ...extraOptions, - fetch_types: doFetchTypes - }) - - existing = new PostgresClientReferenceImpl(connectionString, sql, () => { - connections.delete(key) - }) - connections.set(key, existing) - } - // Add reference and return once closable - existing.addRef() - return new ClientRef(existing) -} - export function convertDoc ( domain: string, doc: T, diff --git a/server/server-pipeline/src/blobStorage.ts b/server/server-pipeline/src/blobStorage.ts index 0f7904e521..efbb09b1df 100644 --- a/server/server-pipeline/src/blobStorage.ts +++ b/server/server-pipeline/src/blobStorage.ts @@ -152,7 +152,6 @@ class StorageBlobAdapter implements DbAdapter { */ export async function createStorageDataAdapter ( ctx: MeasureContext, - contextVars: Record, hierarchy: Hierarchy, url: string, wsIds: WorkspaceIds, diff --git a/server/server-pipeline/src/pipeline.ts b/server/server-pipeline/src/pipeline.ts index d49c6553c2..2ea62086a3 100644 --- a/server/server-pipeline/src/pipeline.ts +++ b/server/server-pipeline/src/pipeline.ts @@ -86,12 +86,6 @@ export function getTxAdapterFactory ( return adapter.factory } -/** - * A pipelice context used by standalong services to hold global variables. - * In case of Durable Objects, it should not be shared and individual context should be created. - */ -export const sharedPipelineContextVars: Record = {} - /** * @public */ @@ -170,7 +164,7 @@ export function createServerPipeline ( hierarchy, queue: opt.queue, storageAdapter: opt.externalStorage, - contextVars: opt.pipelineContextVars ?? sharedPipelineContextVars + contextVars: opt.pipelineContextVars ?? {} } return createPipeline(ctx, middlewares, context) } diff --git a/server/workspace-service/src/service.ts b/server/workspace-service/src/service.ts index 5c740cf994..facf11a44b 100644 --- a/server/workspace-service/src/service.ts +++ b/server/workspace-service/src/service.ts @@ -72,8 +72,7 @@ import { registerServerPlugins, registerStringLoaders, registerTxAdapterFactory, - setAdapterSecurity, - sharedPipelineContextVars + setAdapterSecurity } from '@hcengineering/server-pipeline' import { buildStorageFromConfig, storageConfigFromEnv } from '@hcengineering/server-storage' import { createWorkspace, upgradeWorkspace } from './ws-operations' @@ -95,10 +94,10 @@ export interface WorkspaceOptions { // Register close on process exit. process.on('exit', () => { - shutdownPostgres(sharedPipelineContextVars).catch((err) => { + shutdownPostgres().catch((err) => { console.error(err) }) - shutdownMongo(sharedPipelineContextVars).catch((err) => { + shutdownMongo().catch((err) => { console.error(err) }) }) @@ -422,7 +421,7 @@ export class WorkspaceWorker { async doCleanup (ctx: MeasureContext, workspace: WorkspaceInfoWithStatus, cleanIndexes: boolean): Promise { const { dbUrl } = prepareTools([]) const adapter = getWorkspaceDestroyAdapter(dbUrl) - await adapter.deleteWorkspace(ctx, sharedPipelineContextVars, workspace.uuid, workspace.dataId) + await adapter.deleteWorkspace(ctx, workspace.uuid, workspace.dataId) await this.workspaceQueue.send(workspace.uuid, [workspaceEvents.clearIndex()]) } @@ -627,7 +626,6 @@ export class WorkspaceWorker { this.region, 50000, ['blob'], - sharedPipelineContextVars, doFullCheck, // Do full check based on config, do not do for migration, it is to slow, will perform before migration. (_p: number) => { if (progress !== Math.round(_p)) { diff --git a/server/workspace-service/src/ws-operations.ts b/server/workspace-service/src/ws-operations.ts index ca31a074ca..c5c9714440 100644 --- a/server/workspace-service/src/ws-operations.ts +++ b/server/workspace-service/src/ws-operations.ts @@ -25,7 +25,7 @@ import { type QueueWorkspaceMessage, type StorageAdapter } from '@hcengineering/server-core' -import { getServerPipeline, getTxAdapterFactory, sharedPipelineContextVars } from '@hcengineering/server-pipeline' +import { getServerPipeline, getTxAdapterFactory } from '@hcengineering/server-pipeline' import { buildStorageFromConfig, storageConfigFromEnv } from '@hcengineering/server-storage' import { generateToken } from '@hcengineering/server-token' import { initializeWorkspace, initModel, prepareTools, updateModel, upgradeModel } from '@hcengineering/server-tool' @@ -91,15 +91,7 @@ export async function createWorkspace ( externalStorage: storageAdapter, usePassedCtx: true }) - const txAdapter = await txFactory( - ctx, - sharedPipelineContextVars, - hierarchy, - dbUrl, - wsIds, - modelDb, - storageAdapter - ) + const txAdapter = await txFactory(ctx, hierarchy, dbUrl, wsIds, modelDb, storageAdapter) await childLogger.withLog('init-workspace', {}, (ctx) => initModel(ctx, wsId, txes, txAdapter, storageAdapter, ctxModellogger, async (value) => {}) )