daemon: store/ query layer — tasks + settings (prep for the download.add vertical slice)
The read/write surface the dispatcher handlers need, so download.add persists and download.list / download.get project real rows when the engine lands. - store/tasks — TaskRow (1:1 with the schema), insert / get / remove / set_state / count, and list(filter, sort, offset, limit) that does all the WHERE / ORDER BY / LIMIT in SQL (M1 DoD: a 1000-row list never materialised client-side). Filter covers states / category / queue / case-insensitive filename+url substring / date range; sort is a whitelisted column + direction with NULLs last, default newest-first; the enum spellings in a state IN (...) come from proto::to_string, never from user text. to_summary() projects a row onto proto::TaskSummary including the flattened error block when the task failed / retry_wait / auto-paused. - store/settings — key -> JSON-text with a built-in default table mirroring Settings.schema.json / ADR 0012; get_raw / set_raw / overrides plus typed get_int / get_string / get_string_array for the governor config and saveTo.allowedRoots. Full settings.get/set wire projection lands with those handlers. - veloxd_store now links velox::proto + nlohmann_json for the projection. Test veloxd.store_tasks: insert/get round trip, PK duplicate rejected, the error-block projection, list total+paging+sort+every filter, set_state pause_reason clear-on-unpause, remove, and settings default vs override. ASan+UBSan and TSan clean; 34 daemon/cli tests green. Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Upd9WhG9oppieig5nRDLig
This commit is contained in:
Claude Sonnet 5
@@ -0,0 +1,309 @@
|
||||
#include "store/tasks.hpp"
|
||||
|
||||
#include <sqlite3.h>
|
||||
|
||||
#include <string>
|
||||
|
||||
namespace velox::daemon::store {
|
||||
|
||||
namespace proto = velox::proto;
|
||||
|
||||
namespace {
|
||||
|
||||
// Column order shared by get() and list() — the row reader indexes into this.
|
||||
constexpr const char* kCols =
|
||||
"task_id, url, save_dir, filename, state, start_mode, created_at, "
|
||||
"effective_url, category_id, queue_id, description, pause_reason, "
|
||||
"etag, last_modified, content_type, last_try_at, completed_at, "
|
||||
"checksum_algo, checksum_value, "
|
||||
"size_bytes, downloaded_bytes, resumable, "
|
||||
"req_segments, eff_segments, req_buffer_bytes, eff_buffer_bytes, queue_position, "
|
||||
"error_code, error_message, error_http_status, error_retryable, error_attempt, "
|
||||
"error_next_retry_at";
|
||||
|
||||
DbResult<void> bind_opt(Stmt& s, int i, const std::optional<std::string>& v) {
|
||||
return v ? s.bind(i, std::string_view(*v)) : s.bind_null(i);
|
||||
}
|
||||
DbResult<void> bind_opt(Stmt& s, int i, const std::optional<std::int64_t>& v) {
|
||||
return v ? s.bind(i, *v) : s.bind_null(i);
|
||||
}
|
||||
|
||||
std::optional<std::string> col_opt_text(Stmt& s, int i) {
|
||||
if (s.column_is_null(i)) return std::nullopt;
|
||||
return s.column_text(i);
|
||||
}
|
||||
std::optional<std::int64_t> col_opt_int(Stmt& s, int i) {
|
||||
if (s.column_is_null(i)) return std::nullopt;
|
||||
return s.column_int(i);
|
||||
}
|
||||
|
||||
TaskRow read_row(Stmt& s) {
|
||||
TaskRow r;
|
||||
r.task_id = s.column_text(0);
|
||||
r.url = s.column_text(1);
|
||||
r.save_dir = s.column_text(2);
|
||||
r.filename = s.column_text(3);
|
||||
r.state = s.column_text(4);
|
||||
r.start_mode = s.column_text(5);
|
||||
r.created_at = s.column_text(6);
|
||||
r.effective_url = col_opt_text(s, 7);
|
||||
r.category_id = col_opt_text(s, 8);
|
||||
r.queue_id = col_opt_text(s, 9);
|
||||
r.description = col_opt_text(s, 10);
|
||||
r.pause_reason = col_opt_text(s, 11);
|
||||
r.etag = col_opt_text(s, 12);
|
||||
r.last_modified = col_opt_text(s, 13);
|
||||
r.content_type = col_opt_text(s, 14);
|
||||
r.last_try_at = col_opt_text(s, 15);
|
||||
r.completed_at = col_opt_text(s, 16);
|
||||
r.checksum_algo = col_opt_text(s, 17);
|
||||
r.checksum_value = col_opt_text(s, 18);
|
||||
r.size_bytes = col_opt_int(s, 19);
|
||||
r.downloaded_bytes = s.column_int(20);
|
||||
r.resumable = s.column_int(21) != 0;
|
||||
r.req_segments = col_opt_int(s, 22);
|
||||
r.eff_segments = s.column_int(23);
|
||||
r.req_buffer_bytes = col_opt_int(s, 24);
|
||||
r.eff_buffer_bytes = col_opt_int(s, 25);
|
||||
r.queue_position = col_opt_int(s, 26);
|
||||
r.error_code = col_opt_text(s, 27);
|
||||
r.error_message = col_opt_text(s, 28);
|
||||
r.error_http_status = col_opt_int(s, 29);
|
||||
if (!s.column_is_null(30)) r.error_retryable = s.column_int(30) != 0;
|
||||
r.error_attempt = col_opt_int(s, 31);
|
||||
r.error_next_retry_at = col_opt_text(s, 32);
|
||||
return r;
|
||||
}
|
||||
|
||||
// TaskSort.field -> a whitelisted column. Anything not backed by a stored column (live
|
||||
// speed, eta) sorts by recency instead of erroring.
|
||||
const char* sort_column(proto::TaskSortField f) {
|
||||
switch (f) {
|
||||
case proto::TaskSortField::Filename: return "filename";
|
||||
case proto::TaskSortField::SizeBytes: return "size_bytes";
|
||||
case proto::TaskSortField::State: return "state";
|
||||
case proto::TaskSortField::LastTryAt: return "last_try_at";
|
||||
case proto::TaskSortField::QueuePosition: return "queue_position";
|
||||
case proto::TaskSortField::Description: return "description";
|
||||
case proto::TaskSortField::CreatedAt:
|
||||
case proto::TaskSortField::EtaSeconds:
|
||||
case proto::TaskSortField::SpeedBps:
|
||||
default: return "created_at";
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
DbResult<void> Tasks::insert(const TaskRow& r) {
|
||||
auto st = db_.prepare(
|
||||
"INSERT INTO tasks("
|
||||
"task_id, url, save_dir, filename, state, start_mode, created_at, "
|
||||
"effective_url, category_id, queue_id, description, pause_reason, "
|
||||
"etag, last_modified, content_type, last_try_at, completed_at, "
|
||||
"checksum_algo, checksum_value, size_bytes, downloaded_bytes, resumable, "
|
||||
"req_segments, eff_segments, req_buffer_bytes, eff_buffer_bytes, queue_position, "
|
||||
"error_code, error_message, error_http_status, error_retryable, error_attempt, "
|
||||
"error_next_retry_at) VALUES("
|
||||
"?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20,?21,?22,"
|
||||
"?23,?24,?25,?26,?27,?28,?29,?30,?31,?32,?33)");
|
||||
if (!st) return std::unexpected(st.error());
|
||||
|
||||
auto chk = [](DbResult<void> r) { return r.has_value(); };
|
||||
bool ok = chk(st->bind(1, std::string_view(r.task_id))) &&
|
||||
chk(st->bind(2, std::string_view(r.url))) &&
|
||||
chk(st->bind(3, std::string_view(r.save_dir))) &&
|
||||
chk(st->bind(4, std::string_view(r.filename))) &&
|
||||
chk(st->bind(5, std::string_view(r.state))) &&
|
||||
chk(st->bind(6, std::string_view(r.start_mode))) &&
|
||||
chk(st->bind(7, std::string_view(r.created_at))) &&
|
||||
chk(bind_opt(*st, 8, r.effective_url)) && chk(bind_opt(*st, 9, r.category_id)) &&
|
||||
chk(bind_opt(*st, 10, r.queue_id)) && chk(bind_opt(*st, 11, r.description)) &&
|
||||
chk(bind_opt(*st, 12, r.pause_reason)) && chk(bind_opt(*st, 13, r.etag)) &&
|
||||
chk(bind_opt(*st, 14, r.last_modified)) && chk(bind_opt(*st, 15, r.content_type)) &&
|
||||
chk(bind_opt(*st, 16, r.last_try_at)) && chk(bind_opt(*st, 17, r.completed_at)) &&
|
||||
chk(bind_opt(*st, 18, r.checksum_algo)) &&
|
||||
chk(bind_opt(*st, 19, r.checksum_value)) && chk(bind_opt(*st, 20, r.size_bytes)) &&
|
||||
chk(st->bind(21, r.downloaded_bytes)) &&
|
||||
chk(st->bind(22, static_cast<std::int64_t>(r.resumable))) &&
|
||||
chk(bind_opt(*st, 23, r.req_segments)) && chk(st->bind(24, r.eff_segments)) &&
|
||||
chk(bind_opt(*st, 25, r.req_buffer_bytes)) &&
|
||||
chk(bind_opt(*st, 26, r.eff_buffer_bytes)) &&
|
||||
chk(bind_opt(*st, 27, r.queue_position)) && chk(bind_opt(*st, 28, r.error_code)) &&
|
||||
chk(bind_opt(*st, 29, r.error_message)) &&
|
||||
chk(bind_opt(*st, 30, r.error_http_status)) &&
|
||||
chk(r.error_retryable ? st->bind(31, static_cast<std::int64_t>(*r.error_retryable))
|
||||
: st->bind_null(31)) &&
|
||||
chk(bind_opt(*st, 32, r.error_attempt)) &&
|
||||
chk(bind_opt(*st, 33, r.error_next_retry_at));
|
||||
if (!ok) return std::unexpected(DbError{0, "failed to bind a task column"});
|
||||
|
||||
if (auto r2 = st->step(); !r2) return std::unexpected(r2.error());
|
||||
return {};
|
||||
}
|
||||
|
||||
DbResult<std::optional<TaskRow>> Tasks::get(std::string_view task_id) {
|
||||
auto st = db_.prepare(std::string("SELECT ") + kCols + " FROM tasks WHERE task_id = ?1");
|
||||
if (!st) return std::unexpected(st.error());
|
||||
if (auto r = st->bind(1, task_id); !r) return std::unexpected(r.error());
|
||||
auto row = st->step();
|
||||
if (!row) return std::unexpected(row.error());
|
||||
if (!*row) return std::optional<TaskRow>{};
|
||||
return std::optional<TaskRow>{read_row(*st)};
|
||||
}
|
||||
|
||||
DbResult<Tasks::Page> Tasks::list(const std::optional<proto::TaskFilter>& filter,
|
||||
const std::optional<proto::TaskSort>& sort, std::int64_t offset,
|
||||
std::int64_t limit) {
|
||||
std::string where;
|
||||
std::vector<std::string> params; // bound 1:1 with the '?' placeholders, in order
|
||||
auto clause = [&](std::string c) {
|
||||
where += where.empty() ? " WHERE " : " AND ";
|
||||
where += std::move(c);
|
||||
};
|
||||
|
||||
if (filter) {
|
||||
if (filter->states && !filter->states->empty()) {
|
||||
std::string in;
|
||||
for (const auto st : *filter->states) {
|
||||
in += in.empty() ? "" : ",";
|
||||
in += "'";
|
||||
in += proto::to_string(st); // an enum spelling, never user input
|
||||
in += "'";
|
||||
}
|
||||
clause("state IN (" + in + ")");
|
||||
}
|
||||
if (filter->categoryId) {
|
||||
clause("category_id = ?");
|
||||
params.push_back(*filter->categoryId);
|
||||
}
|
||||
if (filter->queueId) {
|
||||
clause("queue_id = ?");
|
||||
params.push_back(*filter->queueId);
|
||||
}
|
||||
if (filter->query) {
|
||||
clause("(instr(lower(filename), lower(?)) > 0 OR instr(lower(url), lower(?)) > 0)");
|
||||
params.push_back(*filter->query);
|
||||
params.push_back(*filter->query);
|
||||
}
|
||||
if (filter->addedAfter) {
|
||||
clause("created_at >= ?");
|
||||
params.push_back(*filter->addedAfter);
|
||||
}
|
||||
if (filter->addedBefore) {
|
||||
clause("created_at < ?");
|
||||
params.push_back(*filter->addedBefore);
|
||||
}
|
||||
}
|
||||
|
||||
std::int64_t total = 0;
|
||||
{
|
||||
auto st = db_.prepare("SELECT count(*) FROM tasks" + where);
|
||||
if (!st) return std::unexpected(st.error());
|
||||
for (std::size_t i = 0; i < params.size(); ++i)
|
||||
if (auto r = st->bind(static_cast<int>(i + 1), std::string_view(params[i])); !r)
|
||||
return std::unexpected(r.error());
|
||||
auto row = st->step();
|
||||
if (!row) return std::unexpected(row.error());
|
||||
if (*row) total = st->column_int(0);
|
||||
}
|
||||
|
||||
// Default sort is newest-first; an explicit sort is a whitelisted column + direction.
|
||||
// NULLs sort last in both directions so an unsized / unqueued task never floats up.
|
||||
std::string order = "created_at DESC";
|
||||
if (sort) {
|
||||
const char* col = sort_column(sort->field);
|
||||
const bool desc = sort->direction == proto::TaskSortDirection::Desc;
|
||||
order = std::string(col) + " IS NULL, " + col + (desc ? " DESC" : " ASC");
|
||||
}
|
||||
|
||||
const std::int64_t lim = limit > 0 ? limit : 500;
|
||||
const std::int64_t off = offset > 0 ? offset : 0;
|
||||
|
||||
Page page;
|
||||
page.total = total;
|
||||
{
|
||||
auto st = db_.prepare(std::string("SELECT ") + kCols + " FROM tasks" + where +
|
||||
" ORDER BY " + order + " LIMIT ? OFFSET ?");
|
||||
if (!st) return std::unexpected(st.error());
|
||||
int n = 1;
|
||||
for (const auto& p : params)
|
||||
if (auto r = st->bind(n++, std::string_view(p)); !r) return std::unexpected(r.error());
|
||||
if (auto r = st->bind(n++, lim); !r) return std::unexpected(r.error());
|
||||
if (auto r = st->bind(n++, off); !r) return std::unexpected(r.error());
|
||||
for (;;) {
|
||||
auto row = st->step();
|
||||
if (!row) return std::unexpected(row.error());
|
||||
if (!*row) break;
|
||||
page.rows.push_back(read_row(*st));
|
||||
}
|
||||
}
|
||||
return page;
|
||||
}
|
||||
|
||||
DbResult<bool> Tasks::set_state(std::string_view task_id, std::string_view state,
|
||||
const std::optional<std::string>& pause_reason) {
|
||||
auto st = db_.prepare(
|
||||
"UPDATE tasks SET state = ?2, pause_reason = ?3 WHERE task_id = ?1");
|
||||
if (!st) return std::unexpected(st.error());
|
||||
if (auto r = st->bind(1, task_id); !r) return std::unexpected(r.error());
|
||||
if (auto r = st->bind(2, state); !r) return std::unexpected(r.error());
|
||||
const bool paused = state == "paused";
|
||||
if (auto r = paused && pause_reason ? st->bind(3, std::string_view(*pause_reason))
|
||||
: st->bind_null(3);
|
||||
!r)
|
||||
return std::unexpected(r.error());
|
||||
if (auto r = st->step(); !r) return std::unexpected(r.error());
|
||||
return sqlite3_changes(db_.raw()) > 0;
|
||||
}
|
||||
|
||||
DbResult<bool> Tasks::remove(std::string_view task_id) {
|
||||
auto st = db_.prepare("DELETE FROM tasks WHERE task_id = ?1");
|
||||
if (!st) return std::unexpected(st.error());
|
||||
if (auto r = st->bind(1, task_id); !r) return std::unexpected(r.error());
|
||||
if (auto r = st->step(); !r) return std::unexpected(r.error());
|
||||
return sqlite3_changes(db_.raw()) > 0;
|
||||
}
|
||||
|
||||
DbResult<std::int64_t> Tasks::count() {
|
||||
auto st = db_.prepare("SELECT count(*) FROM tasks");
|
||||
if (!st) return std::unexpected(st.error());
|
||||
auto row = st->step();
|
||||
if (!row) return std::unexpected(row.error());
|
||||
return (*row) ? st->column_int(0) : 0;
|
||||
}
|
||||
|
||||
proto::TaskSummary to_summary(const TaskRow& r) {
|
||||
proto::TaskSummary s;
|
||||
s.taskId = r.task_id;
|
||||
s.filename = r.filename;
|
||||
s.saveDir = r.save_dir;
|
||||
s.url = r.url;
|
||||
s.effectiveUrl = r.effective_url;
|
||||
s.sizeBytes = r.size_bytes;
|
||||
s.downloadedBytes = r.downloaded_bytes;
|
||||
if (auto st = proto::parse_TaskState(r.state)) s.state = *st;
|
||||
s.speedBps = 0;
|
||||
s.resumable = r.resumable;
|
||||
s.segments = r.eff_segments;
|
||||
s.categoryId = r.category_id;
|
||||
s.queueId = r.queue_id;
|
||||
s.queuePosition = r.queue_position;
|
||||
s.description = r.description;
|
||||
s.createdAt = r.created_at;
|
||||
s.lastTryAt = r.last_try_at;
|
||||
s.completedAt = r.completed_at;
|
||||
|
||||
if (r.error_code) {
|
||||
proto::TaskError e;
|
||||
if (auto c = proto::parse_TaskErrorCode(*r.error_code)) e.code = *c;
|
||||
e.message = r.error_message.value_or("");
|
||||
e.httpStatus = r.error_http_status;
|
||||
e.retryable = r.error_retryable.value_or(false);
|
||||
e.attempt = r.error_attempt;
|
||||
e.nextRetryAt = r.error_next_retry_at;
|
||||
s.error = std::move(e);
|
||||
}
|
||||
return s;
|
||||
}
|
||||
|
||||
} // namespace velox::daemon::store
|
||||
Reference in New Issue
Block a user