`query_uploads` selects the rows the archives are built from; `estimate_export_bytes` sizes them for the disk preflight. They stated the same WHERE clause separately, and the direction of drift matters: an estimate that MISSES rows the archive writes under-reserves, which is precisely the ENOSPC the preflight exists to prevent. The integration test claimed to guard this and cannot. Both sides of `the_estimate_sums_exactly_the_rows_the_archive_will_contain` are `SRC:`-marked hand-copies in tests/common/mod.rs -- neither is production code -- so drift means production moved while both copies sat still, and the test goes on passing. The convention is sound for pinning behaviour; it is structurally incapable of detecting divergence from the thing it copies. So fix it where it can be fixed. One `export_visibility_where!()` fragment, `concat!`-ed into both queries at compile time (still `&'static str`, no allocation), with the `u`/`usr` alias contract stated. Divergence is now impossible by construction rather than watched for. The tests keep their value and lose the overclaim: the docstrings now say they pin WHICH uploads may be counted -- each excluded row in the fixture is excluded by a different predicate, so weakening any one of them still fails here -- and say plainly that they do not detect drift, with a pointer to what does. No behaviour change. The filters were verified identical before the hoist (`u.event_id = $1 AND u.deleted_at IS NULL AND usr.uploads_hidden = FALSE AND usr.is_banned = FALSE`); 99 backend tests still pass. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
346 lines
11 KiB
Rust
346 lines
11 KiB
Rust
//! Shared fixtures for the DB-backed integration tests.
|
|
//!
|
|
//! Every helper here executes SQL that is **character-for-character identical** to what `src/`
|
|
//! actually runs (see the `// SRC:` markers). That is the whole point: a paraphrased query is a
|
|
//! query nobody runs, and a test that passes against a paraphrase proves nothing about production.
|
|
|
|
#![allow(dead_code)] // each integration-test crate uses a different subset of these helpers
|
|
|
|
use sqlx::PgPool;
|
|
use uuid::Uuid;
|
|
|
|
/// Insert a bare, unreleased event (epoch 0, uploads open).
|
|
pub async fn seed_event(pool: &PgPool, slug: &str) -> Uuid {
|
|
sqlx::query_scalar("INSERT INTO event (slug, name) VALUES ($1, $2) RETURNING id")
|
|
.bind(slug)
|
|
.bind("Hochzeit")
|
|
.fetch_one(pool)
|
|
.await
|
|
.expect("seed event")
|
|
}
|
|
|
|
/// Insert a guest with a zeroed byte total.
|
|
pub async fn seed_user(pool: &PgPool, event_id: Uuid, name: &str) -> Uuid {
|
|
sqlx::query_scalar(
|
|
"INSERT INTO \"user\" (event_id, display_name, recovery_pin_hash)
|
|
VALUES ($1, $2, 'x') RETURNING id",
|
|
)
|
|
.bind(event_id)
|
|
.bind(name)
|
|
.fetch_one(pool)
|
|
.await
|
|
.expect("seed user")
|
|
}
|
|
|
|
/// SRC: `handlers/host.rs::release_gallery` — the claim + epoch bump, verbatim.
|
|
/// Returns the POST-increment epoch, exactly as the handler consumes it.
|
|
pub async fn release_gallery(pool: &PgPool, slug: &str) -> Option<i64> {
|
|
let claimed: Option<(Uuid, String, i64)> = sqlx::query_as(
|
|
"UPDATE event
|
|
SET export_released_at = NOW(),
|
|
uploads_locked_at = COALESCE(uploads_locked_at, NOW()),
|
|
export_epoch = export_epoch + 1
|
|
WHERE slug = $1 AND export_released_at IS NULL
|
|
RETURNING id, name, export_epoch",
|
|
)
|
|
.bind(slug)
|
|
.fetch_optional(pool)
|
|
.await
|
|
.expect("release_gallery");
|
|
|
|
if let Some((event_id, _, epoch)) = claimed {
|
|
// The handler arms both jobs in the SAME transaction; for a single-connection fixture the
|
|
// sequencing is equivalent.
|
|
let mut conn = pool.acquire().await.expect("acquire");
|
|
enqueue_types_at_epoch(&mut conn, event_id, epoch, &["zip", "html"]).await;
|
|
Some(epoch)
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
|
|
/// SRC: `handlers/host.rs::open_event` — the one statement that retires an entire generation.
|
|
/// Returns rows affected.
|
|
pub async fn open_event(pool: &PgPool, slug: &str) -> u64 {
|
|
sqlx::query(
|
|
"UPDATE event
|
|
SET uploads_locked_at = NULL,
|
|
export_released_at = NULL,
|
|
export_epoch = export_epoch + 1
|
|
WHERE slug = $1 AND (uploads_locked_at IS NOT NULL OR export_released_at IS NOT NULL)",
|
|
)
|
|
.bind(slug)
|
|
.execute(pool)
|
|
.await
|
|
.expect("open_event")
|
|
.rows_affected()
|
|
}
|
|
|
|
/// SRC: `services/export.rs::enqueue_types_at_epoch` — verbatim upsert.
|
|
pub async fn enqueue_types_at_epoch(
|
|
conn: &mut sqlx::PgConnection,
|
|
event_id: Uuid,
|
|
epoch: i64,
|
|
types: &[&str],
|
|
) {
|
|
for export_type in types {
|
|
sqlx::query(
|
|
"INSERT INTO export_job (event_id, type, status, progress_pct, epoch)
|
|
VALUES ($1, $2::export_type, 'pending', 0, $3)
|
|
ON CONFLICT (event_id, type) DO UPDATE
|
|
SET status = 'pending', progress_pct = 0, file_path = NULL,
|
|
error_message = NULL, completed_at = NULL,
|
|
epoch = EXCLUDED.epoch
|
|
WHERE export_job.status <> 'done' OR export_job.epoch <> EXCLUDED.epoch",
|
|
)
|
|
.bind(event_id)
|
|
.bind(export_type)
|
|
.bind(epoch)
|
|
.execute(&mut *conn)
|
|
.await
|
|
.expect("enqueue_types_at_epoch");
|
|
}
|
|
}
|
|
|
|
/// SRC: `services/export.rs::claim_job` — verbatim. `true` = we won the generation.
|
|
pub async fn claim_job(pool: &PgPool, event_id: Uuid, export_type: &str, epoch: i64) -> bool {
|
|
sqlx::query(
|
|
"UPDATE export_job SET status = 'running'
|
|
WHERE event_id = $1 AND type = $2::export_type
|
|
AND epoch = $3 AND status = 'pending'",
|
|
)
|
|
.bind(event_id)
|
|
.bind(export_type)
|
|
.bind(epoch)
|
|
.execute(pool)
|
|
.await
|
|
.expect("claim_job")
|
|
.rows_affected()
|
|
> 0
|
|
}
|
|
|
|
/// SRC: `services/export.rs::finalize_job` — verbatim. This IS the publish step.
|
|
pub async fn finalize_job(
|
|
pool: &PgPool,
|
|
event_id: Uuid,
|
|
export_type: &str,
|
|
epoch: i64,
|
|
file_path: &str,
|
|
) -> bool {
|
|
sqlx::query(
|
|
"UPDATE export_job
|
|
SET status = 'done', progress_pct = 100, file_path = $3, completed_at = NOW()
|
|
WHERE event_id = $1 AND type = $2::export_type
|
|
AND epoch = $4 AND status = 'running'",
|
|
)
|
|
.bind(event_id)
|
|
.bind(export_type)
|
|
.bind(file_path)
|
|
.bind(epoch)
|
|
.execute(pool)
|
|
.await
|
|
.expect("finalize_job")
|
|
.rows_affected()
|
|
> 0
|
|
}
|
|
|
|
/// SRC: `services/export.rs::update_progress` — verbatim. Doubles as the worker's liveness check:
|
|
/// `false` means "your generation was retired, stop working".
|
|
pub async fn update_progress(
|
|
pool: &PgPool,
|
|
event_id: Uuid,
|
|
export_type: &str,
|
|
epoch: i64,
|
|
pct: i16,
|
|
) -> bool {
|
|
sqlx::query(
|
|
"UPDATE export_job SET progress_pct = $3
|
|
WHERE event_id = $1 AND type = $2::export_type
|
|
AND epoch = $4 AND status = 'running'",
|
|
)
|
|
.bind(event_id)
|
|
.bind(export_type)
|
|
.bind(pct)
|
|
.bind(epoch)
|
|
.execute(pool)
|
|
.await
|
|
.expect("update_progress")
|
|
.rows_affected()
|
|
> 0
|
|
}
|
|
|
|
/// SRC: `services/export.rs::invalidate_and_arm` — the ViewerOnly ZIP carry-forward, verbatim.
|
|
/// Returns `rows_affected() == 1`, which is what the production code branches on.
|
|
pub async fn carry_zip_forward(pool: &PgPool, event_id: Uuid, epoch: i64) -> bool {
|
|
sqlx::query(
|
|
"UPDATE export_job SET epoch = $2
|
|
WHERE event_id = $1 AND type = 'zip'::export_type
|
|
AND status = 'done' AND epoch = $2 - 1",
|
|
)
|
|
.bind(event_id)
|
|
.bind(epoch)
|
|
.execute(pool)
|
|
.await
|
|
.expect("carry_zip_forward")
|
|
.rows_affected()
|
|
== 1
|
|
}
|
|
|
|
/// SRC: `services/export.rs::invalidate_and_arm` — the epoch bump, verbatim.
|
|
pub async fn bump_epoch(pool: &PgPool, slug: &str) -> Option<(Uuid, String, i64)> {
|
|
sqlx::query_as(
|
|
"UPDATE event SET export_epoch = export_epoch + 1
|
|
WHERE slug = $1 AND export_released_at IS NOT NULL
|
|
RETURNING id, name, export_epoch",
|
|
)
|
|
.bind(slug)
|
|
.fetch_optional(pool)
|
|
.await
|
|
.expect("bump_epoch")
|
|
}
|
|
|
|
/// The event's authoritative epoch.
|
|
pub async fn event_epoch(pool: &PgPool, event_id: Uuid) -> i64 {
|
|
sqlx::query_scalar("SELECT export_epoch FROM event WHERE id = $1")
|
|
.bind(event_id)
|
|
.fetch_one(pool)
|
|
.await
|
|
.expect("event_epoch")
|
|
}
|
|
|
|
/// The raw job row, bypassing `export_current` — what the WORKER sees.
|
|
pub async fn job_row(
|
|
pool: &PgPool,
|
|
event_id: Uuid,
|
|
export_type: &str,
|
|
) -> Option<(String, i64, Option<String>)> {
|
|
sqlx::query_as(
|
|
"SELECT status::text, epoch, file_path FROM export_job
|
|
WHERE event_id = $1 AND type = $2::export_type",
|
|
)
|
|
.bind(event_id)
|
|
.bind(export_type)
|
|
.fetch_optional(pool)
|
|
.await
|
|
.expect("job_row")
|
|
}
|
|
|
|
/// Does `export_current` expose this job at all? (The view itself, without the `status` filter —
|
|
/// it is what `handlers/admin.rs::export_status` reports to the host UI.)
|
|
pub async fn in_export_current(pool: &PgPool, event_id: Uuid, export_type: &str) -> bool {
|
|
sqlx::query_scalar::<_, i64>(
|
|
"SELECT COUNT(*) FROM export_current
|
|
WHERE event_id = $1 AND type = $2::export_type",
|
|
)
|
|
.bind(event_id)
|
|
.bind(export_type)
|
|
.fetch_one(pool)
|
|
.await
|
|
.expect("in_export_current")
|
|
> 0
|
|
}
|
|
|
|
/// THE download predicate. SRC: `handlers/admin.rs::download_export` reads exactly this shape —
|
|
/// `SELECT c.file_path FROM export_current c WHERE ... AND c.status = 'done'`. If this returns
|
|
/// `Some`, a guest can download the keepsake; if `None`, they get a 404.
|
|
pub async fn downloadable(pool: &PgPool, event_id: Uuid, export_type: &str) -> Option<String> {
|
|
sqlx::query_scalar(
|
|
"SELECT c.file_path FROM export_current c
|
|
WHERE c.event_id = $1 AND c.type = $2::export_type AND c.status = 'done'",
|
|
)
|
|
.bind(event_id)
|
|
.bind(export_type)
|
|
.fetch_optional(pool)
|
|
.await
|
|
.expect("downloadable")
|
|
.flatten()
|
|
}
|
|
|
|
/// Insert an upload of `size` bytes, optionally already soft-deleted.
|
|
pub async fn seed_upload(
|
|
pool: &PgPool,
|
|
event_id: Uuid,
|
|
user_id: Uuid,
|
|
size: i64,
|
|
deleted: bool,
|
|
) -> Uuid {
|
|
sqlx::query_scalar(
|
|
"INSERT INTO upload (event_id, user_id, original_path, mime_type,
|
|
original_size_bytes, deleted_at)
|
|
VALUES ($1, $2, 'originals/x.jpg', 'image/jpeg', $3,
|
|
CASE WHEN $4 THEN NOW() ELSE NULL END)
|
|
RETURNING id",
|
|
)
|
|
.bind(event_id)
|
|
.bind(user_id)
|
|
.bind(size)
|
|
.bind(deleted)
|
|
.fetch_one(pool)
|
|
.await
|
|
.expect("seed upload")
|
|
}
|
|
|
|
/// Flip the moderation flags a ban sets.
|
|
pub async fn set_user_moderation(pool: &PgPool, user_id: Uuid, banned: bool, hidden: bool) {
|
|
sqlx::query("UPDATE \"user\" SET is_banned = $2, uploads_hidden = $3 WHERE id = $1")
|
|
.bind(user_id)
|
|
.bind(banned)
|
|
.bind(hidden)
|
|
.execute(pool)
|
|
.await
|
|
.expect("set moderation");
|
|
}
|
|
|
|
/// SRC: `services/export.rs::query_uploads` — the visibility filter, verbatim, projected down to
|
|
/// `(id, original_size_bytes)`. This is the row set that ACTUALLY lands in the archives.
|
|
///
|
|
/// Production builds this WHERE from `export_visibility_where!()`, shared with
|
|
/// `estimate_export_bytes`. A copy here can pin the behaviour but CANNOT detect production moving
|
|
/// away from it — that is what sharing the fragment is for, not this.
|
|
pub async fn export_visible_uploads(pool: &PgPool, event_id: Uuid) -> Vec<(Uuid, i64)> {
|
|
sqlx::query_as(
|
|
"SELECT u.id, u.original_size_bytes
|
|
FROM upload u
|
|
JOIN \"user\" usr ON usr.id = u.user_id
|
|
WHERE u.event_id = $1 AND u.deleted_at IS NULL
|
|
AND usr.uploads_hidden = FALSE AND usr.is_banned = FALSE
|
|
GROUP BY u.id, usr.display_name
|
|
ORDER BY u.created_at ASC",
|
|
)
|
|
.bind(event_id)
|
|
.fetch_all(pool)
|
|
.await
|
|
.expect("export_visible_uploads")
|
|
}
|
|
|
|
/// SRC: `services/export.rs::estimate_export_bytes` — verbatim. Same caveat as above: production
|
|
/// shares its WHERE with `query_uploads` via `export_visibility_where!()`, so these two copies
|
|
/// agreeing proves the behaviour, not the absence of drift.
|
|
pub async fn estimate_export_bytes(pool: &PgPool, event_id: Uuid) -> i64 {
|
|
let (bytes,): (i64,) = sqlx::query_as(
|
|
"SELECT COALESCE(SUM(u.original_size_bytes), 0)::bigint
|
|
FROM upload u
|
|
JOIN \"user\" usr ON usr.id = u.user_id
|
|
WHERE u.event_id = $1 AND u.deleted_at IS NULL
|
|
AND usr.uploads_hidden = FALSE AND usr.is_banned = FALSE",
|
|
)
|
|
.bind(event_id)
|
|
.fetch_one(pool)
|
|
.await
|
|
.expect("estimate_export_bytes");
|
|
bytes
|
|
}
|
|
|
|
/// SRC: `services/export.rs::ensure_export_space` — the armed-job count, verbatim.
|
|
pub async fn armed_job_count(pool: &PgPool, event_id: Uuid) -> i64 {
|
|
let (n,): (i64,) = sqlx::query_as(
|
|
"SELECT COUNT(*) FROM export_job
|
|
WHERE event_id = $1 AND status IN ('pending', 'running')",
|
|
)
|
|
.bind(event_id)
|
|
.fetch_one(pool)
|
|
.await
|
|
.expect("armed_job_count");
|
|
n
|
|
}
|