diff --git a/client/android/client.go b/client/android/client.go index 16daef98a..daf70a8c4 100644 --- a/client/android/client.go +++ b/client/android/client.go @@ -113,8 +113,9 @@ type Client struct { // The file drop handle survives engine restarts so the UI keeps one listener // registration and one history view across reconnects. See fileDropFor. - fileDropMu sync.Mutex - fileDrop *FileDrop + fileDropMu sync.Mutex + fileDrop *FileDrop + fileDropSink FileDropSink } func (c *Client) setState(cfg *profilemanager.Config, cacheDir string, cfgPath string, cc *internal.ConnectClient) { diff --git a/client/android/client_filedrop.go b/client/android/client_filedrop.go index 8576bd6eb..319914a3a 100644 --- a/client/android/client_filedrop.go +++ b/client/android/client_filedrop.go @@ -10,6 +10,15 @@ import ( "github.com/netbirdio/netbird/client/internal" ) +// SetFileDropSink installs the platform sink incoming payloads are staged +// through. It takes effect on the next handle the client opens, so the platform +// sets it before asking for one. +func (c *Client) SetFileDropSink(sink FileDropSink) { + c.fileDropMu.Lock() + defer c.fileDropMu.Unlock() + c.fileDropSink = sink +} + // FileDrop returns the handle of the active profile, creating it on first use. // The UI calls this to list transfers and change settings while disconnected. func (c *Client) FileDrop(configDir string) (*FileDrop, error) { @@ -32,7 +41,7 @@ func (c *Client) fileDropFor(configDir, profileID string) (*FileDrop, error) { return fd, nil } - fd, err := NewFileDrop(configDir, profileID) + fd, err := NewFileDrop(configDir, profileID, c.fileDropSink) if err != nil { c.fileDropMu.Unlock() return nil, err diff --git a/client/android/filedrop.go b/client/android/filedrop.go index daa4f8ada..ec3294093 100644 --- a/client/android/filedrop.go +++ b/client/android/filedrop.go @@ -28,8 +28,10 @@ type FileDrop struct { listener FileDropListener } -// NewFileDrop opens the file drop state of the given profile. -func NewFileDrop(configDir, profileID string) (*FileDrop, error) { +// NewFileDrop opens the file drop state of the given profile. A nil sink leaves +// payloads staged and delivered on the filesystem, under the destination +// directory the policy names. +func NewFileDrop(configDir, profileID string, sink FileDropSink) (*FileDrop, error) { if configDir == "" || profileID == "" { return nil, errors.New("file drop requires a config dir and profile ID") } @@ -39,12 +41,21 @@ func NewFileDrop(configDir, profileID string) (*FileDrop, error) { return nil, err } + var platformSink filedrop.Sink + if sink != nil { + platformSink, err = filedrop.NewPlatformSink(newPlatformSinkAdapter(sink)) + if err != nil { + return nil, fmt.Errorf("wrap file drop sink: %w", err) + } + } + fd := &FileDrop{configDir: configDir, profileID: profileID} manager, err := filedrop.NewManager(filedrop.ManagerConfig{ Profile: profilemanager.ID(profileID), DataDir: filepath.Join(configDir, filedropDataSubdir, profileID), Store: filedrop.NewProfileStore(prefs.prefs), Events: fd.publish, + Sink: platformSink, }) if err != nil { return nil, fmt.Errorf("create file drop manager: %w", err) diff --git a/client/android/filedrop_sink.go b/client/android/filedrop_sink.go new file mode 100644 index 000000000..b1df245fb --- /dev/null +++ b/client/android/filedrop_sink.go @@ -0,0 +1,83 @@ +//go:build android + +package android + +import ( + "fmt" + + "github.com/netbirdio/netbird/client/internal/filedrop" +) + +// FileDropSink stages incoming payloads on the platform's behalf. Android +// cannot address the user's shared storage by path, so the writing itself +// happens in Java and the engine only drives it. +// +// The methods mirror filedrop.PlatformSink; see that interface for the contract. +type FileDropSink interface { + // DestinationLabel names where payloads land, for the UI to show in place of + // a path the platform does not have. + DestinationLabel() string + Prepare(offerID string) error + Received(offerID string, index int) (int64, error) + OpenWriter(offerID string, index int, name string, offset int64, size int64) (FileDropWriter, error) + Deliver(offerID string) (string, error) + Remove(offerID string) + Cleanup(maxAgeSeconds int64) +} + +// FileDropWriter is one payload's destination, opened by a FileDropSink. +// +// WriteChunk takes the bytes rather than filling a caller-supplied buffer, and +// the total is read back through Written: gomobile copies a []byte argument +// into a fresh Java array and never carries a written count back out. +type FileDropWriter interface { + WriteChunk(p []byte) error + Close() error + Written() int64 +} + +// platformSinkAdapter bridges the gomobile-bound FileDropSink onto the +// interface the engine consumes. The two differ only in the writer type, which +// gomobile cannot share between packages. +type platformSinkAdapter struct { + sink FileDropSink +} + +func newPlatformSinkAdapter(sink FileDropSink) filedrop.PlatformSink { + return &platformSinkAdapter{sink: sink} +} + +func (a *platformSinkAdapter) DestinationLabel() string { + return a.sink.DestinationLabel() +} + +func (a *platformSinkAdapter) Prepare(offerID string) error { + return a.sink.Prepare(offerID) +} + +func (a *platformSinkAdapter) Received(offerID string, index int) (int64, error) { + return a.sink.Received(offerID, index) +} + +func (a *platformSinkAdapter) OpenWriter(offerID string, index int, name string, offset int64, size int64) (filedrop.PlatformWriter, error) { + w, err := a.sink.OpenWriter(offerID, index, name, offset, size) + if err != nil { + return nil, err + } + if w == nil { + return nil, fmt.Errorf("no destination for %s item %d", offerID, index) + } + return w, nil +} + +func (a *platformSinkAdapter) Deliver(offerID string) (string, error) { + return a.sink.Deliver(offerID) +} + +func (a *platformSinkAdapter) Remove(offerID string) { + a.sink.Remove(offerID) +} + +func (a *platformSinkAdapter) Cleanup(maxAgeSeconds int64) { + a.sink.Cleanup(maxAgeSeconds) +} diff --git a/client/internal/filedrop/filedrop_test.go b/client/internal/filedrop/filedrop_test.go index 710b75a7c..3419d1724 100644 --- a/client/internal/filedrop/filedrop_test.go +++ b/client/internal/filedrop/filedrop_test.go @@ -166,7 +166,7 @@ func TestAutoAcceptTransfersPayload(t *testing.T) { assert.Equal(t, int64(len(content)), lastSent, "progress must reach the full payload size") - staged, err := os.ReadFile(srv.Spool().Path(id, 0)) + staged, err := os.ReadFile(srv.FileSpool().Path(id, 0)) require.NoError(t, err) assert.Equal(t, content, staged, "staged bytes should match what was sent") @@ -200,7 +200,7 @@ func TestAskModeAcceptReleasesUpload(t *testing.T) { id, err := client.Send(context.Background(), testAddr, []Payload{payload}, nil) require.NoError(t, err) - staged, err := os.ReadFile(srv.Spool().Path(id, 0)) + staged, err := os.ReadFile(srv.FileSpool().Path(id, 0)) require.NoError(t, err) assert.Equal(t, content, staged, "payload should arrive after acceptance") @@ -226,7 +226,7 @@ func TestAskModeDeclineKeepsPayloadOut(t *testing.T) { id, err := client.Send(context.Background(), testAddr, []Payload{filePayload(t, "x.bin", []byte("data"))}, nil) require.ErrorIs(t, err, ErrDeclined, "sender must see the decline") - _, statErr := os.Stat(srv.Spool().Path(id, 0)) + _, statErr := os.Stat(srv.FileSpool().Path(id, 0)) assert.True(t, os.IsNotExist(statErr), "declined payload must never be staged") } @@ -257,7 +257,7 @@ func TestUploadResumesFromConfirmedOffset(t *testing.T) { require.NoError(t, srv.Spool().Prepare(offer.ID)) half := int64(len(content) / 2) - _, err := srv.Spool().Write(offer.ID, 0, 0, strings.NewReader(string(content[:half])), half) + _, err := srv.Spool().Write(offer.ID, 0, "", 0, strings.NewReader(string(content[:half])), half) require.NoError(t, err) srv.mu.RLock() @@ -270,7 +270,7 @@ func TestUploadResumesFromConfirmedOffset(t *testing.T) { require.NoError(t, client.putFile(context.Background(), base, offer.ID, 0, payload, confirmed, nil)) - staged, err := os.ReadFile(srv.Spool().Path(offer.ID, 0)) + staged, err := os.ReadFile(srv.FileSpool().Path(offer.ID, 0)) require.NoError(t, err) assert.Equal(t, content, staged, "resumed upload must reconstruct the full payload") } @@ -302,7 +302,7 @@ func TestUploadIsBoundedByAnnouncedSize(t *testing.T) { _, err = io.ReadAll(conn) require.NoError(t, err) - staged, err := os.ReadFile(srv.Spool().Path(offer.ID, 0)) + staged, err := os.ReadFile(srv.FileSpool().Path(offer.ID, 0)) require.NoError(t, err) assert.Len(t, staged, int(announced), "staged size must be capped at the announced size") } @@ -349,7 +349,7 @@ func TestTextPayloadStaysInline(t *testing.T) { assert.Equal(t, "hello peer", offer.Files[0].Text, "text must arrive in the offer itself") assert.Equal(t, StateCompleted, offer.State, "a text-only offer completes without an upload") - _, statErr := os.Stat(srv.Spool().Path(id, 0)) + _, statErr := os.Stat(srv.FileSpool().Path(id, 0)) assert.True(t, os.IsNotExist(statErr), "text payloads must not be written to the spool") } @@ -583,10 +583,10 @@ func TestSpoolWriteTruncatesStaleTail(t *testing.T) { id := OfferID("offer") require.NoError(t, spool.Prepare(id)) - _, err = spool.Write(id, 0, 0, strings.NewReader("AAAAAAAAAA"), 10) + _, err = spool.Write(id, 0, "", 0, strings.NewReader("AAAAAAAAAA"), 10) require.NoError(t, err) - total, err := spool.Write(id, 0, 2, strings.NewReader("BB"), 10) + total, err := spool.Write(id, 0, "", 2, strings.NewReader("BB"), 10) require.NoError(t, err) assert.Equal(t, int64(4), total, "staged size follows the resumed write") diff --git a/client/internal/filedrop/manager.go b/client/internal/filedrop/manager.go index ae1bfbddb..b3bcef9d2 100644 --- a/client/internal/filedrop/manager.go +++ b/client/internal/filedrop/manager.go @@ -50,6 +50,7 @@ type ManagerConfig struct { Store Store Events EventSink OfferTTL time.Duration + Sink Sink } type sendHandle struct { @@ -69,6 +70,7 @@ type Manager struct { history *History events EventSink offerTTL time.Duration + sink Sink server *Server ports *PortRegistry @@ -95,6 +97,7 @@ func NewManager(cfg ManagerConfig) (*Manager, error) { history: LoadHistory(cfg.Store), events: cfg.Events, offerTTL: cfg.OfferTTL, + sink: cfg.Sink, ports: NewPortRegistry(), sends: make(map[OfferID]*sendHandle), } @@ -141,8 +144,13 @@ func (m *Manager) DeleteTransfer(id OfferID) { m.history.Delete(id) } -// DestinationDir returns the directory received files are delivered to. +// DestinationDir returns where received files are delivered: the platform +// sink's own name for it when one stages payloads, otherwise the configured +// directory. func (m *Manager) DestinationDir() string { + if labeller, ok := m.sink.(interface{ DestinationLabel() string }); ok { + return labeller.DestinationLabel() + } return m.policy.DestinationDir() } @@ -162,6 +170,7 @@ func (m *Manager) StartReceiver(ctx context.Context, addr netip.AddrPort, netsta server, err := NewServer(ServerConfig{ SpoolDir: filepath.Join(m.dataDir, "spool"), + Sink: m.sink, Policy: m.policy, Resolver: resolver, Notifier: m, @@ -577,9 +586,12 @@ func (m *Manager) OnCompleted(offer Offer) { return } - delivered, err := deliver(server.Spool(), offer, m.policy.DestinationDir()) + delivered, err := server.Spool().Deliver(offer, m.policy.DestinationDir()) if err != nil { log.Errorf("failed to deliver file drop payloads: %v", err) + // Nothing will ask for these payloads again, and a sink that cannot + // address them by path has no sweep of its own to fall back on. + server.Spool().Remove(offer.ID) m.finishTransfer(offer.ID, StateFailed, err.Error()) m.emit(EventFailed, m.transferOf(offer.ID)) return diff --git a/client/internal/filedrop/receiver.go b/client/internal/filedrop/receiver.go index 3f5b610ea..45e17fdee 100644 --- a/client/internal/filedrop/receiver.go +++ b/client/internal/filedrop/receiver.go @@ -27,11 +27,11 @@ type receiver struct { resolver PeerResolver notifier Notifier offers *OfferStore - spool *Spool + spool Sink spoolMaxAge time.Duration } -func newReceiver(cfg ServerConfig, spool *Spool, maxAge time.Duration) *receiver { +func newReceiver(cfg ServerConfig, spool Sink, maxAge time.Duration) *receiver { return &receiver{ policy: cfg.Policy, resolver: cfg.Resolver, @@ -162,7 +162,7 @@ func (r *receiver) upload(sender senderIdentity, id OfferID, index int, offset i }, } - received, err := r.spool.Write(id, index, offset, staged, size) + received, err := r.spool.Write(id, index, offer.Files[index].Name, offset, staged, size) // A withdrawn offer is an answer, not a failure: the state it moved to is // the one the user chose, and the spool is already gone. diff --git a/client/internal/filedrop/server.go b/client/internal/filedrop/server.go index ec39535ee..70a04d539 100644 --- a/client/internal/filedrop/server.go +++ b/client/internal/filedrop/server.go @@ -35,9 +35,12 @@ type Notifier interface { OnWithdrawn(offer Offer) } -// ServerConfig configures the receiving side. +// ServerConfig configures the receiving side. Sink overrides the filesystem +// spool, for a platform that stages payloads somewhere the engine cannot +// address; SpoolDir is then unused. type ServerConfig struct { SpoolDir string + Sink Sink Policy *PolicyStore Resolver PeerResolver Notifier Notifier @@ -70,9 +73,13 @@ func NewServer(cfg ServerConfig) (*Server, error) { return nil, errors.New("receiving policy is required") } - spool, err := NewSpool(cfg.SpoolDir) - if err != nil { - return nil, fmt.Errorf("create spool: %w", err) + spool := cfg.Sink + if spool == nil { + fsSpool, err := NewSpool(cfg.SpoolDir) + if err != nil { + return nil, fmt.Errorf("create spool: %w", err) + } + spool = fsSpool } maxAge := cfg.SpoolMaxAge @@ -96,10 +103,17 @@ func (s *Server) Offers() *OfferStore { } // Spool returns the staging area, so the platform layer can deliver completed payloads. -func (s *Server) Spool() *Spool { +func (s *Server) Spool() Sink { return s.recv.spool } +// FileSpool returns the staging area as a filesystem spool, nil when the +// platform staged payloads elsewhere. +func (s *Server) FileSpool() *Spool { + spool, _ := s.recv.spool.(*Spool) + return spool +} + // Policy returns the active profile's receiving policy store. func (s *Server) Policy() *PolicyStore { return s.recv.policy diff --git a/client/internal/filedrop/sink.go b/client/internal/filedrop/sink.go new file mode 100644 index 000000000..0cc327bec --- /dev/null +++ b/client/internal/filedrop/sink.go @@ -0,0 +1,135 @@ +package filedrop + +import ( + "fmt" + "io" + "strings" + "time" +) + +// Sink stages incoming payloads until an offer completes. The receiver writes +// every payload through it and never touches the filesystem directly, so a +// platform whose destination is not a filesystem path can implement its own. +type Sink interface { + Prepare(id OfferID) error + Received(id OfferID, index int) (int64, error) + Write(id OfferID, index int, name string, offset int64, r io.Reader, limit int64) (int64, error) + Deliver(offer Offer, destDir string) ([]string, error) + Remove(id OfferID) + Cleanup(maxAge time.Duration, now time.Time) +} + +// PlatformSink is the platform-facing half of a Sink, bound over gomobile. It +// carries the same calls with only types gomobile can bind, so a platform can +// stage payloads somewhere the engine cannot address by path, such as an +// Android MediaStore entry. +type PlatformSink interface { + // DestinationLabel names where payloads land, for the UI to show in place of + // a path the platform does not have. + DestinationLabel() string + Prepare(offerID string) error + Received(offerID string, index int) (int64, error) + OpenWriter(offerID string, index int, name string, offset int64, size int64) (PlatformWriter, error) + // Deliver publishes an offer's payloads and returns where each one landed, + // newline-separated, in the order the offer announced them, text payloads + // skipped. + Deliver(offerID string) (string, error) + Remove(offerID string) + Cleanup(maxAgeSeconds int64) +} + +// PlatformWriter is one payload's destination, opened by a PlatformSink. +// +// WriteChunk takes the bytes by value rather than filling a caller-supplied +// buffer: gomobile copies a []byte argument into a fresh Java array, which is +// the right direction here, but the written count never crosses back, so the +// writer reports its own total through Written instead, read after Close. +type PlatformWriter interface { + WriteChunk(p []byte) error + Close() error + Written() int64 +} + +// platformSink adapts a PlatformSink to the Sink the receiver uses. +type platformSink struct { + platform PlatformSink +} + +// platformWriter adapts a PlatformWriter to io.Writer. +type platformWriter struct { + w PlatformWriter +} + +// NewPlatformSink wraps a platform-provided sink for the receiver to stage into. +func NewPlatformSink(platform PlatformSink) (Sink, error) { + if platform == nil { + return nil, fmt.Errorf("platform sink is required") + } + return &platformSink{platform: platform}, nil +} + +// DestinationLabel reports where the platform delivers payloads. +func (s *platformSink) DestinationLabel() string { + return s.platform.DestinationLabel() +} + +func (s *platformSink) Prepare(id OfferID) error { + return s.platform.Prepare(string(id)) +} + +func (s *platformSink) Received(id OfferID, index int) (int64, error) { + return s.platform.Received(string(id), index) +} + +func (s *platformSink) Write(id OfferID, index int, name string, offset int64, r io.Reader, limit int64) (int64, error) { + if offset < 0 { + return 0, fmt.Errorf("negative offset %d", offset) + } + + w, err := s.platform.OpenWriter(string(id), index, name, offset, limit) + if err != nil { + return offset, fmt.Errorf("open destination: %w", err) + } + + _, copyErr := io.Copy(&platformWriter{w: w}, io.LimitReader(r, limit-offset)) + closeErr := w.Close() + + written := w.Written() + if copyErr != nil { + return written, fmt.Errorf("write destination: %w", copyErr) + } + if closeErr != nil { + return written, fmt.Errorf("close destination: %w", closeErr) + } + return written, nil +} + +func (s *platformSink) Deliver(offer Offer, _ string) ([]string, error) { + delivered, err := s.platform.Deliver(string(offer.ID)) + if err != nil { + return nil, err + } + return splitDelivered(delivered), nil +} + +func (s *platformSink) Remove(id OfferID) { + s.platform.Remove(string(id)) +} + +func (s *platformSink) Cleanup(maxAge time.Duration, _ time.Time) { + s.platform.Cleanup(int64(maxAge / time.Second)) +} + +func (p *platformWriter) Write(b []byte) (int, error) { + if err := p.w.WriteChunk(b); err != nil { + return 0, err + } + return len(b), nil +} + +func splitDelivered(delivered string) []string { + if delivered == "" { + return nil + } + return strings.Split(delivered, "\n") +} diff --git a/client/internal/filedrop/sink_test.go b/client/internal/filedrop/sink_test.go new file mode 100644 index 000000000..4014e1e20 --- /dev/null +++ b/client/internal/filedrop/sink_test.go @@ -0,0 +1,219 @@ +package filedrop + +import ( + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type fakePlatformWriter struct { + written []byte + offset int64 + closed bool + writeErr error + closeErr error +} + +type fakePlatformSink struct { + writers map[string]*fakePlatformWriter + prepared []string + removed []string + cleanups []int64 + openErr error + closeErr error + openCalls []string + delivered string + label string +} + +func newFakePlatformSink() *fakePlatformSink { + return &fakePlatformSink{writers: map[string]*fakePlatformWriter{}} +} + +func (w *fakePlatformWriter) WriteChunk(p []byte) error { + if w.writeErr != nil { + return w.writeErr + } + w.written = append(w.written, p...) + return nil +} + +func (w *fakePlatformWriter) Close() error { + w.closed = true + return w.closeErr +} + +func (w *fakePlatformWriter) Written() int64 { + return w.offset + int64(len(w.written)) +} + +func (s *fakePlatformSink) DestinationLabel() string { + return s.label +} + +func (s *fakePlatformSink) Prepare(offerID string) error { + s.prepared = append(s.prepared, offerID) + return nil +} + +func (s *fakePlatformSink) Received(offerID string, index int) (int64, error) { + w, ok := s.writers[key(offerID, index)] + if !ok { + return 0, nil + } + return w.Written(), nil +} + +func (s *fakePlatformSink) OpenWriter(offerID string, index int, name string, offset int64, _ int64) (PlatformWriter, error) { + s.openCalls = append(s.openCalls, name) + if s.openErr != nil { + return nil, s.openErr + } + w := &fakePlatformWriter{offset: offset, closeErr: s.closeErr} + s.writers[key(offerID, index)] = w + return w, nil +} + +func (s *fakePlatformSink) Deliver(string) (string, error) { + return s.delivered, nil +} + +func (s *fakePlatformSink) Remove(offerID string) { + s.removed = append(s.removed, offerID) +} + +func (s *fakePlatformSink) Cleanup(maxAgeSeconds int64) { + s.cleanups = append(s.cleanups, maxAgeSeconds) +} + +func key(offerID string, index int) string { + return fmt.Sprintf("%s:%d", offerID, index) +} + +func TestPlatformSinkWritesThroughToTheDestination(t *testing.T) { + platform := newFakePlatformSink() + sink, err := NewPlatformSink(platform) + require.NoError(t, err) + + written, err := sink.Write("offer", 0, "report.bin", 0, strings.NewReader("netbird"), 7) + require.NoError(t, err) + assert.Equal(t, int64(7), written) + + w := platform.writers[key("offer", 0)] + assert.Equal(t, "netbird", string(w.written)) + assert.True(t, w.closed, "the destination must be closed after the write") + assert.Equal(t, []string{"report.bin"}, platform.openCalls, + "the announced name must reach the platform, which owns the final naming") +} + +func TestPlatformSinkStopsAtTheAnnouncedSize(t *testing.T) { + platform := newFakePlatformSink() + sink, err := NewPlatformSink(platform) + require.NoError(t, err) + + written, err := sink.Write("offer", 0, "x.bin", 0, strings.NewReader("more than announced"), 4) + require.NoError(t, err) + assert.Equal(t, int64(4), written, "a sender overrunning its own offer must be cut off") + assert.Equal(t, "more", string(platform.writers[key("offer", 0)].written)) +} + +func TestPlatformSinkResumeReportsTheTotal(t *testing.T) { + platform := newFakePlatformSink() + sink, err := NewPlatformSink(platform) + require.NoError(t, err) + + written, err := sink.Write("offer", 0, "x.bin", 3, strings.NewReader("def"), 6) + require.NoError(t, err) + assert.Equal(t, int64(6), written, + "the platform counts from the offset it resumed at, not from this call alone") +} + +func TestPlatformSinkFailsWhenTheDestinationWillNotOpen(t *testing.T) { + platform := newFakePlatformSink() + platform.openErr = errors.New("no space") + sink, err := NewPlatformSink(platform) + require.NoError(t, err) + + written, err := sink.Write("offer", 0, "x.bin", 5, strings.NewReader("data"), 9) + require.Error(t, err) + assert.Equal(t, int64(5), written, "the offset already staged survives a failed reopen") +} + +func TestPlatformSinkReportsWhatLandedWhenTheCloseFails(t *testing.T) { + platform := newFakePlatformSink() + sink, err := NewPlatformSink(platform) + require.NoError(t, err) + + // A destination that accepts every chunk but fails to flush has to report the + // bytes that did land, or a resume would restart from the wrong offset. + platform.closeErr = errors.New("flush failed") + + written, err := sink.Write("offer", 0, "x.bin", 0, strings.NewReader("netbird"), 7) + require.Error(t, err) + assert.Equal(t, int64(7), written) +} + +func TestPlatformSinkRejectsNegativeOffset(t *testing.T) { + sink, err := NewPlatformSink(newFakePlatformSink()) + require.NoError(t, err) + + _, err = sink.Write("offer", 0, "x.bin", -1, strings.NewReader("data"), 4) + require.Error(t, err) +} + +func TestPlatformSinkDeliverSplitsTheReportedDestinations(t *testing.T) { + platform := newFakePlatformSink() + platform.delivered = "content://media/1\ncontent://media/2" + sink, err := NewPlatformSink(platform) + require.NoError(t, err) + + delivered, err := sink.Deliver(Offer{ID: "offer"}, "") + require.NoError(t, err) + assert.Equal(t, []string{"content://media/1", "content://media/2"}, delivered) +} + +func TestPlatformSinkDeliverWithoutDestinationsIsEmpty(t *testing.T) { + sink, err := NewPlatformSink(newFakePlatformSink()) + require.NoError(t, err) + + delivered, err := sink.Deliver(Offer{ID: "offer"}, "") + require.NoError(t, err) + assert.Empty(t, delivered, "a text-only offer delivers nothing, rather than one empty path") +} + +func TestPlatformSinkCleanupPassesWholeSeconds(t *testing.T) { + platform := newFakePlatformSink() + sink, err := NewPlatformSink(platform) + require.NoError(t, err) + + sink.Cleanup(90*time.Minute, time.Now()) + assert.Equal(t, []int64{5400}, platform.cleanups) +} + +func TestNewPlatformSinkRequiresAPlatform(t *testing.T) { + _, err := NewPlatformSink(nil) + require.Error(t, err) +} + +func TestManagerReportsThePlatformDestinationLabel(t *testing.T) { + platform := newFakePlatformSink() + platform.label = "Download/NetBird" + sink, err := NewPlatformSink(platform) + require.NoError(t, err) + + mgr, err := NewManager(ManagerConfig{ + Profile: testProfile, + DataDir: t.TempDir(), + Sink: sink, + }) + require.NoError(t, err) + + require.NoError(t, mgr.SetDestinationDir("/data/data/io.netbird.client/files")) + assert.Equal(t, "Download/NetBird", mgr.DestinationDir(), + "the platform's own name for the destination wins over the stored path") +} diff --git a/client/internal/filedrop/spool.go b/client/internal/filedrop/spool.go index 646d906fd..91ea53a5b 100644 --- a/client/internal/filedrop/spool.go +++ b/client/internal/filedrop/spool.go @@ -61,8 +61,10 @@ func (s *Spool) Received(id OfferID, index int) (int64, error) { return info.Size(), nil } -// Write appends the payload at offset, truncating any bytes past it first. -func (s *Spool) Write(id OfferID, index int, offset int64, r io.Reader, limit int64) (int64, error) { +// Write appends the payload at offset, truncating any bytes past it first. The +// announced name plays no part: payloads are staged under their index and only +// take their name in Deliver. +func (s *Spool) Write(id OfferID, index int, _ string, offset int64, r io.Reader, limit int64) (int64, error) { if offset < 0 { return 0, fmt.Errorf("negative offset %d", offset) } @@ -92,6 +94,12 @@ func (s *Spool) Write(id OfferID, index int, offset int64, r io.Reader, limit in return offset + written, nil } +// Deliver moves an offer's staged payloads into destDir under their announced +// names and returns where each one landed. +func (s *Spool) Deliver(offer Offer, destDir string) ([]string, error) { + return deliver(s, offer, destDir) +} + // Path returns the staged path of one item for the platform layer to deliver from. func (s *Spool) Path(id OfferID, index int) string { return s.filePath(id, index)