summercms/modules/beachcomber/sync.go

package beachcomber

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"reflect"
	"strconv"

	"git.golem15.com/golem15/summercms/modules/lagoon"
	"gorm.io/gorm"
	"gorm.io/gorm/schema"
)

// operation is what a sync does with the written row's document.
type operation string

const (
	opUpsert operation = "upsert"
	opDelete operation = "delete"
)

const syncSavepoint = "beachcomber_sync"

var (
	searchableType = reflect.TypeFor[Searchable]()
	deletedAtType  = reflect.TypeFor[gorm.DeletedAt]()
)

// pending is one written row waiting for its after-commit sync. It holds
// the primary key, never the model, so the sync always reads the
// committed row.
type pending struct {
	sch   *schema.Schema
	typ   reflect.Type
	pk    any
	key   string
	index string
	op    operation
}

func (s *Service) afterCreate(db *gorm.DB) { s.afterWrite(db, opUpsert) }
func (s *Service) afterUpdate(db *gorm.DB) { s.afterWrite(db, opUpsert) }
func (s *Service) afterDelete(db *gorm.DB) { s.afterWrite(db, opDelete) }

// afterWrite registers one after-commit sync per Searchable row of the
// statement. A statement without a primary key value (a batch update or
// delete through an empty model) is skipped: no document can be built for
// it. Nothing is registered while the engine is not configured.
func (s *Service) afterWrite(db *gorm.DB, op operation) {
	if db.Error != nil || db.Statement == nil || db.Statement.Schema == nil {
		return
	}
	if s.engine == nil || !s.engine.Configured() {
		return
	}
	sch := db.Statement.Schema
	if !reflect.PointerTo(sch.ModelType).Implements(searchableType) {
		return
	}
	pkField := sch.PrioritizedPrimaryField
	if pkField == nil {
		return
	}
	ctx := db.Statement.Context
	if ctx == nil {
		ctx = context.Background()
	}
	index := s.IndexName(reflect.New(sch.ModelType).Interface().(Searchable))
	add := func(v reflect.Value) {
		for v.Kind() == reflect.Pointer {
			if v.IsNil() {
				return
			}
			v = v.Elem()
		}
		if v.Kind() != reflect.Struct || v.Type() != sch.ModelType {
			return
		}
		pk, zero := pkField.ValueOf(ctx, v)
		if zero {
			return
		}
		p := pending{sch: sch, typ: sch.ModelType, pk: pk, key: searchKey(v, pk), index: index, op: op}
		lagoon.AfterCommit(ctx, db, func(ctx context.Context, db *gorm.DB) {
			if err := s.syncOne(ctx, db, p); err != nil {
				s.warn(p, err)
			}
		})
	}
	rv := db.Statement.ReflectValue
	switch rv.Kind() {
	case reflect.Slice, reflect.Array:
		for i := 0; i < rv.Len(); i++ {
			add(rv.Index(i))
		}
	default:
		add(rv)
	}
}

// Sync upserts the document of model (a pointer to a Searchable struct
// with its primary key set) through the same path as the callbacks: the
// gates, a reload by primary key, and a delete when the row is gone, soft
// deleted or not searchable. db may be nil to use the published handle. A
// skipped sync returns nil.
func (s *Service) Sync(ctx context.Context, db *gorm.DB, model any) error {
	return s.explicit(ctx, db, model, opUpsert)
}

// Remove deletes the document of model (a pointer to a Searchable struct
// with its primary key set) through the same gated path as Sync.
func (s *Service) Remove(ctx context.Context, db *gorm.DB, model any) error {
	return s.explicit(ctx, db, model, opDelete)
}

func (s *Service) explicit(ctx context.Context, db *gorm.DB, model any, op operation) error {
	if s == nil {
		return fmt.Errorf("beachcomber: nil service")
	}
	if ctx == nil {
		ctx = context.Background()
	}
	if db == nil {
		published, ok := s.publishedDB()
		if !ok {
			return nil
		}
		db = published
	}
	rv := reflect.ValueOf(model)
	if rv.Kind() != reflect.Pointer || rv.IsNil() || rv.Elem().Kind() != reflect.Struct {
		return fmt.Errorf("beachcomber: %T is not a pointer to a model struct", model)
	}
	if _, ok := model.(Searchable); !ok {
		return fmt.Errorf("beachcomber: %T does not implement Searchable", model)
	}
	stmt := &gorm.Statement{DB: db}
	if err := stmt.Parse(model); err != nil {
		return fmt.Errorf("beachcomber: parse %T: %w", model, err)
	}
	pkField := stmt.Schema.PrioritizedPrimaryField
	if pkField == nil {
		return fmt.Errorf("beachcomber: %T has no primary key", model)
	}
	v := rv.Elem()
	pk, zero := pkField.ValueOf(ctx, v)
	if zero {
		return fmt.Errorf("beachcomber: %T has a zero primary key", model)
	}
	p := pending{
		sch:   stmt.Schema,
		typ:   stmt.Schema.ModelType,
		pk:    pk,
		key:   searchKey(v, pk),
		index: s.IndexName(model.(Searchable)),
		op:    op,
	}
	return s.syncOne(ctx, db, p)
}

// syncOne runs the gates, in order and without a request: the engine is
// configured, a database is published, the application Gate is on. It then
// reloads the row by primary key and upserts its document, or deletes it
// when the operation is a delete or the row is gone, soft deleted or not
// searchable. The engine call is bounded by the engine's own timeout; the
// caller's cancellation does not abandon it.
func (s *Service) syncOne(ctx context.Context, db *gorm.DB, p pending) (err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("panic: %v", r)
		}
	}()
	if s.engine == nil || !s.engine.Configured() {
		return nil
	}
	if _, ok := s.publishedDB(); !ok {
		return nil
	}
	if db == nil {
		return nil
	}
	ctx = context.WithoutCancel(ctx)
	sess := cleanSession(db, ctx)

	var (
		skip   bool
		remove bool
		doc    map[string]any
		idx    map[string]any
	)
	err = inSavepoint(sess, func(tx *gorm.DB) error {
		if g := s.currentGate(); g != nil && !g.Enabled(ctx, tx) {
			skip = true
			return nil
		}
		if p.op == opDelete {
			remove = true
			return nil
		}
		model := reflect.New(p.typ)
		pkCol := tx.Statement.Quote(p.sch.PrioritizedPrimaryField.DBName)
		err := tx.Unscoped().Where(pkCol+" = ?", p.pk).Take(model.Interface()).Error
		if errors.Is(err, gorm.ErrRecordNotFound) {
			remove = true
			return nil
		}
		if err != nil {
			return fmt.Errorf("reload: %w", err)
		}
		if softDeleted(ctx, p.sch, model.Elem()) {
			remove = true
			return nil
		}
		m := model.Interface().(Searchable)
		if !m.ShouldBeSearchable() {
			remove = true
			return nil
		}
		doc, err = m.ToSearchableArray(ctx, tx)
		if err != nil {
			return fmt.Errorf("build document: %w", err)
		}
		if len(doc) == 0 {
			skip = true
			return nil
		}
		if _, ok := doc["id"]; !ok {
			doc["id"] = p.key
		}
		if sp, ok := model.Interface().(IndexSchemaProvider); ok {
			idx = sp.SearchIndexSchema()
		}
		return nil
	})
	if err != nil || skip {
		return err
	}
	if remove {
		if err := s.engine.Delete(ctx, p.index, []string{p.key}); err != nil {
			return fmt.Errorf("delete: %w", err)
		}
		return nil
	}
	if err := s.engine.Upsert(ctx, p.index, idx, []map[string]any{doc}); err != nil {
		return fmt.Errorf("upsert: %w", err)
	}
	return nil
}

// cleanSession returns a handle on db's connection (the pool, or the open
// transaction) with an empty statement. db.Session with NewDB and a Context
// is not enough on a callback's handle: the Context makes it clone the
// write's statement (model, table, clauses), and a later WithContext on the
// result continues from that clone, so a gate query would run against the
// written model's table.
func cleanSession(db *gorm.DB, ctx context.Context) *gorm.DB {
	return db.Session(&gorm.Session{NewDB: true, Context: ctx}).Clauses().Session(&gorm.Session{NewDB: true})
}

// inSavepoint runs fn on db. Inside a transaction fn runs in a savepoint,
// so a failed read (a table that does not exist yet, say) is rolled back
// to it and never aborts the caller's transaction.
func inSavepoint(db *gorm.DB, fn func(tx *gorm.DB) error) error {
	if _, inTx := db.Statement.ConnPool.(gorm.TxCommitter); !inTx {
		return fn(db)
	}
	if err := db.SavePoint(syncSavepoint).Error; err != nil {
		return fmt.Errorf("savepoint: %w", err)
	}
	var err error
	func() {
		defer func() {
			if r := recover(); r != nil {
				err = fmt.Errorf("panic: %v", r)
			}
		}()
		err = fn(db)
	}()
	if err != nil {
		db.RollbackTo(syncSavepoint)
		return err
	}
	if db.Exec("RELEASE SAVEPOINT "+syncSavepoint).Error != nil {
		// A statement inside fn failed although fn did not report it (a
		// Gate counts a failed read as off): the transaction is aborted
		// and only a rollback to the savepoint makes it usable again.
		db.RollbackTo(syncSavepoint)
		db.Exec("RELEASE SAVEPOINT " + syncSavepoint)
	}
	return nil
}

// softDeleted reports whether any gorm.DeletedAt field of v is set.
func softDeleted(ctx context.Context, sch *schema.Schema, v reflect.Value) bool {
	for _, f := range sch.Fields {
		if f.FieldType != deletedAtType {
			continue
		}
		val, zero := f.ValueOf(ctx, v)
		if zero {
			continue
		}
		if d, ok := val.(gorm.DeletedAt); ok && d.Valid {
			return true
		}
	}
	return false
}

// searchKey is SearchKeyer.SearchKey when the model implements it and
// returns a key, else the decimal primary key.
func searchKey(v reflect.Value, pk any) string {
	if v.CanAddr() {
		if k, ok := v.Addr().Interface().(SearchKeyer); ok {
			if key := k.SearchKey(); key != "" {
				return key
			}
		}
	} else if k, ok := v.Interface().(SearchKeyer); ok {
		if key := k.SearchKey(); key != "" {
			return key
		}
	}
	switch n := pk.(type) {
	case uint:
		return strconv.FormatUint(uint64(n), 10)
	case uint32:
		return strconv.FormatUint(uint64(n), 10)
	case uint64:
		return strconv.FormatUint(n, 10)
	case int:
		return strconv.Itoa(n)
	case int32:
		return strconv.FormatInt(int64(n), 10)
	case int64:
		return strconv.FormatInt(n, 10)
	default:
		return fmt.Sprint(pk)
	}
}

func (s *Service) publishedDB() (*gorm.DB, bool) {
	if s.app == nil {
		return nil, false
	}
	gdb, ok := s.app.Lookup[*gorm.DB]()
	if !ok || gdb == nil {
		return nil, false
	}
	return gdb, true
}

// warn logs a failed sync with the index, key and operation. It never logs
// the document or the engine's credentials.
func (s *Service) warn(p pending, err error) {
	s.Logger().Warn("search: sync failed",
		slog.String("index", p.index),
		slog.String("key", p.key),
		slog.String("operation", string(p.op)),
		slog.String("error", err.Error()),
	)
}