feat: snowpipe streaming

router/batchrouter/asyncdestinationmanager/common/utils.go

@@ -3,7 +3,7 @@ package common
 import "slices"
 
 var (
-	asyncDestinations = []string{"MARKETO_BULK_UPLOAD", "BINGADS_AUDIENCE", "ELOQUA", "YANDEX_METRICA_OFFLINE_EVENTS", "BINGADS_OFFLINE_CONVERSIONS", "KLAVIYO_BULK_UPLOAD", "LYTICS_BULK_UPLOAD"}
+	asyncDestinations = []string{"MARKETO_BULK_UPLOAD", "BINGADS_AUDIENCE", "ELOQUA", "YANDEX_METRICA_OFFLINE_EVENTS", "BINGADS_OFFLINE_CONVERSIONS", "KLAVIYO_BULK_UPLOAD", "LYTICS_BULK_UPLOAD", "SNOWPIPE_STREAMING"}
 	sftpDestinations  = []string{"SFTP"}
 )
 

router/batchrouter/asyncdestinationmanager/manager.go

@@ -16,6 +16,7 @@ import (
 	lyticsBulkUpload "github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/lytics_bulk_upload"
 	marketobulkupload "github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/marketo-bulk-upload"
 	"github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/sftp"
+	"github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/snowpipestreaming"
 	"github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/yandexmetrica"
 )
 
@@ -41,6 +42,8 @@ func newRegularManager(
 		return klaviyobulkupload.NewManager(logger, statsFactory, destination)
 	case "LYTICS_BULK_UPLOAD":
 		return lyticsBulkUpload.NewManager(logger, statsFactory, destination)
+	case "SNOWPIPE_STREAMING":
+		return snowpipestreaming.New(conf, logger, statsFactory, destination), nil
 	}
 	return nil, errors.New("invalid destination type")
 }

router/batchrouter/asyncdestinationmanager/snowpipestreaming/createchannel.go

@@ -0,0 +1,69 @@
+package snowpipestreaming
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/rudderlabs/rudder-go-kit/httputil"
+)
+
+type accountConfig struct {
+	Account              string `json:"account"`
+	User                 string `json:"user"`
+	Role                 string `json:"role"`
+	PrivateKey           string `json:"privateKey"`
+	PrivateKeyPassphrase string `json:"privateKeyPassphrase"`
+}
+
+type tableConfig struct {
+	Database string `json:"database"`
+	Schema   string `json:"schema"`
+	Table    string `json:"table"`
+}
+
+type createChannelRequest struct {
+	RudderIdentifier string        `json:"rudderIdentifier"`
+	Partition        string        `json:"partition"`
+	AccountConfig    accountConfig `json:"account"`
+	TableConfig      tableConfig   `json:"table"`
+}
+
+type createChannelResponse struct {
+	ChannelId   string `json:"channelId"`
+	ChannelName string `json:"channelName"`
+	ClientName  string `json:"clientName"`
+	Valid       bool   `json:"valid"`
+}
+
+func (m *Manager) createChannel(ctx context.Context, channelReq *createChannelRequest) (*createChannelResponse, error) {
+	reqJSON, err := json.Marshal(channelReq)
+	if err != nil {
+		return nil, fmt.Errorf("marshalling create channel request: %w", err)
+	}
+
+	channelReqURL := m.config.clientURL + "/channels"
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, channelReqURL, bytes.NewBuffer(reqJSON))
+	if err != nil {
+		return nil, fmt.Errorf("creating request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, reqErr := m.requestDoer.Do(req)
+	if reqErr != nil {
+		return nil, fmt.Errorf("sending request: %w", reqErr)
+	}
+	defer func() { httputil.CloseResponse(resp) }()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("invalid status code: %d", resp.StatusCode)
+	}
+
+	var res createChannelResponse
+	if err := json.NewDecoder(resp.Body).Decode(&res); err != nil {
+		return nil, fmt.Errorf("decoding response: %w", err)
+	}
+
+	return &res, nil
+}

router/batchrouter/asyncdestinationmanager/snowpipestreaming/insert.go

@@ -0,0 +1,60 @@
+package snowpipestreaming
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/rudderlabs/rudder-server/utils/httputil"
+)
+
+type Row map[string]any
+
+type insertRequest struct {
+	Rows   []Row  `json:"rows"`
+	Offset string `json:"offset"`
+}
+
+type insertError struct {
+	RowIndex                    int64    `json:"rowIndex"`
+	ExtraColNames               []string `json:"extraColNames"`
+	MissingNotNullColNames      []string `json:"missingNotNullColNames"`
+	NullValueForNotNullColNames []string `json:"nullvalueForNotNullColNames"`
+}
+
+type insertResponse struct {
+	Success bool          `json:"success"`
+	Errors  []insertError `json:"errors"`
+}
+
+func (m *Manager) insert(ctx context.Context, channelId string, insertRequest *insertRequest) (*insertResponse, error) {
+	reqJSON, err := json.Marshal(insertRequest)
+	if err != nil {
+		return nil, fmt.Errorf("marshalling insert request: %w", err)
+	}
+
+	insertReqURL := m.config.clientURL + "/channels/" + channelId + "/insert"
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, insertReqURL, bytes.NewBuffer(reqJSON))
+	if err != nil {
+		return nil, fmt.Errorf("creating request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, reqErr := m.requestDoer.Do(req)
+	if reqErr != nil {
+		return nil, fmt.Errorf("sending request: %w", reqErr)
+	}
+	defer func() { httputil.CloseResponse(resp) }()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("invalid status code: %d", resp.StatusCode)
+	}
+
+	var res insertResponse
+	if err := json.NewDecoder(resp.Body).Decode(&res); err != nil {
+		return nil, fmt.Errorf("decoding response: %w", err)
+	}
+
+	return &res, nil
+}