Table of Contents#

1. Project Abstract
2. System Architecture
3. Technology Specifications
4. Data Model
5. Core Subsystems
   • Virtual File System (VFS)
   • Provider Adapter Interface
   • Synchronization Engine
   • Transfer Engine
6. Detailed Implementation Plan
7. Infrastructure and Deployment

Federated Virtual Filesystem is a storage orchestration middleware designed to aggregate distinct cloud storage providers (AWS S3, Google Drive, Nextcloud) into a single, federated Virtual File System (VFS). The system operates on a metadata-first principle: file metadata is indexed in a relational database (PostgreSQL) and a search engine (OpenSearch) to enable sub-second search and traversal without engaging provider APIs. Actual file content serves are proxied only when necessary, with a preference for direct-to-provider signed URLs. The backend utilizes an event-driven microservices architecture to handle high-throughput file synchronization and background transfers.

The architecture distinguishes between the Control Plane (synchronous API operations) and the Data Plane (asynchronous workers). These planes are decoupled via Apache Kafka (for event streaming) and Redis (for task queuing).

graph TD
    subgraph Clients
        Web[Web Client]
        CLI[CLI Agent]
    end

    subgraph Load_Balancer
        NGINX[Nginx Reverse Proxy]
    end

    subgraph Control_Plane
        API[FastAPI Gateway]
        Auth[Auth Service]
        ConnMgr[Connection Manager]
        VFSSvc[VFS Service]
        
        API --> Auth
        API --> ConnMgr
        API --> VFSSvc
    end

    subgraph Message_Bus_and_State
        Redis[(Redis - Cache/Broker)]
        Kafka{{Apache Kafka}}
        PG[(PostgreSQL - Metadata)]
        Search[(OpenSearch - Index)]
    end

    subgraph Data_Plane
        SyncWorker[Sync Worker]
        TransferWorker[Transfer Worker]
        IndexWorker[Index Worker]
    end

    subgraph External_Storage
        S3[AWS S3]
        GDrive[Google Drive]
        NC[Nextcloud]
    end

    %% Data Flows
    Clients --> NGINX --> API
    
    API -- Read/Write --> PG
    API -- Cache --> Redis
    API -- Publish Events --> Kafka

    Kafka -- Consume Commands --> SyncWorker
    Kafka -- Consume Jobs --> TransferWorker
    Kafka -- Consume File Events --> IndexWorker

    SyncWorker -- Delta Sync --> S3 & GDrive & NC
    SyncWorker -- Update State --> PG
    SyncWorker -- Publish Changes --> Kafka

    TransferWorker -- Stream Data --> S3 & GDrive & NC
    TransferWorker -- Update Progress --> Redis
    TransferWorker -- Finalize --> PG

    IndexWorker -- Upsert Documents --> Search

The database schema enforces referential integrity between users, their storage connections, and the virtual file tree.

erDiagram
    USERS ||--o{ STORAGE_CONNECTIONS : owns
    USERS ||--o{ VIRTUAL_NODES : owns
    
    STORAGE_CONNECTIONS ||--o{ VIRTUAL_NODES : contains
    STORAGE_CONNECTIONS ||--|| SYNC_STATES : maintains
    
    VIRTUAL_NODES ||--o{ VIRTUAL_NODES : parent_of

    USERS {
        uuid id PK
        string email
        string password_hash
        timestamp created_at
    }

    STORAGE_CONNECTIONS {
        uuid id PK
        uuid user_id FK
        enum provider_type "s3|gdrive|nextcloud"
        jsonb config "bucket, region, root_id"
        string encrypted_credentials
        enum status "active|error|syncing"
    }

    VIRTUAL_NODES {
        uuid id PK
        uuid parent_id FK "Self-referencing"
        uuid connection_id FK
        string name
        boolean is_folder
        ltree path "Materialized Path"
        string provider_id "External ID"
        string etag "Change detection"
        bigint size_bytes
        jsonb raw_metadata
        timestamp indexed_at
    }

    SYNC_STATES {
        uuid connection_id PK
        string cursor "PageToken or DeltaToken"
        timestamp last_successful_sync
    }

    JOBS {
        uuid id PK
        uuid user_id FK
        enum type "transfer|sync"
        enum status "pending|running|completed|failed"
        jsonb progress
    }

Virtual File System (VFS)#

The VFS abstracts external file structures. It uses the PostgreSQL ltree extension to allow efficient subtree queries (e.g., “Select all files inside Folder X”). Operations on the VFS (Move, Rename, Delete) must be transactionally atomic: updates are applied to the provider first, then committed to the database.

Provider Adapter Interface#

All storage providers must implement the BaseStorageAdapter abstract base class to ensure uniformity.

from abc import ABC
from typing import AsyncGenerator, List, Tuple

# Assuming these are defined elsewhere for type hints
class ListResult: pass
class DeltaResult: pass
class UploadSession: pass
class FileMetadata: pass

class BaseStorageAdapter(ABC):
    async def list_children(self, path: str, cursor: str) -> ListResult: ...
    async def get_changes(self, cursor: str) -> DeltaResult: ...
    async def download_stream(self, file_id: str, range: Tuple) -> AsyncGenerator: ...
    async def initiate_upload(self, name: str, size: int) -> UploadSession: ...
    async def move(self, file_id: str, new_parent: str) -> FileMetadata: ...

Synchronization Engine#

Synchronization is event-driven rather than poll-based where possible.

1. Ingest: Webhooks (Drive) or Event Notifications (S3) push events to the API.
2. Queue: Events are published to the sync.events Kafka topic.
3. Process: Sync Workers consume events, fetch metadata from the provider, and upsert records in the virtual_nodes table.
4. Index: Changes in virtual_nodes emit events to file.index, consumed by Index Workers for OpenSearch.

Transfer Engine#

Cross-provider transfers (e.g., Drive to S3) are handled by Transfer Workers.

• Streaming: Data is streamed in chunks (5MB-10MB) through the worker to avoid memory exhaustion.
• Resumability: Transfer progress is checkpointed in Redis. If a worker fails, another can resume from the last successful chunk.
• Multipart: Large files utilize S3 Multipart Upload and Drive Resumable Upload APIs.

This roadmap outlines the sequential development phases required to build the system.

Phase 1: Foundation and Infrastructure#

Objective: Establish the runtime environment and database schema.

1. Container Orchestration:
   • Create docker-compose.yml defining services: postgres, redis, kafka, zookeeper, opensearch.
   • Configure persistent volumes for data durability.
2. Database Definition:
   • Initialize SQLAlchemy Base and models (User, StorageConnection, VirtualNode).
   • Create Alembic migration scripts to enable ltree extension and create tables.
3. Authentication Service:
   • Implement POST /auth/register and POST /auth/login using JWT (JSON Web Tokens).
   • Implement get_current_user dependency for FastAPI.

Phase 2: Connection Management#

Objective: Allow users to link storage accounts.

1. Encryption Utilities:
   • Implement Fernet (symmetric encryption) helpers to store and retrieve provider credentials securely.
2. Connection API:
   • Implement POST /connections to store provider config (bucket names, regions).
   • Implement GET /connections to list active integrations.
3. Adapter Registry:
   • Create the ProviderRegistry pattern to instantiate the correct adapter class based on the provider_type column.

Phase 3: S3 Integration (Base Provider)#

Objective: Implement full read/write capability for S3-compatible stores.

1. S3 Adapter Implementation:
   • Implement list_children using boto3.
   • Implement generate_presigned_url for downloads and uploads.
2. VFS Browse API:
   • Implement GET /files endpoint. Logic: Query virtual_nodes by parent_id.
3. Initial Sync Logic:
   • Create a simple Celery task sync_s3_bucket that lists all objects in a bucket and populates virtual_nodes.

Phase 4: The Sync Engine#

Objective: Handle ongoing changes and multi-provider state.

1. Kafka Configuration:
   • Define topics: system.commands, file.events, file.index.
2. Google Drive Integration:
   • Implement GoogleDriveAdapter.
   • Implement OAuth2 flow handling to acquire and store Refresh Tokens.
   • Implement get_changes using the Drive Activity API.
3. Sync Worker:
   • Create a consumer service using aiokafka.
   • Logic: Listen for SYNC_REQUESTED. Iterate through provider changes. Upsert/Delete rows in Postgres.
4. Delta Sync:
   • Implement logic to store and retrieve sync_cursor tokens from the sync_states table.

Phase 5: Search and Indexing#

Objective: Enable fast search across the VFS.

1. OpenSearch Setup:
   • Define index mappings for files (name, size, MIME type, ltree path).
2. Index Worker:
   • Create a Kafka consumer for file.events.
   • Logic: On FILE_CREATED or FILE_MODIFIED, transform the Postgres row into a JSON document and push to OpenSearch.
3. Search API:
   • Implement GET /search. Logic: Query OpenSearch, return list of matching file IDs, enrich with latest metadata from Postgres if necessary.

Phase 6: Transfer Engine#

Objective: Enable moving files between providers.

1. Transfer API:
   • Implement POST /transfers. Accepts source_node_id and destination_folder_id.
   • Creates a Job record in Postgres with status PENDING.
2. Transfer Worker:
   • Implement process_transfer_job Celery task.
   • Logic:
     1. Acquire source stream (adapter.download_stream).
     2. Acquire destination upload session (adapter.initiate_upload).
     3. Pipe stream from source to destination.
     4. Update Job progress in Redis.
     5. On success, create new VirtualNode and mark Job as COMPLETED.

Phase 7: Optimization and Production Readiness#

Objective: Prepare for scale.

1. Rate Limiting:
   • Implement Redis-based sliding window rate limiter for external API calls to prevent quota exhaustion.
2. Refinement:
   • Implement error handling policies (exponential backoff) for failed sync tasks.
   • Add structural logging (JSON logs) for observability.

Docker Compose Configuration#

For local development and testing, the system creates the following services:

version: '3.8'
services:
  api:
    build: ./backend
    ports: ["8000:8000"]
    depends_on: [postgres, redis, kafka]
    environment:
      - DATABASE_URL=postgresql+asyncpg://user:pass@postgres/unicloud
      - KAFKA_BOOTSTRAP_SERVERS=kafka:9092

  sync_worker:
    build: ./backend
    command: celery -A app.worker.celery worker -Q sync
    depends_on: [redis, kafka]

  transfer_worker:
    build: ./backend
    command: celery -A app.worker.celery worker -Q transfer
    depends_on: [redis, kafka]

  postgres:
    image: postgres:15
    volumes: [pg_data:/var/lib/postgresql/data]

  redis:
    image: redis:7

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092

  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0

  opensearch:
    image: opensearchproject/opensearch:2.11.0
    ports: ["9200:9200"]