Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .dockerignore
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
.git
**/node_modules
dist
storage
**/dist
**/storage
**/coverage
**/.nyc_output
**/.cache
Expand Down
5 changes: 5 additions & 0 deletions .env.example
Original file line number Diff line number Diff line change
@@ -1,6 +1,11 @@
HOST=127.0.0.1
PORT=3190
DATABASE_PATH=./storage/usage.sqlite
# Export staging uses anonymous files, removed on close or process exit.
# Default: ./storage (the writable volume in Docker); choose disk, not /tmp's
# small container tmpfs. Per process: two exports, one per reader, 512 MiB each,
# with a 30-second total deadline. Downloads do not retain DB connections.
# EXPORT_TEMP_PATH=./storage
# Optional PostgreSQL, takes precedence over SQLite:
# DATABASE_URL=postgres://usage:change-me@localhost:5432/picpeak_usage
# Private maintainer access only; set 32+ random characters:
Expand Down
1 change: 1 addition & 0 deletions compose.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ services:
environment:
MAINTAINER_TOKEN: ${MAINTAINER_TOKEN:?Set a random maintainer token of at least 32 characters}
DATABASE_URL: ${DATABASE_URL:-}
EXPORT_TEMP_PATH: ${EXPORT_TEMP_PATH:-/app/storage}
SESSION_SECRET: ${SESSION_SECRET:-}
# The container is meant to run behind exactly one TLS-terminating reverse
# proxy on this host, so one X-Forwarded-For hop is trusted by default.
Expand Down
1 change: 1 addition & 0 deletions scripts/archive-source.js
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ const files = [
// Builds run outside Git too (Docker and downloaded source archives). Do not
// recursively include ignored local files just because their parent is source.
const localNames = new Set([
"storage", "dist", "coverage",
"node_modules", ".git", ".local", ".cache", ".vite", ".nyc_output",
".vscode", ".idea", ".claude", ".codex", ".cursor", ".playwright-mcp",
"test-results", "playwright-report", "blob-report",
Expand Down
4 changes: 3 additions & 1 deletion server/app.js
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ const { rateLimit, ipKeyGenerator } = require("express-rate-limit");
const crypto = require("node:crypto");
const path = require("node:path");
const { Collector } = require("./collector");
const { streamExport } = require("./exports");
const { createExportStreamer } = require("./exports");
const { history } = require("./history");
const { reporters, pageOptions } = require("./maintainer");
const {
Expand All @@ -27,9 +27,11 @@ function createApp({
maintainerToken = process.env.MAINTAINER_TOKEN,
now,
disableRateLimit = false,
exportOptions,
...options
}) {
const app = express();
const streamExport = createExportStreamer(exportOptions);
const collector = new Collector(db, {
now,
sessionSecret: process.env.SESSION_SECRET || maintainerToken,
Expand Down
3 changes: 3 additions & 0 deletions server/collector.js
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ const { readSnapshot } = require("./database");
const { emptyInventory, addInventory } = require("./inventory");
const { recordParticipation, pruneWeeklyActivity } = require("./weeklyActivity");
const PAGE_SIZE = 200;
const { FIRST_REPORT_DATE } = require("./reportDates");

class Collector {
constructor(db, options = {}) {
Expand Down Expand Up @@ -193,6 +194,8 @@ class Collector {
expires_at: now + MAX_AGE_MS * 2,
});
if (packet.action === "report") {
if (packet.payload.report_date < FIRST_REPORT_DATE)
throw new ProtocolError("INVALID_REPORT_DATE");
if (schemaRank(packet.schema_version) > schemaRank(schemaForConsent(installation.consent_version)))
throw new ProtocolError("CONSENT_REQUIRED", 409);
// One current daily report plus one delayed report per receiving day.
Expand Down
236 changes: 141 additions & 95 deletions server/exports.js
Original file line number Diff line number Diff line change
@@ -1,109 +1,155 @@
"use strict";
const crypto = require("node:crypto");
const fs = require("node:fs/promises");
const path = require("node:path");
const { pipeline } = require("node:stream/promises");
const { readSnapshot } = require("./database");
const { contributionRows } = require("./maintainer");
const { ProtocolError } = require("../protocol/protocol.cjs");

// One database snapshot per export, bounded pages and backpressure. A stalled
// browser must not hold a DB connection/SQLite read transaction indefinitely.
async function streamExport({
res,
// Per-process admission, with no waiting queue. Resolve sessions to their
// installation so rotating tokens cannot bypass the per-participant limit.
function createExportStreamer({
directory = process.env.EXPORT_TEMP_PATH || path.join(process.cwd(), "storage"),
maxConcurrent = 2,
maxBytes = 512 * 1024 * 1024,
timeoutMs = 30000,
} = {}) {
const active = new Set();
return async function streamExport(options) {
const { res, collector, credential, installationId, maintainer } = options;
const owner = maintainer ? "maintainer" : installationId || await collector.reader(credential);
if (active.has(owner)) throw new ProtocolError("EXPORT_IN_PROGRESS", 409);
if (active.size >= maxConcurrent) throw new ProtocolError("EXPORT_BUSY", 503);
active.add(owner);
const controller = new AbortController();
const cancel = () => controller.abort();
const disconnect = () => { if (!res.writableFinished) cancel(); };
res.once("close", disconnect);
if (res.destroyed) cancel();
const timeout = setTimeout(() => { cancel(); res.destroy(); }, timeoutMs);
timeout.unref();
let file, filename;
try {
controller.signal.throwIfAborted();
await fs.mkdir(directory, { recursive: true, mode: 0o700 });
filename = path.join(directory, `.usage-export-${crypto.randomUUID()}`);
file = await fs.open(filename, "wx+", 0o600);
// POSIX anonymous temporary file: no pathname remains for serving,
// backups or crash leftovers. Closing the descriptor releases the data.
await fs.unlink(filename);
filename = undefined;
let bytes = 0;
const write = async value => {
controller.signal.throwIfAborted();
bytes += Buffer.byteLength(value);
if (bytes > maxBytes) throw new ProtocolError("EXPORT_TOO_LARGE", 413);
await file.writeFile(value);
};
const metadata = await prepareExport({ ...options, write });
controller.signal.throwIfAborted();
// The snapshot is closed before any bytes can reach a slow client.
res.set({
"X-Dataset-Revision": metadata.revision,
"X-Exported-At": metadata.exportedAt,
"X-Export-Receipt": metadata.receiptId,
});
res.type(metadata.type).attachment(metadata.filename);
await pipeline(file.createReadStream({ start: 0, autoClose: false }), res, {
signal: controller.signal,
});
} catch (error) {
if (["ENOSPC", "EDQUOT"].includes(error.code))
throw new ProtocolError("EXPORT_STORAGE_FULL", 503);
throw error;
} finally {
clearTimeout(timeout);
res.off("close", disconnect);
try {
await file?.close();
} finally {
active.delete(owner);
if (filename) await fs.unlink(filename).catch(() => {});
}
}
};
}

// Bounded pages are serialized into the private file in one consistent
// snapshot. Network backpressure never extends this transaction's lifetime.
async function prepareExport({
write,
db,
collector,
credential,
installationId,
maintainer = false,
}) {
const timeout = setTimeout(() => res.destroy(), 30000);
timeout.unref();
const write = async (value) => {
if (res.destroyed) throw new Error("Export closed");
if (res.write(value)) return;
await new Promise((resolve) => {
const done = () => {
res.off("drain", done);
res.off("close", done);
resolve();
};
res.once("drain", done);
res.once("close", done);
});
if (res.destroyed) throw new Error("Export closed");
};
try {
await readSnapshot(db, async (tx) => {
if (installationId)
await collector.requireInstallation(installationId, tx);
else if (!maintainer) await collector.reader(credential, tx);
const revision = await collector.revision(tx);
const exportedAt = new Date(collector.now()).toISOString();
const receiptId = crypto.randomUUID();
res.set({
"X-Dataset-Revision": revision,
"X-Exported-At": exportedAt,
"X-Export-Receipt": receiptId,
});
let count = 0;
if (maintainer) {
res.type("application/x-ndjson").attachment("picpeak-usage-maintainer.ndjson");
const counts = {};
await write(`${JSON.stringify({ type: "manifest", data: {
format: "maintainer-export.v1", revision, exported_at: exportedAt,
scope: installationId || "all_reporters",
} })}\n`);
for await (const record of contributionRows(tx, installationId)) {
counts[record.type] = (counts[record.type] || 0) + 1;
await write(`${JSON.stringify(record)}\n`);
}
await write(`${JSON.stringify({ type: "export_receipt", data: {
receipt_version: "export.v1", receipt_id: receiptId, exported_at: exportedAt, revision, counts,
} })}\n`);
} else if (installationId) {
res.type("application/json").attachment("picpeak-usage-packets.json");
await write(
`${JSON.stringify({ installation_id: installationId }).slice(0, -1)},"packets":[`,
);
let after = "";
for (;;) {
const rows = await collector.rawPage(installationId, after, tx);
if (!rows.length) break;
for (const row of rows) {
await write(
`${count++ ? "," : ""}${JSON.stringify(collector.rawPacket(row))}`,
);
}
after = rows.at(-1).report_date;
return readSnapshot(db, async (tx) => {
if (installationId)
await collector.requireInstallation(installationId, tx);
else if (!maintainer) await collector.reader(credential, tx);
const revision = await collector.revision(tx);
const exportedAt = new Date(collector.now()).toISOString();
const receiptId = crypto.randomUUID();
let type = "application/x-ndjson", filename = "picpeak-usage-dataset.ndjson";
let count = 0;
if (maintainer) {
filename = "picpeak-usage-maintainer.ndjson";
const counts = {};
await write(`${JSON.stringify({ type: "manifest", data: {
format: "maintainer-export.v1", revision, exported_at: exportedAt,
scope: installationId || "all_reporters",
} })}\n`);
for await (const record of contributionRows(tx, installationId)) {
counts[record.type] = (counts[record.type] || 0) + 1;
await write(`${JSON.stringify(record)}\n`);
}
await write(`${JSON.stringify({ type: "export_receipt", data: {
receipt_version: "export.v1", receipt_id: receiptId, exported_at: exportedAt, revision, counts,
} })}\n`);
} else if (installationId) {
type = "application/json";
filename = "picpeak-usage-packets.json";
await write(
`${JSON.stringify({ installation_id: installationId }).slice(0, -1)},"packets":[`,
);
let after = "";
for (;;) {
const rows = await collector.rawPage(installationId, after, tx);
if (!rows.length) break;
for (const row of rows) {
await write(
`${count++ ? "," : ""}${JSON.stringify(collector.rawPacket(row))}`,
);
}
// User-held audit receipt. No central export/access history is stored.
await write(
`],"export_receipt":${JSON.stringify({
receipt_version: "export.v1",
receipt_id: receiptId,
exported_at: exportedAt,
revision,
scope: "unique accepted usage reports",
packet_count: count,
retries:
"first accepted envelope retained; transport retries deduplicated",
})}}`,
);
} else {
res
.type("application/x-ndjson")
.attachment("picpeak-usage-dataset.ndjson");
let offset = 0;
do {
const page = await collector.dataset(offset, 200, tx);
for (const row of page.records) {
await write(`${JSON.stringify(row)}\n`);
count++;
}
offset = page.next;
} while (offset !== null);
after = rows.at(-1).report_date;
}
});
res.end();
} finally {
clearTimeout(timeout);
}
// User-held audit receipt. No central export/access history is stored.
await write(
`],"export_receipt":${JSON.stringify({
receipt_version: "export.v1",
receipt_id: receiptId,
exported_at: exportedAt,
revision,
scope: "unique accepted usage reports",
packet_count: count,
retries:
"first accepted envelope retained; transport retries deduplicated",
})}}`,
);
} else {
let offset = 0;
do {
const page = await collector.dataset(offset, 200, tx);
for (const row of page.records) {
await write(`${JSON.stringify(row)}\n`);
count++;
}
offset = page.next;
} while (offset !== null);
}
return { revision, exportedAt, receiptId, type, filename };
});
}
module.exports = { streamExport };
module.exports = { createExportStreamer };
6 changes: 5 additions & 1 deletion server/history.js
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ const { ALL_FEATURE_KEYS, ProtocolError } = require("../protocol/protocol.cjs");
const { readSnapshot } = require("./database");
const { emptyInventory, addInventory } = require("./inventory");
const DAY = 86400000;
const { FIRST_REPORT_DATE } = require("./reportDates");

function period(date, interval) {
if (interval === "month") return `${date.slice(0, 7)}-01`;
Expand All @@ -20,6 +21,7 @@ function nextPeriod(date, interval) {
function validDate(value) {
return (
typeof value === "string" &&
value >= FIRST_REPORT_DATE &&
/^\d{4}-\d{2}-\d{2}$/.test(value) &&
Number.isFinite(Date.parse(`${value}T00:00:00.000Z`)) &&
new Date(`${value}T00:00:00.000Z`).toISOString().slice(0, 10) === value
Expand Down Expand Up @@ -86,7 +88,9 @@ async function history(
: null;
if (id) await collector.requireInstallation(id, tx);
const base = () => {
const query = tx("reports");
// Older deployments may have accepted outliers. Keep their original
// envelopes available for private export, but exclude them from history.
const query = tx("reports").where("report_date", ">=", FIRST_REPORT_DATE);
if (id) query.where({ installation_id: id });
return query;
};
Expand Down
6 changes: 6 additions & 0 deletions server/reportDates.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
"use strict";
// Collector policy, independent of the immutable wire schemas. This generous
// floor preserves supported 2020-era backlogs and predates the 2026 launch.
// Never let pre-service timestamps determine a shared history window.
const FIRST_REPORT_DATE = "2020-01-01";
module.exports = { FIRST_REPORT_DATE };
Loading
Loading