You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
247 lines
7.8 KiB
247 lines
7.8 KiB
3 years ago
|
import {FileMetadata, Filesystem, Stat} from "./Filesystem"
|
||
|
import * as ssh2 from "ssh2"
|
||
|
import * as path from "path"
|
||
|
import * as fs from "fs"
|
||
|
import ReadableStream = NodeJS.ReadableStream
|
||
|
import {UniversalPath} from "../path"
|
||
|
|
||
|
/**
|
||
|
* A Filesystem implementation that stores files on remote hosts via SFTP/SSH.
|
||
|
*/
|
||
|
export class SSHFilesystem extends Filesystem {
|
||
|
private _ssh?: ssh2.Client
|
||
|
|
||
|
constructor(
|
||
|
protected readonly baseConfig: { ssh: ssh2.ConnectConfig, baseDir: string },
|
||
|
) { super() }
|
||
|
|
||
|
getPrefix(): string {
|
||
|
return `sftp+${this.baseConfig.ssh.host}://`
|
||
|
}
|
||
|
|
||
|
async getStoreFileAsTemp(args: { storePath: string }): Promise<UniversalPath> {
|
||
|
const temp = this.tempName()
|
||
|
const write = fs.createWriteStream(temp)
|
||
|
const read = await this.getStoreFileAsStream(args)
|
||
|
|
||
|
return new Promise<UniversalPath>((res, rej) => {
|
||
|
write.on('finish', () => {
|
||
|
res(new UniversalPath(temp))
|
||
|
})
|
||
|
|
||
|
write.on('error', rej)
|
||
|
read.on('error', rej)
|
||
|
read.pipe(write)
|
||
|
})
|
||
|
}
|
||
|
|
||
|
async getStoreFileAsStream(args: { storePath: string }): Promise<ReadableStream> {
|
||
|
const sftp = await this.getSFTP()
|
||
|
return sftp.createReadStream(this.storePath(args.storePath))
|
||
|
}
|
||
|
|
||
|
async putLocalFile(args: { localPath: string; storePath: string; mimeType?: string; tags?: string[]; tag?: string }): Promise<void> {
|
||
|
const read = fs.createReadStream(args.localPath)
|
||
|
const write = await this.putStoreFileAsStream({storePath: args.storePath})
|
||
|
|
||
|
// write the metadata first
|
||
|
const sftp = await this.getSFTP()
|
||
|
await sftp.writeFile(this.metadataPath(args.storePath), JSON.stringify({
|
||
|
mimeType: args.mimeType,
|
||
|
tags: this._normalizeTags(args.tag, args.tags)
|
||
|
}))
|
||
|
|
||
|
// pipe the local file to the store
|
||
|
await new Promise<void>((res, rej) => {
|
||
|
write.on('finish', () => res())
|
||
|
write.on('error', rej)
|
||
|
read.on('error', rej)
|
||
|
read.pipe(write)
|
||
|
})
|
||
|
}
|
||
|
|
||
|
async putStoreFileAsStream(args: { storePath: string }): Promise<NodeJS.WritableStream> {
|
||
|
const sftp = await this.getSFTP()
|
||
|
return sftp.createWriteStream(this.storePath(args.storePath))
|
||
|
}
|
||
|
|
||
|
async mkdir(args: { storePath: string }): Promise<void> {
|
||
|
const sftp = await this.getSFTP()
|
||
|
await new Promise<void>((res, rej) => {
|
||
|
sftp.mkdir(this.storePath(args.storePath), err => {
|
||
|
if ( err ) rej(err)
|
||
|
else res()
|
||
|
})
|
||
|
})
|
||
|
}
|
||
|
|
||
|
async remove(args: { storePath: string; recursive?: boolean }): Promise<void> {
|
||
|
const sftp = await this.getSFTP()
|
||
|
|
||
|
await new Promise<void>((res, rej) => {
|
||
|
sftp.unlink(this.storePath(args.storePath), err => {
|
||
|
if ( err ) return rej(err)
|
||
|
else {
|
||
|
sftp.unlink(this.metadataPath(args.storePath), err2 => {
|
||
|
if ( err2 ) rej(err2)
|
||
|
else res()
|
||
|
})
|
||
|
}
|
||
|
})
|
||
|
})
|
||
|
}
|
||
|
|
||
|
async stat(args: { storePath: string }): Promise<Stat> {
|
||
|
const sftp = await this.getSFTP()
|
||
|
|
||
|
try {
|
||
|
const stat = await new Promise<any>((res, rej) => {
|
||
|
sftp.stat(this.storePath(args.storePath), (err, stat) => {
|
||
|
if ( err ) return rej(err)
|
||
|
res(stat)
|
||
|
})
|
||
|
})
|
||
|
|
||
|
const jsonStream = sftp.createReadStream(this.metadataPath(args.storePath))
|
||
|
const json = await this.streamToString(jsonStream)
|
||
|
const meta = JSON.parse(json)
|
||
|
|
||
|
return {
|
||
|
path: new UniversalPath(args.storePath, this),
|
||
|
exists: true,
|
||
|
sizeInBytes: stat.size,
|
||
|
mimeType: meta.mimeType,
|
||
|
tags: meta.tags,
|
||
|
accessed: stat.atime,
|
||
|
modified: stat.mtime,
|
||
|
created: stat.ctime,
|
||
|
}
|
||
|
} catch (e) {
|
||
|
return {
|
||
|
path: new UniversalPath(args.storePath, this),
|
||
|
exists: false,
|
||
|
sizeInBytes: 0,
|
||
|
tags: []
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
|
||
|
async touch(args: { storePath: string }): Promise<void> {
|
||
|
const sftp = await this.getSFTP()
|
||
|
|
||
|
return new Promise<void>((res, rej) => {
|
||
|
const storePath = this.storePath(args.storePath)
|
||
|
const time = new Date()
|
||
|
sftp.utimes(storePath, time, time, err => {
|
||
|
if ( err ) {
|
||
|
sftp.open(storePath, 'w', (err2, fd) => {
|
||
|
if ( err2 ) return rej(err2)
|
||
|
sftp.close(fd, err3 => {
|
||
|
if ( err3 ) return rej(err3)
|
||
|
res()
|
||
|
})
|
||
|
})
|
||
|
} else {
|
||
|
res()
|
||
|
}
|
||
|
})
|
||
|
})
|
||
|
}
|
||
|
|
||
|
async getMetadata(storePath: string): Promise<FileMetadata> {
|
||
|
try {
|
||
|
const sftp = await this.getSFTP()
|
||
|
return new Promise((res, rej) => {
|
||
|
sftp.readFile(this.metadataPath(storePath), (err, buffer) => {
|
||
|
if ( err ) rej(err)
|
||
|
res(JSON.parse(buffer.toString('utf-8')))
|
||
|
})
|
||
|
})
|
||
|
} catch (e) {
|
||
|
return {
|
||
|
tags: []
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
|
||
|
async setMetadata(storePath: string, meta: FileMetadata): Promise<void> {
|
||
|
const sftp = await this.getSFTP()
|
||
|
const metaPath = this.metadataPath(storePath)
|
||
|
await new Promise<void>((res, rej) => {
|
||
|
sftp.writeFile(metaPath, JSON.stringify(meta), err => {
|
||
|
if ( err ) rej(err)
|
||
|
else res()
|
||
|
})
|
||
|
})
|
||
|
}
|
||
|
|
||
|
async close(): Promise<void> {
|
||
|
await this._ssh?.end()
|
||
|
}
|
||
|
|
||
|
/**
|
||
|
* Using the provided `hostConfig`, create and cache the SSH connection to the host.
|
||
|
* If a connection already exists, re-use it.
|
||
|
*/
|
||
|
async getSSH(): Promise<ssh2.Client> {
|
||
|
if ( this._ssh ) return this._ssh
|
||
|
|
||
|
return new Promise((res, rej) => {
|
||
|
const client = new ssh2.Client()
|
||
|
|
||
|
client.on('ready', () => {
|
||
|
this._ssh = client
|
||
|
res(client)
|
||
|
}).connect(this.baseConfig.ssh)
|
||
|
|
||
|
client.on('error', rej)
|
||
|
})
|
||
|
}
|
||
|
|
||
|
/**
|
||
|
* Using `getSSH()`, create a new SFTP helper based on that connection.
|
||
|
*/
|
||
|
async getSFTP(): Promise<ssh2.SFTPWrapper> {
|
||
|
const ssh = await this.getSSH()
|
||
|
|
||
|
return new Promise((res, rej) => {
|
||
|
ssh.sftp((err, sftp) => {
|
||
|
if ( err ) rej(err)
|
||
|
else res(sftp)
|
||
|
})
|
||
|
})
|
||
|
}
|
||
|
|
||
|
/**
|
||
|
* Resolve the given store path to an absolute path on the remote filesystem.
|
||
|
* @param storePath
|
||
|
* @protected
|
||
|
*/
|
||
|
protected storePath(storePath: string): string {
|
||
|
return path.resolve(this.baseConfig.baseDir, 'data', storePath)
|
||
|
}
|
||
|
|
||
|
/**
|
||
|
* Resolve the given store path to an absolute path of a metadata file on the remote filesystem.
|
||
|
* @param storePath
|
||
|
* @protected
|
||
|
*/
|
||
|
protected metadataPath(storePath: string): string {
|
||
|
return path.resolve(this.baseConfig.baseDir, 'meta', storePath + '.json')
|
||
|
}
|
||
|
|
||
|
/**
|
||
|
* Given a readable stream, cast it to a string.
|
||
|
* @param stream
|
||
|
* @protected
|
||
|
*/
|
||
|
protected streamToString(stream: NodeJS.ReadableStream): Promise<string> {
|
||
|
const chunks: Buffer[] = [];
|
||
|
return new Promise((resolve, reject) => {
|
||
|
stream.on('data', (chunk) => chunks.push(Buffer.from(chunk)))
|
||
|
stream.on('error', (err) => reject(err))
|
||
|
stream.on('end', () => resolve(Buffer.concat(chunks).toString('utf8')))
|
||
|
})
|
||
|
}
|
||
|
}
|