2 * This file is part of GNUnet
3 * Copyright (C) 2013 GNUnet e.V.
5 * GNUnet is free software: you can redistribute it and/or modify it
6 * under the terms of the GNU Affero General Public License as published
7 * by the Free Software Foundation, either version 3 of the License,
8 * or (at your option) any later version.
10 * GNUnet is distributed in the hope that it will be useful, but
11 * WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 * Affero General Public License for more details.
15 * You should have received a copy of the GNU Affero General Public License
16 * along with this program. If not, see <http://www.gnu.org/licenses/>.
20 * @file psycstore/plugin_psycstore_sqlite.c
21 * @brief sqlite-based psycstore backend
22 * @author Gabor X Toth
23 * @author Christian Grothoff
27 * FIXME: SQLite3 only supports signed 64-bit integers natively,
28 * thus it can only store 63 bits of the uint64_t's.
32 #include "gnunet_psycstore_plugin.h"
33 #include "gnunet_psycstore_service.h"
34 #include "gnunet_multicast_service.h"
35 #include "gnunet_crypto_lib.h"
36 #include "gnunet_psyc_util_lib.h"
37 #include "psycstore.h"
41 * After how many ms "busy" should a DB operation fail for good? A
42 * low value makes sure that we are more responsive to requests
43 * (especially PUTs). A high value guarantees a higher success rate
44 * (SELECTs in iterate can take several seconds despite LIMIT=1).
46 * The default value of 1s should ensure that users do not experience
47 * huge latencies while at the same time allowing operations to
48 * succeed with reasonable probability.
50 #define BUSY_TIMEOUT_MS 1000
52 #define DEBUG_PSYCSTORE GNUNET_EXTRA_LOGGING
55 * Log an error message at log-level 'level' that indicates
56 * a failure of the command 'cmd' on file 'filename'
57 * with the message given by strerror(errno).
59 #define LOG_SQLITE(db, level, cmd) do { GNUNET_log_from (level, "psycstore-sqlite", _("`%s' failed at %s:%d with error: %s (%d)\n"), cmd, __FILE__, __LINE__, sqlite3_errmsg(db->dbh), sqlite3_errcode(db->dbh)); } while(0)
61 #define LOG(kind,...) GNUNET_log_from (kind, "psycstore-sqlite", __VA_ARGS__)
65 TRANSACTION_STATE_MODIFY,
66 TRANSACTION_STATE_SYNC,
70 * Context for all functions in this plugin.
75 const struct GNUNET_CONFIGURATION_Handle *cfg;
83 * Native SQLite database handle.
88 * Current transaction.
90 enum Transactions transaction;
92 sqlite3_stmt *transaction_begin;
94 sqlite3_stmt *transaction_commit;
96 sqlite3_stmt *transaction_rollback;
99 * Precompiled SQL for channel_key_store()
101 sqlite3_stmt *insert_channel_key;
104 * Precompiled SQL for slave_key_store()
106 sqlite3_stmt *insert_slave_key;
110 * Precompiled SQL for membership_store()
112 sqlite3_stmt *insert_membership;
115 * Precompiled SQL for membership_test()
117 sqlite3_stmt *select_membership;
121 * Precompiled SQL for fragment_store()
123 sqlite3_stmt *insert_fragment;
126 * Precompiled SQL for message_add_flags()
128 sqlite3_stmt *update_message_flags;
131 * Precompiled SQL for fragment_get()
133 sqlite3_stmt *select_fragments;
136 * Precompiled SQL for fragment_get()
138 sqlite3_stmt *select_latest_fragments;
141 * Precompiled SQL for message_get()
143 sqlite3_stmt *select_messages;
146 * Precompiled SQL for message_get()
148 sqlite3_stmt *select_latest_messages;
151 * Precompiled SQL for message_get_fragment()
153 sqlite3_stmt *select_message_fragment;
156 * Precompiled SQL for counters_get_message()
158 sqlite3_stmt *select_counters_message;
161 * Precompiled SQL for counters_get_state()
163 sqlite3_stmt *select_counters_state;
166 * Precompiled SQL for state_modify_end()
168 sqlite3_stmt *update_state_hash_message_id;
171 * Precompiled SQL for state_sync_end()
173 sqlite3_stmt *update_max_state_message_id;
176 * Precompiled SQL for state_modify_op()
178 sqlite3_stmt *insert_state_current;
181 * Precompiled SQL for state_modify_end()
183 sqlite3_stmt *delete_state_empty;
186 * Precompiled SQL for state_set_signed()
188 sqlite3_stmt *update_state_signed;
191 * Precompiled SQL for state_sync()
193 sqlite3_stmt *insert_state_sync;
196 * Precompiled SQL for state_sync()
198 sqlite3_stmt *delete_state;
201 * Precompiled SQL for state_sync()
203 sqlite3_stmt *insert_state_from_sync;
206 * Precompiled SQL for state_sync()
208 sqlite3_stmt *delete_state_sync;
211 * Precompiled SQL for state_get_signed()
213 sqlite3_stmt *select_state_signed;
216 * Precompiled SQL for state_get()
218 sqlite3_stmt *select_state_one;
221 * Precompiled SQL for state_get_prefix()
223 sqlite3_stmt *select_state_prefix;
230 sql_trace (void *cls, const char *sql)
232 LOG (GNUNET_ERROR_TYPE_DEBUG, "SQL query:\n%s\n", sql);
238 * @brief Prepare a SQL statement
240 * @param dbh handle to the database
241 * @param sql SQL statement, UTF-8 encoded
242 * @param stmt set to the prepared statement
243 * @return 0 on success
246 sql_prepare (sqlite3 *dbh, const char *sql, sqlite3_stmt **stmt)
251 result = sqlite3_prepare_v2 (dbh, sql, strlen (sql), stmt,
252 (const char **) &tail);
253 LOG (GNUNET_ERROR_TYPE_DEBUG,
254 "Prepared `%s' / %p: %d\n", sql, *stmt, result);
255 if (result != SQLITE_OK)
256 LOG (GNUNET_ERROR_TYPE_ERROR,
257 _("Error preparing SQL query: %s\n %s\n"),
258 sqlite3_errmsg (dbh), sql);
264 * @brief Prepare a SQL statement
266 * @param dbh handle to the database
267 * @param sql SQL statement, UTF-8 encoded
268 * @return 0 on success
271 sql_exec (sqlite3 *dbh, const char *sql)
275 result = sqlite3_exec (dbh, sql, NULL, NULL, NULL);
276 LOG (GNUNET_ERROR_TYPE_DEBUG,
277 "Executed `%s' / %d\n", sql, result);
278 if (result != SQLITE_OK)
279 LOG (GNUNET_ERROR_TYPE_ERROR,
280 _("Error executing SQL query: %s\n %s\n"),
281 sqlite3_errmsg (dbh), sql);
287 * Initialize the database connections and associated
288 * data structures (create tables and indices
289 * as needed as well).
291 * @param plugin the plugin context (state for this module)
292 * @return GNUNET_OK on success
295 database_setup (struct Plugin *plugin)
300 GNUNET_CONFIGURATION_get_value_filename (plugin->cfg, "psycstore-sqlite",
301 "FILENAME", &filename))
303 GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
304 "psycstore-sqlite", "FILENAME");
305 return GNUNET_SYSERR;
307 if (GNUNET_OK != GNUNET_DISK_file_test (filename))
309 if (GNUNET_OK != GNUNET_DISK_directory_create_for_file (filename))
312 GNUNET_free (filename);
313 return GNUNET_SYSERR;
316 /* filename should be UTF-8-encoded. If it isn't, it's a bug */
317 plugin->fn = filename;
319 /* Open database and precompile statements */
320 if (SQLITE_OK != sqlite3_open (plugin->fn, &plugin->dbh))
322 LOG (GNUNET_ERROR_TYPE_ERROR,
323 _("Unable to initialize SQLite: %s.\n"),
324 sqlite3_errmsg (plugin->dbh));
325 return GNUNET_SYSERR;
329 sqlite3_trace (plugin->dbh, &sql_trace, NULL);
332 sql_exec (plugin->dbh, "PRAGMA temp_store=MEMORY");
333 sql_exec (plugin->dbh, "PRAGMA synchronous=NORMAL");
334 sql_exec (plugin->dbh, "PRAGMA legacy_file_format=OFF");
335 sql_exec (plugin->dbh, "PRAGMA auto_vacuum=INCREMENTAL");
336 sql_exec (plugin->dbh, "PRAGMA encoding=\"UTF-8\"");
337 #if ! DEBUG_PSYCSTORE
338 sql_exec (plugin->dbh, "PRAGMA locking_mode=EXCLUSIVE");
340 sql_exec (plugin->dbh, "PRAGMA page_size=4096");
342 sqlite3_busy_timeout (plugin->dbh, BUSY_TIMEOUT_MS);
346 sql_exec (plugin->dbh,
347 "CREATE TABLE IF NOT EXISTS channels (\n"
348 " id INTEGER PRIMARY KEY,\n"
349 " pub_key BLOB(32) UNIQUE,\n"
350 " max_state_message_id INTEGER,\n" // last applied state message ID
351 " state_hash_message_id INTEGER\n" // last message ID with a state hash
354 sql_exec (plugin->dbh,
355 "CREATE TABLE IF NOT EXISTS slaves (\n"
356 " id INTEGER PRIMARY KEY,\n"
357 " pub_key BLOB(32) UNIQUE\n"
360 sql_exec (plugin->dbh,
361 "CREATE TABLE IF NOT EXISTS membership (\n"
362 " channel_id INTEGER NOT NULL REFERENCES channels(id),\n"
363 " slave_id INTEGER NOT NULL REFERENCES slaves(id),\n"
364 " did_join INTEGER NOT NULL,\n"
365 " announced_at INTEGER NOT NULL,\n"
366 " effective_since INTEGER NOT NULL,\n"
367 " group_generation INTEGER NOT NULL\n"
369 sql_exec (plugin->dbh,
370 "CREATE INDEX IF NOT EXISTS idx_membership_channel_id_slave_id "
371 "ON membership (channel_id, slave_id);");
373 /** @todo messages table: add method_name column */
374 sql_exec (plugin->dbh,
375 "CREATE TABLE IF NOT EXISTS messages (\n"
376 " channel_id INTEGER NOT NULL REFERENCES channels(id),\n"
377 " hop_counter INTEGER NOT NULL,\n"
380 " fragment_id INTEGER NOT NULL,\n"
381 " fragment_offset INTEGER NOT NULL,\n"
382 " message_id INTEGER NOT NULL,\n"
383 " group_generation INTEGER NOT NULL,\n"
384 " multicast_flags INTEGER NOT NULL,\n"
385 " psycstore_flags INTEGER NOT NULL,\n"
387 " PRIMARY KEY (channel_id, fragment_id),\n"
388 " UNIQUE (channel_id, message_id, fragment_offset)\n"
391 sql_exec (plugin->dbh,
392 "CREATE TABLE IF NOT EXISTS state (\n"
393 " channel_id INTEGER NOT NULL REFERENCES channels(id),\n"
394 " name TEXT NOT NULL,\n"
395 " value_current BLOB,\n"
396 " value_signed BLOB,\n"
397 " PRIMARY KEY (channel_id, name)\n"
400 sql_exec (plugin->dbh,
401 "CREATE TABLE IF NOT EXISTS state_sync (\n"
402 " channel_id INTEGER NOT NULL REFERENCES channels(id),\n"
403 " name TEXT NOT NULL,\n"
405 " PRIMARY KEY (channel_id, name)\n"
408 /* Prepare statements */
410 sql_prepare (plugin->dbh, "BEGIN;", &plugin->transaction_begin);
412 sql_prepare (plugin->dbh, "COMMIT;", &plugin->transaction_commit);
414 sql_prepare (plugin->dbh, "ROLLBACK;", &plugin->transaction_rollback);
416 sql_prepare (plugin->dbh,
417 "INSERT OR IGNORE INTO channels (pub_key) VALUES (?);",
418 &plugin->insert_channel_key);
420 sql_prepare (plugin->dbh,
421 "INSERT OR IGNORE INTO slaves (pub_key) VALUES (?);",
422 &plugin->insert_slave_key);
424 sql_prepare (plugin->dbh,
425 "INSERT INTO membership\n"
426 " (channel_id, slave_id, did_join, announced_at,\n"
427 " effective_since, group_generation)\n"
428 "VALUES ((SELECT id FROM channels WHERE pub_key = ?),\n"
429 " (SELECT id FROM slaves WHERE pub_key = ?),\n"
431 &plugin->insert_membership);
433 sql_prepare (plugin->dbh,
434 "SELECT did_join FROM membership\n"
435 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
436 " AND slave_id = (SELECT id FROM slaves WHERE pub_key = ?)\n"
437 " AND effective_since <= ? AND did_join = 1\n"
438 "ORDER BY announced_at DESC LIMIT 1;",
439 &plugin->select_membership);
441 sql_prepare (plugin->dbh,
442 "INSERT OR IGNORE INTO messages\n"
443 " (channel_id, hop_counter, signature, purpose,\n"
444 " fragment_id, fragment_offset, message_id,\n"
445 " group_generation, multicast_flags, psycstore_flags, data)\n"
446 "VALUES ((SELECT id FROM channels WHERE pub_key = ?),\n"
447 " ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);",
448 &plugin->insert_fragment);
450 sql_prepare (plugin->dbh,
452 "SET psycstore_flags = psycstore_flags | ?\n"
453 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
454 " AND message_id = ? AND fragment_offset = 0;",
455 &plugin->update_message_flags);
457 sql_prepare (plugin->dbh,
458 "SELECT hop_counter, signature, purpose, fragment_id,\n"
459 " fragment_offset, message_id, group_generation,\n"
460 " multicast_flags, psycstore_flags, data\n"
462 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
463 " AND ? <= fragment_id AND fragment_id <= ?;",
464 &plugin->select_fragments);
466 /** @todo select_messages: add method_prefix filter */
467 sql_prepare (plugin->dbh,
468 "SELECT hop_counter, signature, purpose, fragment_id,\n"
469 " fragment_offset, message_id, group_generation,\n"
470 " multicast_flags, psycstore_flags, data\n"
472 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
473 " AND ? <= message_id AND message_id <= ?"
475 &plugin->select_messages);
477 sql_prepare (plugin->dbh,
479 "(SELECT hop_counter, signature, purpose, fragment_id,\n"
480 " fragment_offset, message_id, group_generation,\n"
481 " multicast_flags, psycstore_flags, data\n"
483 " WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
484 " ORDER BY fragment_id DESC\n"
486 "ORDER BY fragment_id;",
487 &plugin->select_latest_fragments);
489 /** @todo select_latest_messages: add method_prefix filter */
490 sql_prepare (plugin->dbh,
491 "SELECT hop_counter, signature, purpose, fragment_id,\n"
492 " fragment_offset, message_id, group_generation,\n"
493 " multicast_flags, psycstore_flags, data\n"
495 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
496 " AND message_id IN\n"
497 " (SELECT message_id\n"
499 " WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
500 " GROUP BY message_id\n"
501 " ORDER BY message_id\n"
503 "ORDER BY fragment_id;",
504 &plugin->select_latest_messages);
506 sql_prepare (plugin->dbh,
507 "SELECT hop_counter, signature, purpose, fragment_id,\n"
508 " fragment_offset, message_id, group_generation,\n"
509 " multicast_flags, psycstore_flags, data\n"
511 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
512 " AND message_id = ? AND fragment_offset = ?;",
513 &plugin->select_message_fragment);
515 sql_prepare (plugin->dbh,
516 "SELECT fragment_id, message_id, group_generation\n"
518 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
519 "ORDER BY fragment_id DESC LIMIT 1;",
520 &plugin->select_counters_message);
522 sql_prepare (plugin->dbh,
523 "SELECT max_state_message_id\n"
525 "WHERE pub_key = ? AND max_state_message_id IS NOT NULL;",
526 &plugin->select_counters_state);
528 sql_prepare (plugin->dbh,
530 "SET max_state_message_id = ?\n"
531 "WHERE pub_key = ?;",
532 &plugin->update_max_state_message_id);
534 sql_prepare (plugin->dbh,
536 "SET state_hash_message_id = ?\n"
537 "WHERE pub_key = ?;",
538 &plugin->update_state_hash_message_id);
540 sql_prepare (plugin->dbh,
541 "INSERT OR REPLACE INTO state\n"
542 " (channel_id, name, value_current, value_signed)\n"
543 "SELECT new.channel_id, new.name,\n"
544 " new.value_current, old.value_signed\n"
545 "FROM (SELECT (SELECT id FROM channels WHERE pub_key = ?)\n"
547 " ? AS name, ? AS value_current) AS new\n"
548 "LEFT JOIN (SELECT channel_id, name, value_signed\n"
549 " FROM state) AS old\n"
550 "ON new.channel_id = old.channel_id AND new.name = old.name;",
551 &plugin->insert_state_current);
553 sql_prepare (plugin->dbh,
554 "DELETE FROM state\n"
555 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
556 " AND (value_current IS NULL OR length(value_current) = 0)\n"
557 " AND (value_signed IS NULL OR length(value_signed) = 0);",
558 &plugin->delete_state_empty);
560 sql_prepare (plugin->dbh,
562 "SET value_signed = value_current\n"
563 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
564 &plugin->update_state_signed);
566 sql_prepare (plugin->dbh,
567 "DELETE FROM state\n"
568 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
569 &plugin->delete_state);
571 sql_prepare (plugin->dbh,
572 "INSERT INTO state_sync (channel_id, name, value)\n"
573 "VALUES ((SELECT id FROM channels WHERE pub_key = ?), ?, ?);",
574 &plugin->insert_state_sync);
576 sql_prepare (plugin->dbh,
577 "INSERT INTO state\n"
578 " (channel_id, name, value_current, value_signed)\n"
579 "SELECT channel_id, name, value, value\n"
581 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
582 &plugin->insert_state_from_sync);
584 sql_prepare (plugin->dbh,
585 "DELETE FROM state_sync\n"
586 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
587 &plugin->delete_state_sync);
589 sql_prepare (plugin->dbh,
590 "SELECT value_current\n"
592 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
594 &plugin->select_state_one);
596 sql_prepare (plugin->dbh,
597 "SELECT name, value_current\n"
599 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
600 " AND (name = ? OR substr(name, 1, ?) = ?);",
601 &plugin->select_state_prefix);
603 sql_prepare (plugin->dbh,
604 "SELECT name, value_signed\n"
606 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)"
607 " AND value_signed IS NOT NULL;",
608 &plugin->select_state_signed);
615 * Shutdown database connection and associate data
617 * @param plugin the plugin context (state for this module)
620 database_shutdown (struct Plugin *plugin)
624 while (NULL != (stmt = sqlite3_next_stmt (plugin->dbh, NULL)))
626 result = sqlite3_finalize (stmt);
627 if (SQLITE_OK != result)
628 LOG (GNUNET_ERROR_TYPE_WARNING,
629 "Failed to close statement %p: %d\n", stmt, result);
631 if (SQLITE_OK != sqlite3_close (plugin->dbh))
632 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR, "sqlite3_close");
634 GNUNET_free_non_null (plugin->fn);
638 * Execute a prepared statement with a @a channel_key argument.
640 * @param plugin Plugin handle.
641 * @param stmt Statement to execute.
642 * @param channel_key Public key of the channel.
644 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
647 exec_channel (struct Plugin *plugin, sqlite3_stmt *stmt,
648 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
650 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
651 sizeof (*channel_key), SQLITE_STATIC))
653 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
656 else if (SQLITE_DONE != sqlite3_step (stmt))
658 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
662 if (SQLITE_OK != sqlite3_reset (stmt))
664 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
666 return GNUNET_SYSERR;
673 * Begin a transaction.
676 transaction_begin (struct Plugin *plugin, enum Transactions transaction)
678 sqlite3_stmt *stmt = plugin->transaction_begin;
680 if (SQLITE_DONE != sqlite3_step (stmt))
682 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
685 if (SQLITE_OK != sqlite3_reset (stmt))
687 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
689 return GNUNET_SYSERR;
692 plugin->transaction = transaction;
698 * Commit current transaction.
701 transaction_commit (struct Plugin *plugin)
703 sqlite3_stmt *stmt = plugin->transaction_commit;
705 if (SQLITE_DONE != sqlite3_step (stmt))
707 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
710 if (SQLITE_OK != sqlite3_reset (stmt))
712 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
714 return GNUNET_SYSERR;
717 plugin->transaction = TRANSACTION_NONE;
723 * Roll back current transaction.
726 transaction_rollback (struct Plugin *plugin)
728 sqlite3_stmt *stmt = plugin->transaction_rollback;
730 if (SQLITE_DONE != sqlite3_step (stmt))
732 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
735 if (SQLITE_OK != sqlite3_reset (stmt))
737 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
739 return GNUNET_SYSERR;
741 plugin->transaction = TRANSACTION_NONE;
747 channel_key_store (struct Plugin *plugin,
748 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
750 sqlite3_stmt *stmt = plugin->insert_channel_key;
752 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
753 sizeof (*channel_key), SQLITE_STATIC))
755 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
758 else if (SQLITE_DONE != sqlite3_step (stmt))
760 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
764 if (SQLITE_OK != sqlite3_reset (stmt))
766 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
768 return GNUNET_SYSERR;
776 slave_key_store (struct Plugin *plugin,
777 const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key)
779 sqlite3_stmt *stmt = plugin->insert_slave_key;
781 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, slave_key,
782 sizeof (*slave_key), SQLITE_STATIC))
784 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
787 else if (SQLITE_DONE != sqlite3_step (stmt))
789 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
794 if (SQLITE_OK != sqlite3_reset (stmt))
796 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
798 return GNUNET_SYSERR;
806 * Store join/leave events for a PSYC channel in order to be able to answer
807 * membership test queries later.
809 * @see GNUNET_PSYCSTORE_membership_store()
811 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
814 sqlite_membership_store (void *cls,
815 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
816 const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
818 uint64_t announced_at,
819 uint64_t effective_since,
820 uint64_t group_generation)
822 struct Plugin *plugin = cls;
823 sqlite3_stmt *stmt = plugin->insert_membership;
825 GNUNET_assert (TRANSACTION_NONE == plugin->transaction);
827 if (announced_at > INT64_MAX ||
828 effective_since > INT64_MAX ||
829 group_generation > INT64_MAX)
832 return GNUNET_SYSERR;
835 if (GNUNET_OK != channel_key_store (plugin, channel_key)
836 || GNUNET_OK != slave_key_store (plugin, slave_key))
837 return GNUNET_SYSERR;
839 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
840 sizeof (*channel_key), SQLITE_STATIC)
841 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, slave_key,
842 sizeof (*slave_key), SQLITE_STATIC)
843 || SQLITE_OK != sqlite3_bind_int (stmt, 3, did_join)
844 || SQLITE_OK != sqlite3_bind_int64 (stmt, 4, announced_at)
845 || SQLITE_OK != sqlite3_bind_int64 (stmt, 5, effective_since)
846 || SQLITE_OK != sqlite3_bind_int64 (stmt, 6, group_generation))
848 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
851 else if (SQLITE_DONE != sqlite3_step (stmt))
853 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
857 if (SQLITE_OK != sqlite3_reset (stmt))
859 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
861 return GNUNET_SYSERR;
868 * Test if a member was admitted to the channel at the given message ID.
870 * @see GNUNET_PSYCSTORE_membership_test()
872 * @return #GNUNET_YES if the member was admitted, #GNUNET_NO if not,
873 * #GNUNET_SYSERR if there was en error.
876 membership_test (void *cls,
877 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
878 const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
881 struct Plugin *plugin = cls;
882 sqlite3_stmt *stmt = plugin->select_membership;
883 int ret = GNUNET_SYSERR;
885 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
886 sizeof (*channel_key), SQLITE_STATIC)
887 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, slave_key,
888 sizeof (*slave_key), SQLITE_STATIC)
889 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, message_id))
891 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
896 switch (sqlite3_step (stmt))
906 if (SQLITE_OK != sqlite3_reset (stmt))
908 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
916 * Store a message fragment sent to a channel.
918 * @see GNUNET_PSYCSTORE_fragment_store()
920 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
923 fragment_store (void *cls,
924 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
925 const struct GNUNET_MULTICAST_MessageHeader *msg,
926 uint32_t psycstore_flags)
928 struct Plugin *plugin = cls;
929 sqlite3_stmt *stmt = plugin->insert_fragment;
931 GNUNET_assert (TRANSACTION_NONE == plugin->transaction);
933 uint64_t fragment_id = GNUNET_ntohll (msg->fragment_id);
934 uint64_t fragment_offset = GNUNET_ntohll (msg->fragment_offset);
935 uint64_t message_id = GNUNET_ntohll (msg->message_id);
936 uint64_t group_generation = GNUNET_ntohll (msg->group_generation);
938 if (fragment_id > INT64_MAX || fragment_offset > INT64_MAX ||
939 message_id > INT64_MAX || group_generation > INT64_MAX)
941 LOG (GNUNET_ERROR_TYPE_ERROR,
942 "Tried to store fragment with a field > INT64_MAX: "
943 "%lu, %lu, %lu, %lu\n", fragment_id, fragment_offset,
944 message_id, group_generation);
946 return GNUNET_SYSERR;
949 if (GNUNET_OK != channel_key_store (plugin, channel_key))
950 return GNUNET_SYSERR;
952 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
953 sizeof (*channel_key), SQLITE_STATIC)
954 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, ntohl (msg->hop_counter) )
955 || SQLITE_OK != sqlite3_bind_blob (stmt, 3, (const void *) &msg->signature,
956 sizeof (msg->signature), SQLITE_STATIC)
957 || SQLITE_OK != sqlite3_bind_blob (stmt, 4, (const void *) &msg->purpose,
958 sizeof (msg->purpose), SQLITE_STATIC)
959 || SQLITE_OK != sqlite3_bind_int64 (stmt, 5, fragment_id)
960 || SQLITE_OK != sqlite3_bind_int64 (stmt, 6, fragment_offset)
961 || SQLITE_OK != sqlite3_bind_int64 (stmt, 7, message_id)
962 || SQLITE_OK != sqlite3_bind_int64 (stmt, 8, group_generation)
963 || SQLITE_OK != sqlite3_bind_int64 (stmt, 9, ntohl (msg->flags))
964 || SQLITE_OK != sqlite3_bind_int64 (stmt, 10, psycstore_flags)
965 || SQLITE_OK != sqlite3_bind_blob (stmt, 11, (const void *) &msg[1],
966 ntohs (msg->header.size)
967 - sizeof (*msg), SQLITE_STATIC))
969 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
972 else if (SQLITE_DONE != sqlite3_step (stmt))
974 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
978 if (SQLITE_OK != sqlite3_reset (stmt))
980 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
982 return GNUNET_SYSERR;
989 * Set additional flags for a given message.
991 * They are OR'd with any existing flags set.
993 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
996 message_add_flags (void *cls,
997 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
999 uint32_t psycstore_flags)
1001 struct Plugin *plugin = cls;
1002 sqlite3_stmt *stmt = plugin->update_message_flags;
1003 int ret = GNUNET_SYSERR;
1005 if (SQLITE_OK != sqlite3_bind_int64 (stmt, 1, psycstore_flags)
1006 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, channel_key,
1007 sizeof (*channel_key), SQLITE_STATIC)
1008 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, message_id))
1010 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1015 switch (sqlite3_step (stmt))
1018 ret = sqlite3_total_changes (plugin->dbh) > 0 ? GNUNET_OK : GNUNET_NO;
1021 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1026 if (SQLITE_OK != sqlite3_reset (stmt))
1028 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1030 return GNUNET_SYSERR;
1037 fragment_row (sqlite3_stmt *stmt, GNUNET_PSYCSTORE_FragmentCallback cb,
1040 int data_size = sqlite3_column_bytes (stmt, 9);
1041 struct GNUNET_MULTICAST_MessageHeader *msg
1042 = GNUNET_malloc (sizeof (*msg) + data_size);
1044 msg->header.size = htons (sizeof (*msg) + data_size);
1045 msg->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE);
1046 msg->hop_counter = htonl ((uint32_t) sqlite3_column_int64 (stmt, 0));
1047 GNUNET_memcpy (&msg->signature,
1048 sqlite3_column_blob (stmt, 1),
1049 sqlite3_column_bytes (stmt, 1));
1050 GNUNET_memcpy (&msg->purpose,
1051 sqlite3_column_blob (stmt, 2),
1052 sqlite3_column_bytes (stmt, 2));
1053 msg->fragment_id = GNUNET_htonll (sqlite3_column_int64 (stmt, 3));
1054 msg->fragment_offset = GNUNET_htonll (sqlite3_column_int64 (stmt, 4));
1055 msg->message_id = GNUNET_htonll (sqlite3_column_int64 (stmt, 5));
1056 msg->group_generation = GNUNET_htonll (sqlite3_column_int64 (stmt, 6));
1057 msg->flags = htonl (sqlite3_column_int64 (stmt, 7));
1058 GNUNET_memcpy (&msg[1], sqlite3_column_blob (stmt, 9), data_size);
1060 return cb (cb_cls, (void *) msg, sqlite3_column_int64 (stmt, 8));
1065 fragment_select (struct Plugin *plugin, sqlite3_stmt *stmt,
1066 uint64_t *returned_fragments,
1067 GNUNET_PSYCSTORE_FragmentCallback cb, void *cb_cls)
1069 int ret = GNUNET_SYSERR;
1074 sql_ret = sqlite3_step (stmt);
1078 if (ret != GNUNET_OK)
1082 ret = fragment_row (stmt, cb, cb_cls);
1083 (*returned_fragments)++;
1084 if (ret != GNUNET_YES)
1085 sql_ret = SQLITE_DONE;
1088 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1092 while (sql_ret == SQLITE_ROW);
1098 * Retrieve a message fragment range by fragment ID.
1100 * @see GNUNET_PSYCSTORE_fragment_get()
1102 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1105 fragment_get (void *cls,
1106 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1107 uint64_t first_fragment_id,
1108 uint64_t last_fragment_id,
1109 uint64_t *returned_fragments,
1110 GNUNET_PSYCSTORE_FragmentCallback cb,
1113 struct Plugin *plugin = cls;
1114 sqlite3_stmt *stmt = plugin->select_fragments;
1115 int ret = GNUNET_SYSERR;
1116 *returned_fragments = 0;
1118 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1119 sizeof (*channel_key),
1121 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, first_fragment_id)
1122 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, last_fragment_id))
1124 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1129 ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls);
1132 if (SQLITE_OK != sqlite3_reset (stmt))
1134 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1143 * Retrieve a message fragment range by fragment ID.
1145 * @see GNUNET_PSYCSTORE_fragment_get_latest()
1147 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1150 fragment_get_latest (void *cls,
1151 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1152 uint64_t fragment_limit,
1153 uint64_t *returned_fragments,
1154 GNUNET_PSYCSTORE_FragmentCallback cb,
1157 struct Plugin *plugin = cls;
1158 sqlite3_stmt *stmt = plugin->select_latest_fragments;
1159 int ret = GNUNET_SYSERR;
1160 *returned_fragments = 0;
1162 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1163 sizeof (*channel_key),
1165 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, fragment_limit))
1167 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1172 ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls);
1175 if (SQLITE_OK != sqlite3_reset (stmt))
1177 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1186 * Retrieve all fragments of a message ID range.
1188 * @see GNUNET_PSYCSTORE_message_get()
1190 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1193 message_get (void *cls,
1194 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1195 uint64_t first_message_id,
1196 uint64_t last_message_id,
1197 uint64_t fragment_limit,
1198 uint64_t *returned_fragments,
1199 GNUNET_PSYCSTORE_FragmentCallback cb,
1202 struct Plugin *plugin = cls;
1203 sqlite3_stmt *stmt = plugin->select_messages;
1204 int ret = GNUNET_SYSERR;
1205 *returned_fragments = 0;
1207 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1208 sizeof (*channel_key),
1210 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, first_message_id)
1211 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, last_message_id)
1212 || SQLITE_OK != sqlite3_bind_int64 (stmt, 4,
1213 (0 != fragment_limit)
1217 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1222 ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls);
1225 if (SQLITE_OK != sqlite3_reset (stmt))
1227 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1236 * Retrieve all fragments of the latest messages.
1238 * @see GNUNET_PSYCSTORE_message_get_latest()
1240 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1243 message_get_latest (void *cls,
1244 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1245 uint64_t message_limit,
1246 uint64_t *returned_fragments,
1247 GNUNET_PSYCSTORE_FragmentCallback cb,
1250 struct Plugin *plugin = cls;
1251 sqlite3_stmt *stmt = plugin->select_latest_messages;
1252 int ret = GNUNET_SYSERR;
1253 *returned_fragments = 0;
1255 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1256 sizeof (*channel_key),
1258 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, channel_key,
1259 sizeof (*channel_key),
1261 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, message_limit))
1263 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1268 ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls);
1271 if (SQLITE_OK != sqlite3_reset (stmt))
1273 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1282 * Retrieve a fragment of message specified by its message ID and fragment
1285 * @see GNUNET_PSYCSTORE_message_get_fragment()
1287 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1290 message_get_fragment (void *cls,
1291 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1292 uint64_t message_id,
1293 uint64_t fragment_offset,
1294 GNUNET_PSYCSTORE_FragmentCallback cb,
1297 struct Plugin *plugin = cls;
1298 sqlite3_stmt *stmt = plugin->select_message_fragment;
1299 int ret = GNUNET_SYSERR;
1301 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1302 sizeof (*channel_key),
1304 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, message_id)
1305 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, fragment_offset))
1307 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1312 switch (sqlite3_step (stmt))
1318 ret = fragment_row (stmt, cb, cb_cls);
1321 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1326 if (SQLITE_OK != sqlite3_reset (stmt))
1328 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1336 * Retrieve the max. values of message counters for a channel.
1338 * @see GNUNET_PSYCSTORE_counters_get()
1340 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1343 counters_message_get (void *cls,
1344 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1345 uint64_t *max_fragment_id,
1346 uint64_t *max_message_id,
1347 uint64_t *max_group_generation)
1349 struct Plugin *plugin = cls;
1350 sqlite3_stmt *stmt = plugin->select_counters_message;
1351 int ret = GNUNET_SYSERR;
1353 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1354 sizeof (*channel_key),
1357 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1362 switch (sqlite3_step (stmt))
1368 *max_fragment_id = sqlite3_column_int64 (stmt, 0);
1369 *max_message_id = sqlite3_column_int64 (stmt, 1);
1370 *max_group_generation = sqlite3_column_int64 (stmt, 2);
1374 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1379 if (SQLITE_OK != sqlite3_reset (stmt))
1381 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1389 * Retrieve the max. values of state counters for a channel.
1391 * @see GNUNET_PSYCSTORE_counters_get()
1393 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1396 counters_state_get (void *cls,
1397 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1398 uint64_t *max_state_message_id)
1400 struct Plugin *plugin = cls;
1401 sqlite3_stmt *stmt = plugin->select_counters_state;
1402 int ret = GNUNET_SYSERR;
1404 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1405 sizeof (*channel_key),
1408 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1413 switch (sqlite3_step (stmt))
1419 *max_state_message_id = sqlite3_column_int64 (stmt, 0);
1423 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1428 if (SQLITE_OK != sqlite3_reset (stmt))
1430 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1439 * Assign a value to a state variable.
1441 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1444 state_assign (struct Plugin *plugin, sqlite3_stmt *stmt,
1445 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1446 const char *name, const void *value, size_t value_size)
1448 int ret = GNUNET_SYSERR;
1450 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1451 sizeof (*channel_key), SQLITE_STATIC)
1452 || SQLITE_OK != sqlite3_bind_text (stmt, 2, name, -1, SQLITE_STATIC)
1453 || SQLITE_OK != sqlite3_bind_blob (stmt, 3, value, value_size,
1456 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1461 switch (sqlite3_step (stmt))
1464 ret = 0 < sqlite3_total_changes (plugin->dbh) ? GNUNET_OK : GNUNET_NO;
1467 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1472 if (SQLITE_OK != sqlite3_reset (stmt))
1474 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1476 return GNUNET_SYSERR;
1484 update_message_id (struct Plugin *plugin, sqlite3_stmt *stmt,
1485 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1486 uint64_t message_id)
1488 if (SQLITE_OK != sqlite3_bind_int64 (stmt, 1, message_id)
1489 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, channel_key,
1490 sizeof (*channel_key), SQLITE_STATIC))
1492 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1495 else if (SQLITE_DONE != sqlite3_step (stmt))
1497 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1500 if (SQLITE_OK != sqlite3_reset (stmt))
1502 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1504 return GNUNET_SYSERR;
1511 * Begin modifying current state.
1514 state_modify_begin (void *cls,
1515 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1516 uint64_t message_id, uint64_t state_delta)
1518 struct Plugin *plugin = cls;
1520 if (state_delta > 0)
1523 * We can only apply state modifiers in the current message if modifiers in
1524 * the previous stateful message (message_id - state_delta) were already
1528 uint64_t max_state_message_id = 0;
1529 int ret = counters_state_get (plugin, channel_key, &max_state_message_id);
1533 case GNUNET_NO: // no state yet
1540 if (max_state_message_id < message_id - state_delta)
1541 return GNUNET_NO; /* some stateful messages not yet applied */
1542 else if (message_id - state_delta < max_state_message_id)
1543 return GNUNET_NO; /* changes already applied */
1546 if (TRANSACTION_NONE != plugin->transaction)
1548 /** @todo FIXME: wait for other transaction to finish */
1549 return GNUNET_SYSERR;
1551 return transaction_begin (plugin, TRANSACTION_STATE_MODIFY);
1556 * Set the current value of state variable.
1558 * @see GNUNET_PSYCSTORE_state_modify()
1560 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1563 state_modify_op (void *cls,
1564 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1565 enum GNUNET_PSYC_Operator op,
1566 const char *name, const void *value, size_t value_size)
1568 struct Plugin *plugin = cls;
1569 GNUNET_assert (TRANSACTION_STATE_MODIFY == plugin->transaction);
1573 case GNUNET_PSYC_OP_ASSIGN:
1574 return state_assign (plugin, plugin->insert_state_current, channel_key,
1575 name, value, value_size);
1577 default: /** @todo implement more state operations */
1579 return GNUNET_SYSERR;
1585 * End modifying current state.
1588 state_modify_end (void *cls,
1589 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1590 uint64_t message_id)
1592 struct Plugin *plugin = cls;
1593 GNUNET_assert (TRANSACTION_STATE_MODIFY == plugin->transaction);
1596 GNUNET_OK == exec_channel (plugin, plugin->delete_state_empty, channel_key)
1597 && GNUNET_OK == update_message_id (plugin,
1598 plugin->update_max_state_message_id,
1599 channel_key, message_id)
1600 && GNUNET_OK == transaction_commit (plugin)
1601 ? GNUNET_OK : GNUNET_SYSERR;
1606 * Begin state synchronization.
1609 state_sync_begin (void *cls,
1610 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
1612 struct Plugin *plugin = cls;
1613 return exec_channel (plugin, plugin->delete_state_sync, channel_key);
1618 * Assign current value of a state variable.
1620 * @see GNUNET_PSYCSTORE_state_modify()
1622 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1625 state_sync_assign (void *cls,
1626 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1627 const char *name, const void *value, size_t value_size)
1629 struct Plugin *plugin = cls;
1630 return state_assign (cls, plugin->insert_state_sync, channel_key,
1631 name, value, value_size);
1636 * End modifying current state.
1639 state_sync_end (void *cls,
1640 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1641 uint64_t max_state_message_id,
1642 uint64_t state_hash_message_id)
1644 struct Plugin *plugin = cls;
1645 int ret = GNUNET_SYSERR;
1647 if (TRANSACTION_NONE != plugin->transaction)
1649 /** @todo FIXME: wait for other transaction to finish */
1650 return GNUNET_SYSERR;
1653 GNUNET_OK == transaction_begin (plugin, TRANSACTION_STATE_SYNC)
1654 && GNUNET_OK == exec_channel (plugin, plugin->delete_state, channel_key)
1655 && GNUNET_OK == exec_channel (plugin, plugin->insert_state_from_sync,
1657 && GNUNET_OK == exec_channel (plugin, plugin->delete_state_sync,
1659 && GNUNET_OK == update_message_id (plugin,
1660 plugin->update_state_hash_message_id,
1661 channel_key, state_hash_message_id)
1662 && GNUNET_OK == update_message_id (plugin,
1663 plugin->update_max_state_message_id,
1664 channel_key, max_state_message_id)
1665 && GNUNET_OK == transaction_commit (plugin)
1667 : transaction_rollback (plugin);
1673 * Delete the whole state.
1675 * @see GNUNET_PSYCSTORE_state_reset()
1677 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1680 state_reset (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
1682 struct Plugin *plugin = cls;
1683 return exec_channel (plugin, plugin->delete_state, channel_key);
1688 * Update signed values of state variables in the state store.
1690 * @see GNUNET_PSYCSTORE_state_hash_update()
1692 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1695 state_update_signed (void *cls,
1696 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
1698 struct Plugin *plugin = cls;
1699 return exec_channel (plugin, plugin->update_state_signed, channel_key);
1704 * Retrieve a state variable by name.
1706 * @see GNUNET_PSYCSTORE_state_get()
1708 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1711 state_get (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1712 const char *name, GNUNET_PSYCSTORE_StateCallback cb, void *cb_cls)
1714 struct Plugin *plugin = cls;
1715 int ret = GNUNET_SYSERR;
1717 sqlite3_stmt *stmt = plugin->select_state_one;
1719 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1720 sizeof (*channel_key),
1722 || SQLITE_OK != sqlite3_bind_text (stmt, 2, name, -1, SQLITE_STATIC))
1724 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1729 switch (sqlite3_step (stmt))
1735 ret = cb (cb_cls, name, sqlite3_column_blob (stmt, 0),
1736 sqlite3_column_bytes (stmt, 0));
1739 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1744 if (SQLITE_OK != sqlite3_reset (stmt))
1746 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1755 * Retrieve all state variables for a channel with the given prefix.
1757 * @see GNUNET_PSYCSTORE_state_get_prefix()
1759 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1762 state_get_prefix (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1763 const char *name, GNUNET_PSYCSTORE_StateCallback cb,
1766 struct Plugin *plugin = cls;
1767 int ret = GNUNET_SYSERR;
1768 sqlite3_stmt *stmt = plugin->select_state_prefix;
1769 size_t name_len = strlen (name);
1771 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1772 sizeof (*channel_key), SQLITE_STATIC)
1773 || SQLITE_OK != sqlite3_bind_text (stmt, 2, name, name_len, SQLITE_STATIC)
1774 || SQLITE_OK != sqlite3_bind_int (stmt, 3, name_len)
1775 || SQLITE_OK != sqlite3_bind_text (stmt, 4, name, name_len, SQLITE_STATIC))
1777 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1785 sql_ret = sqlite3_step (stmt);
1789 if (ret != GNUNET_OK)
1793 ret = cb (cb_cls, (const char *) sqlite3_column_text (stmt, 0),
1794 sqlite3_column_blob (stmt, 1),
1795 sqlite3_column_bytes (stmt, 1));
1796 if (ret != GNUNET_YES)
1797 sql_ret = SQLITE_DONE;
1800 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1804 while (sql_ret == SQLITE_ROW);
1806 if (SQLITE_OK != sqlite3_reset (stmt))
1808 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1816 * Retrieve all signed state variables for a channel.
1818 * @see GNUNET_PSYCSTORE_state_get_signed()
1820 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1823 state_get_signed (void *cls,
1824 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1825 GNUNET_PSYCSTORE_StateCallback cb, void *cb_cls)
1827 struct Plugin *plugin = cls;
1828 int ret = GNUNET_SYSERR;
1830 sqlite3_stmt *stmt = plugin->select_state_signed;
1832 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1833 sizeof (*channel_key), SQLITE_STATIC))
1835 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1843 sql_ret = sqlite3_step (stmt);
1847 if (ret != GNUNET_OK)
1851 ret = cb (cb_cls, (const char *) sqlite3_column_text (stmt, 0),
1852 sqlite3_column_blob (stmt, 1),
1853 sqlite3_column_bytes (stmt, 1));
1854 if (ret != GNUNET_YES)
1855 sql_ret = SQLITE_DONE;
1858 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1862 while (sql_ret == SQLITE_ROW);
1865 if (SQLITE_OK != sqlite3_reset (stmt))
1867 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1876 * Entry point for the plugin.
1878 * @param cls The struct GNUNET_CONFIGURATION_Handle.
1879 * @return NULL on error, otherwise the plugin context
1882 libgnunet_plugin_psycstore_sqlite_init (void *cls)
1884 static struct Plugin plugin;
1885 const struct GNUNET_CONFIGURATION_Handle *cfg = cls;
1886 struct GNUNET_PSYCSTORE_PluginFunctions *api;
1888 if (NULL != plugin.cfg)
1889 return NULL; /* can only initialize once! */
1890 memset (&plugin, 0, sizeof (struct Plugin));
1892 if (GNUNET_OK != database_setup (&plugin))
1894 database_shutdown (&plugin);
1897 api = GNUNET_new (struct GNUNET_PSYCSTORE_PluginFunctions);
1899 api->membership_store = &sqlite_membership_store;
1900 api->membership_test = &membership_test;
1901 api->fragment_store = &fragment_store;
1902 api->message_add_flags = &message_add_flags;
1903 api->fragment_get = &fragment_get;
1904 api->fragment_get_latest = &fragment_get_latest;
1905 api->message_get = &message_get;
1906 api->message_get_latest = &message_get_latest;
1907 api->message_get_fragment = &message_get_fragment;
1908 api->counters_message_get = &counters_message_get;
1909 api->counters_state_get = &counters_state_get;
1910 api->state_modify_begin = &state_modify_begin;
1911 api->state_modify_op = &state_modify_op;
1912 api->state_modify_end = &state_modify_end;
1913 api->state_sync_begin = &state_sync_begin;
1914 api->state_sync_assign = &state_sync_assign;
1915 api->state_sync_end = &state_sync_end;
1916 api->state_reset = &state_reset;
1917 api->state_update_signed = &state_update_signed;
1918 api->state_get = &state_get;
1919 api->state_get_prefix = &state_get_prefix;
1920 api->state_get_signed = &state_get_signed;
1922 LOG (GNUNET_ERROR_TYPE_INFO, _("SQLite database running\n"));
1928 * Exit point from the plugin.
1930 * @param cls The plugin context (as returned by "init")
1931 * @return Always NULL
1934 libgnunet_plugin_psycstore_sqlite_done (void *cls)
1936 struct GNUNET_PSYCSTORE_PluginFunctions *api = cls;
1937 struct Plugin *plugin = api->cls;
1939 database_shutdown (plugin);
1942 LOG (GNUNET_ERROR_TYPE_DEBUG, "SQLite plugin is finished\n");
1946 /* end of plugin_psycstore_sqlite.c */