more sql consolidation [CI SKIP]

db/helpers.go

@@ -238,6 +238,29 @@ func EnsureSnapshotTable(ctx context.Context, dbConn *sqlx.DB, tableName string)
 	return err
 }
 
+// BackfillSerialColumn sets missing values in a serial-like column for Postgres tables.
+func BackfillSerialColumn(ctx context.Context, dbConn *sqlx.DB, tableName, columnName string) error {
+	if err := ValidateTableName(tableName); err != nil {
+		return err
+	}
+	if columnName == "" {
+		return fmt.Errorf("column name is empty")
+	}
+	query := fmt.Sprintf(
+		`UPDATE %s SET "%s" = nextval(pg_get_serial_sequence('%s','%s')) WHERE "%s" IS NULL`,
+		tableName, columnName, tableName, columnName, columnName,
+	)
+	_, err := dbConn.ExecContext(ctx, query)
+	if err != nil {
+		errText := strings.ToLower(err.Error())
+		if strings.Contains(errText, "pg_get_serial_sequence") || strings.Contains(errText, "sequence") {
+			return nil
+		}
+		return err
+	}
+	return nil
+}
+
 // BuildDailySummaryInsert returns the SQL to aggregate hourly snapshots into a daily summary table.
 func BuildDailySummaryInsert(tableName string, unionQuery string) (string, error) {
 	if _, err := SafeTableName(tableName); err != nil {

internal/tasks/inventorySnapshots.go

@@ -604,15 +604,7 @@ func ensureSnapshotRowID(ctx context.Context, dbConn *sqlx.DB, tableName string)
 				return err
 			}
 		}
-		_, err = dbConn.ExecContext(ctx, fmt.Sprintf(
+		if err := db.BackfillSerialColumn(ctx, dbConn, tableName, "RowId"); err != nil {
-			`UPDATE %s SET "RowId" = nextval(pg_get_serial_sequence('%s','RowId')) WHERE "RowId" IS NULL`,
-			tableName, tableName,
-		))
-		if err != nil {
-			errText := strings.ToLower(err.Error())
-			if strings.Contains(errText, "pg_get_serial_sequence") || strings.Contains(errText, "sequence") {
-				return nil
-			}
 			return err
 		}
 	case "sqlite":