Backup & Restore Flow

// From AGENTS.md and internal/engine/
1. BackupManager acquires a shared lock
2. Load the previous snapshot (if any) for this source identity
3. Extract the previous HAMT root reference

source.Walk(func(entry SourceEntry) error {
    // Process each file/folder
})

changeToken := previousSnapshot.ChangeToken
source.WalkChanges(changeToken, func(entry SourceEntry) error {
    // Process only changed files
})

for _, queuedFile := range uploadQueue {
    go func(file SourceEntry) {
        // 1. Stream file content
        reader := source.Open(file.FileID)
        
        // 2. Content-defined chunking
        chunks := fastCDC(reader, minSize, avgSize, maxSize)
        
        // 3. For each chunk:
        for chunk := range chunks {
            // Compute hash (HMAC-SHA256 or SHA-256)
            hash := computeChunkHash(chunk.Data, dedupKey)
            
            // Deduplicate
            if store.Exists("chunk/" + hash) {
                continue // Skip, already stored
            }
            
            // Compress and write
            compressed := zstd.Compress(chunk.Data)
            store.Put("chunk/" + hash, compressed)
        }
        
        // 4. Create content object
        contentObj := Content{
            Type: "content",
            Size: file.Size,
            Chunks: chunkRefs,
        }
        contentHash := sha256(file.RawData)
        store.Put("content/" + contentHash, json(contentObj))
        
        // 5. Create filemeta object
        fileMeta := FileMeta{
            FileID: file.FileID,
            Name: file.Name,
            Type: file.Type,
            ContentHash: contentHash,
            Size: file.Size,
            Mtime: file.Mtime,
            Parents: parentRefs,
            ...
        }
        metaRef := "filemeta/" + sha256(json(fileMeta))
        store.Put(metaRef, json(fileMeta))
        
        // 6. Insert into new HAMT
        newRoot = hamtTree.Insert(newRoot, file.FileID, metaRef)
    }(queuedFile)
}

// TransactionalStore buffers nodes in memory
// Only flush the reachable subset from the final root
txnStore.Flush(newRoot)

// 1. Create snapshot object
snapshot := Snapshot{
    Version: 1,
    Created: time.Now().Format(time.RFC3339),
    Root: newRoot,
    Seq: previousSeq + 1,
    Source: sourceInfo,
    ChangeToken: newChangeToken, // For incremental sources
    ...
}
snapshotRef := "snapshot/" + sha256(json(snapshot))
store.Put(snapshotRef, json(snapshot))

// 2. Update index/latest (the commit point)
index := Index{
    LatestSnapshot: snapshotRef,
    Seq: snapshot.Seq,
}
store.Put("index/latest", json(index))

// 3. Update index/snapshots catalog
catalog = append(catalog, SnapshotSummary{...})
store.Put("index/snapshots", json(catalog))

// 4. Update index/packs catalog (if packfiles enabled)
// Automatically handled by the PackStore layer

// Resolve snapshot by:
// - Explicit ref ("snapshot/<hash>")
// - Sequence number (42)
// - "latest" keyword
snapshot := resolveSnapshot(snapshotID)
root := snapshot.Root

var entries []FileMeta
hamtTree.Walk(root, func(key, metaRef string) error {
    // Load filemeta object
    metaData := store.Get(metaRef)
    meta := json.Unmarshal(metaData)
    entries = append(entries, meta)
})

// Build parent dependency graph
graph := buildDependencyGraph(entries)

// Topological sort
sorted := topologicalSort(graph)

func buildPath(meta FileMeta) string {
    if len(meta.Parents) == 0 {
        return meta.Name // Root entry
    }
    
    // Load first parent (arbitrary choice if multi-parent)
    parentRef := meta.Parents[0]
    parentData := store.Get(parentRef)
    parentMeta := json.Unmarshal(parentData)
    
    parentPath := buildPath(parentMeta)
    return parentPath + "/" + meta.Name
}

zipWriter := zip.NewWriter(output)

for _, meta := range sortedEntries {
    path := buildPath(meta)
    
    if meta.Type == "folder" {
        // Create directory entry
        header := &zip.FileHeader{
            Name: path + "/",
            Method: zip.Store,
        }
        header.SetModTime(time.Unix(meta.Mtime, 0))
        zipWriter.CreateHeader(header)
    } else {
        // Create file entry
        header := &zip.FileHeader{
            Name: path,
            Method: zip.Deflate,
        }
        header.SetModTime(time.Unix(meta.Mtime, 0))
        
        writer, _ := zipWriter.CreateHeader(header)
        
        // Load content object
        contentData := store.Get("content/" + meta.ContentHash)
        content := json.Unmarshal(contentData)
        
        // Stream chunks
        for _, chunkRef := range content.Chunks {
            // Fetch and decompress chunk
            compressedData := store.Get(chunkRef)
            rawData := zstd.Decompress(compressedData)
            writer.Write(rawData)
        }
    }
}

zipWriter.Close()

workerPool := make(chan struct{}, 10) // Semaphore

for _, file := range uploadQueue {
    workerPool <- struct{}{} // Acquire
    go func(f SourceEntry) {
        defer func() { <-workerPool }() // Release
        processFile(f)
    }(file)
}

if store.Exists("chunk/" + hash) {
    continue // Skip, already stored
}

if store.Exists("content/" + file.ContentHash) {
    // Entire file upload skipped!
    // Only create new filemeta and HAMT nodes
}

Snapshot 1: 10,000 files → 10,000 filemeta + ~500 HAMT nodes
Snapshot 2: 10 files changed → 10 new filemeta + ~5 new HAMT nodes

Incremental storage: ~15 objects, not 10,510

retry.Do(
    func() error { return store.Put(key, data) },
    retry.Attempts(3),
    retry.Delay(time.Second),
    retry.DelayType(retry.BackOffDelay),
)

if isPermanentError(err) {
    return fmt.Errorf("backup failed: %w", err)
}

hamtTree.Diff(snapshot1.Root, snapshot2.Root, func(entry DiffEntry) error {
    if entry.OldValue == "" {
        fmt.Printf("+ %s\n", entry.Key) // Added
    } else if entry.NewValue == "" {
        fmt.Printf("- %s\n", entry.Key) // Removed
    } else {
        fmt.Printf("M %s\n", entry.Key) // Modified
    }
})