chore: more refactor

Author: miparnisari

internal/datastore/mysql/watch.go

@@ -19,9 +19,8 @@ const (
 
 func (mds *mysqlDatastore) DefaultsWatchOptions() datastore.WatchOptions {
 	return datastore.WatchOptions{
-		WatchBufferLength:              defaultWatchBufferLength,
-		WatchBufferWriteTimeout:        defaultWatchBufferWriteTimeout,
-		MaximumBufferedChangesByteSize: 0, // 0 means no limit
+		WatchBufferLength:       defaultWatchBufferLength,
+		WatchBufferWriteTimeout: defaultWatchBufferWriteTimeout,
 		// MySQL does not use CheckpointInterval or WatchConnectTimeout
 		// MySQL does not support EmitImmediatelyStrategy or WatchSchema
 	}

internal/datastore/postgres/watch.go

@@ -26,10 +26,9 @@ const (
 
 func (pgd *pgDatastore) DefaultsWatchOptions() datastore.WatchOptions {
 	return datastore.WatchOptions{
-		CheckpointInterval:             minimumWatchSleep,
-		WatchBufferLength:              defaultWatchBufferLength,
-		WatchBufferWriteTimeout:        defaultWatchBufferWriteTimeout,
-		MaximumBufferedChangesByteSize: 0, // 0 means no limit
+		CheckpointInterval:      minimumWatchSleep,
+		WatchBufferLength:       defaultWatchBufferLength,
+		WatchBufferWriteTimeout: defaultWatchBufferWriteTimeout,
 		// Postgres does not use WatchConnectTimeout
 		// Postgres does not support EmitImmediatelyStrategy
 	}

internal/datastore/proxy/indexcheck/indexcheck_test.go

@@ -121,7 +121,7 @@ func TestIndexCheckingProxyMethods(t *testing.T) {
 
 	t.Run("Watch", func(t *testing.T) {
 		watchOptions, err := datastore.BuildAndValidateWatchOptions(
-			datastore.ServerWatchConfig{},
+			datastore.ServerWatchOptions{},
 			datastore.ClientWatchOptions{},
 			proxy.DefaultsWatchOptions(),
 		)

internal/datastore/proxy/schemacaching/watchingcache.go

@@ -158,7 +158,7 @@ func (p *watchingCachingProxy) startSync(ctx context.Context) error {
 	}
 
 	watchOptions, err := datastore.BuildAndValidateWatchOptions(
-		datastore.ServerWatchConfig{
+		datastore.ServerWatchOptions{
 			CheckpointInterval: p.watchHeartbeat,
 		},
 		datastore.ClientWatchOptions{

internal/datastore/spanner/watch.go

@@ -55,10 +55,9 @@ func parseDatabaseName(db string) (project, instance, database string, err error
 
 func (sd *spannerDatastore) DefaultsWatchOptions() datastore.WatchOptions {
 	return datastore.WatchOptions{
-		CheckpointInterval:             100 * time.Millisecond,
-		WatchBufferLength:              defaultWatchBufferLength,
-		WatchBufferWriteTimeout:        defaultWatchBufferWriteTimeout,
-		MaximumBufferedChangesByteSize: 0, // 0 means no limit
+		CheckpointInterval:      100 * time.Millisecond,
+		WatchBufferLength:       defaultWatchBufferLength,
+		WatchBufferWriteTimeout: defaultWatchBufferWriteTimeout,
 		// Spanner does not use WatchConnectTimeout
 		// Spanner does not support EmitImmediatelyStrategy
 	}

internal/services/v1/watch.go

@@ -101,7 +101,7 @@ func (ws *watchServer) Watch(req *v1.WatchRequest, stream v1.WatchService_WatchS
 	clientRequest := datastore.ClientWatchOptions{
 		Content: convertWatchKindToContent(req.OptionalUpdateKinds),
 	}
-	dsConfig := datastore.ServerWatchConfig{
+	dsConfig := datastore.ServerWatchOptions{
 		CheckpointInterval:             ws.serverConfig.CheckpointInterval,
 		WatchBufferLength:              ws.serverConfig.WatchBufferLength,
 		WatchBufferWriteTimeout:        ws.serverConfig.WatchBufferWriteTimeout,

pkg/datastore/datastore.go

@@ -602,9 +602,9 @@ const (
 	WatchCheckpoints   WatchContent = 1 << 2
 )
 
-// ServerWatchConfig contains server-level configuration for Watch operations.
+// ServerWatchOptions contains server-level configuration for Watch operations.
 // These values do NOT change during the lifetime of a server.
-type ServerWatchConfig struct {
+type ServerWatchOptions struct {
 	// CheckpointInterval is the interval to use for checkpointing in the watch.
 	CheckpointInterval time.Duration
 
@@ -635,22 +635,21 @@ type ClientWatchOptions struct {
 }
 
 // WatchOptions are ALL options for a Watch call.
-// This struct combines server configuration, client requests, and datastore defaults
-// into a single set of options passed to the datastore Watch implementation.
+// Some datastore implementations may ignore one or more of these.
 type WatchOptions struct {
 	// See ClientWatchOptions.Content
 	Content WatchContent
 	// See ClientWatchOptions.EmissionStrategy
 	EmissionStrategy EmissionStrategy
-	// See ServerWatchConfig.CheckpointInterval
+	// See ServerWatchOptions.CheckpointInterval
 	CheckpointInterval time.Duration
-	// See ServerWatchConfig.WatchBufferLength
+	// See ServerWatchOptions.WatchBufferLength
 	WatchBufferLength uint16
-	// See ServerWatchConfig.WatchBufferWriteTimeout
+	// See ServerWatchOptions.WatchBufferWriteTimeout
 	WatchBufferWriteTimeout time.Duration
-	// See ServerWatchConfig.WatchConnectTimeout
+	// See ServerWatchOptions.WatchConnectTimeout
 	WatchConnectTimeout time.Duration
-	// See ServerWatchConfig.MaximumBufferedChangesByteSize
+	// See ServerWatchOptions.MaximumBufferedChangesByteSize
 	MaximumBufferedChangesByteSize uint64
 }
 
@@ -672,7 +671,7 @@ const (
 // WatchJustRelationships returns watch options for just relationships.
 func WatchJustRelationships(ds Datastore) WatchOptions {
 	v, _ := BuildAndValidateWatchOptions(
-		ServerWatchConfig{},
+		ServerWatchOptions{},
 		ClientWatchOptions{Content: WatchRelationships},
 		ds.DefaultsWatchOptions(),
 	)
@@ -682,7 +681,7 @@ func WatchJustRelationships(ds Datastore) WatchOptions {
 // WatchJustSchema returns watch options for just schema.
 func WatchJustSchema(ds ReadOnlyDatastore) WatchOptions {
 	v, _ := BuildAndValidateWatchOptions(
-		ServerWatchConfig{},
+		ServerWatchOptions{},
 		ClientWatchOptions{Content: WatchSchema},
 		ds.DefaultsWatchOptions(),
 	)
@@ -731,25 +730,27 @@ func watchBufferSize(sizeString string) (size uint64, err error) {
 	return size, nil
 }
 
-// BuildAndValidateWatchOptions constructs complete WatchOptions by merging server configuration,
-// client requests, and datastore defaults.
+// BuildAndValidateWatchOptions constructs complete WatchOptions by merging server options,
+// client options, and datastore defaults.
+// Datastore defaults take precedence over server options.
+// Client options cannot be overridden.
 func BuildAndValidateWatchOptions(
-	serverConfig ServerWatchConfig,
-	clientRequest ClientWatchOptions,
+	serverOptions ServerWatchOptions,
+	clientOptions ClientWatchOptions,
 	datastoreDefaults WatchOptions,
 ) (WatchOptions, error) {
-	watchChangeBufferMaximumSize, err := watchBufferSize(serverConfig.MaximumBufferedChangesByteSize)
+	watchChangeBufferMaximumSize, err := watchBufferSize(serverOptions.MaximumBufferedChangesByteSize)
 	if err != nil {
 		return WatchOptions{}, err
 	}
 
 	options := WatchOptions{
-		Content:                        clientRequest.Content,
-		EmissionStrategy:               clientRequest.EmissionStrategy,
-		CheckpointInterval:             serverConfig.CheckpointInterval,
-		WatchBufferLength:              serverConfig.WatchBufferLength,
-		WatchBufferWriteTimeout:        serverConfig.WatchBufferWriteTimeout,
-		WatchConnectTimeout:            serverConfig.WatchConnectTimeout,
+		Content:                        clientOptions.Content,
+		EmissionStrategy:               clientOptions.EmissionStrategy,
+		CheckpointInterval:             serverOptions.CheckpointInterval,
+		WatchBufferLength:              serverOptions.WatchBufferLength,
+		WatchBufferWriteTimeout:        serverOptions.WatchBufferWriteTimeout,
+		WatchConnectTimeout:            serverOptions.WatchConnectTimeout,
 		MaximumBufferedChangesByteSize: watchChangeBufferMaximumSize,
 	}
 
@@ -825,7 +826,7 @@ type ReadOnlyDatastore interface {
 	Watch(ctx context.Context, afterRevision Revision, options WatchOptions) (<-chan RevisionChanges, <-chan error)
 
 	// DefaultsWatchOptions returns the default watch options for this datastore.
-	// These defaults are used when building WatchOptions from ServerWatchConfig and ClientWatchOptions.
+	// These defaults are used when building WatchOptions from ServerWatchOptions and ClientWatchOptions.
 	// Each datastore should return appropriate defaults based on its capabilities and constraints.
 	DefaultsWatchOptions() WatchOptions
 

pkg/datastore/datastore_test.go

@@ -15,31 +15,31 @@ import (
 
 func TestBuildAndValidateWatchOptions(t *testing.T) {
 	t.Run("WatchBufferWriteTimeout", func(t *testing.T) {
-		c, err := BuildAndValidateWatchOptions(ServerWatchConfig{WatchBufferWriteTimeout: 1 * time.Second}, ClientWatchOptions{}, WatchOptions{WatchBufferWriteTimeout: 2 * time.Second})
+		c, err := BuildAndValidateWatchOptions(ServerWatchOptions{WatchBufferWriteTimeout: 1 * time.Second}, ClientWatchOptions{}, WatchOptions{WatchBufferWriteTimeout: 2 * time.Second})
 		require.NoError(t, err)
 		require.NotNil(t, c)
 		require.Equal(t, 2*time.Second, c.WatchBufferWriteTimeout)
 	})
 	t.Run("WatchConnectTimeout", func(t *testing.T) {
-		c, err := BuildAndValidateWatchOptions(ServerWatchConfig{WatchConnectTimeout: 1 * time.Second}, ClientWatchOptions{}, WatchOptions{WatchConnectTimeout: 2 * time.Second})
+		c, err := BuildAndValidateWatchOptions(ServerWatchOptions{WatchConnectTimeout: 1 * time.Second}, ClientWatchOptions{}, WatchOptions{WatchConnectTimeout: 2 * time.Second})
 		require.NoError(t, err)
 		require.NotNil(t, c)
 		require.Equal(t, 2*time.Second, c.WatchConnectTimeout)
 	})
 	t.Run("WatchBufferLength", func(t *testing.T) {
-		c, err := BuildAndValidateWatchOptions(ServerWatchConfig{WatchBufferLength: 100}, ClientWatchOptions{}, WatchOptions{WatchBufferLength: 200})
+		c, err := BuildAndValidateWatchOptions(ServerWatchOptions{WatchBufferLength: 100}, ClientWatchOptions{}, WatchOptions{WatchBufferLength: 200})
 		require.NoError(t, err)
 		require.NotNil(t, c)
 		require.Equal(t, uint16(200), c.WatchBufferLength)
 	})
 	t.Run("CheckpointInterval", func(t *testing.T) {
-		c, err := BuildAndValidateWatchOptions(ServerWatchConfig{CheckpointInterval: 1}, ClientWatchOptions{}, WatchOptions{CheckpointInterval: 2})
+		c, err := BuildAndValidateWatchOptions(ServerWatchOptions{CheckpointInterval: 1}, ClientWatchOptions{}, WatchOptions{CheckpointInterval: 2})
 		require.NoError(t, err)
 		require.NotNil(t, c)
 		require.Equal(t, time.Duration(2), c.CheckpointInterval)
 	})
 	t.Run("Invalid CheckpointInterval", func(t *testing.T) {
-		_, err := BuildAndValidateWatchOptions(ServerWatchConfig{CheckpointInterval: -1}, ClientWatchOptions{}, WatchOptions{CheckpointInterval: -1})
+		_, err := BuildAndValidateWatchOptions(ServerWatchOptions{CheckpointInterval: -1}, ClientWatchOptions{}, WatchOptions{CheckpointInterval: -1})
 		require.Error(t, err)
 	})
 }