refactor(data): implement BinaryFetcher interface for improved code structure

cmd/main/main.go

@@ -268,6 +268,12 @@ func main() {
 		defer artifactPrivateServiceClientConn.Close()
 	}
 
+	minioxClientWrapper, err := miniox.NewMinioClient(ctx, &config.Config.Minio, logger)
+	if err != nil {
+		logger.Fatal("failed to create miniox client wrapper", zap.Error(err))
+	}
+
+	binaryFetcher := external.NewArtifactBinaryFetcher(artifactPrivateServiceClient, minioxClientWrapper)
 	service := service.NewService(
 		service.ServiceConfig{
 			Repository:               repo,
@@ -281,6 +287,7 @@ func main() {
 			Memory:                   ms,
 			WorkerUID:                workerUID,
 			RetentionHandler:         nil,
+			BinaryFetcher:            binaryFetcher,
 		},
 	)
 
@@ -448,6 +455,7 @@ func main() {
 			WorkerUID:                    workerUID,
 			ArtifactPublicServiceClient:  artifactPublicServiceClient,
 			ArtifactPrivateServiceClient: artifactPrivateServiceClient,
+			BinaryFetcher:                binaryFetcher,
 		},
 	)
 

pkg/component/ai/openai/v0/main.go

@@ -4,6 +4,7 @@ package openai
 import (
 	"bufio"
 	"context"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -462,7 +463,12 @@ func (e *execution) worker(ctx context.Context, client *httpclient.Client, job *
 
 		results := []imageGenerationsOutputResult{}
 		for _, d := range resp.Data {
-			img, err := data.NewImageFromURL(fmt.Sprintf("data:image/webp;base64,%s", d.Image))
+			b, err := base64.StdEncoding.DecodeString(d.Image)
+			if err != nil {
+				job.Error.Error(ctx, err)
+				return
+			}
+			img, err := data.NewImageFromBytes(b, data.PNG, "")
 			if err != nil {
 				job.Error.Error(ctx, err)
 				return

pkg/component/base/executionwrapper.go

@@ -62,7 +62,8 @@ func NewOutputWriter(ow OutputWriter, schema string) *outputWriter {
 	}
 }
 func (ow *outputWriter) WriteData(ctx context.Context, output any) (err error) {
-	outputMap, err := data.Marshal(output)
+	marshaler := data.NewMarshaler()
+	outputMap, err := marshaler.Marshal(output)
 	if err != nil {
 		return err
 	}

pkg/component/operator/audio/v0/task_detect_activity_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/instill-ai/pipeline-backend/pkg/component/base"
 	"github.com/instill-ai/pipeline-backend/pkg/component/internal/mock"
 	"github.com/instill-ai/pipeline-backend/pkg/data"
+	"github.com/instill-ai/pipeline-backend/pkg/external"
 )
 
 func TestDetectActivity(t *testing.T) {
@@ -124,7 +125,9 @@ func TestDetectActivity(t *testing.T) {
 				jsonValue, err := data.NewJSONValue(segmentsMap)
 				c.Assert(err, qt.IsNil)
 
-				c.Assert(data.Unmarshal(jsonValue, &expectedSegmentsStruct), qt.IsNil)
+				binaryFetcher := external.NewBinaryFetcher()
+				unmarshaler := data.NewUnmarshaler(binaryFetcher)
+				c.Assert(unmarshaler.Unmarshal(context.Background(), jsonValue, &expectedSegmentsStruct), qt.IsNil)
 				expectedSegments := expectedSegmentsStruct.Segments
 
 				c.Assert(capturedOutput.Segments, qt.HasLen, len(expectedSegments))

pkg/component/operator/audio/v0/task_segment_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/instill-ai/pipeline-backend/pkg/component/base"
 	"github.com/instill-ai/pipeline-backend/pkg/component/internal/mock"
 	"github.com/instill-ai/pipeline-backend/pkg/data"
+	"github.com/instill-ai/pipeline-backend/pkg/external"
 )
 
 func TestSegment(t *testing.T) {
@@ -67,7 +68,9 @@ func TestSegment(t *testing.T) {
 			jsonValue, err := data.NewJSONValue(segmentsMap)
 			c.Assert(err, qt.IsNil)
 
-			c.Assert(data.Unmarshal(jsonValue, &segmentsStruct), qt.IsNil)
+			binaryFetcher := external.NewBinaryFetcher()
+			unmarshaler := data.NewUnmarshaler(binaryFetcher)
+			c.Assert(unmarshaler.Unmarshal(context.Background(), jsonValue, &segmentsStruct), qt.IsNil)
 			segments := segmentsStruct.Segments
 
 			ir.ReadDataMock.Set(func(ctx context.Context, input any) error {

pkg/component/operator/document/v0/convert_document_to_markdown.go

@@ -2,6 +2,7 @@ package document
 
 import (
 	"context"
+	"encoding/base64"
 
 	"github.com/instill-ai/pipeline-backend/pkg/component/base"
 	"github.com/instill-ai/pipeline-backend/pkg/component/operator/document/v0/transformer"
@@ -40,7 +41,8 @@ func (e *execution) convertDocumentToMarkdown(ctx context.Context, job *base.Job
 		Images: func() []format.Image {
 			images := make([]format.Image, len(transformerOutputStruct.Images))
 			for i, image := range transformerOutputStruct.Images {
-				images[i], _ = data.NewImageFromURL(image)
+				b, _ := base64.StdEncoding.DecodeString(image)
+				images[i], _ = data.NewImageFromBytes(b, data.PNG, "")
 				// TODO: handle error
 			}
 			return images
@@ -49,7 +51,8 @@ func (e *execution) convertDocumentToMarkdown(ctx context.Context, job *base.Job
 		AllPageImages: func() []format.Image {
 			images := make([]format.Image, len(transformerOutputStruct.AllPageImages))
 			for i, image := range transformerOutputStruct.AllPageImages {
-				images[i], _ = data.NewImageFromURL(image)
+				b, _ := base64.StdEncoding.DecodeString(image)
+				images[i], _ = data.NewImageFromBytes(b, data.PNG, "")
 				// TODO: handle error
 			}
 			return images

pkg/component/operator/document/v0/convert_document_to_markdown_test.go

@@ -2,8 +2,6 @@ package document
 
 import (
 	"context"
-	"encoding/base64"
-	"fmt"
 	"os"
 	"testing"
 
@@ -109,15 +107,13 @@ func TestConvertDocumentToMarkdown(t *testing.T) {
 			fileContent, err := os.ReadFile(test.filepath)
 			c.Assert(err, qt.IsNil)
 
-			base64DataURI := fmt.Sprintf("data:%s;base64,%s", mimeTypeByExtension(test.filepath), base64.StdEncoding.EncodeToString(fileContent))
-
 			ir, ow, eh, job := mock.GenerateMockJob(c)
 			ir.ReadDataMock.Set(func(ctx context.Context, input any) error {
 				switch input := input.(type) {
 				case *ConvertDocumentToMarkdownInput:
 					*input = ConvertDocumentToMarkdownInput{
 						Document: func() format.Document {
-							doc, err := data.NewDocumentFromURL(base64DataURI)
+							doc, err := data.NewDocumentFromBytes(fileContent, mimeTypeByExtension(test.filepath), "")
 							if err != nil {
 								return nil
 							}

pkg/component/operator/document/v0/convert_test.go

@@ -2,8 +2,6 @@ package document
 
 import (
 	"context"
-	"encoding/base64"
-	"fmt"
 	"os"
 	"testing"
 
@@ -65,8 +63,6 @@ func TestConvertToText(t *testing.T) {
 			fileContent, err := os.ReadFile(test.filepath)
 			c.Assert(err, qt.IsNil)
 
-			base64DataURI := fmt.Sprintf("data:%s;base64,%s", mimeTypeByExtension(test.filepath), base64.StdEncoding.EncodeToString(fileContent))
-
 			execution, err := component.CreateExecution(base.ComponentExecution{
 				Component: component,
 				Task:      "TASK_CONVERT_TO_TEXT",
@@ -79,7 +75,7 @@ func TestConvertToText(t *testing.T) {
 				case *ConvertToTextInput:
 					*input = ConvertToTextInput{
 						Document: func() format.Document {
-							doc, err := data.NewDocumentFromURL(base64DataURI)
+							doc, err := data.NewDocumentFromBytes(fileContent, mimeTypeByExtension(test.filepath), "")
 							if err != nil {
 								return nil
 							}

pkg/component/operator/document/v0/convert_to_images.go

@@ -2,6 +2,7 @@ package document
 
 import (
 	"context"
+	"encoding/base64"
 
 	"github.com/instill-ai/pipeline-backend/pkg/component/base"
 	"github.com/instill-ai/pipeline-backend/pkg/component/operator/document/v0/transformer"
@@ -22,8 +23,8 @@ func (e *execution) convertDocumentToImages(ctx context.Context, job *base.Job)
 	}
 
 	transformerInputStruct := transformer.ConvertDocumentToImagesTransformerInput{
-		Document: dataURI.String(),
-		Filename: inputStruct.Filename,
+		Document:   dataURI.String(),
+		Filename:   inputStruct.Filename,
 		Resolution: inputStruct.Resolution,
 	}
 
@@ -35,7 +36,9 @@ func (e *execution) convertDocumentToImages(ctx context.Context, job *base.Job)
 		Images: func() []format.Image {
 			images := make([]format.Image, len(transformerOutputStruct.Images))
 			for i, image := range transformerOutputStruct.Images {
-				images[i], _ = data.NewImageFromURL(image)
+				b, _ := base64.StdEncoding.DecodeString(image)
+				images[i], _ = data.NewImageFromBytes(b, data.PNG, "")
+				// TODO: handle error
 			}
 			return images
 		}(),

pkg/component/operator/document/v0/convert_to_images_test.go

@@ -2,8 +2,6 @@ package document
 
 import (
 	"context"
-	"encoding/base64"
-	"fmt"
 	"os"
 	"testing"
 
@@ -44,16 +42,14 @@ func Test_ConvertDocumentToImages(t *testing.T) {
 	fileContent, err := os.ReadFile(test.filepath)
 	c.Assert(err, qt.IsNil)
 
-	base64DataURI := fmt.Sprintf("data:%s;base64,%s", mimeTypeByExtension(test.filepath), base64.StdEncoding.EncodeToString(fileContent))
-
 	ir, ow, eh, job := mock.GenerateMockJob(c)
 
 	ir.ReadDataMock.Times(1).Set(func(ctx context.Context, input any) error {
 		switch input := input.(type) {
 		case *ConvertDocumentToImagesInput:
 			*input = ConvertDocumentToImagesInput{
 				Document: func() format.Document {
-					doc, err := data.NewDocumentFromURL(base64DataURI)
+					doc, err := data.NewDocumentFromBytes(fileContent, mimeTypeByExtension(test.filepath), "")
 					if err != nil {
 						return nil
 					}

pkg/data/audio.go

@@ -1,6 +1,7 @@
 package data
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"os"
@@ -11,6 +12,7 @@ import (
 
 	"github.com/instill-ai/pipeline-backend/pkg/data/format"
 	"github.com/instill-ai/pipeline-backend/pkg/data/path"
+	"github.com/instill-ai/pipeline-backend/pkg/external"
 )
 
 type audioData struct {
@@ -50,8 +52,8 @@ func NewAudioFromBytes(b []byte, contentType, filename string) (*audioData, erro
 	return createAudioData(b, contentType, filename)
 }
 
-func NewAudioFromURL(url string) (*audioData, error) {
-	b, contentType, filename, err := convertURLToBytes(url)
+func NewAudioFromURL(ctx context.Context, binaryFetcher external.BinaryFetcher, url string) (video *audioData, err error) {
+	b, contentType, filename, err := binaryFetcher.FetchFromURL(ctx, url)
 	if err != nil {
 		return nil, err
 	}

pkg/data/audio_test.go

@@ -1,12 +1,15 @@
 package data
 
 import (
+	"context"
 	"os"
 	"testing"
 
 	"github.com/google/go-cmp/cmp/cmpopts"
 
 	qt "github.com/frankban/quicktest"
+
+	"github.com/instill-ai/pipeline-backend/pkg/external"
 )
 
 func TestNewAudioFromBytes(t *testing.T) {
@@ -54,7 +57,9 @@ func TestNewAudioFromBytes(t *testing.T) {
 func TestNewAudioFromURL(t *testing.T) {
 	t.Parallel()
 	c := qt.New(t)
+	ctx := context.Background()
 
+	binaryFetcher := external.NewBinaryFetcher()
 	testCases := []struct {
 		name string
 		url  string
@@ -66,7 +71,7 @@ func TestNewAudioFromURL(t *testing.T) {
 
 	for _, tc := range testCases {
 		c.Run(tc.name, func(c *qt.C) {
-			audio, err := NewAudioFromURL(tc.url)
+			audio, err := NewAudioFromURL(ctx, binaryFetcher, tc.url)
 
 			if tc.name == "Valid audio URL" {
 				c.Assert(err, qt.IsNil)

pkg/data/document.go

@@ -1,12 +1,15 @@
 package data
 
 import (
+	"context"
+	"encoding/base64"
 	"fmt"
 	"strings"
 
 	"github.com/instill-ai/pipeline-backend/pkg/component/operator/document/v0/transformer"
 	"github.com/instill-ai/pipeline-backend/pkg/data/format"
 	"github.com/instill-ai/pipeline-backend/pkg/data/path"
+	"github.com/instill-ai/pipeline-backend/pkg/external"
 )
 
 type documentData struct {
@@ -38,12 +41,11 @@ func NewDocumentFromBytes(b []byte, contentType, filename string) (*documentData
 	return createDocumentData(b, contentType, filename)
 }
 
-func NewDocumentFromURL(url string) (*documentData, error) {
-	b, contentType, filename, err := convertURLToBytes(url)
+func NewDocumentFromURL(ctx context.Context, binaryFetcher external.BinaryFetcher, url string) (*documentData, error) {
+	b, contentType, filename, err := binaryFetcher.FetchFromURL(ctx, url)
 	if err != nil {
 		return nil, err
 	}
-
 	return createDocumentData(b, contentType, filename)
 }
 
@@ -143,7 +145,12 @@ func (d *documentData) PDF() (val format.Document, err error) {
 		return nil, err
 	}
 
-	return NewDocumentFromURL(fmt.Sprintf("data:application/pdf;base64,%s", s))
+	b, err := base64.StdEncoding.DecodeString(s)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewDocumentFromBytes(b, PDF, d.filename)
 }
 
 func (d *documentData) Images() (mp Array, err error) {
@@ -168,8 +175,11 @@ func (d *documentData) Images() (mp Array, err error) {
 	images := make([]format.Value, len(res.Images))
 
 	for idx := range res.Images {
-		// img := strings.Split(res.Images[idx], ",")[1]
-		images[idx], err = NewImageFromURL(res.Images[idx])
+		b, err := base64.StdEncoding.DecodeString(res.Images[idx])
+		if err != nil {
+			return nil, err
+		}
+		images[idx], err = NewImageFromBytes(b, PNG, d.filename)
 		if err != nil {
 			return nil, fmt.Errorf("NewImageFromBytes: %w", err)
 		}

pkg/data/document_test.go

@@ -1,10 +1,13 @@
 package data
 
 import (
+	"context"
 	"os"
 	"testing"
 
 	qt "github.com/frankban/quicktest"
+
+	"github.com/instill-ai/pipeline-backend/pkg/external"
 )
 
 func TestNewDocumentFromBytes(t *testing.T) {
@@ -48,7 +51,8 @@ func TestNewDocumentFromBytes(t *testing.T) {
 func TestNewDocumentFromURL(t *testing.T) {
 	t.Parallel()
 	c := qt.New(t)
-
+	ctx := context.Background()
+	binaryFetcher := external.NewBinaryFetcher()
 	testCases := []struct {
 		name string
 		url  string
@@ -61,7 +65,7 @@ func TestNewDocumentFromURL(t *testing.T) {
 
 	for _, tc := range testCases {
 		c.Run(tc.name, func(c *qt.C) {
-			document, err := NewDocumentFromURL(tc.url)
+			document, err := NewDocumentFromURL(ctx, binaryFetcher, tc.url)
 
 			if tc.name == "Valid PDF URL" || tc.name == "Valid TXT URL" || tc.name == "Valid DOCX URL" {
 				c.Assert(err, qt.IsNil)