tmp

Makefile

@@ -0,0 +1,8 @@
+MOCK_FILE=mock/mock_gen.go
+
+all: mock
+
+mock: $(MOCK_FILE)
+
+$(MOCK_FILE): interfaces.go
+	go run github.com/matryer/moq@latest -pkg mock -out $(MOCK_FILE) . Source Destination

destination/cloud_storage.go

@@ -3,6 +3,8 @@ package destination
 import (
 	"context"
 	"io"
+
+	"github.com/secmon-as-code/hatchery/pkg/metadata"
 )
 
 // CloudStorage is a destination that writes data to a Google cloud storage bucket.
@@ -19,7 +21,10 @@ func NewCloudStorage(bucket, prefix string) *CloudStorage {
 	}
 }
 
-func (c *CloudStorage) NewWriter(ctx context.Context) (io.WriteCloser, error) {
-	// Write data to Google cloud storage bucket.
+func (c *CloudStorage) NewWriter(ctx context.Context, md metadata.MetaData) (io.WriteCloser, error) {
+	// Open a new file in the cloud storage bucket.
+
+	client, err := storage.NewClient(ctx)
+
 	return nil, nil
 }

go.mod

@@ -4,12 +4,47 @@ go 1.22.0
 
 require (
 	github.com/m-mizutani/goerr v0.1.14
+	github.com/m-mizutani/gt v0.0.10
 	github.com/urfave/cli/v2 v2.27.4
 )
 
 require (
+	cloud.google.com/go v0.115.0 // indirect
+	cloud.google.com/go/auth v0.6.1 // indirect
+	cloud.google.com/go/auth/oauth2adapt v0.2.2 // indirect
+	cloud.google.com/go/compute/metadata v0.3.0 // indirect
+	cloud.google.com/go/iam v1.1.8 // indirect
+	cloud.google.com/go/storage v1.43.0 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.4 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
+	github.com/go-logr/logr v1.4.1 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
+	github.com/golang/protobuf v1.5.4 // indirect
+	github.com/google/go-cmp v0.6.0 // indirect
+	github.com/google/s2a-go v0.1.7 // indirect
 	github.com/google/uuid v1.6.0 // indirect
+	github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
+	github.com/googleapis/gax-go/v2 v2.12.5 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 // indirect
+	go.opencensus.io v0.24.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 // indirect
+	go.opentelemetry.io/otel v1.24.0 // indirect
+	go.opentelemetry.io/otel/metric v1.24.0 // indirect
+	go.opentelemetry.io/otel/trace v1.24.0 // indirect
+	golang.org/x/crypto v0.24.0 // indirect
+	golang.org/x/net v0.26.0 // indirect
+	golang.org/x/oauth2 v0.21.0 // indirect
+	golang.org/x/sync v0.7.0 // indirect
+	golang.org/x/sys v0.21.0 // indirect
+	golang.org/x/text v0.16.0 // indirect
+	golang.org/x/time v0.5.0 // indirect
+	google.golang.org/api v0.187.0 // indirect
+	google.golang.org/genproto v0.0.0-20240624140628-dc46fd24d27d // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20240617180043-68d350f18fd4 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240624140628-dc46fd24d27d // indirect
+	google.golang.org/grpc v1.64.0 // indirect
+	google.golang.org/protobuf v1.34.2 // indirect
 )

hatchery.go

@@ -25,17 +25,17 @@ func New(opts ...Option) (*Hatchery, error) {
 	return h, nil
 }
 
-func (h *Hatchery) Run(ctx context.Context, pipelineIDs []string) error {
-	for _, id := range pipelineIDs {
+func (h *Hatchery) Run(ctx context.Context, streamIDs []string) error {
+	for _, id := range streamIDs {
 		if _, ok := h.pipelines[StreamID(id)]; !ok {
 			return ErrStreamNotFound
 		}
 	}
 
 	var wg sync.WaitGroup
-	var errCh = make(chan error, len(pipelineIDs))
+	var errCh = make(chan error, len(streamIDs))
 
-	for _, pID := range pipelineIDs {
+	for _, pID := range streamIDs {
 		wg.Add(1)
 		go func(id string) {
 			defer wg.Done()

source/context.go

@@ -2,24 +2,9 @@ package source
 
 import (
 	"context"
-	"net/http"
 	"time"
 )
 
-// HTTPClient is an interface for HTTP client. It's used to inject mock client for testing.
-type HTTPClient interface {
-	Do(req *http.Request) (*http.Response, error)
-}
-
-type ctxHTTPClientKey struct{}
-
-func httpClientFromCtx(ctx context.Context) HTTPClient {
-	if c, ok := ctx.Value(ctxHTTPClientKey{}).(HTTPClient); ok {
-		return c
-	}
-	return http.DefaultClient
-}
-
 // InjectHTTPClient injects HTTP client to context. It's used to inject mock client for testing.
 func InjectHTTPClient(ctx context.Context, c HTTPClient) context.Context {
 	return context.WithValue(ctx, ctxHTTPClientKey{}, c)

source/interfaces.go

@@ -0,0 +1,20 @@
+package source
+
+import (
+	"context"
+	"net/http"
+)
+
+// HTTPClient is an interface for HTTP client. It's used to inject mock client for testing.
+type HTTPClient interface {
+	Do(req *http.Request) (*http.Response, error)
+}
+
+type ctxHTTPClientKey struct{}
+
+func httpClientFromCtx(ctx context.Context) HTTPClient {
+	if c, ok := ctx.Value(ctxHTTPClientKey{}).(HTTPClient); ok {
+		return c
+	}
+	return http.DefaultClient
+}

source/slack.go

@@ -108,7 +108,11 @@ func (x *Slack) crawl(ctx context.Context, end time.Time, cursor string, dst hat
 		return nil, goerr.Wrap(err, "failed to read response body")
 	}
 
-	var resp apiResponse
+	var resp struct {
+		ResponseMetadata struct {
+			NextCursor string `json:"next_cursor"`
+		} `json:"response_metadata"`
+	}
 	if err := json.Unmarshal(body, &resp); err != nil {
 		return nil, goerr.Wrap(err, "failed to unmarshal response body")
 	}
@@ -134,9 +138,3 @@ func (x *Slack) crawl(ctx context.Context, end time.Time, cursor string, dst hat
 
 	return nil, nil
 }
-
-type apiResponse struct {
-	ResponseMetadata struct {
-		NextCursor string `json:"next_cursor"`
-	} `json:"response_metadata"`
-}