Dropbox / Google Drive: файловая синхронизация

<?php

declare(strict_types=1);

final readonly class ChunkUploadRequest
{
    public function __construct(
        public string $sessionId,
        public int $offset,
        public string $sha256,   // of this chunk
        public int $size,
        public string $payload,  // binary body
    ) {}
}

final readonly class CommitUploadRequest
{
    /**
     * @param list<string> $chunkHashes  // ordered list of chunk sha256
     */
    public function __construct(
        public string $sessionId,
        public string $path,
        public string $fileSha256,
        public int $totalSize,
        public array $chunkHashes,
        public ?string $parentVersionId = null, // for conflict detection
    ) {}
}

final readonly class FileMetadata
{
    /**
     * @param list<string> $chunkHashes
     */
    public function __construct(
        public string $id,
        public string $path,
        public int $size,
        public string $sha256,
        public array $chunkHashes,
        public string $versionId,
        public \DateTimeImmutable $modifiedAt,
        public string $modifiedBy,
    ) {}
}

package api

import "time"

type ChunkUploadRequest struct {
    SessionID string `json:"session_id"`
    Offset    int64  `json:"offset"`
    SHA256    string `json:"sha256"`
    Size      int64  `json:"size"`
}

type CommitUploadRequest struct {
    SessionID       string   `json:"session_id"`
    Path            string   `json:"path"`
    FileSHA256      string   `json:"file_sha256"`
    TotalSize       int64    `json:"total_size"`
    ChunkHashes     []string `json:"chunk_hashes"`
    ParentVersionID string   `json:"parent_version_id,omitempty"`
}

type FileMetadata struct {
    ID           string    `json:"id"`
    Path         string    `json:"path"`
    Size         int64     `json:"size"`
    SHA256       string    `json:"sha256"`
    ChunkHashes  []string  `json:"chunk_hashes"`
    VersionID    string    `json:"version_id"`
    ModifiedAt   time.Time `json:"modified_at"`
    ModifiedBy   string    `json:"modified_by"`
}

namespace Dropbox.Api.Contracts;

using System.Text.Json.Serialization;

// Records give value equality and immutability out of the box.
public sealed record ChunkUploadRequest(
    [property: JsonPropertyName("session_id")] string SessionId,
    [property: JsonPropertyName("offset")] long Offset,
    [property: JsonPropertyName("sha256")] string Sha256,   // of this chunk
    [property: JsonPropertyName("size")] long Size,
    [property: JsonIgnore] ReadOnlyMemory<byte> Payload);   // binary body, not serialized to JSON

public sealed record CommitUploadRequest(
    [property: JsonPropertyName("session_id")] string SessionId,
    [property: JsonPropertyName("path")] string Path,
    [property: JsonPropertyName("file_sha256")] string FileSha256,
    [property: JsonPropertyName("total_size")] long TotalSize,
    // Ordered list of chunk sha256.
    [property: JsonPropertyName("chunk_hashes")] IReadOnlyList<string> ChunkHashes,
    // For conflict detection.
    [property: JsonPropertyName("parent_version_id")] string? ParentVersionId = null);

public sealed record FileMetadata(
    [property: JsonPropertyName("id")] string Id,
    [property: JsonPropertyName("path")] string Path,
    [property: JsonPropertyName("size")] long Size,
    [property: JsonPropertyName("sha256")] string Sha256,
    [property: JsonPropertyName("chunk_hashes")] IReadOnlyList<string> ChunkHashes,
    [property: JsonPropertyName("version_id")] string VersionId,
    [property: JsonPropertyName("modified_at")] DateTimeOffset ModifiedAt,
    [property: JsonPropertyName("modified_by")] string ModifiedBy);

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime

@dataclass(frozen=True, slots=True)
class ChunkUploadRequest:
    session_id: str
    offset: int
    sha256: str  # of this chunk
    size: int
    payload: bytes  # binary body

@dataclass(frozen=True, slots=True)
class CommitUploadRequest:
    session_id: str
    path: str
    file_sha256: str
    total_size: int
    chunk_hashes: tuple[str, ...]  # ordered list of chunk sha256
    parent_version_id: str | None = None  # for conflict detection

@dataclass(frozen=True, slots=True)
class FileMetadata:
    id: str
    path: str
    size: int
    sha256: str
    chunk_hashes: tuple[str, ...]
    version_id: str
    modified_at: datetime
    modified_by: str

package chunker

import (
    "crypto/sha256"
    "encoding/hex"
    "io"
)

const ChunkSize = 4 * 1024 * 1024 // 4 MB

// Chunk represents a single content-addressable block.
type Chunk struct {
    SHA256 string
    Data   []byte
    Size   int
}

// SplitFile reads a stream and yields fixed-size chunks.
// For CDC (content-defined chunking), replace with a rolling-hash splitter.
func SplitFile(r io.Reader) ([]Chunk, error) {
    chunks := make([]Chunk, 0, 16)
    buf := make([]byte, ChunkSize)
    for {
        n, err := io.ReadFull(r, buf)
        if n > 0 {
            sum := sha256.Sum256(buf[:n])
            chunk := Chunk{
                SHA256: hex.EncodeToString(sum[:]),
                Data:   append([]byte(nil), buf[:n]...),
                Size:   n,
            }
            chunks = append(chunks, chunk)
        }
        if err == io.EOF || err == io.ErrUnexpectedEOF {
            return chunks, nil
        }
        if err != nil {
            return nil, err
        }
    }
}

// DedupUploader uploads only missing blocks and returns the ordered hash list.
type BlockAPI interface {
    Exists(sha string) (bool, error)
    Upload(sha string, data []byte) error
}

func Upload(api BlockAPI, chunks []Chunk) ([]string, error) {
    hashes := make([]string, len(chunks))
    for i, c := range chunks {
        exists, err := api.Exists(c.SHA256)
        if err != nil {
            return nil, err
        }
        if !exists {
            if err := api.Upload(c.SHA256, c.Data); err != nil {
                return nil, err
            }
        }
        hashes[i] = c.SHA256
    }
    return hashes, nil
}

namespace Dropbox.Chunker;

using System.Buffers;
using System.Runtime.CompilerServices;
using System.Security.Cryptography;

public static class ChunkerDefaults
{
    public const int ChunkSize = 4 * 1024 * 1024; // 4 MB
}

// Chunk represents a single content-addressable block.
public sealed record Chunk(string Sha256, ReadOnlyMemory<byte> Data, int Size);

public static class FileSplitter
{
    // SplitFile reads a stream and yields fixed-size chunks.
    // For CDC (content-defined chunking), replace with a rolling-hash splitter.
    // IAsyncEnumerable streams chunks so a 50 GB file never sits in memory.
    public static async IAsyncEnumerable<Chunk> SplitFileAsync(
        Stream source,
        [EnumeratorCancellation] CancellationToken ct = default)
    {
        var buffer = ArrayPool<byte>.Shared.Rent(ChunkerDefaults.ChunkSize);
        try
        {
            while (true)
            {
                // ReadAtLeast fills the buffer across partial reads; a short read means EOF.
                int n = await source.ReadAtLeastAsync(
                    buffer.AsMemory(0, ChunkerDefaults.ChunkSize),
                    ChunkerDefaults.ChunkSize,
                    throwOnEndOfStream: false,
                    ct);

                if (n > 0)
                {
                    Span<byte> digest = stackalloc byte[SHA256.HashSizeInBytes];
                    SHA256.HashData(buffer.AsSpan(0, n), digest);
                    yield return new Chunk(
                        Convert.ToHexString(digest).ToLowerInvariant(),
                        buffer.AsSpan(0, n).ToArray(),
                        n);
                }

                if (n < ChunkerDefaults.ChunkSize)
                {
                    yield break;
                }
            }
        }
        finally
        {
            ArrayPool<byte>.Shared.Return(buffer);
        }
    }
}

// DedupUploader uploads only missing blocks and returns the ordered hash list.
public interface IBlockApi
{
    Task<bool> ExistsAsync(string sha, CancellationToken ct = default);
    Task UploadAsync(string sha, ReadOnlyMemory<byte> data, CancellationToken ct = default);
}

public static class DedupUploader
{
    public static async Task<IReadOnlyList<string>> UploadAsync(
        IBlockApi api,
        IAsyncEnumerable<Chunk> chunks,
        CancellationToken ct = default)
    {
        var hashes = new List<string>();
        await foreach (var chunk in chunks.WithCancellation(ct))
        {
            if (!await api.ExistsAsync(chunk.Sha256, ct))
            {
                await api.UploadAsync(chunk.Sha256, chunk.Data, ct);
            }
            hashes.Add(chunk.Sha256);
        }
        return hashes;
    }
}

from __future__ import annotations

import asyncio
import hashlib
from collections.abc import Iterable, Iterator
from dataclasses import dataclass
from typing import BinaryIO, Protocol

CHUNK_SIZE = 4 * 1024 * 1024  # 4 MB

@dataclass(frozen=True, slots=True)
class Chunk:
    """Chunk represents a single content-addressable block."""

    sha256: str
    data: bytes
    size: int

def split_file(stream: BinaryIO) -> Iterator[Chunk]:
    """Read a stream and yield fixed-size chunks.

    For CDC (content-defined chunking), replace with a rolling-hash splitter.
    """
    while True:
        # read() may return short blocks on pipes/sockets -- accumulate until
        # the chunk is full so boundaries stay identical to the Go version.
        parts: list[bytes] = []
        remaining = CHUNK_SIZE
        while remaining > 0:
            block = stream.read(remaining)
            if not block:
                break
            parts.append(block)
            remaining -= len(block)

        buf = b"".join(parts)
        if not buf:
            return
        yield Chunk(sha256=hashlib.sha256(buf).hexdigest(), data=buf, size=len(buf))
        if remaining > 0:  # short final chunk -- EOF reached
            return

class BlockAPI(Protocol):
    async def exists(self, sha: str) -> bool: ...
    async def upload(self, sha: str, data: bytes) -> None: ...

async def upload(
    api: BlockAPI,
    chunks: Iterable[Chunk],
    concurrency: int = 10,
) -> list[str]:
    """Upload only missing blocks and return the ordered hash list.

    Unlike the Go version this fans out with a semaphore -- parallel chunk
    upload is what saturates a 1 Gbps client link.
    """
    ordered = list(chunks)
    limiter = asyncio.Semaphore(concurrency)

    async def put(chunk: Chunk) -> None:
        async with limiter:
            if not await api.exists(chunk.sha256):
                await api.upload(chunk.sha256, chunk.data)

    await asyncio.gather(*(put(c) for c in ordered))
    return [c.sha256 for c in ordered]

<?php

declare(strict_types=1);

final class UploadCommitService
{
    public function __construct(
        private readonly \PDO $db,
        private readonly BlockStore $blocks,
        private readonly ChangeLogPublisher $publisher,
    ) {}

    public function commit(CommitUploadRequest $req, string $userId, string $namespaceId): FileMetadata
    {
        $this->db->beginTransaction();
        try {
            // 1. Verify all chunks exist in block store.
            $missing = $this->blocks->missingChunks($req->chunkHashes);
            if (!empty($missing)) {
                throw new \DomainException('chunks missing: ' . implode(',', $missing));
            }

            // 2. Conflict detection: latest version must match parent_version_id.
            $node = $this->findOrCreateNode($namespaceId, $req->path);

            $latest = $this->db->prepare(
                'SELECT id FROM file_versions WHERE node_id = :nid ORDER BY modified_at DESC LIMIT 1'
            );
            $latest->execute(['nid' => $node['id']]);
            $latestId = $latest->fetchColumn();

            if ($latestId !== false && $latestId !== $req->parentVersionId) {
                // Conflict: create parallel version with "(conflicted copy)" suffix
                $node = $this->createConflictedCopy($node, $namespaceId, $userId);
            }

            // 3. Insert version.
            $versionId = $this->insertVersion(
                nodeId: $node['id'],
                namespaceId: $namespaceId,
                size: $req->totalSize,
                sha256: $req->fileSha256,
                chunks: $req->chunkHashes,
                userId: $userId,
            );

            // 4. Bump refcounts.
            $this->blocks->incrementRefs($req->chunkHashes);

            // 5. Append to change log.
            $this->publisher->publish($namespaceId, $node['id'], ChangeType::Modified, $versionId);

            $this->db->commit();

            return $this->buildMetadata($node, $versionId, $req);
        } catch (\Throwable $e) {
            $this->db->rollBack();
            throw $e;
        }
    }
}

namespace Dropbox.Upload;

using System.Data;
using Dapper;
using Npgsql;

public sealed class UploadCommitService(
    NpgsqlDataSource dataSource,
    IBlockStore blocks,
    IChangeLogPublisher publisher)
{
    public async Task<FileMetadata> CommitAsync(
        CommitUploadRequest req,
        string userId,
        string namespaceId,
        CancellationToken ct = default)
    {
        await using var db = await dataSource.OpenConnectionAsync(ct);
        await using var tx = await db.BeginTransactionAsync(IsolationLevel.ReadCommitted, ct);
        try
        {
            // 1. Verify all chunks exist in block store.
            var missing = await blocks.MissingChunksAsync(req.ChunkHashes, ct);
            if (missing.Count > 0)
            {
                throw new DomainException($"chunks missing: {string.Join(',', missing)}");
            }

            // 2. Conflict detection: latest version must match parent_version_id.
            var node = await FindOrCreateNodeAsync(db, tx, namespaceId, req.Path, ct);

            var latestId = await db.ExecuteScalarAsync<string?>(new CommandDefinition(
                "SELECT id FROM file_versions WHERE node_id = @nid ORDER BY modified_at DESC LIMIT 1",
                new { nid = node.Id }, tx, cancellationToken: ct));

            if (latestId is not null && latestId != req.ParentVersionId)
            {
                // Conflict: create parallel version with "(conflicted copy)" suffix
                node = await CreateConflictedCopyAsync(db, tx, node, namespaceId, userId, ct);
            }

            // 3. Insert version.
            var versionId = await InsertVersionAsync(
                db, tx,
                nodeId: node.Id,
                namespaceId: namespaceId,
                size: req.TotalSize,
                sha256: req.FileSha256,
                chunks: req.ChunkHashes,
                userId: userId,
                ct: ct);

            // 4. Bump refcounts.
            await blocks.IncrementRefsAsync(req.ChunkHashes, ct);

            // 5. Append to change log.
            await publisher.PublishAsync(namespaceId, node.Id, ChangeType.Modified, versionId, ct);

            await tx.CommitAsync(ct);

            return BuildMetadata(node, versionId, req);
        }
        catch
        {
            await tx.RollbackAsync(ct);
            throw;
        }
    }
}

from __future__ import annotations

import asyncpg

class DomainError(Exception):
    """Domain-level failure, distinct from infrastructure errors."""

class UploadCommitService:
    def __init__(
        self,
        pool: asyncpg.Pool,
        blocks: BlockStore,
        publisher: ChangeLogPublisher,
    ) -> None:
        self._pool = pool
        self._blocks = blocks
        self._publisher = publisher

    async def commit(
        self,
        req: CommitUploadRequest,
        user_id: str,
        namespace_id: str,
    ) -> FileMetadata:
        async with self._pool.acquire() as conn:
            # asyncpg's transaction context rolls back on any exception,
            # so no explicit try/except rollback is needed here.
            async with conn.transaction():
                # 1. Verify all chunks exist in block store.
                missing = await self._blocks.missing_chunks(req.chunk_hashes)
                if missing:
                    raise DomainError(f"chunks missing: {','.join(missing)}")

                # 2. Conflict detection: latest version must match parent_version_id.
                node = await self._find_or_create_node(conn, namespace_id, req.path)

                latest_id = await conn.fetchval(
                    "SELECT id FROM file_versions "
                    "WHERE node_id = $1 ORDER BY modified_at DESC LIMIT 1",
                    node.id,
                )

                if latest_id is not None and latest_id != req.parent_version_id:
                    # Conflict: create parallel version with "(conflicted copy)" suffix
                    node = await self._create_conflicted_copy(
                        conn, node, namespace_id, user_id
                    )

                # 3. Insert version.
                version_id = await self._insert_version(
                    conn,
                    node_id=node.id,
                    namespace_id=namespace_id,
                    size=req.total_size,
                    sha256=req.file_sha256,
                    chunks=req.chunk_hashes,
                    user_id=user_id,
                )

                # 4. Bump refcounts.
                await self._blocks.increment_refs(req.chunk_hashes)

                # 5. Append to change log.
                await self._publisher.publish(
                    namespace_id, node.id, ChangeType.MODIFIED, version_id
                )

            return self._build_metadata(node, version_id, req)