mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(workflow): add publish API + workflow / trigger REST controllers
This commit is contained in:
matevip
parent
ce08b311d3
commit
67c0efe8ae
@ -0,0 +1,90 @@
|
||||
package vip.mate.trigger.api;
|
||||
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import io.swagger.v3.oas.annotations.tags.Tag;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.web.bind.annotation.*;
|
||||
import vip.mate.common.result.R;
|
||||
import vip.mate.trigger.ingest.TriggerEventEnvelope;
|
||||
import vip.mate.trigger.ingest.TriggerEventIngestService;
|
||||
import vip.mate.trigger.model.TriggerEntity;
|
||||
import vip.mate.trigger.service.TriggerService;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* REST surface for cron / event triggers. The generic event ingest endpoint
|
||||
* exists so external systems (n8n, GitHub webhooks, ad-hoc curl) can post
|
||||
* events without going through a dedicated channel adapter — useful for
|
||||
* smoke-testing a trigger before the channel integration lands.
|
||||
*/
|
||||
@Tag(name = "触发器管理")
|
||||
@RestController
|
||||
@RequestMapping("/api/v1/triggers")
|
||||
@RequiredArgsConstructor
|
||||
public class TriggerController {
|
||||
|
||||
private final TriggerService triggerService;
|
||||
private final TriggerEventIngestService ingestService;
|
||||
|
||||
@Operation(summary = "List triggers in a workspace.")
|
||||
@GetMapping
|
||||
public R<List<TriggerEntity>> list(@RequestParam("workspaceId") long workspaceId) {
|
||||
return R.ok(triggerService.listByWorkspace(workspaceId));
|
||||
}
|
||||
|
||||
@Operation(summary = "Get a trigger by id.")
|
||||
@GetMapping("/{id}")
|
||||
public R<TriggerEntity> get(@PathVariable long id) {
|
||||
TriggerEntity row = triggerService.get(id);
|
||||
if (row == null) return R.fail("trigger not found: " + id);
|
||||
return R.ok(row);
|
||||
}
|
||||
|
||||
@Operation(summary = "Create a trigger; if enabled, registers it with the scheduler.")
|
||||
@PostMapping
|
||||
public R<TriggerEntity> create(@RequestBody TriggerEntity trigger) {
|
||||
return R.ok(triggerService.create(trigger));
|
||||
}
|
||||
|
||||
@Operation(summary = "Update a trigger; pattern_version bumps when the cron expression changes.")
|
||||
@PutMapping("/{id}")
|
||||
public R<TriggerEntity> update(@PathVariable long id, @RequestBody TriggerEntity trigger) {
|
||||
trigger.setId(id);
|
||||
return R.ok(triggerService.update(trigger));
|
||||
}
|
||||
|
||||
@Operation(summary = "Delete a trigger and unregister its schedule.")
|
||||
@DeleteMapping("/{id}")
|
||||
public R<Void> delete(@PathVariable long id) {
|
||||
triggerService.delete(id);
|
||||
return R.ok();
|
||||
}
|
||||
|
||||
/**
|
||||
* Ingest one event envelope through the dedup / rate-limit / bot-self
|
||||
* pipeline. The endpoint is open to operators; production deployments
|
||||
* SHOULD gate it behind the workspace interceptor / a service token
|
||||
* before exposing it externally.
|
||||
*/
|
||||
@Operation(summary = "Ingest one event envelope; returns per-trigger fire / drop summary.")
|
||||
@PostMapping("/events")
|
||||
public R<List<TriggerEventIngestService.IngestResult>> ingestEvent(
|
||||
@RequestBody EventIngestRequest body) {
|
||||
TriggerEventEnvelope env = new TriggerEventEnvelope(
|
||||
body.workspaceId(),
|
||||
body.patternType(),
|
||||
body.eventId(),
|
||||
body.senderId(),
|
||||
body.data() == null ? Map.of() : body.data());
|
||||
return R.ok(ingestService.ingest(env));
|
||||
}
|
||||
|
||||
public record EventIngestRequest(
|
||||
long workspaceId,
|
||||
String patternType,
|
||||
String eventId,
|
||||
String senderId,
|
||||
Map<String, Object> data) {}
|
||||
}
|
||||
@ -0,0 +1,24 @@
|
||||
package vip.mate.workflow.api;
|
||||
|
||||
import vip.mate.workflow.compiler.CompileError;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Response shape for compile failures returned from publish / preview-compile
|
||||
* endpoints. Surfaces every diagnostic at once so the front-end editor can
|
||||
* highlight all offending fields in a single round trip; mirroring
|
||||
* {@link CompileError} preserves the path / code / message tuple the editor
|
||||
* expects.
|
||||
*/
|
||||
public record CompileErrorResponse(int errorCount, List<Item> errors) {
|
||||
|
||||
public record Item(String code, String path, String message) {}
|
||||
|
||||
public static CompileErrorResponse of(List<CompileError> errors) {
|
||||
List<Item> items = errors.stream()
|
||||
.map(e -> new Item(e.code(), e.path(), e.message()))
|
||||
.toList();
|
||||
return new CompileErrorResponse(items.size(), items);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,151 @@
|
||||
package vip.mate.workflow.api;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import io.swagger.v3.oas.annotations.tags.Tag;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.http.HttpStatus;
|
||||
import org.springframework.http.ResponseEntity;
|
||||
import org.springframework.web.bind.annotation.*;
|
||||
import vip.mate.common.result.R;
|
||||
import vip.mate.workflow.compiler.PublishContext;
|
||||
import vip.mate.workflow.compiler.WorkflowAclPort;
|
||||
import vip.mate.workflow.compiler.WorkflowCompileFailedException;
|
||||
import vip.mate.workflow.compiler.WorkflowCompiler;
|
||||
import vip.mate.workflow.model.WorkflowEntity;
|
||||
import vip.mate.workflow.model.WorkflowRunEntity;
|
||||
import vip.mate.workflow.model.WorkflowRunStepEntity;
|
||||
import vip.mate.workflow.repository.WorkflowRunMapper;
|
||||
import vip.mate.workflow.repository.WorkflowRunStepMapper;
|
||||
import vip.mate.workflow.service.WorkflowService;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* REST surface for workflow CRUD + draft / publish / run inspection.
|
||||
* Endpoints follow the project convention of a single workspace id passed
|
||||
* via query param (production deploys read it from {@code X-Workspace-Id}
|
||||
* via the workspace interceptor; the param fallback keeps tests simple).
|
||||
*/
|
||||
@Tag(name = "工作流管理")
|
||||
@RestController
|
||||
@RequestMapping("/api/v1/workflows")
|
||||
@RequiredArgsConstructor
|
||||
public class WorkflowController {
|
||||
|
||||
private final WorkflowService workflowService;
|
||||
private final WorkflowRunMapper runMapper;
|
||||
private final WorkflowRunStepMapper stepMapper;
|
||||
private final WorkflowCompiler compiler;
|
||||
private final WorkflowAclPort aclPort;
|
||||
|
||||
@Operation(summary = "List workflows in the workspace")
|
||||
@GetMapping
|
||||
public R<List<WorkflowEntity>> list(@RequestParam("workspaceId") long workspaceId) {
|
||||
return R.ok(workflowService.listByWorkspace(workspaceId));
|
||||
}
|
||||
|
||||
@Operation(summary = "Get a workflow by id (includes inline draft).")
|
||||
@GetMapping("/{id}")
|
||||
public R<WorkflowEntity> get(@PathVariable long id) {
|
||||
WorkflowEntity row = workflowService.get(id);
|
||||
if (row == null) return R.fail("workflow not found: " + id);
|
||||
return R.ok(row);
|
||||
}
|
||||
|
||||
@Operation(summary = "Create a workflow row (draft starts empty).")
|
||||
@PostMapping
|
||||
public R<WorkflowEntity> create(@RequestBody WorkflowEntity workflow) {
|
||||
return R.ok(workflowService.create(workflow));
|
||||
}
|
||||
|
||||
@Operation(summary = "Update workflow metadata (name / description / enabled).")
|
||||
@PutMapping("/{id}")
|
||||
public R<WorkflowEntity> update(@PathVariable long id,
|
||||
@RequestBody WorkflowEntity workflow) {
|
||||
workflow.setId(id);
|
||||
return R.ok(workflowService.update(workflow));
|
||||
}
|
||||
|
||||
@Operation(summary = "Save the inline draft graph_json without compiling.")
|
||||
@PutMapping("/{id}/draft")
|
||||
public R<WorkflowEntity> saveDraft(@PathVariable long id,
|
||||
@RequestBody WorkflowDraftRequest body,
|
||||
@RequestParam(value = "userId", required = false) Long userId) {
|
||||
return R.ok(workflowService.saveDraft(id, body.draftJson(), userId));
|
||||
}
|
||||
|
||||
@Operation(summary = "Compile the draft and surface diagnostics without persisting a revision.")
|
||||
@PostMapping("/{id}/compile")
|
||||
public ResponseEntity<?> compileDraft(@PathVariable long id) {
|
||||
WorkflowEntity row = workflowService.get(id);
|
||||
if (row == null || row.getDraftJson() == null) {
|
||||
return ResponseEntity.badRequest()
|
||||
.body(R.fail("workflow has no draft to compile: " + id));
|
||||
}
|
||||
WorkflowCompiler.Result result = compiler.compile(row.getDraftJson(),
|
||||
new PublishContext(0L, row.getWorkspaceId()), aclPort);
|
||||
if (!result.ok()) {
|
||||
return ResponseEntity.unprocessableEntity()
|
||||
.body(buildCompileFailure(result.errors()));
|
||||
}
|
||||
return ResponseEntity.ok(R.ok());
|
||||
}
|
||||
|
||||
@Operation(summary = "Compile the draft and persist a new revision pointed at by latest_revision_id.")
|
||||
@PostMapping("/{id}/publish")
|
||||
public ResponseEntity<?> publish(@PathVariable long id,
|
||||
@RequestBody(required = false) WorkflowPublishRequest body,
|
||||
@RequestParam(value = "userId", required = false) Long userId) {
|
||||
try {
|
||||
WorkflowService.PublishOutcome outcome = workflowService.publish(id, userId,
|
||||
body == null ? null : body.note());
|
||||
return ResponseEntity.ok(R.ok(outcome));
|
||||
} catch (WorkflowCompileFailedException e) {
|
||||
return ResponseEntity.unprocessableEntity().body(buildCompileFailure(e.errors()));
|
||||
} catch (IllegalArgumentException | IllegalStateException e) {
|
||||
return ResponseEntity.status(HttpStatus.BAD_REQUEST).body(R.fail(e.getMessage()));
|
||||
}
|
||||
}
|
||||
|
||||
@Operation(summary = "Soft-delete a workflow row.")
|
||||
@DeleteMapping("/{id}")
|
||||
public R<Void> delete(@PathVariable long id) {
|
||||
workflowService.delete(id);
|
||||
return R.ok();
|
||||
}
|
||||
|
||||
@Operation(summary = "List the most recent runs for a workflow.")
|
||||
@GetMapping("/{id}/runs")
|
||||
public R<List<WorkflowRunEntity>> listRuns(@PathVariable long id,
|
||||
@RequestParam(value = "limit", defaultValue = "50") int limit) {
|
||||
int capped = Math.min(Math.max(limit, 1), 200);
|
||||
List<WorkflowRunEntity> rows = runMapper.selectList(new LambdaQueryWrapper<WorkflowRunEntity>()
|
||||
.eq(WorkflowRunEntity::getWorkflowId, id)
|
||||
.orderByDesc(WorkflowRunEntity::getStartedAt)
|
||||
.last("LIMIT " + capped));
|
||||
return R.ok(rows);
|
||||
}
|
||||
|
||||
@Operation(summary = "Inspect a single run with its step rows for replay / debugging.")
|
||||
@GetMapping("/runs/{runId}")
|
||||
public R<RunDetail> getRun(@PathVariable long runId) {
|
||||
WorkflowRunEntity run = runMapper.selectById(runId);
|
||||
if (run == null) return R.fail("run not found: " + runId);
|
||||
List<WorkflowRunStepEntity> steps = stepMapper.selectList(new LambdaQueryWrapper<WorkflowRunStepEntity>()
|
||||
.eq(WorkflowRunStepEntity::getRunId, runId)
|
||||
.orderByAsc(WorkflowRunStepEntity::getStepIndex)
|
||||
.orderByAsc(WorkflowRunStepEntity::getIterationIndex));
|
||||
return R.ok(new RunDetail(run, steps));
|
||||
}
|
||||
|
||||
public record RunDetail(WorkflowRunEntity run, List<WorkflowRunStepEntity> steps) {}
|
||||
|
||||
private static R<CompileErrorResponse> buildCompileFailure(List<vip.mate.workflow.compiler.CompileError> errors) {
|
||||
R<CompileErrorResponse> r = new R<>();
|
||||
r.setCode(422);
|
||||
r.setMsg("compile failed");
|
||||
r.setData(CompileErrorResponse.of(errors));
|
||||
return r;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,10 @@
|
||||
package vip.mate.workflow.api;
|
||||
|
||||
/**
|
||||
* Request body for {@code PUT /api/v1/workflows/{id}/draft}. The wire format
|
||||
* matches {@code mate_workflow.draft_json} verbatim — the controller does
|
||||
* not reshape this before persisting, so the editor / API caller owns the
|
||||
* exact JSON the publish-time compiler will see.
|
||||
*/
|
||||
public record WorkflowDraftRequest(String draftJson) {
|
||||
}
|
||||
@ -0,0 +1,8 @@
|
||||
package vip.mate.workflow.api;
|
||||
|
||||
/**
|
||||
* Request body for {@code POST /api/v1/workflows/{id}/publish}. {@code note}
|
||||
* is the human-friendly publish note recorded on the new revision row.
|
||||
*/
|
||||
public record WorkflowPublishRequest(String note) {
|
||||
}
|
||||
@ -0,0 +1,66 @@
|
||||
package vip.mate.workflow.service;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import org.springframework.stereotype.Component;
|
||||
import vip.mate.agent.model.AgentEntity;
|
||||
import vip.mate.agent.repository.AgentMapper;
|
||||
import vip.mate.channel.model.ChannelEntity;
|
||||
import vip.mate.channel.repository.ChannelMapper;
|
||||
import vip.mate.workflow.compiler.WorkflowAclPort;
|
||||
|
||||
/**
|
||||
* Production binding for {@link WorkflowAclPort}. Reads agents from
|
||||
* {@code mate_agent}, channels from {@code mate_channel}, and treats every
|
||||
* non-blank {@code employeeId} as a workspace member — until a real
|
||||
* "human employee" registry exists in the system, the workflow's
|
||||
* {@code employeeId} is interpreted as the agent id of the agent that owns
|
||||
* the memory file.
|
||||
*/
|
||||
@Component
|
||||
public class DefaultWorkflowAclPort implements WorkflowAclPort {
|
||||
|
||||
private final AgentMapper agentMapper;
|
||||
private final ChannelMapper channelMapper;
|
||||
|
||||
public DefaultWorkflowAclPort(AgentMapper agentMapper, ChannelMapper channelMapper) {
|
||||
this.agentMapper = agentMapper;
|
||||
this.channelMapper = channelMapper;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean agentExists(long workspaceId, String agentName) {
|
||||
if (agentName == null || agentName.isBlank()) return false;
|
||||
Long count = agentMapper.selectCount(new LambdaQueryWrapper<AgentEntity>()
|
||||
.eq(AgentEntity::getName, agentName.trim())
|
||||
.eq(AgentEntity::getEnabled, true));
|
||||
return count != null && count > 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean agentIdExists(long workspaceId, long agentId) {
|
||||
AgentEntity row = agentMapper.selectById(agentId);
|
||||
return row != null && Boolean.TRUE.equals(row.getEnabled());
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean channelAllowed(long workspaceId, String channelName) {
|
||||
if (channelName == null || channelName.isBlank()) return false;
|
||||
Long count = channelMapper.selectCount(new LambdaQueryWrapper<ChannelEntity>()
|
||||
.eq(ChannelEntity::getChannelType, channelName.trim())
|
||||
.eq(ChannelEntity::getEnabled, true));
|
||||
return count != null && count > 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean employeeInWorkspace(long workspaceId, String employeeId) {
|
||||
if (employeeId == null || employeeId.isBlank()) return false;
|
||||
try {
|
||||
long parsed = Long.parseLong(employeeId);
|
||||
return agentIdExists(workspaceId, parsed);
|
||||
} catch (NumberFormatException e) {
|
||||
// Non-numeric employeeId — let the runtime fail loudly rather
|
||||
// than silently passing publish-time ACL.
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,127 @@
|
||||
package vip.mate.workflow.service;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import vip.mate.workflow.compiler.PublishContext;
|
||||
import vip.mate.workflow.compiler.WorkflowAclPort;
|
||||
import vip.mate.workflow.compiler.WorkflowCompiler;
|
||||
import vip.mate.workflow.model.WorkflowEntity;
|
||||
import vip.mate.workflow.model.WorkflowRevisionEntity;
|
||||
import vip.mate.workflow.repository.WorkflowMapper;
|
||||
import vip.mate.workflow.repository.WorkflowRevisionMapper;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Workflow CRUD + draft / publish lifecycle. Drafts live inline on the
|
||||
* {@code mate_workflow} row; publishing compiles the draft and writes a
|
||||
* fresh row into {@code mate_workflow_revision} with a monotonically
|
||||
* increasing per-workflow revision number, then atomically points
|
||||
* {@code latest_revision_id} at it.
|
||||
*/
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class WorkflowService {
|
||||
|
||||
private final WorkflowMapper workflowMapper;
|
||||
private final WorkflowRevisionMapper revisionMapper;
|
||||
private final WorkflowCompiler compiler;
|
||||
private final WorkflowAclPort aclPort;
|
||||
|
||||
public List<WorkflowEntity> listByWorkspace(long workspaceId) {
|
||||
return workflowMapper.selectList(new LambdaQueryWrapper<WorkflowEntity>()
|
||||
.eq(WorkflowEntity::getWorkspaceId, workspaceId)
|
||||
.orderByDesc(WorkflowEntity::getUpdateTime));
|
||||
}
|
||||
|
||||
public WorkflowEntity get(long id) {
|
||||
return workflowMapper.selectById(id);
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public WorkflowEntity create(WorkflowEntity workflow) {
|
||||
if (workflow.getEnabled() == null) workflow.setEnabled(true);
|
||||
workflowMapper.insert(workflow);
|
||||
return workflow;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public WorkflowEntity update(WorkflowEntity workflow) {
|
||||
WorkflowEntity existing = workflowMapper.selectById(workflow.getId());
|
||||
if (existing == null) {
|
||||
throw new IllegalArgumentException("workflow not found: " + workflow.getId());
|
||||
}
|
||||
// Preserve revision pointer — only the publish path moves it.
|
||||
workflow.setLatestRevisionId(existing.getLatestRevisionId());
|
||||
workflowMapper.updateById(workflow);
|
||||
return workflow;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public WorkflowEntity saveDraft(long id, String draftJson, Long updatedBy) {
|
||||
WorkflowEntity row = workflowMapper.selectById(id);
|
||||
if (row == null) throw new IllegalArgumentException("workflow not found: " + id);
|
||||
row.setDraftJson(draftJson);
|
||||
row.setDraftUpdatedAt(LocalDateTime.now());
|
||||
row.setDraftUpdatedBy(updatedBy);
|
||||
workflowMapper.updateById(row);
|
||||
return row;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public void delete(long id) {
|
||||
workflowMapper.deleteById(id);
|
||||
}
|
||||
|
||||
/**
|
||||
* Compile the workflow's current draft and persist it as a new revision
|
||||
* pointed at by {@code latest_revision_id}. Throws
|
||||
* {@link vip.mate.workflow.compiler.WorkflowCompileFailedException} when
|
||||
* the compiler reports any errors.
|
||||
*/
|
||||
@Transactional
|
||||
public PublishOutcome publish(long workflowId, Long publisherId, String publishedNote) {
|
||||
WorkflowEntity workflow = workflowMapper.selectById(workflowId);
|
||||
if (workflow == null) {
|
||||
throw new IllegalArgumentException("workflow not found: " + workflowId);
|
||||
}
|
||||
String draft = workflow.getDraftJson();
|
||||
if (draft == null || draft.isBlank()) {
|
||||
throw new IllegalStateException("cannot publish workflow " + workflowId
|
||||
+ " without a draft");
|
||||
}
|
||||
PublishContext ctx = new PublishContext(publisherId == null ? 0L : publisherId,
|
||||
workflow.getWorkspaceId());
|
||||
WorkflowCompiler.Result compileResult = compiler.compile(draft, ctx, aclPort);
|
||||
compileResult.requireOk();
|
||||
|
||||
int nextRevision = nextRevisionNumber(workflowId);
|
||||
WorkflowRevisionEntity revision = new WorkflowRevisionEntity();
|
||||
revision.setWorkflowId(workflowId);
|
||||
revision.setRevision(nextRevision);
|
||||
revision.setGraphJson(draft);
|
||||
revision.setSchemaVersion(compileResult.graph().schemaVersion() == null
|
||||
? "1.0" : compileResult.graph().schemaVersion());
|
||||
revision.setPublishedNote(publishedNote);
|
||||
revision.setPublishedBy(publisherId);
|
||||
revisionMapper.insert(revision);
|
||||
|
||||
workflow.setLatestRevisionId(revision.getId());
|
||||
workflowMapper.updateById(workflow);
|
||||
return new PublishOutcome(workflow, revision);
|
||||
}
|
||||
|
||||
private int nextRevisionNumber(long workflowId) {
|
||||
WorkflowRevisionEntity max = revisionMapper.selectOne(new LambdaQueryWrapper<WorkflowRevisionEntity>()
|
||||
.eq(WorkflowRevisionEntity::getWorkflowId, workflowId)
|
||||
.orderByDesc(WorkflowRevisionEntity::getRevision)
|
||||
.last("LIMIT 1"));
|
||||
return max == null ? 1 : max.getRevision() + 1;
|
||||
}
|
||||
|
||||
/** Snapshot returned to controllers after a successful publish. */
|
||||
public record PublishOutcome(WorkflowEntity workflow, WorkflowRevisionEntity revision) {}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user