chore: move warehouse handle to router

runner/runner.go

@@ -339,8 +339,6 @@ func runAllInit() {
 	jobsdb.Init()
 	jobsdb.Init2()
 	warehouse.Init()
-	warehouse.Init2()
-	warehouse.Init3()
 	warehouse.Init4()
 	warehouse.Init6()
 	warehousearchiver.Init()

warehouse/backend_config.go

@@ -100,15 +100,15 @@ func (s *backendConfigManager) Start(ctx context.Context) {
 func (s *backendConfigManager) Subscribe(ctx context.Context) <-chan []model.Warehouse {
 	s.subscriptionsMu.Lock()
 	defer s.subscriptionsMu.Unlock()
-	s.warehousesMu.Lock()
-	defer s.warehousesMu.Unlock()
 
 	ch := make(chan []model.Warehouse, 10)
 	s.subscriptions = append(s.subscriptions, ch)
 
+	s.warehousesMu.Lock()
 	if len(s.warehouses) > 0 {
 		ch <- s.warehouses
 	}
+	s.warehousesMu.Unlock()
 
 	go func() {
 		<-ctx.Done()
@@ -249,6 +249,7 @@ func (s *backendConfigManager) namespace(ctx context.Context, source backendconf
 			logfield.DestinationID, destination.ID,
 			logfield.DestinationType, destination.DestinationDefinition.Name,
 			logfield.WorkspaceID, destination.WorkspaceID,
+			logfield.Error, err.Error(),
 		)
 		return ""
 	}

warehouse/backend_config_test.go

@@ -288,13 +288,15 @@ func TestBackendConfigManager(t *testing.T) {
 
 func TestBackendConfigManager_Namespace(t *testing.T) {
 	testcases := []struct {
+		name              string
 		config            map[string]interface{}
 		source            backendconfig.SourceT
 		destination       backendconfig.DestinationT
 		expectedNamespace string
 		setConfig         bool
 	}{
 		{
+			name:   "clickhouse with database configured in config",
 			source: backendconfig.SourceT{},
 			destination: backendconfig.DestinationT{
 				Config: map[string]interface{}{
@@ -308,6 +310,7 @@ func TestBackendConfigManager_Namespace(t *testing.T) {
 			setConfig:         false,
 		},
 		{
+			name:   "clickhouse without database configured in config",
 			source: backendconfig.SourceT{},
 			destination: backendconfig.DestinationT{
 				Config: map[string]interface{}{},
@@ -319,19 +322,7 @@ func TestBackendConfigManager_Namespace(t *testing.T) {
 			setConfig:         false,
 		},
 		{
-			source: backendconfig.SourceT{},
-			destination: backendconfig.DestinationT{
-				Config: map[string]interface{}{
-					"namespace": "test_namespace",
-				},
-				DestinationDefinition: backendconfig.DestinationDefinitionT{
-					Name: "test-destinationType-1",
-				},
-			},
-			expectedNamespace: "test_namespace",
-			setConfig:         false,
-		},
-		{
+			name:   "namespace only contains letters",
 			source: backendconfig.SourceT{},
 			destination: backendconfig.DestinationT{
 				Config: map[string]interface{}{
@@ -345,6 +336,7 @@ func TestBackendConfigManager_Namespace(t *testing.T) {
 			setConfig:         false,
 		},
 		{
+			name:   "namespace only contains special characters",
 			source: backendconfig.SourceT{},
 			destination: backendconfig.DestinationT{
 				Config: map[string]interface{}{
@@ -358,6 +350,7 @@ func TestBackendConfigManager_Namespace(t *testing.T) {
 			setConfig:         false,
 		},
 		{
+			name:   "namespace contains special characters and letters",
 			source: backendconfig.SourceT{},
 			destination: backendconfig.DestinationT{
 				Config: map[string]interface{}{
@@ -371,6 +364,7 @@ func TestBackendConfigManager_Namespace(t *testing.T) {
 			setConfig:         false,
 		},
 		{
+			name:   "empty namespace but config is set",
 			source: backendconfig.SourceT{},
 			destination: backendconfig.DestinationT{
 				Config: map[string]interface{}{},
@@ -382,6 +376,7 @@ func TestBackendConfigManager_Namespace(t *testing.T) {
 			setConfig:         true,
 		},
 		{
+			name: "empty namespace with picking from cache",
 			source: backendconfig.SourceT{
 				Name: "test-source",
 				ID:   "test-sourceID",
@@ -397,6 +392,7 @@ func TestBackendConfigManager_Namespace(t *testing.T) {
 			setConfig:         false,
 		},
 		{
+			name: "destination config without namespace configured and custom dataset prefix is not configured",
 			source: backendconfig.SourceT{
 				Name: "test-source",
 				ID:   "random-sourceID",
@@ -412,6 +408,7 @@ func TestBackendConfigManager_Namespace(t *testing.T) {
 			setConfig:         false,
 		},
 		{
+			name: "destination config without namespace configured and custom dataset prefix configured",
 			source: backendconfig.SourceT{
 				Name: "test-source",
 				ID:   "test-sourceID",
@@ -431,7 +428,7 @@ func TestBackendConfigManager_Namespace(t *testing.T) {
 	for _, tc := range testcases {
 		tc := tc
 
-		t.Run("should return namespace", func(t *testing.T) {
+		t.Run(tc.name, func(t *testing.T) {
 			pool, err := dockertest.NewPool("")
 			require.NoError(t, err)
 

warehouse/internal/repo/upload.go

@@ -632,3 +632,54 @@ func (uploads *Uploads) PendingTableUploads(ctx context.Context, namespace strin
 	}
 	return pendingTableUploads, nil
 }
+
+func (uploads *Uploads) ResetInProgress(ctx context.Context, destType string) error {
+	_, err := uploads.db.ExecContext(ctx, `
+		UPDATE
+			`+uploadsTableName+`
+		SET
+			in_progress = FALSE
+		WHERE
+			destination_type = $1 AND
+			in_progress = TRUE;
+	`,
+		destType,
+	)
+	if err != nil {
+		return fmt.Errorf("reset in progress: %w", err)
+	}
+	return nil
+}
+
+func (uploads *Uploads) LastCreatedAt(ctx context.Context, sourceID, destinationID string) (time.Time, error) {
+	row := uploads.db.QueryRowContext(ctx, `
+		SELECT
+			created_at
+		FROM
+		`+uploadsTableName+`
+		WHERE
+			source_id = $1 AND
+			destination_id = $2
+		ORDER BY
+			id DESC
+		LIMIT 1;
+	`,
+		sourceID,
+		destinationID,
+	)
+
+	var createdAt sql.NullTime
+
+	err := row.Scan(&createdAt)
+	if err == sql.ErrNoRows {
+		return time.Time{}, nil
+	}
+	if err != nil {
+		return time.Time{}, fmt.Errorf("last created at: %w", err)
+	}
+	if !createdAt.Valid {
+		return time.Time{}, nil
+	}
+
+	return createdAt.Time, nil
+}

warehouse/internal/repo/upload_test.go

@@ -1000,3 +1000,125 @@ func TestUploads_PendingTableUploads(t *testing.T) {
 		require.EqualError(t, err, "pending table uploads: context canceled")
 	})
 }
+
+func TestUploads_ResetInProgress(t *testing.T) {
+	const (
+		sourceID        = "source_id"
+		destinationID   = "destination_id"
+		destinationType = "destination_type"
+	)
+
+	db, ctx := setupDB(t), context.Background()
+
+	now := time.Date(2023, 1, 1, 0, 0, 0, 0, time.UTC)
+	repoUpload := repo.NewUploads(db, repo.WithNow(func() time.Time {
+		return now
+	}))
+
+	t.Run("success", func(t *testing.T) {
+		repoStaging := repo.NewStagingFiles(db, repo.WithNow(func() time.Time {
+			return now
+		}))
+
+		stagingID, err := repoStaging.Insert(ctx, &model.StagingFileWithSchema{})
+		require.NoError(t, err)
+
+		uploadID, err := repoUpload.CreateWithStagingFiles(ctx, model.Upload{
+			SourceID:        sourceID,
+			DestinationID:   destinationID,
+			DestinationType: destinationType,
+			Status:          model.Waiting,
+		}, []*model.StagingFile{
+			{
+				ID:            stagingID,
+				SourceID:      sourceID,
+				DestinationID: destinationID,
+			},
+		})
+		require.NoError(t, err)
+
+		_, err = db.ExecContext(ctx, `UPDATE wh_uploads SET in_progress = TRUE WHERE id = $1;`, uploadID)
+		require.NoError(t, err)
+
+		uploadsToProcess, err := repoUpload.GetToProcess(ctx, destinationType, 1, repo.ProcessOptions{})
+		require.NoError(t, err)
+		require.Len(t, uploadsToProcess, 0)
+
+		err = repoUpload.ResetInProgress(ctx, destinationType)
+		require.NoError(t, err)
+
+		uploadsToProcess, err = repoUpload.GetToProcess(ctx, destinationType, 1, repo.ProcessOptions{})
+		require.NoError(t, err)
+		require.Len(t, uploadsToProcess, 1)
+	})
+
+	t.Run("context cancelled", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(ctx)
+		cancel()
+
+		err := repoUpload.ResetInProgress(ctx, destinationType)
+		require.EqualError(t, err, "reset in progress: context canceled")
+	})
+}
+
+func TestUploads_LastCreatedAt(t *testing.T) {
+	const (
+		sourceID        = "source_id"
+		destinationID   = "destination_id"
+		destinationType = "destination_type"
+	)
+
+	db, ctx := setupDB(t), context.Background()
+
+	now := time.Date(2023, 1, 1, 0, 0, 0, 0, time.UTC)
+	repoUpload := repo.NewUploads(db, repo.WithNow(func() time.Time {
+		return now
+	}))
+
+	t.Run("many uploads", func(t *testing.T) {
+		for i := 0; i < 5; i++ {
+			repoStaging := repo.NewStagingFiles(db, repo.WithNow(func() time.Time {
+				return now
+			}))
+			stagingID, err := repoStaging.Insert(ctx, &model.StagingFileWithSchema{})
+			require.NoError(t, err)
+
+			repoUpload := repo.NewUploads(db, repo.WithNow(func() time.Time {
+				return now.Add(time.Second * time.Duration(i+1))
+			}))
+
+			_, err = repoUpload.CreateWithStagingFiles(ctx, model.Upload{
+				SourceID:        sourceID,
+				DestinationID:   destinationID,
+				DestinationType: destinationType,
+				Status:          model.Waiting,
+			}, []*model.StagingFile{
+				{
+					ID:            stagingID,
+					SourceID:      sourceID,
+					DestinationID: destinationID,
+				},
+			})
+			require.NoError(t, err)
+		}
+
+		lastCreatedAt, err := repoUpload.LastCreatedAt(ctx, sourceID, destinationID)
+		require.NoError(t, err)
+		require.Equal(t, lastCreatedAt.UTC(), now.Add(time.Second*5).UTC())
+	})
+
+	t.Run("no uploads", func(t *testing.T) {
+		lastCreatedAt, err := repoUpload.LastCreatedAt(ctx, "unknown_source", "unknown_destination")
+		require.NoError(t, err)
+		require.Equal(t, lastCreatedAt, time.Time{})
+	})
+
+	t.Run("context cancelled", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(ctx)
+		cancel()
+
+		lastCreatedAt, err := repoUpload.LastCreatedAt(ctx, sourceID, destinationID)
+		require.EqualError(t, err, "last created at: context canceled")
+		require.Equal(t, lastCreatedAt, time.Time{})
+	})
+}