Implement kaya-rbcs: simplified RBCS clone on the Kaya framework

- KayaApp with GET/PUT recursive routes backed by memcached (aiomcache)
- env-based configuration, metadata-aware value encoding, key prefix/digest
- Granian RSGI launcher (rbcs-server / python -m kaya_rbcs)
- pinned requirements via pip-compile, alpine:3.24 Dockerfile with healthcheck
- offline tests using an in-process fake memcached
- Gitea Actions workflow to build and push the multi-arch Docker image

.dockerignore

@@ -0,0 +1,12 @@
+.git
+.gitignore
+.venv
+**/__pycache__
+*.pyc
+.mypy_cache
+build
+dist
+*.egg-info
+tests
+Dockerfile
+.dockerignore

.gitea/workflows/build.yaml

@@ -0,0 +1,38 @@
+name: CI
+on:
+  push:
+    tags:
+      - 'release/*'
+
+jobs:
+  build_and_push_docker_image:
+    name: "Build and push docker image"
+    runs-on: hostinger
+    steps:
+      - name: Checkout sources
+        uses: actions/checkout@v6
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+      - name: Login to Gitea container registry
+        uses: docker/login-action@v3
+        with:
+          registry: gitea.woggioni.net
+          username: woggioni-opencode-agent
+          password: ${{ secrets.PUBLISHER_TOKEN }}
+      - name: Extract metadata for image tags
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: gitea.woggioni.net/woggioni-opencode-agent/kaya-rbcs
+          flavor: latest=false
+          tags: |
+            type=match,pattern=release/(.*),group=1
+            type=raw,value=latest
+      - name: Build and push docker image
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          platforms: linux/amd64,linux/arm64
+          push: true
+          pull: true
+          tags: ${{ steps.meta.outputs.tags }}

.gitignore

@@ -0,0 +1,7 @@
+.venv/
+__pycache__/
+*.pyc
+.mypy_cache/
+build/
+dist/
+*.egg-info/

Dockerfile

@@ -0,0 +1,37 @@
+FROM alpine:3.24
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_DISABLE_PIP_VERSION_CHECK=1 \
+    PIP_NO_CACHE_DIR=1 \
+    RBCS_HOST=0.0.0.0 \
+    RBCS_PORT=8080 \
+    RBCS_MEMCACHE_HOST=memcached \
+    RBCS_MEMCACHE_PORT=11211
+
+RUN apk add --no-cache python3
+
+WORKDIR /opt/kaya-rbcs
+
+COPY requirements.txt ./
+RUN python3 -m venv /opt/venv \
+    && /opt/venv/bin/pip install -r requirements.txt
+
+COPY pyproject.toml README.md ./
+COPY src ./src
+RUN /opt/venv/bin/pip install --no-deps .
+
+RUN addgroup -S rbcs \
+    && adduser -S -G rbcs rbcs \
+    && chown -R rbcs:rbcs /opt/kaya-rbcs /opt/venv
+
+ENV PATH="/opt/venv/bin:${PATH}"
+
+USER rbcs
+
+EXPOSE 8080
+
+HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
+    CMD python3 -c "import socket, os; socket.create_connection(('127.0.0.1', int(os.environ['RBCS_PORT'])), timeout=3).close()"
+
+ENTRYPOINT ["rbcs-server"]

README.md

@@ -0,0 +1,87 @@
+# kaya-rbcs
+
+A simplified Python clone of [RBCS](https://gitea.woggioni.net/woggioni/rbcs)
+(Remote Build Cache Server) built on the [Kaya](https://gitea.woggioni.net/woggioni/kaya)
+web framework.
+
+It implements only the two core cache endpoints — `GET` and `PUT` — persisting
+values to memcached. Authentication, RBAC, rate limiting, TRACE health checks,
+compression and TLS are intentionally left out.
+
+## Behavior
+
+The cache key is the request path relative to a configurable URL prefix
+(default `/`), normalized (`..` segments are resolved). Values are stored in
+memcached together with the `Content-Type` and `Content-Disposition` metadata
+sent on upload, mirroring RBCS' `CacheValueMetadata`.
+
+| Request                | Response                                                        |
+|------------------------|-----------------------------------------------------------------|
+| `PUT /<key>`           | `201 Created`, body = key, `Content-Type: text/plain`           |
+| `GET /<key>` (hit)     | `200 OK`, body = value, stored `Content-Type` / `Content-Disposition` |
+| `GET /<key>` (miss)    | `404 Not Found`, empty body                                     |
+| `GET/PUT` outside prefix | `400 Bad Request`                                            |
+
+## Installation
+
+```bash
+pip install -e .            # + kaya-core, aiomcache
+pip install -e .[server]    # additionally installs granian + kaya-rsgi
+```
+
+Requires Python 3.10+ and a running memcached.
+
+## Running
+
+Start the server with the `rbcs-server` console script or `python -m kaya_rbcs`:
+
+```bash
+rbcs-server
+```
+
+Configuration is read from environment variables:
+
+| Variable                | Default       | Description                         |
+|-------------------------|---------------|-------------------------------------|
+| `RBCS_HOST`             | `127.0.0.1`   | Bind address                        |
+| `RBCS_PORT`             | `8080`        | Bind port                           |
+| `RBCS_PATH_PREFIX`      | `/`           | URL prefix that maps to the cache   |
+| `RBCS_MEMCACHE_HOST`    | `127.0.0.1`   | Memcached host                      |
+| `RBCS_MEMCACHE_PORT`    | `11211`       | Memcached port                      |
+| `RBCS_KEY_PREFIX`       | *(unset)*     | String appended to each cache key   |
+| `RBCS_DIGEST`           | *(unset)*     | Hash the key with this algorithm (e.g. `sha256`) |
+| `RBCS_MAX_AGE`          | `7d`          | Value TTL (`s`/`m`/`h`/`d`)         |
+
+The app is also a plain ASGI application, so it can be served by any ASGI
+server, e.g. `daphne kaya_rbcs.app:app`.
+
+### Usage example
+
+```bash
+curl -X PUT -H 'Content-Type: application/octet-stream' \
+     --data-binary @build-output.tar http://localhost:8080/my-module/build
+curl http://localhost:8080/my-module/build -o build-output.tar
+```
+
+## Tests
+
+```bash
+python -m unittest discover -s tests
+```
+
+Tests use an in-process fake memcached (text protocol), so they run offline
+with no external dependencies.
+
+## Layout
+
+```
+src/kaya_rbcs/
+├── __init__.py    # package exports + `rbcs-server` entry point
+├── __main__.py    # Granian (RSGI) launcher
+├── config.py      # env-based configuration
+├── store.py       # MemcacheStore: value+metadata encoding, key processing
+└── app.py         # create_app(): KayaApp with GET/PUT recursive routes
+tests/
+├── fake_memcached.py
+└── test_app.py
+```

pyproject.toml

@@ -0,0 +1,53 @@
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "kaya-rbcs"
+dynamic = ["version"]
+description = "Simplified clone of RBCS (Remote Build Cache Server) built on the Kaya web framework"
+readme = "README.md"
+requires-python = ">=3.10"
+license = { text = "MIT" }
+classifiers = [
+    'Development Status :: 3 - Alpha',
+    'Topic :: Utilities',
+    'Intended Audience :: Developers',
+    'Environment :: Console',
+    'Programming Language :: Python :: 3',
+]
+
+dependencies = [
+    "kaya-core",
+    "aiomcache",
+]
+
+[project.optional-dependencies]
+server = [
+    "kaya-rsgi",
+    "granian",
+]
+dev = [
+    "httpx",
+    "mypy",
+]
+
+[project.scripts]
+rbcs-server = "kaya_rbcs:main"
+
+[tool.setuptools.packages.find]
+where = ["src"]
+namespaces = true
+
+[tool.mypy]
+python_version = "3.12"
+disallow_untyped_defs = true
+show_error_codes = true
+no_implicit_optional = true
+warn_return_any = true
+warn_unused_ignores = true
+strict = true
+exclude = ["tests"]
+
+[tool.setuptools.dynamic]
+version = { attr = "kaya_rbcs.__version__" }

requirements.in

@@ -0,0 +1,6 @@
+# runtime dependencies of kaya-rbcs
+# (see pyproject.toml: [project] dependencies + [project.optional-dependencies] server)
+kaya-core
+aiomcache
+kaya-rsgi
+granian

requirements.txt

@@ -0,0 +1,31 @@
+#
+# This file is autogenerated by pip-compile with Python 3.14
+# by the following command:
+#
+#    pip-compile --extra-index-url=https://pypi.org/simple --index-url=https://gitea.woggioni.net/api/packages/woggioni/pypi/simple --output-file=requirements.txt requirements.in
+#
+--index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple
+--extra-index-url https://pypi.org/simple
+
+aiomcache==0.8.2
+    # via -r requirements.in
+click==8.4.2
+    # via granian
+granian==2.7.9
+    # via
+    #   -r requirements.in
+    #   kaya-rsgi
+kaya-core==0.0.1
+    # via
+    #   -r requirements.in
+    #   kaya-rsgi
+kaya-rsgi==0.0.1
+    # via -r requirements.in
+pwo==0.1.2
+    # via
+    #   kaya-core
+    #   kaya-rsgi
+typing-extensions==4.16.0
+    # via
+    #   kaya-core
+    #   pwo

src/kaya_rbcs/__init__.py

@@ -0,0 +1,20 @@
+from .app import app, create_app
+from .config import Config
+from .store import CacheValueMetadata, MemcacheStore
+
+__version__ = '0.1.0'
+
+
+def main() -> None:
+    from kaya_rbcs.__main__ import main as _main
+
+    _main()
+
+
+__all__ = [
+    'app',
+    'create_app',
+    'Config',
+    'CacheValueMetadata',
+    'MemcacheStore',
+]

src/kaya_rbcs/__main__.py

@@ -0,0 +1,32 @@
+import sys
+from typing import List, NoReturn
+
+from .config import Config
+
+
+def main() -> NoReturn:
+    try:
+        from granian.cli import entrypoint
+    except ImportError as exc:
+        raise SystemExit(
+            "Granian is required to run the server, "
+            "install it with 'pip install kaya-rbcs[server]'"
+        ) from exc
+
+    config = Config.from_env()
+    args: List[str] = [
+        '--interface',
+        'rsgi',
+        '--host',
+        config.host,
+        '--port',
+        str(config.port),
+        'kaya_rbcs.app:app',
+    ]
+    sys.argv = [sys.argv[0], *args]
+    entrypoint()  # type: ignore[no-untyped-call]
+    raise SystemExit(0)
+
+
+if __name__ == '__main__':
+    main()

src/kaya_rbcs/app.py

@@ -0,0 +1,108 @@
+from logging import getLogger
+from posixpath import normpath
+from typing import List, Mapping, Optional, Sequence
+
+from kaya.core import HttpContext, KayaApp
+
+from .config import Config
+from .store import CacheValueMetadata, MemcacheStore
+
+log = getLogger(__name__)
+
+_APPLICATION_OCTET_STREAM = 'application/octet-stream'
+_TEXT_PLAIN = 'text/plain'
+
+
+def _normalize_prefix(prefix: str) -> str:
+    stripped = prefix.strip('/')
+    return '/' + stripped if stripped else '/'
+
+
+def _route_pattern(prefix: str) -> str:
+    stripped = prefix.strip('/')
+    return '/*' if not stripped else '/' + stripped + '/*'
+
+
+def key_from_path(path: str, prefix: str) -> Optional[str]:
+    """Normalize ``path`` and compute the cache key relative to ``prefix``.
+
+    Returns ``None`` when the path does not fall under the prefix (or resolves
+    to an empty key), mirroring the 400 responses of the original RBCS for
+    unhandled / escaping paths.
+    """
+    normalized = normpath(path)
+    prefix = _normalize_prefix(prefix)
+    if prefix == '/':
+        key = normalized.lstrip('/')
+    else:
+        if normalized != prefix and not normalized.startswith(prefix + '/'):
+            return None
+        key = normalized[len(prefix):].lstrip('/')
+    return key or None
+
+
+def _first_header(headers: Mapping[str, Sequence[str]], name: str) -> Optional[str]:
+    values = headers.get(name)
+    if values is None:
+        return None
+    if isinstance(values, str):
+        return values
+    return values[0] if len(values) > 0 else None
+
+
+def create_app(config: Config) -> KayaApp:
+    store = MemcacheStore(
+        host=config.memcache_host,
+        port=config.memcache_port,
+        key_prefix=config.key_prefix,
+        digest=config.digest,
+        max_age=config.max_age,
+    )
+    prefix = config.path_prefix
+    app = KayaApp()
+    app.store = store  # type: ignore[attr-defined]
+
+    @app.GET(_route_pattern(prefix), recursive=True)
+    async def get_handler(ctx: HttpContext, path: List[str]) -> None:
+        key = key_from_path(ctx.path, prefix)
+        if key is None:
+            log.warning('Got GET request for unhandled path %s', ctx.path)
+            await ctx.send_empty(400)
+            return
+        result = await store.get(key)
+        if result is None:
+            log.debug('Cache miss for key %s', key)
+            await ctx.send_empty(404)
+            return
+        metadata, value = result
+        log.debug('Cache hit for key %s', key)
+        headers: dict[str, str] = {
+            'content-type': metadata.content_type or _APPLICATION_OCTET_STREAM,
+        }
+        if metadata.content_disposition is not None:
+            headers['content-disposition'] = metadata.content_disposition
+        await ctx.send_bytes(200, value, headers)
+
+    @app.PUT(_route_pattern(prefix), recursive=True)
+    async def put_handler(ctx: HttpContext, path: List[str]) -> None:
+        key = key_from_path(ctx.path, prefix)
+        if key is None:
+            log.warning('Got PUT request for unhandled path %s', ctx.path)
+            await ctx.send_empty(400)
+            return
+        chunks: List[bytes] = []
+        async for chunk in ctx.request_body:
+            chunks.append(chunk)
+        value = b''.join(chunks)
+        metadata = CacheValueMetadata(
+            content_type=_first_header(ctx.headers, 'content-type'),
+            content_disposition=_first_header(ctx.headers, 'content-disposition'),
+        )
+        await store.put(key, value, metadata)
+        log.debug('Added value for key %s', key)
+        await ctx.send_str(201, key, {'content-type': _TEXT_PLAIN})
+
+    return app
+
+
+app = create_app(Config.from_env())

src/kaya_rbcs/config.py

@@ -0,0 +1,44 @@
+from dataclasses import dataclass, field
+from datetime import timedelta
+from os import getenv
+
+
+def _parse_max_age(raw: str) -> timedelta:
+    if raw.endswith('s'):
+        return timedelta(seconds=int(raw[:-1]))
+    if raw.endswith('m'):
+        return timedelta(minutes=int(raw[:-1]))
+    if raw.endswith('h'):
+        return timedelta(hours=int(raw[:-1]))
+    if raw.endswith('d'):
+        return timedelta(days=int(raw[:-1]))
+    return timedelta(seconds=int(raw))
+
+
+@dataclass(frozen=True)
+class Config:
+    host: str = '127.0.0.1'
+    port: int = 8080
+    path_prefix: str = '/'
+    memcache_host: str = '127.0.0.1'
+    memcache_port: int = 11211
+    key_prefix: str | None = None
+    digest: str | None = None
+    max_age: timedelta = field(default_factory=lambda: timedelta(days=7))
+
+    @staticmethod
+    def from_env() -> 'Config':
+        def get_int(name: str, default: int) -> int:
+            raw = getenv(name)
+            return int(raw) if raw is not None else default
+
+        return Config(
+            host=getenv('RBCS_HOST', '127.0.0.1'),
+            port=get_int('RBCS_PORT', 8080),
+            path_prefix=getenv('RBCS_PATH_PREFIX', '/'),
+            memcache_host=getenv('RBCS_MEMCACHE_HOST', '127.0.0.1'),
+            memcache_port=get_int('RBCS_MEMCACHE_PORT', 11211),
+            key_prefix=getenv('RBCS_KEY_PREFIX'),
+            digest=getenv('RBCS_DIGEST'),
+            max_age=_parse_max_age(getenv('RBCS_MAX_AGE', '7d')),
+        )

src/kaya_rbcs/store.py

@@ -0,0 +1,117 @@
+import hashlib
+import json
+import struct
+from dataclasses import asdict, dataclass
+from datetime import timedelta
+from time import time
+from typing import Callable, Optional, Tuple
+
+import aiomcache
+
+_THIRTY_DAYS = 30 * 24 * 60 * 60
+
+
+def _exptime(max_age: timedelta, clock: Callable[[], float]) -> int:
+    """Convert a relative TTL to a memcached ``exptime`` value.
+
+    Memcached interprets ``exptime`` values larger than 30 days as absolute
+    Unix timestamps, so large TTLs must be converted explicitly.
+    """
+    seconds = int(max_age.total_seconds())
+    if seconds > _THIRTY_DAYS:
+        return int(clock()) + seconds
+    return seconds
+
+
+@dataclass(frozen=True)
+class CacheValueMetadata:
+    """Metadata stored alongside a cached value, mirroring RBCS' metadata."""
+
+    content_type: Optional[str] = None
+    content_disposition: Optional[str] = None
+
+
+def process_cache_key(key: str, key_prefix: Optional[str], digest: Optional[str]) -> bytes:
+    """Compute the memcached key, mirroring RBCS' ``RBCS.processCacheKey``.
+
+    The key prefix is appended to the key (as in the original implementation)
+    and, if a digest algorithm is configured, the result is hashed and returned
+    as a hexadecimal string so that it stays a valid memcached key.
+    """
+    prefixed_key = key + key_prefix if key_prefix is not None else key
+    data = prefixed_key.encode('utf-8')
+    if digest is not None:
+        return hashlib.new(digest, data).hexdigest().encode('utf-8')
+    return data
+
+
+def encode_value(value: bytes, metadata: CacheValueMetadata) -> bytes:
+    """Encode a cached value as ``[4-byte metadata length][json metadata][value]``."""
+    metadata_bytes = json.dumps(asdict(metadata)).encode('utf-8')
+    return struct.pack('>I', len(metadata_bytes)) + metadata_bytes + value
+
+
+def decode_value(payload: bytes) -> Tuple[CacheValueMetadata, bytes]:
+    """Decode a value previously encoded with :func:`encode_value`."""
+    (size,) = struct.unpack('>I', payload[:4])
+    metadata = json.loads(payload[4:4 + size].decode('utf-8'))
+    return CacheValueMetadata(**metadata), payload[4 + size:]
+
+
+class MemcacheStore:
+    """Memcached-backed cache store with RBCS-compatible value encoding.
+
+    The underlying ``aiomcache.Client`` is created lazily on first use so that
+    it is always bound to the currently running event loop; call :meth:`start`
+    eagerly inside a running loop if you want to fail fast, and :meth:`close`
+    to release the pool.
+    """
+
+    def __init__(
+        self,
+        host: str = '127.0.0.1',
+        port: int = 11211,
+        key_prefix: Optional[str] = None,
+        digest: Optional[str] = None,
+        max_age: timedelta = timedelta(days=7),
+        clock: Callable[[], float] = time,
+    ) -> None:
+        self._host = host
+        self._port = port
+        self._key_prefix = key_prefix
+        self._digest = digest
+        self._max_age = max_age
+        self._clock = clock
+        self._client: Optional[aiomcache.Client] = None
+
+    def _ensure_client(self) -> aiomcache.Client:
+        if self._client is None:
+            self._client = aiomcache.Client(self._host, self._port)
+        return self._client
+
+    async def start(self) -> None:
+        self._ensure_client()
+
+    def _memcache_key(self, key: str) -> bytes:
+        return process_cache_key(key, self._key_prefix, self._digest)
+
+    async def get(self, key: str) -> Optional[Tuple[CacheValueMetadata, bytes]]:
+        client = self._ensure_client()
+        payload = await client.get(self._memcache_key(key))
+        if payload is None:
+            return None
+        return decode_value(payload)
+
+    async def put(self, key: str, value: bytes, metadata: CacheValueMetadata) -> None:
+        client = self._ensure_client()
+        exptime = _exptime(self._max_age, self._clock)
+        await client.set(
+            self._memcache_key(key),
+            encode_value(value, metadata),
+            exptime=exptime,
+        )
+
+    async def close(self) -> None:
+        if self._client is not None:
+            await self._client.close()
+            self._client = None

tests/fake_memcached.py

@@ -0,0 +1,144 @@
+"""A minimal in-process memcached server (text protocol) for offline tests.
+
+It implements the subset of the memcached text protocol used by
+``aiomcache`` (and thus by this project): ``get``, ``set``/``add``/
+``replace``/``append``/``prepend``, ``touch``, ``delete``, ``version``,
+``flush_all`` and ``quit``, including lazy expiration.
+"""
+
+import asyncio
+from time import time
+from typing import Dict, List, Optional, Tuple
+
+_THIRTY_DAYS = 30 * 24 * 60 * 60
+
+
+class FakeMemcachedServer:
+    def __init__(self) -> None:
+        self._data: Dict[bytes, Tuple[int, bytes]] = {}
+        self._server: Optional[asyncio.AbstractServer] = None
+        self.port: int = 0
+
+    async def start(self) -> None:
+        self._server = await asyncio.start_server(self._handle, '127.0.0.1', 0)
+        assert self._server.sockets is not None
+        self.port = self._server.sockets[0].getsockname()[1]
+
+    async def stop(self) -> None:
+        if self._server is not None:
+            self._server.close()
+            await self._server.wait_closed()
+
+    @staticmethod
+    def _expiry_timestamp(exptime: int, now: float) -> int:
+        if exptime == 0:
+            return 0
+        if exptime > _THIRTY_DAYS:
+            return exptime
+        return int(now) + exptime
+
+    def _expired(self, key: bytes, now: float) -> bool:
+        entry = self._data.get(key)
+        if entry is None:
+            return False
+        expiry, _ = entry
+        return 0 < expiry <= now
+
+    async def _handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        try:
+            while True:
+                line = await reader.readline()
+                if not line:
+                    break
+                parts = line.rstrip(b'\r\n').split()
+                if not parts:
+                    continue
+                command = parts[0]
+                if command == b'get':
+                    self._handle_get(parts[1:], writer)
+                elif command in (b'set', b'add', b'replace', b'append', b'prepend'):
+                    await self._handle_store(command, parts, reader, writer)
+                elif command == b'touch':
+                    self._handle_touch(parts, writer)
+                elif command == b'delete':
+                    self._handle_delete(parts, writer)
+                elif command == b'version':
+                    writer.write(b'VERSION 1.6.23\r\n')
+                elif command == b'flush_all':
+                    self._data.clear()
+                    writer.write(b'OK\r\n')
+                elif command == b'quit':
+                    break
+                else:
+                    writer.write(b'ERROR\r\n')
+                await writer.drain()
+        finally:
+            writer.close()
+            await writer.wait_closed()
+
+    def _handle_get(self, keys: List[bytes], writer: asyncio.StreamWriter) -> None:
+        now = time()
+        for key in keys:
+            if key in self._data and not self._expired(key, now):
+                _, value = self._data[key]
+                writer.write(
+                    b'VALUE ' + key + b' 0 ' + str(len(value)).encode('utf-8') + b'\r\n'
+                )
+                writer.write(value + b'\r\n')
+        writer.write(b'END\r\n')
+
+    async def _handle_store(
+        self,
+        command: bytes,
+        parts: List[bytes],
+        reader: asyncio.StreamReader,
+        writer: asyncio.StreamWriter,
+    ) -> None:
+        key = parts[1]
+        exptime = int(parts[3])
+        length = int(parts[4])
+        value = (await reader.readexactly(length + 2))[:-2]
+        now = time()
+        stored = False
+        if command == b'set':
+            self._data[key] = (self._expiry_timestamp(exptime, now), value)
+            stored = True
+        elif command == b'add':
+            if not (key in self._data and not self._expired(key, now)):
+                self._data[key] = (self._expiry_timestamp(exptime, now), value)
+                stored = True
+        elif command == b'replace':
+            if key in self._data and not self._expired(key, now):
+                self._data[key] = (self._expiry_timestamp(exptime, now), value)
+                stored = True
+        elif command in (b'append', b'prepend'):
+            if key in self._data and not self._expired(key, now):
+                expiry, existing = self._data[key]
+                self._data[key] = (
+                    expiry,
+                    existing + value if command == b'append' else value + existing,
+                )
+                stored = True
+        writer.write(b'STORED\r\n' if stored else b'NOT_STORED\r\n')
+
+    def _handle_touch(self, parts: List[bytes], writer: asyncio.StreamWriter) -> None:
+        key = parts[1]
+        exptime = int(parts[2])
+        if key in self._data and not self._expired(key, time()):
+            _, value = self._data[key]
+            self._data[key] = (self._expiry_timestamp(exptime, time()), value)
+            writer.write(b'TOUCHED\r\n')
+        else:
+            writer.write(b'NOT_FOUND\r\n')
+
+    def _handle_delete(self, parts: List[bytes], writer: asyncio.StreamWriter) -> None:
+        key = parts[1]
+        if key in self._data:
+            del self._data[key]
+            writer.write(b'DELETED\r\n')
+        else:
+            writer.write(b'NOT_FOUND\r\n')
+
+    @property
+    def data(self) -> Dict[bytes, Tuple[int, bytes]]:
+        return self._data

tests/test_app.py

@@ -0,0 +1,208 @@
+import asyncio
+import hashlib
+import unittest
+from datetime import timedelta
+from typing import Any
+
+import httpx
+
+from kaya_rbcs.app import create_app, key_from_path
+from kaya_rbcs.config import Config
+from kaya_rbcs.store import MemcacheStore
+
+from fake_memcached import FakeMemcachedServer
+
+
+class RbcsTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self) -> None:
+        self.memcache = FakeMemcachedServer()
+        await self.memcache.start()
+        self.app = create_app(self._config())
+        self.store: MemcacheStore = getattr(self.app, 'store')
+        self.client = httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=self.app),
+            base_url='http://testserver',
+        )
+
+    async def asyncTearDown(self) -> None:
+        await self.client.aclose()
+        await self.store.close()
+        await self.memcache.stop()
+
+    def _config(self, **overrides: Any) -> Config:
+        params: dict[str, Any] = {
+            'host': '127.0.0.1',
+            'port': 8080,
+            'path_prefix': '/',
+            'memcache_host': '127.0.0.1',
+            'memcache_port': self.memcache.port,
+        }
+        params.update(overrides)
+        return Config(**params)
+
+    async def test_put_get_roundtrip(self) -> None:
+        key = 'abc123'
+        value = b'hello world'
+        put = await self.client.put(
+            '/' + key,
+            content=value,
+            headers={'content-type': 'application/octet-stream'},
+        )
+        self.assertEqual(201, put.status_code)
+        self.assertEqual(key, put.text)
+        self.assertEqual('text/plain', put.headers['content-type'])
+
+        get = await self.client.get('/' + key)
+        self.assertEqual(200, get.status_code)
+        self.assertEqual(value, get.content)
+        self.assertEqual('application/octet-stream', get.headers['content-type'])
+
+    async def test_get_missing_key(self) -> None:
+        get = await self.client.get('/does/not/exist')
+        self.assertEqual(404, get.status_code)
+        self.assertEqual(b'', get.content)
+
+    async def test_nested_key(self) -> None:
+        value = b'nested value'
+        put = await self.client.put('/a/b/c', content=value)
+        self.assertEqual(201, put.status_code)
+        get = await self.client.get('/a/b/c')
+        self.assertEqual(200, get.status_code)
+        self.assertEqual(value, get.content)
+
+    async def test_default_content_type(self) -> None:
+        await self.client.put('/no-type', content=b'x')
+        get = await self.client.get('/no-type')
+        self.assertEqual('application/octet-stream', get.headers['content-type'])
+
+    async def test_content_disposition_roundtrip(self) -> None:
+        disposition = 'inline; filename="page.html"'
+        await self.client.put(
+            '/page',
+            content=b'<html></html>',
+            headers={'content-type': 'text/html', 'content-disposition': disposition},
+        )
+        get = await self.client.get('/page')
+        self.assertEqual(200, get.status_code)
+        self.assertEqual('text/html', get.headers['content-type'])
+        self.assertEqual(disposition, get.headers['content-disposition'])
+
+    async def test_key_prefix_appended(self) -> None:
+        app = create_app(self._config(key_prefix='suffix'))
+        store: MemcacheStore = getattr(app, 'store')
+        client = httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app),
+            base_url='http://testserver',
+        )
+        try:
+            put = await client.put('/key1', content=b'x')
+            self.assertEqual(201, put.status_code)
+            self.assertIn(b'key1suffix', self.memcache.data)
+
+            get = await client.get('/key1')
+            self.assertEqual(200, get.status_code)
+            self.assertEqual(b'x', get.content)
+        finally:
+            await client.aclose()
+            await store.close()
+
+    async def test_digest_hashes_key(self) -> None:
+        app = create_app(self._config(digest='sha256'))
+        store: MemcacheStore = getattr(app, 'store')
+        client = httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app),
+            base_url='http://testserver',
+        )
+        try:
+            put = await client.put('/key1', content=b'x')
+            self.assertEqual(201, put.status_code)
+            expected = hashlib.sha256(b'key1').hexdigest().encode('utf-8')
+            self.assertIn(expected, self.memcache.data)
+
+            get = await client.get('/key1')
+            self.assertEqual(200, get.status_code)
+            self.assertEqual(b'x', get.content)
+        finally:
+            await client.aclose()
+            await store.close()
+
+    async def test_path_prefix(self) -> None:
+        app = create_app(self._config(path_prefix='/cache'))
+        store: MemcacheStore = getattr(app, 'store')
+        client = httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app),
+            base_url='http://testserver',
+        )
+        try:
+            await client.put('/cache/entry', content=b'value')
+            get = await client.get('/cache/entry')
+            self.assertEqual(200, get.status_code)
+            self.assertEqual(b'value', get.content)
+            self.assertIn(b'entry', self.memcache.data)
+        finally:
+            await client.aclose()
+            await store.close()
+
+    def test_key_from_path_rejects_escaping(self) -> None:
+        self.assertIsNone(key_from_path('/cache/../../etc/passwd', '/cache'))
+        self.assertIsNone(key_from_path('/cache/../', '/cache'))
+        self.assertIsNone(key_from_path('/', '/'))
+        self.assertEqual('a/b', key_from_path('/cache/a/b', '/cache'))
+        self.assertEqual('foo', key_from_path('/foo', '/'))
+        self.assertEqual('foo', key_from_path('/cache/foo', '/cache'))
+
+    async def test_escaped_path_is_rejected(self) -> None:
+        # ASGI servers (and httpx) normalize dot segments in the request path
+        # before it reaches the app, so the route is never matched. Either way
+        # the request is rejected; a raw, unnormalized escaping path would be
+        # handled by key_from_path and rejected with 400.
+        app = create_app(self._config(path_prefix='/cache'))
+        store: MemcacheStore = getattr(app, 'store')
+        client = httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app),
+            base_url='http://testserver',
+        )
+        try:
+            get = await client.get('/cache/../../etc/passwd')
+            self.assertIn(get.status_code, (400, 404))
+            put = await client.put('/cache/../../etc/passwd', content=b'x')
+            self.assertIn(put.status_code, (400, 404))
+        finally:
+            await client.aclose()
+            await store.close()
+
+    async def test_max_age_expiry(self) -> None:
+        app = create_app(self._config(max_age=timedelta(seconds=1)))
+        store: MemcacheStore = getattr(app, 'store')
+        client = httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app),
+            base_url='http://testserver',
+        )
+        try:
+            put = await client.put('/ttl', content=b'x')
+            self.assertEqual(201, put.status_code)
+
+            get = await client.get('/ttl')
+            self.assertEqual(200, get.status_code)
+
+            await asyncio.sleep(1.2)
+            get = await client.get('/ttl')
+            self.assertEqual(404, get.status_code)
+        finally:
+            await client.aclose()
+            await store.close()
+
+    async def test_overwrite_value(self) -> None:
+        await self.client.put('/same', content=b'first')
+        await self.client.put('/same', content=b'second')
+        get = await self.client.get('/same')
+        self.assertEqual(200, get.status_code)
+        self.assertEqual(b'second', get.content)
+
+    async def test_invalid_method_returns_404(self) -> None:
+        post = await self.client.post('/anything', content=b'x')
+        self.assertEqual(404, post.status_code)
+
+
+if __name__ == '__main__':
+    unittest.main()