oc-datacenter/infrastructure/nats.go

package infrastructure

import (
	"context"
	"encoding/json"
	"fmt"
	"oc-datacenter/infrastructure/minio"
	"sync"

	"cloud.o-forge.io/core/oc-lib/tools"
)

// roleWaiters maps executionID → channel expecting the role-assignment message from OC discovery.
var roleWaiters sync.Map

// ArgoKubeEvent carries the peer-routing metadata for a resource provisioning event.
//
// When MinioID is non-empty the event concerns Minio credential provisioning;
// otherwise it concerns Admiralty kubeconfig provisioning.
type ArgoKubeEvent struct {
	ExecutionsID string         `json:"executions_id"`
	DestPeerID   string         `json:"dest_peer_id"`
	Type         tools.DataType `json:"data_type"`
	SourcePeerID string         `json:"source_peer_id"`
	MinioID      string         `json:"minio_id,omitempty"`
	// OriginID is the peer that initiated the request; the PB_CONSIDERS
	// response is routed back to this peer once provisioning completes.
	OriginID string `json:"origin_id,omitempty"`
}

// ListenNATS starts all NATS subscriptions for the infrastructure layer.
// Must be launched in a goroutine from main.
func ListenNATS() {
	tools.NewNATSCaller().ListenNats(map[tools.NATSMethod]func(tools.NATSResponse){
		// ─── ARGO_KUBE_EVENT ────────────────────────────────────────────────────────
		// Triggered by oc-discovery to notify this peer of a provisioning task.
		// Dispatches to Admiralty or Minio based on whether MinioID is set.
		tools.ARGO_KUBE_EVENT: func(resp tools.NATSResponse) {
			argo := &ArgoKubeEvent{}
			if err := json.Unmarshal(resp.Payload, argo); err != nil {
				return
			}

			if argo.Type == tools.STORAGE_RESOURCE {
				fmt.Println("DETECT STORAGE ARGO_KUBE_EVENT")
				// ── Minio credential provisioning ──────────────────────────────
				setter := minio.NewMinioSetter(argo.ExecutionsID, argo.MinioID)
				if argo.SourcePeerID == argo.DestPeerID {
					fmt.Println("CONFIG MYSELF")
					err := CreateNamespace(argo.ExecutionsID)
					fmt.Println("NS", err)
					// Same peer: source creates credentials and immediately stores them.
					go setter.InitializeAsSource(context.Background(), argo.SourcePeerID, argo.DestPeerID, argo.OriginID)
				} else {
					// Different peers: publish Phase-1 PB_MINIO_CONFIG (Access == "")
					// so oc-discovery routes the role-assignment to the Minio host.
					phase1 := minio.MinioCredentialEvent{
						ExecutionsID: argo.ExecutionsID,
						MinioID:      argo.MinioID,
						SourcePeerID: argo.SourcePeerID,
						DestPeerID:   argo.DestPeerID,
						OriginID:     argo.OriginID,
					}
					if b, err := json.Marshal(phase1); err == nil {
						if b2, err := json.Marshal(&tools.PropalgationMessage{
							Payload: b,
							Action:  tools.PB_MINIO_CONFIG,
						}); err == nil {
							fmt.Println("CONFIG THEM")
							go tools.NewNATSCaller().SetNATSPub(tools.PROPALGATION_EVENT, tools.NATSResponse{
								FromApp:  "oc-datacenter",
								Datatype: -1,
								User:     resp.User,
								Method:   int(tools.PROPALGATION_EVENT),
								Payload:  b2,
							})
						}
					}
				}
			} else {
				fmt.Println("DETECT COMPUTE ARGO_KUBE_EVENT")
				// ── Admiralty kubeconfig provisioning (existing behaviour) ──────
				if argo.SourcePeerID == argo.DestPeerID {
					fmt.Println("CONFIG MYSELF")
					err := CreateNamespace(argo.ExecutionsID)
					fmt.Println("NS", err)
					go NewAdmiraltySetter(argo.ExecutionsID).InitializeAsSource(
						context.Background(), argo.SourcePeerID, argo.DestPeerID, argo.OriginID)
				} else if b, err := json.Marshal(argo); err == nil {
					if b2, err := json.Marshal(&tools.PropalgationMessage{
						Payload: b,
						Action:  tools.PB_ADMIRALTY_CONFIG,
					}); err == nil {
						fmt.Println("CONFIG THEM")
						go tools.NewNATSCaller().SetNATSPub(tools.PROPALGATION_EVENT, tools.NATSResponse{
							FromApp:  "oc-datacenter",
							Datatype: -1,
							User:     resp.User,
							Method:   int(tools.PROPALGATION_EVENT),
							Payload:  b2,
						})
					}
				}
			}
		},

		// ─── ADMIRALTY_CONFIG_EVENT ─────────────────────────────────────────────────
		// Forwarded by oc-discovery after receiving via libp2p ProtocolAdmiraltyConfigResource.
		// Payload is a KubeconfigEvent (phase discriminated by Kubeconfig presence).
		tools.ADMIRALTY_CONFIG_EVENT: func(resp tools.NATSResponse) {

			kubeconfigEvent := KubeconfigEvent{}
			if err := json.Unmarshal(resp.Payload, &kubeconfigEvent); err == nil {
				if kubeconfigEvent.Kubeconfig != "" {
					// Phase 2: kubeconfig present → this peer is the TARGET (scheduler).
					NewAdmiraltySetter(kubeconfigEvent.ExecutionsID).InitializeAsTarget(
						context.Background(), kubeconfigEvent)
				} else {
					err := CreateNamespace(kubeconfigEvent.ExecutionsID)
					fmt.Println("NS", err)
					// Phase 1: no kubeconfig → this peer is the SOURCE (compute).
					NewAdmiraltySetter(kubeconfigEvent.ExecutionsID).InitializeAsSource(
						context.Background(), kubeconfigEvent.SourcePeerID, kubeconfigEvent.DestPeerID, kubeconfigEvent.OriginID)
				}
			}
		},

		// ─── MINIO_CONFIG_EVENT ──────────────────────────────────────────────────────
		// Forwarded by oc-discovery after receiving via libp2p ProtocolMinioConfigResource.
		// Payload is a MinioCredentialEvent (phase discriminated by Access presence).
		tools.MINIO_CONFIG_EVENT: func(resp tools.NATSResponse) {
			minioEvent := minio.MinioCredentialEvent{}
			if err := json.Unmarshal(resp.Payload, &minioEvent); err == nil {
				if minioEvent.Access != "" {
					// Phase 2: credentials present → this peer is the TARGET (compute).
					minio.NewMinioSetter(minioEvent.ExecutionsID, minioEvent.MinioID).InitializeAsTarget(
						context.Background(), minioEvent)
				} else {
					err := CreateNamespace(minioEvent.ExecutionsID)
					fmt.Println("NS", err)
					// Phase 1: no credentials → this peer is the SOURCE (Minio host).
					minio.NewMinioSetter(minioEvent.ExecutionsID, minioEvent.MinioID).InitializeAsSource(
						context.Background(), minioEvent.SourcePeerID, minioEvent.DestPeerID, minioEvent.OriginID)
				}
			}
		},

		// ─── REMOVE_RESOURCE ────────────────────────────────────────────────────────
		// Routed by oc-discovery via ProtocolDeleteResource for datacenter teardown.
		// Only STORAGE_RESOURCE and COMPUTE_RESOURCE deletions are handled here.
		tools.REMOVE_RESOURCE: func(resp tools.NATSResponse) {
			switch resp.Datatype {
			case tools.STORAGE_RESOURCE:
				deleteEvent := minio.MinioDeleteEvent{}
				if err := json.Unmarshal(resp.Payload, &deleteEvent); err == nil && deleteEvent.ExecutionsID != "" {
					go minio.NewMinioSetter(deleteEvent.ExecutionsID, deleteEvent.MinioID).
						TeardownAsSource(context.Background(), deleteEvent)
				}
			case tools.COMPUTE_RESOURCE:
				argo := &ArgoKubeEvent{}
				if err := json.Unmarshal(resp.Payload, argo); err == nil && argo.ExecutionsID != "" {
					go NewAdmiraltySetter(argo.ExecutionsID).TeardownAsSource(context.Background())
				}
			}
		},
	})
}