feat(api): streaming, transcoding, events, history and likes (M3)
ci / api / api (push) Successful in 29s
ci / contract / contract (сгенерированный код совпадает с openapi.yaml) (push) Successful in 35s
ci / ts / ts (packages + web) (push) Failing after 49s
ci / mobile / mobile (Metro-бандл Android) (push) Successful in 1m1s

- tracks carry a stream_url signed for the user for 12 h and the
  user's reaction; /stream checks the signature, expiry and that the
  user is not blocked, then hands the file to nginx by X-Accel-Redirect
  (or serves it with Range support when X_ACCEL=false)
- original by default; formats browsers can't play and explicit
  high/medium/low go through Opus transcoding, never upwards: a lossy
  source under the target bitrate stays original; transcodes are
  cached per file and size, deduplicated in flight, limited by
  TRANSCODE_CONCURRENCY and trimmed hourly by LRU to the size limit
- /events takes up to 500 events with client UUIDs, drops repeats and
  events older than 30 days, counts plays (≥ half or 4 min, or
  finished), early skips and history entries (≥ 30 s)
- history with clearing, likes and dislikes with cursors and totals,
  track reactions, playback state with last-write-wins, tracks batch

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
mrleo1nidandClaude Opus 5.5 committed 2026-10-20 07:25:15 +03:00
1 parent 8b76aa3964
commit c05e523cbe
16 files changed
+3913 -155

No files matched your search

+553
View File
@@ -0,0 +1,553 @@
package playback
import (
"context"
"encoding/base64"
"errors"
"fmt"
"log/slog"
"net/http"
"os"
"path/filepath"
"strings"
"time"
"github.com/google/uuid"
"pipu/api/internal/api/gen"
"pipu/api/internal/auth"
"pipu/api/internal/library"
"pipu/api/internal/platform/apperr"
"pipu/api/internal/platform/httpx"
"pipu/api/internal/platform/sign"
)
// Коды ошибок.
const (
CodeLinkInvalid = "playback.link_invalid"
CodeUnavailable = "playback.track_unavailable"
CodeTranscodeFail = "playback.transcode_failed"
CodeTrackNotFound = "catalog.not_found"
maxEventAge = 30 * 24 * time.Hour
maxEventClockSkew = 5 * time.Minute
defaultCollectPage = 50
)
// Handler реализует playback- и collection-части gen.StrictServerInterface,
// а также library.Personalizer (ссылки на стрим и реакции в треках).
type Handler struct {
store *Store
catalog *library.Store
dto func() library.DTO
signer *sign.Signer
transcoder *Transcoder
// RootFor — каталог хранилища в файловой системе api (для отдачи файлов без nginx и транскода).
rootFor func(ctx context.Context, sourceID uuid.UUID) (string, error)
xAccel bool
log *slog.Logger
now func() time.Time
}
// NewHandler создаёт обработчик. dto — DTO каталога с персонализацией (замыкание: DTO ссылается на сам Handler).
func NewHandler(store *Store, catalog *library.Store, dto func() library.DTO, signer *sign.Signer, tr *Transcoder,
rootFor func(context.Context, uuid.UUID) (string, error), xAccel bool, log *slog.Logger,
) *Handler {
return &Handler{
store: store, catalog: catalog, dto: dto, signer: signer, transcoder: tr, rootFor: rootFor, xAccel: xAccel,
log: log, now: func() time.Time { return time.Now().UTC() },
}
}
func principal(ctx context.Context) (auth.Principal, error) {
p, ok := auth.PrincipalFrom(ctx)
if !ok {
return auth.Principal{}, apperr.Unauthorized(auth.CodeUnauthorized, "not authenticated")
}
return p, nil
}
// Personalize реализует library.Personalizer.
func (h *Handler) Personalize(ctx context.Context, rows []library.TrackRow, out []gen.Track) error {
p, ok := auth.PrincipalFrom(ctx)
if !ok {
return nil
}
ids := make([]uuid.UUID, len(rows))
for i, r := range rows {
ids[i] = r.ID
}
reactions, err := h.store.Reactions(ctx, p.UserID, ids)
if err != nil {
return err
}
now := h.now()
for i := range out {
if out[i].Available {
u := StreamURL(h.signer, p.UserID, rows[i].ID, now)
out[i].StreamUrl = &u
}
if r, ok := reactions[rows[i].ID]; ok {
rr := gen.Reaction(r)
out[i].Reaction = &rr
}
}
return nil
}
// --- стрим ---
// StreamTrack реализует gen.StrictServerInterface.
func (h *Handler) StreamTrack(ctx context.Context, req gen.StreamTrackRequestObject) (gen.StreamTrackResponseObject, error) {
prm := req.Params
if !VerifyStream(h.signer, prm.Sig, prm.U, req.Id, prm.Exp, h.now()) {
return nil, apperr.Forbidden(CodeLinkInvalid, "link is invalid or expired")
}
blocked, err := h.store.UserBlocked(ctx, prm.U)
if err != nil {
return nil, err
}
if blocked {
return nil, apperr.Forbidden(auth.CodeUserBlocked, "user is blocked")
}
f, err := h.store.StreamFile(ctx, req.Id)
if errors.Is(err, ErrNotFound) {
return nil, apperr.NotFound(CodeUnavailable, "track has no available file")
}
if err != nil {
return nil, err
}
if f.SourceStatus != "mounted" {
return nil, apperr.NotFound(CodeUnavailable, "storage is offline")
}
q := QualityAuto
if prm.Q != nil {
q = string(*prm.Q)
}
codec, bitrate := "", 0
if f.Codec != nil {
codec = *f.Codec
}
if f.Bitrate != nil {
bitrate = *f.Bitrate
}
target := Target(q, f.Format, codec, bitrate)
res := streamResponse{req: httpx.RequestFrom(ctx), xAccel: h.xAccel}
if target == "" {
res.contentType = ContentType(f.Format)
res.accel = accelPath(f)
if !h.xAccel {
root, err := h.rootFor(ctx, f.SourceID)
if err != nil {
return nil, err
}
res.file = filepath.Join(root, filepath.FromSlash(f.RelPath))
}
return res, nil
}
root, err := h.rootFor(ctx, f.SourceID)
if err != nil {
return nil, err
}
name := CacheName(f.FileID, f.Size, target)
path, err := h.transcoder.Ensure(ctx, filepath.Join(root, filepath.FromSlash(f.RelPath)), name, target)
if err != nil {
h.log.ErrorContext(ctx, "transcode failed", "track_id", req.Id, "err", err)
return nil, apperr.New(http.StatusBadGateway, CodeTranscodeFail, "transcoding failed")
}
res.contentType = "audio/ogg; codecs=opus"
res.accel = "/_internal/cache/" + name
res.file = path
return res, nil
}
// streamResponse — X-Accel-Redirect для nginx или сам файл с поддержкой Range.
type streamResponse struct {
req *http.Request
contentType string
accel string
file string
xAccel bool
}
// VisitStreamTrackResponse реализует gen.StreamTrackResponseObject.
func (s streamResponse) VisitStreamTrackResponse(w http.ResponseWriter) error {
w.Header().Set("Content-Type", s.contentType)
w.Header().Set("Cache-Control", "private, max-age=43200")
if s.xAccel {
w.Header().Set("X-Accel-Redirect", s.accel)
w.WriteHeader(http.StatusOK)
return nil
}
f, err := os.Open(s.file)
if err != nil {
return err
}
defer func() { _ = f.Close() }()
info, err := f.Stat()
if err != nil {
return err
}
if s.req == nil {
return fmt.Errorf("playback: request missing in context")
}
http.ServeContent(w, s.req, "", info.ModTime(), f)
return nil
}
// --- треки пачкой ---
// GetTracksBatch реализует gen.StrictServerInterface.
func (h *Handler) GetTracksBatch(ctx context.Context, req gen.GetTracksBatchRequestObject) (gen.GetTracksBatchResponseObject, error) {
rows, err := h.catalog.TracksByIDs(ctx, req.Body.Ids)
if err != nil {
return nil, err
}
items, err := h.dto().Tracks(ctx, rows)
if err != nil {
return nil, err
}
return gen.GetTracksBatch200JSONResponse{Items: items}, nil
}
// --- события ---
// PostEvents реализует gen.StrictServerInterface.
func (h *Handler) PostEvents(ctx context.Context, req gen.PostEventsRequestObject) (gen.PostEventsResponseObject, error) {
p, err := principal(ctx)
if err != nil {
return nil, err
}
now := h.now()
ids := make([]uuid.UUID, 0, len(req.Body.Events))
for _, e := range req.Body.Events {
ids = append(ids, e.TrackId)
}
durations, err := h.store.TrackDurations(ctx, ids)
if err != nil {
return nil, err
}
clientType := h.store.SessionClientType(ctx, p.SessionID)
out := gen.PostEvents200JSONResponse{}
tx, err := h.store.Begin(ctx)
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback(ctx) }()
for _, e := range req.Body.Events {
dur, known := durations[e.TrackId]
at := e.OccurredAt.UTC()
if !known || at.Before(now.Add(-maxEventAge)) {
out.Rejected++
continue
}
if at.After(now.Add(maxEventClockSkew)) {
at = now // часы клиента спешат
}
ev := Event{
ID: e.Id, Type: string(e.Type), TrackID: e.TrackId, OccurredAt: at, PositionMs: e.PositionMs,
PlayedMs: e.PlayedMs, ContextID: e.ContextId, WaveSessionID: e.WaveSessionId,
Offline: e.Offline != nil && *e.Offline,
}
if e.ContextType != nil {
ct := string(*e.ContextType)
ev.ContextType = &ct
}
inserted, err := h.store.InsertEvent(ctx, tx, p.UserID, clientType, ev)
if err != nil {
return nil, err
}
if !inserted {
out.Duplicates++
continue
}
out.Accepted++
played := 0
if e.PlayedMs != nil {
played = *e.PlayedMs
}
if o := Judge(ev.Type, played, dur); o != (Outcome{}) {
if err := h.store.ApplyOutcome(ctx, tx, p.UserID, ev, o); err != nil {
return nil, err
}
}
}
if err := tx.Commit(ctx); err != nil {
return nil, err
}
return out, nil
}
// --- состояние плеера ---
func stateDTO(s State) gen.PlaybackState {
out := gen.PlaybackState{
TrackId: s.TrackID, PositionMs: s.PositionMs, Queue: s.Queue, QueueIndex: s.QueueIndex, Shuffle: s.Shuffle,
RepeatMode: gen.PlaybackStateRepeatMode(s.RepeatMode), Volume: s.Volume, DeviceId: s.DeviceID, UpdatedAt: s.UpdatedAt,
}
if out.Queue == nil {
out.Queue = []uuid.UUID{}
}
if s.Context != nil {
c := gen.PlaybackContext{}
str := func(k string) *string {
if v, ok := s.Context[k].(string); ok && v != "" {
return &v
}
return nil
}
c.Type, c.Id, c.Label = str("type"), str("id"), str("label")
if w := str("wave_session_id"); w != nil {
if id, err := uuid.Parse(*w); err == nil {
c.WaveSessionId = &id
}
}
out.Context = &c
}
return out
}
// GetPlaybackState реализует gen.StrictServerInterface.
func (h *Handler) GetPlaybackState(ctx context.Context, _ gen.GetPlaybackStateRequestObject) (gen.GetPlaybackStateResponseObject, error) {
p, err := principal(ctx)
if err != nil {
return nil, err
}
s, err := h.store.GetState(ctx, p.UserID)
if err != nil {
return nil, err
}
return gen.GetPlaybackState200JSONResponse(stateDTO(s)), nil
}
// PutPlaybackState реализует gen.StrictServerInterface.
func (h *Handler) PutPlaybackState(ctx context.Context, req gen.PutPlaybackStateRequestObject) (gen.PutPlaybackStateResponseObject, error) {
p, err := principal(ctx)
if err != nil {
return nil, err
}
b := req.Body
s := State{
TrackID: b.TrackId, PositionMs: max(b.PositionMs, 0), Queue: b.Queue, QueueIndex: b.QueueIndex,
RepeatMode: "off", Volume: 1, DeviceID: b.DeviceId,
}
if s.QueueIndex >= len(s.Queue) {
s.QueueIndex = max(len(s.Queue)-1, 0)
}
if b.Shuffle != nil {
s.Shuffle = *b.Shuffle
}
if b.RepeatMode != nil {
s.RepeatMode = string(*b.RepeatMode)
}
if b.Volume != nil {
s.Volume = min(max(*b.Volume, 0), 1)
}
if c := b.Context; c != nil {
s.Context = map[string]any{}
for k, v := range map[string]*string{"type": c.Type, "id": c.Id, "label": c.Label} {
if v != nil {
s.Context[k] = *v
}
}
if c.WaveSessionId != nil {
s.Context["wave_session_id"] = c.WaveSessionId.String()
}
}
saved, err := h.store.PutState(ctx, p.UserID, s)
if err != nil {
return nil, err
}
return gen.PutPlaybackState200JSONResponse(stateDTO(saved)), nil
}
// --- коллекция ---
func encodeCursor(t time.Time, id uuid.UUID) string {
return base64.RawURLEncoding.EncodeToString([]byte(t.Format(time.RFC3339Nano) + "|" + id.String()))
}
func decodeCursor(c *string) (time.Time, uuid.UUID, bool, error) {
if c == nil || *c == "" {
return time.Time{}, uuid.Nil, false, nil
}
bad := apperr.Validation("cursor", "validation.cursor_invalid", "malformed cursor", nil)
raw, err := base64.RawURLEncoding.DecodeString(*c)
if err != nil {
return time.Time{}, uuid.Nil, false, bad
}
ts, idStr, ok := strings.Cut(string(raw), "|")
t, err1 := time.Parse(time.RFC3339Nano, ts)
id, err2 := uuid.Parse(idStr)
if !ok || err1 != nil || err2 != nil {
return time.Time{}, uuid.Nil, false, bad
}
return t, id, true, nil
}
func pageLimit(l *int) int {
if l == nil {
return defaultCollectPage
}
return min(max(*l, 1), 200)
}
// hydrate — треки каталога по id; удалённые пропускаются.
func (h *Handler) hydrate(ctx context.Context, ids []uuid.UUID) (map[uuid.UUID]gen.Track, error) {
rows, err := h.catalog.TracksByIDs(ctx, ids)
if err != nil {
return nil, err
}
dtos, err := h.dto().Tracks(ctx, rows)
if err != nil {
return nil, err
}
out := make(map[uuid.UUID]gen.Track, len(dtos))
for _, t := range dtos {
out[t.Id] = t
}
return out, nil
}
// GetHistory реализует gen.StrictServerInterface.
func (h *Handler) GetHistory(ctx context.Context, req gen.GetHistoryRequestObject) (gen.GetHistoryResponseObject, error) {
p, err := principal(ctx)
if err != nil {
return nil, err
}
t, id, ok, err := decodeCursor(req.Params.Cursor)
if err != nil {
return nil, err
}
var before *HistoryRow
if ok {
before = &HistoryRow{ID: id, PlayedAt: t}
}
limit := pageLimit(req.Params.Limit)
rows, err := h.store.History(ctx, p.UserID, before, limit+1)
if err != nil {
return nil, err
}
out := gen.GetHistory200JSONResponse{Items: []gen.HistoryItem{}}
if len(rows) > limit {
rows = rows[:limit]
c := encodeCursor(rows[limit-1].PlayedAt, rows[limit-1].ID)
out.NextCursor = &c
}
ids := make([]uuid.UUID, len(rows))
for i, r := range rows {
ids[i] = r.TrackID
}
tracks, err := h.hydrate(ctx, ids)
if err != nil {
return nil, err
}
for _, r := range rows {
if tr, ok := tracks[r.TrackID]; ok {
out.Items = append(out.Items, gen.HistoryItem{Id: r.ID, PlayedAt: r.PlayedAt, Track: tr})
}
}
return out, nil
}
// ClearHistory реализует gen.StrictServerInterface.
func (h *Handler) ClearHistory(ctx context.Context, _ gen.ClearHistoryRequestObject) (gen.ClearHistoryResponseObject, error) {
p, err := principal(ctx)
if err != nil {
return nil, err
}
if err := h.store.ClearHistory(ctx, p.UserID); err != nil {
return nil, err
}
return gen.ClearHistory204Response{}, nil
}
func (h *Handler) reacted(ctx context.Context, reaction string, limitP *int, cursor *string) (gen.ReactedTrackPage, error) {
p, err := principal(ctx)
if err != nil {
return gen.ReactedTrackPage{}, err
}
t, id, ok, err := decodeCursor(cursor)
if err != nil {
return gen.ReactedTrackPage{}, err
}
var before *Reacted
if ok {
before = &Reacted{TrackID: id, CreatedAt: t}
}
limit := pageLimit(limitP)
rows, total, err := h.store.ReactedPage(ctx, p.UserID, reaction, before, limit+1)
if err != nil {
return gen.ReactedTrackPage{}, err
}
out := gen.ReactedTrackPage{Items: []gen.ReactedTrack{}, Total: total}
if len(rows) > limit {
rows = rows[:limit]
c := encodeCursor(rows[limit-1].CreatedAt, rows[limit-1].TrackID)
out.NextCursor = &c
}
ids := make([]uuid.UUID, len(rows))
for i, r := range rows {
ids[i] = r.TrackID
}
tracks, err := h.hydrate(ctx, ids)
if err != nil {
return gen.ReactedTrackPage{}, err
}
for _, r := range rows {
if tr, ok := tracks[r.TrackID]; ok {
out.Items = append(out.Items, gen.ReactedTrack{CreatedAt: r.CreatedAt, Track: tr})
}
}
return out, nil
}
// GetLikes реализует gen.StrictServerInterface.
func (h *Handler) GetLikes(ctx context.Context, req gen.GetLikesRequestObject) (gen.GetLikesResponseObject, error) {
page, err := h.reacted(ctx, "like", req.Params.Limit, req.Params.Cursor)
if err != nil {
return nil, err
}
return gen.GetLikes200JSONResponse(page), nil
}
// GetDislikes реализует gen.StrictServerInterface.
func (h *Handler) GetDislikes(ctx context.Context, req gen.GetDislikesRequestObject) (gen.GetDislikesResponseObject, error) {
page, err := h.reacted(ctx, "dislike", req.Params.Limit, req.Params.Cursor)
if err != nil {
return nil, err
}
return gen.GetDislikes200JSONResponse(page), nil
}
// SetTrackReaction реализует gen.StrictServerInterface.
func (h *Handler) SetTrackReaction(ctx context.Context, req gen.SetTrackReactionRequestObject) (gen.SetTrackReactionResponseObject, error) {
p, err := principal(ctx)
if err != nil {
return nil, err
}
err = h.store.SetReaction(ctx, p.UserID, req.Id, string(req.Body.Reaction))
if errors.Is(err, ErrNotFound) {
return nil, apperr.NotFound(CodeTrackNotFound, "track not found")
}
if err != nil {
return nil, err
}
return gen.SetTrackReaction204Response{}, nil
}
// ClearTrackReaction реализует gen.StrictServerInterface.
func (h *Handler) ClearTrackReaction(ctx context.Context, req gen.ClearTrackReactionRequestObject) (gen.ClearTrackReactionResponseObject, error) {
p, err := principal(ctx)
if err != nil {
return nil, err
}
if err := h.store.ClearReaction(ctx, p.UserID, req.Id); err != nil {
return nil, err
}
return gen.ClearTrackReaction204Response{}, nil
}
var _ library.Personalizer = (*Handler)(nil)