presence stream writer

flake.lock

@@ -0,0 +1,26 @@
+{
+  "nodes": {
+    "nixpkgs-lib": {
+      "locked": {
+        "lastModified": 1673743903,
+        "narHash": "sha256-sloY6KYyVOozJ1CkbgJPpZ99TKIjIvM+04V48C04sMQ=",
+        "owner": "nix-community",
+        "repo": "nixpkgs.lib",
+        "rev": "7555e2dfcbac1533f047021f1744ac8871150f9f",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nix-community",
+        "repo": "nixpkgs.lib",
+        "type": "github"
+      }
+    },
+    "root": {
+      "inputs": {
+        "nixpkgs-lib": "nixpkgs-lib"
+      }
+    }
+  },
+  "root": "root",
+  "version": 7
+}

flake.nix

@@ -1,9 +1,14 @@
 {
   description = "NixOS modules for matrix related services";
 
-  outputs = { self }: {
+  inputs = {
+    nixpkgs-lib.url = github:nix-community/nixpkgs.lib;
+  };
+
+  outputs = { self, nixpkgs-lib }: {
     nixosModules = {
-      synapse = import ./synapse-module;
+      synapse = import ./synapse-module { matrix-lib = self.lib; };
     };
+    lib = import ./lib.nix { lib = nixpkgs-lib.lib; };
   };
 }

lib.nix

@@ -0,0 +1,20 @@
+{ lib }:
+rec {
+  # checks if given listener configuration has type as a resource
+  isListenerType = type: l: lib.any (r: lib.any (n: n == type) r.names) l.resources;
+  # Get the first listener that includes the given resource from worker
+  firstListenerOfType = type: ls: lib.lists.findFirst (isListenerType type)
+    (lib.throw "No listener with resource: ${type} configured")
+    ls;
+  # Get an attrset of the host and port from a listener 
+  connectionInfo = l: {
+    host = lib.head l.bind_addresses;
+    port = l.port;
+  };
+
+  # Get an attrset of the host and port from a worker given a type
+  workerConnectionResource = r: w: let
+    l = firstListenerOfType r w.settings.worker_listeners;
+  in connectionInfo l;
+
+}

synapse-module/default.nix

@@ -1,3 +1,4 @@
+{ matrix-lib }:
 { pkgs, lib, config, ... }:
 let
   cfg = config.services.matrix-synapse-next;
@@ -31,7 +32,7 @@ in
   imports = [
     ./nginx.nix
     (import ./workers.nix {
-      inherit throw' format matrix-synapse-common-config pluginsEnv;
+      inherit matrix-lib throw' format matrix-synapse-common-config pluginsEnv;
     })
   ];
 

synapse-module/nginx.nix

@@ -117,6 +117,7 @@ in
         synapse_federation_transaction synapse_worker_federation;
 
         synapse_client_user-dir synapse_worker_user-dir;
+        synapse_client_presence synapse_worker_stream-presence;
       }
 
       # from https://github.com/tswfi/synapse/commit/b3704b936663cc692241e978dce4ac623276b1a6
@@ -187,6 +188,14 @@ in
       '';
     };
 
+    services.nginx.upstreams.synapse_worker_stream-presence = {
+      servers = let
+        workers = getWorkersOfType "stream-presence";
+        socketAddresses = generateSocketAddresses "client" workers;
+      in if workers != { } then
+        lib.genAttrs socketAddresses (_: { })
+      else config.services.nginx.upstreams.synapse_master.servers;
+    };
 
     services.nginx.upstreams.synapse_worker_user-dir = {
       servers = let

synapse-module/workers.nix

@@ -1,4 +1,5 @@
 { matrix-synapse-common-config,
+  matrix-lib,
   pluginsEnv,
   throw',
   format
@@ -22,17 +23,7 @@
 
   genAttrs' = items: f: g: builtins.listToAttrs (map (i: lib.nameValuePair (f i) (g i)) items);
 
-  isReplicationListener =
-    l: lib.any (r: lib.any (n: n == "replication") r.names) l.resources;
-
-  mainReplicationListener = lib.lists.findFirst isReplicationListener
-    (throw' "No replication listener configured!")
-    cfg.settings.listeners;
-  mainReplicationListenerHost =
-    if mainReplicationListener.bind_addresses == []
-      then throw' "Replication listener had no addresses"
-      else builtins.head mainReplicationListener.bind_addresses;
-  mainReplicationListenerPort = mainReplicationListener.port;
+  mainReplicationListener = matrix-lib.firstListenerOfType "replication" cfg.settings.listeners;
 in {
   # See https://github.com/matrix-org/synapse/blob/develop/docs/workers.md for more info
   options.services.matrix-synapse-next.workers = let
@@ -185,17 +176,20 @@ in {
   in {
     mainReplicationHost = mkOption {
       type = types.str;
-      default =
-        if builtins.elem mainReplicationListenerHost [ "0.0.0.0" "::" ]
+      default = let
+        host = (matrix-lib.connectionInfo mainReplicationListener).host;
+      in
+        # To avoid connecting to 0.0.0.0 and so on
+        if builtins.elem host [ "0.0.0.0" "::" ]
           then "127.0.0.1"
-          else mainReplicationListenerHost;
+          else host;
       # TODO: add defaultText
       description = "Host of the main synapse instance's replication listener";
     };
 
     mainReplicationPort = mkOption {
       type = types.port;
-      default = mainReplicationListenerPort;
+      default = mainReplicationListener.port;
       # TODO: add defaultText
       description = "Port for the main synapse instance's replication listener";
     };
@@ -232,6 +226,7 @@ in {
     eventPersisters = mkWorkerCountOption "event-persister";
 
     useUserDirectoryWorker = mkEnableOption "user directory worker";
+    usePresenceStreamWriter = mkEnableOption "prescence stream writer";
 
     instances = mkOption {
       type = types.attrsOf workerInstanceType;
@@ -260,16 +255,8 @@ in {
       instance_map = genAttrs' (lib.lists.range 1 wcfg.eventPersisters)
         (i: "auto-event-persist${toString i}")
         (i: let
-          wRL = lib.lists.findFirst isReplicationListener
-            (throw' "No replication listener configured!")
-            wcfg.instances."auto-event-persist${toString i}".settings.worker_listeners;
-          wRH = lib.findFirst (x: true) (throw' "Replication listener had no addresses")
-            wRL.bind_addresses;
-          wRP = wRL.port;
-        in {
-          host = wRH;
-          port = wRP;
-        });
+          wRL = matrix-lib.firstListenerOfType "replication" wcfg.instances."auto-event-persist${toString i}".settings.worker_listeners;
+        in matrix-lib.connectionInfo wRL);
 
       stream_writers.events =
         mkIf (wcfg.eventPersisters > 0)
@@ -317,6 +304,11 @@ in {
           numberOfWorkers = 1;
           nameFn = _: "auto-user-dir";
         };
+      }) // (lib.optionalAttrs wcfg.usePresenceStreamWriter {
+        "stream-presence" = {
+          numberOfWorkers = 1;
+          nameFn = _: "auto-stream-presence";
+        };
       });
 
       coerceWorker = { name, value }: if builtins.isInt value then {