Merge branch 'feat/go-fuse-daemon' into refactor/remote-sync-process

Author: egalvis27

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "internxt",
-  "version": "2.5.4",
+  "version": "2.6.0",
   "author": "Internxt <hello@internxt.com>",
   "description": "Internxt Drive client UI",
   "main": "./dist/main/main.js",
@@ -15,7 +15,7 @@
     "build:renderer": "cross-env NODE_ENV=production TS_NODE_TRANSPILE_ONLY=true webpack --config ./.erb/configs/webpack.config.renderer.prod.ts",
     "rebuild": "electron-rebuild --parallel --types prod,dev,optional --module-dir .",
     "reinstall:nautilus-extension": "NODE_ENV=development ts-node src/apps/nautilus-extension/reload.ts",
-    "lint": "cross-env NODE_ENV=development eslint . --ext .ts,.tsx --max-warnings=60",
+    "lint": "cross-env NODE_ENV=development eslint . --ext .ts,.tsx --max-warnings=50",
     "lint:fix": "npm run lint --fix",
     "format": "prettier src --check",
     "format:fix": "prettier src --write",
@@ -81,9 +81,6 @@
     "rpm": {
       "depends": [
         "fuse-libs"
-      ],
-      "fpm": [
-        "--rpm-tag", "Recommends: nautilus-python"
       ]
     },
     "directories": {
@@ -230,4 +227,4 @@
   "engines": {
     "node": ">=18.0.0 <19.0.0"
   }
-}
+}

package.json

@@ -10,21 +10,23 @@ import (
 	"net"
 	"net/http"
 	"time"
+
+	"github.com/hanwen/go-fuse/v2/fuse"
 )
 
+
 type Client struct {
-    http *http.Client
-    socketPath string
+	http       *http.Client
+	socketPath string
 }
 
 func NewClient(socketPath string) *Client {
-    return &Client{
-        http:       NewUnixSocketClient(socketPath),
-        socketPath: socketPath,
-    }
+	return &Client{
+		http:       NewUnixSocketClient(socketPath),
+		socketPath: socketPath,
+	}
 }
 
-
 func NewUnixSocketClient(socketPath string) *http.Client {
 	return &http.Client{
 		Transport: &http.Transport{
@@ -59,35 +61,122 @@ func (client *Client) NotifyReady(logger *slog.Logger) error {
 	return nil
 }
 
- func (client *Client) Post(context context.Context, path OperationPath, in any, out any) error {
-  body, err := json.Marshal(in)
-  if err != nil {
-    return fmt.Errorf("failed to marshal request: %w", err)
-  }
-  url := serverURL + string(path)
-  req, err := http.NewRequestWithContext(context, http.MethodPost, url, bytes.NewBuffer(body))
-  if err != nil {
-		return fmt.Errorf("error creating Post request: %w", err)
+// Post sends a JSON body to the given operation path and returns an errno.
+// A non-200 HTTP response means a transport failure so we return fuse.EIO without reading the body.
+// uppon 200, the response always contains an errno field: non-zero means the operation failed with that errno,
+// zero means success and the remaining fields are the operation's data, unmarshalled into out if non-nil.
+func (client *Client) Post(context context.Context, path OperationPath, in any, out any) fuse.Status {
+	body, err := json.Marshal(in)
+	if err != nil {
+		return fuse.EIO
+	}
+	url := serverURL + string(path)
+	req, err := http.NewRequestWithContext(context, http.MethodPost, url, bytes.NewBuffer(body))
+	if err != nil {
+		return fuse.EIO
 	}
+	req.Header.Set("Content-Type", "application/json")
 
-  req.Header.Set("Content-Type", "application/json")
+	resp, err := client.http.Do(req)
+	if err != nil {
+		return fuse.EIO
+	}
+	defer func() { _ = resp.Body.Close() }()
 
-  resp, err := client.http.Do(req)
+	if resp.StatusCode != http.StatusOK {
+		return fuse.EIO
+	}
+
+	resBody, err := io.ReadAll(resp.Body)
 	if err != nil {
-		return fmt.Errorf("sending Post request: %w", err)
+		return fuse.EIO
+	}
+
+	var errResp ErrorResponse
+	if err = json.Unmarshal(resBody, &errResp); err == nil && errResp.Errno != 0 {
+		return fuse.Status(errResp.Errno)
 	}
-  defer func() { _ = resp.Body.Close() }()
 
-  if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("unexpected status from Post endpoint: %d", resp.StatusCode)
+	if out != nil {
+		if err = json.Unmarshal(resBody, out); err != nil {
+			return fuse.EIO
+		}
+	}
+
+	return fuse.OK
+}
+
+// PostBinary sends a JSON body to the given operation path and returns raw binary data.
+// Errors are signaled via the X-Errno response header (non-zero = fuse.Status error code).
+// On success (X-Errno: 0) the response body is raw bytes copied into dest.
+// Returns the number of bytes read and a fuse.Status.
+func (client *Client) PostBinary(ctx context.Context, path OperationPath, in any, dest []byte) (int, fuse.Status) {
+	body, err := json.Marshal(in)
+	if err != nil {
+		return 0, fuse.EIO
 	}
-  resBody, err := io.ReadAll(resp.Body)
-  if err != nil {
-		return fmt.Errorf("failed to read response body: %w", err)
+	url := serverURL + string(path)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(body))
+	if err != nil {
+		return 0, fuse.EIO
 	}
-  err = json.Unmarshal(resBody, out)
-  if err != nil {
-		return fmt.Errorf("failed to unmarshal response: %w", err)
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := client.http.Do(req)
+	if err != nil {
+		return 0, fuse.EIO
 	}
-  return nil
+	defer func() { _ = resp.Body.Close() }()
+
+	if resp.StatusCode != http.StatusOK {
+		return 0, fuse.EIO
+	}
+
+	if errnoStr := resp.Header.Get("X-Errno"); errnoStr != "" && errnoStr != "0" {
+		var errno int32
+		if _, err := fmt.Sscanf(errnoStr, "%d", &errno); err == nil && errno != 0 {
+			return 0, fuse.Status(errno)
+		}
+	}
+
+	bytesRead, err := io.ReadFull(resp.Body, dest)
+	if err != nil && err != io.ErrUnexpectedEOF {
+		return 0, fuse.EIO
+	}
+	return bytesRead, fuse.OK
+}
+
+// PostBinaryRequest sends raw binary payload to the given operation path.
+// Errors are signaled via the X-Errno response header (non-zero = fuse.Status error code).
+// On success it returns fuse.OK.
+func (client *Client) PostSendBinary(ctx context.Context, path OperationPath, payload []byte, headers map[string]string) fuse.Status {
+	url := serverURL + string(path)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(payload))
+	if err != nil {
+		return fuse.EIO
+	}
+	req.Header.Set("Content-Type", "application/octet-stream")
+	for key, value := range headers {
+		req.Header.Set(key, value)
+	}
+
+	resp, err := client.http.Do(req)
+	if err != nil {
+		return fuse.EIO
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if resp.StatusCode != http.StatusOK {
+		return fuse.EIO
+	}
+
+	if errnoStr := resp.Header.Get("X-Errno"); errnoStr != "" && errnoStr != "0" {
+		var errno int32
+		if _, err := fmt.Sscanf(errnoStr, "%d", &errno); err == nil && errno != 0 {
+			return fuse.Status(errno)
+		}
+		return fuse.EIO
+	}
+
+	return fuse.OK
 }

packages/fuse-daemon/internal/client/client.go

@@ -1,7 +1,24 @@
 package client
+
 type OperationPath string
-const(
-  OperationGetAttr OperationPath = "/op/getattributes"
+
+type ErrorResponse struct {
+	Errno int32 `json:"errno"`
+}
+
+const (
+	OperationGetAttr  OperationPath = "/op/getattributes"
+	OperationOpen     OperationPath = "/op/open"
+	OperationOpenDir  OperationPath = "/op/opendir"
+	OperationRead     OperationPath = "/op/read"
+	OperationTruncate OperationPath = "/op/truncate"
+	OperationCreate   OperationPath = "/op/create"
+	OperationWrite    OperationPath = "/op/write"
+	OperationRelease  OperationPath = "/op/release"
+	OperationMkdir    OperationPath = "/op/mkdir"
+	OperationRename   OperationPath = "/op/rename"
+	OperationUnlink   OperationPath = "/op/unlink"
+	OperationRmdir    OperationPath = "/op/rmdir"
 )
 
 const serverURL = "http://localhost"

packages/fuse-daemon/internal/client/operation_paths.go

@@ -0,0 +1,100 @@
+package filesystem
+
+import (
+	"context"
+	"strconv"
+	"log/slog"
+
+	"internxt/drive-desktop-linux/fuse-daemon/internal/client"
+
+	"github.com/hanwen/go-fuse/v2/fuse"
+	"github.com/hanwen/go-fuse/v2/fuse/nodefs"
+)
+
+type WriteCallbackData struct {
+	Written uint32 `json:"written"`
+}
+
+// InternxtFile is the file handle returned by Open.
+// It holds the context needed for future Read/Write implementation.
+// Operations with a path-based fallback (GetAttr, Chmod, Chown, Truncate, Utimens)
+// are intentionally not overridden — DefaultFile returns ENOSYS which triggers
+// the fallback to InternxtFilesystem automatically.
+type InternxtFile struct {
+	nodefs.File
+	path        string
+	flag        uint32
+	processName string
+	logger      *slog.Logger
+	client      *client.Client
+}
+
+func NewInternxtFile(path string, flag uint32, processName string, logger *slog.Logger, c *client.Client) *InternxtFile {
+	return &InternxtFile{
+		File:        nodefs.NewDefaultFile(),
+		path:        path,
+		flag:        flag,
+		processName: processName,
+		logger:      logger,
+		client:      c,
+	}
+}
+
+func (f *InternxtFile) String() string {
+	return "InternxtFile(" + f.path + ")"
+}
+
+func (f *InternxtFile) Read(dest []byte, off int64) (fuse.ReadResult, fuse.Status) {
+	f.logger.Debug("Received Read call", "path", f.path, "offset", off, "length", len(dest))
+	body := struct {
+		Path        string `json:"path"`
+		Offset      int64  `json:"offset"`
+		Length      int    `json:"length"`
+		ProcessName string `json:"processName"`
+	}{Path: f.path, Offset: off, Length: len(dest), ProcessName: f.processName}
+
+	bytesRead, status := f.client.PostBinary(context.Background(), client.OperationRead, body, dest)
+	if status != fuse.OK {
+		f.logger.Error("Error occurred while reading file", "status", status)
+		return nil, status
+	}
+	return fuse.ReadResultData(dest[:bytesRead]), fuse.OK
+}
+
+func (f *InternxtFile) Write(data []byte, off int64) (uint32, fuse.Status) {
+	f.logger.Debug("Received Write call", "path", f.path, "offset", off, "length", len(data))
+	headers := map[string]string{
+		"X-Path":   f.path,
+		"X-Offset": strconv.FormatInt(off, 10),
+	}
+
+	if status := f.client.PostSendBinary(context.Background(), client.OperationWrite, data, headers); status != fuse.OK {
+		f.logger.Error("Error occurred while writing file", "status", status)
+		return 0, status
+	}
+
+	return uint32(len(data)), fuse.OK
+}
+
+// Flush is called on each close(2) of the file descriptor.
+// Multiple flushes may occur if the file descriptor was duplicated.
+// Data is already persisted to the temporal file via Write, so no action is needed here.
+func (f *InternxtFile) Flush() fuse.Status {
+	return fuse.OK
+}
+
+func (f *InternxtFile) Release() {
+	f.logger.Debug("Received Release call:", "path", f.path)
+	body := struct {
+		Path        string `json:"path"`
+		ProcessName string `json:"processName"`
+	}{Path: f.path, ProcessName: f.processName}
+	if status := f.client.Post(context.Background(), client.OperationRelease, body, nil); status != fuse.OK {
+		f.logger.Warn("Release call failed", "path", f.path, "status", status)
+	}
+}
+
+func (f *InternxtFile) Fsync(flags int) fuse.Status {
+	f.logger.Warn("not implemented", "op", "Fsync", "path", f.path)
+	return fuse.ENOSYS
+}