19e103dafc2ed3bdd481df7640a8c5b1131dc814

Committed 28 May 2025 11:21AM UTC coverage: 59.444% (-0.2%) from 59.605%

Build # 19e103dafc2ed3bdd481df7640a8c5b1131dc814

Build Type

push

github

Committed by NSHkr

Commit Message

fb

Run Details

4787 of 8053 relevant lines covered (59.44%)

3840.58 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

70.29

/lib/elixir_scope/ast_repository/runtime_correlator.ex

defmodule ElixirScope.ASTRepository.RuntimeCorrelator do
  @moduledoc """
  Runtime correlation bridge that maps runtime events to AST nodes.
  
  This module provides the core functionality for correlating runtime execution
  events with static AST analysis data, enabling the hybrid architecture.
  
  Key responsibilities:
  - Map correlation IDs from runtime events to AST node IDs
  - Maintain temporal correlation data
  - Provide fast lookup capabilities (<5ms target)
  - Update AST repository with runtime insights
  """
  
  use GenServer
  require Logger
  
  alias ElixirScope.Utils
  alias ElixirScope.Storage.DataAccess
  alias ElixirScope.ASTRepository.Repository
  
  @type correlation_id :: binary()
  @type ast_node_id :: binary()
  @type runtime_event :: map()
  @type correlation_result :: {:ok, ast_node_id()} | {:error, term()}
  
  defstruct [
    # Core State
    :repository_pid,          # Repository process PID
    :data_access,            # DataAccess instance for events
    :correlation_cache,      # ETS table for fast correlation lookup
    :temporal_index,         # ETS table for temporal queries
    :statistics,             # Correlation statistics
    
    # Configuration
    :cache_size_limit,       # Maximum cache entries
    :correlation_timeout,    # Timeout for correlation operations
    :cleanup_interval,       # Cache cleanup interval
    :performance_tracking,   # Enable performance tracking
    
    # Metadata
    :start_time,             # Start timestamp
    :total_correlations,     # Total correlations performed
    :successful_correlations, # Successful correlations
    :failed_correlations     # Failed correlations
  ]
  
  @type t :: %__MODULE__{}
  
  # Default configuration
  @default_config %{
    cache_size_limit: 100_000,
    correlation_timeout: 5_000,
    cleanup_interval: 300_000,  # 5 minutes
    performance_tracking: true
  }
  
  # ETS table options
  @cache_opts [:set, :public, {:read_concurrency, true}, {:write_concurrency, true}]
  @temporal_opts [:bag, :public, {:read_concurrency, true}, {:write_concurrency, true}]
  
  #############################################################################
  # Public API
  #############################################################################
  
  @doc """
  Starts the RuntimeCorrelator with the given repository and configuration.
  """
  @spec start_link(keyword()) :: GenServer.on_start()
  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end
  
  @doc """
  Correlates a runtime event with AST nodes.
  
  Returns the AST node ID that correlates with the event, or an error if
  no correlation can be established.
  """
  @spec correlate_event(GenServer.server(), runtime_event()) :: correlation_result()
  def correlate_event(correlator \\ __MODULE__, runtime_event) do
    GenServer.call(correlator, {:correlate_event, runtime_event})
  end
  
  @doc """
  Batch correlates multiple runtime events for better performance.
  """
  @spec correlate_events(GenServer.server(), [runtime_event()]) :: 
    {:ok, [{correlation_id(), ast_node_id()}]} | {:error, term()}
  def correlate_events(correlator \\ __MODULE__, runtime_events) do
    GenServer.call(correlator, {:correlate_events, runtime_events})
  end
  
  @doc """
  Gets all runtime events correlated with a specific AST node.
  """
  @spec get_correlated_events(GenServer.server(), ast_node_id()) :: 
    {:ok, [runtime_event()]} | {:error, term()}
  def get_correlated_events(correlator \\ __MODULE__, ast_node_id) do
    GenServer.call(correlator, {:get_correlated_events, ast_node_id})
  end
  
  @doc """
  Gets correlation statistics and performance metrics.
  """
  @spec get_statistics(GenServer.server()) :: {:ok, map()}
  def get_statistics(correlator \\ __MODULE__) do
    GenServer.call(correlator, :get_statistics)
  end
  
  @doc """
  Performs a health check on the correlator.
  """
  @spec health_check(GenServer.server()) :: {:ok, map()} | {:error, term()}
  def health_check(correlator \\ __MODULE__) do
    GenServer.call(correlator, :health_check)
  end
  
  @doc """
  Updates the correlation mapping for an AST node.
  """
  @spec update_correlation_mapping(GenServer.server(), correlation_id(), ast_node_id()) :: :ok
  def update_correlation_mapping(correlator \\ __MODULE__, correlation_id, ast_node_id) do
    GenServer.call(correlator, {:update_correlation_mapping, correlation_id, ast_node_id})
  end
  
  @doc """
  Queries events within a time range for temporal correlation.
  """
  @spec query_temporal_events(GenServer.server(), integer(), integer()) :: 
    {:ok, [runtime_event()]} | {:error, term()}
  def query_temporal_events(correlator \\ __MODULE__, start_time, end_time) do
    GenServer.call(correlator, {:query_temporal_events, start_time, end_time})
  end

  @doc """
  Gets all runtime events correlated with a specific AST node, ordered chronologically.
  
  This is the primary function for AST-centric debugging queries.
  """
  @spec get_events_for_ast_node(GenServer.server(), ast_node_id()) :: 
    {:ok, [runtime_event()]} | {:error, term()}
  def get_events_for_ast_node(correlator \\ __MODULE__, ast_node_id) do
    GenServer.call(correlator, {:get_events_for_ast_node, ast_node_id})
  end
  
  #############################################################################
  # GenServer Callbacks
  #############################################################################
  
  @impl true
  def init(opts) do
    Logger.info("🔧 RuntimeCorrelator.init starting with opts: #{inspect(opts)}")
    Logger.info("🔍 Checking Config GenServer availability...")
    
    config_pid = GenServer.whereis(ElixirScope.Config)
    Logger.info("📍 Config GenServer PID: #{inspect(config_pid)}")
    
    if config_pid do
      Logger.info("✅ Config GenServer found, testing responsiveness...")
      try do
        config_result = ElixirScope.Config.get([:ast_repository])
        Logger.info("✅ Config retrieved successfully: #{inspect(config_result)}")
      rescue
        error ->
          Logger.error("❌ Config GenServer unresponsive: #{inspect(error)}")
          Logger.error("📍 Error details: #{inspect(__STACKTRACE__)}")
          {:stop, {:config_unresponsive, error}}
      end
    else
      Logger.error("❌ Config GenServer not found!")
      Logger.error("📍 Available registered processes: #{inspect(Process.registered())}")
      {:stop, :config_not_found}
    end
    
    repository_pid = Keyword.get(opts, :repository_pid)
    Logger.info("🏗️ Building config with repository_pid: #{inspect(repository_pid)}")
    config = build_config(opts)
    
    Logger.info("🔧 Creating correlator state...")
    case create_correlator_state(repository_pid, config) do
      {:ok, state} ->
        # Schedule periodic cleanup
        schedule_cleanup(state.cleanup_interval)
        
        Logger.info("✅ RuntimeCorrelator started successfully with repository: #{inspect(repository_pid)}")
        {:ok, state}
      
      {:error, reason} ->
        Logger.error("❌ Failed to initialize RuntimeCorrelator: #{inspect(reason)}")
        {:stop, reason}
    end
  end
  
  @impl true
  def handle_call({:correlate_event, runtime_event}, _from, state) do
    start_time = if state.performance_tracking, do: Utils.monotonic_timestamp(), else: nil
    
    result = correlate_event_impl(state, runtime_event)
    
    # Update statistics and performance tracking
    new_state = update_correlation_stats(state, result, start_time)
    
    {:reply, result, new_state}
  end
  
  @impl true
  def handle_call({:correlate_events, runtime_events}, _from, state) do
    start_time = if state.performance_tracking, do: Utils.monotonic_timestamp(), else: nil
    
    results = Enum.map(runtime_events, &correlate_event_impl(state, &1))
    
    # Extract successful correlations
    successful_correlations = results
      |> Enum.filter(&match?({:ok, _}, &1))
      |> Enum.map(fn {:ok, {correlation_id, ast_node_id}} -> {correlation_id, ast_node_id} end)
    
    # Update statistics
    new_state = update_batch_correlation_stats(state, results, start_time)
    
    {:reply, {:ok, successful_correlations}, new_state}
  end
  
  @impl true
  def handle_call({:get_correlated_events, ast_node_id}, _from, state) do
    result = get_correlated_events_impl(state, ast_node_id)
    {:reply, result, state}
  end
  
  @impl true
  def handle_call(:get_statistics, _from, state) do
    stats = collect_statistics(state)
    {:reply, {:ok, stats}, state}
  end
  
  @impl true
  def handle_call(:health_check, _from, state) do
    health = perform_health_check(state)
    {:reply, {:ok, health}, state}
  end
  
  @impl true
  def handle_call({:update_correlation_mapping, correlation_id, ast_node_id}, _from, state) do
    :ets.insert(state.correlation_cache, {correlation_id, ast_node_id})
    {:reply, :ok, state}
  end
  
  @impl true
  def handle_call({:query_temporal_events, start_time, end_time}, _from, state) do
    result = query_temporal_events_impl(state, start_time, end_time)
    {:reply, result, state}
  end

  @impl true
  def handle_call({:get_events_for_ast_node, ast_node_id}, _from, state) do
    result = get_events_for_ast_node_impl(state, ast_node_id)
    {:reply, result, state}
  end
  
  @impl true
  def handle_info(:cleanup, state) do
    new_state = perform_cleanup(state)
    schedule_cleanup(state.cleanup_interval)
    {:noreply, new_state}
  end
  
  @impl true
  def handle_info(msg, state) do
    Logger.debug("RuntimeCorrelator received unexpected message: #{inspect(msg)}")
    {:noreply, state}
  end
  
  #############################################################################
  # Private Implementation Functions
  #############################################################################
  
  defp build_config(opts) do
    user_config = Keyword.get(opts, :config, %{})
    Map.merge(@default_config, user_config)
  end
  
  defp create_correlator_state(repository_pid, config) do
    if is_nil(repository_pid) do
      {:error, :repository_pid_required}
    else
      try do
        # Create ETS tables for caching and temporal indexing
        correlation_cache = :ets.new(:correlation_cache, @cache_opts)
        temporal_index = :ets.new(:temporal_index, @temporal_opts)
        
        # Create DataAccess instance for event storage
        {:ok, data_access} = DataAccess.new([name: :correlator_events])
        
        state = %__MODULE__{
          repository_pid: repository_pid,
          data_access: data_access,
          correlation_cache: correlation_cache,
          temporal_index: temporal_index,
          statistics: %{},
          cache_size_limit: config.cache_size_limit,
          correlation_timeout: config.correlation_timeout,
          cleanup_interval: config.cleanup_interval,
          performance_tracking: config.performance_tracking,
          start_time: Utils.monotonic_timestamp(),
          total_correlations: 0,
          successful_correlations: 0,
          failed_correlations: 0
        }
        
        {:ok, state}
      rescue
        error -> {:error, {:initialization_failed, error}}
      end
    end
  end
  
  defp correlate_event_impl(state, runtime_event) do
    correlation_id = extract_correlation_id(runtime_event)
    
    if correlation_id do
      case lookup_correlation(state, correlation_id) do
        {:ok, ast_node_id} ->
          # Store the event for future analysis
          store_correlated_event(state, runtime_event, ast_node_id)
          
          # Update temporal index
          timestamp = extract_timestamp(runtime_event)
          :ets.insert(state.temporal_index, {timestamp, {correlation_id, ast_node_id}})
          
          {:ok, {correlation_id, ast_node_id}}
        
        {:error, :not_found} ->
          # Try to get correlation from repository
          case Repository.correlate_event(state.repository_pid, runtime_event) do
            {:ok, ast_node_id} ->
              # Cache the correlation for future use
              :ets.insert(state.correlation_cache, {correlation_id, ast_node_id})
              
              # Store the event
              store_correlated_event(state, runtime_event, ast_node_id)
              
              # Update temporal index
              timestamp = extract_timestamp(runtime_event)
              :ets.insert(state.temporal_index, {timestamp, {correlation_id, ast_node_id}})
              
              {:ok, {correlation_id, ast_node_id}}
            
            {:error, reason} ->
              {:error, reason}
          end
        
        {:error, reason} ->
          {:error, reason}
      end
    else
      {:error, :no_correlation_id}
    end
  end
  
  defp lookup_correlation(state, correlation_id) do
    case :ets.lookup(state.correlation_cache, correlation_id) do
      [{^correlation_id, ast_node_id}] -> {:ok, ast_node_id}
      [] -> {:error, :not_found}
    end
  end
  
  defp store_correlated_event(state, runtime_event, ast_node_id) do
    # Enrich the event with correlation information
    enriched_event = Map.merge(runtime_event, %{
      correlated_ast_node_id: ast_node_id,
      correlation_timestamp: Utils.monotonic_timestamp()
    })
    
    # Store in DataAccess for future queries
    case DataAccess.store_event(state.data_access, enriched_event) do
      :ok -> :ok
      {:error, reason} ->
        Logger.warning("Failed to store correlated event: #{inspect(reason)}")
        :ok  # Don't fail correlation due to storage issues
    end
  end
  
  defp get_correlated_events_impl(state, ast_node_id) do
    get_events_for_ast_node_impl(state, ast_node_id)
  end

  defp get_events_for_ast_node_impl(state, ast_node_id) do
    try do
      # Get all correlation IDs for this AST node from temporal index
      correlation_ids = :ets.select(state.temporal_index, [
        {{:_, {:'$1', ast_node_id}}, [], [:'$1']}
      ])
      
      # Get events from DataAccess for each correlation ID
      events = Enum.flat_map(correlation_ids, fn correlation_id ->
        case DataAccess.query_by_correlation(state.data_access, correlation_id) do
          {:ok, events} -> events
          {:error, _} -> []
        end
      end)
      
      # Sort by timestamp for chronological order
      sorted_events = Enum.sort_by(events, fn event ->
        extract_timestamp(event)
      end)
      
      {:ok, sorted_events}
    rescue
      error -> {:error, {:query_failed, error}}
    end
  end
  
  defp query_temporal_events_impl(state, start_time, end_time) do
    try do
      # Query temporal index for events in the time range
      temporal_entries = :ets.select(state.temporal_index, [
        {{:'$1', :'$2'}, 
         [{:andalso, {:>=, :'$1', start_time}, {:'=<', :'$1', end_time}}], 
         [:'$2']}
      ])
      
      # Extract correlation IDs and get the actual events
      correlation_ids = Enum.map(temporal_entries, fn {correlation_id, _ast_node_id} -> correlation_id end)
      
      # Get events from DataAccess
      events = Enum.flat_map(correlation_ids, fn correlation_id ->
        case DataAccess.query_by_correlation(state.data_access, correlation_id) do
          {:ok, events} -> events
          {:error, _} -> []
        end
      end)
      
      {:ok, events}
    rescue
      error -> {:error, {:temporal_query_failed, error}}
    end
  end
  
  defp update_correlation_stats(state, result, start_time) do
    new_total = state.total_correlations + 1
    
    {new_successful, new_failed} = case result do
      {:ok, _} -> {state.successful_correlations + 1, state.failed_correlations}
      {:error, _} -> {state.successful_correlations, state.failed_correlations + 1}
    end
    
    # Update performance statistics if tracking is enabled
    new_statistics = if state.performance_tracking and start_time do
      duration = Utils.monotonic_timestamp() - start_time
      update_performance_stats(state.statistics, duration)
    else
      state.statistics
    end
    
    %{state |
      total_correlations: new_total,
      successful_correlations: new_successful,
      failed_correlations: new_failed,
      statistics: new_statistics
    }
  end
  
  defp update_batch_correlation_stats(state, results, start_time) do
    successful_count = Enum.count(results, &match?({:ok, _}, &1))
    failed_count = length(results) - successful_count
    
    new_total = state.total_correlations + length(results)
    new_successful = state.successful_correlations + successful_count
    new_failed = state.failed_correlations + failed_count
    
    # Update performance statistics if tracking is enabled
    new_statistics = if state.performance_tracking and start_time do
      duration = Utils.monotonic_timestamp() - start_time
      update_batch_performance_stats(state.statistics, duration, length(results))
    else
      state.statistics
    end
    
    %{state |
      total_correlations: new_total,
      successful_correlations: new_successful,
      failed_correlations: new_failed,
      statistics: new_statistics
    }
  end
  
  defp update_performance_stats(statistics, duration) do
    current_avg = Map.get(statistics, :average_correlation_time, 0.0)
    current_count = Map.get(statistics, :correlation_count, 0)
    
    new_count = current_count + 1
    new_avg = (current_avg * current_count + duration) / new_count
    
    Map.merge(statistics, %{
      average_correlation_time: new_avg,
      correlation_count: new_count,
      last_correlation_time: duration
    })
  end
  
  defp update_batch_performance_stats(statistics, duration, batch_size) do
    current_avg = Map.get(statistics, :average_batch_correlation_time, 0.0)
    current_count = Map.get(statistics, :batch_correlation_count, 0)
    
    new_count = current_count + 1
    new_avg = (current_avg * current_count + duration) / new_count
    
    Map.merge(statistics, %{
      average_batch_correlation_time: new_avg,
      batch_correlation_count: new_count,
      last_batch_correlation_time: duration,
      last_batch_size: batch_size
    })
  end
  
  defp collect_statistics(state) do
    uptime = Utils.monotonic_timestamp() - state.start_time
    success_rate = if state.total_correlations > 0 do
      state.successful_correlations / state.total_correlations
    else
      0.0
    end
    
    %{
      uptime_ms: uptime,
      total_correlations: state.total_correlations,
      successful_correlations: state.successful_correlations,
      failed_correlations: state.failed_correlations,
      success_rate: success_rate,
      cache_size: :ets.info(state.correlation_cache, :size),
      temporal_index_size: :ets.info(state.temporal_index, :size),
      performance_stats: state.statistics
    }
  end
  
  defp perform_health_check(state) do
    cache_size = :ets.info(state.correlation_cache, :size)
    _temporal_size = :ets.info(state.temporal_index, :size)
    
    status = cond do
      cache_size > state.cache_size_limit * 0.9 -> :warning
      not Process.alive?(state.repository_pid) -> :error
      true -> :healthy
    end
    
    %{
      status: status,
      uptime_ms: Utils.monotonic_timestamp() - state.start_time,
      cache_utilization: cache_size / state.cache_size_limit,
      repository_alive: Process.alive?(state.repository_pid),
      memory_usage: %{
        correlation_cache: :ets.info(state.correlation_cache, :memory),
        temporal_index: :ets.info(state.temporal_index, :memory)
      }
    }
  end
  
  defp perform_cleanup(state) do
    # Clean up old entries from cache if it's getting too large
    cache_size = :ets.info(state.correlation_cache, :size)
    
    if cache_size > state.cache_size_limit do
      # Simple cleanup: remove oldest 10% of entries
      # In practice, you'd want a more sophisticated LRU strategy
      entries_to_remove = div(cache_size, 10)
      
      # Get all keys and remove the first N (this is a simplified approach)
      all_keys = :ets.select(state.correlation_cache, [{{:'$1', :_}, [], [:'$1']}])
      keys_to_remove = Enum.take(all_keys, entries_to_remove)
      
      Enum.each(keys_to_remove, fn key ->
        :ets.delete(state.correlation_cache, key)
      end)
      
      Logger.debug("RuntimeCorrelator cleanup: removed #{entries_to_remove} cache entries")
    end
    
    state
  end
  
  defp schedule_cleanup(interval) do
    Process.send_after(self(), :cleanup, interval)
  end
  
  defp extract_correlation_id(%{correlation_id: correlation_id}), do: correlation_id
  defp extract_correlation_id(%{"correlation_id" => correlation_id}), do: correlation_id
  defp extract_correlation_id(_), do: nil
  
  defp extract_timestamp(%{timestamp: timestamp}), do: timestamp
  defp extract_timestamp(%{"timestamp" => timestamp}), do: timestamp
  defp extract_timestamp(_), do: Utils.monotonic_timestamp()
end 

1	defmodule ElixirScope.ASTRepository.RuntimeCorrelator do
2	@moduledoc """
3	Runtime correlation bridge that maps runtime events to AST nodes.
4
5	This module provides the core functionality for correlating runtime execution
6	events with static AST analysis data, enabling the hybrid architecture.
7
8	Key responsibilities:
9	- Map correlation IDs from runtime events to AST node IDs
10	- Maintain temporal correlation data
11	- Provide fast lookup capabilities (<5ms target)
12	- Update AST repository with runtime insights
13	"""
14
15	use GenServer
16	require Logger
17
18	alias ElixirScope.Utils
19	alias ElixirScope.Storage.DataAccess
20	alias ElixirScope.ASTRepository.Repository
21
22	@type correlation_id :: binary()
23	@type ast_node_id :: binary()
24	@type runtime_event :: map()
25	@type correlation_result :: {:ok, ast_node_id()} \| {:error, term()}
26
27	defstruct [
28	# Core State
29	:repository_pid, # Repository process PID
30	:data_access, # DataAccess instance for events
31	:correlation_cache, # ETS table for fast correlation lookup
32	:temporal_index, # ETS table for temporal queries
33	:statistics, # Correlation statistics
34
35	# Configuration
36	:cache_size_limit, # Maximum cache entries
37	:correlation_timeout, # Timeout for correlation operations
38	:cleanup_interval, # Cache cleanup interval
39	:performance_tracking, # Enable performance tracking
40
41	# Metadata
42	:start_time, # Start timestamp
43	:total_correlations, # Total correlations performed
44	:successful_correlations, # Successful correlations
45	:failed_correlations # Failed correlations
46	]
47
48	@type t :: %__MODULE__{}
49
50	# Default configuration
51	@default_config %{
52	cache_size_limit: 100_000,
53	correlation_timeout: 5_000,
54	cleanup_interval: 300_000, # 5 minutes
55	performance_tracking: true
56	}
57
58	# ETS table options
59	@cache_opts [:set, :public, {:read_concurrency, true}, {:write_concurrency, true}]
60	@temporal_opts [:bag, :public, {:read_concurrency, true}, {:write_concurrency, true}]
61
62	#############################################################################
63	# Public API
64	#############################################################################
65
66	@doc """
67	Starts the RuntimeCorrelator with the given repository and configuration.
68	"""
69	@spec start_link(keyword()) :: GenServer.on_start()
70	def start_link(opts \\ []) do
71	GenServer.start_link(__MODULE__, opts, name: __MODULE__)	8✔
72	end
73
74	@doc """
75	Correlates a runtime event with AST nodes.
76
77	Returns the AST node ID that correlates with the event, or an error if
78	no correlation can be established.
79	"""
80	@spec correlate_event(GenServer.server(), runtime_event()) :: correlation_result()
81	def correlate_event(correlator \\ __MODULE__, runtime_event) do
82	GenServer.call(correlator, {:correlate_event, runtime_event})	6✔
83	end
84
85	@doc """
86	Batch correlates multiple runtime events for better performance.
87	"""
88	@spec correlate_events(GenServer.server(), [runtime_event()]) ::
89	{:ok, [{correlation_id(), ast_node_id()}]} \| {:error, term()}
90	def correlate_events(correlator \\ __MODULE__, runtime_events) do
91	GenServer.call(correlator, {:correlate_events, runtime_events})	1✔
92	end
93
94	@doc """
95	Gets all runtime events correlated with a specific AST node.
96	"""
97	@spec get_correlated_events(GenServer.server(), ast_node_id()) ::
98	{:ok, [runtime_event()]} \| {:error, term()}
99	def get_correlated_events(correlator \\ __MODULE__, ast_node_id) do
100	GenServer.call(correlator, {:get_correlated_events, ast_node_id})	1✔
101	end
102
103	@doc """
104	Gets correlation statistics and performance metrics.
105	"""
106	@spec get_statistics(GenServer.server()) :: {:ok, map()}
107	def get_statistics(correlator \\ __MODULE__) do
108	GenServer.call(correlator, :get_statistics)	×
109	end
110
111	@doc """
112	Performs a health check on the correlator.
113	"""
114	@spec health_check(GenServer.server()) :: {:ok, map()} \| {:error, term()}
115	def health_check(correlator \\ __MODULE__) do
116	GenServer.call(correlator, :health_check)	1✔
117	end
118
119	@doc """
120	Updates the correlation mapping for an AST node.
121	"""
122	@spec update_correlation_mapping(GenServer.server(), correlation_id(), ast_node_id()) :: :ok
123	def update_correlation_mapping(correlator \\ __MODULE__, correlation_id, ast_node_id) do
124	GenServer.call(correlator, {:update_correlation_mapping, correlation_id, ast_node_id})	1✔
125	end
126
127	@doc """
128	Queries events within a time range for temporal correlation.
129	"""
130	@spec query_temporal_events(GenServer.server(), integer(), integer()) ::
131	{:ok, [runtime_event()]} \| {:error, term()}
132	def query_temporal_events(correlator \\ __MODULE__, start_time, end_time) do
133	GenServer.call(correlator, {:query_temporal_events, start_time, end_time})	1✔
134	end
135
136	@doc """
137	Gets all runtime events correlated with a specific AST node, ordered chronologically.
138
139	This is the primary function for AST-centric debugging queries.
140	"""
141	@spec get_events_for_ast_node(GenServer.server(), ast_node_id()) ::
142	{:ok, [runtime_event()]} \| {:error, term()}
143	def get_events_for_ast_node(correlator \\ __MODULE__, ast_node_id) do
144	GenServer.call(correlator, {:get_events_for_ast_node, ast_node_id})	3✔
145	end
146
147	#############################################################################
148	# GenServer Callbacks
149	#############################################################################
150
151	@impl true
152	def init(opts) do
153	Logger.info("🔧 RuntimeCorrelator.init starting with opts: #{inspect(opts)}")	8✔
154	Logger.info("🔍 Checking Config GenServer availability...")	8✔
155
156	config_pid = GenServer.whereis(ElixirScope.Config)	8✔
157	Logger.info("📍 Config GenServer PID: #{inspect(config_pid)}")	8✔
158
159	if config_pid do	8✔
160	Logger.info("✅ Config GenServer found, testing responsiveness...")	8✔
161	try do	8✔
162	config_result = ElixirScope.Config.get([:ast_repository])	8✔
163	Logger.info("✅ Config retrieved successfully: #{inspect(config_result)}")	8✔
164	rescue
165	error ->	×
166	Logger.error("❌ Config GenServer unresponsive: #{inspect(error)}")	×
167	Logger.error("📍 Error details: #{inspect(__STACKTRACE__)}")	×
168	{:stop, {:config_unresponsive, error}}
169	end
170	else
171	Logger.error("❌ Config GenServer not found!")	×
172	Logger.error("📍 Available registered processes: #{inspect(Process.registered())}")	×
173	{:stop, :config_not_found}
174	end
175
176	repository_pid = Keyword.get(opts, :repository_pid)	8✔
177	Logger.info("🏗️ Building config with repository_pid: #{inspect(repository_pid)}")	8✔
178	config = build_config(opts)	8✔
179
180	Logger.info("🔧 Creating correlator state...")	8✔
181	case create_correlator_state(repository_pid, config) do	8✔
182	{:ok, state} ->
183	# Schedule periodic cleanup
184	schedule_cleanup(state.cleanup_interval)	8✔
185
186	Logger.info("✅ RuntimeCorrelator started successfully with repository: #{inspect(repository_pid)}")	8✔
187	{:ok, state}
188
189	{:error, reason} ->
190	Logger.error("❌ Failed to initialize RuntimeCorrelator: #{inspect(reason)}")	×
191	{:stop, reason}
192	end
193	end
194
195	@impl true
196	def handle_call({:correlate_event, runtime_event}, _from, state) do
197	start_time = if state.performance_tracking, do: Utils.monotonic_timestamp(), else: nil	6✔
198
199	result = correlate_event_impl(state, runtime_event)	6✔
200
201	# Update statistics and performance tracking
202	new_state = update_correlation_stats(state, result, start_time)	6✔
203
204	{:reply, result, new_state}	6✔
205	end
206
207	@impl true
208	def handle_call({:correlate_events, runtime_events}, _from, state) do
209	start_time = if state.performance_tracking, do: Utils.monotonic_timestamp(), else: nil	1✔
210
211	results = Enum.map(runtime_events, &correlate_event_impl(state, &1))	1✔
212
213	# Extract successful correlations
214	successful_correlations = results	1✔
215	\|> Enum.filter(&match?({:ok, _}, &1))	5✔
216	\|> Enum.map(fn {:ok, {correlation_id, ast_node_id}} -> {correlation_id, ast_node_id} end)	×
217
218	# Update statistics
219	new_state = update_batch_correlation_stats(state, results, start_time)	1✔
220
221	{:reply, {:ok, successful_correlations}, new_state}	1✔
222	end
223
224	@impl true
225	def handle_call({:get_correlated_events, ast_node_id}, _from, state) do
226	result = get_correlated_events_impl(state, ast_node_id)	1✔
227	{:reply, result, state}	1✔
228	end
229
230	@impl true
231	def handle_call(:get_statistics, _from, state) do
232	stats = collect_statistics(state)	×
233	{:reply, {:ok, stats}, state}	×
234	end
235
236	@impl true
237	def handle_call(:health_check, _from, state) do
238	health = perform_health_check(state)	1✔
239	{:reply, {:ok, health}, state}	1✔
240	end
241
242	@impl true
243	def handle_call({:update_correlation_mapping, correlation_id, ast_node_id}, _from, state) do
244	:ets.insert(state.correlation_cache, {correlation_id, ast_node_id})	1✔
245	{:reply, :ok, state}	1✔
246	end
247
248	@impl true
249	def handle_call({:query_temporal_events, start_time, end_time}, _from, state) do
250	result = query_temporal_events_impl(state, start_time, end_time)	1✔
251	{:reply, result, state}	1✔
252	end
253
254	@impl true
255	def handle_call({:get_events_for_ast_node, ast_node_id}, _from, state) do
256	result = get_events_for_ast_node_impl(state, ast_node_id)	3✔
257	{:reply, result, state}	3✔
258	end
259
260	@impl true
261	def handle_info(:cleanup, state) do
262	new_state = perform_cleanup(state)	×
263	schedule_cleanup(state.cleanup_interval)	×
264	{:noreply, new_state}
265	end
266
267	@impl true
268	def handle_info(msg, state) do
269	Logger.debug("RuntimeCorrelator received unexpected message: #{inspect(msg)}")	×
270	{:noreply, state}
271	end
272
273	#############################################################################
274	# Private Implementation Functions
275	#############################################################################
276
277	defp build_config(opts) do
278	user_config = Keyword.get(opts, :config, %{})	8✔
279	Map.merge(@default_config, user_config)	8✔
280	end
281
282	defp create_correlator_state(repository_pid, config) do
283	if is_nil(repository_pid) do	8✔
284	{:error, :repository_pid_required}
285	else
286	try do	8✔
287	# Create ETS tables for caching and temporal indexing
288	correlation_cache = :ets.new(:correlation_cache, @cache_opts)	8✔
289	temporal_index = :ets.new(:temporal_index, @temporal_opts)	8✔
290
291	# Create DataAccess instance for event storage
292	{:ok, data_access} = DataAccess.new([name: :correlator_events])	8✔
293
294	state = %__MODULE__{	8✔
295	repository_pid: repository_pid,
296	data_access: data_access,
297	correlation_cache: correlation_cache,
298	temporal_index: temporal_index,
299	statistics: %{},
300	cache_size_limit: config.cache_size_limit,	8✔
301	correlation_timeout: config.correlation_timeout,	8✔
302	cleanup_interval: config.cleanup_interval,	8✔
303	performance_tracking: config.performance_tracking,	8✔
304	start_time: Utils.monotonic_timestamp(),
305	total_correlations: 0,
306	successful_correlations: 0,
307	failed_correlations: 0
308	}
309
310	{:ok, state}
311	rescue
312	error -> {:error, {:initialization_failed, error}}	×
313	end
314	end
315	end
316
317	defp correlate_event_impl(state, runtime_event) do
318	correlation_id = extract_correlation_id(runtime_event)	11✔
319
320	if correlation_id do	11✔
321	case lookup_correlation(state, correlation_id) do	11✔
322	{:ok, ast_node_id} ->
323	# Store the event for future analysis
324	store_correlated_event(state, runtime_event, ast_node_id)	3✔
325
326	# Update temporal index
327	timestamp = extract_timestamp(runtime_event)	3✔
328	:ets.insert(state.temporal_index, {timestamp, {correlation_id, ast_node_id}})	3✔
329
330	{:ok, {correlation_id, ast_node_id}}
331
332	{:error, :not_found} ->
333	# Try to get correlation from repository
334	case Repository.correlate_event(state.repository_pid, runtime_event) do	8✔
335	{:ok, ast_node_id} ->
336	# Cache the correlation for future use
337	:ets.insert(state.correlation_cache, {correlation_id, ast_node_id})	×
338
339	# Store the event
340	store_correlated_event(state, runtime_event, ast_node_id)	×
341
342	# Update temporal index
343	timestamp = extract_timestamp(runtime_event)	×
344	:ets.insert(state.temporal_index, {timestamp, {correlation_id, ast_node_id}})	×
345
346	{:ok, {correlation_id, ast_node_id}}
347
348	{:error, reason} ->	8✔
349	{:error, reason}
350	end
351
352	{:error, reason} ->	×
353	{:error, reason}
354	end
355	else
356	{:error, :no_correlation_id}
357	end
358	end
359
360	defp lookup_correlation(state, correlation_id) do
361	case :ets.lookup(state.correlation_cache, correlation_id) do	11✔
362	[{^correlation_id, ast_node_id}] -> {:ok, ast_node_id}	3✔
363	[] -> {:error, :not_found}	8✔
364	end
365	end
366
367	defp store_correlated_event(state, runtime_event, ast_node_id) do
368	# Enrich the event with correlation information
369	enriched_event = Map.merge(runtime_event, %{	3✔
370	correlated_ast_node_id: ast_node_id,
371	correlation_timestamp: Utils.monotonic_timestamp()
372	})
373
374	# Store in DataAccess for future queries
375	case DataAccess.store_event(state.data_access, enriched_event) do	3✔
376	:ok -> :ok	3✔
377	{:error, reason} ->
378	Logger.warning("Failed to store correlated event: #{inspect(reason)}")	×
379	:ok # Don't fail correlation due to storage issues
380	end
381	end
382
383	defp get_correlated_events_impl(state, ast_node_id) do
384	get_events_for_ast_node_impl(state, ast_node_id)	1✔
385	end
386
387	defp get_events_for_ast_node_impl(state, ast_node_id) do
388	try do	4✔
389	# Get all correlation IDs for this AST node from temporal index
390	correlation_ids = :ets.select(state.temporal_index, [	4✔
391	{{:_, {:'$1', ast_node_id}}, [], [:'$1']}
392	])
393
394	# Get events from DataAccess for each correlation ID
395	events = Enum.flat_map(correlation_ids, fn correlation_id ->	4✔
396	case DataAccess.query_by_correlation(state.data_access, correlation_id) do	3✔
397	{:ok, events} -> events	3✔
398	{:error, _} -> []	×
399	end
400	end)
401
402	# Sort by timestamp for chronological order
403	sorted_events = Enum.sort_by(events, fn event ->	4✔
404	extract_timestamp(event)	×
405	end)
406
407	{:ok, sorted_events}
408	rescue
409	error -> {:error, {:query_failed, error}}	×
410	end
411	end
412
413	defp query_temporal_events_impl(state, start_time, end_time) do
414	try do	1✔
415	# Query temporal index for events in the time range
416	temporal_entries = :ets.select(state.temporal_index, [	1✔
417	{{:'$1', :'$2'},
418	[{:andalso, {:>=, :'$1', start_time}, {:'=<', :'$1', end_time}}],
419	[:'$2']}
420	])
421
422	# Extract correlation IDs and get the actual events
423	correlation_ids = Enum.map(temporal_entries, fn {correlation_id, _ast_node_id} -> correlation_id end)	1✔
424
425	# Get events from DataAccess
426	events = Enum.flat_map(correlation_ids, fn correlation_id ->	1✔
427	case DataAccess.query_by_correlation(state.data_access, correlation_id) do	×
428	{:ok, events} -> events	×
429	{:error, _} -> []	×
430	end
431	end)
432
433	{:ok, events}
434	rescue
435	error -> {:error, {:temporal_query_failed, error}}	×
436	end
437	end
438
439	defp update_correlation_stats(state, result, start_time) do
440	new_total = state.total_correlations + 1	6✔
441
442	{new_successful, new_failed} = case result do	6✔
443	{:ok, _} -> {state.successful_correlations + 1, state.failed_correlations}	3✔
444	{:error, _} -> {state.successful_correlations, state.failed_correlations + 1}	3✔
445	end
446
447	# Update performance statistics if tracking is enabled
448	new_statistics = if state.performance_tracking and start_time do	6✔
449	duration = Utils.monotonic_timestamp() - start_time	6✔
450	update_performance_stats(state.statistics, duration)	6✔
451	else
452	state.statistics	×
453	end
454
455	%{state \|	6✔
456	total_correlations: new_total,
457	successful_correlations: new_successful,
458	failed_correlations: new_failed,
459	statistics: new_statistics
460	}
461	end
462
463	defp update_batch_correlation_stats(state, results, start_time) do
464	successful_count = Enum.count(results, &match?({:ok, _}, &1))	1✔
465	failed_count = length(results) - successful_count	1✔
466
467	new_total = state.total_correlations + length(results)	1✔
468	new_successful = state.successful_correlations + successful_count	1✔
469	new_failed = state.failed_correlations + failed_count	1✔
470
471	# Update performance statistics if tracking is enabled
472	new_statistics = if state.performance_tracking and start_time do	1✔
473	duration = Utils.monotonic_timestamp() - start_time	1✔
474	update_batch_performance_stats(state.statistics, duration, length(results))	1✔
475	else
476	state.statistics	×
477	end
478
479	%{state \|	1✔
480	total_correlations: new_total,
481	successful_correlations: new_successful,
482	failed_correlations: new_failed,
483	statistics: new_statistics
484	}
485	end
486
487	defp update_performance_stats(statistics, duration) do
488	current_avg = Map.get(statistics, :average_correlation_time, 0.0)	6✔
489	current_count = Map.get(statistics, :correlation_count, 0)	6✔
490
491	new_count = current_count + 1	6✔
492	new_avg = (current_avg * current_count + duration) / new_count	6✔
493
494	Map.merge(statistics, %{	6✔
495	average_correlation_time: new_avg,
496	correlation_count: new_count,
497	last_correlation_time: duration
498	})
499	end
500
501	defp update_batch_performance_stats(statistics, duration, batch_size) do
502	current_avg = Map.get(statistics, :average_batch_correlation_time, 0.0)	1✔
503	current_count = Map.get(statistics, :batch_correlation_count, 0)	1✔
504
505	new_count = current_count + 1	1✔
506	new_avg = (current_avg * current_count + duration) / new_count	1✔
507
508	Map.merge(statistics, %{	1✔
509	average_batch_correlation_time: new_avg,
510	batch_correlation_count: new_count,
511	last_batch_correlation_time: duration,
512	last_batch_size: batch_size
513	})
514	end
515
516	defp collect_statistics(state) do
517	uptime = Utils.monotonic_timestamp() - state.start_time	×
518	success_rate = if state.total_correlations > 0 do	×
519	state.successful_correlations / state.total_correlations	×
520	else
521	0.0
522	end
523
524	%{	×
525	uptime_ms: uptime,
526	total_correlations: state.total_correlations,	×
527	successful_correlations: state.successful_correlations,	×
528	failed_correlations: state.failed_correlations,	×
529	success_rate: success_rate,
530	cache_size: :ets.info(state.correlation_cache, :size),	×
531	temporal_index_size: :ets.info(state.temporal_index, :size),	×
532	performance_stats: state.statistics	×
533	}
534	end
535
536	defp perform_health_check(state) do
537	cache_size = :ets.info(state.correlation_cache, :size)	1✔
538	_temporal_size = :ets.info(state.temporal_index, :size)	1✔
539
540	status = cond do	1✔
541	cache_size > state.cache_size_limit * 0.9 -> :warning	1✔
542	not Process.alive?(state.repository_pid) -> :error	1✔
543	true -> :healthy	1✔
544	end
545
546	%{	1✔
547	status: status,
548	uptime_ms: Utils.monotonic_timestamp() - state.start_time,	1✔
549	cache_utilization: cache_size / state.cache_size_limit,	1✔
550	repository_alive: Process.alive?(state.repository_pid),	1✔
551	memory_usage: %{
552	correlation_cache: :ets.info(state.correlation_cache, :memory),	1✔
553	temporal_index: :ets.info(state.temporal_index, :memory)	1✔
554	}
555	}
556	end
557
558	defp perform_cleanup(state) do
559	# Clean up old entries from cache if it's getting too large
560	cache_size = :ets.info(state.correlation_cache, :size)	×
561
562	if cache_size > state.cache_size_limit do	×
563	# Simple cleanup: remove oldest 10% of entries
564	# In practice, you'd want a more sophisticated LRU strategy
565	entries_to_remove = div(cache_size, 10)	×
566
567	# Get all keys and remove the first N (this is a simplified approach)
568	all_keys = :ets.select(state.correlation_cache, [{{:'$1', :_}, [], [:'$1']}])	×
569	keys_to_remove = Enum.take(all_keys, entries_to_remove)	×
570
571	Enum.each(keys_to_remove, fn key ->	×
572	:ets.delete(state.correlation_cache, key)	×
573	end)
574
575	Logger.debug("RuntimeCorrelator cleanup: removed #{entries_to_remove} cache entries")	×
576	end
577
578	state	×
579	end
580
581	defp schedule_cleanup(interval) do
582	Process.send_after(self(), :cleanup, interval)	8✔
583	end
584
585	defp extract_correlation_id(%{correlation_id: correlation_id}), do: correlation_id	11✔
586	defp extract_correlation_id(%{"correlation_id" => correlation_id}), do: correlation_id	×
587	defp extract_correlation_id(_), do: nil	×
588
589	defp extract_timestamp(%{timestamp: timestamp}), do: timestamp	3✔
590	defp extract_timestamp(%{"timestamp" => timestamp}), do: timestamp	×
591	defp extract_timestamp(_), do: Utils.monotonic_timestamp()	×
592	end

nshkrdotcom / ElixirScope / 19e103dafc2ed3bdd481df7640a8c5b1131dc814

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous