Stream writers, event creators, auth worker

6 files changed, 251 insertions, 70 deletions

diff --git a/host/Rory-nginx/services/matrix/synapse/synapse-main.nix b/host/Rory-nginx/services/matrix/synapse/synapse-main.nix
index 5202ac1..d69d7c6 100755
--- a/host/Rory-nginx/services/matrix/synapse/synapse-main.nix
+++ b/host/Rory-nginx/services/matrix/synapse/synapse-main.nix
@@ -17,6 +17,7 @@
     mediaRepoWorkers = 4;
     clientReaders = 4;
     syncWorkers = 4;
+    authWorkers = 1;
 
     federationReaders = 8;
     federationInboundWorkers = 8;
@@ -25,8 +26,8 @@
     enableBackgroundWorker = true;
     enableUserDirWorker = true;
 
-    enableStreamWriters = false;
-    #eventStreamWriters = 8;
+    #enableStreamWriters = false;
+    eventStreamWriters = 8;
 
     # https://matrix-org.github.io/synapse/latest/usage/configuration/config_documentation.html
     settings = {
diff --git a/host/Rory-nginx/services/matrix/synapse/workers/auth.nix b/host/Rory-nginx/services/matrix/synapse/workers/auth.nix
index 6b9e250..37cdc85 100644
--- a/host/Rory-nginx/services/matrix/synapse/workers/auth.nix
+++ b/host/Rory-nginx/services/matrix/synapse/workers/auth.nix
@@ -18,6 +18,8 @@ let
       "~ ^/_matrix/client/(r0|v3|unstable)/password_policy$"
       "~ ^/_matrix/client/(r0|v3|unstable)/capabilities$"
     ];
+    federation = [ ];
+    media = [ ];
   };
 in
 {
@@ -73,38 +75,40 @@ in
       );
     };
 
-    services.nginx.upstreams."${workerName}" = {
-      extraConfig = ''
-        keepalive 32;
-      '';
-      servers = lib.listToAttrs (
-        lib.map (index: {
-          name = "unix:/run/matrix-synapse/${workerName}-client-${toString index}.sock";
+    services.nginx.upstreams = (
+      lib.listToAttrs (
+        lib.map (type: {
+          name = "${workerName}-${type}";
           value = {
-            max_fails = 0;
+            extraConfig = ''
+              keepalive 32;
+            '';
+            servers = lib.listToAttrs (
+              lib.map (index: {
+                name = "unix:/run/matrix-synapse/${workerName}-${type}-${toString index}.sock";
+                value = {
+                  max_fails = 0;
+                };
+              }) workers
+            );
           };
-        }) workers
-      );
-    };
+        }) ([ ] ++ lib.optional (lib.length workerRoutes.client > 0) "client" ++ lib.optional (lib.length workerRoutes.federation > 0) "federation" ++ lib.optional (lib.length workerRoutes.media > 0) "media")
+      )
+    );
 
     services.nginx = {
       virtualHosts."${cfg.nginxVirtualHostName}".locations = lib.listToAttrs (
         lib.flatten (
-          lib.forEach
-            [
-              "client"
-              "federation"
-              "media"
-            ]
-            (
-              type:
-              lib.map (route: {
-                name = route;
-                value = {
-                  proxyPass = "http://unix:/run/matrix-synapse/${workerName}-${type}.sock";
-                };
-              }) workerRoutes.${type}
-            )
+          lib.forEach ([ ] ++ lib.optional (lib.length workerRoutes.client > 0) "client" ++ lib.optional (lib.length workerRoutes.federation > 0) "federation" ++ lib.optional (lib.length workerRoutes.media > 0) "media") (
+            type:
+            lib.map (route: {
+              name = route;
+              value = {
+                proxyPass = "http://${workerName}-${type}";
+
+              };
+            }) workerRoutes.${type}
+          )
         )
       );
     };
diff --git a/host/Rory-nginx/services/matrix/synapse/workers/event-creator.nix b/host/Rory-nginx/services/matrix/synapse/workers/event-creator.nix
new file mode 100644
index 0000000..c4b9c7f
--- /dev/null
+++ b/host/Rory-nginx/services/matrix/synapse/workers/event-creator.nix
@@ -0,0 +1,111 @@
+{ config, lib, ... }:
+
+let
+  cfg = config.services.matrix-synapse;
+  dbGroup = "medium";
+  workers = lib.range 0 (cfg.eventCreators - 1);
+  workerName = "event_creator";
+  workerRoutes = {
+    client =
+      [
+        "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/redact"
+        "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/send"
+        "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/(join|invite|leave|ban|unban|kick)$"
+        "^/_matrix/client/(api/v1|r0|v3|unstable)/join/"
+        "^/_matrix/client/(api/v1|r0|v3|unstable)/knock/"
+        "^/_matrix/client/(api/v1|r0|v3|unstable)/profile/"
+      ];
+    federation = [ ];
+    media = [ ];
+  };
+in
+{
+  config = lib.mkIf (cfg.eventCreators > 0) {
+    services.matrix-synapse = {
+      settings = {
+        instance_map = lib.listToAttrs (
+          lib.map (index: {
+            name = "${workerName}-${toString index}";
+            value = {
+              path = "/run/matrix-synapse/${workerName}-${toString index}.sock";
+            };
+          }) workers
+        );
+      };
+
+      workers = lib.listToAttrs (
+        lib.map (index: {
+          name = "${workerName}-${toString index}";
+          value = {
+            worker_app = "synapse.app.generic_worker";
+            worker_listeners = [
+              {
+                type = "http";
+                path = "/run/matrix-synapse/${workerName}-${toString index}.sock";
+                resources = [
+                  {
+                    names = [ "replication" ];
+                    compress = false;
+                  }
+                ];
+              }
+              {
+                type = "http";
+                path = "/run/matrix-synapse/${workerName}-client-${toString index}.sock";
+                mode = "666";
+                resources = [
+                  {
+                    names = [ "client" ];
+                    compress = false;
+                  }
+                ];
+              }
+            ];
+            database = (
+              import ../db.nix {
+                inherit dbGroup;
+                workerName = "${workerName}-${toString index}";
+              }
+            );
+          };
+        }) workers
+      );
+    };
+
+    services.nginx.upstreams."${workerName}" = {
+      extraConfig = ''
+        keepalive 32;
+      '';
+      servers = lib.listToAttrs (
+        lib.map (index: {
+          name = "unix:/run/matrix-synapse/${workerName}-client-${toString index}.sock";
+          value = {
+            max_fails = 0;
+          };
+        }) workers
+      );
+    };
+
+    services.nginx = {
+      virtualHosts."${cfg.nginxVirtualHostName}".locations = lib.listToAttrs (
+        lib.flatten (
+          lib.forEach
+            [
+              "client"
+              "federation"
+              "media"
+            ]
+            (
+              type:
+              lib.map (route: {
+                name = route;
+                value = {
+                  proxyPass = "http://unix:/run/matrix-synapse/${workerName}-${type}.sock";
+                };
+              }) workerRoutes.${type}
+            )
+        )
+      );
+    };
+  };
+}
diff --git a/host/Rory-nginx/services/matrix/synapse/workers/module.nix b/host/Rory-nginx/services/matrix/synapse/workers/module.nix
index 6615809..87e014e 100644
--- a/host/Rory-nginx/services/matrix/synapse/workers/module.nix
+++ b/host/Rory-nginx/services/matrix/synapse/workers/module.nix
@@ -14,8 +14,10 @@ in
     ./single/appservice.nix
     ./single/background.nix
     ./single/user-dir.nix
+    
     ./auth.nix
     ./client-reader.nix
+    ./event-creator.nix
     ./federation-inbound.nix
     ./federation-reader.nix
     ./federation-sender.nix
@@ -23,7 +25,7 @@ in
     ./pusher.nix
     ./sync.nix
 
-    #./stream-writers/event-stream-writer.nix
+    ./stream-writers/event-stream-writer.nix
   ];
   options.services.matrix-synapse = {
     enableWorkers = lib.mkEnableOption "Enable dedicated workers";
@@ -33,13 +35,14 @@ in
     enableUserDirWorker = lib.mkEnableOption "Enable dedicated user directory worker";
 
     authWorkers = mkIntOption "Number of auth workers";
+    clientReaders = mkIntOption "Number of client readers";
+    eventCreators = mkIntOption "Number of auth workers";
+    federationInboundWorkers = mkIntOption "Number of federation inbound workers";
+    federationReaders = mkIntOption "Number of federation readers";
     federationSenders = mkIntOption "Number of federation senders";
-    pushers = mkIntOption "Number of pushers";
     mediaRepoWorkers = mkIntOption "Number of media repo workers";
+    pushers = mkIntOption "Number of pushers";
     syncWorkers = mkIntOption "Number of sync workers";
-    clientReaders = mkIntOption "Number of client readers";
-    federationReaders = mkIntOption "Number of federation readers";
-    federationInboundWorkers = mkIntOption "Number of federation inbound workers";
 
     #stream writers
     eventStreamWriters = mkIntOption "Number of event stream writers";
diff --git a/host/Rory-nginx/services/matrix/synapse/workers/single/user-dir.nix b/host/Rory-nginx/services/matrix/synapse/workers/single/user-dir.nix
index 4fd3fb3..56dbd5f 100644
--- a/host/Rory-nginx/services/matrix/synapse/workers/single/user-dir.nix
+++ b/host/Rory-nginx/services/matrix/synapse/workers/single/user-dir.nix
@@ -5,11 +5,12 @@ let
   dbGroup = "solo";
   workerName = "user_dir";
   workerRoutes = {
-    client = [
-      "~ ^/_matrix/client/(api/v1|r0|v3|unstable)/user_directory/search$"
-      "~ ^/_matrix/client/v3/profile/.*/(displayname|avatar_url)$"
-      "~ ^/_matrix/client/v3/profile/.*$"
-    ];
+    client =
+      [ "~ ^/_matrix/client/(api/v1|r0|v3|unstable)/user_directory/search$" ]
+      ++ lib.optionals (cfg.authWorkers == 0) [
+        "~ ^/_matrix/client/v3/profile/.*$"
+        "~ ^/_matrix/client/v3/profile/.*/(displayname|avatar_url)$"
+      ];
     federation = [ ];
     media = [ ];
   };
diff --git a/host/Rory-nginx/services/matrix/synapse/workers/stream-writers/event-stream-writer.nix b/host/Rory-nginx/services/matrix/synapse/workers/stream-writers/event-stream-writer.nix
index 1f5b638..e17fa95 100644
--- a/host/Rory-nginx/services/matrix/synapse/workers/stream-writers/event-stream-writer.nix
+++ b/host/Rory-nginx/services/matrix/synapse/workers/stream-writers/event-stream-writer.nix
@@ -1,45 +1,106 @@
 { config, lib, ... }:
+
 let
   cfg = config.services.matrix-synapse;
-  streamWriters = lib.range 0 (cfg.eventStreamWriters - 1);
-  streamWriterType = "events";
+  dbGroup = "medium";
+  streamWriterType = "event";
+  workers = lib.range 0 (cfg.eventStreamWriters - 1);
+  workerName = "event_stream_writer";
+  workerRoutes = {
+    client = [ ];
+    federation = [ ];
+    media = [ ];
+  };
 in
 {
-  services.matrix-synapse = lib.mkIf (cfg.eventStreamWriters > 0) {
-    settings = {
-      instance_map = lib.listToAttrs (
-        lib.map (port: {
-          name = "stream-writer-${streamWriterType}-${toString port}";
+  config = lib.mkIf (cfg.eventStreamWriters > 0) {
+    services.matrix-synapse = {
+      settings = {
+        instance_map = lib.listToAttrs (
+          lib.map (index: {
+            name = "${workerName}-${toString index}";
+            value = {
+              path = "/run/matrix-synapse/${workerName}-${toString index}.sock";
+            };
+          }) workers
+        );
+
+        stream_writers.${streamWriterType} = lib.map (index: "${workerName}-${toString index}") workers;
+      };
+
+      workers = lib.listToAttrs (
+        lib.map (index: {
+          name = "${workerName}-${toString index}";
           value = {
-            path = "/run/matrix-synapse/stream_writer-${streamWriterType}-${toString port}.sock";
+            worker_app = "synapse.app.generic_worker";
+            worker_listeners = [
+              {
+                type = "http";
+                path = "/run/matrix-synapse/${workerName}-${toString index}.sock";
+                resources = [
+                  {
+                    names = [ "replication" ];
+                    compress = false;
+                  }
+                ];
+              }
+              {
+                type = "http";
+                path = "/run/matrix-synapse/${workerName}-client-${toString index}.sock";
+                mode = "666";
+                resources = [
+                  {
+                    names = [ "client" ];
+                    compress = false;
+                  }
+                ];
+              }
+            ];
+            database = (
+              import ../../db.nix {
+                inherit dbGroup;
+                workerName = "${workerName}-${toString index}";
+              }
+            );
           };
-        }) streamWriters
+        }) workers
       );
+    };
 
-      stream_writers = {
-        events = lib.map (index: "stream_writer-${streamWriterType}-${toString index}") streamWriters;
-      };
+    services.nginx.upstreams."${workerName}" = {
+      extraConfig = ''
+        keepalive 32;
+      '';
+      servers = lib.listToAttrs (
+        lib.map (index: {
+          name = "unix:/run/matrix-synapse/${workerName}-client-${toString index}.sock";
+          value = {
+            max_fails = 0;
+          };
+        }) workers
+      );
     };
 
-    workers = lib.listToAttrs (
-      lib.map (index: {
-        name = "stream-writer-${toString index}";
-        value = {
-          worker_app = "synapse.app.generic_worker";
-          worker_listeners = [
-            {
-              type = "http";
-              path = "/run/matrix-synapse/stream_writer-${streamWriterType}-${toString index}.sock";
-              resources = [
-                {
-                  names = [ "replication" ];
-                  compress = false;
-                }
-              ];
-            }
-          ];
-        };
-      }) streamWriters
-    );
+    services.nginx = {
+      virtualHosts."${cfg.nginxVirtualHostName}".locations = lib.listToAttrs (
+        lib.flatten (
+          lib.forEach
+            [
+              "client"
+              "federation"
+              "media"
+            ]
+            (
+              type:
+              lib.map (route: {
+                name = route;
+                value = {
+                  proxyPass = "http://unix:/run/matrix-synapse/${workerName}-${type}.sock";
+                };
+              }) workerRoutes.${type}
+            )
+        )
+      );
+    };
   };
 }