refactor: lay the repository out by role

local/ held six unrelated things under a name that stopped being true once it
became the production runtime: the service, the frontend, the tests, the ops
commands, the container definitions and the dependency lock, 65 files with
nothing to tell them apart.

  app/      the service: api/ routers, core/ for identity, database, models,
            prices and secret loading, and the worker, bootstrap and schema
  tests/    the twelve suites, no longer inside the shipped package
  ops/      backup, readiness, dependency audit, security summary
  infra/    Dockerfiles, gateway templates, ClamAV and storage configuration,
            the requirements and their hash lock
  web/      the Site, Kanban and portal pages with their scripts

deploy/Dockerfile.api now copies app/ alone, so the tests stop shipping to
production; the local image still carries them, because the suites run inside
the stack's network.

Five kinds of reference had to follow, and each was found by something different
rather than by reading. Imports of the form "from . import db" survived a rewrite
that only matched "from .db import". Tests kept relative imports of modules that
had left the package. A mock.patch target names its module in a string, where no
import rewriting can see it. The browser test resolves a fixture by path. And the
release gate's markers pointed at local/runtime.py and local/worker.py, which is
the decay its new marker test exists to catch — it caught it.

Verified from docker compose down -v: the stack starts, all six integration
suites, both browser suites and the twenty-nine unit tests pass.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

app/api/__init__.py

@@ -0,0 +1,2 @@
+"""HTTP routers, one per resource. They import from local.runtime, never
+from each other or from local.app."""

app/api/artwork.py

@@ -0,0 +1,49 @@
+"""Operator artwork handling: final files, and the uploads that carry them.
+
+These reuse the customer upload routes with the operator's derived identity, so
+one transport serves both and there is only one place where parts are signed.
+"""
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException
+
+from ..core import db
+from ..artwork import submit_files
+from ..core.auth import operator
+from ..core.models import ArtworkSubmission, UploadStart
+from ..runtime import file_rows, operator_identity
+from .uploads import begin_upload, complete_upload, part_url, upload_status
+
+router = APIRouter()
+
+@router.post('/api/operator/orders/{oid}/uploads')
+def begin_final(oid: UUID, body: UploadStart, user=Depends(operator)):
+    with db.connect() as c:
+        row = c.execute('SELECT state FROM dtf_local.orders WHERE id=%s', (oid,)).fetchone()
+        if not row: raise HTTPException(404, 'Order not found')
+        if row['state'] not in ('rec','tra','cor'): raise HTTPException(409, 'Order is not in artwork review')
+    return begin_upload(body, session_id=operator_identity(user))
+
+@router.get('/api/operator/uploads/{uid}')
+def final_status(uid: UUID, user=Depends(operator)):
+    return upload_status(uid,session_id=operator_identity(user))
+
+@router.post('/api/operator/uploads/{uid}/parts/{number}')
+def final_part(uid: UUID, number: int, user=Depends(operator)):
+    return part_url(uid,number,session_id=operator_identity(user))
+
+@router.post('/api/operator/uploads/{uid}/complete')
+def final_complete(uid: UUID, user=Depends(operator)):
+    return complete_upload(uid,session_id=operator_identity(user))
+
+@router.get('/api/operator/orders/{oid}/files')
+def operator_files(oid: UUID, user=Depends(operator)):
+    with db.connect() as c:
+        return file_rows(c,oid)
+
+@router.post('/api/operator/orders/{oid}/final-files')
+def final_files(oid: UUID, body: ArtworkSubmission, user=Depends(operator)):
+    with db.connect() as c:
+        order = c.execute('SELECT * FROM dtf_local.orders WHERE id=%s FOR UPDATE', (oid,)).fetchone()
+        if not order: raise HTTPException(404, 'Order not found')
+        return submit_files(c,order,body,operator_identity(user),'final',user)

app/api/customer.py

@@ -0,0 +1,116 @@
+"""Customer accounts, their orders, and the corrections they submit."""
+from datetime import datetime, timedelta, timezone
+from uuid import UUID, uuid4
+
+from fastapi import APIRouter, Depends, HTTPException, Request, Response
+from psycopg.errors import UniqueViolation
+from psycopg.types.json import Jsonb
+
+from ..core import db
+from ..artwork import submit_files
+from ..core.auth import (DUMMY_PASSWORD_HASH, audit, client_ip, new_session, owner,
+                    password_hash, password_matches, session_row, throttle, transfer_guest)
+from ..core.models import ArtworkSubmission, Login, Register
+from ..runtime import STATES, file_rows, owned_order, storage
+from ..scanning import require_clean
+
+router = APIRouter()
+
+
+def current(request):
+    try: return session_row(request)
+    except HTTPException: return None
+
+
+@router.post('/api/account/register')
+def register(body: Register, request: Request, response: Response):
+    email = body.customer.mail.strip().lower()
+    throttle(email, request)
+    previous = current(request)
+    encoded = password_hash(body.password)
+    identity = uuid4()
+    profile = body.customer.model_dump()
+    profile['mail'] = email
+    try:
+        with db.connect() as c:
+            if previous and c.execute('SELECT id FROM dtf_local.accounts WHERE id=%s', (previous['owner'],)).fetchone():
+                raise HTTPException(409, 'Sign out before registering another account')
+            c.execute('INSERT INTO dtf_local.accounts(id,email,password_hash,profile) VALUES(%s,%s,%s,%s)', (identity,email,encoded,Jsonb(profile)))
+            if previous: transfer_guest(c, previous, identity)
+            new_session(c, response, identity)
+    except UniqueViolation:
+        raise HTTPException(409, 'An account already exists; sign in')
+    audit('account_registered', account=str(identity))
+    return {'customer': profile}
+
+@router.post('/api/account/login')
+def login(body: Login, request: Request, response: Response):
+    email = body.email.strip().lower()
+    throttle(email, request)
+    with db.connect() as c:
+        account = c.execute('SELECT * FROM dtf_local.accounts WHERE email=%s', (email,)).fetchone()
+    # Comparable password work even when the email is absent.
+    stored = account['password_hash'] if account else DUMMY_PASSWORD_HASH
+    matches = password_matches(body.password, stored)
+    if not account or not matches:
+        audit('customer_login_failed', ip=client_ip(request))
+        raise HTTPException(401, 'Invalid email or password')
+    previous = current(request)
+    with db.connect() as c:
+        if not stored.startswith('scrypt-v2$'):
+            c.execute('UPDATE dtf_local.accounts SET password_hash=%s WHERE id=%s', (password_hash(body.password),account['id']))
+        if previous:
+            transfer_guest(c, previous, account['id'])
+            c.execute('DELETE FROM dtf_local.sessions WHERE id=%s', (previous['id'],))
+        new_session(c, response, account['id'])
+    audit('customer_login_success', account=str(account['id']))
+    return {'customer': account['profile']}
+
+@router.post('/api/account/logout')
+def logout(request: Request, response: Response):
+    previous = current(request)
+    if previous:
+        with db.connect() as c:
+            c.execute('DELETE FROM dtf_local.sessions WHERE id=%s', (previous['id'],))
+    response.delete_cookie('dtf_session', httponly=True, samesite='strict')
+    response.headers['Clear-Site-Data'] = '"storage"'
+    audit('customer_logout')
+    return {'ok': True}
+
+@router.get('/api/account/me')
+def me(identity=Depends(owner)):
+    with db.connect() as c:
+        row = c.execute('SELECT profile FROM dtf_local.accounts WHERE id=%s', (identity,)).fetchone()
+    return {'customer': row['profile'] if row else None}
+
+@router.get('/api/customer/orders')
+def orders(identity=Depends(owner)):
+    with db.connect() as c:
+        rows = c.execute('SELECT id,number,state,version,snapshot,created_at,updated_at FROM dtf_local.orders WHERE owner=%s ORDER BY created_at DESC', (identity,)).fetchall()
+        quotes = c.execute('''SELECT q.id,q.approved,q.approved_at,q.created_at FROM dtf_local.quotes q
+            LEFT JOIN dtf_local.orders o ON o.quote_id=q.id WHERE q.owner=%s AND o.id IS NULL ORDER BY q.created_at DESC''', (identity,)).fetchall()
+    return {'orders': rows, 'quotes': quotes, 'states': STATES}
+
+@router.get('/api/customer/orders/{oid}')
+def detail(oid: UUID, identity=Depends(owner)):
+    with db.connect() as c:
+        row = owned_order(c, oid, identity)
+        history = c.execute('SELECT from_state,to_state,reason,created_at FROM dtf_local.movements WHERE order_id=%s ORDER BY id', (oid,)).fetchall()
+        return {'id': row['id'], 'number': row['number'], 'state': row['state'], 'version': row['version'],
+                'snapshot': row['snapshot'], 'history': history, 'files': file_rows(c,oid)}
+
+@router.post('/api/customer/orders/{oid}/corrections')
+def correction(oid: UUID, body: ArtworkSubmission, identity=Depends(owner)):
+    with db.connect() as c:
+        order = owned_order(c, oid, identity, lock=True)
+        return submit_files(c,order,body,identity,'correction','customer')
+
+@router.get('/api/customer/orders/{oid}/files/{fid}/download')
+def customer_download(oid: UUID, fid: UUID, identity=Depends(owner)):
+    with db.connect() as c:
+        owned_order(c,oid,identity)
+        row = c.execute('SELECT u.* FROM dtf_local.order_files f JOIN dtf_local.uploads u ON u.id=f.upload_id WHERE f.id=%s AND f.order_id=%s AND f.active', (fid,oid)).fetchone()
+        if not row: raise HTTPException(404, 'Active file not found')
+        if row['expires_at'] <= datetime.now(timezone.utc): raise HTTPException(410, 'File expired')
+        require_clean(row)
+    return {'url':storage.download(row['object_key'],row['name']), 'name':row['name']}

app/api/health.py

@@ -0,0 +1,43 @@
+"""Health, session bootstrap and freight quoting."""
+import os
+
+from fastapi import APIRouter, HTTPException, Request, Response
+
+from ..core import db
+from ..core.auth import client_ip, owner, new_session, rate_limit
+from ..core.models import Freight
+from ..runtime import ENVIRONMENT, GUEST_SESSION_LIMIT, PART_BYTES, freight, storage
+
+router = APIRouter()
+
+@router.get('/health')
+@router.get('/api/health')
+def health():
+    try:
+        with db.connect() as c:
+            c.execute('SELECT 1')
+        storage.health()
+    except Exception:
+        raise HTTPException(503, 'Database or storage unavailable')
+    return {'status': 'ok', 'environment': ENVIRONMENT,
+            'storage': 'minio' if ENVIRONMENT == 'local' else 'r2', 'integrations': 'fake'}
+
+@router.get('/api/session')
+def session(request: Request, response: Response):
+    try:
+        session_id = owner(request)
+    except HTTPException:
+        # Per source, not per deployment: keyed on the environment name this was a
+        # single global bucket, so ~8 new visitors a minute exhausted it site-wide.
+        rate_limit('guest-sessions', client_ip(request), GUEST_SESSION_LIMIT, 900)
+        with db.connect() as c:
+            session_id = new_session(c, response)
+    return {'environment': ENVIRONMENT, 'cart_scope': str(session_id), 'part_bytes': PART_BYTES,
+            'max_upload_bytes': int(os.environ.get('MAX_UPLOAD_BYTES', '5368709120'))}
+
+@router.post('/api/freight')
+def quote_freight(body: Freight):
+    try:
+        return freight.quote(body.service, body.postal_code)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc))

app/api/operator.py

@@ -0,0 +1,146 @@
+"""Kanban: sign-in, the board, commercial review and card movement."""
+import hashlib
+import os
+import secrets
+from datetime import datetime, timedelta, timezone
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, Request, Response
+from psycopg.types.json import Jsonb
+
+from ..core import db
+from ..core.auth import (COOKIE_SECURE, DUMMY_PASSWORD_HASH, audit, client_ip, operator,
+                    password_matches, throttle)
+from ..core.models import Move, OperatorLogin, Review
+from ..core.pricing import price
+from ..runtime import (BOARD_FINISHED_LIMIT, BOARD_QUOTE_LIMIT, STATES, TRANSITIONS,
+                       enqueue, freight, quote_view, storage, upload_row)
+from ..scanning import require_clean
+
+router = APIRouter()
+
+@router.post('/api/operator/login')
+def operator_login(body: OperatorLogin, request: Request, response: Response):
+    email = body.email
+    throttle('operator:'+email, request)
+    with db.connect() as c:
+        account = c.execute('SELECT * FROM dtf_local.operators WHERE email=%s', (email,)).fetchone()
+        if not c.execute('SELECT 1 FROM dtf_local.operators WHERE active LIMIT 1').fetchone():
+            raise HTTPException(503, 'No Kanban operator account is configured')
+    # Comparable password work whether or not the account exists or is active.
+    stored = account['password_hash'] if account else DUMMY_PASSWORD_HASH
+    matches = password_matches(body.password, stored)
+    if not account or not account['active'] or not matches:
+        audit('operator_login_failed', ip=client_ip(request), operator=email)
+        raise HTTPException(401, 'Invalid operator login')
+    token = secrets.token_urlsafe(32)
+    with db.connect() as c:
+        previous = hashlib.sha256(request.cookies.get('dtf_operator','').encode()).hexdigest()
+        c.execute('DELETE FROM dtf_local.operator_sessions WHERE token_hash=%s', (previous,))
+        c.execute('INSERT INTO dtf_local.operator_sessions(token_hash,username) VALUES(%s,%s)',
+                  (hashlib.sha256(token.encode()).hexdigest(), email))
+        c.execute('UPDATE dtf_local.operators SET last_login_at=now() WHERE id=%s', (account['id'],))
+    response.set_cookie('dtf_operator', token, httponly=True, secure=COOKIE_SECURE,
+                        samesite='strict', path='/api/operator', max_age=28800)
+    audit('operator_login_success', operator=email)
+    return {'ok': True}
+
+@router.post('/api/operator/logout')
+def operator_logout(request: Request, response: Response):
+    with db.connect() as c:
+        digest = hashlib.sha256(request.cookies.get('dtf_operator','').encode()).hexdigest()
+        c.execute('DELETE FROM dtf_local.operator_sessions WHERE token_hash=%s', (digest,))
+    response.delete_cookie('dtf_operator', path='/api/operator', httponly=True,
+                           secure=COOKIE_SECURE, samesite='strict')
+    audit('operator_logout')
+    return {'ok': True}
+
+@router.get('/api/operator/board')
+def board(user=Depends(operator)):
+    with db.connect() as c:
+        # Everything still in progress, however old: an operator must never lose a
+        # card they can act on. Finished orders are terminal and only accumulate,
+        # so the board carries a recent window of them and reports the true total.
+        active = c.execute("SELECT * FROM dtf_local.orders WHERE state<>'fin' ORDER BY created_at").fetchall()
+        finished = c.execute("SELECT * FROM dtf_local.orders WHERE state='fin' ORDER BY created_at DESC LIMIT %s",
+                             (BOARD_FINISHED_LIMIT,)).fetchall()
+        finished_total = c.execute("SELECT count(*) AS n FROM dtf_local.orders WHERE state='fin'").fetchone()['n']
+        quotes = c.execute('''SELECT q.* FROM dtf_local.quotes q
+            LEFT JOIN dtf_local.orders o ON o.quote_id=q.id WHERE o.id IS NULL
+            ORDER BY q.created_at LIMIT %s''', (BOARD_QUOTE_LIMIT,)).fetchall()
+        return {'states': STATES, 'transitions': TRANSITIONS,
+                'orders': active + list(reversed(finished)),
+                'finished_shown': len(finished), 'finished_total': finished_total,
+                'quotes': [quote_view(c, q) for q in quotes],
+                'events': c.execute('SELECT * FROM dtf_local.outbox ORDER BY id DESC LIMIT 100').fetchall()}
+
+@router.post('/api/operator/quotes/{uid}/approve')
+def approve(uid: UUID, body: Review, user=Depends(operator)):
+    with db.connect() as c:
+        row = c.execute('SELECT * FROM dtf_local.quotes WHERE id=%s FOR UPDATE', (uid,)).fetchone()
+        if not row:
+            raise HTTPException(404, 'Quote not found')
+        if row['approved']:
+            raise HTTPException(409, 'Approved quotes are immutable; request a new quote')
+        draft = row['draft']
+        if len(body.items) != len(draft['items']):
+            raise HTTPException(422, 'Review must cover every item')
+        items = []
+        for item, original in zip(body.items, draft['items']):
+            if item.mode != original['mode'] or list(map(str,item.uploads)) != original['uploads']:
+                raise HTTPException(422, 'Product mode and attached files cannot change during review')
+            for upload_id in item.uploads:
+                require_clean(upload_row(c, upload_id, row['owner']))
+            items.append({**price(item.mode, str(item.metres), item.grade), 'uploads': original['uploads']})
+        quoted_freight = freight.quote(**draft['freight'])
+        approved = {'customer': draft['customer'], 'items': items, 'freight': quoted_freight,
+                    'total_cents': sum(i['total_cents'] for i in items)+quoted_freight['total_cents']}
+        c.execute('UPDATE dtf_local.quotes SET approved=%s, reviewed_by=%s, approved_at=now() WHERE id=%s', (Jsonb(approved),user,uid))
+    return approved
+
+@router.post('/api/operator/orders/{uid}/move')
+def move(uid: UUID, body: Move, user=Depends(operator)):
+    with db.connect() as c:
+        row = c.execute('SELECT * FROM dtf_local.orders WHERE id=%s FOR UPDATE', (uid,)).fetchone()
+        if not row:
+            raise HTTPException(404, 'Order not found')
+        if body.version != row['version']:
+            raise HTTPException(409, 'Order changed; refresh the board')
+        if body.state == row['state']:
+            return row
+        if body.state not in TRANSITIONS[row['state']]:
+            raise HTTPException(409, 'Move is not allowed from this state')
+        if body.state == 'cor' and not body.reason.strip():
+            raise HTTPException(422, 'Correction requires a reason')
+        if body.state in ('fil','imp'):
+            coverage = c.execute('SELECT DISTINCT f.item_index FROM dtf_local.order_files f JOIN dtf_local.uploads u ON u.id=f.upload_id WHERE f.order_id=%s AND f.kind=\'final\' AND f.active AND u.expires_at>now() AND u.purged_at IS NULL AND u.scan_state=\'clean\'', (uid,)).fetchall()
+            if {r['item_index'] for r in coverage} != set(range(len(row['snapshot']['items']))):
+                raise HTTPException(409, 'Approve a complete final-file set for every item before queueing')
+        if body.state == 'cor':
+            c.execute("UPDATE dtf_local.order_files SET active=false WHERE order_id=%s AND kind='final'", (uid,))
+        c.execute('INSERT INTO dtf_local.movements(order_id,from_state,to_state,operator,reason) VALUES(%s,%s,%s,%s,%s)',
+                  (uid,row['state'],body.state,user,body.reason))
+        changed = c.execute('UPDATE dtf_local.orders SET state=%s, version=version+1, updated_at=now() WHERE id=%s RETURNING *', (body.state,uid)).fetchone()
+        events = {'imp':'production_started','cor':'correction_needed','fin':'ready'}
+        if body.state in events:
+            for provider in ('tiny','whatsapp'):
+                enqueue(c, f'{uid}:{changed["version"]}:{provider}', provider,
+                        {'order_id':str(uid), 'number':row['number'], 'event':events[body.state], 'reason':body.reason,
+                         'customer_path': f'/portal.html?order={uid}'})
+    return changed
+
+@router.get('/api/operator/orders/{uid}/history')
+def history(uid: UUID, user=Depends(operator)):
+    with db.connect() as c:
+        return c.execute('SELECT * FROM dtf_local.movements WHERE order_id=%s ORDER BY id', (uid,)).fetchall()
+
+@router.get('/api/operator/uploads/{uid}/download')
+def download(uid: UUID, user=Depends(operator)):
+    with db.connect() as c:
+        row = c.execute('SELECT * FROM dtf_local.uploads WHERE id=%s AND complete', (uid,)).fetchone()
+        if not row:
+            raise HTTPException(404, 'Completed upload not found')
+        if row['expires_at'] <= datetime.now(timezone.utc):
+            raise HTTPException(410, 'Artwork retention expired')
+        require_clean(row)
+    return {'name':row['name'], 'url':storage.download(row['object_key'],row['name']), 'expires_in':300}

app/api/orders.py

@@ -0,0 +1,41 @@
+"""Paid orders. Local development payment only; no provider is wired yet."""
+from datetime import datetime, timedelta, timezone
+from uuid import UUID, uuid4
+
+from fastapi import APIRouter, Depends, HTTPException
+from psycopg.types.json import Jsonb
+
+from ..core import db
+from ..core.auth import owner
+from ..core.models import Pay
+from ..runtime import ENVIRONMENT, enqueue, payment, upload_row
+from ..scanning import require_clean
+
+router = APIRouter()
+
+@router.post('/api/orders/dev-paid')
+def dev_paid(body: Pay, session_id=Depends(owner)):
+    if ENVIRONMENT != 'local':
+        raise HTTPException(503, 'Checkout is not configured yet')
+    with db.connect() as c:
+        row = c.execute('SELECT * FROM dtf_local.quotes WHERE id=%s AND owner=%s FOR UPDATE', (body.quote_id,session_id)).fetchone()
+        if not row:
+            raise HTTPException(404, 'Quote not found')
+        existing = c.execute('SELECT * FROM dtf_local.orders WHERE quote_id=%s', (body.quote_id,)).fetchone()
+        if existing:
+            return existing
+        if not row['approved']:
+            raise HTTPException(409, 'An operator must verify length and grade first')
+        if row['approved_at'] < datetime.now(timezone.utc)-timedelta(hours=24):
+            raise HTTPException(409, 'Quote expired; request a new quote')
+        approved = row['approved']
+        for item in approved['items']:
+            for upload_id in item['uploads']:
+                require_clean(upload_row(c, UUID(upload_id), session_id))
+        paid = payment.pay(str(body.quote_id), approved['total_cents'])
+        result = c.execute('INSERT INTO dtf_local.orders(id,quote_id,owner,snapshot,payment) VALUES(%s,%s,%s,%s,%s) RETURNING *',
+            (uuid4(),body.quote_id,session_id,Jsonb(approved),Jsonb(paid))).fetchone()
+        for provider in ('tiny','whatsapp'):
+            enqueue(c, f"{result['id']}:paid:{provider}", provider,
+                    {'order_id': str(result['id']), 'number': result['number'], 'event': 'payment_approved', 'order': approved})
+    return result

app/api/quotes.py

@@ -0,0 +1,46 @@
+"""Quotes: the customer's cart, and the operator-reviewed version of it."""
+import hashlib
+import json
+from uuid import UUID, uuid4
+
+from fastapi import APIRouter, Depends, HTTPException
+from psycopg.types.json import Jsonb
+
+from ..core import db
+from ..core.auth import owner
+from ..core.models import QuoteRequest
+from ..runtime import freight, quote_view, upload_row
+from ..scanning import require_clean
+
+router = APIRouter()
+
+@router.post('/api/quotes')
+def create_quote(body: QuoteRequest, session_id=Depends(owner)):
+    draft = body.model_dump(mode='json', exclude={'request_key'})
+    digest = hashlib.sha256(json.dumps(draft, sort_keys=True).encode()).hexdigest()
+    try:
+        freight.quote(body.freight.service, body.freight.postal_code)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc))
+    with db.connect() as c:
+        for item in body.items:
+            for uid in item.uploads:
+                row = upload_row(c, uid, session_id)
+                if not row['complete']:
+                    raise HTTPException(409, 'Complete every upload before requesting a quote')
+                require_clean(row)
+        uid = uuid4()
+        c.execute('INSERT INTO dtf_local.quotes(id,owner,request_key,request_hash,draft) VALUES(%s,%s,%s,%s,%s) ON CONFLICT(owner,request_key) DO NOTHING',
+                  (uid, session_id, body.request_key, digest, Jsonb(draft)))
+        row = c.execute('SELECT * FROM dtf_local.quotes WHERE owner=%s AND request_key=%s', (session_id, body.request_key)).fetchone()
+        if row['request_hash'] != digest:
+            raise HTTPException(409, 'Request key already used for a different cart')
+    return {'id': row['id'], 'status': 'pending_review'}
+
+@router.get('/api/quotes/{uid}')
+def get_quote(uid: UUID, session_id=Depends(owner)):
+    with db.connect() as c:
+        row = c.execute('SELECT * FROM dtf_local.quotes WHERE id=%s AND owner=%s', (uid,session_id)).fetchone()
+        if not row:
+            raise HTTPException(404, 'Quote not found')
+        return quote_view(c, row)

app/api/uploads.py

@@ -0,0 +1,85 @@
+"""Customer uploads: reservation, signed parts, completion.
+
+The operator artwork routes reuse these functions directly with a different
+identity, which is why they are plain functions with a session_id argument.
+"""
+import math
+import os
+from uuid import UUID, uuid4
+
+from botocore.exceptions import ClientError
+from fastapi import APIRouter, Depends, HTTPException
+
+from ..core import db
+from ..core.auth import audit, owner, rate_limit
+from ..core.models import UploadStart
+from ..runtime import PART_BYTES, storage, upload_row
+
+router = APIRouter()
+
+@router.post('/api/uploads')
+def begin_upload(body: UploadStart, session_id=Depends(owner)):
+    if body.size > int(os.environ.get('MAX_UPLOAD_BYTES', '5368709120')):
+        raise HTTPException(413, 'File exceeds the upload limit')
+    uid = uuid4()
+    key = f'originals/{uid}'
+    rate_limit('upload-start', str(session_id), 60, 900)
+    with db.connect() as c:
+        # Serialize reservations across API processes, including changing guest IDs.
+        c.execute('SELECT pg_advisory_xact_lock(804208)')
+        usage = c.execute('''SELECT COALESCE(sum(size),0) AS total,
+            COALESCE(sum(size) FILTER(WHERE owner=%s),0) AS owned,
+            count(*) FILTER(WHERE owner=%s AND NOT complete) AS pending
+            FROM dtf_local.uploads WHERE purged_at IS NULL''', (session_id,session_id)).fetchone()
+        if (usage['total']+body.size > int(os.environ.get('STORAGE_QUOTA_BYTES','53687091200')) or
+            usage['owned']+body.size > int(os.environ.get('OWNER_UPLOAD_QUOTA_BYTES','10737418240')) or
+            usage['pending'] >= int(os.environ.get('MAX_PENDING_UPLOADS','10'))):
+            audit('upload_quota_rejected')
+            raise HTTPException(429, 'Local storage quota or pending upload limit reached')
+        multipart = storage.begin(key)
+        c.execute('INSERT INTO dtf_local.uploads(id,owner,name,size,object_key,multipart_id) VALUES(%s,%s,%s,%s,%s,%s)',
+                  (uid, session_id, body.name, body.size, key, multipart))
+    return {'id': uid, 'part_bytes': PART_BYTES}
+
+@router.get('/api/uploads/{uid}')
+def upload_status(uid: UUID, session_id=Depends(owner)):
+    with db.connect() as c:
+        row = upload_row(c, uid, session_id)
+    parts = [] if row['complete'] else storage.parts(row['object_key'], row['multipart_id'])
+    return {'id': uid, 'complete': row['complete'], 'part_bytes': PART_BYTES,
+            'scan_state':row['scan_state'], 'scan_reason':row['scan_reason'],
+            'parts': [p['PartNumber'] for p in parts]}
+
+@router.post('/api/uploads/{uid}/parts/{part}')
+def part_url(uid: UUID, part: int, session_id=Depends(owner)):
+    with db.connect() as c:
+        row = upload_row(c, uid, session_id)
+    if row['complete'] or not 1 <= part <= math.ceil(row['size'] / PART_BYTES):
+        raise HTTPException(409, 'Invalid part or completed upload')
+    size = min(PART_BYTES, row['size']-(part-1)*PART_BYTES)
+    return {'url': storage.part_url(row['object_key'], row['multipart_id'], part, size)}
+
+@router.post('/api/uploads/{uid}/complete')
+def complete_upload(uid: UUID, session_id=Depends(owner)):
+    with db.connect() as c:
+        row = upload_row(c, uid, session_id, lock=True)
+        if row['complete']:
+            return {'id': uid, 'complete': True}
+        try:
+            existing_size = storage.size(row['object_key'])
+        except ClientError as exc:
+            if exc.response['ResponseMetadata']['HTTPStatusCode'] != 404:
+                raise
+            existing_size = None
+        if existing_size is None:
+            parts = storage.parts(row['object_key'], row['multipart_id'])
+            expected = math.ceil(row['size'] / PART_BYTES)
+            if [p['PartNumber'] for p in parts] != list(range(1, expected+1)) or any(
+                p['Size'] != min(PART_BYTES, row['size'] - i*PART_BYTES) for i,p in enumerate(parts)):
+                raise HTTPException(409, 'Parts are missing or their sizes do not match')
+            storage.complete(row['object_key'], row['multipart_id'], parts)
+            existing_size = storage.size(row['object_key'])
+        if existing_size != row['size']:
+            raise HTTPException(409, 'Stored size differs from declared size')
+        c.execute('UPDATE dtf_local.uploads SET complete=true WHERE id=%s', (uid,))
+    return {'id': uid, 'complete': True}