Initial commit
This commit is contained in:
580
server/utils/drop-store.ts
Normal file
580
server/utils/drop-store.ts
Normal file
@@ -0,0 +1,580 @@
|
||||
import { createHash, randomBytes, timingSafeEqual } from "node:crypto";
|
||||
import {
|
||||
mkdir,
|
||||
readFile,
|
||||
readdir,
|
||||
rename,
|
||||
rm,
|
||||
stat,
|
||||
writeFile,
|
||||
open as openFile,
|
||||
} from "node:fs/promises";
|
||||
import { resolve } from "node:path";
|
||||
import {
|
||||
ROOM_ID_PATTERN,
|
||||
expiresInMs,
|
||||
validateReservation,
|
||||
validateRoomId,
|
||||
} from "../../shared/utils/drop-contract";
|
||||
import {
|
||||
CHUNK_SIZE,
|
||||
LEASE_IDLE_MS,
|
||||
UPLOAD_RESERVATION_MS,
|
||||
type PublicDescriptor,
|
||||
type ReserveRequest,
|
||||
type StoredDrop,
|
||||
type StoredLease,
|
||||
} from "../../shared/types/drop";
|
||||
const ALPHABET = "23456789ABCDEFGHJKLMNPQRSTUVWXYZ";
|
||||
const productionRoot = resolve(
|
||||
process.env.QUICKDROP_STORAGE_DIR || ".data/quickdrop",
|
||||
);
|
||||
export class DropStoreError extends Error {
|
||||
constructor(
|
||||
public code:
|
||||
| "not-found"
|
||||
| "expired"
|
||||
| "conflict"
|
||||
| "incomplete"
|
||||
| "exhausted"
|
||||
| "forbidden",
|
||||
message?: string,
|
||||
) {
|
||||
super(message || code);
|
||||
}
|
||||
}
|
||||
interface LockEntry {
|
||||
tail: Promise<void>;
|
||||
waiters: number;
|
||||
}
|
||||
const locks = new Map<string, LockEntry>();
|
||||
async function withLock<T>(
|
||||
key: string,
|
||||
operation: () => Promise<T>,
|
||||
): Promise<T> {
|
||||
let entry = locks.get(key);
|
||||
if (!entry) {
|
||||
entry = { tail: Promise.resolve(), waiters: 0 };
|
||||
locks.set(key, entry);
|
||||
}
|
||||
entry.waiters++;
|
||||
const previous = entry.tail;
|
||||
let release!: () => void;
|
||||
entry.tail = new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
await previous;
|
||||
try {
|
||||
return await operation();
|
||||
} finally {
|
||||
release();
|
||||
entry.waiters--;
|
||||
if (entry.waiters === 0 && locks.get(key) === entry) locks.delete(key);
|
||||
}
|
||||
}
|
||||
function hash(value: string): string {
|
||||
return createHash("sha256").update(value).digest("hex");
|
||||
}
|
||||
function sameHash(left: string, right: string): boolean {
|
||||
const a = Buffer.from(left, "hex");
|
||||
const b = Buffer.from(right, "hex");
|
||||
return a.length === b.length && timingSafeEqual(a, b);
|
||||
}
|
||||
function bearer(): string {
|
||||
return randomBytes(32).toString("base64url");
|
||||
}
|
||||
function nowValue(clock: (() => number) | { now(): number }): number {
|
||||
return typeof clock === "function" ? clock() : clock.now();
|
||||
}
|
||||
function ensureId(id: string) {
|
||||
try {
|
||||
validateRoomId(id);
|
||||
} catch {
|
||||
throw new DropStoreError("not-found", "Drop not found.");
|
||||
}
|
||||
return id;
|
||||
}
|
||||
|
||||
export interface DropStoreOptions {
|
||||
root?: string;
|
||||
clock?: (() => number) | { now(): number };
|
||||
}
|
||||
const TEMP_FILE_GRACE_MS = 60 * 60 * 1000;
|
||||
const METADATALESS_GRACE_MS = 60 * 60 * 1000;
|
||||
export function pathsFor(id: string, root = productionRoot) {
|
||||
ensureId(id);
|
||||
const directory = resolve(root, id);
|
||||
return {
|
||||
directory,
|
||||
metadata: resolve(directory, "metadata.json"),
|
||||
manifest: resolve(directory, "manifest.bin"),
|
||||
chunks: resolve(directory, "chunks"),
|
||||
};
|
||||
}
|
||||
async function atomicWrite(path: string, data: Uint8Array | string) {
|
||||
const temp = `${path}.tmp-${process.pid}-${randomBytes(6).toString("hex")}`;
|
||||
await writeFile(temp, data, { mode: 0o600 });
|
||||
try {
|
||||
const handle = await openFile(temp, "r");
|
||||
await handle.sync();
|
||||
await handle.close();
|
||||
} catch {
|
||||
/* fsync is best effort on unusual filesystems */
|
||||
}
|
||||
await rename(temp, path);
|
||||
}
|
||||
function chunkPath(paths: ReturnType<typeof pathsFor>, index: number) {
|
||||
return resolve(paths.chunks, `${String(index).padStart(10, "0")}.bin`);
|
||||
}
|
||||
|
||||
export class DropStore {
|
||||
readonly root: string;
|
||||
readonly clock: (() => number) | { now(): number };
|
||||
constructor(options: DropStoreOptions = {}) {
|
||||
this.root = resolve(options.root || productionRoot);
|
||||
this.clock = options.clock || (() => Date.now());
|
||||
}
|
||||
async ensure() {
|
||||
await mkdir(this.root, { recursive: true, mode: 0o700 });
|
||||
}
|
||||
async reserve(
|
||||
input: ReserveRequest,
|
||||
): Promise<{ id: string; uploadExpiresAt: number; access: string }> {
|
||||
validateReservation(input);
|
||||
await this.ensure();
|
||||
for (let attempt = 0; attempt < 100; attempt++) {
|
||||
const id = Array.from(
|
||||
randomBytes(32),
|
||||
(byte) => ALPHABET[byte % ALPHABET.length],
|
||||
).join("");
|
||||
const paths = pathsFor(id, this.root);
|
||||
try {
|
||||
await mkdir(paths.directory, { recursive: false, mode: 0o700 });
|
||||
await mkdir(paths.chunks, { mode: 0o700 });
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === "EEXIST") continue;
|
||||
throw error;
|
||||
}
|
||||
const createdAt = nowValue(this.clock);
|
||||
const access = bearer();
|
||||
const metadata: StoredDrop = {
|
||||
version: 2,
|
||||
id,
|
||||
state: "uploading",
|
||||
createdAt,
|
||||
uploadExpiresAt: createdAt + UPLOAD_RESERVATION_MS,
|
||||
expiresInMs: expiresInMs(input.expiresIn),
|
||||
expiresAt: null,
|
||||
maxReads: input.maxReads,
|
||||
readsClaimed: 0,
|
||||
chunkSize: input.chunkSize,
|
||||
chunkCount: input.chunkCount,
|
||||
noncePrefix: input.noncePrefix,
|
||||
leases: {},
|
||||
accessHash: hash(access),
|
||||
};
|
||||
try {
|
||||
await atomicWrite(paths.metadata, JSON.stringify(metadata));
|
||||
return { id, uploadExpiresAt: metadata.uploadExpiresAt, access };
|
||||
} catch (error) {
|
||||
await rm(paths.directory, { recursive: true, force: true });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
throw new Error("Could not allocate a drop id.");
|
||||
}
|
||||
async readMetadata(id: string): Promise<StoredDrop | null> {
|
||||
const paths = pathsFor(id, this.root);
|
||||
try {
|
||||
return JSON.parse(await readFile(paths.metadata, "utf8")) as StoredDrop;
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
private async metadataLocked(id: string): Promise<StoredDrop> {
|
||||
const metadata = await this.readMetadata(id);
|
||||
if (!metadata || metadata.version !== 2 || metadata.id !== id)
|
||||
throw new DropStoreError("not-found", "Drop not found.");
|
||||
const now = nowValue(this.clock);
|
||||
if (
|
||||
(metadata.state === "uploading" && metadata.uploadExpiresAt <= now) ||
|
||||
(metadata.state === "ready" &&
|
||||
metadata.expiresAt !== null &&
|
||||
metadata.expiresAt <= now)
|
||||
) {
|
||||
await rm(pathsFor(id, this.root).directory, {
|
||||
recursive: true,
|
||||
force: true,
|
||||
});
|
||||
throw new DropStoreError("expired", "Drop expired.");
|
||||
}
|
||||
return metadata;
|
||||
}
|
||||
private async save(id: string, metadata: StoredDrop) {
|
||||
await atomicWrite(
|
||||
pathsFor(id, this.root).metadata,
|
||||
JSON.stringify(metadata),
|
||||
);
|
||||
}
|
||||
async putChunk(
|
||||
id: string,
|
||||
index: number,
|
||||
body: Uint8Array,
|
||||
): Promise<"created" | "same"> {
|
||||
ensureId(id);
|
||||
return withLock(`${this.root}\0${id}`, async () => {
|
||||
const metadata = await this.metadataLocked(id);
|
||||
if (metadata.state !== "uploading")
|
||||
throw new DropStoreError("conflict", "Drop is already finalized.");
|
||||
if (
|
||||
!Number.isSafeInteger(index) ||
|
||||
index < 0 ||
|
||||
index >= metadata.chunkCount
|
||||
)
|
||||
throw new DropStoreError("conflict", "Invalid chunk index.");
|
||||
const paths = pathsFor(id, this.root);
|
||||
const target = chunkPath(paths, index);
|
||||
const expectedFull = metadata.chunkSize + 16;
|
||||
if (
|
||||
body.byteLength < 16 ||
|
||||
body.byteLength > expectedFull ||
|
||||
(index < metadata.chunkCount - 1 && body.byteLength !== expectedFull)
|
||||
)
|
||||
throw new DropStoreError("conflict", "Invalid chunk length.");
|
||||
try {
|
||||
const existing = await readFile(target);
|
||||
if (
|
||||
existing.byteLength === body.byteLength &&
|
||||
sameHash(
|
||||
hash(existing.toString("base64")),
|
||||
hash(Buffer.from(body).toString("base64")),
|
||||
)
|
||||
)
|
||||
return "same";
|
||||
throw new DropStoreError(
|
||||
"conflict",
|
||||
"Chunk already exists with different contents.",
|
||||
);
|
||||
} catch (error) {
|
||||
if (error instanceof DropStoreError) throw error;
|
||||
if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error;
|
||||
}
|
||||
const temp = resolve(
|
||||
paths.directory,
|
||||
`.chunk-${index}-${randomBytes(5).toString("hex")}`,
|
||||
);
|
||||
await writeFile(temp, body, { mode: 0o600 });
|
||||
try {
|
||||
const handle = await openFile(temp, "r");
|
||||
await handle.sync();
|
||||
await handle.close();
|
||||
} catch {}
|
||||
await rename(temp, target);
|
||||
return "created";
|
||||
});
|
||||
}
|
||||
async finalize(id: string, manifest: Uint8Array): Promise<StoredDrop> {
|
||||
ensureId(id);
|
||||
if (manifest.byteLength <= 16)
|
||||
throw new DropStoreError("conflict", "Invalid manifest.");
|
||||
return withLock(`${this.root}\0${id}`, async () => {
|
||||
const metadata = await this.metadataLocked(id);
|
||||
const paths = pathsFor(id, this.root);
|
||||
if (metadata.state === "ready") {
|
||||
const current = await readFile(paths.manifest);
|
||||
if (Buffer.compare(current, Buffer.from(manifest)) !== 0)
|
||||
throw new DropStoreError(
|
||||
"conflict",
|
||||
"Manifest conflicts with finalized drop.",
|
||||
);
|
||||
return metadata;
|
||||
}
|
||||
for (let index = 0; index < metadata.chunkCount; index++) {
|
||||
let info;
|
||||
try {
|
||||
info = await stat(chunkPath(paths, index));
|
||||
} catch {
|
||||
throw new DropStoreError("incomplete", "Some chunks are missing.");
|
||||
}
|
||||
if (
|
||||
info.size < 16 ||
|
||||
(index < metadata.chunkCount - 1 &&
|
||||
info.size !== metadata.chunkSize + 16)
|
||||
)
|
||||
throw new DropStoreError("incomplete", "Chunk geometry is invalid.");
|
||||
}
|
||||
let previous: Buffer | null = null;
|
||||
try {
|
||||
previous = await readFile(paths.manifest);
|
||||
} catch {}
|
||||
if (previous && Buffer.compare(previous, Buffer.from(manifest)) !== 0)
|
||||
throw new DropStoreError(
|
||||
"conflict",
|
||||
"Manifest conflicts with staged data.",
|
||||
);
|
||||
if (!previous) await atomicWrite(paths.manifest, manifest);
|
||||
metadata.state = "ready";
|
||||
metadata.expiresAt = nowValue(this.clock) + metadata.expiresInMs;
|
||||
metadata.manifestBytes = manifest.byteLength;
|
||||
metadata.chunkCipherBytes = [];
|
||||
for (let index = 0; index < metadata.chunkCount; index++)
|
||||
metadata.chunkCipherBytes.push(
|
||||
(await stat(chunkPath(paths, index))).size,
|
||||
);
|
||||
await this.save(id, metadata);
|
||||
return metadata;
|
||||
});
|
||||
}
|
||||
async descriptor(id: string, access: string): Promise<PublicDescriptor> {
|
||||
ensureId(id);
|
||||
return withLock(`${this.root}\0${id}`, async () => {
|
||||
const metadata = await this.metadataLocked(id);
|
||||
if (metadata.state !== "ready")
|
||||
throw new DropStoreError("not-found", "Drop is not ready.");
|
||||
if (!sameHash(hash(access), metadata.accessHash))
|
||||
throw new DropStoreError("forbidden", "Invalid access capability.");
|
||||
const manifest = await readFile(pathsFor(id, this.root).manifest);
|
||||
return {
|
||||
version: 2,
|
||||
id,
|
||||
chunkSize: metadata.chunkSize,
|
||||
chunkCount: metadata.chunkCount,
|
||||
noncePrefix: metadata.noncePrefix,
|
||||
chunkCipherBytes: metadata.chunkCipherBytes || [],
|
||||
manifest: manifest.toString("base64url"),
|
||||
expiresAt: new Date(metadata.expiresAt!).toISOString(),
|
||||
canClaim:
|
||||
metadata.maxReads === null ||
|
||||
metadata.readsClaimed < metadata.maxReads,
|
||||
maxReads: metadata.maxReads,
|
||||
};
|
||||
});
|
||||
}
|
||||
async claim(
|
||||
id: string,
|
||||
access: string,
|
||||
existing?: string,
|
||||
): Promise<{
|
||||
token: string;
|
||||
leaseExpiresAt: number;
|
||||
expiresAt: number;
|
||||
reused: boolean;
|
||||
}> {
|
||||
ensureId(id);
|
||||
return withLock(`${this.root}\0${id}`, async () => {
|
||||
const metadata = await this.metadataLocked(id);
|
||||
if (metadata.state !== "ready")
|
||||
throw new DropStoreError("not-found", "Drop is not ready.");
|
||||
if (!sameHash(hash(access), metadata.accessHash))
|
||||
throw new DropStoreError("forbidden", "Invalid access capability.");
|
||||
const now = nowValue(this.clock);
|
||||
for (const [tokenHash, lease] of Object.entries(metadata.leases)) {
|
||||
if (lease.expiresAt <= now) delete metadata.leases[tokenHash];
|
||||
}
|
||||
if (existing && metadata.leases[hash(existing)]) {
|
||||
const lease = metadata.leases[hash(existing)]!;
|
||||
lease.lastSeenAt = now;
|
||||
lease.expiresAt = Math.min(now + LEASE_IDLE_MS, metadata.expiresAt!);
|
||||
await this.save(id, metadata);
|
||||
return {
|
||||
token: existing,
|
||||
leaseExpiresAt: lease.expiresAt,
|
||||
expiresAt: metadata.expiresAt!,
|
||||
reused: true,
|
||||
};
|
||||
}
|
||||
if (
|
||||
metadata.maxReads !== null &&
|
||||
metadata.readsClaimed >= metadata.maxReads
|
||||
) {
|
||||
await this.save(id, metadata);
|
||||
throw new DropStoreError("exhausted", "No new reads remain.");
|
||||
}
|
||||
const token = bearer();
|
||||
const lease: StoredLease = {
|
||||
createdAt: now,
|
||||
lastSeenAt: now,
|
||||
expiresAt: Math.min(now + LEASE_IDLE_MS, metadata.expiresAt!),
|
||||
};
|
||||
metadata.leases[hash(token)] = lease;
|
||||
metadata.readsClaimed++;
|
||||
await this.save(id, metadata);
|
||||
return {
|
||||
token,
|
||||
leaseExpiresAt: lease.expiresAt,
|
||||
expiresAt: metadata.expiresAt!,
|
||||
reused: false,
|
||||
};
|
||||
});
|
||||
}
|
||||
async readChunk(
|
||||
id: string,
|
||||
index: number,
|
||||
token: string,
|
||||
): Promise<{ handle: Awaited<ReturnType<typeof openFile>>; size: number }> {
|
||||
ensureId(id);
|
||||
return withLock(`${this.root}\0${id}`, async () => {
|
||||
const metadata = await this.metadataLocked(id);
|
||||
const tokenHash = hash(token);
|
||||
if (metadata.state !== "ready" || !metadata.leases[tokenHash])
|
||||
throw new DropStoreError("forbidden", "Invalid lease.");
|
||||
const lease = metadata.leases[tokenHash]!;
|
||||
const now = nowValue(this.clock);
|
||||
if (lease.expiresAt <= now) {
|
||||
delete metadata.leases[hash(token)];
|
||||
await this.save(id, metadata);
|
||||
throw new DropStoreError("forbidden", "Lease expired.");
|
||||
}
|
||||
if (
|
||||
!Number.isSafeInteger(index) ||
|
||||
index < 0 ||
|
||||
index >= metadata.chunkCount
|
||||
)
|
||||
throw new DropStoreError("not-found", "Chunk not found.");
|
||||
lease.lastSeenAt = now;
|
||||
lease.expiresAt = Math.min(now + LEASE_IDLE_MS, metadata.expiresAt!);
|
||||
await this.save(id, metadata);
|
||||
const handle = await openFile(
|
||||
chunkPath(pathsFor(id, this.root), index),
|
||||
"r",
|
||||
);
|
||||
try {
|
||||
const info = await handle.stat();
|
||||
return { handle, size: info.size };
|
||||
} catch (error) {
|
||||
await handle.close().catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
});
|
||||
}
|
||||
async complete(id: string, token: string): Promise<void> {
|
||||
ensureId(id);
|
||||
return withLock(`${this.root}\0${id}`, async () => {
|
||||
const existing = await this.readMetadata(id);
|
||||
if (!existing) return;
|
||||
const now = nowValue(this.clock);
|
||||
if (existing.expiresAt !== null && existing.expiresAt <= now) {
|
||||
await rm(pathsFor(id, this.root).directory, {
|
||||
recursive: true,
|
||||
force: true,
|
||||
});
|
||||
return;
|
||||
}
|
||||
const metadata = await this.metadataLocked(id);
|
||||
const key = hash(token);
|
||||
if (!metadata.leases[key])
|
||||
throw new DropStoreError("forbidden", "Invalid lease.");
|
||||
delete metadata.leases[key];
|
||||
if (
|
||||
metadata.maxReads !== null &&
|
||||
metadata.readsClaimed >= metadata.maxReads &&
|
||||
Object.keys(metadata.leases).length === 0
|
||||
)
|
||||
await rm(pathsFor(id, this.root).directory, {
|
||||
recursive: true,
|
||||
force: true,
|
||||
});
|
||||
else await this.save(id, metadata);
|
||||
});
|
||||
}
|
||||
async delete(id: string) {
|
||||
ensureId(id);
|
||||
await withLock(`${this.root}\0${id}`, () =>
|
||||
rm(pathsFor(id, this.root).directory, { recursive: true, force: true }),
|
||||
);
|
||||
}
|
||||
async cleanup() {
|
||||
await this.ensure();
|
||||
const entries = await readdir(this.root, { withFileTypes: true });
|
||||
for (const entry of entries) {
|
||||
if (!entry.isDirectory() || !ROOM_ID_PATTERN.test(entry.name)) continue;
|
||||
await withLock(`${this.root}\0${entry.name}`, async () => {
|
||||
try {
|
||||
const dropDirectory = pathsFor(entry.name, this.root).directory;
|
||||
const now = nowValue(this.clock);
|
||||
for (const temporary of await readdir(dropDirectory)) {
|
||||
if (
|
||||
!temporary.startsWith(".chunk-") &&
|
||||
!temporary.includes(".tmp-")
|
||||
)
|
||||
continue;
|
||||
const temporaryPath = resolve(dropDirectory, temporary);
|
||||
const temporaryInfo = await stat(temporaryPath).catch(() => null);
|
||||
if (
|
||||
temporaryInfo &&
|
||||
now - temporaryInfo.mtimeMs >= TEMP_FILE_GRACE_MS
|
||||
)
|
||||
await rm(temporaryPath, { force: true });
|
||||
}
|
||||
const metadata = await this.readMetadata(entry.name);
|
||||
if (!metadata) {
|
||||
const directoryInfo = await stat(dropDirectory);
|
||||
if (now - directoryInfo.mtimeMs >= METADATALESS_GRACE_MS)
|
||||
await rm(dropDirectory, { recursive: true, force: true });
|
||||
return;
|
||||
}
|
||||
if (
|
||||
(metadata.state === "uploading" &&
|
||||
metadata.uploadExpiresAt <= now) ||
|
||||
(metadata.state === "ready" &&
|
||||
metadata.expiresAt !== null &&
|
||||
metadata.expiresAt <= now)
|
||||
) {
|
||||
await rm(dropDirectory, { recursive: true, force: true });
|
||||
return;
|
||||
}
|
||||
let changed = false;
|
||||
for (const [key, lease] of Object.entries(metadata.leases))
|
||||
if (
|
||||
lease.expiresAt <= now ||
|
||||
now - lease.lastSeenAt >= LEASE_IDLE_MS
|
||||
) {
|
||||
delete metadata.leases[key];
|
||||
changed = true;
|
||||
}
|
||||
if (
|
||||
metadata.maxReads !== null &&
|
||||
metadata.readsClaimed >= metadata.maxReads &&
|
||||
Object.keys(metadata.leases).length === 0
|
||||
) {
|
||||
await rm(dropDirectory, { recursive: true, force: true });
|
||||
return;
|
||||
}
|
||||
if (changed) await this.save(entry.name, metadata);
|
||||
} catch (error) {
|
||||
console.error(
|
||||
`[quickdrop] ignoring malformed drop ${entry.name}`,
|
||||
error,
|
||||
);
|
||||
await rm(resolve(this.root, entry.name), {
|
||||
recursive: true,
|
||||
force: true,
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
export const dropStore = new DropStore();
|
||||
export async function ensureStore() {
|
||||
return dropStore.ensure();
|
||||
}
|
||||
export async function cleanupExpiredDrops() {
|
||||
return dropStore.cleanup();
|
||||
}
|
||||
export async function deleteDrop(id: string) {
|
||||
return dropStore.delete(id);
|
||||
}
|
||||
export async function createRoomId() {
|
||||
return (
|
||||
await dropStore.reserve({
|
||||
version: 2,
|
||||
chunkSize: CHUNK_SIZE,
|
||||
chunkCount: 1,
|
||||
noncePrefix: randomBytes(8).toString("base64url"),
|
||||
expiresIn: "24h",
|
||||
maxReads: 1,
|
||||
})
|
||||
).id;
|
||||
}
|
||||
Reference in New Issue
Block a user