feat: Implement full network functionality with HTTP and P2P modes
Go Build and Test / build-and-test (pull_request) Successful in 37s
Go Build and Test / build-and-test (pull_request) Successful in 37s
This major update introduces a robust, multi-faceted networking capability, allowing backups to be managed through a central server or via direct peer-to-peer connections. - **Add `serve` command**: Implemented a new `serve` command that runs a persistent HTTP server. This allows for a centralized backup management workflow where multiple clients can upload and download archives. - **HTTP Client Mode**: The `create` and `restore` commands can now function as HTTP clients, targeting a `serve` instance using the `ip:port/filename` address format. - **Restore P2P Mode**: The original peer-to-peer streaming functionality (where `restore` acts as a TCP server and `create` as a client) has been fully restored and integrated into the new, more robust command architecture. - **Accurate Network Progress**: Fixed a critical bug where network progress bars would not display correctly. The `serve` command now sends a `Content-Length` header, and `restore` is refactored to handle asynchronous state updates correctly. - **Smart Address Parsing**: The application now automatically distinguishes between HTTP mode (`ip:port/filename`) and P2P mode (`ip:port`), providing a seamless user experience. - **Update `README.md`**: The project's README has been completely rewritten to document all three operating modes (Local, HTTP Server, P2P) with clear usage examples. - **Add Gitea Actions Workflow**: A new CI pipeline (`.gitea/workflows/build-and-test.yml`) has been added to automatically build and test the application on every push and pull request.
This commit is contained in:
1 parent
668721ab58
commit
6b1a302c18
7 files changed
+543
-240
No files matched your search
+147
-111
@@ -7,6 +7,7 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
@@ -28,7 +29,7 @@ var (
|
||||
var restoreCmd = &cobra.Command{
|
||||
Use: "restore",
|
||||
Short: "Restore a backup",
|
||||
Long: "Restore a backup from a local archive or network stream to a file, disk, or directory.",
|
||||
Long: "Restore a backup from a local archive, P2P stream, or HTTP server.",
|
||||
Run: func(cmd *cobra.Command, args []string) {
|
||||
model, err := initialRestoreModel(restoreSource, restoreTarget)
|
||||
if err != nil {
|
||||
@@ -45,18 +46,27 @@ var restoreCmd = &cobra.Command{
|
||||
}
|
||||
|
||||
type restoreModel struct {
|
||||
progress progress.Model
|
||||
source string
|
||||
target string
|
||||
totalBytes int64
|
||||
processed int64
|
||||
done bool
|
||||
err error
|
||||
startTime time.Time
|
||||
progressCh chan int64
|
||||
isDirectory bool
|
||||
isNetworkSource bool
|
||||
statusMessage string
|
||||
progress progress.Model
|
||||
source string
|
||||
target string
|
||||
totalBytes int64
|
||||
processed int64
|
||||
done bool
|
||||
err error
|
||||
startTime time.Time
|
||||
progressCh chan int64
|
||||
isDirectory bool
|
||||
httpSourceInfo *ServerTargetInfo
|
||||
isP2PSource bool
|
||||
statusMessage string
|
||||
}
|
||||
|
||||
// restoreSourceReadyMsg is sent when the source is ready to be read.
|
||||
// It carries the reader, the total size, and whether it's a directory backup.
|
||||
type restoreSourceReadyMsg struct {
|
||||
reader io.ReadCloser
|
||||
totalBytes int64
|
||||
isDir bool
|
||||
}
|
||||
|
||||
type restoreProgressMsg int64
|
||||
@@ -64,21 +74,24 @@ type restoreDoneMsg struct{}
|
||||
type restoreErrorMsg struct{ err error }
|
||||
|
||||
func initialRestoreModel(src, dst string) (*restoreModel, error) {
|
||||
isNetwork := isNetworkAddress(src)
|
||||
p := progress.New(
|
||||
progress.WithDefaultGradient(),
|
||||
progress.WithWidth(40),
|
||||
)
|
||||
|
||||
var totalBytes int64 = 1 // Placeholder for network source, to avoid division by zero.
|
||||
var isDir bool
|
||||
var totalBytes int64 = 1 // Placeholder, will be updated.
|
||||
var isDir, isP2P bool
|
||||
targetPath := dst
|
||||
status := "Initializing..."
|
||||
|
||||
if isNetwork {
|
||||
if targetPath == "" {
|
||||
return nil, fmt.Errorf("target flag -t is required for a network source")
|
||||
}
|
||||
httpInfo, isHttp := parseServerTarget(src)
|
||||
|
||||
if isHttp {
|
||||
isDir = strings.HasSuffix(httpInfo.Filename, ".tar.gz")
|
||||
status = fmt.Sprintf("Connecting to %s...", httpInfo.Address)
|
||||
src = httpInfo.URL
|
||||
} else if isNetworkAddress(src) {
|
||||
isP2P = true
|
||||
status = fmt.Sprintf("Listening on %s...", src)
|
||||
} else {
|
||||
fileInfo, err := os.Stat(src)
|
||||
@@ -87,89 +100,96 @@ func initialRestoreModel(src, dst string) (*restoreModel, error) {
|
||||
}
|
||||
totalBytes = fileInfo.Size()
|
||||
isDir = strings.HasSuffix(src, ".tar.gz")
|
||||
}
|
||||
|
||||
if isDir && targetPath == "" {
|
||||
targetPath, err = os.Getwd()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get current directory: %w", err)
|
||||
}
|
||||
} else if !isDir && targetPath == "" {
|
||||
return nil, fmt.Errorf("target flag -t is required for file or disk restoration")
|
||||
if isDir && targetPath == "" {
|
||||
// For directory restores, we can default to the current directory.
|
||||
// We do this check here, but also again after getting the P2P type.
|
||||
var err error
|
||||
targetPath, err = os.Getwd()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get current directory: %w", err)
|
||||
}
|
||||
} else if !isDir && targetPath == "" && !isP2P {
|
||||
// Target is mandatory for file/disk restores (except P2P, where we can't know the type yet).
|
||||
return nil, fmt.Errorf("target flag -t is required for this restoration")
|
||||
}
|
||||
|
||||
return &restoreModel{
|
||||
progress: p,
|
||||
source: src,
|
||||
target: targetPath,
|
||||
totalBytes: totalBytes,
|
||||
startTime: time.Now(),
|
||||
progressCh: make(chan int64, 100),
|
||||
isDirectory: isDir,
|
||||
isNetworkSource: isNetwork,
|
||||
statusMessage: status,
|
||||
progress: p,
|
||||
source: src,
|
||||
target: targetPath,
|
||||
totalBytes: totalBytes,
|
||||
startTime: time.Now(),
|
||||
progressCh: make(chan int64, 100),
|
||||
isDirectory: isDir,
|
||||
httpSourceInfo: httpInfo,
|
||||
isP2PSource: isP2P,
|
||||
statusMessage: status,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (m *restoreModel) Init() tea.Cmd {
|
||||
return tea.Batch(
|
||||
m.startRestore,
|
||||
m.restoreProgressListener,
|
||||
)
|
||||
return m.fetchSourceCmd
|
||||
}
|
||||
|
||||
func (m *restoreModel) getSourceReader() (io.ReadCloser, error) {
|
||||
if m.isNetworkSource {
|
||||
func (m *restoreModel) fetchSourceCmd() tea.Msg {
|
||||
if m.httpSourceInfo != nil {
|
||||
// New HTTP Mode
|
||||
client := &http.Client{}
|
||||
resp, err := client.Get(m.source)
|
||||
if err != nil {
|
||||
return restoreErrorMsg{fmt.Errorf("failed to connect to server: %w", err)}
|
||||
}
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
bodyBytes, _ := io.ReadAll(resp.Body)
|
||||
resp.Body.Close()
|
||||
return restoreErrorMsg{fmt.Errorf("server returned non-200 status: %s\n%s", resp.Status, string(bodyBytes))}
|
||||
}
|
||||
return restoreSourceReadyMsg{reader: resp.Body, totalBytes: resp.ContentLength, isDir: m.isDirectory}
|
||||
|
||||
} else if m.isP2PSource {
|
||||
// Old P2P Mode
|
||||
listener, err := net.Listen("tcp", m.source)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to listen on %s: %w", m.source, err)
|
||||
return restoreErrorMsg{fmt.Errorf("failed to listen on %s: %w", m.source, err)}
|
||||
}
|
||||
defer listener.Close() // Close listener after accepting one connection
|
||||
|
||||
m.statusMessage = fmt.Sprintf("Waiting for connection on %s", m.source)
|
||||
// This is a blocking call
|
||||
// Accept is blocking, so it must be in a command.
|
||||
conn, err := listener.Accept()
|
||||
listener.Close() // Close listener after one connection.
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to accept connection: %w", err)
|
||||
return restoreErrorMsg{fmt.Errorf("failed to accept connection: %w", err)}
|
||||
}
|
||||
m.statusMessage = "Client connected. Receiving data..."
|
||||
|
||||
// Protocol: Read 8 bytes for size, 1 byte for type
|
||||
if err := binary.Read(conn, binary.BigEndian, &m.totalBytes); err != nil {
|
||||
var size int64
|
||||
if err := binary.Read(conn, binary.BigEndian, &size); err != nil {
|
||||
conn.Close()
|
||||
return nil, fmt.Errorf("failed to read backup size: %w", err)
|
||||
return restoreErrorMsg{fmt.Errorf("failed to read backup size: %w", err)}
|
||||
}
|
||||
if m.totalBytes == 0 {
|
||||
m.totalBytes = 1 // Avoid division by zero for empty files
|
||||
}
|
||||
|
||||
typeByte := make([]byte, 1)
|
||||
if _, err := io.ReadFull(conn, typeByte); err != nil {
|
||||
conn.Close()
|
||||
return nil, fmt.Errorf("failed to read backup type: %w", err)
|
||||
return restoreErrorMsg{fmt.Errorf("failed to read backup type: %w", err)}
|
||||
}
|
||||
m.isDirectory = (typeByte[0] == 1)
|
||||
|
||||
return conn, nil
|
||||
return restoreSourceReadyMsg{reader: conn, totalBytes: size, isDir: typeByte[0] == 1}
|
||||
}
|
||||
|
||||
// Logic for local file
|
||||
m.statusMessage = fmt.Sprintf("Opening archive %s", m.source)
|
||||
return os.Open(m.source)
|
||||
// Local File Mode
|
||||
fileInfo, err := os.Stat(m.source)
|
||||
if err != nil {
|
||||
return restoreErrorMsg{err}
|
||||
}
|
||||
file, err := os.Open(m.source)
|
||||
if err != nil {
|
||||
return restoreErrorMsg{err}
|
||||
}
|
||||
return restoreSourceReadyMsg{reader: file, totalBytes: fileInfo.Size(), isDir: m.isDirectory}
|
||||
}
|
||||
|
||||
func (m *restoreModel) startRestore() tea.Msg {
|
||||
go func() {
|
||||
sourceReader, err := m.getSourceReader()
|
||||
if err != nil {
|
||||
m.err = err
|
||||
m.progressCh <- -1
|
||||
return
|
||||
}
|
||||
defer sourceReader.Close()
|
||||
|
||||
progressReader := &restoreProgressReader{
|
||||
reader: sourceReader,
|
||||
func (m *restoreModel) processRestoreCmd(r io.ReadCloser) tea.Cmd {
|
||||
return func() tea.Msg {
|
||||
defer r.Close()
|
||||
progressReader := &progressTracker{
|
||||
Reader: r,
|
||||
progressCh: m.progressCh,
|
||||
}
|
||||
|
||||
@@ -183,12 +203,11 @@ func (m *restoreModel) startRestore() tea.Msg {
|
||||
if restoreErr != nil {
|
||||
m.err = restoreErr
|
||||
m.progressCh <- -1
|
||||
return
|
||||
} else {
|
||||
m.progressCh <- -2 // Done signal
|
||||
}
|
||||
|
||||
m.progressCh <- -2 // Done signal
|
||||
}()
|
||||
return nil
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func (m *restoreModel) restoreFileOrDisk(r io.Reader) error {
|
||||
@@ -212,7 +231,8 @@ func (m *restoreModel) restoreFileOrDisk(r io.Reader) error {
|
||||
}
|
||||
defer dstFile.Close()
|
||||
|
||||
if _, err = io.Copy(dstFile, gzipReader); err != nil {
|
||||
_, err = io.Copy(dstFile, gzipReader)
|
||||
if err != nil && err != io.EOF {
|
||||
return err
|
||||
}
|
||||
return dstFile.Sync()
|
||||
@@ -264,37 +284,19 @@ func (m *restoreModel) restoreDirectory(r io.Reader) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
type restoreProgressReader struct {
|
||||
reader io.Reader
|
||||
progressCh chan int64
|
||||
read int64
|
||||
}
|
||||
|
||||
func (r *restoreProgressReader) Read(p []byte) (int, error) {
|
||||
n, err := r.reader.Read(p)
|
||||
if n > 0 {
|
||||
r.read += int64(n)
|
||||
r.progressCh <- r.read
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
|
||||
func (m *restoreModel) restoreProgressListener() tea.Msg {
|
||||
select {
|
||||
case n := <-m.progressCh:
|
||||
switch {
|
||||
case n == -1: // Ошибка
|
||||
n := <-m.progressCh
|
||||
if n < 0 {
|
||||
if n == -1 {
|
||||
if m.err == nil {
|
||||
m.err = fmt.Errorf("unknown restoration error")
|
||||
}
|
||||
return restoreErrorMsg{m.err}
|
||||
case n == -2: // Завершение
|
||||
return restoreDoneMsg{}
|
||||
default: // Прогресс
|
||||
m.processed = n
|
||||
return restoreProgressMsg(n)
|
||||
}
|
||||
return restoreDoneMsg{}
|
||||
}
|
||||
m.processed = n
|
||||
return restoreProgressMsg(n)
|
||||
}
|
||||
|
||||
func (m *restoreModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
|
||||
@@ -309,6 +311,31 @@ func (m *restoreModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
|
||||
m.progress.Width = msg.Width - 4
|
||||
return m, nil
|
||||
|
||||
case restoreSourceReadyMsg:
|
||||
m.totalBytes = msg.totalBytes
|
||||
if m.totalBytes <= 0 {
|
||||
m.totalBytes = 1
|
||||
}
|
||||
m.isDirectory = msg.isDir // Update directory type, especially for P2P.
|
||||
|
||||
// Final check for target directory now that we know the type for sure.
|
||||
if m.isDirectory && m.target == "" {
|
||||
wd, err := os.Getwd()
|
||||
if err != nil {
|
||||
return m, func() tea.Msg { return restoreErrorMsg{err} }
|
||||
}
|
||||
m.target = wd
|
||||
} else if !m.isDirectory && m.target == "" {
|
||||
err := fmt.Errorf("target flag -t is required for file/disk restoration")
|
||||
return m, func() tea.Msg { return restoreErrorMsg{err} }
|
||||
}
|
||||
|
||||
m.statusMessage = "Receiving data..."
|
||||
return m, tea.Batch(
|
||||
m.processRestoreCmd(msg.reader),
|
||||
m.restoreProgressListener,
|
||||
)
|
||||
|
||||
case restoreProgressMsg:
|
||||
progressVal := float64(msg) / float64(m.totalBytes)
|
||||
if progressVal > 1.0 {
|
||||
@@ -340,6 +367,9 @@ func (m *restoreModel) View() string {
|
||||
if m.isDirectory {
|
||||
restoreType = "Directory"
|
||||
}
|
||||
if m.httpSourceInfo != nil || m.isP2PSource {
|
||||
restoreType += " network"
|
||||
}
|
||||
return fmt.Sprintf("\n✅ %s restoration complete!\n\n"+
|
||||
"Source: %s\n"+
|
||||
"Target: %s\n"+
|
||||
@@ -348,12 +378,18 @@ func (m *restoreModel) View() string {
|
||||
duration.Round(time.Millisecond))
|
||||
}
|
||||
|
||||
// Show status message before progress starts
|
||||
if m.processed == 0 && (m.isNetworkSource || m.statusMessage != "") {
|
||||
if m.processed == 0 {
|
||||
return m.statusMessage + "\n\nPress Ctrl+C to cancel"
|
||||
}
|
||||
|
||||
title := fmt.Sprintf("Restoring from %s → %s", m.source, m.target)
|
||||
operation := "Restoring"
|
||||
if m.httpSourceInfo != nil {
|
||||
operation = "Downloading"
|
||||
} else if m.isP2PSource {
|
||||
operation = "Receiving"
|
||||
}
|
||||
|
||||
title := fmt.Sprintf("%s from %s → %s", operation, m.source, m.target)
|
||||
progressVal := float64(m.processed) / float64(m.totalBytes)
|
||||
progressView := m.progress.ViewAs(progressVal)
|
||||
stats := fmt.Sprintf("%s / %s (%.1f%%)", formatBytes(m.processed), formatBytes(m.totalBytes), progressVal*100)
|
||||
@@ -385,7 +421,7 @@ func isBlockDevice(path string) bool {
|
||||
|
||||
func init() {
|
||||
rootCmd.AddCommand(restoreCmd)
|
||||
restoreCmd.Flags().StringVarP(&restoreSource, "source", "s", "", "Source archive or network address (e.g., /path/to/archive.tar.gz or 0.0.0.0:8080) (required)")
|
||||
restoreCmd.Flags().StringVarP(&restoreTarget, "target", "t", "", "Target file, disk, or directory. Required for network sources.")
|
||||
restoreCmd.Flags().StringVarP(&restoreSource, "source", "s", "", "Source archive, P2P address (e.g., 0.0.0.0:8080), or HTTP URL (e.g., 127.0.0.1:8080/backup.gz) (required)")
|
||||
restoreCmd.Flags().StringVarP(&restoreTarget, "target", "t", "", "Target file, disk, or directory. Required for non-directory restores.")
|
||||
restoreCmd.MarkFlagRequired("source")
|
||||
}
|
||||
Reference in new issue
Block a user