From 3c03d1f4820197914c5d794a02b05a643af292bf Mon Sep 17 00:00:00 2001 From: viettranx Date: Mon, 9 Mar 2026 09:44:57 +0700 Subject: [PATCH] feat(cron): add detail page, running status, realtime events, and pagination MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add /cron/:id detail page with job info, payload, run history - Make cron.run async: respond immediately, execute in background - Set last_status="running" before execution, emit CronEvent via WS - Add CronEvent (running/completed/error) broadcast to all WS clients - Add server-side pagination to GetRunLog (offset + total count) - Show loading spinner on Run button when job is running (list + detail) - Enrich CronRunLogEntry with duration, input/output tokens - Make job names clickable in overview card → /cron/:id - Fix refresh button animation using isFetching instead of isPending --- cmd/gateway.go | 16 +- internal/gateway/methods/cron.go | 36 ++- internal/store/cron_store.go | 25 +- internal/store/pg/cron.go | 75 +++-- internal/store/pg/cron_scheduler.go | 8 + internal/tools/cron.go | 3 +- ui/web/src/lib/constants.ts | 1 + ui/web/src/pages/cron/cron-detail-page.tsx | 284 +++++++++++++++++++ ui/web/src/pages/cron/cron-page.tsx | 47 ++- ui/web/src/pages/cron/hooks/use-cron.ts | 16 +- ui/web/src/pages/overview/cron-jobs-card.tsx | 9 +- ui/web/src/routes.tsx | 1 + 12 files changed, 463 insertions(+), 58 deletions(-) create mode 100644 ui/web/src/pages/cron/cron-detail-page.tsx diff --git a/cmd/gateway.go b/cmd/gateway.go index b65d559a..4846e712 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -7,6 +7,7 @@ import ( "os" "os/signal" "path/filepath" + "strings" "syscall" "github.com/google/uuid" @@ -795,9 +796,17 @@ func runGateway() { // Wire pairing approval notification → channel (matching TS notifyPairingApproved). botName := cfg.ResolveDisplayName("default") - pairingMethods.SetOnApprove(func(ctx context.Context, channel, chatID string) { + pairingMethods.SetOnApprove(func(ctx context.Context, channel, chatID, senderID string) { msg := fmt.Sprintf("✅ %s access approved. Send a message to start chatting.", botName) - if err := channelMgr.SendToChannel(ctx, channel, chatID, msg); err != nil { + // Group pairings need group_id metadata so channels (e.g. Zalo) route to group API. + if strings.HasPrefix(senderID, "group:") { + msgBus.PublishOutbound(bus.OutboundMessage{ + Channel: channel, + ChatID: chatID, + Content: msg, + Metadata: map[string]string{"group_id": chatID}, + }) + } else if err := channelMgr.SendToChannel(ctx, channel, chatID, msg); err != nil { slog.Warn("failed to send pairing approval notification", "channel", channel, "chatID", chatID, "error", err) } }) @@ -892,6 +901,9 @@ func runGateway() { // Start cron service with job handler (routes through scheduler's cron lane) pgStores.Cron.SetOnJob(makeCronJobHandler(sched, msgBus, cfg, channelMgr)) + pgStores.Cron.SetOnEvent(func(event store.CronEvent) { + server.BroadcastEvent(*protocol.NewEvent(protocol.EventCron, event)) + }) if err := pgStores.Cron.Start(); err != nil { slog.Warn("cron service failed to start", "error", err) } diff --git a/internal/gateway/methods/cron.go b/internal/gateway/methods/cron.go index 3275477f..5506a5aa 100644 --- a/internal/gateway/methods/cron.go +++ b/internal/gateway/methods/cron.go @@ -3,6 +3,7 @@ package methods import ( "context" "encoding/json" + "log/slog" "regexp" "github.com/nextlevelbuilder/goclaw/internal/gateway" @@ -188,27 +189,33 @@ func (m *CronMethods) handleRun(_ context.Context, client *gateway.Client, req * } force := params.Mode == "force" - ran, reason, err := m.service.RunJob(jobID, force) - if err != nil { - client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error())) + + // Validate job exists before responding + _, ok := m.service.GetJob(jobID) + if !ok { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "job not found")) return } - resp := map[string]interface{}{ + // Respond immediately — job execution happens in background + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{ "ok": true, - "ran": ran, - } - if !ran && reason != "" { - resp["reason"] = reason - } - client.SendResponse(protocol.NewOKResponse(req.ID, resp)) + "ran": true, + })) + + go func() { + if _, _, err := m.service.RunJob(jobID, force); err != nil { + slog.Warn("cron.run background error", "jobId", jobID, "error", err) + } + }() } func (m *CronMethods) handleRuns(_ context.Context, client *gateway.Client, req *protocol.RequestFrame) { var params struct { - JobID string `json:"jobId"` - ID string `json:"id"` - Limit int `json:"limit"` + JobID string `json:"jobId"` + ID string `json:"id"` + Limit int `json:"limit"` + Offset int `json:"offset"` } if req.Params != nil { json.Unmarshal(req.Params, ¶ms) @@ -219,8 +226,9 @@ func (m *CronMethods) handleRuns(_ context.Context, client *gateway.Client, req jobID = params.ID } - entries := m.service.GetRunLog(jobID, params.Limit) + entries, total := m.service.GetRunLog(jobID, params.Limit, params.Offset) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{ "entries": entries, + "total": total, })) } diff --git a/internal/store/cron_store.go b/internal/store/cron_store.go index 5e008c69..0cb14868 100644 --- a/internal/store/cron_store.go +++ b/internal/store/cron_store.go @@ -46,11 +46,14 @@ type CronJobState struct { // CronRunLogEntry records a job execution. type CronRunLogEntry struct { - Ts int64 `json:"ts"` - JobID string `json:"jobId"` - Status string `json:"status,omitempty"` - Error string `json:"error,omitempty"` - Summary string `json:"summary,omitempty"` + Ts int64 `json:"ts"` + JobID string `json:"jobId"` + Status string `json:"status,omitempty"` + Error string `json:"error,omitempty"` + Summary string `json:"summary,omitempty"` + DurationMS int64 `json:"durationMs,omitempty"` + InputTokens int `json:"inputTokens,omitempty"` + OutputTokens int `json:"outputTokens,omitempty"` } // CronJobResult is the output of a cron job handler execution. @@ -74,6 +77,15 @@ type CronJobPatch struct { DeleteAfterRun *bool `json:"deleteAfterRun,omitempty"` } +// CronEvent represents a job lifecycle event sent to subscribers. +type CronEvent struct { + Action string `json:"action"` // "running", "completed", "error" + JobID string `json:"jobId"` + JobName string `json:"jobName,omitempty"` + Status string `json:"status,omitempty"` // final status for completed/error + Error string `json:"error,omitempty"` +} + // CronStore manages scheduled jobs. type CronStore interface { AddJob(name string, schedule CronSchedule, message string, deliver bool, channel, to, agentID, userID string) (*CronJob, error) @@ -82,7 +94,7 @@ type CronStore interface { RemoveJob(jobID string) error UpdateJob(jobID string, patch CronJobPatch) (*CronJob, error) EnableJob(jobID string, enabled bool) error - GetRunLog(jobID string, limit int) []CronRunLogEntry + GetRunLog(jobID string, limit, offset int) ([]CronRunLogEntry, int) Status() map[string]interface{} // Lifecycle @@ -91,6 +103,7 @@ type CronStore interface { // Job execution SetOnJob(handler func(job *CronJob) (*CronJobResult, error)) + SetOnEvent(handler func(event CronEvent)) RunJob(jobID string, force bool) (ran bool, reason string, err error) // Due job detection (for scheduler) diff --git a/internal/store/pg/cron.go b/internal/store/pg/cron.go index f3e0d137..0227925e 100644 --- a/internal/store/pg/cron.go +++ b/internal/store/pg/cron.go @@ -24,6 +24,7 @@ type PGCronStore struct { db *sql.DB mu sync.Mutex onJob func(job *store.CronJob) (*store.CronJobResult, error) + onEvent func(event store.CronEvent) running bool stop chan struct{} @@ -369,27 +370,35 @@ func (s *PGCronStore) EnableJob(jobID string, enabled bool) error { return nil } -func (s *PGCronStore) GetRunLog(jobID string, limit int) []store.CronRunLogEntry { +func (s *PGCronStore) GetRunLog(jobID string, limit, offset int) ([]store.CronRunLogEntry, int) { if limit <= 0 { limit = 20 } + if offset < 0 { + offset = 0 + } + const cols = "job_id, status, error, summary, ran_at, COALESCE(duration_ms, 0), COALESCE(input_tokens, 0), COALESCE(output_tokens, 0)" + + var total int var rows *sql.Rows var err error if jobID != "" { id, parseErr := uuid.Parse(jobID) if parseErr != nil { - return nil + return nil, 0 } + s.db.QueryRow("SELECT COUNT(*) FROM cron_run_logs WHERE job_id = $1", id).Scan(&total) rows, err = s.db.Query( - "SELECT job_id, status, error, summary, ran_at FROM cron_run_logs WHERE job_id = $1 ORDER BY ran_at DESC LIMIT $2", - id, limit) + "SELECT "+cols+" FROM cron_run_logs WHERE job_id = $1 ORDER BY ran_at DESC LIMIT $2 OFFSET $3", + id, limit, offset) } else { + s.db.QueryRow("SELECT COUNT(*) FROM cron_run_logs").Scan(&total) rows, err = s.db.Query( - "SELECT job_id, status, error, summary, ran_at FROM cron_run_logs ORDER BY ran_at DESC LIMIT $1", limit) + "SELECT "+cols+" FROM cron_run_logs ORDER BY ran_at DESC LIMIT $1 OFFSET $2", limit, offset) } if err != nil { - return nil + return nil, 0 } defer rows.Close() @@ -399,18 +408,23 @@ func (s *PGCronStore) GetRunLog(jobID string, limit int) []store.CronRunLogEntry var status string var errStr, summary *string var ranAt time.Time - if err := rows.Scan(&jobUUID, &status, &errStr, &summary, &ranAt); err != nil { + var durationMS int64 + var inputTokens, outputTokens int + if err := rows.Scan(&jobUUID, &status, &errStr, &summary, &ranAt, &durationMS, &inputTokens, &outputTokens); err != nil { continue } result = append(result, store.CronRunLogEntry{ - Ts: ranAt.UnixMilli(), - JobID: jobUUID.String(), - Status: status, - Error: derefStr(errStr), - Summary: derefStr(summary), + Ts: ranAt.UnixMilli(), + JobID: jobUUID.String(), + Status: status, + Error: derefStr(errStr), + Summary: derefStr(summary), + DurationMS: durationMS, + InputTokens: inputTokens, + OutputTokens: outputTokens, }) } - return result + return result, total } func (s *PGCronStore) Status() map[string]interface{} { @@ -454,6 +468,21 @@ func (s *PGCronStore) SetOnJob(handler func(job *store.CronJob) (*store.CronJobR s.onJob = handler } +func (s *PGCronStore) SetOnEvent(handler func(event store.CronEvent)) { + s.mu.Lock() + defer s.mu.Unlock() + s.onEvent = handler +} + +func (s *PGCronStore) emitEvent(event store.CronEvent) { + s.mu.Lock() + fn := s.onEvent + s.mu.Unlock() + if fn != nil { + fn(event) + } +} + func (s *PGCronStore) RunJob(jobID string, force bool) (bool, string, error) { job, ok := s.GetJob(jobID) if !ok { @@ -468,11 +497,21 @@ func (s *PGCronStore) RunJob(jobID string, force bool) (bool, string, error) { return false, "", fmt.Errorf("no job handler configured") } - result, err := handler(job) - content := "" - if result != nil { - content = result.Content + // Mark job as running before execution + if id, parseErr := uuid.Parse(jobID); parseErr == nil { + s.db.Exec("UPDATE cron_jobs SET last_status = 'running', updated_at = $1 WHERE id = $2", time.Now(), id) } - return true, content, err + s.mu.Lock() + s.cacheLoaded = false + s.mu.Unlock() + + s.emitEvent(store.CronEvent{Action: "running", JobID: job.ID, JobName: job.Name}) + + // Use executeOneJob for proper state updates, run logging, and retry + s.executeOneJob(*job, handler) + s.mu.Lock() + s.cacheLoaded = false + s.mu.Unlock() + return true, "", nil } diff --git a/internal/store/pg/cron_scheduler.go b/internal/store/pg/cron_scheduler.go index 7cf16e99..5663aa0c 100644 --- a/internal/store/pg/cron_scheduler.go +++ b/internal/store/pg/cron_scheduler.go @@ -242,4 +242,12 @@ func (s *PGCronStore) executeOneJob(job store.CronJob, handler func(job *store.C now, status, lastError, next, now, id, ) } + + // Emit completion event + evt := store.CronEvent{Action: "completed", JobID: job.ID, JobName: job.Name, Status: status} + if err != nil { + evt.Action = "error" + evt.Error = err.Error() + } + s.emitEvent(evt) } diff --git a/internal/tools/cron.go b/internal/tools/cron.go index 9b309320..7e60aa3f 100644 --- a/internal/tools/cron.go +++ b/internal/tools/cron.go @@ -366,11 +366,12 @@ func (t *CronTool) handleRuns(args map[string]interface{}, agentID, userID strin limit = int(v) } - entries := t.cronStore.GetRunLog(jobID, limit) + entries, total := t.cronStore.GetRunLog(jobID, limit, 0) result := map[string]interface{}{ "entries": entries, "count": len(entries), + "total": total, } data, _ := json.MarshalIndent(result, "", " ") return NewResult(string(data)) diff --git a/ui/web/src/lib/constants.ts b/ui/web/src/lib/constants.ts index 8073de5a..29438d25 100644 --- a/ui/web/src/lib/constants.ts +++ b/ui/web/src/lib/constants.ts @@ -10,6 +10,7 @@ export const ROUTES = { SKILLS: "/skills", SKILL_DETAIL: "/skills/:id", CRON: "/cron", + CRON_DETAIL: "/cron/:id", CONFIG: "/config", TRACES: "/traces", TRACE_DETAIL: "/traces/:id", diff --git a/ui/web/src/pages/cron/cron-detail-page.tsx b/ui/web/src/pages/cron/cron-detail-page.tsx new file mode 100644 index 00000000..2461536a --- /dev/null +++ b/ui/web/src/pages/cron/cron-detail-page.tsx @@ -0,0 +1,284 @@ +import { useState, useEffect, useCallback } from "react"; +import { ArrowLeft, Play, Trash2, Power, Loader2 } from "lucide-react"; +import { Button } from "@/components/ui/button"; +import { Badge } from "@/components/ui/badge"; +import { ConfirmDialog } from "@/components/shared/confirm-dialog"; +import { Pagination } from "@/components/shared/pagination"; +import { formatDate, formatTokens } from "@/lib/format"; +import type { CronJob, CronRunLogEntry } from "./hooks/use-cron"; + +function formatScheduleDetail(job: CronJob): string { + const s = job.schedule; + if (s.kind === "every" && s.everyMs) { + const sec = s.everyMs / 1000; + if (sec < 60) return `Every ${sec} seconds`; + if (sec < 3600) return `Every ${Math.round(sec / 60)} minutes`; + return `Every ${(sec / 3600).toFixed(1)} hours`; + } + if (s.kind === "cron" && s.expr) return s.expr; + if (s.kind === "at" && s.atMs) return `Once at ${new Date(s.atMs).toLocaleString()}`; + return s.kind; +} + +function formatDuration(ms?: number): string { + if (!ms) return "-"; + if (ms < 1000) return `${ms}ms`; + if (ms < 60000) return `${(ms / 1000).toFixed(1)}s`; + return `${(ms / 60000).toFixed(1)}m`; +} + +interface CronDetailPageProps { + job: CronJob; + onBack: () => void; + onRun: (id: string) => Promise; + onToggle: (id: string, enabled: boolean) => Promise; + onDelete: (id: string) => Promise; + getRunLog: (id: string, limit?: number, offset?: number) => Promise<{ entries: CronRunLogEntry[]; total: number }>; + onRefresh: () => void; +} + +export function CronDetailPage({ job, onBack, onRun, onToggle, onDelete, getRunLog, onRefresh }: CronDetailPageProps) { + const [confirmDelete, setConfirmDelete] = useState(false); + const [confirmToggle, setConfirmToggle] = useState(false); + const [runLog, setRunLog] = useState([]); + const [runLogTotal, setRunLogTotal] = useState(0); + const [runLogLoading, setRunLogLoading] = useState(true); + const [running, setRunning] = useState(false); + const isRunning = running || job.state?.lastStatus === "running"; + const [runLogPage, setRunLogPage] = useState(1); + const [runLogPageSize, setRunLogPageSize] = useState(10); + const loadRunLog = useCallback(async (page?: number, pageSize?: number) => { + const p = page ?? runLogPage; + const ps = pageSize ?? runLogPageSize; + setRunLogLoading(true); + try { + const { entries, total } = await getRunLog(job.id, ps, (p - 1) * ps); + setRunLog(entries); + setRunLogTotal(total); + } finally { + setRunLogLoading(false); + } + }, [job.id, getRunLog, runLogPage, runLogPageSize]); + + const runLogTotalPages = Math.ceil(runLogTotal / runLogPageSize); + + useEffect(() => { + loadRunLog(); + }, [loadRunLog]); + + // Poll cron list while job is running to detect completion + useEffect(() => { + if (!isRunning) return; + const interval = setInterval(onRefresh, 3000); + return () => clearInterval(interval); + }, [isRunning, onRefresh]); + + // Clear local running state when backend status changes from running + useEffect(() => { + if (running && job.state?.lastStatus && job.state.lastStatus !== "running") { + setRunning(false); + loadRunLog(1); + } + }, [running, job.state?.lastStatus, loadRunLog]); + + return ( +
+ {/* Header */} +
+
+ +
+

+ {job.name} + + {job.enabled ? "enabled" : "disabled"} + +

+
+ {job.schedule.kind} + {job.agentId && {job.agentId}} + {job.state?.lastStatus && ( + + {job.state.lastStatus === "running" ? "running..." : job.state.lastStatus} + + )} +
+
+
+
+ + + +
+
+ + {/* Content */} +
+
+ {/* Job Info */} +
+ + {job.schedule.tz && } + {job.state?.nextRunAtMs && ( + + )} + {job.state?.lastRunAtMs && ( + + )} + + + {job.deleteAfterRun && } +
+ + {/* Payload */} +
+

Payload

+
+
+ {job.payload?.message || "(empty)"} +
+
+ {job.payload?.deliver && Deliver: direct} + {job.payload?.channel && Channel: {job.payload.channel}} + {job.payload?.to && To: {job.payload.to}} +
+
+
+ + {/* Last Error */} + {job.state?.lastError && ( +
+

Last Error

+

{job.state.lastError}

+
+ )} + + {/* Run History */} +
+
+

Run History

+ +
+ {runLogLoading && runLog.length === 0 ? ( +
+
+
+ ) : runLog.length === 0 ? ( +

No run history yet.

+ ) : ( + <> +
+ {runLog.map((entry, i) => ( +
+
+
+ {formatDate(new Date(entry.ts))} + {entry.durationMs != null && entry.durationMs > 0 && ( + ({formatDuration(entry.durationMs)}) + )} +
+
+ {(entry.inputTokens != null && entry.inputTokens > 0) && ( + + {formatTokens(entry.inputTokens)} in / {formatTokens(entry.outputTokens ?? 0)} out + + )} + + {entry.status || "unknown"} + +
+
+ {entry.summary && ( +

{entry.summary}

+ )} + {entry.error && ( +

{entry.error}

+ )} +
+ ))} +
+ { setRunLogPage(p); loadRunLog(p); }} + onPageSizeChange={(s) => { setRunLogPageSize(s); setRunLogPage(1); loadRunLog(1, s); }} + pageSizes={[10, 20, 50]} + /> + + )} +
+
+
+ + { + await onDelete(job.id); + setConfirmDelete(false); + }} + /> + + { + await onToggle(job.id, !job.enabled); + setConfirmToggle(false); + }} + /> +
+ ); +} + +function InfoRow({ label, value }: { label: string; value: string }) { + return ( +
+ {label} +
{value}
+
+ ); +} diff --git a/ui/web/src/pages/cron/cron-page.tsx b/ui/web/src/pages/cron/cron-page.tsx index c0c96f98..fb763005 100644 --- a/ui/web/src/pages/cron/cron-page.tsx +++ b/ui/web/src/pages/cron/cron-page.tsx @@ -1,5 +1,6 @@ import { useState } from "react"; -import { Clock, Plus, Play, Trash2, History, RefreshCw } from "lucide-react"; +import { useParams, useNavigate } from "react-router"; +import { Clock, Plus, Play, Trash2, History, RefreshCw, Loader2 } from "lucide-react"; import { Button } from "@/components/ui/button"; import { Badge } from "@/components/ui/badge"; import { Switch } from "@/components/ui/switch"; @@ -11,6 +12,7 @@ import { ConfirmDialog } from "@/components/shared/confirm-dialog"; import { useCron, type CronJob, type CronRunLogEntry } from "./hooks/use-cron"; import { CronFormDialog } from "./cron-form-dialog"; import { CronRunLogDialog } from "./cron-run-log-dialog"; +import { CronDetailPage } from "./cron-detail-page"; import { useMinLoading } from "@/hooks/use-min-loading"; import { useDeferredLoading } from "@/hooks/use-deferred-loading"; import { usePagination } from "@/hooks/use-pagination"; @@ -29,8 +31,10 @@ function formatSchedule(job: CronJob): string { } export function CronPage() { - const { jobs, loading, refresh, createJob, toggleJob, deleteJob, runJob, getRunLog } = useCron(); - const spinning = useMinLoading(loading); + const { id: detailId } = useParams<{ id: string }>(); + const navigate = useNavigate(); + const { jobs, loading, refreshing, refresh, createJob, toggleJob, deleteJob, runJob, getRunLog } = useCron(); + const spinning = useMinLoading(refreshing); const showSkeleton = useDeferredLoading(loading && jobs.length === 0); const [showCreate, setShowCreate] = useState(false); const [deleteTarget, setDeleteTarget] = useState(null); @@ -41,11 +45,29 @@ export function CronPage() { const { pageItems, pagination, setPage, setPageSize } = usePagination(jobs); + const detailJob = detailId ? jobs.find((j) => j.id === detailId) : null; + if (detailJob) { + return ( + navigate("/cron")} + onRun={runJob} + onToggle={toggleJob} + onDelete={async (id) => { + await deleteJob(id); + navigate("/cron"); + }} + getRunLog={getRunLog} + onRefresh={refresh} + /> + ); + } + const handleShowRunLog = async (job: CronJob) => { setRunLogTarget(job); setRunLogLoading(true); try { - const entries = await getRunLog(job.id); + const { entries } = await getRunLog(job.id); setRunLogEntries(entries); } finally { setRunLogLoading(false); @@ -105,7 +127,15 @@ export function CronPage() { onCheckedChange={(checked: boolean) => setToggleTarget({ job, enabled: checked })} /> - {job.name} + + + {formatSchedule(job)} @@ -124,10 +154,13 @@ export function CronPage() {
{job.enabled && job.state.nextRunAtMs ? ( diff --git a/ui/web/src/routes.tsx b/ui/web/src/routes.tsx index 9a4815c0..ab93c741 100644 --- a/ui/web/src/routes.tsx +++ b/ui/web/src/routes.tsx @@ -126,6 +126,7 @@ export function AppRoutes() { } /> } /> } /> + } /> } /> } /> } />