GoToSocial/internal/media/processingmedia.go

// GoToSocial
// Copyright (C) GoToSocial Authors admin@gotosocial.org
// SPDX-License-Identifier: AGPL-3.0-or-later
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.

package media

import (
	"context"
	"time"

	errorsv2 "codeberg.org/gruf/go-errors/v2"
	"codeberg.org/gruf/go-runners"

	"github.com/superseriousbusiness/gotosocial/internal/gtscontext"
	"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
// 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
	done   bool                      // done is set when process finishes with non ctx canceled type error
	proc   runners.Processor         // proc helps synchronize only a singular running processing instance
	err    error                     // error stores permanent error value when done
	mgr    *Manager                  // mgr instance (access to db / storage)
}

// ID returns the ID of the underlying media.
func (p *ProcessingMedia) ID() string {
	return p.media.ID // immutable, safe outside mutex.
}

// 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) Load(ctx context.Context) (*gtsmodel.MediaAttachment, error) {
	media, done, err := p.load(ctx)
	if !done {
		// On a context-canceled error (marked as !done), requeue for loading.
		p.mgr.state.Workers.Dereference.Queue.Push(func(ctx context.Context) {
			if _, _, err := p.load(ctx); err != nil {
				log.Errorf(ctx, "error loading media: %v", err)
			}
		})
	}
	return media, err
}

// load is the package private form of load() that is wrapped to catch context canceled.
func (p *ProcessingMedia) load(ctx context.Context) (
	media *gtsmodel.MediaAttachment,
	done bool,
	err error,
) {
	err = p.proc.Process(func() error {
		if done = p.done; done {
			// Already proc'd.
			return p.err
		}

		defer func() {
			// This is only done when ctx NOT cancelled.
			done = (err == nil || !errorsv2.IsV2(err,
				context.Canceled,
				context.DeadlineExceeded,
			))

			if !done {
				return
			}

			// Anything from here, we
			// need to ensure happens
			// (i.e. no ctx canceled).
			ctx = gtscontext.WithValues(
				context.Background(),
				ctx, // values
			)

			// On error or unknown media types, perform error cleanup.
			if err != nil || p.media.Type == gtsmodel.FileTypeUnknown {
				p.cleanup(ctx)
			}

			// Update with latest details, whatever happened.
			e := p.mgr.state.DB.UpdateAttachment(ctx, p.media)
			if e != nil {
				log.Errorf(ctx, "error updating media in db: %v", e)
			}

			// Store final values.
			p.done = true
			p.err = err
		}()

		// TODO: in time update this
		// to perhaps follow a similar
		// freshness window to statuses
		// / accounts? But that's a big
		// maybe, media don't change in
		// the same way so this is largely
		// just to slow down fail retries.
		const maxfreq = 6 * time.Hour

		// Check whether media is uncached but repeatedly failing,
		// specifically limit the frequency at which we allow this.
		if !p.media.UpdatedAt.Equal(p.media.CreatedAt) && // i.e. not new
			p.media.UpdatedAt.Add(maxfreq).Before(time.Now()) {
			return nil
		}

		// Attempt to store media and calculate
		// full-size media attachment details.
		//
		// This will update p.media as it goes.
		err = p.store(ctx)
		return err
	})
	media = p.media
	return
}

// store calls the data function attached to p if it hasn't been called yet,
// and updates the underlying attachment fields as necessary. It will then stream
// bytes from p's reader directly into storage so that it can be retrieved later.
func (p *ProcessingMedia) store(ctx context.Context) error {
	// Load media from data func.
	rc, err := p.dataFn(ctx)
	if err != nil {
		return gtserror.Newf("error executing data function: %w", err)
	}

	var (
		// predfine temporary media
		// file path variables so we
		// can remove them on error.
		temppath  string
		thumbpath string
	)

	defer func() {
		if err := remove(temppath, thumbpath); err != nil {
			log.Errorf(ctx, "error(s) cleaning up files: %v", err)
		}
	}()

	// Drain reader to tmp file
	// (this reader handles close).
	temppath, err = drainToTmp(rc)
	if err != nil {
		return gtserror.Newf("error draining data to tmp: %w", err)
	}

	// Pass input file through ffprobe to
	// parse further metadata information.
	result, err := ffprobe(ctx, temppath)
	if err != nil {
		return gtserror.Newf("error ffprobing data: %w", err)
	}

	switch {
	// No errors parsing data.
	case result.Error == nil:

	// Data type unhandleable by ffprobe.
	case result.Error.Code == -1094995529:
		log.Warn(ctx, "unsupported data type")
		return nil

	default:
		return gtserror.Newf("ffprobe error: %w", err)
	}

	var ext string

	// Set the media type from ffprobe format data.
	p.media.Type, ext = result.Format.GetFileType()
	if p.media.Type == gtsmodel.FileTypeUnknown {

		// Return early (deleting file)
		// for unhandled file types.
		return nil
	}

	switch p.media.Type {
	case gtsmodel.FileTypeImage:
		// Pass file through ffmpeg clearing
		// any excess metadata (e.g. EXIF).
		if err := ffmpegClearMetadata(ctx,
			temppath, ext,
		); err != nil {
			return gtserror.Newf("error cleaning metadata: %w", err)
		}

		// Extract image metadata from streams.
		width, height, err := result.ImageMeta()
		if err != nil {
			return err
		}
		p.media.FileMeta.Original.Width = width
		p.media.FileMeta.Original.Height = height
		p.media.FileMeta.Original.Size = (width * height)
		p.media.FileMeta.Original.Aspect = float32(width) / float32(height)

		// Determine thumbnail dimensions to use.
		thumbWidth, thumbHeight := thumbSize(width, height)
		p.media.FileMeta.Small.Width = thumbWidth
		p.media.FileMeta.Small.Height = thumbHeight
		p.media.FileMeta.Small.Size = (thumbWidth * thumbHeight)
		p.media.FileMeta.Small.Aspect = float32(thumbWidth) / float32(thumbHeight)

		// Generate a thumbnail image from input image path.
		thumbpath, err = ffmpegGenerateThumb(ctx, temppath,
			thumbWidth,
			thumbHeight,
		)
		if err != nil {
			return gtserror.Newf("error generating image thumb: %w", err)
		}

	case gtsmodel.FileTypeVideo:
		// Pass file through ffmpeg clearing
		// any excess metadata (e.g. EXIF).
		if err := ffmpegClearMetadata(ctx,
			temppath, ext,
		); err != nil {
			return gtserror.Newf("error cleaning metadata: %w", err)
		}

		// Extract video metadata we can from streams.
		width, height, framerate, err := result.VideoMeta()
		if err != nil {
			return err
		}
		p.media.FileMeta.Original.Width = width
		p.media.FileMeta.Original.Height = height
		p.media.FileMeta.Original.Size = (width * height)
		p.media.FileMeta.Original.Aspect = float32(width) / float32(height)
		p.media.FileMeta.Original.Framerate = &framerate

		// Extract total duration from format.
		duration := result.Format.GetDuration()
		p.media.FileMeta.Original.Duration = &duration

		// Extract total bitrate from format.
		bitrate := result.Format.GetBitRate()
		p.media.FileMeta.Original.Bitrate = &bitrate

		// Determine thumbnail dimensions to use.
		thumbWidth, thumbHeight := thumbSize(width, height)
		p.media.FileMeta.Small.Width = thumbWidth
		p.media.FileMeta.Small.Height = thumbHeight
		p.media.FileMeta.Small.Size = (thumbWidth * thumbHeight)
		p.media.FileMeta.Small.Aspect = float32(thumbWidth) / float32(thumbHeight)

		// Extract a thumbnail frame from input video path.
		thumbpath, err = ffmpegGenerateThumb(ctx, temppath,
			thumbWidth,
			thumbHeight,
		)
		if err != nil {
			return gtserror.Newf("error extracting video frame: %w", err)
		}

	case gtsmodel.FileTypeAudio:
		// Extract total duration from format.
		duration := result.Format.GetDuration()
		p.media.FileMeta.Original.Duration = &duration

		// Extract total bitrate from format.
		bitrate := result.Format.GetBitRate()
		p.media.FileMeta.Original.Bitrate = &bitrate

		// Extract image metadata from streams (if any),
		// this will only exist for embedded album art.
		width, height, framerate, _ := result.EmbeddedImageMeta()
		if width > 0 && height > 0 {
			// Unlikely to need these but masto API includes them.
			p.media.FileMeta.Original.Width = width
			p.media.FileMeta.Original.Height = height
			if framerate != 0 {
				p.media.FileMeta.Original.Framerate = &framerate
			}

			// Determine thumbnail dimensions to use.
			thumbWidth, thumbHeight := thumbSize(width, height)
			p.media.FileMeta.Small.Width = thumbWidth
			p.media.FileMeta.Small.Height = thumbHeight
			p.media.FileMeta.Small.Size = (thumbWidth * thumbHeight)
			p.media.FileMeta.Small.Aspect = float32(thumbWidth) / float32(thumbHeight)

			// Generate a thumbnail image from input image path.
			thumbpath, err = ffmpegGenerateThumb(ctx, temppath,
				thumbWidth,
				thumbHeight,
			)
			if err != nil {
				return gtserror.Newf("error generating image thumb: %w", err)
			}
		}

	default:
		log.Warnf(ctx, "unsupported type: %s (%s)", p.media.Type, result.Format.FormatName)
		return nil
	}

	// Calculate final media attachment file path.
	p.media.File.Path = uris.StoragePathForAttachment(
		p.media.AccountID,
		string(TypeAttachment),
		string(SizeOriginal),
		p.media.ID,
		ext,
	)

	// Copy temporary file into storage at path.
	filesz, err := p.mgr.state.Storage.PutFile(ctx,
		p.media.File.Path,
		temppath,
	)
	if err != nil {
		return gtserror.Newf("error writing media to storage: %w", err)
	}

	// Set final determined file size.
	p.media.File.FileSize = int(filesz)

	if thumbpath != "" {
		// Note that neither thumbnail storage
		// nor a blurhash are needed for audio.

		if p.media.Blurhash == "" {
			// Generate blurhash (if not already) from thumbnail.
			p.media.Blurhash, err = generateBlurhash(thumbpath)
			if err != nil {
				return gtserror.Newf("error generating thumb blurhash: %w", err)
			}
		}

		// Copy thumbnail file into storage at path.
		thumbsz, err := p.mgr.state.Storage.PutFile(ctx,
			p.media.Thumbnail.Path,
			thumbpath,
		)
		if err != nil {
			return gtserror.Newf("error writing thumb to storage: %w", err)
		}

		// Set final determined thumbnail size.
		p.media.Thumbnail.FileSize = int(thumbsz)
	}

	// Fill in correct attachment
	// data now we've parsed it.
	p.media.URL = uris.URIForAttachment(
		p.media.AccountID,
		string(TypeAttachment),
		string(SizeOriginal),
		p.media.ID,
		ext,
	)

	// Get mimetype for the file container
	// type, falling back to generic data.
	p.media.File.ContentType = getMimeType(ext)

	// We can now consider this cached.
	p.media.Cached = util.Ptr(true)

	// Finally set the attachment as finished processing.
	p.media.Processing = gtsmodel.ProcessingStatusProcessed

	return nil
}

// cleanup will remove any traces of processing media from storage.
// and perform any other necessary cleanup steps after failure.
func (p *ProcessingMedia) cleanup(ctx context.Context) {
	var err error

	if p.media.File.Path != "" {
		// Ensure media file at path is deleted from storage.
		err = p.mgr.state.Storage.Delete(ctx, p.media.File.Path)
		if err != nil && !storage.IsNotFound(err) {
			log.Errorf(ctx, "error deleting %s: %v", p.media.File.Path, err)
		}
	}

	if p.media.Thumbnail.Path != "" {
		// Ensure media thumbnail at path is deleted from storage.
		err = p.mgr.state.Storage.Delete(ctx, p.media.Thumbnail.Path)
		if err != nil && !storage.IsNotFound(err) {
			log.Errorf(ctx, "error deleting %s: %v", p.media.Thumbnail.Path, err)
		}
	}

	// Also ensure marked as unknown and finished
	// processing so gets inserted as placeholder URL.
	p.media.Processing = gtsmodel.ProcessingStatusProcessed
	p.media.Type = gtsmodel.FileTypeUnknown
	p.media.Cached = util.Ptr(false)
}