mirror of
https://github.com/jakejarvis/sofa.git
synced 2026-07-14 18:15:56 -04:00
fix: harden backup restore, first-user race, and several reliability issues
- Block all non-health API requests with 503 while a database restore is in progress (`withDatabaseAccessBlocked` in `@sofa/db/client`); pause and resume cron jobs around the restore window - Replace the two-hook first-user admin promotion with an atomic `claimInitialAdmin` query that uses a DB-level unique constraint so concurrent sign-ups during the first-run window can't each see `userCount === 0` - Fix `refreshAvailability` to always call `replaceAvailabilityTransaction` (clearing stale rows) even when TMDB returns no US providers, instead of returning early and leaving old data in place - Fix `performUpdateCheck` to read `release_url` (snake_case) from the public API response instead of `releaseUrl` - Fix `createJob` import handler to catch `SQLITE_CONSTRAINT_UNIQUE` and surface it as an `IMPORT_ALREADY_RUNNING` conflict error instead of a 500 - Relax public API telemetry schema to accept `string | number` for `users` and `titles` fields - Add tests for availability clearing, import deduplication, and `claimInitialAdmin`
This commit is contained in:
@@ -4,7 +4,7 @@ import { APIError, createAuthMiddleware } from "better-auth/api";
|
||||
import { type BetterAuthOptions, betterAuth } from "better-auth/minimal";
|
||||
import { admin, genericOAuth } from "better-auth/plugins";
|
||||
|
||||
import { getUserCount, isRegistrationOpen, setSetting } from "@sofa/core/settings";
|
||||
import { claimInitialAdmin, isRegistrationOpen } from "@sofa/core/settings";
|
||||
import { db } from "@sofa/db/client";
|
||||
import { createLogger } from "@sofa/logger";
|
||||
|
||||
@@ -89,25 +89,10 @@ export const auth = betterAuth({
|
||||
databaseHooks: {
|
||||
user: {
|
||||
create: {
|
||||
before: async (user) => {
|
||||
// First user becomes admin regardless of auth method
|
||||
const userCount = getUserCount();
|
||||
if (userCount === 0) {
|
||||
return {
|
||||
data: {
|
||||
...user,
|
||||
role: "admin",
|
||||
},
|
||||
};
|
||||
}
|
||||
return { data: user };
|
||||
},
|
||||
after: async () => {
|
||||
// Auto-close registration after first user
|
||||
const userCount = getUserCount();
|
||||
if (userCount === 1) {
|
||||
setSetting("registrationOpen", "false");
|
||||
}
|
||||
after: async (user) => {
|
||||
// Promote exactly one bootstrap user to admin, even if multiple
|
||||
// sign-ups race during the first-run window.
|
||||
claimInitialAdmin(user.id);
|
||||
},
|
||||
},
|
||||
},
|
||||
|
||||
@@ -15,35 +15,37 @@ export async function refreshAvailability(titleId: string) {
|
||||
|
||||
const data = await getWatchProviders(title.tmdbId, title.type);
|
||||
const us = data.results?.US;
|
||||
if (!us) {
|
||||
log.debug(`No US providers for title ${titleId}`);
|
||||
return;
|
||||
}
|
||||
|
||||
const now = new Date();
|
||||
const offerTypes = ["flatrate", "rent", "buy", "free", "ads"] as const;
|
||||
|
||||
// Collect all offer rows, then batch insert in a single transaction
|
||||
const allOfferRows: (typeof availabilityOffers.$inferInsert)[] = [];
|
||||
for (const offerType of offerTypes) {
|
||||
const providers = us[offerType];
|
||||
if (!providers) continue;
|
||||
for (const p of providers) {
|
||||
allOfferRows.push({
|
||||
titleId,
|
||||
region: "US",
|
||||
providerId: p.provider_id,
|
||||
providerName: p.provider_name ?? "",
|
||||
logoPath: p.logo_path ?? "",
|
||||
offerType,
|
||||
link: us.link ?? null,
|
||||
lastFetchedAt: now,
|
||||
});
|
||||
if (us) {
|
||||
for (const offerType of offerTypes) {
|
||||
const providers = us[offerType];
|
||||
if (!providers) continue;
|
||||
for (const p of providers) {
|
||||
allOfferRows.push({
|
||||
titleId,
|
||||
region: "US",
|
||||
providerId: p.provider_id,
|
||||
providerName: p.provider_name ?? "",
|
||||
logoPath: p.logo_path ?? "",
|
||||
offerType,
|
||||
link: us.link ?? null,
|
||||
lastFetchedAt: now,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
replaceAvailabilityTransaction(titleId, "US", allOfferRows);
|
||||
|
||||
if (!us) {
|
||||
log.debug(`No US providers for title ${titleId}; cleared cached offers`);
|
||||
return;
|
||||
}
|
||||
|
||||
const total = offerTypes.reduce((n, t) => n + (us[t]?.length ?? 0), 0);
|
||||
log.debug(`Refreshed availability for title ${titleId}: ${total} offers`);
|
||||
}
|
||||
|
||||
@@ -4,7 +4,7 @@ import { mkdir, readdir } from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
|
||||
import { BACKUP_DIR, DATABASE_URL } from "@sofa/config";
|
||||
import { closeDatabase, vacuumDatabase } from "@sofa/db/client";
|
||||
import { closeDatabase, vacuumDatabase, withDatabaseAccessBlocked } from "@sofa/db/client";
|
||||
import { runMigrations } from "@sofa/db/migrate";
|
||||
import { createLogger } from "@sofa/logger";
|
||||
|
||||
@@ -240,15 +240,17 @@ export async function restoreFromBackup(source: Buffer | string): Promise<void>
|
||||
log.info("Creating pre-restore safety backup...");
|
||||
await createBackupInternal("pre-restore");
|
||||
|
||||
// Keep the close+replace window synchronous to avoid event-loop interleaving.
|
||||
log.info("Replacing database...");
|
||||
closeDatabase();
|
||||
renameSync(tempPath, DATABASE_URL);
|
||||
unlinkIfExistsSync(`${DATABASE_URL}-wal`);
|
||||
unlinkIfExistsSync(`${DATABASE_URL}-shm`);
|
||||
await withDatabaseAccessBlocked(async () => {
|
||||
// Keep the close+replace window synchronous to avoid event-loop interleaving.
|
||||
log.info("Replacing database...");
|
||||
closeDatabase();
|
||||
renameSync(tempPath, DATABASE_URL);
|
||||
unlinkIfExistsSync(`${DATABASE_URL}-wal`);
|
||||
unlinkIfExistsSync(`${DATABASE_URL}-shm`);
|
||||
|
||||
// Ensure restored backups from older app versions are brought up-to-date.
|
||||
runMigrations();
|
||||
// Ensure restored backups from older app versions are brought up-to-date.
|
||||
runMigrations();
|
||||
});
|
||||
log.info("Database restored successfully");
|
||||
} finally {
|
||||
const tempFile = Bun.file(tempPath);
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import {
|
||||
claimInitialAdmin as queryClaimInitialAdmin,
|
||||
getSettingValue,
|
||||
getUserCount as queryGetUserCount,
|
||||
upsertSetting,
|
||||
@@ -16,6 +17,10 @@ export function getUserCount(): number {
|
||||
return queryGetUserCount();
|
||||
}
|
||||
|
||||
export function claimInitialAdmin(userId: string): boolean {
|
||||
return queryClaimInitialAdmin(userId);
|
||||
}
|
||||
|
||||
export function getInstanceId(): string {
|
||||
const existing = getSetting("instanceId");
|
||||
if (existing) return existing;
|
||||
|
||||
@@ -74,11 +74,11 @@ export async function performUpdateCheck(): Promise<UpdateCheckResult> {
|
||||
});
|
||||
if (!res.ok) throw new Error(`Public API ${res.status}`);
|
||||
|
||||
const data = (await res.json()) as { version: string; releaseUrl: string };
|
||||
const data = (await res.json()) as { version: string; release_url: string };
|
||||
const version = data.version;
|
||||
|
||||
setSetting("updateCheckLatestVersion", version);
|
||||
setSetting("updateCheckReleaseUrl", data.releaseUrl);
|
||||
setSetting("updateCheckReleaseUrl", data.release_url);
|
||||
setSetting("updateCheckLastCheckedAt", new Date().toISOString());
|
||||
|
||||
log.info(`Update check complete: current=${APP_VERSION}, latest=${version}`);
|
||||
@@ -87,7 +87,7 @@ export async function performUpdateCheck(): Promise<UpdateCheckResult> {
|
||||
updateAvailable: isNewerVersion(version, APP_VERSION),
|
||||
currentVersion: APP_VERSION,
|
||||
latestVersion: version,
|
||||
releaseUrl: data.releaseUrl,
|
||||
releaseUrl: data.release_url,
|
||||
lastCheckedAt: new Date().toISOString(),
|
||||
};
|
||||
} catch (err) {
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
import { beforeEach, describe, expect, mock, test } from "bun:test";
|
||||
|
||||
import { availabilityOffers } from "@sofa/db/schema";
|
||||
import {
|
||||
clearAllTables,
|
||||
eq,
|
||||
insertAvailabilityOffer,
|
||||
insertTitle,
|
||||
testDb,
|
||||
} from "@sofa/db/test-utils";
|
||||
|
||||
let watchProvidersResponse: { results: Record<string, unknown> } = { results: {} };
|
||||
|
||||
const getWatchProviders = mock(async () => watchProvidersResponse);
|
||||
|
||||
mock.module("@sofa/tmdb/client", () => ({
|
||||
getWatchProviders,
|
||||
}));
|
||||
|
||||
import { refreshAvailability } from "../src/availability";
|
||||
|
||||
beforeEach(() => {
|
||||
clearAllTables();
|
||||
watchProvidersResponse = { results: {} };
|
||||
});
|
||||
|
||||
describe("refreshAvailability", () => {
|
||||
test("clears stale US offers when TMDB returns no US availability", async () => {
|
||||
insertTitle({ id: "movie-1", tmdbId: 101, type: "movie", title: "Movie" });
|
||||
insertAvailabilityOffer("movie-1");
|
||||
|
||||
await refreshAvailability("movie-1");
|
||||
|
||||
const offers = testDb
|
||||
.select()
|
||||
.from(availabilityOffers)
|
||||
.where(eq(availabilityOffers.titleId, "movie-1"))
|
||||
.all();
|
||||
|
||||
expect(offers).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
@@ -344,6 +344,80 @@ describe("processImportJob — state transitions", () => {
|
||||
test("readImportJob throws for non-existent job", () => {
|
||||
expect(() => readImportJob("non-existent-id")).toThrow("Import job non-existent-id not found");
|
||||
});
|
||||
|
||||
test("schema allows only one active import job per user", () => {
|
||||
const userId = insertUser();
|
||||
const createdAt = new Date();
|
||||
|
||||
testDb
|
||||
.insert(importJobs)
|
||||
.values({
|
||||
id: "job-1",
|
||||
userId,
|
||||
source: "trakt",
|
||||
status: "pending",
|
||||
payload: JSON.stringify({
|
||||
source: "trakt",
|
||||
movies: [],
|
||||
episodes: [],
|
||||
watchlist: [],
|
||||
ratings: [],
|
||||
}),
|
||||
importWatches: true,
|
||||
importWatchlist: true,
|
||||
importRatings: true,
|
||||
createdAt,
|
||||
})
|
||||
.run();
|
||||
|
||||
expect(() =>
|
||||
testDb
|
||||
.insert(importJobs)
|
||||
.values({
|
||||
id: "job-2",
|
||||
userId,
|
||||
source: "simkl",
|
||||
status: "running",
|
||||
payload: JSON.stringify({
|
||||
source: "simkl",
|
||||
movies: [],
|
||||
episodes: [],
|
||||
watchlist: [],
|
||||
ratings: [],
|
||||
}),
|
||||
importWatches: true,
|
||||
importWatchlist: true,
|
||||
importRatings: true,
|
||||
createdAt,
|
||||
})
|
||||
.run(),
|
||||
).toThrow();
|
||||
|
||||
testDb.update(importJobs).set({ status: "success" }).where(eq(importJobs.id, "job-1")).run();
|
||||
|
||||
expect(() =>
|
||||
testDb
|
||||
.insert(importJobs)
|
||||
.values({
|
||||
id: "job-3",
|
||||
userId,
|
||||
source: "letterboxd",
|
||||
status: "pending",
|
||||
payload: JSON.stringify({
|
||||
source: "letterboxd",
|
||||
movies: [],
|
||||
episodes: [],
|
||||
watchlist: [],
|
||||
ratings: [],
|
||||
}),
|
||||
importWatches: true,
|
||||
importWatchlist: true,
|
||||
importRatings: true,
|
||||
createdAt,
|
||||
})
|
||||
.run(),
|
||||
).not.toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
// ── Failed Resolution ───────────────────────────────────────────────
|
||||
|
||||
@@ -9,6 +9,8 @@ mock.module("@sofa/db/client", () => ({
|
||||
optimizeDatabase: () => {},
|
||||
vacuumDatabase: () => {},
|
||||
closeDatabase: () => {},
|
||||
isDatabaseAccessBlocked: () => false,
|
||||
withDatabaseAccessBlocked: async (fn: () => Promise<unknown> | unknown) => await fn(),
|
||||
}));
|
||||
|
||||
applyMigrations();
|
||||
|
||||
@@ -1,8 +1,15 @@
|
||||
import { beforeEach, describe, expect, test } from "bun:test";
|
||||
|
||||
import { clearAllTables, insertUser } from "@sofa/db/test-utils";
|
||||
import { user } from "@sofa/db/schema";
|
||||
import { clearAllTables, eq, insertUser, testDb } from "@sofa/db/test-utils";
|
||||
|
||||
import { getSetting, getUserCount, isRegistrationOpen, setSetting } from "../src/settings";
|
||||
import {
|
||||
claimInitialAdmin,
|
||||
getSetting,
|
||||
getUserCount,
|
||||
isRegistrationOpen,
|
||||
setSetting,
|
||||
} from "../src/settings";
|
||||
|
||||
beforeEach(() => {
|
||||
clearAllTables();
|
||||
@@ -65,3 +72,21 @@ describe("isRegistrationOpen", () => {
|
||||
expect(isRegistrationOpen()).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("claimInitialAdmin", () => {
|
||||
test("promotes the earliest user to admin and closes registration", () => {
|
||||
insertUser("user-1");
|
||||
insertUser("user-2");
|
||||
|
||||
expect(claimInitialAdmin("user-2")).toBe(false);
|
||||
expect(claimInitialAdmin("user-1")).toBe(true);
|
||||
expect(claimInitialAdmin("user-1")).toBe(false);
|
||||
|
||||
const firstUser = testDb.select().from(user).where(eq(user.id, "user-1")).get();
|
||||
const secondUser = testDb.select().from(user).where(eq(user.id, "user-2")).get();
|
||||
|
||||
expect(firstUser?.role).toBe("admin");
|
||||
expect(secondUser?.role).toBe("user");
|
||||
expect(getSetting("registrationOpen")).toBe("false");
|
||||
});
|
||||
});
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
CREATE INDEX `episodes_airDate` ON `episodes` (`airDate`);--> statement-breakpoint
|
||||
CREATE UNIQUE INDEX `importJobs_active_user` ON `importJobs` (`userId`) WHERE "importJobs"."status" in ('pending', 'running');
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,4 +1,5 @@
|
||||
import { Database } from "bun:sqlite";
|
||||
import { AsyncLocalStorage } from "node:async_hooks";
|
||||
|
||||
import type { Logger } from "drizzle-orm";
|
||||
import { drizzle } from "drizzle-orm/bun-sqlite";
|
||||
@@ -16,6 +17,13 @@ const drizzleLogger: Logger = {
|
||||
},
|
||||
};
|
||||
|
||||
export class DatabaseRestoreInProgressError extends Error {
|
||||
constructor(message = "Database is temporarily unavailable during restore") {
|
||||
super(message);
|
||||
this.name = "DatabaseRestoreInProgressError";
|
||||
}
|
||||
}
|
||||
|
||||
// Lazy-init singleton via globalThis. Next.js evaluates module-level code at
|
||||
// build time (when no database exists) and re-imports modules on HMR in dev
|
||||
// (which would create duplicate connections). Stashing the instances on
|
||||
@@ -30,9 +38,19 @@ const drizzleLogger: Logger = {
|
||||
const globalForDb = globalThis as unknown as {
|
||||
_db: ReturnType<typeof drizzle> | undefined;
|
||||
_client: Database | undefined;
|
||||
_accessBlocked: boolean | undefined;
|
||||
};
|
||||
|
||||
const dbAccessBypass = new AsyncLocalStorage<boolean>();
|
||||
|
||||
function assertDatabaseAccessible() {
|
||||
if (globalForDb._accessBlocked && !dbAccessBypass.getStore()) {
|
||||
throw new DatabaseRestoreInProgressError();
|
||||
}
|
||||
}
|
||||
|
||||
function getClient() {
|
||||
assertDatabaseAccessible();
|
||||
if (!globalForDb._client) {
|
||||
globalForDb._client = new Database(DATABASE_URL);
|
||||
globalForDb._client.run("PRAGMA journal_mode = WAL");
|
||||
@@ -47,6 +65,7 @@ function getClient() {
|
||||
}
|
||||
|
||||
function getDb() {
|
||||
assertDatabaseAccessible();
|
||||
if (!globalForDb._db) {
|
||||
globalForDb._db = drizzle({
|
||||
client: getClient(),
|
||||
@@ -59,6 +78,7 @@ function getDb() {
|
||||
|
||||
export const db = new Proxy({} as ReturnType<typeof drizzle>, {
|
||||
get(_, prop) {
|
||||
assertDatabaseAccessible();
|
||||
return Reflect.get(getDb(), prop);
|
||||
},
|
||||
});
|
||||
@@ -72,6 +92,19 @@ export function vacuumDatabase(into: string): void {
|
||||
getClient().run("VACUUM INTO ?", [into.replace(/'/g, "''")]);
|
||||
}
|
||||
|
||||
export function isDatabaseAccessBlocked(): boolean {
|
||||
return globalForDb._accessBlocked === true;
|
||||
}
|
||||
|
||||
export async function withDatabaseAccessBlocked<T>(fn: () => Promise<T> | T): Promise<T> {
|
||||
globalForDb._accessBlocked = true;
|
||||
try {
|
||||
return await dbAccessBypass.run(true, fn);
|
||||
} finally {
|
||||
globalForDb._accessBlocked = false;
|
||||
}
|
||||
}
|
||||
|
||||
/** Close the current connection, and clear singletons so the Proxy re-initializes on next access. */
|
||||
export function closeDatabase() {
|
||||
globalForDb._client?.close();
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
import { count, eq } from "drizzle-orm";
|
||||
import { asc, count, eq } from "drizzle-orm";
|
||||
|
||||
import { db } from "../client";
|
||||
import { appSettings, user } from "../schema";
|
||||
|
||||
const INITIAL_ADMIN_KEY = "initialAdminAssigned";
|
||||
|
||||
export function getSettingValue(key: string): string | null {
|
||||
const row = db.select().from(appSettings).where(eq(appSettings.key, key)).get();
|
||||
return row?.value ?? null;
|
||||
@@ -19,3 +21,42 @@ export function getUserCount(): number {
|
||||
const result = db.select({ count: count() }).from(user).get();
|
||||
return result?.count ?? 0;
|
||||
}
|
||||
|
||||
export function claimInitialAdmin(userId: string): boolean {
|
||||
let claimed = false;
|
||||
|
||||
db.transaction((tx) => {
|
||||
const firstUser = tx
|
||||
.select({ id: user.id })
|
||||
.from(user)
|
||||
.orderBy(asc(user.createdAt), asc(user.id))
|
||||
.limit(1)
|
||||
.get();
|
||||
if (!firstUser || firstUser.id !== userId) {
|
||||
return;
|
||||
}
|
||||
|
||||
const lock = tx
|
||||
.insert(appSettings)
|
||||
.values({ key: INITIAL_ADMIN_KEY, value: userId })
|
||||
.onConflictDoNothing()
|
||||
.returning({ key: appSettings.key })
|
||||
.get();
|
||||
if (!lock) {
|
||||
return;
|
||||
}
|
||||
|
||||
tx.update(user).set({ role: "admin" }).where(eq(user.id, userId)).run();
|
||||
tx.insert(appSettings)
|
||||
.values({ key: "registrationOpen", value: "false" })
|
||||
.onConflictDoUpdate({
|
||||
target: appSettings.key,
|
||||
set: { value: "false" },
|
||||
})
|
||||
.run();
|
||||
|
||||
claimed = true;
|
||||
});
|
||||
|
||||
return claimed;
|
||||
}
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { sql } from "drizzle-orm";
|
||||
import { index, int, real, sqliteTable, text, uniqueIndex } from "drizzle-orm/sqlite-core";
|
||||
|
||||
// Helper for UUID primary keys
|
||||
@@ -472,6 +473,9 @@ export const importJobs = sqliteTable(
|
||||
(table) => [
|
||||
index("importJobs_userId_createdAt").on(table.userId, table.createdAt),
|
||||
index("importJobs_status").on(table.status),
|
||||
uniqueIndex("importJobs_active_user")
|
||||
.on(table.userId)
|
||||
.where(sql`${table.status} in ('pending', 'running')`),
|
||||
],
|
||||
);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user