<commit_before>part of eventsourcing;

/**
 *  Hauptklasse, die für eine Instanz des EventSourcing-Systems steht.
 *  Integriert Datenbankzugriff (MySQL mit sqljocky), Eventsourcing in der Datei [eventsFile]
 *  und einen Webserver mit speziellem WebSocket-Protokoll (siehe [WebSocketClient])
 *  sowie Auslieferung von statischen Dateien in "html".
 *  Bei Start wird die aktuelle EventQueue in [eventFile] in eine neue Datei in
 *  [backupDirectory] gesichert. Zum Aufbau des Dateiformats siehe [readEventFile].
 *  */
class EventRouter {
  /// Pool an SQL-Verbindungen (Standardmaximum 50)
  final ConnectionPool db;

  /// EventQueue, in der die verarbeiteten Events gespeichert werden
  final File eventFile;

  /// Ordner, in den die alte EventQueue bei Start gesichert wird
  final Directory backupDirectory;

  /// Bekannte [EventHandler], die bei Ausführung eines bestimmten Events ausgeführt werden sollen
  final Map<String, List<EventHandler>> eventHandler;

  /// Bekannte [QueryHandler], die bei Ausführung eines bestimmten Queries ausgeführt werden sollen
  final Map<String, List<QueryHandler>> queryHandler;

  /// Höchste EventID, die bereits benutzt wurde. Neue Events erhalten eine höhere Nummer.
  int _biggestKnownEventId = 0;

  int _eventCount = 0;

  final Authenticator authenticator;

  final List<WebSocketConnection> connections = [];

  /// Sink für die Events
  EventFileWriter _eventQueueSink = null;

  /// Werden zusätzlich ausgeführt, wenn *neue* Events eingehen
  final List<EventHandler> pushHooks = [];

  /// Sichert die aktuelle EventQueue in [eventFile] in eine neue Datei in [backupDirectory] ("events_milliSecondsSinceEpoch.dat")
  Future backupEvents() async {
    final File tempEventFile = new File(backupDirectory.path + "/tmp.dat");
    await tempEventFile.create(recursive: true);
    final Stream<List<int>> eventsFile = eventFile.openRead();
    final IOSink tmpFile = tempEventFile.openWrite();

    await eventsFile.transform(GZIP.encoder).pipe(tmpFile);
    await tempEventFile.rename(backupDirectory.path +
        "/events_${new DateTime.now().millisecondsSinceEpoch}.dat");
  }

  /**
   * Spielt ein Event neu ab. Dabei werden Ausführungsfehler in einzelnen [EventHandler]n ignoriert
   * und das Ergebnis der anderen Handler trotzdem verwendet und in die Datenbank gesichert.
   * Die EventHandler werden alle *nacheinander* in *einer* Transaktion ausgeführt.
   * Unbekannte Events lösen eine Exception aus. Jedes Event sollte die Felder "action", "id" und "timestamp"
   * definieren.
   * */
  Future replayEvent(Map e) async {
    assert(e.containsKey("action") &&
        e.containsKey("id") &&
        e.containsKey("timestamp"));
    if (e.containsKey("id") && e["id"] >= _biggestKnownEventId)
      _biggestKnownEventId = e["id"];

    if (!eventHandler.containsKey(e["action"]))
      throw "Keine Verarbeitung für ${e["action"]} hinterlegt";

    final List<EventHandler> handlers = eventHandler[e["action"]];
    final Transaction trans = await db.startTransaction();

    try {
      for (EventHandler h in handlers) {
        final Map ep = new Map.from(e);

        try {
          await h(ep, trans);
        } catch (err) {
          // Beim Abspielen der Events Fehler ignorieren
          print(
              "$_eventCount Events verarbeitet; Eventfehler bei ${new DateTime.fromMillisecondsSinceEpoch(e["timestamp"])} von ${e.containsKey("user") ? e["user"] : "Unbekannt"}, ${e["action"]}: $err");
        }
      }
      await trans.commit();
    } catch (e) {
      await trans.rollback();
      rethrow;
    } finally {
      _eventCount++;
    }
  }

  /**
   * Führt eine Query aus. Löst ein [QueryHandler] eine Exception aus, wird eine Map mit {"error": "msg"}
   * zurückgegeben.
   * Aktuell werden die QueryHandler in einer konsistenten Transaction *parallel* ausgeführt.
   * */
  Future<Map> submitQuery(Map q, int user) async {
    if (!queryHandler.containsKey(q["action"]))
      throw new Exception("Keine Verarbeitung für ${q["action"]} hinterlegt");

    final List<QueryHandler> handlers = queryHandler[q["action"]];
    final Map finalOutput = {};
    q["timestamp"] = new DateTime.now().millisecondsSinceEpoch;
    q["user"] = user;

    // final Transaction trans = await db.startTransaction(consistent: true);

    final Iterable<Future> handlerRunners = handlers.map((h) async {
      var qo;
      try {
        qo = await h(new Map.from(q), db);
      } catch (e) {
        rethrow;
      }

      if (qo is Map) qo.forEach((k, v) => finalOutput[k] = v);
    });

    await Future.wait(handlerRunners, eagerError: true);

    // FinalOutput enthält alle Ausgaben (return to sender)
    return finalOutput;
  }

  /**
   * Spielt ein Event neu ab. Die EventHandler werden alle *nacheinander* in *einer* Transaktion ausgeführt.
   * Wirft einer der [EventHandler] eine Exception, wird ein Rollback ausgeführt und eine Map mit {"error": "msg"}
   * zurückgegeben.
   * Unbekannte Events lösen eine Exception aus. Jedes Event sollte die Felder "action", "id" und "timestamp"
   * definieren.
   * TODO: Die Events dürfen ja eigentlich auch parallel ausgeführt werden, da sie voneinander unabhängig sind
   * */
  Future<Map> submitEvent(Map e, Map hookData, int user) async {
    if (!eventHandler.containsKey(e["action"]))
      throw new Exception("Keine Verarbeitung für ${e["action"]} hinterlegt");

    final List<EventHandler> handlers = eventHandler[e["action"]];
    final Map finalInput = {
      "timestamp": new DateTime.now().millisecondsSinceEpoch,
      "id": ++_biggestKnownEventId,
      "user": user
    };
    e.addAll(finalInput);
    final Transaction trans = await db.startTransaction();

    try {
      for (EventHandler h in handlers) {
        await h(new Map.from(e), trans);
      }

      for (EventHandler hook in pushHooks) {<commit_msg>Fix event-transactions not being marked as consistent<commit_after>part of eventsourcing;

/**
 *  Hauptklasse, die für eine Instanz des EventSourcing-Systems steht.
 *  Integriert Datenbankzugriff (MySQL mit sqljocky), Eventsourcing in der Datei [eventsFile]
 *  und einen Webserver mit speziellem WebSocket-Protokoll (siehe [WebSocketClient])
 *  sowie Auslieferung von statischen Dateien in "html".
 *  Bei Start wird die aktuelle EventQueue in [eventFile] in eine neue Datei in
 *  [backupDirectory] gesichert. Zum Aufbau des Dateiformats siehe [readEventFile].
 *  */
class EventRouter {
  /// Pool an SQL-Verbindungen (Standardmaximum 50)
  final ConnectionPool db;

  /// EventQueue, in der die verarbeiteten Events gespeichert werden
  final File eventFile;

  /// Ordner, in den die alte EventQueue bei Start gesichert wird
  final Directory backupDirectory;

  /// Bekannte [EventHandler], die bei Ausführung eines bestimmten Events ausgeführt werden sollen
  final Map<String, List<EventHandler>> eventHandler;

  /// Bekannte [QueryHandler], die bei Ausführung eines bestimmten Queries ausgeführt werden sollen
  final Map<String, List<QueryHandler>> queryHandler;

  /// Höchste EventID, die bereits benutzt wurde. Neue Events erhalten eine höhere Nummer.
  int _biggestKnownEventId = 0;

  int _eventCount = 0;

  final Authenticator authenticator;

  final List<WebSocketConnection> connections = [];

  /// Sink für die Events
  EventFileWriter _eventQueueSink = null;

  /// Werden zusätzlich ausgeführt, wenn *neue* Events eingehen
  final List<EventHandler> pushHooks = [];

  /// Sichert die aktuelle EventQueue in [eventFile] in eine neue Datei in [backupDirectory] ("events_milliSecondsSinceEpoch.dat")
  Future backupEvents() async {
    final File tempEventFile = new File(backupDirectory.path + "/tmp.dat");
    await tempEventFile.create(recursive: true);
    final Stream<List<int>> eventsFile = eventFile.openRead();
    final IOSink tmpFile = tempEventFile.openWrite();

    await eventsFile.transform(GZIP.encoder).pipe(tmpFile);
    await tempEventFile.rename(backupDirectory.path +
        "/events_${new DateTime.now().millisecondsSinceEpoch}.dat");
  }

  /**
   * Spielt ein Event neu ab. Dabei werden Ausführungsfehler in einzelnen [EventHandler]n ignoriert
   * und das Ergebnis der anderen Handler trotzdem verwendet und in die Datenbank gesichert.
   * Die EventHandler werden alle *nacheinander* in *einer* Transaktion ausgeführt.
   * Unbekannte Events lösen eine Exception aus. Jedes Event sollte die Felder "action", "id" und "timestamp"
   * definieren.
   * */
  Future replayEvent(Map e) async {
    assert(e.containsKey("action") &&
        e.containsKey("id") &&
        e.containsKey("timestamp"));
    if (e.containsKey("id") && e["id"] >= _biggestKnownEventId)
      _biggestKnownEventId = e["id"];

    if (!eventHandler.containsKey(e["action"]))
      throw "Keine Verarbeitung für ${e["action"]} hinterlegt";

    final List<EventHandler> handlers = eventHandler[e["action"]];
    final Transaction trans = await db.startTransaction();

    try {
      for (EventHandler h in handlers) {
        final Map ep = new Map.from(e);

        try {
          await h(ep, trans);
        } catch (err) {
          // Beim Abspielen der Events Fehler ignorieren
          print(
              "$_eventCount Events verarbeitet; Eventfehler bei ${new DateTime.fromMillisecondsSinceEpoch(e["timestamp"])} von ${e.containsKey("user") ? e["user"] : "Unbekannt"}, ${e["action"]}: $err");
        }
      }
      await trans.commit();
    } catch (e) {
      await trans.rollback();
      rethrow;
    } finally {
      _eventCount++;
    }
  }

  /**
   * Führt eine Query aus. Löst ein [QueryHandler] eine Exception aus, wird eine Map mit {"error": "msg"}
   * zurückgegeben.
   * Aktuell werden die QueryHandler in einer konsistenten Transaction *parallel* ausgeführt.
   * */
  Future<Map> submitQuery(Map q, int user) async {
    if (!queryHandler.containsKey(q["action"]))
      throw new Exception("Keine Verarbeitung für ${q["action"]} hinterlegt");

    final List<QueryHandler> handlers = queryHandler[q["action"]];
    final Map finalOutput = {};
    q["timestamp"] = new DateTime.now().millisecondsSinceEpoch;
    q["user"] = user;

    // final Transaction trans = await db.startTransaction(consistent: true);

    final Iterable<Future> handlerRunners = handlers.map((h) async {
      var qo;
      try {
        qo = await h(new Map.from(q), db);
      } catch (e) {
        rethrow;
      }

      if (qo is Map) qo.forEach((k, v) => finalOutput[k] = v);
    });

    await Future.wait(handlerRunners, eagerError: true);

    // FinalOutput enthält alle Ausgaben (return to sender)
    return finalOutput;
  }

  int getNewEventId() => ++_biggestKnownEventId;

  /**
   * Spielt ein Event neu ab. Die EventHandler werden alle *nacheinander* in *einer* Transaktion ausgeführt.
   * Wirft einer der [EventHandler] eine Exception, wird ein Rollback ausgeführt und eine Map mit {"error": "msg"}
   * zurückgegeben.
   * Unbekannte Events lösen eine Exception aus. Jedes Event sollte das Feld "action" definieren.
   * **NOTE**: This method does not check auth.
   * */
  Future<Map> submitEvent(Map e, Map hookData, int user) async {
    if (!eventHandler.containsKey(e["action"]))
      throw new Exception("No event handlers defined for ${e["action"]}");

    final List<EventHandler> handlers = eventHandler[e["action"]];
    final Map finalInput = {
      "timestamp": new DateTime.now().millisecondsSinceEpoch,
      "id": getNewEventId(),
      "user": user
    };
    e.addAll(finalInput);
    final Transaction trans = await db.startTransaction(consistent: true);

    try {
      for (EventHandler h in handlers) {
        await h(new Map.from(e), trans);
      }

      for (EventHandler hook in pushHooks) {
{'fraction_non_alphanumeric': 0.11103536429484448, 'fraction_numerical': 0.0006817213463996591, 'mean_word_length': 3.550601008142691, 'pattern_counts': {'":': 10, '<': 31, '<?xml version=': 0, '>': 36, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3481382', 'n_tokens_mistral': 3575, 'n_tokens_neox': 3590, 'n_words': 1076}