summercms/modules/lagoon/deferred_test.go

package lagoon

import (
	"bytes"
	"context"
	"fmt"
	"image"
	"image/png"
	"strconv"
	"strings"
	"testing"
	"time"

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

// TestDeferredUploadPurgeTracer runs the 12.2 storage path end to end: a
// guarded upload held by one admin's deferred binding, then purged with its
// row and, after commit, its blob.
func TestDeferredUploadPurgeTracer(t *testing.T) {
	db, _ := dedicatedDB(t, "lagoon_deferred_tracer")
	gdb, err := Use(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	if err := Migrate(gdb, nil); err != nil {
		t.Fatal(err)
	}
	ctx := t.Context()
	bucket := memblob.OpenBucket(nil)
	t.Cleanup(func() { _ = bucket.Close() })

	var img bytes.Buffer
	if err := png.Encode(&img, image.NewRGBA(image.Rect(0, 0, 2, 2))); err != nil {
		t.Fatal(err)
	}
	f, err := attach.Store(ctx, gdb, bucket, attach.Upload{FileName: "cover.png", Body: &img, Public: true}, attach.Limits{Image: true})
	if err != nil {
		t.Fatal(err)
	}
	key := DeferredKey{SessionKey: "tracer-session-key-0123456789abcdef", AdminID: 7, MasterType: "acme_posts"}
	slaveID := strconv.FormatUint(uint64(f.ID), 10)
	if err := Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
		if err := DeferredBind(ctx, tx, key, "cover", DeferredFileType, slaveID, nil); err != nil {
			return err
		}
		// A repeated bind writes nothing.
		return DeferredBind(ctx, tx, key, "cover", DeferredFileType, slaveID, nil)
	}); err != nil {
		t.Fatal(err)
	}
	var rows []DeferredBinding
	if err := gdb.Find(&rows).Error; err != nil {
		t.Fatal(err)
	}
	if len(rows) != 1 || !rows[0].IsBind || rows[0].BackendUserID != 7 || rows[0].MasterType != "acme_posts" {
		t.Fatalf("bindings %+v", rows)
	}
	// Another admin's key sees nothing.
	other := key
	other.AdminID = 8
	var seen []DeferredBinding
	if err := Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
		var err error
		seen, err = DeferredBindings(ctx, tx, other, []string{"cover"})
		return err
	}); err != nil || len(seen) != 0 {
		t.Fatalf("foreign admin saw %v (%v)", seen, err)
	}

	if err := gdb.Exec(`UPDATE deferred_bindings SET created_at = NOW() - INTERVAL '6 days'`).Error; err != nil {
		t.Fatal(err)
	}
	res, err := PurgeDeferred(ctx, gdb, bucket, PurgeOptions{Before: time.Now().Add(-5 * 24 * time.Hour)})
	if err != nil {
		t.Fatal(err)
	}
	if res.Bindings != 1 || res.Files != 1 || res.Skipped != 0 {
		t.Fatalf("purge result %+v", res)
	}
	var count int64
	if err := gdb.Model(&DeferredBinding{}).Count(&count).Error; err != nil || count != 0 {
		t.Fatalf("bindings left %d (%v)", count, err)
	}
	if err := gdb.Model(&attach.File{}).Where("id = ?", f.ID).Count(&count).Error; err != nil || count != 0 {
		t.Fatalf("file rows left %d (%v)", count, err)
	}
	exists, err := bucket.Exists(ctx, attach.BlobKey(f.DiskName))
	if err != nil || exists {
		t.Fatalf("blob still exists=%v (%v)", exists, err)
	}

	// A bind followed by an unbind of the same slave cancels the pair and
	// hands the cancelled bind back.
	var cancelled *DeferredBinding
	if err := Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
		env := &DeferredEnvelope{Created: true}
		if err := DeferredBind(ctx, tx, key, "comments", "acme_comments", "41", env); err != nil {
			return err
		}
		var err error
		cancelled, err = DeferredUnbind(ctx, tx, key, "comments", "acme_comments", "41")
		return err
	}); err != nil {
		t.Fatal(err)
	}
	if cancelled == nil || !cancelled.IsBind {
		t.Fatalf("cancelled bind %+v", cancelled)
	}
	if env, err := cancelled.Envelope(); err != nil || !env.Created {
		t.Fatalf("envelope %+v (%v)", env, err)
	}
	if err := gdb.Model(&DeferredBinding{}).Count(&count).Error; err != nil || count != 0 {
		t.Fatalf("bindings after cancel %d (%v)", count, err)
	}
}

// deferredDB is a migrated dedicated database for the deferred-binding
// tests.
func deferredDB(t *testing.T, name string) *gorm.DB {
	t.Helper()
	db, _ := 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)
	}
	return gdb
}

// TestDeferredMigrations checks the deferred_bindings shape (D-01): the
// WinterCMS columns plus a NOT NULL backend_user_id, the six indexes, the
// summercms.deferred history, and a rollback that drops the table.
func TestDeferredMigrations(t *testing.T) {
	gdb := deferredDB(t, "lagoon_deferred_migrations")
	type column struct {
		Name     string `gorm:"column:column_name"`
		Type     string `gorm:"column:data_type"`
		Nullable string `gorm:"column:is_nullable"`
	}
	var cols []column
	if err := gdb.Raw(`SELECT column_name, data_type, is_nullable FROM information_schema.columns WHERE table_name = 'deferred_bindings' ORDER BY ordinal_position`).Scan(&cols).Error; err != nil {
		t.Fatal(err)
	}
	want := []column{
		{"id", "integer", "NO"},
		{"master_type", "text", "NO"},
		{"master_field", "text", "NO"},
		{"slave_type", "text", "NO"},
		{"slave_id", "text", "NO"},
		{"pivot_data", "text", "YES"},
		{"session_key", "text", "NO"},
		{"is_bind", "boolean", "NO"},
		{"backend_user_id", "integer", "NO"},
		{"created_at", "timestamp with time zone", "NO"},
		{"updated_at", "timestamp with time zone", "NO"},
	}
	if fmt.Sprint(cols) != fmt.Sprint(want) {
		t.Fatalf("columns\n got %v\nwant %v", cols, want)
	}
	var indexes []string
	if err := gdb.Raw(`SELECT indexname FROM pg_indexes WHERE tablename = 'deferred_bindings' AND indexname <> 'deferred_bindings_pkey' ORDER BY indexname`).Scan(&indexes).Error; err != nil {
		t.Fatal(err)
	}
	wantIdx := []string{
		"deferred_bindings_created_at_index",
		"deferred_bindings_master_field_index",
		"deferred_bindings_master_type_index",
		"deferred_bindings_session_lookup_index",
		"deferred_bindings_slave_id_index",
		"deferred_bindings_slave_type_index",
	}
	if strings.Join(indexes, ",") != strings.Join(wantIdx, ",") {
		t.Fatalf("indexes %v", indexes)
	}
	table, err := HistoryTableName(DeferredHistoryID)
	if err != nil || table != "summer_migrations_summercms_deferred" {
		t.Fatalf("history table %q (%v)", table, err)
	}
	var ids []string
	if err := gdb.Raw(`SELECT id FROM ` + table).Scan(&ids).Error; err != nil || len(ids) != 1 || ids[0] != DeferredBindingMigrations[0].ID {
		t.Fatalf("history ids %v (%v)", ids, err)
	}
	if err := gdb.Exec(`INSERT INTO deferred_bindings (master_type, master_field, slave_type, slave_id, session_key) VALUES ('a', 'b', 'c', '1', 'k')`).Error; err == nil {
		t.Fatal("a binding without backend_user_id was accepted")
	}
	m, err := migrator(gdb, DeferredHistoryID, DeferredBindingMigrations)
	if err != nil {
		t.Fatal(err)
	}
	if err := m.RollbackLast(); err != nil {
		t.Fatal(err)
	}
	if gdb.Migrator().HasTable("deferred_bindings") {
		t.Fatal("rollback left deferred_bindings")
	}
	if err := Migrate(gdb, nil); err != nil || !gdb.Migrator().HasTable("deferred_bindings") {
		t.Fatalf("re-migrate: %v", err)
	}
}

// bindingRows lists every binding in id order.
func bindingRows(t *testing.T, gdb *gorm.DB) []DeferredBinding {
	t.Helper()
	var rows []DeferredBinding
	if err := gdb.Order("id").Find(&rows).Error; err != nil {
		t.Fatal(err)
	}
	return rows
}

// inTx runs fn in a lagoon transaction and fails the test on error.
func inTx(t *testing.T, gdb *gorm.DB, fn func(ctx context.Context, tx *gorm.DB) error) {
	t.Helper()
	if err := Transaction(t.Context(), gdb, fn); err != nil {
		t.Fatal(err)
	}
}

// TestDeferredStore covers the binding store (D-01, D-02, D-22): bind
// dedupe, unbind dedupe, bind/unbind and unbind/bind cancellation, admin and
// master type isolation, argument refusals, DeferredSlaves and
// DeferredForget, and the pivot_data envelope.
func TestDeferredStore(t *testing.T) {
	gdb := deferredDB(t, "lagoon_deferred_store")
	key := DeferredKey{SessionKey: "store-session-key-0123456789abcdef", AdminID: 3, MasterType: "acme_posts"}

	t.Run("refusals", func(t *testing.T) {
		ctx := t.Context()
		for name, k := range map[string]DeferredKey{
			"empty key":         {SessionKey: " ", AdminID: 3, MasterType: "acme_posts"},
			"zero admin":        {SessionKey: key.SessionKey, MasterType: "acme_posts"},
			"empty master type": {SessionKey: key.SessionKey, AdminID: 3},
		} {
			if err := DeferredBind(ctx, gdb, k, "tags", "acme_tags", "1", nil); err == nil {
				t.Fatalf("bind with %s accepted", name)
			}
			if _, err := DeferredUnbind(ctx, gdb, k, "tags", "acme_tags", "1"); err == nil {
				t.Fatalf("unbind with %s accepted", name)
			}
			if _, err := DeferredBindings(ctx, gdb, k, []string{"tags"}); err == nil {
				t.Fatalf("read with %s accepted", name)
			}
		}
		for name, args := range map[string][3]string{"field": {"", "acme_tags", "1"}, "slave type": {"tags", "", "1"}, "slave id": {"tags", "acme_tags", ""}} {
			if err := DeferredBind(ctx, gdb, key, args[0], args[1], args[2], nil); err == nil {
				t.Fatalf("bind with an empty %s accepted", name)
			}
		}
		if err := DeferredBind(ctx, nil, key, "tags", "acme_tags", "1", nil); err == nil {
			t.Fatal("bind with a nil db accepted")
		}
		if err := DeferredForget(ctx, nil, []uint{1}); err == nil {
			t.Fatal("forget with a nil db accepted")
		}
		if rows, err := DeferredBindings(ctx, gdb, key, nil); err != nil || rows != nil {
			t.Fatalf("no fields = %v %v", rows, err)
		}
		if len(bindingRows(t, gdb)) != 0 {
			t.Fatal("a refused call wrote a binding")
		}
	})

	t.Run("dedupe and cancel", func(t *testing.T) {
		inTx(t, gdb, func(ctx context.Context, tx *gorm.DB) error {
			for i := 0; i < 2; i++ {
				if err := DeferredBind(ctx, tx, key, "tags", "acme_tags", "1", nil); err != nil {
					return err
				}
				if _, err := DeferredUnbind(ctx, tx, key, "tags", "acme_tags", "2"); err != nil {
					return err
				}
			}
			return nil
		})
		rows := bindingRows(t, gdb)
		if len(rows) != 2 || !rows[0].IsBind || rows[0].SlaveID != "1" || rows[1].IsBind || rows[1].SlaveID != "2" {
			t.Fatalf("dedupe rows %+v", rows)
		}
		// unbind of a pending bind cancels it and returns it; bind of a
		// pending unbind cancels the unbind and writes nothing.
		var cancelled *DeferredBinding
		inTx(t, gdb, func(ctx context.Context, tx *gorm.DB) error {
			var err error
			if cancelled, err = DeferredUnbind(ctx, tx, key, "tags", "acme_tags", "1"); err != nil {
				return err
			}
			return DeferredBind(ctx, tx, key, "tags", "acme_tags", "2", nil)
		})
		if cancelled == nil || cancelled.SlaveID != "1" || !cancelled.IsBind {
			t.Fatalf("cancelled %+v", cancelled)
		}
		if rows := bindingRows(t, gdb); len(rows) != 0 {
			t.Fatalf("rows after cancelling both pairs %+v", rows)
		}
	})

	t.Run("isolation", func(t *testing.T) {
		otherAdmin := key
		otherAdmin.AdminID = 4
		otherMaster := key
		otherMaster.MasterType = "acme_pages"
		inTx(t, gdb, func(ctx context.Context, tx *gorm.DB) error {
			if err := DeferredBind(ctx, tx, key, "tags", "acme_tags", "7", nil); err != nil {
				return err
			}
			// The same slave under another admin or master type is a
			// separate binding, and does not cancel the first.
			if _, err := DeferredUnbind(ctx, tx, otherAdmin, "tags", "acme_tags", "7"); err != nil {
				return err
			}
			return DeferredBind(ctx, tx, otherMaster, "tags", "acme_tags", "8", nil)
		})
		inTx(t, gdb, func(ctx context.Context, tx *gorm.DB) error {
			for k, wantIDs := range map[DeferredKey]string{key: "7", otherAdmin: "7", otherMaster: "8"} {
				rows, err := DeferredBindings(ctx, tx, k, []string{"tags", "other"})
				if err != nil {
					return err
				}
				if len(rows) != 1 || rows[0].SlaveID != wantIDs || rows[0].BackendUserID != k.AdminID || rows[0].MasterType != k.MasterType {
					t.Errorf("%+v sees %+v", k, rows)
				}
			}
			var ids []string
			if err := DeferredSlaves(tx, key, "tags", "acme_tags", true).Scan(&ids).Error; err != nil || fmt.Sprint(ids) != "[7]" {
				t.Errorf("bind slaves %v %v", ids, err)
			}
			ids = nil
			if err := DeferredSlaves(tx, otherAdmin, "tags", "acme_tags", false).Scan(&ids).Error; err != nil || fmt.Sprint(ids) != "[7]" {
				t.Errorf("unbind slaves %v %v", ids, err)
			}
			var none []string
			if err := DeferredSlaves(tx, otherAdmin, "tags", "acme_tags", true).Scan(&none).Error; err != nil || len(none) != 0 {
				t.Errorf("other admin's bind slaves %v %v", none, err)
			}
			return nil
		})
		rows := bindingRows(t, gdb)
		if len(rows) != 3 {
			t.Fatalf("rows %+v", rows)
		}
		inTx(t, gdb, func(ctx context.Context, tx *gorm.DB) error {
			return DeferredForget(ctx, tx, []uint{rows[0].ID})
		})
		if left := bindingRows(t, gdb); len(left) != 2 || left[0].ID != rows[1].ID {
			t.Fatalf("forget deleted more than its id: %+v", left)
		}
		if err := gdb.Exec(`DELETE FROM deferred_bindings`).Error; err != nil {
			t.Fatal(err)
		}
	})

	t.Run("envelope", func(t *testing.T) {
		inTx(t, gdb, func(ctx context.Context, tx *gorm.DB) error {
			if err := DeferredBind(ctx, tx, key, "parts", "acme_parts", "1", &DeferredEnvelope{Created: true}); err != nil {
				return err
			}
			if err := DeferredBind(ctx, tx, key, "members", "acme_members", "2", &DeferredEnvelope{Pivot: map[string]any{"note": "hi"}}); err != nil {
				return err
			}
			// An empty envelope stores no pivot_data; upload_id alone is stored.
			if err := DeferredBind(ctx, tx, key, "members", "acme_members", "3", &DeferredEnvelope{}); err != nil {
				return err
			}
			return DeferredBind(ctx, tx, key, "photos", "system_files", "4", &DeferredEnvelope{UploadID: "retry-one"})
		})
		rows := bindingRows(t, gdb)
		if len(rows) != 4 || rows[0].PivotData == nil || *rows[0].PivotData != `{"created":true}` ||
			rows[1].PivotData == nil || *rows[1].PivotData != `{"pivot":{"note":"hi"}}` || rows[2].PivotData != nil ||
			rows[3].PivotData == nil || *rows[3].PivotData != `{"upload_id":"retry-one"}` {
			t.Fatalf("pivot_data %+v", rows)
		}
		for i, want := range []DeferredEnvelope{{Created: true}, {Pivot: map[string]any{"note": "hi"}}, {}, {UploadID: "retry-one"}} {
			env, err := rows[i].Envelope()
			if err != nil || fmt.Sprint(env) != fmt.Sprint(want) {
				t.Fatalf("envelope %d = %+v %v", i, env, err)
			}
		}
		bad := "{not json"
		if _, err := (DeferredBinding{ID: 9, PivotData: &bad}).Envelope(); err == nil {
			t.Fatal("invalid pivot_data decoded")
		}
		blank := "  "
		if env, err := (DeferredBinding{PivotData: &blank}).Envelope(); err != nil || env.Created || env.Pivot != nil {
			t.Fatalf("blank pivot_data = %+v %v", env, err)
		}
		// A foreign JSON shape is a plain link.
		winter := `{"sort_order":3}`
		if env, err := (DeferredBinding{PivotData: &winter}).Envelope(); err != nil || env.Created || env.Pivot != nil {
			t.Fatalf("winter pivot_data = %+v %v", env, err)
		}
	})
}

// morphOwner implements attach.Owner.
type morphOwner struct{ name string }

func (o morphOwner) MorphName() string { return o.name }

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

func (morphTable) TableName() string { return "acme_morph_rows" }

// TestDeferredMorphType: MorphName wins, else the GORM table name; empty
// and nil are errors.
func TestDeferredMorphType(t *testing.T) {
	gdb := deferredDB(t, "lagoon_deferred_morph")
	if got, err := MorphType(gdb, morphOwner{name: "acme.blog.post"}); err != nil || got != "acme.blog.post" {
		t.Fatalf("owner morph %q %v", got, err)
	}
	if got, err := MorphType(gdb, &morphTable{}); err != nil || got != "acme_morph_rows" {
		t.Fatalf("table morph %q %v", got, err)
	}
	for name, model := range map[string]any{"nil": nil, "empty morph name": morphOwner{name: " "}} {
		if _, err := MorphType(gdb, model); err == nil {
			t.Fatalf("%s: no error", name)
		}
	}
	if _, err := MorphType(nil, &morphTable{}); err == nil {
		t.Fatal("nil db: no error")
	}
}

// TestDeferredConcurrentFirstBind documents a gap WinterCMS shares: two
// transactions binding the same slave for the first time can both insert,
// because nothing makes (key, field, slave) unique. The duplicate is
// harmless: a save applies both rows idempotently and deletes both. No
// unique index is added without a decision (12.2-01 open item).
func TestDeferredConcurrentFirstBind(t *testing.T) {
	gdb := deferredDB(t, "lagoon_deferred_concurrent")
	key := DeferredKey{SessionKey: "race-session-key-0123456789abcdef", AdminID: 5, MasterType: "acme_posts"}
	ctx := t.Context()
	tx1 := gdb.Begin()
	tx2 := gdb.Begin()
	defer tx1.Rollback()
	defer tx2.Rollback()
	if err := DeferredBind(ctx, tx1, key, "tags", "acme_tags", "1", nil); err != nil {
		t.Fatal(err)
	}
	// tx2 cannot see tx1's uncommitted row, so it inserts its own.
	if err := DeferredBind(ctx, tx2, key, "tags", "acme_tags", "1", nil); err != nil {
		t.Fatal(err)
	}
	if err := tx1.Commit().Error; err != nil {
		t.Fatal(err)
	}
	if err := tx2.Commit().Error; err != nil {
		t.Fatal(err)
	}
	rows := bindingRows(t, gdb)
	if len(rows) != 2 || rows[0].SlaveID != "1" || rows[1].SlaveID != "1" {
		t.Fatalf("concurrent first binds = %+v", rows)
	}
	// Reading them for a save returns both, so the commit sees and forgets
	// both.
	inTx(t, gdb, func(ctx context.Context, tx *gorm.DB) error {
		got, err := DeferredBindings(ctx, tx, key, []string{"tags"})
		if err != nil {
			return err
		}
		if len(got) != 2 {
			t.Errorf("bindings read for a save = %d", len(got))
		}
		return DeferredForget(ctx, tx, []uint{got[0].ID, got[1].ID})
	})
	if rows := bindingRows(t, gdb); len(rows) != 0 {
		t.Fatalf("rows after forget %+v", rows)
	}
}