diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md index 3d3d8e7..588553a 100644 --- a/daemon/docs/deferrals.md +++ b/daemon/docs/deferrals.md @@ -7,7 +7,11 @@ close. Kept here (not buried in commit messages) so the next pass can see them a |---|---|---|---|---| | D1 | Pairing prompt is `EnvAutoApprover` (needs `VELOX_PAIR_AUTO=1`) | `rpc/pairing.hpp`, `main.cpp` | A GUI dialog / `org.freedesktop.Notifications` approver is integration work | Build step 7 (systemd + notifications) | | ~~D2~~ | **Closed** — `download.probe` is real on both transports. It's genuinely async (the engine's probe pool, up to the schema's 30s `x-deadlineMs`) and so cannot fit `VeloxDispatcher::on_download_probe`'s synchronous `HandlerResult` return — `uds_server.cpp`/`ws_server.cpp` special-case `"download.probe"` before the generic `dispatch()`, exactly the way they already special-case `session.hello`/`session.subscribe`, and queue the reply whenever the callback fires. `rpc::TaskActionPort::probe_now` (kept in proto/std terms, no `vdm::net::*`, so `veloxd_rpc` never needs `core/include`'s vdm headers) is what both transports call; `sched::Scheduler::probe_now` is the implementation — builds a `vdm::net::ProbeRequest`, runs it on the engine's probe pool, maps a failure to `-32013 ProbeFailed` (with `data.httpStatus` when there was one), and fills `suggestedCategoryId`/`suggestedSaveDir` with a plain extension match against the categories table (not the real rules engine — that's still D3). Verified live: a real probe answers in ~5ms; a bad host maps to `-32013`; a connection issuing a 10s `slow-loris` probe does not block a second connection's `download.list` (answered in ~1ms) — confirms the async design actually keeps the loop free, not just compiles. | `rpc/task_action_port.hpp`, `rpc/{uds_server,ws_server}.{hpp,cpp}`, `sched/scheduler.{cpp,hpp}` | — | done | -| D3 | Stub handlers for the rest: `download.remove/addBatch/refreshUrl/provideAuth/update`, `rules.*`, `settings.*`, `limiter.*`, `schedule.*`, `queue.upsert/reorder`, `category.upsert/remove`, `grabber.*`, `media.*`, `capture.*` | `rpc/dispatcher.cpp` | No store/scheduler wiring behind them yet. `category.list`, `queue.list`, `queue.start`/`stop` are done | Per method, as each wires to the store/scheduler | +| D3 | Stub handlers for the rest: `download.refreshUrl/update`, `rules.*`, `settings.*`, `limiter.*`, `schedule.*`, `queue.reorder`, `grabber.*`, `media.*`, `capture.*` | `rpc/dispatcher.cpp` | No store/scheduler wiring behind them yet, or (`settings.*`) sound but large — see the note below. `category.list/upsert/remove`, `queue.list/upsert/start/stop`, `download.remove/addBatch/provideAuth` are done | Per method, as each wires to the store/scheduler | +| — | **`settings.get`/`settings.set` specifically, not started:** `proto::Settings` is a flat struct of ~43 `std::optional` fields, one per `SettingKey` (~50 keys) in `Settings.schema.json`; `store::Settings` already has `get_raw`/`set_raw`/`overrides` keyed by the same dotted strings the JSON uses (`"connection.maxSegmentsPerDownload"`, …). The handlers are a mechanical field <-> key <-> JSON-type mapping table in both directions (get: row-or-default -> struct field; set: struct field -> validate against the key's schema type -> `set_raw`, collecting `changed`) — real work, just long and repetitive rather than hard. Left alone this pass rather than rushed; every other read of settings in this codebase already goes through `store::Settings`'s typed helpers directly (`reload_config`, `on_download_add`'s segment default, `capture.offer`'s allowed roots when that lands), so nothing downstream is blocked on the RPC surface existing. | `rpc/dispatcher.cpp`, `store/settings.{hpp,cpp}` | the mapping table is genuinely large, not genuinely hard | its own pass | +| ~~D3a~~ | **Closed** — `category.upsert`/`category.remove`: `store/categories.hpp` gains `get`/`upsert`/`remove`. `upsert` generates an id when absent (create) and always ignores the payload's `builtin` (preserved from the existing row on replace, false on create — a client can never mint or revoke it); the `saveDir` goes through the same `fs::resolve_target` canonicalize-and-root-check as `download.add` (`-32011` on failure). `remove` refuses a builtin at both layers (dispatcher pre-checks for the `-32602` error text; the store's own `DELETE ... AND builtin = 0` is defense in depth) and reassigns member tasks to `reassignTo` (default `"general"`) inside one transaction before deleting the row. Note: the `categories` table (0001) has no columns for `Category.mimeTypes`/`.sortOrder` — accepted on `upsert` but not persisted. | `store/categories.{hpp,cpp}`, `rpc/dispatcher.cpp` | — | done, `mimeTypes`/`sortOrder` gap noted | +| ~~D3b~~ | **Closed** — `queue.upsert`: `store/queues.hpp` gains `get`/`upsert` (`set_state` already existed from D4b). Same create-generates-id pattern as categories; `taskIds` in the payload is ignored (schema's own note) and a create always starts `'stopped'` while a replace keeps the queue's current run state — `queue.upsert` edits config, not run state (that's `queue.start`/`stop`). Also fixed: `on_complete` was a real column since 0001 but `Queues::list`/`get` never projected it onto `Queue.onComplete` — now they do. | `store/queues.{hpp,cpp}`, `rpc/dispatcher.cpp` | — | done | +| ~~D3c~~ | **Closed** — `download.remove`: cancels with `discard_partial=true` through `TaskActionPort` (always drops any `.veloxpart`/`.veloxpart.meta` — the row is gone either way, unlike `download.cancel`, which keeps them), deletes the finished file only when `deleteFile` is true and the task was `complete` (best-effort — a missing file doesn't fail the call), deletes the row (segments cascade via the FK), and publishes `event.task.removed` (closing the last open note under D5). `download.addBatch`: `on_download_add`'s body is now a shared `add_one()`, called once per item after merging each item's unset fields against `params.defaults`. `download.provideAuth`: forwards to `EnginePort::provide_auth` through a new `TaskActionPort::provide_auth`; `remember`/persisting to the Secret Service is accepted but not acted on — nothing in this build talks to libsecret yet (verified: no such integration exists anywhere in the tree). Verified against real `veloxd` + `tools/testserver`: category create/replace/remove-with-reassignment, queue create/replace-keeps-state, a batch add with shared `defaults.saveDir`, and remove-with-deleteFile actually deleting the file and the task then 404ing `download.get` with `-32010`. | `rpc/dispatcher.{hpp,cpp}`, `rpc/task_action_port.hpp`, `sched/scheduler.{cpp,hpp}` | `download.provideAuth`'s `remember` (needs the Secret Service, unbuilt) | done, `remember` persistence gap noted | | ~~D4a~~ | **Closed** — `sched/engine_port_core.hpp` wraps `vdm::Engine` + `segment_budget()`; `main.cpp` constructs `Engine` + `Scheduler`, calls `reconcile_after_restart` / `reload_config` / `tick` at startup | — | — | done (`lane/core` stage 8 merged) | | ~~D4b~~ | **Closed** — `download.pause`/`resume`/`start`/`cancel` and `queue.start`/`stop` all drive the scheduler now, and apply *immediately* (not deferred to the next tick — pausing/resuming/cancelling a live transfer can't wait up to 1s, and per ADR 0013 §3 the governor never touches a user-owned pause on its own). New `rpc::TaskActionPort` interface (owned by `rpc/`, implemented by `sched::Scheduler`) is the seam dispatcher.hpp depends on instead of `sched/scheduler.hpp` directly — avoids a real `veloxd_rpc` <-> `veloxd_sched` circular library dependency (`veloxd_sched` already links `veloxd_rpc` for `EventHub`). `Scheduler::user_pause/resume/start/cancel` + `pause_queue` engine-call-then-eager-transition, matching `tick()`'s existing `to_pause` pattern. Fixed a real bug hit while building this: `transition()` always overwrote `pause_reason` to NULL when the engine's own delayed pause-ack callback arrived with no explicit reason, clobbering whatever the actual initiator (user or governor) had just written — now it preserves the stored reason when none is supplied. Verified against real `veloxd` + `tools/testserver`: pausing a live single-segment throttled transfer freezes `downloadedBytes`, resume continues it from that point, cancel stops it; `queue.stop(pauseRunning:true)` pauses the queue's running task immediately. NOTE: `download.start`'s contract "a task in 'queued' jumps its queue" (priority bump) is not implemented — admission is still plain FIFO by `created_at`. | `sched/scheduler.{cpp,hpp}`, `rpc/task_action_port.hpp`, `rpc/dispatcher.{hpp,cpp}`, `store/queues.{cpp,hpp}` | — | done, except the queue-jump priority bump noted above | | ~~D5~~ | **Mostly closed** — `rpc/event_hub` fans out per-subscription; `session.subscribe` on both transports registers/updates/tears down a real subscription; `Scheduler::transition()` publishes `event.task.state` (with `previousState`) on every state change, scheduler-driven or engine-reported; `dispatcher::on_download_add` publishes `event.task.added`; a 250 ms timer batches `Scheduler::progress_snapshot()` into one `event.task.progress` array per AGENT-DAEMON.md item 5 / the schema's `x-maxRateHz: 4`. Verified live end to end. | — | `event.task.removed` has no source yet (`download.remove` is D3); `event.speed.global`, `event.notify`, `event.auth.required`, `event.settings.changed`, `event.grabber.progress` are unpublished — each lands with its owning handler | as each owning D3 handler lands | diff --git a/daemon/src/rpc/dispatcher.cpp b/daemon/src/rpc/dispatcher.cpp index 9f11f5d..c88d8be 100644 --- a/daemon/src/rpc/dispatcher.cpp +++ b/daemon/src/rpc/dispatcher.cpp @@ -1,5 +1,6 @@ #include "rpc/dispatcher.hpp" +#include #include #include @@ -141,9 +142,13 @@ VeloxDispatcher::on_download_list(const proto::DownloadListParams& params) { } // --- download.add : canonicalise + root-check the destination, then persist ----------- +// +// add_one() is the whole of download.add's body; on_download_add and on_download_addBatch +// (each item merged against `defaults` first) both call it, so there is exactly one place +// that turns a DownloadSpec into a stored, admitted task. -proto::HandlerResult -VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) { +proto::HandlerResult VeloxDispatcher::add_one( + const proto::DownloadSpec& spec) { store::Settings settings(db_); std::string save_dir = spec.saveDir && !spec.saveDir->empty() @@ -226,6 +231,11 @@ VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) { return r; } +proto::HandlerResult +VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) { + return add_one(spec); +} + // --- everything else : not implemented until the store and scheduler land ------------- proto::HandlerResult @@ -248,16 +258,94 @@ VeloxDispatcher::on_category_list(const proto::CategoryListParams&) { return r; } proto::HandlerResult -VeloxDispatcher::on_category_remove(const proto::CategoryRemoveParams&) { - return not_implemented("category.remove"); +VeloxDispatcher::on_category_remove(const proto::CategoryRemoveParams& params) { + store::Categories categories(db_); + // A builtin category refuses with -32602 (category.remove's own words) — checked here + // rather than inferred from RemoveResult.removed=false, which also covers "no such + // category" and would otherwise collapse two different error stories into one. + auto existing = categories.get(params.categoryId); + if (!existing) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "category.remove: " + existing.error().message}); + if (!existing->has_value()) + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidParams, "no such category", + nlohmann::json{{"categoryId", params.categoryId}}}); + if ((*existing)->builtin) + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidParams, "builtin categories cannot be removed", + nlohmann::json{{"categoryId", params.categoryId}}}); + + auto removed = categories.remove(params.categoryId, params.reassignTo); + if (!removed) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "category.remove: " + removed.error().message}); + proto::CategoryRemoveResult r; + r.removed = removed->removed; + r.reassignedTaskIds = removed->reassigned_task_ids; + if (on_mutation_) on_mutation_(); // download.list's categoryId column just moved + return r; } proto::HandlerResult -VeloxDispatcher::on_category_upsert(const proto::CategoryUpsertParams&) { - return not_implemented("category.upsert"); +VeloxDispatcher::on_category_upsert(const proto::CategoryUpsertParams& params) { + store::Settings settings(db_); + std::string save_dir = expand_tilde(params.category.saveDir); + + std::vector roots; + for (const auto& r : settings.get_string_array("saveTo.allowedRoots")) { + if (auto c = fs::canonicalize_root(r)) roots.push_back(*c); + } + // No file is ever written for this marker leaf — resolve_target only validates/creates + // the directory chain (fs/safepath.hpp); it never creates the leaf itself. + auto target = fs::resolve_target(save_dir, ".category-marker", roots); + if (!target) { + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidPath, target.error().message, + nlohmann::json{{"path", params.category.saveDir}}}); + } + + store::Categories categories(db_); + proto::Category to_store = params.category; + to_store.saveDir = target->dir; + auto stored = categories.upsert(std::move(to_store)); + if (!stored) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "category.upsert: " + stored.error().message}); + proto::CategoryUpsertResult r; + r.category = std::move(*stored); + return r; } proto::HandlerResult -VeloxDispatcher::on_download_addBatch(const proto::DownloadAddBatchParams&) { - return not_implemented("download.addBatch"); +VeloxDispatcher::on_download_addBatch(const proto::DownloadAddBatchParams& params) { + proto::DownloadAddBatchResult r; + for (std::size_t i = 0; i < params.items.size(); ++i) { + proto::DownloadSpec spec = params.items[i]; // "its url is ignored" only for defaults + if (params.defaults) { + const auto& d = *params.defaults; + if (!spec.headers) spec.headers = d.headers; + if (!spec.cookies) spec.cookies = d.cookies; + if (!spec.referrer) spec.referrer = d.referrer; + if (!spec.userAgent) spec.userAgent = d.userAgent; + if (!spec.filename) spec.filename = d.filename; + if (!spec.saveDir) spec.saveDir = d.saveDir; + if (!spec.categoryId) spec.categoryId = d.categoryId; + if (!spec.queueId) spec.queueId = d.queueId; + if (!spec.segments) spec.segments = d.segments; + if (!spec.bufferBytes) spec.bufferBytes = d.bufferBytes; + if (!spec.startMode) spec.startMode = d.startMode; + if (!spec.description) spec.description = d.description; + if (!spec.checksum) spec.checksum = d.checksum; + } + + auto added = add_one(spec); + if (added) { + r.taskIds.push_back(added->taskId); + } else { + r.failed.push_back({static_cast(i), added.error().code, + added.error().message}); + } + } + return r; } proto::HandlerResult VeloxDispatcher::on_download_cancel(const proto::DownloadCancelParams& params) { @@ -313,16 +401,56 @@ VeloxDispatcher::on_download_probe(const proto::DownloadProbeParams&) { return not_implemented("download.probe"); } proto::HandlerResult -VeloxDispatcher::on_download_provideAuth(const proto::DownloadProvideAuthParams&) { - return not_implemented("download.provideAuth"); +VeloxDispatcher::on_download_provideAuth(const proto::DownloadProvideAuthParams& params) { + if (!actions_) return not_implemented("download.provideAuth"); + proto::DownloadProvideAuthResult r; + r.ok = actions_->provide_auth(params.taskId, params.username, params.password, + params.save.value_or(false)); + return r; } proto::HandlerResult VeloxDispatcher::on_download_refreshUrl(const proto::DownloadRefreshUrlParams&) { return not_implemented("download.refreshUrl"); } proto::HandlerResult -VeloxDispatcher::on_download_remove(const proto::DownloadRemoveParams&) { - return not_implemented("download.remove"); +VeloxDispatcher::on_download_remove(const proto::DownloadRemoveParams& params) { + store::Tasks tasks(db_); + proto::DownloadRemoveResult r; + for (const auto& id : params.taskIds) { + auto got = tasks.get(id); + if (!got || !got->has_value()) { + r.failed.push_back({id, proto::ErrorCode::TaskNotFound, "no such task"}); + continue; + } + const store::TaskRow row = **got; // copy: row is gone once tasks.remove() runs + + // Always discard the .veloxpart/.veloxpart.meta pair (the schema's own words) — + // download.remove deletes the row outright, so there is no reason to leave a + // partial behind the way download.cancel does. A no-op if the task never had a + // live engine handle (already terminal, or never started). + if (actions_) (void)actions_->user_cancel(id, /*discard_partial=*/true); + + // The finished file only when deleteFile is true — never touched for a partial or + // failed transfer (there is nothing there but what discard_partial above already + // took care of). Best-effort: a missing file is not a reason to fail the remove. + bool deleted_file = false; + if (params.deleteFile && row.state == "complete") { + std::error_code ec; + deleted_file = + std::filesystem::remove(std::filesystem::path(row.save_dir) / row.filename, ec); + } + + (void)tasks.remove(id); // segments cascade via the FK (ON DELETE CASCADE) + r.removed.push_back(id); + + hub_.publish(proto::Event::TaskRemoved, + proto::make_notification( + proto::Event::TaskRemoved, + nlohmann::json{{"taskId", id}, {"deletedFile", deleted_file}}), + id); + } + if (on_mutation_) on_mutation_(); + return r; } proto::HandlerResult VeloxDispatcher::on_download_resume(const proto::DownloadResumeParams& params) { @@ -436,8 +564,15 @@ VeloxDispatcher::on_queue_stop(const proto::QueueStopParams& params) { return r; } proto::HandlerResult -VeloxDispatcher::on_queue_upsert(const proto::QueueUpsertParams&) { - return not_implemented("queue.upsert"); +VeloxDispatcher::on_queue_upsert(const proto::QueueUpsertParams& params) { + store::Queues queues(db_); + auto stored = queues.upsert(params.queue); + if (!stored) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "queue.upsert: " + stored.error().message}); + proto::QueueUpsertResult r; + r.queue = std::move(*stored); + return r; } proto::HandlerResult VeloxDispatcher::on_rules_list(const proto::RulesListParams&) { diff --git a/daemon/src/rpc/dispatcher.hpp b/daemon/src/rpc/dispatcher.hpp index 2c4ccc5..89c1aad 100644 --- a/daemon/src/rpc/dispatcher.hpp +++ b/daemon/src/rpc/dispatcher.hpp @@ -116,6 +116,12 @@ public: on_settings_set(const velox::proto::SettingsSetParams&) override; private: + // The whole of download.add's body; on_download_add and on_download_addBatch (each + // item merged against DownloadAddBatchParams.defaults first) both call this — exactly + // one place turns a DownloadSpec into a stored, admitted task. + velox::proto::HandlerResult add_one( + const velox::proto::DownloadSpec& spec); + velox::daemon::store::Db& db_; EventHub& hub_; TaskActionPort* actions_; diff --git a/daemon/src/rpc/task_action_port.hpp b/daemon/src/rpc/task_action_port.hpp index ce39e67..a036e96 100644 --- a/daemon/src/rpc/task_action_port.hpp +++ b/daemon/src/rpc/task_action_port.hpp @@ -43,6 +43,14 @@ public: // Returns the wire ids actually paused. virtual std::vector pause_queue(const std::string& queue_id) = 0; + // download.provideAuth: answers a task auto-paused on a 401/407. false if the task + // isn't currently holding a live engine handle (nothing waiting on credentials). + // `remember` is accepted but not yet acted on — persisting to the Secret Service isn't + // wired anywhere in this build yet (CLAUDE.md §4: never SQLite, never logs); this + // always does the "this retry only" half. Noted in deferrals.md. + virtual bool provide_auth(const std::string& wire_id, const std::string& username, + const std::string& password, bool remember) = 0; + // download.probe (D2), the File Info dialog's own network round trip — no task row // involved. Genuinely async (the engine's probe pool; up to the schema's 30s // x-deadlineMs) and so cannot fit VeloxDispatcher's synchronous on_download_probe: diff --git a/daemon/src/sched/scheduler.cpp b/daemon/src/sched/scheduler.cpp index ac296b5..61e128c 100644 --- a/daemon/src/sched/scheduler.cpp +++ b/daemon/src/sched/scheduler.cpp @@ -571,6 +571,15 @@ std::vector Scheduler::pause_queue(const std::string& queue_id) { return paused; } +bool Scheduler::provide_auth(const std::string& wire_id, const std::string& username, + const std::string& password, bool remember) { + (void)remember; // not yet wired to the Secret Service anywhere in this build + auto eid = engine_id_of(wire_id); + if (!eid) return false; + engine_.provide_auth(*eid, username, password, remember); + return true; +} + namespace { // Extension match against the categories table (categories.extensions, per category.list), diff --git a/daemon/src/sched/scheduler.hpp b/daemon/src/sched/scheduler.hpp index 29aba9a..c020ac2 100644 --- a/daemon/src/sched/scheduler.hpp +++ b/daemon/src/sched/scheduler.hpp @@ -132,6 +132,9 @@ public: // instead of User. Returns the wire ids actually paused. std::vector pause_queue(const std::string& queue_id) override; + bool provide_auth(const std::string& wire_id, const std::string& username, + const std::string& password, bool remember) override; + // rpc::TaskActionPort. Builds a vdm::net::ProbeRequest from `params`, runs it on the // engine's probe pool (outside the segment budget, ADR 0011 §5), and converts the // result back to proto terms — including the suggestedCategoryId/-SaveDir guess (a diff --git a/daemon/src/store/categories.cpp b/daemon/src/store/categories.cpp index 8532e8d..d585b4b 100644 --- a/daemon/src/store/categories.cpp +++ b/daemon/src/store/categories.cpp @@ -1,11 +1,35 @@ #include "store/categories.hpp" +#include + +#include +#include +#include + #include namespace velox::daemon::store { namespace proto = velox::proto; +namespace { + +proto::Category project_row(Stmt& st) { + proto::Category c; + c.categoryId = st.column_text(0); + c.name = st.column_text(1); + c.saveDir = st.column_text(2); + auto j = nlohmann::json::parse(st.column_text(3), nullptr, false); + if (j.is_array()) { + for (const auto& e : j) + if (e.is_string()) c.extensions.push_back(e.get()); + } + c.builtin = st.column_int(4) != 0; + return c; +} + +} // namespace + DbResult> Categories::list() { auto st = db_.prepare( "SELECT category_id, name, save_dir, extensions, builtin FROM categories " @@ -17,20 +41,111 @@ DbResult> Categories::list() { auto row = st->step(); if (!row) return std::unexpected(row.error()); if (!*row) break; - - proto::Category c; - c.categoryId = st->column_text(0); - c.name = st->column_text(1); - c.saveDir = st->column_text(2); - auto j = nlohmann::json::parse(st->column_text(3), nullptr, false); - if (j.is_array()) { - for (const auto& e : j) - if (e.is_string()) c.extensions.push_back(e.get()); - } - c.builtin = st->column_int(4) != 0; - out.push_back(std::move(c)); + out.push_back(project_row(*st)); } return out; } +DbResult> Categories::get(std::string_view category_id) { + auto st = db_.prepare( + "SELECT category_id, name, save_dir, extensions, builtin FROM categories " + "WHERE category_id = ?1"); + if (!st) return std::unexpected(st.error()); + if (auto b = st->bind(1, category_id); !b) return std::unexpected(b.error()); + auto row = st->step(); + if (!row) return std::unexpected(row.error()); + if (!*row) return std::optional{}; + return std::optional{project_row(*st)}; +} + +DbResult Categories::upsert(proto::Category category) { + // A replace keeps the existing row's builtin flag; a create is never builtin. Either + // way the payload's own `builtin` is ignored — a client cannot mint or revoke it. + bool builtin = false; + if (!category.categoryId.empty()) { + auto existing = get(category.categoryId); + if (!existing) return std::unexpected(existing.error()); + if (existing->has_value()) builtin = (*existing)->builtin; + } else { + // Reuse Tasks' id scheme (v4 UUID) would need a cross-module include for one + // function; a category id has no wire format requirement beyond "a string", so a + // timestamp-free random hex id keeps this module self-contained. + std::random_device rd; + std::uniform_int_distribution d; + char buf[17]; + std::snprintf(buf, sizeof(buf), "%016llx", static_cast(d(rd))); + category.categoryId = std::string(buf); + } + + nlohmann::json ext = nlohmann::json::array(); + for (const auto& e : category.extensions) ext.push_back(e); + + auto st = db_.prepare( + "INSERT INTO categories(category_id, name, save_dir, extensions, builtin) " + "VALUES(?1,?2,?3,?4,?5) " + "ON CONFLICT(category_id) DO UPDATE SET " + "name=excluded.name, save_dir=excluded.save_dir, extensions=excluded.extensions"); + if (!st) return std::unexpected(st.error()); + if (auto b = st->bind(1, std::string_view(category.categoryId)); !b) + return std::unexpected(b.error()); + if (auto b = st->bind(2, std::string_view(category.name)); !b) return std::unexpected(b.error()); + if (auto b = st->bind(3, std::string_view(category.saveDir)); !b) + return std::unexpected(b.error()); + if (auto b = st->bind(4, std::string_view(ext.dump())); !b) return std::unexpected(b.error()); + if (auto b = st->bind(5, static_cast(builtin)); !b) + return std::unexpected(b.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + + category.builtin = builtin; + return category; +} + +DbResult Categories::remove( + std::string_view category_id, const std::optional& reassign_to) { + RemoveResult out; + const std::string target = reassign_to.value_or("general"); + + // Db::transaction only threads a DbResult lambda; `out` is filled in-place and + // returned once the transaction (which may still fail and roll back) succeeds. + auto txn = db_.transaction([&]() -> DbResult { + { + // A builtin category is never removed, and — since it was never going to be + // removed — its tasks must not be reassigned away from it either. + auto chk = db_.prepare("SELECT builtin FROM categories WHERE category_id = ?1"); + if (!chk) return std::unexpected(chk.error()); + if (auto b = chk->bind(1, category_id); !b) return std::unexpected(b.error()); + auto row = chk->step(); + if (!row) return std::unexpected(row.error()); + if (!*row) return {}; // no such category: removed stays false + if (chk->column_int(0) != 0) return {}; // builtin: removed stays false + } + { + auto sel = db_.prepare("SELECT task_id FROM tasks WHERE category_id = ?1"); + if (!sel) return std::unexpected(sel.error()); + if (auto b = sel->bind(1, category_id); !b) return std::unexpected(b.error()); + for (;;) { + auto row = sel->step(); + if (!row) return std::unexpected(row.error()); + if (!*row) break; + out.reassigned_task_ids.push_back(sel->column_text(0)); + } + } + if (!out.reassigned_task_ids.empty()) { + auto upd = db_.prepare("UPDATE tasks SET category_id = ?2 WHERE category_id = ?1"); + if (!upd) return std::unexpected(upd.error()); + if (auto b = upd->bind(1, category_id); !b) return std::unexpected(b.error()); + if (auto b = upd->bind(2, std::string_view(target)); !b) return std::unexpected(b.error()); + if (auto r = upd->step(); !r) return std::unexpected(r.error()); + } + auto del = db_.prepare("DELETE FROM categories WHERE category_id = ?1 AND builtin = 0"); + if (!del) return std::unexpected(del.error()); + if (auto b = del->bind(1, category_id); !b) return std::unexpected(b.error()); + if (auto r = del->step(); !r) return std::unexpected(r.error()); + out.removed = sqlite3_changes(db_.raw()) > 0; + return {}; + }); + if (!txn) return std::unexpected(txn.error()); + return out; +} + } // namespace velox::daemon::store diff --git a/daemon/src/store/categories.hpp b/daemon/src/store/categories.hpp index b4439e8..1c3b57c 100644 --- a/daemon/src/store/categories.hpp +++ b/daemon/src/store/categories.hpp @@ -3,7 +3,13 @@ // Read access to the `categories` table, projected onto proto::Category. Owned here // rather than duplicated per handler since category.list and download.add (rule // matching, later) both need it. +// +// The table has no columns for Category.mimeTypes / .sortOrder (0001_initial.sql predates +// those fields); upsert() accepts them but they are not persisted — round-tripped as unset +// on the next list()/get(). Noted in daemon/docs/deferrals.md. +#include +#include #include #include "store/sqlite.hpp" @@ -16,6 +22,24 @@ public: explicit Categories(Db& db) : db_(db) {} DbResult> list(); + DbResult> get(std::string_view category_id); + + // "Omit categoryId to create; supply it to replace" (category.upsert's own words) — + // the caller (dispatcher) decides create vs replace by whether `category.categoryId` + // is empty and generates the id; this just writes the row. `builtin` is never taken + // from the payload: preserved from the existing row on a replace, always false on a + // create (a client can never mint a builtin category). + DbResult upsert(velox::proto::Category category); + + // False for "no such category". A builtin category is never removed — the caller + // checks that (category.remove -> -32602) before calling this, since that check needs + // ErrorCode, which this module (like the rest of store/) does not depend on. + struct RemoveResult { + bool removed = false; + std::vector reassigned_task_ids; + }; + DbResult remove(std::string_view category_id, + const std::optional& reassign_to); private: Db& db_; diff --git a/daemon/src/store/queues.cpp b/daemon/src/store/queues.cpp index 9fb469f..8db8771 100644 --- a/daemon/src/store/queues.cpp +++ b/daemon/src/store/queues.cpp @@ -2,6 +2,10 @@ #include +#include +#include +#include + #include namespace velox::daemon::store { @@ -10,8 +14,8 @@ namespace proto = velox::proto; namespace { -// One queue row (columns queue_id, name, state, max_concurrent, schedule, in that order) -// plus its member taskIds, read off the row a caller has already step()'d to. +// One queue row (columns queue_id, name, state, max_concurrent, schedule, on_complete, in +// that order) plus its member taskIds, read off the row a caller has already step()'d to. DbResult project_row(Db& db, Stmt& st) { proto::Queue q; q.queueId = st.column_text(0); @@ -22,6 +26,7 @@ DbResult project_row(Db& db, Stmt& st) { auto j = nlohmann::json::parse(st.column_text(4), nullptr, false); if (auto sched = proto::parse(j, "schedule")) q.schedule = *sched; } + if (auto oc = proto::parse_QueueOnComplete(st.column_text(5))) q.onComplete = *oc; auto ts = db.prepare("SELECT task_id FROM tasks WHERE queue_id = ?1 ORDER BY queue_position"); if (!ts) return std::unexpected(ts.error()); @@ -41,7 +46,8 @@ DbResult project_row(Db& db, Stmt& st) { DbResult> Queues::list() { auto st = db_.prepare( - "SELECT queue_id, name, state, max_concurrent, schedule FROM queues ORDER BY name"); + "SELECT queue_id, name, state, max_concurrent, schedule, on_complete FROM queues " + "ORDER BY name"); if (!st) return std::unexpected(st.error()); std::vector out; @@ -58,7 +64,8 @@ DbResult> Queues::list() { DbResult> Queues::get(std::string_view queue_id) { auto st = db_.prepare( - "SELECT queue_id, name, state, max_concurrent, schedule FROM queues WHERE queue_id = ?1"); + "SELECT queue_id, name, state, max_concurrent, schedule, on_complete FROM queues " + "WHERE queue_id = ?1"); if (!st) return std::unexpected(st.error()); if (auto b = st->bind(1, queue_id); !b) return std::unexpected(b.error()); auto row = st->step(); @@ -78,4 +85,48 @@ DbResult Queues::set_state(std::string_view queue_id, std::string_view sta return sqlite3_changes(db_.raw()) > 0; } +DbResult Queues::upsert(proto::Queue queue) { + if (queue.queueId.empty()) { + std::random_device rd; + std::uniform_int_distribution d; + char buf[17]; + std::snprintf(buf, sizeof(buf), "%016llx", static_cast(d(rd))); + queue.queueId = std::string(buf); + } + + // A create defaults to 'stopped' (never auto-runs a brand-new queue); a replace keeps + // whatever run state the queue is already in — queue.upsert edits the config, not the + // run state (that's queue.start/stop). + std::string state = "stopped"; + if (auto existing = get(queue.queueId); existing && existing->has_value()) + state = std::string(proto::to_string((*existing)->state)); + + const std::string schedule_json = + queue.schedule ? nlohmann::json(*queue.schedule).dump() : std::string(); + const std::string on_complete = + std::string(proto::to_string(queue.onComplete.value_or(proto::QueueOnComplete::Nothing))); + + auto st = db_.prepare( + "INSERT INTO queues(queue_id, name, state, max_concurrent, schedule, on_complete) " + "VALUES(?1,?2,?3,?4,?5,?6) " + "ON CONFLICT(queue_id) DO UPDATE SET " + "name=excluded.name, max_concurrent=excluded.max_concurrent, " + "schedule=excluded.schedule, on_complete=excluded.on_complete"); + if (!st) return std::unexpected(st.error()); + if (auto b = st->bind(1, std::string_view(queue.queueId)); !b) return std::unexpected(b.error()); + if (auto b = st->bind(2, std::string_view(queue.name)); !b) return std::unexpected(b.error()); + if (auto b = st->bind(3, std::string_view(state)); !b) return std::unexpected(b.error()); + if (auto b = st->bind(4, queue.maxConcurrent); !b) return std::unexpected(b.error()); + if (auto r = queue.schedule ? st->bind(5, std::string_view(schedule_json)) : st->bind_null(5); !r) + return std::unexpected(r.error()); + if (auto b = st->bind(6, std::string_view(on_complete)); !b) return std::unexpected(b.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + + auto stored = get(queue.queueId); + if (!stored) return std::unexpected(stored.error()); + if (!stored->has_value()) + return std::unexpected(DbError{0, "queue.upsert: row vanished after insert"}); + return **stored; +} + } // namespace velox::daemon::store diff --git a/daemon/src/store/queues.hpp b/daemon/src/store/queues.hpp index dd1c8ef..a284aea 100644 --- a/daemon/src/store/queues.hpp +++ b/daemon/src/store/queues.hpp @@ -27,6 +27,12 @@ public: // id doesn't exist. DbResult set_state(std::string_view queue_id, std::string_view state); + // "Omit queueId to create" (queue.upsert's own words) — an empty id generates one. + // taskIds is ignored (membership changes only through download.update / queue.reorder, + // per the schema's own note); a create defaults to 'stopped', a replace keeps the + // queue's current run state (queue.upsert edits config, not run state). + DbResult upsert(velox::proto::Queue queue); + private: Db& db_; }; diff --git a/daemon/tests/store_categories_queues_test.cpp b/daemon/tests/store_categories_queues_test.cpp index bd4f696..dbd376e 100644 --- a/daemon/tests/store_categories_queues_test.cpp +++ b/daemon/tests/store_categories_queues_test.cpp @@ -1,6 +1,8 @@ // store/categories + store/queues: the two D3 handlers GUI's category panel and queue // view need against a real daemon. +#include +#include #include #include "check.hpp" @@ -73,6 +75,105 @@ void run() { CHECK_EQ(ids[2], std::string("t0")); } } + + // --- Categories::upsert: create generates an id, builtin is never settable ------ + { + Categories categories(*db); + + velox::proto::Category in; + in.name = "ISOs"; + in.saveDir = "/tmp/isos"; + in.extensions = {"iso"}; + in.builtin = true; // ignored on create: a client cannot mint a builtin category + auto created = categories.upsert(in); + CHECK(created.has_value()); + if (created) { + CHECK(!created->categoryId.empty()); + CHECK(!created->builtin); + + // A replace keeps builtin=false too, and can rename/re-point. + velox::proto::Category patch = *created; + patch.name = "ISO Images"; + patch.builtin = true; // still ignored + auto replaced = categories.upsert(patch); + CHECK(replaced.has_value()); + if (replaced) { + CHECK_EQ(replaced->categoryId, created->categoryId); + CHECK_EQ(replaced->name, std::string("ISO Images")); + CHECK(!replaced->builtin); + } + + // A builtin category is untouched by remove(), and its tasks are not + // reassigned away from it — the store enforces this even without the + // dispatcher's own -32602 pre-check. + auto builtin_attempt = categories.remove("general", std::nullopt); + CHECK(builtin_attempt.has_value()); + if (builtin_attempt) CHECK(!builtin_attempt->removed); + + // remove() reassigns member tasks (default target: "general") and deletes + // the row. + Tasks tasks(*db); + TaskRow r; + r.task_id = "cat-owner"; + r.url = "https://example.com/x"; + r.save_dir = "/tmp"; + r.filename = "x"; + r.created_at = "2026-09-11T00:00:00Z"; + r.category_id = created->categoryId; + CHECK(tasks.insert(r).has_value()); + + auto removed = categories.remove(created->categoryId, std::nullopt); + CHECK(removed.has_value()); + if (removed) { + CHECK(removed->removed); + CHECK_EQ(removed->reassigned_task_ids.size(), std::size_t{1}); + CHECK_EQ(removed->reassigned_task_ids[0], std::string("cat-owner")); + } + auto owner = tasks.get("cat-owner"); + CHECK(owner.has_value() && owner->has_value()); + if (owner && *owner) + CHECK_EQ((*owner)->category_id.value_or(""), std::string("general")); + + // Gone: a second remove() finds nothing. + auto gone = categories.remove(created->categoryId, std::nullopt); + CHECK(gone.has_value()); + if (gone) CHECK(!gone->removed); + + tasks.remove("cat-owner"); + } + } + + // --- Queues::upsert: create generates an id; replace keeps the run state -------- + { + Queues queues(*db); + + velox::proto::Queue in; + in.name = "Nightly"; + in.state = velox::proto::QueueState::Running; // ignored on create: always 'stopped' + in.maxConcurrent = 3; + auto created = queues.upsert(in); + CHECK(created.has_value()); + if (created) { + CHECK(!created->queueId.empty()); + CHECK(created->state == velox::proto::QueueState::Stopped); + + CHECK(queues.set_state(created->queueId, "running").has_value()); + + velox::proto::Queue patch = *created; + patch.name = "Nightly Batch"; + patch.maxConcurrent = 5; + patch.state = velox::proto::QueueState::Stopped; // ignored on replace too + auto replaced = queues.upsert(patch); + CHECK(replaced.has_value()); + if (replaced) { + CHECK_EQ(replaced->name, std::string("Nightly Batch")); + CHECK_EQ(replaced->maxConcurrent, std::int64_t{5}); + // Run state survived the config edit — still 'running' from set_state above, + // not reset by the payload's (ignored) 'stopped'. + CHECK(replaced->state == velox::proto::QueueState::Running); + } + } + } } TEST_MAIN()