diff options
author | 2023-11-10 19:29:26 +0100 | |
---|---|---|
committer | 2023-11-10 19:29:26 +0100 | |
commit | ba9d6b467a1f03447789844048d913738c843569 (patch) | |
tree | 5a464ee4a33f26e3284179582ab6d3332d9d5388 /internal/media/processingmedia.go | |
parent | [chore/bugfix/horror] Allow `expires_in` and poll choices to be parsed from s... (diff) | |
download | gotosocial-ba9d6b467a1f03447789844048d913738c843569.tar.xz |
[feature] Media attachment placeholders (#2331)
* [feature] Use placeholders for unknown media types
* fix read of underreported small files
* switch to reduce nesting
* simplify cleanup
Diffstat (limited to 'internal/media/processingmedia.go')
-rw-r--r-- | internal/media/processingmedia.go | 344 |
1 files changed, 246 insertions, 98 deletions
diff --git a/internal/media/processingmedia.go b/internal/media/processingmedia.go index 591cbac74..3627d8cd3 100644 --- a/internal/media/processingmedia.go +++ b/internal/media/processingmedia.go @@ -20,12 +20,12 @@ package media import ( "bytes" "context" - "fmt" + "errors" "image/jpeg" "io" "time" - "codeberg.org/gruf/go-errors/v2" + errorsv2 "codeberg.org/gruf/go-errors/v2" "codeberg.org/gruf/go-runners" "github.com/disintegration/imaging" "github.com/h2non/filetype" @@ -33,11 +33,14 @@ import ( "github.com/superseriousbusiness/gotosocial/internal/gtserror" "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" "github.com/superseriousbusiness/gotosocial/internal/log" + "github.com/superseriousbusiness/gotosocial/internal/storage" "github.com/superseriousbusiness/gotosocial/internal/uris" + "github.com/superseriousbusiness/gotosocial/internal/util" ) -// ProcessingMedia represents a piece of media that is currently being processed. It exposes -// various functions for retrieving data from the process. +// ProcessingMedia represents a piece of media +// currently being processed. It exposes functions +// for retrieving data from the process. type ProcessingMedia struct { media *gtsmodel.MediaAttachment // processing media attachment details dataFn DataFunc // load-data function, returns media stream @@ -48,39 +51,56 @@ type ProcessingMedia struct { mgr *Manager // mgr instance (access to db / storage) } -// AttachmentID returns the ID of the underlying media attachment without blocking processing. +// AttachmentID returns the ID of the underlying +// media attachment without blocking processing. func (p *ProcessingMedia) AttachmentID() string { return p.media.ID // immutable, safe outside mutex. } -// LoadAttachment blocks until the thumbnail and fullsize content has been processed, and then returns the completed attachment. +// LoadAttachment blocks until the thumbnail and +// fullsize content has been processed, and then +// returns the attachment. +// +// If processing could not be completed fully +// then an error will be returned. The attachment +// will still be returned in that case, but it will +// only be partially complete and should be treated +// as a placeholder. func (p *ProcessingMedia) LoadAttachment(ctx context.Context) (*gtsmodel.MediaAttachment, error) { // Attempt to load synchronously. media, done, err := p.load(ctx) - if err == nil { // No issue, return media. return media, nil } if !done { - // Provided context was cancelled, e.g. request cancelled - // early. Queue this item for asynchronous processing. + // Provided context was cancelled, + // e.g. request aborted early before + // its context could be used to finish + // loading the attachment. Enqueue for + // asynchronous processing, which will + // use a background context. log.Warnf(ctx, "reprocessing media %s after canceled ctx", p.media.ID) go p.mgr.state.Workers.Media.Enqueue(p.Process) } - return nil, err + // Media could not be retrieved FULLY, + // but partial attachment should be present. + return media, err } -// Process allows the receiving object to fit the runners.WorkerFunc signature. It performs a (blocking) load and logs on error. +// Process allows the receiving object to fit the +// runners.WorkerFunc signature. It performs a +// (blocking) load and logs on error. func (p *ProcessingMedia) Process(ctx context.Context) { if _, _, err := p.load(ctx); err != nil { - log.Errorf(ctx, "error processing media: %v", err) + log.Errorf(ctx, "error(s) processing media: %v", err) } } -// load performs a concurrency-safe load of ProcessingMedia, only marking itself as complete when returned error is NOT a context cancel. +// load performs a concurrency-safe load of ProcessingMedia, only +// marking itself as complete when returned error is NOT a context cancel. func (p *ProcessingMedia) load(ctx context.Context) (*gtsmodel.MediaAttachment, bool, error) { var ( done bool @@ -95,7 +115,7 @@ func (p *ProcessingMedia) load(ctx context.Context) (*gtsmodel.MediaAttachment, defer func() { // This is only done when ctx NOT cancelled. - done = err == nil || !errors.Comparable(err, + done = err == nil || !errorsv2.Comparable(err, context.Canceled, context.DeadlineExceeded, ) @@ -109,34 +129,61 @@ func (p *ProcessingMedia) load(ctx context.Context) (*gtsmodel.MediaAttachment, p.err = err }() + // Gather errors as we proceed. + var errs = gtserror.NewMultiError(4) + // Attempt to store media and calculate // full-size media attachment details. - if err = p.store(ctx); err != nil { - return err + // + // This will update p.media as it goes. + storeErr := p.store(ctx) + if storeErr != nil { + errs.Append(storeErr) } // Finish processing by reloading media into // memory to get dimension and generate a thumb. - if err = p.finish(ctx); err != nil { - return err + // + // This will update p.media as it goes. + if finishErr := p.finish(ctx); finishErr != nil { + errs.Append(finishErr) + } + + // If this isn't a file we were able to process, + // we may have partially stored it (eg., it's a + // jpeg, which is fine, but streaming it to storage + // was interrupted halfway through and so it was + // never decoded). Try to clean up in this case. + if p.media.Type == gtsmodel.FileTypeUnknown { + deleteErr := p.mgr.state.Storage.Delete(ctx, p.media.File.Path) + if deleteErr != nil && !errors.Is(deleteErr, storage.ErrNotFound) { + errs.Append(deleteErr) + } + } + + var dbErr error + switch { + case !p.recache: + // First time caching this attachment, insert it. + dbErr = p.mgr.state.DB.PutAttachment(ctx, p.media) + + case p.recache && len(errs) == 0: + // Existing attachment we're recaching, update it. + // + // (We only want to update if everything went OK so far, + // otherwise we'd better leave previous version alone.) + dbErr = p.mgr.state.DB.UpdateAttachment(ctx, p.media) } - if p.recache { - // Existing attachment we're recaching, so only update. - err = p.mgr.state.DB.UpdateAttachment(ctx, p.media) - return err + if dbErr != nil { + errs.Append(dbErr) } - // First time caching this attachment, insert it. - err = p.mgr.state.DB.PutAttachment(ctx, p.media) + err = errs.Combine() return err }) - if err != nil { - return nil, done, err - } - - return p.media, done, nil + return p.media, done, err } // store calls the data function attached to p if it hasn't been called yet, @@ -156,17 +203,47 @@ func (p *ProcessingMedia) store(ctx context.Context) error { } }() - // Byte buffer to read file header into. - // See: https://en.wikipedia.org/wiki/File_format#File_header - // and https://github.com/h2non/filetype - hdrBuf := make([]byte, 261) + // Assume we're given correct file + // size, we can overwrite this later + // once we know THE TRUTH. + fileSize := int(sz) + p.media.File.FileSize = fileSize + + // Prepare to read bytes from + // file header or magic number. + hdrBuf := newHdrBuf(fileSize) + + // Read into buffer as much as possible. + // + // UnexpectedEOF means we couldn't read up to the + // given size, but we may still have read something. + // + // EOF means we couldn't read anything at all. + // + // Any other error likely means the connection messed up. + // + // In other words, rather counterintuitively, we + // can only proceed on no error or unexpected error! + n, err := io.ReadFull(rc, hdrBuf) + if err != nil { + if err != io.ErrUnexpectedEOF { + return gtserror.Newf("error reading first bytes of incoming media: %w", err) + } - // Read the first 261 header bytes into buffer as much as possible. - if _, err := rc.Read(hdrBuf); err != nil { - return gtserror.Newf("error reading incoming media: %w", err) + // Initial file size was misreported, so we didn't read + // fully into hdrBuf. Reslice it to the size we did read. + log.Warnf(ctx, + "recovered from misreported file size; reported %d; read %d", + fileSize, n, + ) + hdrBuf = hdrBuf[:n] + fileSize = n + p.media.File.FileSize = fileSize } // Parse file type info from header buffer. + // This should only ever error if the buffer + // is empty (ie., the attachment is 0 bytes). info, err := filetype.Match(hdrBuf) if err != nil { return gtserror.Newf("error parsing file type: %w", err) @@ -175,38 +252,77 @@ func (p *ProcessingMedia) store(ctx context.Context) error { // Recombine header bytes with remaining stream r := io.MultiReader(bytes.NewReader(hdrBuf), rc) + // Assume we'll put + // this file in storage. + store := true + switch info.Extension { case "mp4": - p.media.Type = gtsmodel.FileTypeVideo + // No problem. case "gif": - p.media.Type = gtsmodel.FileTypeImage + // No problem case "jpg", "jpeg", "png", "webp": - p.media.Type = gtsmodel.FileTypeImage - if sz > 0 { - // A file size was provided so we can clean exif data from image. - r, err = terminator.Terminate(r, int(sz), info.Extension) + if fileSize > 0 { + // A file size was provided so we can clean + // exif data from image as we're streaming it. + r, err = terminator.Terminate(r, fileSize, info.Extension) if err != nil { return gtserror.Newf("error cleaning exif data: %w", err) } } default: - return gtserror.Newf("unsupported file type: %s", info.Extension) + // The file is not a supported format that + // we can process, so we can't do much with it. + log.Warnf(ctx, + "media extension '%s' not officially supported, will be processed as "+ + "type '%s' with minimal metadata, and will not be cached locally", + info.Extension, gtsmodel.FileTypeUnknown, + ) + + // Don't bother storing this. + store = false } + // Fill in correct attachment + // data now we're parsed it. + p.media.URL = uris.URIForAttachment( + p.media.AccountID, + string(TypeAttachment), + string(SizeOriginal), + p.media.ID, + info.Extension, + ) + + // Prefer discovered mime type, fall back to + // generic "this contains some bytes" type. + mime := info.MIME.Value + if mime == "" { + mime = "application/octet-stream" + } + p.media.File.ContentType = mime + // Calculate attachment file path. - p.media.File.Path = fmt.Sprintf( - "%s/%s/%s/%s.%s", + p.media.File.Path = uris.StoragePathForAttachment( p.media.AccountID, - TypeAttachment, - SizeOriginal, + string(TypeAttachment), + string(SizeOriginal), p.media.ID, info.Extension, ) - // This shouldn't already exist, but we do a check as it's worth logging. + // We should only try to store the file if it's + // a format we can keep processing, otherwise be + // a bit cheeky: don't store it and let users + // click through to the remote server instead. + if !store { + return nil + } + + // File shouldn't already exist in storage at this point, + // but we do a check as it's worth logging / cleaning up. if have, _ := p.mgr.state.Storage.Has(ctx, p.media.File.Path); have { log.Warnf(ctx, "media already exists at storage path: %s", p.media.File.Path) @@ -216,59 +332,99 @@ func (p *ProcessingMedia) store(ctx context.Context) error { } } - // Write the final image reader stream to our storage. - sz, err = p.mgr.state.Storage.PutStream(ctx, p.media.File.Path, r) + // Write the final reader stream to our storage. + wroteSize, err := p.mgr.state.Storage.PutStream(ctx, p.media.File.Path, r) if err != nil { return gtserror.Newf("error writing media to storage: %w", err) } - // Set written image size. - p.media.File.FileSize = int(sz) + // Set actual written size + // as authoritative file size. + p.media.File.FileSize = int(wroteSize) + + // We can now consider this cached. + p.media.Cached = util.Ptr(true) + + return nil +} - // Fill in remaining attachment data now it's stored. - p.media.URL = uris.GenerateURIForAttachment( +func (p *ProcessingMedia) finish(ctx context.Context) error { + // Make a jolly assumption about thumbnail type. + p.media.Thumbnail.ContentType = mimeImageJpeg + + // Calculate attachment thumbnail file path + p.media.Thumbnail.Path = uris.StoragePathForAttachment( p.media.AccountID, string(TypeAttachment), - string(SizeOriginal), + string(SizeSmall), p.media.ID, - info.Extension, + // Always encode attachment + // thumbnails as jpg. + "jpg", ) - p.media.File.ContentType = info.MIME.Value - p.media.Cached = func() *bool { - ok := true - return &ok - }() - return nil -} + // Calculate attachment thumbnail serve path. + p.media.Thumbnail.URL = uris.URIForAttachment( + p.media.AccountID, + string(TypeAttachment), + string(SizeSmall), + p.media.ID, + // Always encode attachment + // thumbnails as jpg. + "jpg", + ) -func (p *ProcessingMedia) finish(ctx context.Context) error { - // Fetch a stream to the original file in storage. + // If original file hasn't been stored, there's + // likely something wrong with the data, or we + // don't want to store it. Skip everything else. + if !*p.media.Cached { + p.media.Processing = gtsmodel.ProcessingStatusProcessed + return nil + } + + // Get a stream to the original file for further processing. rc, err := p.mgr.state.Storage.GetStream(ctx, p.media.File.Path) if err != nil { return gtserror.Newf("error loading file from storage: %w", err) } defer rc.Close() + // fullImg is the processed version of + // the original (stripped + reoriented). var fullImg *gtsImage + // Depending on the content type, we + // can do various types of decoding. switch p.media.File.ContentType { + // .jpeg, .gif, .webp image type case mimeImageJpeg, mimeImageGif, mimeImageWebp: - fullImg, err = decodeImage(rc, imaging.AutoOrientation(true)) + fullImg, err = decodeImage( + rc, + imaging.AutoOrientation(true), + ) if err != nil { return gtserror.Newf("error decoding image: %w", err) } + // Mark as no longer unknown type now + // we know for sure we can decode it. + p.media.Type = gtsmodel.FileTypeImage + // .png image (requires ancillary chunk stripping) case mimeImagePng: - fullImg, err = decodeImage(&pngAncillaryChunkStripper{ - Reader: rc, - }, imaging.AutoOrientation(true)) + fullImg, err = decodeImage( + &pngAncillaryChunkStripper{Reader: rc}, + imaging.AutoOrientation(true), + ) if err != nil { return gtserror.Newf("error decoding image: %w", err) } + // Mark as no longer unknown type now + // we know for sure we can decode it. + p.media.Type = gtsmodel.FileTypeImage + // .mp4 video type case mimeVideoMp4: video, err := decodeVideoFrame(rc) @@ -283,9 +439,14 @@ func (p *ProcessingMedia) finish(ctx context.Context) error { p.media.FileMeta.Original.Duration = &video.duration p.media.FileMeta.Original.Framerate = &video.framerate p.media.FileMeta.Original.Bitrate = &video.bitrate + + // Mark as no longer unknown type now + // we know for sure we can decode it. + p.media.Type = gtsmodel.FileTypeVideo } - // The image should be in-memory by now. + // fullImg should be in-memory by + // now so we're done with storage. if err := rc.Close(); err != nil { return gtserror.Newf("error closing file: %w", err) } @@ -296,15 +457,6 @@ func (p *ProcessingMedia) finish(ctx context.Context) error { p.media.FileMeta.Original.Size = int(fullImg.Size()) p.media.FileMeta.Original.Aspect = fullImg.AspectRatio() - // Calculate attachment thumbnail file path - p.media.Thumbnail.Path = fmt.Sprintf( - "%s/%s/%s/%s.jpg", - p.media.AccountID, - TypeAttachment, - SizeSmall, - p.media.ID, - ) - // Get smaller thumbnail image thumbImg := fullImg.Thumbnail() @@ -312,16 +464,20 @@ func (p *ProcessingMedia) finish(ctx context.Context) error { // now take our large son. fullImg = nil - // Blurhash needs generating from thumb. - hash, err := thumbImg.Blurhash() - if err != nil { - return gtserror.Newf("error generating blurhash: %w", err) - } + // Only generate blurhash + // from thumb if necessary. + if p.media.Blurhash == "" { + hash, err := thumbImg.Blurhash() + if err != nil { + return gtserror.Newf("error generating blurhash: %w", err) + } - // Set the attachment blurhash. - p.media.Blurhash = hash + // Set the attachment blurhash. + p.media.Blurhash = hash + } - // This shouldn't already exist, but we do a check as it's worth logging. + // Thumbnail shouldn't already exist in storage at this point, + // but we do a check as it's worth logging / cleaning up. if have, _ := p.mgr.state.Storage.Has(ctx, p.media.Thumbnail.Path); have { log.Warnf(ctx, "thumbnail already exists at storage path: %s", p.media.Thumbnail.Path) @@ -333,7 +489,9 @@ func (p *ProcessingMedia) finish(ctx context.Context) error { // Create a thumbnail JPEG encoder stream. enc := thumbImg.ToJPEG(&jpeg.Options{ - Quality: 70, // enough for a thumbnail. + // Good enough for + // a thumbnail. + Quality: 70, }) // Stream-encode the JPEG thumbnail image into storage. @@ -342,16 +500,6 @@ func (p *ProcessingMedia) finish(ctx context.Context) error { return gtserror.Newf("error stream-encoding thumbnail to storage: %w", err) } - // Fill in remaining thumbnail now it's stored - p.media.Thumbnail.ContentType = mimeImageJpeg - p.media.Thumbnail.URL = uris.GenerateURIForAttachment( - p.media.AccountID, - string(TypeAttachment), - string(SizeSmall), - p.media.ID, - "jpg", // always jpeg - ) - // Set thumbnail dimensions in attachment info. p.media.FileMeta.Small = gtsmodel.Small{ Width: int(thumbImg.Width()), |