v0.3.1

go.mod

@@ -13,12 +13,10 @@ require (
 	github.com/joho/godotenv v1.4.0
 	github.com/nsf/jsondiff v0.0.0-20210926074059-1e845ec5d249
 	github.com/prometheus/client_golang v1.13.1
-	github.com/satimoto/go-datastore v0.3.1-0.20221207141414-56f9531eb6ea
+	github.com/satimoto/go-datastore v0.3.1-0.20221215215552-467363aef4a9
 	google.golang.org/grpc v1.47.0
 )
 
-require github.com/satimoto/go-lsp v0.2.1-0.20221115190646-b981e98c7ba4 // indirect
-
 require (
 	github.com/99designs/gqlgen v0.17.2 // indirect
 	github.com/appleboy/go-fcm v0.1.5

go.sum

@@ -394,6 +394,14 @@ github.com/satimoto/go-datastore v0.3.1-0.20221207124238-2ae95d5797d4 h1:dV1+hbH
 github.com/satimoto/go-datastore v0.3.1-0.20221207124238-2ae95d5797d4/go.mod h1:SvM8losYPwH6hJgUKLkjNtQwVv06lZgCMWl8cGdllCM=
 github.com/satimoto/go-datastore v0.3.1-0.20221207141414-56f9531eb6ea h1:CivzL7OWIhSdnZ6meu50n6NlDcYelphbquZe5930M3s=
 github.com/satimoto/go-datastore v0.3.1-0.20221207141414-56f9531eb6ea/go.mod h1:SvM8losYPwH6hJgUKLkjNtQwVv06lZgCMWl8cGdllCM=
+github.com/satimoto/go-datastore v0.3.1-0.20221211195231-cf0e5e78480d h1:wkTwNt8gWY4a9Fuvk+xZEKGIo+sWxeFsnMaeOFc9HIo=
+github.com/satimoto/go-datastore v0.3.1-0.20221211195231-cf0e5e78480d/go.mod h1:SvM8losYPwH6hJgUKLkjNtQwVv06lZgCMWl8cGdllCM=
+github.com/satimoto/go-datastore v0.3.1-0.20221211212503-67cd89cf5af6 h1:mp63StobDDHg6sQL19afbPI55cslXodxPKNcShcA6Mg=
+github.com/satimoto/go-datastore v0.3.1-0.20221211212503-67cd89cf5af6/go.mod h1:SvM8losYPwH6hJgUKLkjNtQwVv06lZgCMWl8cGdllCM=
+github.com/satimoto/go-datastore v0.3.1-0.20221211215042-d5318660b0e1 h1:j8SMRJQwSu7noWT2TEt+33URmPXpFw1+B/dc0aIRr4I=
+github.com/satimoto/go-datastore v0.3.1-0.20221211215042-d5318660b0e1/go.mod h1:SvM8losYPwH6hJgUKLkjNtQwVv06lZgCMWl8cGdllCM=
+github.com/satimoto/go-datastore v0.3.1-0.20221215215552-467363aef4a9 h1:vmWEl6/uFLQND2y0IE47Ym8iBDRhfF/z8AdFH9ls8LI=
+github.com/satimoto/go-datastore v0.3.1-0.20221215215552-467363aef4a9/go.mod h1:SvM8losYPwH6hJgUKLkjNtQwVv06lZgCMWl8cGdllCM=
 github.com/satimoto/go-lsp v0.2.1-0.20221115190646-b981e98c7ba4 h1:3QKFoeOVSDU8bvrVfj/zAMDq+hg3KJBLaMgEgutJExk=
 github.com/satimoto/go-lsp v0.2.1-0.20221115190646-b981e98c7ba4/go.mod h1:bleUpVumUC8QAFi0CZdnbtDRGWRFKETMH5JWcLAolBs=
 github.com/sergi/go-diff v1.1.0 h1:we8PVUC3FE2uYfodKH/nBHMSetSfHDR6scGdBi+erh0=

internal/cdr/v2.1.1/mocks/mock_resolve.go

@@ -2,6 +2,7 @@ package mocks
 
 import (
 	cdrMocks "github.com/satimoto/go-datastore/pkg/cdr/mocks"
+	command "github.com/satimoto/go-datastore/pkg/command/mocks"
 	mocks "github.com/satimoto/go-datastore/pkg/db/mocks"
 	node "github.com/satimoto/go-datastore/pkg/node/mocks"
 	session "github.com/satimoto/go-datastore/pkg/session/mocks"
@@ -19,8 +20,10 @@ func NewResolver(repositoryService *mocks.MockRepositoryService, services *servi
 	return &cdr.CdrResolver{
 		Repository:             cdrMocks.NewRepository(repositoryService),
 		OcpiService:            services.OcpiService,
+		AsyncService:           services.AsyncService,
 		CalibrationResolver:    calibration.NewResolver(repositoryService),
 		ChargingPeriodResolver: chargingperiod.NewResolver(repositoryService),
+		CommandRepository:      command.NewRepository(repositoryService),
 		LocationResolver:       location.NewResolver(repositoryService, services),
 		NodeRepository:         node.NewRepository(repositoryService),
 		SessionRepository:      session.NewRepository(repositoryService),

internal/cdr/v2.1.1/monitor.go

@@ -0,0 +1,41 @@
+package cdr
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/satimoto/go-datastore/pkg/db"
+	"github.com/satimoto/go-ocpi/internal/async"
+	coreCommand "github.com/satimoto/go-ocpi/internal/command"
+	metrics "github.com/satimoto/go-ocpi/internal/metric"
+)
+
+func (r *CdrResolver) updateCommand(cdr db.Cdr, session db.Session) {
+	ctx := context.Background()
+
+	if cdr.AuthorizationID.Valid {
+		updateCommandStopBySessionIDParams := db.UpdateCommandStopBySessionIDParams{
+			SessionID:   session.Uid,
+			Status:      db.CommandResponseTypeACCEPTED,
+			LastUpdated: time.Now().UTC(),
+		}
+
+		command, err := r.CommandRepository.UpdateCommandStopBySessionID(ctx, updateCommandStopBySessionIDParams)
+
+		if err != nil {
+			metrics.RecordError("OCPI329", "Error updating command stop", err)
+			log.Printf("OCPI329: SessionUid=%#v", session.Uid)
+			return
+		}
+
+		asyncKey := fmt.Sprintf(coreCommand.STOP_COMMAND_ASYNC_KEY, command.ID)
+		asyncResult := async.AsyncResult{
+			String: string(command.Status),
+			Bool:   command.Status == db.CommandResponseTypeACCEPTED,
+		}
+
+		r.AsyncService.Set(asyncKey, asyncResult)
+	}
+}

internal/cdr/v2.1.1/process.go

@@ -133,6 +133,8 @@ func (r *CdrResolver) ReplaceCdrByIdentifier(ctx context.Context, credential db.
 					metrics.RecordError("OCPI283", "Error updating session", err)
 					log.Printf("OCPI283: Params=%#v", sessionParams)
 				}
+
+				r.updateCommand(cdr, session)
 			} else {
 				// A session was never received for this cdr, create it
 				createSessionParams := db.CreateSessionParams{

internal/cdr/v2.1.1/resolve.go

@@ -2,10 +2,12 @@ package cdr
 
 import (
 	"github.com/satimoto/go-datastore/pkg/cdr"
+	"github.com/satimoto/go-datastore/pkg/command"
 	"github.com/satimoto/go-datastore/pkg/db"
 	"github.com/satimoto/go-datastore/pkg/node"
 	"github.com/satimoto/go-datastore/pkg/session"
 	"github.com/satimoto/go-datastore/pkg/token"
+	"github.com/satimoto/go-ocpi/internal/async"
 	"github.com/satimoto/go-ocpi/internal/calibration"
 	"github.com/satimoto/go-ocpi/internal/chargingperiod"
 	location "github.com/satimoto/go-ocpi/internal/location/v2.1.1"
@@ -18,8 +20,10 @@ import (
 type CdrResolver struct {
 	Repository             cdr.CdrRepository
 	OcpiService            *transportation.OcpiService
+	AsyncService           *async.AsyncService
 	CalibrationResolver    *calibration.CalibrationResolver
 	ChargingPeriodResolver *chargingperiod.ChargingPeriodResolver
+	CommandRepository      command.CommandRepository
 	LocationResolver       *location.LocationResolver
 	NodeRepository         node.NodeRepository
 	SessionRepository      session.SessionRepository
@@ -32,8 +36,10 @@ func NewResolver(repositoryService *db.RepositoryService, services *service.Serv
 	return &CdrResolver{
 		Repository:             cdr.NewRepository(repositoryService),
 		OcpiService:            services.OcpiService,
+		AsyncService:           services.AsyncService,
 		CalibrationResolver:    calibration.NewResolver(repositoryService),
 		ChargingPeriodResolver: chargingperiod.NewResolver(repositoryService),
+		CommandRepository:      command.NewRepository(repositoryService),
 		LocationResolver:       location.NewResolver(repositoryService, services),
 		NodeRepository:         node.NewRepository(repositoryService),
 		SessionRepository:      session.NewRepository(repositoryService),

internal/command/constant.go

@@ -0,0 +1,8 @@
+package command
+
+const (
+	RESERVE_NOW_ASYNC_KEY      = "commands/RESERVE_NOW/%v"
+	START_COMMAND_ASYNC_KEY    = "commands/START_COMMAND/%v"
+	STOP_COMMAND_ASYNC_KEY     = "commands/STOP_COMMAND/%v"
+	UNLOCK_CONNECTOR_ASYNC_KEY = "commands/UNLOCK_CONNECTOR/%v"
+)

internal/command/v2.1.1/mocks/mock_resolve.go

@@ -4,15 +4,19 @@ import (
 	commandMocks "github.com/satimoto/go-datastore/pkg/command/mocks"
 	mocks "github.com/satimoto/go-datastore/pkg/db/mocks"
 	command "github.com/satimoto/go-ocpi/internal/command/v2.1.1"
+	evse "github.com/satimoto/go-ocpi/internal/evse/v2.1.1/mocks"
 	"github.com/satimoto/go-ocpi/internal/service"
 	token "github.com/satimoto/go-ocpi/internal/token/v2.1.1/mocks"
+	versiondetail "github.com/satimoto/go-ocpi/internal/versiondetail/mocks"
 )
 
-
 func NewResolver(repositoryService *mocks.MockRepositoryService, services *service.ServiceResolver) *command.CommandResolver {
 	return &command.CommandResolver{
-		Repository:    commandMocks.NewRepository(repositoryService),
-		OcpiService:   services.OcpiService,
-		TokenResolver: token.NewResolver(repositoryService, services),
+		Repository:            commandMocks.NewRepository(repositoryService),
+		OcpiService:           services.OcpiService,
+		AsyncService:          services.AsyncService,
+		EvseResolver:          evse.NewResolver(repositoryService, services),
+		TokenResolver:         token.NewResolver(repositoryService, services),
+		VersionDetailResolver: versiondetail.NewResolver(repositoryService, services),
 	}
 }

internal/command/v2.1.1/process.go

@@ -2,11 +2,14 @@ package command
 
 import (
 	"context"
+	"fmt"
 	"log"
 	"time"
 
 	"github.com/satimoto/go-datastore/pkg/db"
 	"github.com/satimoto/go-datastore/pkg/param"
+	"github.com/satimoto/go-ocpi/internal/async"
+	coreCommand "github.com/satimoto/go-ocpi/internal/command"
 	dto "github.com/satimoto/go-ocpi/internal/dto/v2.1.1"
 	metrics "github.com/satimoto/go-ocpi/internal/metric"
 )
@@ -17,27 +20,49 @@ func (r *CommandResolver) UpdateCommandReservation(ctx context.Context, command
 		commandParams.Status = *commandResponseDto.Result
 		commandParams.LastUpdated = time.Now().UTC()
 
-		_, err := r.Repository.UpdateCommandReservation(ctx, commandParams)
+		updatedCommand, err := r.Repository.UpdateCommandReservation(ctx, commandParams)
 
 		if err != nil {
 			metrics.RecordError("OCPI038", "Error updating command reservation", err)
 			log.Printf("OCPI038: Params=%#v", commandParams)
 		}
+
+		statusAccepted := updatedCommand.Status == db.CommandResponseTypeACCEPTED
+		asyncKey := fmt.Sprintf(coreCommand.RESERVE_NOW_ASYNC_KEY, updatedCommand.ID)
+		asyncResult := async.AsyncResult{
+			String: string(updatedCommand.Status),
+			Bool:   statusAccepted,
+		}
+
+		r.AsyncService.Set(asyncKey, asyncResult)
 	}
 }
 
-func (r *CommandResolver) UpdateCommandStart(ctx context.Context, command db.CommandStart, commandResponseDto *dto.CommandResponseDto) {
+func (r *CommandResolver) UpdateCommandStart(ctx context.Context, credential db.Credential, command db.CommandStart, commandResponseDto *dto.CommandResponseDto) {
 	if commandResponseDto != nil {
 		commandParams := param.NewUpdateCommandStartParams(command)
 		commandParams.Status = *commandResponseDto.Result
 		commandParams.LastUpdated = time.Now().UTC()
 
-		_, err := r.Repository.UpdateCommandStart(ctx, commandParams)
+		updatedCommand, err := r.Repository.UpdateCommandStart(ctx, commandParams)
 
 		if err != nil {
 			metrics.RecordError("OCPI039", "Error updating command start", err)
 			log.Printf("OCPI039: Params=%#v", commandParams)
 		}
+
+		statusAccepted := updatedCommand.Status == db.CommandResponseTypeACCEPTED
+		asyncKey := fmt.Sprintf(coreCommand.START_COMMAND_ASYNC_KEY, updatedCommand.ID)
+		asyncResult := async.AsyncResult{
+			String: string(updatedCommand.Status),
+			Bool:   statusAccepted,
+		}
+
+		r.AsyncService.Set(asyncKey, asyncResult)
+
+		if statusAccepted && updatedCommand.AuthorizationID.Valid {
+			go r.waitForEvseStatus(credential, updatedCommand)
+		}
 	}
 }
 
@@ -47,12 +72,21 @@ func (r *CommandResolver) UpdateCommandStop(ctx context.Context, command db.Comm
 		commandParams.Status = *commandResponseDto.Result
 		commandParams.LastUpdated = time.Now().UTC()
 
-		_, err := r.Repository.UpdateCommandStop(ctx, commandParams)
+		updatedCommand, err := r.Repository.UpdateCommandStop(ctx, commandParams)
 
 		if err != nil {
 			metrics.RecordError("OCPI040", "Error updating command stop", err)
 			log.Printf("OCPI040: Params=%#v", commandParams)
 		}
+
+		statusAccepted := updatedCommand.Status == db.CommandResponseTypeACCEPTED
+		asyncKey := fmt.Sprintf(coreCommand.STOP_COMMAND_ASYNC_KEY, updatedCommand.ID)
+		asyncResult := async.AsyncResult{
+			String: string(updatedCommand.Status),
+			Bool:   statusAccepted,
+		}
+
+		r.AsyncService.Set(asyncKey, asyncResult)
 	}
 }
 
@@ -62,11 +96,48 @@ func (r *CommandResolver) UpdateCommandUnlock(ctx context.Context, command db.Co
 		commandParams.Status = *commandResponseDto.Result
 		commandParams.LastUpdated = time.Now().UTC()
 
-		_, err := r.Repository.UpdateCommandUnlock(ctx, commandParams)
+		updatedCommand, err := r.Repository.UpdateCommandUnlock(ctx, commandParams)
 
 		if err != nil {
 			metrics.RecordError("OCPI041", "Error updating command unlock", err)
 			log.Printf("OCPI041: Params=%#v", commandParams)
 		}
+
+		statusAccepted := updatedCommand.Status == db.CommandResponseTypeACCEPTED
+		asyncKey := fmt.Sprintf(coreCommand.UNLOCK_CONNECTOR_ASYNC_KEY, updatedCommand.ID)
+		asyncResult := async.AsyncResult{
+			String: string(updatedCommand.Status),
+			Bool:   statusAccepted,
+		}
+
+		r.AsyncService.Set(asyncKey, asyncResult)
+	}
+}
+
+func (r *CommandResolver) waitForEvseStatus(credential db.Credential, command db.CommandStart) {
+	ctx := context.Background()
+	token, err := r.TokenResolver.Repository.GetToken(ctx, command.TokenID)
+
+	if err != nil {
+		metrics.RecordError("OCPI325", "Error getting token", err)
+		log.Printf("OCPI325: CommandID=%v, TokenID=%v", command.ID, command.TokenID)
+		return
+	}
+
+	tokenAuthorization, err := r.TokenResolver.TokenAuthorizationResolver.Repository.GetTokenAuthorizationByAuthorizationID(ctx, command.AuthorizationID.String)
+
+	if err != nil {
+		metrics.RecordError("OCPI326", "Error updating command start", err)
+		log.Printf("OCPI326: CommandID=%v, AuthorizationID=%#v", command.ID, command.AuthorizationID)
+		return
+	}
+
+	if command.EvseUid.Valid {
+		cancelCtx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		r.EvseResolver.WaitForEvseStatus(credential, token, tokenAuthorization, command.LocationID, command.EvseUid.String, db.EvseStatusCHARGING, cancelCtx, cancel, 150)
+
+		<-cancelCtx.Done()
 	}
 }