hub/docs/drives-phase-2.md

Phase 2: Storage Service

Objective

Implement the backend service layer that handles all disk operations: uploading files to hashed paths, resolving file metadata, and serving content. This module is backend/src/drive_service.py.

Concepts

Database-First Metadata

Files are database records pointing to disk locations. Access paths are constructed lazily via a resolve function so internal path formats can change without breaking external references.

Minimal Backend Layers

Access flows through three strict stages:

1. Auth: Bearer token validation.
2. DB Lookup: Verify ownership (user_id) and Drive existence.
3. Serve: If requesting raw content, pass the resolved disk path to Quart's send functions.

The router stays thin; the database handles all authorization before any file system calls.

Tasks

2.1 Implement drive_service.py

Create backend/src/drive_service.py with the following core functions:

import os
from pathlib import Path
import secrets
from datetime import timezone
import database as db

DRIVES_ROOT = None

async def init_drives(data_directory):
    """Initialize DRIVES_ROOT global from app startup."""
    global DRIVES_ROOT
    DRIVES_ROOT = os.path.join(data_directory, "drives")
    os.makedirs(DRIVES_ROOT, exist_ok=True)

def _generate_disk_hash():
    """Generate a random hex string for filesystem-level filenames."""
    return secrets.token_hex(8)

async def upload_file(user_id, drive_name, readable_filename, stream, mime_type=None):
    """Write an incoming stream to disk and register it in the DB. Returns file ID."""
    # 1. Locate or create Drive
    drive = await db.Drive.get_or_none(user_id=user_id, name=drive_name)
    if not drive:
        drive = await db.Drive.create(user_id=user_id, name=drive_name)
    
    # 2. Prepare disk path
    disk_hash = _generate_disk_hash()
    drive_dir = os.path.join(DRIVES_ROOT, str(user_id), drive_name)
    os.makedirs(drive_dir, exist_ok=True)
    file_path = os.path.join(drive_dir, disk_hash)
    
    # 3. Write stream and calculate size
    size = 0
    with open(file_path, "wb") as f:
        while chunk := stream.read(8192):
            f.write(chunk)
            size += len(chunk)
            
    # 4. Register in database
    record = await db.DriveFile.create(
        drive_id=drive.id,
        filename=readable_filename,
        disk_hash=disk_hash,
        mime_type=mime_type or "application/octet-stream",
        size_bytes=size,
    )
    return record.id

async def get_drive_file_metadata(file_id):
    """Return JSON metadata for frontend display."""
    file = await db.DriveFile.get_or_none(id=file_id)
    if not file:
        raise Exception("File not found")
    return {
        "id": file.id,
        "name": file.filename,
        "size": file.size_bytes,
        "mime": file.mime_type,
        "created": file.created_at.isoformat(),
        "accessed": file.accessed_at.isoformat() if file.accessed_at else None
    }

async def resolve_drive_file_for_serving(file_id, current_user_id):
    """Resolve file to absolute disk path after permission check. Returns (path, mime)."""
    # Eager load drive and verify ownership
    file = await db.DriveFile.filter(id=file_id).select_related("drive__user").get_or_none()
    if not file:
        raise FileNotFoundError("File not found")
    if file.drive.user_id != current_user_id:
        raise PermissionError("Access denied")
        
    # Update access time asynchronously
    from asyncio import create_task
    create_task(db.DriveFile.filter(id=file_id).update(accessed_at=timezone.now()))
    
    abs_path = os.path.join(DRIVES_ROOT, str(file.drive.user_id), file.drive.name, file.disk_hash)
    if not os.path.exists(abs_path):
        raise FileNotFoundError("File missing on disk")
        
    return abs_path, file.mime_type