Refactor/no diags

Makefile

@@ -4,4 +4,8 @@ test:
 
 .PHONY: lint
 lint:
-	golangci-lint run
+	golangci-lint run
+
+.PHONY: generate-protobuf
+generate-protobuf:
+	protoc --proto_path=. --go_out . --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative internal/pb/base.proto internal/pb/source.proto internal/pb/destination.proto

clients/destination.go

@@ -0,0 +1,97 @@
+package clients
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cloudquery/cq-provider-sdk/internal/pb"
+	"github.com/cloudquery/cq-provider-sdk/plugins"
+	"github.com/cloudquery/cq-provider-sdk/schema"
+	"github.com/cloudquery/cq-provider-sdk/spec"
+	"github.com/vmihailenco/msgpack/v5"
+	"google.golang.org/grpc"
+	"gopkg.in/yaml.v3"
+)
+
+type DestinationClient struct {
+	pbClient pb.DestinationClient
+	// this can be used if we have a plugin which is compiled in so we dont need to do any grpc requests
+	localClient plugins.DestinationPlugin
+}
+
+func NewDestinationClient(cc grpc.ClientConnInterface) *DestinationClient {
+	return &DestinationClient{
+		pbClient: pb.NewDestinationClient(cc),
+	}
+}
+
+func NewLocalDestinationClient(p plugins.DestinationPlugin) *DestinationClient {
+	return &DestinationClient{
+		localClient: p,
+	}
+}
+
+func (c *DestinationClient) Configure(ctx context.Context, spec spec.DestinationSpec) error {
+	if c.localClient != nil {
+		return c.localClient.Configure(ctx, spec)
+	}
+	b, err := yaml.Marshal(spec)
+	if err != nil {
+		return fmt.Errorf("failed to marshal spec: %w", err)
+	}
+	if _, err := c.pbClient.Configure(ctx, &pb.Configure_Request{Config: b}); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (c *DestinationClient) GetExampleConfig(ctx context.Context) (string, error) {
+	if c.localClient != nil {
+		return c.localClient.GetExampleConfig(ctx), nil
+	}
+	res, err := c.pbClient.GetExampleConfig(ctx, &pb.GetExampleConfig_Request{})
+	if err != nil {
+		return "", err
+	}
+	return string(res.Config), nil
+}
+
+func (c *DestinationClient) Save(ctx context.Context, msg *FetchResultMessage) error {
+	var saveClient pb.Destination_SaveClient
+	var err error
+	if c.pbClient != nil {
+		saveClient, err = c.pbClient.Save(ctx)
+		if err != nil {
+			return fmt.Errorf("failed to create save client: %w", err)
+		}
+	}
+	if c.localClient != nil {
+		var resource schema.Resource
+		if err := msgpack.Unmarshal(msg.Resource, &resource); err != nil {
+			return fmt.Errorf("failed to unmarshal resources: %w", err)
+		}
+		if err := c.localClient.Save(ctx, []*schema.Resource{&resource}); err != nil {
+			return fmt.Errorf("failed to save resources: %w", err)
+		}
+	} else {
+		if err := saveClient.Send(&pb.Save_Request{Resources: msg.Resource}); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *DestinationClient) CreateTables(ctx context.Context, tables []*schema.Table) error {
+	if c.localClient != nil {
+		return c.localClient.CreateTables(ctx, tables)
+	}
+	b, err := yaml.Marshal(tables)
+	if err != nil {
+		return fmt.Errorf("failed to marshal tables: %w", err)
+	}
+	if _, err := c.pbClient.CreateTables(ctx, &pb.CreateTables_Request{Tables: b}); err != nil {
+		return err
+	}
+	return nil
+}

clients/source.go

@@ -0,0 +1,109 @@
+// package clients is a wrapper around grpc clients so clients can work
+// with non protobuf structs and handle unmarshaling
+package clients
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"text/template"
+
+	"github.com/cloudquery/cq-provider-sdk/internal/pb"
+	"github.com/cloudquery/cq-provider-sdk/schema"
+	"github.com/cloudquery/cq-provider-sdk/spec"
+	"github.com/pkg/errors"
+	"github.com/vmihailenco/msgpack/v5"
+	"github.com/xeipuuv/gojsonschema"
+	"google.golang.org/grpc"
+	"gopkg.in/yaml.v3"
+)
+
+type SourceClient struct {
+	pbClient pb.SourceClient
+}
+
+type FetchResultMessage struct {
+	Resource []byte
+}
+
+const sourcePluginExampleConfigTemplate = `kind: source
+spec:
+  name: {{.Name}}
+  version: {{.Version}}
+  configuration:
+  {{.PluginExampleConfig | indent 4}}
+`
+
+func NewSourceClient(cc grpc.ClientConnInterface) *SourceClient {
+	return &SourceClient{
+		pbClient: pb.NewSourceClient(cc),
+	}
+}
+
+func (c *SourceClient) GetTables(ctx context.Context) ([]*schema.Table, error) {
+	res, err := c.pbClient.GetTables(ctx, &pb.GetTables_Request{})
+	if err != nil {
+		return nil, err
+	}
+	var tables []*schema.Table
+	if err := msgpack.Unmarshal(res.Tables, &tables); err != nil {
+		return nil, err
+	}
+	return tables, nil
+}
+
+func (c *SourceClient) Configure(ctx context.Context, s spec.SourceSpec) (*gojsonschema.Result, error) {
+	b, err := yaml.Marshal(s)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to marshal source spec")
+	}
+	res, err := c.pbClient.Configure(ctx, &pb.Configure_Request{Config: b})
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to configure source")
+	}
+	var validationResult gojsonschema.Result
+	if err := msgpack.Unmarshal(res.JsonschemaResult, &validationResult); err != nil {
+		return nil, errors.Wrap(err, "failed to unmarshal validation result")
+	}
+	return &validationResult, nil
+}
+
+func (c *SourceClient) GetExampleConfig(ctx context.Context) (string, error) {
+	res, err := c.pbClient.GetExampleConfig(ctx, &pb.GetExampleConfig_Request{})
+	if err != nil {
+		return "", fmt.Errorf("failed to get example config: %w", err)
+	}
+	t, err := template.New("source_plugin").Funcs(templateFuncMap()).Parse(sourcePluginExampleConfigTemplate)
+	if err != nil {
+		return "", fmt.Errorf("failed to parse template: %w", err)
+	}
+	var tpl bytes.Buffer
+	if err := t.Execute(&tpl, map[string]interface{}{
+		"Name":                res.Name,
+		"Version":             res.Version,
+		"PluginExampleConfig": string(res.Config),
+	}); err != nil {
+		return "", fmt.Errorf("failed to generate example config: %w", err)
+	}
+	return tpl.String(), nil
+}
+
+func (c *SourceClient) Fetch(ctx context.Context, spec spec.SourceSpec, res chan<- *FetchResultMessage) error {
+	stream, err := c.pbClient.Fetch(ctx, &pb.Fetch_Request{})
+	if err != nil {
+		return fmt.Errorf("failed to fetch resources: %w", err)
+	}
+	for {
+		r, err := stream.Recv()
+		if err != nil {
+			if err == io.EOF {
+				return nil
+			}
+			return fmt.Errorf("failed to fetch resources from stream: %w", err)
+		}
+		res <- &FetchResultMessage{
+			Resource: r.Resource,
+		}
+	}
+}

clients/template.go

@@ -0,0 +1,17 @@
+package clients
+
+import (
+	"strings"
+	"text/template"
+)
+
+func templateFuncMap() template.FuncMap {
+	return template.FuncMap{
+		"indent": indent,
+	}
+}
+
+func indent(spaces int, v string) string {
+	pad := strings.Repeat(" ", spaces)
+	return pad + strings.Replace(v, "\n", "\n"+pad, -1)
+}