oc-lib/models/workflow/workflow.go

package workflow

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"mime/multipart"
	"strings"
	"time"

	"cloud.o-forge.io/core/oc-lib/models/collaborative_area/shallow_collaborative_area"
	"cloud.o-forge.io/core/oc-lib/models/common"
	"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"
	"cloud.o-forge.io/core/oc-lib/models/utils"
	"cloud.o-forge.io/core/oc-lib/models/workflow/graph"
	"cloud.o-forge.io/core/oc-lib/tools"
	"github.com/google/uuid"
)

/*
* Workflow is a struct that represents a workflow
* it defines the native workflow
 */
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
}

func (d *Workflow) GetAccessor(request *tools.APIRequest) utils.Accessor {
	return NewAccessor(request) // Create a new instance of the accessor
}

func (d *Workflow) GetResources(dt tools.DataType) []resources.ResourceInterface {
	itf := []resources.ResourceInterface{}
	switch dt {
	case tools.DATA_RESOURCE:
		for _, d := range d.DataResources {
			itf = append(itf, d)
		}
		return itf
	case tools.PROCESSING_RESOURCE:
		for _, d := range d.ProcessingResources {
			itf = append(itf, d)
		}
		return itf
	case tools.COMPUTE_RESOURCE:
		for _, d := range d.ComputeResources {
			itf = append(itf, d)
		}
		return itf
	case tools.WORKFLOW_RESOURCE:
		for _, d := range d.WorkflowResources {
			itf = append(itf, d)
		}
		return itf
	case tools.STORAGE_RESOURCE:
		for _, d := range d.StorageResources {
			itf = append(itf, d)
		}
		return itf
	}
	return itf
}

func (d *Workflow) ExtractFromPlantUML(plantUML multipart.File, request *tools.APIRequest) (*Workflow, error) {
	if plantUML == nil {
		return d, errors.New("no file available to export")
	}

	defer plantUML.Close()

	d.Datas = []string{}
	d.Storages = []string{}
	d.Processings = []string{}
	d.Computes = []string{}
	d.Workflows = []string{}

	d.DataResources = []*resources.DataResource{}
	d.StorageResources = []*resources.StorageResource{}
	d.ProcessingResources = []*resources.ProcessingResource{}
	d.ComputeResources = []*resources.ComputeResource{}
	d.WorkflowResources = []*resources.WorkflowResource{}

	d.Graph = graph.NewGraph()
	resourceCatalog := map[string]func() resources.ResourceInterface{
		"Processing": func() resources.ResourceInterface {
			return &resources.ProcessingResource{
				AbstractInstanciatedResource: resources.AbstractInstanciatedResource[*resources.ProcessingInstance]{
					Instances: []*resources.ProcessingInstance{},
				},
			}
		},
		"Storage": func() resources.ResourceInterface {
			return &resources.StorageResource{
				AbstractInstanciatedResource: resources.AbstractInstanciatedResource[*resources.StorageResourceInstance]{
					Instances: []*resources.StorageResourceInstance{},
				},
			}
		},
		"Data": func() resources.ResourceInterface {
			return &resources.DataResource{
				AbstractInstanciatedResource: resources.AbstractInstanciatedResource[*resources.DataInstance]{
					Instances: []*resources.DataInstance{},
				},
			}
		},
		"ComputeUnit": func() resources.ResourceInterface {
			return &resources.ComputeResource{
				AbstractInstanciatedResource: resources.AbstractInstanciatedResource[*resources.ComputeResourceInstance]{
					Instances: []*resources.ComputeResourceInstance{},
				},
			}
		},
	}
	graphVarName := map[string]*graph.GraphItem{}
	scanner := bufio.NewScanner(plantUML)

	for scanner.Scan() {
		line := scanner.Text()
		for n, new := range resourceCatalog {
			if strings.Contains(line, n+"(") && !strings.Contains(line, "!procedure") { // should exclude declaration of type.
				newRes := new()
				varName, graphItem, err := d.extractResourcePlantUML(line, newRes, n, request.PeerID)
				if err != nil {
					return d, err
				}
				graphVarName[varName] = graphItem
				continue
			} else if strings.Contains(line, n+"-->") {
				err := d.extractLink(line, graphVarName, "-->")
				if err != nil {
					fmt.Println(err)
					continue
				}
			} else if strings.Contains(line, n+"--") {
				err := d.extractLink(line, graphVarName, "--")
				if err != nil {
					fmt.Println(err)
					continue
				}
			} else if strings.Contains(line, n+"-") {
				err := d.extractLink(line, graphVarName, "-")
				if err != nil {
					fmt.Println(err)
					continue
				}
			}
		}
	}
	if err := scanner.Err(); err != nil {
		return d, err
	}
	d.generateResource(d.GetResources(tools.DATA_RESOURCE), request)
	d.generateResource(d.GetResources(tools.PROCESSING_RESOURCE), request)
	d.generateResource(d.GetResources(tools.STORAGE_RESOURCE), request)
	d.generateResource(d.GetResources(tools.COMPUTE_RESOURCE), request)
	d.generateResource(d.GetResources(tools.WORKFLOW_RESOURCE), request)
	return d, nil
}

func (d *Workflow) generateResource(datas []resources.ResourceInterface, request *tools.APIRequest) error {
	for _, d := range datas {
		access := d.GetAccessor(request)
		if _, code, err := access.LoadOne(d.GetID()); err != nil && code == 200 {
			continue
		}
		access.StoreOne(d)
	}
	return nil
}

func (d *Workflow) extractLink(line string, graphVarName map[string]*graph.GraphItem, pattern string) error {
	splitted := strings.Split(line, pattern)
	if len(splitted) < 2 {
		return errors.New("links elements not found")
	}
	if graphVarName[splitted[0]] != nil {
		return errors.New("links elements not found -> " + strings.Trim(splitted[0], " "))
	}
	if graphVarName[splitted[1]] != nil {
		return errors.New("links elements not found -> " + strings.Trim(splitted[1], " "))
	}
	link := &graph.GraphLink{
		Source: graph.Position{
			ID: graphVarName[splitted[0]].ID,
			X:  0,
			Y:  0,
		},
		Destination: graph.Position{
			ID: graphVarName[splitted[1]].ID,
			X:  0,
			Y:  0,
		},
	}
	splittedComments := strings.Split(line, "'")
	if len(splittedComments) <= 1 {
		return errors.New("Can't deserialize Object, there's no commentary")
	}
	comment := strings.ReplaceAll(splittedComments[1], "'", "") // for now it's a json.
	json.Unmarshal([]byte(comment), link)
	d.Graph.Links = append(d.Graph.Links, *link)
	return nil
}

func (d *Workflow) extractResourcePlantUML(line string, resource resources.ResourceInterface, dataName string, peerID string) (string, *graph.GraphItem, error) {
	splittedFunc := strings.Split(line, "(")
	if len(splittedFunc) <= 1 {
		return "", nil, errors.New("Can't deserialize Object, there's no func")
	}
	splittedParams := strings.Split(splittedFunc[1], ",")
	if len(splittedFunc) <= 1 {
		return "", nil, errors.New("Can't deserialize Object, there's no params")
	}

	varName := splittedParams[0]
	splitted := strings.Split(splittedParams[1], "\"")

	if len(splitted) <= 1 {
		return "", nil, errors.New("Can't deserialize Object, there's no name")
	}
	resource.SetName(splitted[1])

	splittedComments := strings.Split(line, "'")
	if len(splittedComments) <= 1 {
		return "", nil, errors.New("Can't deserialize Object, there's no commentary")
	}
	comment := strings.ReplaceAll(splittedComments[1], "'", "") // for now it's a json.
	instance := d.getNewInstance(dataName, splitted[1], peerID)
	if instance == nil {
		return "", nil, errors.New("No instance found.")
	}
	resource.AddInstances(instance)

	json.Unmarshal([]byte(comment), instance)
	// deserializer les instances... une instance doit par défaut avoir certaines valeurs d'accès.
	graphID := uuid.New()
	graphItem := &graph.GraphItem{
		ID: graphID.String(),
	}
	graphItem = d.getNewGraphItem(dataName, graphItem, resource)

	d.Graph.Items[graphID.String()] = *graphItem
	return varName, graphItem, nil
}

func (d *Workflow) getNewGraphItem(dataName string, graphItem *graph.GraphItem, resource resources.ResourceInterface) *graph.GraphItem {
	switch dataName {
	case "Data":
		d.Datas = append(d.Datas, resource.GetID())
		d.DataResources = append(d.DataResources, resource.(*resources.DataResource))
		graphItem.Data = resource.(*resources.DataResource)
	case "Processing", "Event":
		d.Processings = append(d.Processings, resource.GetID())
		d.ProcessingResources = append(d.ProcessingResources, resource.(*resources.ProcessingResource))
		graphItem.Processing = resource.(*resources.ProcessingResource)
	case "Storage":
		d.Storages = append(d.Storages, resource.GetID())
		d.StorageResources = append(d.StorageResources, resource.(*resources.StorageResource))
		graphItem.Storage = resource.(*resources.StorageResource)
	case "ComputeUnit":
		d.Computes = append(d.Computes, resource.GetID())
		d.ComputeResources = append(d.ComputeResources, resource.(*resources.ComputeResource))
		graphItem.Compute = resource.(*resources.ComputeResource)
	default:
		return graphItem
	}
	return graphItem
}

func (d *Workflow) getNewInstance(dataName string, name string, peerID string) resources.ResourceInstanceITF {
	switch dataName {
	case "Data":
		return resources.NewDataInstance(name, peerID)
	case "Processing":
		return resources.NewProcessingInstance(name, peerID)
	case "Storage":
		return resources.NewStorageResourceInstance(name, peerID)
	case "ComputeUnit":
		return resources.NewComputeResourceInstance(name, peerID)
	default:
		return nil
	}
}

type Deps struct {
	Source string
	Dest   string
}

func (w *Workflow) IsDependancy(id string) []Deps {
	dependancyOfIDs := []Deps{}
	for _, link := range w.Graph.Links {
		if _, ok := w.Graph.Items[link.Destination.ID]; !ok {
			continue
		}
		source := w.Graph.Items[link.Destination.ID].Processing
		if id == link.Source.ID && source != nil {
			dependancyOfIDs = append(dependancyOfIDs, Deps{Source: source.GetName(), Dest: link.Destination.ID})
		}
		sourceWF := w.Graph.Items[link.Destination.ID].Workflow
		if id == link.Source.ID && sourceWF != nil {
			dependancyOfIDs = append(dependancyOfIDs, Deps{Source: sourceWF.GetName(), Dest: link.Destination.ID})
		}
	}
	return dependancyOfIDs
}

func (w *Workflow) GetDependencies(id string) (dependencies []Deps) {
	for _, link := range w.Graph.Links {
		if _, ok := w.Graph.Items[link.Source.ID]; !ok {
			continue
		}
		source := w.Graph.Items[link.Source.ID].Processing
		if id == link.Destination.ID && source != nil {
			dependencies = append(dependencies, Deps{Source: source.GetName(), Dest: link.Source.ID})
			continue
		}
	}
	return
}

func (w *Workflow) GetGraphItems(f func(item graph.GraphItem) bool) (list_datas []graph.GraphItem) {
	for _, item := range w.Graph.Items {
		if f(item) {
			list_datas = append(list_datas, item)
		}
	}
	return
}

func (w *Workflow) GetPricedItem(
	f func(item graph.GraphItem) bool, request *tools.APIRequest, buyingStrategy int, pricingStrategy int) map[string]pricing.PricedItemITF {
	list_datas := map[string]pricing.PricedItemITF{}
	for _, item := range w.Graph.Items {
		if f(item) {
			dt, res := item.GetResource()
			ord := res.ConvertToPricedResource(dt, request)
			list_datas[res.GetID()] = ord
		}
	}
	return list_datas
}

type Related struct {
	Node  resources.ResourceInterface
	Links []graph.GraphLink
}

func (w *Workflow) GetByRelatedProcessing(processingID string, g func(item graph.GraphItem) bool) map[string]Related {
	related := map[string]Related{}
	for _, link := range w.Graph.Links {
		nodeID := link.Destination.ID
		var node resources.ResourceInterface
		if g(w.Graph.Items[link.Source.ID]) {
			item := w.Graph.Items[link.Source.ID]
			_, node = item.GetResource()
		}
		if node == nil && g(w.Graph.Items[link.Destination.ID]) { // if the source is not a storage, we consider that the destination is the storage
			nodeID = link.Source.ID
			item := w.Graph.Items[link.Destination.ID] // and the processing is the source
			_, node = item.GetResource()               // we are looking for the storage as destination
		}
		if processingID == nodeID && node != nil { // if the storage is linked to the processing
			relID := node.GetID()
			rel := Related{}
			rel.Node = node
			rel.Links = append(rel.Links, link)
			related[relID] = rel
		}
	}
	return related
}

func (ao *Workflow) VerifyAuth(request *tools.APIRequest) bool {
	isAuthorized := false
	if len(ao.Shared) > 0 {
		for _, shared := range ao.Shared {
			shared, code, _ := shallow_collaborative_area.NewAccessor(request).LoadOne(shared)
			if code != 200 || shared == nil {
				isAuthorized = false
			} else {
				isAuthorized = shared.VerifyAuth(request)
			}
		}
	}
	return ao.AbstractObject.VerifyAuth(request) || isAuthorized
}

/*
* CheckBooking is a function that checks the booking of the workflow on peers (even ourselves)
 */
func (wfa *Workflow) CheckBooking(caller *tools.HTTPCaller) (bool, error) {
	// check if
	if wfa.Graph == nil { // no graph no booking
		return false, nil
	}
	accessor := (&resources.ComputeResource{}).GetAccessor(&tools.APIRequest{Caller: caller})
	for _, link := range wfa.Graph.Links {
		if ok, compute_id := link.IsComputeLink(*wfa.Graph); ok { // check if the link is a link between a compute and a resource
			compute, code, _ := accessor.LoadOne(compute_id)
			if code != 200 {
				continue
			}
			// CHECK BOOKING ON PEER, compute could be a remote one
			peerID := compute.(*resources.ComputeResource).CreatorID
			if peerID == "" {
				return false, errors.New("no peer id")
			} // no peer id no booking, we need to know where to book
			_, err := (&peer.Peer{}).LaunchPeerExecution(peerID, compute_id, tools.BOOKING, tools.GET, nil, caller)
			if err != nil {
				return false, err
			}
		}
	}
	return true, nil
}

func (wf *Workflow) Planify(start time.Time, end *time.Time, request *tools.APIRequest) (float64, map[tools.DataType]map[string]pricing.PricedItemITF, *Workflow, error) {
	priceds := map[tools.DataType]map[string]pricing.PricedItemITF{}
	ps, priceds, err := plan[*resources.ProcessingResource](tools.PROCESSING_RESOURCE, wf, priceds, request, wf.Graph.IsProcessing,
		func(res resources.ResourceInterface, priced pricing.PricedItemITF) (time.Time, float64) {
			return start.Add(time.Duration(wf.Graph.GetAverageTimeProcessingBeforeStart(0, res.GetID(), request)) * time.Second), priced.GetExplicitDurationInS()
		}, func(started time.Time, duration float64) *time.Time {
			s := started.Add(time.Duration(duration))
			return &s
		})
	if err != nil {
		return 0, priceds, nil, err
	}
	if _, priceds, err = plan[resources.ResourceInterface](tools.DATA_RESOURCE, wf, priceds, request,
		wf.Graph.IsData, func(res resources.ResourceInterface, priced pricing.PricedItemITF) (time.Time, float64) {
			return start, 0
		}, func(started time.Time, duration float64) *time.Time {
			return end
		}); err != nil {
		return 0, priceds, nil, err
	}
	for k, f := range map[tools.DataType]func(graph.GraphItem) bool{tools.STORAGE_RESOURCE: wf.Graph.IsStorage, tools.COMPUTE_RESOURCE: wf.Graph.IsCompute} {
		if _, priceds, err = plan[resources.ResourceInterface](k, wf, priceds, request,
			f, func(res resources.ResourceInterface, priced pricing.PricedItemITF) (time.Time, float64) {
				nearestStart, longestDuration := wf.Graph.GetAverageTimeRelatedToProcessingActivity(start, ps, res, func(i graph.GraphItem) (r resources.ResourceInterface) {
					if f(i) {
						_, r = i.GetResource()
					}
					return r
				}, request)
				return start.Add(time.Duration(nearestStart) * time.Second), longestDuration
			}, func(started time.Time, duration float64) *time.Time {
				s := started.Add(time.Duration(duration))
				return &s
			}); err != nil {
			return 0, priceds, nil, err
		}
	}
	longest := common.GetPlannerLongestTime(end, priceds, request)
	if _, priceds, err = plan[resources.ResourceInterface](tools.WORKFLOW_RESOURCE, wf, priceds, request, wf.Graph.IsWorkflow,
		func(res resources.ResourceInterface, priced pricing.PricedItemITF) (time.Time, float64) {
			start := start.Add(time.Duration(common.GetPlannerNearestStart(start, priceds, request)) * time.Second)
			longest := float64(-1)
			r, code, err := res.GetAccessor(request).LoadOne(res.GetID())
			if code != 200 || err != nil {
				return start, longest
			}
			if neoLongest, _, _, err := r.(*Workflow).Planify(start, end, request); err != nil {
				return start, longest
			} else if neoLongest > longest {
				longest = neoLongest
			}
			return start.Add(time.Duration(common.GetPlannerNearestStart(start, priceds, request)) * time.Second), longest
		}, func(start time.Time, longest float64) *time.Time {
			s := start.Add(time.Duration(longest) * time.Second)
			return &s
		}); err != nil {
		return 0, priceds, nil, err
	}
	return longest, priceds, wf, nil
}

// Returns a map of DataType (processing,computing,data,storage,worfklow) where each resource (identified by its UUID)
// is mapped to the list of its items (different appearance) in the graph
// ex: if the same Minio storage is represented by several nodes in the graph, in [tools.STORAGE_RESSOURCE] its UUID will be mapped to
// the list of GraphItem ID that correspond to the ID of each node
func (w *Workflow) GetItemsByResources() map[tools.DataType]map[string][]string {
	res := make(map[tools.DataType]map[string][]string)
	dtMethodMap := map[tools.DataType]func() []graph.GraphItem{
		tools.STORAGE_RESOURCE:    func() []graph.GraphItem { return w.GetGraphItems(w.Graph.IsStorage) },
		tools.DATA_RESOURCE:       func() []graph.GraphItem { return w.GetGraphItems(w.Graph.IsData) },
		tools.COMPUTE_RESOURCE:    func() []graph.GraphItem { return w.GetGraphItems(w.Graph.IsCompute) },
		tools.PROCESSING_RESOURCE: func() []graph.GraphItem { return w.GetGraphItems(w.Graph.IsProcessing) },
		tools.WORKFLOW_RESOURCE:   func() []graph.GraphItem { return w.GetGraphItems(w.Graph.IsWorkflow) },
	}

	for dt, meth := range dtMethodMap {
		res[dt] = make(map[string][]string)
		items := meth()
		for _, i := range items {
			_, r := i.GetResource()
			rId := r.GetID()
			res[dt][rId] = append(res[dt][rId], i.ID)
		}
	}

	return res
}

func plan[T resources.ResourceInterface](
	dt tools.DataType, wf *Workflow, priceds map[tools.DataType]map[string]pricing.PricedItemITF, request *tools.APIRequest,
	f func(graph.GraphItem) bool, start func(resources.ResourceInterface, pricing.PricedItemITF) (time.Time, float64), end func(time.Time, float64) *time.Time) ([]T, map[tools.DataType]map[string]pricing.PricedItemITF, error) {
	resources := []T{}
	for _, item := range wf.GetGraphItems(f) {
		if priceds[dt] == nil {
			priceds[dt] = map[string]pricing.PricedItemITF{}
		}
		dt, realItem := item.GetResource()
		if realItem == nil {
			return resources, priceds, errors.New("could not load the processing resource")
		}
		priced := realItem.ConvertToPricedResource(dt, request)
		// Should be commented once the Pricing selection feature has been implemented, related to the commit d35ad440fa77763ec7f49ab34a85e47e75581b61
		// if priced.SelectPricing() == nil {
		// 	return resources, priceds, errors.New("no pricings are selected... can't proceed")
		// }
		started, duration := start(realItem, priced)
		priced.SetLocationStart(started)
		if duration >= 0 {
			if e := end(started, duration); e != nil {
				priced.SetLocationEnd(*e)
			}
		}
		if e := end(started, priced.GetExplicitDurationInS()); e != nil {
			priced.SetLocationEnd(*e)
		}
		resources = append(resources, realItem.(T))
		priceds[dt][item.ID] = priced
	}
	return resources, priceds, nil
}