explewd/flit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add flit watch: one-way folder sync to a trusted peer

Browse Source 
Watches local directories and auto-pushes new/changed files to a
known daemon peer via fsnotify, with startup reconciliation, a
stable_after debounce, and per-file JSON state so failed sends retry
on the next connection. Multiple watch entries targeting the same
peer share a single connection/room (watchPeerGroup), resolving the
room-name-collision question flagged in PROPOSAL-watch.md. Push-only,
never propagates deletes.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
explewd
2026-07-12 21:50:18 +02:00
parent 0ffe9fc29d
commit 932a4bd7bb
6 changed files with 602 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
README.md
View File

@@ -20,6 +20,7 @@ It is deliberately not a replacement for [Zipline](https://github.com/diced/zipl
- Auto-downloads on the receiver side — no "click to save" step.
- In-app QR scanner (Chrome/Android) — scan the sender's QR without leaving flit.
- Android share sheet integration — "Share → flit" works from Photos, Files, any app.
- One-way folder sync (`flit watch`) — drop a file into a watched folder, it gets pushed to a trusted peer automatically, no manual send.
---
@@ -101,6 +102,7 @@ For headless machines (homelab boxes, servers) that can't run a browser.
flit send path/to/file.tar.gz # print QR + invite, wait for peer, send
flit recv "flit:eyJhbmNob3..." # join from invite string, accept file
flit daemon # persistent receiver for trusted peers
flit watch [--dry-run] # watch folders, auto-push new files to trusted peers
```
**Built with:** Go, pion/webrtc, libsodium (via nhooyr.io/websocket + internal crypto).
@@ -132,6 +134,38 @@ Copy `cli/daemon.toml.example` to `~/.flit/daemon.toml` to get started. The daem
**Bootstrapping:** to get a peer's `id`, pair once via QR (`flit send` / PWA "Invite new") and copy the hex ID from the "peer connected" log line, or read it from the PWA's Known devices list after tapping "Remember this device."
### Watch mode
`flit watch` is the send-side sibling of `flit daemon`: it watches one or more local folders and automatically pushes new or changed files to a specific trusted peer — no manual `flit send`, no QR, no per-file interaction. It's **one-way sync only** (source → destination, push only) — deliberately not a general bidirectional sync tool, and it never propagates deletes.
The receiving side needs no changes: an already-running `flit daemon`, configured with the watcher's device as a trusted peer, already does the right thing on receipt.
Config lives at `~/.flit/watch.toml`:
```toml
signal_url = "wss://your-anchor.example.com/ws"
turn_url = "" # optional
turn_secret = "" # optional
[[watch]]
dir = "/home/user/backups/documents"
peer_id = "<hex Ed25519 pubkey of the destination device>"
peer_label = "home-server"
pattern = "*.pdf,*.docx" # optional glob include-list; default "*" if omitted
stable_after = "10s" # optional quiet period before considering a file ready
```
Multiple `[[watch]]` blocks are allowed — different folders can go to different peers, or the same peer (entries sharing a `peer_id` share a single connection).
Copy `cli/watch.toml.example` to `~/.flit/watch.toml` to get started. Notes on behavior:
- **Startup reconciliation** — on start, before subscribing to filesystem events, `flit watch` walks each `dir` and diffs it against local state, so anything that changed while the watcher wasn't running gets picked up (not just live events going forward).
- **Debounce (`stable_after`)** — a file is only considered ready to send after this quiet period has passed since its last write, so a large file being actively written isn't offered mid-write.
- **State tracking** — one JSON state file per `[[watch]]` entry lives in `~/.flit/watch-state/`, recording what's been sent. If a send fails partway (peer disconnects mid-transfer), the file is retried on the next successful connection — no separate retry logic needed. Losing this state is harmless (worst case: the whole directory gets re-sent) but not something to engineer around.
- **`flit watch --dry-run`** walks the configured directories and prints what *would* be sent, without connecting to anything — use it to sanity-check a new `watch.toml` before it goes live.
- **`pattern` should be as narrow as your use case allows.** `*` is the default if omitted, but an unconstrained watch on a broad directory turns "sync my documents" into "sync everything that ever lands in this tree." There's no per-file confirmation in watch mode — the directory and pattern in `watch.toml` are the only safety knob.
- Reconnects with the same exponential backoff (2s → 30s cap) as daemon mode, and reuses the same trusted-peer identity model — a `peer_id` here is the same kind of already-known device as anything in the PWA's Known devices list or `daemon.toml`.
---
## Self-hosting

8
cli/cmd/flit/main.go
View File

@@ -4,6 +4,7 @@
// flit send <path> generate a one-shot room, print QR + invite, wait for a peer, send
// flit recv <invite|code> join a room from an invite string, accept the offered file
// flit daemon run as a persistent receiver for trusted peers (reads ~/.flit/daemon.toml)
// flit watch [--dry-run] watch folders and auto-push new files to trusted peers (reads ~/.flit/watch.toml)
package main
import (
@@ -93,7 +94,7 @@ func cfg() transport.Config {
func main() {
if len(os.Args) < 2 {
fmt.Println("usage: flit send <path> | flit recv <invite> | flit daemon")
fmt.Println("usage: flit id | flit send <path> | flit recv <invite> | flit daemon | flit watch [--dry-run]")
os.Exit(1)
}
@@ -117,8 +118,11 @@ func main() {
recv(ctx, os.Args[2])
case "daemon":
runDaemon(ctx)
case "watch":
dryRun := len(os.Args) > 2 && os.Args[2] == "--dry-run"
runWatch(ctx, dryRun)
default:
fmt.Println("usage: flit id | flit send <path> | flit recv <invite> | flit daemon")
fmt.Println("usage: flit id | flit send <path> | flit recv <invite> | flit daemon | flit watch [--dry-run]")
os.Exit(1)
}
}

532
cli/cmd/flit/watch.go Normal file
View File

@@ -0,0 +1,532 @@
package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"log"
"os"
"path/filepath"
"sort"
"strings"
"sync"
"time"
"github.com/BurntSushi/toml"
"github.com/fsnotify/fsnotify"
"flit/internal/transport"
)
// watchFileConfig is read from ~/.flit/watch.toml.
type watchFileConfig struct {
SignalURL string `toml:"signal_url"`
TurnURL string `toml:"turn_url"`
TurnSecret string `toml:"turn_secret"`
Watches []watchEntryConfig `toml:"watch"`
}
type watchEntryConfig struct {
Dir string `toml:"dir"`
PeerID string `toml:"peer_id"`
PeerLabel string `toml:"peer_label"`
Pattern string `toml:"pattern"`
StableAfter string `toml:"stable_after"`
}
func watchConfigPath() string {
home, _ := os.UserHomeDir()
return filepath.Join(home, ".flit", "watch.toml")
}
func watchStateDir() string {
home, _ := os.UserHomeDir()
return filepath.Join(home, ".flit", "watch-state")
}
func loadWatchConfig() (*watchFileConfig, error) {
path := watchConfigPath()
var cfg watchFileConfig
if _, err := toml.DecodeFile(path, &cfg); err != nil {
return nil, fmt.Errorf("reading %s: %w", path, err)
}
if cfg.SignalURL == "" {
return nil, fmt.Errorf("%s: signal_url is required", path)
}
if len(cfg.Watches) == 0 {
return nil, fmt.Errorf("%s: no [[watch]] entries configured", path)
}
for i := range cfg.Watches {
w := &cfg.Watches[i]
if w.Dir == "" {
return nil, fmt.Errorf("%s: watch entry %d missing dir", path, i)
}
w.Dir = expandHome(w.Dir)
if w.PeerID == "" {
return nil, fmt.Errorf("%s: watch entry for %s missing peer_id", path, w.Dir)
}
if w.PeerLabel == "" {
w.PeerLabel = w.PeerID[:min16(w.PeerID)]
}
if w.Pattern == "" {
w.Pattern = "*"
}
if w.StableAfter == "" {
w.StableAfter = "10s"
}
}
return &cfg, nil
}
// fileState is the per-file record kept in a watch entry's state file.
// SentAt is nil until SendFile has returned successfully for exactly this
// size/mtime — that's the entire "needs sending" queue (see PROPOSAL-watch.md §3).
type fileState struct {
Size int64 `json:"size"`
Mtime int64 `json:"mtime"`
SHA256 string `json:"sha256,omitempty"`
SentAt *int64 `json:"sent_at"`
}
// watchEntry is one [[watch]] block: a directory watched for one peer.
type watchEntry struct {
cfg watchEntryConfig
stableAfter time.Duration
statePath string
mu sync.Mutex
state map[string]fileState
}
func newWatchEntry(cfg watchEntryConfig) (*watchEntry, error) {
d, err := time.ParseDuration(cfg.StableAfter)
if err != nil {
return nil, fmt.Errorf("dir %s: invalid stable_after %q: %w", cfg.Dir, cfg.StableAfter, err)
}
sum := sha256.Sum256([]byte(cfg.Dir + cfg.PeerID))
e := &watchEntry{
cfg: cfg,
stableAfter: d,
statePath: filepath.Join(watchStateDir(), hex.EncodeToString(sum[:])+".json"),
state: map[string]fileState{},
}
return e, nil
}
func (e *watchEntry) load() error {
data, err := os.ReadFile(e.statePath)
if os.IsNotExist(err) {
return nil
}
if err != nil {
return err
}
e.mu.Lock()
defer e.mu.Unlock()
return json.Unmarshal(data, &e.state)
}
func (e *watchEntry) save() error {
e.mu.Lock()
data, err := json.MarshalIndent(e.state, "", " ")
e.mu.Unlock()
if err != nil {
return err
}
if err := os.MkdirAll(filepath.Dir(e.statePath), 0700); err != nil {
return err
}
return os.WriteFile(e.statePath, data, 0600)
}
// matches checks a file's base name against the entry's comma-separated
// glob pattern list.
func (e *watchEntry) matches(name string) bool {
for _, pat := range strings.Split(e.cfg.Pattern, ",") {
pat = strings.TrimSpace(pat)
if pat == "" {
continue
}
if ok, _ := filepath.Match(pat, name); ok {
return true
}
}
return false
}
// noteObserved records the current size/mtime for relPath and, if they
// differ from what was last successfully sent (or nothing was ever sent),
// marks the file ready to send by clearing sent_at.
func (e *watchEntry) noteObserved(relPath string, size, mtime int64) (readyNow bool) {
e.mu.Lock()
defer e.mu.Unlock()
cur, ok := e.state[relPath]
if ok && cur.SentAt != nil && cur.Size == size && cur.Mtime == mtime {
return false // already sent this exact version
}
e.state[relPath] = fileState{Size: size, Mtime: mtime, SHA256: cur.SHA256, SentAt: nil}
return true
}
func (e *watchEntry) markSent(relPath string, size, mtime int64, sha string) {
e.mu.Lock()
defer e.mu.Unlock()
now := time.Now().Unix()
e.state[relPath] = fileState{Size: size, Mtime: mtime, SHA256: sha, SentAt: &now}
}
// pendingSorted returns relative paths currently marked as needing a send,
// in a deterministic order.
func (e *watchEntry) pendingSorted() []string {
e.mu.Lock()
defer e.mu.Unlock()
var out []string
for rel, st := range e.state {
if st.SentAt == nil {
out = append(out, rel)
}
}
sort.Strings(out)
return out
}
// reconcile walks the watched directory and updates in-memory state for
// anything that's new or changed since it was last recorded (PROPOSAL-watch.md §4).
// It does not touch the filesystem beyond reading; call save() afterward to persist.
func (e *watchEntry) reconcile() error {
entries, err := os.ReadDir(e.cfg.Dir)
if err != nil {
return err
}
for _, de := range entries {
if de.IsDir() || !e.matches(de.Name()) {
continue
}
info, err := de.Info()
if err != nil {
continue
}
e.noteObserved(de.Name(), info.Size(), info.ModTime().Unix())
}
return nil
}
// watchPeerGroup shares one persistent connection across every [[watch]]
// entry targeting the same peer_id (PROPOSAL-watch.md open question #2).
type watchPeerGroup struct {
peerID string
label string
entries []*watchEntry
mu sync.Mutex
sess *transport.Session
sendMu sync.Mutex
}
func (g *watchPeerGroup) currentSession() *transport.Session {
g.mu.Lock()
defer g.mu.Unlock()
return g.sess
}
// sendOne sends one file and, on success, advances that file's state past
// "needs sending". On failure the state is left untouched, so the file
// remains eligible for retry on the next drain (PROPOSAL-watch.md §3).
func (g *watchPeerGroup) sendOne(sess *transport.Session, e *watchEntry, rel string) error {
g.sendMu.Lock()
defer g.sendMu.Unlock()
full := filepath.Join(e.cfg.Dir, rel)
info, err := os.Stat(full)
if err != nil {
return err
}
log.Printf("[%s] → %s", g.label, rel)
if err := sess.SendFile(full); err != nil {
return err
}
e.markSent(rel, info.Size(), info.ModTime().Unix(), "")
if err := e.save(); err != nil {
log.Printf("[%s] warning: could not persist state for %s: %v", g.label, e.cfg.Dir, err)
}
log.Printf("[%s] ✓ sent %s", g.label, rel)
return nil
}
// drainAll sends every currently-pending file across every entry in the
// group, sequentially (PROPOSAL-watch.md §5). Stops early on the first
// send error, since that almost always means the connection just dropped;
// remaining files stay queued for the next successful connect.
func (g *watchPeerGroup) drainAll() {
sess := g.currentSession()
if sess == nil {
return
}
for _, e := range g.entries {
for _, rel := range e.pendingSorted() {
if g.currentSession() != sess {
return // disconnected mid-drain
}
if err := g.sendOne(sess, e, rel); err != nil {
log.Printf("[%s] send failed for %s: %v", g.label, rel, err)
return
}
}
}
}
// attemptSendNow sends rel immediately if the group is currently connected.
// If not connected, the file simply stays marked pending — the next
// successful connect's drainAll will pick it up.
func (g *watchPeerGroup) attemptSendNow(e *watchEntry, rel string) {
sess := g.currentSession()
if sess == nil {
return
}
if err := g.sendOne(sess, e, rel); err != nil {
log.Printf("[%s] send failed for %s: %v", g.label, rel, err)
}
}
// runWatchGroup mirrors daemon.go's runPeerLoop: a persistent connection to
// one trusted peer, reconnecting with the same backoff. It differs only in
// what drives sends — pending state across the group's entries, rather than
// incoming OnFileOffer.
func runWatchGroup(ctx context.Context, tc transport.Config, myID string, g *watchPeerGroup) {
room := transport.PairRoomName(myID, g.peerID)
log.Printf("[%s] room: %s", g.label, room)
backoff := 2 * time.Second
for {
if ctx.Err() != nil {
return
}
log.Printf("[%s] connecting…", g.label)
disconnected := make(chan struct{})
sess, err := transport.NewSession(dataDir(), tc)
if err != nil {
log.Printf("[%s] session error: %v", g.label, err)
} else {
sess.OnConnected = func(verified bool) {
status := "unverified"
if verified {
status = "verified"
}
log.Printf("[%s] connected (%s)", g.label, status)
backoff = 2 * time.Second
g.mu.Lock()
g.sess = sess
g.mu.Unlock()
go g.drainAll()
}
sess.OnDisconnected = func() {
g.mu.Lock()
if g.sess == sess {
g.sess = nil
}
g.mu.Unlock()
select {
case <-disconnected:
default:
close(disconnected)
}
}
if err := sess.Join(ctx, room, g.peerID); err != nil {
log.Printf("[%s] join error: %v", g.label, err)
} else {
select {
case <-ctx.Done():
return
case <-disconnected:
log.Printf("[%s] disconnected", g.label)
}
}
}
select {
case <-ctx.Done():
return
case <-time.After(backoff):
}
if backoff < 30*time.Second {
backoff *= 2
}
}
}
// watchDirLive subscribes to filesystem events for one entry and, after a
// stable_after quiet period with no further writes to a given file, marks
// it ready and (if connected) sends it immediately. Deletes/renames are
// deliberately ignored (PROPOSAL-watch.md §7) — this is push-only, one-way.
func watchDirLive(ctx context.Context, e *watchEntry, g *watchPeerGroup) error {
w, err := fsnotify.NewWatcher()
if err != nil {
return err
}
defer w.Close()
if err := w.Add(e.cfg.Dir); err != nil {
return err
}
var mu sync.Mutex
timers := map[string]*time.Timer{}
fire := func(name string) {
full := filepath.Join(e.cfg.Dir, name)
info, err := os.Stat(full)
if err != nil {
return // removed again before it went stable
}
if e.noteObserved(name, info.Size(), info.ModTime().Unix()) {
_ = e.save()
g.attemptSendNow(e, name)
}
}
for {
select {
case <-ctx.Done():
mu.Lock()
for _, t := range timers {
t.Stop()
}
mu.Unlock()
return nil
case ev, ok := <-w.Events:
if !ok {
return nil
}
if ev.Op&(fsnotify.Write|fsnotify.Create) == 0 {
continue
}
name := filepath.Base(ev.Name)
if !e.matches(name) {
continue
}
mu.Lock()
if t, ok := timers[name]; ok {
t.Stop()
}
timers[name] = time.AfterFunc(e.stableAfter, func() { fire(name) })
mu.Unlock()
case err, ok := <-w.Errors:
if !ok {
return nil
}
log.Printf("watch %s: fsnotify error: %v", e.cfg.Dir, err)
}
}
}
func runWatch(ctx context.Context, dryRun bool) {
cfg, err := loadWatchConfig()
if err != nil {
fmt.Fprintln(os.Stderr, "flit watch:", err)
fmt.Fprintf(os.Stderr, "create %s — see watch.toml.example\n", watchConfigPath())
os.Exit(1)
}
entries := make([]*watchEntry, 0, len(cfg.Watches))
for _, wc := range cfg.Watches {
if _, err := os.Stat(wc.Dir); err != nil {
fmt.Fprintf(os.Stderr, "flit watch: %s: %v\n", wc.Dir, err)
os.Exit(1)
}
e, err := newWatchEntry(wc)
if err != nil {
fmt.Fprintln(os.Stderr, "flit watch:", err)
os.Exit(1)
}
entries = append(entries, e)
}
if dryRun {
for _, e := range entries {
if err := e.load(); err != nil {
fmt.Fprintf(os.Stderr, "flit watch: %s: reading state: %v\n", e.cfg.Dir, err)
os.Exit(1)
}
if err := e.reconcile(); err != nil {
fmt.Fprintf(os.Stderr, "flit watch: %s: %v\n", e.cfg.Dir, err)
os.Exit(1)
}
pending := e.pendingSorted()
fmt.Printf("%s → %s (%d file(s) would be sent):\n", e.cfg.Dir, e.cfg.PeerLabel, len(pending))
for _, rel := range pending {
fmt.Printf(" %s\n", rel)
}
}
return
}
for _, e := range entries {
if err := e.load(); err != nil {
fmt.Fprintf(os.Stderr, "flit watch: %s: reading state: %v\n", e.cfg.Dir, err)
os.Exit(1)
}
}
tc := transport.Config{SignalURL: cfg.SignalURL, TurnURL: cfg.TurnURL, TurnSecret: cfg.TurnSecret}
probe, err := transport.NewSession(dataDir(), tc)
if err != nil {
fmt.Fprintln(os.Stderr, "flit watch: identity error:", err)
os.Exit(1)
}
myID := probe.PeerID()
groups := map[string]*watchPeerGroup{}
var order []string
for _, e := range entries {
g, ok := groups[e.cfg.PeerID]
if !ok {
g = &watchPeerGroup{peerID: e.cfg.PeerID, label: e.cfg.PeerLabel}
groups[e.cfg.PeerID] = g
order = append(order, e.cfg.PeerID)
}
g.entries = append(g.entries, e)
}
log.Printf("flit watch starting — id %s…", myID[:16])
log.Printf(" anchor : %s", cfg.SignalURL)
log.Printf(" dirs : %d, peers : %d", len(entries), len(groups))
var wg sync.WaitGroup
for _, e := range entries {
if err := e.reconcile(); err != nil {
fmt.Fprintf(os.Stderr, "flit watch: %s: %v\n", e.cfg.Dir, err)
os.Exit(1)
}
if err := e.save(); err != nil {
log.Printf("warning: could not persist state for %s: %v", e.cfg.Dir, err)
}
}
for _, peerID := range order {
g := groups[peerID]
wg.Add(1)
go func(g *watchPeerGroup) {
defer wg.Done()
runWatchGroup(ctx, tc, myID, g)
}(g)
for _, e := range g.entries {
wg.Add(1)
go func(e *watchEntry, g *watchPeerGroup) {
defer wg.Done()
if err := watchDirLive(ctx, e, g); err != nil {
log.Printf("watch %s: %v", e.cfg.Dir, err)
}
}(e, g)
}
}
wg.Wait()
log.Println("flit watch stopped")
}

1
cli/go.mod
View File

@@ -13,6 +13,7 @@ require (
require (
github.com/BurntSushi/toml v1.6.0 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/fsnotify/fsnotify v1.10.1 // indirect
github.com/google/uuid v1.3.1 // indirect
github.com/pion/datachannel v1.5.8 // indirect
github.com/pion/dtls/v2 v2.2.12 // indirect

2
cli/go.sum
View File

@@ -5,6 +5,8 @@ github.com/BurntSushi/toml v1.6.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/fsnotify/fsnotify v1.10.1 h1:b0/UzAf9yR5rhf3RPm9gf3ehBPpf0oZKIjtpKrx59Ho=
github.com/fsnotify/fsnotify v1.10.1/go.mod h1:TLheqan6HD6GBK6PrDWyDPBaEV8LspOxvPSjC+bVfgo=
github.com/google/uuid v1.3.1 h1:KjJaJ9iWZ3jOFZIf1Lqf4laDRCasjl0BCmnEGxkdLb4=
github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=

27
cli/watch.toml.example Normal file
View File

@@ -0,0 +1,27 @@
# flit watch config — copy to ~/.flit/watch.toml and fill in.
#
# Run with: flit watch
# Sanity-check what would be sent, without connecting to anything:
# flit watch --dry-run
#
# One-way sync only: new/changed files in `dir` are pushed to `peer_id`.
# Deletes are never propagated. The receiving device just needs to be
# running `flit daemon` with this device listed as a trusted peer.
# Required — WebSocket URL of your waste-go anchor
signal_url = "wss://your-anchor.example.com/ws"
# Optional — TURN relay for cross-network transfers
turn_url = "turn:your-anchor.example.com:3478"
turn_secret = "" # coturn use-auth-secret; leave empty to use STUN only
# One [[watch]] block per watched directory. Multiple entries may target
# different peers, or the same peer — entries sharing a peer_id share one
# connection.
[[watch]]
dir = "~/backups/documents"
peer_id = "aabbccddeeff..." # hex Ed25519 pubkey of the destination device
peer_label = "home-server"
pattern = "*.pdf,*.docx" # comma-separated glob include-list; default "*" if omitted
stable_after = "10s" # quiet period after the last write before a file is considered ready