package order import ( "errors" "fmt" "sync" "time" "cloud.o-forge.io/core/oc-lib/dbs" "cloud.o-forge.io/core/oc-lib/models/booking" "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/purchase_resource" "cloud.o-forge.io/core/oc-lib/models/utils" "cloud.o-forge.io/core/oc-lib/models/workflow_execution" "cloud.o-forge.io/core/oc-lib/tools" ) /* * Booking is a struct that represents a booking */ type OrderStatus = int const ( DRAFT OrderStatus = iota PENDING CANCELLED PARTIAL PAID DISPUTED OVERDUE REFUND ) type Order struct { utils.AbstractObject OrderBy string `json:"order_by" bson:"order_by" validate:"required"` WorkflowExecutionIDs []string `json:"workflow_execution_ids" bson:"workflow_execution_ids" validate:"required"` Status OrderStatus `json:"status" bson:"status" default:"0"` SubOrders map[string]*PeerOrder `json:"sub_orders" bson:"sub_orders"` Total float64 `json:"total" bson:"total" validate:"required"` } func (r *Order) StoreDraftDefault() { r.IsDraft = true } func (r *Order) CanUpdate(set utils.DBObject) (bool, utils.DBObject) { if !r.IsDraft && r.Status != set.(*Order).Status { return true, &Order{Status: set.(*Order).Status} // only state can be updated } return r.IsDraft, set } func (r *Order) CanDelete() bool { return r.IsDraft // only draft order can be deleted } func (o *Order) DraftOrder(scheduler *workflow_execution.WorkflowSchedule, request *tools.APIRequest) error { // set the draft order from the model if err := o.draftStoreFromModel(scheduler, request); err != nil { return err } return nil } func (o *Order) Pay(scheduler *workflow_execution.WorkflowSchedule, request *tools.APIRequest) error { if _, err := o.draftBookOrder(scheduler, request); err != nil { return err } o.Status = PENDING _, code, err := o.GetAccessor(request).UpdateOne(o, o.GetID()) if code != 200 || err != nil { return errors.New("could not update the order" + fmt.Sprintf("%v", err)) } if err := o.pay(request); err != nil { // pay the order return err } else { o.IsDraft = false } for _, exec := range scheduler.WorkflowExecutions { exec.IsDraft = false _, code, err := utils.GenericUpdateOne(exec, exec.GetID(), workflow_execution.NewAccessor(request), &workflow_execution.WorkflowExecutions{}) if code != 200 || err != nil { return errors.New("could not update the workflow execution" + fmt.Sprintf("%v", err)) } } _, code, err = o.GetAccessor(request).UpdateOne(o, o.GetID()) if code != 200 || err != nil { return errors.New("could not update the order" + fmt.Sprintf("%v", err)) } /* TODO : TEMPORARY SET BOOKINGS TO UNDRAFT TO AVOID DELETION BUT NEXT ONLY WHO IS PAYED WILL BE ALLOWED TO CHANGE IT */ return nil } func (o *Order) draftStoreFromModel(scheduler *workflow_execution.WorkflowSchedule, request *tools.APIRequest) error { if request == nil { return errors.New("no request found") } if scheduler.Workflow.Graph == nil { // if the workflow has no graph, return an error return errors.New("no graph found") } o.SetName() o.IsDraft = true o.OrderBy = request.Username o.WorkflowExecutionIDs = []string{} // create an array of ids for _, exec := range scheduler.WorkflowExecutions { o.WorkflowExecutionIDs = append(o.WorkflowExecutionIDs, exec.GetID()) } // set the name of the order resourcesByPeer := map[string][]pricing.PricedItemITF{} // create a map of resources by peer processings := scheduler.Workflow.GetPricedItem(scheduler.Workflow.IsProcessing) // get the processing items datas := scheduler.Workflow.GetPricedItem(scheduler.Workflow.IsData) // get the data items storages := scheduler.Workflow.GetPricedItem(scheduler.Workflow.IsStorage) // get the storage items workflows := scheduler.Workflow.GetPricedItem(scheduler.Workflow.IsWorkflow) // get the workflow items for _, items := range []map[string]pricing.PricedItemITF{processings, datas, storages, workflows} { for _, item := range items { if _, ok := resourcesByPeer[item.GetCreatorID()]; !ok { resourcesByPeer[item.GetCreatorID()] = []pricing.PricedItemITF{} } resourcesByPeer[item.GetCreatorID()] = append(resourcesByPeer[item.GetCreatorID()], item) } } for peerID, resources := range resourcesByPeer { peerOrder := &PeerOrder{ Status: DRAFT, PeerID: peerID, } peerOrder.GenerateID() for _, resource := range resources { peerOrder.AddItem(resource, len(scheduler.WorkflowExecutions)) // TODO SPECIALS REF ADDITIONALS NOTES } o.SubOrders[peerOrder.GetID()] = peerOrder } // search an order with same user name and same session id err := o.sumUpBill(request) if err != nil { return err } // should store the order res, code, err := o.GetAccessor(request).Search(&dbs.Filters{ And: map[string][]dbs.Filter{ "order_by": {{Operator: dbs.EQUAL.String(), Value: request.Username}}, }, }, "", o.IsDraft) if code != 200 || err != nil { return errors.New("could not search the order" + fmt.Sprintf("%v", err)) } if len(res) > 0 { _, code, err := utils.GenericUpdateOne(o, res[0].GetID(), o.GetAccessor(request), o) if code != 200 || err != nil { return errors.New("could not update the order" + fmt.Sprintf("%v", err)) } } else { _, code, err := utils.GenericStoreOne(o, o.GetAccessor(request)) if code != 200 || err != nil { return errors.New("could not store the order" + fmt.Sprintf("%v", err)) } } return nil } func (o *Order) draftBookOrder(scheduler *workflow_execution.WorkflowSchedule, request *tools.APIRequest) ([]*booking.Booking, error) { draftedBookings := []*booking.Booking{} if request == nil { return draftedBookings, errors.New("no request found") } for _, exec := range scheduler.WorkflowExecutions { bookings := exec.Book(scheduler.Workflow) for _, booking := range bookings { _, err := (&peer.Peer{}).LaunchPeerExecution(booking.DestPeerID, "", tools.BOOKING, tools.POST, booking.Serialize(booking), request.Caller) if err != nil { return draftedBookings, errors.New("could not launch the peer execution : " + fmt.Sprintf("%v", err)) } draftedBookings = append(draftedBookings, booking) } } return draftedBookings, nil } func (o *Order) Quantity() int { return len(o.WorkflowExecutionIDs) } func (d *Order) SetName() { d.Name = d.UUID + "_order_" + "_" + time.Now().UTC().Format("2006-01-02T15:04:05") } func (d *Order) GetAccessor(request *tools.APIRequest) utils.Accessor { return NewAccessor(request) // Create a new instance of the accessor } func (d *Order) sumUpBill(request *tools.APIRequest) error { for _, b := range d.SubOrders { err := b.SumUpBill(request) if err != nil { return err } d.Total += b.Total } return nil } // TO FINISH func (d *Order) pay(request *tools.APIRequest) error { responses := make(chan *PeerOrder, len(d.SubOrders)) var wg *sync.WaitGroup wg.Add(len(d.SubOrders)) for _, b := range d.SubOrders { go b.Pay(request, responses, wg) } wg.Wait() errs := "" gotAnUnpaid := false count := 0 for range responses { res := <-responses count++ if res != nil { if res.Error != "" { errs += res.Error } if res.Status != PAID { gotAnUnpaid = true } d.Status = PARTIAL d.SubOrders[res.GetID()] = res if count == len(d.SubOrders) && !gotAnUnpaid { d.Status = PAID } } } if errs != "" { return errors.New(errs) } return nil } type PeerOrder struct { utils.AbstractObject Error string `json:"error,omitempty" bson:"error,omitempty"` PeerID string `json:"peer_id,omitempty" bson:"peer_id,omitempty"` Status OrderStatus `json:"status" bson:"status" default:"0"` BillingAddress string `json:"billing_address,omitempty" bson:"billing_address,omitempty"` Items []*PeerItemOrder `json:"items,omitempty" bson:"items,omitempty"` Total float64 `json:"total,omitempty" bson:"total,omitempty"` } func (d *PeerOrder) Pay(request *tools.APIRequest, response chan *PeerOrder, wg *sync.WaitGroup) { d.Status = PENDING go func() { // DO SOMETHING TO PAY ON BLOCKCHAIN OR WHATEVER ON RETURN UPDATE STATUS d.Status = PAID // TO REMOVE LATER IT'S A MOCK if d.Status == PAID { for _, b := range d.Items { if !b.Item.IsPurchased(request) { continue } accessor := purchase_resource.NewAccessor(request) accessor.StoreOne(&purchase_resource.PurchaseResource{ ResourceID: b.Item.GetID(), ResourceType: b.Item.GetType(), EndDate: b.Item.GetLocationEnd(), }) } } if d.Status != PENDING { response <- d } wg.Done() }() } func (d *PeerOrder) SumUpBill(request *tools.APIRequest) error { for _, b := range d.Items { tot, err := b.GetPrice(request) // missing something if err != nil { return err } d.Total += tot } return nil } func (d *PeerOrder) AddItem(item pricing.PricedItemITF, quantity int) { d.Items = append(d.Items, &PeerItemOrder{ Quantity: quantity, Item: item, }) } func (d *PeerOrder) SetName() { d.Name = d.UUID + "_order_" + d.PeerID + "_" + time.Now().UTC().Format("2006-01-02T15:04:05") } type PeerItemOrder struct { Quantity int `json:"quantity,omitempty" bson:"quantity,omitempty"` Purchase purchase_resource.PurchaseResource `json:"purchase,omitempty" bson:"purchase,omitempty"` Item pricing.PricedItemITF `json:"item,omitempty" bson:"item,omitempty"` } func (d *PeerItemOrder) GetPrice(request *tools.APIRequest) (float64, error) { accessor := purchase_resource.NewAccessor(request) search, code, _ := accessor.Search(&dbs.Filters{ And: map[string][]dbs.Filter{ "resource_id": {{Operator: dbs.EQUAL.String(), Value: d.Item.GetID()}}, }, }, "", d.Purchase.IsDraft) if code == 200 && len(search) > 0 { for _, s := range search { if s.(*purchase_resource.PurchaseResource).EndDate == nil || time.Now().UTC().After(*s.(*purchase_resource.PurchaseResource).EndDate) { return 0, nil } } } p, err := d.Item.GetPrice(request) if err != nil { return 0, err } return p * float64(d.Quantity), nil } // SHOULD SET A BUYING STATUS WHEN PAYMENT IS VALIDATED