new indexing mechanisms

mod/all/mods.go

@@ -16,8 +16,9 @@ import (
 	_ "github.com/cryptopunkscc/astrald/mod/fs/src"
 	_ "github.com/cryptopunkscc/astrald/mod/fwd/src"
 	_ "github.com/cryptopunkscc/astrald/mod/gateway/src"
+	_ "github.com/cryptopunkscc/astrald/mod/indexing/src"
 	_ "github.com/cryptopunkscc/astrald/mod/ip/src"
-	_ "github.com/cryptopunkscc/astrald/mod/kcp/src"
+	//_ "github.com/cryptopunkscc/astrald/mod/kcp/src"
 	_ "github.com/cryptopunkscc/astrald/mod/keys/src"
 	_ "github.com/cryptopunkscc/astrald/mod/kos/src"
 	_ "github.com/cryptopunkscc/astrald/mod/log/src"

mod/indexing/errors.go

@@ -0,0 +1,5 @@
+package indexing
+
+import "errors"
+
+var ErrObjectAlreadyAdded = errors.New("object already added")

mod/indexing/indexer.go

@@ -0,0 +1,11 @@
+package indexing
+
+import (
+	"github.com/cryptopunkscc/astrald/astral"
+	"github.com/cryptopunkscc/astrald/mod/objects"
+)
+
+type Indexer interface {
+	Add(id *astral.ObjectID, repo objects.Repository) error
+	Remove(id *astral.ObjectID) error
+}

mod/indexing/module.go

@@ -0,0 +1,7 @@
+package indexing
+
+const ModuleName = "indexing"
+const DBPrefix = "indexing__"
+
+type Module interface {
+}

mod/indexing/src/config.go

@@ -0,0 +1,6 @@
+package indexing
+
+type Config struct {
+}
+
+var defaultConfig = Config{}

mod/indexing/src/db.go

@@ -0,0 +1,127 @@
+package indexing
+
+import (
+	"errors"
+
+	"github.com/cryptopunkscc/astrald/astral"
+	"github.com/cryptopunkscc/astrald/mod/indexing"
+	"gorm.io/gorm"
+)
+
+type DB struct {
+	*gorm.DB
+}
+
+func (db *DB) autoMigrate() error {
+	return db.AutoMigrate(&dbRepoEntry{})
+}
+
+func newDB(gormDB *gorm.DB) (*DB, error) {
+	db := &DB{DB: gormDB}
+
+	err := db.DB.AutoMigrate()
+	if err != nil {
+		return nil, err
+	}
+
+	return db, nil
+}
+
+func (db *DB) addToRepo(repoName string, objectID *astral.ObjectID) (err error) {
+	err = db.Transaction(func(tx *gorm.DB) error {
+		// get next version
+		var ver int
+		err = tx.Model(&dbRepoEntry{}).
+			Select("MAX(version)+1").
+			Where("repo = ?", repoName, objectID).
+			First(&ver).
+			Error
+
+		var row dbRepoEntry
+		err = tx.Where("repo = ? and object_id = ?", repoName, objectID).First(&row).Error
+		if err == nil {
+			if row.Exist {
+				return indexing.ErrObjectAlreadyAdded
+			}
+			row.Version = ver
+			row.Exist = true
+			return tx.Save(&row).Error
+		}
+
+		return tx.Create(&dbRepoEntry{
+			Repo:     repoName,
+			ObjectID: objectID,
+			Version:  ver,
+			Exist:    true,
+		}).Error
+	})
+
+	return
+}
+
+func (db *DB) removeFromRepo(repoName string, objectID *astral.ObjectID) error {
+	return db.Transaction(func(tx *gorm.DB) (err error) {
+		var row dbRepoEntry
+		err = tx.Where("repo = ? and object_id = ?", repoName, objectID).First(&row).Error
+		switch {
+		case err != nil:
+			return err
+		case !row.Exist:
+			return errors.New("row already marked as removed")
+		}
+
+		err = tx.Model(&dbRepoEntry{}).
+			Select("MAX(version)+1").
+			Where("repo = ?", repoName, objectID).
+			First(&row.Version).
+			Error
+		if err != nil {
+			return err
+		}
+
+		row.Exist = false
+
+		err = tx.Save(&row).Error
+
+		return err
+	})
+}
+
+func (db *DB) findExcessObjectIDs(repoName string, set []*astral.ObjectID) (excess []*astral.ObjectID, err error) {
+	err = db.DB.
+		Model(&dbRepoEntry{}).
+		Select("object_id").
+		Where("repo = ? and object_id not in (?) and exist = true", repoName, set).
+		Find(&excess).
+		Error
+
+	return
+}
+
+func (db *DB) findMissingObjectIDs(repoName string, set []*astral.ObjectID) (missing []*astral.ObjectID, err error) {
+	// fetch existing
+	var existing []*astral.ObjectID
+	err = db.DB.
+		Model(&dbRepoEntry{}).
+		Where("repo = ? and object_id in (?) and exist = true", repoName, set).
+		Pluck("object_id", &existing).
+		Error
+	if err != nil {
+		return nil, err
+	}
+
+	// prepare a lookup set
+	present := make(map[string]struct{}, len(existing))
+	for _, id := range existing {
+		present[id.String()] = struct{}{}
+	}
+
+	// find missing
+	for _, id := range set {
+		if _, ok := present[id.String()]; !ok {
+			missing = append(missing, id)
+		}
+	}
+
+	return
+}

mod/indexing/src/db_repo_entry.go

@@ -0,0 +1,19 @@
+package indexing
+
+import (
+	"time"
+
+	"github.com/cryptopunkscc/astrald/astral"
+	"github.com/cryptopunkscc/astrald/mod/indexing"
+)
+
+type dbRepoEntry struct {
+	Repo      string           `gorm:"index;primaryKey;uniqueIndex:idx_repo_version"`
+	ObjectID  *astral.ObjectID `gorm:"index;primaryKey"`
+	Version   int              `gorm:"index;uniqueIndex:idx_repo_version"`
+	Exist     bool             `gorm:"index"`
+	CreatedAt time.Time
+	UpdatedAt time.Time
+}
+
+func (dbRepoEntry) TableName() string { return indexing.DBPrefix + "repo_entries" }

mod/indexing/src/deps.go

@@ -0,0 +1,28 @@
+package indexing
+
+import (
+	"github.com/cryptopunkscc/astrald/astral"
+	"github.com/cryptopunkscc/astrald/core"
+	"github.com/cryptopunkscc/astrald/mod/tree"
+)
+
+func (mod *Module) LoadDependencies() (err error) {
+	ctx := astral.NewContext(nil)
+
+	err = core.Inject(mod.node, &mod.Deps)
+	if err != nil {
+		return
+	}
+
+	mod.repos, err = tree.Query(ctx, mod.Tree.Root(), "/mod/indexing/repos", true)
+	if err != nil {
+		return err
+	}
+
+	mod.indexes, err = tree.Query(ctx, mod.Tree.Root(), "/mod/indexing/indexes", true)
+	if err != nil {
+		return err
+	}
+
+	return err
+}

mod/indexing/src/loader.go

@@ -0,0 +1,43 @@
+package indexing
+
+import (
+	"github.com/cryptopunkscc/astrald/astral"
+	"github.com/cryptopunkscc/astrald/astral/log"
+	"github.com/cryptopunkscc/astrald/core"
+	"github.com/cryptopunkscc/astrald/core/assets"
+	"github.com/cryptopunkscc/astrald/mod/indexing"
+)
+
+type Loader struct{}
+
+func (Loader) Load(node astral.Node, assets assets.Assets, log *log.Logger) (core.Module, error) {
+	var err error
+	var mod = &Module{
+		node:   node,
+		config: defaultConfig,
+		log:    log,
+		assets: assets,
+	}
+
+	_ = assets.LoadYAML(indexing.ModuleName, &mod.config)
+
+	mod.ops.AddStruct(mod, "Op")
+
+	mod.db, err = newDB(assets.Database())
+	if err != nil {
+		return nil, err
+	}
+
+	err = mod.db.autoMigrate()
+	if err != nil {
+		return nil, err
+	}
+
+	return mod, err
+}
+
+func init() {
+	if err := core.RegisterModule(indexing.ModuleName, Loader{}); err != nil {
+		panic(err)
+	}
+}