v0.5.0: parallel uploads

Author: pcarrier

download.go

@@ -0,0 +1,104 @@
+package main
+
+import (
+	"bytes"
+	"fmt"
+	"log"
+	"os"
+	"path/filepath"
+	"strconv"
+	"sync"
+
+	"github.com/fxamacker/cbor/v2"
+	"github.com/xmit-co/xmit/protocol"
+	"github.com/zeebo/blake3"
+)
+
+func download(key, domain, id, destination string) error {
+	downloadParallelism := 3
+	if s := os.Getenv("DOWNLOAD_PARALLELISM"); s != "" {
+		if v, err := strconv.Atoi(s); err == nil {
+			downloadParallelism = v
+		}
+	}
+
+	// Discover endpoint
+	log.Print("🔍 Discovering endpoint…")
+	discovery, err := protocol.Discover()
+	if err != nil {
+		return fmt.Errorf("discovering endpoint: %w", err)
+	}
+	log.Printf("🌐 Using URL: %s", discovery.URL)
+
+	downloader, err := protocol.NewParallelDownloader(discovery.URL, downloadParallelism)
+	if err != nil {
+		return fmt.Errorf("creating parallel downloader: %w", err)
+	}
+
+	resp, err := downloader.DownloadBundle(key, domain, id)
+	if err != nil {
+		return fmt.Errorf("downloading bundle: %w", err)
+	}
+	if !resp.Response.Success {
+		return fmt.Errorf("downloading bundle, server-side: %v", resp.Response.Errors)
+	}
+	var node protocol.Node
+	if err := cbor.NewDecoder(bytes.NewReader(resp.Bundle)).Decode(&node); err != nil {
+		return fmt.Errorf("unmarshaling bundle: %w", err)
+	}
+
+	return downloadTraversal(downloader, key, domain, &node, destination)
+}
+
+func downloadTraversal(downloader *protocol.ParallelDownloader, key, domain string, node *protocol.Node, destination string) error {
+	if node.Hash != nil {
+		hash := *node.Hash
+		if b, err := os.ReadFile(destination); err == nil {
+			h2 := protocol.Hash(blake3.Sum256(b))
+			if bytes.Equal(h2[:], hash[:]) {
+				return nil
+			}
+		}
+		log.Printf("🎁 Downloading %s", destination)
+		resp, err := downloader.DownloadParts(key, domain, []protocol.Hash{hash})
+		if err != nil {
+			return fmt.Errorf("downloading part: %w", err)
+		}
+		if !resp.Response.Success {
+			return fmt.Errorf("downloading part, server-side: %v", resp.Response.Errors)
+		}
+		if len(resp.Parts) == 0 {
+			return fmt.Errorf("no part found for %s", hash)
+		}
+		if err := os.WriteFile(destination, resp.Parts[0], 0644); err != nil {
+			return fmt.Errorf("writing file: %w", err)
+		}
+		log.Printf("✅ Downloaded %s", destination)
+	} else {
+		if err := os.MkdirAll(destination, 0755); err != nil {
+			return err
+		}
+		var errors []error
+		var mu sync.Mutex
+		var wg sync.WaitGroup
+		for name, child := range node.Children {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				if err := downloadTraversal(downloader, key, domain, child, filepath.Join(destination, name)); err != nil {
+					mu.Lock()
+					errors = append(errors, err)
+					mu.Unlock()
+				}
+			}()
+		}
+		wg.Wait()
+		if len(errors) > 0 {
+			for _, err := range errors {
+				log.Printf("🛑 Failed to download: %v", err)
+			}
+			return fmt.Errorf("%d subtraversals failed", len(errors))
+		}
+	}
+	return nil
+}

flake.nix

@@ -15,7 +15,7 @@
           };
           packages.default = pkgs.buildGoModule {
             pname = "xmit";
-            version = "0.4.0";  
+            version = "0.5.0";  
             src = ./.;
             vendorHash = "sha256-beSUbtBgJHlXmgv+cLSOaGP4evXdxN0sOpasw+v8xgs=";
           };

ingest.go

@@ -0,0 +1,59 @@
+package main
+
+import (
+	"log"
+	"os"
+	"path/filepath"
+
+	"github.com/xmit-co/xmit/protocol"
+	"github.com/zeebo/blake3"
+)
+
+type ingestion struct {
+	protocol.Node
+	contents map[protocol.Hash][]byte
+}
+
+func ingest(directory string) (*ingestion, error) {
+	bundle := ingestion{
+		contents: make(map[protocol.Hash][]byte),
+	}
+	err := traverse(directory, &bundle.Node, &bundle.contents)
+	return &bundle, err
+}
+
+func traverse(directory string, node *protocol.Node, contents *map[protocol.Hash][]byte) error {
+	entries, err := os.ReadDir(directory)
+	if err != nil {
+		return err
+	}
+	if node.Children == nil {
+		node.Children = make(map[string]*protocol.Node)
+	}
+	for _, entry := range entries {
+		p := filepath.Join(directory, entry.Name())
+		if entry.IsDir() {
+			if entry.Name() == ".git" {
+				log.Printf("😇 Skipping %s", p)
+				continue
+			}
+			child := protocol.Node{}
+			err := traverse(p, &child, contents)
+			if err != nil {
+				return err
+			}
+			node.Children[entry.Name()] = &child
+		} else {
+			bytes, err := os.ReadFile(p)
+			if err != nil {
+				return err
+			}
+			hash := protocol.Hash(blake3.Sum256(bytes))
+			node.Children[entry.Name()] = &protocol.Node{
+				Hash: &hash,
+			}
+			(*contents)[hash] = bytes
+		}
+	}
+	return nil
+}

key.go

@@ -0,0 +1,31 @@
+package main
+
+import (
+	"os"
+	"path"
+	"path/filepath"
+	"strings"
+
+	"github.com/kirsle/configdir"
+)
+
+var (
+	keyPath = path.Join(configdir.LocalConfig("xmit"), "key")
+)
+
+func findKey() string {
+	if key, found := os.LookupEnv("XMIT_KEY"); found {
+		return key
+	}
+	if b, err := os.ReadFile(keyPath); err == nil {
+		return strings.TrimSpace(string(b))
+	}
+	return ""
+}
+
+func storeKey(key string) error {
+	if err := os.MkdirAll(filepath.Dir(keyPath), 0700); err != nil {
+		return err
+	}
+	return os.WriteFile(keyPath, []byte(key), 0600)
+}