refac: atomic merge, lost-update race, RunPurge cost, duplicated walks
This commit is contained in:
1 parent
0fc3511e2b
commit
074944c99a
45 files changed
+2204
-267
No files matched your search
@@ -115,6 +115,21 @@ func (f Field) validDefault() bool {
|
||||
return !f.Unique && f.Violation(f.Default) == ""
|
||||
}
|
||||
|
||||
type MediaRef struct {
|
||||
Field Field
|
||||
ID uint32
|
||||
}
|
||||
|
||||
func MediaRefs(fields []Field, data map[string]any) []MediaRef {
|
||||
var refs []MediaRef
|
||||
for _, field := range MediaFields(fields) {
|
||||
if id, set := MediaID(data[field.Key]); set {
|
||||
refs = append(refs, MediaRef{Field: field, ID: id})
|
||||
}
|
||||
}
|
||||
return refs
|
||||
}
|
||||
|
||||
func MediaFields(fields []Field) []Field {
|
||||
var media []Field
|
||||
for _, field := range fields {
|
||||
|
||||
@@ -51,12 +51,12 @@ func validateField(field contenttype.Field, data map[string]any) error {
|
||||
value := data[field.Key]
|
||||
if value == nil {
|
||||
if field.Required {
|
||||
return invalidField(field.Key, RuleRequired)
|
||||
return InvalidField(field.Key, RuleRequired)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if rule := field.Violation(value); rule != "" {
|
||||
return invalidField(field.Key, rule)
|
||||
return InvalidField(field.Key, rule)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -76,7 +76,7 @@ func firstUnknownField(fields []contenttype.Field, data map[string]any) error {
|
||||
return nil
|
||||
}
|
||||
sort.Strings(unknown)
|
||||
return invalidField(unknown[0], RuleUnknownField)
|
||||
return InvalidField(unknown[0], RuleUnknownField)
|
||||
}
|
||||
|
||||
func UniqueValues(fields []contenttype.Field, data map[string]any) []UniqueValue {
|
||||
@@ -91,10 +91,15 @@ func UniqueValues(fields []contenttype.Field, data map[string]any) []UniqueValue
|
||||
|
||||
func MediaIDs(fields []contenttype.Field, data map[string]any) map[string]uint32 {
|
||||
ids := map[string]uint32{}
|
||||
for _, field := range contenttype.MediaFields(fields) {
|
||||
if id, ok := contenttype.MediaID(data[field.Key]); ok {
|
||||
ids[field.Key] = id
|
||||
}
|
||||
for _, ref := range contenttype.MediaRefs(fields, data) {
|
||||
ids[ref.Field.Key] = ref.ID
|
||||
}
|
||||
return ids
|
||||
}
|
||||
|
||||
func MissingMedia(field contenttype.Field, unchanged bool) error {
|
||||
if unchanged {
|
||||
return InvalidField(field.Key, RuleRequired)
|
||||
}
|
||||
return InvalidField(field.Key, contenttype.RuleType)
|
||||
}
|
||||
@@ -55,12 +55,8 @@ func (e FieldError) Unwrap() error {
|
||||
return e.Cause
|
||||
}
|
||||
|
||||
func invalidField(key string, rule string) error {
|
||||
return FieldError{Key: key, Rule: rule, Cause: ErrInvalidEntry}
|
||||
}
|
||||
|
||||
func InvalidField(key string, rule string) error {
|
||||
return invalidField(key, rule)
|
||||
return FieldError{Key: key, Rule: rule, Cause: ErrInvalidEntry}
|
||||
}
|
||||
|
||||
func UniqueViolation(key string) error {
|
||||
|
||||
@@ -246,3 +246,52 @@ type Image struct {
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
|
||||
func isStoredNameByte(char byte) bool {
|
||||
return char == '-' || char == '_' || char == '.' || char >= '0' && char <= '9' || char >= 'a' && char <= 'z' || char >= 'A' && char <= 'Z'
|
||||
}
|
||||
|
||||
func continuesName(body string, end int) bool {
|
||||
if end >= len(body) || !isStoredNameByte(body[end]) {
|
||||
return false
|
||||
}
|
||||
return body[end] != '.' || end+1 < len(body) && isStoredNameByte(body[end+1])
|
||||
}
|
||||
|
||||
func storedNameMatches(body string, name string) []int {
|
||||
var starts []int
|
||||
if name == "" {
|
||||
return starts
|
||||
}
|
||||
for from := 0; from < len(body); {
|
||||
found := strings.Index(body[from:], name)
|
||||
if found < 0 {
|
||||
break
|
||||
}
|
||||
start := from + found
|
||||
end := start + len(name)
|
||||
from = start + 1
|
||||
if start > 0 && isStoredNameByte(body[start-1]) || continuesName(body, end) {
|
||||
continue
|
||||
}
|
||||
starts = append(starts, start)
|
||||
from = end
|
||||
}
|
||||
return starts
|
||||
}
|
||||
|
||||
func ContainsStoredName(body string, name string) bool {
|
||||
return len(storedNameMatches(body, name)) > 0
|
||||
}
|
||||
|
||||
func ReplaceStoredName(body string, from string, to string) string {
|
||||
var rewritten strings.Builder
|
||||
last := 0
|
||||
for _, start := range storedNameMatches(body, from) {
|
||||
rewritten.WriteString(body[last:start])
|
||||
rewritten.WriteString(to)
|
||||
last = start + len(from)
|
||||
}
|
||||
rewritten.WriteString(body[last:])
|
||||
return rewritten.String()
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
package transaction
|
||||
|
||||
import (
|
||||
"RenewCMS/internal/domain/article"
|
||||
"RenewCMS/internal/domain/author"
|
||||
"RenewCMS/internal/domain/entry"
|
||||
"RenewCMS/internal/domain/image"
|
||||
"RenewCMS/internal/domain/site"
|
||||
"context"
|
||||
)
|
||||
|
||||
type Repositories struct {
|
||||
Articles article.Repository
|
||||
Sites site.Repository
|
||||
Authors author.Repository
|
||||
Entries entry.Repository
|
||||
Images image.Repository
|
||||
}
|
||||
|
||||
type Transactor interface {
|
||||
Run(ctx context.Context, work func(Repositories) error) error
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
package persistence
|
||||
|
||||
import (
|
||||
domain "RenewCMS/internal/domain/transaction"
|
||||
"context"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
var _ domain.Transactor = &Transactor{}
|
||||
|
||||
type Transactor struct {
|
||||
db *gorm.DB
|
||||
}
|
||||
|
||||
func NewTransactor(db *gorm.DB) *Transactor {
|
||||
return &Transactor{db}
|
||||
}
|
||||
|
||||
func (t *Transactor) Run(ctx context.Context, work func(domain.Repositories) error) error {
|
||||
return t.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
return work(domain.Repositories{
|
||||
Articles: NewArticleRepository(tx),
|
||||
Sites: NewSiteRepository(tx),
|
||||
Authors: NewAuthorRepository(tx),
|
||||
Entries: NewEntryRepository(tx),
|
||||
Images: NewImageRepository(tx),
|
||||
})
|
||||
})
|
||||
}
|
||||
@@ -16,7 +16,7 @@ const (
|
||||
)
|
||||
|
||||
type Job interface {
|
||||
Tick(now time.Time) (purge.RunResult, bool, error)
|
||||
Tick(ctx context.Context, now time.Time) (purge.RunResult, bool, error)
|
||||
}
|
||||
|
||||
type CleanupJob interface {
|
||||
@@ -24,15 +24,15 @@ type CleanupJob interface {
|
||||
}
|
||||
|
||||
type task interface {
|
||||
tick(now time.Time) (summary string, ran bool, err error)
|
||||
tick(ctx context.Context, now time.Time) (summary string, ran bool, err error)
|
||||
}
|
||||
|
||||
type purgeTask struct {
|
||||
job Job
|
||||
}
|
||||
|
||||
func (p purgeTask) tick(now time.Time) (string, bool, error) {
|
||||
result, ran, err := p.job.Tick(now)
|
||||
func (p purgeTask) tick(ctx context.Context, now time.Time) (string, bool, error) {
|
||||
result, ran, err := p.job.Tick(ctx, now)
|
||||
if errors.Is(err, purge.ErrPurgeBusy) {
|
||||
return "", false, nil
|
||||
}
|
||||
@@ -44,7 +44,7 @@ type cleanupTask struct {
|
||||
job CleanupJob
|
||||
}
|
||||
|
||||
func (c cleanupTask) tick(now time.Time) (string, bool, error) {
|
||||
func (c cleanupTask) tick(_ context.Context, now time.Time) (string, bool, error) {
|
||||
result, ran, err := c.job.Tick(now)
|
||||
return fmt.Sprintf("removed=%d", result.Removed), ran, err
|
||||
}
|
||||
@@ -119,7 +119,7 @@ func (r *Runner) tick(ctx context.Context) {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
summary, ran, err := r.task.tick(r.now())
|
||||
summary, ran, err := r.task.tick(ctx, r.now())
|
||||
if err != nil {
|
||||
r.logger.Printf("scheduler: %s failed: %v", r.label, err)
|
||||
return
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
"RenewCMS/internal/domain/image"
|
||||
"RenewCMS/internal/domain/site"
|
||||
"slices"
|
||||
"strings"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -111,7 +110,7 @@ func (u *GetMediaUsageUseCase) articleUsages(img image.Image) ([]MediaUsage, err
|
||||
}
|
||||
byID := map[uint32]*MediaUsage{}
|
||||
for _, found := range inBody {
|
||||
if !strings.Contains(found.Body, storedName) {
|
||||
if !image.ContainsStoredName(found.Body, storedName) {
|
||||
continue
|
||||
}
|
||||
entry := usageFor(byID, found)
|
||||
@@ -172,5 +171,5 @@ func (u *GetMediaUsageUseCase) siteUsages(img image.Image) ([]MediaUsage, error)
|
||||
}
|
||||
|
||||
func articleMentionsImage(a article.Article, img image.Image) bool {
|
||||
return strings.Contains(a.Body, image.StoredName(img.Path)) || a.UsesImage(img.ID)
|
||||
return image.ContainsStoredName(a.Body, image.StoredName(img.Path)) || a.UsesImage(img.ID)
|
||||
}
|
||||
@@ -3,17 +3,14 @@ package useCases
|
||||
import (
|
||||
"RenewCMS/internal/domain/article"
|
||||
"RenewCMS/internal/domain/audit"
|
||||
"RenewCMS/internal/domain/author"
|
||||
"RenewCMS/internal/domain/contenttype"
|
||||
"RenewCMS/internal/domain/entry"
|
||||
"RenewCMS/internal/domain/image"
|
||||
"RenewCMS/internal/domain/site"
|
||||
"cmp"
|
||||
"RenewCMS/internal/domain/transaction"
|
||||
"context"
|
||||
"errors"
|
||||
"log"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type MergeMediaResult struct {
|
||||
@@ -30,13 +27,13 @@ type MergeMediaUseCase struct {
|
||||
articleRepository article.Repository
|
||||
remover mediaRemover
|
||||
recorder audit.Recorder
|
||||
sites site.Repository
|
||||
authors author.Repository
|
||||
entries *entryMediaScanner
|
||||
transactor transaction.Transactor
|
||||
logger *log.Logger
|
||||
}
|
||||
|
||||
func NewMergeMediaUseCase(imageRepository image.Repository, articleRepository article.Repository, storage image.Storage, variantFiles image.VariantFileRepository, recorder audit.Recorder, sites site.Repository, authors author.Repository, contentTypes contenttype.Repository, entries entry.Repository) *MergeMediaUseCase {
|
||||
return &MergeMediaUseCase{imageRepository: imageRepository, articleRepository: articleRepository, remover: newMediaRemover(imageRepository, storage, variantFiles), recorder: recorder, sites: sites, authors: authors, entries: newEntryMediaScanner(contentTypes, entries)}
|
||||
func NewMergeMediaUseCase(imageRepository image.Repository, articleRepository article.Repository, storage image.Storage, variantFiles image.VariantFileRepository, recorder audit.Recorder, contentTypes contenttype.Repository, entries entry.Repository, transactor transaction.Transactor, logger *log.Logger) *MergeMediaUseCase {
|
||||
return &MergeMediaUseCase{imageRepository: imageRepository, articleRepository: articleRepository, remover: newMediaRemover(imageRepository, storage, variantFiles), recorder: recorder, entries: newEntryMediaScanner(contentTypes, entries), transactor: transactor, logger: logger}
|
||||
}
|
||||
|
||||
func copyIdsOf(copies []image.Image) []uint32 {
|
||||
@@ -47,22 +44,8 @@ func copyIdsOf(copies []image.Image) []uint32 {
|
||||
return copyIds
|
||||
}
|
||||
|
||||
func (u *MergeMediaUseCase) repointSites(copies []image.Image, keptId uint32) (int, error) {
|
||||
return u.sites.RepointImage(copyIdsOf(copies), keptId)
|
||||
}
|
||||
|
||||
func (u *MergeMediaUseCase) repointAuthors(copies []image.Image, keptId uint32) (int, error) {
|
||||
return u.authors.RepointAvatar(copyIdsOf(copies), keptId)
|
||||
}
|
||||
|
||||
type bodyRewrite struct {
|
||||
articleId uint32
|
||||
original string
|
||||
rewritten string
|
||||
}
|
||||
|
||||
func (u *MergeMediaUseCase) Merge(ctx context.Context, keepId uint32, mergeIds []uint32) (MergeMediaResult, error) {
|
||||
result, err := u.merge(keepId, mergeIds)
|
||||
result, err := u.merge(ctx, keepId, mergeIds, nil)
|
||||
event := audit.NewEvent(audit.ActionMediaMerge, audit.CategoryMedia)
|
||||
if kept, found := u.keptImage(keepId); found {
|
||||
event.Target = imageTarget(kept)
|
||||
@@ -84,7 +67,7 @@ func (u *MergeMediaUseCase) keptImage(keepId uint32) (image.Image, bool) {
|
||||
return kept, err == nil
|
||||
}
|
||||
|
||||
func (u *MergeMediaUseCase) merge(keepId uint32, mergeIds []uint32) (MergeMediaResult, error) {
|
||||
func (u *MergeMediaUseCase) merge(ctx context.Context, keepId uint32, mergeIds []uint32, index *entryMediaIndex) (MergeMediaResult, error) {
|
||||
if err := validateMergeRequest(keepId, mergeIds); err != nil {
|
||||
return MergeMediaResult{}, err
|
||||
}
|
||||
@@ -98,51 +81,94 @@ func (u *MergeMediaUseCase) merge(keepId uint32, mergeIds []uint32) (MergeMediaR
|
||||
return MergeMediaResult{}, err
|
||||
}
|
||||
|
||||
rewrites, err := u.planRewrites(kept, copies)
|
||||
planned, err := u.planRewrites(copies)
|
||||
if err != nil {
|
||||
return MergeMediaResult{}, err
|
||||
}
|
||||
if err := u.applyRewrites(rewrites); err != nil {
|
||||
return MergeMediaResult{}, err
|
||||
}
|
||||
repointedArticles, err := u.articleRepository.RepointImage(copyIdsOf(copies), kept.ID)
|
||||
if err != nil {
|
||||
u.restoreBodies(rewrites)
|
||||
return MergeMediaResult{}, err
|
||||
}
|
||||
sitesUpdated, err := u.repointSites(copies, kept.ID)
|
||||
if err != nil {
|
||||
u.restoreBodies(rewrites)
|
||||
return MergeMediaResult{}, err
|
||||
}
|
||||
|
||||
authorsUpdated, err := u.repointAuthors(copies, kept.ID)
|
||||
if err != nil {
|
||||
u.restoreBodies(rewrites)
|
||||
return MergeMediaResult{}, err
|
||||
}
|
||||
|
||||
repointedEntries, err := u.entries.repoint(copyIdsOf(copies), kept.ID)
|
||||
if err != nil {
|
||||
u.restoreBodies(rewrites)
|
||||
return MergeMediaResult{}, err
|
||||
}
|
||||
|
||||
for _, duplicate := range copies {
|
||||
if err := u.remover.remove(duplicate); err != nil {
|
||||
if index == nil {
|
||||
if index, err = u.entries.index(); err != nil {
|
||||
return MergeMediaResult{}, err
|
||||
}
|
||||
}
|
||||
|
||||
updatedIds := updatedArticleIds(rewrites, repointedArticles)
|
||||
return MergeMediaResult{ArticlesUpdated: len(updatedIds), Merged: len(copies), SitesUpdated: sitesUpdated, AuthorsUpdated: authorsUpdated, EntriesUpdated: len(repointedEntries), UpdatedArticleIDs: updatedIds}, nil
|
||||
result, err := u.applyMerge(ctx, kept, copies, planned, index.referencing(copyIdsOf(copies)))
|
||||
if err != nil {
|
||||
return MergeMediaResult{}, err
|
||||
}
|
||||
u.discardCopies(copies)
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func updatedArticleIds(rewrites []bodyRewrite, repointed []uint32) []uint32 {
|
||||
ids := slices.Clone(repointed)
|
||||
for _, plan := range rewrites {
|
||||
ids = append(ids, plan.articleId)
|
||||
func (u *MergeMediaUseCase) applyMerge(ctx context.Context, kept image.Image, copies []image.Image, planned []uint32, candidates []mediaEntry) (MergeMediaResult, error) {
|
||||
var result MergeMediaResult
|
||||
copyIds := copyIdsOf(copies)
|
||||
err := u.transactor.Run(ctx, func(tx transaction.Repositories) error {
|
||||
rewritten, err := rewriteBodies(tx.Articles, planned, copies, kept)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
repointedArticles, err := tx.Articles.RepointImage(copyIds, kept.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if result.SitesUpdated, err = tx.Sites.RepointImage(copyIds, kept.ID); err != nil {
|
||||
return err
|
||||
}
|
||||
if result.AuthorsUpdated, err = tx.Authors.RepointAvatar(copyIds, kept.ID); err != nil {
|
||||
return err
|
||||
}
|
||||
if result.EntriesUpdated, err = repointEntries(tx.Entries, candidates, copyIds, kept.ID); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, duplicate := range copies {
|
||||
if err := tx.Images.Delete(duplicate.ID); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
result.UpdatedArticleIDs = updatedArticleIds(rewritten, repointedArticles)
|
||||
result.ArticlesUpdated = len(result.UpdatedArticleIDs)
|
||||
result.Merged = len(copies)
|
||||
return nil
|
||||
})
|
||||
return result, err
|
||||
}
|
||||
|
||||
func (u *MergeMediaUseCase) discardCopies(copies []image.Image) {
|
||||
for _, duplicate := range copies {
|
||||
if err := u.remover.discardFiles(duplicate); err != nil {
|
||||
u.logger.Printf("media merge: could not remove the files of %s: %v", image.StoredName(duplicate.Path), err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func rewriteBodies(articles article.Repository, planned []uint32, copies []image.Image, kept image.Image) ([]uint32, error) {
|
||||
keptName := image.StoredName(kept.Path)
|
||||
var rewritten []uint32
|
||||
for _, id := range planned {
|
||||
current, err := articles.Get(id)
|
||||
if errors.Is(err, article.ErrArticleNotFound) {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
body := current.Body
|
||||
for _, duplicate := range copies {
|
||||
body = image.ReplaceStoredName(body, image.StoredName(duplicate.Path), keptName)
|
||||
}
|
||||
if body == current.Body {
|
||||
continue
|
||||
}
|
||||
if _, err := articles.UpdateBody(id, body); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
rewritten = append(rewritten, id)
|
||||
}
|
||||
return rewritten, nil
|
||||
}
|
||||
|
||||
func updatedArticleIds(rewritten []uint32, repointed []uint32) []uint32 {
|
||||
ids := append(slices.Clone(repointed), rewritten...)
|
||||
slices.Sort(ids)
|
||||
return slices.Compact(ids)
|
||||
}
|
||||
@@ -179,10 +205,8 @@ func (u *MergeMediaUseCase) loadCopies(kept image.Image, mergeIds []uint32) ([]i
|
||||
return copies, nil
|
||||
}
|
||||
|
||||
func (u *MergeMediaUseCase) planRewrites(kept image.Image, copies []image.Image) ([]bodyRewrite, error) {
|
||||
keptName := image.StoredName(kept.Path)
|
||||
rewritten := map[uint32]*bodyRewrite{}
|
||||
|
||||
func (u *MergeMediaUseCase) planRewrites(copies []image.Image) ([]uint32, error) {
|
||||
seen := map[uint32]bool{}
|
||||
for _, duplicate := range copies {
|
||||
copyName := image.StoredName(duplicate.Path)
|
||||
articles, err := u.articleRepository.FindByBodyContaining(copyName)
|
||||
@@ -190,39 +214,15 @@ func (u *MergeMediaUseCase) planRewrites(kept image.Image, copies []image.Image)
|
||||
return nil, err
|
||||
}
|
||||
for _, a := range articles {
|
||||
if !strings.Contains(a.Body, copyName) {
|
||||
continue
|
||||
if image.ContainsStoredName(a.Body, copyName) {
|
||||
seen[a.ID] = true
|
||||
}
|
||||
plan, ok := rewritten[a.ID]
|
||||
if !ok {
|
||||
plan = &bodyRewrite{articleId: a.ID, original: a.Body, rewritten: a.Body}
|
||||
rewritten[a.ID] = plan
|
||||
}
|
||||
plan.rewritten = strings.ReplaceAll(plan.rewritten, copyName, keptName)
|
||||
}
|
||||
}
|
||||
|
||||
plans := make([]bodyRewrite, 0, len(rewritten))
|
||||
for _, plan := range rewritten {
|
||||
plans = append(plans, *plan)
|
||||
}
|
||||
slices.SortFunc(plans, func(a, b bodyRewrite) int { return cmp.Compare(a.articleId, b.articleId) })
|
||||
return plans, nil
|
||||
}
|
||||
|
||||
func (u *MergeMediaUseCase) applyRewrites(rewrites []bodyRewrite) error {
|
||||
for applied, plan := range rewrites {
|
||||
if _, err := u.articleRepository.UpdateBody(plan.articleId, plan.rewritten); err != nil {
|
||||
u.restoreBodies(rewrites[:applied])
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (u *MergeMediaUseCase) restoreBodies(applied []bodyRewrite) {
|
||||
for _, plan := range applied {
|
||||
_, err := u.articleRepository.UpdateBody(plan.articleId, plan.original)
|
||||
logCompensation("restore article body "+strconv.FormatUint(uint64(plan.articleId), 10), err)
|
||||
planned := make([]uint32, 0, len(seen))
|
||||
for id := range seen {
|
||||
planned = append(planned, id)
|
||||
}
|
||||
slices.Sort(planned)
|
||||
return planned, nil
|
||||
}
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
)
|
||||
|
||||
type PurgeRunner interface {
|
||||
Execute(fingerprint *string) (purge.RunResult, error)
|
||||
Execute(ctx context.Context, fingerprint *string) (purge.RunResult, error)
|
||||
}
|
||||
|
||||
type PurgeSchedulerUseCase struct {
|
||||
@@ -22,7 +22,7 @@ func NewPurgeSchedulerUseCase(repository purge.Repository, runner PurgeRunner, r
|
||||
return &PurgeSchedulerUseCase{repository, runner, recorder}
|
||||
}
|
||||
|
||||
func (u *PurgeSchedulerUseCase) Tick(now time.Time) (purge.RunResult, bool, error) {
|
||||
func (u *PurgeSchedulerUseCase) Tick(ctx context.Context, now time.Time) (purge.RunResult, bool, error) {
|
||||
settings, err := u.repository.Get()
|
||||
if err != nil {
|
||||
return purge.RunResult{}, false, err
|
||||
@@ -31,7 +31,7 @@ func (u *PurgeSchedulerUseCase) Tick(now time.Time) (purge.RunResult, bool, erro
|
||||
if next == nil || now.Before(*next) {
|
||||
return purge.RunResult{}, false, nil
|
||||
}
|
||||
result, err := u.runner.Execute(nil)
|
||||
result, err := u.runner.Execute(ctx, nil)
|
||||
if err != nil {
|
||||
u.recordFailedRun(err)
|
||||
return purge.RunResult{}, false, err
|
||||
|
||||
@@ -36,7 +36,7 @@ func NewRunPurgeUseCase(imageRepository image.Repository, articleRepository arti
|
||||
}
|
||||
|
||||
func (u *RunPurgeUseCase) Run(ctx context.Context, fingerprint *string) (purge.RunResult, error) {
|
||||
result, err := u.Execute(fingerprint)
|
||||
result, err := u.Execute(ctx, fingerprint)
|
||||
event := audit.NewEvent(audit.ActionMediaPurge, audit.CategoryMedia)
|
||||
if err != nil {
|
||||
recordFailure(ctx, u.recorder, event, err)
|
||||
@@ -55,7 +55,7 @@ func purgeParams(result purge.RunResult) map[string]any {
|
||||
return params
|
||||
}
|
||||
|
||||
func (u *RunPurgeUseCase) Execute(fingerprint *string) (purge.RunResult, error) {
|
||||
func (u *RunPurgeUseCase) Execute(ctx context.Context, fingerprint *string) (purge.RunResult, error) {
|
||||
if !u.guard.TryLock() {
|
||||
return purge.RunResult{}, purge.ErrPurgeBusy
|
||||
}
|
||||
@@ -69,7 +69,10 @@ func (u *RunPurgeUseCase) Execute(fingerprint *string) (purge.RunResult, error)
|
||||
return purge.RunResult{}, purge.ErrPurgeConflict
|
||||
}
|
||||
|
||||
result := u.mergeGroups(plan)
|
||||
result, err := u.mergeGroups(ctx, plan)
|
||||
if err != nil {
|
||||
return purge.RunResult{}, err
|
||||
}
|
||||
if err := u.recordRun(result); err != nil {
|
||||
u.logger.Printf("purge: could not record the run result: %v", err)
|
||||
return purge.RunResult{}, err
|
||||
@@ -77,11 +80,14 @@ func (u *RunPurgeUseCase) Execute(fingerprint *string) (purge.RunResult, error)
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (u *RunPurgeUseCase) mergeGroups(plan purgePlan) purge.RunResult {
|
||||
func (u *RunPurgeUseCase) mergeGroups(ctx context.Context, plan purgePlan) (purge.RunResult, error) {
|
||||
result := purge.RunResult{GroupsTotal: len(plan.groups)}
|
||||
updated := map[uint32]bool{}
|
||||
for _, group := range plan.groups {
|
||||
merged, err := u.merge.merge(group.kept.ID, slices.Clone(group.mergedIds))
|
||||
if err := ctx.Err(); err != nil {
|
||||
return purge.RunResult{}, err
|
||||
}
|
||||
merged, err := u.merge.merge(ctx, group.kept.ID, slices.Clone(group.mergedIds), plan.entries)
|
||||
if err != nil {
|
||||
result.FailedGroups++
|
||||
continue
|
||||
@@ -95,7 +101,7 @@ func (u *RunPurgeUseCase) mergeGroups(plan purgePlan) purge.RunResult {
|
||||
}
|
||||
}
|
||||
result.ArticlesUpdated = len(updated)
|
||||
return result
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (u *RunPurgeUseCase) recordRun(result purge.RunResult) error {
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"RenewCMS/internal/domain/image"
|
||||
"cmp"
|
||||
"errors"
|
||||
"maps"
|
||||
"slices"
|
||||
)
|
||||
|
||||
@@ -30,16 +31,12 @@ func newMediaResolver(images image.Repository) *mediaResolver {
|
||||
|
||||
func (r *mediaResolver) resolve(owner contenttype.ContentType, record entry.Entry) (map[string]EntryMedia, error) {
|
||||
resolved := map[string]EntryMedia{}
|
||||
for _, field := range contenttype.MediaFields(owner.Fields) {
|
||||
id, set := contenttype.MediaID(record.Data[field.Key])
|
||||
if !set {
|
||||
continue
|
||||
}
|
||||
media, err := r.lookup(id)
|
||||
for _, ref := range contenttype.MediaRefs(owner.Fields, record.Data) {
|
||||
media, err := r.lookup(ref.ID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resolved[field.Key] = media
|
||||
resolved[ref.Field.Key] = media
|
||||
}
|
||||
return resolved, nil
|
||||
}
|
||||
@@ -73,6 +70,63 @@ type mediaEntry struct {
|
||||
record entry.Entry
|
||||
}
|
||||
|
||||
func (m mediaEntry) keysPointingAt(ids map[uint32]bool) []string {
|
||||
keys := []string{}
|
||||
for _, ref := range contenttype.MediaRefs(m.owner.Fields, m.record.Data) {
|
||||
if ids[ref.ID] {
|
||||
keys = append(keys, ref.Field.Key)
|
||||
}
|
||||
}
|
||||
return keys
|
||||
}
|
||||
|
||||
func compareMediaEntries(a, b mediaEntry) int {
|
||||
return cmp.Or(cmp.Compare(a.owner.Name, b.owner.Name), cmp.Compare(a.owner.ID, b.owner.ID), cmp.Compare(a.record.ID, b.record.ID))
|
||||
}
|
||||
|
||||
type entryMediaIndex struct {
|
||||
byImage map[uint32][]mediaEntry
|
||||
}
|
||||
|
||||
func newEntryMediaIndex(found []mediaEntry) *entryMediaIndex {
|
||||
index := &entryMediaIndex{byImage: map[uint32][]mediaEntry{}}
|
||||
for _, candidate := range found {
|
||||
distinct := map[uint32]bool{}
|
||||
for _, ref := range contenttype.MediaRefs(candidate.owner.Fields, candidate.record.Data) {
|
||||
distinct[ref.ID] = true
|
||||
}
|
||||
for id := range distinct {
|
||||
index.byImage[id] = append(index.byImage[id], candidate)
|
||||
}
|
||||
}
|
||||
return index
|
||||
}
|
||||
|
||||
func (i *entryMediaIndex) usageCounts() map[uint32]int {
|
||||
counts := make(map[uint32]int, len(i.byImage))
|
||||
for id, found := range i.byImage {
|
||||
counts[id] = len(found)
|
||||
}
|
||||
return counts
|
||||
}
|
||||
|
||||
func (i *entryMediaIndex) referencing(imageIDs []uint32) []mediaEntry {
|
||||
type entryKey struct{ typeID, entryID uint32 }
|
||||
seen := map[entryKey]bool{}
|
||||
var found []mediaEntry
|
||||
for _, id := range imageIDs {
|
||||
for _, candidate := range i.byImage[id] {
|
||||
key := entryKey{candidate.owner.ID, candidate.record.ID}
|
||||
if !seen[key] {
|
||||
seen[key] = true
|
||||
found = append(found, candidate)
|
||||
}
|
||||
}
|
||||
}
|
||||
slices.SortFunc(found, compareMediaEntries)
|
||||
return found
|
||||
}
|
||||
|
||||
type entryMediaScanner struct {
|
||||
contentTypes contenttype.Repository
|
||||
entries entry.Repository
|
||||
@@ -82,7 +136,7 @@ func newEntryMediaScanner(contentTypes contenttype.Repository, entries entry.Rep
|
||||
return &entryMediaScanner{contentTypes: contentTypes, entries: entries}
|
||||
}
|
||||
|
||||
func (s *entryMediaScanner) scan() ([]mediaEntry, error) {
|
||||
func (s *entryMediaScanner) index() (*entryMediaIndex, error) {
|
||||
types, err := s.contentTypes.List()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -100,91 +154,51 @@ func (s *entryMediaScanner) scan() ([]mediaEntry, error) {
|
||||
found = append(found, mediaEntry{owner: owner, record: record})
|
||||
}
|
||||
}
|
||||
slices.SortFunc(found, func(a, b mediaEntry) int {
|
||||
return cmp.Or(cmp.Compare(a.owner.Name, b.owner.Name), cmp.Compare(a.owner.ID, b.owner.ID), cmp.Compare(a.record.ID, b.record.ID))
|
||||
})
|
||||
return found, nil
|
||||
}
|
||||
|
||||
func (m mediaEntry) keysPointingAt(ids map[uint32]bool) []string {
|
||||
keys := []string{}
|
||||
for _, field := range contenttype.MediaFields(m.owner.Fields) {
|
||||
if id, set := contenttype.MediaID(m.record.Data[field.Key]); set && ids[id] {
|
||||
keys = append(keys, field.Key)
|
||||
}
|
||||
}
|
||||
return keys
|
||||
return newEntryMediaIndex(found), nil
|
||||
}
|
||||
|
||||
func (s *entryMediaScanner) usageCounts() (map[uint32]int, error) {
|
||||
found, err := s.scan()
|
||||
index, err := s.index()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
counts := map[uint32]int{}
|
||||
for _, candidate := range found {
|
||||
distinct := map[uint32]bool{}
|
||||
for _, id := range entry.MediaIDs(candidate.owner.Fields, candidate.record.Data) {
|
||||
distinct[id] = true
|
||||
}
|
||||
for id := range distinct {
|
||||
counts[id]++
|
||||
}
|
||||
}
|
||||
return counts, nil
|
||||
return index.usageCounts(), nil
|
||||
}
|
||||
|
||||
func (s *entryMediaScanner) referencing(imageID uint32) ([]mediaEntry, error) {
|
||||
found, err := s.scan()
|
||||
index, err := s.index()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
targets := map[uint32]bool{imageID: true}
|
||||
return slices.DeleteFunc(found, func(candidate mediaEntry) bool {
|
||||
return len(candidate.keysPointingAt(targets)) == 0
|
||||
}), nil
|
||||
return index.referencing([]uint32{imageID}), nil
|
||||
}
|
||||
|
||||
type entryRepoint struct {
|
||||
original entry.Entry
|
||||
updated entry.Entry
|
||||
}
|
||||
|
||||
func (s *entryMediaScanner) repoint(copyIDs []uint32, keptID uint32) ([]entryRepoint, error) {
|
||||
found, err := s.scan()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
func repointEntries(entries entry.Repository, candidates []mediaEntry, copyIDs []uint32, keptID uint32) (int, error) {
|
||||
copies := map[uint32]bool{}
|
||||
for _, id := range copyIDs {
|
||||
copies[id] = true
|
||||
}
|
||||
applied := []entryRepoint{}
|
||||
for _, candidate := range found {
|
||||
keys := candidate.keysPointingAt(copies)
|
||||
updated := 0
|
||||
for _, candidate := range candidates {
|
||||
current, err := entries.Get(candidate.owner.ID, candidate.record.ID)
|
||||
if errors.Is(err, entry.ErrEntryNotFound) {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return updated, err
|
||||
}
|
||||
keys := mediaEntry{owner: candidate.owner, record: current}.keysPointingAt(copies)
|
||||
if len(keys) == 0 {
|
||||
continue
|
||||
}
|
||||
change := entryRepoint{original: candidate.record, updated: candidate.record}
|
||||
change.updated.Data = make(map[string]any, len(candidate.record.Data))
|
||||
for key, value := range candidate.record.Data {
|
||||
change.updated.Data[key] = value
|
||||
}
|
||||
current.Data = maps.Clone(current.Data)
|
||||
for _, key := range keys {
|
||||
change.updated.Data[key] = float64(keptID)
|
||||
current.Data[key] = float64(keptID)
|
||||
}
|
||||
if _, err := s.entries.Update(change.updated); err != nil {
|
||||
s.restore(applied)
|
||||
return nil, err
|
||||
if _, err := entries.Update(current); err != nil {
|
||||
return updated, err
|
||||
}
|
||||
applied = append(applied, change)
|
||||
}
|
||||
return applied, nil
|
||||
}
|
||||
|
||||
func (s *entryMediaScanner) restore(applied []entryRepoint) {
|
||||
for _, change := range applied {
|
||||
_, err := s.entries.Update(change.original)
|
||||
logCompensation("restore entry media", err)
|
||||
updated++
|
||||
}
|
||||
return updated, nil
|
||||
}
|
||||
@@ -24,26 +24,18 @@ func typedEntry(owner contenttype.ContentType, record entry.Entry, resolver *med
|
||||
|
||||
func ensureMediaExists(images image.Repository, owner contenttype.ContentType, data map[string]any, stored map[string]any) error {
|
||||
storedIDs := entry.MediaIDs(owner.Fields, stored)
|
||||
for _, field := range contenttype.MediaFields(owner.Fields) {
|
||||
id, set := contenttype.MediaID(data[field.Key])
|
||||
if !set {
|
||||
for _, ref := range contenttype.MediaRefs(owner.Fields, data) {
|
||||
unchanged := stored != nil && storedIDs[ref.Field.Key] == ref.ID
|
||||
if unchanged && !ref.Field.Required {
|
||||
continue
|
||||
}
|
||||
unchanged := stored != nil && storedIDs[field.Key] == id
|
||||
if unchanged && !field.Required {
|
||||
continue
|
||||
}
|
||||
exists, err := imageExists(images, id)
|
||||
exists, err := imageExists(images, ref.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if exists {
|
||||
continue
|
||||
if !exists {
|
||||
return entry.MissingMedia(ref.Field, unchanged)
|
||||
}
|
||||
if unchanged {
|
||||
return entry.InvalidField(field.Key, entry.RuleRequired)
|
||||
}
|
||||
return entry.InvalidField(field.Key, contenttype.RuleType)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
package useCases
|
||||
|
||||
import "RenewCMS/internal/domain/image"
|
||||
import (
|
||||
"RenewCMS/internal/domain/image"
|
||||
"errors"
|
||||
)
|
||||
|
||||
type mediaRemover struct {
|
||||
imageRepository image.Repository
|
||||
@@ -22,3 +25,8 @@ func (r mediaRemover) remove(img image.Image) error {
|
||||
}
|
||||
return r.imageRepository.Delete(img.ID)
|
||||
}
|
||||
|
||||
func (r mediaRemover) discardFiles(img image.Image) error {
|
||||
storedName := image.StoredName(img.Path)
|
||||
return errors.Join(r.generated.purgeSource(storedName), r.storage.Delete(storedName))
|
||||
}
|
||||
@@ -23,6 +23,7 @@ type purgePlan struct {
|
||||
copies int
|
||||
articles int
|
||||
fingerprint string
|
||||
entries *entryMediaIndex
|
||||
}
|
||||
|
||||
type purgePlanner struct {
|
||||
@@ -43,7 +44,15 @@ func (p *purgePlanner) plan() (purgePlan, error) {
|
||||
return purgePlan{}, err
|
||||
}
|
||||
|
||||
siteUsage, err := imageUsageCount(p.sites, p.authors, p.entries)
|
||||
var entryIndex *entryMediaIndex
|
||||
siteUsage, err := combinedImageUsage(p.sites, p.authors, func() (map[uint32]int, error) {
|
||||
built, err := p.entries.index()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
entryIndex = built
|
||||
return built.usageCounts(), nil
|
||||
})
|
||||
if err != nil {
|
||||
return purgePlan{}, err
|
||||
}
|
||||
@@ -64,7 +73,7 @@ func (p *purgePlanner) plan() (purgePlan, error) {
|
||||
}
|
||||
slices.SortFunc(groups, func(a, b purgeGroup) int { return int(a.kept.ID) - int(b.kept.ID) })
|
||||
|
||||
return purgePlan{groups: groups, copies: copies, articles: len(touched), fingerprint: fingerprintOf(groups)}, nil
|
||||
return purgePlan{groups: groups, copies: copies, articles: len(touched), fingerprint: fingerprintOf(groups), entries: entryIndex}, nil
|
||||
}
|
||||
|
||||
func groupByHash(images []image.Image) [][]image.Image {
|
||||
|
||||
@@ -20,6 +20,10 @@ func siteImageUsage(sites site.Repository) (map[uint32]int, error) {
|
||||
}
|
||||
|
||||
func imageUsageCount(sites site.Repository, authors author.Repository, entries *entryMediaScanner) (map[uint32]int, error) {
|
||||
return combinedImageUsage(sites, authors, entries.usageCounts)
|
||||
}
|
||||
|
||||
func combinedImageUsage(sites site.Repository, authors author.Repository, entryCounts func() (map[uint32]int, error)) (map[uint32]int, error) {
|
||||
usage, err := siteImageUsage(sites)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -31,11 +35,11 @@ func imageUsageCount(sites site.Repository, authors author.Repository, entries *
|
||||
for imageID, total := range avatars {
|
||||
usage[imageID] += total
|
||||
}
|
||||
entryCounts, err := entries.usageCounts()
|
||||
counts, err := entryCounts()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for imageID, total := range entryCounts {
|
||||
for imageID, total := range counts {
|
||||
usage[imageID] += total
|
||||
}
|
||||
return usage, nil
|
||||
|
||||
Reference in new issue
Block a user