From d712ac0d229df478d9de9920089efab02d67d27d Mon Sep 17 00:00:00 2001 From: Clifford Tawiah Date: Thu, 24 Sep 2026 13:55:55 -0400 Subject: [PATCH] feat(sync): orchestrate prompt synchronization --- go.mod | 2 +- internal/sync/interactive/form.go | 30 +- internal/sync/prompt/conflict.go | 229 ++++++++++++++ internal/sync/prompt/conflict_test.go | 235 ++++++++++++++ internal/sync/prompt/runner.go | 334 ++++++++++++++++++++ internal/sync/prompt/runner_test.go | 221 +++++++++++++ internal/sync/prompt/terminal.go | 56 ++++ internal/sync/prompt/terminal_test.go | 39 +++ internal/sync/prompt/watch.go | 434 ++++++++++++++++++++++++++ internal/sync/prompt/watch_test.go | 434 ++++++++++++++++++++++++++ 10 files changed, 2009 insertions(+), 5 deletions(-) create mode 100644 internal/sync/prompt/conflict.go create mode 100644 internal/sync/prompt/conflict_test.go create mode 100644 internal/sync/prompt/runner.go create mode 100644 internal/sync/prompt/runner_test.go create mode 100644 internal/sync/prompt/terminal.go create mode 100644 internal/sync/prompt/terminal_test.go create mode 100644 internal/sync/prompt/watch.go create mode 100644 internal/sync/prompt/watch_test.go diff --git a/go.mod b/go.mod index 33c502b6..cbf9b245 100644 --- a/go.mod +++ b/go.mod @@ -14,6 +14,7 @@ require ( github.com/charmbracelet/huh v1.0.0 github.com/charmbracelet/lipgloss v1.1.1-0.20250404203927-76690c660834 github.com/charmbracelet/x/ansi v0.9.3 + github.com/fsnotify/fsnotify v1.9.0 github.com/getkin/kin-openapi v0.144.0 github.com/google/uuid v1.6.0 github.com/gorilla/handlers v1.5.2 @@ -62,7 +63,6 @@ require ( github.com/dustin/go-humanize v1.0.1 // indirect github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f // indirect github.com/felixge/httpsnoop v1.0.4 // indirect - github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/go-openapi/jsonpointer v0.22.5 // indirect github.com/go-openapi/swag/jsonname v0.25.5 // indirect github.com/go-viper/mapstructure/v2 v2.4.0 // indirect diff --git a/internal/sync/interactive/form.go b/internal/sync/interactive/form.go index 419a5ac8..88cad5c9 100644 --- a/internal/sync/interactive/form.go +++ b/internal/sync/interactive/form.go @@ -1,11 +1,11 @@ package interactive import ( + "context" "errors" "fmt" "io" - tea "github.com/charmbracelet/bubbletea" "github.com/charmbracelet/huh" "github.com/charmbracelet/lipgloss" ) @@ -25,6 +25,17 @@ func Select[T any]( output io.Writer, title string, choices []Choice[T], +) (T, bool, error) { + return SelectContext(context.Background(), input, output, title, choices) +} + +// SelectContext asks the user to choose one value and stops when the context ends. +func SelectContext[T any]( + ctx context.Context, + input io.Reader, + output io.Writer, + title string, + choices []Choice[T], ) (T, bool, error) { var zero T if len(choices) == 0 { @@ -37,7 +48,8 @@ func Select[T any]( options[index] = huh.NewOption(choiceLabel(choice), index) } - canceled, err := RunForm( + canceled, err := RunFormContext( + ctx, input, output, huh.NewSelect[int](). @@ -97,13 +109,23 @@ func RunForm( input io.Reader, output io.Writer, fields ...huh.Field, +) (bool, error) { + return RunFormContext(context.Background(), input, output, fields...) +} + +// RunFormContext runs fields until they complete, abort, or the context ends. +func RunFormContext( + ctx context.Context, + input io.Reader, + output io.Writer, + fields ...huh.Field, ) (bool, error) { err := huh.NewForm(huh.NewGroup(fields...)). WithInput(input). WithOutput(output). + WithAccessible(false). WithTheme(formTheme()). - WithProgramOptions(tea.WithAltScreen()). - Run() + RunWithContext(ctx) if errors.Is(err, huh.ErrUserAborted) { return true, nil } diff --git a/internal/sync/prompt/conflict.go b/internal/sync/prompt/conflict.go new file mode 100644 index 00000000..24a40c8e --- /dev/null +++ b/internal/sync/prompt/conflict.go @@ -0,0 +1,229 @@ +package prompt + +import ( + "context" + "crypto/sha256" + "errors" + "fmt" + "io" + "time" + + syncconsole "github.com/launchdarkly/ldcli/internal/sync/console" + syncinteractive "github.com/launchdarkly/ldcli/internal/sync/interactive" +) + +type conflictResolution string + +const ( + useLaunchDarkly conflictResolution = "launchdarkly" + useLocal conflictResolution = "local" + abortConflict conflictResolution = "abort" +) + +type conflictResolutionResult struct { + resolutions map[ResourceID]conflictResolution + aborted bool + sourcesChanged bool +} + +type conflictChoice struct { + resolution conflictResolution + aborted bool + sourcesChanged bool +} + +// resolveConflicts shows each conflict and asks which side should win. +func resolveConflicts( + options Options, + plan Plan, + reader io.Reader, + interactive bool, + watched *watchedSources, +) (conflictResolutionResult, error) { + var conflicts []PlannedResource + for _, resource := range plan.Resources { + if resource.Action == ActionConflict { + conflicts = append(conflicts, resource) + } + } + if len(conflicts) == 0 { + return conflictResolutionResult{}, nil + } + if !interactive { + return conflictResolutionResult{}, fmt.Errorf( + "interactive conflict resolution requires a terminal; rerun in a terminal or resolve the conflict manually", + ) + } + + result := conflictResolutionResult{resolutions: make(map[ResourceID]conflictResolution, len(conflicts))} + for _, resource := range conflicts { + conflict := Plan{Resources: []PlannedResource{resource}} + if err := writePlanReview(options.ErrorOutput, "plaintext", conflict, terminalWidth(options.ErrorOutput)); err != nil { + return conflictResolutionResult{}, err + } + + choice, err := readConflictChoice(options, reader, watched) + if err != nil { + return conflictResolutionResult{}, err + } + if choice.sourcesChanged { + result.sourcesChanged = true + return result, nil + } + if choice.aborted { + result.aborted = true + return result, nil + } + result.resolutions[resource.ID] = choice.resolution + } + return result, nil +} + +// readConflictChoice waits for a regular terminal choice or a watch-aware +// choice that can be interrupted by another source change. +func readConflictChoice(options Options, reader io.Reader, watched *watchedSources) (conflictChoice, error) { + if watched != nil { + return promptWatchedConflictResolution(options.Context, options.Input, options.ErrorOutput, *watched) + } + return promptConflictResolution(context.Background(), reader, options.ErrorOutput) +} + +// promptConflictResolution asks which side should win one conflict. +func promptConflictResolution( + ctx context.Context, + input io.Reader, + output io.Writer, +) (conflictChoice, error) { + resolution, canceled, err := syncinteractive.SelectContext( + ctx, + input, + output, + "Choose how to resolve this conflict", + []syncinteractive.Choice[conflictResolution]{ + {Title: "Use LaunchDarkly", Value: useLaunchDarkly}, + {Title: "Use local", Value: useLocal}, + {Title: "Abort and resolve manually", Value: abortConflict}, + }, + ) + if err != nil { + return conflictChoice{}, err + } + choice := conflictChoice{ + resolution: resolution, + aborted: canceled || resolution == abortConflict, + } + writeConflictChoice(output, choice) + return choice, nil +} + +// promptWatchedConflictResolution refreshes the plan if a source changes +// while the conflict selector is open. +func promptWatchedConflictResolution( + ctx context.Context, + input io.Reader, + output io.Writer, + watched watchedSources, +) (conflictChoice, error) { + promptContext, cancel := context.WithCancel(ctx) + defer cancel() + + changeResult := make(chan error, 1) + // Race the form against filesystem changes. Canceling the shared child + // context guarantees exactly one path wins and the other exits promptly. + go func() { + changeResult <- watched.WaitForChange(promptContext) + cancel() + }() + + choice, promptErr := promptConflictResolution(promptContext, input, output) + cancel() + changeErr := <-changeResult + if changeErr == nil { + _ = syncconsole.New(output).Line("\nA watched file changed; refreshing...") + return conflictChoice{sourcesChanged: true}, nil + } + if !errors.Is(changeErr, context.Canceled) { + return conflictChoice{}, changeErr + } + if promptErr == nil && choice.aborted { + return conflictChoice{}, context.Canceled + } + if promptErr != nil { + if ctx.Err() != nil { + return conflictChoice{}, context.Canceled + } + return conflictChoice{}, promptErr + } + return choice, nil +} + +// writeConflictChoice confirms the selected resolution after the form exits. +func writeConflictChoice(output io.Writer, choice conflictChoice) { + console := syncconsole.New(output) + switch { + case choice.aborted: + _ = console.Line("Sync canceled; conflict left unresolved.") + case choice.resolution == useLaunchDarkly: + _ = console.Line("Using LaunchDarkly.") + case choice.resolution == useLocal: + _ = console.Line("Using local.") + } +} + +type watchedSources struct { + watcher *sourceWatcher + snapshot [sha256.Size]byte + debounce time.Duration +} + +// WaitForChange waits until the watched source content differs from the state +// used to build the current plan. +func (watched watchedSources) WaitForChange(ctx context.Context) error { + for { + if err := watched.watcher.WaitForChange(ctx, watched.debounce); err != nil { + return err + } + current, err := sourceSnapshot(watched.watcher.root) + if err != nil { + return err + } + if current != watched.snapshot { + return nil + } + } +} + +// applyConflictResolutions replaces conflict actions with the selected direction. +func applyConflictResolutions(plan Plan, resolutions map[ResourceID]conflictResolution) Plan { + resolved := Plan{Resources: append([]PlannedResource(nil), plan.Resources...)} + for index := range resolved.Resources { + resource := &resolved.Resources[index] + resolution, ok := resolutions[resource.ID] + if !ok || resource.Action != ActionConflict { + continue + } + resource.Action = resolvedConflictAction(*resource, resolution) + } + return resolved +} + +// resolvedConflictAction returns the operation needed for the chosen side to win. +func resolvedConflictAction(resource PlannedResource, resolution conflictResolution) Action { + switch resolution { + case useLaunchDarkly: + if resource.Server == nil { + return ActionDeleteLocal + } + return ActionUpdateLocal + case useLocal: + if resource.Local == nil { + return ActionArchiveServer + } + if resource.Server == nil { + return ActionCreateServer + } + return ActionUpdateServer + default: + return ActionConflict + } +} diff --git a/internal/sync/prompt/conflict_test.go b/internal/sync/prompt/conflict_test.go new file mode 100644 index 00000000..f1241a6c --- /dev/null +++ b/internal/sync/prompt/conflict_test.go @@ -0,0 +1,235 @@ +package prompt + +import ( + "bytes" + "encoding/json" + "io" + "net/url" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + syncdomain "github.com/launchdarkly/ldcli/internal/sync" + synclocal "github.com/launchdarkly/ldcli/internal/sync/local" + syncmanifest "github.com/launchdarkly/ldcli/internal/sync/manifest" +) + +func TestResolvedConflictAction(t *testing.T) { + local := testVariation("local") + server := testVariation("server") + + tests := []struct { + name string + resource PlannedResource + resolution conflictResolution + expected Action + }{ + {"LaunchDarkly updates an existing local resource", PlannedResource{Local: &local, Server: &server}, useLaunchDarkly, ActionUpdateLocal}, + {"LaunchDarkly restores a missing local resource", PlannedResource{Server: &server}, useLaunchDarkly, ActionUpdateLocal}, + {"LaunchDarkly deletion removes the local resource", PlannedResource{Local: &local}, useLaunchDarkly, ActionDeleteLocal}, + {"local updates an existing server resource", PlannedResource{Local: &local, Server: &server}, useLocal, ActionUpdateServer}, + {"local creates a missing server resource", PlannedResource{Local: &local}, useLocal, ActionCreateServer}, + {"local deletion archives the server resource", PlannedResource{Server: &server}, useLocal, ActionArchiveServer}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.expected, resolvedConflictAction(test.resource, test.resolution)) + }) + } +} + +func TestApplyConflictResolutionsDoesNotChangeReviewedPlan(t *testing.T) { + id := testResourceID() + local, server := testVariation("local"), testVariation("server") + reviewed := Plan{Resources: []PlannedResource{{ + ID: id, Action: ActionConflict, Local: &local, Server: &server, + }}} + + resolved := applyConflictResolutions(reviewed, map[ResourceID]conflictResolution{id: useLocal}) + + assert.Equal(t, ActionConflict, reviewed.Resources[0].Action) + assert.Equal(t, ActionUpdateServer, resolved.Resources[0].Action) +} + +func TestApplyLocalChangeRestoresMissingConflictFile(t *testing.T) { + root := t.TempDir() + server := testVariation("server") + resource := PlannedResource{ + ID: testResourceID(), + Action: ActionUpdateLocal, + Server: &server, + } + + require.NoError(t, applyLocalChange(synclocal.NewStore(root), resource)) + + resources, err := synclocal.CompileWorkspace(root) + require.NoError(t, err) + require.Len(t, resources, 1) + assert.Equal(t, resource.ID.LookupKey, resources[0].LookupKey) + assert.True(t, resources[0].Upsert) +} + +func TestWriteConflictChoice(t *testing.T) { + var output bytes.Buffer + writeConflictChoice(&output, conflictChoice{resolution: useLocal}) + writeConflictChoice(&output, conflictChoice{aborted: true}) + assert.Equal(t, "Using local.\nSync canceled; conflict left unresolved.\n", output.String()) +} + +func TestResolveConflictsShowsDiffBeforePrompt(t *testing.T) { + plan := divergentPlan(t) + var output bytes.Buffer + input := strings.NewReader("\x1b[B\r") + options := Options{Input: input, ErrorOutput: &output} + + result, err := resolveConflicts(options, plan, input, true, nil) + + require.NoError(t, err) + assert.False(t, result.aborted) + assert.Equal(t, useLocal, result.resolutions[testResourceID()]) + rendered := output.String() + assert.Contains(t, rendered, "LaunchDarkly now") + assert.Contains(t, rendered, "Local file now") + assert.Less(t, strings.Index(rendered, "LaunchDarkly now"), strings.Index(rendered, "Using local.")) +} + +func TestResolveConflictsRequiresTerminalEvenWithYes(t *testing.T) { + plan := divergentPlan(t) + input := strings.NewReader("") + + _, err := resolveConflicts(Options{Input: input, ErrorOutput: &bytes.Buffer{}, Yes: true}, plan, input, false, nil) + + require.ErrorContains(t, err, "interactive conflict resolution requires a terminal") +} + +func TestRunWorkspaceSyncAppliesConflictChoiceAfterRevalidation(t *testing.T) { + root := t.TempDir() + baseline, local, server := testVariation("baseline"), testVariation("local"), testVariation("server") + localStore, manifestStore := writeConflictWorkspace(t, root, baseline, local) + api := &conflictAPI{variation: &server} + runner := NewRunner(api) + input := strings.NewReader("\x1b[B\r") + runner.isTerminal = func(actual io.Reader, _ io.Writer) bool { return actual == input } + var output bytes.Buffer + + err := runner.runWorkspaceSync( + Options{ + AccessToken: "token", BaseURI: "https://example.test", Yes: true, Input: input, + Output: &output, ErrorOutput: &output, + }, + syncWorkspace{root: root, local: localStore, manifest: manifestStore}, + ) + + require.NoError(t, err) + require.NotNil(t, api.variation) + assert.Equal(t, local.Name, api.variation.Name) + assert.Contains(t, api.requests, "PATCH") +} + +func TestRunWorkspaceSyncAbortsConflictWithoutWriting(t *testing.T) { + root := t.TempDir() + baseline, local, server := testVariation("baseline"), testVariation("local"), testVariation("server") + localStore, manifestStore := writeConflictWorkspace(t, root, baseline, local) + api := &conflictAPI{variation: &server} + runner := NewRunner(api) + input := strings.NewReader("\x1b[B\x1b[B\r") + runner.isTerminal = func(actual io.Reader, _ io.Writer) bool { return actual == input } + var output bytes.Buffer + + err := runner.runWorkspaceSync( + Options{ + AccessToken: "token", BaseURI: "https://example.test", Yes: true, Input: input, + Output: &output, ErrorOutput: &output, + }, + syncWorkspace{root: root, local: localStore, manifest: manifestStore}, + ) + + require.NoError(t, err) + require.NotNil(t, api.variation) + assert.Equal(t, server.Name, api.variation.Name) + assert.NotContains(t, api.requests, "PATCH") + assert.Contains(t, output.String(), "Sync canceled; conflict left unresolved.") +} + +func divergentPlan(t *testing.T) Plan { + t.Helper() + + id := testResourceID() + baseline, local, server := testVariation("baseline"), testVariation("local"), testVariation("server") + baselineFingerprint, err := syncdomain.FingerprintVariation(id.ProjectKey, id.LookupKey, baseline) + require.NoError(t, err) + + manifest := syncmanifest.Manifest{ + FormatVersion: syncmanifest.FormatVersion, + Resources: []syncmanifest.Resource{{ + ResourceKind: id.Kind, + ProjectKey: id.ProjectKey, + LookupKey: id.LookupKey, + Fingerprint: baselineFingerprint, + }}, + } + return BuildPlan(manifest, localResources(&local, true), map[ResourceID]ServerResource{ + id: {Variation: &server, ConfigMode: syncdomain.VariationModeAgent}, + }) +} + +func writeConflictWorkspace(t *testing.T, root string, baseline, local syncdomain.Variation) (synclocal.Store, syncmanifest.Store) { + t.Helper() + + localStore := synclocal.NewStore(root) + _, err := localStore.Add([]synclocal.VariationFile{{ + ProjectKey: "production", ConfigKey: "support", Upsert: true, Variation: local, + }}) + require.NoError(t, err) + + id := testResourceID() + fingerprint, err := syncdomain.FingerprintVariation(id.ProjectKey, id.LookupKey, baseline) + require.NoError(t, err) + manifestStore := syncmanifest.NewStore(root) + require.NoError(t, manifestStore.Write(syncmanifest.Manifest{ + FormatVersion: syncmanifest.FormatVersion, + Resources: []syncmanifest.Resource{{ + ResourceKind: id.Kind, ProjectKey: id.ProjectKey, LookupKey: id.LookupKey, Fingerprint: fingerprint, + }}, + })) + return localStore, manifestStore +} + +type conflictAPI struct { + variation *syncdomain.Variation + requests []string +} + +func (api *conflictAPI) MakeRequest( + _ string, + method string, + _ string, + _ string, + _ url.Values, + body []byte, + _ bool, +) ([]byte, error) { + api.requests = append(api.requests, method) + if method == "GET" { + return json.Marshal(map[string]any{ + "key": "support", "name": "Support", "mode": "agent", "variations": []syncdomain.Variation{*api.variation}, + }) + } + if method == "PATCH" { + var update syncdomain.Variation + if err := json.Unmarshal(body, &update); err != nil { + return nil, err + } + update.Mode = syncdomain.VariationModeAgent + update.Key = api.variation.Key + api.variation = &update + } + return []byte(`{}`), nil +} + +func (*conflictAPI) MakeUnauthenticatedRequest(string, string, []byte) ([]byte, error) { + return nil, nil +} diff --git a/internal/sync/prompt/runner.go b/internal/sync/prompt/runner.go new file mode 100644 index 00000000..7f250acb --- /dev/null +++ b/internal/sync/prompt/runner.go @@ -0,0 +1,334 @@ +package prompt + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "os/signal" + "syscall" + + "github.com/launchdarkly/ldcli/internal/resources" + syncdomain "github.com/launchdarkly/ldcli/internal/sync" + syncapi "github.com/launchdarkly/ldcli/internal/sync/api" + syncbootstrap "github.com/launchdarkly/ldcli/internal/sync/bootstrap" + syncconsole "github.com/launchdarkly/ldcli/internal/sync/console" + syncdetach "github.com/launchdarkly/ldcli/internal/sync/detach" + syncinteractive "github.com/launchdarkly/ldcli/internal/sync/interactive" + synclink "github.com/launchdarkly/ldcli/internal/sync/link" + synclocal "github.com/launchdarkly/ldcli/internal/sync/local" + syncmanifest "github.com/launchdarkly/ldcli/internal/sync/manifest" + syncreference "github.com/launchdarkly/ldcli/internal/sync/reference" + syncsource "github.com/launchdarkly/ldcli/internal/sync/source" +) + +// Options contains command input and streams for one prompt synchronization. +type Options struct { + WorkingDirectory string + AccessToken string + BaseURI string + OutputKind string + Add bool + Detach bool + DryRun bool + Format string + Link string + Watch bool + Yes bool + Context context.Context + Input io.Reader + Output io.Writer + ErrorOutput io.Writer + watcher *sourceWatcher +} + +type bootstrapRunner func(syncbootstrap.Options) error +type detachRunner func(syncdetach.Options) error +type linkRunner func(synclink.Options) (string, error) + +type syncWorkspace struct { + root string + local synclocal.Store + manifest syncmanifest.Store +} + +// Runner coordinates prompt synchronization using existing config APIs. +type Runner struct { + client resources.Client + bootstrap bootstrapRunner + detach detachRunner + link linkRunner + watch watchRunner + isTerminal terminalCheck +} + +// NewRunner creates a prompt synchronization runner. +func NewRunner(client resources.Client) Runner { + return Runner{ + client: client, bootstrap: syncbootstrap.Run, detach: syncdetach.Run, link: synclink.Run, + watch: watchWorkspace, isTerminal: syncinteractive.StreamsAreTerminal, + } +} + +// Run resolves the Git workspace and performs the requested prompt sync flow. +func (runner Runner) Run(options Options) error { + if err := validateOptions(options); err != nil { + return err + } + // Every path stored in wrappers or the manifest is repository-relative, so + // resolve the canonical Git root before dispatching any command mode. + resolvedWorkspace, err := syncsource.NewResolver().Resolve(options.WorkingDirectory) + if err != nil { + return err + } + workspace := syncWorkspace{ + root: resolvedWorkspace.Root, + local: synclocal.NewStore(resolvedWorkspace.Root), + manifest: syncmanifest.NewStore(resolvedWorkspace.Root), + } + catalog := syncapi.NewCatalogClient(runner.client, options.AccessToken, options.BaseURI) + + if options.Detach { + return runner.detach(syncdetach.Options{ + RepositoryRoot: workspace.root, + Store: workspace.local, + Manifest: workspace.manifest, + Input: options.Input, + Output: options.Output, + }) + } + if options.Link != "" { + path, err := runner.link(synclink.Options{ + Catalog: catalog, + Store: workspace.local, + RepositoryRoot: workspace.root, + WorkingDirectory: options.WorkingDirectory, + File: options.Link, + Format: options.Format, + Input: options.Input, + Output: options.Output, + }) + if err != nil { + return err + } + if path == "" { + return nil + } + _ = syncconsole.New(options.Output).Printf( + "Linked %s/%s.\n", + syncdomain.RootDir, + path, + ) + } + localDirectoryExists, err := workspace.local.Exists() + if err != nil { + return err + } + + if !localDirectoryExists || options.Add { + if err := runner.bootstrap(syncbootstrap.Options{ + Catalog: catalog, + Store: workspace.local, + Manifest: workspace.manifest, + Input: options.Input, + Output: options.Output, + Initial: !localDirectoryExists, + DryRun: options.DryRun, + }); err != nil { + return err + } + if !options.Watch { + return nil + } + localDirectoryExists, err = workspace.local.Exists() + if err != nil { + return err + } + if !localDirectoryExists { + return nil + } + } + + if options.Watch { + ctx := options.Context + if ctx == nil { + ctx = context.Background() + } + ctx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) + defer stop() + + syncOptions := options + syncOptions.Add = false + syncOptions.Format = "" + syncOptions.Link = "" + syncOptions.Yes = true + syncOptions.Context = ctx + // Watch owns the retry loop. Each callback still runs the exact same + // plan, review, revalidation, and execution pipeline as a normal sync. + return runner.watch(ctx, workspace.root, watchDebounce, func(watcher *sourceWatcher) error { + syncOptions.watcher = watcher + return runner.runWorkspaceSync(syncOptions, workspace) + }, options.ErrorOutput) + } + + return runner.runWorkspaceSync(options, workspace) +} + +// runWorkspaceSync plans, reviews, revalidates, and executes one workspace sync. +func (runner Runner) runWorkspaceSync(options Options, workspace syncWorkspace) error { + var watched *watchedSources + if options.Watch { + if options.watcher == nil { + return fmt.Errorf("watch mode requires an initialized file watcher") + } + snapshot, err := sourceSnapshot(workspace.root) + if err != nil { + return err + } + watched = &watchedSources{ + watcher: options.watcher, + snapshot: snapshot, + debounce: watchDebounce, + } + } + + apiClient := syncapi.NewClient(runner.client, options.AccessToken, options.BaseURI) + // The manifest is the common ancestor in a three-way comparison between + // current local files and current LaunchDarkly state. + baseline, _, err := workspace.manifest.Load() + if err != nil { + return err + } + reviewedPlan, err := loadWorkspacePlan(workspace.root, baseline, apiClient) + if err != nil { + return err + } + + if options.DryRun { + if err := writePlanOutput(options.Output, options.OutputKind, reviewedPlan); err != nil { + return err + } + return reviewedPlan.BlockingError() + } + + interactive := runner.isTerminal(options.Input, options.ErrorOutput) + conflictResult, err := resolveConflicts(options, reviewedPlan, options.Input, interactive, watched) + if err != nil { + return err + } + if conflictResult.sourcesChanged { + return errRefreshWatchPlan + } + if conflictResult.aborted { + return nil + } + + resolvedPlan := applyConflictResolutions(reviewedPlan, conflictResult.resolutions) + shouldContinue, err := reviewAndConfirmPlan(options, resolvedPlan, interactive) + if err != nil || !shouldContinue { + return err + } + + // Re-read both sides after review so no action uses stale state. + currentManifest, _, err := workspace.manifest.Load() + if err != nil { + return err + } + currentPlan, err := loadWorkspacePlan(workspace.root, currentManifest, apiClient) + if err != nil { + return err + } + if !samePlanState(reviewedPlan, currentPlan) { + if options.Watch { + return errRefreshWatchPlan + } + return fmt.Errorf("sync state changed after review; run sync again") + } + + // Apply the user's conflict choices to freshly read state, never to the + // potentially stale objects that were rendered during review. + currentPlan = applyConflictResolutions(currentPlan, conflictResult.resolutions) + outcomes, updatedManifest, executionErr := executePlan(workspace.root, workspace.local, apiClient, currentManifest, currentPlan) + if err := workspace.manifest.Write(updatedManifest); err != nil { + executionErr = errors.Join(executionErr, err) + } + if err := workspace.local.RemoveEmptyDirectories(); err != nil { + executionErr = errors.Join(executionErr, err) + } + if err := writeOutcomeOutput(options.Output, options.OutputKind, outcomes); err != nil { + executionErr = errors.Join(executionErr, err) + } + return executionErr +} + +// validateOptions rejects command modes whose side effects or UX conflict. +func validateOptions(options Options) error { + switch { + case options.Detach && (options.Add || options.DryRun || options.Link != "" || options.Format != "" || options.Watch || options.Yes): + return fmt.Errorf("--detach cannot be combined with other sync actions") + case options.Link == "" && options.Format != "": + return fmt.Errorf("--format requires --link") + case options.Link != "" && options.Format == "": + return fmt.Errorf("--link requires --format") + case options.Link != "" && (options.Add || options.DryRun): + return fmt.Errorf("--link cannot be used with --add or --dry-run") + case options.Watch && options.DryRun: + return fmt.Errorf("--watch cannot be used with --dry-run") + } + if options.Link != "" { + return syncreference.ValidateFormat(options.Format) + } + return nil +} + +// loadWorkspacePlan reads local and server state before building a three-way plan. +func loadWorkspacePlan(repositoryRoot string, baseline syncmanifest.Manifest, client syncapi.Client) (Plan, error) { + localResources, err := synclocal.CompileWorkspace(repositoryRoot) + if err != nil { + return Plan{}, err + } + + resourceIDs := make(map[ResourceID]struct{}, len(localResources)+len(baseline.Resources)) + for _, resource := range localResources { + resourceIDs[ResourceID{Kind: resource.Kind, ProjectKey: resource.ProjectKey, LookupKey: resource.LookupKey}] = struct{}{} + } + for _, resource := range baseline.Resources { + resourceIDs[resource.ID()] = struct{}{} + } + + serverResources := make(map[ResourceID]ServerResource, len(resourceIDs)) + for id := range resourceIDs { + resource, err := readServerResource(client, id) + if err != nil { + return Plan{}, err + } + serverResources[id] = resource + } + return BuildPlan(baseline, localResources, serverResources), nil +} + +// samePlanState reports whether every reviewed decision still has the same inputs. +func samePlanState(reviewed, current Plan) bool { + if len(reviewed.Resources) != len(current.Resources) { + return false + } + for index := range reviewed.Resources { + if !samePlannedResourceState(reviewed.Resources[index], current.Resources[index]) { + return false + } + } + return true +} + +// samePlannedResourceState compares every input that can change a reviewed +// action. Rendered diffs and decoded payload pointers are derived from these values. +func samePlannedResourceState(reviewed, current PlannedResource) bool { + return reviewed.ID == current.ID && + reviewed.Action == current.Action && + reviewed.BaselineFingerprint == current.BaselineFingerprint && + reviewed.LocalFingerprint == current.LocalFingerprint && + reviewed.ServerFingerprint == current.ServerFingerprint && + reviewed.ServerMode == current.ServerMode && + reviewed.Upsert == current.Upsert +} diff --git a/internal/sync/prompt/runner_test.go b/internal/sync/prompt/runner_test.go new file mode 100644 index 00000000..5f1217b9 --- /dev/null +++ b/internal/sync/prompt/runner_test.go @@ -0,0 +1,221 @@ +package prompt + +import ( + "context" + "io" + "net/url" + "os" + "os/exec" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/launchdarkly/ldcli/internal/resources" + syncdomain "github.com/launchdarkly/ldcli/internal/sync" + syncbootstrap "github.com/launchdarkly/ldcli/internal/sync/bootstrap" + syncdetach "github.com/launchdarkly/ldcli/internal/sync/detach" + synclink "github.com/launchdarkly/ldcli/internal/sync/link" + syncreference "github.com/launchdarkly/ldcli/internal/sync/reference" + syncsource "github.com/launchdarkly/ldcli/internal/sync/source" +) + +func TestRunnerBootstrapsMissingWorkspaceAndAddsToExistingWorkspace(t *testing.T) { + tests := map[string]struct { + createDirectory bool + add bool + dryRun bool + wantInitial bool + }{ + "missing workspace": { + wantInitial: true, + }, + "missing workspace dry run": { + dryRun: true, + wantInitial: true, + }, + "add to existing workspace": { + createDirectory: true, + add: true, + }, + } + + for name, test := range tests { + t.Run(name, func(t *testing.T) { + root := initGitRepository(t) + if test.createDirectory { + require.NoError(t, os.Mkdir(filepath.Join(root, syncdomain.RootDir), 0o755)) + } + + called := false + runner := NewRunner(noopResourceClient{}) + runner.bootstrap = func(options syncbootstrap.Options) error { + called = true + assert.Equal(t, test.wantInitial, options.Initial) + assert.Equal(t, test.dryRun, options.DryRun) + assert.NotNil(t, options.Catalog) + assert.NotNil(t, options.Input) + assert.NotNil(t, options.Output) + return nil + } + + err := runner.Run(Options{ + WorkingDirectory: root, + AccessToken: "token", + BaseURI: "https://example.com", + Add: test.add, + DryRun: test.dryRun, + Input: os.Stdin, + Output: io.Discard, + ErrorOutput: io.Discard, + }) + + require.NoError(t, err) + assert.True(t, called) + }) + } +} + +func TestRunnerRequiresGit(t *testing.T) { + runner := NewRunner(noopResourceClient{}) + called := false + runner.bootstrap = func(syncbootstrap.Options) error { + called = true + return nil + } + + err := runner.Run(Options{ + WorkingDirectory: t.TempDir(), + Input: os.Stdin, + Output: io.Discard, + ErrorOutput: io.Discard, + }) + + require.ErrorIs(t, err, syncsource.ErrGitRequired) + assert.False(t, called) +} + +func TestRunnerWatchDoesNotRunInitialSync(t *testing.T) { + root := initGitRepository(t) + require.NoError(t, os.Mkdir(filepath.Join(root, syncdomain.RootDir), 0o755)) + runner := NewRunner(noopResourceClient{}) + watchCalled := false + runner.watch = func( + _ context.Context, + repositoryRoot string, + _ time.Duration, + _ func(*sourceWatcher) error, + _ io.Writer, + ) error { + watchCalled = true + expectedRoot, err := filepath.EvalSymlinks(root) + require.NoError(t, err) + assert.Equal(t, expectedRoot, repositoryRoot) + return nil + } + + err := runner.Run(Options{ + WorkingDirectory: root, + Watch: true, + Input: os.Stdin, + Output: io.Discard, + ErrorOutput: io.Discard, + }) + + require.NoError(t, err) + assert.True(t, watchCalled) +} + +func TestRunnerLinksBeforeWatching(t *testing.T) { + root := initGitRepository(t) + runner := NewRunner(noopResourceClient{}) + linkCalled := false + runner.link = func(options synclink.Options) (string, error) { + linkCalled = true + assert.Equal(t, "prompt.md", options.File) + assert.Equal(t, syncreference.PlainMarkdown, options.Format) + require.NoError(t, os.Mkdir(filepath.Join(root, syncdomain.RootDir), 0o755)) + return "project/configs/config/prompt.prompt.md", nil + } + watchCalled := false + runner.watch = func(context.Context, string, time.Duration, func(*sourceWatcher) error, io.Writer) error { + watchCalled = true + return nil + } + + err := runner.Run(Options{ + WorkingDirectory: root, + Link: "prompt.md", + Format: syncreference.PlainMarkdown, + Watch: true, + Input: os.Stdin, + Output: io.Discard, + ErrorOutput: io.Discard, + }) + + require.NoError(t, err) + assert.True(t, linkCalled) + assert.True(t, watchCalled) +} + +func TestRunnerDetachesWithoutCallingTheAPI(t *testing.T) { + root := initGitRepository(t) + runner := NewRunner(noopResourceClient{}) + called := false + runner.detach = func(options syncdetach.Options) error { + called = true + assert.NotZero(t, options.Store) + assert.NotZero(t, options.Manifest) + assert.Equal(t, os.Stdin, options.Input) + assert.Equal(t, io.Discard, options.Output) + return nil + } + + err := runner.Run(Options{ + WorkingDirectory: root, + Detach: true, + Input: os.Stdin, + Output: io.Discard, + ErrorOutput: io.Discard, + }) + + require.NoError(t, err) + assert.True(t, called) +} + +func TestValidateOptions(t *testing.T) { + require.ErrorContains(t, validateOptions(Options{Format: syncreference.PlainMarkdown}), "--format requires --link") + require.ErrorContains(t, validateOptions(Options{Link: "prompt.md"}), "--link requires --format") + require.ErrorContains(t, validateOptions(Options{Watch: true, DryRun: true}), "--watch cannot be used with --dry-run") + require.ErrorContains(t, validateOptions(Options{Detach: true, Add: true}), "--detach cannot be combined") +} + +type noopResourceClient struct{} + +var _ resources.Client = noopResourceClient{} + +func (noopResourceClient) MakeRequest(string, string, string, string, url.Values, []byte, bool) ([]byte, error) { + return nil, nil +} + +func (noopResourceClient) MakeUnauthenticatedRequest(string, string, []byte) ([]byte, error) { + return nil, nil +} + +func initGitRepository(t *testing.T) string { + t.Helper() + + root := t.TempDir() + for _, args := range [][]string{ + {"init", "--quiet"}, + {"remote", "add", "origin", "git@github.com:launchdarkly/example.git"}, + } { + command := exec.Command("git", args...) + command.Dir = root + output, err := command.CombinedOutput() + require.NoError(t, err, "%s", output) + } + return root +} diff --git a/internal/sync/prompt/terminal.go b/internal/sync/prompt/terminal.go new file mode 100644 index 00000000..c62bc9a6 --- /dev/null +++ b/internal/sync/prompt/terminal.go @@ -0,0 +1,56 @@ +package prompt + +import ( + "bufio" + "fmt" + "io" + "strings" + + syncconsole "github.com/launchdarkly/ldcli/internal/sync/console" +) + +// reviewAndConfirmPlan renders a plan and decides whether execution should continue. +func reviewAndConfirmPlan(options Options, plan Plan, interactive bool) (bool, error) { + if err := writePlanReview(options.ErrorOutput, "plaintext", plan, terminalWidth(options.ErrorOutput)); err != nil { + return false, err + } + if err := plan.BlockingError(); err != nil { + return false, err + } + if !plan.HasChanges() { + if options.OutputKind != "" && options.OutputKind != "plaintext" { + return false, writePlanOutput(options.Output, options.OutputKind, plan) + } + return false, nil + } + if options.Yes || !plan.RequiresConfirmation() { + return true, nil + } + + confirmed, err := confirmApply(options.Input, options.ErrorOutput, interactive) + if err != nil { + return false, err + } + if !confirmed { + _ = syncconsole.New(options.ErrorOutput).Line("Sync canceled.") + } + return confirmed, nil +} + +type terminalCheck func(io.Reader, io.Writer) bool + +// confirmApply asks an interactive user to approve planned changes. +func confirmApply(input io.Reader, prompt io.Writer, interactive bool) (bool, error) { + if !interactive { + return false, fmt.Errorf("interactive apply confirmation requires a terminal; rerun with --yes to apply non-interactively") + } + if err := syncconsole.New(prompt).Write("\nSync these changes? [y/N] "); err != nil { + return false, err + } + answer, err := bufio.NewReader(input).ReadString('\n') + if err != nil && err != io.EOF { + return false, fmt.Errorf("read apply confirmation: %w", err) + } + answer = strings.ToLower(strings.TrimSpace(answer)) + return answer == "y" || answer == "yes", nil +} diff --git a/internal/sync/prompt/terminal_test.go b/internal/sync/prompt/terminal_test.go new file mode 100644 index 00000000..16ee4063 --- /dev/null +++ b/internal/sync/prompt/terminal_test.go @@ -0,0 +1,39 @@ +package prompt + +import ( + "bytes" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestConfirmApply(t *testing.T) { + tests := []struct { + name string + input string + terminal bool + confirmed bool + wantError string + }{ + {name: "yes", input: "yes\n", terminal: true, confirmed: true}, + {name: "declined", input: "n\n", terminal: true}, + {name: "non-terminal", wantError: "rerun with --yes"}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + var prompt bytes.Buffer + confirmed, err := confirmApply(strings.NewReader(test.input), &prompt, test.terminal) + + assert.Equal(t, test.confirmed, confirmed) + if test.wantError != "" { + require.ErrorContains(t, err, test.wantError) + } else { + require.NoError(t, err) + assert.Contains(t, prompt.String(), "Sync these changes?") + } + }) + } +} diff --git a/internal/sync/prompt/watch.go b/internal/sync/prompt/watch.go new file mode 100644 index 00000000..78cf60e5 --- /dev/null +++ b/internal/sync/prompt/watch.go @@ -0,0 +1,434 @@ +package prompt + +import ( + "context" + "crypto/sha256" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "time" + + "github.com/fsnotify/fsnotify" + + syncdomain "github.com/launchdarkly/ldcli/internal/sync" + syncconsole "github.com/launchdarkly/ldcli/internal/sync/console" + synclocal "github.com/launchdarkly/ldcli/internal/sync/local" +) + +const watchDebounce = 300 * time.Millisecond + +var errRefreshWatchPlan = errors.New("refresh watch plan") + +type watchRunner func(context.Context, string, time.Duration, func(*sourceWatcher) error, io.Writer) error + +// watchWorkspace waits for a stable source change before running sync. The +// initial filesystem state is only a baseline and never triggers a sync. +func watchWorkspace( + ctx context.Context, + repositoryRoot string, + debounce time.Duration, + syncWorkspace func(*sourceWatcher) error, + output io.Writer, +) error { + // Show the exact initial watch set, but do not treat it as a change. Watch + // mode intentionally waits for an edit before running its first sync. + files, err := synclocal.SourceFiles(repositoryRoot) + if err != nil { + return err + } + console := syncconsole.New(output) + _ = console.Line("Watching files for sync:") + for _, file := range files { + _ = console.Printf("- %s\n", file) + } + + watcher, err := newSourceWatcher(repositoryRoot) + if err != nil { + return err + } + defer watcher.Close() + + // Filesystem APIs commonly emit duplicate events for one editor save. A + // content snapshot lets us ignore events that do not change sync inputs. + handledSnapshot, err := sourceSnapshot(repositoryRoot) + if err != nil { + return err + } + + for { + if err := watcher.WaitForChange(ctx, debounce); err != nil { + if errors.Is(err, context.Canceled) { + return nil + } + _ = console.Printf("Watch error: %s\n", err) + continue + } + + current, err := sourceSnapshot(repositoryRoot) + if err != nil { + _ = console.Printf("Watch error: %s\n", err) + continue + } + if current == handledSnapshot { + continue + } + + _ = console.Line("A watched file changed; syncing...") + // Keep rebuilding until one sync attempt covers a stable source + // snapshot. This closes the gap where an edit arrives during sync. + for { + err = syncWorkspace(watcher) + if errors.Is(err, context.Canceled) { + return nil + } + if errors.Is(err, errRefreshWatchPlan) { + // A source changed during review or the reviewed server state + // became stale. Rebuild immediately instead of waiting for a + // second filesystem event that may never arrive. + if err := watcher.Refresh(); err != nil { + return err + } + current, err = sourceSnapshot(repositoryRoot) + if err != nil { + return err + } + continue + } + if err != nil { + _ = console.Printf("Sync failed: %s\n", err) + } + + // Record the exact state used by this attempt. If sync or the user + // changed another source while it ran, rebuild immediately. + handledSnapshot = current + if err := watcher.Refresh(); err != nil { + return err + } + latestSnapshot, err := sourceSnapshot(repositoryRoot) + if err != nil { + return err + } + if latestSnapshot != current { + current = latestSnapshot + continue + } + break + } + } +} + +type sourceWatcher struct { + root string + managedRoot string + files map[string]struct{} + directories map[string]struct{} + watcher *fsnotify.Watcher +} + +// newSourceWatcher creates the fsnotify watcher and registers the initial +// managed and referenced source directories. +func newSourceWatcher(repositoryRoot string) (*sourceWatcher, error) { + watcher, err := fsnotify.NewWatcher() + if err != nil { + return nil, fmt.Errorf("create sync file watcher: %w", err) + } + + result := &sourceWatcher{ + root: repositoryRoot, + managedRoot: filepath.Join(repositoryRoot, syncdomain.RootDir), + files: make(map[string]struct{}), + directories: make(map[string]struct{}), + watcher: watcher, + } + if err := result.Refresh(); err != nil { + _ = watcher.Close() + return nil, err + } + return result, nil +} + +// Close releases the underlying operating-system watcher. +func (watcher *sourceWatcher) Close() error { + return watcher.watcher.Close() +} + +// Refresh discovers referenced files and adds any newly relevant directories. +func (watcher *sourceWatcher) Refresh() error { + files, err := synclocal.SourceFiles(watcher.root) + if err != nil { + return err + } + + // Rebuild registrations from source-of-truth state because a wrapper edit + // may add, remove, or redirect an external reference. + watcher.resetDirectories() + watcher.files = make(map[string]struct{}, len(files)) + // Watching the repository root lets us observe recreation of a deleted + // .launchdarkly tree and creation of a missing referenced-file parent. + if err := watcher.addDirectory(watcher.root); err != nil { + return err + } + for _, file := range files { + absolute := filepath.Join(watcher.root, filepath.FromSlash(file)) + watcher.files[filepath.Clean(absolute)] = struct{}{} + if err := watcher.addClosestExistingDirectory(filepath.Dir(absolute)); err != nil { + return err + } + } + // fsnotify is not recursive, so every existing managed directory needs its + // own registration. + return watcher.addDirectoryTree(watcher.managedRoot) +} + +// resetDirectories removes stale OS registrations before rebuilding the +// desired watch set from current workspace state. +func (watcher *sourceWatcher) resetDirectories() { + for directory := range watcher.directories { + _ = watcher.watcher.Remove(directory) + } + watcher.directories = make(map[string]struct{}) +} + +// WaitForChange waits until relevant filesystem events have been quiet for the +// debounce period. +func (watcher *sourceWatcher) WaitForChange( + ctx context.Context, + debounce time.Duration, +) error { + var timer *time.Timer + // A nil channel disables the timer select case until the first relevant + // event starts the debounce window. + var timerChannel <-chan time.Time + defer func() { + if timer != nil { + timer.Stop() + } + }() + + for { + select { + case <-ctx.Done(): + return context.Canceled + case err, ok := <-watcher.watcher.Errors: + if !ok { + return errors.New("sync file watcher closed") + } + return fmt.Errorf("watch sync files: %w", err) + case event, ok := <-watcher.watcher.Events: + if !ok { + return errors.New("sync file watcher closed") + } + // Maintain directory registrations for all events, even ones that + // do not represent a source-content change. + watcher.forgetRemovedDirectories(event) + if err := watcher.addCreatedDirectory(event); err != nil { + return err + } + if !watcher.relevant(event) { + continue + } + if timer == nil { + timer = time.NewTimer(debounce) + timerChannel = timer.C + continue + } + // Editors often save through several writes or a temp-file rename. + // Restart the timer until that burst has gone quiet. + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + timer.Reset(debounce) + case <-timerChannel: + return nil + } + } +} + +// addCreatedDirectory recursively watches a newly created directory when it +// belongs to the managed tree or leads to a referenced source. +func (watcher *sourceWatcher) addCreatedDirectory(event fsnotify.Event) error { + if !event.Has(fsnotify.Create) { + return nil + } + // The path may disappear between the event and Stat when an editor uses a + // short-lived temporary directory. There is nothing left to register. + info, err := os.Stat(event.Name) + if errors.Is(err, os.ErrNotExist) || err == nil && !info.IsDir() { + return nil + } + if err != nil { + return fmt.Errorf("inspect watched path %q: %w", event.Name, err) + } + if !watcher.shouldWatchDirectory(event.Name) { + return nil + } + return watcher.addDirectoryTree(event.Name) +} + +// forgetRemovedDirectories drops registrations invalidated by rename or removal. +func (watcher *sourceWatcher) forgetRemovedDirectories(event fsnotify.Event) { + if event.Op&(fsnotify.Remove|fsnotify.Rename) == 0 { + return + } + + removed := filepath.Clean(event.Name) + for directory := range watcher.directories { + if directory == removed || pathWithin(removed, directory) { + // Remove descendants from our bookkeeping as well; their OS watches + // are no longer useful after an ancestor moves or disappears. + _ = watcher.watcher.Remove(directory) + delete(watcher.directories, directory) + } + } +} + +// shouldWatchDirectory reports whether a directory contains managed files or +// is an ancestor of a referenced file that may not exist yet. +func (watcher *sourceWatcher) shouldWatchDirectory(directory string) bool { + if watcher.insideManagedRoot(directory) { + return true + } + for file := range watcher.files { + if pathWithin(directory, file) { + return true + } + } + return false +} + +// relevant filters noisy filesystem events down to tracked sources and newly +// created managed resources. +func (watcher *sourceWatcher) relevant(event fsnotify.Event) bool { + if event.Op&(fsnotify.Write|fsnotify.Create|fsnotify.Remove|fsnotify.Rename) == 0 { + return false + } + + name := filepath.Clean(event.Name) + if _, tracked := watcher.files[name]; tracked { + // Tracked paths remain relevant even after remove or rename events, when + // the path can no longer be inspected. + return true + } + if event.Has(fsnotify.Create) { + if info, err := os.Stat(name); err == nil && info.IsDir() && watcher.shouldWatchDirectory(name) { + return true + } + // New resource kinds may use different filenames. Treat any new file in + // a project subtree as relevant so watch mode does not need to know each + // resource format. Root-level files are sync metadata such as the manifest. + if watcher.insideManagedRoot(name) { + relative, err := filepath.Rel(watcher.managedRoot, name) + return err == nil && filepath.Dir(relative) != "." + } + } + return false +} + +// insideManagedRoot reports whether a path belongs to .launchdarkly. +func (watcher *sourceWatcher) insideManagedRoot(name string) bool { + return pathWithin(watcher.managedRoot, name) +} + +// pathWithin performs a path-aware containment check without prefix ambiguity. +func pathWithin(parent, child string) bool { + relative, err := filepath.Rel(parent, child) + return err == nil && + relative != ".." && + !strings.HasPrefix(relative, ".."+string(filepath.Separator)) +} + +// addDirectoryTree registers every existing directory below a root. +func (watcher *sourceWatcher) addDirectoryTree(root string) error { + err := filepath.WalkDir(root, func(path string, entry os.DirEntry, walkErr error) error { + if walkErr != nil { + return walkErr + } + if entry.IsDir() { + return watcher.addDirectory(path) + } + return nil + }) + // A missing managed tree is valid while a user deletes or recreates it. The + // repository-root watch will report its next creation. + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return fmt.Errorf("watch sync directories: %w", err) + } + return nil +} + +// addDirectory registers one directory once and records the OS watch. +func (watcher *sourceWatcher) addDirectory(path string) error { + path = filepath.Clean(path) + if _, exists := watcher.directories[path]; exists { + return nil + } + if err := watcher.watcher.Add(path); err != nil { + return fmt.Errorf("watch sync directory %q: %w", path, err) + } + watcher.directories[path] = struct{}{} + return nil +} + +// addClosestExistingDirectory walks toward the repository root until it finds +// a directory that can observe creation of the missing descendants. +func (watcher *sourceWatcher) addClosestExistingDirectory(path string) error { + path = filepath.Clean(path) + // A referenced file may not exist yet. Its nearest existing ancestor is + // enough to observe creation of the next missing path component. + for pathWithin(watcher.root, path) { + info, err := os.Stat(path) + if err == nil { + if !info.IsDir() { + return fmt.Errorf("watch sync directory %q: not a directory", path) + } + return watcher.addDirectory(path) + } + if !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("inspect sync directory %q: %w", path, err) + } + if path == watcher.root { + break + } + path = filepath.Dir(path) + } + return watcher.addDirectory(watcher.root) +} + +// sourceSnapshot hashes sorted source paths, bytes, and read failures so watch +// can distinguish meaningful state changes from duplicate fsnotify events. +func sourceSnapshot(repositoryRoot string) ([sha256.Size]byte, error) { + files, err := synclocal.SourceFiles(repositoryRoot) + if err != nil { + return [sha256.Size]byte{}, err + } + + hash := sha256.New() + for _, file := range files { + // Include both the path and bytes with separators. This distinguishes + // renames and avoids ambiguous concatenations across adjacent files. + _, _ = io.WriteString(hash, file) + _, _ = hash.Write([]byte{0}) + content, err := os.ReadFile(filepath.Join(repositoryRoot, filepath.FromSlash(file))) + if err != nil { + // Missing or unreadable files are still meaningful states. Hash the + // error so deletion and later recreation produce different snapshots. + _, _ = io.WriteString(hash, "!"+err.Error()) + } else { + _, _ = hash.Write(content) + } + _, _ = hash.Write([]byte{0}) + } + + var snapshot [sha256.Size]byte + copy(snapshot[:], hash.Sum(nil)) + return snapshot, nil +} diff --git a/internal/sync/prompt/watch_test.go b/internal/sync/prompt/watch_test.go new file mode 100644 index 00000000..0dfe2fb4 --- /dev/null +++ b/internal/sync/prompt/watch_test.go @@ -0,0 +1,434 @@ +package prompt + +import ( + "bytes" + "context" + "errors" + "io" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/fsnotify/fsnotify" + "github.com/stretchr/testify/require" + + syncdomain "github.com/launchdarkly/ldcli/internal/sync" + synclocal "github.com/launchdarkly/ldcli/internal/sync/local" + syncreference "github.com/launchdarkly/ldcli/internal/sync/reference" +) + +func TestWatchWorkspaceWaitsForDebouncedChange(t *testing.T) { + root := t.TempDir() + wrapper := filepath.Join(root, syncdomain.RootDir, "project", "configs", "config", "prompt.prompt.md") + require.NoError(t, os.MkdirAll(filepath.Dir(wrapper), 0o755)) + require.NoError(t, os.WriteFile(wrapper, []byte("initial"), 0o644)) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var syncCount atomic.Int32 + synced := make(chan struct{}, 2) + done := make(chan error, 1) + var output bytes.Buffer + go func() { + done <- watchWorkspace(ctx, root, 20*time.Millisecond, func(*sourceWatcher) error { + syncCount.Add(1) + synced <- struct{}{} + return nil + }, &output) + }() + + select { + case <-synced: + t.Fatal("watch ran an initial sync") + case <-time.After(40 * time.Millisecond): + } + + require.NoError(t, os.WriteFile(wrapper, []byte("first"), 0o644)) + require.NoError(t, os.WriteFile(wrapper, []byte("second"), 0o644)) + select { + case <-synced: + case <-time.After(time.Second): + t.Fatal("watch did not sync changed file") + } + require.Equal(t, int32(1), syncCount.Load()) + + cancel() + require.NoError(t, <-done) + require.Contains(t, output.String(), "Watching files for sync:\n- .launchdarkly/project/configs/config/prompt.prompt.md\n") +} + +func TestWatchWorkspaceKeepsWatchingAfterSyncError(t *testing.T) { + root := t.TempDir() + wrapper := filepath.Join(root, syncdomain.RootDir, "project", "configs", "config", "prompt.prompt.md") + require.NoError(t, os.MkdirAll(filepath.Dir(wrapper), 0o755)) + require.NoError(t, os.WriteFile(wrapper, []byte("initial"), 0o644)) + + ctx, cancel := context.WithCancel(context.Background()) + var attempts atomic.Int32 + attempted := make(chan struct{}, 2) + done := make(chan error, 1) + go func() { + done <- watchWorkspace(ctx, root, 15*time.Millisecond, func(*sourceWatcher) error { + attempt := attempts.Add(1) + attempted <- struct{}{} + if attempt == 1 { + return errors.New("temporary failure") + } + return nil + }, io.Discard) + }() + + time.Sleep(30 * time.Millisecond) + require.NoError(t, os.WriteFile(wrapper, []byte("first"), 0o644)) + select { + case <-attempted: + case <-time.After(time.Second): + t.Fatal("watch did not attempt first sync") + } + require.NoError(t, os.WriteFile(wrapper, []byte("second"), 0o644)) + select { + case <-attempted: + case <-time.After(time.Second): + t.Fatal("watch did not attempt second sync") + } + require.Equal(t, int32(2), attempts.Load()) + + cancel() + require.NoError(t, <-done) +} + +func TestWatchWorkspaceRefreshesPlanAfterSourceChangesDuringSync(t *testing.T) { + root := t.TempDir() + wrapper := filepath.Join(root, syncdomain.RootDir, "project", "configs", "config", "prompt.prompt.md") + require.NoError(t, os.MkdirAll(filepath.Dir(wrapper), 0o755)) + require.NoError(t, os.WriteFile(wrapper, []byte("initial"), 0o644)) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var attempts atomic.Int32 + attempted := make(chan struct{}, 2) + done := make(chan error, 1) + go func() { + done <- watchWorkspace(ctx, root, 15*time.Millisecond, func(*sourceWatcher) error { + attempt := attempts.Add(1) + attempted <- struct{}{} + if attempt == 1 { + return errRefreshWatchPlan + } + return nil + }, io.Discard) + }() + + time.Sleep(30 * time.Millisecond) + require.NoError(t, os.WriteFile(wrapper, []byte("changed"), 0o644)) + for attempt := 0; attempt < 2; attempt++ { + select { + case <-attempted: + case <-time.After(time.Second): + t.Fatal("watch did not refresh the changed plan") + } + } + require.Equal(t, int32(2), attempts.Load()) + + cancel() + require.NoError(t, <-done) +} + +func TestWatchWorkspaceRefreshesReferencedFilesBeforeRebuildingPlan(t *testing.T) { + root := t.TempDir() + store := synclocal.NewStore(root) + _, err := store.Add([]synclocal.VariationFile{{ + ProjectKey: "project", + ConfigKey: "config", + Variation: syncdomain.Variation{ + Mode: syncdomain.VariationModeAgent, Key: "prompt", Name: "Prompt", Instructions: "Initial", + }, + }}) + require.NoError(t, err) + wrapper := filepath.Join(root, syncdomain.RootDir, "project", "configs", "config", "prompt.prompt.md") + referencePath := filepath.Join(root, "prompts", "prompt.md") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var attempts atomic.Int32 + attempted := make(chan struct{}, 3) + done := make(chan error, 1) + go func() { + done <- watchWorkspace(ctx, root, 15*time.Millisecond, func(*sourceWatcher) error { + attempt := attempts.Add(1) + if attempt == 1 { + if err := os.MkdirAll(filepath.Dir(referencePath), 0o755); err != nil { + return err + } + if err := os.WriteFile(referencePath, []byte("Referenced"), 0o644); err != nil { + return err + } + content := `--- +formatVersion: 1 +upsert: true +ref: + file: prompts/prompt.md + format: plain-markdown +mode: agent +key: prompt +name: Prompt +--- +` + if err := os.WriteFile(wrapper, []byte(content), 0o644); err != nil { + return err + } + attempted <- struct{}{} + return errRefreshWatchPlan + } + attempted <- struct{}{} + return nil + }, io.Discard) + }() + + time.Sleep(30 * time.Millisecond) + require.NoError(t, os.WriteFile(wrapper, []byte("trigger"), 0o644)) + for attempt := 0; attempt < 2; attempt++ { + select { + case <-attempted: + case <-time.After(time.Second): + t.Fatal("watch did not rebuild the plan") + } + } + + require.NoError(t, os.WriteFile(referencePath, []byte("Changed"), 0o644)) + select { + case <-attempted: + case <-time.After(time.Second): + t.Fatal("watch did not track the newly referenced file") + } + + cancel() + require.NoError(t, <-done) +} + +func TestWatchWorkspaceRetriesEditsMadeDuringSuccessfulSync(t *testing.T) { + root := t.TempDir() + wrapper := filepath.Join(root, syncdomain.RootDir, "project", "configs", "config", "prompt.prompt.md") + require.NoError(t, os.MkdirAll(filepath.Dir(wrapper), 0o755)) + require.NoError(t, os.WriteFile(wrapper, []byte("initial"), 0o644)) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var attempts atomic.Int32 + attempted := make(chan struct{}, 2) + done := make(chan error, 1) + go func() { + done <- watchWorkspace(ctx, root, 15*time.Millisecond, func(*sourceWatcher) error { + if attempts.Add(1) == 1 { + if err := os.WriteFile(wrapper, []byte("changed during sync"), 0o644); err != nil { + return err + } + } + attempted <- struct{}{} + return nil + }, io.Discard) + }() + + time.Sleep(30 * time.Millisecond) + require.NoError(t, os.WriteFile(wrapper, []byte("trigger"), 0o644)) + for attempt := 0; attempt < 2; attempt++ { + select { + case <-attempted: + case <-time.After(time.Second): + t.Fatal("watch dropped a source edit made during sync") + } + } + require.Equal(t, int32(2), attempts.Load()) + + cancel() + require.NoError(t, <-done) +} + +func TestWatchedConflictPromptRefreshesWhenSourceChanges(t *testing.T) { + root := t.TempDir() + wrapper := filepath.Join(root, syncdomain.RootDir, "project", "configs", "config", "prompt.prompt.md") + require.NoError(t, os.MkdirAll(filepath.Dir(wrapper), 0o755)) + require.NoError(t, os.WriteFile(wrapper, []byte("initial"), 0o644)) + snapshot, err := sourceSnapshot(root) + require.NoError(t, err) + watcher, err := newSourceWatcher(root) + require.NoError(t, err) + defer watcher.Close() + + input, inputWriter := io.Pipe() + defer input.Close() + defer inputWriter.Close() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + results := make(chan conflictChoice, 1) + errs := make(chan error, 1) + go func() { + result, err := promptWatchedConflictResolution(ctx, input, io.Discard, watchedSources{ + watcher: watcher, snapshot: snapshot, debounce: 15 * time.Millisecond, + }) + results <- result + errs <- err + }() + + require.NoError(t, os.WriteFile(wrapper, []byte("changed"), 0o644)) + select { + case result := <-results: + require.NoError(t, <-errs) + require.True(t, result.sourcesChanged) + case <-ctx.Done(): + t.Fatal("conflict prompt did not refresh after the source changed") + } +} + +func TestWatchedConflictPromptCancellationStopsWatch(t *testing.T) { + root := t.TempDir() + wrapper := filepath.Join(root, syncdomain.RootDir, "project", "configs", "config", "prompt.prompt.md") + require.NoError(t, os.MkdirAll(filepath.Dir(wrapper), 0o755)) + require.NoError(t, os.WriteFile(wrapper, []byte("initial"), 0o644)) + snapshot, err := sourceSnapshot(root) + require.NoError(t, err) + watcher, err := newSourceWatcher(root) + require.NoError(t, err) + defer watcher.Close() + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _, err = promptWatchedConflictResolution( + ctx, + strings.NewReader("\x03"), + io.Discard, + watchedSources{watcher: watcher, snapshot: snapshot, debounce: 15 * time.Millisecond}, + ) + + require.ErrorIs(t, err, context.Canceled) +} + +func TestWatchWorkspaceTracksMissingReferencedFileParent(t *testing.T) { + root := t.TempDir() + _, err := synclocal.NewStore(root).Add([]synclocal.VariationFile{{ + ProjectKey: "project", ConfigKey: "config", + Ref: &synclocal.Reference{File: "prompts/nested/prompt.md", Format: syncreference.PlainMarkdown}, + Variation: syncdomain.Variation{ + Mode: syncdomain.VariationModeAgent, Key: "prompt", Name: "Prompt", + }, + }}) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + synced := make(chan struct{}, 1) + done := make(chan error, 1) + go func() { + done <- watchWorkspace(ctx, root, 15*time.Millisecond, func(*sourceWatcher) error { + synced <- struct{}{} + return nil + }, io.Discard) + }() + + time.Sleep(30 * time.Millisecond) + referencePath := filepath.Join(root, "prompts", "nested", "prompt.md") + require.NoError(t, os.MkdirAll(filepath.Dir(referencePath), 0o755)) + require.NoError(t, os.WriteFile(referencePath, []byte("created"), 0o644)) + select { + case <-synced: + case <-time.After(time.Second): + t.Fatal("watch did not detect a referenced file created under a missing parent") + } + + cancel() + require.NoError(t, <-done) +} + +func TestWatchWorkspaceRewatchesRecreatedManagedTree(t *testing.T) { + root := t.TempDir() + wrapper := filepath.Join(root, syncdomain.RootDir, "project", "configs", "config", "prompt.prompt.md") + require.NoError(t, os.MkdirAll(filepath.Dir(wrapper), 0o755)) + require.NoError(t, os.WriteFile(wrapper, []byte("initial"), 0o644)) + + ctx, cancel := context.WithCancel(context.Background()) + attempted := make(chan struct{}, 2) + done := make(chan error, 1) + go func() { + done <- watchWorkspace(ctx, root, 15*time.Millisecond, func(*sourceWatcher) error { + attempted <- struct{}{} + return nil + }, io.Discard) + }() + + time.Sleep(30 * time.Millisecond) + require.NoError(t, os.RemoveAll(filepath.Join(root, syncdomain.RootDir))) + select { + case <-attempted: + case <-time.After(time.Second): + t.Fatal("watch did not detect the managed tree deletion") + } + + require.NoError(t, os.MkdirAll(filepath.Dir(wrapper), 0o755)) + require.NoError(t, os.WriteFile(wrapper, []byte("recreated"), 0o644)) + select { + case <-attempted: + case <-time.After(time.Second): + t.Fatal("watch did not detect the recreated managed tree") + } + + cancel() + require.NoError(t, <-done) +} + +func TestWatchWorkspaceDetectsReferencedFileChange(t *testing.T) { + root := t.TempDir() + referencePath := filepath.Join(root, "prompt.md") + require.NoError(t, os.WriteFile(referencePath, []byte("initial"), 0o644)) + _, err := synclocal.NewStore(root).Add([]synclocal.VariationFile{{ + ProjectKey: "project", ConfigKey: "config", + Ref: &synclocal.Reference{File: "prompt.md", Format: syncreference.PlainMarkdown}, + Variation: syncdomain.Variation{ + Mode: syncdomain.VariationModeAgent, Key: "prompt", Name: "Prompt", + }, + }}) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + synced := make(chan struct{}, 1) + done := make(chan error, 1) + var output bytes.Buffer + go func() { + done <- watchWorkspace(ctx, root, 15*time.Millisecond, func(*sourceWatcher) error { + synced <- struct{}{} + return nil + }, &output) + }() + + time.Sleep(30 * time.Millisecond) + require.NoError(t, os.WriteFile(referencePath, []byte("changed"), 0o644)) + select { + case <-synced: + case <-time.After(time.Second): + t.Fatal("watch did not detect referenced file change") + } + cancel() + require.NoError(t, <-done) + require.Contains(t, output.String(), "- .launchdarkly/project/configs/config/prompt.prompt.md\n") + require.Contains(t, output.String(), "- prompt.md\n") +} + +func TestSourceWatcherRecognizesNewManagedResourceKinds(t *testing.T) { + root := t.TempDir() + managedRoot := filepath.Join(root, syncdomain.RootDir) + resourceFile := filepath.Join(managedRoot, "project", "skills", "review.skill.md") + require.NoError(t, os.MkdirAll(filepath.Dir(resourceFile), 0o755)) + require.NoError(t, os.WriteFile(resourceFile, []byte("Review carefully."), 0o644)) + + watcher := sourceWatcher{ + managedRoot: managedRoot, + files: make(map[string]struct{}), + } + + require.True(t, watcher.relevant(fsnotify.Event{Name: resourceFile, Op: fsnotify.Create})) + require.False(t, watcher.relevant(fsnotify.Event{ + Name: filepath.Join(managedRoot, "manifest.yaml"), + Op: fsnotify.Create, + })) +}