Remove commanders from datastores

.github/workflows/gen.yaml

@@ -12,7 +12,7 @@ jobs:
       uses: actions/checkout@v3
     - name: Generate code
       # NOTE: Keep this container image as same as defined in Makefile
-      uses: docker://ghcr.io/pipe-cd/codegen@sha256:831f2dda2f56b1d12e90f88c0cb4168f51aa4eb5907b468e74bc42670939fff2 #v0.50.0-215-g3f6a738
+      uses: docker://ghcr.io/pipe-cd/codegen@sha256:3aa25a5abafe40419861ce1f1667580d4274e144370d03ce9f1d00e9b391d7fd #v0.52.0-135-gcefd641
       with:
         entrypoint: ./tool/codegen/codegen.sh
         args: /github/workspace

Makefile

@@ -258,7 +258,7 @@ update/copyright:
 .PHONY: gen/code
 gen/code:
 	# NOTE: Keep this container image as same as defined in .github/workflows/gen.yml
-	docker run --rm -v ${PWD}:/repo -it --entrypoint ./tool/codegen/codegen.sh ghcr.io/pipe-cd/codegen@sha256:831f2dda2f56b1d12e90f88c0cb4168f51aa4eb5907b468e74bc42670939fff2 /repo # v0.50.0-215-g3f6a738
+	docker run --rm -v ${PWD}:/repo -it --entrypoint ./tool/codegen/codegen.sh ghcr.io/pipe-cd/codegen@sha256:3aa25a5abafe40419861ce1f1667580d4274e144370d03ce9f1d00e9b391d7fd /repo # v0.52.0-135-gcefd641
 
 .PHONY: gen/test-tls
 gen/test-tls:

cmd/pipecd/ops.go

@@ -204,14 +204,14 @@ func (s *ops) run(ctx context.Context, input cli.Input) error {
 
 	insightMetricsCollector := insightmetrics.NewInsightMetricsCollector(
 		insight.NewProvider(insightStore),
-		datastore.NewProjectStore(ds, datastore.OpsCommander),
+		datastore.NewProjectStore(ds),
 	)
 
 	// Start running HTTP server.
 	{
 		handler := handler.NewHandler(
 			s.httpPort,
-			datastore.NewProjectStore(ds, datastore.OpsCommander),
+			datastore.NewProjectStore(ds),
 			cfg.SharedSSOConfigs,
 			s.gracePeriod,
 			input.Logger,

cmd/pipecd/server.go

@@ -211,9 +211,9 @@ func (s *server) run(ctx context.Context, input cli.Input) error {
 			verifier = pipedverifier.NewVerifier(
 				ctx,
 				cfg,
-				// These stores are used to handle PipedAPI request, thus the writer should be PipedWriter.
-				datastore.NewProjectStore(ds, datastore.PipedCommander),
-				datastore.NewPipedStore(ds, datastore.PipedCommander),
+				// These stores are used to handle PipedAPI request.
+				datastore.NewProjectStore(ds),
+				datastore.NewPipedStore(ds),
 				input.Logger,
 			)
 			service = grpcapi.NewPipedAPI(ctx, ds, cache, sls, alss, las, statCache, cmdOutputStore, unregisteredAppStore, aso, cfg.Address, input.Logger)
@@ -247,7 +247,7 @@ func (s *server) run(ctx context.Context, input cli.Input) error {
 		var (
 			verifier = apikeyverifier.NewVerifier(
 				ctx,
-				datastore.NewAPIKeyStore(ds, datastore.PipectlCommander),
+				datastore.NewAPIKeyStore(ds),
 				apiKeyLastUsedCache,
 				input.Logger,
 			)
@@ -345,7 +345,7 @@ func (s *server) run(ctx context.Context, input cli.Input) error {
 			cfg.StateKey,
 			cfg.ProjectMap(),
 			cfg.SharedSSOConfigMap(),
-			datastore.NewProjectStore(ds, datastore.WebCommander),
+			datastore.NewProjectStore(ds),
 			!s.insecureCookie,
 			input.Logger,
 		)
@@ -385,9 +385,9 @@ func (s *server) run(ctx context.Context, input cli.Input) error {
 			verifier = pipedverifier.NewVerifier(
 				ctx,
 				cfg,
-				// These stores are used to handle request from piped over envoy ext_authz, thus the writer should be PipedWriter.
-				datastore.NewProjectStore(ds, datastore.PipedCommander),
-				datastore.NewPipedStore(ds, datastore.PipedCommander),
+				// These stores are used to handle request from piped over envoy ext_authz.
+				datastore.NewProjectStore(ds),
+				datastore.NewPipedStore(ds),
 				input.Logger,
 			)
 			service = grpcapi.NewEnvoyAuthorizationServer(verifier)

pkg/app/ops/apikeylastusedtimeupdater/apikeylastusedtimeupdater.go

@@ -54,7 +54,7 @@ func NewAPIKeyLastUsedTimeUpdater(
 	logger *zap.Logger,
 ) *APIKeyLastUsedTimeUpdater {
 	return &APIKeyLastUsedTimeUpdater{
-		apiKeyStore:             datastore.NewAPIKeyStore(ds, datastore.OpsCommander),
+		apiKeyStore:             datastore.NewAPIKeyStore(ds),
 		apiKeyLastUsedTimeCache: rediscache.NewHashCache(rd, apiKeyLastUsedCacheHashKey),
 		logger:                  logger.Named("api-key-last-used-time-updater"),
 	}

pkg/app/ops/deploymentchaincontroller/controller.go

@@ -61,10 +61,9 @@ func NewDeploymentChainController(
 	ds datastore.DataStore,
 	logger *zap.Logger,
 ) *DeploymentChainController {
-	w := datastore.OpsCommander
 	return &DeploymentChainController{
-		deploymentStore:      datastore.NewDeploymentStore(ds, w),
-		deploymentChainStore: datastore.NewDeploymentChainStore(ds, w),
+		deploymentStore:      datastore.NewDeploymentStore(ds),
+		deploymentChainStore: datastore.NewDeploymentChainStore(ds),
 		updaters:             make(map[string]*updater),
 		logger:               logger.Named("deployment-chain-controller"),
 	}

pkg/app/ops/insightcollector/collector.go

@@ -46,9 +46,8 @@ func NewCollector(ds datastore.DataStore, store insight.Store, cfg config.Contro
 	logger = logger.Named("insight-collector")
 
 	var (
-		w            = datastore.OpsCommander
-		appLister    = datastore.NewApplicationStore(ds, w)
-		deployLister = datastore.NewDeploymentStore(ds, w)
+		appLister    = datastore.NewApplicationStore(ds)
+		deployLister = datastore.NewDeploymentStore(ds)
 		appCol       = newApplicationDataCollector(appLister, store, logger)
 		comDepCol    = newCompletedDeploymentDataCollector(deployLister, store, logger)
 	)

pkg/app/ops/orphancommandcleaner/orphancommandcleaner.go

@@ -45,7 +45,7 @@ func NewOrphanCommandCleaner(
 	logger *zap.Logger,
 ) *OrphanCommandCleaner {
 	return &OrphanCommandCleaner{
-		commandstore: datastore.NewCommandStore(ds, datastore.OpsCommander),
+		commandstore: datastore.NewCommandStore(ds),
 		logger:       logger.Named("orphan-command-cleaner"),
 	}
 }

pkg/app/server/commandstore/store.go

@@ -38,9 +38,9 @@ type store struct {
 	logger  *zap.Logger
 }
 
-func NewStore(w datastore.Commander, ds datastore.DataStore, c cache.Cache, logger *zap.Logger) Store {
+func NewStore(ds datastore.DataStore, c cache.Cache, logger *zap.Logger) Store {
 	return &store{
-		backend: datastore.NewCommandStore(ds, w),
+		backend: datastore.NewCommandStore(ds),
 		cache: &commandCache{
 			backend: c,
 		},

pkg/app/server/grpcapi/api.go

@@ -107,14 +107,13 @@ func NewAPI(
 	webBaseURL string,
 	logger *zap.Logger,
 ) *API {
-	w := datastore.PipectlCommander
 	a := &API{
-		applicationStore:     datastore.NewApplicationStore(ds, w),
-		deploymentStore:      datastore.NewDeploymentStore(ds, w),
-		pipedStore:           datastore.NewPipedStore(ds, w),
-		eventStore:           datastore.NewEventStore(ds, w),
-		deploymentTraceStore: datastore.NewDeploymentTraceStore(ds, w),
-		commandStore:         commandstore.NewStore(w, ds, sc, logger),
+		applicationStore:     datastore.NewApplicationStore(ds),
+		deploymentStore:      datastore.NewDeploymentStore(ds),
+		pipedStore:           datastore.NewPipedStore(ds),
+		eventStore:           datastore.NewEventStore(ds),
+		deploymentTraceStore: datastore.NewDeploymentTraceStore(ds),
+		commandStore:         commandstore.NewStore(ds, sc, logger),
 		stageLogStore:        stagelogstore.NewStore(fs, sc, logger),
 		commandOutputGetter:  cog,
 		// Public key is variable but likely to be accessed multiple times in a short period.

pkg/app/server/grpcapi/piped_api.go

@@ -113,17 +113,16 @@ type PipedAPI struct {
 
 // NewPipedAPI creates a new PipedAPI instance.
 func NewPipedAPI(ctx context.Context, ds datastore.DataStore, sc cache.Cache, sls stagelogstore.Store, alss applicationlivestatestore.Store, las analysisresultstore.Store, hc cache.Cache, cop commandOutputPutter, uas unregisteredappstore.Store, aso applicationsharedobjectstore.Store, webBaseURL string, logger *zap.Logger) *PipedAPI {
-	w := datastore.PipedCommander
 	a := &PipedAPI{
-		applicationStore:          datastore.NewApplicationStore(ds, w),
-		deploymentStore:           datastore.NewDeploymentStore(ds, w),
-		deploymentChainStore:      datastore.NewDeploymentChainStore(ds, w),
-		pipedStore:                datastore.NewPipedStore(ds, w),
-		eventStore:                datastore.NewEventStore(ds, w),
+		applicationStore:          datastore.NewApplicationStore(ds),
+		deploymentStore:           datastore.NewDeploymentStore(ds),
+		deploymentChainStore:      datastore.NewDeploymentChainStore(ds),
+		pipedStore:                datastore.NewPipedStore(ds),
+		eventStore:                datastore.NewEventStore(ds),
 		stageLogStore:             sls,
 		applicationLiveStateStore: alss,
 		analysisResultStore:       las,
-		commandStore:              commandstore.NewStore(w, ds, sc, logger),
+		commandStore:              commandstore.NewStore(ds, sc, logger),
 		commandOutputPutter:       cop,
 		unregisteredAppStore:      uas,
 		appSharedObjectStore:      aso,

pkg/app/server/grpcapi/web_api.go

@@ -163,20 +163,19 @@ func NewWebAPI(
 	encrypter encrypter,
 	logger *zap.Logger,
 ) *WebAPI {
-	w := datastore.WebCommander
 	a := &WebAPI{
-		applicationStore:          datastore.NewApplicationStore(ds, w),
-		deploymentChainStore:      datastore.NewDeploymentChainStore(ds, w),
-		deploymentStore:           datastore.NewDeploymentStore(ds, w),
-		deploymentTraceStore:      datastore.NewDeploymentTraceStore(ds, w),
-		pipedStore:                datastore.NewPipedStore(ds, w),
-		projectStore:              datastore.NewProjectStore(ds, w),
-		apiKeyStore:               datastore.NewAPIKeyStore(ds, w),
+		applicationStore:          datastore.NewApplicationStore(ds),
+		deploymentChainStore:      datastore.NewDeploymentChainStore(ds),
+		deploymentStore:           datastore.NewDeploymentStore(ds),
+		deploymentTraceStore:      datastore.NewDeploymentTraceStore(ds),
+		pipedStore:                datastore.NewPipedStore(ds),
+		projectStore:              datastore.NewProjectStore(ds),
+		apiKeyStore:               datastore.NewAPIKeyStore(ds),
 		apiKeyLastUsedStore:       akluc,
-		eventStore:                datastore.NewEventStore(ds, w),
+		eventStore:                datastore.NewEventStore(ds),
 		stageLogStore:             sls,
 		applicationLiveStateStore: alss,
-		commandStore:              commandstore.NewStore(w, ds, sc, logger),
+		commandStore:              commandstore.NewStore(ds, sc, logger),
 		insightProvider:           ip,
 		unregisteredAppStore:      uas,
 		projectsInConfig:          projs,

pkg/datastore/apikey.go

@@ -24,7 +24,6 @@ import (
 )
 
 type apiKeyCollection struct {
-	requestedBy Commander
 }
 
 func (a *apiKeyCollection) Kind() string {
@@ -44,12 +43,7 @@ func (a *apiKeyCollection) ListInUsedShards() []Shard {
 }
 
 func (a *apiKeyCollection) GetUpdatableShard() (Shard, error) {
-	switch a.requestedBy {
-	case WebCommander:
-		return ClientShard, nil
-	default:
-		return "", ErrUnsupported
-	}
+	return ClientShard, nil
 }
 
 func (a *apiKeyCollection) Encode(e interface{}) (map[Shard][]byte, error) {
@@ -79,18 +73,16 @@ type APIKeyStore interface {
 
 type apiKeyStore struct {
 	backend
-	commander Commander
-	nowFunc   func() time.Time
+	nowFunc func() time.Time
 }
 
-func NewAPIKeyStore(ds DataStore, c Commander) APIKeyStore {
+func NewAPIKeyStore(ds DataStore) APIKeyStore {
 	return &apiKeyStore{
 		backend: backend{
 			ds:  ds,
-			col: &apiKeyCollection{requestedBy: c},
+			col: &apiKeyCollection{},
 		},
-		commander: c,
-		nowFunc:   time.Now,
+		nowFunc: time.Now,
 	}
 }
 

pkg/datastore/apikey_test.go

@@ -64,7 +64,7 @@ func TestAddAPIKey(t *testing.T) {
 
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
-			s := NewAPIKeyStore(tc.dsFactory(tc.apiKey), TestCommander)
+			s := NewAPIKeyStore(tc.dsFactory(tc.apiKey))
 			err := s.Add(context.Background(), tc.apiKey)
 			assert.Equal(t, tc.wantErr, err != nil)
 		})
@@ -119,7 +119,7 @@ func TestListAPIKeys(t *testing.T) {
 
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
-			s := NewAPIKeyStore(tc.ds, TestCommander)
+			s := NewAPIKeyStore(tc.ds)
 			_, err := s.List(context.Background(), tc.opts)
 			assert.Equal(t, tc.wantErr, err)
 		})

pkg/datastore/applicationstore.go

@@ -24,7 +24,6 @@ import (
 )
 
 type applicationCollection struct {
-	requestedBy Commander
 }
 
 func (a *applicationCollection) Kind() string {
@@ -45,14 +44,7 @@ func (a *applicationCollection) ListInUsedShards() []Shard {
 }
 
 func (a *applicationCollection) GetUpdatableShard() (Shard, error) {
-	switch a.requestedBy {
-	case WebCommander, PipectlCommander:
-		return ClientShard, nil
-	case PipedCommander:
-		return AgentShard, nil
-	default:
-		return "", ErrUnsupported
-	}
+	return ClientShard, nil
 }
 
 func (a *applicationCollection) Decode(e interface{}, parts map[Shard][]byte) error {
@@ -203,18 +195,16 @@ type ApplicationStore interface {
 
 type applicationStore struct {
 	backend
-	commander Commander
-	nowFunc   func() time.Time
+	nowFunc func() time.Time
 }
 
-func NewApplicationStore(ds DataStore, c Commander) ApplicationStore {
+func NewApplicationStore(ds DataStore) ApplicationStore {
 	return &applicationStore{
 		backend: backend{
 			ds:  ds,
-			col: &applicationCollection{requestedBy: c},
+			col: &applicationCollection{},
 		},
-		commander: c,
-		nowFunc:   time.Now,
+		nowFunc: time.Now,
 	}
 }
 

pkg/datastore/applicationstore_test.go

@@ -70,7 +70,7 @@ func TestAddApplication(t *testing.T) {
 
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
-			s := NewApplicationStore(tc.dsFactory(tc.application), TestCommander)
+			s := NewApplicationStore(tc.dsFactory(tc.application))
 			err := s.Add(context.Background(), tc.application)
 			assert.Equal(t, tc.wantErr, err != nil)
 		})
@@ -115,7 +115,7 @@ func TestGetApplication(t *testing.T) {
 
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
-			s := NewApplicationStore(tc.ds, TestCommander)
+			s := NewApplicationStore(tc.ds)
 			_, err := s.Get(context.Background(), tc.id)
 			assert.Equal(t, tc.wantErr, err != nil)
 		})
@@ -170,15 +170,15 @@ func TestListApplications(t *testing.T) {
 
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
-			s := NewApplicationStore(tc.ds, TestCommander)
+			s := NewApplicationStore(tc.ds)
 			_, _, err := s.List(context.Background(), tc.opts)
 			assert.Equal(t, tc.wantErr, err != nil)
 		})
 	}
 }
 
 func TestApplicationDecode(t *testing.T) {
-	col := &applicationCollection{requestedBy: TestCommander}
+	col := &applicationCollection{}
 
 	testcases := []struct {
 		name      string