Compute distance between airport and user

README.md

@@ -20,3 +20,4 @@
 - BANK_ENDPOINT
 - BANK_CALLBACK
 - BANK_TOKEN
+- GEODISTANCE_API

bpmn/acmesky.bpmn

@@ -19,8 +19,8 @@
     <bpmn:messageFlow id="Flow_1mf9ie3" sourceRef="Participant_Rent" targetRef="TM_Ask_For_Rent" />
     <bpmn:messageFlow id="Flow_020v9in" sourceRef="TM_Find_Nearest_Available_Rent_Company" targetRef="Participant_Geolocation" />
     <bpmn:messageFlow id="Flow_057on4g" sourceRef="Participant_Geolocation" targetRef="TM_Find_Nearest_Available_Rent_Company" />
-    <bpmn:messageFlow id="Flow_01fwi9s" sourceRef="TM_Computer_Distance_User_Airport" targetRef="Participant_Geolocation" />
-    <bpmn:messageFlow id="Flow_11v886u" sourceRef="Participant_Geolocation" targetRef="TM_Computer_Distance_User_Airport" />
+    <bpmn:messageFlow id="Flow_01fwi9s" sourceRef="TM_Compute_Distance_User_Airport" targetRef="Participant_Geolocation" />
+    <bpmn:messageFlow id="Flow_11v886u" sourceRef="Participant_Geolocation" targetRef="TM_Compute_Distance_User_Airport" />
     <bpmn:messageFlow id="Flow_0eonj29" sourceRef="Participant_Bank" targetRef="Event_0wdh6he" />
     <bpmn:messageFlow id="Flow_0055psj" sourceRef="TM_Error_On_Book_Journey" targetRef="CM_Received_Bank_Error" />
     <bpmn:messageFlow id="Flow_00yxmjl" sourceRef="TM_Error_On_Check_Offer" targetRef="CM_Received_Bank_Error" />
@@ -327,7 +327,7 @@
       <bpmn:endEvent id="End_Book_Journey">
         <bpmn:incoming>Flow_0kuzdys</bpmn:incoming>
       </bpmn:endEvent>
-      <bpmn:exclusiveGateway id="EG_Payment_Success" name="Payment success?">
+      <bpmn:exclusiveGateway id="EG_Payment_Success" name="Payment success?" default="Flow_1xgd3n1">
         <bpmn:incoming>Flow_1kig1jb</bpmn:incoming>
         <bpmn:outgoing>Flow_1xgd3n1</bpmn:outgoing>
         <bpmn:outgoing>Flow_01oh7g4</bpmn:outgoing>
@@ -349,9 +349,7 @@
       <bpmn:sequenceFlow id="Flow_0kpaa08" sourceRef="PG_Send_Payment_Link_Start" targetRef="Event_0wdh6he" />
       <bpmn:sequenceFlow id="Flow_1kig1jb" sourceRef="Event_0wdh6he" targetRef="EG_Payment_Success" />
       <bpmn:sequenceFlow id="Flow_0kuzdys" sourceRef="PG_Send_Payment_Link_End" targetRef="End_Book_Journey" />
-      <bpmn:sequenceFlow id="Flow_1xgd3n1" name="No" sourceRef="EG_Payment_Success" targetRef="EE_Payment">
-        <bpmn:conditionExpression xsi:type="bpmn:tFormalExpression">=payment_status != "Ok"</bpmn:conditionExpression>
-      </bpmn:sequenceFlow>
+      <bpmn:sequenceFlow id="Flow_1xgd3n1" name="No" sourceRef="EG_Payment_Success" targetRef="EE_Payment" />
       <bpmn:sequenceFlow id="Flow_01oh7g4" name="Yes" sourceRef="EG_Payment_Success" targetRef="PG_Send_Payment_Link_End">
         <bpmn:conditionExpression xsi:type="bpmn:tFormalExpression">=payment_status = "OK"</bpmn:conditionExpression>
       </bpmn:sequenceFlow>
@@ -422,16 +420,16 @@
           <bpmn:timeDuration xsi:type="bpmn:tFormalExpression">=PT30S</bpmn:timeDuration>
         </bpmn:timerEventDefinition>
       </bpmn:intermediateCatchEvent>
-      <bpmn:sendTask id="TM_Computer_Distance_User_Airport" name="Compute distance user-airport">
+      <bpmn:sendTask id="TM_Compute_Distance_User_Airport" name="Compute distance user-airport">
         <bpmn:extensionElements>
-          <zeebe:taskDefinition type="TM_Computer_Distance_User_Airport" />
+          <zeebe:taskDefinition type="TM_Compute_Distance_User_Airport" />
         </bpmn:extensionElements>
         <bpmn:incoming>Flow_1ajjpq1</bpmn:incoming>
         <bpmn:outgoing>Flow_08h7gb2</bpmn:outgoing>
       </bpmn:sendTask>
-      <bpmn:sequenceFlow id="Flow_08h7gb2" sourceRef="TM_Computer_Distance_User_Airport" targetRef="EG_Distance_Less_Than_30km" />
+      <bpmn:sequenceFlow id="Flow_08h7gb2" sourceRef="TM_Compute_Distance_User_Airport" targetRef="EG_Distance_Less_Than_30km" />
       <bpmn:sequenceFlow id="Flow_1i8vff5" sourceRef="TE_30s_Distance_Airport" targetRef="EG_Loop_Rent" />
-      <bpmn:boundaryEvent id="Event_078zn83" attachedToRef="TM_Computer_Distance_User_Airport">
+      <bpmn:boundaryEvent id="Event_078zn83" attachedToRef="TM_Compute_Distance_User_Airport">
         <bpmn:outgoing>Flow_05e47ko</bpmn:outgoing>
         <bpmn:errorEventDefinition id="ErrorEventDefinition_104ycy0" />
       </bpmn:boundaryEvent>
@@ -441,7 +439,7 @@
         <bpmn:incoming>Flow_0ye5jwj</bpmn:incoming>
         <bpmn:outgoing>Flow_1ajjpq1</bpmn:outgoing>
       </bpmn:exclusiveGateway>
-      <bpmn:sequenceFlow id="Flow_1ajjpq1" sourceRef="EG_Loop_Rent" targetRef="TM_Computer_Distance_User_Airport" />
+      <bpmn:sequenceFlow id="Flow_1ajjpq1" sourceRef="EG_Loop_Rent" targetRef="TM_Compute_Distance_User_Airport" />
       <bpmn:startEvent id="Start_Rent_Service">
         <bpmn:outgoing>Flow_0ye5jwj</bpmn:outgoing>
       </bpmn:startEvent>
@@ -1133,21 +1131,6 @@
           <dc:Bounds x="256" y="1438" width="87" height="14" />
         </bpmndi:BPMNLabel>
       </bpmndi:BPMNShape>
-      <bpmndi:BPMNShape id="Event_1vdq7fa_di" bpmnElement="Event_1vdq7fa" bioc:stroke="#831311" bioc:fill="#ffcdd2" color:background-color="#ffcdd2" color:border-color="#831311">
-        <dc:Bounds x="779" y="632" width="36" height="36" />
-      </bpmndi:BPMNShape>
-      <bpmndi:BPMNShape id="Activity_15gzw5g_di" bpmnElement="Activity_0caef9i" bioc:stroke="#831311" bioc:fill="#ffcdd2" color:background-color="#ffcdd2" color:border-color="#831311">
-        <dc:Bounds x="870" y="610" width="100" height="80" />
-      </bpmndi:BPMNShape>
-      <bpmndi:BPMNShape id="Event_0em1zuc_di" bpmnElement="Event_0em1zuc" bioc:stroke="#831311" bioc:fill="#ffcdd2" color:background-color="#ffcdd2" color:border-color="#831311">
-        <dc:Bounds x="1032" y="632" width="36" height="36" />
-      </bpmndi:BPMNShape>
-      <bpmndi:BPMNShape id="DataStoreReference_1bsvx5n_di" bpmnElement="DataStoreReference_1bsvx5n">
-        <dc:Bounds x="385" y="1025" width="50" height="50" />
-        <bpmndi:BPMNLabel>
-          <dc:Bounds x="387" y="1003" width="46" height="14" />
-        </bpmndi:BPMNLabel>
-      </bpmndi:BPMNShape>
       <bpmndi:BPMNShape id="BPMNShape_16hdy8m" bpmnElement="Activity_Foreach_Journey" isExpanded="true">
         <dc:Bounds x="800" y="1025" width="520" height="250" />
         <bpmndi:BPMNLabel />
@@ -1383,7 +1366,7 @@
           <dc:Bounds x="5240" y="694" width="57" height="14" />
         </bpmndi:BPMNLabel>
       </bpmndi:BPMNShape>
-      <bpmndi:BPMNShape id="Activity_0k14vde_di" bpmnElement="TM_Computer_Distance_User_Airport">
+      <bpmndi:BPMNShape id="Activity_0k14vde_di" bpmnElement="TM_Compute_Distance_User_Airport">
         <dc:Bounds x="5228" y="770" width="100" height="80" />
         <bpmndi:BPMNLabel />
       </bpmndi:BPMNShape>
@@ -1569,6 +1552,21 @@
           <dc:Bounds x="1668" y="835" width="85" height="14" />
         </bpmndi:BPMNLabel>
       </bpmndi:BPMNShape>
+      <bpmndi:BPMNShape id="Event_1vdq7fa_di" bpmnElement="Event_1vdq7fa" bioc:stroke="#831311" bioc:fill="#ffcdd2" color:background-color="#ffcdd2" color:border-color="#831311">
+        <dc:Bounds x="779" y="632" width="36" height="36" />
+      </bpmndi:BPMNShape>
+      <bpmndi:BPMNShape id="Activity_15gzw5g_di" bpmnElement="Activity_0caef9i" bioc:stroke="#831311" bioc:fill="#ffcdd2" color:background-color="#ffcdd2" color:border-color="#831311">
+        <dc:Bounds x="870" y="610" width="100" height="80" />
+      </bpmndi:BPMNShape>
+      <bpmndi:BPMNShape id="Event_0em1zuc_di" bpmnElement="Event_0em1zuc" bioc:stroke="#831311" bioc:fill="#ffcdd2" color:background-color="#ffcdd2" color:border-color="#831311">
+        <dc:Bounds x="1032" y="632" width="36" height="36" />
+      </bpmndi:BPMNShape>
+      <bpmndi:BPMNShape id="DataStoreReference_1bsvx5n_di" bpmnElement="DataStoreReference_1bsvx5n">
+        <dc:Bounds x="385" y="1025" width="50" height="50" />
+        <bpmndi:BPMNLabel>
+          <dc:Bounds x="387" y="1003" width="46" height="14" />
+        </bpmndi:BPMNLabel>
+      </bpmndi:BPMNShape>
       <bpmndi:BPMNShape id="Event_0ieguxi_di" bpmnElement="Event_0jl8r24">
         <dc:Bounds x="4050" y="701" width="36" height="36" />
       </bpmndi:BPMNShape>
@@ -1963,6 +1961,11 @@
         <di:waypoint x="700" y="1220" />
         <di:waypoint x="700" y="1190" />
       </bpmndi:BPMNEdge>
+      <bpmndi:BPMNEdge id="DataInputAssociation_0zzoevr_di" bpmnElement="DataInputAssociation_0zzoevr">
+        <di:waypoint x="435" y="1050" />
+        <di:waypoint x="700" y="1050" />
+        <di:waypoint x="700" y="1110" />
+      </bpmndi:BPMNEdge>
       <bpmndi:BPMNEdge id="DataInputAssociation_0tb22zr_di" bpmnElement="DataInputAssociation_0tb22zr">
         <di:waypoint x="1001" y="505" />
         <di:waypoint x="1010" y="527" />
@@ -2096,11 +2099,6 @@
         <di:waypoint x="1150" y="1110" />
         <di:waypoint x="1150" y="498" />
       </bpmndi:BPMNEdge>
-      <bpmndi:BPMNEdge id="DataInputAssociation_0zzoevr_di" bpmnElement="DataInputAssociation_0zzoevr">
-        <di:waypoint x="435" y="1050" />
-        <di:waypoint x="700" y="1050" />
-        <di:waypoint x="700" y="1110" />
-      </bpmndi:BPMNEdge>
     </bpmndi:BPMNPlane>
   </bpmndi:BPMNDiagram>
 </bpmn:definitions>

go.mod

@@ -3,12 +3,14 @@ module github.com/acme-sky/workers
 go 1.22.0
 
 require (
+	github.com/acme-sky/geodistance-api v0.0.0-20240430172918-82d1991cdf9f
 	github.com/camunda/zeebe/clients/go/v8 v8.5.0
 	github.com/charmbracelet/log v0.4.0
 	github.com/getsentry/sentry-go v0.27.0
 	github.com/knadh/koanf/providers/env v0.1.0
 	github.com/knadh/koanf/v2 v2.1.1
 	github.com/rabbitmq/amqp091-go v1.9.0
+	google.golang.org/grpc v1.62.1
 	gorm.io/driver/postgres v1.5.7
 	gorm.io/gorm v1.25.10
 )
@@ -51,7 +53,6 @@ require (
 	golang.org/x/text v0.14.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240123012728-ef4313101c80 // indirect
-	google.golang.org/grpc v1.62.1 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -1,3 +1,5 @@
+github.com/acme-sky/geodistance-api v0.0.0-20240430172918-82d1991cdf9f h1:cdidp+mqYQLspkf1oPNi76CHE9xA3NJCSBgF/on/gxc=
+github.com/acme-sky/geodistance-api v0.0.0-20240430172918-82d1991cdf9f/go.mod h1:l2vd6NctcHlym6EsoYxwYMFSjTbEqz+HeEXVA/Z76/U=
 github.com/asaskevich/govalidator v0.0.0-20200108200545-475eaeb16496 h1:zV3ejI06GQ59hwDQAvmK1qxOQGB3WuVTRoY0okPTAv0=
 github.com/asaskevich/govalidator v0.0.0-20200108200545-475eaeb16496/go.mod h1:oGkLhpf+kjZl6xBf758TQhh5XrAeiJv/7FRz/2spLIg=
 github.com/aymanbagabas/go-osc52/v2 v2.0.1 h1:HwpRHbFMcZLEVr42D4p7XBqjyuxQH5SMiErDT4WkJ2k=

internal/handlers/acmesky/st_prepare_offer.go

@@ -31,7 +31,7 @@ func STPrepareOffer(client worker.JobClient, job entities.Job) {
 	db, _ := db.GetDb()
 	var journey models.Journey
 
-	if err := db.Where("id = ?", int(journeys[index].(float64))).Preload("User").First(&journey).Error; err != nil {
+	if err := db.Where("id = ?", int(journeys[index].(float64))).Preload("Flight1").Preload("Flight2").Preload("User").First(&journey).Error; err != nil {
 		log.Errorf("[%s] [%d] Journey not found", job.Type, jobKey)
 		acmejob.FailJob(client, job)
 		return

internal/handlers/acmesky/tm_compute_distance_user_airport.go

@@ -2,14 +2,25 @@ package handlers
 
 import (
 	"context"
+	"fmt"
+	"time"
+
 	"github.com/charmbracelet/log"
 
+	pb "github.com/acme-sky/geodistance-api/pkg/distance/proto"
+	"github.com/acme-sky/workers/internal/config"
+	"github.com/acme-sky/workers/internal/db"
+	"github.com/acme-sky/workers/internal/http"
 	acmejob "github.com/acme-sky/workers/internal/job"
+	"github.com/acme-sky/workers/internal/models"
 	"github.com/camunda/zeebe/clients/go/v8/pkg/entities"
 	"github.com/camunda/zeebe/clients/go/v8/pkg/worker"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 )
 
-func TMComputerDistanceUserAirport(client worker.JobClient, job entities.Job) {
+// Task used to find distance between departaure airport and user.
+func TMComputeDistanceUserAirport(client worker.JobClient, job entities.Job) {
 	jobKey := job.GetKey()
 
 	variables, err := job.GetVariablesAsMap()
@@ -18,17 +29,69 @@ func TMComputerDistanceUserAirport(client worker.JobClient, job entities.Job) {
 		return
 	}
 
-	variables["distance"] = 28
+	db, _ := db.GetDb()
+	var offer models.Offer
 
-	request, err := client.NewCompleteJobCommand().JobKey(jobKey).VariablesFromMap(variables)
+	if err := db.Where("id = ?", int(variables["offer_id"].(float64))).Preload("User").Preload("Journey").Preload("Journey.Flight1").First(&offer).Error; err != nil {
+		log.Errorf("[%s] [%d] Error on getting offer %s", job.Type, jobKey, err.Error())
+	}
+
+	if offer.User.Address == nil {
+		log.Warnf("[%s] [%d] User does not have an address", job.Type, jobKey)
+		acmejob.FailJob(client, job)
+		return
+	}
+
+	conf, _ := config.GetConfig()
+
+	conn, err := grpc.Dial(conf.String("geodistance.api"), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		log.Errorf("[%s] [%d] Can't connect to Geodistance url: %s", job.Type, jobKey, err.Error())
+		acmejob.FailJob(client, job)
+		return
+	}
+
+	defer conn.Close()
+	c := pb.NewDistanceClient(conn)
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	userGeometry, err := c.FindGeometry(ctx, &pb.AddressRequest{
+		Address: *offer.User.Address,
+	})
 	if err != nil {
+		log.Errorf("[%s] [%d] Can't find geometry for user: %s", job.Type, jobKey, err.Error())
 		acmejob.FailJob(client, job)
 		return
 	}
 
-	log.Debug("Processing data:", variables)
+	endpoint := fmt.Sprintf("%s/airports/code/%s/", offer.Journey.Flight1.Airline, offer.Journey.Flight1.DepartaureAirport)
+	airport, err := http.GetAirportInfo(endpoint)
+	if err != nil {
+		log.Errorf("[%s] [%d] Can't find info for departaure airport: %s", job.Type, jobKey, err.Error())
+		acmejob.FailJob(client, job)
+		return
+	}
+
+	distance, err := c.FindDistance(ctx, &pb.DistanceRequest{
+		Origin:      &pb.MapPosition{Latitude: userGeometry.Latitude, Longitude: userGeometry.Longitude},
+		Destination: &pb.MapPosition{Latitude: airport.Latitude, Longitude: airport.Longitude},
+	})
+	if err != nil {
+		log.Errorf("[%s] [%d] Can't find distance: %s", job.Type, jobKey, err.Error())
+		acmejob.FailJob(client, job)
+		return
+	}
+	variables["distance"] = distance.GetDistance() / 1000
+	log.Info("[%s] [%d] Found a distance of: %d km", job.Type, jobKey, variables["distance"])
+
+	request, err := client.NewCompleteJobCommand().JobKey(jobKey).VariablesFromMap(variables)
+	if err != nil {
+		acmejob.FailJob(client, job)
+		return
+	}
 
-	ctx := context.Background()
 	_, err = request.Send(ctx)
 	if err != nil {
 		acmejob.FailJob(client, job)

internal/handlers/acmesky/tm_journey.go

@@ -9,6 +9,7 @@ import (
 	"github.com/camunda/zeebe/clients/go/v8/pkg/worker"
 )
 
+// Make a message request to the user for "journey receipt"
 func TMJourney(client worker.JobClient, job entities.Job) {
 	jobKey := job.GetKey()
 
@@ -24,8 +25,6 @@ func TMJourney(client worker.JobClient, job entities.Job) {
 		return
 	}
 
-	log.Debug("Processing data:", variables)
-
 	ctx := context.Background()
 	_, err = request.Send(ctx)
 	if err != nil {

internal/handlers/acmesky/tm_journey_and_rent.go

@@ -9,6 +9,7 @@ import (
 	"github.com/camunda/zeebe/clients/go/v8/pkg/worker"
 )
 
+// Make a message request to the user for "journey and rent receipts"
 func TMJourneyAndRent(client worker.JobClient, job entities.Job) {
 	jobKey := job.GetKey()
 
@@ -24,8 +25,6 @@ func TMJourneyAndRent(client worker.JobClient, job entities.Job) {
 		return
 	}
 
-	log.Debug("Processing data:", variables)
-
 	ctx := context.Background()
 	_, err = request.Send(ctx)
 	if err != nil {

internal/handlers/acmesky/tm_journey_rent_error.go

@@ -9,6 +9,7 @@ import (
 	"github.com/camunda/zeebe/clients/go/v8/pkg/worker"
 )
 
+// Make a message request to the user for "journey receipt but rent error"
 func TMJourneyRentError(client worker.JobClient, job entities.Job) {
 	jobKey := job.GetKey()
 
@@ -24,8 +25,6 @@ func TMJourneyRentError(client worker.JobClient, job entities.Job) {
 		return
 	}
 
-	log.Debug("Processing data:", variables)
-
 	ctx := context.Background()
 	_, err = request.Send(ctx)
 	if err != nil {

internal/handlers/acmesky/tm_send_payment_link.go

@@ -19,7 +19,7 @@ func TMSendPaymentLink(client worker.JobClient, job entities.Job) {
 		return
 	}
 
-	if variables["payment_link"] != nil {
+	if variables["payment_link"] == nil {
 		log.Errorf("[%s] [%d] `payment_link` is a not a valid variable to send", job.Type, jobKey)
 		acmejob.FailJob(client, job)
 		return

internal/http/http.go

@@ -40,6 +40,15 @@ type PaymentResponseBody struct {
 	CreatedAt string `json:"created_at"`
 }
 
+type AirportInfoResponseBody struct {
+	Id        uint    `json:"id"`
+	Name      string  `json:"name"`
+	Code      string  `json:"code"`
+	Location  string  `json:"location"`
+	Latitude  float32 `json:"latitude"`
+	Longitude float32 `json:"longitude"`
+}
+
 // Make a new request to an endpoint with a `body` and returns a response body
 // or an error.
 func MakeRequest(endpoint string, body map[string]interface{}) (*ResponseBody, error) {
@@ -194,3 +203,27 @@ func NewPaymentRequest(endpoint string, body map[string]interface{}, auth string
 
 	return &responseBody, nil
 }
+
+// Make a new request to an endpoint to get info about an airport.
+func GetAirportInfo(endpoint string) (*AirportInfoResponseBody, error) {
+	res, err := http.Get(endpoint)
+	if err != nil {
+		return nil, err
+	}
+
+	resBody, err := io.ReadAll(res.Body)
+	if err != nil {
+		return nil, errors.New(fmt.Sprintf("Could not read response body: %s", err.Error()))
+	}
+
+	if res.StatusCode != 200 {
+		return nil, errors.New(fmt.Sprintf("HTTP request returned a status %d and response `%s`", res.StatusCode, resBody))
+	}
+
+	var responseBody AirportInfoResponseBody
+	if err := json.Unmarshal(resBody, &responseBody); err != nil {
+		return nil, errors.New(fmt.Sprintf("Could not unmarshal response body: %s", err))
+	}
+
+	return &responseBody, nil
+}

main.go

@@ -89,7 +89,7 @@ func main() {
 		{Name: "ST_Offer_Still_Valid", Handler: acmeskyHandlers.STOfferStillValid},
 		{Name: "TM_Error_On_Book_Journey", Handler: acmeskyHandlers.TMErrorOnBookJourney, Message: &acmejob.MessageCommand{Name: "CM_Received_Bank_Error", CorrelationKey: "0"}},
 		{Name: "TM_Journey", Handler: acmeskyHandlers.TMJourney, Message: &acmejob.MessageCommand{Name: "CM_Journey", CorrelationKey: "0"}},
-		{Name: "TM_Computer_Distance_User_Airport", Handler: acmeskyHandlers.TMComputerDistanceUserAirport},
+		{Name: "TM_Compute_Distance_User_Airport", Handler: acmeskyHandlers.TMComputeDistanceUserAirport},
 		{Name: "TM_Find_Nearest_Available_Rent_Company", Handler: acmeskyHandlers.TMFindNearestAvailableRentCompany},
 		{Name: "TM_Ask_For_Rent", Handler: acmeskyHandlers.TMAskForRent},
 		{Name: "TM_Journey_And_Rent", Handler: acmeskyHandlers.TMJourneyAndRent, Message: &acmejob.MessageCommand{Name: "CM_Journey_And_Rent", CorrelationKey: "0"}},