work on #9303: make aggregator logic more asynchronous to support running AML program, move shared logic into exchangedb

7 files changed, 915 insertions, 608 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index b039fd98c..3af49638d 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -25,8 +25,8 @@ SUBDIRS = \
   mhd \
   templating \
   bank-lib \
-  exchangedb \
   kyclogic \
+  exchangedb \
   exchange \
   auditordb \
   auditor \
diff --git a/src/exchange/taler-exchange-aggregator.c b/src/exchange/taler-exchange-aggregator.c
index bf13325f7..eb117800e 100644
--- a/src/exchange/taler-exchange-aggregator.c
+++ b/src/exchange/taler-exchange-aggregator.c
@@ -98,6 +98,21 @@ struct AggregationUnit
   const struct TALER_EXCHANGEDB_AccountInfo *wa;
 
   /**
+   * Handle for asynchronously running AML program.
+   */
+  struct TALER_KYCLOGIC_AmlProgramRunnerHandle *amlh;
+
+  /**
+   * Shard this aggregation unit is part of.
+   */
+  struct Shard *shard;
+
+  /**
+   * Currently active rule set.
+   */
+  struct TALER_KYCLOGIC_LegitimizationRuleSet *lrs;
+
+  /**
    * Row in KYC table for legitimization requirements
    * that are pending for this aggregation, or 0 if none.
    */
@@ -180,6 +195,11 @@ static int kyc_off;
 static const struct GNUNET_CONFIGURATION_Handle *cfg;
 
 /**
+ * Key used to encrypt KYC attribute data in our database.
+ */
+static struct TALER_AttributeEncryptionKeyP attribute_key;
+
+/**
  * Our database plugin.
  */
 static struct TALER_EXCHANGEDB_Plugin *db_plugin;
@@ -232,18 +252,23 @@ drain_kyc_alerts (void *cls);
 
 
 /**
- * Free data stored in @a au, but not @a au itself (stack allocated).
+ * Free data stored in @a au, including @a au itself.
  *
- * @param au aggregation unit to clean up
+ * @param[in] au aggregation unit to clean up
  */
 static void
 cleanup_au (struct AggregationUnit *au)
 {
   GNUNET_assert (NULL != au);
+  if (NULL != au->amlh)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                "Aborting AML program during aggregation cleanup\n");
+    TALER_KYCLOGIC_run_aml_program_cancel (au->amlh);
+    au->amlh = NULL;
+  }
   GNUNET_free (au->payto_uri.full_payto);
-  memset (au,
-          0,
-          sizeof (*au));
+  GNUNET_free (au);
 }
 
 
@@ -293,7 +318,29 @@ parse_aggregator_config (void)
     return GNUNET_SYSERR;
   }
   if (GNUNET_NO == enable_kyc)
+  {
     kyc_off = true;
+  }
+  else
+  {
+    char *attr_enc_key_str;
+
+    if (GNUNET_OK !=
+        GNUNET_CONFIGURATION_get_value_string (cfg,
+                                               "exchange",
+                                               "ATTRIBUTE_ENCRYPTION_KEY",
+                                               &attr_enc_key_str))
+    {
+      GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
+                                 "exchange",
+                                 "ATTRIBUTE_ENCRYPTION_KEY");
+      return GNUNET_SYSERR;
+    }
+    GNUNET_CRYPTO_hash (attr_enc_key_str,
+                        strlen (attr_enc_key_str),
+                        &attribute_key.hash);
+    GNUNET_free (attr_enc_key_str);
+  }
   if (GNUNET_OK !=
       GNUNET_CONFIGURATION_get_value_string (cfg,
                                              "exchange",
@@ -408,28 +455,206 @@ release_shard (struct Shard *s)
 
 
 /**
- * Trigger the wire transfer for the @a au_active
- * and delete the record of the aggregation.
+ * Callback to return all applicable amounts for the KYC
+ * decision to @ a cb.
  *
- * @param au_active information about the aggregation
+ * @param cls a `struct AggregationUnit *`
+ * @param limit time limit for the iteration
+ * @param cb function to call with the amounts
+ * @param cb_cls closure for @a cb
+ * @return transaction status
  */
 static enum GNUNET_DB_QueryStatus
-trigger_wire_transfer (const struct AggregationUnit *au_active)
+return_relevant_amounts (void *cls,
+                         struct GNUNET_TIME_Absolute limit,
+                         TALER_EXCHANGEDB_KycAmountCallback cb,
+                         void *cb_cls)
+{
+  const struct AggregationUnit *au = cls;
+  enum GNUNET_DB_QueryStatus qs;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Returning amount %s in KYC check\n",
+              TALER_amount2s (&au->total_amount));
+  if (GNUNET_OK !=
+      cb (cb_cls,
+          &au->total_amount,
+          GNUNET_TIME_absolute_get ()))
+    return GNUNET_DB_STATUS_SUCCESS_NO_RESULTS;
+  qs = db_plugin->select_aggregation_amounts_for_kyc_check (
+    db_plugin->cls,
+    &au->h_normalized_payto,
+    limit,
+    cb,
+    cb_cls);
+  if (GNUNET_DB_STATUS_HARD_ERROR == qs)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "Failed to select aggregation amounts for KYC limit check!\n");
+  }
+  return qs;
+}
+
+
+/**
+ * The aggregation process failed hard, shut down the program.
+ *
+ * @param[in] au aggregation that failed hard
+ */
+static void
+fail_aggregation (struct AggregationUnit *au)
+{
+  struct Shard *s = au->shard;
+
+  cleanup_au (au);
+  global_ret = EXIT_FAILURE;
+  GNUNET_SCHEDULER_shutdown ();
+  db_plugin->rollback (db_plugin->cls);
+  release_shard (s);
+}
+
+
+/**
+ * The aggregation process failed with a serialization
+ * issue.  Rollback the transaction and try again.
+ *
+ * @param[in] au aggregation that needs to be rolled back
+ */
+static void
+rollback_aggregation (struct AggregationUnit *au)
+{
+  struct Shard *s = au->shard;
+
+  cleanup_au (au);
+  db_plugin->rollback (db_plugin->cls);
+  GNUNET_assert (NULL == task);
+  task = GNUNET_SCHEDULER_add_now (&run_aggregation,
+                                   s);
+}
+
+
+/**
+ * The aggregation process succeeded and should be finally committed.
+ *
+ * @param[in] au aggregation that needs to be committed
+ */
+static void
+commit_aggregation (struct AggregationUnit *au)
+{
+  struct Shard *s = au->shard;
+
+  cleanup_au (au);
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Committing aggregation result\n");
+
+  /* Now we can finally commit the overall transaction, as we are
+     again consistent if all of this passes. */
+  switch (commit_or_warn ())
+  {
+  case GNUNET_DB_STATUS_SOFT_ERROR:
+    /* try again */
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                "Serialization issue on commit; trying again later!\n");
+    GNUNET_assert (NULL == task);
+    task = GNUNET_SCHEDULER_add_now (&run_aggregation,
+                                     s);
+    return;
+  case GNUNET_DB_STATUS_HARD_ERROR:
+    GNUNET_break (0);
+    global_ret = EXIT_FAILURE;
+    GNUNET_SCHEDULER_shutdown ();
+    db_plugin->rollback (db_plugin->cls); /* just in case */
+    release_shard (s);
+    return;
+  case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                "Commit complete, going again\n");
+    GNUNET_assert (NULL == task);
+    task = GNUNET_SCHEDULER_add_now (&run_aggregation,
+                                     s);
+    return;
+  default:
+    GNUNET_break (0);
+    global_ret = EXIT_FAILURE;
+    GNUNET_SCHEDULER_shutdown ();
+    db_plugin->rollback (db_plugin->cls); /* just in case */
+    release_shard (s);
+    return;
+  }
+}
+
+
+/**
+ * The aggregation process could not be concluded and its progress state
+ * should be remembered in a transient aggregation.
+ *
+ * @param[in] au aggregation that needs to be committed
+ *     into a transient aggregation
+ */
+static void
+commit_to_transient (struct AggregationUnit *au)
+{
+  enum GNUNET_DB_QueryStatus qs;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Not ready for wire transfer (%d/%s)\n",
+              qs,
+              TALER_amount2s (&au->final_amount));
+  if (au->have_transient)
+    qs = db_plugin->update_aggregation_transient (db_plugin->cls,
+                                                  &au->h_full_payto,
+                                                  &au->wtid,
+                                                  au->requirement_row,
+                                                  &au->total_amount);
+  else
+    qs = db_plugin->create_aggregation_transient (db_plugin->cls,
+                                                  &au->h_full_payto,
+                                                  au->wa->section_name,
+                                                  &au->merchant_pub,
+                                                  &au->wtid,
+                                                  au->requirement_row,
+                                                  &au->total_amount);
+  if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                "Serialization issue, trying again later!\n");
+    rollback_aggregation (au);
+    return;
+  }
+  if (GNUNET_DB_STATUS_HARD_ERROR == qs)
+  {
+    GNUNET_break (0);
+    fail_aggregation (au);
+    return;
+  }
+  /* commit */
+  commit_aggregation (au);
+}
+
+
+/**
+ * Trigger the wire transfer for the @a au
+ * and delete the record of the aggregation.
+ *
+ * @param[in] au information about the aggregation
+ */
+static void
+trigger_wire_transfer (struct AggregationUnit *au)
 {
   enum GNUNET_DB_QueryStatus qs;
 
   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
               "Preparing wire transfer of %s to %s\n",
-              TALER_amount2s (&au_active->final_amount),
-              TALER_B2S (&au_active->merchant_pub));
+              TALER_amount2s (&au->final_amount),
+              TALER_B2S (&au->merchant_pub));
   {
     void *buf;
     size_t buf_size;
 
-    TALER_BANK_prepare_transfer (au_active->payto_uri,
-                                 &au_active->final_amount,
+    TALER_BANK_prepare_transfer (au->payto_uri,
+                                 &au->final_amount,
                                  exchange_base_url,
-                                 &au_active->wtid,
+                                 &au->wtid,
                                  &buf,
                                  &buf_size);
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -437,7 +662,7 @@ trigger_wire_transfer (const struct AggregationUnit *au_active)
                 (unsigned int) buf_size);
     /* Commit our intention to execute the wire transfer! */
     qs = db_plugin->wire_prepare_data_insert (db_plugin->cls,
-                                              au_active->wa->method,
+                                              au->wa->method,
                                               buf,
                                               buf_size);
     GNUNET_free (buf);
@@ -446,104 +671,212 @@ trigger_wire_transfer (const struct AggregationUnit *au_active)
   if (qs >= 0)
     qs = db_plugin->store_wire_transfer_out (
       db_plugin->cls,
-      au_active->execution_time,
-      &au_active->wtid,
-      &au_active->h_full_payto,
-      au_active->wa->section_name,
-      &au_active->final_amount);
+      au->execution_time,
+      &au->wtid,
+      &au->h_full_payto,
+      au->wa->section_name,
+      &au->final_amount);
 
   if ( (qs >= 0) &&
-       au_active->have_transient)
+       au->have_transient)
     qs = db_plugin->delete_aggregation_transient (
       db_plugin->cls,
-      &au_active->h_full_payto,
-      &au_active->wtid);
-  return qs;
+      &au->h_full_payto,
+      &au->wtid);
+
+  switch (qs)
+  {
+  case GNUNET_DB_STATUS_SOFT_ERROR:
+    GNUNET_log (
+      GNUNET_ERROR_TYPE_INFO,
+      "Serialization issue during aggregation; trying again later!\n");
+    rollback_aggregation (au);
+    return;
+  case GNUNET_DB_STATUS_HARD_ERROR:
+    GNUNET_break (0);
+    fail_aggregation (au);
+    return;
+  default:
+    break;
+  }
+  {
+    struct TALER_CoinDepositEventP rep = {
+      .header.size = htons (sizeof (rep)),
+      .header.type = htons (TALER_DBEVENT_EXCHANGE_DEPOSIT_STATUS_CHANGED),
+      .merchant_pub = au->merchant_pub
+    };
+
+    db_plugin->event_notify (db_plugin->cls,
+                             &rep.header,
+                             NULL,
+                             0);
+  }
+  commit_aggregation (au);
 }
 
 
 /**
- * Callback to return all applicable amounts for the KYC
- * decision to @ a cb.
+ * Function called with legitimization rule set. Check
+ * how that affects the aggregation process.
+ *
+ * @param[in] au active aggregation
+ * @param[in] lrs legitimization rule set to evaluate, NULL for defaults
+ */
+static void
+evaluate_lrs (
+  struct AggregationUnit *au,
+  struct TALER_KYCLOGIC_LegitimizationRuleSet *lrs);
+
+
+/**
+ * Function called after AML program was run.  Decides how to
+ * continue with the aggregation based on the AML result.
  *
  * @param cls a `struct AggregationUnit *`
- * @param limit time limit for the iteration
- * @param cb function to call with the amounts
- * @param cb_cls closure for @a cb
- * @return transaction status
+ * @param apr result of the AML program.
  */
-static enum GNUNET_DB_QueryStatus
-return_relevant_amounts (void *cls,
-                         struct GNUNET_TIME_Absolute limit,
-                         TALER_EXCHANGEDB_KycAmountCallback cb,
-                         void *cb_cls)
+static void
+aml_result_callback (
+  void *cls,
+  const struct TALER_KYCLOGIC_AmlProgramResult *apr);
+
+
+/**
+ * Run the given measure @a m of the @a lrs for the
+ * given aggregation process @a au.
+ *
+ * @param lrs a legitimization rule set containing @a m
+ * @param m measure to run
+ * @param au aggregation unit we are processing
+ */
+static void
+run_measure (
+  struct TALER_KYCLOGIC_LegitimizationRuleSet *lrs,
+  const struct TALER_KYCLOGIC_Measure *m,
+  struct AggregationUnit *au)
 {
-  const struct AggregationUnit *au_active = cls;
-  enum GNUNET_DB_QueryStatus qs;
+  if ( (NULL == m->check_name) ||
+       (0 ==
+        strcasecmp ("skip",
+                    m->check_name)) )
+  {
+    struct TALER_EXCHANGEDB_HistoryBuilderContext hbc = {
+      .account = &au->h_normalized_payto,
+      .db_plugin = db_plugin,
+      .attribute_key = &attribute_key
+    };
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-              "Returning amount %s in KYC check\n",
-              TALER_amount2s (&au_active->total_amount));
-  if (GNUNET_OK !=
-      cb (cb_cls,
-          &au_active->total_amount,
-          GNUNET_TIME_absolute_get ()))
-    return GNUNET_DB_STATUS_SUCCESS_NO_RESULTS;
-  qs = db_plugin->select_aggregation_amounts_for_kyc_check (
-    db_plugin->cls,
-    &au_active->h_normalized_payto,
-    limit,
-    cb,
-    cb_cls);
-  if (GNUNET_DB_STATUS_HARD_ERROR == qs)
+    au->lrs = lrs;
+    au->amlh = TALER_KYCLOGIC_run_aml_program3 (
+      m,
+      NULL /* no attributes */,
+      &TALER_EXCHANGEDB_current_rule_builder,
+      &hbc,
+      &TALER_EXCHANGEDB_aml_history_builder,
+      &hbc,
+      &TALER_EXCHANGEDB_kyc_history_builder,
+      &hbc,
+      &aml_result_callback,
+      au);
+    return;
+  }
+  /* User MUST pass interactive check (odd): we cannot continue here */
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "Fallback measure %s involves check %s, blocking aggreation\n",
+              m->measure_name,
+              m->check_name);
+  TALER_KYCLOGIC_rules_free (lrs);
+  commit_to_transient (au);
+}
+
+
+static void
+aml_result_callback (
+  void *cls,
+  const struct TALER_KYCLOGIC_AmlProgramResult *apr)
+{
+  struct AggregationUnit *au = cls;
+
+  au->amlh = NULL;
+  // FIXME: database update based on result!
+  switch (apr->status)
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                "Failed to select aggregation amounts for KYC limit check!\n");
+  case TALER_KYCLOGIC_AMLR_SUCCESS:
+    {
+      struct TALER_KYCLOGIC_LegitimizationRuleSet *lrs;
+
+      TALER_KYCLOGIC_rules_free (au->lrs);
+      au->lrs = NULL;
+      lrs = TALER_KYCLOGIC_rules_parse (apr->details.success.new_rules);
+      GNUNET_break (NULL != lrs);
+      /* Fall back to default rules on parse error! */
+      evaluate_lrs (au,
+                    lrs);
+      return;
+    }
+  case TALER_KYCLOGIC_AMLR_FAILURE:
+    {
+      struct TALER_KYCLOGIC_LegitimizationRuleSet *lrs = au->lrs;
+      const char *fmn = apr->details.failure.fallback_measure;
+      const struct TALER_KYCLOGIC_Measure *m;
+
+      au->lrs = NULL;
+      m = TALER_KYCLOGIC_get_measure (lrs,
+                                      fmn);
+      if (NULL == m)
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                    "Fallback measure `%s' does not exist (anymore?).\n",
+                    fmn);
+        TALER_KYCLOGIC_rules_free (lrs);
+        trigger_wire_transfer (au);
+        return;
+      }
+      run_measure (lrs,
+                   m,
+                   au);
+      return;
+    }
   }
-  return qs;
+  /* This should be impossible */
+  GNUNET_assert (0);
 }
 
 
 /**
- * Test if legitimization rules are satisfied for a transfer to @a h_payto.
+ * Function called with legitimization rule set. Check
+ * how that affects the aggregation process.
  *
- * @param[in,out] au_active aggregation unit to check for
- * @return true if KYC checks are satisfied
+ * @param[in] au active aggregation
+ * @param[in] lrs legitimization rule set to evaluate, NULL for defaults
  */
-static bool
-legitimization_satisfied (struct AggregationUnit *au_active)
+static void
+evaluate_lrs (
+  struct AggregationUnit *au,
+  struct TALER_KYCLOGIC_LegitimizationRuleSet *lrs)
 {
-  struct TALER_KYCLOGIC_LegitimizationRuleSet *lrs = NULL;
-  const struct TALER_KYCLOGIC_KycRule *requirement;
   enum GNUNET_DB_QueryStatus qs;
-  json_t *jrule;
-
-  if (kyc_off)
-  {
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                "KYC checks are off, legitimization satisfied\n");
-    return true;
-  }
+  const struct TALER_KYCLOGIC_KycRule *requirement;
 
+  if ( (NULL != lrs) &&
+       GNUNET_TIME_absolute_is_past
+         (TALER_KYCLOGIC_rules_get_expiration (lrs).abs_time) )
   {
-    json_t *jrules;
+    const struct TALER_KYCLOGIC_Measure *m;
 
-    qs = db_plugin->get_kyc_rules2 (db_plugin->cls,
-                                    &au_active->h_normalized_payto,
-                                    &jrules);
-    if (qs < 0)
-    {
-      GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
-      return false;
-    }
-    if (qs > 0)
+    m = TALER_KYCLOGIC_rules_get_successor (lrs);
+    if (NULL != m)
     {
-      lrs = TALER_KYCLOGIC_rules_parse (jrules);
-      GNUNET_break (NULL != lrs);
-      /* Fall back to default rules on parse error! */
-      json_decref (jrules);
+      run_measure (lrs,
+                   m,
+                   au);
+      return;
     }
+    /* fall back to default rules */
+    TALER_KYCLOGIC_rules_free (lrs);
+    lrs = NULL;
   }
+
   {
     struct TALER_Amount next_threshold;
 
@@ -551,7 +884,7 @@ legitimization_satisfied (struct AggregationUnit *au_active)
       TALER_KYCLOGIC_KYC_TRIGGER_AGGREGATE,
       lrs,
       &return_relevant_amounts,
-      (void *) au_active,
+      (void *) au,
       &requirement,
       &next_threshold);
   }
@@ -559,43 +892,91 @@ legitimization_satisfied (struct AggregationUnit *au_active)
   {
     TALER_KYCLOGIC_rules_free (lrs);
     GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
-    return false;
+    rollback_aggregation (au);
+    return;
   }
   if (NULL == requirement)
   {
     TALER_KYCLOGIC_rules_free (lrs);
-    return true;
+    commit_aggregation (au);
+    return;
   }
   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
               "KYC requirement for %s is %s\n",
-              TALER_amount2s (&au_active->total_amount),
+              TALER_amount2s (&au->total_amount),
               TALER_KYCLOGIC_rule2s (requirement));
-  jrule = TALER_KYCLOGIC_rule_to_measures (requirement);
-  qs = db_plugin->trigger_kyc_rule_for_account (
-    db_plugin->cls,
-    au_active->payto_uri,
-    &au_active->h_normalized_payto,
-    NULL,
-    &au_active->merchant_pub,
-    jrule,
-    TALER_KYCLOGIC_rule2priority (requirement),
-    &au_active->requirement_row,
-    &au_active->bad_kyc_auth);
-  json_decref (jrule);
+  {
+    json_t *jrule;
+
+    jrule = TALER_KYCLOGIC_rule_to_measures (requirement);
+    qs = db_plugin->trigger_kyc_rule_for_account (
+      db_plugin->cls,
+      au->payto_uri,
+      &au->h_normalized_payto,
+      NULL,
+      &au->merchant_pub,
+      jrule,
+      TALER_KYCLOGIC_rule2priority (requirement),
+      &au->requirement_row,
+      &au->bad_kyc_auth);
+    json_decref (jrule);
+  }
   if (qs < 0)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                 "Failed to persist KYC requirement `%s' in DB!\n",
                 TALER_KYCLOGIC_rule2s (requirement));
+    rollback_aggregation (au);
+    return;
   }
-  else
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Legitimization process %llu started\n",
+              (unsigned long long) au->requirement_row);
+  TALER_KYCLOGIC_rules_free (lrs);
+  commit_to_transient (au);
+}
+
+
+/**
+ * Test if legitimization rules are satisfied for a transfer to @a h_payto.
+ *
+ * @param[in] au aggregation unit to check for
+ */
+static void
+check_legitimization_satisfied (struct AggregationUnit *au)
+{
+  struct TALER_KYCLOGIC_LegitimizationRuleSet *lrs = NULL;
+  enum GNUNET_DB_QueryStatus qs;
+
+  if (kyc_off)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                "Legitimization process %llu started\n",
-                (unsigned long long) au_active->requirement_row);
+                "KYC checks are off, legitimization satisfied\n");
+    trigger_wire_transfer (au);
+    return;
   }
-  TALER_KYCLOGIC_rules_free (lrs);
-  return false;
+  {
+    json_t *jrules;
+
+    qs = db_plugin->get_kyc_rules2 (db_plugin->cls,
+                                    &au->h_normalized_payto,
+                                    &jrules);
+    if (qs < 0)
+    {
+      GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+      rollback_aggregation (au);
+      return;
+    }
+    if (qs > 0)
+    {
+      lrs = TALER_KYCLOGIC_rules_parse (jrules);
+      GNUNET_break (NULL != lrs);
+      /* Fall back to default rules on parse error! */
+      json_decref (jrules);
+    }
+  }
+  evaluate_lrs (au,
+                lrs);
 }
 
 
@@ -609,7 +990,7 @@ legitimization_satisfied (struct AggregationUnit *au_active)
  *         #GNUNET_NO to rollback and try again (serialization issue)
  *         #GNUNET_SYSERR hard error, terminate aggregator process
  */
-static enum GNUNET_GenericReturnValue
+static void
 do_aggregate (struct AggregationUnit *au)
 {
   enum GNUNET_DB_QueryStatus qs;
@@ -622,7 +1003,8 @@ do_aggregate (struct AggregationUnit *au)
                 "No exchange account configured for `%s', please fix your setup to continue!\n",
                 au->payto_uri.full_payto);
     global_ret = EXIT_FAILURE;
-    return GNUNET_SYSERR;
+    fail_aggregation (au);
+    return;
   }
 
   {
@@ -643,8 +1025,8 @@ do_aggregate (struct AggregationUnit *au)
                   "Could not get wire fees for %s at %s. Aborting run.\n",
                   au->wa->method,
                   GNUNET_TIME_timestamp2s (au->execution_time));
-      global_ret = EXIT_FAILURE;
-      return GNUNET_SYSERR;
+      fail_aggregation (au);
+      return;
     }
   }
 
@@ -664,13 +1046,14 @@ do_aggregate (struct AggregationUnit *au)
   case GNUNET_DB_STATUS_HARD_ERROR:
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                 "Failed to lookup transient aggregates!\n");
-    global_ret = EXIT_FAILURE;
-    return GNUNET_SYSERR;
+    fail_aggregation (au);
+    return;
   case GNUNET_DB_STATUS_SOFT_ERROR:
     /* serializiability issue, try again */
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                 "Serialization issue, trying again later!\n");
-    return GNUNET_NO;
+    rollback_aggregation (au);
+    return;
   case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
     GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
                                 &au->wtid,
@@ -696,15 +1079,16 @@ do_aggregate (struct AggregationUnit *au)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                 "Failed to execute aggregation!\n");
-    global_ret = EXIT_FAILURE;
-    return GNUNET_SYSERR;
+    fail_aggregation (au);
+    return;
   }
   if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
   {
     /* serializiability issue, try again */
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                 "Serialization issue, trying again later!\n");
-    return GNUNET_NO;
+    rollback_aggregation (au);
+    return;
   }
   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
               "Aggregation total is %s.\n",
@@ -729,72 +1113,12 @@ do_aggregate (struct AggregationUnit *au)
        (GNUNET_SYSERR ==
         TALER_amount_round_down (&au->final_amount,
                                  &currency_round_unit)) ||
-       (TALER_amount_is_zero (&au->final_amount)) ||
-       (! legitimization_satisfied (au)) )
+       (TALER_amount_is_zero (&au->final_amount)) )
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                "Not ready for wire transfer (%d/%s)\n",
-                qs,
-                TALER_amount2s (&au->final_amount));
-    if (au->have_transient)
-      qs = db_plugin->update_aggregation_transient (db_plugin->cls,
-                                                    &au->h_full_payto,
-                                                    &au->wtid,
-                                                    au->requirement_row,
-                                                    &au->total_amount);
-    else
-      qs = db_plugin->create_aggregation_transient (db_plugin->cls,
-                                                    &au->h_full_payto,
-                                                    au->wa->section_name,
-                                                    &au->merchant_pub,
-                                                    &au->wtid,
-                                                    au->requirement_row,
-                                                    &au->total_amount);
-    if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
-    {
-      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                  "Serialization issue, trying again later!\n");
-      return GNUNET_NO;
-    }
-    if (GNUNET_DB_STATUS_HARD_ERROR == qs)
-    {
-      GNUNET_break (0);
-      global_ret = EXIT_FAILURE;
-      return GNUNET_SYSERR;
-    }
-    /* commit */
-    return GNUNET_OK;
-  }
-
-  qs = trigger_wire_transfer (au);
-  switch (qs)
-  {
-  case GNUNET_DB_STATUS_SOFT_ERROR:
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                "Serialization issue during aggregation; trying again later!\n")
-    ;
-    return GNUNET_NO;
-  case GNUNET_DB_STATUS_HARD_ERROR:
-    GNUNET_break (0);
-    global_ret = EXIT_FAILURE;
-    return GNUNET_SYSERR;
-  default:
-    break;
-  }
-  {
-    struct TALER_CoinDepositEventP rep = {
-      .header.size = htons (sizeof (rep)),
-      .header.type = htons (TALER_DBEVENT_EXCHANGE_DEPOSIT_STATUS_CHANGED),
-      .merchant_pub = au->merchant_pub
-    };
-
-    db_plugin->event_notify (db_plugin->cls,
-                             &rep.header,
-                             NULL,
-                             0);
+    commit_to_transient (au);
+    return;
   }
-  return GNUNET_OK;
-
+  check_legitimization_satisfied (au);
 }
 
 
@@ -802,18 +1126,16 @@ static void
 run_aggregation (void *cls)
 {
   struct Shard *s = cls;
-  struct AggregationUnit au_active;
+  struct AggregationUnit *au;
   enum GNUNET_DB_QueryStatus qs;
-  enum GNUNET_GenericReturnValue ret;
 
   task = NULL;
   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
               "Checking for ready deposits to aggregate\n");
   /* make sure we have current fees */
-  memset (&au_active,
-          0,
-          sizeof (au_active));
-  au_active.execution_time = GNUNET_TIME_timestamp_get ();
+  au = GNUNET_new (struct AggregationUnit);
+  au->execution_time = GNUNET_TIME_timestamp_get ();
+  au->shard = s;
   if (GNUNET_OK !=
       db_plugin->start_deferred_wire_out (db_plugin->cls))
   {
@@ -828,12 +1150,12 @@ run_aggregation (void *cls)
     db_plugin->cls,
     s->shard_start,
     s->shard_end,
-    &au_active.merchant_pub,
-    &au_active.payto_uri);
+    &au->merchant_pub,
+    &au->payto_uri);
   switch (qs)
   {
   case GNUNET_DB_STATUS_HARD_ERROR:
-    cleanup_au (&au_active);
+    cleanup_au (au);
     db_plugin->rollback (db_plugin->cls);
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                 "Failed to begin deposit iteration!\n");
@@ -842,7 +1164,7 @@ run_aggregation (void *cls)
     release_shard (s);
     return;
   case GNUNET_DB_STATUS_SOFT_ERROR:
-    cleanup_au (&au_active);
+    cleanup_au (au);
     db_plugin->rollback (db_plugin->cls);
     GNUNET_assert (NULL == task);
     task = GNUNET_SCHEDULER_add_now (&run_aggregation,
@@ -854,7 +1176,7 @@ run_aggregation (void *cls)
       struct GNUNET_TIME_Relative duration
         = GNUNET_TIME_absolute_get_duration (s->start_time.abs_time);
 
-      cleanup_au (&au_active);
+      cleanup_au (au);
       db_plugin->rollback (db_plugin->cls);
       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                   "Completed shard [%u,%u] after %s with %llu deposits\n",
@@ -900,68 +1222,11 @@ run_aggregation (void *cls)
     break;
   }
 
-  TALER_full_payto_hash (au_active.payto_uri,
-                         &au_active.h_full_payto);
-  TALER_full_payto_normalize_and_hash (au_active.payto_uri,
-                                       &au_active.h_normalized_payto);
-  ret = do_aggregate (&au_active);
-  cleanup_au (&au_active);
-  switch (ret)
-  {
-  case GNUNET_SYSERR:
-    global_ret = EXIT_FAILURE;
-    GNUNET_SCHEDULER_shutdown ();
-    db_plugin->rollback (db_plugin->cls);
-    release_shard (s);
-    return;
-  case GNUNET_NO:
-    db_plugin->rollback (db_plugin->cls);
-    GNUNET_assert (NULL == task);
-    task = GNUNET_SCHEDULER_add_now (&run_aggregation,
-                                     s);
-    return;
-  case GNUNET_OK:
-    /* continued below */
-    break;
-  }
-
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-              "Committing aggregation result\n");
-
-  /* Now we can finally commit the overall transaction, as we are
-     again consistent if all of this passes. */
-  switch (commit_or_warn ())
-  {
-  case GNUNET_DB_STATUS_SOFT_ERROR:
-    /* try again */
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                "Serialization issue on commit; trying again later!\n");
-    GNUNET_assert (NULL == task);
-    task = GNUNET_SCHEDULER_add_now (&run_aggregation,
-                                     s);
-    return;
-  case GNUNET_DB_STATUS_HARD_ERROR:
-    GNUNET_break (0);
-    global_ret = EXIT_FAILURE;
-    GNUNET_SCHEDULER_shutdown ();
-    db_plugin->rollback (db_plugin->cls); /* just in case */
-    release_shard (s);
-    return;
-  case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                "Commit complete, going again\n");
-    GNUNET_assert (NULL == task);
-    task = GNUNET_SCHEDULER_add_now (&run_aggregation,
-                                     s);
-    return;
-  default:
-    GNUNET_break (0);
-    global_ret = EXIT_FAILURE;
-    GNUNET_SCHEDULER_shutdown ();
-    db_plugin->rollback (db_plugin->cls); /* just in case */
-    release_shard (s);
-    return;
-  }
+  TALER_full_payto_hash (au->payto_uri,
+                         &au->h_full_payto);
+  TALER_full_payto_normalize_and_hash (au->payto_uri,
+                                       &au->h_normalized_payto);
+  do_aggregate (au);
 }
 
 
@@ -1041,7 +1306,7 @@ run_shard (void *cls)
  * @param total amount aggregated so far
  * @return true to continue to iterate
  */
-static bool
+static void
 handle_transient_cb (
   void *cls,
   const struct TALER_FullPayto payto_uri,
@@ -1056,16 +1321,16 @@ handle_transient_cb (
     GNUNET_break (0);
     return false;
   }
-  au->payto_uri = payto_uri;
+  au->payto_uri.full_payto
+    = GNUNET_strdup (payto_uri.full_payto);
   TALER_full_payto_hash (payto_uri,
                          &au->h_full_payto);
   au->wtid = *wtid;
   au->merchant_pub = *merchant_pub;
   au->trans = *total;
   au->have_transient = true;
-  au->ret = do_aggregate (au);
-  au->payto_uri.full_payto = NULL;
-  return (GNUNET_OK == au->ret);
+  do_aggregate (au);
+  return false;
 }
 
 
@@ -1073,16 +1338,14 @@ static void
 drain_kyc_alerts (void *cls)
 {
   enum GNUNET_DB_QueryStatus qs;
-  struct AggregationUnit au;
+  struct AggregationUnit *au;
 
   (void) cls;
   task = NULL;
   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
               "Draining KYC alerts\n");
-  memset (&au,
-          0,
-          sizeof (au));
-  au.execution_time = GNUNET_TIME_timestamp_get ();
+  au = GNUNET_new (struct AggregationUnit);
+  au->execution_time = GNUNET_TIME_timestamp_get ();
   if (GNUNET_SYSERR ==
       db_plugin->preflight (db_plugin->cls))
   {
@@ -1106,7 +1369,7 @@ drain_kyc_alerts (void *cls)
   {
     qs = db_plugin->drain_kyc_alert (db_plugin->cls,
                                      1,
-                                     &au.h_normalized_payto);
+                                     &au->h_normalized_payto);
     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                 "Found %d KYC alerts\n",
                 (int) qs);
@@ -1115,6 +1378,7 @@ drain_kyc_alerts (void *cls)
     case GNUNET_DB_STATUS_HARD_ERROR:
       GNUNET_break (0);
       db_plugin->rollback (db_plugin->cls);
+      GNUNET_free (au);
       GNUNET_assert (NULL == task);
       task = GNUNET_SCHEDULER_add_now (&drain_kyc_alerts,
                                        NULL);
@@ -1122,10 +1386,12 @@ drain_kyc_alerts (void *cls)
     case GNUNET_DB_STATUS_SOFT_ERROR:
       db_plugin->rollback (db_plugin->cls);
       GNUNET_assert (NULL == task);
+      GNUNET_free (au);
       task = GNUNET_SCHEDULER_add_now (&drain_kyc_alerts,
                                        NULL);
       return;
     case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
+      GNUNET_free (au);
       qs = db_plugin->commit (db_plugin->cls);
       if (qs < 0)
         GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
@@ -1139,9 +1405,10 @@ drain_kyc_alerts (void *cls)
       break;
     }
 
-    au.ret = GNUNET_OK;
+    au->ret = GNUNET_OK;
+    /* FIXME: should be replaced with a query that has a LIMIT 1... */
     qs = db_plugin->find_aggregation_transient (db_plugin->cls,
-                                                &au.h_normalized_payto,
+                                                &au->h_normalized_payto,
                                                 &handle_transient_cb,
                                                 &au);
     switch (qs)
@@ -1166,66 +1433,11 @@ drain_kyc_alerts (void *cls)
     case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
       continue; /* while (1) */
     default:
-      break;
-    }
-    break;
-  } /* while(1) */
-
-  {
-    enum GNUNET_GenericReturnValue ret;
-
-    ret = au.ret;
-    cleanup_au (&au);
-    switch (ret)
-    {
-    case GNUNET_SYSERR:
-      GNUNET_break (0);
-      global_ret = EXIT_FAILURE;
-      GNUNET_SCHEDULER_shutdown ();
-      db_plugin->rollback (db_plugin->cls); /* just in case */
+      /* handle_transient_cb has various continuations... */
       return;
-    case GNUNET_NO:
-      db_plugin->rollback (db_plugin->cls);
-      GNUNET_assert (NULL == task);
-      task = GNUNET_SCHEDULER_add_now (&drain_kyc_alerts,
-                                       NULL);
-      return;
-    case GNUNET_OK:
-      /* continued below */
-      break;
     }
-  }
-
-  switch (commit_or_warn ())
-  {
-  case GNUNET_DB_STATUS_SOFT_ERROR:
-    /* try again */
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                "Serialization issue on commit; trying again later!\n");
-    GNUNET_assert (NULL == task);
-    task = GNUNET_SCHEDULER_add_now (&drain_kyc_alerts,
-                                     NULL);
-    return;
-  case GNUNET_DB_STATUS_HARD_ERROR:
-    GNUNET_break (0);
-    global_ret = EXIT_FAILURE;
-    GNUNET_SCHEDULER_shutdown ();
-    db_plugin->rollback (db_plugin->cls); /* just in case */
-    return;
-  case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                "Commit complete, going again\n");
-    GNUNET_assert (NULL == task);
-    task = GNUNET_SCHEDULER_add_now (&drain_kyc_alerts,
-                                     NULL);
-    return;
-  default:
-    GNUNET_break (0);
-    global_ret = EXIT_FAILURE;
-    GNUNET_SCHEDULER_shutdown ();
-    db_plugin->rollback (db_plugin->cls); /* just in case */
-    return;
-  }
+    GNUNET_assert (0);
+  } /* while(1) */
 }
 
 
diff --git a/src/exchange/taler-exchange-httpd_common_kyc.c b/src/exchange/taler-exchange-httpd_common_kyc.c
index 823bcb26d..eaf7fc1c1 100644
--- a/src/exchange/taler-exchange-httpd_common_kyc.c
+++ b/src/exchange/taler-exchange-httpd_common_kyc.c
@@ -25,6 +25,7 @@
 #include "taler_error_codes.h"
 #include "taler_kyclogic_lib.h"
 #include "taler_exchangedb_plugin.h"
+#include "taler_exchangedb_lib.h"
 #include <gnunet/gnunet_common.h>
 
 /**
@@ -322,247 +323,6 @@ done:
 }
 
 
-/**
- * Function called to expand AML history for the account.
- *
- * @param cls a `json_t *` array to build
- * @param decision_time when was the decision taken
- * @param justification what was the given justification
- * @param decider_pub which key signed the decision
- * @param jproperties what are the new account properties
- * @param jnew_rules what are the new account rules
- * @param to_investigate should AML staff investigate
- *          after the decision
- * @param is_active is this the active decision
- */
-static void
-add_aml_history_entry (
-  void *cls,
-  struct GNUNET_TIME_Timestamp decision_time,
-  const char *justification,
-  const struct TALER_AmlOfficerPublicKeyP *decider_pub,
-  const json_t *jproperties,
-  const json_t *jnew_rules,
-  bool to_investigate,
-  bool is_active)
-{
-  json_t *aml_history = cls;
-  json_t *e;
-
-  e = GNUNET_JSON_PACK (
-    GNUNET_JSON_pack_timestamp ("decision_time",
-                                decision_time),
-    GNUNET_JSON_pack_string ("justification",
-                             justification),
-    GNUNET_JSON_pack_data_auto ("decider_pub",
-                                decider_pub),
-    GNUNET_JSON_pack_object_incref ("properties",
-                                    (json_t *) jproperties),
-    GNUNET_JSON_pack_object_incref ("new_rules",
-                                    (json_t *) jnew_rules),
-    GNUNET_JSON_pack_bool ("to_investigate",
-                           to_investigate),
-    GNUNET_JSON_pack_bool ("is_active",
-                           is_active)
-    );
-  GNUNET_assert (0 ==
-                 json_array_append_new (aml_history,
-                                        e));
-}
-
-
-/**
- * Function called to obtain an AML
- * history in JSON on-demand if needed.
- *
- * @param cls must be a `struct TALER_NormalizedPaytoHashP account_id *`
- * @return AML history in JSON format, NULL on error
- */
-static json_t *
-aml_history_builder_cb (void *cls)
-{
-  const struct TALER_NormalizedPaytoHashP *acc = cls;
-  enum GNUNET_DB_QueryStatus qs;
-  json_t *aml_history;
-
-  aml_history = json_array ();
-  GNUNET_assert (NULL != aml_history);
-  qs = TEH_plugin->lookup_aml_history (
-    TEH_plugin->cls,
-    acc,
-    &add_aml_history_entry,
-    aml_history);
-  switch (qs)
-  {
-  case GNUNET_DB_STATUS_HARD_ERROR:
-  case GNUNET_DB_STATUS_SOFT_ERROR:
-    GNUNET_break (0);
-    json_decref (aml_history);
-    return NULL;
-  case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
-    /* empty history is fine! */
-    break;
-  case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
-    break;
-  }
-  return aml_history;
-}
-
-
-/**
- * Function called to expand KYC history for the account.
- *
- * @param cls a `json_t *` array to build
- * @param provider_name name of the KYC provider
- *    or NULL for none
- * @param finished did the KYC process finish
- * @param error_code error code from the KYC process
- * @param error_message error message from the KYC process,
- *    or NULL for none
- * @param provider_user_id user ID at the provider
- *    or NULL for none
- * @param provider_legitimization_id legitimization process ID at the provider
- *    or NULL for none
- * @param collection_time when was the data collected
- * @param expiration_time when does the collected data expire
- * @param encrypted_attributes_len number of bytes in @a encrypted_attributes
- * @param encrypted_attributes encrypted KYC attributes
- */
-static void
-add_kyc_history_entry (
-  void *cls,
-  const char *provider_name,
-  bool finished,
-  enum TALER_ErrorCode error_code,
-  const char *error_message,
-  const char *provider_user_id,
-  const char *provider_legitimization_id,
-  struct GNUNET_TIME_Timestamp collection_time,
-  struct GNUNET_TIME_Absolute expiration_time,
-  size_t encrypted_attributes_len,
-  const void *encrypted_attributes)
-{
-  json_t *kyc_history = cls;
-  json_t *attributes;
-  json_t *e;
-
-  attributes = TALER_CRYPTO_kyc_attributes_decrypt (
-    &TEH_attribute_key,
-    encrypted_attributes,
-    encrypted_attributes_len);
-  e = GNUNET_JSON_PACK (
-    GNUNET_JSON_pack_string (
-      "provider_name",
-      provider_name),
-    GNUNET_JSON_pack_bool (
-      "finished",
-      finished),
-    TALER_JSON_pack_ec (error_code),
-    GNUNET_JSON_pack_allow_null (
-      GNUNET_JSON_pack_string (
-        "error_message",
-        error_message)),
-    GNUNET_JSON_pack_allow_null (
-      GNUNET_JSON_pack_string (
-        "provider_user_id",
-        provider_user_id)),
-    GNUNET_JSON_pack_allow_null (
-      GNUNET_JSON_pack_string (
-        "provider_legitimization_id",
-        provider_legitimization_id)),
-    GNUNET_JSON_pack_allow_null (
-      GNUNET_JSON_pack_timestamp (
-        "collection_time",
-        collection_time)),
-    GNUNET_JSON_pack_allow_null (
-      GNUNET_JSON_pack_timestamp (
-        "expiration_time",
-        GNUNET_TIME_absolute_to_timestamp (
-          expiration_time))),
-    GNUNET_JSON_pack_allow_null (
-      GNUNET_JSON_pack_object_steal (
-        "attributes",
-        attributes))
-    );
-
-  GNUNET_assert (0 ==
-                 json_array_append_new (kyc_history,
-                                        e));
-}
-
-
-/**
- * Function called to obtain a KYC
- * history in JSON on-demand if needed.
- *
- * @param cls must be a `struct TALER_NormalizedPaytoHashP account_id *`
- * @return KYC history in JSON format, NULL on error
- */
-static json_t *
-kyc_history_builder_cb (void *cls)
-{
-  const struct TALER_NormalizedPaytoHashP *acc = cls;
-  enum GNUNET_DB_QueryStatus qs;
-  json_t *kyc_history;
-
-  kyc_history = json_array ();
-  GNUNET_assert (NULL != kyc_history);
-  qs = TEH_plugin->lookup_kyc_history (
-    TEH_plugin->cls,
-    acc,
-    &add_kyc_history_entry,
-    kyc_history);
-  switch (qs)
-  {
-  case GNUNET_DB_STATUS_HARD_ERROR:
-  case GNUNET_DB_STATUS_SOFT_ERROR:
-    GNUNET_break (0);
-    json_decref (kyc_history);
-    return NULL;
-  case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
-    /* empty history is fine! */
-    break;
-  case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
-    break;
-  }
-  return kyc_history;
-}
-
-
-/**
- * Function called to obtain the current ``LegitimizationRuleSet``
- * in JSON for an account on-demand if needed.
- *
- * @param cls must be a `struct TALER_NormalizedPaytoHashP *`
- * @return KYC history in JSON format, NULL on error
- */
-static json_t *
-current_rule_builder_cb (void *cls)
-{
-  const struct TALER_NormalizedPaytoHashP *acc = cls;
-  enum GNUNET_DB_QueryStatus qs;
-  json_t *jlrs;
-
-  qs = TEH_plugin->get_kyc_rules2 (
-    TEH_plugin->cls,
-    acc,
-    &jlrs);
-  switch (qs)
-  {
-  case GNUNET_DB_STATUS_HARD_ERROR:
-  case GNUNET_DB_STATUS_SOFT_ERROR:
-    GNUNET_break (0);
-    return NULL;
-  case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
-    jlrs = json_incref ((json_t *) TALER_KYCLOGIC_get_default_legi_rules ());
-    break;
-  case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
-    break;
-  }
-  return jlrs;
-}
-
-
 void
 TEH_kyc_run_measure_cancel (struct TEH_KycMeasureRunContext *kat)
 {
@@ -648,20 +408,27 @@ TEH_kyc_run_measure_for_attributes (
   case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
     break;
   }
-  kat->kyc_aml
-    = TALER_KYCLOGIC_run_aml_program (
-        kat->jmeasures,
-        kat->measure_index,
-        kat->attributes,
-        &current_rule_builder_cb,
-        &kat->account_id,
-        &aml_history_builder_cb,
-        &kat->account_id,
-        &kyc_history_builder_cb,
-        &kat->account_id,
-        &kyc_aml_finished,
-        kat);
+  {
+    struct TALER_EXCHANGEDB_HistoryBuilderContext hbc = {
+      .account = &kat->account_id,
+      .db_plugin = TEH_plugin,
+      .attribute_key = &TEH_attribute_key
+    };
 
+    kat->kyc_aml
+      = TALER_KYCLOGIC_run_aml_program (
+          kat->jmeasures,
+          kat->measure_index,
+          kat->attributes,
+          &TALER_EXCHANGEDB_current_rule_builder,
+          &hbc,
+          &TALER_EXCHANGEDB_aml_history_builder,
+          &hbc,
+          &TALER_EXCHANGEDB_kyc_history_builder,
+          &hbc,
+          &kyc_aml_finished,
+          kat);
+  }
   if (NULL == kat->kyc_aml)
   {
     GNUNET_break (0);
@@ -772,19 +539,26 @@ TEH_kyc_run_measure_directly (
     TEH_kyc_run_measure_cancel (kat);
     return NULL;
   }
+  {
+    struct TALER_EXCHANGEDB_HistoryBuilderContext hbc = {
+      .account = &kat->account_id,
+      .db_plugin = TEH_plugin,
+      .attribute_key = &TEH_attribute_key
+    };
 
-  kat->kyc_aml
-    = TALER_KYCLOGIC_run_aml_program3 (
-        instant_ms,
-        NULL,                                /* no attributes */
-        &current_rule_builder_cb,
-        &kat->account_id,
-        &aml_history_builder_cb,
-        &kat->account_id,
-        &kyc_history_builder_cb,
-        &kat->account_id,
-        &kyc_aml_finished,
-        kat);
+    kat->kyc_aml
+      = TALER_KYCLOGIC_run_aml_program3 (
+          instant_ms,
+          NULL,                              /* no attributes */
+          &TALER_EXCHANGEDB_current_rule_builder,
+          &hbc,
+          &TALER_EXCHANGEDB_aml_history_builder,
+          &hbc,
+          &TALER_EXCHANGEDB_kyc_history_builder,
+          &hbc,
+          &kyc_aml_finished,
+          kat);
+  }
   if (NULL == kat->kyc_aml)
   {
     GNUNET_break (0);
@@ -983,18 +757,24 @@ TEH_kyc_fallback (
   fb->cb_cls = cb_cls;
   if (NULL == kcc.check)
   {
+    struct TALER_EXCHANGEDB_HistoryBuilderContext hbc = {
+      .account = &fb->account_id,
+      .db_plugin = TEH_plugin,
+      .attribute_key = &TEH_attribute_key
+    };
+
     /* check was set to 'SKIP', run program immediately */
     fb->aprh
       = TALER_KYCLOGIC_run_aml_program2 (
           kcc.prog_name,
           attributes,
           kcc.context,
-          &current_rule_builder_cb,
-          &fb->account_id,
-          &aml_history_builder_cb,
-          &fb->account_id,
-          &kyc_history_builder_cb,
-          &fb->account_id,
+          &TALER_EXCHANGEDB_current_rule_builder,
+          &hbc,
+          &TALER_EXCHANGEDB_aml_history_builder,
+          &hbc,
+          &TALER_EXCHANGEDB_kyc_history_builder,
+          &hbc,
           &handle_aml_fallback_result,
           fb);
     if (NULL == fb->aprh)
diff --git a/src/exchangedb/Makefile.am b/src/exchangedb/Makefile.am
index 9560eefc8..a6c8be7c0 100644
--- a/src/exchangedb/Makefile.am
+++ b/src/exchangedb/Makefile.am
@@ -340,11 +340,15 @@ lib_LTLIBRARIES = \
 
 libtalerexchangedb_la_SOURCES = \
   exchangedb_accounts.c \
+  exchangedb_history.c \
   exchangedb_plugin.c \
   exchangedb_transactions.c
 libtalerexchangedb_la_LIBADD = \
   $(top_builddir)/src/bank-lib/libtalerbank.la \
+  $(top_builddir)/src/kyclogic/libtalerkyclogic.la \
+  $(top_builddir)/src/json/libtalerjson.la \
   $(top_builddir)/src/util/libtalerutil.la \
+  -lgnunetjson \
   -lgnunetutil \
   $(XLIB)
 libtalerexchangedb_la_LDFLAGS = \
diff --git a/src/exchangedb/exchangedb_history.c b/src/exchangedb/exchangedb_history.c
new file mode 100644
index 000000000..8a1014a0e
--- /dev/null
+++ b/src/exchangedb/exchangedb_history.c
@@ -0,0 +1,267 @@
+/*
+  This file is part of TALER
+  Copyright (C) 2023, 2024 Taler Systems SA
+
+  TALER is free software; you can redistribute it and/or modify it under the
+  terms of the GNU Affero General Public License as published by the Free Software
+  Foundation; either version 3, or (at your option) any later version.
+
+  TALER is distributed in the hope that it will be useful, but WITHOUT ANY
+  WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+  A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more details.
+
+  You should have received a copy of the GNU Affero General Public License along with
+  TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
+*/
+/**
+ * @file exchangedb_history.c
+ * @brief helper function to build AML inputs from account histories
+ * @author Christian Grothoff
+ */
+#include "taler_exchangedb_plugin.h"
+#include "taler_exchangedb_lib.h"
+#include "taler_kyclogic_lib.h"
+#include "taler_json_lib.h"
+#include <gnunet/gnunet_common.h>
+
+/**
+ * Function called to expand AML history for the account.
+ *
+ * @param cls a `json_t *` array to build
+ * @param decision_time when was the decision taken
+ * @param justification what was the given justification
+ * @param decider_pub which key signed the decision
+ * @param jproperties what are the new account properties
+ * @param jnew_rules what are the new account rules
+ * @param to_investigate should AML staff investigate
+ *          after the decision
+ * @param is_active is this the active decision
+ */
+static void
+add_aml_history_entry (
+  void *cls,
+  struct GNUNET_TIME_Timestamp decision_time,
+  const char *justification,
+  const struct TALER_AmlOfficerPublicKeyP *decider_pub,
+  const json_t *jproperties,
+  const json_t *jnew_rules,
+  bool to_investigate,
+  bool is_active)
+{
+  json_t *aml_history = cls;
+  json_t *e;
+
+  e = GNUNET_JSON_PACK (
+    GNUNET_JSON_pack_timestamp ("decision_time",
+                                decision_time),
+    GNUNET_JSON_pack_string ("justification",
+                             justification),
+    GNUNET_JSON_pack_data_auto ("decider_pub",
+                                decider_pub),
+    GNUNET_JSON_pack_object_incref ("properties",
+                                    (json_t *) jproperties),
+    GNUNET_JSON_pack_object_incref ("new_rules",
+                                    (json_t *) jnew_rules),
+    GNUNET_JSON_pack_bool ("to_investigate",
+                           to_investigate),
+    GNUNET_JSON_pack_bool ("is_active",
+                           is_active)
+    );
+  GNUNET_assert (0 ==
+                 json_array_append_new (aml_history,
+                                        e));
+}
+
+
+json_t *
+TALER_EXCHANGEDB_aml_history_builder (void *cls)
+{
+  struct TALER_EXCHANGEDB_HistoryBuilderContext *hbc = cls;
+  const struct TALER_NormalizedPaytoHashP *acc = hbc->account;
+  enum GNUNET_DB_QueryStatus qs;
+  json_t *aml_history;
+
+  aml_history = json_array ();
+  GNUNET_assert (NULL != aml_history);
+  qs = hbc->db_plugin->lookup_aml_history (
+    hbc->db_plugin->cls,
+    acc,
+    &add_aml_history_entry,
+    aml_history);
+  switch (qs)
+  {
+  case GNUNET_DB_STATUS_HARD_ERROR:
+  case GNUNET_DB_STATUS_SOFT_ERROR:
+    GNUNET_break (0);
+    json_decref (aml_history);
+    return NULL;
+  case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
+    /* empty history is fine! */
+    break;
+  case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
+    break;
+  }
+  return aml_history;
+}
+
+
+/**
+ * Closure for #add_kyc_history_entry.
+ */
+struct KycContext
+{
+  /**
+   * JSON array we are building.
+   */
+  json_t *kyc_history;
+
+  /**
+   * Key to use to decrypt KYC attributes.
+   */
+  const struct TALER_AttributeEncryptionKeyP *attribute_key;
+};
+
+
+/**
+ * Function called to expand KYC history for the account.
+ *
+ * @param cls a `json_t *` array to build
+ * @param provider_name name of the KYC provider
+ *    or NULL for none
+ * @param finished did the KYC process finish
+ * @param error_code error code from the KYC process
+ * @param error_message error message from the KYC process,
+ *    or NULL for none
+ * @param provider_user_id user ID at the provider
+ *    or NULL for none
+ * @param provider_legitimization_id legitimization process ID at the provider
+ *    or NULL for none
+ * @param collection_time when was the data collected
+ * @param expiration_time when does the collected data expire
+ * @param encrypted_attributes_len number of bytes in @a encrypted_attributes
+ * @param encrypted_attributes encrypted KYC attributes
+ */
+static void
+add_kyc_history_entry (
+  void *cls,
+  const char *provider_name,
+  bool finished,
+  enum TALER_ErrorCode error_code,
+  const char *error_message,
+  const char *provider_user_id,
+  const char *provider_legitimization_id,
+  struct GNUNET_TIME_Timestamp collection_time,
+  struct GNUNET_TIME_Absolute expiration_time,
+  size_t encrypted_attributes_len,
+  const void *encrypted_attributes)
+{
+  struct KycContext *kc = cls;
+  json_t *kyc_history = kc->kyc_history;
+  json_t *attributes;
+  json_t *e;
+
+  attributes = TALER_CRYPTO_kyc_attributes_decrypt (
+    kc->attribute_key,
+    encrypted_attributes,
+    encrypted_attributes_len);
+  e = GNUNET_JSON_PACK (
+    GNUNET_JSON_pack_string (
+      "provider_name",
+      provider_name),
+    GNUNET_JSON_pack_bool (
+      "finished",
+      finished),
+    TALER_JSON_pack_ec (error_code),
+    GNUNET_JSON_pack_allow_null (
+      GNUNET_JSON_pack_string (
+        "error_message",
+        error_message)),
+    GNUNET_JSON_pack_allow_null (
+      GNUNET_JSON_pack_string (
+        "provider_user_id",
+        provider_user_id)),
+    GNUNET_JSON_pack_allow_null (
+      GNUNET_JSON_pack_string (
+        "provider_legitimization_id",
+        provider_legitimization_id)),
+    GNUNET_JSON_pack_allow_null (
+      GNUNET_JSON_pack_timestamp (
+        "collection_time",
+        collection_time)),
+    GNUNET_JSON_pack_allow_null (
+      GNUNET_JSON_pack_timestamp (
+        "expiration_time",
+        GNUNET_TIME_absolute_to_timestamp (
+          expiration_time))),
+    GNUNET_JSON_pack_allow_null (
+      GNUNET_JSON_pack_object_steal (
+        "attributes",
+        attributes))
+    );
+
+  GNUNET_assert (0 ==
+                 json_array_append_new (kyc_history,
+                                        e));
+}
+
+
+json_t *
+TALER_EXCHANGEDB_kyc_history_builder (void *cls)
+{
+  struct TALER_EXCHANGEDB_HistoryBuilderContext *hbc = cls;
+  const struct TALER_NormalizedPaytoHashP *acc = hbc->account;
+  enum GNUNET_DB_QueryStatus qs;
+  struct KycContext kc = {
+    .kyc_history = json_array (),
+    .attribute_key = hbc->attribute_key
+  };
+
+  GNUNET_assert (NULL != kc.kyc_history);
+  qs = hbc->db_plugin->lookup_kyc_history (
+    hbc->db_plugin->cls,
+    acc,
+    &add_kyc_history_entry,
+    &kc);
+  switch (qs)
+  {
+  case GNUNET_DB_STATUS_HARD_ERROR:
+  case GNUNET_DB_STATUS_SOFT_ERROR:
+    GNUNET_break (0);
+    json_decref (kc.kyc_history);
+    return NULL;
+  case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
+    /* empty history is fine! */
+    break;
+  case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
+    break;
+  }
+  return kc.kyc_history;
+}
+
+
+json_t *
+TALER_EXCHANGEDB_current_rule_builder (void *cls)
+{
+  struct TALER_EXCHANGEDB_HistoryBuilderContext *hbc = cls;
+  const struct TALER_NormalizedPaytoHashP *acc = hbc->account;
+  enum GNUNET_DB_QueryStatus qs;
+  json_t *jlrs;
+
+  qs = hbc->db_plugin->get_kyc_rules2 (
+    hbc->db_plugin->cls,
+    acc,
+    &jlrs);
+  switch (qs)
+  {
+  case GNUNET_DB_STATUS_HARD_ERROR:
+  case GNUNET_DB_STATUS_SOFT_ERROR:
+    GNUNET_break (0);
+    return NULL;
+  case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
+    jlrs = json_incref ((json_t *) TALER_KYCLOGIC_get_default_legi_rules ());
+    break;
+  case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
+    break;
+  }
+  return jlrs;
+}
diff --git a/src/include/taler_crypto_lib.h b/src/include/taler_crypto_lib.h
index c3867d18b..096ab8147 100644
--- a/src/include/taler_crypto_lib.h
+++ b/src/include/taler_crypto_lib.h
@@ -179,18 +179,6 @@ struct TALER_ReserveSignatureP
 
 
 /**
- * (Symmetric) key used to encrypt KYC attribute data in the database.
- */
-struct TALER_AttributeKeyP
-{
-  /**
-   * Actual key material.
-   */
-  struct GNUNET_HashCode key;
-};
-
-
-/**
  * @brief Type of public keys to for merchant authorizations.
  * Merchants can issue refunds using the corresponding
  * private key.
diff --git a/src/include/taler_exchangedb_lib.h b/src/include/taler_exchangedb_lib.h
index 825578c95..e5644fa63 100644
--- a/src/include/taler_exchangedb_lib.h
+++ b/src/include/taler_exchangedb_lib.h
@@ -202,4 +202,60 @@ TALER_EXCHANGEDB_load_accounts (
 void
 TALER_EXCHANGEDB_unload_accounts (void);
 
+
+/**
+ * Closure for various history building functions.
+ */
+struct TALER_EXCHANGEDB_HistoryBuilderContext
+{
+  /**
+   * Account to build history for.
+   */
+  const struct TALER_NormalizedPaytoHashP *account;
+
+  /**
+   * Database plugin to build history with.
+   */
+  struct TALER_EXCHANGEDB_Plugin *db_plugin;
+
+  /**
+   * Key to use to decrypt KYC attributes.
+   */
+  struct TALER_AttributeEncryptionKeyP *attribute_key;
+};
+
+
+/**
+ * Function called to obtain an AML
+ * history in JSON on-demand if needed.
+ *
+ * @param cls must be a `struct TALER_EXCHANGEDB_HistoryBuilderContext *`
+ * @return AML history in JSON format, NULL on error
+ */
+json_t *
+TALER_EXCHANGEDB_aml_history_builder (void *cls);
+
+
+/**
+ * Function called to obtain a KYC
+ * history in JSON on-demand if needed.
+ *
+ * @param cls must be a `struct TALER_EXCHANGEDB_HistoryBuilderContext *`
+ * @return KYC history in JSON format, NULL on error
+ */
+json_t *
+TALER_EXCHANGEDB_kyc_history_builder (void *cls);
+
+
+/**
+ * Function called to obtain the current ``LegitimizationRuleSet``
+ * in JSON for an account on-demand if needed.
+ *
+ * @param cls must be a `struct TALER_EXCHANGEDB_HistoryBuilderContext *`
+ * @return KYC history in JSON format, NULL on error
+ */
+json_t *
+TALER_EXCHANGEDB_current_rule_builder (void *cls);
+
+
 #endif