mrleo1nid
64dd8ed05e
ci / module boundaries / boundaries (push) Successful in 7s
ci / contract / contract (сгенерированный код совпадает с openapi.yaml) (push) Successful in 55s
ci / ts / ts (packages + web) (push) Failing after 4m4s
ci / api / api (push) Successful in 55s
ci / db / db (push) Failing after 1m14s
ci / mobile / mobile (Metro-бандл Android) (push) Successful in 1m11s
631 lines
23 KiB
Go
631 lines
23 KiB
Go
package uploads
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
// ErrNotFound — файла (пачки) нет.
|
|
var ErrNotFound = errors.New("uploads: not found")
|
|
|
|
// ErrWrongStatus — переход из текущего статуса не разрешён (гонка или повторное решение).
|
|
var ErrWrongStatus = errors.New("uploads: wrong status")
|
|
|
|
// Store — таблицы очереди и чтение каталога для проверки дублей.
|
|
type Store struct {
|
|
pool *pgxpool.Pool
|
|
}
|
|
|
|
// NewStore создаёт Store.
|
|
func NewStore(pool *pgxpool.Pool) *Store { return &Store{pool: pool} }
|
|
|
|
func newID() uuid.UUID { return uuid.Must(uuid.NewV7()) }
|
|
|
|
// --- Настройки ---
|
|
|
|
const settingsKey = "uploads.settings"
|
|
|
|
// Settings — текущие настройки очереди (по умолчанию, если не сохранены).
|
|
func (st *Store) Settings(ctx context.Context) (Settings, error) {
|
|
var raw []byte
|
|
err := st.pool.QueryRow(ctx, `SELECT value FROM app_settings WHERE key = $1`, settingsKey).Scan(&raw)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return DefaultSettings, nil
|
|
}
|
|
if err != nil {
|
|
return Settings{}, fmt.Errorf("uploads: settings: %w", err)
|
|
}
|
|
s := DefaultSettings
|
|
if err := json.Unmarshal(raw, &s); err != nil {
|
|
return Settings{}, fmt.Errorf("uploads: settings: %w", err)
|
|
}
|
|
return s, nil
|
|
}
|
|
|
|
// SaveSettings сохраняет настройки.
|
|
func (st *Store) SaveSettings(ctx context.Context, s Settings) error {
|
|
raw, _ := json.Marshal(s)
|
|
_, err := st.pool.Exec(ctx, `
|
|
INSERT INTO app_settings (key, value, updated_at) VALUES ($1, $2, now())
|
|
ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`, settingsKey, raw)
|
|
if err != nil {
|
|
return fmt.Errorf("uploads: save settings: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// --- Квота ---
|
|
|
|
// occupyingStatuses — для SQL: статусы, занимающие место.
|
|
var occupyingStatuses = []string{"uploading", "processing", "pending_review", "approved", "importing", "import_failed"}
|
|
|
|
// Usage — занято всего и пользователем.
|
|
func (st *Store) Usage(ctx context.Context, userID uuid.UUID) (used, userUsed int64, err error) {
|
|
err = st.pool.QueryRow(ctx, `
|
|
SELECT (SELECT used_bytes FROM upload_quota),
|
|
coalesce((SELECT sum(size_bytes) FROM upload_files WHERE author_id = $1 AND status::text = ANY($2)), 0)`,
|
|
userID, occupyingStatuses).Scan(&used, &userUsed)
|
|
if err != nil {
|
|
return 0, 0, fmt.Errorf("uploads: usage: %w", err)
|
|
}
|
|
return used, userUsed, nil
|
|
}
|
|
|
|
// NewFile — файл, заявленный при создании пачки.
|
|
type NewFile struct {
|
|
ClientID string
|
|
Name string
|
|
RelPath *string
|
|
Size int64
|
|
SHA256 []byte
|
|
}
|
|
|
|
// Reserved — итог по файлу: либо id в очереди, либо код отказа.
|
|
type Reserved struct {
|
|
ClientID string
|
|
FileID uuid.UUID
|
|
Rejected string
|
|
Track *uuid.UUID // для дубля — с каким треком совпал
|
|
}
|
|
|
|
// CreateBatch создаёт пачку и резервирует место под файлы: проверка «занято + запрошено ≤ лимит»
|
|
// под блокировкой строки счётчика (docs/uploads.md#3). Дубли по SHA-256 отклоняются, не занимая места.
|
|
func (st *Store) CreateBatch(ctx context.Context, author uuid.UUID, comment *string, files []NewFile, s Settings, coverPath *string) (uuid.UUID, []Reserved, error) {
|
|
tx, err := st.pool.Begin(ctx)
|
|
if err != nil {
|
|
return uuid.Nil, nil, fmt.Errorf("uploads: create batch: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
|
|
var used, userUsed int64
|
|
if err := tx.QueryRow(ctx, `SELECT used_bytes FROM upload_quota WHERE id FOR UPDATE`).Scan(&used); err != nil {
|
|
return uuid.Nil, nil, fmt.Errorf("uploads: lock quota: %w", err)
|
|
}
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT coalesce(sum(size_bytes), 0) FROM upload_files WHERE author_id = $1 AND status::text = ANY($2)`,
|
|
author, occupyingStatuses).Scan(&userUsed); err != nil {
|
|
return uuid.Nil, nil, fmt.Errorf("uploads: user usage: %w", err)
|
|
}
|
|
|
|
batchID := newID()
|
|
if _, err := tx.Exec(ctx, `INSERT INTO upload_batches (id, author_id, comment, cover_path) VALUES ($1, $2, $3, $4)`,
|
|
batchID, author, comment, coverPath); err != nil {
|
|
return uuid.Nil, nil, fmt.Errorf("uploads: insert batch: %w", err)
|
|
}
|
|
out := make([]Reserved, 0, len(files))
|
|
seen := map[string]bool{}
|
|
var reserved int64
|
|
for _, f := range files {
|
|
r := Reserved{ClientID: f.ClientID}
|
|
key := string(f.SHA256)
|
|
switch {
|
|
case f.Size > s.MaxFileBytes:
|
|
r.Rejected = CodeTooLarge
|
|
case seen[key]:
|
|
r.Rejected = CodeDuplicate // второй такой же файл в одной пачке
|
|
default:
|
|
track, dup, err := duplicateBySHA(ctx, tx, f.SHA256, uuid.Nil)
|
|
if err != nil {
|
|
return uuid.Nil, nil, err
|
|
}
|
|
switch {
|
|
case dup:
|
|
r.Rejected, r.Track = CodeDuplicate, track
|
|
case used+reserved+f.Size > s.QueueLimitBytes:
|
|
r.Rejected = CodeQueueFull
|
|
case s.PerUserLimitBytes != nil && userUsed+reserved+f.Size > *s.PerUserLimitBytes:
|
|
r.Rejected = CodeUserQuotaFull
|
|
}
|
|
}
|
|
seen[key] = true
|
|
if r.Rejected == "" {
|
|
r.FileID = newID()
|
|
staging := batchID.String() + "/" + r.FileID.String()
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO upload_files (id, batch_id, author_id, status, original_name, client_rel_path, size_bytes, sha256,
|
|
staging_path)
|
|
VALUES ($1, $2, $3, 'uploading', $4, $5, $6, $7, $8)`,
|
|
r.FileID, batchID, author, f.Name, f.RelPath, f.Size, f.SHA256, staging); err != nil {
|
|
return uuid.Nil, nil, fmt.Errorf("uploads: insert file: %w", err)
|
|
}
|
|
reserved += f.Size
|
|
}
|
|
out = append(out, r)
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE upload_quota SET used_bytes = used_bytes + $1 WHERE id`, reserved); err != nil {
|
|
return uuid.Nil, nil, fmt.Errorf("uploads: reserve: %w", err)
|
|
}
|
|
return batchID, out, tx.Commit(ctx)
|
|
}
|
|
|
|
// transition меняет статус под блокировкой строки и освобождает место, если новый статус его не занимает.
|
|
func transition(ctx context.Context, tx pgx.Tx, id uuid.UUID, from []Status, to Status) (Status, error) {
|
|
var (
|
|
prev Status
|
|
size int64
|
|
)
|
|
err := tx.QueryRow(ctx, `SELECT status, size_bytes FROM upload_files WHERE id = $1 FOR UPDATE`, id).Scan(&prev, &size)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return "", ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return "", fmt.Errorf("uploads: lock file: %w", err)
|
|
}
|
|
ok := false
|
|
for _, f := range from {
|
|
ok = ok || f == prev
|
|
}
|
|
if !ok {
|
|
return prev, ErrWrongStatus
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE upload_files SET status = $2, updated_at = now() WHERE id = $1`, id, to); err != nil {
|
|
return "", fmt.Errorf("uploads: set status: %w", err)
|
|
}
|
|
if prev.Occupies() && !to.Occupies() {
|
|
if _, err := tx.Exec(ctx, `UPDATE upload_quota SET used_bytes = greatest(0, used_bytes - $1) WHERE id`, size); err != nil {
|
|
return "", fmt.Errorf("uploads: release quota: %w", err)
|
|
}
|
|
}
|
|
return prev, nil
|
|
}
|
|
|
|
// Transition — смена статуса с дополнительным обновлением полей в той же транзакции (extra может быть nil).
|
|
func (st *Store) Transition(ctx context.Context, id uuid.UUID, from []Status, to Status, extra func(tx pgx.Tx) error) error {
|
|
tx, err := st.pool.Begin(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("uploads: transition: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
if _, err := transition(ctx, tx, id, from, to); err != nil {
|
|
return err
|
|
}
|
|
if extra != nil {
|
|
if err := extra(tx); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
// SetReceived — смещение tus после принятого куска.
|
|
func (st *Store) SetReceived(ctx context.Context, id uuid.UUID, offset int64) error {
|
|
_, err := st.pool.Exec(ctx, `
|
|
UPDATE upload_files SET received_bytes = $2, updated_at = now() WHERE id = $1 AND status = 'uploading'`, id, offset)
|
|
if err != nil {
|
|
return fmt.Errorf("uploads: set received: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// --- Чтение ---
|
|
|
|
const fileColumns = `f.id, f.batch_id, f.author_id, u.username, b.comment, f.status, f.original_name, f.client_rel_path,
|
|
f.size_bytes, f.received_bytes, f.sha256, coalesce(f.staging_path, ''), f.format, f.codec, f.bitrate, f.sample_rate,
|
|
f.bit_depth, f.duration_ms, f.tags, f.metadata, f.metadata_edited, f.matched_artist_id, f.matched_album_id,
|
|
f.duplicate_of_track_id, f.duplicate_level, f.possible_duplicate, f.quality_upgrade, f.target_source_id, f.target_rel_path,
|
|
f.status_reason, f.reviewed_by, f.reviewed_at, f.imported_file_id, b.cover_path, f.created_at, f.updated_at, src.name`
|
|
|
|
const fileFrom = ` FROM upload_files f JOIN upload_batches b ON b.id = f.batch_id JOIN users u ON u.id = f.author_id
|
|
LEFT JOIN library_sources src ON src.id = f.target_source_id`
|
|
|
|
func scanFile(row pgx.Row) (File, error) {
|
|
var (
|
|
f File
|
|
tags, meta []byte
|
|
)
|
|
err := row.Scan(&f.ID, &f.BatchID, &f.AuthorID, &f.AuthorName, &f.BatchComment, &f.Status, &f.OriginalName,
|
|
&f.ClientRelPath, &f.Size, &f.Received, &f.SHA256, &f.StagingPath, &f.Format, &f.Codec, &f.Bitrate, &f.SampleRate,
|
|
&f.BitDepth, &f.DurationMs, &tags, &meta, &f.MetadataEdited, &f.MatchedArtistID, &f.MatchedAlbumID,
|
|
&f.DuplicateOfTrackID, &f.DuplicateLevel, &f.PossibleDuplicate, &f.QualityUpgrade, &f.TargetSourceID, &f.TargetRelPath,
|
|
&f.StatusReason, &f.ReviewedBy, &f.ReviewedAt, &f.ImportedFileID, &f.CoverPath, &f.CreatedAt, &f.UpdatedAt, &f.TargetSourceName)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return File{}, ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return File{}, fmt.Errorf("uploads: scan file: %w", err)
|
|
}
|
|
if len(tags) > 0 {
|
|
_ = json.Unmarshal(tags, &f.Tags)
|
|
}
|
|
if len(meta) > 0 {
|
|
var m Meta
|
|
if json.Unmarshal(meta, &m) == nil {
|
|
f.Meta = &m
|
|
}
|
|
}
|
|
return f, nil
|
|
}
|
|
|
|
// File — файл по id.
|
|
func (st *Store) File(ctx context.Context, id uuid.UUID) (File, error) {
|
|
return scanFile(st.pool.QueryRow(ctx, `SELECT `+fileColumns+fileFrom+` WHERE f.id = $1`, id))
|
|
}
|
|
|
|
// ListFilter — фильтры списка очереди.
|
|
type ListFilter struct {
|
|
AuthorID *uuid.UUID
|
|
Statuses []Status
|
|
PossibleDuplicate *bool
|
|
BatchID *uuid.UUID
|
|
Cursor string // "<unix-nano>|<id>" последнего элемента
|
|
Limit int
|
|
}
|
|
|
|
// List — файлы очереди, новые первыми, страницами.
|
|
func (st *Store) List(ctx context.Context, fl ListFilter) ([]File, string, error) {
|
|
where := []string{"true"}
|
|
args := []any{}
|
|
add := func(cond string, v any) {
|
|
args = append(args, v)
|
|
where = append(where, strings.ReplaceAll(cond, "?", fmt.Sprintf("$%d", len(args))))
|
|
}
|
|
if fl.AuthorID != nil {
|
|
add("f.author_id = ?", *fl.AuthorID)
|
|
}
|
|
if fl.BatchID != nil {
|
|
add("f.batch_id = ?", *fl.BatchID)
|
|
}
|
|
if len(fl.Statuses) > 0 {
|
|
ss := make([]string, len(fl.Statuses))
|
|
for i, s := range fl.Statuses {
|
|
ss[i] = string(s)
|
|
}
|
|
add("f.status::text = ANY(?)", ss)
|
|
}
|
|
if fl.PossibleDuplicate != nil {
|
|
add("f.possible_duplicate = ?", *fl.PossibleDuplicate)
|
|
}
|
|
if fl.Cursor != "" {
|
|
ts, id, ok := strings.Cut(fl.Cursor, "|")
|
|
var nano int64
|
|
cid, err := uuid.Parse(id)
|
|
if _, serr := fmt.Sscan(ts, &nano); !ok || err != nil || serr != nil {
|
|
return nil, "", errors.New("uploads: bad cursor")
|
|
}
|
|
args = append(args, time.Unix(0, nano).UTC(), cid)
|
|
where = append(where, fmt.Sprintf("(f.created_at, f.id) < ($%d, $%d)", len(args)-1, len(args)))
|
|
}
|
|
limit := min(max(fl.Limit, 1), 200)
|
|
args = append(args, limit+1)
|
|
rows, err := st.pool.Query(ctx, `SELECT `+fileColumns+fileFrom+` WHERE `+strings.Join(where, " AND ")+
|
|
fmt.Sprintf(` ORDER BY f.created_at DESC, f.id DESC LIMIT $%d`, len(args)), args...)
|
|
if err != nil {
|
|
return nil, "", fmt.Errorf("uploads: list: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var out []File
|
|
for rows.Next() {
|
|
f, err := scanFile(rows)
|
|
if err != nil {
|
|
return nil, "", err
|
|
}
|
|
out = append(out, f)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, "", err
|
|
}
|
|
next := ""
|
|
if len(out) > limit {
|
|
out = out[:limit]
|
|
last := out[limit-1]
|
|
next = fmt.Sprintf("%d|%s", last.CreatedAt.UnixNano(), last.ID)
|
|
}
|
|
return out, next, nil
|
|
}
|
|
|
|
// NextWithStatus — самый старый файл в статусе (для фоновых обработчиков).
|
|
func (st *Store) NextWithStatus(ctx context.Context, s Status) (File, error) {
|
|
return scanFile(st.pool.QueryRow(ctx, `SELECT `+fileColumns+fileFrom+
|
|
` WHERE f.status = $1 ORDER BY f.updated_at, f.id LIMIT 1`, s))
|
|
}
|
|
|
|
// StagingInUse — пути staging файлов, которым он ещё нужен (сверка при старте).
|
|
func (st *Store) StagingInUse(ctx context.Context) (map[string]bool, error) {
|
|
rows, err := st.pool.Query(ctx, `SELECT staging_path FROM upload_files WHERE status::text = ANY($1) AND staging_path IS NOT NULL
|
|
UNION SELECT b.cover_path FROM upload_batches b JOIN upload_files f ON f.batch_id = b.id
|
|
WHERE f.status::text = ANY($1) AND b.cover_path IS NOT NULL`,
|
|
occupyingStatuses)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("uploads: staging in use: %w", err)
|
|
}
|
|
paths, err := pgx.CollectRows(rows, pgx.RowTo[string])
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := make(map[string]bool, len(paths))
|
|
for _, p := range paths {
|
|
out[p] = true
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// UnusedBatchCover returns a cover only after every file in its batch has released staging.
|
|
func (st *Store) UnusedBatchCover(ctx context.Context, batch uuid.UUID) (*string, error) {
|
|
var cover *string
|
|
err := st.pool.QueryRow(ctx, `SELECT b.cover_path FROM upload_batches b WHERE b.id = $1
|
|
AND NOT EXISTS (SELECT 1 FROM upload_files f WHERE f.batch_id = b.id AND f.status::text = ANY($2))`, batch, occupyingStatuses).Scan(&cover)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, nil
|
|
}
|
|
return cover, err
|
|
}
|
|
|
|
// Stale — недокачанные файлы старше before.
|
|
func (st *Store) Stale(ctx context.Context, before time.Time) ([]uuid.UUID, error) {
|
|
rows, err := st.pool.Query(ctx, `SELECT id FROM upload_files WHERE status = 'uploading' AND updated_at < $1`, before)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("uploads: stale: %w", err)
|
|
}
|
|
return pgx.CollectRows(rows, pgx.RowTo[uuid.UUID])
|
|
}
|
|
|
|
// Audit пишет действие в журнал.
|
|
func (st *Store) Audit(ctx context.Context, actor *uuid.UUID, action string, target uuid.UUID, payload map[string]any) {
|
|
raw, _ := json.Marshal(payload)
|
|
_, _ = st.pool.Exec(ctx, `
|
|
INSERT INTO audit_log (id, actor_id, action, target_type, target_id, payload) VALUES ($1, $2, $3, 'upload_file', $4, $5)`,
|
|
newID(), actor, action, target, raw)
|
|
}
|
|
|
|
// --- Каталог: дубли, сопоставление ---
|
|
|
|
// duplicateBySHA — точная копия: файл библиотеки или другой файл очереди, ещё занимающий место.
|
|
func duplicateBySHA(ctx context.Context, q interface {
|
|
QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
|
|
}, sum []byte, exclude uuid.UUID) (*uuid.UUID, bool, error) {
|
|
if len(sum) == 0 {
|
|
return nil, false, nil
|
|
}
|
|
var track *uuid.UUID
|
|
err := q.QueryRow(ctx, `
|
|
SELECT track_id FROM media_files WHERE sha256 = $1 AND missing_since IS NULL LIMIT 1`, sum).Scan(&track)
|
|
if err == nil {
|
|
return track, true, nil
|
|
}
|
|
if !errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, false, fmt.Errorf("uploads: sha duplicate: %w", err)
|
|
}
|
|
var id uuid.UUID
|
|
err = q.QueryRow(ctx, `
|
|
SELECT id FROM upload_files WHERE sha256 = $1 AND id <> $2 AND status::text = ANY($3) LIMIT 1`,
|
|
sum, exclude, occupyingStatuses).Scan(&id)
|
|
if err == nil {
|
|
return nil, true, nil
|
|
}
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, false, nil
|
|
}
|
|
return nil, false, fmt.Errorf("uploads: sha duplicate: %w", err)
|
|
}
|
|
|
|
// DuplicateBySHA — см. duplicateBySHA.
|
|
func (st *Store) DuplicateBySHA(ctx context.Context, sum []byte, exclude uuid.UUID) (*uuid.UUID, bool, error) {
|
|
return duplicateBySHA(ctx, st.pool, sum, exclude)
|
|
}
|
|
|
|
// PrintCandidate — отпечаток для сравнения: трек библиотеки или другой файл очереди.
|
|
type PrintCandidate struct {
|
|
TrackID *uuid.UUID
|
|
Points []uint32
|
|
}
|
|
|
|
// FingerprintCandidates — отпечатки библиотеки и очереди с длительностью в пределах tolMs.
|
|
func (st *Store) FingerprintCandidates(ctx context.Context, durationMs, tolMs int, exclude uuid.UUID) ([]PrintCandidate, error) {
|
|
rows, err := st.pool.Query(ctx, `
|
|
SELECT mf.track_id, tf.fingerprint FROM track_fingerprints tf JOIN media_files mf ON mf.id = tf.file_id
|
|
WHERE tf.duration_ms BETWEEN $1 AND $2 AND mf.missing_since IS NULL
|
|
UNION ALL
|
|
SELECT NULL, fingerprint FROM upload_files
|
|
WHERE fingerprint IS NOT NULL AND id <> $3 AND status::text = ANY($4) AND duration_ms BETWEEN $1 AND $2`,
|
|
durationMs-tolMs, durationMs+tolMs, exclude, occupyingStatuses)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("uploads: fingerprint candidates: %w", err)
|
|
}
|
|
return pgx.CollectRows(rows, func(r pgx.CollectableRow) (PrintCandidate, error) {
|
|
var (
|
|
c PrintCandidate
|
|
raw []int32
|
|
)
|
|
if err := r.Scan(&c.TrackID, &raw); err != nil {
|
|
return c, err
|
|
}
|
|
c.Points = make([]uint32, len(raw))
|
|
for i, v := range raw {
|
|
c.Points[i] = uint32(v)
|
|
}
|
|
return c, nil
|
|
})
|
|
}
|
|
|
|
// MetadataMatch — трек библиотеки с тем же исполнителем и названием и длительностью ±tolMs.
|
|
func (st *Store) MetadataMatch(ctx context.Context, artistKey, titleKey string, durationMs, tolMs int) (*uuid.UUID, error) {
|
|
var id uuid.UUID
|
|
err := st.pool.QueryRow(ctx, `
|
|
SELECT t.id FROM tracks t
|
|
JOIN track_artists ta ON ta.track_id = t.id
|
|
JOIN artists a ON a.id = ta.artist_id
|
|
WHERE t.title_norm = $1 AND a.name_norm = $2 AND abs(t.duration_ms - $3) <= $4
|
|
LIMIT 1`, titleKey, artistKey, durationMs, tolMs).Scan(&id)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("uploads: metadata match: %w", err)
|
|
}
|
|
return &id, nil
|
|
}
|
|
|
|
// MatchCatalog — существующие исполнитель (по ключу имени) и его альбом (по ключу названия), если есть.
|
|
func (st *Store) MatchCatalog(ctx context.Context, artistKey, albumKey string) (artist, album *uuid.UUID, maxDisc int, err error) {
|
|
var a uuid.UUID
|
|
err = st.pool.QueryRow(ctx, `SELECT id FROM artists WHERE name_norm = $1`, artistKey).Scan(&a)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, nil, 0, nil
|
|
}
|
|
if err != nil {
|
|
return nil, nil, 0, fmt.Errorf("uploads: match artist: %w", err)
|
|
}
|
|
artist = &a
|
|
var (
|
|
al uuid.UUID
|
|
disc *int
|
|
)
|
|
err = st.pool.QueryRow(ctx, `
|
|
SELECT al.id, (SELECT max(disc_no) FROM tracks WHERE album_id = al.id)
|
|
FROM albums al WHERE al.album_artist_id = $1 AND al.title_norm = $2`, a, albumKey).Scan(&al, &disc)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return artist, nil, 0, nil
|
|
}
|
|
if err != nil {
|
|
return nil, nil, 0, fmt.Errorf("uploads: match album: %w", err)
|
|
}
|
|
if disc != nil {
|
|
maxDisc = *disc
|
|
}
|
|
return artist, &al, maxDisc, nil
|
|
}
|
|
|
|
// BatchMaxDisc — наибольший номер диска среди файлов пачки с тем же альбомом (для {disc:1-}).
|
|
func (st *Store) BatchMaxDisc(ctx context.Context, batchID uuid.UUID, album string) (int, error) {
|
|
var n *int
|
|
err := st.pool.QueryRow(ctx, `
|
|
SELECT max((metadata->>'disc')::int) FROM upload_files
|
|
WHERE batch_id = $1 AND metadata->>'album' = $2 AND status::text = ANY($3)`,
|
|
batchID, album, occupyingStatuses).Scan(&n)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("uploads: batch discs: %w", err)
|
|
}
|
|
if n == nil {
|
|
return 0, nil
|
|
}
|
|
return *n, nil
|
|
}
|
|
|
|
// MarkImported связывает файл библиотеки с загрузкой.
|
|
func MarkImported(ctx context.Context, tx pgx.Tx, uploadID, mediaFileID uuid.UUID) error {
|
|
if _, err := tx.Exec(ctx, `UPDATE upload_files SET imported_file_id = $2, status_reason = NULL WHERE id = $1`,
|
|
uploadID, mediaFileID); err != nil {
|
|
return fmt.Errorf("uploads: mark imported: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE media_files SET upload_file_id = $1 WHERE id = $2`, uploadID, mediaFileID); err != nil {
|
|
return fmt.Errorf("uploads: link media file: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ReconcileQuota пересчитывает занятое место по статусам файлов.
|
|
func (st *Store) ReconcileQuota(ctx context.Context) error {
|
|
_, err := st.pool.Exec(ctx, `
|
|
UPDATE upload_quota SET used_bytes = coalesce((
|
|
SELECT sum(size_bytes) FROM upload_files WHERE status::text = ANY($1)
|
|
), 0)`, occupyingStatuses)
|
|
if err != nil {
|
|
return fmt.Errorf("uploads: reconcile quota: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// BestAudio — лучший доступный файл трека (для сравнения качества).
|
|
func (st *Store) BestAudio(ctx context.Context, trackID uuid.UUID) (codec, format string, bitrate int, err error) {
|
|
err = st.pool.QueryRow(ctx, `
|
|
SELECT coalesce(codec, ''), coalesce(format, ''), coalesce(bitrate, 0)
|
|
FROM media_files
|
|
WHERE track_id = $1 AND missing_since IS NULL
|
|
ORDER BY (codec IN ('flac', 'alac', 'wavpack', 'ape', 'tta') OR codec LIKE 'pcm_%') DESC,
|
|
bitrate DESC NULLS LAST
|
|
LIMIT 1`, trackID).Scan(&codec, &format, &bitrate)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return "", "", 0, nil
|
|
}
|
|
if err != nil {
|
|
return "", "", 0, fmt.Errorf("uploads: best audio: %w", err)
|
|
}
|
|
return codec, format, bitrate, nil
|
|
}
|
|
|
|
// PendingInBatch — файлы пачки, которые ещё можно править.
|
|
func (st *Store) PendingInBatch(ctx context.Context, batchID uuid.UUID) ([]File, error) {
|
|
rows, err := st.pool.Query(ctx, `SELECT `+fileColumns+fileFrom+`
|
|
WHERE f.batch_id = $1 AND f.status IN ('pending_review', 'import_failed')
|
|
ORDER BY f.created_at, f.id`, batchID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("uploads: batch files: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var out []File
|
|
for rows.Next() {
|
|
f, err := scanFile(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, f)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// SetCover записывает путь обложки пачки (nil — снять).
|
|
func (st *Store) SetCover(ctx context.Context, batchID uuid.UUID, path *string) error {
|
|
tag, err := st.pool.Exec(ctx, `UPDATE upload_batches SET cover_path = $2 WHERE id = $1`, batchID, path)
|
|
if err != nil {
|
|
return fmt.Errorf("uploads: set cover: %w", err)
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return ErrNotFound
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// UnseenDecisions — решения по файлам автора новее последнего визита.
|
|
func (st *Store) UnseenDecisions(ctx context.Context, userID uuid.UUID) (int, error) {
|
|
var n int
|
|
err := st.pool.QueryRow(ctx, `
|
|
SELECT count(*) FROM upload_files f
|
|
LEFT JOIN upload_visits v ON v.user_id = f.author_id
|
|
WHERE f.author_id = $1
|
|
AND f.status IN ('imported', 'rejected', 'duplicate', 'invalid', 'expired', 'import_failed')
|
|
AND f.updated_at > coalesce(v.seen_at, '-infinity')`, userID).Scan(&n)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("uploads: unseen: %w", err)
|
|
}
|
|
return n, nil
|
|
}
|
|
|
|
// MarkSeen отмечает визит автора.
|
|
func (st *Store) MarkSeen(ctx context.Context, userID uuid.UUID, at time.Time) error {
|
|
_, err := st.pool.Exec(ctx, `
|
|
INSERT INTO upload_visits (user_id, seen_at) VALUES ($1, $2)
|
|
ON CONFLICT (user_id) DO UPDATE SET seen_at = EXCLUDED.seen_at`, userID, at)
|
|
if err != nil {
|
|
return fmt.Errorf("uploads: mark seen: %w", err)
|
|
}
|
|
return nil
|
|
}
|