minstrel/internal/acquire/memory.go

package acquire

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgtype"
	"github.com/jackc/pgx/v5/pgxpool"

	"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
	"git.fabledsword.com/bvandeusen/minstrel/internal/indexer"
	"git.fabledsword.com/bvandeusen/minstrel/internal/release"
)

// Block reasons, matching release_blocks_reason_check.
const (
	BlockOperator = "operator"
	BlockFiles    = "files"
	// BlockImport: the import found its files were not the album (#5522).
	BlockImport = "import"
	// BlockStalled: given up on because it would not finish (#5523); held
	// back until the block's expires_at.
	BlockStalled = "stalled"
)

// Grab origins, matching grabs_origin_check.
const (
	OriginSchedule = "schedule"
	OriginOperator = "operator"
)

// Memory is what searches and grabs leave behind: every candidate of a
// want's last search, the grabs, and the releases set aside.
type Memory struct {
	pool *pgxpool.Pool
}

// NewMemory returns a Memory over pool.
func NewMemory(pool *pgxpool.Pool) *Memory { return &Memory{pool: pool} }

// Keep replaces a want's candidates with this search's, in order.
func (m *Memory) Keep(ctx context.Context, wantID pgtype.UUID, cands []Candidate) error {
	return pgx.BeginFunc(ctx, m.pool, func(tx pgx.Tx) error {
		q := dbq.New(tx)
		if err := q.DeleteWantCandidates(ctx, wantID); err != nil {
			return err
		}
		for i, c := range cands {
			reasons, err := json.Marshal(c.Reasons)
			if err != nil {
				return err
			}
			if c.Reasons == nil {
				reasons = []byte("[]")
			}
			folded := c.FoldedFrom
			if folded == nil {
				folded = []string{}
			}
			if err := q.InsertWantCandidate(ctx, dbq.InsertWantCandidateParams{
				WantID: wantID, Key: Key(c), Name: c.Name, InfoHash: c.InfoHash, Indexer: c.Indexer,
				Source: c.Source.Source, SizeBytes: c.SizeBytes, Seeders: int32(c.Seeders),
				FilesCount: int32(c.FilesCount), Format: release.Describe(c.Release), Quality: string(c.Tier),
				Year: int32(c.Year), Verdict: string(c.Verdict), Reason: c.Reason, Reasons: reasons,
				FoldedFrom: folded, Rank: int32(i),
			}); err != nil {
				return err
			}
		}
		return nil
	})
}

// blocked is every release set aside, by infohash, with the sentence a
// candidate shows. A failed read blocks nothing: offering a release set
// aside costs one grab a person can undo, while blocking everything on a
// database blip would hide every result.
func (m *Memory) blocked(ctx context.Context) map[string]string {
	if m == nil {
		return nil
	}
	rows, err := dbq.New(m.pool).ListReleaseBlocks(ctx)
	if err != nil {
		return nil
	}
	out := make(map[string]string, len(rows))
	for _, r := range rows {
		out[indexer.NormalizeHash(r.InfoHash)] = BlockSentence(r.Reason, r.Detail)
	}
	return out
}

// BlockSentence is what a blocked candidate says.
func BlockSentence(reason, detail string) string {
	switch reason {
	case BlockFiles:
		if detail != "" {
			return "Set aside for what its files hold: " + detail
		}
		return "Set aside for what its files hold."
	case BlockImport:
		if detail != "" {
			return "Refused on import: " + detail
		}
		return "Refused on import: its files were not the album."
	case BlockStalled:
		if detail != "" {
			return "Given up on for now: " + detail
		}
		return "Given up on for now: it would not finish."
	}
	return "You set this release aside."
}

// Block sets a release aside and marks every kept candidate of it blocked.
func (m *Memory) Block(ctx context.Context, p dbq.BlockReleaseParams) error {
	p.InfoHash = indexer.NormalizeHash(p.InfoHash)
	if p.InfoHash == "" {
		return errors.New("a release without an infohash cannot be set aside")
	}
	return pgx.BeginFunc(ctx, m.pool, func(tx pgx.Tx) error {
		q := dbq.New(tx)
		if err := q.BlockRelease(ctx, p); err != nil {
			return err
		}
		return q.MarkCandidatesBlocked(ctx, dbq.MarkCandidatesBlockedParams{
			InfoHash: p.InfoHash, Reason: BlockSentence(p.Reason, p.Detail),
		})
	})
}

// Unblock lets a release be taken again. Its kept candidates wait for a
// person until the next search judges them afresh.
func (m *Memory) Unblock(ctx context.Context, hash string) (bool, error) {
	hash = indexer.NormalizeHash(hash)
	var n int64
	err := pgx.BeginFunc(ctx, m.pool, func(tx pgx.Tx) error {
		q := dbq.New(tx)
		var err error
		if n, err = q.UnblockRelease(ctx, hash); err != nil {
			return err
		}
		return q.MarkCandidatesUnblocked(ctx, dbq.MarkCandidatesUnblockedParams{
			InfoHash: hash, Reason: "Allowed again. The next search judges it afresh.",
		})
	})
	return n > 0, err
}

// LiveGrab is the download running for a want, if there is one.
func (m *Memory) LiveGrab(ctx context.Context, wantID pgtype.UUID) (dbq.Grab, bool, error) {
	g, err := dbq.New(m.pool).GetLiveGrab(ctx, wantID)
	if errors.Is(err, pgx.ErrNoRows) {
		return g, false, nil
	}
	if err != nil {
		return g, false, err
	}
	return g, true, nil
}

// Grabbed records a download started, ending the grab it replaced first so
// the one-live-grab index holds.
func (m *Memory) Grabbed(ctx context.Context, p dbq.InsertGrabParams, replaced pgtype.UUID, why string) (dbq.Grab, error) {
	var g dbq.Grab
	err := pgx.BeginFunc(ctx, m.pool, func(tx pgx.Tx) error {
		q := dbq.New(tx)
		if replaced.Valid {
			if err := q.EndGrab(ctx, dbq.EndGrabParams{ID: replaced, EndReason: "superseded", EndDetail: why}); err != nil {
				return err
			}
		}
		var err error
		g, err = q.InsertGrab(ctx, p)
		return err
	})
	if err != nil {
		return g, fmt.Errorf("record grab: %w", err)
	}
	return g, nil
}