Oak-Team
/
oak-backend-base
13
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 18 Wiki Activity

feat: 支持根据配置文件进行DbStore的自动选择，现支持mysql与postgres

This commit is contained in:
Pan Qiancheng 2025-12-30 15:46:11 +08:00
parent d772fb80e4
commit 5f2af054fc
14 changed files with 443 additions and 317 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
lib/AppLoader.d.ts vendored
View File

@ -1,7 +1,5 @@
/// <reference types="node" />
import { EntityDict as BaseEntityDict } from 'oak-domain/lib/base-app-domain';
import { AppLoader as GeneralAppLoader, Trigger, EntityDict, Watcher, OpRecord, FreeTimer, OperationResult } from "oak-domain/lib/types";
import { DbStore } from "./DbStore";
import { BackendRuntimeContext } from 'oak-frontend-base/lib/context/BackendRuntimeContext';
import { IncomingHttpHeaders, IncomingMessage } from 'http';
import { Namespace } from 'socket.io';
@ -9,13 +7,14 @@ import DataSubscriber from './cluster/DataSubscriber';
import Synchronizer from './Synchronizer';
import { AsyncContext } from 'oak-domain/lib/store/AsyncRowStore';
import { InternalErrorHandler } from './types';
import { AppDbStore } from './DbStore';
export declare class AppLoader<ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>> extends GeneralAppLoader<ED, Cxt> {
protected dbStore: DbStore<ED, Cxt>;
protected dbStore: AppDbStore<ED, Cxt>;
private aspectDict;
private externalDependencies;
protected dataSubscriber?: DataSubscriber<ED, Cxt>;
protected synchronizer?: Synchronizer<ED, Cxt>;
protected contextBuilder: (store: DbStore<ED, Cxt>) => Cxt;
protected contextBuilder: (store: AppDbStore<ED, Cxt>) => Cxt;
private nsSocket?;
private watcherTimerId?;
private scheduledJobs;
@ -48,11 +47,11 @@ export declare class AppLoader<ED extends EntityDict & BaseEntityDict, Cxt exten
message?: string;
}>;
initialize(ifExists?: 'drop' | 'omit' | 'dropIfNotStatic'): Promise<void>;
getStore(): DbStore<ED, Cxt>;
getStore(): AppDbStore<ED, Cxt>;
getEndpoints(prefix: string): [string, "post" | "get" | "put" | "delete", string, (params: Record<string, string>, headers: IncomingHttpHeaders, req: IncomingMessage, body?: any) => Promise<{
headers?: Record<string, string | string[]> | undefined;
headers?: Record<string, string | string[]>;
data: any;
statusCode?: number | undefined;
statusCode?: number;
}>][];
protected operateInWatcher<T extends keyof ED>(entity: T, operation: ED[T]['Update'], context: Cxt, singleton?: true): Promise<OperationResult<ED>>;
protected selectInWatcher<T extends keyof ED>(entity: T, selection: ED[T]['Selection'], context: Cxt, forUpdate?: true, singleton?: true): Promise<Partial<ED[T]["Schema"]>[]>;

10
lib/AppLoader.js
View File

@ -9,7 +9,6 @@ const IntrinsicLogics_1 = require("oak-domain/lib/store/IntrinsicLogics");
const lodash_1 = require("oak-domain/lib/utils/lodash");
const uuid_1 = require("oak-domain/lib/utils/uuid");
const types_1 = require("oak-domain/lib/types");
const DbStore_1 = require("./DbStore");
const index_1 = tslib_1.__importStar(require("oak-common-aspect/lib/index"));
const assert_1 = tslib_1.__importDefault(require("assert"));
const dependencyBuilder_1 = require("oak-domain/lib/compiler/dependencyBuilder");
@ -18,6 +17,8 @@ const env_1 = require("./cluster/env");
const Synchronizer_1 = tslib_1.__importDefault(require("./Synchronizer"));
const i18n_1 = tslib_1.__importDefault(require("oak-domain/lib/data/i18n"));
const requirePrj_1 = tslib_1.__importDefault(require("./utils/requirePrj"));
const dbPriority_1 = require("./utils/dbPriority");
const DbStore_1 = require("./DbStore");
class AppLoader extends types_1.AppLoader {
dbStore;
aspectDict;
@ -95,8 +96,7 @@ class AppLoader extends types_1.AppLoader {
* 后台启动的configuration，统一放在这里读取
*/
getConfiguration() {
const dbConfigFile = (0, path_1.join)(this.path, 'configuration', 'mysql.json');
const dbConfig = require(dbConfigFile);
const dbConfig = (0, dbPriority_1.getDbConfig)(this.path);
const syncConfigFile = (0, path_1.join)(this.path, 'lib', 'configuration', 'sync.js');
const syncConfigs = (0, fs_1.existsSync)(syncConfigFile) && require(syncConfigFile).default;
return {
@ -112,7 +112,7 @@ class AppLoader extends types_1.AppLoader {
this.externalDependencies = depGraph.ascOrder;
const { authDeduceRelationMap, selectFreeEntities, updateFreeDict } = this.requireSth('lib/configuration/relation');
this.aspectDict = Object.assign({}, index_1.default, this.requireSth('lib/aspects/index'));
this.dbStore = new DbStore_1.DbStore(storageSchema, () => this.contextBuilder(this.dbStore), dbConfig, authDeduceRelationMap, selectFreeEntities, updateFreeDict);
this.dbStore = (0, DbStore_1.createDbStore)(storageSchema, () => this.contextBuilder(this.dbStore), dbConfig, authDeduceRelationMap, selectFreeEntities, updateFreeDict);
if (nsSubscribe) {
this.dataSubscriber = new DataSubscriber_1.default(nsSubscribe, nsServer);
}
@ -290,7 +290,7 @@ class AppLoader extends types_1.AppLoader {
}
}
}
await this.dbStore.disconnect();
// await this.dbStore.disconnect(); // 不需要马上断开连接，在initialize后可能还会有操作，unmount时会断开
}
getStore() {
return this.dbStore;

22
lib/DbStore.d.ts vendored
View File

@ -1,22 +1,16 @@
import { MysqlStore, MySqlSelectOption, MysqlOperateOption } from 'oak-db';
import { EntityDict, StorageSchema, Trigger, Checker, SelectOption, SelectFreeEntities, UpdateFreeDict, AuthDeduceRelationMap, VolatileTrigger, OperateOption } from 'oak-domain/lib/types';
import { DbConfiguration } from 'oak-db/src/types/configuration';
import { EntityDict, StorageSchema, Trigger, Checker, SelectFreeEntities, UpdateFreeDict, AuthDeduceRelationMap, VolatileTrigger, OperateOption } from 'oak-domain/lib/types';
import { EntityDict as BaseEntityDict } from 'oak-domain/lib/base-app-domain';
import { MySQLConfiguration } from 'oak-db/lib/MySQL/types/Configuration';
import { BackendRuntimeContext } from 'oak-frontend-base/lib/context/BackendRuntimeContext';
import { AsyncRowStore, AsyncContext } from 'oak-domain/lib/store/AsyncRowStore';
export declare class DbStore<ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>> extends MysqlStore<ED, Cxt> implements AsyncRowStore<ED, Cxt> {
private executor;
private relationAuth;
constructor(storageSchema: StorageSchema<ED>, contextBuilder: () => Cxt, mysqlConfiguration: MySQLConfiguration, authDeduceRelationMap: AuthDeduceRelationMap<ED>, selectFreeEntities?: SelectFreeEntities<ED>, updateFreeDict?: UpdateFreeDict<ED>, onVolatileTrigger?: <T extends keyof ED>(entity: T, trigger: VolatileTrigger<ED, T, Cxt>, ids: string[], cxtStr: string, option: OperateOption) => Promise<void>);
checkRelationAsync<T extends keyof ED, Cxt extends AsyncContext<ED>>(entity: T, operation: Omit<ED[T]["Operation"] | ED[T]["Selection"], "id">, context: Cxt): Promise<void>;
protected cascadeUpdateAsync<T extends keyof ED>(entity: T, operation: ED[T]['Operation'], context: AsyncContext<ED>, option: MysqlOperateOption): Promise<import("oak-domain/lib/types").OperationResult<ED>>;
operate<T extends keyof ED>(entity: T, operation: ED[T]['Operation'], context: Cxt, option: MysqlOperateOption): Promise<import("oak-domain/lib/types").OperationResult<ED>>;
select<T extends keyof ED>(entity: T, selection: ED[T]['Selection'], context: Cxt, option: MySqlSelectOption): Promise<Partial<ED[T]["Schema"]>[]>;
count<T extends keyof ED>(entity: T, selection: Pick<ED[T]['Selection'], 'filter' | 'count'>, context: Cxt, option: SelectOption): Promise<number>;
import { CascadeStore } from 'oak-domain/lib/store/CascadeStore';
import { DbStore } from 'oak-db/lib/types/dbStore';
export type TriggerStore<ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>> = {
registerTrigger<T extends keyof ED>(trigger: Trigger<ED, T, Cxt>): void;
registerChecker<T extends keyof ED>(checker: Checker<ED, T, Cxt>): void;
setOnVolatileTrigger(onVolatileTrigger: <T extends keyof ED>(entity: T, trigger: VolatileTrigger<ED, T, Cxt>, ids: string[], cxtStr: string, option: OperateOption) => Promise<void>): void;
execVolatileTrigger<T extends keyof ED>(entity: T, name: string, ids: string[], context: Cxt, option: OperateOption): Promise<void>;
checkpoint(ts: number): Promise<number>;
independentCheckPoint(name: string, ts: number, instanceCount?: number, instanceId?: number): Promise<number>;
}
};
export type AppDbStore<ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>> = DbStore<ED, Cxt> & CascadeStore<ED> & TriggerStore<ED, Cxt>;
export declare const createDbStore: <ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>>(storageSchema: StorageSchema<ED>, contextBuilder: () => Cxt, dbConfiguration: DbConfiguration, authDeduceRelationMap: AuthDeduceRelationMap<ED>, selectFreeEntities?: SelectFreeEntities<ED>, updateFreeDict?: UpdateFreeDict<ED>, onVolatileTrigger?: <T extends keyof ED>(entity: T, trigger: VolatileTrigger<ED, T, Cxt>, ids: string[], cxtStr: string, option: OperateOption) => Promise<void>) => AppDbStore<ED, Cxt>;

235
lib/DbStore.js
View File

@ -1,128 +1,133 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.DbStore = void 0;
const oak_db_1 = require("oak-db");
exports.createDbStore = void 0;
const TriggerExecutor_1 = require("oak-domain/lib/store/TriggerExecutor");
const RelationAuth_1 = require("oak-domain/lib/store/RelationAuth");
class DbStore extends oak_db_1.MysqlStore {
executor;
relationAuth;
constructor(storageSchema, contextBuilder, mysqlConfiguration, authDeduceRelationMap, selectFreeEntities = [], updateFreeDict = {}, onVolatileTrigger) {
super(storageSchema, mysqlConfiguration);
this.executor = new TriggerExecutor_1.TriggerExecutor(contextBuilder, undefined, onVolatileTrigger);
this.relationAuth = new RelationAuth_1.RelationAuth(storageSchema, authDeduceRelationMap, selectFreeEntities, updateFreeDict);
}
checkRelationAsync(entity, operation, context) {
return this.relationAuth.checkRelationAsync(entity, operation, context);
}
async cascadeUpdateAsync(entity, operation, context, option) {
// 如果是在modi处理过程中，所有的trigger也可以延时到apply时再处理（这时候因为modi中的数据并不实际存在，处理会有问题）
if (!option.blockTrigger) {
await this.executor.preOperation(entity, operation, context, option);
const dbPriority_1 = require("./utils/dbPriority");
const createDbStore = (storageSchema, contextBuilder, dbConfiguration, authDeduceRelationMap, selectFreeEntities = [], updateFreeDict = {}, onVolatileTrigger) => {
const BaseStoreClass = (0, dbPriority_1.getDbStoreClass)(dbConfiguration);
// 动态创建继承类
class DynamicDbStore extends BaseStoreClass {
executor;
relationAuth;
constructor() {
super(storageSchema, dbConfiguration);
this.executor = new TriggerExecutor_1.TriggerExecutor(contextBuilder, undefined, onVolatileTrigger);
this.relationAuth = new RelationAuth_1.RelationAuth(storageSchema, authDeduceRelationMap, selectFreeEntities, updateFreeDict);
}
const result = await super.cascadeUpdateAsync(entity, operation, context, option);
if (!option.blockTrigger) {
await this.executor.postOperation(entity, operation, context, option);
checkRelationAsync(entity, operation, context) {
return this.relationAuth.checkRelationAsync(entity, operation, context);
}
return result;
}
async operate(entity, operation, context, option) {
const autoCommit = !context.getCurrentTxnId();
let result;
if (autoCommit) {
await context.begin();
}
try {
await this.relationAuth.checkRelationAsync(entity, operation, context);
result = await super.operate(entity, operation, context, option);
}
catch (err) {
await context.rollback();
throw err;
}
if (autoCommit) {
await context.commit();
}
return result;
}
async select(entity, selection, context, option) {
const autoCommit = !context.getCurrentTxnId();
if (autoCommit) {
await context.begin();
}
let result;
// select的trigger应加在根select之前，cascade的select不加处理
Object.assign(selection, {
action: 'select',
});
if (!option.blockTrigger) {
await this.executor.preOperation(entity, selection, context, option);
}
if (!option.dontCollect) {
await this.relationAuth.checkRelationAsync(entity, selection, context);
}
try {
result = await super.select(entity, selection, context, option);
async cascadeUpdateAsync(entity, operation, context, option) {
// 如果是在modi处理过程中，所有的trigger也可以延时到apply时再处理（这时候因为modi中的数据并不实际存在，处理会有问题）
if (!option.blockTrigger) {
await this.executor.postOperation(entity, selection, context, option, result);
await this.executor.preOperation(entity, operation, context, option);
}
}
catch (err) {
await context.rollback();
throw err;
}
if (autoCommit) {
await context.commit();
}
return result;
}
async count(entity, selection, context, option) {
const autoCommit = !context.getCurrentTxnId();
let result;
if (autoCommit) {
await context.begin();
}
try {
// count不用检查权限，因为检查权限中本身要用到count
// const selection2 = Object.assign({
// action: 'select',
// }, selection) as ED[T]['Operation'];
// await this.relationAuth.checkRelationAsync(entity, selection2, context);
// if (!option.blockTrigger) {
// await this.executor.preOperation(entity, selection2, context, option);
// }
result = await super.count(entity, selection, context, option);
/* count应该不存在后trigger吧
const result = await super.cascadeUpdateAsync(entity, operation, context, option);
if (!option.blockTrigger) {
await this.executor.postOperation(entity, selection2, context, option);
} */
await this.executor.postOperation(entity, operation, context, option);
}
return result;
}
catch (err) {
await context.rollback();
throw err;
async operate(entity, operation, context, option) {
const autoCommit = !context.getCurrentTxnId();
let result;
if (autoCommit) {
await context.begin();
}
try {
await this.relationAuth.checkRelationAsync(entity, operation, context);
result = await super.operate(entity, operation, context, option);
}
catch (err) {
await context.rollback();
throw err;
}
if (autoCommit) {
await context.commit();
}
return result;
}
if (autoCommit) {
await context.commit();
async select(entity, selection, context, option) {
const autoCommit = !context.getCurrentTxnId();
if (autoCommit) {
await context.begin();
}
let result;
// select的trigger应加在根select之前，cascade的select不加处理
Object.assign(selection, {
action: 'select',
});
if (!option.blockTrigger) {
await this.executor.preOperation(entity, selection, context, option);
}
if (!option.dontCollect) {
await this.relationAuth.checkRelationAsync(entity, selection, context);
}
try {
result = await super.select(entity, selection, context, option);
if (!option.blockTrigger) {
await this.executor.postOperation(entity, selection, context, option, result);
}
}
catch (err) {
await context.rollback();
throw err;
}
if (autoCommit) {
await context.commit();
}
return result;
}
async count(entity, selection, context, option) {
const autoCommit = !context.getCurrentTxnId();
let result;
if (autoCommit) {
await context.begin();
}
try {
// count不用检查权限，因为检查权限中本身要用到count
// const selection2 = Object.assign({
// action: 'select',
// }, selection) as ED[T]['Operation'];
// await this.relationAuth.checkRelationAsync(entity, selection2, context);
// if (!option.blockTrigger) {
// await this.executor.preOperation(entity, selection2, context, option);
// }
result = await super.count(entity, selection, context, option);
/* count应该不存在后trigger吧
if (!option.blockTrigger) {
await this.executor.postOperation(entity, selection2, context, option);
} */
}
catch (err) {
await context.rollback();
throw err;
}
if (autoCommit) {
await context.commit();
}
return result;
}
registerTrigger(trigger) {
this.executor.registerTrigger(trigger);
}
registerChecker(checker) {
this.executor.registerChecker(checker, this.getSchema());
}
setOnVolatileTrigger(onVolatileTrigger) {
this.executor.setOnVolatileTrigger(onVolatileTrigger);
}
async execVolatileTrigger(entity, name, ids, context, option) {
return this.executor.execVolatileTrigger(entity, name, ids, context, option);
}
checkpoint(ts) {
return this.executor.checkpoint(ts);
}
independentCheckPoint(name, ts, instanceCount, instanceId) {
return this.executor.independentCheckPoint(name, ts, instanceCount, instanceId);
}
return result;
}
registerTrigger(trigger) {
this.executor.registerTrigger(trigger);
}
registerChecker(checker) {
this.executor.registerChecker(checker, this.getSchema());
}
setOnVolatileTrigger(onVolatileTrigger) {
this.executor.setOnVolatileTrigger(onVolatileTrigger);
}
async execVolatileTrigger(entity, name, ids, context, option) {
return this.executor.execVolatileTrigger(entity, name, ids, context, option);
}
checkpoint(ts) {
return this.executor.checkpoint(ts);
}
independentCheckPoint(name, ts, instanceCount, instanceId) {
return this.executor.independentCheckPoint(name, ts, instanceCount, instanceId);
}
}
exports.DbStore = DbStore;
return new DynamicDbStore();
};
exports.createDbStore = createDbStore;

2
lib/Synchronizer.d.ts vendored
View File

@ -27,7 +27,7 @@ export default class Synchronizer<ED extends EntityDict & BaseEntityDict, Cxt ex
* 根据sync的定义，生成对应的 commit triggers
* @returns
*/
getSyncTriggers(): VolatileTrigger<ED, keyof ED, Cxt>[];
getSyncTriggers(): Array<VolatileTrigger<ED, keyof ED, Cxt>>;
getSelfEndpoint(): EndpointItem<ED, Cxt>;
tryCreateSyncProcess(): void;
}

2
lib/Synchronizer.js
View File

@ -114,7 +114,7 @@ class Synchronizer {
remoteEntityId: entityId,
data: queue.map((ele) => ({
entity: ele.oper.targetEntity,
rowIds: ele.oper.filter.id.$in,
rowIds: ele.oper.filter.id.$in, // 暂时应该没什么用
action: ele.oper.action,
data: ele.oper.data,
})),

3
lib/cluster/env.js
View File

@ -1,6 +1,6 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.getClusterInfo = void 0;
exports.getClusterInfo = getClusterInfo;
function getProcessEnvOption(option) {
if (process.env.hasOwnProperty(option)) {
return process.env[option];
@ -54,4 +54,3 @@ const MyClusterInfo = initialize();
function getClusterInfo() {
return MyClusterInfo;
}
exports.getClusterInfo = getClusterInfo;

5
lib/routines/i18n.js
View File

@ -1,6 +1,7 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.checkAndUpdateI18n = exports.checkI18n = void 0;
exports.checkI18n = checkI18n;
exports.checkAndUpdateI18n = checkAndUpdateI18n;
const tslib_1 = require("tslib");
const node_path_1 = require("node:path");
const requirePrj_1 = tslib_1.__importDefault(require("../utils/requirePrj"));
@ -66,7 +67,6 @@ async function checkAndUpdateI18nInner(context, onlyCheck) {
function checkI18n(context) {
return checkAndUpdateI18nInner(context, true);
}
exports.checkI18n = checkI18n;
/**
* 检查项目目录下的i18n数据和数据库中的差异，并更新
* @param context
@ -75,4 +75,3 @@ exports.checkI18n = checkI18n;
function checkAndUpdateI18n(context) {
return checkAndUpdateI18nInner(context);
}
exports.checkAndUpdateI18n = checkAndUpdateI18n;

16
lib/utils/dbPriority.d.ts vendored Normal file
View File

@ -0,0 +1,16 @@
import { MysqlStore, PostgreSQLStore } from "oak-db";
import { DbConfiguration } from "oak-db/src/types/configuration";
import { AsyncRowStore } from "oak-domain/lib/store/AsyncRowStore";
import { EntityDict, StorageSchema } from 'oak-domain/lib/types';
import { EntityDict as BaseEntityDict } from 'oak-domain/lib/base-app-domain';
import { BackendRuntimeContext } from 'oak-frontend-base/lib/context/BackendRuntimeContext';
import { CascadeStore } from "oak-domain/lib/store/CascadeStore";
/**
* 数据库优先级列表，按顺序尝试获取配置文件
*/
export declare const dbList: {
mysql: typeof MysqlStore;
postgres: typeof PostgreSQLStore;
};
export declare const getDbConfig: (path: string) => DbConfiguration;
export declare const getDbStoreClass: <ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>>(dbConfig: DbConfiguration) => new (schema: StorageSchema<ED>, config: DbConfiguration) => AsyncRowStore<ED, Cxt> & CascadeStore<ED>;

37
lib/utils/dbPriority.js Normal file
View File

@ -0,0 +1,37 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.getDbStoreClass = exports.getDbConfig = exports.dbList = void 0;
const oak_db_1 = require("oak-db");
const path_1 = require("path");
/**
* 数据库优先级列表，按顺序尝试获取配置文件
*/
exports.dbList = {
mysql: oak_db_1.MysqlStore,
postgres: oak_db_1.PostgreSQLStore
};
const getDbConfig = (path) => {
for (const db of Object.keys(exports.dbList)) {
try {
// eslint-disable-next-line @typescript-eslint/no-var-requires
const dbConfigFile = (0, path_1.join)(path, 'configuration', `${db}.json`);
const config = require(dbConfigFile);
console.log(`使用${db}作为数据库`);
return Object.assign({}, { type: db }, config);
}
catch (err) {
// do nothing
}
}
throw new Error(`没有找到数据库配置文件，请在configuration目录下添加任一配置文件：${Object.keys(exports.dbList).map(ele => `${ele}.json`).join('、')}`);
};
exports.getDbConfig = getDbConfig;
const getDbStoreClass = (dbConfig) => {
const dbType = dbConfig.type || 'mysql';
const DbStoreClass = exports.dbList[dbType.toLowerCase()];
if (!DbStoreClass) {
throw new Error(`不支持的数据库类型：${dbType}，请确认是否存在以下配置文件：${Object.keys(exports.dbList).map(ele => `${ele}.json`).join('、')}`);
}
return DbStoreClass;
};
exports.getDbStoreClass = getDbStoreClass;

2
lib/utils/requirePrj.js
View File

@ -1,5 +1,6 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.default = requireSth;
const fs_1 = require("fs");
const lodash_1 = require("oak-domain/lib/utils/lodash");
const path_1 = require("path");
@ -20,4 +21,3 @@ function requireSth(prjPath, filePath, dependencies) {
}
return (0, lodash_1.mergeConcatMany)(sthExternal);
}
exports.default = requireSth;

21
src/AppLoader.ts
View File

@ -6,9 +6,7 @@ import { cloneDeep, mergeConcatMany, omit } from 'oak-domain/lib/utils/lodash';
import { EntityDict as BaseEntityDict } from 'oak-domain/lib/base-app-domain';
import { generateNewIdAsync } from 'oak-domain/lib/utils/uuid';
import { AppLoader as GeneralAppLoader, Trigger, Checker, Aspect, CreateOpResult, SyncConfig, EntityDict, Watcher, BBWatcher, WBWatcher, OpRecord, Routine, FreeRoutine, Timer, FreeTimer, StorageSchema, OperationResult, OakPartialSuccess, OakException } from "oak-domain/lib/types";
import { DbStore } from "./DbStore";
import generalAspectDict, { clearPorts, registerPorts } from 'oak-common-aspect/lib/index';
import { MySQLConfiguration } from 'oak-db/lib/MySQL/types/Configuration';
import { BackendRuntimeContext } from 'oak-frontend-base/lib/context/BackendRuntimeContext';
import { Endpoint, EndpointItem } from 'oak-domain/lib/types/Endpoint';
import assert from 'assert';
@ -23,14 +21,16 @@ import { AsyncContext } from 'oak-domain/lib/store/AsyncRowStore';
import domainI18nData from 'oak-domain/lib/data/i18n';
import requireSth from './utils/requirePrj';
import { InternalErrorHandler, InternalErrorType } from './types';
import { getDbConfig } from './utils/dbPriority';
import { createDbStore, AppDbStore } from './DbStore';
export class AppLoader<ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>> extends GeneralAppLoader<ED, Cxt> {
protected dbStore: DbStore<ED, Cxt>;
protected dbStore: AppDbStore<ED, Cxt>;
private aspectDict: Record<string, Aspect<ED, Cxt>>;
private externalDependencies: string[];
protected dataSubscriber?: DataSubscriber<ED, Cxt>;
protected synchronizer?: Synchronizer<ED, Cxt>;
protected contextBuilder: (store: DbStore<ED, Cxt>) => Cxt;
protected contextBuilder: (store: AppDbStore<ED, Cxt>) => Cxt;
private nsSocket?: Namespace;
private watcherTimerId?: NodeJS.Timeout;
private scheduledJobs: Record<string, Job> = {};
@ -109,13 +109,12 @@ export class AppLoader<ED extends EntityDict & BaseEntityDict, Cxt extends Backe
* 后台启动的configuration，统一放在这里读取
*/
private getConfiguration() {
const dbConfigFile = join(this.path, 'configuration', 'mysql.json');
const dbConfig = require(dbConfigFile);
const dbConfig = getDbConfig(this.path);
const syncConfigFile = join(this.path, 'lib', 'configuration', 'sync.js');
const syncConfigs = existsSync(syncConfigFile) && require(syncConfigFile).default;
return {
dbConfig: dbConfig as MySQLConfiguration,
dbConfig: dbConfig,
syncConfig: syncConfigs as SyncConfig<ED, Cxt> | undefined,
};
}
@ -128,7 +127,7 @@ export class AppLoader<ED extends EntityDict & BaseEntityDict, Cxt extends Backe
this.externalDependencies = depGraph.ascOrder;
const { authDeduceRelationMap, selectFreeEntities, updateFreeDict } = this.requireSth('lib/configuration/relation')!;
this.aspectDict = Object.assign({}, generalAspectDict, this.requireSth('lib/aspects/index')!);
this.dbStore = new DbStore<ED, Cxt>(
this.dbStore = createDbStore<ED, Cxt>(
storageSchema,
() => this.contextBuilder(this.dbStore),
dbConfig,
@ -146,7 +145,7 @@ export class AppLoader<ED extends EntityDict & BaseEntityDict, Cxt extends Backe
// 需要重载context上的构造和commit方法，否则程序中执行context.restartToExecute这样的方法中，new一个context出来是无法正确执行的
class BackendRuntimeContextWrapper extends BackendRuntimeContext {
constructor(store: DbStore<ED, Cxt>) {
constructor(store: AppDbStore<ED, Cxt>) {
super(store);
this.clusterInfo = getClusterInfo();
}
@ -354,10 +353,10 @@ export class AppLoader<ED extends EntityDict & BaseEntityDict, Cxt extends Backe
}
}
}
await this.dbStore.disconnect();
// await this.dbStore.disconnect(); // 不需要马上断开连接，在initialize后可能还会有操作，unmount时会断开
}
getStore(): DbStore<ED, Cxt> {
getStore(): AppDbStore<ED, Cxt> {
return this.dbStore;
}

349
src/DbStore.ts
View File

@ -1,151 +1,18 @@
import { MysqlStore, MySqlSelectOption, MysqlOperateOption } from 'oak-db';
import { DbConfiguration } from 'oak-db/src/types/configuration';
import { MySqlSelectOption, MysqlOperateOption } from 'oak-db';
import { EntityDict, StorageSchema, Trigger, Checker, SelectOption, SelectFreeEntities, UpdateFreeDict, AuthDeduceRelationMap, VolatileTrigger, OperateOption } from 'oak-domain/lib/types';
import { EntityDict as BaseEntityDict } from 'oak-domain/lib/base-app-domain';
import { TriggerExecutor } from 'oak-domain/lib/store/TriggerExecutor';
import { MySQLConfiguration, } from 'oak-db/lib/MySQL/types/Configuration';
import { BackendRuntimeContext } from 'oak-frontend-base/lib/context/BackendRuntimeContext';
import { AsyncRowStore, AsyncContext } from 'oak-domain/lib/store/AsyncRowStore';
import { AsyncContext } from 'oak-domain/lib/store/AsyncRowStore';
import { RelationAuth } from 'oak-domain/lib/store/RelationAuth';
import { getDbStoreClass } from './utils/dbPriority';
import { CascadeStore } from 'oak-domain/lib/store/CascadeStore';
import { DbStore } from 'oak-db/lib/types/dbStore';
export class DbStore<ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>> extends MysqlStore<ED, Cxt> implements AsyncRowStore<ED, Cxt> {
private executor: TriggerExecutor<ED, Cxt>;
private relationAuth: RelationAuth<ED>;
constructor(
storageSchema: StorageSchema<ED>,
contextBuilder: () => Cxt,
mysqlConfiguration: MySQLConfiguration,
authDeduceRelationMap: AuthDeduceRelationMap<ED>,
selectFreeEntities: SelectFreeEntities<ED> = [],
updateFreeDict: UpdateFreeDict<ED> = {},
onVolatileTrigger?: <T extends keyof ED>(entity: T, trigger: VolatileTrigger<ED, T, Cxt>, ids: string[], cxtStr: string, option: OperateOption) => Promise<void>) {
super(storageSchema, mysqlConfiguration);
this.executor = new TriggerExecutor(contextBuilder, undefined, onVolatileTrigger);
this.relationAuth = new RelationAuth(storageSchema, authDeduceRelationMap, selectFreeEntities, updateFreeDict);
}
checkRelationAsync<T extends keyof ED, Cxt extends AsyncContext<ED>>(entity: T, operation: Omit<ED[T]["Operation"] | ED[T]["Selection"], "id">, context: Cxt): Promise<void> {
return this.relationAuth.checkRelationAsync(entity, operation, context);
}
protected async cascadeUpdateAsync<T extends keyof ED>(entity: T, operation: ED[T]['Operation'], context: AsyncContext<ED>, option: MysqlOperateOption) {
// 如果是在modi处理过程中，所有的trigger也可以延时到apply时再处理（这时候因为modi中的数据并不实际存在，处理会有问题）
if (!option.blockTrigger) {
await this.executor.preOperation(entity, operation, context as Cxt, option);
}
const result = await super.cascadeUpdateAsync(entity, operation, context, option);
if (!option.blockTrigger) {
await this.executor.postOperation(entity, operation, context as Cxt, option);
}
return result;
}
async operate<T extends keyof ED>(
entity: T,
operation: ED[T]['Operation'],
context: Cxt,
option: MysqlOperateOption
) {
const autoCommit = !context.getCurrentTxnId();
let result;
if (autoCommit) {
await context.begin();
}
try {
await this.relationAuth.checkRelationAsync(entity, operation, context);
result = await super.operate(entity, operation, context, option);
}
catch (err) {
await context.rollback();
throw err;
}
if (autoCommit) {
await context.commit();
}
return result;
}
async select<T extends keyof ED>(
entity: T,
selection: ED[T]['Selection'],
context: Cxt,
option: MySqlSelectOption
) {
const autoCommit = !context.getCurrentTxnId();
if (autoCommit) {
await context.begin();
}
let result: Partial<ED[T]['Schema']>[];
// select的trigger应加在根select之前，cascade的select不加处理
Object.assign(selection, {
action: 'select',
});
if (!option.blockTrigger) {
await this.executor.preOperation(entity, selection as ED[T]['Operation'], context, option);
}
if (!option.dontCollect) {
await this.relationAuth.checkRelationAsync(entity, selection, context);
}
try {
result = await super.select(entity, selection, context, option);
if (!option.blockTrigger) {
await this.executor.postOperation(entity, selection as ED[T]['Operation']
, context, option, result);
}
}
catch (err) {
await context.rollback();
throw err;
}
if (autoCommit) {
await context.commit();
}
return result;
}
async count<T extends keyof ED>(entity: T, selection: Pick<ED[T]['Selection'], 'filter' | 'count'>, context: Cxt, option: SelectOption): Promise<number> {
const autoCommit = !context.getCurrentTxnId();
let result;
if (autoCommit) {
await context.begin();
}
try {
// count不用检查权限，因为检查权限中本身要用到count
// const selection2 = Object.assign({
// action: 'select',
// }, selection) as ED[T]['Operation'];
// await this.relationAuth.checkRelationAsync(entity, selection2, context);
// if (!option.blockTrigger) {
// await this.executor.preOperation(entity, selection2, context, option);
// }
result = await super.count(entity, selection, context, option);
/* count应该不存在后trigger吧
if (!option.blockTrigger) {
await this.executor.postOperation(entity, selection2, context, option);
} */
}
catch (err) {
await context.rollback();
throw err;
}
if (autoCommit) {
await context.commit();
}
return result;
}
registerTrigger<T extends keyof ED>(trigger: Trigger<ED, T, Cxt>) {
this.executor.registerTrigger(trigger);
}
registerChecker<T extends keyof ED>(checker: Checker<ED, T, Cxt>) {
this.executor.registerChecker(checker, this.getSchema());
}
export type TriggerStore<ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>> = {
registerTrigger<T extends keyof ED>(trigger: Trigger<ED, T, Cxt>): void;
registerChecker<T extends keyof ED>(checker: Checker<ED, T, Cxt>): void;
setOnVolatileTrigger(
onVolatileTrigger: <T extends keyof ED>(
entity: T,
@ -153,25 +20,193 @@ export class DbStore<ED extends EntityDict & BaseEntityDict, Cxt extends Backend
ids: string[],
cxtStr: string,
option: OperateOption) => Promise<void>
) {
this.executor.setOnVolatileTrigger(onVolatileTrigger);
}
async execVolatileTrigger<T extends keyof ED>(
): void;
execVolatileTrigger<T extends keyof ED>(
entity: T,
name: string,
ids: string[],
context: Cxt,
option: OperateOption
) {
return this.executor.execVolatileTrigger(entity, name, ids, context, option);
): Promise<void>;
checkpoint(ts: number): Promise<number>;
independentCheckPoint(name: string, ts: number, instanceCount?: number, instanceId?: number): Promise<number>;
}
export type AppDbStore<ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>> = DbStore<ED, Cxt> & CascadeStore<ED> & TriggerStore<ED, Cxt>;
export const createDbStore = <ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>>(
storageSchema: StorageSchema<ED>,
contextBuilder: () => Cxt,
dbConfiguration: DbConfiguration,
authDeduceRelationMap: AuthDeduceRelationMap<ED>,
selectFreeEntities: SelectFreeEntities<ED> = [],
updateFreeDict: UpdateFreeDict<ED> = {},
onVolatileTrigger?: <T extends keyof ED>(entity: T, trigger: VolatileTrigger<ED, T, Cxt>, ids: string[], cxtStr: string, option: OperateOption) => Promise<void>
): AppDbStore<ED, Cxt> => {
const BaseStoreClass = getDbStoreClass<ED, Cxt>(dbConfiguration) as any
// 动态创建继承类
class DynamicDbStore extends BaseStoreClass implements TriggerStore<ED, Cxt> {
private executor: TriggerExecutor<ED, Cxt>;
private relationAuth: RelationAuth<ED>;
constructor() {
super(storageSchema, dbConfiguration);
this.executor = new TriggerExecutor(contextBuilder, undefined, onVolatileTrigger);
this.relationAuth = new RelationAuth(storageSchema, authDeduceRelationMap, selectFreeEntities, updateFreeDict);
}
checkRelationAsync<T extends keyof ED, Cxt extends AsyncContext<ED>>(entity: T, operation: Omit<ED[T]["Operation"] | ED[T]["Selection"], "id">, context: Cxt): Promise<void> {
return this.relationAuth.checkRelationAsync(entity, operation, context);
}
protected async cascadeUpdateAsync<T extends keyof ED>(entity: T, operation: ED[T]['Operation'], context: AsyncContext<ED>, option: MysqlOperateOption) {
// 如果是在modi处理过程中，所有的trigger也可以延时到apply时再处理（这时候因为modi中的数据并不实际存在，处理会有问题）
if (!option.blockTrigger) {
await this.executor.preOperation(entity, operation, context as Cxt, option);
}
const result = await super.cascadeUpdateAsync(entity, operation, context, option);
if (!option.blockTrigger) {
await this.executor.postOperation(entity, operation, context as Cxt, option);
}
return result;
}
async operate<T extends keyof ED>(
entity: T,
operation: ED[T]['Operation'],
context: Cxt,
option: MysqlOperateOption
) {
const autoCommit = !context.getCurrentTxnId();
let result;
if (autoCommit) {
await context.begin();
}
try {
await this.relationAuth.checkRelationAsync(entity, operation, context);
result = await super.operate(entity, operation, context, option);
}
catch (err) {
await context.rollback();
throw err;
}
if (autoCommit) {
await context.commit();
}
return result;
}
async select<T extends keyof ED>(
entity: T,
selection: ED[T]['Selection'],
context: Cxt,
option: MySqlSelectOption
) {
const autoCommit = !context.getCurrentTxnId();
if (autoCommit) {
await context.begin();
}
let result: Partial<ED[T]['Schema']>[];
// select的trigger应加在根select之前，cascade的select不加处理
Object.assign(selection, {
action: 'select',
});
if (!option.blockTrigger) {
await this.executor.preOperation(entity, selection as ED[T]['Operation'], context, option);
}
if (!option.dontCollect) {
await this.relationAuth.checkRelationAsync(entity, selection, context);
}
try {
result = await super.select(entity, selection, context, option);
if (!option.blockTrigger) {
await this.executor.postOperation(entity, selection as ED[T]['Operation']
, context, option, result);
}
}
catch (err) {
await context.rollback();
throw err;
}
if (autoCommit) {
await context.commit();
}
return result;
}
async count<T extends keyof ED>(entity: T, selection: Pick<ED[T]['Selection'], 'filter' | 'count'>, context: Cxt, option: SelectOption): Promise<number> {
const autoCommit = !context.getCurrentTxnId();
let result;
if (autoCommit) {
await context.begin();
}
try {
// count不用检查权限，因为检查权限中本身要用到count
// const selection2 = Object.assign({
// action: 'select',
// }, selection) as ED[T]['Operation'];
// await this.relationAuth.checkRelationAsync(entity, selection2, context);
// if (!option.blockTrigger) {
// await this.executor.preOperation(entity, selection2, context, option);
// }
result = await super.count(entity, selection, context, option);
/* count应该不存在后trigger吧
if (!option.blockTrigger) {
await this.executor.postOperation(entity, selection2, context, option);
} */
}
catch (err) {
await context.rollback();
throw err;
}
if (autoCommit) {
await context.commit();
}
return result;
}
registerTrigger<T extends keyof ED>(trigger: Trigger<ED, T, Cxt>) {
this.executor.registerTrigger(trigger);
}
registerChecker<T extends keyof ED>(checker: Checker<ED, T, Cxt>) {
this.executor.registerChecker(checker, this.getSchema());
}
setOnVolatileTrigger(
onVolatileTrigger: <T extends keyof ED>(
entity: T,
trigger: VolatileTrigger<ED, T, Cxt>,
ids: string[],
cxtStr: string,
option: OperateOption) => Promise<void>
) {
this.executor.setOnVolatileTrigger(onVolatileTrigger);
}
async execVolatileTrigger<T extends keyof ED>(
entity: T,
name: string,
ids: string[],
context: Cxt,
option: OperateOption
) {
return this.executor.execVolatileTrigger(entity, name, ids, context, option);
}
checkpoint(ts: number) {
return this.executor.checkpoint(ts);
}
independentCheckPoint(name: string, ts: number, instanceCount?: number, instanceId?: number) {
return this.executor.independentCheckPoint(name, ts, instanceCount, instanceId);
}
}
checkpoint(ts: number) {
return this.executor.checkpoint(ts);
}
independentCheckPoint(name: string, ts: number, instanceCount?: number, instanceId?: number) {
return this.executor.independentCheckPoint(name, ts, instanceCount, instanceId);
}
}
return new DynamicDbStore() as any as AppDbStore<ED, Cxt>;
};

43
src/utils/dbPriority.ts Normal file
View File

@ -0,0 +1,43 @@
import { MysqlStore, PostgreSQLStore } from "oak-db";
import { DbConfiguration } from "oak-db/src/types/configuration";
import { AsyncContext, AsyncRowStore } from "oak-domain/lib/store/AsyncRowStore";
import { join } from "path";
import { EntityDict, StorageSchema, Trigger, Checker, SelectOption, SelectFreeEntities, UpdateFreeDict, AuthDeduceRelationMap, VolatileTrigger, OperateOption } from 'oak-domain/lib/types';
import { EntityDict as BaseEntityDict } from 'oak-domain/lib/base-app-domain';
import { BackendRuntimeContext } from 'oak-frontend-base/lib/context/BackendRuntimeContext';
import { CascadeStore } from "oak-domain/lib/store/CascadeStore";
/**
* 数据库优先级列表，按顺序尝试获取配置文件
*/
export const dbList = {
mysql: MysqlStore,
postgres: PostgreSQLStore
}
export const getDbConfig = (path: string): DbConfiguration => {
for (const db of Object.keys(dbList)) {
try {
// eslint-disable-next-line @typescript-eslint/no-var-requires
const dbConfigFile = join(path, 'configuration', `${db}.json`);
const config = require(dbConfigFile);
console.log(`使用${db}作为数据库`);
return Object.assign({}, { type: db }, config);
}
catch (err) {
// do nothing
}
}
throw new Error(`没有找到数据库配置文件，请在configuration目录下添加任一配置文件：${Object.keys(dbList).map(ele => `${ele}.json`).join('、')}`);
}
export const getDbStoreClass = <ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>>
(dbConfig: DbConfiguration) => {
const dbType = dbConfig.type || 'mysql';
const DbStoreClass = dbList[dbType.toLowerCase() as keyof typeof dbList];
if (!DbStoreClass) {
throw new Error(`不支持的数据库类型：${dbType}，请确认是否存在以下配置文件：${Object.keys(dbList).map(ele => `${ele}.json`).join('、')}`);
}
return DbStoreClass as new (schema: StorageSchema<ED>, config: DbConfiguration) => AsyncRowStore<ED, Cxt> & CascadeStore<ED>;
}