internal/entity/entity_counts.go - develop - githubmirror/photoprism

internal/entity/entity_counts.go
385 строк13 KB
Keith Martin
Entity: Ensure that UpdateSubjectCounts returns an error if encounted in 2nd SQLite update
30 июл 2026, 17:03
30 июл 2026, 17:037b668f0
package entity

import (
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/dustin/go-humanize/english"
	"github.com/jinzhu/gorm"

	"github.com/photoprism/photoprism/internal/mutex"
	"github.com/photoprism/photoprism/pkg/clean"
	"github.com/photoprism/photoprism/pkg/dsn"
	"github.com/photoprism/photoprism/pkg/time/unix"
)

// countsBusy indicates whether count refresh jobs are currently running.
var countsBusy = atomic.Bool{}

// asyncWG tracks in-flight goroutines spawned by entity package async
// helpers (UpdateCountsAsync and the cover-update goroutine in
// internal/entity/query). Callers that intend to tear down the database
// connection — most importantly config.CloseDb during test shutdown —
// should invoke WaitForAsyncJobs first to avoid a nil-DB race in the
// running goroutines.
var asyncWG sync.WaitGroup

// AsyncJobAdd registers a new in-flight async job spawned by the entity
// package or its sub-packages. It must be paired with AsyncJobDone in a
// deferred call so the WaitForAsyncJobs helper can drain reliably.
func AsyncJobAdd() {
	asyncWG.Add(1)
}

// AsyncJobDone marks an in-flight async job as finished. Always pair it
// with a preceding AsyncJobAdd via defer so panics in the goroutine still
// release the WaitGroup.
func AsyncJobDone() {
	asyncWG.Done()
}

// WaitForAsyncJobs blocks until every async job started via the package's
// async helpers has finished. Callers that need to safely tear down the
// database connection should invoke this before nilling the provider.
func WaitForAsyncJobs() {
	asyncWG.Wait()
}

// WaitForAsyncJobsTimeout waits up to timeout for async jobs to finish and reports
// whether they all drained; a non-positive timeout waits indefinitely. It bounds
// database teardown so a wedged background job cannot hang shutdown forever.
func WaitForAsyncJobsTimeout(timeout time.Duration) bool {
	if timeout <= 0 {
		asyncWG.Wait()
		return true
	}

	done := make(chan struct{})

	go func() {
		asyncWG.Wait()
		close(done)
	}()

	select {
	case <-done:
		return true
	case <-time.After(timeout):
		return false
	}
}

type LabelPhotoCount struct {
	LabelID    int
	PhotoCount int
}

// LabelPhotoCounts groups label count results.
type LabelPhotoCounts []LabelPhotoCount

// LabelCounts returns the number of public, non-deleted photos for each label ID.
func LabelCounts() LabelPhotoCounts {
	result := LabelPhotoCounts{}

	if err := UnscopedDb().Raw(`
		SELECT label_id, SUM(photo_count) AS photo_count FROM (
			SELECT l.id AS label_id, COUNT(*) AS photo_count FROM labels l
		JOIN photos_labels pl ON pl.label_id = l.id
		JOIN photos ph ON pl.photo_id = ph.id
		WHERE pl.uncertainty < 100
		AND ph.photo_quality > -1
		AND ph.photo_private = 0
		AND ph.deleted_at IS NULL GROUP BY l.id
		UNION ALL
		SELECT l.id AS label_id, COUNT(*) AS photo_count FROM labels l
		JOIN categories c ON c.category_id = l.id
		JOIN photos_labels pl ON pl.label_id = c.label_id
		JOIN photos ph ON pl.photo_id = ph.id
		WHERE pl.uncertainty < 100
		AND ph.photo_quality > -1
		AND ph.photo_private = 0
		AND ph.deleted_at IS NULL GROUP BY l.id) counts GROUP BY label_id
		`).Scan(&result).Error; err != nil {
		log.Errorf("label-count: %s", err.Error())
	}

	return result
}

// UpdatePlacesCounts updates the places photo counts.
func UpdatePlacesCounts() (err error) {
	mutex.Index.Lock()
	defer mutex.Index.Unlock()

	start := time.Now()

	// Update places.
	res := Db().Table("places").
		UpdateColumn("photo_count", gorm.Expr("(SELECT COUNT(*) FROM photos p "+
			"WHERE places.id = p.place_id "+
			"AND p.photo_quality > -1 "+
			"AND p.photo_private = 0 "+
			"AND p.deleted_at IS NULL)"))

	if res.Error != nil {
		return res.Error
	}

	log.Debugf("counts: updated %s [%s]", english.Plural(int(res.RowsAffected), "place", "places"), time.Since(start))

	return nil
}

// UpdateSubjectCounts updates the subject file counts.
func UpdateSubjectCounts(public bool) (err error) {
	mutex.Index.Lock()
	defer mutex.Index.Unlock()

	start := time.Now()

	var res *gorm.DB

	subjTable := Subject{}.TableName()

	var photosJoin *gorm.SqlExpr

	// Count people tagged on private pictures?
	// see https://github.com/photoprism/photoprism/issues/4238
	// and https://github.com/photoprism/photoprism/issues/2570#issuecomment-1231690056
	if public {
		photosJoin = gorm.Expr("p.id = f.photo_id AND p.deleted_at IS NULL AND p.photo_private = 0")
	} else {
		photosJoin = gorm.Expr("p.id = f.photo_id AND p.deleted_at IS NULL")
	}

	condition := gorm.Expr("subj_type = ?", SubjPerson)

	switch DbDialect() {
	case dsn.DriverMySQL:
		res = Db().Exec(`UPDATE ? LEFT JOIN (
		SELECT m.subj_uid, COUNT(DISTINCT f.id) AS subj_files, COUNT(DISTINCT f.photo_id) AS subj_photos
			FROM files f
			JOIN photos p ON ?			    
			JOIN markers m ON f.file_uid = m.file_uid AND m.subj_uid IS NOT NULL AND m.subj_uid <> '' AND m.subj_uid IS NOT NULL
			WHERE m.marker_invalid = 0 AND f.deleted_at IS NULL GROUP BY m.subj_uid
		) b ON b.subj_uid = subjects.subj_uid
		SET subjects.file_count = CASE WHEN b.subj_files IS NULL THEN 0 ELSE b.subj_files END, 
			subjects.photo_count = CASE WHEN b.subj_photos IS NULL THEN 0 ELSE b.subj_photos END
		WHERE ?`, gorm.Expr(subjTable), photosJoin, condition)
	case dsn.DriverSQLite3:
		// Update files count.
		res = Db().Table(subjTable).
			UpdateColumn("file_count", gorm.Expr("(SELECT COUNT(DISTINCT f.id)"+
				" FROM files f JOIN photos p ON ?"+
				" JOIN markers m ON f.file_uid = m.file_uid AND m.subj_uid = subjects.subj_uid"+
				" WHERE m.marker_invalid = 0 AND f.deleted_at IS NULL) WHERE ?", photosJoin, condition))

		// Update photo count.
		if res.Error != nil {
			return res.Error
		}
		photosRes := Db().Table(subjTable).
			UpdateColumn("photo_count", gorm.Expr("(SELECT COUNT(DISTINCT f.photo_id)"+
				" FROM files f JOIN photos p ON ?"+
				" JOIN markers m ON f.file_uid = m.file_uid AND m.subj_uid = subjects.subj_uid"+
				" WHERE m.marker_invalid = 0 AND f.deleted_at IS NULL) WHERE ?", photosJoin, condition))
		if photosRes.Error != nil {
			return photosRes.Error
		}
		res.RowsAffected += photosRes.RowsAffected
	default:
		return fmt.Errorf("sql: unsupported dialect %s", DbDialect())
	}

	if res.Error != nil {
		return res.Error
	}

	log.Debugf("counts: updated %s [%s]", english.Plural(int(res.RowsAffected), "subject", "subjects"), time.Since(start))

	return nil
}

// updateLabelCountsLastUpdated stores the unix timestamp of the last successful label
// count refresh. Keeping this in memory avoids expensive recounts when labels are
// requested repeatedly (for example, when browsing the Labels tab).
var updateLabelCountsLastUpdated atomic.Int64

// UpdateLabelCountsInterval defines the minimum number of seconds between two automatic
// label count refreshes. It is exposed for tests so that the interval can be shortened.
var UpdateLabelCountsInterval int64 = 300

// ShouldUpdateLabelCounts returns true when enough time has passed to justify another
// label count refresh. It never mutates state on its own; callers must update the
// timestamp after performing the refresh (see UpdateLabelCounts).
func ShouldUpdateLabelCounts() bool {
	lastUpdated := updateLabelCountsLastUpdated.Load()
	if lastUpdated <= 0 {
		return true
	}

	if (unix.Now() - lastUpdated) > UpdateLabelCountsInterval {
		return true
	}

	return false
}

// UpdateLabelCountsIfNeeded updates the label photo counts if needed.
func UpdateLabelCountsIfNeeded() (err error) {
	if !ShouldUpdateLabelCounts() {
		return nil
	}

	return UpdateLabelCounts()
}

// UpdateLabelCounts updates the label photo counts.
func UpdateLabelCounts() (err error) {
	mutex.Index.Lock()
	defer mutex.Index.Unlock()

	start := time.Now()
	var res *gorm.DB
	if IsDialect(dsn.DriverMySQL) {
		res = Db().Exec(`UPDATE labels LEFT JOIN (
		SELECT p2.label_id, COUNT(DISTINCT photo_id) AS label_photos FROM (
			SELECT pl.label_id as label_id, p.id AS photo_id FROM photos p
				JOIN photos_labels pl ON pl.photo_id = p.id AND pl.uncertainty < 100
			WHERE p.photo_quality > -1 AND p.photo_private = 0 AND p.deleted_at IS NULL
			UNION
			SELECT c.category_id as label_id, p.id AS photo_id FROM photos p
				JOIN photos_labels pl ON pl.photo_id = p.id AND pl.uncertainty < 100
				JOIN categories c ON c.label_id = pl.label_id
			WHERE p.photo_quality > -1 AND p.photo_private = 0 AND p.deleted_at IS NULL
			) p2 GROUP BY p2.label_id
		) b ON b.label_id = labels.id
		SET photo_count = CASE WHEN b.label_photos IS NULL THEN 0 ELSE b.label_photos END`)
	} else if IsDialect(dsn.DriverSQLite3) {
		res = Db().
			Table("labels").
			UpdateColumn("photo_count",
				gorm.Expr(`(SELECT photo_count FROM (SELECT label_id, SUM(photo_count) AS photo_count FROM (
				SELECT l.id AS label_id, COUNT(*) AS photo_count FROM labels l
					JOIN photos_labels pl ON pl.label_id = l.id
					JOIN photos ph ON pl.photo_id = ph.id
					WHERE pl.uncertainty < 100
					AND ph.photo_quality > -1
					AND ph.photo_private = 0
					AND ph.deleted_at IS NULL GROUP BY l.id
					UNION ALL
					SELECT l.id AS label_id, COUNT(*) AS photo_count FROM labels l
					JOIN categories c ON c.category_id = l.id
					JOIN photos_labels pl ON pl.label_id = c.label_id
					JOIN photos ph ON pl.photo_id = ph.id
					WHERE pl.uncertainty < 100
					AND ph.photo_quality > -1
					AND ph.photo_private = 0
					AND ph.deleted_at IS NULL GROUP BY l.id) counts GROUP BY label_id) label_counts WHERE label_id = labels.id)`))
	} else {
		return fmt.Errorf("sql: unsupported dialect %s", DbDialect())
	}

	if res.Error != nil {
		return res.Error
	}

	log.Debugf("counts: updated %s [%s]", english.Plural(int(res.RowsAffected), "label", "labels"), time.Since(start))
	updateLabelCountsLastUpdated.Store(unix.Now())

	return nil
}

// UpdateCountsAsync runs UpdateCounts in a goroutine and logs the
// returned error, if any, as a warning. The launched goroutine is
// registered with the package WaitGroup (via AsyncJobAdd / AsyncJobDone)
// so config.CloseDb can drain in-flight work via WaitForAsyncJobs before
// tearing down the database connection. A deferred recover guards against
// any future shutdown race producing a process-killing panic instead of
// a clean log line.
func UpdateCountsAsync() {
	AsyncJobAdd()
	go func() {
		defer AsyncJobDone()
		defer func() {
			if r := recover(); r != nil {
				log.Errorf("index: recovered from panic in UpdateCountsAsync (%v)", r)
			}
		}()
		if err := UpdateCounts(); err != nil {
			log.Warnf("index: %s (update counts)", clean.Error(err))
		}
	}()
}

// UpdateCounts updates precalculated photo and file counts. It returns
// nil without doing any work when the entity database provider has been
// torn down (e.g. during test shutdown), so a stray async invocation
// after CloseDb does not panic on a nil dialect lookup.
func UpdateCounts() (err error) {
	if !countsBusy.CompareAndSwap(false, true) {
		log.Debugf("index: skipped updating counts because it is already in progress")
		return nil
	}

	defer countsBusy.Store(false)

	if Db() == nil {
		log.Debugf("index: skipped updating counts because database is not connected")
		return nil
	}

	log.Debug("index: updating counts")

	if err = UpdatePlacesCounts(); err != nil {
		if strings.Contains(err.Error(), "Error 1054") {
			log.Errorf("counts: failed to update places, potentially incompatible database version")
			log.Errorf("%s see https://jira.mariadb.org/browse/MDEV-25362", err)
			return nil
		}

		return fmt.Errorf("%s while updating places counts", err)
	}

	if err = UpdateSubjectCounts(true); err != nil {
		if strings.Contains(err.Error(), "Error 1054") {
			log.Errorf("counts: failed to update subjects, potentially incompatible database version")
			log.Errorf("%s see https://jira.mariadb.org/browse/MDEV-25362", err)
			return nil
		}

		return fmt.Errorf("%s while updating subject counts", err)
	}

	if err = UpdateLabelCounts(); err != nil {
		return fmt.Errorf("%s while updating label counts", err)
	}

	/* TODO: Slow with many photos due to missing index.
	start = time.Now()

	// Update calendar album visibility.
	switch DbDialect() {
	default:
		if err = UnscopedDb().Exec(`UPDATE albums SET deleted_at = ? WHERE album_type=? AND id NOT IN (
		SELECT a.id FROM albums a JOIN photos p ON a.album_month = MONTH(p.taken_at) AND a.album_year = YEAR(p.taken_at)
		AND p.deleted_at IS NULL AND p.photo_quality > -1 AND p.photo_private = 0 WHERE album_type=? GROUP BY a.id)`,
			TimeStamp(), AlbumMonth, AlbumMonth).Error; err != nil {
			return err
		}
		if err = UnscopedDb().Exec(`UPDATE albums SET deleted_at = NULL WHERE album_type=? AND id IN (
		SELECT a.id FROM albums a JOIN photos p ON a.album_month = MONTH(p.taken_at) AND a.album_year = YEAR(p.taken_at)
		AND p.deleted_at IS NULL AND p.photo_quality > -1 AND p.photo_private = 0 WHERE album_type=? GROUP BY a.id)`,
			AlbumMonth, AlbumMonth).Error; err != nil {
			return err
		}
	}

	log.Debugf("calendar: updating visibility completed [%s]", time.Since(start))
	*/

	return nil
}