feat: add storage usage tracking with DB counters and reconciliation

Increment users.storageUsed on file upload/save, decrement on delete
(per-user via GREATEST to prevent negatives). Add per-team storage
breakdown to GET /api/v1/admin/usage. Weekly reconciliation job
(3 AM Sunday) recomputes counters from actual userFiles sums.
This commit is contained in:
SnapOtter
2026-06-13 20:43:52 +08:00
parent 7e5843cd40
commit aaa8a37c9b
4 changed files with 122 additions and 7 deletions
@@ -0,0 +1,53 @@
/**
* Weekly storage reconciliation job.
*
* Recomputes each user's storageUsed counter from the actual sum of
* their userFiles rows and corrects any drift caused by race conditions,
* crashes, or bugs. Scheduled for 3 AM Sunday via system-jobs.
*/
import { and, eq, notInArray, sql } from "drizzle-orm";
import { db, schema } from "../db/index.js";
export async function storageReconciliationJob(): Promise<void> {
// Sum actual file sizes per user
const actual = await db
.select({
userId: schema.userFiles.userId,
totalSize: sql<number>`coalesce(sum(${schema.userFiles.size}), 0)::int`,
})
.from(schema.userFiles)
.groupBy(schema.userFiles.userId);
let updated = 0;
for (const row of actual) {
if (!row.userId) continue;
const result = await db
.update(schema.users)
.set({ storageUsed: row.totalSize })
.where(
and(eq(schema.users.id, row.userId), sql`${schema.users.storageUsed} != ${row.totalSize}`),
);
if (result.rowCount) updated++;
}
// Zero out users who have no files but a nonzero storageUsed counter
const usersWithFiles = actual.filter((r) => r.userId != null).map((r) => r.userId as string);
if (usersWithFiles.length > 0) {
await db
.update(schema.users)
.set({ storageUsed: 0 })
.where(
and(sql`${schema.users.storageUsed} > 0`, notInArray(schema.users.id, usersWithFiles)),
);
} else {
// No users have files -- zero everyone
await db
.update(schema.users)
.set({ storageUsed: 0 })
.where(sql`${schema.users.storageUsed} > 0`);
}
console.log(
`Storage reconciliation complete: ${actual.length} users checked, ${updated} corrected`,
);
}
+9
View File
@@ -25,6 +25,7 @@ export const SYSTEM_JOBS = {
retention: "system:retention",
siemForward: "system:siem-forward",
auditArchive: "system:audit-archive",
storageReconciliation: "system:storage-reconciliation",
} as const;
// -- Scheduling ---------------------------------------------------------------
@@ -48,6 +49,10 @@ export async function scheduleSystemJobs(): Promise<void> {
await q.upsertJobScheduler(SYSTEM_JOBS.auditArchive, {
pattern: "0 2 1 * *",
});
// Weekly: 3:00 AM Sunday -- reconcile storageUsed counters
await q.upsertJobScheduler(SYSTEM_JOBS.storageReconciliation, {
pattern: "0 3 * * 0",
});
}
/** Enqueue a one-shot system job (e.g. startup cleanup trigger). */
@@ -71,6 +76,10 @@ export async function runSystemJob(job: Job): Promise<unknown> {
return runSiemForward();
case SYSTEM_JOBS.auditArchive:
return runAuditArchive();
case SYSTEM_JOBS.storageReconciliation: {
const { storageReconciliationJob } = await import("./storage-reconciliation.js");
return storageReconciliationJob();
}
default:
// batch-finalize runs on the system pool too but is routed by the
// worker before calling runSystemJob. Anything else is a bug.
+18 -1
View File
@@ -11,7 +11,7 @@
import { sql } from "drizzle-orm";
import type { FastifyInstance, FastifyReply, FastifyRequest } from "fastify";
import { z } from "zod";
import { db } from "../db/index.js";
import { db, schema } from "../db/index.js";
import { formatZodErrors } from "../lib/errors.js";
import { metricsText } from "../lib/metrics.js";
import { buildSupportBundle } from "../lib/support-bundle.js";
@@ -129,6 +129,16 @@ export async function adminOpsRoutes(app: FastifyInstance): Promise<void> {
sql`SELECT coalesce(sum(size), 0)::text AS bytes, count(*)::int AS files FROM user_files`,
);
// Per-team storage breakdown from pre-computed user counters
const teamStorageRows = await db
.select({
teamName: schema.users.team,
totalBytes: sql<string>`coalesce(sum(${schema.users.storageUsed}), 0)::text`,
userCount: sql<number>`count(*)::int`,
})
.from(schema.users)
.groupBy(schema.users.team);
const jobsPerDay = (jobsPerDayResult.rows as Array<Record<string, unknown>>).map((r) => ({
day: String(r.day),
total: Number(r.total),
@@ -162,6 +172,12 @@ export async function adminOpsRoutes(app: FastifyInstance): Promise<void> {
libraryFiles: Number(storageRow.files),
};
const teamStorage = teamStorageRows.map((r) => ({
teamName: r.teamName,
totalBytes: r.totalBytes,
userCount: r.userCount,
}));
return {
days,
jobsPerDay,
@@ -169,6 +185,7 @@ export async function adminOpsRoutes(app: FastifyInstance): Promise<void> {
perUser,
durations,
storage,
teamStorage,
};
});
}
+42 -6
View File
@@ -232,6 +232,7 @@ export async function userFileRoutes(app: FastifyInstance): Promise<void> {
// Create DB record
const id = randomUUID();
const fileSize = safeBuffer.length;
try {
await db.insert(schema.userFiles).values({
id,
@@ -239,7 +240,7 @@ export async function userFileRoutes(app: FastifyInstance): Promise<void> {
originalName: safeName,
storedName,
mimeType,
size: safeBuffer.length,
size: fileSize,
width: validation.width,
height: validation.height,
version: 1,
@@ -250,6 +251,14 @@ export async function userFileRoutes(app: FastifyInstance): Promise<void> {
return reply.status(409).send({ error: "Failed to save file record" });
}
// Increment the user's pre-computed storage counter
if (userId) {
await db
.update(schema.users)
.set({ storageUsed: sql`${schema.users.storageUsed} + ${fileSize}` })
.where(eq(schema.users.id, userId));
}
const [row] = await db.select().from(schema.userFiles).where(eq(schema.userFiles.id, id));
if (row) created.push(serializeFile(row));
@@ -502,6 +511,8 @@ export async function userFileRoutes(app: FastifyInstance): Promise<void> {
type DeleteChainRow = {
id: string;
stored_name: string;
size: number | null;
user_id: string | null;
};
// Single recursive CTE to collect all chain members for every valid ID
@@ -518,15 +529,15 @@ export async function userFileRoutes(app: FastifyInstance): Promise<void> {
SELECT uf.id, uf.parent_id FROM user_files uf
INNER JOIN ancestors a ON uf.id = a.parent_id
),
chain(id, stored_name) AS (
SELECT f.id, f.stored_name FROM user_files f
chain(id, stored_name, size, user_id) AS (
SELECT f.id, f.stored_name, f.size, f.user_id FROM user_files f
WHERE f.id IN (SELECT id FROM ancestors WHERE parent_id IS NULL)
UNION ALL
SELECT child.id, child.stored_name
SELECT child.id, child.stored_name, child.size, child.user_id
FROM user_files child
INNER JOIN chain c ON child.parent_id = c.id
)
SELECT DISTINCT id, stored_name FROM chain
SELECT DISTINCT id, stored_name, size, user_id FROM chain
`);
const chainRows = cteResult.rows;
@@ -542,6 +553,22 @@ export async function userFileRoutes(app: FastifyInstance): Promise<void> {
await db.delete(schema.userFiles).where(inArray(schema.userFiles.id, chainIds));
}
// Decrement storageUsed per user (group by userId for files:all scenarios)
const perUserSizes = new Map<string, number>();
for (const row of chainRows) {
if (row.user_id && row.size) {
perUserSizes.set(row.user_id, (perUserSizes.get(row.user_id) ?? 0) + row.size);
}
}
for (const [uid, totalSize] of perUserSizes) {
await db
.update(schema.users)
.set({
storageUsed: sql`GREATEST(0, ${schema.users.storageUsed} - ${totalSize})`,
})
.where(eq(schema.users.id, uid));
}
await auditFromRequest(request)("FILE_DELETED", {
userId: user.id,
count: chainRows.length,
@@ -640,6 +667,7 @@ export async function userFileRoutes(app: FastifyInstance): Promise<void> {
// Create DB record
const id = randomUUID();
const fileSize = safeResultBuffer.length;
try {
await db.insert(schema.userFiles).values({
id,
@@ -647,7 +675,7 @@ export async function userFileRoutes(app: FastifyInstance): Promise<void> {
originalName: resultName,
storedName,
mimeType,
size: safeResultBuffer.length,
size: fileSize,
width: validation.width,
height: validation.height,
version: nextVersion,
@@ -658,6 +686,14 @@ export async function userFileRoutes(app: FastifyInstance): Promise<void> {
return reply.status(409).send({ error: "Failed to save result record" });
}
// Increment the user's pre-computed storage counter
if (userId) {
await db
.update(schema.users)
.set({ storageUsed: sql`${schema.users.storageUsed} + ${fileSize}` })
.where(eq(schema.users.id, userId));
}
const [row] = await db.select().from(schema.userFiles).where(eq(schema.userFiles.id, id));
return reply.status(201).send({ file: row ? serializeFile(row) : null });