feat(updates): hide/show demoted/un-demoted releases when updated upstream (#5505)

* feat(updates): hide/show demoted/un-demoted releases when updated upstream

* fix: add missing column

* chore: cleanup unused logic

* chore: feedback

* fix: edge cases around app demotion

* Update pkg/updatechecker/updatechecker.go

Co-authored-by: Salah Al Saleh <[email protected]>

---------

Co-authored-by: Salah Al Saleh <[email protected]>

Author: JGAntunes

migrations/tables/app_version.yaml

@@ -39,6 +39,11 @@ spec:
         constraints:
           notNull: true
         default: 0
+      - name: is_demoted
+        type: integer
+        constraints:
+          notNull: true
+        default: 0
       - name: release_notes
         type: text
       - name: supportbundle_spec

pkg/store/kotsstore/downstream_store.go

@@ -379,7 +379,8 @@ func (s *KOTSStore) GetDownstreamVersions(appID string, clusterID string, downlo
 	 adv.app_id = ado.app_id AND adv.cluster_id = ado.cluster_id AND adv.sequence = ado.downstream_sequence
  WHERE
 	 adv.app_id = ? AND
-	 adv.cluster_id = ?`
+	 adv.cluster_id = ? AND
+	 av.is_demoted = false`
 
 	if downloadedOnly {
 		query += fmt.Sprintf(` AND adv.status != '%s'`, types.VersionPendingDownload)

pkg/store/kotsstore/version_store.go

@@ -975,6 +975,25 @@ func (s *KOTSStore) UpdateAppVersionMetadata(appID string, update upstreamtypes.
 	return nil
 }
 
+// UpdateAppVersionDemotion updates the is_demoted field for a specific app version.
+func (s *KOTSStore) UpdateAppVersionDemotion(appID, channelID, cursor string, isDemoted bool) error {
+	db := persistence.MustGetDBSession()
+
+	query := `UPDATE app_version 
+			 SET is_demoted = ? 
+			 WHERE app_id = ? AND channel_id = ? AND update_cursor = ?`
+
+	_, err := db.WriteOneParameterized(gorqlite.ParameterizedStatement{
+		Query:     query,
+		Arguments: []interface{}{isDemoted, appID, channelID, cursor},
+	})
+	if err != nil {
+		return fmt.Errorf("update app version is demoted: %v", err)
+	}
+
+	return nil
+}
+
 func (s *KOTSStore) HasStrictPreflights(appID string, sequence int64) (bool, error) {
 	var preflightSpecStr gorqlite.NullString
 	db := persistence.MustGetDBSession()

pkg/store/mock/mock.go

@@ -191,6 +191,7 @@ type VersionStore interface {
 	GetNextAppSequence(appID string) (int64, error)
 	GetCurrentUpdateCursor(appID string, channelID string) (string, error)
 	UpdateAppVersionMetadata(appID string, update upstreamtypes.Update) error
+	UpdateAppVersionDemotion(appID, channelID, cursor string, isDemoted bool) error
 	HasStrictPreflights(appID string, sequence int64) (bool, error)
 	GetEmbeddedClusterConfigForVersion(appID string, sequence int64) (*embeddedclusterv1beta1.Config, error)
 }

pkg/store/store_interface.go

@@ -37,6 +37,9 @@ var jobs = make(map[string]*cron.Cron)
 var mtx sync.Mutex
 var store storepkg.Store
 
+// getUpdates is a package-level variable that can be replaced in tests for mocking
+var getUpdates = kotspull.GetUpdates
+
 // Start will start the update checker
 // the frequency of those update checks are app specific and can be modified by the user
 func Start() error {
@@ -251,7 +254,7 @@ func checkForKotsAppUpdates(opts types.CheckForUpdatesOpts, finishedChan chan<-
 	}
 
 	// get updates
-	updates, err := kotspull.GetUpdates(fmt.Sprintf("replicated://%s", latestLicense.Spec.AppSlug), getUpdatesOptions)
+	updates, err := getUpdates(fmt.Sprintf("replicated://%s", latestLicense.Spec.AppSlug), getUpdatesOptions)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get updates")
 	}
@@ -274,7 +277,7 @@ func checkForKotsAppUpdates(opts types.CheckForUpdatesOpts, finishedChan chan<-
 		return nil, errors.Errorf("no app versions found for app %s in downstream %s", opts.AppID, d.ClusterID)
 	}
 
-	if err := maybeUpdatePendingVersionsMetadata(a.ID, getUpdatesOptions, appVersions.CurrentVersion); err != nil {
+	if err := maybeUpdatePendingVersionsMetadata(a.ID, getUpdatesOptions, appVersions.CurrentVersion, appVersions.PendingVersions); err != nil {
 		logger.Error(errors.Wrap(err, "failed to update app version metadata"))
 	}
 
@@ -337,13 +340,18 @@ func checkForKotsAppUpdates(opts types.CheckForUpdatesOpts, finishedChan chan<-
 	return &ucr, nil
 }
 
+// getVersionKey builds a string given a chhanel ID and cursor to be used as a key for version lookup maps with the format <channelID>-<cursor>
+func getVersionKey(channelID, cursor string) string {
+	return fmt.Sprintf("%s-%s", channelID, cursor)
+}
+
 // maybeUpdatePendingVersionsMetadata updates metadata for pending versions since the currently deployed version.
 //
 // Limitations:
 // - Only gets pending releases for the channel of the currently deployed version, even if channel changed in later versions
 // - Does not rerender the application archive, so the Installation object in the archive can become out of sync
 // - This is not in the critical path - errors are logged but don't fail the overall update check
-func maybeUpdatePendingVersionsMetadata(appID string, getUpdatesOptions kotspull.GetUpdatesOptions, currentVersion *downstreamtypes.DownstreamVersion) error {
+func maybeUpdatePendingVersionsMetadata(appID string, getUpdatesOptions kotspull.GetUpdatesOptions, currentVersion *downstreamtypes.DownstreamVersion, pendingVersions []*downstreamtypes.DownstreamVersion) error {
 	if currentVersion == nil {
 		return nil
 	}
@@ -358,16 +366,51 @@ func maybeUpdatePendingVersionsMetadata(appID string, getUpdatesOptions kotspull
 		getUpdatesOptions.CurrentChannelName = ""
 	}
 
-	updates, err := kotspull.GetUpdates(fmt.Sprintf("replicated://%s", getUpdatesOptions.License.Spec.AppSlug), getUpdatesOptions)
+	updates, err := getUpdates(fmt.Sprintf("replicated://%s", getUpdatesOptions.License.Spec.AppSlug), getUpdatesOptions)
 	if err != nil {
 		return errors.Wrap(err, "get updates for metadata refresh")
 	}
 
+	// build a version map of the pending versions for fast lookup
+	pendingVersionsMap := make(map[string]*downstreamtypes.DownstreamVersion)
+	for _, pending := range pendingVersions {
+		// We want to prevent mistakenly updating app versions by:
+		// - only considering pending versions from the same channel as the currently deployed version
+		// - making sure the cursor is different than the currently deployed version cursor, which can be the same when changing channels.
+		if pending.ChannelID == currentVersion.ChannelID && pending.UpdateCursor != currentVersion.UpdateCursor {
+			key := getVersionKey(pending.ChannelID, pending.UpdateCursor)
+			pendingVersionsMap[key] = pending
+		}
+	}
+
+	// build a version map of the updates for fast lookup
+	updateVersionsMap := make(map[string]*upstreamtypes.Update)
 	for _, update := range updates.Updates {
+		// update the app version metadata with the upstream info
 		if err := store.UpdateAppVersionMetadata(appID, update); err != nil {
 			logger.Error(errors.Wrapf(err, "failed to update app version metadata for %s", update.VersionLabel))
 		}
+		key := getVersionKey(update.ChannelID, update.Cursor)
+		updateVersionsMap[key] = &update
+	}
+
+	// Determine versions to demote: pending but not in updates
+	for key, pending := range pendingVersionsMap {
+		if _, exists := updateVersionsMap[key]; !exists {
+			if err := store.UpdateAppVersionDemotion(appID, pending.ChannelID, pending.UpdateCursor, true); err != nil {
+				logger.Error(errors.Wrapf(err, "failed to update app version demotion state for %s", pending.VersionLabel))
+			}
+		}
 	}
+	// Determine versions to un-demote: updates but not in pending
+	for key, update := range updateVersionsMap {
+		if _, exists := pendingVersionsMap[key]; !exists {
+			if err := store.UpdateAppVersionDemotion(appID, update.ChannelID, update.Cursor, false); err != nil {
+				logger.Error(errors.Wrapf(err, "failed to update app version demotion state for %s", update.VersionLabel))
+			}
+		}
+	}
+
 	return nil
 }