2 * This file is part of GNUnet
3 * (C) 2013 Christian Grothoff (and other contributing authors)
5 * GNUnet is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License as published
7 * by the Free Software Foundation; either version 3, or (at your
8 * 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 * General Public License for more details.
15 * You should have received a copy of the GNU General Public License
16 * along with GNUnet; see the file COPYING. If not, write to the
17 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
18 * Boston, MA 02111-1307, USA.
22 * @file psycstore/plugin_psycstore_sqlite.c
23 * @brief sqlite-based psycstore backend
24 * @author Gabor X Toth
25 * @author Christian Grothoff
29 * FIXME: SQLite3 only supports signed 64-bit integers natively,
30 * thus it can only store 63 bits of the uint64_t's.
34 #include "gnunet_psycstore_plugin.h"
35 #include "gnunet_psycstore_service.h"
36 #include "gnunet_multicast_service.h"
37 #include "gnunet_crypto_lib.h"
38 #include "psycstore.h"
42 * After how many ms "busy" should a DB operation fail for good? A
43 * low value makes sure that we are more responsive to requests
44 * (especially PUTs). A high value guarantees a higher success rate
45 * (SELECTs in iterate can take several seconds despite LIMIT=1).
47 * The default value of 1s should ensure that users do not experience
48 * huge latencies while at the same time allowing operations to
49 * succeed with reasonable probability.
51 #define BUSY_TIMEOUT_MS 1000
53 #define DEBUG_PSYCSTORE GNUNET_EXTRA_LOGGING
56 * Log an error message at log-level 'level' that indicates
57 * a failure of the command 'cmd' on file 'filename'
58 * with the message given by strerror(errno).
60 #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)
62 #define LOG(kind,...) GNUNET_log_from (kind, "psycstore-sqlite", __VA_ARGS__)
66 TRANSACTION_STATE_MODIFY
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;
177 * Precompiled SQL for message_modify_begin()
179 sqlite3_stmt *select_message_state_delta;
182 * Precompiled SQL for state_modify_set()
184 sqlite3_stmt *insert_state_current;
187 * Precompiled SQL for state_modify_end()
189 sqlite3_stmt *delete_state_empty;
192 * Precompiled SQL for state_set_signed()
194 sqlite3_stmt *update_state_signed;
197 * Precompiled SQL for state_sync()
199 sqlite3_stmt *insert_state_sync;
202 * Precompiled SQL for state_sync()
204 sqlite3_stmt *delete_state;
207 * Precompiled SQL for state_sync()
209 sqlite3_stmt *insert_state_from_sync;
212 * Precompiled SQL for state_sync()
214 sqlite3_stmt *delete_state_sync;
217 * Precompiled SQL for state_get_signed()
219 sqlite3_stmt *select_state_signed;
222 * Precompiled SQL for state_get()
224 sqlite3_stmt *select_state_one;
227 * Precompiled SQL for state_get_prefix()
229 sqlite3_stmt *select_state_prefix;
236 sql_trace (void *cls, const char *sql)
238 LOG (GNUNET_ERROR_TYPE_DEBUG, "SQL query:\n%s\n", sql);
244 * @brief Prepare a SQL statement
246 * @param dbh handle to the database
247 * @param sql SQL statement, UTF-8 encoded
248 * @param stmt set to the prepared statement
249 * @return 0 on success
252 sql_prepare (sqlite3 *dbh, const char *sql, sqlite3_stmt **stmt)
257 result = sqlite3_prepare_v2 (dbh, sql, strlen (sql), stmt,
258 (const char **) &tail);
259 LOG (GNUNET_ERROR_TYPE_DEBUG,
260 "Prepared `%s' / %p: %d\n", sql, *stmt, result);
261 if (result != SQLITE_OK)
262 LOG (GNUNET_ERROR_TYPE_ERROR,
263 _("Error preparing SQL query: %s\n %s\n"),
264 sqlite3_errmsg (dbh), sql);
270 * @brief Prepare a SQL statement
272 * @param dbh handle to the database
273 * @param sql SQL statement, UTF-8 encoded
274 * @return 0 on success
277 sql_exec (sqlite3 *dbh, const char *sql)
281 result = sqlite3_exec (dbh, sql, NULL, NULL, NULL);
282 LOG (GNUNET_ERROR_TYPE_DEBUG,
283 "Executed `%s' / %d\n", sql, result);
284 if (result != SQLITE_OK)
285 LOG (GNUNET_ERROR_TYPE_ERROR,
286 _("Error executing SQL query: %s\n %s\n"),
287 sqlite3_errmsg (dbh), sql);
293 * Initialize the database connections and associated
294 * data structures (create tables and indices
295 * as needed as well).
297 * @param plugin the plugin context (state for this module)
298 * @return GNUNET_OK on success
301 database_setup (struct Plugin *plugin)
306 GNUNET_CONFIGURATION_get_value_filename (plugin->cfg, "psycstore-sqlite",
307 "FILENAME", &filename))
309 GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
310 "psycstore-sqlite", "FILENAME");
311 return GNUNET_SYSERR;
313 if (GNUNET_OK != GNUNET_DISK_file_test (filename))
315 if (GNUNET_OK != GNUNET_DISK_directory_create_for_file (filename))
318 GNUNET_free (filename);
319 return GNUNET_SYSERR;
322 /* filename should be UTF-8-encoded. If it isn't, it's a bug */
323 plugin->fn = filename;
325 /* Open database and precompile statements */
326 if (SQLITE_OK != sqlite3_open (plugin->fn, &plugin->dbh))
328 LOG (GNUNET_ERROR_TYPE_ERROR,
329 _("Unable to initialize SQLite: %s.\n"),
330 sqlite3_errmsg (plugin->dbh));
331 return GNUNET_SYSERR;
335 sqlite3_trace (plugin->dbh, &sql_trace, NULL);
338 sql_exec (plugin->dbh, "PRAGMA temp_store=MEMORY");
339 sql_exec (plugin->dbh, "PRAGMA synchronous=NORMAL");
340 sql_exec (plugin->dbh, "PRAGMA legacy_file_format=OFF");
341 sql_exec (plugin->dbh, "PRAGMA auto_vacuum=INCREMENTAL");
342 sql_exec (plugin->dbh, "PRAGMA encoding=\"UTF-8\"");
343 #if ! DEBUG_PSYCSTORE
344 sql_exec (plugin->dbh, "PRAGMA locking_mode=EXCLUSIVE");
346 sql_exec (plugin->dbh, "PRAGMA page_size=4096");
348 sqlite3_busy_timeout (plugin->dbh, BUSY_TIMEOUT_MS);
352 sql_exec (plugin->dbh,
353 "CREATE TABLE IF NOT EXISTS channels (\n"
354 " id INTEGER PRIMARY KEY,\n"
355 " pub_key BLOB UNIQUE,\n"
356 " max_state_message_id INTEGER,\n"
357 " state_hash_message_id INTEGER\n"
360 sql_exec (plugin->dbh,
361 "CREATE TABLE IF NOT EXISTS slaves (\n"
362 " id INTEGER PRIMARY KEY,\n"
363 " pub_key BLOB UNIQUE\n"
366 sql_exec (plugin->dbh,
367 "CREATE TABLE IF NOT EXISTS membership (\n"
368 " channel_id INTEGER NOT NULL REFERENCES channels(id),\n"
369 " slave_id INTEGER NOT NULL REFERENCES slaves(id),\n"
370 " did_join INTEGER NOT NULL,\n"
371 " announced_at INTEGER NOT NULL,\n"
372 " effective_since INTEGER NOT NULL,\n"
373 " group_generation INTEGER NOT NULL\n"
375 sql_exec (plugin->dbh,
376 "CREATE INDEX IF NOT EXISTS idx_membership_channel_id_slave_id "
377 "ON membership (channel_id, slave_id);");
379 sql_exec (plugin->dbh,
380 "CREATE TABLE IF NOT EXISTS messages (\n"
381 " channel_id INTEGER NOT NULL REFERENCES channels(id),\n"
382 " hop_counter INTEGER NOT NULL,\n"
385 " fragment_id INTEGER NOT NULL,\n"
386 " fragment_offset INTEGER NOT NULL,\n"
387 " message_id INTEGER NOT NULL,\n"
388 " group_generation INTEGER NOT NULL,\n"
389 " multicast_flags INTEGER NOT NULL,\n"
390 " psycstore_flags INTEGER NOT NULL,\n"
392 " PRIMARY KEY (channel_id, fragment_id),\n"
393 " UNIQUE (channel_id, message_id, fragment_offset)\n"
396 sql_exec (plugin->dbh,
397 "CREATE TABLE IF NOT EXISTS state (\n"
398 " channel_id INTEGER NOT NULL REFERENCES channels(id),\n"
399 " name TEXT NOT NULL,\n"
400 " value_current BLOB,\n"
401 " value_signed BLOB,\n"
402 " PRIMARY KEY (channel_id, name)\n"
405 sql_exec (plugin->dbh,
406 "CREATE TABLE IF NOT EXISTS state_sync (\n"
407 " channel_id INTEGER NOT NULL REFERENCES channels(id),\n"
408 " name TEXT NOT NULL,\n"
410 " PRIMARY KEY (channel_id, name)\n"
413 /* Prepare statements */
415 sql_prepare (plugin->dbh, "BEGIN;", &plugin->transaction_begin);
417 sql_prepare (plugin->dbh, "COMMIT;", &plugin->transaction_commit);
419 sql_prepare (plugin->dbh, "ROLLBACK;", &plugin->transaction_rollback);
421 sql_prepare (plugin->dbh,
422 "INSERT OR IGNORE INTO channels (pub_key) VALUES (?);",
423 &plugin->insert_channel_key);
425 sql_prepare (plugin->dbh,
426 "INSERT OR IGNORE INTO slaves (pub_key) VALUES (?);",
427 &plugin->insert_slave_key);
429 sql_prepare (plugin->dbh,
430 "INSERT INTO membership\n"
431 " (channel_id, slave_id, did_join, announced_at,\n"
432 " effective_since, group_generation)\n"
433 "VALUES ((SELECT id FROM channels WHERE pub_key = ?),\n"
434 " (SELECT id FROM slaves WHERE pub_key = ?),\n"
436 &plugin->insert_membership);
438 sql_prepare (plugin->dbh,
439 "SELECT did_join FROM membership\n"
440 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
441 " AND slave_id = (SELECT id FROM slaves WHERE pub_key = ?)\n"
442 " AND effective_since <= ? AND did_join = 1\n"
443 "ORDER BY announced_at DESC LIMIT 1;",
444 &plugin->select_membership);
446 sql_prepare (plugin->dbh,
447 "INSERT OR IGNORE INTO messages\n"
448 " (channel_id, hop_counter, signature, purpose,\n"
449 " fragment_id, fragment_offset, message_id,\n"
450 " group_generation, multicast_flags, psycstore_flags, data)\n"
451 "VALUES ((SELECT id FROM channels WHERE pub_key = ?),\n"
452 " ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);",
453 &plugin->insert_fragment);
455 sql_prepare (plugin->dbh,
457 "SET psycstore_flags = psycstore_flags | ?\n"
458 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
459 " AND message_id = ? AND fragment_offset = 0;",
460 &plugin->update_message_flags);
462 sql_prepare (plugin->dbh,
463 "SELECT hop_counter, signature, purpose, fragment_id,\n"
464 " fragment_offset, message_id, group_generation,\n"
465 " multicast_flags, psycstore_flags, data\n"
467 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
468 " AND ? <= fragment_id AND fragment_id <= ?;",
469 &plugin->select_fragments);
471 sql_prepare (plugin->dbh,
472 "SELECT hop_counter, signature, purpose, fragment_id,\n"
473 " fragment_offset, message_id, group_generation,\n"
474 " multicast_flags, psycstore_flags, data\n"
476 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
477 " AND ? <= message_id AND message_id <= ?;",
478 &plugin->select_messages);
480 sql_prepare (plugin->dbh,
482 "(SELECT hop_counter, signature, purpose, fragment_id,\n"
483 " fragment_offset, message_id, group_generation,\n"
484 " multicast_flags, psycstore_flags, data\n"
486 " WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
487 " ORDER BY fragment_id DESC\n"
489 "ORDER BY fragment_id;",
490 &plugin->select_latest_fragments);
492 sql_prepare (plugin->dbh,
493 "SELECT hop_counter, signature, purpose, fragment_id,\n"
494 " fragment_offset, message_id, group_generation,\n"
495 " multicast_flags, psycstore_flags, data\n"
497 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
498 " AND message_id IN\n"
499 " (SELECT message_id\n"
501 " WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
502 " ORDER BY message_id\n"
504 "ORDER BY fragment_id;",
505 &plugin->select_latest_messages);
507 sql_prepare (plugin->dbh,
508 "SELECT hop_counter, signature, purpose, fragment_id,\n"
509 " fragment_offset, message_id, group_generation,\n"
510 " multicast_flags, psycstore_flags, data\n"
512 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
513 " AND message_id = ? AND fragment_offset = ?;",
514 &plugin->select_message_fragment);
516 sql_prepare (plugin->dbh,
517 "SELECT fragment_id, message_id, group_generation\n"
519 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
520 "ORDER BY fragment_id DESC LIMIT 1;",
521 &plugin->select_counters_message);
523 sql_prepare (plugin->dbh,
524 "SELECT max_state_message_id\n"
526 "WHERE pub_key = ? AND max_state_message_id IS NOT NULL;",
527 &plugin->select_counters_state);
529 sql_prepare (plugin->dbh,
531 "SET max_state_message_id = ?\n"
532 "WHERE pub_key = ?;",
533 &plugin->update_max_state_message_id);
535 sql_prepare (plugin->dbh,
537 "SET state_hash_message_id = ?\n"
538 "WHERE pub_key = ?;",
539 &plugin->update_state_hash_message_id);
541 sql_prepare (plugin->dbh,
543 "FROM channels AS c\n"
544 "LEFT JOIN messages AS m\n"
545 "ON c.id = m.channel_id\n"
546 "WHERE c.pub_key = ?\n"
547 " AND ((? < c.state_hash_message_id AND c.state_hash_message_id < ?)\n"
548 " OR (m.message_id = ? AND m.psycstore_flags & ?))\n"
550 &plugin->select_message_state_delta);
552 sql_prepare (plugin->dbh,
553 "INSERT OR REPLACE INTO state\n"
554 " (channel_id, name, value_current, value_signed)\n"
555 "SELECT new.channel_id, new.name,\n"
556 " new.value_current, old.value_signed\n"
557 "FROM (SELECT (SELECT id FROM channels WHERE pub_key = ?)\n"
559 " ? AS name, ? AS value_current) AS new\n"
560 "LEFT JOIN (SELECT channel_id, name, value_signed\n"
561 " FROM state) AS old\n"
562 "ON new.channel_id = old.channel_id AND new.name = old.name;",
563 &plugin->insert_state_current);
565 sql_prepare (plugin->dbh,
566 "DELETE FROM state\n"
567 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
568 " AND (value_current IS NULL OR length(value_current) = 0)\n"
569 " AND (value_signed IS NULL OR length(value_signed) = 0);",
570 &plugin->delete_state_empty);
572 sql_prepare (plugin->dbh,
574 "SET value_signed = value_current\n"
575 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
576 &plugin->update_state_signed);
578 sql_prepare (plugin->dbh,
579 "DELETE FROM state\n"
580 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
581 &plugin->delete_state);
583 sql_prepare (plugin->dbh,
584 "INSERT INTO state_sync (channel_id, name, value)\n"
585 "VALUES ((SELECT id FROM channels WHERE pub_key = ?), ?, ?);",
586 &plugin->insert_state_sync);
588 sql_prepare (plugin->dbh,
589 "INSERT INTO state\n"
590 " (channel_id, name, value_current, value_signed)\n"
591 "SELECT channel_id, name, value, value\n"
593 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
594 &plugin->insert_state_from_sync);
596 sql_prepare (plugin->dbh,
597 "DELETE FROM state_sync\n"
598 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
599 &plugin->delete_state_sync);
601 sql_prepare (plugin->dbh,
602 "SELECT value_current\n"
604 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
606 &plugin->select_state_one);
608 sql_prepare (plugin->dbh,
609 "SELECT name, value_current\n"
611 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
612 " AND (name = ? OR name LIKE ?);",
613 &plugin->select_state_prefix);
615 sql_prepare (plugin->dbh,
616 "SELECT name, value_signed\n"
618 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)"
619 " AND value_signed IS NOT NULL;",
620 &plugin->select_state_signed);
627 * Shutdown database connection and associate data
629 * @param plugin the plugin context (state for this module)
632 database_shutdown (struct Plugin *plugin)
636 while (NULL != (stmt = sqlite3_next_stmt (plugin->dbh, NULL)))
638 result = sqlite3_finalize (stmt);
639 if (SQLITE_OK != result)
640 LOG (GNUNET_ERROR_TYPE_WARNING,
641 "Failed to close statement %p: %d\n", stmt, result);
643 if (SQLITE_OK != sqlite3_close (plugin->dbh))
644 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR, "sqlite3_close");
646 GNUNET_free_non_null (plugin->fn);
650 * Execute a prepared statement with a @a channel_key argument.
652 * @param plugin Plugin handle.
653 * @param stmt Statement to execute.
654 * @param channel_key Public key of the channel.
656 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
659 exec_channel (struct Plugin *plugin, sqlite3_stmt *stmt,
660 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
662 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
663 sizeof (*channel_key), SQLITE_STATIC))
665 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
668 else if (SQLITE_DONE != sqlite3_step (stmt))
670 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
674 if (SQLITE_OK != sqlite3_reset (stmt))
676 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
678 return GNUNET_SYSERR;
685 * Begin a transaction.
688 transaction_begin (struct Plugin *plugin, enum Transactions transaction)
690 sqlite3_stmt *stmt = plugin->transaction_begin;
692 if (SQLITE_DONE != sqlite3_step (stmt))
694 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
697 if (SQLITE_OK != sqlite3_reset (stmt))
699 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
701 return GNUNET_SYSERR;
704 plugin->transaction = transaction;
710 * Commit current transaction.
713 transaction_commit (struct Plugin *plugin)
715 sqlite3_stmt *stmt = plugin->transaction_commit;
717 if (SQLITE_DONE != sqlite3_step (stmt))
719 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
722 if (SQLITE_OK != sqlite3_reset (stmt))
724 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
726 return GNUNET_SYSERR;
729 plugin->transaction = TRANSACTION_NONE;
735 * Roll back current transaction.
738 transaction_rollback (struct Plugin *plugin)
740 sqlite3_stmt *stmt = plugin->transaction_rollback;
742 if (SQLITE_DONE != sqlite3_step (stmt))
744 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
747 if (SQLITE_OK != sqlite3_reset (stmt))
749 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
751 return GNUNET_SYSERR;
753 plugin->transaction = TRANSACTION_NONE;
759 channel_key_store (struct Plugin *plugin,
760 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
762 sqlite3_stmt *stmt = plugin->insert_channel_key;
764 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
765 sizeof (*channel_key), SQLITE_STATIC))
767 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
770 else if (SQLITE_DONE != sqlite3_step (stmt))
772 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
776 if (SQLITE_OK != sqlite3_reset (stmt))
778 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
780 return GNUNET_SYSERR;
788 slave_key_store (struct Plugin *plugin,
789 const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key)
791 sqlite3_stmt *stmt = plugin->insert_slave_key;
793 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, slave_key,
794 sizeof (*slave_key), SQLITE_STATIC))
796 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
799 else if (SQLITE_DONE != sqlite3_step (stmt))
801 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
806 if (SQLITE_OK != sqlite3_reset (stmt))
808 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
810 return GNUNET_SYSERR;
818 * Store join/leave events for a PSYC channel in order to be able to answer
819 * membership test queries later.
821 * @see GNUNET_PSYCSTORE_membership_store()
823 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
826 membership_store (void *cls,
827 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
828 const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
830 uint64_t announced_at,
831 uint64_t effective_since,
832 uint64_t group_generation)
834 struct Plugin *plugin = cls;
835 sqlite3_stmt *stmt = plugin->insert_membership;
837 GNUNET_assert (TRANSACTION_NONE == plugin->transaction);
839 if (announced_at > INT64_MAX ||
840 effective_since > INT64_MAX ||
841 group_generation > INT64_MAX)
844 return GNUNET_SYSERR;
847 if (GNUNET_OK != channel_key_store (plugin, channel_key)
848 || GNUNET_OK != slave_key_store (plugin, slave_key))
849 return GNUNET_SYSERR;
851 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
852 sizeof (*channel_key), SQLITE_STATIC)
853 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, slave_key,
854 sizeof (*slave_key), SQLITE_STATIC)
855 || SQLITE_OK != sqlite3_bind_int (stmt, 3, did_join)
856 || SQLITE_OK != sqlite3_bind_int64 (stmt, 4, announced_at)
857 || SQLITE_OK != sqlite3_bind_int64 (stmt, 5, effective_since)
858 || SQLITE_OK != sqlite3_bind_int64 (stmt, 6, group_generation))
860 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
863 else if (SQLITE_DONE != sqlite3_step (stmt))
865 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
869 if (SQLITE_OK != sqlite3_reset (stmt))
871 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
873 return GNUNET_SYSERR;
880 * Test if a member was admitted to the channel at the given message ID.
882 * @see GNUNET_PSYCSTORE_membership_test()
884 * @return #GNUNET_YES if the member was admitted, #GNUNET_NO if not,
885 * #GNUNET_SYSERR if there was en error.
888 membership_test (void *cls,
889 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
890 const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
893 struct Plugin *plugin = cls;
894 sqlite3_stmt *stmt = plugin->select_membership;
895 int ret = GNUNET_SYSERR;
897 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
898 sizeof (*channel_key), SQLITE_STATIC)
899 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, slave_key,
900 sizeof (*slave_key), SQLITE_STATIC)
901 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, message_id))
903 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
908 switch (sqlite3_step (stmt))
918 if (SQLITE_OK != sqlite3_reset (stmt))
920 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
928 * Store a message fragment sent to a channel.
930 * @see GNUNET_PSYCSTORE_fragment_store()
932 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
935 fragment_store (void *cls,
936 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
937 const struct GNUNET_MULTICAST_MessageHeader *msg,
938 uint32_t psycstore_flags)
940 struct Plugin *plugin = cls;
941 sqlite3_stmt *stmt = plugin->insert_fragment;
943 GNUNET_assert (TRANSACTION_NONE == plugin->transaction);
945 uint64_t fragment_id = GNUNET_ntohll (msg->fragment_id);
946 uint64_t fragment_offset = GNUNET_ntohll (msg->fragment_offset);
947 uint64_t message_id = GNUNET_ntohll (msg->message_id);
948 uint64_t group_generation = GNUNET_ntohll (msg->group_generation);
950 if (fragment_id > INT64_MAX || fragment_offset > INT64_MAX ||
951 message_id > INT64_MAX || group_generation > INT64_MAX)
953 LOG (GNUNET_ERROR_TYPE_ERROR,
954 "Tried to store fragment with a field > INT64_MAX: "
955 "%lu, %lu, %lu, %lu\n", fragment_id, fragment_offset,
956 message_id, group_generation);
958 return GNUNET_SYSERR;
961 if (GNUNET_OK != channel_key_store (plugin, channel_key))
962 return GNUNET_SYSERR;
964 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
965 sizeof (*channel_key), SQLITE_STATIC)
966 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, ntohl (msg->hop_counter) )
967 || SQLITE_OK != sqlite3_bind_blob (stmt, 3, (const void *) &msg->signature,
968 sizeof (msg->signature), SQLITE_STATIC)
969 || SQLITE_OK != sqlite3_bind_blob (stmt, 4, (const void *) &msg->purpose,
970 sizeof (msg->purpose), SQLITE_STATIC)
971 || SQLITE_OK != sqlite3_bind_int64 (stmt, 5, fragment_id)
972 || SQLITE_OK != sqlite3_bind_int64 (stmt, 6, fragment_offset)
973 || SQLITE_OK != sqlite3_bind_int64 (stmt, 7, message_id)
974 || SQLITE_OK != sqlite3_bind_int64 (stmt, 8, group_generation)
975 || SQLITE_OK != sqlite3_bind_int64 (stmt, 9, ntohl (msg->flags))
976 || SQLITE_OK != sqlite3_bind_int64 (stmt, 10, psycstore_flags)
977 || SQLITE_OK != sqlite3_bind_blob (stmt, 11, (const void *) &msg[1],
978 ntohs (msg->header.size)
979 - sizeof (*msg), SQLITE_STATIC))
981 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
984 else if (SQLITE_DONE != sqlite3_step (stmt))
986 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
990 if (SQLITE_OK != sqlite3_reset (stmt))
992 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
994 return GNUNET_SYSERR;
1001 * Set additional flags for a given message.
1003 * They are OR'd with any existing flags set.
1005 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1008 message_add_flags (void *cls,
1009 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1010 uint64_t message_id,
1011 uint64_t psycstore_flags)
1013 struct Plugin *plugin = cls;
1014 sqlite3_stmt *stmt = plugin->update_message_flags;
1015 int ret = GNUNET_SYSERR;
1017 if (SQLITE_OK != sqlite3_bind_int64 (stmt, 1, psycstore_flags)
1018 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, channel_key,
1019 sizeof (*channel_key), SQLITE_STATIC)
1020 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, message_id))
1022 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1027 switch (sqlite3_step (stmt))
1030 ret = sqlite3_total_changes (plugin->dbh) > 0 ? GNUNET_OK : GNUNET_NO;
1033 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1038 if (SQLITE_OK != sqlite3_reset (stmt))
1040 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1042 return GNUNET_SYSERR;
1049 fragment_row (sqlite3_stmt *stmt, GNUNET_PSYCSTORE_FragmentCallback cb,
1052 int data_size = sqlite3_column_bytes (stmt, 9);
1053 struct GNUNET_MULTICAST_MessageHeader *msg
1054 = GNUNET_malloc (sizeof (*msg) + data_size);
1056 msg->header.size = htons (sizeof (*msg) + data_size);
1057 msg->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE);
1058 msg->hop_counter = htonl ((uint32_t) sqlite3_column_int64 (stmt, 0));
1059 memcpy (&msg->signature,
1060 sqlite3_column_blob (stmt, 1),
1061 sqlite3_column_bytes (stmt, 1));
1062 memcpy (&msg->purpose,
1063 sqlite3_column_blob (stmt, 2),
1064 sqlite3_column_bytes (stmt, 2));
1065 msg->fragment_id = GNUNET_htonll (sqlite3_column_int64 (stmt, 3));
1066 msg->fragment_offset = GNUNET_htonll (sqlite3_column_int64 (stmt, 4));
1067 msg->message_id = GNUNET_htonll (sqlite3_column_int64 (stmt, 5));
1068 msg->group_generation = GNUNET_htonll (sqlite3_column_int64 (stmt, 6));
1069 msg->flags = htonl (sqlite3_column_int64 (stmt, 7));
1070 memcpy (&msg[1], sqlite3_column_blob (stmt, 9), data_size);
1072 return cb (cb_cls, (void *) msg, sqlite3_column_int64 (stmt, 8));
1077 fragment_select (struct Plugin *plugin, sqlite3_stmt *stmt,
1078 uint64_t *returned_fragments,
1079 GNUNET_PSYCSTORE_FragmentCallback cb, void *cb_cls)
1081 int ret = GNUNET_SYSERR;
1086 sql_ret = sqlite3_step (stmt);
1090 if (ret != GNUNET_OK)
1094 ret = fragment_row (stmt, cb, cb_cls);
1095 (*returned_fragments)++;
1096 if (ret != GNUNET_YES)
1097 sql_ret = SQLITE_DONE;
1100 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1104 while (sql_ret == SQLITE_ROW);
1110 * Retrieve a message fragment range by fragment ID.
1112 * @see GNUNET_PSYCSTORE_fragment_get()
1114 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1117 fragment_get (void *cls,
1118 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1119 uint64_t first_fragment_id,
1120 uint64_t last_fragment_id,
1121 uint64_t *returned_fragments,
1122 GNUNET_PSYCSTORE_FragmentCallback cb,
1125 struct Plugin *plugin = cls;
1126 sqlite3_stmt *stmt = plugin->select_fragments;
1127 int ret = GNUNET_SYSERR;
1128 *returned_fragments = 0;
1130 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1131 sizeof (*channel_key),
1133 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, first_fragment_id)
1134 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, last_fragment_id))
1136 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1141 ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls);
1144 if (SQLITE_OK != sqlite3_reset (stmt))
1146 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1155 * Retrieve a message fragment range by fragment ID.
1157 * @see GNUNET_PSYCSTORE_fragment_get_latest()
1159 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1162 fragment_get_latest (void *cls,
1163 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1164 uint64_t fragment_limit,
1165 uint64_t *returned_fragments,
1166 GNUNET_PSYCSTORE_FragmentCallback cb,
1169 struct Plugin *plugin = cls;
1170 sqlite3_stmt *stmt = plugin->select_latest_fragments;
1171 int ret = GNUNET_SYSERR;
1172 *returned_fragments = 0;
1174 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1175 sizeof (*channel_key),
1177 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, fragment_limit))
1179 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1184 ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls);
1187 if (SQLITE_OK != sqlite3_reset (stmt))
1189 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1198 * Retrieve all fragments of a message ID range.
1200 * @see GNUNET_PSYCSTORE_message_get()
1202 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1205 message_get (void *cls,
1206 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1207 uint64_t first_message_id,
1208 uint64_t last_message_id,
1209 uint64_t *returned_fragments,
1210 GNUNET_PSYCSTORE_FragmentCallback cb,
1213 struct Plugin *plugin = cls;
1214 sqlite3_stmt *stmt = plugin->select_messages;
1215 int ret = GNUNET_SYSERR;
1216 *returned_fragments = 0;
1218 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1219 sizeof (*channel_key),
1221 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, first_message_id)
1222 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, last_message_id))
1224 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1229 ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls);
1232 if (SQLITE_OK != sqlite3_reset (stmt))
1234 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1243 * Retrieve all fragments of the latest messages.
1245 * @see GNUNET_PSYCSTORE_message_get_latest()
1247 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1250 message_get_latest (void *cls,
1251 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1252 uint64_t message_limit,
1253 uint64_t *returned_fragments,
1254 GNUNET_PSYCSTORE_FragmentCallback cb,
1257 struct Plugin *plugin = cls;
1258 sqlite3_stmt *stmt = plugin->select_latest_messages;
1259 int ret = GNUNET_SYSERR;
1260 *returned_fragments = 0;
1262 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1263 sizeof (*channel_key),
1265 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, channel_key,
1266 sizeof (*channel_key),
1268 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, message_limit))
1270 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1275 ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls);
1278 if (SQLITE_OK != sqlite3_reset (stmt))
1280 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1289 * Retrieve a fragment of message specified by its message ID and fragment
1292 * @see GNUNET_PSYCSTORE_message_get_fragment()
1294 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1297 message_get_fragment (void *cls,
1298 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1299 uint64_t message_id,
1300 uint64_t fragment_offset,
1301 GNUNET_PSYCSTORE_FragmentCallback cb,
1304 struct Plugin *plugin = cls;
1305 sqlite3_stmt *stmt = plugin->select_message_fragment;
1306 int ret = GNUNET_SYSERR;
1308 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1309 sizeof (*channel_key),
1311 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, message_id)
1312 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, fragment_offset))
1314 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1319 switch (sqlite3_step (stmt))
1325 ret = fragment_row (stmt, cb, cb_cls);
1328 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1333 if (SQLITE_OK != sqlite3_reset (stmt))
1335 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1343 * Retrieve the max. values of message counters for a channel.
1345 * @see GNUNET_PSYCSTORE_counters_get()
1347 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1350 counters_message_get (void *cls,
1351 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1352 uint64_t *max_fragment_id,
1353 uint64_t *max_message_id,
1354 uint64_t *max_group_generation)
1356 struct Plugin *plugin = cls;
1357 sqlite3_stmt *stmt = plugin->select_counters_message;
1358 int ret = GNUNET_SYSERR;
1360 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1361 sizeof (*channel_key),
1364 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1369 switch (sqlite3_step (stmt))
1375 *max_fragment_id = sqlite3_column_int64 (stmt, 0);
1376 *max_message_id = sqlite3_column_int64 (stmt, 1);
1377 *max_group_generation = sqlite3_column_int64 (stmt, 2);
1381 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1386 if (SQLITE_OK != sqlite3_reset (stmt))
1388 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1396 * Retrieve the max. values of state counters for a channel.
1398 * @see GNUNET_PSYCSTORE_counters_get()
1400 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1403 counters_state_get (void *cls,
1404 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1405 uint64_t *max_state_message_id)
1407 struct Plugin *plugin = cls;
1408 sqlite3_stmt *stmt = plugin->select_counters_state;
1409 int ret = GNUNET_SYSERR;
1411 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1412 sizeof (*channel_key),
1415 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1420 switch (sqlite3_step (stmt))
1426 *max_state_message_id = sqlite3_column_int64 (stmt, 0);
1430 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1435 if (SQLITE_OK != sqlite3_reset (stmt))
1437 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1446 * Set a state variable to the given value.
1448 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1451 state_set (struct Plugin *plugin, sqlite3_stmt *stmt,
1452 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1453 const char *name, const void *value, size_t value_size)
1455 int ret = GNUNET_SYSERR;
1457 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1458 sizeof (*channel_key), SQLITE_STATIC)
1459 || SQLITE_OK != sqlite3_bind_text (stmt, 2, name, -1, SQLITE_STATIC)
1460 || SQLITE_OK != sqlite3_bind_blob (stmt, 3, value, value_size,
1463 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1468 switch (sqlite3_step (stmt))
1471 ret = 0 < sqlite3_total_changes (plugin->dbh) ? GNUNET_OK : GNUNET_NO;
1474 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1479 if (SQLITE_OK != sqlite3_reset (stmt))
1481 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1483 return GNUNET_SYSERR;
1491 update_message_id (struct Plugin *plugin, sqlite3_stmt *stmt,
1492 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1493 uint64_t message_id)
1495 if (SQLITE_OK != sqlite3_bind_int64 (stmt, 1, message_id)
1496 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, channel_key,
1497 sizeof (*channel_key), SQLITE_STATIC))
1499 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1502 else if (SQLITE_DONE != sqlite3_step (stmt))
1504 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1507 if (SQLITE_OK != sqlite3_reset (stmt))
1509 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1511 return GNUNET_SYSERR;
1518 * Begin modifying current state.
1521 state_modify_begin (void *cls,
1522 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1523 uint64_t message_id, uint64_t state_delta)
1525 struct Plugin *plugin = cls;
1526 sqlite3_stmt *stmt = plugin->select_message_state_delta;
1528 if (state_delta > 0)
1530 int ret = GNUNET_SYSERR;
1531 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1532 sizeof (*channel_key), SQLITE_STATIC)
1533 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2,
1534 message_id - state_delta)
1535 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3,
1537 || SQLITE_OK != sqlite3_bind_int64 (stmt, 4,
1538 message_id - state_delta)
1539 || SQLITE_OK != sqlite3_bind_int64 (stmt, 5,
1540 GNUNET_PSYCSTORE_MESSAGE_STATE_APPLIED))
1542 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1547 switch (sqlite3_step (stmt))
1556 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1560 if (SQLITE_OK != sqlite3_reset (stmt))
1562 ret = GNUNET_SYSERR;
1563 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1566 if (GNUNET_OK != ret)
1570 if (TRANSACTION_NONE != plugin->transaction)
1571 if (GNUNET_OK != transaction_rollback (plugin))
1572 return GNUNET_SYSERR;
1574 return transaction_begin (plugin, TRANSACTION_STATE_MODIFY);
1579 * Set the current value of state variable.
1581 * @see GNUNET_PSYCSTORE_state_modify()
1583 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1586 state_modify_set (void *cls,
1587 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1588 const char *name, const void *value, size_t value_size)
1590 struct Plugin *plugin = cls;
1591 GNUNET_assert (TRANSACTION_STATE_MODIFY == plugin->transaction);
1593 return state_set (plugin, plugin->insert_state_current, channel_key,
1594 name, value, value_size);
1600 * End modifying current state.
1603 state_modify_end (void *cls,
1604 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1605 uint64_t message_id)
1607 struct Plugin *plugin = cls;
1608 GNUNET_assert (TRANSACTION_STATE_MODIFY == plugin->transaction);
1611 GNUNET_OK == exec_channel (plugin, plugin->delete_state_empty, channel_key)
1612 && GNUNET_OK == update_message_id (plugin,
1613 plugin->update_max_state_message_id,
1614 channel_key, message_id)
1615 && GNUNET_OK == transaction_commit (plugin)
1616 ? GNUNET_OK : GNUNET_SYSERR;
1621 * Begin state synchronization.
1624 state_sync_begin (void *cls,
1625 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
1627 struct Plugin *plugin = cls;
1628 return exec_channel (plugin, plugin->delete_state_sync, channel_key);
1633 * Set the current value of state variable.
1635 * @see GNUNET_PSYCSTORE_state_modify()
1637 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1640 state_sync_set (void *cls,
1641 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1642 const char *name, const void *value, size_t value_size)
1644 struct Plugin *plugin = cls;
1645 return state_set (cls, plugin->insert_state_sync, channel_key,
1646 name, value, value_size);
1651 * End modifying current state.
1654 state_sync_end (void *cls,
1655 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1656 uint64_t message_id)
1658 struct Plugin *plugin = cls;
1659 int ret = GNUNET_SYSERR;
1661 GNUNET_OK == transaction_begin (plugin, TRANSACTION_NONE)
1662 && GNUNET_OK == exec_channel (plugin, plugin->delete_state, channel_key)
1663 && GNUNET_OK == exec_channel (plugin, plugin->insert_state_from_sync,
1665 && GNUNET_OK == exec_channel (plugin, plugin->delete_state_sync,
1667 && GNUNET_OK == update_message_id (plugin,
1668 plugin->update_state_hash_message_id,
1669 channel_key, message_id)
1670 && GNUNET_OK == transaction_commit (plugin)
1672 : transaction_rollback (plugin);
1678 * Reset the state of a channel.
1680 * @see GNUNET_PSYCSTORE_state_reset()
1682 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1685 state_reset (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
1687 struct Plugin *plugin = cls;
1688 return exec_channel (plugin, plugin->delete_state, channel_key);
1693 * Update signed values of state variables in the state store.
1695 * @see GNUNET_PSYCSTORE_state_hash_update()
1697 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1700 state_update_signed (void *cls,
1701 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
1703 struct Plugin *plugin = cls;
1704 return exec_channel (plugin, plugin->update_state_signed, channel_key);
1709 * Retrieve a state variable by name.
1711 * @see GNUNET_PSYCSTORE_state_get()
1713 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1716 state_get (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1717 const char *name, GNUNET_PSYCSTORE_StateCallback cb, void *cb_cls)
1719 struct Plugin *plugin = cls;
1720 int ret = GNUNET_SYSERR;
1722 sqlite3_stmt *stmt = plugin->select_state_one;
1724 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1725 sizeof (*channel_key),
1727 || SQLITE_OK != sqlite3_bind_text (stmt, 2, name, -1, SQLITE_STATIC))
1729 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1734 switch (sqlite3_step (stmt))
1740 ret = cb (cb_cls, name, sqlite3_column_blob (stmt, 0),
1741 sqlite3_column_bytes (stmt, 0));
1744 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1749 if (SQLITE_OK != sqlite3_reset (stmt))
1751 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1760 * Retrieve all state variables for a channel with the given prefix.
1762 * @see GNUNET_PSYCSTORE_state_get_prefix()
1764 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1767 state_get_prefix (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1768 const char *name, GNUNET_PSYCSTORE_StateCallback cb,
1771 struct Plugin *plugin = cls;
1772 int ret = GNUNET_SYSERR;
1774 sqlite3_stmt *stmt = plugin->select_state_prefix;
1775 size_t name_len = strlen (name);
1776 char *name_prefix = GNUNET_malloc (name_len + 2);
1777 memcpy (name_prefix, name, name_len);
1778 memcpy (name_prefix + name_len, "_%", 2);
1780 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1781 sizeof (*channel_key), SQLITE_STATIC)
1782 || SQLITE_OK != sqlite3_bind_text (stmt, 2, name, name_len, SQLITE_STATIC)
1783 || SQLITE_OK != sqlite3_bind_text (stmt, 3, name_prefix, name_len + 2,
1786 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1794 sql_ret = sqlite3_step (stmt);
1798 if (ret != GNUNET_OK)
1802 ret = cb (cb_cls, (const char *) sqlite3_column_text (stmt, 0),
1803 sqlite3_column_blob (stmt, 1),
1804 sqlite3_column_bytes (stmt, 1));
1805 if (ret != GNUNET_YES)
1806 sql_ret = SQLITE_DONE;
1809 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1813 while (sql_ret == SQLITE_ROW);
1816 if (SQLITE_OK != sqlite3_reset (stmt))
1818 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1827 * Retrieve all signed state variables for a channel.
1829 * @see GNUNET_PSYCSTORE_state_get_signed()
1831 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1834 state_get_signed (void *cls,
1835 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1836 GNUNET_PSYCSTORE_StateCallback cb, void *cb_cls)
1838 struct Plugin *plugin = cls;
1839 int ret = GNUNET_SYSERR;
1841 sqlite3_stmt *stmt = plugin->select_state_signed;
1843 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1844 sizeof (*channel_key), SQLITE_STATIC))
1846 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1854 sql_ret = sqlite3_step (stmt);
1858 if (ret != GNUNET_OK)
1862 ret = cb (cb_cls, (const char *) sqlite3_column_text (stmt, 0),
1863 sqlite3_column_blob (stmt, 1),
1864 sqlite3_column_bytes (stmt, 1));
1865 if (ret != GNUNET_YES)
1866 sql_ret = SQLITE_DONE;
1869 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1873 while (sql_ret == SQLITE_ROW);
1876 if (SQLITE_OK != sqlite3_reset (stmt))
1878 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1887 * Entry point for the plugin.
1889 * @param cls The struct GNUNET_CONFIGURATION_Handle.
1890 * @return NULL on error, otherwise the plugin context
1893 libgnunet_plugin_psycstore_sqlite_init (void *cls)
1895 static struct Plugin plugin;
1896 const struct GNUNET_CONFIGURATION_Handle *cfg = cls;
1897 struct GNUNET_PSYCSTORE_PluginFunctions *api;
1899 if (NULL != plugin.cfg)
1900 return NULL; /* can only initialize once! */
1901 memset (&plugin, 0, sizeof (struct Plugin));
1903 if (GNUNET_OK != database_setup (&plugin))
1905 database_shutdown (&plugin);
1908 api = GNUNET_new (struct GNUNET_PSYCSTORE_PluginFunctions);
1910 api->membership_store = &membership_store;
1911 api->membership_test = &membership_test;
1912 api->fragment_store = &fragment_store;
1913 api->message_add_flags = &message_add_flags;
1914 api->fragment_get = &fragment_get;
1915 api->fragment_get_latest = &fragment_get_latest;
1916 api->message_get = &message_get;
1917 api->message_get_latest = &message_get_latest;
1918 api->message_get_fragment = &message_get_fragment;
1919 api->counters_message_get = &counters_message_get;
1920 api->counters_state_get = &counters_state_get;
1921 api->state_modify_begin = &state_modify_begin;
1922 api->state_modify_set = &state_modify_set;
1923 api->state_modify_end = &state_modify_end;
1924 api->state_sync_begin = &state_sync_begin;
1925 api->state_sync_set = &state_sync_set;
1926 api->state_sync_end = &state_sync_end;
1927 api->state_reset = &state_reset;
1928 api->state_update_signed = &state_update_signed;
1929 api->state_get = &state_get;
1930 api->state_get_prefix = &state_get_prefix;
1931 api->state_get_signed = &state_get_signed;
1933 LOG (GNUNET_ERROR_TYPE_INFO, _("SQLite database running\n"));
1939 * Exit point from the plugin.
1941 * @param cls The plugin context (as returned by "init")
1942 * @return Always NULL
1945 libgnunet_plugin_psycstore_sqlite_done (void *cls)
1947 struct GNUNET_PSYCSTORE_PluginFunctions *api = cls;
1948 struct Plugin *plugin = api->cls;
1950 database_shutdown (plugin);
1953 LOG (GNUNET_ERROR_TYPE_DEBUG, "SQLite plugin is finished\n");
1957 /* end of plugin_psycstore_sqlite.c */