From c2a59b60661187e887f6ac4fa36e30afc757b9ec Mon Sep 17 00:00:00 2001 From: Alexander Komyagin Date: Thu, 2 Apr 2026 10:06:31 -0700 Subject: [PATCH 1/3] Firestore sink --- connectors/firestore/connector.go | 356 +++++++++++++++++ .../firestore/connector_integration_test.go | 365 ++++++++++++++++++ connectors/firestore/connector_test.go | 125 ++++++ go.mod | 33 +- go.sum | 53 +++ internal/app/options/connectorflags.go | 37 ++ 6 files changed, 963 insertions(+), 6 deletions(-) create mode 100644 connectors/firestore/connector.go create mode 100644 connectors/firestore/connector_integration_test.go create mode 100644 connectors/firestore/connector_test.go diff --git a/connectors/firestore/connector.go b/connectors/firestore/connector.go new file mode 100644 index 00000000..2e097a80 --- /dev/null +++ b/connectors/firestore/connector.go @@ -0,0 +1,356 @@ +/* + * Copyright (C) 2024 Adiom, Inc. + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + +package firestore + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "strings" + + "cloud.google.com/go/firestore" + "connectrpc.com/connect" + adiomv1 "github.com/adiom-data/dsync/gen/adiom/v1" + "github.com/adiom-data/dsync/gen/adiom/v1/adiomv1connect" + "go.mongodb.org/mongo-driver/v2/bson" + "google.golang.org/api/option" +) + +const ( + DefaultBatchSize = 500 // Firestore max batch size + connectorDBType = "firestore" +) + +var ( + ErrProjectIDRequired = errors.New("project ID is required in connection string") + ErrInvalidURI = errors.New("invalid firestore connection string format") + ErrUnsupportedType = errors.New("unsupported data type for Firestore connector") + ErrMissingDocumentID = errors.New("document missing _id field") +) + +type ConnectorSettings struct { + Uri string + ProjectID string + DatabaseID string + CredentialsFile string + BatchSize int + ID string +} + +type conn struct { + adiomv1connect.UnimplementedConnectorServiceHandler + + client *firestore.Client + settings ConnectorSettings +} + +func parseFirestoreURI(uri string) (projectID, databaseID string, err error) { + const prefix = "firestore://" + if !strings.HasPrefix(strings.ToLower(uri), prefix) { + return "", "", ErrInvalidURI + } + + path := uri[len(prefix):] + if path == "" { + return "", "", ErrProjectIDRequired + } + + parts := strings.SplitN(path, "/", 2) + projectID = parts[0] + if projectID == "" { + return "", "", ErrProjectIDRequired + } + + databaseID = "(default)" + if len(parts) == 2 && parts[1] != "" { + databaseID = parts[1] + } + + return projectID, databaseID, nil +} + +func namespaceToCollection(namespace string) string { + return strings.ReplaceAll(namespace, ".", "_") +} + +func NewConn(ctx context.Context, settings ConnectorSettings) (adiomv1connect.ConnectorServiceHandler, error) { + projectID, databaseID, err := parseFirestoreURI(settings.Uri) + if err != nil { + return nil, fmt.Errorf("failed to parse firestore URI: %w", err) + } + + settings.ProjectID = projectID + settings.DatabaseID = databaseID + + if settings.BatchSize <= 0 || settings.BatchSize > DefaultBatchSize { + settings.BatchSize = DefaultBatchSize + } + + var opts []option.ClientOption + if settings.CredentialsFile != "" { + opts = append(opts, option.WithCredentialsFile(settings.CredentialsFile)) + } + + client, err := firestore.NewClientWithDatabase(ctx, projectID, databaseID, opts...) + if err != nil { + return nil, fmt.Errorf("failed to create firestore client: %w", err) + } + + return &conn{ + client: client, + settings: settings, + }, nil +} + +func (c *conn) GetInfo(context.Context, *connect.Request[adiomv1.GetInfoRequest]) (*connect.Response[adiomv1.GetInfoResponse], error) { + return connect.NewResponse(&adiomv1.GetInfoResponse{ + Id: c.settings.ID, + DbType: connectorDBType, + Spec: fmt.Sprintf("project=%s,database=%s", c.settings.ProjectID, c.settings.DatabaseID), + Capabilities: &adiomv1.Capabilities{ + Sink: &adiomv1.Capabilities_Sink{ + SupportedDataTypes: []adiomv1.DataType{ + adiomv1.DataType_DATA_TYPE_MONGO_BSON, + adiomv1.DataType_DATA_TYPE_JSON_ID, + }, + }, + }, + }), nil +} + +func (c *conn) GetNamespaceMetadata(context.Context, *connect.Request[adiomv1.GetNamespaceMetadataRequest]) (*connect.Response[adiomv1.GetNamespaceMetadataResponse], error) { + return connect.NewResponse(&adiomv1.GetNamespaceMetadataResponse{ + Count: 0, + }), nil +} + +func (c *conn) GeneratePlan(context.Context, *connect.Request[adiomv1.GeneratePlanRequest]) (*connect.Response[adiomv1.GeneratePlanResponse], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.ErrUnsupported) +} + +func (c *conn) ListData(context.Context, *connect.Request[adiomv1.ListDataRequest]) (*connect.Response[adiomv1.ListDataResponse], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.ErrUnsupported) +} + +func (c *conn) StreamLSN(context.Context, *connect.Request[adiomv1.StreamLSNRequest], *connect.ServerStream[adiomv1.StreamLSNResponse]) error { + return connect.NewError(connect.CodeUnimplemented, errors.ErrUnsupported) +} + +func (c *conn) StreamUpdates(context.Context, *connect.Request[adiomv1.StreamUpdatesRequest], *connect.ServerStream[adiomv1.StreamUpdatesResponse]) error { + return connect.NewError(connect.CodeUnimplemented, errors.ErrUnsupported) +} + +func (c *conn) WriteData(ctx context.Context, r *connect.Request[adiomv1.WriteDataRequest]) (*connect.Response[adiomv1.WriteDataResponse], error) { + data := r.Msg.GetData() + if len(data) == 0 { + return connect.NewResponse(&adiomv1.WriteDataResponse{}), nil + } + + namespace := r.Msg.GetNamespace() + collectionName := namespaceToCollection(namespace) + dataType := r.Msg.GetType() + + for i := 0; i < len(data); i += c.settings.BatchSize { + end := i + c.settings.BatchSize + if end > len(data) { + end = len(data) + } + batch := data[i:end] + + if err := c.writeBatch(ctx, collectionName, batch, dataType); err != nil { + return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("failed to write batch to %s: %w", collectionName, err)) + } + } + + return connect.NewResponse(&adiomv1.WriteDataResponse{}), nil +} + +func (c *conn) WriteUpdates(ctx context.Context, r *connect.Request[adiomv1.WriteUpdatesRequest]) (*connect.Response[adiomv1.WriteUpdatesResponse], error) { + updates := r.Msg.GetUpdates() + if len(updates) == 0 { + return connect.NewResponse(&adiomv1.WriteUpdatesResponse{}), nil + } + + namespace := r.Msg.GetNamespace() + collectionName := namespaceToCollection(namespace) + dataType := r.Msg.GetType() + + for i := 0; i < len(updates); i += c.settings.BatchSize { + end := i + c.settings.BatchSize + if end > len(updates) { + end = len(updates) + } + batch := updates[i:end] + + if err := c.writeUpdatesBatch(ctx, collectionName, batch, dataType); err != nil { + return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("failed to write updates batch to %s: %w", collectionName, err)) + } + } + + return connect.NewResponse(&adiomv1.WriteUpdatesResponse{}), nil +} + +func (c *conn) writeBatch(ctx context.Context, collectionName string, data [][]byte, dataType adiomv1.DataType) error { + batch := c.client.Batch() + collection := c.client.Collection(collectionName) + + for _, raw := range data { + docID, docData, err := extractDocumentIDAndData(raw, dataType) + if err != nil { + return fmt.Errorf("failed to extract document ID: %w", err) + } + + docRef := collection.Doc(docID) + batch.Set(docRef, docData) + } + + _, err := batch.Commit(ctx) + if err != nil { + return fmt.Errorf("failed to commit batch: %w", err) + } + + slog.Debug("wrote batch to firestore", "collection", collectionName, "count", len(data)) + return nil +} + +func (c *conn) writeUpdatesBatch(ctx context.Context, collectionName string, updates []*adiomv1.Update, dataType adiomv1.DataType) error { + batch := c.client.Batch() + collection := c.client.Collection(collectionName) + idKey := getIDFieldName(dataType) + + for _, update := range updates { + docID, err := extractIDFromBsonValues(update.GetId()) + if err != nil { + return fmt.Errorf("failed to extract document ID from update: %w", err) + } + + docRef := collection.Doc(docID) + + switch update.GetType() { + case adiomv1.UpdateType_UPDATE_TYPE_DELETE: + batch.Delete(docRef) + case adiomv1.UpdateType_UPDATE_TYPE_UPDATE, adiomv1.UpdateType_UPDATE_TYPE_INSERT: + docData, err := rawToMap(update.GetData(), dataType) + if err != nil { + return fmt.Errorf("failed to convert update data: %w", err) + } + delete(docData, idKey) + batch.Set(docRef, docData) + default: + slog.Warn("unknown update type, treating as upsert", "type", update.GetType()) + docData, err := rawToMap(update.GetData(), dataType) + if err != nil { + return fmt.Errorf("failed to convert update data: %w", err) + } + delete(docData, idKey) + batch.Set(docRef, docData) + } + } + + _, err := batch.Commit(ctx) + if err != nil { + return fmt.Errorf("failed to commit updates batch: %w", err) + } + + slog.Debug("wrote updates batch to firestore", "collection", collectionName, "count", len(updates)) + return nil +} + +func getIDFieldName(dataType adiomv1.DataType) string { + switch dataType { + case adiomv1.DataType_DATA_TYPE_MONGO_BSON: + return "_id" + case adiomv1.DataType_DATA_TYPE_JSON_ID: + return "id" + default: + return "_id" + } +} + +func extractDocumentIDAndData(raw []byte, dataType adiomv1.DataType) (string, map[string]any, error) { + docData, err := rawToMap(raw, dataType) + if err != nil { + return "", nil, err + } + + idKey := getIDFieldName(dataType) + idVal, ok := docData[idKey] + if !ok { + return "", nil, fmt.Errorf("%w: expected field '%s'", ErrMissingDocumentID, idKey) + } + + docID, err := valueToString(idVal) + if err != nil { + return "", nil, fmt.Errorf("failed to convert %s to string: %w", idKey, err) + } + + delete(docData, idKey) + return docID, docData, nil +} + +func rawToMap(raw []byte, dataType adiomv1.DataType) (map[string]any, error) { + var docData map[string]any + + switch dataType { + case adiomv1.DataType_DATA_TYPE_MONGO_BSON: + if err := bson.Unmarshal(raw, &docData); err != nil { + return nil, fmt.Errorf("failed to unmarshal BSON: %w", err) + } + case adiomv1.DataType_DATA_TYPE_JSON_ID: + if err := json.Unmarshal(raw, &docData); err != nil { + return nil, fmt.Errorf("failed to unmarshal JSON: %w", err) + } + default: + return nil, ErrUnsupportedType + } + + return docData, nil +} + +func valueToString(v any) (string, error) { + switch val := v.(type) { + case string: + return val, nil + case bson.ObjectID: + return val.Hex(), nil + case int, int32, int64: + return fmt.Sprintf("%d", val), nil + case uint, uint32, uint64: + return fmt.Sprintf("%d", val), nil + case float32, float64: + return fmt.Sprintf("%v", val), nil + case bson.Binary: + return fmt.Sprintf("%x", val.Data), nil + default: + return fmt.Sprintf("%v", val), nil + } +} + +func extractIDFromBsonValues(ids []*adiomv1.BsonValue) (string, error) { + if len(ids) == 0 { + return "", ErrMissingDocumentID + } + + // Use the first ID field (typically _id) + id := ids[0] + var v any + if err := bson.UnmarshalValue(bson.Type(id.GetType()), id.GetData(), &v); err != nil { + return "", fmt.Errorf("failed to unmarshal BSON value: %w", err) + } + + return valueToString(v) +} + +func (c *conn) Teardown() { + if c.client != nil { + if err := c.client.Close(); err != nil { + slog.Error("failed to close firestore client", "err", err) + } + } +} diff --git a/connectors/firestore/connector_integration_test.go b/connectors/firestore/connector_integration_test.go new file mode 100644 index 00000000..d89e6e6d --- /dev/null +++ b/connectors/firestore/connector_integration_test.go @@ -0,0 +1,365 @@ +//go:build external +// +build external + +/* + * Copyright (C) 2024 Adiom, Inc. + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + +package firestore + +import ( + "context" + "encoding/json" + "os" + "testing" + + "cloud.google.com/go/firestore" + "connectrpc.com/connect" + adiomv1 "github.com/adiom-data/dsync/gen/adiom/v1" + "github.com/adiom-data/dsync/gen/adiom/v1/adiomv1connect" + pkgtest "github.com/adiom-data/dsync/pkg/test" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/suite" + "go.mongodb.org/mongo-driver/v2/bson" + "google.golang.org/api/iterator" + "google.golang.org/api/option" +) + +const ( + // Set FIRESTORE_EMULATOR_HOST=localhost:8080 to use emulator + // Set FIRESTORE_PROJECT_ID=test-project for the project ID + testProjectID = "test-project" + testDatabase = "(default)" + testCollection = "test_collection" + testNamespace = "test.collection" +) + +func getTestProjectID() string { + if p := os.Getenv("FIRESTORE_PROJECT_ID"); p != "" { + return p + } + return testProjectID +} + +func getEmulatorHost() string { + return os.Getenv("FIRESTORE_EMULATOR_HOST") +} + +func newTestClient(ctx context.Context) (*firestore.Client, error) { + projectID := getTestProjectID() + var opts []option.ClientOption + + // When using emulator, no credentials needed + if getEmulatorHost() != "" { + opts = append(opts, option.WithoutAuthentication()) + } + + return firestore.NewClient(ctx, projectID, opts...) +} + +func clearCollection(ctx context.Context, client *firestore.Client, collectionName string) error { + col := client.Collection(collectionName) + iter := col.Documents(ctx) + defer iter.Stop() + + batch := client.Batch() + count := 0 + + for { + doc, err := iter.Next() + if err == iterator.Done { + break + } + if err != nil { + return err + } + batch.Delete(doc.Ref) + count++ + + // Firestore batch limit is 500 + if count >= 500 { + if _, err := batch.Commit(ctx); err != nil { + return err + } + batch = client.Batch() + count = 0 + } + } + + if count > 0 { + if _, err := batch.Commit(ctx); err != nil { + return err + } + } + + return nil +} + +func TestFirestoreConnectorSuite(t *testing.T) { + if getEmulatorHost() == "" { + t.Skip("Skipping integration test: FIRESTORE_EMULATOR_HOST not set") + } + + ctx := context.Background() + projectID := getTestProjectID() + collectionName := namespaceToCollection(testNamespace) + + // Create a direct client for test setup/verification + testClient, err := newTestClient(ctx) + if err != nil { + t.Fatalf("Failed to create test client: %v", err) + } + defer testClient.Close() + + tSuite := pkgtest.NewConnectorTestSuite( + testNamespace, + func() adiomv1connect.ConnectorServiceClient { + conn, err := NewConn(ctx, ConnectorSettings{ + Uri: "firestore://" + projectID, + BatchSize: 500, + }) + if err != nil { + panic(err) + } + return pkgtest.ClientFromHandler(conn) + }, + // Bootstrap: clear collection + func(ctx context.Context) error { + return clearCollection(ctx, testClient, collectionName) + }, + // InsertUpdates: not applicable for sink-only + nil, + 0, // NumPages - sink only, no source + 0, // NumItems - sink only, no source + ) + + // Sink-only connector, skip source tests + tSuite.SkipDuplicateTest = true + tSuite.SkipWriteUpdatesTest = false + + // Custom assertion to verify document exists in Firestore + tSuite.AssertExists = func(ctx context.Context, a *assert.Assertions, id []*adiomv1.BsonValue, exists bool) error { + docID, err := extractIDFromBsonValues(id) + if err != nil { + return err + } + + docRef := testClient.Collection(collectionName).Doc(docID) + _, err = docRef.Get(ctx) + + if exists { + a.NoError(err, "Document should exist") + } else { + a.Error(err, "Document should not exist") + } + + return nil + } + + suite.Run(t, tSuite) +} + +// TestFirestoreBatchWrite tests efficient batch writing +func TestFirestoreBatchWrite(t *testing.T) { + if getEmulatorHost() == "" { + t.Skip("Skipping integration test: FIRESTORE_EMULATOR_HOST not set") + } + + ctx := context.Background() + projectID := getTestProjectID() + collectionName := "batch_test" + + // Create connector + connector, err := NewConn(ctx, ConnectorSettings{ + Uri: "firestore://" + projectID, + BatchSize: 100, + }) + assert.NoError(t, err) + defer connector.(interface{ Teardown() }).Teardown() + + // Create test client for verification + testClient, err := newTestClient(ctx) + assert.NoError(t, err) + defer testClient.Close() + + // Clear collection first + err = clearCollection(ctx, testClient, collectionName) + assert.NoError(t, err) + + // Create test data - 250 documents to test batching (will need 3 batches with batch size 100) + client := pkgtest.ClientFromHandler(connector) + var data [][]byte + for i := 0; i < 250; i++ { + doc := map[string]any{ + "id": i, // JSON_ID uses "id" field + "value": i * 10, + } + encoded, _ := encodeJSON(doc) + data = append(data, encoded) + } + + // Write data + _, err = client.WriteData(ctx, connect.NewRequest(&adiomv1.WriteDataRequest{ + Namespace: collectionName, + Data: data, + Type: adiomv1.DataType_DATA_TYPE_JSON_ID, + })) + assert.NoError(t, err) + + // Verify all documents were written + iter := testClient.Collection(collectionName).Documents(ctx) + defer iter.Stop() + + count := 0 + for { + _, err := iter.Next() + if err == iterator.Done { + break + } + assert.NoError(t, err) + count++ + } + assert.Equal(t, 250, count, "All 250 documents should be written") +} + +// TestFirestoreDeterministicOverwrite tests that writes deterministically overwrite existing data +func TestFirestoreDeterministicOverwrite(t *testing.T) { + if getEmulatorHost() == "" { + t.Skip("Skipping integration test: FIRESTORE_EMULATOR_HOST not set") + } + + ctx := context.Background() + projectID := getTestProjectID() + collectionName := "overwrite_test" + + // Create connector + connector, err := NewConn(ctx, ConnectorSettings{ + Uri: "firestore://" + projectID, + BatchSize: 500, + }) + assert.NoError(t, err) + defer connector.(interface{ Teardown() }).Teardown() + + // Create test client for verification + testClient, err := newTestClient(ctx) + assert.NoError(t, err) + defer testClient.Close() + + // Clear collection first + err = clearCollection(ctx, testClient, collectionName) + assert.NoError(t, err) + + client := pkgtest.ClientFromHandler(connector) + + // Write initial document - JSON_ID uses "id" field + doc1 := map[string]any{ + "id": "test-doc", + "field1": "value1", + "field2": "value2", + } + encoded1, _ := encodeJSON(doc1) + _, err = client.WriteData(ctx, connect.NewRequest(&adiomv1.WriteDataRequest{ + Namespace: collectionName, + Data: [][]byte{encoded1}, + Type: adiomv1.DataType_DATA_TYPE_JSON_ID, + })) + assert.NoError(t, err) + + // Overwrite with different data (missing field2) - JSON_ID uses "id" field + doc2 := map[string]any{ + "id": "test-doc", + "field1": "updated_value", + "field3": "new_field", + } + encoded2, _ := encodeJSON(doc2) + _, err = client.WriteData(ctx, connect.NewRequest(&adiomv1.WriteDataRequest{ + Namespace: collectionName, + Data: [][]byte{encoded2}, + Type: adiomv1.DataType_DATA_TYPE_JSON_ID, + })) + assert.NoError(t, err) + + // Verify the document was completely overwritten + docSnap, err := testClient.Collection(collectionName).Doc("test-doc").Get(ctx) + assert.NoError(t, err) + + data := docSnap.Data() + assert.Equal(t, "updated_value", data["field1"], "field1 should be updated") + assert.Equal(t, "new_field", data["field3"], "field3 should exist") + _, hasField2 := data["field2"] + assert.False(t, hasField2, "field2 should NOT exist (complete overwrite)") +} + +// TestFirestoreWriteUpdatesDelete tests delete operations +func TestFirestoreWriteUpdatesDelete(t *testing.T) { + if getEmulatorHost() == "" { + t.Skip("Skipping integration test: FIRESTORE_EMULATOR_HOST not set") + } + + ctx := context.Background() + projectID := getTestProjectID() + collectionName := "delete_test" + + // Create connector + connector, err := NewConn(ctx, ConnectorSettings{ + Uri: "firestore://" + projectID, + BatchSize: 500, + }) + assert.NoError(t, err) + defer connector.(interface{ Teardown() }).Teardown() + + // Create test client for verification + testClient, err := newTestClient(ctx) + assert.NoError(t, err) + defer testClient.Close() + + // Clear collection first + err = clearCollection(ctx, testClient, collectionName) + assert.NoError(t, err) + + client := pkgtest.ClientFromHandler(connector) + + // Write a document first - JSON_ID uses "id" field + doc := map[string]any{ + "id": "to-delete", + "data": "some data", + } + encoded, _ := encodeJSON(doc) + _, err = client.WriteData(ctx, connect.NewRequest(&adiomv1.WriteDataRequest{ + Namespace: collectionName, + Data: [][]byte{encoded}, + Type: adiomv1.DataType_DATA_TYPE_JSON_ID, + })) + assert.NoError(t, err) + + // Verify document exists + _, err = testClient.Collection(collectionName).Doc("to-delete").Get(ctx) + assert.NoError(t, err, "Document should exist before delete") + + // Delete using WriteUpdates - need to properly encode BSON value + idType, idData, _ := bson.MarshalValue("to-delete") + _, err = client.WriteUpdates(ctx, connect.NewRequest(&adiomv1.WriteUpdatesRequest{ + Namespace: collectionName, + Updates: []*adiomv1.Update{{ + Id: []*adiomv1.BsonValue{{ + Data: idData, + Type: uint32(idType), + Name: "_id", + }}, + Type: adiomv1.UpdateType_UPDATE_TYPE_DELETE, + }}, + Type: adiomv1.DataType_DATA_TYPE_JSON_ID, + })) + assert.NoError(t, err) + + // Verify document was deleted + _, err = testClient.Collection(collectionName).Doc("to-delete").Get(ctx) + assert.Error(t, err, "Document should not exist after delete") +} + +// Helper to encode map to JSON +func encodeJSON(v any) ([]byte, error) { + return json.Marshal(v) +} diff --git a/connectors/firestore/connector_test.go b/connectors/firestore/connector_test.go new file mode 100644 index 00000000..64166791 --- /dev/null +++ b/connectors/firestore/connector_test.go @@ -0,0 +1,125 @@ +/* + * Copyright (C) 2024 Adiom, Inc. + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + +package firestore + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestParseFirestoreURI(t *testing.T) { + tests := []struct { + name string + uri string + wantProjectID string + wantDatabaseID string + wantErr error + }{ + { + name: "basic project ID", + uri: "firestore://my-project", + wantProjectID: "my-project", + wantDatabaseID: "(default)", + wantErr: nil, + }, + { + name: "project ID with database", + uri: "firestore://my-project/my-database", + wantProjectID: "my-project", + wantDatabaseID: "my-database", + wantErr: nil, + }, + { + name: "uppercase prefix", + uri: "FIRESTORE://my-project", + wantProjectID: "my-project", + wantDatabaseID: "(default)", + wantErr: nil, + }, + { + name: "mixed case prefix", + uri: "Firestore://my-project/db", + wantProjectID: "my-project", + wantDatabaseID: "db", + wantErr: nil, + }, + { + name: "invalid prefix", + uri: "mongodb://my-project", + wantErr: ErrInvalidURI, + }, + { + name: "missing project ID", + uri: "firestore://", + wantErr: ErrProjectIDRequired, + }, + { + name: "empty string", + uri: "", + wantErr: ErrInvalidURI, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + projectID, databaseID, err := parseFirestoreURI(tt.uri) + + if tt.wantErr != nil { + assert.ErrorIs(t, err, tt.wantErr) + return + } + + assert.NoError(t, err) + assert.Equal(t, tt.wantProjectID, projectID) + assert.Equal(t, tt.wantDatabaseID, databaseID) + }) + } +} + +func TestNamespaceToCollection(t *testing.T) { + tests := []struct { + namespace string + want string + }{ + {"mydb.users", "mydb_users"}, + {"db.collection", "db_collection"}, + {"simple", "simple"}, + {"a.b.c", "a_b_c"}, + {"", ""}, + } + + for _, tt := range tests { + t.Run(tt.namespace, func(t *testing.T) { + got := namespaceToCollection(tt.namespace) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestValueToString(t *testing.T) { + tests := []struct { + name string + value any + want string + }{ + {"string", "hello", "hello"}, + {"int", 42, "42"}, + {"int32", int32(42), "42"}, + {"int64", int64(42), "42"}, + {"uint", uint(42), "42"}, + {"float64", 3.14, "3.14"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := valueToString(tt.value) + assert.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +} diff --git a/go.mod b/go.mod index 1cbebc4d..b0c20f2a 100644 --- a/go.mod +++ b/go.mod @@ -41,12 +41,18 @@ require ( go.mongodb.org/mongo-driver/v2 v2.5.0 golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b golang.org/x/net v0.46.0 - golang.org/x/time v0.12.0 - google.golang.org/grpc v1.73.0 - google.golang.org/protobuf v1.36.6 + golang.org/x/time v0.14.0 + google.golang.org/grpc v1.76.0 + google.golang.org/protobuf v1.36.10 ) require ( + cloud.google.com/go v0.123.0 // indirect + cloud.google.com/go/auth v0.17.0 // indirect + cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect + cloud.google.com/go/compute/metadata v0.9.0 // indirect + cloud.google.com/go/firestore v1.21.0 // indirect + cloud.google.com/go/longrunning v0.7.0 // indirect github.com/acobaugh/osrelease v0.0.0-20181218015638-a93a0a55a249 // indirect github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 // indirect github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.3 // indirect @@ -67,7 +73,10 @@ require ( github.com/eapache/go-resiliency v1.7.0 // indirect github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect github.com/eapache/queue v1.1.0 // indirect + github.com/felixge/httpsnoop v1.0.4 // indirect github.com/gdamore/encoding v1.0.1 // indirect + github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect github.com/go-openapi/analysis v0.23.0 // indirect github.com/go-openapi/errors v0.22.1 // indirect github.com/go-openapi/jsonpointer v0.21.1 // indirect @@ -80,6 +89,9 @@ require ( github.com/go-openapi/validate v0.24.0 // indirect github.com/golang-sql/civil v0.0.0-20220223132316-b832511892a9 // indirect github.com/golang-sql/sqlexp v0.1.0 // indirect + github.com/google/s2a-go v0.1.9 // indirect + github.com/googleapis/enterprise-certificate-proxy v0.3.7 // indirect + github.com/googleapis/gax-go/v2 v2.15.0 // indirect github.com/hashicorp/go-uuid v1.0.3 // indirect github.com/jackc/pgio v1.0.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect @@ -107,10 +119,19 @@ require ( github.com/spf13/afero v1.11.0 // indirect github.com/stretchr/objx v0.5.2 // indirect go.mongodb.org/mongo-driver v1.17.4 // indirect - golang.org/x/oauth2 v0.30.0 // indirect + go.opentelemetry.io/auto/sdk v1.1.0 // indirect + go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.61.0 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 // indirect + go.opentelemetry.io/otel v1.37.0 // indirect + go.opentelemetry.io/otel/metric v1.37.0 // indirect + go.opentelemetry.io/otel/trace v1.37.0 // indirect + golang.org/x/oauth2 v0.33.0 // indirect golang.org/x/sys v0.37.0 // indirect golang.org/x/term v0.36.0 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20250603155806-513f23925822 // indirect + google.golang.org/api v0.256.0 // indirect + google.golang.org/genproto v0.0.0-20250603155806-513f23925822 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20251111163417-95abcf5c77ba // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba // indirect gopkg.in/yaml.v2 v2.4.0 // indirect ) @@ -129,7 +150,7 @@ require ( github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 // indirect github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect golang.org/x/crypto v0.43.0 // indirect - golang.org/x/sync v0.17.0 + golang.org/x/sync v0.18.0 golang.org/x/text v0.30.0 // indirect gopkg.in/yaml.v3 v3.0.1 ) diff --git a/go.sum b/go.sum index 72213352..75c3d936 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,15 @@ +cloud.google.com/go v0.123.0 h1:2NAUJwPR47q+E35uaJeYoNhuNEM9kM8SjgRgdeOJUSE= +cloud.google.com/go v0.123.0/go.mod h1:xBoMV08QcqUGuPW65Qfm1o9Y4zKZBpGS+7bImXLTAZU= +cloud.google.com/go/auth v0.17.0 h1:74yCm7hCj2rUyyAocqnFzsAYXgJhrG26XCFimrc/Kz4= +cloud.google.com/go/auth v0.17.0/go.mod h1:6wv/t5/6rOPAX4fJiRjKkJCvswLwdet7G8+UGXt7nCQ= +cloud.google.com/go/auth/oauth2adapt v0.2.8 h1:keo8NaayQZ6wimpNSmW5OPc283g65QNIiLpZnkHRbnc= +cloud.google.com/go/auth/oauth2adapt v0.2.8/go.mod h1:XQ9y31RkqZCcwJWNSx2Xvric3RrU88hAYYbjDWYDL+c= +cloud.google.com/go/compute/metadata v0.9.0 h1:pDUj4QMoPejqq20dK0Pg2N4yG9zIkYGdBtwLoEkH9Zs= +cloud.google.com/go/compute/metadata v0.9.0/go.mod h1:E0bWwX5wTnLPedCKqk3pJmVgCBSM6qQI1yTBdEb3C10= +cloud.google.com/go/firestore v1.21.0 h1:BhopUsx7kh6NFx77ccRsHhrtkbJUmDAxNY3uapWdjcM= +cloud.google.com/go/firestore v1.21.0/go.mod h1:1xH6HNcnkf/gGyR8udd6pFO4Z7GWJSwLKQMx/u6UrP4= +cloud.google.com/go/longrunning v0.7.0 h1:FV0+SYF1RIj59gyoWDRi45GiYUMM3K1qO51qoboQT1E= +cloud.google.com/go/longrunning v0.7.0/go.mod h1:ySn2yXmjbK9Ba0zsQqunhDkYi0+9rlXIwnoAf+h+TPY= connectrpc.com/connect v1.18.1 h1:PAg7CjSAGvscaf6YZKUefjoih5Z/qYkyaTrBW8xvYPw= connectrpc.com/connect v1.18.1/go.mod h1:0292hj1rnx8oFrStN7cB4jjVBeqs+Yx5yDIC2prWDO8= connectrpc.com/grpcreflect v1.3.0 h1:Y4V+ACf8/vOb1XOc251Qun7jMB75gCUNw6llvB9csXc= @@ -99,14 +111,19 @@ github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 h1:Oy0F4A github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3/go.mod h1:YvSRo5mw33fLEx1+DlK6L2VV43tJt5Eyel9n9XBcR+0= github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc= github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I= +github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= +github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw= github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g= github.com/gdamore/encoding v1.0.1 h1:YzKZckdBL6jVt2Gc+5p82qhrGiqMdG/eNs6Wy0u3Uhw= github.com/gdamore/encoding v1.0.1/go.mod h1:0Z0cMFinngz9kS1QfMjCP8TY7em3bZYeeklsSDPivEo= github.com/gdamore/tcell/v2 v2.8.1 h1:KPNxyqclpWpWQlPLx6Xui1pMk8S+7+R37h3g07997NU= github.com/gdamore/tcell/v2 v2.8.1/go.mod h1:bj8ori1BG3OYMjmb3IklZVWfZUJ1UBQt9JXrOCOhGWw= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY= github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/go-openapi/analysis v0.21.2/go.mod h1:HZwRk4RRisyG8vx2Oe6aqeSQcoxRp47Xkp3+K6q+LdY= @@ -192,9 +209,15 @@ github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/ github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/s2a-go v0.1.9 h1:LGD7gtMgezd8a/Xak7mEWL0PjoTQFvpRudN895yqKW0= +github.com/google/s2a-go v0.1.9/go.mod h1:YA0Ei2ZQL3acow2O62kdp9UlnvMmU7kA6Eutn0dXayM= github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/googleapis/enterprise-certificate-proxy v0.3.7 h1:zrn2Ee/nWmHulBx5sAVrGgAa0f2/R35S4DJwfFaUPFQ= +github.com/googleapis/enterprise-certificate-proxy v0.3.7/go.mod h1:MkHOF77EYAE7qfSuSS9PU6g4Nt4e11cnsDUowfwewLA= +github.com/googleapis/gax-go/v2 v2.15.0 h1:SyjDc1mGgZU5LncH8gimWo9lW1DtIfPibOG81vgd/bo= +github.com/googleapis/gax-go/v2 v2.15.0/go.mod h1:zVVkkxAQHa1RQpg9z2AUCMnKhi0Qld9rcmyfL1OZhoc= github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4= github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM= github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= @@ -404,16 +427,28 @@ go.mongodb.org/mongo-driver/v2 v2.5.0 h1:yXUhImUjjAInNcpTcAlPHiT7bIXhshCTL3jVBkF go.mongodb.org/mongo-driver/v2 v2.5.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0= go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A= +go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.61.0 h1:q4XOmH/0opmeuJtPsbFNivyl7bCt7yRBbeEm2sC/XtQ= +go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.61.0/go.mod h1:snMWehoOh2wsEwnvvwtDyFCxVeDAODenXHtn5vzrKjo= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 h1:F7Jx+6hwnZ41NSFTO5q4LYDtJRXBf2PD0rNBkeB/lus= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0/go.mod h1:UHB22Z8QsdRDrnAtX4PntOl36ajSxcdUMt1sF7Y6E7Q= go.opentelemetry.io/otel v1.35.0 h1:xKWKPxrxB6OtMCbmMY021CqC45J+3Onta9MqjhnusiQ= go.opentelemetry.io/otel v1.35.0/go.mod h1:UEqy8Zp11hpkUrL73gSlELM0DupHoiq72dR+Zqel/+Y= +go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ= +go.opentelemetry.io/otel v1.37.0/go.mod h1:ehE/umFRLnuLa/vSccNq9oS1ErUlkkK71gMcN34UG8I= go.opentelemetry.io/otel/metric v1.35.0 h1:0znxYu2SNyuMSQT4Y9WDWej0VpcsxkuklLa4/siN90M= go.opentelemetry.io/otel/metric v1.35.0/go.mod h1:nKVFgxBZ2fReX6IlyW28MgZojkoAkJGaE8CpgeAU3oE= +go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/WgbsdpcPoZE= +go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E= go.opentelemetry.io/otel/sdk v1.35.0 h1:iPctf8iprVySXSKJffSS79eOjl9pvxV9ZqOWT0QejKY= go.opentelemetry.io/otel/sdk v1.35.0/go.mod h1:+ga1bZliga3DxJ3CQGg3updiaAJoNECOgJREo9KHGQg= +go.opentelemetry.io/otel/sdk v1.37.0 h1:ItB0QUqnjesGRvNcmAcU0LyvkVyGJ2xftD29bWdDvKI= go.opentelemetry.io/otel/sdk/metric v1.35.0 h1:1RriWBmCKgkeHEhM7a2uMjMUfP7MsOF5JpUCaEqEI9o= go.opentelemetry.io/otel/sdk/metric v1.35.0/go.mod h1:is6XYCUMpcKi+ZsOvfluY5YstFnhW0BidkR+gL+qN+w= +go.opentelemetry.io/otel/sdk/metric v1.37.0 h1:90lI228XrB9jCMuSdA0673aubgRobVZFhbjxHHspCPc= go.opentelemetry.io/otel/trace v1.35.0 h1:dPpEfJu1sDIqruz7BHFG3c7528f6ddfSWfFDVt/xgMs= go.opentelemetry.io/otel/trace v1.35.0/go.mod h1:WUk7DtFp1Aw2MkvqGdwiXYDZZNvA/1J8o6xRXLrIkyc= +go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4= +go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0= golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20190422162423-af44ce270edf/go.mod h1:WFFai1msRO1wXaEeE5yQxYXgSfI8pQAWXbQop6sCtWE= @@ -451,6 +486,8 @@ golang.org/x/net v0.46.0 h1:giFlY12I07fugqwPuWJi68oOnpfqFnJIJzaIIm2JVV4= golang.org/x/net v0.46.0/go.mod h1:Q9BGdFy1y4nkUwiLvT5qtyhAnEHgnQ/zd8PfU6nc210= golang.org/x/oauth2 v0.30.0 h1:dnDm7JmhM45NNpd8FDDeLhK6FwqbOf4MLCM9zb1BOHI= golang.org/x/oauth2 v0.30.0/go.mod h1:B++QgG3ZKulg6sRPGD/mqlHQs5rB3Ml9erfeDY7xKlU= +golang.org/x/oauth2 v0.33.0 h1:4Q+qn+E5z8gPRJfmRy7C2gGG3T4jIprK6aSYgTXGRpo= +golang.org/x/oauth2 v0.33.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190412183630-56d357773e84/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -463,6 +500,8 @@ golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I= +golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190403152447-81d4e9dc473e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -513,6 +552,8 @@ golang.org/x/text v0.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k= golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM= golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE= golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg= +golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI= +golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20190329151228-23e29df326fe/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= golang.org/x/tools v0.0.0-20190416151739-9c9e1878f421/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= @@ -525,12 +566,24 @@ golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58 golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/api v0.256.0 h1:u6Khm8+F9sxbCTYNoBHg6/Hwv0N/i+V94MvkOSor6oI= +google.golang.org/api v0.256.0/go.mod h1:KIgPhksXADEKJlnEoRa9qAII4rXcy40vfI8HRqcU964= +google.golang.org/genproto v0.0.0-20250603155806-513f23925822 h1:rHWScKit0gvAPuOnu87KpaYtjK5zBMLcULh7gxkCXu4= +google.golang.org/genproto v0.0.0-20250603155806-513f23925822/go.mod h1:HubltRL7rMh0LfnQPkMH4NPDFEWp0jw3vixw7jEM53s= +google.golang.org/genproto/googleapis/api v0.0.0-20251111163417-95abcf5c77ba h1:B14OtaXuMaCQsl2deSvNkyPKIzq3BjfxQp8d00QyWx4= +google.golang.org/genproto/googleapis/api v0.0.0-20251111163417-95abcf5c77ba/go.mod h1:G5IanEx8/PgI9w6CFcYQf7jMtHQhZruvfM1i3qOqk5U= google.golang.org/genproto/googleapis/rpc v0.0.0-20250603155806-513f23925822 h1:fc6jSaCT0vBduLYZHYrBBNY4dsWuvgyff9noRNDdBeE= google.golang.org/genproto/googleapis/rpc v0.0.0-20250603155806-513f23925822/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A= +google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba h1:UKgtfRM7Yh93Sya0Fo8ZzhDP4qBckrrxEr2oF5UIVb8= +google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk= google.golang.org/grpc v1.73.0 h1:VIWSmpI2MegBtTuFt5/JWy2oXxtjJ/e89Z70ImfD2ok= google.golang.org/grpc v1.73.0/go.mod h1:50sbHOUqWoCQGI8V2HQLJM0B+LMlIUjNSZmow7EVBQc= +google.golang.org/grpc v1.76.0 h1:UnVkv1+uMLYXoIz6o7chp59WfQUYA2ex/BXQ9rHZu7A= +google.golang.org/grpc v1.76.0/go.mod h1:Ju12QI8M6iQJtbcsV+awF5a4hfJMLi4X0JLo94ULZ6c= google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY= google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY= +google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= +google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/internal/app/options/connectorflags.go b/internal/app/options/connectorflags.go index 61fcb478..68c34acc 100644 --- a/internal/app/options/connectorflags.go +++ b/internal/app/options/connectorflags.go @@ -15,6 +15,7 @@ import ( "github.com/adiom-data/dsync/connectors/cosmos" "github.com/adiom-data/dsync/connectors/dynamodb" fileconnector "github.com/adiom-data/dsync/connectors/file" + "github.com/adiom-data/dsync/connectors/firestore" "github.com/adiom-data/dsync/connectors/kafka" "github.com/adiom-data/dsync/connectors/mongo" "github.com/adiom-data/dsync/connectors/null" @@ -415,6 +416,21 @@ func GetRegisteredConnectors() []RegisteredConnector { } }), }, + { + Name: "Firestore", + IsConnector: func(s string) bool { + return strings.HasPrefix(strings.ToLower(s), "firestore://") + }, + Create: func(args []string, as AdditionalSettings) (adiomv1connect.ConnectorServiceHandler, []string, error) { + if len(args) == 0 { + return nil, nil, fmt.Errorf("missing firestore connection string: %w", ErrMissingConnector) + } + settings := firestore.ConnectorSettings{Uri: args[0]} + return CreateHelper("Firestore", "firestore://project-id[/database-id] [options]", FirestoreFlags(&settings), func(c *cli.Context, _ []string, _ AdditionalSettings) (adiomv1connect.ConnectorServiceHandler, error) { + return firestore.NewConn(c.Context, settings) + })(args, as) + }, + }, { Name: "s3vectors", IsConnector: func(s string) bool { @@ -1133,3 +1149,24 @@ func insecureClient() *http.Client { }, } } + +func FirestoreFlags(settings *firestore.ConnectorSettings) []cli.Flag { + return []cli.Flag{ + altsrc.NewStringFlag(&cli.StringFlag{ + Name: "credentials-file", + Usage: "Path to GCP service account JSON credentials file", + Destination: &settings.CredentialsFile, + }), + altsrc.NewIntFlag(&cli.IntFlag{ + Name: "batch-size", + Usage: "Maximum documents per batch write (max 500)", + Value: 500, + Destination: &settings.BatchSize, + }), + altsrc.NewStringFlag(&cli.StringFlag{ + Name: "id", + Usage: "A fixed id for the connector", + Destination: &settings.ID, + }), + } +} From 2894d5abff97431e38c84ec898153473e63ac28c Mon Sep 17 00:00:00 2001 From: Alexander Komyagin Date: Thu, 2 Apr 2026 10:22:28 -0700 Subject: [PATCH 2/3] bson types conversion --- connectors/firestore/bson_conversion_test.go | 190 ++++++++++++++++++ connectors/firestore/connector.go | 86 ++++++++ .../firestore/connector_integration_test.go | 153 ++++++++++++++ 3 files changed, 429 insertions(+) create mode 100644 connectors/firestore/bson_conversion_test.go diff --git a/connectors/firestore/bson_conversion_test.go b/connectors/firestore/bson_conversion_test.go new file mode 100644 index 00000000..637a6b9f --- /dev/null +++ b/connectors/firestore/bson_conversion_test.go @@ -0,0 +1,190 @@ +/* + * Copyright (C) 2024 Adiom, Inc. + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + +package firestore + +import ( + "testing" + "time" + + adiomv1 "github.com/adiom-data/dsync/gen/adiom/v1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.mongodb.org/mongo-driver/v2/bson" +) + +func TestBsonTypeConversion(t *testing.T) { + tests := []struct { + name string + input map[string]any + expected map[string]any + }{ + { + name: "ObjectID field", + input: map[string]any{ + "_id": bson.ObjectID{0x66, 0xf4, 0xc6, 0x92, 0x91, 0xab, 0x1a, 0x55, 0x33, 0x94, 0x5e, 0x37}, + "field1": "value1", + }, + expected: map[string]any{ + "_id": "66f4c69291ab1a5533945e37", + "field1": "value1", + }, + }, + { + name: "DateTime field", + input: map[string]any{ + "_id": "doc1", + "createdAt": bson.DateTime(1704067200000), // 2024-01-01 00:00:00 UTC + }, + expected: map[string]any{ + "_id": "doc1", + "createdAt": time.UnixMilli(1704067200000).UTC(), + }, + }, + { + name: "Binary field", + input: map[string]any{ + "_id": "doc1", + "data": bson.Binary{Subtype: bson.TypeBinaryGeneric, Data: []byte{0x01, 0x02, 0x03}}, + }, + expected: map[string]any{ + "_id": "doc1", + "data": []byte{0x01, 0x02, 0x03}, + }, + }, + { + name: "Decimal128 field", + input: map[string]any{ + "_id": "doc1", + "price": mustParseDecimal128("123.45"), + }, + expected: map[string]any{ + "_id": "doc1", + "price": "123.45", + }, + }, + { + name: "Nested document with BSON types", + input: map[string]any{ + "_id": "doc1", + "nested": bson.M{ + "objectId": bson.ObjectID{0x66, 0xf4, 0xc6, 0x92, 0x91, 0xab, 0x1a, 0x55, 0x33, 0x94, 0x5e, 0x37}, + "date": bson.DateTime(1704067200000), + }, + }, + expected: map[string]any{ + "_id": "doc1", + "nested": map[string]any{ + "objectId": "66f4c69291ab1a5533945e37", + "date": time.UnixMilli(1704067200000).UTC(), + }, + }, + }, + { + name: "Array with BSON types", + input: map[string]any{ + "_id": "doc1", + "ids": bson.A{ + bson.ObjectID{0x66, 0xf4, 0xc6, 0x92, 0x91, 0xab, 0x1a, 0x55, 0x33, 0x94, 0x5e, 0x37}, + bson.ObjectID{0x66, 0xf4, 0xc6, 0x92, 0x91, 0xab, 0x1a, 0x55, 0x33, 0x94, 0x5e, 0x38}, + }, + }, + expected: map[string]any{ + "_id": "doc1", + "ids": []any{ + "66f4c69291ab1a5533945e37", + "66f4c69291ab1a5533945e38", + }, + }, + }, + { + name: "bson.D document", + input: map[string]any{ + "_id": "doc1", + "doc": bson.D{ + {Key: "name", Value: "test"}, + {Key: "id", Value: bson.ObjectID{0x66, 0xf4, 0xc6, 0x92, 0x91, 0xab, 0x1a, 0x55, 0x33, 0x94, 0x5e, 0x37}}, + }, + }, + expected: map[string]any{ + "_id": "doc1", + "doc": map[string]any{ + "name": "test", + "id": "66f4c69291ab1a5533945e37", + }, + }, + }, + { + name: "Primitive types unchanged", + input: map[string]any{ + "_id": "doc1", + "str": "hello", + "int32": int32(42), + "int64": int64(9999999999), + "float64": 3.14, + "bool": true, + "null": nil, + }, + expected: map[string]any{ + "_id": "doc1", + "str": "hello", + "int32": int32(42), + "int64": int64(9999999999), + "float64": 3.14, + "bool": true, + "null": nil, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result, err := convertBsonTypes(tt.input) + require.NoError(t, err) + assert.Equal(t, tt.expected, result) + }) + } +} + +func TestRawToMapWithBsonTypes(t *testing.T) { + // Create a BSON document with various types + doc := bson.D{ + {Key: "_id", Value: bson.ObjectID{0x66, 0xf4, 0xc6, 0x92, 0x91, 0xab, 0x1a, 0x55, 0x33, 0x94, 0x5e, 0x37}}, + {Key: "name", Value: "test"}, + {Key: "createdAt", Value: bson.DateTime(1704067200000)}, + {Key: "nested", Value: bson.D{ + {Key: "ref", Value: bson.ObjectID{0x66, 0xf4, 0xc6, 0x92, 0x91, 0xab, 0x1a, 0x55, 0x33, 0x94, 0x5e, 0x38}}, + }}, + } + + raw, err := bson.Marshal(doc) + require.NoError(t, err) + + result, err := rawToMap(raw, adiomv1.DataType_DATA_TYPE_MONGO_BSON) + require.NoError(t, err) + + // Check that ObjectID was converted to string + assert.Equal(t, "66f4c69291ab1a5533945e37", result["_id"]) + assert.Equal(t, "test", result["name"]) + + // Check DateTime was converted to time.Time + createdAt, ok := result["createdAt"].(time.Time) + assert.True(t, ok, "createdAt should be time.Time") + assert.Equal(t, int64(1704067200000), createdAt.UnixMilli()) + + // Check nested ObjectID was converted + nested, ok := result["nested"].(map[string]any) + assert.True(t, ok, "nested should be map[string]any") + assert.Equal(t, "66f4c69291ab1a5533945e38", nested["ref"]) +} + +func mustParseDecimal128(s string) bson.Decimal128 { + d, err := bson.ParseDecimal128(s) + if err != nil { + panic(err) + } + return d +} diff --git a/connectors/firestore/connector.go b/connectors/firestore/connector.go index 2e097a80..ecfba6f2 100644 --- a/connectors/firestore/connector.go +++ b/connectors/firestore/connector.go @@ -13,6 +13,7 @@ import ( "fmt" "log/slog" "strings" + "time" "cloud.google.com/go/firestore" "connectrpc.com/connect" @@ -302,6 +303,12 @@ func rawToMap(raw []byte, dataType adiomv1.DataType) (map[string]any, error) { if err := bson.Unmarshal(raw, &docData); err != nil { return nil, fmt.Errorf("failed to unmarshal BSON: %w", err) } + // Convert BSON-specific types to Firestore-compatible types + converted, err := convertBsonTypes(docData) + if err != nil { + return nil, fmt.Errorf("failed to convert BSON types: %w", err) + } + docData = converted case adiomv1.DataType_DATA_TYPE_JSON_ID: if err := json.Unmarshal(raw, &docData); err != nil { return nil, fmt.Errorf("failed to unmarshal JSON: %w", err) @@ -313,6 +320,85 @@ func rawToMap(raw []byte, dataType adiomv1.DataType) (map[string]any, error) { return docData, nil } +// convertBsonTypes recursively converts BSON-specific types to Firestore-compatible types +func convertBsonTypes(m map[string]any) (map[string]any, error) { + result := make(map[string]any, len(m)) + for k, v := range m { + converted, err := convertBsonValue(v) + if err != nil { + return nil, fmt.Errorf("failed to convert field %q: %w", k, err) + } + result[k] = converted + } + return result, nil +} + +func convertBsonValue(v any) (any, error) { + switch val := v.(type) { + case nil: + return nil, nil + case bool, int32, int64, float64, string: + return val, nil + case bson.ObjectID: + return val.Hex(), nil + case bson.DateTime: + return val.Time().UTC(), nil + case bson.Binary: + return val.Data, nil + case bson.Decimal128: + return val.String(), nil + case bson.Timestamp: + return time.Unix(int64(val.T), 0).UTC(), nil + case bson.Regex: + return map[string]any{"pattern": val.Pattern, "options": val.Options}, nil + case bson.D: + m := make(map[string]any, len(val)) + for _, elem := range val { + converted, err := convertBsonValue(elem.Value) + if err != nil { + return nil, err + } + m[elem.Key] = converted + } + return m, nil + case bson.M: + m := make(map[string]any, len(val)) + for k, elem := range val { + converted, err := convertBsonValue(elem) + if err != nil { + return nil, err + } + m[k] = converted + } + return m, nil + case bson.A: + arr := make([]any, len(val)) + for i, elem := range val { + converted, err := convertBsonValue(elem) + if err != nil { + return nil, err + } + arr[i] = converted + } + return arr, nil + case map[string]any: + return convertBsonTypes(val) + case []any: + arr := make([]any, len(val)) + for i, elem := range val { + converted, err := convertBsonValue(elem) + if err != nil { + return nil, err + } + arr[i] = converted + } + return arr, nil + default: + // For unknown types, return as-is and let Firestore handle it + return val, nil + } +} + func valueToString(v any) (string, error) { switch val := v.(type) { case string: diff --git a/connectors/firestore/connector_integration_test.go b/connectors/firestore/connector_integration_test.go index d89e6e6d..0abb7c3f 100644 --- a/connectors/firestore/connector_integration_test.go +++ b/connectors/firestore/connector_integration_test.go @@ -14,6 +14,7 @@ import ( "encoding/json" "os" "testing" + "time" "cloud.google.com/go/firestore" "connectrpc.com/connect" @@ -21,6 +22,7 @@ import ( "github.com/adiom-data/dsync/gen/adiom/v1/adiomv1connect" pkgtest "github.com/adiom-data/dsync/pkg/test" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" "go.mongodb.org/mongo-driver/v2/bson" "google.golang.org/api/iterator" @@ -359,6 +361,157 @@ func TestFirestoreWriteUpdatesDelete(t *testing.T) { assert.Error(t, err, "Document should not exist after delete") } +// TestFirestoreComplexBsonTypes tests writing complex BSON documents with various types +func TestFirestoreComplexBsonTypes(t *testing.T) { + if getEmulatorHost() == "" { + t.Skip("Skipping integration test: FIRESTORE_EMULATOR_HOST not set") + } + + ctx := context.Background() + projectID := getTestProjectID() + collectionName := "complex_bson_test" + + // Create connector + connector, err := NewConn(ctx, ConnectorSettings{ + Uri: "firestore://" + projectID, + BatchSize: 500, + }) + require.NoError(t, err) + defer connector.(interface{ Teardown() }).Teardown() + + // Create test client for verification + testClient, err := newTestClient(ctx) + require.NoError(t, err) + defer testClient.Close() + + // Clear collection first + err = clearCollection(ctx, testClient, collectionName) + require.NoError(t, err) + + client := pkgtest.ClientFromHandler(connector) + + // Create test data with various BSON types + docID := bson.ObjectID{0x66, 0xf4, 0xc6, 0x92, 0x91, 0xab, 0x1a, 0x55, 0x33, 0x94, 0x5e, 0x37} + refID1 := bson.ObjectID{0x66, 0xf4, 0xc6, 0x92, 0x91, 0xab, 0x1a, 0x55, 0x33, 0x94, 0x5e, 0x38} + refID2 := bson.ObjectID{0x66, 0xf4, 0xc6, 0x92, 0x91, 0xab, 0x1a, 0x55, 0x33, 0x94, 0x5e, 0x39} + createdAt := bson.DateTime(1704067200000) // 2024-01-01 00:00:00 UTC + updatedAt := bson.DateTime(1704153600000) // 2024-01-02 00:00:00 UTC + binaryData := []byte{0xDE, 0xAD, 0xBE, 0xEF} + decimal, _ := bson.ParseDecimal128("12345.6789") + + doc := bson.D{ + {Key: "_id", Value: docID}, + {Key: "name", Value: "complex-document"}, + {Key: "createdAt", Value: createdAt}, + {Key: "updatedAt", Value: updatedAt}, + {Key: "binaryData", Value: bson.Binary{Subtype: bson.TypeBinaryGeneric, Data: binaryData}}, + {Key: "price", Value: decimal}, + {Key: "tags", Value: bson.A{"tag1", "tag2", "tag3"}}, + {Key: "refs", Value: bson.A{refID1, refID2}}, + {Key: "metadata", Value: bson.D{ + {Key: "version", Value: int32(42)}, + {Key: "lastModified", Value: updatedAt}, + {Key: "nested", Value: bson.D{ + {Key: "deep", Value: "value"}, + {Key: "deepRef", Value: refID1}, + }}, + }}, + {Key: "active", Value: true}, + {Key: "count", Value: int64(9999999999)}, + {Key: "score", Value: 3.14159}, + {Key: "nullField", Value: nil}, + {Key: "mixedArray", Value: bson.A{ + "string", + int32(123), + true, + createdAt, + refID1, + }}, + } + + // Marshal to BSON + encoded, err := bson.Marshal(doc) + require.NoError(t, err) + + // Write via connector + _, err = client.WriteData(ctx, connect.NewRequest(&adiomv1.WriteDataRequest{ + Namespace: collectionName, + Data: [][]byte{encoded}, + Type: adiomv1.DataType_DATA_TYPE_MONGO_BSON, + })) + require.NoError(t, err) + + // Read back from Firestore and verify + docSnap, err := testClient.Collection(collectionName).Doc(docID.Hex()).Get(ctx) + require.NoError(t, err) + + data := docSnap.Data() + + // Verify string field + assert.Equal(t, "complex-document", data["name"]) + + // Verify DateTime -> time.Time conversion + createdAtResult, ok := data["createdAt"].(time.Time) + assert.True(t, ok, "createdAt should be time.Time, got %T", data["createdAt"]) + assert.Equal(t, int64(1704067200000), createdAtResult.UnixMilli()) + + updatedAtResult, ok := data["updatedAt"].(time.Time) + assert.True(t, ok, "updatedAt should be time.Time") + assert.Equal(t, int64(1704153600000), updatedAtResult.UnixMilli()) + + // Verify Binary -> []byte conversion + binaryResult, ok := data["binaryData"].([]byte) + assert.True(t, ok, "binaryData should be []byte, got %T", data["binaryData"]) + assert.Equal(t, binaryData, binaryResult) + + // Verify Decimal128 -> string conversion + priceResult, ok := data["price"].(string) + assert.True(t, ok, "price should be string, got %T", data["price"]) + assert.Equal(t, "12345.6789", priceResult) + + // Verify string array + tagsResult, ok := data["tags"].([]any) + assert.True(t, ok, "tags should be []any") + assert.Equal(t, []any{"tag1", "tag2", "tag3"}, tagsResult) + + // Verify ObjectID array -> string array conversion + refsResult, ok := data["refs"].([]any) + assert.True(t, ok, "refs should be []any") + assert.Equal(t, []any{refID1.Hex(), refID2.Hex()}, refsResult) + + // Verify nested document + metadataResult, ok := data["metadata"].(map[string]any) + assert.True(t, ok, "metadata should be map[string]any, got %T", data["metadata"]) + assert.Equal(t, int64(42), metadataResult["version"]) // Firestore converts int32 to int64 + lastModified, ok := metadataResult["lastModified"].(time.Time) + assert.True(t, ok, "lastModified should be time.Time") + assert.Equal(t, int64(1704153600000), lastModified.UnixMilli()) + + // Verify deeply nested document + nestedResult, ok := metadataResult["nested"].(map[string]any) + assert.True(t, ok, "nested should be map[string]any") + assert.Equal(t, "value", nestedResult["deep"]) + assert.Equal(t, refID1.Hex(), nestedResult["deepRef"]) + + // Verify primitives + assert.Equal(t, true, data["active"]) + assert.Equal(t, int64(9999999999), data["count"]) + assert.Equal(t, 3.14159, data["score"]) + assert.Nil(t, data["nullField"]) + + // Verify mixed array with type conversions + mixedResult, ok := data["mixedArray"].([]any) + assert.True(t, ok, "mixedArray should be []any") + assert.Len(t, mixedResult, 5) + assert.Equal(t, "string", mixedResult[0]) + assert.Equal(t, int64(123), mixedResult[1]) // Firestore converts int32 to int64 + assert.Equal(t, true, mixedResult[2]) + mixedTime, ok := mixedResult[3].(time.Time) + assert.True(t, ok, "mixed array datetime should be time.Time") + assert.Equal(t, int64(1704067200000), mixedTime.UnixMilli()) + assert.Equal(t, refID1.Hex(), mixedResult[4]) +} + // Helper to encode map to JSON func encodeJSON(v any) ([]byte, error) { return json.Marshal(v) From e232934c39ae60bf7710eceaea8606a1de6d2ab4 Mon Sep 17 00:00:00 2001 From: Alexander Komyagin Date: Thu, 2 Apr 2026 10:28:40 -0700 Subject: [PATCH 3/3] use bulkwriter API --- connectors/firestore/connector.go | 57 +++++++++++++++---- .../firestore/connector_integration_test.go | 27 +++++---- 2 files changed, 58 insertions(+), 26 deletions(-) diff --git a/connectors/firestore/connector.go b/connectors/firestore/connector.go index ecfba6f2..556b2768 100644 --- a/connectors/firestore/connector.go +++ b/connectors/firestore/connector.go @@ -198,22 +198,33 @@ func (c *conn) WriteUpdates(ctx context.Context, r *connect.Request[adiomv1.Writ } func (c *conn) writeBatch(ctx context.Context, collectionName string, data [][]byte, dataType adiomv1.DataType) error { - batch := c.client.Batch() + bw := c.client.BulkWriter(ctx) collection := c.client.Collection(collectionName) + var jobs []*firestore.BulkWriterJob for _, raw := range data { docID, docData, err := extractDocumentIDAndData(raw, dataType) if err != nil { + bw.End() return fmt.Errorf("failed to extract document ID: %w", err) } docRef := collection.Doc(docID) - batch.Set(docRef, docData) + job, err := bw.Set(docRef, docData) + if err != nil { + bw.End() + return fmt.Errorf("failed to queue set operation: %w", err) + } + jobs = append(jobs, job) } - _, err := batch.Commit(ctx) - if err != nil { - return fmt.Errorf("failed to commit batch: %w", err) + bw.End() + + // Check for errors in jobs + for i, job := range jobs { + if _, err := job.Results(); err != nil { + return fmt.Errorf("failed to write document %d: %w", i, err) + } } slog.Debug("wrote batch to firestore", "collection", collectionName, "count", len(data)) @@ -221,42 +232,64 @@ func (c *conn) writeBatch(ctx context.Context, collectionName string, data [][]b } func (c *conn) writeUpdatesBatch(ctx context.Context, collectionName string, updates []*adiomv1.Update, dataType adiomv1.DataType) error { - batch := c.client.Batch() + bw := c.client.BulkWriter(ctx) collection := c.client.Collection(collectionName) idKey := getIDFieldName(dataType) + var jobs []*firestore.BulkWriterJob for _, update := range updates { docID, err := extractIDFromBsonValues(update.GetId()) if err != nil { + bw.End() return fmt.Errorf("failed to extract document ID from update: %w", err) } docRef := collection.Doc(docID) + var job *firestore.BulkWriterJob switch update.GetType() { case adiomv1.UpdateType_UPDATE_TYPE_DELETE: - batch.Delete(docRef) + job, err = bw.Delete(docRef) + if err != nil { + bw.End() + return fmt.Errorf("failed to queue delete operation: %w", err) + } case adiomv1.UpdateType_UPDATE_TYPE_UPDATE, adiomv1.UpdateType_UPDATE_TYPE_INSERT: docData, err := rawToMap(update.GetData(), dataType) if err != nil { + bw.End() return fmt.Errorf("failed to convert update data: %w", err) } delete(docData, idKey) - batch.Set(docRef, docData) + job, err = bw.Set(docRef, docData) + if err != nil { + bw.End() + return fmt.Errorf("failed to queue set operation: %w", err) + } default: slog.Warn("unknown update type, treating as upsert", "type", update.GetType()) docData, err := rawToMap(update.GetData(), dataType) if err != nil { + bw.End() return fmt.Errorf("failed to convert update data: %w", err) } delete(docData, idKey) - batch.Set(docRef, docData) + job, err = bw.Set(docRef, docData) + if err != nil { + bw.End() + return fmt.Errorf("failed to queue set operation: %w", err) + } } + jobs = append(jobs, job) } - _, err := batch.Commit(ctx) - if err != nil { - return fmt.Errorf("failed to commit updates batch: %w", err) + bw.End() + + // Check for errors in jobs + for i, job := range jobs { + if _, err := job.Results(); err != nil { + return fmt.Errorf("failed to write update %d: %w", i, err) + } } slog.Debug("wrote updates batch to firestore", "collection", collectionName, "count", len(updates)) diff --git a/connectors/firestore/connector_integration_test.go b/connectors/firestore/connector_integration_test.go index 0abb7c3f..e0ec3351 100644 --- a/connectors/firestore/connector_integration_test.go +++ b/connectors/firestore/connector_integration_test.go @@ -66,8 +66,8 @@ func clearCollection(ctx context.Context, client *firestore.Client, collectionNa iter := col.Documents(ctx) defer iter.Stop() - batch := client.Batch() - count := 0 + bw := client.BulkWriter(ctx) + var jobs []*firestore.BulkWriterJob for { doc, err := iter.Next() @@ -75,23 +75,22 @@ func clearCollection(ctx context.Context, client *firestore.Client, collectionNa break } if err != nil { + bw.End() return err } - batch.Delete(doc.Ref) - count++ - - // Firestore batch limit is 500 - if count >= 500 { - if _, err := batch.Commit(ctx); err != nil { - return err - } - batch = client.Batch() - count = 0 + job, err := bw.Delete(doc.Ref) + if err != nil { + bw.End() + return err } + jobs = append(jobs, job) } - if count > 0 { - if _, err := batch.Commit(ctx); err != nil { + bw.End() + + // Check for errors + for _, job := range jobs { + if _, err := job.Results(); err != nil { return err } }