Zippytal-Node/p2p_fs_datachannel_manager.go

package localserver

import (
	"bufio"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"os"
	"sync"
	"time"

	"github.com/pion/webrtc/v3"
)

const (
	HOME = "home"
)

const (
	FS_GET_FOLDER                  = "fs_get_folder"
	FS_GET_FOLDER_RESPONSE         = "fs_get_folder_response"
	FS_UPLOAD_FILE                 = "fs_upload_file"
	FS_UPLOAD_FILE_INIT            = "fs_upload_file_init"
	FS_UPLOAD_FILE_END             = "fs_upload_file_response_end"
	FS_DOWNLOAD_FILE               = "fs_download_file"
	FS_DOWNLOAD_FILE_RESPONSE      = "fs_download_file_response"
	FS_DOWNLOAD_FILE_RESPONSE_INIT = "fs_download_file_response_init"
	FS_DOWNLOAD_FILE_RESPONSE_END  = "fs_download_file_response_end"
	FS_CREATE_FOLDER               = "fs_create_folder"
)

type DataChannelManager interface {
	HandleMessage(message *DatachannelMessage, channel *webrtc.DataChannel) (done chan struct{}, errChan chan error)
}

type P2PFSDatachannelManager struct {
	l          *sync.Mutex
	uploadFile map[string]*os.File
}

type DatachannelMessage struct {
	From    string                     `json:"from"`
	Type    string                     `json:"type"`
	Payload *DatachannelMessagePayload `json:"payload"`
}

type DatachannelMessagePayload struct {
	Path    string
	Content []byte
}

type FSResultInfo struct {
	Name    string
	Type    string
	Path    string
	Size    int64
	ModTime string
}

func NewP2PFSDatachannelManager() (manager *P2PFSDatachannelManager) {
	manager = &P2PFSDatachannelManager{&sync.Mutex{}, make(map[string]*os.File)}
	return
}

func (pdm *P2PFSDatachannelManager) HandleMessage(message *DatachannelMessage, channel *webrtc.DataChannel) (done chan struct{}, errChan chan error) {
	done, errChan = make(chan struct{}), make(chan error)
	go func() {
		switch message.Type {
		case FS_GET_FOLDER:
			// logger.Println("try to get folder")

			done, err := pdm.handleGetFolder(message.Payload.Path, channel)
			select {
			case <-done:
				// logger.Println("operation succeed")
			case e := <-err:
				errChan <- e
				return
			}
		case FS_DOWNLOAD_FILE:
			// logger.Println("tried to download a file")
			done, err := pdm.sendFile(message.Payload.Path, channel)
			select {
			case <-done:
				// logger.Println("operation succeed")
			case e := <-err:
				errChan <- e
				return
			}
		case FS_UPLOAD_FILE_INIT:
			// logger.Println("tried to upload a file")
			if err := pdm.downloadFileInit(message.Payload.Path); err != nil {
				errChan <- err
				return
			}
		case FS_UPLOAD_FILE:
			if err := pdm.downloadFile(message.Payload.Path, message.Payload.Content); err != nil {
				errChan <- err
				return
			}
		case FS_UPLOAD_FILE_END:
			if err := pdm.downloadFileEnd(message.Payload.Path); err != nil {
				errChan <- err
				return
			}
		case FS_CREATE_FOLDER:
			if err := pdm.createFolder(message.Payload.Path); err != nil {
				errChan <- err
				return
			}
		default:
			// logger.Printf("got a new message from %s with payload %v\n", message.From, message.Payload)
		}
		done <- struct{}{}
	}()
	return done, errChan
}

func (pdm *P2PFSDatachannelManager) handleGetFolder(path string, channel *webrtc.DataChannel) (done chan struct{}, errCh chan error) {
	done, errCh = make(chan struct{}), make(chan error)
	go func() {
		var dirPath string
		if path == HOME {
			homeDir, err := os.UserHomeDir()
			// logger.Println(homeDir)
			if err != nil {
				errCh <- err
				return
			}
			dirPath = homeDir
		} else {
			dirPath = path
		}
		dir, err := os.ReadDir(dirPath)
		if err != nil {
			errCh <- err
			return
		}
		index := 1
		for i, v := range dirPath {
			if i > 0 {
				if v == '/' {
					index = i
				}
			}
		}
		dirs := []*FSResultInfo{{Name: "..", Type: "directory", Path: dirPath[:index], Size: 0, ModTime: time.Now().Format("January 2, 2006")}}
		for _, d := range dir {
			f := &FSResultInfo{}
			info, err := d.Info()
			if err != nil {
				// logger.Println(err)
				continue
			}
			f.ModTime = fmt.Sprintf("%s at %s", info.ModTime().Format("January 2, 2006"), info.ModTime().Format("15:04:05"))
			f.Path = fmt.Sprintf("%s/%s", dirPath, info.Name())
			f.Size = info.Size()
			f.Name = info.Name()
			if info.IsDir() {
				f.Type = "directory"
			} else {
				f.Type = "file"
			}
			dirs = append(dirs, f)
		}

		if len(dirs) < 100 {
			// logger.Println("dir smaller than 100")
			bs, err := json.Marshal(map[string]interface{}{
				"type": FS_GET_FOLDER_RESPONSE,
				"from": "lolo_local_serv",
				"payload": map[string]interface{}{
					"folderPath": dirPath,
					"dirContent": dirs,
				},
			})
			if err != nil {
				errCh <- err
				return
			}
			if err := channel.SendText(string(bs)); err != nil {
				errCh <- err
				return
			}
		} else {
			// logger.Println("dir greater than 10000")
			reste := len(dirs) % 100
			x := (len(dirs) - reste) / 100
			for j := 0; j < x; j++ {
				// logger.Println("dir sending packet than 100")
				d := dirs[j*100 : (j+1)*100]
				// logger.Println("length of d :", len(d))
				bs, err := json.Marshal(map[string]interface{}{
					"type": FS_GET_FOLDER_RESPONSE,
					"from": "lolo_local_serv",
					"payload": map[string]interface{}{
						"folderPath": dirPath,
						"dirContent": d,
					},
				})
				if err != nil {
					errCh <- err
					return
				}
				if err := channel.SendText(string(bs)); err != nil {
					errCh <- err
					return
				}
			}
			bs, err := json.Marshal(map[string]interface{}{
				"type": FS_GET_FOLDER_RESPONSE,
				"from": "lolo_local_serv",
				"payload": map[string]interface{}{
					"folderPath": dirPath,
					"dirContent": dirs[len(dirs)-reste:],
				},
			})
			if err != nil {
				errCh <- err
				return
			}
			if err := channel.SendText(string(bs)); err != nil {
				errCh <- err
				return
			}
		}
		done <- struct{}{}
	}()
	return
}

func (pdm *P2PFSDatachannelManager) sendFile(path string, channel *webrtc.DataChannel) (<-chan struct{}, <-chan error) {
	done, errCh := make(chan struct{}), make(chan error)
	go func() {
		f, err := os.Open(path)
		if err != nil {
			errCh <- err
			return
		}
		bsInit, err := json.Marshal(map[string]interface{}{
			"type": FS_DOWNLOAD_FILE_RESPONSE_INIT,
			"from": "lolo_local_serv",
			"payload": map[string]string{
				"path": path,
			},
		})
		if err != nil {
			errCh <- err
			return
		}
		if err := channel.SendText(string(bsInit)); err != nil {
			errCh <- err
			return
		}
		r := bufio.NewReader(f)
		buf := make([]byte, 0, 10000)
		for {
			n, err := r.Read(buf[:cap(buf)])
			buf = buf[:n]
			if n == 0 {
				if err == nil {
					continue
				}
				if err == io.EOF {
					break
				}
				log.Fatal(err)
			}
			bs, err := json.Marshal(map[string]interface{}{
				"type": FS_DOWNLOAD_FILE_RESPONSE,
				"from": "lolo_local_serv",
				"payload": map[string]interface{}{
					"path":    path,
					"content": buf,
				},
			})
			if err != nil {
				errCh <- err
				return
			}
			if err := channel.SendText(string(bs)); err != nil {
				errCh <- err
				return
			}
		}
		bs, err := json.Marshal(map[string]interface{}{
			"type": FS_DOWNLOAD_FILE_RESPONSE_END,
			"from": "lolo_local_serv",
			"payload": map[string]string{
				"path": path,
			},
		})
		if err != nil {
			errCh <- err
			return
		}
		if err := channel.SendText(string(bs)); err != nil {
			errCh <- err
			return
		}
		done <- struct{}{}
	}()
	return done, errCh
}

func (pdm *P2PFSDatachannelManager) downloadFileInit(path string) (err error) {
	// logger.Println("upload path name is", path)
	if _, ok := pdm.uploadFile[path]; ok {
		err = fmt.Errorf("the file %s is already being uploaded", path)
		return
	}
	file, e := os.Create(path)
	if e != nil {
		err = fmt.Errorf("an error occured in download file init : %v", e)
		return
	}
	index := 1
	for i, v := range path {
		if i > 0 {
			if v == '/' {
				index = i
			}
		}
	}
	// logger.Println(path)
	// logger.Println(path[index+1:])
	pdm.uploadFile[path[index+1:]] = file
	return
}

func (pdm *P2PFSDatachannelManager) downloadFile(path string, content []byte) (err error) {
	if _, ok := pdm.uploadFile[path]; !ok {
		err = fmt.Errorf("no upload file open for path %s", path)
		return
	}
	_, err = pdm.uploadFile[path].Write(content)
	return
}

func (pdm *P2PFSDatachannelManager) downloadFileEnd(path string) (err error) {
	// logger.Println("closing file")
	if _, ok := pdm.uploadFile[path]; !ok {
		err = fmt.Errorf("no upload file open for path %s", path)
		return
	}
	err = pdm.uploadFile[path].Close()
	delete(pdm.uploadFile, path)
	return
}

func (pdm *P2PFSDatachannelManager) createFolder(path string) (err error) {
	err = os.Mkdir(path, 0770)
	return
}