oc-datacenter/infrastructure/nats.go

package infrastructure

import (
	"context"
	"encoding/json"
	"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 {
				// ── Minio credential provisioning ──────────────────────────────
				setter := minio.NewMinioSetter(argo.ExecutionsID, argo.MinioID)
				if argo.SourcePeerID == argo.DestPeerID {
					// 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 {
							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 {
				// ── Admiralty kubeconfig provisioning (existing behaviour) ──────
				if argo.SourcePeerID == argo.DestPeerID {
					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 {
						go tools.NewNATSCaller().SetNATSPub(tools.PROPALGATION_EVENT, tools.NATSResponse{
							FromApp:  "oc-datacenter",
							Datatype: -1,
							User:     resp.User,
							Method:   int(tools.PROPALGATION_EVENT),
							Payload:  b2,
						})
					}
				}
			}
		},

		// ─── PROPALGATION_EVENT ─────────────────────────────────────────────────────
		// Routes messages forwarded by oc-discovery to the right handler.
		tools.PROPALGATION_EVENT: func(resp tools.NATSResponse) {
			if resp.FromApp != "oc-discovery" {
				return
			}
			var prop tools.PropalgationMessage
			if err := json.Unmarshal(resp.Payload, &prop); err != nil {
				return
			}
			switch prop.Action {

			// ── Admiralty ──────────────────────────────────────────────────────
			case tools.PB_ADMIRALTY_CONFIG:
				kubeconfigEvent := KubeconfigEvent{}
				if err := json.Unmarshal(prop.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 {
						// Phase 1: no kubeconfig → this peer is the SOURCE (compute).
						NewAdmiraltySetter(kubeconfigEvent.ExecutionsID).InitializeAsSource(
							context.Background(), kubeconfigEvent.SourcePeerID, kubeconfigEvent.DestPeerID, kubeconfigEvent.OriginID)
					}
				}

			// ── Minio ──────────────────────────────────────────────────────────
			case tools.PB_MINIO_CONFIG:
				minioEvent := minio.MinioCredentialEvent{}
				if err := json.Unmarshal(prop.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 {
						// 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)
					}
				}

			// ── Deletion (routed by oc-discovery to the source peer) ───────────
			case tools.PB_DELETE:
				argo := &ArgoKubeEvent{}
				if err := json.Unmarshal(prop.Payload, argo); err != nil || argo.ExecutionsID == "" {
					return
				}
				if argo.Type == tools.STORAGE_RESOURCE {
					// Minio source teardown: revoke credentials + delete bucket.
					deleteEvent := minio.MinioDeleteEvent{}
					if err := json.Unmarshal(prop.Payload, &deleteEvent); err == nil {
						go minio.NewMinioSetter(deleteEvent.ExecutionsID, deleteEvent.MinioID).
							TeardownAsSource(context.Background(), deleteEvent)
					}
				} else {
					// Admiralty source teardown: delete AdmiraltySource + namespace.
					go NewAdmiraltySetter(argo.ExecutionsID).TeardownAsSource(context.Background())
				}
			}
		},
	})
}