581 lines
18 KiB
TypeScript
581 lines
18 KiB
TypeScript
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;
|
|
}
|