nim-datastore/datastore/threads/threadproxyds.nim

when not compileOption("threads"):
  {.error: "This module requires --threads:on compilation flag".}

import pkg/upraises

push: {.upraises: [].}

import std/tables
import std/locks

import pkg/chronos
import pkg/chronos/threadsync
import pkg/questionable
import pkg/questionable/results
import pkg/taskpools
import pkg/chronicles

import ../key
import ../query
import ../datastore
import ../backend
import ../sql/sqliteds

import ./asyncsemaphore
import ./databuffer
import ./threadresult

export threadresult

logScope:
  topics = "datastore threadproxyds"

type

  ThreadBackendKinds* = enum
    Sqlite
    # Filesystem

  ThreadBackend* = object
    case kind*: ThreadBackendKinds
    of Sqlite:
      sql*: SQLiteBackend[KeyId,DataBuffer]

  TaskCtx[T: ThreadTypes] = object
    res: ThreadResult[T]
    signal: ThreadSignalPtr
    running: bool
    cancelled: bool

  ThreadDatastore* = ref object of Datastore
    tp: Taskpool
    backend: ThreadBackend
    semaphore: AsyncSemaphore # semaphore is used for backpressure \
                              # to avoid exhausting file descriptors

var ctxLock: Lock
ctxLock.initLock()

proc setCancelled(ctx: var TaskCtx): bool =
  withLock(ctxLock):
    if ctx.running:
      return false
    else:
      ctx.cancelled = true
      return true

proc setRunning[T](ctx: ptr TaskCtx[T]): bool =
  withLock(ctxLock):
    if ctx.cancelled:
      return
    ctx.running = true
proc setDone[T](ctx: ptr TaskCtx[T]) =
  withLock(ctxLock):
    ctx.running = false

proc acquireSignal(): ?!ThreadSignalPtr =
  let signal = ThreadSignalPtr.new()
  if signal.isErr():
    failure (ref CatchableError)(msg: "failed to aquire ThreadSignalPtr: " & signal.error())
  else:
    success signal.get()

template executeTask[T](ctx: ptr TaskCtx[T], blk: untyped) =
  try:
    if not ctx.setRunning():
      return
    
    ## run backend command
    let res = `blk`
    if res.isOk():
      when T is void:
        ctx.res.ok()
      else:
        ctx.res.ok(res.get())
    else:
      ctx.res.err res.error().toThreadErr()
  except CatchableError as exc:
    trace "Unexpected exception thrown in async task", exc = exc.msg
    ctx[].res.err exc.toThreadErr()
  except Exception as exc:
    trace "Unexpected defect thrown in async task", exc = exc.msg
    ctx[].res.err exc.toThreadErr()
  finally:
    ctx.setDone()
    discard ctx[].signal.fireSync()

template dispatchTask[T](self: ThreadDatastore,
                          signal: ThreadSignalPtr,
                          blk: untyped
                        ): auto =
  var
    ctx {.inject.} = TaskCtx[T](signal: signal)
  try:
    case self.backend.kind:
    of Sqlite:
      var ds {.inject.} = self.backend.sql
      proc runTask() =
        `blk`
      runTask()

      await wait(ctx.signal)
  except CancelledError as exc:
    trace "Cancelling thread future!", exc = exc.msg
    while not ctx.setCancelled():
      warn "waiting to cancel thread future!", fn = astToStr(fn)
      await sleepAsync(10.milliseconds)
    raise exc
  finally:
    discard ctx.signal.close()
    self.semaphore.release()

proc hasTask[T, DB](ctx: ptr TaskCtx[T], ds: DB, key: KeyId) {.gcsafe.} =
  ## run backend command
  executeTask(ctx):
    has(ds, key)

method has*(self: ThreadDatastore,
            key: Key): Future[?!bool] {.async.} =
  await self.semaphore.acquire()
  without signal =? acquireSignal(), err:
    return failure err

  let key = KeyId.new key.id()
  dispatchTask[bool](self, signal):
    self.tp.spawn hasTask(addr ctx, ds, key)

proc deleteTask[T, DB](ctx: ptr TaskCtx[T], ds: DB;
                       key: KeyId) {.gcsafe.} =
  ## run backend command
  executeTask(ctx):
    delete(ds, key)

method delete*(self: ThreadDatastore,
               key: Key): Future[?!void] {.async.} =
  await self.semaphore.acquire()
  without signal =? acquireSignal(), err:
    return failure err

  let key = KeyId.new key.id()
  dispatchTask[void](self, signal):
    self.tp.spawn deleteTask(addr ctx, ds, key)

method delete*(self: ThreadDatastore,
               keys: seq[Key]): Future[?!void] {.async.} =

  for key in keys:
    if err =? (await self.delete(key)).errorOption:
      return failure err

  return success()

proc putTask[T, DB](ctx: ptr TaskCtx[T], ds: DB;
                  key: KeyId,
                  data: DataBuffer) {.gcsafe, nimcall.} =
  ## run backend command
  executeTask(ctx):
    put(ds, key, data)

method put*(self: ThreadDatastore,
            key: Key,
            data: seq[byte]): Future[?!void] {.async.} =
  await self.semaphore.acquire()
  without signal =? acquireSignal(), err:
    return failure err

  let key = KeyId.new key.id()
  let data = DataBuffer.new data
  dispatchTask[void](self, signal):
    self.tp.spawn putTask(addr ctx, ds, key, data)
  
method put*(
  self: ThreadDatastore,
  batch: seq[BatchEntry]): Future[?!void] {.async.} =

  for entry in batch:
    if err =? (await self.put(entry.key, entry.data)).errorOption:
      return failure err

  return success()

proc getTask[T, DB](ctx: ptr TaskCtx[T], ds: DB;
                 key: KeyId) {.gcsafe, nimcall.} =
  ## run backend command
  executeTask(ctx):
    get(ds, key)

method get*(self: ThreadDatastore,
            key: Key,
            ): Future[?!seq[byte]] {.async.} =
  await self.semaphore.acquire()
  without signal =? acquireSignal(), err:
    return failure err

  let key = KeyId.new key.id()
  dispatchTask[void](self, signal):
    self.tp.spawn getTask(addr ctx, ds, key)

method close*(self: ThreadDatastore): Future[?!void] {.async.} =
  await self.semaphore.closeAll()
  case self.backend.kind:
  of Sqlite:
    self.backend.sql.close()

type
  QResult = DbQueryResponse[KeyId, DataBuffer]

proc queryTask[DB](
    ctx: ptr TaskCtx[QResult],
    ds: DB,
    dq: DbQuery[KeyId]
) {.gcsafe, nimcall.} =
  ## run query command
  var qh: typeof(ds.query(dq))
  executeTask(ctx):
    qh = ds.query(dq)
    if qh.isOk():
      (?!QResult).ok(default(QResult))
    else:
      (?!QResult).err(qh.error())
  if qh.isErr():
    return

  var handle = qh.get()
  for item in handle.iter():
    executeTask(ctx):
      item

  executeTask(ctx):
    (?!QResult).err((ref QueryEndedError)(msg: "done").toThreadErr())

method query*(
  self: ThreadDatastore,
  query: Query): Future[?!QueryIter] {.async.} =

  await self.semaphore.acquire()
  without signal =? acquireSignal(), err:
    return failure err

  let dq = dbQuery(
    key= KeyId.new query.key.id(),
    value=query.value,
    limit=query.limit,
    offset=query.offset,
    sort=query.sort,
  )

  dispatchTask[DbQueryResponse[KeyId, DataBuffer]](self, signal):
    self.tp.spawn queryTask(addr ctx, ds, dq)

  var
    lock = newAsyncLock() # serialize querying under threads
    iter = QueryIter.new()

  proc next(): Future[?!QueryResponse] {.async.} =
    defer:
      if lock.locked:
        lock.release()

    trace "About to query"
    if lock.locked:
      return failure (ref DatastoreError)(msg: "Should always await query features")
    if iter.finished == true:
      return failure (ref QueryEndedError)(msg: "Calling next on a finished query!")

    await lock.acquire()

    dispatchTask[void](self, signal):
      discard ctx.signal.fireSync()

    # if ctx.res.isErr() and ctx.res.error()[0] == ErrorEnum.QueryEndedErr:
    #   iter.finished = true
    #   return
    # elif ctx.res.isErr():
    #   return err(ctx.res.error())
    # else:
    #   let qres = ctx.res.get()
    #   return ok(default)

  iter.next = next
  return success iter

proc new*(
  self: type ThreadDatastore,
  ds: Datastore,
  withLocks = static false,
  tp: Taskpool): ?!ThreadDatastore =
  doAssert tp.numThreads > 1, "ThreadDatastore requires at least 2 threads"

  success ThreadDatastore(
    tp: tp,
    ds: ds,
    withLocks: withLocks,
    queryLock: newAsyncLock(),
    semaphore: AsyncSemaphore.new(tp.numThreads - 1))