diff --git a/internal/anchor/client.go b/internal/anchor/client.go index f62b9cc..64b84dd 100644 --- a/internal/anchor/client.go +++ b/internal/anchor/client.go @@ -230,6 +230,14 @@ func offer(peerID proto.PeerID, id *crypto.Identity, m *mesh.Mesh, s *sender) (* mesh.WireDataChannel(dc, pc, peerID, id, m) mesh.WireCandidateTrickle(pc, peerID, s) + // Handle file DataChannels opened by the remote peer (they are the file sender). + pc.OnDataChannel(func(dc *webrtc.DataChannel) { + if strings.HasPrefix(dc.Label(), "f:") { + xid := strings.TrimPrefix(dc.Label(), "f:") + m.HandleInboundFileDC(dc, xid, peerID) + } + }) + sdpOffer, err := pc.CreateOffer(nil) if err != nil { pc.Close() @@ -248,8 +256,12 @@ func answer(payload proto.SignalingPayload, fromID proto.PeerID, id *crypto.Iden return nil, err } pc.OnDataChannel(func(dc *webrtc.DataChannel) { - if dc.Label() == "yaw" { + switch { + case dc.Label() == "yaw": mesh.WireDataChannel(dc, pc, fromID, id, m) + case strings.HasPrefix(dc.Label(), "f:"): + xid := strings.TrimPrefix(dc.Label(), "f:") + m.HandleInboundFileDC(dc, xid, fromID) } }) mesh.WireCandidateTrickle(pc, fromID, s) diff --git a/internal/ipc/ipc.go b/internal/ipc/ipc.go index 2a3c7f4..5d6560a 100644 --- a/internal/ipc/ipc.go +++ b/internal/ipc/ipc.go @@ -247,7 +247,22 @@ func handleClient(conn net.Conn, mgr *netmgr.Manager) { }) case proto.CmdSendFile: - send(errMsg("file transfer not yet implemented")) + n := mgr.Resolve(cmd.NetworkID) + if n == nil { + send(errMsg("send_file: not joined to any network")) + continue + } + if cmd.PeerID == nil { + send(errMsg("send_file: peer_id is required")) + continue + } + if cmd.Path == "" { + send(errMsg("send_file: path is required")) + continue + } + if err := n.Mesh.OfferFile(*cmd.PeerID, cmd.Path); err != nil { + send(errMsg(fmt.Sprintf("send_file: %v", err))) + } default: send(errMsg(fmt.Sprintf("unknown command: %s", cmd.Type))) diff --git a/internal/mesh/mesh.go b/internal/mesh/mesh.go index c6e8796..137b288 100644 --- a/internal/mesh/mesh.go +++ b/internal/mesh/mesh.go @@ -6,6 +6,8 @@ import ( "os" "sync" + "github.com/pion/webrtc/v3" + "github.com/waste-go/internal/crypto" "github.com/waste-go/internal/proto" "github.com/waste-go/internal/store" @@ -16,18 +18,26 @@ type PeerConn struct { Info proto.PeerInfo // Send a line of JSON to this peer (pre-encrypted by the sender goroutine). Send chan<- []byte + // PC is the underlying PeerConnection, used to open additional DataChannels. + PC *webrtc.PeerConnection } // Mesh is the shared state of the local node. // All methods are safe to call from multiple goroutines. type Mesh struct { - Identity *crypto.Identity - Store *store.Store // may be nil if persistence is disabled - ShareDir string // directory whose contents are shared with peers; "" = no sharing + Identity *crypto.Identity + Store *store.Store // may be nil if persistence is disabled + ShareDir string // directory whose contents are shared with peers; "" = no sharing + DownloadDir string // directory where received files are saved mu sync.RWMutex peers map[proto.PeerID]*PeerConn + // file transfer state + transferMu sync.Mutex + outbound map[string]*outboundTransfer // xid → pending outbound + inbound map[string]*inboundTransfer // xid → pending inbound + // subscribers receive a copy of every event (fan-out to IPC clients) subMu sync.Mutex subs []chan proto.IpcMessage @@ -40,6 +50,8 @@ func New(id *crypto.Identity, st *store.Store) *Mesh { Identity: id, Store: st, peers: make(map[proto.PeerID]*PeerConn), + outbound: make(map[string]*outboundTransfer), + inbound: make(map[string]*inboundTransfer), } } diff --git a/internal/mesh/peer.go b/internal/mesh/peer.go index b8a77ea..5e6ca36 100644 --- a/internal/mesh/peer.go +++ b/internal/mesh/peer.go @@ -57,6 +57,7 @@ func WireDataChannel( peerConn := &PeerConn{ Info: proto.PeerInfo{ID: peerID, Alias: string(peerID.Short())}, Send: sendCh, + PC: pc, } m.AddPeer(peerConn) @@ -211,6 +212,16 @@ func dispatchPeerMessage(msg proto.PeerMessage, from proto.PeerID, m *Mesh) { PeerID: peerIDPtr(from), Offer: &proto.FileOffer{Xid: msg.Xid, Name: msg.Name, Size: msg.Size, SHA256: msg.SHA256}, }) + m.acceptIncoming(msg, from) + + case proto.MsgFileAccept: + m.startSend(msg.Xid, from) + + case proto.MsgFileCancel: + log.Printf("mesh: file-cancel from %s xid=%s reason=%s", from.Short(), msg.Xid, msg.Reason) + + case proto.MsgFileDone: + log.Printf("mesh: file-done from %s xid=%s", from.Short(), msg.Xid) case proto.MsgPeerGossip: if msg.Gossip != nil { diff --git a/internal/mesh/transfer.go b/internal/mesh/transfer.go new file mode 100644 index 0000000..fd3f97d --- /dev/null +++ b/internal/mesh/transfer.go @@ -0,0 +1,351 @@ +package mesh + +import ( + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "hash" + "io" + "log" + "os" + "path/filepath" + "strings" + "sync" + + "github.com/pion/webrtc/v3" + + "github.com/waste-go/internal/proto" +) + +const ( + fileChunkSize = 64 * 1024 // 64 KiB per chunk + fileBufferHighWater = 512 * 1024 // pause sending above this + fileBufferLowWater = 256 * 1024 // resume when it drops here +) + +type outboundTransfer struct { + peerID proto.PeerID + path string + size int64 + sha256 string +} + +type inboundTransfer struct { + from proto.PeerID + name string + size int64 + sha256 string + mu sync.Mutex + tmp *os.File + hasher hash.Hash + written int64 +} + +// OfferFile reads filename from ShareDir, computes its SHA-256, and sends a +// file-offer to peerID over the existing "yaw" DataChannel. +func (m *Mesh) OfferFile(peerID proto.PeerID, filename string) error { + if m.ShareDir == "" { + return fmt.Errorf("no share directory configured") + } + if strings.ContainsAny(filename, "/\\") || filename == ".." { + return fmt.Errorf("invalid filename %q", filename) + } + path := filepath.Join(m.ShareDir, filename) + + f, err := os.Open(path) + if err != nil { + return fmt.Errorf("open %q: %w", filename, err) + } + defer f.Close() + + info, err := f.Stat() + if err != nil { + return err + } + + h := sha256.New() + if _, err := io.Copy(h, f); err != nil { + return fmt.Errorf("hash %q: %w", filename, err) + } + shaHex := hex.EncodeToString(h.Sum(nil)) + xid := newXid() + + m.transferMu.Lock() + m.outbound[xid] = &outboundTransfer{ + peerID: peerID, + path: path, + size: info.Size(), + sha256: shaHex, + } + m.transferMu.Unlock() + + wire, err := json.Marshal(proto.PeerMessage{ + Type: proto.MsgFileOffer, + Xid: xid, + Name: filename, + Size: info.Size(), + SHA256: shaHex, + }) + if err != nil { + return err + } + if !m.SendTo(peerID, wire) { + m.transferMu.Lock() + delete(m.outbound, xid) + m.transferMu.Unlock() + return fmt.Errorf("peer %s not connected", peerID.Short()) + } + log.Printf("transfer: offered %s (%d bytes) to %s xid=%s", filename, info.Size(), peerID.Short(), xid[:8]) + return nil +} + +// acceptIncoming is called when we receive file-offer from a peer. +// It stores the inbound state and immediately sends file-accept (auto-accept). +func (m *Mesh) acceptIncoming(msg proto.PeerMessage, from proto.PeerID) { + m.transferMu.Lock() + m.inbound[msg.Xid] = &inboundTransfer{ + from: from, + name: msg.Name, + size: msg.Size, + sha256: msg.SHA256, + } + m.transferMu.Unlock() + + accept, _ := json.Marshal(proto.PeerMessage{ + Type: proto.MsgFileAccept, + Xid: msg.Xid, + }) + 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]) +} + +// startSend opens the binary DataChannel "f:" and streams the file. +// Called when we receive file-accept from the peer. +func (m *Mesh) startSend(xid string, from proto.PeerID) { + m.transferMu.Lock() + t, ok := m.outbound[xid] + m.transferMu.Unlock() + if !ok { + log.Printf("transfer: file-accept for unknown xid %s", xid[:8]) + return + } + if t.peerID != from { + log.Printf("transfer: file-accept from unexpected peer %s", from.Short()) + return + } + + m.mu.RLock() + conn, ok := m.peers[from] + m.mu.RUnlock() + if !ok { + log.Printf("transfer: peer %s not connected for xid %s", from.Short(), xid[:8]) + return + } + + ordered := true + dc, err := conn.PC.CreateDataChannel("f:"+xid, &webrtc.DataChannelInit{Ordered: &ordered}) + if err != nil { + log.Printf("transfer: create file DC xid=%s: %v", xid[:8], err) + return + } + + // Wait for receiver to signal "ok" before sending — this ensures the receiver + // has its OnMessage/OnClose handlers registered before any data arrives. + dc.OnOpen(func() { + dc.OnMessage(func(msg webrtc.DataChannelMessage) { + if msg.IsString && string(msg.Data) == "ok" { + go m.sendFileChunks(dc, t, xid) + } + }) + }) +} + +func (m *Mesh) sendFileChunks(dc *webrtc.DataChannel, t *outboundTransfer, xid string) { + defer func() { + m.transferMu.Lock() + delete(m.outbound, xid) + m.transferMu.Unlock() + }() + + f, err := os.Open(t.path) + if err != nil { + log.Printf("transfer: open file xid=%s: %v", xid[:8], err) + dc.Close() + return + } + defer f.Close() + + // Backpressure: block when the DC buffer is full. + resume := make(chan struct{}, 1) + dc.SetBufferedAmountLowThreshold(fileBufferLowWater) + dc.OnBufferedAmountLow(func() { + select { + case resume <- struct{}{}: + default: + } + }) + + buf := make([]byte, fileChunkSize) + var sent int64 + + for { + n, readErr := f.Read(buf) + if n > 0 { + for dc.BufferedAmount() > fileBufferHighWater { + <-resume + } + if err := dc.Send(buf[:n]); err != nil { + log.Printf("transfer: send chunk xid=%s: %v", xid[:8], err) + return + } + sent += int64(n) + m.Emit(proto.IpcMessage{ + Type: proto.EvtFileProgress, + TransferID: xid, + BytesReceived: sent, + TotalBytes: t.size, + }) + } + if readErr == io.EOF { + break + } + if readErr != nil { + log.Printf("transfer: read file xid=%s: %v", xid[:8], readErr) + return + } + } + + dc.Close() + + done, _ := json.Marshal(proto.PeerMessage{ + Type: proto.MsgFileDone, + Xid: xid, + SHA256: t.sha256, + }) + m.SendTo(t.peerID, done) + log.Printf("transfer: sent %s (%d bytes) xid=%s", filepath.Base(t.path), sent, xid[:8]) +} + +// HandleInboundFileDC is called from the anchor when a "f:" DataChannel +// arrives on a PeerConnection. It writes chunks to a temp file, verifies the +// SHA-256 on close, and emits EvtFileComplete. +func (m *Mesh) HandleInboundFileDC(dc *webrtc.DataChannel, xid string, from proto.PeerID) { + m.transferMu.Lock() + t, ok := m.inbound[xid] + m.transferMu.Unlock() + if !ok { + log.Printf("transfer: no pending inbound for xid %s", xid[:8]) + dc.Close() + return + } + + var once sync.Once + doOpen := func() { + if err := os.MkdirAll(m.DownloadDir, 0o755); err != nil { + 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]) + // 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 + } + + // Register OnMessage first so pion queues any early binary chunks. + dc.OnMessage(func(msg webrtc.DataChannelMessage) { + if msg.IsString { + return // "ok" echo or any other text; ignore + } + t.mu.Lock() + tmp, h := t.tmp, t.hasher + t.mu.Unlock() + if tmp == nil { + return + } + if _, err := tmp.Write(msg.Data); err != nil { + log.Printf("transfer: write xid=%s: %v", xid[:8], err) + return + } + h.Write(msg.Data) + t.mu.Lock() + t.written += int64(len(msg.Data)) + written := t.written + t.mu.Unlock() + m.Emit(proto.IpcMessage{ + Type: proto.EvtFileProgress, + TransferID: xid, + BytesReceived: written, + TotalBytes: t.size, + }) + }) + + dc.OnOpen(func() { once.Do(doOpen) }) + if dc.ReadyState() == webrtc.DataChannelStateOpen { + once.Do(doOpen) + } + + dc.OnClose(func() { + t.mu.Lock() + tmp := t.tmp + actualSha := "" + if t.hasher != nil { + actualSha = hex.EncodeToString(t.hasher.Sum(nil)) + } + name, expectedSha := t.name, t.sha256 + t.mu.Unlock() + + m.transferMu.Lock() + delete(m.inbound, xid) + m.transferMu.Unlock() + + if tmp == nil { + return + } + tmp.Close() + + if actualSha != expectedSha { + os.Remove(tmp.Name()) + log.Printf("transfer: sha256 mismatch xid=%s got=%s want=%s", xid[:8], actualSha[:8], expectedSha[:8]) + m.Emit(proto.IpcMessage{ + Type: proto.EvtError, + TransferID: xid, + ErrorMessage: fmt.Sprintf("file %s: sha256 mismatch", name), + }) + return + } + + 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 { + log.Printf("transfer: rename xid=%s: %v", xid[:8], err) + return + } + log.Printf("transfer: saved %s -> %s", name, final) + m.Emit(proto.IpcMessage{ + Type: proto.EvtFileComplete, + TransferID: xid, + Path: final, + }) + }) +} + +func newXid() string { + b := make([]byte, 16) + rand.Read(b) //nolint:errcheck + return hex.EncodeToString(b) +} diff --git a/internal/netmgr/manager.go b/internal/netmgr/manager.go index 33b0fff..57051c2 100644 --- a/internal/netmgr/manager.go +++ b/internal/netmgr/manager.go @@ -92,6 +92,7 @@ func (mgr *Manager) Join(name string) (string, error) { if mgr.cfg.ShareDir != "" { m.ShareDir = mgr.cfg.ShareDir } + m.DownloadDir = filepath.Join(mgr.cfg.StoreDir, "downloads-"+netID_full) // Forward all mesh events to the Manager's fan-out, tagging with network_id. meshEvents := m.Subscribe() diff --git a/internal/proto/proto.go b/internal/proto/proto.go index 3a670d2..d21dd64 100644 --- a/internal/proto/proto.go +++ b/internal/proto/proto.go @@ -225,6 +225,7 @@ const ( EvtError IpcMsgType = "error" EvtInviteGenerated IpcMsgType = "invite_generated" EvtFileList IpcMsgType = "file_list" + EvtFileComplete IpcMsgType = "file_complete" EvtNetworkJoined IpcMsgType = "network_joined" EvtNetworkLeft IpcMsgType = "network_left" ) diff --git a/test-network.sh b/test-network.sh index a4c4eae..975974f 100755 --- a/test-network.sh +++ b/test-network.sh @@ -114,6 +114,26 @@ pretty() { echo -e "${color}${BOLD}[${label}]${RESET} ${BOLD}💬 #${room}${RESET} ${DIM}<${from}…>${RESET} ${body}" fi ;; + incoming_file) + local name size xid + name=$(echo "$line" | jq -r '.offer.name // "?"' 2>/dev/null) + size=$(echo "$line" | jq -r '.offer.size // 0' 2>/dev/null) + xid=$(echo "$line" | jq -r '.offer.xid[:8] // "?"' 2>/dev/null) + echo -e "${color}${BOLD}[${label}]${RESET} ${YELLOW}↓ incoming_file${RESET} ${BOLD}${name}${RESET} (${size}B) xid=${DIM}${xid}…${RESET}" + ;; + file_progress) + local xid rx total + xid=$(echo "$line" | jq -r '.transfer_id[:8] // "?"' 2>/dev/null) + rx=$(echo "$line" | jq -r '.bytes_received // 0' 2>/dev/null) + total=$(echo "$line" | jq -r '.total_bytes // 0' 2>/dev/null) + echo -e "${color}${BOLD}[${label}]${RESET} ${DIM}file_progress${RESET} xid=${xid}… ${rx}/${total}B" + ;; + file_complete) + local xid path + xid=$(echo "$line" | jq -r '.transfer_id[:8] // "?"' 2>/dev/null) + path=$(echo "$line" | jq -r '.path // "?"' 2>/dev/null) + echo -e "${color}${BOLD}[${label}]${RESET} ${GREEN}✓ file_complete${RESET} xid=${DIM}${xid}…${RESET} → ${BOLD}${path}${RESET}" + ;; error) local msg msg=$(echo "$line" | jq -r '.error_message // .' 2>/dev/null) @@ -364,6 +384,40 @@ for peer_ipc in "$ALICE_IPC" "$BOB_IPC" "$CHARLIE_IPC"; do done sleep 0.5 +# ── file transfer ──────────────────────────────────────────────────────────── +echo "" +echo -e "${DIM}────────────────────────────────────────────────────────${RESET}" +echo -e "file transfer (alice → bob: notes.txt)" +echo -e "${DIM}────────────────────────────────────────────────────────${RESET}" + +ipc "$ALICE_IPC" "$(jq -cn --arg peer "$BOB_ID" --arg path "notes.txt" \ + '{"type":"send_file","peer_id":$peer,"path":$path}')" + +# Wait up to 10s for bob to receive the file (glob over downloads-* dir) +received=0 +for i in $(seq 1 50); do + sleep 0.2 + if ls "$DATA_ROOT/bob/downloads-"*/notes.txt 2>/dev/null | grep -q .; then + received=1; break + fi +done + +if [ "$received" = "1" ]; then + rx_file=$(ls "$DATA_ROOT/bob/downloads-"*/notes.txt 2>/dev/null | head -1) + orig_sha=$(sha256sum "$DATA_ROOT/alice/share/notes.txt" | awk '{print $1}') + rx_sha=$(sha256sum "$rx_file" | awk '{print $1}') + if [ "$orig_sha" = "$rx_sha" ]; then + echo -e " ${GREEN}✓ notes.txt received by bob, sha256 matches${RESET}" + else + echo -e " ${RED}✗ notes.txt received but sha256 mismatch!${RESET}" + echo -e " orig: $orig_sha" + echo -e " got: $rx_sha" + fi +else + echo -e " ${RED}✗ notes.txt not received by bob within 10s${RESET}" +fi +sleep 0.5 + # ── leave ───────────────────────────────────────────────────────────────────── echo "" echo -e "${DIM}────────────────────────────────────────────────────────${RESET}"