val userPasswordAuthService = new UserPasswordAuthService {
  override def authenticate(credentials: UsernamePassword): UserPrincipal = {
    // Perform authentication
    UserPrincipalWithCredentials(credentials)
  }
  // Return true if this module should handle the given credentials
  override def accepts(credentials: UsernamePassword): Boolean = true
}

val directoryDataSourceModule = new DirectoryDataSourceModule
// Set the root directory of the data source
directoryDataSourceModule.setRootDirectory(new File(baseDir))

val modules = Array(
  directoryDataSourceModule,
  new BaseDftpModule,
  new UserPasswordAuthModule(userPasswordAuthService),
  new PutModule,
  new ActionModule
)

// Start the DFTP server
dftpServer = DftpServer.start(DftpServerConfig("0.0.0.0"), modules)

// Connect using anonymous login
val dftpClient = DftpClient.connect("dftp://0.0.0.0:3101", Credentials.ANONYMOUS)

// Connect using username/password authentication
val dftpClient = DftpClient.connect(
  "dftp://0.0.0.0:3101",
  UsernamePassword("user", "password")
)

// Retrieve a DataFrame from a remote resource
val df = dftpClient.get("dftp://0.0.0.0:3101/resourcePath")

// Transformations
val dfMap = df.map(row => row._1).limit(10)
val dfFilter = df.filter(row => row._1 > 100).limit(10)
val dfSelect = df.select("col0", "col1").limit(10)

df.collect()                       // Retrieve all data
df.foreach(row => process(row))     // Process each row individually

val directoryCatalogModule = new DirectoryCatalogModule()
directoryCatalogModule.setRootDirectory(new File(baseDir))

val permissionService = new PermissionService {
  override def accepts(user: UserPrincipal): Boolean = true

  // Validate whether the user has permission to perform certain operations
  override def checkPermission(user: UserPrincipal, dataFrameName: String, opList: List[DataOperationType]): Boolean =
    user.asInstanceOf[UserPrincipalWithCredentials].credentials match {
      case Credentials.ANONYMOUS => false
      case UsernamePassword(username, password) => true
    }
}

val modules = Array(
  new BaseDftpModule,
  new DacpCookModule,
  new DacpCatalogModule,
  new DirectoryDataSourceModule,
  directoryCatalogModule,
  new UserPasswordAuthModule(userPasswordAuthService), // Refer to DFTP server implementation
  new PermissionServiceModule(permissionService)
)

// Start the DACP server
server = DftpServer.start(
  DftpServerConfig("0.0.0.0", 3101)
    .withProtocolScheme("dacp"),
  modules
)

// Connect using anonymous login
val dacpClient = DacpClient.connect("dacp://0.0.0.0:3101", Credentials.ANONYMOUS)

// Connect using username/password authentication
val dacpClient = DacpClient.connect(
  "dacp://0.0.0.0:3101",
  UsernamePassword("user", "password")
)

// List all available datasets
dacpClient.listDataSetNames()

// Retrieve dataset metadata
dacpClient.getDataSetMetaData("dataSet")

// List all data frames under a dataset
dacpClient.listDataFrameNames("dataSet")

// Retrieve metadata for a specific data frame
dacpClient.getDataFrameMetaData("/dataFramePath")

// Retrieve schema of a data frame
dacpClient.getSchema("/dataFramePath")

// Retrieve data frame size
dacpClient.getDataFrameSize("/dataFramePath")

// Retrieve the DataFrameDocument associated with the data frame
dacpClient.getDocument("/dataFramePath")

// Retrieve statistical information of a data frame
dacpClient.getStatistics("/dataFramePath")

// Define a custom transformer
val udf = new Transformer11 {
  override def transform(dataFrame: DataFrame): DataFrame = {
    dataFrame.limit(5)
  }
}

// Define a simple transformation DAG
val recipe = Flow(
  Map(
    "A" -> SourceNode("/dataFrame"),
    "B" -> udf
  ),
  Map(
    "A" -> Seq("B")
  )
)

// Execute the transformation flow
val dfs: ExecutionResult = dacpClient.cook(recipe)
dfs.single().foreach(println)