func GetPeersFromRows(rows *pgx.Rows) (peers []Peer, err error) {
	for rows.Next() {
		var (
			id       int32
			peerId   []byte
			ip       string
			port     int32
			infoHash []byte
			isIpV6   bool
		)

		err = rows.Scan(&id, &peerId, &ip, &port, &infoHash, &isIpV6)
		if err != nil {
			log.Print(err)
			return peers, err
		}

		peer := new(Peer)
		peer.Id = id
		peer.PeerId = peerId
		peer.Ip = ip
		peer.Port = port
		peer.InfoHash = infoHash
		peer.IsIpV6 = isIpV6

		peers = append(peers, *peer)
	}

	return peers, err
}

func (s *ValueStore) FindByKeys(dto *models.ValueDTO, mode string) (res []models.Value, err error) {
	if err := models.V.StructPartial(dto, "Keys"); err != nil {
		return nil, err
	}

	if _, exist := map[string]bool{
		"&&": true,
		"@>": true,
		"=":  true,
	}[mode]; !exist {
		mode = "="
	}

	model := models.NewValue()
	model.TransformFrom(dto)

	fields, modelValues := model.Fields()

	query := SqlSelect(model.TableName(), fields)
	// any - "&&", contains - "@>"", equal - "="
	if mode == "=" {
		query += fmt.Sprintf(" WHERE sort_text_array(keys) %s sort_text_array(?) AND is_removed = false", mode)
	} else {
		query += fmt.Sprintf(" WHERE keys %s ? AND is_removed = false", mode)
	}

	query = FormateToPQuery(query)

	var rows *pgx.Rows

	if dto.Tx != nil {
		rows, err = dto.Tx.Query(query, model.Keys)
	} else {
		rows, err = s.db.Query(query, model.Keys)
	}

	defer rows.Close()

	if err != nil {
		s.ErrorLog("action", "поиск записей по keys", "err", err, "keys", model.Keys)
		return
	}

	for rows.Next() {
		if err = rows.Scan(modelValues...); err != nil {
			s.ErrorLog("action", "поиск записей по keys", "subaction", "сканирование строки", "err", err, "keys", model.Keys)
			return
		}

		// res = append(res, *model)
		_model := models.NewValue()
		_model.TransformFrom(model)
		res = append(res, *_model)

		// TODO: reset for maps
		model.Value = models.InterfaceMap{}
	}

	return res, nil
}

func GetTorrentsFromRows(rows *pgx.Rows) (torrents []Torrent, err error) {
	for rows.Next() {
		var (
			info_hash  []byte
			complete   int32
			incomplete int32
		)

		err = rows.Scan(&info_hash, &incomplete, &complete)

		if err != nil {
			return torrents, err
		}

		torrent := new(Torrent)
		torrent.InfoHash = info_hash
		torrent.Complete = complete
		torrent.Incomplete = incomplete

		torrents = append(torrents, *torrent)
	}

	return torrents, err
}

func releaseList(rows *pgx.Rows) ([]*ct.Release, error) {
	var releases []*ct.Release
	for rows.Next() {
		release, err := scanRelease(rows)
		if err != nil {
			rows.Close()
			return nil, err
		}
		releases = append(releases, release)
	}
	return releases, rows.Err()
}

func scanFormations(rows *pgx.Rows) ([]*ct.Formation, error) {
	var formations []*ct.Formation
	for rows.Next() {
		formation, err := scanFormation(rows)
		if err != nil {
			rows.Close()
			return nil, err
		}
		formations = append(formations, formation)
	}
	return formations, rows.Err()
}

func (r *FormationRepo) listExpanded(rows *pgx.Rows) ([]*ct.ExpandedFormation, error) {
	defer rows.Close()

	var formations []*ct.ExpandedFormation

	// artifactIDs is a list of artifact IDs related to the formation list
	// and is used to populate the formation's artifact fields using a
	// subsequent artifact list query
	artifactIDs := make(map[string]struct{})

	for rows.Next() {
		formation, err := scanExpandedFormation(rows)
		if err != nil {
			return nil, err
		}
		formations = append(formations, formation)

		for _, id := range formation.Release.ArtifactIDs {
			artifactIDs[id] = struct{}{}
		}
	}

	if len(artifactIDs) > 0 {
		ids := make([]string, 0, len(artifactIDs))
		for id := range artifactIDs {
			ids = append(ids, id)
		}
		artifacts, err := r.artifacts.ListIDs(ids...)
		if err != nil {
			return nil, err
		}
		for _, formation := range formations {
			populateFormationArtifacts(formation, artifacts)
		}
	}

	return formations, rows.Err()
}

// returns true if a task has been processed
func fetch_task() (bool, error) {
	var err error
	var rows *pgx.Rows

	if rows, err = db.Query(fetchQuery, fetchQueryArgs...); err != nil {
		return false, fmt.Errorf("Could not query for tasks: ", err.Error())
	}

	if rows.Next() {
		var task_id int64
		var worker string
		var parameters []byte

		if err := rows.Scan(&task_id, &worker, &parameters); err != nil {
			rows.Close()
			return false, fmt.Errorf("Could not retrieve columns: ", err.Error())
		}

		rows.Close()

		log.Println("Starting worker:", worker)
		status, result := work(worker, parameters)
		log.Println("Result:", status)

		if _, err := db.Exec(`UPDATE `+quotedTableName+` SET status=$2,result=$3,end_time=now() WHERE task_id=$1`, task_id, status, result); err != nil {
			return false, fmt.Errorf("Could not update task: ", err.Error())
		}
	} else {
		rows.Close()
		return false, nil
	}

	return true, nil
}