export: Plumb consented_user_ids to export_usermessage_batch in a file.

This allows us to get rid of the call to `get_consented_user_ids` in
`fetch_usermessages`. Now it's only called at the beginning of the
export, eliminating the redundant db query and also resolving the
potential for data consistency issues, if some users change their
consent setting after the export starts.

Now the full export process operates with a single snapshot of these
consenting user ids.

These ids need to be plumbed through via a file rather than normal arg
passing, because this is a separate management command, run in
subprocesses during the export.

zerver/management/commands/export_usermessage_batch.py

@@ -4,6 +4,7 @@ import os
 from argparse import ArgumentParser
 from typing import Any
 
+import orjson
 from typing_extensions import override
 
 from zerver.lib.export import export_usermessages_batch
@@ -26,7 +27,18 @@ class Command(ZulipBaseCommand):
     @override
     def handle(self, *args: Any, **options: Any) -> None:
         logging.info("Starting UserMessage batch thread %s", options["thread"])
-        files = set(glob.glob(os.path.join(options["path"], "messages-*.json.partial")))
+        path = options["path"]
+        files = set(glob.glob(os.path.join(path, "messages-*.json.partial")))
+
+        export_full_with_consent = options["export_full_with_consent"]
+        consented_user_ids = None
+        if export_full_with_consent:
+            consented_user_ids_path = os.path.join(path, "consented_user_ids.json")
+            assert os.path.exists(consented_user_ids_path)
+
+            with open(consented_user_ids_path, "rb") as f:
+                consented_user_ids = set(orjson.loads(f.read()))
+
         for partial_path in files:
             locked_path = partial_path.replace(".json.partial", ".json.locked")
             output_path = partial_path.replace(".json.partial", ".json")
@@ -38,7 +50,10 @@ class Command(ZulipBaseCommand):
             logging.info("Thread %s processing %s", options["thread"], output_path)
             try:
                 export_usermessages_batch(
-                    locked_path, output_path, options["export_full_with_consent"]
+                    locked_path,
+                    output_path,
+                    export_full_with_consent,
+                    consented_user_ids=consented_user_ids,
                 )
             except BaseException:
                 # Put the item back in the free pool when we fail