diff --git a/api/server.go b/api/server.go index 7a792f25..74d9b386 100644 --- a/api/server.go +++ b/api/server.go @@ -593,6 +593,7 @@ func NewApiServer(config config.Config) *ApiServer { g.Post("/tracks/:trackId/reposts", app.requireAuthMiddleware, app.requireWriteScope, app.postV1TrackRepost) g.Delete("/tracks/:trackId/reposts", app.requireAuthMiddleware, app.requireWriteScope, app.deleteV1TrackRepost) g.Get("/tracks/:trackId/stems", app.v1TrackStems) + g.Get("/tracks/:trackId/stems/download", app.v1TrackStemsDownload) g.Get("/tracks/:trackId/favorites", app.v1TrackFavorites) g.Post("/tracks/:trackId/favorites", app.requireAuthMiddleware, app.requireWriteScope, app.postV1TrackFavorite) g.Delete("/tracks/:trackId/favorites", app.requireAuthMiddleware, app.requireWriteScope, app.deleteV1TrackFavorite) diff --git a/api/swagger/swagger-v1.yaml b/api/swagger/swagger-v1.yaml index 80be36e9..7d088043 100644 --- a/api/swagger/swagger-v1.yaml +++ b/api/swagger/swagger-v1.yaml @@ -4381,6 +4381,66 @@ paths: application/json: schema: $ref: "#/components/schemas/stems_response" + /tracks/{track_id}/stems/download: + get: + tags: + - tracks + summary: Download a track's stems as one zip + description: + Streams every stem the requester may download, and optionally the + parent track, as a single uncompressed zip with a known + Content-Length. Files that can't be downloaded are left out. + operationId: Download Track Stems + security: + - {} + - OAuth2: + - read + parameters: + - name: track_id + in: path + description: A Track ID + required: true + schema: + type: string + - name: include_parent + in: query + description: Optional - also include the parent track's file + schema: + type: boolean + default: false + - name: user_id + in: query + description: The user ID of the user making the request + schema: + type: string + - name: user_signature + in: query + description: + Optional - signature from the requesting user's wallet. Needed + for gated stems and for an artist downloading their own files. + schema: + type: string + - name: user_data + in: query + description: + Optional - data which was used to generate the optional signature + argument. + schema: + type: string + responses: + "200": + description: Success + content: + application/zip: {} + "403": + description: Not allowed to download any of the stems + content: {} + "404": + description: Track not found, or no downloadable stems + content: {} + "429": + description: Too many downloads in progress + content: {} /tracks/{track_id}/shares: post: tags: diff --git a/api/v1_track_inspect.go b/api/v1_track_inspect.go index a5b24504..5d8d70f5 100644 --- a/api/v1_track_inspect.go +++ b/api/v1_track_inspect.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "net/http" + "time" "api.audius.co/api/dbv1" "api.audius.co/rendezvous" @@ -22,27 +23,28 @@ type inspectResponse struct { } func inspectTrack(track dbv1.Track, original bool) (*inspectResponse, error) { - var cid string if original { - cid = track.OrigFileCid.String - } else { - cid = track.TrackCid.String + return inspectCid(track.OrigFileCid.String) } + return inspectCid(track.TrackCid.String) +} + +var inspectClient = &http.Client{Timeout: 10 * time.Second} +func inspectCid(cid string) (*inspectResponse, error) { hosts := rendezvous.GlobalHasher.Rank(cid) var info blobInspect var lastErr error for _, host := range hosts { - client := &http.Client{} req, err := http.NewRequest("GET", fmt.Sprintf("%s/internal/blobs/info/%s", host, cid), nil) if err != nil { lastErr = err continue } - resp, err := client.Do(req) + resp, err := inspectClient.Do(req) if err != nil { lastErr = err continue diff --git a/api/v1_track_stems_download.go b/api/v1_track_stems_download.go new file mode 100644 index 00000000..58e6bd58 --- /dev/null +++ b/api/v1_track_stems_download.go @@ -0,0 +1,546 @@ +package api + +import ( + "archive/zip" + "context" + "errors" + "fmt" + "hash/crc32" + "io" + "net" + "net/http" + "net/url" + "path" + "strconv" + "strings" + "sync" + "time" + "unicode/utf8" + + "api.audius.co/api/dbv1" + apiutils "api.audius.co/utils" + "github.com/gofiber/fiber/v2" + "github.com/jackc/pgx/v5" + "go.uber.org/zap" + "golang.org/x/sync/errgroup" +) + +const ( + stemsZipMaxPerPod = 32 + stemsZipMaxPerUser = 2 + + // A content node that stops sending bytes for this long is treated as + // failed, and the file resumes from the next mirror. + stemsZipStallTimeout = 30 * time.Second +) + +var stemsZipSlots = newDownloadSlots(stemsZipMaxPerPod, stemsZipMaxPerUser) + +// No overall timeout: a large stem legitimately takes minutes. Stalls are +// caught per read instead, see copyWithStallTimeout. +var stemsZipClient = &http.Client{ + Transport: &http.Transport{ + Proxy: http.ProxyFromEnvironment, + DialContext: (&net.Dialer{Timeout: 10 * time.Second}).DialContext, + TLSHandshakeTimeout: 10 * time.Second, + ResponseHeaderTimeout: 15 * time.Second, + IdleConnTimeout: 90 * time.Second, + MaxIdleConnsPerHost: 8, + }, +} + +type stemsZipEntry struct { + name string + size int64 + link *dbv1.MediaLink +} + +// v1TrackStemsDownload streams a track's stems, and optionally the parent, as +// one zip. Entries are stored uncompressed so the archive's length is known +// before any audio is fetched, which lets the browser's download manager show +// real progress. +func (app *ApiServer) v1TrackStemsDownload(c *fiber.Ctx) error { + trackId := int32(c.Locals("trackId").(int)) + includeParent := c.QueryBool("include_parent") + + rootName, entries, err := app.stemsZipEntries(c, trackId, includeParent) + if err != nil { + return err + } + + slotKey := apiutils.GetIP(c) + if myId := app.getMyId(c); myId != 0 { + slotKey = strconv.Itoa(int(myId)) + } + if !stemsZipSlots.acquire(slotKey) { + return fiber.NewError(fiber.StatusTooManyRequests, "too many downloads in progress") + } + + headers := stemsZipHeaders(rootName, entries, time.Now()) + length, err := stemsZipLength(headers) + if err != nil { + stemsZipSlots.release(slotKey) + return err + } + + logger := app.logger.With(zap.Int32("track_id", trackId), zap.Int64("bytes", length)) + pr, pw := io.Pipe() + go func() { + defer stemsZipSlots.release(slotKey) + err := writeStemsZip(pw, headers, entries) + if err != nil && !errors.Is(err, io.ErrClosedPipe) { + logger.Warn("stems zip download failed", zap.Error(err)) + } + pw.CloseWithError(err) + }() + + c.Set(fiber.HeaderContentType, "application/zip") + c.Set(fiber.HeaderContentDisposition, attachmentDisposition(rootName+".zip")) + c.Set(fiber.HeaderCacheControl, "private, no-store") + // Keep ingress-nginx from spooling the whole zip to disk for slow clients. + c.Set("X-Accel-Buffering", "no") + // fasthttp closes pr once the response ends, successful or not, which + // fails the next write and stops the goroutine. + c.Context().SetBodyStream(pr, int(length)) + return nil +} + +// stemsZipEntries resolves the files the requester may download, with a +// signed link and size for each. Files that can't be downloaded are left out +// rather than failing the whole archive. +func (app *ApiServer) stemsZipEntries(c *fiber.Ctx, trackId int32, includeParent bool) (string, []stemsZipEntry, error) { + rows, err := app.pool.Query(c.Context(), ` + SELECT s.child_track_id + FROM stems s + JOIN tracks t ON t.track_id = s.child_track_id + WHERE s.parent_track_id = @track_id + AND t.is_current = true + AND t.is_delete = false + ORDER BY s.child_track_id + `, pgx.NamedArgs{"track_id": trackId}) + if err != nil { + return "", nil, err + } + stemIds, err := pgx.CollectRows(rows, pgx.RowTo[int32]) + if err != nil { + return "", nil, err + } + + tracks, err := app.queries.Tracks(c.Context(), dbv1.TracksParams{ + GetTracksParams: dbv1.GetTracksParams{ + MyID: app.getMyId(c), + Ids: append([]int32{trackId}, stemIds...), + AuthedWallet: app.tryGetAuthedWallet(c), + IncludeUnlisted: true, + }, + }) + if err != nil { + return "", nil, err + } + + byId := make(map[int32]dbv1.Track, len(tracks)) + for _, t := range tracks { + byId[t.TrackID] = t + } + + parent, ok := byId[trackId] + if !ok || !parent.IsAudioAllowed { + return "", nil, fiber.NewError(fiber.StatusNotFound, "track not found") + } + if len(stemIds) == 0 { + return "", nil, fiber.NewError(fiber.StatusNotFound, "track has no stems") + } + + var candidates []dbv1.Track + if includeParent { + candidates = append(candidates, parent) + } + for _, id := range stemIds { + if t, ok := byId[id]; ok && t.IsAudioAllowed { + candidates = append(candidates, t) + } + } + type resolved struct { + track dbv1.Track + link *dbv1.MediaLink + size int64 + contentType string + } + var files []*resolved + denied := false + for _, t := range candidates { + link := t.Download + if link == nil { + link, err = app.ownerDownloadLink(c, &t) + if err != nil { + return "", nil, err + } + } + if link == nil { + denied = denied || !t.Access.Download + continue + } + files = append(files, &resolved{track: t, link: link}) + } + + g := errgroup.Group{} + for _, f := range files { + g.Go(func() error { + info, err := inspectCid(f.track.DownloadCid()) + if err != nil { + app.logger.Warn("stems zip: skipping file with no blob info", + zap.Int32("track_id", f.track.TrackID), zap.Error(err)) + return nil + } + f.size = info.Size + f.contentType = info.ContentType + return nil + }) + } + g.Wait() + + rootName := sanitizeZipName(parent.Title.String) + if rootName == "" { + rootName = "stems" + } + + seen := map[string]int{} + var entries []stemsZipEntry + for _, f := range files { + if f.size <= 0 { + continue + } + entries = append(entries, stemsZipEntry{ + name: dedupeZipName(stemsZipFilename(&f.track, f.contentType), seen), + size: f.size, + link: f.link, + }) + } + + if len(entries) == 0 { + if denied { + return "", nil, fiber.NewError(fiber.StatusForbidden, "you are not allowed to download these stems") + } + return "", nil, fiber.NewError(fiber.StatusNotFound, "no downloadable files") + } + return rootName, entries, nil +} + +var audioExtensions = map[string]string{ + "audio/wav": ".wav", + "audio/wave": ".wav", + "audio/x-wav": ".wav", + "audio/flac": ".flac", + "audio/x-flac": ".flac", + "audio/aiff": ".aiff", + "audio/x-aiff": ".aiff", + "audio/mpeg": ".mp3", + "audio/mp4": ".m4a", + "audio/x-m4a": ".m4a", + "audio/ogg": ".ogg", +} + +// stemsZipFilename is the download filename, except that an original with no +// recorded filename gets its extension from the blob's content type rather +// than createFilename's .mp3, since the bytes are the original upload. +func stemsZipFilename(t *dbv1.Track, contentType string) string { + name := sanitizeZipName(createFilename(t)) + if t.OrigFileCid.String != "" && t.OrigFilename.String == "" { + if ext, ok := audioExtensions[strings.ToLower(contentType)]; ok { + name = strings.TrimSuffix(name, path.Ext(name)) + ext + } + } + if name == "" || strings.HasPrefix(name, ".") { + name = "track" + name + } + return name +} + +func sanitizeZipName(name string) string { + name = strings.Map(func(r rune) rune { + switch { + case r == '/' || r == '\\': + return '_' + case r < 0x20 || r == 0x7f: + return -1 + } + return r + }, name) + return strings.Trim(strings.TrimSpace(name), ".") +} + +// dedupeZipName appends " (n)" before the extension when a name repeats, +// ignoring case so the zip extracts cleanly on macOS and Windows. +func dedupeZipName(name string, seen map[string]int) string { + key := strings.ToLower(name) + n := seen[key] + seen[key] = n + 1 + if n == 0 { + return name + } + ext := path.Ext(name) + candidate := fmt.Sprintf("%s (%d)%s", strings.TrimSuffix(name, ext), n, ext) + return dedupeZipName(candidate, seen) +} + +func attachmentDisposition(filename string) string { + fallback := strings.Map(func(r rune) rune { + if r > 0x7e || r == '"' || r == '\\' { + return '_' + } + return r + }, filename) + return fmt.Sprintf(`attachment; filename="%s"; filename*=UTF-8''%s`, fallback, url.PathEscape(filename)) +} + +// stemsZipHeaders builds stored entries with a data descriptor. The CRC is +// filled in after each file is written, which archive/zip reads from the +// header pointer when it writes the descriptor and central directory. +func stemsZipHeaders(rootName string, entries []stemsZipEntry, modified time.Time) []*zip.FileHeader { + date, tm := msDosTime(modified) + headers := make([]*zip.FileHeader, len(entries)) + for i, e := range entries { + name := rootName + "/" + e.name + h := &zip.FileHeader{ + Name: name, + Method: zip.Store, + Flags: 0x8, + CreatorVersion: 3<<8 | 20, + ReaderVersion: 20, + ExternalAttrs: 0o644 << 16, + ModifiedDate: date, + ModifiedTime: tm, + CompressedSize64: uint64(e.size), + UncompressedSize64: uint64(e.size), + } + if !isASCII(name) { + h.Flags |= 0x800 + } + if e.size >= 1<<32-1 { + h.ReaderVersion = 45 + } + headers[i] = h + } + return headers +} + +// stemsZipLength lays the archive out against a writer that only counts, so +// it shares every byte of framing with writeStemsZip, including the zip64 +// records that depend on offsets past 4GiB. +func stemsZipLength(headers []*zip.FileHeader) (int64, error) { + cw := &countingWriter{} + zw := zip.NewWriter(cw) + zeros := make([]byte, 1<<20) + for _, h := range headers { + hc := *h + fw, err := zw.CreateRaw(&hc) + if err != nil { + return 0, err + } + for left := int64(h.UncompressedSize64); left > 0; { + n := min(left, int64(len(zeros))) + if _, err := fw.Write(zeros[:n]); err != nil { + return 0, err + } + left -= n + } + } + if err := zw.Close(); err != nil { + return 0, err + } + return cw.n, nil +} + +func writeStemsZip(w io.Writer, headers []*zip.FileHeader, entries []stemsZipEntry) error { + zw := zip.NewWriter(w) + for i, h := range headers { + fw, err := zw.CreateRaw(h) + if err != nil { + return err + } + sum := crc32.NewIEEE() + if err := fetchStem(io.MultiWriter(fw, sum), entries[i]); err != nil { + return fmt.Errorf("%s: %w", entries[i].name, err) + } + h.CRC32 = sum.Sum32() + } + return zw.Close() +} + +// fetchStem writes exactly e.size bytes of the file, resuming with a Range +// request on the next mirror when a node fails partway through. +func fetchStem(w io.Writer, e stemsZipEntry) error { + urls, err := mediaLinkUrls(e.link) + if err != nil { + return err + } + + var written int64 + var lastErr error + for _, u := range urls { + n, err := fetchRange(w, u, written, e.size) + written += n + if err == nil { + return nil + } + var we *writeError + if errors.As(err, &we) { + return we.err + } + lastErr = err + } + return fmt.Errorf("all mirrors failed after %d of %d bytes: %w", written, e.size, lastErr) +} + +type writeError struct{ err error } + +func (e *writeError) Error() string { return e.err.Error() } + +func fetchRange(w io.Writer, u *url.URL, offset, size int64) (int64, error) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, "GET", u.String(), nil) + if err != nil { + return 0, err + } + if offset > 0 { + req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset)) + } + + resp, err := stemsZipClient.Do(req) + if err != nil { + return 0, err + } + defer resp.Body.Close() + + switch { + case offset == 0 && resp.StatusCode == http.StatusOK: + case offset > 0 && resp.StatusCode == http.StatusPartialContent: + default: + return 0, fmt.Errorf("%s returned %d", u.Host, resp.StatusCode) + } + + remaining := size - offset + n, err := copyWithStallTimeout(w, io.LimitReader(resp.Body, remaining), cancel) + if err != nil { + return n, err + } + if n != remaining { + return n, fmt.Errorf("%s ended after %d of %d bytes", u.Host, offset+n, size) + } + return n, nil +} + +// copyWithStallTimeout copies like io.Copy but cancels the request when no +// bytes arrive for stemsZipStallTimeout. Write failures come back as +// *writeError so the caller can tell a gone client from a bad node. +func copyWithStallTimeout(w io.Writer, r io.Reader, cancel context.CancelFunc) (int64, error) { + timer := time.AfterFunc(stemsZipStallTimeout, cancel) + defer timer.Stop() + + buf := make([]byte, 64*1024) + var total int64 + for { + n, err := r.Read(buf) + if n > 0 { + timer.Reset(stemsZipStallTimeout) + if _, werr := w.Write(buf[:n]); werr != nil { + return total, &writeError{werr} + } + total += int64(n) + } + if err == io.EOF { + return total, nil + } + if err != nil { + return total, err + } + } +} + +// mediaLinkUrls lists the primary URL then each mirror with the same signed +// path, the order tryFindWorkingUrl uses. +func mediaLinkUrls(link *dbv1.MediaLink) ([]*url.URL, error) { + main, err := url.Parse(link.Url) + if err != nil { + return nil, err + } + q := main.Query() + q.Set("skip_play_count", "true") + main.RawQuery = q.Encode() + + urls := []*url.URL{main} + for _, mirror := range link.Mirrors { + m, err := url.Parse(mirror) + if err != nil || m.Host == main.Host { + continue + } + u := *main + u.Scheme = m.Scheme + u.Host = m.Host + urls = append(urls, &u) + } + return urls, nil +} + +func msDosTime(t time.Time) (uint16, uint16) { + t = t.UTC() + if t.Year() < 1980 { + t = time.Date(1980, 1, 1, 0, 0, 0, 0, time.UTC) + } + date := uint16(t.Day() + int(t.Month())<<5 + (t.Year()-1980)<<9) + tm := uint16(t.Second()/2 + t.Minute()<<5 + t.Hour()<<11) + return date, tm +} + +func isASCII(s string) bool { + for i := 0; i < len(s); i++ { + if s[i] >= utf8.RuneSelf { + return false + } + } + return true +} + +type countingWriter struct{ n int64 } + +func (w *countingWriter) Write(p []byte) (int, error) { + w.n += int64(len(p)) + return len(p), nil +} + +// downloadSlots caps concurrent long-lived downloads per pod and per key. +type downloadSlots struct { + mu sync.Mutex + total int + perKey map[string]int + maxTotal int + maxPerUser int +} + +func newDownloadSlots(maxTotal, maxPerUser int) *downloadSlots { + return &downloadSlots{perKey: map[string]int{}, maxTotal: maxTotal, maxPerUser: maxPerUser} +} + +func (s *downloadSlots) acquire(key string) bool { + s.mu.Lock() + defer s.mu.Unlock() + if s.total >= s.maxTotal || s.perKey[key] >= s.maxPerUser { + return false + } + s.total++ + s.perKey[key]++ + return true +} + +func (s *downloadSlots) release(key string) { + s.mu.Lock() + defer s.mu.Unlock() + s.total-- + if s.perKey[key] <= 1 { + delete(s.perKey, key) + } else { + s.perKey[key]-- + } +} diff --git a/api/v1_track_stems_download_test.go b/api/v1_track_stems_download_test.go new file mode 100644 index 00000000..3bf3a4a5 --- /dev/null +++ b/api/v1_track_stems_download_test.go @@ -0,0 +1,355 @@ +package api + +import ( + "archive/zip" + "bytes" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "api.audius.co/api/dbv1" + "api.audius.co/api/testdata" + "api.audius.co/database" + "api.audius.co/rendezvous" + "api.audius.co/trashid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeContentNode serves blob info and cidstream for the given blobs, with +// Range support like a real node. +func fakeContentNode(t *testing.T, blobs map[string][]byte) *httptest.Server { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if cid, ok := strings.CutPrefix(r.URL.Path, "/internal/blobs/info/"); ok { + data, found := blobs[cid] + if !found { + http.NotFound(w, r) + return + } + fmt.Fprintf(w, `{"ContentType":"audio/wav","Size":%d}`, len(data)) + return + } + if cid, ok := strings.CutPrefix(r.URL.Path, "/tracks/cidstream/"); ok { + data, found := blobs[cid] + if !found { + http.NotFound(w, r) + return + } + http.ServeContent(w, r, cid, time.Time{}, bytes.NewReader(data)) + return + } + http.NotFound(w, r) + })) + t.Cleanup(srv.Close) + return srv +} + +func useContentNodes(t *testing.T, hosts ...string) { + prev := rendezvous.GlobalHasher + rendezvous.GlobalHasher = rendezvous.NewRendezvousHasher(hosts) + t.Cleanup(func() { rendezvous.GlobalHasher = prev }) +} + +func blob(size int, seed byte) []byte { + b := make([]byte, size) + for i := range b { + b[i] = byte(i) ^ seed + } + return b +} + +func readZip(t *testing.T, body []byte) map[string][]byte { + zr, err := zip.NewReader(bytes.NewReader(body), int64(len(body))) + require.NoError(t, err) + files := map[string][]byte{} + for _, f := range zr.File { + rc, err := f.Open() + require.NoError(t, err) + data, err := io.ReadAll(rc) + require.NoError(t, err, f.Name) + rc.Close() + files[f.Name] = data + } + return files +} + +func TestStemsZipLengthMatchesOutput(t *testing.T) { + blobs := map[string][]byte{ + "QmA": blob(200_000, 1), + "QmB": blob(1, 2), + "QmC": blob(70_000, 3), + } + srv := fakeContentNode(t, blobs) + + entries := []stemsZipEntry{ + {name: "🍕DRUMS.wav", size: 200_000, link: &dbv1.MediaLink{Url: srv.URL + "/tracks/cidstream/QmA"}}, + {name: "bass.wav", size: 1, link: &dbv1.MediaLink{Url: srv.URL + "/tracks/cidstream/QmB"}}, + {name: "vocals.wav", size: 70_000, link: &dbv1.MediaLink{Url: srv.URL + "/tracks/cidstream/QmC"}}, + } + headers := stemsZipHeaders("Alone", entries, time.Now()) + length, err := stemsZipLength(headers) + require.NoError(t, err) + + var buf bytes.Buffer + require.NoError(t, writeStemsZip(&buf, headers, entries)) + assert.Equal(t, length, int64(buf.Len())) + + files := readZip(t, buf.Bytes()) + assert.Equal(t, blobs["QmA"], files["Alone/🍕DRUMS.wav"]) + assert.Equal(t, blobs["QmB"], files["Alone/bass.wav"]) + assert.Equal(t, blobs["QmC"], files["Alone/vocals.wav"]) +} + +func TestStemsZipResumesOnMirror(t *testing.T) { + data := blob(500_000, 7) + + // The primary sends part of the file, then drops the connection. + primary := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", fmt.Sprint(len(data))) + w.WriteHeader(http.StatusOK) + w.Write(data[:123_456]) + w.(http.Flusher).Flush() + conn, _, _ := w.(http.Hijacker).Hijack() + conn.Close() + })) + t.Cleanup(primary.Close) + mirror := fakeContentNode(t, map[string][]byte{"QmA": data}) + + entries := []stemsZipEntry{{ + name: "stem.wav", + size: int64(len(data)), + link: &dbv1.MediaLink{Url: primary.URL + "/tracks/cidstream/QmA", Mirrors: []string{mirror.URL}}, + }} + headers := stemsZipHeaders("Song", entries, time.Now()) + length, err := stemsZipLength(headers) + require.NoError(t, err) + + var buf bytes.Buffer + require.NoError(t, writeStemsZip(&buf, headers, entries)) + assert.Equal(t, length, int64(buf.Len())) + assert.Equal(t, data, readZip(t, buf.Bytes())["Song/stem.wav"]) +} + +func TestStemsZipFailsWhenNoMirrorHasFile(t *testing.T) { + srv := fakeContentNode(t, map[string][]byte{}) + entries := []stemsZipEntry{{ + name: "stem.wav", + size: 10, + link: &dbv1.MediaLink{Url: srv.URL + "/tracks/cidstream/QmMissing"}, + }} + err := writeStemsZip(io.Discard, stemsZipHeaders("Song", entries, time.Now()), entries) + assert.ErrorContains(t, err, "all mirrors failed") +} + +func TestStemsZipLengthPastZip64Offsets(t *testing.T) { + // Two entries that each fit in 32 bits but together push the second + // header's offset past 4GiB, which changes the central directory. + entries := []stemsZipEntry{ + {name: "a.wav", size: 3 << 30}, + {name: "b.wav", size: 2 << 30}, + } + length, err := stemsZipLength(stemsZipHeaders("Big", entries, time.Now())) + require.NoError(t, err) + assert.Greater(t, length, int64(5<<30)) +} + +func TestDedupeZipName(t *testing.T) { + seen := map[string]int{} + assert.Equal(t, "Kick.wav", dedupeZipName("Kick.wav", seen)) + assert.Equal(t, "kick (1).wav", dedupeZipName("kick.wav", seen)) + assert.Equal(t, "Kick (2).wav", dedupeZipName("Kick.wav", seen)) +} + +func TestStemsZipFilename(t *testing.T) { + track := func(fields map[string]string) *dbv1.Track { + var tr dbv1.Track + tr.Title.String = fields["title"] + tr.OrigFileCid.String = fields["orig_cid"] + tr.OrigFilename.String = fields["orig_filename"] + return &tr + } + assert.Equal(t, "🍕DRUMS.wav", stemsZipFilename(track(map[string]string{"orig_cid": "Qm", "orig_filename": " 🍕DRUMS.wav"}), "audio/wav")) + assert.Equal(t, "Bass.wav", stemsZipFilename(track(map[string]string{"title": "Bass", "orig_cid": "Qm"}), "audio/x-wav")) + assert.Equal(t, "Bass.mp3", stemsZipFilename(track(map[string]string{"title": "Bass"}), "audio/wav")) + assert.Equal(t, "a_b.mp3", stemsZipFilename(track(map[string]string{"title": "a/b"}), "")) +} + +func seedStemsTrack(t *testing.T) *ApiServer { + app := emptyTestApp(t) + database.Seed(app.pool.Replicas[0], database.FixtureMap{ + "users": { + {"user_id": 1, "handle": "artist", "wallet": ownerWallet}, + {"user_id": 2, "handle": "stranger", "wallet": strangerWallet}, + }, + "tracks": { + { + "track_id": 1, + "owner_id": 1, + "title": "Alone", + "orig_file_cid": "QmParent", + "orig_filename": "Alone.wav", + "is_downloadable": true, + }, + { + "track_id": 2, + "owner_id": 1, + "title": "Drums", + "orig_file_cid": "QmDrums", + "orig_filename": " Drums.wav", + "is_downloadable": true, + "stem_of": map[string]any{"parent_track_id": 1, "category": "drums"}, + }, + { + "track_id": 3, + "owner_id": 1, + "title": "Drums again", + "orig_file_cid": "QmDrums2", + "orig_filename": "drums.wav", + "is_downloadable": true, + "stem_of": map[string]any{"parent_track_id": 1, "category": "drums"}, + }, + { + "track_id": 4, + "owner_id": 1, + "title": "Vocals", + "orig_file_cid": "QmVocals", + "is_downloadable": false, + "stem_of": map[string]any{"parent_track_id": 1, "category": "vocals"}, + }, + { + "track_id": 5, + "owner_id": 1, + "title": "No Stems", + "orig_file_cid": "QmLonely", + "is_downloadable": true, + }, + }, + "stems": { + {"child_track_id": 2, "parent_track_id": 1}, + {"child_track_id": 3, "parent_track_id": 1}, + {"child_track_id": 4, "parent_track_id": 1}, + }, + }) + return app +} + +func TestGetTrackStemsDownload(t *testing.T) { + blobs := map[string][]byte{ + "QmParent": blob(90_000, 1), + "QmDrums": blob(40_000, 2), + "QmDrums2": blob(30_000, 3), + "QmVocals": blob(20_000, 4), + } + srv := fakeContentNode(t, blobs) + app := seedStemsTrack(t) + useContentNodes(t, srv.URL) + + req := httptest.NewRequest("GET", "/v1/tracks/"+trashid.MustEncodeHashID(1)+"/stems/download?include_parent=true", nil) + res, err := app.Test(req, -1) + require.NoError(t, err) + body, err := io.ReadAll(res.Body) + require.NoError(t, err) + + require.Equal(t, 200, res.StatusCode, string(body)) + assert.Equal(t, "application/zip", res.Header.Get("Content-Type")) + assert.Equal(t, fmt.Sprint(len(body)), res.Header.Get("Content-Length")) + assert.Contains(t, res.Header.Get("Content-Disposition"), `filename="Alone.zip"`) + + // The vocals stem isn't downloadable, so it's left out. + assert.Equal(t, map[string][]byte{ + "Alone/Alone.wav": blobs["QmParent"], + "Alone/Drums.wav": blobs["QmDrums"], + "Alone/drums (1).wav": blobs["QmDrums2"], + }, readZip(t, body)) +} + +func TestGetTrackStemsDownload_WithoutParent(t *testing.T) { + srv := fakeContentNode(t, map[string][]byte{ + "QmParent": blob(10, 1), + "QmDrums": blob(10, 2), + "QmDrums2": blob(10, 3), + }) + app := seedStemsTrack(t) + useContentNodes(t, srv.URL) + + status, body := testGet(t, app, "/v1/tracks/"+trashid.MustEncodeHashID(1)+"/stems/download") + require.Equal(t, 200, status, string(body)) + files := readZip(t, body) + assert.Len(t, files, 2) + assert.NotContains(t, files, "Alone/Alone.wav") +} + +func TestGetTrackStemsDownload_SkipsMissingBlob(t *testing.T) { + srv := fakeContentNode(t, map[string][]byte{"QmDrums": blob(10, 2)}) + app := seedStemsTrack(t) + useContentNodes(t, srv.URL) + + status, body := testGet(t, app, "/v1/tracks/"+trashid.MustEncodeHashID(1)+"/stems/download?include_parent=true") + require.Equal(t, 200, status, string(body)) + assert.Equal(t, []string{"Alone/Drums.wav"}, keys(readZip(t, body))) +} + +func TestGetTrackStemsDownload_OwnerGetsNonDownloadableStem(t *testing.T) { + blobs := map[string][]byte{ + "QmDrums": blob(10, 2), + "QmDrums2": blob(10, 3), + "QmVocals": blob(10, 4), + } + srv := fakeContentNode(t, blobs) + app := seedStemsTrack(t) + useContentNodes(t, srv.URL) + + get := func(wallet string) []string { + req := httptest.NewRequest("GET", "/v1/tracks/"+trashid.MustEncodeHashID(1)+"/stems/download", nil) + sigData := testdata.GetSignatureData(wallet) + req.Header.Set("Encoded-Data-Message", sigData.Message) + req.Header.Set("Encoded-Data-Signature", sigData.Signature) + res, err := app.Test(req, -1) + require.NoError(t, err) + body, _ := io.ReadAll(res.Body) + require.Equal(t, 200, res.StatusCode, string(body)) + return keys(readZip(t, body)) + } + + assert.Contains(t, get(ownerWallet), "Alone/Vocals.wav") + assert.NotContains(t, get(strangerWallet), "Alone/Vocals.wav") +} + +func TestGetTrackStemsDownload_NotFound(t *testing.T) { + srv := fakeContentNode(t, map[string][]byte{}) + app := seedStemsTrack(t) + useContentNodes(t, srv.URL) + + status, _ := testGet(t, app, "/v1/tracks/"+trashid.MustEncodeHashID(5)+"/stems/download") + assert.Equal(t, 404, status) + + status, _ = testGet(t, app, "/v1/tracks/"+trashid.MustEncodeHashID(999)+"/stems/download") + assert.Equal(t, 404, status) + + // Stems exist but no blob can be found for any of them. + status, _ = testGet(t, app, "/v1/tracks/"+trashid.MustEncodeHashID(1)+"/stems/download") + assert.Equal(t, 404, status) +} + +func TestDownloadSlots(t *testing.T) { + s := newDownloadSlots(3, 2) + assert.True(t, s.acquire("a")) + assert.True(t, s.acquire("a")) + assert.False(t, s.acquire("a")) + assert.True(t, s.acquire("b")) + assert.False(t, s.acquire("c")) + s.release("a") + assert.True(t, s.acquire("c")) +} + +func keys[V any](m map[string]V) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + return out +}