truf-server/app/db_backend.py

import os
import re
import sqlite3
import time
from urllib.parse import parse_qsl, quote, unquote, urlencode, urlsplit, urlunsplit


POSTGRES_SCHEMES = ('postgresql://', 'postgres://')
DEFAULT_POSTGRES_CONNECT_TIMEOUT_SEC = 10
DEFAULT_POSTGRES_STATEMENT_TIMEOUT_MS = 30000
DEFAULT_POSTGRES_LOCK_TIMEOUT_MS = 10000
DEFAULT_POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS = 30000
DEFAULT_POSTGRES_TCP_USER_TIMEOUT_MS = 30000
POSTGRES_APPLICATION_SCHEMA = 'public'
POSTGRES_CHILD_START_RETRY_ATTEMPTS = 3
POSTGRES_CHILD_START_RETRY_MARKERS = (
    'server closed the connection unexpectedly',
    'connection reset by peer',
    'connection was forcibly closed by the remote host',
)
HOST_AGENT_POSTGRES_SOCKET_DIRECTORY = '/run/truf-postgres'
HOST_AGENT_POSTGRES_DATABASE = 'truf'
HOST_AGENT_POSTGRES_USER = 'truf'
HOST_AGENT_POSTGRES_PORT = 5432


class DatabaseUrlError(ValueError):
    pass


def is_postgres_url(value):
    return str(value or '').strip().lower().startswith(POSTGRES_SCHEMES)


def database_url_from_env():
    # Supervised processes receive TRUF_MANAGED_POSTGRES_DSN as the canonical
    # connection authority. Use that same precedence everywhere that derives
    # an endpoint identity or opens a managed connection.
    return (
        os.getenv('TRUF_MANAGED_POSTGRES_DSN')
        or os.getenv('SCANNER_DB_URL')
        or os.getenv('DATABASE_URL')
    )


def parse_postgres_url(value):
    """Parse one URL-form libpq DSN without allowing alternate authorities."""
    text = str(value or '').strip()
    if not text or any(character in text for character in ('\x00', '\r', '\n')):
        raise DatabaseUrlError('invalid PostgreSQL database URL')
    try:
        parsed = urlsplit(text)
        port = parsed.port or 5432
        host = parsed.hostname or ''
        username = unquote(parsed.username or '')
        password = unquote(parsed.password or '')
        database = unquote((parsed.path or '')[1:]) if (parsed.path or '').startswith('/') else ''
    except (TypeError, ValueError) as exc:
        raise DatabaseUrlError('invalid PostgreSQL database URL') from exc
    if parsed.scheme.lower() not in ('postgresql', 'postgres'):
        raise DatabaseUrlError('database URL must use the PostgreSQL scheme')
    if parsed.query:
        raise DatabaseUrlError('PostgreSQL database URL query parameters are forbidden')
    if parsed.fragment:
        raise DatabaseUrlError('PostgreSQL database URL fragments are forbidden')
    if not parsed.netloc or not host or not username or not database:
        raise DatabaseUrlError('PostgreSQL database URL must include one host, user, and database')
    authority = parsed.netloc.rsplit('@', 1)[-1]
    decoded_authority = unquote(authority)
    if (
        ',' in decoded_authority
        or any(character.isspace() for character in decoded_authority)
        or '%' in authority
        or any(character in host for character in (',', '/', '\\', '\x00'))
    ):
        raise DatabaseUrlError('PostgreSQL database URL must contain exactly one literal host authority')
    if not 0 < int(port) <= 65535:
        raise DatabaseUrlError('PostgreSQL database URL port is invalid')
    if any(character in database for character in ('/', '\\', '?', '#', '\x00')):
        raise DatabaseUrlError('PostgreSQL database URL database name contains encoded authority syntax')
    if parsed.path.count('/') != 1:
        raise DatabaseUrlError('PostgreSQL database URL must contain exactly one database path segment')
    return {
        'parsed': parsed,
        'host': host.lower(),
        'port': int(port),
        'database': database,
        'user': username,
        'password': password,
    }


def canonical_postgres_url(value, database, user, port, host='127.0.0.1'):
    """Return one libpq URL whose endpoint cannot be redirected by DSN options."""
    values = parse_postgres_url(value)
    expected_host = str(host).lower()
    if values['host'] != expected_host or values['port'] != int(port):
        raise DatabaseUrlError('PostgreSQL database URL endpoint does not match managed cluster authority')
    if values['database'] != str(database) or values['user'] != str(user):
        raise DatabaseUrlError('PostgreSQL database URL identity does not match managed cluster authority')
    credentials = quote(str(user), safe='')
    if values['parsed'].password is not None:
        credentials += ':' + quote(values['password'], safe='')
    netloc = f'{credentials}@{expected_host}:{int(port)}'
    return urlunsplit(('postgresql', netloc, '/' + quote(str(database), safe=''), '', ''))


def redact_database_url(value):
    text = str(value or '')
    if not is_postgres_url(text):
        if text.strip().lower().startswith(('postgres', 'postgre')):
            return 'postgresql://***'
        if '://' in text:
            return text.split('://', 1)[0] + '://***'
        return text
    try:
        parsed = urlsplit(text)
        username = parsed.username or ''
        host = parsed.hostname or ''
        port = f':{parsed.port}' if parsed.port else ''
        netloc = parsed.netloc
        if parsed.password:
            netloc = f'{username}:***@{host}{port}' if username else f'***@{host}{port}'
        query = []
        for key, value in parse_qsl(parsed.query, keep_blank_values=True):
            key_lower = key.lower()
            sensitive = any(part in key_lower for part in ('password', 'passwd', 'pwd', 'token', 'secret', 'credential'))
            query.append((key, '***' if sensitive else value))
        return urlunsplit((parsed.scheme, netloc, parsed.path, urlencode(query), parsed.fragment))
    except Exception:
        return 'postgresql://***'


def _split_sql_script(script):
    statements = []
    current = []
    quote = None
    escape = False
    for char in str(script or ''):
        current.append(char)
        if escape:
            escape = False
            continue
        if char == '\\':
            escape = True
            continue
        if quote:
            if char == quote:
                quote = None
            continue
        if char in ("'", '"'):
            quote = char
            continue
        if char == ';':
            statement = ''.join(current).strip()
            if statement:
                statements.append(statement[:-1].strip())
            current = []
    tail = ''.join(current).strip()
    if tail:
        statements.append(tail)
    return [statement for statement in statements if statement]


def _convert_qmark_to_psycopg(sql):
    out = []
    quote = None
    escape = False
    for char in str(sql or ''):
        if escape:
            out.append(char)
            escape = False
            continue
        if char == '\\':
            out.append(char)
            escape = True
            continue
        if quote:
            out.append('%%' if char == '%' else char)
            if char == quote:
                quote = None
            continue
        if char in ("'", '"'):
            out.append(char)
            quote = char
            continue
        if char == '?':
            out.append('%s')
        elif char == '%':
            out.append('%%')
        else:
            out.append(char)
    return ''.join(out)


def _postgres_schema_sql(script):
    converted = str(script or '').replace(
        'INTEGER PRIMARY KEY AUTOINCREMENT',
        'BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY',
    )
    # PostgreSQL cannot create either side of the target_queue/target_scans
    # cycle with both inline FKs. The offline migration adds this edge after
    # both tables exist; SQLite can retain it in the base schema.
    converted = converted.replace(
        ',\n    FOREIGN KEY(queue_id) REFERENCES target_queue(id)',
        '',
    )
    for future_foreign_key in (
        ',\n    FOREIGN KEY(result_reservation_id) REFERENCES result_reservations(id)',
        ',\n    FOREIGN KEY(reservation_id) REFERENCES result_reservations(id)',
        ',\n    FOREIGN KEY(current_result_reservation_id) REFERENCES result_reservations(id)',
        ',\n    FOREIGN KEY(candidate_id) REFERENCES keycheck_candidates(id)',
        ',\n    FOREIGN KEY(credential_id) REFERENCES keycheck_credentials(id)',
        ',\n    FOREIGN KEY(last_append_id) REFERENCES projection_appends(id)',
        ',\n    FOREIGN KEY(experiment_id) REFERENCES docker_depth_experiments(id)',
    ):
        converted = converted.replace(future_foreign_key, '')
    for column in (
        'run_id', 'cycle_id', 'target_scan_id', 'finding_id', 'keycheck_result_id',
        'last_run_id', 'last_cycle_id', 'queue_id', 'byte_offset', 'line_number',
        'current_result_reservation_id', 'result_reservation_id', 'reservation_id',
        'projection_job_id', 'keycheck_candidate_id', 'candidate_id', 'credential_id',
        'keycheck_result_id', 'target_scan_id', 'job_id', 'last_append_id',
        'last_job_id', 'last_result_id', 'object_id', 'lease_reservation_id',
        'covered_reservation_id', 'declared_bytes', 'verified_bytes',
        'experiment_id', 'pass_id', 'page_id', 'source_cycle_id', 'retry_work_id',
        'repository_queue_id', 'first_cycle_id', 'last_cycle_id', 'first_page_id',
        'last_page_id', 'target_queue_id', 'manifest_id', 'manifest_layer_id',
        'eligibility_page_id', 'experiment_repository_id', 'experiment_target_id',
        'scan_binding_id', 'manifest_size_bytes', 'layer_size_bytes',
        'fence_generation', 'resolver_generation', 'dispatch_order', 'total_count',
        'user_id', 'remote_user_id', 'remote_device_id',
        'expected_revision', 'resulting_revision', 'revision',
        'before_bytes', 'after_bytes', 'previous_event_id',
    ):
        converted = converted.replace(f'{column} INTEGER', f'{column} BIGINT')
    converted = converted.replace(
        'CREATE VIEW IF NOT EXISTS keycheck_latest_state AS',
        'CREATE OR REPLACE VIEW keycheck_latest_state AS',
    )
    return converted


def _sqlite_check_constraints(sql):
    text = str(sql or '')
    constraints = {}
    index = 0
    position = 0
    quote = None
    while position < len(text):
        char = text[position]
        if quote:
            if char == quote:
                if position + 1 < len(text) and text[position + 1] == quote:
                    position += 2
                    continue
                quote = None
            position += 1
            continue
        if char in ("'", '"', '`'):
            quote = char
            position += 1
            continue
        if (
            text[position:position + 5].lower() != 'check'
            or (position and (text[position - 1].isalnum() or text[position - 1] == '_'))
            or (
                position + 5 < len(text)
                and (text[position + 5].isalnum() or text[position + 5] == '_')
            )
        ):
            position += 1
            continue
        opening = position + 5
        while opening < len(text) and text[opening].isspace():
            opening += 1
        if opening >= len(text) or text[opening] != '(':
            position += 5
            continue
        depth = 1
        closing = opening + 1
        expression_quote = None
        while closing < len(text) and depth:
            current = text[closing]
            if expression_quote:
                if current == expression_quote:
                    if closing + 1 < len(text) and text[closing + 1] == expression_quote:
                        closing += 2
                        continue
                    expression_quote = None
            elif current in ("'", '"', '`'):
                expression_quote = current
            elif current == '(':
                depth += 1
            elif current == ')':
                depth -= 1
            closing += 1
        if depth:
            break
        prefix = text[:position]
        named = re.search(
            r'\bCONSTRAINT\s+(?:"([A-Za-z_][A-Za-z0-9_$]*)"|'
            r'([A-Za-z_][A-Za-z0-9_$]*))\s*$',
            prefix,
            re.IGNORECASE,
        )
        name = (named.group(1) or named.group(2)) if named else f'__unnamed_check_{index}'
        expression = text[opening + 1:closing - 1]
        constraint = {
            'expression': expression,
            'definition': f'CHECK ({expression})',
            'valid': True,
        }
        if name in constraints:
            constraints[name]['valid'] = False
            name = f'__duplicate_check_{index}_{name}'
            constraint['valid'] = False
        constraints[name] = constraint
        index += 1
        position = closing
    return constraints


class DatabaseConnection:
    def __init__(self, dialect, conn, application_schema=None):
        self.dialect = dialect
        self._conn = conn
        self.application_schema = application_schema if dialect == 'postgres' else None

    @property
    def is_postgres(self):
        return self.dialect == 'postgres'

    @property
    def is_sqlite(self):
        return self.dialect == 'sqlite'

    def execute(self, sql, params=None):
        params = tuple(params or ())
        if self.is_postgres:
            params = tuple(value.replace('\x00', '') if isinstance(value, str) else value for value in params)
            cur = self._conn.cursor()
            cur.execute(_convert_qmark_to_psycopg(sql), params)
            return cur
        return self._conn.execute(sql, params)

    def executescript(self, script):
        if self.is_sqlite:
            return self._conn.executescript(script)
        for statement in _split_sql_script(_postgres_schema_sql(script)):
            self.execute(statement)
        return None

    def commit(self):
        return self._conn.commit()

    def rollback(self):
        return self._conn.rollback()

    def close(self):
        return self._conn.close()

    def table_columns(self, table):
        if self.is_postgres:
            rows = self.execute(
                '''SELECT a.attname AS name
                   FROM pg_catalog.pg_attribute a
                   JOIN pg_catalog.pg_class c ON c.oid = a.attrelid
                   JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
                    WHERE n.nspname = ?
                      AND c.relname = ?
                     AND a.attnum > 0
                     AND NOT a.attisdropped''',
                (self.application_schema, table),
            ).fetchall()
            return {row['name'] for row in rows}
        return {row['name'] for row in self.execute(f'PRAGMA table_info({table})').fetchall()}

    def table_column_details(self, table):
        if self.is_postgres:
            rows = self.execute(
                '''SELECT a.attname AS name,
                          pg_catalog.format_type(a.atttypid, a.atttypmod) AS type,
                          a.attnotnull AS not_null,
                          a.attidentity AS identity_generation,
                          a.attgenerated AS generated_kind,
                          pg_catalog.pg_get_expr(d.adbin, d.adrelid) AS default_sql,
                          (d.oid IS NOT NULL) AS has_default,
                          pg_catalog.pg_get_serial_sequence(
                              pg_catalog.quote_ident(n.nspname) || '.' || pg_catalog.quote_ident(c.relname),
                              a.attname
                          ) AS sequence_name,
                          COALESCE(i.indisprimary, false) AS primary_key
                    FROM pg_catalog.pg_attribute a
                    JOIN pg_catalog.pg_class c ON c.oid = a.attrelid
                    JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
                    LEFT JOIN pg_catalog.pg_attrdef d ON d.adrelid = a.attrelid AND d.adnum = a.attnum
                    LEFT JOIN pg_catalog.pg_index i ON i.indrelid = a.attrelid
                       AND i.indisprimary AND a.attnum = ANY(i.indkey)
                    WHERE n.nspname = ?
                      AND c.relname = ?
                     AND a.attnum > 0
                     AND NOT a.attisdropped
                   ORDER BY a.attnum''',
                (self.application_schema, table),
            ).fetchall()
            return {
                row['name']: {
                    'type': str(row['type'] or '').lower(),
                    'not_null': bool(row['not_null']),
                    'default': str(row['default_sql'] or ''),
                    'has_default': bool(row['has_default']),
                    'primary_key': bool(row['primary_key']),
                    'identity': str(row['identity_generation'] or ''),
                    'generated': str(row['generated_kind'] or ''),
                    'sequence': str(row['sequence_name'] or ''),
                }
                for row in rows
            }
        rows = self.execute(f'PRAGMA table_info({table})').fetchall()
        return {
            row['name']: {
                'type': str(row['type'] or '').lower(),
                'not_null': bool(row['notnull']) or bool(row['pk']),
                'default': str(row['dflt_value'] or ''),
                'has_default': row['dflt_value'] is not None,
                'primary_key': bool(row['pk']),
                'identity': '',
                'generated': '',
                'sequence': '',
            }
            for row in rows
        }

    def table_indexes(self, table):
        if self.is_postgres:
            rows = self.execute(
                '''SELECT idx.relname AS name,
                          i.indisunique AS is_unique,
                          i.indisprimary AS is_primary,
                          i.indisvalid AS is_valid,
                          i.indisready AS is_ready,
                          i.indislive AS is_live,
                          pg_catalog.pg_get_expr(i.indpred, i.indrelid) AS predicate,
                          ARRAY(
                              SELECT pg_catalog.pg_get_indexdef(i.indexrelid, position, true)
                               FROM pg_catalog.generate_series(1, i.indnkeyatts) AS position
                              ORDER BY position
                          ) AS columns,
                          pg_catalog.pg_get_indexdef(i.indexrelid) AS sql
                    FROM pg_catalog.pg_index i
                    JOIN pg_catalog.pg_class tbl ON tbl.oid = i.indrelid
                    JOIN pg_catalog.pg_namespace n ON n.oid = tbl.relnamespace
                    JOIN pg_catalog.pg_class idx ON idx.oid = i.indexrelid
                    WHERE n.nspname = ? AND tbl.relname = ?''',
                (self.application_schema, table),
            ).fetchall()
            return {
                row['name']: {
                    'unique': bool(row['is_unique']),
                    'primary': bool(row['is_primary']),
                    'valid': bool(row['is_valid']),
                    'ready': bool(row['is_ready']),
                    'live': bool(row['is_live']),
                    'predicate': str(row['predicate'] or ''),
                    'columns': [str(value).strip('"') for value in (row['columns'] or [])],
                    'sql': str(row['sql'] or ''),
                }
                for row in rows
            }
        output = {}
        for row in self.execute(f'PRAGMA index_list({table})').fetchall():
            name = row['name']
            columns = [item['name'] for item in self.execute(f'PRAGMA index_info({name})').fetchall()]
            sql_row = self.execute(
                "SELECT sql FROM sqlite_master WHERE type = 'index' AND name = ?",
                (name,),
            ).fetchone()
            sql = str(sql_row['sql'] if sql_row else '')
            predicate = sql.split(' WHERE ', 1)[1] if ' WHERE ' in sql.upper() else ''
            if ' WHERE ' in sql.upper():
                position = sql.upper().index(' WHERE ')
                predicate = sql[position + 7:]
            output[name] = {
                'unique': bool(row['unique']),
                'primary': str(row['origin'] or '') == 'pk',
                'valid': True,
                'ready': True,
                'live': True,
                'predicate': predicate,
                'columns': columns,
                'sql': sql,
            }
        return output

    def table_foreign_keys(self, table):
        if self.is_postgres:
            rows = self.execute(
                '''SELECT con.conname AS name,
                          ARRAY(
                              SELECT src.attname
                              FROM pg_catalog.unnest(con.conkey) WITH ORDINALITY AS keys(attnum, position)
                              JOIN pg_catalog.pg_attribute src
                                ON src.attrelid = con.conrelid AND src.attnum = keys.attnum
                              ORDER BY keys.position
                          ) AS columns,
                          ref_n.nspname AS referenced_schema,
                          ref.relname AS referenced_table,
                          ARRAY(
                              SELECT dst.attname
                              FROM pg_catalog.unnest(con.confkey) WITH ORDINALITY AS keys(attnum, position)
                              JOIN pg_catalog.pg_attribute dst
                                ON dst.attrelid = con.confrelid AND dst.attnum = keys.attnum
                              ORDER BY keys.position
                          ) AS referenced_columns,
                          con.confupdtype::text AS update_action,
                          con.confdeltype::text AS delete_action,
                          con.convalidated AS is_valid
                   FROM pg_catalog.pg_constraint con
                   JOIN pg_catalog.pg_class tbl ON tbl.oid = con.conrelid
                   JOIN pg_catalog.pg_namespace n ON n.oid = tbl.relnamespace
                   JOIN pg_catalog.pg_class ref ON ref.oid = con.confrelid
                   JOIN pg_catalog.pg_namespace ref_n ON ref_n.oid = ref.relnamespace
                  WHERE con.contype = 'f' AND n.nspname = ? AND tbl.relname = ?''',
                (self.application_schema, table),
            ).fetchall()
            action_names = {
                'a': 'NO ACTION', 'r': 'RESTRICT', 'c': 'CASCADE',
                'n': 'SET NULL', 'd': 'SET DEFAULT',
            }
            return {
                row['name']: {
                    'columns': [str(value) for value in (row['columns'] or [])],
                    'referenced_schema': str(row['referenced_schema'] or ''),
                    'referenced_table': str(row['referenced_table'] or ''),
                    'referenced_columns': [str(value) for value in (row['referenced_columns'] or [])],
                    'update_action': action_names.get(str(row['update_action'] or ''), str(row['update_action'] or '')),
                    'delete_action': action_names.get(str(row['delete_action'] or ''), str(row['delete_action'] or '')),
                    'valid': bool(row['is_valid']),
                }
                for row in rows
            }

        output = {}
        for row in self.execute(f'PRAGMA foreign_key_list({table})').fetchall():
            name = f'fk_{row["id"]}'
            current = output.setdefault(name, {
                'columns': [],
                'referenced_schema': 'main',
                'referenced_table': str(row['table'] or ''),
                'referenced_columns': [],
                'update_action': str(row['on_update'] or '').upper(),
                'delete_action': str(row['on_delete'] or '').upper(),
                'valid': True,
            })
            current['columns'].append(str(row['from'] or ''))
            current['referenced_columns'].append(str(row['to'] or ''))
        return output

    def table_check_constraints(self, table):
        if self.is_postgres:
            rows = self.execute(
                '''SELECT con.conname AS name,
                          pg_catalog.pg_get_expr(con.conbin, con.conrelid, true) AS expression,
                          pg_catalog.pg_get_constraintdef(con.oid, true) AS definition,
                          con.convalidated AS is_valid
                   FROM pg_catalog.pg_constraint con
                   JOIN pg_catalog.pg_class tbl ON tbl.oid = con.conrelid
                   JOIN pg_catalog.pg_namespace n ON n.oid = tbl.relnamespace
                  WHERE con.contype = 'c' AND n.nspname = ? AND tbl.relname = ?''',
                (self.application_schema, table),
            ).fetchall()
            return {
                str(row['name']): {
                    'expression': str(row['expression'] or ''),
                    'definition': str(row['definition'] or ''),
                    'valid': bool(row['is_valid']),
                }
                for row in rows
            }
        row = self.execute(
            "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?",
            (table,),
        ).fetchone()
        return _sqlite_check_constraints(row['sql'] if row else '')

    def table_triggers(self, table):
        if self.is_postgres:
            rows = self.execute(
                '''SELECT trg.tgname AS name,
                          trg.tgenabled <> 'D' AS enabled,
                          pg_catalog.pg_get_triggerdef(trg.oid, true) AS sql,
                          pg_catalog.pg_get_functiondef(trg.tgfoid) AS function_sql
                   FROM pg_catalog.pg_trigger trg
                   JOIN pg_catalog.pg_class tbl ON tbl.oid = trg.tgrelid
                   JOIN pg_catalog.pg_namespace n ON n.oid = tbl.relnamespace
                  WHERE n.nspname = ? AND tbl.relname = ?
                    AND NOT trg.tgisinternal''',
                (self.application_schema, table),
            ).fetchall()
            return {
                str(row['name']): {
                    'enabled': bool(row['enabled']),
                    'sql': str(row['sql'] or ''),
                    'function_sql': str(row['function_sql'] or ''),
                }
                for row in rows
            }
        rows = self.execute(
            "SELECT name, sql FROM sqlite_master WHERE type = 'trigger' AND tbl_name = ?",
            (table,),
        ).fetchall()
        return {
            str(row['name']): {
                'enabled': True,
                'sql': str(row['sql'] or ''),
                'function_sql': '',
            }
            for row in rows
        }

    def table_exists(self, table):
        if self.is_postgres:
            row = self.execute(
                '''SELECT c.oid AS name FROM pg_catalog.pg_class c
                   JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
                   WHERE n.nspname = ? AND c.relname = ? AND c.relkind IN ('r', 'p')''',
                (self.application_schema, table),
            ).fetchone()
            return bool(row and row['name'])
        row = self.execute("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?", (table,)).fetchone()
        return bool(row)

    def insert_returning_id(self, sql, params=None):
        if self.is_postgres:
            cur = self.execute(f'{sql.rstrip()} RETURNING id', params)
            row = cur.fetchone()
            return row['id'] if row else None
        cur = self.execute(sql, params)
        return cur.lastrowid if int(getattr(cur, 'rowcount', 0) or 0) != 0 else None

    def json_extract(self, column, path):
        if self.is_sqlite:
            return f"json_extract({column}, '{path}')"
        parts = str(path or '').lstrip('$.').split('.')
        pg_path = ','.join(part for part in parts if part)
        return f"(NULLIF({column}, '')::jsonb #>> '{{{pg_path}}}')"


def connect_sqlite(path, timeout_sec=30, read_only=False, immutable=False, check_same_thread=True):
    if read_only:
        params = 'mode=ro&immutable=1' if immutable else 'mode=ro'
        uri = 'file:' + str(path).replace('\\', '/') + '?' + params
        conn = sqlite3.connect(uri, uri=True, timeout=max(1, int(timeout_sec or 30)), check_same_thread=check_same_thread)
    else:
        conn = sqlite3.connect(path, timeout=max(1, int(timeout_sec or 30)), check_same_thread=check_same_thread)
    conn.row_factory = sqlite3.Row
    return DatabaseConnection('sqlite', conn)


def _bounded_int(value, default, minimum=1):
    try:
        return max(minimum, int(value))
    except (TypeError, ValueError):
        return max(minimum, int(default))


def connect_postgres(
    url,
    connect_timeout_sec=None,
    statement_timeout_ms=None,
    lock_timeout_ms=None,
    idle_in_transaction_timeout_ms=None,
    tcp_user_timeout_ms=None,
):
    parse_postgres_url(url)
    try:
        import psycopg
        from psycopg.rows import dict_row
    except ImportError as exc:
        raise RuntimeError('PostgreSQL backend requires psycopg[binary]. Install app requirements first.') from exc
    connect_timeout_sec = _bounded_int(
        connect_timeout_sec if connect_timeout_sec is not None else os.getenv('TRUF_DB_CONNECT_TIMEOUT_SEC'),
        DEFAULT_POSTGRES_CONNECT_TIMEOUT_SEC,
    )
    statement_timeout_ms = _bounded_int(
        statement_timeout_ms if statement_timeout_ms is not None else os.getenv('TRUF_DB_STATEMENT_TIMEOUT_MS'),
        DEFAULT_POSTGRES_STATEMENT_TIMEOUT_MS,
    )
    lock_timeout_ms = _bounded_int(
        lock_timeout_ms if lock_timeout_ms is not None else os.getenv('TRUF_DB_LOCK_TIMEOUT_MS'),
        DEFAULT_POSTGRES_LOCK_TIMEOUT_MS,
    )
    idle_in_transaction_timeout_ms = _bounded_int(
        idle_in_transaction_timeout_ms if idle_in_transaction_timeout_ms is not None else os.getenv('TRUF_DB_IDLE_TRANSACTION_TIMEOUT_MS'),
        DEFAULT_POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS,
    )
    tcp_user_timeout_ms = _bounded_int(
        tcp_user_timeout_ms if tcp_user_timeout_ms is not None else os.getenv('TRUF_DB_TCP_USER_TIMEOUT_MS'),
        DEFAULT_POSTGRES_TCP_USER_TIMEOUT_MS,
        minimum=1000,
    )
    options = ' '.join((
        f'-c search_path={POSTGRES_APPLICATION_SCHEMA}',
        f'-c statement_timeout={statement_timeout_ms}',
        f'-c lock_timeout={lock_timeout_ms}',
        f'-c idle_in_transaction_session_timeout={idle_in_transaction_timeout_ms}',
    ))
    for attempt in range(POSTGRES_CHILD_START_RETRY_ATTEMPTS):
        try:
            conn = psycopg.connect(
                url,
                row_factory=dict_row,
                connect_timeout=connect_timeout_sec,
                options=options,
                tcp_user_timeout=tcp_user_timeout_ms,
                keepalives=1,
                keepalives_idle=5,
                keepalives_interval=5,
                keepalives_count=2,
            )
            break
        except Exception as exc:
            transient_child_start = any(
                marker in str(exc).lower() for marker in POSTGRES_CHILD_START_RETRY_MARKERS
            )
            if not transient_child_start or attempt + 1 >= POSTGRES_CHILD_START_RETRY_ATTEMPTS:
                raise
            time.sleep(0.05 * (attempt + 1))
    try:
        cursor = conn.cursor()
        cursor.execute(
            """SELECT pg_catalog.current_schema() AS schema_name,
                      pg_catalog.current_setting('search_path') AS search_path,
                      EXISTS (
                          SELECT 1
                          FROM pg_catalog.pg_namespace n
                          CROSS JOIN LATERAL pg_catalog.aclexplode(
                              COALESCE(n.nspacl, pg_catalog.acldefault('n', n.nspowner))
                          ) acl
                          WHERE n.nspname = 'public'
                            AND acl.grantee = 0
                            AND acl.privilege_type = 'CREATE'
                      ) AS public_create"""
        )
        row = cursor.fetchone()
        cursor.close()
        schema_name = row.get('schema_name') if isinstance(row, dict) else row[0] if row else None
        search_path = row.get('search_path') if isinstance(row, dict) else row[1] if row else None
        public_create = row.get('public_create', False) if isinstance(row, dict) else row[2] if row and len(row) > 2 else False
        normalized_path = re.sub(r'[\s\"]', '', str(search_path or '').lower())
        if schema_name != POSTGRES_APPLICATION_SCHEMA or normalized_path != 'public' or bool(public_create):
            raise RuntimeError('PostgreSQL application schema/search_path validation failed')
        conn.rollback()
    except Exception:
        try:
            conn.close()
        except Exception:
            pass
        raise
    return DatabaseConnection('postgres', conn, application_schema=POSTGRES_APPLICATION_SCHEMA)


def connect_host_agent_postgres():
    """Open the one fixed peer-authenticated host-agent authority."""
    if os.name != 'posix' or not hasattr(os, 'geteuid') or os.geteuid() != 0:
        raise RuntimeError('PostgreSQL host-agent authority requires root on POSIX')
    try:
        import psycopg
        from psycopg.rows import dict_row
    except ImportError as exc:
        raise RuntimeError(
            'PostgreSQL backend requires psycopg[binary]. Install app requirements first.'
        ) from exc
    options = ' '.join((
        f'-c search_path={POSTGRES_APPLICATION_SCHEMA}',
        f'-c statement_timeout={DEFAULT_POSTGRES_STATEMENT_TIMEOUT_MS}',
        f'-c lock_timeout={DEFAULT_POSTGRES_LOCK_TIMEOUT_MS}',
        f'-c idle_in_transaction_session_timeout={DEFAULT_POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS}',
    ))
    conn = psycopg.connect(
        dbname=HOST_AGENT_POSTGRES_DATABASE,
        user=HOST_AGENT_POSTGRES_USER,
        host=HOST_AGENT_POSTGRES_SOCKET_DIRECTORY,
        port=HOST_AGENT_POSTGRES_PORT,
        row_factory=dict_row,
        connect_timeout=DEFAULT_POSTGRES_CONNECT_TIMEOUT_SEC,
        options=options,
        sslmode='disable',
    )
    try:
        cursor = conn.cursor()
        cursor.execute(
            """SELECT pg_catalog.current_schema() AS schema_name,
                      pg_catalog.current_setting('search_path') AS search_path,
                      CURRENT_USER AS current_user,
                      current_database() AS database_name,
                      pg_catalog.inet_server_addr() IS NULL AS unix_socket,
                      pg_catalog.current_setting('port')::integer AS port,
                      EXISTS (
                          SELECT 1
                          FROM pg_catalog.pg_namespace n
                          CROSS JOIN LATERAL pg_catalog.aclexplode(
                              COALESCE(n.nspacl, pg_catalog.acldefault('n', n.nspowner))
                          ) acl
                          WHERE n.nspname = 'public'
                            AND acl.grantee = 0
                            AND acl.privilege_type = 'CREATE'
                      ) AS public_create"""
        )
        row = cursor.fetchone()
        cursor.close()
        normalized_path = re.sub(
            r'[\s\"]', '', str((row or {}).get('search_path') or '').lower()
        )
        if (
            not isinstance(row, dict)
            or row.get('schema_name') != POSTGRES_APPLICATION_SCHEMA
            or normalized_path != POSTGRES_APPLICATION_SCHEMA
            or row.get('current_user') != HOST_AGENT_POSTGRES_USER
            or row.get('database_name') != HOST_AGENT_POSTGRES_DATABASE
            or row.get('unix_socket') is not True
            or row.get('port') != HOST_AGENT_POSTGRES_PORT
            or bool(row.get('public_create'))
        ):
            raise RuntimeError('PostgreSQL host-agent authority validation failed')
        conn.rollback()
    except Exception:
        try:
            conn.close()
        except Exception:
            pass
        raise
    return DatabaseConnection(
        'postgres', conn, application_schema=POSTGRES_APPLICATION_SCHEMA,
    )