Adapt activity extraction SQL to bytebar schema

src/main/java/at/procon/eventhub/service/AbstractTachographActivityRowMapper.java

@@ -267,6 +267,15 @@ abstract class AbstractTachographActivityRowMapper implements TachographExtracti
             return fallback;
         }
         String normalized = value.trim().toUpperCase(Locale.ROOT).replace('-', '_').replace(' ', '_');
+        if ("CODRIVER".equals(normalized)) {
+            normalized = "CO_DRIVER";
+        } else if ("DRIVING".equals(normalized)) {
+            normalized = "DRIVE";
+        } else if ("BREAK/REST".equals(normalized) || "REST".equals(normalized)) {
+            normalized = "BREAK_REST";
+        } else if ("AVAILABLE".equals(normalized)) {
+            normalized = "AVAILABILITY";
+        }
         try {
             return Enum.valueOf(type, normalized);
         } catch (IllegalArgumentException ignored) {

src/main/java/at/procon/eventhub/service/JdbcTachographExtractionBatchExecutor.java

@@ -105,7 +105,11 @@ public class JdbcTachographExtractionBatchExecutor implements TachographExtracti
                 .filter(event -> event.sourcePackageRef() != null && event.sourcePackageRef().importedIntoSourceAt() != null)
                 .max((left, right) -> left.sourcePackageRef().importedIntoSourceAt().compareTo(right.sourcePackageRef().importedIntoSourceAt()))
                 .map(event -> event.sourcePackageRef().sourcePackageId())
-                .orElse(cursor == null ? null : cursor.lastSourcePackageId());
+                .orElseGet(() -> events.stream()
+                        .map(event -> event.sourcePackageRef() == null ? null : event.sourcePackageRef().sourcePackageId())
+                        .filter(value -> value != null && !value.isBlank())
+                        .max(this::compareSourcePackageId)
+                        .orElse(cursor == null ? null : cursor.lastSourcePackageId()));
 
         return new TachographExtractionBatchResultDto(
                 packageId,
@@ -174,4 +178,24 @@ public class JdbcTachographExtractionBatchExecutor implements TachographExtracti
             throw new IllegalStateException("Cannot load tachograph extraction SQL resource " + location, e);
         }
     }
+
+    private int compareSourcePackageId(String left, String right) {
+        Integer leftInt = parseInteger(left);
+        Integer rightInt = parseInteger(right);
+        if (leftInt != null && rightInt != null) {
+            return leftInt.compareTo(rightInt);
+        }
+        return left.compareTo(right);
+    }
+
+    private Integer parseInteger(String value) {
+        if (value == null || value.isBlank()) {
+            return null;
+        }
+        try {
+            return Integer.parseInt(value.trim());
+        } catch (NumberFormatException ignored) {
+            return null;
+        }
+    }
 }

src/main/resources/sql/tachograph/card-activity.sql

@@ -1,50 +1,83 @@
 /*
- * CardActivity DRIVER_ACTIVITY extraction.
+ * CardActivity DRIVER_ACTIVITY extraction for the bytebar tachograph schema.
  *
- * Adapt table and column names to the concrete tachograph SQL Server schema.
+ * Real join path:
- * Keep the selected aliases stable; CardActivityRowMapper consumes this alias contract.
+ *   CardActivity -> CardDailyActivity -> Card -> Driver/Nation
+ *
+ * CardActivity itself has no direct vehicle reference. The OUTER APPLY resolves
+ * the best matching CardVehiclesUsed row for the activity timestamp when one is
+ * available.
  */
 select
-    cast(ca.Id as varchar(128)) as source_row_id,
+    cast(ca.ID as varchar(128)) as source_row_id,
-    cast(ca.Id as varchar(128)) as card_activity_id,
+    cast(ca.ID as varchar(128)) as card_activity_id,
-    concat('TACHOGRAPH:CARD_ACTIVITY:', ca.Id) as external_source_event_id,
+    concat('TACHOGRAPH:CARD_ACTIVITY:', ca.ID) as external_source_event_id,
 
-    ca.ActivityTime as occurred_at,
+    ca.BeginTime as occurred_at,
-    ca.ReceivedAt as received_partner_at,
+    cast(null as datetime) as received_partner_at,
     ca.Activity as activity_code,
-    ca.ActivityText as activity_text,
+    ca.Activity as activity_text,
-    ca.EventType as event_type,
+    case upper(coalesce(ca.Activity, ''))
-    ca.Lifecycle as lifecycle,
+        when 'DRIVING' then 'DRIVE'
-    ca.CardSlot as card_slot,
+        when 'DRIVE' then 'DRIVE'
+        when 'WORK' then 'WORK'
+        when 'AVAILABILITY' then 'AVAILABILITY'
+        when 'AVAILABLE' then 'AVAILABILITY'
+        when 'BREAK_REST' then 'BREAK_REST'
+        when 'BREAK/REST' then 'BREAK_REST'
+        when 'REST' then 'BREAK_REST'
+        else 'UNKNOWN_ACTIVITY'
+    end as event_type,
+    'SNAPSHOT' as lifecycle,
+    ca.Slot as card_slot,
     ca.CardStatus as card_status,
     ca.DrivingStatus as driving_status,
-    ca.OdometerM as odometer_m,
+    cast(null as bigint) as odometer_m,
 
-    cast(ca.DriverId as varchar(128)) as driver_source_entity_id,
+    cast(d.ID as varchar(128)) as driver_source_entity_id,
-    ca.DriverCardNation as driver_card_nation,
+    cn.AlphaCode as driver_card_nation,
-    ca.DriverCardNumber as driver_card_number,
+    c.CardNumber as driver_card_number,
 
-    cast(ca.VehicleId as varchar(128)) as vehicle_source_entity_id,
+    cast(coalesce(cvu.ID_Vehicle, v.ID) as varchar(128)) as vehicle_source_entity_id,
-    ca.VehicleVin as vehicle_vin,
+    coalesce(cvu.VIN, vi.VIN) as vehicle_vin,
-    ca.VehicleRegistrationNation as vehicle_registration_nation,
+    vn.AlphaCode as vehicle_registration_nation,
-    ca.VehicleRegistrationNumber as vehicle_registration_number,
+    v.VRN as vehicle_registration_number,
 
     'DRIVER_CARD' as source_package_kind,
-    cast(ca.SourcePackageId as varchar(128)) as source_package_id,
+    cast(coalesce(ca.ID_FileLog, cda.ID_FileLog, c.ID_FileLog) as varchar(128)) as source_package_id,
-    cast(ca.DriverId as varchar(128)) as source_package_entity_id,
+    cast(c.ID as varchar(128)) as source_package_entity_id,
-    ca.SourcePackagePeriodFrom as source_package_period_from,
+    cda.RecordDate as source_package_period_from,
-    ca.SourcePackagePeriodTo as source_package_period_to,
+    coalesce(cda.RecordDateTo, dateadd(day, 1, cda.RecordDate)) as source_package_period_to,
-    ca.SourcePackageImportedAt as source_package_imported_at
+    cast(null as datetime) as source_package_imported_at
-from CardActivity ca
+from dbo.CardActivity ca
-where (:occurredFrom is null or ca.ActivityTime >= :occurredFrom)
+join dbo.CardDailyActivity cda on cda.ID = ca.ID_DailyActivity
-  and (:occurredTo is null or ca.ActivityTime < :occurredTo)
+join dbo.Card c on c.ID = cda.ID_Card
+left join dbo.Driver d on d.ID = c.ID_Driver
+left join dbo.Nation cn on cn.ID = c.ID_Nation
+outer apply (
+    select top 1 used.ID_Vehicle,
+                 used.VIN,
+                 used.OdoBegin,
+                 used.ID_VUInstallation
+    from dbo.CardVehiclesUsed used
+    where used.ID_Card = c.ID
+      and (used.FirstUse is null or used.FirstUse <= ca.BeginTime)
+      and (used.LastUse is null or used.LastUse >= ca.BeginTime)
+    order by
+        case when used.FirstUse is null then 1 else 0 end,
+        used.FirstUse desc,
+        used.ID desc
+) cvu
+left join dbo.Vehicle v on v.ID = cvu.ID_Vehicle
+left join dbo.VehicleIdentification vi on vi.ID = v.ID_VehicleIdentification
+left join dbo.Nation vn on vn.ID = v.ID_Nation
+where (:occurredFrom is null or ca.BeginTime >= :occurredFrom)
+  and (:occurredTo is null or ca.BeginTime < :occurredTo)
   and (
-      :lastSourcePackageImportedAt is null
+      :lastSourcePackageId is null
-      or ca.SourcePackageImportedAt > :lastSourcePackageImportedAt
+      or coalesce(ca.ID_FileLog, cda.ID_FileLog, c.ID_FileLog, ca.ID) > try_convert(int, :lastSourcePackageId)
-      or (ca.SourcePackageImportedAt = :lastSourcePackageImportedAt and cast(ca.SourcePackageId as varchar(128)) > :lastSourcePackageId)
   )
 /*
- * Organisation filtering is schema-specific. Once stable joins are known, add:
+ * Organisation filtering can be added through Driver_I_90021 / Vehicle_I_90021
- *   and (:rootOrganisationId is null or ...)
+ * once the exact organisation subtree semantics are confirmed.
- * using :includeChildren to decide whether to match only root or descendants.
  */

src/main/resources/sql/tachograph/vu-activity.sql

@@ -1,50 +1,80 @@
 /*
- * VUActivity DRIVER_ACTIVITY extraction.
+ * VUActivity DRIVER_ACTIVITY extraction for the bytebar tachograph schema.
  *
- * Adapt table and column names to the concrete tachograph SQL Server schema.
+ * Real join path:
- * Keep the selected aliases stable; VuActivityRowMapper consumes this alias contract.
+ *   VUActivity -> VUDailyActivity -> VUInstallation -> VehicleIdentification
+ * Optional driver/card context comes from VUActivity.ID_IWCycle -> IWCycle -> Card.
  */
 select
-    cast(va.Id as varchar(128)) as source_row_id,
+    cast(va.ID as varchar(128)) as source_row_id,
-    cast(va.Id as varchar(128)) as vu_activity_id,
+    cast(va.ID as varchar(128)) as vu_activity_id,
-    concat('TACHOGRAPH:VU_ACTIVITY:', va.Id) as external_source_event_id,
+    concat('TACHOGRAPH:VU_ACTIVITY:', va.ID) as external_source_event_id,
 
-    va.ActivityTime as occurred_at,
+    va.BeginTime as occurred_at,
-    va.ReceivedAt as received_partner_at,
+    cast(null as datetime) as received_partner_at,
     va.Activity as activity_code,
-    va.ActivityText as activity_text,
+    va.Activity as activity_text,
-    va.EventType as event_type,
+    case upper(coalesce(va.Activity, ''))
-    va.Lifecycle as lifecycle,
+        when 'DRIVING' then 'DRIVE'
-    va.CardSlot as card_slot,
+        when 'DRIVE' then 'DRIVE'
+        when 'WORK' then 'WORK'
+        when 'AVAILABILITY' then 'AVAILABILITY'
+        when 'AVAILABLE' then 'AVAILABILITY'
+        when 'BREAK_REST' then 'BREAK_REST'
+        when 'BREAK/REST' then 'BREAK_REST'
+        when 'REST' then 'BREAK_REST'
+        else 'UNKNOWN_ACTIVITY'
+    end as event_type,
+    'SNAPSHOT' as lifecycle,
+    va.Slot as card_slot,
     va.CardStatus as card_status,
     va.DrivingStatus as driving_status,
-    va.OdometerM as odometer_m,
+    cast(iw.OdoBegin as bigint) as odometer_m,
 
-    cast(va.DriverId as varchar(128)) as driver_source_entity_id,
+    cast(d.ID as varchar(128)) as driver_source_entity_id,
-    va.DriverCardNation as driver_card_nation,
+    cn.AlphaCode as driver_card_nation,
-    va.DriverCardNumber as driver_card_number,
+    c.CardNumber as driver_card_number,
 
-    cast(va.VehicleId as varchar(128)) as vehicle_source_entity_id,
+    cast(v.ID as varchar(128)) as vehicle_source_entity_id,
-    va.VehicleVin as vehicle_vin,
+    vi.VIN as vehicle_vin,
-    va.VehicleRegistrationNation as vehicle_registration_nation,
+    vn.AlphaCode as vehicle_registration_nation,
-    va.VehicleRegistrationNumber as vehicle_registration_number,
+    v.VRN as vehicle_registration_number,
 
     'VEHICLE_UNIT' as source_package_kind,
-    cast(va.SourcePackageId as varchar(128)) as source_package_id,
+    cast(coalesce(va.ID_FileLog, vda.ID_FileLog, vui.ID_FileLog) as varchar(128)) as source_package_id,
-    cast(va.VehicleId as varchar(128)) as source_package_entity_id,
+    cast(vui.ID_VehicleIdentification as varchar(128)) as source_package_entity_id,
-    va.SourcePackagePeriodFrom as source_package_period_from,
+    vda.RecordDate as source_package_period_from,
-    va.SourcePackagePeriodTo as source_package_period_to,
+    dateadd(day, 1, vda.RecordDate) as source_package_period_to,
-    va.SourcePackageImportedAt as source_package_imported_at
+    cast(null as datetime) as source_package_imported_at
-from VUActivity va
+from dbo.VUActivity va
-where (:occurredFrom is null or va.ActivityTime >= :occurredFrom)
+join dbo.VUDailyActivity vda on vda.ID = va.ID_VUDailyActivity
-  and (:occurredTo is null or va.ActivityTime < :occurredTo)
+join dbo.VUInstallation vui on vui.ID = vda.ID_VUInstallation
+join dbo.VehicleIdentification vi on vi.ID = vui.ID_VehicleIdentification
+outer apply (
+    select top 1 vehicle.ID,
+                 vehicle.VRN,
+                 vehicle.ID_Nation
+    from dbo.Vehicle vehicle
+    where vehicle.ID_VehicleIdentification = vi.ID
+      and (vehicle.ValidFrom is null or vehicle.ValidFrom <= va.BeginTime)
+      and (vehicle.ValidTo is null or vehicle.ValidTo > va.BeginTime)
+    order by
+        case when vehicle.ValidFrom is null then 1 else 0 end,
+        vehicle.ValidFrom desc,
+        vehicle.ID desc
+) v
+left join dbo.Nation vn on vn.ID = v.ID_Nation
+left join dbo.IWCycle iw on iw.ID = va.ID_IWCycle
+left join dbo.Card c on c.ID = iw.ID_Card
+left join dbo.Driver d on d.ID = c.ID_Driver
+left join dbo.Nation cn on cn.ID = c.ID_Nation
+where (:occurredFrom is null or va.BeginTime >= :occurredFrom)
+  and (:occurredTo is null or va.BeginTime < :occurredTo)
   and (
-      :lastSourcePackageImportedAt is null
+      :lastSourcePackageId is null
-      or va.SourcePackageImportedAt > :lastSourcePackageImportedAt
+      or coalesce(va.ID_FileLog, vda.ID_FileLog, vui.ID_FileLog, va.ID) > try_convert(int, :lastSourcePackageId)
-      or (va.SourcePackageImportedAt = :lastSourcePackageImportedAt and cast(va.SourcePackageId as varchar(128)) > :lastSourcePackageId)
   )
 /*
- * Organisation filtering is schema-specific. Once stable joins are known, add:
+ * Organisation filtering can be added through Vehicle_I_90021 / Driver_I_90021
- *   and (:rootOrganisationId is null or ...)
+ * once the exact organisation subtree semantics are confirmed.
- * using :includeChildren to decide whether to match only root or descendants.
  */