explewd/waste-go
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

feat: download dirs, file transfer resume, Wails desktop app, PWA, CI

Browse Source 
Daemon:
- Per-network download directories (-download-dir flag, set_download_dir IPC)
- File transfer resume after disconnection: .tmp.meta sidecars survive
  interruption; resume_offset in file-accept lets sender seek and continue
- set_download_dir IPC command; download_dir reported in state_snapshot

Protocol:
- PeerMessage.ResumeOffset (EXT-006) for file transfer resume
- IpcMessage.ShareNetworks json tag changed from "networks" to "network_ids"
  to fix duplicate json tag collision with Networks []NetworkInfo

Desktop app (cmd/app):
- Wails v2 shell embedding daemon logic directly (no subprocess)
- System tray on Linux/Windows via getlantern/systray; macOS hides to Dock
- OS notifications for message_received and file_complete via Wails events
- notray build tag for headless/CI builds without GTK tray headers
- build-app.sh: builds web frontend, copies dist, runs wails build

Web / PWA:
- manifest.json + Apple touch icon meta tags for mobile "Add to Home Screen"
- PNG icons (192px, 512px, 180px) generated from SVG
- Wails EventsOn("notify") hook in App.tsx for native OS notifications

CI:
- .gitea/workflows/build.yml: server binaries cross-compiled for 5 platforms,
  desktop app for Linux amd64, release artifacts published on v* tags

Docs:
- README: download dir, file transfer resume, desktop app, PWA, CI sections
- EXTENSIONS.md: EXT-004 daemon mode marked shipped; EXT-006 resume added
- FUTURE.md: roadmap updated

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Fredrik Johansson
2026-06-28 21:38:28 +02:00
parent d233f4d79e
commit 0789cf8840
29 changed files with 951 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
internal/mesh/peer.go
View File

@@ -267,7 +267,7 @@ func dispatchPeerMessage(msg proto.PeerMessage, from proto.PeerID, m *Mesh) {
m.acceptIncoming(msg, from)
case proto.MsgFileAccept:
m.startSend(msg.Xid, from)
m.startSend(msg.Xid, from, msg.ResumeOffset)
case proto.MsgFileCancel:
log.Printf("mesh: file-cancel from %s xid=%s reason=%s", from.Short(), msg.Xid, msg.Reason)

188
internal/mesh/transfer.go
View File

@@ -33,14 +33,62 @@ type outboundTransfer struct {
}
type inboundTransfer struct {
from proto.PeerID
name string
size int64
sha256 string
mu sync.Mutex
tmp *os.File
hasher hash.Hash
written int64
from proto.PeerID
name string
size int64
sha256 string
mu sync.Mutex
tmp *os.File
hasher hash.Hash
written int64 // total bytes received; starts at resumeOffset when resuming
resumePath string // path to existing .tmp when resuming; empty for new transfers
resumeOffset int64 // bytes already present in resumePath
metaPath string // path to the .tmp.meta sidecar
}
// partialMeta is written as a JSON sidecar alongside each in-progress .tmp file.
// It survives interruptions so the receiver can resume later.
type partialMeta struct {
Name string `json:"name"`
SHA256 string `json:"sha256"`
From string `json:"from"`
Size int64 `json:"size"`
}
// findPartial scans dir for a .tmp.meta sidecar whose sha256 matches.
// Returns (tmpPath, metaPath, offset) — all empty/zero if no match.
func findPartial(dir, sha256hex string) (tmpPath, metaPath string, offset int64) {
metas, _ := filepath.Glob(filepath.Join(dir, "*.tmp.meta"))
for _, mp := range metas {
data, err := os.ReadFile(mp)
if err != nil {
continue
}
var m partialMeta
if err := json.Unmarshal(data, &m); err != nil {
continue
}
if m.SHA256 != sha256hex {
continue
}
tp := strings.TrimSuffix(mp, ".meta")
info, err := os.Stat(tp)
if err != nil {
continue
}
return tp, mp, info.Size()
}
return "", "", 0
}
func writePartialMeta(path string, t *inboundTransfer) {
data, _ := json.Marshal(partialMeta{
Name: t.name,
SHA256: t.sha256,
From: string(t.from),
Size: t.size,
})
os.WriteFile(path, data, 0o644) //nolint:errcheck
}
// OfferFile reads filename from ShareDir, computes its SHA-256, and sends a
@@ -102,28 +150,49 @@ func (m *Mesh) OfferFile(peerID proto.PeerID, filename string) error {
}
// acceptIncoming is called when we receive file-offer from a peer.
// It stores the inbound state and immediately sends file-accept (auto-accept).
// Checks for a resumable partial download matching the offer's sha256.
// Sends file-accept immediately, with a non-zero resume_offset when resuming.
func (m *Mesh) acceptIncoming(msg proto.PeerMessage, from proto.PeerID) {
m.transferMu.Lock()
m.inbound[msg.Xid] = &inboundTransfer{
t := &inboundTransfer{
from: from,
name: msg.Name,
size: msg.Size,
sha256: msg.SHA256,
}
var resumeOffset int64
if m.DownloadDir != "" {
if tp, mp, off := findPartial(m.DownloadDir, msg.SHA256); tp != "" {
t.resumePath = tp
t.metaPath = mp
t.resumeOffset = off
t.written = off
resumeOffset = off
log.Printf("transfer: found partial for %s at %s (%d/%d bytes)", msg.Name, tp, off, msg.Size)
}
}
m.transferMu.Lock()
m.inbound[msg.Xid] = t
m.transferMu.Unlock()
accept, _ := json.Marshal(proto.PeerMessage{
Type: proto.MsgFileAccept,
Xid: msg.Xid,
Type: proto.MsgFileAccept,
Xid: msg.Xid,
ResumeOffset: resumeOffset,
})
m.SendTo(from, accept)
log.Printf("transfer: auto-accepted %s (%d bytes) from %s xid=%s", msg.Name, msg.Size, from.Short(), msg.Xid[:8])
if resumeOffset > 0 {
log.Printf("transfer: resuming %s from byte %d xid=%s", msg.Name, resumeOffset, msg.Xid[:8])
} else {
log.Printf("transfer: auto-accepted %s (%d bytes) from %s xid=%s", msg.Name, msg.Size, from.Short(), msg.Xid[:8])
}
}
// startSend opens the binary DataChannel "f:<xid>" and streams the file.
// Called when we receive file-accept from the peer.
func (m *Mesh) startSend(xid string, from proto.PeerID) {
// Called when we receive file-accept from the peer. resumeOffset is non-zero
// when the receiver is resuming a previous partial download.
func (m *Mesh) startSend(xid string, from proto.PeerID, resumeOffset int64) {
m.transferMu.Lock()
t, ok := m.outbound[xid]
m.transferMu.Unlock()
@@ -156,13 +225,13 @@ func (m *Mesh) startSend(xid string, from proto.PeerID) {
dc.OnOpen(func() {
dc.OnMessage(func(msg webrtc.DataChannelMessage) {
if msg.IsString && string(msg.Data) == "ok" {
go m.sendFileChunks(dc, t, xid)
go m.sendFileChunks(dc, t, xid, resumeOffset)
}
})
})
}
func (m *Mesh) sendFileChunks(dc *webrtc.DataChannel, t *outboundTransfer, xid string) {
func (m *Mesh) sendFileChunks(dc *webrtc.DataChannel, t *outboundTransfer, xid string, offset int64) {
defer func() {
m.transferMu.Lock()
delete(m.outbound, xid)
@@ -177,6 +246,15 @@ func (m *Mesh) sendFileChunks(dc *webrtc.DataChannel, t *outboundTransfer, xid s
}
defer f.Close()
if offset > 0 {
if _, err := f.Seek(offset, io.SeekStart); err != nil {
log.Printf("transfer: seek to %d xid=%s: %v", offset, xid[:8], err)
dc.Close()
return
}
log.Printf("transfer: resuming send from byte %d xid=%s", offset, xid[:8])
}
// Backpressure: block when the DC buffer is full.
resume := make(chan struct{}, 1)
dc.SetBufferedAmountLowThreshold(fileBufferLowWater)
@@ -188,7 +266,7 @@ func (m *Mesh) sendFileChunks(dc *webrtc.DataChannel, t *outboundTransfer, xid s
})
buf := make([]byte, fileChunkSize)
var sent int64
sent := offset // start progress reporting from where the receiver left off
for {
n, readErr := f.Read(buf)
@@ -229,8 +307,9 @@ func (m *Mesh) sendFileChunks(dc *webrtc.DataChannel, t *outboundTransfer, xid s
}
// HandleInboundFileDC is called from the anchor when a "f:<xid>" DataChannel
// arrives on a PeerConnection. It writes chunks to a temp file, verifies the
// SHA-256 on close, and emits EvtFileComplete.
// arrives on a PeerConnection. It writes chunks to a temp file (or resumes an
// existing partial), verifies the SHA-256 on close, and emits EvtFileComplete.
// Interrupted transfers keep their .tmp and .meta files for future resume.
func (m *Mesh) HandleInboundFileDC(dc *webrtc.DataChannel, xid string, from proto.PeerID) {
m.transferMu.Lock()
t, ok := m.inbound[xid]
@@ -247,16 +326,43 @@ func (m *Mesh) HandleInboundFileDC(dc *webrtc.DataChannel, xid string, from prot
log.Printf("transfer: mkdir %s: %v", m.DownloadDir, err)
return
}
tmp, err := os.CreateTemp(m.DownloadDir, "dl-*.tmp")
if err != nil {
log.Printf("transfer: create temp file: %v", err)
return
}
t.mu.Lock()
t.tmp = tmp
t.hasher = sha256.New()
t.mu.Unlock()
log.Printf("transfer: receiving %s xid=%s", t.name, xid[:8])
defer t.mu.Unlock()
if t.resumePath != "" {
// Resume: open existing .tmp in read-write mode, seek to end for appending.
f, err := os.OpenFile(t.resumePath, os.O_RDWR, 0o644)
if err != nil {
log.Printf("transfer: open resume file %s: %v", t.resumePath, err)
return
}
// Restore the hasher state by re-hashing the bytes already on disk.
h := sha256.New()
if _, err := f.Seek(0, io.SeekStart); err == nil {
io.Copy(h, f) //nolint:errcheck
}
if _, err := f.Seek(0, io.SeekEnd); err != nil {
log.Printf("transfer: seek to end xid=%s: %v", xid[:8], err)
f.Close()
return
}
t.tmp = f
t.hasher = h
log.Printf("transfer: resuming inbound %s at byte %d xid=%s", t.name, t.resumeOffset, xid[:8])
} else {
tmp, err := os.CreateTemp(m.DownloadDir, "dl-*.tmp")
if err != nil {
log.Printf("transfer: create temp file: %v", err)
return
}
t.tmp = tmp
t.hasher = sha256.New()
t.metaPath = tmp.Name() + ".meta"
writePartialMeta(t.metaPath, t)
log.Printf("transfer: receiving %s xid=%s", t.name, xid[:8])
}
// Signal sender that we are ready — it will not start streaming until
// it receives this, guaranteeing our OnMessage/OnClose are registered first.
dc.SendText("ok") //nolint:errcheck
@@ -298,11 +404,12 @@ func (m *Mesh) HandleInboundFileDC(dc *webrtc.DataChannel, xid string, from prot
dc.OnClose(func() {
t.mu.Lock()
tmp := t.tmp
metaPath := t.metaPath
actualSha := ""
if t.hasher != nil {
actualSha = hex.EncodeToString(t.hasher.Sum(nil))
}
name, expectedSha := t.name, t.sha256
name, expectedSha, size, written := t.name, t.sha256, t.size, t.written
t.mu.Unlock()
m.transferMu.Lock()
@@ -312,10 +419,21 @@ func (m *Mesh) HandleInboundFileDC(dc *webrtc.DataChannel, xid string, from prot
if tmp == nil {
return
}
tmpName := tmp.Name()
tmp.Close()
// Incomplete — keep .tmp and .meta for future resume.
if written < size {
log.Printf("transfer: interrupted %s at %d/%d bytes xid=%s (resumable)", name, written, size, xid[:8])
return
}
// Full receive — verify integrity.
if actualSha != expectedSha {
os.Remove(tmp.Name())
os.Remove(tmpName)
if metaPath != "" {
os.Remove(metaPath)
}
log.Printf("transfer: sha256 mismatch xid=%s got=%s want=%s", xid[:8], actualSha[:8], expectedSha[:8])
m.Emit(proto.IpcMessage{
Type: proto.EvtError,
@@ -325,13 +443,17 @@ func (m *Mesh) HandleInboundFileDC(dc *webrtc.DataChannel, xid string, from prot
return
}
// Success — remove sidecar and move to final path.
if metaPath != "" {
os.Remove(metaPath)
}
final := filepath.Join(m.DownloadDir, name)
if _, err := os.Stat(final); err == nil {
ext := filepath.Ext(name)
base := strings.TrimSuffix(name, ext)
final = filepath.Join(m.DownloadDir, base+"-"+xid[:8]+ext)
}
if err := os.Rename(tmp.Name(), final); err != nil {
if err := os.Rename(tmpName, final); err != nil {
log.Printf("transfer: rename xid=%s: %v", xid[:8], err)
return
}