sqlk.tcl

# sqlk.tcl --
#
# Persistent SQL-backed hierarchical keyed lists for Tcl 8.6.
#
# sqlk provides a common keyed-list API with SQLite and MySQL/InnoDB
# storage backends. SQLite is the default backend. Backend drivers are
# loaded lazily when a database instance is initialized.
#
# See README.md for an overview, MANUAL.md for the API reference, and
# CHANGELOG.md for release history.

package provide sqlk 0.8

##########
##########
##
##
##
namespace eval sqlk {
  variable seq
  variable enc 1

  if {[catch {package require aes}]} {
    set enc 0
  }


  ##########
  #
  #
  #
  proc kinit {filepath args} {
    variable seq
    variable enc

    if {fmod([llength $args],2)} {
      return -code error "Invalid number of options, should be 2 elements list, got \"[llength $args]\""
    }
    set procname ""
    set enckey ""
    set backendname sqlite
    if {[llength $args] > "0"} {
      array set op $args
      if {[info exists op(-procname)]} {
        set procname $op(-procname)
        unset op(-procname)
      }
      if {[info exists op(-enckey)]} {
        set enckey $op(-enckey)
        unset op(-enckey)
      }
      if {[info exists op(-backend)]} {
        set backendname $op(-backend)
        unset op(-backend)
      }
      if {[llength [array names op]] > 0} {
        return -code error "Invalid option \"[lindex [array names op] 0]\", should be -procname, -enckey or -backend"
      }
    }

    switch -- $backendname {
      sqlite {
        set backend ::sqlk::backend::sqlite
      }
      mysql {
        set backend ::sqlk::backend::mysql
      }
      default {
        return -code error "Backend \"$backendname\" is not available"
      }
    }

    ${backend}::package_require

    if {!$enc && [string length $enckey] > 0} {
      return -code error "Package aes needed for encryption"
    }

    ${backend}::validate_target $filepath $enckey
    set public_target [${backend}::target_public $filepath]

    if {![info exists seq]} {set seq 0}
    set dbid [format %08s [incr seq]]

    set db [${backend}::open _db$dbid $filepath]

    ${backend}::schema_init $db


    ##########
    ##########
    ##
    ##
    ##
    namespace eval $dbid {
      variable db ""
      variable backend ""
      variable filepath  ""
      variable kprocname ""
      variable kenckey   ""
      variable encrypt   0

      proc varlist   {}                       {::sqlk::_varlist}
      proc varadd    {name {procname ""}}     {::sqlk::_varadd}
      proc vardel    {name}                   {::sqlk::_vardel}
      proc keys      {name {key ""}}          {::sqlk::_keys}
      proc tree      {name {key ""}}          {::sqlk::_tree}
      proc kget      {name key}               {::sqlk::_kget}
      proc kset      {name args}              {::sqlk::_kset}
      proc kdel      {name key}               {::sqlk::_kdel}
      proc kmove     {name key {afterkey ""}} {::sqlk::_kmove}
      proc kexist    {name key}               {::sqlk::_kexist}
      proc attrset   {name key args}          {::sqlk::_attrset}
      proc attrget   {name key {attr ""}}     {::sqlk::_attrget}
      proc attrdel   {name key attr}          {::sqlk::_attrdel}
      proc serialize {name args}              {::sqlk::_serialize}
      proc parse     {name args}              {::sqlk::_parse}
      proc closefile {}                       {::sqlk::_closefile}
      proc backup    {filename}               {::sqlk::_backup}
      proc restore   {filename}               {::sqlk::_restore}

      proc varcmd    {name {procname ""}}     {::sqlk::_varcmd}

      # Internal usage
      proc doenc     {data}                   {::sqlk::_doenc}
      proc dodec     {data}                   {::sqlk::_dodec}
      proc doencbin  {data}                   {::sqlk::_doencbin}
      proc dodecbin  {data}                   {::sqlk::_dodecbin}
      proc nameid    {name}                   {::sqlk::_nameid}
      proc keyid     {name_id key}            {::sqlk::_keyid}
      proc validname {name}                   {::sqlk::validname $name}
      proc validkey  {name}                   {::sqlk::validkey  $name}
      proc keysort   {name_id parent}         {::sqlk::_keysort}
      namespace export varlist varadd vardel keys kget kset kdel kexist kmove attrset attrget attrdel serialize closefile varcmd parse tree backup restore
      namespace ensemble create
    }
    set ${dbid}::db $db
    set ${dbid}::backend $backend
    set ${dbid}::filepath $public_target
    set ${dbid}::kenckey $enckey

    if {$enckey != "" && $enc} {
      set ${dbid}::encrypt 1
    }

    if {$procname != ""} {
      set ${dbid}::kprocname $procname
      uplevel #0 "rename [namespace current]::$dbid $procname"
    } else {
      return [namespace current]::$dbid
    }
  }

  ##########
  #
  #
  #
  proc klist {} {
    set y [list]
    foreach x [namespace children [namespace current]] {
      if {![info exists ${x}::db]} {
        continue
      }
      set a [list]
      foreach z [namespace children $x] {
        lappend a [list [namespace tail $z] $z [set ${z}::kprocname]]
      }
      lappend y [list $x [set ${x}::filepath] [set ${x}::kprocname] $a]
    }
    return $y
  }

  ##########
  #
  #
  #
  proc kclose {objname} {
    foreach x [namespace children [namespace current]] {
      if {![info exists ${x}::db]} {
        continue
      }
      if {$x == $objname} {
        $objname closefile
        return
      }
    }
    if {[namespace exists ::sqlk::$objname] && [info exists ::sqlk::${objname}::db]} {
      sqlk::${objname} closefile
    } else {
      return -code error "keyed list object \"$objname\" does not exist or is not open"
    }
  }

  ##########
  #
  #
  #
  proc kcloseall {} {
    foreach x [namespace children] {
      if {[info exists ${x}::db]} {
        ${x}::closefile
      }
    }
  }


  ##########
  #
  #
  #
  proc _closefile {} {
    uplevel 1 {
      variable db
      variable backend
      ${backend}::close $db
      namespace delete [namespace current]
    }
  }

  ##########
  #
  #
  #
  proc _backup {} {
    uplevel 1 {
      variable db
      variable backend
      ${backend}::backup $db $filename
    }
  }

  ##########
  #
  #
  #
  proc _restore {} {
    uplevel 1 {
      variable db
      variable backend
      ${backend}::restore $db $filename
    }
  }

  ##########
  #
  #
  #
  proc _varcmd {} {
    uplevel 1 {
      if {[namespace exists [namespace current]::$name]} {
        if {$procname != ""} {
          if {[set [namespace current]::${name}::kprocname] == ""} {
            set [namespace current]::${name}::kprocname $procname
            uplevel #0 "rename [namespace current]::$name $procname"
            return ""
          } else {
            return [set [namespace current]::${name}::kprocname]
          }
        } else {
          return [namespace current]::$name
        }
      }

      ##########
      ##########
      ##
      ##
      ##
      namespace eval $name {
        variable name [namespace tail [namespace current]]
        variable kprocname ""
        proc keys      {{key ""}}          {eval "[namespace parent]::keys      [myname] $key"}
        proc tree      {{key ""}}          {eval "[namespace parent]::tree      [myname] $key"}
        proc kget      {key}               {eval "[namespace parent]::kget      [myname] $key"}
        proc kset      {args}              {eval "[namespace parent]::kset      [myname] $args"}
        proc kdel      {key}               {eval "[namespace parent]::kdel      [myname] $key"}
        proc kexist    {key}               {eval "[namespace parent]::kexist    [myname] $key"}
        proc kmove     {key {afterkey ""}} {eval "[namespace parent]::kmove     [myname] $key $afterkey"}
        proc attrset   {key args}          {eval "[namespace parent]::attrset   [myname] $key $args"}
        proc attrget   {key {attr ""}}     {eval "[namespace parent]::attrget   [myname] $key $attr"}
        proc attrdel   {key attr}          {eval "[namespace parent]::attrdel   [myname] $key $attr"}
        proc serialize {args}              {eval "[namespace parent]::serialize [myname] $args"}
        proc parse     {args}              {eval "[namespace parent]::parse     [myname] $args"}
        proc myname    {}                  {variable name ; return $name}

        namespace export keys kget kset kdel kexist kmove attrset attrget attrdel serialize parse tree
        namespace ensemble create
      }
      if {$procname != ""} {
        set [namespace current]::${name}::kprocname $procname
        uplevel #0 "rename [namespace current]::$name $procname"
      } else {
        return [namespace current]::$name
      }
    }
  }


  ##########
  #
  #
  #
  proc _varlist {} {
    uplevel 1 {
      variable db
      variable backend
      variable encrypt

      if {$encrypt} {
        set l [list]
        foreach stored_name [${backend}::variable_list $db 1] {
          if {[string range [set e [dodec $stored_name]] end-4 end] == "*****"} {
            lappend l [string map {* "" + ""} $e]
          }
        }
        return $l
      } else {
        ${backend}::variable_list $db 0
      }
    }
  }

  ##########
  #
  #
  #
  proc _varadd {} {
    uplevel 1 {
      variable db
      variable backend
      variable encrypt

      if {[validname $name]} {
        if {[string length $name] > 128} {
          return -code error "Invalid size, max 128, got [string length $name]"
        } else {
          return -code error "Invalid characters"
        }
      } elseif {[nameid $name] != ""} {
        return -code error "var name \"$name\" already exists"
      }

      if {$encrypt} {
        set ename [doenc "$name[string repeat "+" [expr 45 - [string length $name]]]*****"]
      } else {
        set ename $name
      }
      ${backend}::transaction_begin_write $db
      ${backend}::variable_insert $db $ename $encrypt
      ${backend}::transaction_commit $db

      varcmd $name
      if {$procname != ""} {
        varcmd $name $procname
      } else {
        varcmd $name
      }
    }
  }

  ##########
  #
  #
  #
  proc _vardel {} {
    uplevel 1 {
      variable db
      variable backend
      variable encrypt

      if {[set name_id [nameid $name]] == ""} {
        return -code error "key name \"$name\" does not exists"
      }

      if {[namespace exists [namespace current]::$name]} {
        namespace delete [namespace current]::$name
      }
      ${backend}::variable_delete $db $name_id
    }
  }

  ##########
  #
  #
  #
  proc _keys {} {
    uplevel 1 {
      variable db
      variable backend
      variable encrypt

      if {[set name_id [nameid $name]] == ""} {
        return -code error "var name \"$name\" does not exists"
      }

      if {$key == ""} {
        return [${backend}::node_children_names $db $name_id 0]
      }

      if {[set p [keyid $name_id $key]] == ""} {
        return -code error "key name \"$key\" does not exists"
      }

      return [${backend}::node_children_names $db $name_id $p]
    }
  }


  ##########
  #
  #
  #
  proc _kget {} {
    uplevel 1 {
      variable db
      variable backend
      variable encrypt

      if {[set name_id [nameid $name]] == ""} {
        return -code error "var name \"$name\" does not exists"
      }

      if {[set p [keyid $name_id $key]] == ""} {
        return -code error "key name \"$key\" does not exists"
      }

      if {$encrypt} {
        return [dodecbin [${backend}::node_value $db $p]]
      } else {
        return [${backend}::node_value $db $p]
      }
    }
  }

  ##########
  #
  #
  #
  proc _kset {} {
    uplevel 1 {
      variable db
      variable backend
      variable encrypt

      if {[set name_id [nameid $name]] == ""} {
        return -code error "var name \"$name\" does not exists"
      } elseif {fmod([llength $args],2)} {
        return -code error "Keyed list must be a valid, 2 elements list, got \"[llength $args]\""
      }

      ${backend}::transaction_begin_write $db
      foreach {k v} $args {
        if {[validkey $k]} {
          ${backend}::transaction_rollback $db
          return -code error "Invalid characters or invalid size in \"$k\""
        }
        set p 0
        foreach s [split $k .] {
          if {$encrypt} {
            set ev [doencbin $v]
          } else {
            set ev $v
          }
          set x [${backend}::node_id_by_child $db $name_id $p $s]
          if {$x == ""} {
            set p [${backend}::node_insert_append $db $name_id $p $s]
          } else {
            set p $x
          }
        }
        ${backend}::node_value_set $db $p $ev
      }
      ${backend}::transaction_commit $db

    }
  }

  ##########
  #
  #
  #
  proc _kdel {} {
    uplevel 1 {
      variable db
      variable backend
      variable encrypt

      if {[set name_id [nameid $name]] == ""} {
        return -code error "var name \"$name\" does not exists"
      }

      if {[set p [keyid $name_id $key]] == ""} {
        return -code error "key name \"$key\" does not exists"
      }

      ${backend}::transaction_begin_write $db

      set parent [${backend}::node_parent $db $p]

      set next_after_id [${backend}::node_follower $db $parent $p]
      if {$next_after_id != ""} {
        set curr_after_id [${backend}::node_after_id $db $p]
        if {$curr_after_id == 0} {
          ${backend}::node_after_id_and_order_set $db $next_after_id $curr_after_id 0
        } else {
          ${backend}::node_after_id_set $db $next_after_id $curr_after_id
        }
      } else {
        set last_id [${backend}::node_after_id $db $p]
        if {$parent == 0} {
          ${backend}::variable_last_id_set $db $name_id $last_id
        } else {
          ${backend}::node_last_id_set $db $parent $last_id
        }
      }

      set r $p
      while {$r != ""} {
        set r [${backend}::node_children_ids $db $r]
        set p [lsort -unique -integer "$p $r"]
      }

      ${backend}::node_delete_ids $db $p

      ${backend}::transaction_commit $db
    }
  }


  ##########
  #
  #
  #
  proc _kexist {} {
    uplevel 1 {
      variable db
      variable encrypt

      if {[set name_id [nameid $name]] == ""} {
        return -code error "var name \"$name\" does not exists"
      }

      if {[set p [keyid $name_id $key]] == ""} {
        return 0
      }
      return 1
    }
  }

  ##########
  #
  #
  #
  proc _kmove {} {
    uplevel 1 {
      variable db
      variable backend
      variable encrypt

      if {[set name_id [nameid $name]] == ""} {
        return -code error "var name \"$name\" does not exists"
      }

      if {[set p [keyid $name_id $key]] == ""} {
        return -code error "key name \"$key\" does not exists"
      }

      set path [split $key .]
      if {[llength $path] == 1} {
        set path ""
      } else {
        set path [join [lrange $path 0 end-1] .]
      }
      set g [${backend}::node_parent $db $p]


      if {$afterkey == ""} {
        set n 0
      } else {
        if {[llength [split $afterkey .]] == 1} {
          set afterkey [join "$path $afterkey" .]
        }
        if {[set n [keyid $name_id $afterkey]] == ""} {
          return -code error "key name \"$afterkey\" does not exists"
        }
        if {[${backend}::node_is_child $db $name_id $g $n] == 0} {
          return -code error "key \"$afterkey\" and $key does not belong to the same parent"
        }
      }


      if {[${backend}::node_child_count $db $g] == 1} {
        return ""
      }

      if {$n == [${backend}::node_after_id $db $p]} {
        # Same positions, dont change
        return ""
      }

      set setorder 0
      set last_id ""

      ${backend}::transaction_begin_write $db

      # First move, to set my follower to follow who i used to follow
      set next_after_id [${backend}::node_follower_in_variable $db $name_id $g $p]
      if {$next_after_id != ""} {
        set curr_after_id [${backend}::node_after_id $db $p]
        if {$curr_after_id == 0} {
          ${backend}::node_after_id_and_order_set $db $next_after_id $curr_after_id 0
        } else {
          ${backend}::node_after_id_set $db $next_after_id $curr_after_id
        }
      } else {
        set last_id [${backend}::node_after_id $db $p]
      }

      # Second move, to set the follower of the one i m going to follow to follow me
      #   AND... set myself to follow the new one
      if {$n == 0} {
        set curr_order 0
        lassign [${backend}::node_first_with_order $db $name_id $g] next_id next_order
        if {$next_order == 0} {
          lassign [${backend}::node_follower_with_order $db $g $next_id] next_next_id next_order
        } else {
          set next_next_id [${backend}::node_follower $db $g $next_id]
        }
        if {$next_next_id == $p} {
          ${backend}::node_after_id_and_order_set $db $next_id $p $next_id
          set last_id $next_id
        } else {
          set next_order [::sqlk::inmiddle $next_order $curr_order]
          if {$next_order == 0} {
            set setorder 1
          }
          ${backend}::node_after_id_and_order_set $db $next_id $p $next_order
        }
      } else {
        lassign [${backend}::node_follower_with_order_in_variable $db $name_id $g $n] prev_after_id curr_order
        set prev_order [${backend}::node_key_order $db $n]
        if {$prev_after_id != ""} {
          set curr_order [::sqlk::inmiddle $prev_order $curr_order]
          if {$curr_order == 0} {
            set setorder 1
          }
          ${backend}::node_after_id_set $db $prev_after_id $p
        } else {
          set last_id $p
          set curr_order [${backend}::node_max_order_append $db $g]
        }
      }
      ${backend}::node_after_id_and_order_set $db $p $n $curr_order

      if {$last_id != ""} {
        if {$g == 0} {
          ${backend}::variable_last_id_set $db $name_id $last_id
        } else {
          ${backend}::node_last_id_set $db $g $last_id
        }
      }

      ${backend}::transaction_commit $db

      if {$setorder} {
        keysort $name_id $g
      }
    }
  }

  ##########
  #
  #
  #
  proc _keysort {} {
    uplevel 1 {
      variable db
      variable backend

      ${backend}::transaction_begin_write $db

      # online sort
      set n 0
      set id 0
      set last_id 0
      set step [${backend}::node_sort_step $db $name_id $parent]
      while {[set id [${backend}::node_follower_in_variable $db $name_id $parent $id]] != ""} {
        incr n $step
        ${backend}::node_key_order_set $db $id $n
        set last_id $id
      }

      if {$parent == 0} {
        ${backend}::variable_last_id_set $db $name_id $last_id
      } else {
        ${backend}::node_last_id_set $db $parent $last_id
      }

      ${backend}::transaction_commit $db
    }
  }

  ##########
  #
  #
  #
  proc _attrset {} {
    uplevel 1 {
      variable db
      variable backend
      variable encrypt

      if {[set name_id [nameid $name]] == ""} {
        return -code error "var name \"$name\" does not exists"
      } elseif {fmod([llength $args],2)} {
        return -code error "Argument list must be a valid, 2 elements list, got \"[llength $args]\""
      }

      if {[set p [keyid $name_id $key]] == ""} {
        return -code error "key name \"$key\" does not exists"
      }

      if {$encrypt} {
        array set a [dodecbin [${backend}::node_attributes $db $p]]
      } else {
        array set a [${backend}::node_attributes $db $p]
      }

      ${backend}::transaction_begin_write $db

      foreach {k v} $args {
        if {[validname $k]} {
          ${backend}::transaction_commit $db
          if {[string length $k] > 128} {
            return -code error "Invalid size, max 128, got [string length $k]"
          } else {
            return -code error "Invalid characters"
          }
        }
        set a($k) $v
      }
      if {$encrypt} {
        set x [doencbin [array get a]]
      } else {
        set x [array get a]
      }
      ${backend}::node_attributes_set $db $p $x

      ${backend}::transaction_commit $db
    }
  }

  ##########
  #
  #
  #
  proc _attrget {} {
    uplevel 1 {
      variable db
      variable backend
      variable encrypt

      if {[set name_id [nameid $name]] == ""} {
        return -code error "var name \"$name\" does not exists"
      }

      if {[set p [keyid $name_id $key]] == ""} {
        return -code error "key name \"$key\" does not exists"
      }

      if {$encrypt} {
        array set a [dodecbin [${backend}::node_attributes $db $p]]
      } else {
        array set a [${backend}::node_attributes $db $p]
      }

      if {$attr == ""} {
        return [array names a]
      } elseif {[info exist a($attr)]} {
        return $a($attr)
      } else {
        return -code error "no attribute name \"$attr\" found in $key"
      }
    }
  }

  ##########
  #
  #
  #
  proc _attrdel {} {
    uplevel 1 {
      variable db
      variable backend
      variable encrypt

      if {[set name_id [nameid $name]] == ""} {
        return -code error "var name \"$name\" does not exists"
      }

      if {[set p [keyid $name_id $key]] == ""} {
        return -code error "key name \"$key\" does not exists"
      }

      if {$encrypt} {
        array set a [dodecbin [${backend}::node_attributes $db $p]]
      } else {
        array set a [${backend}::node_attributes $db $p]
      }

      if {[info exist a($attr)]} {
        unset a($attr)
        set x [array get a]
        if {$encrypt} {
          set x [doencbin $x]
        }

        ${backend}::transaction_begin_write $db
        ${backend}::node_attributes_set $db $p $x
        ${backend}::transaction_commit $db

      } else {
        return -code error "no attribute name \"$attr\" found in $key"
      }
    }
  }

  ##########
  #
  #
  #
  proc _tree {} {
    uplevel 1 {
      variable db
      if {$name ni [varlist]} {
        return -code error "var name does not exist \"$name\""
      }
      ::sqlk::__tree [namespace current] $name $key
    }
  }

  ##########
  #
  #
  #
  proc __tree {ns name key} {
    set y ""
    foreach x [${ns}::keys $name $key] {
      set nkey [join "$key $x" .]
      lappend y $nkey
      if {[set z [::sqlk::__tree $ns $name $nkey]] != ""} {
        lappend y $z
      }
    }
    return [join $y]
  }

  ##########
  #
  #
  #
  proc _serialize {} {
    uplevel 1 {
      variable db
      # nsk serialize name ?-format XML|TCL|TEXT? ?-indent 0|1? ?-key key?
      set oformat TCL
      set oindent 0
      set okey    ""

      # Check for options
      if {[llength $args] > 1} {
        if {[catch {array set op $args} e] || fmod([llength $args],2) != "0"} {
          return -code error "wrong # args: should be \"serialize ?-format XML|TCL|TEXT? ?-indent 0|1? ?-key key?\""
        }
        while {[llength [array names op]] > 0} {
          if {[info exist op(-format)]} {
            if {$op(-format) ni "XML TCL TEXT"} {
              return -code error "invalid format \"$op(-format)\": must be XML, TCL or TEXT"
            }
            set oformat $op(-format)
            unset op(-format)
          } elseif  {[info exist op(-indent)]} {
            if {$op(-indent) ni "0 1"} {
              return -code error "invalid indent value \"$op(-indent)\": must be 0 or 1"
            }
            set oindent $op(-indent)
            unset op(-indent)
          } elseif  {[info exist op(-key)]} {
            if {[catch {keys $name $op(-key)} e]} {
              return -code error $e
            }
            set okey $op(-key)
            unset op(-key)
          } else {
            return -code error "invalid arg \"[lindex [array names op] 0]\": must be -format or -indent"
          }
        }
      }

      # Check if name or procname exists
      if {$name ni [varlist]} {
        return -code error "var name does not exist \"$name\""
      }
      ::sqlk::__serialize [namespace current] $name $oformat $oindent $okey
    }
  }

  ##########
  #
  #
  #
  proc __serialize {ns name oformat oindent okey {level 1}} {
    if {$oformat == "TCL"} {
      set y [list]
      set i ""
      set s ""
      foreach x [${ns}::keys $name $okey] {
        set nkey [join "$okey $x" .]
        set d [__serialize $ns $name $oformat $oindent $nkey [expr $level + 1]]
        if {$oindent == "1"} {
          set s [string repeat " " [expr $level * 2]]
          set dd ""
          foreach n $d {
            append dd $s[list $n]\n
          }
          if {[llength $d] > 0} {
            set d \n$dd[string range $s 0 end-2]
          } else {
            set d ""
          }
        }
        array unset a
        foreach n [${ns}::attrget $name $nkey] {
          set a($n) [${ns}::attrget $name $nkey $n]
        }
        lappend y [list $x [${ns}::kget $name $nkey] [array get a] $d]
      }
      if {$oindent == "1"} {
        if {$level == "1"} {
          set dd ""
          foreach n $y {
            lappend dd [list $n]
          }
          set y [join $dd \n]
        }
        return $y
      } else {
        return $y
      }
    } elseif {$oformat == "TEXT"} {
      set y ""
      set i ""
      set s ""
      foreach x [${ns}::keys $name $okey] {
        set nkey [join "$okey $x" .]
        set d [__serialize $ns $name $oformat $oindent $nkey [expr $level + 1]]
        set a ""
        foreach n [${ns}::attrget $name $nkey] {
          lappend a "$n=\"[${ns}::attrget $name $nkey $n]\""
        }
        append y "$nkey\n [join $a]\n [${ns}::kget $name $nkey]\n$d"
      }
      return $y
    } elseif {$oformat == "XML"} {
      # TODO: XML serialization pending
    }
  }

  ##########
  #
  #
  #
  proc _parse {} {
    uplevel 1 {
      variable db

      # nsk parse ?-into key? ?-format XML|TCL? string
      set oformat TCL
      set oname ""
      set okey ""

      set data [lindex $args end]

      # Check for options
      if {[llength $args] > 1} {
        if {[catch {array set op [lrange $args 0 end-1]} e] || [llength $args] ni "3 5"} {
          return -code error "wrong # args: should be \"parse ?-format XML|TCL? ?-into key? string\""
        }
        while {[llength [array names op]] > 0} {
          if {[info exist op(-format)]} {
            if {$op(-format) ni "XML TCL"} {
              return -code error "invalid format \"$op(-format)\": must be XML or TCL"
            }
            set oformat $op(-format)
            unset op(-format)
          } elseif  {[info exist op(-into)]} {
            set okey $op(-into)
            unset op(-into)
          } else {
            return -code error "invalid arg \"[lindex [array names op] 0]\": must be -format or -into"
          }
        }
      }

      # Check if name or procname exists
      if {$name ni [varlist]} {
        return -code error "var name \"$name\" does not exists"
      }

      ::sqlk::__parse [namespace current] $name $data TCL $okey
      return ""
    }
  }

  ##########
  #
  #
  #
  proc __parse {ns name str oformat {key ""}} {
    if {$oformat == "TCL"} {
      foreach x $str {
        lassign $x k v a c
        set lkey [join "$key $k" .]
        ${ns}::kset $name $lkey $v
        foreach {ak av} $a {
          ${ns}::attrset $name $lkey $ak $av
        }
        if {[llength $c] > 0} {
          __parse $ns $name $c $oformat $lkey
        }
      }
    }
  }


  ##########
  #
  # AES encryption: deterministic IV for searchable names,
  # random IV for values/attributes.
  #
  proc _doenc {} {
    uplevel 1 {
      variable kenckey
      if {$kenckey == ""} {
        return $data
      } else {
        # Deterministic IV (all zeros) for names — must be searchable
        set iv [string repeat \x00 16]
        set key16 [string range [string repeat \x00 16] 0 15]
        set key16 [string replace $key16 0 [expr {[string length $kenckey]-1}] $kenckey]
        return [binary scan [::aes::aes -dir encrypt -key $key16 -iv $iv -- $data] H* hex ; set hex]
      }
    }
  }

  ##########
  #
  #
  #
  proc _dodec {} {
    uplevel 1 {
      variable kenckey
      if {$kenckey == ""} {
        return $data
      } else {
        set iv [string repeat \x00 16]
        set key16 [string range [string repeat \x00 16] 0 15]
        set key16 [string replace $key16 0 [expr {[string length $kenckey]-1}] $kenckey]
        return [::aes::aes -dir decrypt -key $key16 -iv $iv -- [binary format H* $data]]
      }
    }
  }

  ##########
  #
  # Random IV for values/attributes — stored as IV+ciphertext in binary.
  #
  proc _doencbin {} {
    uplevel 1 {
      variable kenckey
      if {$kenckey == ""} {
        return $data
      } else {
        set key16 [string range [string repeat \x00 16] 0 15]
        set key16 [string replace $key16 0 [expr {[string length $kenckey]-1}] $kenckey]
        set iv ""
        for {set i 0} {$i < 16} {incr i} {
          append iv [binary format c [expr {int(rand()*256)}]]
        }
        return $iv[::aes::aes -dir encrypt -key $key16 -iv $iv -- $data]
      }
    }
  }

  ##########
  #
  #
  #
  proc _dodecbin {} {
    uplevel 1 {
      variable kenckey
      if {$kenckey == ""} {
        return $data
      } else {
        set key16 [string range [string repeat \x00 16] 0 15]
        set key16 [string replace $key16 0 [expr {[string length $kenckey]-1}] $kenckey]
        set iv    [string range $data 0 15]
        set ctext [string range $data 16 end]
        return [::aes::aes -dir decrypt -key $key16 -iv $iv -- $ctext]
      }
    }
  }


  ##########
  #
  #
  #
  proc _nameid {} {
    uplevel 1 {
      variable db
      variable backend
      variable encrypt

      if {$encrypt} {
        variable kenckey
        set ename [doenc "$name[string repeat "+" [expr 45 - [string length $name]]]*****"]
        ${backend}::variable_id $db $ename 1
      } else {
        ${backend}::variable_id $db $name 0
      }
    }
  }


  ##########
  #
  #
  #
  proc _keyid {} {
    uplevel 1 {
      variable db
      variable backend
      ${backend}::node_id_by_path $db $name_id $key
    }
  }


  ##########
  #
  #
  #
  proc validname {name} {
    return [expr ![regexp -- {^[[:alnum:]][[:alnum:]_-]{1,40}$} $name]]
  }


  ##########
  #
  #
  #
  proc validkey {name} {
    return [expr ![regexp -- {^[[:alnum:]][[:alnum:]_.-]{0,}$} $name]]
  }

  ##########
  #
  #
  #
  proc inmiddle {a b} {
    # Integer
    set c [expr int(($a + $b) / 2)]
    if {$a > $b && $c > $b || $b > $a && $c > $a} {
      return $c
    }

    # Real
    set c [expr ($a + $b) / 2.0]
    if {$a > $b && $c > $b || $b > $a && $c > $a} {
      return $c
    }

    return 0
  }

  ##########
  #
  #
  #
  namespace eval backend {
    ##########
    #
    #
    #
    namespace eval sqlite {
      ##########
      #
      #
      #
      proc package_require {} {
        package require sqlite3
      }

      ##########
      #
      #
      #
      proc validate_target {target enckey} {
        if {$target == ":memory:" && [string length $enckey] > 0} {
          return -code error "Memory keyed lists can not be encrypted"
        }
      }

      ##########
      #
      #
      #
      proc target_public {target} {
        return $target
      }

      ##########
      #
      #
      #
      proc open {db filepath} {
        if {[catch {sqlite3 $db $filepath} e]} {
          return -code error $e
        }

        $db busy 10000
        $db timeout 10000

        return $db
      }

      ##########
      #
      #
      #
      proc close {db} {
        $db close
      }

      ##########
      #
      #
      #
      proc backup {db filename} {
        $db backup $filename
      }

      ##########
      #
      #
      #
      proc restore {db filename} {
        $db restore $filename
      }

      ##########
      #
      #
      #
      proc transaction_begin_write {db} {
        $db eval "begin immediate transaction"
      }

      ##########
      #
      #
      #
      proc transaction_commit {db} {
        $db eval {commit}
      }

      ##########
      #
      #
      #
      proc transaction_rollback {db} {
        $db eval "rollback transaction"
      }

      ##########
      #
      #
      #
      proc variable_list {db encrypt} {
        if {$encrypt} {
          $db eval {select name from key_name where encrypt = 1}
        } else {
          $db eval {select name from key_name where encrypt = 0}
        }
      }

      ##########
      #
      #
      #
      proc variable_id {db stored_name encrypt} {
        if {$encrypt} {
          $db onecolumn {select id from key_name where name = $stored_name}
        } else {
          $db onecolumn {select id from key_name where name = $stored_name and encrypt = 0}
        }
      }

      ##########
      #
      #
      #
      proc variable_insert {db stored_name encrypt} {
        $db eval {insert into key_name (name,encrypt,ctime,utime) values ($stored_name,$encrypt,strftime('%s'),strftime('%s'))}
      }

      ##########
      #
      #
      #
      proc variable_delete {db name_id} {
        $db eval {delete from key_name where id = $name_id}
      }

      ##########
      #
      #
      #
      proc node_id_by_path {db name_id path} {
        $db onecolumn {select id from key_data where name_id = $name_id and path = $path}
      }

      ##########
      #
      #
      #
      proc node_id_by_child {db name_id parent_id local_name} {
        $db onecolumn {select id from key_data where name_id = $name_id and parent = $parent_id and name = $local_name}
      }

      ##########
      #
      #
      #
      proc node_value {db node_id} {
        $db onecolumn {select val from key_data where id = $node_id}
      }

      ##########
      #
      #
      #
      proc node_insert_append {db name_id parent_id local_name} {
        $db eval {insert into key_data (name_id,parent,name,val,attr) values ($name_id,$parent_id,$local_name,'','')}
        $db last_insert_rowid
      }

      ##########
      #
      #
      #
      proc node_value_set {db node_id value} {
        $db eval {update key_data set val = $value where id = $node_id}
      }

      ##########
      #
      #
      #
      proc node_attributes {db node_id} {
        $db onecolumn {select attr from key_data where id = $node_id}
      }

      ##########
      #
      #
      #
      proc node_attributes_set {db node_id attributes} {
        $db eval {update key_data set attr = $attributes where id = $node_id}
      }

      ##########
      #
      #
      #
      proc node_parent {db node_id} {
        $db onecolumn {select parent from key_data where id = $node_id}
      }

      ##########
      #
      #
      #
      proc node_after_id {db node_id} {
        $db onecolumn {select after_id from key_data where id = $node_id}
      }

      ##########
      #
      #
      #
      proc node_follower {db parent_id after_id} {
        $db onecolumn {select id from key_data where after_id = $after_id and parent = $parent_id}
      }

      ##########
      #
      #
      #
      proc node_after_id_set {db node_id after_id} {
        $db eval {update key_data set after_id = $after_id where id = $node_id}
      }

      ##########
      #
      #
      #
      proc node_after_id_and_order_set {db node_id after_id key_order} {
        $db eval {update key_data set after_id = $after_id, key_order = $key_order where id = $node_id}
      }

      ##########
      #
      #
      #
      proc variable_last_id_set {db name_id last_id} {
        $db eval {update key_name set last_id = $last_id where id = $name_id}
      }

      ##########
      #
      #
      #
      proc node_last_id_set {db node_id last_id} {
        $db eval {update key_data set last_id = $last_id where id = $node_id}
      }

      ##########
      #
      #
      #
      proc node_children_ids {db parent_ids} {
        $db eval "select id from key_data where parent in ([join $parent_ids ,])"
      }

      ##########
      #
      #
      #
      proc node_children_names {db name_id parent_id} {
        $db eval {select name from key_sort where parent = $parent_id and name_id = $name_id}
      }

      ##########
      #
      #
      #
      proc node_delete_ids {db node_ids} {
        $db eval "delete from key_data where id in ([join $node_ids ,])"
      }

      ##########
      #
      #
      #
      proc node_is_child {db name_id parent_id node_id} {
        $db onecolumn {select count(*) from key_data where id = $node_id and parent = $parent_id and name_id = $name_id}
      }

      ##########
      #
      #
      #
      proc node_child_count {db parent_id} {
        $db onecolumn {select count(*) from key_data where parent = $parent_id}
      }

      ##########
      #
      #
      #
      proc node_follower_in_variable {db name_id parent_id after_id} {
        $db onecolumn {select id from key_data where after_id = $after_id and name_id = $name_id and parent = $parent_id}
      }

      ##########
      #
      #
      #
      proc node_first_with_order {db name_id parent_id} {
        $db eval {select id,key_order from key_data where after_id = 0 and parent = $parent_id and name_id = $name_id limit 1}
      }

      ##########
      #
      #
      #
      proc node_follower_with_order {db parent_id after_id} {
        $db eval {select id,key_order from key_data where after_id = $after_id and parent = $parent_id}
      }

      ##########
      #
      #
      #
      proc node_follower_with_order_in_variable {db name_id parent_id after_id} {
        $db eval {select id,key_order from key_data where after_id = $after_id and name_id = $name_id and parent = $parent_id}
      }

      ##########
      #
      #
      #
      proc node_key_order {db node_id} {
        $db onecolumn {select key_order from key_data where id = $node_id}
      }

      ##########
      #
      #
      #
      proc node_max_order_append {db parent_id} {
        $db onecolumn {select max(key_order)+0.0001 from key_data where parent = $parent_id}
      }

      ##########
      #
      #
      #
      proc node_key_order_set {db node_id key_order} {
        $db eval {update key_data set key_order = $key_order where id = $node_id}
      }

      ##########
      #
      #
      #
      proc node_sort_step {db name_id parent_id} {
        $db onecolumn {select max(id)/count(*) from key_data where parent = $parent_id and name_id = $name_id}
      }

      ##########
      #
      #
      #
      proc schema_init {db} {
        set tables [$db eval {
            SELECT name FROM sqlite_master
            WHERE type = 'table'
              AND (name = 'key_data' OR name = 'key_name')
            ORDER BY name
        }]
        if {$tables eq {key_data key_name}} {
          set cols {}
          $db eval "PRAGMA table_info(key_name)" {lappend cols $name}
          if {$cols ne {id name encrypt last_id ctime utime}} {
            error "Column names for key_name table do not match -> $cols"
          }

          set cols {}
          $db eval "PRAGMA table_info(key_data)" {lappend cols $name}
          if {$cols ne {id name_id parent after_id last_id key_order path ctime utime name val attr}} {
            error "Column names for key_data table do not match -> $cols"
          }

        } else {
          $db eval {
            PRAGMA auto_vacuum = 1;

            CREATE TABLE "key_name" (
                "id"      INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                "name"    TEXT NOT NULL,
                "encrypt" INTEGER NOT NULL DEFAULT(0),
                "last_id" INTEGER NOT NULL DEFAULT(0),
                "ctime"   INTEGER,
                "utime"   INTEGER,
                UNIQUE(name)
            );

            CREATE TABLE "key_data" (
                "id"         INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                "name_id"    INTEGER NOT NULL,
                "parent"     INTEGER NOT NULL,
                "after_id"   INTEGER,
                "last_id"    INTEGER NOT NULL DEFAULT(0),
                "key_order"  REAL,
                "path"       TEXT,
                "ctime"      INTEGER,
                "utime"      INTEGER,
                "name"       TEXT    NOT NULL,
                "val"        BLOB,
                "attr"       BLOB,
                UNIQUE(name,parent,name_id)
            );

            CREATE INDEX key_data_key0 on key_data (path);
            CREATE INDEX key_data_key1 on key_data (parent,after_id,key_order);
            CREATE INDEX key_data_key2 on key_data (name_id,path);

            CREATE TRIGGER key_name_delete AFTER DELETE ON key_name
             FOR EACH ROW
              BEGIN
                delete from key_data where name_id = OLD.id;
              END;

            CREATE TRIGGER key_data_insert0 AFTER INSERT on key_data
             FOR EACH ROW
             WHEN NEW.parent = 0
              BEGIN
                update key_data
                  set
                    after_id =
                     (select a.last_id from key_name as a where id = NEW.name_id),
                    key_order =
                     NEW.id,
                    path =
                     ifnull((select p.name || '.' from key_data as p where p.id = NEW.parent),'') || NEW.name,
                    ctime = strftime('%s'),
                    utime = strftime('%s')
                  where
                    id = NEW.id;

                update key_name
                  set
                    utime = strftime('%s'),
                    last_id = NEW.id
                  where
                    id = NEW.name_id;
              END;

            CREATE TRIGGER key_data_insert1 AFTER INSERT on key_data
             FOR EACH ROW
              WHEN NEW.parent <> 0
              BEGIN
                update key_data
                  set
                    after_id =
                     (select a.last_id from key_data as a where a.id = NEW.parent),
                    key_order =
                     NEW.id,
                    path =
                     ifnull((select p.path || '.' from key_data as p where p.id = NEW.parent),'') || NEW.name,
                    ctime = strftime('%s'),
                    utime = strftime('%s')
                  where
                    id = NEW.id;

                update key_data
                  set
                    last_id = NEW.id
                  where
                    id = NEW.parent;

                update key_name
                  set
                    utime = strftime('%s')
                  where
                    id = NEW.name_id;
              END;


            CREATE TRIGGER key_data_update AFTER UPDATE OF name, val, attr ON key_data
             FOR EACH ROW
              BEGIN
                update key_data
                  set
                    utime = strftime('%s')
                  where
                    id = OLD.id;
                update key_name
                  set
                    utime = strftime('%s')
                  where
                    id = OLD.name_id;
              END;

            CREATE VIEW "key_sort" AS
              select * from key_data order by key_order;
          }
        }
      }
    }

    ##########
    #
    #
    #
    namespace eval mysql {
      ##########
      #
      #
      #
      proc package_require {} {
        package require -exact mysqltcl 3.052
      }

      ##########
      #
      #
      #
      proc validate_target {target enckey} {
        if {[catch {llength $target} target_length] || fmod($target_length,2)} {
          return -code error "Invalid MySQL target: expected an option-value list"
        }

        set seen [list]
        foreach {option value} $target {
          if {[lsearch -exact $seen $option] != -1} {
            return -code error "Duplicate MySQL target option \"$option\""
          }
          lappend seen $option
        }

        set have_host 0
        set have_socket 0
        set have_db 0
        foreach {option value} $target {
          switch -- $option {
            -host {
              set have_host 1
            }
            -port -
            -user -
            -password -
            -ssl -
            -compress {
            }
            -socket {
              set have_socket 1
            }
            -db {
              set have_db 1
            }
            -reconnect {
              return -code error "MySQL target option \"-reconnect\" is not supported"
            }
            default {
              return -code error "Invalid MySQL target option \"$option\""
            }
          }
        }

        if {$have_host && $have_socket} {
          return -code error "Conflicting MySQL target options \"-host\" and \"-socket\""
        }
        if {!$have_db} {
          return -code error "Missing required MySQL target option \"-db\""
        }
      }

      ##########
      #
      #
      #
      proc target_public {target} {
        set public_target [list]
        foreach {option value} $target {
          if {$option != "-password"} {
            lappend public_target $option $value
          }
        }
        return $public_target
      }

      ##########
      #
      #
      #
      proc open {db target} {
        ::mysql::connect {*}$target -encoding utf-8
      }

      ##########
      #
      #
      #
      proc close {db} {
        ::mysql::close $db
      }

      ##########
      #
      #
      #
      proc backup {db filename} {
        return -code error "Operation \"backup\" is not supported by backend \"mysql\""
      }

      ##########
      #
      #
      #
      proc restore {db filename} {
        return -code error "Operation \"restore\" is not supported by backend \"mysql\""
      }

      ##########
      #
      #
      #
      proc transaction_begin_write {db} {
        set state [::mysql::sel $db {SELECT @@autocommit} -flatlist]
        set state [lindex $state 0]

        if {$state == 0} {
          return -code error "cannot start a transaction within a transaction"
        }

        set code [catch {
          ::mysql::autocommit $db 0

          set lock [::mysql::sel $db {
            SELECT lock_id
            FROM sqlk_writer_lock
            WHERE lock_id = 1
            FOR UPDATE
          } -flatlist]
          set lock [lindex $lock 0]

          if {$lock ne "1"} {
            error "MySQL sqlk writer lock row is missing"
          }
        } result options]

        if {$code} {
          catch {::mysql::rollback $db}
          catch {::mysql::autocommit $db 1}
          return -options $options $result
        }

        return ""
      }

      ##########
      #
      #
      #
      proc transaction_commit {db} {
        set state [::mysql::sel $db {SELECT @@autocommit} -flatlist]
        set state [lindex $state 0]

        if {$state != 0} {
          return -code error "cannot commit - no transaction is active"
        }

        ::mysql::commit $db
        ::mysql::autocommit $db 1
        return ""
      }

      ##########
      #
      #
      #
      proc transaction_rollback {db} {
        set state [::mysql::sel $db {SELECT @@autocommit} -flatlist]
        set state [lindex $state 0]

        if {$state != 0} {
          return -code error "cannot rollback - no transaction is active"
        }

        ::mysql::rollback $db
        ::mysql::autocommit $db 1
        return ""
      }

      ##########
      #
      #
      #
      proc variable_list {db encrypt} {
        ::mysql::sel $db "SELECT name FROM key_name WHERE encrypt = $encrypt" -flatlist
      }

      ##########
      #
      #
      #
      proc variable_id {db stored_name encrypt} {
        set escaped_name [::mysql::escape $db $stored_name]
        set sql "SELECT id FROM key_name WHERE name_hash = UNHEX(SHA2(CAST('$escaped_name' AS BINARY),256)) AND BINARY name = BINARY '$escaped_name'"
        if {!$encrypt} {
          append sql " AND encrypt = 0"
        }
        append sql " LIMIT 1"

        lindex [::mysql::sel $db $sql -flatlist] 0
      }

      ##########
      #
      #
      #
      proc variable_insert {db stored_name encrypt} {
        set escaped_name [::mysql::escape $db $stored_name]
        ::mysql::exec $db "INSERT INTO key_name (name,encrypt,ctime,utime) VALUES ('$escaped_name',$encrypt,UNIX_TIMESTAMP(),UNIX_TIMESTAMP())"
        return ""
      }

      ##########
      #
      #
      #
      proc variable_delete {db name_id} {
        transaction_begin_write $db

        set code [catch {
          ::mysql::exec $db "DELETE FROM key_name WHERE id = $name_id"
        } result options]

        if {$code} {
          catch {transaction_rollback $db}
          return -options $options $result
        }

        transaction_commit $db
        return ""
      }

      ##########
      #
      #
      #
      proc variable_last_id_set {db name_id last_id} {
        ::mysql::exec $db "UPDATE key_name SET last_id = $last_id WHERE id = $name_id"
        return ""
      }

      ##########
      #
      #
      #
      proc node_id_by_path {db name_id path} {
        set escaped_path [::mysql::escape $db $path]
        set sql "SELECT id FROM key_data WHERE name_id = $name_id AND path_hash = UNHEX(SHA2(CAST('$escaped_path' AS BINARY),256)) AND BINARY path = BINARY '$escaped_path' LIMIT 1"

        lindex [::mysql::sel $db $sql -flatlist] 0
      }

      ##########
      #
      #
      #
      proc node_id_by_child {db name_id parent_id local_name} {
        set escaped_name [::mysql::escape $db $local_name]
        set sql "SELECT id FROM key_data WHERE name_id = $name_id AND parent = $parent_id AND name_hash = UNHEX(SHA2(CAST('$escaped_name' AS BINARY),256)) AND BINARY name = BINARY '$escaped_name' LIMIT 1"

        lindex [::mysql::sel $db $sql -flatlist] 0
      }

      ##########
      #
      #
      #
      proc node_value {db node_id} {
        set result [::mysql::sel $db "SELECT kn.encrypt,CASE WHEN kn.encrypt = 1 THEN HEX(kd.val) ELSE CONVERT(kd.val USING utf8mb4) END FROM key_data AS kd JOIN key_name AS kn ON kn.id = kd.name_id WHERE kd.id = $node_id LIMIT 1" -flatlist]
        if {$result eq ""} {
          return ""
        }

        lassign $result encrypt value
        if {$encrypt} {
          return [binary format H* $value]
        }
        return $value
      }

      ##########
      #
      #
      #
      proc node_insert_append {db name_id parent_id local_name} {
        if {$parent_id == 0} {
          set previous_last_id [lindex [::mysql::sel $db "SELECT last_id FROM key_name WHERE id = $name_id FOR UPDATE" -flatlist] 0]
          set full_path $local_name
        } else {
          set parent [::mysql::sel $db "SELECT IFNULL(CONCAT(path,'.'),''),last_id FROM key_data WHERE id = $parent_id FOR UPDATE" -flatlist]
          lassign $parent path_prefix previous_last_id
          set full_path $path_prefix$local_name
        }

        if {$previous_last_id == ""} {
          set after_id NULL
        } else {
          set after_id $previous_last_id
        }

        set escaped_name [::mysql::escape $db $local_name]
        set escaped_path [::mysql::escape $db $full_path]

        ::mysql::exec $db "INSERT INTO key_data (name_id,parent,after_id,last_id,key_order,path,ctime,utime,name,val,attr) VALUES ($name_id,$parent_id,$after_id,0,NULL,'$escaped_path',UNIX_TIMESTAMP(),UNIX_TIMESTAMP(),'$escaped_name',X'',X'')"
        set node_id [::mysql::insertid $db]

        ::mysql::exec $db "UPDATE key_data SET key_order = id WHERE id = $node_id"

        if {$parent_id == 0} {
          ::mysql::exec $db "UPDATE key_name AS kn JOIN key_data AS kd ON kd.id = $node_id SET kn.last_id = $node_id, kn.utime = kd.ctime WHERE kn.id = $name_id"
        } else {
          ::mysql::exec $db "UPDATE key_data SET last_id = $node_id WHERE id = $parent_id"
          ::mysql::exec $db "UPDATE key_name AS kn JOIN key_data AS kd ON kd.id = $node_id SET kn.utime = kd.ctime WHERE kn.id = $name_id"
        }

        return $node_id
      }

      ##########
      #
      #
      #
      proc node_value_set {db node_id value} {
        set encrypt [lindex [::mysql::sel $db "SELECT kn.encrypt FROM key_data AS kd JOIN key_name AS kn ON kn.id = kd.name_id WHERE kd.id = $node_id LIMIT 1" -flatlist] 0]
        if {$encrypt eq ""} {
          return ""
        }

        if {$encrypt} {
          set value_hex [binary encode hex $value]
          set escaped_value [::mysql::escape $db $value_hex]
          set stored_value "UNHEX('$escaped_value')"
        } else {
          set escaped_value [::mysql::escape $db $value]
          set stored_value "'$escaped_value'"
        }

        ::mysql::exec $db "UPDATE key_data SET val = $stored_value, utime = UNIX_TIMESTAMP() WHERE id = $node_id"
        ::mysql::exec $db "UPDATE key_name AS kn JOIN key_data AS kd ON kd.name_id = kn.id SET kn.utime = kd.utime WHERE kd.id = $node_id"
        return ""
      }

      ##########
      #
      #
      #
      proc node_attributes {db node_id} {
        set result [::mysql::sel $db "SELECT kn.encrypt,CASE WHEN kn.encrypt = 1 THEN HEX(kd.attr) ELSE CONVERT(kd.attr USING utf8mb4) END FROM key_data AS kd JOIN key_name AS kn ON kn.id = kd.name_id WHERE kd.id = $node_id LIMIT 1" -flatlist]
        if {$result eq ""} {
          return ""
        }

        lassign $result encrypt attributes
        if {$encrypt} {
          return [binary format H* $attributes]
        }
        return $attributes
      }

      ##########
      #
      #
      #
      proc node_attributes_set {db node_id attributes} {
        set encrypt [lindex [::mysql::sel $db "SELECT kn.encrypt FROM key_data AS kd JOIN key_name AS kn ON kn.id = kd.name_id WHERE kd.id = $node_id LIMIT 1" -flatlist] 0]
        if {$encrypt eq ""} {
          return ""
        }

        if {$encrypt} {
          set attributes_hex [binary encode hex $attributes]
          set escaped_attributes [::mysql::escape $db $attributes_hex]
          set stored_attributes "UNHEX('$escaped_attributes')"
        } else {
          set escaped_attributes [::mysql::escape $db $attributes]
          set stored_attributes "'$escaped_attributes'"
        }

        ::mysql::exec $db "UPDATE key_data SET attr = $stored_attributes, utime = UNIX_TIMESTAMP() WHERE id = $node_id"
        ::mysql::exec $db "UPDATE key_name AS kn JOIN key_data AS kd ON kd.name_id = kn.id SET kn.utime = kd.utime WHERE kd.id = $node_id"
        return ""
      }

      ##########
      #
      #
      #
      proc node_children_names {db name_id parent_id} {
        ::mysql::sel $db "SELECT name FROM key_data WHERE parent = $parent_id AND name_id = $name_id ORDER BY key_order" -flatlist
      }

      ##########
      #
      #
      #
      proc node_parent {db node_id} {
        lindex [::mysql::sel $db "SELECT parent FROM key_data WHERE id = $node_id FOR UPDATE" -flatlist] 0
      }

      ##########
      #
      #
      #
      proc node_after_id {db node_id} {
        lindex [::mysql::sel $db "SELECT after_id FROM key_data WHERE id = $node_id FOR UPDATE" -flatlist] 0
      }

      ##########
      #
      #
      #
      proc node_follower {db parent_id after_id} {
        lindex [::mysql::sel $db "SELECT id FROM key_data WHERE after_id = $after_id AND parent = $parent_id FOR UPDATE" -flatlist] 0
      }

      ##########
      #
      #
      #
      proc node_after_id_set {db node_id after_id} {
        ::mysql::exec $db "UPDATE key_data SET after_id = $after_id WHERE id = $node_id"
        return ""
      }

      ##########
      #
      #
      #
      proc node_after_id_and_order_set {db node_id after_id key_order} {
        ::mysql::exec $db "UPDATE key_data SET after_id = $after_id, key_order = $key_order WHERE id = $node_id"
        return ""
      }

      ##########
      #
      #
      #
      proc node_last_id_set {db node_id last_id} {
        ::mysql::exec $db "UPDATE key_data SET last_id = $last_id WHERE id = $node_id"
        return ""
      }

      ##########
      #
      #
      #
      proc node_children_ids {db parent_ids} {
        if {$parent_ids eq ""} {
          return ""
        }

        ::mysql::sel $db "SELECT id FROM key_data WHERE parent IN ([join $parent_ids ,]) FOR UPDATE" -flatlist
      }

      ##########
      #
      #
      #
      proc node_delete_ids {db node_ids} {
        if {$node_ids eq ""} {
          return ""
        }

        ::mysql::exec $db "DELETE FROM key_data WHERE id IN ([join $node_ids ,])"
        return ""
      }

      ##########
      #
      #
      #
      proc node_is_child {db name_id parent_id node_id} {
        lindex [::mysql::sel $db "SELECT COUNT(*) FROM key_data WHERE id = $node_id AND parent = $parent_id AND name_id = $name_id FOR UPDATE" -flatlist] 0
      }

      ##########
      #
      #
      #
      proc node_child_count {db parent_id} {
        lindex [::mysql::sel $db "SELECT COUNT(*) FROM key_data WHERE parent = $parent_id FOR UPDATE" -flatlist] 0
      }

      ##########
      #
      #
      #
      proc node_follower_in_variable {db name_id parent_id after_id} {
        lindex [::mysql::sel $db "SELECT id FROM key_data WHERE after_id = $after_id AND name_id = $name_id AND parent = $parent_id FOR UPDATE" -flatlist] 0
      }

      ##########
      #
      #
      #
      proc node_first_with_order {db name_id parent_id} {
        ::mysql::sel $db "SELECT id,key_order FROM key_data WHERE after_id = 0 AND parent = $parent_id AND name_id = $name_id LIMIT 1 FOR UPDATE" -flatlist
      }

      ##########
      #
      #
      #
      proc node_follower_with_order {db parent_id after_id} {
        ::mysql::sel $db "SELECT id,key_order FROM key_data WHERE after_id = $after_id AND parent = $parent_id FOR UPDATE" -flatlist
      }

      ##########
      #
      #
      #
      proc node_follower_with_order_in_variable {db name_id parent_id after_id} {
        ::mysql::sel $db "SELECT id,key_order FROM key_data WHERE after_id = $after_id AND name_id = $name_id AND parent = $parent_id FOR UPDATE" -flatlist
      }

      ##########
      #
      #
      #
      proc node_key_order {db node_id} {
        lindex [::mysql::sel $db "SELECT key_order FROM key_data WHERE id = $node_id FOR UPDATE" -flatlist] 0
      }

      ##########
      #
      #
      #
      proc node_max_order_append {db parent_id} {
        lindex [::mysql::sel $db "SELECT MAX(key_order)+0.0001 FROM key_data WHERE parent = $parent_id FOR UPDATE" -flatlist] 0
      }

      ##########
      #
      #
      #
      proc node_key_order_set {db node_id key_order} {
        ::mysql::exec $db "UPDATE key_data SET key_order = $key_order WHERE id = $node_id"
        return ""
      }

      ##########
      #
      #
      #
      proc node_sort_step {db name_id parent_id} {
        lindex [::mysql::sel $db "SELECT FLOOR(MAX(id)/COUNT(*)) FROM key_data WHERE parent = $parent_id AND name_id = $name_id FOR UPDATE" -flatlist] 0
      }

      ##########
      #
      #
      #
      proc _schema_create {db} {
        ::mysql::exec $db {
          CREATE TABLE sqlk_writer_lock (
            lock_id TINYINT UNSIGNED NOT NULL,
            PRIMARY KEY (lock_id),
            CONSTRAINT sqlk_writer_lock_singleton CHECK (lock_id = 1)
          ) ENGINE=InnoDB
            DEFAULT CHARACTER SET utf8mb4
            COLLATE utf8mb4_0900_bin
        }

        ::mysql::exec $db {
          CREATE TABLE key_name (
            id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
            name LONGTEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin NOT NULL,
            name_hash BINARY(32) GENERATED ALWAYS AS (UNHEX(SHA2(CAST(name AS BINARY),256))) STORED,
            encrypt TINYINT UNSIGNED NOT NULL DEFAULT 0,
            last_id BIGINT UNSIGNED NOT NULL DEFAULT 0,
            ctime BIGINT NULL,
            utime BIGINT NULL,
            PRIMARY KEY (id),
            KEY key_name_name_hash (name_hash)
          ) ENGINE=InnoDB
            DEFAULT CHARACTER SET utf8mb4
            COLLATE utf8mb4_0900_bin
        }

        ::mysql::exec $db {
          CREATE TABLE key_data (
            id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
            name_id BIGINT UNSIGNED NOT NULL,
            parent BIGINT UNSIGNED NOT NULL,
            after_id BIGINT UNSIGNED NULL,
            last_id BIGINT UNSIGNED NOT NULL DEFAULT 0,
            key_order DOUBLE NULL,
            path LONGTEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin NULL,
            path_hash BINARY(32) GENERATED ALWAYS AS (UNHEX(SHA2(CAST(path AS BINARY),256))) STORED,
            ctime BIGINT NULL,
            utime BIGINT NULL,
            name LONGTEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin NOT NULL,
            name_hash BINARY(32) GENERATED ALWAYS AS (UNHEX(SHA2(CAST(name AS BINARY),256))) STORED,
            val LONGBLOB NULL,
            attr LONGBLOB NULL,
            PRIMARY KEY (id),
            KEY key_data_path_hash (path_hash),
            KEY key_data_parent_after_order (parent,after_id,key_order),
            KEY key_data_name_path_hash (name_id,path_hash),
            KEY key_data_child_name_hash (name_id,parent,name_hash)
          ) ENGINE=InnoDB
            DEFAULT CHARACTER SET utf8mb4
            COLLATE utf8mb4_0900_bin
        }

        ::mysql::exec $db {INSERT INTO sqlk_writer_lock (lock_id) VALUES (1)}

        ::mysql::exec $db {CREATE TRIGGER key_name_exact_insert BEFORE INSERT ON key_name FOR EACH ROW BEGIN IF EXISTS (SELECT 1 FROM key_name WHERE name_hash=UNHEX(SHA2(CAST(NEW.name AS BINARY),256)) AND name=NEW.name COLLATE utf8mb4_0900_bin) THEN SIGNAL SQLSTATE '23000' SET MYSQL_ERRNO=1062, MESSAGE_TEXT='duplicate key_name.name'; END IF; END}
        ::mysql::exec $db {CREATE TRIGGER key_name_exact_update BEFORE UPDATE ON key_name FOR EACH ROW BEGIN IF NOT (NEW.name <=> OLD.name) AND EXISTS (SELECT 1 FROM key_name WHERE id<>OLD.id AND name_hash=UNHEX(SHA2(CAST(NEW.name AS BINARY),256)) AND name=NEW.name COLLATE utf8mb4_0900_bin) THEN SIGNAL SQLSTATE '23000' SET MYSQL_ERRNO=1062, MESSAGE_TEXT='duplicate key_name.name'; END IF; END}
        ::mysql::exec $db {CREATE TRIGGER key_name_delete AFTER DELETE ON key_name FOR EACH ROW DELETE FROM key_data WHERE name_id=OLD.id}
        ::mysql::exec $db {CREATE TRIGGER key_data_exact_insert BEFORE INSERT ON key_data FOR EACH ROW BEGIN IF EXISTS (SELECT 1 FROM key_data WHERE name_id=NEW.name_id AND parent=NEW.parent AND name_hash=UNHEX(SHA2(CAST(NEW.name AS BINARY),256)) AND name=NEW.name COLLATE utf8mb4_0900_bin) THEN SIGNAL SQLSTATE '23000' SET MYSQL_ERRNO=1062, MESSAGE_TEXT='duplicate key_data sibling'; END IF; END}
        ::mysql::exec $db {CREATE TRIGGER key_data_before_update BEFORE UPDATE ON key_data FOR EACH ROW BEGIN IF (NOT (NEW.name_id <=> OLD.name_id) OR NOT (NEW.parent <=> OLD.parent) OR NOT (NEW.name <=> OLD.name)) AND EXISTS (SELECT 1 FROM key_data WHERE id<>OLD.id AND name_id=NEW.name_id AND parent=NEW.parent AND name_hash=UNHEX(SHA2(CAST(NEW.name AS BINARY),256)) AND name=NEW.name COLLATE utf8mb4_0900_bin) THEN SIGNAL SQLSTATE '23000' SET MYSQL_ERRNO=1062, MESSAGE_TEXT='duplicate key_data sibling'; END IF; IF NOT (NEW.name <=> OLD.name) OR NOT (NEW.val <=> OLD.val) OR NOT (NEW.attr <=> OLD.attr) THEN SET NEW.utime=UNIX_TIMESTAMP(); END IF; END}
        ::mysql::exec $db {CREATE TRIGGER key_data_after_update AFTER UPDATE ON key_data FOR EACH ROW BEGIN IF NOT (NEW.name <=> OLD.name) OR NOT (NEW.val <=> OLD.val) OR NOT (NEW.attr <=> OLD.attr) THEN UPDATE key_name SET utime=UNIX_TIMESTAMP() WHERE id=OLD.name_id; END IF; END}
      }

      ##########
      #
      #
      #
      proc _schema_validate {db} {
        set tables [::mysql::sel $db {
          SELECT TABLE_NAME,ENGINE,TABLE_COLLATION
          FROM information_schema.TABLES
          WHERE TABLE_SCHEMA=DATABASE()
            AND TABLE_NAME IN ('sqlk_writer_lock','key_name','key_data')
          ORDER BY TABLE_NAME
        } -list]
        set expected_tables [list \
          [list key_data InnoDB utf8mb4_0900_bin] \
          [list key_name InnoDB utf8mb4_0900_bin] \
          [list sqlk_writer_lock InnoDB utf8mb4_0900_bin]]
        if {$tables ne $expected_tables} {
          return -code error "MySQL sqlk schema validation failed: incompatible tables"
        }

        set columns [::mysql::sel $db {
          SELECT TABLE_NAME,COLUMN_NAME,ORDINAL_POSITION,
                 IFNULL(COLUMN_DEFAULT,'<NULL>'),IS_NULLABLE,DATA_TYPE,COLUMN_TYPE,
                 IFNULL(CHARACTER_SET_NAME,'<NULL>'),IFNULL(COLLATION_NAME,'<NULL>'),
                 EXTRA,GENERATION_EXPRESSION
          FROM information_schema.COLUMNS
          WHERE TABLE_SCHEMA=DATABASE()
            AND TABLE_NAME IN ('sqlk_writer_lock','key_name','key_data')
          ORDER BY TABLE_NAME,ORDINAL_POSITION
        } -list]
        set expected_columns [list \
          [list key_data id 1 <NULL> NO bigint {bigint unsigned} <NULL> <NULL> auto_increment {}] \
          [list key_data name_id 2 <NULL> NO bigint {bigint unsigned} <NULL> <NULL> {} {}] \
          [list key_data parent 3 <NULL> NO bigint {bigint unsigned} <NULL> <NULL> {} {}] \
          [list key_data after_id 4 <NULL> YES bigint {bigint unsigned} <NULL> <NULL> {} {}] \
          [list key_data last_id 5 0 NO bigint {bigint unsigned} <NULL> <NULL> {} {}] \
          [list key_data key_order 6 <NULL> YES double double <NULL> <NULL> {} {}] \
          [list key_data path 7 <NULL> YES longtext longtext utf8mb4 utf8mb4_0900_bin {} {}] \
          [list key_data path_hash 8 <NULL> YES binary binary(32) <NULL> <NULL> {STORED GENERATED} {unhex(sha2(cast(`path` as char charset binary),256))}] \
          [list key_data ctime 9 <NULL> YES bigint bigint <NULL> <NULL> {} {}] \
          [list key_data utime 10 <NULL> YES bigint bigint <NULL> <NULL> {} {}] \
          [list key_data name 11 <NULL> NO longtext longtext utf8mb4 utf8mb4_0900_bin {} {}] \
          [list key_data name_hash 12 <NULL> YES binary binary(32) <NULL> <NULL> {STORED GENERATED} {unhex(sha2(cast(`name` as char charset binary),256))}] \
          [list key_data val 13 <NULL> YES longblob longblob <NULL> <NULL> {} {}] \
          [list key_data attr 14 <NULL> YES longblob longblob <NULL> <NULL> {} {}] \
          [list key_name id 1 <NULL> NO bigint {bigint unsigned} <NULL> <NULL> auto_increment {}] \
          [list key_name name 2 <NULL> NO longtext longtext utf8mb4 utf8mb4_0900_bin {} {}] \
          [list key_name name_hash 3 <NULL> YES binary binary(32) <NULL> <NULL> {STORED GENERATED} {unhex(sha2(cast(`name` as char charset binary),256))}] \
          [list key_name encrypt 4 0 NO tinyint {tinyint unsigned} <NULL> <NULL> {} {}] \
          [list key_name last_id 5 0 NO bigint {bigint unsigned} <NULL> <NULL> {} {}] \
          [list key_name ctime 6 <NULL> YES bigint bigint <NULL> <NULL> {} {}] \
          [list key_name utime 7 <NULL> YES bigint bigint <NULL> <NULL> {} {}] \
          [list sqlk_writer_lock lock_id 1 <NULL> NO tinyint {tinyint unsigned} <NULL> <NULL> {} {}]]
        if {$columns ne $expected_columns} {
          return -code error "MySQL sqlk schema validation failed: incompatible columns"
        }

        set indexes [::mysql::sel $db {
          SELECT TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,
                 IFNULL(SUB_PART,'<NULL>'),INDEX_TYPE
          FROM information_schema.STATISTICS
          WHERE TABLE_SCHEMA=DATABASE()
            AND TABLE_NAME IN ('sqlk_writer_lock','key_name','key_data')
          ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX
        } -list]
        set expected_indexes [list \
          [list key_data key_data_child_name_hash 1 1 name_id <NULL> BTREE] \
          [list key_data key_data_child_name_hash 1 2 parent <NULL> BTREE] \
          [list key_data key_data_child_name_hash 1 3 name_hash <NULL> BTREE] \
          [list key_data key_data_name_path_hash 1 1 name_id <NULL> BTREE] \
          [list key_data key_data_name_path_hash 1 2 path_hash <NULL> BTREE] \
          [list key_data key_data_parent_after_order 1 1 parent <NULL> BTREE] \
          [list key_data key_data_parent_after_order 1 2 after_id <NULL> BTREE] \
          [list key_data key_data_parent_after_order 1 3 key_order <NULL> BTREE] \
          [list key_data key_data_path_hash 1 1 path_hash <NULL> BTREE] \
          [list key_data PRIMARY 0 1 id <NULL> BTREE] \
          [list key_name key_name_name_hash 1 1 name_hash <NULL> BTREE] \
          [list key_name PRIMARY 0 1 id <NULL> BTREE] \
          [list sqlk_writer_lock PRIMARY 0 1 lock_id <NULL> BTREE]]
        if {$indexes ne $expected_indexes} {
          return -code error "MySQL sqlk schema validation failed: incompatible indexes"
        }

        set constraints [::mysql::sel $db {
          SELECT TABLE_NAME,CONSTRAINT_NAME,CONSTRAINT_TYPE
          FROM information_schema.TABLE_CONSTRAINTS
          WHERE TABLE_SCHEMA=DATABASE()
            AND TABLE_NAME IN ('sqlk_writer_lock','key_name','key_data')
          ORDER BY TABLE_NAME,CONSTRAINT_NAME
        } -list]
        set expected_constraints [list \
          [list key_data PRIMARY {PRIMARY KEY}] \
          [list key_name PRIMARY {PRIMARY KEY}] \
          [list sqlk_writer_lock PRIMARY {PRIMARY KEY}] \
          [list sqlk_writer_lock sqlk_writer_lock_singleton CHECK]]
        if {$constraints ne $expected_constraints} {
          return -code error "MySQL sqlk schema validation failed: incompatible constraints"
        }

        set checks [::mysql::sel $db {
          SELECT tc.TABLE_NAME,cc.CONSTRAINT_NAME,cc.CHECK_CLAUSE
          FROM information_schema.CHECK_CONSTRAINTS AS cc
          JOIN information_schema.TABLE_CONSTRAINTS AS tc
            ON tc.CONSTRAINT_SCHEMA=cc.CONSTRAINT_SCHEMA
           AND tc.CONSTRAINT_NAME=cc.CONSTRAINT_NAME
          WHERE cc.CONSTRAINT_SCHEMA=DATABASE()
            AND tc.TABLE_NAME IN ('sqlk_writer_lock','key_name','key_data')
          ORDER BY tc.TABLE_NAME,cc.CONSTRAINT_NAME
        } -list]
        set expected_checks [list \
          [list sqlk_writer_lock sqlk_writer_lock_singleton {(`lock_id` = 1)}]]
        if {$checks ne $expected_checks} {
          return -code error "MySQL sqlk schema validation failed: incompatible checks"
        }

        set triggers [::mysql::sel $db {
          SELECT TRIGGER_NAME,EVENT_MANIPULATION,ACTION_TIMING,
                 EVENT_OBJECT_TABLE,ACTION_STATEMENT
          FROM information_schema.TRIGGERS
          WHERE TRIGGER_SCHEMA=DATABASE()
            AND EVENT_OBJECT_TABLE IN ('key_name','key_data')
          ORDER BY TRIGGER_NAME
        } -list]
        set expected_triggers [list \
          [list key_data_after_update UPDATE AFTER key_data {BEGIN IF NOT (NEW.name <=> OLD.name) OR NOT (NEW.val <=> OLD.val) OR NOT (NEW.attr <=> OLD.attr) THEN UPDATE key_name SET utime=UNIX_TIMESTAMP() WHERE id=OLD.name_id; END IF; END}] \
          [list key_data_before_update UPDATE BEFORE key_data {BEGIN IF (NOT (NEW.name_id <=> OLD.name_id) OR NOT (NEW.parent <=> OLD.parent) OR NOT (NEW.name <=> OLD.name)) AND EXISTS (SELECT 1 FROM key_data WHERE id<>OLD.id AND name_id=NEW.name_id AND parent=NEW.parent AND name_hash=UNHEX(SHA2(CAST(NEW.name AS BINARY),256)) AND name=NEW.name COLLATE utf8mb4_0900_bin) THEN SIGNAL SQLSTATE '23000' SET MYSQL_ERRNO=1062, MESSAGE_TEXT='duplicate key_data sibling'; END IF; IF NOT (NEW.name <=> OLD.name) OR NOT (NEW.val <=> OLD.val) OR NOT (NEW.attr <=> OLD.attr) THEN SET NEW.utime=UNIX_TIMESTAMP(); END IF; END}] \
          [list key_data_exact_insert INSERT BEFORE key_data {BEGIN IF EXISTS (SELECT 1 FROM key_data WHERE name_id=NEW.name_id AND parent=NEW.parent AND name_hash=UNHEX(SHA2(CAST(NEW.name AS BINARY),256)) AND name=NEW.name COLLATE utf8mb4_0900_bin) THEN SIGNAL SQLSTATE '23000' SET MYSQL_ERRNO=1062, MESSAGE_TEXT='duplicate key_data sibling'; END IF; END}] \
          [list key_name_delete DELETE AFTER key_name {DELETE FROM key_data WHERE name_id=OLD.id}] \
          [list key_name_exact_insert INSERT BEFORE key_name {BEGIN IF EXISTS (SELECT 1 FROM key_name WHERE name_hash=UNHEX(SHA2(CAST(NEW.name AS BINARY),256)) AND name=NEW.name COLLATE utf8mb4_0900_bin) THEN SIGNAL SQLSTATE '23000' SET MYSQL_ERRNO=1062, MESSAGE_TEXT='duplicate key_name.name'; END IF; END}] \
          [list key_name_exact_update UPDATE BEFORE key_name {BEGIN IF NOT (NEW.name <=> OLD.name) AND EXISTS (SELECT 1 FROM key_name WHERE id<>OLD.id AND name_hash=UNHEX(SHA2(CAST(NEW.name AS BINARY),256)) AND name=NEW.name COLLATE utf8mb4_0900_bin) THEN SIGNAL SQLSTATE '23000' SET MYSQL_ERRNO=1062, MESSAGE_TEXT='duplicate key_name.name'; END IF; END}]]
        if {$triggers ne $expected_triggers} {
          return -code error "MySQL sqlk schema validation failed: incompatible triggers"
        }

        set writer_lock [::mysql::sel $db {
          SELECT lock_id FROM sqlk_writer_lock ORDER BY lock_id
        } -flatlist]
        if {$writer_lock ne "1"} {
          return -code error "MySQL sqlk schema validation failed: invalid sqlk_writer_lock contents"
        }
      }

      ##########
      #
      #
      #
      proc schema_init {db} {
        set code [catch {
          set tables [::mysql::sel $db {
            SELECT TABLE_NAME
            FROM information_schema.TABLES
            WHERE TABLE_SCHEMA=DATABASE()
              AND TABLE_NAME IN ('sqlk_writer_lock','key_name','key_data')
            ORDER BY TABLE_NAME
          } -flatlist]

          if {[llength $tables] == 0} {
            _schema_create $db
          } elseif {$tables ne [list key_data key_name sqlk_writer_lock]} {
            return -code error "MySQL sqlk schema validation failed: incomplete schema"
          }

          _schema_validate $db
        } result options]

        if {$code} {
          catch {::mysql::close $db}
          return -options $options $result
        }
      }
    }
  }
}