2 * This file is part of GNUnet
3 * Copyright (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., 51 Franklin Street, Fifth Floor,
18 * Boston, MA 02110-1301, 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 "gnunet_env_lib.h"
39 #include "psycstore.h"
43 * After how many ms "busy" should a DB operation fail for good? A
44 * low value makes sure that we are more responsive to requests
45 * (especially PUTs). A high value guarantees a higher success rate
46 * (SELECTs in iterate can take several seconds despite LIMIT=1).
48 * The default value of 1s should ensure that users do not experience
49 * huge latencies while at the same time allowing operations to
50 * succeed with reasonable probability.
52 #define BUSY_TIMEOUT_MS 1000
54 #define DEBUG_PSYCSTORE GNUNET_EXTRA_LOGGING
57 * Log an error message at log-level 'level' that indicates
58 * a failure of the command 'cmd' on file 'filename'
59 * with the message given by strerror(errno).
61 #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)
63 #define LOG(kind,...) GNUNET_log_from (kind, "psycstore-sqlite", __VA_ARGS__)
67 TRANSACTION_STATE_MODIFY,
68 TRANSACTION_STATE_SYNC,
72 * Context for all functions in this plugin.
77 const struct GNUNET_CONFIGURATION_Handle *cfg;
85 * Native SQLite database handle.
90 * Current transaction.
92 enum Transactions transaction;
94 sqlite3_stmt *transaction_begin;
96 sqlite3_stmt *transaction_commit;
98 sqlite3_stmt *transaction_rollback;
101 * Precompiled SQL for channel_key_store()
103 sqlite3_stmt *insert_channel_key;
106 * Precompiled SQL for slave_key_store()
108 sqlite3_stmt *insert_slave_key;
112 * Precompiled SQL for membership_store()
114 sqlite3_stmt *insert_membership;
117 * Precompiled SQL for membership_test()
119 sqlite3_stmt *select_membership;
123 * Precompiled SQL for fragment_store()
125 sqlite3_stmt *insert_fragment;
128 * Precompiled SQL for message_add_flags()
130 sqlite3_stmt *update_message_flags;
133 * Precompiled SQL for fragment_get()
135 sqlite3_stmt *select_fragments;
138 * Precompiled SQL for fragment_get()
140 sqlite3_stmt *select_latest_fragments;
143 * Precompiled SQL for message_get()
145 sqlite3_stmt *select_messages;
148 * Precompiled SQL for message_get()
150 sqlite3_stmt *select_latest_messages;
153 * Precompiled SQL for message_get_fragment()
155 sqlite3_stmt *select_message_fragment;
158 * Precompiled SQL for counters_get_message()
160 sqlite3_stmt *select_counters_message;
163 * Precompiled SQL for counters_get_state()
165 sqlite3_stmt *select_counters_state;
168 * Precompiled SQL for state_modify_end()
170 sqlite3_stmt *update_state_hash_message_id;
173 * Precompiled SQL for state_sync_end()
175 sqlite3_stmt *update_max_state_message_id;
178 * Precompiled SQL for state_modify_op()
180 sqlite3_stmt *insert_state_current;
183 * Precompiled SQL for state_modify_end()
185 sqlite3_stmt *delete_state_empty;
188 * Precompiled SQL for state_set_signed()
190 sqlite3_stmt *update_state_signed;
193 * Precompiled SQL for state_sync()
195 sqlite3_stmt *insert_state_sync;
198 * Precompiled SQL for state_sync()
200 sqlite3_stmt *delete_state;
203 * Precompiled SQL for state_sync()
205 sqlite3_stmt *insert_state_from_sync;
208 * Precompiled SQL for state_sync()
210 sqlite3_stmt *delete_state_sync;
213 * Precompiled SQL for state_get_signed()
215 sqlite3_stmt *select_state_signed;
218 * Precompiled SQL for state_get()
220 sqlite3_stmt *select_state_one;
223 * Precompiled SQL for state_get_prefix()
225 sqlite3_stmt *select_state_prefix;
232 sql_trace (void *cls, const char *sql)
234 LOG (GNUNET_ERROR_TYPE_DEBUG, "SQL query:\n%s\n", sql);
240 * @brief Prepare a SQL statement
242 * @param dbh handle to the database
243 * @param sql SQL statement, UTF-8 encoded
244 * @param stmt set to the prepared statement
245 * @return 0 on success
248 sql_prepare (sqlite3 *dbh, const char *sql, sqlite3_stmt **stmt)
253 result = sqlite3_prepare_v2 (dbh, sql, strlen (sql), stmt,
254 (const char **) &tail);
255 LOG (GNUNET_ERROR_TYPE_DEBUG,
256 "Prepared `%s' / %p: %d\n", sql, *stmt, result);
257 if (result != SQLITE_OK)
258 LOG (GNUNET_ERROR_TYPE_ERROR,
259 _("Error preparing SQL query: %s\n %s\n"),
260 sqlite3_errmsg (dbh), sql);
266 * @brief Prepare a SQL statement
268 * @param dbh handle to the database
269 * @param sql SQL statement, UTF-8 encoded
270 * @return 0 on success
273 sql_exec (sqlite3 *dbh, const char *sql)
277 result = sqlite3_exec (dbh, sql, NULL, NULL, NULL);
278 LOG (GNUNET_ERROR_TYPE_DEBUG,
279 "Executed `%s' / %d\n", sql, result);
280 if (result != SQLITE_OK)
281 LOG (GNUNET_ERROR_TYPE_ERROR,
282 _("Error executing SQL query: %s\n %s\n"),
283 sqlite3_errmsg (dbh), sql);
289 * Initialize the database connections and associated
290 * data structures (create tables and indices
291 * as needed as well).
293 * @param plugin the plugin context (state for this module)
294 * @return GNUNET_OK on success
297 database_setup (struct Plugin *plugin)
302 GNUNET_CONFIGURATION_get_value_filename (plugin->cfg, "psycstore-sqlite",
303 "FILENAME", &filename))
305 GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
306 "psycstore-sqlite", "FILENAME");
307 return GNUNET_SYSERR;
309 if (GNUNET_OK != GNUNET_DISK_file_test (filename))
311 if (GNUNET_OK != GNUNET_DISK_directory_create_for_file (filename))
314 GNUNET_free (filename);
315 return GNUNET_SYSERR;
318 /* filename should be UTF-8-encoded. If it isn't, it's a bug */
319 plugin->fn = filename;
321 /* Open database and precompile statements */
322 if (SQLITE_OK != sqlite3_open (plugin->fn, &plugin->dbh))
324 LOG (GNUNET_ERROR_TYPE_ERROR,
325 _("Unable to initialize SQLite: %s.\n"),
326 sqlite3_errmsg (plugin->dbh));
327 return GNUNET_SYSERR;
331 sqlite3_trace (plugin->dbh, &sql_trace, NULL);
334 sql_exec (plugin->dbh, "PRAGMA temp_store=MEMORY");
335 sql_exec (plugin->dbh, "PRAGMA synchronous=NORMAL");
336 sql_exec (plugin->dbh, "PRAGMA legacy_file_format=OFF");
337 sql_exec (plugin->dbh, "PRAGMA auto_vacuum=INCREMENTAL");
338 sql_exec (plugin->dbh, "PRAGMA encoding=\"UTF-8\"");
339 #if ! DEBUG_PSYCSTORE
340 sql_exec (plugin->dbh, "PRAGMA locking_mode=EXCLUSIVE");
342 sql_exec (plugin->dbh, "PRAGMA page_size=4096");
344 sqlite3_busy_timeout (plugin->dbh, BUSY_TIMEOUT_MS);
348 sql_exec (plugin->dbh,
349 "CREATE TABLE IF NOT EXISTS channels (\n"
350 " id INTEGER PRIMARY KEY,\n"
351 " pub_key BLOB UNIQUE,\n"
352 " max_state_message_id INTEGER,\n" // last applied state message ID
353 " state_hash_message_id INTEGER\n" // last message ID with a state hash
356 sql_exec (plugin->dbh,
357 "CREATE TABLE IF NOT EXISTS slaves (\n"
358 " id INTEGER PRIMARY KEY,\n"
359 " pub_key BLOB UNIQUE\n"
362 sql_exec (plugin->dbh,
363 "CREATE TABLE IF NOT EXISTS membership (\n"
364 " channel_id INTEGER NOT NULL REFERENCES channels(id),\n"
365 " slave_id INTEGER NOT NULL REFERENCES slaves(id),\n"
366 " did_join INTEGER NOT NULL,\n"
367 " announced_at INTEGER NOT NULL,\n"
368 " effective_since INTEGER NOT NULL,\n"
369 " group_generation INTEGER NOT NULL\n"
371 sql_exec (plugin->dbh,
372 "CREATE INDEX IF NOT EXISTS idx_membership_channel_id_slave_id "
373 "ON membership (channel_id, slave_id);");
375 /** @todo messages table: add method_name column */
376 sql_exec (plugin->dbh,
377 "CREATE TABLE IF NOT EXISTS messages (\n"
378 " channel_id INTEGER NOT NULL REFERENCES channels(id),\n"
379 " hop_counter INTEGER NOT NULL,\n"
382 " fragment_id INTEGER NOT NULL,\n"
383 " fragment_offset INTEGER NOT NULL,\n"
384 " message_id INTEGER NOT NULL,\n"
385 " group_generation INTEGER NOT NULL,\n"
386 " multicast_flags INTEGER NOT NULL,\n"
387 " psycstore_flags INTEGER NOT NULL,\n"
389 " PRIMARY KEY (channel_id, fragment_id),\n"
390 " UNIQUE (channel_id, message_id, fragment_offset)\n"
393 sql_exec (plugin->dbh,
394 "CREATE TABLE IF NOT EXISTS state (\n"
395 " channel_id INTEGER NOT NULL REFERENCES channels(id),\n"
396 " name TEXT NOT NULL,\n"
397 " value_current BLOB,\n"
398 " value_signed BLOB,\n"
399 " PRIMARY KEY (channel_id, name)\n"
402 sql_exec (plugin->dbh,
403 "CREATE TABLE IF NOT EXISTS state_sync (\n"
404 " channel_id INTEGER NOT NULL REFERENCES channels(id),\n"
405 " name TEXT NOT NULL,\n"
407 " PRIMARY KEY (channel_id, name)\n"
410 /* Prepare statements */
412 sql_prepare (plugin->dbh, "BEGIN;", &plugin->transaction_begin);
414 sql_prepare (plugin->dbh, "COMMIT;", &plugin->transaction_commit);
416 sql_prepare (plugin->dbh, "ROLLBACK;", &plugin->transaction_rollback);
418 sql_prepare (plugin->dbh,
419 "INSERT OR IGNORE INTO channels (pub_key) VALUES (?);",
420 &plugin->insert_channel_key);
422 sql_prepare (plugin->dbh,
423 "INSERT OR IGNORE INTO slaves (pub_key) VALUES (?);",
424 &plugin->insert_slave_key);
426 sql_prepare (plugin->dbh,
427 "INSERT INTO membership\n"
428 " (channel_id, slave_id, did_join, announced_at,\n"
429 " effective_since, group_generation)\n"
430 "VALUES ((SELECT id FROM channels WHERE pub_key = ?),\n"
431 " (SELECT id FROM slaves WHERE pub_key = ?),\n"
433 &plugin->insert_membership);
435 sql_prepare (plugin->dbh,
436 "SELECT did_join FROM membership\n"
437 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
438 " AND slave_id = (SELECT id FROM slaves WHERE pub_key = ?)\n"
439 " AND effective_since <= ? AND did_join = 1\n"
440 "ORDER BY announced_at DESC LIMIT 1;",
441 &plugin->select_membership);
443 sql_prepare (plugin->dbh,
444 "INSERT OR IGNORE INTO messages\n"
445 " (channel_id, hop_counter, signature, purpose,\n"
446 " fragment_id, fragment_offset, message_id,\n"
447 " group_generation, multicast_flags, psycstore_flags, data)\n"
448 "VALUES ((SELECT id FROM channels WHERE pub_key = ?),\n"
449 " ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);",
450 &plugin->insert_fragment);
452 sql_prepare (plugin->dbh,
454 "SET psycstore_flags = psycstore_flags | ?\n"
455 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
456 " AND message_id = ? AND fragment_offset = 0;",
457 &plugin->update_message_flags);
459 sql_prepare (plugin->dbh,
460 "SELECT hop_counter, signature, purpose, fragment_id,\n"
461 " fragment_offset, message_id, group_generation,\n"
462 " multicast_flags, psycstore_flags, data\n"
464 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
465 " AND ? <= fragment_id AND fragment_id <= ?;",
466 &plugin->select_fragments);
468 /** @todo select_messages: add method_prefix filter */
469 sql_prepare (plugin->dbh,
470 "SELECT hop_counter, signature, purpose, fragment_id,\n"
471 " fragment_offset, message_id, group_generation,\n"
472 " multicast_flags, psycstore_flags, data\n"
474 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
475 " AND ? <= message_id AND message_id <= ?;",
476 &plugin->select_messages);
478 sql_prepare (plugin->dbh,
480 "(SELECT hop_counter, signature, purpose, fragment_id,\n"
481 " fragment_offset, message_id, group_generation,\n"
482 " multicast_flags, psycstore_flags, data\n"
484 " WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
485 " ORDER BY fragment_id DESC\n"
487 "ORDER BY fragment_id;",
488 &plugin->select_latest_fragments);
490 /** @todo select_latest_messages: add method_prefix filter */
491 sql_prepare (plugin->dbh,
492 "SELECT hop_counter, signature, purpose, fragment_id,\n"
493 " fragment_offset, message_id, group_generation,\n"
494 " multicast_flags, psycstore_flags, data\n"
496 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
497 " AND message_id IN\n"
498 " (SELECT message_id\n"
500 " WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
501 " GROUP BY message_id\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,
542 "INSERT OR REPLACE INTO state\n"
543 " (channel_id, name, value_current, value_signed)\n"
544 "SELECT new.channel_id, new.name,\n"
545 " new.value_current, old.value_signed\n"
546 "FROM (SELECT (SELECT id FROM channels WHERE pub_key = ?)\n"
548 " ? AS name, ? AS value_current) AS new\n"
549 "LEFT JOIN (SELECT channel_id, name, value_signed\n"
550 " FROM state) AS old\n"
551 "ON new.channel_id = old.channel_id AND new.name = old.name;",
552 &plugin->insert_state_current);
554 sql_prepare (plugin->dbh,
555 "DELETE FROM state\n"
556 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
557 " AND (value_current IS NULL OR length(value_current) = 0)\n"
558 " AND (value_signed IS NULL OR length(value_signed) = 0);",
559 &plugin->delete_state_empty);
561 sql_prepare (plugin->dbh,
563 "SET value_signed = value_current\n"
564 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
565 &plugin->update_state_signed);
567 sql_prepare (plugin->dbh,
568 "DELETE FROM state\n"
569 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
570 &plugin->delete_state);
572 sql_prepare (plugin->dbh,
573 "INSERT INTO state_sync (channel_id, name, value)\n"
574 "VALUES ((SELECT id FROM channels WHERE pub_key = ?), ?, ?);",
575 &plugin->insert_state_sync);
577 sql_prepare (plugin->dbh,
578 "INSERT INTO state\n"
579 " (channel_id, name, value_current, value_signed)\n"
580 "SELECT channel_id, name, value, value\n"
582 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
583 &plugin->insert_state_from_sync);
585 sql_prepare (plugin->dbh,
586 "DELETE FROM state_sync\n"
587 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
588 &plugin->delete_state_sync);
590 sql_prepare (plugin->dbh,
591 "SELECT value_current\n"
593 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
595 &plugin->select_state_one);
597 sql_prepare (plugin->dbh,
598 "SELECT name, value_current\n"
600 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
601 " AND (name = ? OR substr(name, 1, ?) = ? || '_');",
602 &plugin->select_state_prefix);
604 sql_prepare (plugin->dbh,
605 "SELECT name, value_signed\n"
607 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)"
608 " AND value_signed IS NOT NULL;",
609 &plugin->select_state_signed);
616 * Shutdown database connection and associate data
618 * @param plugin the plugin context (state for this module)
621 database_shutdown (struct Plugin *plugin)
625 while (NULL != (stmt = sqlite3_next_stmt (plugin->dbh, NULL)))
627 result = sqlite3_finalize (stmt);
628 if (SQLITE_OK != result)
629 LOG (GNUNET_ERROR_TYPE_WARNING,
630 "Failed to close statement %p: %d\n", stmt, result);
632 if (SQLITE_OK != sqlite3_close (plugin->dbh))
633 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR, "sqlite3_close");
635 GNUNET_free_non_null (plugin->fn);
639 * Execute a prepared statement with a @a channel_key argument.
641 * @param plugin Plugin handle.
642 * @param stmt Statement to execute.
643 * @param channel_key Public key of the channel.
645 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
648 exec_channel (struct Plugin *plugin, sqlite3_stmt *stmt,
649 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
651 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
652 sizeof (*channel_key), SQLITE_STATIC))
654 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
657 else if (SQLITE_DONE != sqlite3_step (stmt))
659 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
663 if (SQLITE_OK != sqlite3_reset (stmt))
665 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
667 return GNUNET_SYSERR;
674 * Begin a transaction.
677 transaction_begin (struct Plugin *plugin, enum Transactions transaction)
679 sqlite3_stmt *stmt = plugin->transaction_begin;
681 if (SQLITE_DONE != sqlite3_step (stmt))
683 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
686 if (SQLITE_OK != sqlite3_reset (stmt))
688 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
690 return GNUNET_SYSERR;
693 plugin->transaction = transaction;
699 * Commit current transaction.
702 transaction_commit (struct Plugin *plugin)
704 sqlite3_stmt *stmt = plugin->transaction_commit;
706 if (SQLITE_DONE != sqlite3_step (stmt))
708 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
711 if (SQLITE_OK != sqlite3_reset (stmt))
713 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
715 return GNUNET_SYSERR;
718 plugin->transaction = TRANSACTION_NONE;
724 * Roll back current transaction.
727 transaction_rollback (struct Plugin *plugin)
729 sqlite3_stmt *stmt = plugin->transaction_rollback;
731 if (SQLITE_DONE != sqlite3_step (stmt))
733 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
736 if (SQLITE_OK != sqlite3_reset (stmt))
738 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
740 return GNUNET_SYSERR;
742 plugin->transaction = TRANSACTION_NONE;
748 channel_key_store (struct Plugin *plugin,
749 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
751 sqlite3_stmt *stmt = plugin->insert_channel_key;
753 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
754 sizeof (*channel_key), SQLITE_STATIC))
756 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
759 else if (SQLITE_DONE != sqlite3_step (stmt))
761 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
765 if (SQLITE_OK != sqlite3_reset (stmt))
767 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
769 return GNUNET_SYSERR;
777 slave_key_store (struct Plugin *plugin,
778 const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key)
780 sqlite3_stmt *stmt = plugin->insert_slave_key;
782 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, slave_key,
783 sizeof (*slave_key), SQLITE_STATIC))
785 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
788 else if (SQLITE_DONE != sqlite3_step (stmt))
790 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
795 if (SQLITE_OK != sqlite3_reset (stmt))
797 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
799 return GNUNET_SYSERR;
807 * Store join/leave events for a PSYC channel in order to be able to answer
808 * membership test queries later.
810 * @see GNUNET_PSYCSTORE_membership_store()
812 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
815 membership_store (void *cls,
816 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
817 const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
819 uint64_t announced_at,
820 uint64_t effective_since,
821 uint64_t group_generation)
823 struct Plugin *plugin = cls;
824 sqlite3_stmt *stmt = plugin->insert_membership;
826 GNUNET_assert (TRANSACTION_NONE == plugin->transaction);
828 if (announced_at > INT64_MAX ||
829 effective_since > INT64_MAX ||
830 group_generation > INT64_MAX)
833 return GNUNET_SYSERR;
836 if (GNUNET_OK != channel_key_store (plugin, channel_key)
837 || GNUNET_OK != slave_key_store (plugin, slave_key))
838 return GNUNET_SYSERR;
840 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
841 sizeof (*channel_key), SQLITE_STATIC)
842 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, slave_key,
843 sizeof (*slave_key), SQLITE_STATIC)
844 || SQLITE_OK != sqlite3_bind_int (stmt, 3, did_join)
845 || SQLITE_OK != sqlite3_bind_int64 (stmt, 4, announced_at)
846 || SQLITE_OK != sqlite3_bind_int64 (stmt, 5, effective_since)
847 || SQLITE_OK != sqlite3_bind_int64 (stmt, 6, group_generation))
849 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
852 else if (SQLITE_DONE != sqlite3_step (stmt))
854 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
858 if (SQLITE_OK != sqlite3_reset (stmt))
860 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
862 return GNUNET_SYSERR;
869 * Test if a member was admitted to the channel at the given message ID.
871 * @see GNUNET_PSYCSTORE_membership_test()
873 * @return #GNUNET_YES if the member was admitted, #GNUNET_NO if not,
874 * #GNUNET_SYSERR if there was en error.
877 membership_test (void *cls,
878 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
879 const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
882 struct Plugin *plugin = cls;
883 sqlite3_stmt *stmt = plugin->select_membership;
884 int ret = GNUNET_SYSERR;
886 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
887 sizeof (*channel_key), SQLITE_STATIC)
888 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, slave_key,
889 sizeof (*slave_key), SQLITE_STATIC)
890 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, message_id))
892 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
897 switch (sqlite3_step (stmt))
907 if (SQLITE_OK != sqlite3_reset (stmt))
909 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
917 * Store a message fragment sent to a channel.
919 * @see GNUNET_PSYCSTORE_fragment_store()
921 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
924 fragment_store (void *cls,
925 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
926 const struct GNUNET_MULTICAST_MessageHeader *msg,
927 uint32_t psycstore_flags)
929 struct Plugin *plugin = cls;
930 sqlite3_stmt *stmt = plugin->insert_fragment;
932 GNUNET_assert (TRANSACTION_NONE == plugin->transaction);
934 uint64_t fragment_id = GNUNET_ntohll (msg->fragment_id);
935 uint64_t fragment_offset = GNUNET_ntohll (msg->fragment_offset);
936 uint64_t message_id = GNUNET_ntohll (msg->message_id);
937 uint64_t group_generation = GNUNET_ntohll (msg->group_generation);
939 if (fragment_id > INT64_MAX || fragment_offset > INT64_MAX ||
940 message_id > INT64_MAX || group_generation > INT64_MAX)
942 LOG (GNUNET_ERROR_TYPE_ERROR,
943 "Tried to store fragment with a field > INT64_MAX: "
944 "%lu, %lu, %lu, %lu\n", fragment_id, fragment_offset,
945 message_id, group_generation);
947 return GNUNET_SYSERR;
950 if (GNUNET_OK != channel_key_store (plugin, channel_key))
951 return GNUNET_SYSERR;
953 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
954 sizeof (*channel_key), SQLITE_STATIC)
955 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, ntohl (msg->hop_counter) )
956 || SQLITE_OK != sqlite3_bind_blob (stmt, 3, (const void *) &msg->signature,
957 sizeof (msg->signature), SQLITE_STATIC)
958 || SQLITE_OK != sqlite3_bind_blob (stmt, 4, (const void *) &msg->purpose,
959 sizeof (msg->purpose), SQLITE_STATIC)
960 || SQLITE_OK != sqlite3_bind_int64 (stmt, 5, fragment_id)
961 || SQLITE_OK != sqlite3_bind_int64 (stmt, 6, fragment_offset)
962 || SQLITE_OK != sqlite3_bind_int64 (stmt, 7, message_id)
963 || SQLITE_OK != sqlite3_bind_int64 (stmt, 8, group_generation)
964 || SQLITE_OK != sqlite3_bind_int64 (stmt, 9, ntohl (msg->flags))
965 || SQLITE_OK != sqlite3_bind_int64 (stmt, 10, psycstore_flags)
966 || SQLITE_OK != sqlite3_bind_blob (stmt, 11, (const void *) &msg[1],
967 ntohs (msg->header.size)
968 - sizeof (*msg), SQLITE_STATIC))
970 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
973 else if (SQLITE_DONE != sqlite3_step (stmt))
975 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
979 if (SQLITE_OK != sqlite3_reset (stmt))
981 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
983 return GNUNET_SYSERR;
990 * Set additional flags for a given message.
992 * They are OR'd with any existing flags set.
994 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
997 message_add_flags (void *cls,
998 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1000 uint64_t psycstore_flags)
1002 struct Plugin *plugin = cls;
1003 sqlite3_stmt *stmt = plugin->update_message_flags;
1004 int ret = GNUNET_SYSERR;
1006 if (SQLITE_OK != sqlite3_bind_int64 (stmt, 1, psycstore_flags)
1007 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, channel_key,
1008 sizeof (*channel_key), SQLITE_STATIC)
1009 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, message_id))
1011 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1016 switch (sqlite3_step (stmt))
1019 ret = sqlite3_total_changes (plugin->dbh) > 0 ? GNUNET_OK : GNUNET_NO;
1022 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1027 if (SQLITE_OK != sqlite3_reset (stmt))
1029 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1031 return GNUNET_SYSERR;
1038 fragment_row (sqlite3_stmt *stmt, GNUNET_PSYCSTORE_FragmentCallback cb,
1041 int data_size = sqlite3_column_bytes (stmt, 9);
1042 struct GNUNET_MULTICAST_MessageHeader *msg
1043 = GNUNET_malloc (sizeof (*msg) + data_size);
1045 msg->header.size = htons (sizeof (*msg) + data_size);
1046 msg->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE);
1047 msg->hop_counter = htonl ((uint32_t) sqlite3_column_int64 (stmt, 0));
1048 memcpy (&msg->signature,
1049 sqlite3_column_blob (stmt, 1),
1050 sqlite3_column_bytes (stmt, 1));
1051 memcpy (&msg->purpose,
1052 sqlite3_column_blob (stmt, 2),
1053 sqlite3_column_bytes (stmt, 2));
1054 msg->fragment_id = GNUNET_htonll (sqlite3_column_int64 (stmt, 3));
1055 msg->fragment_offset = GNUNET_htonll (sqlite3_column_int64 (stmt, 4));
1056 msg->message_id = GNUNET_htonll (sqlite3_column_int64 (stmt, 5));
1057 msg->group_generation = GNUNET_htonll (sqlite3_column_int64 (stmt, 6));
1058 msg->flags = htonl (sqlite3_column_int64 (stmt, 7));
1059 memcpy (&msg[1], sqlite3_column_blob (stmt, 9), data_size);
1061 return cb (cb_cls, (void *) msg, sqlite3_column_int64 (stmt, 8));
1066 fragment_select (struct Plugin *plugin, sqlite3_stmt *stmt,
1067 uint64_t *returned_fragments,
1068 GNUNET_PSYCSTORE_FragmentCallback cb, void *cb_cls)
1070 int ret = GNUNET_SYSERR;
1075 sql_ret = sqlite3_step (stmt);
1079 if (ret != GNUNET_OK)
1083 ret = fragment_row (stmt, cb, cb_cls);
1084 (*returned_fragments)++;
1085 if (ret != GNUNET_YES)
1086 sql_ret = SQLITE_DONE;
1089 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1093 while (sql_ret == SQLITE_ROW);
1099 * Retrieve a message fragment range by fragment ID.
1101 * @see GNUNET_PSYCSTORE_fragment_get()
1103 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1106 fragment_get (void *cls,
1107 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1108 uint64_t first_fragment_id,
1109 uint64_t last_fragment_id,
1110 uint64_t *returned_fragments,
1111 GNUNET_PSYCSTORE_FragmentCallback cb,
1114 struct Plugin *plugin = cls;
1115 sqlite3_stmt *stmt = plugin->select_fragments;
1116 int ret = GNUNET_SYSERR;
1117 *returned_fragments = 0;
1119 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1120 sizeof (*channel_key),
1122 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, first_fragment_id)
1123 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, last_fragment_id))
1125 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1130 ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls);
1133 if (SQLITE_OK != sqlite3_reset (stmt))
1135 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1144 * Retrieve a message fragment range by fragment ID.
1146 * @see GNUNET_PSYCSTORE_fragment_get_latest()
1148 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1151 fragment_get_latest (void *cls,
1152 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1153 uint64_t fragment_limit,
1154 uint64_t *returned_fragments,
1155 GNUNET_PSYCSTORE_FragmentCallback cb,
1158 struct Plugin *plugin = cls;
1159 sqlite3_stmt *stmt = plugin->select_latest_fragments;
1160 int ret = GNUNET_SYSERR;
1161 *returned_fragments = 0;
1163 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1164 sizeof (*channel_key),
1166 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, fragment_limit))
1168 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1173 ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls);
1176 if (SQLITE_OK != sqlite3_reset (stmt))
1178 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1187 * Retrieve all fragments of a message ID range.
1189 * @see GNUNET_PSYCSTORE_message_get()
1191 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1194 message_get (void *cls,
1195 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1196 uint64_t first_message_id,
1197 uint64_t last_message_id,
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))
1213 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1218 ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls);
1221 if (SQLITE_OK != sqlite3_reset (stmt))
1223 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1232 * Retrieve all fragments of the latest messages.
1234 * @see GNUNET_PSYCSTORE_message_get_latest()
1236 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1239 message_get_latest (void *cls,
1240 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1241 uint64_t message_limit,
1242 uint64_t *returned_fragments,
1243 GNUNET_PSYCSTORE_FragmentCallback cb,
1246 struct Plugin *plugin = cls;
1247 sqlite3_stmt *stmt = plugin->select_latest_messages;
1248 int ret = GNUNET_SYSERR;
1249 *returned_fragments = 0;
1251 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1252 sizeof (*channel_key),
1254 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, channel_key,
1255 sizeof (*channel_key),
1257 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, message_limit))
1259 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1264 ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls);
1267 if (SQLITE_OK != sqlite3_reset (stmt))
1269 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1278 * Retrieve a fragment of message specified by its message ID and fragment
1281 * @see GNUNET_PSYCSTORE_message_get_fragment()
1283 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1286 message_get_fragment (void *cls,
1287 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1288 uint64_t message_id,
1289 uint64_t fragment_offset,
1290 GNUNET_PSYCSTORE_FragmentCallback cb,
1293 struct Plugin *plugin = cls;
1294 sqlite3_stmt *stmt = plugin->select_message_fragment;
1295 int ret = GNUNET_SYSERR;
1297 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1298 sizeof (*channel_key),
1300 || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, message_id)
1301 || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, fragment_offset))
1303 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1308 switch (sqlite3_step (stmt))
1314 ret = fragment_row (stmt, cb, cb_cls);
1317 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1322 if (SQLITE_OK != sqlite3_reset (stmt))
1324 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1332 * Retrieve the max. values of message counters for a channel.
1334 * @see GNUNET_PSYCSTORE_counters_get()
1336 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1339 counters_message_get (void *cls,
1340 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1341 uint64_t *max_fragment_id,
1342 uint64_t *max_message_id,
1343 uint64_t *max_group_generation)
1345 struct Plugin *plugin = cls;
1346 sqlite3_stmt *stmt = plugin->select_counters_message;
1347 int ret = GNUNET_SYSERR;
1349 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1350 sizeof (*channel_key),
1353 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1358 switch (sqlite3_step (stmt))
1364 *max_fragment_id = sqlite3_column_int64 (stmt, 0);
1365 *max_message_id = sqlite3_column_int64 (stmt, 1);
1366 *max_group_generation = sqlite3_column_int64 (stmt, 2);
1370 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1375 if (SQLITE_OK != sqlite3_reset (stmt))
1377 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1385 * Retrieve the max. values of state counters for a channel.
1387 * @see GNUNET_PSYCSTORE_counters_get()
1389 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1392 counters_state_get (void *cls,
1393 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1394 uint64_t *max_state_message_id)
1396 struct Plugin *plugin = cls;
1397 sqlite3_stmt *stmt = plugin->select_counters_state;
1398 int ret = GNUNET_SYSERR;
1400 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1401 sizeof (*channel_key),
1404 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1409 switch (sqlite3_step (stmt))
1415 *max_state_message_id = sqlite3_column_int64 (stmt, 0);
1419 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1424 if (SQLITE_OK != sqlite3_reset (stmt))
1426 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1435 * Assign a value to a state variable.
1437 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1440 state_assign (struct Plugin *plugin, sqlite3_stmt *stmt,
1441 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1442 const char *name, const void *value, size_t value_size)
1444 int ret = GNUNET_SYSERR;
1446 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1447 sizeof (*channel_key), SQLITE_STATIC)
1448 || SQLITE_OK != sqlite3_bind_text (stmt, 2, name, -1, SQLITE_STATIC)
1449 || SQLITE_OK != sqlite3_bind_blob (stmt, 3, value, value_size,
1452 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1457 switch (sqlite3_step (stmt))
1460 ret = 0 < sqlite3_total_changes (plugin->dbh) ? GNUNET_OK : GNUNET_NO;
1463 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1468 if (SQLITE_OK != sqlite3_reset (stmt))
1470 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1472 return GNUNET_SYSERR;
1480 update_message_id (struct Plugin *plugin, sqlite3_stmt *stmt,
1481 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1482 uint64_t message_id)
1484 if (SQLITE_OK != sqlite3_bind_int64 (stmt, 1, message_id)
1485 || SQLITE_OK != sqlite3_bind_blob (stmt, 2, channel_key,
1486 sizeof (*channel_key), SQLITE_STATIC))
1488 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1491 else if (SQLITE_DONE != sqlite3_step (stmt))
1493 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1496 if (SQLITE_OK != sqlite3_reset (stmt))
1498 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1500 return GNUNET_SYSERR;
1507 * Begin modifying current state.
1510 state_modify_begin (void *cls,
1511 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1512 uint64_t message_id, uint64_t state_delta)
1514 struct Plugin *plugin = cls;
1516 if (state_delta > 0)
1519 * We can only apply state modifiers in the current message if modifiers in
1520 * the previous stateful message (message_id - state_delta) were already
1524 uint64_t max_state_message_id = 0;
1525 int ret = counters_state_get (plugin, channel_key, &max_state_message_id);
1529 case GNUNET_NO: // no state yet
1536 if (max_state_message_id < message_id - state_delta)
1537 return GNUNET_NO; /* some stateful messages not yet applied */
1538 else if (message_id - state_delta < max_state_message_id)
1539 return GNUNET_NO; /* changes already applied */
1542 if (TRANSACTION_NONE != plugin->transaction)
1544 /** @todo FIXME: wait for other transaction to finish */
1545 return GNUNET_SYSERR;
1547 return transaction_begin (plugin, TRANSACTION_STATE_MODIFY);
1552 * Set the current value of state variable.
1554 * @see GNUNET_PSYCSTORE_state_modify()
1556 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1559 state_modify_op (void *cls,
1560 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1561 enum GNUNET_ENV_Operator op,
1562 const char *name, const void *value, size_t value_size)
1564 struct Plugin *plugin = cls;
1565 GNUNET_assert (TRANSACTION_STATE_MODIFY == plugin->transaction);
1569 case GNUNET_ENV_OP_ASSIGN:
1570 return state_assign (plugin, plugin->insert_state_current, channel_key,
1571 name, value, value_size);
1573 default: /** @todo implement more state operations */
1575 return GNUNET_SYSERR;
1581 * End modifying current state.
1584 state_modify_end (void *cls,
1585 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1586 uint64_t message_id)
1588 struct Plugin *plugin = cls;
1589 GNUNET_assert (TRANSACTION_STATE_MODIFY == plugin->transaction);
1592 GNUNET_OK == exec_channel (plugin, plugin->delete_state_empty, channel_key)
1593 && GNUNET_OK == update_message_id (plugin,
1594 plugin->update_max_state_message_id,
1595 channel_key, message_id)
1596 && GNUNET_OK == transaction_commit (plugin)
1597 ? GNUNET_OK : GNUNET_SYSERR;
1602 * Begin state synchronization.
1605 state_sync_begin (void *cls,
1606 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
1608 struct Plugin *plugin = cls;
1609 return exec_channel (plugin, plugin->delete_state_sync, channel_key);
1614 * Assign current value of a state variable.
1616 * @see GNUNET_PSYCSTORE_state_modify()
1618 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1621 state_sync_assign (void *cls,
1622 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1623 const char *name, const void *value, size_t value_size)
1625 struct Plugin *plugin = cls;
1626 return state_assign (cls, plugin->insert_state_sync, channel_key,
1627 name, value, value_size);
1632 * End modifying current state.
1635 state_sync_end (void *cls,
1636 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1637 uint64_t max_state_message_id,
1638 uint64_t state_hash_message_id)
1640 struct Plugin *plugin = cls;
1641 int ret = GNUNET_SYSERR;
1643 if (TRANSACTION_NONE != plugin->transaction)
1645 /** @todo FIXME: wait for other transaction to finish */
1646 return GNUNET_SYSERR;
1649 GNUNET_OK == transaction_begin (plugin, TRANSACTION_STATE_SYNC)
1650 && GNUNET_OK == exec_channel (plugin, plugin->delete_state, channel_key)
1651 && GNUNET_OK == exec_channel (plugin, plugin->insert_state_from_sync,
1653 && GNUNET_OK == exec_channel (plugin, plugin->delete_state_sync,
1655 && GNUNET_OK == update_message_id (plugin,
1656 plugin->update_state_hash_message_id,
1657 channel_key, state_hash_message_id)
1658 && GNUNET_OK == update_message_id (plugin,
1659 plugin->update_max_state_message_id,
1660 channel_key, max_state_message_id)
1661 && GNUNET_OK == transaction_commit (plugin)
1663 : transaction_rollback (plugin);
1669 * Delete the whole state.
1671 * @see GNUNET_PSYCSTORE_state_reset()
1673 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1676 state_reset (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
1678 struct Plugin *plugin = cls;
1679 return exec_channel (plugin, plugin->delete_state, channel_key);
1684 * Update signed values of state variables in the state store.
1686 * @see GNUNET_PSYCSTORE_state_hash_update()
1688 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1691 state_update_signed (void *cls,
1692 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
1694 struct Plugin *plugin = cls;
1695 return exec_channel (plugin, plugin->update_state_signed, channel_key);
1700 * Retrieve a state variable by name.
1702 * @see GNUNET_PSYCSTORE_state_get()
1704 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1707 state_get (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1708 const char *name, GNUNET_PSYCSTORE_StateCallback cb, void *cb_cls)
1710 struct Plugin *plugin = cls;
1711 int ret = GNUNET_SYSERR;
1713 sqlite3_stmt *stmt = plugin->select_state_one;
1715 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1716 sizeof (*channel_key),
1718 || SQLITE_OK != sqlite3_bind_text (stmt, 2, name, -1, SQLITE_STATIC))
1720 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1725 switch (sqlite3_step (stmt))
1731 ret = cb (cb_cls, name, sqlite3_column_blob (stmt, 0),
1732 sqlite3_column_bytes (stmt, 0));
1735 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1740 if (SQLITE_OK != sqlite3_reset (stmt))
1742 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1751 * Retrieve all state variables for a channel with the given prefix.
1753 * @see GNUNET_PSYCSTORE_state_get_prefix()
1755 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1758 state_get_prefix (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1759 const char *name, GNUNET_PSYCSTORE_StateCallback cb,
1762 struct Plugin *plugin = cls;
1763 int ret = GNUNET_SYSERR;
1764 sqlite3_stmt *stmt = plugin->select_state_prefix;
1765 size_t name_len = strlen (name);
1767 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1768 sizeof (*channel_key), SQLITE_STATIC)
1769 || SQLITE_OK != sqlite3_bind_text (stmt, 2, name, name_len, SQLITE_STATIC)
1770 || SQLITE_OK != sqlite3_bind_int (stmt, 3, name_len + 1)
1771 || SQLITE_OK != sqlite3_bind_text (stmt, 4, name, name_len, SQLITE_STATIC))
1773 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1781 sql_ret = sqlite3_step (stmt);
1785 if (ret != GNUNET_OK)
1789 ret = cb (cb_cls, (const char *) sqlite3_column_text (stmt, 0),
1790 sqlite3_column_blob (stmt, 1),
1791 sqlite3_column_bytes (stmt, 1));
1792 if (ret != GNUNET_YES)
1793 sql_ret = SQLITE_DONE;
1796 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1800 while (sql_ret == SQLITE_ROW);
1802 if (SQLITE_OK != sqlite3_reset (stmt))
1804 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1812 * Retrieve all signed state variables for a channel.
1814 * @see GNUNET_PSYCSTORE_state_get_signed()
1816 * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1819 state_get_signed (void *cls,
1820 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1821 GNUNET_PSYCSTORE_StateCallback cb, void *cb_cls)
1823 struct Plugin *plugin = cls;
1824 int ret = GNUNET_SYSERR;
1826 sqlite3_stmt *stmt = plugin->select_state_signed;
1828 if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
1829 sizeof (*channel_key), SQLITE_STATIC))
1831 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1839 sql_ret = sqlite3_step (stmt);
1843 if (ret != GNUNET_OK)
1847 ret = cb (cb_cls, (const char *) sqlite3_column_text (stmt, 0),
1848 sqlite3_column_blob (stmt, 1),
1849 sqlite3_column_bytes (stmt, 1));
1850 if (ret != GNUNET_YES)
1851 sql_ret = SQLITE_DONE;
1854 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1858 while (sql_ret == SQLITE_ROW);
1861 if (SQLITE_OK != sqlite3_reset (stmt))
1863 LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1872 * Entry point for the plugin.
1874 * @param cls The struct GNUNET_CONFIGURATION_Handle.
1875 * @return NULL on error, otherwise the plugin context
1878 libgnunet_plugin_psycstore_sqlite_init (void *cls)
1880 static struct Plugin plugin;
1881 const struct GNUNET_CONFIGURATION_Handle *cfg = cls;
1882 struct GNUNET_PSYCSTORE_PluginFunctions *api;
1884 if (NULL != plugin.cfg)
1885 return NULL; /* can only initialize once! */
1886 memset (&plugin, 0, sizeof (struct Plugin));
1888 if (GNUNET_OK != database_setup (&plugin))
1890 database_shutdown (&plugin);
1893 api = GNUNET_new (struct GNUNET_PSYCSTORE_PluginFunctions);
1895 api->membership_store = &membership_store;
1896 api->membership_test = &membership_test;
1897 api->fragment_store = &fragment_store;
1898 api->message_add_flags = &message_add_flags;
1899 api->fragment_get = &fragment_get;
1900 api->fragment_get_latest = &fragment_get_latest;
1901 api->message_get = &message_get;
1902 api->message_get_latest = &message_get_latest;
1903 api->message_get_fragment = &message_get_fragment;
1904 api->counters_message_get = &counters_message_get;
1905 api->counters_state_get = &counters_state_get;
1906 api->state_modify_begin = &state_modify_begin;
1907 api->state_modify_op = &state_modify_op;
1908 api->state_modify_end = &state_modify_end;
1909 api->state_sync_begin = &state_sync_begin;
1910 api->state_sync_assign = &state_sync_assign;
1911 api->state_sync_end = &state_sync_end;
1912 api->state_reset = &state_reset;
1913 api->state_update_signed = &state_update_signed;
1914 api->state_get = &state_get;
1915 api->state_get_prefix = &state_get_prefix;
1916 api->state_get_signed = &state_get_signed;
1918 LOG (GNUNET_ERROR_TYPE_INFO, _("SQLite database running\n"));
1924 * Exit point from the plugin.
1926 * @param cls The plugin context (as returned by "init")
1927 * @return Always NULL
1930 libgnunet_plugin_psycstore_sqlite_done (void *cls)
1932 struct GNUNET_PSYCSTORE_PluginFunctions *api = cls;
1933 struct Plugin *plugin = api->cls;
1935 database_shutdown (plugin);
1938 LOG (GNUNET_ERROR_TYPE_DEBUG, "SQLite plugin is finished\n");
1942 /* end of plugin_psycstore_sqlite.c */