package infrastructure import ( "encoding/json" "fmt" "time" oclib "cloud.o-forge.io/core/oc-lib" "cloud.o-forge.io/core/oc-lib/models/booking" "cloud.o-forge.io/core/oc-lib/models/booking/planner" "cloud.o-forge.io/core/oc-lib/models/common/enum" "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/models/workflow" "cloud.o-forge.io/core/oc-lib/tools" ) func handleConfirm(resp tools.NATSResponse) { confirmResource(string(resp.Payload), resp.Datatype) } func handlePlannerExecution(resp tools.NATSResponse) { m := map[string]interface{}{} p := planner.Planner{} if err := json.Unmarshal(resp.Payload, &m); err != nil { return } if err := json.Unmarshal(resp.Payload, &p); err != nil { return } storePlanner(fmt.Sprintf("%v", m["peer_id"]), &p) } func handlePropagationEvent(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 { case tools.PB_CONSIDERS: fmt.Println("PB_CONSIDERS") switch tools.DataType(prop.DataType) { case tools.BOOKING, tools.PURCHASE_RESOURCE: fmt.Println("updateExecutionState", tools.DataType(prop.DataType)) updateExecutionState(prop.Payload, tools.DataType(prop.DataType)) case tools.WORKFLOW_EXECUTION: confirmExecutionDrafts(prop.Payload) } } } func handleRemoveResource(resp tools.NATSResponse) { switch resp.Datatype { case tools.WORKFLOW: wf := workflow.Workflow{} if err := json.Unmarshal(resp.Payload, &wf); err != nil { return } notifyWorkflowWatchers(wf.GetID()) case tools.BOOKING: var p removeResourcePayload if err := json.Unmarshal(resp.Payload, &p); err != nil { return } self, err := oclib.GetMySelf() if err != nil || self == nil { return } adminReq := &tools.APIRequest{Admin: true} res, _, loadErr := booking.NewAccessor(adminReq).LoadOne(p.ID) if loadErr != nil || res == nil { return } existing := res.(*booking.Booking) if existing.SchedulerPeerID != p.SchedulerPeerID || existing.ExecutionsID != p.ExecutionsID { fmt.Println("ListenNATS REMOVE_RESOURCE booking: auth mismatch, ignoring", p.ID) return } booking.NewAccessor(adminReq).DeleteOne(p.ID) go refreshSelfPlanner(self.PeerID, adminReq) case tools.PURCHASE_RESOURCE: var p removeResourcePayload if err := json.Unmarshal(resp.Payload, &p); err != nil { return } adminReq := &tools.APIRequest{Admin: true} res, _, loadErr := purchase_resource.NewAccessor(adminReq).LoadOne(p.ID) if loadErr != nil || res == nil { return } existing := res.(*purchase_resource.PurchaseResource) if existing.SchedulerPeerID != p.SchedulerPeerID || existing.ExecutionsID != p.ExecutionsID { fmt.Println("ListenNATS REMOVE_RESOURCE purchase: auth mismatch, ignoring", p.ID) return } purchase_resource.NewAccessor(adminReq).DeleteOne(p.ID) } } func handleCreateBooking(bk *booking.Booking, self *peer.Peer, adminReq *tools.APIRequest) { // Upsert: if a booking with this ID already exists, verify auth and update. if existing, _, loadErr := booking.NewAccessor(adminReq).LoadOne(bk.GetID()); loadErr == nil && existing != nil { prev := existing.(*booking.Booking) if prev.SchedulerPeerID != bk.SchedulerPeerID || prev.ExecutionsID != bk.ExecutionsID { fmt.Println("ListenNATS CREATE_RESOURCE booking upsert: auth mismatch, ignoring", bk.GetID()) return } if !prev.IsDrafted() && bk.IsDraft { // Already confirmed, refuse downgrade. return } // Expired check only on confirmation (IsDraft→false). if !bk.IsDraft && !prev.ExpectedStartDate.IsZero() && prev.ExpectedStartDate.Before(time.Now()) { fmt.Println("ListenNATS CREATE_RESOURCE booking: expired, deleting", bk.GetID()) booking.NewAccessor(adminReq).DeleteOne(bk.GetID()) return } if _, _, err := utils.GenericRawUpdateOne(bk, bk.GetID(), booking.NewAccessor(adminReq)); err != nil { fmt.Println("ListenNATS CREATE_RESOURCE booking update failed:", err) return } go refreshSelfPlanner(self.PeerID, adminReq) if !bk.IsDraft { go applyConsidersLocal(bk.GetID(), tools.BOOKING) } return } // New booking: standard create flow. if !bk.ExpectedStartDate.IsZero() && bk.ExpectedStartDate.Before(time.Now()) { fmt.Println("ListenNATS: booking start date is in the past, discarding") return } plannerMu.RLock() selfEntry := PlannerCache[self.PeerID] plannerMu.RUnlock() if selfEntry != nil && selfEntry.Planner != nil && !checkInstance(selfEntry.Planner, bk.ResourceID, bk.InstanceID, bk.ExpectedStartDate, bk.ExpectedEndDate) { fmt.Println("ListenNATS: booking conflicts with local planner, discarding") return } bk.IsDraft = true stored, _, err := booking.NewAccessor(adminReq).StoreOne(bk) if err != nil { fmt.Println("ListenNATS: could not store booking:", err) return } storedID := stored.GetID() go refreshSelfPlanner(self.PeerID, adminReq) time.AfterFunc(10*time.Minute, func() { draftTimeout(storedID, tools.BOOKING) }) } func handleCreatePurchase(pr *purchase_resource.PurchaseResource, self *peer.Peer, adminReq *tools.APIRequest) { if pr.DestPeerID != self.GetID() { return } // Upsert: if a purchase with this ID already exists, verify auth and update. if existing, _, loadErr := purchase_resource.NewAccessor(adminReq).LoadOne(pr.GetID()); loadErr == nil && existing != nil { prev := existing.(*purchase_resource.PurchaseResource) if prev.SchedulerPeerID != pr.SchedulerPeerID || prev.ExecutionsID != pr.ExecutionsID { fmt.Println("ListenNATS CREATE_RESOURCE purchase upsert: auth mismatch, ignoring", pr.GetID()) return } if !prev.IsDrafted() && pr.IsDraft { return } if _, _, err := utils.GenericRawUpdateOne(pr, pr.GetID(), purchase_resource.NewAccessor(adminReq)); err != nil { fmt.Println("ListenNATS CREATE_RESOURCE purchase update failed:", err) return } if !pr.IsDraft { go applyConsidersLocal(pr.GetID(), tools.PURCHASE_RESOURCE) } return } // New purchase: standard create flow. pr.IsDraft = true stored, _, err := purchase_resource.NewAccessor(adminReq).StoreOne(pr) if err != nil { fmt.Println("ListenNATS: could not store purchase:", err) return } storedID := stored.GetID() time.AfterFunc(10*time.Minute, func() { draftTimeout(storedID, tools.PURCHASE_RESOURCE) }) } func handleCreateResource(resp tools.NATSResponse) { switch resp.Datatype { case tools.WORKFLOW: wf := workflow.Workflow{} if err := json.Unmarshal(resp.Payload, &wf); err != nil { return } broadcastPlanner(&wf) notifyWorkflowWatchers(wf.GetID()) case tools.BOOKING: var bk booking.Booking if err := json.Unmarshal(resp.Payload, &bk); err != nil { return } self, err := oclib.GetMySelf() /*if err != nil || self == nil || bk.DestPeerID != self.GetID() { return }*/ adminReq := &tools.APIRequest{Admin: true} _ = err handleCreateBooking(&bk, self, adminReq) case tools.PURCHASE_RESOURCE: var pr purchase_resource.PurchaseResource if err := json.Unmarshal(resp.Payload, &pr); err != nil { return } self, err := oclib.GetMySelf() if err != nil || self == nil { return } adminReq := &tools.APIRequest{Admin: true} handleCreatePurchase(&pr, self, adminReq) } } // confirmResource sets IsDraft=false for a booking or purchase resource. // For bookings it also advances State to SCHEDULED and refreshes the local planner. func confirmResource(id string, dt tools.DataType) { adminReq := &tools.APIRequest{Admin: true} switch dt { case tools.BOOKING: res, _, err := booking.NewAccessor(adminReq).LoadOne(id) if err != nil || res == nil { fmt.Printf("confirmResource: could not load booking %s: %v\n", id, err) return } bk := res.(*booking.Booking) bk.IsDraft = false bk.State = enum.SCHEDULED if _, _, err := utils.GenericRawUpdateOne(bk, id, booking.NewAccessor(adminReq)); err != nil { fmt.Printf("confirmResource: could not confirm booking %s: %v\n", id, err) return } createNamespace(bk.ExecutionsID) // create Namespace locally self, err := oclib.GetMySelf() if err == nil && self != nil { go refreshSelfPlanner(self.PeerID, adminReq) } case tools.PURCHASE_RESOURCE: res, _, err := purchase_resource.NewAccessor(adminReq).LoadOne(id) if err != nil || res == nil { fmt.Printf("confirmResource: could not load purchase %s: %v\n", id, err) return } pr := res.(*purchase_resource.PurchaseResource) pr.IsDraft = false if _, _, err := utils.GenericRawUpdateOne(pr, id, purchase_resource.NewAccessor(adminReq)); err != nil { fmt.Printf("confirmResource: could not confirm purchase %s: %v\n", id, err) } } }