"""Sync v1 HTTP 边界；每次访问重新检查设备撤销，内容不进入日志。"""

import hashlib
import json
import secrets
import time
import os
import tempfile
import asyncio
from contextlib import asynccontextmanager
import logging
from pathlib import Path

from fastapi import FastAPI, Header, Query, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse, FileResponse
from starlette.background import BackgroundTask
from starlette.concurrency import run_in_threadpool
from starlette.staticfiles import StaticFiles

from .database import Database, password_hash, row, rows, run
from .models import Commit, CredentialChange, Login, Refresh, Upload, VaultCreate
from .readiness import Readiness


class SyncError(Exception):
    def __init__(self, status, code, details=None):
        self.status, self.code, self.details = status, code, details or {}


class StagingDamaged(Exception):
    """用于在返回 UPLOAD_DAMAGED 之前提交清理的内部信号。"""

    def __init__(self, upload):
        self.upload_id = upload["id"]
        self.vault_id = upload["vault_id"]
        self.device_id = upload["device_id"]


def digest(value: str) -> str:
    return hashlib.sha256(value.encode()).hexdigest()


def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=time.time):
    db.migrate()
    staging.mkdir(parents=True, exist_ok=True)
    @asynccontextmanager
    async def lifespan(_app):
        from .maintenance import cleanup_expired_uploads
        stopping = asyncio.Event()

        async def maintain():
            while not stopping.is_set():
                try:
                    await asyncio.to_thread(cleanup_expired_uploads, db, staging, now=int(clock()))
                except Exception:
                    logging.getLogger(__name__).error("UPLOAD_MAINTENANCE_FAILED")
                try:
                    await asyncio.wait_for(stopping.wait(), timeout=60)
                except TimeoutError:
                    pass

        worker = asyncio.create_task(maintain())
        try:
            yield
        finally:
            stopping.set()
            await worker

    app = FastAPI(title="OpenNexus Sync", version="1.0.0", lifespan=lifespan)
    app.state.database = db
    worker_id = secrets.token_hex(8)
    console_root = Path(__file__).with_name("static")

    @app.middleware("http")
    async def console_security(request: Request, call_next):
        response = await call_next(request)
        if request.url.path == "/" or request.url.path.startswith("/console"):
            response.headers["Cache-Control"] = "no-store"
            response.headers["Content-Security-Policy"] = (
                "default-src 'none'; script-src 'self'; style-src 'self'; "
                "img-src 'self' data:; connect-src 'self'; font-src 'self'; "
                "base-uri 'none'; form-action 'self'; frame-ancestors 'none'"
            )
            response.headers["Referrer-Policy"] = "no-referrer"
            response.headers["X-Content-Type-Options"] = "nosniff"
            response.headers["X-Frame-Options"] = "DENY"
        return response

    @app.get("/", include_in_schema=False)
    def console():
        return FileResponse(console_root / "index.html", media_type="text/html; charset=utf-8")

    app.mount("/console", StaticFiles(directory=console_root, html=True), name="sync-console")

    @app.exception_handler(SyncError)
    async def error(_request, exc):
        headers = {"Retry-After": "60"} if exc.status == 429 else {}
        if exc.status == 503:
            headers["X-OpenNexus-Worker"] = worker_id
        return JSONResponse({"error": {"code": exc.code, "details": exc.details}}, status_code=exc.status, headers=headers)

    @app.exception_handler(RequestValidationError)
    async def invalid(_request, _exc):
        # Pydantic 的原始错误可能带请求正文，禁止回显密码或笔记。
        return JSONResponse({"error": {"code": "INVALID_REQUEST", "details": {}}}, status_code=422)

    def identity(conn, authorization, *, allow_bootstrap=False):
        token = authorization.removeprefix("Bearer ") if authorization.startswith("Bearer ") else ""
        session = row(conn, "SELECT s.*, d.user_id, d.revoked FROM sessions s JOIN devices d ON d.id=s.device_id WHERE s.token=:token", token=digest(token))
        if not session or session["revoked"] or session["expires"] <= clock():
            raise SyncError(401, "SESSION_EXPIRED")
        if not allow_bootstrap and row(conn, "SELECT user_id FROM bootstrap_state WHERE user_id=:user",
                                       user=session["user_id"]):
            raise SyncError(403, "CREDENTIAL_CHANGE_REQUIRED")
        return session

    def vault(conn, vault_id, authorization, *, lock=False):
        session = identity(conn, authorization)
        # Vault 行锁覆盖配额、CAS、路径冲突和序列分配，跨 Worker 也保持一致。
        suffix = " FOR UPDATE" if lock and not db.sqlite else ""
        item = row(conn, "SELECT * FROM vaults WHERE id=:id AND user_id=:owner" + suffix,
                   id=vault_id, owner=session["user_id"])
        if not item:
            raise SyncError(404, "VAULT_NOT_FOUND")
        return session, item

    def issue(conn, device_id):
        access, refresh = secrets.token_urlsafe(32), secrets.token_urlsafe(48)
        run(conn, "INSERT INTO sessions VALUES (:token,:refresh,:device,:expires,:refresh_expires)",
            token=digest(access), refresh=digest(refresh), device=device_id,
            expires=int(clock()) + 900, refresh_expires=int(clock()) + 30 * 86400)
        device = row(conn, "SELECT user_id FROM devices WHERE id=:id", id=device_id)
        must_change = bool(row(conn, "SELECT user_id FROM bootstrap_state WHERE user_id=:user",
                               user=device["user_id"]))
        return {"access_token": access, "refresh_token": refresh, "expires_in": 900,
                "device_id": device_id, "must_change_credentials": must_change}

    @app.get("/health")
    def health(response: Response):
        # 临时标识符可以让部署探测证明两个配置的工作线程都接收流量，而不会暴露主机身份。
        response.headers["X-OpenNexus-Worker"] = worker_id
        return {"status": "ok"}

    def readiness_probe():
        with db.transaction() as conn:
            if row(conn, "SELECT version FROM schema_version")["version"] != 1:
                raise SyncError(503, "SCHEMA_INCOMPATIBLE")
        key = "health-probe/" + secrets.token_hex(16)
        stored = False
        try:
            with tempfile.TemporaryFile(dir=staging) as local:
                local.write(b"opennexus-ready")
                local.flush()
                os.fsync(local.fileno())
                local.seek(0)
                if local.read() != b"opennexus-ready":
                    raise OSError("STAGING_INTEGRITY")
            objects.put(key, b"opennexus-ready")
            stored = True
            if objects.get(key) != b"opennexus-ready":
                raise OSError("STORAGE_INTEGRITY")
        finally:
            if stored:
                objects.delete(key)

    readiness = Readiness(readiness_probe, cache_seconds=1)

    @app.get("/ready")
    async def ready(response: Response):
        if not await readiness.check():
            raise SyncError(503, "DEPENDENCY_UNAVAILABLE")
        response.headers["X-OpenNexus-Worker"] = worker_id
        return {"status": "ready", "schema": 1}

    @app.get("/sync/v1/handshake")
    def handshake(protocol: int = 1):
        if protocol != 1:
            raise SyncError(426, "PROTOCOL_INCOMPATIBLE")
        return {"protocol": 1, "max_object_size": 104857600, "chunk_size": 1048576,
                "encryption": "transport-only", "history_retention": "indefinite",
                "cursor_retention": "indefinite", "sharing": False}

    @app.post("/sync/v1/auth/sessions")
    def login(body: Login, request: Request):
        key = digest((request.client.host if request.client else "unknown") + ":" + body.username.casefold())
        # 失败计数先独立提交，抛出认证异常也不会回滚限流状态。
        with db.transaction() as conn:
            limit = row(conn, "SELECT * FROM login_limits WHERE key=:key", key=key)
            if limit and clock() - limit["started"] < 60:
                if limit["attempts"] >= 10:
                    raise SyncError(429, "RATE_LIMITED")
                run(conn, "UPDATE login_limits SET attempts=attempts+1 WHERE key=:key", key=key)
            else:
                run(conn, "DELETE FROM login_limits WHERE key=:key", key=key)
                run(conn, "INSERT INTO login_limits VALUES (:key,:now,1)", key=key, now=int(clock()))
        with db.transaction() as conn:
            user = row(conn, "SELECT * FROM users WHERE username=:name", name=body.username)
            expected = user["password"] if user else password_hash("unavailable-user-password", "0" * 32)
            if not secrets.compare_digest(password_hash(body.password, expected.split(":")[0]), expected) or not user:
                raise SyncError(401, "LOGIN_FAILED")
            device_id = secrets.token_hex(16)
            run(conn, "INSERT INTO devices VALUES (:id,:user,:name,0)", id=device_id, user=user["id"], name=body.device_name)
            return issue(conn, device_id)

    @app.post("/sync/v1/auth/refresh")
    def refresh(body: Refresh):
        with db.transaction() as conn:
            suffix = " FOR UPDATE OF s" if not db.sqlite else ""
            session = row(conn, "SELECT s.*,d.revoked FROM sessions s JOIN devices d ON d.id=s.device_id WHERE refresh=:refresh" + suffix,
                          refresh=digest(body.refresh_token))
            if not session or session["revoked"] or session["refresh_expires"] <= clock():
                raise SyncError(401, "SESSION_EXPIRED")
            run(conn, "DELETE FROM sessions WHERE token=:token", token=session["token"])
            return issue(conn, session["device_id"])

    @app.delete("/sync/v1/auth/sessions", status_code=204)
    def logout(authorization: str = Header(default="")):
        with db.transaction() as conn:
            session = identity(conn, authorization, allow_bootstrap=True)
            run(conn, "DELETE FROM sessions WHERE token=:token", token=session["token"])

    @app.put("/sync/v1/account/credentials")
    def change_credentials(body: CredentialChange, authorization: str = Header(default="")):
        with db.transaction() as conn:
            session = identity(conn, authorization, allow_bootstrap=True)
            user = row(conn, "SELECT * FROM users WHERE id=:id", id=session["user_id"])
            expected = user["password"]
            if not secrets.compare_digest(
                    password_hash(body.current_password, expected.split(":")[0]), expected):
                raise SyncError(401, "CURRENT_PASSWORD_INVALID")
            conflict = row(conn, "SELECT id FROM users WHERE username=:name AND id<>:id",
                           name=body.username, id=user["id"])
            if conflict:
                raise SyncError(409, "USERNAME_TAKEN")
            run(conn, "UPDATE users SET username=:name,password=:password WHERE id=:id",
                name=body.username, password=password_hash(body.password), id=user["id"])
            run(conn, "DELETE FROM bootstrap_state WHERE user_id=:user", user=user["id"])
            run(conn, "DELETE FROM sessions WHERE device_id IN (SELECT id FROM devices WHERE user_id=:user) AND token<>:token",
                user=user["id"], token=session["token"])
            run(conn, "UPDATE devices SET revoked=1 WHERE user_id=:user AND id<>:device",
                user=user["id"], device=session["device_id"])
            return {"username": body.username, "credentials_fixed": True}

    @app.get("/sync/v1/devices")
    def devices(authorization: str = Header(default="")):
        with db.transaction() as conn:
            session = identity(conn, authorization)
            return {"items": rows(conn, "SELECT id,name,revoked FROM devices WHERE user_id=:user", user=session["user_id"])}

    @app.delete("/sync/v1/devices/{device_id}", status_code=204)
    def revoke(device_id: str, authorization: str = Header(default="")):
        with db.transaction() as conn:
            session = identity(conn, authorization)
            run(conn, "UPDATE devices SET revoked=1 WHERE id=:id AND user_id=:user", id=device_id, user=session["user_id"])

    @app.post("/sync/v1/vaults")
    def create_vault(body: VaultCreate, authorization: str = Header(default="")):
        with db.transaction() as conn:
            session = identity(conn, authorization)
            vault_id = secrets.token_hex(16)
            run(conn, "INSERT INTO vaults VALUES (:id,:user,:name,0,:quota,0)", id=vault_id, user=session["user_id"], name=body.name, quota=quota)
            return {"vault_id": vault_id, "name": body.name}

    @app.get("/sync/v1/vaults")
    def list_vaults(authorization: str = Header(default="")):
        with db.transaction() as conn:
            session = identity(conn, authorization)
            return {"items": rows(conn, "SELECT id,name,sequence,used,quota FROM vaults WHERE user_id=:user", user=session["user_id"])}

    @app.post("/sync/v1/vaults/{vault_id}/uploads")
    def begin_upload(vault_id: str, body: Upload, authorization: str = Header(default="")):
        with db.transaction() as conn:
            session, item = vault(conn, vault_id, authorization, lock=True)
            found = row(conn, "SELECT * FROM objects WHERE vault_id=:v AND hash=:h", v=vault_id, h=body.content_hash)
            if found:
                if found["size"] != body.size:
                    raise SyncError(409, "OBJECT_SIZE_MISMATCH")
                return {"complete": True, "upload_id": None, "offset": body.size}
            reserved = row(conn, "SELECT COALESCE(SUM(size),0) AS size FROM uploads WHERE vault_id=:v AND expires>:now", v=vault_id, now=int(clock()))["size"]
            if item["used"] + reserved + body.size > item["quota"]:
                raise SyncError(413, "QUOTA_EXCEEDED")
            upload_id = secrets.token_hex(16)
            run(conn, "INSERT INTO uploads VALUES (:id,:v,:device,:h,:size,0,:expires)", id=upload_id,
                v=vault_id, device=session["device_id"], h=body.content_hash, size=body.size, expires=int(clock()) + 3600)
            (staging / upload_id).write_bytes(b"")
            return {"complete": False, "upload_id": upload_id, "offset": 0}

    def authorized_upload(conn, vault_id, upload_id, authorization):
        session, _ = vault(conn, vault_id, authorization, lock=True)
        upload = row(conn, "SELECT * FROM uploads WHERE id=:id AND vault_id=:v AND device_id=:device", id=upload_id, v=vault_id, device=session["device_id"])
        if not upload or upload["expires"] <= clock():
            raise SyncError(404, "UPLOAD_EXPIRED")
        return upload

    def reconcile_staging(upload):
        path = staging / upload["id"]
        try:
            length = path.stat().st_size
            if length < upload["offset_bytes"]:
                raise StagingDamaged(upload)
            if length > upload["offset_bytes"]:
                with path.open("r+b") as stream:
                    stream.truncate(upload["offset_bytes"])
                    stream.flush()
                    os.fsync(stream.fileno())
        except OSError:
            raise StagingDamaged(upload) from None
        return path

    def discard_damaged_upload(damaged):
        # 失败的操作释放其事务后重新打开。失败事务中的删除操作将随响应一起回滚。
        with db.transaction() as conn:
            suffix = " FOR UPDATE" if not db.sqlite else ""
            row(conn, "SELECT id FROM vaults WHERE id=:v" + suffix, v=damaged.vault_id)
            upload = row(
                conn,
                "SELECT id FROM uploads WHERE id=:id AND vault_id=:v AND device_id=:device",
                id=damaged.upload_id,
                v=damaged.vault_id,
                device=damaged.device_id,
            )
            if upload:
                # 文件优先：中断留下一行，其配额保留仍可通过到期维护释放。
                (staging / upload["id"]).unlink(missing_ok=True)
                run(conn, "DELETE FROM uploads WHERE id=:id", id=upload["id"])

    def upload_damaged(damaged):
        discard_damaged_upload(damaged)
        raise SyncError(409, "UPLOAD_DAMAGED", {"restart_required": True})

    @app.get("/sync/v1/vaults/{vault_id}/uploads/{upload_id}")
    def upload_status(vault_id: str, upload_id: str, authorization: str = Header(default="")):
        try:
            with db.transaction() as conn:
                upload = authorized_upload(conn, vault_id, upload_id, authorization)
                reconcile_staging(upload)
                return {"offset": upload["offset_bytes"], "size": upload["size"], "expires": upload["expires"]}
        except StagingDamaged as damaged:
            upload_damaged(damaged)

    @app.put("/sync/v1/vaults/{vault_id}/uploads/{upload_id}")
    async def upload_chunk(vault_id: str, upload_id: str, request: Request,
                           offset: int = Query(ge=0), authorization: str = Header(default="")):
        data = bytearray()
        async for chunk in request.stream():
            if len(chunk) > 1048576 - len(data):
                raise SyncError(413, "CHUNK_TOO_LARGE")
            data.extend(chunk)
        # 将事务和持久写入保留在一个工作线程上。缓慢的数据库锁定或 fsync 不得阻止此工作线程的 ASGI 事件循环。
        return await run_in_threadpool(persist_upload_chunk, vault_id, upload_id,
                                       authorization, offset, data)

    def persist_upload_chunk(vault_id, upload_id, authorization, offset, data):
        try:
            with db.transaction() as conn:
                upload = authorized_upload(conn, vault_id, upload_id, authorization)
                reconcile_staging(upload)
                if offset != upload["offset_bytes"]:
                    raise SyncError(409, "UPLOAD_OFFSET", {"offset": upload["offset_bytes"]})
                if offset + len(data) > upload["size"]:
                    raise SyncError(413, "OBJECT_TOO_LARGE")
                # 先刷盘后提交 offset；崩溃重试覆盖未确认尾部，不能重复追加。
                import os
                with (staging / upload_id).open("r+b") as stream:
                    stream.seek(offset)
                    stream.write(data)
                    stream.truncate()
                    stream.flush()
                    os.fsync(stream.fileno())
                run(conn, "UPDATE uploads SET offset_bytes=:offset WHERE id=:id", id=upload_id, offset=offset + len(data))
                return {"offset": offset + len(data)}
        except StagingDamaged as damaged:
            upload_damaged(damaged)

    @app.delete("/sync/v1/vaults/{vault_id}/uploads/{upload_id}", status_code=204)
    def cancel_upload(vault_id: str, upload_id: str, authorization: str = Header(default="")):
        with db.transaction() as conn:
            authorized_upload(conn, vault_id, upload_id, authorization)
            run(conn, "DELETE FROM uploads WHERE id=:id", id=upload_id)
        (staging / upload_id).unlink(missing_ok=True)

    @app.post("/sync/v1/vaults/{vault_id}/uploads/{upload_id}/complete")
    def complete_upload(vault_id: str, upload_id: str, authorization: str = Header(default="")):
        try:
            with db.transaction() as conn:
                session, _ = vault(conn, vault_id, authorization, lock=True)
                receipt = row(conn, "SELECT hash FROM upload_receipts WHERE id=:id AND vault_id=:v AND device_id=:d",
                              id=upload_id, v=vault_id, d=session["device_id"])
                if receipt:
                    return {"complete": True, "content_hash": receipt["hash"]}
                upload = authorized_upload(conn, vault_id, upload_id, authorization)
                path = reconcile_staging(upload)
                with path.open("rb") as stream:
                    content_hash = hashlib.file_digest(stream, "sha256").hexdigest()
                if upload["size"] != upload["offset_bytes"] or content_hash != upload["hash"]:
                    raise SyncError(422, "OBJECT_INTEGRITY")
                exists = row(conn, "SELECT hash FROM objects WHERE vault_id=:v AND hash=:h", v=vault_id, h=upload["hash"])
                if not exists:
                    objects.put_file(vault_id + "/" + upload["hash"], path, upload["hash"])
                    run(conn, "INSERT INTO objects VALUES (:v,:h,:size,:now)", v=vault_id, h=upload["hash"], size=upload["size"], now=int(clock()))
                    run(conn, "UPDATE vaults SET used=used+:size WHERE id=:v", size=upload["size"], v=vault_id)
                run(conn, "INSERT INTO upload_receipts VALUES (:id,:v,:d,:h,:now)", id=upload_id,
                    v=vault_id, d=session["device_id"], h=upload["hash"], now=int(clock()))
                run(conn, "DELETE FROM uploads WHERE id=:id", id=upload_id)
        except StagingDamaged as damaged:
            upload_damaged(damaged)
        (staging / upload_id).unlink(missing_ok=True)
        return {"complete": True, "content_hash": upload["hash"]}

    def perform_commit(conn, vault_id, body, session, item):
        fingerprint = digest(body.model_dump_json())
        previous = row(conn, "SELECT * FROM revisions WHERE vault_id=:v AND operation_id=:op", v=vault_id, op=body.operation_id)
        if previous:
            if previous["fingerprint"] != fingerprint or previous["device_id"] != session["device_id"]:
                raise SyncError(409, "IDEMPOTENCY_REUSED")
            return dict(previous)
        current = row(conn, "SELECT * FROM files WHERE vault_id=:v AND file_id=:f", v=vault_id, f=body.file_id)
        if (current["sequence"] if current else 0) != body.base_revision:
            actual = row(conn, "SELECT * FROM revisions WHERE vault_id=:v AND sequence=:s", v=vault_id, s=current["sequence"]) if current else None
            raise SyncError(409, "REVISION_CONFLICT", {"current": dict(actual) if actual else None})
        if body.operation == "put":
            obj = row(conn, "SELECT * FROM objects WHERE vault_id=:v AND hash=:h", v=vault_id, h=body.content_hash)
            if not obj or obj["size"] != body.size:
                raise SyncError(409, "OBJECT_NOT_READY")
            paths = rows(conn, "SELECT path_key FROM files WHERE vault_id=:v AND deleted=0 AND file_id<>:f", v=vault_id, f=body.file_id)
            key = body.path.casefold()
            if any(p["path_key"] == key or p["path_key"].startswith(key + "/") or key.startswith(p["path_key"] + "/") for p in paths):
                raise SyncError(409, "PATH_CONFLICT")
        elif not current or body.content_hash is not None or body.size != 0:
            raise SyncError(422, "INVALID_DELETE")
        sequence = item["sequence"] + 1
        run(conn, "UPDATE vaults SET sequence=:s WHERE id=:v", s=sequence, v=vault_id)
        run(conn, "INSERT INTO revisions VALUES (:v,:s,:f,:base,:path,:key,:operation,:hash,:size,:device,:op,:fingerprint)",
            v=vault_id, s=sequence, f=body.file_id, base=body.base_revision, path=body.path, key=body.path.casefold(),
            operation=body.operation, hash=body.content_hash, size=body.size, device=session["device_id"], op=body.operation_id, fingerprint=fingerprint)
        run(conn, "DELETE FROM files WHERE vault_id=:v AND file_id=:f", v=vault_id, f=body.file_id)
        run(conn, "INSERT INTO files VALUES (:v,:f,:s,:key,:deleted)", v=vault_id, f=body.file_id, s=sequence,
            key=body.path.casefold(), deleted=int(body.operation == "delete"))
        return dict(row(conn, "SELECT * FROM revisions WHERE vault_id=:v AND sequence=:s", v=vault_id, s=sequence))

    @app.post("/sync/v1/vaults/{vault_id}/revisions")
    def commit(vault_id: str, body: Commit, authorization: str = Header(default="")):
        with db.transaction() as conn:
            session, item = vault(conn, vault_id, authorization, lock=True)
            return perform_commit(conn, vault_id, body, session, item)

    @app.get("/sync/v1/vaults/{vault_id}/changes")
    def changes(vault_id: str, cursor: int = Query(default=0, ge=0), limit: int = Query(default=100, ge=1, le=500),
                boundary: int | None = Query(default=None, ge=0), authorization: str = Header(default="")):
        with db.transaction() as conn:
            _, item = vault(conn, vault_id, authorization)
            end = item["sequence"] if boundary is None else boundary
            if cursor > end or end > item["sequence"]:
                raise SyncError(409, "CURSOR_INVALID")
            items = rows(conn, "SELECT * FROM revisions WHERE vault_id=:v AND sequence>:cursor AND sequence<=:end ORDER BY sequence LIMIT :limit", v=vault_id, cursor=cursor, end=end, limit=limit)
            next_cursor = items[-1]["sequence"] if items else cursor
            return {"items": items, "cursor": next_cursor, "boundary": end, "has_more": next_cursor < end}

    @app.get("/sync/v1/vaults/{vault_id}/history/{file_id}")
    def history(vault_id: str, file_id: str, before: int = Query(default=9223372036854775807, ge=1),
                limit: int = Query(default=100, ge=1, le=500), authorization: str = Header(default="")):
        with db.transaction() as conn:
            vault(conn, vault_id, authorization)
            return {"items": rows(conn, "SELECT * FROM revisions WHERE vault_id=:v AND file_id=:f AND sequence<:before ORDER BY sequence DESC LIMIT :limit", v=vault_id, f=file_id, before=before, limit=limit)}

    @app.get("/sync/v1/vaults/{vault_id}/objects/{content_hash}")
    def get_object(vault_id: str, content_hash: str, authorization: str = Header(default="")):
        with db.transaction() as conn:
            vault(conn, vault_id, authorization)
            obj = row(conn, "SELECT * FROM objects WHERE vault_id=:v AND hash=:h", v=vault_id, h=content_hash)
            if not obj:
                raise SyncError(404, "OBJECT_NOT_FOUND")
            expected_size = obj["size"]
        # 在返回任何字节之前进行验证，而不保留数据库事务或在 RAM 中缓冲整个附件。
        temporary = tempfile.NamedTemporaryFile(prefix="download-", dir=staging, delete=False)
        path = Path(temporary.name)
        try:
            size, checksum = 0, hashlib.sha256()
            with temporary, objects.open(vault_id + "/" + content_hash) as source:
                while chunk := source.read(1048576):
                    size += len(chunk)
                    if size > expected_size:
                        raise SyncError(503, "STORAGE_INTEGRITY")
                    checksum.update(chunk)
                    temporary.write(chunk)
            if size != expected_size or checksum.hexdigest() != content_hash:
                raise SyncError(503, "STORAGE_INTEGRITY")
            return FileResponse(path, media_type="application/octet-stream",
                headers={"ETag": '"' + content_hash + '"', "Cache-Control": "private, no-store"},
                background=BackgroundTask(path.unlink, missing_ok=True))
        except BaseException:
            temporary.close()
            path.unlink(missing_ok=True)
            raise

    return app