(defn rollback
  [conn &opt v]
  (signal 0 [conn @[:rollback v]]))

(defn commit
  [conn &opt v]
  (signal 0 [conn @[:commit v]]))

(defn txn*
  "function form of txn"
  [conn options ftx]
  (def retry (get options :retry false))
  (def mode (get options :mode ""))
  (try
    (do
      (exec conn (string "begin " mode ";"))
      (def fb (fiber/new ftx :i0123))
      (def v (resume fb))
      (if (= (fiber/status fb) :dead)
        (do
          (exec conn "commit;")
          v)
        (if (and (= (fiber/status fb) :user0)
                 (= (get v 0) conn))
          (let [action (get-in v [0 0])
                value (get-in v [0 1])]
            (case action
              :commit
                (do
                  (exec conn "commit;")
                  value)
              :rollback
                (do
                  (exec conn "rollback;")
                  value)
              (error "misuse of txn* error")))
          (do
            (exec conn "rollback;")
            (propagate v fb)))))
  ([err f]
    (if (and
          retry
          (error? err)
          (= (result-error-field err PG_DIAG_SQLSTATE) "40001"))
      (txn* conn options ftx)
      (do
        (unless (= (status conn) CONNECTION_BAD)
          (exec conn "rollback;"))
        (propagate err f))))))