Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
13 kB · 410 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411// Package upload implements resumable user uploads of arbitrary media// content using the TUS protocol. Uploads are pre-created by the// place.stream.media.createUpload XRPC (DPoP-authenticated) which returns a// URL and a short-lived bearer token; clients then PATCH/HEAD chunks against// the URL with only the bearer token, avoiding per-chunk DPoP.//// Two storage backends are supported. When S3 is configured, uploads are// streamed directly into an S3 multipart upload, which lets any node in a// station resume any upload. Otherwise uploads land on local disk under// $DataDir/uploads — that mode requires a single-node deployment.package upload
import ( "context" "errors" "fmt" "net/http" "os" "strings" "time"
"github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/credentials" awss3 "github.com/aws/aws-sdk-go-v2/service/s3" "github.com/google/uuid" "github.com/lestrrat-go/jwx/v2/jwa" "github.com/lestrrat-go/jwx/v2/jwt" tusfilestore "github.com/tus/tusd/v2/pkg/filestore" tushandler "github.com/tus/tusd/v2/pkg/handler" tusmemlocker "github.com/tus/tusd/v2/pkg/memorylocker" tuss3store "github.com/tus/tusd/v2/pkg/s3store"
"stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/pkg/statedb")
// Backend identifies which TUS data store backs an upload.type Backend string
const ( BackendFile Backend = "file" BackendS3 Backend = "s3")
// TokenLifetime is how long an upload bearer token remains valid. Renewed// uploads (long uploads, paused & resumed) will need a fresh token via// createUpload — we deliberately keep this finite to limit token leak blast// radius.const TokenLifetime = 24 * time.Hour
// MaxUploadSize caps the upload size at the protocol layer. Set to zero to// disable enforcement. Currently 50 GiB.const MaxUploadSize int64 = 50 * 1024 * 1024 * 1024
// BasePath is the URL prefix the TUS handler is mounted at. Token claims// embed the upload ID; URLs returned to clients are constructed from this// prefix plus the assigned ID.const BasePath = "/api/upload/"
const tokenAudience = "streamplace-upload"
// Manager owns the TUS store, the unrouted handler, and the bearer-token// signing key. It's safe to use as a singleton across the lifetime of the// server.type Manager struct { cli *config.CLI state *statedb.StatefulDB composer *tushandler.StoreComposer handler *tushandler.UnroutedHandler backend Backend authKey []byte}
// New constructs an upload Manager, choosing the storage backend based on// the CLI's S3 configuration. Returns an error when configured for a// multi-node station without S3 — file-backed uploads cannot survive being// served by a different node than the one that issued the upload URL.func New(ctx context.Context, cli *config.CLI, state *statedb.StatefulDB) (*Manager, error) { authKey, err := state.GetOrCreateUploadAuthKey() if err != nil { return nil, fmt.Errorf("upload: get auth key: %w", err) }
multiNode := cli.ServerHost != "" && cli.BroadcasterHost != "" && cli.ServerHost != cli.BroadcasterHost if multiNode && !cli.S3Configured() { return nil, errors.New("upload: multi-node deployments require S3 storage; file-backed uploads are single-node only") }
composer := tushandler.NewStoreComposer() tusmemlocker.New().UseIn(composer)
var backend Backend if cli.S3Configured() { backend = BackendS3 s3client := awss3.New(awss3.Options{ Region: cli.S3Region, Credentials: credentials.NewStaticCredentialsProvider( cli.S3AccessKeyID, cli.S3SecretAccessKey, "", ), BaseEndpoint: aws.String(cli.S3Endpoint), UsePathStyle: true, }) store := tuss3store.New(cli.S3Bucket, s3client) store.ObjectPrefix = "uploads/" store.UseIn(composer) log.Log(ctx, "upload manager: S3 backend", "bucket", cli.S3Bucket) } else { backend = BackendFile dir := cli.DataFilePath([]string{"uploads"}) if err := os.MkdirAll(dir, 0o755); err != nil { return nil, fmt.Errorf("upload: create upload dir %q: %w", dir, err) } store := tusfilestore.New(dir) store.UseIn(composer) log.Log(ctx, "upload manager: file backend", "dir", dir) }
cfg := tushandler.Config{ StoreComposer: composer, BasePath: BasePath, MaxSize: MaxUploadSize, NotifyCompleteUploads: true, // We pre-create uploads from the createUpload XRPC, so the protocol's // POST path is unused. Disabling download/termination keeps the // surface small; if we need cancellation we can flip DisableTermination. DisableDownload: true, DisableTermination: false, } h, err := tushandler.NewUnroutedHandler(cfg) if err != nil { return nil, fmt.Errorf("upload: create tusd handler: %w", err) }
m := &Manager{ cli: cli, state: state, composer: composer, handler: h, backend: backend, authKey: authKey, }
go m.notifyLoop(ctx) return m, nil}
// CreateResult is what the createUpload XRPC returns to the caller, modulo// JSON field naming.type CreateResult struct { UploadID string UploadURL string UploadToken string ExpiresAt time.Time}
// Create pre-creates a TUS upload for the given user and mints a bearer// token bound to it. baseURL should be the absolute scheme+host the caller// reached (e.g. "https://example.com") so the returned UploadURL is// reachable for the client.func (m *Manager) Create(ctx context.Context, did string, mimeType string, filename string, size int64, baseURL string) (*CreateResult, error) { if size <= 0 { return nil, errors.New("upload: size must be positive") } if MaxUploadSize > 0 && size > MaxUploadSize { return nil, fmt.Errorf("upload: size %d exceeds max %d", size, MaxUploadSize) }
id, err := uuid.NewV7() if err != nil { return nil, fmt.Errorf("upload: generate id: %w", err) }
meta := tushandler.MetaData{ "did": did, "mimeType": mimeType, } if filename != "" { meta["filename"] = filename } info := tushandler.FileInfo{ ID: id.String(), Size: size, MetaData: meta, }
upload, err := m.composer.Core.NewUpload(ctx, info) if err != nil { return nil, fmt.Errorf("upload: create tusd upload: %w", err) } final, err := upload.GetInfo(ctx) if err != nil { return nil, fmt.Errorf("upload: read back upload info: %w", err) }
row := &statedb.Upload{ ID: final.ID, RepoDID: did, MimeType: mimeType, Filename: filename, Size: size, Backend: string(m.backend), } if final.Storage != nil { // Filestore stores "Path"; s3store stores "Bucket"+"Key". if p, ok := final.Storage["Path"]; ok { row.Location = p } else if k, ok := final.Storage["Key"]; ok { row.Location = "s3://" + final.Storage["Bucket"] + "/" + k } } if err := m.state.CreateUpload(ctx, row); err != nil { return nil, fmt.Errorf("upload: persist upload row: %w", err) }
expiresAt := time.Now().Add(TokenLifetime) tok, err := m.mintToken(did, final.ID, expiresAt) if err != nil { return nil, fmt.Errorf("upload: mint token: %w", err) }
url := strings.TrimRight(baseURL, "/") + BasePath + final.ID return &CreateResult{ UploadID: final.ID, UploadURL: url, UploadToken: tok, ExpiresAt: expiresAt, }, nil}
// ServeHTTP implements http.Handler for TUS requests. The caller is// responsible for routing TUS-protocol methods (HEAD/PATCH/DELETE) to this// handler at paths under BasePath. POST is rejected — uploads are// pre-created exclusively via createUpload.func (m *Manager) ServeHTTP(w http.ResponseWriter, r *http.Request) { if r.Method == http.MethodPost { http.Error(w, "POST not supported; call place.stream.media.createUpload to start an upload", http.StatusMethodNotAllowed) return }
id := strings.TrimPrefix(r.URL.Path, BasePath) id = strings.Trim(id, "/") if id == "" { http.Error(w, "missing upload id in path", http.StatusBadRequest) return }
if err := m.checkAuth(r, id); err != nil { log.Warn(r.Context(), "upload auth rejected", "error", err, "uploadId", id) http.Error(w, "unauthorized", http.StatusUnauthorized) return }
// tusd's handler reads the upload ID via strings.Trim(r.URL.Path, "/"), // so we need to strip BasePath before delegating. Cloning the request is // safer than mutating r.URL in-place. r2 := r.Clone(r.Context()) r2.URL.Path = "/" + id
wrapped := m.handler.Middleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodHead: m.handler.HeadFile(w, r) case http.MethodPatch: m.handler.PatchFile(w, r) case http.MethodDelete: m.handler.DelFile(w, r) case http.MethodOptions: // Middleware handles CORS; just terminate. w.WriteHeader(http.StatusOK) default: w.Header().Set("Allow", "HEAD, PATCH, DELETE, OPTIONS") w.WriteHeader(http.StatusMethodNotAllowed) } })) wrapped.ServeHTTP(w, r2)}
// notifyLoop consumes upload completion events from tusd's hook channel and// enqueues a VOD processing task for each finished upload.func (m *Manager) notifyLoop(ctx context.Context) { for { select { case <-ctx.Done(): return case ev := <-m.handler.CompleteUploads: m.onComplete(ctx, ev) } }}
func (m *Manager) onComplete(ctx context.Context, ev tushandler.HookEvent) { id := ev.Upload.ID // The createUpload XRPC stashed the owner's DID in the TUS metadata; // thread it into the log context so every downstream log line (complete, // enqueue, draft creation) carries the user this upload belongs to. did := "" if ev.Upload.MetaData != nil { did = ev.Upload.MetaData["did"] } ctx = log.WithLogValues(ctx, "uploadId", id, "did", did) log.Log(ctx, "upload complete", "size", ev.Upload.Size)
location := "" if ev.Upload.Storage != nil { if p, ok := ev.Upload.Storage["Path"]; ok { location = p } else if k, ok := ev.Upload.Storage["Key"]; ok { location = "s3://" + ev.Upload.Storage["Bucket"] + "/" + k } } if err := m.state.CompleteUpload(ctx, id, location); err != nil { log.Error(ctx, "failed to mark upload complete", "error", err) }
row, err := m.state.GetUpload(ctx, id) if err != nil || row == nil { log.Error(ctx, "failed to look up completed upload", "error", err) return } payload := statedb.VODProcessTask{ UploadID: row.ID, RepoDID: row.RepoDID, MimeType: row.MimeType, Filename: row.Filename, Size: row.Size, Backend: row.Backend, Location: row.Location, } if _, err := m.state.EnqueueTask(ctx, statedb.TaskVODProcess, payload, statedb.WithTaskKey("vod-process:"+row.ID)); err != nil { log.Error(ctx, "failed to enqueue vod-process task", "error", err) }
// Create a draft VOD in the 'processing' state, but only if the upload // isn't already associated with a pre-created draft. The modern client // flow creates the draft up front (place.stream.vod.createDraft) and passes // its URI to createUpload, which re-points the draft's origin_upload_id at // this upload — so a draft already exists for this row. Only fall back to // creating one here for legacy clients / external callers that didn't // supply a draftUri. A failed create is non-fatal: the upload still // processes. if existing, _ := m.state.GetDraftByUpload(ctx, row.ID); existing == nil { if _, err := m.state.CreateDraft(ctx, row.RepoDID, row.ID, &placestream.VodDraftVideo{ LexiconTypeID: "place.stream.vod.draftVideo", Title: filenameOrDefault(row.Filename), Status: "processing", CreatedAt: time.Now().UTC().Format(time.RFC3339), }); err != nil { log.Error(ctx, "failed to create draft for upload", "error", err) } }}
// filenameOrDefault returns a human placeholder title for a freshly-uploaded// draft: the original filename if present, else a generic "Uploaded video".// The user edits this before publishing.func filenameOrDefault(filename string) string { if filename != "" { return filename } return "Uploaded video"}
func (m *Manager) mintToken(did, uploadID string, expiresAt time.Time) (string, error) { now := time.Now() tok, err := jwt.NewBuilder(). Issuer("streamplace"). Audience([]string{tokenAudience}). Subject(did). IssuedAt(now). Expiration(expiresAt). Claim("upload_id", uploadID). Build() if err != nil { return "", err } signed, err := jwt.Sign(tok, jwt.WithKey(jwa.HS256, m.authKey)) if err != nil { return "", err } return string(signed), nil}
func (m *Manager) checkAuth(r *http.Request, uploadID string) error { auth := r.Header.Get("Authorization") if !strings.HasPrefix(auth, "Bearer ") { return errors.New("missing bearer token") } raw := strings.TrimPrefix(auth, "Bearer ") tok, err := jwt.Parse([]byte(raw), jwt.WithKey(jwa.HS256, m.authKey), jwt.WithValidate(true), jwt.WithAudience(tokenAudience), ) if err != nil { return fmt.Errorf("parse token: %w", err) } v, ok := tok.Get("upload_id") if !ok { return errors.New("token missing upload_id claim") } claimID, ok := v.(string) if !ok || claimID != uploadID { return errors.New("token upload_id does not match request") } return nil}