# Licensed under MIT License. Copyright: 2021-2023 Alisa Feistel, PARRRATE TNV.

from __future__ import annotations

__all__ = (
    "VDELETE",
    "KVProtocol",
    "KVFactory",
    "KVJson",
    "VirtualConnection",
    "ExtendedVirtualConnection",
    "DbInterface",
    "AbstractDbConnection",
    "DbConnection",
    "DbManager",
    "DbFactory",
    "Db",
    "Transaction",
    "TransactionView",
    "FutureContext",
)

import abc
import asyncio
import concurrent.futures
import json
import os
import pathlib
import threading
from collections.abc import Hashable
from io import StringIO, UnsupportedOperation
from typing import IO, Any, Protocol, TypeAlias


class Request:
    __slots__ = ("__future",)

    def __init__(self, future: asyncio.Future | None, /) -> None:
        self.__future = future

    def waiting(self, /) -> bool:
        return self.__future is not None

    def set_result(self, result, /) -> None:
        if self.__future is not None:
            self.__future.set_result(result)

    def set_exception(self, exception, /) -> None:
        if self.__future is not None:
            self.__future.set_exception(exception)

    async def wait(self, /) -> None:
        if self.__future is not None:
            await self.__future


class LineRequest(Request):
    __slots__ = ("line",)

    def __init__(self, line: str, /, *, future: asyncio.Future | None) -> None:
        super().__init__(future)
        self.line = line


class KVProtocol(abc.ABC):
    @abc.abstractmethod
    def dbget(self, db: dict, key: Any, default: Any, /) -> Any:
        raise NotImplementedError


VDELETE = object()


class KVFactory(KVProtocol):
    """\
this class is for working with already normalised data values, not for data transformation (e.g. reducing keys to a common form).
that functionality may be added in the future, though, probably, only for custom DbConnection implementations.
note: unstable signature."""

    __slots__ = ()

    @abc.abstractmethod
    def line(self, key: Any, value: Any, /) -> str:
        """\
line must contain exactly one '\\n' at exactly the end if the line is not empty.
note: other forms of requests will later be represented by different methods or by instances of Action class."""
        raise NotImplementedError

    @abc.abstractmethod
    def fromline(self, line: str, /) -> tuple[Any, Any]:
        """\
inverse of line().
note: unstable signature."""
        raise NotImplementedError

    def run(self, line: str, db: dict, reduce: bool, /) -> None:
        """\
run request against the db.
extensible to allow forms of requests other than set.
note: unstable signature."""
        key, value = self.fromline(line)
        self._dbset(db, key, value, reduce)

    def _dbset(self, db: dict, key: Any, value: Any, reduce: bool, /) -> None:
        if reduce and value is VDELETE:
            db.pop(key, None)
        else:
            db[key] = value

    def dbset(self, db: dict, key: Any, value: Any, /) -> None:
        self._dbset(db, key, value, True)

    def dbget(self, db: dict, key: Any, default: Any, /) -> Any:
        value = db.get(key, default)
        return self.filter_value(value, default)

    def filter_value(self, value: Any, default: Any, /) -> Any:
        if value is VDELETE:
            return default
        else:
            return value

    def request(self, key: Any, value: Any, /, *, future: asyncio.Future | None) -> KVRequest:
        """\
form request with Future.
low-level API.
note: unstable signature."""
        return KVRequest(key, value, future=future, factory=self)

    def free(self, key: Any, value: Any, /) -> KVRequest:
        """\
result free from Future.
note: unstable signature."""
        return self.request(key, value, future=None)

    def io2db(self, io: IO[str], db: dict, reduce: bool, /) -> int:
        """\
note: unstable signature."""
        size = 0
        for line in io:
            self.run(line, db, reduce)
            size += len(line)
        return size

    def db2io(self, db: dict, io: IO[str], /) -> int:
        """\
note: unstable signature."""
        size = 0
        for key, value in db.items():
            size += io.write(self.line(key, value))
        return size

    def path2db_sync(self, path: pathlib.Path, db: dict, /) -> int:
        path.touch()
        with path.open("r") as file:
            return self.io2db(file, db, True)

    def db2path_sync(self, db: dict, path: pathlib.Path, /) -> int:
        with path.open("w") as file:
            initial_size = self.db2io(db, file)
            os.fsync(file.fileno())
            return initial_size


class KVRequest(LineRequest):
    __slots__ = (
        "__factory",
        "key",
        "value",
    )

    def __init__(self, key: Any, value: Any, /, *, future: asyncio.Future | None, factory: KVFactory) -> None:
        super().__init__(factory.line(key, value), future=future)
        self.__factory = factory
        self.key = key
        self.value = value


class CommitRequest(Request):
    __slots__ = ()


class UnknownRequestType(TypeError):
    __slots__ = ()


class KVJson(KVFactory):
    """note: unstable signature."""

    __slots__ = ()

    def line(self, key: Any, value: Any, /) -> str:
        if value is VDELETE:
            obj = {"key": key}
        else:
            obj = {"key": key, "value": value}
        return json.dumps(obj) + "\n"

    def _load_key(self, key: Any, /) -> Hashable:
        """note: unstable signature."""
        match key:
            case Hashable():
                return key
            case list():
                return tuple(map(self._load_key, key))
            case dict():
                return tuple((self._load_key(k), self._load_key(v)) for k, v in key.items())
            case _:
                raise TypeError("unknown KVJson key type, cannot convert to hashable")

    def fromline(self, line: str, /) -> tuple[Any, Any]:
        d = json.loads(line)
        return self._load_key(d["key"]), d.get("value", VDELETE)


class TransactionRequest(LineRequest):
    __slots__ = ("buffer",)

    def __init__(self, buffer: StringIO, /, *, future: asyncio.Future | None) -> None:
        super().__init__(buffer.getvalue(), future=future)
        self.buffer = buffer


class DbParameters:
    __slots__ = (
        "path",
        "kvfactory",
        "buffersize",
    )

    def __init__(self, path: pathlib.Path, /, *, kvfactory: KVFactory, buffersize: int) -> None:
        self.path = path
        """note: unstable signature."""
        self.kvfactory = kvfactory
        """note: unstable signature."""
        self.buffersize = buffersize
        """note: unstable signature."""


class RequestToClosedConnection(asyncio.InvalidStateError):
    pass


class VirtualConnection(abc.ABC):
    """minimal intersection of DbConnection and TransactionView functionality"""

    __slots__ = ()

    @abc.abstractmethod
    def get(self, key: Any, default: Any, /) -> Any:
        raise NotImplementedError

    @abc.abstractmethod
    def kvprotocol(self, /) -> KVProtocol:
        raise NotImplementedError

    @abc.abstractmethod
    async def commit_transaction(self, delta: dict, /) -> None:
        raise NotImplementedError

    @abc.abstractmethod
    def submit_transaction_request(self, delta: dict, future: asyncio.Future | None, /) -> None:
        raise NotImplementedError

    @abc.abstractmethod
    def loop(self, /) -> asyncio.AbstractEventLoop:
        raise NotImplementedError

    def transaction(self, /) -> Transaction:
        return Transaction(self)


class ExtendedVirtualConnection(VirtualConnection, abc.ABC):
    """maximal intersection of DbConnection and TransactionView functionality"""

    @abc.abstractmethod
    def set_nowait(self, key: Any, value: Any, /) -> None:
        raise NotImplementedError

    @abc.abstractmethod
    def submit_transaction(self, delta: dict, /) -> None:
        raise NotImplementedError

    @abc.abstractmethod
    async def commit(self, /) -> None:
        raise NotImplementedError


class DbInterface(ExtendedVirtualConnection, abc.ABC):
    @abc.abstractmethod
    async def set(self, key: Any, value: Any, /) -> None:
        raise NotImplementedError


class AbstractDbConnection(Protocol):
    def get(self, key: Any, default: Any, /) -> Any:
        """this method is instant."""
        raise NotImplementedError

    async def set(self, key: Any, value: Any, /) -> None:
        """\
this method may take time to run.
ordering may not be guaranteed (depends on event loop implementation)."""
        raise NotImplementedError

    def set_nowait(self, key: Any, value: Any, /) -> None:
        """\
this method is instant.
ordering is guaranteed."""
        raise NotImplementedError

    async def commit(self, /) -> None:
        """\
this method may take time to run.
respects the ordering of previously called :meth:`~ptvp35.AbstractDbConnection.set_nowait` methods.
will, depending on event loop implementation, also execute later changes."""
        raise NotImplementedError

    def transaction(self, /) -> Transaction:
        raise NotImplementedError


class _Loop:
    __slots__ = ("__loop",)

    def __init__(self, loop: asyncio.AbstractEventLoop, /) -> None:
        self.__loop = loop

    def create_future(self, /) -> asyncio.Future:
        return self.__loop.create_future()

    def loop(self, /) -> asyncio.AbstractEventLoop:
        return self.__loop

    def run_in_thread(self, name: str, fn, /, *args, **kwargs) -> asyncio.Future:
        """\
we are using our own thread to guarantee as much of autonomy and control as possible.
intended for heavy tasks."""
        future = self.create_future()

        def wrap() -> None:
            try:
                result = fn(*args, **kwargs)
            except Exception as exception:
                self.__loop.call_soon_threadsafe(future.set_exception, exception)
            else:
                self.__loop.call_soon_threadsafe(future.set_result, result)

        fname = getattr(fn, "__name__", "?")
        threading.Thread(target=wrap, name=f"persistence5-{name}-{fname}").start()

        return future


class _Errors:
    __slots__ = (
        "__path",
        "__loop",
        "__event_loop",
    )

    def __init__(self, path: pathlib.Path, loop: _Loop, /) -> None:
        self.__path = path.with_name(path.name + ".error")
        self.__loop = loop
        self.__event_loop = loop.loop()

    def _save_sync(self, line: str, /) -> None:
        with self.__path.open("a") as file:
            file.write(line.strip() + "\n")

    async def _save(self, line: str, /) -> None:
        await self.__event_loop.run_in_executor(None, self._save_sync, line)

    def _schedule(self, line: str, /) -> concurrent.futures.Future:
        return asyncio.run_coroutine_threadsafe(self._save(line), self.__event_loop)

    def save_from_thread(self, line: str, /) -> None:
        self._schedule(line).result()


class _File:
    __slots__ = (
        "__path",
        "__file",
    )

    __file: IO[str]

    def __init__(self, path: pathlib.Path, /) -> None:
        self.__path = path

    def path(self, /) -> pathlib.Path:
        return self.__path

    def tell(self, /) -> int:
        return self.__file.tell()

    def write_to_disk_sync(self, line: str, /) -> None:
        self.__file.write(line)
        self.__file.flush()
        try:
            os.fsync(self.__file.fileno())
        except UnsupportedOperation:
            pass

    def open_sync(self, /) -> None:
        self.__file = self.__path.open("a")

    def close_sync(self, /) -> None:
        self.__file.close()
        del self.__file


class _Backup:
    __slots__ = (
        "__file",
        "__kvfactory",
        "__loop",
        "__path",
        "__backup",
        "__recover",
        "__initial_size",
    )

    __initial_size: int

    def __init__(self, path: pathlib.Path, kvfactory: KVFactory, loop: _Loop, /) -> None:
        self.__file = _File(path)
        self.__kvfactory = kvfactory
        self.__loop = loop
        self.__path = path
        self.__backup = path.with_name(path.name + ".backup")
        self.__recover = path.with_name(path.name + ".recover")

    def file(self, /) -> _File:
        return self.__file

    def kvfactory(self, /) -> KVFactory:
        return self.__kvfactory

    def _copy_sync(self, db: dict | None, /) -> None:
        if db is None:
            db = {}
            self.__kvfactory.path2db_sync(self.__backup, db)
        self.__kvfactory.db2path_sync(db, self.__path)

    def _recovery_unset_sync(self, /) -> None:
        self.__recover.unlink()
        self.__backup.unlink()

    def _finish_recovery_sync(self, db: dict | None, /) -> None:
        self._copy_sync(db)
        self._recovery_unset_sync()

    def _recovery_set_sync(self, db: dict, /) -> None:
        self.__initial_size = self.__kvfactory.db2path_sync(db, self.__backup)
        self.__recover.touch()

    def build_file_sync(self, db: dict, /) -> None:
        self._recovery_set_sync(db)
        self._finish_recovery_sync(db)

    def _rebuild_file_sync(self, db: dict, /) -> None:
        if self.__recover.exists():
            self._finish_recovery_sync(None)
        self.__kvfactory.path2db_sync(self.__path, db)
        self.build_file_sync(db)

    def _reload_sync(self, /) -> None:
        self.__file.close_sync()
        self._rebuild_file_sync({})
        self.__file.open_sync()

    def run_in_thread(self, fn, /, *args, **kwargs) -> asyncio.Future:
        return self.__loop.run_in_thread(self.__path.name, fn, *args, **kwargs)

    async def _reload(self, /) -> None:
        await self.run_in_thread(self._reload_sync)

    async def reload_if_oversized(self, /) -> None:
        if self.__file.tell() > 2 * self.__initial_size:
            await self._reload()

    def load_mmdb_sync(self, /) -> dict:
        db = {}
        self._rebuild_file_sync(db)
        return db

    def uninitialize(self, /) -> None:
        del self.__initial_size


class _Guard:
    __slots__ = (
        "__backup",
        "__error",
        "__file",
        "__path",
        "__truncate",
        "__flag",
    )

    def __init__(self, backup: _Backup, error: _Errors, /) -> None:
        self.__backup = backup
        self.__error = error
        self.__file = backup.file()
        self.__path = path = self.__file.path()
        self.__truncate = path.with_name(path.name + ".truncate")
        self.__flag = path.with_name(path.name + ".truncate_flag")

    def backup(self, /) -> _Backup:
        return self.__backup

    def _write_bytes_sync(self, s: bytes, /) -> None:
        # consider subclassing/rewriting to use `os.fsync`
        self.__truncate.write_bytes(s)

    def _write_value_sync(self, value: int, /) -> None:
        self._write_bytes_sync(value.to_bytes(16, "little"))

    def _read_bytes_sync(self, /) -> bytes:
        return self.__truncate.read_bytes()

    def _read_value_sync(self, /) -> int:
        return int.from_bytes(self._read_bytes_sync(), "little")

    def _set_sync(self, /) -> None:
        self._write_value_sync(self.__file.tell())
        self.__flag.touch()

    def _unset_sync(self, /) -> None:
        self.__flag.unlink(missing_ok=True)
        self.__truncate.unlink(missing_ok=True)

    def _truncate_sync(self, /) -> None:
        with self.__path.open("r+") as file:
            self._file_truncate_sync(file, self._read_value_sync())

    def assure_sync(self, /) -> None:
        if self.__flag.exists():
            self._truncate_sync()
        self._unset_sync()

    def _file_truncate_sync(self, file: IO[str], pos: int, /) -> None:
        file.seek(pos)
        self.__error.save_from_thread(file.read())
        file.truncate(pos)

    def file_write_sync(self, line: str, /) -> None:
        self._set_sync()
        self.__file.write_to_disk_sync(line)
        self._unset_sync()


class _ReceivingQueue:
    __all__ = ("__queue",)

    def __init__(self, queue: asyncio.Queue[Request], /) -> None:
        self.__queue: asyncio.Queue[Request] = queue

    def submit(self, request: Request, /) -> None:
        self.__queue.put_nowait(request)


class _WriteableBuffer:
    __slots__ = (
        "__buffersize",
        "__guard",
        "__queue",
        "__backup",
        "__kvfactory",
        "__loop",
        "__event_loop",
        "__buffer",
        "__buffer_future",
        "__buffer_requested",
    )

    __buffer: StringIO
    __buffer_future: asyncio.Future
    __buffer_requested: bool

    def __init__(self, buffersize: int, guard: _Guard, queue: _ReceivingQueue, loop: _Loop, /) -> None:
        self.__buffersize = buffersize
        self.__guard = guard
        self.__queue = queue
        self.__backup = self.__guard.backup()
        self.__kvfactory = self.__backup.kvfactory()
        self.__loop = loop
        self.__event_loop = self.__loop.loop()
        self._clear()

    def writeable(self, /) -> _Guard:
        return self.__guard

    def loop(self, /) -> _Loop:
        return self.__loop

    def _compressed(self, /) -> StringIO:
        self.__buffer.seek(0)
        bufferdb = {}
        self.__kvfactory.io2db(self.__buffer, bufferdb, False)
        buffer = StringIO()
        self.__kvfactory.db2io(bufferdb, buffer)
        return buffer

    def _commit_compressed_sync(self, /) -> None:
        self.__guard.file_write_sync(self._compressed().getvalue())

    async def _commit_compressed(self, /) -> None:
        await self.__event_loop.run_in_executor(None, self._commit_compressed_sync)

    def _clear(self, /) -> None:
        self.__buffer = StringIO()
        self.__buffer_future = self.__loop.create_future()
        self.__buffer_requested = False

    def _satisfy_future(self, /) -> None:
        self.__buffer_future.set_result(None)
        self._clear()

    def _fail_future(self, exception: BaseException, /) -> None:
        self.__buffer_future.set_exception(exception)
        self._clear()

    async def _do_commit_buffer(self, /) -> None:
        try:
            await self._commit_compressed()
        except Exception as e:
            self._fail_future(e)
        else:
            self._satisfy_future()

    def _request_buffer(self, request: Request, /) -> None:
        if request.waiting():

            def callback(bf: asyncio.Future) -> None:
                if (e := bf.exception()) is not None:
                    request.set_exception(e)
                else:
                    request.set_result(None)

            self.__buffer_future.add_done_callback(callback)
        if not self.__buffer_requested:
            self.__buffer_requested = True
            self.__queue.submit(CommitRequest(None))

    async def _commit(self, /) -> None:
        if self.__buffer.tell():
            await self._do_commit_buffer()
            await self.__backup.reload_if_oversized()
        elif self.__buffer_requested:
            self._satisfy_future()

    async def _commit_or_request_so(self, request: Request, /) -> None:
        if self.__buffer.tell() >= self.__buffersize:
            await self._commit()
            request.set_result(None)
        else:
            self._request_buffer(request)

    async def _write(self, line: str, request: Request, /) -> None:
        self.__buffer.write(line)
        await self._commit_or_request_so(request)

    async def _handle_request(self, request: Request, /) -> None:
        match request:
            case LineRequest():
                await self._write(request.line, request)
            case CommitRequest():
                await self._commit()
                request.set_result(None)
            case _:
                raise UnknownRequestType

    async def _close(self, /) -> None:
        await self._commit()
        if not self.__buffer_future.done():
            self.__buffer_future.set_exception(RequestToClosedConnection())
            if not isinstance(self.__buffer_future.exception(), RequestToClosedConnection):
                raise RuntimeError
        del self.__buffer_requested
        del self.__buffer_future
        del self.__buffer


class _Memory:
    __slots__ = (
        "__backup",
        "__guard",
        "__file",
        "__kvfactory",
        "__loop",
        "__mmdb",
    )

    __mmdb: dict

    def __init__(self, guard: _Guard, /) -> None:
        self.__guard = guard
        self.__backup = guard.backup()
        self.__file = self.__backup.file()
        self.__kvfactory = self.__backup.kvfactory()

    def _initialize_sync(self, /) -> None:
        self.__mmdb = self.__backup.load_mmdb_sync()

    def _load_from_file_sync(self, /) -> None:
        self.__guard.assure_sync()
        self._initialize_sync()
        self.__file.open_sync()

    async def _load_from_file(self, /) -> None:
        await self.__backup.run_in_thread(self._load_from_file_sync)

    def _close_sync(self, /) -> None:
        self.__file.close_sync()
        self.__backup.build_file_sync(self.__mmdb)
        del self.__mmdb
        self.__backup.uninitialize()

    async def _close(self, /) -> None:
        await self.__backup.run_in_thread(self._close_sync)

    def _transaction_buffer(self, delta: dict, /) -> StringIO:
        buffer = StringIO()
        self.__kvfactory.db2io(delta, buffer)
        for key, value in delta.items():
            self.__kvfactory.dbset(self.__mmdb, key, value)
        return buffer

    def get(self, key: Any, default: Any, /) -> Any:
        return self.__kvfactory.dbget(self.__mmdb, key, default)

    def set(self, key: Any, value: Any, /) -> None:
        self.__kvfactory.dbset(self.__mmdb, key, value)


class _QueueTask:
    __slots__ = (
        "__queue",
        "__buffer",
        "__event_loop",
        "__task",
    )

    def __init__(self, queue: asyncio.Queue[Request], buffer: _WriteableBuffer, /) -> None:
        self.__queue = queue
        self.__buffer = buffer
        self.__event_loop = buffer.loop().loop()

    async def _background_cycle(self, /) -> None:
        request: Request = await self.__queue.get()
        try:
            await self.__buffer._handle_request(request)
        except Exception as e:
            request.set_exception(e)
        finally:
            self.__queue.task_done()

    async def _background_task(self, /) -> None:
        while True:
            await self._background_cycle()

    async def close(self, /) -> None:
        if not self.__task.done():
            await self.__queue.join()
            self.__task.cancel()
            del self.__task
        if not self.__queue.empty():
            raise RuntimeError
        del self.__queue
        await self.__buffer._close()

    def start(self, /) -> None:
        self.__task = self.__event_loop.create_task(self._background_task())


class _DbConnection(DbInterface):
    """note: unstable constructor signature."""

    __slots__ = (
        "__kvfactory",
        "__buffersize",
        "__path",
        "__error",
        "__not_running",
        "__mmdb",
        "__loop",
        "__queue",
        "__file",
        "__task",
    )

    __mmdb: _Memory
    __loop: _Loop
    __queue: _ReceivingQueue
    __task: _QueueTask

    def __init__(self, parameters: DbParameters, /) -> None:
        self.__kvfactory = parameters.kvfactory
        self.__buffersize = parameters.buffersize
        self.__path = parameters.path
        self.__running = False

    def kvprotocol(self, /) -> KVProtocol:
        return self.__kvfactory

    def get(self, key: Any, default: Any, /) -> Any:
        """dict-like get with mandatory default parameter."""
        return self.__mmdb.get(key, default)

    async def set(self, key: Any, value: Any, /) -> None:
        """set the value and wait until it's written to disk."""
        future = self.__loop.create_future()
        request = self.__kvfactory.request(key, value, future=future)
        self.__mmdb.set(key, value)
        self.__queue.submit(request)
        await future

    def set_nowait(self, key: Any, value: Any, /) -> None:
        """set value and add write-to-disk request to queue."""
        request = self.__kvfactory.free(key, value)
        self.__mmdb.set(key, value)
        self.__queue.submit(request)

    async def _initialize_running(self, /) -> None:
        self.__loop = _Loop(asyncio.get_running_loop())
        guard = _Guard(
            _Backup(self.__path, self.__kvfactory, self.__loop),
            _Errors(self.__path, self.__loop),
        )
        queue: asyncio.Queue[Request] = asyncio.Queue()
        self.__queue = _ReceivingQueue(queue)
        self.__mmdb = _Memory(guard)
        await self.__mmdb._load_from_file()
        self.__task = _QueueTask(queue, _WriteableBuffer(self.__buffersize, guard, self.__queue, self.__loop))
        self.__task.start()

    async def _initialize(self, /) -> None:
        if self.__running:
            raise RuntimeError
        self.__running = True
        await self._initialize_running()

    @classmethod
    async def create(cls, parameters: DbParameters, /) -> _DbConnection:
        """\
connect to the factory.
note: unstable signature."""
        dbconnection = _DbConnection(parameters)
        await dbconnection._initialize()
        return dbconnection

    async def _close_running(self, /) -> None:
        mmdb = self.__mmdb
        del self.__mmdb
        del self.__queue
        del self.__loop
        await self.__task.close()
        del self.__task
        await mmdb._close()

    async def aclose(self, /) -> None:
        """\
close the connection.
note: unstable signature."""
        await self._close_running()
        self.__running = False

    async def commit_transaction(self, delta: dict, /) -> None:
        """\
hybrid of set() and dict.update().
note: unstable signature."""
        if not delta:
            return
        buffer = self.__mmdb._transaction_buffer(delta)
        future = self.__loop.create_future()
        self.__queue.submit(TransactionRequest(buffer, future=future))
        await future

    def submit_transaction(self, delta: dict, /) -> None:
        """\
hybrid of set_nowait() and dict.update().
_nowait analogue of commit_transaction().
note: this method was added only for async-sync symmetry with commit_transaction().
note: unstable signature."""
        if not delta:
            return
        buffer = self.__mmdb._transaction_buffer(delta)
        self.__queue.submit(TransactionRequest(buffer, future=None))

    def submit_transaction_request(self, delta: dict, future: asyncio.Future | None, /) -> None:
        """\
low-level API.
for high-level synchronisation use transaction() instead.
note: unstable signature."""
        if not delta:
            if future:
                future.set_result(None)
            return
        buffer = self.__mmdb._transaction_buffer(delta)
        self.__queue.submit(TransactionRequest(buffer, future=future))

    async def commit(self, /) -> None:
        """wait until all requests queued before are completed."""
        future = self.__loop.create_future()
        self.__queue.submit(CommitRequest(future))
        await future

    def loop(self, /) -> asyncio.AbstractEventLoop:
        return self.__loop.loop()


DbConnection: TypeAlias = DbInterface


class DbManager:
    __slots__ = (
        "__parameters",
        "__db",
    )

    def __init__(self, path: pathlib.Path, /, *, kvfactory: KVFactory, buffersize=1048576) -> None:
        self.__parameters = DbParameters(path, kvfactory=kvfactory, buffersize=buffersize)

    async def __aenter__(self) -> DbInterface:
        self.__db = await _DbConnection.create(self.__parameters)
        return self.__db

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        await self.__db.aclose()


DbFactory: TypeAlias = DbManager


class Db(_DbConnection):
    """simplified usecase combining the factory and the connection in one class."""

    __slots__ = ()

    def __init__(self, path: str | pathlib.Path, /, *, kvfactory: KVFactory, buffersize=1048576) -> None:
        _DbConnection.__init__(self, DbParameters(pathlib.Path(path), kvfactory=kvfactory, buffersize=buffersize))

    async def __aenter__(self) -> _DbConnection:
        await self._initialize()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        await self.aclose()


class FutureContext:
    def __init__(self, future: asyncio.Future | None) -> None:
        self.__future = future

    async def __aenter__(self) -> None:
        pass

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        await self.wait()

    async def wait(self, /) -> None:
        if self.__future is not None:
            await self.__future


class TransactionView(ExtendedVirtualConnection):
    """note: unstable constructor signature."""

    __slots__ = (
        "__delta",
        "__shadow",
        "__connection",
        "__loop",
        "__kvprotocol",
        "__subfuture",
    )

    def __init__(self, delta: dict, connection: VirtualConnection, /) -> None:
        self.__delta = delta
        self.__shadow = {}
        self.__connection = connection
        self.__loop = connection.loop()
        self.__kvprotocol = connection.kvprotocol()
        self.__subfuture: asyncio.Future | None = None

    def future_context(self, /) -> FutureContext:
        """do something (inside of async with), then wait for submitted changes to be committed."""
        return FutureContext(self.__subfuture)

    def rollback(self, /) -> None:
        """clear unsubmitted changes."""
        self.__delta.clear()

    def illuminate(self, /) -> None:
        """clear submitted changes, thus syncing the view (underlying the delta) with the connection."""
        self.__shadow.clear()

    async def ailluminate(self, /) -> None:
        """illuminate, then wait for submitted changes to be committed."""
        async with self.future_context():
            self.illuminate()

    def fork(self, /) -> None:
        """keep delta, but forget about the shadow entirely (including making sure it's committed)."""
        self.illuminate()
        self.__subfuture = None

    async def afork(self, /) -> None:
        """fork, then wait for submitted changes to be committed."""
        async with self.future_context():
            self.fork()

    def clear(self, /) -> None:
        """clear all changes (including the shadow)."""
        self.rollback()
        self.illuminate()

    async def aclear(self, /) -> None:
        """clear, then wait for submitted changes to be committed."""
        async with self.future_context():
            self.clear()

    def reset(self, /) -> None:
        """reset transaction."""
        self.clear()
        self.__subfuture = None

    async def areset(self, /) -> None:
        """reset, then wait for submitted changes to be committed."""
        async with self.future_context():
            self.reset()

    def kvprotocol(self, /) -> KVProtocol:
        return self.__kvprotocol

    def get(self, key: Any, default: Any, /) -> Any:
        """get from the delta (unsubmitted), else from the shadow (submitted), else from the connection."""
        if key in self.__delta:
            return self.__kvprotocol.dbget(self.__delta, key, default)
        elif key in self.__shadow:
            return self.__kvprotocol.dbget(self.__shadow, key, default)
        else:
            return self.__connection.get(key, default)

    def set_nowait(self, key: Any, value: Any, /) -> None:
        """note: unlike the corresponding db method, this one does not catch serialisation errors early."""
        self.__delta[key] = value

    def _delta(self, /) -> dict:
        delta = self.__delta.copy()
        self.__shadow |= delta
        self.__delta.clear()
        return delta

    async def commit(self, /) -> None:
        """bulk analogue of DbConnection.set()."""
        # for persistence5('s forks) developers:
        # q: why not self.__subfuture = None here?
        # a: run two commit calls concurrently. one will quit early and fail semantically.
        #    we also never implicitly reset self.__subfuture because newly created future may depend on it.
        # q: why not self.submit() inside FC block?
        # a: that would require using FC block later once more + that future may include extra submitted changes;
        #    so one would need to do submit, then do an empty FC block. that maybe introduced in the future
        # q: why use if delta?
        # a: to have code symmetric to that of submit + to not create an extra coroutine.
        # note: q&a comments above may become obsolete
        async with self.future_context():
            delta = self._delta()
            if delta:
                await self.__connection.commit_transaction(delta)

    def submit(self, /) -> None:
        """\
submit changes.
_nowait analogue of commit().
bulk analogue of DbConnection.set_nowait()."""
        # for persistence5('s forks) developers:
        # q: why use if delta?
        # a: to have code symmetric to that of commit + to not create an extra future.
        # note: q&a comments above may become obsolete
        delta = self._delta()
        if delta:
            future = self.__loop.create_future()
            self.__connection.submit_transaction_request(delta, future)
            self.__subfuture = self._gather(self.__subfuture, future)

    def _do_gather(self, left: asyncio.Future, right: asyncio.Future) -> asyncio.Future:
        future = self.__loop.create_future()

        def rcallback(fr: asyncio.Future) -> None:
            if (e := fr.exception()) is not None:
                future.set_exception(e)
            else:
                future.set_result(None)

        def lcallback(fl: asyncio.Future) -> None:
            if (e := fl.exception()) is not None:
                future.set_exception(e)
            else:
                right.add_done_callback(rcallback)

        left.add_done_callback(lcallback)

        return future

    def _reduce_future(self, future: asyncio.Future | None) -> asyncio.Future | None:
        if future is None or future.done() and future.exception() is None:
            return None
        else:
            return future

    def _gather(self, left: asyncio.Future | None, right: asyncio.Future | None) -> asyncio.Future | None:
        match (self._reduce_future(left), self._reduce_future(right)):
            case None, ofr:
                return ofr
            case ofl, None:
                return ofl
            case asyncio.Future() as fl, asyncio.Future() as fr:
                return self._do_gather(fl, fr)
            case _:
                raise TypeError

    async def commit_transaction(self, delta: dict, /) -> None:
        if not delta:
            return
        self.__delta.update(delta)
        await self.commit()

    def submit_transaction(self, delta: dict, /) -> None:
        if not delta:
            return
        self.__delta.update(delta)
        self.submit()

    def submit_transaction_request(self, delta: dict, future: asyncio.Future | None, /) -> None:
        def set_result(sf: asyncio.Future | None):
            if future is None:
                pass
            elif sf is None or (e := sf.exception()) is None:
                future.set_result(None)
            else:
                future.set_exception(e)

        if not delta:
            set_result(None)
            return
        self.submit_transaction(delta)
        if self.__subfuture is None:
            set_result(None)
            return
        self.__subfuture.add_done_callback(set_result)

    def loop(self, /) -> asyncio.AbstractEventLoop:
        return self.__loop


class Transaction:
    """note: unstable signature."""

    __slots__ = (
        "__connection",
        "__view",
        "__running",
    )

    __view: TransactionView

    def __init__(self, connection: VirtualConnection, /) -> None:
        self.__connection = connection
        self.__running = False

    async def __aenter__(self) -> TransactionView:
        return self.__enter__()

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        if exc_type is None:
            await self.__view.commit()
        else:
            self.__view.rollback()
        self._clean()

    def _clean(self, /) -> None:
        del self.__view
        self.__running = False

    def __enter__(self) -> TransactionView:
        if self.__running:
            raise RuntimeError
        self.__running = True
        self.__view = TransactionView({}, self.__connection)
        return self.__view

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        if exc_type is None:
            self.__view.submit()
        else:
            self.__view.rollback()
        self._clean()