func (srv *Server) CreateFile(ctx context.Context, file *pb.File) (*pb.FileID, error) {
	name := file.Name
	if name == "" {
		return nil, twirp.RequiredArgumentError("name")
	}
	name = cleanFilename(name)
	if err := validateFilename(name); err != nil {
		return nil, twirp.InvalidArgumentError("name", err.Error())
	}
	var hasPrev bool
	prevInfo, err := srv.db.GetLatestFileVersion(name)
	if errors.Is(err, db.ErrNotFound) {
		hasPrev = false
	} else if err != nil {
		return nil, fmt.Errorf("db GetLatestFileVersion: %w", err)
	} else {
		hasPrev = true
	}
	chunks := make([]object.Chunk, len(file.Sums))
	for i, s := range file.Sums {
		sum, err := sum.FromBytes(s)
		if err != nil {
			msg := fmt.Sprintf("sum %d: %v", i, err)
			return nil, twirp.InvalidArgumentError("sums", msg)
		}
		size, err := srv.db.GetChunkSize(sum)
		if errors.Is(err, db.ErrNotFound) {
			msg := fmt.Sprintf("sum %d %x does not exist", i, sum)
			return nil, twirp.NewError(twirp.FailedPrecondition, msg)
		} else if err != nil {
			return nil, err
		}
		chunks[i] = object.Chunk{Sequence: uint64(i), Size: size, Sum: sum}
	}
	f := object.File{Name: name, Chunks: chunks, CreatedAt: time.Now().UTC(), Versioned: srv.cfg.VersioningEnabled}
	b := f.MarshalBinary()
	sum := sum.Compute(b)
	fkey := sum.AsHex() + ".file"
	if err := srv.store.Put(ctx, srv.cfg.Bucket, fkey, bytes.NewReader(b)); err != nil {
		return nil, err
	}
	if err := srv.db.InsertFile(f, sum); err != nil {
		err = mergeErrors(err, srv.store.Delete(srv.cfg.Bucket, fkey))
		return nil, err
	}
	if hasPrev && !prevInfo.Versioned && !srv.cfg.VersioningEnabled {
		if _, err = srv.Delete(ctx, &pb.FileID{Sum: prevInfo.Sum[:]}); err != nil {
			log.Error(err)
		}
	}
	return &pb.FileID{Sum: sum[:]}, nil
}
CreateFile creates a new file. Returns an error if any chunk referenced by the file
does not exist.
package server
import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"path"
	"strings"
	"sync/atomic"
	"time"
	"github.com/rs/zerolog"
	"github.com/twitchtv/twirp"
	"golang.org/x/sync/errgroup"
	"github.com/jotfs/jotfs/internal/db"
	"github.com/jotfs/jotfs/internal/log"
	"github.com/jotfs/jotfs/internal/object"
	pb "github.com/jotfs/jotfs/internal/protos"
	"github.com/jotfs/jotfs/internal/store"
	"github.com/jotfs/jotfs/internal/sum"
)
const maxFilenameSize = 1024
const (
	stateNotVacuuming int32 = iota
	stateVacuuming
)
type Config struct {
	Bucket string
	VersioningEnabled bool
	MaxChunkSize uint64
	MaxPackfileSize uint64
	DownloadTimeout time.Duration
	Params ChunkerParams
}
type ChunkerParams struct {
	MinChunkSize  uint `json:"min_chunk_size"`
	AvgChunkSize  uint `json:"avg_chunk_size"`
	MaxChunkSize  uint `json:"max_chunk_size"`
	Normalization uint `json:"normalization"`
}
type Server struct {
	db          *db.Adapter
	store       store.Store
	cfg         Config
	logger      zerolog.Logger
	isVacuuming int32
}
func New(db *db.Adapter, s store.Store, cfg Config) *Server {
	logger := zerolog.New(ioutil.Discard).Level(zerolog.Disabled)
	return &Server{db: db, cfg: cfg, store: s, logger: logger}
}
func (srv *Server) SetLogger(logger zerolog.Logger) {
	srv.logger = logger
}
func (srv *Server) PackfileUploadHandler(w http.ResponseWriter, req *http.Request) {
	if req.ContentLength <= 0 {
		http.Error(w, "content-length required", http.StatusBadRequest)
		return
	}
	if req.ContentLength > int64(srv.cfg.MaxPackfileSize) {
		http.Error(w, "content-length exceeds maximum packfile size", http.StatusBadRequest)
		return
	}
	h := req.Header.Get("x-jotfs-checksum")
	if h == "" {
		http.Error(w, "x-jotfs-checksum required", http.StatusBadRequest)
		return
	}
	sum, err := sum.FromBase64(h)
	if err != nil {
		msg := fmt.Sprintf("invalid x-jotfs-checksum: %v", err)
		http.Error(w, msg, http.StatusBadRequest)
		return
	}
	digest := sum.AsHex()
	pkey := digest + ".pack"
	bucket := srv.cfg.Bucket
	ctx, cancel := context.WithCancel(req.Context())
	defer cancel()
	r, pfile := io.Pipe()
	var g errgroup.Group
	g.Go(func() error {
		err := srv.store.Put(ctx, bucket, pkey, r)
		return mergeErrors(err, r.CloseWithError(err))
	})
	rd := io.TeeReader(io.LimitReader(req.Body, req.ContentLength), pfile)
	index, err := object.LoadPackIndex(rd)
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	if index.Sum != sum {
		msg := fmt.Sprintf("provided packfile checksum %x does not match actual checksum %x", sum, index.Sum)
		http.Error(w, msg, http.StatusBadRequest)
		return
	}
	if err = pfile.Close(); err != nil {
		err = fmt.Errorf("closing packfile writer: %w", err)
		internalError(w, err)
		return
	}
	if err = g.Wait(); err != nil {
		err = fmt.Errorf("uploading packfile to store: %w", err)
		internalError(w, err)
		return
	}
	ikey := digest + ".index"
	b := index.MarshalBinary()
	if err = srv.store.Put(ctx, bucket, ikey, bytes.NewReader(b)); err != nil {
		err = mergeErrors(err, srv.store.Delete(bucket, pkey))
		internalError(w, err)
		return
	}
	createdAt := time.Now().UTC()
	if err = srv.db.InsertPackIndex(index, createdAt); err != nil {
		err = mergeErrors(err, srv.store.Delete(bucket, pkey))
		err = mergeErrors(err, srv.store.Delete(bucket, ikey))
		internalError(w, err)
		return
	}
	w.WriteHeader(http.StatusCreated)
}