verstak/verstak-desktop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Harden core snapshot sync and workspace lifecycle

Browse Source
This commit is contained in:
mirivlad
2026-07-17 04:10:59 +08:00
parent ba0ba5f8c4
commit e3d8078ad5
22 changed files with 2891 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/app.go
+487 -52
View File
@@ -3,8 +3,11 @@ package api
import (
"context"
"crypto/sha256"
"encoding/base64"
"encoding/json"
"fmt"
"io"
"log"
"net/url"
"os"
@@ -95,6 +98,9 @@ type App struct {
browserReceiver *browserreceiver.Receiver
secretsSession *coresecrets.VaultSession
fileWatcher *filewatcher.Service
syncRunMu sync.Mutex
syncTimerMu sync.Mutex
syncScanTimer *time.Timer
notifications notificationService
debug bool
activityEvents map[string]bool
@@ -1228,6 +1234,7 @@ func (a *App) CloseVault() error {
if a.fileWatcher != nil {
a.fileWatcher.Stop()
}
a.stopScheduledSnapshotScan()
a.vault.CloseVault()
a.syncSvc = nil
a.secretsSession = nil
@@ -1738,10 +1745,30 @@ func (a *App) externalOpenService() externalOpenService {
}
func (a *App) recordFileSyncOp(entityType, entityID, opType string, payload interface{}) error {
if a.syncSvc == nil {
_, err := a.scanLocalChanges()
return err
}
func (a *App) recordWorkspaceSyncOp(opType, workspaceID, path, previousPath, name string) error {
if a.syncSvc == nil || a.workspace == nil {
return nil
}
return a.syncSvc.RecordOp(entityType, entityID, opType, payload)
meta, err := a.workspace.GetWorkspaceMetadata(path)
if err != nil && opType != syncsvc.OpTrash {
return err
}
payload := syncWorkspacePayload{
WorkspaceID: workspaceID,
Path: path,
PreviousPath: previousPath,
Name: name,
Metadata: meta,
}
if err := a.syncSvc.RecordOp(syncsvc.EntityWorkspace, workspaceID, opType, payload); err != nil {
return err
}
_, err = a.syncSvc.RebaseSnapshot()
return err
}
func (a *App) publishFileActivity(eventName, pluginID, relativePath string, extra map[string]interface{}) {
@@ -2314,9 +2341,69 @@ func (a *App) startFileWatcherForOpenVault() {
if a.fileWatcher == nil {
a.fileWatcher = filewatcher.NewService(a.eventBus, 0)
}
a.fileWatcher.SetOnChange(a.scheduleSnapshotScan)
if err := a.fileWatcher.Start(a.vault.GetVaultPath()); err != nil {
log.Printf("[api] file watcher start failed: %v", err)
}
if _, err := a.scanLocalChanges(); err != nil {
log.Printf("[api] initial sync snapshot scan failed: %v", err)
}
}
const snapshotScanDebounce = 300 * time.Millisecond
func (a *App) scheduleSnapshotScan() {
if a == nil {
return
}
a.syncTimerMu.Lock()
if a.syncScanTimer != nil {
a.syncScanTimer.Stop()
}
a.syncScanTimer = time.AfterFunc(snapshotScanDebounce, func() {
if _, err := a.scanLocalChanges(); err != nil {
log.Printf("[api] watcher sync snapshot scan failed: %v", err)
}
})
a.syncTimerMu.Unlock()
}
func (a *App) stopScheduledSnapshotScan() {
if a == nil {
return
}
a.syncTimerMu.Lock()
if a.syncScanTimer != nil {
a.syncScanTimer.Stop()
a.syncScanTimer = nil
}
a.syncTimerMu.Unlock()
}
func (a *App) scanLocalChanges() ([]string, error) {
if a == nil {
return nil, nil
}
a.syncRunMu.Lock()
defer a.syncRunMu.Unlock()
return a.scanLocalChangesLocked()
}
func (a *App) scanLocalChangesLocked() ([]string, error) {
if a.syncSvc == nil {
return nil, nil
}
warnings, err := a.syncSvc.ScanAndRecord()
if err != nil {
if a.appSettings != nil {
_ = a.updateSyncError("snapshot scan: " + err.Error())
}
return nil, err
}
if err := a.syncSvc.SetLastWarning(strings.Join(warnings, "\n")); err != nil {
return nil, err
}
return warnings, nil
}
// ─── Workspace API ─────────────────────────────────────────
@@ -2373,6 +2460,9 @@ func (a *App) CreateWorkspace(name, templateID string) (workspace.Workspace, str
if err != nil {
return workspace.Workspace{}, err.Error()
}
if err := a.recordWorkspaceSyncOp(syncsvc.OpCreate, ws.ID, ws.RootPath, "", ws.Name); err != nil {
return workspace.Workspace{}, err.Error()
}
a.publishWorkspaceLifecycleEvent(workspaceCreatedEventName, map[string]interface{}{
"operation": "create",
"workspaceId": ws.ID,
@@ -2395,6 +2485,9 @@ func (a *App) RenameWorkspace(oldName, newName string) string {
if err != nil {
return err.Error()
}
if err := a.recordWorkspaceSyncOp(syncsvc.OpRename, identity.WorkspaceID, newName, oldName, newName); err != nil {
return err.Error()
}
a.publishWorkspaceLifecycleEvent(workspaceRenamedEventName, map[string]interface{}{
"operation": "rename",
"workspaceId": identity.WorkspaceID,
@@ -2415,6 +2508,9 @@ func (a *App) TrashWorkspace(name string) (workspace.TrashResult, string) {
if err != nil {
return workspace.TrashResult{}, err.Error()
}
if err := a.recordWorkspaceSyncOp(syncsvc.OpTrash, result.WorkspaceID, name, "", name); err != nil {
return workspace.TrashResult{}, err.Error()
}
a.publishWorkspaceLifecycleEvent(workspaceTrashedEventName, map[string]interface{}{
"operation": "trash",
"workspaceId": result.WorkspaceID,
@@ -2436,6 +2532,9 @@ func (a *App) RestoreWorkspaceTrash(trashID, targetName string) (workspace.Works
if err != nil {
return workspace.Workspace{}, err.Error()
}
if err := a.recordWorkspaceSyncOp(syncsvc.OpRestore, restored.ID, restored.RootPath, "", restored.Name); err != nil {
return workspace.Workspace{}, err.Error()
}
a.publishWorkspaceLifecycleEvent(workspaceRestoredEventName, map[string]interface{}{
"operation": "restore",
"workspaceId": restored.ID,
@@ -2943,6 +3042,7 @@ func (a *App) vaultPath() string {
type SyncStatusDTO struct {
Configured bool `json:"configured"`
ServerURL string `json:"serverUrl"`
VaultID string `json:"vaultId"`
DeviceID string `json:"deviceId"`
DeviceName string `json:"deviceName"`
Connected bool `json:"connected"`
@@ -2952,6 +3052,7 @@ type SyncStatusDTO struct {
LastSyncAt string `json:"lastSyncAt"`
SyncInterval int `json:"syncInterval"`
LastError string `json:"lastError"`
LastWarning string `json:"lastWarning"`
StatusLabel string `json:"statusLabel"`
}
@@ -2972,15 +3073,19 @@ func (a *App) syncStatus() (*SyncStatusDTO, error) {
cfg := a.appSettings.Get()
deviceToken := syncsvc.LoadDeviceToken(vaultPath)
remoteVaultID, _ := a.syncSvc.RemoteVaultID()
lastWarning, _ := a.syncSvc.LastWarning()
dto := &SyncStatusDTO{
Configured: serverURL != "" && (apiKey != "" || deviceToken != ""),
ServerURL: serverURL,
VaultID: remoteVaultID,
LastSyncAt: lastSyncAt,
UnpushedOps: 0,
TokenStored: deviceToken != "",
SyncInterval: cfg.Sync.SyncInterval,
LastError: cfg.Sync.LastError,
LastWarning: lastWarning,
}
if deviceID := a.syncSvc.GetDeviceID(); deviceID != "" {
@@ -3048,7 +3153,7 @@ func (a *App) PluginSyncStatus(pluginID string) (*SyncStatusDTO, string) {
return dto, ""
}
func (a *App) syncConfigure(serverURL, username, password string) error {
func (a *App) syncConfigure(serverURL, username, password, remoteVaultID string) error {
if err := a.requireVault(); err != nil {
return err
}
@@ -3061,8 +3166,28 @@ func (a *App) syncConfigure(serverURL, username, password string) error {
if hostname == "" {
hostname = "unknown"
}
targetVaultID := strings.TrimSpace(remoteVaultID)
if targetVaultID == "" {
targetVaultID = meta.VaultID
}
if a.syncSvc != nil {
previousServerURL, _, _, _, stateErr := a.syncSvc.GetState()
previousVaultID, _ := a.syncSvc.RemoteVaultID()
if previousVaultID == "" {
previousVaultID = meta.VaultID
}
if stateErr == nil && previousServerURL != "" && previousVaultID != targetVaultID {
pending, err := a.syncSvc.GetUnpushedOps()
if err != nil {
return fmt.Errorf("read pending operations before changing remote vault: %w", err)
}
if len(pending) > 0 {
return fmt.Errorf("cannot change remote vault with %d unpushed local operation(s); synchronize or resolve them first", len(pending))
}
}
}
client := newSyncClient(serverURL, "", "", vaultPath)
deviceID, deviceToken, err := client.PairDevice(serverURL, username, password, hostname, "verstak-desktop/v2", meta.VaultID)
deviceID, deviceToken, err := client.PairDevice(serverURL, username, password, hostname, "verstak-desktop/v2", targetVaultID)
if err != nil {
return fmt.Errorf("pair: %w", err)
}
@@ -3073,6 +3198,18 @@ func (a *App) syncConfigure(serverURL, username, password string) error {
if err := a.syncSvc.SetState(serverURL, ""); err != nil {
return err
}
if err := a.syncSvc.SetRemoteVaultID(targetVaultID); err != nil {
return err
}
if err := a.syncSvc.SetLastPullSeq(0); err != nil {
return err
}
if err := a.syncSvc.SetBootstrapComplete(false); err != nil {
return err
}
if err := a.syncSvc.SetLastWarning(""); err != nil {
return err
}
cfg := a.appSettings.Get()
cfg.Sync.Enabled = true
@@ -3087,11 +3224,11 @@ func (a *App) syncConfigure(serverURL, username, password string) error {
}
// PluginSyncConfigure pairs the current vault with a sync server for a plugin.
func (a *App) PluginSyncConfigure(pluginID, serverURL, username, password string) string {
func (a *App) PluginSyncConfigure(pluginID, serverURL, username, password, remoteVaultID string) string {
if err := a.requirePluginSyncAccess(pluginID, true); err != nil {
return err.Error()
}
if err := a.syncConfigure(serverURL, username, password); err != nil {
if err := a.syncConfigure(serverURL, username, password, remoteVaultID); err != nil {
return err.Error()
}
return ""
@@ -3130,7 +3267,13 @@ func (a *App) syncDisconnect() error {
if a.syncSvc == nil {
return nil
}
return a.syncSvc.SetState("", "")
if err := a.syncSvc.SetState("", ""); err != nil {
return err
}
if err := a.syncSvc.SetRemoteVaultID(""); err != nil {
return err
}
return a.syncSvc.SetLastWarning("")
}
// PluginSyncDisconnect disconnects sync for a plugin with sync permission.
@@ -3230,6 +3373,8 @@ func (a *App) PluginSyncResetKey(pluginID string) string {
}
func (a *App) syncNow() (map[string]interface{}, error) {
a.syncRunMu.Lock()
defer a.syncRunMu.Unlock()
if err := a.requireVault(); err != nil {
return nil, err
}
@@ -3240,6 +3385,9 @@ func (a *App) syncNow() (map[string]interface{}, error) {
if a.syncSvc == nil {
return nil, fmt.Errorf("sync service not initialized")
}
if _, err := a.scanLocalChangesLocked(); err != nil {
return nil, fmt.Errorf("snapshot scan: %w", err)
}
serverURL, apiKey, lastPullSeq, _, err := a.syncSvc.GetState()
deviceToken := syncsvc.LoadDeviceToken(vaultPath)
@@ -3270,12 +3418,44 @@ func (a *App) syncNow() (map[string]interface{}, error) {
client.DeviceID = deviceID
}
bootstrapComplete, err := a.syncSvc.BootstrapComplete()
if err != nil {
return nil, fmt.Errorf("read bootstrap state: %w", err)
}
initialSnapshot, err := a.syncSvc.LoadSnapshot()
if err != nil {
return nil, fmt.Errorf("load initial snapshot: %w", err)
}
// Pull before publishing. This makes the first reconciliation safe: remote
// state is applied or reported as a conflict before a pre-existing local
// vault can enqueue its bootstrap creates.
pulled, cursor, serverSequence, err := a.pullRemoteOps(client, lastPullSeq, !bootstrapComplete)
if err != nil {
return nil, err
}
if pulled > 0 {
if warnings, err := a.syncSvc.RebaseSnapshot(); err != nil {
return nil, fmt.Errorf("rebase remote snapshot: %w", err)
} else if err := a.syncSvc.SetLastWarning(strings.Join(warnings, "\n")); err != nil {
return nil, fmt.Errorf("save sync warning: %w", err)
}
}
if !bootstrapComplete {
if err := a.syncSvc.RecordBootstrapOps(initialSnapshot); err != nil {
return nil, fmt.Errorf("record initial local snapshot: %w", err)
}
if err := a.syncSvc.SetBootstrapComplete(true); err != nil {
return nil, fmt.Errorf("save bootstrap state: %w", err)
}
}
unpushed, err := a.syncSvc.GetUnpushedOps()
if err != nil {
return nil, fmt.Errorf("get ops: %w", err)
}
for i := range unpushed {
unpushed[i].LastSeenServerSeq = lastPullSeq
unpushed[i].LastSeenServerSeq = cursor
}
pushResult := &syncsvc.PushResponse{}
if len(unpushed) > 0 {
@@ -3289,50 +3469,35 @@ func (a *App) syncNow() (map[string]interface{}, error) {
}
}
pullResult, err := client.Pull(lastPullSeq)
// Pull once more so this device durably acknowledges its own accepted
// operations and any concurrent remote operations without reapplying its own.
pulledAfterPush, _, finalServerSequence, err := a.pullRemoteOps(client, cursor, false)
if err != nil {
_ = a.updateSyncError(fmt.Sprintf("pull: %v", err))
return nil, fmt.Errorf("pull: %w", err)
return nil, err
}
var applyErrors []string
for _, op := range pullResult.Ops {
if err := a.applyRemoteOp(op); err != nil {
applyErrors = append(applyErrors, fmt.Sprintf("%s/%s: %v", op.EntityType, op.OpID, err))
if pulledAfterPush > 0 {
if warnings, err := a.syncSvc.RebaseSnapshot(); err != nil {
return nil, fmt.Errorf("rebase final remote snapshot: %w", err)
} else if err := a.syncSvc.SetLastWarning(strings.Join(warnings, "\n")); err != nil {
return nil, fmt.Errorf("save sync warning: %w", err)
}
_ = a.syncSvc.RecordRemoteOp(op)
}
if len(pullResult.Ops) > 0 {
opIDs := make([]string, len(pullResult.Ops))
for i, op := range pullResult.Ops {
opIDs[i] = op.OpID
}
_ = a.syncSvc.MarkApplied(opIDs)
if finalServerSequence > serverSequence {
serverSequence = finalServerSequence
}
if len(pushResult.Conflicts) > 0 {
log.Printf("[sync] %d conflict(s) detected on push", len(pushResult.Conflicts))
for _, c := range pushResult.Conflicts {
log.Printf("[sync] conflict: op=%v entity=%v/%v",
c["op_id"], c["entity_type"], c["entity_id"])
}
}
if pullResult.ServerSequence > lastPullSeq {
_ = a.syncSvc.SetLastPullSeq(pullResult.ServerSequence)
}
_ = a.syncSvc.SetLastSyncAt(time.Now().UTC().Format(time.RFC3339))
now := time.Now().UTC().Format(time.RFC3339)
a.updateSyncSuccess(now)
if err := a.syncSvc.SetLastSyncAt(now); err != nil {
return nil, fmt.Errorf("save sync time: %w", err)
}
_ = a.updateSyncSuccess(now)
result := map[string]interface{}{
"pushed": len(pushResult.Accepted),
"pulled": len(pullResult.Ops),
"serverSequence": pullResult.ServerSequence,
}
if len(applyErrors) > 0 {
result["applyErrors"] = applyErrors
"pulled": pulled + pulledAfterPush,
"serverSequence": serverSequence,
}
if len(pushResult.Conflicts) > 0 {
result["conflicts"] = pushResult.Conflicts
@@ -3340,6 +3505,74 @@ func (a *App) syncNow() (map[string]interface{}, error) {
return result, nil
}
func (a *App) pullRemoteOps(client *syncsvc.Client, cursor int, initialReconciliation bool) (pulled, nextCursor, serverSequence int, err error) {
pullResult, err := client.Pull(cursor)
if err != nil {
_ = a.updateSyncError(fmt.Sprintf("pull: %v", err))
return 0, cursor, cursor, fmt.Errorf("pull: %w", err)
}
nextCursor = cursor
lastSequenceInBatch := cursor
for _, op := range pullResult.Ops {
if op.ServerSequence <= cursor {
continue
}
if op.ServerSequence <= lastSequenceInBatch {
errMsg := fmt.Sprintf("pull response is not strictly ordered at sequence %d (%s)", op.ServerSequence, op.OpID)
_ = a.updateSyncError(errMsg)
return pulled, nextCursor, pullResult.ServerSequence, fmt.Errorf("%s", errMsg)
}
lastSequenceInBatch = op.ServerSequence
if err := a.applyRemoteOpForReconciliation(op, initialReconciliation); err != nil {
path := syncOperationPath(op)
errMsg := fmt.Sprintf("pull apply failed at sequence %d for %s %s (%s): %v", op.ServerSequence, op.EntityType, path, op.OpID, err)
_ = a.updateSyncError(errMsg)
return pulled, nextCursor, pullResult.ServerSequence, fmt.Errorf("%s", errMsg)
}
if err := a.syncSvc.RecordRemoteOp(op); err != nil {
errMsg := fmt.Sprintf("record applied remote operation at sequence %d (%s): %v", op.ServerSequence, op.OpID, err)
_ = a.updateSyncError(errMsg)
return pulled, nextCursor, pullResult.ServerSequence, fmt.Errorf("%s", errMsg)
}
if err := a.syncSvc.SetLastPullSeq(op.ServerSequence); err != nil {
errMsg := fmt.Sprintf("save pull cursor at sequence %d (%s): %v", op.ServerSequence, op.OpID, err)
_ = a.updateSyncError(errMsg)
return pulled, nextCursor, pullResult.ServerSequence, fmt.Errorf("%s", errMsg)
}
if warnings, err := a.syncSvc.RebaseSnapshot(); err != nil {
errMsg := fmt.Sprintf("rebase snapshot after sequence %d (%s): %v", op.ServerSequence, op.OpID, err)
_ = a.updateSyncError(errMsg)
return pulled, nextCursor, pullResult.ServerSequence, fmt.Errorf("%s", errMsg)
} else if err := a.syncSvc.SetLastWarning(strings.Join(warnings, "\n")); err != nil {
return pulled, nextCursor, pullResult.ServerSequence, fmt.Errorf("save sync warning: %w", err)
}
nextCursor = op.ServerSequence
pulled++
}
if pullResult.ServerSequence > nextCursor {
if err := a.syncSvc.SetLastPullSeq(pullResult.ServerSequence); err != nil {
_ = a.updateSyncError(fmt.Sprintf("save pull cursor at sequence %d: %v", pullResult.ServerSequence, err))
return pulled, nextCursor, pullResult.ServerSequence, fmt.Errorf("save pull cursor: %w", err)
}
nextCursor = pullResult.ServerSequence
}
return pulled, nextCursor, pullResult.ServerSequence, nil
}
func syncOperationPath(op syncsvc.Op) string {
if op.EntityType == syncsvc.EntityWorkspace {
if payload, err := parseSyncWorkspacePayload(op.PayloadJSON); err == nil && payload.Path != "" {
return payload.Path
}
return op.EntityID
}
payload, _ := parseSyncFilePayload(op.PayloadJSON)
if path := syncPayloadPath(op, payload); path != "" {
return path
}
return op.EntityID
}
// PluginSyncNow triggers sync for a plugin with sync permission.
func (a *App) PluginSyncNow(pluginID string) (map[string]interface{}, string) {
if err := a.requirePluginSyncAccess(pluginID, true); err != nil {
@@ -3368,12 +3601,23 @@ func (a *App) updateSyncSuccess(lastSyncAt string) error {
}
func (a *App) applyRemoteOp(op syncsvc.Op) error {
return a.applyRemoteOpForReconciliation(op, false)
}
func (a *App) applyRemoteOpForReconciliation(op syncsvc.Op, initialReconciliation bool) error {
if a.debug {
log.Printf("[sync] applyRemoteOp: type=%s entity=%s/%s", op.OpType, op.EntityType, op.EntityID)
}
if op.DeviceID != "" && op.DeviceID == a.localSyncDeviceID() {
return nil
}
if op.EntityType == syncsvc.EntityWorkspace {
payload, err := parseSyncWorkspacePayload(op.PayloadJSON)
if err != nil {
return err
}
return a.applyRemoteWorkspaceOp(op, payload)
}
if a.files == nil {
return fmt.Errorf("files service not initialized")
}
@@ -3384,20 +3628,29 @@ func (a *App) applyRemoteOp(op syncsvc.Op) error {
}
switch op.EntityType {
case syncsvc.EntityFile:
return a.applyRemoteFileOp(op, payload)
return a.applyRemoteFileOp(op, payload, initialReconciliation)
case syncsvc.EntityFolder:
return a.applyRemoteFolderOp(op, payload)
return a.applyRemoteFolderOp(op, payload, initialReconciliation)
default:
return fmt.Errorf("unsupported sync entity type: %s", op.EntityType)
}
}
type syncFilePayload struct {
Path string `json:"path"`
Content string `json:"content"`
DataBase64 *string `json:"dataBase64"`
FromPath string `json:"fromPath"`
ToPath string `json:"toPath"`
Path string `json:"path"`
Content string `json:"content"`
DataBase64 *string `json:"dataBase64"`
ContentHash string `json:"contentHash"`
FromPath string `json:"fromPath"`
ToPath string `json:"toPath"`
}
type syncWorkspacePayload struct {
WorkspaceID string `json:"workspaceId"`
Path string `json:"path"`
PreviousPath string `json:"previousPath,omitempty"`
Name string `json:"name"`
Metadata workspace.Metadata `json:"metadata"`
}
func parseSyncFilePayload(payloadJSON string) (syncFilePayload, error) {
@@ -3411,13 +3664,64 @@ func parseSyncFilePayload(payloadJSON string) (syncFilePayload, error) {
return payload, nil
}
func (a *App) applyRemoteFileOp(op syncsvc.Op, payload syncFilePayload) error {
func parseSyncWorkspacePayload(payloadJSON string) (syncWorkspacePayload, error) {
if payloadJSON == "" {
return syncWorkspacePayload{}, fmt.Errorf("workspace sync payload is empty")
}
var payload syncWorkspacePayload
if err := json.Unmarshal([]byte(payloadJSON), &payload); err != nil {
return syncWorkspacePayload{}, fmt.Errorf("invalid workspace sync payload: %w", err)
}
if payload.WorkspaceID == "" {
return syncWorkspacePayload{}, fmt.Errorf("workspace sync payload is missing workspaceId")
}
if payload.Path == "" {
payload.Path = payload.Name
}
return payload, nil
}
func (a *App) applyRemoteWorkspaceOp(op syncsvc.Op, payload syncWorkspacePayload) error {
if a.workspace == nil {
return fmt.Errorf("workspace service not initialized")
}
if payload.WorkspaceID != op.EntityID {
return fmt.Errorf("workspace identity mismatch: entity %s payload %s", op.EntityID, payload.WorkspaceID)
}
switch op.OpType {
case syncsvc.OpCreate:
_, err := a.workspace.CreateWorkspaceFromSync(payload.Path, payload.WorkspaceID, payload.Metadata)
return err
case syncsvc.OpRename:
return a.workspace.RenameWorkspaceFromSync(payload.WorkspaceID, payload.PreviousPath, payload.Path)
case syncsvc.OpTrash:
_, err := a.workspace.TrashWorkspaceFromSync(payload.WorkspaceID, payload.Path)
return err
case syncsvc.OpRestore:
_, err := a.workspace.RestoreWorkspaceFromSync(payload.WorkspaceID, payload.Path)
return err
default:
return fmt.Errorf("unsupported workspace sync op type: %s", op.OpType)
}
}
func (a *App) applyRemoteFileOp(op syncsvc.Op, payload syncFilePayload, initialReconciliation bool) error {
switch op.OpType {
case syncsvc.OpCreate:
path := syncPayloadPath(op, payload)
if path == "" {
return fmt.Errorf("missing file path")
}
matches, exists, err := a.remoteFileMatches(path, payload)
if err != nil {
return err
}
if matches {
return nil
}
if exists {
return fmt.Errorf("conflict: remote create would replace local file %s", path)
}
if payload.DataBase64 != nil {
return a.files.WriteVaultFileBytes(path, *payload.DataBase64, corefiles.WriteOptions{CreateIfMissing: true})
}
@@ -3427,15 +3731,44 @@ func (a *App) applyRemoteFileOp(op syncsvc.Op, payload syncFilePayload) error {
if path == "" {
return fmt.Errorf("missing file path")
}
if payload.DataBase64 != nil {
return a.files.WriteVaultFileBytes(path, *payload.DataBase64, corefiles.WriteOptions{CreateIfMissing: true, Overwrite: true})
matches, exists, err := a.remoteFileMatches(path, payload)
if err != nil {
return err
}
return a.files.WriteVaultTextFile(path, payload.Content, corefiles.WriteOptions{CreateIfMissing: true, Overwrite: true})
if matches {
return nil
}
if initialReconciliation && exists {
return fmt.Errorf("conflict: initial reconciliation would replace local file %s", path)
}
if pending, err := a.syncSvc.HasUnpushedPath(path); err != nil {
return err
} else if pending {
return fmt.Errorf("conflict: remote update would replace unpushed local file %s", path)
}
if payload.DataBase64 != nil {
return a.files.WriteVaultFileBytes(path, *payload.DataBase64, corefiles.WriteOptions{CreateIfMissing: !exists, Overwrite: exists})
}
return a.files.WriteVaultTextFile(path, payload.Content, corefiles.WriteOptions{CreateIfMissing: !exists, Overwrite: exists})
case syncsvc.OpDelete:
path := syncPayloadPath(op, payload)
if path == "" {
return fmt.Errorf("missing file path")
}
if initialReconciliation {
exists, err := a.syncPathExists(path)
if err != nil {
return err
}
if exists {
return fmt.Errorf("conflict: initial reconciliation would delete local file %s", path)
}
}
if pending, err := a.syncSvc.HasUnpushedPath(path); err != nil {
return err
} else if pending {
return fmt.Errorf("conflict: remote delete would remove unpushed local file %s", path)
}
_, err := a.files.TrashVaultPath(path)
if isSyncNotFound(err) {
return nil
@@ -3449,6 +3782,24 @@ func (a *App) applyRemoteFileOp(op syncsvc.Op, payload syncFilePayload) error {
if fromPath == "" || payload.ToPath == "" {
return fmt.Errorf("missing file move path")
}
if initialReconciliation {
fromExists, err := a.syncPathExists(fromPath)
if err != nil {
return err
}
toExists, err := a.syncPathExists(payload.ToPath)
if err != nil {
return err
}
if fromExists || toExists {
return fmt.Errorf("conflict: initial reconciliation would move local file %s", fromPath)
}
}
if pending, err := a.syncSvc.HasUnpushedPath(fromPath); err != nil {
return err
} else if pending {
return fmt.Errorf("conflict: remote move would replace unpushed local file %s", fromPath)
}
err := a.files.MoveVaultPath(fromPath, payload.ToPath, corefiles.MoveOptions{})
if isSyncNotFound(err) {
return nil
@@ -3459,7 +3810,54 @@ func (a *App) applyRemoteFileOp(op syncsvc.Op, payload syncFilePayload) error {
}
}
func (a *App) applyRemoteFolderOp(op syncsvc.Op, payload syncFilePayload) error {
func (a *App) remoteFileMatches(path string, payload syncFilePayload) (matches, exists bool, err error) {
normalized, err := corefiles.NormalizeRelativeFile(path)
if err != nil {
return false, false, err
}
info, err := os.Lstat(filepath.Join(a.vaultPath(), filepath.FromSlash(normalized)))
if err != nil {
if os.IsNotExist(err) {
return false, false, nil
}
return false, false, err
}
if info.Mode()&os.ModeSymlink != 0 || !info.Mode().IsRegular() {
return false, true, fmt.Errorf("conflict: remote file target is not a regular file: %s", normalized)
}
want, err := remotePayloadHash(payload)
if err != nil {
return false, true, err
}
file, err := os.Open(filepath.Join(a.vaultPath(), filepath.FromSlash(normalized)))
if err != nil {
return false, true, err
}
defer file.Close()
hash := sha256.New()
if _, err := io.Copy(hash, file); err != nil {
return false, true, err
}
return fmt.Sprintf("%x", hash.Sum(nil)) == want, true, nil
}
func remotePayloadHash(payload syncFilePayload) (string, error) {
if payload.ContentHash != "" {
return payload.ContentHash, nil
}
data := []byte(payload.Content)
if payload.DataBase64 != nil {
decoded, err := base64.StdEncoding.DecodeString(*payload.DataBase64)
if err != nil {
return "", fmt.Errorf("invalid remote base64 payload: %w", err)
}
data = decoded
}
sum := sha256.Sum256(data)
return fmt.Sprintf("%x", sum[:]), nil
}
func (a *App) applyRemoteFolderOp(op syncsvc.Op, payload syncFilePayload, initialReconciliation bool) error {
switch op.OpType {
case syncsvc.OpCreate:
path := syncPayloadPath(op, payload)
@@ -3476,6 +3874,15 @@ func (a *App) applyRemoteFolderOp(op syncsvc.Op, payload syncFilePayload) error
if path == "" {
return fmt.Errorf("missing folder path")
}
if initialReconciliation {
exists, err := a.syncPathExists(path)
if err != nil {
return err
}
if exists {
return fmt.Errorf("conflict: initial reconciliation would delete local folder %s", path)
}
}
_, err := a.files.TrashVaultPath(path)
if isSyncNotFound(err) {
return nil
@@ -3489,6 +3896,19 @@ func (a *App) applyRemoteFolderOp(op syncsvc.Op, payload syncFilePayload) error
if fromPath == "" || payload.ToPath == "" {
return fmt.Errorf("missing folder move path")
}
if initialReconciliation {
fromExists, err := a.syncPathExists(fromPath)
if err != nil {
return err
}
toExists, err := a.syncPathExists(payload.ToPath)
if err != nil {
return err
}
if fromExists || toExists {
return fmt.Errorf("conflict: initial reconciliation would move local folder %s", fromPath)
}
}
err := a.files.MoveVaultPath(fromPath, payload.ToPath, corefiles.MoveOptions{})
if isSyncNotFound(err) {
return nil
@@ -3499,6 +3919,21 @@ func (a *App) applyRemoteFolderOp(op syncsvc.Op, payload syncFilePayload) error
}
}
func (a *App) syncPathExists(path string) (bool, error) {
normalized, err := corefiles.NormalizeRelativeFile(path)
if err != nil {
return false, err
}
_, err = os.Lstat(filepath.Join(a.vaultPath(), filepath.FromSlash(normalized)))
if err == nil {
return true, nil
}
if os.IsNotExist(err) {
return false, nil
}
return false, err
}
func syncPayloadPath(op syncsvc.Op, payload syncFilePayload) string {
if payload.Path != "" {
return payload.Path
internal/api/app_test.go
+462 -6
View File
@@ -175,6 +175,12 @@ func newSyncFilesTestApp(t *testing.T, perms []string, deviceID string) (*App, s
t.Helper()
app, root := newFilesTestApp(t, perms)
app.syncSvc = syncsvc.NewService(root, deviceID)
if _, err := app.syncSvc.ScanAndRecord(); err != nil {
t.Fatalf("initial sync snapshot: %v", err)
}
if err := app.syncSvc.SetBootstrapComplete(true); err != nil {
t.Fatalf("mark test sync bootstrap complete: %v", err)
}
app.appSettings = appsettings.NewManager(filepath.Join(t.TempDir(), "config.json"))
if err := app.appSettings.Load(); err != nil {
t.Fatalf("settings Load: %v", err)
@@ -1692,6 +1698,63 @@ func TestApplyRemoteOpSkipsLocalDevice(t *testing.T) {
}
}
func TestApplyRemoteWorkspaceLifecyclePreservesDurableIdentity(t *testing.T) {
app, root := newSyncFilesTestApp(t, []string{"files.read", "files.write", "files.delete"}, "local-device")
app.workspace = workspace.NewManager(root)
if err := app.workspace.Load(); err != nil {
t.Fatal(err)
}
workspaceID := "5f0f96d9-61c8-4b6b-8c3a-a1b9a0f40002"
payload := syncWorkspacePayload{
WorkspaceID: workspaceID,
Path: "Remote",
Name: "Remote",
Metadata: workspace.Metadata{
WorkspaceID: workspaceID,
WorkspaceName: "Remote",
Folders: map[string]string{"notes": "Notes"},
Features: map[string]bool{"files": true},
},
}
encoded, err := json.Marshal(payload)
if err != nil {
t.Fatal(err)
}
create := syncsvc.Op{OpID: "workspace-create", DeviceID: "remote-device", EntityType: syncsvc.EntityWorkspace, EntityID: workspaceID, OpType: syncsvc.OpCreate, PayloadJSON: string(encoded)}
if err := app.applyRemoteOp(create); err != nil {
t.Fatalf("apply workspace create: %v", err)
}
if err := app.applyRemoteOp(create); err != nil {
t.Fatalf("replay workspace create: %v", err)
}
if identity, err := app.workspace.GetWorkspaceIdentity("Remote"); err != nil || identity.WorkspaceID != workspaceID {
t.Fatalf("remote workspace identity = %+v err=%v", identity, err)
}
payload.PreviousPath = "Remote"
payload.Path = "Remote-Renamed"
payload.Name = "Remote-Renamed"
encoded, _ = json.Marshal(payload)
rename := syncsvc.Op{OpID: "workspace-rename", DeviceID: "remote-device", EntityType: syncsvc.EntityWorkspace, EntityID: workspaceID, OpType: syncsvc.OpRename, PayloadJSON: string(encoded)}
if err := app.applyRemoteOp(rename); err != nil {
t.Fatalf("apply workspace rename: %v", err)
}
trash := syncsvc.Op{OpID: "workspace-trash", DeviceID: "remote-device", EntityType: syncsvc.EntityWorkspace, EntityID: workspaceID, OpType: syncsvc.OpTrash, PayloadJSON: string(encoded)}
if err := app.applyRemoteOp(trash); err != nil {
t.Fatalf("apply workspace trash: %v", err)
}
payload.Path = "Remote-Restored"
payload.Name = "Remote-Restored"
encoded, _ = json.Marshal(payload)
restore := syncsvc.Op{OpID: "workspace-restore", DeviceID: "remote-device", EntityType: syncsvc.EntityWorkspace, EntityID: workspaceID, OpType: syncsvc.OpRestore, PayloadJSON: string(encoded)}
if err := app.applyRemoteOp(restore); err != nil {
t.Fatalf("apply workspace restore: %v", err)
}
if identity, err := app.workspace.GetWorkspaceIdentity("Remote-Restored"); err != nil || identity.WorkspaceID != workspaceID {
t.Fatalf("restored workspace identity = %+v err=%v", identity, err)
}
}
func TestFileBridgeRecordsSyncOps(t *testing.T) {
app, _ := newSyncFilesTestApp(t, []string{"files.read", "files.write", "files.delete"}, "local-device")
@@ -1718,8 +1781,8 @@ func TestFileBridgeRecordsSyncOps(t *testing.T) {
if err != nil {
t.Fatalf("GetUnpushedOps: %v", err)
}
if len(ops) != 6 {
t.Fatalf("ops len = %d, want 6: %#v", len(ops), ops)
if len(ops) != 7 {
t.Fatalf("ops len = %d, want 7: %#v", len(ops), ops)
}
want := []struct {
@@ -1732,7 +1795,10 @@ func TestFileBridgeRecordsSyncOps(t *testing.T) {
{syncsvc.EntityFile, "Docs/one.txt", syncsvc.OpCreate, `"content":"hello"`},
{syncsvc.EntityFile, "Docs/one.txt", syncsvc.OpUpdate, `"content":"updated"`},
{syncsvc.EntityFile, "Docs/image.bin", syncsvc.OpCreate, `"dataBase64":"AQID"`},
{syncsvc.EntityFile, "Docs/one.txt", syncsvc.OpMove, `"toPath":"Docs/two.txt"`},
// Snapshot reconciliation represents external and API renames uniformly
// as a create at the destination followed by a delete at the source.
{syncsvc.EntityFile, "Docs/two.txt", syncsvc.OpCreate, `"content":"updated"`},
{syncsvc.EntityFile, "Docs/one.txt", syncsvc.OpDelete, `"path":"Docs/one.txt"`},
{syncsvc.EntityFile, "Docs/two.txt", syncsvc.OpDelete, `"path":"Docs/two.txt"`},
}
for i, w := range want {
@@ -1854,8 +1920,8 @@ func TestSyncNowPushesLocalOpsAndAppliesPulledFileOps(t *testing.T) {
t.Fatalf("pushed ops len = %d, want 2", len(pushedOps))
}
for i, op := range pushedOps {
if op.LastSeenServerSeq != 0 {
t.Fatalf("pushed op[%d] last seen = %d, want 0", i, op.LastSeenServerSeq)
if op.LastSeenServerSeq != 2 {
t.Fatalf("pushed op[%d] last seen = %d, want 2 after initial pull", i, op.LastSeenServerSeq)
}
}
@@ -1882,6 +1948,323 @@ func TestSyncNowPushesLocalOpsAndAppliesPulledFileOps(t *testing.T) {
}
}
func TestSyncNowStopsAtFailedRemoteOperationAndRetriesAfterRestart(t *testing.T) {
app, root := newSyncFilesTestApp(t, []string{"files.read", "files.write", "files.delete"}, "local-device")
pullCalls := 0
server := newLocalHTTPTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Authorization") != "Bearer device-token" {
http.Error(w, "missing auth", http.StatusUnauthorized)
return
}
w.Header().Set("Content-Type", "application/json")
switch r.URL.Path {
case "/api/v1/sync/pull":
pullCalls++
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"server_sequence": 3,
"ops": []map[string]interface{}{
{
"op_id": "remote-folder",
"server_sequence": 1,
"device_id": "remote-device",
"entity_type": syncsvc.EntityFolder,
"entity_id": "Remote",
"op_type": syncsvc.OpCreate,
"payload_json": `{"path":"Remote"}`,
},
{
"op_id": "blocked-file",
"server_sequence": 2,
"device_id": "remote-device",
"entity_type": syncsvc.EntityFile,
"entity_id": "Missing/blocked.txt",
"op_type": syncsvc.OpCreate,
"payload_json": `{"path":"Missing/blocked.txt","content":"blocked"}`,
},
{
"op_id": "after-failure",
"server_sequence": 3,
"device_id": "remote-device",
"entity_type": syncsvc.EntityFile,
"entity_id": "Remote/after.txt",
"op_type": syncsvc.OpCreate,
"payload_json": `{"path":"Remote/after.txt","content":"must wait"}`,
},
},
})
case "/api/v1/sync/push":
_ = json.NewEncoder(w).Encode(map[string]interface{}{"accepted": []string{}, "count": 0, "conflicts": []map[string]interface{}{}})
default:
http.NotFound(w, r)
}
}))
defer server.Close()
if err := app.syncSvc.SetState(server.URL, ""); err != nil {
t.Fatal(err)
}
if err := syncsvc.SaveDeviceToken(root, "device-token"); err != nil {
t.Fatal(err)
}
if _, err := app.syncNow(); err == nil || !strings.Contains(err.Error(), "sequence 2") {
t.Fatalf("first sync error = %v, want failed sequence", err)
}
assertLastPullSequence(t, app.syncSvc, 1)
if _, errStr := app.GetVaultFileMetadata("files.plugin", "Remote/after.txt"); !strings.Contains(errStr, "not-found") {
t.Fatalf("operation after failure applied early: %q", errStr)
}
cfg := app.appSettings.Get()
if cfg.Sync.LastError == "" || !strings.Contains(cfg.Sync.LastError, "sequence 2") || !strings.Contains(cfg.Sync.LastError, "Missing/blocked.txt") {
t.Fatalf("sync status after failed apply = %#v", cfg.Sync)
}
if err := os.Mkdir(filepath.Join(root, "Missing"), 0o755); err != nil {
t.Fatal(err)
}
restarted := &App{
vault: app.vault,
files: app.files,
plugins: app.plugins,
appSettings: app.appSettings,
syncSvc: syncsvc.NewService(root, ""),
}
if _, err := restarted.syncNow(); err != nil {
t.Fatalf("retry after restart: %v", err)
}
if pullCalls < 2 {
t.Fatalf("pull calls = %d, want retry", pullCalls)
}
assertLastPullSequence(t, restarted.syncSvc, 3)
expectText(t, restarted, "Missing/blocked.txt", "blocked")
expectText(t, restarted, "Remote/after.txt", "must wait")
}
func TestSyncNowBootstrapsExistingLocalSnapshotAfterInitialPull(t *testing.T) {
app, root := newFilesTestApp(t, []string{"files.read", "files.write", "files.delete"})
if err := os.Mkdir(filepath.Join(root, "Existing"), 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(root, "Existing", "before-connect.txt"), []byte("local before connect"), 0o644); err != nil {
t.Fatal(err)
}
app.syncSvc = syncsvc.NewService(root, "local-device")
if _, err := app.syncSvc.ScanAndRecord(); err != nil {
t.Fatalf("initial baseline scan: %v", err)
}
app.appSettings = appsettings.NewManager(filepath.Join(t.TempDir(), "config.json"))
if err := app.appSettings.Load(); err != nil {
t.Fatal(err)
}
var pushed []syncsvc.PushOp
server := newLocalHTTPTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Authorization") != "Bearer device-token" {
http.Error(w, "missing auth", http.StatusUnauthorized)
return
}
w.Header().Set("Content-Type", "application/json")
switch r.URL.Path {
case "/api/v1/sync/pull":
_ = json.NewEncoder(w).Encode(map[string]interface{}{"server_sequence": 0, "ops": []map[string]interface{}{}})
case "/api/v1/sync/push":
var request struct {
Ops []syncsvc.PushOp `json:"ops"`
}
if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
pushed = request.Ops
accepted := make([]string, 0, len(request.Ops))
for _, op := range request.Ops {
accepted = append(accepted, op.OpID)
}
_ = json.NewEncoder(w).Encode(map[string]interface{}{"accepted": accepted, "count": len(accepted), "conflicts": []map[string]interface{}{}})
default:
http.NotFound(w, r)
}
}))
defer server.Close()
if err := app.syncSvc.SetState(server.URL, ""); err != nil {
t.Fatal(err)
}
if err := syncsvc.SaveDeviceToken(root, "device-token"); err != nil {
t.Fatal(err)
}
result, err := app.syncNow()
if err != nil {
t.Fatalf("syncNow: %v", err)
}
if result["pushed"] != len(pushed) || len(pushed) < 2 {
t.Fatalf("result=%#v pushed=%#v, want initial entries", result, pushed)
}
foundFolder, foundFile := false, false
for _, op := range pushed {
if op.EntityID == "Existing" && op.EntityType == syncsvc.EntityFolder && op.OpType == syncsvc.OpCreate {
foundFolder = true
}
if op.EntityID == "Existing/before-connect.txt" && op.EntityType == syncsvc.EntityFile && op.OpType == syncsvc.OpCreate {
foundFile = true
}
if strings.Contains(op.EntityID, "/.verstak/") {
t.Fatalf("ordinary bootstrap operation leaked internal path: %+v", op)
}
}
if !foundFolder || !foundFile {
t.Fatalf("bootstrap push = %#v, missing existing file or folder", pushed)
}
bootstrapped, err := app.syncSvc.BootstrapComplete()
if err != nil || !bootstrapped {
t.Fatalf("bootstrap complete = %v err=%v", bootstrapped, err)
}
}
func TestInitialSyncConflictDoesNotOverwriteLocalFileOrAdvanceCursor(t *testing.T) {
app, root := newFilesTestApp(t, []string{"files.read", "files.write", "files.delete"})
if err := os.WriteFile(filepath.Join(root, "same-name.txt"), []byte("local value"), 0o644); err != nil {
t.Fatal(err)
}
app.syncSvc = syncsvc.NewService(root, "local-device")
if _, err := app.syncSvc.ScanAndRecord(); err != nil {
t.Fatalf("initial baseline scan: %v", err)
}
app.appSettings = appsettings.NewManager(filepath.Join(t.TempDir(), "config.json"))
if err := app.appSettings.Load(); err != nil {
t.Fatal(err)
}
pushCalls := 0
server := newLocalHTTPTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Authorization") != "Bearer device-token" {
http.Error(w, "missing auth", http.StatusUnauthorized)
return
}
w.Header().Set("Content-Type", "application/json")
switch r.URL.Path {
case "/api/v1/sync/pull":
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"server_sequence": 1,
"ops": []map[string]interface{}{{
"op_id": "remote-update",
"server_sequence": 1,
"device_id": "remote-device",
"entity_type": syncsvc.EntityFile,
"entity_id": "same-name.txt",
"op_type": syncsvc.OpUpdate,
"payload_json": `{"path":"same-name.txt","content":"remote value"}`,
}},
})
case "/api/v1/sync/push":
pushCalls++
_ = json.NewEncoder(w).Encode(map[string]interface{}{"accepted": []string{}, "count": 0, "conflicts": []map[string]interface{}{}})
default:
http.NotFound(w, r)
}
}))
defer server.Close()
if err := app.syncSvc.SetState(server.URL, ""); err != nil {
t.Fatal(err)
}
if err := syncsvc.SaveDeviceToken(root, "device-token"); err != nil {
t.Fatal(err)
}
if _, err := app.syncNow(); err == nil || !strings.Contains(err.Error(), "initial reconciliation") || !strings.Contains(err.Error(), "same-name.txt") {
t.Fatalf("sync error = %v, want explicit initial conflict", err)
}
expectText(t, app, "same-name.txt", "local value")
assertLastPullSequence(t, app.syncSvc, 0)
if pushCalls != 0 {
t.Fatalf("push calls = %d, conflict must not overwrite remote state", pushCalls)
}
bootstrapped, err := app.syncSvc.BootstrapComplete()
if err != nil || bootstrapped {
t.Fatalf("bootstrap state = %v err=%v, conflict must remain unresolved", bootstrapped, err)
}
}
func TestInitialSyncEmptyVaultRestoresRemoteDataWithoutDeletePush(t *testing.T) {
app, root := newFilesTestApp(t, []string{"files.read", "files.write", "files.delete"})
app.syncSvc = syncsvc.NewService(root, "empty-device")
if _, err := app.syncSvc.ScanAndRecord(); err != nil {
t.Fatalf("initial empty baseline: %v", err)
}
app.appSettings = appsettings.NewManager(filepath.Join(t.TempDir(), "config.json"))
if err := app.appSettings.Load(); err != nil {
t.Fatal(err)
}
var pushed []syncsvc.PushOp
server := newLocalHTTPTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Authorization") != "Bearer device-token" {
http.Error(w, "missing auth", http.StatusUnauthorized)
return
}
w.Header().Set("Content-Type", "application/json")
switch r.URL.Path {
case "/api/v1/sync/pull":
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"server_sequence": 1,
"ops": []map[string]interface{}{{
"op_id": "remote-file",
"server_sequence": 1,
"device_id": "remote-device",
"entity_type": syncsvc.EntityFile,
"entity_id": "restored.txt",
"op_type": syncsvc.OpCreate,
"payload_json": `{"path":"restored.txt","content":"from remote"}`,
}},
})
case "/api/v1/sync/push":
var request struct {
Ops []syncsvc.PushOp `json:"ops"`
}
if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
pushed = append(pushed, request.Ops...)
_ = json.NewEncoder(w).Encode(map[string]interface{}{"accepted": []string{}, "count": 0, "conflicts": []map[string]interface{}{}})
default:
http.NotFound(w, r)
}
}))
defer server.Close()
if err := app.syncSvc.SetState(server.URL, ""); err != nil {
t.Fatal(err)
}
if err := syncsvc.SaveDeviceToken(root, "device-token"); err != nil {
t.Fatal(err)
}
result, err := app.syncNow()
if err != nil {
t.Fatalf("sync empty vault: %v", err)
}
expectText(t, app, "restored.txt", "from remote")
for _, op := range pushed {
if op.OpType == syncsvc.OpDelete {
t.Fatalf("empty vault published delete operation: %+v", op)
}
}
if result["pushed"] != 0 && len(pushed) == 0 {
t.Fatalf("result reports pushed operations without a push payload: %#v", result)
}
assertLastPullSequence(t, app.syncSvc, 1)
}
func assertLastPullSequence(t *testing.T, service *syncsvc.Service, want int) {
t.Helper()
_, _, got, _, err := service.GetState()
if err != nil {
t.Fatalf("GetState: %v", err)
}
if got != want {
t.Fatalf("last pull sequence = %d, want %d", got, want)
}
}
func TestSyncConfigurePairsCurrentVaultID(t *testing.T) {
app, _ := newSyncFilesTestApp(t, []string{"files.read", "files.write", "files.delete"}, "local-device")
meta := app.vault.GetVaultMeta()
@@ -1911,7 +2294,7 @@ func TestSyncConfigurePairsCurrentVaultID(t *testing.T) {
}))
defer server.Close()
if err := app.syncConfigure(server.URL, "alice", "secret"); err != nil {
if err := app.syncConfigure(server.URL, "alice", "secret", ""); err != nil {
t.Fatalf("syncConfigure: %v", err)
}
if pairedVaultID != meta.VaultID {
@@ -1919,6 +2302,79 @@ func TestSyncConfigurePairsCurrentVaultID(t *testing.T) {
}
}
func TestSyncConfigurePairsSpecifiedRemoteVaultID(t *testing.T) {
app, _ := newSyncFilesTestApp(t, []string{"files.read", "files.write", "files.delete"}, "local-device")
const remoteVaultID = "existing-remote-vault"
var pairedVaultID string
server := newLocalHTTPTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/api/client/pair" {
http.NotFound(w, r)
return
}
var request struct {
VaultID string `json:"vault_id"`
}
if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
pairedVaultID = request.VaultID
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]string{
"device_id": "paired-device",
"device_token": "paired-token",
})
}))
defer server.Close()
if err := app.syncConfigure(server.URL, "alice", "secret", remoteVaultID); err != nil {
t.Fatalf("syncConfigure: %v", err)
}
if pairedVaultID != remoteVaultID {
t.Fatalf("paired vault ID = %q, want %q", pairedVaultID, remoteVaultID)
}
storedRemoteVaultID, err := app.syncSvc.RemoteVaultID()
if err != nil || storedRemoteVaultID != remoteVaultID {
t.Fatalf("stored remote vault ID = %q err=%v, want %q", storedRemoteVaultID, err, remoteVaultID)
}
bootstrapped, err := app.syncSvc.BootstrapComplete()
if err != nil || bootstrapped {
t.Fatalf("bootstrap state after new pairing = %v err=%v, want false", bootstrapped, err)
}
}
func TestSyncConfigureRefusesRemoteScopeChangeWithUnpushedOperations(t *testing.T) {
app, _ := newSyncFilesTestApp(t, []string{"files.read", "files.write", "files.delete"}, "local-device")
if err := app.syncSvc.SetState("https://old-sync.example.test", ""); err != nil {
t.Fatal(err)
}
if err := app.syncSvc.SetRemoteVaultID("old-remote-vault"); err != nil {
t.Fatal(err)
}
if err := app.syncSvc.RecordOp(syncsvc.EntityFile, "pending.txt", syncsvc.OpCreate, map[string]string{"path": "pending.txt", "content": "local"}); err != nil {
t.Fatal(err)
}
err := app.syncConfigure("https://new-sync.example.test", "alice", "secret", "new-remote-vault")
if err == nil || !strings.Contains(err.Error(), "unpushed local operation") {
t.Fatalf("scope-change error = %v, want pending-operation refusal", err)
}
}
func TestSyncStatusExposesPersistentScannerWarning(t *testing.T) {
app, _ := newSyncFilesTestApp(t, []string{"files.read", "files.write", "files.delete"}, "local-device")
if err := app.syncSvc.SetLastWarning("file-too-large: archive.bin"); err != nil {
t.Fatal(err)
}
status, err := app.syncStatus()
if err != nil {
t.Fatalf("syncStatus: %v", err)
}
if status.LastWarning != "file-too-large: archive.bin" {
t.Fatalf("last warning = %q", status.LastWarning)
}
}
func TestSyncNowHydratesLegacyVaultDeviceID(t *testing.T) {
app, root := newSyncFilesTestApp(t, []string{"files.read", "files.write", "files.delete"}, "wrong-global-device")
var pushedDeviceID string
internal/api/sync_real_server_test.go
+95 -6
View File
@@ -2,10 +2,12 @@ package api
import (
"os"
"path/filepath"
"strings"
"testing"
corefiles "github.com/verstak/verstak-desktop/internal/core/files"
"github.com/verstak/verstak-desktop/internal/core/workspace"
)
func TestSyncNowAgainstRealServerTwoVaults(t *testing.T) {
@@ -18,8 +20,16 @@ func TestSyncNowAgainstRealServerTwoVaults(t *testing.T) {
t.Skip("set VERSTAK_SYNC_SMOKE_* env vars to run the real sync-server smoke test")
}
appA, _ := newSyncFilesTestApp(t, []string{"files.read", "files.write", "files.delete"}, deviceA)
appB, _ := newSyncFilesTestApp(t, []string{"files.read", "files.write", "files.delete"}, deviceB)
appA, rootA := newSyncFilesTestApp(t, []string{"files.read", "files.write", "files.delete"}, deviceA)
appB, rootB := newSyncFilesTestApp(t, []string{"files.read", "files.write", "files.delete"}, deviceB)
appA.workspace = workspace.NewManager(rootA)
appB.workspace = workspace.NewManager(rootB)
if err := appA.workspace.Load(); err != nil {
t.Fatalf("load workspace A: %v", err)
}
if err := appB.workspace.Load(); err != nil {
t.Fatalf("load workspace B: %v", err)
}
if err := appA.syncSvc.SetState(serverURL, apiKeyA); err != nil {
t.Fatalf("appA SetState: %v", err)
}
@@ -43,8 +53,8 @@ func TestSyncNowAgainstRealServerTwoVaults(t *testing.T) {
if errStr := appB.MoveVaultPath("files.plugin", "Shared/one.txt", "Shared/two.txt", corefiles.MoveOptions{}); errStr != "" {
t.Fatalf("appB move: %s", errStr)
}
expectSyncCounts(t, appB, 2, 2)
expectSyncCounts(t, appA, 0, 2)
expectSyncCounts(t, appB, 3, 3)
expectSyncCounts(t, appA, 0, 3)
expectText(t, appA, "Shared/two.txt", "from B")
if _, errStr := appA.TrashVaultPath("files.plugin", "Shared/two.txt"); errStr != "" {
@@ -68,8 +78,8 @@ func TestSyncNowAgainstRealServerTwoVaults(t *testing.T) {
if errStr := appA.MoveVaultPath("files.plugin", "Shared/Folder", "Shared/Archive", corefiles.MoveOptions{}); errStr != "" {
t.Fatalf("appA move folder: %s", errStr)
}
expectSyncCounts(t, appA, 1, 1)
expectSyncCounts(t, appB, 0, 1)
expectSyncCounts(t, appA, 2, 2)
expectSyncCounts(t, appB, 0, 2)
if _, errStr := appB.GetVaultFileMetadata("files.plugin", "Shared/Folder"); !strings.Contains(errStr, "not-found") {
t.Fatalf("appB moved folder old metadata err = %q, want not-found", errStr)
}
@@ -85,6 +95,63 @@ func TestSyncNowAgainstRealServerTwoVaults(t *testing.T) {
if _, errStr := appA.GetVaultFileMetadata("files.plugin", "Shared/Archive"); !strings.Contains(errStr, "not-found") {
t.Fatalf("appA deleted folder metadata err = %q, want not-found", errStr)
}
if err := os.WriteFile(filepath.Join(rootA, "Shared", "external.txt"), []byte("external while running"), 0o644); err != nil {
t.Fatalf("external create: %v", err)
}
expectSyncCounts(t, appA, 1, 1)
expectSyncCounts(t, appB, 0, 1)
expectText(t, appB, "Shared/external.txt", "external while running")
assertNoUnpushedOps(t, appB)
if err := os.WriteFile(filepath.Join(rootB, "Shared", "external.txt"), []byte("external while closed"), 0o644); err != nil {
t.Fatalf("offline external update: %v", err)
}
if err := os.WriteFile(filepath.Join(rootB, "Shared", "offline-created.txt"), []byte("created while closed"), 0o644); err != nil {
t.Fatalf("offline external create: %v", err)
}
if err := os.Remove(filepath.Join(rootB, "Shared", "external.txt")); err != nil {
t.Fatalf("offline external delete: %v", err)
}
expectSyncCounts(t, appB, 2, 2)
expectSyncCounts(t, appA, 0, 2)
if _, errStr := appA.GetVaultFileMetadata("files.plugin", "Shared/external.txt"); !strings.Contains(errStr, "not-found") {
t.Fatalf("offline deleted file remained on appA: %q", errStr)
}
expectText(t, appA, "Shared/offline-created.txt", "created while closed")
assertNoUnpushedOps(t, appA)
deal, errStr := appA.CreateWorkspace("Synced Deal", "minimal")
if errStr != "" {
t.Fatalf("appA CreateWorkspace: %s", errStr)
}
expectSyncCounts(t, appA, 1, 1)
expectSyncCounts(t, appB, 0, 1)
assertWorkspaceIdentity(t, appB, "Synced Deal", deal.ID)
if errStr := appA.RenameWorkspace("Synced Deal", "Renamed Deal"); errStr != "" {
t.Fatalf("appA RenameWorkspace: %s", errStr)
}
expectSyncCounts(t, appA, 1, 1)
expectSyncCounts(t, appB, 0, 1)
assertWorkspaceIdentity(t, appB, "Renamed Deal", deal.ID)
trash, errStr := appA.TrashWorkspace("Renamed Deal")
if errStr != "" {
t.Fatalf("appA TrashWorkspace: %s", errStr)
}
expectSyncCounts(t, appA, 1, 1)
expectSyncCounts(t, appB, 0, 1)
if _, err := appB.workspace.GetWorkspaceIdentity("Renamed Deal"); err == nil {
t.Fatal("trashed workspace is still active on appB")
}
if _, errStr := appA.RestoreWorkspaceTrash(trash.TrashID, "Restored Deal"); errStr != "" {
t.Fatalf("appA RestoreWorkspaceTrash: %s", errStr)
}
expectSyncCounts(t, appA, 1, 1)
expectSyncCounts(t, appB, 0, 1)
assertWorkspaceIdentity(t, appB, "Restored Deal", deal.ID)
}
func expectSyncCounts(t *testing.T, app *App, pushed, pulled int) {
@@ -108,3 +175,25 @@ func expectText(t *testing.T, app *App, path, want string) {
t.Fatalf("ReadVaultTextFile(%s) = %q, want %q", path, text, want)
}
}
func assertNoUnpushedOps(t *testing.T, app *App) {
t.Helper()
ops, err := app.syncSvc.GetUnpushedOps()
if err != nil {
t.Fatalf("GetUnpushedOps: %v", err)
}
if len(ops) != 0 {
t.Fatalf("remote operations were echoed as local operations: %#v", ops)
}
}
func assertWorkspaceIdentity(t *testing.T, app *App, name, wantID string) {
t.Helper()
identity, err := app.workspace.GetWorkspaceIdentity(name)
if err != nil {
t.Fatalf("GetWorkspaceIdentity(%s): %v", name, err)
}
if identity.WorkspaceID != wantID {
t.Fatalf("workspace %s ID = %s, want %s", name, identity.WorkspaceID, wantID)
}
}