diff --git a/apps/yaak-client/commands/commands.test.tsx b/apps/yaak-client/commands/commands.test.tsx new file mode 100644 index 000000000..eef1f85ca --- /dev/null +++ b/apps/yaak-client/commands/commands.test.tsx @@ -0,0 +1,148 @@ +import type { FunctionComponent, ReactElement } from "react"; +import { beforeEach, expect, test, vi } from "vite-plus/test"; +import type { SyncOp } from "@yaakapp-internal/sync/bindings/gen_sync"; +import type { DialogInstance } from "../components/Dialogs"; +import { syncWorkspace } from "./commands"; + +const mocks = vi.hoisted(() => ({ + calculateSync: vi.fn(), + applySync: vi.fn(), + showDialog: vi.fn<(dialog: DialogInstance) => void>(), + hideDialog: vi.fn(), + showToast: vi.fn(), +})); + +vi.mock("@yaakapp-internal/sync", () => mocks); +vi.mock("@yaakapp-internal/models", () => ({ + createWorkspaceModel: vi.fn(), + modelTypeLabel: () => "Request", +})); +vi.mock("@yaakapp-internal/ui", () => ({ + Banner: "div", + InlineCode: "code", + Table: "table", + TableBody: "tbody", + TableCell: "td", + TableHead: "thead", + TableHeaderCell: "th", + TableRow: "tr", + TruncatedWideTableCell: "td", +})); +vi.mock("../components/core/Button", () => ({ Button: "button" })); +vi.mock("../hooks/useActiveWorkspace", () => ({ activeWorkspaceIdAtom: null })); +vi.mock("../lib/dialog", () => mocks); +vi.mock("../lib/jotai", () => ({ jotaiStore: { get: vi.fn() } })); +vi.mock("../lib/prompt", () => ({ showPrompt: vi.fn() })); +vi.mock("../lib/resolvedModelName", () => ({ resolvedModelNameWithFolders: () => "Request" })); +vi.mock("../lib/toast", () => mocks); + +const staleToast = expect.objectContaining({ + message: "Files changed while syncing. Review the changes again.", +}); + +const params = { workspaceId: "wk_test", syncDir: "/sync" }; +const deletion = { + type: "dbDelete", + model: { id: "rq_test", model: "http_request" }, + state: {}, +} as Extract; +const update = { + type: "dbUpdate", + model: { id: "rq_test", model: "http_request" }, + fs: { model: { id: "rq_test", model: "http_request", name: "Updated" } }, + state: {}, +} as Extract; + +async function submit(dialog: DialogInstance) { + const render = dialog.render as FunctionComponent<{ hide: () => void }>; + const form = render({ hide: () => mocks.hideDialog(dialog.id) }) as ReactElement<{ + onSubmit: (event: { preventDefault: () => void }) => Promise; + }>; + await form.props.onSubmit({ preventDefault: vi.fn() }); +} + +function latestDialog() { + const dialog = mocks.showDialog.mock.calls.at(-1)?.[0]; + if (dialog == null) throw new Error("Expected a sync dialog"); + return dialog; +} + +beforeEach(() => { + vi.resetAllMocks(); + mocks.applySync.mockResolvedValue(true); +}); + +test("closes an obsolete dialog when the restored file leaves no changes", async () => { + mocks.calculateSync.mockResolvedValueOnce([deletion]).mockResolvedValueOnce([]); + await syncWorkspace.mutateAsync(params); + mocks.applySync.mockResolvedValueOnce(false); + + await submit(latestDialog()); + + expect(mocks.applySync).toHaveBeenCalledTimes(1); + expect(mocks.applySync).toHaveBeenCalledWith(params.workspaceId, params.syncDir, [deletion]); + expect(mocks.hideDialog).toHaveBeenCalledWith("commit-sync"); + expect(mocks.showDialog).toHaveBeenCalledTimes(1); +}); + +test("requires another confirmation for a changed plan", async () => { + mocks.calculateSync.mockResolvedValueOnce([deletion]).mockResolvedValueOnce([update]); + await syncWorkspace.mutateAsync(params); + const original = latestDialog(); + mocks.applySync.mockResolvedValueOnce(false); + + await submit(original); + + expect(mocks.applySync).toHaveBeenCalledTimes(1); + expect(mocks.showDialog).toHaveBeenCalledTimes(2); + expect(mocks.showToast).toHaveBeenCalledWith(staleToast); + expect(mocks.hideDialog).not.toHaveBeenCalled(); + await submit(latestDialog()); + expect(mocks.applySync).toHaveBeenLastCalledWith(params.workspaceId, params.syncDir, [update]); + expect(mocks.hideDialog).toHaveBeenCalledWith("commit-sync"); +}); + +test("applies an unchanged reviewed plan and closes the dialog", async () => { + mocks.calculateSync.mockResolvedValue([update]); + await syncWorkspace.mutateAsync(params); + await submit(latestDialog()); + + expect(mocks.calculateSync).toHaveBeenCalledTimes(1); + expect(mocks.applySync).toHaveBeenCalledTimes(1); + expect(mocks.hideDialog).toHaveBeenCalledWith("commit-sync"); +}); + +test("forced sync retries the current plan without requesting another confirmation", async () => { + mocks.calculateSync.mockResolvedValueOnce([deletion]).mockResolvedValueOnce([update]); + mocks.applySync.mockResolvedValueOnce(false); + + await syncWorkspace.mutateAsync({ ...params, force: true }); + + expect(mocks.applySync).toHaveBeenCalledTimes(2); + expect(mocks.applySync).toHaveBeenLastCalledWith(params.workspaceId, params.syncDir, [update]); + expect(mocks.showDialog).not.toHaveBeenCalled(); +}); + +test("outbound-only changes clear the old dialog and sync automatically", async () => { + const outbound = { type: "fsUpdate", model: update.model, state: {} } as SyncOp; + mocks.calculateSync.mockResolvedValueOnce([deletion]).mockResolvedValueOnce([outbound]); + await syncWorkspace.mutateAsync(params); + await syncWorkspace.mutateAsync(params); + + expect(mocks.applySync).toHaveBeenCalledWith(params.workspaceId, params.syncDir, [outbound]); + expect(mocks.hideDialog).toHaveBeenCalledWith("commit-sync"); + expect(mocks.showDialog).toHaveBeenCalledTimes(1); +}); + +test("stops retrying a plan that keeps going stale", async () => { + mocks.calculateSync.mockResolvedValue([update]); + mocks.applySync.mockResolvedValue(false); + + await syncWorkspace.mutateAsync({ ...params, force: true }); + + expect(mocks.calculateSync).toHaveBeenCalledTimes(3); + expect(mocks.applySync).toHaveBeenCalledTimes(3); + expect(mocks.showToast).toHaveBeenCalledTimes(1); + expect(mocks.showToast).toHaveBeenCalledWith(staleToast); + expect(mocks.showDialog).not.toHaveBeenCalled(); +}); diff --git a/apps/yaak-client/commands/commands.tsx b/apps/yaak-client/commands/commands.tsx index 6bc57c3ad..dc9a7adbc 100644 --- a/apps/yaak-client/commands/commands.tsx +++ b/apps/yaak-client/commands/commands.tsx @@ -14,10 +14,11 @@ import { } from "@yaakapp-internal/ui"; import { activeWorkspaceIdAtom } from "../hooks/useActiveWorkspace"; import { createFastMutation } from "../hooks/useFastMutation"; -import { showDialog } from "../lib/dialog"; +import { hideDialog, showDialog } from "../lib/dialog"; import { jotaiStore } from "../lib/jotai"; import { showPrompt } from "../lib/prompt"; import { resolvedModelNameWithFolders } from "../lib/resolvedModelName"; +import { showToast } from "../lib/toast"; import { pluralizeCount } from "@yaakapp-internal/lib/pluralize"; export const createFolder = createFastMutation< @@ -52,24 +53,48 @@ export const createFolder = createFastMutation< }, }); +// Files can keep changing under a plan, so stop recalculating after a few stale applies. +const MAX_SYNC_ATTEMPTS = 3; + +function showStaleSyncToast() { + showToast({ + id: "sync-stale", + color: "warning", + message: "Files changed while syncing. Review the changes again.", + }); +} + export const syncWorkspace = createFastMutation< void, void, - { workspaceId: string; syncDir: string; force?: boolean } + { workspaceId: string; syncDir: string; force?: boolean; attempt?: number } >({ mutationKey: [], - mutationFn: async ({ workspaceId, syncDir, force }) => { + mutationFn: async ({ workspaceId, syncDir, force, attempt = 1 }) => { const ops = (await calculateSync(workspaceId, syncDir)) ?? []; if (ops.length === 0) { console.log("Nothing to sync", workspaceId, syncDir); + hideDialog("commit-sync"); return; } console.log("Syncing workspace", workspaceId, syncDir, ops); const dbOps = ops.filter((o) => o.type.startsWith("db")); - if (dbOps.length === 0) { - await applySync(workspaceId, syncDir, ops); + const apply = async () => { + if (await applySync(workspaceId, syncDir, ops)) { + hideDialog("commit-sync"); + } else if (attempt < MAX_SYNC_ATTEMPTS) { + await syncWorkspace.mutateAsync({ workspaceId, syncDir, force, attempt: attempt + 1 }); + } else { + hideDialog("commit-sync"); + showStaleSyncToast(); + } + }; + + if (dbOps.length === 0 || force) { + hideDialog("commit-sync"); + await apply(); return; } @@ -79,9 +104,8 @@ export const syncWorkspace = createFastMutation< console.log("Directory changes detected", { dbOps, ops }); - if (force) { - await applySync(workspaceId, syncDir, ops); - return; + if (attempt > 1) { + showStaleSyncToast(); } showDialog({ @@ -93,8 +117,7 @@ export const syncWorkspace = createFastMutation< className="h-full grid grid-rows-[auto_auto_minmax(0,1fr)_auto] gap-3" onSubmit={async (e) => { e.preventDefault(); - await applySync(workspaceId, syncDir, ops); - hide(); + await apply(); }} > {isDeletingWorkspace ? ( diff --git a/apps/yaak-client/commands/openWorkspaceFromSyncDir.tsx b/apps/yaak-client/commands/openWorkspaceFromSyncDir.tsx index 9c5b6d2f0..b12d90d84 100644 --- a/apps/yaak-client/commands/openWorkspaceFromSyncDir.tsx +++ b/apps/yaak-client/commands/openWorkspaceFromSyncDir.tsx @@ -17,7 +17,7 @@ export const openWorkspaceFromSyncDir = createFastMutation({ return; } - await applySync(workspace.id, dir, ops); + await applySync(workspace.id, dir, ops, { fsOnly: true }); await router.navigate({ to: "/workspaces/$workspaceId", diff --git a/crates-tauri/yaak-app-client/src/rpc_ext.rs b/crates-tauri/yaak-app-client/src/rpc_ext.rs index e62232f74..4a35ea1fa 100644 --- a/crates-tauri/yaak-app-client/src/rpc_ext.rs +++ b/crates-tauri/yaak-app-client/src/rpc_ext.rs @@ -1129,12 +1129,13 @@ async fn cmd_sync_calculate_fs( Ok(crate::sync_ext::cmd_sync_calculate_fs(&req.dir).await?) } -async fn cmd_sync_apply(ctx: ClientCtx, req: CmdSyncApplyReq) -> Result<()> { +async fn cmd_sync_apply(ctx: ClientCtx, req: CmdSyncApplyReq) -> Result { Ok(crate::sync_ext::cmd_sync_apply( ctx.window.app_handle().clone(), req.sync_ops, &req.sync_dir, &req.workspace_id, + req.fs_only, ) .await?) } diff --git a/crates-tauri/yaak-app-client/src/sync_ext.rs b/crates-tauri/yaak-app-client/src/sync_ext.rs index 6ba538096..e8f2a0a64 100644 --- a/crates-tauri/yaak-app-client/src/sync_ext.rs +++ b/crates-tauri/yaak-app-client/src/sync_ext.rs @@ -12,8 +12,8 @@ use tokio::sync::watch; use yaak_rpc_schema::WatchResult; use yaak_sync::error::Error::InvalidSyncDirectory; use yaak_sync::sync::{ - FsCandidate, SyncOp, apply_db_sync_ops, apply_fs_sync_ops, apply_sync_state_ops, - compute_sync_ops, get_db_candidates, get_fs_candidates, + FsCandidate, SyncOp, apply_db_sync_ops, apply_fs_sync_ops, apply_sync_ops, + apply_sync_state_ops, compute_sync_ops, get_db_candidates, get_fs_candidates, }; use yaak_sync::watch::{WatchEvent, watch_directory}; @@ -48,14 +48,28 @@ pub(crate) async fn cmd_sync_apply( sync_ops: Vec, sync_dir: &Path, workspace_id: &str, -) -> Result<()> { + fs_only: bool, +) -> Result { + let blobs = app_handle.blob_manager(); + if !fs_only || sync_ops.iter().any(|op| !matches!(op, SyncOp::DbCreate { .. })) { + let version = app_handle.package_info().version.to_string(); + return Ok(apply_sync_ops( + &app_handle.db_manager(), + &blobs, + &version, + workspace_id, + sync_dir, + sync_ops, + )?); + } + + // Opening a sync directory imports its files without existing DB sync state. // Files first, so the write transaction never waits on the filesystem let pending = apply_fs_sync_ops(workspace_id, sync_dir, sync_ops)?; - let blobs = app_handle.blob_manager(); app_handle.db_manager().with_tx(|tx| { let sync_state_ops = apply_db_sync_ops(tx, &blobs, workspace_id, sync_dir, pending)?; apply_sync_state_ops(tx, workspace_id, sync_dir, sync_state_ops)?; - Ok(()) + Ok(true) }) } diff --git a/crates/common/yaak-rpc-schema/bindings/gen_rpc.ts b/crates/common/yaak-rpc-schema/bindings/gen_rpc.ts index e0718d686..0345336d2 100644 --- a/crates/common/yaak-rpc-schema/bindings/gen_rpc.ts +++ b/crates/common/yaak-rpc-schema/bindings/gen_rpc.ts @@ -197,7 +197,7 @@ export type CmdSendHttpRequestReq = { environmentId: string | null, cookieJarId: export type CmdSetWorkspaceKeyReq = { workspaceId: string, key: string, }; -export type CmdSyncApplyReq = { syncOps: Array, syncDir: string, workspaceId: string, }; +export type CmdSyncApplyReq = { syncOps: Array, syncDir: string, workspaceId: string, fsOnly: boolean, }; export type CmdSyncCalculateFsReq = { dir: string, }; @@ -256,6 +256,6 @@ export type ModelsWebsocketEventsReq = { connectionId: string, }; export type ModelsWorkspaceModelsReq = { workspaceId: string | null, }; -export type RpcSchema = { cmd_metadata: [CmdMetadataReq, AppMetaData], cmd_template_tokens_to_string: [CmdTemplateTokensToStringReq, string], cmd_render_template: [CmdRenderTemplateReq, string], cmd_send_feedback: [CmdSendFeedbackReq, null], cmd_dismiss_notification: [CmdDismissNotificationReq, null], cmd_grpc_reflect: [CmdGrpcReflectReq, Array], cmd_grpc_go: [CmdGrpcGoReq, string], cmd_restart: [CmdRestartReq, null], cmd_send_ephemeral_request: [CmdSendEphemeralRequestReq, EphemeralHttpResponse], cmd_format_json: [CmdFormatJsonReq, string], cmd_format_graphql: [CmdFormatGraphqlReq, string], cmd_http_response_body: [CmdHttpResponseBodyReq, FilterResponse], cmd_http_response_body_path: [CmdHttpResponseBodyPathReq, string | null], cmd_http_request_body: [CmdHttpRequestBodyReq, Array | null], cmd_get_sse_events: [CmdGetSseEventsReq, Array], cmd_get_http_response_events: [CmdGetHttpResponseEventsReq, Array], cmd_import_data: [CmdImportDataReq, ImportPlan], cmd_commit_import: [CmdCommitImportReq, BatchUpsertResult], cmd_list_import_sources: [CmdListImportSourcesReq, Array], cmd_import_sources_for_origin: [CmdImportSourcesForOriginReq, Array], cmd_detect_import_source: [CmdDetectImportSourceReq, string], cmd_http_request_actions: [CmdHttpRequestActionsReq, Array], cmd_websocket_request_actions: [CmdWebsocketRequestActionsReq, Array], cmd_call_websocket_request_action: [CmdCallWebsocketRequestActionReq, null], cmd_workspace_actions: [CmdWorkspaceActionsReq, Array], cmd_call_workspace_action: [CmdCallWorkspaceActionReq, null], cmd_folder_actions: [CmdFolderActionsReq, Array], cmd_call_folder_action: [CmdCallFolderActionReq, null], cmd_grpc_request_actions: [CmdGrpcRequestActionsReq, Array], cmd_template_function_summaries: [CmdTemplateFunctionSummariesReq, Array], cmd_template_function_config: [CmdTemplateFunctionConfigReq, GetTemplateFunctionConfigResponse], cmd_get_http_authentication_summaries: [CmdGetHttpAuthenticationSummariesReq, Array], cmd_get_http_authentication_config: [CmdGetHttpAuthenticationConfigReq, GetHttpAuthenticationConfigResponse], cmd_call_http_request_action: [CmdCallHttpRequestActionReq, null], cmd_call_grpc_request_action: [CmdCallGrpcRequestActionReq, null], cmd_call_http_authentication_action: [CmdCallHttpAuthenticationActionReq, null], cmd_curl_to_request: [CmdCurlToRequestReq, HttpRequest], cmd_export_data: [CmdExportDataReq, string], cmd_create_example_workspace: [CmdCreateExampleWorkspaceReq, BatchUpsertResult], cmd_save_base64_to_binary: [CmdSaveBase64ToBinaryReq, null], cmd_save_response: [CmdSaveResponseReq, null], cmd_send_http_request: [CmdSendHttpRequestReq, HttpResponse], cmd_reload_plugins: [CmdReloadPluginsReq, Array<[string, string]>], cmd_plugin_info: [CmdPluginInfoReq, PluginMetadata], cmd_delete_all_grpc_connections: [CmdDeleteAllGrpcConnectionsReq, null], cmd_delete_send_history: [CmdDeleteSendHistoryReq, null], cmd_delete_all_http_responses: [CmdDeleteAllHttpResponsesReq, null], cmd_get_workspace_meta: [CmdGetWorkspaceMetaReq, WorkspaceMeta], cmd_new_child_window: [CmdNewChildWindowReq, null], cmd_new_main_window: [CmdNewMainWindowReq, null], cmd_check_for_updates: [CmdCheckForUpdatesReq, boolean], cmd_decrypt_template: [CmdDecryptTemplateReq, string], cmd_secure_template: [CmdSecureTemplateReq, string], cmd_get_themes: [CmdGetThemesReq, Array], cmd_enable_encryption: [CmdEnableEncryptionReq, null], cmd_reveal_workspace_key: [CmdRevealWorkspaceKeyReq, string], cmd_set_workspace_key: [CmdSetWorkspaceKeyReq, null], cmd_disable_encryption: [CmdDisableEncryptionReq, null], cmd_default_headers: [CmdDefaultHeadersReq, Array], models_upsert: [ModelsUpsertReq, string], models_delete: [ModelsDeleteReq, string], models_duplicate: [ModelsDuplicateReq, string], models_websocket_events: [ModelsWebsocketEventsReq, Array], models_grpc_events: [ModelsGrpcEventsReq, Array], models_get_settings: [ModelsGetSettingsReq, Settings], models_get_graphql_introspection: [ModelsGetGraphqlIntrospectionReq, GraphQlIntrospection | null], models_upsert_graphql_introspection: [ModelsUpsertGraphqlIntrospectionReq, GraphQlIntrospection], models_workspace_models: [ModelsWorkspaceModelsReq, string], cmd_git_checkout: [CmdGitCheckoutReq, string], cmd_git_branch: [CmdGitBranchReq, null], cmd_git_delete_branch: [CmdGitDeleteBranchReq, BranchDeleteResult], cmd_git_delete_remote_branch: [CmdGitDeleteRemoteBranchReq, null], cmd_git_merge_branch: [CmdGitMergeBranchReq, null], cmd_git_rename_branch: [CmdGitRenameBranchReq, null], cmd_git_status: [CmdGitStatusReq, GitStatusSummary], cmd_git_branch_info: [CmdGitBranchInfoReq, GitBranchInfo], cmd_git_worktree_status: [CmdGitWorktreeStatusReq, GitWorktreeStatus], cmd_git_log: [CmdGitLogReq, Array], cmd_git_log_for_file: [CmdGitLogForFileReq, Array], cmd_git_file_diff_for_commit: [CmdGitFileDiffForCommitReq, GitFileDiff], cmd_git_initialize: [CmdGitInitializeReq, null], cmd_git_clone: [CmdGitCloneReq, CloneResult], cmd_git_commit: [CmdGitCommitReq, null], cmd_git_fetch_all: [CmdGitFetchAllReq, null], cmd_git_push: [CmdGitPushReq, PushResult], cmd_git_pull: [CmdGitPullReq, PullResult], cmd_git_pull_force_reset: [CmdGitPullForceResetReq, PullResult], cmd_git_pull_merge: [CmdGitPullMergeReq, PullResult], cmd_git_add: [CmdGitAddReq, null], cmd_git_unstage: [CmdGitUnstageReq, null], cmd_git_reset_changes: [CmdGitResetChangesReq, null], cmd_git_restore_files: [CmdGitRestoreFilesReq, null], cmd_git_restore_file_from_commit: [CmdGitRestoreFileFromCommitReq, null], cmd_git_add_credential: [CmdGitAddCredentialReq, null], cmd_git_remotes: [CmdGitRemotesReq, Array], cmd_git_add_remote: [CmdGitAddRemoteReq, GitRemote], cmd_git_rm_remote: [CmdGitRmRemoteReq, null], cmd_sync_calculate: [CmdSyncCalculateReq, Array], cmd_sync_calculate_fs: [CmdSyncCalculateFsReq, Array], cmd_sync_apply: [CmdSyncApplyReq, null], cmd_ws_delete_connections: [CmdWsDeleteConnectionsReq, null], cmd_ws_send: [CmdWsSendReq, WebsocketConnection], cmd_ws_close: [CmdWsCloseReq, WebsocketConnection], cmd_ws_connect: [CmdWsConnectReq, WebsocketConnection], cmd_plugins_search: [CmdPluginsSearchReq, PluginSearchResponse], cmd_plugins_install: [CmdPluginsInstallReq, null], cmd_plugins_install_from_directory: [CmdPluginsInstallFromDirectoryReq, Plugin], cmd_plugins_uninstall: [CmdPluginsUninstallReq, Plugin], cmd_plugin_init_errors: [CmdPluginInitErrorsReq, Array<[string, string]>], cmd_plugins_updates: [CmdPluginsUpdatesReq, PluginUpdatesResponse], cmd_plugins_update_all: [CmdPluginsUpdateAllReq, Array], cmd_git_watch_worktree_status: [CmdGitWatchWorktreeStatusReq, GitWatchResult], cmd_sync_watch: [CmdSyncWatchReq, WatchResult], }; +export type RpcSchema = { cmd_metadata: [CmdMetadataReq, AppMetaData], cmd_template_tokens_to_string: [CmdTemplateTokensToStringReq, string], cmd_render_template: [CmdRenderTemplateReq, string], cmd_send_feedback: [CmdSendFeedbackReq, null], cmd_dismiss_notification: [CmdDismissNotificationReq, null], cmd_grpc_reflect: [CmdGrpcReflectReq, Array], cmd_grpc_go: [CmdGrpcGoReq, string], cmd_restart: [CmdRestartReq, null], cmd_send_ephemeral_request: [CmdSendEphemeralRequestReq, EphemeralHttpResponse], cmd_format_json: [CmdFormatJsonReq, string], cmd_format_graphql: [CmdFormatGraphqlReq, string], cmd_http_response_body: [CmdHttpResponseBodyReq, FilterResponse], cmd_http_response_body_path: [CmdHttpResponseBodyPathReq, string | null], cmd_http_request_body: [CmdHttpRequestBodyReq, Array | null], cmd_get_sse_events: [CmdGetSseEventsReq, Array], cmd_get_http_response_events: [CmdGetHttpResponseEventsReq, Array], cmd_import_data: [CmdImportDataReq, ImportPlan], cmd_commit_import: [CmdCommitImportReq, BatchUpsertResult], cmd_list_import_sources: [CmdListImportSourcesReq, Array], cmd_import_sources_for_origin: [CmdImportSourcesForOriginReq, Array], cmd_detect_import_source: [CmdDetectImportSourceReq, string], cmd_http_request_actions: [CmdHttpRequestActionsReq, Array], cmd_websocket_request_actions: [CmdWebsocketRequestActionsReq, Array], cmd_call_websocket_request_action: [CmdCallWebsocketRequestActionReq, null], cmd_workspace_actions: [CmdWorkspaceActionsReq, Array], cmd_call_workspace_action: [CmdCallWorkspaceActionReq, null], cmd_folder_actions: [CmdFolderActionsReq, Array], cmd_call_folder_action: [CmdCallFolderActionReq, null], cmd_grpc_request_actions: [CmdGrpcRequestActionsReq, Array], cmd_template_function_summaries: [CmdTemplateFunctionSummariesReq, Array], cmd_template_function_config: [CmdTemplateFunctionConfigReq, GetTemplateFunctionConfigResponse], cmd_get_http_authentication_summaries: [CmdGetHttpAuthenticationSummariesReq, Array], cmd_get_http_authentication_config: [CmdGetHttpAuthenticationConfigReq, GetHttpAuthenticationConfigResponse], cmd_call_http_request_action: [CmdCallHttpRequestActionReq, null], cmd_call_grpc_request_action: [CmdCallGrpcRequestActionReq, null], cmd_call_http_authentication_action: [CmdCallHttpAuthenticationActionReq, null], cmd_curl_to_request: [CmdCurlToRequestReq, HttpRequest], cmd_export_data: [CmdExportDataReq, string], cmd_create_example_workspace: [CmdCreateExampleWorkspaceReq, BatchUpsertResult], cmd_save_base64_to_binary: [CmdSaveBase64ToBinaryReq, null], cmd_save_response: [CmdSaveResponseReq, null], cmd_send_http_request: [CmdSendHttpRequestReq, HttpResponse], cmd_reload_plugins: [CmdReloadPluginsReq, Array<[string, string]>], cmd_plugin_info: [CmdPluginInfoReq, PluginMetadata], cmd_delete_all_grpc_connections: [CmdDeleteAllGrpcConnectionsReq, null], cmd_delete_send_history: [CmdDeleteSendHistoryReq, null], cmd_delete_all_http_responses: [CmdDeleteAllHttpResponsesReq, null], cmd_get_workspace_meta: [CmdGetWorkspaceMetaReq, WorkspaceMeta], cmd_new_child_window: [CmdNewChildWindowReq, null], cmd_new_main_window: [CmdNewMainWindowReq, null], cmd_check_for_updates: [CmdCheckForUpdatesReq, boolean], cmd_decrypt_template: [CmdDecryptTemplateReq, string], cmd_secure_template: [CmdSecureTemplateReq, string], cmd_get_themes: [CmdGetThemesReq, Array], cmd_enable_encryption: [CmdEnableEncryptionReq, null], cmd_reveal_workspace_key: [CmdRevealWorkspaceKeyReq, string], cmd_set_workspace_key: [CmdSetWorkspaceKeyReq, null], cmd_disable_encryption: [CmdDisableEncryptionReq, null], cmd_default_headers: [CmdDefaultHeadersReq, Array], models_upsert: [ModelsUpsertReq, string], models_delete: [ModelsDeleteReq, string], models_duplicate: [ModelsDuplicateReq, string], models_websocket_events: [ModelsWebsocketEventsReq, Array], models_grpc_events: [ModelsGrpcEventsReq, Array], models_get_settings: [ModelsGetSettingsReq, Settings], models_get_graphql_introspection: [ModelsGetGraphqlIntrospectionReq, GraphQlIntrospection | null], models_upsert_graphql_introspection: [ModelsUpsertGraphqlIntrospectionReq, GraphQlIntrospection], models_workspace_models: [ModelsWorkspaceModelsReq, string], cmd_git_checkout: [CmdGitCheckoutReq, string], cmd_git_branch: [CmdGitBranchReq, null], cmd_git_delete_branch: [CmdGitDeleteBranchReq, BranchDeleteResult], cmd_git_delete_remote_branch: [CmdGitDeleteRemoteBranchReq, null], cmd_git_merge_branch: [CmdGitMergeBranchReq, null], cmd_git_rename_branch: [CmdGitRenameBranchReq, null], cmd_git_status: [CmdGitStatusReq, GitStatusSummary], cmd_git_branch_info: [CmdGitBranchInfoReq, GitBranchInfo], cmd_git_worktree_status: [CmdGitWorktreeStatusReq, GitWorktreeStatus], cmd_git_log: [CmdGitLogReq, Array], cmd_git_log_for_file: [CmdGitLogForFileReq, Array], cmd_git_file_diff_for_commit: [CmdGitFileDiffForCommitReq, GitFileDiff], cmd_git_initialize: [CmdGitInitializeReq, null], cmd_git_clone: [CmdGitCloneReq, CloneResult], cmd_git_commit: [CmdGitCommitReq, null], cmd_git_fetch_all: [CmdGitFetchAllReq, null], cmd_git_push: [CmdGitPushReq, PushResult], cmd_git_pull: [CmdGitPullReq, PullResult], cmd_git_pull_force_reset: [CmdGitPullForceResetReq, PullResult], cmd_git_pull_merge: [CmdGitPullMergeReq, PullResult], cmd_git_add: [CmdGitAddReq, null], cmd_git_unstage: [CmdGitUnstageReq, null], cmd_git_reset_changes: [CmdGitResetChangesReq, null], cmd_git_restore_files: [CmdGitRestoreFilesReq, null], cmd_git_restore_file_from_commit: [CmdGitRestoreFileFromCommitReq, null], cmd_git_add_credential: [CmdGitAddCredentialReq, null], cmd_git_remotes: [CmdGitRemotesReq, Array], cmd_git_add_remote: [CmdGitAddRemoteReq, GitRemote], cmd_git_rm_remote: [CmdGitRmRemoteReq, null], cmd_sync_calculate: [CmdSyncCalculateReq, Array], cmd_sync_calculate_fs: [CmdSyncCalculateFsReq, Array], cmd_sync_apply: [CmdSyncApplyReq, boolean], cmd_ws_delete_connections: [CmdWsDeleteConnectionsReq, null], cmd_ws_send: [CmdWsSendReq, WebsocketConnection], cmd_ws_close: [CmdWsCloseReq, WebsocketConnection], cmd_ws_connect: [CmdWsConnectReq, WebsocketConnection], cmd_plugins_search: [CmdPluginsSearchReq, PluginSearchResponse], cmd_plugins_install: [CmdPluginsInstallReq, null], cmd_plugins_install_from_directory: [CmdPluginsInstallFromDirectoryReq, Plugin], cmd_plugins_uninstall: [CmdPluginsUninstallReq, Plugin], cmd_plugin_init_errors: [CmdPluginInitErrorsReq, Array<[string, string]>], cmd_plugins_updates: [CmdPluginsUpdatesReq, PluginUpdatesResponse], cmd_plugins_update_all: [CmdPluginsUpdateAllReq, Array], cmd_git_watch_worktree_status: [CmdGitWatchWorktreeStatusReq, GitWatchResult], cmd_sync_watch: [CmdSyncWatchReq, WatchResult], }; export type WatchResult = { unlistenEvent: string, }; diff --git a/crates/common/yaak-rpc-schema/bindings/gen_sync.ts b/crates/common/yaak-rpc-schema/bindings/gen_sync.ts index 49b168b50..d0e16a4ee 100644 --- a/crates/common/yaak-rpc-schema/bindings/gen_sync.ts +++ b/crates/common/yaak-rpc-schema/bindings/gen_sync.ts @@ -3,4 +3,7 @@ import type { SyncModel, SyncState } from "./gen_models"; export type FsCandidate = { "type": "FsCandidate", model: SyncModel, relPath: string, checksum: string, }; -export type SyncOp = { "type": "fsCreate", model: SyncModel, } | { "type": "fsUpdate", model: SyncModel, state: SyncState, } | { "type": "fsDelete", state: SyncState, fs: FsCandidate | null, } | { "type": "dbCreate", fs: FsCandidate, } | { "type": "dbUpdate", state: SyncState, fs: FsCandidate, } | { "type": "dbDelete", model: SyncModel, state: SyncState, } | { "type": "ignorePrivate", model: SyncModel, }; +/** + * Outbound operations record the file candidate checksum, or `None` if it was absent. + */ +export type SyncOp = { "type": "fsCreate", model: SyncModel, fsChecksum: string | null, } | { "type": "fsUpdate", model: SyncModel, state: SyncState, fsChecksum: string | null, } | { "type": "fsDelete", state: SyncState, fs: FsCandidate | null, } | { "type": "dbCreate", fs: FsCandidate, } | { "type": "dbUpdate", model: SyncModel, state: SyncState, fs: FsCandidate, } | { "type": "dbDelete", model: SyncModel, state: SyncState, } | { "type": "ignorePrivate", model: SyncModel, }; diff --git a/crates/common/yaak-rpc-schema/src/lib.rs b/crates/common/yaak-rpc-schema/src/lib.rs index 8f6d8ad83..791f0271d 100644 --- a/crates/common/yaak-rpc-schema/src/lib.rs +++ b/crates/common/yaak-rpc-schema/src/lib.rs @@ -834,6 +834,8 @@ pub struct CmdSyncApplyReq { pub sync_ops: Vec, pub sync_dir: PathBuf, pub workspace_id: String, + #[serde(default)] + pub fs_only: bool, } #[derive(Debug, Deserialize, TS)] @@ -1024,7 +1026,7 @@ macro_rules! with_commands { cmd_git_rm_remote(CmdGitRmRemoteReq) -> (), cmd_sync_calculate(CmdSyncCalculateReq) -> Vec, cmd_sync_calculate_fs(CmdSyncCalculateFsReq) -> Vec, - cmd_sync_apply(CmdSyncApplyReq) -> (), + cmd_sync_apply(CmdSyncApplyReq) -> bool, cmd_ws_delete_connections(CmdWsDeleteConnectionsReq) -> (), cmd_ws_send(CmdWsSendReq) -> WebsocketConnection, cmd_ws_close(CmdWsCloseReq) -> WebsocketConnection, diff --git a/crates/yaak-models/src/models.rs b/crates/yaak-models/src/models.rs index 7c8279bc4..c1a6ee77e 100644 --- a/crates/yaak-models/src/models.rs +++ b/crates/yaak-models/src/models.rs @@ -2839,7 +2839,7 @@ impl UpsertModelInfo for Plugin { } } -#[derive(Debug, Clone, Serialize, Deserialize, Default, TS)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default, TS)] #[serde(default, rename_all = "camelCase")] #[ts(export, export_to = "gen_models.ts")] #[enum_def(table_name = "sync_states")] diff --git a/crates/yaak-sync/bindings/gen_sync.ts b/crates/yaak-sync/bindings/gen_sync.ts index 49b168b50..d0e16a4ee 100644 --- a/crates/yaak-sync/bindings/gen_sync.ts +++ b/crates/yaak-sync/bindings/gen_sync.ts @@ -3,4 +3,7 @@ import type { SyncModel, SyncState } from "./gen_models"; export type FsCandidate = { "type": "FsCandidate", model: SyncModel, relPath: string, checksum: string, }; -export type SyncOp = { "type": "fsCreate", model: SyncModel, } | { "type": "fsUpdate", model: SyncModel, state: SyncState, } | { "type": "fsDelete", state: SyncState, fs: FsCandidate | null, } | { "type": "dbCreate", fs: FsCandidate, } | { "type": "dbUpdate", state: SyncState, fs: FsCandidate, } | { "type": "dbDelete", model: SyncModel, state: SyncState, } | { "type": "ignorePrivate", model: SyncModel, }; +/** + * Outbound operations record the file candidate checksum, or `None` if it was absent. + */ +export type SyncOp = { "type": "fsCreate", model: SyncModel, fsChecksum: string | null, } | { "type": "fsUpdate", model: SyncModel, state: SyncState, fsChecksum: string | null, } | { "type": "fsDelete", state: SyncState, fs: FsCandidate | null, } | { "type": "dbCreate", fs: FsCandidate, } | { "type": "dbUpdate", model: SyncModel, state: SyncState, fs: FsCandidate, } | { "type": "dbDelete", model: SyncModel, state: SyncState, } | { "type": "ignorePrivate", model: SyncModel, }; diff --git a/crates/yaak-sync/index.ts b/crates/yaak-sync/index.ts index cb8bf087a..96a1fbe42 100644 --- a/crates/yaak-sync/index.ts +++ b/crates/yaak-sync/index.ts @@ -16,11 +16,17 @@ export async function calculateSyncFsOnly(dir: string) { return platform.rpc("cmd_sync_calculate_fs", { dir }); } -export async function applySync(workspaceId: string, syncDir: string, syncOps: SyncOp[]) { - return platform.rpc("cmd_sync_apply", { +export async function applySync( + workspaceId: string, + syncDir: string, + syncOps: SyncOp[], + options?: { fsOnly: boolean }, +) { + return platform.rpc("cmd_sync_apply", { workspaceId, syncDir, syncOps: syncOps, + fsOnly: options?.fsOnly ?? false, }); } diff --git a/crates/yaak-sync/src/sync.rs b/crates/yaak-sync/src/sync.rs index b24f47728..d136a2b31 100644 --- a/crates/yaak-sync/src/sync.rs +++ b/crates/yaak-sync/src/sync.rs @@ -17,18 +17,24 @@ use yaak_models::models::{ Environment, Folder, GrpcRequest, HttpRequest, SyncState, WebsocketRequest, Workspace, WorkspaceMeta, }; +use yaak_models::query_manager::QueryManager; use yaak_models::util::{UpdateSource, get_workspace_export_resources}; -#[derive(Debug, Clone, Serialize, Deserialize, TS)] +/// Outbound operations record the file candidate checksum, or `None` if it was absent. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TS)] #[serde(rename_all = "camelCase", tag = "type")] #[ts(export, export_to = "gen_sync.ts")] pub enum SyncOp { FsCreate { model: SyncModel, + #[serde(rename = "fsChecksum")] + fs_checksum: Option, }, FsUpdate { model: SyncModel, state: SyncState, + #[serde(rename = "fsChecksum")] + fs_checksum: Option, }, FsDelete { state: SyncState, @@ -38,6 +44,8 @@ pub enum SyncOp { fs: FsCandidate, }, DbUpdate { + // The DB version the user reviewed, so a subsequent edit invalidates the plan. + model: SyncModel, state: SyncState, fs: FsCandidate, }, @@ -56,7 +64,7 @@ impl SyncOp { SyncOp::DbCreate { fs } => fs.model.workspace_id(), SyncOp::DbDelete { model, .. } => model.workspace_id(), SyncOp::DbUpdate { state, .. } => state.workspace_id.clone(), - SyncOp::FsCreate { model } => model.workspace_id(), + SyncOp::FsCreate { model, .. } => model.workspace_id(), SyncOp::FsDelete { state, .. } => state.workspace_id.clone(), SyncOp::FsUpdate { state, .. } => state.workspace_id.clone(), SyncOp::IgnorePrivate { model } => model.workspace_id(), @@ -68,7 +76,7 @@ impl Display for SyncOp { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { f.write_str( match self { - SyncOp::FsCreate { model } => format!("fs_create({})", model.id()), + SyncOp::FsCreate { model, .. } => format!("fs_create({})", model.id()), SyncOp::FsUpdate { model, .. } => format!("fs_update({})", model.id()), SyncOp::FsDelete { state, .. } => format!("fs_delete({})", state.model_id), SyncOp::DbCreate { fs } => format!("db_create({})", fs.model.id()), @@ -101,7 +109,7 @@ impl DbCandidate { } } -#[derive(Debug, Clone, Serialize, Deserialize, TS)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TS)] #[serde(rename_all = "camelCase", tag = "type")] #[ts(export, export_to = "gen_sync.ts")] pub struct FsCandidate { @@ -285,13 +293,15 @@ pub fn compute_sync_ops( } // DB modified <-> FS missing - (Some(DbCandidate::Modified(model, sync_state)), None) => { - SyncOp::FsUpdate { model: model.to_owned(), state: sync_state.to_owned() } - } + (Some(DbCandidate::Modified(model, sync_state)), None) => SyncOp::FsUpdate { + model: model.to_owned(), + state: sync_state.to_owned(), + fs_checksum: None, + }, // DB added <-> FS missing (Some(DbCandidate::Added(model)), None) => { - SyncOp::FsCreate { model: model.to_owned() } + SyncOp::FsCreate { model: model.to_owned(), fs_checksum: None } } // DB deleted <-> FS missing @@ -301,11 +311,12 @@ pub fn compute_sync_ops( } // DB unchanged <-> FS exists - (Some(DbCandidate::Unmodified(_, sync_state)), Some(fs_candidate)) => { + (Some(DbCandidate::Unmodified(model, sync_state)), Some(fs_candidate)) => { if sync_state.checksum == fs_candidate.checksum { return None; } else { SyncOp::DbUpdate { + model: model.to_owned(), state: sync_state.to_owned(), fs: fs_candidate.to_owned(), } @@ -315,23 +326,35 @@ pub fn compute_sync_ops( // DB modified <-> FS exists (Some(DbCandidate::Modified(model, sync_state)), Some(fs_candidate)) => { if sync_state.checksum == fs_candidate.checksum { - SyncOp::FsUpdate { model: model.to_owned(), state: sync_state.to_owned() } + SyncOp::FsUpdate { + model: model.to_owned(), + state: sync_state.to_owned(), + fs_checksum: Some(fs_candidate.checksum.clone()), + } } else if model.updated_at() < fs_candidate.model.updated_at() { // CONFLICT! Write to DB if the fs model is newer SyncOp::DbUpdate { + model: model.to_owned(), state: sync_state.to_owned(), fs: fs_candidate.to_owned(), } } else { // CONFLICT! Write to FS if the db model is newer - SyncOp::FsUpdate { model: model.to_owned(), state: sync_state.to_owned() } + SyncOp::FsUpdate { + model: model.to_owned(), + state: sync_state.to_owned(), + fs_checksum: Some(fs_candidate.checksum.clone()), + } } } // DB added <-> FS anything - (Some(DbCandidate::Added(model)), Some(_)) => { + (Some(DbCandidate::Added(model)), Some(fs_candidate)) => { // This would be super rare (impossible?), so let's follow the user's intention - SyncOp::FsCreate { model: model.to_owned() } + SyncOp::FsCreate { + model: model.to_owned(), + fs_checksum: Some(fs_candidate.checksum.clone()), + } } // DB deleted <-> FS exists @@ -384,6 +407,104 @@ fn workspace_models(db: &ClientDb, version: &str, workspace_id: &str) -> Result< Ok(sync_models) } +/// Apply a reviewed workspace plan only while its inputs still match. +/// Returns `false` when the caller needs to calculate and review a new plan. +pub fn apply_sync_ops( + db: &QueryManager, + blobs: &BlobManager, + version: &str, + workspace_id: &str, + sync_dir: &Path, + mut sync_ops: Vec, +) -> Result { + let db_candidates = get_db_candidates(&db.connect(), version, workspace_id, sync_dir)?; + let fs_candidates = workspace_fs_candidates(sync_dir, workspace_id, &db_candidates)?; + if !sync_ops_match(&sync_ops, compute_sync_ops(db_candidates, fs_candidates.clone()), false) { + return Ok(false); + } + + let pending = apply_fs_sync_ops(workspace_id, sync_dir, sync_ops.clone())?; + let has_db_ops = sync_ops.iter().any(is_db_op); + // File deletions are complete; their remaining work only removes sync state. + sync_ops.retain(|op| !matches!(op, SyncOp::FsDelete { .. })); + + // Inbound changes need a fresh file scan. It happens here, before the write + // transaction, so the transaction never waits on the filesystem. + let fs_candidates = if has_db_ops { + let db_candidates = get_db_candidates(&db.connect(), version, workspace_id, sync_dir)?; + let fs_candidates = workspace_fs_candidates(sync_dir, workspace_id, &db_candidates)?; + if !written_sync_ops_match(&sync_ops, db_candidates, fs_candidates.clone()) { + return Ok(false); + } + fs_candidates + } else { + fs_candidates + }; + + db.with_tx(|tx| { + // Another writer may have changed the DB while files were being written or + // while waiting for the writer. Files were checked above; only the DB is reread. + let db_candidates = get_db_candidates(tx, version, workspace_id, sync_dir)?; + if !written_sync_ops_match(&sync_ops, db_candidates, fs_candidates) { + return Ok(false); + } + let states = apply_db_sync_ops(tx, blobs, workspace_id, sync_dir, pending)?; + apply_sync_state_ops(tx, workspace_id, sync_dir, states)?; + Ok(true) + }) +} + +/// The sync directory's candidates for one workspace. +fn workspace_fs_candidates( + sync_dir: &Path, + workspace_id: &str, + db_candidates: &[DbCandidate], +) -> Result> { + if !sync_dir.exists() { + return Err(Error::InvalidSyncDirectory(sync_dir.to_string_lossy().to_string())); + } + Ok(get_fs_candidates(sync_dir, db_candidates)? + .into_iter() + .filter(|fs| fs.model.workspace_id() == workspace_id) + .collect()) +} + +/// Whether a plan, minus its completed file deletions, still matches once its files are written. +fn written_sync_ops_match( + expected: &[SyncOp], + db_candidates: Vec, + fs_candidates: Vec, +) -> bool { + let current = compute_sync_ops(db_candidates, fs_candidates) + .into_iter() + .filter(|op| !matches!(op, SyncOp::FsDelete { .. })) + .collect(); + sync_ops_match(expected, current, true) +} + +fn sync_ops_match(expected: &[SyncOp], current: Vec, files_written: bool) -> bool { + // The join emits one operation per ID, in arbitrary order. + expected.len() == current.len() + && current.iter().all(|op| { + expected.iter().any(|reviewed| match (op, reviewed) { + // Once outbound files are written, only their DB/state inputs must still match. + ( + SyncOp::FsCreate { model, .. }, + SyncOp::FsCreate { model: reviewed_model, .. }, + ) if files_written => model == reviewed_model, + ( + SyncOp::FsUpdate { model, state, .. }, + SyncOp::FsUpdate { model: reviewed_model, state: reviewed_state, .. }, + ) if files_written => model == reviewed_model && state == reviewed_state, + _ => op == reviewed, + }) + }) +} + +fn is_db_op(op: &SyncOp) -> bool { + matches!(op, SyncOp::DbCreate { .. } | SyncOp::DbUpdate { .. } | SyncOp::DbDelete { .. }) +} + /// The database half of a sync apply, ready to run once the files are on disk. pub struct PendingDbSyncOps { sync_state_ops: Vec, @@ -399,7 +520,7 @@ pub struct PendingDbSyncOps { /// Apply the filesystem half of the sync operations: create, rewrite and /// delete files. Returns the database half, for [`apply_db_sync_ops`]. /// -/// Split this way so the file work, which can be slow, happens before the +/// Split this way so filesystem writes, which can be slow, happen before the /// write transaction is opened rather than inside it. pub fn apply_fs_sync_ops( workspace_id: &str, @@ -432,14 +553,14 @@ pub fn apply_fs_sync_ops( } let state_op = match op { - SyncOp::FsCreate { model } => { + SyncOp::FsCreate { model, .. } => { let rel_path = derive_model_filename(&model); let abs_path = sync_dir.join(rel_path.clone()); let (content, checksum) = model.to_file_contents(&rel_path)?; write_sync_file(&abs_path, |file| file.write_all(&content))?; SyncStateOp::Create { model_id: model.id(), checksum, rel_path } } - SyncOp::FsUpdate { model, state } => { + SyncOp::FsUpdate { model, state, .. } => { // Always write the existing path let rel_path = Path::new(&state.rel_path); let abs_path = Path::new(&state.sync_dir).join(&rel_path); @@ -470,7 +591,7 @@ pub fn apply_fs_sync_ops( rel_path: fs.rel_path.to_owned(), } } - SyncOp::DbUpdate { state, fs } => { + SyncOp::DbUpdate { state, fs, .. } => { pending.push_upsert(fs.model); SyncStateOp::Update { state: state.to_owned(), diff --git a/crates/yaak-sync/tests/sync_files.rs b/crates/yaak-sync/tests/sync_files.rs index 12e1b66e7..22dc7df42 100644 --- a/crates/yaak-sync/tests/sync_files.rs +++ b/crates/yaak-sync/tests/sync_files.rs @@ -2,13 +2,14 @@ use std::fs; use std::path::PathBuf; use tempfile::TempDir; use yaak_models::blob_manager::BlobManager; -use yaak_models::models::{HttpRequest, SyncState, Workspace}; +use yaak_models::models::{HttpRequest, HttpResponse, SyncState, Workspace}; use yaak_models::query_manager::QueryManager; +use yaak_models::util::UpdateSource; use yaak_sync::error::{Error, Result}; use yaak_sync::models::SyncModel; use yaak_sync::sync::{ - SyncOp, apply_db_sync_ops, apply_fs_sync_ops, apply_sync_state_ops, compute_sync_ops, - get_db_candidates, get_fs_candidates, + SyncOp, apply_db_sync_ops, apply_fs_sync_ops, apply_sync_ops, apply_sync_state_ops, + compute_sync_ops, get_db_candidates, get_fs_candidates, }; struct Fixture { @@ -22,8 +23,12 @@ struct Fixture { impl Fixture { fn new(request_filename: &str) -> Self { - let (db, blobs, _events) = yaak_models::init_in_memory().unwrap(); let dir = tempfile::tempdir().unwrap(); + let (db, blobs, _events) = yaak_models::init_standalone( + dir.path().join(".db/models.sqlite"), + dir.path().join(".db/blobs.sqlite"), + ) + .unwrap(); let workspace = Workspace { id: "wk_sync_test".into(), ..Default::default() }; let request = HttpRequest { id: "rq_sync_test".into(), @@ -75,6 +80,11 @@ impl Fixture { .unwrap(); } + fn apply_reviewed(&self, ops: Vec) -> bool { + apply_sync_ops(&self.db, &self.blobs, "test", &self.workspace_id, self.dir.path(), ops) + .unwrap() + } + fn states(&self) -> Vec { let mut states = self .db @@ -215,3 +225,312 @@ fn unrelated_files_are_still_ignored() { fs::create_dir(fixture.dir.path().join("other-files")).unwrap(); assert!(fixture.calculate().unwrap().is_empty()); } + +#[test] +fn restored_file_invalidates_reviewed_deletion_and_preserves_history() { + let fixture = Fixture::new("request.yaml"); + let response = fixture + .db + .with_tx::<_, Error>(|tx| { + Ok(tx.upsert_http_response( + &HttpResponse { + id: "rs_sync_test".into(), + request_id: fixture.request.id.clone(), + workspace_id: fixture.workspace_id.clone(), + status: 200, + ..Default::default() + }, + &UpdateSource::Background, + &fixture.blobs, + )?) + }) + .unwrap(); + let saved = fs::read(&fixture.request_path).unwrap(); + fs::remove_file(&fixture.request_path).unwrap(); + let ops = fixture.calculate().unwrap(); + assert!(matches!(ops.as_slice(), [SyncOp::DbDelete { .. }])); + fs::write(&fixture.request_path, &saved).unwrap(); + let before = fixture.states(); + + assert!(!fixture.apply_reviewed(ops)); + assert_eq!(fixture.db_request().name, "Keep this request"); + assert_eq!( + serde_json::to_value(fixture.db.connect().get_http_response(&response.id).unwrap()) + .unwrap(), + serde_json::to_value(response).unwrap() + ); + assert_eq!(fixture.states(), before); + assert_eq!(fs::read(&fixture.request_path).unwrap(), saved); + assert!(fixture.calculate().unwrap().is_empty()); +} + +#[test] +fn newer_database_edit_invalidates_reviewed_update_even_when_file_timestamp_wins() { + for file_is_newer in [false, true] { + let fixture = Fixture::new("request.yaml"); + fixture.edit(|value| { + value["name"] = "Previously reviewed file version".into(); + if file_is_newer { + value["updatedAt"] = serde_json::to_value( + (chrono::Utc::now() + chrono::Duration::days(1)).naive_utc(), + ) + .unwrap(); + } + }); + let ops = fixture.calculate().unwrap(); + assert!(matches!(ops.as_slice(), [SyncOp::DbUpdate { .. }])); + let before = fixture.states(); + let mut request = fixture.db_request(); + request.name = "New edit from another window".into(); + fixture + .db + .with_tx::<_, Error>(|tx| { + tx.upsert_http_request(&request, &UpdateSource::from_window_label("test"))?; + Ok(()) + }) + .unwrap(); + if file_is_newer { + assert!(matches!(fixture.calculate().unwrap().as_slice(), [SyncOp::DbUpdate { .. }])); + } + + assert!(!fixture.apply_reviewed(ops)); + assert_eq!(fixture.db_request().name, request.name); + assert_eq!(fixture.states(), before); + } +} + +#[test] +fn changed_file_invalidates_reviewed_update_until_reviewed_again() { + let fixture = Fixture::new("request.yaml"); + fixture.edit(|value| value["name"] = "Reviewed version".into()); + let ops = fixture.calculate().unwrap(); + fixture.edit(|value| value["name"] = "New file version".into()); + let before = fixture.states(); + + assert!(!fixture.apply_reviewed(ops)); + assert_eq!(fixture.db_request().name, "Keep this request"); + assert_eq!(fixture.states(), before); + assert!(fixture.apply_reviewed(fixture.calculate().unwrap())); + assert_eq!(fixture.db_request().name, "New file version"); + assert!(fixture.calculate().unwrap().is_empty()); +} + +#[test] +fn unchanged_reviewed_plan_applies_in_any_order_and_cannot_be_replayed() { + let fixture = Fixture::new("request.yaml"); + fs::remove_file(&fixture.request_path).unwrap(); + let path = fixture.dir.path().join("workspace.yaml"); + let mut workspace = fixture.db.connect().get_workspace(&fixture.workspace_id).unwrap(); + workspace.name = "Reviewed workspace".into(); + let (contents, _) = SyncModel::Workspace(workspace).to_file_contents(&path).unwrap(); + fs::write(path, contents).unwrap(); + let mut ops = fixture.calculate().unwrap(); + assert_eq!(ops.len(), 2); + ops.reverse(); + + assert!(fixture.apply_reviewed(ops.clone())); + assert!(fixture.db.connect().get_http_request(&fixture.request.id).is_err()); + assert_eq!( + fixture.db.connect().get_workspace(&fixture.workspace_id).unwrap().name, + "Reviewed workspace" + ); + assert!(fixture.calculate().unwrap().is_empty()); + let before = fixture.states(); + assert!(!fixture.apply_reviewed(ops)); + assert_eq!(fixture.states(), before); +} + +#[test] +fn database_changes_while_waiting_for_writer_invalidate_reviewed_plan() { + use std::sync::mpsc; + use std::time::{Duration, Instant}; + + // Files are checked before the write transaction opens, so only database changes made + // while waiting for the writer are caught here. + let fixture = Fixture::new("request.yaml"); + fixture.edit(|value| value["name"] = "Incoming file version".into()); + fixture + .db + .with_tx::<_, Error>(|tx| { + let mut workspace = tx.get_workspace(&fixture.workspace_id)?; + workspace.name = "Outbound workspace version".into(); + tx.upsert_workspace(&workspace, &UpdateSource::from_window_label("test"))?; + Ok(()) + }) + .unwrap(); + let ops = fixture.calculate().unwrap(); + assert_eq!(ops.len(), 2); + let before = fixture.states(); + let workspace_path = fixture.dir.path().join("workspace.yaml"); + let (ready_tx, ready_rx) = mpsc::channel(); + let (resume_tx, resume_rx) = mpsc::channel(); + std::thread::scope(|scope| { + let fixture = &fixture; + let writer = scope.spawn(move || { + fixture + .db + .with_tx::<_, Error>(|tx| { + ready_tx.send(()).unwrap(); + resume_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + let mut request = tx.get_http_request(&fixture.request.id)?; + request.name = "Concurrent database edit".into(); + tx.upsert_http_request(&request, &UpdateSource::from_window_label("test"))?; + Ok(()) + }) + .unwrap(); + }); + ready_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + let applying = scope.spawn(move || fixture.apply_reviewed(ops)); + let deadline = Instant::now() + Duration::from_secs(5); + let mut published = false; + while Instant::now() < deadline { + if fs::read_to_string(&workspace_path).unwrap().contains("Outbound workspace version") { + published = true; + break; + } + std::thread::sleep(Duration::from_millis(10)); + } + resume_tx.send(()).unwrap(); + writer.join().unwrap(); + assert!(!applying.join().unwrap()); + assert!(published, "filesystem writes should proceed while the DB writer is held"); + }); + assert_eq!(fixture.db_request().name, "Concurrent database edit"); + assert_eq!(fixture.states(), before); +} + +#[test] +fn current_outbound_only_plan_still_syncs_automatically() { + let fixture = Fixture::new("request.yaml"); + fixture + .db + .with_tx::<_, Error>(|tx| { + let mut request = tx.get_http_request(&fixture.request.id)?; + request.name = "New database version".into(); + tx.upsert_http_request(&request, &UpdateSource::from_window_label("test"))?; + Ok(()) + }) + .unwrap(); + let ops = fixture.calculate().unwrap(); + assert!(matches!(ops.as_slice(), [SyncOp::FsUpdate { .. }])); + assert!(fixture.apply_reviewed(ops)); + assert!(fs::read_to_string(&fixture.request_path).unwrap().contains("New database version")); + assert!(fixture.calculate().unwrap().is_empty()); +} + +#[test] +fn reviewed_import_and_file_deletion_apply_together() { + let fixture = Fixture::new("request.yaml"); + let mut added = fixture.request.clone(); + added.id = "rq_sync_added".into(); + let path = fixture.dir.path().join("added.yaml"); + let (contents, _) = SyncModel::HttpRequest(added.clone()).to_file_contents(&path).unwrap(); + fs::write(path, contents).unwrap(); + fixture + .db + .with_tx::<_, Error>(|tx| { + let request = tx.get_http_request(&fixture.request.id)?; + tx.delete_http_request(&request, &UpdateSource::Sync)?; + Ok(()) + }) + .unwrap(); + let ops = fixture.calculate().unwrap(); + assert!(ops.iter().any(|op| matches!(op, SyncOp::DbCreate { .. }))); + assert!(ops.iter().any(|op| matches!(op, SyncOp::FsDelete { fs: Some(_), .. }))); + + assert!(fixture.apply_reviewed(ops)); + assert!(!fixture.request_path.exists()); + assert_eq!(fixture.db.connect().get_http_request(&added.id).unwrap().id, added.id); + assert!(fixture.calculate().unwrap().is_empty()); +} + +#[test] +fn reviewed_inbound_and_outbound_updates_apply_together() { + let fixture = Fixture::new("request.yaml"); + fixture.edit(|value| value["name"] = "Incoming request".into()); + fixture + .db + .with_tx::<_, Error>(|tx| { + let mut workspace = tx.get_workspace(&fixture.workspace_id)?; + workspace.name = "Outgoing workspace".into(); + tx.upsert_workspace(&workspace, &UpdateSource::from_window_label("test"))?; + Ok(()) + }) + .unwrap(); + let ops = fixture.calculate().unwrap(); + assert!(ops.iter().any(|op| matches!(op, SyncOp::DbUpdate { .. }))); + assert!(ops.iter().any(|op| matches!(op, SyncOp::FsUpdate { .. }))); + + assert!(fixture.apply_reviewed(ops)); + assert_eq!(fixture.db_request().name, "Incoming request"); + assert!( + fs::read_to_string(fixture.dir.path().join("workspace.yaml")) + .unwrap() + .contains("Outgoing workspace") + ); + assert!(fixture.calculate().unwrap().is_empty()); +} + +#[test] +fn later_file_edits_invalidate_reviewed_outbound_operations() { + for creating in [false, true] { + for file_present in [true, false] { + let fixture = Fixture::new("yaak.rq_sync_test.yaml"); + let state = fixture + .states() + .into_iter() + .find(|state| state.model_id == fixture.request.id) + .unwrap(); + fixture + .db + .with_tx::<_, Error>(|tx| { + if creating { + tx.delete_sync_state(&state)?; + } + let mut request = tx.get_http_request(&fixture.request.id)?; + request.name = "Reviewed outgoing request".into(); + tx.upsert_http_request(&request, &UpdateSource::from_window_label("test"))?; + Ok(()) + }) + .unwrap(); + let workspace_path = fixture.dir.path().join("workspace.yaml"); + let workspace = fixture.db.connect().get_workspace(&fixture.workspace_id).unwrap(); + let mut incoming = workspace.clone(); + incoming.name = "Reviewed incoming workspace".into(); + let (content, _) = + SyncModel::Workspace(incoming).to_file_contents(&workspace_path).unwrap(); + fs::write(&workspace_path, &content).unwrap(); + if !file_present { + fs::remove_file(&fixture.request_path).unwrap(); + } + let ops = fixture.calculate().unwrap(); + assert!(ops.iter().any(|op| matches!(op, SyncOp::DbUpdate { .. }))); + assert!(ops.iter().any(|op| if creating { + matches!(op, SyncOp::FsCreate { .. }) + } else { + matches!(op, SyncOp::FsUpdate { .. }) + })); + let request = fixture.db_request(); + let states = fixture.states(); + + // Keep updatedAt older than the DB version, so conflict resolution still writes to FS. + fixture.edit(|value| value["name"] = "Later external file edit".into()); + let edited = fs::read(&fixture.request_path).unwrap(); + + assert!( + !fixture.apply_reviewed(ops), + "creating={creating}, file_present={file_present}" + ); + assert_eq!(fs::read(&fixture.request_path).unwrap(), edited); + assert_eq!(fs::read(&workspace_path).unwrap(), content); + assert_eq!(fixture.db_request(), request); + assert_eq!( + fixture.db.connect().get_workspace(&fixture.workspace_id).unwrap(), + workspace + ); + assert_eq!(fixture.states(), states); + assert!(fixture.apply_reviewed(fixture.calculate().unwrap())); + assert!(fixture.calculate().unwrap().is_empty()); + } + } +}