Beta release

config/config.go

@@ -19,11 +19,11 @@ import "crypto/tls"
 
 // Driver contains connection and transport configuration
 type Driver struct {
-	TLS               *tls.Config `json:"tls,omitempty"`
-	ApplicationId     string      `json:"application_id,omitempty"`
-	ApplicationSecret string      `json:"application_secret,omitempty"`
-	Token             string      `json:"token,omitempty"`
-	URL               string      `json:"url,omitempty"`
+	TLS          *tls.Config `json:"tls,omitempty"`
+	ClientID     string      `json:"client_id,omitempty"`
+	ClientSecret string      `json:"client_secret,omitempty"`
+	Token        string      `json:"token,omitempty"`
+	URL          string      `json:"url,omitempty"`
 }
 
 // Database contains database and connection config

driver/driver.go

@@ -285,7 +285,7 @@ func initConfig(cfg *config.Driver) *config.Driver {
 		cfg = &config.Driver{}
 	}
 
-	if cfg.TLS == nil && (cfg.ApplicationId != "" || cfg.ApplicationSecret != "" || cfg.Token != "") {
+	if cfg.TLS == nil && (cfg.ClientID != "" || cfg.ClientSecret != "" || cfg.Token != "") {
 		cfg.TLS = &tls.Config{MinVersion: tls.VersionTLS12}
 	}
 

driver/driver_test.go

@@ -1009,7 +1009,7 @@ func TestNewDriver(t *testing.T) {
 	_, err = NewDriver(context.Background(), nil)
 	require.Error(t, err)
 
-	cfg1 := &config.Driver{URL: test.GRPCURL(4), ApplicationSecret: "aaaa"}
+	cfg1 := &config.Driver{URL: test.GRPCURL(4), ClientSecret: "aaaa"}
 	cfg1 = initConfig(cfg1)
 	require.NotNil(t, cfg1.TLS)
 }

driver/grpc.go

@@ -464,7 +464,7 @@ func (c *grpcDriver) ListApplications(ctx context.Context) ([]*Application, erro
 	return applications, nil
 }
 
-func (c *grpcDriver) RotateApplicationSecret(ctx context.Context, id string) (*Application, error) {
+func (c *grpcDriver) RotateClientSecret(ctx context.Context, id string) (*Application, error) {
 	r, err := c.mgmt.RotateApplicationSecret(ctx, &api.RotateApplicationSecretRequest{Id: id})
 	if err != nil {
 		return nil, GRPCError(err)
@@ -477,7 +477,7 @@ func (c *grpcDriver) RotateApplicationSecret(ctx context.Context, id string) (*A
 	return (*Application)(r.Application), nil
 }
 
-func (c *grpcDriver) GetAccessToken(ctx context.Context, applicationID string, applicationSecret string, refreshToken string) (*TokenResponse, error) {
+func (c *grpcDriver) GetAccessToken(ctx context.Context, clientId string, clientSecret string, refreshToken string) (*TokenResponse, error) {
 	tp := api.GrantType_CLIENT_CREDENTIALS
 	if refreshToken != "" {
 		tp = api.GrantType_REFRESH_TOKEN
@@ -486,8 +486,8 @@ func (c *grpcDriver) GetAccessToken(ctx context.Context, applicationID string, a
 	r, err := c.auth.GetAccessToken(ctx, &api.GetAccessTokenRequest{
 		GrantType:    tp,
 		RefreshToken: refreshToken,
-		ClientId:     applicationID,
-		ClientSecret: applicationSecret,
+		ClientId:     clientId,
+		ClientSecret: clientSecret,
 	})
 	if err != nil {
 		return nil, GRPCError(err)

driver/http.go

@@ -718,7 +718,7 @@ func (c *httpDriver) ListApplications(ctx context.Context) ([]*Application, erro
 	return apps.Applications, nil
 }
 
-func (c *httpDriver) RotateApplicationSecret(ctx context.Context, id string) (*Application, error) {
+func (c *httpDriver) RotateClientSecret(ctx context.Context, id string) (*Application, error) {
 	resp, err := c.api.ManagementRotateApplicationSecret(ctx, apiHTTP.ManagementRotateApplicationSecretJSONBody{Id: &id})
 	if err := HTTPError(err, resp); err != nil {
 		return nil, err
@@ -735,21 +735,21 @@ func (c *httpDriver) RotateApplicationSecret(ctx context.Context, id string) (*A
 	return &app.Application, nil
 }
 
-func (c *httpDriver) GetAccessToken(ctx context.Context, applicationID string, applicationSecret string, refreshToken string) (*TokenResponse, error) {
-	return getAccessToken(ctx, c.tokenURL, c.cfg, applicationID, applicationSecret, refreshToken)
+func (c *httpDriver) GetAccessToken(ctx context.Context, clientId string, clientSecret string, refreshToken string) (*TokenResponse, error) {
+	return getAccessToken(ctx, c.tokenURL, c.cfg, clientId, clientSecret, refreshToken)
 }
 
-func getAccessToken(ctx context.Context, tokenURL string, cfg *config.Driver, applicationID string, applicationSecret string, refreshToken string) (*TokenResponse, error) {
+func getAccessToken(ctx context.Context, tokenURL string, cfg *config.Driver, clientId string, clientSecret string, refreshToken string) (*TokenResponse, error) {
 	data := url.Values{
-		"client_id":     {applicationID},
-		"client_secret": {applicationSecret},
+		"client_id":     {clientId},
+		"client_secret": {clientSecret},
 		"grant_type":    {grantTypeClientCredentials},
 		"scope":         {scope},
 	}
 	if refreshToken != "" {
 		data = url.Values{
 			"refresh_token": {refreshToken},
-			"client_id":     {applicationID},
+			"client_id":     {clientId},
 			"grant_type":    {grantTypeRefreshToken},
 			"scope":         {scope},
 		}

driver/internal.go

@@ -60,14 +60,14 @@ type txWithOptions interface {
 
 //func configAuth(config *config.Driver) (*clientcredentials.Config, context.Context) {
 func configAuth(config *config.Driver) (oauth2.TokenSource, *http.Client, string) {
-	appID := config.ApplicationId
+	clientId := config.ClientID
 	if os.Getenv(ApplicationID) != "" {
-		appID = os.Getenv(ApplicationID)
+		clientId = os.Getenv(ApplicationID)
 	}
 
-	appSecret := config.ApplicationSecret
+	clientSecret := config.ClientSecret
 	if os.Getenv(ApplicationSecret) != "" {
-		appSecret = os.Getenv(ApplicationSecret)
+		clientSecret = os.Getenv(ApplicationSecret)
 	}
 
 	token := config.Token
@@ -101,8 +101,8 @@ func configAuth(config *config.Driver) (oauth2.TokenSource, *http.Client, string
 		ocfg1 := &oauth2.Config{Endpoint: oauth2.Endpoint{TokenURL: tokenURL}}
 		client = ocfg1.Client(ctxClient, t)
 		ts = ocfg1.TokenSource(ctxClient, t)
-	} else if appID != "" || appSecret != "" {
-		oCfg := &clientcredentials.Config{TokenURL: tokenURL, ClientID: appID, ClientSecret: appSecret, AuthStyle: oauth2.AuthStyleInParams}
+	} else if clientId != "" || clientSecret != "" {
+		oCfg := &clientcredentials.Config{TokenURL: tokenURL, ClientID: clientId, ClientSecret: clientSecret, AuthStyle: oauth2.AuthStyleInParams}
 		client = oCfg.Client(ctxClient)
 		ts = oCfg.TokenSource(ctxClient)
 	}

driver/management.go

@@ -13,8 +13,8 @@ type Management interface {
 	DeleteApplication(ctx context.Context, id string) error
 	UpdateApplication(ctx context.Context, id string, name string, description string) (*Application, error)
 	ListApplications(ctx context.Context) ([]*Application, error)
-	RotateApplicationSecret(ctx context.Context, id string) (*Application, error)
-	GetAccessToken(ctx context.Context, applicationID string, applicationSecret string, refreshToken string) (*TokenResponse, error)
+	RotateClientSecret(ctx context.Context, id string) (*Application, error)
+	GetAccessToken(ctx context.Context, clientId string, clientSecret string, refreshToken string) (*TokenResponse, error)
 
 	Close() error
 }

driver/management_test.go

@@ -52,7 +52,7 @@ func testDriverAuthNegative(t *testing.T, c Management, mc *mock.MockManagementS
 		pm(&api.RotateApplicationSecretRequest{Id: "ras_id1"})).Return(
 		&api.RotateApplicationSecretResponse{Application: nil}, nil)
 
-	_, err = c.RotateApplicationSecret(ctx, "ras_id1")
+	_, err = c.RotateClientSecret(ctx, "ras_id1")
 	require.Error(t, err)
 }
 
@@ -117,7 +117,7 @@ func testDriverAuth(t *testing.T, c Management, mc *mock.MockManagementServer, m
 		pm(&api.RotateApplicationSecretRequest{Id: "ras_id1"})).Return(
 		&api.RotateApplicationSecretResponse{Application: lapp2}, nil)
 
-	app, err = c.RotateApplicationSecret(ctx, "ras_id1")
+	app, err = c.RotateClientSecret(ctx, "ras_id1")
 	require.NoError(t, err)
 
 	appEqual(t, lapp2, app)
@@ -208,9 +208,9 @@ func TestGRPCDriverCredentials(t *testing.T) {
 	t.Run("config", func(t *testing.T) {
 		TokenURLOverride = testTokenURLOverride
 		client, _, mockServers, cancel := SetupMgmtGRPCTests(t, &config.Driver{
-			URL:               test.GRPCURL(0),
-			ApplicationId:     "client_id_test",
-			ApplicationSecret: "client_secret_test",
+			URL:          test.GRPCURL(0),
+			ClientID:     "client_id_test",
+			ClientSecret: "client_secret_test",
 		})
 		defer cancel()
 
@@ -249,9 +249,9 @@ func TestHTTPDriverCredentials(t *testing.T) {
 	t.Run("config", func(t *testing.T) {
 		TokenURLOverride = ""
 		client, _, mockServers, cancel := SetupMgmtHTTPTests(t, &config.Driver{
-			URL:               test.HTTPURL(2),
-			ApplicationId:     "client_id_test",
-			ApplicationSecret: "client_secret_test",
+			URL:          test.HTTPURL(2),
+			ClientID:     "client_id_test",
+			ClientSecret: "client_secret_test",
 		})
 		defer cancel()
 

schema/schema.go

@@ -52,6 +52,12 @@ const (
 	id = "ID"
 )
 
+// Collection types
+const (
+	Documents = "documents" // Regular collection containing documents
+	Messages  = "messages"  // Publish/subscribe collection containing messages
+)
+
 // Model represents types supported as collection models
 type Model interface{}
 
@@ -106,6 +112,8 @@ type Schema struct {
 	Desc       string            `json:"description,omitempty"`
 	Fields     map[string]*Field `json:"properties,omitempty"`
 	PrimaryKey []string          `json:"primary_key,omitempty"`
+
+	CollectionType string `json:"collection_type,omitempty"`
 }
 
 // DatabaseModelName returns name of the database derived from the given database model
@@ -356,7 +364,7 @@ func traverseFields(prefix string, t reflect.Type, fields map[string]*Field, pk
 	return nil
 }
 
-func fromCollectionModel(model interface{}) (*Schema, error) {
+func fromCollectionModel(model interface{}, typ string) (*Schema, error) {
 	t := reflect.TypeOf(model)
 	if t.Kind() == reflect.Ptr {
 		t = t.Elem()
@@ -392,7 +400,7 @@ func fromCollectionModel(model interface{}) (*Schema, error) {
 	}
 
 	// No explicit primary key defined and no tigris.Model embedded
-	if len(pk) == 0 {
+	if len(pk) == 0 && typ == Documents {
 		var p *Field
 		var ok bool
 		name := id
@@ -411,22 +419,28 @@ func fromCollectionModel(model interface{}) (*Schema, error) {
 		sch.PrimaryKey = append(sch.PrimaryKey, name)
 	}
 
+	if len(pk) != 0 && typ == Messages {
+		return nil, fmt.Errorf("primary key should not be defined for `messages` collection type")
+	}
+
+	sch.CollectionType = typ
+
 	return &sch, nil
 }
 
 // FromCollectionModels converts provided model(s) to the schema structure
-func FromCollectionModels(model Model, models ...Model) (map[string]*Schema, error) {
+func FromCollectionModels(tp string, model Model, models ...Model) (map[string]*Schema, error) {
 	//models parameter added to require at least one schema to migrate
 	schemas := make(map[string]*Schema)
 
-	schema, err := fromCollectionModel(model)
+	schema, err := fromCollectionModel(model, tp)
 	if err != nil {
 		return nil, err
 	}
 	schemas[schema.Name] = schema
 
 	for _, m := range models {
-		schema, err := fromCollectionModel(m)
+		schema, err := fromCollectionModel(m, tp)
 		if err != nil {
 			return nil, err
 		}
@@ -472,7 +486,7 @@ func FromDatabaseModel(dbModel interface{}) (string, map[string]*Schema, error)
 			tt = field.Type.Elem()
 		}
 
-		sch, err := fromCollectionModel(reflect.New(tt).Elem().Interface())
+		sch, err := fromCollectionModel(reflect.New(tt).Elem().Interface(), Documents)
 		if err != nil {
 			return "", nil, err
 		}

schema/schema_test.go

@@ -345,32 +345,41 @@ func TestCollectionSchema(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(reflect.TypeOf(c.input).Name(), func(t *testing.T) {
-			schema, err := fromCollectionModel(c.input)
+			schema, err := fromCollectionModel(c.input, Documents)
 			assert.Equal(t, c.err, err)
+			if schema != nil {
+				assert.Equal(t, Documents, schema.CollectionType)
+				c.output.CollectionType = Documents
+			}
 			assert.Equal(t, c.output, schema)
 		})
 	}
 
 	t.Run("build", func(t *testing.T) {
-		s, err := fromCollectionModel(allTypes{})
+		s, err := fromCollectionModel(allTypes{}, Documents)
 		require.NoError(t, err)
 
+		assert.Equal(t, Documents, s.CollectionType)
+		s.CollectionType = ""
+
 		b, err := s.Build()
 		require.NoError(t, err)
 
 		require.Equal(t, `{"title":"all_types","properties":{"PtrStruct":{"type":"object","properties":{"ss_field_1":{"type":"string"}}},"Tm":{"type":"string","format":"date-time"},"TmPtr":{"type":"string","format":"date-time"},"UUID":{"type":"string","format":"uuid"},"UUIDPtr":{"type":"string","format":"uuid"},"arr_1":{"type":"array","items":{"type":"string"}},"bool_1":{"type":"boolean"},"bool_123":{"type":"boolean"},"bytes_1":{"type":"string","format":"byte"},"bytes_2":{"type":"string","format":"byte"},"data_1":{"type":"object","properties":{"Nested":{"type":"object","properties":{"ss_field_1":{"type":"string"}}},"field_1":{"type":"string"}}},"float_32":{"type":"number"},"float_64":{"type":"number"},"int_1":{"type":"integer"},"int_32":{"type":"integer","format":"int32"},"int_64":{"type":"integer"},"map_1":{"type":"object"},"map_2":{"type":"object"},"slice_1":{"type":"array","items":{"type":"string"}},"slice_2":{"type":"array","items":{"type":"object","properties":{"Nested":{"type":"object","properties":{"ss_field_1":{"type":"string"}}},"field_1":{"type":"string"}}}},"string_1":{"type":"string"}},"primary_key":["string_1"]}`, string(b))
 	})
 
 	t.Run("multiple_models", func(t *testing.T) {
-		s, err := FromCollectionModels(pk{}, pk1{})
+		s, err := FromCollectionModels(Documents, pk{}, pk1{})
 		require.NoError(t, err)
 
-		assert.Equal(t, map[string]*Schema{"pks": {Name: "pks", Fields: map[string]*Field{"key_1": {Type: typeString}}, PrimaryKey: []string{"key_1"}},
-			"pk_1": {Name: "pk_1", Fields: map[string]*Field{"key_1": {Type: typeString}}, PrimaryKey: []string{"key_1"}}}, s)
+		assert.Equal(t, map[string]*Schema{
+			"pks":  {Name: "pks", Fields: map[string]*Field{"key_1": {Type: typeString}}, PrimaryKey: []string{"key_1"}, CollectionType: Documents},
+			"pk_1": {Name: "pk_1", Fields: map[string]*Field{"key_1": {Type: typeString}}, PrimaryKey: []string{"key_1"}, CollectionType: Documents},
+		}, s)
 	})
 
 	t.Run("duplicate_pk_index", func(t *testing.T) {
-		_, err := FromCollectionModels(pkDup{})
+		_, err := FromCollectionModels(Documents, pkDup{})
 		if err.Error() == "duplicate primary key index 1 set for key_1 and key_2" {
 			require.Equal(t, err, fmt.Errorf("duplicate primary key index 1 set for key_1 and key_2"))
 		} else {
@@ -408,11 +417,11 @@ func TestDatabaseSchema(t *testing.T) {
 
 	_ = Db4{1}
 
-	coll1 := Schema{Name: "Coll1", Fields: map[string]*Field{"Key1": {Type: "integer"}}, PrimaryKey: []string{"Key1"}}
-	c1 := Schema{Name: "c1", Fields: map[string]*Field{"Key1": {Type: "integer"}}, PrimaryKey: []string{"Key1"}}
-	c2 := Schema{Name: "c2", Fields: map[string]*Field{"Key2": {Type: "integer"}}, PrimaryKey: []string{"Key2"}}
-	c3 := Schema{Name: "c3", Fields: map[string]*Field{"Key2": {Type: "integer"}}, PrimaryKey: []string{"Key2"}}
-	c4 := Schema{Name: "coll_4", Fields: map[string]*Field{"Key2": {Type: "integer"}}, PrimaryKey: []string{"Key2"}}
+	coll1 := Schema{Name: "Coll1", Fields: map[string]*Field{"Key1": {Type: "integer"}}, PrimaryKey: []string{"Key1"}, CollectionType: Documents}
+	c1 := Schema{Name: "c1", Fields: map[string]*Field{"Key1": {Type: "integer"}}, PrimaryKey: []string{"Key1"}, CollectionType: Documents}
+	c2 := Schema{Name: "c2", Fields: map[string]*Field{"Key2": {Type: "integer"}}, PrimaryKey: []string{"Key2"}, CollectionType: Documents}
+	c3 := Schema{Name: "c3", Fields: map[string]*Field{"Key2": {Type: "integer"}}, PrimaryKey: []string{"Key2"}, CollectionType: Documents}
+	c4 := Schema{Name: "coll_4", Fields: map[string]*Field{"Key2": {Type: "integer"}}, PrimaryKey: []string{"Key2"}, CollectionType: Documents}
 
 	var i int64
 
@@ -436,3 +445,35 @@ func TestDatabaseSchema(t *testing.T) {
 		})
 	}
 }
+
+func TestMessagesSchema(t *testing.T) {
+	type valid struct {
+		Key1 string `json:"key_1"`
+	}
+
+	type invalid struct {
+		Key1 string `json:"key_1" tigris:"primary_key:1"`
+	}
+
+	cases := []struct {
+		input  interface{}
+		output *Schema
+		err    error
+	}{
+		{valid{}, &Schema{Name: "valids", Fields: map[string]*Field{
+			"key_1": {Type: typeString}}, PrimaryKey: []string{}}, nil},
+		{invalid{}, nil, fmt.Errorf("primary key should not be defined for `messages` collection type")},
+	}
+
+	for _, c := range cases {
+		t.Run(reflect.TypeOf(c.input).Name(), func(t *testing.T) {
+			schema, err := fromCollectionModel(c.input, Messages)
+			assert.Equal(t, c.err, err)
+			if schema != nil {
+				assert.Equal(t, Messages, schema.CollectionType)
+				c.output.CollectionType = Messages
+			}
+			assert.Equal(t, c.output, schema)
+		})
+	}
+}

tigris/client_test.go

@@ -45,7 +45,7 @@ func TestClient(t *testing.T) {
 	mc.EXPECT().CreateOrUpdateCollection(gomock.Any(),
 		pm(&api.CreateOrUpdateCollectionRequest{
 			Db: "db1", Collection: "coll_1",
-			Schema:  []byte(`{"title":"coll_1","properties":{"Key1":{"type":"string"}},"primary_key":["Key1"]}`),
+			Schema:  []byte(`{"title":"coll_1","properties":{"Key1":{"type":"string"}},"primary_key":["Key1"],"collection_type":"documents"}`),
 			Options: &api.CollectionOptions{},
 		})).Do(func(ctx context.Context, r *api.CreateOrUpdateCollectionRequest) {
 	}).Return(&api.CreateOrUpdateCollectionResponse{}, nil)