Oak-Team
/
oak-domain
13
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 33 Wiki Activity

triggerExecutor增加了unregisterTrigger的能力

This commit is contained in:
Xu Chang 2022-04-17 15:45:23 +08:00
parent 26aa84bd45
commit c6697b0ef6
18 changed files with 1074 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
lib/store/TriggerExecutor.d.ts vendored Normal file
View File

@ -0,0 +1,21 @@
import { EntityDict } from "../types/Entity";
import { Logger } from "../types/Logger";
import { Checker } from '../types/Auth';
import { Context } from '../types/Context';
import { Trigger, Executor } from "../types/Trigger";
export declare class TriggerExecutor<ED extends EntityDict> extends Executor<ED> {
private triggerMap;
private triggerNameMap;
private volatileEntities;
private logger;
constructor(logger?: Logger);
registerChecker<T extends keyof ED>(checker: Checker<ED, T>): void;
registerTrigger<T extends keyof ED>(trigger: Trigger<ED, T>): void;
unregisterTrigger<T extends keyof ED>(trigger: Trigger<ED, T>): void;
private preCommitTrigger;
preOperation<T extends keyof ED>(entity: T, operation: ED[T]['Operation'], context: Context<ED>): Promise<void>;
private onCommit;
private postCommitTrigger;
postOperation<T extends keyof ED>(entity: T, operation: ED[T]['Operation'], context: Context<ED>): Promise<void>;
checkpoint(context: Context<ED>, timestamp: number): Promise<number>;
}

237
lib/store/TriggerExecutor.js Normal file
View File

@ -0,0 +1,237 @@
"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.TriggerExecutor = void 0;
const assert_1 = __importDefault(require("assert"));
const lodash_1 = require("lodash");
const filter_1 = require("../store/filter");
const Trigger_1 = require("../types/Trigger");
/**
* update可能会传入多种不同的action，此时都需要检查update trigger
*/
const UnifiedActionMatrix = {
'create': 'create',
'remove': 'remove',
'select': 'select',
'download': 'select',
'count': 'select',
'stat': 'select',
};
class TriggerExecutor extends Trigger_1.Executor {
triggerMap;
triggerNameMap;
volatileEntities;
logger;
constructor(logger = console) {
super();
this.logger = logger;
this.triggerMap = {};
this.triggerNameMap = {};
this.volatileEntities = [];
}
registerChecker(checker) {
const { entity, action, checker: checkFn } = checker;
const ActionNameMatrix = {
'create': '创建',
'remove': '删除',
};
let triggerAction = ActionNameMatrix[action] || '更新';
const triggerName = `${entity}${triggerAction}权限检查`;
const trigger = {
name: triggerName,
entity,
action,
fn: checkFn,
when: 'before',
};
this.registerTrigger(trigger);
}
registerTrigger(trigger) {
// trigger的两种访问方式: by name, by entity/action
if (this.triggerNameMap.hasOwnProperty(trigger.name)) {
throw new Error(`不可有同名的触发器「${trigger.name}」`);
}
(0, lodash_1.assign)(this.triggerNameMap, {
[trigger.name]: trigger,
});
const action = UnifiedActionMatrix[trigger.action] || 'update';
const triggers = this.triggerMap[trigger.entity] && this.triggerMap[trigger.entity][action];
if (triggers) {
triggers.push(trigger);
}
else if (this.triggerMap[trigger.entity]) {
(0, lodash_1.assign)(this.triggerMap[trigger.entity], {
[action]: [trigger],
});
}
else {
(0, lodash_1.assign)(this.triggerMap, {
[trigger.entity]: {
[action]: [trigger],
}
});
}
if (trigger.when === 'commit' && trigger.strict === 'makeSure') {
if (this.volatileEntities.indexOf(trigger.entity) === -1) {
this.volatileEntities.push(trigger.entity);
}
}
}
unregisterTrigger(trigger) {
(0, assert_1.default)(trigger.when !== 'commit' || trigger.strict !== 'makeSure');
const action = UnifiedActionMatrix[trigger.action] || 'update';
const triggers = this.triggerMap[trigger.entity] && this.triggerMap[trigger.entity][action];
if (triggers) {
(0, lodash_1.pull)(triggers, trigger);
}
}
async preCommitTrigger(entity, operation, trigger, context) {
(0, assert_1.default)(trigger.action !== 'select');
if (trigger.strict === 'makeSure') {
switch (operation.action) {
case 'create': {
if (operation.data.hasOwnProperty(Trigger_1.Executor.dataAttr) || operation.data.hasOwnProperty(Trigger_1.Executor.timestampAttr)) {
throw new Error('同一行数据上不能存在两个跨事务约束');
}
break;
}
default: {
const { filter } = operation;
// 此时要保证更新或者删除的行上没有跨事务约束
const filter2 = (0, filter_1.addFilterSegment)({
$or: [
{
$$triggerData$$: {
$exists: true,
},
},
{
$$triggerTimestamp$$: {
$exists: true,
},
}
],
}, filter);
const { rowStore } = context;
const count = await rowStore.count(entity, {
filter: filter2
}, context);
if (count > 0) {
throw new Error(`对象${entity}的行「${JSON.stringify(operation)}」上已经存在未完成的跨事务约束`);
}
break;
}
}
(0, lodash_1.assign)(operation.data, {
[Trigger_1.Executor.dataAttr]: {
name: trigger.name,
operation,
},
[Trigger_1.Executor.timestampAttr]: Date.now(),
});
}
}
async preOperation(entity, operation, context) {
const action = UnifiedActionMatrix[operation.action] || 'update';
const triggers = this.triggerMap[entity] && this.triggerMap[entity][action];
if (triggers) {
const preTriggers = triggers.filter(ele => ele.when === 'before' && (!ele.check || ele.check(operation)));
for (const trigger of preTriggers) {
const number = await trigger.fn({ operation: operation }, context);
if (number > 0) {
this.logger.info(`触发器「${trigger.name}」成功触发了「${number}」行数据更改`);
}
}
const commitTriggers = triggers.filter(ele => ele.when === 'commit' && (!ele.check || ele.check(operation)));
for (const trigger of commitTriggers) {
await this.preCommitTrigger(entity, operation, trigger, context);
}
}
}
onCommit(trigger, operation) {
return async (context) => {
await context.begin();
const number = await trigger.fn({
operation: operation,
}, context);
const { rowStore } = context;
if (trigger.strict === 'makeSure') {
// 如果是必须完成的trigger，在完成成功后要把trigger相关的属性置null;
let filter = {};
if (operation.action === 'create') {
filter = operation.data instanceof Array ? {
filter: {
id: {
$in: operation.data.map(ele => ele.id),
},
},
} : {
filter: {
id: operation.data.id,
}
};
}
else if (operation.filter) {
(0, lodash_1.assign)(filter, { filter: operation.filter });
}
await rowStore.operate(trigger.entity, {
action: 'update',
data: {
$$triggerTimestamp$$: null,
$$triggerData$$: null,
},
...filter /** as Filter<'update', DeduceFilter<ED[T]['Schema']>> */,
}, context);
}
await context.commit();
return;
};
}
async postCommitTrigger(operation, trigger, context) {
context.on('commit', this.onCommit(trigger, operation));
}
async postOperation(entity, operation, context) {
const action = UnifiedActionMatrix[operation.action] || 'update';
const triggers = this.triggerMap[entity] && this.triggerMap[entity][action];
if (triggers) {
const postTriggers = triggers.filter(ele => ele.when === 'after' && (!ele.check || ele.check(operation)));
for (const trigger of postTriggers) {
const number = await trigger.fn({ operation: operation }, context);
if (number > 0) {
this.logger.info(`触发器「${trigger.name}」成功触发了「${number}」行数据更改`);
}
}
const commitTriggers = triggers.filter(ele => ele.when === 'commit' && (!ele.check || ele.check(operation)));
for (const trigger of commitTriggers) {
await this.postCommitTrigger(operation, trigger, context);
}
}
}
async checkpoint(context, timestamp) {
let result = 0;
const { rowStore } = context;
for (const entity of this.volatileEntities) {
const { result: rows } = await rowStore.select(entity, {
data: {
id: 1,
$$triggerData$$: 1,
},
filter: {
$$triggerTimestamp$$: {
$gt: timestamp,
}
},
}, context);
for (const row of rows) {
const { $$triggerData$$ } = row;
const { name, operation } = $$triggerData$$;
const trigger = this.triggerNameMap[name];
await this.onCommit(trigger, operation)(context);
}
}
return result;
}
}
exports.TriggerExecutor = TriggerExecutor;

17
lib/store/UniversalContext.d.ts vendored Normal file
View File

@ -0,0 +1,17 @@
import { EntityDict, OpRecord, RowStore, TxnOption, Context } from "../types";
export declare class UniversalContext<ED extends EntityDict> implements Context<ED> {
rowStore: RowStore<ED>;
uuid?: string;
opRecords: OpRecord<ED>[];
events: {
commit: Array<(context: UniversalContext<ED>) => Promise<void>>;
rollback: Array<(context: UniversalContext<ED>) => Promise<void>>;
};
constructor(store: RowStore<ED>);
private resetEvents;
on(event: 'commit' | 'rollback', callback: (context: UniversalContext<ED>) => Promise<void>): void;
begin(options?: TxnOption): Promise<void>;
commit(): Promise<void>;
rollback(): Promise<void>;
getCurrentTxnId(): string | undefined;
}

61
lib/store/UniversalContext.js Normal file
View File

@ -0,0 +1,61 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.UniversalContext = void 0;
class UniversalContext {
rowStore;
uuid;
opRecords;
events;
constructor(store) {
this.rowStore = store;
this.opRecords = [];
this.events = {
commit: [],
rollback: [],
};
}
resetEvents() {
this.events = {
commit: [],
rollback: [],
};
}
on(event, callback) {
this.uuid && this.events[event].push(callback);
}
async begin(options) {
if (!this.uuid) {
this.uuid = await this.rowStore.begin(options);
}
}
async commit() {
if (this.uuid) {
/**
* todo 这里应该等到提交成功了再做 by Xc
*/
for (const e of this.events.commit) {
await e(this);
}
await this.rowStore.commit(this.uuid);
this.uuid = undefined;
this.resetEvents();
}
}
async rollback() {
if (this.uuid) {
/**
* todo 这里应该等到回滚成功了再做 by Xc
*/
for (const e of this.events.rollback) {
await e(this);
}
await this.rowStore.rollback(this.uuid);
this.uuid = undefined;
this.resetEvents();
}
}
getCurrentTxnId() {
return this.uuid;
}
}
exports.UniversalContext = UniversalContext;

5
lib/types/Aspect.d.ts vendored Normal file
View File

@ -0,0 +1,5 @@
import { EntityDict } from "./Entity";
import { Context } from './Context';
export interface Aspect<ED extends EntityDict, Cxt extends Context<ED>> {
(params: any, context: Cxt): Promise<any>;
}

3
lib/types/Aspect.js Normal file
View File

@ -0,0 +1,3 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
;

23
lib/types/Auth.d.ts vendored Normal file
View File

@ -0,0 +1,23 @@
import { EntityDict } from "../types/Entity";
import { CreateTriggerBase, RemoveTriggerBase, UpdateTriggerBase } from "./Trigger";
export declare class AttrIllegalError extends Error {
private attributes;
constructor(attributes: string[], message?: string);
getAttributes(): string[];
}
export declare type CreateChecker<ED extends EntityDict, T extends keyof ED> = {
action: 'create';
entity: T;
checker: CreateTriggerBase<ED, T>['fn'];
};
export declare type UpdateChecker<ED extends EntityDict, T extends keyof ED> = {
action: UpdateTriggerBase<ED, T>['action'];
entity: T;
checker: UpdateTriggerBase<ED, T>['fn'];
};
export declare type RemoveChecker<ED extends EntityDict, T extends keyof ED> = {
action: 'remove';
entity: T;
checker: RemoveTriggerBase<ED, T>['fn'];
};
export declare type Checker<ED extends EntityDict, T extends keyof ED> = CreateChecker<ED, T> | UpdateChecker<ED, T> | RemoveChecker<ED, T>;

15
lib/types/Auth.js Normal file
View File

@ -0,0 +1,15 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.AttrIllegalError = void 0;
class AttrIllegalError extends Error {
attributes;
constructor(attributes, message) {
super(message);
this.attributes = attributes;
}
getAttributes() {
return this.attributes;
}
}
exports.AttrIllegalError = AttrIllegalError;
;

95
lib/types/Trigger.d.ts vendored Normal file
View File

@ -0,0 +1,95 @@
import { GenericAction } from "../actions/action";
import { DeduceCreateOperation, DeduceRemoveOperation, DeduceSelection, DeduceUpdateOperation, EntityDict } from "../types/Entity";
import { EntityShape, SelectionResult2, TriggerDataAttribute, TriggerTimestampAttribute } from "../types/Entity";
import { Context } from "./Context";
export interface CreateTriggerBase<ED extends EntityDict, T extends keyof ED> {
entity: T;
name: string;
action: 'create';
check?: (operation: DeduceCreateOperation<ED[T]['Schema']>) => boolean;
fn: (event: {
operation: DeduceCreateOperation<ED[T]['Schema']>;
}, context: Context<ED>, params?: Object) => Promise<number>;
}
export interface CreateTriggerInTxn<ED extends EntityDict, T extends keyof ED> extends CreateTriggerBase<ED, T> {
when: 'before' | 'after';
}
export interface CreateTriggerCrossTxn<ED extends EntityDict, T extends keyof ED> extends CreateTriggerBase<ED, T> {
when: 'commit';
strict?: 'takeEasy' | 'makeSure';
}
export declare type CreateTrigger<ED extends EntityDict, T extends keyof ED> = CreateTriggerInTxn<ED, T> | CreateTriggerCrossTxn<ED, T>;
export interface UpdateTriggerBase<ED extends EntityDict, T extends keyof ED> {
entity: T;
name: string;
action: Exclude<ED[T]['Action'], GenericAction> | 'update';
attributes?: keyof ED[T]['OpSchema'] | Array<keyof ED[T]['OpSchema']>;
check?: (operation: DeduceUpdateOperation<ED[T]['Schema']>) => boolean;
fn: (event: {
operation: DeduceUpdateOperation<ED[T]['Schema']>;
}, context: Context<ED>, params?: Object) => Promise<number>;
}
export interface UpdateTriggerInTxn<ED extends EntityDict, T extends keyof ED> extends UpdateTriggerBase<ED, T> {
when: 'before' | 'after';
}
export interface UpdateTriggerCrossTxn<ED extends EntityDict, T extends keyof ED> extends UpdateTriggerBase<ED, T> {
when: 'commit';
strict?: 'takeEasy' | 'makeSure';
}
export declare type UpdateTrigger<ED extends EntityDict, T extends keyof ED> = UpdateTriggerInTxn<ED, T> | UpdateTriggerCrossTxn<ED, T>;
export interface RemoveTriggerBase<ED extends EntityDict, T extends keyof ED> {
entity: T;
name: string;
action: 'remove';
check?: (operation: DeduceRemoveOperation<ED[T]['Schema']>) => boolean;
fn: (event: {
operation: DeduceRemoveOperation<ED[T]['Schema']>;
}, context: Context<ED>, params?: Object) => Promise<number>;
}
export interface RemoveTriggerInTxn<ED extends EntityDict, T extends keyof ED> extends RemoveTriggerBase<ED, T> {
when: 'before' | 'after';
}
export interface RemoveTriggerCrossTxn<ED extends EntityDict, T extends keyof ED> extends RemoveTriggerBase<ED, T> {
when: 'commit';
strict?: 'takeEasy' | 'makeSure';
}
export declare type RemoveTrigger<ED extends EntityDict, T extends keyof ED> = RemoveTriggerInTxn<ED, T> | RemoveTriggerCrossTxn<ED, T>;
export interface SelectTriggerBase<ED extends EntityDict, T extends keyof ED> {
entity: T;
name: string;
action: 'select';
}
/**
* selection似乎不需要支持跨事务？没想清楚
* todo by Xc
*/
export interface SelectTriggerBefore<ED extends EntityDict, T extends keyof ED> extends SelectTriggerBase<ED, T> {
when: 'before';
fn: (event: {
operation: DeduceSelection<ED[T]['Schema']>;
}, context: Context<ED>, params?: Object) => Promise<number>;
}
export interface SelectTriggerAfter<ED extends EntityDict, T extends keyof ED> extends SelectTriggerBase<ED, T> {
when: 'after';
fn: <S extends ED[T]['Selection']>(event: {
operation: S;
result: SelectionResult2<ED[T]['Schema'], S['data']>;
}, context: Context<ED>, params?: Object) => Promise<number>;
}
export declare type SelectTrigger<ED extends EntityDict, T extends keyof ED> = SelectTriggerBefore<ED, T> | SelectTriggerAfter<ED, T>;
export declare type Trigger<ED extends EntityDict, T extends keyof ED> = CreateTrigger<ED, T> | UpdateTrigger<ED, T> | RemoveTrigger<ED, T> | SelectTrigger<ED, T>;
export interface TriggerEntityShape extends EntityShape {
$$triggerData$$?: {
name: string;
operation: object;
};
$$triggerTimestamp$$?: number;
}
export declare abstract class Executor<ED extends EntityDict> {
static dataAttr: TriggerDataAttribute;
static timestampAttr: TriggerTimestampAttribute;
abstract registerTrigger<T extends keyof ED>(trigger: Trigger<ED, T>): void;
abstract preOperation<T extends keyof ED>(entity: T, operation: ED[T]['Operation'], context: Context<ED>): Promise<void>;
abstract postOperation<T extends keyof ED>(entity: T, operation: ED[T]['Operation'], context: Context<ED>): Promise<void>;
abstract checkpoint(context: Context<ED>, timestamp: number): Promise<number>;
}

21
lib/types/Trigger.js Normal file
View File

@ -0,0 +1,21 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.Executor = void 0;
;
;
;
;
;
;
;
;
;
;
;
;
;
class Executor {
static dataAttr = '$$triggerData$$';
static timestampAttr = '$$triggerTimestamp$$';
}
exports.Executor = Executor;

14
lib/types/index.d.ts vendored Normal file
View File

@ -0,0 +1,14 @@
export * from './Action';
export * from './Aspect';
export * from './Auth';
export * from './Context';
export * from './DataType';
export * from './Demand';
export * from './Entity';
export * from './Expression';
export * from './Geo';
export * from './Logger';
export * from './Polyfill';
export * from './RowStore';
export * from './Storage';
export * from './Trigger';

26
lib/types/index.js Normal file
View File

@ -0,0 +1,26 @@
"use strict";
var __createBinding = (this && this.__createBinding) || (Object.create ? (function(o, m, k, k2) {
if (k2 === undefined) k2 = k;
Object.defineProperty(o, k2, { enumerable: true, get: function() { return m[k]; } });
}) : (function(o, m, k, k2) {
if (k2 === undefined) k2 = k;
o[k2] = m[k];
}));
var __exportStar = (this && this.__exportStar) || function(m, exports) {
for (var p in m) if (p !== "default" && !Object.prototype.hasOwnProperty.call(exports, p)) __createBinding(exports, m, p);
};
Object.defineProperty(exports, "__esModule", { value: true });
__exportStar(require("./Action"), exports);
__exportStar(require("./Aspect"), exports);
__exportStar(require("./Auth"), exports);
__exportStar(require("./Context"), exports);
__exportStar(require("./DataType"), exports);
__exportStar(require("./Demand"), exports);
__exportStar(require("./Entity"), exports);
__exportStar(require("./Expression"), exports);
__exportStar(require("./Geo"), exports);
__exportStar(require("./Logger"), exports);
__exportStar(require("./Polyfill"), exports);
__exportStar(require("./RowStore"), exports);
__exportStar(require("./Storage"), exports);
__exportStar(require("./Trigger"), exports);

293
src/store/TriggerExecutor.ts Normal file
View File

@ -0,0 +1,293 @@
import assert from 'assert';
import { assign, pull } from "lodash";
import { addFilterSegment } from "../store/filter";
import { DeduceCreateOperation, DeduceCreateOperationData, EntityDict } from "../types/Entity";
import { Logger } from "../types/Logger";
import { Checker } from '../types/Auth';
import { Context } from '../types/Context';
import { Trigger, Executor, CreateTriggerCrossTxn, CreateTrigger, CreateTriggerInTxn } from "../types/Trigger";
/**
* update可能会传入多种不同的action，此时都需要检查update trigger
*/
const UnifiedActionMatrix: Record<string, string> = {
'create': 'create',
'remove': 'remove',
'select': 'select',
'download': 'select',
'count': 'select',
'stat': 'select',
};
export class TriggerExecutor<ED extends EntityDict> extends Executor<ED> {
private triggerMap: {
[T in keyof ED]?: {
[A: string]: Array<Trigger<ED, T>>;
};
};
private triggerNameMap: {
[N: string]: Trigger<ED, keyof ED>;
};
private volatileEntities: Array<keyof ED>;
private logger: Logger;
constructor(logger: Logger = console) {
super();
this.logger = logger;
this.triggerMap = {};
this.triggerNameMap = {};
this.volatileEntities = [];
}
registerChecker<T extends keyof ED>(checker: Checker<ED, T>): void {
const { entity, action, checker: checkFn } = checker;
const ActionNameMatrix: Record<string, string> = {
'create': '创建',
'remove': '删除',
};
let triggerAction = ActionNameMatrix[action] || '更新';
const triggerName = `${entity}${triggerAction}权限检查`;
const trigger = {
name: triggerName,
entity,
action,
fn: checkFn,
when: 'before',
} as CreateTriggerInTxn<ED, T>;
this.registerTrigger(trigger);
}
registerTrigger<T extends keyof ED>(trigger: Trigger<ED, T>): void {
// trigger的两种访问方式: by name, by entity/action
if (this.triggerNameMap.hasOwnProperty(trigger.name)) {
throw new Error(`不可有同名的触发器「${trigger.name}」`);
}
assign(this.triggerNameMap, {
[trigger.name]: trigger,
});
const action = UnifiedActionMatrix[trigger.action] || 'update';
const triggers = this.triggerMap[trigger.entity] && this.triggerMap[trigger.entity]![action];
if (triggers) {
triggers.push(trigger);
}
else if (this.triggerMap[trigger.entity]) {
assign(this.triggerMap[trigger.entity], {
[action]: [trigger],
});
}
else {
assign(this.triggerMap, {
[trigger.entity]: {
[action]: [trigger],
}
});
}
if (trigger.when === 'commit' && trigger.strict === 'makeSure') {
if (this.volatileEntities.indexOf(trigger.entity) === -1) {
this.volatileEntities.push(trigger.entity);
}
}
}
unregisterTrigger<T extends keyof ED>(trigger: Trigger<ED, T>): void {
assert(trigger.when !== 'commit' || trigger.strict !== 'makeSure');
const action = UnifiedActionMatrix[trigger.action] || 'update';
const triggers = this.triggerMap[trigger.entity] && this.triggerMap[trigger.entity]![action];
if (triggers) {
pull(triggers!, trigger);
}
}
private async preCommitTrigger<T extends keyof ED>(
entity: T,
operation: ED[T]['Operation'],
trigger: Trigger<ED, T>,
context: Context<ED>,
) {
assert(trigger.action !== 'select');
if ((trigger as CreateTriggerCrossTxn<ED, T>).strict === 'makeSure') {
switch (operation.action) {
case 'create': {
if (operation.data.hasOwnProperty(Executor.dataAttr) || operation.data.hasOwnProperty(Executor.timestampAttr)) {
throw new Error('同一行数据上不能存在两个跨事务约束');
}
break;
}
default: {
const { filter } = operation;
// 此时要保证更新或者删除的行上没有跨事务约束
const filter2 = addFilterSegment({
$or: [
{
$$triggerData$$: {
$exists: true,
},
},
{
$$triggerTimestamp$$: {
$exists: true,
},
}
],
}, filter);
const { rowStore } = context;
const count = await rowStore.count(entity, {
filter: filter2
} as Omit<ED[T]['Selection'], 'action' | 'sorter' | 'data'>, context);
if (count > 0) {
throw new Error(`对象${entity}的行「${JSON.stringify(operation)}」上已经存在未完成的跨事务约束`);
}
break;
}
}
assign(operation.data, {
[Executor.dataAttr]: {
name: trigger.name,
operation,
},
[Executor.timestampAttr]: Date.now(),
});
}
}
async preOperation<T extends keyof ED>(
entity: T,
operation: ED[T]['Operation'],
context: Context<ED>
): Promise<void> {
const action = UnifiedActionMatrix[operation.action] || 'update';
const triggers = this.triggerMap[entity] && this.triggerMap[entity]![action];
if (triggers) {
const preTriggers = triggers.filter(
ele => ele.when === 'before' && (!(ele as CreateTrigger<ED, T>).check || (ele as CreateTrigger<ED, T>).check!(operation as DeduceCreateOperation<ED[T]['Schema']>))
);
for (const trigger of preTriggers) {
const number = await (trigger as CreateTrigger<ED, T>).fn({ operation: operation as DeduceCreateOperation<ED[T]['Schema']> }, context);
if (number > 0) {
this.logger.info(`触发器「${trigger.name}」成功触发了「${number}」行数据更改`);
}
}
const commitTriggers = triggers.filter(
ele => ele.when === 'commit' && (!(ele as CreateTrigger<ED, T>).check || (ele as CreateTrigger<ED, T>).check!(operation as DeduceCreateOperation<ED[T]['Schema']>))
);
for (const trigger of commitTriggers) {
await this.preCommitTrigger(entity, operation, trigger, context);
}
}
}
private onCommit<T extends keyof ED>(
trigger: Trigger<ED, T>, operation: ED[T]['Operation']) {
return async (context: Context<ED>) => {
await context.begin();
const number = await (trigger as CreateTrigger<ED, T>).fn({
operation: operation as DeduceCreateOperation<ED[T]['Schema']>,
}, context);
const { rowStore } = context;
if ((trigger as CreateTriggerCrossTxn<ED, T>).strict === 'makeSure') {
// 如果是必须完成的trigger，在完成成功后要把trigger相关的属性置null;
let filter = {};
if (operation.action === 'create') {
filter = operation.data instanceof Array ? {
filter: {
id: {
$in: operation.data.map(ele => (ele.id as string)),
},
},
} : {
filter: {
id: (operation.data.id as string),
}
};
}
else if (operation.filter) {
assign(filter, { filter: operation.filter });
}
await rowStore.operate(trigger.entity, {
action: 'update',
data: {
$$triggerTimestamp$$: null,
$$triggerData$$: null,
} as any,
...filter /** as Filter<'update', DeduceFilter<ED[T]['Schema']>> */,
}, context);
}
await context.commit();
return;
};
}
private async postCommitTrigger<T extends keyof ED>(
operation: ED[T]['Operation'],
trigger: Trigger<ED, T>,
context: Context<ED>
) {
context.on('commit', this.onCommit(trigger, operation));
}
async postOperation<T extends keyof ED>(
entity: T,
operation: ED[T]['Operation'],
context: Context<ED>
): Promise<void> {
const action = UnifiedActionMatrix[operation.action] || 'update';
const triggers = this.triggerMap[entity] && this.triggerMap[entity]![action];
if (triggers) {
const postTriggers = triggers.filter(
ele => ele.when === 'after' && (!(ele as CreateTrigger<ED, T>).check || (ele as CreateTrigger<ED, T>).check!(operation as DeduceCreateOperation<ED[T]['Schema']>))
);
for (const trigger of postTriggers) {
const number = await (trigger as CreateTrigger<ED, T>).fn({ operation: operation as DeduceCreateOperation<ED[T]['Schema']> }, context);
if (number > 0) {
this.logger.info(`触发器「${trigger.name}」成功触发了「${number}」行数据更改`);
}
}
const commitTriggers = (<Array<CreateTrigger<ED, T>>>triggers).filter(
ele => ele.when === 'commit' && (!ele.check || ele.check(operation as DeduceCreateOperation<ED[T]['Schema']>))
);
for (const trigger of commitTriggers) {
await this.postCommitTrigger(operation, trigger, context);
}
}
}
async checkpoint(context: Context<ED>, timestamp: number): Promise<number> {
let result = 0;
const { rowStore } = context;
for (const entity of this.volatileEntities) {
const { result: rows } = await rowStore.select(entity, {
data: {
id: 1,
$$triggerData$$: 1,
},
filter: {
$$triggerTimestamp$$: {
$gt: timestamp,
}
},
} as any, context);
for (const row of rows) {
const { $$triggerData$$ } = row;
const { name, operation } = $$triggerData$$!;
const trigger = this.triggerNameMap[name];
await this.onCommit(trigger, operation as ED[typeof entity]['Operation'])(context);
}
}
return result;
}
}

67
src/store/UniversalContext.ts Normal file
View File

@ -0,0 +1,67 @@
import { EntityDict, OpRecord, RowStore, TxnOption, Context } from "../types";
export class UniversalContext<ED extends EntityDict> implements Context<ED> {
rowStore: RowStore<ED>;
uuid?: string;
opRecords: OpRecord<ED>[];
events: {
commit: Array<(context: UniversalContext<ED>) => Promise<void>>;
rollback: Array<(context: UniversalContext<ED>) => Promise<void>>;
}
constructor(store: RowStore<ED>) {
this.rowStore = store;
this.opRecords = [];
this.events = {
commit: [],
rollback: [],
};
}
private resetEvents() {
this.events = {
commit: [],
rollback: [],
};
}
on(event: 'commit' | 'rollback', callback: (context: UniversalContext<ED>) => Promise<void>): void {
this.uuid && this.events[event].push(callback);
}
async begin(options?: TxnOption): Promise<void> {
if (!this.uuid) {
this.uuid = await this.rowStore.begin(options);
}
}
async commit(): Promise<void> {
if (this.uuid) {
/**
* todo 这里应该等到提交成功了再做 by Xc
*/
for(const e of this.events.commit) {
await e(this);
}
await this.rowStore.commit(this.uuid!);
this.uuid = undefined;
this.resetEvents();
}
}
async rollback(): Promise<void> {
if(this.uuid) {
/**
* todo 这里应该等到回滚成功了再做 by Xc
*/
for(const e of this.events.rollback) {
await e(this);
}
await this.rowStore.rollback(this.uuid!);
this.uuid = undefined;
this.resetEvents();
}
}
getCurrentTxnId() {
return this.uuid;
}
}

6
src/types/Aspect.ts Normal file
View File

@ -0,0 +1,6 @@
import { EntityDict } from "./Entity";
import { Context } from './Context';
export interface Aspect<ED extends EntityDict, Cxt extends Context<ED>>{
(params: any, context: Cxt): Promise<any>;
};

35
src/types/Auth.ts Normal file
View File

@ -0,0 +1,35 @@
import { EntityDict } from "../types/Entity";
import { CreateTriggerBase, RemoveTriggerBase, UpdateTriggerBase } from "./Trigger";
export class AttrIllegalError extends Error {
private attributes: string[];
constructor(attributes: string[], message?: string) {
super(message);
this.attributes = attributes;
}
getAttributes() {
return this.attributes;
}
};
export type CreateChecker<ED extends EntityDict, T extends keyof ED> = {
action: 'create';
entity: T;
checker: CreateTriggerBase<ED, T>['fn'],
};
export type UpdateChecker<ED extends EntityDict, T extends keyof ED> = {
action: UpdateTriggerBase<ED, T>['action'];
entity: T;
checker: UpdateTriggerBase<ED, T>['fn'],
};
export type RemoveChecker<ED extends EntityDict, T extends keyof ED> = {
action: 'remove';
entity: T;
checker: RemoveTriggerBase<ED, T>['fn'],
};
export type Checker<ED extends EntityDict, T extends keyof ED> = CreateChecker<ED, T> | UpdateChecker<ED, T> | RemoveChecker<ED, T>;

121
src/types/Trigger.ts Normal file
View File

@ -0,0 +1,121 @@
import { GenericAction } from "../actions/action";
import { DeduceCreateOperation, DeduceRemoveOperation, DeduceSelection, DeduceUpdateOperation, EntityDict } from "../types/Entity";
import { EntityDef, EntityShape, OperationResult, SelectionResult2, TriggerDataAttribute, TriggerTimestampAttribute } from "../types/Entity";
import { Context } from "./Context";
export interface CreateTriggerBase<ED extends EntityDict, T extends keyof ED> {
entity: T;
name: string;
action: 'create',
check?: (operation: DeduceCreateOperation<ED[T]['Schema']>) => boolean;
fn: (event: { operation: DeduceCreateOperation<ED[T]['Schema']>; }, context: Context<ED>, params?: Object) => Promise<number>;
};
export interface CreateTriggerInTxn<ED extends EntityDict, T extends keyof ED> extends CreateTriggerBase<ED, T> {
when: 'before' | 'after',
};
export interface CreateTriggerCrossTxn<ED extends EntityDict, T extends keyof ED> extends CreateTriggerBase<ED, T> {
when: 'commit',
strict?: 'takeEasy' | 'makeSure';
};
export type CreateTrigger<ED extends EntityDict, T extends keyof ED> = CreateTriggerInTxn<ED, T> | CreateTriggerCrossTxn<ED, T>;
export interface UpdateTriggerBase<ED extends EntityDict, T extends keyof ED> {
entity: T;
name: string;
action: Exclude<ED[T]['Action'], GenericAction> | 'update',
attributes?: keyof ED[T]['OpSchema'] | Array<keyof ED[T]['OpSchema']>;
check?: (operation: DeduceUpdateOperation<ED[T]['Schema']>) => boolean;
fn: (event: { operation: DeduceUpdateOperation<ED[T]['Schema']> }, context: Context<ED>, params?: Object) => Promise<number>;
};
export interface UpdateTriggerInTxn<ED extends EntityDict, T extends keyof ED> extends UpdateTriggerBase<ED, T> {
when: 'before' | 'after',
};
export interface UpdateTriggerCrossTxn<ED extends EntityDict, T extends keyof ED> extends UpdateTriggerBase<ED, T> {
when: 'commit',
strict?: 'takeEasy' | 'makeSure';
};
export type UpdateTrigger<ED extends EntityDict, T extends keyof ED> = UpdateTriggerInTxn<ED, T> | UpdateTriggerCrossTxn<ED, T>;
export interface RemoveTriggerBase<ED extends EntityDict, T extends keyof ED> {
entity: T;
name: string;
action: 'remove',
check?: (operation: DeduceRemoveOperation<ED[T]['Schema']>) => boolean;
fn: (event: { operation: DeduceRemoveOperation<ED[T]['Schema']> }, context: Context<ED>, params?: Object) => Promise<number>;
};
export interface RemoveTriggerInTxn<ED extends EntityDict, T extends keyof ED> extends RemoveTriggerBase<ED, T> {
when: 'before' | 'after',
};
export interface RemoveTriggerCrossTxn<ED extends EntityDict, T extends keyof ED> extends RemoveTriggerBase<ED, T> {
when: 'commit',
strict?: 'takeEasy' | 'makeSure';
};
export type RemoveTrigger<ED extends EntityDict, T extends keyof ED> = RemoveTriggerInTxn<ED, T> | RemoveTriggerCrossTxn<ED, T>;
export interface SelectTriggerBase<ED extends EntityDict, T extends keyof ED> {
entity: T;
name: string;
action: 'select';
};
/**
* selection似乎不需要支持跨事务？没想清楚
* todo by Xc
*/
export interface SelectTriggerBefore<ED extends EntityDict, T extends keyof ED> extends SelectTriggerBase<ED, T> {
when: 'before';
fn: (event: { operation: DeduceSelection<ED[T]['Schema']> }, context: Context<ED>, params?: Object) => Promise<number>;
};
export interface SelectTriggerAfter<ED extends EntityDict, T extends keyof ED> extends SelectTriggerBase<ED, T> {
when: 'after',
fn: <S extends ED[T]['Selection']>(event: {
operation: S;
result: SelectionResult2<ED[T]['Schema'], S['data']>;
}, context: Context<ED>, params?: Object) => Promise<number>;
};
export type SelectTrigger<ED extends EntityDict, T extends keyof ED> = SelectTriggerBefore<ED, T> | SelectTriggerAfter<ED, T>;
export type Trigger<ED extends EntityDict, T extends keyof ED> = CreateTrigger<ED, T> | UpdateTrigger<ED, T> | RemoveTrigger<ED, T> | SelectTrigger<ED, T>;
export interface TriggerEntityShape extends EntityShape {
$$triggerData$$?: {
name: string;
operation: object;
};
$$triggerTimestamp$$?: number;
};
export abstract class Executor<ED extends EntityDict> {
static dataAttr: TriggerDataAttribute = '$$triggerData$$';
static timestampAttr: TriggerTimestampAttribute = '$$triggerTimestamp$$';
abstract registerTrigger<T extends keyof ED>(trigger: Trigger<ED, T>): void;
abstract preOperation<T extends keyof ED>(
entity: T,
operation: ED[T]['Operation'],
context: Context<ED>
): Promise<void>;
abstract postOperation<T extends keyof ED>(
entity: T,
operation: ED[T]['Operation'],
context: Context<ED>
): Promise<void>;
abstract checkpoint(context: Context<ED>, timestamp: number): Promise<number>; // 将所有在timestamp之前存在不一致的数据进行恢复
}

14
src/types/index.ts Normal file
View File

@ -0,0 +1,14 @@
export * from './Action';
export * from './Aspect';
export * from './Auth';
export * from './Context';
export * from './DataType';
export * from './Demand';
export * from './Entity';
export * from './Expression';
export * from './Geo';
export * from './Logger';
export * from './Polyfill';
export * from './RowStore';
export * from './Storage';
export * from './Trigger';