Add push to remote mirror repository (#15157)

* Added push mirror model.

* Integrated push mirror into queue.

* Moved methods into own file.

* Added basic implementation.

* Mirror wiki too.

* Removed duplicated method.

* Get url for different remotes.

* Added migration.

* Unified remote url access.

* Add/Remove push mirror remotes.

* Prevent hangs with missing credentials.

* Moved code between files.

* Changed sanitizer interface.

* Added push mirror backend methods.

* Only update the mirror remote.

* Limit refs on push.

* Added UI part.

* Added missing table.

* Delete mirror if repository gets removed.

* Changed signature. Handle object errors.

* Added upload method.

* Added "upload" unit tests.

* Added transfer adapter unit tests.

* Send correct headers.

* Added pushing of LFS objects.

* Added more logging.

* Simpler body handling.

* Process files in batches to reduce HTTP calls.

* Added created timestamp.

* Fixed invalid column name.

* Changed name to prevent xorm auto setting.

* Remove table header im empty.

* Strip exit code from error message.

* Added docs page about mirroring.

* Fixed date.

* Fixed merge errors.

* Moved test to integrations.

* Added push mirror test.

* Added test.

modules/lfs/http_client.go

@@ -7,17 +7,19 @@ package lfs
 import (
 	"bytes"
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"net/http"
 	"net/url"
 	"strings"
 
 	"code.gitea.io/gitea/modules/log"
+
+	jsoniter "github.com/json-iterator/go"
 )
 
+const batchSize = 20
+
 // HTTPClient is used to communicate with the LFS server
 // https://github.com/git-lfs/git-lfs/blob/main/docs/api/batch.md
 type HTTPClient struct {
@@ -26,6 +28,11 @@ type HTTPClient struct {
 	transfers map[string]TransferAdapter
 }
 
+// BatchSize returns the preferred size of batchs to process
+func (c *HTTPClient) BatchSize() int {
+	return batchSize
+}
+
 func newHTTPClient(endpoint *url.URL) *HTTPClient {
 	hc := &http.Client{}
 
@@ -55,21 +62,25 @@ func (c *HTTPClient) transferNames() []string {
 }
 
 func (c *HTTPClient) batch(ctx context.Context, operation string, objects []Pointer) (*BatchResponse, error) {
+	log.Trace("BATCH operation with objects: %v", objects)
+
 	url := fmt.Sprintf("%s/objects/batch", c.endpoint)
 
 	request := &BatchRequest{operation, c.transferNames(), nil, objects}
 
 	payload := new(bytes.Buffer)
-	err := json.NewEncoder(payload).Encode(request)
+	err := jsoniter.NewEncoder(payload).Encode(request)
 	if err != nil {
-		return nil, fmt.Errorf("lfs.HTTPClient.batch json.Encode: %w", err)
+		log.Error("Error encoding json: %v", err)
+		return nil, err
 	}
 
-	log.Trace("lfs.HTTPClient.batch NewRequestWithContext: %s", url)
+	log.Trace("Calling: %s", url)
 
 	req, err := http.NewRequestWithContext(ctx, "POST", url, payload)
 	if err != nil {
-		return nil, fmt.Errorf("lfs.HTTPClient.batch http.NewRequestWithContext: %w", err)
+		log.Error("Error creating request: %v", err)
+		return nil, err
 	}
 	req.Header.Set("Content-type", MediaType)
 	req.Header.Set("Accept", MediaType)
@@ -81,18 +92,20 @@ func (c *HTTPClient) batch(ctx context.Context, operation string, objects []Poin
 			return nil, ctx.Err()
 		default:
 		}
-		return nil, fmt.Errorf("lfs.HTTPClient.batch http.Do: %w", err)
+		log.Error("Error while processing request: %v", err)
+		return nil, err
 	}
 	defer res.Body.Close()
 
 	if res.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("lfs.HTTPClient.batch: Unexpected servers response: %s", res.Status)
+		return nil, fmt.Errorf("Unexpected server response: %s", res.Status)
 	}
 
 	var response BatchResponse
-	err = json.NewDecoder(res.Body).Decode(&response)
+	err = jsoniter.NewDecoder(res.Body).Decode(&response)
 	if err != nil {
-		return nil, fmt.Errorf("lfs.HTTPClient.batch json.Decode: %w", err)
+		log.Error("Error decoding json: %v", err)
+		return nil, err
 	}
 
 	if len(response.Transfer) == 0 {
@@ -103,27 +116,99 @@ func (c *HTTPClient) batch(ctx context.Context, operation string, objects []Poin
 }
 
 // Download reads the specific LFS object from the LFS server
-func (c *HTTPClient) Download(ctx context.Context, oid string, size int64) (io.ReadCloser, error) {
-	var objects []Pointer
-	objects = append(objects, Pointer{oid, size})
+func (c *HTTPClient) Download(ctx context.Context, objects []Pointer, callback DownloadCallback) error {
+	return c.performOperation(ctx, objects, callback, nil)
+}
 
-	result, err := c.batch(ctx, "download", objects)
+// Upload sends the specific LFS object to the LFS server
+func (c *HTTPClient) Upload(ctx context.Context, objects []Pointer, callback UploadCallback) error {
+	return c.performOperation(ctx, objects, nil, callback)
+}
+
+func (c *HTTPClient) performOperation(ctx context.Context, objects []Pointer, dc DownloadCallback, uc UploadCallback) error {
+	if len(objects) == 0 {
+		return nil
+	}
+
+	operation := "download"
+	if uc != nil {
+		operation = "upload"
+	}
+
+	result, err := c.batch(ctx, operation, objects)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	transferAdapter, ok := c.transfers[result.Transfer]
 	if !ok {
-		return nil, fmt.Errorf("lfs.HTTPClient.Download Transferadapter not found: %s", result.Transfer)
+		return fmt.Errorf("TransferAdapter not found: %s", result.Transfer)
 	}
 
-	if len(result.Objects) == 0 {
-		return nil, errors.New("lfs.HTTPClient.Download: No objects in result")
+	for _, object := range result.Objects {
+		if object.Error != nil {
+			objectError := errors.New(object.Error.Message)
+			log.Trace("Error on object %v: %v", object.Pointer, objectError)
+			if uc != nil {
+				if _, err := uc(object.Pointer, objectError); err != nil {
+					return err
+				}
+			} else {
+				if err := dc(object.Pointer, nil, objectError); err != nil {
+					return err
+				}
+			}
+			continue
+		}
+
+		if uc != nil {
+			if len(object.Actions) == 0 {
+				log.Trace("%v already present on server", object.Pointer)
+				continue
+			}
+
+			link, ok := object.Actions["upload"]
+			if !ok {
+				log.Debug("%+v", object)
+				return errors.New("Missing action 'upload'")
+			}
+
+			content, err := uc(object.Pointer, nil)
+			if err != nil {
+				return err
+			}
+
+			err = transferAdapter.Upload(ctx, link, object.Pointer, content)
+
+			content.Close()
+
+			if err != nil {
+				return err
+			}
+
+			link, ok = object.Actions["verify"]
+			if ok {
+				if err := transferAdapter.Verify(ctx, link, object.Pointer); err != nil {
+					return err
+				}
+			}
+		} else {
+			link, ok := object.Actions["download"]
+			if !ok {
+				log.Debug("%+v", object)
+				return errors.New("Missing action 'download'")
+			}
+
+			content, err := transferAdapter.Download(ctx, link)
+			if err != nil {
+				return err
+			}
+
+			if err := dc(object.Pointer, content, nil); err != nil {
+				return err
+			}
+		}
 	}
 
-	content, err := transferAdapter.Download(ctx, result.Objects[0])
-	if err != nil {
-		return nil, err
-	}
-	return content, nil
+	return nil
 }