feat(jobs/runners): workspace fetch runner with notify dispatch on failure

2026-05-05 23:35:58 +08:00
parent e361670539
commit af1855b753
3 changed files with 223 additions and 0 deletions
@@ -0,0 +1,119 @@
+// Package runners contains periodic background runners. Each file implements
+// one periodic task — runners are NOT job types; they tick directly via the
+// scheduler and do their work synchronously per tick.
+package runners
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/yan1h/agent-coding-workflow/internal/infra/notify"
+)
+
+// WorkspaceFetchTarget is the projection of a workspaces row needed by the
+// fetch runner.
+type WorkspaceFetchTarget struct {
+	ID       uuid.UUID
+	OwnerID  uuid.UUID
+	Name     string
+	MainPath string
+}
+
+// WorkspaceFetchRepo is the minimal repository surface needed by this runner.
+// In production the workspace.PgRepository implements it.
+type WorkspaceFetchRepo interface {
+	ListFetchTargets(ctx context.Context) ([]WorkspaceFetchTarget, error)
+	MarkSyncing(ctx context.Context, id uuid.UUID) (bool, error)
+	MarkFetchResult(ctx context.Context, id uuid.UUID, success bool, errMsg string) error
+}
+
+// WorkspaceFetcher abstracts the underlying git.Runner.Fetch path. The
+// production implementation lives in workspace.GitOpsService and resolves
+// credentials per-workspace.
+type WorkspaceFetcher interface {
+	Fetch(ctx context.Context, mainPath string, workspaceID uuid.UUID) error
+}
+
+// Dispatcher narrows notify.Dispatcher to what runners need.
+type Dispatcher interface {
+	Dispatch(ctx context.Context, msg notify.Message) error
+}
+
+// WorkspaceFetch is the periodic A runner.
+type WorkspaceFetch struct {
+	repo WorkspaceFetchRepo
+	gitr WorkspaceFetcher
+	disp Dispatcher
+	log  *slog.Logger
+}
+
+// NewWorkspaceFetch builds the runner.
+func NewWorkspaceFetch(repo WorkspaceFetchRepo, gitr WorkspaceFetcher, disp Dispatcher, log *slog.Logger) *WorkspaceFetch {
+	if log == nil {
+		log = slog.Default()
+	}
+	return &WorkspaceFetch{repo: repo, gitr: gitr, disp: disp, log: log}
+}
+
+// Run scans workspaces and synchronously fetches each. Sequential by design
+// (avoid disk IO contention). Idempotent: status='cloning'/'syncing' rows are
+// skipped because ListFetchTargets filters them out.
+func (r *WorkspaceFetch) Run(ctx context.Context) {
+	targets, err := r.repo.ListFetchTargets(ctx)
+	if err != nil {
+		r.log.Error("workspace_fetch.list", "err", err.Error())
+		return
+	}
+	for _, t := range targets {
+		r.fetchOne(ctx, t)
+	}
+}
+
+func (r *WorkspaceFetch) fetchOne(ctx context.Context, t WorkspaceFetchTarget) {
+	ok, err := r.repo.MarkSyncing(ctx, t.ID)
+	if err != nil {
+		r.log.Error("workspace_fetch.mark_syncing", "ws_id", t.ID, "err", err.Error())
+		return
+	}
+	if !ok {
+		// 别的 runner / 手工同步抢先了
+		return
+	}
+	if fetchErr := r.gitr.Fetch(ctx, t.MainPath, t.ID); fetchErr != nil {
+		msg := truncate(fetchErr.Error(), 2000)
+		if err := r.repo.MarkFetchResult(ctx, t.ID, false, msg); err != nil {
+			r.log.Error("workspace_fetch.mark_fail", "ws_id", t.ID, "err", err.Error())
+			return
+		}
+		if derr := r.disp.Dispatch(ctx, notify.Message{
+			UserID:   t.OwnerID,
+			Topic:    "workspace.sync_failed",
+			Severity: notify.SeverityError,
+			Title:    "Workspace sync failed",
+			Body:     "fetch failed: " + msg,
+			Metadata: map[string]any{
+				"workspace_id":   t.ID.String(),
+				"workspace_name": t.Name,
+				"error":          msg,
+			},
+			CreatedAt: time.Now().UTC(),
+		}); derr != nil {
+			r.log.Error("workspace_fetch.dispatch", "ws_id", t.ID, "err", derr.Error())
+		}
+		r.log.Warn("workspace_fetch.failed", "ws_id", t.ID, "err", msg)
+		return
+	}
+	if err := r.repo.MarkFetchResult(ctx, t.ID, true, ""); err != nil {
+		r.log.Error("workspace_fetch.mark_ok", "ws_id", t.ID, "err", err.Error())
+	}
+}
+
+func truncate(s string, n int) string {
+	if len(s) <= n {
+		return s
+	}
+	return s[:n]
+}