diff --git a/internal/jobs/runners/helpers_test.go b/internal/jobs/runners/helpers_test.go new file mode 100644 index 0000000..58f334d --- /dev/null +++ b/internal/jobs/runners/helpers_test.go @@ -0,0 +1,10 @@ +package runners_test + +import ( + "io" + "log/slog" +) + +func discardLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} diff --git a/internal/jobs/runners/workspace_fetch.go b/internal/jobs/runners/workspace_fetch.go new file mode 100644 index 0000000..591f9cf --- /dev/null +++ b/internal/jobs/runners/workspace_fetch.go @@ -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] +} diff --git a/internal/jobs/runners/workspace_fetch_test.go b/internal/jobs/runners/workspace_fetch_test.go new file mode 100644 index 0000000..0093fd1 --- /dev/null +++ b/internal/jobs/runners/workspace_fetch_test.go @@ -0,0 +1,94 @@ +package runners_test + +import ( + "context" + "errors" + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/yan1h/agent-coding-workflow/internal/infra/notify" + "github.com/yan1h/agent-coding-workflow/internal/jobs/runners" +) + +// fakeWorkspaceFetchRepo: inline mock holding a few workspaces + recorded calls. +type fakeWorkspaceFetchRepo struct { + listed []runners.WorkspaceFetchTarget + startCAS map[uuid.UUID]bool + finished map[uuid.UUID]string +} + +func (f *fakeWorkspaceFetchRepo) ListFetchTargets(_ context.Context) ([]runners.WorkspaceFetchTarget, error) { + return f.listed, nil +} + +func (f *fakeWorkspaceFetchRepo) MarkSyncing(_ context.Context, id uuid.UUID) (bool, error) { + if f.startCAS == nil { + f.startCAS = map[uuid.UUID]bool{} + } + f.startCAS[id] = true + return true, nil +} + +func (f *fakeWorkspaceFetchRepo) MarkFetchResult(_ context.Context, id uuid.UUID, success bool, errMsg string) error { + if f.finished == nil { + f.finished = map[uuid.UUID]string{} + } + if success { + f.finished[id] = "ok" + } else { + f.finished[id] = errMsg + } + return nil +} + +type fakeFetcher struct{ err error } + +func (f *fakeFetcher) Fetch(_ context.Context, _ string, _ uuid.UUID) error { return f.err } + +type recordingDispatcher struct{ msgs []notify.Message } + +func (r *recordingDispatcher) Dispatch(_ context.Context, m notify.Message) error { + r.msgs = append(r.msgs, m) + return nil +} + +func TestWorkspaceFetch_SuccessUpdatesIdle(t *testing.T) { + t.Parallel() + id := uuid.New() + ownerID := uuid.New() + repo := &fakeWorkspaceFetchRepo{listed: []runners.WorkspaceFetchTarget{ + {ID: id, OwnerID: ownerID, Name: "ws", MainPath: "/x"}, + }} + r := runners.NewWorkspaceFetch(repo, &fakeFetcher{}, &recordingDispatcher{}, discardLogger()) + r.Run(context.Background()) + assert.True(t, repo.startCAS[id]) + assert.Equal(t, "ok", repo.finished[id]) +} + +func TestWorkspaceFetch_FailureMarksErrorAndDispatches(t *testing.T) { + t.Parallel() + id := uuid.New() + ownerID := uuid.New() + repo := &fakeWorkspaceFetchRepo{listed: []runners.WorkspaceFetchTarget{ + {ID: id, OwnerID: ownerID, Name: "ws", MainPath: "/x"}, + }} + disp := &recordingDispatcher{} + r := runners.NewWorkspaceFetch(repo, &fakeFetcher{err: errors.New("network")}, disp, discardLogger()) + r.Run(context.Background()) + assert.Contains(t, repo.finished[id], "network") + require.Len(t, disp.msgs, 1) + assert.Equal(t, "workspace.sync_failed", disp.msgs[0].Topic) + assert.Equal(t, ownerID, disp.msgs[0].UserID) + assert.Equal(t, notify.SeverityError, disp.msgs[0].Severity) +} + +func TestWorkspaceFetch_NoTargetsNoOp(t *testing.T) { + t.Parallel() + repo := &fakeWorkspaceFetchRepo{} + r := runners.NewWorkspaceFetch(repo, &fakeFetcher{}, &recordingDispatcher{}, discardLogger()) + r.Run(context.Background()) + assert.Empty(t, repo.startCAS) +}