From 0da599e93905b2f7711dd5ccac8b9b582a6fb5bd Mon Sep 17 00:00:00 2001 From: sepehr-safari Date: Sat, 11 Jul 2026 13:57:50 +0300 Subject: [PATCH] feat(engine): session timeline correlation + close out S1 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add `src/ocpp/timeline.zig` — the final S1 engine module — correlating normalized events into logical charging sessions, and close out the milestone. Mirrors the toolkit's `core/timeline.ts` (the shared conformance contract), not its source. - Correlate by transactionId (StartTransaction response; StopTransaction / MeterValues request payloads), with responses inheriting their Call's id. - Distribute un-keyed events by connectorId and time proximity; Boot / Heartbeat attach to the first session; a trace with no StartTransaction collapses to a single session. - Session status (completed / aborted / active), start / end times, and time-ordered, renumbered sessions. - Vendor the `normal-session` fixture and add an end-to-end pipeline test (parse -> normalize -> correlate); the toolkit's timeline cases are ported. - Mark S1 complete in ROADMAP.md and CURRENT_STATE.md. Closes #14 --- CURRENT_STATE.md | 44 ++- ROADMAP.md | 15 +- src/ocpp/ocpp.zig | 2 + src/ocpp/testdata/normal-session.json | 33 ++ src/ocpp/timeline.zig | 493 ++++++++++++++++++++++++++ 5 files changed, 562 insertions(+), 25 deletions(-) create mode 100644 src/ocpp/testdata/normal-session.json create mode 100644 src/ocpp/timeline.zig diff --git a/CURRENT_STATE.md b/CURRENT_STATE.md index b2165d3..21b9272 100644 --- a/CURRENT_STATE.md +++ b/CURRENT_STATE.md @@ -8,8 +8,8 @@ ## Active milestone -**S1 — Engine core (0.1.0): in progress.** S0 — Foundation is complete (see -[ROADMAP.md](ROADMAP.md)). +**S1 — Engine core (0.1.0): complete.** Next up: **S2 — Detection + conformance** +(see [ROADMAP.md](ROADMAP.md)). ## What's done @@ -32,24 +32,30 @@ **Exit criteria met:** CI green on macOS + Linux; `native doctor --strict` clean. +### S1 — Engine core ✅ + +The pure-Zig, headless OCPP engine under `src/ocpp/`, mirroring the toolkit's +conformance contract (behavior, not source) and tested via +`native test -Dplatform=null`: + +- **Types & value boundary** (#11) — canonical `Event` / `Session` / trace / + parse-result types, the `Direction` / `MessageType` / `Status` enums, and the + `std.json.Value` payload boundary (ADR-0005). +- **Event normalizer** (#12) — message classification, ISO 8601 / epoch + timestamp normalization, and two-pass direction inference. +- **Trace parser** (#13) — JSON object / JSONL / bare-array formats, structural + validation, per-entry warnings, and untrusted-input limits (10 MB, 10 000 + events), into a caller-owned arena. +- **Session timeline** (#14) — transactionId correlation, connector / time-based + distribution of un-keyed events, and session status. + +**Exit criteria met:** the vendored `normal-session` fixture parses and +correlates end to end (one completed session, transactionId 100001); engine +tests green headlessly on macOS + Linux. + ## What's in progress -**S1 — Engine core (0.1.0).** Building the pure-Zig engine module by module: - -- **Engine types & value boundary** (#11) — the `src/ocpp/` foundation: canonical - `Event` / `Session` / trace / parse-result types, the `Direction` / - `MessageType` / `Status` enums, and the `std.json.Value` payload boundary - (ADR-0005). Headless, imports no UI/runtime modules, tested via - `native test -Dplatform=null`. -- **Event normalizer** (#12) — message classification, ISO 8601 / epoch-second / - epoch-millisecond timestamp normalization, and two-pass direction inference - (the CS↔CSMS action tables), producing canonical `Event`s. The toolkit's - normalizer test cases are ported. -- **Trace parser** (#13) — JSON object / JSONL / bare-array formats with format - detection, structural validation, per-entry warnings for malformed data, and - the untrusted-input limits (10 MB input, 10 000 events). Parses into a - caller-owned arena. The toolkit's parser test cases are ported. -- Next: session timeline (#14). +- Nothing in flight — S1 is closed; S2 is next. ## What's next @@ -67,7 +73,7 @@ checks Studio's detected failures against locked goldens. | --- | --- | | `repo` (tooling, CI) | ✅ done for S0 | | `docs` (docs, ADRs) | ✅ done for S0 | -| `ocpp` (engine) | 🚧 in progress (S1) | +| `ocpp` (engine) | ✅ done for S1 | | `ui` (native views) | ⬜ placeholder (S3) | | `capture` (live proxy) | ⬜ not started (S5) | | `cli` (headless) | ⬜ not started (S4) | diff --git a/ROADMAP.md b/ROADMAP.md index ab374cd..438bf2b 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -9,8 +9,8 @@ the stable part. | Milestone | Version | Theme | Status | | --- | --- | --- | --- | | **S0** | 0.0.x | Foundation | ✅ Complete | -| **S1** | 0.1.0 | Engine core | Next up | -| **S2** | 0.1.0 | Detection + conformance | Planned | +| **S1** | 0.1.0 | Engine core | ✅ Complete | +| **S2** | 0.1.0 | Detection + conformance | Next up | | **S3** | 0.2.0 | Inspector UI | Planned | | **S4** | 0.3.0 | Analysis parity+ | Planned | | **S5** | 0.4.0 | Live capture ⭐ | Planned | @@ -22,12 +22,15 @@ Repo, tooling, and CI. A native window that opens, an automation-driven smoke test, founding docs, and architecture decision records. **Exit criteria met:** CI green on macOS + Linux; `native doctor --strict` clean. -## S1 — Engine core (0.1.0) +## S1 — Engine core (0.1.0) ✅ The pure-Zig OCPP engine, headless and testable: canonical `Event` / `Session` -/ `Failure` types, a parser for the three input formats (JSON object, JSONL, -bare array), the normalizer (message-direction inference and timestamp -normalization), and session correlation by transaction id. +types and the `std.json.Value` value boundary (ADR-0005), a parser for the three +input formats (JSON object, JSONL, bare array) with untrusted-input limits, the +normalizer (message classification, timestamp normalization, and direction +inference), and session correlation by transaction id. **Exit criteria met:** the +vendored `normal-session` fixture parses and correlates end to end; engine tests +run green headlessly (`native test -Dplatform=null`) on macOS + Linux. ## S2 — Detection + conformance (0.1.0) diff --git a/src/ocpp/ocpp.zig b/src/ocpp/ocpp.zig index 9bb9529..42a8401 100644 --- a/src/ocpp/ocpp.zig +++ b/src/ocpp/ocpp.zig @@ -7,9 +7,11 @@ pub const types = @import("types.zig"); pub const normalizer = @import("normalizer.zig"); pub const parser = @import("parser.zig"); +pub const timeline = @import("timeline.zig"); test { _ = types; _ = normalizer; _ = parser; + _ = timeline; } diff --git a/src/ocpp/testdata/normal-session.json b/src/ocpp/testdata/normal-session.json new file mode 100644 index 0000000..72852f8 --- /dev/null +++ b/src/ocpp/testdata/normal-session.json @@ -0,0 +1,33 @@ +{ + "traceId": "fixture-normal-session", + "metadata": { + "stationId": "CS-SYNTHETIC-001", + "ocppVersion": "1.6", + "source": "synthetic-fixture", + "description": "Normal charging session: boot -> authorize -> start transaction -> meter values -> stop transaction. No failures expected." + }, + "events": [ + { "timestamp": "2024-01-15T10:00:00.000Z", "direction": "CS_TO_CSMS", "message": [2, "msg-001", "BootNotification", { "chargePointVendor": "SyntheticVendor", "chargePointModel": "SM-100", "chargePointSerialNumber": "CS-SYNTHETIC-001", "firmwareVersion": "1.0.0" }] }, + { "timestamp": "2024-01-15T10:00:00.500Z", "direction": "CSMS_TO_CS", "message": [3, "msg-001", { "currentTime": "2024-01-15T10:00:00.500Z", "interval": 300, "status": "Accepted" }] }, + { "timestamp": "2024-01-15T10:01:00.000Z", "direction": "CS_TO_CSMS", "message": [2, "msg-002", "StatusNotification", { "connectorId": 0, "status": "Available", "errorCode": "NoError" }] }, + { "timestamp": "2024-01-15T10:01:00.500Z", "direction": "CSMS_TO_CS", "message": [3, "msg-002", {}] }, + { "timestamp": "2024-01-15T10:02:00.000Z", "direction": "CS_TO_CSMS", "message": [2, "msg-003", "StatusNotification", { "connectorId": 1, "status": "Preparing", "errorCode": "NoError" }] }, + { "timestamp": "2024-01-15T10:02:00.500Z", "direction": "CSMS_TO_CS", "message": [3, "msg-003", {}] }, + { "timestamp": "2024-01-15T10:02:15.000Z", "direction": "CS_TO_CSMS", "message": [2, "msg-004", "Authorize", { "idTag": "SYNTHETIC-TAG-001" }] }, + { "timestamp": "2024-01-15T10:02:15.500Z", "direction": "CSMS_TO_CS", "message": [3, "msg-004", { "idTagInfo": { "status": "Accepted", "expiryDate": "2024-12-31T23:59:59.000Z", "parentIdTag": "SYNTHETIC-PARENT-001" } }] }, + { "timestamp": "2024-01-15T10:02:30.000Z", "direction": "CS_TO_CSMS", "message": [2, "msg-005", "StartTransaction", { "connectorId": 1, "idTag": "SYNTHETIC-TAG-001", "meterStart": 0, "timestamp": "2024-01-15T10:02:30.000Z" }] }, + { "timestamp": "2024-01-15T10:02:30.500Z", "direction": "CSMS_TO_CS", "message": [3, "msg-005", { "transactionId": 100001, "idTagInfo": { "status": "Accepted" } }] }, + { "timestamp": "2024-01-15T10:02:31.000Z", "direction": "CS_TO_CSMS", "message": [2, "msg-006", "StatusNotification", { "connectorId": 1, "status": "Charging", "errorCode": "NoError" }] }, + { "timestamp": "2024-01-15T10:02:31.500Z", "direction": "CSMS_TO_CS", "message": [3, "msg-006", {}] }, + { "timestamp": "2024-01-15T10:07:30.000Z", "direction": "CS_TO_CSMS", "message": [2, "msg-hb-1", "Heartbeat", {}] }, + { "timestamp": "2024-01-15T10:07:30.500Z", "direction": "CSMS_TO_CS", "message": [3, "msg-hb-1", { "currentTime": "2024-01-15T10:07:30.500Z" }] }, + { "timestamp": "2024-01-15T10:15:00.000Z", "direction": "CS_TO_CSMS", "message": [2, "msg-007", "MeterValues", { "connectorId": 1, "transactionId": 100001, "meterValue": [{ "timestamp": "2024-01-15T10:15:00.000Z", "sampledValue": [{ "value": "5000", "measurand": "Energy.Active.Import.Register", "unit": "Wh" }] }] }] }, + { "timestamp": "2024-01-15T10:15:00.500Z", "direction": "CSMS_TO_CS", "message": [3, "msg-007", {}] }, + { "timestamp": "2024-01-15T10:30:00.000Z", "direction": "CS_TO_CSMS", "message": [2, "msg-008", "MeterValues", { "connectorId": 1, "transactionId": 100001, "meterValue": [{ "timestamp": "2024-01-15T10:30:00.000Z", "sampledValue": [{ "value": "10000", "measurand": "Energy.Active.Import.Register", "unit": "Wh" }] }] }] }, + { "timestamp": "2024-01-15T10:30:00.500Z", "direction": "CSMS_TO_CS", "message": [3, "msg-008", {}] }, + { "timestamp": "2024-01-15T10:35:00.000Z", "direction": "CS_TO_CSMS", "message": [2, "msg-009", "StopTransaction", { "transactionId": 100001, "idTag": "SYNTHETIC-TAG-001", "meterStop": 10000, "timestamp": "2024-01-15T10:35:00.000Z", "reason": "EVDisconnected" }] }, + { "timestamp": "2024-01-15T10:35:00.500Z", "direction": "CSMS_TO_CS", "message": [3, "msg-009", { "idTagInfo": { "status": "Accepted" } }] }, + { "timestamp": "2024-01-15T10:35:01.000Z", "direction": "CS_TO_CSMS", "message": [2, "msg-010", "StatusNotification", { "connectorId": 1, "status": "Available", "errorCode": "NoError" }] }, + { "timestamp": "2024-01-15T10:35:01.500Z", "direction": "CSMS_TO_CS", "message": [3, "msg-010", {}] } + ] +} diff --git a/src/ocpp/timeline.zig b/src/ocpp/timeline.zig new file mode 100644 index 0000000..09d90f0 --- /dev/null +++ b/src/ocpp/timeline.zig @@ -0,0 +1,493 @@ +//! Session timeline — correlates normalized events into logical charging +//! sessions. Mirrors the toolkit's `core/timeline.ts` (the shared conformance +//! contract), not its source. +//! +//! Correlation strategy: +//! 1. Primary key: transactionId — from the StartTransaction response, and +//! from StopTransaction / MeterValues request payloads. Responses inherit +//! their Call's transactionId by messageId. +//! 2. Un-keyed events are distributed to sessions by connectorId and time +//! proximity; BootNotification / Heartbeat attach to the first session. +//! 3. A trace with no StartTransaction collapses to a single session. + +const std = @import("std"); +const types = @import("types.zig"); + +const Event = types.Event; +const Session = types.Session; +const Status = types.Status; + +// --------------------------------------------------------------------------- +// Payload / event helpers +// --------------------------------------------------------------------------- + +fn payloadInt(payload: std.json.Value, key: []const u8) ?i64 { + if (payload != .object) return null; + const v = payload.object.get(key) orelse return null; + return if (v == .integer) v.integer else null; +} + +fn payloadStr(payload: std.json.Value, key: []const u8) ?[]const u8 { + if (payload != .object) return null; + const v = payload.object.get(key) orelse return null; + return if (v == .string) v.string else null; +} + +fn actionEql(e: Event, name: []const u8) bool { + return e.action != null and std.mem.eql(u8, e.action.?, name); +} + +/// The transactionId an event carries, if any. StartTransaction Calls carry +/// none (their response does); StopTransaction / MeterValues Calls and any +/// CallResult may carry one in the payload. +fn extractTransactionId(e: Event) ?i64 { + switch (e.message_type) { + .call => { + if (actionEql(e, "StartTransaction")) return null; + if (actionEql(e, "StopTransaction") or actionEql(e, "MeterValues")) { + return payloadInt(e.payload, "transactionId"); + } + return null; + }, + .call_result => return payloadInt(e.payload, "transactionId"), + .call_error => return null, + } +} + +fn extractConnectorId(e: Event) ?i64 { + if (e.message_type != .call) return null; + return payloadInt(e.payload, "connectorId"); +} + +fn extractStationId(events: []const Event) []const u8 { + for (events) |e| { + if (e.message_type == .call and actionEql(e, "BootNotification")) { + if (payloadStr(e.payload, "chargePointSerialNumber")) |s| return s; + } + } + return "unknown"; +} + +fn isStartTransaction(e: Event) bool { + return e.message_type == .call and actionEql(e, "StartTransaction"); +} + +fn isStopTransaction(e: Event) bool { + return e.message_type == .call and actionEql(e, "StopTransaction"); +} + +fn isFaultedStatus(e: Event) bool { + if (e.message_type != .call or !actionEql(e, "StatusNotification")) return false; + const s = payloadStr(e.payload, "status") orelse return false; + return std.mem.eql(u8, s, "Faulted"); +} + +fn isUnavailableStatus(e: Event) bool { + if (e.message_type != .call or !actionEql(e, "StatusNotification")) return false; + const s = payloadStr(e.payload, "status") orelse return false; + return std.mem.eql(u8, s, "Unavailable") or std.mem.eql(u8, s, "Offline"); +} + +// --------------------------------------------------------------------------- +// buildSessionTimeline +// --------------------------------------------------------------------------- + +const EventList = std.ArrayList(Event); + +/// Build session timelines from normalized events. Sessions are ordered by +/// start time and numbered `session-0…`. Allocations come from `arena`; the +/// result borrows from it. +pub fn buildSessionTimeline(arena: std.mem.Allocator, events: []const Event) ![]Session { + if (events.len == 0) return &[_]Session{}; + + const station_id = extractStationId(events); + + // messageId → transactionId from CallResults carrying one (StartTransaction + // responses supply the id its Call lacked). + var response_tx = std.StringHashMap(i64).init(arena); + defer response_tx.deinit(); + for (events) |e| { + if (e.message_type == .call_result) { + if (extractTransactionId(e)) |tx| try response_tx.put(e.message_id, tx); + } + } + + var has_start = false; + for (events) |e| { + if (isStartTransaction(e)) { + has_start = true; + break; + } + } + + // No transactions → one session holding every event. + if (!has_start) { + const one = try arena.alloc(Session, 1); + one[0] = try createSession(arena, "session-0", station_id, events, null, null); + return one; + } + + // messageId → transactionId for Call messages: Stop/Meter from payload, + // StartTransaction from its matched response. + var call_tx = std.StringHashMap(i64).init(arena); + defer call_tx.deinit(); + for (events) |e| { + if (e.message_type != .call) continue; + if (extractTransactionId(e)) |tx| try call_tx.put(e.message_id, tx); + if (actionEql(e, "StartTransaction")) { + if (response_tx.get(e.message_id)) |tx| try call_tx.put(e.message_id, tx); + } + } + + // Group events by transactionId, preserving first-seen order. Un-keyed + // events wait in `null_events` for distribution. + var groups: std.AutoArrayHashMapUnmanaged(i64, EventList) = .empty; + defer groups.deinit(arena); + var null_events: EventList = .empty; + + for (events) |e| { + const tx: ?i64 = switch (e.message_type) { + .call, .call_result => call_tx.get(e.message_id) orelse extractTransactionId(e), + .call_error => null, + }; + if (tx) |txid| { + const gop = try groups.getOrPut(arena, txid); + if (!gop.found_existing) gop.value_ptr.* = .empty; + try gop.value_ptr.append(arena, e); + } else { + try null_events.append(arena, e); + } + } + + var used_null = std.StringHashMap(void).init(arena); + defer used_null.deinit(); + + var sessions: std.ArrayList(Session) = .empty; + var session_index: usize = 0; + + var it = groups.iterator(); + while (it.next()) |entry| { + const txid = entry.key_ptr.*; + const tx_events = entry.value_ptr.*.items; + + var connector_id: ?i64 = null; + for (tx_events) |e| { + if (isStartTransaction(e)) { + connector_id = extractConnectorId(e); + break; + } + } + + var tx_msg_ids = std.StringHashMap(void).init(arena); + defer tx_msg_ids.deinit(); + for (tx_events) |e| try tx_msg_ids.put(e.message_id, {}); + + const session_start = if (tx_events.len > 0) tx_events[0].timestamp else null; + const session_end = if (tx_events.len > 0) tx_events[tx_events.len - 1].timestamp else null; + + var related: EventList = .empty; + for (null_events.items) |e| { + if (used_null.contains(e.id)) continue; + if (shouldAttach(e, connector_id, session_start, session_end, session_index, &tx_msg_ids)) { + try related.append(arena, e); + } + } + for (related.items) |e| try used_null.put(e.id, {}); + + var all: EventList = .empty; + try all.appendSlice(arena, tx_events); + try all.appendSlice(arena, related.items); + std.mem.sort(Event, all.items, {}, lessById); + + const label = try std.fmt.allocPrint(arena, "session-{d}", .{session_index}); + try sessions.append(arena, try createSession(arena, label, station_id, all.items, connector_id, txid)); + session_index += 1; + } + + std.mem.sort(Session, sessions.items, {}, lessByStartTime); + for (sessions.items, 0..) |*s, i| { + s.session_id = try std.fmt.allocPrint(arena, "session-{d}", .{i}); + } + + return sessions.toOwnedSlice(arena); +} + +/// Decide whether an un-keyed event belongs to the session under construction — +/// the toolkit's null-event distribution rules, in order. +fn shouldAttach( + e: Event, + connector_id: ?i64, + session_start: ?i64, + session_end: ?i64, + session_index: usize, + tx_msg_ids: *std.StringHashMap(void), +) bool { + // A response to a Call already in the session. + if (e.message_type != .call and tx_msg_ids.contains(e.message_id)) return true; + + // Same connector: within [start, end + 1 min], or include when timestamps + // are missing. + if (extractConnectorId(e)) |eci| { + if (connector_id) |cid| { + if (eci == cid) { + if (session_start) |ss| if (session_end) |se| if (e.timestamp) |ts| { + return ts >= ss and ts <= se + 60_000; + }; + return true; + } + } + } + + const is_boot = actionEql(e, "BootNotification"); + const is_hb = actionEql(e, "Heartbeat"); + if (is_boot or is_hb) return session_index == 0; + + // Connector-less Calls (e.g. Authorize): from 5 min before start to 1 min + // after end. + if (extractConnectorId(e) == null and e.message_type == .call and !is_boot and !is_hb) { + if (session_start) |ss| if (session_end) |se| if (e.timestamp) |ts| { + return ts >= ss - 300_000 and ts <= se + 60_000; + }; + } + return false; +} + +fn createSession( + arena: std.mem.Allocator, + session_id: []const u8, + station_id: []const u8, + evs: []const Event, + connector_id: ?i64, + tx_in: ?i64, +) !Session { + var start: ?i64 = null; + var end: ?i64 = null; + for (evs) |e| { + if (e.timestamp) |t| { + if (start == null or t < start.?) start = t; + if (end == null or t > end.?) end = t; + } + } + + var has_stop = false; + var has_faulted = false; + var has_unavailable = false; + for (evs) |e| { + if (isStopTransaction(e)) has_stop = true; + if (isFaultedStatus(e)) has_faulted = true; + if (isUnavailableStatus(e)) has_unavailable = true; + } + const status: Status = if (has_stop) + .completed + else if (has_faulted or has_unavailable) + .aborted + else + .active; + + var tx = tx_in; + if (tx == null) { + for (evs) |e| { + if (extractTransactionId(e)) |t| { + tx = t; + break; + } + } + } + + return .{ + .session_id = session_id, + .station_id = station_id, + .connector_id = connector_id, + .transaction_id = tx, + .start_time = start, + .end_time = end, + .events = try arena.dupe(Event, evs), + .status = status, + }; +} + +fn lessById(_: void, a: Event, b: Event) bool { + return std.mem.order(u8, a.id, b.id) == .lt; +} + +fn lessByStartTime(_: void, a: Session, b: Session) bool { + const at = a.start_time orelse return false; // nulls sort last + const bt = b.start_time orelse return true; + return at < bt; +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +const testing = std.testing; +const parser = @import("parser.zig"); + +/// Build an `Event` directly (bypassing the parser), mirroring the toolkit's +/// `makeEvent`. Timeline reads message_type / action / payload / message_id / +/// timestamp — not raw_message — so raw_message is left null. +fn ev( + arena: std.mem.Allocator, + id: []const u8, + message_id: []const u8, + message_type: types.MessageType, + action: ?[]const u8, + payload_json: []const u8, + timestamp: ?i64, +) Event { + return .{ + .id = id, + .message_id = message_id, + .timestamp = timestamp, + .direction = .cs_to_csms, + .message_type = message_type, + .action = action, + .payload = std.json.parseFromSliceLeaky(std.json.Value, arena, payload_json, .{}) catch unreachable, + .error_code = null, + .error_description = null, + .raw_message = .null, + }; +} + +fn normalSessionEvents(a: std.mem.Allocator) []Event { + var list: EventList = .empty; + list.append(a, ev(a, "evt-0001", "msg-001", .call, "BootNotification", "{\"chargePointSerialNumber\":\"CS-001\"}", 1000)) catch unreachable; + list.append(a, ev(a, "evt-0002", "msg-001", .call_result, null, "{\"status\":\"Accepted\"}", 1500)) catch unreachable; + list.append(a, ev(a, "evt-0003", "msg-002", .call, "StatusNotification", "{\"connectorId\":0,\"status\":\"Available\"}", 2000)) catch unreachable; + list.append(a, ev(a, "evt-0004", "msg-002", .call_result, null, "{}", 2500)) catch unreachable; + list.append(a, ev(a, "evt-0005", "msg-003", .call, "Authorize", "{\"idTag\":\"TAG-001\"}", 3000)) catch unreachable; + list.append(a, ev(a, "evt-0006", "msg-003", .call_result, null, "{\"idTagInfo\":{\"status\":\"Accepted\"}}", 3500)) catch unreachable; + list.append(a, ev(a, "evt-0007", "msg-004", .call, "StartTransaction", "{\"connectorId\":1,\"idTag\":\"TAG-001\",\"meterStart\":0}", 4000)) catch unreachable; + list.append(a, ev(a, "evt-0008", "msg-004", .call_result, null, "{\"transactionId\":100001,\"idTagInfo\":{\"status\":\"Accepted\"}}", 4500)) catch unreachable; + list.append(a, ev(a, "evt-0009", "msg-005", .call, "StatusNotification", "{\"connectorId\":1,\"status\":\"Charging\"}", 5000)) catch unreachable; + list.append(a, ev(a, "evt-0010", "msg-005", .call_result, null, "{}", 5500)) catch unreachable; + list.append(a, ev(a, "evt-0011", "msg-006", .call, "MeterValues", "{\"connectorId\":1,\"transactionId\":100001,\"meterValue\":[]}", 6000)) catch unreachable; + list.append(a, ev(a, "evt-0012", "msg-006", .call_result, null, "{}", 6500)) catch unreachable; + list.append(a, ev(a, "evt-0013", "msg-007", .call, "StopTransaction", "{\"transactionId\":100001,\"meterStop\":10000,\"reason\":\"EVDisconnected\"}", 7000)) catch unreachable; + list.append(a, ev(a, "evt-0014", "msg-007", .call_result, null, "{\"idTagInfo\":{\"status\":\"Accepted\"}}", 7500)) catch unreachable; + return list.items; +} + +test "a normal session correlates into one completed session" { + var arena_state = std.heap.ArenaAllocator.init(testing.allocator); + defer arena_state.deinit(); + const a = arena_state.allocator(); + + const sessions = try buildSessionTimeline(a, normalSessionEvents(a)); + try testing.expectEqual(@as(usize, 1), sessions.len); + try testing.expectEqual(@as(?i64, 100001), sessions[0].transaction_id); + try testing.expectEqual(Status.completed, sessions[0].status); + try testing.expectEqual(@as(?i64, 1), sessions[0].connector_id); + try testing.expectEqualStrings("CS-001", sessions[0].station_id); + try testing.expectEqual(@as(?i64, 1000), sessions[0].start_time); + try testing.expectEqual(@as(?i64, 7500), sessions[0].end_time); +} + +test "stationId falls back to unknown without a BootNotification" { + var arena_state = std.heap.ArenaAllocator.init(testing.allocator); + defer arena_state.deinit(); + const a = arena_state.allocator(); + + const events = [_]Event{ev(a, "evt-0001", "msg-001", .call, "Authorize", "{\"idTag\":\"TAG-001\"}", 1000)}; + const sessions = try buildSessionTimeline(a, &events); + try testing.expectEqual(@as(usize, 1), sessions.len); + try testing.expectEqualStrings("unknown", sessions[0].station_id); +} + +test "status is active without a StopTransaction" { + var arena_state = std.heap.ArenaAllocator.init(testing.allocator); + defer arena_state.deinit(); + const a = arena_state.allocator(); + + const events = [_]Event{ + ev(a, "evt-0001", "msg-001", .call, "BootNotification", "{\"chargePointSerialNumber\":\"CS-001\"}", 1000), + ev(a, "evt-0002", "msg-002", .call, "StartTransaction", "{\"connectorId\":1,\"meterStart\":0}", 2000), + ev(a, "evt-0003", "msg-002", .call_result, null, "{\"transactionId\":200001}", 2500), + }; + const sessions = try buildSessionTimeline(a, &events); + try testing.expectEqual(Status.active, sessions[0].status); + try testing.expectEqual(@as(?i64, 200001), sessions[0].transaction_id); +} + +test "status is aborted on a fault with no StopTransaction" { + var arena_state = std.heap.ArenaAllocator.init(testing.allocator); + defer arena_state.deinit(); + const a = arena_state.allocator(); + + const events = [_]Event{ + ev(a, "evt-0001", "msg-001", .call, "StartTransaction", "{\"connectorId\":1,\"meterStart\":0}", 2000), + ev(a, "evt-0002", "msg-001", .call_result, null, "{\"transactionId\":400001}", 2500), + ev(a, "evt-0003", "msg-002", .call, "StatusNotification", "{\"connectorId\":1,\"status\":\"Faulted\"}", 3000), + }; + const sessions = try buildSessionTimeline(a, &events); + try testing.expectEqual(@as(usize, 1), sessions.len); + try testing.expectEqual(Status.aborted, sessions[0].status); +} + +test "a trace with no transactions collapses to one session" { + var arena_state = std.heap.ArenaAllocator.init(testing.allocator); + defer arena_state.deinit(); + const a = arena_state.allocator(); + + const events = [_]Event{ + ev(a, "evt-0001", "msg-001", .call, "BootNotification", "{\"chargePointSerialNumber\":\"CS-001\"}", 1000), + ev(a, "evt-0002", "msg-001", .call_result, null, "{\"status\":\"Accepted\"}", 1500), + ev(a, "evt-0003", "msg-002", .call, "Heartbeat", "{}", 2000), + ev(a, "evt-0004", "msg-002", .call_result, null, "{\"currentTime\":\"2024-01-15T10:00:00.000Z\"}", 2500), + }; + const sessions = try buildSessionTimeline(a, &events); + try testing.expectEqual(@as(usize, 1), sessions.len); + try testing.expectEqual(@as(?i64, null), sessions[0].transaction_id); +} + +test "empty events yield no sessions" { + var arena_state = std.heap.ArenaAllocator.init(testing.allocator); + defer arena_state.deinit(); + const sessions = try buildSessionTimeline(arena_state.allocator(), &[_]Event{}); + try testing.expectEqual(@as(usize, 0), sessions.len); +} + +test "two transactions produce two time-ordered sessions" { + var arena_state = std.heap.ArenaAllocator.init(testing.allocator); + defer arena_state.deinit(); + const a = arena_state.allocator(); + + const events = [_]Event{ + ev(a, "evt-0001", "msg-001", .call, "BootNotification", "{\"chargePointSerialNumber\":\"CS-001\"}", 1000), + ev(a, "evt-0002", "msg-001", .call_result, null, "{\"status\":\"Accepted\"}", 1500), + ev(a, "evt-0003", "msg-002", .call, "StartTransaction", "{\"connectorId\":1,\"meterStart\":0}", 2000), + ev(a, "evt-0004", "msg-002", .call_result, null, "{\"transactionId\":100001}", 2500), + ev(a, "evt-0005", "msg-003", .call, "StopTransaction", "{\"transactionId\":100001,\"meterStop\":5000}", 3000), + ev(a, "evt-0006", "msg-003", .call_result, null, "{\"idTagInfo\":{\"status\":\"Accepted\"}}", 3500), + ev(a, "evt-0007", "msg-004", .call, "StartTransaction", "{\"connectorId\":2,\"meterStart\":0}", 4000), + ev(a, "evt-0008", "msg-004", .call_result, null, "{\"transactionId\":100002}", 4500), + ev(a, "evt-0009", "msg-005", .call, "StopTransaction", "{\"transactionId\":100002,\"meterStop\":3000}", 5000), + ev(a, "evt-0010", "msg-005", .call_result, null, "{\"idTagInfo\":{\"status\":\"Accepted\"}}", 5500), + }; + const sessions = try buildSessionTimeline(a, &events); + try testing.expectEqual(@as(usize, 2), sessions.len); + try testing.expectEqual(@as(?i64, 100001), sessions[0].transaction_id); + try testing.expectEqual(@as(?i64, 100002), sessions[1].transaction_id); + try testing.expectEqualStrings("session-0", sessions[0].session_id); + try testing.expectEqualStrings("session-1", sessions[1].session_id); +} + +test "end to end: parse the normal-session fixture and correlate it" { + var arena_state = std.heap.ArenaAllocator.init(testing.allocator); + defer arena_state.deinit(); + const a = arena_state.allocator(); + + const fixture = @embedFile("testdata/normal-session.json"); + const parsed = try parser.parseTrace(a, fixture); + try testing.expectEqual(@as(usize, 22), parsed.events.len); + try testing.expectEqual(@as(usize, 0), parsed.warnings.len); + + const sessions = try buildSessionTimeline(a, parsed.events); + try testing.expectEqual(@as(usize, 1), sessions.len); + try testing.expectEqual(@as(?i64, 100001), sessions[0].transaction_id); + try testing.expectEqual(Status.completed, sessions[0].status); + try testing.expectEqual(@as(?i64, 1), sessions[0].connector_id); + try testing.expectEqualStrings("CS-SYNTHETIC-001", sessions[0].station_id); + // BootNotification (earliest, 10:00:00.000Z) attaches to the first session. + try testing.expectEqual(@as(?i64, 1_705_312_800_000), sessions[0].start_time); +}