Implement buf alpha repo sync

private/buf/bufsync/bufsync.go

@@ -0,0 +1,100 @@
+// Copyright 2020-2023 Buf Technologies, Inc.
+//
+// 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 bufsync
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/bufbuild/buf/private/bufpkg/bufmodule/bufmoduleref"
+	"github.com/bufbuild/buf/private/pkg/git"
+	"github.com/bufbuild/buf/private/pkg/storage"
+	"github.com/bufbuild/buf/private/pkg/storage/storagegit"
+	"go.uber.org/zap"
+)
+
+// Syncer syncs a modules in a git.Repository.
+type Syncer interface {
+	// Sync syncs the repository using the provided PushFunc. It processes
+	// commits in reverse topological order, loads any configured named
+	// modules, extracts any Git metadata for that commit, and invokes
+	// PushFunc with a ModuleCommit.
+	//
+	// Only commits/branches belonging to the remote named 'origin' are
+	// processed. All tags are processed.
+	Sync(context.Context, PushFunc) error
+}
+
+// NewSyncer creates a new Syncer.
+func NewSyncer(
+	logger *zap.Logger,
+	repo git.Repository,
+	storageGitProvider storagegit.Provider,
+	options ...SyncerOption,
+) (Syncer, error) {
+	return newSyncer(
+		logger,
+		repo,
+		storageGitProvider,
+		options...,
+	)
+}
+
+// SyncerOption configures the creation of a new Syncer.
+type SyncerOption func(*syncer) error
+
+// SyncerWithModule configures a Syncer to sync the specified module. The module
+// identity override is optional.
+//
+// This option can be provided multiple times to sync multiple distinct modules.
+func SyncerWithModule(dir string, identityOverride bufmoduleref.ModuleIdentity) SyncerOption {
+	return func(s *syncer) error {
+		for _, existingModule := range s.modulesToSync {
+			if existingModule.dir != dir {
+				continue
+			}
+			if identityOverride == nil && existingModule.identityOverride == nil {
+				return fmt.Errorf("duplicate module %s", dir)
+			}
+			if identityOverride != nil &&
+				existingModule.identityOverride != nil &&
+				identityOverride.IdentityString() == existingModule.identityOverride.IdentityString() {
+				return fmt.Errorf("duplicate module %s:%s", dir, identityOverride.IdentityString())
+			}
+		}
+		s.modulesToSync = append(s.modulesToSync, newSyncableModule(
+			dir,
+			identityOverride,
+		))
+		return nil
+	}
+}
+
+// PushFunc is invoked by Syncer to process a sync point.
+type PushFunc func(ctx context.Context, commit ModuleCommit) error
+
+// ModuleCommit is a module at a particular commit.
+type ModuleCommit interface {
+	// Identity is the identity of the module, accounting for any configured override.
+	Identity() bufmoduleref.ModuleIdentity
+	// Bucket is the bucket for the module.
+	Bucket() storage.ReadBucket
+	// Commit is the commit that the module is sourced from.
+	Commit() git.Commit
+	// Branch is the git branch that this module is sourced from.
+	Branch() string
+	// Tags are the git tags associated with Commit.
+	Tags() []string
+}

private/buf/bufsync/module_commit.go

@@ -0,0 +1,65 @@
+// Copyright 2020-2023 Buf Technologies, Inc.
+//
+// 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 bufsync
+
+import (
+	"github.com/bufbuild/buf/private/bufpkg/bufmodule/bufmoduleref"
+	"github.com/bufbuild/buf/private/pkg/git"
+	"github.com/bufbuild/buf/private/pkg/storage"
+)
+
+type moduleCommit struct {
+	identity bufmoduleref.ModuleIdentity
+	bucket   storage.ReadBucket
+	commit   git.Commit
+	branch   string
+	tags     []string
+}
+
+func newModuleCommit(
+	identity bufmoduleref.ModuleIdentity,
+	bucket storage.ReadBucket,
+	commit git.Commit,
+	branch string,
+	tags []string,
+) ModuleCommit {
+	return &moduleCommit{
+		identity: identity,
+		bucket:   bucket,
+		commit:   commit,
+		branch:   branch,
+		tags:     tags,
+	}
+}
+
+func (m *moduleCommit) Identity() bufmoduleref.ModuleIdentity {
+	return m.identity
+}
+
+func (m *moduleCommit) Bucket() storage.ReadBucket {
+	return m.bucket
+}
+
+func (m *moduleCommit) Commit() git.Commit {
+	return m.commit
+}
+
+func (m *moduleCommit) Branch() string {
+	return m.branch
+}
+
+func (m *moduleCommit) Tags() []string {
+	return m.tags
+}

private/buf/bufsync/syncable_module.go

@@ -0,0 +1,34 @@
+// Copyright 2020-2023 Buf Technologies, Inc.
+//
+// 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 bufsync
+
+import (
+	"github.com/bufbuild/buf/private/bufpkg/bufmodule/bufmoduleref"
+)
+
+type syncableModule struct {
+	dir              string
+	identityOverride bufmoduleref.ModuleIdentity
+}
+
+func newSyncableModule(
+	dir string,
+	identityOverride bufmoduleref.ModuleIdentity,
+) syncableModule {
+	return syncableModule{
+		dir:              dir,
+		identityOverride: identityOverride,
+	}
+}

private/buf/bufsync/syncer.go

@@ -0,0 +1,174 @@
+// Copyright 2020-2023 Buf Technologies, Inc.
+//
+// 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 bufsync
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/bufbuild/buf/private/bufpkg/bufconfig"
+	"github.com/bufbuild/buf/private/bufpkg/bufmodule/bufmodulebuild"
+	"github.com/bufbuild/buf/private/pkg/git"
+	"github.com/bufbuild/buf/private/pkg/storage"
+	"github.com/bufbuild/buf/private/pkg/storage/storagegit"
+	"go.uber.org/zap"
+)
+
+type syncer struct {
+	logger             *zap.Logger
+	repo               git.Repository
+	storageGitProvider storagegit.Provider
+	modulesToSync      []syncableModule
+
+	knownTagsByCommitHash map[string][]string
+}
+
+func newSyncer(
+	logger *zap.Logger,
+	repo git.Repository,
+	storageGitProvider storagegit.Provider,
+	options ...SyncerOption,
+) (Syncer, error) {
+	s := &syncer{
+		logger:             logger,
+		repo:               repo,
+		storageGitProvider: storageGitProvider,
+	}
+	for _, opt := range options {
+		if err := opt(s); err != nil {
+			return nil, err
+		}
+	}
+	return s, nil
+}
+
+func (s *syncer) Sync(ctx context.Context, pushFunc PushFunc) error {
+	s.knownTagsByCommitHash = map[string][]string{}
+	if err := s.repo.ForEachTag(func(tag string, commitHash git.Hash) error {
+		s.knownTagsByCommitHash[commitHash.Hex()] = append(s.knownTagsByCommitHash[commitHash.Hex()], tag)
+		return nil
+	}); err != nil {
+		return fmt.Errorf("process tags: %w", err)
+	}
+	// TODO: sync other branches
+	for _, branch := range []string{s.repo.BaseBranch()} {
+		// TODO: resume from last sync point
+		if err := s.repo.ForEachCommit(branch, func(commit git.Commit) error {
+			for _, module := range s.modulesToSync {
+				if err := s.visitCommit(
+					ctx,
+					module,
+					branch,
+					commit,
+					pushFunc,
+				); err != nil {
+					return fmt.Errorf("process commit %s: %w", commit.Hash().Hex(), err)
+				}
+			}
+			return nil
+		}); err != nil {
+			return fmt.Errorf("process commits: %w", err)
+		}
+	}
+	return nil
+}
+
+// visitCommit looks for the module in the commit, and if found tries to validate it.
+// If it is valid, it invokes `pushFunc`.
+//
+// It does not return errors on invalid modules, but it will return any errors from
+// `pushFunc` as those may be transient.
+func (s *syncer) visitCommit(
+	ctx context.Context,
+	module syncableModule,
+	branch string,
+	commit git.Commit,
+	pushFunc PushFunc,
+) error {
+	sourceBucket, err := s.storageGitProvider.NewReadBucket(
+		commit.Tree(),
+		storagegit.ReadBucketWithSymlinksIfSupported(),
+	)
+	if err != nil {
+		return err
+	}
+	sourceBucket = storage.MapReadBucket(sourceBucket, storage.MapOnPrefix(module.dir))
+	foundModule, err := bufconfig.ExistingConfigFilePath(ctx, sourceBucket)
+	if err != nil {
+		return err
+	}
+	if foundModule == "" {
+		// We did not find a module. Carry on to the next commit.
+		s.logger.Debug(
+			"module not found, skipping commit",
+			zap.String("commit", commit.Hash().String()),
+			zap.String("module", module.dir),
+		)
+		return nil
+	}
+	sourceConfig, err := bufconfig.GetConfigForBucket(ctx, sourceBucket)
+	if err != nil {
+		// We found a module but the module config is invalid. We can warn on this
+		// and carry on. Note that because of resumption, we will typically only come
+		// across this commit once, we will not log this warning again.
+		s.logger.Warn(
+			"invalid module",
+			zap.String("commit", commit.Hash().String()),
+			zap.String("module", module.dir),
+			zap.Error(err),
+		)
+		return nil
+	}
+	if sourceConfig.ModuleIdentity == nil {
+		// Unnamed module. Carry on.
+		s.logger.Debug(
+			"unnamed module, skipping commit",
+			zap.String("commit", commit.Hash().String()),
+			zap.String("module", module.dir),
+		)
+		return nil
+	}
+	moduleIdentity := sourceConfig.ModuleIdentity
+	if module.identityOverride != nil {
+		moduleIdentity = module.identityOverride
+	}
+	builtModule, err := bufmodulebuild.BuildForBucket(
+		ctx,
+		sourceBucket,
+		sourceConfig.Build,
+	)
+	if err != nil {
+		// We failed to build the module. We can warn on this
+		// and carry on. Note that because of resumption, we will typically only come
+		// across this commit once, we will not log this warning again.
+		s.logger.Warn(
+			"invalid module",
+			zap.String("commit", commit.Hash().String()),
+			zap.String("module", module.dir),
+			zap.Error(err),
+		)
+		return nil
+	}
+	return pushFunc(
+		ctx,
+		newModuleCommit(
+			moduleIdentity,
+			builtModule.Bucket,
+			commit,
+			branch,
+			s.knownTagsByCommitHash[commit.Hash().Hex()],
+		),
+	)
+}

private/buf/bufsync/usage.gen.go

@@ -0,0 +1,19 @@
+// Copyright 2020-2023 Buf Technologies, Inc.
+//
+// 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.
+
+// Generated. DO NOT EDIT.
+
+package bufsync
+
+import _ "github.com/bufbuild/buf/private/usage"