Clean ghost transactions
MODIFY
src/locking/LockingHelper.ts
src/locking/LockingHelper.ts
@@ -116,6 +116,14 @@ export default class LockingHelper {
return lock;
}
+ public static async acquireTransactionsCleanLock(tenantID: string): Promise<Lock | null> {
+ const lock = LockingManager.createExclusiveLock(tenantID, LockEntity.TRANSACTION, 'clean-transactions');
+ if (!(await LockingManager.acquire(lock))) {
+ return null;
+ }
+ return lock;
+ }
+
public static async createOCPIPullTokensLock(tenantID: string, ocpiEndpoint: OCPIEndpoint, partial: boolean): Promise<Lock | null> {
return LockingHelper.acquireOCPIEndpointActionLock(tenantID, ocpiEndpoint, `pull-tokens${partial ? '-partial' : ''}`);
}

MODIFY
src/scheduler/SchedulerManager.ts
src/scheduler/SchedulerManager.ts
@@ -9,6 +9,7 @@ import CheckOfflineChargingStationsTask from './tasks/CheckOfflineChargingStatio
import CheckPreparingSessionNotStartedTask from './tasks/CheckPreparingSessionNotStartedTask';
import CheckSessionNotStartedAfterAuthorizeTask from './tasks/CheckSessionNotStartedAfterAuthorizeTask';
import CheckUserAccountInactivityTask from './tasks/CheckUserAccountInactivityTask';
+import CleanTransactionsInProgressTask from './tasks/CleanTransactionsInProgressTask';
import Constants from '../utils/Constants';
import Logging from '../utils/Logging';
import LoggingDatabaseTableCleanupTask from './tasks/LoggingDatabaseTableCleanupTask';
@@ -142,6 +143,9 @@ export default class SchedulerManager {
case 'MigrateSensitiveDataTask':
schedulerTask = new MigrateSensitiveDataTask();
break;
+ case 'CleanTransactionsInProgressTask':
+ schedulerTask = new CleanTransactionsInProgressTask();
+ break;
default:
await Logging.logError({
tenantID: Constants.DEFAULT_TENANT,

ADD
src/scheduler/tasks/CleanTransactionsInProgressTask.ts
+import { ActionsResponse } from '../../types/GlobalType';
+import Constants from '../../utils/Constants';
+import LockingHelper from '../../locking/LockingHelper';
+import LockingManager from '../../locking/LockingManager';
+import Logging from '../../utils/Logging';
+import OCPPUtils from '../../server/ocpp/utils/OCPPUtils';
+import SchedulerTask from '../SchedulerTask';
+import { ServerAction } from '../../types/Server';
+import Tenant from '../../types/Tenant';
+import { TransactionDataResult } from '../../types/DataResult';
+import TransactionStorage from '../../storage/mongodb/TransactionStorage';
+import Utils from '../../utils/Utils';
+
+const MODULE_NAME = 'CleanTransactionsInProgressTask';
+
+export default class CleanTransactionsInProgressTask extends SchedulerTask {
+ public async processTenant(tenant: Tenant): Promise<void> {
+ const cleanTransactionsLock = await LockingHelper.acquireTransactionsCleanLock(tenant.id);
+ if (cleanTransactionsLock) {
+ try {
+ const result: ActionsResponse = {
+ inError: 0,
+ inSuccess: 0,
+ };
+ const startTime = new Date().getTime();
+ let transactionsToClean: TransactionDataResult;
+ do {
+ // Get Transactions to clean
+ transactionsToClean = await TransactionStorage.getTransactions(tenant, { stop: { $exists: false }, transactionsToClean: true }, Constants.DB_PARAMS_MAX_LIMIT,
+ ['id', 'chargeBoxID', 'tagID', 'lastConsumption.timestamp', 'timestamp', 'lastConsumption.value', 'meterStart']);
+ for (const transactionToClean of transactionsToClean.result) {
+ try {
+ // Soft stop transaction
+ await OCPPUtils.softStopTransaction(tenant, transactionToClean) ? result.inSuccess++ : result.inError++;
+ } catch (error) {
+ result.inError++;
+ await Logging.logError({
+ tenantID: tenant.id,
+ action: ServerAction.TRANSACTION_SOFT_STOP,
+ module: MODULE_NAME, method: 'processTenant',
+ message: `Cannot soft stop Transaction ID '${transactionToClean.id}': ${error.message}`,
+ detailedMessages: { transactionToClean, error: error.stack }
+ });
+ }
+ }
+ if (!Utils.isEmptyArray(transactionsToClean.result) && (result.inError + result.inSuccess) > 0) {
+ const intermediateDurationSecs = Math.round((new Date().getTime() - startTime) / 1000);
+ await Logging.logDebug({
+ tenantID: tenant.id,
+ action: ServerAction.TAGS_IMPORT,
+ module: MODULE_NAME, method: 'processTenant',
+ message: `${result.inError + result.inSuccess}/${transactionsToClean.count} Transaction(s) have been processed in ${intermediateDurationSecs}s...`
+ });
+ }
+ } while (!Utils.isEmptyArray(transactionsToClean?.result));
+ // Log final results
+ const executionDurationSecs = Math.round((new Date().getTime() - startTime) / 1000);
+ await Logging.logActionsResponse(tenant.id, ServerAction.TAGS_IMPORT, MODULE_NAME, 'processTenant', result,
+ `{{inSuccess}} Transaction(s) have been stopped successfully in ${executionDurationSecs}s in Tenant ${Utils.buildTenantName(tenant)}`,
+ `{{inError}} Transaction(s) failed to be stopped in ${executionDurationSecs}s in Tenant ${Utils.buildTenantName(tenant)}`,
+ `{{inSuccess}} Transaction(s) have been stopped successfully but {{inError}} failed in ${executionDurationSecs}s in Tenant ${Utils.buildTenantName(tenant)}`,
+ `Not Transaction has been stopped in ${executionDurationSecs}s in Tenant ${Utils.buildTenantName(tenant)}`
+ );
+ } catch (error) {
+ // Log error
+ await Logging.logActionExceptionMessage(tenant.id, ServerAction.TRANSACTION_SOFT_STOP, error);
+ } finally {
+ // Release the lock
+ await LockingManager.release(cleanTransactionsLock);
+ }
+ }
+ }
+}

MODIFY
src/server/ocpp/utils/OCPPUtils.ts
src/server/ocpp/utils/OCPPUtils.ts
import { BillingDataTransactionStart, BillingDataTransactionStop } from '../../../types/Billing';
import { ChargingProfile, ChargingProfilePurposeType } from '../../../types/ChargingProfile';
import ChargingStation, { ChargingStationCapabilities, ChargingStationTemplate, Command, Connector, ConnectorCurrentLimitSource, CurrentType, OcppParameter, SiteAreaLimitSource, StaticLimitAmps, TemplateUpdate, TemplateUpdateResult } from '../../../types/ChargingStation';
-import FeatureToggles, { Feature } from '../../../utils/FeatureToggles';
+import { OCPPAuthorizationStatus, OCPPMeasurand, OCPPNormalizedMeterValue, OCPPPhase, OCPPReadingContext, OCPPStopTransactionRequestExtended, OCPPUnitOfMeasure, OCPPValueFormat } from '../../../types/ocpp/OCPPServer';
import { OCPPChangeConfigurationResponse, OCPPChargingProfileStatus, OCPPConfigurationStatus } from '../../../types/ocpp/OCPPClient';
-import { OCPPMeasurand, OCPPNormalizedMeterValue, OCPPPhase, OCPPReadingContext, OCPPStopTransactionRequestExtended, OCPPUnitOfMeasure, OCPPValueFormat } from '../../../types/ocpp/OCPPServer';
import { OICPIdentification, OICPSessionID } from '../../../types/oicp/OICPIdentification';
import Tenant, { TenantComponents } from '../../../types/Tenant';
import Transaction, { InactivityStatus, TransactionAction } from '../../../types/Transaction';
@@ -13,6 +12,7 @@ import BackendError from '../../../exception/BackendError';
import BillingFactory from '../../../integration/billing/BillingFactory';
import ChargingStationStorage from '../../../storage/mongodb/ChargingStationStorage';
import ChargingStationVendorFactory from '../../../integration/charging-station-vendor/ChargingStationVendorFactory';
+import Configuration from '../../../utils/Configuration';
import Constants from '../../../utils/Constants';
import Consumption from '../../../types/Consumption';
import ConsumptionStorage from '../../../storage/mongodb/ConsumptionStorage';
@@ -25,6 +25,7 @@ import OCPIClientFactory from '../../../client/ocpi/OCPIClientFactory';
import { OCPIRole } from '../../../types/ocpi/OCPIRole';
import OCPPCommon from './OCPPCommon';
import { OCPPHeader } from '../../../types/ocpp/OCPPHeader';
+import OCPPService from '../services/OCPPService';
import OICPClientFactory from '../../../client/oicp/OICPClientFactory';
import { OICPRole } from '../../../types/oicp/OICPRole';
import OICPUtils from '../../oicp/OICPUtils';
@@ -1714,6 +1715,35 @@ export default class OCPPUtils {
}
}
+ public static async softStopTransaction(tenant: Tenant, transaction: Transaction): Promise<boolean> {
+ // Get the Charging Station
+ const chargingStation = await ChargingStationStorage.getChargingStation(tenant, transaction.chargeBoxID, { withSiteArea: true });
+ // Stop Transaction
+ const result = await new OCPPService(Configuration.getChargingStationConfig()).handleStopTransaction(
+ {
+ chargeBoxIdentity: chargingStation.id,
+ chargingStation: chargingStation,
+ companyID: chargingStation.companyID,
+ siteID: chargingStation.siteID,
+ siteAreaID: chargingStation.siteAreaID,
+ tenantID: tenant.id,
+ tenant: tenant,
+ },
+ {
+ transactionId: transaction.id,
+ chargeBoxID: chargingStation.id,
+ idTag: transaction.tagID,
+ timestamp: Utils.convertToDate(transaction.lastConsumption ? transaction.lastConsumption.timestamp : transaction.timestamp).toISOString(),
+ meterStop: transaction.lastConsumption ? transaction.lastConsumption.value : transaction.meterStart
+ },
+ true
+ );
+ if (result.idTagInfo?.status !== OCPPAuthorizationStatus.ACCEPTED) {
+ return false;
+ }
+ return true;
+ }
+
private static async enrichChargingStationWithTemplate(tenant: Tenant, chargingStation: ChargingStation): Promise<TemplateUpdateResult> {
const templateUpdate: TemplateUpdate = {
chargingStationUpdate: false,

MODIFY
src/storage/mongodb/TransactionStorage.ts
src/storage/mongodb/TransactionStorage.ts
@@ -300,7 +300,7 @@ export default class TransactionStorage {
endDateTime?: Date; stop?: any; minimalPrice?: boolean; reportIDs?: string[]; tagIDs?: string[]; inactivityStatus?: string[];
ocpiSessionID?: string; ocpiAuthorizationID?: string; ocpiSessionDateFrom?: Date; ocpiSessionDateTo?: Date; ocpiCdrDateFrom?: Date; ocpiCdrDateTo?: Date;
ocpiSessionChecked?: boolean; ocpiCdrChecked?: boolean; oicpSessionID?: string; withSite?: boolean; withSiteArea?: boolean; withCompany?: boolean;
- statistics?: 'refund' | 'history' | 'ongoing'; refundStatus?: RefundStatus[]; withTag?: boolean; hasUserID?: boolean; withUser?: boolean; withCar?: boolean;
+ statistics?: 'refund' | 'history' | 'ongoing'; refundStatus?: RefundStatus[]; withTag?: boolean; hasUserID?: boolean; withUser?: boolean; withCar?: boolean; transactionsToClean?: boolean
},
dbParams: DbParams, projectFields?: string[]): Promise<TransactionDataResult> {
const startTime = Logging.traceDatabaseRequestStart();
@@ -477,6 +477,51 @@ export default class TransactionStorage {
$match: filters
});
}
+ // Charging Station
+ if (params.withChargingStation || params.transactionsToClean) {
+ DatabaseUtils.pushChargingStationLookupInAggregation({
+ tenantID: tenant.id, aggregation: aggregation, localField: 'chargeBoxID', foreignField: '_id',
+ asField: 'chargeBox', oneToOneCardinality: true, oneToOneCardinalityNotNull: false
+ });
+ DatabaseUtils.pushConvertObjectIDToString(aggregation, 'chargeBox.siteAreaID');
+ // Add Connector and Status
+ if ((projectFields && projectFields.includes('status')) || params.transactionsToClean) {
+ aggregation.push({
+ $addFields: {
+ connector: {
+ '$arrayElemAt': [
+ {
+ '$filter': {
+ input: '$chargeBox.connectors',
+ as: 'connector',
+ cond: {
+ $eq: [
+ '$$connector.connectorId',
+ '$connectorId'
+ ]
+ }
+ }
+ },
+ 0
+ ]
+ }
+ }
+ });
+ if (projectFields && projectFields.includes('status')) {
+ aggregation.push({
+ $addFields: { status: '$connector.status' }
+ });
+ }
+ if (params.transactionsToClean) {
+ aggregation.push({
+ '$addFields': {
+ 'transactionIdEq': { '$eq': ['$connector.currentTransactionID', '$_id'] }
+ }
+ },
+ { '$match': { 'transactionIdEq': false } });
+ }
+ }
+ }
// Limit records?
if (!dbParams.onlyRecordCount) {
// Always limit the nbr of record to avoid perfs issues
@@ -672,30 +717,6 @@ export default class TransactionStorage {
asField: 'siteArea', oneToOneCardinality: true
});
}
- // Charging Station
- if (params.withChargingStation) {
- DatabaseUtils.pushChargingStationLookupInAggregation({
- tenantID: tenant.id, aggregation: aggregation, localField: 'chargeBoxID', foreignField: '_id',
- asField: 'chargeBox', oneToOneCardinality: true, oneToOneCardinalityNotNull: false
- });
- DatabaseUtils.pushConvertObjectIDToString(aggregation, 'chargeBox.siteAreaID');
- // Add Connector and Status
- if (projectFields && projectFields.includes('status')) {
- aggregation.push({
- $addFields: {
- connector: {
- $arrayElemAt: [
- '$chargeBox.connectors', {
- $indexOfArray: ['$chargeBox.connectors.connectorId', '$connectorId']
- }
- ]
- }
- }
- }, {
- $addFields: { status: '$connector.status' }
- });
- }
- }
// User
if (params.withUser) {
DatabaseUtils.pushUserLookupInAggregation({

Clean ghost transactions