src/psycstore/plugin_psycstore_mysql.c

   1 /*
   2  * This file is part of GNUnet
   3  * Copyright (C) 2013 GNUnet e.V.
   4  *
   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.
   9  *
  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.
  14  *
  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.
  19  */
  20
  21 /**
  22  * @file psycstore/plugin_psycstore_mysql.c
  23  * @brief mysql-based psycstore backend
  24  * @author Gabor X Toth
  25  * @author Christian Grothoff
  26  * @author Christophe Genevey
  27  */
  28
  29 #include "platform.h"
  30 #include "gnunet_psycstore_plugin.h"
  31 #include "gnunet_psycstore_service.h"
  32 #include "gnunet_multicast_service.h"
  33 #include "gnunet_crypto_lib.h"
  34 #include "gnunet_psyc_util_lib.h"
  35 #include "psycstore.h"
  36 #include "gnunet_my_lib.h"
  37 #include "gnunet_mysql_lib.h"
  38 #include <mysql/mysql.h>
  39
  40 /**
  41  * After how many ms "busy" should a DB operation fail for good?  A
  42  * low value makes sure that we are more responsive to requests
  43  * (especially PUTs).  A high value guarantees a higher success rate
  44  * (SELECTs in iterate can take several seconds despite LIMIT=1).
  45  *
  46  * The default value of 1s should ensure that users do not experience
  47  * huge latencies while at the same time allowing operations to
  48  * succeed with reasonable probability.
  49  */
  50 #define BUSY_TIMEOUT_MS 1000
  51
  52 #define DEBUG_PSYCSTORE GNUNET_EXTRA_LOGGING
  53
  54 /**
  55  * Log an error message at log-level 'level' that indicates
  56  * a failure of the command 'cmd' on file 'filename'
  57  * with the message given by strerror(errno).
  58  */
  59 #define LOG_MYSQL(db, level, cmd, stmt) do { GNUNET_log_from (level, "psycstore-mysql", _("`%s' failed at %s:%d with error: %s\n"), cmd, __FILE__, __LINE__, mysql_stmt_error (GNUNET_MYSQL_statement_get_stmt(stmt))); } while(0)
  60
  61 #define LOG(kind,...) GNUNET_log_from (kind, "psycstore-mysql", __VA_ARGS__)
  62
  63 enum Transactions {
  64   TRANSACTION_NONE = 0,
  65   TRANSACTION_STATE_MODIFY,
  66   TRANSACTION_STATE_SYNC,
  67 };
  68
  69 /**
  70  * Context for all functions in this plugin.
  71  */
  72 struct Plugin
  73 {
  74
  75   const struct GNUNET_CONFIGURATION_Handle *cfg;
  76
  77   /**
  78    * Database filename.
  79    */
  80   char *fn;
  81
  82   /**
  83     *Handle to talk to Mysql
  84     */
  85   struct GNUNET_MYSQL_Context *mc;
  86
  87   /**
  88    * Current transaction.
  89    */
  90   enum Transactions transaction;
  91
  92   struct GNUNET_MYSQL_StatementHandle *transaction_begin;
  93
  94   struct GNUNET_MYSQL_StatementHandle *transaction_commit;
  95
  96   struct GNUNET_MYSQL_StatementHandle *transaction_rollback;
  97
  98   /**
  99    * Precompiled SQL for channel_key_store()
 100    */
 101   struct GNUNET_MYSQL_StatementHandle *insert_channel_key;
 102
 103
 104   /**
 105    * Precompiled SQL for slave_key_store()
 106    */
 107   struct GNUNET_MYSQL_StatementHandle *insert_slave_key;
 108
 109   /**
 110    * Precompiled SQL for membership_store()
 111    */
 112   struct GNUNET_MYSQL_StatementHandle *insert_membership;
 113
 114   /**
 115    * Precompiled SQL for membership_test()
 116    */
 117   struct GNUNET_MYSQL_StatementHandle *select_membership;
 118
 119   /**
 120    * Precompiled SQL for fragment_store()
 121    */
 122   struct GNUNET_MYSQL_StatementHandle *insert_fragment;
 123
 124   /**
 125    * Precompiled SQL for message_add_flags()
 126    */
 127   struct GNUNET_MYSQL_StatementHandle *update_message_flags;
 128
 129   /**
 130    * Precompiled SQL for fragment_get()
 131    */
 132   struct GNUNET_MYSQL_StatementHandle *select_fragments;
 133
 134   /**
 135    * Precompiled SQL for fragment_get()
 136    */
 137   struct GNUNET_MYSQL_StatementHandle *select_latest_fragments;
 138
 139   /**
 140    * Precompiled SQL for message_get()
 141    */
 142   struct GNUNET_MYSQL_StatementHandle *select_messages;
 143
 144   /**
 145    * Precompiled SQL for message_get()
 146    */
 147   struct GNUNET_MYSQL_StatementHandle *select_latest_messages;
 148
 149   /**
 150    * Precompiled SQL for message_get_fragment()
 151    */
 152   struct GNUNET_MYSQL_StatementHandle *select_message_fragment;
 153
 154   /**
 155    * Precompiled SQL for counters_get_message()
 156    */
 157   struct GNUNET_MYSQL_StatementHandle *select_counters_message;
 158
 159   /**
 160    * Precompiled SQL for counters_get_state()
 161    */
 162   struct GNUNET_MYSQL_StatementHandle *select_counters_state;
 163
 164   /**
 165    * Precompiled SQL for state_modify_end()
 166    */
 167   struct GNUNET_MYSQL_StatementHandle *update_state_hash_message_id;
 168
 169   /**
 170    * Precompiled SQL for state_sync_end()
 171    */
 172   struct GNUNET_MYSQL_StatementHandle *update_max_state_message_id;
 173
 174   /**
 175    * Precompiled SQL for state_modify_op()
 176    */
 177   struct GNUNET_MYSQL_StatementHandle *insert_state_current;
 178
 179   /**
 180    * Precompiled SQL for state_modify_end()
 181    */
 182   struct GNUNET_MYSQL_StatementHandle *delete_state_empty;
 183
 184   /**
 185    * Precompiled SQL for state_set_signed()
 186    */
 187   struct GNUNET_MYSQL_StatementHandle *update_state_signed;
 188
 189   /**
 190    * Precompiled SQL for state_sync()
 191    */
 192   struct GNUNET_MYSQL_StatementHandle *insert_state_sync;
 193
 194   /**
 195    * Precompiled SQL for state_sync()
 196    */
 197   struct GNUNET_MYSQL_StatementHandle *delete_state;
 198
 199   /**
 200    * Precompiled SQL for state_sync()
 201    */
 202   struct GNUNET_MYSQL_StatementHandle *insert_state_from_sync;
 203
 204   /**
 205    * Precompiled SQL for state_sync()
 206    */
 207   struct GNUNET_MYSQL_StatementHandle *delete_state_sync;
 208
 209   /**
 210    * Precompiled SQL for state_get_signed()
 211    */
 212   struct GNUNET_MYSQL_StatementHandle *select_state_signed;
 213
 214   /**
 215    * Precompiled SQL for state_get()
 216    */
 217   struct GNUNET_MYSQL_StatementHandle *select_state_one;
 218
 219   /**
 220    * Precompiled SQL for state_get_prefix()
 221    */
 222   struct GNUNET_MYSQL_StatementHandle *select_state_prefix;
 223
 224 };
 225
 226 #if DEBUG_PSYCSTORE
 227
 228 static void
 229 mysql_trace (void *cls, const char *sql)
 230 {
 231   LOG(GNUNET_ERROR_TYPE_DEBUG, "MYSQL query:\n%s\n", sql);
 232 }
 233
 234 #endif
 235
 236
 237 /**
 238  * @brief Prepare a SQL statement
 239  *
 240  * @param dbh handle to the database
 241  * @param sql SQL statement, UTF-8 encoded
 242  * @param stmt set to the prepared statement
 243  * @return 0 on success
 244  */
 245 static int
 246 mysql_prepare (struct GNUNET_MYSQL_Context *mc,
 247               const char *sql,
 248               struct GNUNET_MYSQL_StatementHandle **stmt)
 249 {
 250   *stmt = GNUNET_MYSQL_statement_prepare (mc,
 251                                           sql);
 252
 253   LOG(GNUNET_ERROR_TYPE_DEBUG,
 254        "Prepared `%s' / %p\n", sql, stmt);
 255   if(NULL == *stmt)
 256     LOG(GNUNET_ERROR_TYPE_ERROR,
 257    _("Error preparing SQL query: %s\n  %s\n"),
 258    mysql_stmt_error (GNUNET_MYSQL_statement_get_stmt (*stmt)), sql);
 259
 260   return 0;
 261 }
 262
 263
 264 /**
 265  * Initialize the database connections and associated
 266  * data structures (create tables and indices
 267  * as needed as well).
 268  *
 269  * @param plugin the plugin context (state for this module)
 270  * @return GNUNET_OK on success
 271  */
 272 static int
 273 database_setup (struct Plugin *plugin)
 274 {
 275   char *filename;
 276
 277   if (GNUNET_OK !=
 278       GNUNET_CONFIGURATION_get_value_filename (plugin->cfg, "psycstore-mysql",
 279                                                "FILENAME", &filename))
 280   {
 281     GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
 282                                "psycstore-mysql", "FILENAME");
 283     return GNUNET_SYSERR;
 284   }
 285
 286   if (GNUNET_OK != GNUNET_DISK_file_test (filename))
 287   {
 288     if (GNUNET_OK != GNUNET_DISK_directory_create_for_file (filename))
 289     {
 290       GNUNET_break (0);
 291       GNUNET_free (filename);
 292       return GNUNET_SYSERR;
 293     }
 294   }
 295   /* filename should be UTF-8-encoded. If it isn't, it's a bug */
 296   plugin->fn = filename;
 297
 298   /* Open database and precompile statements */
 299   plugin->mc = GNUNET_MYSQL_context_create(plugin->cfg, "psycstore-mysql");
 300
 301   if (NULL == plugin->mc)
 302   {
 303     LOG(GNUNET_ERROR_TYPE_ERROR,
 304    _("Unable to initialize Mysql.\n"));
 305     return GNUNET_SYSERR;
 306   }
 307
 308   /* Create tables */
 309
 310   GNUNET_MYSQL_statement_run (plugin->mc,
 311                               "CREATE TABLE IF NOT EXISTS channels (\n"
 312                               " id INT AUTO_INCREMENT,\n"
 313                               " pub_key BLOB,\n"
 314                               " max_state_message_id INT,\n"
 315                               " state_hash_message_id INT,\n"
 316                               " PRIMARY KEY(id),\n"
 317                               " UNIQUE KEY(pub_key(5))\n"
 318                               ");");
 319
 320   GNUNET_MYSQL_statement_run (plugin->mc,
 321                               "CREATE TABLE IF NOT EXISTS slaves (\n"
 322                               " id INT AUTO_INCREMENT,\n"
 323                               " pub_key BLOB,\n"
 324                               " PRIMARY KEY(id),\n"
 325                               " UNIQUE KEY(pub_key(5))\n"
 326                               ");");
 327
 328   GNUNET_MYSQL_statement_run (plugin->mc,
 329                               "CREATE TABLE IF NOT EXISTS membership (\n"
 330                               "  channel_id INT NOT NULL REFERENCES channels(id),\n"
 331                               "  slave_id INT NOT NULL REFERENCES slaves(id),\n"
 332                               "  did_join INT NOT NULL,\n"
 333                               "  announced_at BIGINT UNSIGNED NOT NULL,\n"
 334                               "  effective_since BIGINT UNSIGNED NOT NULL,\n"
 335                               "  group_generation BIGINT UNSIGNED NOT NULL\n"
 336                               ");");
 337
 338 /*** FIX because IF NOT EXISTS doesn't work ***/
 339   GNUNET_MYSQL_statement_run (plugin->mc,
 340                               "CREATE INDEX idx_membership_channel_id_slave_id "
 341                               "ON membership (channel_id, slave_id);");
 342
 343   /** @todo messages table: add method_name column */
 344   GNUNET_MYSQL_statement_run (plugin->mc,
 345                               "CREATE TABLE IF NOT EXISTS messages (\n"
 346                               "  channel_id INT NOT NULL REFERENCES channels(id),\n"
 347                               "  hop_counter BIGINT UNSIGNED NOT NULL,\n"
 348                               "  signature BLOB,\n"
 349                               "  purpose BLOB,\n"
 350                               "  fragment_id BIGINT UNSIGNED NOT NULL,\n"
 351                               "  fragment_offset BIGINT UNSIGNED NOT NULL,\n"
 352                               "  message_id BIGINT UNSIGNED NOT NULL,\n"
 353                               "  group_generation BIGINT UNSIGNED NOT NULL,\n"
 354                               "  multicast_flags BIGINT UNSIGNED NOT NULL,\n"
 355                               "  psycstore_flags BIGINT UNSIGNED NOT NULL,\n"
 356                               "  data BLOB,\n"
 357                               "  PRIMARY KEY (channel_id, fragment_id),\n"
 358                               "  UNIQUE KEY(channel_id, message_id, fragment_offset)\n"
 359                               ");");
 360
 361   GNUNET_MYSQL_statement_run (plugin->mc,
 362                               "CREATE TABLE IF NOT EXISTS state (\n"
 363                               "  channel_id INT NOT NULL REFERENCES channels(id),\n"
 364                               "  name TEXT NOT NULL,\n"
 365                               "  value_current BLOB,\n"
 366                               "  value_signed BLOB,\n"
 367                               "  PRIMARY KEY (channel_id, name(5))\n"
 368                               ");");
 369
 370   GNUNET_MYSQL_statement_run (plugin->mc,
 371                               "CREATE TABLE IF NOT EXISTS state_sync (\n"
 372                               "  channel_id INT NOT NULL REFERENCES channels(id),\n"
 373                               "  name TEXT NOT NULL,\n"
 374                               "  value BLOB,\n"
 375                               "  PRIMARY KEY (channel_id, name(5))\n"
 376                               ");");
 377
 378   /* Prepare statements */
 379   mysql_prepare (plugin->mc,
 380                 "BEGIN",
 381                 &plugin->transaction_begin);
 382
 383   mysql_prepare (plugin->mc,
 384                 "COMMIT",
 385                 &plugin->transaction_commit);
 386
 387   mysql_prepare (plugin->mc,
 388                 "ROLLBACK;",
 389                 &plugin->transaction_rollback);
 390
 391   mysql_prepare (plugin->mc,
 392                 "INSERT IGNORE INTO channels (pub_key) VALUES (?);",
 393                 &plugin->insert_channel_key);
 394
 395   mysql_prepare (plugin->mc,
 396                 "INSERT IGNORE INTO slaves (pub_key) VALUES (?);",
 397                 &plugin->insert_slave_key);
 398
 399   mysql_prepare (plugin->mc,
 400                 "INSERT INTO membership\n"
 401                 " (channel_id, slave_id, did_join, announced_at,\n"
 402                 "  effective_since, group_generation)\n"
 403                 "VALUES ((SELECT id FROM channels WHERE pub_key = ?),\n"
 404                 "        (SELECT id FROM slaves WHERE pub_key = ?),\n"
 405                 "        ?, ?, ?, ?);",
 406                 &plugin->insert_membership);
 407
 408   mysql_prepare (plugin->mc,
 409                 "SELECT did_join FROM membership\n"
 410                "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
 411                "      AND slave_id = (SELECT id FROM slaves WHERE pub_key = ?)\n"
 412                "      AND effective_since <= ? AND did_join = 1\n"
 413                "ORDER BY announced_at DESC LIMIT 1;",
 414                &plugin->select_membership);
 415
 416   mysql_prepare (plugin->mc,
 417                 "INSERT IGNORE INTO messages\n"
 418                " (channel_id, hop_counter, signature, purpose,\n"
 419                "  fragment_id, fragment_offset, message_id,\n"
 420                "  group_generation, multicast_flags, psycstore_flags, data)\n"
 421                "VALUES ((SELECT id FROM channels WHERE pub_key = ?),\n"
 422                "        ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);",
 423                 &plugin->insert_fragment);
 424
 425   mysql_prepare (plugin->mc,
 426                 "UPDATE messages\n"
 427                 "SET psycstore_flags = psycstore_flags | ?\n"
 428                 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
 429                 "      AND message_id = ? AND fragment_offset = 0;",
 430                 &plugin->update_message_flags);
 431
 432   mysql_prepare (plugin->mc,
 433                   "SELECT hop_counter, signature, purpose, fragment_id,\n"
 434                   "       fragment_offset, message_id, group_generation,\n"
 435                   "       multicast_flags, psycstore_flags, data\n"
 436                   "FROM messages\n"
 437                   "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
 438                   "      AND ? <= fragment_id AND fragment_id <= ?;",
 439                 &plugin->select_fragments);
 440
 441   /** @todo select_messages: add method_prefix filter */
 442   mysql_prepare (plugin->mc,
 443                 "SELECT hop_counter, signature, purpose, fragment_id,\n"
 444                 "       fragment_offset, message_id, group_generation,\n"
 445                 "       multicast_flags, psycstore_flags, data\n"
 446                 "FROM messages\n"
 447                 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
 448                 "      AND ? <= message_id AND message_id <= ?"
 449                 "LIMIT ?;",
 450                 &plugin->select_messages);
 451
 452   mysql_prepare (plugin->mc,
 453                 "SELECT * FROM\n"
 454                 "(SELECT hop_counter, signature, purpose, fragment_id,\n"
 455                 "        fragment_offset, message_id, group_generation,\n"
 456                 "        multicast_flags, psycstore_flags, data\n"
 457                 " FROM messages\n"
 458                 " WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
 459                 " ORDER BY fragment_id DESC\n"
 460                 " LIMIT ?)\n"
 461                 "ORDER BY fragment_id;",
 462                 &plugin->select_latest_fragments);
 463
 464   /** @todo select_latest_messages: add method_prefix filter */
 465   mysql_prepare (plugin->mc,
 466                 "SELECT hop_counter, signature, purpose, fragment_id,\n"
 467                 "       fragment_offset, message_id, group_generation,\n"
 468                 "        multicast_flags, psycstore_flags, data\n"
 469                 "FROM messages\n"
 470                 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
 471                 "      AND message_id IN\n"
 472                 "      (SELECT message_id\n"
 473                 "       FROM messages\n"
 474                 "       WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
 475                 "       GROUP BY message_id\n"
 476                 "       ORDER BY message_id\n"
 477                 "       DESC LIMIT ?)\n"
 478                 "ORDER BY fragment_id;",
 479                 &plugin->select_latest_messages);
 480
 481   mysql_prepare (plugin->mc,
 482                 "SELECT hop_counter, signature, purpose, fragment_id,\n"
 483                 "       fragment_offset, message_id, group_generation,\n"
 484                 "       multicast_flags, psycstore_flags, data\n"
 485                 "FROM messages\n"
 486                 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
 487                 "      AND message_id = ? AND fragment_offset = ?;",
 488                 &plugin->select_message_fragment);
 489
 490   mysql_prepare (plugin->mc,
 491                 "SELECT fragment_id, message_id, group_generation\n"
 492                 "FROM messages\n"
 493                 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
 494                 "ORDER BY fragment_id DESC LIMIT 1;",
 495                 &plugin->select_counters_message);
 496
 497   mysql_prepare (plugin->mc,
 498                 "SELECT max_state_message_id\n"
 499                 "FROM channels\n"
 500                 "WHERE pub_key = ? AND max_state_message_id IS NOT NULL;",
 501                 &plugin->select_counters_state);
 502
 503   mysql_prepare (plugin->mc,
 504                 "UPDATE channels\n"
 505                 "SET max_state_message_id = ?\n"
 506                 "WHERE pub_key = ?;",
 507                 &plugin->update_max_state_message_id);
 508
 509   mysql_prepare (plugin->mc,
 510                 "UPDATE channels\n"
 511                 "SET state_hash_message_id = ?\n"
 512                 "WHERE pub_key = ?;",
 513                 &plugin->update_state_hash_message_id);
 514
 515   mysql_prepare (plugin->mc,
 516                 "REPLACE INTO state\n"
 517                 "  (channel_id, name, value_current, value_signed)\n"
 518                 "SELECT new.channel_id, new.name,\n"
 519                 "       new.value_current, old.value_signed\n"
 520                 "FROM (SELECT (SELECT id FROM channels WHERE pub_key = ?)\n"
 521                 "             AS channel_id,\n"
 522                 "             ? AS name, ? AS value_current) AS new\n"
 523                 "LEFT JOIN (SELECT channel_id, name, value_signed\n"
 524                 "           FROM state) AS old\n"
 525                 "ON new.channel_id = old.channel_id AND new.name = old.name;",
 526                 &plugin->insert_state_current);
 527
 528   mysql_prepare (plugin->mc,
 529                 "DELETE FROM state\n"
 530                 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
 531                 "      AND (value_current IS NULL OR length(value_current) = 0)\n"
 532                 "      AND (value_signed IS NULL OR length(value_signed) = 0);",
 533                 &plugin->delete_state_empty);
 534
 535   mysql_prepare (plugin->mc,
 536                 "UPDATE state\n"
 537                 "SET value_signed = value_current\n"
 538                 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
 539                 &plugin->update_state_signed);
 540
 541   mysql_prepare (plugin->mc,
 542                 "DELETE FROM state\n"
 543                 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
 544                 &plugin->delete_state);
 545
 546   mysql_prepare (plugin->mc,
 547                 "INSERT INTO state_sync (channel_id, name, value)\n"
 548                 "VALUES ((SELECT id FROM channels WHERE pub_key = ?), ?, ?);",
 549                 &plugin->insert_state_sync);
 550
 551   mysql_prepare (plugin->mc,
 552                 "INSERT INTO state\n"
 553                 " (channel_id, name, value_current, value_signed)\n"
 554                 "SELECT channel_id, name, value, value\n"
 555                 "FROM state_sync\n"
 556                 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
 557                 &plugin->insert_state_from_sync);
 558
 559   mysql_prepare (plugin->mc,
 560                 "DELETE FROM state_sync\n"
 561                 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);",
 562                 &plugin->delete_state_sync);
 563
 564   mysql_prepare (plugin->mc,
 565                 "SELECT value_current\n"
 566                 "FROM state\n"
 567                 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
 568                 "      AND name = ?;",
 569                 &plugin->select_state_one);
 570
 571   mysql_prepare (plugin->mc,
 572                 "SELECT name, value_current\n"
 573                 "FROM state\n"
 574                 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
 575                 "      AND (name = ? OR substr(name, 1, ?) = ? || '_');",
 576                 &plugin->select_state_prefix);
 577
 578   mysql_prepare (plugin->mc,
 579                 "SELECT name, value_signed\n"
 580                 "FROM state\n"
 581                 "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)"
 582                 "      AND value_signed IS NOT NULL;",
 583                 &plugin->select_state_signed);
 584
 585   return GNUNET_OK;
 586 }
 587
 588
 589 /**
 590  * Shutdown database connection and associate data
 591  * structures.
 592  * @param plugin the plugin context (state for this module)
 593  */
 594 static void
 595 database_shutdown (struct Plugin *plugin)
 596 {
 597   GNUNET_MYSQL_context_destroy (plugin->mc);
 598
 599   GNUNET_free_non_null (plugin->fn);
 600
 601 }
 602
 603
 604 /**
 605  * Execute a prepared statement with a @a channel_key argument.
 606  *
 607  * @param plugin Plugin handle.
 608  * @param stmt Statement to execute.
 609  * @param channel_key Public key of the channel.
 610  *
 611  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
 612  */
 613 static int
 614 exec_channel (struct Plugin *plugin, struct GNUNET_MYSQL_StatementHandle *stmt,
 615               const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
 616 {
 617   struct GNUNET_MY_QueryParam params[] = {
 618     GNUNET_MY_query_param_auto_from_type (channel_key),
 619     GNUNET_MY_query_param_end
 620   };
 621
 622   if(GNUNET_OK != GNUNET_MY_exec_prepared (plugin->mc,
 623                                           stmt,
 624                                           params))
 625   {
 626     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 627                 "mysql exec_channel", stmt);
 628   }
 629
 630   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
 631   {
 632     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 633               "mysql_stmt_reset", stmt);
 634     return GNUNET_SYSERR;
 635   }
 636
 637   return GNUNET_OK;
 638 }
 639
 640
 641 /**
 642  * Begin a transaction.
 643  */
 644 static int
 645 transaction_begin (struct Plugin *plugin, enum Transactions transaction)
 646 {
 647   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->transaction_begin;
 648
 649   struct GNUNET_MY_QueryParam params[] = {
 650     GNUNET_MY_query_param_end
 651   };
 652
 653   if (GNUNET_OK != GNUNET_MY_exec_prepared (plugin->mc,
 654                                             stmt,
 655                                             params))
 656   {
 657     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 658                 "mysql extract_result", stmt);
 659     return GNUNET_SYSERR;
 660   }
 661
 662   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt(stmt)))
 663   {
 664     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 665               "mysql_stmt_reset", stmt);
 666     return GNUNET_SYSERR;
 667   }
 668
 669   plugin->transaction = transaction;
 670   return GNUNET_OK;
 671 }
 672
 673
 674 /**
 675  * Commit current transaction.
 676  */
 677 static int
 678 transaction_commit (struct Plugin *plugin)
 679 {
 680   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->transaction_commit;
 681
 682   struct GNUNET_MY_QueryParam params[] = {
 683     GNUNET_MY_query_param_end
 684   };
 685
 686   if (GNUNET_OK != GNUNET_MY_exec_prepared( plugin->mc,
 687                                             stmt,
 688                                             params))
 689   {
 690     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 691                 "mysql extract_result", stmt);
 692     return GNUNET_SYSERR;
 693   }
 694
 695   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
 696   {
 697     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 698               "mysql_stmt_reset", stmt);
 699     return GNUNET_SYSERR;
 700   }
 701
 702   plugin->transaction = TRANSACTION_NONE;
 703   return GNUNET_OK;
 704 }
 705
 706
 707 /**
 708  * Roll back current transaction.
 709  */
 710 static int
 711 transaction_rollback (struct Plugin *plugin)
 712 {
 713   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->transaction_rollback;
 714
 715   struct GNUNET_MY_QueryParam params[] = {
 716     GNUNET_MY_query_param_end
 717   };
 718
 719   if (GNUNET_OK != GNUNET_MY_exec_prepared (plugin->mc,
 720                                             stmt,
 721                                             params))
 722   {
 723     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 724                 "mysql extract_result", stmt);
 725     return GNUNET_SYSERR;
 726   }
 727
 728   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
 729   {
 730     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 731               "mysql_stmt_reset", stmt);
 732     return GNUNET_SYSERR;
 733   }
 734
 735   plugin->transaction = TRANSACTION_NONE;
 736   return GNUNET_OK;
 737 }
 738
 739
 740 static int
 741 channel_key_store (struct Plugin *plugin,
 742                    const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
 743 {
 744   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->insert_channel_key;
 745
 746   struct GNUNET_MY_QueryParam params[] = {
 747     GNUNET_MY_query_param_auto_from_type (channel_key),
 748     GNUNET_MY_query_param_end
 749   };
 750
 751   if (GNUNET_OK != GNUNET_MY_exec_prepared (plugin->mc,
 752                                             stmt,
 753                                             params))
 754   {
 755     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 756                 "mysql exec_prepared", stmt);
 757     return GNUNET_SYSERR;
 758   }
 759
 760   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
 761   {
 762     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 763               "mysql_stmt_reset", stmt);
 764     return GNUNET_SYSERR;
 765   }
 766
 767   return GNUNET_OK;
 768 }
 769
 770
 771 static int
 772 slave_key_store (struct Plugin *plugin,
 773                  const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key)
 774 {
 775   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->insert_slave_key;
 776
 777   struct GNUNET_MY_QueryParam params[] = {
 778     GNUNET_MY_query_param_auto_from_type (slave_key),
 779     GNUNET_MY_query_param_end
 780   };
 781
 782   if (GNUNET_OK != GNUNET_MY_exec_prepared( plugin->mc,
 783                                             stmt,
 784                                             params))
 785   {
 786     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 787                 "mysql exec_prepared", stmt);
 788     return GNUNET_SYSERR;
 789   }
 790
 791   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
 792   {
 793     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 794               "mysql_stmt_reset", stmt);
 795     return GNUNET_SYSERR;
 796   }
 797
 798   return GNUNET_OK;
 799 }
 800
 801
 802 /**
 803  * Store join/leave events for a PSYC channel in order to be able to answer
 804  * membership test queries later.
 805  *
 806  * @see GNUNET_PSYCSTORE_membership_store()
 807  *
 808  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
 809  */
 810 static int
 811 mysql_membership_store (void *cls,
 812                          const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 813                          const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
 814                          int did_join,
 815                          uint64_t announced_at,
 816                          uint64_t effective_since,
 817                          uint64_t group_generation)
 818 {
 819   struct Plugin *plugin = cls;
 820
 821   uint32_t idid_join = (uint32_t)did_join;
 822
 823   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->insert_membership;
 824
 825   GNUNET_assert (TRANSACTION_NONE == plugin->transaction);
 826
 827   if (announced_at > INT64_MAX ||
 828       effective_since > INT64_MAX ||
 829       group_generation > INT64_MAX)
 830   {
 831     GNUNET_break (0);
 832     return GNUNET_SYSERR;
 833   }
 834
 835   if (GNUNET_OK != channel_key_store (plugin, channel_key)
 836       || GNUNET_OK != slave_key_store (plugin, slave_key))
 837     return GNUNET_SYSERR;
 838
 839   struct GNUNET_MY_QueryParam params[] = {
 840     GNUNET_MY_query_param_auto_from_type (channel_key),
 841     GNUNET_MY_query_param_auto_from_type (slave_key),
 842     GNUNET_MY_query_param_uint32 (&idid_join),
 843     GNUNET_MY_query_param_uint64 (&announced_at),
 844     GNUNET_MY_query_param_uint64 (&effective_since),
 845     GNUNET_MY_query_param_uint64 (&group_generation),
 846     GNUNET_MY_query_param_end
 847   };
 848
 849   if (GNUNET_OK != GNUNET_MY_exec_prepared (plugin->mc,
 850                                             stmt,
 851                                             params))
 852   {
 853     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 854                 "mysql exec_prepared", stmt);
 855     return GNUNET_SYSERR;
 856   }
 857
 858   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
 859   {
 860     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 861               "mysql_stmt_reset", stmt);
 862     return GNUNET_SYSERR;
 863   }
 864   return GNUNET_OK;
 865 }
 866
 867 /**
 868  * Test if a member was admitted to the channel at the given message ID.
 869  *
 870  * @see GNUNET_PSYCSTORE_membership_test()
 871  *
 872  * @return #GNUNET_YES if the member was admitted, #GNUNET_NO if not,
 873  *         #GNUNET_SYSERR if there was en error.
 874  */
 875 static int
 876 membership_test (void *cls,
 877                  const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 878                  const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
 879                  uint64_t message_id)
 880 {
 881   struct Plugin *plugin = cls;
 882
 883   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->select_membership;
 884
 885   uint32_t did_join = 0;
 886
 887   int ret = GNUNET_SYSERR;
 888
 889   struct GNUNET_MY_QueryParam params_select[] = {
 890     GNUNET_MY_query_param_auto_from_type (channel_key),
 891     GNUNET_MY_query_param_auto_from_type (slave_key),
 892     GNUNET_MY_query_param_uint64 (&message_id),
 893     GNUNET_MY_query_param_end
 894   };
 895
 896   if (GNUNET_OK != GNUNET_MY_exec_prepared (plugin->mc,
 897                               stmt,
 898                               params_select))
 899   {
 900     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 901                 "mysql execute prepared", stmt);
 902     return GNUNET_SYSERR;
 903   }
 904
 905   struct GNUNET_MY_ResultSpec results_select[] = {
 906     GNUNET_MY_result_spec_uint32 (&did_join),
 907     GNUNET_MY_result_spec_end
 908   };
 909
 910   switch(GNUNET_MY_extract_result (stmt,
 911                                 results_select))
 912   {
 913     case GNUNET_NO:
 914       ret = GNUNET_NO;
 915       break;
 916     case GNUNET_OK:
 917       ret = GNUNET_YES;
 918       break;
 919     default:
 920       LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 921                 "mysql extract_result", stmt);
 922       return GNUNET_SYSERR;
 923   }
 924
 925   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
 926   {
 927     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 928               "mysql_stmt_reset", stmt);
 929     return GNUNET_SYSERR;
 930   }
 931
 932   return ret;
 933 }
 934
 935 /**
 936  * Store a message fragment sent to a channel.
 937  *
 938  * @see GNUNET_PSYCSTORE_fragment_store()
 939  *
 940  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
 941  */
 942 static int
 943 fragment_store (void *cls,
 944                 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 945                 const struct GNUNET_MULTICAST_MessageHeader *msg,
 946                 uint32_t psycstore_flags)
 947 {
 948   struct Plugin *plugin = cls;
 949
 950   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->insert_fragment;
 951
 952   GNUNET_assert (TRANSACTION_NONE == plugin->transaction);
 953
 954   uint64_t fragment_id = GNUNET_ntohll (msg->fragment_id);
 955
 956   uint64_t fragment_offset = GNUNET_ntohll (msg->fragment_offset);
 957   uint64_t message_id = GNUNET_ntohll (msg->message_id);
 958   uint64_t group_generation = GNUNET_ntohll (msg->group_generation);
 959
 960   uint64_t hop_counter = ntohl(msg->hop_counter);
 961   uint64_t flags = ntohl(msg->flags);
 962
 963   if (fragment_id > INT64_MAX || fragment_offset > INT64_MAX ||
 964       message_id > INT64_MAX || group_generation > INT64_MAX)
 965   {
 966     LOG(GNUNET_ERROR_TYPE_ERROR,
 967          "Tried to store fragment with a field > INT64_MAX: "
 968          "%lu, %lu, %lu, %lu\n", fragment_id, fragment_offset,
 969          message_id, group_generation);
 970     GNUNET_break (0);
 971     return GNUNET_SYSERR;
 972   }
 973
 974   if (GNUNET_OK != channel_key_store (plugin, channel_key))
 975     return GNUNET_SYSERR;
 976
 977   struct GNUNET_MY_QueryParam params_insert[] = {
 978     GNUNET_MY_query_param_auto_from_type (channel_key),
 979     GNUNET_MY_query_param_uint64 (&hop_counter),
 980     GNUNET_MY_query_param_auto_from_type (&msg->signature),
 981     GNUNET_MY_query_param_auto_from_type (&msg->purpose),
 982     GNUNET_MY_query_param_uint64 (&fragment_id),
 983     GNUNET_MY_query_param_uint64 (&fragment_offset),
 984     GNUNET_MY_query_param_uint64 (&message_id),
 985     GNUNET_MY_query_param_uint64 (&group_generation),
 986     GNUNET_MY_query_param_uint64 (&flags),
 987     GNUNET_MY_query_param_uint32 (&psycstore_flags),
 988     GNUNET_MY_query_param_fixed_size (&msg[1], ntohs (msg->header.size)
 989                                                   - sizeof (*msg)),
 990     GNUNET_MY_query_param_end
 991   };
 992
 993   if (GNUNET_OK != GNUNET_MY_exec_prepared (plugin->mc,
 994                               stmt,
 995                               params_insert))
 996   {
 997     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 998               "mysql execute prepared", stmt);
 999     return GNUNET_SYSERR;
1000   }
1001
1002   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
1003   {
1004     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1005               "mysql_stmt_reset", stmt);
1006     return GNUNET_SYSERR;
1007   }
1008
1009   return GNUNET_OK;
1010 }
1011
1012 /**
1013  * Set additional flags for a given message.
1014  *
1015  * They are OR'd with any existing flags set.
1016  *
1017  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1018  */
1019 static int
1020 message_add_flags (void *cls,
1021                    const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1022                    uint64_t message_id,
1023                    uint64_t psycstore_flags)
1024 {
1025   struct Plugin *plugin = cls;
1026
1027   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->update_message_flags;
1028
1029   int sql_ret;
1030   int ret = GNUNET_SYSERR;
1031
1032   struct GNUNET_MY_QueryParam params_update[] = {
1033     GNUNET_MY_query_param_uint64 (&psycstore_flags),
1034     GNUNET_MY_query_param_auto_from_type (channel_key),
1035     GNUNET_MY_query_param_uint64 (&message_id),
1036     GNUNET_MY_query_param_end
1037   };
1038
1039   sql_ret = GNUNET_MY_exec_prepared (plugin->mc,
1040                                             stmt,
1041                                             params_update);
1042   switch(sql_ret)
1043   {
1044     case GNUNET_OK:
1045       ret = GNUNET_OK;
1046       break;
1047     default:
1048        LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1049               "mysql execute prepared", stmt);
1050   }
1051
1052   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
1053   {
1054     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1055               "mysql_stmt_reset", stmt);
1056     return GNUNET_SYSERR;
1057   }
1058
1059   return ret;
1060 }
1061
1062
1063 static int
1064 fragment_row (struct GNUNET_MYSQL_StatementHandle *stmt,
1065               GNUNET_PSYCSTORE_FragmentCallback cb,
1066               void *cb_cls)
1067 {
1068
1069   uint32_t hop_counter;
1070   void *signature = NULL;
1071   void *purpose = NULL;
1072   size_t signature_size;
1073   size_t purpose_size;
1074
1075   uint64_t fragment_id;
1076   uint64_t fragment_offset;
1077   uint64_t message_id;
1078   uint64_t group_generation;
1079   uint64_t flags;
1080   void *buf;
1081   size_t buf_size;
1082   int ret = GNUNET_SYSERR;
1083   int sql_ret;
1084   struct GNUNET_MULTICAST_MessageHeader *mp;
1085
1086   uint64_t msg_flags;
1087
1088
1089   struct GNUNET_MY_ResultSpec results[] = {
1090     GNUNET_MY_result_spec_uint32 (&hop_counter),
1091     GNUNET_MY_result_spec_variable_size (&signature, &signature_size),
1092     GNUNET_MY_result_spec_variable_size (&purpose, &purpose_size),
1093     GNUNET_MY_result_spec_uint64 (&fragment_id),
1094     GNUNET_MY_result_spec_uint64 (&fragment_offset),
1095     GNUNET_MY_result_spec_uint64 (&message_id),
1096     GNUNET_MY_result_spec_uint64 (&group_generation),
1097     GNUNET_MY_result_spec_uint64 (&msg_flags),
1098     GNUNET_MY_result_spec_uint64 (&flags),
1099     GNUNET_MY_result_spec_variable_size (&buf,
1100                                          &buf_size),
1101     GNUNET_MY_result_spec_end
1102   };
1103
1104   sql_ret = GNUNET_MY_extract_result (stmt,
1105                                 results);
1106   switch (sql_ret)
1107   {
1108     case GNUNET_NO:
1109       if (ret != GNUNET_OK)
1110           ret = GNUNET_NO;
1111       break;
1112     case GNUNET_OK:
1113
1114       mp = GNUNET_malloc (sizeof (*mp) + buf_size);
1115
1116       mp->header.size = htons (sizeof (*mp) + buf_size);
1117       mp->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE);
1118       mp->hop_counter = htonl (hop_counter);
1119       GNUNET_memcpy (&mp->signature,
1120                     signature,
1121                     signature_size);
1122       GNUNET_memcpy (&mp->purpose,
1123                     purpose,
1124                     purpose_size);
1125       mp->fragment_id = GNUNET_htonll (fragment_id);
1126       mp->fragment_offset = GNUNET_htonll (fragment_offset);
1127       mp->message_id = GNUNET_htonll (message_id);
1128       mp->group_generation = GNUNET_htonll (group_generation);
1129       mp->flags = htonl(msg_flags);
1130
1131       GNUNET_memcpy (&mp[1],
1132                     buf,
1133                     buf_size);
1134       ret = cb (cb_cls,
1135                 mp,
1136                 (enum GNUNET_PSYCSTORE_MessageFlags) flags);
1137
1138       GNUNET_MY_cleanup_result (results);
1139
1140       break;
1141     default:
1142       LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1143                     "mysql extract_result", stmt);
1144   }
1145
1146   return ret;
1147 }
1148
1149
1150 static int
1151 fragment_select (struct Plugin *plugin, struct GNUNET_MYSQL_StatementHandle *stmt,
1152                  struct GNUNET_MY_QueryParam *params,
1153                  uint64_t *returned_fragments,
1154                  GNUNET_PSYCSTORE_FragmentCallback cb, void *cb_cls)
1155 {
1156   int ret = GNUNET_SYSERR;
1157   int sql_ret;
1158
1159   sql_ret = GNUNET_MY_exec_prepared (plugin->mc,
1160                           stmt,
1161                           params);
1162   switch(sql_ret)
1163   {
1164     case GNUNET_NO:
1165       if (ret != GNUNET_OK)
1166           ret = GNUNET_NO;
1167       break;
1168     case GNUNET_YES:
1169        ret = fragment_row (stmt, cb, cb_cls);
1170        (*returned_fragments)++;
1171       break;
1172     default:
1173       LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1174                 "mysql exec_prepared", stmt);
1175   }
1176
1177   return ret;
1178 }
1179
1180 /**
1181  * Retrieve a message fragment range by fragment ID.
1182  *
1183  * @see GNUNET_PSYCSTORE_fragment_get()
1184  *
1185  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1186  */
1187 static int
1188 fragment_get (void *cls,
1189               const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1190               uint64_t first_fragment_id,
1191               uint64_t last_fragment_id,
1192               uint64_t *returned_fragments,
1193               GNUNET_PSYCSTORE_FragmentCallback cb,
1194               void *cb_cls)
1195 {
1196   struct Plugin *plugin = cls;
1197   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->select_fragments;
1198   int ret = GNUNET_SYSERR;
1199   *returned_fragments = 0;
1200
1201   struct GNUNET_MY_QueryParam params_select[] = {
1202     GNUNET_MY_query_param_auto_from_type (channel_key),
1203     GNUNET_MY_query_param_uint64 (&first_fragment_id),
1204     GNUNET_MY_query_param_uint64 (&last_fragment_id),
1205     GNUNET_MY_query_param_end
1206   };
1207
1208   ret = fragment_select (plugin, stmt, params_select, returned_fragments, cb, cb_cls);
1209
1210   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
1211   {
1212     LOG_MYSQL (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1213               "mysql_stmt_reset", stmt);
1214     return GNUNET_SYSERR;
1215   }
1216
1217   return ret;
1218 }
1219
1220
1221 /**
1222  * Retrieve a message fragment range by fragment ID.
1223  *
1224  * @see GNUNET_PSYCSTORE_fragment_get_latest()
1225  *
1226  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1227  */
1228 static int
1229 fragment_get_latest (void *cls,
1230                      const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1231                      uint64_t fragment_limit,
1232                      uint64_t *returned_fragments,
1233                      GNUNET_PSYCSTORE_FragmentCallback cb,
1234                      void *cb_cls)
1235 {
1236   struct Plugin *plugin = cls;
1237
1238   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->select_latest_fragments;
1239
1240   int ret = GNUNET_SYSERR;
1241   *returned_fragments = 0;
1242
1243   struct GNUNET_MY_QueryParam params_select[] = {
1244     GNUNET_MY_query_param_auto_from_type (channel_key),
1245     GNUNET_MY_query_param_uint64 (&fragment_limit),
1246     GNUNET_MY_query_param_end
1247   };
1248
1249   ret = fragment_select (plugin, stmt, params_select, returned_fragments, cb, cb_cls);
1250
1251   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
1252   {
1253     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1254               "mysql_stmt_reset", stmt);
1255     return GNUNET_SYSERR;
1256   }
1257
1258   return ret;
1259 }
1260
1261
1262 /**
1263  * Retrieve all fragments of a message ID range.
1264  *
1265  * @see GNUNET_PSYCSTORE_message_get()
1266  *
1267  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1268  */
1269 static int
1270 message_get (void *cls,
1271              const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1272              uint64_t first_message_id,
1273              uint64_t last_message_id,
1274              uint64_t fragment_limit,
1275              uint64_t *returned_fragments,
1276              GNUNET_PSYCSTORE_FragmentCallback cb,
1277              void *cb_cls)
1278 {
1279   struct Plugin *plugin = cls;
1280
1281   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->select_messages;
1282
1283   int ret = GNUNET_SYSERR;
1284   *returned_fragments = 0;
1285
1286   struct GNUNET_MY_QueryParam params_select[] = {
1287     GNUNET_MY_query_param_auto_from_type (channel_key),
1288     GNUNET_MY_query_param_uint64 (&first_message_id),
1289     GNUNET_MY_query_param_uint64 (&last_message_id),
1290     GNUNET_MY_query_param_uint64 (&fragment_limit),
1291     GNUNET_MY_query_param_end
1292   };
1293
1294   ret = fragment_select (plugin, stmt, params_select, returned_fragments, cb, cb_cls);
1295
1296   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
1297   {
1298     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1299               "mysql_stmt_reset", stmt);
1300     return GNUNET_SYSERR;
1301   }
1302
1303   return ret;
1304 }
1305
1306
1307 /**
1308  * Retrieve all fragments of the latest messages.
1309  *
1310  * @see GNUNET_PSYCSTORE_message_get_latest()
1311  *
1312  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1313  */
1314 static int
1315 message_get_latest (void *cls,
1316                     const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1317                     uint64_t message_limit,
1318                     uint64_t *returned_fragments,
1319                     GNUNET_PSYCSTORE_FragmentCallback cb,
1320                     void *cb_cls)
1321 {
1322   struct Plugin *plugin = cls;
1323
1324   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->select_latest_messages;
1325
1326   int ret = GNUNET_SYSERR;
1327   *returned_fragments = 0;
1328
1329   struct GNUNET_MY_QueryParam params_select[] = {
1330     GNUNET_MY_query_param_auto_from_type (channel_key),
1331     GNUNET_MY_query_param_auto_from_type (channel_key),
1332     GNUNET_MY_query_param_uint64 (&message_limit),
1333     GNUNET_MY_query_param_end
1334   };
1335
1336   ret = fragment_select (plugin, stmt, params_select, returned_fragments, cb, cb_cls);
1337
1338   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
1339   {
1340     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1341               "mysql_stmt_reset", stmt);
1342     return GNUNET_SYSERR;
1343   }
1344
1345   return ret;
1346 }
1347
1348
1349 /**
1350  * Retrieve a fragment of message specified by its message ID and fragment
1351  * offset.
1352  *
1353  * @see GNUNET_PSYCSTORE_message_get_fragment()
1354  *
1355  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1356  */
1357 static int
1358 message_get_fragment (void *cls,
1359                       const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1360                       uint64_t message_id,
1361                       uint64_t fragment_offset,
1362                       GNUNET_PSYCSTORE_FragmentCallback cb,
1363                       void *cb_cls)
1364 {
1365   struct Plugin *plugin = cls;
1366
1367   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->select_message_fragment;
1368
1369   int ret = GNUNET_SYSERR;
1370
1371   struct GNUNET_MY_QueryParam params_select[] = {
1372     GNUNET_MY_query_param_auto_from_type (channel_key),
1373     GNUNET_MY_query_param_uint64 (&message_id),
1374     GNUNET_MY_query_param_uint64 (&fragment_offset),
1375     GNUNET_MY_query_param_end
1376   };
1377
1378   if (GNUNET_OK != GNUNET_MY_exec_prepared (plugin->mc,
1379                                             stmt,
1380                                             params_select))
1381   {
1382     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1383               "mysql execute prepared", stmt);
1384     return GNUNET_SYSERR;
1385   }
1386
1387   ret = fragment_row (stmt, cb, cb_cls);
1388
1389   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
1390   {
1391     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1392               "mysql_stmt_reset", stmt);
1393     return GNUNET_SYSERR;
1394   }
1395
1396   return ret;
1397 }
1398
1399 /**
1400  * Retrieve the max. values of message counters for a channel.
1401  *
1402  * @see GNUNET_PSYCSTORE_counters_get()
1403  *
1404  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1405  */
1406 static int
1407 counters_message_get (void *cls,
1408                       const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1409                       uint64_t *max_fragment_id,
1410                       uint64_t *max_message_id,
1411                       uint64_t *max_group_generation)
1412 {
1413   struct Plugin *plugin = cls;
1414
1415   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->select_counters_message;
1416
1417   int ret = GNUNET_SYSERR;
1418
1419   struct GNUNET_MY_QueryParam params_select[] = {
1420     GNUNET_MY_query_param_auto_from_type (channel_key),
1421     GNUNET_MY_query_param_end
1422   };
1423
1424   if (GNUNET_OK != GNUNET_MY_exec_prepared (plugin->mc,
1425                                             stmt,
1426                                             params_select))
1427   {
1428     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1429               "mysql execute prepared", stmt);
1430     return GNUNET_SYSERR;
1431   }
1432
1433   struct GNUNET_MY_ResultSpec results_select[] = {
1434     GNUNET_MY_result_spec_uint64 (max_fragment_id),
1435     GNUNET_MY_result_spec_uint64 (max_message_id),
1436     GNUNET_MY_result_spec_uint64 (max_group_generation),
1437     GNUNET_MY_result_spec_end
1438   };
1439
1440   ret = GNUNET_MY_extract_result (stmt,
1441                                   results_select);
1442
1443   if (GNUNET_OK != ret)
1444   {
1445     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1446               "mysql extract_result", stmt);
1447     return GNUNET_SYSERR;
1448   }
1449
1450   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
1451   {
1452     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1453               "mysql_stmt_reset", stmt);
1454     return GNUNET_SYSERR;
1455   }
1456
1457   return ret;
1458 }
1459
1460 /**
1461  * Retrieve the max. values of state counters for a channel.
1462  *
1463  * @see GNUNET_PSYCSTORE_counters_get()
1464  *
1465  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1466  */
1467 static int
1468 counters_state_get (void *cls,
1469                     const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1470                     uint64_t *max_state_message_id)
1471 {
1472   struct Plugin *plugin = cls;
1473
1474   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->select_counters_state;
1475
1476   int ret = GNUNET_SYSERR;
1477
1478   struct GNUNET_MY_QueryParam params_select[] = {
1479     GNUNET_MY_query_param_auto_from_type (channel_key),
1480     GNUNET_MY_query_param_end
1481   };
1482
1483   if (GNUNET_OK != GNUNET_MY_exec_prepared (plugin->mc,
1484                                             stmt,
1485                                             params_select))
1486   {
1487     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1488               "mysql execute prepared", stmt);
1489     return GNUNET_SYSERR;
1490   }
1491
1492   struct GNUNET_MY_ResultSpec results_select[] = {
1493     GNUNET_MY_result_spec_uint64 (max_state_message_id),
1494     GNUNET_MY_result_spec_end
1495   };
1496
1497   ret = GNUNET_MY_extract_result (stmt,
1498                                   results_select);
1499
1500   if (GNUNET_OK != ret)
1501   {
1502     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1503               "mysql extract_result", stmt);
1504     return GNUNET_SYSERR;
1505   }
1506
1507   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
1508   {
1509     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1510               "mysql_stmt_reset", stmt);
1511     return GNUNET_SYSERR;
1512   }
1513
1514   return ret;
1515 }
1516
1517
1518 /**
1519  * Assign a value to a state variable.
1520  *
1521  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1522  */
1523 static int
1524 state_assign (struct Plugin *plugin, struct GNUNET_MYSQL_StatementHandle *stmt,
1525               const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1526               const char *name, const void *value, size_t value_size)
1527 {
1528   int ret = GNUNET_SYSERR;
1529
1530   struct GNUNET_MY_QueryParam params[] = {
1531     GNUNET_MY_query_param_auto_from_type (channel_key),
1532     GNUNET_MY_query_param_string (name),
1533     GNUNET_MY_query_param_auto_from_type (value),
1534     GNUNET_MY_query_param_end
1535   };
1536
1537   ret = GNUNET_MY_exec_prepared (plugin->mc,
1538                                             stmt,
1539                                             params);
1540
1541   if (GNUNET_OK != ret)
1542   {
1543     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1544               "mysql execute prepared", stmt);
1545     return GNUNET_SYSERR;
1546   }
1547
1548   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
1549   {
1550     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1551               "mysql_stmt_reset", stmt);
1552     return GNUNET_SYSERR;
1553   }
1554
1555   return ret;
1556 }
1557
1558
1559 static int
1560 update_message_id (struct Plugin *plugin, struct GNUNET_MYSQL_StatementHandle *stmt,
1561                    const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1562                    uint64_t message_id)
1563 {
1564   struct GNUNET_MY_QueryParam params[] = {
1565     GNUNET_MY_query_param_uint64 (&message_id),
1566     GNUNET_MY_query_param_auto_from_type (channel_key),
1567     GNUNET_MY_query_param_end
1568   };
1569
1570   if (GNUNET_OK != GNUNET_MY_exec_prepared (plugin->mc,
1571                                             stmt,
1572                                             params))
1573   {
1574     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1575               "mysql execute prepared", stmt);
1576     return GNUNET_SYSERR;
1577   }
1578
1579   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
1580   {
1581     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1582               "mysql_stmt_reset", stmt);
1583     return GNUNET_SYSERR;
1584   }
1585
1586   return GNUNET_OK;
1587 }
1588
1589
1590 /**
1591  * Begin modifying current state.
1592  */
1593 static int
1594 state_modify_begin (void *cls,
1595                     const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1596                     uint64_t message_id, uint64_t state_delta)
1597 {
1598   struct Plugin *plugin = cls;
1599
1600   if (state_delta > 0)
1601   {
1602     /**
1603      * We can only apply state modifiers in the current message if modifiers in
1604      * the previous stateful message (message_id - state_delta) were already
1605      * applied.
1606      */
1607
1608     uint64_t max_state_message_id = 0;
1609     int ret = counters_state_get (plugin, channel_key, &max_state_message_id);
1610     switch (ret)
1611     {
1612     case GNUNET_OK:
1613     case GNUNET_NO: // no state yet
1614       ret = GNUNET_OK;
1615       break;
1616     default:
1617       return ret;
1618     }
1619
1620     if (max_state_message_id < message_id - state_delta)
1621       return GNUNET_NO; /* some stateful messages not yet applied */
1622     else if (message_id - state_delta < max_state_message_id)
1623       return GNUNET_NO; /* changes already applied */
1624   }
1625
1626   if (TRANSACTION_NONE != plugin->transaction)
1627   {
1628     /** @todo FIXME: wait for other transaction to finish  */
1629     return GNUNET_SYSERR;
1630   }
1631   return transaction_begin (plugin, TRANSACTION_STATE_MODIFY);
1632 }
1633
1634
1635 /**
1636  * Set the current value of state variable.
1637  *
1638  * @see GNUNET_PSYCSTORE_state_modify()
1639  *
1640  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1641  */
1642 static int
1643 state_modify_op (void *cls,
1644                  const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1645                  enum GNUNET_PSYC_Operator op,
1646                  const char *name, const void *value, size_t value_size)
1647 {
1648   struct Plugin *plugin = cls;
1649   GNUNET_assert (TRANSACTION_STATE_MODIFY == plugin->transaction);
1650
1651   switch (op)
1652   {
1653   case GNUNET_PSYC_OP_ASSIGN:
1654     return state_assign (plugin, plugin->insert_state_current, channel_key,
1655                          name, value, value_size);
1656
1657   default: /** @todo implement more state operations */
1658     GNUNET_break (0);
1659     return GNUNET_SYSERR;
1660   }
1661 }
1662
1663
1664 /**
1665  * End modifying current state.
1666  */
1667 static int
1668 state_modify_end (void *cls,
1669                   const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1670                   uint64_t message_id)
1671 {
1672   struct Plugin *plugin = cls;
1673   GNUNET_assert (TRANSACTION_STATE_MODIFY == plugin->transaction);
1674
1675   return
1676     GNUNET_OK == exec_channel (plugin, plugin->delete_state_empty, channel_key)
1677     && GNUNET_OK == update_message_id (plugin,
1678                                        plugin->update_max_state_message_id,
1679                                        channel_key, message_id)
1680     && GNUNET_OK == transaction_commit (plugin)
1681     ? GNUNET_OK : GNUNET_SYSERR;
1682 }
1683
1684
1685 /**
1686  * Begin state synchronization.
1687  */
1688 static int
1689 state_sync_begin (void *cls,
1690                   const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
1691 {
1692   struct Plugin *plugin = cls;
1693   return exec_channel (plugin, plugin->delete_state_sync, channel_key);
1694 }
1695
1696
1697 /**
1698  * Assign current value of a state variable.
1699  *
1700  * @see GNUNET_PSYCSTORE_state_modify()
1701  *
1702  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1703  */
1704 static int
1705 state_sync_assign (void *cls,
1706                 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1707                 const char *name, const void *value, size_t value_size)
1708 {
1709   struct Plugin *plugin = cls;
1710   return state_assign (cls, plugin->insert_state_sync, channel_key,
1711                        name, value, value_size);
1712 }
1713
1714
1715 /**
1716  * End modifying current state.
1717  */
1718 static int
1719 state_sync_end (void *cls,
1720                 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1721                 uint64_t max_state_message_id,
1722                 uint64_t state_hash_message_id)
1723 {
1724   struct Plugin *plugin = cls;
1725   int ret = GNUNET_SYSERR;
1726
1727   if (TRANSACTION_NONE != plugin->transaction)
1728   {
1729     /** @todo FIXME: wait for other transaction to finish  */
1730     return GNUNET_SYSERR;
1731   }
1732
1733   GNUNET_OK == transaction_begin (plugin, TRANSACTION_STATE_SYNC)
1734     && GNUNET_OK == exec_channel (plugin, plugin->delete_state, channel_key)
1735     && GNUNET_OK == exec_channel (plugin, plugin->insert_state_from_sync,
1736                                   channel_key)
1737     && GNUNET_OK == exec_channel (plugin, plugin->delete_state_sync,
1738                                   channel_key)
1739     && GNUNET_OK == update_message_id (plugin,
1740                                        plugin->update_state_hash_message_id,
1741                                        channel_key, state_hash_message_id)
1742     && GNUNET_OK == update_message_id (plugin,
1743                                        plugin->update_max_state_message_id,
1744                                        channel_key, max_state_message_id)
1745     && GNUNET_OK == transaction_commit (plugin)
1746     ? ret = GNUNET_OK
1747     : transaction_rollback (plugin);
1748   return ret;
1749 }
1750
1751
1752 /**
1753  * Delete the whole state.
1754  *
1755  * @see GNUNET_PSYCSTORE_state_reset()
1756  *
1757  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1758  */
1759 static int
1760 state_reset (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
1761 {
1762   struct Plugin *plugin = cls;
1763   return exec_channel (plugin, plugin->delete_state, channel_key);
1764 }
1765
1766
1767 /**
1768  * Update signed values of state variables in the state store.
1769  *
1770  * @see GNUNET_PSYCSTORE_state_hash_update()
1771  *
1772  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1773  */
1774 static int
1775 state_update_signed (void *cls,
1776                      const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
1777 {
1778   struct Plugin *plugin = cls;
1779   return exec_channel (plugin, plugin->update_state_signed, channel_key);
1780 }
1781
1782
1783 /**
1784  * Retrieve a state variable by name.
1785  *
1786  * @see GNUNET_PSYCSTORE_state_get()
1787  *
1788  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1789  */
1790 static int
1791 state_get (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1792            const char *name, GNUNET_PSYCSTORE_StateCallback cb, void *cb_cls)
1793 {
1794   struct Plugin *plugin = cls;
1795   int ret = GNUNET_SYSERR;
1796   int sql_ret ;
1797
1798   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->select_state_one;
1799
1800   struct GNUNET_MY_QueryParam params_select[] = {
1801     GNUNET_MY_query_param_auto_from_type (channel_key),
1802     GNUNET_MY_query_param_string (name),
1803     GNUNET_MY_query_param_end
1804   };
1805
1806   void *value_current = NULL;
1807   size_t value_size = 0;
1808
1809   struct GNUNET_MY_ResultSpec results[] = {
1810     GNUNET_MY_result_spec_variable_size (&value_current, &value_size),
1811     GNUNET_MY_result_spec_end
1812   };
1813
1814   GNUNET_MY_exec_prepared (plugin->mc,
1815                           stmt,
1816                           params_select);
1817
1818
1819   sql_ret = GNUNET_MY_extract_result (stmt,
1820                                       results);
1821
1822   switch (sql_ret)
1823   {
1824     case GNUNET_NO:
1825       ret = GNUNET_NO;
1826       break;
1827     case GNUNET_YES:
1828       ret = cb (cb_cls, name, value_current,
1829                 value_size);
1830       break;
1831     default:
1832       LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1833               "mysql extract_result", stmt);
1834   }
1835
1836   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
1837   {
1838     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1839               "mysql_stmt_reset", stmt);
1840     return GNUNET_SYSERR;
1841   }
1842
1843   return ret;
1844 }
1845
1846
1847 /**
1848  * Retrieve all state variables for a channel with the given prefix.
1849  *
1850  * @see GNUNET_PSYCSTORE_state_get_prefix()
1851  *
1852  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1853  */
1854 static int
1855 state_get_prefix (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1856                   const char *name, GNUNET_PSYCSTORE_StateCallback cb,
1857                   void *cb_cls)
1858 {
1859   struct Plugin *plugin = cls;
1860   int ret = GNUNET_SYSERR;
1861
1862   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->select_state_prefix;
1863
1864   uint32_t name_len = (uint32_t) strlen (name);
1865
1866   struct GNUNET_MY_QueryParam params_select[] = {
1867     GNUNET_MY_query_param_auto_from_type (channel_key),
1868     GNUNET_MY_query_param_string (name),
1869     GNUNET_MY_query_param_uint32 (&name_len),
1870     GNUNET_MY_query_param_string (name),
1871     GNUNET_MY_query_param_end
1872   };
1873
1874   char *name2 = "";
1875   void *value_current = NULL;
1876   size_t value_size = 0;
1877
1878   struct GNUNET_MY_ResultSpec results[] = {
1879     GNUNET_MY_result_spec_string (&name2),
1880     GNUNET_MY_result_spec_variable_size (&value_current, &value_size),
1881     GNUNET_MY_result_spec_end
1882   };
1883
1884   int sql_ret;
1885
1886   do
1887   {
1888     GNUNET_MY_exec_prepared (plugin->mc,
1889                             stmt,
1890                             params_select);
1891     sql_ret = GNUNET_MY_extract_result (stmt,
1892                                         results);
1893     switch (sql_ret)
1894     {
1895       case GNUNET_NO:
1896         if (ret != GNUNET_OK)
1897           ret = GNUNET_NO;
1898         break;
1899       case GNUNET_YES:
1900         ret = cb (cb_cls, (const char *) name2,
1901                   value_current,
1902                   value_size);
1903
1904         if (ret != GNUNET_YES)
1905           sql_ret = GNUNET_NO;
1906         break;
1907       default:
1908         LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1909               "mysql extract_result", stmt);
1910     }
1911   }
1912   while (sql_ret == GNUNET_YES);
1913
1914   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
1915   {
1916     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1917               "mysql_stmt_reset", stmt);
1918     return GNUNET_SYSERR;
1919   }
1920
1921   return ret;
1922 }
1923
1924
1925 /**
1926  * Retrieve all signed state variables for a channel.
1927  *
1928  * @see GNUNET_PSYCSTORE_state_get_signed()
1929  *
1930  * @return #GNUNET_OK on success, else #GNUNET_SYSERR
1931  */
1932 static int
1933 state_get_signed (void *cls,
1934                   const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1935                   GNUNET_PSYCSTORE_StateCallback cb, void *cb_cls)
1936 {
1937   struct Plugin *plugin = cls;
1938   int ret = GNUNET_SYSERR;
1939
1940   struct GNUNET_MYSQL_StatementHandle *stmt = plugin->select_state_signed;
1941
1942   struct GNUNET_MY_QueryParam params_select[] = {
1943     GNUNET_MY_query_param_auto_from_type (channel_key),
1944     GNUNET_MY_query_param_end
1945   };
1946
1947   int sql_ret;
1948
1949   char *name = "";
1950   void *value_signed = NULL;
1951   size_t value_size = 0;
1952
1953   struct GNUNET_MY_ResultSpec results[] = {
1954     GNUNET_MY_result_spec_string (&name),
1955     GNUNET_MY_result_spec_variable_size (&value_signed, &value_size),
1956     GNUNET_MY_result_spec_end
1957   };
1958
1959   do
1960   {
1961     GNUNET_MY_exec_prepared (plugin->mc,
1962                              stmt,
1963                              params_select);
1964     sql_ret = GNUNET_MY_extract_result (stmt,
1965                                         results);
1966
1967     switch (sql_ret)
1968     {
1969       case GNUNET_NO:
1970         if (ret != GNUNET_OK)
1971           ret = GNUNET_NO;
1972         break;
1973       case GNUNET_YES:
1974         ret = cb (cb_cls, (const char *) name,
1975                   value_signed,
1976                   value_size);
1977
1978         if (ret != GNUNET_YES)
1979             sql_ret = GNUNET_NO;
1980         break;
1981       default:
1982          LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1983               "mysql extract_result", stmt);
1984     }
1985   }
1986   while (sql_ret == GNUNET_YES);
1987
1988   if (0 != mysql_stmt_reset (GNUNET_MYSQL_statement_get_stmt (stmt)))
1989   {
1990     LOG_MYSQL(plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1991               "mysql_stmt_reset", stmt);
1992     return GNUNET_SYSERR;
1993   }
1994
1995   return ret;
1996 }
1997
1998
1999 /**
2000  * Entry point for the plugin.
2001  *
2002  * @param cls The struct GNUNET_CONFIGURATION_Handle.
2003  * @return NULL on error, otherwise the plugin context
2004  */
2005 void *
2006 libgnunet_plugin_psycstore_mysql_init (void *cls)
2007 {
2008   static struct Plugin plugin;
2009   const struct GNUNET_CONFIGURATION_Handle *cfg = cls;
2010   struct GNUNET_PSYCSTORE_PluginFunctions *api;
2011
2012   if (NULL != plugin.cfg)
2013     return NULL;                /* can only initialize once! */
2014   memset (&plugin, 0, sizeof (struct Plugin));
2015   plugin.cfg = cfg;
2016   if (GNUNET_OK != database_setup (&plugin))
2017   {
2018     database_shutdown (&plugin);
2019     return NULL;
2020   }
2021   api = GNUNET_new (struct GNUNET_PSYCSTORE_PluginFunctions);
2022   api->cls = &plugin;
2023   api->membership_store = &mysql_membership_store;
2024   api->membership_test = &membership_test;
2025   api->fragment_store = &fragment_store;
2026   api->message_add_flags = &message_add_flags;
2027   api->fragment_get = &fragment_get;
2028   api->fragment_get_latest = &fragment_get_latest;
2029   api->message_get = &message_get;
2030   api->message_get_latest = &message_get_latest;
2031   api->message_get_fragment = &message_get_fragment;
2032   api->counters_message_get = &counters_message_get;
2033   api->counters_state_get = &counters_state_get;
2034   api->state_modify_begin = &state_modify_begin;
2035   api->state_modify_op = &state_modify_op;
2036   api->state_modify_end = &state_modify_end;
2037   api->state_sync_begin = &state_sync_begin;
2038   api->state_sync_assign = &state_sync_assign;
2039   api->state_sync_end = &state_sync_end;
2040   api->state_reset = &state_reset;
2041   api->state_update_signed = &state_update_signed;
2042   api->state_get = &state_get;
2043   api->state_get_prefix = &state_get_prefix;
2044   api->state_get_signed = &state_get_signed;
2045
2046   LOG (GNUNET_ERROR_TYPE_INFO, _("Mysql database running\n"));
2047   return api;
2048 }
2049
2050
2051 /**
2052  * Exit point from the plugin.
2053  *
2054  * @param cls The plugin context (as returned by "init")
2055  * @return Always NULL
2056  */
2057 void *
2058 libgnunet_plugin_psycstore_mysql_done (void *cls)
2059 {
2060   struct GNUNET_PSYCSTORE_PluginFunctions *api = cls;
2061   struct Plugin *plugin = api->cls;
2062
2063   database_shutdown (plugin);
2064   plugin->cfg = NULL;
2065   GNUNET_free (api);
2066   LOG (GNUNET_ERROR_TYPE_DEBUG, "Mysql plugin is finished\n");
2067   return NULL;
2068 }
2069
2070 /* end of plugin_psycstore_mysql.c */