feat: retention for bucketeer demo site

api-description/web-api.swagger.yaml

@@ -10992,6 +10992,8 @@ definitions:
         $ref: '#/definitions/environmentCreateTrialProjectCommand'
   environmentCreateTrialProjectResponse:
     type: object
+  environmentDeleteBucketeerDataResponse:
+    type: object
   environmentDemoCreationToken:
     type: object
     properties:

manifests/bucketeer/charts/batch/templates/deployment.yaml

@@ -59,6 +59,8 @@ spec:
               value: "{{.Values.env.bigqueryWriterEmulatorHost}}"
             - name: BUCKETEER_BATCH_PROJECT
               value: "{{ .Values.global.pubsub.project }}"
+            - name: DEMO_TRIAL_PERIOD_DAY
+              value: "{{ .Values.env.demo.demoTrialPeriod }}"
             - name: BUCKETEER_BATCH_PROFILE
               value: "{{.Values.env.profile}}"
             - name: BUCKETEER_BATCH_GCP_TRACE_ENABLED

manifests/bucketeer/charts/batch/values.dev.yaml

@@ -5,6 +5,8 @@ global:
 
 env:
   project: bucketeer-test
+  demo:
+    demoTrialPeriod: 1 # days
   profile: false
   gcpEnabled: false
   mysqlUser: bucketeer

manifests/bucketeer/charts/batch/values.yaml

@@ -9,6 +9,8 @@ namespace: default
 env:
   project:
   profile: true
+  demo:
+    demoTrialPeriod: 1 # days
   gcpEnabled: true
   mysqlUser:
   mysqlPass:

pkg/batch/api/api.go

@@ -50,6 +50,7 @@ type batchService struct {
 	experimentCacher          jobs.Job
 	autoOpsRulesCacher        jobs.Job
 	tagDeleter                jobs.Job
+	demoOrganizationDeleter   jobs.Job
 	logger                    *zap.Logger
 }
 
@@ -60,7 +61,7 @@ func NewBatchService(
 	redisCounterDeleter, experimentCalculator,
 	mauSummarizer, mauPartitionDeleter, mauPartitionCreator,
 	featureFlagCacher, segmentUserCacher, apiKeyCacher,
-	experimentCacher, autoOpsRulesCacher, tagDeleter jobs.Job,
+	experimentCacher, autoOpsRulesCacher, tagDeleter, demoOrganizationDeleter jobs.Job,
 	logger *zap.Logger,
 ) *batchService {
 	return &batchService{
@@ -82,6 +83,7 @@ func NewBatchService(
 		experimentCacher:          experimentCacher,
 		autoOpsRulesCacher:        autoOpsRulesCacher,
 		tagDeleter:                tagDeleter,
+		demoOrganizationDeleter:   demoOrganizationDeleter,
 		logger:                    logger.Named("batch-service"),
 	}
 }
@@ -126,6 +128,8 @@ func (s *batchService) ExecuteBatchJob(
 		err = s.autoOpsRulesCacher.Run(ctx)
 	case batch.BatchJob_TagDeleter:
 		err = s.tagDeleter.Run(ctx)
+	case batch.BatchJob_DemoOrganizationDeleter:
+		err = s.demoOrganizationDeleter.Run(ctx)
 	default:
 		s.logger.Error("Unknown job",
 			log.FieldsFromIncomingContext(ctx).AddFields(

pkg/batch/api/api_test.go

@@ -864,6 +864,7 @@ func newBatchService(t *testing.T,
 			redisMockClient,
 		),
 		deleter.NewTagDeleter(mysqlMockClient),
+		deleter.NewDemoOrganizationDeleter(mysqlMockClient, environmentMockClient),
 		logger,
 	)
 	return service

pkg/batch/cmd/server/server.go

@@ -573,6 +573,12 @@ func (s *server) Run(ctx context.Context, metrics metrics.Metrics, logger *zap.L
 			jobs.WithTimeout(5*time.Minute),
 			jobs.WithLogger(logger),
 		),
+		deleter.NewDemoOrganizationDeleter(
+			mysqlClient,
+			environmentClient,
+			jobs.WithTimeout(10*time.Minute),
+			jobs.WithLogger(logger),
+		),
 		logger,
 	)
 

pkg/batch/jobs/deleter/demo_organization_deleter.go

@@ -0,0 +1,122 @@
+// Copyright 2025 The Bucketeer Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package deleter
+
+import (
+	"context"
+	"os"
+	"strconv"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/bucketeer-io/bucketeer/pkg/batch/jobs"
+	environmentclient "github.com/bucketeer-io/bucketeer/pkg/environment/client"
+	v2es "github.com/bucketeer-io/bucketeer/pkg/environment/storage/v2"
+	"github.com/bucketeer-io/bucketeer/pkg/storage/v2/mysql"
+	envproto "github.com/bucketeer-io/bucketeer/proto/environment"
+)
+
+type demoOrganizationDeleter struct {
+	organizationStorage v2es.OrganizationStorage
+	environmentClient   environmentclient.Client
+	opts                *jobs.Options
+	logger              *zap.Logger
+}
+
+func NewDemoOrganizationDeleter(
+	mysqlClient mysql.Client,
+	environmentClient environmentclient.Client,
+	opts ...jobs.Option,
+) jobs.Job {
+	dopts := &jobs.Options{
+		Timeout: 1 * time.Minute,
+		Logger:  zap.NewNop(),
+	}
+	for _, opt := range opts {
+		opt(dopts)
+	}
+
+	return &demoOrganizationDeleter{
+		organizationStorage: v2es.NewOrganizationStorage(mysqlClient),
+		environmentClient:   environmentClient,
+		opts:                dopts,
+		logger:              dopts.Logger.Named("demo-organization-deleter"),
+	}
+}
+
+func (d *demoOrganizationDeleter) Run(ctx context.Context) error {
+	ctx, cancel := context.WithTimeout(ctx, d.opts.Timeout)
+	defer cancel()
+
+	outdatedOrganizations, err := d.getOutdatedOrganizations(ctx)
+	if err != nil {
+		return err
+	}
+	if len(outdatedOrganizations) == 0 {
+		return nil
+	}
+
+	outdatedOrganizationIDs := make([]string, 0, len(outdatedOrganizations))
+	for _, org := range outdatedOrganizations {
+		outdatedOrganizationIDs = append(outdatedOrganizationIDs, org.Id)
+	}
+
+	_, err = d.environmentClient.DeleteBucketeerData(ctx, &envproto.DeleteBucketeerDataRequest{
+		DeleteOrganizationIds: outdatedOrganizationIDs,
+	})
+	if err != nil {
+		d.logger.Error("Could not delete bucketeer data", zap.Error(err))
+		return err
+	}
+	return nil
+}
+
+func (d *demoOrganizationDeleter) getOutdatedOrganizations(ctx context.Context) ([]*envproto.Organization, error) {
+	trialPeriod, err := strconv.Atoi(os.Getenv("DEMO_TRIAL_PERIOD_DAY"))
+	if err != nil {
+		d.logger.Error("Could not parse DEMO_TRIAL_PERIOD_DAY", zap.Error(err))
+		return nil, err
+	}
+	filters := []*mysql.FilterV2{
+		{
+			Column:   "organization.created_at",
+			Operator: mysql.OperatorLessThan,
+			Value:    time.Now().AddDate(0, 0, -trialPeriod).Unix(),
+		},
+		{
+			Column:   "organization.system_admin",
+			Operator: mysql.OperatorEqual,
+			Value:    false,
+		},
+	}
+	options := &mysql.ListOptions{
+		Limit:       10000,
+		Offset:      0,
+		Filters:     filters,
+		InFilters:   nil,
+		NullFilters: nil,
+		JSONFilters: nil,
+		SearchQuery: nil,
+		Orders:      nil,
+	}
+
+	organizations, _, _, err := d.organizationStorage.ListOrganizations(ctx, options)
+	if err != nil {
+		d.logger.Error("Could not list organizations", zap.Error(err))
+		return nil, err
+	}
+	return organizations, nil
+}

pkg/batch/jobs/deleter/demo_organization_deleter_test.go

@@ -0,0 +1,100 @@
+package deleter
+
+import (
+	"context"
+	"errors"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/mock/gomock"
+
+	"github.com/bucketeer-io/bucketeer/pkg/batch/jobs"
+	environmentclient "github.com/bucketeer-io/bucketeer/pkg/environment/client/mock"
+	storagemock "github.com/bucketeer-io/bucketeer/pkg/environment/storage/v2/mock"
+	"github.com/bucketeer-io/bucketeer/pkg/log"
+	envproto "github.com/bucketeer-io/bucketeer/proto/environment"
+)
+
+func TestDemoOrganizationDeleter_Run(t *testing.T) {
+	t.Parallel()
+	mockController := gomock.NewController(t)
+	defer mockController.Finish()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	deleter := newMockDemoOrganizationDeleter(t, mockController)
+	err := os.Setenv("DEMO_TRIAL_PERIOD_DAY", "7")
+	assert.Nil(t, err)
+
+	patterns := []struct {
+		desc     string
+		setup    func(deleter *demoOrganizationDeleter)
+		expected error
+	}{
+		{
+			desc: "Error internal",
+			setup: func(deleter *demoOrganizationDeleter) {
+				deleter.organizationStorage.(*storagemock.MockOrganizationStorage).EXPECT().ListOrganizations(
+					gomock.Any(), gomock.Any(),
+				).Return(nil, 0, int64(0), errors.New("internal error"))
+			},
+			expected: errors.New("internal error"),
+		},
+		{
+			desc: "Success no outdated organizations",
+			setup: func(deleter *demoOrganizationDeleter) {
+				deleter.organizationStorage.(*storagemock.MockOrganizationStorage).EXPECT().ListOrganizations(
+					gomock.Any(), gomock.Any(),
+				).Return([]*envproto.Organization{}, 0, int64(0), nil)
+			},
+		},
+		{
+			desc: "Success delete outdated organizations",
+			setup: func(deleter *demoOrganizationDeleter) {
+				deleter.organizationStorage.(*storagemock.MockOrganizationStorage).EXPECT().ListOrganizations(
+					gomock.Any(), gomock.Any(),
+				).Return([]*envproto.Organization{
+					{
+						Id:          "org-id-1",
+						CreatedAt:   17000000000,
+						SystemAdmin: false,
+					},
+					{
+						Id:          "org-id-2",
+						CreatedAt:   17000000000,
+						SystemAdmin: false,
+					},
+				}, 2, int64(2), nil)
+				deleter.environmentClient.(*environmentclient.MockClient).EXPECT().DeleteBucketeerData(
+					gomock.Any(), gomock.Any(),
+				).Return(nil, nil)
+			},
+		},
+	}
+	for _, p := range patterns {
+		t.Run(p.desc, func(t *testing.T) {
+			p.setup(deleter)
+			err := deleter.Run(ctx)
+			if p.expected != nil {
+				assert.Equal(t, p.expected, err)
+				return
+			}
+		})
+	}
+}
+
+func newMockDemoOrganizationDeleter(t *testing.T, c *gomock.Controller) *demoOrganizationDeleter {
+	t.Helper()
+	logger, err := log.NewLogger()
+	assert.Nil(t, err)
+	return &demoOrganizationDeleter{
+		organizationStorage: storagemock.NewMockOrganizationStorage(c),
+		environmentClient:   environmentclient.NewMockClient(c),
+		opts: &jobs.Options{
+			Timeout: 5 * time.Second,
+		},
+		logger: logger,
+	}
+}