Sign in

    jeeflow-persist

    jeeflow dynamic table persist (DynamicTableWriter + PersistPostInterceptor + Meta)

    Download zip
    Author
    Version
    0.1.34
    License
    Apache-2.0
    Last updated
    2 hours ago
    Downloads
    168

    Dependencies

    #jeeflow-persist

    Workflow → business-table persistence for jeeflow — ARCHIVE / SYNC with field permissions.

    mooncakes License

    PersistPostInterceptor copies form data out of running workflow instances into your own business tables, driven entirely by the process definition — the role jeeflow-persist plays for the MoonBit build of jeeflow-moon.

    #Quickstart

    let provider = @persist.InMemoryMetaProvider::new()
    let meta = @persist.TableMeta::make("expense", "报销")
    // ⚠ 元数据是**白名单**:`filter_editable` 只放行登记过且 permission=2(可编辑)的列。
    // 只登记业务列 `note` 而漏掉 `process_instance_id`,归档行就带着空幂等键落库 ⇒
    // 每次办结都插一行,幂等(C16)静默失效。要写的列请登记全(issues/151 普查 X7 的实话)。
    meta.add_field(@persist.FieldMeta::make("process_instance_id", "BIGINT", "流程实例ID"))
    meta.add_field(@persist.FieldMeta::make("apply_user_id", "VARCHAR(64)", "申请人"))
    meta.add_field(@persist.FieldMeta::make("note", "VARCHAR(200)", "备注"))
    provider.register(meta)

    let interceptor = @persist.PersistPostInterceptor::make(provider, my_writer)
    ctx.register_interceptor(interceptor.as_interceptor()) // order = 100 (post)

    两个装配口按集成方的元数据来源选:自备元数据(框架 dev_schema / 库内省)用 PersistPostInterceptor::make_from_provider;另有 with_system_fields(系统列名与缺省用户值, 见下)与 with_strict_columns(列匹配档位)。

    Process definition drives everything:

    {"persistMode": "ARCHIVE", "relTableName": "expense"}

    • ARCHIVE — at the end node, on FINISHED + agree (submitType=1), insert the full f_* form once; idempotent by the process_instance_id key.
    • SYNC — insert at start → update at each task (business fields filtered by the target node's field.PERMISSION_*: readonly/hidden never write through) → final-state update at the end; a {nodeId}_{state} status column is probed automatically.
    • Safety — dynamic table names are validated (alnum+underscore, wf_/sys_/… prefixes refused) and failures are loud, never silent.
    • Column matching — loose by default: a camelCase form key (merchantName) lands on the snake_case table column (merchant_name), case-insensitively, and the SQL always emits the real column name. Two data keys normalising onto one column resolve deterministically — the first one in insertion order wins. Exact matching is one switch away (MysqlTableWriter::with_strict(true) and PersistPostInterceptor::with_strict_columns(true) are the same ruler's two consumers: the table-structure filter and the metadata whitelist — set both, or the two layers measure with different sticks).
    • System columns — create_time / create_user / update_time / update_user / is_deleted, all renameable and individually disableable (None) via PersistPostInterceptor::with_system_fields. Insert fills them with putIfAbsent semantics (a business-supplied create_time survives); the update leg overwrites update_time only. User columns read apply_user_id first (= the flow's operator), then the current operator, then the configured fallback (default "system") — spec/09 §3, matching the Java reference.
    • Storage types (spec/10) — a field's storageType (name or code 1-5) drives how it lands: NORMAL straight to a column, EXPAND one object → several columns (expandFields maps sub-field → column), JSON object/array → a JSON string column, ONE2ONE/ONE2MANY → a recursive sub-table insert/update keyed by the parent's primary key (inheriting the parent's apply_user_id). Sub-tables never participate in mid-flow updates. Wrap any writer with MetaTableWriter::make(base, provider); read the same shape back with MetaTableReader::make(provider, reader_fns) whose as_biz_data_reader() slots straight into Ctx::with_biz_data_reader.
    • Auto-increment / pk generation — a missing id is filled by the configured with_pk_generator (or this stack's snowflake by default); AUTO_INCREMENT columns are left to the database and read back via LAST_INSERT_ID() on the same connection.

    X12 的实话:PersistPostInterceptor 的两个字段都是必填,所以 java :83-87 那一档「未注入 writer ⇒ ServiceContext.find 兜底、再找不到就静默跳过」在本栈结构上不存在。 这不是免检:哪天把它改成可选字段或可选注册,spec/09 §4.6 的「未注入静默跳过 vs 显性报错」判据 必须连着补回来(普查 jeeflow-hub/issues/151 的 X12 行记着这一句)。

    #License

    Apache-2.0 — part of jeeflow-moon.

    DynamicTableWriter

    pub(open) trait DynamicTableWriter {
    async fn insert(Self, String, Map[String, Json]) -> Int64 raise
    JeeflowError

    async fn update(Self, String, Int64, Map[String, Json]) -> Unit raise
    JeeflowError

    async fn exists(Self, String, Int64) -> Bool raise
    JeeflowError

    async fn exists_by_key(Self, String, String, Int64) -> Bool raise
    JeeflowError

    async fn update_by_key(Self, String, Map[String, Json], String, Int64) -> Unit raise
    JeeflowError

    async fn columns(Self, String, Array[String]) -> Array[String] raise
    JeeflowError

    }

    DynamicTableWriter trait(MoonBit:大接口 trait 形态)

    issues/149:六法全部 async —— 本栈写真库只有 moonmysql 的 async 连接这一条路,同步形状 发不出查询(145-6 记的机理)。async 由 trait 声明承载:impl … with fn 处不重复写 async(实测写了判 parse error),⇒ InMemoryTableWriter 六支 impl 与 &DynamicTableWriter 的六处调用点都不用改;唯一要跟着改的是同步函数里调 writer 的 put_state_field。 同形参照=core/spi/repository.mbt 的 ProcessRepository(async trait + repository-mysql 真库实现)。 ⚠ 列探测(columns)归属裁决=留在 writer:参考实现 java persist/DynamicTableWriter.java:20 的 filterColumns 就挂在 writer 上,而 IDynamicMetaProvider 管字段权限/显示元数据——两者在 java 本就并存,故不并入 DynamicMetaProviderFns(146 缺口二),也不留"同一件事两个端口"。

    DynamicMetaProviderFns

    pub(all) struct DynamicMetaProviderFns {
    load_table_meta : async (String) -> TableMeta? raise
    JeeflowError

    }

    IDynamicMetaProvider 的 MoonBit 同位(issues/146 缺口二)。 java 侧这个端口就定义在 jeeflow-persist/meta(不在 core),消费点是 MetaTableWriter/Reader ⇒ 本栈同位同样落在 persist:归档/SYNC 的写侧与回显读侧共用一份元数据(storageType 语义两侧一致)。 load_table_meta 是 async:集成方的元数据来源是库(mldong dev_schema / information_schema), 同步形状就会重演缺口一——启动期快照、改表结构引擎看不见。 未注入 ⇒ 引擎行为与既往一致(InMemory 档照常,缺省不启用外部元数据)。

    DynamicTableReaderFns

    pub(all) struct DynamicTableReaderFns {
    row_by_key : async (TableKeyQuery) -> Map[String, Json]? raise
    JeeflowError

    rows_by_key : async (TableKeyQuery) -> Array[Map[String, Json]] raise
    JeeflowError

    }

    行读取端口:按「列等值」取单行 / 多行。列名与值都是引擎内部给的(外键列来自元数据), 实现方(宿主/仓储)自己负责防注入。

    FieldMeta

    pub(all) struct FieldMeta {
    column_name : String
    column_type : String
    display_name : String
    nullable : Bool
    default_value : String?
    permission : Int
    name : String
    storage_type : StorageType
    expand_fields : Map[String, String]
    target_table : String
    foreign_key : String
    }

    字段元数据(动态表列)

    FieldMeta::is_editable

    fn FieldMeta::is_editable(self : FieldMeta) -> Bool

    FieldMeta::is_visible

    fn FieldMeta::is_visible(self : FieldMeta) -> Bool

    FieldMeta::make

    fn FieldMeta::make(column_name : String, column_type : String, display_name : String, name? : String, storage_type? : StorageType, expand_fields? : Map[String, String], target_table? : String, foreign_key? : String) -> FieldMeta

    InMemoryMetaProvider

    pub(all) struct InMemoryMetaProvider {
    tables : Map[String, TableMeta]
    }

    IDynamicMetaProvider 内存实现

    InMemoryMetaProvider::as_provider

    内存档 → 端口(纯内存读 ⇒ 经 @spi.async_of_sync 桥,注册点零警告)

    InMemoryMetaProvider::get_table_meta

    fn InMemoryMetaProvider::get_table_meta(self : InMemoryMetaProvider, table_name : String) -> TableMeta?

    InMemoryMetaProvider::list_table_names

    fn InMemoryMetaProvider::list_table_names(self : InMemoryMetaProvider) -> Array[String]

    InMemoryMetaProvider::new

    InMemoryMetaProvider::register

    fn InMemoryMetaProvider::register(self : InMemoryMetaProvider, meta : TableMeta) -> Unit

    InMemoryTableWriter

    pub(all) struct InMemoryTableWriter {
    id_counter : Int64
    tables : Map[String, Map[Int64, Map[String, Json]]]
    }

    InMemoryTableWriter::columns

    async fn InMemoryTableWriter::columns(self : InMemoryTableWriter, _table_name : String, candidates : Array[String]) -> Array[String] raise
    JeeflowError

    InMemoryTableWriter::exists

    async fn InMemoryTableWriter::exists(self : InMemoryTableWriter, table_name : String, id : Int64) -> Bool raise
    JeeflowError

    InMemoryTableWriter::exists_by_key

    async fn InMemoryTableWriter::exists_by_key(self : InMemoryTableWriter, table_name : String, key : String, value : Int64) -> Bool raise
    JeeflowError

    InMemoryTableWriter::get_row

    fn InMemoryTableWriter::get_row(self : InMemoryTableWriter, table_name : String, id : Int64) -> Map[String, Json]?

    InMemoryTableWriter::get_row_by_key

    fn InMemoryTableWriter::get_row_by_key(self : InMemoryTableWriter, table_name : String, key : String, value : Int64) -> Map[String, Json]?

    InMemoryTableWriter::get_table_rows

    fn InMemoryTableWriter::get_table_rows(self : InMemoryTableWriter, table_name : String) -> Array[Map[String, Json]]

    InMemoryTableWriter::insert

    async fn InMemoryTableWriter::insert(self : InMemoryTableWriter, table_name : String, data : Map[String, Json]) -> Int64 raise
    JeeflowError

    InMemoryTableWriter::new

    InMemoryTableWriter::update

    async fn InMemoryTableWriter::update(self : InMemoryTableWriter, table_name : String, id : Int64, data : Map[String, Json]) -> Unit raise
    JeeflowError

    InMemoryTableWriter::update_by_key

    async fn InMemoryTableWriter::update_by_key(self : InMemoryTableWriter, table_name : String, data : Map[String, Json], key : String, value : Int64) -> Unit raise
    JeeflowError

    MetaTableReader

    pub(all) struct MetaTableReader {
    meta : DynamicMetaProviderFns
    reader : DynamicTableReaderFns
    }

    MetaTableReader::as_biz_data_reader

    fn MetaTableReader::as_biz_data_reader(self : MetaTableReader) -> (async (String, Int64) -> Map[String, Json]? raise
    JeeflowError
    )

    包成与 Ctx.biz_data_reader 同签名的闭包 ⇒ 宿主 with_biz_data_reader(...) 一行接好, facade 的 bizData 出口与 issues/137 的泄漏纪律都不动。

    MetaTableReader::assemble

    async fn MetaTableReader::assemble(self : MetaTableReader, meta : TableMeta, row : Map[String, Json]) -> Map[String, Json] raise
    JeeflowError

    按元数据组装回显结果(键=表单字段名;java assemble:50-79 同判据)

    MetaTableReader::make

    MetaTableReader::read_by_process_instance

    async fn MetaTableReader::read_by_process_instance(self : MetaTableReader, table : String, instance_id : Int64) -> Map[String, Json]? raise
    JeeflowError

    按流程实例回显一条业务数据(定位键=process_instance_id,与写入幂等键同款)。 无记录 ⇒ None;无元数据 ⇒ 原始行(列名→值)。

    MetaTableWriter

    pub(all) struct MetaTableWriter {
    base : &DynamicTableWriter
    meta : DynamicMetaProviderFns
    }

    MetaTableWriter::columns

    async fn MetaTableWriter::columns(self : MetaTableWriter, table : String, candidates : Array[String]) -> Array[String] raise
    JeeflowError

    MetaTableWriter::exists

    async fn MetaTableWriter::exists(self : MetaTableWriter, table : String, id : Int64) -> Bool raise
    JeeflowError

    MetaTableWriter::exists_by_key

    async fn MetaTableWriter::exists_by_key(self : MetaTableWriter, table : String, key : String, value : Int64) -> Bool raise
    JeeflowError

    MetaTableWriter::insert

    async fn MetaTableWriter::insert(self : MetaTableWriter, table : String, data : Map[String, Json]) -> Int64 raise
    JeeflowError

    MetaTableWriter::make

    MetaTableWriter::update

    async fn MetaTableWriter::update(self : MetaTableWriter, table : String, id : Int64, data : Map[String, Json]) -> Unit raise
    JeeflowError

    MetaTableWriter::update_by_key

    async fn MetaTableWriter::update_by_key(self : MetaTableWriter, table : String, data : Map[String, Json], key : String, value : Int64) -> Unit raise
    JeeflowError

    PersistMode

    pub enum PersistMode {
    Archive
    Sync
    } derive(Eq,
    Debug
    )

    持久化模式

    PersistMode::equal

    fn PersistMode::equal(PersistMode, PersistMode) -> Bool
    automatically derived

    PersistMode::from_str

    fn PersistMode::from_str(s : String) -> PersistMode?

    PersistMode::not_equal

    fn PersistMode::not_equal(x : PersistMode, y : PersistMode) -> Bool

    PersistMode::to_repr

    automatically derived

    PersistPostInterceptor

    pub(all) struct PersistPostInterceptor {
    meta : DynamicMetaProviderFns
    table_writer : &DynamicTableWriter
    system_fields : SystemFields
    strict_columns : Bool
    }

    PersistPostInterceptor::as_interceptor

    装配为引擎拦截器(order=100,后置)

    PersistPostInterceptor::make

    PersistPostInterceptor::make_from_provider

    端口档装配(issues/146 缺口二):集成方自备元数据来源(dev_schema/库内省)走这一支

    PersistPostInterceptor::with_strict_columns

    fn PersistPostInterceptor::with_strict_columns(self : PersistPostInterceptor, strict_columns : Bool) -> PersistPostInterceptor
    严格列匹配装配口(issues/151 X3)

    PersistPostInterceptor::with_system_fields

    fn PersistPostInterceptor::with_system_fields(self : PersistPostInterceptor, system_fields : SystemFields) -> PersistPostInterceptor
    或不想填 is_deleted 的集成方用它。缺省档见 SystemFields::make_default。

    StorageType

    pub(all) enum StorageType {
    Normal
    Json
    Expand
    One2One
    One2Many
    } derive(Eq,
    Debug
    )

    存储类型(issues/151 X10/spec/10 §2.1):名称/数字双解析,对齐 mldong dev_schema_field 的 1-5 与 java StorageType.fromValue。

    StorageType::equal

    fn StorageType::equal(StorageType, StorageType) -> Bool
    automatically derived

    StorageType::not_equal

    fn StorageType::not_equal(x : StorageType, y : StorageType) -> Bool

    StorageType::to_repr

    automatically derived

    SystemFields

    pub(all) struct SystemFields {
    create_time_col : String?
    create_user_col : String?
    update_time_col : String?
    update_user_col : String?
    is_deleted_col : String?
    default_user_value : String
    }

    系统列配置(issues/150 X5,java JdbcDynamicTableWriter:57-65,211-234 同位)。 列名可配置,None = 不填该列(java 用 null 表达同一意思); default_user_value = 取不到操作人时的回落值(java 缺省 "system")。

    SystemFields::make_default

    fn SystemFields::make_default() -> SystemFields

    TableKeyQuery

    pub(all) struct TableKeyQuery {
    table : String
    column : String
    value : Int64
    }

    一次「按列等值」的行查询请求(表 / 列 / 值)。 三参数打平成一个请求结构,不是省字符:本栈的 async 闭包若体内不 await 就判 unused_async,而 @spi 的同步→异步桥只有 1/2 参两档(async_of_sync / async_of_sync2), 单参数档才接得上 ⇒ 纯内存实现经桥注册、零新增告警(同 InMemoryMetaProvider::as_provider 那条)。

    TableMeta

    pub(all) struct TableMeta {
    table_name : String
    display_name : String
    fields : Array[FieldMeta]
    primary_key : String
    }

    TableMeta::add_field

    fn TableMeta::add_field(self : TableMeta, field : FieldMeta) -> Unit

    TableMeta::find_field

    fn TableMeta::find_field(self : TableMeta, key : String) -> FieldMeta?

    按表单字段名找字段(宽松列名匹配,issues/151 X3 那枚尺子)。 写侧用它判断"这个 data 键是不是已被某个字段消费"(java findField:74)。

    TableMeta::find_field_by_column

    fn TableMeta::find_field_by_column(self : TableMeta, column : String) -> FieldMeta?

    按表列名找字段(读侧判断某列是否已被消费,java findFieldByColumn:74)

    TableMeta::get_field

    fn TableMeta::get_field(self : TableMeta, column_name : String) -> FieldMeta?

    TableMeta::make

    fn TableMeta::make(table_name : String, display_name : String, primary_key? : String) -> TableMeta

    column_names_equal

    fn column_names_equal(a : String, b : String, strict? : Bool) -> Bool

    列名等价判定(java findColumn:293-302 / findDataKey:310-320 的同一枚尺子):
    • 严格档=忽略大小写的逐字相等(java setStrictColumnMatch(true));
    • 宽松档(默认)=归一后相等。 两键撞同一列时的确定性规则也在这里定:遍历方(find_data_key 走 data 插入序、 match_column 走表列序)里第一个命中者胜出,与 java 的 data.keySet() / List<ColumnMeta> 遍历顺序同判据——不写这条,同一份 data 两次运行可能落到不同列(Map 保插入序是 MoonBit 保证, 但"命中即停"必须显式,否则宽松档会静默多写一列)。

    fill_system_fields

    fn fill_system_fields(data : Map[String, Json], operator : String, is_insert : Bool) -> Unit

    缺省列名档(create_time/create_user/update_time/update_user/is_deleted)—— 旧调用形状保持可用,列名不同的集成方走 fill_system_fields_with。

    fill_system_fields_with

    fn fill_system_fields_with(cfg : SystemFields, data : Map[String, Json], operator : String, is_insert : Bool) -> Unit

    系统字段填充(时间走 Clock 注入口,D-M0-3)。 与 java fillSystemFields:211-224 逐条同判据:
    • 插入=create/update 四列 + is_deleted 全填,且一律 putIfAbsent(业务自带的时间/人不覆盖);
    • 更新=update_time 覆写(java 用 put,定稿时刻必须以本次为准)、update_user putIfAbsent;
    • 列名为 None ⇒ 该列不填。

    find_data_key

    fn find_data_key(data : Map[String, Json], column : String, strict? : Bool) -> String?

    在 data 中找匹配指定表列的 key(java findDataKey)。命中即停 ⇒ data 插入序里的第一个键胜出。

    is_table_name_safe

    fn is_table_name_safe(name : String) -> Bool

    动态表名安全检查:仅字母数字下划线 + 保留前缀拒绝

    match_column

    fn match_column(real_cols : Array[String], candidate : String, strict? : Bool) -> String?

    候选列名 → 表里实际存在的真列名(java findColumn)。 返回真列名而不是候选名,是因为写侧要把这个串直接拼进 SQL 的反引号位; java 的 filterColumns 返回候选名(它之后只当 data 键用),本栈两处都按 java 各自的那档实现。

    normalize_column

    fn normalize_column(name : String) -> String

    列名归一(issues/151 X3):转小写 + 去下划线 ⇒ companyName / company_name / COMPANY_NAME 三档等价。逐字照 java JdbcDynamicTableWriter.normalizeColumn:305-307。

    storage_type_of

    fn storage_type_of(v : Json) -> StorageType