Dropbox / Cloud File Sync

-- Core file identity (one row per logical file)
CREATE TABLE files (
  id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id     UUID NOT NULL REFERENCES users(id),
  path        TEXT NOT NULL,                -- relative path within user's Dropbox
  is_deleted  BOOLEAN NOT NULL DEFAULT false,
  created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
  updated_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
  UNIQUE (user_id, path)
);

-- Immutable version history (one row per upload/edit)
CREATE TABLE file_versions (
  id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  file_id     UUID NOT NULL REFERENCES files(id),
  version_num INT  NOT NULL,
  size_bytes  BIGINT NOT NULL,
  vector_clock JSONB NOT NULL,             -- { device_id: lamport_clock }
  created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
  UNIQUE (file_id, version_num)
);

-- Content-addressed chunk registry (global dedup)
CREATE TABLE block_store (
  sha256      CHAR(64) PRIMARY KEY,        -- hex SHA-256 of raw bytes
  size_bytes  INT NOT NULL,
  s3_key      TEXT NOT NULL,               -- s3://bucket/{sha256}
  ref_count   INT NOT NULL DEFAULT 0,      -- GC when 0
  created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Mapping: version → ordered list of chunks
CREATE TABLE chunks (
  version_id  UUID NOT NULL REFERENCES file_versions(id),
  seq_num     INT  NOT NULL,               -- chunk order within file
  sha256      CHAR(64) NOT NULL REFERENCES block_store(sha256),
  PRIMARY KEY (version_id, seq_num)
);

-- Conflict log: divergent versions that need user resolution
CREATE TABLE conflicts (
  id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  file_id     UUID NOT NULL REFERENCES files(id),
  version_a   UUID NOT NULL REFERENCES file_versions(id),
  version_b   UUID NOT NULL REFERENCES file_versions(id),
  resolved_at TIMESTAMPTZ,
  created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

service SyncState {
  rpc GetDeviceState(DeviceStateReq) returns (DeviceState);
  rpc UpdateCursor(UpdateCursorReq) returns (google.protobuf.Empty);
  rpc BroadcastChange(ChangeEvent) returns (google.protobuf.Empty);
}

// ─── Domain Types ───────────────────────────────────────────────

interface Chunk {
  sha256: string;      // hex SHA-256 of raw bytes
  seqNum: number;      // order within file (0-indexed)
  sizeBytes: number;
}

interface FileVersion {
  id: string;
  fileId: string;
  versionNum: number;
  chunks: Chunk[];
  vectorClock: Record<string, number>;  // deviceId → lamport clock
  sizeBytes: number;
  createdAt: Date;
}

interface Delta {
  missingHashes: string[];
  chunkManifest: Chunk[];
  vectorClock: Record<string, number>;
}

// ─── ChunkUploadService ─────────────────────────────────────────

class ChunkUploadService {
  constructor(
    private readonly s3: S3Client,
    private readonly db: Pool,
  ) {}

  /** Returns subset of hashes that are NOT in block_store */
  async checkChunks(hashes: string[]): Promise<string[]> {
    const rows = await this.db.query(
      'SELECT sha256 FROM block_store WHERE sha256 = ANY($1)',
      [hashes]
    );
    const present = new Set(rows.rows.map((r: any) => r.sha256));
    return hashes.filter(h => !present.has(h));
  }

  async storeChunk(sha256: string, data: Buffer): Promise<void> {
    const key = `chunks/${sha256}`;
    await this.s3.putObject({ Bucket: BUCKET, Key: key, Body: data });
    await this.db.query(
      `INSERT INTO block_store (sha256, s3_key, size_bytes)
       VALUES ($1, $2, $3) ON CONFLICT (sha256) DO NOTHING`,
      [sha256, key, data.byteLength]
    );
  }

  async getPresignedURL(sha256: string): Promise<string> {
    return getSignedUrl(this.s3, new GetObjectCommand({
      Bucket: BUCKET, Key: `chunks/${sha256}`
    }), { expiresIn: 3600 });
  }
}

// ─── SyncStateManager ───────────────────────────────────────────

class SyncStateManager {
  constructor(
    private readonly redis: Redis,
    private readonly kafka: Producer,
  ) {}

  async getDeviceCursor(deviceId: string): Promise<number> {
    const val = await this.redis.get(`device:${deviceId}:cursor`);
    return val ? parseInt(val) : 0;
  }

  async updateCursor(deviceId: string, offset: number): Promise<void> {
    await this.redis.set(`device:${deviceId}:cursor`, offset, 'EX', 2592000);
  }

  async broadcastChange(event: {
    userId: string; fileId: string; versionId: string; operation: 'create' | 'update' | 'delete';
  }): Promise<void> {
    await this.kafka.send({
      topic: 'file.changed',
      messages: [{ key: event.userId, value: JSON.stringify(event) }],
    });
  }
}

// ─── Client-side Delta Computer ─────────────────────────────────

class DeltaComputer {
  /** Split file into ~4MB chunks using content-defined chunking */
  computeChunks(buffer: Buffer): Chunk[] {
    const CHUNK_SIZE = 4 * 1024 * 1024;
    const chunks: Chunk[] = [];
    let offset = 0, seq = 0;
    while (offset < buffer.length) {
      const slice = buffer.subarray(offset, offset + CHUNK_SIZE);
      chunks.push({
        sha256: sha256hex(slice),
        seqNum: seq++,
        sizeBytes: slice.length,
      });
      offset += CHUNK_SIZE;
    }
    return chunks;
  }
}