abstract kopia setup code (#4637)

Refactors the repository code to abstract code responsible for kopia related setup.

#### Does this PR need a docs update or release note?

- [ ] ✅ Yes, it's included
- [ ] 🕐 Yes, but in a later PR
- [x] ⛔ No

#### Type of change

- [ ] 🌻 Feature
- [ ] 🐛 Bugfix
- [ ] 🗺️ Documentation
- [ ] 🤖 Supportability/Tests
- [ ] 💻 CI/Deployment
- [ ] 🧹 Tech Debt/Cleanup

#### Issue(s)

* #

#### Test Plan

- [ ] 💪 Manual
- [ ] ⚡ Unit test
- [ ] 💚 E2E

src/pkg/repository/repository.go

@@ -140,18 +140,10 @@ type InitConfig struct {
 //   - update maintenance retention parameters as needed
 //   - store the configuration details
 //   - connect to the provider
-func (r *repository) Initialize(
+func (r *repository) Initialize(ctx context.Context, cfg InitConfig) (err error) {
-	ctx context.Context,
+	ctx = r.addContextClues(ctx)
-	cfg InitConfig,
-) (err error) {
-	ctx = clues.Add(
-		ctx,
-		"acct_provider", r.Account.Provider.String(),
-		"acct_id", clues.Hide(r.Account.ID()),
-		"storage_provider", r.Storage.Provider.String())
-
 	defer func() {
-		if crErr := crash.Recovery(ctx, recover(), "repo init"); crErr != nil {
+		if crErr := crash.Recovery(ctx, recover(), "repo connect"); crErr != nil {
 			err = crErr
 		}
 	}()
@@ -162,31 +154,8 @@ func (r *repository) Initialize(
 
 	observe.Message(ctx, "Initializing repository")
 
-	kopiaRef := kopia.NewConn(r.Storage)
+	if err := r.setupKopia(ctx, cfg.RetentionOpts, true); err != nil {
-	if err := kopiaRef.Initialize(ctx, r.Opts.Repo, cfg.RetentionOpts); err != nil {
+		return err
-		// replace common internal errors so that sdk users can check results with errors.Is()
-		if errors.Is(err, kopia.ErrorRepoAlreadyExists) {
-			return clues.Stack(ErrorRepoAlreadyExists, err).WithClues(ctx)
-		}
-
-		return clues.Wrap(err, "initializing kopia")
-	}
-	// kopiaRef comes with a count of 1 and NewWrapper/NewModelStore bumps it again so safe
-	// to close here.
-	defer kopiaRef.Close(ctx)
-
-	r.dataLayer, err = kopia.NewWrapper(kopiaRef)
-	if err != nil {
-		return clues.Stack(err).WithClues(ctx)
-	}
-
-	r.modelStore, err = kopia.NewModelStore(kopiaRef)
-	if err != nil {
-		return clues.Stack(err).WithClues(ctx)
-	}
-
-	if err := newRepoModel(ctx, r.modelStore, r.ID); err != nil {
-		return clues.Wrap(err, "setting up repository").WithClues(ctx)
 	}
 
 	r.Bus.Event(ctx, events.RepoInit, nil)
@@ -205,16 +174,8 @@ type ConnConfig struct {
 //   - connect to the m365 account
 //   - connect to the provider storage
 //   - return the connected repository
-func (r *repository) Connect(
+func (r *repository) Connect(ctx context.Context, cfg ConnConfig) (err error) {
-	ctx context.Context,
+	ctx = r.addContextClues(ctx)
-	cfg ConnConfig,
-) (err error) {
-	ctx = clues.Add(
-		ctx,
-		"acct_provider", r.Account.Provider.String(),
-		"acct_id", clues.Hide(r.Account.ID()),
-		"storage_provider", r.Storage.Provider.String())
-
 	defer func() {
 		if crErr := crash.Recovery(ctx, recover(), "repo connect"); crErr != nil {
 			err = crErr
@@ -227,31 +188,8 @@ func (r *repository) Connect(
 
 	observe.Message(ctx, "Connecting to repository")
 
-	kopiaRef := kopia.NewConn(r.Storage)
+	if err := r.setupKopia(ctx, ctrlRepo.Retention{}, false); err != nil {
-	if err := kopiaRef.Connect(ctx, r.Opts.Repo); err != nil {
+		return clues.Stack(err)
-		return clues.Wrap(err, "connecting kopia client")
-	}
-	// kopiaRef comes with a count of 1 and NewWrapper/NewModelStore bumps it again so safe
-	// to close here.
-	defer kopiaRef.Close(ctx)
-
-	r.dataLayer, err = kopia.NewWrapper(kopiaRef)
-	if err != nil {
-		return clues.Stack(err).WithClues(ctx)
-	}
-
-	r.modelStore, err = kopia.NewModelStore(kopiaRef)
-	if err != nil {
-		return clues.Stack(err).WithClues(ctx)
-	}
-
-	if r.ID == events.RepoIDNotFound {
-		rm, err := getRepoModel(ctx, r.modelStore)
-		if err != nil {
-			return clues.Wrap(err, "retrieving repo model info")
-		}
-
-		r.ID = string(rm.ID)
 	}
 
 	r.Bus.Event(ctx, events.RepoConnect, nil)
@@ -263,12 +201,7 @@ func (r *repository) Connect(
 // - connect to the provider storage using existing password
 // - update the repo with new password
 func (r *repository) UpdatePassword(ctx context.Context, password string) (err error) {
-	ctx = clues.Add(
+	ctx = r.addContextClues(ctx)
-		ctx,
-		"acct_provider", r.Account.Provider.String(),
-		"acct_id", clues.Hide(r.Account.ID()),
-		"storage_provider", r.Storage.Provider.String())
-
 	defer func() {
 		if crErr := crash.Recovery(ctx, recover(), "repo connect"); crErr != nil {
 			err = crErr
@@ -346,6 +279,62 @@ func (r repository) Counter() *count.Bus {
 	return r.counter
 }
 
+func (r *repository) addContextClues(ctx context.Context) context.Context {
+	return clues.Add(
+		ctx,
+		"acct_provider", r.Account.Provider,
+		"acct_id", clues.Hide(r.Account.ID()),
+		"storage_provider", r.Storage.Provider)
+}
+
+func (r *repository) setupKopia(
+	ctx context.Context,
+	retentionOpts ctrlRepo.Retention,
+	isInitialize bool,
+) error {
+	var err error
+
+	kopiaRef := kopia.NewConn(r.Storage)
+	if isInitialize {
+		if err := kopiaRef.Initialize(ctx, r.Opts.Repo, retentionOpts); err != nil {
+			// Replace common internal errors so that SDK users can check results with errors.Is()
+			if errors.Is(err, kopia.ErrorRepoAlreadyExists) {
+				return clues.Stack(ErrorRepoAlreadyExists, err).WithClues(ctx)
+			}
+
+			return clues.Wrap(err, "initializing kopia")
+		}
+	} else {
+		if err := kopiaRef.Connect(ctx, r.Opts.Repo); err != nil {
+			return clues.Wrap(err, "connecting kopia client")
+		}
+	}
+
+	// kopiaRef comes with a count of 1, and NewWrapper/NewModelStore bumps it again, so it's safe to close here.
+	defer kopiaRef.Close(ctx)
+
+	r.dataLayer, err = kopia.NewWrapper(kopiaRef)
+	if err != nil {
+		return clues.Stack(err).WithClues(ctx)
+	}
+
+	r.modelStore, err = kopia.NewModelStore(kopiaRef)
+	if err != nil {
+		return clues.Stack(err).WithClues(ctx)
+	}
+
+	if r.ID == events.RepoIDNotFound {
+		rm, err := getRepoModel(ctx, r.modelStore)
+		if err != nil {
+			return clues.Wrap(err, "retrieving repo model info")
+		}
+
+		r.ID = string(rm.ID)
+	}
+
+	return nil
+}
+
 // ---------------------------------------------------------------------------
 // Repository ID Model
 // ---------------------------------------------------------------------------

src/pkg/services/m365/sites_test.go

@@ -78,7 +78,7 @@ func (suite *siteIntegrationSuite) TestSites_GetByID() {
 		suite.Run("site_"+s.ID, func() {
 			t := suite.T()
 			site, err := SiteByID(ctx, acct, s.ID)
-			assert.NoError(t, err, clues.ToCore(err))
+			require.NoError(t, err, clues.ToCore(err))
 			assert.NotEmpty(t, site.WebURL)
 			assert.NotEmpty(t, site.ID)
 			assert.NotEmpty(t, site.OwnerType)