summercms/modules/lagoon/purge_test.go

package lagoon

import (
	"bytes"
	"context"
	"encoding/base64"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"git.golem15.com/golem15/summercms/modules/backpack"
	"git.golem15.com/golem15/summercms/modules/bonfire"
	"git.golem15.com/golem15/summercms/modules/compass"
	"git.golem15.com/golem15/summercms/modules/lagoon/attach"
	"git.golem15.com/golem15/summercms/modules/party"
	"gocloud.dev/blob"
	"gocloud.dev/blob/memblob"
	"gorm.io/gorm"
)

// purgeChild is a model created under deferral: soft-deletable, with a
// delete hook that records its calls.
type purgeChild struct {
	ID        uint           `gorm:"column:id;primaryKey"`
	PostID    *uint          `gorm:"column:post_id"`
	Label     string         `gorm:"column:label"`
	DeletedAt gorm.DeletedAt `gorm:"column:deleted_at"`
}

func (purgeChild) TableName() string { return "acme_purge_children" }

var (
	purgeHookMu    sync.Mutex
	purgeHookCalls []uint
)

func (c *purgeChild) BeforeDelete(*gorm.DB) error {
	purgeHookMu.Lock()
	defer purgeHookMu.Unlock()
	purgeHookCalls = append(purgeHookCalls, c.ID)
	return nil
}

// purgeEnv is a migrated database with the child table and a bucket.
type purgeEnv struct {
	gdb    *gorm.DB
	dsn    string
	bucket *blob.Bucket
	key    DeferredKey
}

func newPurgeEnv(t *testing.T, name string) *purgeEnv {
	t.Helper()
	db, dsn := dedicatedDB(t, name)
	gdb, err := Use(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	if err := Migrate(gdb, nil); err != nil {
		t.Fatal(err)
	}
	if err := gdb.AutoMigrate(&purgeChild{}); err != nil {
		t.Fatal(err)
	}
	bucket := memblob.OpenBucket(nil)
	t.Cleanup(func() { _ = bucket.Close() })
	return &purgeEnv{gdb: gdb, dsn: dsn, bucket: bucket, key: DeferredKey{SessionKey: "purge-session-key-0123456789abcdef", AdminID: 1, MasterType: "acme_posts"}}
}

// bind writes a binding through DeferredBind and backdates it by age.
func (e *purgeEnv) bind(t *testing.T, field, slaveType, slaveID string, env *DeferredEnvelope, age time.Duration) uint {
	t.Helper()
	inTx(t, e.gdb, func(ctx context.Context, tx *gorm.DB) error {
		return DeferredBind(ctx, tx, e.key, field, slaveType, slaveID, env)
	})
	var row DeferredBinding
	if err := e.gdb.Where("master_field = ? AND slave_type = ? AND slave_id = ?", field, slaveType, slaveID).Order("id DESC").Take(&row).Error; err != nil {
		t.Fatal(err)
	}
	if err := e.gdb.Exec(`UPDATE deferred_bindings SET created_at = ? WHERE id = ?`, time.Now().Add(-age), row.ID).Error; err != nil {
		t.Fatal(err)
	}
	return row.ID
}

// unbind writes an unbind and backdates it by age.
func (e *purgeEnv) unbind(t *testing.T, field, slaveType, slaveID string, age time.Duration) {
	t.Helper()
	inTx(t, e.gdb, func(ctx context.Context, tx *gorm.DB) error {
		_, err := DeferredUnbind(ctx, tx, e.key, field, slaveType, slaveID)
		return err
	})
	if err := e.gdb.Exec(`UPDATE deferred_bindings SET created_at = ? WHERE slave_id = ? AND NOT is_bind`, time.Now().Add(-age), slaveID).Error; err != nil {
		t.Fatal(err)
	}
}

// file stores a small PNG; attachedTo, when set, attaches it.
func (e *purgeEnv) file(t *testing.T, attachedTo string) attach.File {
	t.Helper()
	f, err := attach.Store(t.Context(), e.gdb, e.bucket, attach.Upload{FileName: "x.png", Body: bytes.NewReader(purgePNG)}, attach.Limits{Image: true})
	if err != nil {
		t.Fatal(err)
	}
	if attachedTo != "" {
		if err := e.gdb.Model(&attach.File{}).Where("id = ?", f.ID).Updates(map[string]any{"attachment_type": "acme_posts", "attachment_id": attachedTo, "field": "cover"}).Error; err != nil {
			t.Fatal(err)
		}
	}
	return *f
}

func (e *purgeEnv) child(t *testing.T, label string) uint {
	t.Helper()
	c := purgeChild{Label: label}
	if err := e.gdb.Create(&c).Error; err != nil {
		t.Fatal(err)
	}
	return c.ID
}

func (e *purgeEnv) blobExists(t *testing.T, f attach.File) bool {
	t.Helper()
	ok, err := e.bucket.Exists(context.Background(), attach.BlobKey(f.DiskName))
	if err != nil {
		t.Fatal(err)
	}
	return ok
}

func (e *purgeEnv) fileExists(t *testing.T, id uint) bool {
	t.Helper()
	var n int64
	if err := e.gdb.Model(&attach.File{}).Where("id = ?", id).Count(&n).Error; err != nil {
		t.Fatal(err)
	}
	return n == 1
}

func (e *purgeEnv) bindingIDs(t *testing.T) []uint {
	t.Helper()
	var ids []uint
	if err := e.gdb.Model(&DeferredBinding{}).Order("id").Pluck("id", &ids).Error; err != nil {
		t.Fatal(err)
	}
	return ids
}

func (e *purgeEnv) models() func(string) (any, bool) {
	return func(slaveType string) (any, bool) {
		if slaveType == "acme_purge_children" {
			return &purgeChild{}, true
		}
		return nil, false
	}
}

var purgePNG = func() []byte {
	// The smallest valid 1x1 PNG.
	return []byte{
		0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0x00, 0x00, 0x00, 0x0d, 0x49, 0x48, 0x44, 0x52,
		0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x01, 0x08, 0x06, 0x00, 0x00, 0x00, 0x1f, 0x15, 0xc4,
		0x89, 0x00, 0x00, 0x00, 0x0d, 0x49, 0x44, 0x41, 0x54, 0x78, 0x9c, 0x63, 0xf8, 0xcf, 0xc0, 0xf0,
		0x1f, 0x00, 0x05, 0x00, 0x01, 0xff, 0x89, 0x99, 0x3d, 0x1d, 0x00, 0x00, 0x00, 0x00, 0x49, 0x45,
		0x4e, 0x44, 0xae, 0x42, 0x60, 0x82,
	}
}()

const day = 24 * time.Hour

// TestPurgeDeferredRules: only bindings older than the cut-off go; an
// unattached pending upload loses its row and blob, an attached file is
// kept, a created child is deleted through its model (hook and soft
// delete), a linked-only record and the slave of an unbind are kept, and an
// unresolvable created child is skipped with its binding left in place.
func TestPurgeDeferredRules(t *testing.T) {
	e := newPurgeEnv(t, "lagoon_purge_rules")
	pending := e.file(t, "")
	attached := e.file(t, "9")
	fresh := e.file(t, "")
	created := e.child(t, "created")
	linked := e.child(t, "linked")

	e.bind(t, "cover", DeferredFileType, fmt.Sprint(pending.ID), nil, 6*day)
	e.bind(t, "cover", DeferredFileType, fmt.Sprint(attached.ID), nil, 6*day)
	freshID := e.bind(t, "cover", DeferredFileType, fmt.Sprint(fresh.ID), nil, 4*day)
	e.bind(t, "children", "acme_purge_children", fmt.Sprint(created), &DeferredEnvelope{Created: true}, 6*day)
	e.bind(t, "children", "acme_purge_children", fmt.Sprint(linked), nil, 6*day)
	e.unbind(t, "children", "acme_purge_children", "77", 6*day)
	skippedID := e.bind(t, "widgets", "acme_unknown", "5", &DeferredEnvelope{Created: true}, 6*day)
	e.bind(t, "cover", DeferredFileType, "not-a-number", nil, 6*day)

	purgeHookMu.Lock()
	purgeHookCalls = nil
	purgeHookMu.Unlock()
	res, err := PurgeDeferred(t.Context(), e.gdb, e.bucket, PurgeOptions{Before: time.Now().Add(-5 * day), Models: e.models()})
	if err != nil {
		t.Fatal(err)
	}
	if res.Bindings != 6 || res.Files != 1 || res.Children != 1 || res.Skipped != 1 {
		t.Fatalf("result %+v", res)
	}
	if ids := e.bindingIDs(t); fmt.Sprint(ids) != fmt.Sprint([]uint{freshID, skippedID}) {
		t.Fatalf("bindings left %v, want the fresh and the skipped one", ids)
	}
	if e.fileExists(t, pending.ID) || e.blobExists(t, pending) {
		t.Fatal("the pending upload's row or blob is left")
	}
	if !e.fileExists(t, attached.ID) || !e.blobExists(t, attached) {
		t.Fatal("an attached file was purged")
	}
	if !e.fileExists(t, fresh.ID) || !e.blobExists(t, fresh) {
		t.Fatal("a binding newer than the cut-off was purged")
	}
	var c purgeChild
	if err := e.gdb.Unscoped().First(&c, created).Error; err != nil || !c.DeletedAt.Valid {
		t.Fatalf("created child %+v (%v), want soft deleted", c, err)
	}
	purgeHookMu.Lock()
	calls := fmt.Sprint(purgeHookCalls)
	purgeHookMu.Unlock()
	if calls != fmt.Sprint([]uint{created}) {
		t.Fatalf("model hook calls %s, want [%d]", calls, created)
	}
	var kept purgeChild
	if err := e.gdb.First(&kept, linked).Error; err != nil {
		t.Fatalf("linked-only child purged: %v", err)
	}

	// A second run finds only the skipped binding again.
	res, err = PurgeDeferred(t.Context(), e.gdb, e.bucket, PurgeOptions{Before: time.Now().Add(-5 * day), Models: e.models()})
	if err != nil || res.Bindings != 0 || res.Skipped != 1 {
		t.Fatalf("second run %+v %v", res, err)
	}
}

// TestPurgeDeferredCutoff: created_at equal to the cut-off is not expired,
// a moment earlier is; a zero cut-off and a nil db are errors.
func TestPurgeDeferredCutoff(t *testing.T) {
	e := newPurgeEnv(t, "lagoon_purge_cutoff")
	cut := time.Now().Add(-5 * day).Truncate(time.Second)
	at := e.bind(t, "children", "acme_purge_children", "1", nil, 0)
	before := e.bind(t, "children", "acme_purge_children", "2", nil, 0)
	if err := e.gdb.Exec(`UPDATE deferred_bindings SET created_at = ? WHERE id = ?`, cut, at).Error; err != nil {
		t.Fatal(err)
	}
	if err := e.gdb.Exec(`UPDATE deferred_bindings SET created_at = ? WHERE id = ?`, cut.Add(-time.Millisecond), before).Error; err != nil {
		t.Fatal(err)
	}
	res, err := PurgeDeferred(t.Context(), e.gdb, e.bucket, PurgeOptions{Before: cut})
	if err != nil || res.Bindings != 1 {
		t.Fatalf("result %+v %v", res, err)
	}
	if ids := e.bindingIDs(t); fmt.Sprint(ids) != fmt.Sprint([]uint{at}) {
		t.Fatalf("left %v, want only the binding at the cut-off", ids)
	}
	if _, err := PurgeDeferred(t.Context(), e.gdb, e.bucket, PurgeOptions{}); err == nil {
		t.Fatal("a zero cut-off was accepted")
	}
	if _, err := PurgeDeferred(t.Context(), nil, e.bucket, PurgeOptions{Before: cut}); err == nil {
		t.Fatal("a nil db was accepted")
	}
}

// TestPurgeDeferredBlobsAfterCommit: inside an outer transaction the purge
// deletes rows but leaves blobs until the commit; a rolled-back purge
// keeps rows, bindings and blobs.
func TestPurgeDeferredBlobsAfterCommit(t *testing.T) {
	e := newPurgeEnv(t, "lagoon_purge_after_commit")
	f := e.file(t, "")
	e.bind(t, "cover", DeferredFileType, fmt.Sprint(f.ID), nil, 6*day)
	opts := PurgeOptions{Before: time.Now().Add(-5 * day)}

	rollback := errors.New("roll back")
	err := Transaction(t.Context(), e.gdb, func(ctx context.Context, tx *gorm.DB) error {
		res, err := PurgeDeferred(ctx, tx, e.bucket, opts)
		if err != nil {
			return err
		}
		if res.Files != 1 {
			t.Errorf("result inside the transaction %+v", res)
		}
		if !e.blobExists(t, f) {
			t.Error("the blob was deleted before the commit")
		}
		return rollback
	})
	if !errors.Is(err, rollback) {
		t.Fatalf("err = %v", err)
	}
	if !e.fileExists(t, f.ID) || !e.blobExists(t, f) || len(e.bindingIDs(t)) != 1 {
		t.Fatal("a rolled-back purge lost the row, the blob or the binding")
	}

	err = Transaction(t.Context(), e.gdb, func(ctx context.Context, tx *gorm.DB) error {
		if _, err := PurgeDeferred(ctx, tx, e.bucket, opts); err != nil {
			return err
		}
		if !e.blobExists(t, f) {
			t.Error("the blob was deleted before the commit")
		}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if e.fileExists(t, f.ID) || e.blobExists(t, f) || len(e.bindingIDs(t)) != 0 {
		t.Fatal("the committed purge left the row, the blob or the binding")
	}
}

// TestPurgeDeferredSkipsLocked: a binding locked by a running save (FOR
// UPDATE in another transaction) is skipped, the rest are purged, and the
// next run takes it once the lock is gone.
func TestPurgeDeferredSkipsLocked(t *testing.T) {
	e := newPurgeEnv(t, "lagoon_purge_skip_locked")
	locked := e.bind(t, "children", "acme_purge_children", "1", nil, 6*day)
	free := e.bind(t, "children", "acme_purge_children", "2", nil, 6*day)

	save := e.gdb.Begin()
	defer save.Rollback()
	var row DeferredBinding
	if err := save.Raw(`SELECT * FROM deferred_bindings WHERE id = ? FOR UPDATE`, locked).Scan(&row).Error; err != nil || row.ID != locked {
		t.Fatalf("lock %+v %v", row, err)
	}
	done := make(chan struct{})
	var res PurgeResult
	var err error
	go func() {
		defer close(done)
		res, err = PurgeDeferred(context.Background(), e.gdb, e.bucket, PurgeOptions{Before: time.Now().Add(-5 * day)})
	}()
	select {
	case <-done:
	case <-time.After(10 * time.Second):
		t.Fatal("the purge waited on a locked binding instead of skipping it")
	}
	if err != nil || res.Bindings != 1 {
		t.Fatalf("result %+v %v", res, err)
	}
	if ids := e.bindingIDs(t); fmt.Sprint(ids) != fmt.Sprint([]uint{locked}) {
		t.Fatalf("left %v, want the locked binding %d (free %d purged)", ids, locked, free)
	}
	if err := save.Rollback().Error; err != nil {
		t.Fatal(err)
	}
	res, err = PurgeDeferred(t.Context(), e.gdb, e.bucket, PurgeOptions{Before: time.Now().Add(-5 * day)})
	if err != nil || res.Bindings != 1 || len(e.bindingIDs(t)) != 0 {
		t.Fatalf("after the lock: %+v %v", res, err)
	}
}

// TestPurgeDeferredNilBucket: without a bucket the purge runs when no
// expired binding points at a file, and refuses before deleting anything
// when one does.
func TestPurgeDeferredNilBucket(t *testing.T) {
	e := newPurgeEnv(t, "lagoon_purge_nil_bucket")
	e.bind(t, "children", "acme_purge_children", "1", nil, 6*day)
	opts := PurgeOptions{Before: time.Now().Add(-5 * day)}
	if res, err := PurgeDeferred(t.Context(), e.gdb, nil, opts); err != nil || res.Bindings != 1 {
		t.Fatalf("no file bindings: %+v %v", res, err)
	}
	f := e.file(t, "")
	e.bind(t, "cover", DeferredFileType, fmt.Sprint(f.ID), nil, 6*day)
	e.bind(t, "children", "acme_purge_children", "2", nil, 6*day)
	if _, err := PurgeDeferred(t.Context(), e.gdb, nil, opts); err == nil || !strings.Contains(err.Error(), "bucket") {
		t.Fatalf("err = %v, want a bucket error", err)
	}
	if len(e.bindingIDs(t)) != 2 || !e.fileExists(t, f.ID) {
		t.Fatal("a refused purge deleted something")
	}
}

// purgeInput is a bonfire.Input with flags only.
type purgeInput map[string]string

func (purgeInput) Args() []string                  { return nil }
func (purgeInput) Argument(string) (string, bool)  { return "", false }
func (in purgeInput) Flag(n string) (string, bool) { v, ok := in[n]; return v, ok }
func (purgeInput) Flags(string) []string           { return nil }
func purgeConfig(t *testing.T, values map[string]any) *compass.Config {
	t.Helper()
	dir := t.TempDir()
	if err := os.WriteFile(filepath.Join(dir, "app.yaml"), []byte("name: lagoon-purge\n"), 0o644); err != nil {
		t.Fatal(err)
	}
	cfg, err := compass.Open(compass.Options{Dir: dir, Env: "development", Environ: []string{"SUMMER_ENV=development"}})
	if err != nil {
		t.Fatal(err)
	}
	for k, v := range values {
		if err := cfg.Set(k, v); err != nil {
			t.Fatal(err)
		}
	}
	return cfg
}

// TestPurgeDeferredDays: --days wins over database.deferred_bindings
// .purge_days, which wins over the default of five; negative and
// non-integer values are refused.
func TestPurgeDeferredDays(t *testing.T) {
	cases := []struct {
		name string
		cfg  map[string]any
		in   purgeInput
		want int
		err  bool
	}{
		{"default", nil, nil, 5, false},
		{"nil app", nil, nil, 5, false},
		{"config int", map[string]any{purgeDaysKey: 9}, nil, 9, false},
		{"config string", map[string]any{purgeDaysKey: "3"}, nil, 3, false},
		{"config whole float", map[string]any{purgeDaysKey: 2.0}, nil, 2, false},
		{"flag wins", map[string]any{purgeDaysKey: 9}, purgeInput{"days": "0"}, 0, false},
		{"flag negative", nil, purgeInput{"days": "-1"}, 0, true},
		{"flag text", nil, purgeInput{"days": "five"}, 0, true},
		{"config negative", map[string]any{purgeDaysKey: -2}, nil, 0, true},
		{"config fraction", map[string]any{purgeDaysKey: 1.5}, nil, 0, true},
		{"config text", map[string]any{purgeDaysKey: "x"}, nil, 0, true},
		{"config bool", map[string]any{purgeDaysKey: true}, nil, 0, true},
	}
	for _, tc := range cases {
		var app *backpack.App
		if tc.name != "nil app" {
			app = backpack.New(purgeConfig(t, tc.cfg))
		}
		var in bonfire.Input
		if tc.in != nil {
			in = tc.in
		}
		got, err := purgeDays(app, in)
		if (err != nil) != tc.err || (!tc.err && got != tc.want) {
			t.Fatalf("%s: days=%d err=%v", tc.name, got, err)
		}
	}
}

// purgePlugin lists the child model for the deferred:purge command.
type purgePlugin struct{ models []any }

func (purgePlugin) ID() string                   { return "acme.purge" }
func (purgePlugin) Requires() []string           { return nil }
func (purgePlugin) Register(*backpack.App) error { return nil }
func (purgePlugin) Boot(*backpack.App) error     { return nil }
func (p purgePlugin) Models() []any              { return p.models }

type otherChild struct {
	ID uint `gorm:"column:id;primaryKey"`
}

func (otherChild) TableName() string { return "acme_purge_children" }

// TestPurgeDeferredCommand runs deferred:purge end to end: it resolves
// created children through the plugins' Models(), uses the published
// bucket and reports the counts; a morph type shared by two model types is
// refused.
func TestPurgeDeferredCommand(t *testing.T) {
	e := newPurgeEnv(t, "lagoon_purge_command")
	f := e.file(t, "")
	created := e.child(t, "created")
	e.bind(t, "cover", DeferredFileType, fmt.Sprint(f.ID), nil, 3*day)
	e.bind(t, "children", "acme_purge_children", fmt.Sprint(created), &DeferredEnvelope{Created: true}, 3*day)

	run := func(plugins []party.Plugin, args ...string) (string, error) {
		app := backpack.New(purgeConfig(t, map[string]any{"database.dsn": e.dsn, "app.key": base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{7}, 32))}))
		if err := attach.Publish(app, e.bucket); err != nil {
			t.Fatal(err)
		}
		var out bytes.Buffer
		root, err := bonfire.NewRoot("app", RuntimeCommands(app, plugins), &out)
		if err != nil {
			t.Fatal(err)
		}
		root.SetArgs(append([]string{"deferred:purge"}, args...))
		root.SetOut(&out)
		root.SetErr(&out)
		err = root.Execute()
		return out.String(), err
	}
	// The default five days keeps three-day-old bindings.
	if out, err := run([]party.Plugin{purgePlugin{models: []any{&purgeChild{}}}}); err != nil || !strings.Contains(out, "purged 0 deferred bindings older than 5 days") {
		t.Fatalf("default run: %q %v", out, err)
	}
	if _, err := run([]party.Plugin{purgePlugin{models: []any{&purgeChild{}, &otherChild{}}}}, "--days", "2"); err == nil || !strings.Contains(err.Error(), "share the morph type") {
		t.Fatalf("shared morph type: %v", err)
	}
	if _, err := run(nil, "--days", "-3"); err == nil {
		t.Fatal("negative --days accepted")
	}
	out, err := run([]party.Plugin{purgePlugin{models: []any{&purgeChild{}}}}, "--days", "2")
	if err != nil || !strings.Contains(out, "purged 2 deferred bindings older than 2 days (1 files, 1 children, 0 skipped)") {
		t.Fatalf("run: %q %v", out, err)
	}
	if e.fileExists(t, f.ID) || e.blobExists(t, f) || len(e.bindingIDs(t)) != 0 {
		t.Fatal("the command left the file, its blob or a binding")
	}
}