497 lines
14 KiB
Go
497 lines
14 KiB
Go
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
|
|
contentStore *store.ContentStore
|
|
logger *slog.Logger
|
|
sourceDir string
|
|
}
|
|
|
|
type FileAccessFunc func(FileEntry) bool
|
|
type PathAccessFunc func(string) bool
|
|
|
|
// NewService creates a new sync service.
|
|
func NewService(repo *Repository, docService *docs.Service, contentStore *store.ContentStore, sourceDir string, logger *slog.Logger) *Service {
|
|
return &Service{
|
|
repo: repo,
|
|
docService: docService,
|
|
contentStore: contentStore,
|
|
logger: logger,
|
|
sourceDir: sourceDir,
|
|
}
|
|
}
|
|
|
|
// InitSync creates a new snapshot from the current server state.
|
|
func (s *Service) InitSync(ctx context.Context, deviceID, userID string) (*Snapshot, error) {
|
|
return s.InitSyncFiltered(ctx, deviceID, userID, nil)
|
|
}
|
|
|
|
func (s *Service) InitSyncFiltered(ctx context.Context, deviceID, userID string, canRead FileAccessFunc) (*Snapshot, error) {
|
|
files, err := s.buildSnapshotFromDisk(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("build snapshot: %w", err)
|
|
}
|
|
files = filterFiles(files, canRead)
|
|
|
|
snap, err := s.repo.CreateSnapshot(ctx, deviceID, userID, files)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create snapshot: %w", err)
|
|
}
|
|
|
|
s.logger.Info("sync initialized", "device", deviceID, "snapshot", snap.ID, "files", len(files))
|
|
return snap, nil
|
|
}
|
|
|
|
// ApplyDelta processes client changes and computes server delta + conflicts.
|
|
func (s *Service) ApplyDelta(ctx context.Context, snapshotID, userID string, clientDelta Delta) (*DeltaResult, error) {
|
|
return s.ApplyDeltaFiltered(ctx, snapshotID, userID, clientDelta, nil, nil)
|
|
}
|
|
|
|
func (s *Service) ApplyDeltaFiltered(ctx context.Context, snapshotID, userID string, clientDelta Delta, canRead FileAccessFunc, canWrite PathAccessFunc) (*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
|
|
}
|
|
for _, clientChange := range clientDelta.Changes {
|
|
if canWrite != nil && !canWrite(clientChange.Path) {
|
|
return nil, ErrForbidden
|
|
}
|
|
if clientChange.OldPath != "" && canWrite != nil && !canWrite(clientChange.OldPath) {
|
|
return nil, ErrForbidden
|
|
}
|
|
}
|
|
|
|
serverFiles := make(map[string]FileEntry)
|
|
for _, f := range snap.Files {
|
|
serverFiles[f.Path] = f
|
|
}
|
|
|
|
clientFiles := make(map[string]FileEntry)
|
|
for _, c := range clientDelta.Changes {
|
|
if c.Type == ChangeDelete || c.Type == ChangeRename {
|
|
continue
|
|
}
|
|
clientFiles[c.Path] = FileEntry{
|
|
Path: c.Path,
|
|
Hash: c.Hash,
|
|
Size: c.Size,
|
|
Modified: c.Modified,
|
|
}
|
|
}
|
|
|
|
// Rebuild current server state (may have changed since snapshot)
|
|
currentFiles, err := s.buildSnapshotFromDisk(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("build current snapshot: %w", err)
|
|
}
|
|
currentFiles = filterFiles(currentFiles, canRead)
|
|
|
|
currentMap := make(map[string]FileEntry)
|
|
for _, f := range currentFiles {
|
|
currentMap[f.Path] = f
|
|
}
|
|
|
|
serverDelta := make([]Change, 0)
|
|
conflicts := make([]Conflict, 0)
|
|
|
|
// Detect server changes since snapshot
|
|
for path, current := range currentMap {
|
|
old, existed := serverFiles[path]
|
|
if !existed {
|
|
// Server created this file
|
|
serverDelta = append(serverDelta, Change{
|
|
Type: ChangeCreate,
|
|
Path: path,
|
|
Hash: current.Hash,
|
|
Size: current.Size,
|
|
Modified: current.Modified,
|
|
})
|
|
} else if old.Hash != current.Hash {
|
|
// Server updated this file
|
|
serverDelta = append(serverDelta, Change{
|
|
Type: ChangeUpdate,
|
|
Path: path,
|
|
Hash: current.Hash,
|
|
Size: current.Size,
|
|
Modified: current.Modified,
|
|
})
|
|
}
|
|
}
|
|
|
|
// Detect server deletions
|
|
for path := range serverFiles {
|
|
if _, exists := currentMap[path]; !exists {
|
|
serverDelta = append(serverDelta, Change{
|
|
Type: ChangeDelete,
|
|
Path: path,
|
|
})
|
|
}
|
|
}
|
|
|
|
// 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 || 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,
|
|
ClientHash: clientChange.Hash,
|
|
ServerModified: serverCurrent.Modified,
|
|
ClientModified: clientChange.Modified,
|
|
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)
|
|
}
|
|
latestFiles = filterFiles(latestFiles, canRead)
|
|
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,
|
|
NewSnapshotID: newSnap.ID,
|
|
}, nil
|
|
}
|
|
|
|
// ResolveConflicts applies resolved changes and creates a new snapshot.
|
|
func (s *Service) ResolveConflicts(ctx context.Context, snapshotID, userID string, resolutions []Resolution) (*Snapshot, error) {
|
|
return s.ResolveConflictsFiltered(ctx, snapshotID, userID, resolutions, nil, nil)
|
|
}
|
|
|
|
func (s *Service) ResolveConflictsFiltered(ctx context.Context, snapshotID, userID string, resolutions []Resolution, canRead FileAccessFunc, canWrite PathAccessFunc) (*Snapshot, 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
|
|
}
|
|
for _, resolution := range resolutions {
|
|
if canWrite != nil && !canWrite(resolution.Path) {
|
|
return nil, ErrForbidden
|
|
}
|
|
if resolution.NewPath != "" && canWrite != nil && !canWrite(resolution.NewPath) {
|
|
return nil, ErrForbidden
|
|
}
|
|
}
|
|
|
|
// Get current server state
|
|
if _, err := s.buildSnapshotFromDisk(ctx); err != nil {
|
|
return nil, fmt.Errorf("build current snapshot: %w", err)
|
|
}
|
|
|
|
for _, res := range resolutions {
|
|
switch res.Strategy {
|
|
case ResolutionClientWins:
|
|
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:
|
|
s.logger.Debug("conflict resolved: server wins", "path", res.Path)
|
|
case ResolutionRenameBoth:
|
|
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:
|
|
if res.Content != "" {
|
|
if err := s.applyResolvedContent(res.Path, res.Hash, res.Content); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
case ResolutionManualMerge:
|
|
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)
|
|
}
|
|
}
|
|
|
|
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)
|
|
}
|
|
files = filterFiles(files, canRead)
|
|
|
|
newSnap, err := s.repo.CreateSnapshot(ctx, snap.DeviceID, snap.UserID, files)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create new snapshot: %w", err)
|
|
}
|
|
|
|
s.logger.Info("conflicts resolved", "snapshot", snapshotID, "newSnapshot", newSnap.ID, "resolutions", len(resolutions))
|
|
return newSnap, nil
|
|
}
|
|
|
|
func filterFiles(files []FileEntry, canRead FileAccessFunc) []FileEntry {
|
|
if canRead == nil {
|
|
return files
|
|
}
|
|
filtered := make([]FileEntry, 0, len(files))
|
|
for _, file := range files {
|
|
if canRead(file) {
|
|
filtered = append(filtered, file)
|
|
}
|
|
}
|
|
return filtered
|
|
}
|
|
|
|
// 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) {
|
|
files := make([]FileEntry, 0)
|
|
|
|
err := filepath.WalkDir(s.sourceDir, func(path string, entry os.DirEntry, err error) error {
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if entry.IsDir() {
|
|
return nil
|
|
}
|
|
if strings.ToLower(filepath.Ext(path)) != ".md" {
|
|
return nil
|
|
}
|
|
|
|
info, err := entry.Info()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
relative, err := filepath.Rel(s.sourceDir, path)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
relative = filepath.ToSlash(relative)
|
|
|
|
content, err := os.ReadFile(path)
|
|
if err != nil {
|
|
return fmt.Errorf("read file %s: %w", path, err)
|
|
}
|
|
|
|
record, err := s.contentStore.PutBytes(content)
|
|
if err != nil {
|
|
return fmt.Errorf("store file %s: %w", path, err)
|
|
}
|
|
|
|
files = append(files, FileEntry{
|
|
Path: relative,
|
|
Hash: record.Hash,
|
|
Size: info.Size(),
|
|
Modified: info.ModTime().UTC(),
|
|
})
|
|
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return files, nil
|
|
}
|