server.go

package main

import (
	"bytes"
	"encoding/gob"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"

	"github.com/girish/storage/p2p"
	"github.com/girish/storage/store"
)

type FileServerOpts struct {
	StorageRoot string //StorageRoot is the folder on hard drive where this node will save files
	HTTPAddr    string
	Peers       []string
	Transport   p2p.Transport
	Store       *store.Store
}

// DataMessage is wire protocol payload!
type DataMessage struct {
	Key  string
	Data []byte
}

type FileServer struct {
	FileServerOpts
	quitCh chan struct{}
}

func NewFileServer(opts FileServerOpts) *FileServer {
	return &FileServer{
		FileServerOpts: opts,
		quitCh:         make(chan struct{}),
	}
}

// Start boots up the transport layer and begins processing messages
func (s *FileServer) Start() error {
	fmt.Printf("Starting File Server... (Storage Root: %s)\n", s.StorageRoot)

	if err := s.Transport.ListenAndAccept(); err != nil {
		return err
	}
	go func() {
		fmt.Printf("HTTP API listening on %s\n", s.HTTPAddr)
		if err := http.ListenAndServe(s.HTTPAddr, s.httpHandler()); err != nil {
			fmt.Printf("HTTP server error: %s\n", err)
		}
	}()

	s.loop()
	return nil
}

func (s *FileServer) httpHandler() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("PUT /files/{key}", s.handleHTTPPut)
	return mux
}

func (s *FileServer) handleHTTPPut(w http.ResponseWriter, r *http.Request) {
	key := r.PathValue("key")
	if key == "" {
		http.Error(w, "file key is required", http.StatusBadRequest)
		return
	}
	data, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "failed to read upload", http.StatusBadRequest)
		return
	}
	if err := s.Store.WriteStream(key, bytes.NewReader(data)); err != nil {
		http.Error(w, "failed to store file", http.StatusInternalServerError)
		return
	}
	if r.Header.Get("X-DFS-Replication") != "true" {
		client := &http.Client{Timeout: 10 * time.Second}
		for _, peer := range s.Peers {
			peer = strings.TrimRight(peer, "/")
			peerURL := peer + "/files/" + url.PathEscape(key)
			request, err := http.NewRequestWithContext(r.Context(), http.MethodPut, peerURL, bytes.NewReader(data))
			if err != nil {
				http.Error(w, "invalid peer URL", http.StatusBadGateway)
				return
			}
			request.Header.Set("X-DFS-Replication", "true")
			response, err := client.Do(request)
			if err != nil {
				http.Error(w, "failed to replicate to peer "+peer, http.StatusBadGateway)
				return
			}
			response.Body.Close()
			if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices {
				http.Error(w, "peer rejected replicated upload: "+peer, http.StatusBadGateway)
				return
			}
		}
	}
	w.WriteHeader(http.StatusCreated)
}

func (s *FileServer) loop() {
	for {
		select {
		case rpc := <-s.Transport.Consume():
			s.handleMessage(rpc)
		case <-s.quitCh:
			fmt.Println("File Server shutting down")
			return
		}
	}
}

func (s *FileServer) handleMessage(rpc p2p.RPC) {
	//1 Decode the binary network payload back into Structured DataMessage
	var msg DataMessage
	if err := gob.NewDecoder(bytes.NewReader(rpc.Payload)).Decode(&msg); err != nil {
		fmt.Printf("Failed to decode network payload: %s\n", err)
		return
	}
	fmt.Printf("FileServer received command to store file: '%s' (%d bytes)\n", msg.Key, len(msg.Data))

	//2 Write the file to disk using CAS store engine
	err := s.Store.WriteStream(msg.Key, bytes.NewReader(msg.Data))
	if err != nil {
		fmt.Printf("Error Storing file to disk: %s\n", err)
		return
	}
	fmt.Println("File successfully saved from the network!")
}