func (s *SQL) AddSDBlob(sdHash string, sdBlobLength int, sdBlob SdBlob) error {
	if s.conn == nil {
		return errors.Err("not connected")
	}
	sdBlobID, err := s.insertBlob(sdHash, sdBlobLength, true)
	if err != nil {
		return err
	}
	streamID, err := s.insertStream(sdBlob.StreamHash, sdBlobID)
	if err != nil {
		return err
	}
	for _, contentBlob := range sdBlob.Blobs {
		if contentBlob.BlobHash == "" {
			continue
		}
		blobID, err := s.insertBlob(contentBlob.BlobHash, contentBlob.Length, false)
		if err != nil {
			return err
		}
		args := []interface{}{streamID, blobID, contentBlob.BlobNum}
		_, err = s.exec(
			"INSERT IGNORE INTO stream_blob (stream_id, blob_id, num) VALUES ("+qt.Qs(len(args))+")",
			args...,
		)
		if err != nil {
			return errors.Err(err)
		}
	}
	return nil
}
AddSDBlob insert the SD blob and all the content blobs. The content blobs are marked as "not stored",
but they are tracked so reflector knows what it is missing.
package db
import (
	"context"
	"database/sql"
	"fmt"
	"runtime"
	"strings"
	"time"
	"github.com/lbryio/lbry.go/v2/dht/bits"
	"github.com/lbryio/lbry.go/v2/extras/errors"
	qt "github.com/lbryio/lbry.go/v2/extras/query"
	"github.com/lbryio/lbry.go/v2/extras/stop"
	"github.com/lbryio/lbry.go/v2/stream"
	"github.com/go-sql-driver/mysql"
	_ "github.com/go-sql-driver/mysql"  
	log "github.com/sirupsen/logrus"
	"github.com/volatiletech/null"
	"go.uber.org/atomic"
)
type SdBlob struct {
	StreamName string `json:"stream_name"`
	Blobs      []struct {
		Length   int    `json:"length"`
		BlobNum  int    `json:"blob_num"`
		BlobHash string `json:"blob_hash,omitempty"`
		IV       string `json:"iv"`
	} `json:"blobs"`
	StreamType        string `json:"stream_type"`
	Key               string `json:"key"`
	SuggestedFileName string `json:"suggested_file_name"`
	StreamHash        string `json:"stream_hash"`
}
type trackAccess int
const (
	TrackAccessNone    trackAccess = iota  
	TrackAccessStreams                     
	TrackAccessBlobs                       
)
type SQL struct {
	conn *sql.DB
	TrackAccess trackAccess
	SoftDelete bool
	LogQueries bool
}
func (s SQL) logQuery(query string, args ...interface{}) {
	if !s.LogQueries {
		return
	}
	qStr, err := qt.InterpolateParams(query, args...)
	if err != nil {
		log.Errorln(err)
	} else {
		log.Debugln(qStr)
	}
}
func (s *SQL) Connect(dsn string) error {
	var err error
	dsn += "?parseTime=1&collation=utf8mb4_unicode_ci&interpolateParams=1"
	s.conn, err = sql.Open("mysql", dsn)
	if err != nil {
		return errors.Err(err)
	}
	s.conn.SetMaxIdleConns(12)
	return errors.Err(s.conn.Ping())
}
func (s *SQL) AddBlob(hash string, length int, isStored bool) error {
	if s.conn == nil {
		return errors.Err("not connected")
	}
	_, err := s.insertBlob(hash, length, isStored)
	return err
}
func (s *SQL) AddBlobs(hash []string) error {
	if s.conn == nil {
		return errors.Err("not connected")
	}
	batch := 10000
	totalBlobs := int64(len(hash))
	work := make(chan []string, 1000)
	stopper := stop.New()
	var totalInserted atomic.Int64
	start := time.Now()
	go func() {
		for i := 0; i < len(hash); i += batch {
			j := i + batch
			if j > len(hash) {
				j = len(hash)
			}
			work <- hash[i:j]
		}
		log.Infof("done loading %d hashes in the work queue", len(hash))
		close(work)
	}()
	for i := 0; i < runtime.NumCPU(); i++ {
		stopper.Add(1)
		go func(worker int) {
			log.Infof("starting worker %d", worker)
			defer stopper.Done()
			for hashes := range work {
				inserted := totalInserted.Load()
				remaining := totalBlobs - inserted
				if inserted > 0 {
					timePerBlob := time.Since(start).Microseconds() / inserted
					remainingTime := time.Duration(remaining*timePerBlob) * time.Microsecond
					log.Infof("[T%d] processing batch of %d items. ETA: %s", worker, len(hashes), remainingTime.String())
				}
				err := s.insertBlobs(hashes)  
				if err != nil {
					log.Errorf("error while inserting batch: %s", errors.FullTrace(err))
				}
				totalInserted.Add(int64(len(hashes)))
			}
		}(i)
	}
	stopper.Wait()
	return nil
}
func (s *SQL) insertBlobs(hashes []string) error {
	var (
		q string
	)
	dayAgo := time.Now().AddDate(0, 0, -1).Format("2006-01-02 15:04:05")
	q = "insert into blob_ (hash, is_stored, length, last_accessed_at) values "
	for _, hash := range hashes {
		q += fmt.Sprintf("('%s',1,%d,'%s'),", hash, stream.MaxBlobSize, dayAgo)
	}
	q = strings.TrimSuffix(q, ",")
	_, err := s.exec(q)
	if err != nil {
		return err
	}
	return nil
}
func (s *SQL) insertBlob(hash string, length int, isStored bool) (int64, error) {
	if length <= 0 {
		return 0, errors.Err("length must be positive")
	}
	var (
		q    string
		args []interface{}
	)
	if s.TrackAccess == TrackAccessBlobs {
		args = []interface{}{hash, isStored, length, time.Now()}
		q = "INSERT INTO blob_ (hash, is_stored, length, last_accessed_at) VALUES (" + qt.Qs(len(args)) + ") ON DUPLICATE KEY UPDATE is_stored = (is_stored or VALUES(is_stored)), last_accessed_at = VALUES(last_accessed_at)"
	} else {
		args = []interface{}{hash, isStored, length}
		q = "INSERT INTO blob_ (hash, is_stored, length) VALUES (" + qt.Qs(len(args)) + ") ON DUPLICATE KEY UPDATE is_stored = (is_stored or VALUES(is_stored))"
	}
	blobID, err := s.exec(q, args...)
	if err != nil {
		return 0, err
	}
	if blobID == 0 {
		err = s.conn.QueryRow("SELECT id FROM blob_ WHERE hash = ?", hash).Scan(&blobID)
		if err != nil {
			return 0, errors.Err(err)
		}
		if blobID == 0 {
			return 0, errors.Err("blob ID is 0 even after INSERTing and SELECTing")
		}
		if s.TrackAccess == TrackAccessBlobs {
			err := s.touchBlobs([]uint64{uint64(blobID)})
			if err != nil {
				return 0, errors.Err(err)
			}
		}
	}
	return blobID, nil
}
func (s *SQL) insertStream(hash string, sdBlobID int64) (int64, error) {
	var (
		q    string
		args []interface{}
	)
	if s.TrackAccess == TrackAccessStreams {
		args = []interface{}{hash, sdBlobID, time.Now()}
		q = "INSERT IGNORE INTO stream (hash, sd_blob_id, last_accessed_at) VALUES (" + qt.Qs(len(args)) + ")"
	} else {
		args = []interface{}{hash, sdBlobID}
		q = "INSERT IGNORE INTO stream (hash, sd_blob_id) VALUES (" + qt.Qs(len(args)) + ")"
	}
	streamID, err := s.exec(q, args...)
	if err != nil {
		return 0, errors.Err(err)
	}
	if streamID == 0 {
		err = s.conn.QueryRow("SELECT id FROM stream WHERE sd_blob_id = ?", sdBlobID).Scan(&streamID)
		if err != nil {
			return 0, errors.Err(err)
		}
		if streamID == 0 {
			return 0, errors.Err("stream ID is 0 even after INSERTing and SELECTing")
		}
		if s.TrackAccess == TrackAccessStreams {
			err := s.touchStreams([]uint64{uint64(streamID)})
			if err != nil {
				return 0, errors.Err(err)
			}
		}
	}
	return streamID, nil
}
func (s *SQL) HasBlob(hash string, touch bool) (bool, error) {
	exists, err := s.HasBlobs([]string{hash}, touch)
	if err != nil {
		return false, err
	}
	return exists[hash], nil
}
func (s *SQL) HasBlobs(hashes []string, touch bool) (map[string]bool, error) {
	exists, idsNeedingTouch, err := s.hasBlobs(hashes)
	if touch {
		if s.TrackAccess == TrackAccessBlobs {
			_ = s.touchBlobs(idsNeedingTouch)
		} else if s.TrackAccess == TrackAccessStreams {
			_ = s.touchStreams(idsNeedingTouch)
		}
	}
	return exists, err
}
func (s *SQL) touchBlobs(blobIDs []uint64) error {
	if len(blobIDs) == 0 {
		return nil
	}
	query := "UPDATE blob_ SET last_accessed_at = ? WHERE id IN (" + qt.Qs(len(blobIDs)) + ")"
	args := make([]interface{}, len(blobIDs)+1)
	args[0] = time.Now()
	for i := range blobIDs {
		args[i+1] = blobIDs[i]
	}
	startTime := time.Now()
	_, err := s.exec(query, args...)
	log.Debugf("touched %d blobs and took %s", len(blobIDs), time.Since(startTime))
	return errors.Err(err)
}
func (s *SQL) touchStreams(streamIDs []uint64) error {
	if len(streamIDs) == 0 {
		return nil
	}
	query := "UPDATE stream SET last_accessed_at = ? WHERE id IN (" + qt.Qs(len(streamIDs)) + ")"
	args := make([]interface{}, len(streamIDs)+1)
	args[0] = time.Now()
	for i := range streamIDs {
		args[i+1] = streamIDs[i]
	}
	startTime := time.Now()
	_, err := s.exec(query, args...)
	log.Debugf("touched %d streams and took %s", len(streamIDs), time.Since(startTime))
	return errors.Err(err)
}
func (s *SQL) hasBlobs(hashes []string) (map[string]bool, []uint64, error) {
	if s.conn == nil {
		return nil, nil, errors.Err("not connected")
	}
	var (
		hash           string
		blobID         uint64
		streamID       null.Uint64
		lastAccessedAt null.Time
	)
	var needsTouch []uint64
	exists := make(map[string]bool)
	touchDeadline := time.Now().Add(-6 * time.Hour)  
	maxBatchSize := 10000
	doneIndex := 0
	for len(hashes) > doneIndex {
		sliceEnd := doneIndex + maxBatchSize
		if sliceEnd > len(hashes) {
			sliceEnd = len(hashes)
		}
		log.Debugf("getting hashes[%d:%d] of %d", doneIndex, sliceEnd, len(hashes))
		batch := hashes[doneIndex:sliceEnd]
		var query string
		if s.TrackAccess == TrackAccessBlobs {
			query = `SELECT b.hash, b.id, NULL, b.last_accessed_at
FROM blob_ b
WHERE b.is_stored = 1 and b.hash IN (` + qt.Qs(len(batch)) + `)`
		} else if s.TrackAccess == TrackAccessStreams {
			query = `SELECT b.hash, b.id, s.id, s.last_accessed_at
FROM blob_ b
LEFT JOIN stream_blob sb ON b.id = sb.blob_id
INNER JOIN stream s on (sb.stream_id = s.id or s.sd_blob_id = b.id)
WHERE b.is_stored = 1 and b.hash IN (` + qt.Qs(len(batch)) + `)`
		} else {
			query = `SELECT b.hash, b.id, NULL, NULL
FROM blob_ b
WHERE b.is_stored = 1 and b.hash IN (` + qt.Qs(len(batch)) + `)`
		}
		args := make([]interface{}, len(batch))
		for i := range batch {
			args[i] = batch[i]
		}
		s.logQuery(query, args...)
		err := func() error {
			startTime := time.Now()
			rows, err := s.conn.Query(query, args...)
			log.Debugf("hashes query took %s", time.Since(startTime))
			if err != nil {
				return errors.Err(err)
			}
			defer closeRows(rows)
			for rows.Next() {
				err := rows.Scan(&hash, &blobID, &streamID, &lastAccessedAt)
				if err != nil {
					return errors.Err(err)
				}
				exists[hash] = true
				if !lastAccessedAt.Valid || lastAccessedAt.Time.Before(touchDeadline) {
					if s.TrackAccess == TrackAccessBlobs {
						needsTouch = append(needsTouch, blobID)
					} else if s.TrackAccess == TrackAccessStreams && !streamID.IsZero() {
						needsTouch = append(needsTouch, streamID.Uint64)
					}
				}
			}
			err = rows.Err()
			if err != nil {
				return errors.Err(err)
			}
			doneIndex += len(batch)
			return nil
		}()
		if err != nil {
			return nil, nil, err
		}
	}
	return exists, needsTouch, nil
}
func (s *SQL) Delete(hash string) error {
	if s.SoftDelete {
		_, err := s.exec("UPDATE blob_ SET is_stored = 0 WHERE hash = ?", hash)
		return errors.Err(err)
	}
	_, err := s.exec("DELETE FROM stream WHERE sd_blob_id = (SELECT id FROM blob_ WHERE hash = ?)", hash)
	if err != nil {
		return errors.Err(err)
	}
	_, err = s.exec("DELETE FROM blob_ WHERE hash = ?", hash)
	return errors.Err(err)
}
func (s *SQL) LeastRecentlyAccessedHashes(maxBlobs int) ([]string, error) {
	if s.conn == nil {
		return nil, errors.Err("not connected")
	}
	if s.TrackAccess != TrackAccessBlobs {
		return nil, errors.Err("blob access tracking is disabled")
	}
	query := "SELECT hash from blob_ where is_stored = 1 order by last_accessed_at limit ?"
	s.logQuery(query, maxBlobs)
	rows, err := s.conn.Query(query, maxBlobs)
	if err != nil {
		return nil, errors.Err(err)
	}
	defer closeRows(rows)
	blobs := make([]string, 0, maxBlobs)
	for rows.Next() {
		var hash string
		err := rows.Scan(&hash)
		if err != nil {
			return nil, errors.Err(err)
		}
		blobs = append(blobs, hash)
	}
	return blobs, nil
}
func (s *SQL) Count() (int, error) {
	if s.conn == nil {
		return 0, errors.Err("not connected")
	}
	query := "SELECT count(id) from blob_"
	if s.SoftDelete {
		query += " where is_stored = 1"
	}
	s.logQuery(query)
	var count int
	err := s.conn.QueryRow(query).Scan(&count)
	return count, errors.Err(err)
}
func (s *SQL) Block(hash string) error {
	query := "INSERT IGNORE INTO blocked SET hash = ?"
	args := []interface{}{hash}
	s.logQuery(query, args...)
	_, err := s.conn.Exec(query, args...)
	return errors.Err(err)
}
func (s *SQL) GetBlocked() (map[string]bool, error) {
	query := "SELECT hash FROM blocked"
	s.logQuery(query)
	rows, err := s.conn.Query(query)
	if err != nil {
		return nil, errors.Err(err)
	}
	defer closeRows(rows)
	blocked := make(map[string]bool)
	var hash string
	for rows.Next() {
		err := rows.Scan(&hash)
		if err != nil {
			return nil, errors.Err(err)
		}
		blocked[hash] = true
	}
	err = rows.Err()
	if err != nil {
		return nil, errors.Err(err)
	}
	return blocked, nil
}
func (s *SQL) MissingBlobsForKnownStream(sdHash string) ([]string, error) {
	if s.conn == nil {
		return nil, errors.Err("not connected")
	}
	query := `
		SELECT b.hash FROM blob_ b
		INNER JOIN stream_blob sb ON b.id = sb.blob_id
		INNER JOIN stream s ON s.id = sb.stream_id
		INNER JOIN blob_ sdb ON sdb.id = s.sd_blob_id AND sdb.hash = ?
		WHERE b.is_stored = 0
	`
	args := []interface{}{sdHash}
	s.logQuery(query, args...)
	rows, err := s.conn.Query(query, args...)
	if err != nil {
		return nil, errors.Err(err)
	}
	defer closeRows(rows)
	var missingBlobs []string
	var hash string
	for rows.Next() {
		err := rows.Scan(&hash)
		if err != nil {
			return nil, errors.Err(err)
		}
		missingBlobs = append(missingBlobs, hash)
	}
	err = rows.Err()
	if err != nil {
		return nil, errors.Err(err)
	}
	return missingBlobs, errors.Err(err)
}