fix(files): a file is added when it is done, not when it was asked for
A download that failed for days and then went through sat days down the library, under its request date, and read as missing. A job now records when it became done (`ready_at`), stamped where the expiry already is (`folders.stamp_done`, `finalize_if_running`) and on an insert that is already done; a check constraint holds every path to done to it. The library sorts and dates its "Added" by it; the queue keeps the request order. The migration dates a file from before by its "ready to watch" notice where the inbox still holds one, else by its request date, where it already sat.
This commit is contained in:
@@ -0,0 +1,44 @@
|
||||
"""A file's "Added" is when it became done
|
||||
|
||||
Revision ID: 0104_job_ready_at
|
||||
Revises: 0103_files_sha256
|
||||
Create Date: 2026-10-12
|
||||
|
||||
`download_jobs.ready_at` is when a job became done, and the Files library sorts and dates by it: a
|
||||
download that failed for days and then went through used to sit days down the list, under its
|
||||
request date. Every path to done stamps it, held by `ck_download_jobs_ready_at`.
|
||||
|
||||
A done job from before has no record of that moment, except where its "ready to watch" notice is
|
||||
still in the owner's inbox (read notices are pruned past a cap): that notice was written the moment
|
||||
it finished. The rest take the request date, which is where they already sat.
|
||||
"""
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
revision: str = "0104_job_ready_at"
|
||||
down_revision: str | None = "0103_files_sha256"
|
||||
branch_labels: str | None = None
|
||||
depends_on: str | None = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.add_column("download_jobs", sa.Column("ready_at", sa.DateTime(timezone=True)))
|
||||
op.execute(
|
||||
"""
|
||||
UPDATE download_jobs j SET ready_at = COALESCE(
|
||||
(SELECT max(n.created_at) FROM notifications n
|
||||
WHERE n.type = 'download_done' AND n.user_id = j.user_id
|
||||
AND n.data->>'job_id' = j.id::text),
|
||||
j.created_at)
|
||||
WHERE j.status = 'done'
|
||||
"""
|
||||
)
|
||||
op.create_check_constraint(
|
||||
"ck_download_jobs_ready_at", "download_jobs", "status <> 'done' OR ready_at IS NOT NULL"
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_constraint("ck_download_jobs_ready_at", "download_jobs", type_="check")
|
||||
op.drop_column("download_jobs", "ready_at")
|
||||
@@ -16,7 +16,7 @@ file in the TikTok folder would be a lie the folder's name tells.
|
||||
Retention is the folder's: a source folder keeps a file `download_retention_days` (the value when the
|
||||
folder was made; the user can change it), Uploads and own folders keep it until it is deleted. A
|
||||
done file carries its own `expires_at`, stamped by `done_expiry` when it finishes — unless the user
|
||||
kept it.
|
||||
kept it — and its `ready_at`, the moment it finished (`stamp_done`).
|
||||
"""
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
@@ -136,8 +136,9 @@ def done_expiry(db: Session, folder_id: int | None, kept: bool) -> datetime | No
|
||||
return None if days is None else datetime.now(timezone.utc) + timedelta(days=days)
|
||||
|
||||
|
||||
def stamp_expiry(db: Session, job: DownloadJob) -> None:
|
||||
"""Stamp a job that just became done (in this session) with its expiry."""
|
||||
def stamp_done(db: Session, job: DownloadJob) -> None:
|
||||
"""Stamp a job that just became done (in this session): when it did, and its expiry."""
|
||||
job.ready_at = datetime.now(timezone.utc)
|
||||
job.expires_at = done_expiry(db, job.folder_id, job.kept)
|
||||
job.expiry_warned = False
|
||||
|
||||
@@ -163,7 +164,7 @@ def upload_target(db: Session, user_id: int, folder_id: int) -> FileFolder:
|
||||
|
||||
def restamp(db: Session, job: DownloadJob) -> None:
|
||||
"""Start a file's clock again from now: it moved to another folder, or its user stopped keeping
|
||||
it. A file not done yet gets its expiry when it finishes (`stamp_expiry`), so it has none now."""
|
||||
it. A file not done yet gets its expiry when it finishes (`stamp_done`), so it has none now."""
|
||||
done = job.status == JobStatus.DONE
|
||||
job.expires_at = done_expiry(db, job.folder_id, job.kept) if done else None
|
||||
job.expiry_warned = False
|
||||
|
||||
@@ -205,7 +205,7 @@ def enqueue(
|
||||
if asset.status == "ready":
|
||||
job.status = JobStatus.DONE
|
||||
job.progress = 100
|
||||
folders.stamp_expiry(db, job)
|
||||
folders.stamp_done(db, job)
|
||||
asset.last_access_at = datetime.now(timezone.utc)
|
||||
|
||||
db.commit()
|
||||
@@ -285,7 +285,7 @@ def enqueue_edit(
|
||||
if asset.status == "ready": # identical clip already produced for this user → instant
|
||||
job.status = JobStatus.DONE
|
||||
job.progress = 100
|
||||
folders.stamp_expiry(db, job)
|
||||
folders.stamp_done(db, job)
|
||||
asset.last_access_at = datetime.now(timezone.utc)
|
||||
|
||||
db.commit()
|
||||
@@ -374,7 +374,7 @@ def enqueue_clip(
|
||||
if asset.status == "ready": # the same range was already cut for this user → instant
|
||||
job.status = JobStatus.DONE
|
||||
job.progress = 100
|
||||
folders.stamp_expiry(db, job)
|
||||
folders.stamp_done(db, job)
|
||||
asset.last_access_at = datetime.now(timezone.utc)
|
||||
|
||||
db.commit()
|
||||
@@ -399,6 +399,7 @@ def finalize_if_running(db: Session, job_id: int, **fields) -> bool:
|
||||
return False
|
||||
db.flush() # asset-then-job lock order: push pending asset writes before we lock the job row
|
||||
if fields.get("status") == JobStatus.DONE:
|
||||
fields = {**fields, "ready_at": datetime.now(timezone.utc)}
|
||||
# Done is when a file's retention starts: its folder's, unless the user kept it.
|
||||
held = db.execute(
|
||||
select(DownloadJob.folder_id, DownloadJob.kept).where(DownloadJob.id == job_id)
|
||||
|
||||
@@ -140,7 +140,7 @@ def ingest_upload(
|
||||
if folder_id is not None:
|
||||
job.folder_id = folder_id
|
||||
if not convert:
|
||||
folders.stamp_expiry(db, job)
|
||||
folders.stamp_done(db, job)
|
||||
service.incr_ref_count(db, asset.id)
|
||||
db.flush()
|
||||
audit.record(db, user_id, AuditAction.FILE_UPLOAD, target_type="file", target_id=job.id)
|
||||
|
||||
+21
-2
@@ -1,4 +1,4 @@
|
||||
from datetime import date, datetime
|
||||
from datetime import date, datetime, timezone
|
||||
|
||||
from sqlalchemy import (
|
||||
BigInteger,
|
||||
@@ -949,6 +949,13 @@ class FileFolder(Base, TimestampMixin):
|
||||
retention_days: Mapped[int | None] = mapped_column(Integer)
|
||||
|
||||
|
||||
def _ready_on_insert(context) -> datetime | None:
|
||||
"""A job inserted already done (an upload kept as it came) was ready the moment it was made; one
|
||||
that becomes done later is stamped then (`folders.stamp_done`, `service.finalize_if_running`)."""
|
||||
done = context.get_current_parameters().get("status") == "done"
|
||||
return datetime.now(timezone.utc) if done else None
|
||||
|
||||
|
||||
class DownloadJob(Base, TimestampMixin, UpdatedAtMixin):
|
||||
"""A user's request to download one source — the per-user layer over `MediaAsset`.
|
||||
|
||||
@@ -967,7 +974,13 @@ class DownloadJob(Base, TimestampMixin, UpdatedAtMixin):
|
||||
point at the parent download the clip was cut from; `edit_spec` is the (trim/crop) recipe."""
|
||||
|
||||
__tablename__ = "download_jobs"
|
||||
__table_args__ = (Index("ix_download_jobs_claim", "status", "queue_pos"),)
|
||||
__table_args__ = (
|
||||
Index("ix_download_jobs_claim", "status", "queue_pos"),
|
||||
# Every way a job becomes done stamps when (`folders.stamp_done`, `finalize_if_running`).
|
||||
CheckConstraint(
|
||||
"status <> 'done' OR ready_at IS NOT NULL", name="ck_download_jobs_ready_at"
|
||||
),
|
||||
)
|
||||
|
||||
id: Mapped[int] = mapped_column(primary_key=True)
|
||||
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
|
||||
@@ -1026,6 +1039,12 @@ class DownloadJob(Base, TimestampMixin, UpdatedAtMixin):
|
||||
kept: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
|
||||
expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), index=True)
|
||||
expiry_warned: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
|
||||
# When the file became done — what the library's "Added" means: a download that failed for days
|
||||
# and then went through is new to its user, so it sorts as new. `created_at` stays when it was
|
||||
# asked for (the queue's order). Set again when a file is made anew (a retry, a re-download).
|
||||
ready_at: Mapped[datetime | None] = mapped_column(
|
||||
DateTime(timezone=True), default=_ready_on_insert
|
||||
)
|
||||
|
||||
|
||||
class DownloadShare(Base, TimestampMixin):
|
||||
|
||||
@@ -78,6 +78,7 @@ def serialize(job: DownloadJob, asset: MediaAsset | None) -> dict:
|
||||
"error_detail": job.error_detail,
|
||||
"queue_pos": job.queue_pos,
|
||||
"created_at": iso(job.created_at),
|
||||
"ready_at": iso(job.ready_at),
|
||||
"source_kind": job.source_kind,
|
||||
"source_ref": job.source_ref,
|
||||
"source_url": service.reference_url(job, asset),
|
||||
@@ -545,10 +546,10 @@ def list_files(
|
||||
func.coalesce(DownloadJob.display_uploader, MediaAsset.uploader, "").ilike(like),
|
||||
)
|
||||
)
|
||||
# A file shared with the caller was added to THEIR list when it was shared, not when its owner
|
||||
# downloaded it.
|
||||
# A file is added to its owner's list when it is done, not when it was asked for (a download that
|
||||
# failed for days and then went through is new); one shared with the caller, when it was shared.
|
||||
key = {
|
||||
"added": DownloadShare.created_at if shared else DownloadJob.created_at,
|
||||
"added": DownloadShare.created_at if shared else DownloadJob.ready_at,
|
||||
"name": collated(name, lang_for_user(user)),
|
||||
"size": MediaAsset.size_bytes,
|
||||
}[sort]
|
||||
|
||||
@@ -374,7 +374,7 @@ def test_a_clip_is_not_redownloadable(db, make_user, as_user, yt_video):
|
||||
c = as_user(make_user())
|
||||
job = _clip(c, source="youtube", ref=yt_video).json()
|
||||
row = db.get(DownloadJob, job["id"])
|
||||
row.status = "done"
|
||||
row.status, row.ready_at = "done", row.created_at
|
||||
db.commit()
|
||||
r = c.post(f"/api/downloads/{job['id']}/redownload")
|
||||
assert (r.status_code, r.json()["detail"]["reason"]) == (409, "not_redownloadable")
|
||||
|
||||
@@ -59,7 +59,7 @@ def _file(db, user_id, *, kind="youtube", ref=None, title="A title", size=10, re
|
||||
db.add(job)
|
||||
db.flush()
|
||||
folders.place(db, job)
|
||||
folders.stamp_expiry(db, job)
|
||||
folders.stamp_done(db, job)
|
||||
service.incr_ref_count(db, asset.id)
|
||||
db.commit()
|
||||
return job
|
||||
@@ -245,6 +245,38 @@ def test_the_list_sorts_by_name_size_or_when_added(db, make_user, as_user):
|
||||
assert ids(sort="size", order="desc") == [b.id, z.id, a.id]
|
||||
|
||||
|
||||
def test_a_file_is_added_when_it_is_done_not_when_it_was_asked_for(db, make_user, as_user):
|
||||
"""A download that failed for days and then went through is new: it lists first, dated today."""
|
||||
alice = make_user()
|
||||
early = _file(db, alice.id, created_at=NOW - timedelta(days=1))
|
||||
early.ready_at = NOW - timedelta(days=1)
|
||||
late = _file(db, alice.id, status=JobStatus.RUNNING, created_at=NOW - timedelta(days=5))
|
||||
late.ready_at = None
|
||||
db.commit()
|
||||
assert service.finalize_if_running(db, late.id, status=JobStatus.DONE)
|
||||
db.commit()
|
||||
db.expire_all() # the finalize is a Core UPDATE; the route reads this same session
|
||||
got = as_user(alice).get("/api/files").json()
|
||||
assert [f["id"] for f in got] == [late.id, early.id]
|
||||
ready = datetime.fromisoformat(got[0]["ready_at"])
|
||||
assert ready > NOW - timedelta(minutes=5)
|
||||
assert datetime.fromisoformat(got[0]["created_at"]) < NOW - timedelta(days=4)
|
||||
|
||||
|
||||
def test_a_job_made_done_is_ready_as_it_is_made(db, make_user):
|
||||
# An upload kept as it came is inserted done; the database refuses a done job without the moment.
|
||||
job = DownloadJob(
|
||||
user_id=make_user().id,
|
||||
source_kind="upload",
|
||||
source_ref="r",
|
||||
profile_snapshot={},
|
||||
status=JobStatus.DONE,
|
||||
)
|
||||
db.add(job)
|
||||
db.commit()
|
||||
assert job.ready_at is not None
|
||||
|
||||
|
||||
def test_shared_with_me_lists_what_others_shared_and_nothing_else(db, make_user, as_user):
|
||||
alice, bob, carol = make_user(), make_user(), make_user()
|
||||
theirs = _file(db, bob.id)
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
"""Migration 0104's data step, DB-lane: a done file from before is dated by when it finished where
|
||||
its "ready to watch" notice says so, else by when it was asked for; a job not done has no date."""
|
||||
|
||||
from alembic import command
|
||||
from alembic.config import Config
|
||||
from sqlalchemy import text
|
||||
|
||||
from app.db import engine
|
||||
|
||||
|
||||
def test_a_done_file_takes_its_notice_time_else_its_request_time(db, make_user):
|
||||
uid = make_user().id
|
||||
cfg = Config("alembic.ini")
|
||||
command.downgrade(cfg, "0103_files_sha256")
|
||||
try:
|
||||
with engine.begin() as conn:
|
||||
|
||||
def job(status, created):
|
||||
return conn.execute(
|
||||
text(
|
||||
"INSERT INTO download_jobs (user_id, source_kind, source_ref,"
|
||||
" profile_snapshot, status, created_at) VALUES (:u, 'url', 'x', '{}', :s,"
|
||||
" CAST(:c AS timestamptz)) RETURNING id"
|
||||
),
|
||||
{"u": uid, "s": status, "c": created},
|
||||
).scalar_one()
|
||||
|
||||
retried = job("done", "2026-10-01T10:00:00+00:00")
|
||||
quiet = job("done", "2026-10-02T10:00:00+00:00")
|
||||
failed = job("error", "2026-10-03T10:00:00+00:00")
|
||||
for at in ("2026-10-05T09:00:00+00:00", "2026-10-06T09:00:00+00:00"):
|
||||
conn.execute(
|
||||
text(
|
||||
"INSERT INTO notifications (user_id, type, title, data, created_at) VALUES"
|
||||
" (:u, 'download_done', 't', CAST(:d AS json), CAST(:c AS timestamptz))"
|
||||
),
|
||||
{"u": uid, "d": f'{{"job_id": {retried}}}', "c": at},
|
||||
)
|
||||
finally:
|
||||
command.upgrade(cfg, "head")
|
||||
|
||||
with engine.connect() as conn:
|
||||
rows = dict(
|
||||
conn.execute(
|
||||
text(
|
||||
"SELECT id, to_char(ready_at AT TIME ZONE 'UTC', 'MM-DD HH24') FROM download_jobs"
|
||||
)
|
||||
).all()
|
||||
)
|
||||
assert rows[retried] == "10-06 09" # its last finish
|
||||
assert rows[quiet] == "10-02 10" # no notice left: where it already sat
|
||||
assert rows[failed] is None
|
||||
@@ -320,7 +320,7 @@ def test_upload_route_media_html_extension_not_served_as_html(
|
||||
assert job["status"] == "queued"
|
||||
row = db.get(DownloadJob, job["id"])
|
||||
db.get(MediaAsset, row.asset_id).status = "ready"
|
||||
row.status = "done"
|
||||
row.status, row.ready_at = "done", row.created_at
|
||||
db.commit()
|
||||
|
||||
# Authenticated endpoint: never text/html.
|
||||
|
||||
@@ -407,7 +407,7 @@ def test_the_sweep_keeps_an_aged_upload_file_its_row_names_in_any_state(
|
||||
_probe(monkeypatch)
|
||||
job, asset = _ingest(db, _root, make_user().id)
|
||||
path = storage.abs_path(str(_root), asset.rel_path)
|
||||
job.status, asset.status = job_status, asset_status
|
||||
job.status, asset.status, job.ready_at = job_status, asset_status, job.created_at
|
||||
db.commit()
|
||||
_aged(path)
|
||||
|
||||
@@ -427,7 +427,11 @@ def _route_moved(db, job, asset, job_status: str, ref_count: int) -> None:
|
||||
"""What the cancel/pause/delete route leaves behind, written as SQL: the session's copy of
|
||||
`ref_count` is stale (the routes count it in SQL), so an ORM assignment of the same value would
|
||||
write nothing."""
|
||||
db.execute(update(DownloadJob).where(DownloadJob.id == job.id).values(status=job_status))
|
||||
db.execute(
|
||||
update(DownloadJob)
|
||||
.where(DownloadJob.id == job.id)
|
||||
.values(status=job_status, ready_at=DownloadJob.created_at)
|
||||
)
|
||||
db.execute(update(MediaAsset).where(MediaAsset.id == asset.id).values(ref_count=ref_count))
|
||||
db.commit()
|
||||
|
||||
@@ -609,6 +613,7 @@ def test_the_admin_storage_view_names_each_users_quota_and_files(
|
||||
done, _ = _ingest(db, _root, u.id, size=10)
|
||||
waiting, _ = _ingest(db, _root, u.id, name="b.mov", size=5)
|
||||
done.status, waiting.status = "done", "queued"
|
||||
done.ready_at = done.created_at
|
||||
db.commit()
|
||||
admin = as_user(make_user(role="admin"))
|
||||
put = admin.put(f"/api/admin/downloads/quota/{u.id}", json={"max_bytes": 1000})
|
||||
|
||||
@@ -11,6 +11,7 @@ import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { act, cleanup, fireEvent, render, screen, waitFor } from "@testing-library/react";
|
||||
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
|
||||
import { api, type FileFolder, type LibraryFile } from "../lib/api";
|
||||
import { formatDateTime } from "../lib/format";
|
||||
import {
|
||||
ALL_FILES,
|
||||
FilesActionsContext,
|
||||
@@ -496,6 +497,32 @@ describe("Files · the view", () => {
|
||||
expect(screen.queryByText("Más")).not.toBeNull();
|
||||
expect(screen.queryByText("Harmadik")).toBeNull();
|
||||
});
|
||||
|
||||
it("dates and sorts a file by when it was done, not when it was asked for", () => {
|
||||
// Done in another order than asked: a download that failed for days and then went through is new.
|
||||
mocks.files = [
|
||||
file(2, { created_at: "2026-10-01T10:00:00Z", ready_at: "2026-10-05T10:00:00Z" }),
|
||||
file(3, { created_at: "2026-10-02T10:00:00Z", ready_at: "2026-10-12T10:00:00Z" }),
|
||||
file(4, { created_at: "2026-10-03T10:00:00Z", ready_at: "2026-10-03T10:00:00Z" }),
|
||||
];
|
||||
mocks.folders = [YT];
|
||||
mount();
|
||||
const order = () =>
|
||||
[...document.querySelectorAll<HTMLElement>("[data-nav-index]")].map(
|
||||
(r) => r.textContent?.match(/Clip \d/)?.[0],
|
||||
);
|
||||
const byDone = [
|
||||
["Clip 4", "Clip 2", "Clip 3"],
|
||||
["Clip 3", "Clip 2", "Clip 4"],
|
||||
];
|
||||
fireEvent.click(screen.getByRole("button", { name: /files\.columns\.added/ }));
|
||||
expect(byDone).toContainEqual(order());
|
||||
fireEvent.click(screen.getByRole("button", { name: /files\.columns\.added/ }));
|
||||
expect(byDone).toContainEqual(order());
|
||||
expect(
|
||||
rowOf("Clip 3").querySelector(`[title="${formatDateTime("2026-10-12T10:00:00Z", "en")}"]`),
|
||||
).not.toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe("Files · the grid", () => {
|
||||
|
||||
@@ -402,9 +402,9 @@ export default function Files() {
|
||||
align: "right",
|
||||
nowrap: true,
|
||||
sortable: true,
|
||||
sortValue: (f) => f.shared_at ?? f.created_at ?? "",
|
||||
sortValue: (f) => f.shared_at ?? f.ready_at ?? f.created_at ?? "",
|
||||
render: (f) => {
|
||||
const at = f.shared_at ?? f.created_at;
|
||||
const at = f.shared_at ?? f.ready_at ?? f.created_at;
|
||||
return (
|
||||
<span className="text-xs text-muted" title={formatDateTime(at, lang)}>
|
||||
{relativeTime(at, lang, t)}
|
||||
|
||||
@@ -94,7 +94,8 @@ export interface DownloadJob {
|
||||
error_code: string | null; // failure category, see backend downloads.errors.classify
|
||||
error_detail: string | null; // host-specific failure-catalog entry, see downloads.errors.diagnose
|
||||
queue_pos: number;
|
||||
created_at: string | null;
|
||||
created_at: string | null; // when it was asked for
|
||||
ready_at?: string | null; // when it became done — the library's "Added"
|
||||
source_kind: string;
|
||||
source_ref: string;
|
||||
source_url: string | null; // clean "downloaded from" URL (null for edit clips)
|
||||
|
||||
Reference in New Issue
Block a user