Neo Oclib + Config Minio For private source

infrastructure/nats/nats.go

@@ -20,9 +20,10 @@ var roleWaiters sync.Map
 
 // ArgoKubeEvent carries the peer-routing metadata for a resource provisioning event.
 //
-// When MinioID is non-empty and Local is false, the event concerns Minio credential provisioning.
-// When Local is true, the event concerns local PVC provisioning.
-// Otherwise it concerns Admiralty kubeconfig provisioning.
+// Dispatch rules (evaluated in order):
+//  1. Type == STORAGE_RESOURCE                                   → Minio credentials or PVC
+//  2. (Type == PROCESSING_RESOURCE || DATA_RESOURCE) && SourceResourceID != "" → Minio++ pre-signed URL (Phase 4)
+//  3. everything else                                            → Admiralty kubeconfig
 type ArgoKubeEvent struct {
 	ExecutionsID string         `json:"executions_id"`
 	DestPeerID   string         `json:"dest_peer_id"`
@@ -36,8 +37,13 @@ type ArgoKubeEvent struct {
 	// response is routed back to this peer once provisioning completes.
 	OriginID string `json:"origin_id,omitempty"`
 	// Images is the list of container images to pre-pull on the target peer
-	// before the workflow starts. Empty for STORAGE_RESOURCE events.
+	// before the workflow starts. Empty for STORAGE_RESOURCE / PROCESSING_RESOURCE source events.
 	Images []string `json:"images,omitempty"`
+	// SourceResourceID is non-empty only for Phase 4 (isReachable=false):
+	// it identifies the Processing or Data resource whose binary/data must be
+	// fetched via a pre-signed Minio URL. When set, the event is a Minio++ request,
+	// NOT an Admiralty compute event.
+	SourceResourceID string `json:"source_resource_id,omitempty"`
 }
 
 // ListenNATS starts all NATS subscriptions for the infrastructure layer.
@@ -54,6 +60,9 @@ func ListenNATS() {
 			}
 			kube := kubernetes.NewKubernetesService(argo.ExecutionsID)
 
+			isSourcePresign := argo.SourceResourceID != "" &&
+				(argo.Type == tools.PROCESSING_RESOURCE || argo.Type == tools.DATA_RESOURCE)
+
 			if argo.Type == tools.STORAGE_RESOURCE {
 				if argo.Local {
 					fmt.Println("DETECT LOCAL PVC ARGO_KUBE_EVENT")
@@ -126,6 +135,40 @@ func ListenNATS() {
 						}
 					}
 				}
+			} else if isSourcePresign {
+				fmt.Println("DETECT SOURCE PRESIGN ARGO_KUBE_EVENT", argo.Type, argo.SourceResourceID)
+				// ── Minio++ : génération d'URL pré-signée pour source privée ──
+				presigner := storage.NewSourcePresigner(argo.ExecutionsID, argo.SourceResourceID)
+				event := storage.SourcePresignEvent{
+					ExecutionsID:     argo.ExecutionsID,
+					SourceResourceID: argo.SourceResourceID,
+					SourcePeerID:     argo.SourcePeerID,
+					DestPeerID:       argo.DestPeerID,
+					OriginID:         argo.OriginID,
+					DataType:         argo.Type.EnumIndex(),
+				}
+				if argo.SourcePeerID == argo.DestPeerID {
+					// Même peer : génère directement l'URL et émet CONSIDERS_EVENT local.
+					go presigner.InitializeAsSource(context.Background(), event, true)
+				} else {
+					// Cross-peer : route via PROPALGATION_EVENT(PB_SOURCE_PRESIGN)
+					// vers le peer propriétaire de la ressource.
+					if b, err := json.Marshal(event); err == nil {
+						if b2, err := json.Marshal(&tools.PropalgationMessage{
+							Payload: b,
+							Action:  tools.PB_SOURCE_PRESIGN,
+						}); err == nil {
+							fmt.Println("ROUTE SOURCE PRESIGN TO", argo.SourcePeerID)
+							go tools.NewNATSCaller().SetNATSPub(tools.PROPALGATION_EVENT, tools.NATSResponse{
+								FromApp:  "oc-datacenter",
+								Datatype: argo.Type,
+								User:     resp.User,
+								Method:   int(tools.PROPALGATION_EVENT),
+								Payload:  b2,
+							})
+						}
+					}
+				}
 			} else {
 				fmt.Println("DETECT COMPUTE ARGO_KUBE_EVENT")
 				// ── Pre-pull + Admiralty kubeconfig provisioning ─────────────
@@ -165,6 +208,20 @@ func ListenNATS() {
 			}
 		},
 
+		// ─── SOURCE_PRESIGN_EVENT ────────────────────────────────────────────────────
+		// Forwarded by oc-discovery after receiving PB_SOURCE_PRESIGN via libp2p
+		// ProtocolSourcePresignResource. This peer is the resource owner (Minio source).
+		// It generates a pre-signed URL and responds via PB_CONSIDERS → OriginID.
+		tools.SOURCE_PRESIGN_EVENT: func(resp tools.NATSResponse) {
+			event := storage.SourcePresignEvent{}
+			if err := json.Unmarshal(resp.Payload, &event); err != nil {
+				return
+			}
+			fmt.Println("SOURCE_PRESIGN_EVENT received resource=", event.SourceResourceID)
+			presigner := storage.NewSourcePresigner(event.ExecutionsID, event.SourceResourceID)
+			go presigner.InitializeAsSource(context.Background(), event, false)
+		},
+
 		// ─── ADMIRALTY_CONFIG_EVENT ─────────────────────────────────────────────────
 		// Forwarded by oc-discovery after receiving via libp2p ProtocolAdmiraltyConfigResource.
 		// Payload is a KubeconfigEvent (phase discriminated by Kubeconfig presence).