149 lines
5.7 KiB
Go
149 lines
5.7 KiB
Go
package app
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"log"
|
||
|
||
"yingce/backend/internal/model"
|
||
"yingce/backend/internal/repository"
|
||
)
|
||
|
||
const (
|
||
// cloudAgentRunEventDeltaLimit 是增量读取（sinceSeq）单次返回的条数上限，
|
||
// 也是 HTTP 上 eventLimit 允许的最大值（越界的请求由 handler 拒绝）。
|
||
// 内部调用方不受它约束：续轮收束按 cloudAgentContinuationEventLimit 显式取更大的页。
|
||
cloudAgentRunEventDeltaLimit = 500
|
||
// cloudAgentRunEventPageLimit 是运行详情默认返回的事件条数：等于内存里的尾部窗口，
|
||
// 因此默认读取不额外查事件表；要更早的记录由客户端通过 eventLimit 显式索取。
|
||
cloudAgentRunEventPageLimit = repository.CloudAgentJournalWindow
|
||
// cloudAgentEventWindowSanityLimit 是内存事件窗口的健全上限（窗口本身加一次转移
|
||
// 新产生的事件），超过它说明窗口没有按 CloudAgentJournalWindow 载入。
|
||
cloudAgentEventWindowSanityLimit = 512
|
||
)
|
||
|
||
// cloudAgentRunEventsForView 组装运行详情要返回的事件。
|
||
//
|
||
// 事件全量在 cloud_agent_event_records（一行一条 EventJSON，主键 run_id + sequence），
|
||
// 内存里只有最近一窗（repository.CloudAgentJournalWindow），所以两条读路径都要显式
|
||
// 说明"这一页在整条日志里的位置"：
|
||
//
|
||
// - sinceSeq > 0：只返回该序号之后的增量（SSE 断线重连 / 滚动拉取），走
|
||
// `sequence > ? ORDER BY sequence LIMIT ?` 的 keyset 读，不载入更早的行。
|
||
// - sinceSeq == 0：返回最近 cloudAgentRunEventPageLimit 条；窗口不够时（eventLimit
|
||
// 大于窗口）再从事件表往前补一页，仍然只按 seq 读。
|
||
//
|
||
// 事件表里没有该运行的行时（升级前的旧检查点把事件放在 state_json 里），退回内存窗口，
|
||
// 保证老运行仍可读。
|
||
func (s *Service) cloudAgentRunEventsForView(userID string, run *model.CloudAgentExecution, state *cloudAgentRuntime, sinceSeq, limit int) []CloudAgentEvent {
|
||
if sinceSeq > 0 {
|
||
pageLimit := cloudAgentRunEventDeltaLimit
|
||
if limit > 0 && limit < pageLimit {
|
||
pageLimit = limit
|
||
}
|
||
rows, err := s.repo.CloudAgentEventRecords(userID, run.ID, sinceSeq, pageLimit)
|
||
if err != nil {
|
||
log.Printf("agent event delta %s: %v", run.ID, err)
|
||
rows = nil
|
||
}
|
||
events := make([]CloudAgentEvent, 0, pageLimit)
|
||
for _, row := range rows {
|
||
events = append(events, cloudAgentEventFromRecord(row))
|
||
}
|
||
// 数据库页已达到上限时不能再拼接内存窗口，否则可能超过请求的页大小，
|
||
// 或在表中间隔着未返回的历史记录直接跳到窗口尾部，制造序号缺口。
|
||
if len(events) == pageLimit {
|
||
return events
|
||
}
|
||
last := sinceSeq
|
||
if len(rows) > 0 {
|
||
last = rows[len(rows)-1].Sequence
|
||
}
|
||
// 表未填满这一页时，内存里可能还有本次转移尚未落库的事件，补在末尾。
|
||
for _, event := range state.Events {
|
||
if event.Seq > last {
|
||
events = append(events, event)
|
||
if len(events) == pageLimit {
|
||
break
|
||
}
|
||
}
|
||
}
|
||
if len(events) > 0 {
|
||
return events
|
||
}
|
||
// 表与窗口都没有更新的内容：按游标过滤内存窗口（旧检查点的运行走这里）。
|
||
pending := make([]CloudAgentEvent, 0, len(state.Events))
|
||
for _, event := range state.Events {
|
||
if event.Seq > sinceSeq {
|
||
pending = append(pending, event)
|
||
if len(pending) == pageLimit {
|
||
break
|
||
}
|
||
}
|
||
}
|
||
return pending
|
||
}
|
||
tail := state.Events
|
||
if limit <= 0 {
|
||
limit = cloudAgentRunEventPageLimit
|
||
}
|
||
if len(tail) >= limit {
|
||
return append([]CloudAgentEvent(nil), tail[len(tail)-limit:]...)
|
||
}
|
||
// 窗口已经是全量（水位 0）：没有更早的记录可补，直接返回。
|
||
if state.EventSeqBase == 0 {
|
||
return append([]CloudAgentEvent(nil), tail...)
|
||
}
|
||
rows, err := s.repo.CloudAgentEventRecordsBefore(userID, run.ID, state.EventSeqBase+1, limit-len(tail))
|
||
if err != nil {
|
||
log.Printf("agent event page %s: %v", run.ID, err)
|
||
return append([]CloudAgentEvent(nil), tail...)
|
||
}
|
||
events := make([]CloudAgentEvent, 0, len(rows)+len(tail))
|
||
for _, row := range rows {
|
||
events = append(events, cloudAgentEventFromRecord(row))
|
||
}
|
||
return append(events, tail...)
|
||
}
|
||
|
||
// cloudAgentEventFromRecord 把事件行解析成对外契约结构。
|
||
func cloudAgentEventFromRecord(row model.CloudAgentEventRecord) CloudAgentEvent {
|
||
var event CloudAgentEvent
|
||
if err := json.Unmarshal([]byte(row.EventJSON), &event); err != nil || event.EventID == "" {
|
||
// 行存在但解不出来：仍按行给出身份与占位类型，避免整页读取失败。
|
||
// 这条事件不代表运行的真实动作，客户端应以服务端画布与任务状态为准。
|
||
return CloudAgentEvent{
|
||
EventID: fmt.Sprintf("%s:%d", row.RunID, row.Sequence), RunID: row.RunID, Seq: row.Sequence,
|
||
Type: "unreadable_event", Payload: map[string]any{"text": "该事件记录无法解析，请以服务端画布与任务状态为准"}, CreatedAt: row.CreatedAt,
|
||
}
|
||
}
|
||
if event.RunID == "" {
|
||
event.RunID = row.RunID
|
||
}
|
||
if event.Seq == 0 {
|
||
event.Seq = row.Sequence
|
||
}
|
||
if event.CreatedAt.IsZero() {
|
||
event.CreatedAt = row.CreatedAt
|
||
}
|
||
return event
|
||
}
|
||
|
||
// cloudAgentRunEventCount 给出"这个运行一共产生过多少事件"：以事件表条数为准，再补上
|
||
// 本次转移还没落库的部分（内存里 seq 高于水位的窗口）与检查点水位，取三者最大值。
|
||
func (s *Service) cloudAgentRunEventCount(userID string, run *model.CloudAgentExecution, state *cloudAgentRuntime) int {
|
||
stored, err := s.repo.CloudAgentEventRecordCount(userID, run.ID)
|
||
if err != nil {
|
||
log.Printf("agent event count %s: %v", run.ID, err)
|
||
stored = 0
|
||
}
|
||
total := int(stored)
|
||
if pending := state.EventSeqBase + len(state.Events); pending > total {
|
||
total = pending
|
||
}
|
||
if run != nil && run.EventCount > total {
|
||
total = run.EventCount
|
||
}
|
||
return total
|
||
}
|