feat: [OCISDEV-906] implement the Kiteworks write path, versions and locking (#739)

Author: LarsJurgensen

pkg/storage/fs/kiteworks/kiteworks.go

@@ -2,6 +2,7 @@ package kwlib
 
 import (
 	"bytes"
+	"context"
 	"crypto/tls"
 	"encoding/json"
 	"errors"
@@ -18,6 +19,8 @@ import (
 	"github.com/rs/zerolog"
 )
 
+const uploadChunkSize = 8 << 20 // 8 MiB
+
 func NewClientFactory(server, agentString string, insecure bool) *APIClientFactory {
 	transport := &http.Transport{
 		Proxy: http.ProxyFromEnvironment,
@@ -36,28 +39,35 @@ func NewClientFactory(server, agentString string, insecure bool) *APIClientFacto
 		// #nosec
 		transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
 	}
+
+	uploadTransport := transport.Clone()
+	uploadTransport.ResponseHeaderTimeout = 30 * time.Second
+
 	return &APIClientFactory{
-		server:      server,
-		agentString: agentString,
-		httpClient:  &http.Client{Transport: transport, Timeout: 15 * time.Second},
+		server:       server,
+		agentString:  agentString,
+		httpClient:   &http.Client{Transport: transport, Timeout: 15 * time.Second},
+		uploadClient: &http.Client{Transport: uploadTransport},
 	}
 }
 
 type APIClientFactory struct {
-	server      string
-	agentString string
-	httpClient  *http.Client
+	server       string
+	agentString  string
+	httpClient   *http.Client
+	uploadClient *http.Client
 }
 
 type APIClient struct {
-	server      string
-	agentString string
-	logger      *zerolog.Logger
-	host        string
-	token       string
-	requestId   string
-	remoteAddr  string
-	httpClient  *http.Client
+	server       string
+	agentString  string
+	logger       *zerolog.Logger
+	host         string
+	token        string
+	requestId    string
+	remoteAddr   string
+	httpClient   *http.Client
+	uploadClient *http.Client
 }
 
 func decodeJSON(body io.ReadCloser, out any) error {
@@ -67,13 +77,14 @@ func decodeJSON(body io.ReadCloser, out any) error {
 
 func (f *APIClientFactory) Build(host, requestId, remoteAddr, token string, l *zerolog.Logger) *APIClient {
 	return &APIClient{
-		token:      token,
-		server:     f.server,
-		host:       host,
-		logger:     l,
-		requestId:  requestId,
-		remoteAddr: remoteAddr,
-		httpClient: f.httpClient,
+		token:        token,
+		server:       f.server,
+		host:         host,
+		logger:       l,
+		requestId:    requestId,
+		remoteAddr:   remoteAddr,
+		httpClient:   f.httpClient,
+		uploadClient: f.uploadClient,
 	}
 }
 
@@ -94,7 +105,7 @@ func (c *APIClient) GetTopFolders() (*DirectoryInfo, error) {
 }
 
 func (c *APIClient) GetFolderByID(id string) (*FileInfo, error) {
-	request, err := c.NewGetRequest(fmt.Sprintf("/rest/folders/%s", id))
+	request, err := c.NewGetRequest(fmt.Sprintf("/rest/folders/%s?with=(permissions)", id))
 	if err != nil {
 		return nil, err
 	}
@@ -148,7 +159,7 @@ func (c *APIClient) Search(path string) (*FileInfo, error) {
 }
 
 func (c *APIClient) GetFileByID(id string) (*FileInfo, error) {
-	request, err := c.NewGetRequest(fmt.Sprintf("/rest/files/%s", id))
+	request, err := c.NewGetRequest(fmt.Sprintf("/rest/files/%s?with=(permissions,lockUser)", id))
 	if err != nil {
 		return nil, err
 	}
@@ -194,16 +205,16 @@ func (c *APIClient) GetUser(id string) (*User, error) {
 	return out, nil
 }
 
-func (c *APIClient) GetQuotaInfo() (*QuotaInfo, error) {
-	request, err := c.NewGetRequest("/rest/quotas")
+func (c *APIClient) GetFolderQuota(folderID string) (*FolderQuota, error) {
+	request, err := c.NewGetRequest(fmt.Sprintf("/rest/folders/%s/quota", folderID))
 	if err != nil {
 		return nil, err
 	}
 	response, err := c.SendRequest(request)
 	if err != nil {
 		return nil, err
 	}
-	out := &QuotaInfo{}
+	out := &FolderQuota{}
 	if err := decodeJSON(response.Body, out); err != nil {
 		return nil, err
 	}
@@ -251,12 +262,20 @@ func (c *APIClient) CreateFolder(id string, payload CreateDirRequest) (string, e
 }
 
 func (c *APIClient) InitializeUpload(parentID, name string, size int64, numberOfChunks int) (*UploadResult, error) {
+	return c.initializeUpload(fmt.Sprintf("/rest/folders/%s/actions/initiateUpload", parentID), name, size, numberOfChunks)
+}
+
+func (c *APIClient) InitializeVersionUpload(fileID, name string, size int64, numberOfChunks int) (*UploadResult, error) {
+	return c.initializeUpload(fmt.Sprintf("/rest/files/%s/actions/initiateUpload", fileID), name, size, numberOfChunks)
+}
+
+func (c *APIClient) initializeUpload(path, name string, size int64, numberOfChunks int) (*UploadResult, error) {
 	payload := InitializeUpload{
 		FileName:    name,
 		TotalSize:   size,
 		TotalChunks: numberOfChunks,
 	}
-	request, err := c.NewPostRequest(fmt.Sprintf("/rest/folders/%s/actions/initiateUpload", parentID), payload)
+	request, err := c.NewPostRequest(path, payload)
 	if err != nil {
 		return nil, err
 	}
@@ -271,7 +290,20 @@ func (c *APIClient) InitializeUpload(parentID, name string, size int64, numberOf
 	return out, nil
 }
 
-func (c *APIClient) UploadChunk(uploadURI, name string, file io.Reader, chunkIndex int, chunk int64, isLastChunk bool) (*FileInfo, error) {
+func (c *APIClient) TerminateUpload(uploadID int64) error {
+	request, err := c.newRequest("DELETE", fmt.Sprintf("/rest/uploads/%d", uploadID), nil)
+	if err != nil {
+		return err
+	}
+	response, err := c.SendRequest(request)
+	if err != nil {
+		return err
+	}
+	response.Body.Close()
+	return nil
+}
+
+func (c *APIClient) UploadChunk(ctx context.Context, uploadURI, name string, file io.Reader, chunkIndex int, chunk int64, isLastChunk bool) (*FileInfo, error) {
 	body := new(bytes.Buffer)
 	writer := multipart.NewWriter(body)
 	part, err := writer.CreateFormFile("content", name)
@@ -296,13 +328,14 @@ func (c *APIClient) UploadChunk(uploadURI, name string, file io.Reader, chunkInd
 	if err != nil {
 		return nil, err
 	}
+	request = request.WithContext(ctx)
 	request.Header.Set("Content-Type", writer.FormDataContentType())
 	if isLastChunk {
 		q := request.URL.Query()
 		q.Add("returnEntity", "true")
 		request.URL.RawQuery = q.Encode()
 	}
-	response, err := c.SendRequest(request)
+	response, err := c.sendWith(c.uploadClient, request)
 	if err != nil {
 		return nil, err
 	}
@@ -313,10 +346,121 @@ func (c *APIClient) UploadChunk(uploadURI, name string, file io.Reader, chunkInd
 		}
 		return out, nil
 	}
+	// drained so the connection can be reused across chunks
+	_, _ = io.Copy(io.Discard, response.Body)
 	response.Body.Close()
 	return nil, nil
 }
 
+func (c *APIClient) MoveFolder(id, destinationFolderID string) error {
+	request, err := c.NewPostRequest(
+		fmt.Sprintf("/rest/folders/%s/actions/move", id),
+		MoveFolderRequest{DestinationFolderID: destinationFolderID},
+	)
+	if err != nil {
+		return err
+	}
+	_, err = c.SendRequest(request)
+	return err
+}
+
+// name must be the file's current name: the server requires it even for a version upload.
+func (c *APIClient) UploadFileVersion(ctx context.Context, fileID, name string, body io.Reader, length int64) error {
+	chunks := chunkCount(length)
+	session, err := c.InitializeVersionUpload(fileID, name, length, chunks)
+	if err != nil {
+		return err
+	}
+	for i := 0; i < chunks; i++ {
+		size := int64(uploadChunkSize)
+		if remaining := length - int64(i)*uploadChunkSize; remaining < size {
+			size = remaining
+		}
+		if _, err := c.UploadChunk(ctx, session.URI, name, body, i, size, i == chunks-1); err != nil {
+			if termErr := c.TerminateUpload(session.ID); termErr != nil {
+				c.logger.Warn().Err(termErr).Int64("uploadID", session.ID).Msg("could not terminate kiteworks upload session")
+			}
+			return err
+		}
+	}
+	return nil
+}
+
+func chunkCount(length int64) int {
+	if length <= 0 {
+		return 1
+	}
+	return int((length + uploadChunkSize - 1) / uploadChunkSize)
+}
+
+func (c *APIClient) GetFileVersions(fileID string) ([]Version, error) {
+	req, err := c.NewGetRequest(fmt.Sprintf("/rest/files/%s/versions", fileID))
+	if err != nil {
+		return nil, err
+	}
+	resp, err := c.SendRequest(req)
+	if err != nil {
+		return nil, err
+	}
+	out := &VersionList{}
+	if err := decodeJSON(resp.Body, out); err != nil {
+		return nil, err
+	}
+	return out.Data, nil
+}
+
+func (c *APIClient) DeleteFileVersion(fileID, versionID string) error {
+	req, err := c.newRequest("DELETE", fmt.Sprintf("/rest/files/%s/versions/%s", fileID, versionID), nil)
+	if err != nil {
+		return err
+	}
+	_, err = c.SendRequest(req)
+	return err
+}
+
+func (c *APIClient) PromoteFileVersion(fileID, versionID string) error {
+	req, err := c.newRequest("POST", fmt.Sprintf("/rest/files/%s/versions/%s/actions/promote", fileID, versionID), nil)
+	if err != nil {
+		return err
+	}
+	_, err = c.SendRequest(req)
+	return err
+}
+
+func (c *APIClient) GetVersionContents(fileID, versionID string) (*http.Response, error) {
+	req, err := c.NewGetRequest(fmt.Sprintf("/rest/files/%s/versions/%s/content", fileID, versionID))
+	if err != nil {
+		return nil, err
+	}
+	return c.SendRequest(req)
+}
+
+func (c *APIClient) LockFile(fileID string) error {
+	req, err := c.newRequest("PATCH", fmt.Sprintf("/rest/files/%s/actions/lock", fileID), nil)
+	if err != nil {
+		return err
+	}
+	resp, err := c.SendRequest(req)
+	if err != nil {
+		return err
+	}
+	resp.Body.Close()
+	return nil
+}
+
+func (c *APIClient) UnlockFile(fileID string) error {
+	req, err := c.newRequest("PATCH", fmt.Sprintf("/rest/files/%s/actions/unlock", fileID), nil)
+	if err != nil {
+		return err
+	}
+	resp, err := c.SendRequest(req)
+	if err != nil {
+		return err
+	}
+	resp.Body.Close()
+	return nil
+}
+
 func (c *APIClient) DeleteFolder(id string) error {
 	request, err := c.newRequest("DELETE", fmt.Sprintf("/rest/folders/%s", id), nil)
 	if err != nil {
@@ -432,7 +576,11 @@ func (c *APIClient) newRequest(method, path string, body io.Reader) (*http.Reque
 }
 
 func (c *APIClient) SendRequest(req *http.Request) (*http.Response, error) {
-	response, err := c.httpClient.Do(req)
+	return c.sendWith(c.httpClient, req)
+}
+
+func (c *APIClient) sendWith(client *http.Client, req *http.Request) (*http.Response, error) {
+	response, err := client.Do(req)
 	if err != nil {
 		c.logger.Debug().Str("method", req.Method).Str("path", req.URL.String()).Err(err).Msg("kiteworks API call errored")
 		return nil, err