Module: Legion::Data::Connection

Extended by:

Logging::Helper

Defined in:

lib/legion/data/connection.rb

Defined Under Namespace

Classes: QueryFileLogger, SegmentedTaggedLogger, SlowQueryLogger, UnresolvedCredentialError

Constant Summary

ADAPTERS =

%i[sqlite mysql2 postgres].freeze

UNRESOLVED_URI_PATTERN =

%r{\A(?:vault|env|lease)://}

GENERIC_KEYS =

%i[max_connections pool_timeout preconnect single_threaded test name].freeze

ADAPTER_KEYS =

{
  sqlite:   %i[timeout readonly disable_dqs],
  postgres: %i[connect_timeout sslmode sslrootcert search_path],
  mysql2:   %i[connect_timeout read_timeout write_timeout encoding sql_mode]
}.freeze

ADAPTER_DEFAULTS =

{
  sqlite:   { timeout: 5000, readonly: false, disable_dqs: true },
  postgres: { connect_timeout: 20, sslmode: 'disable' },
  mysql2:   { connect_timeout: 120, encoding: 'utf8mb4' }
}.freeze

QUERY_LOG_DIR =

File.expand_path('~/.legionio/logs').freeze

Class Attribute Summary

• .sequel ⇒ Object

  Returns the value of attribute sequel.

Class Method Summary

• .adapter ⇒ Object
• .connect_with_replicas ⇒ Object
• .connection_info ⇒ Object

  Returns connection metadata for health checks and diagnostics.

• .creds_builder(final_creds = {}) ⇒ Object
• .expire_all_pooled_connections ⇒ Object
• .fallback_active? ⇒ Boolean

  Returns true if the data layer fell back to SQLite from a configured network database (PostgreSQL/MySQL).

• .merge_tls_creds(creds, adapter:, port:) ⇒ Object
• .pool_stats ⇒ Object
• .read_server ⇒ Object
• .reconnect_with_fresh_creds ⇒ Object
• .replica_servers ⇒ Object
• .setup ⇒ Object
• .shutdown ⇒ Object
• .stats ⇒ Object

Methods included from Logging::Helper

handle_exception

Class Attribute Details

.sequel ⇒ Object

Returns the value of attribute sequel.

# File 'lib/legion/data/connection.rb', line 162

def sequel
  @sequel
end

Class Method Details

.adapter ⇒ Object

# File 'lib/legion/data/connection.rb', line 164

def adapter
  @adapter ||= Legion::Settings[:data][:adapter]&.to_sym || :sqlite
end

.connect_with_replicas ⇒ Object

# File 'lib/legion/data/connection.rb', line 324

def connect_with_replicas
  return unless adapter == :postgres

  replica_url  = Legion::Settings[:data][:read_replica_url]
  replica_list = Array(Legion::Settings[:data][:replicas]).dup

  replica_list.prepend(replica_url) if replica_url && !replica_url.empty?
  replica_list.uniq!
  replica_list.compact!

  return if replica_list.empty?

  @sequel.extension(:server_block)

  replica_list.each_with_index do |url, idx|
    @sequel.add_servers("read_#{idx}": url)
  end

  @replica_servers = replica_list.each_with_index.map { |_, idx| :"read_#{idx}" }
  log.debug "Registered #{@replica_servers.size} read replica(s)"
end

.connection_info ⇒ Object

Returns connection metadata for health checks and diagnostics. Apollo and other services can use this to detect silent fallback.

# File 'lib/legion/data/connection.rb', line 203

def connection_info
  {
    adapter:            adapter,
    connected:          Legion::Settings[:data][:connected],
    fallback_active:    @fallback_active || false,
    configured_adapter: Legion::Settings[:data][:adapter]&.to_sym || :sqlite,
    sequel_alive:       (begin
      !@sequel&.test_connection.nil?
    rescue StandardError => e
      log.debug("connection health check failed: #{e.message}")
      false
    end)
  }
end

.creds_builder(final_creds = {}) ⇒ Object

# File 'lib/legion/data/connection.rb', line 382

def creds_builder(final_creds = {})
  final_creds.merge! Legion::Data::Settings.creds(adapter)
  final_creds.merge! Legion::Settings[:data][:creds] if Legion::Settings[:data][:creds].is_a? Hash

  port = final_creds[:port]
  merge_tls_creds(final_creds, adapter: adapter, port: port)

  final_creds
end

.expire_all_pooled_connections ⇒ Object

# File 'lib/legion/data/connection.rb', line 314

def expire_all_pooled_connections
  pool = @sequel.pool
  return unless pool.instance_variable_defined?(:@connection_expiration_timestamps)

  timestamps = pool.instance_variable_get(:@connection_expiration_timestamps)
  timestamps.each_key { |conn| timestamps[conn] = [0, 0].freeze }
rescue StandardError => e
  handle_exception(e, level: :warn, handled: true, operation: :expire_all_pooled_connections)
end

.fallback_active? ⇒ Boolean

Returns true if the data layer fell back to SQLite from a configured network database (PostgreSQL/MySQL). Services should check this and log warnings when operating in degraded mode.

Returns:

• (Boolean)

# File 'lib/legion/data/connection.rb', line 221

def fallback_active?
  @fallback_active == true
end

.merge_tls_creds(creds, adapter:, port:) ⇒ Object

# File 'lib/legion/data/connection.rb', line 356

def merge_tls_creds(creds, adapter:, port:)
  return creds if adapter == :sqlite
  return creds unless defined?(Legion::Crypt::TLS)

  tls_settings = data_tls_settings
  return creds unless tls_settings[:enabled] == true

  tls = Legion::Crypt::TLS.resolve(tls_settings, port: port)
  return creds unless tls[:enabled]

  case adapter
  when :postgres
    creds[:sslmode]     = tls[:verify] == :none ? 'require' : 'verify-full'
    creds[:sslrootcert] = tls[:ca] if tls[:ca]
    creds[:sslcert]     = tls[:cert] if tls[:cert]
    creds[:sslkey]      = tls[:key] if tls[:key]
  when :mysql2
    creds[:ssl_mode] = tls[:verify] == :none ? 'required' : 'verify_identity'
    creds[:sslca]    = tls[:ca] if tls[:ca]
    creds[:sslcert]  = tls[:cert] if tls[:cert]
    creds[:sslkey]   = tls[:key] if tls[:key]
  end

  creds
end

.pool_stats ⇒ Object

# File 'lib/legion/data/connection.rb', line 241

def pool_stats
  return {} unless @sequel

  pool = @sequel.pool
  stats = {
    type:     pool.pool_type,
    size:     pool.size,
    max_size: pool.respond_to?(:max_size) ? pool.max_size : nil
  }

  case pool.pool_type
  when :timed_queue, :sharded_timed_queue
    queue_size = pool.instance_variable_get(:@queue)&.size || 0
    stats[:available] = queue_size
    stats[:in_use]    = stats[:size] - queue_size
    stats[:waiting]   = pool.num_waiting
  when :threaded, :sharded_threaded
    avail = pool.instance_variable_get(:@available_connections)
    stats[:available] = avail&.size || 0
    stats[:in_use]    = stats[:size] - stats[:available]
    stats[:waiting]   = pool.num_waiting
  when :single, :sharded_single
    stats[:available] = pool.size
    stats[:in_use]    = 0
    stats[:waiting]   = 0
  end

  stats.compact
rescue StandardError => e
  handle_exception(e, level: :warn, handled: true, operation: :data_pool_stats, adapter: adapter)
  {}
end

.read_server ⇒ Object

# File 'lib/legion/data/connection.rb', line 346

def read_server
  return :default if @replica_servers.nil? || @replica_servers.empty?

  :read_0
end

.reconnect_with_fresh_creds ⇒ Object

# File 'lib/legion/data/connection.rb', line 283

def reconnect_with_fresh_creds
  return false unless @sequel
  return false if adapter == :sqlite

  fresh_creds = Legion::Settings[:data][:creds]
  return false unless fresh_creds.is_a?(Hash)

  new_user = fresh_creds[:user] || fresh_creds[:username]
  new_pass = fresh_creds[:password]

  unless new_user && new_pass
    log.error('reconnect_with_fresh_creds: no user/password in Settings[:data][:creds]')
    return false
  end

  old_user = @sequel.opts[:user]
  @sequel.opts[:user] = new_user
  @sequel.opts[:password] = new_pass

  @sequel.disconnect
  expire_all_pooled_connections

  @sequel.test_connection
  log.info("reconnect_with_fresh_creds: rotated credentials (#{old_user} → #{new_user})")
  true
rescue StandardError => e
  handle_exception(e, level: :error, handled: true, operation: :reconnect_with_fresh_creds,
                   old_user: old_user, new_user: new_user)
  false
end

.replica_servers ⇒ Object

# File 'lib/legion/data/connection.rb', line 352

def replica_servers
  @replica_servers || []
end

.setup ⇒ Object

# File 'lib/legion/data/connection.rb', line 168

def setup
  @adapter = Legion::Settings[:data][:adapter]&.to_sym || :sqlite
  opts = sequel_opts
  log.info("Legion::Data::Connection setup adapter=#{adapter}")
  @fallback_active = false
  @sequel = if adapter == :sqlite
              ::Sequel.connect(opts.merge(adapter: :sqlite, database: sqlite_path))
            else
              attempted_adapter = adapter
              begin
                ::Sequel.connect(connection_opts_for(adapter: attempted_adapter, opts: opts))
              rescue UnresolvedCredentialError => e
                handle_exception(e, level: :fatal, handled: false, operation: :data_connect)
                raise
              rescue StandardError => e
                raise unless dev_fallback?

                log.error("Legion::Data FALLING BACK TO SQLITE — #{attempted_adapter} network DB connection failed: #{e.message}")
                log.error("Legion::Data WARNING: Data written to SQLite will NOT be visible when #{attempted_adapter} reconnects. " \
                          'Apollo knowledge, audit logs, and other DB-backed services will use a local-only store.')
                handle_exception(e, level: :error, handled: true, operation: :shared_connect, fallback: :sqlite)
                @adapter = :sqlite
                @fallback_active = true
                sqlite_opts = sequel_opts
                ::Sequel.connect(sqlite_opts.merge(adapter: :sqlite, database: sqlite_path))
              end
            end
  Legion::Settings[:data][:connected] = true
  log_connection_info
  configure_extensions
  connect_with_replicas
end

.shutdown ⇒ Object

# File 'lib/legion/data/connection.rb', line 274

def shutdown
  @sequel&.disconnect
  @query_file_logger&.close
  @query_file_logger = nil
  @fallback_active = false
  Legion::Settings[:data][:connected] = false
  log.info 'Legion::Data connection closed'
end

.stats ⇒ Object

# File 'lib/legion/data/connection.rb', line 225

def stats
  return { connected: false } unless @sequel

  data = Legion::Settings[:data]
  {
    connected: data[:connected],
    adapter:   adapter,
    pool:      pool_stats,
    tuning:    tuning_stats(data),
    database:  database_stats
  }
rescue StandardError => e
  handle_exception(e, level: :warn, handled: true, operation: :data_connection_stats, adapter: adapter)
  { connected: (data[:connected] if data.is_a?(Hash)), adapter: adapter, error: e.message }
end