(define-values (cursor-set-range cursor-first cursor-next)
  (values
   (foreign-value "MDB_SET_RANGE" int)
   (foreign-value "MDB_FIRST" int)
   (foreign-value "MDB_NEXT" int)))

(define (database-fold proc seed dbi #!key from to< to<= limit)
  (let-location ([cursor cursor #f])
    (check-error
     'database-fold
     ((foreign-lambda
       int "mdb_cursor_open"
       nonnull-transaction database (c-pointer cursor))
      (current-transaction) dbi (location cursor)))
    (let ([complete? #f])
      (dynamic-wind
        (lambda ()
          (when complete?
            (error 'database-fold "cannot re-enter cursor walk")))
        (lambda ()
          (let loop ([op (if from cursor-set-range cursor-first)]
                     [key from] [seed seed] [fuel (or limit +inf.0)])
            (if (positive? fuel)
              (let-location ([klen size_t 0] [kptr c-pointer #f]
                             [vlen size_t 0] [vptr c-pointer #f])
                (let ([status
                       ((foreign-lambda*
                         int ([nonnull-cursor cursor] [int op]
                              [size_t k0len] [scheme-pointer k0ptr]
                              [size_t k1len] [scheme-pointer k1ptr]
                              [int inclusive]
                              [(c-pointer size_t) klen] [(c-pointer c-pointer) kptr]
                              [(c-pointer size_t) vlen] [(c-pointer c-pointer) vptr])
                         "int status;"
                         "MDB_val key0, key1, val;"
                         "key0.mv_size = k0len; key0.mv_data = k0ptr;"
                         "key1.mv_size = k1len; key1.mv_data = k1ptr;"
                         "val.mv_size = 0; val.mv_data = NULL;"
                         "status = mdb_cursor_get(cursor, &key0, &val, op);"
                         "if (status == 0 && key1.mv_data != NULL) {"
                         "  if (mdb_cmp(mdb_cursor_txn(cursor), mdb_cursor_dbi(cursor), &key0, &key1) >= inclusive) status = MDB_NOTFOUND;"
                         "}"
                         "if (status == 0) {"
                         "  *klen = key0.mv_size; *kptr = key0.mv_data;"
                         "  *vlen = val.mv_size; *vptr = val.mv_data;"
                         "}"
                         "C_return(status);")
                        cursor op
                        (blob/string-size key) key
                        (blob/string-size (or to< to<=)) (or to< to<=)
                        (if to< 0 1)
                        (location klen) (location kptr)
                        (location vlen) (location vptr))])
                  (cond
                    [(zero? status)
                     (let ([key (make-string klen)]
                           [val (make-string vlen)])
                       (move-memory! kptr key klen)
                       (move-memory! vptr val vlen)
                       (loop cursor-next key (proc key val seed) (sub1 fuel)))]
                    [(eqv? status not-found)
                     seed]
                    [else
                     (check-error 'database-ref status)])))
              seed)))
        (lambda ()
          ((foreign-lambda
            void "mdb_cursor_close"
            nonnull-cursor)
           cursor)
          (set! complete? #t))))))

(define (database-walk proc dbi . args)
  (apply
   database-fold
   (lambda (key val seed) (proc key val) seed) (void)
   dbi args))

(define (database->alist dbi #!rest args #!key from to< to<= limit)
  (fold
   (lambda (key+val rest)
     (let ([val (cdr key+val)])
       (when (pair? val) (set-cdr! key+val (reverse! val))))
     (cons key+val rest))
   '()
   (database-fold
    (if (memq #:duplicate-list args)
        (lambda (key val rest)
          (if (and (pair? rest) (equal? key (caar rest)))
              (begin
                (set-cdr! (car rest) (cons val (cdar rest)))
                rest)
              (cons (cons key (list val)) rest)))
        (lambda (key val rest)
          (cons (cons key val) rest)))
    '() dbi #:from from #:to< to< #:to<= to<= #:limit limit)))

(define (alist->database alist . args)
  (let-keys+flags open-database args
                  ([0 name #f])
                  ([database-flags dbi-flags]
                   [database-set!-flags set-flags])
    (let ([dbi (%open-database name dbi-flags)])

The transaction is committed upon normal return from <tt>(THUNK)</tt>
and aborted if <tt>(THUNK)</tt> throws an exception.

<verbatim>
  (clear-stale-transactions [ENVIRONMENT]) => INTEGER
</verbatim>

Removes stale read-only transactions from the database lockfile (stale
read-write transactions are usually removed automatically). If no
<tt>ENVIRONMENT</tt> is specified explicitly, the current environment is used.

The procedure returns the number of stale lock file entries that were removed.

<h2>Databases</h2>

with one value per key, <tt>VALUE</tt> is ignored.

Database keys and values can be strings or blobs.

<verbatim>
  (database-fold
   PROC SEED DATABASE
   [#:from START] [#:to< | #:to<= STOP] [#:limit LIMIT])
  =>
  (PROC KEY VALUE (... (PROC KEY VALUE SEED)))
</verbatim>

Folds over the records in the database. If <tt>START</tt> is given, only
fold over the records with keys greater than or equal to <tt>START</tt>.
If <tt>STOP</tt> is given, only fold over the records with keys less
than (or equal) to <tt>STOP</tt>. If <tt>LIMIT</tt> is given, process at
most <tt>LIMIT</tt> records before stopping early.

<verbatim>
  (database-walk
   PROC DATABASE
   [#:from START] [#:to< | #:to<= STOP] [#:limit LIMIT])
  =>
  VOID
</verbatim>

Like <tt>database-fold</tt>, but discarding the result.

<verbatim>
  (database->alist
   DATABASE
   [#:from START] [#:to< | #:to<= STOP] [#:limit LIMIT] [#:duplicate-list])
  =>
  ALIST
</verbatim>

Convert the records in the database to an association list. If
<tt>START</tt> is given, only list the records with keys greater than or
equal to <tt>START</tt>. If <tt>STOP</tt> is given, only list the
records with keys less than (or equal) to <tt>STOP</tt>. If
<tt>LIMIT</tt> is given, process at most <tt>LIMIT</tt> records before
stopping early.

Optionally, multiple values for the same key may be folded into lists
rather than producing multiple pairs with the same key in the
association list.

<verbatim>
  (alist->database

(import
  lmdb test)

(test-assert "environment creation"
  (database-environment?
   (open-database-environment
    "test-env" #:max-databases 4)))

        "quirks" #:create #:duplicate-sort #:append/duplicate)))

(test-assert (database? quirks))

(with-transaction
 (lambda ()
   (test
     '()
     (database->alist quirks #:limit 0))
   (test
     '(("bar" . "foo") ("baz" . "badumm") ("baz" . "kawumm"))
     (database->alist quirks #:from "b" #:to< "x"))
   (test
     '(("bar" . "foo"))
     (database->alist quirks #:from "b" #:to< "x" #:limit 1))
   (test
     '(("bar" "foo") ("baz" "badumm" "kawumm"))
     (database->alist quirks #:from "b" #:to< "x" #:duplicate-list))
   (test
     '(("bar" "foo") ("baz" "badumm"))
     (database->alist quirks #:from "b" #:to< "x" #:limit 2 #:duplicate-list))

   (database-delete! quirks "baz" "badumm")

   (test
     '(("baz" . "kawumm"))
     (database->alist quirks #:from "baz" #:to<= "baz"))

   (drop-database quirks)

   (test
     '()
     (database->alist quirks))))

(close-database-environment)

(test-exit)