Feature/service discovery

core/modules.go

@@ -4,15 +4,16 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"reflect"
+	"slices"
+	"strings"
+	"sync"
+
 	"github.com/cryptopunkscc/astrald/astral"
 	log2 "github.com/cryptopunkscc/astrald/astral/log"
 	"github.com/cryptopunkscc/astrald/core/assets"
 	"github.com/cryptopunkscc/astrald/debug"
 	"github.com/cryptopunkscc/astrald/sig"
-	"reflect"
-	"slices"
-	"strings"
-	"sync"
 )
 
 type Modules struct {

lib/astrald/services.go

@@ -0,0 +1,62 @@
+package astrald
+
+import (
+	"github.com/cryptopunkscc/astrald/astral"
+	"github.com/cryptopunkscc/astrald/astral/channel"
+	"github.com/cryptopunkscc/astrald/lib/query"
+	"github.com/cryptopunkscc/astrald/mod/services"
+)
+
+type ServicesClient struct {
+	c        *Client
+	targetID *astral.Identity
+}
+
+func NewServicesClient(c *Client, targetID *astral.Identity) *ServicesClient {
+	return &ServicesClient{c: c, targetID: targetID}
+}
+
+var defaultServicesClient *ServicesClient
+
+func Services() *ServicesClient {
+	if defaultServicesClient == nil {
+		defaultServicesClient = NewServicesClient(DefaultClient(), nil)
+	}
+	return defaultServicesClient
+}
+
+func (client *ServicesClient) Discover(ctx *astral.Context, follow bool) (<-chan *services.Update, error) {
+	ch, err := client.queryCh(ctx, "services.discover", query.Args{
+		"follow": follow,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	var out = make(chan *services.Update)
+
+	go func() {
+		defer ch.Close()
+		defer close(out)
+		for {
+			obj, err := ch.Receive()
+			if err != nil {
+				return
+			}
+			switch obj := obj.(type) {
+			case *services.Update:
+				out <- obj
+			case *astral.EOS:
+				out <- nil
+			default:
+				return
+			}
+		}
+	}()
+
+	return out, nil
+}
+
+func (client *ServicesClient) queryCh(ctx *astral.Context, method string, args any, cfg ...channel.ConfigFunc) (*channel.Channel, error) {
+	return client.c.WithTarget(client.targetID).QueryChannel(ctx, method, args, cfg...)
+}

main.go

@@ -1,3 +1,24 @@
 package main
 
-func main() {}
+import (
+	"fmt"
+
+	"github.com/cryptopunkscc/astrald/lib/astrald"
+)
+
+func main() {
+	ctx := astrald.NewContext()
+
+	ch, err := astrald.Services().Discover(ctx, true)
+	if err != nil {
+		panic(err)
+	}
+
+	for u := range ch {
+		if u == nil {
+			fmt.Println("snapshot done")
+			continue
+		}
+		fmt.Println(u.Available, u.Name, u.ProviderID)
+	}
+}

mod/all/mods.go

@@ -25,6 +25,7 @@ import (
 	_ "github.com/cryptopunkscc/astrald/mod/nodes/src"
 	_ "github.com/cryptopunkscc/astrald/mod/objects/src"
 	_ "github.com/cryptopunkscc/astrald/mod/scheduler/src"
+	_ "github.com/cryptopunkscc/astrald/mod/services/src"
 	_ "github.com/cryptopunkscc/astrald/mod/shell/src"
 	_ "github.com/cryptopunkscc/astrald/mod/tcp/src"
 	_ "github.com/cryptopunkscc/astrald/mod/tor/src"

mod/nat/src/loader.go

@@ -1,6 +1,8 @@
 package nat
 
 import (
+	"sync"
+
 	"github.com/cryptopunkscc/astrald/astral"
 	"github.com/cryptopunkscc/astrald/astral/log"
 	"github.com/cryptopunkscc/astrald/core"
@@ -14,6 +16,7 @@ func (Loader) Load(node astral.Node, assets assets.Assets, l *log.Logger) (core.
 	mod := &Module{
 		node: node,
 		log:  l,
+		cond: sync.NewCond(&sync.Mutex{}),
 	}
 
 	mod.pool = NewPairPool(mod)

mod/nat/src/module.go

@@ -1,6 +1,9 @@
 package nat
 
 import (
+	"sync"
+	"sync/atomic"
+
 	"github.com/cryptopunkscc/astrald/astral"
 	"github.com/cryptopunkscc/astrald/astral/log"
 	"github.com/cryptopunkscc/astrald/mod/dir"
@@ -32,6 +35,9 @@ type Module struct {
 
 	pool *PairPool
 	ops  shell.Scope
+
+	enabled atomic.Bool
+	cond    *sync.Cond
 }
 
 func (mod *Module) Run(ctx *astral.Context) error {
@@ -45,6 +51,12 @@ func (mod *Module) Scope() *shell.Scope {
 	return &mod.ops
 }
 
+func (mod *Module) SetEnabled(enabled bool) {
+	if mod.enabled.Swap(enabled) != enabled {
+		mod.cond.Broadcast()
+	}
+}
+
 func (mod *Module) String() string {
 	return nat.ModuleName
 }

mod/nat/src/op_enable.go

@@ -0,0 +1,22 @@
+package nat
+
+import (
+	"github.com/cryptopunkscc/astrald/astral"
+	"github.com/cryptopunkscc/astrald/astral/channel"
+	"github.com/cryptopunkscc/astrald/mod/shell"
+)
+
+type opSetEnabledArgs struct {
+	Arg bool   `query:"optional"`
+	In  string `query:"optional"`
+	Out string `query:"optional"`
+}
+
+func (mod *Module) OpSetEnabled(ctx *astral.Context, q shell.Query, args opSetEnabledArgs) (err error) {
+	ch := channel.New(q.Accept(), channel.WithFormats(args.In, args.Out))
+	defer ch.Close()
+
+	mod.SetEnabled(args.Arg)
+
+	return ch.Send(&astral.Ack{})
+}

mod/nat/src/service_discoverer.go

@@ -0,0 +1,55 @@
+package nat
+
+import (
+	"github.com/cryptopunkscc/astrald/astral"
+	"github.com/cryptopunkscc/astrald/mod/nat"
+	"github.com/cryptopunkscc/astrald/mod/services"
+)
+
+func (mod *Module) DiscoverServices(
+	ctx *astral.Context,
+	caller *astral.Identity,
+	follow bool,
+) (<-chan *services.Update, error) {
+	var ch = make(chan *services.Update, 2)
+
+	if mod.enabled.Load() {
+		ch <- mod.newServiceUpdate(true)
+	}
+
+	if !follow {
+		close(ch)
+		return ch, nil
+	}
+
+	ch <- nil
+
+	go func() {
+		<-ctx.Done()
+		mod.cond.Broadcast()
+	}()
+
+	go func() {
+		mod.cond.L.Lock()
+		defer mod.cond.L.Unlock()
+		for {
+			mod.cond.Wait()
+			select {
+			case <-ctx.Done():
+				return
+			case ch <- mod.newServiceUpdate(mod.enabled.Load()):
+			}
+
+		}
+	}()
+
+	return ch, nil
+}
+
+func (mod *Module) newServiceUpdate(available bool) *services.Update {
+	return &services.Update{
+		Available:  available,
+		Name:       nat.ModuleName,
+		ProviderID: mod.node.Identity(),
+	}
+}

mod/services/module.go

@@ -0,0 +1,15 @@
+package services
+
+import "github.com/cryptopunkscc/astrald/astral"
+
+const ModuleName = "services"
+const DBPrefix = "services__"
+
+type Module interface {
+	AddDiscoverer(Discoverer) error
+	Discoverer
+}
+
+type Discoverer interface {
+	DiscoverServices(ctx *astral.Context, caller *astral.Identity, follow bool) (<-chan *Update, error)
+}

mod/services/src/db.go

@@ -0,0 +1,58 @@
+package services
+
+import (
+	"github.com/cryptopunkscc/astrald/astral"
+	"gorm.io/gorm"
+)
+
+type DB struct {
+	db *gorm.DB
+}
+
+func (db *DB) Migrate() error {
+	return db.db.AutoMigrate(
+		&dbService{},
+	)
+}
+
+// InTx executes fn inside a database transaction.
+// This is the only transaction API - commit/rollback are fully encapsulated.
+func (db *DB) InTx(fn func(tx *DB) error) error {
+	return db.db.Transaction(func(txGorm *gorm.DB) error {
+		return fn(&DB{db: txGorm})
+	})
+}
+
+// deleteAllProviderServices deletes all cached services for a specific identity
+func (db *DB) deleteAllProviderServices(providerID *astral.Identity) error {
+	return db.db.
+		Delete(&dbService{}, "provider_id = ?", providerID).
+		Error
+}
+
+// deleteProviderService deletes a specific cached service for a specific identity
+func (db *DB) deleteProviderService(providerID *astral.Identity, name string) error {
+	return db.db.
+		Delete(&dbService{}, "name = ? AND provider_id = ?", name, providerID).
+		Error
+}
+
+// createProviderService creates a new service record
+func (db *DB) createProviderService(providerID *astral.Identity, name string, info *astral.Bundle) error {
+	return db.db.Create(&dbService{
+		Name:       name,
+		ProviderID: providerID,
+		Info:       info,
+	}).Error
+}
+
+// findProviderServices returns all current services for a specific identity
+func (db *DB) findProviderServices(providerID *astral.Identity) ([]dbService, error) {
+	var svcList []dbService
+	err := db.db.
+		Where("provider_id = ?", providerID).
+		Order("created_at DESC").
+		Find(&svcList).
+		Error
+	return svcList, err
+}

mod/services/src/db_service.go

@@ -0,0 +1,20 @@
+package services
+
+import (
+	"time"
+
+	"github.com/cryptopunkscc/astrald/astral"
+	"github.com/cryptopunkscc/astrald/mod/services"
+)
+
+// dbService represents discovered service of a specific identity cached in the database
+type dbService struct {
+	Name       string           `gorm:"uniqueIndex:idx_db_service_name_provider_id"`
+	ProviderID *astral.Identity `gorm:"uniqueIndex:idx_db_service_name_provider_id"`
+	Info       *astral.Bundle   `gorm:"serializer:json"`
+	CreatedAt  time.Time
+}
+
+func (dbService) TableName() string {
+	return services.DBPrefix + "services"
+}

mod/services/src/deps.go

@@ -0,0 +1,32 @@
+package services
+
+import (
+	"github.com/cryptopunkscc/astrald/core"
+	"github.com/cryptopunkscc/astrald/mod/dir"
+	"github.com/cryptopunkscc/astrald/mod/services"
+)
+
+type Deps struct {
+	Dir dir.Module
+}
+
+func (mod *Module) LoadDependencies() (err error) {
+	err = core.Inject(mod.node, &mod.Deps)
+	if err != nil {
+		return err
+	}
+
+	if cnode, ok := mod.node.(*core.Node); ok {
+		for _, m := range cnode.Modules().Loaded() {
+			if m == mod {
+				continue
+			}
+
+			if d, ok := m.(services.Discoverer); ok {
+				mod.AddDiscoverer(d)
+			}
+		}
+	}
+
+	return nil
+}