init

2026-02-04 02:21:23 +08:00
commit 208786aa90
112 changed files with 9571 additions and 0 deletions
--- a/internal/transfer/client.go
+++ b/internal/transfer/client.go
@@ -0,0 +1,158 @@
+package transfer
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log/slog"
+	"mesh-drop/internal/discovery"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+
+	"github.com/google/uuid"
+)
+
+func (s *Service) SendFile(target *discovery.Peer, targetIP string, filePath string) {
+	file, err := os.Open(filePath)
+	if err != nil {
+		slog.Error("Failed to open file", "path", filePath, "error", err, "component", "transfer-client")
+		return
+	}
+	defer file.Close()
+
+	stat, err := file.Stat()
+	if err != nil {
+		return
+	}
+
+	task := Transfer{
+		ID:       uuid.New().String(),
+		FileName: filepath.Base(filePath),
+		FileSize: stat.Size(),
+		Sender: Sender{
+			ID:   s.discoveryService.GetID(),
+			Name: s.discoveryService.GetName(),
+		},
+		Type:        TransferTypeSend,
+		Status:      TransferStatusPending,
+		ContentType: ContentTypeFile,
+	}
+
+	s.processTransfer(target, targetIP, task, file)
+}
+
+func (s *Service) SendText(target *discovery.Peer, targetIP string, text string) {
+	reader := bytes.NewReader([]byte(text))
+	task := Transfer{
+		ID:       uuid.New().String(),
+		FileName: "",
+		FileSize: int64(len(text)),
+		Sender: Sender{
+			ID:   s.discoveryService.GetID(),
+			Name: s.discoveryService.GetName(),
+		},
+		Type:        TransferTypeSend,
+		Status:      TransferStatusPending,
+		ContentType: ContentTypeText,
+	}
+
+	s.processTransfer(target, targetIP, task, reader)
+}
+
+func (s *Service) processTransfer(target *discovery.Peer, targetIP string, task Transfer, payload io.Reader) {
+	s.transferList.Store(task.ID, task)
+	s.app.Event.Emit("transfer:refreshList")
+
+	// 发送请求
+	askBody, _ := json.Marshal(task)
+
+	askUrl := fmt.Sprintf("http://%s:%d/transfer/ask", targetIP, target.Port)
+
+	resp, err := http.Post(askUrl, "application/json", bytes.NewReader(askBody))
+	if err != nil {
+		slog.Error("Failed to send ask request", "url", askUrl, "error", err, "component", "transfer-client")
+		// 如果请求发送失败，更新状态为 Error
+		task.Status = TransferStatusError
+		task.ErrorMsg = fmt.Sprintf("Failed to connect to receiver: %v", err)
+		s.transferList.Store(task.ID, task)
+		s.app.Event.Emit("transfer:refreshList")
+		return
+	}
+	defer resp.Body.Close()
+
+	var askResp TransferAskResponse
+	if err := json.NewDecoder(resp.Body).Decode(&askResp); err != nil {
+		return
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		task.Status = TransferStatusError
+		task.ErrorMsg = askResp.Message
+		s.transferList.Store(task.ID, task)
+		s.app.Event.Emit("transfer:refreshList")
+		return
+	}
+
+	if !askResp.Accepted {
+		// 接收方拒绝
+		task.Status = TransferStatusRejected
+		s.transferList.Store(task.ID, task)
+		s.app.Event.Emit("transfer:refreshList")
+		return
+	}
+
+	// 上传
+	uploadUrl, _ := url.Parse(fmt.Sprintf("http://%s:%d/transfer/upload/%s", targetIP, target.Port, task.ID))
+	query := uploadUrl.Query()
+	query.Add("token", askResp.Token)
+	uploadUrl.RawQuery = query.Encode()
+
+	reader := &PassThroughReader{
+		Reader: payload,
+		total:  task.FileSize,
+		callback: func(current, total int64, speed float64) {
+			task.Progress = Progress{
+				Current: current,
+				Total:   total,
+				Speed:   speed,
+			}
+			s.transferList.Store(task.ID, task)
+			s.app.Event.Emit("transfer:refreshList")
+		},
+	}
+
+	req, err := http.NewRequest(http.MethodPut, uploadUrl.String(), reader)
+	if err != nil {
+		return
+	}
+	req.ContentLength = task.FileSize
+	req.Header.Set("Content-Type", "application/octet-stream")
+
+	resp, err = http.DefaultClient.Do(req)
+	if err != nil {
+		slog.Error("Failed to upload file", "url", uploadUrl.String(), "error", err, "component", "transfer-client")
+		return
+	}
+	defer resp.Body.Close()
+
+	var uploadResp TransferUploadResponse
+	if err := json.NewDecoder(resp.Body).Decode(&uploadResp); err != nil {
+		return
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		task.Status = TransferStatusError
+		task.ErrorMsg = uploadResp.Message
+		s.transferList.Store(task.ID, task)
+		s.app.Event.Emit("transfer:refreshList")
+		return
+	}
+
+	// 传输成功，任务结束
+	task.Status = TransferStatusCompleted
+	s.transferList.Store(task.ID, task)
+	s.app.Event.Emit("transfer:refreshList")
+}
--- a/internal/transfer/model.go
+++ b/internal/transfer/model.go
@@ -0,0 +1,78 @@
+package transfer
+
+type TransferStatus string
+
+const (
+	TransferStatusPending   TransferStatus = "pending"
+	TransferStatusAccepted  TransferStatus = "accepted"
+	TransferStatusRejected  TransferStatus = "rejected"
+	TransferStatusCompleted TransferStatus = "completed"
+	TransferStatusError     TransferStatus = "error"
+	TransferStatusCanceled  TransferStatus = "canceled"
+	TransferStatusActive    TransferStatus = "active"
+)
+
+type TransferType string
+
+const (
+	TransferTypeSend    TransferType = "send"
+	TransferTypeReceive TransferType = "receive"
+)
+
+type ContentType string
+
+const (
+	ContentTypeFile   ContentType = "file"
+	ContentTypeText   ContentType = "text"
+	ContentTypeFolder ContentType = "folder"
+)
+
+// Transfer
+type Transfer struct {
+	ID           string         `json:"id" binding:"required"`     // 传输会话 ID
+	Sender       Sender         `json:"sender" binding:"required"` // 发送者
+	FileName     string         `json:"file_name"`                 // 文件名
+	FileSize     int64          `json:"file_size"`                 // 文件大小 (字节)
+	SavePath     string         `json:"savePath"`                  // 保存路径
+	Status       TransferStatus `json:"status"`                    // 传输状态
+	Progress     Progress       `json:"progress"`                  // 传输进度
+	Type         TransferType   `json:"type"`                      // 进度类型
+	ContentType  ContentType    `json:"content_type"`              // 内容类型
+	Text         string         `json:"text"`                      // 文本内容
+	ErrorMsg     string         `json:"error_msg"`                 // 错误信息
+	Token        string         `json:"token"`                     // 用于上传的凭证
+	DecisionChan chan Decision  `json:"-"`                         // 用户决策通道
+}
+
+type Sender struct {
+	ID   string `json:"id" binding:"required"`   // 发送者 ID
+	Name string `json:"name" binding:"required"` // 发送者名称
+}
+
+// Progress 用户前端传输进度
+type Progress struct {
+	Current int64   `json:"current"` // 当前进度
+	Total   int64   `json:"total"`   // 总进度
+	Speed   float64 `json:"speed"`   // 速度
+}
+
+// Decision 用户前端决策
+type Decision struct {
+	ID       string `json:"id"` // 传输会话 ID
+	Accepted bool   `json:"accepted"`
+	SavePath string `json:"save_path"`
+}
+
+// TransferAskResponse 握手回应
+type TransferAskResponse struct {
+	ID       string `json:"id"` // 传输会话 ID
+	Accepted bool   `json:"accepted"`
+	Token    string `json:"token,omitempty"`   // 用于上传的凭证
+	Message  string `json:"message,omitempty"` // 错误信息
+}
+
+// TransferUploadResponse 上传回应
+type TransferUploadResponse struct {
+	ID      string `json:"id"` // 传输会话 ID
+	Message string `json:"message"`
+}
--- a/internal/transfer/progress.go
+++ b/internal/transfer/progress.go
@@ -0,0 +1,37 @@
+package transfer
+
+import (
+	"io"
+	"time"
+)
+
+type ProgressCallback func(current int64, total int64, speed float64)
+
+const (
+	ProgressInterval = 100 * time.Millisecond
+)
+
+// PassThroughReader 包装 io.Reader 以计算读取字节数
+type PassThroughReader struct {
+	io.Reader
+	total      int64
+	currentLen int64
+	lastTime   time.Time
+	lastLen    int64
+	callback   ProgressCallback
+}
+
+func (pt *PassThroughReader) Read(p []byte) (int, error) {
+	n, err := pt.Reader.Read(p)
+	pt.currentLen += int64(n)
+
+	if time.Since(pt.lastTime) > ProgressInterval || err == io.EOF {
+		// 计算速度，单位为字节/秒
+		speed := float64(pt.currentLen-pt.lastLen) / time.Since(pt.lastTime).Seconds()
+		pt.callback(pt.currentLen, pt.total, speed)
+		pt.lastTime = time.Now()
+		pt.lastLen = pt.currentLen
+	}
+
+	return n, err
+}
--- a/internal/transfer/server.go
+++ b/internal/transfer/server.go
@@ -0,0 +1,271 @@
+package transfer
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"log/slog"
+	"mesh-drop/internal/discovery"
+	"net/http"
+	"os"
+	"path/filepath"
+	"sync"
+
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+	"github.com/wailsapp/wails/v3/pkg/application"
+)
+
+type Service struct {
+	app      *application.App
+	port     int
+	savePath string // 默认下载目录
+
+	// pendingRequests 存储等待用户确认的通道
+	// Key: TransferID, Value: Transfer
+	transferList sync.Map
+
+	discoveryService *discovery.Service
+}
+
+func NewService(app *application.App, port int, defaultSavePath string, discoveryService *discovery.Service) *Service {
+	gin.SetMode(gin.ReleaseMode)
+
+	return &Service{
+		app:              app,
+		port:             port,
+		savePath:         defaultSavePath,
+		discoveryService: discoveryService,
+	}
+}
+
+func init() {
+	application.RegisterEvent[application.Void]("transfer:refreshList")
+}
+
+func (s *Service) GetPort() int {
+	return s.port
+}
+
+func (s *Service) Start() {
+	r := gin.Default()
+	transfer := r.Group("/transfer")
+	{
+		transfer.POST("/ask", s.handleAsk)
+		transfer.PUT("/upload/:id", s.handleUpload)
+	}
+
+	go func() {
+		addr := fmt.Sprintf(":%d", s.port)
+		slog.Info("Transfer service listening", "address", addr, "component", "transfer")
+		if err := r.Run(addr); err != nil {
+			slog.Error("Transfer service error", "error", err, "component", "transfer")
+		}
+	}()
+}
+
+// handleAsk 处理接收文件请求
+func (s *Service) handleAsk(c *gin.Context) {
+	var task Transfer
+
+	// Gin 的 BindJSON 自动处理 JSON 解析
+	if err := c.ShouldBindJSON(&task); err != nil {
+		c.JSON(http.StatusBadRequest, TransferAskResponse{
+			ID:      task.ID,
+			Message: "Invalid request",
+		})
+		return
+	}
+
+	// 检查是否已经存在
+	if _, exists := s.transferList.Load(task.ID); exists {
+		// 如果已经存在，说明是网络重试，直接忽略
+		return
+	}
+
+	// 存储请求
+	task.Type = TransferTypeReceive
+	task.Status = TransferStatusPending
+	task.DecisionChan = make(chan Decision)
+	s.transferList.Store(task.ID, task)
+
+	// 通知 Wails 前端
+	s.app.Event.Emit("transfer:refreshList")
+
+	// 等待用户决策或发送端放弃
+	select {
+	case decision := <-task.DecisionChan:
+		// 用户决策
+		if decision.Accepted {
+			task.Status = TransferStatusAccepted
+			task.SavePath = decision.SavePath
+			token := uuid.New().String()
+			task.Token = token
+			s.transferList.Store(task.ID, task)
+		} else {
+			task.Status = TransferStatusRejected
+			s.transferList.Store(task.ID, task)
+		}
+		c.JSON(http.StatusOK, TransferAskResponse{
+			ID:       task.ID,
+			Accepted: decision.Accepted,
+			Token:    task.Token,
+		})
+	case <-c.Done():
+		// 发送端放弃
+		task.Status = TransferStatusCanceled
+		s.transferList.Store(task.ID, task)
+		s.app.Event.Emit("transfer:refreshList")
+	}
+}
+
+// ResolvePendingRequest 外部调用，解决待处理的传输请求
+// 返回 true 表示成功处理，false 表示未找到该 ID 的请求
+func (s *Service) ResolvePendingRequest(id string, accept bool, savePath string) bool {
+	val, ok := s.transferList.Load(id)
+	if !ok {
+		return false
+	}
+
+	task := val.(Transfer)
+	task.DecisionChan <- Decision{
+		ID:       id,
+		Accepted: accept,
+		SavePath: savePath,
+	}
+	return true
+}
+
+// handleUpload 处理接收文件请求
+func (s *Service) handleUpload(c *gin.Context) {
+	id := c.Param("id")
+	token := c.Query("token")
+
+	if id == "" || token == "" {
+		c.JSON(http.StatusBadRequest, TransferUploadResponse{
+			ID:      id,
+			Message: "Invalid request: missing id or token",
+		})
+		return
+	}
+
+	// 获取传输任务
+	val, ok := s.transferList.Load(id)
+	if !ok {
+		c.JSON(http.StatusUnauthorized, TransferUploadResponse{
+			ID:      id,
+			Message: "Invalid request: task not found",
+		})
+		return
+	}
+	task := val.(Transfer)
+
+	// 校验 token
+	if task.Token != token {
+		c.JSON(http.StatusUnauthorized, TransferUploadResponse{
+			ID:      id,
+			Message: "Token mismatch",
+		})
+		return
+	}
+
+	// 校验状态
+	if task.Status != TransferStatusAccepted {
+		c.JSON(http.StatusForbidden, TransferUploadResponse{
+			ID:      id,
+			Message: "Invalid task status",
+		})
+		return
+	}
+
+	// 更新状态为 active
+	task.Status = TransferStatusActive
+	s.transferList.Store(task.ID, task)
+	s.app.Event.Emit("transfer:refreshList")
+
+	savePath := task.SavePath
+	if savePath == "" {
+		savePath = s.savePath
+	}
+
+	switch task.ContentType {
+	case ContentTypeFile:
+		destPath := filepath.Join(savePath, task.FileName)
+		file, err := os.Create(destPath)
+		if err != nil {
+			// 接收方无法创建文件，直接报错，任务结束
+			c.JSON(http.StatusInternalServerError, TransferUploadResponse{
+				ID:      task.ID,
+				Message: "Receiver failed to create file",
+			})
+			slog.Error("Failed to create file", "error", err, "component", "transfer")
+			task.Status = TransferStatusError
+			task.ErrorMsg = fmt.Errorf("receiver failed to create file: %v", err).Error()
+			s.transferList.Store(task.ID, task)
+			// 通知前端传输失败
+			s.app.Event.Emit("transfer:refreshList")
+			return
+		}
+		defer file.Close()
+		s.receive(c, &task, file)
+	case ContentTypeText:
+		var buf bytes.Buffer
+		s.receive(c, &task, &buf)
+		task.Text = buf.String()
+		s.transferList.Store(task.ID, task)
+		s.app.Event.Emit("transfer:refreshList")
+	case ContentTypeFolder:
+		// s.receiveFolder(c, savePath, task)
+	}
+}
+
+func (s *Service) receive(c *gin.Context, task *Transfer, writer io.Writer) {
+	// 包装 reader，用于计算进度
+	reader := &PassThroughReader{
+		Reader: c.Request.Body,
+		total:  task.FileSize,
+		callback: func(current, total int64, speed float64) {
+			task.Progress = Progress{
+				Current: current,
+				Total:   total,
+				Speed:   speed,
+			}
+			s.transferList.Store(task.ID, task)
+			s.app.Event.Emit("transfer:refreshList")
+		},
+	}
+
+	_, err := io.Copy(writer, reader)
+	if err != nil {
+		// 文件写入失败，直接报错，任务结束
+		c.JSON(http.StatusInternalServerError, TransferUploadResponse{
+			ID:      task.ID,
+			Message: "Failed to write file",
+		})
+		slog.Error("Failed to write file", "error", err, "component", "transfer")
+		task.Status = TransferStatusError
+		task.ErrorMsg = fmt.Errorf("failed to write file: %v", err).Error()
+		s.transferList.Store(task.ID, task)
+		// 通知前端传输失败
+		s.app.Event.Emit("transfer:refreshList")
+		return
+	}
+
+	c.JSON(http.StatusOK, TransferUploadResponse{
+		ID:      task.ID,
+		Message: "File received successfully",
+	})
+	// 传输成功，任务结束
+	task.Status = TransferStatusCompleted
+	s.transferList.Store(task.ID, task)
+	s.app.Event.Emit("transfer:refreshList")
+}
+
+func (s *Service) GetTransferList() []Transfer {
+	var requests []Transfer
+	s.transferList.Range(func(key, value any) bool {
+		requests = append(requests, value.(Transfer))
+		return true
+	})
+	return requests
+}