<commit_before>// Copyright 2013 The Chihaya Authors. All rights reserved.
// Use of this source code is governed by the BSD 2-Clause license,
// which can be found in the LICENSE file.

// Package redis implements the storage interface for a BitTorrent tracker.
//
// The client whitelist is represented as a set with the key name "whitelist"
// with an optional prefix. Torrents and users are represented as hashes.
// Torrents' keys are named "torrent:<infohash>" with an optional prefix.
// Users' keys are named "user:<passkey>" with an optional prefix. The
// seeders and leechers attributes of torrent hashes are strings that represent
// the key for those hashes within redis. This is done because redis cannot
// nest their hash data type.
package redis

import (
	"errors"
	"strconv"
	"time"

	"github.com/garyburd/redigo/redis"

	"github.com/pushrax/chihaya/cache"
	"github.com/pushrax/chihaya/config"
	"github.com/pushrax/chihaya/models"
)

var (
	ErrCreateUser    = errors.New("redis: Incorrect reply length for user")
	ErrCreateTorrent = errors.New("redis: Incorrect reply length for torrent")
	ErrCreatePeer    = errors.New("redis: Incorrect reply length for peer")
	ErrMarkActive    = errors.New("redis: Torrent doesn't exist")

	SeedersPrefix  = "seeders:"
	LeechersPrefix = "leechers:"
	TorrentPrefix  = "torrent:"
	UserPrefix     = "user:"
	PeerPrefix     = "peer:"
)

type driver struct{}

func (d *driver) New(conf *config.DataStore) cache.Pool {
	return &Pool{
		conf: conf,
		pool: redis.Pool{
			MaxIdle:      conf.MaxIdleConns,
			IdleTimeout:  conf.IdleTimeout.Duration,
			Dial:         makeDialFunc(conf),
			TestOnBorrow: testOnBorrow,
		},
	}
}

func makeDialFunc(conf *config.DataStore) func() (redis.Conn, error) {
	return func() (conn redis.Conn, err error) {
		conn, err = redis.Dial(conf.Network, conf.Host+":"+conf.Port)
		if err != nil {
			return nil, err
		}
		return conn, nil
	}
}

func testOnBorrow(c redis.Conn, t time.Time) error {
	_, err := c.Do("PING")
	return err
}

type Pool struct {
	conf *config.DataStore
	pool redis.Pool
}

func (p *Pool) Close() error {
	return p.pool.Close()
}

func (p *Pool) Get() (cache.Tx, error) {
	retTx := &Tx{
		conf: p.conf,
		done: false,
		Conn: p.pool.Get(),
	}
	return retTx, nil
}

type Tx struct {
	conf *config.DataStore
	done bool
	redis.Conn
}

func (tx *Tx) close() {
	if tx.done {
		panic("redis: transaction closed twice")
	}
	tx.done = true
	tx.Conn.Close()
}

func createUser(userVals []string) (*models.User, error) {
	if len(userVals) != 7 {
		return nil, ErrCreateUser
	}
	var user models.User
	convErrors := make([]error, 7)
	user.ID, convErrors[0] = strconv.ParseUint(userVals[0], 10, 64)
	user.Passkey = userVals[1]
	user.UpMultiplier, convErrors[2] = strconv.ParseFloat(userVals[2], 64)
	user.DownMultiplier, convErrors[3] = strconv.ParseFloat(userVals[3], 64)
	user.Slots, convErrors[4] = strconv.ParseInt(userVals[4], 10, 64)
	user.SlotsUsed, convErrors[5] = strconv.ParseInt(userVals[5], 10, 64)
	user.Snatches, convErrors[6] = strconv.ParseUint(userVals[6], 10, 64)

	for i := 0; i < 7; i++ {
		if convErrors[i] != nil {
			return nil, convErrors[i]
		}
	}
	return &user, nil
}

// This is a multiple action command, it's not internally atomic
func (tx *Tx) createTorrent(torrentVals []string) (*models.Torrent, error) {
	if len(torrentVals) != 7 {
		return nil, ErrCreateTorrent
	}
	var torrent models.Torrent
	convErrors := make([]error, 9)
	torrent.ID, convErrors[0] = strconv.ParseUint(torrentVals[0], 10, 64)
	torrent.Infohash = torrentVals[1]
	torrent.Active, convErrors[2] = strconv.ParseBool(torrentVals[2])
	torrent.Snatches, convErrors[3] = strconv.ParseUint(torrentVals[3], 10, 32)
	torrent.UpMultiplier, convErrors[4] = strconv.ParseFloat(torrentVals[4], 64)
	torrent.DownMultiplier, convErrors[5] = strconv.ParseFloat(torrentVals[5], 64)
	torrent.LastAction, convErrors[6] = strconv.ParseInt(torrentVals[6], 10, 64)
	torrent.Seeders, convErrors[7] = tx.getPeers(torrent.ID, SeedersPrefix)
	torrent.Leechers, convErrors[8] = tx.getPeers(torrent.ID, LeechersPrefix)

	for i := 0; i < 9; i++ {
		if convErrors[i] != nil {
			return nil, convErrors[i]
		}
	}
	return &torrent, nil
}

// The peer hashkey relies on the combination of peerID, userID, and torrentID being unique
func (tx *Tx) setPeer(peer *models.Peer) error {
	hashKey := tx.conf.Prefix + getPeerHashKey(peer)
	_, err := tx.Do("HMSET", hashKey,
		"id", peer.ID,
		"user_id", peer.UserID,
		"torrent_id", peer.TorrentID,
		"ip", peer.IP,
		"port", peer.Port,
		"uploaded", peer.Uploaded,
		"downloaded", peer.Downloaded,
		"left", peer.Left,
		"last_announce", peer.LastAnnounce)

	if err != nil {
		return err
	}
	return nil
}

// Will not return an error if the peer doesn't exist
func (tx *Tx) removePeer(peer *models.Peer, peerTypePrefix string) error {
	setKey := tx.conf.Prefix + getPeerSetKey(peerTypePrefix, peer)
	_, err := tx.Do("SREM", setKey, getPeerHashKey(peer))
	if err != nil {
		return err
	}
	hashKey := tx.conf.Prefix + getPeerHashKey(peer)
	_, err = tx.Do("DEL", hashKey)
	return nil
}

// This is a multiple action command, it's not internally atomic
func (tx *Tx) removePeers(torrentID uint64, peers map[string]models.Peer, peerTypePrefix string) error {
	for _, peer := range peers {
		hashKey := tx.conf.Prefix + getPeerHashKey(&peer)
		_, err := tx.Do("DEL", hashKey)
		if err != nil {
			return err
		}
		delete(peers, models.PeerMapKey(&peer))
	}
	// Will only delete the set if all the peer deletions were successful
	setKey := tx.conf.Prefix + peerTypePrefix + strconv.FormatUint(torrentID, 36)
	_, err := tx.Do("DEL", setKey)
	if err != nil {
		return err
	}

	return nil
}

func getPeerHashKey(peer *models.Peer) string {
	return peer.ID + ":" + strconv.FormatUint(peer.UserID, 36) + ":" + strconv.FormatUint(peer.TorrentID, 36)
}

func getPeerSetKey(typePrefix string, peer *models.Peer) string {
	return typePrefix + strconv.FormatUint(peer.TorrentID, 36)
}

// This is a multiple action command, it's not internally atomic
func (tx *Tx) addPeers(peers map[string]models.Peer, peerTypePrefix string) error {
	for _, peer := range peers {
		setKey := tx.conf.Prefix + getPeerSetKey(peerTypePrefix, &peer)
		_, err := tx.Do("SADD", setKey, getPeerHashKey(&peer))
		if err != nil {
			return err
		}
		tx.setPeer(&peer)
	}
	return nil
}

func createPeer(peerVals []string) (*models.Peer, error) {
	if len(peerVals) != 9 {
		return nil, ErrCreatePeer
	}
	ID := peerVals[0]
	UserID, err := strconv.ParseUint(peerVals[1], 10, 64)
	if err != nil {
		return nil, err
	}
	TorrentID, err := strconv.ParseUint(peerVals[2], 10, 64)
	if err != nil {
		return nil, err
	}
	IP := peerVals[3]

	Port, err := strconv.ParseUint(peerVals[4], 10, 64)
	if err != nil {
		return nil, err
	}
	Uploaded, err := strconv.ParseUint(peerVals[5], 10, 64)
	if err != nil {
		return nil, err
	}
	Downloaded, err := strconv.ParseUint(peerVals[6], 10, 64)
	if err != nil {
		return nil, err
	}
	Left, err := strconv.ParseUint(peerVals[7], 10, 64)
	if err != nil {
		return nil, err
	}
	LastAnnounce, err := strconv.ParseInt(peerVals[8], 10, 64)
	if err != nil {
		return nil, err
	}
	return &models.Peer{ID: ID, UserID: UserID, TorrentID: TorrentID, IP: IP, Port: Port,
		Uploaded: Uploaded, Downloaded: Downloaded, Left: Left, LastAnnounce: LastAnnounce}, nil

}

// This is a multiple action command, it's not internally atomic
func (tx *Tx) getPeers(torrentID uint64, peerTypePrefix string) (peers map[string]models.Peer, err error) {
	peers = make(map[string]models.Peer)
	setKey := tx.conf.Prefix + peerTypePrefix + strconv.FormatUint(torrentID, 36)
	peerStrings, err := redis.Strings(tx.Do("SMEMBERS", setKey))
	if err != nil {
		return nil, err
	}
	// Keys map to peer objects stored in hashes
	for _, peerHashKey := range peerStrings {
		hashKey := tx.conf.Prefix + peerHashKey
		peerVals, err := redis.Strings(tx.Do("HVALS", hashKey))
		if err != nil {
			return nil, err
		}
		if len(peerVals) == 0 {
			continue
		}
		peer, err := createPeer(peerVals)
		if err != nil {
			return nil, err
		}
		peers[models.PeerMapKey(peer)] = *peer
	}
	return
}

// This is a multiple action command, it's not internally atomic
func (tx *Tx) AddTorrent(t *models.Torrent) error {
	hashkey := tx.conf.Prefix + TorrentPrefix + t.Infohash
	_, err := tx.Do("HMSET", hashkey,
		"id", t.ID,
		"infohash", t.Infohash,
		"active", t.Active,
		"snatches", t.Snatches,
		"up_multiplier", t.UpMultiplier,
		"down_multiplier", t.DownMultiplier,
		"last_action", t.LastAction)
	if err != nil {
		return err
	}

	err = tx.addPeers(t.Seeders, SeedersPrefix)
	if err != nil {
		return err
	}
	err = tx.addPeers(t.Leechers, LeechersPrefix)
	if err != nil {
		return err
	}
	return nil
}

// This is a multiple action command, it's not internally atomic
func (tx *Tx) RemoveTorrent(t *models.Torrent) error {
	hashkey := tx.conf.Prefix + TorrentPrefix + t.Infohash
	_, err := tx.Do("DEL", hashkey)
	if err != nil {
		return err
	}
	// Remove seeders and leechers as well
	err = tx.removePeers(t.ID, t.Seeders, SeedersPrefix)
	if err != nil {
		return err
	}
	err = tx.removePeers(t.ID, t.Leechers, LeechersPrefix)
	if err != nil {
		return err
	}
	return nil
}

func (tx *Tx) AddUser(u *models.User) error {
	hashkey := tx.conf.Prefix + UserPrefix + u.Passkey
	_, err := tx.Do("HMSET", hashkey,
		"id", u.ID,
		"passkey", u.<PASSWORD>key,
		"up_multiplier", u.UpMultiplier,
		"down_multiplier", u.DownMultiplier,
		"slots", u.Slots,
		"slots_used", u.SlotsUsed,
		"snatches", u.Snatches)
	if err != nil {
		return err
	}
	return nil
}

func (tx *Tx) RemoveUser(u *models.User) error {
	hashkey := tx.conf.Prefix + UserPrefix + u.Passkey
	_, err := tx.Do("DEL", hashkey)
	if err != nil {
		return err
	}
	return nil
}

func (tx *Tx) FindUser(passkey string) (*models.User, bool, error) {
	hashkey := tx.conf.Prefix + UserPrefix + passkey
	userStrings, err := redis.Strings(tx.Do("HVALS", hashkey))
	if err != nil {
		return nil, false, err
	} else if len(userStrings) == 0 {
		return nil, false, nil
	}
	foundUser, err := createUser(userStrings)
	if err != nil {
		return nil, false, err
	}
	return foundUser, true, nil
}

// This is a multiple action command, it's not internally atomic
func (tx *Tx) FindTorrent(infohash string) (*models.Torrent, bool, error) {
	hashkey := tx.conf.Prefix + TorrentPrefix + infohash
	torrentStrings, err := redis.Strings(tx.Do("HVALS", hashkey))
	if err != nil {
		return nil, false, err
	} else if len(torrentStrings) == 0 {
		return nil, false, nil
	}

	foundTorrent, err := tx.createTorrent(torrentStrings)
	if err != nil {
		return nil, false, err
	}
	return foundTorrent, true, nil
}

func (tx *Tx) ClientWhitelisted(peerID string) (exists bool, err error) {
	key := tx.conf.Prefix + "whitelist"
	return redis.Bool(tx.Do("SISMEMBER", key, peerID))
}

func (tx *Tx) WhitelistClient(peerID string) error {
	key := tx.conf.Prefix + "whitelist"
	_, err := tx.Do("SADD", key, peerID)
	return err
}

func (tx *Tx) UnWhitelistClient(peerID string) error {
	key := tx.conf.Prefix + "whitelist"
	_, err := tx.Do("SREM", key, peerID)
	return err
}

// This is a multiple action command, it's not internally atomic
func (tx *Tx) RecordSnatch(user *models.User, torrent *models.Torrent) error {

	torrentKey := tx.conf.Prefix + TorrentPrefix + torrent.Infohash
	snatchCount, err := redis.Int(tx.Do("HINCRBY", torrentKey, "snatches", 1))
	if err != nil {
		return err
	}
	torrent.Snatches = uint64(snatchCount)

	userKey := tx.conf.Prefix + UserPrefix + user.Passkey
	snatchCount, err = redis.Int(tx.Do("HINCRBY", userKey, "snatches", 1))
	if err != nil {
		return err
	}
	user.Snatches = uint64(snatchCount)
	return nil
}

func (tx *Tx) MarkActive(torrent *models.Torrent) error {
	hashkey := tx.conf.Prefix + TorrentPrefix + torrent.Infohash
	activeExists, err := redis.Int(tx.Do("HSET", hashkey, "active", true))
	if err != nil {
		return err
	}
	torrent.Active = true
	// HSET returns 1 if hash didn't exist before
	if activeExists == 1 {
		return ErrMarkActive
	}
	return nil
}

func (tx *Tx) MarkInactive(torrent *models.Torrent) error {
	hashkey := tx.conf.Prefix + TorrentPrefix + torrent.Infohash
	activeExists, err := redis.Int(tx.Do("HSET", hashkey, "active", false))
	if err != nil {
		return err
	}
	torrent.Active = false
	// HSET returns 1 if hash didn't exist before
	if activeExists == 1 {
		// Clean-up incomplete torrent
		_, err = tx.Do("DEL", hashkey)
		if err != nil {
			return err
		}
		return ErrMarkActive
	}
	return nil
}

// This is a multiple action command, it's not internally atomic
func (tx *Tx) AddLeecher(torrent *models.Torrent, peer *models.Peer) error {
	setKey := tx.conf.Prefix + LeechersPrefix + strconv.FormatUint(torrent.ID, 36)
	_, err := tx.Do("SADD", setKey, getPeerHashKey(peer))
	if err != nil {
		return err
	}
	err = tx.setPeer(peer)
	if err != nil {
		return err
	}
	if torrent.Leechers == nil {
		torrent.Leechers = make(map[string]models.Peer)
	}
	torrent.Leechers[models.PeerMapKey(peer)] = *peer
	return nil
}

// Setting assumes it is already a leecher, and just needs to be updated
// Maybe eventually there will be a move from leecher to seeder method
func (tx *Tx) SetLeecher(t *models.Torrent, p *models.Peer) error {
	err := tx.setPeer(p)
	if err != nil {
		return err
	}
	t.Leechers[models.PeerMapKey(p)] = *p
	return nil
}

func (tx *Tx) RemoveLeecher(t *models.Torrent, p *models.Peer) error {
	err := tx.removePeer(p, LeechersPrefix)
	if err != nil {
		return err
	}
	delete(t.Leechers, models.PeerMapKey(p))
	return nil
}

func (tx *Tx) LeecherFinished(torrent *models.Torrent, peer *models.Peer) error {
	torrentIdKey := strconv.FormatUint(torrent.ID, 36)
	seederSetKey := tx.conf.Prefix + SeedersPrefix + torrentIdKey
	leecherSetKey := tx.conf.Prefix + LeechersPrefix + torrentIdKey

	_, err := tx.Do("SMOVE", leecherSetKey, seederSetKey, getPeerHashKey(peer))
	if err != nil {
		return err
	}
	torrent.Seeders[models.PeerMapKey(peer)] = *peer
	delete(torrent.Leechers, models.PeerMapKey(peer))

	err = tx.setPeer(peer)
	return err
}

// This is a multiple action command, it's not internally atomic
func (tx *Tx) AddSeeder(torrent *models.Torrent, peer *models.Peer) error {
	setKey := tx.conf.Prefix + SeedersPrefix + strconv.FormatUint(torrent.ID, 36)
	_, err := tx.Do("SADD", setKey, getPeerHashKey(peer))
	if err != nil {
		return err
	}
	err = tx.setPeer(peer)
	if err != nil {
		return err
	}
	if torrent.Seeders == nil {
		torrent.Seeders = make(map[string]models.Peer)
	}
	torrent.Seeders[models.PeerMapKey(peer)] = *peer
	return nil
}

func (tx *Tx) SetSeeder(t *models.Torrent, p *models.Peer) error {
	err := tx.setPeer(p)
	if err != nil {
		return err
	}
	t.Seeders[models.PeerMapKey(p)] = *p
	return nil
}

func (tx *Tx) RemoveSeeder(t *models.Torrent, p *models.Peer) error {
	err := tx.removePeer(p, SeedersPrefix)
	if err != nil {
		return err
	}
	delete(t.Seeders, models.PeerMapKey(p))
	return nil
}

func (tx *Tx) IncrementSlots(u *models.User) error {
	hashkey := tx.conf.Prefix + UserPrefix + u.Passkey
	slotCount, err := redis.Int(tx.Do("HINCRBY", hashkey, "slots", 1))
	if err != nil {
		return err
	}
	u.Slots = int64(slotCount)
	return nil
}

func (tx *Tx) DecrementSlots(u *models.User) error {
	hashkey := tx.conf.Prefix + UserPrefix + u.Passkey
	slotCount, err := redis.Int(tx.Do("HINCRBY", hashkey, "slots", -1))
	if err != nil {
		return err
	}
	u.Slots = int64(slotCount)
	return nil
}

func init() {
	cache.Register("redis", &driver{})
}<commit_msg>Added documentation to redis package<commit_after>// Copyright 2013 The Chihaya Authors. All rights reserved.
// Use of this source code is governed by the BSD 2-Clause license,
// which can be found in the LICENSE file.

// Package redis implements the storage interface for a BitTorrent tracker.
//
// This interface is configured by a config.DataStore.
// To get a handle to this interface, call New on the initialized driver and
// then Get() on returned the cache.Pool.
//
// Torrents, Users, and Peers are all stored in Redis hash types. All Redis
// keys can have an optional prefix specified during configuration.
// The relationship between Torrents and Peers is a Redis set that holds
// the peers' keys. There are two sets per torrent, one for seeders and
// one for leechers. The Redis sets are keyed by type and the torrent's ID.
//
// The whitelist is a Redis set with the key "whitelist" that holds client IDs.
// Operations on the whitelist do not parse the client ID from a peer ID.
//
// Some functions in this interface are not atomic. The data being modified may
// change while the function is executing. This will not cause the function to
// return an error; instead the function will complete and return valid, stale
// data.
package redis

import (
	"errors"
	"strconv"
	"time"

	"github.com/garyburd/redigo/redis"

	"github.com/pushrax/chihaya/cache"
	"github.com/pushrax/chihaya/config"
	"github.com/pushrax/chihaya/models"
)

var (
	ErrCreateUser    = errors.New("redis: Incorrect reply length for user")
	ErrCreateTorrent = errors.New("redis: Incorrect reply length for torrent")
	ErrCreatePeer    = errors.New("redis: Incorrect reply length for peer")
	ErrMarkActive    = errors.New("redis: Torrent doesn't exist")

	SeedersPrefix  = "seeders:"
	LeechersPrefix = "leechers:"
	TorrentPrefix  = "torrent:"
	UserPrefix     = "user:"
	PeerPrefix     = "peer:"
)

type driver struct{}

// New creates and returns a cache.Pool.
func (d *driver) New(conf *config.DataStore) cache.Pool {
	return &Pool{
		conf: conf,
		pool: redis.Pool{
			MaxIdle:      conf.MaxIdleConns,
			IdleTimeout:  conf.IdleTimeout.Duration,
			Dial:         makeDialFunc(conf),
			TestOnBorrow: testOnBorrow,
		},
	}
}

// makeDialFunc configures and returns a new redis.Dial struct using the specified configuration.
func makeDialFunc(conf *config.DataStore) func() (redis.Conn, error) {
	return func() (conn redis.Conn, err error) {
		conn, err = redis.Dial(conf.Network, conf.Host+":"+conf.Port)
		if err != nil {
			return nil, err
		}
		return conn, nil
	}
}

// testOnBorrow pings the Redis instance
func testOnBorrow(c redis.Conn, t time.Time) error {
	_, err := c.Do("PING")
	return err
}

type Pool struct {
	conf *config.DataStore
	pool redis.Pool
}

func (p *Pool) Close() error {
	return p.pool.Close()
}

func (p *Pool) Get() (cache.Tx, error) {
	retTx := &Tx{
		conf: p.conf,
		done: false,
		Conn: p.pool.Get(),
	}
	return retTx, nil
}

type Tx struct {
	conf *config.DataStore
	done bool
	redis.Conn
}

func (tx *Tx) close() {
	if tx.done {
		panic("redis: transaction closed twice")
	}
	tx.done = true
	tx.Conn.Close()
}

// createUser takes a slice of length 7 and returns a pointer to a new models.User or an error.
// This function is used to create a user from a Redis hash response.
// The 7 strings in the slice must be in the specified order.
//
// User.ID
// User.Passkey
// User.UpMultiplier
// User.DownMultiplier
// User.Slots
// User.SlotsUsed
// User.Snatches
//
// If the strings cannot be converted to the correct type,
// createUser will return a nil user and the conversion error.
func createUser(userVals []string) (*models.User, error) {
	if len(userVals) != 7 {
		return nil, ErrCreateUser
	}
	var user models.User
	convErrors := make([]error, 7)
	user.ID, convErrors[0] = strconv.ParseUint(userVals[0], 10, 64)
	user.Passkey = userVals[1]
	user.UpMultiplier, convErrors[2] = strconv.ParseFloat(userVals[2], 64)
	user.DownMultiplier, convErrors[3] = strconv.ParseFloat(userVals[3], 64)
	user.Slots, convErrors[4] = strconv.ParseInt(userVals[4], 10, 64)
	user.SlotsUsed, convErrors[5] = strconv.ParseInt(userVals[5], 10, 64)
	user.Snatches, convErrors[6] = strconv.ParseUint(userVals[6], 10, 64)

	for i := 0; i < 7; i++ {
		if convErrors[i] != nil {
			return nil, convErrors[i]
		}
	}
	return &user, nil
}

// createTorrent takes a slice of length 7 and returns a pointer to a new models.Torrent or an error.
// This function can be used to create a torrent from a Redis hash response.
// The 7 hash fields must be in the specified order.
//
// torrent.ID
// torrent.Infohash
// torrent.Active
// torrent.Snatches
// torrent.UpMultiplier
// torrent.DownMultiplier
// torrent.LastAction
//
// This function calls multiple redis commands, it's not internally atomic.
// After converting the torrent's ID, the seeders and leechers are populated by getPeers
// If the strings cannot be converted to the correct type,
// createTorrent will return a nil user and the conversion error.
func (tx *Tx) createTorrent(torrentVals []string) (*models.Torrent, error) {
	if len(torrentVals) != 7 {
		return nil, ErrCreateTorrent
	}
	var torrent models.Torrent
	convErrors := make([]error, 9)
	torrent.ID, convErrors[0] = strconv.ParseUint(torrentVals[0], 10, 64)
	torrent.Infohash = torrentVals[1]
	torrent.Active, convErrors[2] = strconv.ParseBool(torrentVals[2])
	torrent.Snatches, convErrors[3] = strconv.ParseUint(torrentVals[3], 10, 32)
	torrent.UpMultiplier, convErrors[4] = strconv.ParseFloat(torrentVals[4], 64)
	torrent.DownMultiplier, convErrors[5] = strconv.ParseFloat(torrentVals[5], 64)
	torrent.LastAction, convErrors[6] = strconv.ParseInt(torrentVals[6], 10, 64)
	torrent.Seeders, convErrors[7] = tx.getPeers(torrent.ID, SeedersPrefix)
	torrent.Leechers, convErrors[8] = tx.getPeers(torrent.ID, LeechersPrefix)

	for i := 0; i < 9; i++ {
		if convErrors[i] != nil {
			return nil, convErrors[i]
		}
	}
	return &torrent, nil
}

// setPeer writes or overwrites peer information.
// The hash fields are sent in a specific order
// so that they can be unpacked correctly
func (tx *Tx) setPeer(peer *models.Peer) error {
	hashKey := tx.conf.Prefix + getPeerHashKey(peer)
	_, err := tx.Do("HMSET", hashKey,
		"id", peer.ID,
		"user_id", peer.UserID,
		"torrent_id", peer.TorrentID,
		"ip", peer.IP,
		"port", peer.Port,
		"uploaded", peer.Uploaded,
		"downloaded", peer.Downloaded,
		"left", peer.Left,
		"last_announce", peer.LastAnnounce)

	return err
}

// removePeer removes the given peer from the specified peer set (seeder or leecher),
// and removes the peer information.
// This function calls multiple redis commands, it's not internally atomic.
// This function will not return an error if the peer to remove doesn't exist.
func (tx *Tx) removePeer(peer *models.Peer, peerTypePrefix string) error {
	setKey := tx.conf.Prefix + getPeerSetKey(peerTypePrefix, peer)
	_, err := tx.Do("SREM", setKey, getPeerHashKey(peer))
	if err != nil {
		return err
	}
	hashKey := tx.conf.Prefix + getPeerHashKey(peer)
	_, err = tx.Do("DEL", hashKey)
	return nil
}

// removePeers removes all peers from specified peer set (seeders or leechers),
// removes the peer information, and then removes the associated peer from the given map.
// This function will not return an error if the peer to remove doesn't exist.
// This function will only delete the peer set if all the individual peer deletions were successful
// This function calls multiple redis commands, it's not internally atomic.
func (tx *Tx) removePeers(torrentID uint64, peers map[string]models.Peer, peerTypePrefix string) error {
	for _, peer := range peers {
		hashKey := tx.conf.Prefix + getPeerHashKey(&peer)
		_, err := tx.Do("DEL", hashKey)
		if err != nil {
			return err
		}
		delete(peers, models.PeerMapKey(&peer))
	}
	setKey := tx.conf.Prefix + peerTypePrefix + strconv.FormatUint(torrentID, 36)
	_, err := tx.Do("DEL", setKey)
	if err != nil {
		return err
	}
	return nil
}

// getPeerHashKey returns a string with the peer.ID, encoded peer.UserID, and encoded peer.TorrentID,
// concatenated and delimited by colons
// This key corresponds to a Redis hash type with fields containing a peer's data.
// The peer hashkey relies on the combination of peerID, userID, and torrentID being unique.
func getPeerHashKey(peer *models.Peer) string {
	return peer.ID + ":" + strconv.FormatUint(peer.UserID, 36) + ":" + strconv.FormatUint(peer.TorrentID, 36)
}

// getPeerSetKey returns a string that is the peer's encoded torrentID appended to the typePrefix
// This key corresponds to a torrent's pool of leechers or seeders
func getPeerSetKey(typePrefix string, peer *models.Peer) string {
	return typePrefix + strconv.FormatUint(peer.TorrentID, 36)
}

// addPeers adds each peer's key to the specified peer set and saves the peer's information.
// This function will not return an error if the peer already exists in the set.
// This function calls multiple redis commands, it's not internally atomic.
func (tx *Tx) addPeers(peers map[string]models.Peer, peerTypePrefix string) error {
	for _, peer := range peers {
		setKey := tx.conf.Prefix + getPeerSetKey(peerTypePrefix, &peer)
		_, err := tx.Do("SADD", setKey, getPeerHashKey(&peer))
		if err != nil {
			return err
		}
		tx.setPeer(&peer)
	}
	return nil
}

// createPeer takes a slice of length 9 and returns a pointer to a new models.Peer or an error.
// This function is used to create a peer from a Redis hash response.
// The 9 strings in the slice must be in the specified order.
//
// peer.ID
// peer.UserID
// peer.TorrentID
// peer.IP
// peer.Port
// peer.Uploaded
// peer.Downloaded
// peer.Left
// peer.LastAnnounce
//
// If the strings cannot be converted to the correct type,
// This function will return a nil peer and the conversion error.
func createPeer(peerVals []string) (*models.Peer, error) {
	if len(peerVals) != 9 {
		return nil, ErrCreatePeer
	}
	var peer models.Peer
	convErrors := make([]error, 9)
	peer.ID = peerVals[0]
	peer.UserID, convErrors[1] = strconv.ParseUint(peerVals[1], 10, 64)
	peer.TorrentID, convErrors[2] = strconv.ParseUint(peerVals[2], 10, 64)
	peer.IP = peerVals[3]
	peer.Port, convErrors[4] = strconv.ParseUint(peerVals[4], 10, 64)
	peer.Uploaded, convErrors[5] = strconv.ParseUint(peerVals[5], 10, 64)
	peer.Downloaded, convErrors[6] = strconv.ParseUint(peerVals[6], 10, 64)
	peer.Left, convErrors[7] = strconv.ParseUint(peerVals[7], 10, 64)
	peer.LastAnnounce, convErrors[8] = strconv.ParseInt(peerVals[8], 10, 64)

	for i := 0; i < 9; i++ {
		if convErrors[i] != nil {
			return nil, convErrors[i]
		}
	}
	return &peer, nil
}

// getPeers returns a map of peers from a specified torrent's peer set(seeders or leechers).
// This is a multiple action command, it's not internally atomic
func (tx *Tx) getPeers(torrentID uint64, peerTypePrefix string) (peers map[string]models.Peer, err error) {
	peers = make(map[string]models.Peer)
	setKey := tx.conf.Prefix + peerTypePrefix + strconv.FormatUint(torrentID, 36)
	peerStrings, err := redis.Strings(tx.Do("SMEMBERS", setKey))
	if err != nil {
		return nil, err
	}
	// Keys map to peer objects stored in hashes
	for _, peerHashKey := range peerStrings {
		hashKey := tx.conf.Prefix + peerHashKey
		peerVals, err := redis.Strings(tx.Do("HVALS", hashKey))
		if err != nil {
			return nil, err
		}
		if len(peerVals) == 0 {
			continue
		}
		peer, err := createPeer(peerVals)
		if err != nil {
			return nil, err
		}
		peers[models.PeerMapKey(peer)] = *peer
	}
	return
}

// AddTorrent writes/overwrites torrent information and saves peers from both peer sets.
// The hash fields are sent in a specific order
// so that they can be unpacked correctly
// This is a multiple action command, it's not internally atomic
func (tx *Tx) AddTorrent(t *models.Torrent) error {
	hashkey := tx.conf.Prefix + TorrentPrefix + t.Infohash
	_, err := tx.Do("HMSET", hashkey,
		"id", t.ID,
		"infohash", t.Infohash,
		"active", t.Active,
		"snatches", t.Snatches,
		"up_multiplier", t.UpMultiplier,
		"down_multiplier", t.DownMultiplier,
		"last_action", t.LastAction)
	if err != nil {
		return err
	}

	err = tx.addPeers(t.Seeders, SeedersPrefix)
	if err != nil {
		return err
	}
	err = tx.addPeers(t.Leechers, LeechersPrefix)
	if err != nil {
		return err
	}
	return nil
}

// RemoveTorrent deletes the torrent's Redis hash and then deletes all peers.
// This function will not return an error if the torrent has already been removed.
// This is a multiple action command, it's not internally atomic.
func (tx *Tx) RemoveTorrent(t *models.Torrent) error {
	hashkey := tx.conf.Prefix + TorrentPrefix + t.Infohash
	_, err := tx.Do("DEL", hashkey)
	if err != nil {
		return err
	}
	// Remove seeders and leechers as well
	err = tx.removePeers(t.ID, t.Seeders, SeedersPrefix)
	if err != nil {
		return err
	}
	err = tx.removePeers(t.ID, t.Leechers, LeechersPrefix)
	if err != nil {
		return err
	}
	return nil
}

// AddUser writes/overwrites user information to a Redis hash.
// The hash fields are sent in a specific order
// so that they can be unpacked correctly
func (tx *Tx) AddUser(u *models.User) error {
	hashkey := tx.conf.Prefix + UserPrefix + u.Passkey
	_, err := tx.Do("HMSET", hashkey,
		"id", u.ID,
		"passkey", u.Passkey,
		"up_multiplier", u.UpMultiplier,
		"down_multiplier", u.DownMultiplier,
		"slots", u.Slots,
		"slots_used", u.SlotsUsed,
		"snatches", u.Snatches)
	if err != nil {
		return err
	}
	return nil
}

// RemoveUser removes the user's hash from Redis.
// This function does not return an error if the user doesn't exist.
func (tx *Tx) RemoveUser(u *models.User) error {
	hashkey := tx.conf.Prefix + UserPrefix + u.Passkey
	_, err := tx.Do("DEL", hashkey)
	if err != nil {
		return err
	}
	return nil
}

// FindUser returns true and a pointer to a new user struct, if the user exists
// or nil and false if the user doesn't exist.
// This function does not return an error if the torrent doesn't exist.
func (tx *Tx) FindUser(passkey string) (*models.User, bool, error) {
	hashkey := tx.conf.Prefix + UserPrefix + passkey
	// Consider using HGETALL instead of HVALS here for robustness
	userStrings, err := redis.Strings(tx.Do("HVALS", hashkey))
	if err != nil {
		return nil, false, err
	} else if len(userStrings) == 0 {
		return nil, false, nil
	}
	foundUser, err := createUser(userStrings)
	if err != nil {
		return nil, false, err
	}
	return foundUser, true, nil
}

// FindTorrent returns a pointer to a new torrent struct and true, if the torrent exists
// or nil and false if the torrent doesn't exist.
// This is a multiple action command, it's not internally atomic
func (tx *Tx) FindTorrent(infohash string) (*models.Torrent, bool, error) {
	hashkey := tx.conf.Prefix + TorrentPrefix + infohash
	torrentStrings, err := redis.Strings(tx.Do("HVALS", hashkey))
	if err != nil {
		return nil, false, err
	} else if len(torrentStrings) == 0 {
		return nil, false, nil
	}

	foundTorrent, err := tx.createTorrent(torrentStrings)
	if err != nil {
		return nil, false, err
	}
	return foundTorrent, true, nil
}

// ClientWhitelisted returns true if the ClientID exists in the Client set.
// This function does not parse the client ID from the peer ID.
// The clientID must match exactly to a member of the set.
func (tx *Tx) ClientWhitelisted(peerID string) (exists bool, err error) {
	key := tx.conf.Prefix + "whitelist"
	return redis.Bool(tx.Do("SISMEMBER", key, peerID))
}

// WhitelistClient adds a client ID to the client whitelist set.
// This function does not return an error if the client ID is already in the set.
func (tx *Tx) WhitelistClient(peerID string) error {
	key := tx.conf.Prefix + "whitelist"
	_, err := tx.Do("SADD", key, peerID)
	return err
}

// UnWhitelistClient removes a client ID from the client whitelist set
// This function does not return an error if the client ID is not in the set.
func (tx *Tx) UnWhitelistClient(peerID string) error {
	key := tx.conf.Prefix + "whitelist"
	_, err := tx.Do("SREM", key, peerID)
	return err
}

// RecordSnatch increments the snatch counter on the torrent and user by one.
// This modifies the arguments as well as the hash field in Redis.
// This is a multiple action command, it's not internally atomic
func (tx *Tx) RecordSnatch(user *models.User, torrent *models.Torrent) error {

	torrentKey := tx.conf.Prefix + TorrentPrefix + torrent.Infohash
	snatchCount, err := redis.Int(tx.Do("HINCRBY", torrentKey, "snatches", 1))
	if err != nil {
		return err
	}
	torrent.Snatches = uint64(snatchCount)

	userKey := tx.conf.Prefix + UserPrefix + user.Passkey
	snatchCount, err = redis.Int(tx.Do("HINCRBY", userKey, "snatches", 1))
	if err != nil {
		return err
	}
	user.Snatches = uint64(snatchCount)
	return nil
}

// MarkActive sets the active field of the torrent to true.
// This modifies the argument as well as the hash field in Redis.
// This function will return ErrMarkActive if the torrent does not exist.
func (tx *Tx) MarkActive(torrent *models.Torrent) error {
	hashkey := tx.conf.Prefix + TorrentPrefix + torrent.Infohash
	activeExists, err := redis.Int(tx.Do("HSET", hashkey, "active", true))
	if err != nil {
		return err
	}
	torrent.Active = true
	// HSET returns 1 if hash didn't exist before
	if activeExists == 1 {
		return ErrMarkActive
	}
	return nil
}

// MarkInactive sets the active field of the torrent to false.
// This modifies the argument as well as the hash field in Redis.
// This function will return ErrMarkActive if the torrent does not exist.
func (tx *Tx) MarkInactive(torrent *models.Torrent) error {
	hashkey := tx.conf.Prefix + TorrentPrefix + torrent.Infohash
	activeExists, err := redis.Int(tx.Do("HSET", hashkey, "active", false))
	if err != nil {
		return err
	}
	torrent.Active = false
	// HSET returns 1 if hash didn't exist before
	if activeExists == 1 {
		// Clean-up incomplete torrent
		_, err = tx.Do("DEL", hashkey)
		if err != nil {
			return err
		}
		return ErrMarkActive
	}
	return nil
}

// AddLeecher adds a new peer to a torrent's leecher set.
// This modifies the torrent argument, as well as the torrent's set and peer's hash in Redis.
// This function does not return an error if the leecher already exists.
// This is a multiple action command, it's not internally atomic
func (tx *Tx) AddLeecher(torrent *models.Torrent, peer *models.Peer) error {
	setKey := tx.conf.Prefix + LeechersPrefix + strconv.FormatUint(torrent.ID, 36)
	_, err := tx.Do("SADD", setKey, getPeerHashKey(peer))
	if err != nil {
		return err
	}
	err = tx.setPeer(peer)
	if err != nil {
		return err
	}
	if torrent.Leechers == nil {
		torrent.Leechers = make(map[string]models.Peer)
	}
	torrent.Leechers[models.PeerMapKey(peer)] = *peer
	return nil
}

// SetLeecher updates a torrent's leecher.
// This modifies the torrent argument, as well as the peer's hash in Redis.
// Setting assumes that the peer is already a leecher, and only needs to be updated.
// This function does not return an error if the leecher does not exist or is not in the torrent's leecher set.
func (tx *Tx) SetLeecher(t *models.Torrent, p *models.Peer) error {
	err := tx.setPeer(p)
	if err != nil {
		return err
	}
	t.Leechers[models.PeerMapKey(p)] = *p
	return nil
}

// RemoveLeecher removes the given peer from a torrent's leecher set.
// This modifies the torrent argument, as well as the torrent's set and peer's hash in Redis.
// This function does not return an error if the peer doesn't exist, or is not in the set.
func (tx *Tx) RemoveLeecher(t *models.Torrent, p *models.Peer) error {
	err := tx.removePeer(p, LeechersPrefix)
	if err != nil {
		return err
	}
	delete(t.Leechers, models.PeerMapKey(p))
	return nil
}

// LeecherFinished moves a peer's hashkey from a torrent's leecher set to the seeder set and updates the peer.
// This modifies the torrent argument, as well as the torrent's set and peer's hash in Redis.
// This function does not return an error if the peer doesn't exist or is not in the torrent's leecher set.
func (tx *Tx) LeecherFinished(torrent *models.Torrent, peer *models.Peer) error {
	torrentIdKey := strconv.FormatUint(torrent.ID, 36)
	seederSetKey := tx.conf.Prefix + SeedersPrefix + torrentIdKey
	leecherSetKey := tx.conf.Prefix + LeechersPrefix + torrentIdKey

	_, err := tx.Do("SMOVE", leecherSetKey, seederSetKey, getPeerHashKey(peer))
	if err != nil {
		return err
	}
	torrent.Seeders[models.PeerMapKey(peer)] = *peer
	delete(torrent.Leechers, models.PeerMapKey(peer))

	err = tx.setPeer(peer)
	return err
}

// AddSeeder adds a new peer to a torrent's seeder set.
// This modifies the torrent argument, as well as the torrent's set and peer's hash in Redis.
// This function does not return an error if the seeder already exists.
// This is a multiple action command, it's not internally atomic
func (tx *Tx) AddSeeder(torrent *models.Torrent, peer *models.Peer) error {
	setKey := tx.conf.Prefix + SeedersPrefix + strconv.FormatUint(torrent.ID, 36)
	_, err := tx.Do("SADD", setKey, getPeerHashKey(peer))
	if err != nil {
		return err
	}
	err = tx.setPeer(peer)
	if err != nil {
		return err
	}
	if torrent.Seeders == nil {
		torrent.Seeders = make(map[string]models.Peer)
	}
	torrent.Seeders[models.PeerMapKey(peer)] = *peer
	return nil
}

// SetSeeder updates a torrent's seeder.
// This modifies the torrent argument, as well as the peer's hash in Redis.
// Setting assumes that the peer is already a seeder, and only needs to be updated.
// This function does not return an error if the seeder does not exist or is not in the torrent's seeder set.
func (tx *Tx) SetSeeder(t *models.Torrent, p *models.Peer) error {
	err := tx.setPeer(p)
	if err != nil {
		return err
	}
	t.Seeders[models.PeerMapKey(p)] = *p
	return nil
}

// RemoveSeeder removes the given peer from a torrent's seeder set.
// This modifies the torrent argument, as well as the torrent's set and peer's hash in Redis.
// This function does not return an error if the peer doesn't exist, or is not in the set.
func (tx *Tx) RemoveSeeder(t *models.Torrent, p *models.Peer) error {
	err := tx.removePeer(p, SeedersPrefix)
	if err != nil {
		return err
	}
	delete(t.Seeders, models.PeerMapKey(p))
	return nil
}

// IncrementSlots increment a user's Slots by one.
// This function modifies the argument as well as the hash field in Redis.
func (tx *Tx) IncrementSlots(u *models.User) error {
	hashkey := tx.conf.Prefix + UserPrefix + u.Passkey
	slotCount, err := redis.Int(tx.Do("HINCRBY", hashkey, "slots", 1))
	if err != nil {
		return err
	}
	u.Slots = int64(slotCount)
	return nil
}

// IncrementSlots increment a user's Slots by one.
// This function modifies the argument as well as the hash field in Redis.
func (tx *Tx) DecrementSlots(u *models.User) error {
	hashkey := tx.conf.Prefix + UserPrefix + u.Passkey
	slotCount, err := redis.Int(tx.Do("HINCRBY", hashkey, "slots", -1))
	if err != nil {
		return err
	}
	u.Slots = int64(slotCount)
	return nil
}

// init registers the redis driver
func init() {
	cache.Register("redis", &driver{})
}
{'fraction_non_alphanumeric': 0.12313403774760508, 'fraction_numerical': 0.008411433318621979, 'mean_word_length': 3.9226837060702877, 'pattern_counts': {'":': 6, '<': 11, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 9, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4078396', 'n_tokens_mistral': 14168, 'n_tokens_neox': 13409, 'n_words': 4292}