6 #include <unordered_set>
10 DB::DB(const string &filename)
12 int ret = sqlite3_open(filename.c_str(), &db);
13 if (ret != SQLITE_OK) {
14 fprintf(stderr, "%s: %s\n", filename.c_str(), sqlite3_errmsg(db));
19 CREATE TABLE IF NOT EXISTS state (state BLOB);
21 nullptr, nullptr, nullptr); // Ignore errors.
23 sqlite3_exec(db, "CREATE UNIQUE INDEX only_one_state ON state (1);", nullptr, nullptr, nullptr); // Ignore errors.
26 CREATE TABLE IF NOT EXISTS settings (settings BLOB, uniqifier NOT NULL DEFAULT 1 UNIQUE);
28 nullptr, nullptr, nullptr); // Ignore errors.
30 sqlite3_exec(db, "CREATE UNIQUE INDEX only_one_settings ON settings (1);", nullptr, nullptr, nullptr); // Ignore errors.
35 nullptr, nullptr, nullptr); // Ignore errors.
40 nullptr, nullptr, nullptr); // Ignore errors.
43 CREATE TABLE IF NOT EXISTS filev2 (
44 file INTEGER NOT NULL PRIMARY KEY,
45 filename VARCHAR NOT NULL UNIQUE,
50 nullptr, nullptr, nullptr); // Ignore errors.
52 sqlite3_exec(db, "PRAGMA journal_mode=WAL", nullptr, nullptr, nullptr); // Ignore errors.
53 sqlite3_exec(db, "PRAGMA synchronous=NORMAL", nullptr, nullptr, nullptr); // Ignore errors.
56 StateProto DB::get_state()
61 int ret = sqlite3_prepare_v2(db, "SELECT state FROM state", -1, &stmt, 0);
62 if (ret != SQLITE_OK) {
63 fprintf(stderr, "SELECT prepare: %s\n", sqlite3_errmsg(db));
67 ret = sqlite3_step(stmt);
68 if (ret == SQLITE_ROW) {
69 bool ok = state.ParseFromArray(sqlite3_column_blob(stmt, 0), sqlite3_column_bytes(stmt, 0));
71 fprintf(stderr, "State in database is corrupted!\n");
74 } else if (ret != SQLITE_DONE) {
75 fprintf(stderr, "SELECT step: %s\n", sqlite3_errmsg(db));
79 ret = sqlite3_finalize(stmt);
80 if (ret != SQLITE_OK) {
81 fprintf(stderr, "SELECT finalize: %s\n", sqlite3_errmsg(db));
88 void DB::store_state(const StateProto &state)
91 state.SerializeToString(&serialized);
93 int ret = sqlite3_exec(db, "BEGIN", nullptr, nullptr, nullptr);
94 if (ret != SQLITE_OK) {
95 fprintf(stderr, "BEGIN: %s\n", sqlite3_errmsg(db));
100 ret = sqlite3_prepare_v2(db, "REPLACE INTO state VALUES (?)", -1, &stmt, 0);
101 if (ret != SQLITE_OK) {
102 fprintf(stderr, "REPLACE prepare: %s\n", sqlite3_errmsg(db));
106 sqlite3_bind_blob(stmt, 1, serialized.data(), serialized.size(), SQLITE_STATIC);
108 ret = sqlite3_step(stmt);
109 if (ret == SQLITE_ROW) {
110 fprintf(stderr, "REPLACE step: %s\n", sqlite3_errmsg(db));
114 ret = sqlite3_finalize(stmt);
115 if (ret != SQLITE_OK) {
116 fprintf(stderr, "REPLACE finalize: %s\n", sqlite3_errmsg(db));
120 ret = sqlite3_exec(db, "COMMIT", nullptr, nullptr, nullptr);
121 if (ret != SQLITE_OK) {
122 fprintf(stderr, "COMMIT: %s\n", sqlite3_errmsg(db));
127 SettingsProto DB::get_settings()
129 SettingsProto settings;
132 int ret = sqlite3_prepare_v2(db, "SELECT settings FROM settings", -1, &stmt, 0);
133 if (ret != SQLITE_OK) {
134 fprintf(stderr, "SELECT prepare: %s\n", sqlite3_errmsg(db));
138 ret = sqlite3_step(stmt);
139 if (ret == SQLITE_ROW) {
140 bool ok = settings.ParseFromArray(sqlite3_column_blob(stmt, 0), sqlite3_column_bytes(stmt, 0));
142 fprintf(stderr, "State in database is corrupted!\n");
145 } else if (ret != SQLITE_DONE) {
146 fprintf(stderr, "SELECT step: %s\n", sqlite3_errmsg(db));
150 ret = sqlite3_finalize(stmt);
151 if (ret != SQLITE_OK) {
152 fprintf(stderr, "SELECT finalize: %s\n", sqlite3_errmsg(db));
159 void DB::store_settings(const SettingsProto &settings)
162 settings.SerializeToString(&serialized);
164 int ret = sqlite3_exec(db, "BEGIN", nullptr, nullptr, nullptr);
165 if (ret != SQLITE_OK) {
166 fprintf(stderr, "BEGIN: %s\n", sqlite3_errmsg(db));
171 ret = sqlite3_prepare_v2(db, "REPLACE INTO settings VALUES (?)", -1, &stmt, 0);
172 if (ret != SQLITE_OK) {
173 fprintf(stderr, "REPLACE prepare: %s\n", sqlite3_errmsg(db));
177 sqlite3_bind_blob(stmt, 1, serialized.data(), serialized.size(), SQLITE_STATIC);
179 ret = sqlite3_step(stmt);
180 if (ret == SQLITE_ROW) {
181 fprintf(stderr, "REPLACE step: %s\n", sqlite3_errmsg(db));
185 ret = sqlite3_finalize(stmt);
186 if (ret != SQLITE_OK) {
187 fprintf(stderr, "REPLACE finalize: %s\n", sqlite3_errmsg(db));
191 ret = sqlite3_exec(db, "COMMIT", nullptr, nullptr, nullptr);
192 if (ret != SQLITE_OK) {
193 fprintf(stderr, "COMMIT: %s\n", sqlite3_errmsg(db));
198 vector<DB::FrameOnDiskAndStreamIdx> DB::load_frame_file(const string &filename, size_t size, unsigned filename_idx)
200 FileContentsProto file_contents;
203 int ret = sqlite3_prepare_v2(db, "SELECT frames FROM filev2 WHERE filename=? AND size=?", -1, &stmt, 0);
204 if (ret != SQLITE_OK) {
205 fprintf(stderr, "SELECT prepare: %s\n", sqlite3_errmsg(db));
209 sqlite3_bind_text(stmt, 1, filename.data(), filename.size(), SQLITE_STATIC);
210 sqlite3_bind_int64(stmt, 2, size);
212 ret = sqlite3_step(stmt);
213 if (ret == SQLITE_ROW) {
214 bool ok = file_contents.ParseFromArray(sqlite3_column_blob(stmt, 0), sqlite3_column_bytes(stmt, 0));
216 fprintf(stderr, "Frame list in database is corrupted!\n");
219 } else if (ret != SQLITE_DONE) {
220 fprintf(stderr, "SELECT step: %s\n", sqlite3_errmsg(db));
224 ret = sqlite3_finalize(stmt);
225 if (ret != SQLITE_OK) {
226 fprintf(stderr, "SELECT finalize: %s\n", sqlite3_errmsg(db));
230 vector<FrameOnDiskAndStreamIdx> frames;
231 for (const StreamContentsProto &stream : file_contents.stream()) {
232 FrameOnDiskAndStreamIdx frame;
233 frame.stream_idx = stream.stream_idx();
234 for (int i = 0; i < stream.pts_size(); ++i) {
235 frame.frame.filename_idx = filename_idx;
236 frame.frame.pts = stream.pts(i);
237 frame.frame.offset = stream.offset(i);
238 frame.frame.size = stream.file_size(i);
239 frames.push_back(frame);
246 void DB::store_frame_file(const string &filename, size_t size, const vector<FrameOnDiskAndStreamIdx> &frames)
248 int ret = sqlite3_exec(db, "BEGIN", nullptr, nullptr, nullptr);
249 if (ret != SQLITE_OK) {
250 fprintf(stderr, "BEGIN: %s\n", sqlite3_errmsg(db));
254 // Delete any existing instances with this filename.
257 // Create the protobuf blob for the new row.
258 FileContentsProto file_contents;
259 unordered_set<unsigned> seen_stream_idx; // Usually only one.
260 for (const FrameOnDiskAndStreamIdx &frame : frames) {
261 seen_stream_idx.insert(frame.stream_idx);
263 for (unsigned stream_idx : seen_stream_idx) {
264 StreamContentsProto *stream = file_contents.add_stream();
265 stream->set_stream_idx(stream_idx);
266 stream->mutable_pts()->Reserve(frames.size());
267 stream->mutable_offset()->Reserve(frames.size());
268 stream->mutable_file_size()->Reserve(frames.size());
269 for (const FrameOnDiskAndStreamIdx &frame : frames) {
270 if (frame.stream_idx != stream_idx) {
273 stream->add_pts(frame.frame.pts);
274 stream->add_offset(frame.frame.offset);
275 stream->add_file_size(frame.frame.size);
279 file_contents.SerializeToString(&serialized);
281 // Insert the new row.
282 ret = sqlite3_prepare_v2(db, "REPLACE INTO filev2 (filename, size, frames) VALUES (?, ?, ?)", -1, &stmt, 0);
283 if (ret != SQLITE_OK) {
284 fprintf(stderr, "INSERT prepare: %s\n", sqlite3_errmsg(db));
288 sqlite3_bind_text(stmt, 1, filename.data(), filename.size(), SQLITE_STATIC);
289 sqlite3_bind_int64(stmt, 2, size);
290 sqlite3_bind_blob(stmt, 3, serialized.data(), serialized.size(), SQLITE_STATIC);
292 ret = sqlite3_step(stmt);
293 if (ret == SQLITE_ROW) {
294 fprintf(stderr, "REPLACE step: %s\n", sqlite3_errmsg(db));
298 ret = sqlite3_finalize(stmt);
299 if (ret != SQLITE_OK) {
300 fprintf(stderr, "REPLACE finalize: %s\n", sqlite3_errmsg(db));
305 ret = sqlite3_exec(db, "COMMIT", nullptr, nullptr, nullptr);
306 if (ret != SQLITE_OK) {
307 fprintf(stderr, "COMMIT: %s\n", sqlite3_errmsg(db));
312 void DB::clean_unused_frame_files(const vector<string> &used_filenames)
314 int ret = sqlite3_exec(db, "BEGIN", nullptr, nullptr, nullptr);
315 if (ret != SQLITE_OK) {
316 fprintf(stderr, "BEGIN: %s\n", sqlite3_errmsg(db));
320 ret = sqlite3_exec(db, R"(
321 CREATE TEMPORARY TABLE used_filenames ( filename VARCHAR NOT NULL PRIMARY KEY )
323 nullptr, nullptr, nullptr);
325 if (ret != SQLITE_OK) {
326 fprintf(stderr, "CREATE TEMPORARY TABLE: %s\n", sqlite3_errmsg(db));
330 // Insert the new rows.
332 ret = sqlite3_prepare_v2(db, "INSERT INTO used_filenames (filename) VALUES (?)", -1, &stmt, 0);
333 if (ret != SQLITE_OK) {
334 fprintf(stderr, "INSERT prepare: %s\n", sqlite3_errmsg(db));
338 for (const string &filename : used_filenames) {
339 sqlite3_bind_text(stmt, 1, filename.data(), filename.size(), SQLITE_STATIC);
341 ret = sqlite3_step(stmt);
342 if (ret == SQLITE_ROW) {
343 fprintf(stderr, "INSERT step: %s\n", sqlite3_errmsg(db));
347 ret = sqlite3_reset(stmt);
348 if (ret == SQLITE_ROW) {
349 fprintf(stderr, "INSERT reset: %s\n", sqlite3_errmsg(db));
354 ret = sqlite3_finalize(stmt);
355 if (ret != SQLITE_OK) {
356 fprintf(stderr, "INSERT finalize: %s\n", sqlite3_errmsg(db));
360 ret = sqlite3_exec(db, R"(
361 DELETE FROM filev2 WHERE filename NOT IN ( SELECT filename FROM used_filenames )
363 nullptr, nullptr, nullptr);
365 if (ret != SQLITE_OK) {
366 fprintf(stderr, "DELETE: %s\n", sqlite3_errmsg(db));
370 ret = sqlite3_exec(db, R"(
371 DROP TABLE used_filenames
373 nullptr, nullptr, nullptr);
375 if (ret != SQLITE_OK) {
376 fprintf(stderr, "DROP TABLE: %s\n", sqlite3_errmsg(db));
381 ret = sqlite3_exec(db, "COMMIT", nullptr, nullptr, nullptr);
382 if (ret != SQLITE_OK) {
383 fprintf(stderr, "COMMIT: %s\n", sqlite3_errmsg(db));