逻辑引擎与自动化 —— 事件驱动的后续动作 (Workflow Logic)

MetaForm 低代码引擎系列 · 第 5 篇（完结篇）
技术栈：FastAPI BackgroundTasks / Go Channel + Event Bus

一、从同步到异步

随着业务复杂度攀升，产品需求变成了："请在用户提交问卷后，自动发送感谢邮件并推送到企业微信，随后修改另一张表的计数字段。"

在早期开发中，这些动作很容易堆积在同步的 API 主线程里：

# 灾难的同步堆叠
def create_order(payload):
    insert_to_db(payload)          # 写入 10 毫秒
    send_email(payload["email"])   # 阻断 2000 毫秒！
    call_webhook()                 # 网络抖动，阻塞 5000 毫秒！
    return {"status": "ok"}        # 用户看着白屏转圈 7 秒钟

对于一个成熟的元数据系统：写主表的动作必须极致得快，无论后面挂起什么长程操作，都不应该阻塞前端那轻快的 200 OK 响应。

所有复杂的流转动作必须在异步框架下完成。

二、设计 meta_workflows 结构

由于这是一个完全动态的系统，触发行为本身也作为"软逻辑"记录在字典表中。它回答三个问题：

CREATE TABLE meta_workflows (
    workflow_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    form_id VARCHAR(64) NOT NULL,
    trigger_type VARCHAR(32) NOT NULL,   -- 'AfterInsert', 'AfterUpdate'
    condition_formula TEXT,              -- 条件公式（可为空 = 无条件触发）
    action_type VARCHAR(64) NOT NULL,    -- 'SendEmail', 'CallWebhook', 'UpdateRecord'
    action_config JSONB NOT NULL         -- 动作的具体配置参数
);

有了这份配置表，租户可以通过前端拖拉拽配置出："当【销售单】在【新建】时，如果【金额>100】，则触发【给经理发邮件】" 的无代码自动化规则。

三、内部事件总线 (Event Bus) 构建

在 Python/FastAPI 中，利用 BackgroundTasks 实现单机异步；在 Go 中利用 Channel；到微服务阶段可平滑迁移到 Kafka 或 RabbitMQ。

from fastapi import BackgroundTasks
from dataclasses import dataclass

@dataclass
class RecordEvent:
    form_id: str
    trigger_type: str  # "AfterInsert", "AfterUpdate"
    payload: dict      # 强烈建议：携带完整的 JSONB 数据，避免异步任务中产生二次查库
    execution_depth: int = 1  # ⚠️ 架构师防线：递归深度控制计数器

def event_listener_worker(event: RecordEvent, db: Session):
    """后台独立线程中的逻辑引擎"""

    # 【灾难预防】防循环机制 (Recursion Loop Prevention)
    if event.execution_depth > 5:
        # 预防 A 修改 B，B 又修改 A 导致的死循环，超过阈值强行阻断
        print(f"Trigger recursion limit exceeded for {event.form_id}!")
        return

    # 1. 查询匹配的自动化流程
    workflows = db.execute(
        """SELECT condition_formula, action_type, action_config
           FROM meta_workflows
           WHERE form_id = :fid AND trigger_type = :ttype""",
        {"fid": event.form_id, "ttype": event.trigger_type}
    ).fetchall()

    # 2. 评估条件（复用第 4 篇的沙箱）
    for wf in workflows:
        if not wf.condition_formula or sandbox_eval(wf.condition_formula, event.payload):
            # 3. 执行动作
            execute_action(wf.action_type, wf.action_config, event.payload)


def execute_action(action_type: str, config: dict, payload: dict):
    """动作分发器"""
    if action_type == "SendEmail":
        send_mail(config["to"], config["template_id"], context=payload)
    elif action_type == "CallWebhook":
        requests.post(config["url"], json=payload, timeout=30)
    elif action_type == "UpdateRecord":
        update_related_record(config["target_form_id"], config["updates"], payload)

四、整合到 DML 写入链

将事件投递嵌入到第 3 篇的写入接口中：

@router.post("/api/data/{form_id}")
def insert_record(
    form_id: str,
    payload: dict,
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db)
):
    # Step 1: 加载元数据 + Canonical 编码（第 3 篇）
    canonical_payload = canonical_encode_all(payload, fields_meta)

    # Step 2: 规则引擎拦截校验（第 4 篇）
    execute_validation_rules(form_id, canonical_payload, db)

    # Step 3: 落库
    db.execute("INSERT INTO data_heap (...) VALUES (...)", {...})
    db.commit()

    # Step 4: 抛出异步事件，迅速结束主线程！
    background_tasks.add_task(
        event_listener_worker,
        RecordEvent(form_id=form_id, trigger_type="AfterInsert", payload=canonical_payload),
        db
    )

    return {"status": "ok"}  # 前端瞬间收到响应

五、Workflow 异步触发机制图解

架构分为两个泳道：

主线程 (Sync)：INSERT data_heap → 立即返回前端 200 OK，同时向 Internal Event Bus 投递 RecordCreatedEvent

架构师提示（关于动作上下文）：主线程投递事件时，务必携带 form_id 和完整的 JSONB payload 数据。这样异步的 Workflow 引擎在评估 Condition (如 Status == 'pending') 以及执行发邮件等模板渲染时，就不需要重新发起一次昂贵的数据库查询。

异步工作线程 (Async)：

1. Logic Engine 监听事件
2. 匹配 meta_workflows 的条件（直接复用传入的 payload 上下文）
3. 触发后续 Actions：发送邮件、更新关联数据、调用 Webhook

六、专栏大总结

在这 5 篇文章中，我们见证了一个硬编码的单体应用被彻底重构为灵活至极的元数据驱动引擎：

这就是以 Salesforce 为开端、演进至今的 Metadata-Driven Architecture（元数据驱动架构）——一套把复杂封装到底座里，把无限空间留给上层配置组合的架构哲学。

感谢您的阅读。