sirtimbly/cairnquire
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sync app filled in

Browse Source
This commit is contained in:
Tim Bendt
2026-06-01 10:53:28 -04:00
parent ebe0920f89
commit fb0673a1c5
36 changed files with 4527 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
apps/server/internal/sync/protocol.go
View File

@@ -46,6 +46,7 @@ type Change struct {
Path string `json:"path"`
OldPath string `json:"oldPath,omitempty"`
Hash string `json:"hash,omitempty"`
Content string `json:"content,omitempty"`
Size int64 `json:"size,omitempty"`
Modified time.Time `json:"modified,omitempty"`
}
@@ -67,8 +68,9 @@ type Conflict struct {
// DeltaResult is the server's response to a delta upload.
type DeltaResult struct {
ServerDelta []Change `json:"serverDelta"`
Conflicts []Conflict `json:"conflicts"`
ServerDelta []Change `json:"serverDelta"`
Conflicts []Conflict `json:"conflicts"`
NewSnapshotID string `json:"newSnapshotId,omitempty"`
}
// Resolution is a user's choice for resolving a conflict.
@@ -76,6 +78,8 @@ type Resolution struct {
Path string `json:"path"`
Strategy ResolutionStrategy `json:"strategy"`
NewPath string `json:"newPath,omitempty"` // for rename-both
Hash string `json:"hash,omitempty"`
Content string `json:"content,omitempty"`
}
// InitRequest starts a new sync session.
@@ -86,26 +90,27 @@ type InitRequest struct {
// InitResponse returns the server's current snapshot.
type InitResponse struct {
SnapshotID string `json:"snapshotId"`
ServerSnapshot []FileEntry `json:"serverSnapshot"`
SnapshotID string `json:"snapshotId"`
ServerSnapshot []FileEntry `json:"serverSnapshot"`
}
// DeltaRequest uploads client changes.
type DeltaRequest struct {
SnapshotID string `json:"snapshotId"`
SnapshotID string `json:"snapshotId"`
ClientDelta []Change `json:"clientDelta"`
}
// DeltaResponse returns server changes and conflicts.
type DeltaResponse struct {
ServerDelta []Change `json:"serverDelta"`
Conflicts []Conflict `json:"conflicts"`
ServerDelta []Change `json:"serverDelta"`
Conflicts []Conflict `json:"conflicts"`
NewSnapshotID string `json:"newSnapshotId,omitempty"`
}
// ResolveRequest uploads conflict resolutions.
type ResolveRequest struct {
SnapshotID string `json:"snapshotId"`
Resolutions []Resolution `json:"resolutions"`
SnapshotID string `json:"snapshotId"`
Resolutions []Resolution `json:"resolutions"`
}
// ResolveResponse confirms the new snapshot.

265
apps/server/internal/sync/service.go
View File

@@ -2,22 +2,33 @@ package sync
import (
"context"
"errors"
"fmt"
"log/slog"
"os"
"path/filepath"
"strings"
"github.com/tim/cairnquire/apps/server/internal/docs"
"github.com/tim/cairnquire/apps/server/internal/store"
)
var (
ErrForbidden = errors.New("sync snapshot does not belong to user")
ErrContentRequired = errors.New("sync content is required")
ErrHashMismatch = errors.New("sync content hash mismatch")
ErrInvalidPath = errors.New("invalid sync path")
ErrInvalidHash = errors.New("invalid sync content hash")
ErrContentNotFound = errors.New("sync content hash not found")
)
// Service handles sync protocol business logic.
type Service struct {
repo *Repository
docService *docs.Service
repo *Repository
docService *docs.Service
contentStore *store.ContentStore
logger *slog.Logger
sourceDir string
logger *slog.Logger
sourceDir string
}
// NewService creates a new sync service.
@@ -48,11 +59,14 @@ func (s *Service) InitSync(ctx context.Context, deviceID, userID string) (*Snaps
}
// ApplyDelta processes client changes and computes server delta + conflicts.
func (s *Service) ApplyDelta(ctx context.Context, snapshotID string, clientDelta Delta) (*DeltaResult, error) {
func (s *Service) ApplyDelta(ctx context.Context, snapshotID, userID string, clientDelta Delta) (*DeltaResult, error) {
snap, err := s.repo.GetSnapshot(ctx, snapshotID)
if err != nil {
return nil, fmt.Errorf("get snapshot: %w", err)
}
if snap.UserID != userID {
return nil, ErrForbidden
}
serverFiles := make(map[string]FileEntry)
for _, f := range snap.Files {
@@ -83,8 +97,8 @@ func (s *Service) ApplyDelta(ctx context.Context, snapshotID string, clientDelta
currentMap[f.Path] = f
}
var serverDelta []Change
var conflicts []Conflict
serverDelta := make([]Change, 0)
conflicts := make([]Conflict, 0)
// Detect server changes since snapshot
for path, current := range currentMap {
@@ -121,13 +135,15 @@ func (s *Service) ApplyDelta(ctx context.Context, snapshotID string, clientDelta
}
// Check for conflicts: both client and server changed same file
conflictPaths := make(map[string]struct{})
for _, clientChange := range clientDelta.Changes {
if clientChange.Type == ChangeCreate || clientChange.Type == ChangeUpdate {
if clientChange.Type == ChangeCreate || clientChange.Type == ChangeUpdate || clientChange.Type == ChangeDelete {
serverCurrent, serverHas := currentMap[clientChange.Path]
serverOld, serverHad := serverFiles[clientChange.Path]
if serverHad && serverHas && serverOld.Hash != serverCurrent.Hash &&
serverCurrent.Hash != clientChange.Hash {
conflictPaths[clientChange.Path] = struct{}{}
conflicts = append(conflicts, Conflict{
Path: clientChange.Path,
ServerHash: serverCurrent.Hash,
@@ -137,73 +153,107 @@ func (s *Service) ApplyDelta(ctx context.Context, snapshotID string, clientDelta
Strategy: ResolutionLastWriteWins,
})
}
if !serverHad && serverHas && serverCurrent.Hash != clientChange.Hash {
conflictPaths[clientChange.Path] = struct{}{}
conflicts = append(conflicts, Conflict{
Path: clientChange.Path,
ServerHash: serverCurrent.Hash,
ClientHash: clientChange.Hash,
ServerModified: serverCurrent.Modified,
ClientModified: clientChange.Modified,
Strategy: ResolutionManualMerge,
})
}
}
}
for _, clientChange := range clientDelta.Changes {
if _, conflicted := conflictPaths[clientChange.Path]; conflicted {
continue
}
if err := s.applyClientChange(ctx, clientChange); err != nil {
return nil, err
}
}
if len(clientDelta.Changes) > 0 {
if _, err := s.docService.SyncSourceDir(ctx); err != nil {
return nil, fmt.Errorf("sync documents after client delta: %w", err)
}
}
latestFiles, err := s.buildSnapshotFromDisk(ctx)
if err != nil {
return nil, fmt.Errorf("build post-delta snapshot: %w", err)
}
newSnap, err := s.repo.CreateSnapshot(ctx, snap.DeviceID, snap.UserID, latestFiles)
if err != nil {
return nil, fmt.Errorf("create post-delta snapshot: %w", err)
}
return &DeltaResult{
ServerDelta: serverDelta,
Conflicts: conflicts,
ServerDelta: serverDelta,
Conflicts: conflicts,
NewSnapshotID: newSnap.ID,
}, nil
}
// ResolveConflicts applies resolved changes and creates a new snapshot.
func (s *Service) ResolveConflicts(ctx context.Context, snapshotID string, resolutions []Resolution) (*Snapshot, error) {
func (s *Service) ResolveConflicts(ctx context.Context, snapshotID, userID string, resolutions []Resolution) (*Snapshot, error) {
snap, err := s.repo.GetSnapshot(ctx, snapshotID)
if err != nil {
return nil, fmt.Errorf("get snapshot: %w", err)
}
// Build a map of resolutions by path
resMap := make(map[string]Resolution)
for _, r := range resolutions {
resMap[r.Path] = r
if snap.UserID != userID {
return nil, ErrForbidden
}
// Get current server state
currentFiles, err := s.buildSnapshotFromDisk(ctx)
if err != nil {
if _, err := s.buildSnapshotFromDisk(ctx); err != nil {
return nil, fmt.Errorf("build current snapshot: %w", err)
}
currentMap := make(map[string]FileEntry)
for _, f := range currentFiles {
currentMap[f.Path] = f
}
// Apply resolutions to create the merged state
merged := make(map[string]FileEntry)
for path, f := range currentMap {
merged[path] = f
}
for _, res := range resolutions {
switch res.Strategy {
case ResolutionClientWins:
// In a real implementation, we'd apply the client's content here.
// For now, we keep the server state and mark it resolved.
if err := s.applyResolvedContent(res.Path, res.Hash, res.Content); err != nil {
return nil, err
}
s.logger.Debug("conflict resolved: client wins", "path", res.Path)
case ResolutionServerWins:
// Keep server state (already in merged)
s.logger.Debug("conflict resolved: server wins", "path", res.Path)
case ResolutionRenameBoth:
if existing, ok := merged[res.Path]; ok {
merged[res.NewPath] = existing
delete(merged, res.Path)
if res.NewPath == "" {
res.NewPath = conflictPath(res.Path)
}
if err := s.applyResolvedContent(res.NewPath, res.Hash, res.Content); err != nil {
return nil, err
}
case ResolutionLastWriteWins:
// Keep whichever is newer - in practice server state since
// we haven't received client content yet
s.logger.Debug("conflict resolved: last-write-wins", "path", res.Path)
if res.Content != "" {
if err := s.applyResolvedContent(res.Path, res.Hash, res.Content); err != nil {
return nil, err
}
}
case ResolutionManualMerge:
// Flag for later UI resolution
if res.Content != "" {
if err := s.applyResolvedContent(res.Path, res.Hash, res.Content); err != nil {
return nil, err
}
}
s.logger.Debug("conflict deferred: manual merge", "path", res.Path)
}
}
// Create new snapshot from merged state
var files []FileEntry
for _, f := range merged {
files = append(files, f)
if len(resolutions) > 0 {
if _, err := s.docService.SyncSourceDir(ctx); err != nil {
return nil, fmt.Errorf("sync documents after resolutions: %w", err)
}
}
files, err := s.buildSnapshotFromDisk(ctx)
if err != nil {
return nil, fmt.Errorf("build resolved snapshot: %w", err)
}
newSnap, err := s.repo.CreateSnapshot(ctx, snap.DeviceID, snap.UserID, files)
@@ -217,12 +267,137 @@ func (s *Service) ResolveConflicts(ctx context.Context, snapshotID string, resol
// GetContent returns raw file content by hash.
func (s *Service) GetContent(hash string) ([]byte, error) {
if !isSHA256Hex(hash) {
return nil, ErrInvalidHash
}
return s.contentStore.Read(hash)
}
func (s *Service) applyClientChange(_ context.Context, change Change) error {
switch change.Type {
case ChangeCreate, ChangeUpdate:
return s.applyResolvedContent(change.Path, change.Hash, change.Content)
case ChangeDelete:
path, err := s.safeContentPath(change.Path)
if err != nil {
return err
}
if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
return fmt.Errorf("delete synced file %s: %w", change.Path, err)
}
return nil
case ChangeRename:
oldPath, err := s.safeContentPath(change.OldPath)
if err != nil {
return err
}
newPath, err := s.safeContentPath(change.Path)
if err != nil {
return err
}
if err := os.MkdirAll(filepath.Dir(newPath), 0o755); err != nil {
return fmt.Errorf("create rename directory %s: %w", change.Path, err)
}
if err := os.Rename(oldPath, newPath); err != nil && !errors.Is(err, os.ErrNotExist) {
return fmt.Errorf("rename synced file %s to %s: %w", change.OldPath, change.Path, err)
}
if change.Content != "" || change.Hash != "" {
return s.applyResolvedContent(change.Path, change.Hash, change.Content)
}
return nil
default:
return nil
}
}
func (s *Service) applyResolvedContent(requestPath, expectedHash, content string) error {
path, err := s.safeContentPath(requestPath)
if err != nil {
return err
}
var bytes []byte
if content != "" {
record, err := s.contentStore.PutBytes([]byte(content))
if err != nil {
return fmt.Errorf("store synced content %s: %w", requestPath, err)
}
if expectedHash != "" && record.Hash != expectedHash {
return fmt.Errorf("%w: %s", ErrHashMismatch, requestPath)
}
bytes = []byte(content)
} else {
if expectedHash == "" {
return fmt.Errorf("%w: %s", ErrContentRequired, requestPath)
}
if !isSHA256Hex(expectedHash) {
return fmt.Errorf("%w: %s", ErrInvalidHash, expectedHash)
}
bytes, err = s.contentStore.Read(expectedHash)
if err != nil {
if errors.Is(err, os.ErrNotExist) {
return fmt.Errorf("%w: %s", ErrContentNotFound, expectedHash)
}
return fmt.Errorf("read synced content %s: %w", expectedHash, err)
}
}
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
return fmt.Errorf("create synced file directory %s: %w", requestPath, err)
}
if err := os.WriteFile(path, bytes, 0o644); err != nil {
return fmt.Errorf("write synced file %s: %w", requestPath, err)
}
return nil
}
func (s *Service) safeContentPath(requestPath string) (string, error) {
if requestPath == "" {
return "", ErrInvalidPath
}
clean := filepath.ToSlash(filepath.Clean(filepath.FromSlash(strings.Trim(requestPath, "/"))))
if clean == "." || clean == ".." || strings.HasPrefix(clean, "../") || filepath.IsAbs(clean) {
return "", ErrInvalidPath
}
if !strings.HasSuffix(strings.ToLower(clean), ".md") {
return "", ErrInvalidPath
}
sourceRoot, err := filepath.Abs(s.sourceDir)
if err != nil {
return "", fmt.Errorf("resolve source root: %w", err)
}
target, err := filepath.Abs(filepath.Join(sourceRoot, filepath.FromSlash(clean)))
if err != nil {
return "", ErrInvalidPath
}
relative, err := filepath.Rel(sourceRoot, target)
if err != nil || relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) {
return "", ErrInvalidPath
}
return target, nil
}
func conflictPath(path string) string {
ext := filepath.Ext(path)
stem := strings.TrimSuffix(path, ext)
return stem + " (conflict)" + ext
}
func isSHA256Hex(hash string) bool {
if len(hash) != 64 {
return false
}
for _, c := range hash {
if (c < '0' || c > '9') && (c < 'a' || c > 'f') && (c < 'A' || c > 'F') {
return false
}
}
return true
}
// buildSnapshotFromDisk walks the source directory and builds a file list.
func (s *Service) buildSnapshotFromDisk(ctx context.Context) ([]FileEntry, error) {
var files []FileEntry
files := make([]FileEntry, 0)
err := filepath.WalkDir(s.sourceDir, func(path string, entry os.DirEntry, err error) error {
if err != nil {
@@ -231,7 +406,7 @@ func (s *Service) buildSnapshotFromDisk(ctx context.Context) ([]FileEntry, error
if entry.IsDir() {
return nil
}
if filepath.Ext(path) != ".md" {
if strings.ToLower(filepath.Ext(path)) != ".md" {
return nil
}

125
apps/server/internal/sync/service_test.go
View File

@@ -2,11 +2,15 @@ package sync
import (
"context"
"crypto/sha256"
"database/sql"
"encoding/hex"
"encoding/json"
"fmt"
"log/slog"
"os"
"path/filepath"
"strings"
"testing"
"time"
@@ -112,7 +116,7 @@ func TestApplyDeltaDetectsServerChanges(t *testing.T) {
}
// Client sends empty delta
result, err := service.ApplyDelta(ctx, snap.ID, Delta{Changes: nil})
result, err := service.ApplyDelta(ctx, snap.ID, "user:test", Delta{Changes: nil})
if err != nil {
t.Fatalf("ApplyDelta() error = %v", err)
}
@@ -133,6 +137,31 @@ func TestApplyDeltaDetectsServerChanges(t *testing.T) {
}
}
func TestApplyDeltaMarshalsEmptyCollectionsAsArrays(t *testing.T) {
service, _ := setupTestService(t)
ctx := context.Background()
snap, err := service.InitSync(ctx, "device-1", "user:test")
if err != nil {
t.Fatalf("InitSync() error = %v", err)
}
result, err := service.ApplyDelta(ctx, snap.ID, "user:test", Delta{})
if err != nil {
t.Fatalf("ApplyDelta() error = %v", err)
}
payload, err := json.Marshal(result)
if err != nil {
t.Fatalf("marshal delta result: %v", err)
}
for _, expected := range []string{`"serverDelta":[]`, `"conflicts":[]`} {
if !strings.Contains(string(payload), expected) {
t.Fatalf("delta JSON = %s, want %s", payload, expected)
}
}
}
func TestApplyDeltaDetectsConflicts(t *testing.T) {
service, sourceDir := setupTestService(t)
ctx := context.Background()
@@ -157,7 +186,7 @@ func TestApplyDeltaDetectsConflicts(t *testing.T) {
Modified: time.Now().UTC(),
}
result, err := service.ApplyDelta(ctx, snap.ID, Delta{Changes: []Change{clientChange}})
result, err := service.ApplyDelta(ctx, snap.ID, "user:test", Delta{Changes: []Change{clientChange}})
if err != nil {
t.Fatalf("ApplyDelta() error = %v", err)
}
@@ -189,15 +218,17 @@ func TestApplyDeltaAllowsDifferentFileEditsWithoutConflict(t *testing.T) {
t.Fatalf("update guide.md: %v", err)
}
clientContent := "# Hello\n\nClient-side hello update"
clientChange := Change{
Type: ChangeUpdate,
Path: "hello.md",
Hash: "clienthash-different-file",
Size: 128,
Hash: sha256Hex(clientContent),
Content: clientContent,
Size: int64(len(clientContent)),
Modified: time.Now().UTC(),
}
result, err := service.ApplyDelta(ctx, snap.ID, Delta{Changes: []Change{clientChange}})
result, err := service.ApplyDelta(ctx, snap.ID, "user:test", Delta{Changes: []Change{clientChange}})
if err != nil {
t.Fatalf("ApplyDelta() error = %v", err)
}
@@ -213,6 +244,71 @@ func TestApplyDeltaAllowsDifferentFileEditsWithoutConflict(t *testing.T) {
}
}
func TestApplyDeltaAppliesClientCreateWithContent(t *testing.T) {
service, sourceDir := setupTestService(t)
ctx := context.Background()
snap, err := service.InitSync(ctx, "device-1", "user:test")
if err != nil {
t.Fatalf("InitSync() error = %v", err)
}
content := "# Inbox\n\nCreated from macOS"
result, err := service.ApplyDelta(ctx, snap.ID, "user:test", Delta{Changes: []Change{{
Type: ChangeCreate,
Path: "inbox/from-mac.md",
Hash: sha256Hex(content),
Content: content,
Size: int64(len(content)),
Modified: time.Now().UTC(),
}}})
if err != nil {
t.Fatalf("ApplyDelta() error = %v", err)
}
if len(result.Conflicts) != 0 {
t.Fatalf("expected no conflicts, got %#v", result.Conflicts)
}
if result.NewSnapshotID == "" || result.NewSnapshotID == snap.ID {
t.Fatalf("new snapshot id = %q, old = %q", result.NewSnapshotID, snap.ID)
}
written, err := os.ReadFile(filepath.Join(sourceDir, "inbox", "from-mac.md"))
if err != nil {
t.Fatalf("read synced file: %v", err)
}
if string(written) != content {
t.Fatalf("synced content = %q, want %q", string(written), content)
}
stored, err := service.GetContent(sha256Hex(content))
if err != nil {
t.Fatalf("GetContent() error = %v", err)
}
if string(stored) != content {
t.Fatalf("stored content = %q, want %q", string(stored), content)
}
}
func TestApplyDeltaRejectsHashMismatch(t *testing.T) {
service, _ := setupTestService(t)
ctx := context.Background()
snap, err := service.InitSync(ctx, "device-1", "user:test")
if err != nil {
t.Fatalf("InitSync() error = %v", err)
}
_, err = service.ApplyDelta(ctx, snap.ID, "user:test", Delta{Changes: []Change{{
Type: ChangeCreate,
Path: "bad.md",
Hash: "not-the-content-hash",
Content: "# Bad\n",
}}})
if err == nil {
t.Fatal("expected hash mismatch error")
}
}
func TestResolveConflictsPreservesServerContent(t *testing.T) {
service, sourceDir := setupTestService(t)
ctx := context.Background()
@@ -227,7 +323,7 @@ func TestResolveConflictsPreservesServerContent(t *testing.T) {
t.Fatalf("update hello.md: %v", err)
}
_, err = service.ResolveConflicts(ctx, snap.ID, []Resolution{{
_, err = service.ResolveConflicts(ctx, snap.ID, "user:test", []Resolution{{
Path: "hello.md",
Strategy: ResolutionServerWins,
}})
@@ -260,7 +356,7 @@ func TestResolveConflictsCreatesNewSnapshot(t *testing.T) {
},
}
newSnap, err := service.ResolveConflicts(ctx, snap.ID, resolutions)
newSnap, err := service.ResolveConflicts(ctx, snap.ID, "user:test", resolutions)
if err != nil {
t.Fatalf("ResolveConflicts() error = %v", err)
}
@@ -300,6 +396,14 @@ func TestGetContentReturnsFileBytes(t *testing.T) {
}
}
func TestGetContentRejectsInvalidHash(t *testing.T) {
service, _ := setupTestService(t)
if _, err := service.GetContent("short"); err == nil {
t.Fatal("expected invalid hash error")
}
}
func TestSnapshotDeltaHandles100FilesUnderTarget(t *testing.T) {
service, sourceDir := setupTestService(t)
ctx := context.Background()
@@ -329,7 +433,7 @@ func TestSnapshotDeltaHandles100FilesUnderTarget(t *testing.T) {
}
start = time.Now()
result, err := service.ApplyDelta(ctx, snap.ID, Delta{})
result, err := service.ApplyDelta(ctx, snap.ID, "user:test", Delta{})
if err != nil {
t.Fatalf("ApplyDelta() error = %v", err)
}
@@ -343,3 +447,8 @@ func TestSnapshotDeltaHandles100FilesUnderTarget(t *testing.T) {
t.Fatalf("server delta path = %q, want note-050.md", result.ServerDelta[0].Path)
}
}
func sha256Hex(content string) string {
sum := sha256.Sum256([]byte(content))
return hex.EncodeToString(sum[:])
}