Server Transfers

type Transfer struct {
    ctx    context.Context
    cancel *context.CancelFunc
    
    Server  *server.Server
    status  *system.Atomic[Status]
    archive *Archive
}

const (
    StatusPending     Status = "pending"
    StatusProcessing  Status = "processing"
    StatusCancelling  Status = "cancelling"
    StatusCancelled   Status = "cancelled"
    StatusFailed      Status = "failed"
    StatusCompleted   Status = "completed"
)

POST /api/servers/{server}/transfer

func New(ctx context.Context, s *server.Server) *Transfer {
    ctx, cancel := context.WithCancel(ctx)
    
    return &Transfer{
        ctx:    ctx,
        cancel: &cancel,
        Server: s,
        status: system.NewAtomic(StatusPending),
    }
}

type Archive struct {
    archive *filesystem.Archive
}

func NewArchive(t *Transfer, size uint64) *Archive {
    return &Archive{
        archive: &filesystem.Archive{
            Filesystem: t.Server.Filesystem(),
            Progress:   progress.NewProgress(size),
        },
    }
}

func (t *Transfer) Archive() (*Archive, error) {
    if t.archive == nil {
        // Get server disk usage for progress calculation
        rawSize, err := t.Server.Filesystem().DiskUsage(true)
        if err != nil {
            return nil, fmt.Errorf("transfer: failed to get server disk usage: %w", err)
        }
        
        // Create archive instance
        t.archive = NewArchive(t, uint64(rawSize))
    }
    
    return t.archive, nil
}

func (a *Archive) Stream(ctx context.Context, w io.Writer) error {
    return a.archive.Stream(ctx, w)
}

func (t *Transfer) PushArchiveToTarget(url, token string) ([]byte, error) {
    // 1. Set processing status
    t.SetStatus(StatusProcessing)
    
    // 2. Get archive
    a, _ := t.Archive()
    
    // 3. Create HTTP request
    body, writer := io.Pipe()
    req, _ := http.NewRequestWithContext(ctx, http.MethodPost, url, body)
    req.Header.Set("Authorization", token)
    
    // 4. Create multipart writer
    mp := multipart.NewWriter(writer)
    req.Header.Set("Content-Type", mp.FormDataContentType())
    
    // 5. Stream archive in goroutine
    go func() {
        src, pw := io.Pipe()
        h := sha256.New()
        tee := io.TeeReader(src, h)
        
        // Create form file
        dest, _ := mp.CreateFormFile("archive", "archive.tar.gz")
        
        // Copy with checksum calculation
        go io.Copy(dest, tee)
        
        // Stream archive to pipe
        a.Stream(ctx, pw)
        
        // Add checksum field
        mp.WriteField("checksum", hex.EncodeToString(h.Sum(nil)))
        mp.Close()
    }()
    
    // 6. Send request
    client := http.Client{Timeout: 0} // No timeout
    res, _ := client.Do(req)
    
    return io.ReadAll(res.Body)
}

go func(ctx context.Context, p *progress.Progress, tc *time.Ticker) {
    defer tc.Stop()
    
    for {
        select {
        case <-ctx.Done():
            return
        case <-tc.C:
            t.SendMessage("Uploading " + p.Progress(25))
        }
    }
}(ctx2, a.Progress(), time.NewTicker(5*time.Second))

[Transfer System] [Source Node]: Uploading [=========>          ] 45.2%

req.Header.Set("Authorization", token)

mp := multipart.NewWriter(writer)
req.Header.Set("Content-Type", mp.FormDataContentType())

// Archive file
dest, err := mp.CreateFormFile("archive", "archive.tar.gz")

// Checksum field
mp.WriteField("checksum", hex.EncodeToString(h.Sum(nil)))

POST /api/transfers

func (t *Transfer) Cancel() {
    status := t.Status()
    
    // Don't cancel if already in final state
    if status == StatusCancelling ||
        status == StatusCancelled ||
        status == StatusCompleted ||
        status == StatusFailed {
        return
    }
    
    if t.cancel == nil {
        return
    }
    
    t.SetStatus(StatusCancelling)
    (*t.cancel)()
}

DELETE /api/servers/{server}/transfer

DELETE /api/transfers/{server}

func (t *Transfer) Status() Status {
    return t.status.Load()
}

func (t *Transfer) SetStatus(s Status) {
    t.status.Store(s)
    t.Server.Events().Publish(server.TransferStatusEvent, s)
}

func (t *Transfer) SendMessage(v string) {
    t.Server.Events().Publish(
        server.TransferLogsEvent,
        colorstring.Color("[yellow][bold]"+time.Now().Format(time.RFC1123)+" [Transfer System] [Source Node]:[default] "+v),
    )
}

[yellow][bold]Mon, 02 Jan 2006 15:04:05 MST [Transfer System] [Source Node]:[default] Preparing to stream server data to destination...

func (t *Transfer) Error(err error, v string) {
    t.Log().WithError(err).Error(v)
    t.SendMessage(v)
}

func (t *Transfer) Log() *log.Entry {
    if t.Server == nil {
        return log.WithField("subsystem", "transfer")
    }
    return t.Server.Log().WithField("subsystem", "transfer")
}

func (t *Transfer) Context() context.Context {
    return t.ctx
}

t.Server.Events().Publish(server.TransferStatusEvent, s)

t.Server.Events().Publish(server.TransferLogsEvent, message)

func (s *Server) IsTransferring() bool {
    return s.transferring.Load()
}

func (s *Server) SetTransferring(state bool) {
    s.transferring.Store(state)
}

func (s *Server) HandlePowerAction(action PowerAction, waitSeconds ...int) error {
    if s.IsTransferring() {
        return ErrServerIsTransferring
    }
}

client := http.Client{Timeout: 0}

POST /api/transfers HTTP/1.1
Authorization: {token}
Content-Type: multipart/form-data; boundary={boundary}

h := sha256.New()
tee := io.TeeReader(src, h)

// ... copy data through tee reader ...

checksum := hex.EncodeToString(h.Sum(nil))
mp.WriteField("checksum", checksum)

if res.StatusCode != http.StatusOK {
    return nil, fmt.Errorf("unexpected status code from destination: %d", res.StatusCode)
}