streams: Add API for changing stream-level message_retention_days.

This commit adds backend support for setting message_retention_days
while creating streams and updating it for an existing stream. We only
allow organization owners to set/update it for a stream.

'message_retention_days' field for a stream existed previously also, but
there was no way to set it while creating streams or update it for an
exisiting streams using any endpoint.

templates/zerver/api/changelog.md

@@ -10,6 +10,14 @@ below features are supported.
 
 ## Changes in Zulip 2.2
 
+**Feature level 17**
+
+* [`GET users/me/subscriptions`](/api/get-subscribed-streams), [`GET /streams`]
+  (api/get-all-streams): Added `message_retention_days` to the Stream objects.
+* [`POST users/me/subscriptions`](/api/add-subscriptions), [`PATCH
+  streams/{stream_id}`](/api/update-stream): Added `message_retention_days`
+  parameter.
+
 **Feature level 16**
 
 * [`GET /users/me`]: Removed `pointer` from the response, as the

version.py

@@ -29,7 +29,7 @@ DESKTOP_WARNING_VERSION = "5.2.0"
 #
 # Changes should be accompanied by documentation explaining what the
 # new level means in templates/zerver/api/changelog.md.
-API_FEATURE_LEVEL = 16
+API_FEATURE_LEVEL = 17
 
 # Bump the minor PROVISION_VERSION to indicate that folks should provision
 # only when going from an old version of the code to a newer version. Bump

zerver/lib/actions.py

@@ -3613,6 +3613,20 @@ def do_change_stream_description(stream: Stream, new_description: str) -> None:
     )
     send_event(stream.realm, event, can_access_stream_user_ids(stream))
 
+def do_change_stream_message_retention_days(stream: Stream, message_retention_days: Optional[int]=None) -> None:
+    stream.message_retention_days = message_retention_days
+    stream.save(update_fields=['message_retention_days'])
+
+    event = dict(
+        op="update",
+        type="stream",
+        property="message_retention_days",
+        value=message_retention_days,
+        stream_id=stream.id,
+        name=stream.name,
+    )
+    send_event(stream.realm, event, can_access_stream_user_ids(stream))
+
 def do_create_realm(string_id: str, name: str,
                     emails_restricted_to_domains: Optional[bool]=None) -> Realm:
     if Realm.objects.filter(string_id=string_id).exists():

zerver/lib/streams.py

@@ -58,7 +58,8 @@ def create_stream_if_needed(realm: Realm,
                             invite_only: bool=False,
                             stream_post_policy: int=Stream.STREAM_POST_POLICY_EVERYONE,
                             history_public_to_subscribers: Optional[bool]=None,
-                            stream_description: str="") -> Tuple[Stream, bool]:
+                            stream_description: str="",
+                            message_retention_days: Optional[int]=None) -> Tuple[Stream, bool]:
     history_public_to_subscribers = get_default_value_for_history_public_to_subscribers(
         realm, invite_only, history_public_to_subscribers)
 
@@ -72,6 +73,7 @@ def create_stream_if_needed(realm: Realm,
             stream_post_policy=stream_post_policy,
             history_public_to_subscribers=history_public_to_subscribers,
             is_in_zephyr_realm=realm.is_zephyr_mirror_realm,
+            message_retention_days=message_retention_days,
         ),
     )
 
@@ -104,6 +106,7 @@ def create_streams_if_needed(realm: Realm,
             stream_post_policy=stream_dict.get("stream_post_policy", Stream.STREAM_POST_POLICY_EVERYONE),
             history_public_to_subscribers=stream_dict.get("history_public_to_subscribers"),
             stream_description=stream_dict.get("description", ""),
+            message_retention_days=stream_dict.get("message_retention_days", None)
         )
 
         if created:
@@ -423,10 +426,13 @@ def list_to_streams(streams_raw: Iterable[Mapping[str, Any]],
     missing_stream_dicts: List[Mapping[str, Any]] = []
     existing_stream_map = bulk_get_streams(user_profile.realm, stream_set)
 
+    message_retention_days_not_none = False
     for stream_dict in streams_raw:
         stream_name = stream_dict["name"]
         stream = existing_stream_map.get(stream_name.lower())
         if stream is None:
+            if stream_dict.get('message_retention_days', None) is not None:
+                message_retention_days_not_none = True
             missing_stream_dicts.append(stream_dict)
         else:
             existing_streams.append(stream)
@@ -443,6 +449,10 @@ def list_to_streams(streams_raw: Iterable[Mapping[str, Any]],
             raise JsonableError(_("Stream(s) ({}) do not exist").format(
                 ", ".join(stream_dict["name"] for stream_dict in missing_stream_dicts),
             ))
+        elif message_retention_days_not_none:
+            if not user_profile.is_realm_owner:
+                raise JsonableError(_('User cannot create stream with this settings.'))
+            user_profile.realm.ensure_not_on_limited_plan()
 
         # We already filtered out existing streams, so dup_streams
         # will normally be an empty list below, but we protect against somebody

zerver/models.py

@@ -1507,8 +1507,6 @@ class Stream(models.Model):
     # * "realm" and "recipient" are not exposed to clients via the API.
     # * "date_created" should probably be added here, as it's useful information
     #   to subscribers.
-    # * message_retention_days should be added here once the feature is
-    #   complete.
     API_FIELDS = [
         "name",
         "id",
@@ -1519,6 +1517,7 @@ class Stream(models.Model):
         "stream_post_policy",
         "history_public_to_subscribers",
         "first_message_id",
+        "message_retention_days"
     ]
 
     @staticmethod

zerver/openapi/zulip.yaml

@@ -2337,6 +2337,19 @@ paths:
 
                               **Changes**: New in Zulip 2.2, replacing the previous
                               `is_announcement_only` boolean.
+                          message_retention_days:
+                            type: integer
+                            nullable: true
+                            description: |
+                              Number of days that messages sent to this stream will be stored
+                              before being automatically deleted by the [message retention
+                              policy](/help/message-retention-policy).  There are two special values:
+
+                              * `null`, the default, means the stream will inherit the organization
+                                level setting.
+                              * `-1` encodes retaining messages in this stream forever.
+
+                              **Changes**: New in Zulip 2.2 (feature level 17).
                           history_public_to_subscribers:
                             type: boolean
                             description: |
@@ -2463,6 +2476,7 @@ paths:
         example: false
       - $ref: '#/components/parameters/HistoryPublicToSubscribers'
       - $ref: '#/components/parameters/StreamPostPolicy'
+      - $ref: '#/components/parameters/MessageRetentionDays'
       - name: announce
         in: query
         description: |
@@ -3813,6 +3827,19 @@ paths:
 
                               **Changes**: New in Zulip 2.2, replacing the previous
                               `is_announcement_only` boolean.
+                          message_retention_days:
+                            type: integer
+                            nullable: true
+                            description: |
+                              Number of days that messages sent to this stream will be stored
+                              before being automatically deleted by the [message retention
+                              policy](/help/message-retention-policy).  There are two special values:
+
+                              * `null`, the default, means the stream will inherit the organization
+                                level setting.
+                              * `-1` encodes retaining messages in this stream forever.
+
+                              **Changes**: New in Zulip 2.2 (feature level 17).
                           history_public_to_subscribers:
                             type: boolean
                             description: |
@@ -3988,6 +4015,7 @@ paths:
         required: false
       - $ref: '#/components/parameters/StreamPostPolicy'
       - $ref: '#/components/parameters/HistoryPublicToSubscribers'
+      - $ref: '#/components/parameters/MessageRetentionDays'
       responses:
         '200':
           description: Success.
@@ -4975,3 +5003,22 @@ components:
         type: string
       example: '1f419'
       required: false
+    MessageRetentionDays:
+      name: message_retention_days
+      in: query
+      description: |
+        Number of days that messages sent to this stream will be stored
+        before being automatically deleted by the [message retention
+        policy](/help/message-retention-policy).  Two special string format
+        values are supported:
+
+        * "realm_default" => Return to the organization-level setting.
+        * "forever" => Retain messages forever.
+
+        **Changes**: New in Zulip 2.2 (feature level 17).
+      schema:
+        oneOf:
+          - type: string
+          - type: integer
+      example: '20'
+      required: false

zerver/tests/test_events.py

@@ -46,6 +46,7 @@ from zerver.lib.actions import (
     do_change_realm_domain,
     do_change_stream_description,
     do_change_stream_invite_only,
+    do_change_stream_message_retention_days,
     do_change_stream_post_policy,
     do_change_subscription_property,
     do_change_user_delivery_email,
@@ -1451,6 +1452,7 @@ class EventsRegisterTest(ZulipTestCase):
                     ('is_web_public', check_bool),
                     ('is_announcement_only', check_bool),
                     ('stream_post_policy', check_int_in(Stream.STREAM_POST_POLICY_TYPES)),
+                    ('message_retention_days', check_none_or(check_int)),
                     ('name', check_string),
                     ('stream_id', check_int),
                     ('first_message_id', check_none_or(check_int)),
@@ -2564,6 +2566,7 @@ class EventsRegisterTest(ZulipTestCase):
             ('is_web_public', check_bool),
             ('is_announcement_only', check_bool),
             ('stream_post_policy', check_int_in(Stream.STREAM_POST_POLICY_TYPES)),
+            ('message_retention_days', check_none_or(check_int)),
             ('is_muted', check_bool),
             ('in_home_view', check_bool),
             ('name', check_string),
@@ -2647,6 +2650,14 @@ class EventsRegisterTest(ZulipTestCase):
             ('name', check_string),
             ('value', check_int_in(Stream.STREAM_POST_POLICY_TYPES)),
         ])
+        stream_update_message_retention_days_schema_checker = self.check_events_dict([
+            ('type', equals('stream')),
+            ('op', equals('update')),
+            ('property', equals('message_retention_days')),
+            ('stream_id', check_int),
+            ('name', check_string),
+            ('value', check_none_or(check_int))
+        ])
 
         # Subscribe to a totally new stream, so it's just Hamlet on it
         action: Callable[[], object] = lambda: self.subscribe(self.example_user("hamlet"), "test_stream")
@@ -2717,6 +2728,12 @@ class EventsRegisterTest(ZulipTestCase):
         error = stream_update_stream_post_policy_schema_checker('events[0]', events[0])
         self.assert_on_error(error)
 
+        action = lambda: do_change_stream_message_retention_days(stream, -1)
+        events = self.do_test(action,
+                              include_subscribers=include_subscribers, num_events=1)
+        error = stream_update_message_retention_days_schema_checker('events[0]', events[0])
+        self.assert_on_error(error)
+
         # Subscribe to a totally new invite-only stream, so it's just Hamlet on it
         stream = self.make_stream("private", self.user_profile.realm, invite_only=True)
         user_profile = self.example_user('hamlet')

zerver/tests/test_subs.py

@@ -21,6 +21,7 @@ from zerver.lib.actions import (
     do_add_streams_to_default_stream_group,
     do_change_default_stream_group_description,
     do_change_default_stream_group_name,
+    do_change_plan_type,
     do_change_stream_post_policy,
     do_change_user_role,
     do_create_default_stream_group,
@@ -146,7 +147,8 @@ class TestCreateStreams(ZulipTestCase):
             [{"name": stream_name,
               "description": stream_description,
               "invite_only": True,
-              "stream_post_policy": Stream.STREAM_POST_POLICY_ADMINS}
+              "stream_post_policy": Stream.STREAM_POST_POLICY_ADMINS,
+              "message_retention_days": -1}
              for (stream_name, stream_description) in zip(stream_names, stream_descriptions)])
 
         self.assertEqual(len(new_streams), 3)
@@ -159,6 +161,7 @@ class TestCreateStreams(ZulipTestCase):
         for stream in new_streams:
             self.assertTrue(stream.invite_only)
             self.assertTrue(stream.stream_post_policy == Stream.STREAM_POST_POLICY_ADMINS)
+            self.assertTrue(stream.message_retention_days == -1)
 
         new_streams, existing_streams = create_streams_if_needed(
             realm,
@@ -856,6 +859,159 @@ class StreamAdminTest(ZulipTestCase):
             stream = get_stream('stream_name1', user_profile.realm)
             self.assertEqual(stream.stream_post_policy, policy)
 
+    def test_change_stream_message_retention_days(self) -> None:
+        user_profile = self.example_user('desdemona')
+        self.login_user(user_profile)
+        realm = user_profile.realm
+        do_change_plan_type(realm, Realm.LIMITED)
+        stream = self.subscribe(user_profile, 'stream_name1')
+
+        result = self.client_patch(f'/json/streams/{stream.id}',
+                                   {'message_retention_days': ujson.dumps(2)})
+        self.assert_json_error(result, "Available on Zulip Standard. Upgrade to access.")
+
+        do_change_plan_type(realm, Realm.SELF_HOSTED)
+        events: List[Mapping[str, Any]] = []
+        with tornado_redirected_to_list(events):
+            result = self.client_patch(f'/json/streams/{stream.id}',
+                                       {'message_retention_days': ujson.dumps(2)})
+        self.assert_json_success(result)
+
+        event = events[0]['event']
+        self.assertEqual(event, dict(
+            op='update',
+            type='stream',
+            property='message_retention_days',
+            value=2,
+            stream_id=stream.id,
+            name='stream_name1',
+        ))
+        notified_user_ids = set(events[0]['users'])
+        stream = get_stream('stream_name1', realm)
+
+        self.assertEqual(notified_user_ids, set(active_non_guest_user_ids(realm.id)))
+        self.assertIn(user_profile.id, notified_user_ids)
+        self.assertIn(self.example_user('prospero').id, notified_user_ids)
+        self.assertNotIn(self.example_user('polonius').id, notified_user_ids)
+        self.assertEqual(stream.message_retention_days, 2)
+
+        events = []
+        with tornado_redirected_to_list(events):
+            result = self.client_patch(f'/json/streams/{stream.id}',
+                                       {'message_retention_days': ujson.dumps("forever")})
+        self.assert_json_success(result)
+        event = events[0]['event']
+        self.assertEqual(event, dict(
+            op='update',
+            type='stream',
+            property='message_retention_days',
+            value=-1,
+            stream_id=stream.id,
+            name='stream_name1',
+        ))
+        self.assert_json_success(result)
+        stream = get_stream('stream_name1', realm)
+        self.assertEqual(stream.message_retention_days, -1)
+
+        events = []
+        with tornado_redirected_to_list(events):
+            result = self.client_patch(f'/json/streams/{stream.id}',
+                                       {'message_retention_days': ujson.dumps("realm_default")})
+        self.assert_json_success(result)
+        event = events[0]['event']
+        self.assertEqual(event, dict(
+            op='update',
+            type='stream',
+            property='message_retention_days',
+            value=None,
+            stream_id=stream.id,
+            name='stream_name1',
+        ))
+        stream = get_stream('stream_name1', realm)
+        self.assertEqual(stream.message_retention_days, None)
+
+        result = self.client_patch(f'/json/streams/{stream.id}',
+                                   {'message_retention_days': ujson.dumps("invalid")})
+        self.assert_json_error(result, "Bad value for 'message_retention_days': invalid")
+
+        result = self.client_patch(f'/json/streams/{stream.id}',
+                                   {'message_retention_days': ujson.dumps(-1)})
+        self.assert_json_error(result, "Bad value for 'message_retention_days': -1")
+
+    def test_change_stream_message_retention_days_requires_realm_owner(self) -> None:
+        user_profile = self.example_user('iago')
+        self.login_user(user_profile)
+        realm = user_profile.realm
+        stream = self.subscribe(user_profile, 'stream_name1')
+
+        result = self.client_patch(f'/json/streams/{stream.id}',
+                                   {'message_retention_days': ujson.dumps(2)})
+        self.assert_json_error(result, "Must be an organization owner")
+
+        do_change_user_role(user_profile, UserProfile.ROLE_REALM_OWNER)
+        result = self.client_patch(f'/json/streams/{stream.id}',
+                                   {'message_retention_days': ujson.dumps(2)})
+        self.assert_json_success(result)
+        stream = get_stream('stream_name1', realm)
+        self.assertEqual(stream.message_retention_days, 2)
+
+    def test_stream_message_retention_days_on_stream_creation(self) -> None:
+        """
+        Only admins can create streams with message_retention_days
+        with value other than None.
+        """
+        admin = self.example_user('iago')
+
+        streams_raw = [{
+            'name': 'new_stream',
+            'message_retention_days': 10,
+        }]
+        with self.assertRaisesRegex(JsonableError, "User cannot create stream with this settings."):
+            list_to_streams(streams_raw, admin, autocreate=True)
+
+        streams_raw = [{
+            'name': 'new_stream',
+            'message_retention_days': -1,
+        }]
+        with self.assertRaisesRegex(JsonableError, "User cannot create stream with this settings."):
+            list_to_streams(streams_raw, admin, autocreate=True)
+
+        streams_raw = [{
+            'name': 'new_stream',
+            'message_retention_days': None,
+        }]
+        result = list_to_streams(streams_raw, admin, autocreate=True)
+        self.assert_length(result[0], 0)
+        self.assert_length(result[1], 1)
+        self.assertEqual(result[1][0].name, 'new_stream')
+        self.assertEqual(result[1][0].message_retention_days, None)
+
+        owner = self.example_user('desdemona')
+        realm = owner.realm
+        streams_raw = [
+            {'name': 'new_stream1',
+             'message_retention_days': 10},
+            {'name': 'new_stream2',
+             'message_retention_days': -1},
+            {'name': 'new_stream3'},
+        ]
+
+        do_change_plan_type(realm, Realm.LIMITED)
+        with self.assertRaisesRegex(JsonableError, "Available on Zulip Standard. Upgrade to access."):
+            list_to_streams(streams_raw, owner, autocreate=True)
+
+
+        do_change_plan_type(realm, Realm.SELF_HOSTED)
+        result = list_to_streams(streams_raw, owner, autocreate=True)
+        self.assert_length(result[0], 0)
+        self.assert_length(result[1], 3)
+        self.assertEqual(result[1][0].name, 'new_stream1')
+        self.assertEqual(result[1][0].message_retention_days, 10)
+        self.assertEqual(result[1][1].name, 'new_stream2')
+        self.assertEqual(result[1][1].message_retention_days, -1)
+        self.assertEqual(result[1][2].name, 'new_stream3')
+        self.assertEqual(result[1][2].message_retention_days, None)
+
     def set_up_stream_for_deletion(self, stream_name: str, invite_only: bool=False,
                                    subscribed: bool=True) -> Stream:
         """

zerver/views/streams.py

@@ -33,6 +33,7 @@ from zerver.lib.actions import (
     do_change_default_stream_group_name,
     do_change_stream_description,
     do_change_stream_invite_only,
+    do_change_stream_message_retention_days,
     do_change_stream_post_policy,
     do_change_subscription_property,
     do_create_default_stream_group,
@@ -49,8 +50,8 @@ from zerver.lib.actions import (
     internal_prep_private_message,
     internal_prep_stream_message,
 )
-from zerver.lib.exceptions import ErrorCode, JsonableError
-from zerver.lib.request import REQ, has_request_variables
+from zerver.lib.exceptions import ErrorCode, JsonableError, OrganizationOwnerRequired
+from zerver.lib.request import REQ, RequestVariableConversionError, has_request_variables
 from zerver.lib.response import json_error, json_success
 from zerver.lib.streams import (
     access_default_stream_group_by_id,
@@ -73,6 +74,7 @@ from zerver.lib.validator import (
     check_int_in,
     check_list,
     check_string,
+    check_string_or_int,
     check_variable_type,
     to_non_negative_int,
 )
@@ -125,6 +127,16 @@ def check_if_removing_someone_else(user_profile: UserProfile,
     else:
         return principals[0] != user_profile.email
 
+def parse_message_retention_days(value: Union[int, str]) -> Optional[int]:
+    if value == "forever":
+        return -1
+    if value == "realm_default":
+        return None
+    if isinstance(value, str) or value < 0:
+        raise RequestVariableConversionError('message_retention_days', value)
+    assert isinstance(value, int)
+    return value
+
 @require_realm_admin
 def deactivate_stream_backend(request: HttpRequest,
                               user_profile: UserProfile,
@@ -224,10 +236,19 @@ def update_stream_backend(
             Stream.STREAM_POST_POLICY_TYPES), default=None),
         history_public_to_subscribers: Optional[bool]=REQ(validator=check_bool, default=None),
         new_name: Optional[str]=REQ(validator=check_string, default=None),
+        message_retention_days: Optional[Union[int, str]]=REQ(validator=check_string_or_int, default=None),
 ) -> HttpResponse:
     # We allow realm administrators to to update the stream name and
     # description even for private streams.
     stream = access_stream_for_delete_or_update(user_profile, stream_id)
+
+    if message_retention_days is not None:
+        if not user_profile.is_realm_owner:
+            raise OrganizationOwnerRequired()
+        user_profile.realm.ensure_not_on_limited_plan()
+        message_retention_days_value = parse_message_retention_days(message_retention_days)
+        do_change_stream_message_retention_days(stream, message_retention_days_value)
+
     if description is not None:
         if '\n' in description:
             # We don't allow newline characters in stream descriptions.
@@ -384,6 +405,8 @@ def add_subscriptions_backend(
         stream_post_policy: int=REQ(validator=check_int_in(
             Stream.STREAM_POST_POLICY_TYPES), default=Stream.STREAM_POST_POLICY_EVERYONE),
         history_public_to_subscribers: Optional[bool]=REQ(validator=check_bool, default=None),
+        message_retention_days: Union[str, int]=REQ(validator=check_string_or_int,
+                                                    default="realm_default"),
         announce: bool=REQ(validator=check_bool, default=False),
         principals: Sequence[Union[str, int]]=REQ(validator=check_variable_type([
             check_list(check_string), check_list(check_int)]), default=[]),
@@ -408,6 +431,7 @@ def add_subscriptions_backend(
         stream_dict_copy["invite_only"] = invite_only
         stream_dict_copy["stream_post_policy"] = stream_post_policy
         stream_dict_copy["history_public_to_subscribers"] = history_public_to_subscribers
+        stream_dict_copy["message_retention_days"] = parse_message_retention_days(message_retention_days)
         stream_dicts.append(stream_dict_copy)
 
     # Validation of the streams arguments, including enforcement of