Change bus

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

dbs/dbs.go

@@ -2,6 +2,7 @@ package dbs
 
 import (
 	"fmt"
+	"reflect"
 	"runtime/debug"
 	"strings"
 
@@ -102,12 +103,12 @@ func GetBson(filters *Filters) bson.D {
 			}
 		}
 		if len(orList) > 0 && len(andList) == 0 {
-			f = bson.D{{"$or", orList}}
+			f = bson.D{{Key: "$or", Value: orList}}
 		} else {
 			if len(orList) > 0 {
 				andList = append(andList, bson.M{"$or": orList})
 			}
-			f = bson.D{{"$and", andList}}
+			f = bson.D{{Key: "$and", Value: andList}}
 		}
 	}
 	return f
@@ -148,6 +149,137 @@ type Filter struct {
 	Value    interface{} `json:"value,omitempty"`
 }
 
+// FiltersFromFlatMap builds a *Filters from a map[string]interface{} whose structure
+// mirrors the JSON form of Filters:
+//
+//	{
+//	  "and": { "name": [{"operator":"like","value":"foo"}] },
+//	  "or":  { "source": [{"operator":"equal","value":"bar"}] }
+//	}
+//
+// Keys inside "and"/"or" are json tag names; the function resolves each to its
+// full dotted BSON path using the target struct. Unknown keys are kept as-is.
+func FiltersFromFlatMap(flatMap map[string]interface{}, target interface{}) *Filters {
+	filters := &Filters{
+		And: make(map[string][]Filter),
+		Or:  make(map[string][]Filter),
+	}
+	paths := jsonToBsonPaths(reflect.TypeOf(target), "")
+	resolve := func(jsonKey string) string {
+		if p, ok := paths[jsonKey]; ok {
+			return p
+		}
+		return jsonKey
+	}
+	parseFilters := func(raw interface{}) map[string][]Filter {
+		out := make(map[string][]Filter)
+		m, ok := raw.(map[string]interface{})
+		if !ok {
+			return out
+		}
+		for jsonKey, val := range m {
+			bsonKey := resolve(jsonKey)
+			items, ok := val.([]interface{})
+			if !ok {
+				continue
+			}
+			for _, item := range items {
+				entry, ok := item.(map[string]interface{})
+				if !ok {
+					continue
+				}
+				f := Filter{}
+				if op, ok := entry["operator"].(string); ok {
+					f.Operator = op
+				}
+				if v, ok := entry["value"]; ok {
+					f.Value = v
+				}
+				out[bsonKey] = append(out[bsonKey], f)
+			}
+		}
+		return out
+	}
+	if and, ok := flatMap["and"]; ok {
+		filters.And = parseFilters(and)
+	}
+	if or, ok := flatMap["or"]; ok {
+		filters.Or = parseFilters(or)
+	}
+	return filters
+}
+
+// jsonToBsonPaths recursively walks a struct type and returns a map of
+// json_name → dotted_bson_path for every field reachable from that type.
+//
+// Anonymous embedded fields without any tag follow the BSON convention of this
+// codebase: they are stored as a nested sub-document whose key is the lowercased
+// struct type name (e.g. utils.AbstractObject → "abstractobject").
+func jsonToBsonPaths(t reflect.Type, prefix string) map[string]string {
+	for t.Kind() == reflect.Ptr || t.Kind() == reflect.Slice {
+		t = t.Elem()
+	}
+	result := make(map[string]string)
+	if t.Kind() != reflect.Struct {
+		return result
+	}
+	for i := 0; i < t.NumField(); i++ {
+		field := t.Field(i)
+
+		jsonTag := field.Tag.Get("json")
+		bsonTag := field.Tag.Get("bson")
+		jsonName := strings.Split(jsonTag, ",")[0]
+		bsonName := strings.Split(bsonTag, ",")[0]
+
+		// Anonymous embedded struct with no tags: use lowercase type name as BSON prefix.
+		if field.Anonymous && jsonName == "" && bsonName == "" {
+			ft := field.Type
+			for ft.Kind() == reflect.Ptr {
+				ft = ft.Elem()
+			}
+			if ft.Kind() == reflect.Struct {
+				embedPrefix := strings.ToLower(ft.Name())
+				if prefix != "" {
+					embedPrefix = prefix + "." + embedPrefix
+				}
+				for k, v := range jsonToBsonPaths(ft, embedPrefix) {
+					if _, exists := result[k]; !exists {
+						result[k] = v
+					}
+				}
+			}
+			continue
+		}
+
+		if jsonName == "" || jsonName == "-" {
+			continue
+		}
+		if bsonName == "" || bsonName == "-" {
+			bsonName = jsonName
+		}
+
+		fullPath := bsonName
+		if prefix != "" {
+			fullPath = prefix + "." + bsonName
+		}
+
+		result[jsonName] = fullPath
+
+		ft := field.Type
+		for ft.Kind() == reflect.Ptr || ft.Kind() == reflect.Slice {
+			ft = ft.Elem()
+		}
+		if ft.Kind() == reflect.Struct {
+			for k, v := range jsonToBsonPaths(ft, fullPath) {
+				if _, exists := result[k]; !exists {
+					result[k] = v
+				}
+			}
+		}
+	}
+	return result
+}
+
 type Input = map[string]interface{}
 
 func InputToBson(i Input, isUpdate bool) bson.D {

dbs/mongo/mongo.go

@@ -282,12 +282,11 @@ func (m *MongoDB) LoadOne(id string, collection_name string) (*mongo.SingleResul
 	return res, 200, nil
 }
 
-func (m *MongoDB) Search(filters *dbs.Filters, collection_name string) (*mongo.Cursor, int, error) {
+func (m *MongoDB) Search(filters *dbs.Filters, collection_name string, offset int64, limit int64) (*mongo.Cursor, int, error) {
 	if err := m.createClient(mngoConfig.GetUrl(), false); err != nil {
 		return nil, 503, err
 	}
 	opts := options.Find()
-	opts.SetLimit(1000)
 	targetDBCollection := CollectionMap[collection_name]
 	if targetDBCollection == nil {
 		return nil, 503, errors.New("collection " + collection_name + " not initialized")
@@ -295,6 +294,9 @@ func (m *MongoDB) Search(filters *dbs.Filters, collection_name string) (*mongo.C
 
 	f := dbs.GetBson(filters)
 
+	opts.SetSkip(offset) // OFFSET
+	opts.SetLimit(limit) // LIMIT
+
 	MngoCtx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
 	// defer cancel()
 	if cursor, err := targetDBCollection.Find(
@@ -329,7 +331,8 @@ func (m *MongoDB) LoadFilter(filter map[string]interface{}, collection_name stri
 	return res, 200, nil
 }
 
-func (m *MongoDB) LoadAll(collection_name string) (*mongo.Cursor, int, error) {
+func (m *MongoDB) LoadAll(collection_name string, offset int64, limit int64) (*mongo.Cursor, int, error) {
+
 	if err := m.createClient(mngoConfig.GetUrl(), false); err != nil {
 		return nil, 503, err
 	}
@@ -337,8 +340,10 @@ func (m *MongoDB) LoadAll(collection_name string) (*mongo.Cursor, int, error) {
 
 	MngoCtx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
 	//defer cancel()
-
-	res, err := targetDBCollection.Find(MngoCtx, bson.D{})
+	findOptions := options.Find()
+	findOptions.SetSkip(offset) // OFFSET
+	findOptions.SetLimit(limit) // LIMIT
+	res, err := targetDBCollection.Find(MngoCtx, bson.D{}, findOptions)
 	if err != nil {
 		// m.Logger.Error().Msg("Couldn't find any resources. Error : " + err.Error())
 		return nil, 404, err

entrypoint.go

@@ -64,8 +64,13 @@ const (
 	PURCHASE_RESOURCE                  = tools.PURCHASE_RESOURCE
 	NATIVE_TOOL                        = tools.NATIVE_TOOL
 	EXECUTION_VERIFICATION             = tools.EXECUTION_VERIFICATION
+	ALLOWED_IMAGE                      = tools.ALLOWED_IMAGE
 )
 
+func FiltersFromFlatMap(flatMap map[string]interface{}, target interface{}) *dbs.Filters {
+	return dbs.FiltersFromFlatMap(flatMap, target)
+}
+
 func GetMySelf() (*peer.Peer, error) {
 	pp, err := utils.GetMySelf((&peer.Peer{}).GetAccessor(&tools.APIRequest{Admin: true}))
 	if pp == nil {
@@ -200,24 +205,40 @@ func ExtractTokenInfo(request http.Request) (string, string, []string) {
 		reqToken = splitToken[1]
 	}
 	if reqToken != "" {
-		token := strings.Split(reqToken, ".")
-		if len(token) > 2 {
-			bytes, err := base64.StdEncoding.DecodeString(token[2])
-			if err != nil {
-				return "", "", []string{}
-			}
-			var c Claims
-			err = json.Unmarshal(bytes, &c)
-			if err != nil {
-				return "", "", []string{}
-			}
-			return c.Session.IDToken.UserID, c.Session.IDToken.PeerID, c.Session.IDToken.Groups
-		}
+		return extractFromToken(reqToken, "user_id"), extractFromToken(reqToken, "peer_id"), strings.Split(extractFromToken(reqToken, "groups"), ",")
 	}
 	return "", "", []string{}
 }
 
-func InitAPI(appName string) {
+func extractFromToken(token string, attr string) string {
+	parts := strings.Split(token, ".")
+	if len(parts) < 2 {
+		return ""
+	}
+	payload := parts[1]
+	switch len(payload) % 4 {
+	case 2:
+		payload += "=="
+	case 3:
+		payload += "="
+	}
+	b, err := base64.URLEncoding.DecodeString(payload)
+	if err != nil {
+		return ""
+	}
+	var claims map[string]interface{}
+	if err := json.Unmarshal(b, &claims); err != nil {
+		return ""
+	}
+	ext, ok := claims["ext"].(map[string]interface{})
+	if !ok {
+		return ""
+	}
+	peerID, _ := ext[attr].(string)
+	return peerID
+}
+
+func InitAPI(appName string, extraRoutes ...map[string][]string) {
 	InitDaemon(appName)
 	beego.BConfig.Listen.HTTPPort = config.GetConfig().APIPort
 	beego.BConfig.WebConfig.DirectoryIndex = true
@@ -231,7 +252,7 @@ func InitAPI(appName string) {
 	})
 	beego.InsertFilter("*", beego.BeforeRouter, c)
 	api := &tools.API{}
-	api.Discovered(beego.BeeApp.Handlers.GetAllControllerInfo())
+	api.Discovered(beego.BeeApp.Handlers.GetAllControllerInfo(), extraRoutes...)
 }
 
 //
@@ -341,7 +362,7 @@ func (r *Request) PaymentTunnel(o *order.Order, scheduler *workflow_execution.Wo
 * @param c ...*tools.HTTPCaller
 * @return data LibDataShallow
  */
-func (r *Request) Search(filters *dbs.Filters, word string, isDraft bool) (data LibDataShallow) {
+func (r *Request) Search(filters *dbs.Filters, word string, isDraft bool, offset int64, limit int64) (data LibDataShallow) {
 	defer func() { // recover the panic
 		if r := recover(); r != nil {
 			tools.UncatchedError = append(tools.UncatchedError, errors.New("Panic recovered in Search : "+fmt.Sprintf("%v", r)))
@@ -354,7 +375,7 @@ func (r *Request) Search(filters *dbs.Filters, word string, isDraft bool) (data
 		PeerID:   r.PeerID,
 		Groups:   r.Groups,
 		Admin:    r.admin,
-	}).Search(filters, word, isDraft)
+	}).Search(filters, word, isDraft, offset, limit)
 	if err != nil {
 		data = LibDataShallow{Data: d, Code: code, Err: err.Error()}
 		return
@@ -369,7 +390,7 @@ func (r *Request) Search(filters *dbs.Filters, word string, isDraft bool) (data
 * @param c ...*tools.HTTPCaller
 * @return data LibDataShallow
  */
-func (r *Request) LoadAll(isDraft bool) (data LibDataShallow) {
+func (r *Request) LoadAll(isDraft bool, offset int64, limit int64) (data LibDataShallow) {
 	defer func() { // recover the panic
 		if r := recover(); r != nil {
 			tools.UncatchedError = append(tools.UncatchedError, errors.New("Panic recovered in LoadAll : "+fmt.Sprintf("%v", r)+" - "+string(debug.Stack())))
@@ -382,7 +403,7 @@ func (r *Request) LoadAll(isDraft bool) (data LibDataShallow) {
 		PeerID:   r.PeerID,
 		Groups:   r.Groups,
 		Admin:    r.admin,
-	}).LoadAll(isDraft)
+	}).LoadAll(isDraft, offset, limit)
 	if err != nil {
 		data = LibDataShallow{Data: d, Code: code, Err: err.Error()}
 		return
@@ -712,7 +733,7 @@ func InitNATSDecentralizedEmitter(authorizedDT ...tools.DataType) {
 					return // don't trust anyone... only friends and foes are privilege
 				}
 				access := NewRequestAdmin(LibDataEnum(resp.Datatype), nil)
-				if data := access.Search(nil, fmt.Sprintf("%v", p[resp.SearchAttr]), false); len(data.Data) > 0 {
+				if data := access.Search(nil, fmt.Sprintf("%v", p[resp.SearchAttr]), false, 0, 1); len(data.Data) > 0 {
 					delete(p, "id")
 					access.UpdateOne(p, data.Data[0].GetID())
 				} else {
@@ -731,8 +752,8 @@ func InitNATSDecentralizedEmitter(authorizedDT ...tools.DataType) {
 			access := NewRequestAdmin(LibDataEnum(resp.Datatype), nil)
 			err := json.Unmarshal(resp.Payload, &p)
 			if err == nil {
-				if data := access.Search(nil, fmt.Sprintf("%v", p[resp.SearchAttr]), false); len(data.Data) > 0 {
-					access.DeleteOne(fmt.Sprintf("%v", p[resp.SearchAttr]))
+				if data := access.Search(nil, fmt.Sprintf("%v", p[resp.SearchAttr]), false, 0, 1); len(data.Data) > 0 {
+					access.DeleteOne(data.Data[0].GetID())
 				}
 			}
 

models/allowed_image/allowed_image.go

@@ -0,0 +1,56 @@
+package allowed_image
+
+import (
+	"cloud.o-forge.io/core/oc-lib/models/utils"
+	"cloud.o-forge.io/core/oc-lib/tools"
+)
+
+// AllowedImage représente une image de conteneur autorisée à persister
+// sur un peer après l'exécution d'un workflow.
+//
+// La décision de rétention est entièrement locale au datacenter —
+// le fournisseur de processing n'a aucun levier sur cette liste.
+//
+// Règle de matching (côté oc-datacenter) :
+//   - Registry vide  = toutes les registries
+//   - TagConstraint vide = toutes les versions
+//   - TagConstraint non vide = exact ou glob (ex: "3.*", "1.2.3")
+//
+// Les entrées IsDefault sont créées au bootstrap et ne peuvent pas
+// être supprimées via l'API.
+type AllowedImage struct {
+	utils.AbstractObject
+
+	// Registry source (ex: "docker.io", "registry.example.com").
+	// Vide = wildcard, accepte n'importe quelle registry.
+	Registry string `json:"registry,omitempty" bson:"registry,omitempty"`
+
+	// Image est le nom de l'image sans registry ni tag
+	// (ex: "natsio/nats-box", "library/alpine").
+	Image string `json:"image" bson:"image" validate:"required"`
+
+	// TagConstraint est la contrainte sur le tag.
+	// Vide = toutes les versions autorisées.
+	// Supporte exact ("1.2.3") ou glob ("3.*", "*-alpine").
+	TagConstraint string `json:"tag_constraint,omitempty" bson:"tag_constraint,omitempty"`
+
+	// IsDefault marque les entrées bootstrap insérées au démarrage.
+	// Ces entrées ne peuvent pas être supprimées via l'API.
+	IsDefault bool `json:"is_default,omitempty" bson:"is_default,omitempty"`
+}
+
+func (a *AllowedImage) StoreDraftDefault() {
+	a.IsDraft = false // les allowed images sont actives immédiatement
+}
+
+func (a *AllowedImage) CanUpdate(set utils.DBObject) (bool, utils.DBObject) {
+	return true, set
+}
+
+func (a *AllowedImage) CanDelete() bool {
+	return !a.IsDefault // les entrées bootstrap sont non supprimables
+}
+
+func (a *AllowedImage) GetAccessor(request *tools.APIRequest) utils.Accessor {
+	return NewAccessor(request)
+}

models/allowed_image/allowed_image_mongo_accessor.go

@@ -0,0 +1,23 @@
+package allowed_image
+
+import (
+	"cloud.o-forge.io/core/oc-lib/logs"
+	"cloud.o-forge.io/core/oc-lib/models/utils"
+	"cloud.o-forge.io/core/oc-lib/tools"
+)
+
+type allowedImageMongoAccessor struct {
+	utils.AbstractAccessor[*AllowedImage]
+}
+
+func NewAccessor(request *tools.APIRequest) *allowedImageMongoAccessor {
+	return &allowedImageMongoAccessor{
+		AbstractAccessor: utils.AbstractAccessor[*AllowedImage]{
+			Logger:         logs.CreateLogger(tools.ALLOWED_IMAGE.String()),
+			Request:        request,
+			Type:           tools.ALLOWED_IMAGE,
+			New:            func() *AllowedImage { return &AllowedImage{} },
+			NotImplemented: []string{"CopyOne"},
+		},
+	}
+}

models/bill/bill.go

@@ -221,7 +221,7 @@ func (d *PeerItemOrder) GetPriceHT(request *tools.APIRequest) (float64, error) {
 		And: map[string][]dbs.Filter{
 			"resource_id": {{Operator: dbs.EQUAL.String(), Value: priced.GetID()}},
 		},
-	}, "", d.Purchase.IsDraft)
+	}, "", d.Purchase.IsDraft, 0, 10000)
 	if code == 200 && len(search) > 0 {
 		for _, s := range search {
 			if s.(*purchase_resource.PurchaseResource).EndDate == nil || time.Now().UTC().After(*s.(*purchase_resource.PurchaseResource).EndDate) {

models/booking/booking.go

@@ -37,6 +37,15 @@ type Booking struct {
 	// Authorization: identifies who created this draft and the Check session it belongs to.
 	// Used to verify UPDATE and DELETE orders from remote schedulers.
 	SchedulerPeerID string `json:"scheduler_peer_id,omitempty" bson:"scheduler_peer_id,omitempty"`
+
+	// Peerless is true when the booked resource has no destination peer
+	// (e.g. a public Docker Hub image). No peer confirmation or pricing
+	// negotiation is needed; the booking is stored locally only.
+	Peerless bool `json:"peerless,omitempty" bson:"peerless,omitempty"`
+
+	// OriginRef carries the registry reference of a peerless resource
+	// (e.g. "docker.io/pytorch/pytorch:2.1") so schedulers can validate it.
+	OriginRef string `json:"origin_ref,omitempty" bson:"origin_ref,omitempty"`
 }
 
 func (b *Booking) CalcDeltaOfExecution() map[string]map[string]models.MetricResume {

models/booking/planner/planner.go

@@ -74,7 +74,7 @@ func generate(request *tools.APIRequest, shallow bool) (*Planner, error) {
 		And: map[string][]dbs.Filter{
 			"expected_start_date": {{Operator: dbs.GTE.String(), Value: time.Now().UTC()}},
 		},
-	}, "*", false)
+	}, "*", false, 0, 10000)
 	if code != 200 || err != nil {
 		return nil, err
 	}
@@ -82,7 +82,7 @@ func generate(request *tools.APIRequest, shallow bool) (*Planner, error) {
 		And: map[string][]dbs.Filter{
 			"expected_start_date": {{Operator: dbs.GTE.String(), Value: time.Now().UTC()}},
 		},
-	}, "*", true)
+	}, "*", true, 0, 10000)
 	bookings := append(confirmed, drafts...)
 
 	p := &Planner{

models/collaborative_area/collaborative_area_mongo_accessor.go

@@ -91,7 +91,7 @@ func filterEnrich[T utils.ShallowDBObject](arr []string, isDrafted bool, a utils
 		Or: map[string][]dbs.Filter{
 			"abstractobject.id": {{Operator: dbs.IN.String(), Value: arr}},
 		},
-	}, "", isDrafted)
+	}, "", isDrafted, 0, int64(len(arr)))
 	if code == 200 {
 		for _, r := range res {
 			new = append(new, r.(T))

models/common/enum/status.go

@@ -32,6 +32,7 @@ const (
 	FORGOTTEN
 	DELAYED
 	CANCELLED
+	IN_PREPARATION
 )
 
 var str = [...]string{
@@ -43,6 +44,7 @@ var str = [...]string{
 	"forgotten",
 	"delayed",
 	"cancelled",
+	"in_preparation",
 }
 
 func FromInt(i int) string {
@@ -60,5 +62,5 @@ func (d BookingStatus) EnumIndex() int {
 
 // List
 func StatusList() []BookingStatus {
-	return []BookingStatus{DRAFT, SCHEDULED, STARTED, FAILURE, SUCCESS, FORGOTTEN, DELAYED, CANCELLED}
+	return []BookingStatus{DRAFT, SCHEDULED, STARTED, FAILURE, SUCCESS, FORGOTTEN, DELAYED, CANCELLED, IN_PREPARATION}
 }

models/common/models/access_configuration.go

@@ -6,8 +6,6 @@ type Container struct {
 	Args    string            `json:"args,omitempty" bson:"args,omitempty"`       // Args is the container arguments
 	Env     map[string]string `json:"env,omitempty" bson:"env,omitempty"`         // Env is the container environment variables
 	Volumes map[string]string `json:"volumes,omitempty" bson:"volumes,omitempty"` // Volumes is the container volumes
-
-	Exposes []Expose `bson:"exposes,omitempty" json:"exposes,omitempty"` // Expose is the execution
 }
 
 type Expose struct {

models/common/pricing/interfaces.go

@@ -9,6 +9,7 @@ import (
 
 type PricedItemITF interface {
 	GetID() string
+	GetName() string
 	GetInstanceID() string
 	GetType() tools.DataType
 	IsPurchasable() bool

models/common/pricing/pricing_profile.go

@@ -28,7 +28,25 @@ func RefundTypeList() []RefundType {
 	return []RefundType{REFUND_DEAD_END, REFUND_ON_ERROR, REFUND_ON_EARLY_END}
 }
 
+type PaymentType int
+
+const (
+	PAY_ONCE PaymentType = iota
+	PAY_EVERY_WEEK
+	PAY_EVERY_MONTH
+	PAY_EVERY_YEAR
+)
+
+func (t PaymentType) String() string {
+	return [...]string{"PAY ONCE", "PAY_EVERY_WEEK", "PAY_EVERY_MONTH", "PAY_EVERY_YEAR"}[t]
+}
+
+func PaymentTypeList() []PaymentType {
+	return []PaymentType{PAY_ONCE, PAY_EVERY_WEEK, PAY_EVERY_MONTH, PAY_EVERY_YEAR}
+}
+
 type AccessPricingProfile[T Strategy] struct { // only use for acces such as : DATA && PROCESSING
+	AllowedPaymentType []PaymentType      `json:"allowed_payment_type,omitempty" bson:"allowed_payment_type,omitempty"` // Price is the price of the resource
 	Pricing            PricingStrategy[T] `json:"pricing,omitempty" bson:"pricing,omitempty"`                           // Price is the price of the resource
 	DefaultRefund      RefundType         `json:"default_refund" bson:"default_refund"`                                 // DefaultRefund is the default refund type of the pricing
 	RefundRatio        int32              `json:"refund_ratio" bson:"refund_ratio" default:"0"`                         // RefundRatio is the refund ratio if missing

models/common/pricing/pricing_strategy.go

@@ -154,6 +154,8 @@ func BookingEstimation(t TimePricingStrategy, price float64, locationDurationInS
 type PricingStrategy[T Strategy] struct {
 	Price    float64 `json:"price" bson:"price" default:"0"`         // Price is the Price of the pricing
 	Currency string  `json:"currency" bson:"currency" default:"USD"` // Currency is the currency of the pricing
+
+	// NO NEED ?
 	BuyingStrategy      BuyingStrategy      `json:"buying_strategy" bson:"buying_strategy" default:"0"`             // BuyingStrategy is the buying strategy of the pricing
 	TimePricingStrategy TimePricingStrategy `json:"time_pricing_strategy" bson:"time_pricing_strategy" default:"0"` // TimePricingStrategy is the time pricing strategy of the pricing
 	OverrideStrategy    T                   `json:"override_strategy" bson:"override_strategy" default:"-1"`        // Modulation is the modulation of the pricing

models/live/live_datacenter.go

@@ -33,7 +33,7 @@ func (d *LiveDatacenter) GetAccessor(request *tools.APIRequest) utils.Accessor {
 	return NewAccessor[*LiveDatacenter](tools.LIVE_DATACENTER, request) // Create a new instance of the accessor
 }
 func (d *LiveDatacenter) GetResourceAccessor(request *tools.APIRequest) utils.Accessor {
-	return resources.NewAccessor[*resources.ComputeResource](tools.COMPUTE_RESOURCE, request, func() utils.DBObject { return &resources.ComputeResource{} })
+	return resources.NewAccessor[*resources.ComputeResource](tools.COMPUTE_RESOURCE, request)
 }
 
 func (d *LiveDatacenter) GetResource() resources.ResourceInterface {

models/live/live_mongo_accessor.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 
+	"cloud.o-forge.io/core/oc-lib/dbs"
 	"cloud.o-forge.io/core/oc-lib/logs"
 	"cloud.o-forge.io/core/oc-lib/models/utils"
 	"cloud.o-forge.io/core/oc-lib/tools"
@@ -89,3 +90,25 @@ func (a *liveMongoAccessor[T]) CopyOne(data utils.DBObject) (utils.DBObject, int
 		}
 	}
 }
+
+func (wfa *liveMongoAccessor[T]) LoadAll(isDraft bool, offset int64, limit int64) ([]utils.ShallowDBObject, int, error) {
+	return utils.GenericLoadAll[T](wfa.GetExec(isDraft), isDraft, wfa, offset, limit)
+}
+
+func (wfa *liveMongoAccessor[T]) Search(filters *dbs.Filters, search string, isDraft bool, offset int64, limit int64) ([]utils.ShallowDBObject, int, error) {
+	if filters == nil && search == "*" {
+		return utils.GenericLoadAll[T](func(d utils.DBObject) utils.ShallowDBObject {
+			return d
+		}, isDraft, wfa, offset, limit)
+	}
+	return utils.GenericSearch[T](filters, search, wfa.New().GetObjectFilters(search),
+		func(d utils.DBObject) utils.ShallowDBObject {
+			return d
+		}, isDraft, wfa, offset, limit)
+}
+
+func (a *liveMongoAccessor[T]) GetExec(isDraft bool) func(utils.DBObject) utils.ShallowDBObject {
+	return func(d utils.DBObject) utils.ShallowDBObject {
+		return d
+	}
+}

models/live/live_storage.go

@@ -29,7 +29,7 @@ func (d *LiveStorage) GetAccessor(request *tools.APIRequest) utils.Accessor {
 	return NewAccessor[*LiveStorage](tools.LIVE_STORAGE, request) // Create a new instance of the accessor
 }
 func (d *LiveStorage) GetResourceAccessor(request *tools.APIRequest) utils.Accessor {
-	return resources.NewAccessor[*resources.ComputeResource](tools.STORAGE_RESOURCE, request, func() utils.DBObject { return &resources.StorageResource{} })
+	return resources.NewAccessor[*resources.StorageResource](tools.STORAGE_RESOURCE, request)
 }
 
 func (d *LiveStorage) GetResource() resources.ResourceInterface {

models/models.go

@@ -2,6 +2,7 @@ package models
 
 import (
 	"cloud.o-forge.io/core/oc-lib/logs"
+	"cloud.o-forge.io/core/oc-lib/models/allowed_image"
 	"cloud.o-forge.io/core/oc-lib/models/bill"
 	"cloud.o-forge.io/core/oc-lib/models/execution_verification"
 	"cloud.o-forge.io/core/oc-lib/models/live"
@@ -46,6 +47,7 @@ var ModelsCatalog = map[string]func() utils.DBObject{
 	tools.LIVE_STORAGE.String():           func() utils.DBObject { return &live.LiveStorage{} },
 	tools.BILL.String():                   func() utils.DBObject { return &bill.Bill{} },
 	tools.EXECUTION_VERIFICATION.String(): func() utils.DBObject { return &execution_verification.ExecutionVerification{} },
+	tools.ALLOWED_IMAGE.String():          func() utils.DBObject { return &allowed_image.AllowedImage{} },
 }
 
 // Model returns the model object based on the model type

models/peer/peer.go

@@ -3,6 +3,7 @@ package peer
 import (
 	"fmt"
 	"strings"
+	"time"
 
 	"cloud.o-forge.io/core/oc-lib/models/utils"
 	"cloud.o-forge.io/core/oc-lib/tools"
@@ -41,11 +42,38 @@ func (m PeerRelation) EnumIndex() int {
 	return int(m)
 }
 
+// BehaviorWarning records a single misbehavior observed by a trusted service.
+type BehaviorWarning struct {
+	At          time.Time              `json:"at" bson:"at"`
+	ReporterApp string                 `json:"reporter_app" bson:"reporter_app"`
+	Severity    tools.BehaviorSeverity `json:"severity" bson:"severity"`
+	Reason      string                 `json:"reason" bson:"reason"`
+	Evidence    string                 `json:"evidence,omitempty" bson:"evidence,omitempty"`
+}
+
+// PeerLocation holds the voluntarily disclosed geographic position of a node.
+// Granularity controls how precise the location is:
+//
+//	0 = not disclosed
+//	1 = continent   (±15°)
+//	2 = country     (±3°)   — default
+//	3 = region      (±0.5°)
+//	4 = city        (±0.05°)
+//
+// The coordinates are always fuzzed by oc-discovery before publication,
+// so a granularity-2 location identifies only the rough country area.
+type PeerLocation struct {
+	Latitude    float64 `json:"latitude" bson:"latitude"`
+	Longitude   float64 `json:"longitude" bson:"longitude"`
+	Granularity int     `json:"granularity" bson:"granularity"`
+}
+
 // Peer is a struct that represents a peer
 type Peer struct {
 	utils.AbstractObject
 
 	Verify         bool   `json:"verify" bson:"verify"`
+	OrganizationID string `json:"organization_id" bson:"organization_id"`
 	PeerID         string `json:"peer_id" bson:"peer_id" validate:"required"`
 
 	APIUrl          string          `json:"api_url" bson:"api_url" validate:"required"`               // Url is the URL of the peer (base64url)
@@ -56,12 +84,56 @@ type Peer struct {
 	Relation        PeerRelation    `json:"relation" bson:"relation" default:"0"`
 	ServicesState   map[string]int  `json:"services_state,omitempty" bson:"services_state,omitempty"`
 	FailedExecution []PeerExecution `json:"failed_execution" bson:"failed_execution"` // FailedExecution is the list of failed executions, to be retried
+
+	// Location is the voluntarily disclosed (and fuzzed) geographic position.
+	Location *PeerLocation `json:"location,omitempty" bson:"location,omitempty"`
+
+	// Trust scoring — maintained by oc-discovery from PEER_BEHAVIOR_EVENT reports.
+	TrustScore       float64           `json:"trust_score" bson:"trust_score" default:"100"`
+	BlacklistReason  string            `json:"blacklist_reason,omitempty" bson:"blacklist_reason,omitempty"`
+	BehaviorWarnings []BehaviorWarning `json:"behavior_warnings,omitempty" bson:"behavior_warnings,omitempty"`
 }
 
 func (ao *Peer) VerifyAuth(callName string, request *tools.APIRequest) bool {
 	return true
 }
 
+// BlacklistThreshold is the trust score below which a peer is auto-blacklisted.
+const BlacklistThreshold = 20.0
+
+// ApplyBehaviorReport records a misbehavior, deducts the trust penalty, and
+// returns true when the trust score has fallen below BlacklistThreshold so the
+// caller can trigger the relation change.
+func (p *Peer) ApplyBehaviorReport(r tools.PeerBehaviorReport) (shouldBlacklist bool) {
+	p.BehaviorWarnings = append(p.BehaviorWarnings, BehaviorWarning{
+		At:          r.At,
+		ReporterApp: r.ReporterApp,
+		Severity:    r.Severity,
+		Reason:      r.Reason,
+		Evidence:    r.Evidence,
+	})
+	if p.TrustScore == 0 {
+		p.TrustScore = 100 // initialise if never set
+	}
+	p.TrustScore -= r.Severity.Penalty()
+	if p.TrustScore < 0 {
+		p.TrustScore = 0
+	}
+	if p.TrustScore <= BlacklistThreshold {
+		p.BlacklistReason = r.Reason
+		return true
+	}
+	return false
+}
+
+// ResetTrust clears all behavior history and resets the trust score to 100.
+// Must be called when a peer relation is manually set to NONE or PARTNER.
+func (p *Peer) ResetTrust() {
+	p.TrustScore = 100
+	p.BlacklistReason = ""
+	p.BehaviorWarnings = nil
+}
+
 // AddExecution adds an execution to the list of failed executions
 func (ao *Peer) AddExecution(exec PeerExecution) {
 	found := false
@@ -96,7 +168,3 @@ func (d *Peer) GetAccessor(request *tools.APIRequest) utils.Accessor {
 	data := NewAccessor(request) // Create a new instance of the accessor
 	return data
 }
-
-func (r *Peer) CanDelete() bool {
-	return false // only draft order can be deleted
-}

models/peer/peer_mongo_accessor.go

@@ -42,6 +42,23 @@ func (wfa *peerMongoAccessor) ShouldVerifyAuth() bool {
 	return !wfa.OverrideAuth
 }
 
+/*
+TODO : organization_ID est un peer_ID duquel on se revendique faire parti.
+Ca implique une clé d'organisation + une demande d'intégration.
+
+Slave-Master IOT
+*/
+func (dca *peerMongoAccessor) StoreOne(data utils.DBObject) (utils.DBObject, int, error) {
+	pp, _ := utils.GetMySelf(NewAccessor(&tools.APIRequest{Admin: true}))
+	if data != nil {
+		d := data.(*Peer)
+		if pp != nil && d.OrganizationID != "" && d.OrganizationID == pp.(*Peer).OrganizationID {
+			d.Relation = PARTNER // defaulting on partner if same organization.
+		}
+	}
+	return utils.GenericStoreOne(data, dca)
+}
+
 /*
 * Nothing special here, just the basic CRUD operations
  */

models/resources/compute.go

@@ -25,7 +25,7 @@ type ComputeResource struct {
 }
 
 func (d *ComputeResource) GetAccessor(request *tools.APIRequest) utils.Accessor {
-	return NewAccessor[*ComputeResource](tools.COMPUTE_RESOURCE, request, func() utils.DBObject { return &ComputeResource{} })
+	return NewAccessor[*ComputeResource](tools.COMPUTE_RESOURCE, request)
 }
 
 func (r *ComputeResource) GetType() string {
@@ -65,6 +65,10 @@ type ComputeResourceInstance struct {
 	Nodes              []*ComputeNode         `json:"nodes,omitempty" bson:"nodes,omitempty"`
 }
 
+// IsPeerless is always false for compute instances: a compute resource is
+// infrastructure owned by a peer and can never be declared peerless.
+func (ri *ComputeResourceInstance) IsPeerless() bool { return false }
+
 func NewComputeResourceInstance(name string, peerID string) ResourceInstanceITF {
 	return &ComputeResourceInstance{
 		ResourceInstance: ResourceInstance[*ComputeResourcePartnership]{

models/resources/data.go

@@ -30,13 +30,22 @@ type DataResource struct {
 }
 
 func (d *DataResource) GetAccessor(request *tools.APIRequest) utils.Accessor {
-	return NewAccessor[*DataResource](tools.DATA_RESOURCE, request, func() utils.DBObject { return &DataResource{} }) // Create a new instance of the accessor
+	return NewAccessor[*DataResource](tools.DATA_RESOURCE, request) // Create a new instance of the accessor
 }
 
 func (r *DataResource) GetType() string {
 	return tools.DATA_RESOURCE.String()
 }
 
+func (ri *DataResource) StoreDraftDefault() {
+	ri.AbstractObject.StoreDraftDefault()
+	ri.Env = append(ri.Env, models.Param{
+		Attr:     "source",
+		Value:    "[resource]instance.source",
+		Readonly: true,
+	})
+}
+
 func (abs *DataResource) ConvertToPricedResource(t tools.DataType, selectedInstance *int, selectedPartnership *int, selectedBuyingStrategy *int, selectedStrategy *int, selectedBookingModeIndex *int, request *tools.APIRequest) (pricing.PricedItemITF, error) {
 	if t != tools.DATA_RESOURCE {
 		return nil, errors.New("not the proper type expected : cannot convert to priced resource : have " + t.String() + " wait Data")
@@ -67,24 +76,6 @@ func NewDataInstance(name string, peerID string) ResourceInstanceITF {
 	}
 }
 
-func (ri *DataInstance) StoreDraftDefault() {
-	found := false
-	for _, p := range ri.ResourceInstance.Env {
-		if p.Attr == "source" {
-			found = true
-			break
-		}
-	}
-	if !found {
-		ri.ResourceInstance.Env = append(ri.ResourceInstance.Env, models.Param{
-			Attr:     "source",
-			Value:    ri.Source,
-			Readonly: true,
-		})
-	}
-	ri.ResourceInstance.StoreDraftDefault()
-}
-
 type DataResourcePartnership struct {
 	ResourcePartnerShip[*DataResourcePricingProfile]
 	MaxDownloadableGbAllowed      float64 `json:"allowed_gb,omitempty" bson:"allowed_gb,omitempty"`
@@ -95,7 +86,7 @@ type DataResourcePartnership struct {
 type DataResourcePricingStrategy int
 
 const (
-	PER_DOWNLOAD DataResourcePricingStrategy = iota + 6
+	PER_DOWNLOAD DataResourcePricingStrategy = iota + 7
 	PER_TB_DOWNLOADED
 	PER_GB_DOWNLOADED
 	PER_MB_DOWNLOADED

models/resources/interfaces.go

@@ -3,6 +3,7 @@ package resources
 import (
 	"cloud.o-forge.io/core/oc-lib/dbs"
 	"cloud.o-forge.io/core/oc-lib/models/booking"
+	"cloud.o-forge.io/core/oc-lib/models/common/models"
 	"cloud.o-forge.io/core/oc-lib/models/common/pricing"
 	"cloud.o-forge.io/core/oc-lib/models/utils"
 	"cloud.o-forge.io/core/oc-lib/tools"
@@ -19,17 +20,23 @@ type ResourceInterface interface {
 	ConvertToPricedResource(t tools.DataType, a *int, selectedPartnership *int, selectedBuyingStrategy *int, selectedStrategy *int, b *int, request *tools.APIRequest) (pricing.PricedItemITF, error)
 	GetType() string
 	ClearEnv() utils.DBObject
+	VerifyBuy()
 	SetAllowedInstances(request *tools.APIRequest, instance_id ...string) []ResourceInstanceITF
 	AddInstances(instance ResourceInstanceITF)
 	GetSelectedInstance(index *int) ResourceInstanceITF
+	StoreDraftDefault()
+
+	GetEnv() []models.Param
+	GetInputs() []models.Param
+	GetOutputs() []models.Param
 }
 
 type ResourceInstanceITF interface {
 	utils.DBObject
 	GetID() string
 	GetName() string
-	StoreDraftDefault()
-	ClearEnv()
+	GetOrigin() OriginMeta
+	IsPeerless() bool
 	FilterInstance(peerID string)
 	GetProfile(peerID string, partnershipIndex *int, buying *int, strategy *int) pricing.PricingProfileITF
 	GetPricingsProfiles(peerID string, groups []string) []pricing.PricingProfileITF

models/resources/native_tools.go

@@ -23,7 +23,7 @@ func (d *NativeTool) SetName(name string) {
 }
 
 func (d *NativeTool) GetAccessor(request *tools.APIRequest) utils.Accessor {
-	return NewAccessor[*NativeTool](tools.NATIVE_TOOL, request, func() utils.DBObject { return &NativeTool{} })
+	return NewAccessor[*NativeTool](tools.NATIVE_TOOL, request)
 }
 
 func (r *NativeTool) AddInstances(instance ResourceInstanceITF) {
@@ -61,7 +61,7 @@ func InitNative() {
 	for _, kind := range []native_tools.NativeToolsEnum{native_tools.WORKFLOW_EVENT} {
 		newNative := &NativeTool{}
 		access := newNative.GetAccessor(&tools.APIRequest{Admin: true})
-		l, _, err := access.Search(nil, kind.String(), false)
+		l, _, err := access.Search(nil, kind.String(), false, 0, 10)
 		if err != nil || len(l) == 0 {
 			newNative.Name = kind.String()
 			newNative.Kind = int(kind)

models/resources/origin.go

@@ -0,0 +1,31 @@
+package resources
+
+// OriginType qualifies where a resource instance comes from.
+type OriginType int
+
+const (
+	// OriginPeer: instance offered by a known network peer (default).
+	OriginPeer OriginType = iota
+	// OriginPublic: instance from a public registry (Docker Hub, HuggingFace, etc.).
+	// No peer confirmation is needed; access is unrestricted.
+	OriginPublic
+	// OriginSelf: self-hosted instance with no third-party peer.
+	OriginSelf
+)
+
+// OriginMeta carries provenance information for a resource instance.
+type OriginMeta struct {
+	Type     OriginType `json:"origin_type" bson:"origin_type"`
+	Ref      string     `json:"origin_ref,omitempty" bson:"origin_ref,omitempty"`         // e.g. "docker.io/pytorch/pytorch:2.1"
+	License  string     `json:"origin_license,omitempty" bson:"origin_license,omitempty"` // SPDX identifier or free-form
+	Verified bool       `json:"origin_verified" bson:"origin_verified"`                   // manually vetted by an OC admin
+}
+
+// IsPeerless MUST NOT be used for authorization decisions.
+// Use ResourceInstance.IsPeerless() instead, which derives the property
+// from structural invariants rather than this self-declared field.
+//
+// This method is kept only for display/logging purposes.
+func (o OriginMeta) DeclaredPeerless() bool {
+	return o.Type != OriginPeer
+}

models/resources/priced_resource.go

@@ -47,6 +47,10 @@ func (abs *PricedResource[T]) GetID() string {
 	return abs.ResourceID
 }
 
+func (abs *PricedResource[T]) GetName() string {
+	return abs.Name
+}
+
 func (abs *PricedResource[T]) GetInstanceID() string {
 	return abs.InstanceID
 }

models/resources/processing.go

@@ -48,6 +48,8 @@ type ProcessingResourceAccess struct {
 type ProcessingInstance struct {
 	ResourceInstance[*ResourcePartnerShip[*ProcessingResourcePricingProfile]]
 	Access      *ProcessingResourceAccess `json:"access,omitempty" bson:"access,omitempty"` // Access is the access
+	SizeGB      int                       `json:"size_gb,omitempty" bson:"size_gb,omitempty"`
+	ContentType string                    `json:"content_type,omitempty" bson:"content_type,omitempty"`
 }
 
 func NewProcessingInstance(name string, peerID string) ResourceInstanceITF {
@@ -112,7 +114,7 @@ func (a *PricedProcessingResource) GetExplicitDurationInS() float64 {
 }
 
 func (d *ProcessingResource) GetAccessor(request *tools.APIRequest) utils.Accessor {
-	return NewAccessor[*ProcessingResource](tools.PROCESSING_RESOURCE, request, func() utils.DBObject { return &ProcessingResource{} }) // Create a new instance of the accessor
+	return NewAccessor[*ProcessingResource](tools.PROCESSING_RESOURCE, request) // Create a new instance of the accessor
 }
 
 func (abs *ProcessingResource) ConvertToPricedResource(t tools.DataType, selectedInstance *int, selectedPartnership *int, selectedBuyingStrategy *int, selectedStrategy *int, selectedBookingModeIndex *int, request *tools.APIRequest) (pricing.PricedItemITF, error) {

models/resources/resource.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"slices"
+	"time"
 
 	"cloud.o-forge.io/core/oc-lib/config"
 	"cloud.o-forge.io/core/oc-lib/dbs"
@@ -11,6 +12,7 @@ import (
 	"cloud.o-forge.io/core/oc-lib/models/common/models"
 	"cloud.o-forge.io/core/oc-lib/models/common/pricing"
 	"cloud.o-forge.io/core/oc-lib/models/peer"
+	"cloud.o-forge.io/core/oc-lib/models/resources/purchase_resource"
 	"cloud.o-forge.io/core/oc-lib/models/utils"
 	"cloud.o-forge.io/core/oc-lib/tools"
 	"github.com/biter777/countries"
@@ -20,7 +22,7 @@ import (
 // AbstractResource is the struct containing all of the attributes commons to all ressources
 type AbstractResource struct {
 	utils.AbstractObject                                                   // AbstractObject contains the basic fields of an object (id, name)
-
+	PurchaseInfo         *purchase_resource.PurchaseResource               `json:"purchase_info,omitempty"`                                        // is_buy precise if a resource is buy or not
 	Type                 string                                            `json:"type,omitempty" bson:"type,omitempty"`                           // Type is the type of the resource
 	Logo                 string                                            `json:"logo,omitempty" bson:"logo,omitempty"`                           // Logo is the logo of the resource
 	Description          string                                            `json:"description,omitempty" bson:"description,omitempty"`             // Description is the description of the resource
@@ -28,12 +30,47 @@ type AbstractResource struct {
 	Owners               []utils.Owner                                     `json:"owners,omitempty" bson:"owners,omitempty"`                       // Owners is the list of owners of the resource
 	UsageRestrictions    string                                            `bson:"usage_restrictions,omitempty" json:"usage_restrictions,omitempty"`
 	AllowedBookingModes  map[booking.BookingMode]*pricing.PricingVariation `bson:"allowed_booking_modes" json:"allowed_booking_modes"`
+
+	Env     []models.Param `json:"env,omitempty" bson:"env,omitempty"`
+	Inputs  []models.Param `json:"inputs,omitempty" bson:"inputs,omitempty"`
+	Outputs []models.Param `json:"outputs,omitempty" bson:"outputs,omitempty"`
+}
+
+func (abs *AbstractResource) VerifyBuy() {
+	p := &purchase_resource.PurchaseResource{}
+	access := p.GetAccessor(&tools.APIRequest{Admin: true})
+
+	purchase, _, _ := access.Search(&dbs.Filters{
+		And: map[string][]dbs.Filter{
+			"resource_id": {{Operator: dbs.EQUAL.String(), Value: abs.GetID()}},
+		},
+	}, "", false, 0, 1)
+	if len(purchase) > 0 {
+		abs.PurchaseInfo = purchase[0].(*purchase_resource.PurchaseResource)
+	}
+}
+
+func (abs *AbstractResource) GetEnv() []models.Param {
+	return abs.Env
+}
+func (abs *AbstractResource) GetInputs() []models.Param {
+	return abs.Inputs
+}
+func (abs *AbstractResource) GetOutputs() []models.Param {
+	return abs.Outputs
 }
 
 func (abs *AbstractResource) FilterPeer(peerID string) *dbs.Filters {
 	return nil
 }
 
+func (ri *AbstractResource) ClearEnv() utils.DBObject {
+	ri.Env = []models.Param{}
+	ri.Inputs = []models.Param{}
+	ri.Outputs = []models.Param{}
+	return ri
+}
+
 func (r *AbstractResource) GetBookingModes() map[booking.BookingMode]*pricing.PricingVariation {
 	if len(r.AllowedBookingModes) == 0 {
 		return map[booking.BookingMode]*pricing.PricingVariation{
@@ -52,17 +89,13 @@ func (r *AbstractResource) GetType() string {
 }
 
 func (r *AbstractResource) StoreDraftDefault() {
-	r.IsDraft = true
+	//r.IsDraft = true pour le moment on passe outre.
 }
 
 func (r *AbstractResource) CanUpdate(set utils.DBObject) (bool, utils.DBObject) {
 	return r.IsDraft, set
 }
 
-func (r *AbstractResource) CanDelete() bool {
-	return r.IsDraft // only draft bookings can be deleted
-}
-
 type AbstractInstanciatedResource[T ResourceInstanceITF] struct {
 	AbstractResource // AbstractResource contains the basic fields of an object (id, name)
 
@@ -136,13 +169,6 @@ func ConvertToPricedResource[T pricing.PricingProfileITF](t tools.DataType,
 	}, nil
 }
 
-func (abs *AbstractInstanciatedResource[T]) ClearEnv() utils.DBObject {
-	for _, instance := range abs.Instances {
-		instance.ClearEnv()
-	}
-	return abs
-}
-
 func (r *AbstractInstanciatedResource[T]) GetSelectedInstance(selected *int) ResourceInstanceITF {
 	if selected != nil && len(r.Instances) > *selected {
 		return r.Instances[*selected]
@@ -175,13 +201,20 @@ func VerifyAuthAction[T ResourceInstanceITF](baseInstance []T, request *tools.AP
 		if len(instanceID) > 0 && !slices.Contains(instanceID, instance.GetID()) {
 			continue
 		}
+		// Structurally peerless instances (no creator, no partnerships, non-empty Ref)
+		// are freely accessible by any requester.
+		if instance.IsPeerless() {
+			instances = append(instances, instance)
+			continue
+		}
 		_, peerGroups := instance.GetPeerGroups()
 		for _, peers := range peerGroups {
 			if request == nil {
 				continue
 			}
-			if grps, ok := peers[request.PeerID]; ok || config.GetConfig().Whitelist {
-				if (ok && slices.Contains(grps, "*")) || (!ok && config.GetConfig().Whitelist) {
+			_, allOK := peers["*"]
+			if grps, ok := peers[request.PeerID]; ok || allOK || config.GetConfig().Whitelist {
+				if allOK || (ok && slices.Contains(grps, "*")) || (!ok && config.GetConfig().Whitelist) {
 					instance.FilterInstance(request.PeerID)
 					instances = append(instances, instance)
 					// TODO filter Partners + Profiles...
@@ -206,12 +239,12 @@ type GeoPoint struct {
 
 type ResourceInstance[T ResourcePartnerITF] struct {
 	utils.AbstractObject
+	ContentType    string                `json:"content_type,omitempty" bson:"content_type,omitempty"`
+	LastUpdate     time.Time             `json:"last_update,omitempty" bson:"last_update,omitempty"`
+	Origin         OriginMeta            `json:"origin,omitempty" bson:"origin,omitempty"`
 	Location       GeoPoint              `json:"location,omitempty" bson:"location,omitempty"`
 	Country        countries.CountryCode `json:"country,omitempty" bson:"country,omitempty"`
 	AccessProtocol string                `json:"access_protocol,omitempty" bson:"access_protocol,omitempty"`
-	Env            []models.Param        `json:"env,omitempty" bson:"env,omitempty"`
-	Inputs         []models.Param        `json:"inputs,omitempty" bson:"inputs,omitempty"`
-	Outputs        []models.Param        `json:"outputs,omitempty" bson:"outputs,omitempty"`
 
 	Partnerships []T `json:"partnerships,omitempty" bson:"partnerships,omitempty"`
 
@@ -231,10 +264,23 @@ func NewInstance[T ResourcePartnerITF](name string) *ResourceInstance[T] {
 	}
 }
 
+func (ri *ResourceInstance[T]) GetOrigin() OriginMeta {
+	return ri.Origin
+}
+
+// IsPeerless returns true when the instance has no owning peer and a non-empty
+// registry reference. This is derived from structural invariants — NOT from the
+// self-declared Origin.Type field — to prevent auth bypass via metadata manipulation:
+//
+//	CreatorID == "" ∧ len(Partnerships) == 0 ∧ Origin.Ref != ""
+func (ri *ResourceInstance[T]) IsPeerless() bool {
+	return ri.CreatorID == "" && len(ri.Partnerships) == 0 && ri.Origin.Ref != ""
+}
+
 func (ri *ResourceInstance[T]) FilterInstance(peerID string) {
 	partnerships := []T{}
 	for _, p := range ri.Partnerships {
-		if p.GetPeerGroups()[peerID] != nil {
+		if p.GetPeerGroups()["*"] != nil || p.GetPeerGroups()[peerID] != nil {
 			p.FilterPartnership(peerID)
 			partnerships = append(partnerships, p)
 		}
@@ -242,13 +288,10 @@ func (ri *ResourceInstance[T]) FilterInstance(peerID string) {
 	ri.Partnerships = partnerships
 }
 
-func (ri *ResourceInstance[T]) ClearEnv() {
-	ri.Env = []models.Param{}
-	ri.Inputs = []models.Param{}
-	ri.Outputs = []models.Param{}
-}
-
 func (ri *ResourceInstance[T]) GetProfile(peerID string, partnershipIndex *int, buyingIndex *int, strategyIndex *int) pricing.PricingProfileITF {
+	if ri.IsPeerless() {
+		return pricing.GetDefaultPricingProfile()
+	}
 	if partnershipIndex != nil && len(ri.Partnerships) > *partnershipIndex {
 		prts := ri.Partnerships[*partnershipIndex]
 		return prts.GetProfile(buyingIndex, strategyIndex)
@@ -262,6 +305,9 @@ func (ri *ResourceInstance[T]) GetProfile(peerID string, partnershipIndex *int,
 }
 
 func (ri *ResourceInstance[T]) GetPricingsProfiles(peerID string, groups []string) []pricing.PricingProfileITF {
+	if ri.IsPeerless() {
+		return []pricing.PricingProfileITF{pricing.GetDefaultPricingProfile()}
+	}
 	pricings := []pricing.PricingProfileITF{}
 	for _, p := range ri.Partnerships {
 		pricings = append(pricings, p.GetPricingsProfiles(peerID, groups)...)
@@ -277,6 +323,10 @@ func (ri *ResourceInstance[T]) GetPricingsProfiles(peerID string, groups []strin
 }
 
 func (ri *ResourceInstance[T]) GetPeerGroups() ([]ResourcePartnerITF, []map[string][]string) {
+	// Structurally peerless: universally accessible — wildcard on all peers.
+	if ri.IsPeerless() {
+		return []ResourcePartnerITF{}, []map[string][]string{{"*": {"*"}}}
+	}
 	groups := []map[string][]string{}
 	partners := []ResourcePartnerITF{}
 	for _, p := range ri.Partnerships {
@@ -325,11 +375,17 @@ type ResourcePartnerShip[T pricing.PricingProfileITF] struct {
 }
 
 func (ri *ResourcePartnerShip[T]) FilterPartnership(peerID string) {
-	if ri.PeerGroups[peerID] == nil {
-		ri.PeerGroups = map[string][]string{}
-	} else {
+	if ri.PeerGroups["*"] == nil && ri.PeerGroups[peerID] == nil {
 		ri.PeerGroups = map[string][]string{
-			peerID: ri.PeerGroups[peerID],
+			"*": {"*"},
+		}
+	} else {
+		ri.PeerGroups = map[string][]string{}
+		if ri.PeerGroups["*"] != nil {
+			ri.PeerGroups["*"] = ri.PeerGroups["*"]
+		}
+		if ri.PeerGroups[peerID] != nil {
+			ri.PeerGroups[peerID] = ri.PeerGroups[peerID]
 		}
 	}
 }
@@ -355,7 +411,15 @@ Une bill (facture) représente alors... l'emission d'une facture à un instant T
 */
 func (ri *ResourcePartnerShip[T]) GetPricingsProfiles(peerID string, groups []string) []pricing.PricingProfileITF {
 	profiles := []pricing.PricingProfileITF{}
-	if ri.PeerGroups[peerID] == nil {
+	if ri.PeerGroups["*"] == nil && ri.PeerGroups[peerID] == nil {
+		return profiles
+	}
+	if ri.PeerGroups["*"] != nil {
+		for _, ri := range ri.PricingProfiles {
+			for _, i := range ri {
+				profiles = append(profiles, i)
+			}
+		}
 		return profiles
 	}
 	for _, p := range ri.PeerGroups[peerID] {
@@ -387,6 +451,7 @@ func (rp *ResourcePartnerShip[T]) GetPeerGroups() map[string][]string {
 			return rp.PeerGroups
 		}
 		return map[string][]string{
+			"*":        {"*"},
 			pp.GetID(): {"*"},
 		}
 	}

models/resources/resource_accessor.go

@@ -12,11 +12,10 @@ import (
 
 type ResourceMongoAccessor[T ResourceInterface] struct {
 	utils.AbstractAccessor[ResourceInterface] // AbstractAccessor contains the basic fields of an accessor (model, caller)
-	generateData                              func() utils.DBObject
 }
 
 // New creates a new instance of the computeMongoAccessor
-func NewAccessor[T ResourceInterface](t tools.DataType, request *tools.APIRequest, g func() utils.DBObject) *ResourceMongoAccessor[T] {
+func NewAccessor[T ResourceInterface](t tools.DataType, request *tools.APIRequest) *ResourceMongoAccessor[T] {
 	if !slices.Contains([]tools.DataType{
 		tools.COMPUTE_RESOURCE, tools.STORAGE_RESOURCE,
 		tools.PROCESSING_RESOURCE, tools.WORKFLOW_RESOURCE,
@@ -47,7 +46,6 @@ func NewAccessor[T ResourceInterface](t tools.DataType, request *tools.APIReques
 				return nil
 			},
 		},
-		generateData: g,
 	}
 }
 
@@ -55,6 +53,16 @@ func NewAccessor[T ResourceInterface](t tools.DataType, request *tools.APIReques
 * Nothing special here, just the basic CRUD operations
  */
 
+func (dca *ResourceMongoAccessor[T]) LoadOne(id string) (utils.DBObject, int, error) {
+	data, code, err := dca.AbstractAccessor.LoadOne(id)
+	if err == nil {
+		data.(T).VerifyBuy()
+		data.(T).SetAllowedInstances(dca.Request)
+		return data, code, err
+	}
+	return data, code, err
+}
+
 func (dca *ResourceMongoAccessor[T]) UpdateOne(set map[string]interface{}, id string) (utils.DBObject, int, error) {
 	if dca.GetType() == tools.COMPUTE_RESOURCE {
 		return nil, 404, errors.New("can't update a non existing computing units resource not reported onto compute units catalog")
@@ -80,26 +88,29 @@ func (dca *ResourceMongoAccessor[T]) CopyOne(data utils.DBObject) (utils.DBObjec
 	return dca.StoreOne(data)
 }
 
-func (wfa *ResourceMongoAccessor[T]) LoadAll(isDraft bool) ([]utils.ShallowDBObject, int, error) {
-	return utils.GenericLoadAll[T](wfa.GetExec(isDraft), isDraft, wfa)
+func (wfa *ResourceMongoAccessor[T]) LoadAll(isDraft bool, offset int64, limit int64) ([]utils.ShallowDBObject, int, error) {
+	return utils.GenericLoadAll[T](wfa.GetExec(isDraft), isDraft, wfa, offset, limit)
 }
 
-func (wfa *ResourceMongoAccessor[T]) Search(filters *dbs.Filters, search string, isDraft bool) ([]utils.ShallowDBObject, int, error) {
+func (wfa *ResourceMongoAccessor[T]) Search(filters *dbs.Filters, search string, isDraft bool, offset int64, limit int64) ([]utils.ShallowDBObject, int, error) {
 	if filters == nil && search == "*" {
 		return utils.GenericLoadAll[T](func(d utils.DBObject) utils.ShallowDBObject {
+			d.(T).VerifyBuy()
 			d.(T).SetAllowedInstances(wfa.Request)
 			return d
-		}, isDraft, wfa)
+		}, isDraft, wfa, offset, limit)
 	}
 	return utils.GenericSearch[T](filters, search, wfa.GetObjectFilters(search),
 		func(d utils.DBObject) utils.ShallowDBObject {
+			d.(T).VerifyBuy()
 			d.(T).SetAllowedInstances(wfa.Request)
 			return d
-		}, isDraft, wfa)
+		}, isDraft, wfa, offset, limit)
 }
 
 func (a *ResourceMongoAccessor[T]) GetExec(isDraft bool) func(utils.DBObject) utils.ShallowDBObject {
 	return func(d utils.DBObject) utils.ShallowDBObject {
+		d.(T).VerifyBuy()
 		d.(T).SetAllowedInstances(a.Request)
 		return d
 	}
@@ -108,12 +119,11 @@ func (a *ResourceMongoAccessor[T]) GetExec(isDraft bool) func(utils.DBObject) ut
 func (abs *ResourceMongoAccessor[T]) GetObjectFilters(search string) *dbs.Filters {
 	return &dbs.Filters{
 		Or: map[string][]dbs.Filter{ // filter by like name, short_description, description, owner, url if no filters are provided
-			"abstractintanciatedresource.abstractresource.abstractobject.name":       {{Operator: dbs.LIKE.String(), Value: search}},
-			"abstractintanciatedresource.abstractresource.type":                      {{Operator: dbs.LIKE.String(), Value: search}},
-			"abstractintanciatedresource.abstractresource.short_description":         {{Operator: dbs.LIKE.String(), Value: search}},
-			"abstractintanciatedresource.abstractresource.description":               {{Operator: dbs.LIKE.String(), Value: search}},
-			"abstractintanciatedresource.abstractresource.owners.name":               {{Operator: dbs.LIKE.String(), Value: search}},
-			"abstractintanciatedresource.abstractresource.abstractobject.creator_id": {{Operator: dbs.EQUAL.String(), Value: search}},
+			"abstractinstanciatedresource.abstractresource.abstractobject.name": {{Operator: dbs.LIKE.String(), Value: search}},
+			"abstractinstanciatedresource.abstractresource.type":                {{Operator: dbs.LIKE.String(), Value: search}},
+			"abstractinstanciatedresource.abstractresource.short_description":   {{Operator: dbs.LIKE.String(), Value: search}},
+			"abstractinstanciatedresource.abstractresource.description":         {{Operator: dbs.LIKE.String(), Value: search}},
+			"abstractinstanciatedresource.abstractresource.owners.name":         {{Operator: dbs.LIKE.String(), Value: search}},
 		},
 	}
 }

models/resources/storage.go

@@ -23,7 +23,7 @@ type StorageResource struct {
 }
 
 func (d *StorageResource) GetAccessor(request *tools.APIRequest) utils.Accessor {
-	return NewAccessor[*StorageResource](tools.STORAGE_RESOURCE, request, func() utils.DBObject { return &StorageResource{} }) // Create a new instance of the accessor
+	return NewAccessor[*StorageResource](tools.STORAGE_RESOURCE, request) // Create a new instance of the accessor
 }
 
 func (r *StorageResource) GetType() string {
@@ -44,6 +44,15 @@ func (abs *StorageResource) ConvertToPricedResource(t tools.DataType, selectedIn
 	}, nil
 }
 
+func (ri *StorageResource) StoreDraftDefault() {
+	ri.AbstractObject.StoreDraftDefault()
+	ri.Env = append(ri.Env, models.Param{
+		Attr:     "source",
+		Value:    "[resource]instance.source",
+		Readonly: true,
+	})
+}
+
 type StorageResourceInstance struct {
 	ResourceInstance[*StorageResourcePartnership]
 	Source        string           `bson:"source,omitempty" json:"source,omitempty"` // Source is the source of the storage
@@ -57,6 +66,10 @@ type StorageResourceInstance struct {
 	Throughput    string           `bson:"throughput,omitempty" json:"throughput,omitempty"` // Throughput is the throughput of the storage
 }
 
+// IsPeerless is always false for storage instances: a storage resource is
+// infrastructure owned by a peer and can never be declared peerless.
+func (ri *StorageResourceInstance) IsPeerless() bool { return false }
+
 func NewStorageResourceInstance(name string, peerID string) ResourceInstanceITF {
 	return &StorageResourceInstance{
 		ResourceInstance: ResourceInstance[*StorageResourcePartnership]{
@@ -68,30 +81,6 @@ func NewStorageResourceInstance(name string, peerID string) ResourceInstanceITF
 	}
 }
 
-func (ri *StorageResourceInstance) ClearEnv() {
-	ri.Env = []models.Param{}
-	ri.Inputs = []models.Param{}
-	ri.Outputs = []models.Param{}
-}
-
-func (ri *StorageResourceInstance) StoreDraftDefault() {
-	found := false
-	for _, p := range ri.ResourceInstance.Env {
-		if p.Attr == "source" {
-			found = true
-			break
-		}
-	}
-	if !found {
-		ri.ResourceInstance.Env = append(ri.ResourceInstance.Env, models.Param{
-			Attr:     "source",
-			Value:    ri.Source,
-			Readonly: true,
-		})
-	}
-	ri.ResourceInstance.StoreDraftDefault()
-}
-
 type StorageResourcePartnership struct {
 	ResourcePartnerShip[*StorageResourcePricingProfile]
 	MaxSizeGBAllowed     float64 `json:"allowed_gb,omitempty" bson:"allowed_gb,omitempty"`
@@ -117,7 +106,7 @@ func (t PrivilegeStoragePricingStrategy) String() string {
 type StorageResourcePricingStrategy int
 
 const (
-	PER_DATA_STORED StorageResourcePricingStrategy = iota + 6
+	PER_DATA_STORED StorageResourcePricingStrategy = iota + 7
 	PER_TB_STORED
 	PER_GB_STORED
 	PER_MB_STORED

models/resources/tests/data_test.go

@@ -4,7 +4,6 @@ import (
 	"testing"
 	"time"
 
-	"cloud.o-forge.io/core/oc-lib/models/common/models"
 	"cloud.o-forge.io/core/oc-lib/models/common/pricing"
 	"cloud.o-forge.io/core/oc-lib/models/resources"
 	"cloud.o-forge.io/core/oc-lib/tools"
@@ -34,23 +33,6 @@ func TestDataResource_ConvertToPricedResource(t *testing.T) {
 	assert.Nil(t, nilRes)
 }
 
-func TestDataInstance_StoreDraftDefault(t *testing.T) {
-	di := &resources.DataInstance{
-		Source: "test-src",
-		ResourceInstance: resources.ResourceInstance[*resources.DataResourcePartnership]{
-			Env: []models.Param{},
-		},
-	}
-	di.StoreDraftDefault()
-	assert.Len(t, di.ResourceInstance.Env, 1)
-	assert.Equal(t, "source", di.ResourceInstance.Env[0].Attr)
-	assert.Equal(t, "test-src", di.ResourceInstance.Env[0].Value)
-
-	// Call again, should not duplicate
-	di.StoreDraftDefault()
-	assert.Len(t, di.ResourceInstance.Env, 1)
-}
-
 func TestDataResourcePricingStrategy_GetQuantity(t *testing.T) {
 	tests := []struct {
 		strategy resources.DataResourcePricingStrategy

models/resources/tests/resource_test.go

@@ -114,8 +114,6 @@ func (f *FakeResource) ConvertToPricedResource(t tools.DataType, a *int, b *int,
 func (f *FakeResource) VerifyAuth(string, *tools.APIRequest) bool { return true }
 
 func TestNewAccessor_ReturnsValid(t *testing.T) {
-	acc := resources.NewAccessor[*FakeResource](tools.COMPUTE_RESOURCE, &tools.APIRequest{}, func() utils.DBObject {
-		return &FakeResource{}
-	})
+	acc := resources.NewAccessor[*FakeResource](tools.COMPUTE_RESOURCE, &tools.APIRequest{})
 	assert.NotNil(t, acc)
 }

models/resources/tests/storage_test.go

@@ -3,7 +3,6 @@ package resources_test
 import (
 	"testing"
 
-	"cloud.o-forge.io/core/oc-lib/models/common/models"
 	"cloud.o-forge.io/core/oc-lib/tools"
 	"github.com/stretchr/testify/assert"
 
@@ -37,36 +36,6 @@ func TestStorageResource_ConvertToPricedResource_InvalidType(t *testing.T) {
 	assert.Nil(t, priced)
 }
 
-func TestStorageResourceInstance_ClearEnv(t *testing.T) {
-	inst := &resources.StorageResourceInstance{
-		ResourceInstance: resources.ResourceInstance[*resources.StorageResourcePartnership]{
-			Env:     []models.Param{{Attr: "A"}},
-			Inputs:  []models.Param{{Attr: "B"}},
-			Outputs: []models.Param{{Attr: "C"}},
-		},
-	}
-
-	inst.ClearEnv()
-	assert.Empty(t, inst.Env)
-	assert.Empty(t, inst.Inputs)
-	assert.Empty(t, inst.Outputs)
-}
-
-func TestStorageResourceInstance_StoreDraftDefault(t *testing.T) {
-	inst := &resources.StorageResourceInstance{
-		Source: "my-source",
-		ResourceInstance: resources.ResourceInstance[*resources.StorageResourcePartnership]{
-			Env: []models.Param{},
-		},
-	}
-
-	inst.StoreDraftDefault()
-	assert.Len(t, inst.Env, 1)
-	assert.Equal(t, "source", inst.Env[0].Attr)
-	assert.Equal(t, "my-source", inst.Env[0].Value)
-	assert.True(t, inst.Env[0].Readonly)
-}
-
 func TestStorageResourcePricingStrategy_GetQuantity(t *testing.T) {
 	tests := []struct {
 		strategy resources.StorageResourcePricingStrategy

models/resources/workflow.go

@@ -16,7 +16,7 @@ type WorkflowResource struct {
 }
 
 func (d *WorkflowResource) GetAccessor(request *tools.APIRequest) utils.Accessor {
-	return NewAccessor[*WorkflowResource](tools.WORKFLOW_RESOURCE, request, func() utils.DBObject { return &WorkflowResource{} })
+	return NewAccessor[*WorkflowResource](tools.WORKFLOW_RESOURCE, request)
 }
 
 func (r *WorkflowResource) AddInstances(instance ResourceInstanceITF) {

models/utils/abstracts.go

@@ -31,6 +31,7 @@ const (
  */
 type AbstractObject struct {
 	UUID          string     `json:"id,omitempty" bson:"id,omitempty" validate:"required"`
+	NotInCatalog  bool       `json:"not_in_catalog" bson:"not_in_catalog" default:"false"`
 	Name          string     `json:"name,omitempty" bson:"name,omitempty" validate:"required"`
 	IsDraft       bool       `json:"is_draft" bson:"is_draft" default:"false"`
 	CreatorID     string     `json:"creator_id,omitempty" bson:"creator_id,omitempty"`
@@ -46,6 +47,12 @@ type AbstractObject struct {
 func (ri *AbstractObject) GetAccessor(request *tools.APIRequest) Accessor {
 	return nil
 }
+func (r *AbstractObject) SetNotInCatalog(ok bool) {
+	r.NotInCatalog = ok
+}
+func (r *AbstractObject) IsNotInCatalog() bool {
+	return r.NotInCatalog
+}
 
 func (r *AbstractObject) Unsign() {
 	r.Signature = nil
@@ -259,12 +266,12 @@ func (a *AbstractAccessor[T]) LoadOne(id string) (DBObject, int, error) {
 	}, a)
 }
 
-func (a *AbstractAccessor[T]) LoadAll(isDraft bool) ([]ShallowDBObject, int, error) {
-	return GenericLoadAll[T](a.GetExec(isDraft), isDraft, a)
+func (a *AbstractAccessor[T]) LoadAll(isDraft bool, offset int64, limit int64) ([]ShallowDBObject, int, error) {
+	return GenericLoadAll[T](a.GetExec(isDraft), isDraft, a, offset, limit)
 }
 
-func (a *AbstractAccessor[T]) Search(filters *dbs.Filters, search string, isDraft bool) ([]ShallowDBObject, int, error) {
-	return GenericSearch[T](filters, search, a.New().GetObjectFilters(search), a.GetExec(isDraft), isDraft, a)
+func (a *AbstractAccessor[T]) Search(filters *dbs.Filters, search string, isDraft bool, offset int64, limit int64) ([]ShallowDBObject, int, error) {
+	return GenericSearch[T](filters, search, a.New().GetObjectFilters(search), a.GetExec(isDraft), isDraft, a, offset, limit)
 }
 
 func (a *AbstractAccessor[T]) GetExec(isDraft bool) func(DBObject) ShallowDBObject {

models/utils/change_bus.go

@@ -0,0 +1,60 @@
+package utils
+
+import (
+	"sync"
+
+	"cloud.o-forge.io/core/oc-lib/tools"
+)
+
+// ChangeEvent is fired whenever a DB object is created, updated or deleted
+// within this process. Deleted=true means the object was removed; Object is
+// the last known snapshot before deletion.
+type ChangeEvent struct {
+	DataType tools.DataType
+	ID       string
+	Object   ShallowDBObject // nil only when the load after the write failed
+	Deleted  bool
+}
+
+var (
+	changeBusMu sync.RWMutex
+	changeBus   = map[tools.DataType][]chan ChangeEvent{}
+)
+
+// SubscribeChanges returns a channel that receives ChangeEvents for dt
+// whenever an object of that type is written or deleted in this process.
+// Call the returned cancel function to unsubscribe; after that the channel
+// will no longer receive events (it is not closed — use a context to stop
+// reading).
+func SubscribeChanges(dt tools.DataType) (<-chan ChangeEvent, func()) {
+	ch := make(chan ChangeEvent, 32)
+	changeBusMu.Lock()
+	changeBus[dt] = append(changeBus[dt], ch)
+	changeBusMu.Unlock()
+	return ch, func() {
+		changeBusMu.Lock()
+		subs := changeBus[dt]
+		for i, c := range subs {
+			if c == ch {
+				changeBus[dt] = append(subs[:i], subs[i+1:]...)
+				break
+			}
+		}
+		changeBusMu.Unlock()
+	}
+}
+
+// NotifyChange broadcasts a ChangeEvent to all current subscribers for dt.
+// Non-blocking: events are dropped for subscribers whose buffer is full.
+func NotifyChange(dt tools.DataType, id string, obj ShallowDBObject, deleted bool) {
+	changeBusMu.RLock()
+	subs := changeBus[dt]
+	changeBusMu.RUnlock()
+	evt := ChangeEvent{DataType: dt, ID: id, Object: obj, Deleted: deleted}
+	for _, ch := range subs {
+		select {
+		case ch <- evt:
+		default:
+		}
+	}
+}

models/utils/common.go

@@ -51,7 +51,7 @@ func GenericStoreOne(data DBObject, a Accessor) (DBObject, int, error) {
 	if a.ShouldVerifyAuth() && !data.VerifyAuth("store", a.GetRequest()) {
 		return nil, 403, errors.New("you are not allowed to access : " + a.GetType().String())
 	}
-	if cursor, _, _ := a.Search(&f, "", data.IsDrafted()); len(cursor) > 0 {
+	if cursor, _, _ := a.Search(&f, "", data.IsDrafted(), 0, 10); len(cursor) > 0 {
 		return nil, 409, errors.New(a.GetType().String() + " with name " + data.GetName() + " already exists")
 	}
 	err := validate.Struct(data)
@@ -63,7 +63,11 @@ func GenericStoreOne(data DBObject, a Accessor) (DBObject, int, error) {
 		a.GetLogger().Error().Msg("Could not store " + data.GetName() + " to db. Error: " + err.Error())
 		return nil, code, err
 	}
-	return a.LoadOne(id)
+	result, rcode, rerr := a.LoadOne(id)
+	if rerr == nil && result != nil {
+		go NotifyChange(a.GetType(), result.GetID(), result, false)
+	}
+	return result, rcode, rerr
 }
 
 // GenericLoadOne loads one object from the database (generic)
@@ -78,9 +82,6 @@ func GenericDeleteOne(id string, a Accessor) (DBObject, int, error) {
 	if !res.CanDelete() {
 		return nil, 403, errors.New("you are not allowed to delete :" + a.GetType().String())
 	}
-	if err != nil {
-		return nil, code, err
-	}
 	if a.ShouldVerifyAuth() && !res.VerifyAuth("delete", a.GetRequest()) {
 		return nil, 403, errors.New("you are not allowed to access " + a.GetType().String())
 	}
@@ -89,6 +90,7 @@ func GenericDeleteOne(id string, a Accessor) (DBObject, int, error) {
 		a.GetLogger().Error().Msg("Could not delete " + id + " to db. Error: " + err.Error())
 		return nil, code, err
 	}
+	go NotifyChange(a.GetType(), id, res, true)
 	return res, 200, nil
 }
 
@@ -117,17 +119,26 @@ func ModelGenericUpdateOne(change map[string]interface{}, id string, a Accessor)
 	}
 
 	loaded := r.Serialize(r)   // get the loaded object
-
 	for k, v := range change { // apply the changes, with a flatten method
 		loaded[k] = v
 	}
-	return r, loaded, 200, nil
+	newObj := a.NewObj()
+	b, err = json.Marshal(loaded)
+	if err != nil {
+		return nil, loaded, 400, nil
+	}
+	err = json.Unmarshal(b, newObj)
+	if err != nil {
+		return nil, loaded, 400, nil
+	}
+	return newObj, loaded, 200, nil
 }
 
 // GenericLoadOne loads one object from the database (generic)
 // json expected in entry is a flatted object no need to respect the inheritance hierarchy
 func GenericUpdateOne(change map[string]interface{}, id string, a Accessor) (DBObject, int, error) {
 	obj, loaded, c, err := ModelGenericUpdateOne(change, id, a)
+
 	if err != nil {
 		return nil, c, err
 	}
@@ -136,7 +147,11 @@ func GenericUpdateOne(change map[string]interface{}, id string, a Accessor) (DBO
 		a.GetLogger().Error().Msg("Could not update " + id + " to db. Error: " + err.Error())
 		return nil, code, err
 	}
-	return a.LoadOne(id)
+	result, rcode, rerr := a.LoadOne(id)
+	if rerr == nil && result != nil {
+		go NotifyChange(a.GetType(), result.GetID(), result, false)
+	}
+	return result, rcode, rerr
 }
 
 func GenericLoadOne[T DBObject](id string, data T, f func(DBObject) (DBObject, int, error), a Accessor) (DBObject, int, error) {
@@ -172,17 +187,27 @@ func genericLoadAll[T DBObject](res *mgb.Cursor, code int, err error, onlyDraft
 	return objs, 200, nil
 }
 
-func GenericLoadAll[T DBObject](f func(DBObject) ShallowDBObject, onlyDraft bool, wfa Accessor) ([]ShallowDBObject, int, error) {
-	res_mongo, code, err := mongo.MONGOService.LoadAll(wfa.GetType().String())
+func GenericLoadAll[T DBObject](f func(DBObject) ShallowDBObject, onlyDraft bool, wfa Accessor, opts ...int64) ([]ShallowDBObject, int, error) {
+	offset := int64(0)
+	limit := int64(0)
+	if len(opts) > 1 {
+		offset = opts[0]
+	}
+	res_mongo, code, err := mongo.MONGOService.LoadAll(wfa.GetType().String(), offset, limit)
 	return genericLoadAll[T](res_mongo, code, err, onlyDraft, f, wfa)
 }
 
 func GenericSearch[T DBObject](filters *dbs.Filters, search string, defaultFilters *dbs.Filters,
-	f func(DBObject) ShallowDBObject, onlyDraft bool, wfa Accessor) ([]ShallowDBObject, int, error) {
+	f func(DBObject) ShallowDBObject, onlyDraft bool, wfa Accessor, opts ...int64) ([]ShallowDBObject, int, error) {
 	if filters == nil && search != "" {
 		filters = defaultFilters
 	}
-	res_mongo, code, err := mongo.MONGOService.Search(filters, wfa.GetType().String())
+	offset := int64(0)
+	limit := int64(0)
+	if len(opts) > 1 {
+		offset = opts[0]
+	}
+	res_mongo, code, err := mongo.MONGOService.Search(filters, wfa.GetType().String(), offset, limit)
 	return genericLoadAll[T](res_mongo, code, err, onlyDraft, f, wfa)
 }
 
@@ -194,7 +219,11 @@ func GenericRawUpdateOne(set DBObject, id string, a Accessor) (DBObject, int, er
 		a.GetLogger().Error().Msg("Could not update " + id + " to db. Error: " + err.Error())
 		return nil, code, err
 	}
-	return a.LoadOne(id)
+	result, rcode, rerr := a.LoadOne(id)
+	if rerr == nil && result != nil {
+		go NotifyChange(a.GetType(), result.GetID(), result, false)
+	}
+	return result, rcode, rerr
 }
 
 func GetMySelf(wfa Accessor) (ShallowDBObject, error) {
@@ -202,7 +231,7 @@ func GetMySelf(wfa Accessor) (ShallowDBObject, error) {
 		And: map[string][]dbs.Filter{
 			"relation": {{Operator: dbs.EQUAL.String(), Value: 1}},
 		},
-	}, "", false)
+	}, "", false, 0, 1)
 	if len(datas) > 0 && datas[0] != nil {
 		return datas[0], nil
 	}

models/utils/interfaces.go

@@ -18,6 +18,8 @@ type ShallowDBObject interface {
 // DBObject is an interface that defines the basic methods for a DBObject
 type DBObject interface {
 	GenerateID()
+	SetNotInCatalog(bool)
+	IsNotInCatalog() bool
 	SetID(id string)
 	GetID() string
 	GetName() string
@@ -53,8 +55,8 @@ type Accessor interface {
 	DeleteOne(id string) (DBObject, int, error)
 	CopyOne(data DBObject) (DBObject, int, error)
 	StoreOne(data DBObject) (DBObject, int, error)
-	LoadAll(isDraft bool) ([]ShallowDBObject, int, error)
+	LoadAll(isDraft bool, offset int64, limit int64) ([]ShallowDBObject, int, error)
 	UpdateOne(set map[string]interface{}, id string) (DBObject, int, error)
-	Search(filters *dbs.Filters, search string, isDraft bool) ([]ShallowDBObject, int, error)
+	Search(filters *dbs.Filters, search string, isDraft bool, offset int64, limit int64) ([]ShallowDBObject, int, error)
 	GetExec(isDraft bool) func(DBObject) ShallowDBObject
 }

models/workflow/graph/item.go

@@ -25,6 +25,8 @@ func (g *GraphItem) GetResource() (tools.DataType, resources.ResourceInterface)
 		return tools.PROCESSING_RESOURCE, g.Processing
 	} else if g.Storage != nil {
 		return tools.STORAGE_RESOURCE, g.Storage
+	} else if g.NativeTool != nil {
+		return tools.NATIVE_TOOL, g.NativeTool
 	}
 	return tools.INVALID, nil
 }

models/workflow/workflow.go

@@ -44,11 +44,14 @@ type Workflow struct {
 	utils.AbstractObject // AbstractObject contains the basic fields of an object (id, name)
 	resources.ResourceSet
 	Graph *graph.Graph `bson:"graph,omitempty" json:"graph,omitempty"` // Graph UI & logic representation of the workflow
-	ScheduleActive bool         `json:"schedule_active" bson:"schedule_active"` // ScheduleActive is a flag that indicates if the schedule is active, if not the workflow is not scheduled and no execution or booking will be set
 	// Schedule       *WorkflowSchedule `bson:"schedule,omitempty" json:"schedule,omitempty"` // Schedule is the schedule of the workflow
 	Shared []string `json:"shared,omitempty" bson:"shared,omitempty"` // Shared is the ID of the shared workflow     // AbstractWorkflow contains the basic fields of a workflow
-	Env    []models.Param `json:"env,omitempty" bson:"env,omitempty"`
-	Inputs []models.Param `json:"inputs,omitempty" bson:"inputs,omitempty"`
+
+	Env     map[string][]models.Param  `json:"env" bson:"env"`
+	Inputs  map[string][]models.Param  `json:"inputs" bson:"inputs"`
+	Outputs map[string][]models.Param  `json:"outputs" bson:"outputs"`
+	Args    map[string][]string        `json:"args" bson:"args"`
+	Exposes map[string][]models.Expose `bson:"exposes" json:"exposes"` // Expose is the execution
 }
 
 func (d *Workflow) GetAccessor(request *tools.APIRequest) utils.Accessor {

models/workflow/workflow_mongo_accessor.go

@@ -150,7 +150,7 @@ func (a *workflowMongoAccessor) execute(workflow *Workflow, delete bool, active
 			"abstractobject.name": {{Operator: dbs.LIKE.String(), Value: workflow.Name + "_workspace"}},
 		},
 	}
-	resource, _, err := a.workspaceAccessor.Search(filters, "", workflow.IsDraft)
+	resource, _, err := a.workspaceAccessor.Search(filters, "", workflow.IsDraft, 0, 10)
 	if delete { // if delete is set to true, delete the workspace
 		for _, r := range resource {
 			a.workspaceAccessor.DeleteOne(r.GetID())
@@ -192,9 +192,9 @@ func (a *workflowMongoAccessor) LoadOne(id string) (utils.DBObject, int, error)
 	}, a)
 }
 
-func (a *workflowMongoAccessor) Search(filters *dbs.Filters, search string, isDraft bool) ([]utils.ShallowDBObject, int, error) {
+func (a *workflowMongoAccessor) Search(filters *dbs.Filters, search string, isDraft bool, offset int64, limit int64) ([]utils.ShallowDBObject, int, error) {
 	return utils.GenericSearch[*Workflow](filters, search, a.New().GetObjectFilters(search),
-		func(d utils.DBObject) utils.ShallowDBObject { return a.verifyResource(d) }, isDraft, a)
+		func(d utils.DBObject) utils.ShallowDBObject { return a.verifyResource(d) }, isDraft, a, offset, limit)
 }
 
 func (a *workflowMongoAccessor) verifyResource(obj utils.DBObject) utils.DBObject {
@@ -210,15 +210,15 @@ func (a *workflowMongoAccessor) verifyResource(obj utils.DBObject) utils.DBObjec
 		var access utils.Accessor
 		switch t {
 		case tools.COMPUTE_RESOURCE:
-			access = resources.NewAccessor[*resources.ComputeResource](t, a.GetRequest(), func() utils.DBObject { return &resources.ComputeResource{} })
+			access = resources.NewAccessor[*resources.ComputeResource](t, a.GetRequest())
 		case tools.PROCESSING_RESOURCE:
-			access = resources.NewAccessor[*resources.ProcessingResource](t, a.GetRequest(), func() utils.DBObject { return &resources.ProcessingResource{} })
+			access = resources.NewAccessor[*resources.ProcessingResource](t, a.GetRequest())
 		case tools.STORAGE_RESOURCE:
-			access = resources.NewAccessor[*resources.StorageResource](t, a.GetRequest(), func() utils.DBObject { return &resources.StorageResource{} })
+			access = resources.NewAccessor[*resources.StorageResource](t, a.GetRequest())
 		case tools.WORKFLOW_RESOURCE:
-			access = resources.NewAccessor[*resources.WorkflowResource](t, a.GetRequest(), func() utils.DBObject { return &resources.WorkflowResource{} })
+			access = resources.NewAccessor[*resources.WorkflowResource](t, a.GetRequest())
 		case tools.DATA_RESOURCE:
-			access = resources.NewAccessor[*resources.DataResource](t, a.GetRequest(), func() utils.DBObject { return &resources.DataResource{} })
+			access = resources.NewAccessor[*resources.DataResource](t, a.GetRequest())
 		default:
 			wf.Graph.Clear(resource.GetID())
 		}

models/workflow_execution/workflow_execution.go

@@ -17,6 +17,17 @@ import (
 	"go.mongodb.org/mongo-driver/bson/primitive"
 )
 
+// BookingState tracks the reservation and completion status of a single booking
+// within a workflow execution.
+//   - IsBooked: true while the resource is actively reserved (set on WORKFLOW_STARTED_EVENT,
+//     cleared on WORKFLOW_STEP_DONE_EVENT / WORKFLOW_DONE_EVENT).
+//   - IsDone: true once the booking has been confirmed by the remote peer (CONSIDERS_EVENT)
+//     or completed (WORKFLOW_STEP_DONE_EVENT / WORKFLOW_DONE_EVENT).
+type BookingState struct {
+	IsBooked bool `json:"is_booked" bson:"is_booked"`
+	IsDone   bool `json:"is_done" bson:"is_done"`
+}
+
 /*
 * WorkflowExecution is a struct that represents a list of workflow executions
 * Warning: No user can write (del, post, put) a workflow execution, it is only used by the system
@@ -33,8 +44,8 @@ type WorkflowExecution struct {
 	State                enum.BookingStatus             `json:"state" bson:"state" default:"0"`                                               // TEMPORARY TODO DEFAULT 1 -> 0 State is the state of the workflow
 	WorkflowID           string                         `json:"workflow_id" bson:"workflow_id,omitempty"`                                     // WorkflowID is the ID of the workflow
 
-	BookingsState  map[string]bool `json:"bookings_state" bson:"bookings_state,omitempty"`   // WorkflowID is the ID of the workflow
-	PurchasesState map[string]bool `json:"purchases_state" bson:"purchases_state,omitempty"` // WorkflowID is the ID of the workflow
+	BookingsState  map[string]BookingState `json:"bookings_state" bson:"bookings_state,omitempty"`   // booking_id → reservation+completion status
+	PurchasesState map[string]bool         `json:"purchases_state" bson:"purchases_state,omitempty"` // purchase_id → confirmed
 
 	SelectedInstances    workflow.ConfigItem `json:"selected_instances"`
 	SelectedPartnerships workflow.ConfigItem `json:"selected_partnerships"`
@@ -78,7 +89,7 @@ func (ws *WorkflowExecution) PurgeDraft(request *tools.APIRequest) error {
 				{Operator: dbs.GTE.String(), Value: primitive.NewDateTimeFromTime(ws.ExecDate)},
 			},
 		},
-	}, "", ws.IsDraft)
+	}, "", ws.IsDraft, 0, 10000)
 	if code != 200 || err != nil {
 		return err
 	}
@@ -189,9 +200,9 @@ func (d *WorkflowExecution) Book(executionsID string, wfID string, priceds map[t
 	booking = append(booking, d.bookEach(executionsID, wfID, tools.DATA_RESOURCE, priceds[tools.DATA_RESOURCE])...)
 	for _, p := range booking {
 		if d.BookingsState == nil {
-			d.BookingsState = map[string]bool{}
+			d.BookingsState = map[string]BookingState{}
 		}
-		d.BookingsState[p.GetID()] = false
+		d.BookingsState[p.GetID()] = BookingState{}
 	}
 	return booking
 }
@@ -229,10 +240,14 @@ func (d *WorkflowExecution) bookEach(executionsID string, wfID string, dt tools.
 		var m map[string]interface{}
 		b, _ := json.Marshal(priced)
 		json.Unmarshal(b, &m)
+		name := priced.GetName()
+		if len(executionsID) > 8 {
+			name += " " + executionsID[:8]
+		}
 		bookingItem := &booking.Booking{
 			AbstractObject: utils.AbstractObject{
 				UUID:    uuid.New().String(),
-				Name:    d.GetName() + "_" + executionsID + "_" + wfID,
+				Name:    name + " " + start.Format("2006-01-02 15:04"),
 				IsDraft: true,
 			},
 			PricedItem:        m,
@@ -242,6 +257,7 @@ func (d *WorkflowExecution) bookEach(executionsID string, wfID string, dt tools.
 			InstanceID:        priced.GetInstanceID(),
 			ResourceType:      dt,
 			DestPeerID:        priced.GetCreatorID(),
+			Peerless:          priced.GetCreatorID() == "",
 			WorkflowID:        wfID,
 			ExecutionID:       d.GetID(),
 			ExpectedStartDate: start,

models/workspace/workspace_mongo_accessor.go

@@ -49,7 +49,7 @@ func (a *workspaceMongoAccessor) DeleteOne(id string) (utils.DBObject, int, erro
 
 // UpdateOne updates a workspace in the database, given its ID, it automatically share to peers if the workspace is shared
 func (a *workspaceMongoAccessor) UpdateOne(set map[string]interface{}, id string) (utils.DBObject, int, error) {
-	if set["active"] == true { // If the workspace is active, deactivate all the other workspaces
+	/*if set["active"] == true { // If the workspace is active, deactivate all the other workspaces
 		res, _, err := a.LoadAll(true)
 		if err == nil {
 			for _, r := range res {
@@ -59,7 +59,7 @@ func (a *workspaceMongoAccessor) UpdateOne(set map[string]interface{}, id string
 				}
 			}
 		}
-	}
+	}*/
 	res, code, err := utils.GenericUpdateOne(set, id, a)
 	if code == 200 && res != nil {
 		a.share(res.(*Workspace), tools.PUT, a.GetCaller())
@@ -76,7 +76,7 @@ func (a *workspaceMongoAccessor) StoreOne(data utils.DBObject) (utils.DBObject,
 		},
 	}
 	// filters *dbs.Filters, word string, isDraft bool
-	res, _, err := a.Search(filters, "", true) // Search for the workspace
+	res, _, err := a.Search(filters, "", true, 0, 10) // Search for the workspace
 	if err == nil && len(res) > 0 {                   // If the workspace already exists, return an error
 		return nil, 409, errors.New("a workspace with the same name already exists")
 	}
@@ -87,24 +87,24 @@ func (a *workspaceMongoAccessor) StoreOne(data utils.DBObject) (utils.DBObject,
 }
 
 func (a *workspaceMongoAccessor) LoadOne(id string) (utils.DBObject, int, error) {
-	return utils.GenericLoadOne[*Workspace](id, a.New(), func(d utils.DBObject) (utils.DBObject, int, error) {
+	return utils.GenericLoadOne(id, a.New(), func(d utils.DBObject) (utils.DBObject, int, error) {
 		d.(*Workspace).Fill(a.GetRequest())
 		return d, 200, nil
 	}, a)
 }
 
-func (a *workspaceMongoAccessor) LoadAll(isDraft bool) ([]utils.ShallowDBObject, int, error) {
+func (a *workspaceMongoAccessor) LoadAll(isDraft bool, offset int64, limit int64) ([]utils.ShallowDBObject, int, error) {
 	return utils.GenericLoadAll[*Workspace](func(d utils.DBObject) utils.ShallowDBObject {
 		d.(*Workspace).Fill(a.GetRequest())
 		return d
-	}, isDraft, a)
+	}, isDraft, a, offset, limit)
 }
 
-func (a *workspaceMongoAccessor) Search(filters *dbs.Filters, search string, isDraft bool) ([]utils.ShallowDBObject, int, error) {
+func (a *workspaceMongoAccessor) Search(filters *dbs.Filters, search string, isDraft bool, offset int64, limit int64) ([]utils.ShallowDBObject, int, error) {
 	return utils.GenericSearch[*Workspace](filters, search, (&Workspace{}).GetObjectFilters(search), func(d utils.DBObject) utils.ShallowDBObject {
 		d.(*Workspace).Fill(a.GetRequest())
 		return d
-	}, isDraft, a)
+	}, isDraft, a, offset, limit)
 }
 
 /*

tools/api.go

@@ -60,16 +60,16 @@ func (s State) String() string {
 
 type API struct{}
 
-func (a *API) Discovered(infos []*beego.ControllerInfo) {
+func (a *API) Discovered(infos []*beego.ControllerInfo, extra ...map[string][]string) {
 	respondToDiscovery := func(resp NATSResponse) {
 		var m map[string]interface{}
 		json.Unmarshal(resp.Payload, &m)
 		if len(m) == 0 {
-			a.SubscribeRouter(infos)
+			a.SubscribeRouter(infos, extra...)
 		}
 	}
 	a.ListenRouter(respondToDiscovery)
-	a.SubscribeRouter(infos)
+	a.SubscribeRouter(infos, extra...)
 }
 
 // GetState returns the state of the API
@@ -99,11 +99,12 @@ func (a *API) ListenRouter(exec func(msg NATSResponse)) {
 	})
 }
 
-func (a *API) SubscribeRouter(infos []*beego.ControllerInfo) {
+func (a *API) SubscribeRouter(infos []*beego.ControllerInfo, extra ...map[string][]string) {
 	nats := NewNATSCaller()
+	appPrefix := "/" + strings.ReplaceAll(config.GetAppName(), "oc-", "")
 	discovery := map[string][]string{}
 	for _, info := range infos {
-		path := strings.ReplaceAll(info.GetPattern(), "/oc/", "/"+strings.ReplaceAll(config.GetAppName(), "oc-", ""))
+		path := strings.ReplaceAll(info.GetPattern(), "/oc/", appPrefix+"/")
 		for k, v := range info.GetMethod() {
 			if discovery[path] == nil {
 				discovery[path] = []string{}
@@ -115,6 +116,15 @@ func (a *API) SubscribeRouter(infos []*beego.ControllerInfo) {
 			}
 		}
 	}
+	for _, extraRoutes := range extra {
+		for rawPath, methods := range extraRoutes {
+			path := strings.ReplaceAll(rawPath, "/oc/", appPrefix+"/")
+			if discovery[path] == nil {
+				discovery[path] = []string{}
+			}
+			discovery[path] = append(discovery[path], methods...)
+		}
+	}
 	b, _ := json.Marshal(discovery)
 
 	go nats.SetNATSPub(DISCOVERY, NATSResponse{

tools/enums.go

@@ -32,6 +32,7 @@ const (
 	BILL
 	NATIVE_TOOL
 	EXECUTION_VERIFICATION
+	ALLOWED_IMAGE
 )
 
 var NOAPI = func() string {
@@ -88,6 +89,7 @@ var InnerDefaultAPI = [...]func() string{
 	NOAPI,
 	CATALOGAPI,
 	SCHEDULERAPI,
+	DATACENTERAPI,
 }
 
 // Bind the standard data name to the data type
@@ -114,6 +116,7 @@ var Str = [...]string{
 	"bill",
 	"native_tool",
 	"execution_verification",
+	"allowed_image",
 }
 
 func FromString(comp string) int {
@@ -149,7 +152,7 @@ func DataTypeList() []DataType {
 	return []DataType{DATA_RESOURCE, PROCESSING_RESOURCE, STORAGE_RESOURCE, COMPUTE_RESOURCE, WORKFLOW_RESOURCE,
 		WORKFLOW, WORKFLOW_EXECUTION, WORKSPACE, PEER, COLLABORATIVE_AREA, RULE, BOOKING, WORKFLOW_HISTORY, WORKSPACE_HISTORY,
 		ORDER, PURCHASE_RESOURCE,
-		LIVE_DATACENTER, LIVE_STORAGE, BILL, NATIVE_TOOL}
+		LIVE_DATACENTER, LIVE_STORAGE, BILL, NATIVE_TOOL, EXECUTION_VERIFICATION, ALLOWED_IMAGE}
 }
 
 type PropalgationMessage struct {
@@ -171,6 +174,7 @@ const (
 	PB_CONSIDERS
 	PB_ADMIRALTY_CONFIG
 	PB_MINIO_CONFIG
+	PB_PVC_CONFIG
 	PB_CLOSE_SEARCH
 	NONE
 )

tools/kubernetes.go

@@ -14,6 +14,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -97,8 +98,8 @@ func (k *KubernetesService) CreateNamespace(ctx context.Context, ns string) erro
 	namespace := &v1.Namespace{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: ns,
-			Labels: map[string]string{
-				"multicluster-scheduler": "enabled",
+			Annotations: map[string]string{
+				"multicluster.admiralty.io/elect": "",
 			},
 		},
 	}
@@ -199,9 +200,9 @@ func (k *KubernetesService) ProvisionExecutionNamespace(ctx context.Context, ns
 	}
 	role := "argo-role"
 	if err := k.CreateRole(ctx, ns, role,
-		[][]string{{"coordination.k8s.io"}, {""}, {""}},
-		[][]string{{"leases"}, {"secrets"}, {"pods"}},
-		[][]string{{"get", "create", "update"}, {"get"}, {"patch"}},
+		[][]string{{"coordination.k8s.io"}, {""}, {""}, {"multicluster.admiralty.io"}, {"argoproj.io"}},
+		[][]string{{"leases"}, {"secrets"}, {"pods"}, {"podchaperons"}, {"workflowtaskresults"}},
+		[][]string{{"get", "create", "update"}, {"get"}, {"patch"}, {"get", "list", "watch", "create", "update", "patch", "delete"}, {"create", "patch"}},
 	); err != nil {
 		return err
 	}
@@ -598,6 +599,75 @@ func (k *KubernetesService) CreateSecret(context context.Context, minioId string
 	return nil
 }
 
+// CreatePVC creates a static PersistentVolume + PersistentVolumeClaim in the given namespace.
+// Static provisioning (no StorageClass) avoids the WaitForFirstConsumer deadlock
+// with Admiralty virtual nodes — the PVC binds immediately.
+func (k *KubernetesService) CreatePVC(ctx context.Context, name, namespace, storageSize string) error {
+	storageClassName := ""
+	pv := &v1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: v1.PersistentVolumeSpec{
+			Capacity: v1.ResourceList{
+				v1.ResourceStorage: resource.MustParse(storageSize),
+			},
+			AccessModes:                   []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
+			StorageClassName:              storageClassName,
+			PersistentVolumeReclaimPolicy: v1.PersistentVolumeReclaimDelete,
+			PersistentVolumeSource: v1.PersistentVolumeSource{
+				HostPath: &v1.HostPathVolumeSource{
+					Path: "/var/lib/oc-storage/" + name,
+					Type: func() *v1.HostPathType { t := v1.HostPathDirectoryOrCreate; return &t }(),
+				},
+			},
+			ClaimRef: &v1.ObjectReference{
+				Namespace: namespace,
+				Name:      name,
+			},
+		},
+	}
+	_, err := k.Set.CoreV1().PersistentVolumes().Create(ctx, pv, metav1.CreateOptions{})
+	if err != nil && !apierrors.IsAlreadyExists(err) {
+		return fmt.Errorf("CreatePV %s: %w", name, err)
+	}
+
+	pvc := &v1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Spec: v1.PersistentVolumeClaimSpec{
+			AccessModes:      []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
+			StorageClassName: &storageClassName,
+			VolumeName:       name,
+			Resources: v1.VolumeResourceRequirements{
+				Requests: v1.ResourceList{
+					v1.ResourceStorage: resource.MustParse(storageSize),
+				},
+			},
+		},
+	}
+	_, err = k.Set.CoreV1().PersistentVolumeClaims(namespace).Create(ctx, pvc, metav1.CreateOptions{})
+	if err != nil && !apierrors.IsAlreadyExists(err) {
+		return fmt.Errorf("CreatePVC %s/%s: %w", namespace, name, err)
+	}
+	return nil
+}
+
+// DeletePVC deletes a PersistentVolumeClaim and its associated PersistentVolume.
+func (k *KubernetesService) DeletePVC(ctx context.Context, name, namespace string) error {
+	err := k.Set.CoreV1().PersistentVolumeClaims(namespace).Delete(ctx, name, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("DeletePVC %s/%s: %w", namespace, name, err)
+	}
+	err = k.Set.CoreV1().PersistentVolumes().Delete(ctx, name, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("DeletePV %s: %w", name, err)
+	}
+	return nil
+}
+
 // ============== ADMIRALTY ==============
 // Returns a concatenation of the peerId and namespace in order for
 // kubernetes ressources to have a unique name, under 63 characters

tools/nats_caller.go

@@ -29,8 +29,9 @@ type NATSMethod int
 var meths = []string{"remove execution", "create execution", "planner execution", "discovery",
 	"workflow event", "argo kube event", "create resource", "remove resource",
 	"propalgation event", "search event", "confirm event",
-	"considers event", "admiralty config event", "minio config event",
+	"considers event", "admiralty config event", "minio config event", "pvc config event",
 	"workflow started event", "workflow step done event", "workflow done event",
+	"peer behavior event",
 }
 
 const (
@@ -53,6 +54,7 @@ const (
 	CONSIDERS_EVENT
 	ADMIRALTY_CONFIG_EVENT
 	MINIO_CONFIG_EVENT
+	PVC_CONFIG_EVENT
 
 	// Workflow lifecycle events emitted by oc-monitord.
 	// oc-scheduler listens to STARTED and DONE to maintain WorkflowExecution state.
@@ -60,6 +62,12 @@ const (
 	WORKFLOW_STARTED_EVENT
 	WORKFLOW_STEP_DONE_EVENT
 	WORKFLOW_DONE_EVENT
+
+	// PEER_BEHAVIOR_EVENT is emitted by any trusted service (oc-scheduler,
+	// oc-datacenter, …) when a peer exhibits suspicious or fraudulent behavior.
+	// oc-discovery consumes it to update the peer's trust score and auto-blacklist
+	// below threshold.
+	PEER_BEHAVIOR_EVENT
 )
 
 func (n NATSMethod) String() string {
@@ -70,7 +78,7 @@ func (n NATSMethod) String() string {
 func NameToMethod(name string) NATSMethod {
 	for _, v := range [...]NATSMethod{REMOVE_EXECUTION, CREATE_EXECUTION, PLANNER_EXECUTION, DISCOVERY, WORKFLOW_EVENT, ARGO_KUBE_EVENT,
 		CREATE_RESOURCE, REMOVE_RESOURCE, PROPALGATION_EVENT, SEARCH_EVENT, CONFIRM_EVENT,
-		CONSIDERS_EVENT, ADMIRALTY_CONFIG_EVENT, MINIO_CONFIG_EVENT,
+		CONSIDERS_EVENT, ADMIRALTY_CONFIG_EVENT, MINIO_CONFIG_EVENT, PVC_CONFIG_EVENT,
 		WORKFLOW_STARTED_EVENT, WORKFLOW_STEP_DONE_EVENT, WORKFLOW_DONE_EVENT} {
 		if strings.Contains(strings.ToLower(v.String()), strings.ToLower(name)) {
 			return v

tools/peer_behavior.go

@@ -0,0 +1,49 @@
+package tools
+
+import "time"
+
+// BehaviorSeverity qualifies the gravity of a peer misbehavior.
+type BehaviorSeverity int
+
+const (
+	// BehaviorWarn: minor inconsistency — slight trust penalty.
+	BehaviorWarn BehaviorSeverity = iota
+	// BehaviorFraud: deliberate data manipulation (e.g. fake peerless Ref,
+	// invalid booking) — significant trust penalty.
+	BehaviorFraud
+	// BehaviorCritical: severe abuse (secret exfiltration, data corruption,
+	// system-level attack) — heavy penalty, near-immediate blacklist.
+	BehaviorCritical
+)
+
+// scorePenalties maps each severity to a trust-score deduction (out of 100).
+var scorePenalties = map[BehaviorSeverity]float64{
+	BehaviorWarn:     5,
+	BehaviorFraud:    20,
+	BehaviorCritical: 40,
+}
+
+// Penalty returns the trust-score deduction for this severity.
+func (s BehaviorSeverity) Penalty() float64 {
+	if p, ok := scorePenalties[s]; ok {
+		return p
+	}
+	return 5
+}
+
+// PeerBehaviorReport is the payload carried by PEER_BEHAVIOR_EVENT.
+// Any trusted service can emit it; oc-discovery is the sole consumer.
+type PeerBehaviorReport struct {
+	// ReporterApp identifies the emitting service (e.g. "oc-scheduler", "oc-datacenter").
+	ReporterApp string `json:"reporter_app"`
+	// TargetPeerID is the MongoDB DID (_id) of the offending peer.
+	TargetPeerID string `json:"target_peer_id"`
+	// Severity drives how much the trust score drops.
+	Severity BehaviorSeverity `json:"severity"`
+	// Reason is a human-readable description shown in the blacklist warning.
+	Reason string `json:"reason"`
+	// Evidence is an optional reference (booking ID, resource Ref, …).
+	Evidence string `json:"evidence,omitempty"`
+	// At is the timestamp of the observed misbehavior.
+	At time.Time `json:"at"`
+}

tools/remote_caller.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"net/url"
 	"strings"
+	"sync"
 )
 
 // HTTP Method Enum defines the different methods that can be used to interact with the HTTP server
@@ -57,6 +58,7 @@ type HTTPCallerITF interface {
 type HTTPCaller struct {
 	URLS        map[DataType]map[METHOD]string // Map of the different methods and their urls
 	Disabled    bool                           // Disabled flag
+	Mu          sync.RWMutex
 	LastResults map[string]interface{} // Used to store information regarding the last execution of a given method on a given data type
 }
 
@@ -217,6 +219,8 @@ func (caller *HTTPCaller) CallForm(method string, url string, subpath string,
 }
 
 func (caller *HTTPCaller) StoreResp(resp *http.Response) error {
+	caller.Mu.Lock()
+	defer caller.Mu.Unlock()
 	caller.LastResults = make(map[string]interface{})
 	caller.LastResults["header"] = resp.Header
 	caller.LastResults["code"] = resp.StatusCode