<commit_before>package server

import (
	"bytes"
	"fmt"
	"io"
	"math/rand"
	"strings"

	"google.golang.org/grpc"

	"golang.org/x/net/context"

	"github.com/pachyderm/pachyderm/src/pfs"
	"github.com/pachyderm/pachyderm/src/pfs/drive"
	"github.com/pachyderm/pachyderm/src/pfs/route"
	"go.pedge.io/google-protobuf"
	"go.pedge.io/proto/stream"
)

var (
	emptyInstance = &google_protobuf.Empty{}
)

type combinedAPIServer struct {
	sharder route.Sharder
	router  route.Router
	driver  drive.Driver
}

func newCombinedAPIServer(
	sharder route.Sharder,
	router route.Router,
	driver drive.Driver,
) *combinedAPIServer {
	return &combinedAPIServer{
		sharder,
		router,
		driver,
	}
}

func (a *combinedAPIServer) RepoCreate(ctx context.Context, request *pfs.RepoCreateRequest) (*google_protobuf.Empty, error) {
	masterShards, err := a.router.GetMasterShards()
	if err != nil {
		return nil, err
	}
	if err := a.driver.RepoCreate(request.Repo, masterShards); err != nil {
		return nil, err
	}
	replicaShards, err := a.router.GetReplicaShards()
	if err != nil {
		return nil, err
	}
	if err := a.driver.RepoCreate(request.Repo, replicaShards); err != nil {
		return nil, err
	}
	if !request.Redirect {
		clientConns, err := a.router.GetAllClientConns()
		if err != nil {
			return nil, err
		}
		for _, clientConn := range clientConns {
			if _, err := pfs.NewApiClient(clientConn).RepoCreate(
				ctx,
				&pfs.RepoCreateRequest{
					Repo:     request.Repo,
					Redirect: true,
				},
			); err != nil {
				return nil, err
			}
		}
		// Create the initial commit
		if _, err = a.CommitStart(ctx, &pfs.CommitStartRequest{
			Parent: nil,
			Commit: &pfs.Commit{
				Repo: request.Repo,
				Id:   InitialCommitID,
			},
			Redirect: false,
		}); err != nil {
			return nil, err
		}
		if _, err = a.CommitFinish(ctx, &pfs.CommitFinishRequest{
			Commit: &pfs.Commit{
				Repo: request.Repo,
				Id:   InitialCommitID,
			},
			Redirect: false,
		}); err != nil {
			return nil, err
		}
	}
	return emptyInstance, nil
}

func (a *combinedAPIServer) RepoInspect(ctx context.Context, request *pfs.RepoInspectRequest) (*pfs.RepoInspectResponse, error) {
	masterShards, err := a.router.GetMasterShards()
	if err != nil {
		return nil, err
	}
	for shard := range masterShards {
		repoInfo, ok, err := a.driver.RepoInspect(request.Repo, shard)
		if err != nil {
			return nil, err
		}
		if !ok {
			return &pfs.RepoInspectResponse{}, nil
		}
		return &pfs.RepoInspectResponse{RepoInfo: repoInfo}, nil
	}
	if !request.Redirect {
		clientConns, err := a.router.GetAllClientConns()
		if err != nil {
			return nil, err
		}
		for _, clientConn := range clientConns {
			return pfs.NewApiClient(clientConn).RepoInspect(
				ctx,
				&pfs.RepoInspectRequest{
					Repo:     request.Repo,
					Redirect: true,
				},
			)
		}
	}
	return nil, fmt.Errorf("pachyderm: no available masters")
}

func (a *combinedAPIServer) RepoList(ctx context.Context, request *pfs.RepoListRequest) (*pfs.RepoListResponse, error) {
	masterShards, err := a.router.GetMasterShards()
	if err != nil {
		return nil, err
	}
	for shard := range masterShards {
		repos, err := a.driver.RepoList(shard)
		if err != nil {
			return nil, err
		}
		return &pfs.RepoListResponse{RepoInfo: repos}, nil
	}
	if !request.Redirect {
		clientConns, err := a.router.GetAllClientConns()
		if err != nil {
			return nil, err
		}
		for _, clientConn := range clientConns {
			return pfs.NewApiClient(clientConn).RepoList(ctx, &pfs.RepoListRequest{Redirect: true})
		}
	}
	return nil, fmt.Errorf("pachyderm: no available masters")
}

func (a *combinedAPIServer) FileGet(request *pfs.FileGetRequest, apiFileGetServer pfs.Api_FileGetServer) (retErr error) {
	shard, clientConn, err := a.getShardAndClientConnIfNecessary(request.File, false)
	if err != nil {
		return err
	}
	if clientConn != nil {
		apiFileGetClient, err := pfs.NewApiClient(clientConn).FileGet(context.Background(), request)
		if err != nil {
			return err
		}
		return protostream.RelayFromStreamingBytesClient(apiFileGetClient, apiFileGetServer)
	}
	file, err := a.driver.FileGet(request.File, shard)
	if err != nil {
		return err
	}
	defer func() {
		if err := file.Close(); err != nil && retErr == nil {<commit_msg>Adds RepoDelete method to combinedAPIServer.<commit_after>package server

import (
	"bytes"
	"fmt"
	"io"
	"math/rand"
	"strings"

	"google.golang.org/grpc"

	"golang.org/x/net/context"

	"github.com/pachyderm/pachyderm/src/pfs"
	"github.com/pachyderm/pachyderm/src/pfs/drive"
	"github.com/pachyderm/pachyderm/src/pfs/route"
	"go.pedge.io/google-protobuf"
	"go.pedge.io/proto/stream"
)

var (
	emptyInstance = &google_protobuf.Empty{}
)

type combinedAPIServer struct {
	sharder route.Sharder
	router  route.Router
	driver  drive.Driver
}

func newCombinedAPIServer(
	sharder route.Sharder,
	router route.Router,
	driver drive.Driver,
) *combinedAPIServer {
	return &combinedAPIServer{
		sharder,
		router,
		driver,
	}
}

func (a *combinedAPIServer) RepoCreate(ctx context.Context, request *pfs.RepoCreateRequest) (*google_protobuf.Empty, error) {
	masterShards, err := a.router.GetMasterShards()
	if err != nil {
		return nil, err
	}
	if err := a.driver.RepoCreate(request.Repo, masterShards); err != nil {
		return nil, err
	}
	replicaShards, err := a.router.GetReplicaShards()
	if err != nil {
		return nil, err
	}
	if err := a.driver.RepoCreate(request.Repo, replicaShards); err != nil {
		return nil, err
	}
	if !request.Redirect {
		clientConns, err := a.router.GetAllClientConns()
		if err != nil {
			return nil, err
		}
		for _, clientConn := range clientConns {
			if _, err := pfs.NewApiClient(clientConn).RepoCreate(
				ctx,
				&pfs.RepoCreateRequest{
					Repo:     request.Repo,
					Redirect: true,
				},
			); err != nil {
				return nil, err
			}
		}
		// Create the initial commit
		if _, err = a.CommitStart(ctx, &pfs.CommitStartRequest{
			Parent: nil,
			Commit: &pfs.Commit{
				Repo: request.Repo,
				Id:   InitialCommitID,
			},
			Redirect: false,
		}); err != nil {
			return nil, err
		}
		if _, err = a.CommitFinish(ctx, &pfs.CommitFinishRequest{
			Commit: &pfs.Commit{
				Repo: request.Repo,
				Id:   InitialCommitID,
			},
			Redirect: false,
		}); err != nil {
			return nil, err
		}
	}
	return emptyInstance, nil
}

func (a *combinedAPIServer) RepoInspect(ctx context.Context, request *pfs.RepoInspectRequest) (*pfs.RepoInspectResponse, error) {
	masterShards, err := a.router.GetMasterShards()
	if err != nil {
		return nil, err
	}
	for shard := range masterShards {
		repoInfo, ok, err := a.driver.RepoInspect(request.Repo, shard)
		if err != nil {
			return nil, err
		}
		if !ok {
			return &pfs.RepoInspectResponse{}, nil
		}
		return &pfs.RepoInspectResponse{RepoInfo: repoInfo}, nil
	}
	if !request.Redirect {
		clientConns, err := a.router.GetAllClientConns()
		if err != nil {
			return nil, err
		}
		for _, clientConn := range clientConns {
			return pfs.NewApiClient(clientConn).RepoInspect(
				ctx,
				&pfs.RepoInspectRequest{
					Repo:     request.Repo,
					Redirect: true,
				},
			)
		}
	}
	return nil, fmt.Errorf("pachyderm: no available masters")
}

func (a *combinedAPIServer) RepoList(ctx context.Context, request *pfs.RepoListRequest) (*pfs.RepoListResponse, error) {
	masterShards, err := a.router.GetMasterShards()
	if err != nil {
		return nil, err
	}
	for shard := range masterShards {
		repos, err := a.driver.RepoList(shard)
		if err != nil {
			return nil, err
		}
		return &pfs.RepoListResponse{RepoInfo: repos}, nil
	}
	if !request.Redirect {
		clientConns, err := a.router.GetAllClientConns()
		if err != nil {
			return nil, err
		}
		for _, clientConn := range clientConns {
			return pfs.NewApiClient(clientConn).RepoList(ctx, &pfs.RepoListRequest{Redirect: true})
		}
	}
	return nil, fmt.Errorf("pachyderm: no available masters")
}

func (a *combinedAPIServer) RepoDelete(ctx context.Context, request *pfs.RepoDeleteRequest) (*google_protobuf.Empty, error) {
	masterShards, err := a.router.GetMasterShards()
	if err != nil {
		return nil, err
	}
	if err := a.driver.RepoDelete(request.Repo, masterShards); err != nil {
		return nil, err
	}
	replicaShards, err := a.router.GetReplicaShards()
	if err != nil {
		return nil, err
	}
	if err := a.driver.RepoDelete(request.Repo, replicaShards); err != nil {
		return nil, err
	}
	if !request.Redirect {
		clientConns, err := a.router.GetAllClientConns()
		if err != nil {
			return nil, err
		}
		for _, clientConn := range clientConns {
			if _, err := pfs.NewApiClient(clientConn).RepoCreate(
				ctx,
				&pfs.RepoCreateRequest{
					Repo:     request.Repo,
					Redirect: true,
				},
			); err != nil {
				return nil, err
			}
		}
	}
	return emptyInstance, nil

}

func (a *combinedAPIServer) FileGet(request *pfs.FileGetRequest, apiFileGetServer pfs.Api_FileGetServer) (retErr error) {
	shard, clientConn, err := a.getShardAndClientConnIfNecessary(request.File, false)
	if err != nil {
		return err
	}
	if clientConn != nil {
		apiFileGetClient, err := pfs.NewApiClient(clientConn).FileGet(context.Background(), request)
		if err != nil {
			return err
		}
		return protostream.RelayFromStreamingBytesClient(apiFileGetClient, apiFileGetServer)
	}
	file, err := a.driver.FileGet(request.File, shard)
	if err != nil {
		return err
	}
	defer func() {
		if err := file.Close(); err != nil && retErr == nil {
{'fraction_non_alphanumeric': 0.12882189437673722, 'fraction_numerical': 0.0, 'mean_word_length': 3.6565455450472872, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2349640', 'n_tokens_mistral': 3768, 'n_tokens_neox': 3486, 'n_words': 763}