;;; clutch-db-jdbc.el --- JDBC backend over the Java sidecar -*- lexical-binding: t; -*-

;; Copyright (C) 2025-2026 Lucius Chen
;; SPDX-License-Identifier: GPL-3.0-or-later


;; This file is part of clutch.

;; clutch is free software: you can redistribute it and/or modify
;; it under the terms of the GNU General Public License as published by
;; the Free Software Foundation, either version 3 of the License, or
;; (at your option) any later version.

;; clutch is distributed in the hope that it will be useful,
;; but WITHOUT ANY WARRANTY; without even the implied warranty of
;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
;; GNU General Public License for more details.

;; You should have received a copy of the GNU General Public License
;; along with clutch.  If not, see <https://www.gnu.org/licenses/>.

;;; Commentary:

;; JDBC backend for the clutch generic database interface.
;; Delegates to clutch-jdbc-agent (a JVM sidecar process) via a
;; single-line JSON protocol on stdin/stdout.
;;
;; Usage:
;;   (require 'clutch-db-jdbc)
;;   (clutch-connect '(:driver oracle :host db.corp.com :port 1521
;;                     :database ORCL :user scott :pass-entry my-entry))
;;
;; The agent jar is downloaded on first use.  JDBC drivers must be placed
;; manually in `clutch-jdbc-drivers-dir' (Oracle, DB2) or installed via
;; `clutch-jdbc-install-driver' (drivers available on Maven Central).

;;; Code:

(require 'cl-lib)
(require 'clutch-backend)
(require 'json)
(require 'seq)
(require 'sql)

;;;; Configuration

(defgroup clutch-jdbc nil
  "JDBC backend for clutch."
  :group 'clutch)

(defcustom clutch-jdbc-agent-dir
  (expand-file-name "clutch-jdbc" user-emacs-directory)
  "Directory containing clutch-jdbc-agent.jar and drivers/ subdirectory."
  :type 'directory
  :group 'clutch-jdbc)

(defcustom clutch-jdbc-agent-version "0.2.20"
  "Version of clutch-jdbc-agent to use."
  :type 'string
  :group 'clutch-jdbc)

(defcustom clutch-jdbc-agent-sha256
  "6f9f2cd76f9524e6bf3e9c9f22988a83d3ae8e223aeb2b6f97407af492540b28"
  "Expected SHA-256 for the configured clutch-jdbc-agent jar.
Set this to nil to disable checksum verification for a locally built jar."
  :type '(choice (const :tag "Disable verification" nil) string)
  :group 'clutch-jdbc)

(defcustom clutch-jdbc-agent-java-executable "java"
  "Java executable used to launch clutch-jdbc-agent."
  :type 'string
  :group 'clutch-jdbc)

(defcustom clutch-jdbc-agent-jvm-args '("-Xss512k"
                                        "-Dpolyglot.engine.WarnInterpreterOnly=false")
  "Extra JVM arguments passed when starting clutch-jdbc-agent.
Examples:
  (\"-Xss512k\")          — smaller thread stack, faster startup (default)
  (\"-Xss512k\" \"-Xmx256m\") — also cap heap at 256 MB"
  :type '(repeat string)
  :group 'clutch-jdbc)

(defcustom clutch-jdbc-fetch-size 500
  "Number of rows fetched per batch from the agent, from 1 through 10000."
  :type 'integer
  :group 'clutch-jdbc)

(defcustom clutch-jdbc-validate-after-idle-seconds 300
  "Validate a JDBC primary session before executing after this idle period.
The agent validates immediately before statement creation.  When that check
proves the session dead, Clutch may reconnect and execute once on a clean
transaction.  Set this to nil or 0 to disable idle validation."
  :type '(choice (const :tag "Disable" nil) natnum)
  :group 'clutch-jdbc)

(defconst clutch-jdbc--max-fetch-size 10000
  "Maximum batch size accepted by the JDBC agent protocol.")

(defcustom clutch-jdbc-oracle-manual-commit t
  "When non-nil, Oracle JDBC connections default to manual-commit mode.
Set this to nil to keep Oracle in auto-commit by default.  Per-connection
`:manual-commit' still overrides this default when explicitly present."
  :type 'boolean
  :group 'clutch-jdbc)

(defvar clutch-jdbc-cancel-timeout-seconds 5
  "Seconds to wait for a cancel acknowledgement from the JDBC agent.
Shorter than `clutch-jdbc-rpc-timeout-seconds' because a slow cancel
should degrade to disconnect, not block the user.")

(defvar clutch-jdbc-disconnect-timeout-seconds 5
  "Seconds to wait for a disconnect acknowledgement from the JDBC agent.
A stuck disconnect should not block the user or kill the agent.")

(defvar clutch-debug-mode nil
  "Forward declaration; defined as a global minor mode in clutch-diagnostics.el.")

(defvar clutch-debug-buffer-name "*clutch-debug*"
  "Forward declaration; defined in clutch-diagnostics.el.")

;;;; Driver sources (for automatic installation from Maven Central)

(defconst clutch-jdbc--driver-sources
  '((sqlserver . (:maven "com.microsoft.sqlserver:mssql-jdbc:13.4.0.jre11"
                  :filename "mssql-jdbc.jar"
                  :class "com.microsoft.sqlserver.jdbc.SQLServerDriver"))
    (snowflake . (:maven "net.snowflake:snowflake-jdbc:3.14.4"
                  :filename "snowflake-jdbc.jar"
                  :class "net.snowflake.client.jdbc.SnowflakeDriver"))
    ;; ojdbc8 (19c driver) is the safest default across Oracle 11g/12c/19c.
    (oracle    . (:maven "com.oracle.database.jdbc:ojdbc8:19.21.0.0"
                  :filename "ojdbc8.jar"
                  :class "oracle.jdbc.OracleDriver"
                  :companions (oracle-i18n)))
    (oracle-8  . (:maven "com.oracle.database.jdbc:ojdbc8:19.21.0.0"
                  :filename "ojdbc8.jar"
                  :class "oracle.jdbc.OracleDriver"
                  :companions (oracle-i18n)))
    ;; ojdbc11 remains available for users who explicitly want the newer line.
    (oracle-11 . (:maven "com.oracle.database.jdbc:ojdbc11:21.13.0.0"
                  :filename "ojdbc11.jar"
                  :class "oracle.jdbc.OracleDriver"
                  :companions (oracle-i18n)))
    (oracle-i18n . (:maven "com.oracle.database.nls:orai18n:21.13.0.0"
                    :filename "orai18n.jar"))
    (db2       . (:manual "https://www.ibm.com/support/pages/db2-jdbc-driver-versions-and-downloads"
                  :filename "db2jcc4.jar"
                  :class "com.ibm.db2.jcc.DB2Driver"))
    (redshift  . (:maven "com.amazon.redshift:redshift-jdbc42:2.1.0.30"
                  :filename "redshift-jdbc42.jar"
                  :class "com.amazon.redshift.jdbc.Driver"))
    (clickhouse . (:maven "com.clickhouse:clickhouse-jdbc:0.9.8:all"
                   :filename "clickhouse-jdbc.jar"
                   :class "com.clickhouse.jdbc.ClickHouseDriver"
                   :companions (slf4j-api slf4j-nop)))
    (duckdb     . (:maven "org.duckdb:duckdb_jdbc:1.5.3.0"
                  :filename "duckdb_jdbc.jar"
                  :class "org.duckdb.DuckDBDriver"))
    (mongodb    . (:maven "org.mongodb:mongodb-jdbc:3.0.6:all"
                  :filename "mongodb-jdbc.jar"
                  :class "com.mongodb.jdbc.MongoDriver"))
    (slf4j-api  . (:maven "org.slf4j:slf4j-api:2.0.16"
                   :filename "slf4j-api.jar"))
    (slf4j-nop  . (:maven "org.slf4j:slf4j-nop:2.0.16"
                   :filename "slf4j-nop.jar")))
  "Known JDBC driver sources.
All entries support auto-download via `clutch-jdbc-install-driver'.")

(defconst clutch-jdbc--driver-metadata
  '((jdbc       . (:display-name "JDBC"
                  :support-level basic
                  :data-model relational
                  :manual-choice nil))
    (oracle     . (:display-name "Oracle"
                  :default-port 1521
                  :support-level core
                  :data-model relational
                  :update-default t
                  :sql-product oracle))
    (sqlserver  . (:display-name "SQL Server"
                  :default-port 1433
                  :support-level core
                  :data-model relational
                  :update-default t
                  :sql-product ms))
    (db2        . (:display-name "DB2"
                  :default-port 50000
                  :support-level basic
                  :data-model relational
                  :update-default t
                  :sql-product db2))
    ;; Snowflake and ClickHouse have no `sql-mode' product, and Redshift's
    ;; PostgreSQL 8.0 lineage predates standard_conforming_strings, so all
    ;; three read a backslash as an escape and support dollar-quoted bodies.
    ;; Their `:sql-dialect' entries state that directly.
    (snowflake  . (:display-name "Snowflake"
                  :default-port 443
                  :support-level basic
                  :data-model relational
                  :update-default t
                  :sql-dialect (:backslash-escapes t :dollar-quotes t)))
    (redshift   . (:display-name "Redshift"
                  :default-port 5439
                  :support-level basic
                  :data-model relational
                  :update-default t
                  :sql-product postgres
                  :sql-dialect (:backslash-escapes t :dollar-quotes t)))
    (clickhouse . (:display-name "ClickHouse"
                  :default-port 8123
                  :support-level basic
                  :data-model relational
                  :sql-dialect (:backslash-escapes t :dollar-quotes t))))
  "Metadata for JDBC-backed concrete drivers.")

(defconst clutch-jdbc--oracle-driver-filenames
  '("ojdbc8.jar" "ojdbc11.jar")
  "Oracle JDBC driver jar filenames that conflict with each other.")

;;;; Connection struct

(cl-defstruct clutch-jdbc-conn
  "A JDBC connection managed by clutch-jdbc-agent."
  process   ; the shared agent process
  conn-id   ; integer handle in the agent's ConnectionManager
  driver    ; internal JDBC driver implementation symbol
  params    ; original connection plist (for metadata)
  busy)     ; non-nil while a query is running

(defun clutch-jdbc--conn-driver (conn)
  "Return the internal JDBC driver implementation for CONN."
  (or (clutch-jdbc-conn-driver conn)
      (plist-get (clutch-jdbc-conn-params conn) :driver)
      'jdbc))

;;;; Agent process (one shared process for all JDBC connections)

(defvar clutch-jdbc--agent-process nil
  "The running clutch-jdbc-agent process, or nil if not started.")

(defvar clutch-jdbc--response-queue nil
  "List of parsed synchronous JSON responses, oldest first.")

(defvar-local clutch-jdbc--agent-scan-position nil
  "Buffer position where the JDBC agent filter should resume line scanning.")

(defvar clutch-jdbc--async-callbacks (make-hash-table :test 'eql)
  "Map of JDBC request ids to asynchronous callbacks.")

(defvar clutch-jdbc--busy-request-ids (make-hash-table :test 'eq)
  "Map of JDBC connection objects to their current in-flight request id.")

(defvar clutch-jdbc--ignored-response-ids (make-hash-table :test 'eql)
  "Set of JDBC response ids to drop because the request was interrupted.")

(defvar clutch-jdbc--connections-by-id (make-hash-table :test 'eql)
  "Map of JDBC connection ids to their live connection structs.")

(defvar clutch-jdbc--error-details-by-conn (make-hash-table :test 'eq)
  "Map of JDBC connection objects to their latest structured error details.")

(defconst clutch-jdbc--json-false (make-symbol "clutch-jdbc-json-false")
  "Sentinel used to represent JSON false distinctly from nil.")

(defun clutch-jdbc--protocol-error-response (message)
  "Return a synthetic response for JDBC protocol error MESSAGE."
  (list :protocol-error t
        :ok clutch-jdbc--json-false
        :error message))

(defconst clutch-jdbc--object-category-specs
  '((indexes . (:op "get-indexes" :key :indexes))
    (sequences . (:op "get-sequences" :key :sequences))
    (procedures . (:op "get-procedures" :key :procedures))
    (functions . (:op "get-functions" :key :functions))
    (triggers . (:op "get-triggers" :key :triggers)))
  "JDBC object category to agent operation/result-key mapping.")

(defun clutch-jdbc--agent-jar ()
  "Return the path to the clutch-jdbc-agent jar."
  (expand-file-name
   (format "clutch-jdbc-agent-%s.jar" clutch-jdbc-agent-version)
   clutch-jdbc-agent-dir))

(defun clutch-jdbc--drivers-dir ()
  "Return the drivers/ directory path."
  (expand-file-name "drivers" clutch-jdbc-agent-dir))

(defun clutch-jdbc--agent-jar-sha256 (&optional jar)
  "Return the SHA-256 of JAR.
JAR defaults to `clutch-jdbc--agent-jar'."
  (with-temp-buffer
    (set-buffer-multibyte nil)
    (insert-file-contents-literally (or jar (clutch-jdbc--agent-jar)))
    (secure-hash 'sha256 (current-buffer))))

(defun clutch-jdbc--agent-jar-valid-p (&optional jar)
  "Return non-nil if JAR matches `clutch-jdbc-agent-sha256'.
If verification is disabled, return non-nil."
  (or (null clutch-jdbc-agent-sha256)
      (string-equal (clutch-jdbc--agent-jar-sha256 jar)
                    clutch-jdbc-agent-sha256)))

(defun clutch-jdbc--validate-agent-jar (&optional jar)
  "Signal `user-error' unless JAR exists and passes checksum verification.
JAR defaults to `clutch-jdbc--agent-jar'."
  (let ((jar (or jar (clutch-jdbc--agent-jar))))
    (unless (file-exists-p jar)
      (user-error "JDBC agent jar not found: %s\nRun M-x clutch-jdbc-ensure-agent" jar))
    (unless (clutch-jdbc--agent-jar-valid-p jar)
      (user-error (concat "JDBC agent checksum mismatch: %s\n"
                          "Run M-x clutch-jdbc-ensure-agent to refresh it,\n"
                          "or set `clutch-jdbc-agent-sha256' to nil for a custom jar")
                  jar))))

(defun clutch-jdbc--cleanup-stale-agent-jars ()
  "Delete stale versioned clutch-jdbc-agent jars from `clutch-jdbc-agent-dir'."
  (let ((current (file-name-nondirectory (clutch-jdbc--agent-jar)))
        (files (directory-files clutch-jdbc-agent-dir t
                                "\\`clutch-jdbc-agent-.*\\.jar\\'")))
    (dolist (file files)
      (unless (string-equal (file-name-nondirectory file) current)
        (delete-file file)))))

(defun clutch-jdbc--agent-live-p ()
  "Return non-nil if the agent process is running."
  (and clutch-jdbc--agent-process
       (process-live-p clutch-jdbc--agent-process)))

(defun clutch-jdbc--clear-async-callbacks (&optional conn)
  "Cancel pending asynchronous JDBC callbacks.
When CONN is non-nil, clear only callbacks for that connection and ignore their
late responses.  Otherwise clear every callback."
  (let (ids)
    (maphash (lambda (id entry)
               (when (or (null conn) (eq conn (plist-get entry :conn)))
                 (when-let* ((timer (plist-get entry :timer)))
                   (cancel-timer timer))
                 (when conn
                   (puthash id t clutch-jdbc--ignored-response-ids))
                 (push id ids)))
             clutch-jdbc--async-callbacks)
    (dolist (id ids)
      (remhash id clutch-jdbc--async-callbacks))))

(defun clutch-jdbc--clear-request-state ()
  "Clear in-flight and ignored JDBC request bookkeeping."
  (clrhash clutch-jdbc--busy-request-ids)
  (clrhash clutch-jdbc--ignored-response-ids))

(defun clutch-jdbc--dispatch-async-response (response)
  "Dispatch asynchronous RESPONSE when a callback is registered.
Return non-nil when RESPONSE was consumed asynchronously."
  (let* ((id (plist-get response :id))
         (entry (and id (gethash id clutch-jdbc--async-callbacks))))
    (when entry
      (remhash id clutch-jdbc--async-callbacks)
      (when-let* ((timer (plist-get entry :timer)))
        (cancel-timer timer))
      (let ((callback (plist-get entry :callback))
            (errback (plist-get entry :errback))
            (conn (plist-get entry :conn))
            (op (plist-get entry :op)))
        (run-at-time
         0 nil
         (lambda ()
           (when (or (not (clutch-jdbc-conn-p conn))
                     (eq conn
                         (gethash (clutch-jdbc-conn-conn-id conn)
                                  clutch-jdbc--connections-by-id)))
             (condition-case err
                 (if (eq t (plist-get response :ok))
                     (when callback
                       (funcall callback (plist-get response :result)))
                   (clutch-jdbc--remember-error-response conn op response)
                   (when (clutch-jdbc--connection-invalidated-p response)
                     (clutch-jdbc--retire-invalidated-connection conn))
                   (let ((message (clutch-jdbc--rpc-error-message op response)))
                     (if errback
                         (funcall errback message)
                       (message "clutch-jdbc async error: %s" message))))
               (error
                (message "clutch-jdbc async callback failed: %s"
                         (error-message-string err))))))))
      t)))

(defun clutch-jdbc--agent-filter (proc string)
  "Process filter: collect complete JSON lines from PROC output STRING."
  (let ((buf (process-buffer proc)))
    (when (buffer-live-p buf)
      (with-current-buffer buf
        (goto-char (point-max))
        (insert string)
        ;; Collect complete lines into the response queue.
        (goto-char (min (or clutch-jdbc--agent-scan-position (point-min))
                        (point-max)))
        (while (search-forward "\n" nil t)
          (let ((line (string-trim (buffer-substring (point-min) (point)))))
            (delete-region (point-min) (point))
            (goto-char (point-min))
            (unless (string-empty-p line)
              (let ((parsed (condition-case err
                                (json-parse-string line :object-type 'plist
                                                   :array-type 'list
                                                   :null-object nil
                                                   :false-object clutch-jdbc--json-false)
                              (error
                               (clutch-jdbc--protocol-error-response
                                (format "clutch-jdbc-agent emitted invalid JSON: %s"
                                        (error-message-string err)))))))
                (when parsed
                  (let ((id (plist-get parsed :id)))
                    (cond
                     ((clutch-jdbc--dispatch-async-response parsed)
                      nil)
                     ((and id (gethash id clutch-jdbc--ignored-response-ids))
                      (remhash id clutch-jdbc--ignored-response-ids))
                     (t
                      (setq clutch-jdbc--response-queue
                            (nconc clutch-jdbc--response-queue
                                   (list parsed)))))))))))
        (setq clutch-jdbc--agent-scan-position (point-max))))))

(defun clutch-jdbc--start-agent ()
  "Start the clutch-jdbc-agent process and wait for its ready signal."
  (let ((jar (clutch-jdbc--agent-jar)))
    (clutch-jdbc--validate-agent-jar jar)
    (unless (executable-find clutch-jdbc-agent-java-executable)
      (user-error "Java not found.  Set `clutch-jdbc-agent-java-executable'"))
    (let* ((buf (generate-new-buffer " *clutch-jdbc-agent*"))
           (stderr (get-buffer-create "*clutch-jdbc-agent-stderr*"))
           (proc (progn
                   ;; The stderr buffer is reused across agent restarts, and
                   ;; startup diagnostics read its tail.  Drop the previous
                   ;; agent's output so it cannot be reported as this one's.
                   (with-current-buffer stderr
                     (let ((inhibit-read-only t))
                       (erase-buffer)))
                   (make-process
                    :name "clutch-jdbc-agent"
                    :buffer buf
                    :command (append (list clutch-jdbc-agent-java-executable)
                                     clutch-jdbc-agent-jvm-args
                                     (list "-jar" jar (clutch-jdbc--drivers-dir)))
                    :connection-type 'pipe
                    ;; The protocol is line-delimited JSON, so both directions
                    ;; are UTF-8 regardless of the locale Emacs was started in.
                    :coding '(utf-8-unix . utf-8-unix)
                    :filter #'clutch-jdbc--agent-filter
                    :stderr stderr
                    :noquery t))))
      (setq clutch-jdbc--agent-process proc)
      (setq clutch-jdbc--response-queue nil)
      (clutch-jdbc--clear-async-callbacks)
      (clutch-jdbc--clear-request-state)
      ;; Wait for the ready message (id=0).
      (let ((ready (clutch-jdbc--recv-response 0)))
        (unless (plist-get ready :ok)
          (error "JDBC agent failed to start: %s" (plist-get ready :error))))
      proc)))

(defun clutch-jdbc--stop-agent ()
  "Stop the shared clutch-jdbc-agent process, if running."
  (when (clutch-jdbc--agent-live-p)
    (delete-process clutch-jdbc--agent-process))
  (setq clutch-jdbc--agent-process nil
        clutch-jdbc--response-queue nil)
  (clutch-jdbc--clear-async-callbacks)
  (clutch-jdbc--clear-request-state))

(defun clutch-jdbc--ensure-agent ()
  "Ensure the agent process is running, starting it if necessary."
  (unless (clutch-jdbc--agent-live-p)
    (setq clutch-jdbc--response-queue nil)
    (clutch-jdbc--start-agent)))

(defun clutch-jdbc--agent-stderr-buffer ()
  "Return the JDBC agent stderr buffer, or nil when absent."
  (get-buffer "*clutch-jdbc-agent-stderr*"))

(defun clutch-jdbc--agent-stderr-string ()
  "Return the full JDBC agent stderr buffer as a string, or nil when empty."
  (when-let* ((buf (clutch-jdbc--agent-stderr-buffer))
              ((buffer-live-p buf)))
    (with-current-buffer buf
      (let ((text (string-trim (buffer-string))))
        (unless (string-empty-p text)
          text)))))

(defun clutch-jdbc--agent-stderr-tail (&optional max-lines)
  "Return the last non-empty MAX-LINES of agent stderr as a string.
Defaults to 8 lines.  Return nil when stderr is empty."
  (when-let* ((text (clutch-jdbc--agent-stderr-string)))
    (let* ((lines (seq-filter
                   (lambda (line) (not (string-empty-p (string-trim line))))
                   (split-string text "\n" t)))
           (count (or max-lines 8)))
      (when lines
        (string-join
         (last lines (min count (length lines)))
         "\n")))))

(defun clutch-jdbc--agent-exit-error-message ()
  "Return a user-facing error string when the JDBC agent exited early."
  (let ((stderr (clutch-jdbc--agent-stderr-string))
        (stderr-tail (clutch-jdbc--agent-stderr-tail)))
    (cond
     ((and stderr
           (string-match-p "UnsupportedClassVersionError" stderr))
      (format "clutch-jdbc-agent requires a newer Java runtime than `%s'. Update `clutch-jdbc-agent-java-executable' or JAVA_HOME.\n%s"
              clutch-jdbc-agent-java-executable
              stderr-tail))
     (stderr-tail
      (format "clutch-jdbc-agent exited before replying:\n%s" stderr-tail))
     (t
      "clutch-jdbc-agent exited before replying"))))

;;;; Synchronous RPC

(defvar clutch-jdbc--next-request-id 1
  "Auto-incrementing request id counter.")

(defun clutch-jdbc--request-params (params)
  "Return PARAMS with opt-in debug capture flags when enabled."
  (if (or (not clutch-debug-mode)
          (assq 'debug params))
      params
    (append params '((debug . t)))))

(defun clutch-jdbc--send (op params)
  "Send OP with PARAMS to the agent and return the request id."
  (let* ((id (cl-incf clutch-jdbc--next-request-id))
         (request `((id . ,id)
                    (op . ,op)
                    (params . ,(clutch-jdbc--request-params params))))
         (msg (let ((json-false clutch-jdbc--json-false))
                (json-encode request))))
    (process-send-string clutch-jdbc--agent-process (concat msg "\n"))
    id))

(defun clutch-jdbc--take-queued-response (id)
  "Remove and return queued response matching ID, preserving other responses."
  (let (response remaining)
    (while (and (not response) clutch-jdbc--response-queue)
      (let ((parsed (pop clutch-jdbc--response-queue)))
        (cond
         ((and parsed
               (plist-get parsed :protocol-error))
          (setq response parsed))
         ((and parsed
               (gethash (plist-get parsed :id) clutch-jdbc--ignored-response-ids))
          (remhash (plist-get parsed :id) clutch-jdbc--ignored-response-ids))
         ((and parsed (eql (plist-get parsed :id) id))
          (setq response parsed))
         (t
          (push parsed remaining)))))
    (setq clutch-jdbc--response-queue
          (nconc (nreverse remaining) clutch-jdbc--response-queue))
    response))

(defun clutch-jdbc--recv-response (id &optional timeout-seconds op conn)
  "Wait for and return the response with matching ID as a plist.
TIMEOUT-SECONDS defaults to `clutch-jdbc-rpc-timeout-seconds'.
OP, when non-nil, names the RPC for context-sensitive timeout errors.
CONN, when non-nil, is the connection the request ran on; a timeout with
a live agent then condemns only that connection instead of the process."
  (let ((deadline (+ (float-time)
                     (or timeout-seconds clutch-jdbc-rpc-timeout-seconds)))
        response
        failure-message)
    (while (and (not response) (< (float-time) deadline))
      ;; Drain any queued responses while preserving unmatched entries.
      (when clutch-jdbc--response-queue
        (setq response (clutch-jdbc--take-queued-response id)))
      (unless response
        (if (and clutch-jdbc--agent-process
                 (not (process-live-p clutch-jdbc--agent-process)))
            (setq failure-message (clutch-jdbc--agent-exit-error-message)
                  response :agent-exited)
          (accept-process-output clutch-jdbc--agent-process 0.05)
          (sit-for 0 t))))
    (when (eq response :agent-exited)
      (setq response nil))
    (when (and (not response)
               (not failure-message)
               clutch-jdbc--agent-process
               (not (process-live-p clutch-jdbc--agent-process)))
      (setq failure-message (clutch-jdbc--agent-exit-error-message)))
    (when (and response (plist-get response :protocol-error))
      (when (and clutch-jdbc--agent-process
                 (process-live-p clutch-jdbc--agent-process))
        (delete-process clutch-jdbc--agent-process))
      (clutch-jdbc--clear-async-callbacks)
      (clutch-jdbc--clear-request-state)
      (setq clutch-jdbc--agent-process nil
            clutch-jdbc--response-queue nil)
      (signal 'clutch-db-error
              (list (plist-get response :error))))
    (unless response
      (if (and (clutch-jdbc-conn-p conn)
               (not failure-message)
               (process-live-p clutch-jdbc--agent-process))
          ;; One request went silent but the agent process is alive.  The
          ;; agent serves requests on a thread pool, so a stuck JDBC call
          ;; wedges only the session it ran on; killing the process would
          ;; also destroy every other connection's sessions and any open
          ;; transactions.  Condemn the owning connection alone.
          (progn
            (puthash id t clutch-jdbc--ignored-response-ids)
            (clutch-jdbc--release-stuck-connection conn)
            (signal 'clutch-db-error
                    (list "Connection lost — reconnect with C-c C-e")))
        ;; The agent process died, or the silent request has no owning
        ;; connection (startup handshake, connect), so there is nothing
        ;; narrower to reset than the process and every registration
        ;; hanging off it.
        (when (process-live-p clutch-jdbc--agent-process)
          (delete-process clutch-jdbc--agent-process))
        (clutch-jdbc--clear-async-callbacks)
        (clutch-jdbc--clear-request-state)
        (setq clutch-jdbc--agent-process nil
              clutch-jdbc--response-queue nil)
        (signal 'clutch-db-error
                (list (or failure-message
                          (if (equal op "connect")
                              "Connection attempt timed out or JDBC agent became unresponsive"
                            "Connection lost — reconnect with C-c C-e"))))))
    response))

(defun clutch-jdbc--recv-response-nonfatal (id timeout-seconds)
  "Wait for response ID up to TIMEOUT-SECONDS.
Return the response plist, or nil on timeout/quit.
Unlike `clutch-jdbc--recv-response', this never kills the agent process."
  (let ((inhibit-quit t))
    (let ((deadline (+ (float-time) timeout-seconds))
          response
          agent-exited
          gave-up)
      (while (and (not response) (not agent-exited) (not gave-up)
                  (< (float-time) deadline))
        ;; Drain any queued responses while preserving unmatched entries.
        (when clutch-jdbc--response-queue
          (setq response (clutch-jdbc--take-queued-response id)))
        (unless response
          (if (or (null clutch-jdbc--agent-process)
                  (not (process-live-p clutch-jdbc--agent-process)))
              (setq agent-exited t)
            (let ((output
                   (with-local-quit
                     (accept-process-output clutch-jdbc--agent-process 0.05))))
              (sit-for 0 t)
              (when (and (not output) quit-flag)
                (setq gave-up t
                      quit-flag nil))))))
      (when (and (not response)
                 (not agent-exited)
                 clutch-jdbc--agent-process
                 (not (process-live-p clutch-jdbc--agent-process)))
        (setq agent-exited t))
      (when (and (not response) (not agent-exited))
        (puthash id t clutch-jdbc--ignored-response-ids))
      (unless agent-exited
        response))))

(defun clutch-jdbc--rpc (op params &optional timeout-seconds)
  "Send OP with PARAMS to the agent and return the result plist.
TIMEOUT-SECONDS overrides the default wait time.  Signals
`clutch-db-error' on agent-reported errors."
  (clutch-jdbc--ensure-agent)
  (let* ((conn (clutch-jdbc--conn-from-params params))
         (id (clutch-jdbc--send op params))
         (timeout (or timeout-seconds
                      (and conn (clutch-jdbc--conn-rpc-timeout conn))))
         (response (clutch-jdbc--recv-response id timeout op conn)))
    (clutch-jdbc--response-result-or-signal conn op response)))

(defun clutch-jdbc--rpc-error-message (op response)
  "Return a user-facing error string for OP from RESPONSE."
  (let ((message (or (plist-get response :error)
                     (format "agent error on op %s" op))))
    (if (and (plist-get response :diag)
             (not (string-match-p (regexp-quote clutch-debug-buffer-name) message)))
        (if clutch-debug-mode
            (format "%s See %s for details." message clutch-debug-buffer-name)
          (format
           "%s Enable clutch-debug-mode, reproduce the failure, then inspect %s."
           message clutch-debug-buffer-name))
      message)))

(defun clutch-jdbc--error-details-from-response (op response)
  "Build a structured error-details plist for OP from RESPONSE."
  (when-let* ((diag (plist-get response :diag)))
    (list :backend 'jdbc
          :summary (or (plist-get response :error)
                       (and op (format "agent error on op %s" op)))
          :diag (copy-tree diag)
          :debug (copy-tree (plist-get response :debug))
          :stderr-tail (clutch-jdbc--agent-stderr-tail))))

(defun clutch-jdbc--remember-error-response (conn op response)
  "Remember JDBC error RESPONSE for CONN and OP, and return its details plist.
When CONN is nil, return the details snapshot for the current failure."
  (when-let* ((details (clutch-jdbc--error-details-from-response op response)))
    (when conn
      (puthash conn details clutch-jdbc--error-details-by-conn))
    details))

(defun clutch-jdbc--connection-invalidated-p (response)
  "Return non-nil if RESPONSE has an invalidated logical connection."
  (eq t (plist-get (plist-get response :diag) :connection-invalidated)))

(defun clutch-jdbc--retire-invalidated-connection (conn)
  "Retire invalidated JDBC CONN from local live-request state.
The agent has already removed the logical connection, so do not send a
disconnect request.  Preserve connection-scoped diagnostics for the caller."
  (when (and (clutch-jdbc-conn-p conn)
             (eq conn
                 (gethash (clutch-jdbc-conn-conn-id conn)
                          clutch-jdbc--connections-by-id)))
    (remhash conn clutch-jdbc--busy-request-ids)
    (clutch-jdbc--clear-async-callbacks conn)
    (remhash (clutch-jdbc-conn-conn-id conn)
             clutch-jdbc--connections-by-id)))

(defun clutch-jdbc--release-stuck-connection (conn)
  "Retire CONN after a request on it went silent, asking the agent to drop it.
The release uses force-disconnect, which bypasses the connection's locks
in the agent: the ordinary disconnect queues behind them, so the very
call that went silent would block the release forever and pin an agent
thread.  The request is sent without waiting and its reply, if any, is
ignored.  An older agent answers with an unknown-op error, which lands
in the ignored table the same way."
  (clutch-jdbc--retire-invalidated-connection conn)
  (when (clutch-jdbc--agent-live-p)
    (puthash (clutch-jdbc--send
              "force-disconnect"
              `((conn-id . ,(clutch-jdbc-conn-conn-id conn))))
             t clutch-jdbc--ignored-response-ids)))

(defun clutch-jdbc--response-result-or-signal (conn op response)
  "Return RESPONSE's result or signal `clutch-db-error' for OP.
When CONN is non-nil, remember structured diagnostics on the connection."
  (if (eq t (plist-get response :ok))
      (plist-get response :result)
    (let* ((details (clutch-jdbc--remember-error-response conn op response))
           (diag (plist-get response :diag))
           (execution-not-started
            (and (clutch-jdbc-conn-p conn)
                 (eql (clutch-jdbc-conn-conn-id conn)
                      (plist-get diag :conn-id))
                 (member op '("execute" "execute-params"))
                 (equal (plist-get diag :category) "query")
                 (equal (plist-get diag :op) op)
                 (eq t (plist-get diag :connection-invalidated))
                 (eq t (plist-get diag :execution-not-started)))))
      (when (clutch-jdbc--connection-invalidated-p response)
        (clutch-jdbc--retire-invalidated-connection conn))
      (signal (if execution-not-started
                  'clutch-db-execution-not-started
                'clutch-db-error)
              (if details
                  (list (clutch-jdbc--rpc-error-message op response) details)
                (list (clutch-jdbc--rpc-error-message op response)))))))

(defun clutch-jdbc--conn-from-params (params)
  "Return the JDBC connection object referenced by PARAMS, or nil."
  (when-let* ((conn-id (alist-get 'conn-id params)))
    (gethash conn-id clutch-jdbc--connections-by-id)))

(defun clutch-jdbc--rpc-on-conn (conn op params &optional timeout-seconds)
  "Send OP with PARAMS while tracking the in-flight request for CONN."
  (clutch-jdbc--ensure-agent)
  (let* ((id (clutch-jdbc--send op params))
         (clear-request-id t)
         response)
    (puthash conn id clutch-jdbc--busy-request-ids)
    (unwind-protect
        (condition-case nil
            (progn
              (setq response (clutch-jdbc--recv-response id timeout-seconds op conn))
              (clutch-jdbc--response-result-or-signal conn op response))
          (quit
           (setq clear-request-id nil)
           (signal 'quit nil)))
      (when (and clear-request-id
                 (eql (gethash conn clutch-jdbc--busy-request-ids) id))
        (remhash conn clutch-jdbc--busy-request-ids)))))

(defun clutch-jdbc--rpc-async (op params callback &optional errback timeout-seconds conn)
  "Send OP with PARAMS to the agent asynchronously.
CALLBACK receives the result plist on success.  ERRBACK receives a
string error message on failure.  TIMEOUT-SECONDS defaults to
`clutch-jdbc-rpc-timeout-seconds'.  CONN tracks connection-scoped
diagnostics when non-nil.  Return the request id."
  (clutch-jdbc--ensure-agent)
  (let* ((id (clutch-jdbc--send op params))
         (timeout (or timeout-seconds clutch-jdbc-rpc-timeout-seconds))
         (timer (run-at-time
                 timeout nil
                 (lambda ()
                   (let ((entry (gethash id clutch-jdbc--async-callbacks)))
                     (when entry
                       (remhash id clutch-jdbc--async-callbacks)
                       (puthash id t clutch-jdbc--ignored-response-ids)
                       (when-let* ((timeout-errback (plist-get entry :errback)))
                         (funcall timeout-errback
                                  (format "clutch-jdbc-agent: timeout waiting for async response to request %d"
                                          id)))))))))
    (puthash id (list :callback callback :errback errback :timer timer
                      :conn conn :op op)
             clutch-jdbc--async-callbacks)
    id))

;;;; JDBC URL builder

(defun clutch-jdbc--sql-interface-url (params)
  "Build a MongoDB SQL Interface JDBC URL from PARAMS.
The MongoDB JDBC driver's query database is passed separately as the
driver-specific `database' property; the optional URL path remains available
for the MongoDB default auth database."
  (let* ((host (or (plist-get params :host) "localhost"))
         (port (plist-get params :port))
         (auth-db (or (plist-get params :auth-database)
                      (plist-get params :auth-source))))
    (format "jdbc:mongodb://%s%s%s"
            host
            (if port (format ":%d" port) "")
            (if (and auth-db (not (string-empty-p auth-db)))
                (format "/%s" auth-db)
              ""))))

(defun clutch-jdbc--sql-interface-jdbc-url-p (url)
  "Return non-nil when URL is a MongoDB SQL Interface JDBC URL."
  (and (stringp url)
       (or (string-prefix-p "jdbc:mongodb:" url)
           (string-prefix-p "jdbc:mongodb+srv:" url))))

(defun clutch-jdbc--build-url (driver params)
  "Build a JDBC URL for DRIVER using connection PARAMS plist.
If :url is present in PARAMS it is used as-is (allows full override).
Otherwise constructs a URL from :host, :port, and :database (service name)
or :sid (Oracle SID-style connection)."
  (or (and (eq driver 'mongodb)
           (when-let* ((url (plist-get params :url)))
             (unless (clutch-jdbc--sql-interface-jdbc-url-p url)
               (signal 'clutch-db-error
                       (list "MongoDB SQL Interface JDBC URLs must start with jdbc:mongodb:")))
             url))
      (and (not (eq driver 'mongodb))
           (plist-get params :url))
      (let ((host     (or (plist-get params :host) "localhost"))
            (port     (or (plist-get params :port)
                          (clutch-backend-default-port driver)))
            (database (plist-get params :database))
            (sid      (plist-get params :sid)))
        (pcase driver
          ('oracle
           (if sid
               (format "jdbc:oracle:thin:@%s:%d:%s"
                       host port sid)
             (format "jdbc:oracle:thin:@//%s:%d/%s"
                     host port database)))
          ('sqlserver
           (format "jdbc:sqlserver://%s:%d;databaseName=%s"
                   host port database))
          ('db2
           (format "jdbc:db2://%s:%d/%s"
                   host port database))
          ('snowflake
           (format "jdbc:snowflake://%s.snowflakecomputing.com/?db=%s"
                   host database))
          ('redshift
           (format "jdbc:redshift://%s:%d/%s"
                   host port database))
          ('clickhouse
           (format "jdbc:clickhouse://%s:%d/%s"
                   host port database))
          ('mongodb
           (clutch-jdbc--sql-interface-url params))
          (_
           (error "Unknown JDBC driver %s; provide :url directly" driver))))))

(defun clutch-jdbc--prop-value (props key)
  "Return string-keyed connection property KEY from PROPS, or nil."
  (cdr (assoc-string key props t)))

(defun clutch-jdbc--sql-interface-props (params)
  "Return JDBC properties for MongoDB SQL Interface PARAMS.
The official MongoDB JDBC driver requires the driver-specific `database'
property even when the URL includes a default auth database."
  (let* ((props (copy-sequence
                 (clutch-jdbc--normalize-props (plist-get params :props))))
         (database (plist-get params :database))
         (prop-database (clutch-jdbc--prop-value props "database")))
    (cond
     ((and prop-database database
           (not (string-equal prop-database database)))
      (signal 'clutch-db-error
              (list (format "Conflicting MongoDB database values: :database %S and :props database %S"
                            database prop-database))))
     (prop-database props)
     ((and database (not (string-empty-p database)))
      (cons (cons "database" database) props))
     (t
      (signal 'clutch-db-error
              (list "MongoDB SQL Interface JDBC connections require :database or :props ((\"database\" . ...))"))))))

(defun clutch-jdbc--driver-props (driver params)
  "Return normalized JDBC connection properties for DRIVER and PARAMS."
  (if (clutch-jdbc--mongodb-driver-p driver params)
      (clutch-jdbc--sql-interface-props params)
    (clutch-jdbc--normalize-props (plist-get params :props))))

(defun clutch-jdbc--driver-class (driver params)
  "Return the concrete JDBC driver class for DRIVER and PARAMS."
  (let ((explicit (plist-get params :driver-class)))
    (cond
     ((and (stringp explicit)
           (not (string-empty-p explicit)))
      explicit)
     ((and explicit
           (not (stringp explicit)))
      (signal 'clutch-db-error
              (list ":driver-class must be a non-empty string")))
     ((plist-get (alist-get driver clutch-jdbc--driver-sources) :class))
     (t
      (signal 'clutch-db-error
              (list (format "JDBC backend %S requires :driver-class" driver)))))))

(defun clutch-jdbc--effective-driver (driver params)
  "Return internal JDBC driver for user-facing DRIVER and PARAMS."
  (let ((url (plist-get params :url)))
    (cond
     ((eq driver 'mongodb)
      (if (clutch-backend-jdbc-transport-p 'mongodb params)
          'mongodb
        (signal 'clutch-db-error
                (list "Ordinary MongoDB uses the native mongodb backend; JDBC is only for :surface sql-interface"))))
     ((and (eq driver 'jdbc)
           (stringp url)
           (string-prefix-p "jdbc:oracle:" url t))
      (signal 'clutch-db-error
              (list "Oracle JDBC URLs require :backend oracle, not :backend jdbc")))
     (t driver))))

;;;; Connect function

(defun clutch-jdbc--mongodb-driver-p (driver params)
  "Return non-nil when DRIVER/PARAMS represent MongoDB SQL Interface JDBC."
  (or (and (eq driver 'mongodb)
           (clutch-backend-jdbc-transport-p 'mongodb params))
      (clutch-jdbc--sql-interface-jdbc-url-p
       (plist-get params :url))))

(defun clutch-jdbc--manual-commit-capable-p (driver params)
  "Return non-nil when DRIVER/PARAMS can expose JDBC transaction controls."
  (not (clutch-jdbc--mongodb-driver-p driver params)))

(defun clutch-jdbc--manual-commit-mode (driver params)
  "Return non-nil when DRIVER with PARAMS should use manual-commit mode.
Oracle defaults to manual-commit when `clutch-jdbc-oracle-manual-commit' is
non-nil.  Any driver opts in explicitly via `:manual-commit t' in PARAMS."
  (cond
   ((not (clutch-jdbc--manual-commit-capable-p driver params))
    (when (plist-get params :manual-commit)
      (signal 'clutch-db-error
              (list "MongoDB JDBC does not support Clutch manual-commit controls")))
    nil)
   ((plist-member params :manual-commit)
    (plist-get params :manual-commit))
   (t
    (and (eq driver 'oracle)
         clutch-jdbc-oracle-manual-commit))))

(defun clutch-jdbc--apply-timeout-defaults (params)
  "Return a copy of PARAMS with absent timeout keys set to their global defaults."
  (clutch-db--apply-connect-defaults
   (clutch-db--reject-removed-connect-params params)
   `((:connect-timeout . ,clutch-connect-timeout-seconds)
     (:read-idle-timeout . ,clutch-read-idle-timeout-seconds)
     (:query-timeout . ,clutch-query-timeout-seconds)
     (:rpc-timeout . ,clutch-jdbc-rpc-timeout-seconds))))

(defun clutch-jdbc--setup-prerequisites (driver)
  "Ensure agent jar and DRIVER jar are present."
  ;; Existence check only: the jar's checksum is verified where it counts,
  ;; when the shared JVM actually starts (`clutch-jdbc--start-agent') and
  ;; right after download.  Re-hashing 2.4MB per logical connect bought no
  ;; integrity, and metadata-keyed caching could wave through a replaced
  ;; file whose size and mtime were preserved.
  (let ((jar (clutch-jdbc--agent-jar)))
    (unless (file-exists-p jar)
      (user-error "JDBC agent not found.  Run M-x clutch-jdbc-ensure-agent")))
  (when-let* ((spec (alist-get driver clutch-jdbc--driver-sources))
              (filename (plist-get spec :filename))
              (dest (expand-file-name filename (clutch-jdbc--drivers-dir))))
    (unless (file-exists-p dest)
      (let ((install-driver (clutch-jdbc--driver-install-symbol driver))
            (label (clutch-jdbc--driver-install-label driver)))
        (cond
       ((plist-get spec :maven)
        (user-error
         "%s driver not found.  Run M-x clutch-jdbc-install-driver RET %s"
         label install-driver))
       ((plist-get spec :url)
        (user-error
         "%s driver not found.  Run M-x clutch-jdbc-install-driver RET %s"
         label install-driver))
       ((plist-get spec :manual)
        (user-error "%s driver requires manual download.\nURL: %s\nPlace as: %s"
                    label
                    (plist-get spec :manual) dest)))))))

(defun clutch-jdbc--driver-install-symbol (driver)
  "Return the public install command symbol for internal DRIVER."
  driver)

(defun clutch-jdbc--driver-install-label (driver)
  "Return a user-facing driver label for DRIVER."
  (pcase driver
    ('mongodb "MongoDB SQL Interface")
    (_ (capitalize (symbol-name driver)))))

(defun clutch-db-jdbc-connect (driver params)
  "Connect to a JDBC data source of type DRIVER using PARAMS plist.
DRIVER is a symbol (e.g. \\='oracle, \\='sqlserver) captured by the
registration closure — users do not pass it directly.
Returns a `clutch-jdbc-conn'."
  (let ((driver (clutch-jdbc--effective-driver driver params)))
    (clutch-db-jdbc-connect--internal driver params)))

(defun clutch-db-jdbc-connect--internal (driver params)
  "Connect to JDBC DRIVER using PARAMS after driver dispatch is resolved."
  (unless (or (null clutch-jdbc-validate-after-idle-seconds)
              (and (integerp clutch-jdbc-validate-after-idle-seconds)
                   (>= clutch-jdbc-validate-after-idle-seconds 0)))
    (user-error
     "JDBC idle validation interval must be nil or a non-negative integer"))
  (clutch-jdbc--setup-prerequisites driver)
  (clutch-jdbc--ensure-agent)
  (let* ((normalized-params (clutch-jdbc--apply-timeout-defaults params))
         (manual-commit-p (clutch-jdbc--manual-commit-mode driver normalized-params))
         (url      (clutch-jdbc--build-url driver normalized-params))
         (user     (plist-get normalized-params :user))
         (password (plist-get normalized-params :password))
         (props    (clutch-jdbc--driver-props driver normalized-params))
         (driver-class (clutch-jdbc--driver-class driver normalized-params))
         (connect-timeout (plist-get normalized-params :connect-timeout))
         (read-idle-timeout (plist-get normalized-params :read-idle-timeout))
         (rpc-timeout (plist-get normalized-params :rpc-timeout))
         (result   (clutch-jdbc--rpc
                    "connect"
                    `((url      . ,url)
                      (driver-class . ,driver-class)
                      (user     . ,user)
                      (password . ,password)
                      (auto-commit . ,(if manual-commit-p
                                          clutch-jdbc--json-false
                                        t))
                      (connect-timeout-seconds . ,connect-timeout)
                      ,@(when read-idle-timeout
                          `((network-timeout-seconds . ,read-idle-timeout)))
                      ,@(when (and (integerp clutch-jdbc-validate-after-idle-seconds)
                                   (> clutch-jdbc-validate-after-idle-seconds 0))
                          `((validate-after-idle-seconds
                             . ,clutch-jdbc-validate-after-idle-seconds)))
                      ,@(when props `((props . ,props))))
                    rpc-timeout)))
    (let* ((conn-params (plist-put normalized-params :driver driver))
           (conn-params (if (eq driver 'mongodb)
                            (plist-put conn-params :backend 'mongodb)
                          conn-params))
           (conn-params (if (eq driver 'mongodb)
                            (plist-put conn-params :surface 'sql-interface)
                          conn-params))
           (conn (make-clutch-jdbc-conn
                  :process  clutch-jdbc--agent-process
                  :conn-id  (plist-get result :conn-id)
                  :driver   driver
                  :params   conn-params
                  :busy     nil)))
      (puthash (clutch-jdbc-conn-conn-id conn) conn clutch-jdbc--connections-by-id)
      conn)))

;;;; Register backend

;; Each JDBC driver gets its own closure so the driver type is available
;; inside clutch-db-jdbc-connect without requiring a redundant :driver key
;; in the user's params plist (:backend is stripped by clutch--build-conn
;; before the connect-fn is called).
(dolist (entry clutch-jdbc--driver-metadata)
  (let ((driver (car entry)))
    (unless (alist-get driver clutch-backend--registry)
      (add-to-list
       'clutch-backend--registry
       (cons driver
             (append (cdr entry)
                     (list :require 'clutch-db-jdbc
                           :connect-fn
                           (lambda (p)
                             (clutch-db-jdbc-connect driver p)))))
       t))))

;;;; Lifecycle methods

(cl-defmethod clutch-db-disconnect ((conn clutch-jdbc-conn))
  "Disconnect JDBC CONN, releasing it in the agent."
  (remhash conn clutch-jdbc--busy-request-ids)
  (clutch-jdbc--clear-async-callbacks conn)
  (remhash conn clutch-jdbc--error-details-by-conn)
  (remhash (clutch-jdbc-conn-conn-id conn) clutch-jdbc--connections-by-id)
  (when (clutch-jdbc--agent-live-p)
    (let ((id (clutch-jdbc--send
               "disconnect"
               `((conn-id . ,(clutch-jdbc-conn-conn-id conn))))))
      (clutch-jdbc--recv-response-nonfatal
       id clutch-jdbc-disconnect-timeout-seconds))))

(cl-defmethod clutch-db-live-p ((conn clutch-jdbc-conn))
  "Return non-nil if the agent process is running and CONN belongs to it.
Checks both that the stored process is live AND that it is still the
current agent process.  After a timeout kill, `clutch-jdbc--agent-process'
is set to nil before the old JVM fully exits, so the old process may still
pass `process-live-p' briefly; the identity check closes that window."
  (and (clutch-jdbc-conn-p conn)
       clutch-jdbc--agent-process
       (eq (clutch-jdbc-conn-process conn) clutch-jdbc--agent-process)
       (eq conn
           (gethash (clutch-jdbc-conn-conn-id conn)
                    clutch-jdbc--connections-by-id))
       (process-live-p (clutch-jdbc-conn-process conn))))

(cl-defmethod clutch-db-backend-key ((conn clutch-jdbc-conn))
  "Return the registered backend key for JDBC CONN."
  (let ((driver (clutch-jdbc--conn-driver conn)))
    (if (eq driver 'mongodb)
        'mongodb
      (or driver 'jdbc))))

(cl-defmethod clutch-db-error-details ((conn clutch-jdbc-conn))
  "Return the latest structured error details snapshot for JDBC CONN."
  (when-let* ((details (gethash conn clutch-jdbc--error-details-by-conn)))
    (copy-tree details)))

(cl-defmethod clutch-db-clear-error-details ((conn clutch-jdbc-conn))
  "Forget the latest structured error details snapshot for JDBC CONN."
  (remhash conn clutch-jdbc--error-details-by-conn))

(cl-defmethod clutch-db-init-connection ((_conn clutch-jdbc-conn))
  "No post-connect initialization needed for JDBC connections.")

(defun clutch-jdbc--conn-rpc-timeout (conn)
  "Return the RPC timeout in seconds for CONN.
Always non-nil: `clutch-jdbc--apply-timeout-defaults' ensures the value is
stored in params at connect time."
  (plist-get (clutch-jdbc-conn-params conn) :rpc-timeout))

(defun clutch-jdbc--conn-effective-query-timeout (conn)
  "Return the effective query timeout in seconds for CONN, or nil.
The timeout is clamped so the agent-side timeout fires before the outer
Emacs RPC timeout."
  (let* ((query-timeout (plist-get (clutch-jdbc-conn-params conn) :query-timeout))
         (rpc-timeout   (clutch-jdbc--conn-rpc-timeout conn)))
    (when (and query-timeout (> query-timeout 0))
      (min query-timeout (max 1 (- rpc-timeout 5))))))

(defun clutch-jdbc--oracle-conn-p (conn)
  "Return non-nil when CONN is an Oracle JDBC connection."
  (eq (clutch-jdbc--conn-driver conn) 'oracle))

(defun clutch-jdbc--url-prefix-p (conn prefix)
  "Return non-nil for JDBC CONN URLs with PREFIX."
  (let ((url (plist-get (clutch-jdbc-conn-params conn) :url)))
    (and (stringp url)
         (string-prefix-p prefix url))))

(defun clutch-jdbc--duckdb-conn-p (conn)
  "Return non-nil when CONN is a DuckDB JDBC connection."
  (clutch-jdbc--url-prefix-p conn "jdbc:duckdb:"))

(defun clutch-jdbc--limit-offset-conn-p (conn)
  "Return non-nil when JDBC CONN should use LIMIT/OFFSET pagination."
  (let ((driver (clutch-jdbc--conn-driver conn)))
    (or (memq driver '(redshift clickhouse))
        (clutch-jdbc--mongodb-conn-p conn)
        (clutch-jdbc--duckdb-conn-p conn)
        (clutch-jdbc--url-prefix-p conn "jdbc:redshift:")
        (clutch-jdbc--url-prefix-p conn "jdbc:clickhouse:"))))

(defun clutch-jdbc--clickhouse-conn-p (conn)
  "Return non-nil when CONN is a ClickHouse JDBC connection."
  (eq (clutch-jdbc--conn-driver conn) 'clickhouse))

(defun clutch-jdbc--mongodb-conn-p (conn)
  "Return non-nil when CONN is any MongoDB JDBC connection."
  (clutch-jdbc--mongodb-driver-p
   (clutch-jdbc--conn-driver conn)
   (clutch-jdbc-conn-params conn)))

(defun clutch-jdbc--metadata-table-comments-supported-p (conn)
  "Return non-nil when CONN can use JDBC metadata table remarks."
  (not (or (clutch-jdbc--oracle-conn-p conn)
           (clutch-jdbc--clickhouse-conn-p conn))))

(defun clutch-jdbc--metadata-conn-for-scope (conn schema &optional catalog)
  "Return a JDBC metadata view of CONN scoped to SCHEMA and CATALOG."
  (if (and (or (not schema)
               (equal schema (clutch-jdbc--conn-schema conn)))
           (or (not catalog)
               (equal catalog (clutch-jdbc--conn-catalog conn))))
      conn
    (let ((metadata-conn (copy-clutch-jdbc-conn conn)))
      (let ((params (copy-sequence (clutch-jdbc-conn-params conn))))
        (when schema
          (setq params (plist-put params :schema schema)))
        (when catalog
          (setq params (plist-put params :catalog catalog)))
        (setf (clutch-jdbc-conn-params metadata-conn) params))
      metadata-conn)))

(defun clutch-jdbc--table-like-entry-p (conn entry)
  "Return non-nil when ENTRY should be used as a table-like object for CONN."
  (ignore conn)
  (clutch-jdbc--entry-type= entry "TABLE"))

(defun clutch-jdbc--table-entry-names (conn entries)
  "Return table names from ENTRIES visible as table-like objects for CONN."
  (mapcar (lambda (entry) (plist-get entry :name))
          (seq-filter (lambda (entry)
                        (clutch-jdbc--table-like-entry-p conn entry))
                      entries)))

(defun clutch-jdbc--clickhouse-simple-identifier-p (name)
  "Return non-nil when NAME is a bare ClickHouse identifier."
  (and (stringp name)
       (string-match-p "\\`[A-Za-z_][A-Za-z0-9_]*\\'" name)))

(defun clutch-jdbc--clickhouse-escape-identifier (name)
  "Escape ClickHouse identifier NAME.
Leave simple identifiers bare so generated SQL matches common ClickHouse usage;
fall back to backticks when quoting is required."
  (if (clutch-jdbc--clickhouse-simple-identifier-p name)
      name
    (format "`%s`" (replace-regexp-in-string "`" "``" name))))

(defun clutch-jdbc--url-metadata (url)
  "Return endpoint metadata parsed from JDBC URL, or nil.
Supports common `jdbc:subprotocol://host[:port]/database' URLs."
  (when (and url
             (string-match
              "\\`\\(?:jdbc:\\)?[^:]+://\\([^/:;?]+\\)\\(?::\\([0-9]+\\)\\)?/\\([^/?;]+\\)"
              url))
    (list :host (match-string 1 url)
          :port (when-let* ((port (match-string 2 url)))
                  (string-to-number port))
          :database (match-string 3 url))))

(defconst clutch-jdbc--oracle-system-schemas
  '("SYS" "SYSTEM" "XDB" "MDSYS" "CTXSYS" "LBACSYS" "OLAPSYS"
    "WMSYS" "DBSNMP" "APPQOSSYS" "AUDSYS" "DVSYS"
    "GSMADMIN_INTERNAL" "OJVMSYS" "OUTLN")
  "Oracle schemas hidden from interactive schema switching by default.")

(cl-defmethod clutch-db-manual-commit-p ((conn clutch-jdbc-conn))
  "Return non-nil for JDBC CONN with auto-commit disabled."
  (let ((params (clutch-jdbc-conn-params conn)))
    (and (clutch-jdbc--manual-commit-capable-p
          (clutch-jdbc--conn-driver conn) params)
         (clutch-jdbc--manual-commit-mode
          (clutch-jdbc--conn-driver conn) params))))

(cl-defmethod clutch-db-manual-commit-supported-p ((conn clutch-jdbc-conn))
  "Return non-nil for JDBC CONN runtime auto-commit toggling."
  (let ((params (clutch-jdbc-conn-params conn)))
    (clutch-jdbc--manual-commit-capable-p
     (clutch-jdbc--conn-driver conn) params)))

(cl-defmethod clutch-db-commit ((conn clutch-jdbc-conn))
  "Commit the current transaction on JDBC CONN."
  (unless (clutch-db-manual-commit-supported-p conn)
    (user-error "Manual commit is not supported by this connection"))
  (clutch-jdbc--rpc "commit"
                    `((conn-id . ,(clutch-jdbc-conn-conn-id conn)))
                    (clutch-jdbc--conn-rpc-timeout conn)))

(cl-defmethod clutch-db-rollback ((conn clutch-jdbc-conn))
  "Roll back the current transaction on JDBC CONN."
  (unless (clutch-db-manual-commit-supported-p conn)
    (user-error "Manual commit is not supported by this connection"))
  (clutch-jdbc--rpc "rollback"
                    `((conn-id . ,(clutch-jdbc-conn-conn-id conn)))
                    (clutch-jdbc--conn-rpc-timeout conn)))

(defun clutch-jdbc--set-auto-commit-rpc (conn auto-commit)
  "Set JDBC CONN's remote auto-commit state to AUTO-COMMIT.
Unlike `clutch-db-set-auto-commit', this internal primitive does not change the
user-selected transaction mode stored in CONN."
  (clutch-jdbc--rpc "set-auto-commit"
                    `((conn-id    . ,(clutch-jdbc-conn-conn-id conn))
                      (auto-commit . ,(if auto-commit
                                          t
                                        clutch-jdbc--json-false)))
                    (clutch-jdbc--conn-rpc-timeout conn)))

(cl-defmethod clutch-db-set-auto-commit ((conn clutch-jdbc-conn) auto-commit)
  "Set auto-commit mode on JDBC CONN.
AUTO-COMMIT non-nil enables auto-commit (disables manual-commit); nil
enables manual-commit.  When switching to auto-commit, the JDBC driver
commits any pending transaction per the JDBC specification."
  (unless (clutch-db-manual-commit-supported-p conn)
    (user-error "Manual commit is not supported by this connection"))
  (clutch-jdbc--set-auto-commit-rpc conn auto-commit)
  (setf (clutch-jdbc-conn-params conn)
        (plist-put (clutch-jdbc-conn-params conn) :manual-commit (not auto-commit))))

(cl-defmethod clutch-db-call-with-atomic-batch
  ((conn clutch-jdbc-conn) function)
  "Call FUNCTION atomically in the selected JDBC transaction mode on CONN."
  (unless (clutch-db-manual-commit-supported-p conn)
    (user-error "Atomic mutation submission is not supported by this connection"))
  (if (clutch-db-manual-commit-p conn)
      (let (savepoint-id)
        (clutch-db--call-with-savepoint-boundary
         (lambda ()
           (setq savepoint-id
                 (plist-get
                  (clutch-jdbc--rpc
                   "create-savepoint"
                   `((conn-id . ,(clutch-jdbc-conn-conn-id conn)))
                   (clutch-jdbc--conn-rpc-timeout conn))
                  :savepoint-id)))
         function
         (lambda ()
           (clutch-jdbc--rpc
            "release-savepoint"
            `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
              (savepoint-id . ,savepoint-id))
            (clutch-jdbc--conn-rpc-timeout conn)))
         (lambda ()
           (clutch-jdbc--rpc
            "rollback-savepoint"
            `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
              (savepoint-id . ,savepoint-id))
            (clutch-jdbc--conn-rpc-timeout conn)))))
    (clutch-jdbc--set-auto-commit-rpc conn nil)
    (let (outcome)
      (unwind-protect
          (clutch-db--call-with-transaction-boundary
           #'ignore
           function
           (lambda ()
             (clutch-db-commit conn)
             (setq outcome 'committed))
           (lambda ()
             (clutch-db-rollback conn)
             (setq outcome 'rolled-back)))
        (if outcome
            (condition-case restore-error
                (clutch-jdbc--set-auto-commit-rpc conn t)
              ((error quit)
               (setf (clutch-jdbc-conn-params conn)
                     (plist-put (clutch-jdbc-conn-params conn)
                                :manual-commit t))
               (signal
                'clutch-db-session-restore-error
                (list
                 (format "Mutation batch was %s, but restoring auto-commit failed: %s"
                         (if (eq outcome 'committed) "committed" "rolled back")
                         (error-message-string restore-error))
                 :outcome outcome))))
          ;; An unknown transaction outcome leaves the remote session in
          ;; Manual mode until the user explicitly recovers it.
          (setf (clutch-jdbc-conn-params conn)
                (plist-put (clutch-jdbc-conn-params conn)
                           :manual-commit t)))))))

(cl-defmethod clutch-db-schema-transaction-effect ((conn clutch-jdbc-conn) _sql)
  "Return schema SQL transaction effect for JDBC CONN.
Oracle DDL commits the transaction; other JDBC drivers keep the default
unknown effect."
  (when (clutch-jdbc--oracle-conn-p conn)
    'clear))

(cl-defmethod clutch-db-eager-schema-refresh-p ((conn clutch-jdbc-conn))
  "Return non-nil when CONN should refresh schema eagerly.
Oracle JDBC schema enumeration is too slow to block connect."
  (not (clutch-jdbc--oracle-conn-p conn)))

(cl-defmethod clutch-db-completion-sync-columns-p ((conn clutch-jdbc-conn))
  "Return non-nil when CONN may synchronously load completion columns.
This is allowed in the hot path."
  (not (clutch-jdbc--oracle-conn-p conn)))

;;;; Query methods

(defun clutch-jdbc--effective-fetch-size ()
  "Return the validated JDBC fetch batch size."
  (unless (and (integerp clutch-jdbc-fetch-size)
               (<= 1 clutch-jdbc-fetch-size clutch-jdbc--max-fetch-size))
    (user-error "JDBC fetch size must be between 1 and %d"
                clutch-jdbc--max-fetch-size))
  clutch-jdbc-fetch-size)

(defun clutch-jdbc--fetch-all (conn cursor-id)
  "Fetch all remaining rows for CURSOR-ID on CONN, returning a flat list."
  (let ((rpc-timeout (clutch-jdbc--conn-rpc-timeout conn))
        (effective-qt (clutch-jdbc--conn-effective-query-timeout conn))
        (fetch-size (clutch-jdbc--effective-fetch-size))
        batches done)
    (while (not done)
      (let ((result (clutch-jdbc--rpc-on-conn
                     conn
                     "fetch"
                     `((cursor-id  . ,cursor-id)
                       (fetch-size . ,fetch-size)
                       ,@(when effective-qt
                           `((query-timeout-seconds . ,effective-qt))))
                     rpc-timeout)))
        (push (plist-get result :rows) batches)
        (setq done (eq t (plist-get result :done)))))
    (apply #'nconc (nreverse batches))))

(defun clutch-jdbc--collect-table-entries (conn result)
  "Return normalized table entry plists from get-tables RESULT on CONN.
Supports both the current plist-list payload under :tables and the older
cursor-style :rows format used in tests."
  (mapcar (lambda (entry)
            (clutch-jdbc--normalize-table-entry conn entry))
          (or (plist-get result :tables)
              (mapcar
               #'clutch-jdbc--table-entry-from-row
               (let* ((first-rows (plist-get result :rows))
                      (cursor-id  (plist-get result :cursor-id)))
                 (if (eq t (plist-get result :done))
                     first-rows
                   (nconc first-rows (clutch-jdbc--fetch-all conn cursor-id))))))))

(defun clutch-jdbc--normalize-table-entry (conn entry)
  "Return table ENTRY normalized for JDBC CONN."
  (ignore conn)
  (copy-sequence entry))

(defun clutch-jdbc--entry-type= (entry type)
  "Return non-nil when ENTRY has TYPE, case-insensitively."
  (string-equal (upcase (or (plist-get entry :type) ""))
                (upcase type)))

(defun clutch-jdbc--table-entry-from-row (row)
  "Convert a get-tables ROW into a table entry plist."
  (pcase-let ((`(,name ,type ,schema ,src-schema . ,rest) row))
    (let ((entry (list :name name
                       :type type
                       :schema schema
                       :source-schema (or src-schema schema))))
      (if rest
          (append entry (list :comment (car rest)))
        entry))))

(defun clutch-jdbc--table-comment-from-entries (table entries &optional schema)
  "Return TABLE's unambiguous comment from JDBC ENTRIES in SCHEMA."
  (let ((matches
         (cl-loop for entry in entries
                  when (and (string= (downcase table)
                                     (downcase (or (plist-get entry :name) "")))
                            (or (not schema)
                                (string= (downcase schema)
                                         (downcase (or (plist-get entry :schema)
                                                       "")))))
                  collect entry)))
    (when (= (length matches) 1)
      (plist-get (car matches) :comment))))

(defun clutch-jdbc--type-category (jdbc-type-name)
  "Map JDBC-TYPE-NAME to a `clutch-db' type-category symbol."
  (let ((t-upper (upcase (or jdbc-type-name ""))))
    (cond
     ((string-match-p "INT\\|SMALLINT\\|BIGINT\\|TINYINT\\|NUMBER\\|NUMERIC\\|DECIMAL\\|FLOAT\\|DOUBLE\\|REAL" t-upper) 'numeric)
     ((string-match-p "BOOL" t-upper)                           'text)
     ((string-match-p "JSON\\|DOCUMENT\\|ARRAY" t-upper)        'json)
     ((string-match-p "BLOB\\|BINARY\\|VARBINARY\\|RAW\\|IMAGE\\|BINDATA" t-upper) 'blob)
     ((string-match-p "TIMESTAMP\\|DATETIME" t-upper)           'datetime)
     ((string-match-p "DATE" t-upper)                           'date)
     ((string-match-p "TIME$" t-upper)                          'time)
     (t                                                         'text))))

(defun clutch-jdbc--make-columns (col-names col-types)
  "Build clutch-db column plists from agent COL-NAMES and COL-TYPES lists."
  (cl-mapcar (lambda (name type)
               (list :name name
                     :backend-type type
                     :type-category (clutch-jdbc--type-category type)))
             col-names col-types))

(defun clutch-jdbc--normalize-props (props)
  "Return PROPS as an alist with string keys, suitable for JSON encoding.
Accepts either an alist ((\"key\" . val) ...) or a plist (:key val ...)."
  (when props
    (if (consp (car props))
        props
      (cl-loop for (k v) on props by #'cddr
               collect (cons (substring (symbol-name k) 1) v)))))

(defun clutch-jdbc--normalize-json-false (value)
  "Return VALUE with JDBC JSON false sentinels replaced by `:false'.
Recurses into lists and vectors so nested JSON cell values are normalized."
  (cond
   ((eq value clutch-jdbc--json-false) :false)
   ((consp value)
    (mapcar #'clutch-jdbc--normalize-json-false value))
   ((vectorp value)
    (vconcat (mapcar #'clutch-jdbc--normalize-json-false value)))
   (t value)))

(defun clutch-jdbc--normalize-row (row)
  "Convert JDBC-specific value representations in ROW to generic forms.
Blob plists with :text content become plain strings.
Clob plists become their :preview string.
JDBC JSON false sentinels become `:false'."
  (mapcar (lambda (val)
            (clutch-jdbc--normalize-json-false
             (cond
              ((and (listp val)
                    (equal (plist-get val :__type) "blob")
                    (plist-get val :text))
               (let ((text (copy-sequence (plist-get val :text)))
                     (encoding (plist-get val :encoding)))
                 (when (and encoding (> (length text) 0))
                   (put-text-property 0 (length text)
                                      'clutch-jdbc-blob-encoding encoding text))
                 text))
              ((and (listp val)
                    (equal (plist-get val :__type) "clob"))
               (plist-get val :preview))
              (t val))))
          row))

(defun clutch-jdbc--normalize-column-details (columns)
  "Return JDBC COLUMNS with exact types on the generic :backend-type key."
  (mapcar
   (lambda (column)
     (let ((normalized (copy-sequence column)))
       (when-let* ((type (plist-get normalized :type)))
         (setq normalized (plist-put normalized :backend-type type)))
       normalized))
   columns))

(defun clutch-jdbc--json-bool (value)
  "Normalize VALUE decoded from JSON into an Elisp boolean."
  (and value (not (eq value clutch-jdbc--json-false))))

(defun clutch-jdbc--normalize-object-entry (entry)
  "Normalize agent ENTRY field names for the generic clutch object schema."
  (let ((normalized (copy-sequence entry)))
    (when-let* ((table (plist-get normalized :table)))
      (setq normalized (plist-put normalized :target-table table)))
    (when (plist-member normalized :unique)
      (setq normalized
            (plist-put normalized :unique
                       (clutch-jdbc--json-bool (plist-get normalized :unique)))))
    normalized))

(defun clutch-jdbc--object-category-spec (category)
  "Return JDBC object RPC spec plist for CATEGORY, or nil."
  (alist-get category clutch-jdbc--object-category-specs))

(defun clutch-jdbc--rpc-result (conn result)
  "Convert JDBC RPC RESULT for CONN to a `clutch-db-result'."
  (if (equal (plist-get result :type) "dml")
      (make-clutch-db-result
       :connection conn
       :affected-rows (plist-get result :affected-rows))
    (let* ((first-rows (plist-get result :rows))
           (cursor-id (plist-get result :cursor-id))
           (done (eq t (plist-get result :done)))
           (all-rows (if done first-rows
                       (nconc first-rows
                              (clutch-jdbc--fetch-all conn cursor-id)))))
      (make-clutch-db-result
       :connection conn
       :columns (clutch-jdbc--make-columns
                 (plist-get result :columns)
                 (plist-get result :col-types))
       :rows (mapcar #'clutch-jdbc--normalize-row all-rows)))))

(defun clutch-jdbc--execute-rpc (conn op payload)
  "Execute JDBC RPC OP with PAYLOAD on CONN and return a database result."
  (setf (clutch-jdbc-conn-busy conn) t)
  (unwind-protect
      (let* ((rpc-timeout   (clutch-jdbc--conn-rpc-timeout conn))
             (effective-qt  (clutch-jdbc--conn-effective-query-timeout conn))
             (fetch-size (clutch-jdbc--effective-fetch-size))
             (result (clutch-jdbc--rpc-on-conn
                      conn
                      op
                      (append
                       `((conn-id . ,(clutch-jdbc-conn-conn-id conn)))
                       payload
                       `((fetch-size . ,fetch-size))
                       (when effective-qt
                         `((query-timeout-seconds . ,effective-qt))))
                      rpc-timeout)))
        (clutch-jdbc--rpc-result conn result))
    (setf (clutch-jdbc-conn-busy conn) nil)))

(cl-defmethod clutch-db-query ((conn clutch-jdbc-conn) sql)
  "Execute SQL on JDBC CONN and return a `clutch-db-result'."
  (clutch-jdbc--execute-rpc conn "execute" `((sql . ,sql))))

(defconst clutch-jdbc--binary-param-type-names
  '("BLOB" "TINYBLOB" "MEDIUMBLOB" "LONGBLOB" "BINARY LARGE OBJECT"
    "RAW" "LONG RAW" "BINARY" "VARBINARY" "LONGVARBINARY"
    "BINARY VARYING" "IMAGE" "BINDATA" "BYTEA")
  "Exact JDBC type names supported by the binary parameter protocol.")

(defun clutch-jdbc--binary-param-p (param)
  "Return non-nil when PARAM carries a JDBC binary backend type."
  (when-let* ((type (clutch-db-param-type param))
              ((stringp type)))
    (and (member (upcase (string-trim type))
                 clutch-jdbc--binary-param-type-names)
         t)))

(defun clutch-jdbc--blob-value-coding-system (value)
  "Return the coding system for text BLOB VALUE."
  (let* ((encoding (and (> (length value) 0)
                        (get-text-property
                         0 'clutch-jdbc-blob-encoding value)))
         (coding (and encoding (coding-system-from-name encoding))))
    (when (and encoding (not coding))
      (user-error "Unsupported JDBC BLOB encoding: %s" encoding))
    (or coding 'utf-8)))

(defun clutch-jdbc--binary-param-bytes (value)
  "Return VALUE as an unibyte string for JDBC binary parameter transport."
  (cond
   ((stringp value)
    (if (multibyte-string-p value)
        (encode-coding-string
         value (clutch-jdbc--blob-value-coding-system value))
      value))
   ((vectorp value)
    (condition-case nil
        (apply #'unibyte-string (append value nil))
      (error
       (user-error "JDBC binary vector values must contain bytes 0..255"))))
   (t
    (user-error "JDBC binary parameters require text, bytes, or NULL"))))

(defun clutch-jdbc--wire-param (param)
  "Return PARAM encoded for the JDBC agent prepared-value protocol."
  (let ((value (clutch-db-param-value param)))
    (if (clutch-jdbc--binary-param-p param)
        `((__clutch_jdbc_param . "binary")
          (jdbc-type . ,(clutch-db-param-type param))
          (base64 . ,(and value
                          (base64-encode-string
                           (clutch-jdbc--binary-param-bytes value) t))))
      value)))

(cl-defmethod clutch-db-execute-params ((conn clutch-jdbc-conn) sql params)
  "Execute SQL with positional PARAMS on CONN through JDBC prepared binding."
  (clutch-jdbc--execute-rpc
   conn "execute-params"
   `((sql . ,sql)
     (values . ,(mapcar #'clutch-jdbc--wire-param params)))))

(cl-defmethod clutch-db-interrupt-query ((conn clutch-jdbc-conn))
  "Interrupt the active JDBC request on CONN without dropping the session."
  (when-let* ((request-id (gethash conn clutch-jdbc--busy-request-ids)))
    (puthash request-id t clutch-jdbc--ignored-response-ids)
    (remhash conn clutch-jdbc--busy-request-ids)
    (let* ((id (clutch-jdbc--send "cancel"
                                  `((conn-id . ,(clutch-jdbc-conn-conn-id conn)))))
           (response (clutch-jdbc--recv-response-nonfatal
                      id clutch-jdbc-cancel-timeout-seconds)))
      (when (eq t (plist-get response :ok))
        (let ((result (plist-get response :result)))
          (and (eq t (plist-get result :cancelled))
               (eql request-id (plist-get result :request-id))))))))

(defun clutch-jdbc--build-oracle-paged-sql (conn base offset page-size order-by)
  "Build Oracle ROWNUM-based pagination SQL for CONN.
BASE is the unpaginated SQL, and PAGE-SIZE bounds each page.
Compatible with all Oracle versions (9i+).  Page N (OFFSET>0) adds
an rn column as a side effect."
  (let ((inner (if order-by
                   (format "%s ORDER BY %s %s" base
                           (clutch-db-escape-identifier conn (car order-by))
                           (cdr order-by))
                 base)))
    (if (= offset 0)
        (format "SELECT * FROM (%s) WHERE ROWNUM <= %d" inner page-size)
      (format "SELECT * FROM (SELECT t.*, ROWNUM rn FROM (%s) t WHERE ROWNUM <= %d) WHERE rn > %d"
              inner (+ offset page-size) offset))))

(cl-defmethod clutch-db-build-paged-sql ((conn clutch-jdbc-conn) base-sql
                                         page-num page-size
                                         &optional order-by page-offset)
  "Build a paginated SQL query for JDBC CONN from BASE-SQL.
PAGE-NUM is zero-based, and PAGE-SIZE limits each page.  Oracle uses
ROWNUM subquery syntax compatible with all Oracle versions.  ORDER-BY
controls the optional sort clause.  PAGE-OFFSET overrides PAGE-NUM
when non-nil.  DuckDB, Redshift, and ClickHouse JDBC use LIMIT/OFFSET.
Other databases use SQL:2011 OFFSET/FETCH (Oracle 12c+, SQL Server
2012+, DB2)."
  (cond
   ((clutch-jdbc--limit-offset-conn-p conn)
    ;; Keep this dialect family explicit; generic JDBC covers engines with
    ;; materially different pagination syntax under one transport.
    (clutch-db--build-limit-offset-paged-sql
     base-sql page-num page-size order-by
     (lambda (name) (clutch-db-escape-identifier conn name))
     page-offset))
   ((clutch-db-sql-has-top-level-limit-p base-sql)
    base-sql)
   (t
    (let* ((trimmed (string-trim-right
                     (replace-regexp-in-string ";\\s-*\\'" "" base-sql)))
           (has-user-order-by
            (clutch-db-sql-has-top-level-clause-p trimmed "ORDER\\s-+BY"))
           (sortable-sql (if order-by
                             (clutch-db-sql-strip-top-level-order-by trimmed)
                           trimmed))
           (offset  (or page-offset (* page-num page-size)))
           (oracle-p (clutch-jdbc--oracle-conn-p conn)))
      (if oracle-p
          (clutch-jdbc--build-oracle-paged-sql conn sortable-sql offset page-size order-by)
        (let ((order-clause (cond
                             (order-by
                              (format " ORDER BY %s %s"
                                      (clutch-db-escape-identifier conn (car order-by))
                                      (cdr order-by)))
                             (has-user-order-by "")
                             (t " ORDER BY (SELECT NULL)"))))
          (format "%s%s OFFSET %d ROWS FETCH NEXT %d ROWS ONLY"
                  sortable-sql order-clause offset page-size)))))))

;;;; SQL dialect methods

(cl-defmethod clutch-db-escape-identifier ((conn clutch-jdbc-conn) name)
  "Escape NAME as a SQL identifier for CONN using double quotes (ANSI standard)."
  (if (clutch-jdbc--clickhouse-conn-p conn)
      (clutch-jdbc--clickhouse-escape-identifier name)
    (format "\"%s\"" (replace-regexp-in-string "\"" "\"\"" name))))

(cl-defmethod clutch-db--source-table-name ((conn clutch-jdbc-conn) token)
  "Return source table name for JDBC CONN and SQL table TOKEN."
  (let ((name (cl-call-next-method)))
    (if (and (clutch-jdbc--oracle-conn-p conn)
             (not (string-match-p "\\`\\s-*\""
                                  (clutch-db-sql-table-qualifier token))))
        (upcase name)
      name)))

(cl-defmethod clutch-db--source-table-schema ((conn clutch-jdbc-conn) token)
  "Return source schema for JDBC CONN and SQL table TOKEN."
  (let ((schema (clutch-db-sql-table-schema token)))
    (if (and schema
             (clutch-jdbc--oracle-conn-p conn)
             (not (string-prefix-p "\"" token)))
        (upcase schema)
      schema)))

(cl-defmethod clutch-db--source-table-catalog ((_conn clutch-jdbc-conn) token)
  "Return source catalog for JDBC SQL table TOKEN."
  (clutch-db-sql-table-catalog token))

(cl-defmethod clutch-db-derived-table-alias ((conn clutch-jdbc-conn) alias)
  "Return a JDBC derived-table alias clause for ALIAS on CONN.
Oracle does not accept AS before a subquery alias."
  (if (clutch-jdbc--oracle-conn-p conn)
      (replace-regexp-in-string "\\`_+" "" alias)
    (cl-call-next-method)))

(defun clutch-jdbc--oracle-display-identifier (name)
  "Return Oracle DDL display form for identifier NAME.
Leave simple uppercase identifiers unquoted so reconstructed DDL reads
closer to Oracle's native style."
  (let ((case-fold-search nil))
    (if (and (string-match-p "\\`[A-Z][A-Z0-9_$#]*\\'" name)
             (not (clutch-jdbc--oracle-display-keyword-p name)))
        name
      (format "\"%s\"" (replace-regexp-in-string "\"" "\"\"" name)))))

(defvar clutch-jdbc--oracle-display-keyword-cache (make-hash-table :test 'equal)
  "Cache of Oracle identifiers that should remain quoted in display DDL.")

(defun clutch-jdbc--oracle-display-keyword-p (name)
  "Return non-nil when bare NAME would be fontified as Oracle syntax.
Such identifiers should remain quoted in reconstructed Oracle DDL."
  (let ((cached (gethash name clutch-jdbc--oracle-display-keyword-cache 'missing)))
    (if (eq cached 'missing)
        (let ((keywordp
               (with-temp-buffer
                 (sql-mode)
                 (make-local-variable 'sql-product)
                 (sql-set-product 'oracle)
                 (insert name)
                 (font-lock-ensure)
                 (let ((face (get-text-property (point-min) 'face)))
                   (or (eq face 'font-lock-keyword-face)
                       (eq face 'font-lock-builtin-face)
                       (and (listp face)
                            (or (memq 'font-lock-keyword-face face)
                                (memq 'font-lock-builtin-face face))))))))
          (puthash name keywordp clutch-jdbc--oracle-display-keyword-cache)
          keywordp)
      cached)))

(defconst clutch-jdbc--backslash-escape-drivers '(clickhouse redshift snowflake)
  "JDBC drivers whose string literals read a backslash as an escape.
Doubling the quote alone leaves a trailing backslash escaping the closing
quote on these engines.  Redshift is here because its PostgreSQL 8.0
lineage predates standard_conforming_strings, so escape processing is
always on.  Drivers absent from this list keep backslash literal, where
doubling it would store a character the user never typed.")

(cl-defmethod clutch-db-escape-literal ((conn clutch-jdbc-conn) value)
  "Escape VALUE as a SQL string literal for CONN.
Quote doubling is accepted by every supported engine; backslash handling
follows `clutch-jdbc--backslash-escape-drivers'."
  (let ((escaped (replace-regexp-in-string "'" "''" value)))
    (when (memq (clutch-jdbc--conn-driver conn)
                clutch-jdbc--backslash-escape-drivers)
      (setq escaped (replace-regexp-in-string "\\\\" "\\\\\\\\" escaped)))
    (format "'%s'" escaped)))

;;;; Schema methods

(defun clutch-jdbc--default-schema (conn)
  "Return a default schema filter for CONN, or nil for no filtering.
Oracle uses the username as schema (uppercased).  Other backends return nil."
  (when (clutch-jdbc--oracle-conn-p conn)
    (when-let* ((user (plist-get (clutch-jdbc-conn-params conn) :user)))
      (upcase user))))

(defun clutch-jdbc--conn-schema (conn)
  "Return the effective schema for CONN: explicit :schema or the Oracle default."
  (or (plist-get (clutch-jdbc-conn-params conn) :schema)
      (clutch-jdbc--default-schema conn)))

(defun clutch-jdbc--conn-catalog (conn)
  "Return the effective catalog for CONN, or nil when unused.
ClickHouse maps its current database to JDBC catalog, not schema."
  (or (plist-get (clutch-jdbc-conn-params conn) :catalog)
      (when (clutch-jdbc--clickhouse-conn-p conn)
        (or (plist-get (clutch-jdbc-conn-params conn) :database)
            "default"))))

(defun clutch-jdbc--metadata-scope-params (conn)
  "Return optional JDBC metadata scope params for CONN.
ClickHouse omits catalog because its JDBC metadata filter drops rows
when a catalog is supplied."
  (let ((catalog (unless (clutch-jdbc--clickhouse-conn-p conn)
                   (clutch-jdbc--conn-catalog conn)))
        (schema (clutch-jdbc--conn-schema conn)))
    (append (when catalog `((catalog . ,catalog)))
            (when schema `((schema . ,schema))))))

(defun clutch-jdbc--table-metadata-params (conn table)
  "Return JDBC metadata params for TABLE on CONN."
  `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
    (table . ,table)
    ,@(clutch-jdbc--metadata-scope-params conn)))

(defun clutch-jdbc--visible-schemas (conn schemas)
  "Normalize visible SCHEMAS for CONN."
  (let* ((current (clutch-jdbc--conn-schema conn))
         (schemas (delete-dups
                   (seq-filter
                    (lambda (schema)
                      (and (stringp schema)
                           (not (string-empty-p schema))
                           (or (not (clutch-jdbc--oracle-conn-p conn))
                               (not (member-ignore-case schema
                                                        clutch-jdbc--oracle-system-schemas)))))
                    schemas))))
    (sort schemas
          (lambda (a b)
            (cond
             ((and current (string= (downcase a) (downcase current))) t)
             ((and current (string= (downcase b) (downcase current))) nil)
             (t (string-collate-lessp a b)))))))

(defun clutch-jdbc--remember-duckdb-namespace (conn catalog schema)
  "Store DuckDB CATALOG and SCHEMA on CONN, then return SCHEMA."
  (let ((params (copy-sequence (clutch-jdbc-conn-params conn))))
    (setq params (plist-put params :catalog catalog))
    (setq params (plist-put params :schema schema))
    (setf (clutch-jdbc-conn-params conn) params))
  schema)

(cl-defmethod clutch-db-list-tables ((conn clutch-jdbc-conn))
  "Return table names for JDBC CONN.
For Oracle, defaults the schema filter to the connected username to avoid
returning tables from SYS/SYSTEM and other visible schemas.  ClickHouse uses
system.tables because its JDBC metadata does not reliably enumerate the
current database."
  (clutch-jdbc--table-entry-names conn (clutch-db-list-table-entries conn)))

(defun clutch-jdbc--clickhouse-table-entries (conn)
  "Return ClickHouse table entries for CONN from system.tables."
  (let* ((database (or (plist-get (clutch-jdbc-conn-params conn) :database)
                       "default"))
         (sql (format "SELECT name, engine FROM system.tables WHERE database = %s ORDER BY name"
                      (clutch-db-escape-literal conn database))))
    (mapcar
     (lambda (row)
       (pcase-let ((`(,name ,engine . ,_) row))
         (list :name name
               :type (if (and (stringp engine)
                              (string-match-p "View\\'" engine))
                         "VIEW"
                       "TABLE")
               :schema database
               :source-schema database)))
     (clutch-db-result-rows (clutch-db-query conn sql)))))

(cl-defmethod clutch-db-list-table-entries ((conn clutch-jdbc-conn))
  "Return table entry plists for JDBC CONN."
  (if (clutch-jdbc--clickhouse-conn-p conn)
      (clutch-jdbc--clickhouse-table-entries conn)
    (let* ((result  (clutch-jdbc--rpc
                     "get-tables"
                     `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
                       ,@(clutch-jdbc--metadata-scope-params conn)))))
      (clutch-jdbc--collect-table-entries conn result))))

(cl-defmethod clutch-db-list-schemas ((conn clutch-jdbc-conn))
  "Return visible schema names for JDBC CONN when supported."
  (cond
   ((clutch-jdbc--clickhouse-conn-p conn)
    (sort
     (delete-dups
      (cl-loop for row in (clutch-db-result-rows
                           (clutch-db-query conn "SHOW DATABASES"))
               for database = (car row)
               when (and (stringp database) (not (string-empty-p database)))
               collect database))
     #'string-collate-lessp))
   ((and (clutch-jdbc--duckdb-conn-p conn)
         (not (member (plist-get (clutch-jdbc-conn-params conn) :url)
                      '("jdbc:duckdb:" "jdbc:duckdb::memory:"))))
    ;; DuckDB marks the default `main' schema as internal, so filter by
    ;; catalog instead; `system' and `temp' databases remain hidden.
    (cl-loop
     for row in
     (clutch-db-result-rows
      (clutch-db-query
       conn
       (concat "SELECT schema_name FROM duckdb_schemas() "
               "WHERE database_name = current_catalog() "
               "ORDER BY schema_name")))
     for schema = (car row)
     when (and (stringp schema) (not (string-empty-p schema)))
     collect schema))
   ((clutch-jdbc--oracle-conn-p conn)
    (let* ((rpc-timeout (clutch-jdbc--conn-rpc-timeout conn))
           (result (clutch-jdbc--rpc
                    "get-schemas"
                    `((conn-id . ,(clutch-jdbc-conn-conn-id conn)))
                    rpc-timeout)))
      (clutch-jdbc--visible-schemas conn (plist-get result :schemas))))))

(cl-defmethod clutch-db-current-schema ((conn clutch-jdbc-conn))
  "Return the effective schema for JDBC CONN."
  (cond
   ((clutch-jdbc--clickhouse-conn-p conn)
    (clutch-db-database conn))
   ((clutch-jdbc--duckdb-conn-p conn)
    (pcase (clutch-db-result-rows
            (clutch-db-query conn "SELECT current_catalog(), current_schema()"))
      (`((,catalog ,schema . ,_) . ,_)
       (unless (and (stringp catalog) (not (string-empty-p catalog))
                    (stringp schema) (not (string-empty-p schema)))
         (error "DuckDB returned an invalid current namespace"))
       (clutch-jdbc--remember-duckdb-namespace conn catalog schema))
      (_ (error "DuckDB did not return a current namespace"))))
   (t
    (clutch-jdbc--conn-schema conn))))

(cl-defmethod clutch-db-set-current-schema ((conn clutch-jdbc-conn) schema)
  "Switch JDBC CONN to SCHEMA."
  (cond
   ((clutch-jdbc--duckdb-conn-p conn)
    (if (member schema (clutch-db-list-schemas conn))
       (let ((catalog (clutch-jdbc--conn-catalog conn)))
         (unless catalog
           (error "DuckDB did not report a current catalog"))
         (clutch-db-query
          conn
          (format "USE %s.%s"
                  (clutch-db-escape-identifier conn catalog)
                  (clutch-db-escape-identifier conn schema)))
         (clutch-jdbc--remember-duckdb-namespace conn catalog schema))
      (user-error "DuckDB schema is no longer available: %s" schema)))
   ((clutch-jdbc--oracle-conn-p conn)
    (let ((schema (upcase schema)))
      (clutch-jdbc--rpc
       "set-current-schema"
       `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
         (schema . ,schema))
       (clutch-jdbc--conn-rpc-timeout conn))
       (setf (clutch-jdbc-conn-params conn)
             (plist-put (clutch-jdbc-conn-params conn) :schema schema))
       schema))
   (t
    (user-error "Schema switching is currently supported only for Oracle JDBC"))))

(cl-defmethod clutch-db-update-namespace-params ((conn clutch-jdbc-conn) params)
  "Store JDBC CONN's current namespace in a copy of connection PARAMS."
  (if (clutch-jdbc--duckdb-conn-p conn)
      (let ((catalog (clutch-jdbc--conn-catalog conn))
            (schema (clutch-jdbc--conn-schema conn))
            (params (copy-sequence params)))
        (unless (and catalog schema)
          (error "DuckDB switched namespace without catalog/schema state"))
        (setq params (plist-put params :catalog catalog))
        (plist-put params :schema schema))
    (cl-call-next-method)))

(cl-defmethod clutch-db-namespace-reconnect-params
    ((conn clutch-jdbc-conn) params namespace)
  "Return ClickHouse replacement PARAMS for NAMESPACE, or nil for JDBC CONN."
  (when (clutch-jdbc--clickhouse-conn-p conn)
    (plist-put (copy-sequence (or params (clutch-jdbc-conn-params conn)))
               :database namespace)))

(cl-defmethod clutch-db-browseable-object-entries ((conn clutch-jdbc-conn))
  "Return the fast browseable object snapshot for JDBC CONN.
Oracle/JDBC already surfaces synonym-aware entries through `get-tables', so do
not issue an additional empty-prefix `search-tables' scan here."
  (clutch-db-list-table-entries conn))

(cl-defmethod clutch-db-refresh-schema-async ((conn clutch-jdbc-conn) callback
                                              &optional errback idle-delay)
  "Refresh JDBC table names for CONN asynchronously.
Call CALLBACK on success or ERRBACK on failure.  When IDLE-DELAY is
positive, wait for that many seconds of Emacs idle time before starting
the metadata request."
  (cl-labels
      ((start ()
         (if (clutch-db-live-p conn)
             (clutch-jdbc--rpc-async
              "get-tables"
              `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
                ,@(clutch-jdbc--metadata-scope-params conn))
              (lambda (result)
                (when callback
                  (funcall callback
                           (clutch-jdbc--table-entry-names
                            conn (clutch-jdbc--collect-table-entries
                                  conn result)))))
              errback
              (clutch-jdbc--conn-rpc-timeout conn)
              conn)
           (when errback
             (funcall errback "Connection closed")))))
    (if (and idle-delay (> idle-delay 0))
        (run-with-idle-timer idle-delay nil #'start)
      (start))
    t))

(cl-defmethod clutch-db-column-details-async ((conn clutch-jdbc-conn) table callback
                                              &optional errback)
  "Fetch JDBC column details for TABLE on CONN asynchronously."
  (let ((rpc-timeout (clutch-jdbc--conn-rpc-timeout conn)))
    (clutch-jdbc--rpc-async
     "get-columns"
     (clutch-jdbc--table-metadata-params conn table)
     (lambda (result)
       (when callback
         (funcall callback
                  (clutch-jdbc--normalize-column-details
                   (plist-get result :columns)))))
     errback
     rpc-timeout
     conn)
    t))

(cl-defmethod clutch-db-list-columns-async ((conn clutch-jdbc-conn) table callback
                                            &optional errback)
  "Fetch JDBC column names for TABLE on CONN asynchronously."
  (let ((rpc-timeout (clutch-jdbc--conn-rpc-timeout conn)))
    (clutch-jdbc--rpc-async
     "get-columns"
     (clutch-jdbc--table-metadata-params conn table)
     (lambda (result)
       (when callback
         (funcall callback
                  (mapcar (lambda (col) (plist-get col :name))
                          (plist-get result :columns)))))
     errback
     rpc-timeout
     conn)
    t))

(defun clutch-jdbc--foreign-keys-from-result (result)
  "Return normalized foreign keys from JDBC metadata RESULT."
  (mapcar (lambda (fk)
            (cons (plist-get fk :fk-column)
                  (list :ref-table (plist-get fk :pk-table)
                        :ref-column (plist-get fk :pk-column))))
          (plist-get result :foreign-keys)))

(cl-defmethod clutch-db-foreign-keys-async ((conn clutch-jdbc-conn) table callback
                                            &optional errback)
  "Fetch foreign-key info for TABLE on JDBC CONN asynchronously."
  (let ((rpc-timeout (clutch-jdbc--conn-rpc-timeout conn)))
    (clutch-jdbc--rpc-async
     "get-foreign-keys"
     (clutch-jdbc--table-metadata-params conn table)
     (lambda (result)
       (when callback
         (funcall callback (clutch-jdbc--foreign-keys-from-result result))))
     errback
     rpc-timeout
     conn)
    t))

(cl-defmethod clutch-db-list-columns ((conn clutch-jdbc-conn) table)
  "Return column names for TABLE on JDBC CONN using DatabaseMetaData."
  (let* ((result  (clutch-jdbc--rpc
                   "get-columns"
                   (clutch-jdbc--table-metadata-params conn table))))
    (mapcar (lambda (col) (plist-get col :name))
            (plist-get result :columns))))

(cl-defmethod clutch-db-complete-tables ((conn clutch-jdbc-conn) prefix)
  "Return table name candidates matching PREFIX for JDBC CONN."
  (when (clutch-jdbc--oracle-conn-p conn)
    (let ((result (clutch-jdbc--rpc
                   "search-tables"
                   `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
                     (prefix  . ,prefix)
                     ,@(clutch-jdbc--metadata-scope-params conn)))))
      (mapcar (lambda (tbl) (plist-get tbl :name))
              (plist-get result :tables)))))

(cl-defmethod clutch-db-search-table-entries ((conn clutch-jdbc-conn) prefix)
  "Return table entry plists matching PREFIX for JDBC CONN."
  (if (clutch-jdbc--clickhouse-conn-p conn)
      (seq-filter
       (lambda (entry)
         (string-prefix-p
          (downcase prefix)
          (downcase (plist-get entry :name))))
       (clutch-jdbc--clickhouse-table-entries conn))
    (let* ((result (clutch-jdbc--rpc
                    "search-tables"
                    `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
                      (prefix  . ,prefix)
                      ,@(clutch-jdbc--metadata-scope-params conn)))))
      (mapcar (lambda (entry)
                (clutch-jdbc--normalize-table-entry conn entry))
              (plist-get result :tables)))))

(cl-defmethod clutch-db-table-comment-async ((conn clutch-jdbc-conn) table callback
                                             &optional errback)
  "Fetch TABLE comment for CONN asynchronously when metadata has it."
  (when (clutch-jdbc--metadata-table-comments-supported-p conn)
    (let ((rpc-timeout (clutch-jdbc--conn-rpc-timeout conn))
          (schema (clutch-jdbc--conn-schema conn)))
      (clutch-jdbc--rpc-async
       "search-tables"
       `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
         (prefix  . ,table)
         ,@(clutch-jdbc--metadata-scope-params conn))
       (lambda (result)
         (when callback
           (funcall callback
                    (clutch-jdbc--table-comment-from-entries
                     table
                     (mapcar (lambda (entry)
                               (clutch-jdbc--normalize-table-entry conn entry))
                             (plist-get result :tables))
                     schema))))
       errback
       rpc-timeout
       conn)
      t)))

(cl-defmethod clutch-db-complete-columns ((conn clutch-jdbc-conn) table prefix)
  "Return column name candidates for TABLE matching PREFIX on JDBC CONN."
  (let ((driver (clutch-jdbc--conn-driver conn)))
    (cond
     ((eq driver 'oracle)
      (let* ((result (clutch-jdbc--rpc
                      "search-columns"
                      (append (clutch-jdbc--table-metadata-params conn table)
                              `((prefix . ,prefix))))))
        (mapcar (lambda (col) (plist-get col :name))
                (plist-get result :columns)))))))

(cl-defmethod clutch-db-list-objects ((conn clutch-jdbc-conn) category)
  "Return object entry plists for CATEGORY on JDBC CONN."
  (when-let* ((spec (clutch-jdbc--object-category-spec category)))
    (let* ((op (plist-get spec :op))
           (key (plist-get spec :key))
           (result (clutch-jdbc--rpc
                    op
                    `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
                      ,@(clutch-jdbc--metadata-scope-params conn)))))
      (mapcar #'clutch-jdbc--normalize-object-entry
              (plist-get result key)))))

(cl-defmethod clutch-db-list-objects-async ((conn clutch-jdbc-conn) category callback
                                            &optional errback)
  "Fetch object entry plists for CATEGORY on JDBC CONN asynchronously."
  (when-let* ((spec (clutch-jdbc--object-category-spec category)))
    (let ((op (plist-get spec :op))
          (key (plist-get spec :key)))
      (clutch-jdbc--rpc-async
       op
       `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
         ,@(clutch-jdbc--metadata-scope-params conn))
       (lambda (result)
         (when callback
           (funcall callback
                    (mapcar #'clutch-jdbc--normalize-object-entry
                            (plist-get result key)))))
       errback
       (clutch-jdbc--conn-rpc-timeout conn)
       conn)
      t)))

(cl-defmethod clutch-db-object-details ((conn clutch-jdbc-conn) entry)
  "Return detail plists for JDBC object ENTRY on CONN."
  (let* ((type (upcase (or (plist-get entry :type) ""))))
    (pcase type
      ("INDEX"
       (let ((result
              (clutch-jdbc--rpc
               "get-index-columns"
               `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
                 (index   . ,(plist-get entry :name))
                 ,@(when (plist-get entry :target-table)
                     `((table . ,(plist-get entry :target-table))))
                 ,@(clutch-jdbc--metadata-scope-params conn)))))
         (plist-get result :columns)))
      ((or "PROCEDURE" "FUNCTION")
       (let ((result
              (clutch-jdbc--rpc
               (if (string= type "PROCEDURE")
                   "get-procedure-params"
                 "get-function-params")
               `((conn-id  . ,(clutch-jdbc-conn-conn-id conn))
                 (name     . ,(plist-get entry :name))
                 ,@(when (plist-get entry :identity)
                     `((identity . ,(plist-get entry :identity))))
                 ,@(clutch-jdbc--metadata-scope-params conn)))))
         (plist-get result :params)))
      (_ nil))))

(cl-defmethod clutch-db-object-source ((conn clutch-jdbc-conn) entry)
  "Return source text for JDBC object ENTRY on CONN."
  (let* ((result (clutch-jdbc--rpc
                  "get-object-source"
                  `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
                    (name    . ,(plist-get entry :name))
                    (type    . ,(plist-get entry :type))
                    ,@(when (plist-get entry :identity)
                        `((identity . ,(plist-get entry :identity))))
                    ,@(clutch-jdbc--metadata-scope-params conn)))))
    (plist-get result :source)))

(cl-defmethod clutch-db-object-definition ((conn clutch-jdbc-conn) entry)
  "Return definition or source text for JDBC object ENTRY on CONN."
  (let ((type (upcase (or (plist-get entry :type) "")))
        (name (plist-get entry :name)))
    (pcase type
      ((or "PROCEDURE" "FUNCTION" "TRIGGER")
       (clutch-db-object-source conn entry))
      ((or "TABLE" "COLLECTION")
       (let* ((driver (clutch-jdbc--conn-driver conn))
              (result (clutch-jdbc--rpc
                       "get-columns"
                       `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
                         (table   . ,name)
                         ,@(clutch-jdbc--metadata-scope-params conn))))
              (cols (plist-get result :columns))
              (display-ident
               (if (eq driver 'oracle)
                   #'clutch-jdbc--oracle-display-identifier
                 (lambda (identifier)
                   (clutch-db-escape-identifier conn identifier)))))
         (format "-- DDL reconstructed from DatabaseMetaData\nCREATE TABLE %s (\n%s\n);"
                 (funcall display-ident name)
                 (mapconcat
                  (lambda (col)
                    (format "    %s %s%s"
                            (funcall display-ident (plist-get col :name))
                            (plist-get col :type)
                            (if (clutch-jdbc--json-bool (plist-get col :nullable))
                                ""
                              " NOT NULL")))
                  cols
                  ",\n"))))
      (_
       (let* ((result (clutch-jdbc--rpc
                       "get-object-ddl"
                       `((conn-id . ,(clutch-jdbc-conn-conn-id conn))
                         (name    . ,name)
                         (type    . ,(plist-get entry :type))
                         ,@(when (plist-get entry :identity)
                             `((identity . ,(plist-get entry :identity))))
                         ,@(clutch-jdbc--metadata-scope-params conn)))))
         (plist-get result :ddl))))))

(cl-defmethod clutch-db-table-comment ((conn clutch-jdbc-conn) table
                                       &optional schema)
  "Return TABLE comment for CONN when metadata has one."
  (let ((metadata-conn (clutch-jdbc--metadata-conn-for-scope conn schema)))
    (when (clutch-jdbc--metadata-table-comments-supported-p metadata-conn)
      (clutch-jdbc--table-comment-from-entries
       table
       (clutch-db-search-table-entries metadata-conn table)
       (clutch-jdbc--conn-schema metadata-conn)))))

(cl-defmethod clutch-db-primary-key-columns ((conn clutch-jdbc-conn) table)
  "Return primary key columns for TABLE on JDBC CONN."
  (let* ((result (clutch-jdbc--rpc
                  "get-primary-keys"
                  (clutch-jdbc--table-metadata-params conn table))))
    (plist-get result :primary-keys)))

(defun clutch-jdbc--index-column-name (column)
  "Return a column name string from JDBC index COLUMN metadata."
  (cond
   ((stringp column) column)
   ((and (listp column) (plist-get column :name))
    (plist-get column :name))
   ((and (listp column) (plist-get column :column))
    (plist-get column :column))
   (t (format "%s" column))))

(defun clutch-jdbc--unique-not-null-identities (conn table)
  "Return unique-not-null row identity candidates for TABLE on CONN."
  (let* ((details (clutch-db-column-details conn table))
         (not-null (make-hash-table :test 'equal))
         (indexes (clutch-db-list-objects conn 'indexes)))
    (dolist (detail details)
      (puthash (plist-get detail :name)
               (not (plist-get detail :nullable))
               not-null))
    (cl-loop for index in indexes
             when (and (plist-get index :unique)
                       (string= (or (plist-get index :target-table)
                                    table)
                                table))
             for cols = (mapcar
                         #'clutch-jdbc--index-column-name
                         (clutch-db-object-details conn index))
             when (and cols
                       (cl-every (lambda (col)
                                   (gethash col not-null))
                                 cols))
             collect (list :kind 'unique-key
                           :name (plist-get index :name)
                           :columns cols))))

(defun clutch-jdbc--rowid-identity (conn _table)
  "Return a JDBC row locator candidate for TABLE on CONN, or nil."
  (when (clutch-jdbc--oracle-conn-p conn)
    (list :kind 'row-locator
          :name "ROWID"
          :select-expressions '("ROWID")
          :where-sql "ROWID = ?")))

(cl-defmethod clutch-db-row-identity-candidates ((conn clutch-jdbc-conn) table
                                                 &optional schema catalog)
  "Return row identity candidates for TABLE on JDBC CONN."
  (let ((metadata-conn
         (clutch-jdbc--metadata-conn-for-scope conn schema catalog)))
    (when (or (not (clutch-jdbc--oracle-conn-p metadata-conn))
              (let ((effective-schema
                     (clutch-jdbc--conn-schema metadata-conn)))
                (cl-loop for entry in (clutch-db-search-table-entries
                                       metadata-conn table)
                         thereis
                         (and (string= (downcase table)
                                       (downcase
                                        (or (plist-get entry :name) "")))
                              (clutch-jdbc--entry-type= entry "TABLE")
                              (string= (downcase (or effective-schema ""))
                                       (downcase
                                        (or (plist-get entry :schema) "")))
                              (string= (downcase (or effective-schema ""))
                                       (downcase
                                        (or (plist-get entry :source-schema)
                                            "")))))))
      (or (cl-call-next-method metadata-conn table schema catalog)
          (clutch-jdbc--unique-not-null-identities metadata-conn table)
          (when-let* ((rowid (clutch-jdbc--rowid-identity
                              metadata-conn table)))
            (list rowid))))))

(cl-defmethod clutch-db-foreign-keys ((conn clutch-jdbc-conn) table)
  "Return foreign key info for TABLE on JDBC CONN."
  (let* ((result (clutch-jdbc--rpc
                  "get-foreign-keys"
                  (clutch-jdbc--table-metadata-params conn table))))
    (clutch-jdbc--foreign-keys-from-result result)))

(cl-defmethod clutch-db-referencing-objects ((conn clutch-jdbc-conn) table)
  "Return objects that reference TABLE on JDBC CONN."
  (let* ((result (clutch-jdbc--rpc
                  "get-referencing-objects"
                  (clutch-jdbc--table-metadata-params conn table))))
    (mapcar (lambda (entry)
              (list :name (plist-get entry :name)
                    :type "TABLE"
                    :schema (plist-get entry :schema)
                    :source-schema (or (plist-get entry :source-schema)
                                       (plist-get entry :schema))))
            (plist-get result :objects))))

(cl-defmethod clutch-db-column-details ((conn clutch-jdbc-conn) table)
  "Return detailed column info for TABLE on JDBC CONN."
  (let* ((pk-cols (clutch-db-primary-key-columns conn table))
         (fks     (clutch-db-foreign-keys conn table))
         (result  (clutch-jdbc--rpc
                   "get-columns"
                   (clutch-jdbc--table-metadata-params conn table)))
         (cols    (clutch-jdbc--normalize-column-details
                   (plist-get result :columns))))
    (mapcar (lambda (col)
              (let ((name (plist-get col :name)))
                (list :name        name
                      :type        (plist-get col :type)
                      :backend-type (plist-get col :backend-type)
                      :nullable    (clutch-jdbc--json-bool (plist-get col :nullable))
                      :primary-key (and (member name pk-cols) t)
                      :foreign-key (cdr (assoc name fks))
                      :comment     nil
                      :default     (plist-get col :default))))
            cols)))

;;;; Re-entrancy guard

(cl-defmethod clutch-db-busy-p ((conn clutch-jdbc-conn))
  "Return non-nil if JDBC CONN is executing a query."
  (clutch-jdbc-conn-busy conn))

;;;; Metadata methods

(cl-defmethod clutch-db-user ((conn clutch-jdbc-conn))
  "Return the user for JDBC CONN."
  (plist-get (clutch-jdbc-conn-params conn) :user))

(cl-defmethod clutch-db-host ((conn clutch-jdbc-conn))
  "Return the host for JDBC CONN."
  (or (plist-get (clutch-jdbc-conn-params conn) :host)
      (plist-get (clutch-jdbc--url-metadata
                  (plist-get (clutch-jdbc-conn-params conn) :url))
                 :host)))

(cl-defmethod clutch-db-port ((conn clutch-jdbc-conn))
  "Return the port for JDBC CONN."
  (or (plist-get (clutch-jdbc-conn-params conn) :port)
      (plist-get (clutch-jdbc--url-metadata
                  (plist-get (clutch-jdbc-conn-params conn) :url))
                 :port)))

(cl-defmethod clutch-db-database ((conn clutch-jdbc-conn))
  "Return the database for JDBC CONN."
  (or (plist-get (clutch-jdbc-conn-params conn) :database)
      (plist-get (clutch-jdbc--url-metadata
                  (plist-get (clutch-jdbc-conn-params conn) :url))
                 :database)))

(cl-defmethod clutch-db-display-name ((conn clutch-jdbc-conn))
  "Return a display name for CONN based on the JDBC driver type."
  (or (plist-get (clutch-jdbc-conn-params conn) :display-name)
      (and (eq (clutch-jdbc--conn-driver conn) 'mongodb)
           "MongoDB")
      (clutch-backend-display-name
       (clutch-jdbc--conn-driver conn))
      "JDBC"))

;;;; Agent installation helpers

(defun clutch-jdbc--download-agent-jar ()
  "Download the agent jar from GitHub Releases and verify its checksum."
  (let ((jar (clutch-jdbc--agent-jar))
        (url (format
              "https://github.com/LuciusChen/clutch-jdbc-agent/releases/download/v%s/clutch-jdbc-agent-%s.jar"
              clutch-jdbc-agent-version
              clutch-jdbc-agent-version)))
    (make-directory clutch-jdbc-agent-dir t)
    (make-directory (clutch-jdbc--drivers-dir) t)
    (message "Downloading clutch-jdbc-agent %s..." clutch-jdbc-agent-version)
    (url-copy-file url jar t)
    (unless (clutch-jdbc--agent-jar-valid-p jar)
      (error "Downloaded clutch-jdbc-agent checksum mismatch: %s" jar))
    (clutch-jdbc--cleanup-stale-agent-jars)
    (message "Downloaded to %s" jar)))

;;;###autoload
(defun clutch-jdbc-ensure-agent ()
  "Download clutch-jdbc-agent.jar if not present.
Fetches from GitHub Releases."
  (interactive)
  (let ((jar (clutch-jdbc--agent-jar)))
    (if (and (file-exists-p jar)
             (clutch-jdbc--agent-jar-valid-p jar))
        (progn
          (clutch-jdbc--cleanup-stale-agent-jars)
          (message "clutch-jdbc-agent already at %s" jar))
      (clutch-jdbc--download-agent-jar))))

;;;###autoload
(defun clutch-jdbc-install-driver (driver)
  "Download the JDBC driver for DRIVER symbol."
  (interactive
   (list (intern (completing-read "Driver: "
                                  (clutch-jdbc--installable-drivers)
                                  nil t))))
  (let* ((spec       (alist-get driver clutch-jdbc--driver-sources))
         (filename   (plist-get spec :filename))
         (dest       (expand-file-name filename (clutch-jdbc--drivers-dir)))
         (companions (plist-get spec :companions)))
    (make-directory (clutch-jdbc--drivers-dir) t)
    (cond
     ((file-exists-p dest)
      (message "Driver already installed: %s" dest))
     ((plist-get spec :maven)
      (clutch-jdbc--download-maven-driver (plist-get spec :maven) dest))
     ((plist-get spec :url)
      (clutch-jdbc--download-url-driver (plist-get spec :url) dest))
     (t
      (message "Manual download required for %s.\nURL: %s\nPlace as: %s"
               driver (plist-get spec :manual) dest)))
    (when (clutch-jdbc--oracle-driver-symbol-p driver)
      (clutch-jdbc--disable-conflicting-oracle-jars filename))
    (dolist (companion companions)
      (unless (file-exists-p
               (expand-file-name
                (plist-get (alist-get companion clutch-jdbc--driver-sources) :filename)
                (clutch-jdbc--drivers-dir)))
        (clutch-jdbc-install-driver companion)))
    (when (clutch-jdbc--agent-live-p)
      (clutch-jdbc--stop-agent)
      (message "Installed JDBC driver(s); shared clutch-jdbc-agent restarted on next use"))))

(defun clutch-jdbc--installable-drivers ()
  "Return public driver symbols accepted by `clutch-jdbc-install-driver'."
  (cl-loop for (driver . spec) in clutch-jdbc--driver-sources
           unless (plist-get spec :internal)
           collect driver))

(defun clutch-jdbc--download-maven-driver (coords dest)
  "Download a Maven artifact at COORDS to DEST.
COORDS is \"group:artifact:version\" or \"group:artifact:version:classifier\"."
  (pcase-let ((`(,group ,artifact ,version ,classifier . ,_)
               (split-string coords ":")))
    (let* ((group-path (replace-regexp-in-string "\\." "/" group))
           (jar-name   (if classifier
                           (format "%s-%s-%s.jar" artifact version classifier)
                         (format "%s-%s.jar" artifact version)))
           (url        (format "https://repo1.maven.org/maven2/%s/%s/%s/%s"
                               group-path artifact version jar-name)))
      (message "Downloading %s from Maven Central..." coords)
      (url-copy-file url dest)
      (message "Downloaded driver to %s" dest))))

(defun clutch-jdbc--download-url-driver (url dest)
  "Download a JDBC driver from URL to DEST."
  (message "Downloading JDBC driver from %s..." url)
  (url-copy-file url dest)
  (message "Downloaded driver to %s" dest))

(defun clutch-jdbc--oracle-driver-symbol-p (driver)
  "Return non-nil for DRIVER symbols that use an Oracle JDBC jar."
  (memq driver '(oracle oracle-8 oracle-11)))

(defun clutch-jdbc--disable-conflicting-oracle-jars (selected-filename)
  "Remove Oracle JDBC jars that conflict with SELECTED-FILENAME."
  (dolist (filename clutch-jdbc--oracle-driver-filenames)
    (unless (string-equal filename selected-filename)
      (let ((path (expand-file-name filename (clutch-jdbc--drivers-dir))))
        (when (file-exists-p path)
          (delete-file path))))))

(provide 'clutch-db-jdbc)
;;; clutch-db-jdbc.el ends here
