coopgo-platform
/
carpool-service
7
0
Fork 0
Code Issues Pull Requests Actions Projects Releases Wiki Activity

Add PostgreSQL database option and more booking flow functionalities

This commit is contained in:
Arnaud Delcasse 2023-05-08 01:30:50 +02:00
parent e2e6759dc0
commit f5938d23df
10 changed files with 2318 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
handler/bookings.go Normal file
View File

@ -0,0 +1,101 @@
package handler
import (
"errors"
"fmt"
"time"
"git.coopgo.io/coopgo-platform/carpool-service/internal"
"git.coopgo.io/coopgo-platform/carpool-service/interoperability/ocss"
"github.com/google/uuid"
"github.com/rs/zerolog/log"
)
// Book handles the booking flow
func (h *CarpoolServiceHandler) Book(booking ocss.Booking) (*internal.Booking, error) {
log.Debug().Any("booking", booking).Msg("handler - Book")
log.Debug().Str("passengerPickupDate", booking.PassengerPickupDate.ToTime().Format(time.RFC3339)).Msg("handler - Book")
futureBooking := internal.Booking{}
roles := []string{}
if booking.Driver.Operator == h.InternalOperatorID {
roles = append(roles, "driver")
previous_search_result, err := h.Storage.GetRouteSchedule(booking.DriverJourneyID)
if err == nil {
futureBooking.DriverRoute = previous_search_result.Route
}
}
if booking.Passenger.Operator == h.InternalOperatorID {
roles = append(roles, "passenger")
previous_search_result, err := h.Storage.GetRouteSchedule(booking.PassengerJourneyID)
if err != nil {
log.Error().Err(err).Msg("could not get previous result")
}
if err == nil {
futureBooking.PassengerRoute = previous_search_result.Route
}
}
if len(roles) == 0 {
return nil, fmt.Errorf("couldn't find the right operator id : \"%s\" should be set for driver or passenger", h.InternalOperatorID)
}
if _, err := uuid.Parse(booking.ID); err != nil {
return nil, errors.New("bookingid is not a valid uuid")
}
futureBooking.Booking = booking
err := h.Storage.CreateBooking(futureBooking)
if err != nil {
log.Error().Err(err).Msg("issue creating booking in database")
return nil, err
}
return &futureBooking, nil
}
// GetBooking returns the booking with the given ID
func (h *CarpoolServiceHandler) GetBooking(id string) (*internal.Booking, error) {
booking, err := h.Storage.GetBooking(id)
if err != nil {
log.Error().Err(err).Msg("issue retrieving booking in storage")
return nil, errors.New("booking id not found")
}
return booking, nil
}
// GetUserBookings retrieves all the bookings for a specified user id
func (h *CarpoolServiceHandler) GetUserBookings(user_id string, mindate *time.Time, maxdate *time.Time) ([]internal.Booking, error) {
bookings, err := h.Storage.GetUserBookings(user_id)
if err != nil {
return nil, err
}
results := []internal.Booking{}
for _, b := range bookings {
if mindate != nil {
if b.PassengerPickupDate.ToTime().Before(*mindate) {
continue
}
}
if maxdate != nil {
if b.PassengerPickupDate.ToTime().After(*maxdate) {
continue
}
}
results = append(results, b)
}
return results, nil
}
func (h *CarpoolServiceHandler) UpdateBookingStatus(id string, status ocss.BookingStatus) error {
err := h.Storage.UpdateBookingStatus(id, status.String())
if err != nil {
log.Error().Err(err).Msg("not able to update booking status")
return err
}
return nil
}

15
internal/planned-route-schedules.go Normal file
View File

@ -0,0 +1,15 @@
package internal
import (
"time"
"git.coopgo.io/coopgo-platform/routing-service"
"github.com/paulmach/orb/geojson"
)
type PlannedRouteSchedule struct {
ID string `bson:"_id" json:"id"`
Route *geojson.FeatureCollection
DepartureDate time.Time `bson:"departureDate"`
Itinerary *routing.Route `bson:"itinerary,omitempty"`
}

136
interoperability/ocss/journeys_driver_functions.go Normal file
View File

@ -0,0 +1,136 @@
package ocss
import (
"time"
geojson "github.com/paulmach/orb/geojson"
)
func NewDriverJourney(
id string,
operator string,
driver User,
passengerPickup geojson.Feature,
passengerDrop geojson.Feature,
duration time.Duration,
passengerPickupDate time.Time,
journeyType JourneyScheduleType,
) *DriverJourney {
var pickupAddress *string
if addr := passengerPickup.Properties.MustString("label", ""); addr != "" {
pickupAddress = &addr
}
var dropAddress *string
if addr := passengerDrop.Properties.MustString("label", ""); addr != "" {
dropAddress = &addr
}
return &DriverJourney{
DriverTrip: DriverTrip{
Driver: driver,
Trip: Trip{
Operator: operator,
PassengerPickupLat: passengerPickup.Point().Lat(),
PassengerPickupLng: passengerPickup.Point().Lon(),
PassengerPickupAddress: pickupAddress,
PassengerDropLat: passengerDrop.Point().Lat(),
PassengerDropLng: passengerDrop.Point().Lon(),
PassengerDropAddress: dropAddress,
Duration: duration,
},
},
JourneySchedule: JourneySchedule{
ID: &id,
PassengerPickupDate: OCSSTime(passengerPickupDate),
Type: journeyType,
},
}
}
func (j *DriverJourney) AddDepartureToPickupWalkingDistance(distance int64) *DriverJourney {
j.DepartureToPickupWalkingDistance = &distance
return j
}
func (j *DriverJourney) AddDepartureToPickupWalkingDuration(duration time.Duration) *DriverJourney {
j.DepartureToPickupWalkingDuration = &duration
return j
}
func (j *DriverJourney) AddDepartureToPickupWalkingPolyline(polyline string) *DriverJourney {
j.DepartureToPickupWalkingPolyline = &polyline
return j
}
func (j *DriverJourney) AddDropoffToArrivalWalkingDistance(distance int64) *DriverJourney {
j.DropoffToArrivalWalkingDistance = &distance
return j
}
func (j *DriverJourney) AddDropoffToArrivalWalkingDuration(duration time.Duration) *DriverJourney {
j.DropoffToArrivalWalkingDuration = &duration
return j
}
func (j *DriverJourney) AddDropoffToArrivalWalkingPolyline(polyline string) *DriverJourney {
j.DropoffToArrivalWalkingPolyline = &polyline
return j
}
func (j *DriverJourney) AddCar(car Car) *DriverJourney {
j.Car = &car
return j
}
func (j *DriverJourney) AddDriverDeparture(location geojson.Feature) *DriverJourney {
lat := location.Point().Lat()
lon := location.Point().Lon()
j.DriverDepartureLat = &lat
j.DriverDepartureLng = &lon
if addr := location.Properties.MustString("label", ""); addr != "" {
j.DriverDepartureAddress = &addr
}
return j
}
func (j *DriverJourney) AddDriverArrival(location geojson.Feature) *DriverJourney {
lat := location.Point().Lat()
lon := location.Point().Lon()
j.DriverArrivalLat = &lat
j.DriverArrivalLng = &lon
if addr := location.Properties.MustString("label", ""); addr != "" {
j.DriverArrivalAddress = &addr
}
return j
}
func (j *DriverJourney) AddDistance(distance int64) *DriverJourney {
j.Distance = &distance
return j
}
func (j *DriverJourney) AddJourneyPolyline(polyline string) *DriverJourney {
j.JourneyPolyline = &polyline
return j
}
func (j *DriverJourney) AddPreferences(preferences Preferences) *DriverJourney {
j.Preferences = &preferences
return j
}
func (j *DriverJourney) AddAvailableSeats(seats int64) *DriverJourney {
j.AvailableSteats = &seats
return j
}
func (j *DriverJourney) AddDriverDepartureDate(date time.Time) *DriverJourney {
d := OCSSTime(date)
j.DriverDepartureDate = &d
return j
}
func (j *DriverJourney) AddWebUrl(url string) *DriverJourney {
j.WebUrl = &url
return j
}

105
interoperability/ocss/journeys_passenger_functions.go Normal file
View File

@ -0,0 +1,105 @@
package ocss
import (
"time"
geojson "github.com/paulmach/orb/geojson"
)
func NewPassengerJourney(
id string,
operator string,
driver User,
passengerPickup geojson.Feature,
passengerDrop geojson.Feature,
duration time.Duration,
passengerPickupDate time.Time,
journeyType JourneyScheduleType,
) *PassengerJourney {
var pickupAddress *string
if addr := passengerPickup.Properties.MustString("label", ""); addr != "" {
pickupAddress = &addr
}
var dropAddress *string
if addr := passengerDrop.Properties.MustString("label", ""); addr != "" {
dropAddress = &addr
}
return &PassengerJourney{
PassengerTrip: PassengerTrip{
Passenger: driver,
Trip: Trip{
Operator: operator,
PassengerPickupLat: passengerPickup.Point().Lat(),
PassengerPickupLng: passengerPickup.Point().Lon(),
PassengerPickupAddress: pickupAddress,
PassengerDropLat: passengerDrop.Point().Lat(),
PassengerDropLng: passengerDrop.Point().Lon(),
PassengerDropAddress: dropAddress,
Duration: duration,
},
},
JourneySchedule: JourneySchedule{
ID: &id,
PassengerPickupDate: OCSSTime(passengerPickupDate),
Type: journeyType,
},
}
}
func (j *PassengerJourney) AddDriverDeparture(location geojson.Feature) *PassengerJourney {
lat := location.Point().Lat()
lon := location.Point().Lon()
j.DriverDepartureLat = &lat
j.DriverDepartureLng = &lon
if addr := location.Properties.MustString("label", ""); addr != "" {
j.DriverDepartureAddress = &addr
}
return j
}
func (j *PassengerJourney) AddDriverArrival(location geojson.Feature) *PassengerJourney {
lat := location.Point().Lat()
lon := location.Point().Lon()
j.DriverArrivalLat = &lat
j.DriverArrivalLng = &lon
if addr := location.Properties.MustString("label", ""); addr != "" {
j.DriverArrivalAddress = &addr
}
return j
}
func (j *PassengerJourney) AddDistance(distance int64) *PassengerJourney {
j.Distance = &distance
return j
}
func (j *PassengerJourney) AddJourneyPolyline(polyline string) *PassengerJourney {
j.JourneyPolyline = &polyline
return j
}
func (j *PassengerJourney) AddPreferences(preferences Preferences) *PassengerJourney {
j.Preferences = &preferences
return j
}
func (j *PassengerJourney) AddAvailableSeats(seats int64) *PassengerJourney {
j.RequestedSteats = &seats
return j
}
func (j *PassengerJourney) AddDriverDepartureDate(date time.Time) *PassengerJourney {
d := OCSSTime(date)
j.DriverDepartureDate = &d
return j
}
func (j *PassengerJourney) AddWebUrl(url string) *PassengerJourney {
j.WebUrl = &url
return j
}

62
interoperability/ocss/time.go Normal file
View File

@ -0,0 +1,62 @@
package ocss
import (
"encoding/json"
"fmt"
"time"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/bson/bsontype"
"go.mongodb.org/mongo-driver/x/bsonx/bsoncore"
)
type OCSSTime time.Time
func (t OCSSTime) MarshalJSON() ([]byte, error) {
//do your serializing here
stamp := fmt.Sprintf("%v", time.Time(t).Unix())
return []byte(stamp), nil
}
func (v OCSSTime) MarshalBSONValue() (bsontype.Type, []byte, error) {
return bson.MarshalValue(time.Time(v))
}
func (t *OCSSTime) UnmarshalJSON(b []byte) error {
var timestamp int64
err := json.Unmarshal(b, &timestamp)
if err != nil {
return err
}
parsed := time.Unix(timestamp, 0)
if err != nil {
return err
}
ocsstime := OCSSTime(parsed)
*t = ocsstime
return nil
}
func (t *OCSSTime) UnmarshalBSONValue(bt bsontype.Type, b []byte) error {
if bt == bsontype.Null || len(b) == 0 {
return nil
}
datetime, _, ok := bsoncore.ReadTime(b)
if !ok {
return fmt.Errorf("cannot parse time")
}
*t = OCSSTime(datetime)
return nil
}
func (t *OCSSTime) ToTime() *time.Time {
if t == nil {
return nil
}
time := time.Time(*t)
return &time
}

75
servers/grpc/server/book.go Normal file
View File

@ -0,0 +1,75 @@
package grpcserver
import (
"context"
"time"
"git.coopgo.io/coopgo-platform/carpool-service/servers/grpc/proto"
"github.com/rs/zerolog/log"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
func (s *CarpoolServiceServerImpl) CreateBooking(ctx context.Context, req *proto.CreateBookingRequest) (*proto.CreateBookingResponse, error) {
booking := req.Booking.ToOCSS()
_, err := s.Handler.Book(booking)
if err != nil {
return nil, status.Errorf(codes.Internal, "could not create booking - %s", err.Error())
}
return &proto.CreateBookingResponse{}, nil
}
func (s *CarpoolServiceServerImpl) GetUserBookings(ctx context.Context, req *proto.GetUserBookingsRequest) (*proto.GetUserBookingsResponse, error) {
log.Debug().
Str("user", req.UserId).
Str("mindate", req.MinDate.AsTime().Format(time.RFC3339)).
Any("maxdate", req.MaxDate).
Msg("grpc server - GetUserBookings")
userid := req.UserId
var mindate *time.Time
if req.MinDate != nil {
d := req.MinDate.AsTime()
mindate = &d
}
var maxdate *time.Time
if req.MaxDate != nil {
d := req.MaxDate.AsTime()
maxdate = &d
}
bookings, err := s.Handler.GetUserBookings(userid, mindate, maxdate)
if err != nil {
return nil, status.Errorf(codes.Internal, "error retrieving user bookings - %s", err.Error())
}
results := []*proto.CarpoolServiceBooking{}
for _, b := range bookings {
nb := proto.BookingFromInternal(b)
results = append(results, nb)
}
return &proto.GetUserBookingsResponse{
Bookings: results,
}, nil
}
func (s *CarpoolServiceServerImpl) UpdateBooking(ctx context.Context, req *proto.UpdateBookingRequest) (*proto.UpdateBookingResponse, error) {
err := s.Handler.UpdateBookingStatus(req.BookingId, req.Status.ToOCSS())
if err != nil {
return nil, status.Errorf(codes.Internal, "could not update booking status")
}
return &proto.UpdateBookingResponse{}, nil
}
func (s *CarpoolServiceServerImpl) GetBooking(ctx context.Context, req *proto.GetBookingRequest) (*proto.GetBookingResponse, error) {
result, err := s.Handler.GetBooking(req.BookingId)
if err != nil {
log.Error().Err(err).Msg("issue retrieving booking in handler")
return nil, err
}
return &proto.GetBookingResponse{
Booking: proto.BookingFromInternal(*result),
}, nil
}

494
storage/postgresql.go Normal file
View File

@ -0,0 +1,494 @@
package storage
import (
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"os"
"strconv"
"ariga.io/atlas/sql/postgres"
"ariga.io/atlas/sql/schema"
"git.coopgo.io/coopgo-platform/carpool-service/internal"
"git.coopgo.io/coopgo-platform/carpool-service/interoperability/ocss"
"github.com/lib/pq"
_ "github.com/lib/pq"
"github.com/paulmach/orb/geojson"
"github.com/rs/zerolog/log"
"github.com/spf13/viper"
"github.com/stretchr/objx"
)
type PostgresqlStorage struct {
DbConnection *sql.DB
Schema string
Tables map[string]string
}
func NewPostgresqlStorage(cfg *viper.Viper) (PostgresqlStorage, error) {
var (
host = cfg.GetString("storage.db.psql.host")
port = cfg.GetString("storage.db.psql.port")
user = cfg.GetString("storage.db.psql.user")
password = cfg.GetString("storage.db.psql.password")
dbname = cfg.GetString("storage.db.psql.dbname")
sslmode = cfg.GetString("storage.db.psql.sslmode")
pg_schema = cfg.GetString("storage.db.psql.schema")
pgtables_regular_routes = cfg.GetString("storage.db.psql.tables.regular_routes")
pgtables_regular_route_schedules = cfg.GetString("storage.db.psql.tables.regular_route_schedules")
pgtables_punctual_routes = cfg.GetString("storage.db.psql.tables.punctual_routes")
pgtables_bookings = cfg.GetString("storage.db.psql.tables.bookings")
pgtables_journeys_cache = cfg.GetString("storage.db.psql.tables.journeys_cache")
)
portInt, _ := strconv.Atoi(port)
psqlconn := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=%s", host, portInt, user, password, dbname, sslmode)
db, err := sql.Open("postgres", psqlconn)
if err != nil {
log.Error().Err(err).Msg("opening connection to postgresql failed")
return PostgresqlStorage{}, fmt.Errorf("connection to postgresql failed")
}
err = db.Ping()
if err != nil {
log.Error().Err(err).Msg("ping to postgresql failed")
return PostgresqlStorage{}, fmt.Errorf("connection to postgresql database failed")
}
return PostgresqlStorage{
DbConnection: db,
Schema: pg_schema,
Tables: map[string]string{
"regular_routes": fmt.Sprintf("%s.%s", pg_schema, pgtables_regular_routes),
"regular_route_schedules": fmt.Sprintf("%s.%s", pg_schema, pgtables_regular_route_schedules),
"punctual_routes": fmt.Sprintf("%s.%s", pg_schema, pgtables_punctual_routes),
"bookings": fmt.Sprintf("%s.%s", pg_schema, pgtables_bookings),
"journeys_cache": fmt.Sprintf("%s.%s", pg_schema, pgtables_journeys_cache),
},
}, nil
}
func (s PostgresqlStorage) CreateRegularRoutes(routes []*geojson.FeatureCollection) error {
log.Debug().Msg("Postgresql Storage - CreateRegularRoutes")
tx, err := s.DbConnection.Begin()
if err != nil {
log.Error().Err(err).Msg("error initializing transaction")
return err
}
defer tx.Rollback()
req_routes := fmt.Sprintf(`INSERT INTO %s (id, user_id, routes_group_id, grid_ids, is_driver, is_passenger, route)
VALUES ($1, $2, $3, $4,$5, $6, $7)`, s.Tables["regular_routes"])
stmt_routes, err := tx.Prepare(req_routes)
if err != nil {
log.Error().Err(err).Msg("error preparing regular routes statement for multiple inserts")
return err
}
defer stmt_routes.Close()
req_schedules := fmt.Sprintf(`INSERT INTO %s (route_id, day, time_of_day) VALUES ($1, $2, $3)`, s.Tables["regular_route_schedules"])
stmt_schedules, err := tx.Prepare(req_schedules)
if err != nil {
log.Error().Err(err).Msg("error preparing schedules statement for multiple inserts")
return err
}
defer stmt_schedules.Close()
for _, fc := range routes {
if fc != nil {
id := fc.ExtraMembers.MustString("id")
properties := objx.New(fc.ExtraMembers["properties"])
userid := properties.Get("user.id").Str()
gridids := fc.ExtraMembers["grid_ids"]
groupid := fc.ExtraMembers["routes_group_id"]
isdriver := properties.Get("is_driver").Bool()
ispassenger := properties.Get("is_passenger").Bool()
route, err := fc.MarshalJSON()
if err != nil {
log.Error().Err(err).Msg("error creating regular route")
return err
}
_, err = stmt_routes.Exec(id, userid, groupid, pq.Array(gridids), isdriver, ispassenger, route)
if err != nil {
log.Error().Err(err).Msg("error inserting regular route")
return err
}
schedules := properties.Get("schedules")
if schedules == nil {
err = errors.New("could not get schedules for route")
log.Error().Err(err).Str("route id", id).Msg("issue in CreateRegularRoute")
return err
}
for _, schedule := range schedules.MustObjxMapSlice() {
day := schedule.Get("day").Str()
timeOfDay := schedule.Get("time_of_day").Str()
_, err = stmt_schedules.Exec(id, day, timeOfDay)
if err != nil {
log.Error().Err(err).Msg("issue creating route schedule")
return err
}
}
}
}
if err = tx.Commit(); err != nil {
log.Error().Err(err).Msg("issue committing transaction")
return err
}
return nil
}
func (s PostgresqlStorage) GetUserRegularRoutes(userid string) ([]*geojson.FeatureCollection, error) {
req := fmt.Sprintf(`select id, route
from %s where user_id = $1`, s.Tables["regular_routes"])
rows, err := s.DbConnection.Query(req, userid)
if err != nil {
log.Error().Err(err).Str("request", req).Str("user id", userid).Msg("GetUserRegularRoutes query issue")
return nil, err
}
results := []*geojson.FeatureCollection{}
for rows.Next() {
var id string
var route []byte
err := rows.Scan(
&id,
&route,
)
if err != nil {
return nil, err
}
fc, err := geojson.UnmarshalFeatureCollection(route)
if err != nil {
return nil, err
}
results = append(results, fc)
}
return results, nil
}
func (s PostgresqlStorage) GetDriverRegularRoutesForTile(day string, gridId int64) (regular_routes []*geojson.FeatureCollection, err error) {
req := fmt.Sprintf(`select id, route
from %s inner join %s on id = route_id
where is_driver = true and day = $1 and $2 = ANY (grid_ids)`, s.Tables["regular_routes"], s.Tables["regular_route_schedules"])
rows, err := s.DbConnection.Query(req, day, gridId)
if err != nil {
log.Error().Err(err).Msg("GetDriverRegularRoutesForTile query error")
return nil, err
}
results := []*geojson.FeatureCollection{}
for rows.Next() {
var id string
var route []byte
err := rows.Scan(
&id,
&route,
)
if err != nil {
return nil, err
}
fc, err := geojson.UnmarshalFeatureCollection(route)
if err != nil {
return nil, err
}
results = append(results, fc)
}
return results, nil
}
func (s PostgresqlStorage) GetPassengerRegularRoutesForTile(day string, gridId int64) (regular_routes []*geojson.FeatureCollection, err error) {
req := fmt.Sprintf(`select id, route
from %s join %s on id = route_id
where is_passenger = true and day = $1 and $2 = ANY (grid_ids)`, s.Tables["regular_routes"], s.Tables["regular_route_schedules"])
rows, err := s.DbConnection.Query(req, day, gridId)
if err != nil {
return nil, err
}
results := []*geojson.FeatureCollection{}
for rows.Next() {
var id string
var route []byte
err := rows.Scan(
&id,
&route,
)
if err != nil {
return nil, err
}
fc, err := geojson.UnmarshalFeatureCollection(route)
if err != nil {
return nil, err
}
results = append(results, fc)
}
return results, nil
}
func (s PostgresqlStorage) CreateBooking(booking internal.Booking) error {
req := fmt.Sprintf(`insert into %s (id, booking, status, driver_route, passenger_route)
values($1, $2, $3, $4, $5)`, s.Tables["bookings"])
id := booking.ID
status := booking.Status.String()
jsonbooking, err := json.Marshal(booking)
if err != nil {
log.Error().Err(err).Msg("issue marshalling booking to json")
return err
}
var jsondriverroute, jsonpassengerroute *[]byte
if booking.DriverRoute != nil {
jdr, err := booking.DriverRoute.MarshalJSON()
if err != nil {
log.Error().Err(err).Msg("error marshalling driver route")
return err
}
jsondriverroute = &jdr
}
if booking.PassengerRoute != nil {
jpr, err := booking.PassengerRoute.MarshalJSON()
if err != nil {
log.Error().Err(err).Msg("error marshalling passenger route")
return err
}
jsonpassengerroute = &jpr
}
_, err = s.DbConnection.Exec(req, id, jsonbooking, status, jsondriverroute, jsonpassengerroute)
if err != nil {
log.Error().Err(err).Str("request", req).Msg("error creating booking")
return err
}
return nil
}
func (s PostgresqlStorage) GetBooking(id string) (*internal.Booking, error) {
req := fmt.Sprintf(`select booking, status, driver_route, passenger_route
from %s where id=$1`, s.Tables["bookings"])
var booking ocss.Booking
var status string
var bookingbytes, driverroute, passengerroute []byte
err := s.DbConnection.QueryRow(req, id).Scan(
&bookingbytes,
&status,
&driverroute,
&passengerroute,
)
if err != nil {
log.Error().Err(err).Str("booking id", id).Msg("not able to get and scan booking")
return nil, err
}
err = json.Unmarshal(bookingbytes, &booking)
if err != nil {
log.Error().Err(err).Msg("issue unmarshalling booking")
return nil, err
}
// Override booking status
booking.Status = ocss.BookingStatusFromString(status)
var dr, pr *geojson.FeatureCollection
if driverroute != nil {
dr, err = geojson.UnmarshalFeatureCollection(driverroute)
if err != nil {
log.Error().Err(err).Msg("could not unmarshal driver route feature collection")
return nil, err
}
}
if passengerroute != nil {
pr, err = geojson.UnmarshalFeatureCollection(passengerroute)
if err != nil {
log.Error().Err(err).Msg("could not unmarshal passenger route feature collection")
return nil, err
}
}
return &internal.Booking{
Booking: booking,
DriverRoute: dr,
PassengerRoute: pr,
}, nil
}
func (s PostgresqlStorage) UpdateBookingStatus(bookingid string, status string) error {
req := fmt.Sprintf(`update %s set status = $1 where id=$2`, s.Tables["bookings"])
_, err := s.DbConnection.Exec(req, status, bookingid)
if err != nil {
log.Error().Err(err).Str("request", req).Str("booking id", bookingid).Str("status", status).Msg("error while updating booking status")
}
return nil
}
func (s PostgresqlStorage) GetUserBookings(userid string) ([]internal.Booking, error) {
req := fmt.Sprintf(`select booking, status, driver_route, passenger_route from %s
where booking->'driver'->>'id' = $1 or booking->'passenger'->>'id' = $2`, s.Tables["bookings"])
rows, err := s.DbConnection.Query(req, userid, userid)
if err != nil {
log.Error().Err(err).Str("user id", userid).Msg("GetUserBookings query issue")
}
results := []internal.Booking{}
for rows.Next() {
var booking ocss.Booking
var status string
var bookingbytes, driverroute, passengerroute []byte
err := rows.Scan(
&bookingbytes,
&status,
&driverroute,
&passengerroute,
)
if err != nil {
log.Error().Err(err).Msg("not able to get and scan booking in GetUsersBooking")
return nil, err
}
err = json.Unmarshal(bookingbytes, &booking)
if err != nil {
log.Error().Err(err).Msg("issue unmarshalling booking in GetUsersBooking")
return nil, err
}
// Override booking status
booking.Status = ocss.BookingStatusFromString(status)
var dr, pr *geojson.FeatureCollection
if driverroute != nil {
dr, err = geojson.UnmarshalFeatureCollection(driverroute)
if err != nil {
log.Error().Err(err).Msg("could not unmarshal driver route feature collection")
return nil, err
}
}
if passengerroute != nil {
pr, err = geojson.UnmarshalFeatureCollection(passengerroute)
if err != nil {
log.Error().Err(err).Msg("could not unmarshal passenger route feature collection")
return nil, err
}
}
results = append(results, internal.Booking{
Booking: booking,
DriverRoute: dr,
PassengerRoute: pr,
})
}
return results, nil
}
func (s PostgresqlStorage) StoreRouteSchedules(journeys []internal.PlannedRouteSchedule) error {
tx, err := s.DbConnection.Begin()
if err != nil {
log.Error().Err(err).Msg("issue starting pg transaction")
return err
}
defer tx.Rollback()
req := fmt.Sprintf("insert into %s (id, data) values ($1, $2)", s.Tables["journeys_cache"])
stmt, err := tx.Prepare(req)
if err != nil {
log.Error().Err(err).Msg("issue creating prepared statement in StoreRouteSchedules")
return err
}
for _, j := range journeys {
jsonjourney, err := json.Marshal(j)
if err != nil {
log.Error().Err(err).Msg("error unmarshalling Route Schedule")
return err
}
stmt.Exec(j.ID, jsonjourney)
}
if err = tx.Commit(); err != nil {
log.Error().Err(err).Msg("issue while commiting transaction in StoreRouteSchedules")
return err
}
return nil
}
func (s PostgresqlStorage) GetRouteSchedule(id string) (*internal.PlannedRouteSchedule, error) {
req := fmt.Sprintf("select data from %s where id = $1", s.Tables["journeys_cache"])
var jsonjourney []byte
err := s.DbConnection.QueryRow(req, id).Scan(
&jsonjourney,
)
if err != nil {
log.Error().Err(err).Msg("issue scanning result in GetRouteSchedule")
return nil, err
}
var result internal.PlannedRouteSchedule
err = json.Unmarshal(jsonjourney, &result)
if err != nil {
log.Error().Err(err).Msg("error unmarshalling returned route schedule")
}
return &result, nil
}
func (s PostgresqlStorage) Migrate() error {
ctx := context.Background()
driver, err := postgres.Open(s.DbConnection)
if err != nil {
return err
}
existing, err := driver.InspectRealm(ctx, &schema.InspectRealmOption{Schemas: []string{s.Schema}})
if err != nil {
return err
}
var desired schema.Realm
hcl, err := os.ReadFile("postgresql/schema.hcl")
if err != nil {
return err
}
err = postgres.EvalHCLBytes(hcl, &desired, nil)
if err != nil {
return err
}
diff, err := driver.RealmDiff(existing, &desired)
if err != nil {
return err
}
err = driver.ApplyChanges(ctx, diff)
if err != nil {
return err
}
return nil
}

104
storage/postgresql/schema.hcl Normal file
View File

@ -0,0 +1,104 @@
table "bookings" {
schema = schema.carpool_service
column "id" {
null = false
type = uuid
}
column "booking" {
null = false
type = jsonb
}
column "status" {
null = false
type = enum.booking_status
}
column "driver_route" {
null = true
type = jsonb
}
column "passenger_route" {
null = true
type = jsonb
}
primary_key {
columns = [column.id]
}
}
table "journeys_cache" {
schema = schema.carpool_service
column "id" {
null = false
type = uuid
}
column "data" {
null = false
type = jsonb
}
column "created_at" {
null = false
type = timestamptz
default = sql("now()")
}
primary_key {
columns = [column.id]
}
}
table "regular_route_schedules" {
schema = schema.carpool_service
column "route_id" {
null = false
type = uuid
}
column "day" {
null = false
type = enum.day
}
column "time_of_day" {
null = false
type = time
}
}
table "regular_routes" {
schema = schema.carpool_service
column "id" {
null = false
type = uuid
}
column "user_id" {
null = false
type = uuid
}
column "routes_group_id" {
null = true
type = uuid
}
column "grid_ids" {
null = true
type = sql("numeric[]")
}
column "is_driver" {
null = true
type = boolean
}
column "is_passenger" {
null = true
type = boolean
}
column "route" {
null = false
type = jsonb
}
primary_key {
columns = [column.id]
}
}
enum "booking_status" {
schema = schema.carpool_service
values = ["INITIATED", "WAITING_PASSENGER_CONFIRMATION", "WAITING_DRIVER_CONFIRMATION", "CONFIRMED", "CANCELLED", "COMPLETED_PENDING_VALIDATION", "VALIDATED"]
}
enum "day" {
schema = schema.carpool_service
values = ["MON", "TUE", "WED", "THU", "FRI", "SAT", "SUN"]
}
schema "carpool_service" {
}

134
storage/postgresql_test.go Normal file
View File

@ -0,0 +1,134 @@
package storage
import (
"fmt"
"testing"
"github.com/paulmach/orb/geojson"
"github.com/spf13/viper"
"github.com/stretchr/testify/require"
)
var cfg *viper.Viper
func init() {
cfg = viper.New()
cfg.SetDefault("storage.db.psql.host", "localhost")
cfg.SetDefault("storage.db.psql.port", "5432")
cfg.SetDefault("storage.db.psql.user", "postgres")
cfg.SetDefault("storage.db.psql.password", "postgres")
cfg.SetDefault("storage.db.psql.dbname", "carpool_service_tests")
cfg.SetDefault("storage.db.psql.sslmode", "disable")
cfg.SetDefault("storage.db.psql.schema", "carpool_service")
cfg.SetDefault("storage.db.psql.tables.regular_routes", "regular_routes")
cfg.SetDefault("storage.db.psql.tables.regular_route_schedules", "regular_route_schedules")
cfg.SetDefault("storage.db.psql.tables.punctual_routes", "punctual_routes")
cfg.SetDefault("storage.db.psql.tables.bookings", "bookings")
cfg.SetDefault("storage.db.psql.tables.journeys_cache", "journeys_cache")
cfg.SetConfigName("config") // Override default values in a config.yaml file within this directory
cfg.AddConfigPath(".")
cfg.ReadInConfig()
}
func TestPostgresqlStorage_Initialize(t *testing.T) {
storage, err := NewPostgresqlStorage(cfg)
require.NoError(t, err)
defer storage.DbConnection.Close()
err = storage.Migrate()
require.NoError(t, err)
_, err = storage.DbConnection.Exec(fmt.Sprintf("DELETE FROM %s;", storage.Tables["regular_routes"]))
require.NoError(t, err)
_, err = storage.DbConnection.Exec(fmt.Sprintf("DELETE FROM %s;", storage.Tables["regular_route_schedules"]))
require.NoError(t, err)
// _, err = storage.DbConnection.Exec(fmt.Sprintf("DELETE FROM %s;", storage.Tables["punctual_routes"]))
// require.NoError(t, err)
_, err = storage.DbConnection.Exec(fmt.Sprintf("DELETE FROM %s;", storage.Tables["journeys_cache"]))
require.NoError(t, err)
}
func TestPostgresqlStorage_CreateRegularRoutes(t *testing.T) {
storage, err := NewPostgresqlStorage(cfg)
require.NoError(t, err)
defer storage.DbConnection.Close()
fc1, err := geojson.UnmarshalFeatureCollection([]byte(regularroute1))
require.NoError(t, err)
fc2, err := geojson.UnmarshalFeatureCollection([]byte(regularroute2))
require.NoError(t, err)
fc3, err := geojson.UnmarshalFeatureCollection([]byte(regularroute3))
require.NoError(t, err)
fc4, err := geojson.UnmarshalFeatureCollection([]byte(regularroute4))
require.NoError(t, err)
fc5, err := geojson.UnmarshalFeatureCollection([]byte(regularroute5))
require.NoError(t, err)
fc6, err := geojson.UnmarshalFeatureCollection([]byte(regularroute6))
require.NoError(t, err)
err = storage.CreateRegularRoutes([]*geojson.FeatureCollection{fc1, fc2, fc3, fc4, fc5, fc6})
require.NoError(t, err)
}
func TestPostgresqlStorage_GetUserRegularRoutes(t *testing.T) {
storage, err := NewPostgresqlStorage(cfg)
require.NoError(t, err)
defer storage.DbConnection.Close()
results, err := storage.GetUserRegularRoutes("ca435b62-3162-4c79-983a-2a0f4fb16aa0")
require.NoError(t, err)
require.Len(t, results, 2)
}
func TestPostgresqlStorage_GetRegularRoutesForTile(t *testing.T) {
storage, err := NewPostgresqlStorage(cfg)
require.NoError(t, err)
defer storage.DbConnection.Close()
results, err := storage.GetDriverRegularRoutesForTile("MON", 48067)
require.NoError(t, err)
require.Len(t, results, 3)
results, err = storage.GetPassengerRegularRoutesForTile("MON", 48067)
require.NoError(t, err)
require.Len(t, results, 3)
}
func TestPostgresqlStorage_CreateBooking(t *testing.T) {
storage, err := NewPostgresqlStorage(cfg)
require.NoError(t, err)
defer storage.DbConnection.Close()
err = storage.CreateBooking(booking1)
require.NoError(t, err)
err = storage.CreateBooking(booking2)
require.NoError(t, err)
err = storage.CreateBooking(booking3)
require.NoError(t, err)
}
func TestPostgresqlStorage_GetBooking(t *testing.T) {
storage, err := NewPostgresqlStorage(cfg)
require.NoError(t, err)
defer storage.DbConnection.Close()
_, err = storage.GetBooking(booking1id)
require.NoError(t, err)
}
func TestPostgresqlStorage_GetUserBookings(t *testing.T) {
storage, err := NewPostgresqlStorage(cfg)
require.NoError(t, err)
defer storage.DbConnection.Close()
results, err := storage.GetUserBookings(bookinguser1)
require.NoError(t, err)
require.Len(t, results, 2)
}

1092
storage/storage_test.go Normal file
View File

File diff suppressed because it is too large Load Diff